armdb 0.14.2

use std::sync::Barrier;
use std::thread;
use std::time::{Duration, Instant};

use armdb::armour::Db;
use armdb::{CollectionMeta, DbError, FixedConfig, NoHook, QueueConfig, TxConfig, queue};
use armour_core::GetType;
use rapira::Rapira;

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

queue!(VarJobs, payload = VarJob, name = "queue_var_jobs", var);
queue!(
    OtherVarJobs,
    payload = VarJob,
    name = "queue_var_other_jobs",
    var
);
queue!(
    VarJobsSchemaDrift,
    payload = VarJob,
    name = "queue_var_schema_drift",
    var
);

#[derive(Clone, Debug, PartialEq, Eq, GetType, Rapira)]
pub struct ChangedVarJob {
    body: String,
    priority: i32,
}
queue!(
    VarJobsSchemaDriftChanged,
    payload = ChangedVarJob,
    name = "queue_var_schema_drift",
    var
);

queue!(
    PayloadNameCollision,
    payload = VarJob,
    name = "queue_var_collision",
    var
);
queue!(
    PayloadOccupier,
    payload = u64,
    name = "queue_var_collision.payload"
);

#[test]
fn open_queue_rejects_unbuilt_invalid_var_policies() {
    let assert_invalid = |config| {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        assert!(matches!(
            db.open_queue::<VarJobs>(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));
}

struct CompatibleStagedHeader;

impl armdb::QueueSpec for CompatibleStagedHeader {
    type Payload = VarJob;
    type Mode = armdb::VarQueueMode;

    const NAME: &'static str = "queue_var_collision";
    const PAYLOAD_NAME: &'static str = "queue_var_collision.compatible_payload";
}

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

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

#[test]
fn lease_does_not_rewrite_the_var_payload_and_ack_removes_both_rows() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    q.push(
        VarJob {
            body: "once".into(),
        },
        None,
    )
    .unwrap();
    let row = q
        .lease(1, Some(Duration::from_secs(10)))
        .unwrap()
        .pop()
        .unwrap();
    assert_eq!(row.payload().body, "once");
    assert_eq!(q.debug_payload_write_count(), 1);
    q.ack(row.token()).unwrap();
    assert_eq!(q.debug_payload_len(), 0);
    assert_eq!(q.len(), 0);
}

#[test]
fn reopen_collects_an_orphan_var_payload() {
    let dir = tempfile::tempdir().unwrap();
    {
        let db = Db::open_test(dir.path()).unwrap();
        let q = db
            .open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
            .unwrap();
        q.debug_insert_orphan_payload(
            42,
            VarJob {
                body: "orphan".into(),
            },
        )
        .unwrap();
        assert_eq!(q.debug_payload_len(), 1);
    }

    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    assert_eq!(q.debug_payload_len(), 0);
}

#[test]
fn nack_retains_the_var_payload_across_retry_and_dead_letter_moves() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let retry = db
        .open_queue::<VarJobs>(
            QueueConfig::balanced()
                .max_attempts(2)
                .backoff_secs(&[0])
                .build()
                .unwrap(),
        )
        .unwrap();
    retry
        .push(
            VarJob {
                body: "retry".into(),
            },
            None,
        )
        .unwrap();
    let first = retry.lease(1, None).unwrap().pop().unwrap();
    retry.nack(first.token()).unwrap();
    let second = retry.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(second.payload().body, "retry");
    retry.nack(second.token()).unwrap();
    assert_eq!(retry.len(), 0);
    assert_eq!(retry.dead_len(), 1);
    assert_eq!(retry.debug_payload_len(), 1);
}

#[test]
fn nack_invalidates_the_token_before_the_var_retry_is_leased_again() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let queue = db
        .open_queue::<VarJobs>(
            QueueConfig::balanced()
                .max_attempts(2)
                .backoff_secs(&[0])
                .build()
                .unwrap(),
        )
        .unwrap();
    queue
        .push(
            VarJob {
                body: "retry".into(),
            },
            None,
        )
        .unwrap();
    let leased = queue.lease(1, None).unwrap().pop().unwrap();

    queue.nack(leased.token()).unwrap();

    assert!(matches!(
        queue.ack(leased.token()),
        Err(DbError::LeaseExpired)
    ));
}

#[test]
fn transaction_push_and_ack_pair_var_payload_with_the_header() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    db.tx()
        .push(
            &q,
            VarJob {
                body: "atomic".into(),
            },
            None,
        )
        .unwrap()
        .commit()
        .unwrap();
    assert_eq!(q.len(), 1);
    assert_eq!(q.debug_payload_len(), 1);
    let row = q.lease(1, None).unwrap().pop().unwrap();
    db.tx().ack(&q, row.token()).unwrap().commit().unwrap();
    assert_eq!(q.len(), 0);
    assert_eq!(q.debug_payload_len(), 0);
}

#[test]
fn transactional_var_pushes_past_max_depth_are_refused_at_commit() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(QueueConfig::balanced().max_depth(Some(1)).build().unwrap())
        .unwrap();

    let tx = db
        .tx()
        .push(
            &q,
            VarJob {
                body: "first".into(),
            },
            None,
        )
        .unwrap()
        .push(
            &q,
            VarJob {
                body: "second".into(),
            },
            None,
        )
        .unwrap();

    assert!(matches!(tx.commit(), Err(DbError::QueueFull)));
    assert_eq!(q.len(), 0);
    assert_eq!(q.debug_payload_len(), 0);
}

/// Spec: `armdb/docs/bugs/26-09-10-transactional-queue-depth-ack-push.md`.
fn var_job(body: &str) -> VarJob {
    VarJob { body: body.into() }
}

fn open_full_var_queue_with_one_leased_row() -> (
    tempfile::TempDir,
    Db,
    std::sync::Arc<armdb::Queue<VarJobs>>,
    armdb::LeaseToken,
) {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(
            QueueConfig::balanced()
                .max_depth(Some(1))
                .max_attempts(2)
                .backoff_secs(&[0])
                .build()
                .unwrap(),
        )
        .unwrap();
    q.push(var_job("first"), None).unwrap();
    let token = q.lease(1, None).unwrap().pop().unwrap().token();
    (dir, db, q, token)
}

#[test]
fn transactional_var_ack_then_push_commits_at_max_depth() {
    let (_dir, db, q, token) = open_full_var_queue_with_one_leased_row();

    db.tx()
        .ack(&q, token)
        .unwrap()
        .push(&q, var_job("second"), None)
        .unwrap()
        .commit()
        .unwrap();

    assert_eq!(q.len(), 1);
    assert_eq!(q.debug_payload_len(), 1);
    let next = q.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(next.payload().body, "second");
}

#[test]
fn transactional_var_push_then_ack_commits_at_max_depth() {
    let (_dir, db, q, token) = open_full_var_queue_with_one_leased_row();

    db.tx()
        .push(&q, var_job("second"), None)
        .unwrap()
        .ack(&q, token)
        .unwrap()
        .commit()
        .unwrap();

    assert_eq!(q.len(), 1);
    assert_eq!(q.debug_payload_len(), 1);
    let next = q.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(next.payload().body, "second");
}

#[test]
fn transactional_var_ack_credits_exactly_one_push_at_max_depth() {
    let (_dir, db, q, token) = open_full_var_queue_with_one_leased_row();

    let tx = db
        .tx()
        .ack(&q, token)
        .unwrap()
        .push(&q, var_job("second"), None)
        .unwrap()
        .push(&q, var_job("third"), None)
        .unwrap();

    assert!(matches!(tx.commit(), Err(DbError::QueueFull)));
    assert_eq!(q.len(), 1);
    assert_eq!(q.debug_payload_len(), 1);
    q.ack(token).unwrap();
    assert_eq!(q.len(), 0);
}

#[test]
fn transactional_var_push_without_ack_is_refused_at_max_depth() {
    let (_dir, db, q, _token) = open_full_var_queue_with_one_leased_row();

    let tx = db.tx().push(&q, var_job("second"), None).unwrap();

    assert!(matches!(tx.commit(), Err(DbError::QueueFull)));
    assert_eq!(q.len(), 1);
    assert_eq!(q.debug_payload_len(), 1);
}

#[test]
fn transactional_var_ack_on_one_queue_does_not_admit_a_push_on_another() {
    let (_dir, db, q, token) = open_full_var_queue_with_one_leased_row();
    let other = db
        .open_queue::<OtherVarJobs>(QueueConfig::balanced().max_depth(Some(1)).build().unwrap())
        .unwrap();
    other.push(var_job("other-first"), None).unwrap();

    let tx = db
        .tx()
        .ack(&q, token)
        .unwrap()
        .push(&other, var_job("other-second"), None)
        .unwrap();

    assert!(matches!(tx.commit(), Err(DbError::QueueFull)));
    assert_eq!(q.len(), 1);
    assert_eq!(other.len(), 1);
    assert_eq!(other.debug_payload_len(), 1);
    q.ack(token).unwrap();
}

#[test]
fn transactional_var_stale_ack_plus_push_fails_with_lease_expired_not_queue_full() {
    let (_dir, db, q, stale) = open_full_var_queue_with_one_leased_row();
    q.nack(stale).unwrap();
    assert_eq!(q.len(), 1);

    let tx = db
        .tx()
        .ack(&q, stale)
        .unwrap()
        .push(&q, var_job("second"), None)
        .unwrap();

    assert!(matches!(tx.commit(), Err(DbError::LeaseExpired)));
    assert_eq!(q.len(), 1);
    assert_eq!(q.debug_payload_len(), 1);
    let still_there = q.lease(1, None).unwrap().pop().unwrap();
    assert_eq!(still_there.payload().body, "first");
}

#[test]
fn dropped_var_transaction_with_ack_and_push_changes_nothing() {
    let (_dir, db, q, token) = open_full_var_queue_with_one_leased_row();

    let tx = db
        .tx()
        .ack(&q, token)
        .unwrap()
        .push(&q, var_job("second"), None)
        .unwrap();
    drop(tx);

    assert_eq!(q.len(), 1);
    assert_eq!(q.debug_payload_len(), 1);
    q.ack(token).unwrap();
    assert_eq!(q.len(), 0);
}

#[test]
fn transactional_var_dead_letter_nack_grants_no_push_credit() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(
            QueueConfig::balanced()
                .max_depth(Some(1))
                .max_attempts(1)
                .backoff_secs(&[0])
                .build()
                .unwrap(),
        )
        .unwrap();
    q.push(var_job("first"), None).unwrap();
    let token = q.lease(1, None).unwrap().pop().unwrap().token();

    let tx = db
        .tx()
        .nack(&q, token)
        .unwrap()
        .push(&q, var_job("second"), None)
        .unwrap();

    assert!(matches!(tx.commit(), Err(DbError::QueueFull)));
    assert_eq!(q.len(), 1);
    assert_eq!(q.dead_len(), 0);
    assert_eq!(q.debug_payload_len(), 1);
}

#[test]
fn transactional_var_ack_plus_push_commits_on_a_queue_already_above_max_depth() {
    let dir = tempfile::tempdir().unwrap();
    {
        let db = Db::open_test(dir.path()).unwrap();
        let q = db
            .open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
            .unwrap();
        q.push(var_job("first"), None).unwrap();
        q.push(var_job("second"), None).unwrap();
        drop(q);
        drop(db);
    }
    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(QueueConfig::balanced().max_depth(Some(1)).build().unwrap())
        .unwrap();
    assert_eq!(q.len(), 2);
    let token = q.lease(1, None).unwrap().pop().unwrap().token();

    db.tx()
        .ack(&q, token)
        .unwrap()
        .push(&q, var_job("third"), None)
        .unwrap()
        .commit()
        .unwrap();
    assert_eq!(q.len(), 2);
    // The ack dropped one payload and the push added one: no leak, no loss.
    assert_eq!(q.debug_payload_len(), 2);

    let tx = db.tx().push(&q, var_job("fourth"), None).unwrap();
    assert!(matches!(tx.commit(), Err(DbError::QueueFull)));
    assert_eq!(q.len(), 2);
    assert_eq!(q.debug_payload_len(), 2);
}

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

    let result = Db::open_test(dir.path())
        .unwrap()
        .open_queue::<VarJobsSchemaDriftChanged>(QueueConfig::balanced().build().unwrap());
    let Err(err) = result else {
        panic!("payload type drift must fail managed reopen");
    };
    assert!(matches!(err, DbError::SchemaMismatch { .. }), "{err}");
}

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

    assert!(
        db.open_queue::<PayloadNameCollision>(QueueConfig::balanced().build().unwrap())
            .is_err()
    );

    // A metadata-compatible Var queue at the staged header name proves the
    // failed composite open did not publish its header into Db's live registries.
    db.open_queue::<CompatibleStagedHeader>(QueueConfig::balanced().build().unwrap())
        .expect("failed composite Var open must release its unpublished header name");
}

#[test]
fn failed_transaction_with_var_push_keeps_both_rows_and_count() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_test(dir.path()).unwrap();
    let q = db
        .open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let effects = db
        .open_zero_tree_fixed::<Effect, 8, _>(FixedConfig::test(), NoHook, &[])
        .unwrap();
    let result = db
        .tx()
        .push(
            &q,
            VarJob {
                body: "atomic".into(),
            },
            None,
        )
        .unwrap()
        .guard(&effects, &1_u64.to_be_bytes(), |_| {
            Err(DbError::Client("abort var queue tx"))
        })
        .unwrap()
        .commit();

    assert!(matches!(result, Err(DbError::Client("abort var queue tx"))));
    assert_eq!(q.len(), 0);
    assert_eq!(q.debug_payload_len(), 0);
    assert!(q.lease(1, None).unwrap().is_empty());
}

#[test]
fn var_ack_and_nack_reject_max_ops_before_any_mutation() {
    for nack in [false, true] {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::builder(dir.path())
            .for_tests()
            .tx_config(TxConfig {
                max_ops: 1,
                ..TxConfig::default()
            })
            .open()
            .unwrap();
        let q = db
            .open_queue::<VarJobs>(QueueConfig::balanced().backoff_secs(&[0]).build().unwrap())
            .unwrap();
        q.push(
            VarJob {
                body: "limit".into(),
            },
            None,
        )
        .unwrap();
        let leased = q.lease(1, None).unwrap().pop().unwrap();
        let result = if nack {
            db.tx().nack(&q, leased.token())
        } else {
            db.tx().ack(&q, leased.token())
        };
        assert!(matches!(
            result,
            Err(DbError::Client("transaction exceeds max_ops"))
        ));
        assert_eq!(q.len(), 1);
        assert_eq!(q.debug_payload_len(), 1);
    }
}

/// 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:?}"),
    }
}

fn race_job(round: u64) -> VarJob {
    VarJob {
        body: format!("renewed-{round}"),
    }
}

#[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::<VarJobs>(
            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 = VarJob {
        body: "renewed".into(),
    };
    q.push(expected.clone(), 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, 1);
    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::<VarJobs>(QueueConfig::balanced().backoff_secs(&[0]).build().unwrap())
        .unwrap();
    q.push(
        VarJob {
            body: "renewed".into(),
        },
        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(),
        &VarJob {
            body: "renewed".into()
        }
    );
    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::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    q.push(
        VarJob {
            body: "renewed".into(),
        },
        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::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    q.push(
        VarJob {
            body: "renewed".into(),
        },
        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::<VarJobs>(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(),
        &VarJob {
            body: "renewed".into()
        }
    );
    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::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let (mut renewed, mut acked, mut reclaimed) = (0u32, 0u32, 0u32);
    for round in 0..RACE_ROUNDS {
        q.push(race_job(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());
            assert_eq!(fresh.payload(), &race_job(round));
            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}");
        assert_eq!(q.debug_payload_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::<VarJobs>(QueueConfig::balanced().build().unwrap())
        .unwrap();
    let (mut renewed, mut reclaimed, mut neither) = (0u32, 0u32, 0u32);
    for round in 0..RACE_ROUNDS {
        q.push(race_job(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_eq!(fresh.payload(), &race_job(round));
                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}");
        assert_eq!(q.debug_payload_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::<VarJobs>(QueueConfig::balanced().max_attempts(1).build().unwrap())
        .unwrap();
    q.push(
        VarJob {
            body: "renewed".into(),
        },
        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());
}