remora/crates/remora-coordination/tests/substrate.rs

use std::sync::Arc;
use std::time::{Duration, Instant};

use remora_coordination::{
    ActivateOptions, CoordinationError, CoordinationStore, LeaseOwner, LocalPermitGate,
    MAX_SAFE_INTEGER, OpCode, StoreConfig, ensure_safe_integer,
};
use tokio::sync::Barrier;
use tokio::time::sleep;

fn coordination_url() -> Option<String> {
    std::env::var("REMORA_COORDINATION_URL")
        .ok()
        .filter(|value| !value.is_empty())
}

fn coordination_namespace(suite: &str) -> String {
    let base = std::env::var("REMORA_COORDINATION_NAMESPACE")
        .ok()
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| "remora:test:".into());
    format!("{base}coordination-substrate:{suite}:")
}

async fn test_store(
    suite: &str,
    global_limit: u64,
    lease_ttl: Duration,
) -> Option<CoordinationStore> {
    let Some(url) = coordination_url() else {
        if std::env::var("REMORA_COORDINATION_REQUIRE").is_ok() {
            panic!("REMORA_COORDINATION_URL is required when REMORA_COORDINATION_REQUIRE is set");
        }
        eprintln!(
            "skipping coordination substrate integration test: REMORA_COORDINATION_URL unset"
        );
        return None;
    };
    let store = CoordinationStore::connect(
        &url,
        coordination_namespace(suite),
        StoreConfig {
            global_limit,
            lease_ttl,
            reserve_ttl: lease_ttl,
            progress_ttl: lease_ttl,
            cancel_ttl: None,
        },
    )
    .await
    .expect("connect to coordination store");
    Some(store)
}

fn lease_owner(attempt: &str, fence: u64, worker: &str) -> LeaseOwner {
    LeaseOwner::new(attempt, fence, worker).expect("safe fence")
}

#[tokio::test]
async fn concurrent_activate_at_most_one_live_lease() {
    let Some(store) = test_store("r1-activate", 16, Duration::from_secs(10)).await else {
        return;
    };
    let job = "job-activate-aaaa";
    store.drop_job_state(&[job]).await.expect("clean");
    store.enqueue(job).await.expect("enqueue");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");

    let barrier = Arc::new(Barrier::new(2));
    let store_a = store.clone();
    let store_b = store.clone();
    let b1 = barrier.clone();
    let b2 = barrier.clone();
    let a = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    let b = lease_owner("attempt-bbbbbbbb", 2, "worker-bbbbbbbb");

    let t1 = tokio::spawn(async move {
        b1.wait().await;
        store_a
            .activate(job, &a, ActivateOptions::default())
            .await
            .expect("activate a")
    });
    let t2 = tokio::spawn(async move {
        b2.wait().await;
        store_b
            .activate(
                job,
                &b,
                ActivateOptions {
                    seed_progress: true,
                    require_reserve: false,
                },
            )
            .await
            .expect("activate b")
    });
    let r1 = t1.await.expect("join a");
    let r2 = t2.await.expect("join b");
    let wins = [r1.code, r2.code]
        .into_iter()
        .filter(|code| matches!(code, OpCode::Ok | OpCode::OkIdempotent))
        .count();
    let holds = [r1.code, r2.code]
        .into_iter()
        .filter(|code| *code == OpCode::FailHeld || *code == OpCode::FailReserve)
        .count();
    assert_eq!(wins, 1, "exactly one activate succeeds: {r1:?} {r2:?}");
    assert_eq!(holds, 1, "peer fails held/reserve: {r1:?} {r2:?}");

    let live = store.get_live_lease(job).await.expect("get live");
    assert!(live.is_some());
    assert_eq!(store.global_used().await.expect("used"), 1);

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn reserve_contention_respects_global_limit() {
    let Some(store) = test_store("g1-limit", 1, Duration::from_secs(10)).await else {
        return;
    };
    let job_a = "job-limit-aaaaaaa";
    let job_b = "job-limit-bbbbbbb";
    store.drop_job_state(&[job_a, job_b]).await.expect("clean");

    let barrier = Arc::new(Barrier::new(2));
    let s1 = store.clone();
    let s2 = store.clone();
    let b1 = barrier.clone();
    let b2 = barrier.clone();
    let t1 = tokio::spawn(async move {
        b1.wait().await;
        s1.reserve(job_a, "worker-aaaaaaaa")
            .await
            .expect("reserve a")
    });
    let t2 = tokio::spawn(async move {
        b2.wait().await;
        s2.reserve(job_b, "worker-bbbbbbbb")
            .await
            .expect("reserve b")
    });
    let r1 = t1.await.expect("join");
    let r2 = t2.await.expect("join");
    let oks = [r1.code, r2.code]
        .into_iter()
        .filter(|c| matches!(c, OpCode::Ok | OpCode::OkIdempotent))
        .count();
    let fulls = [r1.code, r2.code]
        .into_iter()
        .filter(|c| *c == OpCode::FailFull)
        .count();
    assert_eq!(oks, 1, "{r1:?} {r2:?}");
    assert_eq!(fulls, 1, "{r1:?} {r2:?}");
    assert!(store.global_used().await.expect("used") <= 1);

    store
        .drop_job_state(&[job_a, job_b])
        .await
        .expect("cleanup");
}

#[tokio::test]
async fn permit_contention_benchmark_preserves_global_bound() {
    if std::env::var("REMORA_QUALIFICATION_REQUIRE").is_err() {
        return;
    }
    const TASKS: usize = 16;
    const OPERATIONS_PER_TASK: usize = 16;
    const GLOBAL_LIMIT: u64 = 8;
    let Some(store) = test_store(
        "qualification-permits",
        GLOBAL_LIMIT,
        Duration::from_secs(30),
    )
    .await
    else {
        return;
    };
    let jobs = (0..TASKS * OPERATIONS_PER_TASK)
        .map(|index| format!("job-qualification-{index:04}"))
        .collect::<Vec<_>>();
    let job_refs = jobs.iter().map(String::as_str).collect::<Vec<_>>();
    store.drop_job_state(&job_refs).await.expect("clean");

    let barrier = Arc::new(Barrier::new(TASKS));
    let started = Instant::now();
    let mut tasks = Vec::with_capacity(TASKS);
    for task_index in 0..TASKS {
        let store = store.clone();
        let barrier = Arc::clone(&barrier);
        let worker = format!("worker-qualification-{task_index:02}");
        let task_jobs =
            jobs[task_index * OPERATIONS_PER_TASK..(task_index + 1) * OPERATIONS_PER_TASK].to_vec();
        tasks.push(tokio::spawn(async move {
            let mut durations = Vec::with_capacity(OPERATIONS_PER_TASK);
            barrier.wait().await;
            for job in task_jobs {
                let operation_started = Instant::now();
                loop {
                    let outcome = store.reserve(&job, &worker).await.expect("reserve");
                    if outcome.ok() {
                        break;
                    }
                    assert_eq!(outcome.code, OpCode::FailFull);
                    tokio::task::yield_now().await;
                }
                assert!(store.global_used().await.expect("used") <= GLOBAL_LIMIT);
                assert!(
                    store
                        .free_reserve(&job, Some(&worker))
                        .await
                        .expect("free")
                        .ok()
                );
                durations.push(operation_started.elapsed());
            }
            durations
        }));
    }
    let mut durations = Vec::with_capacity(TASKS * OPERATIONS_PER_TASK);
    for task in tasks {
        durations.extend(task.await.expect("join"));
    }
    durations.sort_unstable();
    let elapsed = started.elapsed();
    let p50 = durations[durations.len() / 2];
    let p99 = durations[durations.len() * 99 / 100];
    let operations_per_second = durations.len() as f64 / elapsed.as_secs_f64();
    eprintln!(
        "permit benchmark operations={} elapsed_ms={} p50_us={} p99_us={} operations_per_second={operations_per_second:.1}",
        durations.len(),
        elapsed.as_millis(),
        p50.as_micros(),
        p99.as_micros()
    );
    assert!(elapsed < Duration::from_secs(30));
    assert_eq!(store.global_used().await.expect("used"), 0);
    store.drop_job_state(&job_refs).await.expect("cleanup");
}

#[tokio::test]
async fn release_and_expire_preserve_reserved_only_global_slot() {
    let Some(store) = test_store("permit-reserved", 1, Duration::from_secs(10)).await else {
        return;
    };
    let job_a = "job-rsvd-aaaaaaaa";
    let job_b = "job-rsvd-bbbbbbbb";
    store.drop_job_state(&[job_a, job_b]).await.expect("clean");

    let reserved = store
        .reserve(job_a, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    assert_eq!(reserved.code, OpCode::Ok);
    assert_eq!(store.global_used().await.expect("used"), 1);

    // No live lease: release/expire must not free reserved occupancy.
    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    let release = store.release(job_a, &owner).await.expect("release");
    assert_eq!(release.code, OpCode::OkGone);
    assert_eq!(store.global_used().await.expect("used"), 1);
    assert!(store.get_reservation(job_a).await.expect("res").is_some());

    let expire = store.expire_lease(job_a).await.expect("expire");
    assert_eq!(expire.code, OpCode::OkGone);
    assert_eq!(store.global_used().await.expect("used"), 1);

    let peer = store.reserve(job_b, "worker-bbbbbbbb").await.expect("peer");
    assert_eq!(peer.code, OpCode::FailFull, "{peer:?}");

    let free = store
        .free_reserve(job_a, Some("worker-aaaaaaaa"))
        .await
        .expect("free reserve");
    assert_eq!(free.code, OpCode::Ok);
    assert_eq!(store.global_used().await.expect("used"), 0);
    let peer_ok = store
        .reserve(job_b, "worker-bbbbbbbb")
        .await
        .expect("peer after free");
    assert_eq!(peer_ok.code, OpCode::Ok);

    store
        .drop_job_state(&[job_a, job_b])
        .await
        .expect("cleanup");
}

#[tokio::test]
async fn free_reserve_does_not_drop_slot_under_live_lease() {
    let Some(store) = test_store("free-under-lease", 4, Duration::from_secs(10)).await else {
        return;
    };
    let job = "job-freelease-aaa";
    store.drop_job_state(&[job]).await.expect("clean");
    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store
        .activate(job, &owner, ActivateOptions::default())
        .await
        .expect("activate");
    assert_eq!(store.global_used().await.expect("used"), 1);
    // Activate already cleared the reservation. Freeing it must not drop the
    // global unit while the live lease still holds the job.
    let free = store.free_reserve(job, None).await.expect("free");
    assert_eq!(free.code, OpCode::OkGone);
    assert_eq!(store.global_used().await.expect("used"), 1);
    assert!(store.get_live_lease(job).await.expect("lease").is_some());
    store.release(job, &owner).await.expect("release");
    assert_eq!(store.global_used().await.expect("used"), 0);
    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn reserve_activate_is_net_zero_global_occupancy() {
    let Some(store) = test_store("g2-netzero", 4, Duration::from_secs(10)).await else {
        return;
    };
    let job = "job-netzero-aaaaa";
    store.drop_job_state(&[job]).await.expect("clean");
    let res = store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    assert_eq!(res.code, OpCode::Ok);
    assert_eq!(store.global_used().await.expect("used"), 1);

    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    let act = store
        .activate(job, &owner, ActivateOptions::default())
        .await
        .expect("activate");
    assert!(act.ok(), "{act:?}");
    assert_eq!(store.global_used().await.expect("used"), 1);
    assert!(store.get_reservation(job).await.expect("res").is_none());
    assert!(store.get_live_lease(job).await.expect("lease").is_some());

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn expire_frees_global_slot_for_peer() {
    let Some(store) = test_store("g3-expire", 1, Duration::from_millis(80)).await else {
        return;
    };
    let job_a = "job-expire-aaaaaa";
    let job_b = "job-expire-bbbbbb";
    store.drop_job_state(&[job_a, job_b]).await.expect("clean");

    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    store
        .reserve(job_a, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store
        .activate(job_a, &owner, ActivateOptions::default())
        .await
        .expect("activate");
    assert_eq!(store.global_used().await.expect("used"), 1);

    sleep(Duration::from_millis(120)).await;
    let exp = store.expire_lease(job_a).await.expect("expire");
    assert!(exp.ok(), "{exp:?}");
    assert_eq!(store.global_used().await.expect("used"), 0);

    let peer = store
        .reserve(job_b, "worker-bbbbbbbb")
        .await
        .expect("peer reserve");
    assert_eq!(peer.code, OpCode::Ok);

    store
        .drop_job_state(&[job_a, job_b])
        .await
        .expect("cleanup");
}

#[tokio::test]
async fn double_release_and_idempotent_activate() {
    let Some(store) = test_store("r6-idemp", 4, Duration::from_secs(10)).await else {
        return;
    };
    let job = "job-idemp-aaaaaaa";
    store.drop_job_state(&[job]).await.expect("clean");
    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    let first = store
        .activate(job, &owner, ActivateOptions::default())
        .await
        .expect("activate");
    assert_eq!(first.code, OpCode::Ok);
    let second = store
        .activate(job, &owner, ActivateOptions::default())
        .await
        .expect("activate again");
    assert_eq!(second.code, OpCode::OkIdempotent);
    assert_eq!(store.global_used().await.expect("used"), 1);

    let rel1 = store.release(job, &owner).await.expect("release");
    assert_eq!(rel1.code, OpCode::Ok);
    let rel2 = store.release(job, &owner).await.expect("release again");
    assert_eq!(rel2.code, OpCode::OkGone);
    assert_eq!(store.global_used().await.expect("used"), 0);

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn stale_owner_renew_fails() {
    let Some(store) = test_store("r5-stolen", 4, Duration::from_millis(80)).await else {
        return;
    };
    let job = "job-stolen-aaaaaa";
    store.drop_job_state(&[job]).await.expect("clean");
    let old = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    let new = lease_owner("attempt-bbbbbbbb", 2, "worker-bbbbbbbb");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store
        .activate(job, &old, ActivateOptions::default())
        .await
        .expect("activate");
    sleep(Duration::from_millis(120)).await;
    let exp = store.expire_lease(job).await.expect("expire");
    assert!(exp.ok());
    store
        .activate(job, &new, ActivateOptions::default())
        .await
        .expect("peer activate");
    let renew = store.renew(job, &old).await.expect("renew");
    assert!(
        matches!(
            renew.code,
            OpCode::FailStolen | OpCode::FailMissing | OpCode::FailExpired
        ),
        "{renew:?}"
    );
    let live = store
        .get_live_lease(job)
        .await
        .expect("live")
        .expect("present");
    assert_eq!(live.record.attempt_id, "attempt-bbbbbbbb");

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn renew_after_expiry_does_not_resurrect() {
    let Some(store) = test_store("r10-expired", 4, Duration::from_millis(60)).await else {
        return;
    };
    let job = "job-expired-aaaaa";
    store.drop_job_state(&[job]).await.expect("clean");
    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store
        .activate(job, &owner, ActivateOptions::default())
        .await
        .expect("activate");
    sleep(Duration::from_millis(100)).await;
    let renew = store.renew(job, &owner).await.expect("renew");
    // Dual TTL: key may already be gone (PEXPIRE) or still present past
    // expires_at. Neither path may resurrect a live lease.
    assert!(
        matches!(renew.code, OpCode::FailExpired | OpCode::FailMissing),
        "{renew:?}"
    );
    assert!(store.get_live_lease(job).await.expect("live").is_none());

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn progress_fenced_and_stale_clear() {
    let Some(store) = test_store("prog-hint", 4, Duration::from_millis(80)).await else {
        return;
    };
    let job = "job-progress-aaaa";
    store.drop_job_state(&[job]).await.expect("clean");
    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store
        .activate(
            job,
            &owner,
            ActivateOptions {
                seed_progress: true,
                require_reserve: true,
            },
        )
        .await
        .expect("activate");
    let seeded = store
        .get_progress(job)
        .await
        .expect("progress")
        .expect("seed");
    assert_eq!(seeded.attempt_id, "attempt-aaaaaaaa");
    assert_eq!(seeded.bytes, 0);

    let report = store
        .report_progress(job, &owner, 4096)
        .await
        .expect("report");
    assert_eq!(report.code, OpCode::Ok);
    // Idempotent activate must refresh the lease without zeroing progress.
    let again = store
        .activate(
            job,
            &owner,
            ActivateOptions {
                seed_progress: true,
                require_reserve: false,
            },
        )
        .await
        .expect("activate again");
    assert_eq!(again.code, OpCode::OkIdempotent);
    let after_retry = store
        .get_progress(job)
        .await
        .expect("progress")
        .expect("kept");
    assert_eq!(after_retry.bytes, 4096);

    let prog = store
        .get_progress(job)
        .await
        .expect("progress")
        .expect("hint");
    assert_eq!(prog.bytes, 4096);

    let foreign = lease_owner("attempt-bbbbbbbb", 2, "worker-bbbbbbbb");
    let stolen = store
        .report_progress(job, &foreign, 99)
        .await
        .expect("foreign");
    assert_eq!(stolen.code, OpCode::FailStolen);

    store.release(job, &owner).await.expect("release");
    let cleared = store.clear_stale_progress(job).await.expect("clear");
    assert!(cleared.ok(), "{cleared:?}");
    assert!(store.get_progress(job).await.expect("progress").is_none());

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn activate_without_seed_clears_foreign_progress() {
    let Some(store) = test_store("prog-compat", 4, Duration::from_millis(80)).await else {
        return;
    };
    let job = "job-progcompat-aa";
    store.drop_job_state(&[job]).await.expect("clean");
    let first = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    let second = lease_owner("attempt-bbbbbbbb", 2, "worker-bbbbbbbb");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store
        .activate(job, &first, ActivateOptions::default())
        .await
        .expect("activate");
    store
        .report_progress(job, &first, 100)
        .await
        .expect("report");
    store.release(job, &first).await.expect("release");
    // Progress may outlive its lease until the next owner activates.
    assert!(store.get_progress(job).await.expect("progress").is_some());

    store
        .activate(
            job,
            &second,
            ActivateOptions {
                seed_progress: false,
                require_reserve: false,
            },
        )
        .await
        .expect("peer activate");
    assert!(
        store.get_progress(job).await.expect("progress").is_none(),
        "foreign progress must not remain under a new live lease"
    );

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn cancel_notice_and_queue_projection_idempotent() {
    let Some(store) = test_store("proj-queue", 4, Duration::from_secs(10)).await else {
        return;
    };
    let job = "job-project-aaaa";
    store.drop_job_state(&[job]).await.expect("clean");

    let e1 = store.project_enqueue(job).await.expect("enqueue");
    assert_eq!(e1.code, OpCode::Ok);
    let e2 = store.project_enqueue(job).await.expect("enqueue again");
    assert_eq!(e2.code, OpCode::OkIdempotent);

    let c1 = store.project_cancel(job).await.expect("cancel");
    assert_eq!(c1.code, OpCode::Ok);
    let c2 = store.project_cancel(job).await.expect("cancel again");
    assert_eq!(c2.code, OpCode::OkIdempotent);
    assert!(store.cancel_notice(job).await.expect("notice"));

    let claimed = store
        .claim_and_reserve("worker-aaaaaaaa")
        .await
        .expect("claim");
    assert_eq!(claimed.as_deref(), Some(job));
    assert!(store.get_reservation(job).await.expect("res").is_some());

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn clean_terminal_and_reconstruction() {
    let Some(store) = test_store("clean-term", 4, Duration::from_secs(10)).await else {
        return;
    };
    let job = "job-terminal-aaa";
    store.drop_job_state(&[job]).await.expect("clean");
    store.project_enqueue(job).await.expect("enqueue");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store.project_cancel(job).await.expect("cancel");
    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    store
        .activate(job, &owner, ActivateOptions::default())
        .await
        .expect("activate");
    store.release(job, &owner).await.expect("release");

    // Reconstruct: wipe job keys then re-project from durable facts.
    store.drop_job_state(&[job]).await.expect("wipe");
    assert_eq!(store.global_used().await.expect("used"), 0);
    assert!(store.get_live_lease(job).await.expect("lease").is_none());

    // Terminal reconstruction only needs clean + optional cancel/enqueue for non-terminals.
    let re_q = store.project_enqueue(job).await.expect("re-enqueue");
    assert_eq!(re_q.code, OpCode::Ok);
    let clean = store.clean_terminal(job).await.expect("clean");
    assert!(clean.ok(), "{clean:?}");
    assert!(!store.cancel_notice(job).await.expect("notice"));
    assert!(store.get_reservation(job).await.expect("res").is_none());
    assert_eq!(store.dequeue().await.expect("deq"), None);

    store.drop_job_state(&[job]).await.expect("cleanup");
}

/// Re-projection and admission rebuild slot membership after coordination loss.
/// PermitSafety holds across repeated wipes and concurrent admission.
#[tokio::test]
async fn repeated_loss_reconstruction_preserves_permit_safety() {
    let Some(store) = test_store("rebuild-g", 2, Duration::from_secs(10)).await else {
        return;
    };
    let jobs = ["job-rebuild-aaaa", "job-rebuild-bbbb", "job-rebuild-cccc"];
    store.drop_job_state(&jobs).await.expect("clean");

    for pass in 0..3 {
        // Durable facts re-projected after each coordination loss.
        for job in jobs {
            let enq = store.project_enqueue(job).await.expect("enqueue");
            assert!(
                matches!(enq.code, OpCode::Ok | OpCode::OkIdempotent),
                "pass {pass}: {enq:?}"
            );
        }

        let barrier = Arc::new(Barrier::new(3));
        let mut handles = Vec::new();
        for (i, job) in jobs.iter().enumerate() {
            let store = store.clone();
            let barrier = barrier.clone();
            let job = (*job).to_string();
            let worker = format!("worker-{:08}", i);
            handles.push(tokio::spawn(async move {
                barrier.wait().await;
                store.reserve(&job, &worker).await.expect("reserve")
            }));
        }
        let mut results = Vec::new();
        for handle in handles {
            results.push(handle.await.expect("join"));
        }
        let oks = results
            .iter()
            .filter(|r| matches!(r.code, OpCode::Ok | OpCode::OkIdempotent))
            .count();
        let fulls = results
            .iter()
            .filter(|r| r.code == OpCode::FailFull)
            .count();
        assert_eq!(oks, 2, "pass {pass}: {results:?}");
        assert_eq!(fulls, 1, "pass {pass}: {results:?}");
        let used = store.global_used().await.expect("used");
        assert!(used <= 2, "pass {pass}: used={used}");
        assert_eq!(used, 2, "pass {pass}: used={used}");

        // Repeated coordination loss (double wipe) clears occupancy.
        store.drop_job_state(&jobs).await.expect("wipe1");
        assert_eq!(store.global_used().await.expect("used"), 0);
        store.drop_job_state(&jobs).await.expect("wipe2");
        assert_eq!(store.global_used().await.expect("used"), 0);
        // Queue membership is gone; re-enqueue next pass is OK not IDEMPOTENT.
    }

    store.drop_job_state(&jobs).await.expect("cleanup");
}

#[tokio::test]
async fn safe_integer_fence_and_progress_bounds() {
    assert!(ensure_safe_integer(MAX_SAFE_INTEGER).is_ok());
    assert!(matches!(
        ensure_safe_integer(MAX_SAFE_INTEGER + 1),
        Err(CoordinationError::UnsafeInteger { .. })
    ));
    assert!(LeaseOwner::new("attempt-aaaaaaaa", MAX_SAFE_INTEGER, "worker-aaaaaaaa").is_ok());
    assert!(matches!(
        LeaseOwner::new("attempt-aaaaaaaa", MAX_SAFE_INTEGER + 1, "worker-aaaaaaaa"),
        Err(CoordinationError::UnsafeInteger { .. })
    ));

    let Some(store) = test_store("safe-int", 4, Duration::from_secs(10)).await else {
        return;
    };
    let job = "job-safeint-aaaa";
    store.drop_job_state(&[job]).await.expect("clean");
    let owner = LeaseOwner::new("attempt-aaaaaaaa", MAX_SAFE_INTEGER, "worker-aaaaaaaa")
        .expect("max fence");
    store
        .reserve(job, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    let act = store
        .activate(job, &owner, ActivateOptions::default())
        .await
        .expect("activate max fence");
    assert!(act.ok(), "{act:?}");
    let report = store
        .report_progress(job, &owner, MAX_SAFE_INTEGER)
        .await
        .expect("progress max");
    assert_eq!(report.code, OpCode::Ok);

    // Construct an out-of-range owner via struct literal to hit store validation.
    let bad = LeaseOwner {
        attempt_id: "attempt-bbbbbbbb".into(),
        fence: MAX_SAFE_INTEGER + 1,
        worker_id: "worker-bbbbbbbb".into(),
    };
    let err = store
        .renew(job, &bad)
        .await
        .expect_err("unsafe fence rejected");
    assert!(matches!(err, CoordinationError::UnsafeInteger { .. }));
    let err = store
        .report_progress(job, &owner, MAX_SAFE_INTEGER + 1)
        .await
        .expect_err("unsafe bytes rejected");
    assert!(matches!(err, CoordinationError::UnsafeInteger { .. }));

    store.drop_job_state(&[job]).await.expect("cleanup");
}

#[tokio::test]
async fn local_permit_holds_through_lease_expiry_until_stop() {
    let gate = LocalPermitGate::new(1);
    assert!(gate.try_acquire("job-local-aaaaaaa").expect("acquire"));
    assert!(!gate.try_acquire("job-local-bbbbbbb").expect("second"));
    assert_eq!(gate.used(), 1);
    // Simulated lease expiry does not free local capacity.
    assert!(gate.holds("job-local-aaaaaaa"));
    assert!(!gate.try_acquire("job-local-bbbbbbb").expect("still full"));
    // Cooperative stop frees capacity.
    assert!(gate.release_all("job-local-aaaaaaa"));
    assert!(gate.try_acquire("job-local-bbbbbbb").expect("after stop"));
    assert_eq!(gate.used(), 1);
}

#[tokio::test]
async fn zombie_global_free_local_hold() {
    let Some(store) = test_store("zombie-g", 1, Duration::from_millis(60)).await else {
        return;
    };
    let job_a = "job-zombie-aaaaaa";
    let job_b = "job-zombie-bbbbbb";
    store.drop_job_state(&[job_a, job_b]).await.expect("clean");
    let local = LocalPermitGate::new(1);
    local.acquire(job_a).expect("local");

    let owner = lease_owner("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    store
        .reserve(job_a, "worker-aaaaaaaa")
        .await
        .expect("reserve");
    store
        .activate(job_a, &owner, ActivateOptions::default())
        .await
        .expect("activate");
    sleep(Duration::from_millis(100)).await;
    store.expire_lease(job_a).await.expect("expire");
    // Global free after expire: peer job may reserve.
    let peer = store.reserve(job_b, "worker-bbbbbbbb").await.expect("peer");
    assert_eq!(peer.code, OpCode::Ok);
    // Local still held by zombie task.
    assert!(local.holds(job_a));
    assert!(!local.try_acquire(job_b).expect("local full"));
    local.release_all(job_a);
    assert!(local.try_acquire(job_b).expect("after stop"));

    store
        .drop_job_state(&[job_a, job_b])
        .await
        .expect("cleanup");
}

#[tokio::test]
async fn keyspace_prefix_isolation() {
    let Some(url) = coordination_url() else {
        if std::env::var("REMORA_COORDINATION_REQUIRE").is_ok() {
            panic!("REMORA_COORDINATION_URL is required when REMORA_COORDINATION_REQUIRE is set");
        }
        return;
    };
    let a = CoordinationStore::connect(
        &url,
        coordination_namespace("iso-a"),
        StoreConfig {
            global_limit: 1,
            ..StoreConfig::default()
        },
    )
    .await
    .expect("connect a");
    let b = CoordinationStore::connect(
        &url,
        coordination_namespace("iso-b"),
        StoreConfig {
            global_limit: 1,
            ..StoreConfig::default()
        },
    )
    .await
    .expect("connect b");
    let job = "job-iso-aaaaaaaa";
    a.drop_job_state(&[job]).await.expect("clean a");
    b.drop_job_state(&[job]).await.expect("clean b");
    a.reserve(job, "worker-aaaaaaaa").await.expect("reserve a");
    b.reserve(job, "worker-bbbbbbbb").await.expect("reserve b");
    assert_eq!(a.global_used().await.expect("a used"), 1);
    assert_eq!(b.global_used().await.expect("b used"), 1);
    a.drop_job_state(&[job]).await.expect("clean a");
    b.drop_job_state(&[job]).await.expect("clean b");
}