new aggregator mega transaction logic

3 files changed, 159 insertions, 505 deletions

diff --git a/src/exchange/taler-exchange-aggregator.c b/src/exchange/taler-exchange-aggregator.c
index c34d47f9c..04cf426de 100644
--- a/src/exchange/taler-exchange-aggregator.c
+++ b/src/exchange/taler-exchange-aggregator.c
@@ -28,18 +28,6 @@
 #include "taler_json_lib.h"
 #include "taler_bank_service.h"
 
-struct AdditionalDeposit
-{
-  /**
-   * Public key of the coin.
-   */
-  struct TALER_CoinSpendPublicKeyP coin_pub;
-
-  /**
-   * Row of the deposit.
-   */
-  uint64_t row;
-};
 
 /**
  * Information about one aggregation process to be executed.  There is
@@ -55,11 +43,6 @@ struct AggregationUnit
   struct TALER_MerchantPublicKeyP merchant_pub;
 
   /**
-   * Public key of the coin.
-   */
-  struct TALER_CoinSpendPublicKeyP coin_pub;
-
-  /**
    * Total amount to be transferred, before subtraction of @e fees.wire and rounding down.
    */
   struct TALER_Amount total_amount;
@@ -80,11 +63,6 @@ struct AggregationUnit
   struct TALER_WireTransferIdentifierRawP wtid;
 
   /**
-   * Row ID of the transaction that started it all.
-   */
-  uint64_t row_id;
-
-  /**
    * The current time (which triggered the aggregation and
    * defines the wire fee).
    */
@@ -101,32 +79,11 @@ struct AggregationUnit
   struct TALER_PaytoHashP h_payto;
 
   /**
-   * Serial number of the wire target.
-   */
-  uint64_t wire_target;
-
-  /**
    * Exchange wire account to be used for the preparation and
    * eventual execution of the aggregate wire transfer.
    */
   const struct TALER_EXCHANGEDB_AccountInfo *wa;
 
-  /**
-   * Array of row_ids from the aggregation.
-   */
-  struct AdditionalDeposit
-    additional_rows[TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT];
-
-  /**
-   * Offset specifying how many @e additional_rows are in use.
-   */
-  unsigned int rows_offset;
-
-  /**
-   * Set to true if we encountered a refund during #refund_by_coin_cb.
-   * Used to wave the deposit fee.
-   */
-  bool have_refund;
 };
 
 
@@ -341,331 +298,6 @@ parse_wirewatch_config (void)
 
 
 /**
- * Callback invoked with information about refunds applicable
- * to a particular coin.  Subtract refunded amount(s) from
- * the aggregation unit's total amount.
- *
- * @param cls closure with a `struct AggregationUnit *`
- * @param amount_with_fee what was the refunded amount with the fee
- * @return #GNUNET_OK to continue to iterate, #GNUNET_SYSERR to stop
- */
-static enum GNUNET_GenericReturnValue
-refund_by_coin_cb (void *cls,
-                   const struct TALER_Amount *amount_with_fee)
-{
-  struct AggregationUnit *aux = cls;
-
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Aggregator subtracts applicable refund of amount %s\n",
-              TALER_amount2s (amount_with_fee));
-  aux->have_refund = true;
-  if (0 >
-      TALER_amount_subtract (&aux->total_amount,
-                             &aux->total_amount,
-                             amount_with_fee))
-  {
-    GNUNET_break (0);
-    return GNUNET_SYSERR;
-  }
-  return GNUNET_OK;
-}
-
-
-/**
- * Function called with details about deposits that have been made,
- * with the goal of executing the corresponding wire transaction.
- *
- * @param cls a `struct AggregationUnit`
- * @param row_id identifies database entry
- * @param merchant_pub public key of the merchant
- * @param coin_pub public key of the coin
- * @param amount_with_fee amount that was deposited including fee
- * @param deposit_fee amount the exchange gets to keep as transaction fees
- * @param h_contract_terms hash of the proposal data known to merchant and customer
- * @param wire_target target account for the wire transfer
- * @param payto_uri URI of the target account
- * @return transaction status code,  #GNUNET_DB_STATUS_SUCCESS_ONE_RESULT to continue to iterate
- */
-static enum GNUNET_DB_QueryStatus
-deposit_cb (void *cls,
-            uint64_t row_id,
-            const struct TALER_MerchantPublicKeyP *merchant_pub,
-            const struct TALER_CoinSpendPublicKeyP *coin_pub,
-            const struct TALER_Amount *amount_with_fee,
-            const struct TALER_Amount *deposit_fee,
-            const struct TALER_PrivateContractHashP *h_contract_terms,
-            uint64_t wire_target,
-            const char *payto_uri)
-{
-  struct AggregationUnit *au = cls;
-  enum GNUNET_DB_QueryStatus qs;
-
-  au->merchant_pub = *merchant_pub;
-  au->coin_pub = *coin_pub;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Aggregator processing payment %s with amount %s\n",
-              TALER_B2S (coin_pub),
-              TALER_amount2s (amount_with_fee));
-  au->row_id = row_id;
-  au->total_amount = *amount_with_fee;
-  au->have_refund = false;
-  qs = db_plugin->select_refunds_by_coin (db_plugin->cls,
-                                          coin_pub,
-                                          &au->merchant_pub,
-                                          h_contract_terms,
-                                          &refund_by_coin_cb,
-                                          au);
-  if (0 > qs)
-  {
-    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-    return qs;
-  }
-  if (! au->have_refund)
-  {
-    struct TALER_Amount ntotal;
-
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Non-refunded transaction, subtracting deposit fee %s\n",
-                TALER_amount2s (deposit_fee));
-    if (0 >
-        TALER_amount_subtract (&ntotal,
-                               amount_with_fee,
-                               deposit_fee))
-    {
-      /* This should never happen, issue a warning, but continue processing
-         with an amount of zero, least we hang here for good. */
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Fatally malformed record at row %llu over %s (deposit fee exceeds deposited value)\n",
-                  (unsigned long long) row_id,
-                  TALER_amount2s (amount_with_fee));
-      GNUNET_assert (GNUNET_OK ==
-                     TALER_amount_set_zero (au->total_amount.currency,
-                                            &au->total_amount));
-    }
-    else
-    {
-      au->total_amount = ntotal;
-    }
-  }
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Amount after fee is %s\n",
-              TALER_amount2s (&au->total_amount));
-
-  GNUNET_assert (NULL == au->payto_uri);
-  au->payto_uri = GNUNET_strdup (payto_uri);
-  TALER_payto_hash (payto_uri,
-                    &au->h_payto);
-  au->wire_target = wire_target;
-  GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
-                              &au->wtid,
-                              sizeof (au->wtid));
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Starting aggregation under H(WTID)=%s, starting amount %s at %llu\n",
-              TALER_B2S (&au->wtid),
-              TALER_amount2s (amount_with_fee),
-              (unsigned long long) row_id);
-  au->wa = TALER_EXCHANGEDB_find_account_by_payto_uri (payto_uri);
-  if (NULL == au->wa)
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "No exchange account configured for `%s', please fix your setup to continue!\n",
-                payto_uri);
-    return GNUNET_DB_STATUS_HARD_ERROR;
-  }
-
-  /* make sure we have current fees */
-  au->execution_time = GNUNET_TIME_timestamp_get ();
-  {
-    struct GNUNET_TIME_Timestamp start_date;
-    struct GNUNET_TIME_Timestamp end_date;
-    struct TALER_MasterSignatureP master_sig;
-    enum GNUNET_DB_QueryStatus qs;
-
-    qs = db_plugin->get_wire_fee (db_plugin->cls,
-                                  au->wa->method,
-                                  au->execution_time,
-                                  &start_date,
-                                  &end_date,
-                                  &au->fees,
-                                  &master_sig);
-    if (0 >= qs)
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Could not get wire fees for %s at %s. Aborting run.\n",
-                  au->wa->method,
-                  GNUNET_TIME_timestamp2s (au->execution_time));
-      return GNUNET_DB_STATUS_HARD_ERROR;
-    }
-  }
-
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Aggregator starts aggregation for deposit %llu to %s with wire fee %s\n",
-              (unsigned long long) row_id,
-              TALER_B2S (&au->wtid),
-              TALER_amount2s (&au->fees.wire));
-  qs = db_plugin->insert_aggregation_tracking (db_plugin->cls,
-                                               &au->wtid,
-                                               row_id);
-  if (qs <= 0)
-  {
-    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-    return qs;
-  }
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Aggregator marks deposit %llu as done\n",
-              (unsigned long long) row_id);
-  qs = db_plugin->mark_deposit_done (db_plugin->cls,
-                                     coin_pub,
-                                     row_id);
-  if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
-  {
-    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-    return qs;
-  }
-  return qs;
-}
-
-
-/**
- * Function called with details about another deposit we
- * can aggregate into an existing aggregation unit.
- *
- * @param cls a `struct AggregationUnit`
- * @param row_id identifies database entry
- * @param coin_pub public key of the coin
- * @param amount_with_fee amount that was deposited including fee
- * @param deposit_fee amount the exchange gets to keep as transaction fees
- * @param h_contract_terms hash of the proposal data known to merchant and customer
- * @return transaction status code
- */
-static enum GNUNET_DB_QueryStatus
-aggregate_cb (void *cls,
-              uint64_t row_id,
-              const struct TALER_CoinSpendPublicKeyP *coin_pub,
-              const struct TALER_Amount *amount_with_fee,
-              const struct TALER_Amount *deposit_fee,
-              const struct TALER_PrivateContractHashP *h_contract_terms)
-{
-  struct AggregationUnit *au = cls;
-  struct TALER_Amount old;
-  enum GNUNET_DB_QueryStatus qs;
-
-  if (row_id == au->row_id)
-    return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
-  if (au->rows_offset >= TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT)
-  {
-    /* Bug: we asked for at most #TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT results! */
-    GNUNET_break (0);
-    /* Skip this one, but keep going with the overall transaction */
-    return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
-  }
-
-  /* add to total */
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Adding transaction amount %s from row %llu to aggregation\n",
-              TALER_amount2s (amount_with_fee),
-              (unsigned long long) row_id);
-  /* save the existing total aggregate in 'old', for later */
-  old = au->total_amount;
-  /* we begin with the total contribution of the current coin */
-  au->total_amount = *amount_with_fee;
-  /* compute contribution of this coin (after fees) */
-  au->have_refund = false;
-  qs = db_plugin->select_refunds_by_coin (db_plugin->cls,
-                                          coin_pub,
-                                          &au->merchant_pub,
-                                          h_contract_terms,
-                                          &refund_by_coin_cb,
-                                          au);
-  if (0 > qs)
-  {
-    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-    return qs;
-  }
-  if (! au->have_refund)
-  {
-    struct TALER_Amount tmp;
-
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Subtracting deposit fee %s for non-refunded coin\n",
-                TALER_amount2s (deposit_fee));
-    if (0 >
-        TALER_amount_subtract (&tmp,
-                               &au->total_amount,
-                               deposit_fee))
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Fatally malformed record at %llu over amount %s (deposit fee exceeds deposited value)\n",
-                  (unsigned long long) row_id,
-                  TALER_amount2s (&au->total_amount));
-      GNUNET_assert (GNUNET_OK ==
-                     TALER_amount_set_zero (old.currency,
-                                            &au->total_amount));
-    }
-    else
-    {
-      au->total_amount = tmp;
-    }
-  }
-
-  /* now add the au->total_amount with the (remaining) contribution of
-     the current coin to the 'old' value with the current aggregate value */
-  {
-    struct TALER_Amount tmp;
-
-    if (0 >
-        TALER_amount_add (&tmp,
-                          &au->total_amount,
-                          &old))
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Overflow or currency incompatibility during aggregation at %llu\n",
-                  (unsigned long long) row_id);
-      /* Skip this one, but keep going! */
-      au->total_amount = old;
-      return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
-    }
-    au->total_amount = tmp;
-  }
-
-  /* "append" to our list of rows */
-  au->additional_rows[au->rows_offset].coin_pub = *coin_pub;
-  au->additional_rows[au->rows_offset].row = row_id;
-  au->rows_offset++;
-  /* insert into aggregation tracking table */
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Adding %llu to aggregate %s\n",
-              (unsigned long long) row_id,
-              TALER_B2S (&au->wtid));
-  qs = db_plugin->insert_aggregation_tracking (db_plugin->cls,
-                                               &au->wtid,
-                                               row_id);
-  if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                "Failed to add %llu to aggregate %s: %d\n",
-                (unsigned long long) row_id,
-                TALER_B2S (&au->wtid),
-                qs);
-    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-    return qs;
-  }
-  qs = db_plugin->mark_deposit_done (db_plugin->cls,
-                                     coin_pub,
-                                     row_id);
-  if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
-  {
-    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
-    return qs;
-  }
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Aggregator marked deposit %llu as DONE\n",
-              (unsigned long long) row_id);
-  return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
-}
-
-
-/**
  * Perform a database commit. If it fails, print a warning.
  *
  * @return status of commit
@@ -727,10 +359,17 @@ run_aggregation (void *cls)
   struct Shard *s = cls;
   struct AggregationUnit au_active;
   enum GNUNET_DB_QueryStatus qs;
+  struct TALER_Amount trans;
+  bool have_transient;
 
   task = NULL;
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "Checking for ready deposits to aggregate\n");
+  /* make sure we have current fees */
+  memset (&au_active,
+          0,
+          sizeof (au_active));
+  au_active.execution_time = GNUNET_TIME_timestamp_get ();
   if (GNUNET_OK !=
       db_plugin->start_deferred_wire_out (db_plugin->cls))
   {
@@ -741,16 +380,13 @@ run_aggregation (void *cls)
     release_shard (s);
     return;
   }
-  memset (&au_active,
-          0,
-          sizeof (au_active));
   qs = db_plugin->get_ready_deposit (
     db_plugin->cls,
     s->shard_start,
     s->shard_end,
     kyc_off ? true : false,
-    &deposit_cb,
-    &au_active);
+    &au_active.merchant_pub,
+    &au_active.payto_uri);
   switch (qs)
   {
   case GNUNET_DB_STATUS_HARD_ERROR:
@@ -808,22 +444,98 @@ run_aggregation (void *cls)
     /* continued below */
     break;
   }
+  au_active.wa = TALER_EXCHANGEDB_find_account_by_payto_uri (
+    au_active.payto_uri);
+  if (NULL == au_active.wa)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "No exchange account configured for `%s', please fix your setup to continue!\n",
+                au_active.payto_uri);
+    global_ret = EXIT_FAILURE;
+    GNUNET_SCHEDULER_shutdown ();
+    release_shard (s);
+    return;
+  }
+
+  {
+    struct GNUNET_TIME_Timestamp start_date;
+    struct GNUNET_TIME_Timestamp end_date;
+    struct TALER_MasterSignatureP master_sig;
+
+    qs = db_plugin->get_wire_fee (db_plugin->cls,
+                                  au_active.wa->method,
+                                  au_active.execution_time,
+                                  &start_date,
+                                  &end_date,
+                                  &au_active.fees,
+                                  &master_sig);
+    if (0 >= qs)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Could not get wire fees for %s at %s. Aborting run.\n",
+                  au_active.wa->method,
+                  GNUNET_TIME_timestamp2s (au_active.execution_time));
+      global_ret = EXIT_FAILURE;
+      GNUNET_SCHEDULER_shutdown ();
+      release_shard (s);
+      return;
+    }
+  }
+
 
   /* Now try to find other deposits to aggregate */
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Found ready deposit for %s, aggregating by target %llu\n",
+              "Found ready deposit for %s, aggregating by target %s\n",
               TALER_B2S (&au_active.merchant_pub),
-              (unsigned long long) au_active.wire_target);
-  qs = db_plugin->iterate_matching_deposits (db_plugin->cls,
-                                             &au_active.h_payto,
-                                             &au_active.merchant_pub,
-                                             &aggregate_cb,
-                                             &au_active,
-                                             TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT);
+              au_active.payto_uri);
+  TALER_payto_hash (au_active.payto_uri,
+                    &au_active.h_payto);
+
+  qs = db_plugin->select_aggregation_transient (db_plugin->cls,
+                                                &au_active.h_payto,
+                                                au_active.wa->section_name,
+                                                &au_active.wtid,
+                                                &trans);
+  switch (qs)
+  {
+  case GNUNET_DB_STATUS_HARD_ERROR:
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "Failed to lookup transient aggregates!\n");
+    cleanup_au (&au_active);
+    db_plugin->rollback (db_plugin->cls);
+    global_ret = EXIT_FAILURE;
+    GNUNET_SCHEDULER_shutdown ();
+    release_shard (s);
+    return;
+  case GNUNET_DB_STATUS_SOFT_ERROR:
+    /* serializiability issue, try again */
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Serialization issue, trying again later!\n");
+    db_plugin->rollback (db_plugin->cls);
+    cleanup_au (&au_active);
+    GNUNET_assert (NULL == task);
+    task = GNUNET_SCHEDULER_add_now (&run_aggregation,
+                                     s);
+    return;
+  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+    GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
+                                &au_active.wtid,
+                                sizeof (au_active.wtid));
+    have_transient = false;
+    break;
+  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+    have_transient = true;
+    break;
+  }
+  qs = db_plugin->aggregate (db_plugin->cls,
+                             &au_active.h_payto,
+                             &au_active.merchant_pub,
+                             &au_active.wtid,
+                             &au_active.total_amount);
   if (GNUNET_DB_STATUS_HARD_ERROR == qs)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "Failed to execute deposit iteration!\n");
+                "Failed to execute aggregation!\n");
     cleanup_au (&au_active);
     db_plugin->rollback (db_plugin->cls);
     global_ret = EXIT_FAILURE;
@@ -844,13 +556,17 @@ run_aggregation (void *cls)
     return;
   }
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Found %d other deposits to combine into wire transfer with fee %s.\n",
-              qs,
-              TALER_amount2s (&au_active.fees.wire));
+              "Aggregation total is %s.\n",
+              TALER_amount2s (&au_active.total_amount));
 
   /* Subtract wire transfer fee and round to the unit supported by the
      wire transfer method; Check if after rounding down, we still have
      an amount to transfer, and if not mark as 'tiny'. */
+  if (have_transient)
+    GNUNET_assert (0 <=
+                   TALER_amount_add (&au_active.total_amount,
+                                     &au_active.total_amount,
+                                     &trans));
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Rounding aggregate of %s\n",
               TALER_amount2s (&au_active.total_amount));
@@ -867,45 +583,17 @@ run_aggregation (void *cls)
                 "Aggregate value too low for transfer (%d/%s)\n",
                 qs,
                 TALER_amount2s (&au_active.final_amount));
-    /* Rollback ongoing transaction, as we will not use the respective
-       WTID and thus need to remove the tracking data */
-    db_plugin->rollback (db_plugin->cls);
-
-    /* There were results, just the value was too low.  Start another
-       transaction to mark all* of the selected deposits as minor! */
-    if (GNUNET_OK !=
-        db_plugin->start (db_plugin->cls,
-                          "aggregator mark tiny transactions"))
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Failed to start database transaction!\n");
-      global_ret = EXIT_FAILURE;
-      cleanup_au (&au_active);
-      GNUNET_SCHEDULER_shutdown ();
-      release_shard (s);
-      return;
-    }
-    /* Mark transactions by row_id as minor */
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Marking %s (%llu) as tiny\n",
-                TALER_B2S (&au_active.coin_pub),
-                (unsigned long long) au_active.row_id);
-    qs = db_plugin->mark_deposit_tiny (db_plugin->cls,
-                                       &au_active.coin_pub,
-                                       au_active.row_id);
-    if (0 < qs)
-    {
-      for (unsigned int i = 0; i<au_active.rows_offset; i++)
-      {
-        qs = db_plugin->mark_deposit_tiny (db_plugin->cls,
-                                           &au_active.additional_rows[i].
-                                           coin_pub,
-                                           au_active.additional_rows[i].row);
-        if (0 >= qs)
-          break;
-      }
-    }
-    GNUNET_break (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs);
+    if (have_transient)
+      qs = db_plugin->update_aggregation_transient (db_plugin->cls,
+                                                    &au_active.h_payto,
+                                                    &au_active.wtid,
+                                                    &au_active.total_amount);
+    else
+      qs = db_plugin->create_aggregation_transient (db_plugin->cls,
+                                                    &au_active.h_payto,
+                                                    au_active.wa->section_name,
+                                                    &au_active.wtid,
+                                                    &au_active.total_amount);
     if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@@ -962,8 +650,7 @@ run_aggregation (void *cls)
                                               buf_size);
     GNUNET_free (buf);
   }
-  /* Commit the WTID data to 'wire_out' to finally satisfy aggregation
-     table constraints */
+  /* Commit the WTID data to 'wire_out'  */
   if (qs >= 0)
     qs = db_plugin->store_wire_transfer_out (db_plugin->cls,
                                              au_active.execution_time,
@@ -971,6 +658,12 @@ run_aggregation (void *cls)
                                              &au_active.h_payto,
                                              au_active.wa->section_name,
                                              &au_active.final_amount);
+
+  if ( (qs >= 0) &&
+       have_transient)
+    qs = db_plugin->delete_aggregation_transient (db_plugin->cls,
+                                                  &au_active.h_payto,
+                                                  &au_active.wtid);
   cleanup_au (&au_active);
 
   if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
diff --git a/src/exchangedb/plugin_exchangedb_postgres.c b/src/exchangedb/plugin_exchangedb_postgres.c
index 1709f17e4..36a5e48b2 100644
--- a/src/exchangedb/plugin_exchangedb_postgres.c
+++ b/src/exchangedb/plugin_exchangedb_postgres.c
@@ -1160,29 +1160,17 @@ prepare_statements (struct PostgresClosure *pg)
     GNUNET_PQ_make_prepare (
       "deposits_get_ready",
       "SELECT"
-      " dep.deposit_serial_id"
-      ",amount_with_fee_val"
-      ",amount_with_fee_frac"
-      ",denom.fee_deposit_val"
-      ",denom.fee_deposit_frac"
-      ",h_contract_terms"
-      ",payto_uri"
-      ",wire_target_serial_id"
+      " payto_uri"
       ",merchant_pub"
-      ",kc.coin_pub"
       " FROM deposits_by_ready dbr"
       "  JOIN deposits dep"
       "    ON (dbr.coin_pub = dep.coin_pub AND dbr.deposit_serial_id = dep.deposit_serial_id)"
-      "  JOIN wire_targets "
+      "  JOIN wire_targets wt"
       "    USING (wire_target_h_payto)"
-      "  JOIN known_coins kc"
-      "    ON (kc.coin_pub = dep.coin_pub)"
-      "  JOIN denominations denom"
-      "    USING (denominations_serial)"
       " WHERE dbr.wire_deadline<=$1"
       "   AND dbr.shard >= $2"
       "   AND dbr.shard <= $3"
-      "   AND (kyc_ok OR $4)"
+      "   AND (wt.kyc_ok OR $4)"
       " ORDER BY "
       "   dbr.wire_deadline ASC"
       "  ,dbr.shard ASC"
@@ -1218,22 +1206,23 @@ prepare_statements (struct PostgresClosure *pg)
     /* Used in #postgres_aggregate() */
     GNUNET_PQ_make_prepare (
       "aggregate",
-      "WITH rdy AS (" /* find deposits ready */
+      "WITH rdy AS (" /* find deposits ready by merchant */
       "  SELECT"
       "    coin_pub"
       "    FROM deposits_for_matching"
-      "    WHERE refund_deadline<$1"
-      "      AND merchant_pub=$2"
+      "    WHERE refund_deadline<$1" /* filter by shard, only actually executable deposits */
+      "      AND merchant_pub=$2" /* filter by target merchant */
       "    ORDER BY refund_deadline ASC" /* ordering is not critical */
       "    LIMIT "
-      TALER_QUOTE (TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT)
+      TALER_QUOTE (TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT) /* limits transaction size */
       " )"
-      " ,dep AS (" /* restrict to our merchant and account */
+      " ,dep AS (" /* restrict to our merchant and account and mark as done */
       "  UPDATE deposits"
       "     SET done=TRUE"
       "   WHERE coin_pub IN (SELECT coin_pub FROM rdy)"
-      "     AND merchant_pub=$2"
-      "     AND wire_target_h_payto=$3"
+      "     AND merchant_pub=$2" /* theoretically, same coin could be spent at another merchant */
+      "     AND wire_target_h_payto=$3" /* merchant could have a 2nd bank account */
+      "     AND done=FALSE" /* theoretically, same coin could be spend at the same merchant a 2nd time */
       "   RETURNING"
       "     deposit_serial_id"
       "    ,coin_pub"
@@ -1244,18 +1233,26 @@ prepare_statements (struct PostgresClosure *pg)
       "    amount_with_fee_val AS refund_val"
       "   ,amount_with_fee_frac AS refund_frac"
       "   ,coin_pub"
+      "   ,deposit_serial_id" /* theoretically, coin could be in multiple refunded transactions */
       "    FROM refunds"
       "   WHERE coin_pub IN (SELECT coin_pub FROM dep)"
       "     AND deposit_serial_id IN (SELECT deposit_serial_id FROM dep))"
+      " ,coins_with_fees AS (" /* find coins for which deposit fees apply */
+      "  SELECT"
+      "     coin_pub"
+      "    ,deposit_serial_id" /* ensures that if the same coin is deposited twice, it is in the list twice */
+      "    FROM dep"
+      "   WHERE deposit_serial_id NOT IN (SELECT deposit_serial_id FROM ref))"
       " ,fees AS (" /* find deposit fees for non-refunded deposits */
       "  SELECT"
       "    denom.fee_deposit_val AS fee_val"
       "   ,denom.fee_deposit_frac AS fee_frac"
-      "    FROM known_coins kc"
+      "   ,cs.deposit_serial_id" /* ensures we get the fee for each coin, not once per denomination */
+      "    FROM coins_with_fees cs"
+      "    JOIN known_coins kc"
+      "      USING (coin_pub)"
       "    JOIN denominations denom"
-      "      USING (denominations_serial)"
-      "    WHERE coin_pub IN (SELECT coin_pub FROM dep)"
-      "      AND coin_pub NOT IN (SELECT coin_pub FROM ref))"
+      "      USING (denominations_serial))"
       " ,dummy AS (" /* add deposits to aggregation_tracking */
       "    INSERT INTO aggregation_tracking"
       "    (deposit_serial_id"
@@ -1263,14 +1260,14 @@ prepare_statements (struct PostgresClosure *pg)
       "    SELECT deposit_serial_id,$4"
       "      FROM dep)"
       "SELECT" /* calculate totals (deposits, refunds and fees) */
-      "  CAST(COALESCE(SUM(dep.amount_val),0) AS INT8) AS sum_deposit_value"
-      " ,COALESCE(SUM(dep.amount_frac),0) AS sum_deposit_fraction"
+      "  CAST(COALESCE(SUM(dep.amount_val),0) AS INT8) AS sum_deposit_value" /* cast needed, otherwise we get NUMBER */
+      " ,COALESCE(SUM(dep.amount_frac),0) AS sum_deposit_fraction" /* SUM over INT returns INT8 */
       " ,CAST(COALESCE(SUM(ref.refund_val),0) AS INT8) AS sum_refund_value"
       " ,COALESCE(SUM(ref.refund_frac),0) AS sum_refund_fraction"
       " ,CAST(COALESCE(SUM(fees.fee_val),0) AS INT8) AS sum_fee_value"
       " ,COALESCE(SUM(fees.fee_frac),0) AS sum_fee_fraction"
       " FROM dep "
-      "   FULL OUTER JOIN ref ON (FALSE)"
+      "   FULL OUTER JOIN ref ON (FALSE)"    /* We just want all sums */
       "   FULL OUTER JOIN fees ON (FALSE);",
       4),
 
@@ -6270,8 +6267,8 @@ postgres_mark_deposit_done (void *cls,
  * @param end_shard_row maximum shard row to select (inclusive)
  * @param kyc_off true if we should not check the KYC status because
  *                this exchange does not need/support KYC checks.
- * @param deposit_cb function to call for ONE such deposit
- * @param deposit_cb_cls closure for @a deposit_cb
+ * @param[out] merchant_pub set to the public key of a merchant with a ready deposit
+ * @param[out] payto_uri set to the account of the merchant, to be freed by caller
  * @return transaction status code
  */
 static enum GNUNET_DB_QueryStatus
@@ -6279,8 +6276,8 @@ postgres_get_ready_deposit (void *cls,
                             uint64_t start_shard_row,
                             uint64_t end_shard_row,
                             bool kyc_off,
-                            TALER_EXCHANGEDB_DepositIterator deposit_cb,
-                            void *deposit_cb_cls)
+                            struct TALER_MerchantPublicKeyP *merchant_pub,
+                            char **payto_uri)
 {
   struct PostgresClosure *pg = cls;
   struct GNUNET_TIME_Absolute now = {0};
@@ -6291,34 +6288,13 @@ postgres_get_ready_deposit (void *cls,
     GNUNET_PQ_query_param_bool (kyc_off),
     GNUNET_PQ_query_param_end
   };
-  struct TALER_Amount amount_with_fee;
-  struct TALER_Amount deposit_fee;
-  struct TALER_PrivateContractHashP h_contract_terms;
-  struct TALER_MerchantPublicKeyP merchant_pub;
-  struct TALER_CoinSpendPublicKeyP coin_pub;
-  uint64_t serial_id;
-  uint64_t wire_target;
-  char *payto_uri;
   struct GNUNET_PQ_ResultSpec rs[] = {
-    GNUNET_PQ_result_spec_uint64 ("deposit_serial_id",
-                                  &serial_id),
-    GNUNET_PQ_result_spec_uint64 ("wire_target_serial_id",
-                                  &wire_target),
-    TALER_PQ_RESULT_SPEC_AMOUNT ("amount_with_fee",
-                                 &amount_with_fee),
-    TALER_PQ_RESULT_SPEC_AMOUNT ("fee_deposit",
-                                 &deposit_fee),
-    GNUNET_PQ_result_spec_auto_from_type ("h_contract_terms",
-                                          &h_contract_terms),
     GNUNET_PQ_result_spec_auto_from_type ("merchant_pub",
-                                          &merchant_pub),
-    GNUNET_PQ_result_spec_auto_from_type ("coin_pub",
-                                          &coin_pub),
+                                          merchant_pub),
     GNUNET_PQ_result_spec_string ("payto_uri",
-                                  &payto_uri),
+                                  payto_uri),
     GNUNET_PQ_result_spec_end
   };
-  enum GNUNET_DB_QueryStatus qs;
 
   now = GNUNET_TIME_absolute_round_down (GNUNET_TIME_absolute_get (),
                                          pg->aggregator_shift);
@@ -6328,25 +6304,10 @@ postgres_get_ready_deposit (void *cls,
               "Finding ready deposits by deadline %s (%llu)\n",
               GNUNET_TIME_absolute2s (now),
               (unsigned long long) now.abs_value_us);
-
-  qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
-                                                 "deposits_get_ready",
-                                                 params,
-                                                 rs);
-  if (qs <= 0)
-    return qs;
-
-  qs = deposit_cb (deposit_cb_cls,
-                   serial_id,
-                   &merchant_pub,
-                   &coin_pub,
-                   &amount_with_fee,
-                   &deposit_fee,
-                   &h_contract_terms,
-                   wire_target,
-                   payto_uri);
-  GNUNET_PQ_cleanup_result (rs);
-  return qs;
+  return GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
+                                                   "deposits_get_ready",
+                                                   params,
+                                                   rs);
 }
 
 
diff --git a/src/include/taler_exchangedb_plugin.h b/src/include/taler_exchangedb_plugin.h
index 4ca6905e0..06810a7de 100644
--- a/src/include/taler_exchangedb_plugin.h
+++ b/src/include/taler_exchangedb_plugin.h
@@ -3097,8 +3097,8 @@ struct TALER_EXCHANGEDB_Plugin
    * @param end_shard_row maximum shard row to select (inclusive)
    * @param kyc_off true if we should not check the KYC status because
    *                this exchange does not need/support KYC checks.
-   * @param deposit_cb function to call for ONE such deposit
-   * @param deposit_cb_cls closure for @a deposit_cb
+   * @param[out] merchant_pub set to the public key of a merchant with a ready deposit
+   * @param[out] payto_uri set to the account of the merchant, to be freed by caller
    * @return transaction status code
    */
   enum GNUNET_DB_QueryStatus
@@ -3106,8 +3106,8 @@ struct TALER_EXCHANGEDB_Plugin
                        uint64_t start_shard_row,
                        uint64_t end_shard_row,
                        bool kyc_off,
-                       TALER_EXCHANGEDB_DepositIterator deposit_cb,
-                       void *deposit_cb_cls);
+                       struct TALER_MerchantPublicKeyP *merchant_pub,
+                       char **payto_uri);
 
 
 /**