Make message delivery push-only#180
Merged
Merged
Conversation
stidsborg
force-pushed
the
restart-not-running-flows-on-push
branch
from
June 28, 2026 10:32
bde3271 to
e18b535
Compare
RestartExecutions claimed via UPDATE..WHERE owner IS NULL and then SELECTed all requested rows, filtering by owner = replica. A flow claimed by an earlier call from the same replica was therefore returned to a concurrent caller as if it had just been claimed, letting two claimers (e.g. two watchdogs) restart the same flow concurrently - surfacing as UnexpectedStateException: concurrent modification. Run a locking SELECT..FOR UPDATE and the claiming UPDATE inside one transaction so the caller receives exactly the rows its own call claimed, mirroring the already-precise UPDATE..RETURNING (PostgreSQL) and UPDATE..OUTPUT (SqlServer) implementations.
The batch RestartExecutions/RestartExecutionsWithoutMessages claimed any ownerless flow, so a caller targeting a flow that had meanwhile completed would resurrect it - e.g. a message arriving after its target succeeded, once message-driven restarts land. Guard the claim with status IN (Postponed, Suspended) in all stores. The singular RestartExecution (manual restart) intentionally keeps claiming completed flows.
PrepareForReInvocation received the messages loaded by the restart claim but discarded them, so a restarted flow had to re-fetch its own messages from the store. Push them straight into the queue manager's delivery pipeline instead. Push now initializes the queue manager first so the idempotency-key state is loaded before processing. Initialize additionally seeds the fetched-positions set from the persisted delivered positions: the in-hand messages are loaded at claim time - before Initialize clears the delivered positions from the store - so without the seed a message the previous incarnation already delivered (but had not yet cleared) would be delivered again.
Extract the MessageWatchdog's fetch-and-push cycle into PushOnce and invoke it when a queue manager initializes through CreateQueueClient, so a freshly started/resumed flow receives its pending messages immediately instead of waiting for the next watchdog poll. The pending-push is skipped when initialization is triggered by an incoming Push: the pusher already holds an in-flight batch and a nested fetch would stage newer messages ahead of it. Staged messages are additionally kept position-sorted so delivery order stays append order even when concurrent pushers stage batches out of order.
GetMessagesForReplica only returns messages assigned to the fetching replica, so messages published by a crashed replica to ownerless flows were never fetched again once the replica died. The ReplicaWatchdog now reassigns a crashed replica's message rows to itself alongside rescheduling its crashed functions - before deleting the crashed replica from the replica store, so an interrupted handover is retried on a later iteration.
Three tests constructed their own QueueManager inside the flow, which exercised the store-pull directly and duplicated runtime wiring: - QueueManagerFailsOnMessageDeserializationError now registers the exception-throwing serializer on the registry and uses workflow.Message - RegisteredTimeoutIsRemovedWhenPullingMessage now asserts through the observable postpone: a delivered message's timeout must not linger and shorten a subsequent delay's postpone - PullEnvelopeReturnsEnvelopeWithReceiverAndSender uses the flow's own queue manager via a new internal Workflow.QueueManager accessor The tests now stay valid when the queue manager's store-pull is removed and message delivery becomes push-only.
A push whose target flow was suspended, not yet queue-manager-attached, or absent from the manager's dictionary was silently dropped while its positions stayed in the clearer's ignore-set. Later fetches then only returned higher positions, so a later-positioned duplicate consumed the idempotency key before the original message - observed on CI as MultipleIterationsWithDuplicateIdempotencyKeysProcessCorrectly delivering the second-copy values. The queue manager now attaches to its flow state at construction, and undeliverable pushes reopen their positions so they are re-fetched instead of stranded.
The invocation body runs on a Task that captures the scheduler's execution context. When a flow is restarted from a watchdog chain that has itself touched the ambient EffectContext, all invocations spawned from that chain inherit - and mutate - the same context instance, so implicit effect ids shift between incarnations and replay diverges: a restarted flow re-subscribes under a different effect id, filters on the wrong message and suspends, restarting endlessly. Give every invocation a fresh context at start so implicit ids are deterministic regardless of the scheduling context.
MessagesPullFrequency (default 250ms) lost its consumer when the per-flow fetch loop was removed. The MessageWatchdog is the message- delivery loop, so it should run at that frequency - the slower watchdog check frequency made every push-restarted message exchange poll-bound.
Messages pushed to a flow that is not live are now delivered by claiming and restarting the flow (RestartExecutionsWithoutMessages) with the pushed messages handed to the new invocation - the message itself becomes the wake-up signal instead of relying on the interrupt flag. A flow-state entry whose flow has suspended counts as not live: a suspended flow's parked invocation never reaches RemoveFlow by design, and treating the lingering entry as live swallowed the wake-up. Claim failures distinguish retryable targets (reopened for the next poll) from completed/deleted flows, whose positions stay ignored so their messages are not refetched forever. A failed claim call reopens everything. Positions that reach a dying queue manager are reopened rather than dropped: pushes hitting a disposed instance, staged-but-undelivered messages at dispose, and messages for unregistered flow types. A stranded position otherwise loses the flow's wake-up once the message watchdog has marked it as pushed. The tests' thread pool is pre-sized since every test's registry now runs its message poll at 250ms, which starves the default pool growth on small CI runners.
Delete FetchAndNotify and the queue manager's message-store dependency. Messages now arrive exclusively via Push - the MessageWatchdog poll, the immediate fetch at initialization and the restart hand-over - so Subscribe, Interrupt and FetchMessagesOnce only re-evaluate the already staged messages against the current subscriptions.
Message appends no longer set the interrupted flag: the message row itself is the durable wake-up signal - the MessageWatchdog fetches it and either pushes to the live flow or claims and restarts the parked one with the message in hand. The public Interrupt API and the suspend-time interrupted-guard remain. A claim failure for a flow that does not exist (yet) now reopens the positions instead of stranding them: messages may legally precede their flow's creation, and the delivery must be retried once the flow is scheduled.
Message delivery to live flows is handled by the MessageWatchdog's push, and parked flows are woken by push-restarts - the interrupt polling loop no longer had a purpose. Remove it together with its supporting plumbing: FlowsManagers/FlowsManager Interrupt and FilterOwned, FlowExecutionState.Interrupt, QueueManager.Interrupt and the GetInterruptedFunctions/ResetInterrupted store operations. The public Interrupt API and the suspend-time interrupted-guard remain.
Message-driven flow wake-up is push-restart based; the interrupt-only scheduling helper has no callers left.
The tests rely on workflow.Delay suspending the flow so the captured work is re-executed on restart. A 10ms delay can fully elapse before the suspension check on a loaded machine (an effect-persist roundtrip sits in between), letting the capture complete inline on the first attempt. Use 500ms so the suspension reliably happens.
The init-time PushOnce ran on the initializing flow's own async branch, which made restart chains executable from inside another flow's execution context - the trigger for the EffectContext id-shift bug - and coupled queue manager construction to the MessageWatchdog. Message delivery is now driven solely by the watchdog poll and the restart in-hand hand-over, simplifying the wiring (no push delegate threaded through FunctionsRegistry/InvocationHelper/QueueManager). The trade-off is latency: a flow awaiting an already-present message suspends once and is restarted by the next poll instead of receiving it inline. Tests encoding the stronger inline guarantee are adjusted: initial-state flows are awaited tolerantly via Schedule+Completion, AwaitMessageAfterAppendShouldNotCauseSuspension is removed, and WorkflowMessageIsIdempotentAcrossRestarts counts deliveries instead of delivery-timing-dependent invocations.
Add MessageWatchdog.Notify: appending a message completes a wake signal that cuts the watchdog's inter-poll sleep short, so locally appended messages are delivered immediately instead of waiting out the poll interval. The signal is re-armed before each fetch, so a notify arriving mid-push is never lost - it simply makes the next wait return at once. Notify only completes the signal: the fetch-and-push always executes on the watchdog's own loop, so - unlike the removed initialization-time push - no flow execution context is ever captured by another flow's restart. Wired into MessageWriter/MessageWriters appends and the child-to-parent completion message; reopened positions deliberately do not notify, as a persistently unclaimable flow would otherwise spin the poll loop. The faster delivery exposed a stranding hazard: a push whose message fails deserialization poisons the queue manager without staging the positions, and when that races the flow's suspension the message stayed marked forever and the flow never failed. A poisoned queue manager now reopens the batch's unstaged positions so a restarted incarnation receives the message and surfaces the failure.
stidsborg
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Makes message delivery fully push-based and moves flow wake-up into the MessageWatchdog's responsibility.
Architecture
QueueManagerno longer reads from the message store: messages arrive exclusively via push — the MessageWatchdog poll (running atMessagesPullFrequency) and the restart in-hand hand-over.FlowsManager.Pushdelivers to live flows and claims + restarts parked ones with the pushed messages in hand — the message row itself is the durable wake-up signal.AppendMessagesno longer interrupts target flows; theInterruptedWatchdogis removed (the publicInterruptAPI and the suspend-time guard remain).MessageWatchdog.Notify: appends complete a wake signal that cuts the inter-poll sleep short, so locally appended messages are delivered immediately. Notify only completes a signal — the fetch-and-push always runs on the watchdog's own loop.Root-cause fixes along the way
UPDATE..WHERE owner IS NULLfollowed bySELECTreturned rows claimed by an earlier same-replica call as newly claimed, letting two watchdogs restart the same flow concurrently (the CI-onlyconcurrent modificationflakiness). Now a lockingSELECT..FOR UPDATE+UPDATEin one transaction, mirroring PostgreSQL'sRETURNING/SqlServer'sOUTPUTprecision. Regression-tested against all stores.Postponed/Suspendedonly) so late messages cannot resurrect completed flows.EffectContextis reset at the start of every invocation: the invocationTask.Runcaptures the scheduler's execution context, and inheriting a mutable ambient context from the spawning chain shifted implicit effect ids across incarnations, breaking replay determinism.Every commit was verified green on all four CI store jobs before the next.
🤖 Generated with Claude Code