Job queue & leases¶

Architecture

Telha's async work (ingestion parses, embeddings, cascade journals, system maintenance) runs through one internal job queue, stored in the jobs column family and leased out to stateless pollers. This page documents the state machine, the lease-fencing protocol that lets crash-prone workers share work safely without a distributed lock manager, and the reaper that reclaims work from workers that die mid-flight.

Overview & purpose¶

Stateless workers polling shared work need exactly one guarantee: no job is processed by two workers at once, and no job is lost when a worker dies mid job. The module doc in core-engine/src/jobs/mod.rs states the design in one line: every transition is an atomic state-move, one WriteBatch deletes the old state-keyed row and puts the new one. The jobs CF is operational state, not history: the append-only doctrine that governs node_versions and edge_versions does not apply here. The audit trail instead lives inside the job's own value, as an embedded event log.

Two things fall out of that design choice:

• Ready-job discovery is a bounded prefix scan. Because the job's state is the leading byte of its key (composite-key-layout spec), "find pending work" never needs a secondary index or a full-table filter, just a scan from [scope][Pending].
• Fencing, not distributed locking, prevents double-processing. A worker that leases a job increments a generation counter stored in the row. A worker whose lease has been reaped and reassigned discovers this the moment it tries to submit, because its generation number is stale.

Design¶

JobState and the key-order rationale¶

#[repr(u8)]
pub enum JobState {
    Pending = 0x00,   // ready to lease (subject to `not_before` backoff gate)
    Leased = 0x01,    // owned by a worker until `lease_deadline`
    Completed = 0x02, // finished successfully
    Failed = 0x03,    // reserved resting state; v1 retryable failures re-enter Pending directly
    Dead = 0x04,      // dead-lettered after max_attempts (alerting metric)
}

Pending = 0x00 sorts first under RocksDB's ascending byte comparator, so a scan from the state head of a tenant's jobs prefix finds ready work without skipping over completed or dead rows first. Failed is a reserved resting state in the key space; as built, a retryable failure moves the record straight back to Pending with a not_before gate rather than parking at Failed and requiring a second transition.

JobRecord: msgpack value with an embedded audit trail¶

The stored value is [VALUE_VERSION byte][msgpack-encoded JobRecord], the same version-byte convention as every other value in the engine (the version-record-model doctrine). JobRecord carries both the queue's operational state and its own history:

pub struct JobRecord {
    pub job_id: Uuid,
    pub kind: String,              // "ingest:{format}", "embed", "cascade", "system:{task}"
    pub payload: Vec<u8>,          // opaque; validated by consumers
    pub priority: u8,              // lower = sooner; part of the key
    pub created_at: Ts,            // enqueue time, µs; part of the key (FIFO within priority)
    pub state: JobState,           // mirrors the key's state byte
    pub attempts: u32,             // lease/fail cycles consumed
    pub lease_generation: u64,     // fencing token, incremented on every lease acquisition
    pub worker_id: Option<String>,
    pub lease_deadline: Option<Ts>,
    pub not_before: Option<Ts>,    // backoff gate
    pub idempotency_key: Option<String>,
    pub max_attempts: Option<u32>, // per-job ceiling override (PR-060)
    pub continuation: Option<Vec<u8>>, // lease-scoped checkpoint (streaming ingestion)
    pub events: Vec<JobEvent>,     // the audit trail
}

pub struct JobEvent {
    pub at: Ts,
    pub transition: String, // enqueued/leased/extended/completed/failed/reaped/dead/retried/checkpoint/stats
    pub detail: Option<String>,
}

Because history lives in events rather than in a chain of superseded rows, telha jobs show <id> (operator UI, spec §7) can print a job's full lifecycle from a single point lookup.

QueueScope: tenant jobs and the reserved system namespace¶

pub enum QueueScope<'a> {
    Tenant(&'a TenantScope),
    System,
}

QueueScope::System resolves to the reserved nil-tenant prefix (SystemScope, storage-engine spec) for deployment-global work: queue-wide maintenance jobs that are not tied to any tenant. System jobs sort to the absolute head of the jobs CF, the same "fast head scan" property the storage-engine page documents for the embedding-model registry. QueueScope is resolved to key bytes internally by KeyCodec::job_key / KeyCodec::system_job_key; no caller outside the jobs module ever forges a system-scoped key.

Data model¶

Key layout¶

Per the composite-key-layout spec (key_len::JOB, 58 bytes), reproduced here because it drives every access pattern this module implements:

[ scope 32 ][ state 1 ][ priority 1 ][ created_at 8 ][ job_id 16 ]
                                                        total: 58 bytes

created_at is not inverted, unlike every temporal-index key elsewhere in the engine. FIFO within (state, priority) is the wanted order here, not latest-first, so the timestamp is stored big-endian and ascending.

Job kinds (v1)¶

The kind string is also the lease filter: lease_next(scope, kinds, worker_id) accepts an optional list of kind prefixes, so a worker fleet that only handles ingest:* never leases an embed job.

Algorithms & invariants¶

enqueue and lease_next¶

enqueue writes one Pending row with attempts = 0, lease_generation = 0, and a single enqueued event. lease_next is the atomic move at the center of the module:

for row in scan_prefix(cf::JOBS, &scope.state_range(JobState::Pending)) {
    let mut record = decode(&value)?;
    if record.not_before.is_some_and(|nb| nb > now) { continue; }       // backoff gate
    if kinds.is_some_and(|ks| !ks.iter().any(|k| record.kind.starts_with(k))) { continue; }
    record.state = JobState::Leased;
    record.lease_generation += 1;
    record.worker_id = Some(worker_id);
    record.lease_deadline = Some(now + lease_ttl_secs * 1_000_000);
    batch_write(vec![Delete(old_key), Put(new_leased_key, record)]);   // one WriteBatch
    return Ok(Some(record));
}

Because the scan walks the Pending prefix in key order (created_at ascending within priority ascending), this is FIFO within priority: two jobs enqueued at the same priority lease out in enqueue order, and a lower-priority row is never leased ahead of an eligible higher-priority one. The not_before check is the backoff gate: a job that failed recently and is waiting out its exponential backoff is skipped even though it is structurally Pending, and scanning continues past it to the next eligible row.

An in-process Mutex<()> serializes lease acquisition (the same single-writer posture the temporal write paths use), so the atomic delete+put pair is never interleaved with another lease_next call inside one process. Across processes, correctness rests on batch_write being an atomic RocksDB write, not on the mutex.

Lease-generation fencing¶

Every fenced operation (complete, fail, checkpoint) requires the caller to supply the worker_id and lease_generation it was granted at lease time. If either has drifted from the stored record, the call returns JobError::LeaseConflict instead of applying the transition:

if record.worker_id.as_deref() != Some(worker_id) {
    return Err(JobError::LeaseConflict { job: job_id, reason: format!(
        "leased by {:?}, submitted by {worker_id:?}", record.worker_id) });
}
if record.lease_generation != lease_generation {
    return Err(JobError::LeaseConflict { job: job_id, reason: format!(
        "lease generation {lease_generation} superseded by {}", record.lease_generation) });
}

This is the minimal fencing-token scheme the spec's Alternatives Considered section adopts explicitly in place of full fencing machinery (§9): it covers exactly the race the spec calls out, a stale worker whose job was reaped and re-leased submitting late. JobError::LeaseConflict is the type that maps to gRPC ABORTED per the spec's API contract; the late submission is discarded and the re-leased attempt owns the job.

The reaper¶

stateDiagram-v2
    [*] --> Pending: enqueue
    Pending --> Leased: lease_next (atomic move,\nnot_before gate, kind filter)
    Leased --> Leased: extend (Heartbeat,\nfencing checked)
    Leased --> Completed: complete (fenced)
    Leased --> Pending: fail, retryable\n(backoff not_before set)
    Leased --> Dead: fail, not retryable\nOR attempts >= max_attempts
    Leased --> Pending: reaper reclaims\n(lease_deadline expired,\nattempts < max_attempts)
    Leased --> Dead: reaper reclaims\n(attempts >= max_attempts)
    Dead --> Pending: operator retry\n(attempts reset to 0)
    Completed --> [*]
    Dead --> [*]

spawn_reaper runs a tokio::interval tick (default every 30s, jobs.reaper_interval_secs) that calls reap_expired, which scans ScanRange::system_full_cf(), a documented exception to per-tenant confinement (storage-engine spec) reserved for exactly this: the reaper has no tenant registry, so it sweeps every namespace, tenant and system, in one pass. For every Leased row whose lease_deadline <= now, it increments attempts, clears worker_id/lease_deadline, and either:

• moves the job back to Pending with a fresh not_before backoff gate and a reaped event, if attempts < max_attempts; or
• moves it to Dead with a dead event and a tracing::warn!, if the retry budget is exhausted.

Because the reaper has no TenantScope in hand for a row it finds via the full-CF sweep (and must not forge one), it rebuilds the destination key by patching the state byte of the original key bytes (rekey) rather than re-deriving the key through KeyCodec.

Backoff schedule and dead-lettering¶

fn backoff_us(&self, kind: &str, attempts: u32) -> Ts {
    let (base, cap) = match self.config.backoff_base_secs_by_kind.get(kind) {
        Some(&base) => (base, base.max(self.config.backoff_cap_secs)),
        None => (self.config.backoff_base_secs, self.config.backoff_cap_secs),
    };
    let secs = base.saturating_mul(self.config.backoff_factor.saturating_pow(attempts - 1)).min(cap);
    secs * 1_000_000
}

With the defaults (base 10s, factor 4, cap 3,600s), the schedule is 10s, 40s, 160s, 640s, then capped at 3,600s (1h) from the fifth attempt on. A per-kind override in backoff_base_secs_by_kind (PR-060, populated internally rather than as an env-layer key) supplies both the base and a floor for the cap: clarify:temporal uses the escalation window as its base precisely so its retries pace at the escalation window rather than the generic exponential schedule, even though that base exceeds the global cap.

A job dead-letters (moves to Dead) the moment attempts >= max_attempts, where max_attempts is the job's own max_attempts override if set, otherwise jobs.max_attempts (default 5). enqueue_with_max_attempts sets that per-job ceiling; the spec's PR-060 addition documents its motivating use case as clarification jobs, which budget one attempt per ranked candidate plus one steward-escalation attempt, a ceiling that does not fit the global default.

Streaming continuation checkpoint¶

JobRecord.continuation is lease-scoped, opaque storage the queue itself never interprets. The ingestion subsystem uses it to persist an IngestContinuation (expected seq, pinned source version, blob id, accumulated length) between partial submissions of a streamed parse. See Ingestion & provenance for what is stored there. Two invariants the queue enforces on the queue's side of that contract:

• checkpoint is fenced exactly like complete/fail (worker id + lease generation must match), and it also extends the lease in the same write, so a worker mid-stream never has to choose between checkpointing and heartbeating.
• continuation is cleared whenever a lease is newly acquired (lease_next) and whenever a job leaves the leased state (complete, fail, reaper reclaim), so a re-leased job always starts from a clean slate: a worker that inherits a reaped job never sees a stale checkpoint from the previous attempt.

Configuration¶

Config validation rejects lease_ttl_secs == 0 or max_attempts == 0 at load (TelhaConfig::validate, core-engine/src/config.rs).

As-built notes¶

From spec §14 (Changelog), reconciled against core-engine/src/jobs/mod.rs:

• v0.2 (2026-07-05, PR-060) landed exactly as specified. JobRecord.max_attempts: Option<u32> and enqueue_with_max_attempts exist for the per-job ceiling override; backoff_base_secs_by_kind exists as an internal (non-env) config map, consulted by both the fail path (finish_at) and the reaper (reap_expired_at), both routed through the shared backoff_us helper.
• The Failed state is reserved but not reachable in v1, as the spec's §3 prose itself notes ("v1 retryable failures re-enter Pending directly"). The discriminant Failed = 0x03 exists in the key-space value range for forward compatibility but no code path in jobs/mod.rs ever sets state = JobState::Failed.
• System-namespace cross-tenant operations (lease_next_any, extend_any, find_leased_any, reap_expired) all reuse ScanRange::system_full_cf(), the documented storage-engine exception to per-tenant confinement, reserved specifically for the job-queue reaper and the worker fleet's cross-tenant poll (grpc-contracts spec §4: workers are internal infrastructure, not tenant-scoped callers).
• Two job kinds beyond the spec's original four appear in the code and its own doc comments: clarify:temporal (PR-060, temporal-clarification spec) and sync:{source} (PR-065, connector chains, via enqueue_chained). Both are additive; neither changes the state machine or key layout.
• Contention and recovery tests (spec §12) are present as unit tests in jobs/mod.rs: lease_contention_never_double_leases (8 threads, 400 jobs, asserts every job is leased exactly once), and complete_happy_path_and_stale_generation_rejected (kill-then-reap-then- stale-submit sequence, asserting the stale submit is rejected via LeaseConflict while the live lease completes).

Related¶

• Storage engine & key layout · the jobs CF, SystemScope, and system_full_cf.
• Ingestion & provenance · the streaming continuation this queue carries.
• Tombstone cascade · the cascade journal that rides this queue.