grace

Author: seruman

internal/daemon/session_clients.go

@@ -6,12 +6,18 @@ import (
 	"log/slog"
 	"math"
 	"syscall"
+	"time"
 
 	"code.selman.me/hauntty/internal/config"
 	"code.selman.me/hauntty/internal/protocol"
 	"github.com/creack/pty"
 )
 
+const (
+	sessionClientOutBufferSize = 256
+	slowClientGracePeriod      = 100 * time.Millisecond
+)
+
 func (c *sessionClient) writeLoop() {
 	for msg := range c.outCh {
 		if err := c.conn.WriteMessage(msg); err != nil {
@@ -170,8 +176,20 @@ func broadcastOutput(clients []*sessionClient, name string, msg *protocol.Output
 		case c.outCh <- msg:
 			clients[i] = c
 			i++
+			continue
 		default:
-			slog.Debug("evicting slow client", "session", name)
+		}
+
+		timer := time.NewTimer(slowClientGracePeriod)
+		select {
+		case c.outCh <- msg:
+			if !timer.Stop() {
+				<-timer.C
+			}
+			clients[i] = c
+			i++
+		case <-timer.C:
+			slog.Debug("evicting slow client", "session", name, "grace", slowClientGracePeriod)
 			close(c.outCh)
 			_ = c.closeConn()
 		}

internal/daemon/session_lifecycle.go

@@ -257,7 +257,7 @@ func (s *Session) run() {
 					size:      a.spec.size,
 					version:   a.spec.version,
 					readOnly:  a.spec.readOnly,
-					outCh:     make(chan protocol.Message, 64),
+					outCh:     make(chan protocol.Message, sessionClientOutBufferSize),
 				}
 				go sc.writeLoop()
 

internal/daemon/session_loop_test.go

@@ -256,6 +256,39 @@ func TestBroadcastOutputEvictsSlowClients(t *testing.T) {
 	assert.DeepEqual(t, got, msg)
 }
 
+func TestBroadcastOutputAllowsTransientBackpressure(t *testing.T) {
+	msg := &protocol.Output{Data: []byte("hello")}
+
+	var closeCount atomic.Int32
+	slow := &sessionClient{
+		id: "slow",
+		closeConn: func() error {
+			closeCount.Add(1)
+			return nil
+		},
+		outCh: make(chan protocol.Message, 1),
+	}
+	busy := &protocol.Output{Data: []byte("busy")}
+	slow.outCh <- busy
+
+	drainDone := make(chan struct{})
+	go func() {
+		time.Sleep(slowClientGracePeriod / 4)
+		<-slow.outCh
+		close(drainDone)
+	}()
+
+	clients := broadcastOutput([]*sessionClient{slow}, "demo", msg)
+	<-drainDone
+
+	assert.Equal(t, len(clients), 1)
+	assert.Assert(t, clients[0] == slow)
+	assert.Equal(t, closeCount.Load(), int32(0))
+
+	got := <-slow.outCh
+	assert.DeepEqual(t, got, msg)
+}
+
 func TestNotifyClientsChangedSkipsBlockedClients(t *testing.T) {
 	ready := &sessionClient{outCh: make(chan protocol.Message, 1)}
 	blocked := &sessionClient{outCh: make(chan protocol.Message, 1)}