fix(client): keep resume keys across finish, drop at picker; kitty cancel; ack unqueued reads
- finish stays lossless; the supervisor drops unclaimed input when the terminal returns to the picker, so keys never cross owner classes. - Decode the kitty keyboard protocol before matching Ctrl-C/Esc on resume. - Ack a read that queued nothing at once, so an overlay is not held behind a stalled write.
- commit
- f5e4a3eb1543dd2e9b9093b93fde904c733729a6
- author
- bnema · · 2026-10-06 09:48 UTC
- parent
- a4e74c2
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +117 -10
M internal/usecase/client/attachment_outbox_test.go +24 -0
| @@ -107,6 +107,9 @@ func TestResumeWatch(t *testing.T) { | |||
| 107 | 107 | }{ | |
| 108 | 108 | {name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true}, | |
| 109 | 109 | {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true}, | |
| 110 | + | {name: "kitty esc cancels", reads: []string{"\x1b[27u"}, wantCancel: true}, | |
| 111 | + | {name: "kitty ctrl-c cancels", reads: []string{"\x1b[99;5u"}, wantCancel: true}, | |
| 112 | + | {name: "kitty ctrl-c with lock bits cancels", reads: []string{"\x1b[99;69u"}, wantCancel: true}, | |
| 110 | 113 | {name: "escape sequence is kept", reads: []string{"\x1b[A"}, wantResidual: "\x1b[A"}, | |
| 111 | 114 | {name: "typed keys are kept in order", reads: []string{"l", "s", "\r"}, wantResidual: "ls\r"}, | |
| 112 | 115 | } | |
| @@ -192,6 +195,27 @@ func TestAttachmentWorkerStalledLink(t *testing.T) { | |||
| 192 | 195 | h.end(t) | |
| 193 | 196 | }, | |
| 194 | 197 | }, | |
| 198 | + | { | |
| 199 | + | name: "a read that queues nothing is acked while an earlier write is stalled", | |
| 200 | + | run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) { | |
| 201 | + | // A queued output Ack stalls behind the link. | |
| 202 | + | next := sessionTestOutput(2, "\x1b[Hmore") | |
| 203 | + | next.Epoch = 2 | |
| 204 | + | h.stream.deliver(next) | |
| 205 | + | <-s.entered | |
| 206 | + | // An open bracketed paste is held by the coalescer and queues | |
| 207 | + | // nothing, so there is no write to wait for. | |
| 208 | + | h.send("\x1b[200~abc") | |
| 209 | + | h.send("def") | |
| 210 | + | require.Eventually(t, func() bool { | |
| 211 | + | h.pump.mu.Lock() | |
| 212 | + | defer h.pump.mu.Unlock() | |
| 213 | + | return h.pump.pending == nil && h.pump.delivering == 0 && len(h.reader.chunks) == 0 | |
| 214 | + | }, 5*time.Second, time.Millisecond, "an unqueued read held the pump behind a stalled write") | |
| 215 | + | close(s.release) | |
| 216 | + | h.end(t) | |
| 217 | + | }, | |
| 218 | + | }, | |
| 195 | 219 | { | |
| 196 | 220 | name: "keys behind a failed write are preserved for the next attempt", | |
| 197 | 221 | run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) { | |
| @@ -107,6 +107,9 @@ func TestResumeWatch(t *testing.T) { | |||||
| 107 | }{ | 107 | }{ | ||
| 108 | {name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true}, | 108 | {name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true}, | ||
| 109 | {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true}, | 109 | {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true}, | ||
| 110 | + | {name: "kitty esc cancels", reads: []string{"\x1b[27u"}, wantCancel: true}, | |||
| 111 | + | {name: "kitty ctrl-c cancels", reads: []string{"\x1b[99;5u"}, wantCancel: true}, | |||
| 112 | + | {name: "kitty ctrl-c with lock bits cancels", reads: []string{"\x1b[99;69u"}, wantCancel: true}, | |||
| 110 | {name: "escape sequence is kept", reads: []string{"\x1b[A"}, wantResidual: "\x1b[A"}, | 113 | {name: "escape sequence is kept", reads: []string{"\x1b[A"}, wantResidual: "\x1b[A"}, | ||
| 111 | {name: "typed keys are kept in order", reads: []string{"l", "s", "\r"}, wantResidual: "ls\r"}, | 114 | {name: "typed keys are kept in order", reads: []string{"l", "s", "\r"}, wantResidual: "ls\r"}, | ||
| 112 | } | 115 | } | ||
| @@ -192,6 +195,27 @@ func TestAttachmentWorkerStalledLink(t *testing.T) { | |||||
| 192 | h.end(t) | 195 | h.end(t) | ||
| 193 | }, | 196 | }, | ||
| 194 | }, | 197 | }, | ||
| 198 | + | { | |||
| 199 | + | name: "a read that queues nothing is acked while an earlier write is stalled", | |||
| 200 | + | run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) { | |||
| 201 | + | // A queued output Ack stalls behind the link. | |||
| 202 | + | next := sessionTestOutput(2, "\x1b[Hmore") | |||
| 203 | + | next.Epoch = 2 | |||
| 204 | + | h.stream.deliver(next) | |||
| 205 | + | <-s.entered | |||
| 206 | + | // An open bracketed paste is held by the coalescer and queues | |||
| 207 | + | // nothing, so there is no write to wait for. | |||
| 208 | + | h.send("\x1b[200~abc") | |||
| 209 | + | h.send("def") | |||
| 210 | + | require.Eventually(t, func() bool { | |||
| 211 | + | h.pump.mu.Lock() | |||
| 212 | + | defer h.pump.mu.Unlock() | |||
| 213 | + | return h.pump.pending == nil && h.pump.delivering == 0 && len(h.reader.chunks) == 0 | |||
| 214 | + | }, 5*time.Second, time.Millisecond, "an unqueued read held the pump behind a stalled write") | |||
| 215 | + | close(s.release) | |||
| 216 | + | h.end(t) | |||
| 217 | + | }, | |||
| 218 | + | }, | |||
| 195 | { | 219 | { | ||
| 196 | name: "keys behind a failed write are preserved for the next attempt", | 220 | name: "keys behind a failed write are preserved for the next attempt", | ||
| 197 | run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) { | 221 | run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) { | ||
M internal/usecase/client/attachment_session.go +12 -1
| @@ -382,7 +382,18 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||
| 382 | 382 | // await makes the delivery just routed pending until the outbox wrote | |
| 383 | 383 | // every message accepted since from. | |
| 384 | 384 | await := func(from, actionID uint64) error { | |
| 385 | - | pending = &pendingInputAck{from: from, mark: stream.accepted(), actionID: actionID} | |
| 385 | + | mark := stream.accepted() | |
| 386 | + | if mark == from { | |
| 387 | + | // Nothing was queued (an overlay consumed the keys), so there is | |
| 388 | + | // no write to wait for: an earlier stalled message must not hold | |
| 389 | + | // the overlay's next key. | |
| 390 | + | fg.AckInput() | |
| 391 | + | if actionID != 0 { | |
| 392 | + | return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID}) | |
| 393 | + | } | |
| 394 | + | return nil | |
| 395 | + | } | |
| 396 | + | pending = &pendingInputAck{from: from, mark: mark, actionID: actionID} | |
| 386 | 397 | return ackWritten() | |
| 387 | 398 | } | |
| 388 | 399 | for { | |
| @@ -382,7 +382,18 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||||
| 382 | // await makes the delivery just routed pending until the outbox wrote | 382 | // await makes the delivery just routed pending until the outbox wrote | ||
| 383 | // every message accepted since from. | 383 | // every message accepted since from. | ||
| 384 | await := func(from, actionID uint64) error { | 384 | await := func(from, actionID uint64) error { | ||
| 385 | - | pending = &pendingInputAck{from: from, mark: stream.accepted(), actionID: actionID} | 385 | + | mark := stream.accepted() |
| 386 | + | if mark == from { | |||
| 387 | + | // Nothing was queued (an overlay consumed the keys), so there is | |||
| 388 | + | // no write to wait for: an earlier stalled message must not hold | |||
| 389 | + | // the overlay's next key. | |||
| 390 | + | fg.AckInput() | |||
| 391 | + | if actionID != 0 { | |||
| 392 | + | return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID}) | |||
| 393 | + | } | |||
| 394 | + | return nil | |||
| 395 | + | } | |||
| 396 | + | pending = &pendingInputAck{from: from, mark: mark, actionID: actionID} | |||
| 386 | return ackWritten() | 397 | return ackWritten() | ||
| 387 | } | 398 | } | ||
| 388 | for { | 399 | for { | ||
M internal/usecase/client/attachment_worker.go +4 -6
| @@ -1231,12 +1231,10 @@ func (f *attachmentForeground) finish() { | |||
| 1231 | 1231 | return | |
| 1232 | 1232 | } | |
| 1233 | 1233 | f.finishedOnce.Do(func() { | |
| 1234 | - | // Autonomous supervisor foregrounds have a host-lifetime geometry | |
| 1235 | - | // collector. At that owner-class boundary, no attachment byte may reach | |
| 1236 | - | // the picker. Attachment-to-attachment replacement instead delivers pending input. | |
| 1237 | - | if f.input != nil && f.host != nil && f.host.ownerBoundary { | |
| 1238 | - | f.input.dropOwned(f.consumer) | |
| 1239 | - | } | |
| 1234 | + | // Finish is lossless: input kept for a resume must survive it. The | |
| 1235 | + | // supervisor drops unclaimed bytes (dropUnclaimed) when the terminal | |
| 1236 | + | // returns to the picker, the owner-class boundary no attachment byte | |
| 1237 | + | // may cross. | |
| 1240 | 1238 | f.authority.revoke(f) | |
| 1241 | 1239 | }) | |
| 1242 | 1240 | } | |
| @@ -1231,12 +1231,10 @@ func (f *attachmentForeground) finish() { | |||||
| 1231 | return | 1231 | return | ||
| 1232 | } | 1232 | } | ||
| 1233 | f.finishedOnce.Do(func() { | 1233 | f.finishedOnce.Do(func() { | ||
| 1234 | - | // Autonomous supervisor foregrounds have a host-lifetime geometry | 1234 | + | // Finish is lossless: input kept for a resume must survive it. The |
| 1235 | - | // collector. At that owner-class boundary, no attachment byte may reach | 1235 | + | // supervisor drops unclaimed bytes (dropUnclaimed) when the terminal |
| 1236 | - | // the picker. Attachment-to-attachment replacement instead delivers pending input. | 1236 | + | // returns to the picker, the owner-class boundary no attachment byte |
| 1237 | - | if f.input != nil && f.host != nil && f.host.ownerBoundary { | 1237 | + | // may cross. |
| 1238 | - | f.input.dropOwned(f.consumer) | |||
| 1239 | - | } | |||
| 1240 | f.authority.revoke(f) | 1238 | f.authority.revoke(f) | ||
| 1241 | }) | 1239 | }) | ||
| 1242 | } | 1240 | } | ||
M internal/usecase/client/lifecycle_p54a_test.go +49 -0
| @@ -311,6 +311,9 @@ func TestP54AttachmentFinalizeDropsUnleasedPendingBeforePicker(t *testing.T) { | |||
| 311 | 311 | return lifetime.pump.pending != nil && lifetime.pump.delivering == 0 | |
| 312 | 312 | }, 5*time.Second, time.Millisecond) | |
| 313 | 313 | fg.finish() | |
| 314 | + | // The supervisor drops unclaimed session input as the terminal returns to | |
| 315 | + | // the picker; finish itself stays lossless for a resume. | |
| 316 | + | lifetime.pump.dropUnclaimed() | |
| 314 | 317 | lifetime.acquirePicker() | |
| 315 | 318 | assertNoPickerRead(t, consumer.reads) | |
| 316 | 319 | } | |
| @@ -330,6 +333,9 @@ func TestP54AttachmentFinalizeDropsPreservedResidualBeforePicker(t *testing.T) { | |||
| 330 | 333 | require.True(t, ok) | |
| 331 | 334 | fg.PreserveInput(event.Data) | |
| 332 | 335 | fg.finish() | |
| 336 | + | // The supervisor drops unclaimed session input as the terminal returns to | |
| 337 | + | // the picker; finish itself stays lossless for a resume. | |
| 338 | + | lifetime.pump.dropUnclaimed() | |
| 333 | 339 | lifetime.acquirePicker() | |
| 334 | 340 | assertNoPickerRead(t, consumer.reads) | |
| 335 | 341 | } | |
| @@ -468,3 +474,46 @@ func TestP54LifecycleInterleavingsRace(t *testing.T) { | |||
| 468 | 474 | host.stopGeometry() | |
| 469 | 475 | } | |
| 470 | 476 | } | |
| 477 | + | ||
| 478 | + | // TestResumeKeysSurviveFinishButNotThePicker runs the production owner | |
| 479 | + | // boundary: input kept for a resume survives the attachment's finish, then is | |
| 480 | + | // dropped when the terminal returns to the picker instead of resuming. | |
| 481 | + | func TestResumeKeysSurviveFinishButNotThePicker(t *testing.T) { | |
| 482 | + | tests := []struct { | |
| 483 | + | name string | |
| 484 | + | backToPicker bool | |
| 485 | + | }{ | |
| 486 | + | {name: "resumed attachment replays kept keys"}, | |
| 487 | + | {name: "picker never sees kept keys", backToPicker: true}, | |
| 488 | + | } | |
| 489 | + | for _, tt := range tests { | |
| 490 | + | t.Run(tt.name, func(t *testing.T) { | |
| 491 | + | reader := newAttachTestReader() | |
| 492 | + | consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)} | |
| 493 | + | lifetime := startTerminalInputLifetime(reader, consumer) | |
| 494 | + | defer lifetime.stop() | |
| 495 | + | lifetime.releasePicker() | |
| 496 | + | host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump}) | |
| 497 | + | host.ownerBoundary = true | |
| 498 | + | fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream()) | |
| 499 | + | require.True(t, host.authority.grant(fg)) | |
| 500 | + | reader.push([]byte("ls\r")) | |
| 501 | + | event, ok := fg.Input(context.Background()) | |
| 502 | + | require.True(t, ok) | |
| 503 | + | fg.PreserveInput(event.Data) | |
| 504 | + | fg.finish() | |
| 505 | + | ||
| 506 | + | if tt.backToPicker { | |
| 507 | + | lifetime.pump.dropUnclaimed() | |
| 508 | + | lifetime.acquirePicker() | |
| 509 | + | assertNoPickerRead(t, consumer.reads) | |
| 510 | + | return | |
| 511 | + | } | |
| 512 | + | next := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream()) | |
| 513 | + | require.True(t, host.authority.grant(next)) | |
| 514 | + | replayed, ok := next.Input(context.Background()) | |
| 515 | + | require.True(t, ok) | |
| 516 | + | require.Equal(t, []byte("ls\r"), replayed.Data) | |
| 517 | + | }) | |
| 518 | + | } | |
| 519 | + | } | |
| @@ -311,6 +311,9 @@ func TestP54AttachmentFinalizeDropsUnleasedPendingBeforePicker(t *testing.T) { | |||||
| 311 | return lifetime.pump.pending != nil && lifetime.pump.delivering == 0 | 311 | return lifetime.pump.pending != nil && lifetime.pump.delivering == 0 | ||
| 312 | }, 5*time.Second, time.Millisecond) | 312 | }, 5*time.Second, time.Millisecond) | ||
| 313 | fg.finish() | 313 | fg.finish() | ||
| 314 | + | // The supervisor drops unclaimed session input as the terminal returns to | |||
| 315 | + | // the picker; finish itself stays lossless for a resume. | |||
| 316 | + | lifetime.pump.dropUnclaimed() | |||
| 314 | lifetime.acquirePicker() | 317 | lifetime.acquirePicker() | ||
| 315 | assertNoPickerRead(t, consumer.reads) | 318 | assertNoPickerRead(t, consumer.reads) | ||
| 316 | } | 319 | } | ||
| @@ -330,6 +333,9 @@ func TestP54AttachmentFinalizeDropsPreservedResidualBeforePicker(t *testing.T) { | |||||
| 330 | require.True(t, ok) | 333 | require.True(t, ok) | ||
| 331 | fg.PreserveInput(event.Data) | 334 | fg.PreserveInput(event.Data) | ||
| 332 | fg.finish() | 335 | fg.finish() | ||
| 336 | + | // The supervisor drops unclaimed session input as the terminal returns to | |||
| 337 | + | // the picker; finish itself stays lossless for a resume. | |||
| 338 | + | lifetime.pump.dropUnclaimed() | |||
| 333 | lifetime.acquirePicker() | 339 | lifetime.acquirePicker() | ||
| 334 | assertNoPickerRead(t, consumer.reads) | 340 | assertNoPickerRead(t, consumer.reads) | ||
| 335 | } | 341 | } | ||
| @@ -468,3 +474,46 @@ func TestP54LifecycleInterleavingsRace(t *testing.T) { | |||||
| 468 | host.stopGeometry() | 474 | host.stopGeometry() | ||
| 469 | } | 475 | } | ||
| 470 | } | 476 | } | ||
| 477 | + | ||||
| 478 | + | // TestResumeKeysSurviveFinishButNotThePicker runs the production owner | |||
| 479 | + | // boundary: input kept for a resume survives the attachment's finish, then is | |||
| 480 | + | // dropped when the terminal returns to the picker instead of resuming. | |||
| 481 | + | func TestResumeKeysSurviveFinishButNotThePicker(t *testing.T) { | |||
| 482 | + | tests := []struct { | |||
| 483 | + | name string | |||
| 484 | + | backToPicker bool | |||
| 485 | + | }{ | |||
| 486 | + | {name: "resumed attachment replays kept keys"}, | |||
| 487 | + | {name: "picker never sees kept keys", backToPicker: true}, | |||
| 488 | + | } | |||
| 489 | + | for _, tt := range tests { | |||
| 490 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 491 | + | reader := newAttachTestReader() | |||
| 492 | + | consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)} | |||
| 493 | + | lifetime := startTerminalInputLifetime(reader, consumer) | |||
| 494 | + | defer lifetime.stop() | |||
| 495 | + | lifetime.releasePicker() | |||
| 496 | + | host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump}) | |||
| 497 | + | host.ownerBoundary = true | |||
| 498 | + | fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream()) | |||
| 499 | + | require.True(t, host.authority.grant(fg)) | |||
| 500 | + | reader.push([]byte("ls\r")) | |||
| 501 | + | event, ok := fg.Input(context.Background()) | |||
| 502 | + | require.True(t, ok) | |||
| 503 | + | fg.PreserveInput(event.Data) | |||
| 504 | + | fg.finish() | |||
| 505 | + | ||||
| 506 | + | if tt.backToPicker { | |||
| 507 | + | lifetime.pump.dropUnclaimed() | |||
| 508 | + | lifetime.acquirePicker() | |||
| 509 | + | assertNoPickerRead(t, consumer.reads) | |||
| 510 | + | return | |||
| 511 | + | } | |||
| 512 | + | next := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream()) | |||
| 513 | + | require.True(t, host.authority.grant(next)) | |||
| 514 | + | replayed, ok := next.Input(context.Background()) | |||
| 515 | + | require.True(t, ok) | |||
| 516 | + | require.Equal(t, []byte("ls\r"), replayed.Data) | |||
| 517 | + | }) | |||
| 518 | + | } | |||
| 519 | + | } | |||
M internal/usecase/client/supervisor.go +4 -3
| @@ -1524,10 +1524,11 @@ func (w *resumeWatch) hold(data []byte) { | |||
| 1524 | 1524 | } | |
| 1525 | 1525 | } | |
| 1526 | 1526 | ||
| 1527 | - | // isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc. A | |
| 1528 | - | // longer read starting with Esc is an escape sequence (arrows, focus), not a | |
| 1529 | - | // cancel. | |
| 1527 | + | // isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc, | |
| 1528 | + | // after decoding the kitty keyboard protocol (Esc is \x1b[27u there). A longer | |
| 1529 | + | // read starting with Esc is an escape sequence (arrows, focus), not a cancel. | |
| 1530 | 1530 | func isResumeCancelKey(data []byte) bool { | |
| 1531 | + | data = kittykey.Translate(data, 0) | |
| 1531 | 1532 | return len(data) == 1 && (data[0] == 0x03 || data[0] == 0x1b) | |
| 1532 | 1533 | } | |
| 1533 | 1534 | ||
| @@ -1524,10 +1524,11 @@ func (w *resumeWatch) hold(data []byte) { | |||||
| 1524 | } | 1524 | } | ||
| 1525 | } | 1525 | } | ||
| 1526 | 1526 | ||||
| 1527 | - | // isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc. A | 1527 | + | // isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc, |
| 1528 | - | // longer read starting with Esc is an escape sequence (arrows, focus), not a | 1528 | + | // after decoding the kitty keyboard protocol (Esc is \x1b[27u there). A longer |
| 1529 | - | // cancel. | 1529 | + | // read starting with Esc is an escape sequence (arrows, focus), not a cancel. |
| 1530 | func isResumeCancelKey(data []byte) bool { | 1530 | func isResumeCancelKey(data []byte) bool { | ||
| 1531 | + | data = kittykey.Translate(data, 0) | |||
| 1531 | return len(data) == 1 && (data[0] == 0x03 || data[0] == 0x1b) | 1532 | return len(data) == 1 && (data[0] == 0x03 || data[0] == 0x1b) | ||
| 1532 | } | 1533 | } | ||
| 1533 | 1534 | ||||
M internal/usecase/client/supervisor_attach.go +3 -0
| @@ -583,6 +583,9 @@ func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalI | |||
| 583 | 583 | // picker catalogue. | |
| 584 | 584 | picker.ApplySnapshot(service.Snapshot()) | |
| 585 | 585 | picker.SetOwnsInput(true) | |
| 586 | + | if input != nil && input.pump != nil { | |
| 587 | + | input.pump.dropUnclaimed() | |
| 588 | + | } | |
| 586 | 589 | input.acquirePicker() | |
| 587 | 590 | }() | |
| 588 | 591 | // budget.resuming is true while the loop reconnects a lost attachment: | |
| @@ -583,6 +583,9 @@ func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalI | |||||
| 583 | // picker catalogue. | 583 | // picker catalogue. | ||
| 584 | picker.ApplySnapshot(service.Snapshot()) | 584 | picker.ApplySnapshot(service.Snapshot()) | ||
| 585 | picker.SetOwnsInput(true) | 585 | picker.SetOwnsInput(true) | ||
| 586 | + | if input != nil && input.pump != nil { | |||
| 587 | + | input.pump.dropUnclaimed() | |||
| 588 | + | } | |||
| 586 | input.acquirePicker() | 589 | input.acquirePicker() | ||
| 587 | }() | 590 | }() | ||
| 588 | // budget.resuming is true while the loop reconnects a lost attachment: | 591 | // budget.resuming is true while the loop reconnects a lost attachment: | ||
M internal/usecase/client/terminal_input_pump.go +21 -0
| @@ -284,6 +284,27 @@ func (p *terminalInputPump) dropOwned(consumer uint64) { | |||
| 284 | 284 | } | |
| 285 | 285 | } | |
| 286 | 286 | ||
| 287 | + | // dropUnclaimed disposes pending and preserved bytes while no consumer holds | |
| 288 | + | // the claim. The supervisor calls it when the terminal goes back to the | |
| 289 | + | // picker, so session input kept for a resume never reaches the picker. | |
| 290 | + | func (p *terminalInputPump) dropUnclaimed() { | |
| 291 | + | p.mu.Lock() | |
| 292 | + | if p.consumer != 0 { | |
| 293 | + | p.mu.Unlock() | |
| 294 | + | return | |
| 295 | + | } | |
| 296 | + | hadPending := p.pending != nil | |
| 297 | + | p.pending = nil | |
| 298 | + | p.residual = nil | |
| 299 | + | p.mu.Unlock() | |
| 300 | + | if hadPending { | |
| 301 | + | select { | |
| 302 | + | case p.space <- struct{}{}: | |
| 303 | + | default: | |
| 304 | + | } | |
| 305 | + | } | |
| 306 | + | } | |
| 307 | + | ||
| 287 | 308 | func (p *terminalInputPump) ack(consumer uint64) { | |
| 288 | 309 | p.mu.Lock() | |
| 289 | 310 | if p.consumer != consumer || p.delivering != consumer { | |
| @@ -284,6 +284,27 @@ func (p *terminalInputPump) dropOwned(consumer uint64) { | |||||
| 284 | } | 284 | } | ||
| 285 | } | 285 | } | ||
| 286 | 286 | ||||
| 287 | + | // dropUnclaimed disposes pending and preserved bytes while no consumer holds | |||
| 288 | + | // the claim. The supervisor calls it when the terminal goes back to the | |||
| 289 | + | // picker, so session input kept for a resume never reaches the picker. | |||
| 290 | + | func (p *terminalInputPump) dropUnclaimed() { | |||
| 291 | + | p.mu.Lock() | |||
| 292 | + | if p.consumer != 0 { | |||
| 293 | + | p.mu.Unlock() | |||
| 294 | + | return | |||
| 295 | + | } | |||
| 296 | + | hadPending := p.pending != nil | |||
| 297 | + | p.pending = nil | |||
| 298 | + | p.residual = nil | |||
| 299 | + | p.mu.Unlock() | |||
| 300 | + | if hadPending { | |||
| 301 | + | select { | |||
| 302 | + | case p.space <- struct{}{}: | |||
| 303 | + | default: | |||
| 304 | + | } | |||
| 305 | + | } | |||
| 306 | + | } | |||
| 307 | + | ||||
| 287 | func (p *terminalInputPump) ack(consumer uint64) { | 308 | func (p *terminalInputPump) ack(consumer uint64) { | ||
| 288 | p.mu.Lock() | 309 | p.mu.Lock() | ||
| 289 | if p.consumer != consumer || p.delivering != consumer { | 310 | if p.consumer != consumer || p.delivering != consumer { | ||