git_policy_cli/

work_item.rs

//! What:
//!  The durable work-item record: a small file in the Git directory that says "this landed object needs this
//!  asynchronous work", written by the wrapper and drained by a separate resident consumer.
//! Why:
//!  A wrapper process exits when its command ends, and it may start no program, so work that must not delay a landing
//!  (the standards review, later auto-push and maintenance) is left as a record instead. The record is generic: a kind,
//!  the landed object and an id derived from both, so a retry cannot duplicate it. The design, its failure cases and
//!  the consumer contract are in `doc/planning/cli-git-standards-review-port.md`. Nothing calls this yet.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! await enqueue(workDir, { kind: 'review', oid, ... }); const item = await claimNext(workDir); await finish(workDir, item);
//! ```

use super::claim_time::ClaimTime;
use super::private_storage::{create_private_directory_all, write_private_file_unsynced};
use super::random_id::random_uuid;
use serde_json::{Map, Value};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime};

/// The record format version.
pub const VERSION: u64 = 1;

/// Directory under the worktree's own Git directory (`WorkItem::git_dir`) that holds the state directories.
pub const WORK_DIRECTORY: &str = "cli-git/work";

/// The state directories, in the order an item moves through them.
pub const PENDING: &str = "pending";
/// Held by one consumer, with a lease.
pub const CLAIMED: &str = "claimed";
/// Held by one instance that is returning an expired claim to `pending`; see `requeue_with`.
pub const RECOVERING: &str = "recovering";
/// Finished.
pub const DONE: &str = "done";
/// Given up after too many tries.
pub const FAILED: &str = "failed";

/// What:
///  The kind of work an item asks for.
/// Why:
///   One record format carries every asynchronous job; the kind picks the consumer's handler.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type WorkKind = 'review' | 'push' | 'maintenance';
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WorkKind {
    /// The standards-review shadow run.
    Review,
    /// An auto-push of the branch.
    Push,
    /// Repository maintenance.
    Maintenance,
}

/// Methods over a kind.
impl WorkKind {
    /// The kind's name, which is also the start of an item's id.
    pub fn name(self) -> &'static str {
        return match self {
            WorkKind::Review => "review",
            WorkKind::Push => "push",
            WorkKind::Maintenance => "maintenance",
        };
    }

    /// The kind a name spells.
    pub fn from_name(name: &str) -> Option<WorkKind> {
        return [WorkKind::Review, WorkKind::Push, WorkKind::Maintenance].into_iter().find(|kind| return kind.name() == name);
    }
}

/// What:
///  One work item: what to do, to which landed object, and where.
/// Why:
///   The consumer needs only the object name and the places to find the repository; it reads the patch itself.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type WorkItem = { kind; repository: string; gitDir: string; ref: string; oid: string; parent: string; created: string; tries: number };
/// ```
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkItem {
    /// What to do.
    pub kind: WorkKind,
    /// The worktree root.
    pub repository: String,
    /// The Git directory of the worktree that made the commit and whose service instance runs the item: the common directory
    ///  for the main worktree, `<common>/worktrees/<name>` for a linked one. The work records are under it, and
    ///  `is_acceptable` refuses an item whose value is not the directory it is read from.
    pub git_dir: String,
    /// The branch reference that received the commit.
    pub branch_ref: String,
    /// The landed commit.
    pub oid: String,
    /// Its first parent, or empty for a root commit.
    pub parent: String,
    /// `toISOString` text of the enqueue time.
    pub created: String,
    /// Failed or abandoned attempts so far.
    pub tries: u32,
}

/// Methods over an item.
impl WorkItem {
    /// The id and file name: the kind and the landed object, so the same landing always yields the same id.
    pub fn id(&self) -> String {
        return format!("{}-{}", self.kind.name(), self.oid);
    }

    /// The record's one JSON line.
    pub fn encode(&self) -> String {
        let mut fields: Map<String, Value> = Map::new();
        fields.insert(String::from("v"), Value::from(VERSION));
        fields.insert(String::from("tries"), Value::from(self.tries));
        for (name, text) in [
            ("kind", self.kind.name()),
            ("id", self.id().as_str()),
            ("repository", self.repository.as_str()),
            ("gitDir", self.git_dir.as_str()),
            ("ref", self.branch_ref.as_str()),
            ("oid", self.oid.as_str()),
            ("parent", self.parent.as_str()),
            ("created", self.created.as_str()),
        ] {
            fields.insert(String::from(name), Value::from(text));
        }
        return format!("{}\n", Value::Object(fields));
    }

    /// Read a record; any other version, an unknown kind, a missing field or an id that is not derived from the kind and
    /// object is refused with a message.
    pub fn decode(text: &str) -> Result<WorkItem, String> {
        let value: Value = serde_json::from_str(text).map_err(|error| return format!("the record is not JSON: {error}"))?;
        let field = |name: &str| -> Result<String, String> {
            return value.get(name).and_then(Value::as_str).map(String::from).ok_or_else(|| return format!("the record has no text field {name}"));
        };
        if value.get("v").and_then(Value::as_u64) != Some(VERSION) {
            return Err(String::from("the record has an unsupported version"));
        }
        let kind: WorkKind = WorkKind::from_name(field("kind")?.as_str()).ok_or_else(|| return String::from("the record has an unknown kind"))?;
        let tries: u32 = value.get("tries").and_then(Value::as_u64).and_then(|count| return u32::try_from(count).ok()).ok_or_else(|| return String::from("the record has no tries"))?;
        let item: WorkItem = WorkItem {
            kind,
            repository: field("repository")?,
            git_dir: field("gitDir")?,
            branch_ref: field("ref")?,
            oid: field("oid")?,
            parent: field("parent")?,
            created: field("created")?,
            tries,
        };
        if field("id")? != item.id() {
            return Err(String::from("the record's id is not derived from its kind and object"));
        }
        return Ok(item);
    }
}

/// What:
///  Whether an enqueue added the item or found it already known.
/// Why:
///   A landing that is replayed, or a daemon that restarts, enqueues again; the second time must change nothing.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type Enqueued = 'added' | 'known';
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Enqueued {
    /// A new pending item was written.
    Added,
    /// The id already exists in some state directory.
    Known,
}

/// The directory of one state.
pub fn state_directory(work_dir: &Path, state: &str) -> PathBuf {
    return work_dir.join(state);
}

/// What:
///  Write the item as a pending record, whole or absent, unless its id is already in any state.
///  The producer never waits for a consumer and never reads anything back.
/// Why:
///   The id is the idempotency key; the exclusive publish makes two racing producers agree on one winner.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// await writeExclusive(join(workDir, 'pending', item.id), item.encode())
/// ```
pub fn enqueue(work_dir: &Path, item: &WorkItem) -> Result<Enqueued, String> {
    for state in [PENDING, CLAIMED, DONE, FAILED] {
        create_private_directory_all(state_directory(work_dir, state).as_path()).map_err(|error| return format!("creating the {state} directory failed: {error}"))?;
    }
    let id: String = item.id();
    // The states are looked at in the order an item moves through them, so an item that moves on while they are looked at is
    //  found in the next one: looking at `done` first could miss an item that settled after that look and left `claimed`
    //  before the look at it.
    if [CLAIMED, RECOVERING].iter().any(|state| { return entries_oldest_first(work_dir, state).iter().any(|(name, _)| { return id_of_entry(name.as_str()) == id; }); }) {
        return Ok(Enqueued::Known);
    }
    if state_directory(work_dir, DONE).join(id.as_str()).exists() || state_directory(work_dir, FAILED).join(id.as_str()).exists() {
        return Ok(Enqueued::Known);
    }
    return match write_private_file_unsynced(state_directory(work_dir, PENDING).join(id.as_str()).as_path(), item.encode().as_bytes()) {
        Ok(()) => Ok(Enqueued::Added),
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(Enqueued::Known),
        Err(error) => Err(format!("writing the work item {id} failed: {error}")),
    };
}

/// What:
///  A held item: the decoded record, its id, and the name of the file that holds it.
/// Why:
///   Finishing needs the claim's file name, not the id: the name carries the generation of the claim, so a settlement or a
///  renewal reaches only the claim it was made for. The consumer needs the item.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type Claimed = { id: string; name: string; item: WorkItem };
/// ```
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Claimed {
    /// File name in every state directory but `claimed` and `recovering`, where the file name is `name`.
    pub id: String,
    /// File name in `claimed` as of the claim or the last renewal: the id, the claim's generation and the clock reading.
    pub name: String,
    /// The decoded record.
    pub item: WorkItem,
}

/// Separates the parts of a claim's file name; an id is a kind and a hexadecimal object name and holds none.
const NAME_SEPARATOR: char = '~';

/// The most a stray temporary file of a recovery may age before a recovery removes it.
const STRAY_TEMPORARY_AGE: Duration = Duration::from_secs(24 * 60 * 60);

/// The end of the name of a recovery's temporary file.
const TEMPORARY_SUFFIX: &str = ".retry";

/// What:
///  The id an entry of `claimed` or `recovering` is the claim of: the part of its name before the generation.
/// Why:
///   A claim by an earlier release is named by the id alone, so a name with no separator is its own id.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// name.split('~')[0]
/// ```
pub fn id_of_entry(name: &str) -> &str {
    return name.split_once(NAME_SEPARATOR).map_or(name, |(id, _)| { return id; });
}

/// What:
///  The name of the file that holds the claim of `id` of the generation `token` as of the reading `stamp`.
/// Why:
///   The generation is unique to one acquisition and the reading says when the holder last proved it was alive, so the name
///  alone tells every other instance whose claim this is and whether it ran out, and a rename of exactly that name is the only
///  way to take it: a holder that renews changes the name, so a take made from an older look finds nothing to take.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// `${id}~${token}~${stamp.encode()}`
/// ```
pub fn claim_file_name(id: &str, token: &str, stamp: &ClaimTime) -> String {
    return format!("{id}{NAME_SEPARATOR}{token}{NAME_SEPARATOR}{}", stamp.encode());
}

/// The generation and the reading in a claim's file name, or nothing for a name by an earlier release.
fn parse_claim_name(name: &str) -> Option<(&str, ClaimTime)> {
    let (_, rest) = name.split_once(NAME_SEPARATOR)?;
    let (token, stamp) = rest.split_once(NAME_SEPARATOR)?;
    if token.is_empty() || !token.bytes().all(|byte| { return byte.is_ascii_hexdigit(); }) {
        return None;
    }
    return Some((token, ClaimTime::decode(stamp)?));
}

/// A fresh generation: 32 random hexadecimal digits.
fn new_generation() -> Result<String, String> {
    return random_uuid()
        .map(|uuid| { return uuid.replace('-', ""); })
        .map_err(|error| { return format!("making the generation of a claim failed: {}", error.0); });
}

/// The names in a state directory with their modification times, oldest first.
fn entries_oldest_first(work_dir: &Path, state: &str) -> Vec<(String, SystemTime)> {
    let mut found: Vec<(String, SystemTime)> = Vec::new();
    if let Ok(listing) = std::fs::read_dir(state_directory(work_dir, state)) {
        for entry in listing.flatten() {
            let modified: SystemTime = entry.metadata().and_then(|meta| return meta.modified()).unwrap_or(SystemTime::UNIX_EPOCH);
            found.push((entry.file_name().to_string_lossy().into_owned(), modified));
        }
    }
    found.sort_by(|left, right| return left.1.cmp(&right.1).then_with(|| return left.0.cmp(&right.0)));
    return found;
}

/// The path of the file that holds the claim of `id` now, or nothing when it is not claimed or being recovered.
pub fn find_claim(work_dir: &Path, id: &str) -> Option<PathBuf> {
    return [CLAIMED, RECOVERING]
        .into_iter()
        .flat_map(|state| { return entries_oldest_first(work_dir, state).into_iter().map(move |(name, _)| { return (state, name); }); })
        .find(|(_, name)| { return id_of_entry(name.as_str()) == id; })
        .map(|(state, name)| { return state_directory(work_dir, state).join(name); });
}

/// What:
///  Claim the oldest pending item by renaming it into `claimed` under a name that carries a fresh generation and the
///  reading `now`. A record that does not decode is moved to `failed` so it cannot block the queue.
/// Why:
///   A rename is atomic: of two consumers racing for one item, the loser sees the file gone and takes the next one. The
///  generation and the reading are in the name, so the claim is whole from the instant it exists; nothing is written to it.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// await rename(join(workDir, 'pending', id), join(workDir, 'claimed', claimFileName(id, newGeneration(), now)))
/// ```
pub fn claim_next(work_dir: &Path, now: &ClaimTime) -> Result<Option<Claimed>, String> {
    return claim_next_of(work_dir, now, &[WorkKind::Review, WorkKind::Push, WorkKind::Maintenance]);
}

/// What:
///  Claim the oldest pending item of the given kinds, leaving every other kind where it is.
/// Why:
///   Consumers of different kinds share one queue; the kind starts the file name, so a consumer never opens, moves or
///   stamps an item that belongs to another.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// await rename(join(workDir, 'pending', id), join(workDir, 'claimed', name)) // only ids that start with a handled kind
/// ```
pub fn claim_next_of(work_dir: &Path, now: &ClaimTime, kinds: &[WorkKind]) -> Result<Option<Claimed>, String> {
    for (id, _) in entries_oldest_first(work_dir, PENDING) {
        if !kinds.iter().any(|kind| return id.starts_with(format!("{}-", kind.name()).as_str())) {
            continue;
        }
        let name: String = claim_file_name(id.as_str(), new_generation()?.as_str(), now);
        let claimed_path: PathBuf = state_directory(work_dir, CLAIMED).join(name.as_str());
        if std::fs::rename(state_directory(work_dir, PENDING).join(id.as_str()), claimed_path.as_path()).is_err() {
            continue;
        }
        let text: String = std::fs::read_to_string(claimed_path.as_path()).map_err(|error| return format!("reading the claim of {id} failed: {error}"))?;
        match WorkItem::decode(text.as_str()) {
            Ok(item) => return Ok(Some(Claimed { id, name, item })),
            Err(_) => {
                let _ = std::fs::rename(claimed_path.as_path(), state_directory(work_dir, FAILED).join(id.as_str()));
            }
        }
    }
    return Ok(None);
}

/// What:
///  Move a claimed item to `done` (or to `failed`), after the consumer's result is durable.
/// Why:
///   A crash before this rename leaves the item claimed, and its lease runs out; the work is then done again, so a
///   handler must tolerate a repeat. The rename names the claim's generation, so a holder whose claim was taken over and
///  claimed again by another settles nothing of the other's.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// await rename(join(workDir, 'claimed', claimed.name), join(workDir, 'done', claimed.id))
/// ```
pub fn settle(work_dir: &Path, claimed: &Claimed, state: &str) -> Result<(), String> {
    return settle_named(work_dir, claimed.name.as_str(), state);
}

/// `settle` for the claim file `name`, whose id is the part of the name before the generation.
pub fn settle_named(work_dir: &Path, name: &str, state: &str) -> Result<(), String> {
    let id: &str = id_of_entry(name);
    return std::fs::rename(state_directory(work_dir, CLAIMED).join(name), state_directory(work_dir, state).join(id))
        .map_err(|error| return format!("moving {id} to {state} failed: {error}"));
}

/// What:
///  Start the lease of the claim file `name` again at `now`, returning its new name; nothing when the claim is no longer in
///  `claimed` under that name.
/// Why:
///   A holder that is still working must not look abandoned: every consumer of a directory that several consumers drain reads
///   the reading in the name as the proof of life. The renewal is a rename of the exact name the holder last wrote, so a
///  claim that was taken over, settled, or taken over and claimed again by someone else is not found and is never brought
///  back or renewed on the other's behalf; and an instance that looked at the old name and means to take it finds nothing.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// await rename(join(workDir, 'claimed', name), join(workDir, 'claimed', claimFileName(id, token, now)))
/// ```
pub fn renew_claim(work_dir: &Path, name: &str, now: &ClaimTime) -> Option<String> {
    let id: &str = id_of_entry(name);
    // A claim by an earlier release has no generation; its renewal gives it one.
    let token: String = match parse_claim_name(name) {
        Some((token, _)) => token.to_owned(),
        None => new_generation().ok()?,
    };
    let renewed: String = claim_file_name(id, token.as_str(), now);
    let directory: PathBuf = state_directory(work_dir, CLAIMED);
    return std::fs::rename(directory.join(name), directory.join(renewed.as_str())).ok().map(|()| { return renewed; });
}

/// What:
///  Return claims whose lease ran out to `pending` with one more try, or to `failed` at `max_tries`.
///  Returns how many were moved.
/// Why:
///   This is how a consumer killed at any instruction is noticed when nothing says its successor started: by its lease,
///  never by a process id.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// for (const stale of claimed.filter((c) => now.elapsedSince(c.stamp) > lease)) requeue(stale)
/// ```
pub fn requeue_expired(work_dir: &Path, now: &ClaimTime, lease: Duration, max_tries: u32) -> usize {
    return requeue_expired_reporting(work_dir, now, lease, max_tries).moved;
}

/// What:
///  What a recovery pass did: how many claims it moved, and the items it gave up on.
/// Why:
///   An item that reaches `failed` through recovery was never settled by a handler, so nothing recorded a notice for it; the
///  caller needs the items to tell the person.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type Recovered = { moved: number; givenUp: WorkItem[] };
/// ```
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct Recovered {
    /// How many claims this pass moved to `pending` or `failed`.
    pub moved: usize,
    /// The decoded items it moved to `failed` because they reached the try limit.
    pub given_up: Vec<WorkItem>,
}

/// `requeue_expired` that also gives the items it moved to `failed` for want of tries.
pub fn requeue_expired_reporting(work_dir: &Path, now: &ClaimTime, lease: Duration, max_tries: u32) -> Recovered {
    return requeue_with(work_dir, now, max_tries, &|stamp| { return now.elapsed_since(stamp) > lease; });
}

/// What:
///  How long until the first claim that is still held runs out its lease, or nothing when no claim is held.
/// Why:
///   An instance that finds claims it may not take yet has nothing to wake it when they run out, so the service asks when to
///  look again; without this, work whose holder died would wait for the next commit.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// Math.min(...claimed.map((c) => lease - now.elapsedSince(c.stamp)))
/// ```
pub fn next_claim_expiry(work_dir: &Path, now: &ClaimTime, lease: Duration) -> Option<Duration> {
    return [CLAIMED, RECOVERING]
        .into_iter()
        .flat_map(|state| { return entries_oldest_first(work_dir, state); })
        .filter(|(name, _)| { return !name.ends_with(TEMPORARY_SUFFIX); })
        .map(|(name, modified)| { return lease.saturating_sub(now.elapsed_since(&stamp_of_entry(name.as_str(), modified))); })
        .min();
}

/// The reading in a claim's file name, or for a claim by an earlier release the wall-clock time of its last stamp.
fn stamp_of_entry(name: &str, modified: SystemTime) -> ClaimTime {
    return parse_claim_name(name).map_or_else(|| { return ClaimTime::from_wall(modified); }, |(_, stamp)| { return stamp; });
}

/// What:
///  Take each claim whose reading `expired` accepts, by renaming the exact file name that was looked at into `recovering`
///  under a generation of its own, then return it to `pending` with one more try, or to `failed` at `max_tries`. Returns how
///  many were moved.
/// Why:
///   The look and the take are separate moments, and in between another instance may have returned the same claim and claimed
///  it again, or its holder may have renewed it. Both change the name, so the take of the old name finds nothing and takes
///  nothing: the rename is the only check and it is atomic. The file in `recovering` is this instance's alone, so its record
///  is rewritten there whole (a temporary file renamed over it) and the item reaches `pending` by one rename; a recovery
///  killed anywhere leaves the item in exactly one place, and an item left in `recovering` runs out its lease and is
///  recovered like a claim, which costs it one more try.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// for (const seen of claimed.filter(expired)) { if (await tryRename(seen.name, owned)) { await rewrite(owned); await rename(owned, pending) } }
/// ```
fn requeue_with(work_dir: &Path, now: &ClaimTime, max_tries: u32, expired: &dyn Fn(&ClaimTime) -> bool) -> Recovered {
    let mut recovered: Recovered = Recovered::default();
    for state in [CLAIMED, RECOVERING] {
        for (name, modified) in entries_oldest_first(work_dir, state) {
            if name.ends_with(TEMPORARY_SUFFIX) {
                remove_stray_temporary(work_dir, state, name.as_str(), modified, now);
                continue;
            }
            if !expired(&stamp_of_entry(name.as_str(), modified)) {
                continue;
            }
            if let Some(given_up) = recover(work_dir, state, name.as_str(), now, max_tries) {
                recovered.moved += 1;
                recovered.given_up.extend(given_up);
            }
        }
    }
    return recovered;
}

/// Removes a recovery's temporary file that a killed recovery left, once it is old enough that nobody can still be using it.
fn remove_stray_temporary(work_dir: &Path, state: &str, name: &str, modified: SystemTime, now: &ClaimTime) {
    if now.elapsed_since(&ClaimTime::from_wall(modified)) > STRAY_TEMPORARY_AGE {
        let _removed = std::fs::remove_file(state_directory(work_dir, state).join(name));
    }
}

/// Returns the claim file `name` of `state` to `pending` or `failed`; nothing when another instance moved it first, otherwise
/// the item when it was given up on (moved to `failed` for want of tries).
fn recover(work_dir: &Path, state: &str, name: &str, now: &ClaimTime, max_tries: u32) -> Option<Option<WorkItem>> {
    let id: &str = id_of_entry(name);
    let Ok(token) = new_generation() else {
        return None;
    };
    let owned_name: String = claim_file_name(id, token.as_str(), now);
    let recovering: PathBuf = state_directory(work_dir, RECOVERING);
    if create_private_directory_all(recovering.as_path()).is_err() {
        return None;
    }
    let owned: PathBuf = recovering.join(owned_name.as_str());
    // The take: only one instance can rename this exact name, and a claim that was renewed or claimed again is not this name.
    if std::fs::rename(state_directory(work_dir, state).join(name), owned.as_path()).is_err() {
        return None;
    }
    // A record that cannot be read after it was taken is settled like one that cannot be decoded: leaving it in `recovering`
    //  would have every later pass take it, fail the same way, and keep the service waiting on it for good.
    let Ok(text) = std::fs::read_to_string(owned.as_path()) else {
        return std::fs::rename(owned.as_path(), state_directory(work_dir, FAILED).join(id)).is_ok().then_some(None);
    };
    let (target, given_up): (&str, Option<WorkItem>) = match WorkItem::decode(text.as_str()) {
        Ok(mut item) => {
            item.tries += 1;
            let rewritten: PathBuf = recovering.join(format!("{owned_name}{TEMPORARY_SUFFIX}"));
            let destination: &str = if item.tries >= max_tries { FAILED } else { PENDING };
            if std::fs::write(rewritten.as_path(), item.encode()).is_err() || std::fs::rename(rewritten.as_path(), owned.as_path()).is_err() {
                return None;
            }
            (destination, (destination == FAILED).then_some(item))
        }
        // The record never changes once it is pending, so this is a record that was never readable.
        Err(_) => (FAILED, None),
    };
    return std::fs::rename(owned.as_path(), state_directory(work_dir, target).join(id)).is_ok().then_some(given_up);
}

/// What:
///  How long the oldest pending item has waited, or `None` when nothing is pending.
/// Why:
///   The wrapper's one-line notice that no consumer is draining the queue is this age past a bound; a process id in a
///   file would be meaningless across process-id namespaces.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// now - oldestPending.mtime
/// ```
pub fn oldest_pending_age(work_dir: &Path, now: SystemTime) -> Option<Duration> {
    return entries_oldest_first(work_dir, PENDING).first().map(|(_, created)| return now.duration_since(*created).unwrap_or(Duration::ZERO));
}

/// Controls stay out of the release executable.
#[cfg(test)]
#[path = "work_item_tests.rs"]
mod tests;