diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 57db9b6a9..33ca9f07b 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -523,8 +523,8 @@ jobs: -e DB_USERNAME=root -e DB_PASSWORD=parity-disposable-password \ -e CACHE_STORE=file -e QUEUE_CONNECTION=database -e DW_AUTH_TOKEN=parity-test-token \ "$layout" artisan server:bootstrap --force > parity-evidence/published-bootstrap.txt - # Compiling the library tests with two jobs peaks near 4.2 GB; a 4 GB limit OOM-killed rustc (exit 101). - docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=6g --memory-swap=6g \ + # Compiling the test targets with two jobs peaked at 6.8 GB cold (cgroup memory.peak, October 27); a 6 GB limit OOM-killed rustc (signal 9, PR #801). + docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=8g --memory-swap=8g \ -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target -e CARGO_BUILD_JOBS=2 -e DW_PARITY_PART -e DW_EXECUTION_SHARDS \ -e DW_TEST_MYSQL_URL=mysql://root:parity-disposable-password@parity-mysql:3306/php_ref \ -e DW_EXECUTION_MYSQL_URL=mysql://root:parity-disposable-password@parity-mysql:3306/php_ref \ @@ -1183,8 +1183,8 @@ jobs: -e DB_USERNAME=parity -e DB_PASSWORD=parity-disposable-password \ -e CACHE_STORE=file -e QUEUE_CONNECTION=database -e DW_AUTH_TOKEN=parity-test-token \ "$layout" artisan server:bootstrap --force > parity-evidence/published-bootstrap.txt - # Compiling the library tests with two jobs peaks near 4.2 GB; a 4 GB limit OOM-killed rustc (exit 101). - docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=6g --memory-swap=6g \ + # Compiling the test targets with two jobs peaked at 6.8 GB cold (cgroup memory.peak, October 27); a 6 GB limit OOM-killed rustc (signal 9, PR #801). + docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=8g --memory-swap=8g \ -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target -e CARGO_BUILD_JOBS=2 -e DW_PARITY_PART \ -e DW_TEST_POSTGRES_URL=postgres://parity:parity-disposable-password@parity-pg:5432/php_ref \ -e DW_EXECUTION_POSTGRES_URL=postgres://parity:parity-disposable-password@parity-pg:5432/php_ref \ @@ -1862,8 +1862,8 @@ jobs: } trap rust_test_failures ERR mkdir -p cargo-home rust-target parity-evidence - # Compiling the library tests with two jobs peaks near 4.2 GB; a 4 GB limit OOM-killed rustc (exit 101). - docker run --rm --user="$(id -u):$(id -g)" --cpus=2 --memory=6g --memory-swap=6g \ + # Compiling the test targets with two jobs peaked at 6.8 GB cold (cgroup memory.peak, October 27); a 6 GB limit OOM-killed rustc (signal 9, PR #801). + docker run --rm --user="$(id -u):$(id -g)" --cpus=2 --memory=8g --memory-swap=8g \ -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target -e CARGO_BUILD_JOBS=2 \ -e DW_UPGRADE_TEST_WRITABLE=/evidence -e DW_UPGRADE_TEST_READONLY=/source/parity-evidence \ -v "$PWD:/source:ro" -v "$PWD/cargo-home:/cargo" -v "$PWD/rust-target:/target" \ diff --git a/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md new file mode 100644 index 000000000..aa41777f1 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md @@ -0,0 +1,76 @@ +# October 27, message-stream retention + +- **Before.** Native's retention pass held any run that had + `workflow_messages` rows or whose instance had inbound message streams, with + `message_stream_cleanup_unavailable`. Published PHP's + `MessageStreamRetentionCleanup` was never ported. So a namespace that used + message streams kept those runs' details forever, and takeover refused any + PHP database with stream rows. +- **Change.** `prune_run` plans and applies the stream cleanup in the run's + own prune transaction, under the transition lock. The plan follows PHP's + `planForRun`, read before the run's summary goes: + - **No inbound streams on the instance:** nothing to do. + - **The instance's current run is terminal and no other run summary + remains:** every item of the instance's streams is deleted, then the + streams. Reported as `message_stream_items_deleted` and + `message_streams_deleted`. + - **Otherwise:** on each stream whose `cursor_checkpoint_run_id` is a + retained later run (another run summary of the instance, or its current + run), the items this run consumed lose their payloads. That covers items + at or before the stream's cursor that are not released yet: `payload_blob` + becomes `''` and `payload_released_at` is set. The item rows stay as + identity tombstones. Reported as `message_stream_items_compacted`. + `cleanup_blocked_*` markers naming this run are cleared. + - The run's `workflow_messages` rows go with the run's other details + (`messages_deleted`), as in PHP. Before, such a run was held. + - The current run counts as retained only when its run row exists, as in + PHP. + - A run stays held with `external_payload_cleanup_unavailable` when an + inbound item on its instance holds an external reference. Nothing writes + one today; the hold keeps a later change from stranding stored objects. +- **Not ported.** PHP's `markBlocked`, which marks a stream when + external-payload cleanup blocks a prune. Native never deletes external + objects: a run holding an external reference stays held as + `external_payload_cleanup_unavailable`, and native refuses external + references as message input. +- **Tests** on SQLite, PostgreSQL, MySQL and MariaDB: + - `retention_releases_consumed_messages_and_removes_the_last_runs_streams`: + pruning a continued run compacts the two messages it consumed and keeps the + third. Pruning the last run deletes the stream and its three items. + - `retention_keeps_consumed_messages_no_retained_checkpoint_covers`: without + a retained checkpoint, nothing is compacted and the stream stays. + +## Takeover keeps message and workflow streams + +- **Before.** Takeover refused any source with rows in + `workflow_messages`, either kind of stream, or their items. +- **Change.** Stream rows carry over unchanged. + - Each closed or errored durable stream gets its `dw_stream_reclaims` + deadline: 2.11.3's `reclaim_at`, or `closed_at` plus `retention_seconds` + (3600 unless set) for earlier layouts. Reported as + `stream_reclaims_recorded`; MySQL and MariaDB do it in journal step 10. + - Inspection reports `stream_reference_inspection`. A stream row holding an + external payload reference refuses with + `upgrade_source_external_payloads_unsupported`. + - Only `workflow_messages` stays refused, now as + `upgrade_source_workflow_messages_unsupported`. +- **Tests.** + - `takeover_keeps_php_streams_and_records_their_reclaim_deadlines` (SQLite). + - The 2.11.3 takeover test now copies a recorded deadline. + - The refusal list covers a durable stream item holding an external + reference. + - Takeover qualification passes on PostgreSQL, MySQL and MariaDB, + including a MySQL takeover killed after step 9. + +## Review fixes before the PR + +An independent review found no serious defects. Fixed: +- A resumed MySQL takeover decides step 10's deadline source by layout, not + by whether `reclaim_at` still exists. A 2.11.3 takeover killed after that + column's drop had already copied every deadline, so it adopts nothing more + instead of adding `closed_at + retention` rows for streams PHP had emptied. +- The current run's missing row and the inbound-item hold above. +- The payload scan's comment no longer says stream item rows are refused. +- The retention test now puts the cursor exactly on the pruned run's last + consumed message (so `<=` matters) and points the instance at a missing + current run. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 5c83a58a6..917de2887 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -14,92 +14,60 @@ qualify a later tree. ## Current state -Updated October 24, 2026. - -- **Merged October 24.** Train - [#729](https://github.com/durable-workflow/server/pull/729) (merge - `61219609b`): - - MySQL statement closes are sent with the next command - ([#717](https://github.com/durable-workflow/server/pull/717)), ending the - MySQL-family stalls of [#714](https://github.com/durable-workflow/server/issues/714); - - policy maps that are not JSON objects are refused, like PHP #705, and a - namespace quota map that is not UTF-8 is refused under either name - ([#719](https://github.com/durable-workflow/server/pull/719)); - - worker history shows consumed signal sequences, hydrating only each page's - own signals, plus full task snapshots - ([#710](https://github.com/durable-workflow/server/pull/710)); - - a newer push cancels a superseded Shared Server fixtures run - ([#724](https://github.com/durable-workflow/server/pull/724)). -- **Merged October 20-23.** These are summarized in the checkpoint history: - - backup-first takeover on every database; - - multi-node kill and restart; - - ownership guards, search attributes, namespace update, history export, - retention and activity timeouts; - - the October 21-23 trains; - - the move to the published 2.11.1 reference; - - the GHSA-6q48-6c7w-9gw2 fixes. - - Each train's PR holds its evidence. -- **In review or qualifying, in merge order.** - 1. CI speed and the merge queue: - - [#741](https://github.com/durable-workflow/server/pull/741) runs the - SDK direction profiles in parallel lanes, about 14 minutes off every - database job, and fixes a timestamp-comparison test flake; - - [#742](https://github.com/durable-workflow/server/pull/742) adds the - merge-queue prerequisites (`merge_group` triggers, one fixtures - result, and fixture cells skipped when a change cannot affect them). - 2. This train: operator metrics read across PHP's redispatch steps - ([#708](https://github.com/durable-workflow/server/pull/708)), - query-authority clients that wait for the acknowledgement - ([#720](https://github.com/durable-workflow/server/pull/720)), and the - machine-checked route list below. - 3. Condition waits without a key, task snapshots, the SQLite patch cohort and - message streams (#713, #718, #722, #709). - 4. Update validation with PHP's poll request ID contract - ([#712](https://github.com/durable-workflow/server/pull/712)), then the - held-poll registration fence - ([#739](https://github.com/durable-workflow/server/pull/739), #734) and - worker sessions - ([#740](https://github.com/durable-workflow/server/pull/740), #736), - stacked in that order. - 5. Activity cancellation acknowledgement, and pip, Composer and Cargo trust - in CI ([#673](https://github.com/durable-workflow/server/pull/673)). +Updated October 27, 2026. + +- **Merged October 26-27, through the merge queue.** + - [#790](https://github.com/durable-workflow/server/pull/790): prepared + local-activity refusals. + - [#793](https://github.com/durable-workflow/server/pull/793): Nexus catalog + administration, service-call execute, show and cancel, shared admission + limits, and catalog call reads. + - [#795](https://github.com/durable-workflow/server/pull/795): takeover of + published 2.11.1 databases (fresh and upgraded from 2.9.x), keeping search + attribute definitions, worker sessions and service calls, plus the repair + pass's service-call deadline sweep. + - [#797](https://github.com/durable-workflow/server/pull/797): native's own + maintenance repair loop (opt-in, `DW_RUST_MAINTENANCE_LOOP=1`); takeover + keeps runtime credentials and update validation tasks; takeover of + published 2.11.3, fresh and upgraded from 2.9.x. +- **In the queue or next, in order.** + 1. [#799](https://github.com/durable-workflow/server/pull/799): the fixture + adapters' Cargo cache, saved only by pushes to main. + 2. Message-stream retention (PHP's `MessageStreamRetentionCleanup`), and + takeover that keeps message and workflow streams. + 3. The native release-candidate image (`Dockerfile.native`), its release + workflow and the takeover runbook's image commands. + 4. The parity reference moves to published 2.11.3. Native already matches it + on 62 of 65 fixtures. The move waits on two published 2.11.3 defects + General owns: the storage diagnostic (#789) and cross-caller idempotent + replay (Workflow #931). +- **Still refused at takeover:** workflow-to-workflow messages + (`workflow_messages`, written only by embedded PHP workflows) and external + payload references. PHP keeps references in about 30 table and column + locations in their URI form, so taking them over is its own project; the + recommendation is that the release candidate keeps refusing them. - **Shared Server fixtures.** Native, embedded and published PHP 2.11.1 are compared on SQLite, MySQL 8.0, MariaDB 10.11 and PostgreSQL 17. Where the published reference itself misbehaves, an explicit boundary separates it - from native, and the PHP owner has an issue for each case: - - `operator-metrics` across PHP's redispatch window (#708); - - the published-SDK patch cohort's SQLite poll refusal - ([#721](https://github.com/durable-workflow/server/issues/721), #722); - - update validation under SQLite write locks - ([#726](https://github.com/durable-workflow/server/issues/726), fixed in - PHP by #730); - - `worker-sdk-persistent-pressure`'s shutdown window - ([#728](https://github.com/durable-workflow/server/issues/728)); - - a retried activity's cooperative cancellation - ([Workflow #923](https://github.com/durable-workflow/workflow/issues/923)); - - a reacquired worker session keeping its old failure reason - ([#737](https://github.com/durable-workflow/server/issues/737)). + from native, and the PHP owner has an issue for each case. Recent ones: + - #796, `service-calls` under full-corpus SQLite load; + - #794, workflow-bound Nexus handlers. No expectation is edited to make native pass. - **Parity surface.** Native answers 174 of the 175 API route methods in - published 2.11.1's route table, with message streams, durable workflow - stream writes, update validation, activity cancellation acknowledgement, - worker sessions, workflow-task cancellation scopes, service catalog - administration, Nexus service-call execution, show and cancel for the - handler kinds that stay accepted, caller-indexed Nexus operation reads, - bridge adapter webhooks, and embedded v2 and Waterline v1 workflow import. - Eight of those are the prepared local-activity routes, which it answers - with PHP's refusal for a bridge without the prepared role. An execution - test sends each one through the native router, and the ones still - answering `501 rust_capability_not_implemented` must match + published 2.11.1's route table. Eight of those are the prepared + local-activity routes, which it answers with PHP's refusal for a bridge + without the prepared role. Nexus workflow-bound handler kinds answer 501 + until #794 settles their contract. An execution test sends each route + through the native router, and the ones still answering + `501 rust_capability_not_implemented` must match [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt). - Unlike published PHP, native's embedded v2 import refuses a bundle whose - rows reference rows outside it, links only runs of one namespace, and reads - the signing key only from the request body - ([#765](https://github.com/durable-workflow/server/issues/765)). -- **Next action.** Land the items above in order, keep porting the unserved - routes fixture first, and package the native binary into the server image. +- **Remaining for the release candidate:** + - the 2.11.3 reference move; + - the scope question with the operator: prepared local activities, + workflow-bound Nexus handlers and external payload takeover; + - publishing the native image (General runs the release); + - differential and performance qualification of the published image. ## API surface still missing @@ -412,14 +380,14 @@ contract fixture before its correction. workflow task guard does not check registration freshness yet, and the local activity and cancellation scope checkpoints still use the older fence. Add a fixture for each before changing them. -- **Retention of runs with external payloads or message streams.** - Published PHP deletes a pruned run's external payload objects and releases - its message stream items. It also compacts its instance's consumed inbound - stream items, and removes the instance's streams with its last run. Native - does not clean any of these up yet. It skips such a run, including any run - whose instance has inbound streams, with - `external_payload_cleanup_unavailable` or - `message_stream_cleanup_unavailable`, and keeps its details. +- **Retention of runs with external payloads.** Published PHP deletes a + pruned run's external payload objects. Native does not delete them yet: it + skips such a run with `external_payload_cleanup_unavailable` and keeps its + details. A run whose instance has inbound message streams is pruned as PHP's + `MessageStreamRetentionCleanup` prunes it (October 27): the messages it + consumed lose their payloads once a retained later run's cursor checkpoint + covers them, and the instance's last retained run, once terminal, takes the + instance's streams with it. - **Deeply nested JSON bodies.** Published PHP decodes JSON up to 512 levels deep. Native checks every body as PHP does before any route, but its handlers parse with serde_json's 128-level limit. A body nested 129 to 512 diff --git a/docs/rust-server-upgrade-takeover.md b/docs/rust-server-upgrade-takeover.md index 7439f085a..a121504f5 100644 --- a/docs/rust-server-upgrade-takeover.md +++ b/docs/rust-server-upgrade-takeover.md @@ -77,6 +77,20 @@ inspection is `unqualified` and takeover refuses with `upgrade_source_external_payload_storage_unsupported` before any change. A disabled policy does not block takeover. +Inbound message streams and durable workflow streams carry over unchanged: +native serves them from PHP's own tables. Each closed or errored durable +stream gets the reclaim deadline native's retention pass reads, +`dw_stream_reclaims`: PHP 2.11.3's recorded `reclaim_at`, or for earlier +releases its `closed_at` plus its retention (an hour unless set), as 2.11.3's +migration backfills it. Every inspection reports +`stream_reference_inspection`: the number of stream items, durable streams +and inbound messages holding an external payload reference. Native does not +resolve those, so any such row makes the inspection `unqualified` and takeover +refuses with `upgrade_source_external_payloads_unsupported`. +Workflow-to-workflow messages (`workflow_messages`) are not delivered by +native yet, so a source holding any refuses with +`upgrade_source_workflow_messages_unsupported`. + PHP 2.11.3 (Core 2.14) schedules a service call's handler in the `workflow_service_call_dispatches` outbox and dispatches it in a later pass. Native runs the handler when it accepts the call, so it cannot resume a diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs index a23e2ada0..1e583efed 100644 --- a/rust/src/config_audit.rs +++ b/rust/src/config_audit.rs @@ -502,7 +502,10 @@ pub fn audit(environment: &BTreeMap) -> Report { /// Names native reads that PHP's contract does not list. fn native_only(name: &str) -> bool { - matches!(name, "DW_BIND_ADDRESS" | "DW_ENV" | "DW_RUST_EXPERIMENTAL" | "DW_UPGRADE_BACKUP_FINGERPRINT") + matches!( + name, + "DW_BIND_ADDRESS" | "DW_ENV" | "DW_RUST_EXPERIMENTAL" | "DW_RUST_MAINTENANCE_LOOP" | "DW_UPGRADE_BACKUP_FINGERPRINT" + ) || ["DW_RUST_UPGRADE_QUEUE_", "DW_SCHEMA_TEST_", "DW_TAKEOVER_TEST_", "DW_TEST_"] .iter() .any(|prefix| name.starts_with(prefix)) @@ -820,6 +823,7 @@ mod tests { ("DW_WORKFLOW_MEMO_MIGRATION_RECOVERY", ""), ("DW_ENV_AUDIT_STRICT", "1"), ("DW_RUST_EXPERIMENTAL", "1"), + ("DW_RUST_MAINTENANCE_LOOP", "1"), ("DW_BIND_ADDRESS", "0.0.0.0:8080"), ]; let report = audit(&environment(&shipped)); diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 179490386..75340c6a7 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -6,6 +6,26 @@ use std::borrow::Cow; use super::{Result, mysql, postgres, schema}; +/// Counts every durable stream item, durable stream and inbound message that +/// holds an external payload reference. The arguments name the JSON payload +/// and metadata columns as the backend reads them as text. +macro_rules! stream_references { + ($payload:literal, $metadata:literal) => { + concat!( + "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ", + super::retention::reference!($payload), + ")+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ", + super::retention::reference!($metadata), + ")+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ", + super::retention::reference!("payload_blob"), + ") AS n" + ) + }; +} + +/// `STREAM_REFERENCES` for backends that store JSON as text. +const PLAIN_STREAM_REFERENCES: &str = stream_references!("payload", "metadata"); + pub(super) trait Backend: Database { const PHP_CONNECTION: &str; type EncodedTime: for<'q> Encode<'q, Self> + Type + Send + Sync + 'static; @@ -94,6 +114,10 @@ pub(super) trait Backend: Database { // reference, which native does not settle (Server #774). PostgreSQL // stores these payloads as `json`, matched through its text. const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload::text"), ")"); + // Every stream item, durable stream and inbound message that holds an + // external payload reference, which takeover refuses as it refuses any + // other external payload. + const STREAM_REFERENCES: &str = stream_references!("payload::text", "metadata::text"); const RUN_STREAM_REFERENCES: &str = concat!( "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload::text"), "))", "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata::text"), ")" @@ -217,6 +241,7 @@ ORDER BY t.available_order,t.id LIMIT 100 "#; const SELECTION_RESOLUTION_SQL: &str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND json_extract(payload,'$.selection_group_id')=$2 ORDER BY sequence LIMIT 1"; const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")"); + const STREAM_REFERENCES: &str = PLAIN_STREAM_REFERENCES; const RUN_STREAM_REFERENCES: &str = concat!( "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), "))", "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata"), ")" @@ -477,6 +502,7 @@ impl Backend for MySql { const STREAM_SWEEP_BOUNDARY: &str = "SELECT `offset` FROM workflow_durable_stream_items WHERE stream_id=$1 ORDER BY `offset` LIMIT 1 OFFSET $2"; const STREAM_SWEEP_THROUGH: &str = "DELETE FROM workflow_durable_stream_items WHERE stream_id=$1 AND `offset`<=$2"; const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")"); + const STREAM_REFERENCES: &str = PLAIN_STREAM_REFERENCES; const RUN_STREAM_REFERENCES: &str = concat!( "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), "))", "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata"), ")" diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs index 815835b3c..3409aede0 100644 --- a/rust/src/runtime/operator_metrics.rs +++ b/rust/src/runtime/operator_metrics.rs @@ -1353,7 +1353,7 @@ where return Ok(Skipped(reason)); } let mut tx = self.begin().await?; - let run = Self::query("SELECT status,closed_at FROM workflow_runs WHERE id=$1") + let run = Self::query("SELECT status,closed_at,workflow_instance_id FROM workflow_runs WHERE id=$1") .bind(run_id) .fetch_one(&mut *tx) .await?; @@ -1366,10 +1366,8 @@ where } let mut deleted = serde_json::Map::new(); deleted.insert("run_deleted".into(), json!(0)); - // A run with message streams is held above, so none are released. - for key in retention::STREAM_COUNTS { - deleted.insert(key.into(), json!(0)); - } + // Planned before the run's own summary goes, as PHP plans it. + let streams = Self::stream_retention_plan(&mut tx, self.namespace.as_ref(), run_id, &DB::string(&run, "workflow_instance_id")?).await?; for (key, sql, binds) in retention::PRUNED { let mut query = Self::query(sql); for _ in 0..*binds { @@ -1383,6 +1381,10 @@ where deleted.insert((*key).into(), json!(affected)); } } + let [streams_deleted, items_deleted, items_compacted] = Self::apply_stream_retention(&mut tx, run_id, &streams).await?; + for (key, value) in retention::STREAM_COUNTS.into_iter().zip([streams_deleted, items_deleted, items_compacted]) { + deleted.insert(key.into(), json!(value)); + } // The run's history is gone; no cached read of it may survive. tx.scope_history = None; Self::query("UPDATE workflow_runs SET details_pruned_at=$1 WHERE id=$2 AND details_pruned_at IS NULL") @@ -1394,9 +1396,119 @@ where Ok(Pruned(deleted)) } + /// `MessageStreamRetentionCleanup::planForRun`: what pruning `run_id` + /// releases of its instance's inbound message streams. + /// + /// An active or continued instance keeps its streams, cursor checkpoint, + /// pending items and identity rows. An item the run consumed loses its + /// payload only once a retained later run's cursor checkpoint covers it, + /// so replay never needs it again. A terminal instance keeps all stream + /// state until its last retained run goes, which removes every stream. + async fn stream_retention_plan( + tx: &mut DB::Connection, + namespace: &str, + run_id: &str, + instance_id: &str, + ) -> Result { + let streams = Self::query("SELECT id,cursor_position,cursor_checkpoint_run_id FROM workflow_inbound_streams WHERE namespace=$1 AND workflow_instance_id=$2 ORDER BY id") + .bind(namespace) + .bind(instance_id) + .fetch_all(&mut *tx) + .await?; + let mut plan = retention::StreamPlan::default(); + if streams.is_empty() { + return Ok(plan); + } + let current = Self::query("SELECT current_run_id FROM workflow_instances WHERE id=$1 AND namespace=$2") + .bind(instance_id) + .bind(namespace) + .fetch_optional(&mut *tx) + .await? + .map(|row| DB::optional_string(&row, "current_run_id")) + .transpose()? + .flatten(); + // The current run counts only when its row exists, as in PHP: an + // instance pointing at a missing run retains nothing through it. + let current_status = match ¤t { + Some(current) => Self::query("SELECT status FROM workflow_runs WHERE id=$1") + .bind(current) + .fetch_optional(&mut *tx) + .await? + .map(|row| DB::string(&row, "status")) + .transpose()?, + None => None, + }; + let current_terminal = current_status.as_deref().is_some_and(retention::terminal); + let current = current.filter(|_| current_status.is_some()); + let mut retained = Vec::new(); + for row in Self::query("SELECT id FROM workflow_run_summaries WHERE namespace=$1 AND workflow_instance_id=$2 AND id<>$3") + .bind(namespace) + .bind(instance_id) + .bind(run_id) + .fetch_all(&mut *tx) + .await? + { + retained.push(DB::string(&row, "id")?); + } + for row in &streams { + plan.stream_ids.push(DB::number(row, "id")?); + } + if current_terminal && retained.is_empty() { + plan.delete_instance_state = true; + return Ok(plan); + } + if let Some(current) = current.filter(|current| current != run_id && !retained.contains(current)) { + retained.push(current); + } + for row in &streams { + if DB::optional_string(row, "cursor_checkpoint_run_id")?.is_some_and(|checkpoint| retained.contains(&checkpoint)) { + plan.checkpointed.push((DB::number(row, "id")?, DB::number(row, "cursor_position")?)); + } + } + Ok(plan) + } + + /// `MessageStreamRetentionCleanup::apply`, in the prune's transaction: + /// streams deleted, items deleted and items compacted. + async fn apply_stream_retention(tx: &mut DB::Connection, run_id: &str, plan: &retention::StreamPlan) -> Result<[u64; 3]> { + if plan.stream_ids.is_empty() { + return Ok([0, 0, 0]); + } + let mut counts = [0, 0, 0]; + if plan.delete_instance_state { + for stream_id in &plan.stream_ids { + counts[1] += DB::affected(Self::query("DELETE FROM workflow_inbound_stream_items WHERE stream_id=$1").bind(*stream_id).execute(&mut *tx).await?); + counts[0] += DB::affected(Self::query("DELETE FROM workflow_inbound_streams WHERE id=$1").bind(*stream_id).execute(&mut *tx).await?); + } + return Ok(counts); + } + let at = now(); + for (stream_id, cursor) in &plan.checkpointed { + counts[2] += DB::affected( + Self::query("UPDATE workflow_inbound_stream_items SET payload_blob='',payload_released_at=$1,updated_at=$2 WHERE stream_id=$3 AND consumed_run_id=$4 AND consumed_at IS NOT NULL AND payload_released_at IS NULL AND position<=$5") + .bind(DB::bind_time(at)) + .bind(DB::bind_time(at)) + .bind(*stream_id) + .bind(run_id) + .bind(*cursor) + .execute(&mut *tx) + .await?, + ); + } + for stream_id in &plan.stream_ids { + Self::query("UPDATE workflow_inbound_streams SET cleanup_blocked_at=NULL,cleanup_blocked_reason=NULL,cleanup_blocked_run_id=NULL,updated_at=$1 WHERE id=$2 AND cleanup_blocked_run_id=$3") + .bind(DB::bind_time(at)) + .bind(*stream_id) + .bind(run_id) + .execute(&mut *tx) + .await?; + } + Ok(counts) + } + /// What keeps a closed run's details: an activity it abandoned that is - /// still open, a message stream, or a stored external payload reference, - /// whose cleanup native does not perform yet. + /// still open, or a stored external payload reference, whose cleanup + /// native does not perform yet. async fn retention_hold(&self, run_id: &str) -> Result> { if self.abandoned_activity_open(run_id).await? { return Ok(Some("detached_activity_still_open")); diff --git a/rust/src/runtime/retention.rs b/rust/src/runtime/retention.rs index 3053e0c8b..1db35af65 100644 --- a/rust/src/runtime/retention.rs +++ b/rust/src/runtime/retention.rs @@ -3,9 +3,10 @@ //! with `details_pruned_at`. //! //! Native never deletes stored external payload objects. A run that holds a -//! reference to one, an open abandoned activity or a message stream is -//! skipped with its reason, so a prune never strands or loses data PHP would -//! have handled. +//! reference to one, or an open abandoned activity, is skipped with its +//! reason, so a prune never strands or loses data PHP would have handled. +//! Inbound message streams are released as PHP's +//! `MessageStreamRetentionCleanup` releases them. use super::laravel_validation::{Rule, failure_messages, rule_errors}; use chrono::{DateTime, Duration, Utc}; use serde_json::{Map, Value, json}; @@ -64,13 +65,18 @@ macro_rules! reference { pub(super) use reference; /// The run's payload columns that can hold an external reference. History -/// events repeat these payloads, so they are not scanned again. +/// events repeat these payloads, so they are not scanned again. Inbound +/// message stream items on the run's instance count too: retention blanks or +/// deletes them, and native never deletes the stored object behind one. +/// Nothing writes such an item today (append and takeover refuse external +/// references), so this only keeps a later change from stranding objects. const REFERENCES_SQL: &str = concat!( "SELECT (SELECT COUNT(*) FROM workflow_runs WHERE id=$1 AND (", reference!("arguments"), " OR ", reference!("output"), "))", "+(SELECT COUNT(*) FROM workflow_commands WHERE workflow_run_id=$2 AND ", reference!("payload"), ")", "+(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$3 AND ", reference!("arguments"), ")", "+(SELECT COUNT(*) FROM workflow_updates WHERE workflow_run_id=$4 AND (", reference!("arguments"), " OR ", reference!("result"), "))", "+(SELECT COUNT(*) FROM activity_executions WHERE workflow_run_id=$5 AND (", reference!("arguments"), " OR ", reference!("result"), "))", + "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE stream_id IN (SELECT id FROM workflow_inbound_streams WHERE workflow_instance_id IN (SELECT workflow_instance_id FROM workflow_runs WHERE id=$6)) AND ", reference!("payload_blob"), ")", ); /// The run's durable workflow streams, deleted with it as PHP's @@ -87,16 +93,21 @@ pub(super) const STREAM_COUNTS: [&str; 3] = ["message_streams_deleted", "message_stream_items_deleted", "message_stream_items_compacted"]; /// Counts that hold a run's details, the run-id binds of each, and the -/// skip reason. An abandoned activity's hold is read from history. A run -/// with messages or whose instance has inbound message streams is held too: -/// native does not compact or remove that state as PHP's -/// MessageStreamRetentionCleanup does yet, so it keeps the run rather than -/// orphan it. Durable workflow streams go with the run (`DURABLE_STREAMS`) -/// unless they hold an external reference (`Backend::RUN_STREAM_REFERENCES`). -pub(super) const HOLDS: &[(&str, usize, &str)] = &[ - ("SELECT (SELECT COUNT(*) FROM workflow_messages WHERE workflow_run_id=$1)+(SELECT COUNT(*) FROM workflow_inbound_streams WHERE workflow_instance_id IN (SELECT workflow_instance_id FROM workflow_runs WHERE id=$2))", 2, "message_stream_cleanup_unavailable"), - (REFERENCES_SQL, 5, "external_payload_cleanup_unavailable"), -]; +/// skip reason. An abandoned activity's hold is read from history. Durable +/// workflow streams go with the run (`DURABLE_STREAMS`) unless they hold an +/// external reference (`Backend::RUN_STREAM_REFERENCES`). +pub(super) const HOLDS: &[(&str, usize, &str)] = &[(REFERENCES_SQL, 6, "external_payload_cleanup_unavailable")]; + +/// `MessageStreamRetentionCleanup::planForRun`'s plan for one run: the +/// instance's inbound streams, whether they all go with the instance's last +/// retained run, and otherwise each stream whose cursor checkpoint belongs +/// to a retained later run, with that cursor. +#[derive(Default)] +pub(super) struct StreamPlan { + pub(super) stream_ids: Vec, + pub(super) delete_instance_state: bool, + pub(super) checkpointed: Vec<(i64, i64)>, +} /// `server.history.retention_days`. fn default_days() -> i64 { diff --git a/rust/src/runtime/upgrade.rs b/rust/src/runtime/upgrade.rs index cbd0ffa98..310bd7b34 100644 --- a/rust/src/runtime/upgrade.rs +++ b/rust/src/runtime/upgrade.rs @@ -3,6 +3,7 @@ pub mod external_reference; pub mod fingerprint; pub mod takeover; mod dispatch_scan; +mod stream_scan; mod payload_scan; mod schedule_scan; mod storage_scan; @@ -255,6 +256,7 @@ fn report( schedule_inspection: Value, storage_inspection: Value, dispatch_inspection: Value, + stream_inspection: Value, ) -> Result { let digest = Sha256::digest(serde_json::to_vec(catalog)?) .iter() @@ -291,6 +293,10 @@ fn report( report["status"] = json!("unqualified"); } report["service_call_dispatch_inspection"] = dispatch_inspection; + if stream_inspection["references"] != 0 { + report["status"] = json!("unqualified"); + } + report["stream_reference_inspection"] = stream_inspection; Ok(report) } @@ -466,6 +472,7 @@ pub(super) async fn sqlite_inventory( let schedule_inspection = schedule_scan::inspect::(&mut *tx).await?; let storage_inspection = storage_scan::inspect::(&mut *tx).await?; let dispatch_inspection = dispatch_scan::inspect::(&mut *tx, &counts).await?; + let stream_inspection = stream_scan::inspect::(&mut *tx).await?; report( "sqlite", &catalog, @@ -476,6 +483,7 @@ pub(super) async fn sqlite_inventory( schedule_inspection, storage_inspection, dispatch_inspection, + stream_inspection, ) } @@ -741,6 +749,7 @@ pub(super) async fn mysql_inventory( let schedule_inspection = schedule_scan::inspect::(&mut *tx).await?; let storage_inspection = storage_scan::inspect::(&mut *tx).await?; let dispatch_inspection = dispatch_scan::inspect::(&mut *tx, &counts).await?; + let stream_inspection = stream_scan::inspect::(&mut *tx).await?; // MySQL's metadata is not an MVCC snapshot. Refuse a changed visible // catalog rather than joining counts to a different schema. if mysql_source_catalog(&mut *tx, fenced).await?.0 != catalog { @@ -759,6 +768,7 @@ pub(super) async fn mysql_inventory( schedule_inspection, storage_inspection, dispatch_inspection, + stream_inspection, ) } @@ -828,6 +838,7 @@ pub(super) async fn postgres_inventory( let schedule_inspection = schedule_scan::inspect::(&mut *tx).await?; let storage_inspection = storage_scan::inspect::(&mut *tx).await?; let dispatch_inspection = dispatch_scan::inspect::(&mut *tx, &counts).await?; + let stream_inspection = stream_scan::inspect::(&mut *tx).await?; report( "pgsql", &catalog, @@ -838,6 +849,7 @@ pub(super) async fn postgres_inventory( schedule_inspection, storage_inspection, dispatch_inspection, + stream_inspection, ) } diff --git a/rust/src/runtime/upgrade/payload_scan.rs b/rust/src/runtime/upgrade/payload_scan.rs index 412c2b91e..f3cf64122 100644 --- a/rust/src/runtime/upgrade/payload_scan.rs +++ b/rust/src/runtime/upgrade/payload_scan.rs @@ -222,8 +222,10 @@ where "references_metadata_supported":references_supported, "diagnostics":diagnostics, "diagnostics_omitted":omitted, "inline_values_verified":inline_supported, "nested_payloads_verified":false, - // These tables also hold codec-tagged payloads; takeover refuses a - // source with any of their rows, since native lacks the features. + // These tables also hold codec-tagged payloads. Takeover carries + // their rows over unchanged without validating each inline value; + // a stream row holding an external reference is refused through + // `stream_reference_inspection`. "other_root_payload_columns_verified":false, "unscanned_payload_codec_tables":["workflow_durable_stream_items","workflow_inbound_stream_items","workflow_update_validation_tasks"], "inline_validation_limits":{"encoded_frame_bytes":65536,"traversal_depth":64, diff --git a/rust/src/runtime/upgrade/stream_scan.rs b/rust/src/runtime/upgrade/stream_scan.rs new file mode 100644 index 000000000..7f9b5d8da --- /dev/null +++ b/rust/src/runtime/upgrade/stream_scan.rs @@ -0,0 +1,19 @@ +//! Message and workflow stream rows that hold an external payload +//! reference. Native carries streams over unchanged, but it neither resolves +//! nor deletes stored external payload objects, so takeover refuses such a +//! row as it refuses any other external payload. The report counts rows and +//! never names streams. +use crate::runtime::{Result, backend::Backend}; +use serde_json::{Value, json}; +use sqlx::{Executor, IntoArguments}; + +pub(super) async fn inspect(connection: &mut DB::Connection) -> Result +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, +{ + let row = sqlx::query(sqlx::AssertSqlSafe(DB::statement(DB::STREAM_REFERENCES))) + .fetch_one(&mut *connection) + .await?; + Ok(json!({"references": DB::number(&row, "n")?})) +} diff --git a/rust/src/runtime/upgrade/takeover.rs b/rust/src/runtime/upgrade/takeover.rs index 7c0c251ab..1ec5748bb 100644 --- a/rust/src/runtime/upgrade/takeover.rs +++ b/rust/src/runtime/upgrade/takeover.rs @@ -221,16 +221,9 @@ fn report(backend: &str, layout: &str, digest: &Value, inventory: &Value, conver /// Published PHP features native does not implement yet. A source holding any /// of their rows is refused, rather than taken over without them. Derived /// projections and PHP's own queue and poll bookkeeping are left in place. -const UNSUPPORTED_FEATURES: [(&str, &[&str]); 1] = [ ( - "upgrade_source_message_streams_unsupported", - &[ - "workflow_messages", - "workflow_durable_streams", - "workflow_durable_stream_items", - "workflow_inbound_streams", - "workflow_inbound_stream_items", - ], - ), +const UNSUPPORTED_FEATURES: [(&str, &[&str]); 1] = [ + // Workflow-to-workflow messages, which native does not deliver yet. + ("upgrade_source_workflow_messages_unsupported", &["workflow_messages"]), ]; /// Takeover needs every inspection result it relies on, and refuses what @@ -294,6 +287,7 @@ fn qualified(inventory: &Value) -> Result<&str> { || total("unqualified_references") != 0 || total("other_unvalidated") != 0 || inventory["table_rows"]["runtime_external_payloads"] != 0 + || inventory["stream_reference_inspection"]["references"] != 0 { return Err(refuse( StatusCode::CONFLICT, @@ -406,6 +400,7 @@ where } else { 0 }; + let reclaims = adopt_stream_reclaims::(tx, super::has_release(layout, "php-2.11.3")).await?; let release_2113 = if super::has_release(layout, "php-2.11.3") { let messages = count::(tx, FULL_FAILURE_MESSAGES).await?; let dispatches = count::(tx, COMPLETED_DISPATCHES).await?; @@ -419,7 +414,8 @@ where "session_holders_recorded":holders, "service_call_deadlines_copied":deadlines, "queue_delivery_references_removed":delivery_references, - "observability_contexts_removed":observability_contexts + "observability_contexts_removed":observability_contexts, + "stream_reclaims_recorded":reclaims }); if let (Value::Object(converted), Value::Object(release)) = (&mut converted, release_2113) { converted.extend(release); @@ -427,6 +423,44 @@ where Ok(converted) } +/// Native keeps each closed or errored durable stream's reclaim deadline in +/// `dw_stream_reclaims`, which its retention pass reads. PHP 2.11.3 records it +/// as `workflow_durable_streams.reclaim_at` (null once the stream's items are +/// gone). Earlier releases recorded none, so the deadline is `closed_at` plus +/// the stream's retention, an hour unless set, as 2.11.3's own migration +/// backfills it. A repeated call records only streams not recorded yet. +/// Returns how many it recorded. +pub(super) async fn adopt_stream_reclaims(tx: &mut DB::Connection, recorded: bool) -> Result +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, + for<'q> String: Encode<'q, DB> + Type, +{ + let rows = query::(if recorded { + "SELECT id,namespace,workflow_run_id,reclaim_at FROM workflow_durable_streams WHERE reclaim_at IS NOT NULL AND id NOT IN (SELECT stream_id FROM dw_stream_reclaims) ORDER BY id" + } else { + "SELECT id,namespace,workflow_run_id,closed_at,retention_seconds FROM workflow_durable_streams WHERE status IN ('closed','errored') AND closed_at IS NOT NULL AND id NOT IN (SELECT stream_id FROM dw_stream_reclaims) ORDER BY id" + }) + .fetch_all(&mut *tx) + .await?; + for row in &rows { + let reclaim_at = if recorded { + DB::instant(row, "reclaim_at")? + } else { + DB::instant(row, "closed_at")? + + chrono::Duration::seconds(DB::optional_number(row, "retention_seconds")?.unwrap_or(3600)) + }; + query::("INSERT INTO dw_stream_reclaims(stream_id,namespace,workflow_run_id,reclaim_at) VALUES ($1,$2,$3,$4)") + .bind(DB::optional_integer(Some(DB::number(row, "id")?))) + .bind(DB::string(row, "namespace")?) + .bind(DB::string(row, "workflow_run_id")?) + .bind(DB::bind_time(reclaim_at)) + .execute(&mut *tx) + .await?; + } + Ok(rows.len()) +} + /// Core 2.14's full failure messages: a message beyond the old TEXT capacity, /// or with a NUL, keeps a labelled preview in `message` and its full bytes in /// `message_full`. Native has no second column; it reads the preview, as diff --git a/rust/src/runtime/upgrade/takeover/mysql_journal.rs b/rust/src/runtime/upgrade/takeover/mysql_journal.rs index fad95eab0..e55b82596 100644 --- a/rust/src/runtime/upgrade/takeover/mysql_journal.rs +++ b/rust/src/runtime/upgrade/takeover/mysql_journal.rs @@ -390,6 +390,17 @@ async fn perform(connection: &mut MySqlConnection, run: &mut Run<'_>, step: i64) } } 10 => { + // Durable stream reclaim deadlines: 2.11.3 recorded them in + // `reclaim_at`; earlier layouts get each closed stream's + // retention. The layout decides, not the column: a resumed + // 2.11.3 takeover whose column is already dropped copied every + // deadline before the drop (each insert commits), so it adopts + // nothing more. + let recorded = super::super::has_release(&run.layout, "php-2.11.3"); + if !recorded || column_exists(connection, "workflow_durable_streams", "reclaim_at").await? { + let reclaims = super::adopt_stream_reclaims::(connection, recorded).await?; + run.converted.insert("stream_reclaims_recorded".into(), json!(reclaims)); + } // PHP 2.11.3's additions. Count before dropping, as step 9 does. if column_exists(connection, "workflow_failures", "message_full").await? { let messages = count::(connection, super::FULL_FAILURE_MESSAGES).await?; diff --git a/rust/src/runtime/upgrade_tests.rs b/rust/src/runtime/upgrade_tests.rs index b9bcdb4d7..db7309eb5 100644 --- a/rust/src/runtime/upgrade_tests.rs +++ b/rust/src/runtime/upgrade_tests.rs @@ -1108,6 +1108,7 @@ async fn takeover_converts_a_php_2113_database(release: &str, changes: &[&[&'sta "INSERT INTO workflow_failures(id,workflow_run_id,source_kind,source_id,propagation_kind,exception_class,message,file,message_full,created_at,updated_at) VALUES ('01m5wvy8d1j3dtr13es05kfail','01jrun0000000000000000000a','activity_execution','a','activity','RuntimeException','preview\n[full message stored separately]','','base64:ZnVsbA==','2026-10-26 10:00:00','2026-10-26 10:00:00')".to_owned(), "INSERT INTO workflow_service_calls(id,namespace,endpoint_name,service_name,operation_name,caller_namespace,target_namespace,status,outcome,operation_mode,idempotency_key,idempotency_scope_hash,accepted_at,completed_at,created_at,updated_at) VALUES ('01m5wvy8d1j3dtr13es05kcall','default','pay','billing','charge','default','default','completed','succeeded','async','order-1','0000000000000000000000000000000000000000000000000000000000000001','2026-10-26 10:00:00','2026-10-26 10:00:01','2026-10-26 10:00:00','2026-10-26 10:00:01')".to_owned(), "INSERT INTO workflow_service_call_dispatches(service_call_id,state,dispatch_options,available_at,completed_at,created_at,updated_at) VALUES ('01m5wvy8d1j3dtr13es05kcall','completed','{}','2026-10-26 10:00:00','2026-10-26 10:00:01','2026-10-26 10:00:00','2026-10-26 10:00:01')".to_owned(), + "INSERT INTO workflow_durable_streams(namespace,workflow_instance_id,workflow_run_id,stream_name,status,retention_seconds,closed_at,reclaim_at) VALUES ('default','i','01jrun0000000000000000000a','progress','closed',600,'2026-10-26 10:00:00','2026-10-26 10:10:00')".to_owned(), ]; for statement in &statements { mutate(&path, Box::leak(statement.clone().into_boxed_str())).await; @@ -1117,8 +1118,8 @@ async fn takeover_converts_a_php_2113_database(release: &str, changes: &[&[&'sta assert_eq!(report["status"], "taken_over", "{report}"); assert_eq!(report["source_layout"], release); assert_eq!( - [&report["converted"]["session_holders_recorded"], &report["converted"]["full_failure_messages_removed"], &report["converted"]["service_call_dispatches_removed"]], - [&json!(2), &json!(1), &json!(1)], + [&report["converted"]["session_holders_recorded"], &report["converted"]["full_failure_messages_removed"], &report["converted"]["service_call_dispatches_removed"], &report["converted"]["stream_reclaims_recorded"]], + [&json!(2), &json!(1), &json!(1), &json!(1)], "{report}" ); let mut connection = SqliteConnection::connect(&format!("sqlite://{path}")).await.unwrap(); @@ -1278,7 +1279,12 @@ async fn takeover_refusals_leave_the_php_database_unchanged() { // Features native lacks: their rows would be left behind unread. ( "INSERT INTO workflow_messages(workflow_instance_id,direction,channel,stream_key,sequence) VALUES ('w','inbound','c','k',1)", - "upgrade_source_message_streams_unsupported", + "upgrade_source_workflow_messages_unsupported", + ), + // A stream item native would carry over but cannot resolve. + ( + "INSERT INTO workflow_durable_stream_items(stream_id,namespace,workflow_run_id,stream_name,\"offset\",payload_reference,emitted_at) VALUES (1,'default','r','s',0,'dw-external-payload:v1:x','2026-10-26 10:00:00')", + "upgrade_source_external_payloads_unsupported", ), // A namespace whose uploads would all be refused: native serves // only the local driver. @@ -1402,6 +1408,42 @@ async fn takeover_keeps_php_service_calls_and_their_deadlines() { runtime.close().await; } +/// Message streams and durable workflow streams carry over: native serves +/// them from PHP's own tables. A closed durable stream gets the reclaim +/// deadline native's retention pass reads, its `closed_at` plus its +/// retention, as PHP 2.11.3's migration backfills it; an open one gets none. +#[tokio::test] +async fn takeover_keeps_php_streams_and_records_their_reclaim_deadlines() { + let (_directory, path) = php_release_layout(&[PHP_280_CHANGES, TRACE_CARRIERS, QUEUE_DELIVERY]).await; + for statement in [ + "INSERT INTO workflow_durable_streams(id,namespace,workflow_instance_id,workflow_run_id,stream_name,status,last_offset,total_items,retention_seconds,closed_at) VALUES (7,'default','i','01jrun0000000000000000000a','progress','closed',0,1,600,'2026-10-26 10:00:00')", + "INSERT INTO workflow_durable_streams(id,namespace,workflow_instance_id,workflow_run_id,stream_name,status) VALUES (8,'default','i','01jrun0000000000000000000a','log','open')", + "INSERT INTO workflow_durable_stream_items(stream_id,namespace,workflow_run_id,stream_name,\"offset\",payload,emitted_at) VALUES (7,'default','01jrun0000000000000000000a','progress',0,'{\"step\":1}','2026-10-26 09:59:00')", + "INSERT INTO workflow_inbound_streams(id,namespace,workflow_instance_id,stream_name,last_position) VALUES (3,'default','i','orders',1)", + "INSERT INTO workflow_inbound_stream_items(stream_id,namespace,workflow_instance_id,stream_name,message_id,position,payload_codec,payload_blob,payload_hash) VALUES (3,'default','i','orders','m1',1,'avro','AA==','h')", + ] { + mutate(&path, statement).await; + } + let inspection = inspect_sqlite(&path).await.unwrap(); + assert_eq!(inspection["stream_reference_inspection"], json!({"references": 0}), "{inspection}"); + let fingerprint = backup_fingerprint(&path).await; + let report = takeover::take_over_sqlite(&path, &fingerprint, &queue_config()).await.unwrap(); + assert_eq!(report["status"], "taken_over", "{report}"); + assert_eq!(report["converted"]["stream_reclaims_recorded"], 1, "{report}"); + let mut connection = SqliteConnection::connect(&format!("sqlite://{path}")).await.unwrap(); + let reclaims: Vec<(i64, String)> = sqlx::query_as("SELECT stream_id,reclaim_at FROM dw_stream_reclaims") + .fetch_all(&mut connection).await.unwrap(); + assert_eq!(reclaims.len(), 1); + assert_eq!(reclaims[0].0, 7); + assert!(reclaims[0].1.starts_with("2026-10-26 10:10:00"), "{reclaims:?}"); + let kept: i64 = sqlx::query_scalar("SELECT (SELECT COUNT(*) FROM workflow_durable_streams)+(SELECT COUNT(*) FROM workflow_durable_stream_items)+(SELECT COUNT(*) FROM workflow_inbound_streams)+(SELECT COUNT(*) FROM workflow_inbound_stream_items)") + .fetch_one(&mut connection).await.unwrap(); + assert_eq!(kept, 5, "every stream row carries over"); + connection.close().await.unwrap(); + let runtime = crate::runtime::Runtime::open(&path, "test-token".into()).await.unwrap(); + runtime.close().await; +} + /// PHP's update validation tasks carry over unchanged: native validates /// updates in the same `workflow_update_validation_tasks` rows, with no state /// of its own. A pending task whose caller was waiting on PHP is then what a diff --git a/rust/tests/execution/message_streams.rs b/rust/tests/execution/message_streams.rs index 254cfb6f8..f1fe2b763 100644 --- a/rust/tests/execution/message_streams.rs +++ b/rust/tests/execution/message_streams.rs @@ -176,3 +176,103 @@ async fn cluster_info_and_workers_see_the_contract() { assert_eq!(capability["supported"], true, "{capability}"); assert_eq!(capability["minimum_worker_protocol_version"], "1.15"); } + +/// Counts rows with a query built for this test. +async fn rows(database: &TestDatabase, sql: String) -> i64 { + task_retry::rows(database, Box::leak(sql.into_boxed_str())).await +} + +// Published PHP's MessageStreamRetentionCleanup. Pruning a continued run +// releases the payloads of the messages it consumed once the next run's +// cursor checkpoint covers them, and keeps their identity rows. Pruning the +// instance's last retained run, once it is terminal, removes its streams. +#[tokio::test] +async fn retention_releases_consumed_messages_and_removes_the_last_runs_streams() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + assert_eq!(request(&app, "POST", "/api/worker/register", stream_worker()).await.0, StatusCode::CREATED); + let started = start(&app, "streams-kept").await; + let first = started["run_id"].as_str().unwrap().to_owned(); + for (id, n) in [("m1", 1), ("m2", 2)] { + assert_eq!(append(&app, "streams-kept", "orders", json!({"message_id":id,"input":[{"n":n}]})).await.0, StatusCode::ACCEPTED); + } + // The first run consumes both messages and continues as new. + let task = poll(&app, "streams", "workflow").await; + let cursor = json!({"message_stream_cursors":[{"stream_name":"orders","through_position":2}]}); + let (status, body) = complete(&app, &task, json!([{"type":"continue_as_new","arguments":envelope(Payload::Array(vec![]))}]), cursor, "1.20").await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(append(&app, "streams-kept", "orders", json!({"message_id":"m3","input":[{"n":3}]})).await.0, StatusCode::ACCEPTED); + let task = poll(&app, "streams", "workflow").await; + let second = task["run_id"].as_str().unwrap().to_owned(); + assert_ne!(second, first); + let cursor = json!({"message_stream_cursors":[{"stream_name":"orders","through_position":3}]}); + let (status, body) = complete(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::String("done".into()))}]), cursor, "1.20").await; + assert_eq!(status, StatusCode::OK, "{body}"); + let expire = |run: &str| format!("UPDATE {{table}} SET closed_at='2020-01-01 00:00:00.000000' WHERE id='{run}'"); + for table in ["workflow_runs", "workflow_run_summaries"] { + database.execute(sqlx::AssertSqlSafe(expire(&first).replace("{table}", table))).await; + } + + // The first run goes; the second run's checkpoint covers what it consumed. + // Put the cursor exactly on the last message the first run consumed: that + // message is covered too (PHP compares with <=). + database.execute(sqlx::AssertSqlSafe("UPDATE workflow_inbound_streams SET cursor_position=2".to_owned())).await; + let (status, pruned) = request(&app, "POST", "/api/system/retention/pass", json!({"run_ids":[first]})).await; + assert_eq!(status, StatusCode::OK, "{pruned}"); + let deleted = &pruned["results"][0]["deleted"]; + assert_eq!(pruned["results"][0]["outcome"], "pruned", "{pruned}"); + assert_eq!( + [&deleted["message_stream_items_compacted"], &deleted["message_stream_items_deleted"], &deleted["message_streams_deleted"]], + [&json!(2), &json!(0), &json!(0)], + "{pruned}" + ); + assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE payload_released_at IS NOT NULL AND payload_blob=''".into()).await, 2); + assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE message_id='m3' AND payload_released_at IS NULL AND payload_blob<>''".into()).await, 1); + database.execute(sqlx::AssertSqlSafe("UPDATE workflow_inbound_streams SET cursor_position=3".to_owned())).await; + let orders = &streams(&app, "streams-kept").await["streams"][0]; + assert_eq!((&orders["last_position"], &orders["cursor_position"]), (&json!(3), &json!(3)), "{orders}"); + + // The last run goes too, and the instance's stream with it. + for table in ["workflow_runs", "workflow_run_summaries"] { + database.execute(sqlx::AssertSqlSafe(expire(&second).replace("{table}", table))).await; + } + let (status, pruned) = request(&app, "POST", "/api/system/retention/pass", json!({"run_ids":[second]})).await; + assert_eq!(status, StatusCode::OK, "{pruned}"); + let deleted = &pruned["results"][0]["deleted"]; + assert_eq!( + [&deleted["message_stream_items_compacted"], &deleted["message_stream_items_deleted"], &deleted["message_streams_deleted"]], + [&json!(0), &json!(3), &json!(1)], + "{pruned}" + ); + assert_eq!(rows(&database, "SELECT (SELECT COUNT(*) FROM workflow_inbound_streams)+(SELECT COUNT(*) FROM workflow_inbound_stream_items)".into()).await, 0); +} + +// Without a retained checkpoint past them, a run's consumed messages keep +// their payloads: replaying the instance's current run still needs them. +#[tokio::test] +async fn retention_keeps_consumed_messages_no_retained_checkpoint_covers() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + assert_eq!(request(&app, "POST", "/api/worker/register", stream_worker()).await.0, StatusCode::CREATED); + let started = start(&app, "streams-open").await; + let first = started["run_id"].as_str().unwrap().to_owned(); + assert_eq!(append(&app, "streams-open", "orders", json!({"message_id":"m1","input":[{"n":1}]})).await.0, StatusCode::ACCEPTED); + let task = poll(&app, "streams", "workflow").await; + let cursor = json!({"message_stream_cursors":[{"stream_name":"orders","through_position":1}]}); + let (status, body) = complete(&app, &task, json!([{"type":"continue_as_new","arguments":envelope(Payload::Array(vec![]))}]), cursor, "1.20").await; + assert_eq!(status, StatusCode::OK, "{body}"); + // The checkpoint names a run that is no longer retained. The instance + // names it as its current run too, but without a run row it retains + // nothing, as in PHP. + database.execute(sqlx::AssertSqlSafe("UPDATE workflow_inbound_streams SET cursor_checkpoint_run_id='gone'".to_owned())).await; + database.execute(sqlx::AssertSqlSafe("UPDATE workflow_instances SET current_run_id='gone'".to_owned())).await; + for table in ["workflow_runs", "workflow_run_summaries"] { + database.execute(sqlx::AssertSqlSafe(format!("UPDATE {table} SET closed_at='2020-01-01 00:00:00.000000' WHERE id='{first}'"))).await; + } + let (status, pruned) = request(&app, "POST", "/api/system/retention/pass", json!({"run_ids":[first]})).await; + assert_eq!(status, StatusCode::OK, "{pruned}"); + assert_eq!(pruned["results"][0]["outcome"], "pruned", "{pruned}"); + assert_eq!(pruned["results"][0]["deleted"]["message_stream_items_compacted"], 0, "{pruned}"); + assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE payload_released_at IS NULL AND payload_blob<>''".into()).await, 1); + assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_streams".into()).await, 1, "the open instance keeps its stream"); +}