fix(daemon): harden bounded per-source registry races

Stamp host_last_use before publishing a bucket key and treat an unstamped
(last_use == 0) bucket as live, so a just-claimed bucket can no longer be
stolen by a concurrent reclaimer.

After a successful eviction CAS, re-scan for the interned key and, when an
earlier bucket already holds it, zero the duplicate's active count and
return the canonical bucket, preventing orphaned per-host counts and cap
overshoot under full-table concurrency.

Add a message-carrying EXPECT_FAIL primitive and use it for the daemon-conf
buffer-overflow guard, and add a fork-based test that records auth failures
from forked children and asserts the parent observes the shared lockout.

src/shared/daemon_limits.c

@@ -141,7 +141,11 @@ static bool host_bucket_reclaimable(DaemonLimitRegistry* registry, size_t idx, l
   if (until != 0)
     return until <= now;
   long long last_use = atomic_load_explicit(&registry->host_last_use[idx], memory_order_relaxed);
-  return last_use == 0 || now - last_use >= DAEMON_LIMITS_HOST_EVICT_IDLE_SEC;
+  /* A bucket whose key is published but whose last_use has not yet been stamped
+   * (last_use == 0) must be treated as live: reclaiming it here would steal a
+   * bucket a racing child just claimed.  The claim path also stamps last_use
+   * before publishing the key, so this window cannot persist. */
+  return last_use != 0 && now - last_use >= DAEMON_LIMITS_HOST_EVICT_IDLE_SEC;
 }
 
 /* Emit at most one "per-source table full" warning per
@@ -191,14 +195,18 @@ static int host_intern(DaemonLimitRegistry* registry, const char* peer_ip) {
         return (int)idx;
       }
       if (current == 0) {
+        /* Stamp last_use *before* publishing the key so a reclaimer racing the
+         * claim can never observe a claimed bucket with last_use == 0 and
+         * evict it.  A pre-stamp is harmless if the CAS loses: the bucket is
+         * either still empty (never inspected for reclaim) or has just been
+         * taken by another source that wants a fresh timestamp anyway. */
+        atomic_store_explicit(&registry->host_last_use[idx], now, memory_order_relaxed);
         uint64_t expected = 0;
         if (atomic_compare_exchange_strong_explicit(&registry->host_key[idx], &expected, key,
                                                     memory_order_acq_rel, memory_order_acquire)) {
-          atomic_store_explicit(&registry->host_last_use[idx], now, memory_order_relaxed);
           return (int)idx;
         }
         if (atomic_load_explicit(&registry->host_key[idx], memory_order_acquire) == key) {
-          atomic_store_explicit(&registry->host_last_use[idx], now, memory_order_relaxed);
           return (int)idx;
         }
         continue; /* another child won this empty bucket; keep probing */
@@ -209,6 +217,9 @@ static int host_intern(DaemonLimitRegistry* registry, const char* peer_ip) {
       }
     }
     if (evict >= 0) {
+      /* Refresh the timestamp before the key changes hands so the reused bucket
+       * is not seen as immediately idle by a racing reclaimer. */
+      atomic_store_explicit(&registry->host_last_use[evict], now, memory_order_relaxed);
       uint64_t expected = evict_key;
       if (atomic_compare_exchange_strong_explicit(&registry->host_key[evict], &expected, key,
                                                   memory_order_acq_rel, memory_order_acquire)) {
@@ -217,7 +228,27 @@ static int host_intern(DaemonLimitRegistry* registry, const char* peer_ip) {
         atomic_store_explicit(&registry->host_active[evict], 0, memory_order_relaxed);
         atomic_store_explicit(&registry->host_fail[evict], 0, memory_order_relaxed);
         atomic_store_explicit(&registry->host_until[evict], 0, memory_order_relaxed);
-        atomic_store_explicit(&registry->host_last_use[evict], now, memory_order_relaxed);
+        /* Two children can race to intern the same brand-new key into different
+         * eviction targets, leaving the table with duplicate buckets for `key`.
+         * Re-scan for the first (canonical) bucket holding `key`; when it
+         * precedes `evict`, drop our duplicate's occupancy and hand back the
+         * canonical bucket so per-source counts are not orphaned on the
+         * duplicate.  The duplicate keeps its key, so no tombstone hole is
+         * created and probe chains stay intact; it ages out normally. */
+        for (size_t i = 0; i < (size_t)registry->host_slots; i++) {
+          size_t candidate = (start + i) & mask;
+          uint64_t found =
+              atomic_load_explicit(&registry->host_key[candidate], memory_order_acquire);
+          if (found == key) {
+            if (candidate != (size_t)evict) {
+              atomic_store_explicit(&registry->host_active[evict], 0, memory_order_relaxed);
+              return (int)candidate;
+            }
+            break;
+          }
+          if (found == 0)
+            break; /* the key is present at `evict`, so this cannot happen first */
+        }
         return evict;
       }
       continue; /* lost the race; re-probe with fresh observations */

tests/test_daemon_conf.c

@@ -550,7 +550,7 @@ static void test_daemon_conf_module_count_capped() {
     int n = snprintf(line, sizeof(line), "[m%zu]\npath = /x\n", i);
     if (n < 0 || (size_t)n >= sizeof(line) || used + (size_t)n >= len) {
       free(body);
-      EXPECT_TRUE(0 && "module-count test buffer overflow");
+      EXPECT_FAIL("module-count test buffer overflow");
       return;
     }
     memcpy(body + used, line, (size_t)n);

tests/test_daemon_limits.c

@@ -189,6 +189,44 @@ static void test_daemon_limits_fork_shared() {
   daemon_limits_destroy(registry);
 }
 
+/* Cross-process auth lockout: failures recorded by forked children against the
+ * shared mmap must lock the source out for the parent.  This is the
+ * cross-process path the integration test can no longer cover because trusted
+ * loopback peers are exempt from the per-host limits. */
+static void test_daemon_limits_fork_auth_lockout() {
+  if (is_running_under_valgrind())
+    return; /* fork + shared mapping is slow/noisy under valgrind */
+  DaemonLimitRegistry* registry = daemon_limits_create(DAEMON_LIMITS_MIN_SLOTS, 1, 0, 2, 300);
+  EXPECT_NOT_NULL(registry);
+
+  int remaining = 0;
+  EXPECT_FALSE(daemon_limits_auth_locked(registry, "10.0.0.1", &remaining));
+
+  /* One failure from each of two children reaches the threshold of 2 in the
+   * shared mapping; atomics only, no mtx/malloc, so fork-safe. */
+  for (int i = 0; i < 2; i++) {
+    pid_t pid = fork();
+    if (pid == 0) {
+      daemon_limits_auth_record_failure(registry, "10.0.0.1");
+      _exit(0);
+    }
+    EXPECT_TRUE(pid > 0);
+    int status = 0;
+    EXPECT_TRUE(waitpid(pid, &status, 0) == pid);
+    EXPECT_TRUE(WIFEXITED(status) && WEXITSTATUS(status) == 0);
+  }
+
+  /* The parent observes the lockout the children established. */
+  EXPECT_TRUE(daemon_limits_auth_locked(registry, "10.0.0.1", &remaining));
+  EXPECT_TRUE(remaining > 0 && remaining <= 300);
+  /* A different source is unaffected across processes. */
+  EXPECT_FALSE(daemon_limits_auth_locked(registry, "10.0.0.2", &remaining));
+  /* The parent clears the shared lockout on a successful authentication. */
+  daemon_limits_auth_record_success(registry, "10.0.0.1");
+  EXPECT_FALSE(daemon_limits_auth_locked(registry, "10.0.0.1", &remaining));
+  daemon_limits_destroy(registry);
+}
+
 /* The occupancy arrays are derived from the slot table: recompute rebuilds them
  * and is the self-heal path the SIGCHLD handler uses after a child dies. */
 static void test_daemon_limits_recompute() {
@@ -269,4 +307,5 @@ void test_daemon_limits() {
   test_daemon_limits_auth_lockout();
   test_daemon_limits_host_table_eviction();
   test_daemon_limits_fork_shared();
+  test_daemon_limits_fork_auth_lockout();
 }

tests/test_utils.h

@@ -112,4 +112,12 @@ extern bool current_test_failed;
     }                                                                                              \
   } while (0)
 
+/* Unconditional test failure carrying an explanatory message. */
+#define EXPECT_FAIL(message)                                                                       \
+  do {                                                                                             \
+    printf("    \033[1;31m[FAIL]\033[0m %s:%d: %s\n", __FILE__, __LINE__, (message));              \
+    current_test_failed = true;                                                                    \
+    return;                                                                                        \
+  } while (0)
+
 #endif