fe83619 fix(client): replay the held prefix after a resume input overflow · vev

M internal/usecase/client/attachment_outbox_test.go +30 -0


@@ -282,6 +282,36 @@ func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
282	282		require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")
283	283		}
284	284
	285	+	// TestResumeWatchOverflowKeepsHeldPrefix overflows one phase and checks the
	286	+	// next phase still replays the gap-free prefix the first phase handed back.
	287	+	// TestResumeWatchHeldOverflowIsSticky covers dropping keys typed after it.
	288	+	func TestResumeWatchOverflowKeepsHeldPrefix(t *testing.T) {
	289	+	reader := &inputTestReader{chunks: make(chan []byte, 2)}
	290	+	pump := newTerminalInputPump(reader)
	291	+	pump.start()
	292	+	t.Cleanup(func() {
	293	+	pump.stop()
	294	+	close(reader.chunks)
	295	+	})
	296	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
	297	+	clock := newSupervisorTestClock()
	298	+
	299	+	first := lifetime.watchResume(clock, nil)
	300	+	reader.chunks <- []byte("ls")
	301	+	require.Eventually(t, func() bool { return first.heldLen() == 2 }, 5*time.Second, time.Millisecond)
	302	+	first.hold(bytes.Repeat([]byte("x"), resumeHeldInputLimit))
	303	+	require.True(t, lifetime.resumeOverflowed.Load())
	304	+	first.release()
	305	+
	306	+	second := lifetime.watchResume(clock, nil)
	307	+	require.Eventually(t, func() bool { return second.heldLen() == 2 }, 5*time.Second, time.Millisecond)
	308	+	second.release()
	309	+
	310	+	pump.mu.Lock()
	311	+	defer pump.mu.Unlock()
	312	+	require.Equal(t, "ls", string(pump.residual))
	313	+	}
	314	+
285	315		// stallingSessionStream passes writes through until stall is set; stalled
286	316		// writes block until release closes, then fail with failErr when set.
287	317		type stallingSessionStream struct {


@@ -282,6 +282,36 @@ func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
282	require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")	282		require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")
283	}	283		}
284		284
		285	+	// TestResumeWatchOverflowKeepsHeldPrefix overflows one phase and checks the
		286	+	// next phase still replays the gap-free prefix the first phase handed back.
		287	+	// TestResumeWatchHeldOverflowIsSticky covers dropping keys typed after it.
		288	+	func TestResumeWatchOverflowKeepsHeldPrefix(t *testing.T) {
		289	+	reader := &inputTestReader{chunks: make(chan []byte, 2)}
		290	+	pump := newTerminalInputPump(reader)
		291	+	pump.start()
		292	+	t.Cleanup(func() {
		293	+	pump.stop()
		294	+	close(reader.chunks)
		295	+	})
		296	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
		297	+	clock := newSupervisorTestClock()
		298	+
		299	+	first := lifetime.watchResume(clock, nil)
		300	+	reader.chunks <- []byte("ls")
		301	+	require.Eventually(t, func() bool { return first.heldLen() == 2 }, 5*time.Second, time.Millisecond)
		302	+	first.hold(bytes.Repeat([]byte("x"), resumeHeldInputLimit))
		303	+	require.True(t, lifetime.resumeOverflowed.Load())
		304	+	first.release()
		305	+
		306	+	second := lifetime.watchResume(clock, nil)
		307	+	require.Eventually(t, func() bool { return second.heldLen() == 2 }, 5*time.Second, time.Millisecond)
		308	+	second.release()
		309	+
		310	+	pump.mu.Lock()
		311	+	defer pump.mu.Unlock()
		312	+	require.Equal(t, "ls", string(pump.residual))
		313	+	}
		314	+
285	// stallingSessionStream passes writes through until stall is set; stalled	315		// stallingSessionStream passes writes through until stall is set; stalled
286	// writes block until release closes, then fail with failErr when set.	316		// writes block until release closes, then fail with failErr when set.
287	type stallingSessionStream struct {	317		type stallingSessionStream struct {

M internal/usecase/client/supervisor.go +27 -7


@@ -1515,9 +1515,10 @@ type resumeWatch struct {
1515	1515		clock ports.Clock
1516	1516		logger *slog.Logger
1517	1517
1518		-	// backlog counts deliveries already waiting at claim: session input the
1519		-	// lost attachment kept, held verbatim and never read as a cancel key.
1520		-	backlog int
	1518	+	// backlog marks deliveries already waiting at claim: session input the
	1519	+	// lost attachment or an earlier phase kept, held verbatim and never read
	1520	+	// as a cancel key.
	1521	+	backlog claimBacklog
1521	1522		releaseOnce sync.Once
1522	1523
1523	1524		mu sync.Mutex

@@ -1531,12 +1532,24 @@ func (w *resumeWatch) heldLen() int {
1531	1532		return len(w.held)
1532	1533		}
1533	1534
1534		-	// hold keeps data within the bound.
	1535	+	// hold keeps newly typed data within the bound. After an overflow in this
	1536	+	// outage it drops every new key, so the session never sees a gap.
1535	1537		func (w *resumeWatch) hold(data []byte) {
	1538	+	w.holdInput(data, false)
	1539	+	}
	1540	+
	1541	+	// holdResidual keeps the residual an earlier owner handed back. It was kept
	1542	+	// before any overflow, so it precedes the dropped keys and the sticky
	1543	+	// overflow must not discard it.
	1544	+	func (w *resumeWatch) holdResidual(data []byte) {
	1545	+	w.holdInput(data, true)
	1546	+	}
	1547	+
	1548	+	func (w *resumeWatch) holdInput(data []byte, residual bool) {
1536	1549		w.mu.Lock()
1537	1550		defer w.mu.Unlock()
1538	1551		overflowed := &w.lifetime.resumeOverflowed
1539		-	if overflowed.Load() \|\| len(w.held)+len(data) > resumeHeldInputLimit {
	1552	+	if (!residual && overflowed.Load()) \|\| len(w.held)+len(data) > resumeHeldInputLimit {
1540	1553		if len(data) != 0 {
1541	1554		overflowed.Store(true)
1542	1555		}

@@ -1615,8 +1628,15 @@ func (w *resumeWatch) run() {
1615	1628		return
1616	1629		}
1617	1630		pump.ack(w.id)
1618		-	if w.backlog > 0 {
1619		-	w.backlog--
	1631	+	if w.backlog.residual {
	1632	+	w.backlog.residual = false
	1633	+	w.holdResidual(result.data)
	1634	+	continue
	1635	+	}
	1636	+	if w.backlog.pending {
	1637	+	// Read after the earlier owner stopped: newly typed, so an
	1638	+	// overflow drops it, but it is still never a cancel key.
	1639	+	w.backlog.pending = false
1620	1640		w.hold(result.data)
1621	1641		continue
1622	1642		}

 	clock     ports.Clock
 	logger    *slog.Logger
-	// backlog counts deliveries already waiting at claim: session input the
+	// backlog marks deliveries already waiting at claim: session input the
-	// lost attachment kept, held verbatim and never read as a cancel key.
+	// lost attachment or an earlier phase kept, held verbatim and never read
-	backlog     int
+	// as a cancel key.
+	backlog     claimBacklog
 	releaseOnce sync.Once
 	mu   sync.Mutex
 	return len(w.held)
 }
-// hold keeps data within the bound.
+// hold keeps newly typed data within the bound. After an overflow in this
+// outage it drops every new key, so the session never sees a gap.
 func (w *resumeWatch) hold(data []byte) {
+	w.holdInput(data, false)
+}
+// holdResidual keeps the residual an earlier owner handed back. It was kept
+// before any overflow, so it precedes the dropped keys and the sticky
+// overflow must not discard it.
+func (w *resumeWatch) holdResidual(data []byte) {
+	w.holdInput(data, true)
+}
+func (w *resumeWatch) holdInput(data []byte, residual bool) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	overflowed := &w.lifetime.resumeOverflowed
-	if overflowed.Load() || len(w.held)+len(data) > resumeHeldInputLimit {
+	if (!residual && overflowed.Load()) || len(w.held)+len(data) > resumeHeldInputLimit {
 		if len(data) != 0 {
 			overflowed.Store(true)
 		}
 			return
 		}
 		pump.ack(w.id)
-		if w.backlog > 0 {
+		if w.backlog.residual {
-			w.backlog--
+			w.backlog.residual = false
+			w.holdResidual(result.data)
+			continue
+		}
+		if w.backlog.pending {
+			// Read after the earlier owner stopped: newly typed, so an
+			// overflow drops it, but it is still never a cancel key.
+			w.backlog.pending = false
 			w.hold(result.data)
 			continue
 		}

M internal/usecase/client/terminal_input_pump.go +12 -12


@@ -109,22 +109,22 @@ func (p *terminalInputPump) tryClaim() (uint64, bool) {
109	109		return consumer, ok
110	110		}
111	111
112		-	// claimWithBacklog claims the pump and reports, atomically with the claim, how
113		-	// many deliveries were already waiting: the residual, then the pending read.
114		-	// They were read for an earlier owner, not typed for this one.
115		-	func (p *terminalInputPump) claimWithBacklog() (uint64, int, bool) {
	112	+	// claimBacklog describes deliveries already waiting at claim. They are taken
	113	+	// in this order: the residual first, then the pending read.
	114	+	type claimBacklog struct {
	115	+	residual, pending bool
	116	+	}
	117	+
	118	+	// claimWithBacklog claims the pump and reports, atomically with the claim,
	119	+	// which deliveries were already waiting. They were read for an earlier owner,
	120	+	// not typed for this one.
	121	+	func (p *terminalInputPump) claimWithBacklog() (uint64, claimBacklog, bool) {
116	122		p.mu.Lock()
117	123		if p.consumer != 0 {
118	124		p.mu.Unlock()
119		-	return 0, 0, false
120		-	}
121		-	backlog := 0
122		-	if len(p.residual) != 0 {
123		-	backlog++
124		-	}
125		-	if p.pending != nil {
126		-	backlog++
	125	+	return 0, claimBacklog{}, false
127	126		}
	127	+	backlog := claimBacklog{residual: len(p.residual) != 0, pending: p.pending != nil}
128	128		p.nextID++
129	129		p.consumer = p.nextID
130	130		consumer := p.consumer

 	return consumer, ok
 }
-// claimWithBacklog claims the pump and reports, atomically with the claim, how
+// claimBacklog describes deliveries already waiting at claim. They are taken
-// many deliveries were already waiting: the residual, then the pending read.
+// in this order: the residual first, then the pending read.
-// They were read for an earlier owner, not typed for this one.
+type claimBacklog struct {
-func (p *terminalInputPump) claimWithBacklog() (uint64, int, bool) {
+	residual, pending bool
+}
+// claimWithBacklog claims the pump and reports, atomically with the claim,
+// which deliveries were already waiting. They were read for an earlier owner,
+// not typed for this one.
+func (p *terminalInputPump) claimWithBacklog() (uint64, claimBacklog, bool) {
 	p.mu.Lock()
 	if p.consumer != 0 {
 		p.mu.Unlock()
-		return 0, 0, false
+		return 0, claimBacklog{}, false
-	}
-	backlog := 0
-	if len(p.residual) != 0 {
-		backlog++
-	}
-	if p.pending != nil {
-		backlog++
 	}
+	backlog := claimBacklog{residual: len(p.residual) != 0, pending: p.pending != nil}
 	p.nextID++
 	p.consumer = p.nextID
 	consumer := p.consumer