Skip to content

[adapters] Fix test_clock_advance_survives_checkpoint. #6581

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ryzhyk merged 1 commit into from
Jul 3, 2026
Merged

[adapters] Fix test_clock_advance_survives_checkpoint. #6581

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 29 additions & 24 deletions crates/adapters/src/controller.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4434,31 +4434,36 @@ impl FtState {
let mut changed_inputs = HashMap::new();
let inputs = self.controller.status.input_status();

// Stop recording if the controller is shutting down. Avoid race with
// `stop`, which removes input endpoints from the pipeline. Without this
// check we may end up recording these endpoints in `remove_inputs`.
if self.controller.state() == PipelineState::Terminated {
return Ok(());
}
self.input_endpoints
.retain(|endpoint_id, (endpoint_name, paused)| {
if let Some(endpoint) = inputs.get(endpoint_id) {
let now_paused = endpoint.is_paused_by_user();
if *paused != now_paused {
changed_inputs.insert(endpoint_name.clone(), now_paused);
*paused = now_paused;
// Compute the endpoint lifecycle diff (added/removed/paused since
// the previous step) that this step's journal entry records, but
// skip it while the controller is shutting down: `stop`
// concurrently disconnects endpoints, so the diff would spuriously
// record still-live endpoints as removed and replay them away.
//
// We must still journal this step's already-computed input, or a
// clean stop would silently drop the last step and a subsequent
// restart would resume from before it.
if self.controller.state() != PipelineState::Terminated {
self.input_endpoints
.retain(|endpoint_id, (endpoint_name, paused)| {
if let Some(endpoint) = inputs.get(endpoint_id) {
let now_paused = endpoint.is_paused_by_user();
if *paused != now_paused {
changed_inputs.insert(endpoint_name.clone(), now_paused);
*paused = now_paused;
}
true
} else {
remove_inputs.insert(endpoint_name.clone());
false
}
true
} else {
remove_inputs.insert(endpoint_name.clone());
false
}
});
for (endpoint_id, status) in inputs.iter() {
self.input_endpoints.entry(*endpoint_id).or_insert_with(|| {
add_inputs.insert(status.endpoint_name.clone(), status.config.clone());
(status.endpoint_name.clone(), status.is_paused_by_user())
});
});
for (endpoint_id, status) in inputs.iter() {
self.input_endpoints.entry(*endpoint_id).or_insert_with(|| {
add_inputs.insert(status.endpoint_name.clone(), status.config.clone());
(status.endpoint_name.clone(), status.is_paused_by_user())
});
}
}
drop(inputs);

Expand Down
19 changes: 19 additions & 0 deletions crates/adapters/src/transport/clock.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -891,6 +891,13 @@ mod test {
}

/// After checkpoint/restart, `NOW()` resumes from the last replayed value, not the anchor.
///
/// Regression coverage for a fault-tolerance journaling race: a step is
/// journaled after its circuit evaluation (so the entry can carry the
/// transaction id assigned there), and a `stop` landing between a step's
/// output being emitted and its journal write must not drop that step, or
/// the post-checkpoint advance is absent from the replay log and `NOW()`
/// regresses to the checkpoint value on restart.
#[test]
fn test_clock_advance_survives_checkpoint() {
use dbsp::circuit::tokio::TOKIO;
Expand Down Expand Up @@ -987,6 +994,18 @@ mod test {
let reader = controller.get_input_endpoint("now").unwrap();
let clock_reader = reader.as_any().downcast::<super::ClockReader>().unwrap();

// Replay runs asynchronously on the circuit thread, feeding the journaled

@gz gz Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks, I guess it would be nice to hide this complexity somehow, I am pretty sure I wont remember this needs to be done next time I have to work with this API.

maybe get_input_endpoint could block until replay is complete

@ryzhyk ryzhyk Jul 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, I guess this test should have checked for controller status. Can't spend more time on this now.

// steps one at a time. Wait for it to finish before reading `NOW()`, or we
// could observe an intermediate value from before the last replayed step.
let deadline = std::time::Instant::now() + Duration::from_secs(30);
while controller.is_replaying() {
assert!(
std::time::Instant::now() < deadline,
"replay did not complete within 30s"
);
sleep(Duration::from_millis(10));
}

let post_replay = TOKIO.block_on(clock_reader.advance(Some(0))).unwrap();
assert_eq!(
post_replay, last_emitted_in_run1,
Expand Down