fix(remote): bound catalog command lifetime

Signed-off-by: brice <b@bnema.dev>

commit: a13c2fe27392d0e7e9d7a3e33ed13ad86da4c292
author: brice · 2mo ago · 2026-07-30 10:13 UTC
parent: b71ae30
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +160 -32

M internal/adapters/remote/catalog.go +25 -6


@@ -10,6 +10,7 @@ import (
10	10		"log/slog"
11	11		"os/exec"
12	12		"strings"
	13	+	"time"
13	14		"unicode"
14	15		"unicode/utf8"
15	16

@@ -21,6 +22,8 @@ import (
21	22		const (
22	23		maxCatalogBytes = 1 << 20
23	24		maxCatalogDiagnosticBytes = 4 << 10
	25	+	catalogCommandTimeout = 30 * time.Second
	26	+	catalogCommandWaitDelay = 500 * time.Millisecond
24	27		)
25	28
26	29		var (

@@ -35,6 +38,8 @@ var (
35	38		// CatalogClient fetches remote session catalogs over SSH.
36	39		type CatalogClient struct {
37	40		command func(ctx context.Context, name string, args ...string) *exec.Cmd
	41	+	// timeout bounds each List when > 0; otherwise catalogCommandTimeout is used.
	42	+	timeout time.Duration
38	43		}
39	44
40	45		var _ ports.RemoteCatalogClient = (*CatalogClient)(nil)

@@ -44,8 +49,17 @@ func NewCatalogClient() ports.RemoteCatalogClient {
44	49		return &CatalogClient{command: exec.CommandContext}
45	50		}
46	51
	52	+	func (c *CatalogClient) listTimeout() time.Duration {
	53	+	if c != nil && c.timeout > 0 {
	54	+	return c.timeout
	55	+	}
	56	+	return catalogCommandTimeout
	57	+	}
	58	+
47	59		// List runs `ssh -- <target> 'vev' 'cmd' 'remote-catalog' '--json'` and decodes
48		-	// exactly one versioned catalog envelope from stdout.
	60	+	// exactly one versioned catalog envelope from stdout. List always applies a
	61	+	// bounded command timeout derived from ctx so a caller with no deadline cannot
	62	+	// hang indefinitely, while still honoring caller cancellation.
49	63		func (c *CatalogClient) List(ctx context.Context, target string) (ports.RemoteCatalog, error) {
50	64		if err := ctx.Err(); err != nil {
51	65		return ports.RemoteCatalog{}, err

@@ -55,20 +69,25 @@ func (c *CatalogClient) List(ctx context.Context, target string) (ports.RemoteCa
55	69		return ports.RemoteCatalog{}, err
56	70		}
57	71
	72	+	runCtx, cancel := context.WithTimeout(ctx, c.listTimeout())
	73	+	defer cancel()
	74	+
58	75		command := c.command
59	76		if command == nil {
60	77		command = exec.CommandContext
61	78		}
62	79		spec := sshstdio.BuildCommandForRemoteCommand(target, "vev", "cmd", "remote-catalog", "--json")
63		-	cmd := command(ctx, spec.Path, spec.Args...)
	80	+	cmd := command(runCtx, spec.Path, spec.Args...)
64	81		stdout := boundedBuffer{limit: maxCatalogBytes}
65	82		stderr := boundedBuffer{limit: maxCatalogDiagnosticBytes}
66	83		cmd.Stdout = &stdout
67	84		cmd.Stderr = &stderr
68		-	if err := cmd.Run(); err != nil {
69		-	if ctxErr := ctx.Err(); ctxErr != nil {
70		-	return ports.RemoteCatalog{}, ctxErr
71		-	}
	85	+	cmd.WaitDelay = catalogCommandWaitDelay
	86	+	err := cmd.Run()
	87	+	if ctxErr := runCtx.Err(); ctxErr != nil {
	88	+	return ports.RemoteCatalog{}, ctxErr
	89	+	}
	90	+	if err != nil {
72	91		if stdout.overflow \|\| stderr.overflow {
73	92		slog.Debug("remote catalog output too large", "target", target, "stdout_limit", maxCatalogBytes, "stderr_limit", maxCatalogDiagnosticBytes)
74	93		return ports.RemoteCatalog{}, errCatalogTooLarge

 	"log/slog"
 	"os/exec"
 	"strings"
+	"time"
 	"unicode"
 	"unicode/utf8"
 const (
 	maxCatalogBytes           = 1 << 20
 	maxCatalogDiagnosticBytes = 4 << 10
+	catalogCommandTimeout     = 30 * time.Second
+	catalogCommandWaitDelay   = 500 * time.Millisecond
 )
 var (
 // CatalogClient fetches remote session catalogs over SSH.
 type CatalogClient struct {
 	command func(ctx context.Context, name string, args ...string) *exec.Cmd
+	// timeout bounds each List when > 0; otherwise catalogCommandTimeout is used.
+	timeout time.Duration
 }
 var _ ports.RemoteCatalogClient = (*CatalogClient)(nil)
 	return &CatalogClient{command: exec.CommandContext}
 }
+func (c *CatalogClient) listTimeout() time.Duration {
+	if c != nil && c.timeout > 0 {
+		return c.timeout
+	}
+	return catalogCommandTimeout
+}
 // List runs `ssh -- <target> 'vev' 'cmd' 'remote-catalog' '--json'` and decodes
-// exactly one versioned catalog envelope from stdout.
+// exactly one versioned catalog envelope from stdout. List always applies a
+// bounded command timeout derived from ctx so a caller with no deadline cannot
+// hang indefinitely, while still honoring caller cancellation.
 func (c *CatalogClient) List(ctx context.Context, target string) (ports.RemoteCatalog, error) {
 	if err := ctx.Err(); err != nil {
 		return ports.RemoteCatalog{}, err
 		return ports.RemoteCatalog{}, err
 	}
+	runCtx, cancel := context.WithTimeout(ctx, c.listTimeout())
+	defer cancel()
 	command := c.command
 	if command == nil {
 		command = exec.CommandContext
 	}
 	spec := sshstdio.BuildCommandForRemoteCommand(target, "vev", "cmd", "remote-catalog", "--json")
-	cmd := command(ctx, spec.Path, spec.Args...)
+	cmd := command(runCtx, spec.Path, spec.Args...)
 	stdout := boundedBuffer{limit: maxCatalogBytes}
 	stderr := boundedBuffer{limit: maxCatalogDiagnosticBytes}
 	cmd.Stdout = &stdout
 	cmd.Stderr = &stderr
-	if err := cmd.Run(); err != nil {
+	cmd.WaitDelay = catalogCommandWaitDelay
-		if ctxErr := ctx.Err(); ctxErr != nil {
+	err := cmd.Run()
-			return ports.RemoteCatalog{}, ctxErr
+	if ctxErr := runCtx.Err(); ctxErr != nil {
-		}
+		return ports.RemoteCatalog{}, ctxErr
+	}
+	if err != nil {
 		if stdout.overflow || stderr.overflow {
 			slog.Debug("remote catalog output too large", "target", target, "stdout_limit", maxCatalogBytes, "stderr_limit", maxCatalogDiagnosticBytes)
 			return ports.RemoteCatalog{}, errCatalogTooLarge

M internal/adapters/remote/catalog_test.go +104 -0


@@ -8,6 +8,7 @@ import (
8	8		"os/exec"
9	9		"strings"
10	10		"testing"
	11	+	"time"
11	12
12	13		"github.com/bnema/vev/internal/adapters/sshstdio"
13	14		"github.com/bnema/vev/internal/ports"

@@ -220,6 +221,109 @@ func TestRemoteCatalogClientContextCanceled(t *testing.T) {
220	221		}
221	222		}
222	223
	224	+	func TestRemoteCatalogClientAppliesLifetimeAndWaitDelay(t *testing.T) {
	225	+	t.Parallel()
	226	+
	227	+	var commandCtx context.Context
	228	+	var cmd *exec.Cmd
	229	+	client := &CatalogClient{
	230	+	command: func(ctx context.Context, name string, args ...string) *exec.Cmd {
	231	+	commandCtx = ctx
	232	+	cmd = stdoutCmd(ctx, mustCatalogJSON(t, ports.RemoteCatalog{
	233	+	ProtocolVersion: ports.ProtocolVersion,
	234	+	Sessions: []ports.RemoteCatalogSession{},
	235	+	}))
	236	+	return cmd
	237	+	},
	238	+	}
	239	+	if _, err := client.List(context.Background(), "arch"); err != nil {
	240	+	t.Fatalf("List() error = %v", err)
	241	+	}
	242	+	deadline, ok := commandCtx.Deadline()
	243	+	if !ok {
	244	+	t.Fatal("command context missing default lifetime deadline")
	245	+	}
	246	+	if remaining := time.Until(deadline); remaining <= 0 \|\| remaining > catalogCommandTimeout {
	247	+	t.Fatalf("deadline remaining = %v, want within (0, %v]", remaining, catalogCommandTimeout)
	248	+	}
	249	+	if cmd.WaitDelay != catalogCommandWaitDelay {
	250	+	t.Fatalf("WaitDelay = %v, want %v", cmd.WaitDelay, catalogCommandWaitDelay)
	251	+	}
	252	+	}
	253	+
	254	+	func TestRemoteCatalogClientClassifiesRunContextErrors(t *testing.T) {
	255	+	tests := []struct {
	256	+	name string
	257	+	newContext func() (context.Context, context.CancelFunc)
	258	+	timeout time.Duration
	259	+	cancel bool
	260	+	want error
	261	+	}{
	262	+	{
	263	+	name: "own deadline",
	264	+	newContext: func() (context.Context, context.CancelFunc) {
	265	+	return context.WithCancel(context.Background())
	266	+	},
	267	+	timeout: 20 * time.Millisecond,
	268	+	want: context.DeadlineExceeded,
	269	+	},
	270	+	{
	271	+	name: "parent deadline",
	272	+	newContext: func() (context.Context, context.CancelFunc) {
	273	+	return context.WithTimeout(context.Background(), 20*time.Millisecond)
	274	+	},
	275	+	timeout: time.Minute,
	276	+	want: context.DeadlineExceeded,
	277	+	},
	278	+	{
	279	+	name: "parent cancellation",
	280	+	newContext: func() (context.Context, context.CancelFunc) {
	281	+	return context.WithCancel(context.Background())
	282	+	},
	283	+	timeout: time.Minute,
	284	+	cancel: true,
	285	+	want: context.Canceled,
	286	+	},
	287	+	}
	288	+
	289	+	for _, tt := range tests {
	290	+	t.Run(tt.name, func(t *testing.T) {
	291	+	ctx, cancel := tt.newContext()
	292	+	defer cancel()
	293	+	started := make(chan struct{})
	294	+	client := &CatalogClient{
	295	+	timeout: tt.timeout,
	296	+	command: func(ctx context.Context, name string, args ...string) *exec.Cmd {
	297	+	close(started)
	298	+	return exec.CommandContext(ctx, "sleep", "10")
	299	+	},
	300	+	}
	301	+
	302	+	errCh := make(chan error, 1)
	303	+	go func() {
	304	+	_, err := client.List(ctx, "arch")
	305	+	errCh <- err
	306	+	}()
	307	+	select {
	308	+	case <-started:
	309	+	case <-time.After(2 * time.Second):
	310	+	t.Fatal("command did not start")
	311	+	}
	312	+	if tt.cancel {
	313	+	cancel()
	314	+	}
	315	+	select {
	316	+	case err := <-errCh:
	317	+	if !errors.Is(err, tt.want) {
	318	+	t.Fatalf("List() error = %v, want %v", err, tt.want)
	319	+	}
	320	+	case <-time.After(2 * time.Second):
	321	+	t.Fatal("List() did not return after context ended")
	322	+	}
	323	+	})
	324	+	}
	325	+	}
	326	+
223	327		func TestCatalogOutputSafety(t *testing.T) {
224	328		t.Parallel()
225	329


@@ -8,6 +8,7 @@ import (
8	"os/exec"	8		"os/exec"
9	"strings"	9		"strings"
10	"testing"	10		"testing"
		11	+	"time"
11		12
12	"github.com/bnema/vev/internal/adapters/sshstdio"	13		"github.com/bnema/vev/internal/adapters/sshstdio"
13	"github.com/bnema/vev/internal/ports"	14		"github.com/bnema/vev/internal/ports"

@@ -220,6 +221,109 @@ func TestRemoteCatalogClientContextCanceled(t *testing.T) {
220	}	221		}
221	}	222		}
222		223
		224	+	func TestRemoteCatalogClientAppliesLifetimeAndWaitDelay(t *testing.T) {
		225	+	t.Parallel()
		226	+
		227	+	var commandCtx context.Context
		228	+	var cmd *exec.Cmd
		229	+	client := &CatalogClient{
		230	+	command: func(ctx context.Context, name string, args ...string) *exec.Cmd {
		231	+	commandCtx = ctx
		232	+	cmd = stdoutCmd(ctx, mustCatalogJSON(t, ports.RemoteCatalog{
		233	+	ProtocolVersion: ports.ProtocolVersion,
		234	+	Sessions: []ports.RemoteCatalogSession{},
		235	+	}))
		236	+	return cmd
		237	+	},
		238	+	}
		239	+	if _, err := client.List(context.Background(), "arch"); err != nil {
		240	+	t.Fatalf("List() error = %v", err)
		241	+	}
		242	+	deadline, ok := commandCtx.Deadline()
		243	+	if !ok {
		244	+	t.Fatal("command context missing default lifetime deadline")
		245	+	}
		246	+	if remaining := time.Until(deadline); remaining <= 0 \|\| remaining > catalogCommandTimeout {
		247	+	t.Fatalf("deadline remaining = %v, want within (0, %v]", remaining, catalogCommandTimeout)
		248	+	}
		249	+	if cmd.WaitDelay != catalogCommandWaitDelay {
		250	+	t.Fatalf("WaitDelay = %v, want %v", cmd.WaitDelay, catalogCommandWaitDelay)
		251	+	}
		252	+	}
		253	+
		254	+	func TestRemoteCatalogClientClassifiesRunContextErrors(t *testing.T) {
		255	+	tests := []struct {
		256	+	name string
		257	+	newContext func() (context.Context, context.CancelFunc)
		258	+	timeout time.Duration
		259	+	cancel bool
		260	+	want error
		261	+	}{
		262	+	{
		263	+	name: "own deadline",
		264	+	newContext: func() (context.Context, context.CancelFunc) {
		265	+	return context.WithCancel(context.Background())
		266	+	},
		267	+	timeout: 20 * time.Millisecond,
		268	+	want: context.DeadlineExceeded,
		269	+	},
		270	+	{
		271	+	name: "parent deadline",
		272	+	newContext: func() (context.Context, context.CancelFunc) {
		273	+	return context.WithTimeout(context.Background(), 20*time.Millisecond)
		274	+	},
		275	+	timeout: time.Minute,
		276	+	want: context.DeadlineExceeded,
		277	+	},
		278	+	{
		279	+	name: "parent cancellation",
		280	+	newContext: func() (context.Context, context.CancelFunc) {
		281	+	return context.WithCancel(context.Background())
		282	+	},
		283	+	timeout: time.Minute,
		284	+	cancel: true,
		285	+	want: context.Canceled,
		286	+	},
		287	+	}
		288	+
		289	+	for _, tt := range tests {
		290	+	t.Run(tt.name, func(t *testing.T) {
		291	+	ctx, cancel := tt.newContext()
		292	+	defer cancel()
		293	+	started := make(chan struct{})
		294	+	client := &CatalogClient{
		295	+	timeout: tt.timeout,
		296	+	command: func(ctx context.Context, name string, args ...string) *exec.Cmd {
		297	+	close(started)
		298	+	return exec.CommandContext(ctx, "sleep", "10")
		299	+	},
		300	+	}
		301	+
		302	+	errCh := make(chan error, 1)
		303	+	go func() {
		304	+	_, err := client.List(ctx, "arch")
		305	+	errCh <- err
		306	+	}()
		307	+	select {
		308	+	case <-started:
		309	+	case <-time.After(2 * time.Second):
		310	+	t.Fatal("command did not start")
		311	+	}
		312	+	if tt.cancel {
		313	+	cancel()
		314	+	}
		315	+	select {
		316	+	case err := <-errCh:
		317	+	if !errors.Is(err, tt.want) {
		318	+	t.Fatalf("List() error = %v, want %v", err, tt.want)
		319	+	}
		320	+	case <-time.After(2 * time.Second):
		321	+	t.Fatal("List() did not return after context ended")
		322	+	}
		323	+	})
		324	+	}
		325	+	}
		326	+
223	func TestCatalogOutputSafety(t *testing.T) {	327		func TestCatalogOutputSafety(t *testing.T) {
224	t.Parallel()	328		t.Parallel()
225		329

M internal/ports/remote.go +11 -0


@@ -48,6 +48,17 @@ type RemoteCatalogSession struct {
48	48		Attached bool `json:"attached"`
49	49		}
50	50
	51	+	// SaturateUint16 clamps n to the uint16 range.
	52	+	func SaturateUint16(n int) uint16 {
	53	+	if n <= 0 {
	54	+	return 0
	55	+	}
	56	+	if n >= 1<<16 {
	57	+	return ^uint16(0)
	58	+	}
	59	+	return uint16(n)
	60	+	}
	61	+
51	62		// RemoteCatalog is the versioned JSON envelope returned by remote-catalog --json.
52	63		type RemoteCatalog struct {
53	64		ProtocolVersion uint16 `json:"protocol_version"`


@@ -48,6 +48,17 @@ type RemoteCatalogSession struct {
48	Attached bool `json:"attached"`	48		Attached bool `json:"attached"`
49	}	49		}
50		50
		51	+	// SaturateUint16 clamps n to the uint16 range.
		52	+	func SaturateUint16(n int) uint16 {
		53	+	if n <= 0 {
		54	+	return 0
		55	+	}
		56	+	if n >= 1<<16 {
		57	+	return ^uint16(0)
		58	+	}
		59	+	return uint16(n)
		60	+	}
		61	+
51	// RemoteCatalog is the versioned JSON envelope returned by remote-catalog --json.	62		// RemoteCatalog is the versioned JSON envelope returned by remote-catalog --json.
52	type RemoteCatalog struct {	63		type RemoteCatalog struct {
53	ProtocolVersion uint16 `json:"protocol_version"`	64		ProtocolVersion uint16 `json:"protocol_version"`

M internal/usecase/daemon/control.go +3 -13


@@ -616,13 +616,14 @@ func (e controlExec) RemoteCatalog(asJSON bool) (string, error) {
616	616		}
617	617		e.d.mu.Unlock()
618	618		rows := make([]ports.RemoteCatalogSession, 0, len(sessions))
	619	+	// Stopped sessions live in d.stopped and are intentionally not in this listing.
619	620		for _, sess := range sessions {
620	621		snap := sess.snapshotView(viewOptions{})
621	622		rows = append(rows, ports.RemoteCatalogSession{
622	623		Name: snap.name,
623		-	State: remoteCatalogState(ports.SessionRunning),
	624	+	State: "running",
624	625		Ephemeral: snap.ephemeral,
625		-	Tabs: uint16(snap.tabCount),
	626	+	Tabs: ports.SaturateUint16(snap.tabCount),
626	627		Attached: snap.attached,
627	628		})
628	629		}

@@ -633,17 +634,6 @@ func (e controlExec) RemoteCatalog(asJSON bool) (string, error) {
633	634		})
634	635		}
635	636
636		-	func remoteCatalogState(state ports.SessionState) string {
637		-	switch state {
638		-	case ports.SessionStopped:
639		-	return "stopped"
640		-	case ports.SessionBroken:
641		-	return "broken"
642		-	default:
643		-	return "running"
644		-	}
645		-	}
646		-
647	637		func (e controlExec) ListTabs(asJSON bool) (string, error) {
648	638		type row struct {
649	639		Index int `json:"index"`

 	}
 	e.d.mu.Unlock()
 	rows := make([]ports.RemoteCatalogSession, 0, len(sessions))
+	// Stopped sessions live in d.stopped and are intentionally not in this listing.
 	for _, sess := range sessions {
 		snap := sess.snapshotView(viewOptions{})
 		rows = append(rows, ports.RemoteCatalogSession{
 			Name:      snap.name,
-			State:     remoteCatalogState(ports.SessionRunning),
+			State:     "running",
 			Ephemeral: snap.ephemeral,
-			Tabs:      uint16(snap.tabCount),
+			Tabs:      ports.SaturateUint16(snap.tabCount),
 			Attached:  snap.attached,
 		})
 	}
 	})
 }
-func remoteCatalogState(state ports.SessionState) string {
-	switch state {
-	case ports.SessionStopped:
-		return "stopped"
-	case ports.SessionBroken:
-		return "broken"
-	default:
-		return "running"
-	}
-}
 func (e controlExec) ListTabs(asJSON bool) (string, error) {
 	type row struct {
 		Index  int    `json:"index"`

M internal/usecase/daemon/control_test.go +17 -13


@@ -712,20 +712,24 @@ func TestRemoteCatalogJSONOutput(t *testing.T) {
712	712		require.NotContains(t, listAfter.Output, "old")
713	713		}
714	714
715		-	func TestRemoteCatalogStateMapping(t *testing.T) {
716		-	tests := []struct {
717		-	state ports.SessionState
718		-	want string
719		-	}{
720		-	{state: ports.SessionRunning, want: "running"},
721		-	{state: ports.SessionStopped, want: "stopped"},
722		-	{state: ports.SessionBroken, want: "broken"},
723		-	}
724		-	for _, tt := range tests {
725		-	t.Run(tt.want, func(t *testing.T) {
726		-	require.Equal(t, tt.want, remoteCatalogState(tt.state))
727		-	})
	715	+	func TestRemoteCatalogTabCountSaturates(t *testing.T) {
	716	+	d := newTestDaemon(t, nil, stubClock{})
	717	+	sess := addControlSession(d, "work", "t_work", "p_work")
	718	+	sess.mu.Lock()
	719	+	base := sess.tabs[0]
	720	+	sess.tabs = make([]*tab, int(^uint16(0))+1)
	721	+	for i := range sess.tabs {
	722	+	sess.tabs[i] = base
728	723		}
	724	+	sess.mu.Unlock()
	725	+
	726	+	result := sendCommand(t, d, ports.CommandRequest{Slug: "remote-catalog", JSON: true})
	727	+	require.True(t, result.OK, result.Text)
	728	+	var catalog ports.RemoteCatalog
	729	+	require.NoError(t, json.Unmarshal([]byte(result.Output), &catalog))
	730	+	require.Equal(t, []ports.RemoteCatalogSession{{
	731	+	Name: "work", State: "running", Ephemeral: true, Tabs: ^uint16(0), Attached: false,
	732	+	}}, catalog.Sessions)
729	733		}
730	734
731	735		func TestHandleCommandSerializesSelfTargetOnNonActiveTab(t *testing.T) {

 	require.NotContains(t, listAfter.Output, "old")
 }
-func TestRemoteCatalogStateMapping(t *testing.T) {
+func TestRemoteCatalogTabCountSaturates(t *testing.T) {
-	tests := []struct {
+	d := newTestDaemon(t, nil, stubClock{})
-		state ports.SessionState
+	sess := addControlSession(d, "work", "t_work", "p_work")
-		want  string
+	sess.mu.Lock()
-	}{
+	base := sess.tabs[0]
-		{state: ports.SessionRunning, want: "running"},
+	sess.tabs = make([]*tab, int(^uint16(0))+1)
-		{state: ports.SessionStopped, want: "stopped"},
+	for i := range sess.tabs {
-		{state: ports.SessionBroken, want: "broken"},
+		sess.tabs[i] = base
-	}
-	for _, tt := range tests {
-		t.Run(tt.want, func(t *testing.T) {
-			require.Equal(t, tt.want, remoteCatalogState(tt.state))
-		})
 	}
+	sess.mu.Unlock()
+	result := sendCommand(t, d, ports.CommandRequest{Slug: "remote-catalog", JSON: true})
+	require.True(t, result.OK, result.Text)
+	var catalog ports.RemoteCatalog
+	require.NoError(t, json.Unmarshal([]byte(result.Output), &catalog))
+	require.Equal(t, []ports.RemoteCatalogSession{{
+		Name: "work", State: "running", Ephemeral: true, Tabs: ^uint16(0), Attached: false,
+	}}, catalog.Sessions)
 }
 func TestHandleCommandSerializesSelfTargetOnNonActiveTab(t *testing.T) {