//go:build !js && !wasip1
package obsduck
import (
"context"
"path/filepath"
"sync/atomic"
"testing"
"time"
"craftdesign.group/quarry/internal/obs"
)
// stallingSink opens a sink whose writes block until released or until
// their context ends, standing in for a DuckDB write that is slow.
func stallingSink(t *testing.T, batch int) (*Sink, chan struct{}, *atomic.Int32) {
s, release, started, _ := stallingSinkCounting(t, batch)
return s, release, started
}
// stallingSinkCounting is stallingSink that also counts the rows handed
// to the writer.
func stallingSinkCounting(t *testing.T, batch int) (*Sink, chan struct{}, *atomic.Int32, *atomic.Int64) {
t.Helper()
s, err := Open(filepath.Join(t.TempDir(), "ops.duckdb"), batch)
if err != nil {
t.Skipf("duckdb unavailable: %v", err)
}
release := make(chan struct{})
var started atomic.Int32
var handed atomic.Int64
s.write = func(ctx context.Context, rs []obs.Render, ss []obs.Subprocess) error {
started.Add(1)
handed.Add(int64(len(rs) + len(ss)))
select {
case <-release:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
return s, release, &started, &handed
}
// The row that fills a batch was written by the request that filed it,
// so every batch-th request waited on a DuckDB write before its handler
// returned — more often once API calls were recorded too.
func TestTheRowThatFillsABatchDoesNotWaitForTheWrite(t *testing.T) {
s, release, started := stallingSink(t, 4)
defer func() { close(release); s.Close() }()
done := make(chan struct{})
go func() {
for i := 0; i < 4; i++ {
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
}
close(done)
}()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("filing the batch-filling row blocked on the write")
}
deadline := time.Now().Add(2 * time.Second)
for started.Load() == 0 && time.Now().Before(deadline) {
time.Sleep(5 * time.Millisecond)
}
if started.Load() == 0 {
t.Fatal("a full batch never reached the writer")
}
}
// While the writer is stalled the buffer stays bounded: rows past the
// bound are dropped and counted, not held without limit.
func TestAFullBufferDropsAndCounts(t *testing.T) {
s, release, started, handed := stallingSinkCounting(t, 4)
defer func() { close(release); s.Close() }()
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
for i := 0; i < 3; i++ {
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
}
// Let the first batch reach the stalled writer, so the count below
// covers a row in flight as well as held and dropped ones.
for deadline := time.Now().Add(2 * time.Second); started.Load() == 0 && time.Now().Before(deadline); {
time.Sleep(5 * time.Millisecond)
}
for i := 4; i < 1000; i++ {
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
}
s.mu.Lock()
held := len(s.renders)
s.mu.Unlock()
if got := int64(held) + handed.Load() + int64(s.Dropped()); got != 1000 {
t.Errorf("held %d + in flight %d + dropped %d = %d, want every one of 1000 rows accounted for",
held, handed.Load(), s.Dropped(), got)
}
if held > s.max {
t.Errorf("buffer holds %d rows, bound is %d", held, s.max)
}
if s.Dropped() == 0 {
t.Error("rows past the bound were not counted as dropped")
}
r := obs.NewRecorder(8, s)
if r.Health().Dropped != s.Dropped() {
t.Errorf("health reports %d dropped, sink %d", r.Health().Dropped, s.Dropped())
}
}
// Close lands what is buffered but gives up at its deadline, so a stuck
// write cannot hold shutdown past its budget.
func TestCloseStaysInsideItsDeadline(t *testing.T) {
s, release, _ := stallingSink(t, 4)
defer close(release)
s.closeBudget = 200 * time.Millisecond
for i := 0; i < 6; i++ {
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
}
start := time.Now()
s.Close()
if el := time.Since(start); el > 2*time.Second {
t.Errorf("Close took %s with a stuck write, budget 200ms", el)
}
// Every row filed was lost to the cancelled writes, and says so.
if d := s.Dropped(); d != 6 {
t.Errorf("dropped %d after Close cancelled the writes, want all 6", d)
}
}
// A write that fails loses its rows; they are counted as dropped, not
// only remembered as the last error.
func TestAFailedWriteCountsItsRowsAsDropped(t *testing.T) {
s, err := Open(filepath.Join(t.TempDir(), "ops.duckdb"), 256)
if err != nil {
t.Skipf("duckdb unavailable: %v", err)
}
defer s.Close()
s.write = func(context.Context, []obs.Render, []obs.Subprocess) error {
return context.DeadlineExceeded
}
for i := 0; i < 3; i++ {
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
}
s.Subprocess(obs.Subprocess{At: time.Now(), Verb: "ls"})
s.Flush()
if d := s.Dropped(); d != 4 {
t.Errorf("dropped %d after a failed write of 4 rows, want 4", d)
}
if s.Err() == nil {
t.Error("the failed write left no error for the health panel")
}
}
// With a writer that works, Close lands what was buffered.
func TestCloseLandsTheBuffer(t *testing.T) {
path := filepath.Join(t.TempDir(), "ops.duckdb")
s, err := Open(path, 256)
if err != nil {
t.Skipf("duckdb unavailable: %v", err)
}
for i := 0; i < 3; i++ {
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
}
if err := s.Close(); err != nil {
t.Fatal(err)
}
s, err = Open(path, 256)
if err != nil {
t.Fatal(err)
}
defer s.Close()
if n := rows(t, s, "render"); n != 3 {
t.Errorf("%d rows after Close, want 3", n)
}
}