diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index ea8cea43b19..5f3627574c2 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); @@ -4572,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, @@ -7904,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, @@ -7932,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 bd2ab59103a..8cb9def7883 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -3465,7 +3465,9 @@ mod test_http_helpers { }; use feldera_types::{ completion_token::{CompletionStatus, CompletionStatusResponse, CompletionTokenResponse}, + config::ProgramIr, runtime_status::{BootstrapPolicy, RuntimeStatus}, + transaction::ConcurrentBootstrapPhase, }; use futures::{Stream, StreamExt}; use std::{ @@ -3530,7 +3532,7 @@ outputs: Uuid::now_v7(), BootstrapConfig::from(BootstrapPolicy::Allow), &[Some("v0")], - false, + None, ) .await; @@ -3859,30 +3861,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 +3946,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 || { @@ -4248,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. /// @@ -4294,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, 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, @@ -4312,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; @@ -4679,6 +4722,135 @@ 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. + /// + /// `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(); + 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 + send_snapshot: {} + transport: + name: file_output + config: + path: "{}" + format: + name: csv + config: {{}} +"#, + storage_dir.display(), + send_snapshot, + 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; + } + + #[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. @@ -4824,7 +4996,7 @@ outputs: Uuid::new_v4(), BootstrapConfig::from(BootstrapPolicy::Allow), &[Some("v0")], - true, + Some(test_program_ir("v0")), ) .await; start_pipeline(&server).await; @@ -4845,7 +5017,7 @@ outputs: Uuid::new_v4(), BootstrapConfig::from(BootstrapPolicy::Allow), &[Some("v0")], - true, + Some(test_program_ir("v0")), ) .await; start_pipeline(&server).await;