adapt /admin/add/incoming to #5010

4 files changed, 138 insertions, 155 deletions

diff --git a/src/exchange/taler-exchange-httpd_admin.c b/src/exchange/taler-exchange-httpd_admin.c
index e40775f51..83bfedf8b 100644
--- a/src/exchange/taler-exchange-httpd_admin.c
+++ b/src/exchange/taler-exchange-httpd_admin.c
@@ -28,66 +28,96 @@
 
 
 /**
+ * Closure for #admin_add_incoming_transaction()
+ */
+struct AddIncomingContext
+{
+  /**
+   * public key of the reserve
+   */
+  struct TALER_ReservePublicKeyP reserve_pub;
+
+  /**
+   * amount to add to the reserve
+   */
+  struct TALER_Amount amount;
+
+  /**
+   * When did we receive the wire transfer
+   */
+  struct GNUNET_TIME_Absolute execution_time;
+
+  /**
+   * which account send the funds
+   */
+  json_t *sender_account_details;
+
+  /**
+   * Information that uniquely identifies the transfer
+   */
+  json_t *transfer_details;
+
+  /**
+   * Set to the transaction status.
+   */
+  enum GNUNET_DB_QueryStatus qs;
+};
+
+
+/** 
  * Add an incoming transaction to the database.  Checks if the
  * transaction is fresh (not a duplicate) and if so adds it to
  * the database.
  *
- * @param connection the MHD connection to handle
- * @param reserve_pub public key of the reserve
- * @param amount amount to add to the reserve
- * @param execution_time when did we receive the wire transfer
- * @param sender_account_details which account send the funds
- * @param transfer_details information that uniquely identifies the transfer
- * @return MHD result code
+ * If it returns a non-error code, the transaction logic MUST
+ * NOT queue a MHD response.  IF it returns an hard error, the
+ * transaction logic MUST queue a MHD response and set @a mhd_ret.  IF
+ * it returns the soft error code, the function MAY be called again to
+ * retry and MUST not queue a MHD response.
+ *
+ * @param cls closure with the `struct AddIncomingContext *`
+ * @param connection MHD request which triggered the transaction
+ * @param session database session to use
+ * @param[out] mhd_ret set to MHD response status for @a connection,
+ *             if transaction failed (!)
+ * @return transaction status
  */
-static int
-execute_admin_add_incoming (struct MHD_Connection *connection,
-			    const struct TALER_ReservePublicKeyP *reserve_pub,
-			    const struct TALER_Amount *amount,
-			    struct GNUNET_TIME_Absolute execution_time,
-			    const json_t *sender_account_details,
-			    const json_t *transfer_details)
+static enum GNUNET_DB_QueryStatus
+admin_add_incoming_transaction (void *cls,
+				struct MHD_Connection *connection,
+				struct TALER_EXCHANGEDB_Session *session,
+				int *mhd_ret)
 {
-  struct TALER_EXCHANGEDB_Session *session;
-  int ret;
+  struct AddIncomingContext *aic = cls;
   void *json_str;
 
-  if (NULL == (session = TEH_plugin->get_session (TEH_plugin->cls)))
-  {
-    GNUNET_break (0);
-    return TEH_RESPONSE_reply_internal_db_error (connection,
-						 TALER_EC_DB_SETUP_FAILED);
-  }
-  json_str = json_dumps (transfer_details,
+  json_str = json_dumps (aic->transfer_details,
                          JSON_INDENT(2));
   if (NULL == json_str)
   {
     GNUNET_break (0);
-    return TEH_RESPONSE_reply_internal_db_error (connection,
-						 TALER_EC_PARSER_OUT_OF_MEMORY);
+    *mhd_ret = TEH_RESPONSE_reply_internal_db_error (connection,
+						     TALER_EC_PARSER_OUT_OF_MEMORY);
+    return GNUNET_DB_STATUS_HARD_ERROR;
   }
-  ret = TEH_plugin->reserves_in_insert (TEH_plugin->cls,
-                                        session,
-                                        reserve_pub,
-                                        amount,
-                                        execution_time,
-                                        sender_account_details,
-                                        json_str,
-                                        strlen (json_str));
+  aic->qs = TEH_plugin->reserves_in_insert (TEH_plugin->cls,
+					    session,
+					    &aic->reserve_pub,
+					    &aic->amount,
+					    aic->execution_time,
+					    aic->sender_account_details,
+					    json_str,
+					    strlen (json_str));
   free (json_str);
-  if (GNUNET_SYSERR == ret)
+  
+  if (GNUNET_DB_STATUS_HARD_ERROR == aic->qs)
   {
     GNUNET_break (0);
-    return TEH_RESPONSE_reply_internal_db_error (connection,
-						 TALER_EC_ADMIN_ADD_INCOMING_DB_STORE);
+    *mhd_ret = TEH_RESPONSE_reply_internal_db_error (connection,
+						     TALER_EC_ADMIN_ADD_INCOMING_DB_STORE);
+    return GNUNET_DB_STATUS_HARD_ERROR;
   }
-  return TEH_RESPONSE_reply_json_pack (connection,
-                                       MHD_HTTP_OK,
-                                       "{s:s}",
-                                       "status",
-                                       (GNUNET_OK == ret)
-                                       ? "NEW"
-                                       : "DUP");
+  return aic->qs;
 }
 
 
@@ -110,23 +140,20 @@ TEH_ADMIN_handler_admin_add_incoming (struct TEH_RequestHandler *rh,
                                       const char *upload_data,
                                       size_t *upload_data_size)
 {
-  struct TALER_ReservePublicKeyP reserve_pub;
-  struct TALER_Amount amount;
-  struct GNUNET_TIME_Absolute at;
+  struct AddIncomingContext aic;
   enum TALER_ErrorCode ec;
   char *emsg;
-  json_t *sender_account_details;
-  json_t *transfer_details;
   json_t *root;
   struct GNUNET_JSON_Specification spec[] = {
-    GNUNET_JSON_spec_fixed_auto ("reserve_pub", &reserve_pub),
-    TALER_JSON_spec_amount ("amount", &amount),
-    GNUNET_JSON_spec_absolute_time ("execution_date", &at),
-    GNUNET_JSON_spec_json ("sender_account_details", &sender_account_details),
-    GNUNET_JSON_spec_json ("transfer_details", &transfer_details),
+    GNUNET_JSON_spec_fixed_auto ("reserve_pub", &aic.reserve_pub),
+    TALER_JSON_spec_amount ("amount", &aic.amount),
+    GNUNET_JSON_spec_absolute_time ("execution_date", &aic.execution_time),
+    GNUNET_JSON_spec_json ("sender_account_details", &aic.sender_account_details),
+    GNUNET_JSON_spec_json ("transfer_details", &aic.transfer_details),
     GNUNET_JSON_spec_end ()
   };
   int res;
+  int mhd_ret;
 
   res = TEH_PARSE_post_json (connection,
                              connection_cls,
@@ -135,7 +162,8 @@ TEH_ADMIN_handler_admin_add_incoming (struct TEH_RequestHandler *rh,
                              &root);
   if (GNUNET_SYSERR == res)
     return MHD_NO;
-  if ( (GNUNET_NO == res) || (NULL == root) )
+  if ( (GNUNET_NO == res) ||
+       (NULL == root) )
     return MHD_YES;
   res = TEH_PARSE_json_data (connection,
                              root,
@@ -148,37 +176,43 @@ TEH_ADMIN_handler_admin_add_incoming (struct TEH_RequestHandler *rh,
     return (GNUNET_SYSERR == res) ? MHD_NO : MHD_YES;
   }
   if (TALER_EC_NONE !=
-      (ec = TEH_json_validate_wireformat (sender_account_details,
+      (ec = TEH_json_validate_wireformat (aic.sender_account_details,
                                           GNUNET_NO,
                                           &emsg)))
   {
     GNUNET_JSON_parse_free (spec);
-    res = TEH_RESPONSE_reply_external_error (connection,
-                                             ec,
-                                             emsg);
+    mhd_ret = TEH_RESPONSE_reply_external_error (connection,
+						 ec,
+						 emsg);
     GNUNET_free (emsg);
-    return res;
+    return mhd_ret;
   }
-  if (0 != strcasecmp (amount.currency,
+  if (0 != strcasecmp (aic.amount.currency,
                        TEH_exchange_currency_string))
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Exchange uses currency `%s', but /admin/add/incoming tried to use currency `%s'\n",
                 TEH_exchange_currency_string,
-                amount.currency);
+                aic.amount.currency);
     GNUNET_JSON_parse_free (spec);
     return TEH_RESPONSE_reply_arg_invalid (connection,
 					   TALER_EC_ADMIN_ADD_INCOMING_CURRENCY_UNSUPPORTED,
                                            "amount:currency");
   }
-  res = execute_admin_add_incoming (connection,
-				    &reserve_pub,
-				    &amount,
-				    at,
-				    sender_account_details,
-				    transfer_details);
+  res = TEH_DB_run_transaction (connection,
+				&mhd_ret,
+				&admin_add_incoming_transaction,
+				&aic);
   GNUNET_JSON_parse_free (spec);
-  return res;
+  if (GNUNET_OK != res)
+    return mhd_ret;
+  return TEH_RESPONSE_reply_json_pack (connection,
+                                       MHD_HTTP_OK,
+                                       "{s:s}",
+                                       "status",
+                                       (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == aic.qs)
+                                       ? "NEW"
+                                       : "DUP");
 }
 
 /* end of taler-exchange-httpd_admin.c */
diff --git a/src/exchangedb/plugin_exchangedb_postgres.c b/src/exchangedb/plugin_exchangedb_postgres.c
index 2a47f2503..6758a308a 100644
--- a/src/exchangedb/plugin_exchangedb_postgres.c
+++ b/src/exchangedb/plugin_exchangedb_postgres.c
@@ -1952,11 +1952,9 @@ reserves_update (void *cls,
  * @param sender_account_details account information for the sender
  * @param wire_reference unique reference identifying the wire transfer (binary blob)
  * @param wire_reference_size number of bytes in @a wire_reference
- * @return #GNUNET_OK upon success; #GNUNET_NO if the given
- *         @a details are already known for this @a reserve_pub,
- *         #GNUNET_SYSERR upon failures (DB error, incompatible currency)
+ * @return transaction status code
  */
-static int
+static enum GNUNET_DB_QueryStatus
 postgres_reserves_in_insert (void *cls,
                              struct TALER_EXCHANGEDB_Session *session,
                              const struct TALER_ReservePublicKeyP *reserve_pub,
@@ -1967,18 +1965,11 @@ postgres_reserves_in_insert (void *cls,
                              size_t wire_reference_size)
 {
   struct PostgresClosure *pg = cls;
-  PGresult *result;
   enum GNUNET_DB_QueryStatus reserve_exists;
+  enum GNUNET_DB_QueryStatus qs;
   struct TALER_EXCHANGEDB_Reserve reserve;
   struct GNUNET_TIME_Absolute expiry;
 
-  if (GNUNET_OK !=
-      postgres_start (cls,
-                      session))
-  {
-    GNUNET_break (0);
-    return GNUNET_SYSERR;
-  }
   reserve.pub = *reserve_pub;
   reserve_exists = postgres_reserve_get (cls,
                                          session,
@@ -1986,7 +1977,7 @@ postgres_reserves_in_insert (void *cls,
   if (0 > reserve_exists)
   {
     GNUNET_break (0);
-    goto rollback;
+    return reserve_exists;
   }
   if ( (0 == reserve.balance.value) &&
        (0 == reserve.balance.fraction) )
@@ -2030,22 +2021,22 @@ postgres_reserves_in_insert (void *cls,
 
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Reserve does not exist; creating a new one\n");
-    result = GNUNET_PQ_exec_prepared (session->conn,
-                                      "reserve_create",
-                                      params);
-    if (PGRES_COMMAND_OK != PQresultStatus(result))
+    qs = GNUNET_PQ_eval_prepared_non_select (session->conn,
+					     "reserve_create",
+					     params);
+    if (0 > qs)
+      return qs;
+    if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
     {
-      QUERY_ERR (result, session->conn);
-      PQclear (result);
-      goto rollback;
+      /* Maybe DB did not detect serializiability error already,
+	 but clearly there must be one. Still odd. */
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_SOFT_ERROR;
     }
-    PQclear (result);
   }
   /* Create new incoming transaction, SQL "primary key" logic
      is used to guard against duplicates.  If a duplicate is
-     detected, we rollback (which really shouldn't undo
-     anything) and return #GNUNET_NO to indicate that this failure
-     is kind-of harmless (already executed). */
+     detected, we just "succeed" with no changes. */
   {
     struct GNUNET_PQ_QueryParam params[] = {
       GNUNET_PQ_query_param_auto_from_type (&reserve.pub),
@@ -2057,34 +2048,12 @@ postgres_reserves_in_insert (void *cls,
       GNUNET_PQ_query_param_end
     };
 
-    result = GNUNET_PQ_exec_prepared (session->conn,
-                                      "reserves_in_add_transaction",
-                                      params);
-  }
-  if (PGRES_COMMAND_OK != PQresultStatus(result))
-  {
-    const char *efield;
-
-    efield = PQresultErrorField (result,
-				 PG_DIAG_SQLSTATE);
-    if ( (PGRES_FATAL_ERROR == PQresultStatus(result)) &&
-	 (NULL != strstr ("23505", /* unique violation */
-			  efield)) )
-    {
-      /* This means we had the same reserve/justification/details
-	 before */
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Uniqueness violation, deposit details already known\n");
-      PQclear (result);
-      postgres_rollback (cls,
-			 session);
-      return GNUNET_NO;
-    }
-    QUERY_ERR (result, session->conn);
-    PQclear (result);
-    goto rollback;
+    qs = GNUNET_PQ_eval_prepared_non_select (session->conn,
+					     "reserves_in_add_transaction",
+					     params);
+    if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
+      return qs;
   }
-  PQclear (result);
 
   if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == reserve_exists)
   {
@@ -2094,7 +2063,6 @@ postgres_reserves_in_insert (void *cls,
        back for duplicate transactions; like this, we should virtually
        never actually have to rollback anything. */
     struct TALER_EXCHANGEDB_Reserve updated_reserve;
-    enum GNUNET_DB_QueryStatus qs;
     
     updated_reserve.pub = reserve.pub;
     if (GNUNET_OK !=
@@ -2105,38 +2073,15 @@ postgres_reserves_in_insert (void *cls,
       /* currency overflow or incompatible currency */
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                   "Attempt to deposit incompatible amount into reserve\n");
-      goto rollback;
+      return GNUNET_DB_STATUS_HARD_ERROR;
     }
     updated_reserve.expiry = GNUNET_TIME_absolute_max (expiry,
                                                        reserve.expiry);
-    qs = reserves_update (cls,
-			  session,
-			  &updated_reserve);
-    if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
-      goto rollback; /* FIXME: #5010 */
-    if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
-    {
-      postgres_rollback (cls,
-			 session);
-      return GNUNET_SYSERR;
-    }
-  }
-  if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
-      postgres_commit (cls,
-                       session))
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                "Failed to commit transaction adding amount to reserve\n");
-    return GNUNET_SYSERR;
+    return reserves_update (cls,
+			    session,
+			    &updated_reserve);
   }
-  return GNUNET_OK;
-
- rollback:
-  GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-              "Transaction failed, doing rollback\n");
-  postgres_rollback (cls,
-                     session);
-  return GNUNET_SYSERR;
+  return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
 }
 
 
diff --git a/src/exchangedb/test_exchangedb.c b/src/exchangedb/test_exchangedb.c
index f79839246..701c20c79 100644
--- a/src/exchangedb/test_exchangedb.c
+++ b/src/exchangedb/test_exchangedb.c
@@ -162,7 +162,6 @@ check_reserve (struct TALER_EXCHANGEDB_Session *session,
   struct TALER_EXCHANGEDB_Reserve reserve;
 
   reserve.pub = *pub;
-
   FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
           plugin->reserve_get (plugin->cls,
                                session,
@@ -1485,6 +1484,10 @@ run (void *cls)
     goto drop;
   }
 
+  FAILIF (GNUNET_OK !=
+          plugin->start (plugin->cls,
+                         session));
+
   /* test DB is empty */
   FAILIF (GNUNET_NO !=
           plugin->select_payback_above_serial_id (plugin->cls,
@@ -1520,7 +1523,7 @@ run (void *cls)
                                                    session,
                                                    &rr,
                                                    &rr_size));
-  FAILIF (GNUNET_OK !=
+  FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
           plugin->reserves_in_insert (plugin->cls,
                                       session,
                                       &reserve_pub,
@@ -1543,7 +1546,7 @@ run (void *cls)
                          value.value,
                          value.fraction,
                          value.currency));
-  FAILIF (GNUNET_OK !=
+  FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
           plugin->reserves_in_insert (plugin->cls,
                                       session,
                                       &reserve_pub,
@@ -1810,6 +1813,9 @@ run (void *cls)
                                      session,
                                      &deposit_cb,
                                      &deposit));
+  FAILIF (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
+          plugin->commit (plugin->cls,
+                          session));
   FAILIF (GNUNET_OK !=
           plugin->start (plugin->cls,
                          session));
diff --git a/src/include/taler_exchangedb_plugin.h b/src/include/taler_exchangedb_plugin.h
index 7b7fffef2..7c245e224 100644
--- a/src/include/taler_exchangedb_plugin.h
+++ b/src/include/taler_exchangedb_plugin.h
@@ -1190,11 +1190,9 @@ struct TALER_EXCHANGEDB_Plugin
    * @param sender_account_details information about the sender's bank account
    * @param wire_reference unique reference identifying the wire transfer (binary blob)
    * @param wire_reference_size number of bytes in @a wire_reference
-   * @return #GNUNET_OK upon success; #GNUNET_NO if the given
-   *         @a details are already known for this @a reserve_pub,
-   *         #GNUNET_SYSERR upon failures (DB error, incompatible currency)
+   * @return transaction status code
    */
-  int
+  enum GNUNET_DB_QueryStatus
   (*reserves_in_insert) (void *cls,
                          struct TALER_EXCHANGEDB_Session *db,
                          const struct TALER_ReservePublicKeyP *reserve_pub,