Stop discarding accepted shares when the DB is momentarily locked

src/store.c

@@ -33,6 +33,12 @@ static const char *SCHEMA_SQL_PARTS[] = {
     "PRAGMA journal_mode = WAL;\n"
     "PRAGMA synchronous = NORMAL;\n"
     "PRAGMA foreign_keys = ON;\n"
+    /* Without this SQLite returns SQLITE_BUSY the instant another connection
+     * holds the write lock — no waiting at all. The writer thread has already
+     * dequeued its batch by then, so a single concurrent writer (a manual
+     * sqlite3 session, a backup, a maintenance script) silently destroyed
+     * shares the miner had been told were accepted. Wait instead. */
+    "PRAGMA busy_timeout = 5000;\n"
     "CREATE TABLE IF NOT EXISTS workers ("
     "  id              INTEGER PRIMARY KEY AUTOINCREMENT,"
     "  name            TEXT UNIQUE NOT NULL,"
@@ -270,6 +276,12 @@ static const char *MIGRATIONS_SQL[] = {
     "UPDATE templates SET last_seen = ts WHERE last_seen = 0",
 };
 
+/* Retries for one batch. busy_timeout (5s) bounds each attempt, so the worst
+ * case is a long stall rather than a fast loop — which is the right trade:
+ * enqueue-side overflow is counted in shares_dropped and visible, whereas a
+ * dropped batch here is credited work vanishing. */
+#define STORE_COMMIT_ATTEMPTS 3
+
 #define EV_SHARE   1
 #define EV_REJECT  2
 #define EV_BLOCK   3
@@ -346,6 +358,7 @@ struct store {
     _Atomic uint64_t credits_committed;
     _Atomic uint64_t batches;
     _Atomic uint64_t pg_errors;
+    _Atomic uint64_t events_lost;
 
     /* Sequence: monotonically increasing counter of enqueued events.
      * 'committed_seq' tracks the highest sequence that has been
@@ -362,6 +375,12 @@ void store_test_set_ring_capacity(size_t cap) { g_test_ring_cap = cap; }
 
 /* ---- helpers ---------------------------------------------------------- */
 
+/* Linear backoff between commit attempts: 25ms, 50ms, ... */
+static void backoff_sleep(int attempt) {
+    struct timespec ts = { .tv_sec = 0, .tv_nsec = 25L * 1000000L * attempt };
+    nanosleep(&ts, NULL);
+}
+
 static uint64_t now_ms(void) {
     struct timespec ts;
     clock_gettime(CLOCK_REALTIME, &ts);
@@ -541,6 +560,44 @@ static void process_event(store_t *s, const event_t *ev) {
     }
 }
 
+/* Attempts to land one batch. The events are already out of the ring, so a
+ * failure here destroys accepted work — retry rather than count and move on.
+ *
+ * busy_timeout already makes SQLITE_BUSY rare; these attempts cover a lock
+ * held longer than that, and an I/O error that clears. Counters advance only
+ * on the attempt that actually commits, so a retried batch is counted once.
+ * Returns 0 committed, -1 out of attempts. */
+static int commit_batch(store_t *s, event_t *batch, size_t take) {
+    for (int attempt = 1; attempt <= STORE_COMMIT_ATTEMPTS; ++attempt) {
+        char *err = NULL;
+        if (sqlite3_exec(s->db, "BEGIN IMMEDIATE", NULL, NULL, &err) != SQLITE_OK) {
+            LOG_WARN("store: BEGIN failed (attempt %d/%d): %s",
+                     attempt, STORE_COMMIT_ATTEMPTS, err ? err : "?");
+            sqlite3_free(err);
+            atomic_fetch_add(&s->pg_errors, 1);
+            backoff_sleep(attempt);
+            continue;
+        }
+
+        for (size_t i = 0; i < take; ++i) process_event(s, &batch[i]);
+
+        if (sqlite3_exec(s->db, "COMMIT", NULL, NULL, &err) == SQLITE_OK) {
+            atomic_fetch_add(&s->batches, 1);
+            return 0;
+        }
+        LOG_WARN("store: COMMIT failed (attempt %d/%d): %s",
+                 attempt, STORE_COMMIT_ATTEMPTS, err ? err : "?");
+        sqlite3_free(err);
+        /* Nothing was durably written, so replaying the batch is safe. The
+         * per-event counters process_event() bumped are lost accuracy we
+         * accept: they describe attempts, the ledger describes reality. */
+        sqlite3_exec(s->db, "ROLLBACK", NULL, NULL, NULL);
+        atomic_fetch_add(&s->pg_errors, 1);
+        backoff_sleep(attempt);
+    }
+    return -1;
+}
+
 static void *writer_main(void *arg) {
     store_t *s = (store_t *)arg;
 
@@ -585,21 +642,15 @@ static void *writer_main(void *arg) {
         pthread_mutex_unlock(&s->mu);
 
         /* BEGIN/COMMIT outside the producer mutex */
-        char *err = NULL;
-        if (sqlite3_exec(s->db, "BEGIN IMMEDIATE", NULL, NULL, &err) != SQLITE_OK) {
-            LOG_ERROR("store: BEGIN failed: %s", err ? err : "?");
-            sqlite3_free(err);
-            atomic_fetch_add(&s->pg_errors, 1);
-        } else {
-            for (size_t i = 0; i < take; ++i) process_event(s, &batch[i]);
-            if (sqlite3_exec(s->db, "COMMIT", NULL, NULL, &err) != SQLITE_OK) {
-                LOG_ERROR("store: COMMIT failed: %s", err ? err : "?");
-                sqlite3_free(err);
-                sqlite3_exec(s->db, "ROLLBACK", NULL, NULL, NULL);
-                atomic_fetch_add(&s->pg_errors, 1);
-            } else {
-                atomic_fetch_add(&s->batches, 1);
-            }
+        if (commit_batch(s, batch, take) != 0) {
+            /* Out of retries. These events left the ring before the
+             * transaction opened and cannot be put back, so say so plainly —
+             * this is accepted work that will never be credited, not a
+             * transient blip. */
+            LOG_ERROR("store: LOST %zu event(s) after %d failed commit attempts"
+                      " — accepted shares in this batch are not credited",
+                      take, STORE_COMMIT_ATTEMPTS);
+            atomic_fetch_add(&s->events_lost, (uint64_t)take);
         }
 
         pthread_mutex_lock(&s->mu);
@@ -1193,4 +1244,5 @@ void store_get_stats(store_t *s, store_stats_t *out) {
     out->blocks_committed = atomic_load(&s->blocks_committed);
     out->batches          = atomic_load(&s->batches);
     out->pg_errors        = atomic_load(&s->pg_errors);
+    out->events_lost      = atomic_load(&s->events_lost);
 }

src/store.h

@@ -164,6 +164,11 @@ typedef struct {
     uint64_t blocks_committed;
     uint64_t batches;
     uint64_t pg_errors;        /* poorly named; sqlite errors */
+    /* Events that left the ring but never reached the DB, after every
+     * commit retry failed. Accepted work that will never be credited —
+     * distinct from shares_dropped, which is enqueue-side overflow.
+     * Must be 0; anything else is a ledger shortfall against miners. */
+    uint64_t events_lost;
 } store_stats_t;
 void store_get_stats(store_t *s, store_stats_t *out);
 

tests/test_store.c

@@ -496,6 +496,61 @@ static void test_template_history(void) {
     printf("  ok test_template_history\n");
 }
 
+/* A concurrent writer must not cost shares.
+ *
+ * This reproduces a real incident: a maintenance script took the write lock
+ * for a moment, and because the connection had no busy_timeout the writer
+ * thread got SQLITE_BUSY instantly. Its batch was already out of the ring, so
+ * accepted shares — already acknowledged to the miner — were logged and
+ * discarded. Holding the lock here for longer than one commit window forces
+ * exactly that race. */
+static void test_commit_survives_a_locked_db(void) {
+    const char *path = fresh_db_path();
+    store_cfg_t cfg;
+    memset(&cfg, 0, sizeof(cfg));
+    snprintf(cfg.path, sizeof(cfg.path), "%s", path);
+    cfg.commit_window_ms  = 20;    /* wake often, so we hit the lock */
+    cfg.commit_max_shares = 10;    /* several batches, not one */
+    store_t *s = NULL;
+    assert(store_open(&cfg, &s) == 0);
+
+    /* A second connection grabs the write lock, as `sqlite3 < script.sql`
+     * would. */
+    sqlite3 *hog = NULL;
+    assert(sqlite3_open(path, &hog) == SQLITE_OK);
+    assert(sqlite3_exec(hog, "BEGIN IMMEDIATE", NULL, NULL, NULL) == SQLITE_OK);
+
+    const int N = 40;
+    for (int i = 0; i < N; ++i) {
+        char w[32];
+        snprintf(w, sizeof(w), "miner%d", i % 4);
+        assert(store_record_share(s, w, 1700000000000ULL + i, 1.0, 0, NULL) == 0);
+    }
+
+    /* Hold it well past several commit windows, then let go. */
+    struct timespec hold = { .tv_sec = 0, .tv_nsec = 300L * 1000000L };
+    nanosleep(&hold, NULL);
+    assert(sqlite3_exec(hog, "COMMIT", NULL, NULL, NULL) == SQLITE_OK);
+    sqlite3_close(hog);
+
+    assert(store_flush(s) == 0);
+
+    store_stats_t st;
+    store_get_stats(s, &st);
+    assert(st.events_lost == 0);
+    assert(st.shares_dropped == 0);
+
+    sqlite3 *db = NULL;
+    assert(sqlite3_open(path, &db) == SQLITE_OK);
+    /* Every accepted share is on the ledger. Before the fix this came back
+     * short, with the shortfall visible only as an ERROR line. */
+    assert(scalar_i64(db, "SELECT count(*) FROM shares") == N);
+
+    sqlite3_close(db);
+    store_close(s);
+    printf("  ok test_commit_survives_a_locked_db\n");
+}
+
 /* Retention keeps the table bounded. Pruning runs when a new row is opened
  * and is driven off the template's own clock, so this is deterministic. */
 static void test_template_retention(void) {
@@ -564,6 +619,7 @@ int main(void) {
     test_rate_history();
     test_template_history();
     test_template_retention();
+    test_commit_survives_a_locked_db();
     cleanup_dbs();
     printf("all tests passed\n");
     return 0;