🧪 test(driver): close the jobs and rate-limit survivors

crates/peryx-driver/src/jobs/mod.rs

@@ -60,6 +60,30 @@ pub struct JobReport {
     pub quota_remaining: u64,
 }
 
+impl JobReport {
+    /// A run that changed every item it examined.
+    #[must_use]
+    pub const fn changed(count: u64) -> Self {
+        Self {
+            processed: count,
+            changed: count,
+            quota_released: 0,
+            quota_remaining: 0,
+        }
+    }
+
+    /// A run that examined items and changed none of them.
+    #[must_use]
+    pub const fn examined(count: u64) -> Self {
+        Self {
+            processed: count,
+            changed: 0,
+            quota_released: 0,
+            quota_remaining: 0,
+        }
+    }
+}
+
 /// The result of a job that stopped without failing.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum JobRunOutcome {
@@ -282,11 +306,7 @@ impl NodeJob for IdleReclaimJob {
             tracing::info!(ecosystem = %self.ecosystem, reclaimed, "idle resources reclaimed");
         }
         let reclaimed = u64::try_from(reclaimed).expect("reclaimed count fits in u64");
-        Ok(JobRunOutcome::succeeded(JobReport {
-            processed: reclaimed,
-            changed: reclaimed,
-            ..JobReport::default()
-        }))
+        Ok(JobRunOutcome::succeeded(JobReport::changed(reclaimed)))
     }
 }
 
@@ -321,11 +341,7 @@ impl NodeJob for IntentFinalizeJob {
         if finalized > 0 {
             tracing::info!(ecosystem = %self.ecosystem, finalized, "admitted writes finalized at home");
         }
-        Ok(JobRunOutcome::succeeded(JobReport {
-            processed: finalized,
-            changed: finalized,
-            ..JobReport::default()
-        }))
+        Ok(JobRunOutcome::succeeded(JobReport::changed(finalized)))
     }
 }
 
@@ -406,11 +422,7 @@ impl NodeJob for JobHistoryCleanup {
         let mut removed = 0_u64;
         loop {
             if ctx.is_cancelled() {
-                return Ok(JobRunOutcome::cancelled(JobReport {
-                    processed: removed,
-                    changed: removed,
-                    ..JobReport::default()
-                }));
+                return Ok(JobRunOutcome::cancelled(JobReport::changed(removed)));
             }
             let batch = ctx
                 .state()
@@ -419,11 +431,7 @@ impl NodeJob for JobHistoryCleanup {
                 .map_err(|error| JobFailure::new("storage", error.to_string()))?;
             removed += u64::try_from(batch).expect("bounded batch fits in u64");
             if batch == 0 {
-                return Ok(JobRunOutcome::succeeded(JobReport {
-                    processed: removed,
-                    changed: removed,
-                    ..JobReport::default()
-                }));
+                return Ok(JobRunOutcome::succeeded(JobReport::changed(removed)));
             }
         }
     }
@@ -503,11 +511,7 @@ impl NodeJob for WriteLedgerReap {
         let mut reaped = 0_u64;
         loop {
             if ctx.is_cancelled() {
-                return Ok(JobRunOutcome::cancelled(JobReport {
-                    processed: reaped,
-                    changed: reaped,
-                    ..JobReport::default()
-                }));
+                return Ok(JobRunOutcome::cancelled(JobReport::changed(reaped)));
             }
             let now = (ctx.state().clock)();
             let expired = reap_storage_result(ctx.state().meta.expire_stale_intents(
@@ -609,15 +613,8 @@ impl NodeJob for SearchRebuildJob {
             })
             .map_err(|error| JobFailure::new("search_rebuild", error.to_string()))?;
         Ok(match outcome {
-            RebuildOutcome::Published { documents } => JobRunOutcome::succeeded(JobReport {
-                processed: documents,
-                changed: documents,
-                ..JobReport::default()
-            }),
-            RebuildOutcome::Aborted { documents } => JobRunOutcome::cancelled(JobReport {
-                processed: documents,
-                ..JobReport::default()
-            }),
+            RebuildOutcome::Published { documents } => JobRunOutcome::succeeded(JobReport::changed(documents)),
+            RebuildOutcome::Aborted { documents } => JobRunOutcome::cancelled(JobReport::examined(documents)),
         })
     }
 }

crates/peryx-driver/tests/unit/jobs/tests.rs

@@ -1333,6 +1333,32 @@ fn test_job_failure_parts_preserve_code_and_message() {
     );
 }
 
+#[test]
+fn test_a_report_of_changed_items_counts_them_as_examined_too() {
+    assert_eq!(
+        JobReport::changed(3),
+        JobReport {
+            processed: 3,
+            changed: 3,
+            quota_released: 0,
+            quota_remaining: 0,
+        }
+    );
+}
+
+#[test]
+fn test_a_report_of_examined_items_changes_none() {
+    assert_eq!(
+        JobReport::examined(3),
+        JobReport {
+            processed: 3,
+            changed: 0,
+            quota_released: 0,
+            quota_remaining: 0,
+        }
+    );
+}
+
 #[tokio::test]
 async fn test_job_history_cleanup_removes_every_excess_terminal_attempt() {
     let (_dir, state) = serving();
@@ -1363,6 +1389,31 @@ async fn test_job_history_cleanup_removes_every_excess_terminal_attempt() {
     assert_eq!(job_runs(&state.meta).len(), 16);
 }
 
+/// The store prunes at most 128 runs per batch, so a backlog past that takes more than one pass. The
+/// sweep keeps going until a pass removes nothing, rather than reporting the first batch as the
+/// whole job and leaving the excess for the next tick.
+#[tokio::test]
+async fn test_job_history_cleanup_drains_a_backlog_wider_than_one_batch() {
+    let (_dir, state) = serving();
+    for _ in 0..145 {
+        let id = start_corruptible_attempt(&state.meta);
+        state
+            .meta
+            .finish_job_run(&id, JobOutcome::succeeded(100, 0, 0))
+            .unwrap();
+    }
+
+    let report = JobHistoryCleanup { retain: 16 }
+        .run(&context(state.clone(), CancellationToken::new()))
+        .await
+        .unwrap();
+
+    assert_eq!(
+        (report, job_runs(&state.meta).len()),
+        (JobRunOutcome::succeeded(JobReport::changed(129)), 16)
+    );
+}
+
 #[tokio::test]
 async fn test_job_history_cleanup_honors_cancellation_before_writing() {
     let (_dir, state) = serving();
@@ -2571,6 +2622,61 @@ async fn test_write_ledger_reap_stops_when_cancelled() {
     assert_eq!(report, JobRunOutcome::cancelled(JobReport::default()));
 }
 
+/// The three ledgers drain at different rates, so a sweep that stopped once any one of them came up
+/// empty would leave the others holding settled rows until a later tick. One row per ledger per pass
+/// makes the difference visible: the expiry ledger empties first, then the intents, and the outcomes
+/// last.
+#[tokio::test]
+async fn test_write_ledger_reap_keeps_going_until_every_ledger_is_drained() {
+    let (_dir, state) = serving();
+    let past = -3000;
+    stage_refused_intent(
+        &state,
+        "stranded",
+        1_000 - super::INGRESS_STAGING_DEADLINE_SECS,
+        super::MAX_INTENT_REFUSALS,
+    );
+    stage_refused_intent(&state, "done", past, 0);
+    state
+        .meta
+        .advance_intent("done", peryx_storage::meta::IntentPhase::Admitted, past)
+        .unwrap();
+    let operations = ["op-a", "op-b", "op-c"];
+    for operation in operations {
+        state.meta.claim_operation(operation, Some(0), past).unwrap();
+        state
+            .meta
+            .finalize_operation(
+                operation,
+                peryx_storage::meta::OperationResult::Published,
+                b"body",
+                past,
+            )
+            .unwrap();
+    }
+
+    let report = super::WriteLedgerReap { batch: 1 }
+        .run(&context(state.clone(), CancellationToken::new()))
+        .await
+        .unwrap();
+
+    let outcomes = operations.map(|operation| state.meta.operation_outcome(operation).unwrap());
+    assert_eq!(
+        (
+            report,
+            state.meta.staged_intent("stranded").unwrap().unwrap().phase,
+            state.meta.staged_intent("done").unwrap(),
+            outcomes,
+        ),
+        (
+            JobRunOutcome::succeeded(JobReport::changed(5)),
+            peryx_storage::meta::IntentPhase::Expired,
+            None,
+            [None, None, None],
+        )
+    );
+}
+
 #[tokio::test]
 async fn test_write_ledger_reap_surfaces_a_storage_fault() {
     let dir = tempfile::tempdir().unwrap();

crates/peryx-driver/tests/unit/rate_limit/tests.rs

@@ -310,6 +310,22 @@ fn test_zero_limit_is_unbounded() {
     assert!(limiter.check_client(RouteClass::Listing, IpAddr::V4(Ipv4Addr::LOCALHOST)));
 }
 
+/// The hot path asks this before touching forwarded headers at all, so a limiter that trusted
+/// nobody yet answered "some proxy" would parse headers it then ignores, and one that answered
+/// "none" over a configured proxy would bucket every proxied client by the proxy's address.
+#[test]
+fn test_a_limiter_reports_whether_any_proxy_is_trusted() {
+    let proxied = RateLimiter::new(RateLimitConfig {
+        trusted_proxies: vec!["10.0.0.0/8".parse().unwrap()],
+        ..RateLimitConfig::enabled_defaults()
+    });
+
+    assert_eq!(
+        (RateLimiter::default().trusts_any_proxy(), proxied.trusts_any_proxy()),
+        (false, true)
+    );
+}
+
 #[test]
 fn test_proxy_trust_canonicalizes_addresses() {
     let limiter = RateLimiter::new(RateLimitConfig {