armdb 0.14.3

use std::mem::size_of;
use std::process::Command;
use std::sync::{Arc, Barrier};
use std::thread;
use std::time::{Duration, Instant, UNIX_EPOCH};

use armdb::armour::Db;
use armdb::{DEAD_CLASS, DbError, Key, QueueConfig, QueueKey, QueueSpec, queue};
use armour_core::GetType;

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    GetType,
    zerocopy::FromBytes,
    zerocopy::IntoBytes,
    zerocopy::Immutable,
    zerocopy::KnownLayout,
)]
#[repr(transparent)]
#[allow(dead_code)]
pub struct FixedJob(u64);

queue!(FixedJobs, payload = FixedJob, name = "queue_fixed_jobs");
queue!(
    FixedJobsSchemaDrift,
    payload = FixedJob,
    name = "queue_fixed_schema_drift"
);

#[derive(
    Clone,
    Copy,
    GetType,
    zerocopy::FromBytes,
    zerocopy::IntoBytes,
    zerocopy::Immutable,
    zerocopy::KnownLayout,
)]
#[repr(transparent)]
pub struct ChangedFixedJob(u64);

queue!(
    FixedJobsSchemaDriftChanged,
    payload = ChangedFixedJob,
    name = "queue_fixed_schema_drift"
);

#[test]
fn queue_macro_sets_the_exact_physical_names_and_key_order() {
    assert_eq!(FixedJobs::NAME, "queue_fixed_jobs");
    assert_eq!(FixedJobs::PAYLOAD_NAME, "queue_fixed_jobs.payload");
    let a = QueueKey::live(7, 3);
    let b = QueueKey::live(8, 1);
    assert!(a.as_bytes() < b.as_bytes());
    assert_eq!(
        a.as_bytes(),
        &[0, 0, 0, 0, 0, 0, 0, 7, 0, 0, 0, 0, 0, 0, 0, 3]
    );
    assert_eq!(size_of::<QueueKey>(), 16);
    assert_eq!(a.as_bytes().len(), 16);

    let dead = QueueKey::dead(7, 3);
    assert_eq!(dead.class(), DEAD_CLASS);
    assert_eq!(dead.visible_at_ms(), 7);
    assert_eq!(dead.id(), 3);
    assert_eq!(QueueKey::from_bytes(dead.as_bytes()), dead);
    assert!(std::panic::catch_unwind(|| QueueKey::live(1 << 63, 3)).is_err());
}

#[test]
fn queue_config_rejects_an_unusable_retry_policy() {
    assert!(QueueConfig::balanced().max_attempts(0).build().is_err());
    assert!(QueueConfig::balanced().backoff_secs(&[]).build().is_err());
    assert!(matches!(DbError::LeaseExpired, DbError::LeaseExpired));
}

#[test]
fn open_queue_rejects_unbuilt_invalid_fixed_policies_before_nack_can_panic() {
    let assert_invalid = |config| {
        let dir = tempfile::tempdir().unwrap();
        let db = armdb::armour::Db::open_test(dir.path()).unwrap();
        assert!(matches!(
            db.open_queue::<FixedJobs>(config),
            Err(DbError::Config(_))
        ));
    };

    assert_invalid(QueueConfig::balanced().max_attempts(0));
    assert_invalid(QueueConfig::balanced().backoff_secs(&[]));
    assert_invalid(QueueConfig::balanced().lease_timeout(Duration::ZERO));
    assert_invalid(QueueConfig::balanced().wakeup_recheck(Duration::ZERO));
}

#[test]
fn push_is_durable_and_open_seeds_live_count() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let first = q.push(FixedJob(11), None).unwrap();
    let second = q.push(FixedJob(12), None).unwrap();
    assert_eq!(q.len(), 2);
    drop(q);
    drop(db);

    let reopened = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = reopened
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    assert_eq!(q.len(), 2);
    assert!(q.push(FixedJob(13), None).unwrap() > second);
    assert!(second > first);
}

#[test]
fn fixed_payload_schema_drift_is_rejected_on_reopen() {
    let dir = tempfile::tempdir().unwrap();
    Db::open_test(dir.path())
        .unwrap()
        .open_queue::<FixedJobsSchemaDrift>(QueueConfig::balanced().build().unwrap())
        .unwrap();

    let result = Db::open_test(dir.path())
        .unwrap()
        .open_queue::<FixedJobsSchemaDriftChanged>(QueueConfig::balanced().build().unwrap());
    assert!(matches!(result, Err(DbError::SchemaMismatch { .. })));
}

#[test]
fn queue_header_config_is_ascending_and_has_small_queue_baseline() {
    let cfg = QueueConfig::balanced().build().unwrap().fixed_config();
    assert!(!cfg.reversed);
    assert_eq!(cfg.grow_step, 100_000);
}

#[test]
fn depth_cap_refuses_without_creating_a_row() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().max_depth(Some(1)).build().unwrap())
        .unwrap();
    q.push(FixedJob(1), None).unwrap();
    assert!(matches!(q.push(FixedJob(2), None), Err(DbError::QueueFull)));
    assert_eq!(q.len(), 1);
}

#[test]
fn concurrent_pushes_do_not_exceed_the_documented_soft_depth_bound() {
    const MAX_DEPTH: u64 = 1;
    const PUSHERS: usize = 8;

    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let queue = db
        .open_queue::<FixedJobs>(
            QueueConfig::balanced()
                .max_depth(Some(MAX_DEPTH))
                .build()
                .unwrap(),
        )
        .unwrap();
    let barrier = Arc::new(Barrier::new(PUSHERS));
    armdb::queue_test_support::set_fixed_queue_push_admission_barrier(Some(Arc::clone(&barrier)));
    let pushers: Vec<_> = (0..PUSHERS)
        .map(|id| {
            let queue = Arc::clone(&queue);
            let barrier = Arc::clone(&barrier);
            thread::spawn(move || {
                barrier.wait();
                queue.push(FixedJob(id as u64 + 1), None)
            })
        })
        .collect();
    let results: Vec<_> = pushers
        .into_iter()
        .map(|pusher| pusher.join().unwrap())
        .collect();
    armdb::queue_test_support::set_fixed_queue_push_admission_barrier(None);

    assert_eq!(
        results.iter().filter(|result| result.is_ok()).count(),
        PUSHERS,
        "every producer passed the same below-cap admission check"
    );
    assert!(
        queue.len() <= MAX_DEPTH + (PUSHERS as u64 - 1),
        "concurrent pushers exceeded the documented soft-cap race bound"
    );
    assert_eq!(queue.len(), PUSHERS as u64);
}

#[test]
fn fixed_queue_leases_due_rows_by_visible_at_across_retry_and_expiry() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let queue = db
        .open_queue::<FixedJobs>(
            QueueConfig::balanced()
                .lease_timeout(Duration::from_millis(1))
                .max_attempts(2)
                .backoff_secs(&[0])
                .build()
                .unwrap(),
        )
        .unwrap();
    let first_visible = UNIX_EPOCH + Duration::from_millis(100);
    let second_visible = UNIX_EPOCH + Duration::from_millis(200);
    let third_visible = UNIX_EPOCH + Duration::from_millis(300);
    let future = std::time::SystemTime::now() + Duration::from_secs(60);
    queue.push(FixedJob(10), Some(first_visible)).unwrap();
    queue.push(FixedJob(20), Some(second_visible)).unwrap();
    queue.push(FixedJob(30), Some(third_visible)).unwrap();
    queue.push(FixedJob(40), Some(future)).unwrap();

    let first = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(*first.payload(), FixedJob(10));
    assert_eq!(first.key().visible_at_ms(), 100);
    queue.nack(first.token()).unwrap();

    let second = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(*second.payload(), FixedJob(20));
    assert_eq!(second.key().visible_at_ms(), 200);
    queue.ack(second.token()).unwrap();

    let expired = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(*expired.payload(), FixedJob(30));
    assert_eq!(expired.key().visible_at_ms(), 300);
    thread::sleep(Duration::from_millis(3));

    let released = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(*released.payload(), FixedJob(30));
    assert_eq!(released.key().visible_at_ms(), 300);
    queue.ack(released.token()).unwrap();

    let retried = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(*retried.payload(), FixedJob(10));
    assert!(
        retried.key().visible_at_ms()
            > third_visible
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64
    );
    queue.ack(retried.token()).unwrap();

    assert!(queue.lease(1, None).unwrap().is_empty());
    assert_eq!(queue.len(), 1, "the future row stays unleased");
}

#[test]
fn child_crash_after_first_push_never_reuses_a_published_id_block() {
    let dir = tempfile::tempdir().unwrap();
    let status = Command::new(std::env::current_exe().unwrap())
        .arg("--exact")
        .arg("queue_fixed_crash_child_after_first_push")
        .arg("--nocapture")
        .env("ARMDB_QUEUE_CRASH_DIR", dir.path())
        .status()
        .unwrap();
    assert_eq!(status.code(), Some(9));

    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    assert!(
        q.push(FixedJob(2), None).unwrap() >= armdb::ID_BLOCK,
        "a hard crash must not permit reuse of the first published id block"
    );
}

#[test]
fn queue_fixed_crash_child_after_first_push() {
    let Ok(dir) = std::env::var("ARMDB_QUEUE_CRASH_DIR") else {
        return;
    };
    let db = armdb::armour::Db::open_test(dir).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    q.push(FixedJob(1), None).unwrap();
    unsafe { libc::_exit(9) }
}

#[test]
fn push_rejects_not_before_outside_the_queue_timestamp_domain() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let out_of_range = UNIX_EPOCH + Duration::from_millis(1 << 63);

    assert!(matches!(
        q.push(FixedJob(1), Some(out_of_range)),
        Err(DbError::Client(_))
    ));
    assert_eq!(q.len(), 0);
}

#[test]
fn an_expired_lease_does_not_consume_the_retry_budget() {
    // `attempts` counts *failed deliveries*, and only `nack` reports one. A
    // lease that simply expires is a delivery that never completed, not a
    // failure, so it must leave the budget untouched — with `max_attempts(2)`
    // the job below survives an expiry plus one nack and is still retried.
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let queue = db
        .open_queue::<FixedJobs>(
            QueueConfig::balanced()
                .lease_timeout(Duration::from_millis(1))
                .max_attempts(2)
                .backoff_secs(&[0])
                .build()
                .unwrap(),
        )
        .unwrap();
    queue.push(FixedJob(1), None).unwrap();

    let first = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(
        first.attempts(),
        0,
        "a first delivery has no failures behind it"
    );
    thread::sleep(Duration::from_millis(3)); // the lease expires, unclaimed

    let after_expiry = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(
        after_expiry.attempts(),
        0,
        "an expired lease must not be charged to the retry budget"
    );
    queue.nack(after_expiry.token()).unwrap();
    assert_eq!(queue.dead_len(), 0, "one nack is not the final attempt");

    let retried = queue.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(retried.attempts(), 1, "exactly one failed delivery so far");
    queue.nack(retried.token()).unwrap();
    assert_eq!(queue.dead_len(), 1, "the second nack is terminal");
    assert_eq!(queue.len(), 0);
}

#[test]
fn an_expired_lease_nobody_re_took_still_acks() {
    // Nothing re-leased the row, so its epoch never advanced and the original
    // token still matches. The work was done; the ack must land.
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let queue = db
        .open_queue::<FixedJobs>(
            QueueConfig::balanced()
                .lease_timeout(Duration::from_millis(1))
                .build()
                .unwrap(),
        )
        .unwrap();
    queue.push(FixedJob(1), None).unwrap();
    let leased = queue.lease(1, None).unwrap().pop().unwrap();
    thread::sleep(Duration::from_millis(5));

    queue.ack(leased.token()).unwrap();
    assert_eq!(queue.len(), 0);
    assert_eq!(queue.dead_len(), 0);
}

#[test]
fn a_re_leased_expiry_fences_the_original_token() {
    // The counterpart of the test above: once someone else re-leases the row,
    // the epoch advances and the first holder's token is fenced.
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let queue = db
        .open_queue::<FixedJobs>(
            QueueConfig::balanced()
                .lease_timeout(Duration::from_millis(1))
                .build()
                .unwrap(),
        )
        .unwrap();
    queue.push(FixedJob(1), None).unwrap();
    let first = queue.lease(1, None).unwrap().pop().unwrap();
    thread::sleep(Duration::from_millis(5));
    let second = queue.lease(1, None).unwrap().pop().unwrap();

    assert!(matches!(
        queue.ack(first.token()),
        Err(DbError::LeaseExpired)
    ));
    queue.ack(second.token()).unwrap();
    assert_eq!(queue.len(), 0);
}

/// Lease used by the deadline races: short enough that every round stays
/// cheap, long enough that a renewer released right after the lease can still
/// reach the row before it expires.
const RACE_LEASE: Duration = Duration::from_millis(5);
/// Far longer than a round, so a renewal that wins keeps the row out of the
/// claimer's reach for the rest of that round.
const RACE_RENEWAL: Duration = Duration::from_secs(30);
const RACE_ROUNDS: u64 = 120;

/// When round `round` releases its racers, measured from the end of the lease
/// call: swept in 0.5 ms steps from immediately (a renewal can still win) to
/// 2 ms past the latest possible deadline (a renewal cannot), so about a third
/// of the rounds land on the deadline itself.
fn race_release_offset(round: u64) -> Duration {
    Duration::from_micros((round % 15) * 500)
}

fn sleep_until(target: Instant) {
    let now = Instant::now();
    if target > now {
        thread::sleep(target - now);
    }
}

/// `true` for success, `false` for a lost lease; any other error fails the test.
fn owned_or_expired(round: u64, op: &str, result: Result<(), DbError>) -> bool {
    match result {
        Ok(()) => true,
        Err(DbError::LeaseExpired) => false,
        Err(error) => panic!("round {round}: {op} failed with {error:?}"),
    }
}

#[test]
fn queue_renew_extends_and_keeps_token_payload_and_depth() {
    // Only the first renew below has to land inside the original lease.
    const ORIGINAL_LEASE: Duration = Duration::from_millis(500);
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(
            QueueConfig::balanced()
                .lease_timeout(Duration::from_secs(30))
                // Longer than the lease, so `next_delay` reports the deadline itself.
                .wakeup_recheck(Duration::from_secs(60))
                .build()
                .unwrap(),
        )
        .unwrap();
    let expected = FixedJob(91);
    q.push(expected, None).unwrap();
    let entry = q.lease(1, Some(ORIGINAL_LEASE)).unwrap().pop().unwrap();
    let leased_at = Instant::now();
    let token = entry.token();
    q.renew(token, None).unwrap();
    assert!(q.next_delay().unwrap().unwrap() > Duration::from_secs(20));
    q.renew(token, Some(Duration::from_millis(1))).unwrap();
    assert!(q.next_delay().unwrap().unwrap() > Duration::from_secs(20));
    sleep_until(leased_at + ORIGINAL_LEASE + Duration::from_millis(100));
    assert!(q.lease(1, None).unwrap().is_empty());
    assert!(q.next_delay().unwrap().unwrap() > Duration::from_secs(20));
    assert!(!q.wait_due_timeout(Duration::from_millis(10)).unwrap());
    assert_eq!(entry.attempts(), 0);
    assert_eq!(entry.payload(), &expected);
    assert_eq!(q.len(), 1);
    assert_eq!(q.dead_len(), 0);
    assert_eq!(q.stats().payload_writes, 0);
    q.ack(token).unwrap();
    assert_eq!(q.len(), 0);
    assert!(matches!(q.renew(token, None), Err(DbError::LeaseExpired)));
}

#[test]
fn queue_renew_rejects_expired_reclaimed_and_nacked_tokens() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().backoff_secs(&[0]).build().unwrap())
        .unwrap();
    q.push(FixedJob(91), None).unwrap();
    let old = q
        .lease(1, Some(Duration::from_millis(20)))
        .unwrap()
        .pop()
        .unwrap();
    std::thread::sleep(Duration::from_millis(40));
    assert!(matches!(
        q.renew(old.token(), None),
        Err(DbError::LeaseExpired)
    ));
    let fresh = q.lease(1, None).unwrap().pop().unwrap();
    assert_ne!(old.token().epoch(), fresh.token().epoch());
    assert_eq!(fresh.attempts(), 0);
    assert!(matches!(
        q.renew(old.token(), None),
        Err(DbError::LeaseExpired)
    ));
    assert!(matches!(q.ack(old.token()), Err(DbError::LeaseExpired)));
    q.renew(fresh.token(), None).unwrap();
    q.nack(fresh.token()).unwrap();
    assert!(matches!(
        q.renew(fresh.token(), None),
        Err(DbError::LeaseExpired)
    ));
    let retry = q.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(retry.attempts(), 1);
    assert_eq!(retry.payload(), &FixedJob(91));
    q.ack(retry.token()).unwrap();
}

#[test]
fn queue_renew_rejects_invalid_durations_without_losing_ownership() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    q.push(FixedJob(91), None).unwrap();
    let entry = q.lease(1, None).unwrap().pop().unwrap();
    for duration in [
        Duration::ZERO,
        Duration::from_nanos(1),
        Duration::MAX,
        Duration::from_millis(u64::MAX),
    ] {
        assert!(matches!(
            q.renew(entry.token(), Some(duration)),
            Err(DbError::Client(_))
        ));
    }
    q.renew(entry.token(), None).unwrap();
    q.ack(entry.token()).unwrap();
}

#[test]
fn queue_renew_reopen_preserves_deadline_but_invalidates_instance_token() {
    // Only the renew has to land inside the original lease; the reopen has to
    // finish inside the renewal.
    const ORIGINAL_LEASE: Duration = Duration::from_millis(500);
    const RENEWAL: Duration = Duration::from_millis(1500);
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    q.push(FixedJob(91), None).unwrap();
    let old = q
        .lease(1, Some(ORIGINAL_LEASE))
        .unwrap()
        .pop()
        .unwrap()
        .token();
    let leased_at = Instant::now();
    q.renew(old, Some(RENEWAL)).unwrap();
    let renewed_at = Instant::now();
    drop(q);
    drop(db);
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    // Past the original deadline, inside the renewed one: the renewal is durable.
    sleep_until(leased_at + ORIGINAL_LEASE + Duration::from_millis(100));
    assert!(q.lease(1, None).unwrap().is_empty());
    assert!(matches!(q.renew(old, None), Err(DbError::LeaseExpired)));
    sleep_until(renewed_at + RENEWAL + Duration::from_millis(100));
    let recovered = q.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(recovered.attempts(), 0);
    assert_eq!(recovered.payload(), &FixedJob(91));
    assert!(matches!(q.ack(old), Err(DbError::LeaseExpired)));
    q.ack(recovered.token()).unwrap();
}

#[test]
fn queue_renew_competes_atomically_with_claim_and_ack() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let (mut renewed, mut acked, mut reclaimed) = (0u32, 0u32, 0u32);
    for round in 0..RACE_ROUNDS {
        q.push(FixedJob(round), None).unwrap();
        let token = q.lease(1, Some(RACE_LEASE)).unwrap().pop().unwrap().token();
        let leased_at = Instant::now();
        let barrier = Barrier::new(4);
        let (renew, ack, mut claim) = thread::scope(|scope| {
            let renew = scope.spawn(|| {
                barrier.wait();
                q.renew(token, Some(RACE_RENEWAL))
            });
            let ack = scope.spawn(|| {
                barrier.wait();
                q.ack(token)
            });
            let claim = scope.spawn(|| {
                barrier.wait();
                q.lease(1, None)
            });
            sleep_until(leased_at + race_release_offset(round));
            barrier.wait();
            (
                renew.join().unwrap(),
                ack.join().unwrap(),
                claim.join().unwrap().unwrap(),
            )
        });
        let renew_won = owned_or_expired(round, "renew", renew);
        let ack_won = owned_or_expired(round, "ack", ack);
        let claimed = claim.pop();
        // The row is settled exactly once: by the holder's ack or by the claimer.
        assert_ne!(
            ack_won,
            claimed.is_some(),
            "round {round}: ack and claim must settle the row exactly once"
        );
        assert!(
            !(renew_won && claimed.is_some()),
            "round {round}: a renewal and a reclaim both won the same lease"
        );
        if let Some(fresh) = claimed {
            assert_ne!(fresh.token().epoch(), token.epoch());
            q.ack(fresh.token()).unwrap();
            reclaimed += 1;
        }
        renewed += u32::from(renew_won);
        acked += u32::from(ack_won);
        assert!(matches!(q.renew(token, None), Err(DbError::LeaseExpired)));
        assert_eq!(q.len(), 0, "round {round}");
    }
    eprintln!("renewed={renewed} acked={acked} reclaimed={reclaimed} of {RACE_ROUNDS}");
}

#[test]
fn queue_renew_expired_holder_racing_reclaim_cannot_steal_new_lease() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let (mut renewed, mut reclaimed, mut neither) = (0u32, 0u32, 0u32);
    for round in 0..RACE_ROUNDS {
        q.push(FixedJob(round), None).unwrap();
        let old = q.lease(1, Some(RACE_LEASE)).unwrap().pop().unwrap().token();
        let leased_at = Instant::now();
        let barrier = Barrier::new(3);
        let (renew, mut claim) = thread::scope(|scope| {
            let renew = scope.spawn(|| {
                barrier.wait();
                q.renew(old, Some(RACE_RENEWAL))
            });
            let claim = scope.spawn(|| {
                barrier.wait();
                q.lease(1, None)
            });
            sleep_until(leased_at + race_release_offset(round));
            barrier.wait();
            (renew.join().unwrap(), claim.join().unwrap().unwrap())
        });
        let renew_won = owned_or_expired(round, "renew", renew);
        match (renew_won, claim.pop()) {
            (true, Some(_)) => {
                panic!("round {round}: a renewal and a reclaim both won the same lease")
            }
            (true, None) => {
                renewed += 1;
                // Past the original deadline the renewal still holds the row.
                sleep_until(leased_at + RACE_LEASE + Duration::from_millis(2));
                assert!(q.lease(1, None).unwrap().is_empty(), "round {round}");
                q.ack(old).unwrap();
            }
            (false, Some(fresh)) => {
                reclaimed += 1;
                assert_ne!(fresh.token().epoch(), old.epoch());
                assert_eq!(fresh.attempts(), 0);
                assert!(matches!(
                    q.renew(old, Some(RACE_RENEWAL)),
                    Err(DbError::LeaseExpired)
                ));
                assert!(matches!(q.ack(old), Err(DbError::LeaseExpired)));
                q.renew(fresh.token(), None).unwrap();
                q.ack(fresh.token()).unwrap();
            }
            (false, None) => {
                // The claimer sampled the clock before the deadline, the renewer
                // after it; the expired row is still reclaimable.
                neither += 1;
                let fresh = q.lease(1, None).unwrap().pop().unwrap();
                assert_ne!(fresh.token().epoch(), old.epoch());
                assert!(matches!(q.ack(old), Err(DbError::LeaseExpired)));
                q.ack(fresh.token()).unwrap();
            }
        }
        assert_eq!(q.len(), 0, "round {round}");
    }
    eprintln!("renewed={renewed} reclaimed={reclaimed} neither={neither} of {RACE_ROUNDS}");
    assert!(
        renewed > 0 && reclaimed > 0,
        "the sweep never straddled the deadline: \
         renewed={renewed} reclaimed={reclaimed} neither={neither}"
    );
}

#[test]
fn queue_renew_dead_lettered_token_cannot_restore_live_depth() {
    let dir = tempfile::tempdir().unwrap();
    let db = armdb::armour::Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().max_attempts(1).build().unwrap())
        .unwrap();
    q.push(FixedJob(91), None).unwrap();
    let token = q.lease(1, None).unwrap().pop().unwrap().token();
    q.nack(token).unwrap();
    assert!(matches!(q.renew(token, None), Err(DbError::LeaseExpired)));
    assert_eq!(q.len(), 0);
    assert_eq!(q.dead_len(), 1);
    assert!(q.lease(1, None).unwrap().is_empty());
}