armdb 0.14.1

//! Change-stream queue transport crash and post-log replay coverage.
#![cfg(all(
    feature = "test-faults",
    feature = "armour",
    feature = "typed-tree",
    feature = "rapira-codec",
    feature = "var-collections"
))]

use std::fs::{self, OpenOptions};
use std::future::Future;
use std::io::Write;
use std::process::Command;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Condvar, Mutex};
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use std::time::{Duration, UNIX_EPOCH};

use armdb::armour::{Db, WorkerIntervals};
use armdb::txn::failpoints::{self, ApplyFailGuard, LogSyncFailGuard};
use armdb::{
    CollectionMeta, Config, DbError, Events, IndexTargets, NoHook, Projected, Queue, QueueConfig,
    QueueKey, QueueTargets, RapiraCodec, TtlHook, TxFailStage, TypedTree, queue,
};
use armour_core::GetType;
use rapira::Rapira;

#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct Source(u64);

impl CollectionMeta for Source {
    type SelfId = [u8; 8];
    const NAME: &'static str = "queue_crash_stream_source";
}

#[derive(Clone, Debug, PartialEq, Eq, GetType, Rapira)]
pub struct Job {
    body: String,
}

queue!(Jobs, payload = Job, name = "queue_crash_stream_jobs", var);

type StreamHook = Projected<[u8; 8], Source, NoHook, Events<[u8; 8], Source>>;
type SourceTree = TypedTree<[u8; 8], Source, RapiraCodec, TtlHook<Source, StreamHook>>;

fn key(id: u64) -> [u8; 8] {
    id.to_be_bytes()
}

fn open_source(
    db: &Arc<Db>,
    jobs: &Arc<Queue<Jobs>>,
    callback_log: &std::path::Path,
) -> Arc<SourceTree> {
    let queue = Arc::clone(jobs);
    let callback_log = callback_log.to_owned();
    db.open_typed_tree::<Source, RapiraCodec, _>(
        Config::test(),
        Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
            QueueTargets::new().with(jobs),
            move |_, _, new, sink| {
                let mut log = OpenOptions::new()
                    .create(true)
                    .append(true)
                    .open(&callback_log)
                    .map_err(|_| DbError::Internal("stream callback log"))?;
                log.write_all(b"called\n")
                    .map_err(|_| DbError::Internal("stream callback log"))?;
                log.sync_all()
                    .map_err(|_| DbError::Internal("stream callback log"))?;
                if let Some(new) = new {
                    sink.emit(
                        &queue,
                        Job {
                            body: format!("job-{}", new.0),
                        },
                    )?;
                }
                Ok(())
            },
        ),
        &[],
    )
    .unwrap()
}

fn open_fixture(
    dir: &std::path::Path,
    callback_log: &std::path::Path,
) -> (Arc<Db>, Arc<SourceTree>, Arc<Queue<Jobs>>) {
    let db = Arc::new(Db::open(dir).unwrap());
    let jobs = db
        .open_queue::<Jobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let source = open_source(&db, &jobs, callback_log);
    (db, source, jobs)
}

fn assert_converged(
    source: &Arc<SourceTree>,
    jobs: &Arc<Queue<Jobs>>,
    callback_log: &std::path::Path,
) {
    assert_eq!(source.get(&key(1)).map(|row| *row), Some(Source(7)));
    assert_eq!(jobs.len(), 1, "one header must survive/replay");
    assert_eq!(
        jobs.debug_payload_len(),
        1,
        "one payload must survive/replay"
    );
    let leased = jobs.lease(2, None).unwrap();
    assert_eq!(leased.len(), 1, "replay must not duplicate the queue row");
    assert_eq!(
        leased[0].payload(),
        &Job {
            body: "job-7".into()
        }
    );
    assert_eq!(
        fs::read(callback_log).unwrap(),
        b"called\n",
        "replay must not invoke the stream closure"
    );
}

fn run_child(dir: &std::path::Path, callback_log: &std::path::Path) -> std::process::ExitStatus {
    Command::new(std::env::current_exe().unwrap())
        .arg("--exact")
        .arg("queue_crash_child_after_log_append")
        .arg("--nocapture")
        .env("ARMDB_QUEUE_CRASH_CHILD", "1")
        .env("ARMDB_QUEUE_CRASH_DIR", dir)
        .env("ARMDB_QUEUE_CRASH_CALLBACK_LOG", callback_log)
        .status()
        .unwrap()
}

fn run_apply_fail_child(
    dir: &std::path::Path,
    callback_log: &std::path::Path,
) -> std::process::ExitStatus {
    Command::new(std::env::current_exe().unwrap())
        .arg("--exact")
        .arg("queue_crash_child_apply_failure")
        .arg("--nocapture")
        .env("ARMDB_QUEUE_CRASH_APPLY_CHILD", "1")
        .env("ARMDB_QUEUE_CRASH_DIR", dir)
        .env("ARMDB_QUEUE_CRASH_CALLBACK_LOG", callback_log)
        .status()
        .unwrap()
}

fn run_queue_child(action: &str, dir: &std::path::Path) -> std::process::ExitStatus {
    Command::new(std::env::current_exe().unwrap())
        .arg("--exact")
        .arg("queue_crash_queue_child")
        .arg("--nocapture")
        .env("ARMDB_QUEUE_CRASH_QUEUE_CHILD", action)
        .env("ARMDB_QUEUE_CRASH_DIR", dir)
        .status()
        .unwrap()
}

/// Lease deadline for the crash tests that assert an outstanding lease is
/// *still held* after a child exit plus a parent open. That parent open runs
/// replay and, for a Var queue, a full orphan pass; on a loaded box a
/// one-second budget loses that race before the test can make its assertion.
const CRASH_LEASE_TIMEOUT: Duration = Duration::from_secs(5);

fn crash_queue_config(action: &str) -> QueueConfig {
    let config = QueueConfig::balanced().backoff_secs(&[0]);
    match action {
        "lease" => config.lease_timeout(CRASH_LEASE_TIMEOUT),
        "nack-cross-shard-final" | "nack-cross-shard-final-mid-apply" => config
            .max_attempts(1)
            .lease_timeout(Duration::from_millis(1)),
        "seed-counts" => config.max_attempts(1).lease_timeout(CRASH_LEASE_TIMEOUT),
        "retry" => config
            .max_attempts(1)
            .dead_retention(Some(Duration::from_millis(1))),
        _ => config,
    }
    .build()
    .unwrap()
}

fn open_crash_queue(dir: &std::path::Path, action: &str) -> (Arc<Db>, Arc<Queue<Jobs>>) {
    let db = Arc::new(Db::open_test(dir).unwrap());
    let jobs = db.open_queue::<Jobs>(crash_queue_config(action)).unwrap();
    (db, jobs)
}

fn prepare_cross_shard_final_nack(jobs: &Queue<Jobs>) -> armdb::LeaseToken {
    const VISIBLE_AT: Duration = Duration::from_millis(1);
    for _ in 0..64 {
        let id = jobs
            .push(
                Job {
                    body: "nack-cross-shard-final".into(),
                },
                Some(UNIX_EPOCH + VISIBLE_AT),
            )
            .unwrap();
        let source = QueueKey::live(VISIBLE_AT.as_millis() as u64, id);
        let destination = QueueKey::dead(source.visible_at_ms(), source.id());
        let token = jobs.lease(1, None).unwrap().pop().unwrap().token();
        if jobs.debug_header_shard(&source) != jobs.debug_header_shard(&destination) {
            return token;
        }
        jobs.ack(token).unwrap();
    }
    panic!("must find a cross-shard final nack source key");
}

#[test]
fn queue_crash_child_after_log_append() {
    if std::env::var("ARMDB_QUEUE_CRASH_CHILD").is_err() {
        return;
    }
    let dir = std::env::var("ARMDB_QUEUE_CRASH_DIR").unwrap();
    let callback_log = std::env::var("ARMDB_QUEUE_CRASH_CALLBACK_LOG").unwrap();
    failpoints::STAGE.store(
        failpoints::AFTER_LOG_APPEND,
        std::sync::atomic::Ordering::SeqCst,
    );
    let (db, source, _) = open_fixture(
        std::path::Path::new(&dir),
        std::path::Path::new(&callback_log),
    );
    let _ = db
        .tx()
        .put(&source, &key(1), &Source(7))
        .unwrap()
        .durable()
        .commit();
    unreachable!("AFTER_LOG_APPEND must hard-exit the child");
}

#[test]
fn queue_crash_child_apply_failure() {
    if std::env::var("ARMDB_QUEUE_CRASH_APPLY_CHILD").is_err() {
        return;
    }
    let dir = std::env::var("ARMDB_QUEUE_CRASH_DIR").unwrap();
    let callback_log = std::env::var("ARMDB_QUEUE_CRASH_CALLBACK_LOG").unwrap();
    let (db, source, _jobs) = open_fixture(
        std::path::Path::new(&dir),
        std::path::Path::new(&callback_log),
    );
    let failure = ApplyFailGuard::at(0);
    let result = db.tx().put(&source, &key(1), &Source(7)).unwrap().commit();
    assert!(matches!(result, Err(DbError::TxFailedAfterLog { .. })));
    drop(result);
    drop(failure);
}

/// Child-only entry point for queue operations that deliberately leave the
/// process without running database destructors or a graceful checkpoint.
#[test]
fn queue_crash_queue_child() {
    let Ok(action) = std::env::var("ARMDB_QUEUE_CRASH_QUEUE_CHILD") else {
        return;
    };
    let dir = std::env::var("ARMDB_QUEUE_CRASH_DIR").unwrap();
    let (db, jobs) = open_crash_queue(std::path::Path::new(&dir), &action);

    match action.as_str() {
        "payload-before-header" => {
            let (written, _release) = jobs.debug_pause_after_payload();
            let pushing = Arc::clone(&jobs);
            std::thread::spawn(move || {
                pushing.push(
                    Job {
                        body: "orphan".into(),
                    },
                    None,
                )
            });
            written.wait();
        }
        "ack" => {
            let token = jobs.lease(1, None).unwrap().pop().unwrap().token();
            failpoints::STAGE.store(
                failpoints::AFTER_LOG_APPEND,
                std::sync::atomic::Ordering::SeqCst,
            );
            let _ = db.tx().ack(&jobs, token).unwrap().durable().commit();
            unreachable!("AFTER_LOG_APPEND must hard-exit the child");
        }
        "lease" => {
            assert_eq!(jobs.lease(1, None).unwrap().len(), 1);
        }
        "retry" => {
            jobs.push(
                Job {
                    body: "retry".into(),
                },
                None,
            )
            .unwrap();
            let token = jobs.lease(1, None).unwrap().pop().unwrap().token();
            jobs.nack(token).unwrap();
            // The first failure is from the dead-header TTL hook; the second
            // keeps the just-enqueued retry from being drained in this sweep.
            jobs.debug_fail_payload_deletes(2);
            std::thread::sleep(Duration::from_millis(5));
            assert!(db.sweep_expired_now().is_err());
            assert_eq!(jobs.debug_payload_len(), 1);
        }
        "nack" => {
            let token = jobs.lease(1, None).unwrap().pop().unwrap().token();
            failpoints::STAGE.store(
                failpoints::AFTER_LOG_APPEND,
                std::sync::atomic::Ordering::SeqCst,
            );
            let _ = db.tx().nack(&jobs, token).unwrap().durable().commit();
            unreachable!("AFTER_LOG_APPEND must hard-exit the child");
        }
        "nack-cross-shard-final" | "nack-cross-shard-final-mid-apply" => {
            // The parent holds epoch 1 as the stale token. Re-lease after its
            // 1 ms deadline so the child holds the actual token it nacks.
            std::thread::sleep(Duration::from_millis(5));
            let token = jobs.lease(1, None).unwrap().pop().unwrap().token();
            let source = token.key();
            let destination = QueueKey::dead(source.visible_at_ms(), source.id());
            assert_ne!(
                jobs.debug_header_shard(&source),
                jobs.debug_header_shard(&destination),
                "the final nack must exercise two FixedStore header shards"
            );
            let stage = if action == "nack-cross-shard-final-mid-apply" {
                failpoints::MID_APPLY
            } else {
                failpoints::AFTER_LOG_APPEND
            };
            failpoints::STAGE.store(stage, std::sync::atomic::Ordering::SeqCst);
            let _ = db.tx().nack(&jobs, token).unwrap().durable().commit();
            unreachable!("selected failpoint must hard-exit the child");
        }
        "seed-counts" => {
            let leased_at = UNIX_EPOCH + Duration::from_millis(1);
            db.tx()
                .push(
                    &jobs,
                    Job {
                        body: "leased".into(),
                    },
                    Some(leased_at),
                )
                .unwrap()
                .durable()
                .commit()
                .unwrap();
            assert_eq!(jobs.lease(1, None).unwrap().len(), 1);

            db.tx()
                .push(
                    &jobs,
                    Job {
                        body: "dead".into(),
                    },
                    Some(UNIX_EPOCH + Duration::from_millis(2)),
                )
                .unwrap()
                .durable()
                .commit()
                .unwrap();
            let dead = jobs.lease(1, None).unwrap().pop().unwrap().token();
            db.tx()
                .nack(&jobs, dead)
                .unwrap()
                .durable()
                .commit()
                .unwrap();

            db.tx()
                .push(
                    &jobs,
                    Job {
                        body: "ready".into(),
                    },
                    Some(UNIX_EPOCH + Duration::from_millis(3)),
                )
                .unwrap()
                .durable()
                .commit()
                .unwrap();
        }
        _ => panic!("unknown queue crash action: {action}"),
    }
    unsafe { libc::_exit(9) }
}

#[test]
fn hard_exit_after_log_append_replays_source_and_both_var_queue_rows_once() {
    let dir = tempfile::tempdir().unwrap();
    let callback_log = dir.path().join("stream-callback");
    assert_eq!(run_child(dir.path(), &callback_log).code(), Some(9));

    let (_db, source, jobs) = open_fixture(dir.path(), &callback_log);
    assert_converged(&source, &jobs, &callback_log);
}

#[test]
fn post_log_apply_failure_reopens_to_one_source_header_and_payload_without_reinvoking_stream() {
    let dir = tempfile::tempdir().unwrap();
    let callback_log = dir.path().join("stream-callback");
    assert_eq!(
        run_apply_fail_child(dir.path(), &callback_log).code(),
        Some(0)
    );

    let (_db, source, jobs) = open_fixture(dir.path(), &callback_log);
    assert_converged(&source, &jobs, &callback_log);
}

#[test]
fn post_log_var_apply_failure_does_not_pin_ttl_shutdown_and_recovers_after_reopen() {
    let dir = tempfile::tempdir().unwrap();
    let mut db = Db::builder(dir.path())
        .for_tests()
        .workers(WorkerIntervals {
            compaction: None,
            flush: None,
            ttl_sweep: Some(Duration::from_millis(1)),
            ttl_budget: 4096,
        })
        .open()
        .unwrap();
    let jobs = db
        .open_queue::<Jobs>(
            QueueConfig::balanced()
                .max_attempts(1)
                .dead_retention(Some(Duration::ZERO))
                .build()
                .unwrap(),
        )
        .unwrap();

    let maintenance_started = jobs.debug_notify_maintenance_wait();
    let tx = db
        .tx()
        .push(
            &jobs,
            Job {
                body: "replay".into(),
            },
            None,
        )
        .unwrap();
    jobs.debug_mark_full_orphan_pass_pending();
    let failure = ApplyFailGuard::at(0);
    let error = tx.commit().unwrap_err();
    drop(failure);
    assert!(matches!(error, DbError::TxFailedAfterLog { .. }));

    maintenance_started
        .recv_timeout(Duration::from_secs(1))
        .expect("the TTL sweeper must observe the pinned producer");
    std::thread::scope(|scope| {
        let (done, complete) = mpsc::channel();
        let db = &mut db;
        scope.spawn(move || done.send(db.shutdown()).unwrap());
        complete
            .recv_timeout(Duration::from_secs(1))
            .expect("shutdown must not wait for a post-log pinned producer")
            .unwrap();
    });
    drop(jobs);
    drop(db);

    let db = Db::open_test(dir.path()).unwrap();
    let jobs = db
        .open_queue::<Jobs>(
            QueueConfig::balanced()
                .max_attempts(1)
                .dead_retention(Some(Duration::ZERO))
                .build()
                .unwrap(),
        )
        .unwrap();
    let leased = jobs.lease(1, None).unwrap();
    assert_eq!(leased.len(), 1, "recovery must replay the failed enqueue");
    assert_eq!(leased[0].payload().body, "replay");
}

#[test]
fn tx_var_push_does_not_deadlock_its_own_stream_preflight_when_cleanup_is_pending() {
    let dir = tempfile::tempdir().unwrap();
    let callback_log = dir.path().join("stream-callback");
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let jobs = db
        .open_queue::<Jobs>(
            QueueConfig::balanced()
                .max_attempts(1)
                .dead_retention(Some(Duration::ZERO))
                .build()
                .unwrap(),
        )
        .unwrap();
    let source = open_source(&db, &jobs, &callback_log);

    jobs.push(
        Job {
            body: "stale".into(),
        },
        None,
    )
    .unwrap();
    let stale = jobs.lease(1, None).unwrap().pop().unwrap().token();
    jobs.debug_fail_payload_deletes(usize::MAX);
    jobs.nack(stale).unwrap();
    assert!(db.sweep_expired_now().is_err());

    let (done, complete) = mpsc::channel();
    let db_for_tx = Arc::clone(&db);
    let jobs_for_tx = Arc::clone(&jobs);
    let source_for_tx = Arc::clone(&source);
    std::thread::spawn(move || {
        let result = db_for_tx
            .tx()
            .push(
                &jobs_for_tx,
                Job {
                    body: "direct".into(),
                },
                None,
            )
            .and_then(|tx| tx.put(&source_for_tx, &key(1), &Source(7)))
            .and_then(|tx| tx.commit());
        done.send(result).unwrap();
    });
    complete
        .recv_timeout(Duration::from_secs(1))
        .expect("stream preflight must defer maintenance held behind its own producer permit")
        .unwrap();

    jobs.debug_fail_payload_deletes(0);
    let leased = jobs.lease(3, None).unwrap();
    assert_eq!(
        leased.len(),
        2,
        "both direct and stream payloads must survive"
    );
    assert!(leased.iter().any(|entry| entry.payload().body == "direct"));
    assert!(leased.iter().any(|entry| entry.payload().body == "job-7"));
}

/// A direct Var push writes the payload first. A crash at that boundary must
/// never make its body visible as a header, and full orphan maintenance must
/// be able to reclaim the unreferenced payload after reopen.
#[test]
fn hard_exit_after_direct_var_payload_write_never_publishes_a_header_and_orphan_pass_reclaims_it() {
    let dir = tempfile::tempdir().unwrap();
    assert_eq!(
        run_queue_child("payload-before-header", dir.path()).code(),
        Some(9)
    );

    let (_db, jobs) = open_crash_queue(dir.path(), "payload-before-header");
    assert_eq!(jobs.len(), 0, "an unwritten header must not become visible");
    assert_eq!(jobs.lease(1, None).unwrap().len(), 0);
    assert_eq!(
        jobs.debug_payload_len(),
        0,
        "open must collect the unreferenced payload before exposing the queue"
    );
    jobs.debug_force_full_orphan_pass().unwrap();
    assert_eq!(
        jobs.debug_payload_len(),
        0,
        "orphan payload must be collected"
    );
}

#[test]
fn hard_exit_after_log_append_of_ack_replays_the_removal_of_header_and_payload() {
    let dir = tempfile::tempdir().unwrap();
    {
        let (_db, jobs) = open_crash_queue(dir.path(), "ack");
        jobs.push(Job { body: "ack".into() }, None).unwrap();
    }
    assert_eq!(run_queue_child("ack", dir.path()).code(), Some(9));

    let (_db, jobs) = open_crash_queue(dir.path(), "ack");
    assert_eq!(jobs.len(), 0);
    assert_eq!(jobs.debug_payload_len(), 0);
    assert!(jobs.lease(1, None).unwrap().is_empty());
}

#[test]
fn hard_exit_with_an_outstanding_lease_keeps_it_unavailable_until_its_deadline() {
    let dir = tempfile::tempdir().unwrap();
    {
        let (_db, jobs) = open_crash_queue(dir.path(), "lease");
        jobs.push(
            Job {
                body: "lease".into(),
            },
            None,
        )
        .unwrap();
    }
    assert_eq!(run_queue_child("lease", dir.path()).code(), Some(9));

    let (_db, jobs) = open_crash_queue(dir.path(), "lease");
    assert!(jobs.lease(1, None).unwrap().is_empty());
    std::thread::sleep(CRASH_LEASE_TIMEOUT + Duration::from_millis(200));
    assert_eq!(jobs.lease(1, None).unwrap().len(), 1);
}

#[test]
fn hard_exit_reopen_seeds_ready_leased_and_dead_queue_counts() {
    let dir = tempfile::tempdir().unwrap();
    assert_eq!(run_queue_child("seed-counts", dir.path()).code(), Some(9));

    let (_db, jobs) = open_crash_queue(dir.path(), "seed-counts");
    assert_eq!(
        jobs.len(),
        2,
        "ready and leased headers must seed live count"
    );
    assert_eq!(jobs.dead_len(), 1, "one final nack must seed dead count");
    assert_eq!(
        jobs.debug_payload_len(),
        3,
        "all three rows retain one payload"
    );

    let ready = jobs.lease(8, None).unwrap();
    assert_eq!(
        ready.len(),
        1,
        "the outstanding lease must stay unavailable"
    );
    assert_eq!(ready[0].payload().body, "ready");
    jobs.ack(ready[0].token()).unwrap();
    assert!(jobs.lease(1, None).unwrap().is_empty());

    std::thread::sleep(CRASH_LEASE_TIMEOUT + Duration::from_millis(200));
    let formerly_leased = jobs.lease(1, None).unwrap();
    assert_eq!(formerly_leased.len(), 1);
    assert_eq!(formerly_leased[0].payload().body, "leased");
    jobs.ack(formerly_leased[0].token()).unwrap();
    assert_eq!(jobs.len(), 0);
    assert_eq!(jobs.dead_len(), 1);
    assert_eq!(
        jobs.debug_payload_len(),
        1,
        "only the retained dead payload remains"
    );
}

#[test]
fn process_death_with_a_pending_var_payload_delete_is_recovered_by_a_full_orphan_pass() {
    let dir = tempfile::tempdir().unwrap();
    assert_eq!(run_queue_child("retry", dir.path()).code(), Some(9));

    let (_db, jobs) = open_crash_queue(dir.path(), "retry");
    assert_eq!(jobs.len(), 0);
    assert_eq!(jobs.dead_len(), 0);
    assert_eq!(
        jobs.debug_payload_len(),
        0,
        "open must recover a lost in-memory retry through its full orphan pass"
    );
    jobs.debug_force_full_orphan_pass().unwrap();
    assert_eq!(jobs.debug_payload_len(), 0);
}

/// `nack` rewrites one header as delete plus insert. This process-level
/// after-log crash proves replay leaves exactly one retry row and its payload,
/// instead of losing or duplicating the job.
#[test]
fn hard_exit_after_log_append_of_nack_replays_exactly_one_var_header_and_payload() {
    let dir = tempfile::tempdir().unwrap();
    {
        let (_db, jobs) = open_crash_queue(dir.path(), "nack");
        jobs.push(
            Job {
                body: "nack".into(),
            },
            None,
        )
        .unwrap();
    }
    assert_eq!(run_queue_child("nack", dir.path()).code(), Some(9));

    let (_db, jobs) = open_crash_queue(dir.path(), "nack");
    assert_eq!(jobs.len(), 1);
    assert_eq!(jobs.dead_len(), 0);
    assert_eq!(jobs.debug_payload_len(), 1);
    let leased = jobs.lease(1, None).unwrap();
    assert_eq!(leased.len(), 1);
    assert_eq!(
        leased[0].payload(),
        &Job {
            body: "nack".into()
        }
    );
}

/// A final nack rekeys the header from live to dead. The selector deliberately
/// chooses an id whose real source and dead destination route to different
/// FixedStore shards before the child exits immediately after the log append.
#[test]
fn cross_shard_final_nack_after_log_append_replays_one_dead_header_and_payload() {
    let dir = tempfile::tempdir().unwrap();
    let stale = {
        let (_db, jobs) = open_crash_queue(dir.path(), "nack-cross-shard-final");
        prepare_cross_shard_final_nack(&jobs)
    };
    assert_eq!(
        run_queue_child("nack-cross-shard-final", dir.path()).code(),
        Some(9)
    );

    let (_db, jobs) = open_crash_queue(dir.path(), "nack-cross-shard-final");
    assert_eq!(jobs.len(), 0, "the old live source must not survive replay");
    assert_eq!(
        jobs.dead_len(),
        1,
        "exactly one dead destination must remain"
    );
    assert_eq!(
        jobs.debug_payload_len(),
        1,
        "the retained payload must remain once"
    );
    assert!(jobs.lease(1, None).unwrap().is_empty());
    assert!(matches!(jobs.nack(stale), Err(DbError::LeaseExpired)));
}

#[test]
fn cross_shard_final_nack_after_first_apply_replays_one_dead_header_and_payload() {
    let dir = tempfile::tempdir().unwrap();
    let stale = {
        let (_db, jobs) = open_crash_queue(dir.path(), "nack-cross-shard-final");
        prepare_cross_shard_final_nack(&jobs)
    };
    assert_eq!(
        run_queue_child("nack-cross-shard-final-mid-apply", dir.path()).code(),
        Some(9)
    );

    let (_db, jobs) = open_crash_queue(dir.path(), "nack-cross-shard-final");
    assert_eq!(jobs.len(), 0, "the applied source delete must stay deleted");
    assert_eq!(
        jobs.dead_len(),
        1,
        "replay must restore only the destination"
    );
    assert_eq!(jobs.debug_payload_len(), 1);
    assert!(matches!(jobs.nack(stale), Err(DbError::LeaseExpired)));
}

#[test]
fn sync_failed_var_push_publishes_waiters_and_releases_the_producer_permit() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let jobs = db
        .open_queue::<Jobs>(
            QueueConfig::balanced()
                .wakeup_recheck(Duration::from_secs(5))
                .build()
                .unwrap(),
        )
        .unwrap();
    let counter = Arc::new(WakeCounter::default());
    let mut due = Box::pin(jobs.due());
    let waker = counting_waker(Arc::clone(&counter));
    let mut cx = Context::from_waker(&waker);
    assert!(matches!(due.as_mut().poll(&mut cx), Poll::Pending));

    let fail = LogSyncFailGuard::arm();
    let err = db
        .tx()
        .push(
            &jobs,
            Job {
                body: "sync".into(),
            },
            None,
        )
        .unwrap()
        .durable()
        .commit()
        .unwrap_err();
    drop(fail);
    assert!(matches!(
        err,
        DbError::TxFailedAfterLog {
            stage: TxFailStage::Sync,
            ..
        }
    ));
    assert_eq!(jobs.len(), 1);
    counter.wait_for(1);

    let maintenance_jobs = Arc::clone(&jobs);
    let (done, ready) = mpsc::channel();
    let maintenance = std::thread::spawn(move || {
        done.send(maintenance_jobs.debug_force_full_orphan_pass())
            .unwrap();
    });
    assert!(
        ready.recv_timeout(Duration::from_secs(1)).unwrap().is_ok(),
        "successful apply must release the Var producer permit"
    );
    maintenance.join().unwrap();
}

#[derive(Default)]
struct WakeCounter {
    wakes: AtomicUsize,
    observed: Mutex<usize>,
    changed: Condvar,
}

impl WakeCounter {
    fn wake(&self) {
        let wakes = self.wakes.fetch_add(1, Ordering::AcqRel) + 1;
        *self.observed.lock().unwrap() = wakes;
        self.changed.notify_all();
    }

    fn wait_for(&self, expected: usize) {
        let mut observed = self.observed.lock().unwrap();
        while *observed < expected {
            let (next, result) = self
                .changed
                .wait_timeout(observed, Duration::from_secs(1))
                .unwrap();
            assert!(!result.timed_out(), "waker was not called");
            observed = next;
        }
    }
}

fn counting_waker(counter: Arc<WakeCounter>) -> Waker {
    unsafe {
        Waker::from_raw(RawWaker::new(
            Arc::into_raw(counter).cast(),
            &COUNTING_WAKER_VTABLE,
        ))
    }
}

unsafe fn counting_clone(raw: *const ()) -> RawWaker {
    unsafe { Arc::increment_strong_count(raw.cast::<WakeCounter>()) };
    RawWaker::new(raw, &COUNTING_WAKER_VTABLE)
}

unsafe fn counting_wake(raw: *const ()) {
    let counter = unsafe { Arc::from_raw(raw.cast::<WakeCounter>()) };
    counter.wake();
}

unsafe fn counting_wake_by_ref(raw: *const ()) {
    let counter = unsafe { Arc::from_raw(raw.cast::<WakeCounter>()) };
    counter.wake();
    let _ = Arc::into_raw(counter);
}

unsafe fn counting_drop(raw: *const ()) {
    drop(unsafe { Arc::from_raw(raw.cast::<WakeCounter>()) });
}

static COUNTING_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
    counting_clone,
    counting_wake,
    counting_wake_by_ref,
    counting_drop,
);