Restart flows via empty-message pokes; restarts no longer pull messages#184
Merged
Conversation
…ole message-fetch path RestartExecution returns effects only; the with-messages RestartExecutions, StoredFlowWithEffectsAndMessages and the RestartFunction delegate are removed. Completed flows' undeliverable positions are parked per flow (park-then-check ordering to avoid stranding them when a concurrent explicit restart reopens parked positions) and released again on explicit re-invocation. ExistingMessages stamps appended messages with the publisher replica so the watchdog routes them; QueueManager reopens delivered-but-not-deleted positions on unwind.
… parking positions A message fetched for a completed flow is moved into the flow's reserved pending-messages effect entry (serializer-independent BinaryPacker encoding) and its row deleted - any later re-invocation, on any replica and via any restart path, receives it from the effect snapshot the restart hands over. The QueueManager stages the entry at initialization (deduped by position and idempotency key) and prunes it atomically with each delivery's capture. The effect write demands the flow is unowned (owner IS NULL) so it cannot interleave with a claim; rows are deleted only while the flow is observed still completed afterwards, otherwise the positions are reopened and normal delivery takes over. The inline uses the store's current rows, not the fetched copies, so concurrent control-panel Replace/Clear/Remove edits win. PostgreSQL now stores effects in a column on the flows table (fillfactor 80) matching MariaDB/SqlServer, gaining the owner-guarded effect write it lacked and making claim + effect snapshot a single atomic UPDATE..RETURNING; the separate effects table, its unused version machinery and the orphaned ReadMessagesForMultipleStores reader are removed. The in-memory store gains the unowned-flow guard. ExistingMessages surfaces and edits the union of both message carriers. The per-replica parked-positions state is deleted.
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.
Summary
StoredMessage(zero-length content and type) acts as a pure restart-poke: it restarts a not-live flow, is never delivered, and is deleted once the restart it forced has happened. Live flows reopen the poke's position instead of deleting it so the restart guarantee survives a concurrent suspend.IFunctionStore.RestartExecutionreturnsStoredFlowWithEffects(no messages); the with-messages batchRestartExecutions,StoredFlowWithEffectsAndMessagesand theRestartFunctiondelegate are removed across all stores.MessageClearer(kept in the ignore-set, so no re-fetch churn) and released again when the flow is explicitly re-invoked.FlowsManagerparks before re-reading the flow status (park-then-check): paired with the restart's claim-then-reopen ordering, whichever side acts second sees the other's write - closing a race that otherwise strands positions in the ignore-set forever.ExistingMessages.Append/Replacestamp messages with the publisher's replica id (wasReplicaId.Empty) so the watchdog routes them.QueueManagerunwind also reopens delivered-but-not-yet-deleted positions, since restarts no longer hand messages over.Testing
🤖 Generated with Claude Code