1package client
2
3import (
4 "context"
5 "strings"
6 "testing"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10 "github.com/bnema/vev/internal/protocol"
11 "github.com/stretchr/testify/require"
12)
13
14// sentInput joins every Input message a stream carried, in order.
15func sentInput(stream *sessionTestStream) string {
16 var out strings.Builder
17 for _, message := range stream.messages() {
18 if input, ok := message.(protocol.Input); ok {
19 out.Write(input.Data)
20 }
21 }
22 return out.String()
23}
24
25// pushKeys feeds the terminal one byte per read, like a keyboard, and waits
26// until the shared reader consumed them all.
27func pushKeys(t *testing.T, h *attachTestHarness, keys string) {
28 t.Helper()
29 for _, b := range []byte(keys) {
30 h.reader.push([]byte{b})
31 }
32 require.Eventually(t, func() bool { return len(h.reader.chunks) == 0 }, 5*time.Second, time.Millisecond)
33}
34
35// TestResumeReplaysKeysHeldWhileConnecting types while a lost attachment waits
36// for its backoff and while the resume stream opens (including a failed open
37// that is retried), then proves the resumed session receives every key, once,
38// in order, and the lost attachment none.
39func TestResumeReplaysKeysHeldWhileConnecting(t *testing.T) {
40 for _, tt := range []struct {
41 name string
42 backoffKeys string
43 openKeys string
44 failFirst bool
45 retryKeys string
46 }{
47 {name: "typed during backoff", backoffKeys: "HELD_ONE"},
48 {name: "typed during stream open", openKeys: "HELD_TWO\r"},
49 {name: "typed during backoff and open", backoffKeys: "HELD_ONE", openKeys: "HELD_TWO\r"},
50 {name: "kept across a failed open", openKeys: "HELD_ONE", failFirst: true, retryKeys: "HELD_TWO\r"},
51 } {
52 t.Run(tt.name, func(t *testing.T) {
53 picker := newAttachTestPicker()
54 h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
55 first := newSessionTestStream()
56 second := newSessionTestStream()
57 opened := make(chan int, 4)
58 failed := make(chan struct{})
59 release := make(chan struct{})
60 count := 0
61 h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
62 count++
63 switch {
64 case count == 1:
65 return first, nil
66 case tt.failFirst && count == 2:
67 opened <- count
68 select {
69 case <-failed:
70 case <-ctx.Done():
71 }
72 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "route down"}
73 }
74 opened <- count
75 select {
76 case <-release:
77 case <-ctx.Done():
78 return nil, ctx.Err()
79 }
80 return second, nil
81 })
82 awaitOpen := func() {
83 select {
84 case <-opened:
85 case <-time.After(5 * time.Second):
86 t.Fatal("resume did not open")
87 }
88 }
89 picker.commit(sessionTestRequest(false))
90 deliverReadyStream(t, first)
91 awaitAttachedState(t, h.sup)
92 first.fail(testStreamLoss())
93
94 require.Eventually(t, func() bool { return h.sup.State().Presentation == PresentConnecting }, 5*time.Second, time.Millisecond)
95 pushKeys(t, h, tt.backoffKeys)
96 fireResumeTimerAfter(t, h.clock, 1, func() {})
97 awaitOpen()
98 pushKeys(t, h, tt.openKeys)
99 if tt.failFirst {
100 close(failed)
101 fireResumeTimerAfter(t, h.clock, 2, func() {})
102 awaitOpen()
103 pushKeys(t, h, tt.retryKeys)
104 }
105 close(release)
106 deliverReadyStream(t, second)
107 awaitAttachedState(t, h.sup)
108
109 want := tt.backoffKeys + tt.openKeys + tt.retryKeys
110 require.Eventually(t, func() bool { return sentInput(second) == want }, 5*time.Second, time.Millisecond,
111 "resumed session input = %q, want %q", sentInput(second), want)
112 require.Empty(t, sentInput(first))
113 })
114 }
115}