server/lib/src/jobs.rs
//! # The job queue — the four long operations, off the RPC and onto a durable ledger
//!
//! Stinky 01 §1.3 measured the state this module ends: **promote, replicate,
//! upload and gate all run INLINE on the RPC**, nothing resumes after a restart,
//! and a `running` row is permanent once the process dies. And `promote_ledger`
//! measured **606 seals/s on 12 cores against 349 on 32** — the seal path was
//! strictly serial, so a customer who bought a bigger box got faster reads and
//! SLOWER writes.
//!
//! ## What this is, and what it is NOT
//!
//! It is **not a second ledger**. The house already has one — `jera::jobs`
//! (`edda/crates/jera`), the durable redb list of every long-running operation,
//! which holger's promotion engine has written to since `with_job_ledger`. jera
//! is a LEDGER: it records `running → done|failed`. What was missing was the
//! half that DEQUEUES. This module is exactly that half, written ON jera's rows:
//!
//! * a [`Job`] enum over the four long operations ([`Job::Promote`],
//! [`Job::Replicate`], [`Job::Upload`], [`Job::Gate`]) — the payload is
//! serialized into the jera row's `detail_json`, so the queue needs no table
//! of its own and a reader that only knows jera still sees every job;
//! * a [`JobQueue`] over `<data_root>/jobs` that admits a job as `queued`,
//! returns its [`JobId`] AT ONCE, and runs it on a worker pool;
//! * the pool is **gatling** (`gatling::gatling_forkjoin::gatling_run` on one
//! `gatling::background::Job` supervisor thread) — ROOT LAW #0: the house
//! engine, never rayon. `the_worker_pool_is_gatling_not_rayon` is a census
//! test over this file's own source, the way the engine guards do it;
//! * **crash recovery**: at open, every non-terminal row is either RESUMED (the
//! idempotent kinds) or moved to [`status::INTERRUPTED`] with a reason. A
//! `running` row never survives a restart as `running` — that is
//! `a_running_job_does_not_survive_a_restart_as_running`.
//!
//! ## Idempotency
//!
//! Every job carries a [`Job::content_id`] — a stable key over what it acts on
//! (`promote:dev->prod:ns:name:version`). [`JobQueue::enqueue`] refuses to admit
//! a second non-terminal job with the same content id and hands back the FIRST
//! job's id instead. Two clients racing the same coordinate get one job, one
//! seal, two identical answers — `two_jobs_for_the_same_coordinate_do_not_both_seal`.
//! The seal's own `create_new` ref remains the last word; this only stops the
//! work from being done twice.
//!
//! ## Additive by construction
//!
//! The whole module is behind the `jobs` feature. Promote and upload reach it
//! when a caller asks for the async shape (`async_job = true` on the RPC); an
//! EXECUTING replication reaches it whenever the server has a queue (stinky 7,
//! 2026-10-22), because a replication is the bulk copy a restart must resume.
//! With the feature off, or no queue configured, they run exactly as they
//! did — same bytes, same audit rows, same synchronous answer. Nothing published
//! changed: `holger-traits` is untouched by this lane.
use crate::lock::{recover, LockOrRecover as _, RwLockOrRecover as _};
use std::collections::{HashMap, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
/// The statuses a holger job row carries. Four of the five are jera's own
/// (`queued`, `running`, `done`, `failed`); the fifth is holger's and exists
/// because stinky 01 §6.2 named the hole it fills.
pub mod status {
pub use jera::jobs::status::{DONE, FAILED, QUEUED, RUNNING};
/// **A job that was running when the process died and cannot be resumed.**
///
/// jera has no such status because jera is a ledger written by a live
/// process: its `stopped` means "intentionally stopped, and that is not a
/// failure". An interrupted job is neither — it is a job whose outcome
/// nobody knows. Spelling it as `failed` would claim the work was tried and
/// refused; leaving it `running` is the defect stinky 01 §6.2 measured.
///
/// Terminal for holger ([`is_terminal`]), and jera's own `is_terminal` says
/// `false` for it, which is deliberate: jera's pruner never drops a row it
/// does not understand, so an interrupted job outlives the rotation until an
/// operator has seen it.
pub const INTERRUPTED: &str = "interrupted";
/// **A retryable job whose last attempt failed and whose next attempt is
/// scheduled** ([`super::Job::max_attempts`] > 1: a pull-through
/// [`super::Job::Fetch`], a mirror [`super::Job::MirrorAsset`]).
///
/// NOT terminal: the job still owns its content id (a second request for the
/// same artifact joins it), a restart resumes it, and the row carries the
/// reason and the time of the next attempt so the admin view can say
/// `failed(reason, next attempt at …)` while it waits.
pub const RETRY_WAIT: &str = "retry_wait";
/// Whether `s` is terminal **to holger** — jera's three plus
/// [`INTERRUPTED`].
pub fn is_terminal(s: &str) -> bool {
jera::jobs::status::is_terminal(s) || s == INTERRUPTED
}
}
/// The jera `kind` tag each variant writes. jera kinds are free strings by
/// design ("a new op needs no schema change"), so holger's four are namespaced
/// with its own prefix and never collide with nornir's.
pub mod kind {
pub const PROMOTE: &str = "holger_promote";
pub const REPLICATE: &str = "holger_replicate";
pub const UPLOAD: &str = "holger_upload";
pub const GATE: &str = "holger_gate";
pub const FETCH: &str = "holger_fetch";
pub const MIRROR_ASSET: &str = "holger_mirror_asset";
pub const IMPORT_INBOX: &str = "holger_import_inbox";
pub const INTEL_REFRESH: &str = "holger_intel_refresh";
pub const PULL_INTAKE: &str = "holger_pull_intake";
pub const SYNC_TICK: &str = "holger_sync_tick";
pub const EXPORT_PUSH: &str = "holger_export_push";
}
/// **What the job is doing, in the words the admin view prints** — derived
/// from the row's status and kind, never stored twice.
///
/// A pull-through reads `requested → fetching → done | failed`; every other
/// kind reads the ledger's own word (`queued`, `running`, …).
pub fn stage_of(kind: &str, status_word: &str) -> &'static str {
let fetchy = kind == kind::FETCH || kind == kind::MIRROR_ASSET || kind == kind::PULL_INTAKE;
match status_word {
s if s == status::QUEUED => {
if fetchy {
"requested"
} else {
"queued"
}
}
s if s == status::RUNNING => {
if fetchy {
"fetching"
} else {
"running"
}
}
s if s == status::RETRY_WAIT => "failed",
s if s == status::DONE => "done",
s if s == status::FAILED => "failed",
s if s == status::INTERRUPTED => "interrupted",
_ => "unknown",
}
}
/// **A failure retrying cannot fix** — attach it (`.context(JobPermanent(..))`
/// or `Err(anyhow!(JobPermanent(..)))`) and a retryable job goes straight to
/// `failed` instead of waiting for its next attempt. "No upstream publishes this
/// coordinate" is the case: asking again in a minute gets the same answer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JobPermanent(pub String);
impl std::fmt::Display for JobPermanent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl std::error::Error for JobPermanent {}
fn is_permanent(e: &anyhow::Error) -> bool {
e.chain()
.any(|c| c.downcast_ref::<JobPermanent>().is_some())
}
/// **Admission priority** on the pending list: lower runs first. Every job
/// that is not a pull-through is [`PRIORITY_FULL`], so the order among them is
/// FIFO exactly as before; a pull-through admitted for a `Reduced`, `Floor` or
/// `Queued` consumer waits behind the `Full` ones.
pub const PRIORITY_FULL: u8 = 0;
pub const PRIORITY_REDUCED: u8 = 1;
pub const PRIORITY_FLOOR: u8 = 2;
pub const PRIORITY_QUEUED: u8 = 3;
/// The first retry delay of a retryable job ([`Job::max_attempts`] > 1).
pub const DEFAULT_RETRY_BASE: Duration = Duration::from_secs(2);
/// The longest a retry waits: the delay doubles per attempt up to this.
pub const RETRY_CAP: Duration = Duration::from_secs(300);
/// The delay before attempt `attempts + 1`, after `attempts` failures.
pub fn retry_delay(base: Duration, attempts: u32) -> Duration {
let factor = 1u32
.checked_shl(attempts.saturating_sub(1).min(16))
.unwrap_or(u32::MAX);
base.saturating_mul(factor).min(RETRY_CAP.max(base))
}
/// A job's durable identity — what an RPC returns before the work is done.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct JobId(pub String);
impl JobId {
pub fn as_str(&self) -> &str {
&self.0
}
}
impl std::fmt::Display for JobId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
/// The coordinate a job acts on, flattened out of `traits::ArtifactId` so the
/// payload serializes without dragging the trait crate's shape into the ledger.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct JobArtifact {
#[serde(default)]
pub namespace: String,
pub name: String,
pub version: String,
}
impl JobArtifact {
pub fn label(&self) -> String {
if self.namespace.is_empty() {
format!("{}:{}", self.name, self.version)
} else {
format!("{}:{}:{}", self.namespace, self.name, self.version)
}
}
}
/// **The four long operations**, as one durable value.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub enum Job {
/// The DEV → PROD seal (`PromotionService.PromoteArtifact`).
Promote {
from_repo: String,
to_repo: String,
artifact: JobArtifact,
/// The authenticated principal the seal is recorded under.
requested_by: String,
},
/// A repository-to-repository copy (`ReplicationService.Replicate`).
Replicate {
from_repo: String,
to_repo: String,
execute: bool,
limit: usize,
requested_by: String,
},
/// **The gated finalize half of an upload** — the bytes are already staged
/// on disk by the door; this is the digest check plus the backend `put`
/// that used to hold the request thread for 1.75 s per GiB (stinky §2.2).
Upload {
repo: String,
artifact: JobArtifact,
/// Where the door staged the body.
staged: PathBuf,
/// Lower-case hex sha256 the client claimed, or `""` for unchecked.
expect_sha256: String,
requested_by: String,
},
/// Gate evaluation on its own (the OSV scan promote used to run inline).
Gate {
from_repo: String,
to_repo: String,
artifact: JobArtifact,
requested_by: String,
},
/// **A pull-through of an artifact that is not local** (`crate::pull_through`).
/// The first request for it enqueues this and is answered `503` +
/// `Retry-After`; the job walks the proxy's upstream chain (which writes the
/// bytes through into the cache), retrying with backoff; the client's retry
/// is then a local hit. Deduplicated on the coordinate, never on who asked.
Fetch {
repo: String,
artifact: JobArtifact,
/// The consumer the upstream bytes are charged to (`Cost.upstream_fetch`).
requested_by: String,
},
/// **One mirror-cycle asset that did not land** (`crate::mirror_cycle`). The
/// cycle hands it here and moves on; this job owns its retry and its status.
MirrorAsset {
source: String,
from_repo: String,
to_repo: String,
/// The holger coordinate (`mirror_cycle::id_to_coord`).
coord: String,
requested_by: String,
},
/// **Sweep an import drop directory** (lane W4, [`crate::import_door`]):
/// every bundle an operator put in `<data_root>/intel.inbox` or
/// `<data_root>/bundle.inbox` is imported through the same door as the
/// RPCs, then moved to `archive/` or `refused/`. Enqueued by the server's
/// inbox timer only when a file waits; one open sweep per inbox.
#[cfg(feature = "bundle-import")]
ImportInbox {
kind: crate::import_door::ImportKind,
},
/// **`intel.refresh`** (lane W4): with `intel.offline: false` the box fetches
/// OSV, NVD and KEV itself and lands the set through the import door
/// ([`crate::import_door::ImportDoor::refresh_intel`]). Queued once at boot
/// and daily; one open refresh at a time.
IntelRefresh,
/// **A pull-through INTO THE INTAKE CHAIN** ([`crate::pull_intake`]): fetch
/// what a client asked a remote source for, ingest it into `incoming` as
/// `pull-through:<source>`, and walk it as far as policy lets it. One job
/// per want, deduplicated; who asked is not part of it.
#[cfg(feature = "intake")]
PullIntake {
source: String,
want: crate::pull_intake::Want,
},
/// **One tick of the sync protocol** ([`crate::sync`], H4): the alarms due on an
/// upstream; on a downstream with an upstream `url`, one pull and its ack.
/// Queued by the server's sync timer (`sync.tick_secs`); one open at a time.
SyncTick,
/// **One push into the diode** (lane HOL-CDS, [`crate::export_push`]): the new
/// AVAILABLE artifacts as one signed bundle into `export.push_to`. Queued by the
/// server's push timer (`export.every`); one open at a time.
ExportPush,
}
/// **Where a job writes artifact bytes** (SEC-REVIEW 2026-11-01, the guard for the class). Every
/// kind answers in [`Job::writes`], with no catch-all arm, so a new kind does not compile until
/// it says which door its bytes take; the runner asks the intake chain about the answer before
/// the job runs ([`ServerJobRunner`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobWrites<'a> {
/// No artifact bytes land in a repository (a gate run, an intel refresh, an export push).
Nothing,
/// The bytes land through the intake chain's own door (a pull-through into incoming, a
/// bundle import, a sync pull through the import door).
ThroughTheChain,
/// Raw bytes into `repo`'s backend (a queued upload, a mirror asset).
Raw { repo: &'a str },
/// A copy of `from`'s artifacts into `to` (a replication).
Copy { from: &'a str, to: &'a str },
/// A seal of `from`'s artifact into `to` (the promotion engine).
Promote { from: &'a str, to: &'a str },
/// A proxy's cache fill: its upstream's bytes into `repo`'s cache (the boot refuses an
/// unscanned proxy where a scan is required).
CacheFill { repo: &'a str },
}
impl Job {
/// Where this job writes artifact bytes ([`JobWrites`]). Exhaustive on purpose.
pub fn writes(&self) -> JobWrites<'_> {
match self {
Job::Promote {
from_repo, to_repo, ..
} => JobWrites::Promote {
from: from_repo,
to: to_repo,
},
Job::Replicate {
from_repo, to_repo, ..
} => JobWrites::Copy {
from: from_repo,
to: to_repo,
},
Job::Upload { repo, .. } => JobWrites::Raw { repo },
Job::Gate { .. } => JobWrites::Nothing,
Job::Fetch { repo, .. } => JobWrites::CacheFill { repo },
Job::MirrorAsset { to_repo, .. } => JobWrites::Raw { repo: to_repo },
#[cfg(feature = "bundle-import")]
Job::ImportInbox { .. } => JobWrites::ThroughTheChain,
Job::IntelRefresh => JobWrites::Nothing,
#[cfg(feature = "intake")]
Job::PullIntake { .. } => JobWrites::ThroughTheChain,
Job::SyncTick => JobWrites::ThroughTheChain,
Job::ExportPush => JobWrites::Nothing,
}
}
/// The jera `kind` tag for this job.
pub fn kind(&self) -> &'static str {
match self {
Job::Promote { .. } => kind::PROMOTE,
Job::Replicate { .. } => kind::REPLICATE,
Job::Upload { .. } => kind::UPLOAD,
Job::Gate { .. } => kind::GATE,
Job::Fetch { .. } => kind::FETCH,
Job::MirrorAsset { .. } => kind::MIRROR_ASSET,
#[cfg(feature = "bundle-import")]
Job::ImportInbox { .. } => kind::IMPORT_INBOX,
Job::IntelRefresh => kind::INTEL_REFRESH,
#[cfg(feature = "intake")]
Job::PullIntake { .. } => kind::PULL_INTAKE,
Job::SyncTick => kind::SYNC_TICK,
Job::ExportPush => kind::EXPORT_PUSH,
}
}
/// **How many attempts this job gets** before a failure is terminal. The
/// four original kinds get one — they ran once before this field existed and
/// still do. A pull-through and a mirror asset live on somebody else's
/// network, so they are retried with backoff.
pub fn max_attempts(&self) -> u32 {
match self {
Job::Fetch { .. } | Job::MirrorAsset { .. } => 5,
#[cfg(feature = "intake")]
Job::PullIntake { .. } => 5,
_ => 1,
}
}
/// The human target the ledger row names.
pub fn target(&self) -> String {
match self {
Job::Promote {
from_repo,
to_repo,
artifact,
..
}
| Job::Gate {
from_repo,
to_repo,
artifact,
..
} => {
format!("{from_repo}->{to_repo}:{}", artifact.label())
}
Job::Replicate {
from_repo, to_repo, ..
} => format!("{from_repo}->{to_repo}"),
Job::Upload { repo, artifact, .. } | Job::Fetch { repo, artifact, .. } => {
format!("{repo}:{}", artifact.label())
}
Job::MirrorAsset {
source,
from_repo,
to_repo,
coord,
..
} => {
format!("{source}:{from_repo}->{to_repo}:{coord}")
}
#[cfg(feature = "bundle-import")]
Job::ImportInbox { kind } => kind.inbox_name().to_string(),
Job::IntelRefresh => "intel.refresh".to_string(),
#[cfg(feature = "intake")]
Job::PullIntake { source, want } => format!("{source}:{}", want.key()),
Job::SyncTick => "sync.tick".to_string(),
Job::ExportPush => "export.push".to_string(),
}
}
/// Who asked.
pub fn requested_by(&self) -> &str {
match self {
Job::Promote { requested_by, .. }
| Job::Replicate { requested_by, .. }
| Job::Upload { requested_by, .. }
| Job::Gate { requested_by, .. }
| Job::Fetch { requested_by, .. }
| Job::MirrorAsset { requested_by, .. } => requested_by,
#[cfg(feature = "bundle-import")]
Job::ImportInbox { .. } => "inbox",
Job::IntelRefresh => "intel.refresh",
#[cfg(feature = "intake")]
Job::PullIntake { .. } => "pull-through",
Job::SyncTick => "sync",
Job::ExportPush => "export.push",
}
}
/// **The idempotency key.** Two jobs with the same content id are the same
/// work: the queue admits the first and hands its id to the second.
///
/// Deliberately excludes the principal — WHO asked for a seal does not make
/// it a different seal, and the whole point is that two clients racing one
/// coordinate do not both seal it.
pub fn content_id(&self) -> String {
match self {
Job::Promote {
from_repo,
to_repo,
artifact,
..
} => {
format!("promote:{from_repo}->{to_repo}:{}", artifact.label())
}
Job::Gate {
from_repo,
to_repo,
artifact,
..
} => {
format!("gate:{from_repo}->{to_repo}:{}", artifact.label())
}
Job::Replicate {
from_repo,
to_repo,
execute,
limit,
..
} => {
format!("replicate:{from_repo}->{to_repo}:{execute}:{limit}")
}
Job::Upload {
repo,
artifact,
expect_sha256,
..
} => {
format!("upload:{repo}:{}:{expect_sha256}", artifact.label())
}
// WHO asked is excluded on purpose: ten clients asking for one
// artifact is one fetch.
Job::Fetch { repo, artifact, .. } => format!("fetch:{repo}:{}", artifact.label()),
Job::MirrorAsset {
source,
from_repo,
to_repo,
coord,
..
} => {
format!("mirror:{source}:{from_repo}->{to_repo}:{coord}")
}
#[cfg(feature = "bundle-import")]
Job::ImportInbox { kind } => format!("inbox:{}", kind.inbox_name()),
Job::IntelRefresh => "intel:refresh".to_string(),
#[cfg(feature = "intake")]
Job::PullIntake { source, want } => format!("pull-intake:{source}:{}", want.key()),
Job::SyncTick => "sync:tick".to_string(),
Job::ExportPush => "export:push".to_string(),
}
}
/// **Whether a restart may RESUME this job rather than interrupt it.**
///
/// `Replicate`, `Upload` and `Gate` are idempotent: re-running them either
/// re-does work that lands on the same bytes (replication skips by content
/// id, an upload of staged bytes is a `put` of the same blob) or decides the
/// same verdict again. `Promote` is NOT resumable and never will be: the
/// seal is the immutability commit, the source bytes may have moved under
/// it, and an interrupted promotion is precisely the case the operator must
/// be told about by name rather than have silently retried.
pub fn is_resumable(&self) -> bool {
!matches!(self, Job::Promote { .. })
}
}
/// One job as a reader sees it (`JobService.Get`/`List`, the CLI, a test).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JobView {
pub id: JobId,
pub kind: String,
pub target: String,
pub status: String,
pub content_id: String,
/// The terminal result pointer (a content address for a seal, a count for a
/// replication), or `""`.
pub result_ref: String,
/// Why it failed or was interrupted, or `""`.
pub reason: String,
pub requested_by: String,
pub ts_start_micros: i64,
pub ts_end_micros: Option<i64>,
pub elapsed_ms: Option<i64>,
/// [`stage_of`] the row: `requested`/`fetching`/`done`/`failed` for a
/// pull-through, the ledger's word for the rest.
#[serde(default)]
pub stage: String,
/// Attempts made so far (0 while it has never run).
#[serde(default)]
pub attempts: u32,
/// The attempt ceiling ([`Job::max_attempts`]).
#[serde(default)]
pub max_attempts: u32,
/// When a [`status::RETRY_WAIT`] job runs again (µs since the epoch).
#[serde(default)]
pub next_attempt_micros: Option<i64>,
/// Pending-list priority (`PRIORITY_*`; lower runs first).
#[serde(default)]
pub priority: u8,
}
impl JobView {
/// Waiting to run: admitted and never started, or failed with a next
/// attempt scheduled.
pub fn is_waiting(&self) -> bool {
self.status == status::QUEUED || self.status == status::RETRY_WAIT
}
pub fn is_terminal(&self) -> bool {
status::is_terminal(&self.status)
}
}
/// What a worker does with a dequeued [`Job`]. The queue owns durability,
/// dispatch and recovery; the runner owns the work, so the queue can be tested
/// with a runner that only counts.
pub trait JobRunner: Send + Sync + 'static {
/// Run the job. `Ok(result_ref)` is written onto the row as its terminal
/// pointer; `Err` moves the row to `failed` with the error chain as reason.
fn run(&self, job: &Job) -> Result<String>;
}
/// The envelope actually stored in the jera row's `detail_json`. One shape for
/// every status, so a reader never has to guess which fields exist.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Envelope {
job: Job,
content_id: String,
#[serde(default)]
reason: String,
/// Attempts made (absent on a row written before retries existed ⇒ 0).
#[serde(default)]
attempts: u32,
/// The next attempt of a [`status::RETRY_WAIT`] job, µs since the epoch.
#[serde(default)]
next_attempt_micros: Option<i64>,
/// Pending-list priority, kept so a restart re-queues it where it was.
#[serde(default)]
priority: u8,
}
impl Envelope {
fn new(job: Job, content_id: String, priority: u8) -> Self {
Self {
job,
content_id,
reason: String::new(),
attempts: 0,
next_attempt_micros: None,
priority,
}
}
}
/// What [`JobQueue::open`]'s recovery sweep did — the answer to "what happened
/// to the work that was in flight when we died".
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RecoverySummary {
/// Non-terminal rows re-admitted as `queued`.
pub resumed: usize,
/// Non-terminal rows moved to [`status::INTERRUPTED`].
pub interrupted: usize,
/// Rows whose payload could not be decoded (an older row, or a row written
/// by the ledger-only promotion path). Also interrupted — never left
/// `running` — but they cannot be resumed because nobody knows what they were.
pub undecodable: usize,
}
impl RecoverySummary {
pub fn touched(&self) -> usize {
self.resumed + self.interrupted + self.undecodable
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The pool
// ─────────────────────────────────────────────────────────────────────────────
struct Pending {
id: JobId,
job: Job,
/// `PRIORITY_*`; lower runs first, FIFO within one priority.
priority: u8,
/// Not before this instant — a retry's backoff. `None` = ready now.
not_before: Option<Instant>,
/// Attempts already made (carried across a retry and a restart).
attempts: u32,
}
impl Pending {
fn fresh(id: JobId, job: Job, priority: u8) -> Self {
Self {
id,
job,
priority,
not_before: None,
attempts: 0,
}
}
}
struct Shared {
state: Mutex<PoolState>,
/// Woken when work arrives or the pool is asked to stop.
work: Condvar,
/// Woken when any job reaches a terminal status — what
/// [`JobQueue::await_completion`] waits on.
settled: Condvar,
}
struct PoolState {
pending: VecDeque<Pending>,
stopping: bool,
/// **The open-work index**: `content_id -> job_id` for every job that is
/// admitted and not yet terminal. Idempotency is answered from here, in
/// memory, and NOT by re-listing the ledger per admission — a scan per
/// enqueue is quadratic in the ledger, which is the very shape stinky 01
/// §7.1 measured and named. The ledger is still the durable truth; this is
/// its index, rebuilt from it at [`JobQueue::recover`].
open_content: HashMap<String, String>,
/// `job_id -> status` for the same reason: `await_completion` must not scan
/// the ledger in a loop.
statuses: HashMap<String, String>,
}
impl Shared {
fn push(&self, item: Pending) {
let mut st = self.state.lock_or_recover();
st.pending.push_back(item);
drop(st);
self.work.notify_one();
}
/// Block until there is work or the pool is stopping. `None` = wind down.
///
/// The READY item with the lowest priority number, first-come within it. A
/// list of priority-0, never-delayed jobs — everything before pull-through
/// existed — therefore pops in exactly the FIFO order it always did.
fn pop(&self) -> Option<Pending> {
let mut st = self.state.lock_or_recover();
loop {
let now = Instant::now();
let mut best: Option<(usize, u8)> = None;
let mut earliest: Option<Instant> = None;
for (i, p) in st.pending.iter().enumerate() {
match p.not_before {
Some(t) if t > now => {
earliest = Some(earliest.map_or(t, |e: Instant| e.min(t)));
}
_ => {
if best.is_none_or(|(_, bp)| p.priority < bp) {
best = Some((i, p.priority));
}
}
}
}
if let Some((i, _)) = best {
return st.pending.remove(i);
}
// Ready work drains first, as it always did; a DELAYED retry is left
// on the list (its row is durable) when the pool winds down.
if st.stopping {
return None;
}
st = match earliest {
Some(t) => recover(self.work.wait_timeout(st, t.saturating_duration_since(now))).0,
None => recover(self.work.wait(st)),
};
}
}
/// A second caller for an open job with a more urgent priority moves the
/// pending item up — a `Full` consumer is not held behind the `Floor` one
/// who happened to ask first.
fn raise(&self, id: &str, priority: u8) {
let mut st = self.state.lock_or_recover();
if let Some(p) = st.pending.iter_mut().find(|p| p.id.as_str() == id) {
if priority < p.priority {
p.priority = priority;
}
}
}
fn stop(&self) {
let mut st = self.state.lock_or_recover();
st.stopping = true;
drop(st);
self.work.notify_all();
}
fn announce_settled(&self) {
self.settled.notify_all();
}
}
/// **The durable queue.** One per server, opened under `<data_root>/jobs`.
pub struct JobQueue {
/// The ledger (`<root>/jobs.redb`), `None` while it is closed for a data-set
/// growth ([`JobQueue::release`]) — one of the files that made `/data`
/// unmountable while holger ran. Read through [`JobQueue::ledger`].
store: std::sync::RwLock<Option<Arc<jera::jobs::JobStore>>>,
/// The runner [`start`](Self::start) attached, kept so the pool can be
/// started again after a release.
runner: Mutex<Option<Arc<dyn JobRunner>>>,
shared: Arc<Shared>,
root: PathBuf,
workers: usize,
seq: AtomicU64,
/// The first retry delay of a retryable job; doubles per attempt, capped
/// at [`RETRY_CAP`]. [`DEFAULT_RETRY_BASE`] in production; a test shortens it.
retry_base: Duration,
/// The supervisor thread running the gatling pool; `None` until
/// [`JobQueue::start`] and after [`JobQueue::shutdown`].
pump: Mutex<Option<gatling::background::Job<()>>>,
/// **The box's role, read live** ([`JobQueue::follow_role`]): a TWIN runs
/// no job and admits none — it only follows its primary (stinky SB #34).
role: std::sync::OnceLock<Arc<crate::twin::TwinPosture>>,
}
impl JobQueue {
/// Open (or create) the queue under `<data_root>/jobs` and run the recovery
/// sweep. The returned queue accepts [`enqueue`](Self::enqueue) at once;
/// nothing RUNS until [`start`](Self::start) attaches a runner, which is what
/// lets a test write a ledger, drop the process state and re-open it.
pub fn open(data_root: &Path, workers: usize) -> Result<(Self, RecoverySummary)> {
let root = data_root.join("jobs");
std::fs::create_dir_all(&root)
.with_context(|| format!("create job queue root {}", root.display()))?;
std::fs::create_dir_all(root.join("staging"))
.with_context(|| format!("create job staging dir under {}", root.display()))?;
let store = Arc::new(
jera::jobs::JobStore::open(&root)
.context("open the jera job ledger for holger jobs")?,
);
let workers = if workers == 0 {
gatling::gatling_forkjoin::default_workers()
} else {
workers
};
let queue = Self {
store: std::sync::RwLock::new(Some(store)),
runner: Mutex::new(None),
shared: Arc::new(Shared {
state: Mutex::new(PoolState {
pending: VecDeque::new(),
stopping: false,
open_content: HashMap::new(),
statuses: HashMap::new(),
}),
work: Condvar::new(),
settled: Condvar::new(),
}),
root,
workers,
seq: AtomicU64::new(0),
retry_base: DEFAULT_RETRY_BASE,
pump: Mutex::new(None),
role: std::sync::OnceLock::new(),
};
let summary = queue.recover()?;
Ok((queue, summary))
}
/// **Follow this box's role** (the posture the doors ask). Set once, at
/// wiring, before [`start`](Self::start).
pub fn follow_role(self: &Arc<Self>, posture: Arc<crate::twin::TwinPosture>) {
if self.role.set(Arc::clone(&posture)).is_err() {
return;
}
// Elected a twin LIVE, by any road: the pool stands down on gatling's
// sanctioned background job, detached (it joins the workers; the caller
// may be on the executor).
let me = Arc::downgrade(self);
posture.on_become_twin(Box::new(move || {
if let Some(q) = me.upgrade() {
// A32-14, stays a task: the sanctioned gatling background job that stands the queue down for a twin; it joins the workers.
drop(gatling::background::Job::spawn(move || {
q.stand_down_for_twin()
}));
}
}));
}
/// Whether the worker pool runs now.
pub fn is_running(&self) -> bool {
self.pump.lock_or_recover().is_some()
}
/// **A primary elected a twin live** (`BecomeTwin`): its workers stand down
/// — the job each is on finishes, the pending rows stay on the ledger — and
/// the runner is forgotten, so a growth's [`reacquire`](Self::reacquire)
/// starts none again. Blocking (it joins the pool): off the executor.
pub fn stand_down_for_twin(&self) {
self.shutdown();
*self.runner.lock_or_recover() = None;
}
/// The refusal a TWIN gives every job (`holger-twin-20`), or `None` on a
/// primary — read live, so a box elected a twin refuses from that instant.
pub fn twin_refusal(&self) -> Option<String> {
let posture = self.role.get()?;
let role = posture.role();
role.primary().map(|p| {
holger_errcode::TWIN_RUNS_NO_JOBS.tag(format!(
"this box is a TWIN of `{p}`: it runs no job and admits none — it only follows \
its primary; the job belongs there"
))
})
}
/// Where the door stages an upload body before enqueuing [`Job::Upload`].
pub fn staging_dir(&self) -> PathBuf {
self.root.join("staging")
}
/// The worker count this queue fans over.
pub fn workers(&self) -> usize {
self.workers
}
/// **Crash recovery — no row is left `running`.**
///
/// Every non-terminal row is read once: a resumable job goes back to
/// `queued` and into the in-memory pending list; everything else (a promote,
/// or a row whose payload will not decode) moves to [`status::INTERRUPTED`]
/// with a reason naming the restart.
pub fn recover(&self) -> Result<RecoverySummary> {
let rows = self
.ledger()?
.list(&jera::jobs::JobSelector::All)
.context("list the job ledger for recovery")?;
let mut summary = RecoverySummary::default();
for mut row in rows {
if !row.kind.starts_with("holger_") || status::is_terminal(&row.status) {
continue;
}
match serde_json::from_str::<Envelope>(&row.detail_json) {
Ok(env) if env.job.is_resumable() => {
// A retry that was WAITING keeps waiting — its backoff is
// the upstream's breathing room, and a restart is no reason
// to hammer it. Everything else goes back to `queued`.
let waiting = row.status == status::RETRY_WAIT;
let not_before = if waiting {
env.next_attempt_micros.map(|at| {
Instant::now()
+ Duration::from_micros((at - now_micros()).max(0) as u64)
})
} else {
None
};
if !waiting {
row.status = status::QUEUED.to_string();
}
row.ts_end_micros = None;
row.elapsed_ms = None;
self.ledger()?
.submit(&row)
.context("re-queue a resumable job")?;
{
let mut st = self.shared.state.lock_or_recover();
st.open_content
.insert(env.content_id.clone(), row.job_id.clone());
st.statuses.insert(row.job_id.clone(), row.status.clone());
}
self.shared.push(Pending {
id: JobId(row.job_id.clone()),
job: env.job,
priority: env.priority,
not_before,
attempts: env.attempts,
});
summary.resumed += 1;
}
Ok(env) => {
self.mark_interrupted(
&mut row,
Some(env),
"the server restarted while this job was in flight; a promotion \
is not resumed because the seal is the immutability commit",
)?;
summary.interrupted += 1;
}
Err(e) => {
self.mark_interrupted(
&mut row,
None,
&format!(
"the server restarted while this job was in flight and its \
payload could not be decoded ({e}); it cannot be resumed"
),
)?;
summary.undecodable += 1;
}
}
}
Ok(summary)
}
fn mark_interrupted(
&self,
row: &mut jera::jobs::JobRecord,
env: Option<Envelope>,
reason: &str,
) -> Result<()> {
let end = now_micros();
row.status = status::INTERRUPTED.to_string();
row.ts_end_micros = Some(end);
row.elapsed_ms = Some((end - row.ts_start_micros).max(0) / 1000);
if let Some(mut env) = env {
env.reason = reason.to_string();
row.detail_json = serde_json::to_string(&env).unwrap_or_default();
} else {
row.detail_json = serde_json::json!({ "reason": reason }).to_string();
}
self.ledger()?
.submit(row)
.context("mark an in-flight job interrupted")
}
/// Attach the runner and fan N workers over the pending list.
///
/// ROOT LAW #0: the pool is `gatling_run` on one `gatling::background::Job`
/// supervisor — the house engine and its sanctioned single background
/// thread, never rayon and never a hand-rolled pool.
pub fn start(&self, runner: Arc<dyn JobRunner>) {
// A TWIN only follows (stinky SB #34): no worker, and no runner kept for
// a later restart to start.
if let Some(why) = self.twin_refusal() {
log::warn!("job queue: no worker started — {why}");
return;
}
*self.runner.lock_or_recover() = Some(Arc::clone(&runner));
let mut slot = self.pump.lock_or_recover();
if slot.is_some() {
return;
}
let shared = Arc::clone(&self.shared);
let Ok(store) = self.ledger() else { return };
let workers = self.workers;
let retry_base = self.retry_base;
let role = self.role.get().cloned();
// A32-14, stays a task: the sanctioned gatling background job; its handle is kept in `slot`.
*slot = Some(gatling::background::Job::spawn(move || {
gatling::gatling_forkjoin::gatling_run(workers, workers, |_worker| {
while let Some(item) = shared.pop() {
// A box that became a twin runs nothing more: the item goes
// back (its row is on the ledger) and this worker ends.
if role.as_ref().is_some_and(|p| p.is_twin()) {
shared.push(item);
break;
}
run_one(&store, &shared, &*runner, item, retry_base);
}
});
}));
}
/// **Admit a job and return its id at once.**
///
/// Writes the `queued` row BEFORE the in-memory push, so a crash between the
/// two leaves a durable row the next [`recover`](Self::recover) picks up —
/// never a job that was accepted and forgotten.
///
/// Idempotent by [`Job::content_id`]: a non-terminal job for the same
/// content id is returned instead of a second one being admitted.
pub fn enqueue(&self, job: Job) -> Result<JobId> {
self.enqueue_with_priority(job, PRIORITY_FULL)
}
/// [`enqueue`](Self::enqueue) at an admission priority (`PRIORITY_*`). A
/// job already open for the same content id is JOINED — its id is returned,
/// and its pending priority is raised when this caller's is more urgent.
///
/// The content id is RESERVED under the pool lock before the row is
/// written, so two callers racing the same coordinate cannot both miss the
/// open-work index and admit two jobs.
pub fn enqueue_with_priority(&self, job: Job, priority: u8) -> Result<JobId> {
if let Some(why) = self.twin_refusal() {
anyhow::bail!(why);
}
let content_id = job.content_id();
let id = {
let mut st = self.shared.state.lock_or_recover();
if let Some(existing) = st.open_content.get(&content_id) {
let existing = JobId(existing.clone());
drop(st);
self.shared.raise(existing.as_str(), priority);
return Ok(existing);
}
let id = JobId(format!(
"{}-{}-{}",
job.kind(),
now_micros(),
self.seq.fetch_add(1, Ordering::Relaxed)
));
st.open_content.insert(content_id.clone(), id.0.clone());
st.statuses.insert(id.0.clone(), status::QUEUED.to_string());
id
};
let env = Envelope::new(job.clone(), content_id, priority);
let row = jera::jobs::JobRecord {
job_id: id.0.clone(),
kind: job.kind().to_string(),
target: job.target(),
workspace: "holger".to_string(),
status: status::QUEUED.to_string(),
ts_start_micros: now_micros(),
ts_end_micros: None,
elapsed_ms: None,
detail_json: serde_json::to_string(&env).unwrap_or_default(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: Some(job.requested_by().to_string()),
actor_method: None,
};
let written = self
.ledger()
.and_then(|l| l.submit(&row).context("admit a job to the ledger"));
if let Err(e) = written {
let mut st = self.shared.state.lock_or_recover();
st.open_content.remove(&env.content_id);
st.statuses.remove(id.as_str());
return Err(e);
}
self.shared.push(Pending::fresh(id.clone(), job, priority));
Ok(id)
}
/// The id of a non-terminal job with this content id, if one is open.
pub fn open_job_for(&self, content_id: &str) -> Option<JobId> {
let st = self.shared.state.lock_or_recover();
st.open_content.get(content_id).map(|id| JobId(id.clone()))
}
/// How many admitted jobs have not yet reached a terminal status.
/// The in-memory status of a job this process admitted or recovered —
/// cheap, no ledger read. `None` for a job it never saw.
pub fn status_of(&self, id: &JobId) -> Option<String> {
let st = self.shared.state.lock_or_recover();
st.statuses.get(id.as_str()).cloned()
}
/// The first-retry delay (doubling per attempt, capped at [`RETRY_CAP`]).
/// Set before [`start`](Self::start); a test shortens it to milliseconds.
pub fn with_retry_base(mut self, base: Duration) -> Self {
self.retry_base = base;
self
}
pub fn open_jobs(&self) -> usize {
let st = self.shared.state.lock_or_recover();
st.open_content.len()
}
/// One job, by id.
pub fn get(&self, id: &JobId) -> Result<Option<JobView>> {
Ok(self
.list(0)?
.into_iter()
.find(|v| v.id.as_str() == id.as_str()))
}
/// Every holger job, newest first. `limit == 0` = all.
pub fn list(&self, limit: usize) -> Result<Vec<JobView>> {
let mut out: Vec<JobView> = self
.ledger()?
.list(&jera::jobs::JobSelector::All)?
.into_iter()
.filter(|r| r.kind.starts_with("holger_"))
.map(view_of)
.collect();
if limit > 0 && out.len() > limit {
out.truncate(limit);
}
Ok(out)
}
/// **The synchronous shape, kept available.** Block until this job is
/// terminal or `timeout` elapses; `None` on timeout.
///
/// This is what lets `async_job = true` coexist with the old behaviour
/// without a second code path: the RPC that wants the old answer enqueues
/// and waits here, and the one that wants the id returns it.
pub fn await_completion(&self, id: &JobId, timeout: Duration) -> Result<Option<JobView>> {
let deadline = Instant::now() + timeout;
loop {
{
let st = self.shared.state.lock_or_recover();
let settled = st
.statuses
.get(id.as_str())
.map(|s| status::is_terminal(s))
.unwrap_or(false);
if settled {
drop(st);
return self.get(id);
}
let now = Instant::now();
if now >= deadline {
return Ok(None);
}
let _ = recover(
self.shared
.settled
.wait_timeout(st, (deadline - now).min(Duration::from_millis(50))),
);
}
}
}
/// Wind the pool down and join the supervisor. Idempotent.
pub fn shutdown(&self) {
self.shared.stop();
let pump = self.pump.lock_or_recover().take();
if let Some(pump) = pump {
let _ = pump.join();
}
}
/// The open ledger — refused by name while it is closed for a growth.
pub fn ledger(&self) -> Result<Arc<jera::jobs::JobStore>> {
self.store.read_or_recover().clone().ok_or_else(|| {
anyhow::anyhow!(holger_errcode::GROW_STORES_CLOSED.tag(format!(
"the job ledger {}/jobs.redb is closed while the data set is unmounted for a growth",
self.root.display()
)))
})
}
/// **Close the ledger for a data-set growth** (`crate::grow`): the pool is
/// wound down (the job each worker is on finishes; the pending list stays in
/// memory, and its rows are on the ledger), then `jobs.redb` is dropped.
pub fn release(&self) -> std::result::Result<(), String> {
self.shutdown();
let taken = self.store.write_or_recover().take();
if let Some(store) = taken {
if Arc::strong_count(&store) > 1 {
*self.store.write_or_recover() = Some(store);
self.restart();
return Err(format!(
"the job ledger {}/jobs.redb is still in use",
self.root.display()
));
}
}
Ok(())
}
/// Open `jobs.redb` again and start the pool with the runner it had.
pub fn reacquire(&self) -> std::result::Result<(), String> {
{
let mut slot = self.store.write_or_recover();
if slot.is_none() {
let store = jera::jobs::JobStore::open(&self.root).map_err(|e| {
format!(
"reopen the job ledger {}/jobs.redb: {e}",
self.root.display()
)
})?;
*slot = Some(Arc::new(store));
}
}
self.restart();
Ok(())
}
fn restart(&self) {
self.shared.state.lock_or_recover().stopping = false;
let runner = self.runner.lock_or_recover().clone();
if let Some(runner) = runner {
self.start(runner);
}
}
}
impl Drop for JobQueue {
fn drop(&mut self) {
self.shutdown();
}
}
fn run_one(
store: &jera::jobs::JobStore,
shared: &Shared,
runner: &dyn JobRunner,
item: Pending,
retry_base: Duration,
) {
let Pending {
id,
job,
priority,
attempts,
..
} = item;
let content_id = job.content_id();
let started = now_micros();
let mut env = Envelope::new(job.clone(), content_id, priority);
env.attempts = attempts + 1;
let mut row = jera::jobs::JobRecord {
job_id: id.0.clone(),
kind: job.kind().to_string(),
target: job.target(),
workspace: "holger".to_string(),
status: status::RUNNING.to_string(),
ts_start_micros: started,
ts_end_micros: None,
elapsed_ms: None,
detail_json: serde_json::to_string(&env).unwrap_or_default(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: Some(job.requested_by().to_string()),
actor_method: None,
};
if let Err(e) = store.submit(&row) {
log::warn!("job {id}: could not mark running ({e:#}); running it anyway");
}
{
let mut st = shared.state.lock_or_recover();
st.statuses
.insert(id.0.clone(), status::RUNNING.to_string());
}
let outcome = runner.run(&job);
let end = now_micros();
row.ts_end_micros = Some(end);
row.elapsed_ms = Some((end - started).max(0) / 1000);
// A retryable failure with attempts left is NOT terminal: the row says
// `retry_wait` with the reason and the next attempt, the job keeps its
// content id (a new request joins it), and it goes back on the list
// behind its backoff.
let mut retry: Option<Duration> = None;
match outcome {
Ok(result_ref) => {
row.status = status::DONE.to_string();
row.result_ref = result_ref;
}
Err(e) => {
env.reason = format!("{e:#}");
if env.attempts < job.max_attempts() && !is_permanent(&e) {
let delay = retry_delay(retry_base, env.attempts);
env.next_attempt_micros = Some(end + delay.as_micros() as i64);
row.status = status::RETRY_WAIT.to_string();
row.ts_end_micros = None;
retry = Some(delay);
log::warn!(
"job {id} ({}) attempt {}/{} failed, next attempt in {:?}: {}",
job.target(),
env.attempts,
job.max_attempts(),
delay,
env.reason
);
} else {
row.status = status::FAILED.to_string();
}
row.detail_json = serde_json::to_string(&env).unwrap_or_default();
}
}
if let Err(e) = store.submit(&row) {
log::warn!("job {id}: could not write the {} row ({e:#})", row.status);
}
{
let mut st = shared.state.lock_or_recover();
st.statuses.insert(id.0.clone(), row.status.clone());
if retry.is_none() {
st.open_content.remove(&env.content_id);
}
}
if let Some(delay) = retry {
shared.push(Pending {
id,
job,
priority,
not_before: Some(Instant::now() + delay),
attempts: env.attempts,
});
}
shared.announce_settled();
}
fn view_of(row: jera::jobs::JobRecord) -> JobView {
let env = serde_json::from_str::<Envelope>(&row.detail_json).ok();
let stage = stage_of(&row.kind, &row.status).to_string();
JobView {
id: JobId(row.job_id),
stage,
kind: row.kind,
target: row.target,
status: row.status,
content_id: env
.as_ref()
.map(|e| e.content_id.clone())
.unwrap_or_default(),
result_ref: row.result_ref,
reason: env.as_ref().map(|e| e.reason.clone()).unwrap_or_default(),
requested_by: row.actor.unwrap_or_default(),
ts_start_micros: row.ts_start_micros,
ts_end_micros: row.ts_end_micros,
elapsed_ms: row.elapsed_ms,
attempts: env.as_ref().map(|e| e.attempts).unwrap_or_default(),
max_attempts: env.as_ref().map(|e| e.job.max_attempts()).unwrap_or(1),
next_attempt_micros: env.as_ref().and_then(|e| e.next_attempt_micros),
priority: env.as_ref().map(|e| e.priority).unwrap_or_default(),
}
}
fn now_micros() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or_default()
}
// ─────────────────────────────────────────────────────────────────────────────
// The server's runner — the four operations, off the request thread
// ─────────────────────────────────────────────────────────────────────────────
/// The runner the served process attaches: it owns the repository routes and
/// the promotion engine, and does exactly what the inline RPC used to do.
pub struct ServerJobRunner {
routes: Arc<crate::exposed::fast_routes::FastRoutes>,
#[cfg(feature = "intake")]
promotion: Option<Arc<crate::promote::PromotionEngine>>,
/// The mirror's transport, for [`Job::MirrorAsset`]. `None` on a box that
/// declares no mirror (such a job then fails by name, never silently).
/// It is the mirror's own connector ([`crate::mirror_cycle::MirrorJobs`]),
/// so a retried asset travels the road the cycle did: the same credential,
/// the same endpoint.
mirror: Option<Arc<crate::mirror_cycle::MirrorJobs>>,
/// The import door [`Job::ImportInbox`] sweeps through (lane W4).
#[cfg(feature = "bundle-import")]
imports: Option<Arc<crate::import_door::ImportDoor>>,
/// The intake chain a [`Job::PullIntake`] ingests into (its pull-through attached).
#[cfg(feature = "intake")]
intake: Option<Arc<crate::promote::intake::IntakeChain>>,
}
impl ServerJobRunner {
#[cfg(feature = "intake")]
pub fn new(
routes: Arc<crate::exposed::fast_routes::FastRoutes>,
promotion: Option<Arc<crate::promote::PromotionEngine>>,
) -> Self {
Self {
routes,
promotion,
mirror: None,
imports: None,
intake: None,
}
}
/// The runner of a build with no intake chain (`jobs` without `supply-chain`, H2 #3):
/// a promotion, a gate and a pull-through into the chain are refused by name.
#[cfg(not(feature = "intake"))]
pub fn new(routes: Arc<crate::exposed::fast_routes::FastRoutes>) -> Self {
Self {
routes,
mirror: None,
#[cfg(feature = "bundle-import")]
imports: None,
}
}
/// Attach the intake chain [`Job::PullIntake`] pulls into.
#[cfg(feature = "intake")]
pub fn with_intake(mut self, chain: Arc<crate::promote::intake::IntakeChain>) -> Self {
self.intake = Some(chain);
self
}
/// Attach the import door [`Job::ImportInbox`] sweeps through.
#[cfg(feature = "bundle-import")]
pub fn with_imports(mut self, door: Arc<crate::import_door::ImportDoor>) -> Self {
self.imports = Some(door);
self
}
/// Attach the mirror transport [`Job::MirrorAsset`] fetches through.
pub fn with_mirror(mut self, mirror: Arc<crate::mirror_cycle::MirrorJobs>) -> Self {
self.mirror = Some(mirror);
self
}
fn backend(&self, name: &str) -> Result<Arc<dyn traits::RepositoryBackendTrait>> {
self.routes
.lookup(name)
.ok_or_else(|| anyhow::anyhow!("repository '{name}' is not configured on this server"))
}
#[cfg(feature = "intake")]
fn engine(&self) -> Result<&crate::promote::PromotionEngine> {
self.promotion
.as_deref()
.ok_or_else(|| anyhow::anyhow!("promotion is not configured on this server"))
}
}
fn trait_id(a: &JobArtifact) -> traits::ArtifactId {
traits::ArtifactId {
namespace: if a.namespace.is_empty() {
None
} else {
Some(a.namespace.clone())
},
name: a.name.clone(),
version: a.version.clone(),
}
}
impl ServerJobRunner {
/// The intake chain's answer to where `job` writes (SEC-REVIEW 2026-11-01 P4, P9): the
/// same refusals the doors that queued it ask, asked again on the worker, so a job queued
/// before a refusal existed (a resumed one) or by a door that forgot cannot land bytes
/// past the chain.
#[cfg(feature = "intake")]
fn chain_refusal(&self, job: &Job) -> Option<crate::promote::intake::IntakeError> {
let chain = self.intake.as_ref()?;
match job.writes() {
JobWrites::Nothing | JobWrites::ThroughTheChain | JobWrites::CacheFill { .. } => None,
JobWrites::Raw { repo } => chain.raw_write_refusal(repo),
JobWrites::Copy { from, to } => chain.replication_refusal(from, to),
JobWrites::Promote { from, to } => chain.promotion_refusal(from, to),
}
}
}
impl JobRunner for ServerJobRunner {
fn run(&self, job: &Job) -> Result<String> {
#[cfg(feature = "intake")]
if let Some(refusal) = self.chain_refusal(job) {
if let Job::Upload { staged, .. } = job {
let _ = std::fs::remove_file(staged);
}
return Err(anyhow::anyhow!(JobPermanent(refusal.tagged())));
}
match job {
#[cfg(not(feature = "intake"))]
Job::Promote { .. } | Job::Gate { .. } => Err(anyhow::anyhow!(JobPermanent(
"this build has no intake chain (`supply-chain`): no promotion and no gate run here".into()
))),
#[cfg(feature = "intake")]
Job::Promote { from_repo, to_repo, artifact, requested_by } => {
let engine = self.engine()?;
let source = self.backend(from_repo)?;
let id = trait_id(artifact);
let data = source
.fetch(&id)
.map_err(|e| anyhow::anyhow!("promotion source fetch failed: {e}"))?
.ok_or_else(|| {
anyhow::anyhow!(
"artifact '{}' not found in source repo '{from_repo}'",
artifact.label()
)
})?;
// THE GATES, on the worker and not on the request thread — but
// still BEFORE the seal. Moving a promotion onto a queue must
// not move it past its gates; a job that sealed ungated would
// be a security regression dressed as a performance win.
if engine.has_gate() {
let ecosystem = crate::scanner::ecosystem_for_format(source.format());
let digest = crate::attest::ArtifactDigest::of_sha256(&data);
let attestations: Vec<crate::attest::Statement> = Vec::new();
let set = [crate::promote::PromotedArtifact {
id: &id,
ecosystem: &ecosystem,
digest,
bytes: &data,
attestations: &attestations,
}];
engine
.gate_check_hop(from_repo, to_repo, &set, requested_by)
.map_err(|e| anyhow::anyhow!("{e}"))?;
}
let at = crate::audit::now_nanos() / 1_000_000_000;
let record = engine
.seal(from_repo, to_repo, &id, &data, requested_by, at)
.map_err(|e| anyhow::anyhow!("{e}"))?;
Ok(record.content_address)
}
Job::Replicate { from_repo, to_repo, execute, limit, .. } => {
// The source through the chain's view (SEC-REVIEW P4): only what a reader may
// be served is copied.
let source = self.backend(from_repo)?;
#[cfg(feature = "intake")]
let source = match &self.intake {
Some(chain) => chain.read_view(from_repo, source),
None => source,
};
let dest = self.backend(to_repo)?;
let pairs = vec![crate::replication::RepoPair {
source_name: from_repo.clone(),
source,
dest_name: to_repo.clone(),
dest,
}];
let opts =
crate::replication::ReplicationOptions { execute: *execute, limit: *limit };
let report = crate::replication::replicate_pairs(&pairs, &opts);
Ok(format!(
"examined={} copied={} skipped={} would_copy={} failed={} bytes={}",
report.examined(),
report.copied(),
report.skipped(),
report.would_copy(),
report.failed(),
report.copied_bytes()
))
}
Job::Upload { repo, artifact, staged, expect_sha256, .. } => {
let dest = self.backend(repo)?;
let bytes = std::fs::read(staged)
.with_context(|| format!("read staged upload {}", staged.display()))?;
if !expect_sha256.is_empty() {
let got = crate::sha256_hex(&bytes);
if !got.eq_ignore_ascii_case(expect_sha256) {
let _ = std::fs::remove_file(staged);
anyhow::bail!(
"staged upload digest mismatch: claimed {expect_sha256}, computed {got}"
);
}
}
let id = trait_id(artifact);
dest.put(&id, &bytes)
.map_err(|e| anyhow::anyhow!("upload finalize failed: {e}"))?;
let _ = std::fs::remove_file(staged);
Ok(format!("{} bytes", bytes.len()))
}
#[cfg(feature = "intake")]
Job::Gate { from_repo, to_repo, artifact, requested_by } => {
let engine = self.engine()?;
if !engine.has_gate() {
return Ok("no gates configured".to_string());
}
let source = self.backend(from_repo)?;
let id = trait_id(artifact);
let data = source
.fetch(&id)
.map_err(|e| anyhow::anyhow!("gate source fetch failed: {e}"))?
.ok_or_else(|| {
anyhow::anyhow!(
"artifact '{}' not found in source repo '{from_repo}'",
artifact.label()
)
})?;
let ecosystem = crate::scanner::ecosystem_for_format(source.format());
let digest = crate::attest::ArtifactDigest::of_sha256(&data);
let attestations: Vec<crate::attest::Statement> = Vec::new();
let set = [crate::promote::PromotedArtifact {
id: &id,
ecosystem: &ecosystem,
digest,
bytes: &data,
attestations: &attestations,
}];
let decision = engine
.gate_check_hop(from_repo, to_repo, &set, requested_by)
.map_err(|e| anyhow::anyhow!("{e}"))?;
Ok(decision.summary())
}
Job::Fetch { repo, artifact, requested_by } => {
let backend = self.backend(repo)?;
let id = trait_id(artifact);
// The walk the request thread used to do inline, now here. It
// writes the bytes through into the cache (the proxy's cache-fill
// gate still decides), and its upstream bytes are charged to the
// consumer that asked (`Cost.upstream_fetch`).
let got = crate::pull_through::charged_to(requested_by, || backend.fetch(&id))
.with_context(|| format!("pull-through of {repo}:{}", artifact.label()))?;
match got {
Some(bytes) => Ok(format!("{} bytes", bytes.len())),
None => Err(anyhow::anyhow!(JobPermanent(format!(
"no upstream of '{repo}' publishes {}",
artifact.label()
)))),
}
}
Job::MirrorAsset { source, from_repo, to_repo, coord, .. } => {
let mirror = self.mirror.as_ref().ok_or_else(|| {
anyhow::anyhow!(JobPermanent(format!(
"mirror asset {coord}: this server runs no mirror transport"
)))
})?;
let id = crate::mirror_cycle::coord_to_id(coord).ok_or_else(|| {
anyhow::anyhow!(JobPermanent(format!("'{coord}' is not a holger coordinate")))
})?;
let dest = self.backend(to_repo)?;
let bytes = mirror
.fetch_asset(source, from_repo, coord)
.with_context(|| format!("mirror {source}:{from_repo}: fetch {coord}"))?
.ok_or_else(|| {
anyhow::anyhow!(JobPermanent(format!(
"mirror {source}:{from_repo}: the upstream no longer has {coord}"
)))
})?;
dest.put(&id, &bytes)
.map_err(|e| anyhow::anyhow!("mirror {source} -> {to_repo}: put {coord}: {e}"))?;
Ok(format!("{} bytes", bytes.len()))
}
#[cfg(feature = "bundle-import")]
Job::ImportInbox { kind } => {
let door = self.imports.as_ref().ok_or_else(|| {
anyhow::anyhow!(JobPermanent(format!("{}: this server runs no import door", kind.inbox_name())))
})?;
Ok(door.sweep(*kind)?.summary())
}
#[cfg(not(feature = "import-door"))]
Job::IntelRefresh => Err(anyhow::anyhow!(JobPermanent("intel.refresh: this build has no import door".into()))),
#[cfg(feature = "import-door")]
Job::IntelRefresh => {
let door = self.imports.as_ref().ok_or_else(|| {
anyhow::anyhow!(JobPermanent("intel.refresh: this server runs no import door".into()))
})?;
#[cfg(feature = "intel-refresh")]
let net = holger_intel::IntelNet::http();
#[cfg(not(feature = "intel-refresh"))]
let net = holger_intel::IntelNet::offline();
let r = door.refresh_intel(&net).map_err(|e| anyhow::anyhow!(JobPermanent(e.tagged())))?;
Ok(format!("intel.refresh: ({}) {} seq {} dated {}: {} advisories", r.outcome, r.bundle_id, r.sequence, r.snapshot_at, r.advisories))
}
#[cfg(feature = "intake")]
Job::PullIntake { source, want } => {
let chain = self.intake.as_ref().ok_or_else(|| {
anyhow::anyhow!(JobPermanent(format!("pull-through {source}: this server runs no intake chain")))
})?;
let pull = chain.pull().ok_or_else(|| {
anyhow::anyhow!(JobPermanent(format!("pull-through {source}: the intake chain has no pull-through")))
})?;
pull.run(chain, source, want)
}
Job::SyncTick => {
#[cfg(feature = "sync")]
{
let hub = self.intake.as_ref().and_then(|c| c.sync()).ok_or_else(|| {
anyhow::anyhow!(JobPermanent("sync.tick: this server declares no `sync` block".into()))
})?;
#[cfg(all(feature = "online", feature = "fetch"))]
let link = crate::sync::HttpLink::new();
#[cfg(all(feature = "online", feature = "fetch"))]
let link: Option<&dyn crate::sync::SyncLink> = Some(&link);
#[cfg(not(all(feature = "online", feature = "fetch")))]
let link: Option<&dyn crate::sync::SyncLink> = None;
Ok(hub.tick(link))
}
#[cfg(not(feature = "sync"))]
Err(anyhow::anyhow!(JobPermanent("sync.tick: this build has no sync protocol (`sync`)".into())))
}
Job::ExportPush => {
#[cfg(feature = "export-push")]
{
let push = crate::export_push::live().ok_or_else(|| anyhow::anyhow!(JobPermanent("export.push: this server declares no `export:` block".into())))?;
let line = push.push_once()?.line();
log::info!("{line}");
Ok(line)
}
#[cfg(not(feature = "export-push"))]
Err(anyhow::anyhow!(JobPermanent("export.push: this build has no push (`export-push`)".into())))
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::AtomicUsize;
struct Counting {
ran: Arc<AtomicUsize>,
}
impl JobRunner for Counting {
fn run(&self, job: &Job) -> Result<String> {
self.ran.fetch_add(1, Ordering::SeqCst);
match job {
Job::Replicate { limit, .. } if *limit == 999 => {
anyhow::bail!("refused on purpose")
}
_ => Ok("ok".to_string()),
}
}
}
fn promote_job(name: &str) -> Job {
Job::Promote {
from_repo: "dev".into(),
to_repo: "prod".into(),
artifact: JobArtifact {
namespace: String::new(),
name: name.into(),
version: "1.0.0".into(),
},
requested_by: "tester".into(),
}
}
fn replicate_job() -> Job {
Job::Replicate {
from_repo: "dev".into(),
to_repo: "prod".into(),
execute: true,
limit: 0,
requested_by: "tester".into(),
}
}
/// ★ **A twin runs no jobs; it only follows** (stinky SB #34). A queue that
/// follows a TWIN's posture starts no worker and admits no job, refused by
/// name (`holger-twin-20`); a primary elected a twin LIVE stands its workers
/// down, and a growth's reacquire does not start them again.
#[test]
fn a_twin_starts_no_job_workers_and_admits_no_job() {
use crate::twin::{NodeRole, TwinPosture};
let twin = NodeRole::Twin {
primary: "10.0.0.1:50051".into(),
poll_seconds: 60,
};
let ran = Arc::new(AtomicUsize::new(0));
// Booted a twin: no pool, no admission.
let tmp = tempfile::tempdir().unwrap();
let queue = Arc::new(JobQueue::open(tmp.path(), 2).unwrap().0);
queue.follow_role(Arc::new(TwinPosture::new(twin.clone())));
queue.start(Arc::new(Counting {
ran: Arc::clone(&ran),
}));
assert!(!queue.is_running(), "a twin starts no job worker");
let refused = queue
.enqueue(replicate_job())
.expect_err("a twin admits no job");
assert!(
refused.to_string().contains("holger-twin-20")
&& refused.to_string().contains("10.0.0.1:50051"),
"{refused:#}"
);
// A primary runs its jobs, then is elected a twin live: the pool stands down.
let tmp = tempfile::tempdir().unwrap();
let queue = Arc::new(JobQueue::open(tmp.path(), 2).unwrap().0);
let posture = Arc::new(TwinPosture::new(NodeRole::Primary));
queue.follow_role(Arc::clone(&posture));
queue.start(Arc::new(Counting {
ran: Arc::clone(&ran),
}));
assert!(queue.is_running(), "a primary runs its workers");
let id = queue.enqueue(replicate_job()).unwrap();
queue
.await_completion(&id, Duration::from_secs(10))
.unwrap()
.expect("a primary's job settles");
// The posture alone — the road the pairing loop takes — stands the pool down.
posture.become_twin(twin);
let t0 = Instant::now();
while queue.is_running() && t0.elapsed() < Duration::from_secs(10) {
std::thread::sleep(Duration::from_millis(10));
}
assert!(
!queue.is_running(),
"a primary elected a twin stops its workers"
);
queue.reacquire().unwrap();
assert!(
!queue.is_running(),
"a growth's reacquire starts no worker on a twin"
);
assert!(
queue.enqueue(promote_job("late")).is_err(),
"and admits nothing"
);
}
/// § THE HEADLINE: the RPC's answer is the id, and it is available before
/// any work has run — the queue is opened without a runner, so nothing CAN
/// have sealed yet.
#[test]
fn a_promotion_returns_a_job_id_before_the_seal_is_written() {
let tmp = tempfile::tempdir().unwrap();
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
let id = queue.enqueue(promote_job("alpha")).unwrap();
assert!(!id.as_str().is_empty(), "an id is returned");
let view = queue.get(&id).unwrap().expect("the row is durable at once");
assert_eq!(view.status, status::QUEUED, "admitted, not yet run");
assert_eq!(view.result_ref, "", "nothing sealed yet");
}
/// § A job that was in flight when the process died is picked up by the next
/// process and finishes — durability, not merely bookkeeping.
#[test]
fn a_job_survives_a_restart_and_finishes() {
let tmp = tempfile::tempdir().unwrap();
let id = {
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
queue.enqueue(replicate_job()).unwrap()
// dropped: the pool never ran, exactly like a crash before dispatch
};
let ran = Arc::new(AtomicUsize::new(0));
let (queue, summary) = JobQueue::open(tmp.path(), 2).unwrap();
assert_eq!(summary.resumed, 1, "the replication was resumed");
queue.start(Arc::new(Counting {
ran: Arc::clone(&ran),
}));
let view = queue
.await_completion(&id, Duration::from_secs(10))
.unwrap()
.expect("the resumed job settles");
assert_eq!(view.status, status::DONE, "it finished after the restart");
assert_eq!(ran.load(Ordering::SeqCst), 1, "the runner ran it once");
queue.shutdown();
}
/// § No `running` row outlives the process that wrote it. The sweep either
/// resumes or interrupts, never leaves.
#[test]
fn a_running_job_does_not_survive_a_restart_as_running() {
let tmp = tempfile::tempdir().unwrap();
let (promote_id, replicate_id) = {
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
let p = queue.enqueue(promote_job("beta")).unwrap();
let r = queue.enqueue(replicate_job()).unwrap();
// Force both rows to `running`, which is what a crash mid-work leaves.
for id in [&p, &r] {
let mut row = queue
.ledger()
.unwrap()
.list(&jera::jobs::JobSelector::All)
.unwrap()
.into_iter()
.find(|row| row.job_id == id.0)
.unwrap();
row.status = status::RUNNING.to_string();
queue.ledger().unwrap().submit(&row).unwrap();
}
(p, r)
};
let (queue, summary) = JobQueue::open(tmp.path(), 2).unwrap();
assert_eq!(summary.interrupted, 1, "the promotion was interrupted");
assert_eq!(summary.resumed, 1, "the replication was resumed");
for id in [&promote_id, &replicate_id] {
let view = queue.get(id).unwrap().unwrap();
assert_ne!(
view.status,
status::RUNNING,
"job {id} is no longer running"
);
}
}
/// § Idempotency by content id: the same coordinate admitted twice is ONE
/// job, so two racing clients cannot both seal it.
#[test]
fn two_jobs_for_the_same_coordinate_do_not_both_seal() {
let tmp = tempfile::tempdir().unwrap();
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
let first = queue.enqueue(promote_job("gamma")).unwrap();
let second = queue.enqueue(promote_job("gamma")).unwrap();
assert_eq!(first, second, "the second caller gets the FIRST job's id");
assert_eq!(queue.list(0).unwrap().len(), 1, "one row, not two");
// A different coordinate is a different job.
let other = queue.enqueue(promote_job("delta")).unwrap();
assert_ne!(first, other);
assert_eq!(queue.list(0).unwrap().len(), 2);
}
/// § An interrupted job is reported BY NAME with a reason, not left running
/// and not laundered into `failed` (which would claim it was tried and refused).
#[test]
fn an_interrupted_job_is_reported_by_name_not_left_running() {
let tmp = tempfile::tempdir().unwrap();
let id = {
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
let id = queue.enqueue(promote_job("epsilon")).unwrap();
let mut row = queue
.ledger()
.unwrap()
.list(&jera::jobs::JobSelector::All)
.unwrap()
.into_iter()
.find(|row| row.job_id == id.0)
.unwrap();
row.status = status::RUNNING.to_string();
queue.ledger().unwrap().submit(&row).unwrap();
id
};
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
let view = queue.get(&id).unwrap().unwrap();
assert_eq!(
view.status,
status::INTERRUPTED,
"named, not `running`, not `failed`"
);
assert!(
view.reason.contains("restarted"),
"the reason says what happened: {}",
view.reason
);
assert!(status::is_terminal(&view.status), "terminal to holger");
assert!(
!jera::jobs::status::is_terminal(&view.status),
"and deliberately NOT terminal to jera's pruner, so it outlives rotation"
);
}
/// § A failing job lands on `failed` with the error chain as its reason.
#[test]
fn a_failing_job_carries_its_reason() {
let tmp = tempfile::tempdir().unwrap();
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
let id = queue
.enqueue(Job::Replicate {
from_repo: "dev".into(),
to_repo: "prod".into(),
execute: true,
limit: 999,
requested_by: "tester".into(),
})
.unwrap();
queue.start(Arc::new(Counting {
ran: Arc::new(AtomicUsize::new(0)),
}));
let view = queue
.await_completion(&id, Duration::from_secs(10))
.unwrap()
.unwrap();
assert_eq!(view.status, status::FAILED);
assert!(
view.reason.contains("refused on purpose"),
"reason: {}",
view.reason
);
queue.shutdown();
}
/// § ROOT LAW #0, as a census over this module's own source — the way the
/// engine guards in skade (`rayon_free_law.rs`) do it: the pool is gatling,
/// and rayon is not named here at all.
#[test]
fn the_worker_pool_is_gatling_not_rayon() {
// The census reads the SHIPPED half of this module — everything above
// `#[cfg(test)]`. Censusing the tests too would count this very
// assertion's own needle as a hit, which is how a census lies.
let whole = include_str!("jobs.rs");
let cut = whole.find("#[cfg(test)]").expect("the test module marker");
let src = &whole[..cut];
let banned = ["use rayon", concat!("rayon", "::"), "par_iter"];
for line in src.lines() {
let l = line.trim_start();
if l.starts_with("//") {
continue;
}
for needle in banned {
assert!(
!l.contains(needle),
"ROOT LAW #0: the job pool must not reach for rayon — found `{needle}` in: {l}"
);
}
}
assert!(
src.contains("gatling::gatling_forkjoin::gatling_run"),
"the pool fans over gatling_run"
);
assert!(
src.contains("gatling::background::Job::spawn"),
"its supervisor is gatling's sanctioned background job, not a hand-rolled thread"
);
assert!(
!src.contains("std::thread::spawn"),
"no private thread pool beside the engine"
);
}
fn fetch_job(name: &str, who: &str) -> Job {
Job::Fetch {
repo: "cache".into(),
artifact: JobArtifact {
namespace: String::new(),
name: name.into(),
version: "1".into(),
},
requested_by: who.into(),
}
}
struct Recording {
order: Arc<Mutex<Vec<String>>>,
}
impl JobRunner for Recording {
fn run(&self, job: &Job) -> Result<String> {
self.order.lock().unwrap().push(job.target());
Ok("ok".into())
}
}
/// § Class Queued/Reduced consumers wait behind Full ones: with one worker,
/// the Full fetch admitted LAST runs FIRST; FIFO within a class.
#[test]
fn a_full_fetch_runs_before_reduced_and_queued_ones() {
let tmp = tempfile::tempdir().unwrap();
let (queue, _) = JobQueue::open(tmp.path(), 1).unwrap();
queue
.enqueue_with_priority(fetch_job("q1", "anon"), PRIORITY_QUEUED)
.unwrap();
queue
.enqueue_with_priority(fetch_job("r1", "over"), PRIORITY_REDUCED)
.unwrap();
queue
.enqueue_with_priority(fetch_job("r2", "over"), PRIORITY_REDUCED)
.unwrap();
let last = queue
.enqueue_with_priority(fetch_job("f1", "paying"), PRIORITY_FULL)
.unwrap();
let order = Arc::new(Mutex::new(Vec::new()));
queue.start(Arc::new(Recording {
order: Arc::clone(&order),
}));
queue
.await_completion(&last, Duration::from_secs(10))
.unwrap()
.unwrap();
for _ in 0..200 {
if order.lock().unwrap().len() == 4 {
break;
}
std::thread::sleep(Duration::from_millis(10));
}
let order = order.lock().unwrap().clone();
assert_eq!(
order,
vec!["cache:f1:1", "cache:r1:1", "cache:r2:1", "cache:q1:1"]
);
queue.shutdown();
}
/// § A second, more urgent request for an open fetch JOINS it (same id)
/// and raises its priority.
#[test]
fn a_full_caller_joining_a_floor_fetch_raises_it() {
let tmp = tempfile::tempdir().unwrap();
let (queue, _) = JobQueue::open(tmp.path(), 1).unwrap();
let a = queue
.enqueue_with_priority(fetch_job("x", "anon"), PRIORITY_FLOOR)
.unwrap();
let b = queue
.enqueue_with_priority(fetch_job("x", "paying"), PRIORITY_FULL)
.unwrap();
assert_eq!(a, b);
let st = queue.shared.state.lock().unwrap();
assert_eq!(
st.pending.iter().find(|p| p.id == a).unwrap().priority,
PRIORITY_FULL
);
}
#[derive(Default)]
struct MemRepo {
blobs: Mutex<HashMap<String, Vec<u8>>>,
}
fn mem_key(id: &traits::ArtifactId) -> String {
format!("{:?}/{}@{}", id.namespace, id.name, id.version)
}
impl traits::RepositoryBackendTrait for MemRepo {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
fn name(&self) -> &str {
"mirrored"
}
fn format(&self) -> traits::ArtifactFormat {
traits::ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, id: &traits::ArtifactId) -> Result<Option<traits::Bytes>> {
Ok(self
.blobs
.lock()
.unwrap()
.get(&mem_key(id))
.cloned()
.map(traits::Bytes::from))
}
fn put(&self, id: &traits::ArtifactId, data: &[u8]) -> Result<()> {
self.blobs
.lock()
.unwrap()
.insert(mem_key(id), data.to_vec());
Ok(())
}
fn handle_http2_request(
&self,
_: &str,
_: &str,
_: &[u8],
) -> Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], vec![]))
}
}
/// An upstream whose first fetch fails, and every later one answers.
#[derive(Clone)]
struct FlakyMirror {
calls: Arc<AtomicUsize>,
}
#[async_trait::async_trait]
impl crate::mirror_cycle::MirrorUpstream for FlakyMirror {
async fn list(&self, _: &str, _: u32) -> Result<Vec<traits::RemoteAsset>> {
Ok(Vec::new())
}
async fn fetch(&self, _: &str, id: &traits::ArtifactId) -> Result<Option<traits::Bytes>> {
if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
anyhow::bail!("upstream holger reset the stream");
}
Ok(Some(traits::Bytes::from(
format!("bytes of {}", id.name).into_bytes(),
)))
}
}
#[async_trait::async_trait]
impl crate::mirror_cycle::UpstreamConnector for FlakyMirror {
async fn connect(
&self,
_: &crate::mirror_cycle::MirrorSource,
) -> Result<Box<dyn crate::mirror_cycle::MirrorUpstream>> {
Ok(Box::new(self.clone()))
}
}
/// § A mirror asset that failed gets its OWN job: failed with its reason,
/// retried, then done — and the bytes are in the local repository.
#[test]
fn a_mirror_asset_job_retries_then_lands() {
let tmp = tempfile::tempdir().unwrap();
let local: Arc<dyn traits::RepositoryBackendTrait> = Arc::new(MemRepo::default());
let routes = Arc::new(crate::exposed::fast_routes::FastRoutes::new(vec![(
"mirrored".to_string(),
Arc::clone(&local),
)]));
let (queue, _) = JobQueue::open(tmp.path(), 2).unwrap();
let queue = queue.with_retry_base(Duration::from_millis(20));
let mirror = FlakyMirror {
calls: Arc::new(AtomicUsize::new(0)),
};
// The job's road is the mirror's own: the declared source, through a
// connector, on the server's runtime.
let rt = tokio::runtime::Runtime::new().unwrap();
let source: crate::mirror_cycle::MirrorSource =
ron::from_str(r#"( ron_name: "cloud", ron_kind: Holger, ron_endpoint: "http://holger.rs:50051", ron_repositories: [] )"#)
.expect("a source declaration");
let transport = crate::mirror_cycle::MirrorJobs::with_connector(
vec![source],
rt.handle().clone(),
Arc::new(mirror.clone()),
);
queue.start(Arc::new(
ServerJobRunner::new(routes, None).with_mirror(Arc::new(transport)),
));
let coord = "docs/readme.txt@1";
let id = queue
.enqueue(Job::MirrorAsset {
source: "cloud".into(),
from_repo: "bundles".into(),
to_repo: "mirrored".into(),
coord: coord.into(),
requested_by: "mirror:cloud".into(),
})
.unwrap();
let v = queue
.await_completion(&id, Duration::from_secs(10))
.unwrap()
.unwrap();
assert_eq!(v.status, status::DONE, "{v:?}");
assert_eq!(v.attempts, 2);
assert_eq!(mirror.calls.load(Ordering::SeqCst), 2);
let back = crate::mirror_cycle::coord_to_id(coord).unwrap();
assert!(local.fetch(&back).unwrap().is_some(), "the asset landed");
queue.shutdown();
}
/// § A retry that was WAITING when the process died keeps waiting, with
/// its attempt count, after the restart — and `retry_delay` doubles, capped.
#[test]
fn retry_delay_doubles_and_is_capped() {
let b = Duration::from_secs(2);
assert_eq!(retry_delay(b, 1), Duration::from_secs(2));
assert_eq!(retry_delay(b, 2), Duration::from_secs(4));
assert_eq!(retry_delay(b, 3), Duration::from_secs(8));
assert_eq!(retry_delay(b, 30), RETRY_CAP);
}
}