Skip to content

Time travel #6220

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

Open
gz wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Time travel #6220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
79a2856
adapters: add now_offset for testing purposes
gz May 12, 2026
55b6550
adapters: add clock/advance API
gz May 12, 2026
a051bf1
py+fda: Add clock advance commands
gz May 12, 2026
02ed7e2
[ci] apply automatic fixes
feldera-bot May 12, 2026
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
39 changes: 39 additions & 0 deletions crates/adapters/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1252,6 +1252,7 @@ where
.service(coordination_adhoc_scan)
.service(coordination_labels_incomplete)
.service(coordination_restart)
.service(clock_advance)
}

/// Implements `/start`, `/pause`, `/activate`:
Expand Down Expand Up @@ -2045,6 +2046,44 @@ async fn commit_transaction(state: WebData<ServerState>) -> Result<impl Responde
Ok(HttpResponse::Ok().json("Transaction commit initiated"))
}

/// Advance the externally-driven `NOW()` clock and report its new value.
///
/// Requires `dev_tweaks.now_http_driven = true`.
#[post("/clock/advance")]
async fn clock_advance(
state: WebData<ServerState>,
body: web::Json<feldera_types::transport::clock::ClockAdvanceRequest>,
) -> Result<impl Responder, PipelineError> {
use feldera_types::transport::clock::ClockAdvanceResponse;
let controller = state.controller()?;
let reader =
controller
.get_input_endpoint("now")
.ok_or_else(|| PipelineError::InvalidParam {
error: "no `now` clock connector is registered".to_string(),
})?;
let clock_reader = reader
.as_any()
.downcast::<crate::transport::clock::ClockReader>()
.map_err(|_| PipelineError::InvalidParam {
error: "`now` connector is not a clock reader".to_string(),
})?;

let now_ms =
clock_reader
.advance(body.delta_ms)
.await
.map_err(|e| PipelineError::InvalidParam {
error: e.to_string(),
})?;

let now = chrono::DateTime::<Utc>::from_timestamp_millis(now_ms as i64)
.map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
.unwrap_or_else(|| format!("{now_ms} ms since epoch"));

Ok(HttpResponse::Ok().json(ClockAdvanceResponse { now_ms, now }))
}

#[derive(Debug, Deserialize)]
struct IngressArgs {
#[serde(default = "HttpInputTransport::default_format")]
Expand Down
Loading
Loading