1package daemonmux
2
3import (
4 "context"
5 "encoding/binary"
6 "errors"
7 "net"
8 "runtime"
9 "sync"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "github.com/stretchr/testify/require"
15
16 "github.com/bnema/vev/internal/adapters/ipc"
17 "github.com/bnema/vev/internal/adapters/streamframe"
18 "github.com/bnema/vev/internal/protocol/wire"
19)
20
21// errFakeRawClosed is the synthetic raw-carriage error a fakeRawTransport
22// returns once Close released its blocked I/O, mirroring the adapter
23// prompt-close contract.
24var errFakeRawClosed = errors.New("fakeRawTransport: closed")
25
26// fakeRawTransport is one end of a scripted raw framed carriage. RecvBounded
27// records every caller limit and refuses a body above it exactly like a real
28// streamframe framer, so a bridge test observes the bound the bridge applied.
29type fakeRawTransport struct {
30 mu sync.Mutex
31 limits []uint64
32 sends [][]byte
33 inbound chan []byte
34 closedCh chan struct{}
35 closeOnce sync.Once
36 closes atomic.Int64
37 closeErr error
38 // sendHold, when non-nil, blocks Send until released or the carriage
39 // closes, so a test can observe a synchronous send in flight.
40 sendHold chan struct{}
41}
42
43func newFakeRawTransport() *fakeRawTransport {
44 return &fakeRawTransport{
45 inbound: make(chan []byte, 8),
46 closedCh: make(chan struct{}),
47 }
48}
49
50func (f *fakeRawTransport) Send(envelope wire.Envelope) error {
51 f.mu.Lock()
52 hold := f.sendHold
53 f.mu.Unlock()
54 if hold != nil {
55 select {
56 case <-hold:
57 case <-f.closedCh:
58 return errFakeRawClosed
59 }
60 }
61 select {
62 case <-f.closedCh:
63 return errFakeRawClosed
64 default:
65 }
66 f.mu.Lock()
67 f.sends = append(f.sends, append([]byte(nil), envelope.Payload...))
68 f.mu.Unlock()
69 return nil
70}
71
72func (f *fakeRawTransport) RecvBounded(limit uint64) (wire.Envelope, error) {
73 f.mu.Lock()
74 f.limits = append(f.limits, limit)
75 f.mu.Unlock()
76 select {
77 case payload := <-f.inbound:
78 if uint64(len(payload)) > limit {
79 return wire.Envelope{}, streamframe.ErrTooLarge
80 }
81 return wire.Envelope{Payload: payload}, nil
82 case <-f.closedCh:
83 return wire.Envelope{}, errFakeRawClosed
84 }
85}
86
87func (f *fakeRawTransport) Close() error {
88 f.closes.Add(1)
89 f.closeOnce.Do(func() { close(f.closedCh) })
90 return f.closeErr
91}
92
93func (f *fakeRawTransport) recordedLimits() []uint64 {
94 f.mu.Lock()
95 defer f.mu.Unlock()
96 return append([]uint64(nil), f.limits...)
97}
98
99func (f *fakeRawTransport) recordedSends() [][]byte {
100 f.mu.Lock()
101 defer f.mu.Unlock()
102 return append([][]byte(nil), f.sends...)
103}
104
105// carrierTestCeilings is a valid advertisement with the smallest envelope
106// ceiling, so the negotiated bound stays clearly above wire.PreambleLimit.
107func carrierTestCeilings() MuxCeilings {
108 ceilings := DefaultMuxCeilings()
109 ceilings.MaxReceiveEnvelopeBytes = MinMuxEnvelopeBytes
110 return ceilings
111}
112
113// TestFramedCarrierBridgeBoundsPreambleThenNegotiated proves the bridge bounds
114// every receive to wire.PreambleLimit before Negotiate and to the negotiated
115// envelope ceiling afterwards, refusing an oversized frame at each phase.
116func TestFramedCarrierBridgeBoundsPreambleThenNegotiated(t *testing.T) {
117 raw := newFakeRawTransport()
118 bridge, err := NewPreambleCarrier(raw)
119 require.NoError(t, err)
120 require.True(t, bridge.Preamble())
121 require.Equal(t, uint64(wire.PreambleLimit), bridge.ReceiveLimit())
122
123 small := []byte("preamble-sized request")
124 raw.inbound <- small
125 got, err := bridge.Receive(context.Background())
126 require.NoError(t, err)
127 require.Equal(t, small, got)
128 require.Equal(t, []uint64{wire.PreambleLimit}, raw.recordedLimits())
129
130 oversize := make([]byte, wire.PreambleLimit+1)
131 raw.inbound <- oversize
132 _, err = bridge.Receive(context.Background())
133 require.ErrorIs(t, err, streamframe.ErrTooLarge)
134
135 ceilings := carrierTestCeilings()
136 require.NoError(t, bridge.Negotiate(ceilings))
137 require.False(t, bridge.Preamble())
138 require.Equal(t, ceilings.MaxReceiveEnvelopeBytes, bridge.ReceiveLimit())
139
140 raw.inbound <- oversize
141 got, err = bridge.Receive(context.Background())
142 require.NoError(t, err)
143 require.Equal(t, oversize, got)
144 limits := raw.recordedLimits()
145 require.Equal(t, ceilings.MaxReceiveEnvelopeBytes, limits[len(limits)-1])
146}
147
148// TestFramedCarrierBridgeNegotiateIsExplicitAndOneWay proves the state
149// transition is explicit: an invalid advertisement is refused, a second
150// negotiation is refused, and negotiation after Close is refused.
151func TestFramedCarrierBridgeNegotiateIsExplicitAndOneWay(t *testing.T) {
152 raw := newFakeRawTransport()
153 bridge, err := NewPreambleCarrier(raw)
154 require.NoError(t, err)
155
156 require.ErrorIs(t, bridge.Negotiate(MuxCeilings{}), ErrInvalidCeilings)
157
158 require.NoError(t, bridge.Negotiate(carrierTestCeilings()))
159 require.ErrorIs(t, bridge.Negotiate(carrierTestCeilings()), ErrCarrierState)
160
161 require.NoError(t, bridge.Close())
162 require.ErrorIs(t, bridge.Negotiate(carrierTestCeilings()), ErrCarrierClosed)
163}
164
165// TestNewPreambleCarrierRefusesNil proves construction validates its raw
166// transport.
167func TestNewPreambleCarrierRefusesNil(t *testing.T) {
168 _, err := NewPreambleCarrier(nil)
169 require.ErrorIs(t, err, ErrCarrierState)
170}
171
172// TestFramedCarrierBridgeSendIsSynchronousAndExact proves Send returns the raw
173// write result: it does not return while the write is held, records the payload
174// byte for byte, and propagates the raw error.
175func TestFramedCarrierBridgeSendIsSynchronousAndExact(t *testing.T) {
176 raw := newFakeRawTransport()
177 bridge, err := NewPreambleCarrier(raw)
178 require.NoError(t, err)
179 defer func() { _ = bridge.Close() }()
180
181 raw.sendHold = make(chan struct{})
182 payload := []byte("exact outbound envelope")
183 sendDone := make(chan error, 1)
184 go func() { sendDone <- bridge.Send(context.Background(), payload) }()
185
186 select {
187 case err := <-sendDone:
188 t.Fatalf("Send returned %v while the raw write was held", err)
189 case <-time.After(50 * time.Millisecond):
190 }
191 require.Empty(t, raw.recordedSends())
192
193 close(raw.sendHold)
194 require.NoError(t, <-sendDone)
195 require.Equal(t, [][]byte{payload}, raw.recordedSends())
196
197 raw.closeErr = errors.New("raw close failed")
198 require.ErrorIs(t, bridge.Close(), raw.closeErr)
199 require.ErrorIs(t, bridge.Send(context.Background(), payload), ErrCarrierClosed)
200 _, err = bridge.Receive(context.Background())
201 require.ErrorIs(t, err, ErrCarrierClosed)
202}
203
204// TestFramedCarrierBridgeReceiveContextCancelReleasesRead proves a cancelled
205// receive closes the raw carriage to release the blocked call, returns the
206// context error, and leaves no goroutine behind.
207func TestFramedCarrierBridgeReceiveContextCancelReleasesRead(t *testing.T) {
208 raw := newFakeRawTransport()
209 bridge, err := NewPreambleCarrier(raw)
210 require.NoError(t, err)
211
212 before := runtime.NumGoroutine()
213 ctx, cancel := context.WithCancel(context.Background())
214 recvDone := make(chan error, 1)
215 go func() {
216 _, err := bridge.Receive(ctx)
217 recvDone <- err
218 }()
219 require.Eventually(t, func() bool { return len(raw.recordedLimits()) == 1 }, time.Second, time.Millisecond)
220 cancel()
221 select {
222 case err := <-recvDone:
223 require.ErrorIs(t, err, context.Canceled)
224 case <-time.After(5 * time.Second):
225 t.Fatal("Receive did not return after context cancellation")
226 }
227 require.Equal(t, int64(1), raw.closes.Load(), "cancellation must release the blocked raw read")
228 require.NoError(t, bridge.Close())
229 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= before+1 }, 5*time.Second, time.Millisecond,
230 "the bridge must not abandon its receive worker (before=%d now=%d)", before, runtime.NumGoroutine())
231}
232
233// TestFramedCarrierBridgeCloseInterruptsBlockedReceiveAndIsConcurrentSafe
234// proves Close unblocks a blocked receive, is idempotent across concurrent
235// callers, closes the raw carriage exactly once, and joins its workers.
236func TestFramedCarrierBridgeCloseInterruptsBlockedReceiveAndIsConcurrentSafe(t *testing.T) {
237 raw := newFakeRawTransport()
238 bridge, err := NewPreambleCarrier(raw)
239 require.NoError(t, err)
240
241 before := runtime.NumGoroutine()
242 recvDone := make(chan error, 1)
243 go func() {
244 _, err := bridge.Receive(context.Background())
245 recvDone <- err
246 }()
247 require.Eventually(t, func() bool { return len(raw.recordedLimits()) == 1 }, time.Second, time.Millisecond)
248
249 const closers = 8
250 var wg sync.WaitGroup
251 closeErrs := make([]error, closers)
252 for i := range closers {
253 wg.Add(1)
254 go func() {
255 defer wg.Done()
256 closeErrs[i] = bridge.Close()
257 }()
258 }
259 wg.Wait()
260 for _, err := range closeErrs {
261 require.NoError(t, err)
262 }
263 require.Equal(t, int64(1), raw.closes.Load(), "Close must close the raw carriage exactly once")
264
265 select {
266 case err := <-recvDone:
267 require.ErrorIs(t, err, ErrCarrierClosed)
268 case <-time.After(5 * time.Second):
269 t.Fatal("Close did not unblock the receive")
270 }
271 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= before+1 }, 5*time.Second, time.Millisecond,
272 "Close must join every bridge worker (before=%d now=%d)", before, runtime.NumGoroutine())
273}
274
275// TestFramedCarrierBridgeOverIPCFramer proves the bridge drives a real vev raw
276// framed transport: the shared framing round-trips through the bridge, the
277// preamble bound refuses an oversized frame before its body, and Negotiate
278// raises the bound for a larger negotiated frame.
279func TestFramedCarrierBridgeOverIPCFramer(t *testing.T) {
280 t.Run("preamble and negotiated frames round trip", func(t *testing.T) {
281 left, right := net.Pipe()
282 t.Cleanup(func() {
283 _ = left.Close()
284 _ = right.Close()
285 })
286 raw, ok := ipc.NewTransport(left).(RawFramedTransport)
287 require.True(t, ok, "the IPC transport must satisfy the bounded raw capability")
288 bridge, err := NewPreambleCarrier(raw)
289 require.NoError(t, err)
290 peer := ipc.NewTransport(right)
291 defer func() {
292 _ = bridge.Close()
293 _ = peer.Close()
294 }()
295
296 small := []byte("small preamble frame")
297 go func() { _ = peer.Send(wire.Envelope{Payload: small}) }()
298 got, err := bridge.Receive(context.Background())
299 require.NoError(t, err)
300 require.Equal(t, small, got)
301
302 require.NoError(t, bridge.Negotiate(carrierTestCeilings()))
303 large := make([]byte, wire.PreambleLimit+4096)
304 for i := range large {
305 large[i] = byte('a' + i%26)
306 }
307 go func() { _ = peer.Send(wire.Envelope{Payload: large}) }()
308 got, err = bridge.Receive(context.Background())
309 require.NoError(t, err)
310 require.Equal(t, large, got)
311
312 outbound := []byte("exact outbound through the real framer")
313 type peerRecv struct {
314 envelope wire.Envelope
315 err error
316 }
317 peerResult := make(chan peerRecv, 1)
318 go func() {
319 envelope, err := peer.Recv()
320 peerResult <- peerRecv{envelope: envelope, err: err}
321 }()
322 require.NoError(t, bridge.Send(context.Background(), outbound))
323 result := <-peerResult
324 require.NoError(t, result.err)
325 require.Equal(t, outbound, result.envelope.Payload)
326 })
327
328 t.Run("oversized preamble frame is refused before its body", func(t *testing.T) {
329 left, right := net.Pipe()
330 t.Cleanup(func() {
331 _ = left.Close()
332 _ = right.Close()
333 })
334 raw, ok := ipc.NewTransport(left).(RawFramedTransport)
335 require.True(t, ok)
336 bridge, err := NewPreambleCarrier(raw)
337 require.NoError(t, err)
338 defer func() { _ = bridge.Close() }()
339
340 // Advertise a 1 MiB frame and never send its body: only the four-byte
341 // length prefix exists on the wire.
342 go func() { _, _ = right.Write(binary.BigEndian.AppendUint32(nil, 1<<20)) }()
343 _, err = bridge.Receive(context.Background())
344 require.ErrorIs(t, err, streamframe.ErrTooLarge)
345 })
346}