1package daemonmux
2
3import (
4 "context"
5 "errors"
6 "testing"
7 "time"
8
9 "github.com/stretchr/testify/require"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13)
14
15// TestPumpApplyOpenDuplicateIsolation proves a duplicate or replayed inbound
16// Open never disturbs the stream it names and never produces an outbound frame:
17// a live stream keeps its state, its watch, and its queued outbound work, a
18// retired stream stays terminal, and a replayed identity below the high-water
19// mark stays unrecorded.
20func TestPumpApplyOpenDuplicateIsolation(t *testing.T) {
21 t.Run("live stream", func(t *testing.T) {
22 pump, carrier := newTestPump(t, DirectionClient)
23 require.NoError(t, pump.Engine().Open(openFor(1)))
24 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
25 // A queued outbound frame proves the duplicate Open neither resets the
26 // engine record nor discards the scheduler's queued work.
27 require.NoError(t, pump.SendData(1, []byte("out"), nil))
28 before := mustStatus(t, pump.Engine(), 1)
29 watch := pump.Watch(1)
30
31 require.NoError(t, pump.applyOpen(openFor(1)))
32
33 after := mustStatus(t, pump.Engine(), 1)
34 require.Equal(t, before, after, "a duplicate open left the live stream untouched")
35 require.Equal(t, StreamOpen, after.State)
36 require.Equal(t, 1, pump.Engine().Live())
37 require.False(t, channelClosed(after.Done))
38 require.Equal(t, watch, pump.Watch(1), "a duplicate open never rotated the watch channel")
39 require.Equal(t, 1, pump.scheduler.QueuedFrames(1), "the queued outbound frame survived")
40 carrier.requireNoSent(t, 50*time.Millisecond)
41
42 pump.Start(context.Background())
43 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
44 require.NoError(t, err)
45 require.Equal(t, Data{Physical: 1, Data: []byte("out")}, message)
46 carrier.requireNoSent(t, 50*time.Millisecond)
47 })
48
49 t.Run("retired stream", func(t *testing.T) {
50 pump, carrier := newTestPump(t, DirectionClient)
51 require.NoError(t, pump.Engine().Open(openFor(1)))
52 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
53 disposition, err := pump.Engine().Close(Close{Physical: 1})
54 require.NoError(t, err)
55 require.Equal(t, StreamAccepted, disposition)
56 require.Equal(t, StreamTerminal, mustStatus(t, pump.Engine(), 1).State)
57
58 require.NoError(t, pump.applyOpen(openFor(1)))
59
60 status := mustStatus(t, pump.Engine(), 1)
61 require.Equal(t, StreamTerminal, status.State, "a replayed open never revives a retired stream")
62 require.NoError(t, status.Err)
63 require.Zero(t, pump.Engine().Live())
64 carrier.requireNoSent(t, 50*time.Millisecond)
65 })
66
67 t.Run("replayed identity below the high-water mark", func(t *testing.T) {
68 pump, carrier := newTestPump(t, DirectionClient)
69 require.NoError(t, pump.Engine().Open(openFor(3)))
70 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(3)}))
71
72 // Identity 1 is at or below the high-water mark but was never admitted:
73 // it can only be a replay, so it is ignored instead of refused.
74 require.NoError(t, pump.applyOpen(openFor(1)))
75 _, tracked := pump.Engine().Status(1)
76 require.False(t, tracked, "a replayed identity is never recorded")
77 require.Equal(t, 1, pump.Engine().Live())
78 carrier.requireNoSent(t, 50*time.Millisecond)
79 })
80}
81
82// TestPumpFreshOpenRefusedSendsTypedRefused proves a fresh Open the engine
83// refused for capacity is signalled with exactly one typed Mux Refused carrying
84// its admission code, records no stream, and leaves the live sibling and the
85// physical connection untouched.
86func TestPumpFreshOpenRefusedSendsTypedRefused(t *testing.T) {
87 ceilings := DefaultMuxCeilings()
88 ceilings.MaxStreams = 1
89 pump, carrier := newTestPumpWithCeilings(t, DirectionClient, ceilings)
90 pump.Start(context.Background())
91 require.NoError(t, pump.Engine().Open(openFor(1)))
92 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
93 require.Equal(t, 1, pump.Engine().Live())
94
95 // The fresh identity is refused by the stream ceiling before it is recorded.
96 deliverClientFrame(t, carrier, openFor(2))
97
98 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
99 require.NoError(t, err)
100 refused, ok := message.(Refused)
101 require.True(t, ok, "a fresh refusal is a Mux Refused")
102 require.Equal(t, testRef(2), refused.Ref)
103 require.Equal(t, ports.BrokerErrorUnavailable, refused.Error.Code)
104 require.Equal(t, uint32(1), refused.Error.AdmissionCode, "capacity refusal carries the limit code")
105 require.Equal(t, domain.RemoteFailureNone, refused.Error.FailureKind)
106 require.NotEmpty(t, refused.Error.Text)
107 require.LessOrEqual(t, len(refused.Error.Text), ports.BrokerMaxErrorBytes)
108 require.NoError(t, refused.Error.validate())
109
110 // No stream was recorded for the refused identity, and the sibling and the
111 // physical connection are untouched.
112 _, tracked := pump.Engine().Status(2)
113 require.False(t, tracked, "a fresh refusal records no stream")
114 require.Equal(t, 1, pump.Engine().Live())
115 require.False(t, channelClosed(mustStatus(t, pump.Engine(), 1).Done))
116 require.False(t, pump.Engine().Closed())
117 require.False(t, channelClosed(pump.Done()))
118 carrier.requireNoSent(t, 50*time.Millisecond)
119}
120
121// TestOpenRefusalDetail proves the closed admission taxonomy the broker
122// classifies without matching text: capacity is a limit, a closed physical
123// connection is closed, and every other refusal is an invalid request.
124func TestOpenRefusalDetail(t *testing.T) {
125 tests := []struct {
126 name string
127 cause error
128 code uint32
129 text string
130 }{
131 {"capacity", ErrTooManyStreams, 1, "stream limit reached"},
132 {"closed", ErrPhysicalClosed, 2, "physical connection closed"},
133 {"invalid authority", ErrInvalidStream, 4, "invalid open request"},
134 {"unknown refusal", errors.New("boom"), 4, "invalid open request"},
135 }
136 for _, tt := range tests {
137 t.Run(tt.name, func(t *testing.T) {
138 detail := openRefusalDetail(tt.cause)
139 require.Equal(t, tt.code, detail.AdmissionCode)
140 require.Equal(t, tt.text, detail.Text)
141 require.Equal(t, ports.BrokerErrorUnavailable, detail.Code)
142 require.Equal(t, domain.RemoteFailureNone, detail.FailureKind)
143 require.NoError(t, detail.validate())
144 })
145 }
146}