Make workflow-code message sends and receives act once across replays

CHANGELOG.md

@@ -2,6 +2,41 @@
 
 ## Unreleased
 
+- Message stream calls in workflow code act once instead of on every replay
+  (#1010). Workflow code runs again on each later workflow task, on every
+  query and in a redrive successor's reused prefix.
+  - `outbox()->sendReference()` used to write a new outbound and inbound
+    message each time. A replay now returns the row the original send wrote,
+    including a row written by the run a successor was redriven from.
+  - `inbox()->receive()` and `receiveOne()` now return the messages that call
+    consumed at that step, by this run or by the run it was redriven from.
+    Before, a replay could return different messages or none, and could
+    consume a newer message under the earlier step.
+  - A query, a replay check or a redrive definition check never writes or
+    consumes a message. A stream kept across a replayed update handler checks
+    this on every call, so later sends still write.
+  - Rows that earlier releases duplicated are not removed. Reads don't
+    deduplicate them, and no migration can tell a replay duplicate from an
+    intended identical send. A consumer that read the stream on an affected
+    release should deduplicate by `idempotency_key`. A run with such rows
+    matches its sends to the earliest rows and writes only new sends.
+  - Messages that a replay or query on an affected release (2.20.0 and
+    earlier) consumed under the wrong step stay consumed, and workflow code
+    won't receive them again. Check streams that workflow code receives from
+    for consumed messages the workflow never handled.
+  - A replayed receive whose original call found no message still reads the
+    stream again, because nothing records an empty receive. It can return a
+    newer message, or a message that a later call consumed at the same step,
+    for example through a stream kept in a property, which keeps the step it
+    was opened at. Receive only after a signal or condition says a message is
+    there, and send that signal before the message: a signal moves the run's
+    shared message cursor past older pending messages (#1011).
+  - A send is matched by its exact content (channel, stream, target,
+    reference, correlation id and idempotency key), whatever the column
+    collation. An identical row that code outside
+    the workflow wrote for the same run can satisfy it, so give workflow sends
+    their own idempotency key.
+
 ## 2.20.0 - 2026-11-03
 
 - Rollback, embedded Workflow package:

docs/architecture/execution-guarantees.md

@@ -250,10 +250,16 @@ own identifier and its own durable row.
   cursor, and each `MessageCursorAdvanced` event names exactly one
   `stream_key`. Cursor advance is monotonic; a consumed message cannot
   be un-consumed.
-- External senders that may retry the same logical message should
-  populate the optional `idempotencyKey` on `sendReference()` so the
-  durable message stream can recognise and drop duplicates at the
-  ingress layer.
+- `idempotencyKey` on `sendReference()` is stored on both rows. Ingress
+  does not drop duplicates, so a consumer that can see a retried send
+  should deduplicate by the key.
+- In workflow code, `outbox()->sendReference()` and
+  `inbox()->receive()` / `receiveOne()` run again on every replay. A
+  send returns the row this run (or the run it was redriven from)
+  already wrote for the same content, and a receive returns the
+  messages already consumed at that step. A query or other read-only
+  replay never writes or consumes. A replayed receive whose original
+  call found nothing reads the stream again.
 
 ## Side effects and version markers
 

src/V2/MessageStream.php

@@ -4,12 +4,16 @@
 
 namespace Workflow\V2;
 
+use Closure;
 use DateTimeInterface;
 use Illuminate\Database\Eloquent\Collection;
 use InvalidArgumentException;
 use Workflow\V2\Enums\MessageChannel;
+use Workflow\V2\Enums\MessageConsumeState;
+use Workflow\V2\Enums\MessageDirection;
 use Workflow\V2\Models\WorkflowMessage;
 use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Support\MessageReplayLedger;
 use Workflow\V2\Support\MessageService;
 use Workflow\V2\Support\MessageStreamCursor;
 
@@ -25,6 +29,8 @@
         private readonly string $streamKey,
         private readonly MessageService $messages = new MessageService(),
         private readonly ?int $defaultConsumedBySequence = null,
+        private readonly ?MessageReplayLedger $replayLedger = null,
+        private readonly bool|Closure $readOnly = false,
     ) {
         if ($streamKey === '') {
             throw new InvalidArgumentException('Message stream key must not be empty.');
@@ -45,6 +51,32 @@
         );
     }
 
+    /**
+     * The stream workflow code uses. Workflow code runs again on every
+     * replay and query, so a send or receive it already made returns the row
+     * it wrote or consumed instead of acting again, and a read-only replay
+     * (a query or a replay check) never writes.
+     *
+     * @internal
+     */
+    public static function forWorkflowCode(
+        WorkflowRun $run,
+        ?string $streamKey,
+        ?MessageService $messages,
+        int $defaultConsumedBySequence,
+        MessageReplayLedger $replayLedger,
+        Closure $readOnly,
+    ): self {
+        return new self(
+            $run,
+            $streamKey ?? MessageStreamCursor::defaultStreamKey($run),
+            $messages ?? new MessageService(),
+            $defaultConsumedBySequence,
+            $replayLedger,
+            $readOnly,
+        );
+    }
+
     public function key(): string
     {
         return $this->streamKey;
@@ -80,8 +112,29 @@
      *
      * @return Collection<int, WorkflowMessage>
      */
     public function receive(int $limit = 1, ?int $consumedBySequence = null): Collection
     {
+        $limit = $this->positiveLimit($limit);
+
+        if ($this->replayLedger instanceof MessageReplayLedger) {
+            $sequence = $this->consumedBySequence($consumedBySequence);
+            $recorded = $this->recordedReceive($sequence, $limit);
+
+            // A receive this step already made returns what it consumed.
+            if ($recorded->isNotEmpty() || $this->readOnly()) {
+                return $recorded;
+            }
+
+            $messages = $this->peek($limit);
+
+            if ($messages->isNotEmpty()) {
+                $this->messages->consumeMessageBatch($this->run, $messages->all(), $sequence);
+                $this->replayLedger->advanceReceive($this->streamKey, $sequence, $messages->count());
+            }
+
+            return $messages;
+        }
+
         $messages = $this->peek($limit);
 
         if ($messages->isEmpty()) {
@@ -123,7 +176,60 @@
         array $metadata = [],
         ?DateTimeInterface $expiresAt = null,
     ): WorkflowMessage {
-        return $this->messages->sendMessage(
+        $identity = null;
+
+        if ($this->replayLedger instanceof MessageReplayLedger) {
+            $channelValue = is_string($channel) ? $channel : $channel->value;
+            $identity = json_encode(
+                [
+                    $channelValue,
+                    $this->streamKey,
+                    $targetInstanceId,
+                    $payloadReference,
+                    $correlationId,
+                    $idempotencyKey,
+                ],
+                JSON_THROW_ON_ERROR,
+            );
+            $sent = $this->recordedSend(
+                $identity,
+                $channelValue,
+                $targetInstanceId,
+                $payloadReference,
+                $correlationId,
+                $idempotencyKey
+            );
+
+            // A send this run, or a run it was redriven from, already made
+            // returns the row it wrote.
+            if ($sent instanceof WorkflowMessage) {
+                $this->replayLedger->advanceSend($identity);
+
+                return $sent;
+            }
+
+            // A read-only replay describes the send without writing it.
+            if ($this->readOnly()) {
+                return new WorkflowMessage([
+                    'workflow_instance_id' => $this->run->workflow_instance_id,
+                    'workflow_run_id' => $this->run->id,
+                    'direction' => MessageDirection::Outbound,
+                    'channel' => $channelValue,
+                    'stream_key' => $this->streamKey,
+                    'source_workflow_instance_id' => $this->run->workflow_instance_id,
+                    'source_workflow_run_id' => $this->run->id,
+                    'target_workflow_instance_id' => $targetInstanceId,
+                    'payload_reference' => $payloadReference,
+                    'correlation_id' => $correlationId,
+                    'idempotency_key' => $idempotencyKey,
+                    'metadata' => $metadata,
+                    'expires_at' => $expiresAt,
+                    'consume_state' => MessageConsumeState::Pending,
+                ]);
+            }
+        }
+
+        $message = $this->messages->sendMessage(
             $this->run,
             $channel,
             $targetInstanceId,
@@ -134,6 +240,122 @@
             $metadata,
             $expiresAt,
         );
+        if ($identity !== null) {
+            $this->replayLedger?->advanceSend($identity);
+        }
+
+        return $message;
+    }
+
+    /**
+     * The next outbound row for the same send, in sequence order, that this
+     * run or a run it was redriven from wrote and this execution has not
+     * matched yet.
+     */
+    private function recordedSend(
+        string $identity,
+        string $channel,
+        string $targetInstanceId,
+        ?string $payloadReference,
+        ?string $correlationId,
+        ?string $idempotencyKey,
+    ): ?WorkflowMessage {
+        $query = WorkflowMessage::query()
+            ->where('workflow_instance_id', $this->run->workflow_instance_id)
+            ->whereIn('source_workflow_run_id', $this->runIds())
+            ->where('direction', MessageDirection::Outbound->value)
+            ->where('channel', $channel)
+            ->where('stream_key', $this->streamKey)
+            ->where('target_workflow_instance_id', $targetInstanceId);
+
+        foreach ([
+            'payload_reference' => $payloadReference,
+            'correlation_id' => $correlationId,
+            'idempotency_key' => $idempotencyKey,
+        ] as $column => $value) {
+            $value === null ? $query->whereNull($column) : $query->where($column, $value);
+        }
+
+        // The query only narrows the candidates: a case- or accent-insensitive
+        // collation (MySQL's default) treats different text as equal, so the
+        // match itself compares each field exactly.
+        $exact = [
+            'channel' => $channel,
+            'stream_key' => $this->streamKey,
+            'target_workflow_instance_id' => $targetInstanceId,
+            'payload_reference' => $payloadReference,
+            'correlation_id' => $correlationId,
+            'idempotency_key' => $idempotencyKey,
+        ];
+        $matches = $query->orderBy('sequence')
+            ->orderBy('id')
+            ->get()
+            ->filter(static function (WorkflowMessage $message) use ($exact): bool {
+                foreach ($exact as $column => $value) {
+                    $stored = $message->getRawOriginal($column);
+
+                    if (($stored === null ? null : (string) $stored) !== $value) {
+                        return false;
+                    }
+                }
+
+                return true;
+            })
+            ->values();
+
+        /** @var WorkflowMessage|null $message */
+        $message = $matches->get($this->replayLedger?->sendOffset($identity) ?? 0);
+
+        return $message;
+    }
+
+    /**
+     * The messages this run, or a run it was redriven from, consumed at this
+     * step for this stream that this execution has not returned yet.
+     *
+     * @return Collection<int, WorkflowMessage>
+     */
+    private function recordedReceive(int $consumedBySequence, int $limit): Collection
+    {
+        $offset = $this->replayLedger?->receiveOffset($this->streamKey, $consumedBySequence) ?? 0;
+
+        // As with sends, the stream key is compared exactly, not by collation.
+        /** @var Collection<int, WorkflowMessage> $messages */
+        $messages = WorkflowMessage::query()
+            ->whereIn('workflow_run_id', $this->runIds())
+            ->where('stream_key', $this->streamKey)
+            ->where('direction', MessageDirection::Inbound->value)
+            ->where('consume_state', MessageConsumeState::Consumed->value)
+            ->where('consumed_by_sequence', $consumedBySequence)
+            ->orderBy('sequence')
+            ->orderBy('id')
+            ->get()
+            ->filter(fn (WorkflowMessage $message): bool => $message->getRawOriginal('stream_key') === $this->streamKey)
+            ->slice($offset, $limit)
+            ->values();
+
+        if ($messages->isNotEmpty()) {
+            $this->replayLedger?->advanceReceive($this->streamKey, $consumedBySequence, $messages->count());
+        }
+
+        return $messages;
+    }
+
+    /**
+     * Checked on every call: a stream kept across a replayed update handler,
+     * where dispatch is off, must still write once dispatch is back on.
+     */
+    private function readOnly(): bool
+    {
+        return $this->readOnly instanceof Closure ? (bool) ($this->readOnly)() : $this->readOnly;
+    }
+
+    /**
+     * @return list<string>
+     */
+    private function runIds(): array
+    {
+        return $this->replayLedger?->runIds($this->run) ?? [$this->run->id];
     }
 
     private function consumedBySequence(?int $override): int