armdb 0.10.0

//! Queue throughput and scan-risk benchmarks.
//!
//! ```sh
//! cargo bench -p armdb --bench queue -- --sample-size 10
//! ```
//!
//! The `lease_scan/10k_outstanding` result is compared against the explicit
//! operational target printed by this benchmark: at most 10 ms per scan on the
//! benchmark host. It deliberately has no finger cache; the result is the cost
//! of the v1 stable-key design with 10,000 outstanding leases.

use std::hint::black_box;
use std::time::Duration;

use armdb::armour::Db;
use armdb::{QueueConfig, queue};
use armour_core::GetType;
use criterion::{Criterion, Throughput, criterion_group, criterion_main};
use rapira::Rapira;
use tempfile::TempDir;

const OUTSTANDING_LEASES: u64 = 10_000;
const OUTSTANDING_SCAN_TARGET: &str = "<= 10 ms per lease scan";

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

#[derive(Clone, GetType, Rapira)]
pub struct VarJob(Vec<u8>);

queue!(FixedJobs, payload = FixedJob, name = "bench_queue_fixed");
queue!(VarJobs, payload = VarJob, name = "bench_queue_var", var);

fn bench_fixed_cycle(c: &mut Criterion) {
    let dir = TempDir::new().expect("tempdir");
    let db = Db::open_test(dir.path()).expect("open database");
    let queue = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().expect("config"))
        .expect("open fixed queue");
    // Correctness is asserted around the measurement, not inside `b.iter`:
    // an assertion in the hot loop both perturbs the sample and is unreachable
    // from every test command, since benches are a separate build target.
    cycle_once(&queue);
    assert_eq!(queue.stats().compaction_bytes, 0);

    let mut group = c.benchmark_group("queue_cycle");
    group.throughput(Throughput::Elements(1));
    group.bench_function("fixed/push_lease_ack", |b| b.iter(|| cycle_once(&queue)));
    group.finish();

    assert_eq!(
        queue.stats().compaction_bytes,
        0,
        "a fixed queue updates in place and never compacts"
    );
}

fn cycle_once(queue: &armdb::Queue<FixedJobs>) {
    queue.push(FixedJob(7), None).expect("push");
    let row = queue
        .lease(1, Some(Duration::from_secs(60)))
        .expect("lease")
        .pop()
        .expect("one row");
    queue.ack(row.token()).expect("ack");
}

fn bench_var_cycle(c: &mut Criterion) {
    let dir = TempDir::new().expect("tempdir");
    let db = Db::open_test(dir.path()).expect("open database");
    let queue = db
        .open_queue::<VarJobs>(QueueConfig::balanced().build().expect("config"))
        .expect("open variable queue");
    // One payload write per push, asserted over a controlled run outside the
    // measurement — see `bench_fixed_cycle` for why it is not in `b.iter`.
    const CALIBRATION: u64 = 16;
    for _ in 0..CALIBRATION {
        var_cycle_once(&queue);
    }
    assert_eq!(
        queue.stats().payload_writes,
        CALIBRATION,
        "a lease must not rewrite the payload"
    );

    let mut group = c.benchmark_group("queue_cycle");
    group.throughput(Throughput::Elements(1));
    group.bench_function("var/push_lease_ack", |b| b.iter(|| var_cycle_once(&queue)));
    group.finish();
}

fn var_cycle_once(queue: &armdb::Queue<VarJobs>) {
    queue.push(VarJob(vec![7; 128]), None).expect("push");
    let row = queue
        .lease(1, Some(Duration::from_secs(60)))
        .expect("lease")
        .pop()
        .expect("one row");
    queue.ack(row.token()).expect("ack");
}

fn bench_outstanding_lease_scan(c: &mut Criterion) {
    eprintln!(
        "queue lease_scan/{OUTSTANDING_LEASES}_outstanding target: {OUTSTANDING_SCAN_TARGET}"
    );
    let dir = TempDir::new().expect("tempdir");
    let db = Db::open_test(dir.path()).expect("open database");
    let queue = db
        .open_queue::<FixedJobs>(QueueConfig::balanced().build().expect("config"))
        .expect("open fixed queue");
    for id in 0..OUTSTANDING_LEASES {
        queue
            .push(FixedJob(id), None)
            .expect("push outstanding job");
    }
    assert_eq!(
        queue
            .lease(
                OUTSTANDING_LEASES as usize,
                Some(Duration::from_secs(60 * 60)),
            )
            .expect("lease outstanding jobs")
            .len(),
        OUTSTANDING_LEASES as usize
    );

    let mut group = c.benchmark_group("lease_scan");
    group.throughput(Throughput::Elements(OUTSTANDING_LEASES));
    group.bench_function("10k_outstanding", |b| {
        b.iter(|| black_box(queue.lease(1, Some(Duration::from_secs(60))).expect("scan")));
    });
    group.finish();
}

criterion_group!(
    benches,
    bench_fixed_cycle,
    bench_var_cycle,
    bench_outstanding_lease_scan
);
criterion_main!(benches);