armdb 0.11.2

use std::mem::size_of;
use std::process::Command;
use std::sync::{Arc, Barrier};
use std::thread;
use std::time::{Duration, 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);
}