fix(client): never read kept session input as a resume cancel key

Input the lost attachment kept, or read before the loss was noticed, was typed for the remote shell: hold it verbatim instead of treating a Ctrl-C or trailing Esc in it as a cancel. Make the held-input bound sticky so the resumed session never receives input with a gap.

commit: 3980b1367e2f9c86b47d585c8a042c62ed700411
author: brice · 15d ago · 2026-10-07 07:20 UTC
parent: 27f03fe
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +106 -2

M internal/usecase/client/attachment_outbox_test.go +70 -0

@@ -1,6 +1,7 @@
1	1		package client
2	2
3	3		import (
	4	+	"bytes"
4	5		"errors"
5	6		"strings"
6	7		"sync/atomic"

@@ -150,6 +151,75 @@ func TestResumeWatch(t *testing.T) {
150	151		}
151	152		}
152	153
	154	+	// TestResumeWatchKeepsBacklogVerbatim starts a watch while session input the
	155	+	// lost attachment kept is still waiting: a Ctrl-C or trailing Esc in it was
	156	+	// typed for the remote shell, so it is replayed, not read as a cancel key.
	157	+	// Keys typed after the watch started still cancel.
	158	+	func TestResumeWatchKeepsBacklogVerbatim(t *testing.T) {
	159	+	for _, tt := range []struct {
	160	+	name, residual, pending, typed string
	161	+	wantCancel bool
	162	+	wantResidual string
	163	+	}{
	164	+	{name: "kept ctrl-c", residual: "\x03", wantResidual: "\x03"},
	165	+	{name: "kept trailing esc", residual: "ihello\x1b", wantResidual: "ihello\x1b"},
	166	+	{name: "pending ctrl-c", pending: "x\x03", wantResidual: "x\x03"},
	167	+	{name: "residual then pending", residual: "a", pending: "\x03", wantResidual: "a\x03"},
	168	+	{name: "new ctrl-c still cancels", residual: "\x03", typed: "\x03", wantCancel: true},
	169	+	} {
	170	+	t.Run(tt.name, func(t *testing.T) {
	171	+	reader := &inputTestReader{chunks: make(chan []byte, 2)}
	172	+	pump := newTerminalInputPump(reader)
	173	+	pump.start()
	174	+	t.Cleanup(func() {
	175	+	pump.stop()
	176	+	close(reader.chunks)
	177	+	})
	178	+	pump.mu.Lock()
	179	+	pump.residual = []byte(tt.residual)
	180	+	pump.mu.Unlock()
	181	+	if tt.pending != "" {
	182	+	reader.chunks <- []byte(tt.pending)
	183	+	require.Eventually(t, func() bool {
	184	+	pump.mu.Lock()
	185	+	defer pump.mu.Unlock()
	186	+	return pump.pending != nil
	187	+	}, 5*time.Second, time.Millisecond)
	188	+	}
	189	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
	190	+	watch := lifetime.watchResume(newSupervisorTestClock())
	191	+	require.Eventually(t, func() bool { return watch.heldLen() == len(tt.residual)+len(tt.pending) }, 5*time.Second, time.Millisecond)
	192	+	if tt.typed != "" {
	193	+	reader.chunks <- []byte(tt.typed)
	194	+	}
	195	+	if tt.wantCancel {
	196	+	select {
	197	+	case <-watch.cancelled:
	198	+	case <-time.After(5 * time.Second):
	199	+	t.Fatal("new cancel key ignored")
	200	+	}
	201	+	}
	202	+	watch.release()
	203	+	select {
	204	+	case <-watch.cancelled:
	205	+	require.True(t, tt.wantCancel, "backlog cancelled the resume")
	206	+	default:
	207	+	}
	208	+	pump.mu.Lock()
	209	+	defer pump.mu.Unlock()
	210	+	require.Equal(t, tt.wantResidual, string(pump.residual))
	211	+	})
	212	+	}
	213	+	}
	214	+
	215	+	func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
	216	+	w := &resumeWatch{}
	217	+	w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
	218	+	w.hold([]byte("bb"))
	219	+	w.hold([]byte("c"))
	220	+	require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
	221	+	}
	222	+
153	223		// stallingSessionStream passes writes through until stall is set; stalled
154	224		// writes block until release closes, then fail with failErr when set.
155	225		type stallingSessionStream struct {

@@ -1,6 +1,7 @@
1	package client	1		package client
2		2
3	import (	3		import (
		4	+	"bytes"
4	"errors"	5		"errors"
5	"strings"	6		"strings"
6	"sync/atomic"	7		"sync/atomic"

@@ -150,6 +151,75 @@ func TestResumeWatch(t *testing.T) {
150	}	151		}
151	}	152		}
152		153
		154	+	// TestResumeWatchKeepsBacklogVerbatim starts a watch while session input the
		155	+	// lost attachment kept is still waiting: a Ctrl-C or trailing Esc in it was
		156	+	// typed for the remote shell, so it is replayed, not read as a cancel key.
		157	+	// Keys typed after the watch started still cancel.
		158	+	func TestResumeWatchKeepsBacklogVerbatim(t *testing.T) {
		159	+	for _, tt := range []struct {
		160	+	name, residual, pending, typed string
		161	+	wantCancel bool
		162	+	wantResidual string
		163	+	}{
		164	+	{name: "kept ctrl-c", residual: "\x03", wantResidual: "\x03"},
		165	+	{name: "kept trailing esc", residual: "ihello\x1b", wantResidual: "ihello\x1b"},
		166	+	{name: "pending ctrl-c", pending: "x\x03", wantResidual: "x\x03"},
		167	+	{name: "residual then pending", residual: "a", pending: "\x03", wantResidual: "a\x03"},
		168	+	{name: "new ctrl-c still cancels", residual: "\x03", typed: "\x03", wantCancel: true},
		169	+	} {
		170	+	t.Run(tt.name, func(t *testing.T) {
		171	+	reader := &inputTestReader{chunks: make(chan []byte, 2)}
		172	+	pump := newTerminalInputPump(reader)
		173	+	pump.start()
		174	+	t.Cleanup(func() {
		175	+	pump.stop()
		176	+	close(reader.chunks)
		177	+	})
		178	+	pump.mu.Lock()
		179	+	pump.residual = []byte(tt.residual)
		180	+	pump.mu.Unlock()
		181	+	if tt.pending != "" {
		182	+	reader.chunks <- []byte(tt.pending)
		183	+	require.Eventually(t, func() bool {
		184	+	pump.mu.Lock()
		185	+	defer pump.mu.Unlock()
		186	+	return pump.pending != nil
		187	+	}, 5*time.Second, time.Millisecond)
		188	+	}
		189	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
		190	+	watch := lifetime.watchResume(newSupervisorTestClock())
		191	+	require.Eventually(t, func() bool { return watch.heldLen() == len(tt.residual)+len(tt.pending) }, 5*time.Second, time.Millisecond)
		192	+	if tt.typed != "" {
		193	+	reader.chunks <- []byte(tt.typed)
		194	+	}
		195	+	if tt.wantCancel {
		196	+	select {
		197	+	case <-watch.cancelled:
		198	+	case <-time.After(5 * time.Second):
		199	+	t.Fatal("new cancel key ignored")
		200	+	}
		201	+	}
		202	+	watch.release()
		203	+	select {
		204	+	case <-watch.cancelled:
		205	+	require.True(t, tt.wantCancel, "backlog cancelled the resume")
		206	+	default:
		207	+	}
		208	+	pump.mu.Lock()
		209	+	defer pump.mu.Unlock()
		210	+	require.Equal(t, tt.wantResidual, string(pump.residual))
		211	+	})
		212	+	}
		213	+	}
		214	+
		215	+	func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
		216	+	w := &resumeWatch{}
		217	+	w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
		218	+	w.hold([]byte("bb"))
		219	+	w.hold([]byte("c"))
		220	+	require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
		221	+	}
		222	+
153	// stallingSessionStream passes writes through until stall is set; stalled	223		// stallingSessionStream passes writes through until stall is set; stalled
154	// writes block until release closes, then fail with failErr when set.	224		// writes block until release closes, then fail with failErr when set.
155	type stallingSessionStream struct {	225		type stallingSessionStream struct {

M internal/usecase/client/supervisor.go +17 -2


@@ -1505,8 +1505,15 @@ type resumeWatch struct {
1505	1505		id uint64
1506	1506		clock ports.Clock
1507	1507
	1508	+	// backlog counts deliveries already waiting at claim: session input the
	1509	+	// lost attachment kept, held verbatim and never read as a cancel key.
	1510	+	backlog int
	1511	+
1508	1512		mu sync.Mutex
1509	1513		held []byte
	1514	+	// overflowed drops every key after the first one past the bound, so the
	1515	+	// resumed session never receives input with a gap in the middle.
	1516	+	overflowed bool
1510	1517		}
1511	1518
1512	1519		// heldLen reports how many kept bytes the watch holds.

@@ -1520,9 +1527,11 @@ func (w *resumeWatch) heldLen() int {
1520	1527		func (w *resumeWatch) hold(data []byte) {
1521	1528		w.mu.Lock()
1522	1529		defer w.mu.Unlock()
1523		-	if len(w.held)+len(data) <= resumeHeldInputLimit {
1524		-	w.held = append(w.held, data...)
	1530	+	if w.overflowed \|\| len(w.held)+len(data) > resumeHeldInputLimit {
	1531	+	w.overflowed = w.overflowed \|\| len(data) != 0
	1532	+	return
1525	1533		}
	1534	+	w.held = append(w.held, data...)
1526	1535		}
1527	1536
1528	1537		// watchResume claims terminal input for one resume backoff. When the claim

@@ -1539,6 +1548,7 @@ func (l terminalInputLifetime) watchResume(clock ports.Clock) resumeWatch {
1539	1548		return w
1540	1549		}
1541	1550		w.id = id
	1551	+	w.backlog = l.pump.backlog(id)
1542	1552		go w.run()
1543	1553		return w
1544	1554		}

@@ -1582,6 +1592,11 @@ func (w *resumeWatch) run() {
1582	1592		return
1583	1593		}
1584	1594		pump.ack(w.id)
	1595	+	if w.backlog > 0 {
	1596	+	w.backlog--
	1597	+	w.hold(result.data)
	1598	+	continue
	1599	+	}
1585	1600		held, cancel := decoder.feed(result.data)
1586	1601		w.hold(held)
1587	1602		if cancel {

 	id        uint64
 	clock     ports.Clock
+	// backlog counts deliveries already waiting at claim: session input the
+	// lost attachment kept, held verbatim and never read as a cancel key.
+	backlog int
 	mu   sync.Mutex
 	held []byte
+	// overflowed drops every key after the first one past the bound, so the
+	// resumed session never receives input with a gap in the middle.
+	overflowed bool
 }
 // heldLen reports how many kept bytes the watch holds.
 func (w *resumeWatch) hold(data []byte) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	if len(w.held)+len(data) <= resumeHeldInputLimit {
+	if w.overflowed || len(w.held)+len(data) > resumeHeldInputLimit {
-		w.held = append(w.held, data...)
+		w.overflowed = w.overflowed || len(data) != 0
+		return
 	}
+	w.held = append(w.held, data...)
 }
 // watchResume claims terminal input for one resume backoff. When the claim
 		return w
 	}
 	w.id = id
+	w.backlog = l.pump.backlog(id)
 	go w.run()
 	return w
 }
 			return
 		}
 		pump.ack(w.id)
+		if w.backlog > 0 {
+			w.backlog--
+			w.hold(result.data)
+			continue
+		}
 		held, cancel := decoder.feed(result.data)
 		w.hold(held)
 		if cancel {

M internal/usecase/client/terminal_input_pump.go +19 -0


@@ -125,6 +125,25 @@ func (p *terminalInputPump) tryClaim() (uint64, bool) {
125	125		return consumer, true
126	126		}
127	127
	128	+	// backlog reports how many deliveries were already waiting when consumer
	129	+	// claimed the pump: the residual, then the pending read. They were read for an
	130	+	// earlier owner, not typed for consumer.
	131	+	func (p *terminalInputPump) backlog(consumer uint64) int {
	132	+	p.mu.Lock()
	133	+	defer p.mu.Unlock()
	134	+	if p.consumer != consumer {
	135	+	return 0
	136	+	}
	137	+	n := 0
	138	+	if len(p.residual) != 0 {
	139	+	n++
	140	+	}
	141	+	if p.pending != nil {
	142	+	n++
	143	+	}
	144	+	return n
	145	+	}
	146	+
128	147		// revoke invalidates an attempt before its replacement is allowed to claim
129	148		// input. Pending bytes are deliberately retained for that replacement.
130	149		func (p *terminalInputPump) revoke(consumer uint64) {


@@ -125,6 +125,25 @@ func (p *terminalInputPump) tryClaim() (uint64, bool) {
125	return consumer, true	125		return consumer, true
126	}	126		}
127		127
		128	+	// backlog reports how many deliveries were already waiting when consumer
		129	+	// claimed the pump: the residual, then the pending read. They were read for an
		130	+	// earlier owner, not typed for consumer.
		131	+	func (p *terminalInputPump) backlog(consumer uint64) int {
		132	+	p.mu.Lock()
		133	+	defer p.mu.Unlock()
		134	+	if p.consumer != consumer {
		135	+	return 0
		136	+	}
		137	+	n := 0
		138	+	if len(p.residual) != 0 {
		139	+	n++
		140	+	}
		141	+	if p.pending != nil {
		142	+	n++
		143	+	}
		144	+	return n
		145	+	}
		146	+
128	// revoke invalidates an attempt before its replacement is allowed to claim	147		// revoke invalidates an attempt before its replacement is allowed to claim
129	// input. Pending bytes are deliberately retained for that replacement.	148		// input. Pending bytes are deliberately retained for that replacement.
130	func (p *terminalInputPump) revoke(consumer uint64) {	149		func (p *terminalInputPump) revoke(consumer uint64) {