1package dgram
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "reflect"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/bnema/vev/internal/ports"
14)
15
16type fakeTransport struct {
17 recv chan recvResult
18 sent chan ports.Frame
19 sendErr chan error
20 linkState ports.LinkState
21 linkEvents chan ports.LinkEvent
22 closed chan struct{}
23 closeOnce sync.Once
24}
25
26type recvResult struct {
27 frame ports.Frame
28 err error
29}
30
31func newFakeTransport() *fakeTransport {
32 return &fakeTransport{recv: make(chan recvResult, 4), sent: make(chan ports.Frame, 4), sendErr: make(chan error, 4), linkEvents: make(chan ports.LinkEvent, 4), linkState: ports.LinkStateConnected, closed: make(chan struct{})}
33}
34func (f *fakeTransport) Send(fr ports.Frame) error {
35 select {
36 case err := <-f.sendErr:
37 return err
38 default:
39 }
40 f.sent <- fr
41 return nil
42}
43func (f *fakeTransport) Recv() (ports.Frame, error) {
44 select {
45 case r := <-f.recv:
46 return r.frame, r.err
47 case <-f.closed:
48 return ports.Frame{}, io.EOF
49 }
50}
51func (f *fakeTransport) Close() error {
52 f.closeOnce.Do(func() { close(f.closed) })
53 return nil
54}
55func (f *fakeTransport) LinkState() ports.LinkState { return f.linkState }
56func (f *fakeTransport) LinkEvents() <-chan ports.LinkEvent { return f.linkEvents }
57
58func TestProxyRuntimeCopiesFramesAndDaemonEOFIsTerminal(t *testing.T) {
59 client := newFakeTransport()
60 daemon := newFakeTransport()
61 errCh := make(chan error, 1)
62 go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()
63 client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgInput, Payload: []byte("x")}}
64 select {
65 case got := <-daemon.sent:
66 if got.Type != ports.MsgInput {
67 t.Fatalf("daemon got type %v", got.Type)
68 }
69 case <-time.After(time.Second):
70 t.Fatal("timeout waiting for copied frame")
71 }
72 daemon.recv <- recvResult{err: io.EOF}
73 select {
74 case err := <-errCh:
75 if err != nil {
76 t.Fatalf("Run err=%v, want nil", err)
77 }
78 case <-time.After(time.Second):
79 t.Fatal("timeout waiting for terminal EOF")
80 }
81}
82
83func TestProxyRuntimeClampsDatagramHelloOutputWindowAndPreservesControl(t *testing.T) {
84 for _, requested := range []uint8{0, 1, 8} {
85 t.Run(fmt.Sprintf("requested_%d", requested), func(t *testing.T) {
86 client := newFakeTransport()
87 daemon := newFakeTransport()
88 errCh := make(chan error, 1)
89 go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()
90 hello := ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", MaxOutputInFlight: requested}
91 client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}}
92
93 select {
94 case frame := <-daemon.sent:
95 got, err := ports.UnmarshalHello(frame.Payload)
96 if err != nil {
97 t.Fatal(err)
98 }
99 if got.MaxOutputInFlight != 1 {
100 t.Fatalf("proxied output window=%d, want datagram-safe 1", got.MaxOutputInFlight)
101 }
102 case <-time.After(time.Second):
103 t.Fatal("timeout waiting for proxied Hello")
104 }
105
106 input := ports.Frame{Type: ports.MsgInput, Payload: []byte("still flowing")}
107 ack := ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: 7})}
108 client.recv <- recvResult{frame: input}
109 client.recv <- recvResult{frame: ack}
110 for _, want := range []ports.Frame{input, ack} {
111 select {
112 case got := <-daemon.sent:
113 if !reflect.DeepEqual(got, want) {
114 t.Fatalf("proxied control frame=%+v, want %+v", got, want)
115 }
116 case <-time.After(time.Second):
117 t.Fatalf("timeout waiting for proxied %v", want.Type)
118 }
119 }
120
121 daemon.recv <- recvResult{err: io.EOF}
122 if err := <-errCh; err != nil {
123 t.Fatalf("Run err=%v, want nil", err)
124 }
125 })
126 }
127}
128
129func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
130 tests := []struct {
131 name string
132 state ports.LinkState
133 err error
134 }{
135 {name: "connected pending full", state: ports.LinkStateConnected, err: ErrPendingFull},
136 {name: "degraded transient error", state: ports.LinkStateDegraded, err: errors.New("degraded write drop")},
137 }
138 for _, tt := range tests {
139 t.Run(tt.name, func(t *testing.T) {
140 client := newFakeTransport()
141 daemon := newFakeTransport()
142 // No link event is emitted, so the idle timer is never armed; the
143 // retry alone must recover. IdleTTL is large to make that explicit.
144 client.linkState = tt.state
145 errCh := make(chan error, 1)
146 go func() {
147 errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
148 }()
149 client.sendErr <- tt.err
150 daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
151 select {
152 case got := <-client.sent:
153 if got.Type != ports.MsgOutput || string(got.Payload) != "preserved during outage" {
154 t.Fatalf("client got %+v, want retried output", got)
155 }
156 case <-time.After(time.Second):
157 t.Fatal("timeout waiting for recovered daemon output")
158 }
159 daemon.recv <- recvResult{err: io.EOF}
160 select {
161 case err := <-errCh:
162 if err != nil {
163 t.Fatalf("Run err=%v, want nil", err)
164 }
165 case <-time.After(time.Second):
166 t.Fatal("timeout waiting for daemon EOF")
167 }
168 })
169 }
170}
171
172// TestProxyRuntimeSurvivesPendingFullAndPreservesOrder is the congestion
173// regression: a persistently pending-full client window must not tear the proxy
174// down (the old idle TTL killed it after 30s) and every frame must arrive in
175// order. IdleTTL is a nanosecond: if a retry armed the idle timer the proxy
176// would die at once, so surviving proves retries no longer arm it.
177func TestProxyRuntimeSurvivesPendingFullAndPreservesOrder(t *testing.T) {
178 client := newFakeTransport()
179 daemon := newFakeTransport()
180 client.linkState = ports.LinkStateConnected
181 errCh := make(chan error, 1)
182 go func() {
183 errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Nanosecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
184 }()
185 const frames = 4
186 for i := range frames {
187 client.sendErr <- ErrPendingFull
188 daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}}
189 select {
190 case got := <-client.sent:
191 if got.Type != ports.MsgOutput || len(got.Payload) != 1 || got.Payload[0] != byte(i) {
192 t.Fatalf("frame %d delivered out of order or dropped: %+v", i, got)
193 }
194 case <-time.After(time.Second):
195 t.Fatalf("frame %d dropped or proxy killed under pending-full", i)
196 }
197 }
198 select {
199 case err := <-errCh:
200 t.Fatalf("proxy returned under sustained pending-full: %v", err)
201 case <-time.After(20 * time.Millisecond):
202 }
203 daemon.recv <- recvResult{err: io.EOF}
204 select {
205 case err := <-errCh:
206 if err != nil {
207 t.Fatalf("Run err=%v, want nil", err)
208 }
209 case <-time.After(time.Second):
210 t.Fatal("timeout waiting for daemon EOF")
211 }
212}
213
214func TestProxyRuntimeOfflineLinkEventExpiresViaIdleTTL(t *testing.T) {
215 client := newFakeTransport()
216 daemon := newFakeTransport()
217 client.linkState = ports.LinkStateOffline
218 errCh := make(chan error, 1)
219 go func() {
220 errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond}.Run(t.Context())
221 }()
222 client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
223 select {
224 case err := <-errCh:
225 if !errors.Is(err, ErrLinkDead) {
226 t.Fatalf("Run err=%v, want ErrLinkDead", err)
227 }
228 case <-time.After(time.Second):
229 t.Fatal("timeout waiting for idle expiry after offline link event")
230 }
231}
232
233type proxyManualClock struct {
234 mu sync.Mutex
235 now time.Time
236 timers chan *proxyManualTimer
237}
238
239func newProxyManualClock(now time.Time) *proxyManualClock {
240 return &proxyManualClock{now: now, timers: make(chan *proxyManualTimer, 4)}
241}
242
243func (c *proxyManualClock) Now() time.Time {
244 c.mu.Lock()
245 defer c.mu.Unlock()
246 return c.now
247}
248
249func (c *proxyManualClock) NewTimer(d time.Duration) ports.Timer {
250 tm := &proxyManualTimer{c: make(chan time.Time, 1), d: d}
251 c.timers <- tm
252 return tm
253}
254
255func (c *proxyManualClock) advance(d time.Duration) {
256 c.mu.Lock()
257 c.now = c.now.Add(d)
258 now := c.now
259 c.mu.Unlock()
260 for {
261 select {
262 case tm := <-c.timers:
263 if tm.duration() <= d {
264 tm.fire(now)
265 } else {
266 c.timers <- tm
267 return
268 }
269 default:
270 return
271 }
272 }
273}
274
275type proxyManualTimer struct {
276 mu sync.Mutex
277 c chan time.Time
278 d time.Duration
279}
280
281func (t *proxyManualTimer) C() <-chan time.Time { return t.c }
282func (t *proxyManualTimer) Reset(d time.Duration) bool {
283 t.mu.Lock()
284 defer t.mu.Unlock()
285 t.d = d
286 return true
287}
288func (t *proxyManualTimer) Stop() bool { return true }
289func (t *proxyManualTimer) duration() time.Duration {
290 t.mu.Lock()
291 defer t.mu.Unlock()
292 return t.d
293}
294func (t *proxyManualTimer) fire(now time.Time) {
295 select {
296 case t.c <- now:
297 default:
298 }
299}
300
301func waitForProxyManualTimers(t *testing.T, clk *proxyManualClock, n int) {
302 t.Helper()
303 deadline := time.Now().Add(time.Second)
304 for time.Now().Before(deadline) {
305 if len(clk.timers) >= n {
306 return
307 }
308 time.Sleep(time.Millisecond)
309 }
310 t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)
311}
312
313func TestProxyRuntimeClientDeadWaitsForIdleTTL(t *testing.T) {
314 client := newFakeTransport()
315 daemon := newFakeTransport()
316 clk := newProxyManualClock(time.Unix(0, 0))
317 errCh := make(chan error, 1)
318 go func() {
319 errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
320 }()
321
322 client.linkEvents <- ports.LinkEvent{State: ports.LinkStateDead}
323 waitForProxyManualTimers(t, clk, 1)
324 select {
325 case err := <-errCh:
326 t.Fatalf("Run returned before IdleTTL after client LinkStateDead: %v", err)
327 default:
328 }
329
330 clk.advance(time.Minute)
331 select {
332 case err := <-errCh:
333 if !errors.Is(err, ErrLinkDead) {
334 t.Fatalf("Run err=%v, want ErrLinkDead", err)
335 }
336 case <-time.After(time.Second):
337 t.Fatal("timeout waiting for idle expiry after client LinkStateDead")
338 }
339}
340
341func TestProxyRuntimeClientRecvDeadWaitsForIdleTTL(t *testing.T) {
342 client := newFakeTransport()
343 daemon := newFakeTransport()
344 clk := newProxyManualClock(time.Unix(0, 0))
345 errCh := make(chan error, 1)
346 go func() {
347 errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
348 }()
349
350 client.recv <- recvResult{err: ErrLinkDead}
351 waitForProxyManualTimers(t, clk, 1)
352 select {
353 case err := <-errCh:
354 t.Fatalf("Run returned before IdleTTL after client Recv ErrLinkDead: %v", err)
355 default:
356 }
357
358 clk.advance(time.Minute)
359 select {
360 case err := <-errCh:
361 if !errors.Is(err, ErrLinkDead) {
362 t.Fatalf("Run err=%v, want ErrLinkDead", err)
363 }
364 case <-time.After(time.Second):
365 t.Fatal("timeout waiting for idle expiry after client Recv ErrLinkDead")
366 }
367}
368
369func TestProxyRuntimeClientSendDeadWaitsForIdleTTL(t *testing.T) {
370 client := newFakeTransport()
371 daemon := newFakeTransport()
372 clk := newProxyManualClock(time.Unix(0, 0))
373 errCh := make(chan error, 1)
374 go func() {
375 errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Minute, Clock: clk}.Run(t.Context())
376 }()
377
378 client.sendErr <- ErrLinkDead
379 daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("client is gone")}}
380 waitForProxyManualTimers(t, clk, 1)
381 select {
382 case err := <-errCh:
383 t.Fatalf("Run returned before IdleTTL after client Send ErrLinkDead: %v", err)
384 default:
385 }
386
387 clk.advance(time.Minute)
388 select {
389 case err := <-errCh:
390 if !errors.Is(err, ErrLinkDead) {
391 t.Fatalf("Run err=%v, want ErrLinkDead", err)
392 }
393 case <-time.After(time.Second):
394 t.Fatal("timeout waiting for idle expiry after client Send ErrLinkDead")
395 }
396}
397
398func TestProxyRuntimeDaemonRecvDeadTearsDown(t *testing.T) {
399 client := newFakeTransport()
400 daemon := newFakeTransport()
401 errCh := make(chan error, 1)
402 go func() {
403 errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context())
404 }()
405 daemon.recv <- recvResult{err: ErrLinkDead}
406 select {
407 case err := <-errCh:
408 if !errors.Is(err, ErrLinkDead) {
409 t.Fatalf("Run err=%v, want ErrLinkDead", err)
410 }
411 case <-time.After(time.Second):
412 t.Fatal("daemon recv dead did not tear down proxy")
413 }
414}
415
416func TestProxyRuntimeClosedLinkEventsDoNotSpin(t *testing.T) {
417 client := newFakeTransport()
418 daemon := newFakeTransport()
419 close(client.linkEvents)
420 errCh := make(chan error, 1)
421 go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()
422 daemon.recv <- recvResult{err: io.EOF}
423 select {
424 case err := <-errCh:
425 if err != nil {
426 t.Fatalf("Run err=%v, want nil", err)
427 }
428 case <-time.After(time.Second):
429 t.Fatal("timeout waiting for daemon EOF")
430 }
431}
432
433type retryingOfflineTransport struct {
434 closed chan struct{}
435 firstSend chan struct{}
436 closeOnce sync.Once
437 firstSendOnce sync.Once
438}
439
440func newRetryingOfflineTransport() *retryingOfflineTransport {
441 return &retryingOfflineTransport{closed: make(chan struct{}), firstSend: make(chan struct{})}
442}
443
444func (t *retryingOfflineTransport) Send(ports.Frame) error {
445 t.firstSendOnce.Do(func() { close(t.firstSend) })
446 return errors.New("offline")
447}
448
449func (t *retryingOfflineTransport) Recv() (ports.Frame, error) {
450 <-t.closed
451 return ports.Frame{}, io.EOF
452}
453
454func (t *retryingOfflineTransport) Close() error {
455 t.closeOnce.Do(func() { close(t.closed) })
456 return nil
457}
458
459func (t *retryingOfflineTransport) LinkState() ports.LinkState { return ports.LinkStateOffline }
460func (t *retryingOfflineTransport) LinkEvents() <-chan ports.LinkEvent {
461 return nil
462}
463
464func TestProxyCopierStopsRetryingWhenContextCanceled(t *testing.T) {
465 ctx, cancel := context.WithCancel(t.Context())
466 client := newRetryingOfflineTransport()
467 daemon := newFakeTransport()
468 done := make(chan struct{})
469 copier := proxyCopier{ctx: ctx, errCh: make(chan<- proxyCopyErr), retryBackoff: time.Hour, clk: realClock{}}
470 go func() {
471 copier.copyFrames(proxyCopyDirection{src: daemon, dst: client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryRecoverable: true})
472 close(done)
473 }()
474
475 daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("retry until cancel")}}
476 select {
477 case <-client.firstSend:
478 case <-time.After(time.Second):
479 t.Fatal("timeout waiting for proxy send attempt")
480 }
481 cancel()
482 select {
483 case <-done:
484 case <-time.After(time.Second):
485 t.Fatal("retrying copier did not stop after context cancellation")
486 }
487}
488
489func TestClampDatagramHelloOutputWindowPreservesProtocolValidation(t *testing.T) {
490 t.Run("malformed payload remains malformed", func(t *testing.T) {
491 payload := append(ports.MarshalHello(ports.Hello{Version: ports.ProtocolVersion, MaxOutputInFlight: 8}), 0xff)
492 frame := ports.Frame{Type: ports.MsgHello, Payload: payload}
493 got := clampDatagramHelloOutputWindow(frame)
494 if !reflect.DeepEqual(got, frame) {
495 t.Fatalf("malformed Hello changed: got %#v, want %#v", got, frame)
496 }
497 if _, err := ports.UnmarshalHello(got.Payload); err == nil {
498 t.Fatal("malformed Hello became decodable")
499 }
500 })
501
502 t.Run("mismatched version remains mismatched", func(t *testing.T) {
503 wantVersion := ports.ProtocolVersion + 1
504 frame := ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
505 Version: wantVersion, MaxOutputInFlight: 8,
506 })}
507 got, err := ports.UnmarshalHello(clampDatagramHelloOutputWindow(frame).Payload)
508 if err != nil {
509 t.Fatal(err)
510 }
511 if got.Version != wantVersion {
512 t.Fatalf("version=%d, want unchanged mismatch %d", got.Version, wantVersion)
513 }
514 if got.MaxOutputInFlight != 1 {
515 t.Fatalf("output window=%d, want 1", got.MaxOutputInFlight)
516 }
517 })
518}