do all DB operations from downloading /keys in one shot

3 files changed, 212 insertions, 219 deletions

diff --git a/src/backend/taler-merchant-httpd_exchanges.c b/src/backend/taler-merchant-httpd_exchanges.c
index 7e99704c..78b4eea6 100644
--- a/src/backend/taler-merchant-httpd_exchanges.c
+++ b/src/backend/taler-merchant-httpd_exchanges.c
@@ -422,6 +422,8 @@ set_exchange_accounts (
   unsigned int accounts_len,
   const struct TALER_EXCHANGE_WireAccount accounts[static accounts_len])
 {
+  enum GNUNET_GenericReturnValue ret = GNUNET_OK;
+
   purge_exchange_accounts (exchange);
   for (unsigned int i = 0; i<accounts_len; i++)
   {
@@ -456,7 +458,9 @@ set_exchange_accounts (
                           REG_NOSUB | REG_EXTENDED))
         {
           GNUNET_break_op (0);
-          return GNUNET_SYSERR;
+          GNUNET_free (r);
+          ret = GNUNET_SYSERR;
+          continue;
         }
         break;
       }
@@ -465,7 +469,7 @@ set_exchange_accounts (
                                    r);
     }
   }
-  return GNUNET_OK;
+  return ret;
 }
 
 
@@ -555,15 +559,15 @@ process_find_operations (struct TMH_Exchange *exchange)
 
 /**
  * Function called with information about the wire fees for each wire method.
- * Stores the wire fees with the exchange for later use.
+ * Stores the wire fees within our internal data structures for later use.
  *
  * @param exchange connection to the exchange
  * @param master_pub public key of the exchange
  * @param num_methods number of wire methods supported
  * @param fbm wire fees by method
- * @return #TALER_EC_NONE on success
+ * @return #GNUNET_OK on success
  */
-static enum TALER_ErrorCode
+static enum GNUNET_GenericReturnValue
 process_wire_fees (
   struct TMH_Exchange *exchange,
   const struct TALER_MasterPublicKeyP *master_pub,
@@ -576,7 +580,6 @@ process_wire_fees (
     const struct TALER_EXCHANGE_WireAggregateFees *fees = fbm[i].fees_head;
     struct FeesByWireMethod *f;
     struct TALER_EXCHANGE_WireAggregateFees *endp;
-    struct TALER_EXCHANGE_WireAggregateFees *af;
 
     for (f = exchange->wire_fees_head; NULL != f; f = f->next)
       if (0 == strcasecmp (wire_method,
@@ -606,73 +609,16 @@ process_wire_fees (
                                      !=,
                                      endp->end_date)) )
     {
-      /* Hole in the fee structure, not allowed! */
+      /* Hole or overlap in the fee structure, not allowed! */
       GNUNET_break_op (0);
-      return TALER_EC_MERCHANT_GENERIC_HOLE_IN_WIRE_FEE_STRUCTURE;
+      return GNUNET_SYSERR;
     }
     while (NULL != fees)
     {
-      struct GNUNET_HashCode h_wire_method;
-      enum GNUNET_DB_QueryStatus qs;
+      struct TALER_EXCHANGE_WireAggregateFees *af;
 
       af = GNUNET_new (struct TALER_EXCHANGE_WireAggregateFees);
       *af = *fees;
-      GNUNET_CRYPTO_hash (wire_method,
-                          strlen (wire_method) + 1,
-                          &h_wire_method);
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "Storing wire fee for `%s' and method `%s' at %s in DB; the fee is %s\n",
-                  TALER_B2S (master_pub),
-                  wire_method,
-                  GNUNET_TIME_timestamp2s (af->start_date),
-                  TALER_amount2s (&af->fees.wire));
-      TMH_db->preflight (TMH_db->cls);
-      if (GNUNET_OK !=
-          TMH_db->start (TMH_db->cls,
-                         "store wire fee"))
-      {
-        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                    "Failed to start database transaction to store exchange wire fees (will try to continue anyway)!\n");
-        GNUNET_free (af);
-        fees = fees->next;
-        continue;
-      }
-      qs = TMH_db->store_wire_fee_by_exchange (TMH_db->cls,
-                                               master_pub,
-                                               &h_wire_method,
-                                               &af->fees,
-                                               af->start_date,
-                                               af->end_date,
-                                               &af->master_sig);
-      if (0 > qs)
-      {
-        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                    "Failed to persist exchange wire fees in merchant DB (will try to continue anyway)!\n");
-        GNUNET_free (af);
-        fees = fees->next;
-        TMH_db->rollback (TMH_db->cls);
-        continue;
-      }
-      if (0 == qs)
-      {
-        /* Entry was already in DB, fine, continue as if we had succeeded */
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                    "Fees already in DB, rolling back transaction attempt!\n");
-        TMH_db->rollback (TMH_db->cls);
-      }
-      if (0 < qs)
-      {
-        /* Inserted into DB, make sure transaction completes */
-        qs = TMH_db->commit (TMH_db->cls);
-        if (0 > qs)
-        {
-          GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                      "Failed to persist exchange wire fees in merchant DB (will try to continue anyway)!\n");
-          GNUNET_free (af);
-          fees = fees->next;
-          continue;
-        }
-      }
       af->next = NULL;
       if (NULL == endp)
         f->af = af;
@@ -682,7 +628,7 @@ process_wire_fees (
       fees = fees->next;
     } /* all fees for this method */
   } /* for all methods (i) */
-  return TALER_EC_NONE;
+  return GNUNET_OK;
 }
 
 
@@ -742,117 +688,6 @@ add_restriction (json_t *restrictions,
 
 
 /**
- * Function called with information about the wire accounts of the exchange.
- *
- * @param exchange the exchange
- * @param master_pub public key of the exchange
- * @param accounts_len length of the @a accounts array
- * @param accounts list of wire accounts of the exchange
- * @return #TALER_EC_NONE on success
- */
-static enum TALER_ErrorCode
-process_wire_accounts (
-  struct TMH_Exchange *exchange,
-  const struct TALER_MasterPublicKeyP *master_pub,
-  unsigned int accounts_len,
-  const struct TALER_EXCHANGE_WireAccount accounts[static accounts_len])
-{
-  for (unsigned int r = 0; r<MAX_RETRIES; r++)
-  {
-    enum GNUNET_DB_QueryStatus qs;
-
-    if (GNUNET_OK !=
-        TMH_db->start (TMH_db->cls,
-                       "update_exchange_accounts"))
-    {
-      TMH_db->rollback (TMH_db->cls);
-      GNUNET_break (0);
-      return TALER_EC_GENERIC_DB_START_FAILED;
-    }
-    qs = TMH_db->delete_exchange_accounts (TMH_db->cls,
-                                           master_pub);
-    if (qs < 0)
-    {
-      TMH_db->rollback (TMH_db->cls);
-      if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
-        continue;
-      GNUNET_break (0);
-      return TALER_EC_GENERIC_DB_STORE_FAILED;
-    }
-    for (unsigned int i = 0; i<accounts_len; i++)
-    {
-      const struct TALER_EXCHANGE_WireAccount *account = &accounts[i];
-      json_t *debit_restrictions;
-      json_t *credit_restrictions;
-
-      debit_restrictions = json_array ();
-      GNUNET_assert (NULL != debit_restrictions);
-      credit_restrictions = json_array ();
-      GNUNET_assert (NULL != credit_restrictions);
-      for (unsigned int j = 0; j<account->debit_restrictions_length; j++)
-      {
-        if (GNUNET_OK !=
-            add_restriction (debit_restrictions,
-                             &account->debit_restrictions[j]))
-        {
-          TMH_db->rollback (TMH_db->cls);
-          GNUNET_break (0);
-          json_decref (debit_restrictions);
-          json_decref (credit_restrictions);
-          return TALER_EC_MERCHANT_GENERIC_EXCHANGE_WIRE_REQUEST_FAILED;
-        }
-      }
-      for (unsigned int j = 0; j<account->credit_restrictions_length; j++)
-      {
-        if (GNUNET_OK !=
-            add_restriction (credit_restrictions,
-                             &account->credit_restrictions[j]))
-        {
-          TMH_db->rollback (TMH_db->cls);
-          GNUNET_break (0);
-          json_decref (debit_restrictions);
-          json_decref (credit_restrictions);
-          return TALER_EC_MERCHANT_GENERIC_EXCHANGE_WIRE_REQUEST_FAILED;
-        }
-      }
-      qs = TMH_db->insert_exchange_account (TMH_db->cls,
-                                            master_pub,
-                                            account->payto_uri,
-                                            account->conversion_url,
-                                            debit_restrictions,
-                                            credit_restrictions,
-                                            &account->master_sig);
-      json_decref (debit_restrictions);
-      json_decref (credit_restrictions);
-      if (qs < 0)
-      {
-        TMH_db->rollback (TMH_db->cls);
-        if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
-          goto outer;
-        GNUNET_break (0);
-        return TALER_EC_GENERIC_DB_STORE_FAILED;
-      }
-    }
-    qs = TMH_db->commit (TMH_db->cls);
-    if (qs < 0)
-    {
-      TMH_db->rollback (TMH_db->cls);
-      if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
-        continue;
-      GNUNET_break (0);
-      return TALER_EC_GENERIC_DB_COMMIT_FAILED;
-    }
-    set_exchange_accounts (exchange,
-                           accounts_len,
-                           accounts);
-    return TALER_EC_NONE;
-outer:;
-  }
-  return TALER_EC_GENERIC_DB_SOFT_FAILURE;
-}
-
-
-/**
  * Retry getting keys from the given exchange in the closure.
  *
  * @param cls the `struct TMH_Exchange *`
@@ -1114,28 +949,26 @@ fail_and_retry (struct TMH_Exchange *exchange)
 }
 
 
-static void
-keys_mgmt_cb (void *cls,
-              const struct TALER_EXCHANGE_KeysResponse *kr,
-              struct TALER_EXCHANGE_Keys *keys)
+/**
+ * Update our information in the database about the
+ * /keys of an exchange. Run inside of a database
+ * transaction scope that will re-try and/or commit
+ * depending on the return value.
+ *
+ * @param keys information to persist
+ * @return transaction status
+ */
+static enum GNUNET_DB_QueryStatus
+insert_keys_data (const struct TALER_EXCHANGE_Keys *keys)
 {
-  struct TMH_Exchange *exchange = cls;
   enum GNUNET_DB_QueryStatus qs;
 
-  exchange->conn = NULL;
-  if (MHD_HTTP_OK != kr->hr.http_status)
-  {
-    fail_and_retry (exchange);
-    return;
-  }
-
   /* store exchange online signing keys in our DB */
   for (unsigned int i = 0; i<keys->num_sign_keys; i++)
   {
     struct TALER_EXCHANGE_SigningPublicKey *sign_key = &keys->sign_keys[i];
     enum GNUNET_DB_QueryStatus qs;
 
-    TMH_db->preflight (TMH_db->cls);
     qs = TMH_db->insert_exchange_signkey (
       TMH_db->cls,
       &keys->master_pub,
@@ -1147,17 +980,109 @@ keys_mgmt_cb (void *cls,
     /* 0 is OK, we may already have the key in the DB! */
     if (0 > qs)
     {
-      GNUNET_break (0);
-      fail_and_retry (exchange);
-      return;
+      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+      return qs;
     }
   }
 
   qs = TMH_db->insert_exchange_keys (TMH_db->cls,
                                      keys);
-  TALER_EXCHANGE_keys_decref (keys);
-  GNUNET_break (qs > 0);
-  if (qs > 0)
+  if (0 > qs)
+  {
+    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+    return qs;
+  }
+
+  qs = TMH_db->delete_exchange_accounts (TMH_db->cls,
+                                         &keys->master_pub);
+  if (0 > qs)
+  {
+    GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+    return qs;
+  }
+
+  for (unsigned int i = 0; i<keys->accounts_len; i++)
+  {
+    const struct TALER_EXCHANGE_WireAccount *account = &keys->accounts[i];
+    json_t *debit_restrictions;
+    json_t *credit_restrictions;
+
+    debit_restrictions = json_array ();
+    GNUNET_assert (NULL != debit_restrictions);
+    credit_restrictions = json_array ();
+    GNUNET_assert (NULL != credit_restrictions);
+    for (unsigned int j = 0; j<account->debit_restrictions_length; j++)
+    {
+      if (GNUNET_OK !=
+          add_restriction (debit_restrictions,
+                           &account->debit_restrictions[j]))
+      {
+        TMH_db->rollback (TMH_db->cls);
+        GNUNET_break (0);
+        json_decref (debit_restrictions);
+        json_decref (credit_restrictions);
+        return GNUNET_DB_STATUS_HARD_ERROR;
+      }
+    }
+    for (unsigned int j = 0; j<account->credit_restrictions_length; j++)
+    {
+      if (GNUNET_OK !=
+          add_restriction (credit_restrictions,
+                           &account->credit_restrictions[j]))
+      {
+        TMH_db->rollback (TMH_db->cls);
+        GNUNET_break (0);
+        json_decref (debit_restrictions);
+        json_decref (credit_restrictions);
+        return GNUNET_DB_STATUS_HARD_ERROR;
+      }
+    }
+    qs = TMH_db->insert_exchange_account (
+      TMH_db->cls,
+      &keys->master_pub,
+      account->payto_uri,
+      account->conversion_url,
+      debit_restrictions,
+      credit_restrictions,
+      &account->master_sig);
+    json_decref (debit_restrictions);
+    json_decref (credit_restrictions);
+    if (qs < 0)
+    {
+      GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+      return qs;
+    }
+  } /* end 'for all accounts' */
+
+  for (unsigned int i = 0; i<keys->fees_len; i++)
+  {
+    const struct TALER_EXCHANGE_WireFeesByMethod *fbm = &keys->fees[i];
+    const char *wire_method = fbm->method;
+    const struct TALER_EXCHANGE_WireAggregateFees *fees = fbm->fees_head;
+
+    while (NULL != fees)
+    {
+      struct GNUNET_HashCode h_wire_method;
+
+      GNUNET_CRYPTO_hash (wire_method,
+                          strlen (wire_method) + 1,
+                          &h_wire_method);
+      qs = TMH_db->store_wire_fee_by_exchange (TMH_db->cls,
+                                               &keys->master_pub,
+                                               &h_wire_method,
+                                               &fees->fees,
+                                               fees->start_date,
+                                               fees->end_date,
+                                               &fees->master_sig);
+      if (0 > qs)
+      {
+        GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
+        return qs;
+      }
+      fees = fees->next;
+    } /* all fees for this method */
+  } /* for all methods (i) */
+
   {
     struct GNUNET_DB_EventHeaderP es = {
       .size = ntohs (sizeof (es)),
@@ -1166,10 +1091,75 @@ keys_mgmt_cb (void *cls,
 
     TMH_db->event_notify (TMH_db->cls,
                           &es,
-                          exchange->url,
-                          strlen (exchange->url) + 1);
-    exchange->retry_delay = GNUNET_TIME_UNIT_ZERO;
+                          keys->exchange_url,
+                          strlen (keys->exchange_url) + 1);
+  }
+  return qs;
+}
+
+
+static void
+keys_mgmt_cb (void *cls,
+              const struct TALER_EXCHANGE_KeysResponse *kr,
+              struct TALER_EXCHANGE_Keys *keys)
+{
+  struct TMH_Exchange *exchange = cls;
+  enum GNUNET_DB_QueryStatus qs;
+
+  exchange->conn = NULL;
+  if (MHD_HTTP_OK != kr->hr.http_status)
+  {
+    fail_and_retry (exchange);
+    TALER_EXCHANGE_keys_decref (keys);
+    return;
   }
+  TMH_db->preflight (TMH_db->cls);
+  for (unsigned int r = 0; r<MAX_RETRIES; r++)
+  {
+    if (GNUNET_OK !=
+        TMH_db->start (TMH_db->cls,
+                       "update exchange key data"))
+    {
+      TMH_db->rollback (TMH_db->cls);
+      GNUNET_break (0);
+      fail_and_retry (exchange);
+      TALER_EXCHANGE_keys_decref (keys);
+      return;
+    }
+
+    qs = insert_keys_data (keys);
+    if (qs < 0)
+    {
+      TMH_db->rollback (TMH_db->cls);
+      if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
+        continue;
+      GNUNET_break (0);
+      fail_and_retry (exchange);
+      TALER_EXCHANGE_keys_decref (keys);
+      return;
+    }
+
+    qs = TMH_db->commit (TMH_db->cls);
+    if (qs < 0)
+    {
+      TMH_db->rollback (TMH_db->cls);
+      if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
+        continue;
+      GNUNET_break (0);
+      fail_and_retry (exchange);
+      TALER_EXCHANGE_keys_decref (keys);
+      return;
+    }
+  } /* end of retry loop */
+  if (qs < 0)
+  {
+    GNUNET_break (0);
+    fail_and_retry (exchange);
+    TALER_EXCHANGE_keys_decref (keys);
+    return;
+  }
+  TALER_EXCHANGE_keys_decref (keys);
+  exchange->retry_delay = GNUNET_TIME_UNIT_ZERO;
 }
 
 
@@ -1381,7 +1371,6 @@ update_exchange_keys (void *cls,
   const char *url = extra;
   struct TMH_Exchange *exchange;
   struct TALER_EXCHANGE_Keys *keys;
-  enum TALER_ErrorCode ecx;
 
   if ( (NULL == extra) ||
        (0 == extra_len) )
@@ -1410,23 +1399,26 @@ update_exchange_keys (void *cls,
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "Loaded /keys from database with %u accounts\n",
               keys->accounts_len);
-  // FIXME: split into parsing part (do here)
-  // and DB part (do when receiving /keys!)
-  ecx = process_wire_accounts (exchange,
-                               &keys->master_pub,
-                               keys->accounts_len,
-                               keys->accounts);
-  if (TALER_EC_NONE == ecx)
-    ecx = process_wire_fees (exchange,
-                             &keys->master_pub,
-                             keys->fees_len,
-                             keys->fees);
-  if (TALER_EC_NONE != ecx)
+  if (GNUNET_OK !=
+      set_exchange_accounts (exchange,
+                             keys->accounts_len,
+                             keys->accounts))
   {
-    /* Report hard failure to all callbacks! */
+    /* invalid account specification given */
     GNUNET_break_op (0);
-    fail_and_retry (exchange);
-    TALER_EXCHANGE_keys_decref (keys);
+    /* but: we can continue anyway, things may just not
+       work, but probably better than to not keep going. */
+  }
+  if (GNUNET_OK !=
+      process_wire_fees (exchange,
+                         &keys->master_pub,
+                         keys->fees_len,
+                         keys->fees))
+  {
+    /* invalid wire fee specification given */
+    GNUNET_break_op (0);
+    /* but: we can continue anyway, things may just not
+       work, but probably better than to not keep going. */
     return;
   }
 
diff --git a/src/backenddb/pg_insert_exchange_signkey.c b/src/backenddb/pg_insert_exchange_signkey.c
index 81d27d60..b32a22b0 100644
--- a/src/backenddb/pg_insert_exchange_signkey.c
+++ b/src/backenddb/pg_insert_exchange_signkey.c
@@ -47,7 +47,6 @@ TMH_PG_insert_exchange_signkey (
   };
 
   check_connection (pg);
-  postgres_preflight (pg);
   PREPARE (pg,
            "insert_exchange_signkey",
            "INSERT INTO merchant_exchange_signing_keys"
@@ -58,7 +57,8 @@ TMH_PG_insert_exchange_signkey (
            ",end_date"
            ",master_sig)"
            "VALUES"
-           "($1, $2, $3, $4, $5, $6)");
+           "($1, $2, $3, $4, $5, $6)"
+           " ON CONFLICT DO NOTHING;");
   return GNUNET_PQ_eval_prepared_non_select (pg->conn,
                                              "insert_exchange_signkey",
                                              params);
diff --git a/src/backenddb/plugin_merchantdb_postgres.c b/src/backenddb/plugin_merchantdb_postgres.c
index 975fae93..f129b90b 100644
--- a/src/backenddb/plugin_merchantdb_postgres.c
+++ b/src/backenddb/plugin_merchantdb_postgres.c
@@ -5180,7 +5180,8 @@ postgres_connect (void *cls)
                             ",end_date"
                             ",master_sig)"
                             " VALUES "
-                            "($1, $2, $3, $4, $5, $6, $7, $8, $9)"),
+                            "($1, $2, $3, $4, $5, $6, $7, $8, $9)"
+                            " ON CONFLICT DO NOTHING"),
     /* For postgres_insert_reserve() */
     GNUNET_PQ_make_prepare ("insert_reserve",
                             "INSERT INTO merchant_reward_reserves"