refactor: trim job framework comments

Author: St4NNi

core/src/structs/job.rs

@@ -11,9 +11,7 @@ use crate::errors::ConversionError;
 use crate::structs::invert_timestamp_ms;
 use crate::types::{Key, UserId};
 
-/// Version prefix inside the `jobs` keyspace. A single decode chokepoint plus this
-/// prefix keep the record wrappable in a version envelope later (#286) as a
-/// one-site change.
+/// Version prefix keeping the record wrappable in a version envelope later (#286).
 pub const JOB_RECORD_KEY_PREFIX: &[u8] = b"jobs-v1/";
 
 pub const JOB_DUE_INDEX_PREFIX: &[u8] = b"due/";
@@ -107,10 +105,8 @@ impl JobState {
 /// `DocumentSyncOutboxEvent`. Additive-only until a version envelope lands (#286).
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum JobPayload {
-    /// Test-only executor exercising the substrate before #256 ships a real
-    /// payload. Idempotency key: the `cleanup_marker` file path — a re-driven or
-    /// cancelled Probe removes that file, so partial state is always reconciled by
-    /// re-running from scratch.
+    /// Test-only executor. Idempotency key: the `cleanup_marker` file, which a
+    /// re-driven or cancelled Probe removes so re-running from scratch is safe.
     Probe {
         steps: u32,
         step_sleep_ms: u64,
@@ -212,8 +208,7 @@ impl JobProgress {
     }
 }
 
-/// Lease held by the executor currently owning the job. `claim_token` fences
-/// zombie executors: every executor-side write requires it to match.
+/// Lease on the job; `claim_token` fences zombie executors on every write.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct JobClaim {
     pub holder_node_id: NodeId,

operations/src/jobs/drain.rs

@@ -27,8 +27,7 @@ pub struct JobDrainResult {
     pub next_due_after: Option<Duration>,
 }
 
-/// Claim up to `capacity` due jobs and re-queue expired leases. Never runs a job
-/// inline: claimed records are returned for the caller to hand to the runtime.
+/// Claim up to `capacity` due jobs and re-queue expired leases; claimed records are returned.
 pub async fn process_job_queue_batch(
     storage: &StorageHandle,
     holder_node_id: NodeId,
@@ -77,8 +76,7 @@ pub async fn process_job_queue_batch(
     Ok(result)
 }
 
-/// Earliest `due/` or `lease/` head as a delay from now; `ZERO` when work is
-/// already due.
+/// Earliest `due/`/`lease/` head as a delay from now; `ZERO` when work is due.
 pub async fn next_job_drain_timer_after(
     storage: &StorageHandle,
 ) -> Result<Option<Duration>, String> {
@@ -94,8 +92,7 @@ pub async fn next_job_drain_timer_after(
     Ok(next.map(|ts| Duration::from_millis(ts.saturating_sub(now_ms))))
 }
 
-/// ShortenTimer restore run at startup and inside the durable-queue re-arm loop, so
-/// it must never push an existing deadline later.
+/// ShortenTimer restore (startup + re-arm loop); never pushes a deadline later.
 pub async fn restore_job_queue_timer(storage: &StorageHandle, task_handle: &TaskHandle) {
     let after = match next_job_drain_timer_after(storage).await {
         Ok(Some(after)) => after,

operations/src/jobs/mod.rs

@@ -8,18 +8,15 @@ pub mod service;
 pub mod store;
 pub mod submit;
 
-/// Lease held by a claiming executor. A claim is fenced by its `claim_token`; a
-/// zombie executor whose lease lapsed and whose job was re-claimed has every write
-/// rejected.
+/// Claim lease; fenced by `claim_token` so a lapsed holder's writes are rejected.
 pub const JOB_LEASE_MS: u64 = 60_000;
-/// Executor heartbeat interval. Each renew piggybacks a progress flush; must stay
-/// well under `JOB_LEASE_MS`.
+/// Heartbeat interval (renew + progress flush); must stay well under the lease.
 pub const JOB_HEARTBEAT_MS: u64 = 20_000;
 /// Maximum jobs executing locally at once; excess stays queued.
 pub const JOB_CONCURRENCY_CAP: usize = 8;
-/// Attempts (inclusive) before a retryable failure becomes terminal `Failed`.
+/// Attempts before a retryable failure becomes terminal `Failed`.
 pub const JOB_MAX_ATTEMPTS: u32 = 5;
-/// Uniform terminal-state retention before a job record is pruned.
+/// Uniform terminal-state retention before pruning.
 pub const JOB_RETENTION_MS: u64 = 7 * 24 * 60 * 60 * 1000;
 /// Minimum spacing between throttled progress flushes.
 pub const JOB_PROGRESS_FLUSH_INTERVAL_MS: u64 = 500;

operations/src/jobs/prune.rs

@@ -95,8 +95,7 @@ pub(crate) async fn process_job_prune_batch_with_page_size(
     })
 }
 
-/// ShortenTimer restore keyed off the earliest `prune/` entry, run at startup and
-/// in the durable-queue re-arm loop.
+/// ShortenTimer restore keyed off the earliest `prune/` entry.
 pub async fn restore_job_prune_timer(storage: &StorageHandle, task_handle: &TaskHandle) {
     let after = match first_schedule_entry(storage, JOB_PRUNE_INDEX_PREFIX).await {
         Ok(Some((expiry_ms, _))) => {

operations/src/jobs/runtime.rs

@@ -36,8 +36,7 @@ struct RunningJob {
     completion: watch::Sender<bool>,
 }
 
-/// Registry of locally executing jobs: the cancellation tokens, the completion
-/// signals backing `wait_for_terminal`, and the concurrency cap.
+/// Registry of locally executing jobs, their cancellation tokens, and the concurrency cap.
 pub struct JobsRuntime {
     running: Mutex<HashMap<JobId, RunningJob>>,
     cap: usize,
@@ -72,8 +71,7 @@ impl JobsRuntime {
         self.cap.saturating_sub(self.running_count())
     }
 
-    /// Poke a locally running job's cancellation token. Returns `false` if the job
-    /// is not running here; the persisted `cancel_requested` flag then drives it.
+    /// Poke a locally running job's cancel token; `false` if it is not running here.
     pub fn request_cancel(&self, job_id: JobId) -> bool {
         match self
             .running
@@ -124,8 +122,7 @@ impl JobsRuntime {
         }
     }
 
-    /// Block until the job is terminal or `timeout` elapses. Uses the local
-    /// completion signal when the job runs here, else polls storage every 250ms.
+    /// Block until the job is terminal or `timeout` elapses (local signal, else 250ms poll).
     pub async fn wait_for_terminal(
         &self,
         storage: &StorageHandle,
@@ -159,8 +156,7 @@ impl JobsRuntime {
         }
     }
 
-    /// Fjall is single-process, so at startup every claimed/running job's holder is
-    /// definitionally dead: re-queue them all immediately.
+    /// At startup every claimed/running holder is definitionally dead: re-queue them all.
     pub async fn recover_stale_jobs(&self, storage: &StorageHandle) -> Result<usize, String> {
         let now_ms = unix_timestamp_millis();
         let mut job_ids = Vec::new();
@@ -462,8 +458,7 @@ mod tests {
         std::fs::write(&marker, b"partial").unwrap();
         let mut record = probe_record(job_id, 5, 0, Some(marker.to_str().unwrap().to_string()));
         record.cancel_requested = true;
-        // A prior attempt means partial state is possible, so the job is claimed and
-        // its cleanup hook runs (unlike a never-attempted fresh cancel).
+        // A prior attempt means it is claimed (not fresh-cancelled) so cleanup runs.
         record.attempts = 1;
         let claimed = claim(&storage, record).await;
 

operations/src/jobs/service.rs

@@ -13,8 +13,7 @@ use super::store::{
 use super::submit::schedule_job_drain_effect;
 use crate::driver::DriverContext;
 
-/// API-facing helpers so REST handlers never orchestrate storage or task effects
-/// directly (enforced by the api effect-boundary guard).
+/// API-facing helpers so REST handlers never orchestrate storage/task effects directly.
 pub async fn list_owned_jobs(
     context: &DriverContext,
     user_id: UserId,

operations/src/jobs/store.rs

@@ -23,8 +23,7 @@ use crate::queue_backoff::queue_retry_after_ms;
 type JobWrites = Vec<(KeySpace, Key, Value)>;
 type JobDeletes = Vec<(KeySpace, Key)>;
 
-/// Single decode chokepoint for persisted job records. Wrapping the record in a
-/// version envelope later (#286) is a one-site change here.
+/// Single decode chokepoint; wrappable in a version envelope later (#286).
 pub fn decode_job_record(bytes: &[u8]) -> Result<JobRecord, ConversionError> {
     JobRecord::from_bytes(bytes)
 }
@@ -41,8 +40,7 @@ pub enum JobMutationError {
     Storage(String),
 }
 
-/// The single schedule-index entry for a record follows its state: queued jobs
-/// live under `due/`, claimed/running under `lease/`, terminal under `prune/`.
+/// Schedule-index key by state: queued -> due/, claimed/running -> lease/, terminal -> prune/.
 fn schedule_index_key_for(record: &JobRecord) -> Key {
     match record.state {
         JobState::Queued => job_due_index_key(record.due_at_ms, record.job_id),
@@ -65,9 +63,7 @@ fn empty_value() -> Value {
     ByteView::from(Vec::new())
 }
 
-/// Storage writes for creating a fresh job: record + schedule index + owner index
-/// (+ dedup index when a key is set). This is the composable primitive a producer
-/// can splice into its own transaction; the ≤4 writes of the submit budget.
+/// Writes creating a fresh job (<=4 keys); composable into a producer transaction.
 pub fn job_insert_entries(record: &JobRecord) -> Result<JobWrites, ConversionError> {
     let mut writes = vec![
         (
@@ -96,8 +92,7 @@ pub fn job_insert_entries(record: &JobRecord) -> Result<JobWrites, ConversionErr
     Ok(writes)
 }
 
-/// Record + owner index + prune index deletes for a pruned terminal job. The
-/// dedup entry was already removed when the job first went terminal.
+/// Deletes for a pruned terminal job (its dedup entry is already gone).
 pub fn job_prune_delete_entries(record: &JobRecord) -> JobDeletes {
     vec![
         (JOB_KEYSPACE.to_string(), job_record_key(record.job_id)),
@@ -160,9 +155,7 @@ fn guard_token(record: &JobRecord, token: Ulid) -> Result<(), JobMutationError>
     }
 }
 
-/// Read a job under a write transaction, apply `mutate`, and persist the record
-/// plus its index deltas atomically. State changes are validated by the pure
-/// transition guard so index and record can never disagree.
+/// Read, mutate, and persist a job with its index deltas in one transaction.
 pub async fn mutate_job<F>(
     storage: &StorageHandle,
     job_id: JobId,
@@ -249,8 +242,7 @@ pub enum ClaimOutcome {
     NotEligible,
 }
 
-/// Claim a queued job for `holder_node_id`, or move a never-attempted
-/// cancel-requested job straight to `Cancelled`.
+/// Claim a queued job, or cancel it directly if it was never-attempted and cancel-requested.
 pub async fn claim_job(
     storage: &StorageHandle,
     job_id: JobId,
@@ -414,10 +406,8 @@ pub enum RequeueOutcome {
     Skipped,
 }
 
-/// Return a claimed/running job to the queue with `queue_retry_after_ms` backoff, or
-/// move it to `Failed` once it has exhausted `JOB_MAX_ATTEMPTS`. `token` is `None`
-/// for the lease-expiry sweep and startup recovery, which reclaim whatever holder is
-/// recorded.
+/// Re-queue with backoff, or fail once `JOB_MAX_ATTEMPTS` is spent. `token` is `None`
+/// for the lease sweep and startup recovery.
 pub async fn requeue_job(
     storage: &StorageHandle,
     job_id: JobId,
@@ -461,8 +451,7 @@ pub enum CancelRequestOutcome {
     AlreadyTerminal(JobRecord),
 }
 
-/// Set `cancel_requested` without changing state. Idempotent; a terminal job is a
-/// no-op returning the current record.
+/// Idempotently set `cancel_requested`; a terminal job is a no-op.
 pub async fn set_cancel_requested(
     storage: &StorageHandle,
     job_id: JobId,
@@ -485,8 +474,7 @@ pub async fn set_cancel_requested(
     })
 }
 
-/// Owner-scoped listing, newest first, with an opaque 24-byte cursor and an
-/// optional server-side state filter applied per page.
+/// Owner-scoped listing, newest first, with an opaque cursor and optional state filter.
 pub async fn list_jobs_for_user(
     storage: &StorageHandle,
     user_id: UserId,
@@ -578,8 +566,7 @@ pub async fn iter_prefix_page(
     }
 }
 
-/// Head of a schedule-index prefix, i.e. the earliest `(timestamp, job_id)` still
-/// enqueued under `due/`, `lease/`, or `prune/`.
+/// Earliest `(timestamp, job_id)` under a schedule-index prefix.
 pub async fn first_schedule_entry(
     storage: &StorageHandle,
     prefix: &[u8],

operations/src/jobs/submit.rs

@@ -15,9 +15,7 @@ use tracing::warn;
 
 use super::store::job_insert_entries;
 
-/// Kick the drain so a freshly submitted job is claimed without waiting for the
-/// next poll. `DrainJobQueue` is re-armed from its own keyspace, so this timer is
-/// never persisted.
+/// Kick the drain so a submitted job is claimed promptly; this timer is never persisted.
 pub fn schedule_job_drain_effect() -> Effect {
     Effect::Task(TaskEffect::ResetTimer {
         key: TaskKey::DrainJobQueue,
@@ -61,9 +59,7 @@ enum SubmitState {
     Error,
 }
 
-/// Effect-driven submit. Dedup follows the identity-key idempotency of
-/// `QueueBlobReplicationOperation`: a live job's presence in `job_dedup_index`
-/// (entries are removed on terminal transition) short-circuits to the existing id.
+/// Effect-driven submit; a live `job_dedup_index` entry short-circuits to the existing id.
 #[derive(Debug, PartialEq)]
 pub struct SubmitJobOperation {
     record: JobRecord,

operations/src/task_incoming.rs

@@ -1059,8 +1059,7 @@ impl OperationsTaskHandler {
             self.jobs_runtime.spawn(self.context.clone(), record);
         }
 
-        // When work is due but every slot is taken, wait for a completion kick
-        // rather than hot-looping on a ZERO re-arm.
+        // At capacity with work due, wait for a completion kick, not a ZERO hot-loop.
         match result.next_due_after {
             Some(after) if after.is_zero() && self.jobs_runtime.available_slots() == 0 => {
                 self.reschedule_timer(TaskKey::DrainJobQueue, JOB_DRAIN_RETRY_AFTER)