From b91feb698658a836549207bbaa817abf44016f81 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Mon, 5 Oct 2026 20:40:48 +0000 Subject: [PATCH 1/2] Clean up old generations that are building up in active long-running (still-running) workflows --- Cargo.lock | 7 +- 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 | 258 ++++++++++++++++++++++++- tests/e2e/sql/89_execution_pruning.sql | 191 ++++++++++++++++++ 9 files changed, 512 insertions(+), 23 deletions(-) create mode 100644 tests/e2e/sql/89_execution_pruning.sql diff --git a/Cargo.lock b/Cargo.lock index 998df174..5582d335 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -713,9 +713,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", @@ -3093,7 +3092,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.2", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index ba9cd10a..936f8978 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 5233784e..a77d747e 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -2531,19 +2531,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. Satellite maintenance visits `_origins` registrations made by activity routing for submitted work, in bounded batches under the shared origin connection budget. @@ -2562,6 +2568,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 @@ -2579,12 +2590,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 @@ -2596,9 +2608,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 f969dd47..1e5ca766 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -412,6 +412,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 9b7672e1..28605c6a 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) + 54_reconcile_orphans|76_multi_database_reconcile|89_execution_pruning) echo "reconcile" ;; 88_multi_database_maintenance_backlog) diff --git a/src/lib.rs b/src/lib.rs index fba8c1ef..927161fd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -53,11 +53,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. @@ -290,8 +292,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, @@ -302,7 +304,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 e2c287b7..bc166d33 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; @@ -1126,6 +1126,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 pending = MaintenanceProgress::new(); @@ -1208,6 +1209,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)"); @@ -1278,6 +1300,7 @@ struct MaintenanceProgress { origins: bool, engine: bool, registrations: bool, + executions: bool, } impl MaintenanceProgress { @@ -1288,11 +1311,17 @@ impl MaintenanceProgress { origins: true, engine: true, registrations: true, + executions: true, } } fn has_work(&self) -> bool { - self.control || self.legacy || self.origins || self.engine || self.registrations + self.control + || self.legacy + || self.origins + || self.engine + || self.registrations + || self.executions } fn finish_slice(&mut self, origins: &mut OriginRetentionCursor) -> bool { @@ -1886,6 +1915,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 @@ -2263,6 +2355,166 @@ 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, + 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") @@ -2744,6 +2996,7 @@ mod tests { origins: false, engine: false, registrations: false, + executions: false, }; assert!(!progress.finish_slice(&mut cursor)); assert!(cursor.enter(origin)); @@ -2889,6 +3142,7 @@ mod tests { origins: false, engine: true, registrations: 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 af3dcc864336b0e6ccf69f2dce892306ec00d1e3 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Thu, 15 Oct 2026 17:07:34 +0000 Subject: [PATCH 2/2] Robustness improvements --- Cargo.lock | 11 +- Cargo.toml | 4 +- USER_GUIDE.md | 23 ++- docs/upgrade-testing.md | 25 ++- scripts/test-e2e-docker.sh | 1 + src/lib.rs | 2 +- src/monitoring.rs | 178 ++++++++++++++++---- src/worker.rs | 329 +++++++++++++++++++++++++++---------- 8 files changed, 434 insertions(+), 139 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5582d335..78a4d8a2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -694,9 +694,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", @@ -713,8 +713,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", @@ -3092,7 +3092,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.2", "once_cell", "rustix", "windows-sys 0.61.2", @@ -3287,6 +3287,7 @@ dependencies = [ "bytes", "futures-core", "futures-sink", + "futures-util", "pin-project-lite", "tokio", ] diff --git a/Cargo.toml b/Cargo.toml index 936f8978..573da777 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 a77d747e..572b5a36 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -2568,11 +2568,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 @@ -2612,12 +2616,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 1e5ca766..f06b3c75 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -418,17 +418,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 18873c29..1594d024 100755 --- a/scripts/test-e2e-docker.sh +++ b/scripts/test-e2e-docker.sh @@ -50,6 +50,7 @@ SKIP_TESTS=( "54_reconcile_orphans" "76_multi_database_reconcile" "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/lib.rs b/src/lib.rs index 927161fd..c46dceac 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -95,7 +95,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 = 2; +pub const WORKER_SCHEMA_VERSION: i32 = 3; ::pgrx::pg_module_magic!(name, version); 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 bc166d33..5a6af879 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; @@ -1126,7 +1126,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 pending = MaintenanceProgress::new(); @@ -1211,24 +1211,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 => { @@ -1868,11 +1861,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 @@ -1915,67 +1905,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 @@ -2355,6 +2360,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") @@ -2380,7 +2435,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) @@ -2434,38 +2488,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), @@ -2486,10 +2549,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 @@ -2498,6 +2561,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 {