1package daemonmux
2
3import (
4 "sync"
5 "testing"
6 "time"
7
8 "github.com/stretchr/testify/require"
9
10 "github.com/bnema/vev/internal/ports"
11)
12
13// aggregateSink is the test's daemon-style consumer: it takes typed connections
14// from the aggregate until Accept fails and serves each one with servePongs.
15type aggregateSink struct {
16 mu sync.Mutex
17 conns []ports.ServerConnection
18 errs []error
19}
20
21func startAggregateSink(a *AggregateListener) *aggregateSink {
22 sink := &aggregateSink{}
23 go func() {
24 for {
25 connection, err := a.Accept()
26 if err != nil {
27 sink.mu.Lock()
28 sink.errs = append(sink.errs, err)
29 sink.mu.Unlock()
30 return
31 }
32 sink.mu.Lock()
33 sink.conns = append(sink.conns, connection)
34 sink.mu.Unlock()
35 go func() { _ = servePongs(connection) }()
36 }
37 }()
38 return sink
39}
40
41func (s *aggregateSink) await(t *testing.T, count int) []ports.ServerConnection {
42 t.Helper()
43 require.Eventually(t, func() bool {
44 s.mu.Lock()
45 defer s.mu.Unlock()
46 return len(s.conns) >= count
47 }, listenerTestDeadline, time.Millisecond, "aggregate accepted %d of %d connections", count, count)
48 s.mu.Lock()
49 defer s.mu.Unlock()
50 out := make([]ports.ServerConnection, len(s.conns))
51 copy(out, s.conns)
52 return out
53}
54
55func (s *aggregateSink) errors() []error {
56 s.mu.Lock()
57 defer s.mu.Unlock()
58 return append([]error(nil), s.errs...)
59}
60
61// aggregateChildCount reports the children the aggregate still forwards from.
62func aggregateChildCount(a *AggregateListener) int {
63 a.mu.Lock()
64 defer a.mu.Unlock()
65 return len(a.children)
66}
67
68// aggregateQueued reports the accepted-but-undelivered connections the
69// aggregate holds.
70func aggregateQueued(a *AggregateListener) int {
71 a.mu.Lock()
72 defer a.mu.Unlock()
73 return len(a.queue)
74}
75
76// TestNewAggregateListenerValidation covers the nil-receiver and registration
77// guards without any physical listener.
78func TestNewAggregateListenerValidation(t *testing.T) {
79 var nilAggregate *AggregateListener
80 _, err := nilAggregate.Accept()
81 require.ErrorIs(t, err, ErrAggregateConfig)
82 require.ErrorIs(t, nilAggregate.Register(nil), ErrAggregateConfig)
83 require.ErrorIs(t, nilAggregate.Remove(nil), ErrAggregateConfig)
84 require.NoError(t, nilAggregate.Close())
85
86 a := newAggregateListener(MaxAggregateQueue)
87 require.Equal(t, "daemonmux", a.Addr())
88 require.Equal(t, MaxAggregateQueue, a.limit)
89 require.ErrorIs(t, a.Register(nil), ErrAggregateConfig)
90
91 first := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
92 second := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
93 require.NoError(t, a.Register(first.listener))
94 require.ErrorIs(t, a.Register(first.listener), ErrAggregateChildTaken)
95 require.NoError(t, a.Close())
96 require.ErrorIs(t, a.Register(second.listener), ErrAggregateClosed)
97 _, err = a.Accept()
98 require.ErrorIs(t, err, ErrAggregateClosed)
99}
100
101// TestAggregateTwoPhysicalChildrenAdmitIndependently proves the aggregate
102// delivers both children's independently accepted typed connections through one
103// ports.ServerListener, and that one child's stream lifecycle never touches the
104// other child or its physical carriage.
105func TestAggregateTwoPhysicalChildrenAdmitIndependently(t *testing.T) {
106 a := newAggregateListener(MaxAggregateQueue)
107 t.Cleanup(func() { _ = a.Close() })
108
109 first := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
110 second := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
111 require.NoError(t, a.Register(first.listener))
112 require.NoError(t, a.Register(second.listener))
113
114 sink := startAggregateSink(a)
115 firstBroker := first.mustOpen(1)
116 secondBroker := second.mustOpen(1)
117 accepted := sink.await(t, 2)
118
119 // Each physical child contributed exactly one typed daemonmux connection.
120 pumps := map[*Pump]bool{}
121 for _, connection := range accepted {
122 listenerConnection, ok := connection.(*ListenerConnection)
123 require.True(t, ok, "the aggregate must forward the child's typed daemonmux connection")
124 pumps[listenerConnection.pump] = true
125 }
126 require.True(t, pumps[first.daemonPump], "the first child never contributed a connection")
127 require.True(t, pumps[second.daemonPump], "the second child never contributed a connection")
128 require.Len(t, pumps, 2)
129
130 listenerRoundTrip(t, firstBroker)
131 listenerRoundTrip(t, secondBroker)
132
133 // One stream ending leaves its sibling and both physical carriages live.
134 require.NoError(t, firstBroker.Close())
135 require.True(t, channelClosed(firstBroker.Done()))
136 listenerRoundTrip(t, secondBroker)
137 require.False(t, channelClosed(first.daemonPump.Done()))
138 require.False(t, channelClosed(second.daemonPump.Done()))
139 require.Empty(t, sink.errors())
140}
141
142// TestAggregateChildLossDoesNotStopAccept proves a physical loss of one child
143// is isolated: the aggregate deregisters it without a terminal error, keeps
144// accepting from the surviving child, and accepts from a child registered after
145// the loss.
146func TestAggregateChildLossDoesNotStopAccept(t *testing.T) {
147 a := newAggregateListener(MaxAggregateQueue)
148 t.Cleanup(func() { _ = a.Close() })
149
150 first := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
151 second := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
152 require.NoError(t, a.Register(first.listener))
153 require.NoError(t, a.Register(second.listener))
154
155 sink := startAggregateSink(a)
156 firstBroker := first.mustOpen(1)
157 secondBroker := second.mustOpen(1)
158 sink.await(t, 2)
159 listenerRoundTrip(t, firstBroker)
160 listenerRoundTrip(t, secondBroker)
161
162 // Failing the first physical carriage makes its child's Accept terminal.
163 require.NoError(t, first.brokerPump.Close())
164 require.Eventually(t, func() bool { return channelClosed(first.daemonPump.Done()) }, listenerTestDeadline, time.Millisecond)
165
166 // The aggregate isolates the loss: the child is deregistered and the loss
167 // is never surfaced to the shared accept loop.
168 require.Eventually(t, func() bool { return aggregateChildCount(a) == 1 }, listenerTestDeadline, time.Millisecond,
169 "the lost child was not deregistered")
170 require.Empty(t, sink.errors(), "a child loss must not surface as an aggregate Accept error")
171
172 // The surviving physical child still admits.
173 survivorBroker := second.mustOpen(2)
174 survivorDaemon := sink.await(t, 3)[2].(*ListenerConnection)
175 require.Equal(t, second.daemonPump, survivorDaemon.pump)
176 listenerRoundTrip(t, survivorBroker)
177
178 // A child registered after the loss also admits.
179 third := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
180 require.NoError(t, a.Register(third.listener))
181 thirdBroker := third.mustOpen(1)
182 thirdDaemon := sink.await(t, 4)[3].(*ListenerConnection)
183 require.Equal(t, third.daemonPump, thirdDaemon.pump)
184 listenerRoundTrip(t, thirdBroker)
185 require.Empty(t, sink.errors())
186}
187
188// TestAggregateCloseDrainsQueuedConnections proves Close unblocks Accept,
189// closes the owned child, joins the forwarder, and cleans every connection it
190// was holding: the queued ones and the single in-flight one blocked on the
191// bounded queue.
192func TestAggregateCloseDrainsQueuedConnections(t *testing.T) {
193 const limit = 2
194 a := newAggregateListener(limit)
195 child := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
196 require.NoError(t, a.Register(child.listener))
197
198 // Three admitted connections, none accepted: the aggregate holds its
199 // bounded two, and the forwarder holds the third while backpressured.
200 first := child.mustOpen(1)
201 second := child.mustOpen(2)
202 third := child.mustOpen(3)
203 require.Eventually(t, func() bool { return aggregateQueued(a) == limit }, listenerTestDeadline, time.Millisecond,
204 "the aggregate queue never filled to its bound")
205
206 blocked := make(chan error, 1)
207 go func() {
208 _, err := a.Accept()
209 blocked <- err
210 }()
211
212 require.NoError(t, a.Close())
213 require.NoError(t, a.Close(), "Close is idempotent")
214
215 select {
216 case err := <-blocked:
217 require.ErrorIs(t, err, ErrAggregateClosed)
218 case <-time.After(listenerTestDeadline):
219 t.Fatal("Close did not unblock Accept")
220 }
221 _, err := a.Accept()
222 require.ErrorIs(t, err, ErrAggregateClosed)
223 require.Zero(t, aggregateQueued(a), "Close must clear the aggregate queue")
224 require.Zero(t, aggregateChildCount(a), "Close must drop every owned child")
225
226 // Every connection the aggregate held is released, including the one its
227 // forwarder was holding while backpressured; the child itself is closed.
228 for i, broker := range []typedLogical{first, second, third} {
229 require.Eventually(t, func() bool { return channelClosed(broker.Done()) }, listenerTestDeadline, time.Millisecond,
230 "queued connection %d was not cleaned by Close", i+1)
231 }
232 _, err = child.listener.Accept()
233 require.ErrorIs(t, err, ErrListenerClosed)
234}
235
236// TestAggregateRemoveStopsChildOnly proves Remove closes and joins exactly the
237// removed child, leaves the aggregate accepting from a sibling, and reports an
238// unknown child afterwards.
239func TestAggregateRemoveStopsChildOnly(t *testing.T) {
240 a := newAggregateListener(MaxAggregateQueue)
241 t.Cleanup(func() { _ = a.Close() })
242
243 first := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
244 second := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
245 require.NoError(t, a.Register(first.listener))
246 require.NoError(t, a.Register(second.listener))
247 sink := startAggregateSink(a)
248
249 require.NoError(t, a.Remove(first.listener))
250 require.Equal(t, 1, aggregateChildCount(a))
251 _, err := first.listener.Accept()
252 require.ErrorIs(t, err, ErrListenerClosed)
253 require.ErrorIs(t, a.Remove(first.listener), ErrAggregateUnknownChild)
254
255 broker := second.mustOpen(1)
256 sink.await(t, 1)
257 listenerRoundTrip(t, broker)
258 require.Empty(t, sink.errors())
259}
260
261// TestAggregateRemoveUnblocksBackpressuredForwarder proves Remove does not
262// deadlock when the removed child's forwarder is waiting for queue space: it
263// releases the in-flight connection, while a connection already queued stays
264// deliverable to the daemon.
265func TestAggregateRemoveUnblocksBackpressuredForwarder(t *testing.T) {
266 a := newAggregateListener(1)
267 t.Cleanup(func() { _ = a.Close() })
268 child := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
269 require.NoError(t, a.Register(child.listener))
270
271 queuedBroker := child.mustOpen(1)
272 heldBroker := child.mustOpen(2)
273 require.Eventually(t, func() bool { return aggregateQueued(a) == 1 }, listenerTestDeadline, time.Millisecond,
274 "the aggregate queue never filled to its bound")
275
276 require.NoError(t, a.Remove(child.listener))
277 require.Zero(t, aggregateChildCount(a))
278
279 // The connection the forwarder was holding is released; the connection
280 // already queued from the removed child stays deliverable.
281 require.Eventually(t, func() bool { return channelClosed(heldBroker.Done()) }, listenerTestDeadline, time.Millisecond,
282 "the backpressured in-flight connection was not released")
283 queued, err := a.Accept()
284 require.NoError(t, err)
285 queuedConnection, ok := queued.(*ListenerConnection)
286 require.True(t, ok, "the aggregate must forward the child's typed daemonmux connection")
287 require.False(t, channelClosed(queuedConnection.Done()), "a queued connection from a removed child must stay deliverable")
288 require.NoError(t, queued.Close())
289 require.True(t, channelClosed(queuedConnection.Done()))
290 require.Eventually(t, func() bool { return channelClosed(queuedBroker.Done()) }, listenerTestDeadline, time.Millisecond)
291}