test(client): drop held resume keys at the picker through the supervisor

Run the real supervisor through a cancelled and a refused resume and check the picker never receives held keys; remove tests that called dropUnclaimed by hand instead of exercising it.

commit: 767263f9071d4598f3e8c81c0443658a580cbdcb
author: brice · 15d ago · 2026-10-07 07:01 UTC
parent: 0a8685a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +96 -43

M internal/usecase/client/lifecycle_p54a_test.go +23 -40


@@ -475,45 +475,28 @@ func TestP54LifecycleInterleavingsRace(t *testing.T) {
475	475		}
476	476		}
477	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()
	478	+	// TestResumeKeysSurviveFinish runs the production owner boundary: input kept
	479	+	// for a resume survives the attachment's finish and reaches the next
	480	+	// foreground. TestResumeHeldKeysNeverReachThePicker covers the picker side.
	481	+	func TestResumeKeysSurviveFinish(t *testing.T) {
	482	+	reader := newAttachTestReader()
	483	+	consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
	484	+	lifetime := startTerminalInputLifetime(reader, consumer)
	485	+	defer lifetime.stop()
	486	+	lifetime.releasePicker()
	487	+	host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump})
	488	+	host.ownerBoundary = true
	489	+	fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
	490	+	require.True(t, host.authority.grant(fg))
	491	+	reader.push([]byte("ls\r"))
	492	+	event, ok := fg.Input(context.Background())
	493	+	require.True(t, ok)
	494	+	fg.PreserveInput(event.Data)
	495	+	fg.finish()
505	496
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		-	}
	497	+	next := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream())
	498	+	require.True(t, host.authority.grant(next))
	499	+	replayed, ok := next.Input(context.Background())
	500	+	require.True(t, ok)
	501	+	require.Equal(t, []byte("ls\r"), replayed.Data)
519	502		}


@@ -475,45 +475,28 @@ func TestP54LifecycleInterleavingsRace(t *testing.T) {
475		}	475		}
476		}	476		}
477			477
478	-	// TestResumeKeysSurviveFinishButNotThePicker runs the production owner	478	+	// TestResumeKeysSurviveFinish runs the production owner boundary: input kept
479	-	// boundary: input kept for a resume survives the attachment's finish, then is	479	+	// for a resume survives the attachment's finish and reaches the next
480	-	// dropped when the terminal returns to the picker instead of resuming.	480	+	// foreground. TestResumeHeldKeysNeverReachThePicker covers the picker side.
481	-	func TestResumeKeysSurviveFinishButNotThePicker(t *testing.T) {	481	+	func TestResumeKeysSurviveFinish(t *testing.T) {
482	-	tests := []struct {	482	+	reader := newAttachTestReader()
483	-	name string	483	+	consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
484	-	backToPicker bool	484	+	lifetime := startTerminalInputLifetime(reader, consumer)
485	-	}{	485	+	defer lifetime.stop()
486	-	{name: "resumed attachment replays kept keys"},	486	+	lifetime.releasePicker()
487	-	{name: "picker never sees kept keys", backToPicker: true},	487	+	host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump})
488	-	}	488	+	host.ownerBoundary = true
489	-	for _, tt := range tests {	489	+	fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
490	-	t.Run(tt.name, func(t *testing.T) {	490	+	require.True(t, host.authority.grant(fg))
491	-	reader := newAttachTestReader()	491	+	reader.push([]byte("ls\r"))
492	-	consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}	492	+	event, ok := fg.Input(context.Background())
493	-	lifetime := startTerminalInputLifetime(reader, consumer)	493	+	require.True(t, ok)
494	-	defer lifetime.stop()	494	+	fg.PreserveInput(event.Data)
495	-	lifetime.releasePicker()	495	+	fg.finish()
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			496
506	-	if tt.backToPicker {	497	+	next := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream())
507	-	lifetime.pump.dropUnclaimed()	498	+	require.True(t, host.authority.grant(next))
508	-	lifetime.acquirePicker()	499	+	replayed, ok := next.Input(context.Background())
509	-	assertNoPickerRead(t, consumer.reads)	500	+	require.True(t, ok)
510	-	return	501	+	require.Equal(t, []byte("ls\r"), replayed.Data)
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		}	502		}

M internal/usecase/client/resume_open_test.go +67 -0


@@ -65,3 +65,70 @@ func TestResumeCancelDuringStreamOpen(t *testing.T) {
65	65		})
66	66		}
67	67		}
	68	+
	69	+	// TestResumeHeldKeysNeverReachThePicker types while a lost attachment
	70	+	// resumes, then returns to the picker either by cancelling or because the
	71	+	// resume fails for good, and proves the picker never receives the held keys
	72	+	// while it still receives keys typed after it took the terminal back.
	73	+	func TestResumeHeldKeysNeverReachThePicker(t *testing.T) {
	74	+	for _, tt := range []struct {
	75	+	name string
	76	+	cancel bool
	77	+	}{
	78	+	{name: "cancelled", cancel: true},
	79	+	{name: "resume refused"},
	80	+	} {
	81	+	t.Run(tt.name, func(t *testing.T) {
	82	+	assertHeldKeysDroppedAtPicker(t, tt.cancel)
	83	+	})
	84	+	}
	85	+	}
	86	+
	87	+	func assertHeldKeysDroppedAtPicker(t *testing.T, cancel bool) {
	88	+	picker := newAttachTestPicker()
	89	+	h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
	90	+	first := newSessionTestStream()
	91	+	opened := make(chan struct{}, 1)
	92	+	refuse := make(chan struct{})
	93	+	count := 0
	94	+	h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
	95	+	count++
	96	+	if count == 1 {
	97	+	return first, nil
	98	+	}
	99	+	opened <- struct{}{}
	100	+	select {
	101	+	case <-ctx.Done():
	102	+	return nil, ctx.Err()
	103	+	case <-refuse:
	104	+	return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "refused"}
	105	+	}
	106	+	})
	107	+	picker.commit(sessionTestRequest(false))
	108	+	deliverReadyStream(t, first)
	109	+	awaitAttachedState(t, h.sup)
	110	+	first.fail(testStreamLoss())
	111	+	require.Eventually(t, func() bool { return h.sup.State().Presentation == PresentConnecting }, 5*time.Second, time.Millisecond)
	112	+	pushKeys(t, h, "HELD\r")
	113	+	if cancel {
	114	+	h.reader.push([]byte("\x03"))
	115	+	} else {
	116	+	fireResumeTimerAfter(t, h.clock, 1, func() {})
	117	+	select {
	118	+	case <-opened:
	119	+	case <-time.After(5 * time.Second):
	120	+	t.Fatal("resume did not open")
	121	+	}
	122	+	close(refuse)
	123	+	}
	124	+	require.Eventually(t, picker.owns, 5*time.Second, time.Millisecond)
	125	+	h.reader.push([]byte("j"))
	126	+	require.Eventually(t, func() bool {
	127	+	picker.mu.Lock()
	128	+	defer picker.mu.Unlock()
	129	+	return len(picker.consumedBytes) > 0
	130	+	}, 5*time.Second, time.Millisecond)
	131	+	picker.mu.Lock()
	132	+	defer picker.mu.Unlock()
	133	+	require.Equal(t, "j", string(picker.consumedBytes), "held resume keys reached the picker")
	134	+	}


@@ -65,3 +65,70 @@ func TestResumeCancelDuringStreamOpen(t *testing.T) {
65	})	65		})
66	}	66		}
67	}	67		}
		68	+
		69	+	// TestResumeHeldKeysNeverReachThePicker types while a lost attachment
		70	+	// resumes, then returns to the picker either by cancelling or because the
		71	+	// resume fails for good, and proves the picker never receives the held keys
		72	+	// while it still receives keys typed after it took the terminal back.
		73	+	func TestResumeHeldKeysNeverReachThePicker(t *testing.T) {
		74	+	for _, tt := range []struct {
		75	+	name string
		76	+	cancel bool
		77	+	}{
		78	+	{name: "cancelled", cancel: true},
		79	+	{name: "resume refused"},
		80	+	} {
		81	+	t.Run(tt.name, func(t *testing.T) {
		82	+	assertHeldKeysDroppedAtPicker(t, tt.cancel)
		83	+	})
		84	+	}
		85	+	}
		86	+
		87	+	func assertHeldKeysDroppedAtPicker(t *testing.T, cancel bool) {
		88	+	picker := newAttachTestPicker()
		89	+	h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
		90	+	first := newSessionTestStream()
		91	+	opened := make(chan struct{}, 1)
		92	+	refuse := make(chan struct{})
		93	+	count := 0
		94	+	h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
		95	+	count++
		96	+	if count == 1 {
		97	+	return first, nil
		98	+	}
		99	+	opened <- struct{}{}
		100	+	select {
		101	+	case <-ctx.Done():
		102	+	return nil, ctx.Err()
		103	+	case <-refuse:
		104	+	return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "refused"}
		105	+	}
		106	+	})
		107	+	picker.commit(sessionTestRequest(false))
		108	+	deliverReadyStream(t, first)
		109	+	awaitAttachedState(t, h.sup)
		110	+	first.fail(testStreamLoss())
		111	+	require.Eventually(t, func() bool { return h.sup.State().Presentation == PresentConnecting }, 5*time.Second, time.Millisecond)
		112	+	pushKeys(t, h, "HELD\r")
		113	+	if cancel {
		114	+	h.reader.push([]byte("\x03"))
		115	+	} else {
		116	+	fireResumeTimerAfter(t, h.clock, 1, func() {})
		117	+	select {
		118	+	case <-opened:
		119	+	case <-time.After(5 * time.Second):
		120	+	t.Fatal("resume did not open")
		121	+	}
		122	+	close(refuse)
		123	+	}
		124	+	require.Eventually(t, picker.owns, 5*time.Second, time.Millisecond)
		125	+	h.reader.push([]byte("j"))
		126	+	require.Eventually(t, func() bool {
		127	+	picker.mu.Lock()
		128	+	defer picker.mu.Unlock()
		129	+	return len(picker.consumedBytes) > 0
		130	+	}, 5*time.Second, time.Millisecond)
		131	+	picker.mu.Lock()
		132	+	defer picker.mu.Unlock()
		133	+	require.Equal(t, "j", string(picker.consumedBytes), "held resume keys reached the picker")
		134	+	}

M internal/usecase/client/supervisor_attach_test.go +6 -3


@@ -265,8 +265,10 @@ type attachTestPicker struct {
265	265		opsReady chan struct{}
266	266		ownsInput bool
267	267		consumed int
268		-	op pickerOp
269		-	key string
	268	+	// consumedBytes records every byte the picker consumed, in order.
	269	+	consumedBytes []byte
	270	+	op pickerOp
	271	+	key string
270	272		// resolveErr makes ResolveCommit refuse locally, which is how a test reaches
271	273		// the local-refusal classification without dialing anything.
272	274		resolveErr error

@@ -288,13 +290,14 @@ func newAttachTestPicker() *attachTestPicker {
288	290
289	291		func (p *attachTestPicker) ApplySnapshot(ports.BrokerSnapshot) {}
290	292
291		-	func (p *attachTestPicker) ConsumeTerminalRead([]byte) bool {
	293	+	func (p *attachTestPicker) ConsumeTerminalRead(data []byte) bool {
292	294		p.mu.Lock()
293	295		defer p.mu.Unlock()
294	296		if !p.ownsInput {
295	297		return false
296	298		}
297	299		p.consumed++
	300	+	p.consumedBytes = append(p.consumedBytes, data...)
298	301		if p.consumeOp != (pickerOp{}) {
299	302		p.op, p.key, p.request = p.consumeOp, "row", p.consumeRequest
300	303		p.consumeOp = pickerOp{}

 	opsReady  chan struct{}
 	ownsInput bool
 	consumed  int
-	op        pickerOp
+	// consumedBytes records every byte the picker consumed, in order.
-	key       string
+	consumedBytes []byte
+	op            pickerOp
+	key           string
 	// resolveErr makes ResolveCommit refuse locally, which is how a test reaches
 	// the local-refusal classification without dialing anything.
 	resolveErr error
 func (p *attachTestPicker) ApplySnapshot(ports.BrokerSnapshot) {}
-func (p *attachTestPicker) ConsumeTerminalRead([]byte) bool {
+func (p *attachTestPicker) ConsumeTerminalRead(data []byte) bool {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	if !p.ownsInput {
 		return false
 	}
 	p.consumed++
+	p.consumedBytes = append(p.consumedBytes, data...)
 	if p.consumeOp != (pickerOp{}) {
 		p.op, p.key, p.request = p.consumeOp, "row", p.consumeRequest
 		p.consumeOp = pickerOp{}