Skip to content

Commit 5ef50f8

Browse files
jeremydmillerclaude
jeremydmiller
and
claude
committed
GH-3882: opt-in Buffered mode for global partitioned topology slots
GlobalPartitionedMessageTopology.SetExternalTopology force-set EndpointMode.Durable on every external slot and companion local queue after the user's configure callback ran, with no escape hatch. For lossy, re-reported traffic — CritterWatch telemetry being the motivating case — that store-and-forwards every observability envelope through the monitored application's own message store, multiplying its database traffic through the exact path the partitioned topology is recommended for. New: topology.Mode(EndpointMode.BufferedInMemory) opts the slots (external + companion local) out of Durable. Order-independent — the mode is stored on the topology and applied both retroactively (UseSharded*Queues builds the external topology eagerly, so a later Mode() call reaches back into the built slots) and at SetExternalTopology / LocalQueues time. Default stays Durable; Inline is rejected because partitioned slots depend on the external-listener → companion-local-queue bridge that inline endpoints bypass. CoreTests: 2336 passed / 0 failed (net9.0), including 7 new GlobalPartitionedMessageTopologyTests covering both orderings, LocalQueues after Mode, opt-back-to-Durable, the Inline rejection, and fluency. Closes GH-3882. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ri7jHRKhaFNBesQM1B5i1k
1 parent d495c6a commit 5ef50f8

2 files changed

Lines changed: 141 additions & 11 deletions

File tree

‎src/Testing/CoreTests/Runtime/Partitioning/global_partitioning_tests.cs‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,95 @@ public void set_external_topology_forces_durable_mode_on_all_local_endpoints()
8686
}
8787
}
8888

89+
[Fact]
90+
public void mode_buffered_before_external_topology_applies_to_external_and_companion_slots()
91+
{
92+
// GH-3882: the Buffered opt-in must survive SetExternalTopology's stamp, which runs
93+
// after the user's configure callback.
94+
var topology = CreateTopology();
95+
topology.Mode(EndpointMode.BufferedInMemory);
96+
97+
var external = CreateLocalTopology("ext", 3);
98+
topology.SetExternalTopology(external, "test");
99+
100+
foreach (var slot in external.Slots)
101+
{
102+
slot.Mode.ShouldBe(EndpointMode.BufferedInMemory);
103+
}
104+
105+
foreach (var slot in topology.LocalTopology!.Slots)
106+
{
107+
slot.Mode.ShouldBe(EndpointMode.BufferedInMemory);
108+
}
109+
}
110+
111+
[Fact]
112+
public void mode_buffered_after_external_topology_applies_retroactively()
113+
{
114+
// GH-3882: order-independent — UseSharded*Queues() runs SetExternalTopology eagerly, so a
115+
// Mode() call later in the same configure lambda has to reach back into the built slots.
116+
var topology = CreateTopology();
117+
var external = CreateLocalTopology("ext", 3);
118+
topology.SetExternalTopology(external, "test");
119+
120+
topology.Mode(EndpointMode.BufferedInMemory);
121+
122+
foreach (var slot in external.Slots)
123+
{
124+
slot.Mode.ShouldBe(EndpointMode.BufferedInMemory);
125+
}
126+
127+
foreach (var slot in topology.LocalTopology!.Slots)
128+
{
129+
slot.Mode.ShouldBe(EndpointMode.BufferedInMemory);
130+
}
131+
}
132+
133+
[Fact]
134+
public void mode_applies_to_local_queues_declared_after_the_mode_call()
135+
{
136+
var topology = CreateTopology();
137+
topology.Mode(EndpointMode.BufferedInMemory);
138+
139+
topology.LocalQueues("mine", 4);
140+
141+
foreach (var slot in topology.LocalTopology!.Slots)
142+
{
143+
slot.Mode.ShouldBe(EndpointMode.BufferedInMemory);
144+
}
145+
}
146+
147+
[Fact]
148+
public void mode_can_opt_back_into_durable()
149+
{
150+
var topology = CreateTopology();
151+
var external = CreateLocalTopology("ext", 3);
152+
topology.SetExternalTopology(external, "test");
153+
154+
topology.Mode(EndpointMode.BufferedInMemory);
155+
topology.Mode(EndpointMode.Durable);
156+
157+
foreach (var slot in external.Slots)
158+
{
159+
slot.Mode.ShouldBe(EndpointMode.Durable);
160+
}
161+
}
162+
163+
[Fact]
164+
public void mode_rejects_inline()
165+
{
166+
var topology = CreateTopology();
167+
168+
Should.Throw<ArgumentOutOfRangeException>(() => topology.Mode(EndpointMode.Inline));
169+
}
170+
171+
[Fact]
172+
public void mode_is_fluent()
173+
{
174+
var topology = CreateTopology();
175+
topology.Mode(EndpointMode.BufferedInMemory).ShouldBeSameAs(topology);
176+
}
177+
89178
[Fact]
90179
public void set_external_topology_tags_external_endpoints_with_companion_local_queue_uris()
91180
{

‎src/Wolverine/Runtime/Partitioning/GlobalPartitionedMessageTopology.cs‎

Lines changed: 52 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ public class GlobalPartitionedMessageTopology
1313
private readonly HashSet<string> _messageTypeNames = new(StringComparer.OrdinalIgnoreCase);
1414
private PartitionedMessageTopology? _externalTopology;
1515
private LocalPartitionedMessageTopology? _localTopology;
16+
private EndpointMode _mode = EndpointMode.Durable;
1617

1718
public GlobalPartitionedMessageTopology(WolverineOptions options)
1819
{
@@ -22,9 +23,54 @@ public GlobalPartitionedMessageTopology(WolverineOptions options)
2223
internal PartitionedMessageTopology? ExternalTopology => _externalTopology;
2324
internal LocalPartitionedMessageTopology? LocalTopology => _localTopology;
2425

26+
/// <summary>
27+
/// Opt the partitioned slots — the external endpoints AND their companion local queues — out of
28+
/// the default <see cref="EndpointMode.Durable"/>. Use <see cref="EndpointMode.BufferedInMemory"/>
29+
/// for lossy, re-reported traffic (telemetry, metrics) where at-most-once delivery with
30+
/// sender-side FIFO ordering is sufficient and store-and-forwarding every envelope through the
31+
/// application's message store is the wrong trade (GH-3882). The default remains Durable.
32+
/// Order-independent: may be called before or after the external topology is configured.
33+
/// </summary>
34+
/// <exception cref="ArgumentOutOfRangeException">
35+
/// <see cref="EndpointMode.Inline"/> is not valid here — partitioned slots depend on queueing
36+
/// (the external listener bridges into the companion local queue) that inline endpoints bypass.
37+
/// </exception>
38+
public GlobalPartitionedMessageTopology Mode(EndpointMode mode)
39+
{
40+
if (mode == EndpointMode.Inline)
41+
{
42+
throw new ArgumentOutOfRangeException(nameof(mode),
43+
"EndpointMode.Inline is not supported for global partitioned topologies. Partitioned slots depend on queueing (the external listener bridges into the companion local queue), which inline endpoints bypass. Use Durable (default) or BufferedInMemory.");
44+
}
45+
46+
_mode = mode;
47+
applyMode();
48+
return this;
49+
}
50+
51+
private void applyMode()
52+
{
53+
if (_externalTopology != null)
54+
{
55+
foreach (var slot in _externalTopology.Slots)
56+
{
57+
slot.Mode = _mode;
58+
}
59+
}
60+
61+
if (_localTopology != null)
62+
{
63+
foreach (var slot in _localTopology.Slots)
64+
{
65+
slot.Mode = _mode;
66+
}
67+
}
68+
}
69+
2570
public void LocalQueues(string baseQueueName, int numberOfEndpoints)
2671
{
2772
_localTopology = new LocalPartitionedMessageTopology(_options, baseQueueName, numberOfEndpoints);
73+
applyMode();
2874
}
2975

3076
internal void SetExternalTopology(Func<WolverineOptions, PartitionedMessageTopology> factory, string baseName)
@@ -43,17 +89,12 @@ internal void SetExternalTopology(PartitionedMessageTopology topology, string ba
4389
_localTopology = new LocalPartitionedMessageTopology(_options, localBaseName, topology.Slots.Count);
4490
}
4591

46-
// Force durable mode on all external endpoints
47-
foreach (var slot in topology.Slots)
48-
{
49-
slot.Mode = EndpointMode.Durable;
50-
}
51-
52-
// Force durable mode on all local endpoints
53-
foreach (var slot in _localTopology.Slots)
54-
{
55-
slot.Mode = EndpointMode.Durable;
56-
}
92+
// Stamp the topology's mode — Durable unless the user opted the slots into
93+
// BufferedInMemory via Mode() (GH-3882) — on every external endpoint and companion local
94+
// queue. Runs after the user's configure callback, which is why the opt-out lives on this
95+
// class rather than on the individual slots: a per-slot Mode set in the callback would be
96+
// overwritten here.
97+
applyMode();
5798

5899
// Tag each external slot endpoint with its companion local queue URI
59100
// Only tag if slot counts match; mismatches will be caught by AssertValidity()

0 commit comments

Comments
 (0)