store: bound what a reject flood can write, and never drop a found block

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

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);

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];

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);

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;
 }

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

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();