fd82dd0 fix(client): keep the held-input overflow sticky for a whole outage · vev

M internal/usecase/client/attachment_outbox_test.go +11 -4


@@ -126,7 +126,7 @@ func TestResumeWatch(t *testing.T) {
126	126		})
127	127		lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
128	128		clock := newSupervisorTestClock()
129		-	watch := lifetime.watchResume(clock)
	129	+	watch := lifetime.watchResume(clock, nil)
130	130		for _, read := range tt.reads {
131	131		reader.chunks <- []byte(read)
132	132		}

@@ -187,7 +187,7 @@ func TestResumeWatchKeepsBacklogVerbatim(t *testing.T) {
187	187		}, 5*time.Second, time.Millisecond)
188	188		}
189	189		lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
190		-	watch := lifetime.watchResume(newSupervisorTestClock())
	190	+	watch := lifetime.watchResume(newSupervisorTestClock(), nil)
191	191		require.Eventually(t, func() bool { return watch.heldLen() == len(tt.residual)+len(tt.pending) }, 5*time.Second, time.Millisecond)
192	192		if tt.typed != "" {
193	193		reader.chunks <- []byte(tt.typed)

@@ -234,7 +234,7 @@ func TestResumeWatchResolvesEscapeAtRelease(t *testing.T) {
234	234		})
235	235		lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
236	236		clock := newSupervisorTestClock()
237		-	watch := lifetime.watchResume(clock)
	237	+	watch := lifetime.watchResume(clock, nil)
238	238		reader.chunks <- []byte("\x1b")
239	239		escape := clock.awaitTimer(t)
240	240		released := make(chan struct{})

@@ -268,11 +268,18 @@ func TestResumeWatchResolvesEscapeAtRelease(t *testing.T) {
268	268		}
269	269
270	270		func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
271		-	w := &resumeWatch{}
	271	+	lifetime := &terminalInputLifetime{}
	272	+	w := &resumeWatch{lifetime: lifetime}
272	273		w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
273	274		w.hold([]byte("bb"))
274	275		w.hold([]byte("c"))
275	276		require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
	277	+
	278	+	// The next resume phase of the same outage starts a new watch; it must
	279	+	// keep dropping so the session never receives input with a gap.
	280	+	next := &resumeWatch{lifetime: lifetime}
	281	+	next.hold([]byte("d"))
	282	+	require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")
276	283		}
277	284
278	285		// stallingSessionStream passes writes through until stall is set; stalled

 			})
 			lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
 			clock := newSupervisorTestClock()
-			watch := lifetime.watchResume(clock)
+			watch := lifetime.watchResume(clock, nil)
 			for _, read := range tt.reads {
 				reader.chunks <- []byte(read)
 			}
 				}, 5*time.Second, time.Millisecond)
 			}
 			lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
-			watch := lifetime.watchResume(newSupervisorTestClock())
+			watch := lifetime.watchResume(newSupervisorTestClock(), nil)
 			require.Eventually(t, func() bool { return watch.heldLen() == len(tt.residual)+len(tt.pending) }, 5*time.Second, time.Millisecond)
 			if tt.typed != "" {
 				reader.chunks <- []byte(tt.typed)
 			})
 			lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
 			clock := newSupervisorTestClock()
-			watch := lifetime.watchResume(clock)
+			watch := lifetime.watchResume(clock, nil)
 			reader.chunks <- []byte("\x1b")
 			escape := clock.awaitTimer(t)
 			released := make(chan struct{})
 }
 func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
-	w := &resumeWatch{}
+	lifetime := &terminalInputLifetime{}
+	w := &resumeWatch{lifetime: lifetime}
 	w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
 	w.hold([]byte("bb"))
 	w.hold([]byte("c"))
 	require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
+	// The next resume phase of the same outage starts a new watch; it must
+	// keep dropping so the session never receives input with a gap.
+	next := &resumeWatch{lifetime: lifetime}
+	next.hold([]byte("d"))
+	require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")
 }
 // stallingSessionStream passes writes through until stall is set; stalled

M internal/usecase/client/supervisor.go +22 -11


@@ -1387,6 +1387,10 @@ type terminalInputLifetime struct {
1387	1387
1388	1388		mu sync.Mutex
1389	1389		pickerID uint64
	1390	+	// resumeOverflowed spans every watch of one outage: once held keys
	1391	+	// overflowed, later phases drop keys too, so the resumed session never
	1392	+	// receives input with a gap. The picker taking input clears it.
	1393	+	resumeOverflowed atomic.Bool
1390	1394		}
1391	1395
1392	1396		// startTerminalInputLifetime starts the single terminal read. A nil reader is

@@ -1436,6 +1440,7 @@ func (l *terminalInputLifetime) acquirePicker() {
1436	1440		}
1437	1441		if id, ok := l.pump.tryClaim(); ok {
1438	1442		l.pickerID = id
	1443	+	l.resumeOverflowed.Store(false)
1439	1444		l.mu.Unlock()
1440	1445		return
1441	1446		}

@@ -1514,9 +1519,8 @@ func (l *terminalInputLifetime) stop() {
1514	1519		}
1515	1520
1516	1521		// resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
1517		-	// Past it further keys are dropped, like a full terminal buffer.
1518		-	// TODO: the overflow flag is per resume watch; a later phase can accept keys
1519		-	// again and leave a gap. Make it sticky on the input lifetime until attach.
	1522	+	// Past it further keys are dropped, like a full terminal buffer, until the
	1523	+	// picker takes input again.
1520	1524		const resumeHeldInputLimit = 64 << 10
1521	1525
1522	1526		// resumeWatch reads the terminal while a lost attachment waits to resume.

@@ -1529,6 +1533,7 @@ type resumeWatch struct {
1529	1533		lifetime *terminalInputLifetime
1530	1534		id uint64
1531	1535		clock ports.Clock
	1536	+	logger *slog.Logger
1532	1537
1533	1538		// backlog counts deliveries already waiting at claim: session input the
1534	1539		// lost attachment kept, held verbatim and never read as a cancel key.

@@ -1537,9 +1542,6 @@ type resumeWatch struct {
1537	1542
1538	1543		mu sync.Mutex
1539	1544		held []byte
1540		-	// overflowed drops every key after the first one past the bound, so the
1541		-	// resumed session never receives input with a gap in the middle.
1542		-	overflowed bool
1543	1545		}
1544	1546
1545	1547		// heldLen reports how many kept bytes the watch holds.

@@ -1553,8 +1555,11 @@ func (w *resumeWatch) heldLen() int {
1553	1555		func (w *resumeWatch) hold(data []byte) {
1554	1556		w.mu.Lock()
1555	1557		defer w.mu.Unlock()
1556		-	if w.overflowed \|\| len(w.held)+len(data) > resumeHeldInputLimit {
1557		-	w.overflowed = w.overflowed \|\| len(data) != 0
	1558	+	overflowed := &w.lifetime.resumeOverflowed
	1559	+	if overflowed.Load() \|\| len(w.held)+len(data) > resumeHeldInputLimit {
	1560	+	if len(data) != 0 {
	1561	+	overflowed.Store(true)
	1562	+	}
1558	1563		return
1559	1564		}
1560	1565		w.held = append(w.held, data...)

@@ -1562,8 +1567,8 @@ func (w *resumeWatch) hold(data []byte) {
1562	1567
1563	1568		// watchResume claims terminal input for one resume backoff. When the claim
1564	1569		// is unavailable it returns a watch that never cancels.
1565		-	func (l terminalInputLifetime) watchResume(clock ports.Clock) resumeWatch {
1566		-	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock}
	1570	+	func (l terminalInputLifetime) watchResume(clock ports.Clock, logger slog.Logger) *resumeWatch {
	1571	+	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock, logger: logger}
1567	1572		if l == nil \|\| l.pump == nil {
1568	1573		close(w.done)
1569	1574		return w

@@ -1672,7 +1677,13 @@ func (w *resumeWatch) releaseClaim() {
1672	1677		case <-w.cancelled:
1673	1678		default:
1674	1679		w.mu.Lock()
1675		-	pump.appendResidual(w.id, w.held, resumeHeldInputLimit)
	1680	+	if !pump.appendResidual(w.id, w.held, resumeHeldInputLimit) && len(w.held) != 0 {
	1681	+	// Replaying only part of the held keys would leave a gap.
	1682	+	w.lifetime.resumeOverflowed.Store(true)
	1683	+	if w.logger != nil {
	1684	+	w.logger.Warn("client_resume_input_dropped", "bytes", len(w.held))
	1685	+	}
	1686	+	}
1676	1687		w.mu.Unlock()
1677	1688		}
1678	1689		pump.revoke(w.id)

 	mu       sync.Mutex
 	pickerID uint64
+	// resumeOverflowed spans every watch of one outage: once held keys
+	// overflowed, later phases drop keys too, so the resumed session never
+	// receives input with a gap. The picker taking input clears it.
+	resumeOverflowed atomic.Bool
 }
 // startTerminalInputLifetime starts the single terminal read. A nil reader is
 	}
 	if id, ok := l.pump.tryClaim(); ok {
 		l.pickerID = id
+		l.resumeOverflowed.Store(false)
 		l.mu.Unlock()
 		return
 	}
 }
 // resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
-// Past it further keys are dropped, like a full terminal buffer.
+// Past it further keys are dropped, like a full terminal buffer, until the
-// TODO: the overflow flag is per resume watch; a later phase can accept keys
+// picker takes input again.
-// again and leave a gap. Make it sticky on the input lifetime until attach.
 const resumeHeldInputLimit = 64 << 10
 // resumeWatch reads the terminal while a lost attachment waits to resume.
 	lifetime  *terminalInputLifetime
 	id        uint64
 	clock     ports.Clock
+	logger    *slog.Logger
 	// backlog counts deliveries already waiting at claim: session input the
 	// lost attachment kept, held verbatim and never read as a cancel key.
 	mu   sync.Mutex
 	held []byte
-	// overflowed drops every key after the first one past the bound, so the
-	// resumed session never receives input with a gap in the middle.
-	overflowed bool
 }
 // heldLen reports how many kept bytes the watch holds.
 func (w *resumeWatch) hold(data []byte) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	if w.overflowed || len(w.held)+len(data) > resumeHeldInputLimit {
+	overflowed := &w.lifetime.resumeOverflowed
-		w.overflowed = w.overflowed || len(data) != 0
+	if overflowed.Load() || len(w.held)+len(data) > resumeHeldInputLimit {
+		if len(data) != 0 {
+			overflowed.Store(true)
+		}
 		return
 	}
 	w.held = append(w.held, data...)
 // watchResume claims terminal input for one resume backoff. When the claim
 // is unavailable it returns a watch that never cancels.
-func (l *terminalInputLifetime) watchResume(clock ports.Clock) *resumeWatch {
+func (l *terminalInputLifetime) watchResume(clock ports.Clock, logger *slog.Logger) *resumeWatch {
-	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock}
+	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock, logger: logger}
 	if l == nil || l.pump == nil {
 		close(w.done)
 		return w
 	case <-w.cancelled:
 	default:
 		w.mu.Lock()
-		pump.appendResidual(w.id, w.held, resumeHeldInputLimit)
+		if !pump.appendResidual(w.id, w.held, resumeHeldInputLimit) && len(w.held) != 0 {
+			// Replaying only part of the held keys would leave a gap.
+			w.lifetime.resumeOverflowed.Store(true)
+			if w.logger != nil {
+				w.logger.Warn("client_resume_input_dropped", "bytes", len(w.held))
+			}
+		}
 		w.mu.Unlock()
 	}
 	pump.revoke(w.id)

M internal/usecase/client/supervisor_attach.go +6 -2


@@ -744,7 +744,7 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
744	744		var cancelOpen context.CancelFunc
745	745		var watchDone chan struct{}
746	746		if resuming {
747		-	keys = input.watchResume(s.cfg.Clock)
	747	+	keys = input.watchResume(s.cfg.Clock, s.logger)
748	748		openCtx, cancelOpen = context.WithCancel(openCtx)
749	749		watchDone = make(chan struct{})
750	750		go func() {

@@ -779,6 +779,10 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
779	779		if supervisorNil(stream) {
780	780		return s.attemptFailed(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"}, resuming)
781	781		}
	782	+	if input != nil {
	783	+	// The outage ended: keys typed from now on reach the session again.
	784	+	input.resumeOverflowed.Store(false)
	785	+	}
782	786		if err := deadline.adopt(stream); err != nil {
783	787		// The accepted absolute deadline already elapsed before this client
784	788		// saw the stream. Fail the attachment instead of restarting a budget.

@@ -1043,7 +1047,7 @@ func (s Supervisor) waitResume(ctx context.Context, input terminalInputLifetim
1043	1047		delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
1044	1048		timer := s.cfg.Clock.NewTimer(delay)
1045	1049		defer stopSupervisorTimer(timer)
1046		-	keys := input.watchResume(s.cfg.Clock)
	1050	+	keys := input.watchResume(s.cfg.Clock, s.logger)
1047	1051		defer keys.release()
1048	1052		for {
1049	1053		select {

 	var cancelOpen context.CancelFunc
 	var watchDone chan struct{}
 	if resuming {
-		keys = input.watchResume(s.cfg.Clock)
+		keys = input.watchResume(s.cfg.Clock, s.logger)
 		openCtx, cancelOpen = context.WithCancel(openCtx)
 		watchDone = make(chan struct{})
 		go func() {
 	if supervisorNil(stream) {
 		return s.attemptFailed(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"}, resuming)
 	}
+	if input != nil {
+		// The outage ended: keys typed from now on reach the session again.
+		input.resumeOverflowed.Store(false)
+	}
 	if err := deadline.adopt(stream); err != nil {
 		// The accepted absolute deadline already elapsed before this client
 		// saw the stream. Fail the attachment instead of restarting a budget.
 	delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
 	timer := s.cfg.Clock.NewTimer(delay)
 	defer stopSupervisorTimer(timer)
-	keys := input.watchResume(s.cfg.Clock)
+	keys := input.watchResume(s.cfg.Clock, s.logger)
 	defer keys.release()
 	for {
 		select {