Skip to content

Commit b1a4ab0

Browse files
test(precompute): add a wall-clock override hook to PrecomputeEngine (#537)
Stacked on the #474 fix. Worker already has a fake-clock override (set_now_ms_fn) for its own unit tests, but it's #[cfg(test)]-gated and invisible to integration tests under tests/, which link the library normally. Widen it to pub(crate) and add PrecomputeEngine::with_now_ms_fn, a builder that threads a shared clock override down to every spawned worker, so an integration test can drive the wall-clock fallback deterministically through the real ingest pipeline instead of racing a real clock with sleeps. No behavioral change: the override defaults to None and every worker keeps its default SystemTime::now-backed clock unless a caller opts in. Production code never calls with_now_ms_fn. Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
1 parent 9a018cb commit b1a4ab0

2 files changed

Lines changed: 30 additions & 8 deletions

File tree

asap-query-engine/src/precompute_engine/engine.rs

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,10 @@ pub struct PrecomputeEngine {
7474
receivers: Option<Vec<mpsc::Receiver<WorkerMessage>>>,
7575
/// Shared ingest agg_configs, swappable at runtime.
7676
ingest_agg_configs: Arc<ArcSwap<Vec<Arc<AggregationConfig>>>>,
77+
/// Test-support wall-clock override, applied to every spawned worker.
78+
/// See `with_now_ms_fn`. `None` in production — each worker keeps its
79+
/// default `SystemTime::now`-backed clock.
80+
now_ms_fn: Option<Arc<dyn Fn() -> i64 + Send + Sync>>,
7781
}
7882

7983
impl PrecomputeEngine {
@@ -119,6 +123,7 @@ impl PrecomputeEngine {
119123
senders,
120124
receivers: Some(receivers),
121125
ingest_agg_configs,
126+
now_ms_fn: None,
122127
}
123128
}
124129

@@ -127,6 +132,16 @@ impl PrecomputeEngine {
127132
self.diagnostics.clone()
128133
}
129134

135+
/// Test-support builder: override the wall-clock source used by every
136+
/// worker's wall-clock fallback (`flush_all`). Lets integration tests
137+
/// drive the fallback deterministically through the real ingest pipeline
138+
/// instead of racing a real clock with `tokio::time::sleep`. Must be
139+
/// called before `run()`. Production code never calls this.
140+
pub fn with_now_ms_fn(mut self, f: impl Fn() -> i64 + Send + Sync + 'static) -> Self {
141+
self.now_ms_fn = Some(Arc::new(f));
142+
self
143+
}
144+
130145
/// Return a handle for applying runtime config updates to this engine.
131146
/// Must be called before `run()`.
132147
pub fn handle(&self) -> PrecomputeEngineHandle {
@@ -160,7 +175,7 @@ impl PrecomputeEngine {
160175
// Spawn workers
161176
let mut worker_handles = Vec::with_capacity(num_workers);
162177
for (id, rx) in receivers.into_iter().enumerate() {
163-
let worker = Worker::new(
178+
let mut worker = Worker::new(
164179
id,
165180
rx,
166181
self.output_sink.clone(),
@@ -177,6 +192,10 @@ impl PrecomputeEngine {
177192
self.diagnostics.worker_watermarks[id].clone(),
178193
self.diagnostics.worker_watermarks.to_vec(),
179194
);
195+
if let Some(now_ms_fn) = &self.now_ms_fn {
196+
let now_ms_fn = now_ms_fn.clone();
197+
worker.set_now_ms_fn(Box::new(move || now_ms_fn()));
198+
}
180199
let handle = tokio::spawn(async move {
181200
worker.run().await;
182201
});

asap-query-engine/src/precompute_engine/worker.rs

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,8 @@ pub struct Worker {
112112
wall_clock_grace_period_ms: i64,
113113
/// Injectable clock returning current wall-clock time in milliseconds
114114
/// since the unix epoch. Production uses `SystemTime::now`; tests
115-
/// override with a deterministic fake via `set_now_ms_fn`.
115+
/// override with a deterministic fake via `set_now_ms_fn` (directly, or
116+
/// via `PrecomputeEngine::with_now_ms_fn` for integration tests).
116117
now_ms_fn: Box<dyn Fn() -> i64 + Send + Sync>,
117118
}
118119

@@ -154,12 +155,14 @@ impl Worker {
154155
}
155156
}
156157

157-
/// Test/diagnostic-only setter for the wall-clock source. Replaces the
158-
/// default `SystemTime::now`-backed clock with a deterministic fake so
159-
/// unit tests can drive the wall-clock fallback in `flush_all` without
160-
/// `std::thread::sleep`. Production code never calls this.
161-
#[cfg(test)]
162-
pub fn set_now_ms_fn(&mut self, f: Box<dyn Fn() -> i64 + Send + Sync>) {
158+
/// Test-support setter for the wall-clock source. Replaces the default
159+
/// `SystemTime::now`-backed clock with a deterministic fake so tests can
160+
/// drive the wall-clock fallback in `flush_all` without real sleeping.
161+
/// Crate-visible (not `#[cfg(test)]`-gated) so `PrecomputeEngine::
162+
/// with_now_ms_fn` can reach it from integration tests in `tests/`,
163+
/// which link the library normally and don't see `#[cfg(test)]` items.
164+
/// Production code never calls this.
165+
pub(crate) fn set_now_ms_fn(&mut self, f: Box<dyn Fn() -> i64 + Send + Sync>) {
163166
self.now_ms_fn = f;
164167
}
165168

0 commit comments

Comments
 (0)