Merge pull request #87 from bnema/fix/next-review-followups

fix(runtime): address blocking review findings

commit: 93b77cacae9dbafd30c3dcc748aadd6eb0f475c4
author: bnema · 3mo ago · 2026-07-14 09:01 UTC
committer: GitHub · 3mo ago · 2026-07-14 09:01 UTC
merge of: bb5f6e1 + 6eb7ce5 · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

5 files · +330 -22

M cmd/vev-perf-harness/options_manifest.go +26 -3


@@ -10,6 +10,24 @@ import (
10	10		"path/filepath"
11	11		)
12	12
	13	+	var (
	14	+	canonicalTopologyIDs = []string{"1x4", "4x1", "4x4", "8x1"}
	15	+	canonicalTopologySet = stringSet(canonicalTopologyIDs)
	16	+	canonicalWorkloadIDs = []string{
	17	+	"idle", "active_output", "all_output", "inactive_output", "interactive_flood",
	18	+	"copy_search", "resize_sweep", "snapshot_output_resize", "attach_restore_tab_switch",
	19	+	}
	20	+	canonicalWorkloadSet = stringSet(canonicalWorkloadIDs)
	21	+	)
	22	+
	23	+	func stringSet(ids []string) map[string]bool {
	24	+	set := make(map[string]bool, len(ids))
	25	+	for _, id := range ids {
	26	+	set[id] = true
	27	+	}
	28	+	return set
	29	+	}
	30	+
13	31		func parseOptions(args []string) (options, error) {
14	32		var o options
15	33		fs := flag.NewFlagSet("vev-perf-harness", flag.ContinueOnError)

@@ -93,13 +111,15 @@ func validateManifest(m manifest) error {
93	111		if tops[t.ID] {
94	112		return fmt.Errorf("duplicate topology id %q", t.ID)
95	113		}
	114	+	if !canonicalTopologySet[t.ID] {
	115	+	return fmt.Errorf("unknown topology id %q", t.ID)
	116	+	}
96	117		if t.Geometry != "120x40" \|\| t.RowsPerPane != 10000 {
97	118		return fmt.Errorf("invalid topology %q", t.ID)
98	119		}
99	120		tops[t.ID] = true
100	121		}
101		-	wantT := []string{"1x4", "4x1", "4x4", "8x1"}
102		-	for _, v := range wantT {
	122	+	for _, v := range canonicalTopologyIDs {
103	123		if !tops[v] {
104	124		return fmt.Errorf("missing canonical topology %s", v)
105	125		}

@@ -112,9 +132,12 @@ func validateManifest(m manifest) error {
112	132		if works[workload] {
113	133		return fmt.Errorf("duplicate workload id %q", workload)
114	134		}
	135	+	if !canonicalWorkloadSet[workload] {
	136	+	return fmt.Errorf("unknown workload id %q", workload)
	137	+	}
115	138		works[workload] = true
116	139		}
117		-	for _, v := range []string{"idle", "active_output", "all_output", "inactive_output", "interactive_flood", "copy_search", "resize_sweep", "snapshot_output_resize", "attach_restore_tab_switch"} {
	140	+	for _, v := range canonicalWorkloadIDs {
118	141		if !works[v] {
119	142		return fmt.Errorf("missing canonical workload %s", v)
120	143		}

 	"path/filepath"
 )
+var (
+	canonicalTopologyIDs = []string{"1x4", "4x1", "4x4", "8x1"}
+	canonicalTopologySet = stringSet(canonicalTopologyIDs)
+	canonicalWorkloadIDs = []string{
+		"idle", "active_output", "all_output", "inactive_output", "interactive_flood",
+		"copy_search", "resize_sweep", "snapshot_output_resize", "attach_restore_tab_switch",
+	}
+	canonicalWorkloadSet = stringSet(canonicalWorkloadIDs)
+)
+func stringSet(ids []string) map[string]bool {
+	set := make(map[string]bool, len(ids))
+	for _, id := range ids {
+		set[id] = true
+	}
+	return set
+}
 func parseOptions(args []string) (options, error) {
 	var o options
 	fs := flag.NewFlagSet("vev-perf-harness", flag.ContinueOnError)
 		if tops[t.ID] {
 			return fmt.Errorf("duplicate topology id %q", t.ID)
 		}
+		if !canonicalTopologySet[t.ID] {
+			return fmt.Errorf("unknown topology id %q", t.ID)
+		}
 		if t.Geometry != "120x40" || t.RowsPerPane != 10000 {
 			return fmt.Errorf("invalid topology %q", t.ID)
 		}
 		tops[t.ID] = true
 	}
-	wantT := []string{"1x4", "4x1", "4x4", "8x1"}
+	for _, v := range canonicalTopologyIDs {
-	for _, v := range wantT {
 		if !tops[v] {
 			return fmt.Errorf("missing canonical topology %s", v)
 		}
 		if works[workload] {
 			return fmt.Errorf("duplicate workload id %q", workload)
 		}
+		if !canonicalWorkloadSet[workload] {
+			return fmt.Errorf("unknown workload id %q", workload)
+		}
 		works[workload] = true
 	}
-	for _, v := range []string{"idle", "active_output", "all_output", "inactive_output", "interactive_flood", "copy_search", "resize_sweep", "snapshot_output_resize", "attach_restore_tab_switch"} {
+	for _, v := range canonicalWorkloadIDs {
 		if !works[v] {
 			return fmt.Errorf("missing canonical workload %s", v)
 		}

M cmd/vev-perf-harness/options_manifest_test.go +48 -0


@@ -149,6 +149,54 @@ func TestValidateManifestRejectsEmptyAndDuplicateMatrixIDs(t *testing.T) {
149	149		}
150	150		}
151	151
	152	+	func TestValidateManifestRejectsFullyCoveredUnknownMatrixIDs(t *testing.T) {
	153	+	canonical, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
	154	+	if err != nil {
	155	+	t.Fatal(err)
	156	+	}
	157	+	for _, tc := range []struct {
	158	+	name string
	159	+	mutate func(*manifest)
	160	+	want string
	161	+	}{
	162	+	{
	163	+	name: "topology",
	164	+	mutate: func(m *manifest) {
	165	+	m.Topologies = append(m.Topologies, topology{ID: "9x9", Geometry: "120x40", RowsPerPane: 10000})
	166	+	for _, s := range canonical.Scenarios {
	167	+	s.ID = "9x9-" + s.ID
	168	+	s.Topology = "9x9"
	169	+	m.Scenarios = append(m.Scenarios, s)
	170	+	}
	171	+	},
	172	+	want: "unknown topology id \"9x9\"",
	173	+	},
	174	+	{
	175	+	name: "workload",
	176	+	mutate: func(m *manifest) {
	177	+	m.Workloads = append(m.Workloads, "unknown_workload")
	178	+	for _, s := range canonical.Scenarios {
	179	+	s.ID = "unknown-workload-" + s.ID
	180	+	s.Workload = "unknown_workload"
	181	+	m.Scenarios = append(m.Scenarios, s)
	182	+	}
	183	+	},
	184	+	want: "unknown workload id \"unknown_workload\"",
	185	+	},
	186	+	} {
	187	+	t.Run(tc.name, func(t *testing.T) {
	188	+	m := canonical
	189	+	m.Topologies = append([]topology(nil), canonical.Topologies...)
	190	+	m.Workloads = append([]string(nil), canonical.Workloads...)
	191	+	m.Scenarios = append([]scenario(nil), canonical.Scenarios...)
	192	+	tc.mutate(&m)
	193	+	if err := validateManifest(m); err == nil \|\| !strings.Contains(err.Error(), tc.want) {
	194	+	t.Fatalf("validateManifest() error = %v, want %q", err, tc.want)
	195	+	}
	196	+	})
	197	+	}
	198	+	}
	199	+
152	200		func TestHarnessManifestCoversCanonicalMatrix(t *testing.T) {
153	201		m, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
154	202		if err != nil {


@@ -149,6 +149,54 @@ func TestValidateManifestRejectsEmptyAndDuplicateMatrixIDs(t *testing.T) {
149	}	149		}
150	}	150		}
151		151
		152	+	func TestValidateManifestRejectsFullyCoveredUnknownMatrixIDs(t *testing.T) {
		153	+	canonical, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
		154	+	if err != nil {
		155	+	t.Fatal(err)
		156	+	}
		157	+	for _, tc := range []struct {
		158	+	name string
		159	+	mutate func(*manifest)
		160	+	want string
		161	+	}{
		162	+	{
		163	+	name: "topology",
		164	+	mutate: func(m *manifest) {
		165	+	m.Topologies = append(m.Topologies, topology{ID: "9x9", Geometry: "120x40", RowsPerPane: 10000})
		166	+	for _, s := range canonical.Scenarios {
		167	+	s.ID = "9x9-" + s.ID
		168	+	s.Topology = "9x9"
		169	+	m.Scenarios = append(m.Scenarios, s)
		170	+	}
		171	+	},
		172	+	want: "unknown topology id \"9x9\"",
		173	+	},
		174	+	{
		175	+	name: "workload",
		176	+	mutate: func(m *manifest) {
		177	+	m.Workloads = append(m.Workloads, "unknown_workload")
		178	+	for _, s := range canonical.Scenarios {
		179	+	s.ID = "unknown-workload-" + s.ID
		180	+	s.Workload = "unknown_workload"
		181	+	m.Scenarios = append(m.Scenarios, s)
		182	+	}
		183	+	},
		184	+	want: "unknown workload id \"unknown_workload\"",
		185	+	},
		186	+	} {
		187	+	t.Run(tc.name, func(t *testing.T) {
		188	+	m := canonical
		189	+	m.Topologies = append([]topology(nil), canonical.Topologies...)
		190	+	m.Workloads = append([]string(nil), canonical.Workloads...)
		191	+	m.Scenarios = append([]scenario(nil), canonical.Scenarios...)
		192	+	tc.mutate(&m)
		193	+	if err := validateManifest(m); err == nil \|\| !strings.Contains(err.Error(), tc.want) {
		194	+	t.Fatalf("validateManifest() error = %v, want %q", err, tc.want)
		195	+	}
		196	+	})
		197	+	}
		198	+	}
		199	+
152	func TestHarnessManifestCoversCanonicalMatrix(t *testing.T) {	200		func TestHarnessManifestCoversCanonicalMatrix(t *testing.T) {
153	m, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))	201		m, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
154	if err != nil {	202		if err != nil {

M internal/adapters/ipc/transport.go +59 -12


@@ -63,15 +63,18 @@ type unixTransport struct {
63	63		observedClosing bool
64	64		operationsDone chan struct{}
65	65
66		-	// egressMu synchronizes Close with nonblocking enqueue. Once closed, no
67		-	// request can be left behind after the writer drains the queue.
68		-	egressMu sync.Mutex
69		-	egressClosing bool
70		-	egress chan sendRequest
71		-	done chan struct{}
72		-	writerDone chan struct{}
73		-	closeOnce sync.Once
74		-	closeErr error
	66	+	// egressMu synchronizes Close with enqueue. Synchronous enqueuers are
	67	+	// tracked separately so the writer cannot drain the queue until each has
	68	+	// observed closure or finished enqueueing.
	69	+	egressMu sync.Mutex
	70	+	egressClosing bool
	71	+	egressSenders int
	72	+	egressSendersDone chan struct{}
	73	+	egress chan sendRequest
	74	+	done chan struct{}
	75	+	writerDone chan struct{}
	76	+	closeOnce sync.Once
	77	+	closeErr error
75	78
76	79		// readBuf is reused across Recv calls (grow-once strategy): it grows
77	80		// to fit the largest frame seen so far and is never shrunk. The Frame

@@ -112,7 +115,7 @@ func (t *unixTransport) Send(f ports.Frame) error {
112	115		}
113	116		end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
114	117		result := make(chan error, 1)
115		-	if err := t.enqueue(sendRequest{data: data, done: result, end: end}); err != nil {
	118	+	if err := t.enqueueWait(sendRequest{data: data, done: result, end: end}); err != nil {
116	119		end(false)
117	120		return err
118	121		}

@@ -132,7 +135,7 @@ func (t *unixTransport) SendAsync(f ports.Frame) error {
132	135		return err
133	136		}
134	137		end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
135		-	if err := t.enqueue(sendRequest{data: data, end: end}); err != nil {
	138	+	if err := t.enqueueAsync(sendRequest{data: data, end: end}); err != nil {
136	139		end(false)
137	140		return err
138	141		}

@@ -151,7 +154,32 @@ func marshalFrame(f ports.Frame) ([]byte, error) {
151	154		return buf, nil
152	155		}
153	156
154		-	func (t *unixTransport) enqueue(req sendRequest) error {
	157	+	// enqueueWait waits for space in the egress queue. Close wakes waiters and
	158	+	// the writer waits for all in-flight enqueues before draining queued requests.
	159	+	func (t *unixTransport) enqueueWait(req sendRequest) error {
	160	+	t.egressMu.Lock()
	161	+	if t.egressClosing {
	162	+	t.egressMu.Unlock()
	163	+	return errClosed
	164	+	}
	165	+	if t.egressSenders == 0 {
	166	+	t.egressSendersDone = make(chan struct{})
	167	+	}
	168	+	t.egressSenders++
	169	+	t.egressMu.Unlock()
	170	+
	171	+	select {
	172	+	case <-t.done:
	173	+	t.finishEgressSender()
	174	+	return errClosed
	175	+	case t.egress <- req:
	176	+	t.finishEgressSender()
	177	+	return nil
	178	+	}
	179	+	}
	180	+
	181	+	// enqueueAsync performs a bounded, nonblocking enqueue for SendAsync.
	182	+	func (t *unixTransport) enqueueAsync(req sendRequest) error {
155	183		t.egressMu.Lock()
156	184		defer t.egressMu.Unlock()
157	185		if t.egressClosing {

@@ -165,6 +193,24 @@ func (t *unixTransport) enqueue(req sendRequest) error {
165	193		}
166	194		}
167	195
	196	+	func (t *unixTransport) finishEgressSender() {
	197	+	t.egressMu.Lock()
	198	+	defer t.egressMu.Unlock()
	199	+	t.egressSenders--
	200	+	if t.egressSenders == 0 {
	201	+	close(t.egressSendersDone)
	202	+	}
	203	+	}
	204	+
	205	+	func (t *unixTransport) waitForEgressSenders() {
	206	+	t.egressMu.Lock()
	207	+	done := t.egressSendersDone
	208	+	t.egressMu.Unlock()
	209	+	if done != nil {
	210	+	<-done
	211	+	}
	212	+	}
	213	+
168	214		func (t *unixTransport) writeLoop() {
169	215		defer close(t.writerDone)
170	216		for {

@@ -172,6 +218,7 @@ func (t *unixTransport) writeLoop() {
172	218		case req := <-t.egress:
173	219		t.completeWrite(req, writeAll(t.conn, req.data))
174	220		case <-t.done:
	221	+	t.waitForEgressSenders()
175	222		t.egressMu.Lock()
176	223		for {
177	224		select {

 	observedClosing bool
 	operationsDone  chan struct{}
-	// egressMu synchronizes Close with nonblocking enqueue. Once closed, no
+	// egressMu synchronizes Close with enqueue. Synchronous enqueuers are
-	// request can be left behind after the writer drains the queue.
+	// tracked separately so the writer cannot drain the queue until each has
-	egressMu      sync.Mutex
+	// observed closure or finished enqueueing.
-	egressClosing bool
+	egressMu          sync.Mutex
-	egress        chan sendRequest
+	egressClosing     bool
-	done          chan struct{}
+	egressSenders     int
-	writerDone    chan struct{}
+	egressSendersDone chan struct{}
-	closeOnce     sync.Once
+	egress            chan sendRequest
-	closeErr      error
+	done              chan struct{}
+	writerDone        chan struct{}
+	closeOnce         sync.Once
+	closeErr          error
 	// readBuf is reused across Recv calls (grow-once strategy): it grows
 	// to fit the largest frame seen so far and is never shrunk. The Frame
 	}
 	end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
 	result := make(chan error, 1)
-	if err := t.enqueue(sendRequest{data: data, done: result, end: end}); err != nil {
+	if err := t.enqueueWait(sendRequest{data: data, done: result, end: end}); err != nil {
 		end(false)
 		return err
 	}
 		return err
 	}
 	end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
-	if err := t.enqueue(sendRequest{data: data, end: end}); err != nil {
+	if err := t.enqueueAsync(sendRequest{data: data, end: end}); err != nil {
 		end(false)
 		return err
 	}
 	return buf, nil
 }
-func (t *unixTransport) enqueue(req sendRequest) error {
+// enqueueWait waits for space in the egress queue. Close wakes waiters and
+// the writer waits for all in-flight enqueues before draining queued requests.
+func (t *unixTransport) enqueueWait(req sendRequest) error {
+	t.egressMu.Lock()
+	if t.egressClosing {
+		t.egressMu.Unlock()
+		return errClosed
+	}
+	if t.egressSenders == 0 {
+		t.egressSendersDone = make(chan struct{})
+	}
+	t.egressSenders++
+	t.egressMu.Unlock()
+	select {
+	case <-t.done:
+		t.finishEgressSender()
+		return errClosed
+	case t.egress <- req:
+		t.finishEgressSender()
+		return nil
+	}
+}
+// enqueueAsync performs a bounded, nonblocking enqueue for SendAsync.
+func (t *unixTransport) enqueueAsync(req sendRequest) error {
 	t.egressMu.Lock()
 	defer t.egressMu.Unlock()
 	if t.egressClosing {
 	}
 }
+func (t *unixTransport) finishEgressSender() {
+	t.egressMu.Lock()
+	defer t.egressMu.Unlock()
+	t.egressSenders--
+	if t.egressSenders == 0 {
+		close(t.egressSendersDone)
+	}
+}
+func (t *unixTransport) waitForEgressSenders() {
+	t.egressMu.Lock()
+	done := t.egressSendersDone
+	t.egressMu.Unlock()
+	if done != nil {
+		<-done
+	}
+}
 func (t *unixTransport) writeLoop() {
 	defer close(t.writerDone)
 	for {
 		case req := <-t.egress:
 			t.completeWrite(req, writeAll(t.conn, req.data))
 		case <-t.done:
+			t.waitForEgressSenders()
 			t.egressMu.Lock()
 			for {
 				select {

M internal/adapters/ipc/transport_test.go +190 -6


@@ -6,6 +6,7 @@ import (
6	6		"io"
7	7		"net"
8	8		"reflect"
	9	+	"runtime"
9	10		"sync"
10	11		"testing"
11	12		"time"

@@ -234,12 +235,22 @@ func TestTransportSendOversizePayloadRejected(t *testing.T) {
234	235		}
235	236
236	237		func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {
	238	+	const timeout = time.Second
	239	+
237	240		serverConn, clientConn := net.Pipe()
238	241		defer func() { _ = serverConn.Close() }()
239	242		defer func() { _ = clientConn.Close() }()
	243	+	if err := serverConn.SetDeadline(time.Now().Add(timeout)); err != nil {
	244	+	t.Fatalf("SetDeadline server connection: %v", err)
	245	+	}
	246	+	if err := clientConn.SetDeadline(time.Now().Add(timeout)); err != nil {
	247	+	t.Fatalf("SetDeadline client connection: %v", err)
	248	+	}
240	249
241	250		server := NewTransport(serverConn)
	251	+	defer closeTransport(t, server)
242	252		client := NewTransport(clientConn)
	253	+	defer closeTransport(t, client)
243	254		async, ok := server.(ports.AsyncTransport)
244	255		if !ok {
245	256		t.Fatal("IPC transport does not implement AsyncTransport")

@@ -255,8 +266,13 @@ func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {
255	266		if !reflect.DeepEqual(got, welcome) {
256	267		t.Fatalf("welcome = %#v, want %#v", got, welcome)
257	268		}
258		-	if err := <-sent; err != nil {
259		-	t.Fatalf("Send welcome: %v", err)
	269	+	select {
	270	+	case err := <-sent:
	271	+	if err != nil {
	272	+	t.Fatalf("Send welcome: %v", err)
	273	+	}
	274	+	case <-time.After(timeout):
	275	+	t.Fatal("Send welcome did not return")
260	276		}
261	277
262	278		payload := []byte("first")

@@ -281,11 +297,181 @@ func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {
281	297		}
282	298		}
283	299
	300	+	func TestTransportSendWaitsForEgressCapacity(t *testing.T) {
	301	+	const timeout = time.Second
	302	+
	303	+	serverConn, clientConn := net.Pipe()
	304	+	defer func() { _ = serverConn.Close() }()
	305	+	defer func() { _ = clientConn.Close() }()
	306	+	if err := clientConn.SetDeadline(time.Now().Add(timeout)); err != nil {
	307	+	t.Fatalf("SetDeadline client connection: %v", err)
	308	+	}
	309	+
	310	+	server := NewTransport(serverConn)
	311	+	defer closeTransport(t, server)
	312	+	transport, ok := server.(*unixTransport)
	313	+	if !ok {
	314	+	t.Fatal("IPC transport is not a unixTransport")
	315	+	}
	316	+	client := NewTransport(clientConn)
	317	+	defer closeTransport(t, client)
	318	+	async, ok := server.(ports.AsyncTransport)
	319	+	if !ok {
	320	+	t.Fatal("IPC transport does not implement AsyncTransport")
	321	+	}
	322	+
	323	+	queued := fillAsyncEgress(t, transport, async, timeout)
	324	+	want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("synchronous")}
	325	+	sent := make(chan error, 1)
	326	+	go func() { sent <- server.Send(want) }()
	327	+	waitForEgressSender(t, transport, timeout)
	328	+
	329	+	select {
	330	+	case err := <-sent:
	331	+	t.Fatalf("Send returned before capacity was available: %v", err)
	332	+	default:
	333	+	}
	334	+
	335	+	for _, want := range append(queued, want) {
	336	+	got, err := client.Recv()
	337	+	if err != nil {
	338	+	t.Fatalf("Recv: %v", err)
	339	+	}
	340	+	if !reflect.DeepEqual(got, want) {
	341	+	t.Fatalf("Recv = %#v, want %#v", got, want)
	342	+	}
	343	+	}
	344	+	select {
	345	+	case err := <-sent:
	346	+	if err != nil {
	347	+	t.Fatalf("Send after draining: %v", err)
	348	+	}
	349	+	case <-time.After(timeout):
	350	+	t.Fatal("Send did not return after draining")
	351	+	}
	352	+	}
	353	+
	354	+	func TestTransportCloseInterruptsSendWaitingForEgressCapacity(t *testing.T) {
	355	+	const timeout = time.Second
	356	+
	357	+	serverConn, clientConn := net.Pipe()
	358	+	defer func() { _ = clientConn.Close() }()
	359	+
	360	+	server := NewTransport(serverConn)
	361	+	defer closeTransport(t, server)
	362	+	transport, ok := server.(*unixTransport)
	363	+	if !ok {
	364	+	t.Fatal("IPC transport is not a unixTransport")
	365	+	}
	366	+	async, ok := server.(ports.AsyncTransport)
	367	+	if !ok {
	368	+	t.Fatal("IPC transport does not implement AsyncTransport")
	369	+	}
	370	+	fillAsyncEgress(t, transport, async, timeout)
	371	+
	372	+	sent := make(chan error, 1)
	373	+	go func() { sent <- server.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("synchronous")}) }()
	374	+	waitForEgressSender(t, transport, timeout)
	375	+
	376	+	closeTransport(t, server)
	377	+	select {
	378	+	case err := <-sent:
	379	+	if !errors.Is(err, errClosed) {
	380	+	t.Fatalf("Send after Close = %v, want errClosed", err)
	381	+	}
	382	+	case <-time.After(timeout):
	383	+	t.Fatal("Close did not interrupt Send")
	384	+	}
	385	+	}
	386	+
	387	+	func waitForEgressSender(t testing.T, transport unixTransport, timeout time.Duration) {
	388	+	t.Helper()
	389	+
	390	+	timer := time.NewTimer(timeout)
	391	+	defer timer.Stop()
	392	+	for {
	393	+	transport.egressMu.Lock()
	394	+	senders := transport.egressSenders
	395	+	transport.egressMu.Unlock()
	396	+	if senders > 0 {
	397	+	return
	398	+	}
	399	+	select {
	400	+	case <-timer.C:
	401	+	t.Fatal("Send did not register as an egress capacity waiter")
	402	+	default:
	403	+	runtime.Gosched()
	404	+	}
	405	+	}
	406	+	}
	407	+
	408	+	func closeTransport(t *testing.T, transport ports.Transport) {
	409	+	t.Helper()
	410	+
	411	+	closed := make(chan error, 1)
	412	+	go func() { closed <- transport.Close() }()
	413	+	select {
	414	+	case err := <-closed:
	415	+	if err != nil {
	416	+	t.Fatalf("Close: %v", err)
	417	+	}
	418	+	case <-time.After(time.Second):
	419	+	t.Fatal("Close did not return")
	420	+	}
	421	+	}
	422	+
	423	+	func fillAsyncEgress(t testing.T, transport unixTransport, async ports.AsyncTransport, timeout time.Duration) []ports.Frame {
	424	+	t.Helper()
	425	+
	426	+	first := ports.Frame{Type: ports.MsgOutput, Payload: []byte{0}}
	427	+	if err := async.SendAsync(first); err != nil {
	428	+	t.Fatalf("SendAsync blocked frame: %v", err)
	429	+	}
	430	+	waitForEgressDequeue(t, transport, timeout)
	431	+
	432	+	queued := []ports.Frame{first}
	433	+	for i := 1; ; i++ {
	434	+	frame := ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
	435	+	err := async.SendAsync(frame)
	436	+	switch {
	437	+	case err == nil:
	438	+	queued = append(queued, frame)
	439	+	case errors.Is(err, ErrBackpressure):
	440	+	return queued
	441	+	default:
	442	+	t.Fatalf("SendAsync queued frame: %v", err)
	443	+	}
	444	+	}
	445	+	}
	446	+
	447	+	func waitForEgressDequeue(t testing.T, transport unixTransport, timeout time.Duration) {
	448	+	t.Helper()
	449	+
	450	+	timer := time.NewTimer(timeout)
	451	+	defer timer.Stop()
	452	+	for {
	453	+	if len(transport.egress) == 0 {
	454	+	return
	455	+	}
	456	+	select {
	457	+	case <-timer.C:
	458	+	t.Fatal("IPC writer did not dequeue the blocked frame")
	459	+	default:
	460	+	runtime.Gosched()
	461	+	}
	462	+	}
	463	+	}
	464	+
284	465		func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
285	466		serverConn, clientConn := net.Pipe()
286	467		defer func() { _ = clientConn.Close() }()
287	468
288	469		tr := NewTransport(serverConn)
	470	+	defer closeTransport(t, tr)
	471	+	transport, ok := tr.(*unixTransport)
	472	+	if !ok {
	473	+	t.Fatal("IPC transport is not a unixTransport")
	474	+	}
289	475		async, ok := tr.(ports.AsyncTransport)
290	476		if !ok {
291	477		t.Fatal("IPC transport does not implement AsyncTransport")

@@ -315,9 +501,7 @@ func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
315	501		_, err := tr.Recv()
316	502		recvDone <- err
317	503		}()
318		-	if err := tr.Close(); err != nil {
319		-	t.Fatalf("Close: %v", err)
320		-	}
	504	+	closeTransport(t, tr)
321	505		select {
322	506		case err := <-recvDone:
323	507		if err == nil {

@@ -327,7 +511,7 @@ func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
327	511		t.Fatal("Close did not interrupt Recv")
328	512		}
329	513		select {
330		-	case <-tr.(*unixTransport).writerDone:
	514	+	case <-transport.writerDone:
331	515		default:
332	516		t.Fatal("Close returned before the IPC writer terminated")
333	517		}


@@ -6,6 +6,7 @@ import (
6		"io"	6		"io"
7		"net"	7		"net"
8		"reflect"	8		"reflect"
			9	+	"runtime"
9		"sync"	10		"sync"
10		"testing"	11		"testing"
11		"time"	12		"time"

@@ -234,12 +235,22 @@ func TestTransportSendOversizePayloadRejected(t *testing.T) {
234		}	235		}
235			236
236		func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {	237		func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {
			238	+	const timeout = time.Second
			239	+
237		serverConn, clientConn := net.Pipe()	240		serverConn, clientConn := net.Pipe()
238		defer func() { _ = serverConn.Close() }()	241		defer func() { _ = serverConn.Close() }()
239		defer func() { _ = clientConn.Close() }()	242		defer func() { _ = clientConn.Close() }()
			243	+	if err := serverConn.SetDeadline(time.Now().Add(timeout)); err != nil {
			244	+	t.Fatalf("SetDeadline server connection: %v", err)
			245	+	}
			246	+	if err := clientConn.SetDeadline(time.Now().Add(timeout)); err != nil {
			247	+	t.Fatalf("SetDeadline client connection: %v", err)
			248	+	}
240			249
241		server := NewTransport(serverConn)	250		server := NewTransport(serverConn)
			251	+	defer closeTransport(t, server)
242		client := NewTransport(clientConn)	252		client := NewTransport(clientConn)
			253	+	defer closeTransport(t, client)
243		async, ok := server.(ports.AsyncTransport)	254		async, ok := server.(ports.AsyncTransport)
244		if !ok {	255		if !ok {
245		t.Fatal("IPC transport does not implement AsyncTransport")	256		t.Fatal("IPC transport does not implement AsyncTransport")

@@ -255,8 +266,13 @@ func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {
255		if !reflect.DeepEqual(got, welcome) {	266		if !reflect.DeepEqual(got, welcome) {
256		t.Fatalf("welcome = %#v, want %#v", got, welcome)	267		t.Fatalf("welcome = %#v, want %#v", got, welcome)
257		}	268		}
258	-	if err := <-sent; err != nil {	269	+	select {
259	-	t.Fatalf("Send welcome: %v", err)	270	+	case err := <-sent:
			271	+	if err != nil {
			272	+	t.Fatalf("Send welcome: %v", err)
			273	+	}
			274	+	case <-time.After(timeout):
			275	+	t.Fatal("Send welcome did not return")
260		}	276		}
261			277
262		payload := []byte("first")	278		payload := []byte("first")

@@ -281,11 +297,181 @@ func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {
281		}	297		}
282		}	298		}
283			299
			300	+	func TestTransportSendWaitsForEgressCapacity(t *testing.T) {
			301	+	const timeout = time.Second
			302	+
			303	+	serverConn, clientConn := net.Pipe()
			304	+	defer func() { _ = serverConn.Close() }()
			305	+	defer func() { _ = clientConn.Close() }()
			306	+	if err := clientConn.SetDeadline(time.Now().Add(timeout)); err != nil {
			307	+	t.Fatalf("SetDeadline client connection: %v", err)
			308	+	}
			309	+
			310	+	server := NewTransport(serverConn)
			311	+	defer closeTransport(t, server)
			312	+	transport, ok := server.(*unixTransport)
			313	+	if !ok {
			314	+	t.Fatal("IPC transport is not a unixTransport")
			315	+	}
			316	+	client := NewTransport(clientConn)
			317	+	defer closeTransport(t, client)
			318	+	async, ok := server.(ports.AsyncTransport)
			319	+	if !ok {
			320	+	t.Fatal("IPC transport does not implement AsyncTransport")
			321	+	}
			322	+
			323	+	queued := fillAsyncEgress(t, transport, async, timeout)
			324	+	want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("synchronous")}
			325	+	sent := make(chan error, 1)
			326	+	go func() { sent <- server.Send(want) }()
			327	+	waitForEgressSender(t, transport, timeout)
			328	+
			329	+	select {
			330	+	case err := <-sent:
			331	+	t.Fatalf("Send returned before capacity was available: %v", err)
			332	+	default:
			333	+	}
			334	+
			335	+	for _, want := range append(queued, want) {
			336	+	got, err := client.Recv()
			337	+	if err != nil {
			338	+	t.Fatalf("Recv: %v", err)
			339	+	}
			340	+	if !reflect.DeepEqual(got, want) {
			341	+	t.Fatalf("Recv = %#v, want %#v", got, want)
			342	+	}
			343	+	}
			344	+	select {
			345	+	case err := <-sent:
			346	+	if err != nil {
			347	+	t.Fatalf("Send after draining: %v", err)
			348	+	}
			349	+	case <-time.After(timeout):
			350	+	t.Fatal("Send did not return after draining")
			351	+	}
			352	+	}
			353	+
			354	+	func TestTransportCloseInterruptsSendWaitingForEgressCapacity(t *testing.T) {
			355	+	const timeout = time.Second
			356	+
			357	+	serverConn, clientConn := net.Pipe()
			358	+	defer func() { _ = clientConn.Close() }()
			359	+
			360	+	server := NewTransport(serverConn)
			361	+	defer closeTransport(t, server)
			362	+	transport, ok := server.(*unixTransport)
			363	+	if !ok {
			364	+	t.Fatal("IPC transport is not a unixTransport")
			365	+	}
			366	+	async, ok := server.(ports.AsyncTransport)
			367	+	if !ok {
			368	+	t.Fatal("IPC transport does not implement AsyncTransport")
			369	+	}
			370	+	fillAsyncEgress(t, transport, async, timeout)
			371	+
			372	+	sent := make(chan error, 1)
			373	+	go func() { sent <- server.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("synchronous")}) }()
			374	+	waitForEgressSender(t, transport, timeout)
			375	+
			376	+	closeTransport(t, server)
			377	+	select {
			378	+	case err := <-sent:
			379	+	if !errors.Is(err, errClosed) {
			380	+	t.Fatalf("Send after Close = %v, want errClosed", err)
			381	+	}
			382	+	case <-time.After(timeout):
			383	+	t.Fatal("Close did not interrupt Send")
			384	+	}
			385	+	}
			386	+
			387	+	func waitForEgressSender(t testing.T, transport unixTransport, timeout time.Duration) {
			388	+	t.Helper()
			389	+
			390	+	timer := time.NewTimer(timeout)
			391	+	defer timer.Stop()
			392	+	for {
			393	+	transport.egressMu.Lock()
			394	+	senders := transport.egressSenders
			395	+	transport.egressMu.Unlock()
			396	+	if senders > 0 {
			397	+	return
			398	+	}
			399	+	select {
			400	+	case <-timer.C:
			401	+	t.Fatal("Send did not register as an egress capacity waiter")
			402	+	default:
			403	+	runtime.Gosched()
			404	+	}
			405	+	}
			406	+	}
			407	+
			408	+	func closeTransport(t *testing.T, transport ports.Transport) {
			409	+	t.Helper()
			410	+
			411	+	closed := make(chan error, 1)
			412	+	go func() { closed <- transport.Close() }()
			413	+	select {
			414	+	case err := <-closed:
			415	+	if err != nil {
			416	+	t.Fatalf("Close: %v", err)
			417	+	}
			418	+	case <-time.After(time.Second):
			419	+	t.Fatal("Close did not return")
			420	+	}
			421	+	}
			422	+
			423	+	func fillAsyncEgress(t testing.T, transport unixTransport, async ports.AsyncTransport, timeout time.Duration) []ports.Frame {
			424	+	t.Helper()
			425	+
			426	+	first := ports.Frame{Type: ports.MsgOutput, Payload: []byte{0}}
			427	+	if err := async.SendAsync(first); err != nil {
			428	+	t.Fatalf("SendAsync blocked frame: %v", err)
			429	+	}
			430	+	waitForEgressDequeue(t, transport, timeout)
			431	+
			432	+	queued := []ports.Frame{first}
			433	+	for i := 1; ; i++ {
			434	+	frame := ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
			435	+	err := async.SendAsync(frame)
			436	+	switch {
			437	+	case err == nil:
			438	+	queued = append(queued, frame)
			439	+	case errors.Is(err, ErrBackpressure):
			440	+	return queued
			441	+	default:
			442	+	t.Fatalf("SendAsync queued frame: %v", err)
			443	+	}
			444	+	}
			445	+	}
			446	+
			447	+	func waitForEgressDequeue(t testing.T, transport unixTransport, timeout time.Duration) {
			448	+	t.Helper()
			449	+
			450	+	timer := time.NewTimer(timeout)
			451	+	defer timer.Stop()
			452	+	for {
			453	+	if len(transport.egress) == 0 {
			454	+	return
			455	+	}
			456	+	select {
			457	+	case <-timer.C:
			458	+	t.Fatal("IPC writer did not dequeue the blocked frame")
			459	+	default:
			460	+	runtime.Gosched()
			461	+	}
			462	+	}
			463	+	}
			464	+
284		func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {	465		func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
285		serverConn, clientConn := net.Pipe()	466		serverConn, clientConn := net.Pipe()
286		defer func() { _ = clientConn.Close() }()	467		defer func() { _ = clientConn.Close() }()
287			468
288		tr := NewTransport(serverConn)	469		tr := NewTransport(serverConn)
			470	+	defer closeTransport(t, tr)
			471	+	transport, ok := tr.(*unixTransport)
			472	+	if !ok {
			473	+	t.Fatal("IPC transport is not a unixTransport")
			474	+	}
289		async, ok := tr.(ports.AsyncTransport)	475		async, ok := tr.(ports.AsyncTransport)
290		if !ok {	476		if !ok {
291		t.Fatal("IPC transport does not implement AsyncTransport")	477		t.Fatal("IPC transport does not implement AsyncTransport")

@@ -315,9 +501,7 @@ func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
315		_, err := tr.Recv()	501		_, err := tr.Recv()
316		recvDone <- err	502		recvDone <- err
317		}()	503		}()
318	-	if err := tr.Close(); err != nil {	504	+	closeTransport(t, tr)
319	-	t.Fatalf("Close: %v", err)
320	-	}
321		select {	505		select {
322		case err := <-recvDone:	506		case err := <-recvDone:
323		if err == nil {	507		if err == nil {

@@ -327,7 +511,7 @@ func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
327		t.Fatal("Close did not interrupt Recv")	511		t.Fatal("Close did not interrupt Recv")
328		}	512		}
329		select {	513		select {
330	-	case <-tr.(*unixTransport).writerDone:	514	+	case <-transport.writerDone:
331		default:	515		default:
332		t.Fatal("Close returned before the IPC writer terminated")	516		t.Fatal("Close returned before the IPC writer terminated")
333		}	517		}

M internal/usecase/daemon/snapshot_test.go +7 -1


@@ -1041,6 +1041,8 @@ func newSnapshotTestSession(t *testing.T, name string, ephemeral bool, cwd strin
1041	1041
1042	1042		func awaitSnapshotIdle(t testing.TB, sess *session) {
1043	1043		t.Helper()
	1044	+	timer := time.NewTimer(testWaitTimeout)
	1045	+	defer timer.Stop()
1044	1046		for {
1045	1047		sess.snapshotMu.Lock()
1046	1048		pending := sess.snapshotPending

@@ -1049,7 +1051,11 @@ func awaitSnapshotIdle(t testing.TB, sess *session) {
1049	1051		if !pending {
1050	1052		return
1051	1053		}
1052		-	<-changed
	1054	+	select {
	1055	+	case <-changed:
	1056	+	case <-timer.C:
	1057	+	t.Fatal("timed out waiting for snapshot to become idle")
	1058	+	}
1053	1059		}
1054	1060		}
1055	1061

 func awaitSnapshotIdle(t testing.TB, sess *session) {
 	t.Helper()
+	timer := time.NewTimer(testWaitTimeout)
+	defer timer.Stop()
 	for {
 		sess.snapshotMu.Lock()
 		pending := sess.snapshotPending
 		if !pending {
 			return
 		}
-		<-changed
+		select {
+		case <-changed:
+		case <-timer.C:
+			t.Fatal("timed out waiting for snapshot to become idle")
+		}
 	}
 }