change sharding strategy for refund table

6 files changed, 65 insertions, 81 deletions

diff --git a/src/exchangedb/exchange-0001.sql b/src/exchangedb/exchange-0001.sql
index a42baa1f..b2fb52ac 100644
--- a/src/exchangedb/exchange-0001.sql
+++ b/src/exchangedb/exchange-0001.sql
@@ -686,7 +686,7 @@ CREATE TABLE IF NOT EXISTS extension_details_default
 CREATE TABLE IF NOT EXISTS deposits
   (deposit_serial_id BIGINT GENERATED BY DEFAULT AS IDENTITY -- PRIMARY KEY
   ,shard INT8 NOT NULL
-  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) -- REFERENCES known_coins (coin_pub) ON DELETE CASCADE
+  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) REFERENCES known_coins (coin_pub) ON DELETE CASCADE
   ,known_coin_id BIGINT NOT NULL -- REFERENCES known_coins (known_coin_id) ON DELETE CASCADE 
   ,amount_with_fee_val INT8 NOT NULL
   ,amount_with_fee_frac INT4 NOT NULL
@@ -754,7 +754,7 @@ SELECT add_constraints_to_deposits_partition('default');
 CREATE TABLE IF NOT EXISTS deposits_by_ready
   (wire_deadline INT8 NOT NULL
   ,shard INT8 NOT NULL
-  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32)
+  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) REFERENCES known_coins (coin_pub) ON DELETE CASCADE
   ,deposit_serial_id INT8
   )
   PARTITION BY RANGE (wire_deadline);
@@ -773,7 +773,7 @@ CREATE TABLE IF NOT EXISTS deposits_by_ready_default
 CREATE TABLE IF NOT EXISTS deposits_for_matching
   (refund_deadline INT8 NOT NULL
   ,shard INT8 NOT NULL
-  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32)
+  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) REFERENCES known_coins (coin_pub) ON DELETE CASCADE
   ,deposit_serial_id INT8
   )
   PARTITION BY RANGE (refund_deadline);
@@ -947,11 +947,9 @@ CREATE TRIGGER deposits_on_delete
 
 -- ------------------------------ refunds ----------------------------------------
 
--- FIXME-URGENT: very bad structure, should replace 'shard' by 'coin_pub'
--- as deposits is sharded by that now!
 CREATE TABLE IF NOT EXISTS refunds
   (refund_serial_id BIGINT GENERATED BY DEFAULT AS IDENTITY -- UNIQUE
-  ,shard INT8 NOT NULL -- REFERENCES deposits (shard)
+  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) REFERENCES known_coins (coin_pub) ON DELETE CASCADE
   ,deposit_serial_id INT8 NOT NULL -- REFERENCES deposits (deposit_serial_id) ON DELETE CASCADE
   ,merchant_sig BYTEA NOT NULL CHECK(LENGTH(merchant_sig)=64)
   ,rtransaction_id INT8 NOT NULL
@@ -959,7 +957,7 @@ CREATE TABLE IF NOT EXISTS refunds
   ,amount_with_fee_frac INT4 NOT NULL
   -- ,PRIMARY KEY (deposit_serial_id, rtransaction_id) -- done per shard!
   )
-  PARTITION BY HASH (shard);
+  PARTITION BY HASH (coin_pub);
 COMMENT ON TABLE refunds
   IS 'Data on coins that were refunded. Technically, refunds always apply against specific deposit operations involving a coin. The combination of coin_pub, merchant_pub, h_contract_terms and rtransaction_id MUST be unique, and we usually select by coin_pub so that one goes first.';
 COMMENT ON COLUMN refunds.deposit_serial_id
@@ -967,6 +965,10 @@ COMMENT ON COLUMN refunds.deposit_serial_id
 COMMENT ON COLUMN refunds.rtransaction_id
   IS 'used by the merchant to make refunds unique in case the same coin for the same deposit gets a subsequent (higher) refund';
 
+CREATE INDEX IF NOT EXISTS refunds_by_coin_pub_index
+  ON refunds
+  (coin_pub);
+
 CREATE TABLE IF NOT EXISTS refunds_default
   PARTITION OF refunds
   FOR VALUES WITH (MODULUS 1, REMAINDER 0);
@@ -989,9 +991,6 @@ $$;
 
 SELECT add_constraints_to_refunds_partition('default');
 
-CREATE INDEX IF NOT EXISTS refunds_by_deposit_serial_id_index
-  ON refunds
-  (shard,deposit_serial_id);
 
 
 -- ------------------------------ wire_out ----------------------------------------
@@ -1146,7 +1145,7 @@ CREATE INDEX IF NOT EXISTS global_fee_by_end_date_index
 
 CREATE TABLE IF NOT EXISTS recoup
   (recoup_uuid BIGINT GENERATED BY DEFAULT AS IDENTITY -- UNIQUE
-  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) -- REFERENCES known_coins (coin_pub)
+  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) REFERENCES known_coins (coin_pub)
   ,coin_sig BYTEA NOT NULL CHECK(LENGTH(coin_sig)=64)
   ,coin_blind BYTEA NOT NULL CHECK(LENGTH(coin_blind)=32)
   ,amount_val INT8 NOT NULL
@@ -1193,7 +1192,7 @@ SELECT add_constraints_to_recoup_partition('default');
 
 CREATE TABLE IF NOT EXISTS recoup_by_reserve
   (reserve_out_serial_id INT8 NOT NULL -- REFERENCES reserves (reserve_out_serial_id) ON DELETE CASCADE
-  ,coin_pub BYTEA CHECK (LENGTH(coin_pub)=32) 
+  ,coin_pub BYTEA CHECK (LENGTH(coin_pub)=32) REFERENCES known_coins (coin_pub)
   )
   PARTITION BY HASH (reserve_out_serial_id);
 COMMENT ON TABLE recoup_by_reserve
@@ -1251,8 +1250,8 @@ CREATE TRIGGER recoup_on_delete
 
 CREATE TABLE IF NOT EXISTS recoup_refresh
   (recoup_refresh_uuid BIGINT GENERATED BY DEFAULT AS IDENTITY -- UNIQUE
-  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) -- REFERENCES known_coins (coin_pub)
-  ,known_coin_id BIGINT NOT NULL -- REFERENCES known_coins (known_coin_id) ON DELETE CASCADE
+  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) REFERENCES known_coins (coin_pub)
+  ,known_coin_id BIGINT NOT NULL -- REFERENCES known_coins (known_coin_id)
   ,coin_sig BYTEA NOT NULL CHECK(LENGTH(coin_sig)=64)
   ,coin_blind BYTEA NOT NULL CHECK(LENGTH(coin_blind)=32)
   ,amount_val INT8 NOT NULL
@@ -2709,8 +2708,8 @@ DECLARE
   deposit_frac INT8; -- amount that was originally deposited
 BEGIN
 -- Shards: SELECT deposits (coin_pub, shard, h_contract_terms, merchant_pub)
---         INSERT refunds (by deposit_serial_id, rtransaction_id) ON CONFLICT DO NOTHING
---         SELECT refunds (by deposit_serial_id)
+--         INSERT refunds (by coin_pub, rtransaction_id) ON CONFLICT DO NOTHING
+--         SELECT refunds (by coin_pub)
 --         UPDATE known_coins (by coin_pub)
 
 SELECT
@@ -2741,7 +2740,7 @@ END IF;
 
 INSERT INTO refunds
   (deposit_serial_id
-  ,shard
+  ,coin_pub
   ,merchant_sig
   ,rtransaction_id
   ,amount_with_fee_val
@@ -2749,7 +2748,7 @@ INSERT INTO refunds
   )
   VALUES
   (dsi
-  ,in_deposit_shard
+  ,in_coin_pub
   ,in_merchant_sig
   ,in_rtransaction_id
   ,in_amount_with_fee_val
@@ -2765,7 +2764,7 @@ THEN
   -- primarily here to maximally use the existing index.
    PERFORM
    FROM refunds
-   WHERE shard=in_deposit_shard
+   WHERE coin_pub=in_coin_pub
      AND deposit_serial_id=dsi
      AND rtransaction_id=in_rtransaction_id
      AND amount_with_fee_val=in_amount_with_fee_val
@@ -2805,7 +2804,7 @@ SELECT
    tmp_val
   ,tmp_frac
   FROM refunds
-  WHERE shard=in_deposit_shard
+  WHERE coin_pub=in_coin_pub
     AND deposit_serial_id=dsi;
 IF tmp_val IS NULL
 THEN
diff --git a/src/exchangedb/irbt_callbacks.c b/src/exchangedb/irbt_callbacks.c
index fd6e49b1..835c0ea7 100644
--- a/src/exchangedb/irbt_callbacks.c
+++ b/src/exchangedb/irbt_callbacks.c
@@ -554,7 +554,7 @@ irbt_cb_table_refunds (struct PostgresClosure *pg,
 {
   struct GNUNET_PQ_QueryParam params[] = {
     GNUNET_PQ_query_param_uint64 (&td->serial),
-    GNUNET_PQ_query_param_uint64 (&td->details.refunds.shard),
+    GNUNET_PQ_query_param_auto_from_type (&td->details.refunds.coin_pub),
     GNUNET_PQ_query_param_auto_from_type (&td->details.refunds.merchant_sig),
     GNUNET_PQ_query_param_uint64 (&td->details.refunds.rtransaction_id),
     TALER_PQ_query_param_amount (&td->details.refunds.amount_with_fee),
diff --git a/src/exchangedb/lrbt_callbacks.c b/src/exchangedb/lrbt_callbacks.c
index 011b6a3e..0fec486e 100644
--- a/src/exchangedb/lrbt_callbacks.c
+++ b/src/exchangedb/lrbt_callbacks.c
@@ -1010,9 +1010,9 @@ lrbt_cb_table_refunds (void *cls,
       GNUNET_PQ_result_spec_uint64 (
         "serial",
         &td.serial),
-      GNUNET_PQ_result_spec_uint64 (
-        "shard",
-        &td.details.refunds.shard),
+      GNUNET_PQ_result_spec_auto_from_type (
+        "coin_pub",
+        &td.details.refunds.coin_pub),
       GNUNET_PQ_result_spec_auto_from_type (
         "merchant_sig",
         &td.details.refunds.merchant_sig),
diff --git a/src/exchangedb/plugin_exchangedb_postgres.c b/src/exchangedb/plugin_exchangedb_postgres.c
index 8dc201a2..c7bdae39 100644
--- a/src/exchangedb/plugin_exchangedb_postgres.c
+++ b/src/exchangedb/plugin_exchangedb_postgres.c
@@ -43,12 +43,6 @@
 #define AUTO_EXPLAIN 1
 
 /**
- * Should we explicitly lock certain individual tables prior to SELECT+INSERT
- * combis?
- */
-#define EXPLICIT_LOCKS 0
-
-/**
  * Wrapper macro to add the currency from the plugin's state
  * when fetching amounts from the database.
  *
@@ -991,13 +985,14 @@ prepare_statements (struct PostgresClosure *pg)
     GNUNET_PQ_make_prepare (
       "insert_refund",
       "INSERT INTO refunds "
-      "(deposit_serial_id "
+      "(coin_pub "
+      ",deposit_serial_id"
       ",merchant_sig "
       ",rtransaction_id "
       ",amount_with_fee_val "
       ",amount_with_fee_frac "
-      ") SELECT deposit_serial_id, $3, $5, $6, $7"
-      "    FROM deposits" /* FIXME: check if adding additional AND on the 'shard' would help (possibly after reviewing indices on deposits!) */
+      ") SELECT $1, deposit_serial_id, $3, $5, $6, $7"
+      "    FROM deposits"
       "   WHERE coin_pub=$1"
       "     AND h_contract_terms=$4"
       "     AND merchant_pub=$2",
@@ -1015,11 +1010,14 @@ prepare_statements (struct PostgresClosure *pg)
       ",denom.fee_refund_val "
       ",denom.fee_refund_frac "
       ",ref.refund_serial_id"
-      " FROM deposits dep"
-      " JOIN refunds ref USING (deposit_serial_id)"
-      " JOIN known_coins kc ON (dep.coin_pub = kc.coin_pub)"
-      " JOIN denominations denom USING (denominations_serial)"
-      " WHERE dep.coin_pub=$1;",
+      " FROM refunds ref"
+      " JOIN deposits dep"
+      "   ON (ref.coin_pub = dep.coin_pub AND ref.deposit_serial_id = dep.deposit_serial_id)"
+      " JOIN known_coins kc"
+      "   ON (ref.coin_pub = kc.coin_pub)"
+      " JOIN denominations denom"
+      "   USING (denominations_serial)"
+      " WHERE ref.coin_pub=$1;",
       1),
     /* Query the 'refunds' by coin public key, merchant_pub and contract hash */
     GNUNET_PQ_make_prepare (
@@ -1027,9 +1025,10 @@ prepare_statements (struct PostgresClosure *pg)
       "SELECT"
       " ref.amount_with_fee_val"
       ",ref.amount_with_fee_frac"
-      " FROM deposits dep"
-      " JOIN refunds ref USING (shard,deposit_serial_id)"
-      " WHERE dep.coin_pub=$1"
+      " FROM refunds ref"
+      " JOIN deposits dep"
+      "   USING (coin_pub,deposit_serial_id)"
+      " WHERE ref.coin_pub=$1"
       "   AND dep.merchant_pub=$2"
       "   AND dep.h_contract_terms=$3;",
       3),
@@ -1037,30 +1036,26 @@ prepare_statements (struct PostgresClosure *pg)
     GNUNET_PQ_make_prepare (
       "audit_get_refunds_incr",
       "SELECT"
-      " merchant_pub"
-      ",merchant_sig"
-      ",h_contract_terms"
-      ",rtransaction_id"
+      " dep.merchant_pub"
+      ",ref.merchant_sig"
+      ",dep.h_contract_terms"
+      ",ref.rtransaction_id"
       ",denom.denom_pub"
       ",kc.coin_pub"
-      ",refunds.amount_with_fee_val"
-      ",refunds.amount_with_fee_frac"
-      ",refund_serial_id"
-      " FROM refunds"
-      "   JOIN deposits USING (shard, deposit_serial_id)"
-      "   JOIN known_coins kc USING (coin_pub)"
-      "   JOIN denominations denom ON (kc.denominations_serial = denom.denominations_serial)"
-      " WHERE refund_serial_id>=$1"
-      " ORDER BY refund_serial_id ASC;",
+      ",ref.amount_with_fee_val"
+      ",ref.amount_with_fee_frac"
+      ",ref.refund_serial_id"
+      " FROM refunds ref"
+      "   JOIN deposits dep"
+      "     ON (ref.coin_pub=dep.coin_pub AND ref.deposit_serial_id=dep.deposit_serial_id)"
+      "   JOIN known_coins kc"
+      "     ON (dep.coin_pub=kc.coin_pub)"
+      "   JOIN denominations denom"
+      "     ON (kc.denominations_serial=denom.denominations_serial)"
+      " WHERE ref.refund_serial_id>=$1"
+      " ORDER BY ref.refund_serial_id ASC;",
       1),
-    /* Lock deposit table; NOTE: we may want to eventually shard the
-       deposit table to avoid this lock being the main point of
-       contention limiting transaction performance. */
-    // FIXME: check if this query is even still used!
-    GNUNET_PQ_make_prepare (
-      "lock_deposit",
-      "LOCK TABLE deposits;",
-      0),
+
     /* Store information about a /deposit the exchange is to execute.
        Used in #postgres_insert_deposit(). */
     GNUNET_PQ_make_prepare (
@@ -1542,9 +1537,8 @@ prepare_statements (struct PostgresClosure *pg)
       " WHERE wire_deadline >= $1"
       " AND wire_deadline < $2"
       " AND NOT (EXISTS (SELECT 1"
-      "            FROM refunds"
-      "            JOIN deposits dx USING (deposit_serial_id)"
-      "            WHERE (dx.coin_pub = d.coin_pub))"
+      "            FROM refunds r"
+      "            WHERE (r.coin_pub = d.coin_pub) AND (r.deposit_serial_id = d.deposit_serial_id))"
       "       OR EXISTS (SELECT 1"
       "            FROM aggregation_tracking"
       "            WHERE (aggregation_tracking.deposit_serial_id = d.deposit_serial_id)))"
@@ -2509,7 +2503,7 @@ prepare_statements (struct PostgresClosure *pg)
       "select_above_serial_by_table_refunds",
       "SELECT"
       " refund_serial_id AS serial"
-      ",shard"
+      ",coin_pub"
       ",merchant_sig"
       ",rtransaction_id"
       ",amount_with_fee_val"
@@ -2841,7 +2835,7 @@ prepare_statements (struct PostgresClosure *pg)
     GNUNET_PQ_make_prepare (
       "insert_into_table_refunds",
       "INSERT INTO refunds"
-      "(shard"
+      "(coin_pub"
       ",refund_serial_id"
       ",merchant_sig"
       ",rtransaction_id"
@@ -5842,16 +5836,7 @@ postgres_have_deposit2 (
   };
   enum GNUNET_DB_QueryStatus qs;
   struct TALER_MerchantWireHashP h_wire2;
-#if EXPLICIT_LOCKS
-  struct GNUNET_PQ_QueryParam no_params[] = {
-    GNUNET_PQ_query_param_end
-  };
 
-  if (0 > (qs = GNUNET_PQ_eval_prepared_non_select (pg->conn,
-                                                    "lock_deposit",
-                                                    no_params)))
-    return qs;
-#endif
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Getting deposits for coin %s\n",
               TALER_B2S (coin_pub));
@@ -9314,6 +9299,7 @@ refunds_serial_helper_cb (void *cls,
   struct RefundsSerialContext *rsc = cls;
   struct PostgresClosure *pg = rsc->pg;
 
+  fprintf (stderr, "Got %u results\n", num_results);
   for (unsigned int i = 0; i<num_results; i++)
   {
     struct TALER_EXCHANGEDB_Refund refund;
@@ -9338,7 +9324,7 @@ refunds_serial_helper_cb (void *cls,
                                     &rowid),
       GNUNET_PQ_result_spec_end
     };
-    int ret;
+    enum GNUNET_GenericReturnValue ret;
 
     if (GNUNET_OK !=
         GNUNET_PQ_extract_result (result,
diff --git a/src/include/taler_exchangedb_plugin.h b/src/include/taler_exchangedb_plugin.h
index 2a462aba..b2ea240e 100644
--- a/src/include/taler_exchangedb_plugin.h
+++ b/src/include/taler_exchangedb_plugin.h
@@ -359,7 +359,7 @@ struct TALER_EXCHANGEDB_TableData
 
     struct
     {
-      uint64_t shard;
+      struct TALER_CoinSpendPublicKeyP coin_pub;
       uint64_t deposit_serial_id;
       struct TALER_MerchantSignatureP merchant_sig;
       uint64_t rtransaction_id;
diff --git a/src/testing/test_exchange_api.c b/src/testing/test_exchange_api.c
index 71f9ab7c..4f1e2a61 100644
--- a/src/testing/test_exchange_api.c
+++ b/src/testing/test_exchange_api.c
@@ -83,6 +83,7 @@ static bool uses_cs;
  * @param label label to use for the command.
  */
 #define CMD_EXEC_AGGREGATOR(label) \
+  TALER_TESTING_cmd_sleep ("sleep-before-aggregator", 2), \
   TALER_TESTING_cmd_exec_aggregator (label "-aggregator", config_file), \
   TALER_TESTING_cmd_exec_transfer (label "-transfer", config_file)
 
@@ -453,8 +454,6 @@ run (void *cls,
     TALER_TESTING_cmd_track_transfer_empty ("wire-deposit-failing",
                                             NULL,
                                             MHD_HTTP_NOT_FOUND),
-    TALER_TESTING_cmd_sleep ("sleep-before-aggregator",
-                             1),
     /* Run transfers. Note that _actual_ aggregation will NOT
      * happen here, as each deposit operation is run with a
      * fresh merchant public key, so the aggregator will treat
@@ -759,7 +758,7 @@ run (void *cls,
      * Note, this operation takes two commands: one to "flush"
      * the preliminary transfer (used to withdraw) from the
      * fakebank and the second to actually check there are not
-     * other transfers around. *///
+     * other transfers around. */
     TALER_TESTING_cmd_check_bank_empty ("check_bank_transfer-pre-refund"),
     TALER_TESTING_cmd_refund_with_id ("refund-ok",
                                       MHD_HTTP_OK,