Rust #325: serve durable workflow stream writes (items, close)

.github/workflows/server-parity.yml

@@ -66,6 +66,7 @@ on:
       - tests/Unit/ServerParityKilledAuthorPollTest.mjs
       - tests/Unit/ServerParityImagePullTest.mjs
       - tests/Unit/ServerParityMessageStreamsContractTest.mjs
+      - tests/Unit/ServerParityWorkflowStreamWritesContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
       - tests/Unit/ServerParityTaskRetryContractTest.mjs
       - tests/Unit/ServerParityPublishedPatchContractTest.mjs
@@ -1806,7 +1807,7 @@ jobs:
           bash scripts/ci/run-composer.sh --rm -v "$PWD:/source" -w /source -- \
             install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress
           docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \
-            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs > parity-evidence/comparator-tests.txt
+            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs tests/Unit/ServerParityWorkflowStreamWritesContractTest.mjs > parity-evidence/comparator-tests.txt
           docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \
             --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \
             "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt

docs/rust-server-port.md

@@ -82,13 +82,14 @@ Updated October 24, 2026.
     ([#737](https://github.com/durable-workflow/server/issues/737)).
 
   No expectation is edited to make native pass.
-- **Parity surface.** Native serves 146 of the 175 API route methods in
-  published 2.11.1's route table, with message streams, update validation,
-  activity cancellation acknowledgement, worker sessions, workflow-task
-  cancellation scopes, caller-indexed Nexus operation reads, bridge adapter
-  webhooks, and embedded v2 and Waterline v1 workflow import. An execution
-  test sends each one through the native router, and the ones still
-  answering `501 rust_capability_not_implemented` must match
+- **Parity surface.** Native serves 148 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, caller-indexed Nexus
+  operation reads, bridge adapter webhooks, and embedded v2 and Waterline v1
+  workflow import. An execution test sends each one 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
@@ -160,7 +161,6 @@ any other unknown path or method answers PHP's own 404 or 405
     ([checkpoint](rust-server-port-checkpoints/2026-10-23-message-streams.md)),
     except external payload references as input, stream retention cleanup
     and takeover of PHP data with streams.
-  - Durable workflow stream items and close.
   - Service endpoints. (The service catalog reads from Server 2.11 are served:
     native never holds catalog rows, so every list is PHP's empty page and every
     identity its not-found answer; the `service-catalog` fixture checks 37 cases.
@@ -931,12 +931,38 @@ October 21, Waterline reads:
 - **Task queues.** `GET /api/task-queues` and `GET /api/task-queues/{name}`
   follow PHP's `TaskQueueController`: pollers, stats, current leases, repair
   candidates, recent task flow, activity attempt states and admission.
-- **Durable streams.** The run's stream list, stream description and item
-  reads answer as PHP does, including the name pattern, the `from`,
-  `max_items` and `wait_seconds` rules and the refusal bodies. Native runs
-  never produce durable streams, and takeover refuses a database that holds
-  any, so every list is empty. Producer appends and closes are not
-  implemented.
+- **Durable streams.** The run's stream list, stream description, item
+  reads, appends and closes answer as PHP does, with the
+  `workflow-stream-writes` fixture recorded from published 2.11.1: offsets
+  from 0, idempotent retries, bounded waits, final closes, whole-second times
+  and every refusal body. The shared contract fix of Server #774 is native
+  too:
+  - `max_pending_items` is at most 10000 and `retention_seconds` at most
+    2592000, both refused with 422;
+  - a batch whose fresh items would pass the cap is refused whole, so
+    nothing of it is written (published 2.11.1 committed the items before the
+    cap; the fixture bounds that answer as a published defect);
+  - a closed stream past its retention reads as not found, append
+    included, and keeps its row until its run is pruned, so the name is
+    never opened again;
+  - each deadline is a row in the native-owned `dw_stream_reclaims`
+    (schema 6), indexed `(namespace, reclaim_at, stream_id)`. A bounded pass
+    reads due streams from that index, from the worker-heartbeat retention
+    pass and from a general `POST /api/system/retention/pass`. It deletes
+    items in batches under the transition lock and stops blocked when
+    items hold external references, which native does not delete yet;
+  - run retention deletes a pruned run's streams unless they hold external
+    references, and an append to a run whose details are pruned answers
+    404 `stream_not_found`.
+  A continued run keeps its streams, as in PHP. Native differs in three
+  places, documented in `rust/src/runtime/workflow_streams.rs`:
+  - a batch whose items hold no item fields gets 400, not PHP's 500;
+  - a read at the head never clears the count of a later append;
+  - a runtime external payload reference object in `payload_reference`
+    gets 501.
+  Still missing: the workflow-side `workflow_stream` command, so cluster
+  info does not advertise `workflow_streams_contract`, and takeover still
+  refuses a PHP database that holds streams.
 - **Last progress.** Runs record `last_progress_at` when they start, as PHP
   does at start and child start.
 - **Deliberate differences.**

rust/migrations/mariadb/0005_catalog.json → rust/migrations/mariadb/0006_catalog.json

@@ -19,6 +19,10 @@
         "dw_server_schema",
         "CREATE TABLE `dw_server_schema` (\n  `engine` varchar(64) NOT NULL,\n  `version` bigint(20) NOT NULL,\n  `bootstrap_checksum` varbinary(48) NOT NULL,\n  PRIMARY KEY (`engine`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"
     ],
+    [
+        "dw_stream_reclaims",
+        "CREATE TABLE `dw_stream_reclaims` (\n  `stream_id` bigint(20) unsigned NOT NULL,\n  `namespace` varchar(128) NOT NULL,\n  `workflow_run_id` varchar(26) NOT NULL,\n  `reclaim_at` datetime(6) NOT NULL,\n  PRIMARY KEY (`stream_id`),\n  KEY `dw_stream_reclaims_due` (`namespace`,`reclaim_at`,`stream_id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
+    ],
     [
         "dw_task_completions",
         "CREATE TABLE `dw_task_completions` (\n  `task_id` varchar(26) NOT NULL,\n  `receipt` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL CHECK (json_valid(`receipt`)),\n  PRIMARY KEY (`task_id`),\n  CONSTRAINT `dw_task_completions_task` FOREIGN KEY (`task_id`) REFERENCES `workflow_tasks` (`id`) ON DELETE CASCADE\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"

rust/migrations/mariadb/0006_stream_reclaims.sql

@@ -0,0 +1,13 @@
+-- Native development bootstrap version 6: the reclaim deadline of each closed
+-- or errored durable workflow stream (Server #774), closed_at plus its
+-- retention. The stream retention pass reads a namespace's due streams in
+-- deadline order from (namespace, reclaim_at, stream_id). A stream leaves the
+-- table once its items are reclaimed; its row stays until its run is pruned.
+CREATE TABLE dw_stream_reclaims (
+    stream_id BIGINT UNSIGNED NOT NULL PRIMARY KEY,
+    namespace VARCHAR(128) NOT NULL,
+    workflow_run_id VARCHAR(26) NOT NULL,
+    reclaim_at DATETIME(6) NOT NULL,
+    KEY dw_stream_reclaims_due (namespace, reclaim_at, stream_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+UPDATE dw_server_schema SET version=6 WHERE engine='rust-development' AND version=5;

rust/migrations/mysql/0005_catalog.json → rust/migrations/mysql/0006_catalog.json

@@ -23,6 +23,10 @@
         "dw_server_schema",
         "CREATE TABLE `dw_server_schema` (\n  `engine` varchar(64) COLLATE utf8mb4_bin NOT NULL,\n  `version` bigint NOT NULL,\n  `bootstrap_checksum` varbinary(48) NOT NULL,\n  PRIMARY KEY (`engine`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"
     ],
+    [
+        "dw_stream_reclaims",
+        "CREATE TABLE `dw_stream_reclaims` (\n  `stream_id` bigint unsigned NOT NULL,\n  `namespace` varchar(128) COLLATE utf8mb4_unicode_ci NOT NULL,\n  `workflow_run_id` varchar(26) COLLATE utf8mb4_unicode_ci NOT NULL,\n  `reclaim_at` datetime(6) NOT NULL,\n  PRIMARY KEY (`stream_id`),\n  KEY `dw_stream_reclaims_due` (`namespace`,`reclaim_at`,`stream_id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
+    ],
     [
         "dw_task_completions",
         "CREATE TABLE `dw_task_completions` (\n  `task_id` varchar(26) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,\n  `receipt` json NOT NULL,\n  PRIMARY KEY (`task_id`),\n  CONSTRAINT `dw_task_completions_task` FOREIGN KEY (`task_id`) REFERENCES `workflow_tasks` (`id`) ON DELETE CASCADE\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"

rust/migrations/mysql/0006_stream_reclaims.sql

@@ -0,0 +1,13 @@
+-- Native development bootstrap version 6: the reclaim deadline of each closed
+-- or errored durable workflow stream (Server #774), closed_at plus its
+-- retention. The stream retention pass reads a namespace's due streams in
+-- deadline order from (namespace, reclaim_at, stream_id). A stream leaves the
+-- table once its items are reclaimed; its row stays until its run is pruned.
+CREATE TABLE dw_stream_reclaims (
+    stream_id BIGINT UNSIGNED NOT NULL PRIMARY KEY,
+    namespace VARCHAR(128) NOT NULL,
+    workflow_run_id VARCHAR(26) NOT NULL,
+    reclaim_at DATETIME(6) NOT NULL,
+    KEY dw_stream_reclaims_due (namespace, reclaim_at, stream_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+UPDATE dw_server_schema SET version=6 WHERE engine='rust-development' AND version=5;

rust/migrations/postgres/0005_catalog.json → rust/migrations/postgres/0006_catalog.json

@@ -766,6 +766,58 @@
             "generated": ""
         }
     ],
+    [
+        "column",
+        "dw_stream_reclaims.1",
+        {
+            "collation": null,
+            "default": null,
+            "generated": "",
+            "identity": "",
+            "name": "stream_id",
+            "not_null": true,
+            "type": "bigint"
+        }
+    ],
+    [
+        "column",
+        "dw_stream_reclaims.2",
+        {
+            "collation": "default",
+            "default": null,
+            "generated": "",
+            "identity": "",
+            "name": "namespace",
+            "not_null": true,
+            "type": "text"
+        }
+    ],
+    [
+        "column",
+        "dw_stream_reclaims.3",
+        {
+            "collation": "default",
+            "default": null,
+            "generated": "",
+            "identity": "",
+            "name": "workflow_run_id",
+            "not_null": true,
+            "type": "text"
+        }
+    ],
+    [
+        "column",
+        "dw_stream_reclaims.4",
+        {
+            "collation": null,
+            "default": null,
+            "generated": "",
+            "identity": "",
+            "name": "reclaim_at",
+            "not_null": true,
+            "type": "timestamp(6) without time zone"
+        }
+    ],
     [
         "column",
         "dw_task_completions.1",
@@ -11987,6 +12039,16 @@
             "definition": "PRIMARY KEY (key)"
         }
     ],
+    [
+        "constraint",
+        "dw_stream_reclaims.dw_stream_reclaims_pkey",
+        {
+            "deferrable": false,
+            "deferred": false,
+            "definition": "PRIMARY KEY (stream_id)",
+            "validated": true
+        }
+    ],
     [
         "constraint",
         "dw_task_completions.dw_task_completions_pkey",
@@ -13082,6 +13144,28 @@
             "definition": "CREATE UNIQUE INDEX dw_query_cache_pkey ON public.dw_query_cache USING btree (key)"
         }
     ],
+    [
+        "index",
+        "dw_stream_reclaims_due",
+        {
+            "definition": "CREATE INDEX dw_stream_reclaims_due ON public.dw_stream_reclaims USING btree (namespace, reclaim_at, stream_id)",
+            "primary": false,
+            "ready": true,
+            "unique": false,
+            "valid": true
+        }
+    ],
+    [
+        "index",
+        "dw_stream_reclaims_pkey",
+        {
+            "definition": "CREATE UNIQUE INDEX dw_stream_reclaims_pkey ON public.dw_stream_reclaims USING btree (stream_id)",
+            "primary": true,
+            "ready": true,
+            "unique": true,
+            "valid": true
+        }
+    ],
     [
         "index",
         "dw_task_completions_pkey",
@@ -17308,6 +17392,21 @@
             "force_row_security": false
         }
     ],
+    [
+        "relation",
+        "dw_stream_reclaims",
+        {
+            "access_method": "heap",
+            "force_row_security": false,
+            "kind": "r",
+            "options": null,
+            "partition": false,
+            "partition_key": null,
+            "persistence": "p",
+            "row_security": false,
+            "view": null
+        }
+    ],
     [
         "relation",
         "dw_task_completions",

rust/migrations/postgres/0006_stream_reclaims.sql

@@ -0,0 +1,13 @@
+-- Native development bootstrap version 6: the reclaim deadline of each closed
+-- or errored durable workflow stream (Server #774), closed_at plus its
+-- retention. The stream retention pass reads a namespace's due streams in
+-- deadline order from (namespace, reclaim_at, stream_id). A stream leaves the
+-- table once its items are reclaimed; its row stays until its run is pruned.
+CREATE TABLE dw_stream_reclaims (
+    stream_id bigint NOT NULL PRIMARY KEY,
+    namespace text NOT NULL,
+    workflow_run_id text NOT NULL,
+    reclaim_at timestamp(6) without time zone NOT NULL
+);
+CREATE INDEX dw_stream_reclaims_due ON dw_stream_reclaims (namespace, reclaim_at, stream_id);
+UPDATE dw_server_schema SET version=6 WHERE engine='rust-development' AND version=5;