fix(phase4): bound cancellation and remove dead command APIs

commit: 68ba3515bce0ab42c7b194092d271b830af53264
author: brice · 2mo ago · 2026-08-03 13:59 UTC
committer: brice · 2mo ago · 2026-08-04 07:16 UTC
parent: 7739dc9
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

14 files · +291 -418

M internal/adapters/sshstdio/transport.go +28 -16


@@ -70,6 +70,8 @@ type transport struct {
70	70		operationCount int
71	71		closing bool
72	72		operationsDone chan struct{}
	73	+	closeOnce sync.Once
	74	+	closeErr error
73	75
74	76		mu sync.Mutex
75	77		readBuf []byte

@@ -160,22 +162,32 @@ func (t *transport) mapEOFError(err error) error {
160	162		}
161	163
162	164		func (t *transport) Close() error {
163		-	done := t.beginShutdown()
164		-	// Closing the reader first releases a Recv blocked in io.ReadFull even if
165		-	// the subprocess shutdown callback reports an error. This lets Recv emit
166		-	// its failed end mark before the process observer is closed.
167		-	var readErr error
168		-	if r, ok := t.r.(io.Closer); ok {
169		-	readErr = r.Close()
170		-	}
171		-	closeErr := t.close()
172		-	if done != nil {
173		-	<-done
174		-	}
175		-	if closeErr != nil {
176		-	return closeErr
177		-	}
178		-	return readErr
	165	+	t.closeOnce.Do(func() {
	166	+	done := t.beginShutdown()
	167	+	// Close owned streams before waiting for in-flight operations. The
	168	+	// writer is what releases a Send blocked on a child stdin pipe; the
	169	+	// reader releases a Recv blocked in io.ReadFull.
	170	+	var writeErr, readErr error
	171	+	if w, ok := t.w.(io.Closer); ok {
	172	+	writeErr = w.Close()
	173	+	}
	174	+	if r, ok := t.r.(io.Closer); ok {
	175	+	readErr = r.Close()
	176	+	}
	177	+	closeErr := t.close()
	178	+	if done != nil {
	179	+	<-done
	180	+	}
	181	+	switch {
	182	+	case closeErr != nil:
	183	+	t.closeErr = closeErr
	184	+	case readErr != nil:
	185	+	t.closeErr = readErr
	186	+	default:
	187	+	t.closeErr = writeErr
	188	+	}
	189	+	})
	190	+	return t.closeErr
179	191		}
180	192
181	193		func (t *transport) beginObservedOperation() bool {


@@ -70,6 +70,8 @@ type transport struct {
70		operationCount int	70		operationCount int
71		closing bool	71		closing bool
72		operationsDone chan struct{}	72		operationsDone chan struct{}
			73	+	closeOnce sync.Once
			74	+	closeErr error
73			75
74		mu sync.Mutex	76		mu sync.Mutex
75		readBuf []byte	77		readBuf []byte

@@ -160,22 +162,32 @@ func (t *transport) mapEOFError(err error) error {
160		}	162		}
161			163
162		func (t *transport) Close() error {	164		func (t *transport) Close() error {
163	-	done := t.beginShutdown()	165	+	t.closeOnce.Do(func() {
164	-	// Closing the reader first releases a Recv blocked in io.ReadFull even if	166	+	done := t.beginShutdown()
165	-	// the subprocess shutdown callback reports an error. This lets Recv emit	167	+	// Close owned streams before waiting for in-flight operations. The
166	-	// its failed end mark before the process observer is closed.	168	+	// writer is what releases a Send blocked on a child stdin pipe; the
167	-	var readErr error	169	+	// reader releases a Recv blocked in io.ReadFull.
168	-	if r, ok := t.r.(io.Closer); ok {	170	+	var writeErr, readErr error
169	-	readErr = r.Close()	171	+	if w, ok := t.w.(io.Closer); ok {
170	-	}	172	+	writeErr = w.Close()
171	-	closeErr := t.close()	173	+	}
172	-	if done != nil {	174	+	if r, ok := t.r.(io.Closer); ok {
173	-	<-done	175	+	readErr = r.Close()
174	-	}	176	+	}
175	-	if closeErr != nil {	177	+	closeErr := t.close()
176	-	return closeErr	178	+	if done != nil {
177	-	}	179	+	<-done
178	-	return readErr	180	+	}
			181	+	switch {
			182	+	case closeErr != nil:
			183	+	t.closeErr = closeErr
			184	+	case readErr != nil:
			185	+	t.closeErr = readErr
			186	+	default:
			187	+	t.closeErr = writeErr
			188	+	}
			189	+	})
			190	+	return t.closeErr
179		}	191		}
180			192
181		func (t *transport) beginObservedOperation() bool {	193		func (t *transport) beginObservedOperation() bool {

M internal/adapters/sshstdio/transport_test.go +37 -0


@@ -9,6 +9,7 @@ import (
9	9		"log/slog"
10	10		"os/exec"
11	11		"strings"
	12	+	"sync"
12	13		"testing"
13	14		"time"
14	15

@@ -106,6 +107,42 @@ func TestBuildCommandForRemoteCommandQuotesEveryWord(t *testing.T) {
106	107		}
107	108		}
108	109
	110	+	func TestCloseInterruptsBlockedSend(t *testing.T) {
	111	+	writer := &blockedWriter{started: make(chan struct{}), released: make(chan struct{})}
	112	+	transport := NewTransport(nil, writer, nil)
	113	+
	114	+	sendDone := make(chan error, 1)
	115	+	go func() { sendDone <- transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("blocked")}) }()
	116	+	<-writer.started
	117	+	if err := transport.Close(); err != nil {
	118	+	t.Fatalf("Close() error = %v", err)
	119	+	}
	120	+	if err := <-sendDone; err == nil {
	121	+	t.Fatal("Send() error = nil after Close()")
	122	+	}
	123	+	}
	124	+
	125	+	type blockedWriter struct {
	126	+	started chan struct{}
	127	+	released chan struct{}
	128	+	once sync.Once
	129	+	}
	130	+
	131	+	func (w *blockedWriter) Write([]byte) (int, error) {
	132	+	w.once.Do(func() { close(w.started) })
	133	+	<-w.released
	134	+	return 0, io.ErrClosedPipe
	135	+	}
	136	+
	137	+	func (w *blockedWriter) Close() error {
	138	+	select {
	139	+	case <-w.released:
	140	+	default:
	141	+	close(w.released)
	142	+	}
	143	+	return nil
	144	+	}
	145	+
109	146		func TestTransportRoundTripAndVersionMismatchFrame(t *testing.T) {
110	147		clientRead, serverWrite := io.Pipe()
111	148		serverRead, clientWrite := io.Pipe()


@@ -9,6 +9,7 @@ import (
9	"log/slog"	9		"log/slog"
10	"os/exec"	10		"os/exec"
11	"strings"	11		"strings"
		12	+	"sync"
12	"testing"	13		"testing"
13	"time"	14		"time"
14		15

@@ -106,6 +107,42 @@ func TestBuildCommandForRemoteCommandQuotesEveryWord(t *testing.T) {
106	}	107		}
107	}	108		}
108		109
		110	+	func TestCloseInterruptsBlockedSend(t *testing.T) {
		111	+	writer := &blockedWriter{started: make(chan struct{}), released: make(chan struct{})}
		112	+	transport := NewTransport(nil, writer, nil)
		113	+
		114	+	sendDone := make(chan error, 1)
		115	+	go func() { sendDone <- transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("blocked")}) }()
		116	+	<-writer.started
		117	+	if err := transport.Close(); err != nil {
		118	+	t.Fatalf("Close() error = %v", err)
		119	+	}
		120	+	if err := <-sendDone; err == nil {
		121	+	t.Fatal("Send() error = nil after Close()")
		122	+	}
		123	+	}
		124	+
		125	+	type blockedWriter struct {
		126	+	started chan struct{}
		127	+	released chan struct{}
		128	+	once sync.Once
		129	+	}
		130	+
		131	+	func (w *blockedWriter) Write([]byte) (int, error) {
		132	+	w.once.Do(func() { close(w.started) })
		133	+	<-w.released
		134	+	return 0, io.ErrClosedPipe
		135	+	}
		136	+
		137	+	func (w *blockedWriter) Close() error {
		138	+	select {
		139	+	case <-w.released:
		140	+	default:
		141	+	close(w.released)
		142	+	}
		143	+	return nil
		144	+	}
		145	+
109	func TestTransportRoundTripAndVersionMismatchFrame(t *testing.T) {	146		func TestTransportRoundTripAndVersionMismatchFrame(t *testing.T) {
110	clientRead, serverWrite := io.Pipe()	147		clientRead, serverWrite := io.Pipe()
111	serverRead, clientWrite := io.Pipe()	148		serverRead, clientWrite := io.Pipe()

M internal/app/run.go +19 -1


@@ -1049,10 +1049,28 @@ func runStdio(ctx context.Context) (retErr error) {
1049	1049		return err
1050	1050		}
1051	1051		defer func() { _ = transport.Close() }()
1052		-	stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil, sshstdio.WithRuntimeObserver(observer))
	1052	+
	1053	+	// sshstdio.Close owns and closes writers that implement io.Closer so a
	1054	+	// blocked Send can unwind. Keep process-global stdin/stdout outside that
	1055	+	// ownership boundary: input is a non-closing view and output is an owned
	1056	+	// pipe drained by a small forwarding goroutine.
	1057	+	stdin, stdout := os.Stdin, os.Stdout
	1058	+	stdoutReader, stdoutWriter := io.Pipe()
	1059	+	go func() {
	1060	+	if _, err := io.Copy(stdout, stdoutReader); err != nil {
	1061	+	_ = stdoutReader.CloseWithError(err)
	1062	+	}
	1063	+	}()
	1064	+	defer func() {
	1065	+	_ = stdoutWriter.Close()
	1066	+	_ = stdoutReader.Close()
	1067	+	}()
	1068	+	stdio := sshstdio.NewTransport(nonClosingReader{Reader: stdin}, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer))
1053	1069		return runStdioProxy(ctx, stdio, transport, log)
1054	1070		}
1055	1071
	1072	+	type nonClosingReader struct{ io.Reader }
	1073	+
1056	1074		var (
1057	1075		udpBootstrapTimeout = 5 * time.Second
1058	1076		udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd {


@@ -1049,10 +1049,28 @@ func runStdio(ctx context.Context) (retErr error) {
1049		return err	1049		return err
1050		}	1050		}
1051		defer func() { _ = transport.Close() }()	1051		defer func() { _ = transport.Close() }()
1052	-	stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil, sshstdio.WithRuntimeObserver(observer))	1052	+
			1053	+	// sshstdio.Close owns and closes writers that implement io.Closer so a
			1054	+	// blocked Send can unwind. Keep process-global stdin/stdout outside that
			1055	+	// ownership boundary: input is a non-closing view and output is an owned
			1056	+	// pipe drained by a small forwarding goroutine.
			1057	+	stdin, stdout := os.Stdin, os.Stdout
			1058	+	stdoutReader, stdoutWriter := io.Pipe()
			1059	+	go func() {
			1060	+	if _, err := io.Copy(stdout, stdoutReader); err != nil {
			1061	+	_ = stdoutReader.CloseWithError(err)
			1062	+	}
			1063	+	}()
			1064	+	defer func() {
			1065	+	_ = stdoutWriter.Close()
			1066	+	_ = stdoutReader.Close()
			1067	+	}()
			1068	+	stdio := sshstdio.NewTransport(nonClosingReader{Reader: stdin}, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer))
1053		return runStdioProxy(ctx, stdio, transport, log)	1069		return runStdioProxy(ctx, stdio, transport, log)
1054		}	1070		}
1055			1071
			1072	+	type nonClosingReader struct{ io.Reader }
			1073	+
1056		var (	1074		var (
1057		udpBootstrapTimeout = 5 * time.Second	1075		udpBootstrapTimeout = 5 * time.Second
1058		udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd {	1076		udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd {

M internal/usecase/client/handshake.go +4 -3


@@ -91,14 +91,15 @@ func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, o
91	91		go func() { completed <- operation() }()
92	92		select {
93	93		case err := <-completed:
94		-	if ctx.Err() != nil {
	94	+	if ctxErr := ctx.Err(); ctxErr != nil {
95	95		_ = transport.Close()
96		-	return ctx.Err()
	96	+	return ctxErr
97	97		}
98	98		return err
99	99		case <-ctx.Done():
100	100		_ = transport.Close()
101		-	<-completed
	101	+	// The result channel is buffered, so the operation can publish its
	102	+	// completion after Close without keeping this cancellation path stuck.
102	103		return ctx.Err()
103	104		}
104	105		}

 	go func() { completed <- operation() }()
 	select {
 	case err := <-completed:
-		if ctx.Err() != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
 			_ = transport.Close()
-			return ctx.Err()
+			return ctxErr
 		}
 		return err
 	case <-ctx.Done():
 		_ = transport.Close()
-		<-completed
+		// The result channel is buffered, so the operation can publish its
+		// completion after Close without keeping this cancellation path stuck.
 		return ctx.Err()
 	}
 }

M internal/usecase/client/session_connection.go +0 -11


@@ -60,14 +60,6 @@ func (c *SessionConnection) Transport() ports.Transport {
60	60		return c.transport
61	61		}
62	62
63		-	// Target returns the immutable daemon-facing session selection.
64		-	func (c *SessionConnection) Target() SessionTarget {
65		-	if c == nil {
66		-	return SessionTarget{}
67		-	}
68		-	return c.target
69		-	}
70		-
71	63		// AttachRequest returns the common request shape used after transport
72	64		// selection. It contains no remote/proxy semantic.
73	65		func (c *SessionConnection) AttachRequest() AttachRequest {

@@ -77,9 +69,6 @@ func (c *SessionConnection) AttachRequest() AttachRequest {
77	69		return AttachRequest{Intent: c.target.Intent, SessionName: c.target.SessionName}
78	70		}
79	71
80		-	// Request is the short form of AttachRequest for connection callers.
81		-	func (c *SessionConnection) Request() AttachRequest { return c.AttachRequest() }
82		-
83	72		// Close releases the owned transport.
84	73		func (c *SessionConnection) Close() error {
85	74		if c == nil \|\| c.transport == nil {


@@ -60,14 +60,6 @@ func (c *SessionConnection) Transport() ports.Transport {
60		return c.transport	60	return c.transport
61		}	61	}
62			62
63	-	// Target returns the immutable daemon-facing session selection.
64	-	func (c *SessionConnection) Target() SessionTarget {
65	-	if c == nil {
66	-	return SessionTarget{}
67	-	}
68	-	return c.target
69	-	}
70	-
71		// AttachRequest returns the common request shape used after transport	63	// AttachRequest returns the common request shape used after transport
72		// selection. It contains no remote/proxy semantic.	64	// selection. It contains no remote/proxy semantic.
73		func (c *SessionConnection) AttachRequest() AttachRequest {	65	func (c *SessionConnection) AttachRequest() AttachRequest {

@@ -77,9 +69,6 @@ func (c *SessionConnection) AttachRequest() AttachRequest {
77		return AttachRequest{Intent: c.target.Intent, SessionName: c.target.SessionName}	69	return AttachRequest{Intent: c.target.Intent, SessionName: c.target.SessionName}
78		}	70	}
79			71
80	-	// Request is the short form of AttachRequest for connection callers.
81	-	func (c *SessionConnection) Request() AttachRequest { return c.AttachRequest() }
82	-
83		// Close releases the owned transport.	72	// Close releases the owned transport.
84		func (c *SessionConnection) Close() error {	73	func (c *SessionConnection) Close() error {
85		if c == nil \|\| c.transport == nil {	74	if c == nil \|\| c.transport == nil {

D internal/usecase/daemon/attachment_commands_test.go +0 -287

@@ -1,287 +0,0 @@
1	-	package daemon
2	-
3	-	import (
4	-	"context"
5	-	"errors"
6	-	"sync"
7	-	"testing"
8	-	"time"
9	-
10	-	"github.com/stretchr/testify/require"
11	-
12	-	"github.com/bnema/vev/internal/ports"
13	-	)
14	-
15	-	type attachmentCommandTestTransport struct {
16	-	sent chan ports.Frame
17	-	entered chan struct{}
18	-	gate <-chan struct{}
19	-	}
20	-
21	-	func newAttachmentCommandTestTransport() *attachmentCommandTestTransport {
22	-	return &attachmentCommandTestTransport{sent: make(chan ports.Frame, 8)}
23	-	}
24	-
25	-	func (t *attachmentCommandTestTransport) Send(frame ports.Frame) error {
26	-	if t.entered != nil {
27	-	select {
28	-	case t.entered <- struct{}{}:
29	-	default:
30	-	}
31	-	}
32	-	if t.gate != nil {
33	-	<-t.gate
34	-	}
35	-	t.sent <- frame
36	-	return nil
37	-	}
38	-
39	-	func (t *attachmentCommandTestTransport) Recv() (ports.Frame, error) {
40	-	return ports.Frame{}, nil
41	-	}
42	-
43	-	func (t *attachmentCommandTestTransport) Close() error { return nil }
44	-
45	-	type attachmentCommandTestTimer struct {
46	-	ch chan time.Time
47	-	once sync.Once
48	-	}
49	-
50	-	func (t *attachmentCommandTestTimer) C() <-chan time.Time { return t.ch }
51	-	func (t *attachmentCommandTestTimer) Reset(time.Duration) bool { return false }
52	-	func (t *attachmentCommandTestTimer) Stop() bool { return true }
53	-	func (t *attachmentCommandTestTimer) fire() { t.once.Do(func() { t.ch <- time.Time{} }) }
54	-
55	-	type attachmentCommandTestClock struct {
56	-	timers chan *attachmentCommandTestTimer
57	-	}
58	-
59	-	func newAttachmentCommandTestClock() *attachmentCommandTestClock {
60	-	return &attachmentCommandTestClock{timers: make(chan *attachmentCommandTestTimer, 8)}
61	-	}
62	-
63	-	func (c *attachmentCommandTestClock) Now() time.Time { return time.Time{} }
64	-
65	-	func (c *attachmentCommandTestClock) NewTimer(delay time.Duration) ports.Timer {
66	-	if delay != attachedCommandTimeout {
67	-	return stubTimer{}
68	-	}
69	-	timer := &attachmentCommandTestTimer{ch: make(chan time.Time, 1)}
70	-	c.timers <- timer
71	-	return timer
72	-	}
73	-
74	-	func awaitAttachedCommandFrame(t testing.T, transport attachmentCommandTestTransport) ports.CommandRequest {
75	-	t.Helper()
76	-	select {
77	-	case frame := <-transport.sent:
78	-	require.Equal(t, ports.MsgCommand, frame.Type)
79	-	request, err := ports.UnmarshalCommandRequest(frame.Payload)
80	-	require.NoError(t, err)
81	-	return request
82	-	case <-time.After(time.Second):
83	-	t.Fatal("attached command was not published")
84	-	return ports.CommandRequest{}
85	-	}
86	-	}
87	-
88	-	func awaitAttachedCommandTimer(t testing.T, clock attachmentCommandTestClock) *attachmentCommandTestTimer {
89	-	t.Helper()
90	-	select {
91	-	case timer := <-clock.timers:
92	-	return timer
93	-	case <-time.After(time.Second):
94	-	t.Fatal("attached command timeout was not armed")
95	-	return nil
96	-	}
97	-	}
98	-
99	-	func awaitAttachedCommandResult(t *testing.T, done <-chan attachedCommandOutcome) attachedCommandOutcome {
100	-	t.Helper()
101	-	select {
102	-	case result := <-done:
103	-	return result
104	-	case <-time.After(time.Second):
105	-	t.Fatal("attached command did not finish")
106	-	return attachedCommandOutcome{}
107	-	}
108	-	}
109	-
110	-	func startAttachedCommand(t testing.T, ac attachedClient, clock attachmentCommandTestClock, slug string) (ports.CommandRequest, attachmentCommandTestTimer, <-chan attachedCommandOutcome) {
111	-	t.Helper()
112	-	done := make(chan attachedCommandOutcome, 1)
113	-	go func() {
114	-	result, err := ac.sendCommand(context.Background(), clock, slug, nil)
115	-	done <- attachedCommandOutcome{result: result, err: err}
116	-	}()
117	-	request := awaitAttachedCommandFrame(t, ac.transport().(*attachmentCommandTestTransport))
118	-	return request, awaitAttachedCommandTimer(t, clock), done
119	-	}
120	-
121	-	func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) {
122	-	tracker := NewCommandRequestTracker()
123	-	clock := newAttachmentCommandTestClock()
124	-	requestID, outcome := tracker.Publish(4)
125	-	waitDone := make(chan error, 1)
126	-	go func() {
127	-	_, err := tracker.Wait(context.Background(), clock, requestID, 4, outcome)
128	-	waitDone <- err
129	-	}()
130	-	timer := awaitAttachedCommandTimer(t, clock)
131	-	timer.fire()
132	-	require.ErrorIs(t, <-waitDone, ErrCommandRequestTimeout)
133	-	require.Zero(t, tracker.PendingCount())
134	-
135	-	tracker.Complete(4, ports.CommandResult{RequestID: requestID, OK: true})
136	-	secondID, secondOutcome := tracker.Publish(8)
137	-	tracker.Complete(4, ports.CommandResult{RequestID: secondID, OK: true})
138	-	select {
139	-	case <-secondOutcome:
140	-	t.Fatal("old-generation result completed a newer request")
141	-	default:
142	-	}
143	-	tracker.Complete(8, ports.CommandResult{RequestID: secondID, OK: true})
144	-	result := <-secondOutcome
145	-	require.True(t, result.Result.OK)
146	-	}
147	-
148	-	func TestAttachedCommandWaitReleasesSenderLockAfterPublication(t *testing.T) {
149	-	transport := newAttachmentCommandTestTransport()
150	-	gate := make(chan struct{})
151	-	transport.gate = gate
152	-	transport.entered = make(chan struct{}, 1)
153	-	ac := &attachedClient{tr: transport}
154	-	ac.connectionGeneration.Store(1)
155	-	clock := newAttachmentCommandTestClock()
156	-	done := make(chan attachedCommandOutcome, 1)
157	-	go func() {
158	-	result, err := ac.sendCommand(context.Background(), clock, "next-tab", nil)
159	-	done <- attachedCommandOutcome{result: result, err: err}
160	-	}()
161	-
162	-	select {
163	-	case <-transport.entered:
164	-	case <-time.After(time.Second):
165	-	t.Fatal("command publication did not reach the transport")
166	-	}
167	-	close(gate)
168	-	request := awaitAttachedCommandFrame(t, transport)
169	-	_ = awaitAttachedCommandTimer(t, clock)
170	-
171	-	require.True(t, ac.sendMu.TryLock(), "command wait retained sender lock")
172	-	ac.sendMu.Unlock()
173	-	require.NoError(t, ac.sendExpectedTransport(ac.transportSnapshot(), ports.Frame{Type: ports.MsgPing}))
174	-	select {
175	-	case frame := <-transport.sent:
176	-	require.Equal(t, ports.MsgPing, frame.Type)
177	-	case <-time.After(time.Second):
178	-	t.Fatal("ping was blocked by command wait")
179	-	}
180	-
181	-	ac.completeCommandResult(1, ports.CommandResult{RequestID: request.RequestID, OK: true})
182	-	result := awaitAttachedCommandResult(t, done)
183	-	require.NoError(t, result.err)
184	-	require.True(t, result.result.OK)
185	-	}
186	-
187	-	func TestAttachedCommandTimeoutRemovesOnlyRequestAndIgnoresLateResult(t *testing.T) {
188	-	transport := newAttachmentCommandTestTransport()
189	-	ac := &attachedClient{tr: transport}
190	-	ac.connectionGeneration.Store(4)
191	-	clock := newAttachmentCommandTestClock()
192	-
193	-	first, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
194	-	timer.fire()
195	-	result := awaitAttachedCommandResult(t, done)
196	-	require.ErrorIs(t, result.err, errAttachedCommandTimeout)
197	-	require.Zero(t, ac.pendingAttachedCommandCount())
198	-
199	-	ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
200	-	select {
201	-	case <-done:
202	-	t.Fatal("late result completed a timed-out command")
203	-	default:
204	-	}
205	-
206	-	second, secondTimer, secondDone := startAttachedCommand(t, ac, clock, "previous-tab")
207	-	require.Equal(t, first.RequestID+1, second.RequestID)
208	-	ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
209	-	select {
210	-	case <-secondDone:
211	-	t.Fatal("old request ID completed a newer command")
212	-	default:
213	-	}
214	-	ac.completeCommandResult(4, ports.CommandResult{RequestID: second.RequestID, OK: true})
215	-	result = awaitAttachedCommandResult(t, secondDone)
216	-	require.NoError(t, result.err)
217	-	require.True(t, result.result.OK)
218	-	secondTimer.fire()
219	-	}
220	-
221	-	func TestAttachedCommandIgnoresOldGenerationResults(t *testing.T) {
222	-	transport := newAttachmentCommandTestTransport()
223	-	ac := &attachedClient{tr: transport}
224	-	ac.connectionGeneration.Store(7)
225	-	clock := newAttachmentCommandTestClock()
226	-
227	-	request, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
228	-	ac.connectionGeneration.Store(8)
229	-	ac.completeCommandResult(7, ports.CommandResult{RequestID: request.RequestID, OK: true})
230	-	select {
231	-	case <-done:
232	-	t.Fatal("old-generation result completed a command")
233	-	default:
234	-	}
235	-	timer.fire()
236	-	result := awaitAttachedCommandResult(t, done)
237	-	require.ErrorIs(t, result.err, errAttachedCommandTimeout)
238	-	}
239	-
240	-	func TestAttachedCommandTrackingIsIndependentPerAttachment(t *testing.T) {
241	-	firstTransport := newAttachmentCommandTestTransport()
242	-	secondTransport := newAttachmentCommandTestTransport()
243	-	first := &attachedClient{tr: firstTransport}
244	-	second := &attachedClient{tr: secondTransport}
245	-	first.connectionGeneration.Store(1)
246	-	second.connectionGeneration.Store(1)
247	-	clock := newAttachmentCommandTestClock()
248	-
249	-	firstDone := make(chan attachedCommandOutcome, 1)
250	-	go func() {
251	-	result, err := first.sendCommand(context.Background(), clock, "next-tab", nil)
252	-	firstDone <- attachedCommandOutcome{result: result, err: err}
253	-	}()
254	-	secondDone := make(chan attachedCommandOutcome, 1)
255	-	go func() {
256	-	result, err := second.sendCommand(context.Background(), clock, "previous-tab", nil)
257	-	secondDone <- attachedCommandOutcome{result: result, err: err}
258	-	}()
259	-
260	-	firstRequest := awaitAttachedCommandFrame(t, firstTransport)
261	-	secondRequest := awaitAttachedCommandFrame(t, secondTransport)
262	-	require.Equal(t, uint64(1), firstRequest.RequestID)
263	-	require.Equal(t, uint64(1), secondRequest.RequestID)
264	-	_ = awaitAttachedCommandTimer(t, clock)
265	-	_ = awaitAttachedCommandTimer(t, clock)
266	-
267	-	first.completeCommandResult(1, ports.CommandResult{RequestID: firstRequest.RequestID, OK: true})
268	-	second.completeCommandResult(1, ports.CommandResult{RequestID: secondRequest.RequestID, OK: true})
269	-	require.NoError(t, awaitAttachedCommandResult(t, firstDone).err)
270	-	require.NoError(t, awaitAttachedCommandResult(t, secondDone).err)
271	-	require.Equal(t, 0, first.pendingAttachedCommandCount())
272	-	require.Equal(t, 0, second.pendingAttachedCommandCount())
273	-	}
274	-
275	-	func TestAttachedCommandUnavailableAndContextCancellation(t *testing.T) {
276	-	var nilAttachment *attachedClient
277	-	clock := newAttachmentCommandTestClock()
278	-	_, err := nilAttachment.sendCommand(context.Background(), clock, "next-tab", nil)
279	-	require.True(t, errors.Is(err, errAttachedCommandUnavailable))
280	-
281	-	transport := newAttachmentCommandTestTransport()
282	-	ac := &attachedClient{tr: transport}
283	-	ctx, cancel := context.WithCancel(context.Background())
284	-	cancel()
285	-	_, err = ac.sendCommand(ctx, clock, "next-tab", nil)
286	-	require.ErrorIs(t, err, context.Canceled)
287	-	}

@@ -1,287 +0,0 @@
1	-	package daemon
2	-
3	-	import (
4	-	"context"
5	-	"errors"
6	-	"sync"
7	-	"testing"
8	-	"time"
9	-
10	-	"github.com/stretchr/testify/require"
11	-
12	-	"github.com/bnema/vev/internal/ports"
13	-	)
14	-
15	-	type attachmentCommandTestTransport struct {
16	-	sent chan ports.Frame
17	-	entered chan struct{}
18	-	gate <-chan struct{}
19	-	}
20	-
21	-	func newAttachmentCommandTestTransport() *attachmentCommandTestTransport {
22	-	return &attachmentCommandTestTransport{sent: make(chan ports.Frame, 8)}
23	-	}
24	-
25	-	func (t *attachmentCommandTestTransport) Send(frame ports.Frame) error {
26	-	if t.entered != nil {
27	-	select {
28	-	case t.entered <- struct{}{}:
29	-	default:
30	-	}
31	-	}
32	-	if t.gate != nil {
33	-	<-t.gate
34	-	}
35	-	t.sent <- frame
36	-	return nil
37	-	}
38	-
39	-	func (t *attachmentCommandTestTransport) Recv() (ports.Frame, error) {
40	-	return ports.Frame{}, nil
41	-	}
42	-
43	-	func (t *attachmentCommandTestTransport) Close() error { return nil }
44	-
45	-	type attachmentCommandTestTimer struct {
46	-	ch chan time.Time
47	-	once sync.Once
48	-	}
49	-
50	-	func (t *attachmentCommandTestTimer) C() <-chan time.Time { return t.ch }
51	-	func (t *attachmentCommandTestTimer) Reset(time.Duration) bool { return false }
52	-	func (t *attachmentCommandTestTimer) Stop() bool { return true }
53	-	func (t *attachmentCommandTestTimer) fire() { t.once.Do(func() { t.ch <- time.Time{} }) }
54	-
55	-	type attachmentCommandTestClock struct {
56	-	timers chan *attachmentCommandTestTimer
57	-	}
58	-
59	-	func newAttachmentCommandTestClock() *attachmentCommandTestClock {
60	-	return &attachmentCommandTestClock{timers: make(chan *attachmentCommandTestTimer, 8)}
61	-	}
62	-
63	-	func (c *attachmentCommandTestClock) Now() time.Time { return time.Time{} }
64	-
65	-	func (c *attachmentCommandTestClock) NewTimer(delay time.Duration) ports.Timer {
66	-	if delay != attachedCommandTimeout {
67	-	return stubTimer{}
68	-	}
69	-	timer := &attachmentCommandTestTimer{ch: make(chan time.Time, 1)}
70	-	c.timers <- timer
71	-	return timer
72	-	}
73	-
74	-	func awaitAttachedCommandFrame(t testing.T, transport attachmentCommandTestTransport) ports.CommandRequest {
75	-	t.Helper()
76	-	select {
77	-	case frame := <-transport.sent:
78	-	require.Equal(t, ports.MsgCommand, frame.Type)
79	-	request, err := ports.UnmarshalCommandRequest(frame.Payload)
80	-	require.NoError(t, err)
81	-	return request
82	-	case <-time.After(time.Second):
83	-	t.Fatal("attached command was not published")
84	-	return ports.CommandRequest{}
85	-	}
86	-	}
87	-
88	-	func awaitAttachedCommandTimer(t testing.T, clock attachmentCommandTestClock) *attachmentCommandTestTimer {
89	-	t.Helper()
90	-	select {
91	-	case timer := <-clock.timers:
92	-	return timer
93	-	case <-time.After(time.Second):
94	-	t.Fatal("attached command timeout was not armed")
95	-	return nil
96	-	}
97	-	}
98	-
99	-	func awaitAttachedCommandResult(t *testing.T, done <-chan attachedCommandOutcome) attachedCommandOutcome {
100	-	t.Helper()
101	-	select {
102	-	case result := <-done:
103	-	return result
104	-	case <-time.After(time.Second):
105	-	t.Fatal("attached command did not finish")
106	-	return attachedCommandOutcome{}
107	-	}
108	-	}
109	-
110	-	func startAttachedCommand(t testing.T, ac attachedClient, clock attachmentCommandTestClock, slug string) (ports.CommandRequest, attachmentCommandTestTimer, <-chan attachedCommandOutcome) {
111	-	t.Helper()
112	-	done := make(chan attachedCommandOutcome, 1)
113	-	go func() {
114	-	result, err := ac.sendCommand(context.Background(), clock, slug, nil)
115	-	done <- attachedCommandOutcome{result: result, err: err}
116	-	}()
117	-	request := awaitAttachedCommandFrame(t, ac.transport().(*attachmentCommandTestTransport))
118	-	return request, awaitAttachedCommandTimer(t, clock), done
119	-	}
120	-
121	-	func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) {
122	-	tracker := NewCommandRequestTracker()
123	-	clock := newAttachmentCommandTestClock()
124	-	requestID, outcome := tracker.Publish(4)
125	-	waitDone := make(chan error, 1)
126	-	go func() {
127	-	_, err := tracker.Wait(context.Background(), clock, requestID, 4, outcome)
128	-	waitDone <- err
129	-	}()
130	-	timer := awaitAttachedCommandTimer(t, clock)
131	-	timer.fire()
132	-	require.ErrorIs(t, <-waitDone, ErrCommandRequestTimeout)
133	-	require.Zero(t, tracker.PendingCount())
134	-
135	-	tracker.Complete(4, ports.CommandResult{RequestID: requestID, OK: true})
136	-	secondID, secondOutcome := tracker.Publish(8)
137	-	tracker.Complete(4, ports.CommandResult{RequestID: secondID, OK: true})
138	-	select {
139	-	case <-secondOutcome:
140	-	t.Fatal("old-generation result completed a newer request")
141	-	default:
142	-	}
143	-	tracker.Complete(8, ports.CommandResult{RequestID: secondID, OK: true})
144	-	result := <-secondOutcome
145	-	require.True(t, result.Result.OK)
146	-	}
147	-
148	-	func TestAttachedCommandWaitReleasesSenderLockAfterPublication(t *testing.T) {
149	-	transport := newAttachmentCommandTestTransport()
150	-	gate := make(chan struct{})
151	-	transport.gate = gate
152	-	transport.entered = make(chan struct{}, 1)
153	-	ac := &attachedClient{tr: transport}
154	-	ac.connectionGeneration.Store(1)
155	-	clock := newAttachmentCommandTestClock()
156	-	done := make(chan attachedCommandOutcome, 1)
157	-	go func() {
158	-	result, err := ac.sendCommand(context.Background(), clock, "next-tab", nil)
159	-	done <- attachedCommandOutcome{result: result, err: err}
160	-	}()
161	-
162	-	select {
163	-	case <-transport.entered:
164	-	case <-time.After(time.Second):
165	-	t.Fatal("command publication did not reach the transport")
166	-	}
167	-	close(gate)
168	-	request := awaitAttachedCommandFrame(t, transport)
169	-	_ = awaitAttachedCommandTimer(t, clock)
170	-
171	-	require.True(t, ac.sendMu.TryLock(), "command wait retained sender lock")
172	-	ac.sendMu.Unlock()
173	-	require.NoError(t, ac.sendExpectedTransport(ac.transportSnapshot(), ports.Frame{Type: ports.MsgPing}))
174	-	select {
175	-	case frame := <-transport.sent:
176	-	require.Equal(t, ports.MsgPing, frame.Type)
177	-	case <-time.After(time.Second):
178	-	t.Fatal("ping was blocked by command wait")
179	-	}
180	-
181	-	ac.completeCommandResult(1, ports.CommandResult{RequestID: request.RequestID, OK: true})
182	-	result := awaitAttachedCommandResult(t, done)
183	-	require.NoError(t, result.err)
184	-	require.True(t, result.result.OK)
185	-	}
186	-
187	-	func TestAttachedCommandTimeoutRemovesOnlyRequestAndIgnoresLateResult(t *testing.T) {
188	-	transport := newAttachmentCommandTestTransport()
189	-	ac := &attachedClient{tr: transport}
190	-	ac.connectionGeneration.Store(4)
191	-	clock := newAttachmentCommandTestClock()
192	-
193	-	first, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
194	-	timer.fire()
195	-	result := awaitAttachedCommandResult(t, done)
196	-	require.ErrorIs(t, result.err, errAttachedCommandTimeout)
197	-	require.Zero(t, ac.pendingAttachedCommandCount())
198	-
199	-	ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
200	-	select {
201	-	case <-done:
202	-	t.Fatal("late result completed a timed-out command")
203	-	default:
204	-	}
205	-
206	-	second, secondTimer, secondDone := startAttachedCommand(t, ac, clock, "previous-tab")
207	-	require.Equal(t, first.RequestID+1, second.RequestID)
208	-	ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
209	-	select {
210	-	case <-secondDone:
211	-	t.Fatal("old request ID completed a newer command")
212	-	default:
213	-	}
214	-	ac.completeCommandResult(4, ports.CommandResult{RequestID: second.RequestID, OK: true})
215	-	result = awaitAttachedCommandResult(t, secondDone)
216	-	require.NoError(t, result.err)
217	-	require.True(t, result.result.OK)
218	-	secondTimer.fire()
219	-	}
220	-
221	-	func TestAttachedCommandIgnoresOldGenerationResults(t *testing.T) {
222	-	transport := newAttachmentCommandTestTransport()
223	-	ac := &attachedClient{tr: transport}
224	-	ac.connectionGeneration.Store(7)
225	-	clock := newAttachmentCommandTestClock()
226	-
227	-	request, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
228	-	ac.connectionGeneration.Store(8)
229	-	ac.completeCommandResult(7, ports.CommandResult{RequestID: request.RequestID, OK: true})
230	-	select {
231	-	case <-done:
232	-	t.Fatal("old-generation result completed a command")
233	-	default:
234	-	}
235	-	timer.fire()
236	-	result := awaitAttachedCommandResult(t, done)
237	-	require.ErrorIs(t, result.err, errAttachedCommandTimeout)
238	-	}
239	-
240	-	func TestAttachedCommandTrackingIsIndependentPerAttachment(t *testing.T) {
241	-	firstTransport := newAttachmentCommandTestTransport()
242	-	secondTransport := newAttachmentCommandTestTransport()
243	-	first := &attachedClient{tr: firstTransport}
244	-	second := &attachedClient{tr: secondTransport}
245	-	first.connectionGeneration.Store(1)
246	-	second.connectionGeneration.Store(1)
247	-	clock := newAttachmentCommandTestClock()
248	-
249	-	firstDone := make(chan attachedCommandOutcome, 1)
250	-	go func() {
251	-	result, err := first.sendCommand(context.Background(), clock, "next-tab", nil)
252	-	firstDone <- attachedCommandOutcome{result: result, err: err}
253	-	}()
254	-	secondDone := make(chan attachedCommandOutcome, 1)
255	-	go func() {
256	-	result, err := second.sendCommand(context.Background(), clock, "previous-tab", nil)
257	-	secondDone <- attachedCommandOutcome{result: result, err: err}
258	-	}()
259	-
260	-	firstRequest := awaitAttachedCommandFrame(t, firstTransport)
261	-	secondRequest := awaitAttachedCommandFrame(t, secondTransport)
262	-	require.Equal(t, uint64(1), firstRequest.RequestID)
263	-	require.Equal(t, uint64(1), secondRequest.RequestID)
264	-	_ = awaitAttachedCommandTimer(t, clock)
265	-	_ = awaitAttachedCommandTimer(t, clock)
266	-
267	-	first.completeCommandResult(1, ports.CommandResult{RequestID: firstRequest.RequestID, OK: true})
268	-	second.completeCommandResult(1, ports.CommandResult{RequestID: secondRequest.RequestID, OK: true})
269	-	require.NoError(t, awaitAttachedCommandResult(t, firstDone).err)
270	-	require.NoError(t, awaitAttachedCommandResult(t, secondDone).err)
271	-	require.Equal(t, 0, first.pendingAttachedCommandCount())
272	-	require.Equal(t, 0, second.pendingAttachedCommandCount())
273	-	}
274	-
275	-	func TestAttachedCommandUnavailableAndContextCancellation(t *testing.T) {
276	-	var nilAttachment *attachedClient
277	-	clock := newAttachmentCommandTestClock()
278	-	_, err := nilAttachment.sendCommand(context.Background(), clock, "next-tab", nil)
279	-	require.True(t, errors.Is(err, errAttachedCommandUnavailable))
280	-
281	-	transport := newAttachmentCommandTestTransport()
282	-	ac := &attachedClient{tr: transport}
283	-	ctx, cancel := context.WithCancel(context.Background())
284	-	cancel()
285	-	_, err = ac.sendCommand(ctx, clock, "next-tab", nil)
286	-	require.ErrorIs(t, err, context.Canceled)
287	-	}

M internal/usecase/daemon/client.go +0 -1


@@ -85,7 +85,6 @@ type attachedClient struct {
85	85		previousSession Guarded[attachmentSession]
86	86		linkMu sync.Mutex
87	87		sendMu sync.Mutex
88		-	commands attachedCommandTracker
89	88		// routeCreatedSession marks a session created by this attachment's route.
90	89		// A handshake that never commits Welcome must tear down that exact empty
91	90		// session, while an attachment routed to an existing session must not.


@@ -85,7 +85,6 @@ type attachedClient struct {
85		previousSession Guarded[attachmentSession]	85	previousSession Guarded[attachmentSession]
86		linkMu sync.Mutex	86	linkMu sync.Mutex
87		sendMu sync.Mutex	87	sendMu sync.Mutex
88	-	commands attachedCommandTracker
89		// routeCreatedSession marks a session created by this attachment's route.	88	// routeCreatedSession marks a session created by this attachment's route.
90		// A handshake that never commits Welcome must tear down that exact empty	89	// A handshake that never commits Welcome must tear down that exact empty
91		// session, while an attachment routed to an existing session must not.	90	// session, while an attachment routed to an existing session must not.

R internal/usecase/daemon/attachment_commands.go → internal/usecase/daemon/command_tracker.go +0 -89 renamed, 62% similar


@@ -159,92 +159,3 @@ func (t *CommandRequestTracker) PendingCount() int {
159	159		defer t.mu.Unlock()
160	160		return len(t.pending)
161	161		}
162		-
163		-	// attachedCommandTracker adds ordering for interactive commands while reusing
164		-	// the same generic per-connection request tracker.
165		-	type attachedCommandTracker struct {
166		-	commandMu sync.Mutex
167		-	CommandRequestTracker
168		-	}
169		-
170		-	const attachedCommandTimeout = CommandRequestTimeout
171		-
172		-	var (
173		-	errAttachedCommandTimeout = ErrCommandRequestTimeout
174		-	errAttachedCommandUnavailable = ErrCommandRequestUnavailable
175		-	)
176		-
177		-	type attachedCommandOutcome struct {
178		-	result ports.CommandResult
179		-	err error
180		-	}
181		-
182		-	// sendCommand publishes one attached command and waits for its exact result.
183		-	// commandMu keeps interactive palette commands ordered for one attachment;
184		-	// other attachment traffic uses only sendMu during request publication.
185		-	func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, slug string, args []string) (ports.CommandResult, error) {
186		-	if ac == nil \|\| clock == nil {
187		-	return ports.CommandResult{}, errAttachedCommandUnavailable
188		-	}
189		-	if ctx == nil {
190		-	ctx = context.Background()
191		-	}
192		-	if err := ctx.Err(); err != nil {
193		-	return ports.CommandResult{}, err
194		-	}
195		-
196		-	ac.commands.commandMu.Lock()
197		-	defer ac.commands.commandMu.Unlock()
198		-	if err := ctx.Err(); err != nil {
199		-	return ports.CommandResult{}, err
200		-	}
201		-
202		-	ac.sendMu.Lock()
203		-	expected := ac.transportSnapshot()
204		-	generation := ac.connectionGeneration.Load()
205		-	if expected.transport == nil {
206		-	ac.sendMu.Unlock()
207		-	return ports.CommandResult{}, errAttachedCommandUnavailable
208		-	}
209		-	requestID, outcome := ac.commands.Publish(generation)
210		-	if !ac.transportSnapshotCurrent(expected) \|\| ac.connectionGeneration.Load() != generation {
211		-	ac.commands.Remove(requestID, generation)
212		-	ac.sendMu.Unlock()
213		-	return ports.CommandResult{}, errAttachedCommandUnavailable
214		-	}
215		-
216		-	payload, err := ports.MarshalCommandRequest(ports.CommandRequest{
217		-	Version: ports.ProtocolVersion, RequestID: requestID, Attached: true,
218		-	Slug: slug, Args: append([]string(nil), args...),
219		-	})
220		-	if err != nil {
221		-	ac.commands.Remove(requestID, generation)
222		-	ac.sendMu.Unlock()
223		-	return ports.CommandResult{}, err
224		-	}
225		-	if err := expected.transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
226		-	ac.commands.Remove(requestID, generation)
227		-	ac.sendMu.Unlock()
228		-	return ports.CommandResult{}, err
229		-	}
230		-	ac.sendMu.Unlock()
231		-
232		-	return ac.commands.Wait(ctx, clock, requestID, generation, outcome)
233		-	}
234		-
235		-	// completeCommandResult accepts only a result for the exact request and
236		-	// connection generation. Unknown, late, and old-generation results are safe
237		-	// no-ops.
238		-	func (ac *attachedClient) completeCommandResult(generation uint64, result ports.CommandResult) {
239		-	if ac == nil \|\| ac.connectionGeneration.Load() != generation {
240		-	return
241		-	}
242		-	ac.commands.Complete(generation, result)
243		-	}
244		-
245		-	func (ac *attachedClient) pendingAttachedCommandCount() int {
246		-	if ac == nil {
247		-	return 0
248		-	}
249		-	return ac.commands.PendingCount()
250		-	}


@@ -159,92 +159,3 @@ func (t *CommandRequestTracker) PendingCount() int {
159		defer t.mu.Unlock()	159	defer t.mu.Unlock()
160		return len(t.pending)	160	return len(t.pending)
161		}	161	}
162	-
163	-	// attachedCommandTracker adds ordering for interactive commands while reusing
164	-	// the same generic per-connection request tracker.
165	-	type attachedCommandTracker struct {
166	-	commandMu sync.Mutex
167	-	CommandRequestTracker
168	-	}
169	-
170	-	const attachedCommandTimeout = CommandRequestTimeout
171	-
172	-	var (
173	-	errAttachedCommandTimeout = ErrCommandRequestTimeout
174	-	errAttachedCommandUnavailable = ErrCommandRequestUnavailable
175	-	)
176	-
177	-	type attachedCommandOutcome struct {
178	-	result ports.CommandResult
179	-	err error
180	-	}
181	-
182	-	// sendCommand publishes one attached command and waits for its exact result.
183	-	// commandMu keeps interactive palette commands ordered for one attachment;
184	-	// other attachment traffic uses only sendMu during request publication.
185	-	func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, slug string, args []string) (ports.CommandResult, error) {
186	-	if ac == nil \|\| clock == nil {
187	-	return ports.CommandResult{}, errAttachedCommandUnavailable
188	-	}
189	-	if ctx == nil {
190	-	ctx = context.Background()
191	-	}
192	-	if err := ctx.Err(); err != nil {
193	-	return ports.CommandResult{}, err
194	-	}
195	-
196	-	ac.commands.commandMu.Lock()
197	-	defer ac.commands.commandMu.Unlock()
198	-	if err := ctx.Err(); err != nil {
199	-	return ports.CommandResult{}, err
200	-	}
201	-
202	-	ac.sendMu.Lock()
203	-	expected := ac.transportSnapshot()
204	-	generation := ac.connectionGeneration.Load()
205	-	if expected.transport == nil {
206	-	ac.sendMu.Unlock()
207	-	return ports.CommandResult{}, errAttachedCommandUnavailable
208	-	}
209	-	requestID, outcome := ac.commands.Publish(generation)
210	-	if !ac.transportSnapshotCurrent(expected) \|\| ac.connectionGeneration.Load() != generation {
211	-	ac.commands.Remove(requestID, generation)
212	-	ac.sendMu.Unlock()
213	-	return ports.CommandResult{}, errAttachedCommandUnavailable
214	-	}
215	-
216	-	payload, err := ports.MarshalCommandRequest(ports.CommandRequest{
217	-	Version: ports.ProtocolVersion, RequestID: requestID, Attached: true,
218	-	Slug: slug, Args: append([]string(nil), args...),
219	-	})
220	-	if err != nil {
221	-	ac.commands.Remove(requestID, generation)
222	-	ac.sendMu.Unlock()
223	-	return ports.CommandResult{}, err
224	-	}
225	-	if err := expected.transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
226	-	ac.commands.Remove(requestID, generation)
227	-	ac.sendMu.Unlock()
228	-	return ports.CommandResult{}, err
229	-	}
230	-	ac.sendMu.Unlock()
231	-
232	-	return ac.commands.Wait(ctx, clock, requestID, generation, outcome)
233	-	}
234	-
235	-	// completeCommandResult accepts only a result for the exact request and
236	-	// connection generation. Unknown, late, and old-generation results are safe
237	-	// no-ops.
238	-	func (ac *attachedClient) completeCommandResult(generation uint64, result ports.CommandResult) {
239	-	if ac == nil \|\| ac.connectionGeneration.Load() != generation {
240	-	return
241	-	}
242	-	ac.commands.Complete(generation, result)
243	-	}
244	-
245	-	func (ac *attachedClient) pendingAttachedCommandCount() int {
246	-	if ac == nil {
247	-	return 0
248	-	}
249	-	return ac.commands.PendingCount()
250	-	}

A internal/usecase/daemon/command_tracker_test.go +71 -0

@@ -0,0 +1,71 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	)
	13	+
	14	+	type commandTrackerTestTimer struct {
	15	+	ch chan time.Time
	16	+	once sync.Once
	17	+	}
	18	+
	19	+	func (t *commandTrackerTestTimer) C() <-chan time.Time { return t.ch }
	20	+	func (t *commandTrackerTestTimer) Reset(time.Duration) bool { return false }
	21	+	func (t *commandTrackerTestTimer) Stop() bool { return true }
	22	+	func (t *commandTrackerTestTimer) fire() { t.once.Do(func() { t.ch <- time.Time{} }) }
	23	+
	24	+	type commandTrackerTestClock struct {
	25	+	timers chan *commandTrackerTestTimer
	26	+	delay time.Duration
	27	+	}
	28	+
	29	+	func newCommandTrackerTestClock() *commandTrackerTestClock {
	30	+	return &commandTrackerTestClock{timers: make(chan *commandTrackerTestTimer, 8)}
	31	+	}
	32	+
	33	+	func (c *commandTrackerTestClock) Now() time.Time { return time.Time{} }
	34	+
	35	+	func (c *commandTrackerTestClock) NewTimer(delay time.Duration) ports.Timer {
	36	+	c.delay = delay
	37	+	timer := &commandTrackerTestTimer{ch: make(chan time.Time, 1)}
	38	+	c.timers <- timer
	39	+	return timer
	40	+	}
	41	+
	42	+	func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) {
	43	+	tracker := NewCommandRequestTracker()
	44	+	clock := newCommandTrackerTestClock()
	45	+	requestID, outcome := tracker.Publish(4)
	46	+	waitDone := make(chan error, 1)
	47	+	go func() {
	48	+	_, err := tracker.Wait(context.Background(), clock, requestID, 4, outcome)
	49	+	waitDone <- err
	50	+	}()
	51	+
	52	+	timer := <-clock.timers
	53	+	require.Equal(t, CommandRequestTimeout, clock.delay)
	54	+	timer.fire()
	55	+	require.ErrorIs(t, <-waitDone, ErrCommandRequestTimeout)
	56	+	require.Zero(t, tracker.PendingCount())
	57	+
	58	+	// Reuse the timed-out wire ID with a new generation. A late result from the
	59	+	// old connection must not complete the new request.
	60	+	secondOutcome, ok := tracker.Track(requestID, 8)
	61	+	require.True(t, ok)
	62	+	tracker.Complete(4, ports.CommandResult{RequestID: requestID, OK: true})
	63	+	select {
	64	+	case <-secondOutcome:
	65	+	t.Fatal("late result from the old generation completed a newer request")
	66	+	default:
	67	+	}
	68	+	tracker.Complete(8, ports.CommandResult{RequestID: requestID, OK: true})
	69	+	result := <-secondOutcome
	70	+	require.True(t, result.Result.OK)
	71	+	}

@@ -0,0 +1,71 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	)
	13	+
	14	+	type commandTrackerTestTimer struct {
	15	+	ch chan time.Time
	16	+	once sync.Once
	17	+	}
	18	+
	19	+	func (t *commandTrackerTestTimer) C() <-chan time.Time { return t.ch }
	20	+	func (t *commandTrackerTestTimer) Reset(time.Duration) bool { return false }
	21	+	func (t *commandTrackerTestTimer) Stop() bool { return true }
	22	+	func (t *commandTrackerTestTimer) fire() { t.once.Do(func() { t.ch <- time.Time{} }) }
	23	+
	24	+	type commandTrackerTestClock struct {
	25	+	timers chan *commandTrackerTestTimer
	26	+	delay time.Duration
	27	+	}
	28	+
	29	+	func newCommandTrackerTestClock() *commandTrackerTestClock {
	30	+	return &commandTrackerTestClock{timers: make(chan *commandTrackerTestTimer, 8)}
	31	+	}
	32	+
	33	+	func (c *commandTrackerTestClock) Now() time.Time { return time.Time{} }
	34	+
	35	+	func (c *commandTrackerTestClock) NewTimer(delay time.Duration) ports.Timer {
	36	+	c.delay = delay
	37	+	timer := &commandTrackerTestTimer{ch: make(chan time.Time, 1)}
	38	+	c.timers <- timer
	39	+	return timer
	40	+	}
	41	+
	42	+	func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) {
	43	+	tracker := NewCommandRequestTracker()
	44	+	clock := newCommandTrackerTestClock()
	45	+	requestID, outcome := tracker.Publish(4)
	46	+	waitDone := make(chan error, 1)
	47	+	go func() {
	48	+	_, err := tracker.Wait(context.Background(), clock, requestID, 4, outcome)
	49	+	waitDone <- err
	50	+	}()
	51	+
	52	+	timer := <-clock.timers
	53	+	require.Equal(t, CommandRequestTimeout, clock.delay)
	54	+	timer.fire()
	55	+	require.ErrorIs(t, <-waitDone, ErrCommandRequestTimeout)
	56	+	require.Zero(t, tracker.PendingCount())
	57	+
	58	+	// Reuse the timed-out wire ID with a new generation. A late result from the
	59	+	// old connection must not complete the new request.
	60	+	secondOutcome, ok := tracker.Track(requestID, 8)
	61	+	require.True(t, ok)
	62	+	tracker.Complete(4, ports.CommandResult{RequestID: requestID, OK: true})
	63	+	select {
	64	+	case <-secondOutcome:
	65	+	t.Fatal("late result from the old generation completed a newer request")
	66	+	default:
	67	+	}
	68	+	tracker.Complete(8, ports.CommandResult{RequestID: requestID, OK: true})
	69	+	result := <-secondOutcome
	70	+	require.True(t, result.Result.OK)
	71	+	}

M internal/usecase/daemon/control.go +12 -1


@@ -81,13 +81,24 @@ func (d *Daemon) handleCommand(tr ports.Transport, f ports.Frame) error {
81	81		// sendCommandResult gives one-shot control responses the same observable
82	82		// transport-failure behavior regardless of which validation path produced it.
83	83		func (d *Daemon) sendCommandResult(tr ports.Transport, result ports.CommandResult) error {
84		-	if err := tr.Send(frameCommandResult(result)); err != nil {
	84	+	if err := d.boundedControlSend(tr, frameCommandResult(result)); err != nil {
85	85		d.log.Warn("command response send failed", "err", err)
86	86		return err
87	87		}
88	88		return nil
89	89		}
90	90
	91	+	// boundedControlSend keeps one-shot control handlers from waiting forever on a
	92	+	// client that stopped reading. A timeout closes the exact transport so a
	93	+	// blocked Send can unwind before the handler returns.
	94	+	func (d *Daemon) boundedControlSend(tr ports.Transport, frame ports.Frame) error {
	95	+	_, err := d.boundedSendWith(tr, func() error { return tr.Send(frame) })
	96	+	if errors.Is(err, errSendTimedOut) {
	97	+	_ = tr.Close()
	98	+	}
	99	+	return err
	100	+	}
	101	+
91	102		func frameCommandResult(result ports.CommandResult) ports.Frame {
92	103		return ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(result)}
93	104		}

 // sendCommandResult gives one-shot control responses the same observable
 // transport-failure behavior regardless of which validation path produced it.
 func (d *Daemon) sendCommandResult(tr ports.Transport, result ports.CommandResult) error {
-	if err := tr.Send(frameCommandResult(result)); err != nil {
+	if err := d.boundedControlSend(tr, frameCommandResult(result)); err != nil {
 		d.log.Warn("command response send failed", "err", err)
 		return err
 	}
 	return nil
 }
+// boundedControlSend keeps one-shot control handlers from waiting forever on a
+// client that stopped reading. A timeout closes the exact transport so a
+// blocked Send can unwind before the handler returns.
+func (d *Daemon) boundedControlSend(tr ports.Transport, frame ports.Frame) error {
+	_, err := d.boundedSendWith(tr, func() error { return tr.Send(frame) })
+	if errors.Is(err, errSendTimedOut) {
+		_ = tr.Close()
+	}
+	return err
+}
 func frameCommandResult(result ports.CommandResult) ports.Frame {
 	return ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(result)}
 }

M internal/usecase/daemon/control_test.go +51 -0


@@ -149,6 +149,30 @@ func TestHandleCommandTimesOutWithRequestGeneration(t *testing.T) {
149	149		close(factory.release)
150	150		}
151	151
	152	+	func TestHandleCommandResponseSendTimeoutClosesTransport(t *testing.T) {
	153	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	154	+	d := newTestDaemon(t, nil, clock)
	155	+	tr := newBlockingControlSendTransport()
	156	+	frame := commandFrame(t, ports.CommandRequest{RequestID: 41, Attached: true})
	157	+	done := make(chan error, 1)
	158	+	go func() { done <- d.handleCommand(tr, frame) }()
	159	+
	160	+	<-tr.started
	161	+	timer := <-clock.timers
	162	+	timer.ch <- time.Time{}
	163	+	select {
	164	+	case err := <-done:
	165	+	require.ErrorIs(t, err, errSendTimedOut)
	166	+	case <-time.After(testWaitTimeout):
	167	+	t.Fatal("bounded control response did not return")
	168	+	}
	169	+	select {
	170	+	case <-tr.closed:
	171	+	default:
	172	+	t.Fatal("timed-out control response did not close its transport")
	173	+	}
	174	+	}
	175	+
152	176		func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {
153	177		d := newTestDaemon(t, nil, stubClock{})
154	178		var logs bytes.Buffer

@@ -1297,6 +1321,33 @@ func (f *controlPTYFactory) close() {
1297	1321		}
1298	1322		}
1299	1323
	1324	+	type blockingControlSendTransport struct {
	1325	+	started chan struct{}
	1326	+	closed chan struct{}
	1327	+	once sync.Once
	1328	+	}
	1329	+
	1330	+	func newBlockingControlSendTransport() *blockingControlSendTransport {
	1331	+	return &blockingControlSendTransport{started: make(chan struct{}), closed: make(chan struct{})}
	1332	+	}
	1333	+
	1334	+	func (tr *blockingControlSendTransport) Send(ports.Frame) error {
	1335	+	tr.once.Do(func() { close(tr.started) })
	1336	+	<-tr.closed
	1337	+	return io.ErrClosedPipe
	1338	+	}
	1339	+
	1340	+	func (*blockingControlSendTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
	1341	+	func (tr *blockingControlSendTransport) Close() error {
	1342	+	tr.once.Do(func() { close(tr.started) })
	1343	+	select {
	1344	+	case <-tr.closed:
	1345	+	default:
	1346	+	close(tr.closed)
	1347	+	}
	1348	+	return nil
	1349	+	}
	1350	+
1300	1351		type commandSendErrorTransport struct {
1301	1352		err error
1302	1353		closed bool


@@ -149,6 +149,30 @@ func TestHandleCommandTimesOutWithRequestGeneration(t *testing.T) {
149	close(factory.release)	149		close(factory.release)
150	}	150		}
151		151
		152	+	func TestHandleCommandResponseSendTimeoutClosesTransport(t *testing.T) {
		153	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
		154	+	d := newTestDaemon(t, nil, clock)
		155	+	tr := newBlockingControlSendTransport()
		156	+	frame := commandFrame(t, ports.CommandRequest{RequestID: 41, Attached: true})
		157	+	done := make(chan error, 1)
		158	+	go func() { done <- d.handleCommand(tr, frame) }()
		159	+
		160	+	<-tr.started
		161	+	timer := <-clock.timers
		162	+	timer.ch <- time.Time{}
		163	+	select {
		164	+	case err := <-done:
		165	+	require.ErrorIs(t, err, errSendTimedOut)
		166	+	case <-time.After(testWaitTimeout):
		167	+	t.Fatal("bounded control response did not return")
		168	+	}
		169	+	select {
		170	+	case <-tr.closed:
		171	+	default:
		172	+	t.Fatal("timed-out control response did not close its transport")
		173	+	}
		174	+	}
		175	+
152	func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {	176		func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {
153	d := newTestDaemon(t, nil, stubClock{})	177		d := newTestDaemon(t, nil, stubClock{})
154	var logs bytes.Buffer	178		var logs bytes.Buffer

@@ -1297,6 +1321,33 @@ func (f *controlPTYFactory) close() {
1297	}	1321		}
1298	}	1322		}
1299		1323
		1324	+	type blockingControlSendTransport struct {
		1325	+	started chan struct{}
		1326	+	closed chan struct{}
		1327	+	once sync.Once
		1328	+	}
		1329	+
		1330	+	func newBlockingControlSendTransport() *blockingControlSendTransport {
		1331	+	return &blockingControlSendTransport{started: make(chan struct{}), closed: make(chan struct{})}
		1332	+	}
		1333	+
		1334	+	func (tr *blockingControlSendTransport) Send(ports.Frame) error {
		1335	+	tr.once.Do(func() { close(tr.started) })
		1336	+	<-tr.closed
		1337	+	return io.ErrClosedPipe
		1338	+	}
		1339	+
		1340	+	func (*blockingControlSendTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
		1341	+	func (tr *blockingControlSendTransport) Close() error {
		1342	+	tr.once.Do(func() { close(tr.started) })
		1343	+	select {
		1344	+	case <-tr.closed:
		1345	+	default:
		1346	+	close(tr.closed)
		1347	+	}
		1348	+	return nil
		1349	+	}
		1350	+
1300	type commandSendErrorTransport struct {	1351		type commandSendErrorTransport struct {
1301	err error	1352		err error
1302	closed bool	1353		closed bool

M internal/usecase/daemon/daemon.go +5 -5


@@ -1034,7 +1034,7 @@ func (d *Daemon) handleList(tr ports.Transport) {
1034	1034		d.mu.Unlock()
1035	1035
1036	1036		sort.Slice(infos, func(i, j int) bool { return infos[i].Name < infos[j].Name })
1037		-	_ = tr.Send(frameSessions(infos))
	1037	+	_ = d.boundedControlSend(tr, frameSessions(infos))
1038	1038		}
1039	1039
1040	1040		// handleKill terminates the requested live session or stopped named session,

@@ -1045,7 +1045,7 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1045	1045
1046	1046		k, err := ports.UnmarshalKill(f.Payload)
1047	1047		if err != nil {
1048		-	_ = tr.Send(frameError(ports.ErrInternal, "malformed kill request"))
	1048	+	_ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "malformed kill request"))
1049	1049		return
1050	1050		}
1051	1051		if k.All {

@@ -1062,7 +1062,7 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1062	1062		// deletion order as live and offline purges.
1063	1063		if err := d.retryStoppedPurge(k.Name); err != nil {
1064	1064		d.log.Warn("deleting stopped session failed", "err", err, "session", k.Name)
1065		-	_ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session failed"))
	1065	+	_ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting stopped session failed"))
1066	1066		}
1067	1067		return
1068	1068		}

@@ -1070,11 +1070,11 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1070	1070		d.mu.Unlock()
1071	1071
1072	1072		if target == nil {
1073		-	_ = tr.Send(frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))
	1073	+	_ = d.boundedControlSend(tr, frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))
1074	1074		return
1075	1075		}
1076	1076		if err := d.killSession(target, ports.ReasonSessionKilled, true); err != nil {
1077		-	_ = tr.Send(frameError(ports.ErrInternal, "deleting persisted session failed"))
	1077	+	_ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting persisted session failed"))
1078	1078		}
1079	1079		}
1080	1080


@@ -1034,7 +1034,7 @@ func (d *Daemon) handleList(tr ports.Transport) {
1034		d.mu.Unlock()	1034		d.mu.Unlock()
1035			1035
1036		sort.Slice(infos, func(i, j int) bool { return infos[i].Name < infos[j].Name })	1036		sort.Slice(infos, func(i, j int) bool { return infos[i].Name < infos[j].Name })
1037	-	_ = tr.Send(frameSessions(infos))	1037	+	_ = d.boundedControlSend(tr, frameSessions(infos))
1038		}	1038		}
1039			1039
1040		// handleKill terminates the requested live session or stopped named session,	1040		// handleKill terminates the requested live session or stopped named session,

@@ -1045,7 +1045,7 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1045			1045
1046		k, err := ports.UnmarshalKill(f.Payload)	1046		k, err := ports.UnmarshalKill(f.Payload)
1047		if err != nil {	1047		if err != nil {
1048	-	_ = tr.Send(frameError(ports.ErrInternal, "malformed kill request"))	1048	+	_ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "malformed kill request"))
1049		return	1049		return
1050		}	1050		}
1051		if k.All {	1051		if k.All {

@@ -1062,7 +1062,7 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1062		// deletion order as live and offline purges.	1062		// deletion order as live and offline purges.
1063		if err := d.retryStoppedPurge(k.Name); err != nil {	1063		if err := d.retryStoppedPurge(k.Name); err != nil {
1064		d.log.Warn("deleting stopped session failed", "err", err, "session", k.Name)	1064		d.log.Warn("deleting stopped session failed", "err", err, "session", k.Name)
1065	-	_ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session failed"))	1065	+	_ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting stopped session failed"))
1066		}	1066		}
1067		return	1067		return
1068		}	1068		}

@@ -1070,11 +1070,11 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1070		d.mu.Unlock()	1070		d.mu.Unlock()
1071			1071
1072		if target == nil {	1072		if target == nil {
1073	-	_ = tr.Send(frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))	1073	+	_ = d.boundedControlSend(tr, frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))
1074		return	1074		return
1075		}	1075		}
1076		if err := d.killSession(target, ports.ReasonSessionKilled, true); err != nil {	1076		if err := d.killSession(target, ports.ReasonSessionKilled, true); err != nil {
1077	-	_ = tr.Send(frameError(ports.ErrInternal, "deleting persisted session failed"))	1077	+	_ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting persisted session failed"))
1078		}	1078		}
1079		}	1079		}
1080			1080

M internal/usecase/daemon/handshake.go +4 -4


@@ -83,15 +83,15 @@ func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, o
83	83		go func() { completed <- operation() }()
84	84		select {
85	85		case err := <-completed:
86		-	if ctx.Err() != nil {
	86	+	if ctxErr := ctx.Err(); ctxErr != nil {
87	87		_ = transport.Close()
88		-	<-completed
89		-	return ctx.Err()
	88	+	return ctxErr
90	89		}
91	90		return err
92	91		case <-ctx.Done():
93	92		_ = transport.Close()
94		-	<-completed
	93	+	// The result channel is buffered, so the operation can publish its
	94	+	// completion after Close without keeping this cancellation path stuck.
95	95		return ctx.Err()
96	96		}
97	97		}

 	go func() { completed <- operation() }()
 	select {
 	case err := <-completed:
-		if ctx.Err() != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
 			_ = transport.Close()
-			<-completed
+			return ctxErr
-			return ctx.Err()
 		}
 		return err
 	case <-ctx.Done():
 		_ = transport.Close()
-		<-completed
+		// The result channel is buffered, so the operation can publish its
+		// completion after Close without keeping this cancellation path stuck.
 		return ctx.Err()
 	}
 }

M internal/usecase/daemon/handshake_test.go +60 -0


@@ -47,6 +47,66 @@ func (t *handshakeBlockingTransport) Close() error {
47	47		return nil
48	48		}
49	49
	50	+	func TestBoundedHandshakeCompletionCancellationDoesNotDoubleReceive(t *testing.T) {
	51	+	ctx := &completionCancellationContext{}
	52	+	tr := &closeTrackingTransport{}
	53	+	done := make(chan error, 1)
	54	+	go func() {
	55	+	done <- boundedHandshakeOperation(ctx, tr, func() error {
	56	+	ctx.canceled = true
	57	+	return nil
	58	+	})
	59	+	}()
	60	+
	61	+	select {
	62	+	case err := <-done:
	63	+	require.ErrorIs(t, err, context.Canceled)
	64	+	case <-time.After(testWaitTimeout):
	65	+	t.Fatal("handshake completion/cancellation race deadlocked")
	66	+	}
	67	+	}
	68	+
	69	+	type completionCancellationContext struct{ canceled bool }
	70	+
	71	+	func (c *completionCancellationContext) Deadline() (time.Time, bool) { return time.Time{}, false }
	72	+	func (c *completionCancellationContext) Done() <-chan struct{} { return nil }
	73	+	func (c *completionCancellationContext) Err() error {
	74	+	if c.canceled {
	75	+	return context.Canceled
	76	+	}
	77	+	return nil
	78	+	}
	79	+	func (*completionCancellationContext) Value(any) any { return nil }
	80	+
	81	+	func TestBoundedHandshakeCancellationDoesNotWaitForUninterruptibleOperation(t *testing.T) {
	82	+	ctx, cancel := context.WithCancel(context.Background())
	83	+	defer cancel()
	84	+	tr := &closeTrackingTransport{}
	85	+	started := make(chan struct{})
	86	+	release := make(chan struct{})
	87	+	operationDone := make(chan struct{})
	88	+	result := make(chan error, 1)
	89	+	go func() {
	90	+	result <- boundedHandshakeOperation(ctx, tr, func() error {
	91	+	close(started)
	92	+	<-release
	93	+	close(operationDone)
	94	+	return nil
	95	+	})
	96	+	}()
	97	+
	98	+	<-started
	99	+	cancel()
	100	+	select {
	101	+	case err := <-result:
	102	+	require.ErrorIs(t, err, context.Canceled)
	103	+	case <-time.After(testWaitTimeout):
	104	+	t.Fatal("handshake cancellation waited for the transport operation")
	105	+	}
	106	+	close(release)
	107	+	awaitTestCompletion(t, operationDone, "handshake operation did not finish")
	108	+	}
	109	+
50	110		func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) {
51	111		clock := &signalClock{timers: make(chan *signalTimer, 1)}
52	112		d := newTestDaemon(t, nil, clock)


@@ -47,6 +47,66 @@ func (t *handshakeBlockingTransport) Close() error {
47	return nil	47		return nil
48	}	48		}
49		49
		50	+	func TestBoundedHandshakeCompletionCancellationDoesNotDoubleReceive(t *testing.T) {
		51	+	ctx := &completionCancellationContext{}
		52	+	tr := &closeTrackingTransport{}
		53	+	done := make(chan error, 1)
		54	+	go func() {
		55	+	done <- boundedHandshakeOperation(ctx, tr, func() error {
		56	+	ctx.canceled = true
		57	+	return nil
		58	+	})
		59	+	}()
		60	+
		61	+	select {
		62	+	case err := <-done:
		63	+	require.ErrorIs(t, err, context.Canceled)
		64	+	case <-time.After(testWaitTimeout):
		65	+	t.Fatal("handshake completion/cancellation race deadlocked")
		66	+	}
		67	+	}
		68	+
		69	+	type completionCancellationContext struct{ canceled bool }
		70	+
		71	+	func (c *completionCancellationContext) Deadline() (time.Time, bool) { return time.Time{}, false }
		72	+	func (c *completionCancellationContext) Done() <-chan struct{} { return nil }
		73	+	func (c *completionCancellationContext) Err() error {
		74	+	if c.canceled {
		75	+	return context.Canceled
		76	+	}
		77	+	return nil
		78	+	}
		79	+	func (*completionCancellationContext) Value(any) any { return nil }
		80	+
		81	+	func TestBoundedHandshakeCancellationDoesNotWaitForUninterruptibleOperation(t *testing.T) {
		82	+	ctx, cancel := context.WithCancel(context.Background())
		83	+	defer cancel()
		84	+	tr := &closeTrackingTransport{}
		85	+	started := make(chan struct{})
		86	+	release := make(chan struct{})
		87	+	operationDone := make(chan struct{})
		88	+	result := make(chan error, 1)
		89	+	go func() {
		90	+	result <- boundedHandshakeOperation(ctx, tr, func() error {
		91	+	close(started)
		92	+	<-release
		93	+	close(operationDone)
		94	+	return nil
		95	+	})
		96	+	}()
		97	+
		98	+	<-started
		99	+	cancel()
		100	+	select {
		101	+	case err := <-result:
		102	+	require.ErrorIs(t, err, context.Canceled)
		103	+	case <-time.After(testWaitTimeout):
		104	+	t.Fatal("handshake cancellation waited for the transport operation")
		105	+	}
		106	+	close(release)
		107	+	awaitTestCompletion(t, operationDone, "handshake operation did not finish")
		108	+	}
		109	+
50	func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) {	110		func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) {
51	clock := &signalClock{timers: make(chan *signalTimer, 1)}	111		clock := &signalClock{timers: make(chan *signalTimer, 1)}
52	d := newTestDaemon(t, nil, clock)	112		d := newTestDaemon(t, nil, clock)