[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[GNUnet-SVN] [taler-exchange] branch master updated: refactor wire audit
|
From: |
gnunet |
|
Subject: |
[GNUnet-SVN] [taler-exchange] branch master updated: refactor wire auditor to properly handle multiple accounts |
|
Date: |
Sat, 24 Aug 2019 22:49:39 +0200 |
This is an automated email from the git hooks/post-receive script.
grothoff pushed a commit to branch master
in repository exchange.
The following commit(s) were added to refs/heads/master by this push:
new 71ae493c refactor wire auditor to properly handle multiple accounts
71ae493c is described below
commit 71ae493c7a7601b418de4f5c83159d46abf677b8
Author: Christian Grothoff <address@hidden>
AuthorDate: Sat Aug 24 22:49:35 2019 +0200
refactor wire auditor to properly handle multiple accounts
---
src/auditor/auditor.conf | 4 +
src/auditor/taler-wire-auditor.c | 735 ++++++++++++++++++------------
src/auditor/test-auditor.conf | 1 +
src/auditordb/plugin_auditordb_postgres.c | 217 ++++++---
src/include/taler_auditordb_plugin.h | 116 +++--
5 files changed, 691 insertions(+), 382 deletions(-)
diff --git a/src/auditor/auditor.conf b/src/auditor/auditor.conf
index c3b9a00c..a099f9b7 100644
--- a/src/auditor/auditor.conf
+++ b/src/auditor/auditor.conf
@@ -4,6 +4,10 @@
# Which database backend do we use for the auditor?
DB = postgres
+# Option specifying which amount is considered 'tiny'
+# and thus doesn't need to be wire-transferred.
+#TINY_AMOUNT = KUDOS:0.01
+
# Where do we store the auditor's private key?
AUDITOR_PRIV_FILE = ${TALER_DATA_HOME}/auditor/offline-keys/auditor.priv
diff --git a/src/auditor/taler-wire-auditor.c b/src/auditor/taler-wire-auditor.c
index 3b84a641..16104a64 100644
--- a/src/auditor/taler-wire-auditor.c
+++ b/src/auditor/taler-wire-auditor.c
@@ -67,6 +67,31 @@ struct WireAccount
char *section_name;
/**
+ * Active wire request for the transaction history.
+ */
+ struct TALER_WIRE_HistoryHandle *hh;
+
+ /**
+ * Progress point for this account.
+ */
+ struct TALER_AUDITORDB_WireAccountProgressPoint pp;
+
+ /**
+ * Where we are in the inbound (CREDIT) transaction history.
+ */
+ void *in_wire_off;
+
+ /**
+ * Where we are in the inbound (DEBIT) transaction history.
+ */
+ void *out_wire_off;
+
+ /**
+ * Number of bytes in #in_wire_off and #out_wire_off.
+ */
+ size_t wire_off_size;
+
+ /**
* We should check for inbound transactions to this account.
*/
int watch_credit;
@@ -148,22 +173,6 @@ static struct WireAccount *wa_head;
static struct WireAccount *wa_tail;
/**
- * Handle to the wire plugin for wire operations.
- */
-static struct TALER_WIRE_Plugin *wp;
-
-/**
- * Name of the section that configures the account
- * we are currently processing (matches #wp).
- */
-static char *wp_section_name;
-
-/**
- * Active wire request for the transaction history.
- */
-static struct TALER_WIRE_HistoryHandle *hh;
-
-/**
* Query status for the incremental processing status in the auditordb.
*/
static enum GNUNET_DB_QueryStatus qsx;
@@ -174,21 +183,6 @@ static enum GNUNET_DB_QueryStatus qsx;
static struct TALER_AUDITORDB_WireProgressPoint pp;
/**
- * Where we are in the inbound (CREDIT) transaction history.
- */
-static void *in_wire_off;
-
-/**
- * Where we are in the inbound (DEBIT) transaction history.
- */
-static void *out_wire_off;
-
-/**
- * Number of bytes in #in_wire_off and #out_wire_off.
- */
-static size_t wire_off_size;
-
-/**
* Array of reports about row inconsitencies in wire_out table.
*/
static json_t *report_wire_out_inconsistencies;
@@ -226,6 +220,11 @@ static json_t *report_row_minor_inconsistencies;
static json_t *report_lags;
/**
+ * Amount that is considered "tiny"
+ */
+static struct TALER_Amount tiny_amount;
+
+/**
* Total amount that was transferred too much from the exchange.
*/
static struct TALER_Amount total_bad_amount_out_plus;
@@ -434,12 +433,6 @@ do_shutdown (void *cls)
report_lags = NULL;
report_wire_format_inconsistencies = NULL;
}
- if (NULL != hh)
- {
- wp->get_history_cancel (wp->cls,
- hh);
- hh = NULL;
- }
if (NULL != in_map)
{
GNUNET_CONTAINER_multihashmap_iterate (in_map,
@@ -456,23 +449,23 @@ do_shutdown (void *cls)
GNUNET_CONTAINER_multihashmap_destroy (out_map);
out_map = NULL;
}
- if (NULL != wp)
- {
- TALER_WIRE_plugin_unload (wp);
- wp = NULL;
- }
- if (NULL != wp_section_name)
- {
- GNUNET_free (wp_section_name);
- wp_section_name = NULL;
- }
while (NULL != (wa = wa_head))
{
+ if (NULL != wa->hh)
+ {
+ struct TALER_WIRE_Plugin *wp = wa->wire_plugin;
+
+ wp->get_history_cancel (wp->cls,
+ wa->hh);
+ wa->hh = NULL;
+ }
GNUNET_CONTAINER_DLL_remove (wa_head,
wa_tail,
wa);
TALER_WIRE_plugin_unload (wa->wire_plugin);
GNUNET_free (wa->section_name);
+ GNUNET_free_non_null (wa->in_wire_off);
+ GNUNET_free_non_null (wa->out_wire_off);
GNUNET_free (wa);
}
if (NULL != adb)
@@ -534,25 +527,46 @@ commit (enum GNUNET_DB_QueryStatus qs)
esession);
return qs;
}
+ for (struct WireAccount *wa = wa_head;
+ NULL != wa;
+ wa = wa->next)
+ {
+ if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qsx)
+ qs = adb->update_wire_auditor_account_progress (adb->cls,
+ asession,
+ &master_pub,
+ wa->section_name,
+ &wa->pp,
+ wa->in_wire_off,
+ wa->out_wire_off,
+ wa->wire_off_size);
+ else
+ qs = adb->insert_wire_auditor_account_progress (adb->cls,
+ asession,
+ &master_pub,
+ wa->section_name,
+ &wa->pp,
+ wa->in_wire_off,
+ wa->out_wire_off,
+ wa->wire_off_size);
+ if (0 >= qs)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Failed to update auditor DB, not recording progress\n");
+ GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+ return qs;
+ }
+ }
if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qsx)
qs = adb->update_wire_auditor_progress (adb->cls,
asession,
&master_pub,
- wp_section_name,
- &pp,
- in_wire_off,
- out_wire_off,
- wire_off_size);
+ &pp);
else
qs = adb->insert_wire_auditor_progress (adb->cls,
asession,
&master_pub,
- wp_section_name,
- &pp,
- in_wire_off,
- out_wire_off,
- wire_off_size);
-
+ &pp);
if (0 >= qs)
{
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@@ -561,9 +575,8 @@ commit (enum GNUNET_DB_QueryStatus qs)
return qs;
}
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- _("Concluded audit step at %llu/%llu\n"),
- (unsigned long long) pp.last_reserve_in_serial_id,
- (unsigned long long) pp.last_wire_out_serial_id);
+ "Concluded audit step at %s\n",
+ GNUNET_STRINGS_absolute_time_to_string (pp.last_timestamp));
if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs)
{
@@ -602,16 +615,111 @@ commit (enum GNUNET_DB_QueryStatus qs)
}
+/* ***************************** Analyze required transfers
************************ */
+
+/**
+ * Function called on deposits that are past their due date
+ * and have not yet seen a wire transfer.
+ *
+ * @param cls closure
+ * @param rowid deposit table row of the coin's deposit
+ * @param coin_pub public key of the coin
+ * @param amount value of the deposit, including fee
+ * @param wire where should the funds be wired
+ * @param deadline what was the requested wire transfer deadline
+ * @param tiny did the exchange defer this transfer because it is too small?
+ * @param done did the exchange claim that it made a transfer?
+ */
+static void
+wire_missing_cb (void *cls,
+ uint64_t rowid,
+ const struct TALER_CoinSpendPublicKeyP *coin_pub,
+ const struct TALER_Amount *amount,
+ const json_t *wire,
+ struct GNUNET_TIME_Absolute deadline,
+ /* bool? */ int tiny,
+ /* bool? */ int done)
+{
+ GNUNET_break (GNUNET_OK ==
+ TALER_amount_add (&total_amount_lag,
+ &total_amount_lag,
+ amount));
+ if ( (GNUNET_YES == tiny) &&
+ (0 > TALER_amount_cmp (amount,
+ &tiny_amount)) )
+ return; /* acceptable, amount was tiny */
+ report (report_lags,
+ json_pack ("{s:I, s:o, s:s, s:s, s:o, s:O}",
+ "row", (json_int_t) rowid,
+ "amount", TALER_JSON_from_amount (amount),
+ "deadline", GNUNET_STRINGS_absolute_time_to_string
(deadline),
+ "claimed_done", (done) ? "yes" : "no",
+ "coin_pub", GNUNET_JSON_from_data_auto (coin_pub),
+ "account", wire));
+
+}
+
+
+/**
+ * Checks that all wire transfers that should have happened
+ * (based on deposits) have indeed happened.
+ *
+ * FIXME: this check _might_ rather belong with the
+ * taler-auditor logic.
+ */
+static void
+check_for_required_transfers ()
+{
+ struct GNUNET_TIME_Absolute next_timestamp;
+ enum GNUNET_DB_QueryStatus qs;
+
+ next_timestamp = GNUNET_TIME_absolute_get ();
+ /* Subtract #GRACE_PERIOD, so we can be a bit behind in processing
+ without immediately raising undue concern */
+ next_timestamp = GNUNET_TIME_absolute_subtract (next_timestamp,
+ GRACE_PERIOD);
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Analyzing exchange's unfinished deposits\n");
+ qs = edb->select_deposits_missing_wire (edb->cls,
+ esession,
+ pp.last_timestamp,
+ next_timestamp,
+ &wire_missing_cb,
+ &next_timestamp);
+ if (0 > qs)
+ {
+ GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+ global_ret = 1;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ pp.last_timestamp = next_timestamp;
+ /* conclude with success */
+ commit (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT);
+}
+
+
/* ***************************** Analyze reserves_out ************************
*/
+/**
+ * Clean up after processing wire out data.
+ */
+static void
+conclude_wire_out ()
+{
+ GNUNET_CONTAINER_multihashmap_destroy (out_map);
+ out_map = NULL;
+ check_for_required_transfers ();
+}
+
/**
* Function called with details about outgoing wire transfers
* as claimed by the exchange DB.
*
- * @param cls NULL
+ * @param cls a `struct WireAccount`
* @param rowid unique serial ID for the refresh session in our DB
- * @param date timestamp of the wire transfer (roughly)
+ * @param date timestamp of the transfer (roughly)
* @param wtid wire transfer subject
* @param wire wire transfer details of the receiver
* @param amount amount that was wired
@@ -625,6 +733,7 @@ wire_out_cb (void *cls,
const json_t *wire,
const struct TALER_Amount *amount)
{
+ struct WireAccount *wa = cls;
struct GNUNET_HashCode key;
struct ReserveOutInfo *roi;
@@ -757,6 +866,7 @@ cleanup:
free_roi (NULL,
&key,
roi));
+ wa->pp.last_wire_out_serial_id = rowid + 1;
return GNUNET_OK;
}
@@ -765,7 +875,7 @@ cleanup:
* Complain that we failed to match an entry from #out_map. This
* means a wire transfer was made without proper justification.
*
- * @param cls NULL
+ * @param cls a `struct WireAccount`
* @param key unused key
* @param value the `struct ReserveOutInfo` to report
* @return #GNUNET_OK
@@ -775,8 +885,10 @@ complain_out_not_found (void *cls,
const struct GNUNET_HashCode *key,
void *value)
{
+ struct WireAccount *wa = cls;
struct ReserveOutInfo *roi = value;
+ (void) wa; // FIXME: log which account is affected...
report (report_wire_out_inconsistencies,
json_pack ("{s:I, s:o, s:o, s:o, s:s, s:s}",
"row", (json_int_t) 0,
@@ -796,84 +908,36 @@ complain_out_not_found (void *cls,
/**
- * Function called on deposits that are past their due date
- * and have not yet seen a wire transfer.
- *
- * @param cls closure
- * @param rowid deposit table row of the coin's deposit
- * @param coin_pub public key of the coin
- * @param amount value of the deposit, including fee
- * @param wire where should the funds be wired
- * @param deadline what was the requested wire transfer deadline
- * @param tiny did the exchange defer this transfer because it is too small?
- * @param done did the exchange claim that it made a transfer?
- */
-static void
-wire_missing_cb (void *cls,
- uint64_t rowid,
- const struct TALER_CoinSpendPublicKeyP *coin_pub,
- const struct TALER_Amount *amount,
- const json_t *wire,
- struct GNUNET_TIME_Absolute deadline,
- /* bool? */ int tiny,
- /* bool? */ int done)
-{
- GNUNET_break (GNUNET_OK ==
- TALER_amount_add (&total_amount_lag,
- &total_amount_lag,
- amount));
- if (GNUNET_YES == tiny)
- {
- struct TALER_Amount rounded;
-
- rounded = *amount;
- GNUNET_break (GNUNET_SYSERR !=
- wp->amount_round (wp->cls,
- &rounded));
- if (0 == TALER_amount_cmp (&rounded,
- &zero))
- return; /* acceptable, amount was tiny */
- }
- report (report_lags,
- json_pack ("{s:I, s:o, s:s, s:s, s:o, s:O}",
- "row", (json_int_t) rowid,
- "amount", TALER_JSON_from_amount (amount),
- "deadline", GNUNET_STRINGS_absolute_time_to_string
(deadline),
- "claimed_done", (done) ? "yes" : "no",
- "coin_pub", GNUNET_JSON_from_data_auto (coin_pub),
- "account", wire));
-
-}
-
-
-/**
- * Start processing the next wire account.
- * Shuts down if we are done.
+ * Main function for processing 'reserves_out' data. We start by going over
+ * the DEBIT transactions this time, and then verify that all of them are
+ * justified by 'reserves_out'.
*
- * @param cls NULL
+ * @param cls `struct WireAccount` with a wire account list to process
*/
static void
-process_next_account (void *cls);
+process_debits (void *cls);
/**
* Go over the "wire_out" table of the exchange and
* verify that all wire outs are in that table.
+ *
+ * @param wa wire account we are processing
*/
static void
-check_exchange_wire_out ()
+check_exchange_wire_out (struct WireAccount *wa)
{
enum GNUNET_DB_QueryStatus qs;
- struct GNUNET_TIME_Absolute next_timestamp;
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Analyzing exchange's wire OUT table\n");
+ "Analyzing exchange's wire OUT table for account `%s'\n",
+ wa->section_name);
qs = edb->select_wire_out_above_serial_id_by_account (edb->cls,
esession,
- wp_section_name,
-
pp.last_wire_out_serial_id,
+ wa->section_name,
+
wa->pp.last_wire_out_serial_id,
&wire_out_cb,
- NULL);
+ wa);
if (0 > qs)
{
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
@@ -882,41 +946,13 @@ check_exchange_wire_out ()
return;
}
GNUNET_CONTAINER_multihashmap_iterate (out_map,
- &complain_out_not_found,
- NULL);
+ &complain_out_not_found,
+ wa);
/* clean up */
GNUNET_CONTAINER_multihashmap_iterate (out_map,
- &free_roi,
- NULL);
- GNUNET_CONTAINER_multihashmap_destroy (out_map);
- out_map = NULL;
-
- /* now check that all wire transfers that should have happened,
- have indeed happened */
- next_timestamp = GNUNET_TIME_absolute_get ();
- /* Subtract #GRACE_PERIOD, so we can be a bit behind in processing
- without immediately raising undue concern */
- next_timestamp = GNUNET_TIME_absolute_subtract (next_timestamp,
- GRACE_PERIOD);
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Analyzing exchange's unfinished deposits\n");
- qs = edb->select_deposits_missing_wire (edb->cls,
- esession,
- pp.last_timestamp,
- next_timestamp,
- &wire_missing_cb,
- &next_timestamp);
- if (0 > qs)
- {
- GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
- global_ret = 1;
- GNUNET_SCHEDULER_shutdown ();
- return;
- }
- pp.last_timestamp = next_timestamp;
-
- /* continue with next account: */
- process_next_account (NULL);
+ &free_roi,
+ NULL);
+ process_debits (wa->next);
}
@@ -925,7 +961,7 @@ check_exchange_wire_out ()
* are credited to the exchange's account (incoming
* transactions).
*
- * @param cls closure
+ * @param cls `struct WireAccount` with current wire account to process
* @param ec error code in case something went wrong
* @param dir direction of the transfer
* @param row_off identification of the position at which we are querying
@@ -941,10 +977,10 @@ history_debit_cb (void *cls,
size_t row_off_size,
const struct TALER_WIRE_TransferDetails *details)
{
+ struct WireAccount *wa = cls;
struct ReserveOutInfo *roi;
struct GNUNET_HashCode rowh;
-
if (TALER_BANK_DIRECTION_NONE == dir)
{
if (TALER_EC_NONE != ec)
@@ -954,10 +990,8 @@ history_debit_cb (void *cls,
"Error fetching history: %u!\n",
(unsigned int) ec);
}
- /* end of iteration, now check wire_out to see
- if it matches #out_map */
- hh = NULL;
- check_exchange_wire_out ();
+ wa->hh = NULL;
+ check_exchange_wire_out (wa);
return GNUNET_OK;
}
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@@ -987,19 +1021,38 @@ history_debit_cb (void *cls,
GNUNET_free (diagnostic);
return GNUNET_OK;
}
+
+ /* Update offset */
+ if (NULL == wa->out_wire_off)
+ {
+ wa->wire_off_size = row_off_size;
+ wa->out_wire_off = GNUNET_malloc (row_off_size);
+ }
+ if (wa->wire_off_size != row_off_size)
+ {
+ GNUNET_break (0);
+ commit (GNUNET_DB_STATUS_HARD_ERROR);
+ wa->hh = NULL;
+ GNUNET_SCHEDULER_shutdown ();
+ return GNUNET_SYSERR;
+ }
+ memcpy (wa->out_wire_off,
+ row_off,
+ row_off_size);
+
roi = GNUNET_new (struct ReserveOutInfo);
GNUNET_CRYPTO_hash (&details->wtid,
- sizeof (details->wtid),
- &roi->subject_hash);
+ sizeof (details->wtid),
+ &roi->subject_hash);
roi->details.amount = details->amount;
roi->details.execution_date = details->execution_date;
roi->details.wtid = details->wtid;
roi->details.account_url = GNUNET_strdup (details->account_url);
if (GNUNET_OK !=
GNUNET_CONTAINER_multihashmap_put (out_map,
- &roi->subject_hash,
- roi,
-
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
+ &roi->subject_hash,
+ roi,
+
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
{
char *diagnostic;
@@ -1026,31 +1079,43 @@ history_debit_cb (void *cls,
/**
- * Main function for processing 'reserves_out' data.
- * We start by going over the DEBIT transactions this
- * time, and then verify that all of them are justified
- * by 'reserves_out'.
+ * Main function for processing 'reserves_out' data. We start by going over
+ * the DEBIT transactions this time, and then verify that all of them are
+ * justified by 'reserves_out'.
+ *
+ * @param cls `struct WireAccount` with a wire account list to process
*/
static void
-process_debits ()
+process_debits (void *cls)
{
+ struct WireAccount *wa = cls;
+ struct TALER_WIRE_Plugin *wp;
+
+ if (NULL == wa)
+ {
+ /* end of iteration, now check wire_out to see
+ if it matches #out_map */
+ conclude_wire_out ();
+ return;
+ }
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Checking bank DEBIT records\n");
- GNUNET_assert (NULL == hh);
- out_map = GNUNET_CONTAINER_multihashmap_create (1024,
- GNUNET_YES);
- hh = wp->get_history (wp->cls,
- wp_section_name,
- TALER_BANK_DIRECTION_DEBIT,
- out_wire_off,
- wire_off_size,
- INT64_MAX,
- &history_debit_cb,
- NULL);
- if (NULL == hh)
+ "Checking bank DEBIT records of account `%s'\n",
+ wa->section_name);
+ GNUNET_assert (NULL == wa->hh);
+ wp = wa->wire_plugin;
+ wa->hh = wp->get_history (wp->cls,
+ wa->section_name,
+ TALER_BANK_DIRECTION_DEBIT,
+ wa->out_wire_off,
+ wa->wire_off_size,
+ INT64_MAX,
+ &history_debit_cb,
+ wa);
+ if (NULL == wa->hh)
{
fprintf (stderr,
- "Failed to obtain bank transaction history\n");
+ "Failed to obtain bank transaction history for `%s'\n",
+ wa->section_name);
commit (GNUNET_DB_STATUS_HARD_ERROR);
global_ret = 1;
GNUNET_SCHEDULER_shutdown ();
@@ -1059,14 +1124,40 @@ process_debits ()
}
+/**
+ * Begin analyzing wire_out.
+ */
+static void
+begin_debit_audit ()
+{
+ out_map = GNUNET_CONTAINER_multihashmap_create (1024,
+ GNUNET_YES);
+ process_debits (wa_head);
+}
+
+
/* ***************************** Analyze reserves_in ************************
*/
+/**
+ * Conclude the credit history check by logging entries that
+ * were not found and freeing resources. Then move on to
+ * processing debits.
+ */
+static void
+conclude_credit_history ()
+{
+ GNUNET_CONTAINER_multihashmap_destroy (in_map);
+ in_map = NULL;
+ /* credit done, now check debits */
+ begin_debit_audit ();
+}
+
/**
* Function called with details about incoming wire transfers
* as claimed by the exchange DB.
*
- * @param cls NULL
+ * @param cls a `struct WireAccount` we are processing
* @param rowid unique serial ID for the refresh session in our DB
* @param reserve_pub public key of the reserve (also the WTID)
* @param credit amount that was received
@@ -1086,6 +1177,7 @@ reserve_in_cb (void *cls,
size_t wire_reference_size,
struct GNUNET_TIME_Absolute execution_date)
{
+ struct WireAccount *wa = cls;
struct ReserveInInfo *rii;
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@@ -1125,7 +1217,7 @@ reserve_in_cb (void *cls,
GNUNET_free (rii);
return GNUNET_OK;
}
- pp.last_reserve_in_serial_id = rowid + 1;
+ wa->pp.last_reserve_in_serial_id = rowid + 1;
return GNUNET_OK;
}
@@ -1133,7 +1225,7 @@ reserve_in_cb (void *cls,
/**
* Complain that we failed to match an entry from #in_map.
*
- * @param cls NULL
+ * @param cls a `struct WireAccount`
* @param key unused key
* @param value the `struct ReserveInInfo` to free
* @return #GNUNET_OK
@@ -1143,8 +1235,10 @@ complain_in_not_found (void *cls,
const struct GNUNET_HashCode *key,
void *value)
{
+ struct WireAccount *wa = cls;
struct ReserveInInfo *rii = value;
+ (void) wa; // FIXME: log which account is affected...
report (report_reserve_in_inconsistencies,
json_pack ("{s:I, s:o, s:o, s:o, s:s, s:s}",
"row", (json_int_t) rii->rowid,
@@ -1162,24 +1256,13 @@ complain_in_not_found (void *cls,
/**
- * Conclude the credit history check by logging entries that
- * were not found and freeing resources. Then move on to
- * processing debits.
+ * Start processing the next wire account.
+ * Shuts down if we are done.
+ *
+ * @param cls `struct WireAccount` with a wire account list to process
*/
static void
-conclude_credit_history ()
-{
- GNUNET_CONTAINER_multihashmap_iterate (in_map,
- &complain_in_not_found,
- NULL);
- /* clean up before 2nd phase */
- GNUNET_CONTAINER_multihashmap_iterate (in_map,
- &free_rii,
- NULL);
- GNUNET_CONTAINER_multihashmap_destroy (in_map);
- in_map = NULL;
- process_debits ();
-}
+process_credits (void *cls);
/**
@@ -1187,7 +1270,7 @@ conclude_credit_history ()
* are credited to the exchange's account (incoming
* transactions).
*
- * @param cls closure
+ * @param cls `struct WireAccount` we are processing
* @param ec error code in case something went wrong
* @param dir direction of the transfer
* @param row_off identification of the position at which we are querying
@@ -1203,6 +1286,7 @@ history_credit_cb (void *cls,
size_t row_off_size,
const struct TALER_WIRE_TransferDetails *details)
{
+ struct WireAccount *wa = cls;
struct ReserveInInfo *rii;
struct GNUNET_HashCode key;
@@ -1216,8 +1300,18 @@ history_credit_cb (void *cls,
(unsigned int) ec);
}
/* end of operation */
- hh = NULL;
- conclude_credit_history ();
+ wa->hh = NULL;
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Reconciling CREDIT processing of account `%s'\n",
+ wa->section_name);
+ GNUNET_CONTAINER_multihashmap_iterate (in_map,
+ &complain_in_not_found,
+ wa);
+ /* clean up before 2nd phase */
+ GNUNET_CONTAINER_multihashmap_iterate (in_map,
+ &free_rii,
+ NULL);
+ process_credits (wa->next);
return GNUNET_OK;
}
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@@ -1235,29 +1329,29 @@ history_credit_cb (void *cls,
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Failed to find wire transfer at `%s' in exchange database.
Audit ends at this point in time.\n",
GNUNET_STRINGS_absolute_time_to_string
(details->execution_date));
- hh = NULL;
- conclude_credit_history ();
+ wa->hh = NULL;
+ process_credits (wa->next);
return GNUNET_SYSERR; /* not an error, just end of processing */
}
/* Update offset */
- if (NULL == in_wire_off)
+ if (NULL == wa->in_wire_off)
{
- wire_off_size = row_off_size;
- in_wire_off = GNUNET_malloc (row_off_size);
+ wa->wire_off_size = row_off_size;
+ wa->in_wire_off = GNUNET_malloc (row_off_size);
}
- if (wire_off_size != row_off_size)
+ if (wa->wire_off_size != row_off_size)
{
GNUNET_break (0);
commit (GNUNET_DB_STATUS_HARD_ERROR);
GNUNET_SCHEDULER_shutdown ();
- hh = NULL;
return GNUNET_SYSERR;
}
- memcpy (in_wire_off,
+ memcpy (wa->in_wire_off,
row_off,
row_off_size);
+
/* compare records with expected data */
if (row_off_size != rii->row_off_size)
{
@@ -1384,42 +1478,83 @@ history_credit_cb (void *cls,
* Start processing the next wire account.
* Shuts down if we are done.
*
- * @param cls NULL
+ * @param cls `struct WireAccount` with a wire account list to process
*/
static void
-process_next_account (void *cls)
+process_credits (void *cls)
{
- struct WireAccount *wa;
+ struct WireAccount *wa = cls;
+ struct TALER_WIRE_Plugin *wp;
enum GNUNET_DB_QueryStatus qs;
- int ret;
-
- (void) cls;
- // FIXME: this logic is broken at a high level,
- // as it iterates over the exchange's incoming
- // transactions once PER bank account, so for
- // multiple bank accounts this cannot work!
- if (NULL == (wa = wa_head))
+
+ if (NULL == wa)
{
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Finished with all accounts, shutting down\n");
- if (NULL != wp)
- commit (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT);
+ /* done with all accounts, conclude check */
+ conclude_credit_history ();
+ return;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Analyzing exchange's wire IN table for account `%s'\n",
+ wa->section_name);
+ qs = edb->select_reserves_in_above_serial_id_by_account (edb->cls,
+ esession,
+ wa->section_name,
+
wa->pp.last_reserve_in_serial_id,
+ &reserve_in_cb,
+ wa);
+ if (0 > qs)
+ {
+ GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+ global_ret = 1;
GNUNET_SCHEDULER_shutdown ();
return;
}
- GNUNET_CONTAINER_DLL_remove (wa_head,
- wa_tail,
- wa);
+
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Starting audit of account `%s'\n",
+ "Starting bank CREDIT history of account `%s'\n",
wa->section_name);
- /* setup globals */
- if (NULL != wp)
- TALER_WIRE_plugin_unload (wp);
wp = wa->wire_plugin;
- GNUNET_free_non_null (wp_section_name);
- wp_section_name = wa->section_name;
- GNUNET_free (wa);
+ wa->hh = wp->get_history (wp->cls,
+ wa->section_name,
+ TALER_BANK_DIRECTION_CREDIT,
+ wa->in_wire_off,
+ wa->wire_off_size,
+ INT64_MAX,
+ &history_credit_cb,
+ wa);
+ if (NULL == wa->hh)
+ {
+ fprintf (stderr,
+ "Failed to obtain bank transaction history\n");
+ commit (GNUNET_DB_STATUS_HARD_ERROR);
+ global_ret = 1;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+}
+
+
+/**
+ * Begin audit of CREDITs to the exchange.
+ */
+static void
+begin_credit_audit ()
+{
+ in_map = GNUNET_CONTAINER_multihashmap_create (1024,
+ GNUNET_YES);
+ /* now go over all bank accounts and check delta with in_map */
+ process_credits (wa_head);
+}
+
+
+/**
+ * Start the database transactions and begin the audit.
+ */
+static void
+begin_transaction ()
+{
+ enum GNUNET_DB_QueryStatus qsx;
+ int ret;
ret = adb->start (adb->cls,
asession);
@@ -1442,14 +1577,30 @@ process_next_account (void *cls)
GNUNET_SCHEDULER_shutdown ();
return;
}
+ for (struct WireAccount *wa = wa_head;
+ NULL != wa;
+ wa = wa->next)
+ {
+ qsx = adb->get_wire_auditor_account_progress (adb->cls,
+ asession,
+ &master_pub,
+ wa->section_name,
+ &wa->pp,
+ &wa->in_wire_off,
+ &wa->out_wire_off,
+ &wa->wire_off_size);
+ if (0 > qsx)
+ {
+ GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qsx);
+ global_ret = 1;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ }
qsx = adb->get_wire_auditor_progress (adb->cls,
asession,
&master_pub,
- wp_section_name,
- &pp,
- &in_wire_off,
- &out_wire_off,
- &wire_off_size);
+ &pp);
if (0 > qsx)
{
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qsx);
@@ -1465,54 +1616,10 @@ process_next_account (void *cls)
else
{
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- _("Resuming audit at %llu/%llu\n"),
- (unsigned long long) pp.last_reserve_in_serial_id,
- (unsigned long long) pp.last_wire_out_serial_id);
- }
-
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Analyzing exchange's wire IN table\n");
- in_map = GNUNET_CONTAINER_multihashmap_create (1024,
- GNUNET_YES);
- qs = edb->select_reserves_in_above_serial_id_by_account (edb->cls,
- esession,
- wp_section_name,
-
pp.last_reserve_in_serial_id,
- &reserve_in_cb,
- NULL);
- if (0 > qs)
- {
- GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
- global_ret = 1;
- GNUNET_SCHEDULER_shutdown ();
- return;
- }
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE,
- "No new incoming transactions available, skipping CREDIT
phase\n");
- process_debits ();
- return;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Checking bank CREDIT records\n");
- hh = wp->get_history (wp->cls,
- wp_section_name,
- TALER_BANK_DIRECTION_CREDIT,
- in_wire_off,
- wire_off_size,
- INT64_MAX,
- &history_credit_cb,
- NULL);
- if (NULL == hh)
- {
- fprintf (stderr,
- "Failed to obtain bank transaction history\n");
- commit (GNUNET_DB_STATUS_HARD_ERROR);
- global_ret = 1;
- GNUNET_SCHEDULER_shutdown ();
- return;
+ "Resuming audit at %s\n",
+ GNUNET_STRINGS_absolute_time_to_string (pp.last_timestamp));
}
+ begin_credit_audit ();
}
@@ -1568,10 +1675,36 @@ run (void *cls,
const struct GNUNET_CONFIGURATION_Handle *c)
{
static const struct TALER_MasterPublicKeyP zeromp;
+ char *tinys;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Launching auditor\n");
cfg = c;
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_string (cfg,
+ "auditor",
+ "TINY_AMOUNT",
+ &tinys))
+ {
+ GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
+ "auditor",
+ "TINY_AMOUNT");
+ global_ret = 1;
+ return;
+ }
+ if (GNUNET_OK !=
+ TALER_string_to_amount (tinys,
+ &tiny_amount))
+ {
+ GNUNET_free (tinys);
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "auditor",
+ "TINY_AMOUNT",
+ "invalid amount");
+ global_ret = 1;
+ return;
+ }
+ GNUNET_free (tinys);
if (0 == GNUNET_memcmp (&zeromp,
&master_pub))
{
@@ -1716,7 +1849,7 @@ run (void *cls,
TALER_EXCHANGEDB_find_accounts (cfg,
&process_account_cb,
NULL);
- process_next_account (NULL);
+ begin_transaction ();
}
@@ -1755,12 +1888,12 @@ main (int argc,
NULL));
if (GNUNET_OK !=
GNUNET_PROGRAM_run (argc,
- argv,
+ argv,
"taler-wire-auditor",
- "Audit exchange database for consistency with the
bank's wire transfers",
- options,
- &run,
- NULL))
+ "Audit exchange database for consistency with the
bank's wire transfers",
+ options,
+ &run,
+ NULL))
return 1;
return global_ret;
}
diff --git a/src/auditor/test-auditor.conf b/src/auditor/test-auditor.conf
index 781cbbce..420b68ba 100644
--- a/src/auditor/test-auditor.conf
+++ b/src/auditor/test-auditor.conf
@@ -1,5 +1,6 @@
[auditor]
DB = postgres
+TINY_AMOUNT = TESTKUDOS:0.01
[auditordb-postgres]
CONFIG = postgres:///taler-auditor-test
diff --git a/src/auditordb/plugin_auditordb_postgres.c
b/src/auditordb/plugin_auditordb_postgres.c
index dfd8d64e..f1eaa6f2 100644
--- a/src/auditordb/plugin_auditordb_postgres.c
+++ b/src/auditordb/plugin_auditordb_postgres.c
@@ -177,6 +177,7 @@ postgres_drop_tables (void *cls,
GNUNET_PQ_make_execute ("DROP TABLE IF EXISTS
auditor_progress_deposit_confirmation;"),
GNUNET_PQ_make_execute ("DROP TABLE IF EXISTS auditor_progress_coin;"),
GNUNET_PQ_make_execute ("DROP TABLE IF EXISTS wire_auditor_progress;"),
+ GNUNET_PQ_make_execute ("DROP TABLE IF EXISTS
wire_auditor_account_progress;"),
GNUNET_PQ_make_execute ("DROP TABLE IF EXISTS deposit_confirmations
CASCADE;"),
GNUNET_PQ_EXECUTE_STATEMENT_END
};
@@ -283,15 +284,18 @@ postgres_create_tables (void *cls)
",last_payback_serial_id INT8 NOT NULL DEFAULT 0"
",last_payback_refresh_serial_id INT8 NOT NULL
DEFAULT 0"
")"),
- GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS wire_auditor_progress"
+ GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS
wire_auditor_account_progress"
"(master_pub BYTEA CONSTRAINT master_pub_ref
REFERENCES auditor_exchanges(master_pub) ON DELETE CASCADE"
",account_name TEXT NOT NULL"
",last_wire_reserve_in_serial_id INT8 NOT NULL
DEFAULT 0"
",last_wire_wire_out_serial_id INT8 NOT NULL
DEFAULT 0"
- ",last_timestamp INT8 NOT NULL"
",wire_in_off BYTEA"
",wire_out_off BYTEA"
")"),
+ GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS wire_auditor_progress"
+ "(master_pub BYTEA CONSTRAINT master_pub_ref
REFERENCES auditor_exchanges(master_pub) ON DELETE CASCADE"
+ ",last_timestamp INT8 NOT NULL"
+ ")"),
/* Table with all of the customer reserves and their respective
balances that the auditor is aware of.
"last_reserve_out_serial_id" marks the last withdrawal from
@@ -686,39 +690,56 @@ postgres_prepare (PGconn *db_conn)
",last_payback_refresh_serial_id"
") VALUES ($1,$2,$3,$4,$5,$6,$7);",
7),
- /* Used in #postgres_insert_wire_auditor_progress() */
- GNUNET_PQ_make_prepare ("wire_auditor_progress_insert",
- "INSERT INTO wire_auditor_progress "
+ /* Used in #postgres_insert_wire_auditor_account_progress() */
+ GNUNET_PQ_make_prepare ("wire_auditor_account_progress_insert",
+ "INSERT INTO wire_auditor_account_progress "
"(master_pub"
",account_name"
",last_wire_reserve_in_serial_id"
",last_wire_wire_out_serial_id"
- ",last_timestamp"
",wire_in_off"
",wire_out_off"
- ") VALUES ($1,$2,$3,$4,$5,$6,$7);",
- 7),
- /* Used in #postgres_update_wire_auditor_progress() */
- GNUNET_PQ_make_prepare ("wire_auditor_progress_update",
- "UPDATE wire_auditor_progress SET "
+ ") VALUES ($1,$2,$3,$4,$5,$6);",
+ 6),
+ /* Used in #postgres_update_wire_auditor_account_progress() */
+ GNUNET_PQ_make_prepare ("wire_auditor_account_progress_update",
+ "UPDATE wire_auditor_account_progress SET "
" last_wire_reserve_in_serial_id=$1"
",last_wire_wire_out_serial_id=$2"
- ",last_timestamp=$3"
- ",wire_in_off=$4"
- ",wire_out_off=$5"
- " WHERE master_pub=$6 AND account_name=$7",
- 7),
- /* Used in #postgres_get_wire_auditor_progress() */
- GNUNET_PQ_make_prepare ("wire_auditor_progress_select",
+ ",wire_in_off=$3"
+ ",wire_out_off=$4"
+ " WHERE master_pub=$5 AND account_name=$6",
+ 6),
+ /* Used in #postgres_get_wire_auditor_account_progress() */
+ GNUNET_PQ_make_prepare ("wire_auditor_account_progress_select",
"SELECT"
" last_wire_reserve_in_serial_id"
",last_wire_wire_out_serial_id"
- ",last_timestamp"
",wire_in_off"
",wire_out_off"
- " FROM wire_auditor_progress"
+ " FROM wire_auditor_account_progress"
" WHERE master_pub=$1 AND account_name=$2;",
2),
+ /* Used in #postgres_insert_wire_auditor_progress() */
+ GNUNET_PQ_make_prepare ("wire_auditor_progress_insert",
+ "INSERT INTO wire_auditor_progress "
+ "(master_pub"
+ ",last_timestamp"
+ ") VALUES ($1,$2);",
+ 2),
+ /* Used in #postgres_update_wire_auditor_progress() */
+ GNUNET_PQ_make_prepare ("wire_auditor_progress_update",
+ "UPDATE wire_auditor_progress SET "
+ " last_timestamp=$1"
+ " WHERE master_pub=$2",
+ 2),
+ /* Used in #postgres_get_wire_auditor_progress() */
+ GNUNET_PQ_make_prepare ("wire_auditor_progress_select",
+ "SELECT"
+ " last_timestamp"
+ " FROM wire_auditor_progress"
+ " WHERE master_pub=$1;",
+ 1),
/* Used in #postgres_insert_reserve_info() */
GNUNET_PQ_make_prepare ("auditor_reserves_insert",
"INSERT INTO auditor_reserves "
@@ -2105,25 +2126,23 @@ postgres_get_auditor_progress_coin (void *cls,
* @param session connection to use
* @param master_pub master key of the exchange
* @param account_name name of the wire account we are auditing
- * @param pp where is the auditor in processing
* @return transaction status code
*/
static enum GNUNET_DB_QueryStatus
-postgres_insert_wire_auditor_progress (void *cls,
- struct TALER_AUDITORDB_Session *session,
- const struct TALER_MasterPublicKeyP
*master_pub,
- const char *account_name,
- const struct
TALER_AUDITORDB_WireProgressPoint *pp,
- const void *in_wire_off,
- const void *out_wire_off,
- size_t wire_off_size)
+postgres_insert_wire_auditor_account_progress (void *cls,
+ struct TALER_AUDITORDB_Session
*session,
+ const struct
TALER_MasterPublicKeyP *master_pub,
+ const char *account_name,
+ const struct
TALER_AUDITORDB_WireAccountProgressPoint *pp,
+ const void *in_wire_off,
+ const void *out_wire_off,
+ size_t wire_off_size)
{
struct GNUNET_PQ_QueryParam params[] = {
GNUNET_PQ_query_param_auto_from_type (master_pub),
GNUNET_PQ_query_param_string (account_name),
GNUNET_PQ_query_param_uint64 (&pp->last_reserve_in_serial_id),
GNUNET_PQ_query_param_uint64 (&pp->last_wire_out_serial_id),
- TALER_PQ_query_param_absolute_time (&pp->last_timestamp),
GNUNET_PQ_query_param_fixed_size (in_wire_off,
wire_off_size),
GNUNET_PQ_query_param_fixed_size (out_wire_off,
@@ -2132,7 +2151,7 @@ postgres_insert_wire_auditor_progress (void *cls,
};
return GNUNET_PQ_eval_prepared_non_select (session->conn,
- "wire_auditor_progress_insert",
+
"wire_auditor_account_progress_insert",
params);
}
@@ -2145,23 +2164,21 @@ postgres_insert_wire_auditor_progress (void *cls,
* @param session connection to use
* @param master_pub master key of the exchange
* @param account_name name of the wire account we are auditing
- * @param pp where is the auditor in processing
* @return transaction status code
*/
static enum GNUNET_DB_QueryStatus
-postgres_update_wire_auditor_progress (void *cls,
- struct TALER_AUDITORDB_Session *session,
- const struct TALER_MasterPublicKeyP
*master_pub,
- const char *account_name,
- const struct
TALER_AUDITORDB_WireProgressPoint *pp,
- const void *in_wire_off,
- const void *out_wire_off,
- size_t wire_off_size)
+postgres_update_wire_auditor_account_progress (void *cls,
+ struct TALER_AUDITORDB_Session
*session,
+ const struct
TALER_MasterPublicKeyP *master_pub,
+ const char *account_name,
+ const struct
TALER_AUDITORDB_WireAccountProgressPoint *pp,
+ const void *in_wire_off,
+ const void *out_wire_off,
+ size_t wire_off_size)
{
struct GNUNET_PQ_QueryParam params[] = {
GNUNET_PQ_query_param_uint64 (&pp->last_reserve_in_serial_id),
GNUNET_PQ_query_param_uint64 (&pp->last_wire_out_serial_id),
- TALER_PQ_query_param_absolute_time (&pp->last_timestamp),
GNUNET_PQ_query_param_fixed_size (in_wire_off,
wire_off_size),
GNUNET_PQ_query_param_fixed_size (out_wire_off,
@@ -2172,7 +2189,7 @@ postgres_update_wire_auditor_progress (void *cls,
};
return GNUNET_PQ_eval_prepared_non_select (session->conn,
- "wire_auditor_progress_update",
+
"wire_auditor_account_progress_update",
params);
}
@@ -2184,18 +2201,21 @@ postgres_update_wire_auditor_progress (void *cls,
* @param session connection to use
* @param master_pub master key of the exchange
* @param account_name name of the wire account we are auditing
- * @param[out] pp set to where the auditor is in processing
+ * @param[out] pp where is the auditor in processing
+ * @param[out] in_wire_off how far are we in the incoming wire transaction
history
+ * @param[out] out_wire_off how far are we in the outgoing wire transaction
history
+ * @param[out] wire_off_size how many bytes do @a in_wire_off and @a
out_wire_off take?
* @return transaction status code
*/
static enum GNUNET_DB_QueryStatus
-postgres_get_wire_auditor_progress (void *cls,
- struct TALER_AUDITORDB_Session *session,
- const struct TALER_MasterPublicKeyP
*master_pub,
- const char *account_name,
- struct TALER_AUDITORDB_WireProgressPoint
*pp,
- void **in_wire_off,
- void **out_wire_off,
- size_t *wire_off_size)
+postgres_get_wire_auditor_account_progress (void *cls,
+ struct TALER_AUDITORDB_Session
*session,
+ const struct
TALER_MasterPublicKeyP *master_pub,
+ const char *account_name,
+ struct
TALER_AUDITORDB_WireAccountProgressPoint *pp,
+ void **in_wire_off,
+ void **out_wire_off,
+ size_t *wire_off_size)
{
size_t xsize;
enum GNUNET_DB_QueryStatus qs;
@@ -2209,8 +2229,6 @@ postgres_get_wire_auditor_progress (void *cls,
&pp->last_reserve_in_serial_id),
GNUNET_PQ_result_spec_uint64 ("last_wire_wire_out_serial_id",
&pp->last_wire_out_serial_id),
- TALER_PQ_result_spec_absolute_time ("last_timestamp",
- &pp->last_timestamp),
GNUNET_PQ_result_spec_variable_size ("wire_in_off",
in_wire_off,
wire_off_size),
@@ -2221,7 +2239,7 @@ postgres_get_wire_auditor_progress (void *cls,
};
qs = GNUNET_PQ_eval_prepared_singleton_select (session->conn,
-
"wire_auditor_progress_select",
+
"wire_auditor_account_progress_select",
params,
rs);
if (qs <= 0)
@@ -2235,6 +2253,94 @@ postgres_get_wire_auditor_progress (void *cls,
/**
+ * Insert information about the auditor's progress with an exchange's
+ * data.
+ *
+ * @param cls the @e cls of this struct with the plugin-specific state
+ * @param session connection to use
+ * @param master_pub master key of the exchange
+ * @param pp where is the auditor in processing
+ * @return transaction status code
+ */
+static enum GNUNET_DB_QueryStatus
+postgres_insert_wire_auditor_progress (void *cls,
+ struct TALER_AUDITORDB_Session *session,
+ const struct TALER_MasterPublicKeyP
*master_pub,
+ const struct
TALER_AUDITORDB_WireProgressPoint *pp)
+{
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_auto_from_type (master_pub),
+ TALER_PQ_query_param_absolute_time (&pp->last_timestamp),
+ GNUNET_PQ_query_param_end
+ };
+
+ return GNUNET_PQ_eval_prepared_non_select (session->conn,
+ "wire_auditor_progress_insert",
+ params);
+}
+
+
+/**
+ * Update information about the progress of the auditor. There
+ * must be an existing record for the exchange.
+ *
+ * @param cls the @e cls of this struct with the plugin-specific state
+ * @param session connection to use
+ * @param master_pub master key of the exchange
+ * @param pp where is the auditor in processing
+ * @return transaction status code
+ */
+static enum GNUNET_DB_QueryStatus
+postgres_update_wire_auditor_progress (void *cls,
+ struct TALER_AUDITORDB_Session *session,
+ const struct TALER_MasterPublicKeyP
*master_pub,
+ const struct
TALER_AUDITORDB_WireProgressPoint *pp)
+{
+ struct GNUNET_PQ_QueryParam params[] = {
+ TALER_PQ_query_param_absolute_time (&pp->last_timestamp),
+ GNUNET_PQ_query_param_auto_from_type (master_pub),
+ GNUNET_PQ_query_param_end
+ };
+
+ return GNUNET_PQ_eval_prepared_non_select (session->conn,
+ "wire_auditor_progress_update",
+ params);
+}
+
+
+/**
+ * Get information about the progress of the auditor.
+ *
+ * @param cls the @e cls of this struct with the plugin-specific state
+ * @param session connection to use
+ * @param master_pub master key of the exchange
+ * @param[out] pp set to where the auditor is in processing
+ * @return transaction status code
+ */
+static enum GNUNET_DB_QueryStatus
+postgres_get_wire_auditor_progress (void *cls,
+ struct TALER_AUDITORDB_Session *session,
+ const struct TALER_MasterPublicKeyP
*master_pub,
+ struct TALER_AUDITORDB_WireProgressPoint
*pp)
+{
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_auto_from_type (master_pub),
+ GNUNET_PQ_query_param_end
+ };
+ struct GNUNET_PQ_ResultSpec rs[] = {
+ TALER_PQ_result_spec_absolute_time ("last_timestamp",
+ &pp->last_timestamp),
+ GNUNET_PQ_result_spec_end
+ };
+
+ return GNUNET_PQ_eval_prepared_singleton_select (session->conn,
+
"wire_auditor_progress_select",
+ params,
+ rs);
+}
+
+
+/**
* Insert information about a reserve. There must not be an
* existing record for the reserve.
*
@@ -3329,6 +3435,9 @@ libtaler_plugin_auditordb_postgres_init (void *cls)
plugin->update_auditor_progress_coin =
&postgres_update_auditor_progress_coin;
plugin->insert_auditor_progress_coin =
&postgres_insert_auditor_progress_coin;
+ plugin->get_wire_auditor_account_progress =
&postgres_get_wire_auditor_account_progress;
+ plugin->update_wire_auditor_account_progress =
&postgres_update_wire_auditor_account_progress;
+ plugin->insert_wire_auditor_account_progress =
&postgres_insert_wire_auditor_account_progress;
plugin->get_wire_auditor_progress = &postgres_get_wire_auditor_progress;
plugin->update_wire_auditor_progress =
&postgres_update_wire_auditor_progress;
plugin->insert_wire_auditor_progress =
&postgres_insert_wire_auditor_progress;
diff --git a/src/include/taler_auditordb_plugin.h
b/src/include/taler_auditordb_plugin.h
index dab548e6..98fce62d 100644
--- a/src/include/taler_auditordb_plugin.h
+++ b/src/include/taler_auditordb_plugin.h
@@ -108,6 +108,21 @@ typedef int
*/
struct TALER_AUDITORDB_WireProgressPoint
{
+
+ /**
+ * Time until which we have confirmed that all wire transactions
+ * that the exchange should do, have indeed been done.
+ */
+ struct GNUNET_TIME_Absolute last_timestamp;
+};
+
+
+/**
+ * Structure for remembering the wire auditor's progress over the
+ * various tables and (auditor) transactions per wire account.
+ */
+struct TALER_AUDITORDB_WireAccountProgressPoint
+{
/**
* serial ID of the last reserve_in transfer the wire auditor processed
*/
@@ -118,11 +133,6 @@ struct TALER_AUDITORDB_WireProgressPoint
*/
uint64_t last_wire_out_serial_id;
- /**
- * Time until which we have confirmed that all wire transactions
- * that the exchange should do, have indeed been done.
- */
- struct GNUNET_TIME_Absolute last_timestamp;
};
@@ -801,14 +811,14 @@ struct TALER_AUDITORDB_Plugin
* @return transaction status code
*/
enum GNUNET_DB_QueryStatus
- (*insert_wire_auditor_progress)(void *cls,
- struct TALER_AUDITORDB_Session *session,
- const struct TALER_MasterPublicKeyP
*master_pub,
- const char *account_name,
- const struct
TALER_AUDITORDB_WireProgressPoint *pp,
- const void *in_wire_off,
- const void *out_wire_off,
- size_t wire_off_size);
+ (*insert_wire_auditor_account_progress)(void *cls,
+ struct TALER_AUDITORDB_Session
*session,
+ const struct TALER_MasterPublicKeyP
*master_pub,
+ const char *account_name,
+ const struct
TALER_AUDITORDB_WireAccountProgressPoint *pp,
+ const void *in_wire_off,
+ const void *out_wire_off,
+ size_t wire_off_size);
/**
@@ -826,16 +836,75 @@ struct TALER_AUDITORDB_Plugin
* @return transaction status code
*/
enum GNUNET_DB_QueryStatus
- (*update_wire_auditor_progress)(void *cls,
+ (*update_wire_auditor_account_progress)(void *cls,
+ struct TALER_AUDITORDB_Session
*session,
+ const struct TALER_MasterPublicKeyP
*master_pub,
+ const char *account_name,
+ const struct
TALER_AUDITORDB_WireAccountProgressPoint *pp,
+ const void *in_wire_off,
+ const void *out_wire_off,
+ size_t wire_off_size);
+
+
+ /**
+ * Get information about the progress of the wire auditor.
+ *
+ * @param cls the @e cls of this struct with the plugin-specific state
+ * @param session connection to use
+ * @param master_pub master key of the exchange
+ * @param account_name name of the wire account we are auditing
+ * @param[out] pp where is the auditor in processing
+ * @param[out] in_wire_off how far are we in the incoming wire transaction
history
+ * @param[out] out_wire_off how far are we in the outgoing wire transaction
history
+ * @param[out] wire_off_size how many bytes do @a in_wire_off and @a
out_wire_off take?
+ * @return transaction status code
+ */
+ enum GNUNET_DB_QueryStatus
+ (*get_wire_auditor_account_progress)(void *cls,
+ struct TALER_AUDITORDB_Session *session,
+ const struct TALER_MasterPublicKeyP
*master_pub,
+ const char *account_name,
+ struct
TALER_AUDITORDB_WireAccountProgressPoint *pp,
+ void **in_wire_off,
+ void **out_wire_off,
+ size_t *wire_off_size);
+
+
+ /**
+ * Insert information about the wire auditor's progress with an exchange's
+ * data.
+ *
+ * @param cls the @e cls of this struct with the plugin-specific state
+ * @param session connection to use
+ * @param master_pub master key of the exchange
+ * @param account_name name of the wire account we are auditing
+ * @param pp where is the auditor in processing
+ * @return transaction status code
+ */
+ enum GNUNET_DB_QueryStatus
+ (*insert_wire_auditor_progress)(void *cls,
struct TALER_AUDITORDB_Session *session,
const struct TALER_MasterPublicKeyP
*master_pub,
- const char *account_name,
- const struct
TALER_AUDITORDB_WireProgressPoint *pp,
- const void *in_wire_off,
- const void *out_wire_off,
- size_t wire_off_size);
+ const struct
TALER_AUDITORDB_WireProgressPoint *pp);
+ /**
+ * Update information about the progress of the wire auditor. There
+ * must be an existing record for the exchange.
+ *
+ * @param cls the @e cls of this struct with the plugin-specific state
+ * @param session connection to use
+ * @param master_pub master key of the exchange
+ * @param account_name name of the wire account we are auditing
+ * @param pp where is the auditor in processing
+ * @return transaction status code
+ */
+ enum GNUNET_DB_QueryStatus
+ (*update_wire_auditor_progress)(void *cls,
+ struct TALER_AUDITORDB_Session *session,
+ const struct TALER_MasterPublicKeyP
*master_pub,
+ const struct
TALER_AUDITORDB_WireProgressPoint *pp);
+
/**
* Get information about the progress of the wire auditor.
@@ -845,20 +914,13 @@ struct TALER_AUDITORDB_Plugin
* @param master_pub master key of the exchange
* @param account_name name of the wire account we are auditing
* @param[out] pp set to where the auditor is in processing
- * @param[out] in_wire_off how far are we in the incoming wire transaction
history
- * @param[out] out_wire_off how far are we in the outgoing wire transaction
history
- * @param[out] wire_off_size how many bytes do @a in_wire_off and @a
out_wire_off take?
* @return transaction status code
*/
enum GNUNET_DB_QueryStatus
(*get_wire_auditor_progress)(void *cls,
struct TALER_AUDITORDB_Session *session,
const struct TALER_MasterPublicKeyP *master_pub,
- const char *account_name,
- struct TALER_AUDITORDB_WireProgressPoint *pp,
- void **in_wire_off,
- void **out_wire_off,
- size_t *wire_off_size);
+ struct TALER_AUDITORDB_WireProgressPoint *pp);
/**
--
To stop receiving notification emails like this one, please contact
address@hidden.
| [Prev in Thread] |
Current Thread |
[Next in Thread] |
- [GNUnet-SVN] [taler-exchange] branch master updated: refactor wire auditor to properly handle multiple accounts,
gnunet <=