//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) {
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
s.write = func(ctx context.Context, _ []obs.Render, _ []obs.Subprocess) error {
started.Add(1)
select {
case <-release:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
return s, release, &started
}
// 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, _ := stallingSink(t, 4)
defer func() { close(release); s.Close() }()
for i := 0; i < 1000; i++ {
s.Render(obs.Render{At: time.Now(), Path: "GET /x"})
}
s.mu.Lock()
held := len(s.renders)
s.mu.Unlock()
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)
}
}
// 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)
}
}