diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b200e8..b298433 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- A task-oriented Message Streams guide and shipped PHP examples now show + client-role appends with stable message identities alongside deterministic + single-message and bounded-batch worker consumption. + ## [2.0.0] - 2026-09-01 ### Added diff --git a/README.md b/README.md index 68a98fe..6777e7a 100644 --- a/README.md +++ b/README.md @@ -492,6 +492,18 @@ task kind, consecutive attempt, selected delay, and typed server exception. Authentication failures, malformed responses, and generic server errors remain fatal. +## Receive repeated input with Message Streams + +Inbound Message Streams deliver repeated, ordered application input to a stable +workflow instance. The application appends a stable message identity through +`WorkflowHandle::appendMessage()`, while workflow code consumes one message or +a bounded ordered batch through `messageStream()`. Runtime-owned cursors survive +replay, worker replacement, server restart, and continue-as-new. + +See the task-oriented [Message Streams guide](https://php.durable-workflow.com/build/message-streams/) +and the shipped [client](examples/message-stream-client.php) and +[worker](examples/message-stream-worker.php) examples. + ## Laravel service mode Laravel 9 through 13 auto-discover the service provider from the same SDK package. diff --git a/docs/portal/_data/navigation.json b/docs/portal/_data/navigation.json index 65a75fc..9b1755e 100644 --- a/docs/portal/_data/navigation.json +++ b/docs/portal/_data/navigation.json @@ -13,6 +13,7 @@ {"label": "Workflows & activities", "url": "/build/workflows-activities/"}, {"label": "Workers", "url": "/build/workers/"}, {"label": "Signals, queries & updates", "url": "/build/messages/"}, + {"label": "Message Streams", "url": "/build/message-streams/"}, {"label": "Failures & retries", "url": "/build/failures-retries/"}, {"label": "Testing", "url": "/build/testing/"} ] diff --git a/docs/portal/build/failures-retries.md b/docs/portal/build/failures-retries.md index 3900a2a..d6b1871 100644 --- a/docs/portal/build/failures-retries.md +++ b/docs/portal/build/failures-retries.md @@ -4,8 +4,8 @@ title: Failures & retries description: Model activity failures, transport failures, workflow terminal states, cancellation, timeouts, and retry safety in the PHP SDK. lead: Durable execution does not make every error retryable. Classify transport delivery, activity attempts, deterministic replay, and terminal workflow outcomes separately. previous: - label: Signals, queries & updates - url: /build/messages/ + label: Message Streams + url: /build/message-streams/ next: label: Testing url: /build/testing/ diff --git a/docs/portal/build/message-streams.md b/docs/portal/build/message-streams.md new file mode 100644 index 0000000..b6d17af --- /dev/null +++ b/docs/portal/build/message-streams.md @@ -0,0 +1,174 @@ +--- +layout: layout.njk +title: Inbound Message Streams +description: Append repeated input to a Durable Workflow PHP service and consume it in deterministic, ordered batches. +lead: Give every logical application event a stable identity, then let workflow history own its ordered consumption cursor across replay and worker replacement. +previous: + label: Signals, queries & updates + url: /build/messages/ +next: + label: Failures & retries + url: /build/failures-retries/ +--- +## Choose repeated input deliberately + +Inbound Message Streams carry application input to a stable workflow instance. +They fit order events, human replies, device readings, and other inputs that can +arrive repeatedly and must be processed in Server-assigned order. + +| Contract | Direction and scope | Delivery model | +| --- | --- | --- | +| Inbound Message Stream | Application to workflow instance | Repeated input with stable message identity, ordered positions, and a durable cursor | +| Workflow Stream | Workflow output associated with one run | Run-scoped output with offset-based subscribers and an explicit lifecycle | +| Signal | Application to workflow | One-shot input recorded in history and read through `signals()` | + +Use a signal for a one-time event when admission is enough. Use a tracked update +when the caller needs an applied result. Use an inbound Message Stream when the +workflow must consume a repeated series deterministically. Do not substitute a +Workflow Stream: its output lifecycle and run scope solve a different problem. + +## Consume one message or a bounded batch + +Start the `orders.message-inbox` workflow with input `[1]` when it should consume +one message through `receiveOne()`. Pass a value from `2` through `20` when it +should consume the currently available ordered batch through `receive()`. A +batch receiver waits until at least one message is available, then returns no +more than its bound; it does not wait for the batch to fill. + +This shipped example connects the worker with only its worker credential: + + +```php + */ + #[Workflow('orders.message-inbox')] + public function run(WorkflowContext $context, int $batchSize = 1): array + { + $stream = $context->messageStream('order-events'); + $messages = $batchSize === 1 + ? [$stream->receiveOne()] + : $stream->receive(maxItems: min(max($batchSize, 2), 20)); + + return array_map( + static fn (MessageStreamMessage $message): array => [ + 'message_id' => $message->messageId, + 'position' => $message->position, + 'event' => $message->arguments[0], + ], + $messages, + ); + } +} + +$client = new Client( + quickstartEnvironment('DURABLE_WORKFLOW_RUNTIME_URL'), + namespace: quickstartEnvironment('DURABLE_WORKFLOW_NAMESPACE'), + workerToken: quickstartEnvironment('DURABLE_WORKFLOW_WORKER_TOKEN'), +); + +Worker::create($client, quickstartEnvironment('DURABLE_WORKFLOW_TASK_QUEUE')) + ->register(OrderMessageInboxWorkflow::class) + ->run(); +``` + +The returned `MessageStreamMessage` exposes its stable `messageId`, ordered +`position`, and decoded positional `arguments`. Keep the stream name and payload +shape backward-compatible while active workflow histories can still reach this +code. + +## Append from an application service + +Start an `orders.message-inbox` instance using the normal +[`startWorkflow()` path](/getting-started/client-setup/#start-once-then-retain-the-handle), +then append from a command, controller, listener, job, or service. Persist a +business-event identity before the call and reuse it after any timeout whose +outcome is unknown. Do not generate a new random ID for a retry. + +The application example connects with only the client credential: + + +```php +workflowHandle('order:'.$orderId) + ->appendMessage( + 'order-events', + $messageId, + [[ + 'event_id' => $eventId, + 'kind' => 'item-added', + 'sku' => quickstartEnvironment('ORDER_SKU'), + 'quantity' => (int) quickstartEnvironment('ORDER_QUANTITY'), + ]], + ); + +echo json_encode($outcome, JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES).PHP_EOL; +``` + +In Laravel or Symfony, inject `WorkflowClientInterface` into the application +service and call the same `workflowHandle()->appendMessage()` path. Container +resolution remains credential-lazy in Laravel; the append is the first real +application-client operation and therefore requires the client role. Keep the +worker role credential out of application processes. + +## Reuse identities safely + +The first accepted append receives the next ordered stream position. Repeating +the same message ID and identical payload returns the original position with a +duplicate outcome and does not deliver another item. Reusing that ID with a +different payload is rejected as `message_identity_conflict`; choose whether to +repair the upstream event record or submit a genuinely new event with a new +stable identity. + +The SDK records cursor advancement in workflow-task completion. Replay resumes +from the recorded position, and a replacement worker reconstructs the same +state from history instead of relying on process memory. Application code uses +the public append and receive APIs; the SDK and runtime own their internal +transport and wait mechanics. + +## Keep the instance across run transitions + +Inbound Message Streams are instance-scoped. Continue-as-new transfers the +consumed cursor and pending input to the successor run, so the application keeps +addressing the same workflow ID and cannot accidentally re-consume an +acknowledged position. Workflow Streams remain run-scoped output and therefore +do not replace this inbound handoff contract. + +For managed Cloud, use the complete provisioned runtime URL and its separate +client and worker credentials; Cloud operates the runtime storage and service. +For self-hosted Server, your team also owns Server deployment, database +durability, authentication, backups, and upgrades. In both cases the PHP +application owns stable message IDs and payload compatibility, the worker owns +deterministic consumption code, and the runtime owns ordering, waits, and cursor +durability. See [deployment ownership](/operate/deployment/) for the complete +process split. diff --git a/docs/portal/build/messages.md b/docs/portal/build/messages.md index e2b7f6c..058359a 100644 --- a/docs/portal/build/messages.md +++ b/docs/portal/build/messages.md @@ -7,8 +7,8 @@ previous: label: Workers url: /build/workers/ next: - label: Failures & retries - url: /build/failures-retries/ + label: Message Streams + url: /build/message-streams/ --- ## Choose the message contract @@ -46,6 +46,10 @@ $client->workflowHandle('order-1001')->signal('approve', ['Ada']); Signals are appropriate when the caller only needs acknowledgement that the runtime accepted the event. +Use [inbound Message Streams](/build/message-streams/) instead when one workflow +instance needs repeated ordered input with stable message identities and a +replay-safe consumption cursor. + ## Register a read-only query ```php diff --git a/docs/portal/build/workflows-activities.md b/docs/portal/build/workflows-activities.md index 7081529..e30b1a6 100644 --- a/docs/portal/build/workflows-activities.md +++ b/docs/portal/build/workflows-activities.md @@ -14,6 +14,10 @@ next: A workflow handler runs as ordinary straight-line PHP inside an isolated Fiber. Calls on `WorkflowContext` pause internally when durable work is pending; on replay, the SDK walks committed history and returns recorded values directly from those calls. +For repeated ordered application input, open an +[inbound Message Stream](/build/message-streams/) from `WorkflowContext`. Its +durable cursor resumes after replay, worker replacement, and continue-as-new. + Use `WorkflowContext::waitCondition()` when progress depends on workflow state rather than an external activity. Its deterministic predicate is re-evaluated when committed signals or updates produce another workflow task. A stable key identifies the wait across replay, and the optional timeout returns `false` instead of requiring an application timer loop. ```php diff --git a/examples/message-stream-client.php b/examples/message-stream-client.php new file mode 100644 index 0000000..5d6ba1d --- /dev/null +++ b/examples/message-stream-client.php @@ -0,0 +1,32 @@ +workflowHandle('order:'.$orderId) + ->appendMessage( + 'order-events', + $messageId, + [[ + 'event_id' => $eventId, + 'kind' => 'item-added', + 'sku' => quickstartEnvironment('ORDER_SKU'), + 'quantity' => (int) quickstartEnvironment('ORDER_QUANTITY'), + ]], + ); + +echo json_encode($outcome, JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES).PHP_EOL; diff --git a/examples/message-stream-worker.php b/examples/message-stream-worker.php new file mode 100644 index 0000000..8d3a2c4 --- /dev/null +++ b/examples/message-stream-worker.php @@ -0,0 +1,43 @@ + */ + #[Workflow('orders.message-inbox')] + public function run(WorkflowContext $context, int $batchSize = 1): array + { + $stream = $context->messageStream('order-events'); + $messages = $batchSize === 1 + ? [$stream->receiveOne()] + : $stream->receive(maxItems: min(max($batchSize, 2), 20)); + + return array_map( + static fn (MessageStreamMessage $message): array => [ + 'message_id' => $message->messageId, + 'position' => $message->position, + 'event' => $message->arguments[0], + ], + $messages, + ); + } +} + +$client = new Client( + quickstartEnvironment('DURABLE_WORKFLOW_RUNTIME_URL'), + namespace: quickstartEnvironment('DURABLE_WORKFLOW_NAMESPACE'), + workerToken: quickstartEnvironment('DURABLE_WORKFLOW_WORKER_TOKEN'), +); + +Worker::create($client, quickstartEnvironment('DURABLE_WORKFLOW_TASK_QUEUE')) + ->register(OrderMessageInboxWorkflow::class) + ->run(); diff --git a/scripts/check-docs-examples-contract.json b/scripts/check-docs-examples-contract.json index 9f59a2a..e449ee9 100644 --- a/scripts/check-docs-examples-contract.json +++ b/scripts/check-docs-examples-contract.json @@ -29,6 +29,27 @@ } ] } + }, + { + "id": "php.message-stream.worker.portal", + "path": "docs/portal/build/message-streams.md", + "language": "php", + "source": "examples/message-stream-worker.php", + "publicApiMethods": [ + {"declaration": "src/Worker/WorkflowContext.php", "method": "messageStream"}, + {"declaration": "src/Worker/MessageStream.php", "method": "receive"}, + {"declaration": "src/Worker/MessageStream.php", "method": "receiveOne"} + ] + }, + { + "id": "php.message-stream.client.portal", + "path": "docs/portal/build/message-streams.md", + "language": "php", + "source": "examples/message-stream-client.php", + "publicApiMethods": [ + {"declaration": "src/WorkflowClientInterface.php", "method": "workflowHandle"}, + {"declaration": "src/WorkflowHandleInterface.php", "method": "appendMessage"} + ] } ] } diff --git a/scripts/check-docs-examples.mjs b/scripts/check-docs-examples.mjs index 3d6538b..24f3c68 100644 --- a/scripts/check-docs-examples.mjs +++ b/scripts/check-docs-examples.mjs @@ -34,6 +34,14 @@ function examplePattern(id) { ); } +function publicMethodPattern(method) { + return new RegExp(`\\bpublic\\s+function\\s+${escapeRegExp(method)}\\s*\\(`); +} + +function methodCallPattern(method) { + return new RegExp(`->\\s*${escapeRegExp(method)}\\s*\\(`); +} + function patternOccurrences(block, source, context) { try { return [...block.matchAll(new RegExp(source, 'g'))].length; @@ -218,6 +226,17 @@ for (const example of contract.examples || []) { if (`${match[2]}\n` !== executableSource) { throw new Error(`${context} must render the shipped executable ${example.source} without drift`); } + for (const apiMethod of example.publicApiMethods || []) { + const declaration = await readFile(new URL(apiMethod.declaration, repoRoot), 'utf8'); + assert( + publicMethodPattern(apiMethod.method).test(declaration), + `${context} references missing public API ${apiMethod.declaration}::${apiMethod.method}()`, + ); + assert( + methodCallPattern(apiMethod.method).test(executableSource), + `${context} does not exercise declared public API method ${apiMethod.method}()`, + ); + } if (example.workflowIdentity) { checkWorkflowIdentity(match[2], example.workflowIdentity, context); }