taler/exchange
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

add wire_out tracking to exchangedb, including deferred constraint, and to aggregator

Browse Source
This commit is contained in:
Christian Grothoff 2017-03-18 03:44:59 +01:00
parent 5a9d7ac835
commit 6a98b07ff2
No known key found for this signature in database
GPG Key ID: 939E6BE1E29FC3CC
5 changed files with 212 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ChangeLog
View File

@ -1,3 +1,7 @@
Sat Mar 18 03:44:38 CET 2017
Add 'wire_out' table to exchange DB to track outgoing
wire transfers. -CG
Fri Nov 18 18:53:30 CET 2016 Fri Nov 18 18:53:30 CET 2016
Improved error reporting for bogus wire specifications. Improved error reporting for bogus wire specifications.
Releasing taler-exchange 0.2.0. -CG Releasing taler-exchange 0.2.0. -CG

50
src/exchange/taler-exchange-aggregator.c
View File

@ -705,8 +705,8 @@ run_aggregation (void *cls)
return; return;
} }
if (GNUNET_OK != if (GNUNET_OK !=
db_plugin->start (db_plugin->cls, db_plugin->start_deferred_wire_out (db_plugin->cls,
session)) session))
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to start database transaction!\n"); "Failed to start database transaction!\n");
@ -908,11 +908,6 @@ prepare_cb (void *cls,
{ {
struct TALER_EXCHANGEDB_Session *session = au->session; struct TALER_EXCHANGEDB_Session *session = au->session;
if (NULL != au->wire)
{
json_decref (au->wire);
au->wire = NULL;
}
GNUNET_free_non_null (au->additional_rows); GNUNET_free_non_null (au->additional_rows);
if (NULL == buf) if (NULL == buf)
{ {
@ -922,6 +917,11 @@ prepare_cb (void *cls,
/* start again */ /* start again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
if (NULL != au->wire)
{
json_decref (au->wire);
au->wire = NULL;
}
GNUNET_free (au); GNUNET_free (au);
au = NULL; au = NULL;
return; return;
@ -941,10 +941,46 @@ prepare_cb (void *cls,
/* start again */ /* start again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
if (NULL != au->wire)
{
json_decref (au->wire);
au->wire = NULL;
}
GNUNET_free (au); GNUNET_free (au);
au = NULL; au = NULL;
return; return;
} }
/* Commit the WTID data to 'wire_out' to finally satisfy aggregation
table constraints */
if (GNUNET_OK !=
db_plugin->store_wire_transfer_out (db_plugin->cls,
session,
au->execution_time,
&au->wtid,
au->wire,
&au->total_amount))
{
GNUNET_break (0); /* why? how to best recover? */
db_plugin->rollback (db_plugin->cls,
session);
/* start again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL);
if (NULL != au->wire)
{
json_decref (au->wire);
au->wire = NULL;
}
GNUNET_free (au);
au = NULL;
return;
}
if (NULL != au->wire)
{
json_decref (au->wire);
au->wire = NULL;
}
GNUNET_free (au); GNUNET_free (au);
au = NULL; au = NULL;

188
src/exchangedb/plugin_exchangedb_postgres.c
View File

@ -40,13 +40,23 @@
/** /**
* Log a really unexpected PQ error. * Log a really unexpected PQ error with all the details we can get hold of.
* *
* @param result PQ result object of the PQ operation that failed * @param result PQ result object of the PQ operation that failed
* @param conn SQL connection that was used
*/ */
#define BREAK_DB_ERR(result) do { \ #define BREAK_DB_ERR(result,conn) do { \
char *err = PQresultVerboseErrorMessage (result, PQERRORS_VERBOSE, PQSHOW_CONTEXT_ALWAYS); \
GNUNET_break (0); \ GNUNET_break (0); \
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Database failure: %s (%s)\n", PQresultErrorMessage (result), PQresStatus (PQresultStatus (result))); \ GNUNET_log (GNUNET_ERROR_TYPE_ERROR, \
"Database failure: %s/%s/%s/%s/%s/%s", \
PQresultErrorField (result, PG_DIAG_MESSAGE_PRIMARY), \
PQresultErrorField (result, PG_DIAG_MESSAGE_DETAIL), \
PQresultErrorMessage (result), \
PQresStatus (PQresultStatus (result)), \
PQerrorMessage(conn), \
err); \
PQfreemem (err); \
} while (0) } while (0)
@ -75,7 +85,7 @@
PGresult *result = PQexec (conn, sql); \ PGresult *result = PQexec (conn, sql); \
if (PGRES_COMMAND_OK != PQresultStatus (result)) \ if (PGRES_COMMAND_OK != PQresultStatus (result)) \
{ \ { \
BREAK_DB_ERR (result); \ BREAK_DB_ERR (result, conn); \
PQclear (result); \ PQclear (result); \
goto SQLEXEC_fail; \ goto SQLEXEC_fail; \
} \ } \
@ -134,7 +144,7 @@ struct PostgresClosure
* We already log whenever we care, so this function does nothing * We already log whenever we care, so this function does nothing
* and merely exists to silence the libpq logging. * and merely exists to silence the libpq logging.
* *
* @param arg NULL * @param arg the SQL connection that was used
* @param res information about some libpq event * @param res information about some libpq event
*/ */
static void static void
@ -149,7 +159,7 @@ pq_notice_receiver_cb (void *arg,
* Function called by libpq whenever it wants to log something. * Function called by libpq whenever it wants to log something.
* We log those using the Taler logger. * We log those using the Taler logger.
* *
* @param arg NULL * @param arg the SQL connection that was used
* @param message information about some libpq event * @param message information about some libpq event
*/ */
static void static void
@ -186,10 +196,10 @@ connect_to_postgres (struct PostgresClosure *pc)
} }
PQsetNoticeReceiver (conn, PQsetNoticeReceiver (conn,
&pq_notice_receiver_cb, &pq_notice_receiver_cb,
NULL); conn);
PQsetNoticeProcessor (conn, PQsetNoticeProcessor (conn,
&pq_notice_processor_cb, &pq_notice_processor_cb,
NULL); conn);
return conn; return conn;
} }
@ -465,11 +475,23 @@ postgres_create_tables (void *cls)
",PRIMARY KEY (coin_pub, merchant_pub, h_proposal_data, rtransaction_id)" /* this combo must be unique, and we usually select by coin_pub */ ",PRIMARY KEY (coin_pub, merchant_pub, h_proposal_data, rtransaction_id)" /* this combo must be unique, and we usually select by coin_pub */
") "); ") ");
/* This table contains the data for
wire transfers the exchange has executed. */
SQLEXEC("CREATE TABLE IF NOT EXISTS wire_out "
"(wireout_uuid BIGSERIAL PRIMARY KEY"
",execution_date INT8 NOT NULL"
",wtid_raw BYTEA UNIQUE NOT NULL CHECK (LENGTH(wtid_raw)=" TALER_WIRE_TRANSFER_IDENTIFIER_LEN_STR ")"
",wire_target TEXT NOT NULL"
",amount_val INT8 NOT NULL"
",amount_frac INT4 NOT NULL"
",amount_curr VARCHAR("TALER_CURRENCY_LEN_STR") NOT NULL"
")");
/* Table for the tracking API, mapping from wire transfer identifiers /* Table for the tracking API, mapping from wire transfer identifiers
to transactions and back */ to transactions and back */
SQLEXEC("CREATE TABLE IF NOT EXISTS aggregation_tracking " SQLEXEC("CREATE TABLE IF NOT EXISTS aggregation_tracking "
"(deposit_serial_id INT8 PRIMARY KEY REFERENCES deposits (deposit_serial_id) ON DELETE CASCADE" "(deposit_serial_id INT8 PRIMARY KEY REFERENCES deposits (deposit_serial_id) ON DELETE CASCADE"
",wtid_raw BYTEA NOT NULL CHECK (LENGTH(wtid_raw)=" TALER_WIRE_TRANSFER_IDENTIFIER_LEN_STR ")" ",wtid_raw BYTEA CONSTRAINT wire_out_ref REFERENCES wire_out(wtid_raw) ON DELETE CASCADE DEFERRABLE"
",execution_time INT8 NOT NULL" ",execution_time INT8 NOT NULL"
")"); ")");
/* Index for lookup_transactions statement on wtid */ /* Index for lookup_transactions statement on wtid */
@ -505,18 +527,6 @@ postgres_create_tables (void *cls)
SQLEXEC_INDEX("CREATE INDEX prepare_iteration_index " SQLEXEC_INDEX("CREATE INDEX prepare_iteration_index "
"ON prewire(type,finished)"); "ON prewire(type,finished)");
/* This table contains the data for
wire transfers the exchange has executed. */
SQLEXEC("CREATE TABLE IF NOT EXISTS wire_out "
"(wireout_uuid BIGSERIAL PRIMARY KEY"
",execution_date INT8 NOT NULL"
",wtid_raw BYTEA NOT NULL CHECK (LENGTH(wtid_raw)=" TALER_WIRE_TRANSFER_IDENTIFIER_LEN_STR ")"
",wire_target TEXT NOT NULL"
",amount_val INT8 NOT NULL"
",amount_frac INT4 NOT NULL"
",amount_curr VARCHAR("TALER_CURRENCY_LEN_STR") NOT NULL"
")");
#undef SQLEXEC #undef SQLEXEC
#undef SQLEXEC_INDEX #undef SQLEXEC_INDEX
@ -545,7 +555,7 @@ postgres_prepare (PGconn *db_conn)
result = PQprepare (db_conn, name, sql, __VA_ARGS__); \ result = PQprepare (db_conn, name, sql, __VA_ARGS__); \
if (PGRES_COMMAND_OK != PQresultStatus (result)) \ if (PGRES_COMMAND_OK != PQresultStatus (result)) \
{ \ { \
BREAK_DB_ERR (result); \ BREAK_DB_ERR (result, db_conn); \
PQclear (result); result = NULL; \ PQclear (result); result = NULL; \
return GNUNET_SYSERR; \ return GNUNET_SYSERR; \
} \ } \
@ -1270,7 +1280,7 @@ postgres_prepare (PGconn *db_conn)
"INSERT INTO aggregation_tracking " "INSERT INTO aggregation_tracking "
"(deposit_serial_id" "(deposit_serial_id"
",wtid_raw" ",wtid_raw"
",execution_time" ",execution_time" /* TODO: this field should be eliminated and obtained from wire_out */
") VALUES " ") VALUES "
"($1, $2, $3)", "($1, $2, $3)",
3, NULL); 3, NULL);
@ -1601,7 +1611,7 @@ postgres_insert_denomination_info (void *cls,
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
ret = GNUNET_SYSERR; ret = GNUNET_SYSERR;
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
} }
else else
{ {
@ -2311,7 +2321,7 @@ postgres_have_deposit (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2402,7 +2412,7 @@ postgres_mark_deposit_tiny (void *cls,
if (PGRES_COMMAND_OK != if (PGRES_COMMAND_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2441,7 +2451,7 @@ postgres_test_deposit_done (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2507,7 +2517,7 @@ postgres_mark_deposit_done (void *cls,
if (PGRES_COMMAND_OK != if (PGRES_COMMAND_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2549,7 +2559,7 @@ postgres_get_ready_deposit (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2651,7 +2661,7 @@ postgres_iterate_matching_deposits (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2742,7 +2752,7 @@ get_known_coin (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2809,7 +2819,7 @@ insert_known_coin (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -2874,7 +2884,7 @@ postgres_insert_deposit (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
ret = GNUNET_SYSERR; ret = GNUNET_SYSERR;
} }
else else
@ -2919,7 +2929,7 @@ postgres_insert_refund (void *cls,
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
ret = GNUNET_SYSERR; ret = GNUNET_SYSERR;
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
} }
else else
{ {
@ -2959,7 +2969,7 @@ postgres_get_refresh_session (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3076,7 +3086,7 @@ postgres_create_refresh_session (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3124,7 +3134,7 @@ postgres_insert_refresh_order (void *cls,
} }
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3199,7 +3209,7 @@ postgres_get_refresh_order (void *cls,
} }
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
free_dpk_result (denom_pubs, i); free_dpk_result (denom_pubs, i);
return GNUNET_SYSERR; return GNUNET_SYSERR;
@ -3275,7 +3285,7 @@ postgres_insert_refresh_commit_coins (void *cls,
} }
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3354,7 +3364,7 @@ postgres_get_refresh_commit_coins (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
postgres_free_refresh_commit_coins (cls, postgres_free_refresh_commit_coins (cls,
i, i,
@ -3426,7 +3436,7 @@ postgres_insert_refresh_transfer_public_key (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3470,7 +3480,7 @@ postgres_get_refresh_transfer_public_key (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3538,7 +3548,7 @@ postgres_get_refresh_out (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3595,7 +3605,7 @@ postgres_insert_refresh_out (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3633,7 +3643,7 @@ postgres_get_link_data_list (void *cls,
ldl = NULL; ldl = NULL;
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return NULL; return NULL;
} }
@ -3719,7 +3729,7 @@ postgres_get_transfer (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -3872,7 +3882,7 @@ postgres_get_coin_transactions (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
goto cleanup; goto cleanup;
} }
@ -3943,7 +3953,7 @@ postgres_get_coin_transactions (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
goto cleanup; goto cleanup;
} }
@ -4043,7 +4053,7 @@ postgres_lookup_wire_transfer (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4159,7 +4169,7 @@ postgres_wire_lookup_deposit_wtid (void *cls,
params); params);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4187,7 +4197,7 @@ postgres_wire_lookup_deposit_wtid (void *cls,
params2); params2);
if (PGRES_TUPLES_OK != PQresultStatus (result)) if (PGRES_TUPLES_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4301,7 +4311,7 @@ postgres_insert_aggregation_tracking (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4361,7 +4371,7 @@ postgres_get_wire_fee (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4465,7 +4475,7 @@ postgres_insert_wire_fee (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4509,7 +4519,7 @@ postgres_wire_prepare_data_insert (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4543,7 +4553,7 @@ postgres_wire_prepare_data_mark_finished (void *cls,
if (PGRES_COMMAND_OK != if (PGRES_COMMAND_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4633,6 +4643,51 @@ postgres_wire_prepare_data_get (void *cls,
} }
/**
* Start a transaction where we transiently violate the foreign
* constraints on the "wire_out" table as we insert aggregations
* and only add the wire transfer out at the end.
*
* @param cls the @e cls of this struct with the plugin-specific state
* @param session connection to use
* @return #GNUNET_OK on success
*/
static int
postgres_start_deferred_wire_out (void *cls,
struct TALER_EXCHANGEDB_Session *session)
{
PGresult *result;
ExecStatusType ex;
result = PQexec (session->conn,
"START TRANSACTION ISOLATION LEVEL SERIALIZABLE");
if (PGRES_COMMAND_OK !=
(ex = PQresultStatus (result)))
{
TALER_LOG_ERROR ("Failed to start transaction (%s): %s\n",
PQresStatus (ex),
PQerrorMessage (session->conn));
GNUNET_break (0);
PQclear (result);
return GNUNET_SYSERR;
}
result = PQexec (session->conn,
"SET CONSTRAINTS wire_out_ref DEFERRED");
if (PGRES_COMMAND_OK !=
(ex = PQresultStatus (result)))
{
TALER_LOG_ERROR ("Failed to defer wire_out_ref constraint on transaction (%s): %s\n",
PQresStatus (ex),
PQerrorMessage (session->conn));
GNUNET_break (0);
PQclear (result);
return GNUNET_SYSERR;
}
PQclear (result);
return GNUNET_OK;
}
/** /**
* Store information about an outgoing wire transfer that was executed. * Store information about an outgoing wire transfer that was executed.
* *
@ -4667,7 +4722,7 @@ postgres_store_wire_transfer_out (void *cls,
params); params);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4714,7 +4769,7 @@ postgres_gc (void *cls)
params_none); params_none);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, conn);
PQclear (result); PQclear (result);
PQfinish (conn); PQfinish (conn);
return GNUNET_SYSERR; return GNUNET_SYSERR;
@ -4725,7 +4780,7 @@ postgres_gc (void *cls)
params_time); params_time);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, conn);
PQclear (result); PQclear (result);
PQfinish (conn); PQfinish (conn);
return GNUNET_SYSERR; return GNUNET_SYSERR;
@ -4736,7 +4791,7 @@ postgres_gc (void *cls)
params_time); params_time);
if (PGRES_COMMAND_OK != PQresultStatus (result)) if (PGRES_COMMAND_OK != PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, conn);
PQclear (result); PQclear (result);
PQfinish (conn); PQfinish (conn);
return GNUNET_SYSERR; return GNUNET_SYSERR;
@ -4777,7 +4832,7 @@ postgres_select_deposits_above_serial_id (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4881,7 +4936,7 @@ postgres_select_refreshs_above_serial_id (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -4978,7 +5033,7 @@ postgres_select_refunds_above_serial_id (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -5069,7 +5124,7 @@ postgres_select_reserves_in_above_serial_id (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -5164,7 +5219,7 @@ postgres_select_reserves_out_above_serial_id (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -5266,7 +5321,7 @@ postgres_select_wire_out_above_serial_id (void *cls,
if (PGRES_TUPLES_OK != if (PGRES_TUPLES_OK !=
PQresultStatus (result)) PQresultStatus (result))
{ {
BREAK_DB_ERR (result); BREAK_DB_ERR (result, session->conn);
PQclear (result); PQclear (result);
return GNUNET_SYSERR; return GNUNET_SYSERR;
} }
@ -5417,6 +5472,7 @@ libtaler_plugin_exchangedb_postgres_init (void *cls)
plugin->wire_prepare_data_insert = &postgres_wire_prepare_data_insert; plugin->wire_prepare_data_insert = &postgres_wire_prepare_data_insert;
plugin->wire_prepare_data_mark_finished = &postgres_wire_prepare_data_mark_finished; plugin->wire_prepare_data_mark_finished = &postgres_wire_prepare_data_mark_finished;
plugin->wire_prepare_data_get = &postgres_wire_prepare_data_get; plugin->wire_prepare_data_get = &postgres_wire_prepare_data_get;
plugin->start_deferred_wire_out = &postgres_start_deferred_wire_out;
plugin->store_wire_transfer_out = &postgres_store_wire_transfer_out; plugin->store_wire_transfer_out = &postgres_store_wire_transfer_out;
plugin->gc = &postgres_gc; plugin->gc = &postgres_gc;
plugin->select_deposits_above_serial_id = &postgres_select_deposits_above_serial_id; plugin->select_deposits_above_serial_id = &postgres_select_deposits_above_serial_id;

41
src/exchangedb/test_exchangedb.c
View File

@ -1241,20 +1241,13 @@ test_wire_out (struct TALER_EXCHANGEDB_Session *session,
GNUNET_assert (GNUNET_OK == GNUNET_assert (GNUNET_OK ==
TALER_string_to_amount (CURRENCY ":1", TALER_string_to_amount (CURRENCY ":1",
&wire_out_amount)); &wire_out_amount));
/* we will transiently violate the wtid constraint on
the aggregation table, so we need to start the special
transaction where this is allowed... */
FAILIF (GNUNET_OK != FAILIF (GNUNET_OK !=
plugin->store_wire_transfer_out (plugin->cls, plugin->start_deferred_wire_out (plugin->cls,
session, session));
wire_out_date,
&wire_out_wtid,
wire_out_account,
&wire_out_amount));
FAILIF (GNUNET_OK !=
plugin->select_wire_out_above_serial_id (plugin->cls,
session,
0,
&audit_wire_cb,
NULL));
FAILIF (1 != auditor_row_cnt);
/* setup values for wire transfer aggregation data */ /* setup values for wire transfer aggregation data */
merchant_pub_wt = deposit->merchant_pub; merchant_pub_wt = deposit->merchant_pub;
@ -1289,6 +1282,7 @@ test_wire_out (struct TALER_EXCHANGEDB_Session *session,
&cb_wtid_never, &cb_wtid_never,
NULL)); NULL));
} }
wtid_wt = wire_out_wtid; /* to statisfy foreign constraint */
/* insert WT data */ /* insert WT data */
FAILIF (GNUNET_OK != FAILIF (GNUNET_OK !=
plugin->insert_aggregation_tracking (plugin->cls, plugin->insert_aggregation_tracking (plugin->cls,
@ -1312,6 +1306,27 @@ test_wire_out (struct TALER_EXCHANGEDB_Session *session,
&cb_wtid_check, &cb_wtid_check,
&cb_wtid_never)); &cb_wtid_never));
/* Now let's fix the transient constraint violation by
putting in the WTID into the wire_out table */
FAILIF (GNUNET_OK !=
plugin->store_wire_transfer_out (plugin->cls,
session,
wire_out_date,
&wire_out_wtid,
wire_out_account,
&wire_out_amount));
FAILIF (GNUNET_OK !=
plugin->select_wire_out_above_serial_id (plugin->cls,
session,
0,
&audit_wire_cb,
NULL));
FAILIF (1 != auditor_row_cnt);
/* And now the commit should still succeed! */
FAILIF (GNUNET_OK !=
plugin->commit (plugin->cls,
session));
return GNUNET_OK; return GNUNET_OK;
drop: drop:

15
src/include/taler_exchangedb_plugin.h
View File

@ -1639,6 +1639,21 @@ struct TALER_EXCHANGEDB_Plugin
void *cb_cls); void *cb_cls);
/**
* Start a transaction where we transiently violate the foreign
* constraints on the "wire_out" table as we insert aggregations
* and only add the wire transfer out at the end.
*
* @param cls the @e cls of this struct with the plugin-specific state
* @param session connection to use
* @return #GNUNET_OK on success
*/
int
(*start_deferred_wire_out) (void *cls,
struct TALER_EXCHANGEDB_Session *session);
/** /**
* Store information about an outgoing wire transfer that was executed. * Store information about an outgoing wire transfer that was executed.
* *