diff --git a/src/store.c b/src/store.c index 0338dee..4369f62 100644 --- a/src/store.c +++ b/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); } diff --git a/src/store.h b/src/store.h index 5c43e32..d6db71e 100644 --- a/src/store.h +++ b/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); diff --git a/tests/test_store.c b/tests/test_store.c index a2f706f..3a23ae3 100644 --- a/tests/test_store.c +++ b/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;