render_coordinator_handshake_test.go Raw History Folder

1package daemon
2
3import (
4	"io"
5	"sync"
6	"testing"
7	"time"
8
9	"github.com/stretchr/testify/require"
10
11	"github.com/bnema/vev/internal/domain"
12	"github.com/bnema/vev/internal/protocol/wire"
13)
14
15// fanoutBlockingTransport blocks only the first output send. The paired
16// healthy attachment must still receive its frame while this peer is stuck.
17type fanoutBlockingTransport struct {
18	sent       chan wire.Envelope
19	entered    chan struct{}
20	release    chan struct{}
21	done       chan struct{}
22	enteredOne sync.Once
23	releaseOne sync.Once
24	doneOne    sync.Once
25}
26
27func newFanoutBlockingTransport() *fanoutBlockingTransport {
28	return &fanoutBlockingTransport{
29		sent:    make(chan wire.Envelope, 8),
30		entered: make(chan struct{}),
31		release: make(chan struct{}),
32		done:    make(chan struct{}),
33	}
34}
35
36func (t *fanoutBlockingTransport) Send(frame wire.Envelope) error {
37	t.sent <- frame
38	if envelopeMessageName(nil, frame.Payload) == "Output" {
39		t.enteredOne.Do(func() { close(t.entered) })
40		<-t.release
41	}
42	t.doneOne.Do(func() { close(t.done) })
43	return nil
44}
45
46func (t *fanoutBlockingTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
47
48func (t *fanoutBlockingTransport) Close() error {
49	t.releaseOne.Do(func() { close(t.release) })
50	return nil
51}
52
53// Fresh producer state may arrive after route has published an attachment but
54// before the transport has accepted Welcome. It must remain pending until the
55// exact attachment incarnation completes that handshake.
56func TestRenderCoordinatorFreshWakeWaitsForWelcome(t *testing.T) {
57	h := newCoordinatorHarness(t)
58	ac := &attachedClient{}
59	h.rc.attachWithReadiness(ac, false)
60	lease := h.rc.attachmentLease(ac)
61	require.NotNil(t, lease)
62
63	h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true, producer: "pty"})
64	timer := awaitCoordinatorScheduledTimer(t, h.clk)
65	h.rc.mu.Lock()
66	workerDone := h.rc.normalLane.token.done
67	h.rc.mu.Unlock()
68	require.NotNil(t, workerDone)
69	timer.ch <- time.Time{}
70	requireWorkerExit(t, workerDone)
71	requireNoWake(t, h.wakes)
72
73	require.True(t, h.rc.markAttachmentReady(lease))
74	h.rc.fireCurrent(false)
75	wake := awaitWake(t, h.wakes)
76	require.True(t, wake.reset)
77	require.Equal(t, 1, wake.coalesced)
78	require.Nil(t, wake.lease, "shared wakes do not carry an attachment lease")
79	requireNoWake(t, h.wakes)
80}
81
82func TestRenderCoordinatorFanoutDoesNotWaitForSlowTransport(t *testing.T) {
83	d, sess, healthy, healthySends, _ := newManualTabSession(t, 1)
84	slowTransport := newFanoutBlockingTransport()
85	slow := &attachedClient{
86		tr:     slowTransport,
87		output: newOutputStateStream(),
88		size:   domain.Size{Cols: 80, Rows: 24},
89	}
90	slow.initOverlays()
91	slow.setSession(sess)
92	sess.core().mu.Lock()
93	sess.core().attachments[slow] = struct{}{}
94	sess.core().mu.Unlock()
95
96	rc := d.ensureRenderCoordinator(sess)
97	rc.opts.clock = nil // publish, then fire synchronously below
98	rc.attach(healthy)
99	rc.attach(slow)
100	rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "fanout"})
101	rc.fireCurrent(false)
102
103	select {
104	case frame := <-healthySends:
105		require.Equal(t, "Output", envelopeMessageName(t, frame.Payload), "healthy attachment did not receive output")
106	case <-time.After(2 * time.Second):
107		t.Fatal("healthy attachment remained blocked behind the slow transport")
108	}
109	select {
110	case <-slowTransport.entered:
111	case <-time.After(2 * time.Second):
112		t.Fatal("slow transport did not enter its blocked output send")
113	}
114	slowTransport.releaseOne.Do(func() { close(slowTransport.release) })
115	select {
116	case <-slowTransport.done:
117	case <-time.After(2 * time.Second):
118		t.Fatal("slow output send did not finish after release")
119	}
120}
121
122func TestRenderCoordinatorWelcomeReadinessIsAttachmentScoped(t *testing.T) {
123	h := newCoordinatorHarness(t)
124	slow, healthy := &attachedClient{}, &attachedClient{}
125	h.rc.attachWithReadiness(slow, false)
126	h.rc.attachWithReadiness(healthy, true)
127	slowLease := h.rc.attachmentLease(slow)
128	require.NotNil(t, slowLease)
129
130	h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true, producer: "pty"})
131	h.rc.fireCurrent(false)
132	wake := awaitTestValue(t, h.wakes, "ready attachment did not receive output while peer handshakes")
133	require.Contains(t, wake.attachmentLeases, healthy)
134	_, slowSelected := wake.attachmentLeases[slow]
135	require.False(t, slowSelected)
136	h.rc.mu.Lock()
137	require.True(t, h.rc.pending, "the handshaking peer must retain the shared mutation")
138	h.rc.mu.Unlock()
139
140	require.True(t, h.rc.markAttachmentReady(slowLease))
141	h.rc.fireCurrentForLease(slowLease)
142	wake = awaitTestValue(t, h.wakes, "handshaken attachment did not receive deferred output")
143	require.Contains(t, wake.attachmentLeases, slow)
144	_, healthySelected := wake.attachmentLeases[healthy]
145	require.False(t, healthySelected, "a deferred peer must not replay output to a healthy peer")
146	h.rc.mu.Lock()
147	require.False(t, h.rc.pending, "the shared mutation must clear after the handshake catches up")
148	h.rc.mu.Unlock()
149}
150
151// A wake captured for an old attachment generation must not become valid after
152// the same attachment object is rebound to a replacement connection.
153func TestRenderCoordinatorStaleGenerationWakeIsFenced(t *testing.T) {
154	h := newCoordinatorHarness(t)
155	ac := &attachedClient{}
156	h.rc.attachWithReadiness(ac, true)
157	oldLease := h.rc.attachmentLease(ac)
158	require.NotNil(t, oldLease)
159	h.rc.invalidateForLease(ac, oldLease, renderInvalidation{class: invalidateUrgent, reset: true, producer: "old"})
160	h.rc.fireCurrentForLease(oldLease)
161	wake := awaitTestValue(t, h.wakes, "old attachment wake was not published")
162	require.True(t, h.rc.wakeCurrent(wake))
163
164	h.rc.noteDetach(ac)
165	h.rc.attachWithReadiness(ac, true)
166	require.False(t, h.rc.wakeCurrent(wake), "a stale generation wake must not target the replacement connection")
167}
168
169// Parking and resuming the same client object is a new incarnation: a failed
170// Welcome on the replacement transport must not revive readiness from before
171// the park.
172func TestRenderCoordinatorParkedResumeRequiresNewWelcome(t *testing.T) {
173	h := newCoordinatorHarness(t)
174	ac := &attachedClient{}
175	h.rc.attachWithReadiness(ac, false)
176	firstLease := h.rc.attachmentLease(ac)
177	require.True(t, h.rc.markAttachmentReady(firstLease))
178	h.rc.noteDetach(ac)
179	h.rc.attachWithReadiness(ac, false)
180	resumedLease := h.rc.attachmentLease(ac)
181	require.NotNil(t, resumedLease)
182	require.NotSame(t, firstLease, resumedLease)
183	require.False(t, h.rc.markAttachmentReady(firstLease), "stale Welcome must not bless the reused attachment")
184
185	h.rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "session"})
186	timer := awaitCoordinatorScheduledTimer(t, h.clk)
187	h.rc.mu.Lock()
188	workerDone := h.rc.normalLane.token.done
189	h.rc.mu.Unlock()
190	require.NotNil(t, workerDone)
191	timer.ch <- time.Time{}
192	requireWorkerExit(t, workerDone)
193	requireNoWake(t, h.wakes)
194
195	require.True(t, h.rc.markAttachmentReady(resumedLease))
196	h.rc.fireCurrent(false)
197	wake := awaitWake(t, h.wakes)
198	require.True(t, wake.reset)
199	require.Equal(t, 1, wake.coalesced)
200}
render_coordinator_handshake_test.go RawHistoryFolder

render_coordinator_handshake_test.go Raw History Folder
