Merge branch 'master' of git+ssh://git.taler.net/exchange

src/exchangedb/.gitignore

@@ -9,3 +9,4 @@ shard-drop0001.sqltest-exchangedb-by-j-postgres
 test-exchangedb-by-j-postgres
 perf-exchangedb-reserves-in-insert-postgres
 exchange-0002.sql
+procedures.sql

src/exchangedb/Makefile.am

@@ -243,6 +243,7 @@ libtaler_plugin_exchangedb_postgres_la_SOURCES = \
   pg_select_account_merges_above_serial_id.h pg_select_account_merges_above_serial_id.c \
   pg_select_all_purse_decisions_above_serial_id.h pg_select_all_purse_decisions_above_serial_id.c \
   pg_batch_reserves_in_insert.h pg_batch_reserves_in_insert.c \
+  pg_batch2_reserves_in_insert.h pg_batch2_reserves_in_insert.c \
   pg_select_reserve_open_above_serial_id.c pg_select_reserve_open_above_serial_id.h
 libtaler_plugin_exchangedb_postgres_la_LIBADD = \
   $(LTLIBINTL)
@@ -280,12 +281,16 @@ check_PROGRAMS = \
   test-exchangedb-postgres \
   bench-db-postgres\
   perf-exchangedb-reserves-in-insert-postgres\
-  test-exchangedb-by-j-postgres
+  test-exchangedb-by-j-postgres\
+  test-exchangedb-batch-reserves-in-insert-postgres
 AM_TESTS_ENVIRONMENT=export TALER_PREFIX=$${TALER_PREFIX:-@libdir@};export PATH=$${TALER_PREFIX:-@prefix@}/bin:$$PATH;
 TESTS = \
   test-exchangedb-postgres\
   test-exchangedb-by-j-postgres\
-  perf-exchangedb-reserves-in-insert-postgres
+  perf-exchangedb-reserves-in-insert-postgres\
+  test-exchangedb-batch-reserves-in-insert-postgres
+
+
 test_exchangedb_postgres_SOURCES = \
   test_exchangedb.c
 test_exchangedb_postgres_LDADD = \
@@ -333,5 +338,29 @@ bench_db_postgres_LDADD = \
   -lgnunetutil \
   $(XLIB)
 
+
+test_exchangedb_batch_reserves_in_insert_postgres_SOURCES = \
+  test_exchangedb_batch_reserves_in_insert.c
+test_exchangedb_batch_reserves_in_insert_postgres_LDADD = \
+  libtalerexchangedb.la \
+  $(top_builddir)/src/json/libtalerjson.la \
+  $(top_builddir)/src/util/libtalerutil.la \
+  $(top_builddir)/src/pq/libtalerpq.la \
+  -ljansson \
+  -lgnunetjson \
+  -lgnunetutil \
+  $(XLIB)
+
+bench_db_postgres_SOURCES = \
+  bench_db.c
+bench_db_postgres_LDADD = \
+  libtalerexchangedb.la \
+  $(top_builddir)/src/util/libtalerutil.la \
+  $(top_builddir)/src/pq/libtalerpq.la \
+  -lgnunetpq \
+  -lgnunetutil \
+  $(XLIB)
+
+
 EXTRA_test_exchangedb_postgres_DEPENDENCIES = \
   libtaler_plugin_exchangedb_postgres.la

src/exchangedb/pg_batch2_reserves_in_insert.c

@@ -0,0 +1,282 @@
+/*
+   This file is part of TALER
+   Copyright (C) 2022 Taler Systems SA
+
+   TALER is free software; you can redistribute it and/or modify it under the
+   terms of the GNU General Public License as published by the Free Software
+   Foundation; either version 3, or (at your option) any later version.
+
+   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+   A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file exchangedb/pg_batch_reserves_in_insert.c
+ * @brief Implementation of the reserves_in_insert function for Postgres
+ * @author Joseph Xu
+ */
+#include "platform.h"
+#include "taler_error_codes.h"
+#include "taler_dbevents.h"
+#include "taler_pq_lib.h"
+#include "pg_batch_reserves_in_insert.h"
+#include "pg_helper.h"
+#include "pg_start.h"
+#include "pg_rollback.h"
+#include "pg_start_read_committed.h"
+#include "pg_commit.h"
+#include "pg_reserves_get.h"
+#include "pg_reserves_update.h"
+#include "pg_setup_wire_target.h"
+#include "pg_event_notify.h"
+#include "pg_preflight.h"
+
+/**
+ * Generate event notification for the reserve change.
+ *
+ * @param reserve_pub reserve to notfiy on
+ */
+static char *
+compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
+{
+  struct TALER_ReserveEventP rep = {
+    .header.size = htons (sizeof (rep)),
+    .header.type = htons (TALER_DBEVENT_EXCHANGE_RESERVE_INCOMING),
+    .reserve_pub = *reserve_pub
+  };
+
+  return GNUNET_PG_get_event_notify_channel (&rep.header);
+}
+
+
+enum GNUNET_DB_QueryStatus
+TEH_PG_batch2_reserves_in_insert (void *cls,
+                                 const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
+                                 unsigned int reserves_length,
+                                 enum GNUNET_DB_QueryStatus *results)
+{
+  struct PostgresClosure *pg = cls;
+  enum GNUNET_DB_QueryStatus qs1;
+  struct GNUNET_TIME_Timestamp expiry;
+  struct GNUNET_TIME_Timestamp gc;
+  struct TALER_PaytoHashP h_payto;
+  uint64_t reserve_uuid;
+  uint64_t reserve_uuid2;
+  bool conflicted;
+  bool conflicted2;
+  bool transaction_duplicate;
+  bool transaction_duplicate2;
+  bool need_update = false;
+  bool need_update2 = false;
+  struct GNUNET_TIME_Timestamp reserve_expiration
+    = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
+  bool conflicts[reserves_length];
+  bool conflicts2[reserves_length];
+  char *notify_s[reserves_length];
+
+  if (GNUNET_OK !=
+      TEH_PG_preflight (pg))
+  {
+    GNUNET_break (0);
+    return GNUNET_DB_STATUS_HARD_ERROR;
+  }
+  PREPARE (pg,
+           "reserve_create",
+           "SELECT "
+           "out_reserve_found AS conflicted"
+           ",out_reserve_found2 AS conflicted2"
+           ",transaction_duplicate"
+           ",transaction_duplicate2"
+           ",ruuid AS reserve_uuid"
+           ",ruuid2 AS reserve_uuid2"
+           " FROM batch2_reserves_insert"
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21);");
+  expiry = GNUNET_TIME_absolute_to_timestamp (
+    GNUNET_TIME_absolute_add (reserves->execution_time.abs_time,
+                              pg->idle_reserve_expiration_time));
+  gc = GNUNET_TIME_absolute_to_timestamp (
+    GNUNET_TIME_absolute_add (GNUNET_TIME_absolute_get (),
+                              pg->legal_reserve_expiration_time));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Creating reserve %s with expiration in %s\n",
+              TALER_B2S (&(reserves->reserve_pub)),
+              GNUNET_STRINGS_relative_time_to_string (
+                pg->idle_reserve_expiration_time,
+                GNUNET_NO));
+
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start_read_committed(pg,
+                                   "READ_COMMITED"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
+  /* Optimistically assume this is a new reserve, create balance for the first
+     time; we do this before adding the actual transaction to "reserves_in",
+     as for a new reserve it can't be a duplicate 'add' operation, and as
+     the 'add' operation needs the reserve entry as a foreign key. */
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+    notify_s[i] = compute_notify_on_reserve (&reserve->reserve_pub);
+  }
+
+  for (unsigned int i=0;i<(reserves_length & ~1);i+=2)
+  {
+    const struct TALER_EXCHANGEDB_ReserveInInfo *reserve0 = &reserves[i];
+    const struct TALER_EXCHANGEDB_ReserveInInfo *reserve1 = &reserves[i+1];
+    struct GNUNET_PQ_QueryParam params[] = {
+      GNUNET_PQ_query_param_auto_from_type (&reserve0->reserve_pub),
+      GNUNET_PQ_query_param_timestamp (&expiry),
+      GNUNET_PQ_query_param_timestamp (&gc),
+      GNUNET_PQ_query_param_uint64 (&reserve0->wire_reference),
+      TALER_PQ_query_param_amount (&reserve0->balance),
+      GNUNET_PQ_query_param_string (reserve0->exchange_account_name),
+      GNUNET_PQ_query_param_timestamp (&reserve0->execution_time),
+      GNUNET_PQ_query_param_auto_from_type (&h_payto),
+      GNUNET_PQ_query_param_string (reserve0->sender_account_details),
+      GNUNET_PQ_query_param_timestamp (&reserve_expiration),
+      GNUNET_PQ_query_param_string (notify_s[i]),
+      GNUNET_PQ_query_param_auto_from_type (&reserve1->reserve_pub),
+      GNUNET_PQ_query_param_uint64 (&reserve1->wire_reference),
+      TALER_PQ_query_param_amount (&reserve1->balance),
+      GNUNET_PQ_query_param_string (reserve1->exchange_account_name),
+      GNUNET_PQ_query_param_timestamp (&reserve1->execution_time),
+      GNUNET_PQ_query_param_auto_from_type (&h_payto),
+      GNUNET_PQ_query_param_string (reserve1->sender_account_details),
+      GNUNET_PQ_query_param_timestamp (&reserve_expiration),
+
+      GNUNET_PQ_query_param_end
+    };
+
+    struct GNUNET_PQ_ResultSpec rs[] = {
+      GNUNET_PQ_result_spec_bool ("conflicted",
+                                  &conflicted),
+      GNUNET_PQ_result_spec_bool ("conflicted2",
+                                  &conflicted2),
+      GNUNET_PQ_result_spec_bool ("transaction_duplicate",
+                                  &transaction_duplicate),
+      GNUNET_PQ_result_spec_bool ("transaction_duplicate2",
+                                  &transaction_duplicate2),
+      GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
+                                    &reserve_uuid),
+      GNUNET_PQ_result_spec_uint64 ("reserve_uuid2",
+                                    &reserve_uuid2),
+      GNUNET_PQ_result_spec_end
+    };
+
+    TALER_payto_hash (reserve0->sender_account_details,
+                      &h_payto);
+    TALER_payto_hash (reserve1->sender_account_details,
+                      &h_payto);
+
+    /* Note: query uses 'on conflict do nothing' */
+    qs1 = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
+                                                    "reserve_create",
+                                                    params,
+                                                    rs);
+
+    if (qs1 < 0)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Failed to create reserves (%d)\n",
+                  qs1);
+      return qs1;
+    }
+    if (reserves_length & 1)
+    {
+      const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[reserves_length-1];
+      // single insert logic here
+    }
+   GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs1);
+   results[i] = (transaction_duplicate)
+      ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
+      : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
+   conflicts[i] = conflicted;
+   conflicts2[i] = conflicted2;
+   //    fprintf(stdout, "%d", conflicts[i]);
+   // fprintf(stdout, "%d", conflicts2[i]);
+   if ((!conflicts[i] && transaction_duplicate) ||(!conflicts2[i] && transaction_duplicate2))
+   {
+     GNUNET_break (0);
+     TEH_PG_rollback (pg);
+     return GNUNET_DB_STATUS_HARD_ERROR;
+   }
+   need_update |= conflicted |= conflicted2;
+  }
+  // commit
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+  if (!need_update)
+    goto exit;
+  // begin serializable
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start(pg,
+                     "reserve-insert-continued"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
+
+  enum GNUNET_DB_QueryStatus qs2;
+  PREPARE (pg,
+           "reserves_in_add_transaction",
+           "SELECT batch_reserves_update"
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9);");
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    if (! conflicts[i])
+      continue;
+    {
+      const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+      struct GNUNET_PQ_QueryParam params[] = {
+        GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
+        GNUNET_PQ_query_param_timestamp (&expiry),
+        GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
+        TALER_PQ_query_param_amount (&reserve->balance),
+        GNUNET_PQ_query_param_string (reserve->exchange_account_name),
+        GNUNET_PQ_query_param_bool (conflicted),
+        GNUNET_PQ_query_param_auto_from_type (&h_payto),
+        GNUNET_PQ_query_param_string (notify_s[i]),
+        GNUNET_PQ_query_param_end
+      };
+
+      qs2 = GNUNET_PQ_eval_prepared_non_select (pg->conn,
+                                                "reserves_in_add_transaction",
+                                                params);
+      if (qs2<0)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "Failed to update reserves (%d)\n",
+                    qs2);
+        return qs2;
+      }
+    }
+  }
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+ exit:
+  for (unsigned int i=0;i<reserves_length;i++)
+    GNUNET_free (notify_s[i]);
+
+  return reserves_length;
+}

src/exchangedb/pg_batch2_reserves_in_insert.h

@@ -0,0 +1,33 @@
+/*
+   This file is part of TALER
+   Copyright (C) 2022 Taler Systems SA
+
+   TALER is free software; you can redistribute it and/or modify it under the
+   terms of the GNU General Public License as published by the Free Software
+   Foundation; either version 3, or (at your option) any later version.
+
+   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+   A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file exchangedb/pg_batch2_reserves_in_insert.h
+ * @brief implementation of the batch2_reserves_in_insert function for Postgres
+ * @author Christian Grothoff
+ */
+#ifndef PG_BATCH2_RESERVES_IN_INSERT_H
+#define PG_BATCH2_RESERVES_IN_INSERT_H
+
+#include "taler_util.h"
+#include "taler_json_lib.h"
+#include "taler_exchangedb_plugin.h"
+enum GNUNET_DB_QueryStatus
+TEH_PG_batch2_reserves_in_insert (void *cls,
+                                 const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
+                                 unsigned int reserves_length,
+                                 enum GNUNET_DB_QueryStatus *results);
+
+#endif

src/exchangedb/pg_batch4_reserves_in_insert.c

@@ -0,0 +1,266 @@
+/*
+   This file is part of TALER
+   Copyright (C) 2022 Taler Systems SA
+
+   TALER is free software; you can redistribute it and/or modify it under the
+   terms of the GNU General Public License as published by the Free Software
+   Foundation; either version 3, or (at your option) any later version.
+
+   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+   A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file exchangedb/pg_batch4_reserves_in_insert.c
+ * @brief Implementation of the reserves_in_insert function for Postgres
+ * @author Joseph Xu
+ */
+#include "platform.h"
+#include "taler_error_codes.h"
+#include "taler_dbevents.h"
+#include "taler_pq_lib.h"
+#include "pg_batch_reserves_in_insert.h"
+#include "pg_helper.h"
+#include "pg_start.h"
+#include "pg_rollback.h"
+#include "pg_start_read_committed.h"
+#include "pg_commit.h"
+#include "pg_reserves_get.h"
+#include "pg_reserves_update.h"
+#include "pg_setup_wire_target.h"
+#include "pg_event_notify.h"
+#include "pg_preflight.h"
+
+/**
+ * Generate event notification for the reserve change.
+ *
+ * @param reserve_pub reserve to notfiy on
+ */
+static char *
+compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
+{
+  struct TALER_ReserveEventP rep = {
+    .header.size = htons (sizeof (rep)),
+    .header.type = htons (TALER_DBEVENT_EXCHANGE_RESERVE_INCOMING),
+    .reserve_pub = *reserve_pub
+  };
+
+  return GNUNET_PG_get_event_notify_channel (&rep.header);
+}
+
+
+enum GNUNET_DB_QueryStatus
+TEH_PG_batch4_reserves_in_insert (void *cls,
+                                 const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
+                                 unsigned int reserves_length,
+                                 enum GNUNET_DB_QueryStatus *results)
+{
+  struct PostgresClosure *pg = cls;
+  enum GNUNET_DB_QueryStatus qs1;
+  struct GNUNET_TIME_Timestamp expiry;
+  struct GNUNET_TIME_Timestamp gc;
+  struct TALER_PaytoHashP h_payto;
+  uint64_t reserve_uuid;
+  bool conflicted;
+  bool conflicted2;
+  bool transaction_duplicate;
+  bool transaction_duplicate2;
+  bool need_update = false;
+  bool need_update2 = false;
+  struct GNUNET_TIME_Timestamp reserve_expiration
+    = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
+  bool conflicts[reserves_length];
+  bool conflicts2[reserves_length];
+  char *notify_s[reserves_length];
+
+  if (GNUNET_OK !=
+      TEH_PG_preflight (pg))
+  {
+    GNUNET_break (0);
+    return GNUNET_DB_STATUS_HARD_ERROR;
+  }
+  PREPARE (pg,
+           "reserve_create",
+           "SELECT "
+           "out_reserve_found AS conflicted"
+           ",out_reserve_found2 AS conflicted2"
+           ",transaction_duplicate"
+           ",transaction_duplicate2"
+           ",ruuid AS reserve_uuid"
+           " FROM batch2_reserves_insert"
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21);");
+  expiry = GNUNET_TIME_absolute_to_timestamp (
+    GNUNET_TIME_absolute_add (reserves->execution_time.abs_time,
+                              pg->idle_reserve_expiration_time));
+  gc = GNUNET_TIME_absolute_to_timestamp (
+    GNUNET_TIME_absolute_add (GNUNET_TIME_absolute_get (),
+                              pg->legal_reserve_expiration_time));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Creating reserve %s with expiration in %s\n",
+              TALER_B2S (&(reserves->reserve_pub)),
+              GNUNET_STRINGS_relative_time_to_string (
+                pg->idle_reserve_expiration_time,
+                GNUNET_NO));
+
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start_read_committed(pg,
+                                   "READ_COMMITED"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
+  /* Optimistically assume this is a new reserve, create balance for the first
+     time; we do this before adding the actual transaction to "reserves_in",
+     as for a new reserve it can't be a duplicate 'add' operation, and as
+     the 'add' operation needs the reserve entry as a foreign key. */
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+    notify_s[i] = compute_notify_on_reserve (&reserve->reserve_pub);
+  }
+
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+    struct GNUNET_PQ_QueryParam params[] = {
+      GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
+      GNUNET_PQ_query_param_timestamp (&expiry),
+      GNUNET_PQ_query_param_timestamp (&gc),
+      GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
+      TALER_PQ_query_param_amount (&reserve->balance),
+      GNUNET_PQ_query_param_string (reserve->exchange_account_name),
+      GNUNET_PQ_query_param_timestamp (&reserve->execution_time),
+      GNUNET_PQ_query_param_auto_from_type (&h_payto),
+      GNUNET_PQ_query_param_string (reserve->sender_account_details),
+      GNUNET_PQ_query_param_timestamp (&reserve_expiration),
+      GNUNET_PQ_query_param_string (notify_s[i]),
+      GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
+      GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
+      TALER_PQ_query_param_amount (&reserve->balance),
+      GNUNET_PQ_query_param_string (reserve->exchange_account_name),
+      GNUNET_PQ_query_param_timestamp (&reserve->execution_time),
+      GNUNET_PQ_query_param_auto_from_type (&h_payto),
+      GNUNET_PQ_query_param_string (reserve->sender_account_details),
+      GNUNET_PQ_query_param_timestamp (&reserve_expiration),
+      GNUNET_PQ_query_param_end
+    };
+
+    struct GNUNET_PQ_ResultSpec rs[] = {
+      GNUNET_PQ_result_spec_bool ("conflicted",
+                                  &conflicted),
+            GNUNET_PQ_result_spec_bool ("conflicted2",
+                                  &conflicted2),
+      GNUNET_PQ_result_spec_bool ("transaction_duplicate",
+                                  &transaction_duplicate),
+      GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
+                                    &reserve_uuid),
+      GNUNET_PQ_result_spec_end
+    };
+
+    TALER_payto_hash (reserve->sender_account_details,
+                      &h_payto);
+
+    /* Note: query uses 'on conflict do nothing' */
+    qs1 = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
+                                                    "reserve_create",
+                                                    params,
+                                                    rs);
+
+    if (qs1 < 0)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Failed to create reserves (%d)\n",
+                  qs1);
+      return qs1;
+    }
+   GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs1);
+   results[i] = (transaction_duplicate)
+      ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
+      : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
+   conflicts[i] = conflicted;
+   conflicts2[i] = conflicted2;
+   //  fprintf(stdout, "%d", conflicts[i]);
+   if (!conflicts[i] && !conflicts2[i]&& transaction_duplicate)
+   {
+     GNUNET_break (0);
+     TEH_PG_rollback (pg);
+     return GNUNET_DB_STATUS_HARD_ERROR;
+   }
+   need_update |= conflicted |= conflicted2;
+  }
+  // commit
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+  if (!need_update)
+    goto exit;
+  // begin serializable
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start(pg,
+                     "reserve-insert-continued"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
+
+  enum GNUNET_DB_QueryStatus qs2;
+  PREPARE (pg,
+           "reserves_in_add_transaction",
+           "SELECT batch_reserves_update"
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9);");
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    if (! conflicts[i])
+      continue;
+    {
+      const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+      struct GNUNET_PQ_QueryParam params[] = {
+        GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
+        GNUNET_PQ_query_param_timestamp (&expiry),
+        GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
+        TALER_PQ_query_param_amount (&reserve->balance),
+        GNUNET_PQ_query_param_string (reserve->exchange_account_name),
+        GNUNET_PQ_query_param_bool (conflicted),
+        GNUNET_PQ_query_param_auto_from_type (&h_payto),
+        GNUNET_PQ_query_param_string (notify_s[i]),
+        GNUNET_PQ_query_param_end
+      };
+
+      qs2 = GNUNET_PQ_eval_prepared_non_select (pg->conn,
+                                                "reserves_in_add_transaction",
+                                                params);
+      if (qs2<0)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "Failed to update reserves (%d)\n",
+                    qs2);
+        return qs2;
+      }
+    }
+  }
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+ exit:
+  for (unsigned int i=0;i<reserves_length;i++)
+    GNUNET_free (notify_s[i]);
+
+  return reserves_length;
+}

src/exchangedb/pg_batch4_reserves_in_insert.h

@@ -0,0 +1,34 @@
+/*
+   This file is part of TALER
+   Copyright (C) 2022 Taler Systems SA
+
+   TALER is free software; you can redistribute it and/or modify it under the
+   terms of the GNU General Public License as published by the Free Software
+   Foundation; either version 3, or (at your option) any later version.
+
+   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+   A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file exchangedb/pg_batch4_reserves_in_insert.h
+ * @brief implementation of the batch4_reserves_in_insert function for Postgres
+ * @author XU Joseph
+ */
+#ifndef PG_BATCH4_RESERVES_IN_INSERT_H
+#define PG_BATCH4_RESERVES_IN_INSERT_H
+
+#include "taler_util.h"
+#include "taler_json_lib.h"
+#include "taler_exchangedb_plugin.h"
+
+enum GNUNET_DB_QueryStatus
+TEH_PG_batch4_reserves_in_insert (void *cls,
+                                 const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
+                                 unsigned int reserves_length,
+                                  enum GNUNET_DB_QueryStatus *results);
+
+#endif

src/exchangedb/pg_batch_reserves_in_insert.c

@@ -16,7 +16,7 @@
 /**
  * @file exchangedb/pg_batch_reserves_in_insert.c
  * @brief Implementation of the reserves_in_insert function for Postgres
- * @author JOSEPHxu
+ * @author Joseph Xu
  */
 #include "platform.h"
 #include "taler_error_codes.h"
@@ -32,18 +32,16 @@
 #include "pg_reserves_update.h"
 #include "pg_setup_wire_target.h"
 #include "pg_event_notify.h"
+#include "pg_preflight.h"
 
 
 /**
- * Generate event notification for the reserve
+ * Generate event notification for the reserve change.
- * change.
  *
- * @param pg plugin state
  * @param reserve_pub reserve to notfiy on
  */
-static void
+static char *
-notify_on_reserve (struct PostgresClosure *pg,
+compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
-                   const struct TALER_ReservePublicKeyP *reserve_pub)
 {
   struct TALER_ReserveEventP rep = {
     .header.size = htons (sizeof (rep)),
@@ -51,12 +49,7 @@ notify_on_reserve (struct PostgresClosure *pg,
     .reserve_pub = *reserve_pub
   };
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+  return GNUNET_PG_get_event_notify_channel (&rep.header);
-              "Notifying on reserve!\n");
-  TEH_PG_event_notify (pg,
-                       &rep.header,
-                       NULL,
-                       0);
 }
 
 
@@ -75,17 +68,26 @@ TEH_PG_batch_reserves_in_insert (void *cls,
   uint64_t reserve_uuid;
   bool conflicted;
   bool transaction_duplicate;
+  bool need_update = false;
   struct GNUNET_TIME_Timestamp reserve_expiration
     = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
+  bool conflicts[reserves_length];
+  char *notify_s[reserves_length];
 
+  if (GNUNET_OK !=
+      TEH_PG_preflight (pg))
+  {
+    GNUNET_break (0);
+    return GNUNET_DB_STATUS_HARD_ERROR;
+  }
   PREPARE (pg,
            "reserve_create",
            "SELECT "
            "out_reserve_found AS conflicted"
            ",transaction_duplicate"
            ",ruuid AS reserve_uuid"
-           " FROM exchange_do_batch_reserves_in"
+           " FROM batch_reserves_insert"
-           " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11);");
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12);");
   expiry = GNUNET_TIME_absolute_to_timestamp (
     GNUNET_TIME_absolute_add (reserves->execution_time.abs_time,
                               pg->idle_reserve_expiration_time));
@@ -98,54 +100,152 @@ TEH_PG_batch_reserves_in_insert (void *cls,
               GNUNET_STRINGS_relative_time_to_string (
                 pg->idle_reserve_expiration_time,
                 GNUNET_NO));
+
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start_read_committed(pg,
+                                   "READ_COMMITED"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
   /* Optimistically assume this is a new reserve, create balance for the first
      time; we do this before adding the actual transaction to "reserves_in",
      as for a new reserve it can't be a duplicate 'add' operation, and as
      the 'add' operation needs the reserve entry as a foreign key. */
   for (unsigned int i = 0; i<reserves_length; i++)
+  {
+    const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+    notify_s[i] = compute_notify_on_reserve (&reserve->reserve_pub);
+  }
+
+  for (unsigned int i=0;i<reserves_length;i++)
   {
     const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
     struct GNUNET_PQ_QueryParam params[] = {
-      GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub), /*$1*/
+      GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
-      GNUNET_PQ_query_param_timestamp (&expiry),  /*$4*/
+      GNUNET_PQ_query_param_timestamp (&expiry),
-      GNUNET_PQ_query_param_timestamp (&gc),  /*$5*/
+      GNUNET_PQ_query_param_timestamp (&gc),
-      GNUNET_PQ_query_param_uint64 (&reserve->wire_reference), /*6*/
+      GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
-      TALER_PQ_query_param_amount (&reserve->balance), /*7+8*/
+      TALER_PQ_query_param_amount (&reserve->balance),
-      GNUNET_PQ_query_param_string (reserve->exchange_account_name), /*9*/
+      GNUNET_PQ_query_param_string (reserve->exchange_account_name),
-      GNUNET_PQ_query_param_timestamp (&reserve->execution_time), /*10*/
+      GNUNET_PQ_query_param_timestamp (&reserve->execution_time),
-      GNUNET_PQ_query_param_auto_from_type (&h_payto), /*11*/
+      GNUNET_PQ_query_param_auto_from_type (&h_payto),
-      GNUNET_PQ_query_param_string (reserve->sender_account_details),/*12*/
+      GNUNET_PQ_query_param_string (reserve->sender_account_details),
-      GNUNET_PQ_query_param_timestamp (&reserve_expiration),/*13*/
+      GNUNET_PQ_query_param_timestamp (&reserve_expiration),
+      GNUNET_PQ_query_param_string (notify_s[i]),
       GNUNET_PQ_query_param_end
     };
-    /* We should get all our results into results[]*/
+
     struct GNUNET_PQ_ResultSpec rs[] = {
-      GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
-                                    &reserve_uuid),
       GNUNET_PQ_result_spec_bool ("conflicted",
                                   &conflicted),
       GNUNET_PQ_result_spec_bool ("transaction_duplicate",
                                   &transaction_duplicate),
+      GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
+                                    &reserve_uuid),
       GNUNET_PQ_result_spec_end
     };
 
     TALER_payto_hash (reserve->sender_account_details,
                       &h_payto);
+
     /* Note: query uses 'on conflict do nothing' */
     qs1 = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
                                                     "reserve_create",
                                                     params,
                                                     rs);
+
     if (qs1 < 0)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Failed to create reserves (%d)\n",
+                  qs1);
       return qs1;
-    notify_on_reserve (pg,
+    }
-                       &reserve->reserve_pub);
+   GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs1);
-    GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs1);
+   results[i] = (transaction_duplicate)
-    results[i] = (transaction_duplicate)
       ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
       : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
-    if ( (! conflicted) && transaction_duplicate)
+   conflicts[i] = conflicted;
-      TEH_PG_rollback (pg);
+   //  fprintf(stdout, "%d", conflicts[i]);
+   if (!conflicts[i] && transaction_duplicate)
+   {
+     GNUNET_break (0);
+     TEH_PG_rollback (pg);
+     return GNUNET_DB_STATUS_HARD_ERROR;
+   }
+   need_update |= conflicted;
   }
+  // commit
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+  if (!need_update)
+    goto exit;
+  // begin serializable
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start(pg,
+                     "reserve-insert-continued"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
+
+  enum GNUNET_DB_QueryStatus qs2;
+  PREPARE (pg,
+           "reserves_in_add_transaction",
+           "SELECT batch_reserves_update"
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9);");
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    if (! conflicts[i])
+      continue;
+    {
+      const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+      struct GNUNET_PQ_QueryParam params[] = {
+        GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
+        GNUNET_PQ_query_param_timestamp (&expiry),
+        GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
+        TALER_PQ_query_param_amount (&reserve->balance),
+        GNUNET_PQ_query_param_string (reserve->exchange_account_name),
+        GNUNET_PQ_query_param_bool (conflicted),
+        GNUNET_PQ_query_param_auto_from_type (&h_payto),
+        GNUNET_PQ_query_param_string (notify_s[i]),
+        GNUNET_PQ_query_param_end
+      };
+
+      qs2 = GNUNET_PQ_eval_prepared_non_select (pg->conn,
+                                                "reserves_in_add_transaction",
+                                                params);
+      if (qs2<0)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "Failed to update reserves (%d)\n",
+                    qs2);
+        return qs2;
+      }
+    }
+  }
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+ exit:
+  for (unsigned int i=0;i<reserves_length;i++)
+    GNUNET_free (notify_s[i]);
+
   return reserves_length;
 }

src/exchangedb/pg_get_link_data.c

@@ -178,25 +178,55 @@ TEH_PG_get_link_data (void *cls,
   enum GNUNET_DB_QueryStatus qs;
   struct LinkDataContext ldctx;
 
-  PREPARE (pg,
+  if (NULL == getenv ("NEW_LOGIC"))
-           "get_link",
+  {
-           "SELECT "
+    PREPARE (pg,
-           " tp.transfer_pub"
+             "get_link",
-           ",denoms.denom_pub"
+             "SELECT "
-           ",rrc.ev_sig"
+             " tp.transfer_pub"
-           ",rrc.ewv"
+             ",denoms.denom_pub"
-           ",rrc.link_sig"
+             ",rrc.ev_sig"
-           ",rrc.freshcoin_index"
+             ",rrc.ewv"
-           ",rrc.coin_ev"
+             ",rrc.link_sig"
-           " FROM refresh_commitments"
+             ",rrc.freshcoin_index"
-           "     JOIN refresh_revealed_coins rrc"
+             ",rrc.coin_ev"
-           "       USING (melt_serial_id)"
+             " FROM refresh_commitments"
-           "     JOIN refresh_transfer_keys tp"
+             "     JOIN refresh_revealed_coins rrc"
-           "       USING (melt_serial_id)"
+             "       USING (melt_serial_id)"
-           "     JOIN denominations denoms"
+             "     JOIN refresh_transfer_keys tp"
-           "       ON (rrc.denominations_serial = denoms.denominations_serial)"
+             "       USING (melt_serial_id)"
-           " WHERE old_coin_pub=$1"
+             "     JOIN denominations denoms"
-           " ORDER BY tp.transfer_pub, rrc.freshcoin_index ASC");
+             "       ON (rrc.denominations_serial = denoms.denominations_serial)"
+             " WHERE old_coin_pub=$1"
+             " ORDER BY tp.transfer_pub, rrc.freshcoin_index ASC");
+  }
+
+  else
+  {
+    PREPARE (pg,
+             "get_link",
+             "WITH rc AS MATERIALIZED ("
+             "SELECT"
+             "* FROM refresh_commitments"
+             "WHERE old_coin_pub=$1"
+             ")"
+             "SELECT "
+             " tp.transfer_pub"
+             ",denoms.denom_pub"
+             ",rrc.ev_sig"
+             ",rrc.ewv"
+             ",rrc.link_sig"
+             ",rrc.freshcoin_index"
+             ",rrc.coin_ev"
+             " FROM refresh_revealed_coins rrc"
+             "       USING (melt_serial_id)"
+             "     JOIN refresh_transfer_keys tp"
+             "       USING (melt_serial_id)"
+             "     JOIN denominations denoms"
+             "       USING (denominations_serial)"
+             " ORDER BY tp.transfer_pub, rrc.freshcoin_index ASC");
+  }
+
   ldctx.ldc = ldc;
   ldctx.ldc_cls = ldc_cls;
   ldctx.last = NULL;

src/exchangedb/pg_select_refunds_by_coin.c

@@ -138,15 +138,19 @@ TEH_PG_select_refunds_by_coin (
     // FIXME-Joseph
     PREPARE (pg,
              "get_refunds_by_coin_and_contract",
+             "WITH rc AS MATERIALIZED("
              "SELECT"
-             " ref.amount_with_fee_val"
+             " * FROM refunds ref"
-             ",ref.amount_with_fee_frac"
+             "WHERE ref.coin_pub=$1"
-             " FROM refunds ref"
+                "AND dep.merchant_pub=$2"
-             " JOIN deposits dep"
+                "AND dep.h_contract_terms=$3"
-             "   USING (coin_pub,deposit_serial_id)"
+             ")"
-             " WHERE ref.coin_pub=$1"
+             "SELECT"
-             "   AND dep.merchant_pub=$2"
+             "   ref.amount_with_fee_val"
-             "   AND dep.h_contract_terms=$3;");
+             "  ,ref.amount_with_fee_frac"
+             "FROM deposits dep"
+             "JOIN rc"
+             "   USING (coin_pub,deposit_serial_id)");
   }
   qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
                                              "get_refunds_by_coin_and_contract",

src/exchangedb/test_exchangedb_batch_reserves_in_insert.c

@@ -0,0 +1,201 @@
+/*
+  This file is part of TALER
+  Copyright (C) 2014-2022 Taler Systems SA
+
+  TALER is free software; you can redistribute it and/or modify it under the
+  terms of the GNU General Public License as published by the Free Software
+  Foundation; either version 3, or (at your option) any later version.
+
+  TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+  A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License along with
+  TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+*/
+/**
+ * @file exchangedb/test_exchangedb_by_j.c
+ * @brief test cases for DB interaction functions
+ * @author Joseph Xu
+ */
+#include "platform.h"
+#include "taler_exchangedb_lib.h"
+#include "taler_json_lib.h"
+#include "taler_exchangedb_plugin.h"
+
+/**o
+ * Global result from the testcase.
+ */
+static int result;
+
+/**
+ * Report line of error if @a cond is true, and jump to label "drop".
+ */
+#define FAILIF(cond)                            \
+  do {                                          \
+      if (! (cond)) {break;}                    \
+    GNUNET_break (0);                           \
+    goto drop;                                  \
+  } while (0)
+
+
+/**
+ * Initializes @a ptr with random data.
+ */
+#define RND_BLK(ptr)                                                    \
+  GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_WEAK, ptr, sizeof (*ptr))
+
+/**
+ * Initializes @a ptr with zeros.
+ */
+#define ZR_BLK(ptr) \
+  memset (ptr, 0, sizeof (*ptr))
+
+
+/**
+ * Currency we use.  Must match test-exchange-db-*.conf.
+ */
+#define CURRENCY "EUR"
+
+/**
+ * Database plugin under test.
+ */
+static struct TALER_EXCHANGEDB_Plugin *plugin;
+
+
+/**
+ * Main function that will be run by the scheduler.
+ *
+ * @param cls closure with config
+ */
+static void
+run (void *cls)
+{
+  struct GNUNET_CONFIGURATION_Handle *cfg = cls;
+  const uint32_t num_partitions = 10;
+
+  if (NULL ==
+      (plugin = TALER_EXCHANGEDB_plugin_load (cfg)))
+  {
+    GNUNET_break (0);
+    result = 77;
+    return;
+  }
+  (void) plugin->drop_tables (plugin->cls);
+  if (GNUNET_OK !=
+      plugin->create_tables (plugin->cls))
+  {
+    GNUNET_break (0);
+    result = 77;
+    goto cleanup;
+  }
+  if (GNUNET_OK !=
+      plugin->setup_partitions (plugin->cls,
+                                num_partitions))
+  {
+    GNUNET_break (0);
+    result = 77;
+    goto cleanup;
+  }
+
+  for (unsigned int i = 0; i< 8; i++)
+  {
+    static unsigned int batches[] = {1, 1,0, 2, 4, 16, 64, 256};
+    const char *sndr = "payto://x-taler-bank/localhost:8080/1";
+    struct TALER_Amount value;
+    unsigned int batch_size = batches[i];
+    struct GNUNET_TIME_Absolute now;
+    struct GNUNET_TIME_Timestamp ts;
+    struct GNUNET_TIME_Relative duration;
+    struct TALER_EXCHANGEDB_ReserveInInfo reserves[batch_size];
+    enum GNUNET_DB_QueryStatus results[batch_size];
+    GNUNET_assert (GNUNET_OK ==
+                   TALER_string_to_amount (CURRENCY ":1.000010",
+                                           &value));
+    now = GNUNET_TIME_absolute_get ();
+    ts = GNUNET_TIME_timestamp_get ();
+    for (unsigned int r=0;r<10;r++)
+    {
+
+      for (unsigned int k = 0; k<batch_size; k++)
+      {
+        RND_BLK (&reserves[k].reserve_pub);
+        reserves[k].balance = value;
+        reserves[k].execution_time = ts;
+        reserves[k].sender_account_details = sndr;
+        reserves[k].exchange_account_name = "name";
+        reserves[k].wire_reference = k;
+      }
+      FAILIF (batch_size !=
+            plugin->batch_reserves_in_insert (plugin->cls,
+                                              reserves,
+                                              batch_size,
+                                              results));
+      }
+
+    duration = GNUNET_TIME_absolute_get_duration (now);
+    fprintf (stdout,
+             "for a batchsize equal to %d it took %s\n",
+             batch_size,
+             GNUNET_STRINGS_relative_time_to_string (duration,
+                                                     GNUNET_NO) );
+
+  }
+  result = 0;
+drop:
+  GNUNET_break (GNUNET_OK ==
+                plugin->drop_tables (plugin->cls));
+cleanup:
+  TALER_EXCHANGEDB_plugin_unload (plugin);
+  plugin = NULL;
+}
+
+
+int
+main (int argc,
+      char *const argv[])
+{
+  const char *plugin_name;
+  char *config_filename;
+  char *testname;
+  struct GNUNET_CONFIGURATION_Handle *cfg;
+  (void) argc;
+  result = -1;
+   if (NULL == (plugin_name = strrchr (argv[0], (int) '-')))
+  {
+    GNUNET_break (0);
+    return -1;
+    }
+
+  GNUNET_log_setup (argv[0],
+                    "WARNING",
+                    NULL);
+   plugin_name++;
+    (void) GNUNET_asprintf (&testname,
+                          "test-exchange-db-%s",
+                          plugin_name);
+    (void) GNUNET_asprintf (&config_filename,
+                          "%s.conf",
+                          testname);
+  fprintf (stdout,
+           "Using config: %s\n",
+           config_filename);
+  cfg = GNUNET_CONFIGURATION_create ();
+  if (GNUNET_OK !=
+      GNUNET_CONFIGURATION_parse (cfg,
+                                  config_filename))
+  {
+    GNUNET_break (0);
+    GNUNET_free (config_filename);
+    GNUNET_free (testname);
+    return 2;
+  }
+  GNUNET_SCHEDULER_run (&run,
+                        cfg);
+  GNUNET_CONFIGURATION_destroy (cfg);
+  GNUNET_free (config_filename);
+  GNUNET_free (testname);
+  return result;
+}
+
+/* end of test_exchangedb_by_j.c */

src/exchangedb/test_exchangedb_by_j.c

@@ -91,9 +91,22 @@ run (void *cls)
     result = 77;
     goto cleanup;
   }
+
   for (unsigned int i = 0; i< 7; i++)
+
+  if (GNUNET_OK !=
+      plugin->setup_partitions (plugin->cls,
+                                num_partitions))
   {
-    static unsigned int batches[] = {1, 1, 2, 4, 16, 64, 256};
+    GNUNET_break (0);
+    result = 77;
+    goto cleanup;
+  }
+
+  for (unsigned int i = 0; i< 8; i++)
+
+  {
+    static unsigned int batches[] = {1, 1,0, 2, 4, 16, 64, 256};
     const char *sndr = "payto://x-taler-bank/localhost:8080/1";
     struct TALER_Amount value;
     unsigned int batch_size = batches[i];
@@ -101,6 +114,7 @@ run (void *cls)
     struct GNUNET_TIME_Timestamp ts;
     struct GNUNET_TIME_Relative duration;
     struct TALER_EXCHANGEDB_ReserveInInfo reserves[batch_size];
+    /*   struct TALER_EXCHANGEDB_ReserveInInfo reserves2[batch_size];*/
     enum GNUNET_DB_QueryStatus results[batch_size];
     GNUNET_assert (GNUNET_OK ==
                    TALER_string_to_amount (CURRENCY ":1.000010",
@@ -109,10 +123,8 @@ run (void *cls)
     ts = GNUNET_TIME_timestamp_get ();
     for (unsigned int r = 0; r<10; r++)
     {
-      plugin->start_read_committed (plugin->cls,
-                                    "test_by_j");
 
-      for (unsigned int k = 0; k<batch_size; k++)
+       for (unsigned int k = 0; k<batch_size; k++)
       {
         RND_BLK (&reserves[k].reserve_pub);
         reserves[k].balance = value;
@@ -120,22 +132,21 @@ run (void *cls)
         reserves[k].sender_account_details = sndr;
         reserves[k].exchange_account_name = "name";
         reserves[k].wire_reference = k;
-
       }
       FAILIF (batch_size !=
-              plugin->batch_reserves_in_insert (plugin->cls,
+            plugin->batch_reserves_in_insert (plugin->cls,
-                                                reserves,
+                                              reserves,
-                                                batch_size,
+                                              batch_size,
-                                                results));
+                                              results));
-
-      plugin->commit (plugin->cls);
     }
+
     duration = GNUNET_TIME_absolute_get_duration (now);
     fprintf (stdout,
              "for a batchsize equal to %d it took %s\n",
              batch_size,
              GNUNET_STRINGS_relative_time_to_string (duration,
                                                      GNUNET_NO) );
+
   }
   result = 0;
 drop:
@@ -155,7 +166,6 @@ main (int argc,
   char *config_filename;
   char *testname;
   struct GNUNET_CONFIGURATION_Handle *cfg;
-
   (void) argc;
   result = -1;
   if (NULL == (plugin_name = strrchr (argv[0], (int) '-')))
@@ -163,6 +173,7 @@ main (int argc,
     GNUNET_break (0);
     return -1;
   }
+
   GNUNET_log_setup (argv[0],
                     "WARNING",
                     NULL);

src/include/taler_exchangedb_plugin.h

@@ -3453,6 +3453,21 @@ struct TALER_EXCHANGEDB_Plugin
                               unsigned int reserves_length,
                               enum GNUNET_DB_QueryStatus *results);
 
+  /**
+   * Insert a batch of incoming transaction into reserves.  New reserves are
+   * also created through this function.
+   *
+   * @param cls the @e cls of this struct with the plugin-specific state
+   * @param reserves
+   * @param reserves_length length of the @a reserves array
+   * @param[out] results array of transaction status codes of length @a reserves_length,
+   *             set to the status of the
+   */
+  enum GNUNET_DB_QueryStatus
+  (*batch2_reserves_in_insert)(void *cls,
+                              const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
+                              unsigned int reserves_length,
+                              enum GNUNET_DB_QueryStatus *results);
 
   /**
    * Locate a nonce for use with a particular public key.