1package brokerwire
2
3import (
4 "sync"
5 "testing"
6
7 "github.com/stretchr/testify/assert"
8 "github.com/stretchr/testify/require"
9
10 "github.com/bnema/vev/internal/ports"
11)
12
13func testBrokerScope() Scope {
14 epoch, connection := testSnapshotScope()
15 return Scope{Epoch: epoch, Connection: connection}
16}
17
18func testForeignScope() Scope {
19 return Scope{Epoch: 8, Connection: testConnectionID(0x32)}
20}
21
22// connectionOperations and connectionStreams expose the trackers bound to a
23// connection to in-package tests. Production code never touches them: the
24// Connection lifecycle methods hold the connection lock across the ready check
25// and the tracker mutation, so no caller can bypass the connection state
26// machine. The trackers are assigned once by NewConnection and never replaced,
27// so these reads need no lock.
28func connectionOperations(c *Connection) *OperationTracker { return c.operations }
29
30func connectionStreams(c *Connection) *StreamTracker { return c.streams }
31
32// operationIDAt builds one unique non-zero operation identity.
33func operationIDAt(i int) ports.BrokerOperationID {
34 var id ports.BrokerOperationID
35 id[0] = byte(i >> 8)
36 id[1] = byte(i)
37 id[2] = 0x5a
38 id[15] = 0x01
39 return id
40}
41
42func newTestConnection(t *testing.T) *Connection {
43 t.Helper()
44 conn, err := NewConnection(testBrokerScope())
45 require.NoError(t, err)
46 require.Equal(t, ConnNew, conn.State())
47 return conn
48}
49
50func awaitRegisterConnection(t *testing.T) *Connection {
51 t.Helper()
52 conn := newTestConnection(t)
53 require.NoError(t, conn.SignalPreamble())
54 require.Equal(t, ConnAwaitRegister, conn.State())
55 return conn
56}
57
58func readyTestConnection(t *testing.T) *Connection {
59 t.Helper()
60 conn := awaitRegisterConnection(t)
61 require.NoError(t, conn.Register())
62 require.Equal(t, ConnReady, conn.State())
63 require.True(t, conn.Ready())
64 return conn
65}
66
67func closingTestConnection(t *testing.T) *Connection {
68 t.Helper()
69 conn := readyTestConnection(t)
70 require.NoError(t, conn.BeginClose())
71 require.Equal(t, ConnClosing, conn.State())
72 return conn
73}
74
75func closedTestConnection(t *testing.T) *Connection {
76 t.Helper()
77 conn := closingTestConnection(t)
78 require.NoError(t, conn.FinishClose())
79 require.Equal(t, ConnClosed, conn.State())
80 return conn
81}
82
83func TestNewConnectionScope(t *testing.T) {
84 cases := []struct {
85 name string
86 scope Scope
87 wantErr bool
88 }{
89 {"valid", testBrokerScope(), false},
90 {"zero epoch", Scope{Connection: testConnectionID(0x31)}, true},
91 {"zero connection", Scope{Epoch: 7}, true},
92 {"zero scope", Scope{}, true},
93 }
94 for _, tc := range cases {
95 t.Run(tc.name, func(t *testing.T) {
96 conn, err := NewConnection(tc.scope)
97 if tc.wantErr {
98 require.ErrorIs(t, err, ErrInvalidScope)
99 require.Nil(t, conn)
100 return
101 }
102 require.NoError(t, err)
103 require.Equal(t, tc.scope, conn.Scope())
104 require.Equal(t, ConnNew, conn.State())
105 require.False(t, conn.Ready())
106 require.Zero(t, connectionOperations(conn).Pending())
107 require.Zero(t, connectionStreams(conn).Active())
108 })
109 }
110}
111
112func TestConnStateString(t *testing.T) {
113 require.Equal(t, "new", ConnNew.String())
114 require.Equal(t, "await_register", ConnAwaitRegister.String())
115 require.Equal(t, "ready", ConnReady.String())
116 require.Equal(t, "closing", ConnClosing.String())
117 require.Equal(t, "closed", ConnClosed.String())
118 require.Equal(t, "unknown", ConnState(200).String())
119}
120
121// TestConnectionLifecycle is the handshake and shutdown table: New ->
122// AwaitRegister after the preamble signal -> Ready after exactly one
123// Register -> Closing -> Closed.
124func TestConnectionLifecycle(t *testing.T) {
125 op := operationIDAt(1)
126 part := snapshotPartsFor(1, 1, sampleSnapshotDaemons(), nil)[0]
127 cases := []struct {
128 name string
129 setup func(*testing.T) *Connection
130 run func(*Connection) error
131 wantErr error
132 wantState ConnState
133 }{
134 {"preamble in new", newTestConnection, func(c *Connection) error { return c.SignalPreamble() }, nil, ConnAwaitRegister},
135 {"register before preamble", newTestConnection, func(c *Connection) error { return c.Register() }, ErrConnectionState, ConnNew},
136 {"subscribe before preamble", newTestConnection, func(c *Connection) error { return c.Subscribe(1) }, ErrConnectionState, ConnNew},
137 {"operation before preamble", newTestConnection, func(c *Connection) error { return c.AdmitOperation(op) }, ErrConnectionState, ConnNew},
138 {"stream before preamble", newTestConnection, func(c *Connection) error { return c.OpenStream(1) }, ErrConnectionState, ConnNew},
139 {"snapshot part before preamble", newTestConnection, func(c *Connection) error {
140 _, _, err := c.AddSnapshotPart(part)
141 return err
142 }, ErrConnectionState, ConnNew},
143 {"preamble twice", awaitRegisterConnection, func(c *Connection) error { return c.SignalPreamble() }, ErrConnectionState, ConnAwaitRegister},
144 {"subscribe before register", awaitRegisterConnection, func(c *Connection) error { return c.Subscribe(1) }, ErrConnectionState, ConnAwaitRegister},
145 {"register after preamble", awaitRegisterConnection, func(c *Connection) error { return c.Register() }, nil, ConnReady},
146 {"register twice", readyTestConnection, func(c *Connection) error { return c.Register() }, ErrDuplicateRegister, ConnReady},
147 {"preamble when ready", readyTestConnection, func(c *Connection) error { return c.SignalPreamble() }, ErrConnectionState, ConnReady},
148 {"finish close before closing", readyTestConnection, func(c *Connection) error { return c.FinishClose() }, ErrConnectionState, ConnReady},
149 {"finish close in new", newTestConnection, func(c *Connection) error { return c.FinishClose() }, ErrConnectionState, ConnNew},
150 {"begin close when ready", readyTestConnection, func(c *Connection) error { return c.BeginClose() }, nil, ConnClosing},
151 {"begin close idempotent", closingTestConnection, func(c *Connection) error { return c.BeginClose() }, nil, ConnClosing},
152 {"finish close after begin", closingTestConnection, func(c *Connection) error { return c.FinishClose() }, nil, ConnClosed},
153 {"finish close idempotent", closedTestConnection, func(c *Connection) error { return c.FinishClose() }, nil, ConnClosed},
154 {"subscribe while closing", closingTestConnection, func(c *Connection) error { return c.Subscribe(1) }, ErrConnectionClosed, ConnClosing},
155 {"operation while closing", closingTestConnection, func(c *Connection) error { return c.AdmitOperation(op) }, ErrConnectionClosed, ConnClosing},
156 {"stream while closing", closingTestConnection, func(c *Connection) error { return c.OpenStream(1) }, ErrConnectionClosed, ConnClosing},
157 {"part while closing", closingTestConnection, func(c *Connection) error {
158 _, _, err := c.AddSnapshotPart(part)
159 return err
160 }, ErrConnectionClosed, ConnClosing},
161 {"data while closing", closingTestConnection, func(c *Connection) error {
162 _, err := c.StreamData(1)
163 return err
164 }, ErrConnectionClosed, ConnClosing},
165 {"operation while closed", closedTestConnection, func(c *Connection) error { return c.AdmitOperation(op) }, ErrConnectionClosed, ConnClosed},
166 {"preamble while closed", closedTestConnection, func(c *Connection) error { return c.SignalPreamble() }, ErrConnectionClosed, ConnClosed},
167 {"begin close while closed", closedTestConnection, func(c *Connection) error { return c.BeginClose() }, nil, ConnClosed},
168 }
169 for _, tc := range cases {
170 t.Run(tc.name, func(t *testing.T) {
171 conn := tc.setup(t)
172 err := tc.run(conn)
173 if tc.wantErr == nil {
174 require.NoError(t, err)
175 } else {
176 require.ErrorIs(t, err, tc.wantErr)
177 }
178 require.Equal(t, tc.wantState, conn.State())
179 })
180 }
181}
182
183// TestConnectionLifecycleSequence walks the exact accepted lifecycle and
184// proves close is idempotent in both steps.
185func TestConnectionLifecycleSequence(t *testing.T) {
186 conn := newTestConnection(t)
187 require.Equal(t, ConnNew, conn.State())
188 require.NoError(t, conn.SignalPreamble())
189 require.Equal(t, ConnAwaitRegister, conn.State())
190 require.NoError(t, conn.Register())
191 require.Equal(t, ConnReady, conn.State())
192 require.NoError(t, conn.BeginClose())
193 require.Equal(t, ConnClosing, conn.State())
194 require.NoError(t, conn.BeginClose())
195 require.Equal(t, ConnClosing, conn.State())
196 require.NoError(t, conn.FinishClose())
197 require.Equal(t, ConnClosed, conn.State())
198 require.NoError(t, conn.FinishClose())
199 require.Equal(t, ConnClosed, conn.State())
200 require.False(t, conn.Ready())
201}
202
203// TestConnectionSubscriptionGenerations is the subscription generation table:
204// resubscribe advances strictly, resync and unsubscribe must carry the active
205// generation exactly.
206func TestConnectionSubscriptionGenerations(t *testing.T) {
207 type step struct {
208 name string
209 run func(*Connection) error
210 want error
211 check func(*testing.T, *Connection)
212 }
213 cases := []struct {
214 name string
215 steps []step
216 }{
217 {
218 name: "zero generation refused",
219 steps: []step{
220 {name: "subscribe zero", run: func(c *Connection) error { return c.Subscribe(0) }, want: ErrInvalidGeneration},
221 {name: "no active subscription", run: func(c *Connection) error { return c.Unsubscribe(0) }, want: ErrNotSubscribed},
222 },
223 },
224 {
225 name: "resubscribe advances strictly",
226 steps: []step{
227 {name: "subscribe one", run: func(c *Connection) error { return c.Subscribe(1) }},
228 {name: "subscribe one again", run: func(c *Connection) error { return c.Subscribe(1) }, want: ErrStaleGeneration},
229 {name: "subscribe two", run: func(c *Connection) error { return c.Subscribe(2) }},
230 {name: "subscribe zero", run: func(c *Connection) error { return c.Subscribe(0) }, want: ErrInvalidGeneration},
231 },
232 },
233 {
234 name: "resync must match the active generation",
235 steps: []step{
236 {name: "resync before subscribe", run: func(c *Connection) error { return c.Resync(1) }, want: ErrNotSubscribed},
237 {name: "subscribe", run: func(c *Connection) error { return c.Subscribe(3) }},
238 {name: "resync active", run: func(c *Connection) error { return c.Resync(3) }},
239 {name: "resync stale", run: func(c *Connection) error { return c.Resync(2) }, want: ErrStaleGeneration},
240 {name: "resync future", run: func(c *Connection) error { return c.Resync(4) }, want: ErrFutureGeneration},
241 },
242 },
243 {
244 name: "unsubscribe must match and generations are never reused",
245 steps: []step{
246 {name: "subscribe", run: func(c *Connection) error { return c.Subscribe(2) }},
247 {name: "unsubscribe stale", run: func(c *Connection) error { return c.Unsubscribe(1) }, want: ErrStaleGeneration},
248 {name: "unsubscribe future", run: func(c *Connection) error { return c.Unsubscribe(3) }, want: ErrFutureGeneration},
249 {name: "unsubscribe active", run: func(c *Connection) error { return c.Unsubscribe(2) }},
250 {name: "unsubscribe twice", run: func(c *Connection) error { return c.Unsubscribe(2) }, want: ErrNotSubscribed},
251 {name: "resubscribe consumed generation", run: func(c *Connection) error { return c.Subscribe(2) }, want: ErrStaleGeneration},
252 {name: "resubscribe newer generation", run: func(c *Connection) error { return c.Subscribe(3) }},
253 },
254 },
255 }
256 for _, tc := range cases {
257 t.Run(tc.name, func(t *testing.T) {
258 conn := readyTestConnection(t)
259 for _, s := range tc.steps {
260 err := s.run(conn)
261 if s.want == nil {
262 require.NoError(t, err, s.name)
263 } else {
264 require.ErrorIs(t, err, s.want, s.name)
265 }
266 if s.check != nil {
267 s.check(t, conn)
268 }
269 }
270 })
271 }
272}
273
274// stageConnectionPart stages one snapshot part through the connection and
275// requires acceptance.
276func stageConnectionPart(t *testing.T, conn *Connection, part SnapshotPart) {
277 t.Helper()
278 _, _, err := conn.AddSnapshotPart(part)
279 require.NoError(t, err)
280}
281
282// TestConnectionSubscriptionStagingRetention proves a generation advance
283// discards in-flight staging, keeps the committed snapshot, and that close
284// retains the commit until the connection is finally closed.
285func TestConnectionSubscriptionStagingRetention(t *testing.T) {
286 conn := readyTestConnection(t)
287 require.NoError(t, conn.Subscribe(1))
288 for _, part := range snapshotPartsFor(1, 4, sampleSnapshotDaemons(), sampleSnapshotTombstones()) {
289 stageConnectionPart(t, conn, part)
290 }
291 committed, ok := conn.Snapshot()
292 require.True(t, ok)
293 require.Equal(t, ports.BrokerRevision(4), committed.Revision)
294
295 // In-flight staging of the active generation is discarded on advance.
296 inFlight := snapshotPartsFor(1, 6, sampleSnapshotDaemons(), nil)
297 stageConnectionPart(t, conn, inFlight[0])
298 require.True(t, conn.StagingActive())
299 require.NoError(t, conn.Subscribe(2))
300 require.False(t, conn.StagingActive())
301 committed, ok = conn.Snapshot()
302 require.True(t, ok, "generation advance keeps the committed snapshot")
303 require.Equal(t, ports.BrokerRevision(4), committed.Revision)
304
305 // A part of the superseded generation is refused and never applied, a
306 // part of a future generation is refused, and the fresh generation must
307 // start at index 0.
308 stale := inFlight[1]
309 _, _, err := conn.AddSnapshotPart(stale)
310 require.ErrorIs(t, err, ErrStaleGeneration)
311 future := inFlight[1]
312 future.Generation = 3
313 _, _, err = conn.AddSnapshotPart(future)
314 require.ErrorIs(t, err, ErrFutureGeneration)
315 next := snapshotPartsFor(2, 5, sampleSnapshotDaemons(), nil)
316 _, _, err = conn.AddSnapshotPart(next[1])
317 require.ErrorIs(t, err, ErrSnapshotInvalid)
318 require.False(t, conn.StagingActive())
319 for _, part := range next {
320 stageConnectionPart(t, conn, part)
321 }
322 committed, ok = conn.Snapshot()
323 require.True(t, ok)
324 require.Equal(t, ports.BrokerRevision(5), committed.Revision)
325
326 require.NoError(t, conn.BeginClose())
327 _, ok = conn.Snapshot()
328 require.True(t, ok, "closing keeps the committed snapshot")
329 require.NoError(t, conn.FinishClose())
330 _, ok = conn.Snapshot()
331 require.False(t, ok, "closed releases the snapshot")
332 require.False(t, conn.StagingActive())
333}
334
335// TestConnectionSnapshotParts proves snapshot staging requires Ready plus an
336// active subscription, and that every part is fenced by scope and generation.
337func TestConnectionSnapshotParts(t *testing.T) {
338 t.Run("part without subscription", func(t *testing.T) {
339 conn := readyTestConnection(t)
340 _, _, err := conn.AddSnapshotPart(snapshotPartsFor(1, 1, nil, nil)[0])
341 require.ErrorIs(t, err, ErrNotSubscribed)
342 })
343
344 t.Run("unsubscribe stops staging", func(t *testing.T) {
345 conn := readyTestConnection(t)
346 require.NoError(t, conn.Subscribe(1))
347 require.NoError(t, conn.Unsubscribe(1))
348 _, _, err := conn.AddSnapshotPart(snapshotPartsFor(1, 1, nil, nil)[0])
349 require.ErrorIs(t, err, ErrNotSubscribed)
350 })
351
352 t.Run("future generation rejected", func(t *testing.T) {
353 conn := readyTestConnection(t)
354 require.NoError(t, conn.Subscribe(2))
355 part := snapshotPartsFor(1, 1, sampleSnapshotDaemons(), nil)[0]
356 part.Generation = 3
357 _, _, err := conn.AddSnapshotPart(part)
358 require.ErrorIs(t, err, ErrFutureGeneration)
359 part.Generation = 1
360 _, _, err = conn.AddSnapshotPart(part)
361 require.ErrorIs(t, err, ErrStaleGeneration)
362 })
363
364 t.Run("foreign scope rejected", func(t *testing.T) {
365 conn := readyTestConnection(t)
366 require.NoError(t, conn.Subscribe(2))
367 part := snapshotPartsFor(2, 1, sampleSnapshotDaemons(), nil)[0]
368 part.Epoch = 8
369 _, _, err := conn.AddSnapshotPart(part)
370 require.ErrorIs(t, err, ErrScopeMismatch)
371 part = snapshotPartsFor(2, 1, sampleSnapshotDaemons(), nil)[0]
372 part.Connection = testConnectionID(0x77)
373 _, _, err = conn.AddSnapshotPart(part)
374 require.ErrorIs(t, err, ErrScopeMismatch)
375 })
376
377 t.Run("complete transfer through the connection", func(t *testing.T) {
378 conn := readyTestConnection(t)
379 require.NoError(t, conn.Subscribe(9))
380 parts := snapshotPartsFor(9, 2, sampleSnapshotDaemons(), sampleSnapshotTombstones())
381 var published bool
382 for _, part := range parts {
383 completed, _, err := conn.AddSnapshotPart(part)
384 require.NoError(t, err)
385 if completed {
386 published = true
387 }
388 }
389 require.True(t, published)
390 committed, ok := conn.Snapshot()
391 require.True(t, ok)
392 require.Equal(t, ports.BrokerRevision(2), committed.Revision)
393 })
394}
395
396// TestConnectionExactScopeRegistration proves the connection keeps exactly
397// the scope assigned at accept: every frame must carry it, refused frames
398// change nothing, and the scope survives the whole lifecycle.
399func TestConnectionExactScopeRegistration(t *testing.T) {
400 conn := readyTestConnection(t)
401 assigned := conn.Scope()
402 require.Equal(t, testBrokerScope(), assigned)
403 other := Scope{Epoch: assigned.Epoch + 1, Connection: assigned.Connection}
404
405 require.ErrorIs(t, connectionOperations(conn).Admit(other, operationIDAt(1)), ErrScopeMismatch)
406 require.ErrorIs(t, connectionOperations(conn).Complete(other, operationIDAt(1), ports.BrokerOutcomeOK), ErrScopeMismatch)
407 require.ErrorIs(t, connectionStreams(conn).Open(other, 1), ErrScopeMismatch)
408 _, err := connectionStreams(conn).Data(other, 1)
409 require.ErrorIs(t, err, ErrScopeMismatch)
410 _, err = connectionStreams(conn).Close(other, 1)
411 require.ErrorIs(t, err, ErrScopeMismatch)
412
413 part := snapshotPartsFor(1, 1, sampleSnapshotDaemons(), nil)[0]
414 part.Connection = other.Connection
415 part.Epoch = other.Epoch
416 require.NoError(t, conn.Subscribe(1))
417 _, _, err = conn.AddSnapshotPart(part)
418 require.ErrorIs(t, err, ErrScopeMismatch)
419
420 // Refused frames leave no state behind, and the accepted scope still works.
421 require.Equal(t, assigned, conn.Scope())
422 require.Zero(t, connectionOperations(conn).Pending())
423 require.Zero(t, connectionStreams(conn).Active())
424 require.False(t, conn.StagingActive())
425 require.NoError(t, conn.AdmitOperation(operationIDAt(2)))
426 require.Equal(t, 1, connectionOperations(conn).Pending())
427
428 require.NoError(t, conn.BeginClose())
429 require.NoError(t, conn.FinishClose())
430 require.Equal(t, assigned, conn.Scope())
431}
432
433// TestOperationTrackerErrors is the operation admission taxonomy table.
434func TestOperationTrackerErrors(t *testing.T) {
435 scope := testBrokerScope()
436 op := operationIDAt(3)
437 cases := []struct {
438 name string
439 setup func(*testing.T, *OperationTracker)
440 run func(*OperationTracker) error
441 want error
442 }{
443 {"admit accepted", nil, func(tr *OperationTracker) error { return tr.Admit(scope, op) }, nil},
444 {"admit zero identity", nil, func(tr *OperationTracker) error { return tr.Admit(scope, ports.BrokerOperationID{}) }, ErrInvalidOperation},
445 {"admit foreign scope", nil, func(tr *OperationTracker) error { return tr.Admit(testForeignScope(), op) }, ErrScopeMismatch},
446 {"admit duplicate pending", func(t *testing.T, tr *OperationTracker) {
447 require.NoError(t, tr.Admit(scope, op))
448 }, func(tr *OperationTracker) error { return tr.Admit(scope, op) }, ErrOperationPending},
449 {"admit completed", func(t *testing.T, tr *OperationTracker) {
450 require.NoError(t, tr.Admit(scope, op))
451 require.NoError(t, tr.Complete(scope, op, ports.BrokerOutcomeOK))
452 }, func(tr *OperationTracker) error { return tr.Admit(scope, op) }, ErrOperationCompleted},
453 {"complete unknown", nil, func(tr *OperationTracker) error {
454 return tr.Complete(scope, op, ports.BrokerOutcomeOK)
455 }, ErrUnknownOperation},
456 {"complete invalid outcome", func(t *testing.T, tr *OperationTracker) {
457 require.NoError(t, tr.Admit(scope, op))
458 }, func(tr *OperationTracker) error { return tr.Complete(scope, op, 0) }, ErrInvalidOutcome},
459 {"complete foreign scope", func(t *testing.T, tr *OperationTracker) {
460 require.NoError(t, tr.Admit(scope, op))
461 }, func(tr *OperationTracker) error {
462 return tr.Complete(testForeignScope(), op, ports.BrokerOutcomeOK)
463 }, ErrScopeMismatch},
464 {"complete twice", func(t *testing.T, tr *OperationTracker) {
465 require.NoError(t, tr.Admit(scope, op))
466 require.NoError(t, tr.Complete(scope, op, ports.BrokerOutcomeFailed))
467 }, func(tr *OperationTracker) error { return tr.Complete(scope, op, ports.BrokerOutcomeOK) }, ErrOperationCompleted},
468 }
469 for _, tc := range cases {
470 t.Run(tc.name, func(t *testing.T) {
471 tr, err := newOperationTracker(scope)
472 require.NoError(t, err)
473 if tc.setup != nil {
474 tc.setup(t, tr)
475 }
476 err = tc.run(tr)
477 if tc.want == nil {
478 require.NoError(t, err)
479 } else {
480 require.ErrorIs(t, err, tc.want)
481 }
482 })
483 }
484}
485
486func TestNewOperationTrackerValidatesScope(t *testing.T) {
487 tr, err := newOperationTracker(Scope{})
488 require.ErrorIs(t, err, ErrInvalidScope)
489 require.Nil(t, tr)
490}
491
492// TestOperationTrackerPendingBound proves at most 64 operations are in flight.
493func TestOperationTrackerPendingBound(t *testing.T) {
494 scope := testBrokerScope()
495 tr, err := newOperationTracker(scope)
496 require.NoError(t, err)
497 for i := 0; i < MaxPendingOperations; i++ {
498 require.NoError(t, tr.Admit(scope, operationIDAt(i)))
499 }
500 require.Equal(t, MaxPendingOperations, tr.Pending())
501 require.ErrorIs(t, tr.Admit(scope, operationIDAt(MaxPendingOperations)), ErrTooManyPendingOperations)
502 require.Equal(t, MaxPendingOperations, tr.Pending())
503
504 // Completing one frees exactly one slot.
505 require.NoError(t, tr.Complete(scope, operationIDAt(0), ports.BrokerOutcomeOK))
506 require.Equal(t, MaxPendingOperations-1, tr.Pending())
507 require.NoError(t, tr.Admit(scope, operationIDAt(MaxPendingOperations)))
508 require.Equal(t, MaxPendingOperations, tr.Pending())
509 require.ErrorIs(t, tr.Admit(scope, operationIDAt(MaxPendingOperations+1)), ErrTooManyPendingOperations)
510}
511
512// TestOperationTrackerDedupBound proves the dedup set is bounded to 4096
513// identities and 64 KiB, evicting the oldest record first.
514func TestOperationTrackerDedupBound(t *testing.T) {
515 scope := testBrokerScope()
516 tr, err := newOperationTracker(scope)
517 require.NoError(t, err)
518 require.Equal(t, 4096, MaxTrackedOperations)
519 require.Equal(t, 65536, MaxOperationDedupBytes)
520
521 total := MaxTrackedOperations + 128
522 for i := 0; i < total; i++ {
523 op := operationIDAt(i)
524 require.NoError(t, tr.Admit(scope, op))
525 require.NoError(t, tr.Complete(scope, op, ports.BrokerOutcomeOK))
526 require.LessOrEqual(t, tr.DedupBytes(), MaxOperationDedupBytes)
527 }
528 require.Equal(t, MaxTrackedOperations, tr.Tracked())
529 require.Equal(t, MaxOperationDedupBytes, tr.DedupBytes())
530 require.Zero(t, tr.Pending())
531
532 // The oldest records were evicted: they are no longer deduped, while the
533 // newest records still are.
534 evicted := operationIDAt(0)
535 _, ok := tr.Outcome(evicted)
536 require.False(t, ok)
537 require.NoError(t, tr.Admit(scope, evicted))
538 require.ErrorIs(t, tr.Admit(scope, operationIDAt(total-1)), ErrOperationCompleted)
539}
540
541// TestOperationTrackerOutcomeUnknownNoReplay proves outcome-unknown is
542// recorded like any outcome and a completed identity is never replayed.
543func TestOperationTrackerOutcomeUnknownNoReplay(t *testing.T) {
544 scope := testBrokerScope()
545 tr, err := newOperationTracker(scope)
546 require.NoError(t, err)
547 cases := []ports.BrokerMutationOutcome{ports.BrokerOutcomeOK, ports.BrokerOutcomeFailed, ports.BrokerOutcomeUnknown}
548 for i, outcome := range cases {
549 op := operationIDAt(100 + i)
550 require.NoError(t, tr.Admit(scope, op))
551 require.NoError(t, tr.Complete(scope, op, outcome))
552 recorded, ok := tr.Outcome(op)
553 require.True(t, ok)
554 require.Equal(t, outcome, recorded)
555 require.ErrorIs(t, tr.Admit(scope, op), ErrOperationCompleted, "never replays %s", outcome)
556 require.ErrorIs(t, tr.Complete(scope, op, ports.BrokerOutcomeFailed), ErrOperationCompleted)
557 }
558 // An unknown outcome still frees its pending slot.
559 require.Zero(t, tr.Pending())
560}
561
562// streamAction is one stream lifecycle event under test.
563type streamAction func(*StreamTracker, Scope, ports.BrokerStreamID) (StreamDisposition, error)
564
565func actionOpen(tr *StreamTracker, scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
566 if err := tr.Open(scope, stream); err != nil {
567 return 0, err
568 }
569 return StreamAccepted, nil
570}
571
572func actionOpened(tr *StreamTracker, scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
573 if err := tr.Opened(scope, stream); err != nil {
574 return 0, err
575 }
576 return StreamAccepted, nil
577}
578
579func actionData(tr *StreamTracker, scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
580 return tr.Data(scope, stream)
581}
582
583func actionProgress(tr *StreamTracker, scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
584 return tr.Progress(scope, stream)
585}
586
587func actionClose(tr *StreamTracker, scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
588 return tr.Close(scope, stream)
589}
590
591func TestStreamStateStrings(t *testing.T) {
592 require.Equal(t, "opening", StreamOpening.String())
593 require.Equal(t, "open", StreamOpen.String())
594 require.Equal(t, "retired", StreamRetired.String())
595 require.Equal(t, "unknown", StreamState(0).String())
596}
597
598func TestNewStreamTrackerValidatesScope(t *testing.T) {
599 tr, err := newStreamTracker(Scope{})
600 require.ErrorIs(t, err, ErrInvalidScope)
601 require.Nil(t, tr)
602}
603
604// TestStreamTrackerLifecycle is the stream state table: opening -> open ->
605// retired, with legal progress/data/close and late retired discard.
606func TestStreamTrackerLifecycle(t *testing.T) {
607 cases := []struct {
608 name string
609 stream ports.BrokerStreamID
610 steps []streamStep
611 wantState StreamState
612 wantFound bool
613 }{
614 {
615 name: "open then data then close",
616 stream: 1,
617 steps: []streamStep{
618 {name: "open", action: actionOpen, want: StreamAccepted},
619 {name: "data while opening", action: actionData, wantErr: ErrStreamState},
620 {name: "progress while opening", action: actionProgress, want: StreamAccepted},
621 {name: "opened", action: actionOpened, want: StreamAccepted},
622 {name: "opened twice", action: actionOpened, wantErr: ErrStreamState},
623 {name: "data when open", action: actionData, want: StreamAccepted},
624 {name: "progress when open", action: actionProgress, want: StreamAccepted},
625 {name: "close", action: actionClose, want: StreamAccepted},
626 {name: "late data", action: actionData, want: StreamDiscarded},
627 {name: "late progress", action: actionProgress, want: StreamDiscarded},
628 {name: "late close", action: actionClose, want: StreamDiscarded},
629 {name: "opened after close", action: actionOpened, wantErr: ErrStreamState},
630 },
631 wantState: StreamRetired,
632 wantFound: true,
633 },
634 {
635 name: "close while opening is terminal",
636 stream: 4,
637 steps: []streamStep{
638 {name: "open", action: actionOpen, want: StreamAccepted},
639 {name: "close", action: actionClose, want: StreamAccepted},
640 {name: "data after close", action: actionData, want: StreamDiscarded},
641 {name: "opened after close", action: actionOpened, wantErr: ErrStreamState},
642 },
643 wantState: StreamRetired,
644 wantFound: true,
645 },
646 }
647 for _, tc := range cases {
648 t.Run(tc.name, func(t *testing.T) {
649 scope := testBrokerScope()
650 tr, err := newStreamTracker(scope)
651 require.NoError(t, err)
652 for _, s := range tc.steps {
653 disposition, err := s.action(tr, scope, tc.stream)
654 if s.wantErr != nil {
655 require.ErrorIs(t, err, s.wantErr, s.name)
656 require.Zero(t, disposition, s.name)
657 continue
658 }
659 require.NoError(t, err, s.name)
660 require.Equal(t, s.want, disposition, s.name)
661 }
662 state, ok := tr.State(tc.stream)
663 require.Equal(t, tc.wantFound, ok)
664 require.Equal(t, tc.wantState, state)
665 require.Zero(t, tr.Active())
666 })
667 }
668}
669
670type streamStep struct {
671 name string
672 action streamAction
673 want StreamDisposition
674 wantErr error
675}
676
677// TestStreamTrackerOpenRules proves a stream identity is admitted at most once
678// inside the bounded anti-replay window, that a concurrent lower identity is not
679// mistaken for a replay, and that at most MaxStreams streams are concurrent.
680func TestStreamTrackerOpenRules(t *testing.T) {
681 scope := testBrokerScope()
682
683 t.Run("zero identity refused", func(t *testing.T) {
684 tr, err := newStreamTracker(scope)
685 require.NoError(t, err)
686 require.ErrorIs(t, tr.Open(scope, 0), ErrInvalidStream)
687 })
688
689 t.Run("each identity consumed at most once", func(t *testing.T) {
690 tr, err := newStreamTracker(scope)
691 require.NoError(t, err)
692 require.NoError(t, tr.Open(scope, 5))
693 require.ErrorIs(t, tr.Open(scope, 5), ErrStreamIDReused)
694 // A concurrent lower identity that was allocated first is admitted, not
695 // refused as a replay: two racing opens may be scheduled in either
696 // order.
697 require.NoError(t, tr.Open(scope, 1))
698 require.NoError(t, tr.Opened(scope, 5))
699 require.NoError(t, tr.Open(scope, 6))
700 if disposition, err := tr.Close(scope, 5); err != nil || disposition != StreamAccepted {
701 t.Fatalf("close: %v %v", disposition, err)
702 }
703 require.ErrorIs(t, tr.Open(scope, 5), ErrStreamIDReused)
704 require.ErrorIs(t, tr.Open(scope, 6), ErrStreamIDReused)
705 require.ErrorIs(t, tr.Open(scope, 1), ErrStreamIDReused)
706 require.NoError(t, tr.Open(scope, 7))
707 })
708
709 t.Run("identity evicted past the window is stale", func(t *testing.T) {
710 tr, err := newStreamTracker(scope)
711 require.NoError(t, err)
712 require.NoError(t, tr.Open(scope, 1))
713 require.NoError(t, tr.Open(scope, ports.BrokerStreamWindowSize+1))
714 require.ErrorIs(t, tr.Open(scope, 1), ErrStreamIDReused,
715 "an identity that trails the newest admitted one by the window is stale")
716 })
717
718 t.Run("concurrent ceiling", func(t *testing.T) {
719 tr, err := newStreamTracker(scope)
720 require.NoError(t, err)
721 for i := 1; i <= MaxStreams; i++ {
722 require.NoError(t, tr.Open(scope, ports.BrokerStreamID(i)))
723 }
724 require.Equal(t, MaxStreams, tr.Active())
725 require.ErrorIs(t, tr.Open(scope, ports.BrokerStreamID(MaxStreams+1)), ErrTooManyStreams)
726 // Retiring one stream frees exactly one slot for a fresh higher ID.
727 disposition, err := tr.Close(scope, 1)
728 require.NoError(t, err)
729 require.Equal(t, StreamAccepted, disposition)
730 require.Equal(t, MaxStreams-1, tr.Active())
731 require.NoError(t, tr.Open(scope, ports.BrokerStreamID(MaxStreams+2)))
732 require.ErrorIs(t, tr.Open(scope, ports.BrokerStreamID(MaxStreams+3)), ErrTooManyStreams)
733 })
734
735 t.Run("scope fence", func(t *testing.T) {
736 tr, err := newStreamTracker(scope)
737 require.NoError(t, err)
738 require.ErrorIs(t, tr.Open(testForeignScope(), 1), ErrScopeMismatch)
739 require.ErrorIs(t, tr.Opened(testForeignScope(), 1), ErrScopeMismatch)
740 _, err = tr.Data(testForeignScope(), 1)
741 require.ErrorIs(t, err, ErrScopeMismatch)
742 _, err = tr.Progress(testForeignScope(), 1)
743 require.ErrorIs(t, err, ErrScopeMismatch)
744 _, err = tr.Close(testForeignScope(), 1)
745 require.ErrorIs(t, err, ErrScopeMismatch)
746 require.Zero(t, tr.Active())
747 })
748}
749
750// TestStreamTrackerFutureReject proves a stream identity that was never
751// allocated is refused rather than treated as retired, and that a retired
752// record evicted after the retention bound is still discarded.
753func TestStreamTrackerFutureReject(t *testing.T) {
754 scope := testBrokerScope()
755 tr, err := newStreamTracker(scope)
756 require.NoError(t, err)
757 require.NoError(t, tr.Open(scope, 2))
758 require.NoError(t, tr.Opened(scope, 2))
759 for _, action := range []streamAction{actionOpened, actionData, actionProgress, actionClose} {
760 disposition, err := action(tr, scope, 3)
761 require.ErrorIs(t, err, ErrFutureStream)
762 require.Zero(t, disposition)
763 }
764 disposition, err := tr.Close(scope, 2)
765 require.NoError(t, err)
766 require.Equal(t, StreamAccepted, disposition)
767
768 // Retain more retired records than the bound so the oldest is evicted.
769 for i := 0; i < MaxRetiredStreams+8; i++ {
770 stream := ports.BrokerStreamID(10 + i)
771 require.NoError(t, tr.Open(scope, stream))
772 require.NoError(t, tr.Opened(scope, stream))
773 disposition, err := tr.Close(scope, stream)
774 require.NoError(t, err)
775 require.Equal(t, StreamAccepted, disposition)
776 }
777 evicted := ports.BrokerStreamID(2)
778 _, tracked := tr.State(evicted)
779 require.False(t, tracked, "oldest retired record is evicted")
780 disposition, err = tr.Data(scope, evicted)
781 require.NoError(t, err)
782 require.Equal(t, StreamDiscarded, disposition)
783 disposition, err = tr.Close(scope, evicted)
784 require.NoError(t, err)
785 require.Equal(t, StreamDiscarded, disposition)
786 _, err = tr.Data(scope, ports.BrokerStreamID(10+MaxRetiredStreams+8))
787 require.ErrorIs(t, err, ErrFutureStream)
788}
789
790// TestStreamTrackerNilSafety proves the nil-receiver paths.
791func TestStreamTrackerNilSafety(t *testing.T) {
792 var tr *StreamTracker
793 scope := testBrokerScope()
794 require.ErrorIs(t, tr.Open(scope, 1), ErrConnectionClosed)
795 require.ErrorIs(t, tr.Opened(scope, 1), ErrConnectionClosed)
796 _, err := tr.Data(scope, 1)
797 require.ErrorIs(t, err, ErrConnectionClosed)
798 _, err = tr.Progress(scope, 1)
799 require.ErrorIs(t, err, ErrConnectionClosed)
800 _, err = tr.Close(scope, 1)
801 require.ErrorIs(t, err, ErrConnectionClosed)
802 require.Zero(t, tr.Active())
803 _, ok := tr.State(1)
804 require.False(t, ok)
805}
806
807// TestConnectionStreamsAndOperations proves the connection-bound trackers
808// share the connection scope while every mutation still travels through the
809// Connection lifecycle methods.
810func TestConnectionStreamsAndOperations(t *testing.T) {
811 conn := readyTestConnection(t)
812 scope := conn.Scope()
813 op := operationIDAt(7)
814 require.NoError(t, conn.AdmitOperation(op))
815 require.Equal(t, 1, connectionOperations(conn).Pending())
816 require.NoError(t, conn.CompleteOperation(op, ports.BrokerOutcomeUnknown))
817 require.Zero(t, connectionOperations(conn).Pending())
818 recorded, ok := connectionOperations(conn).Outcome(op)
819 require.True(t, ok)
820 require.Equal(t, ports.BrokerOutcomeUnknown, recorded)
821
822 require.NoError(t, conn.OpenStream(1))
823 require.NoError(t, conn.StreamOpened(1))
824 disposition, err := conn.StreamProgress(1)
825 require.NoError(t, err)
826 require.Equal(t, StreamAccepted, disposition)
827 disposition, err = conn.StreamData(1)
828 require.NoError(t, err)
829 require.Equal(t, StreamAccepted, disposition)
830 disposition, err = conn.CloseStream(1)
831 require.NoError(t, err)
832 require.Equal(t, StreamAccepted, disposition)
833 disposition, err = conn.StreamData(1)
834 require.NoError(t, err)
835 require.Equal(t, StreamDiscarded, disposition)
836 require.Zero(t, connectionStreams(conn).Active())
837 require.Equal(t, scope, testBrokerScope())
838
839 // FinishClose releases both trackers.
840 require.NoError(t, conn.BeginClose())
841 require.NoError(t, conn.FinishClose())
842 require.Zero(t, connectionOperations(conn).Tracked())
843 require.Zero(t, connectionStreams(conn).Active())
844}
845
846// TestOperationTrackerConcurrentAdmitComplete is a deterministic race test:
847// exactly MaxPendingOperations distinct operations are admitted concurrently
848// and completed concurrently.
849func TestOperationTrackerConcurrentAdmitComplete(t *testing.T) {
850 scope := testBrokerScope()
851 tr, err := newOperationTracker(scope)
852 require.NoError(t, err)
853 var wg sync.WaitGroup
854 errs := make(chan error, 2*MaxPendingOperations)
855 for i := 0; i < MaxPendingOperations; i++ {
856 op := operationIDAt(i)
857 wg.Add(1)
858 go func(op ports.BrokerOperationID) {
859 defer wg.Done()
860 if err := tr.Admit(scope, op); err != nil {
861 errs <- err
862 }
863 }(op)
864 }
865 wg.Wait()
866 require.Equal(t, MaxPendingOperations, tr.Pending())
867 for i := 0; i < MaxPendingOperations; i++ {
868 op := operationIDAt(i)
869 wg.Add(1)
870 go func(op ports.BrokerOperationID) {
871 defer wg.Done()
872 if err := tr.Complete(scope, op, ports.BrokerOutcomeOK); err != nil {
873 errs <- err
874 }
875 }(op)
876 }
877 wg.Wait()
878 close(errs)
879 for err := range errs {
880 require.NoError(t, err)
881 }
882 require.Zero(t, tr.Pending())
883 require.Equal(t, MaxPendingOperations, tr.Tracked())
884 require.Equal(t, MaxPendingOperations*operationIDBytes, tr.DedupBytes())
885}
886
887// TestStreamTrackerConcurrentFrames is a deterministic race test: one stream
888// per goroutine progresses and retires concurrently.
889func TestStreamTrackerConcurrentFrames(t *testing.T) {
890 scope := testBrokerScope()
891 tr, err := newStreamTracker(scope)
892 require.NoError(t, err)
893 for i := 1; i <= MaxStreams; i++ {
894 require.NoError(t, tr.Open(scope, ports.BrokerStreamID(i)))
895 }
896 var wg sync.WaitGroup
897 errs := make(chan error, MaxStreams)
898 for i := 1; i <= MaxStreams; i++ {
899 stream := ports.BrokerStreamID(i)
900 wg.Add(1)
901 go func(stream ports.BrokerStreamID) {
902 defer wg.Done()
903 if _, err := tr.Progress(scope, stream); err != nil {
904 errs <- err
905 }
906 if err := tr.Opened(scope, stream); err != nil {
907 errs <- err
908 }
909 if disposition, err := tr.Data(scope, stream); err != nil || disposition != StreamAccepted {
910 errs <- err
911 }
912 if disposition, err := tr.Close(scope, stream); err != nil || disposition != StreamAccepted {
913 errs <- err
914 }
915 }(stream)
916 }
917 wg.Wait()
918 close(errs)
919 for err := range errs {
920 require.NoError(t, err)
921 }
922 require.Zero(t, tr.Active())
923}
924
925// TestConnectionConcurrentWork is a deterministic race test: the connection
926// publishes sequential revisions while readers observe it and operations
927// complete concurrently.
928func TestConnectionConcurrentWork(t *testing.T) {
929 conn := readyTestConnection(t)
930 require.NoError(t, conn.Subscribe(1))
931 hosts := sampleSnapshotDaemons()
932 const revisions = 20
933 const operations = 32
934 for i := 0; i < operations; i++ {
935 require.NoError(t, conn.AdmitOperation(operationIDAt(i)))
936 }
937
938 var wg sync.WaitGroup
939 done := make(chan struct{})
940 errs := make(chan error, operations)
941 for i := 0; i < 4; i++ {
942 wg.Add(1)
943 go func() {
944 defer wg.Done()
945 for {
946 select {
947 case <-done:
948 return
949 default:
950 }
951 assert.Equal(t, ConnReady, conn.State())
952 assert.Equal(t, testBrokerScope(), conn.Scope())
953 if snapshot, ok := conn.Snapshot(); ok {
954 assert.NotZero(t, snapshot.Revision)
955 }
956 }
957 }()
958 }
959 for i := 0; i < operations; i++ {
960 op := operationIDAt(i)
961 wg.Add(1)
962 go func(op ports.BrokerOperationID) {
963 defer wg.Done()
964 if err := conn.CompleteOperation(op, ports.BrokerOutcomeOK); err != nil {
965 errs <- err
966 }
967 }(op)
968 }
969 for revision := 1; revision <= revisions; revision++ {
970 for _, part := range snapshotPartsFor(1, ports.BrokerRevision(revision), hosts, nil) {
971 _, _, err := conn.AddSnapshotPart(part)
972 require.NoError(t, err)
973 }
974 }
975 close(done)
976 wg.Wait()
977 close(errs)
978 for err := range errs {
979 require.NoError(t, err)
980 }
981 require.Zero(t, connectionOperations(conn).Pending())
982 require.Equal(t, operations, connectionOperations(conn).Tracked())
983 snapshot, ok := conn.Snapshot()
984 require.True(t, ok)
985 require.Equal(t, ports.BrokerRevision(revisions), snapshot.Revision)
986}
987
988// TestConnectionCloseRaceNoPostCloseAdmission proves the connection lock
989// covers both the ready check and the tracker mutation: whenever a close and
990// an admit/open race, the close wins the state transition and leaves no
991// admitted operation or opened stream behind, and every attempt after close
992// is refused.
993func TestConnectionCloseRaceNoPostCloseAdmission(t *testing.T) {
994 t.Run("admit", func(t *testing.T) {
995 for i := 0; i < 500; i++ {
996 conn := readyTestConnection(t)
997 op := operationIDAt(1)
998 var wg sync.WaitGroup
999 start := make(chan struct{})
1000 var admitErr error
1001 wg.Add(2)
1002 go func() { defer wg.Done(); <-start; admitErr = conn.AdmitOperation(op) }()
1003 go func() { defer wg.Done(); <-start; _ = conn.BeginClose(); _ = conn.FinishClose() }()
1004 close(start)
1005 wg.Wait()
1006 require.Equal(t, ConnClosed, conn.State(), "iteration %d", i)
1007 require.Zero(t, connectionOperations(conn).Pending(), "iteration %d: no admitted operation survives close", i)
1008 if admitErr != nil {
1009 require.ErrorIs(t, admitErr, ErrConnectionClosed, "iteration %d", i)
1010 }
1011 // The closed connection admits nothing, and a refused attempt
1012 // mutates nothing.
1013 require.ErrorIs(t, conn.AdmitOperation(op), ErrConnectionClosed)
1014 require.Zero(t, connectionOperations(conn).Pending())
1015 }
1016 })
1017
1018 t.Run("open", func(t *testing.T) {
1019 for i := 0; i < 500; i++ {
1020 conn := readyTestConnection(t)
1021 var wg sync.WaitGroup
1022 start := make(chan struct{})
1023 var openErr error
1024 wg.Add(2)
1025 go func() { defer wg.Done(); <-start; openErr = conn.OpenStream(1) }()
1026 go func() { defer wg.Done(); <-start; _ = conn.BeginClose(); _ = conn.FinishClose() }()
1027 close(start)
1028 wg.Wait()
1029 require.Equal(t, ConnClosed, conn.State(), "iteration %d", i)
1030 require.Zero(t, connectionStreams(conn).Active(), "iteration %d: no opened stream survives close", i)
1031 if openErr != nil {
1032 require.ErrorIs(t, openErr, ErrConnectionClosed, "iteration %d", i)
1033 }
1034 require.ErrorIs(t, conn.OpenStream(1), ErrConnectionClosed)
1035 require.Zero(t, connectionStreams(conn).Active())
1036 }
1037 })
1038
1039 t.Run("burst", func(t *testing.T) {
1040 conn := readyTestConnection(t)
1041 const attempts = MaxPendingOperations
1042 var wg sync.WaitGroup
1043 start := make(chan struct{})
1044 errs := make([]error, attempts)
1045 wg.Add(attempts + 1)
1046 for i := 0; i < attempts; i++ {
1047 go func(i int) {
1048 defer wg.Done()
1049 <-start
1050 errs[i] = conn.AdmitOperation(operationIDAt(i))
1051 }(i)
1052 }
1053 go func() {
1054 defer wg.Done()
1055 <-start
1056 _ = conn.BeginClose()
1057 _ = conn.FinishClose()
1058 }()
1059 close(start)
1060 wg.Wait()
1061 require.Equal(t, ConnClosed, conn.State())
1062 require.Zero(t, connectionOperations(conn).Pending(), "no admitted operation survives the racing close")
1063 for i, err := range errs {
1064 if err != nil {
1065 require.ErrorIs(t, err, ErrConnectionClosed, "attempt %d", i)
1066 }
1067 }
1068 require.ErrorIs(t, conn.AdmitOperation(operationIDAt(attempts+1)), ErrConnectionClosed)
1069 })
1070}
1071
1072// TestConnectionCloseRaceNoPostCloseSnapshotCommit proves the connection lock
1073// covers both the ready check and the assembler mutation: whenever a close and
1074// an Add race, no snapshot can commit after BeginClose returned, no part can be
1075// accepted after the close fence, and an Add that wins the race holds the lock
1076// across its commit, so its revision is already visible when BeginClose
1077// returns.
1078func TestConnectionCloseRaceNoPostCloseSnapshotCommit(t *testing.T) {
1079 const revision = 6
1080 const attempts = 500
1081
1082 t.Run("staged end", func(t *testing.T) {
1083 for i := 0; i < attempts; i++ {
1084 conn := readyTestConnection(t)
1085 require.NoError(t, conn.Subscribe(2))
1086 parts := snapshotPartsFor(2, revision, sampleSnapshotDaemons(), nil)
1087 end := parts[len(parts)-1]
1088 for _, part := range parts[:len(parts)-1] {
1089 stageConnectionPart(t, conn, part)
1090 }
1091 require.True(t, conn.StagingActive())
1092 _, committedBefore := conn.Snapshot()
1093 require.False(t, committedBefore, "iteration %d: nothing publishes before End", i)
1094
1095 var completed bool
1096 var addedRevision ports.BrokerRevision
1097 committedAtClose, closeRevision, errs := raceConnectionClose(t, conn, func() error {
1098 done, published, err := conn.AddSnapshotPart(end)
1099 if done {
1100 completed = true
1101 addedRevision = published.Revision
1102 }
1103 return err
1104 })
1105 addErr := errs[0]
1106
1107 if completed {
1108 // A committed End is already visible when BeginClose returned;
1109 // it never lands in the closing window.
1110 require.NoError(t, addErr, "iteration %d", i)
1111 require.True(t, committedAtClose, "iteration %d: commit landed after close", i)
1112 require.Equal(t, ports.BrokerRevision(revision), addedRevision, "iteration %d", i)
1113 require.Equal(t, addedRevision, closeRevision, "iteration %d: commit changed after close", i)
1114 } else {
1115 require.ErrorIs(t, addErr, ErrConnectionClosed, "iteration %d", i)
1116 require.False(t, committedAtClose, "iteration %d: refused part must not publish", i)
1117 }
1118 requireClosedConnectionRejects(t, conn, i, end)
1119 }
1120 })
1121
1122 t.Run("empty transfer", func(t *testing.T) {
1123 // An empty transfer's Begin stages and its End publishes. Racing the
1124 // close against both parts proves no commit can land after BeginClose
1125 // returned, whatever order the two Adds run in: an End that runs before
1126 // its Begin is refused as an invalid transfer, and a part that runs
1127 // after the close is refused as closed.
1128 for i := 0; i < attempts; i++ {
1129 conn := readyTestConnection(t)
1130 require.NoError(t, conn.Subscribe(2))
1131 parts := snapshotPartsFor(2, revision, nil, nil)
1132 begin, end := parts[0], parts[1]
1133
1134 var completed bool
1135 var addedRevision ports.BrokerRevision
1136 committedAtClose, _, errs := raceConnectionClose(t, conn,
1137 func() error {
1138 _, _, err := conn.AddSnapshotPart(begin)
1139 return err
1140 },
1141 func() error {
1142 done, published, err := conn.AddSnapshotPart(end)
1143 if done {
1144 completed = true
1145 addedRevision = published.Revision
1146 }
1147 return err
1148 },
1149 )
1150 beginErr, endErr := errs[0], errs[1]
1151
1152 if beginErr != nil {
1153 require.ErrorIs(t, beginErr, ErrConnectionClosed, "iteration %d", i)
1154 }
1155 if completed {
1156 require.NoError(t, endErr, "iteration %d", i)
1157 require.True(t, committedAtClose, "iteration %d: commit landed after close", i)
1158 require.Equal(t, ports.BrokerRevision(revision), addedRevision, "iteration %d", i)
1159 } else {
1160 require.Error(t, endErr, "iteration %d", i)
1161 require.False(t, committedAtClose, "iteration %d: refused transfer must not publish", i)
1162 }
1163 requireClosedConnectionRejects(t, conn, i, end)
1164 }
1165 })
1166}
1167
1168// raceConnectionClose runs every supplied Add concurrently with one full close
1169// and reports whether a snapshot was committed at the moment BeginClose
1170// returned, that revision, and each Add's error in argument order.
1171func raceConnectionClose(t *testing.T, conn *Connection, adds ...func() error) (bool, ports.BrokerRevision, []error) {
1172 t.Helper()
1173 errs := make([]error, len(adds))
1174 var committedAtClose bool
1175 var closeRevision ports.BrokerRevision
1176 var wg sync.WaitGroup
1177 start := make(chan struct{})
1178 wg.Add(len(adds) + 1)
1179 for i, add := range adds {
1180 go func(i int, add func() error) {
1181 defer wg.Done()
1182 <-start
1183 errs[i] = add()
1184 }(i, add)
1185 }
1186 go func() {
1187 defer wg.Done()
1188 <-start
1189 _ = conn.BeginClose()
1190 published, ok := conn.Snapshot()
1191 committedAtClose = ok
1192 closeRevision = published.Revision
1193 _ = conn.FinishClose()
1194 }()
1195 close(start)
1196 wg.Wait()
1197 return committedAtClose, closeRevision, errs
1198}
1199
1200// requireClosedConnectionRejects proves a closed connection retains no snapshot
1201// and accepts no further part.
1202func requireClosedConnectionRejects(t *testing.T, conn *Connection, iteration int, part SnapshotPart) {
1203 t.Helper()
1204 require.Equal(t, ConnClosed, conn.State(), "iteration %d", iteration)
1205 require.False(t, conn.StagingActive(), "iteration %d", iteration)
1206 _, ok := conn.Snapshot()
1207 require.False(t, ok, "iteration %d: closed releases the snapshot", iteration)
1208 _, _, err := conn.AddSnapshotPart(part)
1209 require.ErrorIs(t, err, ErrConnectionClosed, "iteration %d", iteration)
1210}
1211
1212// TestConnectionNilSafety proves the nil-receiver paths fail closed.
1213func TestConnectionNilSafety(t *testing.T) {
1214 var conn *Connection
1215 require.Equal(t, ConnClosed, conn.State())
1216 require.False(t, conn.Ready())
1217 require.Equal(t, Scope{}, conn.Scope())
1218 require.ErrorIs(t, conn.SignalPreamble(), ErrConnectionClosed)
1219 require.ErrorIs(t, conn.Register(), ErrConnectionClosed)
1220 require.ErrorIs(t, conn.Subscribe(1), ErrConnectionClosed)
1221 require.ErrorIs(t, conn.Resync(1), ErrConnectionClosed)
1222 require.ErrorIs(t, conn.Unsubscribe(1), ErrConnectionClosed)
1223 require.ErrorIs(t, conn.AdmitOperation(operationIDAt(1)), ErrConnectionClosed)
1224 require.ErrorIs(t, conn.CompleteOperation(operationIDAt(1), ports.BrokerOutcomeOK), ErrConnectionClosed)
1225 require.ErrorIs(t, conn.OpenStream(1), ErrConnectionClosed)
1226 _, err := conn.StreamData(1)
1227 require.ErrorIs(t, err, ErrConnectionClosed)
1228 _, err = conn.StreamProgress(1)
1229 require.ErrorIs(t, err, ErrConnectionClosed)
1230 _, err = conn.CloseStream(1)
1231 require.ErrorIs(t, err, ErrConnectionClosed)
1232 _, _, err = conn.AddSnapshotPart(snapshotPartsFor(1, 1, nil, nil)[0])
1233 require.ErrorIs(t, err, ErrConnectionClosed)
1234 _, ok := conn.Snapshot()
1235 require.False(t, ok)
1236 require.NoError(t, conn.BeginClose())
1237 require.NoError(t, conn.FinishClose())
1238}
1239
1240// TestOperationTrackerNilSafety and TestStreamTrackerNilSafety above cover the
1241// tracker-receiver paths; this proves the nil tracker pointer paths.
1242func TestNilOperationTracker(t *testing.T) {
1243 var tr *OperationTracker
1244 scope := testBrokerScope()
1245 require.ErrorIs(t, tr.Admit(scope, operationIDAt(1)), ErrConnectionClosed)
1246 require.ErrorIs(t, tr.Complete(scope, operationIDAt(1), ports.BrokerOutcomeOK), ErrConnectionClosed)
1247 require.Zero(t, tr.Pending())
1248 require.Zero(t, tr.Tracked())
1249 require.Zero(t, tr.DedupBytes())
1250 _, ok := tr.Outcome(operationIDAt(1))
1251 require.False(t, ok)
1252}