broker_stream_window_test.go RawHistoryFolder

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}