//go:build !js && !wasip1
// Package obsduck stores observations for analysis.
//
// It is separate from internal/obs, and behind a build constraint, for
// one reason: DuckDB requires cgo and refuses to build for wasm. Both
// were measured rather than assumed — CGO_ENABLED=0 fails on
// go-duckdb's bindings, and GOOS=js reports "build constraints exclude
// all Go files". quarry is otherwise CGO-free on purpose (its SQLite
// driver is a wasm build for exactly this reason), so that the forge
// can theoretically compile to wasm for an embedded UI. That build
// ships no analytics, wires no sink, and must never reach this package.
//
// Nothing on the request path may import this. obs.Sink is the seam;
// only the composition root knows which implementation exists.
//
// Columnar because the workload is columnar: append-only, time-ordered,
// aggregate-queried, never point-looked-up by a page. RFC 0001 defers
// the lake to "when hydration and analytics — the things columnar is
// for — exist", and reserves the operational surface for SQLite. This
// is that analytics workload, not a second operational store.
package obsduck
import (
"context"
"database/sql"
"fmt"
"strings"
"sync"
"sync/atomic"
"time"
_ "github.com/marcboeker/go-duckdb/v2"
"craftdesign.group/quarry/internal/obs"
)
// Sink writes observations to a DuckDB file.
//
// Buffered, never a write per request: an analytics store on the hot
// path would be the forge paying at serve time for a question nobody is
// asking yet. Rows accumulate in memory and land in batches, which is
// also the shape a Parquet COPY TO wants when the lake arrives.
type Sink struct {
db *sql.DB
mu sync.Mutex
renders []obs.Render
subs []obs.Subprocess
batch int
closed bool
flushErr error
// max bounds each buffer. A write slower than the traffic filing
// rows would otherwise grow them without limit; past it a row is
// dropped and counted, because an observation is not worth the
// forge's memory.
max int
dropped atomic.Uint64
// kick asks the flusher for a batch. A full batch used to be
// written by the request that filled it, so every batch-th request
// paid for a DuckDB write before its handler returned.
kick chan struct{}
stop chan struct{}
done chan struct{}
// ctx bounds every write; Close cancels it once closeBudget has
// passed, so a stuck write cannot hold shutdown past its budget.
ctx context.Context
cancel context.CancelFunc
closeBudget time.Duration
write func(context.Context, []obs.Render, []obs.Subprocess) error
}
// Open creates or opens the analytics store at path.
func Open(path string, batch int) (*Sink, error) {
if batch <= 0 {
batch = 256
}
db, err := sql.Open("duckdb", path)
if err != nil {
return nil, fmt.Errorf("obsduck: open %s: %w", path, err)
}
if err := db.Ping(); err != nil {
db.Close()
return nil, fmt.Errorf("obsduck: open %s: %w", path, err)
}
// Wide, flat and denormalised, which is what columnar scans want and
// what a Parquet export wants. No indexes and no keys: nothing here
// is looked up by identity, and a row is never updated.
for _, ddl := range []string{
// ts, not "at": AT is reserved in DuckDB (AT TIME ZONE, time
// travel), and a quoted identifier would need quoting at every
// call site and in every ad-hoc query someone writes later.
`CREATE TABLE IF NOT EXISTS render (
ts TIMESTAMP, path VARCHAR, repo VARCHAR,
elapsed_ms BIGINT, ops_ms BIGINT, zit_ms BIGINT, status INTEGER)`,
// url is the path actually requested, added after the fact: the
// route groups rows and this identifies one. Additive so an
// existing store keeps its history with the column empty,
// rather than a schema bump that would throw the history away
// to gain a column describing it.
`ALTER TABLE render ADD COLUMN IF NOT EXISTS url VARCHAR`,
// surface is the door a request came in by, web or api. Added
// the same additive way; every row from before it was a page,
// so history reads an empty one as web.
`ALTER TABLE render ADD COLUMN IF NOT EXISTS surface VARCHAR`,
`CREATE TABLE IF NOT EXISTS subprocess (
ts TIMESTAMP, verb VARCHAR, repo VARCHAR,
duration_ms BIGINT, failed BOOLEAN)`,
} {
if _, err := db.Exec(ddl); err != nil {
db.Close()
return nil, fmt.Errorf("obsduck: schema: %w", err)
}
}
sk := &Sink{db: db, batch: batch, max: 8 * batch,
kick: make(chan struct{}, 1), stop: make(chan struct{}), done: make(chan struct{}),
closeBudget: 2 * time.Second}
sk.ctx, sk.cancel = context.WithCancel(context.Background())
sk.write = sk.writeDB
go sk.flusher()
return sk, nil
}
// Render buffers one page. It never writes: a full batch is handed to
// the flusher, so the request that filed the row does not wait on it.
func (s *Sink) Render(v obs.Render) {
s.mu.Lock()
if s.closed || len(s.renders) >= s.max {
s.mu.Unlock()
s.dropped.Add(1)
return
}
s.renders = append(s.renders, v)
full := len(s.renders) >= s.batch
s.mu.Unlock()
if full {
s.nudge()
}
}
// Subprocess buffers one store call, the same way.
func (s *Sink) Subprocess(v obs.Subprocess) {
s.mu.Lock()
if s.closed || len(s.subs) >= s.max {
s.mu.Unlock()
s.dropped.Add(1)
return
}
s.subs = append(s.subs, v)
full := len(s.subs) >= s.batch
s.mu.Unlock()
if full {
s.nudge()
}
}
// nudge wakes the flusher without waiting. A wake-up already pending
// covers this one: the flusher takes everything buffered when it runs.
func (s *Sink) nudge() {
select {
case s.kick <- struct{}{}:
default:
}
}
// flusher is the one goroutine that writes full batches.
func (s *Sink) flusher() {
defer close(s.done)
for {
select {
case <-s.kick:
s.Flush()
case <-s.stop:
return
}
}
}
// Dropped is how many observations were refused because the buffer was
// full or the sink closed.
func (s *Sink) Dropped() uint64 { return s.dropped.Load() }
// Flush writes what is buffered. Safe to call at any time and from any
// goroutine; a failure is REMEMBERED rather than returned, because the
// callers are observation funnels that must not care whether analytics
// is healthy.
func (s *Sink) Flush() {
s.mu.Lock()
renders, subs := s.renders, s.subs
s.renders, s.subs = nil, nil
s.mu.Unlock()
if len(renders) == 0 && len(subs) == 0 {
return
}
if err := s.write(s.ctx, renders, subs); err != nil {
s.mu.Lock()
s.flushErr = err
s.mu.Unlock()
}
}
func (s *Sink) writeDB(ctx context.Context, renders []obs.Render, subs []obs.Subprocess) error {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
if len(renders) > 0 {
st, err := tx.Prepare(`INSERT INTO render
(ts, path, repo, elapsed_ms, ops_ms, zit_ms, status, url, surface)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`)
if err != nil {
return err
}
for _, v := range renders {
if _, err := st.ExecContext(ctx, v.At, v.Path, v.Repo,
v.Elapsed.Milliseconds(), v.Ops.Milliseconds(), v.Zit.Milliseconds(),
v.Status, v.URL, v.Surface); err != nil {
st.Close()
return err
}
}
st.Close()
}
if len(subs) > 0 {
st, err := tx.Prepare(`INSERT INTO subprocess VALUES (?, ?, ?, ?, ?)`)
if err != nil {
return err
}
for _, v := range subs {
if _, err := st.ExecContext(ctx, v.At, v.Verb, v.Repo,
v.Duration.Milliseconds(), v.Err); err != nil {
st.Close()
return err
}
}
st.Close()
}
return tx.Commit()
}
// Trend reports a route's daily percentiles over the last days, oldest
// first; an empty path trends the whole forge.
//
// quantile_disc, not quantile_cont: the live panel computes percentiles
// by nearest rank, and a history that interpolated would disagree with
// the snapshot above it about the same window. Two numbers on one page
// that are both "p95" and do not match is the kind of small
// contradiction that costs a reader their trust in the rest.
func (s *Sink) Trend(path string, days int) ([]obs.TrendPoint, error) {
if days <= 0 {
days = 14
}
// Today's rows are the ones an operator is looking for after a
// deploy, and they are the ones most likely to still be buffered.
s.Flush()
where, args := "ts >= now() - INTERVAL (?) DAY", []any{days}
if path != "" {
where, args = "path = ? AND "+where, []any{path, days}
}
rows, err := s.db.Query(`SELECT strftime(ts, '%Y-%m-%d') AS day, count(*),
CAST(quantile_disc(elapsed_ms, 0.5) AS BIGINT),
CAST(quantile_disc(elapsed_ms, 0.95) AS BIGINT),
CAST(max(elapsed_ms) AS BIGINT)
FROM render WHERE `+where+`
GROUP BY day ORDER BY day`, args...)
if err != nil {
return nil, fmt.Errorf("obsduck: trend: %w", err)
}
defer rows.Close()
var out []obs.TrendPoint
for rows.Next() {
var p obs.TrendPoint
if err := rows.Scan(&p.Day, &p.N, &p.P50Ms, &p.P95Ms, &p.MaxMs); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
// History reads renders back, newest or slowest first.
//
// Bounded three ways — a window, a limit, and an optional filter —
// because an unbounded read of an append-only store is the query that
// works fine until the day it matters.
func (s *Sink) History(q obs.HistoryQuery) ([]obs.HistoryRow, error) {
if q.Days <= 0 {
q.Days = 14
}
if q.Limit <= 0 || q.Limit > 500 {
q.Limit = 100
}
s.Flush()
where := []string{"ts >= now() - INTERVAL (?) DAY"}
args := []any{q.Days}
if q.Find != "" {
// The route AND the requested path: a reader who types a
// repository name means the URL, and one who types a route
// means the pattern. Answering only one of those makes the box
// look broken half the time.
where = append(where, "(lower(path) LIKE ? OR lower(coalesce(url, '')) LIKE ?)")
like := "%" + strings.ToLower(q.Find) + "%"
args = append(args, like, like)
}
if q.Repo != "" {
where = append(where, "repo = ?")
args = append(args, q.Repo)
}
order := "ts DESC"
if q.Sort == "slowest" {
order = "elapsed_ms DESC, ts DESC"
}
args = append(args, q.Limit)
rows, err := s.db.Query(`SELECT strftime(ts, '%Y-%m-%d %H:%M:%S'), path,
coalesce(url, ''), coalesce(repo, ''), elapsed_ms, ops_ms, zit_ms, status,
coalesce(nullif(surface, ''), 'web')
FROM render WHERE `+strings.Join(where, " AND ")+`
ORDER BY `+order+` LIMIT ?`, args...)
if err != nil {
return nil, fmt.Errorf("obsduck: history: %w", err)
}
defer rows.Close()
var out []obs.HistoryRow
for rows.Next() {
var h obs.HistoryRow
if err := rows.Scan(&h.At, &h.Path, &h.URL, &h.Repo,
&h.ElapsedMs, &h.OpsMs, &h.ZitMs, &h.Status, &h.Surface); err != nil {
return nil, err
}
out = append(out, h)
}
return out, rows.Err()
}
// Prune drops observations older than keep and reports how many rows
// went. A keep of zero or less keeps everything.
//
// Retention is the operator's call, not a default nobody was asked
// about — but "append for ever" is also a default nobody was asked
// about, and it is the one that ends with a forge whose disk fills
// because of the thing that was supposed to be watching it. So there
// is a window, it is stated, and it can be switched off.
func (s *Sink) Prune(keep time.Duration) (int64, error) {
if keep <= 0 {
return 0, nil
}
// Flush first: rows still in the buffer are the newest ones and
// would survive anyway, but pruning around an unflushed buffer
// makes the count reported to an operator a guess.
s.Flush()
cutoff := time.Now().Add(-keep)
var total int64
for _, table := range []string{"render", "subprocess"} {
res, err := s.db.Exec(`DELETE FROM `+table+` WHERE ts < ?`, cutoff)
if err != nil {
return total, fmt.Errorf("obsduck: prune %s: %w", table, err)
}
if n, err := res.RowsAffected(); err == nil {
total += n
}
}
return total, nil
}
// PruneEvery runs Prune on a timer. Daily is the right cadence for a
// window measured in days: pruning more often costs writes to delete
// rows that a later pass would have deleted anyway.
func (s *Sink) PruneEvery(every, keep time.Duration, stop <-chan struct{}) {
if every <= 0 {
every = 24 * time.Hour
}
t := time.NewTicker(every)
defer t.Stop()
for {
select {
case <-t.C:
if _, err := s.Prune(keep); err != nil {
s.mu.Lock()
s.flushErr = err
s.mu.Unlock()
}
case <-stop:
return
}
}
}
// Err reports the last write failure, so an admin surface can say the
// analytics store is unhealthy rather than silently recording nothing.
func (s *Sink) Err() error {
s.mu.Lock()
defer s.mu.Unlock()
return s.flushErr
}
// Close stops the flusher, lands what is buffered, and shuts down —
// within closeBudget. Past it the write in flight is cancelled and its
// rows are lost, which is the right trade: shutdown has a budget, and
// analytics is the last thing that should spend it.
func (s *Sink) Close() error {
s.mu.Lock()
if s.closed {
s.mu.Unlock()
return nil
}
s.closed = true
s.mu.Unlock()
give := time.AfterFunc(s.closeBudget, s.cancel)
close(s.stop)
<-s.done
s.Flush()
// The deadline stays armed through db.Close, which waits for any
// write still in flight — FlushEvery's last batch among them — and
// that write must be allowed to finish inside the budget.
err := s.db.Close()
give.Stop()
s.cancel()
return err
}
// FlushEvery drains the buffer on a timer, so a quiet forge still lands
// its rows. Without it a low-traffic instance would hold observations
// in memory until a batch filled, which on a personal forge could be
// days — and a restart would lose exactly the window most worth having.
func (s *Sink) FlushEvery(d time.Duration, stop <-chan struct{}) {
if d <= 0 {
d = 30 * time.Second
}
t := time.NewTicker(d)
defer t.Stop()
for {
select {
case <-t.C:
s.Flush()
case <-stop:
s.Flush()
return
}
}
}