diff --git a/CHANGELOG.md b/CHANGELOG.md index 5677e7e1..0da4bb65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,33 @@ 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:** 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; 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 - **Start admission quotas (#139):** `df.max_concurrent_per_user` (default diff --git a/Cargo.lock b/Cargo.lock index e821bc7a..fec32c05 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,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.38" +source = "git+https://github.com/microsoft/duroxide-pg?rev=79870890d1a4a14eb840ec1d31d0b6dff0716c0a#79870890d1a4a14eb840ec1d31d0b6dff0716c0a" dependencies = [ "anyhow", "async-trait", @@ -3320,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 c6a7eee5..29c8703b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,8 +36,9 @@ uuid = { version = "1.0", features = ["v4", "serde"] } percent-encoding = "2.3" # duroxide integration -duroxide = "=0.1.30" -duroxide-pg = "=0.1.34" +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 = "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 e9a5fe70..18f15de9 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 @@ -2533,19 +2534,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 +2577,29 @@ 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 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. 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; 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. + +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 @@ -2587,12 +2617,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 +2635,44 @@ 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. +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. 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. 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: -- "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..9e76e32a 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -358,6 +358,95 @@ 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. + 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. + 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). + 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 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. +- **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:** 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). + 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) - **DDL/B1:** The upgrade adds `idx_instances_submitted_by` on 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/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..a8c5f807 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. @@ -95,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); @@ -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/monitoring.rs b/src/monitoring.rs index 649cceaa..28ea5829 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,308 @@ 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. + // execution_snapshot_matches_provider_api checks these column assumptions + // against the real provider migrations and its public metadata API. + 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_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") + .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/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..cc13c9bb 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -1149,6 +1149,8 @@ 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(), 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(); @@ -1232,6 +1234,24 @@ 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 prune_old_executions( + maintenance_pool, &schema, + retention_cutoff_ms(retention), &mut execution_cursor, + &mut execution_batch_size, + ).await { + Ok(more) => more, + Err(error) => { + log!("pg_durable: execution pruning deferred: {error}"); + execution_cursor != previous_cursor + }, + }; + 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 => { log!("pg_durable: reclaimed {reclaimed} orphaned engine record(s)"); @@ -1305,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, @@ -1313,6 +1334,7 @@ struct MaintenanceProgress { engine: bool, registrations: bool, submissions: bool, + executions: bool, } impl MaintenanceProgress { @@ -1324,6 +1346,7 @@ impl MaintenanceProgress { engine: true, registrations: true, submissions: true, + executions: true, } } @@ -1334,6 +1357,7 @@ impl MaintenanceProgress { || self.engine || self.registrations || self.submissions + || self.executions } fn finish_slice(&mut self, origins: &mut OriginRetentionCursor) -> bool { @@ -1880,11 +1904,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 @@ -1927,6 +1948,104 @@ 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, + 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())?; + crate::origin::configure_metadata_transaction(&mut tx) + .await + .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 + 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.0) + .bind(cursor.1) + .bind(i64::from(*batch_size)) + .fetch_optional(&mut *tx) + .await + .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(*batch_size) + .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}") + })?; + 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}")); + } + }; + tx.commit().await.map_err(|error| error.to_string())?; + 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 batch.executions_deleted > 0 { + log!( + "pg_durable: pruned {} old execution(s) and {} history event(s)", + batch.executions_deleted, + batch.events_deleted, + ); + } + Ok(batch.executions_scanned == i64::from(*batch_size)) +} + /// 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 +2578,481 @@ 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,3) + ON CONFLICT(singleton) DO UPDATE SET schema_version=3;", + ) + .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") + .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(); + 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(), 0); + let mut batch_size = EXECUTION_PRUNE_BATCH_SIZE; + assert!(prune_old_executions( + 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, &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, &mut batch_size) + .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 = (String::new(), 0); + pages = 0; + while prune_old_executions(pool, "execution_pruning_test", cutoff, &mut cursor, &mut batch_size) + .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), + (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, "execution_pruning_test", cutoff, &mut cursor, &mut batch_size, + ) + .await + .unwrap(), "an exhausted scan must not restart from protected histories"); + sqlx::query("DROP SCHEMA execution_pruning_test CASCADE") + .execute(pool) + .await + .unwrap(); + pool.close().await; + }); + } + + #[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") + .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, table) in [("a", "history"), ("b", "history"), ("a", "instances"), ("b", "instances")] { + let mut blocker = store.pool().begin().await.unwrap(); + 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 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); + 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 mut batch_size = 1; + let error = tokio::time::timeout(Duration::from_secs(7), + 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") + .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); + 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(); + 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; + }); + } + + #[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 +3564,7 @@ mod tests { engine: false, registrations: false, submissions: false, + executions: false, }; assert!(!progress.finish_slice(&mut cursor)); assert!(cursor.enter(origin)); @@ -3116,6 +3711,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;