/*
 * test_commit_admission_pool.c — raft-drive-async task 1.6/1.7.
 *
 * The commit-wait admission cap is FOLDED onto the vtaskmgr pool's single
 * LivenessAdmission reservation. Asserts:
 *
 *   (1) with a pool wired and R = VTASK_LIVENESS_RESERVED set, commit_submit
 *       (WAIT) fast-rejects with -XCE_ERR_NOT_READY at the POOL cap
 *       (n_vcpus - R), not the old local COMMIT_WAIT_RESERVED cap. This is the
 *       intended cap-value change: n_peers=5 → R=4 → cap=1 (TIGHTER than the
 *       old n_vcpus-2=6).
 *   (2) commit-wait still fast-rejects (never blocks) at the cap — backpressure
 *       preserved.
 *   (3) releasing a pool slot (a waiter completing) re-admits the next wait.
 *   (4) no-pool fallback still uses the local COMMIT_WAIT_RESERVED cap
 *       (regression guard for the existing commit fixtures).
 *
 * Deterministic: a background thread parks in commit_submit(WAIT) on an LSN
 * that cannot commit (no acks, n_peers>1) with a long timeout, holding the
 * admission slot; the test polls ce->inflight_waits to confirm saturation
 * before probing the next admit. No correctness-by-sleep on assertions.
 *
 * Requirements: 3.1, 3.2, 3.3, 4.5.
 */
#include <pthread.h>
#include <stdatomic.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include <unity/unity.h>

#include "xce/errnox.h"

#include "commit.h"
#include "repl.h"
#include "vtaskmgr.h"

static persist_t       g_p;
static commit_engine_t g_ce;
static vtaskmgr_t      g_mgr;
static int             g_eps[MAX_STANDBYS]; /* opaque repl_ep_t tokens */

static void pool_setup(int n_peers, int n_standbys, int n_vcpus, int reserved)
{
    char name[16];

    memset(&g_p, 0, sizeof(g_p));
    TEST_ASSERT_EQUAL_INT(0, repl_init(&g_p.repl));
    for (int i = 0; i < n_standbys; i++) {
        snprintf(name, sizeof(name), "sb%d", i);
        TEST_ASSERT_EQUAL_INT(
            i, repl_add_standby(&g_p.repl, &g_eps[i], name, 7000 + (uint32_t)i));
    }
    /* Entry 1 appended + durable, but no acks → cannot commit (n_peers>1). */
    g_p.current_lsn = 2;
    atomic_store(&g_p.committed_lsn, 0);
    atomic_store(&g_p.durable_lsn, 1);

    TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_mgr, n_vcpus));
    vtaskmgr_set_liveness_reserved(&g_mgr, reserved);

    TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_p, n_peers, n_vcpus));
    if (reserved >= 0) {
        commit_set_pool(&g_ce, &g_mgr);
    }
}

static void pool_teardown(void)
{
    commit_engine_fini(&g_ce);
    vtaskmgr_fini(&g_mgr);
    repl_fini(&g_p.repl);
}

/* Background waiter: parks in commit_submit(WAIT) holding a slot. */
typedef struct {
    uint64_t lsn;
    uint32_t timeout_ms;
    int      rc;
} waiter_t;

static void* waiter_fn(void* a)
{
    waiter_t* w = (waiter_t*)a;
    w->rc       = commit_submit(&g_ce, w->lsn, COMMIT_MODE_WAIT, w->timeout_ms);
    return NULL;
}

static uint64_t mono_ms(void)
{
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (uint64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
}

static int poll_inflight(int want, uint32_t deadline_ms)
{
    uint64_t t0 = mono_ms();
    int      v;

    for (;;) {
        pthread_mutex_lock(&g_ce.lock);
        v = g_ce.inflight_waits;
        pthread_mutex_unlock(&g_ce.lock);
        if (v >= want || mono_ms() - t0 >= deadline_ms) {
            return v;
        }
        usleep(500);
    }
}

/* (1)+(2): the fold uses the POOL cap (n_vcpus - R), which differs from the
 * old local COMMIT_WAIT_RESERVED cap. n_vcpus=4, R=3 → POOL cap = 4-3 = 1
 * (TIGHTER than the old local n_vcpus-2 = 2). One parked waiter saturates the
 * pool slot; the next commit_submit(WAIT) fast-rejects without blocking.
 * (MAX_STANDBYS = 3 bounds the cluster, so a small cap is exercised here; the
 * cap-value formula is what matters, not the absolute count.) */
static void test_commit_fold_uses_pool_cap(void)
{
    const int N_VCPUS  = 4;
    const int RESERVED = 3;                  /* pool cap = 4 - 3 = 1 */
    const int POOL_CAP = N_VCPUS - RESERVED; /* 1 */
    const int OLD_LOCAL_CAP = N_VCPUS - COMMIT_WAIT_RESERVED; /* 2 — what the
                                              old local cap would have allowed */
    pthread_t tids[1];
    waiter_t  wargs[1];
    int       i;

    /* Sanity: the pool cap is genuinely TIGHTER than the old local cap — this
     * is the intended behavior change the fold introduces. */
    TEST_ASSERT_TRUE(POOL_CAP < OLD_LOCAL_CAP);

    pool_setup(/*n_peers=*/3, /*n_standbys=*/2, N_VCPUS, RESERVED);
    TEST_ASSERT_EQUAL_INT_MESSAGE(POOL_CAP, vtaskmgr_blocking_cap(&g_mgr),
        "fold must use the pool cap n_vcpus - VTASK_LIVENESS_RESERVED");

    /* Saturate the pool cap with POOL_CAP parked waiters. */
    for (i = 0; i < POOL_CAP; i++) {
        wargs[i].lsn        = 1;
        wargs[i].timeout_ms = 4000; /* long — held until we release via ack */
        wargs[i].rc         = -9999;
        TEST_ASSERT_EQUAL_INT(0,
            pthread_create(&tids[i], NULL, waiter_fn, &wargs[i]));
    }
    TEST_ASSERT_EQUAL_INT_MESSAGE(POOL_CAP, poll_inflight(POOL_CAP, 3000),
        "all POOL_CAP waiters must occupy admission slots");

    /* The over-cap wait fast-rejects WITHOUT blocking (backpressure). Under the
     * OLD local cap (2) this second wait would have been ADMITTED — proving the
     * fold now governs admission. */
    uint64_t t0 = mono_ms();
    int rc      = commit_submit(&g_ce, /*lsn=*/1, COMMIT_MODE_WAIT, /*to=*/4000);
    uint64_t dt = mono_ms() - t0;
    TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc,
        "over-pool-cap commit-wait must fast-reject with -XCE_ERR_NOT_READY");
    TEST_ASSERT_TRUE_MESSAGE(dt < 500,
        "admission rejection must not block the caller");

    /* Release the parked waiter: ack entry 1 from a majority so it commits.
     * n_peers=3 → self + 1 standby = 2 > 3/2. */
    repl_on_ack(&g_p.repl, &g_eps[0], 1);
    commit_on_ack(&g_ce, 0, 1);
    for (i = 0; i < POOL_CAP; i++) {
        pthread_join(tids[i], NULL);
        TEST_ASSERT_EQUAL_INT_MESSAGE(0, wargs[i].rc,
            "admitted waiter must complete with success after majority ack");
    }

    /* (3) Slot released → a fresh wait is admitted again (commits immediately
     * since entry 1 is now committed). */
    rc = commit_submit(&g_ce, /*lsn=*/1, COMMIT_MODE_WAIT, /*to=*/1000);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
        "after the slot is released a fresh wait must be admitted (and commit)");

    pool_teardown();
    printf("test_commit_fold_uses_pool_cap: PASS\n");
}

/* (4) No pool wired → local COMMIT_WAIT_RESERVED cap still governs. n_vcpus=2,
 * COMMIT_WAIT_RESERVED=2 → local cap = max(1, 0) = 1. One parked waiter
 * saturates; the second fast-rejects. Regression guard for existing fixtures. */
static void test_commit_local_fallback_cap_preserved(void)
{
    pthread_t tid;
    waiter_t  warg;

    /* reserved = -1 sentinel → pool_setup skips commit_set_pool (no pool). */
    pool_setup(/*n_peers=*/3, /*n_standbys=*/2, /*n_vcpus=*/2, /*reserved=*/-1);
    /* Confirm no pool is wired: the engine uses the local cap path. */

    warg.lsn        = 1;
    warg.timeout_ms = 4000;
    warg.rc         = -9999;
    TEST_ASSERT_EQUAL_INT(0, pthread_create(&tid, NULL, waiter_fn, &warg));
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, poll_inflight(1, 3000),
        "local-cap waiter must occupy the only slot (cap=1)");

    int rc = commit_submit(&g_ce, /*lsn=*/1, COMMIT_MODE_WAIT, /*to=*/4000);
    TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc,
        "over-local-cap commit-wait must fast-reject (cap = n_vcpus-RESERVED)");

    /* Release: majority ack (n_peers=3 → self + 1 standby = 2 > 3/2). */
    repl_on_ack(&g_p.repl, &g_eps[0], 1);
    commit_on_ack(&g_ce, 0, 1);
    pthread_join(tid, NULL);
    TEST_ASSERT_EQUAL_INT(0, warg.rc);

    pool_teardown();
    printf("test_commit_local_fallback_cap_preserved: PASS\n");
}

void test_commit_admission_pool(void)
{
    UnitySetTestFile(__FILE__);
    RUN_TEST(test_commit_fold_uses_pool_cap);
    RUN_TEST(test_commit_local_fallback_cap_preserved);
}