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