Repository navigation
Conversation
A buffered local queue has no ListeningAgent, so once its block faulted terminally (JasperFx#506) every later message was rejected until the process restarted. On the "agents" queue this left a node heartbeating while it could not run any agent command. The queue now replaces a faulted block on the next enqueue. Claude-Session: https://claude.ai/code/session_0176D1FySkqxczVGAYmFjNK7
There was a problem hiding this comment.
🟡 Changes recommended
Block replacement races can lose buffered messages, leave handlers running after shutdown, and violate per-group execution guarantees.
4 open issues
What changed in this PR
Addresses #5358 by allowing buffered local queues, including the agents queue, to rebuild a terminally faulted receiving block instead of rejecting messages until restart.
Changes:
- Extracts receiving-block construction and adds opt-in recovery on the next enqueue.
- Enables recovery for buffered local queues while retaining listener-owned recovery behavior.
- Adds regression tests for message handling and
agentsqueue recovery.
| File | Description |
|---|---|
src/Wolverine/Transports/Local/BufferedLocalQueue.cs |
Enables automatic replacement of faulted receiving blocks. |
src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs |
Adds block reconstruction, replacement locking, and stale-fault detection. |
src/Testing/CoreTests/Transports/Local/faulted_buffered_local_queue.cs |
Tests recovery after terminal faults in application and agents queues. |
Review effort: Balanced
Comment on lines
+139
to
+143
| if (!HasFaulted || _drained) return _receivingBlock; | ||
|
|
||
| // Whatever the faulted block still held is lost, as it would be in a crash; the queue is in-memory. | ||
| var faulted = _receivingBlock; | ||
| _receivingBlock = buildReceivingBlock(); |
Comment on lines
+143
to
+146
| _receivingBlock = buildReceivingBlock(); | ||
| HasFaulted = false; | ||
|
|
||
| _ = faulted.DisposeAsync().AsTask().ContinueWith(_ => { }, TaskScheduler.Default); |
Comment on lines
+163
to
165
| if (!ReferenceEquals(block, _receivingBlock)) return; | ||
|
|
||
| HasFaulted = true; |
|
|
||
| _ = faulted.DisposeAsync().AsTask().ContinueWith(_ => { }, TaskScheduler.Default); | ||
|
|
||
| _logger.LogWarning("The faulted local worker queue for {Uri} was replaced with a new one", Uri); |
Member
|
Thank you, Anne. This is superseded by #5382, which is now merged. Your commit went in unchanged, so you're its author, and you're co-author on the follow-up commit. The follow-up added:
Closing this one in favor of #5382. |
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
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.
Fixes #5358
Problem
A buffered local queue whose
Block<Envelope>faulted terminally (#506) rejected every later message until the process restarted. Listener-backed receivers are rebuilt byListeningAgent/BackPressureAgentwhenHasFaultedis set, but nothing owns aBufferedLocalQueuethat way. On theagentsqueue this left a node heartbeating and getting agents assigned while it could not run one agent command.Fix
BufferedReceiverbuilds its receiving block inbuildReceivingBlock(), so it can build it again.RebuildsFaultedBlockis true and the block has faulted, the nextEnqueue/EnqueueAsyncswaps in a new block (under a lock) and clearsHasFaulted. OnlyBufferedLocalQueueopts in; receivers behind a listener keep being rebuilt whole by theirListeningAgent.onBlockErrorignores a terminal fault from a block that was already replaced, so it cannot mark the new block as faulted. For the partitioned case the check uses the outer block, as before.Items still buffered in the faulted block are lost, like in a crash; the queue is in-memory.
DurableLocalQueuehas the same gap and is not part of this PR (see the issue).Tests
CoreTests.Transports.Local.faulted_buffered_local_queuefaults the real block (the same path its consumer loop takes when an exception escapes it) and checks that a message published afterwards is handled, and that theagentsqueue takes envelopes again. Both fail on main with the exception from the issue.CoreTests.Runtime.WorkerQueues*,CoreTests.Runtime.Partitioning*,CoreTests.Transports.Local*andBackPressureAgentTestspass on net9.0 and net10.0.https://claude.ai/code/session_0176D1FySkqxczVGAYmFjNK7