bench-scenarios/src/promote.rs
//! Promotion-through-jera throughput scenario.
//!
//! Seals a burst of **distinct** artifacts through the REAL
//! [`server_lib::promote::PromotionEngine`] with a durable
//! [`jera::jobs::JobStore`] ledger attached, then reports the throughput of the
//! whole hot path a live promotion runs:
//!
//! 1. blake3 **content-address** the bytes,
//! 2. write the content-addressed **blob** (atomic tmp+rename),
//! 3. create the exclusive **ref** (the immutability seal),
//! 4. write the `running → done` **job row** into the redb ledger.
//!
//! Distinct bytes per artifact defeat content-address dedup, so every seal does
//! the full blob write + a fresh ledger row — the realistic per-promotion cost.
//! The ledger is a single-writer redb file (exactly how the server drives it), so
//! this is a serial writer measurement, not a fan-out.
//!
//! Twin correctness coverage lives in `server/lib/src/promote/mod.rs` unit tests
//! (`seal_records_a_done_job_in_the_ledger` et al.); this module is the *numbers*.
use std::sync::Arc;
use std::time::Instant;
use anyhow::{Context, Result};
use server_lib::promote::PromotionEngine;
use traits::ArtifactId;
use crate::{Scale, Throughput};
/// How many artifacts to seal for a given [`Scale`]. Capped so a `full`/`huge`
/// run doesn't try to write half a million real blobs to disk — the ledger write
/// rate saturates well before that, and the cap keeps the bench bounded while
/// still being HEAVY (tens of thousands of seals + ledger rows).
fn seal_count(scale: &Scale) -> usize {
scale.entries.clamp(2_000, 20_000)
}
/// Bytes per sealed artifact. Small enough that thousands fit on disk, large
/// enough that the blake3 + blob-write cost is real (not a rounding error next to
/// the ledger write).
const ARTIFACT_BYTES: usize = 16 * 1024;
/// Build a distinct payload for artifact `i` — a deterministic fill keyed by `i`
/// so no two artifacts share a content address (no dedup short-circuit).
fn payload(i: usize) -> Vec<u8> {
let mut v = vec![0u8; ARTIFACT_BYTES];
let seed = (i as u64).wrapping_mul(0x9E3779B97F4A7C15).wrapping_add(1);
for (j, b) in v.iter_mut().enumerate() {
*b = ((seed ^ (j as u64).wrapping_mul(131)) & 0xFF) as u8;
}
v
}
/// Seal [`seal_count`] distinct artifacts through a real [`PromotionEngine`] with
/// a jera job ledger attached, returning the throughput (seals/sec = `ops_sec`,
/// bytes/sec = `mb_per_sec`). Every seal is a durable `done` job in the ledger;
/// the count is asserted at the end so a broken ledger fails the bench instead of
/// silently reporting a hollow number.
pub fn promote_ledger_throughput(scale: &Scale) -> Result<Throughput> {
let n = seal_count(scale);
// The prod store and the ledger both live under a scratch dir the OS reclaims.
let store_dir = tempfile::tempdir().context("promote bench: tempdir (prod store)")?;
let led_dir = tempfile::tempdir().context("promote bench: tempdir (job ledger)")?;
let ledger = Arc::new(
jera::jobs::JobStore::open(led_dir.path()).context("promote bench: open jera ledger")?,
);
let engine = PromotionEngine::new(store_dir.path()).with_job_ledger(ledger.clone());
// Pre-build the payloads so payload synthesis is not in the timed region.
let payloads: Vec<Vec<u8>> = (0..n).map(payload).collect();
let start = Instant::now();
let mut bytes: u64 = 0;
for (i, data) in payloads.iter().enumerate() {
let id = ArtifactId {
namespace: None,
name: format!("bench-crate-{i}"),
version: "1.0.0".into(),
};
engine
.seal(
"dev",
"prod",
&id,
data,
"bench@holger",
1_700_000_000 + i as i64,
)
.map_err(|e| anyhow::anyhow!("promote bench: seal {i} failed: {e}"))?;
bytes += data.len() as u64;
}
let elapsed = start.elapsed();
// Every seal must have produced a durable `done` job — the ledger is the point.
// jera bounds `jobs.redb` to the most recent `CAP` terminal jobs (a long-running
// server's job list is intentionally bounded; holger's *permanent* promotion
// record is its separate audit log), so once the burst exceeds the cap the
// ledger retains exactly `CAP` done rows. Assert the retained count matches that
// bound — proving every seal wrote a done row AND the cap prunes as designed.
let done = ledger
.list(&jera::jobs::JobSelector::Kind("promote".into()))
.context("promote bench: list ledger")?
.into_iter()
.filter(|j| j.status == "done")
.count();
let want = n.min(jera::jobs::CAP);
anyhow::ensure!(
done == want,
"promote bench: expected {want} retained done jobs (of {n} seals, jera cap {}), found {done}",
jera::jobs::CAP
);
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
Ok(Throughput {
ops: n as u64,
elapsed,
bytes,
ops_sec: n as f64 / s,
mb_per_sec: bytes as f64 / s / 1_048_576.0,
})
}
/// **The same seals from every core at once** (stinky 30): [`seal_count`]
/// distinct artifacts sealed through ONE engine + ledger by a gatling fan-out —
/// what concurrent `PromoteArtifact` calls and the job queue's workers do. Every
/// seal must land as a `done` row, as in [`promote_ledger_throughput`].
pub fn promote_ledger_concurrent_throughput(scale: &Scale) -> Result<Throughput> {
let n = seal_count(scale);
let store_dir = tempfile::tempdir().context("promote bench: tempdir (prod store)")?;
let led_dir = tempfile::tempdir().context("promote bench: tempdir (job ledger)")?;
let ledger = Arc::new(
jera::jobs::JobStore::open(led_dir.path()).context("promote bench: open jera ledger")?,
);
let engine = PromotionEngine::new(store_dir.path()).with_job_ledger(ledger.clone());
let payloads: Vec<Vec<u8>> = (0..n).map(payload).collect();
let start = Instant::now();
let failed = gatling::gatling_forkjoin::gatling_for_each(n, 0, |i| {
let id = ArtifactId {
namespace: None,
name: format!("bench-crate-{i}"),
version: "1.0.0".into(),
};
engine
.seal(
"dev",
"prod",
&id,
&payloads[i],
"bench@holger",
1_700_000_000 + i as i64,
)
.err()
.map(|e| e.to_string())
});
let elapsed = start.elapsed();
if let Some(e) = failed.into_iter().flatten().next() {
anyhow::bail!("promote bench: a concurrent seal failed: {e}");
}
let done = ledger
.list(&jera::jobs::JobSelector::Kind("promote".into()))
.context("promote bench: list ledger")?
.into_iter()
.filter(|j| j.status == "done")
.count();
let want = n.min(jera::jobs::CAP);
anyhow::ensure!(
done == want,
"promote bench: expected {want} retained done jobs, found {done}"
);
let bytes = (n * ARTIFACT_BYTES) as u64;
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
Ok(Throughput {
ops: n as u64,
elapsed,
bytes,
ops_sec: n as f64 / s,
mb_per_sec: bytes as f64 / s / 1_048_576.0,
})
}
// ─────────────────────────────────────────────────────────────────────────────
// The job queue — the same seals, admitted instead of run (stinky 01 §1.3)
// ─────────────────────────────────────────────────────────────────────────────
/// A runner that does exactly what the inline promote did: seal through the REAL
/// engine. It is the same work; only WHERE it runs changes.
struct SealRunner {
engine: Arc<PromotionEngine>,
payloads: Vec<Vec<u8>>,
}
impl server_lib::jobs::JobRunner for SealRunner {
fn run(&self, job: &server_lib::jobs::Job) -> Result<String> {
let server_lib::jobs::Job::Promote {
from_repo,
to_repo,
artifact,
requested_by,
} = job
else {
anyhow::bail!("this bench admits promotions only");
};
// The payload index is carried in the artifact name, which is how the
// bench keeps the runner free of a lookup that would be measured.
let i: usize = artifact
.name
.rsplit('-')
.next()
.and_then(|s| s.parse().ok())
.context("bench payload index")?;
let id = ArtifactId {
namespace: None,
name: artifact.name.clone(),
version: artifact.version.clone(),
};
let rec = self
.engine
.seal(
from_repo,
to_repo,
&id,
&self.payloads[i],
requested_by,
1_700_000_000 + i as i64,
)
.map_err(|e| anyhow::anyhow!("bench seal {i}: {e}"))?;
Ok(rec.content_address)
}
}
fn bench_job(i: usize) -> server_lib::jobs::Job {
server_lib::jobs::Job::Promote {
from_repo: "dev".into(),
to_repo: "prod".into(),
artifact: server_lib::jobs::JobArtifact {
namespace: String::new(),
name: format!("bench-crate-{i}"),
version: "1.0.0".into(),
},
requested_by: "bench@holger".into(),
}
}
/// **What the RPC caller now waits for.** Admits [`seal_count`] promotions to the
/// durable queue and times the ADMISSION alone — the durable `queued` row plus
/// the in-memory hand-off, with no runner attached so not one seal can have been
/// done. `ops_sec` is promotions ADMITTED per second, which is the number that
/// replaces "the client blocks for the whole seal" from stinky 01 §1.3.
///
/// This is deliberately NOT comparable to `promote_ledger` as "faster sealing":
/// it measures a different thing on purpose, and the two together are the whole
/// story — the caller is released at the admission rate and the work drains at
/// [`promote_queue_drain_throughput`].
pub fn promote_queue_admission_throughput(scale: &Scale) -> Result<Throughput> {
let n = seal_count(scale);
let queue_dir = tempfile::tempdir().context("promote queue bench: tempdir")?;
let (queue, _recovered) = server_lib::jobs::JobQueue::open(queue_dir.path(), 0)
.context("promote queue bench: open the queue")?;
let jobs: Vec<_> = (0..n).map(bench_job).collect();
let start = Instant::now();
for job in jobs {
queue.enqueue(job).context("promote queue bench: admit")?;
}
let elapsed = start.elapsed();
anyhow::ensure!(
queue.open_jobs() == n,
"promote queue bench: expected {n} open jobs, found {}",
queue.open_jobs()
);
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
Ok(Throughput {
ops: n as u64,
elapsed,
bytes: (n * ARTIFACT_BYTES) as u64,
ops_sec: n as f64 / s,
mb_per_sec: (n * ARTIFACT_BYTES) as f64 / s / 1_048_576.0,
})
}
/// **End to end through the queue**: admit [`seal_count`] promotions, then run
/// them on the gatling worker pool and wait for the last one to settle.
/// `ops_sec` is seals/second including the queue's own durable rows
/// (`queued → running → done`, three ledger writes per job against
/// `promote_ledger`'s two), so it is a PESSIMISTIC comparison — the queue is
/// carrying more bookkeeping than the inline path ever did.
pub fn promote_queue_drain_throughput(scale: &Scale) -> Result<Throughput> {
let n = seal_count(scale);
let store_dir = tempfile::tempdir().context("promote queue bench: tempdir (prod store)")?;
let queue_dir = tempfile::tempdir().context("promote queue bench: tempdir (queue)")?;
let engine = Arc::new(PromotionEngine::new(store_dir.path()));
let payloads: Vec<Vec<u8>> = (0..n).map(payload).collect();
let (queue, _recovered) = server_lib::jobs::JobQueue::open(queue_dir.path(), 0)
.context("promote queue bench: open the queue")?;
let workers = queue.workers();
let jobs: Vec<_> = (0..n).map(bench_job).collect();
let start = Instant::now();
let mut last = None;
for job in jobs {
last = Some(queue.enqueue(job).context("promote queue bench: admit")?);
}
queue.start(Arc::new(SealRunner { engine, payloads }));
let last = last.context("promote queue bench: nothing admitted")?;
let settled = queue
.await_completion(&last, std::time::Duration::from_secs(600))
.context("promote queue bench: await")?
.context("promote queue bench: the last job never settled")?;
anyhow::ensure!(
settled.status == server_lib::jobs::status::DONE,
"promote queue bench: last job ended {} ({})",
settled.status,
settled.reason
);
// The queue drains in admission order per worker, but "the last admitted job
// is done" is not "every job is done" with N workers, so wait the rest out.
while queue.open_jobs() > 0 {
std::thread::yield_now();
}
let elapsed = start.elapsed();
queue.shutdown();
let bytes = (n * ARTIFACT_BYTES) as u64;
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
let _ = workers;
Ok(Throughput {
ops: n as u64,
elapsed,
bytes,
ops_sec: n as f64 / s,
mb_per_sec: bytes as f64 / s / 1_048_576.0,
})
}