commit 9458a7ddacc7b3939777fa9a5b802484451849a5
parent cafe3b2c93136904859a5ccbd54c1249beba0983
Author: Christian Grothoff <christian@grothoff.org>
Date: Thu, 6 Aug 2026 00:13:34 +0200
fix statistics update to user proper INSERT ... ON CONFICT DO UPDATE-pattern
Diffstat:
2 files changed, 232 insertions(+), 104 deletions(-)
diff --git a/src/backenddb/pg_statistics_helpers.sql b/src/backenddb/pg_statistics_helpers.sql
@@ -52,24 +52,19 @@ BEGIN
INTO my_bucket_start
FROM merchant.interval_to_start (in_timestamp, my_range);
- UPDATE merchant_statistic_bucket_counter
- SET cumulative_number = cumulative_number + in_delta
- WHERE bmeta_serial_id=my_meta
- AND bucket_start=my_bucket_start
- AND bucket_range=my_range;
- IF NOT FOUND
- THEN
- INSERT INTO merchant_statistic_bucket_counter
- (bmeta_serial_id
- ,bucket_start
- ,bucket_range
- ,cumulative_number
- ) VALUES (
- my_meta
- ,my_bucket_start
- ,my_range
- ,in_delta);
- END IF;
+ INSERT INTO merchant_statistic_bucket_counter AS msbc
+ (bmeta_serial_id
+ ,bucket_start
+ ,bucket_range
+ ,cumulative_number
+ ) VALUES (
+ my_meta
+ ,my_bucket_start
+ ,my_range
+ ,in_delta)
+ ON CONFLICT (bmeta_serial_id, bucket_start, bucket_range)
+ DO UPDATE SET
+ cumulative_number = msbc.cumulative_number + in_delta;
END LOOP;
CLOSE my_curs;
END $$;
@@ -112,41 +107,34 @@ BEGIN
INTO my_bucket_start
FROM merchant.interval_to_start (in_timestamp, my_range);
- UPDATE merchant_statistic_bucket_amount
- SET
- cumulative_value = cumulative_value + (in_delta).val
- + CASE
- WHEN (in_delta).frac + cumulative_frac >= 100000000
- THEN 1
- ELSE 0
- END,
- cumulative_frac = cumulative_frac + (in_delta).frac
- - CASE
- WHEN (in_delta).frac + cumulative_frac >= 100000000
- THEN 100000000
- ELSE 0
- END
- WHERE bmeta_serial_id=my_meta
- AND curr=(in_delta).curr
- AND bucket_start=my_bucket_start
- AND bucket_range=my_range;
- IF NOT FOUND
- THEN
- INSERT INTO merchant_statistic_bucket_amount
- (bmeta_serial_id
- ,bucket_start
- ,bucket_range
- ,curr
- ,cumulative_value
- ,cumulative_frac
- ) VALUES (
- my_meta
- ,my_bucket_start
- ,my_range
- ,(in_delta).curr
- ,(in_delta).val
- ,(in_delta).frac);
- END IF;
+ INSERT INTO merchant_statistic_bucket_amount AS msba
+ (bmeta_serial_id
+ ,bucket_start
+ ,bucket_range
+ ,curr
+ ,cumulative_value
+ ,cumulative_frac
+ ) VALUES (
+ my_meta
+ ,my_bucket_start
+ ,my_range
+ ,(in_delta).curr
+ ,(in_delta).val
+ ,(in_delta).frac)
+ ON CONFLICT (bmeta_serial_id, curr, bucket_start, bucket_range)
+ DO UPDATE SET
+ cumulative_value = msba.cumulative_value + (in_delta).val
+ + CASE
+ WHEN (in_delta).frac + msba.cumulative_frac >= 100000000
+ THEN 1
+ ELSE 0
+ END,
+ cumulative_frac = msba.cumulative_frac + (in_delta).frac
+ - CASE
+ WHEN (in_delta).frac + msba.cumulative_frac >= 100000000
+ THEN 100000000
+ ELSE 0
+ END;
END LOOP;
CLOSE my_curs;
END $$;
@@ -223,23 +211,19 @@ BEGIN
RETURNING nevent_serial_id
INTO my_event;
- UPDATE merchant_statistic_interval_counter
- SET cumulative_number = cumulative_number + in_delta
- WHERE imeta_serial_id = my_meta
- AND range=my_rangex;
- IF NOT FOUND
- THEN
- INSERT INTO merchant_statistic_interval_counter
- (imeta_serial_id
- ,range
- ,event_delimiter
- ,cumulative_number
- ) VALUES (
- my_meta
- ,my_rangex
- ,my_event
- ,in_delta);
- END IF;
+ INSERT INTO merchant_statistic_interval_counter AS msic
+ (imeta_serial_id
+ ,range
+ ,event_delimiter
+ ,cumulative_number
+ ) VALUES (
+ my_meta
+ ,my_rangex
+ ,my_event
+ ,in_delta)
+ ON CONFLICT (imeta_serial_id, range)
+ DO UPDATE SET
+ cumulative_number = msic.cumulative_number + in_delta;
END $$;
COMMENT ON PROCEDURE merchant_do_bump_number_interval_stat
@@ -330,40 +314,34 @@ BEGIN
RETURNING aevent_serial_id
INTO my_event;
- UPDATE merchant_statistic_interval_amount
- SET
- cumulative_value = cumulative_value + (in_delta).val
- + CASE
- WHEN (in_delta).frac + cumulative_frac >= 100000000
- THEN 1
- ELSE 0
- END,
- cumulative_frac = cumulative_frac + (in_delta).frac
- - CASE
- WHEN (in_delta).frac + cumulative_frac >= 100000000
- THEN 100000000
- ELSE 0
- END
- WHERE imeta_serial_id=my_meta
- AND range=my_rangex
- AND curr=(in_delta).curr;
- IF NOT FOUND
- THEN
- INSERT INTO merchant_statistic_interval_amount
- (imeta_serial_id
- ,range
- ,event_delimiter
- ,curr
- ,cumulative_value
- ,cumulative_frac
- ) VALUES (
- my_meta
- ,my_rangex
- ,my_event
- ,(in_delta).curr
- ,(in_delta).val
- ,(in_delta).frac);
- END IF;
+ INSERT INTO merchant_statistic_interval_amount AS msia
+ (imeta_serial_id
+ ,range
+ ,event_delimiter
+ ,curr
+ ,cumulative_value
+ ,cumulative_frac
+ ) VALUES (
+ my_meta
+ ,my_rangex
+ ,my_event
+ ,(in_delta).curr
+ ,(in_delta).val
+ ,(in_delta).frac)
+ ON CONFLICT (imeta_serial_id, curr, range)
+ DO UPDATE SET
+ cumulative_value = msia.cumulative_value + (in_delta).val
+ + CASE
+ WHEN (in_delta).frac + msia.cumulative_frac >= 100000000
+ THEN 1
+ ELSE 0
+ END,
+ cumulative_frac = msia.cumulative_frac + (in_delta).frac
+ - CASE
+ WHEN (in_delta).frac + msia.cumulative_frac >= 100000000
+ THEN 100000000
+ ELSE 0
+ END;
END $$;
COMMENT ON PROCEDURE merchant_do_bump_amount_interval_stat
IS 'Updates an amount statistic tracked over an interval';
diff --git a/src/backenddb/test_merchantdb.c b/src/backenddb/test_merchantdb.c
@@ -30,6 +30,7 @@
#include "merchantdb_lib.h"
#include "helper.h"
#include "merchant-database/gc.h"
+#include "merchant-database/start.h"
#include "merchant-database/iterate_statistic_bucket_amounts.h"
#include "merchant-database/iterate_statistic_bucket_counters.h"
#include "merchant-database/iterate_kyc_statuses.h"
@@ -130,6 +131,12 @@ static int result;
static struct TALER_MERCHANTDB_PostgresContext *pg;
/**
+ * Configuration of the database we are testing. Used by tests that
+ * need a second, concurrent connection.
+ */
+static const struct GNUNET_CONFIGURATION_Handle *test_cfg;
+
+/**
* @param test 0 on success, non-zero on failure
*/
#define TEST_WITH_FAIL_CLAUSE(test, on_fail) \
@@ -10248,6 +10255,147 @@ test_statistics_counter_gc_delimiter (const struct InstanceData *instance)
/**
+ * SQL bumping the 'tokens-used' statistic by one.
+ */
+#define BUMP_TOKENS_USED \
+ "CALL merchant_do_bump_number_stat" \
+ " ('tokens-used'" \
+ " ,CURRENT_TIMESTAMP(0)::TIMESTAMP" \
+ " ,1)"
+
+
+/**
+ * Determines the largest counter value stored for @a slug.
+ *
+ * @param instance the instance to look at.
+ * @param slug the statistic to look at.
+ * @param[out] value set to the largest bucket counter.
+ * @return 0 on success, 1 otherwise.
+ */
+static int
+statistics_max_bucket_counter (const struct InstanceData *instance,
+ const char *slug,
+ uint64_t *value)
+{
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_string (slug),
+ GNUNET_PQ_query_param_end
+ };
+ struct GNUNET_PQ_ResultSpec rs[] = {
+ GNUNET_PQ_result_spec_uint64 ("num",
+ value),
+ GNUNET_PQ_result_spec_end
+ };
+
+ TEST_SET_INSTANCE (instance->instance.id,
+ GNUNET_DB_STATUS_SUCCESS_ONE_RESULT);
+ TEST_COND_RET_ON_FAIL (GNUNET_OK ==
+ GNUNET_PQ_prepare_anon (
+ pg->conn,
+ "SELECT COALESCE(MAX(cumulative_number),0) AS num"
+ " FROM merchant_statistic_bucket_counter"
+ " WHERE bmeta_serial_id ="
+ " (SELECT bmeta_serial_id"
+ " FROM merchant_statistic_bucket_meta"
+ " WHERE slug=$1"
+ " AND stype='number')"),
+ "Preparing counter lookup failed\n");
+ TEST_COND_RET_ON_FAIL (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT ==
+ GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
+ "",
+ params,
+ rs),
+ "Looking up bucket counter failed\n");
+ return 0;
+}
+
+
+/**
+ * Bumps the 'tokens-used' statistic from a second process while this
+ * process still holds the very first (uncommitted) bump of the same
+ * bucket. Regression test: the bucket statistics used to be written
+ * as UPDATE-then-INSERT, so the second writer got a unique violation
+ * (which GNUnet maps to SUCCESS_NO_RESULTS) and its event was lost;
+ * in the pay path the very same pattern surfaces to the wallet as a
+ * bogus 409 on a perfectly valid payment.
+ *
+ * @param instance the instance to collect statistics for.
+ * @return 0 on success, 1 otherwise.
+ */
+static int
+test_statistics_bucket_upsert_race (const struct InstanceData *instance)
+{
+ pid_t child;
+ int status;
+ uint64_t value;
+
+ TEST_SET_INSTANCE (instance->instance.id,
+ GNUNET_DB_STATUS_SUCCESS_ONE_RESULT);
+ /* Make sure the bucket does not exist yet, only then do both writers
+ race to create it. */
+ TEST_RET_ON_FAIL (statistics_exec_sql (
+ "DELETE FROM merchant_statistic_bucket_counter"
+ " WHERE bmeta_serial_id ="
+ " (SELECT bmeta_serial_id"
+ " FROM merchant_statistic_bucket_meta"
+ " WHERE slug='tokens-used'"
+ " AND stype='number')"));
+ TEST_COND_RET_ON_FAIL (GNUNET_OK ==
+ TALER_MERCHANTDB_start_read_committed (
+ pg,
+ "bucket statistics upsert race"),
+ "Failed to start transaction\n");
+ TEST_RET_ON_FAIL (statistics_exec_sql (BUMP_TOKENS_USED));
+ child = fork ();
+ if (-1 == child)
+ {
+ TALER_MERCHANTDB_rollback (pg);
+ GNUNET_break (0);
+ return 1;
+ }
+ if (0 == child)
+ {
+ struct TALER_MERCHANTDB_PostgresContext *cpg;
+ int ret = 1;
+
+ /* Second writer: blocks on the uncommitted row of the parent until
+ the parent commits below. */
+ cpg = TALER_MERCHANTDB_connect (test_cfg);
+ if ( (NULL != cpg) &&
+ (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT ==
+ TALER_MERCHANTDB_set_instance (cpg,
+ instance->instance.id)) )
+ {
+ struct GNUNET_PQ_ExecuteStatement es[] = {
+ GNUNET_PQ_make_execute (BUMP_TOKENS_USED),
+ GNUNET_PQ_EXECUTE_STATEMENT_END
+ };
+
+ if (GNUNET_OK ==
+ GNUNET_PQ_exec_statements (cpg->conn,
+ es))
+ ret = 0;
+ }
+ _exit (ret);
+ }
+ sleep (1); /* give the child time to reach the conflicting INSERT */
+ TEST_COND_RET_ON_FAIL (0 <=
+ TALER_MERCHANTDB_commit (pg),
+ "Failed to commit transaction\n");
+ if (0 >= waitpid (child,
+ &status,
+ 0))
+ sleep (2); /* cannot reap the child, just give it time to finish */
+ TEST_RET_ON_FAIL (statistics_max_bucket_counter (instance,
+ "tokens-used",
+ &value));
+ TEST_COND_RET_ON_FAIL (2 == value,
+ "Concurrent statistics bump was lost\n");
+ return 0;
+}
+
+
+/**
* Prepares for the statistics tests.
*
* @param[out] cls the closure to initialize.
@@ -10306,6 +10454,7 @@ run_test_statistics (struct TestStatistics_Closure *cls)
"deposits-fees-paid"));
TEST_RET_ON_FAIL (test_statistics_counter_gc_delimiter (&cls->instance));
TEST_RET_ON_FAIL (test_statistics_dual_type_slug (&cls->instance));
+ TEST_RET_ON_FAIL (test_statistics_bucket_upsert_race (&cls->instance));
return 0;
}
@@ -10365,6 +10514,7 @@ run (void *cls)
{
struct GNUNET_CONFIGURATION_Handle *cfg = cls;
+ test_cfg = cfg;
if (NULL == (pg = TALER_MERCHANTDB_connect (cfg)))
{
result = 77;
