1package ports
2
3import (
4 "sync"
5 "testing"
6
7 "github.com/stretchr/testify/require"
8)
9
10// TestBrokerStreamWindowAdmission pins the bounded anti-replay window: a
11// concurrent lower identity that was allocated first is admitted after a higher
12// one, a duplicate inside the window is refused, an identity evicted past the
13// bound is stale, and zero is never a valid identity.
14func TestBrokerStreamWindowAdmission(t *testing.T) {
15 for _, tc := range []struct {
16 name string
17 admit []BrokerStreamID
18 wantErr []error
19 }{
20 {
21 name: "concurrent 2 then 1",
22 admit: []BrokerStreamID{2, 1},
23 wantErr: []error{nil, nil},
24 },
25 {
26 name: "duplicate refused",
27 admit: []BrokerStreamID{2, 2},
28 wantErr: []error{nil, BrokerAdmissionStale},
29 },
30 {
31 name: "abandoned identity stays consumable",
32 admit: []BrokerStreamID{3, 1, 2},
33 wantErr: []error{nil, nil, nil},
34 },
35 {
36 name: "identity inside the window after a high one",
37 admit: []BrokerStreamID{BrokerStreamWindowSize, 1},
38 wantErr: []error{nil, nil},
39 },
40 {
41 name: "identity evicted past the window is stale",
42 admit: []BrokerStreamID{1, BrokerStreamWindowSize + 1, 1},
43 wantErr: []error{nil, nil, BrokerAdmissionStale},
44 },
45 {
46 name: "boundary identity still admitted",
47 admit: []BrokerStreamID{1, BrokerStreamWindowSize, 2},
48 wantErr: []error{nil, nil, nil},
49 },
50 {
51 name: "zero identity refused",
52 admit: []BrokerStreamID{0},
53 wantErr: []error{BrokerAdmissionInvalid},
54 },
55 } {
56 t.Run(tc.name, func(t *testing.T) {
57 var window BrokerStreamWindow
58 for i, stream := range tc.admit {
59 require.ErrorIs(t, window.Admit(stream), tc.wantErr[i], "admission %d of %d", i, stream)
60 }
61 })
62 }
63}
64
65// TestBrokerStreamWindowConcurrentMonotone proves concurrent allocations admit
66// every distinct identity exactly once and never collide, under the race
67// detector.
68func TestBrokerStreamWindowConcurrentMonotone(t *testing.T) {
69 var window BrokerStreamWindow
70 const count = 512
71 var wg sync.WaitGroup
72 errs := make([]error, count)
73 for i := range count {
74 wg.Add(1)
75 go func() {
76 defer wg.Done()
77 errs[i] = window.Admit(BrokerStreamID(i + 1))
78 }()
79 }
80 wg.Wait()
81 for i, err := range errs {
82 require.NoError(t, err, "identity %d", i+1)
83 }
84 // Replaying one admitted identity is refused.
85 require.ErrorIs(t, window.Admit(1), BrokerAdmissionStale)
86}