fix(client): preserve palette read generation

commit: d885d0f550c9306980d1c12ce676280d4067f2e1
author: brice · 3mo ago · 2026-07-18 06:54 UTC
parent: d059917
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +85 -16

M internal/usecase/client/client.go +22 -16


@@ -919,15 +919,16 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
919	919		// preserving all ordinary terminal input byte-for-byte. It never publishes a
920	920		// Theme and never writes terminal output; attachAttempt owns both operations.
921	921		type stdinPump struct {
922		-	ctx context.Context
923		-	cancel context.CancelFunc
924		-	in io.Reader
925		-	out chan<- ports.Frame
926		-	clock ports.Clock
927		-	clipboard ports.ClipboardReader
928		-	logger *slog.Logger
929		-	paletteEvents chan<- paletteGenerationEvent
930		-	activeGeneration *atomic.Uint64
	922	+	ctx context.Context
	923	+	cancel context.CancelFunc
	924	+	in io.Reader
	925	+	out chan<- ports.Frame
	926	+	clock ports.Clock
	927	+	clipboard ports.ClipboardReader
	928	+	logger *slog.Logger
	929	+	paletteEvents chan<- paletteGenerationEvent
	930	+	activeGeneration *atomic.Uint64
	931	+	afterPaletteEvent func(paletteGenerationEvent) // test synchronization hook
931	932		}
932	933
933	934		func (p *stdinPump) run() {

@@ -948,6 +949,9 @@ func (p *stdinPump) run() {
948	949		sendEvent := func(event paletteGenerationEvent) {
949	950		select {
950	951		case p.paletteEvents <- event:
	952	+	if p.afterPaletteEvent != nil {
	953	+	p.afterPaletteEvent(event)
	954	+	}
951	955		case <-p.ctx.Done():
952	956		sendOK.Store(false)
953	957		}

@@ -974,25 +978,27 @@ func (p *stdinPump) run() {
974	978		}
975	979		sink = ci.Scan
976	980		}
977		-	id := func() paletteGenerationID {
978		-	return paletteGenerationID(p.activeGeneration.Load())
979		-	}
980	981		for {
981	982		n, rerr := p.in.Read(buf)
	983	+	// All scanner callbacks from this Read must retain this one generation.
	984	+	// A scheme notification can start its replacement before a later byte in
	985	+	// the same read is scanned; reloading there would misclassify an old
	986	+	// completion marker as the replacement's drain response.
	987	+	readGeneration := paletteGenerationID(p.activeGeneration.Load())
982	988		if n > 0 {
983	989		scanner.Scan(buf[:n], func(kind int, rgb renderer.RGB) {
984	990		kindEvent := paletteEventForeground
985	991		if kind == 11 {
986	992		kindEvent = paletteEventBackground
987	993		}
988		-	sendEvent(paletteGenerationEvent{id: id(), kind: kindEvent, rgb: rgb})
	994	+	sendEvent(paletteGenerationEvent{id: readGeneration, kind: kindEvent, rgb: rgb})
989	995		}, func(slot int, rgb renderer.RGB) {
990		-	sendEvent(paletteGenerationEvent{id: id(), kind: paletteEventPalette, slot: uint8(slot), rgb: rgb})
	996	+	sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventPalette, slot: uint8(slot), rgb: rgb})
991	997		}, func(light bool) {
992		-	sendEvent(paletteGenerationEvent{id: id(), kind: paletteEventScheme, light: light})
	998	+	sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventScheme, light: light})
993	999		}, func(data []byte) {
994	1000		markers.scan(data, sink, func() {
995		-	sendEvent(paletteGenerationEvent{id: id(), kind: paletteEventMarker})
	1001	+	sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventMarker})
996	1002		})
997	1003		})
998	1004		if !sendOK.Load() {

 // preserving all ordinary terminal input byte-for-byte. It never publishes a
 // Theme and never writes terminal output; attachAttempt owns both operations.
 type stdinPump struct {
-	ctx              context.Context
+	ctx               context.Context
-	cancel           context.CancelFunc
+	cancel            context.CancelFunc
-	in               io.Reader
+	in                io.Reader
-	out              chan<- ports.Frame
+	out               chan<- ports.Frame
-	clock            ports.Clock
+	clock             ports.Clock
-	clipboard        ports.ClipboardReader
+	clipboard         ports.ClipboardReader
-	logger           *slog.Logger
+	logger            *slog.Logger
-	paletteEvents    chan<- paletteGenerationEvent
+	paletteEvents     chan<- paletteGenerationEvent
-	activeGeneration *atomic.Uint64
+	activeGeneration  *atomic.Uint64
+	afterPaletteEvent func(paletteGenerationEvent) // test synchronization hook
 }
 func (p *stdinPump) run() {
 	sendEvent := func(event paletteGenerationEvent) {
 		select {
 		case p.paletteEvents <- event:
+			if p.afterPaletteEvent != nil {
+				p.afterPaletteEvent(event)
+			}
 		case <-p.ctx.Done():
 			sendOK.Store(false)
 		}
 		}
 		sink = ci.Scan
 	}
-	id := func() paletteGenerationID {
-		return paletteGenerationID(p.activeGeneration.Load())
-	}
 	for {
 		n, rerr := p.in.Read(buf)
+		// All scanner callbacks from this Read must retain this one generation.
+		// A scheme notification can start its replacement before a later byte in
+		// the same read is scanned; reloading there would misclassify an old
+		// completion marker as the replacement's drain response.
+		readGeneration := paletteGenerationID(p.activeGeneration.Load())
 		if n > 0 {
 			scanner.Scan(buf[:n], func(kind int, rgb renderer.RGB) {
 				kindEvent := paletteEventForeground
 				if kind == 11 {
 					kindEvent = paletteEventBackground
 				}
-				sendEvent(paletteGenerationEvent{id: id(), kind: kindEvent, rgb: rgb})
+				sendEvent(paletteGenerationEvent{id: readGeneration, kind: kindEvent, rgb: rgb})
 			}, func(slot int, rgb renderer.RGB) {
-				sendEvent(paletteGenerationEvent{id: id(), kind: paletteEventPalette, slot: uint8(slot), rgb: rgb})
+				sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventPalette, slot: uint8(slot), rgb: rgb})
 			}, func(light bool) {
-				sendEvent(paletteGenerationEvent{id: id(), kind: paletteEventScheme, light: light})
+				sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventScheme, light: light})
 			}, func(data []byte) {
 				markers.scan(data, sink, func() {
-					sendEvent(paletteGenerationEvent{id: id(), kind: paletteEventMarker})
+					sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventMarker})
 				})
 			})
 			if !sendOK.Load() {

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


@@ -5,7 +5,9 @@ import (
5	5		"errors"
6	6		"io"
7	7		"log/slog"
	8	+	"strings"
8	9		"sync"
	10	+	"sync/atomic"
9	11		"testing"
10	12		"time"
11	13

@@ -276,6 +278,67 @@ func TestAttachInitialClearedThemeSendReturnsTransportError(t *testing.T) {
276	278		require.ErrorContains(t, err, "publishing theme")
277	279		}
278	280
	281	+	func TestStdinPumpSameReadSchemeAndOldMarkerCannotAdvanceReplacement(t *testing.T) {
	282	+	ctx, cancel := context.WithCancel(context.Background())
	283	+	defer cancel()
	284	+
	285	+	coordinator := newPaletteGenerationCoordinator()
	286	+	coordinator.start(ports.Theme{}, false)
	287	+	var activeGeneration atomic.Uint64
	288	+	activeGeneration.Store(uint64(coordinator.current.id))
	289	+
	290	+	// An unbuffered event channel forces the attach loop to process the scheme
	291	+	// notification and publish its replacement generation before the scanner
	292	+	// can deliver the old marker from the same terminal Read.
	293	+	events := make(chan paletteGenerationEvent)
	294	+	out := make(chan ports.Frame, 1)
	295	+	schemeHandled := make(chan struct{})
	296	+	pump := &stdinPump{
	297	+	ctx: ctx,
	298	+	cancel: cancel,
	299	+	in: strings.NewReader("\x1b[?997;2n\x1b[?2031;1$y"),
	300	+	out: out,
	301	+	clock: newAttachPaletteClock(),
	302	+	logger: slog.New(slog.DiscardHandler),
	303	+	paletteEvents: events,
	304	+	activeGeneration: &activeGeneration,
	305	+	afterPaletteEvent: func(event paletteGenerationEvent) {
	306	+	if event.kind == paletteEventScheme {
	307	+	<-schemeHandled
	308	+	}
	309	+	},
	310	+	}
	311	+	pumpDone := make(chan struct{})
	312	+	go func() {
	313	+	pump.run()
	314	+	close(pumpDone)
	315	+	}()
	316	+
	317	+	scheme := <-events
	318	+	require.Equal(t, paletteEventScheme, scheme.kind)
	319	+	oldID := scheme.id
	320	+
	321	+	// This models attachAttempt handling the scheme event. The replacement is
	322	+	// draining, so a marker tagged with its ID would incorrectly start its
	323	+	// color batch before the real drain boundary arrives.
	324	+	coordinator.start(ports.Theme{SchemeKnown: true, Light: scheme.light}, true)
	325	+	newID := coordinator.current.id
	326	+	activeGeneration.Store(uint64(newID))
	327	+	close(schemeHandled)
	328	+	require.NotEqual(t, oldID, newID)
	329	+	require.Equal(t, generationDraining, coordinator.current.phase)
	330	+
	331	+	oldMarker := <-events
	332	+	require.Equal(t, paletteEventMarker, oldMarker.kind)
	333	+	require.Equal(t, oldID, oldMarker.id, "one read must retain its generation tag")
	334	+	require.Empty(t, coordinator.handle(oldMarker))
	335	+	require.True(t, coordinator.active)
	336	+	require.Equal(t, generationDraining, coordinator.current.phase)
	337	+	require.Equal(t, newID, coordinator.current.id)
	338	+
	339	+	<-pumpDone
	340	+	}
	341	+
279	342		func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) {
280	343		t.Setenv("COLORTERM", "truecolor")
281	344		h := startAttachPaletteHarness(&terminalThemeState{})


@@ -5,7 +5,9 @@ import (
5	"errors"	5		"errors"
6	"io"	6		"io"
7	"log/slog"	7		"log/slog"
		8	+	"strings"
8	"sync"	9		"sync"
		10	+	"sync/atomic"
9	"testing"	11		"testing"
10	"time"	12		"time"
11		13

@@ -276,6 +278,67 @@ func TestAttachInitialClearedThemeSendReturnsTransportError(t *testing.T) {
276	require.ErrorContains(t, err, "publishing theme")	278		require.ErrorContains(t, err, "publishing theme")
277	}	279		}
278		280
		281	+	func TestStdinPumpSameReadSchemeAndOldMarkerCannotAdvanceReplacement(t *testing.T) {
		282	+	ctx, cancel := context.WithCancel(context.Background())
		283	+	defer cancel()
		284	+
		285	+	coordinator := newPaletteGenerationCoordinator()
		286	+	coordinator.start(ports.Theme{}, false)
		287	+	var activeGeneration atomic.Uint64
		288	+	activeGeneration.Store(uint64(coordinator.current.id))
		289	+
		290	+	// An unbuffered event channel forces the attach loop to process the scheme
		291	+	// notification and publish its replacement generation before the scanner
		292	+	// can deliver the old marker from the same terminal Read.
		293	+	events := make(chan paletteGenerationEvent)
		294	+	out := make(chan ports.Frame, 1)
		295	+	schemeHandled := make(chan struct{})
		296	+	pump := &stdinPump{
		297	+	ctx: ctx,
		298	+	cancel: cancel,
		299	+	in: strings.NewReader("\x1b[?997;2n\x1b[?2031;1$y"),
		300	+	out: out,
		301	+	clock: newAttachPaletteClock(),
		302	+	logger: slog.New(slog.DiscardHandler),
		303	+	paletteEvents: events,
		304	+	activeGeneration: &activeGeneration,
		305	+	afterPaletteEvent: func(event paletteGenerationEvent) {
		306	+	if event.kind == paletteEventScheme {
		307	+	<-schemeHandled
		308	+	}
		309	+	},
		310	+	}
		311	+	pumpDone := make(chan struct{})
		312	+	go func() {
		313	+	pump.run()
		314	+	close(pumpDone)
		315	+	}()
		316	+
		317	+	scheme := <-events
		318	+	require.Equal(t, paletteEventScheme, scheme.kind)
		319	+	oldID := scheme.id
		320	+
		321	+	// This models attachAttempt handling the scheme event. The replacement is
		322	+	// draining, so a marker tagged with its ID would incorrectly start its
		323	+	// color batch before the real drain boundary arrives.
		324	+	coordinator.start(ports.Theme{SchemeKnown: true, Light: scheme.light}, true)
		325	+	newID := coordinator.current.id
		326	+	activeGeneration.Store(uint64(newID))
		327	+	close(schemeHandled)
		328	+	require.NotEqual(t, oldID, newID)
		329	+	require.Equal(t, generationDraining, coordinator.current.phase)
		330	+
		331	+	oldMarker := <-events
		332	+	require.Equal(t, paletteEventMarker, oldMarker.kind)
		333	+	require.Equal(t, oldID, oldMarker.id, "one read must retain its generation tag")
		334	+	require.Empty(t, coordinator.handle(oldMarker))
		335	+	require.True(t, coordinator.active)
		336	+	require.Equal(t, generationDraining, coordinator.current.phase)
		337	+	require.Equal(t, newID, coordinator.current.id)
		338	+
		339	+	<-pumpDone
		340	+	}
		341	+
279	func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) {	342		func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) {
280	t.Setenv("COLORTERM", "truecolor")	343		t.Setenv("COLORTERM", "truecolor")
281	h := startAttachPaletteHarness(&terminalThemeState{})	344		h := startAttachPaletteHarness(&terminalThemeState{})