From faf3537c0f736dd6f188eaff201c94ce53ee81ed Mon Sep 17 00:00:00 2001 From: asilgulsar1 Date: Wed, 4 Nov 2026 00:24:14 +0530 Subject: [PATCH] store: bound what a reject flood can write, and never drop a found block A reject costs the sender no work: authorize with any valid address, then submit bogus job ids at up to max_submits_per_sec per connection. Each wrote a rejects row (nothing deletes them), and the flood filled the event ring that block events share, so store_record_block could fail and a found block's blocks_found row was lost. - rejects_max_per_sec (default 0 = no limit): reject rows written per second, pool-wide. - Rejects stop at three quarters of the ring, leaving room for shares and blocks. - store_record_block waits (up to 30 s) for room instead of dropping the block's row. Co-Authored-By: Claude Opus 5.5 --- proxy.conf.example | 6 ++++++ src/config.c | 1 + src/config.h | 1 + src/main.c | 1 + src/store.c | 51 ++++++++++++++++++++++++++++++++++++++++------ src/store.h | 3 +++ tests/test_store.c | 27 ++++++++++++++++++++++++ 7 files changed, 84 insertions(+), 6 deletions(-) diff --git a/proxy.conf.example b/proxy.conf.example index 56510f9..c0653e0 100644 --- a/proxy.conf.example +++ b/proxy.conf.example @@ -181,6 +181,12 @@ commit_max_shares = 100 # reads this table; the ledger lives in shares and rate_history. templates_retention_days = 30 +# Reject rows written per second, pool-wide; the rest are not recorded (the +# miner still gets its error). A reject costs the sender no work, so without a +# cap a flood of them grows the rejects table without bound. Default 0 = no +# limit. +# rejects_max_per_sec = 0 + # Redis broadcast (optional). When set, accepted shares / rejects / blocks / # tip changes are PUBLISHed on channels pool:shares / pool:rejects / # pool:blocks / pool:tip as JSON. SQLite remains the source of truth — the diff --git a/src/config.c b/src/config.c index f63e27d..3eecc1d 100644 --- a/src/config.c +++ b/src/config.c @@ -306,6 +306,7 @@ int proxy_config_load(const char *path, proxy_config_t *cfg, else if (strcmp(k, "commit_window_ms") == 0) cfg->commit_window_ms = atoi(v); else if (strcmp(k, "commit_max_shares") == 0) cfg->commit_max_shares = atoi(v); else if (strcmp(k, "templates_retention_days") == 0) cfg->templates_retention_days = atoi(v); + else if (strcmp(k, "rejects_max_per_sec") == 0) cfg->rejects_max_per_sec = atoi(v); else if (strcmp(k, "redis_url") == 0) copy_str(cfg->redis_url, sizeof cfg->redis_url, v); else if (strcmp(k, "redis_publish_timeout_ms") == 0) cfg->redis_publish_timeout_ms = atoi(v); else if (strcmp(k, "redis_reconnect_backoff_ms")== 0) cfg->redis_reconnect_backoff_ms = atoi(v); diff --git a/src/config.h b/src/config.h index f6d9032..2bfb876 100644 --- a/src/config.h +++ b/src/config.h @@ -123,6 +123,7 @@ typedef struct { int commit_window_ms; int commit_max_shares; int templates_retention_days; /* template history kept; 0 = forever */ + int rejects_max_per_sec; /* reject rows written per second; 0 = no limit */ /* redis broadcast — optional. Empty url disables the module. */ char redis_url[256]; diff --git a/src/main.c b/src/main.c index b059098..d42159d 100644 --- a/src/main.c +++ b/src/main.c @@ -1325,6 +1325,7 @@ int main(int argc, char **argv) { scfg.commit_window_ms = cfg.commit_window_ms; scfg.commit_max_shares = cfg.commit_max_shares; scfg.templates_retention_days = cfg.templates_retention_days; + scfg.rejects_max_per_sec = cfg.rejects_max_per_sec; store_t *store = NULL; if (store_open(&scfg, &store) < 0) { fprintf(stderr, "store_open failed for %s\n", cfg.db_path); diff --git a/src/store.c b/src/store.c index 07a0c8d..460202a 100644 --- a/src/store.c +++ b/src/store.c @@ -480,6 +480,9 @@ struct store { int commit_window_ms; int commit_max_shares; int templates_retention_days; /* 0 = keep every row */ + int rejects_max_per_sec; /* 0 = no limit */ + uint64_t reject_sec; /* under mu: the second being counted */ + int rejects_in_sec; /* under mu */ worker_slot_t cache[WORKER_CACHE_SLOTS]; @@ -826,9 +829,10 @@ static void *writer_main(void *arg) { /* ---- enqueue ---------------------------------------------------------- */ -static int enqueue(store_t *s, const event_t *ev) { +/* Enqueues only while fewer than `limit` events are waiting. */ +static int enqueue_below(store_t *s, const event_t *ev, size_t limit) { pthread_mutex_lock(&s->mu); - if (s->ring_count == s->ring_cap) { + if (s->ring_count >= limit) { pthread_mutex_unlock(&s->mu); return -1; } @@ -841,6 +845,30 @@ static int enqueue(store_t *s, const event_t *ev) { return 0; } +static int enqueue(store_t *s, const event_t *ev) { + return enqueue_below(s, ev, s->ring_cap); +} + +/* Rejects take no work to send, so a flood of them must not crowd shares and + * blocks out of the ring: they stop at three quarters full. */ +static int enqueue_reject(store_t *s, const event_t *ev) { + return enqueue_below(s, ev, s->ring_cap - s->ring_cap / 4); +} + +static int reject_within_rate(store_t *s, uint64_t ts_ms) { + if (s->rejects_max_per_sec <= 0) return 1; + pthread_mutex_lock(&s->mu); + uint64_t sec = ts_ms / 1000; + if (sec != s->reject_sec) { + s->reject_sec = sec; + s->rejects_in_sec = 0; + } + int ok = s->rejects_in_sec < s->rejects_max_per_sec; + if (ok) s->rejects_in_sec++; + pthread_mutex_unlock(&s->mu); + return ok; +} + /* ---- public API ------------------------------------------------------- */ int store_open(const store_cfg_t *cfg, store_t **out) { @@ -852,6 +880,8 @@ int store_open(const store_cfg_t *cfg, store_t **out) { s->commit_max_shares = cfg->commit_max_shares > 0 ? cfg->commit_max_shares : 100; s->templates_retention_days = cfg->templates_retention_days > 0 ? cfg->templates_retention_days : 0; + s->rejects_max_per_sec = + cfg->rejects_max_per_sec > 0 ? cfg->rejects_max_per_sec : 0; s->ring_cap = g_test_ring_cap > 0 ? g_test_ring_cap : 65536; s->ring = calloc(s->ring_cap, sizeof(event_t)); if (!s->ring) { free(s); return -1; } @@ -1046,13 +1076,14 @@ int store_record_reject(store_t *s, const char *worker_name, uint64_t ts_ms, const char *reason) { if (!s || !reason) return -1; + if (!reject_within_rate(s, ts_ms)) return 0; event_t ev; memset(&ev, 0, sizeof(ev)); ev.kind = EV_REJECT; ev.ts_ms = ts_ms; if (worker_name) strncpy(ev.worker_name, worker_name, WORKER_NAME_MAX - 1); strncpy(ev.reason, reason, REASON_MAX - 1); - if (enqueue(s, &ev) != 0) { + if (enqueue_reject(s, &ev) != 0) { atomic_fetch_add(&s->shares_dropped, 1); return -1; } @@ -1301,9 +1332,17 @@ int store_record_block(store_t *s, uint64_t ts_ms, int height, if (finder_name) strncpy(ev.worker_name, finder_name, WORKER_NAME_MAX - 1); if (finder_address) strncpy(ev.payout_address, finder_address, ADDR_MAX - 1); - if (enqueue(s, &ev) != 0) { - atomic_fetch_add(&s->shares_dropped, 1); - return -1; + /* A found block is never dropped for a full ring: wait (up to 30 s) for + * the writer to make room. */ + for (int waited_ms = 0; enqueue(s, &ev) != 0; waited_ms += 10) { + if (waited_ms >= 30000) { + LOG_ERROR("store: ring full for 30 s; block %s was NOT recorded", + hash); + atomic_fetch_add(&s->shares_dropped, 1); + return -1; + } + struct timespec ts = { 0, 10 * 1000 * 1000 }; + nanosleep(&ts, NULL); } return 0; } diff --git a/src/store.h b/src/store.h index 47ad68e..f15e8a4 100644 --- a/src/store.h +++ b/src/store.h @@ -15,6 +15,9 @@ typedef struct { * new template row is opened, so a busy pool trims itself and an idle * one leaves the table alone. */ int templates_retention_days; + /* Most reject rows written per second, pool-wide; the rest are dropped + * (rejects cost a miner nothing to send). 0 = no limit. */ + int rejects_max_per_sec; } store_cfg_t; /* Open the DB (creates file + applies schema if missing). Starts a writer diff --git a/tests/test_store.c b/tests/test_store.c index 7fcafaf..9ae4fe4 100644 --- a/tests/test_store.c +++ b/tests/test_store.c @@ -374,6 +374,32 @@ static void test_rejects(void) { printf(" ok test_rejects\n"); } +/* rejects_max_per_sec bounds the rows a reject flood can write: rejects cost + * the sender no work. Only the per-second budget lands; the next second has a + * fresh one. */ +static void test_rejects_rate_limited(void) { + const char *path = fresh_db_path(); + store_cfg_t cfg = {0}; + snprintf(cfg.path, sizeof(cfg.path), "%s", path); + cfg.commit_window_ms = 20; + cfg.commit_max_shares = 100; + cfg.rejects_max_per_sec = 5; + + store_t *s = NULL; + assert(store_open(&cfg, &s) == 0); + for (int i = 0; i < 50; ++i) + assert(store_record_reject(s, "flood", 5000 + (uint64_t)i, "unknown job") == 0); + assert(store_record_reject(s, "flood", 6000, "unknown job") == 0); + assert(store_flush(s) == 0); + + sqlite3 *db = NULL; + assert(sqlite3_open(path, &db) == SQLITE_OK); + assert(scalar_i64(db, "SELECT count(*) FROM rejects") == 6); + sqlite3_close(db); + store_close(s); + printf(" ok test_rejects_rate_limited\n"); +} + typedef struct { store_t *s; int tid; @@ -2047,6 +2073,7 @@ int main(void) { printf("running test_store...\n"); test_basic(); test_rejects(); + test_rejects_rate_limited(); test_concurrent(); test_drop(); test_credited_sats();