feat(ha-convergence-recovery): commit-gated scheduler egress — submit_lsn tagging + watermark gate (task 1.2)

src/controller/egress.c

@@ -0,0 +1,109 @@
+/*
+ * egress.c — commit-gated scheduler egress (ha-convergence-recovery, Req 3).
+ *
+ * The scheduler is the one consumer whose actions cannot be revoked, so it must
+ * never observe a job whose submit entry is uncommitted (orphan-dispatch
+ * prevention).  This file holds the commit-watermark predicate and the
+ * scheduler-bound collect wrapper that applies it.
+ *
+ *   job_submit_committed(c, j)
+ *       submit_lsn == 0 || submit_lsn <= committed_lsn
+ *       One atomic load + one compare; lock-free and infallible.
+ *       submit_lsn == 0 (pre-feature / direct create) is always exportable so
+ *       legacy and replayed jobs are never stranded (Req 3.1, 3.2).
+ *
+ *   collect_jobs_committed(c, fn, l)
+ *       Identical contract to collect_jobs (dataex.c) — collects matching jobs,
+ *       references and READ-locks each, leaving them for unlock_collect_jobs —
+ *       additionally excluding any job whose submit entry is uncommitted.  For
+ *       scheduler-bound callers ONLY (xcesched_push_jobs).  Status/API read
+ *       paths keep plain collect_jobs (Req 3.4 — read consistency out of scope).
+ *
+ * Lives in its own translation unit (not dataex.c) so the controller unit test
+ * binary can link the REAL predicate without pulling in dataex.c's notify() /
+ * collect_confvars() definitions, which the test suite already stubs.  This is
+ * the same egress-filter component the design assigns to dataex.c, factored out
+ * for testability (D3 — refinement, not contradiction).
+ */
+
+/* rtti.h first — defines 'var'/'typeinfo_t' used by kvmgr.h (via cluster.h). */
+#include <rtti.h>
+
+#include <errno.h>
+#include <stdatomic.h>
+#include <stdbool.h>
+#include <stdint.h>
+
+#include "cluster.h"
+#include "persist.h"
+#include "utils.h"   /* CONTAINER_OF */
+#include "xce/cluster_def.h"
+#include "xce/types.h"
+#include "xlist.h"
+
+/* Commit-watermark predicate (Req 3.1, 3.2).  Lock-free: one atomic load of
+ * committed_lsn and one comparison.  submit_lsn == 0 means unknown/pre-feature
+ * and is always exportable. */
+bool job_submit_committed(cluster_t* c, job_t* j)
+{
+    if (j == NULL)
+        return false;
+    if (j->submit_lsn == 0)
+        return true;
+    uint64_t committed
+        = atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire);
+    return j->submit_lsn <= committed;
+}
+
+/* Read-lock each collected job, mirroring collect_jobs' lock_collect_jobs
+ * (dataex.c) so the contract — caller must release via unlock_collect_jobs — is
+ * identical for scheduler-bound callers. */
+static void egress_lock_collected(xlist_t* l, int op)
+{
+    for (int i = 0; i < l->count; i++) {
+        xce_job_t* xj = (xce_job_t*)xlist_get(l, i);
+        if (xj == NULL)
+            continue;
+        job_t* j = CONTAINER_OF(xj, job_t, job);
+        lock(j, op);
+    }
+}
+
+/* Scheduler-bound collect: same iteration + reference + read-lock contract as
+ * collect_jobs (dataex.c:294), additionally gated on the commit watermark.
+ * `fn`, when non-NULL, is an additional caller predicate applied first (matches
+ * collect_jobs' check_job_ usage in xcesched_push_jobs). */
+void collect_jobs_committed(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
+{
+    iterator_t it;
+
+    lock(c->jobmgr, RTTI_LOCK_READ);
+    if (iter_make(&it, c->jobmgr) < 0) {
+        unlock(c->jobmgr);
+        return;
+    }
+
+    for (;;) {
+        job_t* j  = NULL;
+        int    rc = iter_next(&it, (void**)&j);
+        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA))
+            break;
+        if (j == NULL)
+            continue;
+
+        /* Caller predicate first (parity with collect_jobs), then the commit
+         * watermark gate — exclude jobs whose submit entry is uncommitted. */
+        if ((fn != NULL) && (!fn(j)))
+            continue;
+        if (!job_submit_committed(c, j))
+            continue;
+
+        reference(j);
+        xlist_store(l, &(j->job));
+    }
+
+    iter_destroy(&it);
+    unlock(c->jobmgr);
+
+    egress_lock_collected(l, RTTI_LOCK_READ);
+}