myxce/src/controller/testsuite/test_persist_checkpoint_offload.c

/* test_persist_checkpoint_offload.c — xce-audit-invariants controller pass,
 * dimension 1 async-I/O thread-role fix, bug-1694.
 *
 * Proves checkpoint_task_fn (persist.c) no longer blocks the delaymgr thread
 * with real (unbounded) checkpoint I/O: the trigger now only releases the
 * "scheduled" slot and submits the real work to the cluster's slow lane
 * (slowlane_submit), returning immediately regardless of how long the real
 * checkpoint body takes. Mirrors test_ha_liveness.c's reconnect/configpush
 * slow-lane offload tests (rc_run_offload_tick / test_configpush_offload_
 * runs_on_lane) exactly: a gated XCE_TESTING hook (persist_checkpoint_
 * test_hook) substitutes for the real (disk-backed) persist_checkpoint()
 * call so the test can park the checkpoint body in flight and observe which
 * thread it runs on, without needing a full WAL/snapshot fixture.
 *
 * checkpoint_task_fn's only production call site is delay_task (1s
 * coalescing delay) — persist_checkpoint_task_fn_for_test drives it directly
 * so the test doesn't have to wait out a real delaymgr timer.
 *
 * Fixture is deliberately minimal (bare zeroed persist_t + cluster_t, no
 * persist_init/persist_start): every field checkpoint_task_fn/checkpoint_
 * lane_fn/run_checkpoint_body touch — checkpoint_in_progress,
 * wal_record_count (plain atomics), cluster->slowlane — is satisfied without
 * a real WAL. The XCE_TESTING hook path never reaches the real
 * persist_checkpoint() (which WOULD need a real disk-backed persist_t), so
 * this is safe. Mirrors test_ha_liveness.c's rc_setup_primary /
 * test_configpush_offload_runs_on_lane, which likewise use a bare
 * memset-zeroed cluster_t with only ha_init + slowlane_init.
 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <string.h>
#include <time.h>

#include <unity/unity.h>

#include "persist.h"
#include "slowlane.h"
#include "xce/cluster_def.h"

/* Test seams (persist.c, compiled under XCE_TESTING — controller_unit_test
 * always builds with it). */
extern int  persist_checkpoint_task_fn_for_test(persist_t *p);
extern void (*persist_checkpoint_test_hook)(persist_t *);

/* ── park/release gate + thread-recording probe ──────────────────────────
 * Mirrors test_ha_liveness.c's cp_probe/cp_wait_parked/cp_wait_runs/
 * cp_release_gate/cp_reset_probe (configpush slow-lane offload, raft-drive-
 * async task 3.3) exactly, renamed for checkpoint. */
static _Atomic int       g_ckpt_runs = 0;
static _Atomic pthread_t g_ckpt_last_thread;
static _Atomic int       g_ckpt_gate_on = 0;
static pthread_mutex_t   g_ckpt_mu         = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t    g_ckpt_release_cv = PTHREAD_COND_INITIALIZER;
static int               g_ckpt_parked  = 0;
static int               g_ckpt_release = 0;

static void ckpt_probe(persist_t *p)
{
    (void)p;
    atomic_store_explicit(&g_ckpt_last_thread, pthread_self(),
                          memory_order_relaxed);
    if (atomic_load(&g_ckpt_gate_on)) {
        pthread_mutex_lock(&g_ckpt_mu);
        g_ckpt_parked = 1;
        while (!g_ckpt_release)
            pthread_cond_wait(&g_ckpt_release_cv, &g_ckpt_mu);
        pthread_mutex_unlock(&g_ckpt_mu);
    }
    atomic_fetch_add(&g_ckpt_runs, 1);
}

/* Bounded poll until the gated body parks; 0 on park, -1 on timeout. */
static int ckpt_wait_parked(int timeout_ms)
{
    pthread_mutex_lock(&g_ckpt_mu);
    int parked = 0;
    for (int i = 0; i < timeout_ms; i++) {
        if (g_ckpt_parked) { parked = 1; break; }
        pthread_mutex_unlock(&g_ckpt_mu);
        struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
        nanosleep(&ts, NULL);
        pthread_mutex_lock(&g_ckpt_mu);
    }
    pthread_mutex_unlock(&g_ckpt_mu);
    return parked ? 0 : -1;
}

static int ckpt_wait_runs(int want, int timeout_ms)
{
    for (int i = 0; i < timeout_ms; i++) {
        if (atomic_load(&g_ckpt_runs) >= want) break;
        struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
        nanosleep(&ts, NULL);
    }
    return atomic_load(&g_ckpt_runs);
}

static void ckpt_release_gate(void)
{
    pthread_mutex_lock(&g_ckpt_mu);
    g_ckpt_release = 1;
    pthread_cond_broadcast(&g_ckpt_release_cv);
    pthread_mutex_unlock(&g_ckpt_mu);
}

static void ckpt_reset_probe(void)
{
    atomic_store(&g_ckpt_runs, 0);
    atomic_store(&g_ckpt_gate_on, 0);
    pthread_mutex_lock(&g_ckpt_mu);
    g_ckpt_parked  = 0;
    g_ckpt_release = 0;
    pthread_mutex_unlock(&g_ckpt_mu);
}

/* Elapsed milliseconds between two CLOCK_MONOTONIC timespecs. */
static double ms_since(const struct timespec *t0, const struct timespec *t1)
{
    return (double)(t1->tv_sec - t0->tv_sec) * 1000.0 +
           (double)(t1->tv_nsec - t0->tv_nsec) / 1e6;
}

/* Minimal fixture: bare zeroed persist_t + cluster_t, real slow lane. See
 * file header for why this is sufficient (the hook path never reaches real
 * disk I/O). */
static void ckpt_setup(persist_t *p, cluster_t *c)
{
    memset(p, 0, sizeof(*p));
    memset(c, 0, sizeof(*c));
    TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c->slowlane, 2, 4));
    p->cluster = c;
    atomic_store(&p->checkpoint_in_progress, 1); /* mirrors schedule_checkpoint's
                                                   * CAS having claimed the slot
                                                   * before delay_task fires the
                                                   * trigger */
    atomic_store(&p->wal_record_count, 999);
}

static void ckpt_teardown(cluster_t *c)
{
    slowlane_fini(&c->slowlane);
}

/*
 * GREEN (Req: delaymgr thread must never block on checkpoint I/O): with the
 * offload flag ON (the production default), triggering checkpoint_task_fn
 * returns promptly — well under the gate's hold duration — while the real
 * checkpoint body is still parked in flight on a SLOW-LANE worker thread,
 * not the caller's own thread. This is the direct proof that delaymgr's own
 * worker thread (which, in production, is the one calling checkpoint_task_fn)
 * would never observe the checkpoint body's blocking duration.
 */
static void test_checkpoint_offload_trigger_returns_promptly(void)
{
    persist_t p;
    cluster_t c;
    ckpt_setup(&p, &c);

    ckpt_reset_probe();
    atomic_store(&g_ckpt_gate_on, 1); /* lane body parks at the gate */
    persist_checkpoint_test_hook = ckpt_probe;
    persist_set_checkpoint_offload_for_test(1);

    pthread_t caller = pthread_self();

    struct timespec t0, t1;
    clock_gettime(CLOCK_MONOTONIC, &t0);
    int rc = persist_checkpoint_task_fn_for_test(&p);
    clock_gettime(CLOCK_MONOTONIC, &t1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    double trigger_ms = ms_since(&t0, &t1);
    TEST_ASSERT_TRUE_MESSAGE(trigger_ms < 200.0,
        "checkpoint_task_fn (the delaymgr-thread trigger) must return "
        "promptly — it must never block on the real checkpoint body");

    /* The "scheduled" slot must already be released so a future
     * schedule_checkpoint() call is not permanently locked out. */
    TEST_ASSERT_EQUAL_INT(0, atomic_load(&p.checkpoint_in_progress));

    /* The real checkpoint body must actually be running (parked) — proving
     * the trigger's prompt return is NOT because the work was dropped. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, ckpt_wait_parked(2000),
        "the offloaded checkpoint body must reach the gate, running "
        "somewhere off the trigger's own call stack");
    TEST_ASSERT_FALSE_MESSAGE(
        pthread_equal(atomic_load_explicit(&g_ckpt_last_thread,
                                            memory_order_relaxed), caller),
        "the checkpoint body must run on a SLOW-LANE worker thread, NOT the "
        "trigger's own (delaymgr, in production) thread");

    /* wal_record_count must NOT have been reset yet — the body hasn't
     * finished (still parked at the gate). */
    TEST_ASSERT_EQUAL_UINT(999, atomic_load(&p.wal_record_count));

    ckpt_release_gate();
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, ckpt_wait_runs(1, 2000),
        "the checkpoint body must complete exactly once after release");
    /* Only after the body actually completes does the counter reset. */
    TEST_ASSERT_EQUAL_UINT(0, atomic_load(&p.wal_record_count));

    persist_checkpoint_test_hook = NULL;
    persist_set_checkpoint_offload_for_test(1); /* restore production default */
    ckpt_teardown(&c);
    printf("test_checkpoint_offload_trigger_returns_promptly: PASS\n");
}

/*
 * RED (pre-fix behavior, demonstrated via the flag's OFF/kill-switch path):
 * with the offload flag OFF, checkpoint_task_fn falls back to running the
 * checkpoint body INLINE, on the calling thread — exactly the delaymgr-
 * blocking violation this fix closes. The trigger call itself now BLOCKS
 * until the gate is released, and the body runs on the CALLER's own thread.
 * This is what checkpoint_task_fn did unconditionally before this fix; with
 * the flag flipped OFF the same legacy code path is still reachable, so this
 * test demonstrates the exact violation the GREEN test above proves fixed.
 */
typedef struct { persist_t *p; _Atomic int done; int rc; } ckpt_trigger_arg_t;

static void *ckpt_trigger_thread(void *arg)
{
    ckpt_trigger_arg_t *a = (ckpt_trigger_arg_t *)arg;
    a->rc = persist_checkpoint_task_fn_for_test(a->p);
    atomic_store(&a->done, 1);
    return NULL;
}

static void test_checkpoint_inline_blocks_delaymgr_thread_RED(void)
{
    persist_t p;
    cluster_t c;
    ckpt_setup(&p, &c);

    ckpt_reset_probe();
    atomic_store(&g_ckpt_gate_on, 1);
    persist_checkpoint_test_hook = ckpt_probe;
    persist_set_checkpoint_offload_for_test(0); /* RED: legacy inline path */

    ckpt_trigger_arg_t arg = { .p = &p };
    atomic_store(&arg.done, 0);

    pthread_t th;
    pthread_create(&th, NULL, ckpt_trigger_thread, &arg);

    /* The body must reach the gate — proving it ran (inline, synchronously,
     * as part of the trigger call itself), unlike the offloaded case. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, ckpt_wait_parked(2000),
        "RED: the inline checkpoint body must still run (on the delaymgr/"
        "trigger thread itself) with the offload flag OFF");

    /* While parked, the trigger call has NOT returned — this is the
     * blocking-delaymgr-thread violation. */
    struct timespec ts = { .tv_sec = 0, .tv_nsec = 100 * 1000000L }; /* 100ms */
    nanosleep(&ts, NULL);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&arg.done),
        "RED: with the offload flag OFF, checkpoint_task_fn must still be "
        "blocked inside the checkpoint body — reproducing the pre-fix "
        "delaymgr-thread-blocking violation");

    /* The body ran on the SAME thread as the trigger — no offload. */
    TEST_ASSERT_TRUE_MESSAGE(
        pthread_equal(atomic_load_explicit(&g_ckpt_last_thread,
                                            memory_order_relaxed), th),
        "RED: with the offload flag OFF the checkpoint body must run on the "
        "trigger's own (delaymgr) thread, not a lane worker");

    ckpt_release_gate();
    pthread_join(th, NULL);
    TEST_ASSERT_EQUAL_INT(0, arg.rc);
    TEST_ASSERT_EQUAL_INT(1, atomic_load(&arg.done));

    persist_checkpoint_test_hook = NULL;
    persist_set_checkpoint_offload_for_test(1); /* restore production default */
    ckpt_teardown(&c);
    printf("test_checkpoint_inline_blocks_delaymgr_thread_RED: PASS\n");
}

/*
 * No-cluster fallback (unit-test / bare persist_t, e.g. most of this file's
 * siblings): checkpoint_task_fn's own production call site is only ever
 * reached via schedule_checkpoint's cluster branch, but this proves the
 * defensive persist->cluster == NULL check inside checkpoint_task_fn itself
 * also takes the inline path (no crash dereferencing a NULL cluster's
 * slowlane).
 */
static void test_checkpoint_no_cluster_runs_inline(void)
{
    persist_t p;
    memset(&p, 0, sizeof(p));
    p.cluster = NULL;
    atomic_store(&p.checkpoint_in_progress, 1);
    atomic_store(&p.wal_record_count, 42);

    ckpt_reset_probe();
    persist_checkpoint_test_hook = ckpt_probe;
    persist_set_checkpoint_offload_for_test(1); /* even with flag ON */

    int rc = persist_checkpoint_task_fn_for_test(&p);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* No lane to submit to (cluster == NULL) — must run synchronously,
     * inline, before the trigger call returns. */
    TEST_ASSERT_EQUAL_INT(1, atomic_load(&g_ckpt_runs));
    TEST_ASSERT_EQUAL_UINT(0, atomic_load(&p.wal_record_count));
    TEST_ASSERT_EQUAL_INT(0, atomic_load(&p.checkpoint_in_progress));

    persist_checkpoint_test_hook = NULL;
    printf("test_checkpoint_no_cluster_runs_inline: PASS\n");
}

void test_persist_checkpoint_offload(void)
{
    UnitySetTestFile(__FILE__);
    RUN_TEST(test_checkpoint_offload_trigger_returns_promptly);
    RUN_TEST(test_checkpoint_inline_blocks_delaymgr_thread_RED);
    RUN_TEST(test_checkpoint_no_cluster_runs_inline);
}