From 8c46ed55c1c611dceecb7aea6a06ad2ddf7c38d6 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Mon, 5 Oct 2026 20:40:48 +0000 Subject: [PATCH 1/8] Clean up old generations that are building up in active long-running (still-running) workflows --- Cargo.lock | 5 +- Cargo.toml | 3 +- USER_GUIDE.md | 41 +++- docs/upgrade-testing.md | 18 ++ scripts/test-e2e-local.sh | 2 +- src/lib.rs | 12 +- src/types.rs | 3 +- src/worker.rs | 253 ++++++++++++++++++++++++- tests/e2e/sql/89_execution_pruning.sql | 191 +++++++++++++++++++ 9 files changed, 507 insertions(+), 21 deletions(-) create mode 100644 tests/e2e/sql/89_execution_pruning.sql diff --git a/Cargo.lock b/Cargo.lock index e821bc7a..2ec31fbb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -731,9 +731,8 @@ dependencies = [ [[package]] name = "duroxide-pg" -version = "0.1.34" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbd3d51000bd6b0433be07bc2f4ba5dd7d8fc4336fc90ff2321f527cd51a3eac" +version = "0.1.35" +source = "git+https://github.com/microsoft/duroxide-pg?branch=thomcc%2Fcontinue_as_new_completed_at#e7773b34bee0127efafd796cdf8c732cd537cd2d" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index c6a7eee5..87e62f34 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,7 +37,8 @@ percent-encoding = "2.3" # duroxide integration duroxide = "=0.1.30" -duroxide-pg = "=0.1.34" +# DONOTLAND: replace this temporary continue_as_new_completed_at pin with a crates.io release. +duroxide-pg = { git = "https://github.com/microsoft/duroxide-pg", branch = "thomcc/continue_as_new_completed_at" } tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] } # For ExecuteSQL activity (connecting back to PostgreSQL) diff --git a/USER_GUIDE.md b/USER_GUIDE.md index e9a5fe70..bd019199 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -2533,19 +2533,25 @@ satellite `df._worker_epoch` is normal; satellites do not have separate workers. pg_durable tracks each workflow in its own `df` tables, while the durable engine keeps workflow state in a separate schema. To keep those two stores consistent — and to keep `df.instances`/`df.nodes` from growing without bound — the background -worker runs a **best-effort reconciliation pass** that does two things: +worker runs a **best-effort reconciliation pass** that: - **Removes expired terminal instances.** Old **terminal** instances (status `completed`, `failed`, or `cancelled`) and their `df.nodes` rows are deleted, - along with their engine records. Retention does not remove running or pending - instances from an existing installation, regardless of age. + along with their engine records. Retention does not remove running or pending + instances from an existing installation, regardless of age. - **Reclaims orphaned engine records.** `df.start()` writes the `df` rows in the caller's transaction but hands the workflow to the engine over a separate connection; if that transaction **rolls back**, the `df` rows vanish while the engine keeps an inert record (it can never load its rolled-back graph, so it ends up failed). Reconciliation deletes such df-less engine records once they - age past `retention_days`. Anything the engine is still tracking with a live - `df` row is left untouched. + age past `retention_days`. Instances with a live `df` row are not orphans. +- **Prunes old execution generations.** Loops start a new execution generation + on each iteration. Reconciliation removes historical generations and their + history events once their completion is older than `retention_days`, including + those belonging to still-running workflows and nested loops in any origin. + The current execution and any running execution are always preserved. This + does not remove the workflow's `df` rows or its current results, but old + generations will no longer appear in execution-history inspection. After the corresponding engine work, history, and origin metadata are gone, reconciliation removes the protected submission record and frees its retained @@ -2570,6 +2576,11 @@ for at most one 1,000-candidate page, then other origins get a turn before a bus origin resumes. Only age-expired or excess-over-10,000 rows consume the retention candidate budget. Legacy orphan candidates exclude satellite roots and children in SQL before their bounded page is loaded into memory. +Execution pruning likewise pages through at most 1,000 eligible engine instances +per slice; an instance that fails pruning does not block later candidates and is +retried on the next scan. Each pruning slice has a 15-second overall deadline. +The page limit bounds instance candidates, not the number of old generations or +history events an individual provider pruning call may delete. Metadata queries have server-side deadlines and satellite operations have a 15-second overall deadline. The candidate limit bounds materialized rows, not @@ -2587,12 +2598,13 @@ Two Postmaster-context GUCs govern it (set in `postgresql.conf`, restart to appl pg_durable.reconcile_interval = 3600 # Days a terminal instance is retained before reconciliation removes it (and its -# engine record); also the age bound for reclaiming orphaned engine records. -# 0 removes terminal instances as soon as the next pass runs. +# engine record); also the age bound for orphaned engine records and completed +# historical execution generations. The current execution is never pruned. +# 0 makes cleanup eligible on the next pass. pg_durable.retention_days = 30 ``` -Retention combines that window with a fixed hard cap: +Terminal-instance retention combines that window with a fixed hard cap: - **Hard cap per origin — at most 10,000 terminal instances are retained, regardless of age.** The newest 10,000 terminal instances are kept; any beyond that are @@ -2604,9 +2616,20 @@ Equivalently, a terminal instance is retained only while it is **both** among th newest 10,000 terminal instances **and** younger than `retention_days`; otherwise it is eligible for removal. +Historical execution generations use only the age window, not a count cap. +Pruning bounds how long completed history is retained, not its byte size: a +busy workflow can still retain many generations within the window. An execution +that has not continued into a new generation retains its current history. +Setting `reconcile_interval = 0` also disables execution pruning. + +On binary upgrade, the worker recovers missing completion timestamps for older +continued generations from their recorded terminal history events. If an event +is missing, it logs a migration warning and leaves that generation ineligible for +age-based pruning rather than guessing its age. + Notes: -- "Age" is measured from `completed_at` when set (instances that reached +- Terminal-instance "age" is measured from `completed_at` when set (instances that reached `completed`), otherwise from `created_at`. `updated_at` is intentionally **not** used, because it is user-writable and would let a low-privilege user influence what is removed. diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 73e62319..25ab8ba4 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -473,6 +473,24 @@ the shipped 0.2.7 to 0.2.8 script must remain byte-identical. See [deferred lifetime guarantees](multi-database-installation.md#release-blockers) for the separate runtime shutdown and server-side cancellation limitations. +#### Historical execution pruning + +- **B1:** background reconciliation prunes completed non-current execution + generations using `pg_durable.retention_days`, including running workflows and + nested loops in every origin. The query uses existing provider tables, so no + extension upgrade DDL or new `df` schema detection is needed. +- **Provider migration:** the worker's startup `ApplyAll` applies migration `0024` + before starting the runtime. It timestamps `ContinuedAsNew` executions and + backfills older NULL timestamps from terminal history events. Unrecoverable + timestamps remain NULL with a warning; those generations are not age-pruned. + Stop any other runtimes sharing the provider schema during this one-time + backfill. Existing histories may make startup slower. +- **Replay:** no orchestration code, durable-operation inputs, or continuation + payloads change. Pruning preserves the current and running executions. +- **Release blocker:** the provider is temporarily pinned to the timestamp-fix + git revision in `Cargo.toml`. Replace the `DONOTLAND` pin with a released + compatible version before landing. + #### Typed HTTP endpoints - Adds composite type `df.http_endpoint(server text, path text)`, diff --git a/scripts/test-e2e-local.sh b/scripts/test-e2e-local.sh index 7f3515e1..2aa099d0 100755 --- a/scripts/test-e2e-local.sh +++ b/scripts/test-e2e-local.sh @@ -221,7 +221,7 @@ phase_for_test() { 46_connection_limit_startup_validation) echo "connlimit-startup" ;; - 54_reconcile_orphans|76_multi_database_reconcile|95_start_limits_rollback_cleanup|96_start_limits_failed_metadata_cleanup) + 54_reconcile_orphans|76_multi_database_reconcile|89_execution_pruning|95_start_limits_rollback_cleanup|96_start_limits_failed_metadata_cleanup) echo "reconcile" ;; 88_multi_database_maintenance_backlog) diff --git a/src/lib.rs b/src/lib.rs index 617763b2..422ba28e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,11 +55,13 @@ pub static LIST_INSTANCES_MAX_LIMIT: GucSetting = GucSetting::::new(10 /// Days a terminal ('completed'/'failed'/'cancelled') instance is retained /// before reconciliation removes it and its engine record. The same age bound /// governs when orphaned engine records left by a rolled-back `df.start()` are -/// reclaimed. `0` removes terminal instances as soon as the next pass runs. +/// reclaimed and completed historical executions are pruned. The current +/// execution is always preserved. `0` makes cleanup eligible on the next pass. pub static RETENTION_DAYS: GucSetting = GucSetting::::new(30); /// Seconds between background reconciliation passes. Each pass removes expired -/// terminal instances and reclaims orphaned engine records. `0` disables it. +/// terminal instances, reclaims orphaned engine records, and prunes historical +/// executions. `0` disables it. pub static RECONCILE_INTERVAL: GucSetting = GucSetting::::new(3600); /// When `false`, the worker log omits the SQL text of executed workflow nodes. @@ -314,8 +316,8 @@ pub extern "C-unwind" fn _PG_init() { GucRegistry::define_int_guc( c"pg_durable.retention_days", - c"Days a terminal instance is retained before reconciliation removes it", - c"Background reconciliation removes 'completed'/'failed'/'cancelled' df.instances rows (and their engine records) once they are older than this many days, subject to a fixed hard cap on the number retained. The same age bound governs when orphaned engine records left by a rolled-back df.start() are reclaimed. 0 removes terminal instances as soon as the next pass runs.", + c"Days to retain terminal instances and completed historical executions", + c"Background reconciliation removes 'completed'/'failed'/'cancelled' df.instances rows (and their engine records) once they are older than this many days, subject to a fixed hard cap on the number retained. The same age bound governs orphaned engine records and completed historical executions, including those of running workflows. Pruning always preserves the current execution. 0 makes cleanup eligible on the next pass.", &RETENTION_DAYS, 0, 36500, @@ -326,7 +328,7 @@ pub extern "C-unwind" fn _PG_init() { GucRegistry::define_int_guc( c"pg_durable.reconcile_interval", c"Idle seconds between reconciliation scans (0 disables reconciliation)", - c"Unfinished bounded pages continue promptly between scans. Reconciliation removes expired terminal instances, orphaned engine records, and confirmed removed registrations with no engine work. Set to 0 to disable all reconciliation.", + c"Unfinished bounded pages continue promptly between scans. Reconciliation removes expired terminal instances, orphaned engine records, completed historical executions, and confirmed removed registrations with no engine work. Set to 0 to disable all reconciliation.", &RECONCILE_INTERVAL, 0, 86400, diff --git a/src/types.rs b/src/types.rs index df5aa6db..516fbd99 100644 --- a/src/types.rs +++ b/src/types.rs @@ -98,7 +98,8 @@ pub fn get_new_transaction_start_timeout() -> Duration { } /// Days a terminal instance is retained before reconciliation removes it and its -/// engine record; also the age bound for reclaiming orphaned engine records. +/// engine record; also the age bound for orphaned engine records and completed +/// historical executions. pub fn get_retention_days() -> i32 { crate::RETENTION_DAYS.get() } diff --git a/src/worker.rs b/src/worker.rs index 4bdbca18..d9b5540e 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use std::time::Duration; use duroxide::runtime; -use duroxide::{Client, ClientError, InstanceFilter}; +use duroxide::{Client, ClientError, InstanceFilter, PruneOptions, PruneResult}; use duroxide_pg::PostgresProvider; use tracing_subscriber::EnvFilter; @@ -1149,6 +1149,7 @@ async fn run_until_extension_dropped_or_shutdown( let mut origin_cursor = OriginRetentionCursor::default(); let mut engine_cursor = String::new(); let mut legacy_cursor = String::new(); + let mut execution_cursor = String::new(); let mut registration_cursor = (0i64, uuid::Uuid::nil()); let mut submission_cursor = String::new(); let mut pending = MaintenanceProgress::new(); @@ -1232,6 +1233,27 @@ async fn run_until_extension_dropped_or_shutdown( } } let retention = Duration::from_secs(retention_days as u64 * 86_400); + if pending.executions { + let previous_cursor = execution_cursor.clone(); + pending.executions = match tokio::time::timeout( + ORIGIN_OPERATION_TIMEOUT, + prune_old_executions( + maintenance_pool, &client, &schema, + retention_cutoff_ms(retention), &mut execution_cursor, + ), + ).await { + Ok(Ok(more)) => more, + Ok(Err(error)) => { + log!("pg_durable: execution pruning deferred: {error}"); + execution_cursor != previous_cursor + }, + Err(_) => { + log!("pg_durable: execution pruning timed out"); + execution_cursor != previous_cursor + }, + }; + if !pending.executions { execution_cursor.clear(); } + } if pending.legacy { match reclaim_orphaned_instances(maintenance_pool, &client, &schema, retention, &mut legacy_cursor).await { Ok(reclaimed) if reclaimed > 0 => { log!("pg_durable: reclaimed {reclaimed} orphaned engine record(s)"); @@ -1313,6 +1335,7 @@ struct MaintenanceProgress { engine: bool, registrations: bool, submissions: bool, + executions: bool, } impl MaintenanceProgress { @@ -1324,6 +1347,7 @@ impl MaintenanceProgress { engine: true, registrations: true, submissions: true, + executions: true, } } @@ -1334,6 +1358,7 @@ impl MaintenanceProgress { || self.engine || self.registrations || self.submissions + || self.executions } fn finish_slice(&mut self, origins: &mut OriginRetentionCursor) -> bool { @@ -1927,6 +1952,69 @@ fn retention_cutoff_ms(retention: Duration) -> u64 { since_epoch.saturating_sub(retention).as_millis() as u64 } +async fn prune_old_executions( + pool: &sqlx::PgPool, + client: &Client, + schema: &str, + completed_before: u64, + cursor: &mut String, +) -> Result { + let mut tx = pool.begin().await.map_err(|error| error.to_string())?; + crate::origin::configure_metadata_transaction(&mut tx) + .await + .map_err(|error| error.to_string())?; + // The bulk API has no cursor; its limit can repeatedly select the same + // instances. Page only eligible histories, including child orchestrations. + let candidates: Vec = sqlx::query_scalar(&format!( + "SELECT i.instance_id FROM {schema}.instances i + WHERE i.instance_id OPERATOR(pg_catalog.>) $1 + AND EXISTS ( + SELECT 1 FROM {schema}.executions e + WHERE e.instance_id OPERATOR(pg_catalog.=) i.instance_id + AND e.execution_id OPERATOR(pg_catalog.<>) i.current_execution_id + AND e.status OPERATOR(pg_catalog.<>) 'Running' + AND e.completed_at OPERATOR(pg_catalog.<) + pg_catalog.to_timestamp($2::bigint OPERATOR(pg_catalog./) 1000.0) + ) + ORDER BY i.instance_id LIMIT $3" + )) + .bind(cursor.as_str()) + .bind(completed_before as i64) + .bind(i64::from(RECLAIM_BATCH)) + .fetch_all(&mut *tx) + .await + .map_err(|error| format!("select execution pruning candidates: {error}"))?; + tx.commit().await.map_err(|error| error.to_string())?; + let more = candidates.len() == RECLAIM_BATCH as usize; + let mut total = PruneResult::default(); + for id in candidates { + // Advance before I/O so an error or timeout cannot starve later IDs. + *cursor = id; + let result = client + .prune_executions( + cursor, + PruneOptions { + completed_before: Some(completed_before), + ..Default::default() + }, + ) + .await + .map_err(|error| format!("prune executions for {cursor}: {error}"))?; + total.instances_processed += result.instances_processed; + total.executions_deleted += result.executions_deleted; + total.events_deleted += result.events_deleted; + } + if total.executions_deleted > 0 { + log!( + "pg_durable: pruned {} old execution(s) and {} history event(s) across {} instance(s)", + total.executions_deleted, + total.events_deleted, + total.instances_processed + ); + } + Ok(more) +} + /// From the engine's `Failed` instance ids and the set that still have a /// df.instances row, select the orphans to reclaim: those with no df row and that /// are not sub-orchestrations. Both legacy engine-named children and current @@ -2459,6 +2547,167 @@ mod tests { }); } + #[pg_test] + fn execution_pruning_retention_and_failure_progress() { + let admin = Spi::get_one::("SELECT current_user::text") + .unwrap() + .unwrap(); + let database = Spi::get_one::("SELECT current_database()::text") + .unwrap() + .unwrap(); + let mut url = url::Url::parse(&postgres_connection_string()).unwrap(); + url.set_username(&admin).unwrap(); + url.set_path(&database); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let store = Arc::new( + PostgresProvider::new_with_config(worker_provider_config( + url.as_str(), + "execution_pruning_test", + )) + .await + .unwrap(), + ); + let pool = store.pool(); + let client = Client::new(store.clone()); + sqlx::raw_sql( + "INSERT INTO execution_pruning_test.instances + (instance_id, orchestration_name, current_execution_id, created_at, updated_at) + SELECT 'a' || lpad(n::text,8,'0'), 'test', 1, now(), now() + FROM generate_series(1,1100) n; + INSERT INTO execution_pruning_test.executions + (instance_id,execution_id,status,completed_at,started_at) + SELECT instance_id,1,'Completed','2000-01-01 UTC','2000-01-01 UTC' + FROM execution_pruning_test.instances; + INSERT INTO execution_pruning_test.instances + (instance_id, orchestration_name, current_execution_id, created_at, updated_at) + SELECT 'b' || lpad(n::text,8,'0'), 'test', 6, now(), now() + FROM generate_series(1,1001) n; + INSERT INTO execution_pruning_test.instances + (instance_id, orchestration_name, current_execution_id, created_at, updated_at) + SELECT id,'test',6,now(),now() FROM (VALUES + ('pgdf-42-00000000000000000000000000000001-cafe0001'), + ('pgdf-42-00000000000000000000000000000001-cafe0001::1::child'), + ('sub::legacy'), ('cafe0001::1::child'), ('zterminal') + ) AS cases(id); + INSERT INTO execution_pruning_test.executions + (instance_id,execution_id,status,completed_at,started_at) + SELECT i.instance_id,e.id,e.status,e.completed,'2000-01-01 UTC' + FROM execution_pruning_test.instances i CROSS JOIN (VALUES + (1,'ContinuedAsNew','2023-12-31 23:59:59.999 UTC'::timestamptz), + (2,'ContinuedAsNew','2024-01-01 UTC'::timestamptz), + (3,'Completed','2024-01-02 UTC'::timestamptz), + (4,'Running','2000-01-01 UTC'::timestamptz), + (5,'Completed',NULL), (6,'Running',NULL) + ) AS e(id,status,completed) WHERE i.current_execution_id=6; + UPDATE execution_pruning_test.executions + SET status='Completed',completed_at='2000-01-01 UTC' + WHERE instance_id='zterminal' AND execution_id=6; + INSERT INTO execution_pruning_test.history + (instance_id,execution_id,event_id,event_type,event_data,created_at) + SELECT instance_id,execution_id,1,'test','{}',now() + FROM execution_pruning_test.executions; + CREATE FUNCTION execution_pruning_test.reject_prune() RETURNS trigger + LANGUAGE plpgsql AS $fn$ BEGIN + IF OLD.instance_id='b00000001' THEN + RAISE EXCEPTION 'intentional pruning failure'; + END IF; + RETURN OLD; + END $fn$; + CREATE TRIGGER reject_prune BEFORE DELETE + ON execution_pruning_test.executions FOR EACH ROW + EXECUTE FUNCTION execution_pruning_test.reject_prune();", + ) + .execute(pool) + .await + .unwrap(); + + let cutoff = 1_704_067_200_000; + let mut cursor = String::new(); + let error = prune_old_executions( + pool, &client, "execution_pruning_test", cutoff, &mut cursor, + ) + .await + .unwrap_err(); + assert!(error.contains("intentional pruning failure"), "{error}"); + assert_eq!(cursor, "b00000001"); + assert!(prune_old_executions( + pool, &client, "execution_pruning_test", cutoff, &mut cursor, + ) + .await + .unwrap()); + assert_eq!(cursor, "b00001001"); + assert!(!prune_old_executions( + pool, &client, "execution_pruning_test", cutoff, &mut cursor, + ) + .await + .unwrap()); + assert_eq!(cursor, "zterminal"); + + sqlx::query("DROP TRIGGER reject_prune ON execution_pruning_test.executions") + .execute(pool) + .await + .unwrap(); + cursor.clear(); + assert!(!prune_old_executions( + pool, &client, "execution_pruning_test", cutoff, &mut cursor, + ) + .await + .unwrap()); + assert_eq!(cursor, "b00000001", "retry the failed instance next cycle"); + let counts: (i64, i64, i64) = sqlx::query_as( + "SELECT + (SELECT count(*) FROM execution_pruning_test.instances), + (SELECT count(*) FROM execution_pruning_test.executions), + (SELECT count(*) FROM execution_pruning_test.history)", + ) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(counts, (2106, 6130, 6130)); + let survivors: Vec = sqlx::query_scalar( + "SELECT DISTINCT e.execution_id FROM execution_pruning_test.executions e + JOIN execution_pruning_test.instances i USING(instance_id) + WHERE i.current_execution_id=6 ORDER BY e.execution_id", + ) + .fetch_all(pool) + .await + .unwrap(); + assert_eq!(survivors, [2, 3, 4, 5, 6]); + assert!(!prune_old_executions( + pool, &client, "execution_pruning_test", cutoff, &mut String::new(), + ) + .await + .unwrap(), "an exhausted scan must not select protected histories"); + sqlx::query("DROP SCHEMA execution_pruning_test CASCADE") + .execute(pool) + .await + .unwrap(); + pool.close().await; + }); + } + + #[test] + fn execution_pruning_keeps_maintenance_cycle_open() { + let mut progress = MaintenanceProgress { + control: false, + legacy: false, + origins: false, + engine: false, + registrations: false, + submissions: false, + executions: true, + }; + let mut origins = OriginRetentionCursor::default(); + assert!(progress.finish_slice(&mut origins)); + progress.executions = false; + assert!(!progress.finish_slice(&mut origins)); + assert!(progress.executions, "pruning must resume on the next scan"); + } + #[pg_test] fn maintenance_queries_bound_candidates_and_preserve_work() { let admin = Spi::get_one::("SELECT current_user::text") @@ -2970,6 +3219,7 @@ mod tests { engine: false, registrations: false, submissions: false, + executions: false, }; assert!(!progress.finish_slice(&mut cursor)); assert!(cursor.enter(origin)); @@ -3116,6 +3366,7 @@ mod tests { engine: true, registrations: false, submissions: false, + executions: false, }; assert!(progress.finish_slice(&mut cursor)); for origin in &origins { diff --git a/tests/e2e/sql/89_execution_pruning.sql b/tests/e2e/sql/89_execution_pruning.sql new file mode 100644 index 00000000..96a9cffe --- /dev/null +++ b/tests/e2e/sql/89_execution_pruning.sql @@ -0,0 +1,191 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +-- Real continue-as-new histories must be pruned without breaking replay of the +-- current generation, including a child loop and a satellite-origin workflow. +CREATE EXTENSION IF NOT EXISTS dblink; +DO $$ +BEGIN + IF current_setting('pg_durable.reconcile_interval') <> '2' + OR current_setting('pg_durable.retention_days') <> '0' THEN + RAISE EXCEPTION 'TEST SETUP ERROR: requires reconcile phase'; + END IF; +END $$; + +CREATE TABLE test_execution_pruning (scenario TEXT PRIMARY KEY, iterations INT NOT NULL); +INSERT INTO test_execution_pruning VALUES ('root',0),('nested',0); +CREATE TEMP TABLE _pruning_cases ( + scenario TEXT PRIMARY KEY, local_id TEXT, engine_id TEXT, + loop_id TEXT, current_history JSONB +); +DROP DATABASE IF EXISTS _e89_pruning WITH (FORCE); +CREATE DATABASE _e89_pruning; +SELECT dblink_connect('pruning', format( + 'host=localhost port=%s dbname=_e89_pruning user=postgres', current_setting('port'))); +SELECT dblink_exec('pruning', $remote$ + CREATE EXTENSION pg_durable; + CREATE TABLE test_execution_pruning (scenario TEXT PRIMARY KEY, iterations INT NOT NULL); + INSERT INTO test_execution_pruning VALUES ('satellite',0); +$remote$); + +DO $$ +DECLARE scenario TEXT; graph TEXT; id TEXT; engine TEXT; +BEGIN + FOREACH scenario IN ARRAY ARRAY['root','nested','satellite'] LOOP + graph := df.loop( + format('UPDATE test_execution_pruning SET iterations=iterations+1 WHERE scenario=%L', scenario) + ~> df.wait_for_signal('pruning-next') + ); + IF scenario = 'nested' THEN graph := 'SELECT 1' ~> graph; END IF; + IF scenario = 'satellite' THEN + SELECT remote.id INTO id FROM dblink('pruning', + format('SELECT df.start(%L,%L)',graph,'execution-pruning')) + AS remote(id TEXT); + SELECT prefix || id INTO engine FROM dblink('pruning', $remote$ + SELECT 'pgdf-' || (SELECT oid::text FROM pg_database WHERE datname=current_database()) + || '-' || (SELECT replace(id::text,'-','') FROM df._installation) || '-' + $remote$) AS remote(prefix TEXT); + ELSE + id := df.start(graph,'execution-pruning'); + engine := id; + END IF; + INSERT INTO _pruning_cases(scenario,local_id,engine_id) VALUES (scenario,id,engine); + END LOOP; +END $$; + +CREATE FUNCTION pg_temp.wait_pruning_generation(generation BIGINT) RETURNS void +LANGUAGE plpgsql AS $$ +DECLARE item RECORD; waiting_id TEXT; iterations INT; +BEGIN + FOR item IN SELECT * FROM _pruning_cases LOOP + waiting_id := NULL; + FOR attempt IN 1..300 LOOP + EXECUTE format( + 'SELECT i.instance_id FROM %1$I.instances i + WHERE (i.instance_id=$1 OR i.parent_instance_id=$1) + AND i.current_execution_id=$2 AND EXISTS ( + SELECT 1 FROM %1$I.history h + WHERE h.instance_id=i.instance_id AND h.execution_id=i.current_execution_id + AND h.event_data::jsonb->>''type''=''ExternalSubscribed'' + AND h.event_data::jsonb->>''name''=''pruning-next'')', + df.duroxide_schema()) + INTO waiting_id USING item.engine_id,generation; + EXIT WHEN waiting_id IS NOT NULL; + PERFORM pg_sleep(0.1); + END LOOP; + IF waiting_id IS NULL THEN + RAISE EXCEPTION 'TEST FAILED: % did not reach waiting generation %',item.scenario,generation; + END IF; + UPDATE _pruning_cases SET loop_id=waiting_id WHERE scenario=item.scenario; + IF item.scenario='satellite' THEN + SELECT n INTO iterations FROM dblink('pruning', + 'SELECT iterations FROM test_execution_pruning') AS remote(n INT); + ELSE + SELECT p.iterations INTO iterations FROM test_execution_pruning p WHERE p.scenario=item.scenario; + END IF; + IF iterations IS DISTINCT FROM generation THEN + RAISE EXCEPTION 'TEST FAILED: % ran % iterations in generation %',item.scenario,iterations,generation; + END IF; + END LOOP; +END $$; + +CREATE FUNCTION pg_temp.advance_pruning_generation() RETURNS void LANGUAGE plpgsql AS $$ +DECLARE item RECORD; +BEGIN + FOR item IN SELECT * FROM _pruning_cases LOOP + IF item.scenario='satellite' THEN + PERFORM result FROM dblink('pruning', + format('SELECT df.signal(%L,''pruning-next'')',item.local_id)) AS remote(result TEXT); + ELSE + PERFORM df.signal(item.local_id,'pruning-next'); + END IF; + END LOOP; +END $$; + +SELECT pg_temp.wait_pruning_generation(1); +SELECT pg_temp.advance_pruning_generation(); +SELECT pg_temp.wait_pruning_generation(2); +SELECT pg_temp.advance_pruning_generation(); +SELECT pg_temp.wait_pruning_generation(3); + +DO $$ +DECLARE item RECORD; snapshot JSONB; +BEGIN + FOR item IN SELECT * FROM _pruning_cases LOOP + EXECUTE format( + 'SELECT jsonb_agg(to_jsonb(h) ORDER BY event_id) FROM %I.history h + WHERE instance_id=$1 AND execution_id=3',df.duroxide_schema()) + INTO snapshot USING item.loop_id; + IF snapshot IS NULL THEN RAISE EXCEPTION 'TEST FAILED: no current history for %',item.scenario; END IF; + UPDATE _pruning_cases SET current_history=snapshot WHERE scenario=item.scenario; + END LOOP; +END $$; + +DO $$ +DECLARE item RECORD; remaining INT; snapshot JSONB; live BOOLEAN; +BEGIN + FOR item IN SELECT * FROM _pruning_cases LOOP + FOR attempt IN 1..300 LOOP + EXECUTE format( + 'SELECT count(*) FROM %I.executions WHERE instance_id=$1 AND execution_id<3', + df.duroxide_schema()) INTO remaining USING item.loop_id; + EXIT WHEN remaining=0; + PERFORM pg_sleep(0.1); + END LOOP; + IF remaining<>0 THEN + RAISE EXCEPTION 'TEST FAILED: % still has % old executions',item.scenario,remaining; + END IF; + EXECUTE format('SELECT count(*) FROM %I.history WHERE instance_id=$1 AND execution_id<3', + df.duroxide_schema()) INTO remaining USING item.loop_id; + IF remaining<>0 THEN RAISE EXCEPTION 'TEST FAILED: old history survived for %',item.scenario; END IF; + EXECUTE format( + 'SELECT jsonb_agg(to_jsonb(h) ORDER BY event_id) FROM %I.history h + WHERE instance_id=$1 AND execution_id=3',df.duroxide_schema()) + INTO snapshot USING item.loop_id; + IF snapshot IS DISTINCT FROM item.current_history THEN + RAISE EXCEPTION 'TEST FAILED: current history changed for %',item.scenario; + END IF; + IF item.scenario='satellite' THEN + SELECT result INTO live FROM dblink('pruning',format( + 'SELECT df.status(%1$L)=''running'' AND EXISTS(SELECT 1 FROM df.nodes WHERE instance_id=%1$L)', + item.local_id)) AS remote(result BOOLEAN); + ELSE + SELECT df.status(item.local_id)='running' + AND EXISTS(SELECT 1 FROM df.nodes WHERE instance_id=item.local_id) INTO live; + END IF; + IF live IS DISTINCT FROM true THEN + RAISE EXCEPTION 'TEST FAILED: pruning removed live workflow state for %',item.scenario; + END IF; + END LOOP; +END $$; + +SELECT pg_temp.advance_pruning_generation(); +SELECT pg_temp.wait_pruning_generation(4); + +DO $$ +DECLARE item RECORD; running BOOLEAN; +BEGIN + FOR item IN SELECT * FROM _pruning_cases LOOP + IF item.scenario='satellite' THEN + PERFORM result FROM dblink('pruning', + format('SELECT df.cancel(%L)',item.local_id)) AS remote(result TEXT); + ELSE + PERFORM df.cancel(item.local_id); + END IF; + FOR attempt IN 1..300 LOOP + EXECUTE format( + 'SELECT EXISTS(SELECT 1 FROM %1$I.instances i JOIN %1$I.executions e + ON e.instance_id=i.instance_id AND e.execution_id=i.current_execution_id + WHERE (i.instance_id=$1 OR i.parent_instance_id=$1) AND e.status=''Running'')', + df.duroxide_schema()) INTO running USING item.engine_id; + EXIT WHEN NOT running; + PERFORM pg_sleep(0.1); + END LOOP; + IF running THEN RAISE EXCEPTION 'TEST FAILED: cleanup left % running',item.scenario; END IF; + END LOOP; +END $$; +SELECT dblink_disconnect('pruning'); +DROP DATABASE _e89_pruning WITH (FORCE); +DROP TABLE _pruning_cases; +DROP TABLE test_execution_pruning; +SELECT 'TEST PASSED: execution pruning' AS result; From 1a427ab76df0c1bba86820244b0ae64990a57778 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Thu, 15 Oct 2026 17:07:34 +0000 Subject: [PATCH 2/8] Robustness improvements --- Cargo.lock | 9 +- Cargo.toml | 4 +- USER_GUIDE.md | 23 ++- docs/upgrade-testing.md | 25 ++- scripts/test-e2e-docker.sh | 1 + src/monitoring.rs | 178 ++++++++++++++++---- src/worker.rs | 329 +++++++++++++++++++++++++++---------- 7 files changed, 432 insertions(+), 137 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2ec31fbb..2c98c614 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -712,9 +712,9 @@ checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" [[package]] name = "duroxide" -version = "0.1.30" +version = "0.1.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92b17ebe10f702644ad65a9c624b4060023955b4cd145c23ed4c6942b17fdac9" +checksum = "75573b31ca6fd3918acbe9af7c0ed32d3b181675cd9b01d9e09f2b8436c3d192" dependencies = [ "async-trait", "futures", @@ -731,8 +731,8 @@ dependencies = [ [[package]] name = "duroxide-pg" -version = "0.1.35" -source = "git+https://github.com/microsoft/duroxide-pg?branch=thomcc%2Fcontinue_as_new_completed_at#e7773b34bee0127efafd796cdf8c732cd537cd2d" +version = "0.1.36" +source = "git+https://github.com/microsoft/duroxide-pg?branch=thomcc%2Fcontinue_as_new_completed_at#c89f75484c3da68599d2de3381b1cbca7f901c57" dependencies = [ "anyhow", "async-trait", @@ -3319,6 +3319,7 @@ dependencies = [ "bytes", "futures-core", "futures-sink", + "futures-util", "libc", "pin-project-lite", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 87e62f34..77d994f1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,8 +36,8 @@ uuid = { version = "1.0", features = ["v4", "serde"] } percent-encoding = "2.3" # duroxide integration -duroxide = "=0.1.30" -# DONOTLAND: replace this temporary continue_as_new_completed_at pin with a crates.io release. +duroxide = "0.1.32" +# DONOTLAND: requires the timestamp and batch-pruning fixes; replace with a crates.io release. duroxide-pg = { git = "https://github.com/microsoft/duroxide-pg", branch = "thomcc/continue_as_new_completed_at" } tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] } diff --git a/USER_GUIDE.md b/USER_GUIDE.md index bd019199..5bdd772c 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -2576,11 +2576,15 @@ for at most one 1,000-candidate page, then other origins get a turn before a bus origin resumes. Only age-expired or excess-over-10,000 rows consume the retention candidate budget. Legacy orphan candidates exclude satellite roots and children in SQL before their bounded page is loaded into memory. -Execution pruning likewise pages through at most 1,000 eligible engine instances -per slice; an instance that fails pruning does not block later candidates and is -retried on the next scan. Each pruning slice has a 15-second overall deadline. -The page limit bounds instance candidates, not the number of old generations or -history events an individual provider pruning call may delete. +Execution pruning scans at most 1,000 execution keys per slice, including young +and protected generations. Each batch deletes at most those 1,000 generations; +this is a work limit, not a generation-count retention policy. Failed pages are +logged, skipped for the rest of the scan, and retried on the next scan so they +cannot block later pages. Pruning uses the maintenance connection, not the +workflow runtime's connection pool, with server-side five-second statement and +1.5-second lock deadlines. A timeout rolls back the batch. The number of history +events in one generation is not capped; a batch that cannot finish within the +deadline is deferred. Metadata queries have server-side deadlines and satellite operations have a 15-second overall deadline. The candidate limit bounds materialized rows, not @@ -2620,12 +2624,19 @@ Historical execution generations use only the age window, not a count cap. Pruning bounds how long completed history is retained, not its byte size: a busy workflow can still retain many generations within the window. An execution that has not continued into a new generation retains its current history. +Completed child instances created by earlier loop iterations retain their current +execution until the owning workflow is removed. Setting `reconcile_interval = 0` also disables execution pruning. On binary upgrade, the worker recovers missing completion timestamps for older continued generations from their recorded terminal history events. If an event is missing, it logs a migration warning and leaves that generation ineligible for -age-based pruning rather than guessing its age. +age-based pruning rather than guessing its age. This one-time backfill runs +before the worker becomes ready and can delay workflow processing on a large +history store. Allow sufficient time for it during upgrade; a role or database +`statement_timeout` shorter than the backfill will cause startup to retry rather +than finish. Adjust the worker's migration-session timeout for the upgrade if +necessary. Notes: diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 25ab8ba4..a8ebc73f 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -479,17 +479,34 @@ for the separate runtime shutdown and server-side cancellation limitations. generations using `pg_durable.retention_days`, including running workflows and nested loops in every origin. The query uses existing provider tables, so no extension upgrade DDL or new `df` schema detection is needed. -- **Provider migration:** the worker's startup `ApplyAll` applies migration `0024` +- **Provider migrations:** the worker's startup `ApplyAll` applies migration `0024` before starting the runtime. It timestamps `ContinuedAsNew` executions and backfills older NULL timestamps from terminal history events. Unrecoverable timestamps remain NULL with a warning; those generations are not age-pruned. Stop any other runtimes sharing the provider schema during this one-time - backfill. Existing histories may make startup slower. + backfill. Existing histories may make startup slower; any worker-role/database + statement timeout must permit that backfill to finish, or startup retries it. + This dependency pair also includes `0023` (partial indexes), `0025` (fetch + leases), and `0026` (cursor-based batch pruning). Readiness schema version `3` + prevents backends from treating a pre-migration worker marker as ready. +- **Bounded maintenance:** each slice scans at most 1,000 execution keys, including + protected rows. The provider owns deletion predicates and atomically prunes + that page on the maintenance connection under server-side deadlines. Failed + pages are revisited next scan; runtime provider connections are not borrowed. +- **Inspection:** `df.instance_executions()` reads execution details and event + counts in one MVCC snapshot, so concurrent pruning does not produce spurious + per-generation not-found errors. Genuine query failures still raise errors. - **Replay:** no orchestration code, durable-operation inputs, or continuation payloads change. Pruning preserves the current and running executions. +- **Dependency pair:** duroxide `0.1.32` is required by the provider's current + branch. Its shutdown operation now cancels and joins owned tasks; zero means + immediate cancellation rather than a drain grace period. Validate the real + previous-binary lifecycle and worker shutdown when updating this pair. - **Release blocker:** the provider is temporarily pinned to the timestamp-fix - git revision in `Cargo.toml`. Replace the `DONOTLAND` pin with a released - compatible version before landing. + git branch in `Cargo.toml`. The local review fixes add migration `0026`; until + they are pushed, validation requires a local Cargo patch override. Refresh the + git lockfile after publishing those fixes and replace the `DONOTLAND` pin with + a compatible release before landing. #### Typed HTTP endpoints diff --git a/scripts/test-e2e-docker.sh b/scripts/test-e2e-docker.sh index 2c3d0901..d11fba8f 100755 --- a/scripts/test-e2e-docker.sh +++ b/scripts/test-e2e-docker.sh @@ -52,6 +52,7 @@ SKIP_TESTS=( "95_start_limits_rollback_cleanup" "96_start_limits_failed_metadata_cleanup" "88_multi_database_maintenance_backlog" + "89_execution_pruning" # Require the force-drop phase (one execution slot, reconciliation disabled). "78_multi_database_force_drop" "79_multi_database_remote_origin" diff --git a/src/monitoring.rs b/src/monitoring.rs index 649cceaa..428fb0d4 100644 --- a/src/monitoring.rs +++ b/src/monitoring.rs @@ -719,8 +719,8 @@ pub fn instance_info( /// /// Distinguishes "this instance genuinely has no execution history yet" (empty /// rowset) from "the execution-history lookup failed" (explicit error). The -/// latter — failing to build the runtime, connect to the duroxide store, list -/// executions, or fetch a specific execution's info — now raises an error +/// latter — failing to build the runtime, connect to the duroxide store, or read +/// execution details — raises an error /// instead of being silently swallowed into an empty rowset. A completed /// instance always has at least one execution row, so an empty result for one /// previously masked a real lookup failure. See issue #168. @@ -779,38 +779,9 @@ pub fn instance_executions( rt.block_on(async { let store = new_backend_provider(&pg_conn_str, provider_schema).await?; - let client = Client::new(store); - - let execution_ids = client - .list_executions(&engine_id) + read_execution_snapshot(store.pool(), provider_schema, &engine_id, limit_count) .await - .map_err(|e| format!("failed to list executions: {e:?}"))?; - - let mut sorted_ids: Vec<_> = execution_ids.into_iter().collect(); - sorted_ids.sort_by(|a, b| b.cmp(a)); - let limited: Vec<_> = sorted_ids.into_iter().take(limit_count as usize).collect(); - - let mut rows = Vec::new(); - for exec_id in limited { - let info = client - .get_execution_info(&engine_id, exec_id) - .await - .map_err(|e| format!("failed to fetch info for execution {exec_id}: {e:?}"))?; - - let duration_ms = info - .completed_at - .map(|end| end.saturating_sub(info.started_at)) - .unwrap_or(0); - - rows.push(( - info.execution_id as i64, - info.status, - info.event_count as i64, - duration_ms as i64, - info.output, - )); - } - Ok(rows) + .map_err(|error| format!("failed to read executions: {error}")) }); match results { @@ -819,6 +790,147 @@ pub fn instance_executions( } } +async fn read_execution_snapshot( + pool: &sqlx::PgPool, + schema: &str, + instance_id: &str, + limit: i32, +) -> Result)>, sqlx::Error> { + let schema = format!("\"{}\"", schema.replace('"', "\"\"")); + // One statement gives details and event counts the same MVCC snapshot, + // including when maintenance removes older generations during this read. + let rows: Vec<( + i64, + String, + i64, + chrono::DateTime, + Option>, + Option, + )> = sqlx::query_as(&format!( + "SELECT e.execution_id, e.status, + (SELECT pg_catalog.count(*) FROM {schema}.history h + WHERE h.instance_id OPERATOR(pg_catalog.=) e.instance_id + AND h.execution_id OPERATOR(pg_catalog.=) e.execution_id), + e.started_at, e.completed_at, e.output + FROM ( + SELECT instance_id, execution_id, status, started_at, completed_at, output + FROM {schema}.executions + WHERE instance_id OPERATOR(pg_catalog.=) $1 + ORDER BY execution_id DESC LIMIT $2 + ) e ORDER BY e.execution_id DESC" + )) + .bind(instance_id) + .bind(limit) + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .map(|(id, status, events, started, completed, output)| { + let duration = completed + .map(|end| { + end.timestamp_millis() + .saturating_sub(started.timestamp_millis()) + .max(0) + }) + .unwrap_or(0); + (id, status, events, duration, output) + }) + .collect()) +} + +#[cfg(any(test, feature = "pg_test"))] +#[pg_schema] +mod tests { + use super::*; + use sqlx::Connection; + use std::str::FromStr; + use std::time::Duration; + + #[pg_test] + fn execution_snapshot_survives_concurrent_pruning() { + let admin = Spi::get_one::("SELECT current_user::text") + .unwrap() + .unwrap(); + let database = Spi::get_one::("SELECT current_database()::text") + .unwrap() + .unwrap(); + tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async { + let mut control = crate::origin::metadata_test_connection(&admin, &database).await; + let pool = sqlx::postgres::PgPoolOptions::new().max_connections(1).connect_with( + sqlx::postgres::PgConnectOptions::from_str(&postgres_connection_string()).unwrap() + .username(&admin).database(&database), + ).await.unwrap(); + sqlx::raw_sql( + "CREATE SCHEMA execution_snapshot; + CREATE TABLE execution_snapshot.rows(instance_id text,execution_id bigint,status text, + started_at timestamptz,completed_at timestamptz,output text); + CREATE TABLE execution_snapshot.history(instance_id text,execution_id bigint); + INSERT INTO execution_snapshot.rows VALUES + ('live',1,'ContinuedAsNew',to_timestamp(1),to_timestamp(2.25),'old'), + ('live',2,'Running',to_timestamp(3),NULL,NULL); + INSERT INTO execution_snapshot.history VALUES ('live',1),('live',1),('live',2); + CREATE FUNCTION execution_snapshot.gate(status text) RETURNS text LANGUAGE plpgsql AS $fn$ + BEGIN PERFORM pg_advisory_xact_lock(477832); RETURN status; END $fn$; + CREATE VIEW execution_snapshot.executions AS + SELECT instance_id,execution_id,execution_snapshot.gate(status) AS status, + started_at,completed_at,output FROM execution_snapshot.rows; + CREATE FUNCTION execution_snapshot.unexpected(text,text) RETURNS boolean + LANGUAGE plpgsql AS $fn$ BEGIN RAISE EXCEPTION 'snapshot operator trap'; END $fn$; + CREATE OPERATOR execution_snapshot.= (LEFTARG=text,RIGHTARG=text, + FUNCTION=execution_snapshot.unexpected); + SELECT pg_advisory_lock(477832);" + ).execute(&mut control).await.unwrap(); + sqlx::raw_sql( + "SET search_path=execution_snapshot,pg_catalog; + CREATE TEMP TABLE history(instance_id text,execution_id bigint); + CREATE DOMAIN pg_temp.text AS pg_catalog.text CHECK (false);" + ).execute(&pool).await.unwrap(); + let trap = sqlx::query("SELECT 'x'::pg_catalog.text = 'x'::pg_catalog.text") + .execute(&pool).await.unwrap_err(); + assert!(trap.to_string().contains("snapshot operator trap")); + let blocked_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pool).await.unwrap(); + let read = read_execution_snapshot(&pool, "execution_snapshot", "live", 5); + tokio::pin!(read); + let deadline = tokio::time::sleep(Duration::from_secs(3)); + tokio::pin!(deadline); + loop { + tokio::select! { + result = &mut read => panic!("snapshot must wait at test barrier: {result:?}"), + _ = &mut deadline => panic!("snapshot barrier was not reached"), + _ = tokio::time::sleep(Duration::from_millis(20)) => { + let waiting: bool = sqlx::query_scalar("SELECT cardinality(pg_blocking_pids($1)) > 0") + .bind(blocked_pid).fetch_one(&mut control).await.unwrap(); + if waiting { break; } + } + } + } + sqlx::raw_sql( + "BEGIN; + DELETE FROM execution_snapshot.history WHERE execution_id=1; + DELETE FROM execution_snapshot.rows WHERE execution_id=1; + COMMIT; + SELECT pg_advisory_unlock(477832);" + ).execute(&mut control).await.unwrap(); + let rows = tokio::time::timeout(Duration::from_secs(3), &mut read).await.unwrap().unwrap(); + assert_eq!(rows, [ + (2, "Running".into(), 1, 0, None), + (1, "ContinuedAsNew".into(), 2, 1250, Some("old".into())), + ], "details and history counts must use one snapshot"); + assert_eq!(read_execution_snapshot(&pool, "execution_snapshot", "live", 1).await.unwrap(), + [(2, "Running".into(), 1, 0, None)]); + sqlx::query("ALTER TABLE execution_snapshot.history RENAME TO missing_history") + .execute(&mut control).await.unwrap(); + let error = read_execution_snapshot(&pool, "execution_snapshot", "live", 1).await.unwrap_err(); + assert_eq!(error.as_database_error().unwrap().code().as_deref(), Some("42P01"), + "real lookup failures must not become empty history"); + pool.close().await; + sqlx::query("DROP SCHEMA execution_snapshot CASCADE").execute(&mut control).await.unwrap(); + control.close().await.unwrap(); + }); + } +} + /// Get system-wide durable function metrics. /// /// Access is controlled by PostgreSQL function privileges. Roles with ordinary diff --git a/src/worker.rs b/src/worker.rs index d9b5540e..ec64d67d 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use std::time::Duration; use duroxide::runtime; -use duroxide::{Client, ClientError, InstanceFilter, PruneOptions, PruneResult}; +use duroxide::{Client, ClientError, InstanceFilter}; use duroxide_pg::PostgresProvider; use tracing_subscriber::EnvFilter; @@ -1149,7 +1149,7 @@ async fn run_until_extension_dropped_or_shutdown( let mut origin_cursor = OriginRetentionCursor::default(); let mut engine_cursor = String::new(); let mut legacy_cursor = String::new(); - let mut execution_cursor = String::new(); + let mut execution_cursor = (String::new(), 0); let mut registration_cursor = (0i64, uuid::Uuid::nil()); let mut submission_cursor = String::new(); let mut pending = MaintenanceProgress::new(); @@ -1235,24 +1235,17 @@ async fn run_until_extension_dropped_or_shutdown( let retention = Duration::from_secs(retention_days as u64 * 86_400); if pending.executions { let previous_cursor = execution_cursor.clone(); - pending.executions = match tokio::time::timeout( - ORIGIN_OPERATION_TIMEOUT, - prune_old_executions( - maintenance_pool, &client, &schema, - retention_cutoff_ms(retention), &mut execution_cursor, - ), + pending.executions = match prune_old_executions( + maintenance_pool, &schema, + retention_cutoff_ms(retention), &mut execution_cursor, ).await { - Ok(Ok(more)) => more, - Ok(Err(error)) => { + Ok(more) => more, + Err(error) => { log!("pg_durable: execution pruning deferred: {error}"); execution_cursor != previous_cursor }, - Err(_) => { - log!("pg_durable: execution pruning timed out"); - execution_cursor != previous_cursor - }, }; - if !pending.executions { execution_cursor.clear(); } + if !pending.executions { execution_cursor = (String::new(), 0); } } if pending.legacy { match reclaim_orphaned_instances(maintenance_pool, &client, &schema, retention, &mut legacy_cursor).await { Ok(reclaimed) if reclaimed > 0 => { @@ -1905,11 +1898,8 @@ async fn teardown_runtime( const POOL_CLOSE_GRACE: Duration = Duration::from_millis(100); let _ = tokio::time::timeout(POOL_CLOSE_GRACE, duroxide_store.pool().close()).await; - // Some(0) is load-bearing on two duroxide internals: a non-zero timeout - // is an unconditional sleep rather than a deadline, and the zero branch - // returns without setting duroxide's cooperative shutdown_flag. Leaving - // that flag unset is only safe because proc_exit() follows immediately. - // Re-verify both when upgrading duroxide. + // In duroxide 0.1.32, zero signals cancellation, aborts owned tasks, and + // waits for their cancellation cleanup without a drain grace period. duroxide_runtime.shutdown(Some(0)).await; } else { // Live process: the runtime is being restarted after an extension drop @@ -1952,67 +1942,82 @@ fn retention_cutoff_ms(retention: Duration) -> u64 { since_epoch.saturating_sub(retention).as_millis() as u64 } +#[derive(sqlx::FromRow)] +struct ExecutionPruneBatch { + last_instance_id: Option, + last_execution_id: Option, + executions_scanned: i64, + executions_deleted: i64, + events_deleted: i64, +} + async fn prune_old_executions( pool: &sqlx::PgPool, - client: &Client, schema: &str, completed_before: u64, - cursor: &mut String, + cursor: &mut (String, i64), ) -> Result { + let completed_before = i64::try_from(completed_before) + .map_err(|error| format!("invalid pruning cutoff: {error}"))?; let mut tx = pool.begin().await.map_err(|error| error.to_string())?; crate::origin::configure_metadata_transaction(&mut tx) .await .map_err(|error| error.to_string())?; - // The bulk API has no cursor; its limit can repeatedly select the same - // instances. Page only eligible histories, including child orchestrations. - let candidates: Vec = sqlx::query_scalar(&format!( - "SELECT i.instance_id FROM {schema}.instances i - WHERE i.instance_id OPERATOR(pg_catalog.>) $1 - AND EXISTS ( - SELECT 1 FROM {schema}.executions e - WHERE e.instance_id OPERATOR(pg_catalog.=) i.instance_id - AND e.execution_id OPERATOR(pg_catalog.<>) i.current_execution_id - AND e.status OPERATOR(pg_catalog.<>) 'Running' - AND e.completed_at OPERATOR(pg_catalog.<) - pg_catalog.to_timestamp($2::bigint OPERATOR(pg_catalog./) 1000.0) - ) - ORDER BY i.instance_id LIMIT $3" + // Remember the scanned page before deletion, so a failed page cannot starve + // later keys. Neither eligibility nor the number of deletions drives progress. + let scanned_through: Option<(String, i64)> = sqlx::query_as(&format!( + "SELECT instance_id, execution_id FROM ( + SELECT instance_id, execution_id FROM {schema}.executions + WHERE (instance_id, execution_id) OPERATOR(pg_catalog.>) ($1, $2) + ORDER BY instance_id, execution_id LIMIT $3 + ) scanned + ORDER BY instance_id DESC, execution_id DESC LIMIT 1" )) - .bind(cursor.as_str()) - .bind(completed_before as i64) + .bind(&cursor.0) + .bind(cursor.1) .bind(i64::from(RECLAIM_BATCH)) - .fetch_all(&mut *tx) + .fetch_optional(&mut *tx) .await - .map_err(|error| format!("select execution pruning candidates: {error}"))?; + .map_err(|error| format!("page execution pruning keys: {error}"))?; + let Some(scanned_through) = scanned_through else { + tx.commit().await.map_err(|error| error.to_string())?; + return Ok(false); + }; + let after = std::mem::replace(cursor, scanned_through); + // Use the maintenance connection, not the runtime provider pool. PostgreSQL + // enforces the deadlines; await rollback instead of abandoning a live query. + let result = sqlx::query_as::<_, ExecutionPruneBatch>(&format!( + "SELECT * FROM {schema}.prune_executions_batch($1, $2, $3, $4)" + )) + .bind(&after.0) + .bind(after.1) + .bind(completed_before) + .bind(RECLAIM_BATCH as i32) + .fetch_one(&mut *tx) + .await; + let batch = match result { + Ok(result) => result, + Err(error) => { + tx.rollback().await.map_err(|rollback| { + format!("execution pruning failed: {error}; rollback failed: {rollback}") + })?; + return Err(format!("prune execution batch: {error}")); + } + }; tx.commit().await.map_err(|error| error.to_string())?; - let more = candidates.len() == RECLAIM_BATCH as usize; - let mut total = PruneResult::default(); - for id in candidates { - // Advance before I/O so an error or timeout cannot starve later IDs. - *cursor = id; - let result = client - .prune_executions( - cursor, - PruneOptions { - completed_before: Some(completed_before), - ..Default::default() - }, - ) - .await - .map_err(|error| format!("prune executions for {cursor}: {error}"))?; - total.instances_processed += result.instances_processed; - total.executions_deleted += result.executions_deleted; - total.events_deleted += result.events_deleted; + match (batch.last_instance_id, batch.last_execution_id) { + (Some(id), Some(execution)) => *cursor = (id, execution), + (None, None) if batch.executions_scanned == 0 => {} + _ => return Err("execution pruning returned an inconsistent cursor".to_string()), } - if total.executions_deleted > 0 { + if batch.executions_deleted > 0 { log!( - "pg_durable: pruned {} old execution(s) and {} history event(s) across {} instance(s)", - total.executions_deleted, - total.events_deleted, - total.instances_processed + "pg_durable: pruned {} old execution(s) and {} history event(s)", + batch.executions_deleted, + batch.events_deleted, ); } - Ok(more) + Ok(batch.executions_scanned == i64::from(RECLAIM_BATCH)) } /// From the engine's `Failed` instance ids and the set that still have a @@ -2547,6 +2552,56 @@ mod tests { }); } + #[pg_test] + fn worker_readiness_requires_pruning_migrations() { + let admin = Spi::get_one::("SELECT current_user::text") + .unwrap() + .unwrap(); + let database = Spi::get_one::("SELECT current_database()::text") + .unwrap() + .unwrap(); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let mut connection = + crate::origin::metadata_test_connection(&admin, &database).await; + let mut tx = connection.begin().await.unwrap(); + sqlx::raw_sql( + "CREATE SCHEMA IF NOT EXISTS _duroxide; + CREATE TABLE IF NOT EXISTS _duroxide._worker_ready( + singleton boolean PRIMARY KEY DEFAULT true, schema_version int NOT NULL, + initialized_at timestamptz NOT NULL DEFAULT now()); + INSERT INTO _duroxide._worker_ready(singleton,schema_version) VALUES (true,2) + ON CONFLICT(singleton) DO UPDATE SET schema_version=2;", + ) + .execute(&mut *tx) + .await + .unwrap(); + assert!( + !crate::types::read_backend_control_state(&mut tx) + .await + .unwrap() + .ready, + "a marker from before the pruning migrations must not admit backends" + ); + sqlx::query("UPDATE _duroxide._worker_ready SET schema_version=$1") + .bind(crate::WORKER_SCHEMA_VERSION) + .execute(&mut *tx) + .await + .unwrap(); + assert!( + crate::types::read_backend_control_state(&mut tx) + .await + .unwrap() + .ready + ); + tx.rollback().await.unwrap(); + connection.close().await.unwrap(); + }); + } + #[pg_test] fn execution_pruning_retention_and_failure_progress() { let admin = Spi::get_one::("SELECT current_user::text") @@ -2572,7 +2627,6 @@ mod tests { .unwrap(), ); let pool = store.pool(); - let client = Client::new(store.clone()); sqlx::raw_sql( "INSERT INTO execution_pruning_test.instances (instance_id, orchestration_name, current_execution_id, created_at, updated_at) @@ -2626,38 +2680,47 @@ mod tests { .unwrap(); let cutoff = 1_704_067_200_000; - let mut cursor = String::new(); - let error = prune_old_executions( - pool, &client, "execution_pruning_test", cutoff, &mut cursor, - ) - .await - .unwrap_err(); - assert!(error.contains("intentional pruning failure"), "{error}"); - assert_eq!(cursor, "b00000001"); + let mut cursor = (String::new(), 0); assert!(prune_old_executions( - pool, &client, "execution_pruning_test", cutoff, &mut cursor, + pool, "execution_pruning_test", cutoff, &mut cursor, ) .await .unwrap()); - assert_eq!(cursor, "b00001001"); - assert!(!prune_old_executions( - pool, &client, "execution_pruning_test", cutoff, &mut cursor, + assert_eq!(cursor, ("a00001000".to_string(), 1), + "protected executions must consume and advance the scanned page"); + let error = prune_old_executions( + pool, "execution_pruning_test", cutoff, &mut cursor, ) .await - .unwrap()); - assert_eq!(cursor, "zterminal"); + .unwrap_err(); + assert!(error.contains("intentional pruning failure"), "{error}"); + assert_eq!(cursor, ("b00000150".to_string(), 6)); + let mut pages = 0; + while prune_old_executions(pool, "execution_pruning_test", cutoff, &mut cursor) + .await.unwrap() + { + pages += 1; + assert!(pages < 20, "a failed page must not prevent scan exhaustion"); + } + assert_eq!(cursor, ("zterminal".to_string(), 6)); + let retained_failed_page: i64 = sqlx::query_scalar( + "SELECT count(*) FROM execution_pruning_test.executions + WHERE instance_id LIKE 'b%' AND execution_id=1", + ).fetch_one(pool).await.unwrap(); + assert_eq!(retained_failed_page, 150, "a failed batch is atomic"); sqlx::query("DROP TRIGGER reject_prune ON execution_pruning_test.executions") .execute(pool) .await .unwrap(); - cursor.clear(); - assert!(!prune_old_executions( - pool, &client, "execution_pruning_test", cutoff, &mut cursor, - ) - .await - .unwrap()); - assert_eq!(cursor, "b00000001", "retry the failed instance next cycle"); + cursor = (String::new(), 0); + pages = 0; + while prune_old_executions(pool, "execution_pruning_test", cutoff, &mut cursor) + .await.unwrap() + { + pages += 1; + assert!(pages < 20, "retry failed pages on the next cycle"); + } let counts: (i64, i64, i64) = sqlx::query_as( "SELECT (SELECT count(*) FROM execution_pruning_test.instances), @@ -2678,10 +2741,10 @@ mod tests { .unwrap(); assert_eq!(survivors, [2, 3, 4, 5, 6]); assert!(!prune_old_executions( - pool, &client, "execution_pruning_test", cutoff, &mut String::new(), + pool, "execution_pruning_test", cutoff, &mut cursor, ) .await - .unwrap(), "an exhausted scan must not select protected histories"); + .unwrap(), "an exhausted scan must not restart from protected histories"); sqlx::query("DROP SCHEMA execution_pruning_test CASCADE") .execute(pool) .await @@ -2690,6 +2753,96 @@ mod tests { }); } + #[pg_test] + fn execution_pruning_deadlines_release_connections() { + let admin = Spi::get_one::("SELECT current_user::text") + .unwrap() + .unwrap(); + let database = Spi::get_one::("SELECT current_database()::text") + .unwrap() + .unwrap(); + let mut url = url::Url::parse(&postgres_connection_string()).unwrap(); + url.set_username(&admin).unwrap(); + url.set_path(&database); + tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async { + let store = PostgresProvider::new_with_config(worker_provider_config( + url.as_str(), "pruning_deadlines", + )).await.unwrap(); + let maintenance = sqlx::postgres::PgPoolOptions::new().max_connections(1) + .connect_with(sqlx::postgres::PgConnectOptions::from_str(url.as_str()).unwrap() + .application_name("pgd_pruning_deadline_test")).await.unwrap(); + sqlx::raw_sql( + "INSERT INTO pruning_deadlines.instances + (instance_id,orchestration_name,current_execution_id,created_at,updated_at) + VALUES ('a','test',2,now(),now()),('b','test',2,now(),now()); + INSERT INTO pruning_deadlines.executions(instance_id,execution_id,status,started_at,completed_at) + SELECT id,n,CASE WHEN n=1 THEN 'ContinuedAsNew' ELSE 'Running' END, + to_timestamp(0),CASE WHEN n=1 THEN to_timestamp(1) END + FROM (VALUES ('a'),('b')) ids(id),generate_series(1,2) n; + INSERT INTO pruning_deadlines.history(instance_id,execution_id,event_id,event_type,event_data,created_at) + SELECT instance_id,execution_id,1,'Event','{}',now() FROM pruning_deadlines.executions;" + ).execute(store.pool()).await.unwrap(); + for id in ["a", "b"] { + let mut blocker = store.pool().begin().await.unwrap(); + sqlx::query("SELECT 1 FROM pruning_deadlines.history WHERE instance_id=$1 AND execution_id=1 FOR UPDATE") + .bind(id).fetch_one(&mut *blocker).await.unwrap(); + let mut cursor = (id.to_string(), 0); + let pruning = prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor); + tokio::pin!(pruning); + let deadline = tokio::time::sleep(Duration::from_secs(3)); + tokio::pin!(deadline); + loop { + tokio::select! { + result = &mut pruning => panic!("prune must encounter the held history lock: {result:?}"), + _ = &mut deadline => panic!("pruning lock wait was not observed"), + _ = tokio::time::sleep(Duration::from_millis(20)) => { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity + WHERE application_name='pgd_pruning_deadline_test' AND wait_event_type='Lock')" + ).fetch_one(store.pool()).await.unwrap(); + if waiting { break; } + } + } + } + let one: i32 = tokio::time::timeout(Duration::from_millis(500), + sqlx::query_scalar("SELECT 1").fetch_one(store.pool())).await.unwrap().unwrap(); + assert_eq!(one, 1, "maintenance must not occupy runtime connections"); + let error = tokio::time::timeout(Duration::from_secs(3), &mut pruning).await.unwrap().unwrap_err(); + assert!(error.contains("lock timeout"), "{error}"); + tokio::time::timeout(Duration::from_millis(500), + sqlx::query("SELECT 1").execute(&maintenance)).await.unwrap().unwrap(); + blocker.rollback().await.unwrap(); + } + sqlx::raw_sql( + "CREATE FUNCTION pruning_deadlines.slow_prune() RETURNS trigger LANGUAGE plpgsql AS $fn$ + BEGIN PERFORM pg_sleep(10); RETURN OLD; END $fn$; + CREATE TRIGGER slow_prune BEFORE DELETE ON pruning_deadlines.history + FOR EACH ROW EXECUTE FUNCTION pruning_deadlines.slow_prune();" + ).execute(store.pool()).await.unwrap(); + let mut cursor = (String::new(), 0); + let error = tokio::time::timeout(Duration::from_secs(7), + prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor)) + .await.unwrap().unwrap_err(); + assert!(error.contains("statement timeout"), "{error}"); + tokio::time::timeout(Duration::from_millis(500), + sqlx::query("SELECT 1").execute(&maintenance)).await.unwrap().unwrap(); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM pruning_deadlines.history") + .fetch_one(store.pool()).await.unwrap(); + assert_eq!(count, 4, "timeout must roll back the whole batch"); + sqlx::query("DROP TRIGGER slow_prune ON pruning_deadlines.history") + .execute(store.pool()).await.unwrap(); + cursor = (String::new(), 0); + assert!(!prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor) + .await.unwrap()); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM pruning_deadlines.history") + .fetch_one(store.pool()).await.unwrap(); + assert_eq!(count, 2, "only current history survives successful retry"); + maintenance.close().await; + sqlx::query("DROP SCHEMA pruning_deadlines CASCADE").execute(store.pool()).await.unwrap(); + store.pool().close().await; + }); + } + #[test] fn execution_pruning_keeps_maintenance_cycle_open() { let mut progress = MaintenanceProgress { From e53da8dd2e2308f018c7f2768e24c3d093e5ca29 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Thu, 29 Oct 2026 20:57:46 +0000 Subject: [PATCH 3/8] Account for provider migrations, handle worker readiness, correct docs --- Cargo.lock | 4 +-- Cargo.toml | 6 ++-- docs/upgrade-testing.md | 77 ++++++++++++++++++++++------------------- src/lib.rs | 2 +- src/worker.rs | 13 ++++--- 5 files changed, 57 insertions(+), 45 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2c98c614..a61e78ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -731,8 +731,8 @@ dependencies = [ [[package]] name = "duroxide-pg" -version = "0.1.36" -source = "git+https://github.com/microsoft/duroxide-pg?branch=thomcc%2Fcontinue_as_new_completed_at#c89f75484c3da68599d2de3381b1cbca7f901c57" +version = "0.1.38" +source = "git+https://github.com/microsoft/duroxide-pg?rev=7e5472726c5dbbabf78a0eb08032b1fe605c6760#7e5472726c5dbbabf78a0eb08032b1fe605c6760" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 77d994f1..d7195232 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,9 +36,9 @@ uuid = { version = "1.0", features = ["v4", "serde"] } percent-encoding = "2.3" # duroxide integration -duroxide = "0.1.32" -# DONOTLAND: requires the timestamp and batch-pruning fixes; replace with a crates.io release. -duroxide-pg = { git = "https://github.com/microsoft/duroxide-pg", branch = "thomcc/continue_as_new_completed_at" } +duroxide = "=0.1.32" +# DONOTLAND: migrations 0027/0028 are not released; replace with an exact compatible crates.io pin. +duroxide-pg = { git = "https://github.com/microsoft/duroxide-pg", rev = "7e5472726c5dbbabf78a0eb08032b1fe605c6760" } tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] } # For ExecuteSQL activity (connecting back to PostgreSQL) diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index a8ebc73f..959c09dd 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -358,6 +358,48 @@ what the upgrade script handles, and any backward compatibility considerations. ### v0.2.9 → v0.2.10 (unreleased) +#### Historical execution pruning + +- **B1:** background reconciliation prunes completed non-current execution + generations using `pg_durable.retention_days`, including running workflows and + nested loops in every origin. The query uses existing provider tables, so no + extension upgrade DDL or new `df` schema detection is needed. +- **Provider migrations:** the worker's startup `ApplyAll` applies migration `0027` + before starting the runtime. It timestamps `ContinuedAsNew` executions and + backfills older NULL timestamps from terminal history events. Unrecoverable + timestamps remain NULL with a warning; those generations are not age-pruned. + Stop any other runtimes sharing the provider schema during this one-time + backfill. Existing histories may make startup slower; any worker-role/database + statement timeout must permit that backfill to finish, or startup retries it. + This dependency pair also includes `0023` (partial indexes), `0025` (fetch + leases), `0026` (renewal and acknowledgement lease locking), and `0028` + (cursor-based pruning and consistent instance-lock ordering). Readiness schema + version `4` supersedes main's version `3` submission ledger and prevents + backends from treating a pre-migration worker marker as ready. +- **Bounded maintenance:** each slice scans at most 1,000 execution keys, including + protected rows. The provider owns deletion predicates and atomically prunes + that page on the maintenance connection under server-side deadlines. It locks + instances before history and rechecks eligibility after acquiring the locks, + protecting replacement instances and coordinating with acknowledgement and + deletion. Failed pages are revisited next scan; runtime provider connections + are not borrowed. +- **Inspection:** `df.instance_executions()` reads execution details and event + counts in one MVCC snapshot, so concurrent pruning does not produce spurious + per-generation not-found errors. Genuine query failures still raise errors. +- **Replay:** no orchestration code, durable-operation inputs, or continuation + payloads change. Pruning preserves the current and running executions. +- **Dependency pair:** duroxide is pinned exactly to `0.1.32`, as required by the + provider. Its shutdown operation cancels and joins owned tasks; zero means + immediate cancellation rather than a drain grace period. Validate the real + previous-binary lifecycle and worker shutdown when updating this pair. +- **Release blocker:** the provider is pinned to immutable commit + `7e5472726c5dbbabf78a0eb08032b1fe605c6760` of + [microsoft/duroxide-pg#32](https://github.com/microsoft/duroxide-pg/pull/32). + Released provider `0.1.38` does not contain migrations `0027`/`0028`. + Replace the `DONOTLAND` git pin with an exact compatible released version and + refresh both dependency resolutions together before landing. No local Cargo + override is needed to build or test the current pinned commit. + #### `df.start()` admission quotas (#139) - **DDL/B1:** The upgrade adds `idx_instances_submitted_by` on @@ -473,41 +515,6 @@ the shipped 0.2.7 to 0.2.8 script must remain byte-identical. See [deferred lifetime guarantees](multi-database-installation.md#release-blockers) for the separate runtime shutdown and server-side cancellation limitations. -#### Historical execution pruning - -- **B1:** background reconciliation prunes completed non-current execution - generations using `pg_durable.retention_days`, including running workflows and - nested loops in every origin. The query uses existing provider tables, so no - extension upgrade DDL or new `df` schema detection is needed. -- **Provider migrations:** the worker's startup `ApplyAll` applies migration `0024` - before starting the runtime. It timestamps `ContinuedAsNew` executions and - backfills older NULL timestamps from terminal history events. Unrecoverable - timestamps remain NULL with a warning; those generations are not age-pruned. - Stop any other runtimes sharing the provider schema during this one-time - backfill. Existing histories may make startup slower; any worker-role/database - statement timeout must permit that backfill to finish, or startup retries it. - This dependency pair also includes `0023` (partial indexes), `0025` (fetch - leases), and `0026` (cursor-based batch pruning). Readiness schema version `3` - prevents backends from treating a pre-migration worker marker as ready. -- **Bounded maintenance:** each slice scans at most 1,000 execution keys, including - protected rows. The provider owns deletion predicates and atomically prunes - that page on the maintenance connection under server-side deadlines. Failed - pages are revisited next scan; runtime provider connections are not borrowed. -- **Inspection:** `df.instance_executions()` reads execution details and event - counts in one MVCC snapshot, so concurrent pruning does not produce spurious - per-generation not-found errors. Genuine query failures still raise errors. -- **Replay:** no orchestration code, durable-operation inputs, or continuation - payloads change. Pruning preserves the current and running executions. -- **Dependency pair:** duroxide `0.1.32` is required by the provider's current - branch. Its shutdown operation now cancels and joins owned tasks; zero means - immediate cancellation rather than a drain grace period. Validate the real - previous-binary lifecycle and worker shutdown when updating this pair. -- **Release blocker:** the provider is temporarily pinned to the timestamp-fix - git branch in `Cargo.toml`. The local review fixes add migration `0026`; until - they are pushed, validation requires a local Cargo patch override. Refresh the - git lockfile after publishing those fixes and replace the `DONOTLAND` pin with - a compatible release before landing. - #### Typed HTTP endpoints - Adds composite type `df.http_endpoint(server text, path text)`, diff --git a/src/lib.rs b/src/lib.rs index 422ba28e..a8c5f807 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -97,7 +97,7 @@ pub use types::Durofut; /// by the background worker after successful initialization. Increment whenever /// a new binary introduces new duroxide-pg migration scripts or any other /// BGW-applied duroxide schema change. -pub const WORKER_SCHEMA_VERSION: i32 = 3; +pub const WORKER_SCHEMA_VERSION: i32 = 4; ::pgrx::pg_module_magic!(name, version); diff --git a/src/worker.rs b/src/worker.rs index ec64d67d..a4fbf716 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -2573,8 +2573,8 @@ mod tests { CREATE TABLE IF NOT EXISTS _duroxide._worker_ready( singleton boolean PRIMARY KEY DEFAULT true, schema_version int NOT NULL, initialized_at timestamptz NOT NULL DEFAULT now()); - INSERT INTO _duroxide._worker_ready(singleton,schema_version) VALUES (true,2) - ON CONFLICT(singleton) DO UPDATE SET schema_version=2;", + INSERT INTO _duroxide._worker_ready(singleton,schema_version) VALUES (true,3) + ON CONFLICT(singleton) DO UPDATE SET schema_version=3;", ) .execute(&mut *tx) .await @@ -2782,9 +2782,14 @@ mod tests { INSERT INTO pruning_deadlines.history(instance_id,execution_id,event_id,event_type,event_data,created_at) SELECT instance_id,execution_id,1,'Event','{}',now() FROM pruning_deadlines.executions;" ).execute(store.pool()).await.unwrap(); - for id in ["a", "b"] { + for (id, table) in [("a", "history"), ("b", "history"), ("a", "instances"), ("b", "instances")] { let mut blocker = store.pool().begin().await.unwrap(); - sqlx::query("SELECT 1 FROM pruning_deadlines.history WHERE instance_id=$1 AND execution_id=1 FOR UPDATE") + let lock_sql = if table == "history" { + "SELECT 1 FROM pruning_deadlines.history WHERE instance_id=$1 AND execution_id=1 FOR UPDATE" + } else { + "SELECT 1 FROM pruning_deadlines.instances WHERE instance_id=$1 FOR NO KEY UPDATE" + }; + sqlx::query(lock_sql) .bind(id).fetch_one(&mut *blocker).await.unwrap(); let mut cursor = (id.to_string(), 0); let pruning = prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor); From f405ae44d86932a713a19a713c5b166f891e9dab Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 11:32:20 +0000 Subject: [PATCH 4/8] Fix pruning progress --- CHANGELOG.md | 22 +++++ USER_GUIDE.md | 31 ++++--- docs/upgrade-testing.md | 7 +- src/worker.rs | 200 +++++++++++++++++++++++++++++++++++++--- 4 files changed, 235 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5677e7e1..3b81c4f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,28 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc ## [Unreleased] +### Changed + +- **Historical execution retention (#415):** reconciliation now removes completed, + non-current execution generations and their history after + `pg_durable.retention_days`, including generations of still-running workflows + and nested loops. Current and Running executions remain protected. Cleanup + advances through bounded pages under server-side deadlines, reducing the page + size after query cancellation so otherwise-prunable generations do not remain + stuck in an oversized batch. `pg_durable.reconcile_interval = 0` disables it. +- **Execution inspection (#415):** `df.instance_executions()` reads execution + details and event counts in one snapshot so concurrent pruning does not fail + the listing. Pruned generations no longer appear in subsequent inspections. + +> **Upgrade:** The worker backfills missing completion timestamps for older +> continued generations before publishing readiness. Large history stores can +> extend this initialization window, during which new submissions and engine +> operations must wait. The `statement_timeout` in effect for +> `pg_durable.worker_role` in the control database must allow the backfill to +> complete. Missing terminal history leaves the timestamp NULL with a warning +> rather than guessing its age. No `ALTER EXTENSION UPDATE` is needed for this +> provider-managed change. See [Automatic Reconciliation](USER_GUIDE.md#automatic-reconciliation). + ### Security - **Start admission quotas (#139):** `df.max_concurrent_per_user` (default diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 5bdd772c..ea17593c 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -2011,8 +2011,9 @@ runtime. Distinguish three roles: `database` argument selects another database on the same cluster; that target does not need the extension unless the SQL itself calls `df` APIs. -Satellite installation requires a compatible control worker with readiness schema -version `3` (including its origin registry and protected submission accounting). +Satellite installation requires a compatible control worker with the current +readiness schema version, including its provider migrations, origin registry, +and protected submission accounting. Provider migrations run only in control through the worker's `ApplyAll`; satellite creation never creates control automatically. The provider namespace is extension-owned, its contents are created @@ -2579,12 +2580,18 @@ in SQL before their bounded page is loaded into memory. Execution pruning scans at most 1,000 execution keys per slice, including young and protected generations. Each batch deletes at most those 1,000 generations; this is a work limit, not a generation-count retention policy. Failed pages are -logged, skipped for the rest of the scan, and retried on the next scan so they -cannot block later pages. Pruning uses the maintenance connection, not the -workflow runtime's connection pool, with server-side five-second statement and +logged. Query cancellations, including statement timeouts, retry the same position +with smaller pages, down to one execution. The smaller limit is retained for the +rest of that scan; the next scan starts at 1,000 again. Lock timeouts and other +failures, or cancellation of a single-execution page, skip that page until the +next scan so they cannot block later pages. Pruning uses the maintenance +connection, not the workflow runtime's connection pool, with server-side +five-second statement and 1.5-second lock deadlines. A timeout rolls back the batch. The number of history -events in one generation is not capped; a batch that cannot finish within the -deadline is deferred. +events in one generation is not capped; an individual generation that cannot +finish within the deadline can remain unpruned despite aging past the retention +window. Smaller pages can lengthen the scan, but other pending maintenance work +still gets a turn between pruning slices. Metadata queries have server-side deadlines and satellite operations have a 15-second overall deadline. The candidate limit bounds materialized rows, not @@ -2633,10 +2640,12 @@ continued generations from their recorded terminal history events. If an event is missing, it logs a migration warning and leaves that generation ineligible for age-based pruning rather than guessing its age. This one-time backfill runs before the worker becomes ready and can delay workflow processing on a large -history store. Allow sufficient time for it during upgrade; a role or database -`statement_timeout` shorter than the backfill will cause startup to retry rather -than finish. Adjust the worker's migration-session timeout for the upgrade if -necessary. +history store. New submissions and engine operations such as signals and +cancellation must wait for worker readiness. Allow sufficient time for the +backfill during upgrade: the `statement_timeout` in effect for +`pg_durable.worker_role` in the control database must permit it to finish. +Adjust the applicable role or database timeout defaults for the upgrade if +necessary; otherwise startup retries the backfill instead of completing. Notes: diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 959c09dd..24194772 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -381,8 +381,11 @@ what the upgrade script handles, and any backward compatibility considerations. that page on the maintenance connection under server-side deadlines. It locks instances before history and rechecks eligibility after acquiring the locks, protecting replacement instances and coordinating with acknowledgement and - deletion. Failed pages are revisited next scan; runtime provider connections - are not borrowed. + deletion. Query cancellation retries the same cursor with a smaller page only + after rollback succeeds; the reduced limit lasts until that scan ends. + Lock failures and other errors, or a cancelled single-execution page, retain + the skip-and-retry-next-scan behavior. Runtime provider connections are not + borrowed. - **Inspection:** `df.instance_executions()` reads execution details and event counts in one MVCC snapshot, so concurrent pruning does not produce spurious per-generation not-found errors. Genuine query failures still raise errors. diff --git a/src/worker.rs b/src/worker.rs index a4fbf716..66316cfe 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -1150,6 +1150,7 @@ async fn run_until_extension_dropped_or_shutdown( let mut engine_cursor = String::new(); let mut legacy_cursor = String::new(); let mut execution_cursor = (String::new(), 0); + let mut execution_batch_size = EXECUTION_PRUNE_BATCH_SIZE; let mut registration_cursor = (0i64, uuid::Uuid::nil()); let mut submission_cursor = String::new(); let mut pending = MaintenanceProgress::new(); @@ -1238,6 +1239,7 @@ async fn run_until_extension_dropped_or_shutdown( pending.executions = match prune_old_executions( maintenance_pool, &schema, retention_cutoff_ms(retention), &mut execution_cursor, + &mut execution_batch_size, ).await { Ok(more) => more, Err(error) => { @@ -1245,7 +1247,10 @@ async fn run_until_extension_dropped_or_shutdown( execution_cursor != previous_cursor }, }; - if !pending.executions { execution_cursor = (String::new(), 0); } + if !pending.executions { + execution_cursor = (String::new(), 0); + execution_batch_size = EXECUTION_PRUNE_BATCH_SIZE; + } } if pending.legacy { match reclaim_orphaned_instances(maintenance_pool, &client, &schema, retention, &mut legacy_cursor).await { Ok(reclaimed) if reclaimed > 0 => { @@ -1320,6 +1325,7 @@ const ORIGIN_BATCH: i64 = 1; const ORIGIN_ENGINE_BATCH: i64 = 100; const ORIGIN_OPERATION_TIMEOUT: Duration = Duration::from_secs(15); const MAINTENANCE_CONTINUATION_DELAY: Duration = Duration::from_millis(100); +const EXECUTION_PRUNE_BATCH_SIZE: i32 = 1_000; struct MaintenanceProgress { control: bool, @@ -1956,7 +1962,9 @@ async fn prune_old_executions( schema: &str, completed_before: u64, cursor: &mut (String, i64), + batch_size: &mut i32, ) -> Result { + assert!((1..=EXECUTION_PRUNE_BATCH_SIZE).contains(batch_size)); let completed_before = i64::try_from(completed_before) .map_err(|error| format!("invalid pruning cutoff: {error}"))?; let mut tx = pool.begin().await.map_err(|error| error.to_string())?; @@ -1975,7 +1983,7 @@ async fn prune_old_executions( )) .bind(&cursor.0) .bind(cursor.1) - .bind(i64::from(RECLAIM_BATCH)) + .bind(i64::from(*batch_size)) .fetch_optional(&mut *tx) .await .map_err(|error| format!("page execution pruning keys: {error}"))?; @@ -1992,7 +2000,7 @@ async fn prune_old_executions( .bind(&after.0) .bind(after.1) .bind(completed_before) - .bind(RECLAIM_BATCH as i32) + .bind(*batch_size) .fetch_one(&mut *tx) .await; let batch = match result { @@ -2001,6 +2009,22 @@ async fn prune_old_executions( tx.rollback().await.map_err(|rollback| { format!("execution pruning failed: {error}; rollback failed: {rollback}") })?; + if *batch_size > 1 + && error + .as_database_error() + .and_then(|error| error.code()) + .as_deref() + == Some("57014") + { + // Query cancellation includes statement timeouts. Retry only after + // rollback, keeping the smaller limit for the rest of this scan. + *batch_size = (*batch_size / 2).max(1); + *cursor = after; + log!( + "pg_durable: execution pruning query cancelled: {error}; retrying with at most {batch_size} execution(s)" + ); + return Ok(true); + } return Err(format!("prune execution batch: {error}")); } }; @@ -2017,7 +2041,7 @@ async fn prune_old_executions( batch.events_deleted, ); } - Ok(batch.executions_scanned == i64::from(RECLAIM_BATCH)) + Ok(batch.executions_scanned == i64::from(*batch_size)) } /// From the engine's `Failed` instance ids and the set that still have a @@ -2681,22 +2705,24 @@ mod tests { let cutoff = 1_704_067_200_000; let mut cursor = (String::new(), 0); + let mut batch_size = EXECUTION_PRUNE_BATCH_SIZE; assert!(prune_old_executions( - pool, "execution_pruning_test", cutoff, &mut cursor, + pool, "execution_pruning_test", cutoff, &mut cursor, &mut batch_size, ) .await .unwrap()); assert_eq!(cursor, ("a00001000".to_string(), 1), "protected executions must consume and advance the scanned page"); let error = prune_old_executions( - pool, "execution_pruning_test", cutoff, &mut cursor, + pool, "execution_pruning_test", cutoff, &mut cursor, &mut batch_size, ) .await .unwrap_err(); assert!(error.contains("intentional pruning failure"), "{error}"); + assert_eq!(batch_size, EXECUTION_PRUNE_BATCH_SIZE); assert_eq!(cursor, ("b00000150".to_string(), 6)); let mut pages = 0; - while prune_old_executions(pool, "execution_pruning_test", cutoff, &mut cursor) + while prune_old_executions(pool, "execution_pruning_test", cutoff, &mut cursor, &mut batch_size) .await.unwrap() { pages += 1; @@ -2715,7 +2741,7 @@ mod tests { .unwrap(); cursor = (String::new(), 0); pages = 0; - while prune_old_executions(pool, "execution_pruning_test", cutoff, &mut cursor) + while prune_old_executions(pool, "execution_pruning_test", cutoff, &mut cursor, &mut batch_size) .await.unwrap() { pages += 1; @@ -2741,7 +2767,7 @@ mod tests { .unwrap(); assert_eq!(survivors, [2, 3, 4, 5, 6]); assert!(!prune_old_executions( - pool, "execution_pruning_test", cutoff, &mut cursor, + pool, "execution_pruning_test", cutoff, &mut cursor, &mut batch_size, ) .await .unwrap(), "an exhausted scan must not restart from protected histories"); @@ -2753,6 +2779,152 @@ mod tests { }); } + #[pg_test] + fn execution_pruning_splits_cancelled_batches() { + let admin = Spi::get_one::("SELECT current_user::text") + .unwrap() + .unwrap(); + let database = Spi::get_one::("SELECT current_database()::text") + .unwrap() + .unwrap(); + let mut url = url::Url::parse(&postgres_connection_string()).unwrap(); + url.set_username(&admin).unwrap(); + url.set_path(&database); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let store = PostgresProvider::new_with_config(worker_provider_config( + url.as_str(), + "pruning_batch_reduction", + )) + .await + .unwrap(); + let pool = store.pool(); + sqlx::raw_sql( + "INSERT INTO pruning_batch_reduction.instances + (instance_id,orchestration_name,current_execution_id,created_at,updated_at) + VALUES ('aggregate','test',4,now(),now()); + INSERT INTO pruning_batch_reduction.executions + (instance_id,execution_id,status,started_at,completed_at) + SELECT 'aggregate',n, + CASE WHEN n=4 THEN 'Running' ELSE 'ContinuedAsNew' END, + to_timestamp(0),CASE WHEN n<4 THEN to_timestamp(1) END + FROM generate_series(1,4) n; + INSERT INTO pruning_batch_reduction.history + (instance_id,execution_id,event_id,event_type,event_data,created_at) + SELECT instance_id,execution_id,1,'Event','{}',now() + FROM pruning_batch_reduction.executions; + CREATE FUNCTION pruning_batch_reduction.slow_prune() RETURNS trigger + LANGUAGE plpgsql AS $fn$ BEGIN + IF OLD.instance_id='blocked' THEN + RAISE EXCEPTION 'cancelled single generation' USING ERRCODE='57014'; + ELSIF OLD.instance_id='aggregate' THEN + PERFORM pg_sleep(2); + END IF; + RETURN OLD; + END $fn$; + CREATE TRIGGER slow_prune BEFORE DELETE ON pruning_batch_reduction.history + FOR EACH ROW EXECUTE FUNCTION pruning_batch_reduction.slow_prune();", + ) + .execute(pool) + .await + .unwrap(); + + let mut cursor = (String::new(), 0); + let mut batch_size = 4; + assert!(tokio::time::timeout( + Duration::from_secs(8), + prune_old_executions(pool, "pruning_batch_reduction", 2000, &mut cursor, &mut batch_size), + ) + .await + .unwrap() + .unwrap(), "the timed-out aggregate must schedule a smaller retry"); + assert_eq!(cursor, (String::new(), 0)); + assert_eq!(batch_size, 2); + let histories: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pruning_batch_reduction.history", + ) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(histories, 4, "retry only after atomic rollback"); + let mut slices = 0; + while prune_old_executions( + pool, "pruning_batch_reduction", 2000, &mut cursor, &mut batch_size, + ) + .await + .unwrap() + { + slices += 1; + assert!(slices < 8, "prunable generations must not retry indefinitely"); + assert!(batch_size <= 2, "success must not immediately increase the limit"); + } + let survivors: Vec = sqlx::query_scalar( + "SELECT execution_id FROM pruning_batch_reduction.executions ORDER BY execution_id", + ) + .fetch_all(pool) + .await + .unwrap(); + assert_eq!(survivors, [4]); + + sqlx::raw_sql( + "INSERT INTO pruning_batch_reduction.instances + (instance_id,orchestration_name,current_execution_id,created_at,updated_at) + VALUES ('blocked','test',2,now(),now()),('tail','test',2,now(),now()); + INSERT INTO pruning_batch_reduction.executions + (instance_id,execution_id,status,started_at,completed_at) + SELECT id,n,CASE WHEN n=2 THEN 'Running' ELSE 'ContinuedAsNew' END, + to_timestamp(0),CASE WHEN n=1 THEN to_timestamp(1) END + FROM (VALUES ('blocked'),('tail')) ids(id),generate_series(1,2) n; + INSERT INTO pruning_batch_reduction.history + (instance_id,execution_id,event_id,event_type,event_data,created_at) + SELECT instance_id,execution_id,1,'Event','{}',now() + FROM pruning_batch_reduction.executions WHERE instance_id<>'aggregate';", + ) + .execute(pool) + .await + .unwrap(); + batch_size = 1; + let error = prune_old_executions( + pool, "pruning_batch_reduction", 2000, &mut cursor, &mut batch_size, + ) + .await + .unwrap_err(); + assert!(error.contains("cancelled single generation"), "{error}"); + assert_eq!(cursor, ("blocked".to_string(), 1)); + slices = 0; + while prune_old_executions( + pool, "pruning_batch_reduction", 2000, &mut cursor, &mut batch_size, + ) + .await + .unwrap() + { + slices += 1; + assert!(slices < 8, "a cancelled singleton must not starve later keys"); + } + let survivors: Vec<(String, i64)> = sqlx::query_as( + "SELECT instance_id,execution_id FROM pruning_batch_reduction.executions + ORDER BY instance_id,execution_id", + ) + .fetch_all(pool) + .await + .unwrap(); + assert_eq!(survivors, [ + ("aggregate".to_string(), 4), + ("blocked".to_string(), 1), + ("blocked".to_string(), 2), + ("tail".to_string(), 2), + ]); + sqlx::query("DROP SCHEMA pruning_batch_reduction CASCADE") + .execute(pool) + .await + .unwrap(); + pool.close().await; + }); + } + #[pg_test] fn execution_pruning_deadlines_release_connections() { let admin = Spi::get_one::("SELECT current_user::text") @@ -2792,7 +2964,8 @@ mod tests { sqlx::query(lock_sql) .bind(id).fetch_one(&mut *blocker).await.unwrap(); let mut cursor = (id.to_string(), 0); - let pruning = prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor); + let mut batch_size = EXECUTION_PRUNE_BATCH_SIZE; + let pruning = prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor, &mut batch_size); tokio::pin!(pruning); let deadline = tokio::time::sleep(Duration::from_secs(3)); tokio::pin!(deadline); @@ -2825,10 +2998,12 @@ mod tests { FOR EACH ROW EXECUTE FUNCTION pruning_deadlines.slow_prune();" ).execute(store.pool()).await.unwrap(); let mut cursor = (String::new(), 0); + let mut batch_size = 1; let error = tokio::time::timeout(Duration::from_secs(7), - prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor)) + prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor, &mut batch_size)) .await.unwrap().unwrap_err(); assert!(error.contains("statement timeout"), "{error}"); + assert_eq!(batch_size, 1, "an indivisible execution must be deferred"); tokio::time::timeout(Duration::from_millis(500), sqlx::query("SELECT 1").execute(&maintenance)).await.unwrap().unwrap(); let count: i64 = sqlx::query_scalar("SELECT count(*) FROM pruning_deadlines.history") @@ -2837,7 +3012,8 @@ mod tests { sqlx::query("DROP TRIGGER slow_prune ON pruning_deadlines.history") .execute(store.pool()).await.unwrap(); cursor = (String::new(), 0); - assert!(!prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor) + batch_size = EXECUTION_PRUNE_BATCH_SIZE; + assert!(!prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor, &mut batch_size) .await.unwrap()); let count: i64 = sqlx::query_scalar("SELECT count(*) FROM pruning_deadlines.history") .fetch_one(store.pool()).await.unwrap(); From e868ac4b2ceab1de0cb18574a8270170aa037721 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Tue, 3 Nov 2026 17:50:44 +0000 Subject: [PATCH 5/8] Update to latest provider change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Also: - Updated upgrade guidance for fully drained callers, malformed-history diagnostics, WAL/retention impact, and timestamp repair. - Documented that databases using earlier pre-release migrations won’t automatically receive revised function bodies. - Added a regression confirming protected generations advance the cursor without waiting on instance locks. --- CHANGELOG.md | 17 +++++++++------ Cargo.lock | 2 +- Cargo.toml | 2 +- USER_GUIDE.md | 23 ++++++++++++++++---- docs/upgrade-testing.md | 47 +++++++++++++++++++++++++++++++++-------- src/worker.rs | 9 ++++++++ 6 files changed, 79 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b81c4f0..0da4bb65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,14 +19,19 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc details and event counts in one snapshot so concurrent pruning does not fail the listing. Pruned generations no longer appear in subsequent inspections. -> **Upgrade:** The worker backfills missing completion timestamps for older -> continued generations before publishing readiness. Large history stores can -> extend this initialization window, during which new submissions and engine -> operations must wait. The `statement_timeout` in effect for +> **Upgrade:** Stop and drain all provider callers, not just workers, before +> upgrading. A full PostgreSQL stop/start drains pg_durable's old server +> processes; coordinate any external clients and deletion jobs sharing the +> provider store too. The new worker backfills missing completion timestamps +> before publishing readiness. Large stores can extend this window and generate +> substantial WAL, while submissions and engine operations must wait. Review +> retention cutoffs before restarting because recovered timestamps can make old +> generations eligible for cleanup immediately. The `statement_timeout` in effect for > `pg_durable.worker_role` in the control database must allow the backfill to > complete. Missing terminal history leaves the timestamp NULL with a warning -> rather than guessing its age. No `ALTER EXTENSION UPDATE` is needed for this -> provider-managed change. See [Automatic Reconciliation](USER_GUIDE.md#automatic-reconciliation). +> rather than guessing its age; malformed JSONB aborts the migration with the +> affected history key in the error. No `ALTER EXTENSION UPDATE` is needed for +> this provider-managed change. See [upgrade and repair guidance](docs/upgrade-testing.md#historical-execution-pruning). ### Security diff --git a/Cargo.lock b/Cargo.lock index a61e78ed..fec32c05 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -732,7 +732,7 @@ dependencies = [ [[package]] name = "duroxide-pg" version = "0.1.38" -source = "git+https://github.com/microsoft/duroxide-pg?rev=7e5472726c5dbbabf78a0eb08032b1fe605c6760#7e5472726c5dbbabf78a0eb08032b1fe605c6760" +source = "git+https://github.com/microsoft/duroxide-pg?rev=79870890d1a4a14eb840ec1d31d0b6dff0716c0a#79870890d1a4a14eb840ec1d31d0b6dff0716c0a" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index d7195232..29c8703b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,7 +38,7 @@ percent-encoding = "2.3" # duroxide integration duroxide = "=0.1.32" # DONOTLAND: migrations 0027/0028 are not released; replace with an exact compatible crates.io pin. -duroxide-pg = { git = "https://github.com/microsoft/duroxide-pg", rev = "7e5472726c5dbbabf78a0eb08032b1fe605c6760" } +duroxide-pg = { git = "https://github.com/microsoft/duroxide-pg", rev = "79870890d1a4a14eb840ec1d31d0b6dff0716c0a" } tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] } # For ExecuteSQL activity (connecting back to PostgreSQL) diff --git a/USER_GUIDE.md b/USER_GUIDE.md index ea17593c..563e05e6 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -2638,14 +2638,29 @@ Setting `reconcile_interval = 0` also disables execution pruning. On binary upgrade, the worker recovers missing completion timestamps for older continued generations from their recorded terminal history events. If an event is missing, it logs a migration warning and leaves that generation ineligible for -age-based pruning rather than guessing its age. This one-time backfill runs -before the worker becomes ready and can delay workflow processing on a large +age-based pruning rather than guessing its age. Malformed history that cannot be +decoded as JSONB instead aborts the migration and identifies the affected schema, +instance, execution, and event in the initialization error. The worker retries +initialization; it does not skip the malformed history. + +This upgrade requires all old provider calls and transactions to finish before +migrations run, not a rolling upgrade. Pause deletion/pruning jobs and stop and +drain every provider caller, including clients and direct SQL callers as well as +workers. A full PostgreSQL stop/start drains pg_durable's old server processes; +also coordinate any external callers sharing the provider store. The new +background worker applies migrations automatically. Wait for its readiness +before resuming provider traffic, and use only updated external callers. + +The one-time backfill can delay readiness and generate substantial WAL on a large history store. New submissions and engine operations such as signals and -cancellation must wait for worker readiness. Allow sufficient time for the -backfill during upgrade: the `statement_timeout` in effect for +cancellation must wait. Review retention settings before restarting: recovered +timestamps may make many old generations eligible for deletion on the next scan. +Allow sufficient time for the backfill: the `statement_timeout` in effect for `pg_durable.worker_role` in the control database must permit it to finish. Adjust the applicable role or database timeout defaults for the upgrade if necessary; otherwise startup retries the backfill instead of completing. +See [upgrade and repair guidance](docs/upgrade-testing.md#historical-execution-pruning), +including the special handling of databases that used earlier pre-release pins. Notes: diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 24194772..f64bf07d 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -368,21 +368,42 @@ what the upgrade script handles, and any backward compatibility considerations. before starting the runtime. It timestamps `ContinuedAsNew` executions and backfills older NULL timestamps from terminal history events. Unrecoverable timestamps remain NULL with a warning; those generations are not age-pruned. - Stop any other runtimes sharing the provider schema during this one-time - backfill. Existing histories may make startup slower; any worker-role/database - statement timeout must permit that backfill to finish, or startup retries it. + Malformed history that cannot be decoded as JSONB aborts migration `0027` + atomically and reports the schema, instance, execution, and event key. The + worker logs the initialization error and retries; malformed history is not + silently skipped or treated as a missing terminal event. This dependency pair also includes `0023` (partial indexes), `0025` (fetch leases), `0026` (renewal and acknowledgement lease locking), and `0028` (cursor-based pruning and consistent instance-lock ordering). Readiness schema version `4` supersedes main's version `3` submission ledger and prevents backends from treating a pre-migration worker marker as ready. +- **Drained upgrade:** pause delete/prune jobs and stop and drain every provider + caller, including workers, clients, maintenance jobs, and direct SQL callers. + Wait for in-flight calls and transactions to finish; stopping only workers + does not prevent an old client's deletion logic from removing a pending + continuation after backfill. A full PostgreSQL stop/start drains pg_durable's + old backend and worker processes. Also coordinate any external callers sharing + the provider schema. Restart PostgreSQL with the new extension library and + let its worker apply migrations before resuming provider traffic; + wait for readiness and restart only updated external callers. No separate + migration process is required for pg_durable. +- **Backfill operations:** allow for initialization time and WAL growth on large + stores. Any worker-role/database statement timeout must permit the backfill to + finish, or startup retries it. Review retention cutoffs before restarting: + recovered timestamps can make many old generations eligible on the next scan. + If an improperly drained upgrade left NULL timestamps behind, drain all + callers first and use the provider's + [documented data-only repair](https://github.com/microsoft/duroxide-pg/blob/79870890d1a4a14eb840ec1d31d0b6dff0716c0a/migrations/README.md#repairing-missing-continued-execution-timestamps). + Do not delete migration-ledger rows or re-run whole migrations to repair data. - **Bounded maintenance:** each slice scans at most 1,000 execution keys, including protected rows. The provider owns deletion predicates and atomically prunes - that page on the maintenance connection under server-side deadlines. It locks - instances before history and rechecks eligibility after acquiring the locks, - protecting replacement instances and coordinating with acknowledgement and - deletion. Query cancellation retries the same cursor with a smaller page only - after rollback succeeds; the reduced limit lasts until that scan ends. + that page on the maintenance connection under server-side deadlines. It first + checks eligibility without locking, so protected rows advance the cursor + without locking their instances. Deletion candidates still lock instances + before history and recheck eligibility after acquiring the locks, protecting + replacement instances and coordinating with acknowledgement and deletion. + Query cancellation retries the same cursor with a smaller page only after + rollback succeeds; the reduced limit lasts until that scan ends. Lock failures and other errors, or a cancelled single-execution page, retain the skip-and-retry-next-scan behavior. Runtime provider connections are not borrowed. @@ -396,12 +417,20 @@ what the upgrade script handles, and any backward compatibility considerations. immediate cancellation rather than a drain grace period. Validate the real previous-binary lifecycle and worker shutdown when updating this pair. - **Release blocker:** the provider is pinned to immutable commit - `7e5472726c5dbbabf78a0eb08032b1fe605c6760` of + `79870890d1a4a14eb840ec1d31d0b6dff0716c0a` of [microsoft/duroxide-pg#32](https://github.com/microsoft/duroxide-pg/pull/32). Released provider `0.1.38` does not contain migrations `0027`/`0028`. Replace the `DONOTLAND` git pin with an exact compatible released version and refresh both dependency resolutions together before landing. No local Cargo override is needed to build or test the current pinned commit. +- **Pre-release databases:** revisions of microsoft/duroxide-pg#32 changed migration + `0027`/`0028` bodies without changing their numbers. A store that already + recorded those migrations from an earlier pin will not install the revised + functions just by loading this binary. Use fresh disposable provider schemas + for development validation. Preserving such a store requires a deliberate + forward migration; changing the readiness marker or deleting ledger entries + is not a substitute. The data-only repair above does not replace functions. + Check migration identity again when replacing the pre-release pin. #### `df.start()` admission quotas (#139) diff --git a/src/worker.rs b/src/worker.rs index 66316cfe..34705108 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -3018,6 +3018,15 @@ mod tests { let count: i64 = sqlx::query_scalar("SELECT count(*) FROM pruning_deadlines.history") .fetch_one(store.pool()).await.unwrap(); assert_eq!(count, 2, "only current history survives successful retry"); + let mut blocker = store.pool().begin().await.unwrap(); + sqlx::query("SELECT 1 FROM pruning_deadlines.instances ORDER BY instance_id FOR NO KEY UPDATE") + .fetch_all(&mut *blocker).await.unwrap(); + cursor = (String::new(), 0); + assert!(!tokio::time::timeout(Duration::from_secs(3), + prune_old_executions(&maintenance, "pruning_deadlines", 2000, &mut cursor, &mut batch_size)) + .await.unwrap().expect("protected generations must not wait for instance locks")); + assert_eq!(cursor, ("b".to_string(), 2), "protected keys must still advance the cursor"); + blocker.rollback().await.unwrap(); maintenance.close().await; sqlx::query("DROP SCHEMA pruning_deadlines CASCADE").execute(store.pool()).await.unwrap(); store.pool().close().await; From 47b951f60a1cade0cb2a7869aefbdec8e60a5e88 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Wed, 4 Nov 2026 12:41:08 +0000 Subject: [PATCH 6/8] Test the provider execution schema contract Compare snapshot inspection against the Client metadata API using tables created by the pinned provider's migrations and populated through provider operations. Document the existing migrated-schema coverage of pruning keys. Ref: #415 review finding 415-1 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/upgrade-testing.md | 6 ++ src/monitoring.rs | 161 ++++++++++++++++++++++++++++++++++++++++ src/worker.rs | 2 + 3 files changed, 169 insertions(+) diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index f64bf07d..e59840d4 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -410,6 +410,12 @@ what the upgrade script handles, and any backward compatibility considerations. - **Inspection:** `df.instance_executions()` reads execution details and event counts in one MVCC snapshot, so concurrent pruning does not produce spurious per-generation not-found errors. Genuine query failures still raise errors. + The snapshot and pruning pre-scan depend on provider column names and types. + `execution_snapshot_matches_provider_api` checks snapshot rows against the + provider's public metadata API using a store created by its actual migrations + and populated through provider operations. The existing pruning retention and + failure-progress test exercises pre-scan key columns against the same migrated + schema contract. Run these regressions when changing the provider dependency. - **Replay:** no orchestration code, durable-operation inputs, or continuation payloads change. Pruning preserves the current and running executions. - **Dependency pair:** duroxide is pinned exactly to `0.1.32`, as required by the diff --git a/src/monitoring.rs b/src/monitoring.rs index 428fb0d4..28ea5829 100644 --- a/src/monitoring.rs +++ b/src/monitoring.rs @@ -799,6 +799,8 @@ async fn read_execution_snapshot( let schema = format!("\"{}\"", schema.replace('"', "\"\"")); // One statement gives details and event counts the same MVCC snapshot, // including when maintenance removes older generations during this read. + // execution_snapshot_matches_provider_api checks these column assumptions + // against the real provider migrations and its public metadata API. let rows: Vec<( i64, String, @@ -846,6 +848,165 @@ mod tests { use std::str::FromStr; use std::time::Duration; + #[pg_test] + fn execution_snapshot_matches_provider_api() { + use duroxide::providers::{ExecutionMetadata, Provider, WorkItem}; + use duroxide::{Event, EventKind}; + use std::sync::Arc; + + let admin = Spi::get_one::("SELECT current_user::text") + .unwrap() + .unwrap(); + let database = Spi::get_one::("SELECT current_database()::text") + .unwrap() + .unwrap(); + let mut url = url::Url::parse(&postgres_connection_string()).unwrap(); + url.set_username(&admin).unwrap(); + url.set_path(&database); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let schema = "execution_snapshot_contract"; + let store = Arc::new( + duroxide_pg::PostgresProvider::new_with_config( + crate::types::worker_provider_config(url.as_str(), schema), + ) + .await + .expect("initialize the real provider schema for the column contract"), + ); + for (instance, execution, output) in [ + ("contract", 1, Some("finished")), + ("contract", 2, None), + ("unrelated", 1, Some("not this instance")), + ] { + store + .enqueue_for_orchestrator( + WorkItem::StartOrchestration { + instance: instance.into(), + orchestration: "contract".into(), + input: "{}".into(), + version: Some("1.0.0".into()), + parent_instance: None, + parent_id: None, + parent_execution_id: None, + execution_id: execution, + }, + None, + ) + .await + .unwrap(); + let (_, token, _) = store + .fetch_orchestration_item(Duration::from_secs(30), Duration::ZERO, None) + .await + .unwrap() + .expect("fetch the seeded execution"); + let mut history = vec![Event::with_event_id( + 1, + instance, + execution, + None, + EventKind::OrchestrationStarted { + name: "contract".into(), + version: "1.0.0".into(), + input: "{}".into(), + parent_instance: None, + parent_id: None, + parent_execution_id: None, + carry_forward_events: None, + initial_custom_status: None, + }, + )]; + if let Some(output) = output { + history.push(Event::with_event_id( + 2, + instance, + execution, + None, + EventKind::OrchestrationCompleted { + output: output.into(), + }, + )); + } + store + .ack_orchestration_item( + &token, + execution, + history, + vec![], + vec![], + ExecutionMetadata { + orchestration_name: Some("contract".into()), + orchestration_version: Some("1.0.0".into()), + status: Some( + if output.is_some() { + "Completed" + } else { + "Running" + } + .into(), + ), + output: output.map(str::to_string), + ..Default::default() + }, + vec![], + ) + .await + .unwrap(); + } + + let client = Client::new(store.clone()); + let mut expected = Vec::new(); + for execution in [2, 1] { + let info = client + .get_execution_info("contract", execution) + .await + .unwrap(); + expected.push(( + i64::try_from(info.execution_id).unwrap(), + info.status, + i64::try_from(info.event_count).unwrap(), + i64::try_from( + info.completed_at + .map(|end| end.saturating_sub(info.started_at)) + .unwrap_or(0), + ) + .unwrap(), + info.output, + )); + } + assert_eq!(expected[0], (2, "Running".into(), 1, 0, None)); + assert_eq!(expected[1].1, "Completed"); + assert_eq!(expected[1].2, 2); + assert_eq!(expected[1].4.as_deref(), Some("finished")); + assert_eq!( + read_execution_snapshot(store.pool(), schema, "contract", 5) + .await + .expect( + "provider executions/history columns must support snapshot inspection" + ), + expected, + ); + assert_eq!( + read_execution_snapshot(store.pool(), schema, "contract", 1) + .await + .unwrap(), + expected[..1], + "limit and newest-first ordering must match the provider API", + ); + assert!(read_execution_snapshot(store.pool(), schema, "missing", 5) + .await + .unwrap() + .is_empty()); + sqlx::query("DROP SCHEMA execution_snapshot_contract CASCADE") + .execute(store.pool()) + .await + .unwrap(); + store.pool().close().await; + }); + } + #[pg_test] fn execution_snapshot_survives_concurrent_pruning() { let admin = Spi::get_one::("SELECT current_user::text") diff --git a/src/worker.rs b/src/worker.rs index 34705108..cc13c9bb 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -1973,6 +1973,8 @@ async fn prune_old_executions( .map_err(|error| error.to_string())?; // Remember the scanned page before deletion, so a failed page cannot starve // later keys. Neither eligibility nor the number of deletions drives progress. + // execution_pruning_retention_and_failure_progress tests these key columns + // against the real provider schema, not a local imitation. let scanned_through: Option<(String, i64)> = sqlx::query_as(&format!( "SELECT instance_id, execution_id FROM ( SELECT instance_id, execution_id FROM {schema}.executions From bde75fc49e93af5691b2580b21b05ad24963aba6 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Wed, 4 Nov 2026 12:41:57 +0000 Subject: [PATCH 7/8] Document pruning-related workflow latency Explain that pruning holds candidate instance locks until commit and can briefly delay acknowledgements, particularly during the first post-backfill scans. Preserve the existing batch and timeout policies. Ref: #415 review finding 415-2 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- USER_GUIDE.md | 8 ++++++++ docs/upgrade-testing.md | 5 +++++ 2 files changed, 13 insertions(+) diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 563e05e6..18f15de9 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -2593,6 +2593,14 @@ finish within the deadline can remain unpruned despite aging past the retention window. Smaller pages can lengthen the scan, but other pending maintenance work still gets a turn between pruning slices. +Pruning old generations holds instance-row locks until the batch transaction +ends. Acknowledgements for those same workflows can wait even though their +current generations are preserved and pruning uses a separate connection. +Expect brief turn-latency spikes during scans; an expensive batch can hold those +locks for close to the five-second statement deadline. The first scans after +upgrade can cause more contention because backfill makes accumulated history +eligible at once. + Metadata queries have server-side deadlines and satellite operations have a 15-second overall deadline. The candidate limit bounds materialized rows, not every row PostgreSQL might examine when determining the newest retained set. diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index e59840d4..f01f8b1a 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -391,6 +391,11 @@ what the upgrade script handles, and any backward compatibility considerations. stores. Any worker-role/database statement timeout must permit the backfill to finish, or startup retries it. Review retention cutoffs before restarting: recovered timestamps can make many old generations eligible on the next scan. + Expect brief workflow turn-latency spikes during those first scans: pruning + holds candidate instance-row locks until each batch transaction ends, and + acknowledgements for the same running workflows wait on those locks. An + expensive batch can hold them for close to the five-second statement deadline; + using a maintenance connection does not remove this row-lock contention. If an improperly drained upgrade left NULL timestamps behind, drain all callers first and use the provider's [documented data-only repair](https://github.com/microsoft/duroxide-pg/blob/79870890d1a4a14eb840ec1d31d0b6dff0716c0a/migrations/README.md#repairing-missing-continued-execution-timestamps). From 65d683824b18f60d5f4ef602a636b39e7cde286b Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Wed, 4 Nov 2026 12:42:43 +0000 Subject: [PATCH 8/8] Clarify ownership of the exact duroxide pin Distinguish the provider's ^0.1.32 requirement from pg_durable's deliberate =0.1.32 pin. Record that the combined release with PR 429 needs later runtime changes and coordinated exact pins with replay validation. Ref: #415 review finding 415-3 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/upgrade-testing.md | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index f01f8b1a..9e76e32a 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -423,10 +423,14 @@ what the upgrade script handles, and any backward compatibility considerations. schema contract. Run these regressions when changing the provider dependency. - **Replay:** no orchestration code, durable-operation inputs, or continuation payloads change. Pruning preserves the current and running executions. -- **Dependency pair:** duroxide is pinned exactly to `0.1.32`, as required by the - provider. Its shutdown operation cancels and joins owned tasks; zero means - immediate cancellation rather than a drain grace period. Validate the real - previous-binary lifecycle and worker shutdown when updating this pair. +- **Dependency pair:** the provider accepts duroxide `^0.1.32`; pg_durable chooses + the exact `=0.1.32` pin so replay-sensitive dependency updates remain deliberate. + The combined release with #429 needs later duroxide changes, so select exact, + compatible released runtime/provider pins for both PRs rather than treating + `0.1.32` as an upstream requirement. In the currently pinned runtime, shutdown + cancels and joins owned tasks; zero means immediate cancellation rather than + a drain grace period. Validate the real previous-binary lifecycle and worker + shutdown when updating this pair. - **Release blocker:** the provider is pinned to immutable commit `79870890d1a4a14eb840ec1d31d0b6dff0716c0a` of [microsoft/duroxide-pg#32](https://github.com/microsoft/duroxide-pg/pull/32).