panel/agent/internal/dispatch/sendbuffer_test.go

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.
}