From 8c8f212623c5dcb5c6d99f188af09257f8d7d829 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 12 Aug 2026 10:06:14 -0700 Subject: [PATCH 1/3] [adapters] let a server test inject a program IR The in-process server tests build their circuit from a Rust closure, so the pipeline config carries no program information and `compute_pipeline_diff` finds none to compare. `start_test_server_with_state` papered over that with a `with_program_ir` flag that injected one fixed empty IR, which only ever produces an empty diff. Take the IR itself, so a test can also inject IRs that differ across a restart and exercise the paths a recompiled program takes. `test_program_ir` builds one for the test circuit's single view, keyed on the view's persistent id. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/server.rs | 56 ++++++++++++++++++++++++----------- 1 file changed, 39 insertions(+), 17 deletions(-) diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index bd2ab59103a..4a9625ffcc3 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -3465,6 +3465,7 @@ mod test_http_helpers { }; use feldera_types::{ completion_token::{CompletionStatus, CompletionStatusResponse, CompletionTokenResponse}, + config::ProgramIr, runtime_status::{BootstrapPolicy, RuntimeStatus}, }; use futures::{Stream, StreamExt}; @@ -3530,7 +3531,7 @@ outputs: Uuid::now_v7(), BootstrapConfig::from(BootstrapPolicy::Allow), &[Some("v0")], - false, + None, ) .await; @@ -3859,30 +3860,54 @@ outputs: deployment_id, bootstrap_config, persistent_output_ids, - false, + None, ) .await .0 } + /// A `program_ir` describing one input table and one view, the shape of the + /// test circuit. `view_persistent_id` identifies the view's state: giving + /// two restarts different ids makes `compute_pipeline_diff` report the view + /// as modified, exactly as a changed view definition does in a compiled + /// pipeline. + pub(super) fn test_program_ir(view_persistent_id: &str) -> ProgramIr { + ProgramIr { + mir: serde_json::from_value(serde_json::json!({ + "0": { + "operation": "source", + "table": "test_input1", + "persistent_id": "test_input1", + }, + "1": { + "operation": "sink", + "view": "test_output1", + "persistent_id": view_persistent_id, + }, + })) + .unwrap(), + program_schema: serde_json::json!({ "inputs": [], "outputs": [] }), + } + } + /// Like [start_test_server_with_options], but also returns the /// [ServerState] so a test can reach the controller directly (e.g. to /// simulate an ungraceful crash via [crash_pipeline]). /// - /// When `with_program_ir` is set, a minimal (empty) `program_ir` is injected - /// into the pipeline config. The test circuit is a Rust closure with no - /// compiled program, so its checkpoint normally carries no program info, - /// which makes `compute_pipeline_diff` fail and forces the journal to be - /// discarded on restart (no replay). Injecting an identical empty + /// `program_ir` is injected into the pipeline config. The test circuit is a + /// Rust closure with no compiled program, so its checkpoint normally carries + /// no program info, which makes `compute_pipeline_diff` fail and forces the + /// journal to be discarded on restart (no replay). Injecting an identical /// `program_ir` makes the diff empty across a same-program restart, so the /// journal is replayed -- required to exercise the fault-tolerant replay - /// path in-process. + /// path in-process. Injecting IRs that differ makes the diff report the + /// change, as it does for a recompiled program. pub(super) async fn start_test_server_with_state( config_str: &str, deployment_id: Uuid, bootstrap_config: BootstrapConfig, persistent_output_ids: &'static [Option<&'static str>], - with_program_ir: bool, + program_ir: Option, ) -> (TestServer, WebData) { let mut config_file = NamedTempFile::new().unwrap(); config_file.write_all(config_str.as_bytes()).unwrap(); @@ -3920,11 +3945,8 @@ outputs: }; let mut config = parse_config(&args.config_file).unwrap(); - if with_program_ir { - config.program_ir = Some(feldera_types::config::ProgramIr { - mir: std::collections::HashMap::new(), - program_schema: serde_json::json!({ "inputs": [], "outputs": [] }), - }); + if let Some(program_ir) = program_ir { + config.program_ir = Some(program_ir); } let builder = ControllerBuilder::new(&config).unwrap(); thread::spawn(move || { @@ -4296,7 +4318,7 @@ mod test_http { adhoc_query_count, assert_no_file_output, batch_num_records, commit_transaction, crash_pipeline, pause_pipeline, send_input, send_input_no_wait, start_pipeline, start_test_server_with_options, start_test_server_with_state, start_transaction, - suspend_pipeline, test_batches, wait_for_file_output, + suspend_pipeline, test_batches, test_program_ir, wait_for_file_output, }, test::{ TestStruct, async_wait, generate_test_batches, @@ -4824,7 +4846,7 @@ outputs: Uuid::new_v4(), BootstrapConfig::from(BootstrapPolicy::Allow), &[Some("v0")], - true, + Some(test_program_ir("v0")), ) .await; start_pipeline(&server).await; @@ -4845,7 +4867,7 @@ outputs: Uuid::new_v4(), BootstrapConfig::from(BootstrapPolicy::Allow), &[Some("v0")], - true, + Some(test_program_ir("v0")), ) .await; start_pipeline(&server).await; From 3c89040f51bf1dbb70064346c153b471007828da Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 12 Aug 2026 10:07:01 -0700 Subject: [PATCH 2/3] [adapters] withhold output progress until the bootstrap re-emits `total_processed_input_records` promises that an output endpoint's output equals the circuit's output after that many input records. A concurrent bootstrap broke the promise for the endpoints whose relations it rebuilds: the pre-existing views stay live, so the primary circuit keeps stepping throughout the backfill, and `push_output` tagged each of those steps' empty batches with the pipeline's full record count. The endpoint stored it, claiming it had delivered output that the cutover had not yet re-emitted. A reader waiting on the counter to learn the sink is caught up -- `wait_for_output_progress` in the platform tests -- then read a transmitted count missing the whole re-emission. Withhold the progress figure from an endpoint the bootstrap still owes a re-emission, so the counter reaches the pipeline's count on the step that carries that output, not before. Fixes: https://github.com/feldera/cloud/issues/1832 Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/controller.rs | 37 ++++++++ crates/adapters/src/server.rs | 140 +++++++++++++++++++++++++++++- 2 files changed, 173 insertions(+), 4 deletions(-) diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index ea8cea43b19..4ce9ffd53de 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -4516,6 +4516,32 @@ impl CircuitThread { let silent_bootstrap = self.silent_bootstrap && self.controller.status.bootstrap_in_progress(); + // Whether the bootstrap still owes a re-emission to the endpoints whose + // relations it rebuilds. Until it arrives, those endpoints receive only + // empty batches: a concurrent bootstrap keeps the pre-existing views + // live and steps the primary circuit throughout the backfill. Reporting + // the pipeline's progress on those batches would claim output the + // endpoint never received, so `processed_records` is withheld from them + // until the step that carries the re-emission. + // + // A silent bootstrap owes nothing: it suppresses the re-emission, and + // discarding it is what progress means for those endpoints. + // + // The step that carries the re-emission is excluded. For a concurrent + // bootstrap that is the `Finalizing` step: cutover has installed the new + // views, and the transaction that commits during this phase drains + // their full contents. `self.bootstrapping` is the stop-the-world + // analogue, cleared earlier in `step` on the step that completes the + // bootstrap. + let bootstrap_owes_reemission = !silent_bootstrap + && (self.bootstrapping + || matches!( + self.concurrent_phase, + ConcurrentPhase::Backfill + | ConcurrentPhase::AwaitingSync + | ConcurrentPhase::Synchronize + )); + let outputs = self.controller.outputs.read().unwrap(); for (_stream, (output_handles, endpoints)) in outputs.iter_by_stream() { let (mut delta_batch, num_delta_records) = if silent_bootstrap { @@ -4532,6 +4558,17 @@ impl CircuitThread { let endpoint = outputs.lookup_by_id(endpoint_id).unwrap(); let transaction = self.controller.get_transaction_number(); + let owed_reemission = bootstrap_owes_reemission + && self + .controller + .bootstrapped_output_endpoints + .contains(&endpoint.endpoint_name); + let processed_records = if owed_reemission { + None + } else { + processed_records + }; + // Silent bootstrap: send empty batch for progress tracking only. if silent_bootstrap { self.controller.status.enqueue_batch(*endpoint_id, 0); diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index 4a9625ffcc3..381ed9d718b 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -3467,6 +3467,7 @@ mod test_http_helpers { completion_token::{CompletionStatus, CompletionStatusResponse, CompletionTokenResponse}, config::ProgramIr, runtime_status::{BootstrapPolicy, RuntimeStatus}, + transaction::ConcurrentBootstrapPhase, }; use futures::{Stream, StreamExt}; use std::{ @@ -4270,6 +4271,22 @@ outputs: .expect("test_output1 output endpoint") } + /// `(total_processed_input_records, transmitted_records, queued_records)` of + /// the `test_output1` endpoint, with the pipeline's concurrent-bootstrap + /// phase. + pub(super) async fn output_progress( + server: &TestServer, + ) -> (u64, u64, u64, ConcurrentBootstrapPhase) { + let stats = get_stats(server).await; + let metrics = output_metrics(&stats); + ( + metrics.total_processed_input_records, + metrics.transmitted_records, + metrics.queued_records, + stats.global_metrics.concurrent_bootstrap_phase, + ) + } + /// After a silent-bootstrap restart, verify that backfilled records were /// processed but not written to the output file. /// @@ -4316,9 +4333,10 @@ mod test_http { ensure_default_crypto_provider, server::test_http_helpers::{ adhoc_query_count, assert_no_file_output, batch_num_records, commit_transaction, - crash_pipeline, pause_pipeline, send_input, send_input_no_wait, start_pipeline, - start_test_server_with_options, start_test_server_with_state, start_transaction, - suspend_pipeline, test_batches, test_program_ir, wait_for_file_output, + crash_pipeline, output_progress, pause_pipeline, send_input, send_input_no_wait, + start_pipeline, start_test_server_with_options, start_test_server_with_state, + start_transaction, suspend_pipeline, test_batches, test_program_ir, + wait_for_file_output, }, test::{ TestStruct, async_wait, generate_test_batches, @@ -4334,7 +4352,10 @@ mod test_http { strategy::{Strategy, ValueTree}, test_runner::TestRunner, }; - use std::{thread::sleep, time::Duration}; + use std::{ + thread::sleep, + time::{Duration, Instant}, + }; use tempfile::TempDir; use tokio::time::timeout; use uuid::Uuid; @@ -4701,6 +4722,117 @@ outputs: suspend_pipeline(&server, Some(&storage_dir)).await; } + /// `total_processed_input_records` must not reach the pipeline's record + /// count while the concurrent-bootstrap cutover still owes the endpoint the + /// re-emission of its view. + /// + /// The counter promises that the endpoint's output equals the circuit's + /// output after that many input records, so a reader waiting on it concludes + /// the sink is up to date. During a concurrent bootstrap the pre-existing + /// view stays live, and the empty batches those steps push to the endpoint + /// carry the pipeline's full record count -- which advances the counter + /// before the cutover re-emission reaches the transport. + #[actix_web::test] + async fn test_concurrent_bootstrap_output_progress_waits_for_cutover() { + ensure_default_crypto_provider(); + + let tempdir = TempDir::new().unwrap(); + let storage_dir = tempdir.path().join("storage"); + let output_path = tempdir.path().join("output.csv"); + std::fs::create_dir(&storage_dir).unwrap(); + + // A one-record replay chunk stretches the background backfill over many + // pumps, so the pre-cutover window spans several samples of the + // endpoint's counters. + let config_str = format!( + r#" +name: test +workers: 4 +storage_config: + path: "{}" +storage: true +clock_resolution_usecs: +dev_tweaks: + splitter_chunk_size_records: 1 +inputs: +outputs: + test_output1: + stream: test_output1 + transport: + name: file_output + config: + path: "{}" + format: + name: csv + config: {{}} +"#, + storage_dir.display(), + output_path.display() + ); + + let first_batch = test_batches(0, 2_000); + let ingested = batch_num_records(&first_batch); + + let (server, _) = start_test_server_with_state( + &config_str, + Uuid::new_v4(), + BootstrapConfig::from(BootstrapPolicy::Allow), + &[Some("v0")], + Some(test_program_ir("v0")), + ) + .await; + start_pipeline(&server).await; + send_input(&server, &first_batch).await; + wait_for_file_output(&output_path, &first_batch).await; + let (_, transmitted_before, _, _) = output_progress(&server).await; + assert_eq!(transmitted_before, ingested); + suspend_pipeline(&server, Some(&storage_dir)).await; + drop(server); + + // Restart with a changed view and concurrent bootstrap, feeding no new + // input: every record the endpoint still owes is derived from the + // `ingested` records the pipeline processed before the restart. + let (server, _) = start_test_server_with_state( + &config_str, + Uuid::new_v4(), + BootstrapConfig::from(BootstrapPolicy::Allow).with_concurrent_bootstrap(true), + &[Some("v1")], + Some(test_program_ir("v1")), + ) + .await; + + // Sample until the endpoint reports the pipeline's full progress. That + // reading promises the re-emission has reached the transport, so the + // transmitted count must already include it. Every sample before it + // must report less than full progress: the endpoint transmits the + // re-emission and takes its new progress reading in one move, leaving no + // sample in between. + let expected_transmitted = 2 * ingested; + let mut trace = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(60); + loop { + let sample = output_progress(&server).await; + trace.push(sample); + let (processed, transmitted, _queued, _phase) = sample; + if processed >= ingested { + assert_eq!( + transmitted, expected_transmitted, + "the endpoint reported {processed} processed records with the cutover \ + re-emission still owed; trace = {trace:?}" + ); + break; + } + assert!( + Instant::now() < deadline, + "the endpoint never caught up with the cutover re-emission; trace = {trace:?}" + ); + } + + // The re-emission is also on disk: the modified connector re-truncates + // the file, so it holds the new view's full contents. + wait_for_file_output(&output_path, &first_batch).await; + } + /// Regression test: after a concurrent-bootstrap cutover, an ad-hoc query of /// the backfilled view must observe its full contents WITHOUT any /// post-cutover input. From 7c7ba0613253f89e755850c8f91232688eaac960 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 14 Aug 2026 18:59:56 -0700 Subject: [PATCH 3/3] [adapters] stop the snapshot path from inventing output progress Review caught the withheld progress figure coming back to life one branch over: `enqueue_latest_snapshot` replaced a `None` with the pipeline's current count, so a connector configured with `send_snapshot: true` on a relation the bootstrap rebuilds reported full progress while it was still owed the re-emission -- the same defect as on the delta path, reached through a different door. `None` on the batch queue means "leave the counter alone" everywhere else, and it now means that here too. The caller supplies the figure it wants reported, including the one this function used to synthesize: a snapshot reflects the last committed transaction, so an endpoint receiving one is up to date with the pipeline's count even mid-transaction, where the delta path has nothing to report. The regression test now runs both paths. Reported-by: Ben Pfaff Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/controller.rs | 33 +++++++++++++++++++++++++++---- crates/adapters/src/server.rs | 22 +++++++++++++++++++-- 2 files changed, 49 insertions(+), 6 deletions(-) diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 4ce9ffd53de..5f3627574c2 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -4609,6 +4609,30 @@ impl CircuitThread { // delivered its initial snapshot. Deliver the snapshot and // skip the normal delta path. if endpoint.control.is_snapshot_pending() { + // A snapshot reflects the state as of the last committed + // transaction, so an endpoint that receives one is up to + // date with the pipeline's own count even mid-transaction, + // where the delta path has no figure to report. An endpoint + // still owed a re-emission is the exception: it gets the + // pre-cutover snapshot, so its counter has to keep waiting. + let processed_records = if owed_reemission { + None + } else { + processed_records.or_else(|| { + Some(ProcessedRecords { + total_processed_input_records: self + .controller + .status + .num_total_processed_records(), + total_processed_steps: self + .controller + .status + .global_metrics + .total_completed_steps(), + }) + }) + }; + self.controller.enqueue_latest_snapshot( *endpoint_id, &endpoint.stream_name, @@ -7941,6 +7965,11 @@ impl ControllerInner { /// endpoint's batch queue. Returns `true` if a snapshot was found and /// enqueued, `false` if no cached snapshot exists yet (e.g., the pipeline /// hasn't completed its first step). + /// + /// `processed_records` means the same here as everywhere else on the batch + /// queue: `None` leaves the endpoint's progress counter alone. The caller + /// decides what to report; it is the one that knows whether the endpoint is + /// still owed output. #[allow(clippy::too_many_arguments)] fn enqueue_latest_snapshot( &self, @@ -7969,10 +7998,6 @@ impl ControllerInner { return false; }; - let processed_records = processed_records.or(Some(ProcessedRecords { - total_processed_input_records: self.status.num_total_processed_records(), - total_processed_steps: self.status.global_metrics.total_completed_steps(), - })); let step = step.unwrap_or_else(|| { processed_records .as_ref() diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index 381ed9d718b..8cb9def7883 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -4732,8 +4732,12 @@ outputs: /// view stays live, and the empty batches those steps push to the endpoint /// carry the pipeline's full record count -- which advances the counter /// before the cutover re-emission reaches the transport. - #[actix_web::test] - async fn test_concurrent_bootstrap_output_progress_waits_for_cutover() { + /// + /// `send_snapshot` picks the path the endpoint takes to that same promise: + /// the ordinary delta path, or `enqueue_latest_snapshot`, which hands the + /// endpoint the pre-cutover snapshot of a view the bootstrap has yet to + /// rebuild. + async fn concurrent_bootstrap_output_progress_case(send_snapshot: bool) { ensure_default_crypto_provider(); let tempdir = TempDir::new().unwrap(); @@ -4758,6 +4762,7 @@ inputs: outputs: test_output1: stream: test_output1 + send_snapshot: {} transport: name: file_output config: @@ -4767,6 +4772,7 @@ outputs: config: {{}} "#, storage_dir.display(), + send_snapshot, output_path.display() ); @@ -4833,6 +4839,18 @@ outputs: wait_for_file_output(&output_path, &first_batch).await; } + #[actix_web::test] + async fn test_concurrent_bootstrap_output_progress_waits_for_cutover() { + concurrent_bootstrap_output_progress_case(false).await; + } + + /// The `send_snapshot` variant: `enqueue_latest_snapshot` must not report + /// progress the endpoint has not earned either. + #[actix_web::test] + async fn test_concurrent_bootstrap_snapshot_progress_waits_for_cutover() { + concurrent_bootstrap_output_progress_case(true).await; + } + /// Regression test: after a concurrent-bootstrap cutover, an ad-hoc query of /// the backfilled view must observe its full contents WITHOUT any /// post-cutover input.