quarry: Artifact [05ca8bb805]

Artifact 05ca8bb8051e69b622fb9532aef2287913431cc98f37a789fc2f51b2ce70a427:

File internal/obsduck/obsduck.go — part of check-in [a0fa952495] at 2026-08-26 11:24:13 on branch landing-merge — ops: read the history back — a route across days
The store recorded renders and nothing read them. The panels are one process's memory, so the page could say how fast a route is now and never whether it is worse than it was — which is the question after a deploy.
The drilldown gains a per-day table from the store: p50, p95, max. Nearest-rank percentiles, matching how the snapshot computes them, so two numbers on one page both called p95 agree.
Empty for no traffic and empty for history-off are different answers and the page gives the right one. (user: dmestas, size: 8781) [annotate] [blame] [check-ins using] [more...]
//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 (
	"database/sql"
	"fmt"
	"sync"
	"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
}

// 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)`,
		`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)
		}
	}
	return &Sink{db: db, batch: batch}, nil
}

// Render buffers one page.
func (s *Sink) Render(v obs.Render) {
	s.mu.Lock()
	s.renders = append(s.renders, v)
	full := len(s.renders) >= s.batch
	s.mu.Unlock()
	if full {
		s.Flush()
	}
}

// Subprocess buffers one store call.
func (s *Sink) Subprocess(v obs.Subprocess) {
	s.mu.Lock()
	s.subs = append(s.subs, v)
	full := len(s.subs) >= s.batch
	s.mu.Unlock()
	if full {
		s.Flush()
	}
}

// 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(renders, subs); err != nil {
		s.mu.Lock()
		s.flushErr = err
		s.mu.Unlock()
	}
}

func (s *Sink) write(renders []obs.Render, subs []obs.Subprocess) error {
	tx, err := s.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if len(renders) > 0 {
		st, err := tx.Prepare(`INSERT INTO render VALUES (?, ?, ?, ?, ?, ?, ?)`)
		if err != nil {
			return err
		}
		for _, v := range renders {
			if _, err := st.Exec(v.At, v.Path, v.Repo,
				v.Elapsed.Milliseconds(), v.Ops.Milliseconds(), v.Zit.Milliseconds(),
				v.Status); 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.Exec(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()
}

// 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 flushes and shuts down.
func (s *Sink) Close() error {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return nil
	}
	s.closed = true
	s.mu.Unlock()
	s.Flush()
	return s.db.Close()
}

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