aion-server 0.26.0

//! `run_server` durable-outbox bootstrap integration tests.
//!
//! The tests launch the public `aion_server::run::run` entrypoint in a child
//! copy of this integration-test executable. That keeps the production
//! bootstrap intact (config load, `ServerState::build`, outbox dispatcher gate,
//! HTTP/gRPC transports) without adding public test APIs or manually spawning an
//! `OutboxDispatcher`.
//!
//! The end-to-end boots a real server over the durable haematite backend and
//! cold-opens the same data directory for independent assertions.

#[path = "run_server_outbox_support/helpers.rs"]
mod helpers;
#[path = "run_server_outbox_support/worker.rs"]
mod worker;

use std::io::Write;
use std::path::PathBuf;
use std::process::ExitCode;

/// Opens the runtime gate on the live-reconciliation test once the store
/// grants a concurrent observation seam (see that test's doc for the exact
/// capability). Absent, the test announces its skip and proves nothing.
const CONCURRENT_SEAM_ENV: &str = "AION_E2E_CONCURRENT_STORE_SEAM";

use aion_core::Event;
use aion_server::config::CliOverrides;
use aion_store::{OutboxRow, OutboxStatus, ReadableEventStore};
use aion_store_haematite::HaematiteStore;
use chrono::Utc;
use helpers::{
    FAN_OUT, TestError, assert_fan_out_settled, assert_task_set, count_completed,
    count_completed_for, count_kind, row_states, run_server_harness,
    run_server_harness_with_reconciliation, start_over_http, task_ordinal, test_error,
    unique_temp_dir, wait_for_history, wait_for_history_over_http, wait_for_rows, worker_result,
    write_package_archive,
};
use worker::WorkerSession;

/// The outbox — not the live in-process path — carries a fan-out, and every
/// member is dispatched and completed exactly once.
///
/// Split across two boots for one reason: haematite holds an EXCLUSIVE writer
/// lock on the data directory for the life of the server process, so the durable
/// state can only be read once that process has exited. Every assertion the
/// single-boot version made survives; only the moment of looking moved. The
/// first boot drives the system to "staged, nothing dispatched" and is then
/// stopped so the bytes can say so; the second dispatches and completes.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_outbox_happy_path_fan_out_completes_once() -> Result<(), TestError> {
    let dir = unique_temp_dir("happy")?;
    let db_path = dir.path().join("aion.db");
    let package_path = write_package_archive(dir.path())?;

    // Boot 1: start the fan-out with NO worker in existence, anywhere.
    let (server, http, _grpc) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let (workflow_id, run_id) = start_over_http(http).await?;
    wait_for_history_over_http(http, &workflow_id, "fan-out scheduled", |events| {
        count_kind(events, |event| {
            matches!(event, Event::ActivityScheduled { .. })
        }) == FAN_OUT
    })
    .await?;
    // The drain is the barrier: when it returns, the process is gone and the
    // writer lock with it. No sleep is involved in that argument.
    server.stop_gracefully()?;

    // At the bytes, with nobody holding the directory: the rows were staged by
    // the outbox path before any worker could exist, and the dispatcher worked
    // on them while waiting for one.
    {
        let reader =
            HaematiteStore::open_or_create(db_path.clone(), haematite::NodeCacheBudget::Unlimited)
                .await?;
        let history = reader.read_history(&workflow_id).await?;
        assert_eq!(
            count_kind(&history, |event| matches!(
                event,
                Event::ActivityScheduled { .. }
            )),
            FAN_OUT,
            "every fan-out member must be scheduled durably"
        );
        assert_eq!(
            count_completed(&history),
            0,
            "no member may be completed before any worker has ever registered"
        );

        let states = row_states(&reader, &workflow_id, &[0, 1, 2, 3]).await?;
        // `record_fan_out_dispatch` stages these atomically with the scheduling
        // events; the live in-process path would never write them at all. That
        // they exist is the cutover proof.
        assert_eq!(
            states.len(),
            FAN_OUT,
            "every ordinal must have a staged row"
        );
        assert!(
            states
                .iter()
                .all(|state| state.status != OutboxStatus::Done),
            "no row may be Done — a row settles on DISPATCH, and no worker ever \
             registered to be dispatched to: {states:?}"
        );
        // The single-boot version also watched, live, for the dispatcher to CLAIM
        // one of these rows while no worker existed. That claim leaves no trace
        // an outside process can read after the fact: claiming does not consume
        // an attempt, and the drain re-arms a claimed row to `Pending` with the
        // attempt budget preserved. Waiting for it here would be a race against
        // the dispatcher's own poll with no observable barrier to close it, so
        // the assertion lives where the sweep can be DRIVEN rather than awaited:
        // `worker::outbox_dispatcher::tests::failed_dispatch_retries_with_backoff_and_bumps_attempt`
        // pins that the sweep claims a staged row and offers it to the dispatch
        // seam even when that seam cannot deliver — which is how the production
        // no-worker path reports itself, since `OutboxRowDispatch::dispatch`
        // returns no-worker as an ordinary dispatch error.
    }

    // Boot 2: a worker appears, and the staged work is dispatched and completed.
    let (server, http, grpc) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let mut worker = WorkerSession::connect(grpc).await?;
    let mut tasks = Vec::with_capacity(FAN_OUT);
    for _ in 0..FAN_OUT {
        tasks.push(worker.next_task().await?);
    }
    assert_task_set(&tasks, &[0, 1, 2, 3])?;
    for task in &tasks {
        let ordinal = task_ordinal(task)?;
        worker
            .complete(task, worker_result(ordinal).as_bytes())
            .await?;
    }
    wait_for_history_over_http(http, &workflow_id, "fan-out settled", |events| {
        count_completed(events) == FAN_OUT
            && count_kind(events, |event| {
                matches!(event, Event::WorkflowCompleted { .. })
            }) == 1
    })
    .await?;
    drop(worker);
    server.stop_gracefully()?;

    let reader =
        HaematiteStore::open_or_create(db_path, haematite::NodeCacheBudget::Unlimited).await?;
    let history = assert_fan_out_settled(&reader, &workflow_id).await?;
    assert_eq!(count_completed(&history), FAN_OUT);
    std::hint::black_box(run_id);
    Ok(())
}

/// A server that dies with dispatches in flight strands their outbox rows; the
/// next boot re-arms them, re-dispatches, and the fan-out still settles exactly
/// once despite a duplicate completion.
///
/// The stranded state is asserted at the BYTES, in the window between the kill
/// and the re-boot — the one moment nothing holds haematite's writer lock. That
/// window is also where this file proves a property no other test here states:
/// an outbox row settles `Done` on DISPATCH, not on completion. All four rows
/// are `Done` while ordinals 2 and 3 have no completion recorded at all.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_outbox_restart_rearms_stranded_rows() -> Result<(), TestError> {
    let dir = unique_temp_dir("restart")?;
    let db_path = dir.path().join("aion.db");
    let package_path = write_package_archive(dir.path())?;
    let (server1, http1, grpc1) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let mut worker1 = WorkerSession::connect(grpc1).await?;

    let (workflow_id, run_id) = start_over_http(http1).await?;
    let mut tasks = Vec::with_capacity(FAN_OUT);
    for _ in 0..FAN_OUT {
        tasks.push(worker1.next_task().await?);
    }
    tasks.sort_by_key(|task| task_ordinal(task).unwrap_or(u64::MAX));
    assert_task_set(&tasks, &[0, 1, 2, 3])?;

    complete_recorded_prefix(&worker1, &tasks).await?;
    wait_for_history_over_http(http1, &workflow_id, "ordinals 0 and 1 recorded", |events| {
        count_completed_for(events, 0) == 1 && count_completed_for(events, 1) == 1
    })
    .await?;
    drop(worker1);
    // An abrupt death with ordinals 2 and 3 in flight — the production way a
    // dispatch is stranded. `stop` waits for the process, so the lock is gone.
    server1.stop()?;

    // Between the stop and the re-boot, at the bytes.
    {
        let reader =
            HaematiteStore::open_or_create(db_path.clone(), haematite::NodeCacheBudget::Unlimited)
                .await?;
        let states = row_states(&reader, &workflow_id, &[0, 1, 2, 3]).await?;
        assert!(
            states
                .iter()
                .all(|state| state.status == OutboxStatus::Done),
            "every row was dispatched, so every row is Done — a row settles on \
             dispatch, not on completion: {states:?}"
        );
        let stranded = reader.read_history(&workflow_id).await?;
        assert_eq!(count_completed_for(&stranded, 0), 1);
        assert_eq!(count_completed_for(&stranded, 1), 1);
        assert_eq!(
            count_completed_for(&stranded, 2),
            0,
            "ordinal 2's dispatch died in flight and must have recorded nothing"
        );
        assert_eq!(
            count_completed_for(&stranded, 3),
            0,
            "ordinal 3's dispatch died in flight and must have recorded nothing"
        );
    }

    // The re-boot re-arms the two stranded rows and re-dispatches them. Their
    // arrival at a fresh worker IS the re-arm: rows 2 and 3 were `Done` at the
    // bytes a moment ago, so nothing could reach a worker again without one.
    // Reading the intermediate `Pending`/`Claimed` status directly would mean
    // stopping the server inside the window between re-arm and re-dispatch,
    // which the dispatcher's own poll can close at any instant.
    let (server2, http2, grpc2) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let mut worker2 = WorkerSession::connect(grpc2).await?;
    let revived = collect_revived_tasks(&mut worker2).await?;
    assert_task_set(&revived, &[2, 3])?;
    complete_with_duplicate_first(&worker2, &revived).await?;
    wait_for_history_over_http(http2, &workflow_id, "fan-out settled", |events| {
        count_completed(events) == FAN_OUT
            && count_kind(events, |event| {
                matches!(event, Event::WorkflowCompleted { .. })
            }) == 1
    })
    .await?;
    drop(worker2);
    server2.stop_gracefully()?;

    let reader =
        HaematiteStore::open_or_create(db_path, haematite::NodeCacheBudget::Unlimited).await?;
    let history = assert_fan_out_settled(&reader, &workflow_id).await?;
    assert_eq!(count_completed(&history), FAN_OUT);
    std::hint::black_box(run_id);
    Ok(())
}

/// LIVE reconciliation re-arms a stale claim exactly once — **RED, and it must
/// stay red until haematite can serve a concurrent reader.**
///
/// ## Why this one could not be relocated like its siblings
///
/// The other two tests in this file were rewritten to drive the system with the
/// server up and assert at the bytes once it has exited. That works because
/// their subjects — the outbox carrying a fan-out, and a restart re-arming a
/// stranded row — leave durable state that outlives the process.
///
/// This test's subject does not. A stale claim is a row left `Claimed` with an
/// old claim stamp, and the LIVE reconciler is the thing under test: it must
/// find that row and re-arm it *while its own server serves*. Two facts close
/// every door:
///
/// 1. **The fixture cannot be planted from outside.** `set_outbox_claimed_at`
///    needs a writer handle, and haematite grants exactly one per data
///    directory — held by the server process for its whole life. libSQL allowed
///    the second handle this test was written against; haematite refuses it,
///    and the refusal is correct (a second writer would corrupt the shard WALs).
/// 2. **Planting it BEFORE the boot makes the test vacuous.** Measured: with the
///    stale rows planted in the window between two servers, the test passes with
///    the reconciler's guard mutated to `false`, and passes again with
///    reconciliation left entirely unconfigured. The boot-time re-arm settles
///    those rows before the reconciler ever sweeps, so a "green" would be
///    attributable to the wrong mechanism — and would duplicate
///    `run_server_outbox_restart_rearms_stranded_rows`, which already owns it.
///
/// A stale claim cannot arise naturally on a live server either: with a worker
/// connected the dispatch settles the row `Done`; with none it fails and retries
/// to `Pending`. Only a death between claim and settle leaves `Claimed`, and
/// that death ends the server this test needs alive.
///
/// ## What would make it green
///
/// Either a haematite **concurrent reader/writer seam** — a second handle that
/// can perform ROUTED reads (aion co-locates an outbox row on its workflow's
/// shard, and `ReadOnlyDatabase` routes by the key itself, so today's observer
/// cannot read these rows at all, let alone write one) — or a server-side test
/// seam that can age a claim in place. Both are product decisions, not
/// something this lane may invent; tracked as aion#114 (the citation once read
/// "the aion#35 report" — a report attached to the libsql-retirement umbrella,
/// which closed 2026-08-18 and took the pointer with it).
///
/// Left in its ORIGINAL single-boot form deliberately: that form is the honest
/// statement of the contract. Runtime-gated (house convention) rather than
/// permanently red: a known-red on main would make every future battery's
/// extraction non-empty and cost the "empty = green" reading estate-wide. The
/// full assertion body is kept armed; the skip line names the missing
/// capability and the switch that opens the gate the day it ships.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_outbox_live_reconciliation_rearms_stranded_claims_once() -> Result<(), TestError>
{
    if std::env::var_os(CONCURRENT_SEAM_ENV).is_none() {
        let mut stderr = std::io::stderr();
        let _announced = writeln!(
            stderr,
            "SKIP (PROVES NOTHING): live reconciliation of a stale claim — not \
             established by this run. Needs a haematite concurrent seam with ROUTED \
             reads, or a server-side claim-aging seam (aion#114). No pinnable \
             surface exists today — the absent capability IS the absence — so this \
             announcement stands unpinned until #114 ships a surface. Set \
             {CONCURRENT_SEAM_ENV} once the capability exists."
        );
        return Ok(());
    }
    let dir = unique_temp_dir("live-reconcile")?;
    let db_path = dir.path().join("aion.db");
    let package_path = write_package_archive(dir.path())?;
    let (server, http, grpc) = run_server_harness_with_reconciliation(
        dir.path(),
        &db_path,
        &package_path,
        Some((50, 100)),
    )
    .await?;
    let reader =
        HaematiteStore::open_or_create(db_path.clone(), haematite::NodeCacheBudget::Unlimited)
            .await?;
    let mut worker = WorkerSession::connect(grpc).await?;

    let (workflow_id, run_id) = start_over_http(http).await?;
    let mut tasks = Vec::with_capacity(FAN_OUT);
    for _ in 0..FAN_OUT {
        tasks.push(worker.next_task().await?);
    }
    tasks.sort_by_key(|task| task_ordinal(task).unwrap_or(u64::MAX));
    assert_task_set(&tasks, &[0, 1, 2, 3])?;
    wait_for_rows(
        &reader,
        &workflow_id,
        &[0, 1, 2, 3],
        "initial rows done before simulated stale claim",
        |statuses| statuses.iter().all(|status| *status == OutboxStatus::Done),
    )
    .await?;

    complete_recorded_prefix(&worker, &tasks).await?;
    wait_for_history(
        &reader,
        &workflow_id,
        "ordinals 0 and 1 recorded before live reconciliation",
        |events| count_completed_for(events, 0) == 1 && count_completed_for(events, 1) == 1,
    )
    .await?;

    force_rows_to_stale_claimed(&db_path, &workflow_id, &[2, 3]).await?;

    let revived = collect_revived_tasks(&mut worker).await?;
    complete_with_duplicate_first(&worker, &revived).await?;
    complete_original_late(&worker, &tasks[2..]).await?;

    let history = assert_fan_out_settled(&reader, &workflow_id).await?;
    assert_eq!(count_completed(&history), FAN_OUT);
    assert_eq!(
        count_kind(&history, |event| matches!(
            event,
            Event::WorkflowCompleted { .. }
        )),
        1
    );
    std::hint::black_box(run_id);
    server.stop()?;
    Ok(())
}

async fn complete_recorded_prefix(
    worker: &WorkerSession,
    tasks: &[aion_proto::generated::ActivityTask],
) -> Result<(), TestError> {
    for task in tasks.iter().take(2) {
        let ordinal = task_ordinal(task)?;
        worker
            .complete(task, worker_result(ordinal).as_bytes())
            .await?;
    }
    Ok(())
}

async fn collect_revived_tasks(
    worker: &mut WorkerSession,
) -> Result<Vec<aion_proto::generated::ActivityTask>, TestError> {
    let mut revived = Vec::with_capacity(2);
    for _ in 0..2 {
        revived.push(worker.next_task().await?);
    }
    revived.sort_by_key(|task| task_ordinal(task).unwrap_or(u64::MAX));
    assert_task_set(&revived, &[2, 3])?;
    Ok(revived)
}

async fn complete_with_duplicate_first(
    worker: &WorkerSession,
    revived: &[aion_proto::generated::ActivityTask],
) -> Result<(), TestError> {
    let first = revived
        .first()
        .ok_or_else(|| test_error("missing first revived task"))?;
    let first_ordinal = task_ordinal(first)?;
    worker
        .complete(first, worker_result(first_ordinal).as_bytes())
        .await?;
    worker
        .complete(first, worker_result(first_ordinal).as_bytes())
        .await?;
    let second = revived
        .get(1)
        .ok_or_else(|| test_error("missing second revived task"))?;
    let second_ordinal = task_ordinal(second)?;
    worker
        .complete(second, worker_result(second_ordinal).as_bytes())
        .await?;
    Ok(())
}

async fn complete_original_late(
    worker: &WorkerSession,
    tasks: &[aion_proto::generated::ActivityTask],
) -> Result<(), TestError> {
    for task in tasks {
        let ordinal = task_ordinal(task)?;
        worker
            .complete(task, worker_result(ordinal).as_bytes())
            .await?;
    }
    Ok(())
}

/// Back-date `ordinals`' claim stamps so the LIVE reconciler sees them as stale.
///
/// Opens its own writer on a directory the running server already holds, which
/// is exactly the capability haematite does not grant — see the doc on
/// `run_server_outbox_live_reconciliation_rearms_stranded_claims_once`.
async fn force_rows_to_stale_claimed(
    data_dir: &std::path::Path,
    workflow_id: &aion_core::WorkflowId,
    ordinals: &[u64],
) -> Result<(), TestError> {
    let store = HaematiteStore::open(data_dir, haematite::NodeCacheBudget::Unlimited, |_, _| {})?;
    let claimed_at = Utc::now() - chrono::Duration::seconds(60);
    for ordinal in ordinals {
        let dispatch_key = OutboxRow::dispatch_key_for(workflow_id, *ordinal);
        store
            .set_outbox_claimed_at(&dispatch_key, claimed_at)
            .await?;
    }
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_child_process() -> Result<(), TestError> {
    if std::env::var_os("AION_RUN_SERVER_CHILD").is_none() {
        return Ok(());
    }
    let config_path = std::env::var_os("AION_RUN_SERVER_CONFIG")
        .map(PathBuf::from)
        .ok_or_else(|| test_error("AION_RUN_SERVER_CONFIG is required"))?;
    let code = aion_server::run::run(CliOverrides {
        config_path: Some(config_path),
        ..CliOverrides::default()
    })
    .await;
    if code == ExitCode::SUCCESS {
        Ok(())
    } else {
        Err(test_error(format!("run_server exited with {code:?}")))
    }
}