fix(client): guard late palette drain responses
- commit
- d05991761665d637f59dad407896f971d95ee831
- author
- brice · · 2026-07-18 06:48 UTC
- parent
- 9394293
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +74 -113
M internal/usecase/client/client.go +11 -8
| @@ -613,32 +613,35 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||
| 613 | 613 | delete(timers, id) | |
| 614 | 614 | } | |
| 615 | 615 | } | |
| 616 | - | publish := func(snapshot ports.Theme) bool { | |
| 616 | + | publish := func(snapshot ports.Theme) error { | |
| 617 | 617 | frame := ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(snapshot)} | |
| 618 | 618 | // The initial cleared snapshot is sent before the sender exists. This | |
| 619 | 619 | // keeps the generation publication ordered before the first query and | |
| 620 | 620 | // avoids leaving a queued Theme behind an immediate detach. | |
| 621 | 621 | if !senderStarted { | |
| 622 | - | return transport.Send(frame) == nil | |
| 622 | + | if err := transport.Send(frame); err != nil { | |
| 623 | + | return fmt.Errorf("sending initial theme: %w", err) | |
| 624 | + | } | |
| 625 | + | return nil | |
| 623 | 626 | } | |
| 624 | 627 | select { | |
| 625 | 628 | case sendCh <- frame: | |
| 626 | - | return true | |
| 629 | + | return nil | |
| 627 | 630 | case <-loopCtx.Done(): | |
| 628 | - | return false | |
| 631 | + | return context.Canceled | |
| 629 | 632 | } | |
| 630 | 633 | } | |
| 631 | 634 | processPaletteActions := func(actions []paletteGenerationAction) error { | |
| 632 | 635 | for _, action := range actions { | |
| 633 | 636 | switch action.kind { | |
| 634 | 637 | case paletteActionPublishCleared: | |
| 635 | - | if !publish(action.theme) { | |
| 636 | - | return context.Canceled | |
| 638 | + | if err := publish(action.theme); err != nil { | |
| 639 | + | return fmt.Errorf("publishing theme: %w", err) | |
| 637 | 640 | } | |
| 638 | 641 | case paletteActionPublishFinal: | |
| 639 | 642 | themeState.update(func(current *ports.Theme) { *current = action.theme }) | |
| 640 | - | if !publish(action.theme) { | |
| 641 | - | return context.Canceled | |
| 643 | + | if err := publish(action.theme); err != nil { | |
| 644 | + | return fmt.Errorf("publishing theme: %w", err) | |
| 642 | 645 | } | |
| 643 | 646 | case paletteActionWriteDrain, paletteActionWriteBatch: | |
| 644 | 647 | if _, err := term.Out().Write([]byte(action.bytes)); err != nil { | |
| @@ -613,32 +613,35 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||||
| 613 | delete(timers, id) | 613 | delete(timers, id) | ||
| 614 | } | 614 | } | ||
| 615 | } | 615 | } | ||
| 616 | - | publish := func(snapshot ports.Theme) bool { | 616 | + | publish := func(snapshot ports.Theme) error { |
| 617 | frame := ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(snapshot)} | 617 | frame := ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(snapshot)} | ||
| 618 | // The initial cleared snapshot is sent before the sender exists. This | 618 | // The initial cleared snapshot is sent before the sender exists. This | ||
| 619 | // keeps the generation publication ordered before the first query and | 619 | // keeps the generation publication ordered before the first query and | ||
| 620 | // avoids leaving a queued Theme behind an immediate detach. | 620 | // avoids leaving a queued Theme behind an immediate detach. | ||
| 621 | if !senderStarted { | 621 | if !senderStarted { | ||
| 622 | - | return transport.Send(frame) == nil | 622 | + | if err := transport.Send(frame); err != nil { |
| 623 | + | return fmt.Errorf("sending initial theme: %w", err) | |||
| 624 | + | } | |||
| 625 | + | return nil | |||
| 623 | } | 626 | } | ||
| 624 | select { | 627 | select { | ||
| 625 | case sendCh <- frame: | 628 | case sendCh <- frame: | ||
| 626 | - | return true | 629 | + | return nil |
| 627 | case <-loopCtx.Done(): | 630 | case <-loopCtx.Done(): | ||
| 628 | - | return false | 631 | + | return context.Canceled |
| 629 | } | 632 | } | ||
| 630 | } | 633 | } | ||
| 631 | processPaletteActions := func(actions []paletteGenerationAction) error { | 634 | processPaletteActions := func(actions []paletteGenerationAction) error { | ||
| 632 | for _, action := range actions { | 635 | for _, action := range actions { | ||
| 633 | switch action.kind { | 636 | switch action.kind { | ||
| 634 | case paletteActionPublishCleared: | 637 | case paletteActionPublishCleared: | ||
| 635 | - | if !publish(action.theme) { | 638 | + | if err := publish(action.theme); err != nil { |
| 636 | - | return context.Canceled | 639 | + | return fmt.Errorf("publishing theme: %w", err) |
| 637 | } | 640 | } | ||
| 638 | case paletteActionPublishFinal: | 641 | case paletteActionPublishFinal: | ||
| 639 | themeState.update(func(current *ports.Theme) { *current = action.theme }) | 642 | themeState.update(func(current *ports.Theme) { *current = action.theme }) | ||
| 640 | - | if !publish(action.theme) { | 643 | + | if err := publish(action.theme); err != nil { |
| 641 | - | return context.Canceled | 644 | + | return fmt.Errorf("publishing theme: %w", err) |
| 642 | } | 645 | } | ||
| 643 | case paletteActionWriteDrain, paletteActionWriteBatch: | 646 | case paletteActionWriteDrain, paletteActionWriteBatch: | ||
| 644 | if _, err := term.Out().Write([]byte(action.bytes)); err != nil { | 647 | if _, err := term.Out().Write([]byte(action.bytes)); err != nil { | ||
M internal/usecase/client/palette_attach_generation_test.go +33 -11
| @@ -2,6 +2,7 @@ package client | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "errors" | |
| 5 | 6 | "io" | |
| 6 | 7 | "log/slog" | |
| 7 | 8 | "sync" | |
| @@ -83,13 +84,14 @@ func (t *attachPaletteTerminalHarness) Out() io.Writer { retur | |||
| 83 | 84 | func (*attachPaletteTerminalHarness) Flush() error { return nil } | |
| 84 | 85 | ||
| 85 | 86 | type attachPaletteTransport struct { | |
| 86 | - | mu sync.Mutex | |
| 87 | - | frames []ports.Frame | |
| 88 | - | themes []ports.Theme | |
| 89 | - | themeCh chan ports.Theme | |
| 90 | - | inputCh chan ports.Input | |
| 91 | - | detached chan ports.Frame | |
| 92 | - | welcomed bool | |
| 87 | + | mu sync.Mutex | |
| 88 | + | frames []ports.Frame | |
| 89 | + | themes []ports.Theme | |
| 90 | + | themeCh chan ports.Theme | |
| 91 | + | inputCh chan ports.Input | |
| 92 | + | detached chan ports.Frame | |
| 93 | + | themeSendErr error | |
| 94 | + | welcomed bool | |
| 93 | 95 | } | |
| 94 | 96 | ||
| 95 | 97 | func newAttachPaletteTransport() *attachPaletteTransport { | |
| @@ -102,9 +104,13 @@ func newAttachPaletteTransport() *attachPaletteTransport { | |||
| 102 | 104 | func (t *attachPaletteTransport) Send(frame ports.Frame) error { | |
| 103 | 105 | t.mu.Lock() | |
| 104 | 106 | t.frames = append(t.frames, frame) | |
| 107 | + | themeSendErr := t.themeSendErr | |
| 105 | 108 | t.mu.Unlock() | |
| 106 | 109 | switch frame.Type { | |
| 107 | 110 | case ports.MsgTheme: | |
| 111 | + | if themeSendErr != nil { | |
| 112 | + | return themeSendErr | |
| 113 | + | } | |
| 108 | 114 | got, err := ports.UnmarshalTheme(frame.Payload) | |
| 109 | 115 | if err != nil { | |
| 110 | 116 | return err | |
| @@ -148,11 +154,14 @@ type attachPaletteHarness struct { | |||
| 148 | 154 | } | |
| 149 | 155 | ||
| 150 | 156 | func startAttachPaletteHarness(state *terminalThemeState) *attachPaletteHarness { | |
| 157 | + | return startAttachPaletteHarnessWithTransport(state, newAttachPaletteTransport()) | |
| 158 | + | } | |
| 159 | + | ||
| 160 | + | func startAttachPaletteHarnessWithTransport(state *terminalThemeState, transport *attachPaletteTransport) *attachPaletteHarness { | |
| 151 | 161 | clock := newAttachPaletteClock() | |
| 152 | 162 | reader := newAttachPaletteReader() | |
| 153 | 163 | writer := newAttachPaletteWriter() | |
| 154 | 164 | term := &attachPaletteTerminalHarness{in: reader, out: writer, resize: make(chan domain.Size)} | |
| 155 | - | transport := newAttachPaletteTransport() | |
| 156 | 165 | ms := &milestones{} | |
| 157 | 166 | runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)} | |
| 158 | 167 | attempt := &attachAttempt{ | |
| @@ -255,6 +264,18 @@ func TestAttachSchemeReplacementAndDeadlinesProgressWhileReadBlocks(t *testing.T | |||
| 255 | 264 | h.detach(t) | |
| 256 | 265 | } | |
| 257 | 266 | ||
| 267 | + | func TestAttachInitialClearedThemeSendReturnsTransportError(t *testing.T) { | |
| 268 | + | t.Setenv("COLORTERM", "truecolor") | |
| 269 | + | want := errors.New("initial theme send failed") | |
| 270 | + | transport := newAttachPaletteTransport() | |
| 271 | + | transport.themeSendErr = want | |
| 272 | + | h := startAttachPaletteHarnessWithTransport(&terminalThemeState{}, transport) | |
| 273 | + | ||
| 274 | + | err := (<-h.done).err | |
| 275 | + | require.ErrorIs(t, err, want) | |
| 276 | + | require.ErrorContains(t, err, "publishing theme") | |
| 277 | + | } | |
| 278 | + | ||
| 258 | 279 | func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) { | |
| 259 | 280 | t.Setenv("COLORTERM", "truecolor") | |
| 260 | 281 | h := startAttachPaletteHarness(&terminalThemeState{}) | |
| @@ -272,9 +293,10 @@ func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) { | |||
| 272 | 293 | require.Equal(t, paletteColorBatch, h.nextWrite(t)) | |
| 273 | 294 | completionDeadline := h.nextTimer(t) | |
| 274 | 295 | ||
| 275 | - | // The first marker is the late drain response. The second is the batch | |
| 276 | - | // completion marker. Colors between them prove the first cannot finalize. | |
| 277 | - | h.reader.send("\x1b[?2031;1$y\x1b]10;#111213\a\x1b]11;#141516\a\x1b]4;3;#212223\a\x1b[?2031;1$y") | |
| 296 | + | // The drain's stale reports arrive before its late boundary and must not | |
| 297 | + | // contaminate the replacement accumulator. Reports after that boundary are | |
| 298 | + | // the batch's reports and are collected through the second marker. | |
| 299 | + | h.reader.send("\x1b]10;#a1a2a3\a\x1b]11;#a4a5a6\a\x1b]4;2;#a7a8a9\a\x1b[?2031;1$y\x1b]10;#111213\a\x1b]11;#141516\a\x1b]4;3;#212223\a\x1b[?2031;1$y") | |
| 278 | 300 | final := h.nextTheme(t) | |
| 279 | 301 | require.Equal(t, renderer.RGB{R: 17, G: 18, B: 19}, final.Foreground) | |
| 280 | 302 | require.Equal(t, renderer.RGB{R: 20, G: 21, B: 22}, final.Background) | |
| @@ -2,6 +2,7 @@ package client | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "errors" | |||
| 5 | "io" | 6 | "io" | ||
| 6 | "log/slog" | 7 | "log/slog" | ||
| 7 | "sync" | 8 | "sync" | ||
| @@ -83,13 +84,14 @@ func (t *attachPaletteTerminalHarness) Out() io.Writer { retur | |||||
| 83 | func (*attachPaletteTerminalHarness) Flush() error { return nil } | 84 | func (*attachPaletteTerminalHarness) Flush() error { return nil } | ||
| 84 | 85 | ||||
| 85 | type attachPaletteTransport struct { | 86 | type attachPaletteTransport struct { | ||
| 86 | - | mu sync.Mutex | 87 | + | mu sync.Mutex |
| 87 | - | frames []ports.Frame | 88 | + | frames []ports.Frame |
| 88 | - | themes []ports.Theme | 89 | + | themes []ports.Theme |
| 89 | - | themeCh chan ports.Theme | 90 | + | themeCh chan ports.Theme |
| 90 | - | inputCh chan ports.Input | 91 | + | inputCh chan ports.Input |
| 91 | - | detached chan ports.Frame | 92 | + | detached chan ports.Frame |
| 92 | - | welcomed bool | 93 | + | themeSendErr error |
| 94 | + | welcomed bool | |||
| 93 | } | 95 | } | ||
| 94 | 96 | ||||
| 95 | func newAttachPaletteTransport() *attachPaletteTransport { | 97 | func newAttachPaletteTransport() *attachPaletteTransport { | ||
| @@ -102,9 +104,13 @@ func newAttachPaletteTransport() *attachPaletteTransport { | |||||
| 102 | func (t *attachPaletteTransport) Send(frame ports.Frame) error { | 104 | func (t *attachPaletteTransport) Send(frame ports.Frame) error { | ||
| 103 | t.mu.Lock() | 105 | t.mu.Lock() | ||
| 104 | t.frames = append(t.frames, frame) | 106 | t.frames = append(t.frames, frame) | ||
| 107 | + | themeSendErr := t.themeSendErr | |||
| 105 | t.mu.Unlock() | 108 | t.mu.Unlock() | ||
| 106 | switch frame.Type { | 109 | switch frame.Type { | ||
| 107 | case ports.MsgTheme: | 110 | case ports.MsgTheme: | ||
| 111 | + | if themeSendErr != nil { | |||
| 112 | + | return themeSendErr | |||
| 113 | + | } | |||
| 108 | got, err := ports.UnmarshalTheme(frame.Payload) | 114 | got, err := ports.UnmarshalTheme(frame.Payload) | ||
| 109 | if err != nil { | 115 | if err != nil { | ||
| 110 | return err | 116 | return err | ||
| @@ -148,11 +154,14 @@ type attachPaletteHarness struct { | |||||
| 148 | } | 154 | } | ||
| 149 | 155 | ||||
| 150 | func startAttachPaletteHarness(state *terminalThemeState) *attachPaletteHarness { | 156 | func startAttachPaletteHarness(state *terminalThemeState) *attachPaletteHarness { | ||
| 157 | + | return startAttachPaletteHarnessWithTransport(state, newAttachPaletteTransport()) | |||
| 158 | + | } | |||
| 159 | + | ||||
| 160 | + | func startAttachPaletteHarnessWithTransport(state *terminalThemeState, transport *attachPaletteTransport) *attachPaletteHarness { | |||
| 151 | clock := newAttachPaletteClock() | 161 | clock := newAttachPaletteClock() | ||
| 152 | reader := newAttachPaletteReader() | 162 | reader := newAttachPaletteReader() | ||
| 153 | writer := newAttachPaletteWriter() | 163 | writer := newAttachPaletteWriter() | ||
| 154 | term := &attachPaletteTerminalHarness{in: reader, out: writer, resize: make(chan domain.Size)} | 164 | term := &attachPaletteTerminalHarness{in: reader, out: writer, resize: make(chan domain.Size)} | ||
| 155 | - | transport := newAttachPaletteTransport() | |||
| 156 | ms := &milestones{} | 165 | ms := &milestones{} | ||
| 157 | runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)} | 166 | runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)} | ||
| 158 | attempt := &attachAttempt{ | 167 | attempt := &attachAttempt{ | ||
| @@ -255,6 +264,18 @@ func TestAttachSchemeReplacementAndDeadlinesProgressWhileReadBlocks(t *testing.T | |||||
| 255 | h.detach(t) | 264 | h.detach(t) | ||
| 256 | } | 265 | } | ||
| 257 | 266 | ||||
| 267 | + | func TestAttachInitialClearedThemeSendReturnsTransportError(t *testing.T) { | |||
| 268 | + | t.Setenv("COLORTERM", "truecolor") | |||
| 269 | + | want := errors.New("initial theme send failed") | |||
| 270 | + | transport := newAttachPaletteTransport() | |||
| 271 | + | transport.themeSendErr = want | |||
| 272 | + | h := startAttachPaletteHarnessWithTransport(&terminalThemeState{}, transport) | |||
| 273 | + | ||||
| 274 | + | err := (<-h.done).err | |||
| 275 | + | require.ErrorIs(t, err, want) | |||
| 276 | + | require.ErrorContains(t, err, "publishing theme") | |||
| 277 | + | } | |||
| 278 | + | ||||
| 258 | func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) { | 279 | func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) { | ||
| 259 | t.Setenv("COLORTERM", "truecolor") | 280 | t.Setenv("COLORTERM", "truecolor") | ||
| 260 | h := startAttachPaletteHarness(&terminalThemeState{}) | 281 | h := startAttachPaletteHarness(&terminalThemeState{}) | ||
| @@ -272,9 +293,10 @@ func TestAttachLateDrainCannotFinalizeReplacementEarly(t *testing.T) { | |||||
| 272 | require.Equal(t, paletteColorBatch, h.nextWrite(t)) | 293 | require.Equal(t, paletteColorBatch, h.nextWrite(t)) | ||
| 273 | completionDeadline := h.nextTimer(t) | 294 | completionDeadline := h.nextTimer(t) | ||
| 274 | 295 | ||||
| 275 | - | // The first marker is the late drain response. The second is the batch | 296 | + | // The drain's stale reports arrive before its late boundary and must not |
| 276 | - | // completion marker. Colors between them prove the first cannot finalize. | 297 | + | // contaminate the replacement accumulator. Reports after that boundary are |
| 277 | - | h.reader.send("\x1b[?2031;1$y\x1b]10;#111213\a\x1b]11;#141516\a\x1b]4;3;#212223\a\x1b[?2031;1$y") | 298 | + | // the batch's reports and are collected through the second marker. |
| 299 | + | h.reader.send("\x1b]10;#a1a2a3\a\x1b]11;#a4a5a6\a\x1b]4;2;#a7a8a9\a\x1b[?2031;1$y\x1b]10;#111213\a\x1b]11;#141516\a\x1b]4;3;#212223\a\x1b[?2031;1$y") | |||
| 278 | final := h.nextTheme(t) | 300 | final := h.nextTheme(t) | ||
| 279 | require.Equal(t, renderer.RGB{R: 17, G: 18, B: 19}, final.Foreground) | 301 | require.Equal(t, renderer.RGB{R: 17, G: 18, B: 19}, final.Foreground) | ||
| 280 | require.Equal(t, renderer.RGB{R: 20, G: 21, B: 22}, final.Background) | 302 | require.Equal(t, renderer.RGB{R: 20, G: 21, B: 22}, final.Background) | ||
M internal/usecase/client/palette_generation.go +15 -8
| @@ -31,10 +31,11 @@ const ( | |||
| 31 | 31 | ) | |
| 32 | 32 | ||
| 33 | 33 | type paletteGeneration struct { | |
| 34 | - | id paletteGenerationID | |
| 35 | - | phase paletteGenerationPhase | |
| 36 | - | theme ports.Theme | |
| 37 | - | expectedMarkers int | |
| 34 | + | id paletteGenerationID | |
| 35 | + | phase paletteGenerationPhase | |
| 36 | + | theme ports.Theme | |
| 37 | + | expectedMarkers int | |
| 38 | + | awaitingLateDrainBoundary bool | |
| 38 | 39 | } | |
| 39 | 40 | ||
| 40 | 41 | type paletteGenerationEventKind uint8 | |
| @@ -156,8 +157,11 @@ func (c *paletteGenerationCoordinator) handle(event paletteGenerationEvent) []pa | |||
| 156 | 157 | return nil | |
| 157 | 158 | } | |
| 158 | 159 | c.current.phase = generationCollecting | |
| 159 | - | // A late drain response may still arrive after the batch marker. | |
| 160 | + | // A late drain response may still arrive after the batch marker. Exclude | |
| 161 | + | // OSC reports until its boundary is consumed: otherwise stale drain-era | |
| 162 | + | // reports could become part of this generation's definitive snapshot. | |
| 160 | 163 | c.current.expectedMarkers = 2 | |
| 164 | + | c.current.awaitingLateDrainBoundary = true | |
| 161 | 165 | return []paletteGenerationAction{ | |
| 162 | 166 | {kind: paletteActionWriteBatch, id: event.id, bytes: paletteColorBatch}, | |
| 163 | 167 | {kind: paletteActionArmCompletionDeadline, id: event.id, deadline: paletteGenerationDeadline}, | |
| @@ -168,17 +172,17 @@ func (c *paletteGenerationCoordinator) handle(event paletteGenerationEvent) []pa | |||
| 168 | 172 | } | |
| 169 | 173 | return c.finalize() | |
| 170 | 174 | case paletteEventForeground: | |
| 171 | - | if c.current.phase == generationCollecting { | |
| 175 | + | if c.current.phase == generationCollecting && !c.current.awaitingLateDrainBoundary { | |
| 172 | 176 | c.current.theme.HasForeground = true | |
| 173 | 177 | c.current.theme.Foreground = event.rgb | |
| 174 | 178 | } | |
| 175 | 179 | case paletteEventBackground: | |
| 176 | - | if c.current.phase == generationCollecting { | |
| 180 | + | if c.current.phase == generationCollecting && !c.current.awaitingLateDrainBoundary { | |
| 177 | 181 | c.current.theme.HasBackground = true | |
| 178 | 182 | c.current.theme.Background = event.rgb | |
| 179 | 183 | } | |
| 180 | 184 | case paletteEventPalette: | |
| 181 | - | if c.current.phase == generationCollecting && event.slot < 16 { | |
| 185 | + | if c.current.phase == generationCollecting && !c.current.awaitingLateDrainBoundary && event.slot < 16 { | |
| 182 | 186 | c.current.theme.PaletteKnown |= uint16(1) << event.slot | |
| 183 | 187 | c.current.theme.Palette[event.slot] = event.rgb | |
| 184 | 188 | } | |
| @@ -200,6 +204,9 @@ func (c *paletteGenerationCoordinator) onMarker() []paletteGenerationAction { | |||
| 200 | 204 | if c.current.expectedMarkers == 0 { | |
| 201 | 205 | c.current.expectedMarkers = 1 | |
| 202 | 206 | } | |
| 207 | + | if c.current.awaitingLateDrainBoundary { | |
| 208 | + | c.current.awaitingLateDrainBoundary = false | |
| 209 | + | } | |
| 203 | 210 | c.current.expectedMarkers-- | |
| 204 | 211 | if c.current.expectedMarkers == 0 { | |
| 205 | 212 | return c.finalize() | |
| @@ -31,10 +31,11 @@ const ( | |||||
| 31 | ) | 31 | ) | ||
| 32 | 32 | ||||
| 33 | type paletteGeneration struct { | 33 | type paletteGeneration struct { | ||
| 34 | - | id paletteGenerationID | 34 | + | id paletteGenerationID |
| 35 | - | phase paletteGenerationPhase | 35 | + | phase paletteGenerationPhase |
| 36 | - | theme ports.Theme | 36 | + | theme ports.Theme |
| 37 | - | expectedMarkers int | 37 | + | expectedMarkers int |
| 38 | + | awaitingLateDrainBoundary bool | |||
| 38 | } | 39 | } | ||
| 39 | 40 | ||||
| 40 | type paletteGenerationEventKind uint8 | 41 | type paletteGenerationEventKind uint8 | ||
| @@ -156,8 +157,11 @@ func (c *paletteGenerationCoordinator) handle(event paletteGenerationEvent) []pa | |||||
| 156 | return nil | 157 | return nil | ||
| 157 | } | 158 | } | ||
| 158 | c.current.phase = generationCollecting | 159 | c.current.phase = generationCollecting | ||
| 159 | - | // A late drain response may still arrive after the batch marker. | 160 | + | // A late drain response may still arrive after the batch marker. Exclude |
| 161 | + | // OSC reports until its boundary is consumed: otherwise stale drain-era | |||
| 162 | + | // reports could become part of this generation's definitive snapshot. | |||
| 160 | c.current.expectedMarkers = 2 | 163 | c.current.expectedMarkers = 2 | ||
| 164 | + | c.current.awaitingLateDrainBoundary = true | |||
| 161 | return []paletteGenerationAction{ | 165 | return []paletteGenerationAction{ | ||
| 162 | {kind: paletteActionWriteBatch, id: event.id, bytes: paletteColorBatch}, | 166 | {kind: paletteActionWriteBatch, id: event.id, bytes: paletteColorBatch}, | ||
| 163 | {kind: paletteActionArmCompletionDeadline, id: event.id, deadline: paletteGenerationDeadline}, | 167 | {kind: paletteActionArmCompletionDeadline, id: event.id, deadline: paletteGenerationDeadline}, | ||
| @@ -168,17 +172,17 @@ func (c *paletteGenerationCoordinator) handle(event paletteGenerationEvent) []pa | |||||
| 168 | } | 172 | } | ||
| 169 | return c.finalize() | 173 | return c.finalize() | ||
| 170 | case paletteEventForeground: | 174 | case paletteEventForeground: | ||
| 171 | - | if c.current.phase == generationCollecting { | 175 | + | if c.current.phase == generationCollecting && !c.current.awaitingLateDrainBoundary { |
| 172 | c.current.theme.HasForeground = true | 176 | c.current.theme.HasForeground = true | ||
| 173 | c.current.theme.Foreground = event.rgb | 177 | c.current.theme.Foreground = event.rgb | ||
| 174 | } | 178 | } | ||
| 175 | case paletteEventBackground: | 179 | case paletteEventBackground: | ||
| 176 | - | if c.current.phase == generationCollecting { | 180 | + | if c.current.phase == generationCollecting && !c.current.awaitingLateDrainBoundary { |
| 177 | c.current.theme.HasBackground = true | 181 | c.current.theme.HasBackground = true | ||
| 178 | c.current.theme.Background = event.rgb | 182 | c.current.theme.Background = event.rgb | ||
| 179 | } | 183 | } | ||
| 180 | case paletteEventPalette: | 184 | case paletteEventPalette: | ||
| 181 | - | if c.current.phase == generationCollecting && event.slot < 16 { | 185 | + | if c.current.phase == generationCollecting && !c.current.awaitingLateDrainBoundary && event.slot < 16 { |
| 182 | c.current.theme.PaletteKnown |= uint16(1) << event.slot | 186 | c.current.theme.PaletteKnown |= uint16(1) << event.slot | ||
| 183 | c.current.theme.Palette[event.slot] = event.rgb | 187 | c.current.theme.Palette[event.slot] = event.rgb | ||
| 184 | } | 188 | } | ||
| @@ -200,6 +204,9 @@ func (c *paletteGenerationCoordinator) onMarker() []paletteGenerationAction { | |||||
| 200 | if c.current.expectedMarkers == 0 { | 204 | if c.current.expectedMarkers == 0 { | ||
| 201 | c.current.expectedMarkers = 1 | 205 | c.current.expectedMarkers = 1 | ||
| 202 | } | 206 | } | ||
| 207 | + | if c.current.awaitingLateDrainBoundary { | |||
| 208 | + | c.current.awaitingLateDrainBoundary = false | |||
| 209 | + | } | |||
| 203 | c.current.expectedMarkers-- | 210 | c.current.expectedMarkers-- | ||
| 204 | if c.current.expectedMarkers == 0 { | 211 | if c.current.expectedMarkers == 0 { | ||
| 205 | return c.finalize() | 212 | return c.finalize() | ||
M internal/usecase/client/palette_generation_test.go +15 -2
| @@ -149,7 +149,7 @@ func TestPaletteGenerationReplacementWaitsForDrain(t *testing.T) { | |||
| 149 | 149 | }, actionKinds(batch)) | |
| 150 | 150 | } | |
| 151 | 151 | ||
| 152 | - | func TestPaletteGenerationLateDrainMarkerNeedsCompletionMarker(t *testing.T) { | |
| 152 | + | func TestPaletteGenerationLateDrainBoundaryExcludesStaleOSC(t *testing.T) { | |
| 153 | 153 | g := newPaletteGenerationCoordinator() | |
| 154 | 154 | clock := newManualPaletteClock() | |
| 155 | 155 | start := g.start(generationTheme(), true) | |
| @@ -163,11 +163,24 @@ func TestPaletteGenerationLateDrainMarkerNeedsCompletionMarker(t *testing.T) { | |||
| 163 | 163 | paletteActionArmCompletionDeadline, | |
| 164 | 164 | }, actionKinds(timedOut)) | |
| 165 | 165 | ||
| 166 | + | // The drain reply may be late, so reports received before its boundary are | |
| 167 | + | // conservatively excluded from this generation's accumulator. | |
| 168 | + | g.foreground(id, renderer.RGB{R: 1, G: 2, B: 3}) | |
| 169 | + | g.background(id, renderer.RGB{R: 4, G: 5, B: 6}) | |
| 170 | + | g.palette(id, 2, renderer.RGB{R: 7, G: 8, B: 9}) | |
| 171 | + | ||
| 166 | 172 | lateDrain := g.marker(id) | |
| 167 | 173 | require.Empty(t, lateDrain, "the late drain response must not finalize the batch") | |
| 174 | + | g.foreground(id, renderer.RGB{R: 10, G: 11, B: 12}) | |
| 175 | + | g.background(id, renderer.RGB{R: 13, G: 14, B: 15}) | |
| 176 | + | g.palette(id, 3, renderer.RGB{R: 16, G: 17, B: 18}) | |
| 168 | 177 | complete := g.marker(id) | |
| 169 | 178 | clock.apply(complete) | |
| 170 | - | require.Equal(t, paletteActionPublishFinal, complete[len(complete)-1].kind) | |
| 179 | + | final := findAction(t, complete, paletteActionPublishFinal).theme | |
| 180 | + | require.Equal(t, renderer.RGB{R: 10, G: 11, B: 12}, final.Foreground) | |
| 181 | + | require.Equal(t, renderer.RGB{R: 13, G: 14, B: 15}, final.Background) | |
| 182 | + | require.Equal(t, uint16(1<<3), final.PaletteKnown) | |
| 183 | + | require.Equal(t, renderer.RGB{R: 16, G: 17, B: 18}, final.Palette[3]) | |
| 171 | 184 | } | |
| 172 | 185 | ||
| 173 | 186 | func TestPaletteGenerationCompletionDeadlineFinalizesAfterMissingLateDrain(t *testing.T) { | |
| @@ -149,7 +149,7 @@ func TestPaletteGenerationReplacementWaitsForDrain(t *testing.T) { | |||||
| 149 | }, actionKinds(batch)) | 149 | }, actionKinds(batch)) | ||
| 150 | } | 150 | } | ||
| 151 | 151 | ||||
| 152 | - | func TestPaletteGenerationLateDrainMarkerNeedsCompletionMarker(t *testing.T) { | 152 | + | func TestPaletteGenerationLateDrainBoundaryExcludesStaleOSC(t *testing.T) { |
| 153 | g := newPaletteGenerationCoordinator() | 153 | g := newPaletteGenerationCoordinator() | ||
| 154 | clock := newManualPaletteClock() | 154 | clock := newManualPaletteClock() | ||
| 155 | start := g.start(generationTheme(), true) | 155 | start := g.start(generationTheme(), true) | ||
| @@ -163,11 +163,24 @@ func TestPaletteGenerationLateDrainMarkerNeedsCompletionMarker(t *testing.T) { | |||||
| 163 | paletteActionArmCompletionDeadline, | 163 | paletteActionArmCompletionDeadline, | ||
| 164 | }, actionKinds(timedOut)) | 164 | }, actionKinds(timedOut)) | ||
| 165 | 165 | ||||
| 166 | + | // The drain reply may be late, so reports received before its boundary are | |||
| 167 | + | // conservatively excluded from this generation's accumulator. | |||
| 168 | + | g.foreground(id, renderer.RGB{R: 1, G: 2, B: 3}) | |||
| 169 | + | g.background(id, renderer.RGB{R: 4, G: 5, B: 6}) | |||
| 170 | + | g.palette(id, 2, renderer.RGB{R: 7, G: 8, B: 9}) | |||
| 171 | + | ||||
| 166 | lateDrain := g.marker(id) | 172 | lateDrain := g.marker(id) | ||
| 167 | require.Empty(t, lateDrain, "the late drain response must not finalize the batch") | 173 | require.Empty(t, lateDrain, "the late drain response must not finalize the batch") | ||
| 174 | + | g.foreground(id, renderer.RGB{R: 10, G: 11, B: 12}) | |||
| 175 | + | g.background(id, renderer.RGB{R: 13, G: 14, B: 15}) | |||
| 176 | + | g.palette(id, 3, renderer.RGB{R: 16, G: 17, B: 18}) | |||
| 168 | complete := g.marker(id) | 177 | complete := g.marker(id) | ||
| 169 | clock.apply(complete) | 178 | clock.apply(complete) | ||
| 170 | - | require.Equal(t, paletteActionPublishFinal, complete[len(complete)-1].kind) | 179 | + | final := findAction(t, complete, paletteActionPublishFinal).theme |
| 180 | + | require.Equal(t, renderer.RGB{R: 10, G: 11, B: 12}, final.Foreground) | |||
| 181 | + | require.Equal(t, renderer.RGB{R: 13, G: 14, B: 15}, final.Background) | |||
| 182 | + | require.Equal(t, uint16(1<<3), final.PaletteKnown) | |||
| 183 | + | require.Equal(t, renderer.RGB{R: 16, G: 17, B: 18}, final.Palette[3]) | |||
| 171 | } | 184 | } | ||
| 172 | 185 | ||||
| 173 | func TestPaletteGenerationCompletionDeadlineFinalizesAfterMissingLateDrain(t *testing.T) { | 186 | func TestPaletteGenerationCompletionDeadlineFinalizesAfterMissingLateDrain(t *testing.T) { | ||
M internal/usecase/client/reconnect_toast_test.go +0 -84
| @@ -511,90 +511,6 @@ func TestRemoteReconnectToastLifecycleWithWrappedTransportError(t *testing.T) { | |||
| 511 | 511 | require.Contains(t, out, strings.Repeat(" ", reconnectToastBounds(term.size).Width)) | |
| 512 | 512 | } | |
| 513 | 513 | ||
| 514 | - | type reconnectThemeTransport struct { | |
| 515 | - | resumeToken uint64 | |
| 516 | - | disconnectAfterTheme bool | |
| 517 | - | recvCount int | |
| 518 | - | fullTheme chan ports.Theme | |
| 519 | - | observedTheme ports.Theme | |
| 520 | - | } | |
| 521 | - | ||
| 522 | - | func newReconnectThemeTransport(token uint64, disconnectAfterTheme bool) *reconnectThemeTransport { | |
| 523 | - | return &reconnectThemeTransport{ | |
| 524 | - | resumeToken: token, | |
| 525 | - | disconnectAfterTheme: disconnectAfterTheme, | |
| 526 | - | fullTheme: make(chan ports.Theme, 1), | |
| 527 | - | } | |
| 528 | - | } | |
| 529 | - | ||
| 530 | - | func (t *reconnectThemeTransport) Send(f ports.Frame) error { | |
| 531 | - | if f.Type != ports.MsgTheme { | |
| 532 | - | return nil | |
| 533 | - | } | |
| 534 | - | theme, err := ports.UnmarshalTheme(f.Payload) | |
| 535 | - | if err != nil { | |
| 536 | - | return err | |
| 537 | - | } | |
| 538 | - | if theme.HasForeground && theme.HasBackground { | |
| 539 | - | select { | |
| 540 | - | case t.fullTheme <- theme: | |
| 541 | - | default: | |
| 542 | - | } | |
| 543 | - | } | |
| 544 | - | return nil | |
| 545 | - | } | |
| 546 | - | ||
| 547 | - | func (t *reconnectThemeTransport) Recv() (ports.Frame, error) { | |
| 548 | - | t.recvCount++ | |
| 549 | - | if t.recvCount == 1 { | |
| 550 | - | return reconnectToastWelcome(t.resumeToken), nil | |
| 551 | - | } | |
| 552 | - | t.observedTheme = <-t.fullTheme | |
| 553 | - | if t.disconnectAfterTheme { | |
| 554 | - | return ports.Frame{}, io.EOF | |
| 555 | - | } | |
| 556 | - | return reconnectToastDetach(ports.ReasonDetach), nil | |
| 557 | - | } | |
| 558 | - | ||
| 559 | - | func (t *reconnectThemeTransport) Close() error { return nil } | |
| 560 | - | ||
| 561 | - | type blockingThemeRestoreTransport struct { | |
| 562 | - | started chan struct{} | |
| 563 | - | closed chan struct{} | |
| 564 | - | recvs int | |
| 565 | - | } | |
| 566 | - | ||
| 567 | - | func newBlockingThemeRestoreTransport() *blockingThemeRestoreTransport { | |
| 568 | - | return &blockingThemeRestoreTransport{started: make(chan struct{}), closed: make(chan struct{})} | |
| 569 | - | } | |
| 570 | - | ||
| 571 | - | func (t *blockingThemeRestoreTransport) Send(f ports.Frame) error { | |
| 572 | - | if f.Type != ports.MsgTheme { | |
| 573 | - | return nil | |
| 574 | - | } | |
| 575 | - | close(t.started) | |
| 576 | - | <-t.closed | |
| 577 | - | return io.ErrClosedPipe | |
| 578 | - | } | |
| 579 | - | ||
| 580 | - | func (t *blockingThemeRestoreTransport) Recv() (ports.Frame, error) { | |
| 581 | - | t.recvs++ | |
| 582 | - | if t.recvs == 1 { | |
| 583 | - | return reconnectToastWelcome(44), nil | |
| 584 | - | } | |
| 585 | - | <-t.closed | |
| 586 | - | return ports.Frame{}, io.EOF | |
| 587 | - | } | |
| 588 | - | ||
| 589 | - | func (t *blockingThemeRestoreTransport) Close() error { | |
| 590 | - | select { | |
| 591 | - | case <-t.closed: | |
| 592 | - | default: | |
| 593 | - | close(t.closed) | |
| 594 | - | } | |
| 595 | - | return nil | |
| 596 | - | } | |
| 597 | - | ||
| 598 | 514 | func TestRemoteEphemeralReconnectUsesAssignedSessionName(t *testing.T) { | |
| 599 | 515 | linkDead := errors.New("remote link dead") | |
| 600 | 516 | oldSleep := reconnectSleep | |
| @@ -511,90 +511,6 @@ func TestRemoteReconnectToastLifecycleWithWrappedTransportError(t *testing.T) { | |||||
| 511 | require.Contains(t, out, strings.Repeat(" ", reconnectToastBounds(term.size).Width)) | 511 | require.Contains(t, out, strings.Repeat(" ", reconnectToastBounds(term.size).Width)) | ||
| 512 | } | 512 | } | ||
| 513 | 513 | ||||
| 514 | - | type reconnectThemeTransport struct { | |||
| 515 | - | resumeToken uint64 | |||
| 516 | - | disconnectAfterTheme bool | |||
| 517 | - | recvCount int | |||
| 518 | - | fullTheme chan ports.Theme | |||
| 519 | - | observedTheme ports.Theme | |||
| 520 | - | } | |||
| 521 | - | ||||
| 522 | - | func newReconnectThemeTransport(token uint64, disconnectAfterTheme bool) *reconnectThemeTransport { | |||
| 523 | - | return &reconnectThemeTransport{ | |||
| 524 | - | resumeToken: token, | |||
| 525 | - | disconnectAfterTheme: disconnectAfterTheme, | |||
| 526 | - | fullTheme: make(chan ports.Theme, 1), | |||
| 527 | - | } | |||
| 528 | - | } | |||
| 529 | - | ||||
| 530 | - | func (t *reconnectThemeTransport) Send(f ports.Frame) error { | |||
| 531 | - | if f.Type != ports.MsgTheme { | |||
| 532 | - | return nil | |||
| 533 | - | } | |||
| 534 | - | theme, err := ports.UnmarshalTheme(f.Payload) | |||
| 535 | - | if err != nil { | |||
| 536 | - | return err | |||
| 537 | - | } | |||
| 538 | - | if theme.HasForeground && theme.HasBackground { | |||
| 539 | - | select { | |||
| 540 | - | case t.fullTheme <- theme: | |||
| 541 | - | default: | |||
| 542 | - | } | |||
| 543 | - | } | |||
| 544 | - | return nil | |||
| 545 | - | } | |||
| 546 | - | ||||
| 547 | - | func (t *reconnectThemeTransport) Recv() (ports.Frame, error) { | |||
| 548 | - | t.recvCount++ | |||
| 549 | - | if t.recvCount == 1 { | |||
| 550 | - | return reconnectToastWelcome(t.resumeToken), nil | |||
| 551 | - | } | |||
| 552 | - | t.observedTheme = <-t.fullTheme | |||
| 553 | - | if t.disconnectAfterTheme { | |||
| 554 | - | return ports.Frame{}, io.EOF | |||
| 555 | - | } | |||
| 556 | - | return reconnectToastDetach(ports.ReasonDetach), nil | |||
| 557 | - | } | |||
| 558 | - | ||||
| 559 | - | func (t *reconnectThemeTransport) Close() error { return nil } | |||
| 560 | - | ||||
| 561 | - | type blockingThemeRestoreTransport struct { | |||
| 562 | - | started chan struct{} | |||
| 563 | - | closed chan struct{} | |||
| 564 | - | recvs int | |||
| 565 | - | } | |||
| 566 | - | ||||
| 567 | - | func newBlockingThemeRestoreTransport() *blockingThemeRestoreTransport { | |||
| 568 | - | return &blockingThemeRestoreTransport{started: make(chan struct{}), closed: make(chan struct{})} | |||
| 569 | - | } | |||
| 570 | - | ||||
| 571 | - | func (t *blockingThemeRestoreTransport) Send(f ports.Frame) error { | |||
| 572 | - | if f.Type != ports.MsgTheme { | |||
| 573 | - | return nil | |||
| 574 | - | } | |||
| 575 | - | close(t.started) | |||
| 576 | - | <-t.closed | |||
| 577 | - | return io.ErrClosedPipe | |||
| 578 | - | } | |||
| 579 | - | ||||
| 580 | - | func (t *blockingThemeRestoreTransport) Recv() (ports.Frame, error) { | |||
| 581 | - | t.recvs++ | |||
| 582 | - | if t.recvs == 1 { | |||
| 583 | - | return reconnectToastWelcome(44), nil | |||
| 584 | - | } | |||
| 585 | - | <-t.closed | |||
| 586 | - | return ports.Frame{}, io.EOF | |||
| 587 | - | } | |||
| 588 | - | ||||
| 589 | - | func (t *blockingThemeRestoreTransport) Close() error { | |||
| 590 | - | select { | |||
| 591 | - | case <-t.closed: | |||
| 592 | - | default: | |||
| 593 | - | close(t.closed) | |||
| 594 | - | } | |||
| 595 | - | return nil | |||
| 596 | - | } | |||
| 597 | - | ||||
| 598 | func TestRemoteEphemeralReconnectUsesAssignedSessionName(t *testing.T) { | 514 | func TestRemoteEphemeralReconnectUsesAssignedSessionName(t *testing.T) { | ||
| 599 | linkDead := errors.New("remote link dead") | 515 | linkDead := errors.New("remote link dead") | ||
| 600 | oldSleep := reconnectSleep | 516 | oldSleep := reconnectSleep | ||