panel — open-source game server manager (public release)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

agent/internal/dispatch/sendbuffer_test.go

@@ -0,0 +1,131 @@
+package dispatch
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	panelv1 "github.com/dbledeez/panel/proto/panel/v1"
+)
+
+// blockingSender simulates a controller that has stopped draining the stream:
+// every Send blocks until released. This is the backpressure condition that
+// used to wedge the whole agent when sendEnv called Send inline.
+type blockingSender struct {
+	release chan struct{}
+	mu      sync.Mutex
+	sent    int
+}
+
+func (b *blockingSender) Send(*panelv1.AgentEnvelope) error {
+	<-b.release // block until the test lets it through
+	b.mu.Lock()
+	b.sent++
+	b.mu.Unlock()
+	return nil
+}
+
+func logEnv() *panelv1.AgentEnvelope {
+	return &panelv1.AgentEnvelope{Payload: &panelv1.AgentEnvelope_Log{Log: &panelv1.LogLine{Line: "x"}}}
+}
+
+func stateEnv() *panelv1.AgentEnvelope {
+	return &panelv1.AgentEnvelope{Payload: &panelv1.AgentEnvelope_InstanceState{InstanceState: &panelv1.InstanceStateUpdate{InstanceId: "i"}}}
+}
+
+// newTestDispatcher builds a Dispatcher with just the send machinery wired —
+// enough to exercise sendEnv / sendLoop without a real runtime.
+func newTestDispatcher() *Dispatcher {
+	d := &Dispatcher{sendCh: make(chan *panelv1.AgentEnvelope, 4096)}
+	go d.sendLoop()
+	return d
+}
+
+// TestSendEnvDropsTelemetryUnderBackpressure is the core regression guard for
+// the VEIN-exposed wedge: with the sender blocked, a flood of log envelopes far
+// exceeding the buffer must NOT block the producer. Pre-fix, sendEnv called
+// Send inline and the producer (and the whole agent) froze here.
+func TestSendEnvDropsTelemetryUnderBackpressure(t *testing.T) {
+	d := newTestDispatcher()
+	bs := &blockingSender{release: make(chan struct{})}
+	d.SetSender(bs)
+
+	done := make(chan struct{})
+	go func() {
+		// 50k log frames — 12x the 4096 buffer. Must all return promptly.
+		for i := 0; i < 50000; i++ {
+			d.sendEnv(logEnv())
+		}
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// Producer completed without blocking on the stalled sender. Pass.
+	case <-time.After(5 * time.Second):
+		t.Fatal("sendEnv blocked under backpressure — the wedge regressed")
+	}
+}
+
+// TestCriticalEnvelopeNotDropped verifies an instance-state envelope is queued
+// (not silently dropped) even when telemetry is being shed. We fill the buffer
+// with the sender blocked, then confirm a critical send blocks until drained
+// (i.e. it's preserved, not discarded).
+func TestCriticalEnvelopeNotDropped(t *testing.T) {
+	d := &Dispatcher{sendCh: make(chan *panelv1.AgentEnvelope, 2)}
+	// No drainer started yet — fill the buffer.
+	d.sendEnv(logEnv())
+	d.sendEnv(logEnv())
+	// Buffer is full (cap 2). A telemetry frame must drop (non-blocking).
+	dropDone := make(chan struct{})
+	go func() { d.sendEnv(logEnv()); close(dropDone) }()
+	select {
+	case <-dropDone:
+	case <-time.After(time.Second):
+		t.Fatal("telemetry send blocked on full buffer — should have dropped")
+	}
+	// A critical frame must NOT drop: it blocks until space frees. Start a
+	// drainer to free space and confirm the critical send then completes.
+	critDone := make(chan struct{})
+	go func() { d.sendEnv(stateEnv()); close(critDone) }()
+	select {
+	case <-critDone:
+		t.Fatal("critical send returned before buffer had space — it was dropped")
+	case <-time.After(100 * time.Millisecond):
+		// Correctly still blocking. Now drain and it should complete.
+	}
+	go d.sendLoop()
+	select {
+	case <-critDone:
+		// Drained and delivered. Pass.
+	case <-time.After(time.Second):
+		t.Fatal("critical send never completed after drain")
+	}
+}
+
+// TestConcurrentProducersNoRace runs many producers through sendEnv at once.
+// Pre-fix, each called stream.Send directly — concurrent Sends on one gRPC
+// stream are a data race. The single-drainer design serializes them. Run with
+// -race to catch a regression.
+func TestConcurrentProducersNoRace(t *testing.T) {
+	d := newTestDispatcher()
+	bs := &blockingSender{release: make(chan struct{})}
+	close(bs.release) // never block — just count
+	d.SetSender(bs)
+
+	var wg sync.WaitGroup
+	for p := 0; p < 32; p++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < 1000; i++ {
+				d.sendEnv(logEnv())
+			}
+		}()
+	}
+	wg.Wait()
+	// Give the drainer a moment to flush.
+	time.Sleep(200 * time.Millisecond)
+	// No assertion on exact count (some may still be in flight / dropped);
+	// the point is the -race detector finds no concurrent Send.
+}