/*
 * 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);
}