fix(client): preserve scanner residual before reconnect claim handoff

commit: aae268046728c77e40b7e63e3adf2b82b91c1640
author: brice · 3mo ago · 2026-07-18 11:50 UTC
parent: afb8c1a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +66 -5

M internal/usecase/client/client.go +10 -5


@@ -764,14 +764,15 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
764	764		defer close(stdinDone)
765	765		(&stdinPump{ctx: loopCtx, cancel: cancel, input: input, consumer: inputConsumer, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
766	766		}()
767		-	// Revoke this scanner's dequeue permission before waiting for it to exit.
768		-	// The lifecycle-owned reader retains queued bytes for the next attempt.
	767	+	// Scanner cancellation can leave an undecided marker suffix (including a
	768	+	// standalone Escape) that it must hand back to the lifecycle-owned reader.
	769	+	// Wait for that handoff before revoking the claim: otherwise a replacement
	770	+	// scanner could claim the reader between revocation and preservation and
	771	+	// lose the suffix.
769	772		defer func() {
770		-	// Revoke before scanner teardown so replacement attempts can claim the
771		-	// lifecycle-owned reader as soon as this attempt releases it.
772		-	revokeInputClaim()
773	773		cancel()
774	774		<-stdinDone
	775	+	revokeInputClaim()
775	776		}()
776	777		go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
777	778

@@ -1007,6 +1008,7 @@ type terminalInputPump struct {
1007	1008		space chan struct{}
1008	1009		state chan struct{}
1009	1010		exited chan struct{}
	1011	+	afterRevoke func() // test synchronization hook
1010	1012		startMu sync.Once
1011	1013		stopMu sync.Once
1012	1014		}

@@ -1053,6 +1055,9 @@ func (p *terminalInputPump) revoke(consumer uint64) {
1053	1055		}
1054	1056		p.mu.Unlock()
1055	1057		p.signalReady()
	1058	+	if p.afterRevoke != nil {
	1059	+	p.afterRevoke()
	1060	+	}
1056	1061		}
1057	1062
1058	1063		func (p *terminalInputPump) signalReady() {

 		defer close(stdinDone)
 		(&stdinPump{ctx: loopCtx, cancel: cancel, input: input, consumer: inputConsumer, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
 	}()
-	// Revoke this scanner's dequeue permission before waiting for it to exit.
+	// Scanner cancellation can leave an undecided marker suffix (including a
-	// The lifecycle-owned reader retains queued bytes for the next attempt.
+	// standalone Escape) that it must hand back to the lifecycle-owned reader.
+	// Wait for that handoff before revoking the claim: otherwise a replacement
+	// scanner could claim the reader between revocation and preservation and
+	// lose the suffix.
 	defer func() {
-		// Revoke before scanner teardown so replacement attempts can claim the
-		// lifecycle-owned reader as soon as this attempt releases it.
-		revokeInputClaim()
 		cancel()
 		<-stdinDone
+		revokeInputClaim()
 	}()
 	go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
 	space              chan struct{}
 	state              chan struct{}
 	exited             chan struct{}
+	afterRevoke        func() // test synchronization hook
 	startMu            sync.Once
 	stopMu             sync.Once
 }
 	}
 	p.mu.Unlock()
 	p.signalReady()
+	if p.afterRevoke != nil {
+		p.afterRevoke()
+	}
 }
 func (p *terminalInputPump) signalReady() {

M internal/usecase/client/palette_attach_generation_test.go +56 -0


@@ -255,6 +255,62 @@ func TestAttachReconnectClearsRetainedPaletteInOnePublication(t *testing.T) {
255	255		require.Len(t, reconnected.transport.snapshots(), 2, "reconnect must not send a restore Theme before its cleared publication")
256	256		}
257	257
	258	+	func TestAttachReconnectPreservesStandaloneEscapeBeforeClaimRevocation(t *testing.T) {
	259	+	t.Setenv("COLORTERM", "truecolor")
	260	+	reader := newAttachPaletteReader()
	261	+	input := newTerminalInputPump(reader)
	262	+	input.start()
	263	+	t.Cleanup(input.stop)
	264	+
	265	+	first := startAttachPaletteHarnessWithInput(&terminalThemeState{}, newAttachPaletteTransport(), input)
	266	+	require.Equal(t, paletteColorBatch, first.nextWrite(t))
	267	+	_ = first.nextTheme(t)
	268	+	_ = first.nextTimer(t) // initial palette completion deadline
	269	+
	270	+	// A standalone Escape is withheld as a possible DECRQM marker prefix. Its
	271	+	// raw read is committed, leaving only the scanner-local residual to hand
	272	+	// over when this attach is replaced.
	273	+	reader.send("\x1b")
	274	+	_ = first.nextTimer(t) // standalone Escape ambiguity deadline
	275	+
	276	+	revoked := make(chan []byte, 1)
	277	+	allowRevocationToReturn := make(chan struct{})
	278	+	var releaseRevocation sync.Once
	279	+	release := func() { releaseRevocation.Do(func() { close(allowRevocationToReturn) }) }
	280	+	defer release()
	281	+	input.afterRevoke = func() {
	282	+	input.mu.Lock()
	283	+	residual := append([]byte(nil), input.residual...)
	284	+	input.mu.Unlock()
	285	+	revoked <- residual
	286	+	<-allowRevocationToReturn
	287	+	}
	288	+
	289	+	// Detach starts reconnect teardown. The residual must already be owned by
	290	+	// the lifecycle reader at the precise point the old claim is revoked.
	291	+	first.transport.detached <- ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach})}
	292	+	require.Equal(t, []byte("\x1b"), <-revoked)
	293	+	release()
	294	+	require.NoError(t, (<-first.done).err)
	295	+	input.afterRevoke = nil
	296	+
	297	+	// The replacement attach receives the retained byte exactly once.
	298	+	second := startAttachPaletteHarnessWithInput(&terminalThemeState{}, newAttachPaletteTransport(), input)
	299	+	require.Equal(t, paletteColorBatch, second.nextWrite(t))
	300	+	_ = second.nextTheme(t)
	301	+	_ = second.nextTimer(t) // initial palette completion deadline
	302	+	second.nextTimer(t).fire()
	303	+	second.nextTimer(t).fire()
	304	+	got := <-second.transport.inputCh
	305	+	require.Equal(t, []byte("\x1b"), got.Data)
	306	+	select {
	307	+	case duplicate := <-second.transport.inputCh:
	308	+	require.Failf(t, "duplicate input", "unexpected duplicate %q", duplicate.Data)
	309	+	default:
	310	+	}
	311	+	second.detach(t)
	312	+	}
	313	+
258	314		func TestAttachSchemeReplacementAndDeadlinesProgressWhileReadBlocks(t *testing.T) {
259	315		t.Setenv("COLORTERM", "truecolor")
260	316		h := startAttachPaletteHarness(&terminalThemeState{})


@@ -255,6 +255,62 @@ func TestAttachReconnectClearsRetainedPaletteInOnePublication(t *testing.T) {
255	require.Len(t, reconnected.transport.snapshots(), 2, "reconnect must not send a restore Theme before its cleared publication")	255		require.Len(t, reconnected.transport.snapshots(), 2, "reconnect must not send a restore Theme before its cleared publication")
256	}	256		}
257		257
		258	+	func TestAttachReconnectPreservesStandaloneEscapeBeforeClaimRevocation(t *testing.T) {
		259	+	t.Setenv("COLORTERM", "truecolor")
		260	+	reader := newAttachPaletteReader()
		261	+	input := newTerminalInputPump(reader)
		262	+	input.start()
		263	+	t.Cleanup(input.stop)
		264	+
		265	+	first := startAttachPaletteHarnessWithInput(&terminalThemeState{}, newAttachPaletteTransport(), input)
		266	+	require.Equal(t, paletteColorBatch, first.nextWrite(t))
		267	+	_ = first.nextTheme(t)
		268	+	_ = first.nextTimer(t) // initial palette completion deadline
		269	+
		270	+	// A standalone Escape is withheld as a possible DECRQM marker prefix. Its
		271	+	// raw read is committed, leaving only the scanner-local residual to hand
		272	+	// over when this attach is replaced.
		273	+	reader.send("\x1b")
		274	+	_ = first.nextTimer(t) // standalone Escape ambiguity deadline
		275	+
		276	+	revoked := make(chan []byte, 1)
		277	+	allowRevocationToReturn := make(chan struct{})
		278	+	var releaseRevocation sync.Once
		279	+	release := func() { releaseRevocation.Do(func() { close(allowRevocationToReturn) }) }
		280	+	defer release()
		281	+	input.afterRevoke = func() {
		282	+	input.mu.Lock()
		283	+	residual := append([]byte(nil), input.residual...)
		284	+	input.mu.Unlock()
		285	+	revoked <- residual
		286	+	<-allowRevocationToReturn
		287	+	}
		288	+
		289	+	// Detach starts reconnect teardown. The residual must already be owned by
		290	+	// the lifecycle reader at the precise point the old claim is revoked.
		291	+	first.transport.detached <- ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach})}
		292	+	require.Equal(t, []byte("\x1b"), <-revoked)
		293	+	release()
		294	+	require.NoError(t, (<-first.done).err)
		295	+	input.afterRevoke = nil
		296	+
		297	+	// The replacement attach receives the retained byte exactly once.
		298	+	second := startAttachPaletteHarnessWithInput(&terminalThemeState{}, newAttachPaletteTransport(), input)
		299	+	require.Equal(t, paletteColorBatch, second.nextWrite(t))
		300	+	_ = second.nextTheme(t)
		301	+	_ = second.nextTimer(t) // initial palette completion deadline
		302	+	second.nextTimer(t).fire()
		303	+	second.nextTimer(t).fire()
		304	+	got := <-second.transport.inputCh
		305	+	require.Equal(t, []byte("\x1b"), got.Data)
		306	+	select {
		307	+	case duplicate := <-second.transport.inputCh:
		308	+	require.Failf(t, "duplicate input", "unexpected duplicate %q", duplicate.Data)
		309	+	default:
		310	+	}
		311	+	second.detach(t)
		312	+	}
		313	+
258	func TestAttachSchemeReplacementAndDeadlinesProgressWhileReadBlocks(t *testing.T) {	314		func TestAttachSchemeReplacementAndDeadlinesProgressWhileReadBlocks(t *testing.T) {
259	t.Setenv("COLORTERM", "truecolor")	315		t.Setenv("COLORTERM", "truecolor")
260	h := startAttachPaletteHarness(&terminalThemeState{})	316		h := startAttachPaletteHarness(&terminalThemeState{})