Documentation
¶
Overview ¶
Package sqlite is obsdb's default backend: the observability schema in a single SQLite file on the CGO-free modernc.org/sqlite driver (thread/sqlite's choice). Open returns an obsdb.DB that also implements interface{ Hub() obsdb.Hub } — every Write publishes its frames to that in-process hub before returning, which is what setup A's live lane runs on (weft/otel's local sink writes, Studio's SSE handler subscribes, no network).
The handle shape is one writer connection plus a read pool: the local sink's synchronous processors write on the run's goroutine, Studio reads concurrently; WAL lets a reader in another process see the same data through its own Open.
Index ¶
- Variables
- func Open(path string, opts ...Option) (obsdb.DB, error)
- type DB
- func (d *DB) Breakpoints(ctx context.Context, runtimeID string) (_ []string, _ bool, err error)
- func (d *DB) Catalogs(ctx context.Context, runID string) (_ []obsdb.ToolsRecord, err error)
- func (d *DB) Close() error
- func (d *DB) Commands(ctx context.Context, q obsdb.CommandQuery) (_ []obsdb.StoredCommand, err error)
- func (d *DB) Compactions(ctx context.Context, runID string) (_ []obsdb.Compaction, err error)
- func (d *DB) DeleteRun(ctx context.Context, id string) (err error)
- func (d *DB) DeleteSession(ctx context.Context, id string) (err error)
- func (d *DB) Events(ctx context.Context, runID string, after int64, limit int) (_ obsdb.EventPage, err error)
- func (d *DB) Experiment(ctx context.Context, id string) (_ obsdb.Experiment, err error)
- func (d *DB) Experiments(ctx context.Context) (_ []obsdb.Experiment, err error)
- func (d *DB) Hub() obsdb.Hub
- func (d *DB) OtherLogs(ctx context.Context, runID string, q obsdb.LogQuery) (_ obsdb.LogPage, err error)
- func (d *DB) Path() string
- func (d *DB) Prompt(ctx context.Context, runID, hash string) (_ obsdb.PromptRecord, err error)
- func (d *DB) PruneCommands(ctx context.Context, before time.Time, keep int) (n int, err error)
- func (d *DB) PutBreakpoints(ctx context.Context, runtimeID string, tools []string) (err error)
- func (d *DB) PutCommand(ctx context.Context, c obsdb.StoredCommand) (err error)
- func (d *DB) PutRoute(ctx context.Context, r obsdb.RunRoute) (err error)
- func (d *DB) PutRuntime(ctx context.Context, r obsdb.StoredRuntime) (err error)
- func (d *DB) Requests(ctx context.Context, runID string, q obsdb.RequestQuery) (_ []obsdb.RequestRecord, err error)
- func (d *DB) ResolvePublicID(ctx context.Context, publicID string) (_ string, err error)
- func (d *DB) Route(ctx context.Context, runID string) (_ obsdb.RunRoute, _ bool, err error)
- func (d *DB) Run(ctx context.Context, id string) (_ obsdb.RunDetail, err error)
- func (d *DB) RunSpans(ctx context.Context, runID string) (_ []obsdb.Span, err error)
- func (d *DB) Runs(ctx context.Context, q obsdb.RunQuery) (_ obsdb.RunPage, err error)
- func (d *DB) Runtimes(ctx context.Context) (_ []obsdb.StoredRuntime, err error)
- func (d *DB) SaveExperiment(ctx context.Context, e obsdb.Experiment) (err error)
- func (d *DB) Search(ctx context.Context, q obsdb.SearchQuery) (_ obsdb.RunPage, err error)
- func (d *DB) Session(ctx context.Context, id string) (_ obsdb.SessionDetail, err error)
- func (d *DB) Sessions(ctx context.Context, q obsdb.SessionQuery) (_ obsdb.SessionPage, err error)
- func (d *DB) Sweep(ctx context.Context, before time.Time) (n int, err error)
- func (d *DB) Tools(ctx context.Context, runID, hash string) (obsdb.ToolsRecord, error)
- func (d *DB) Trace(ctx context.Context, traceID string) (_ []obsdb.Span, err error)
- func (d *DB) Traces(ctx context.Context, q obsdb.TraceQuery) (_ obsdb.TracePage, err error)
- func (d *DB) Transcript(ctx context.Context, runID string) ([]json.RawMessage, error)
- func (d *DB) TranscriptBatches(ctx context.Context, runID string) (_ []obsdb.TranscriptBatch, err error)
- func (d *DB) Vacuum(ctx context.Context) (err error)
- func (d *DB) Write(ctx context.Context, b obsdb.Batch) (err error)
- type Option
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ErrNewerSchema = errors.New("sqlite: database schema is newer than this obsdb")
ErrNewerSchema is returned by Open when the file's obsdb_migrations is ahead of this binary's highest migration: a database written by a newer weft fails loudly instead of silently ignoring tables it does not know.
Functions ¶
func Open ¶
Open opens (creating if needed) the database at path and brings its schema up to date. ":memory:" works — one in-process database. File connections carry journal_mode=WAL (a persistent property of the file, switched once), synchronous=NORMAL, busy_timeout=30000, foreign_keys=ON and _txlock=immediate (preventing deferred-to-writer upgrade deadlocks); the writer handle allows a single connection so every write is serialized, the read pool serves queries.
Types ¶
type DB ¶
type DB struct {
// contains filtered or unexported fields
}
DB is the SQLite backend. Create it with Open; the zero value is not usable.
func (*DB) Breakpoints ¶ added in v0.15.0
Breakpoints answers runtimeID's stored set.
func (*DB) Commands ¶ added in v0.15.0
func (d *DB) Commands(ctx context.Context, q obsdb.CommandQuery) (_ []obsdb.StoredCommand, err error)
Commands answers the rows q selects, newest first.
func (*DB) Compactions ¶ added in v0.10.0
func (*DB) DeleteRun ¶ added in v0.14.0
DeleteRun removes a run and its subagent children (obsdb.DB).
func (*DB) DeleteSession ¶ added in v0.14.0
DeleteSession removes every run of a session, children included (obsdb.DB).
func (*DB) Experiment ¶
Experiment returns one experiment; ErrNotFound otherwise.
func (*DB) Experiments ¶
Experiments lists the saved experiments, newest update first.
func (*DB) Hub ¶
Hub returns the DB's in-process live hub: every Write publishes its frames to it before returning (D4 — this handle is what setup A's studio.DB(otel.LocalDB()) shares).
func (*DB) OtherLogs ¶ added in v0.10.0
func (d *DB) OtherLogs(ctx context.Context, runID string, q obsdb.LogQuery) (_ obsdb.LogPage, err error)
OtherLogs reads a run's app log records from other_logs, where Write puts every record with no weft.run.id (obsdb.ReadOtherLogs: the attribution through the run's spans, the order, the paging). The candidates are read by trace id (other_logs_trace) inside the run's time window, the first obsdb.MaxLogCandidates by time.
func (*DB) Path ¶ added in v0.11.0
Path returns the database file's absolute path, as Open resolved it against the working directory at open time; "" for ":memory:". Studio reads it for api/meta's db.path (through interface{ Path() string }, so another backend may answer too).
Example ¶
ExampleDB_Path: the file a database lives in, through the interface Studio's api/meta reads (db.path).
package main
import (
"fmt"
"os"
"path/filepath"
"github.com/weftgo/weft/obsdb/sqlite"
// The test writes a fake newer migration by hand.
_ "modernc.org/sqlite"
)
func main() {
dir, _ := os.MkdirTemp("", "obsdb")
defer func() { _ = os.RemoveAll(dir) }()
db, err := sqlite.Open(filepath.Join(dir, "weft.db"))
if err != nil {
panic(err)
}
defer func() { _ = db.Close() }()
p := db.(interface{ Path() string }).Path()
fmt.Println(filepath.Base(p), filepath.IsAbs(p))
}
Output: weft.db true
func (*DB) PruneCommands ¶ added in v0.15.0
PruneCommands applies the store's retention. The terminal commands go in passes of at most obsdb.SweepBatch rows, each its own writer transaction with ctx checked between them — the age rule's passes, then the keep rule's, which recounts in each — so a large backlog never holds the writer for the whole prune, as Sweep never does; the routes, breakpoints and runtimes clean-up is one transaction after them. On an error the passes before it stay deleted and are counted.
func (*DB) PutBreakpoints ¶ added in v0.15.0
PutBreakpoints replaces runtimeID's set whole.
func (*DB) PutCommand ¶ added in v0.15.0
PutCommand upserts one command row whole.
func (*DB) PutRuntime ¶ added in v0.15.0
PutRuntime upserts one runtime's registration whole.
func (*DB) Requests ¶ added in v0.10.0
func (d *DB) Requests(ctx context.Context, runID string, q obsdb.RequestQuery) (_ []obsdb.RequestRecord, err error)
func (*DB) ResolvePublicID ¶
func (*DB) Runtimes ¶ added in v0.15.0
Runtimes answers every stored runtime, most recently seen first.
func (*DB) SaveExperiment ¶
SaveExperiment upserts one experiment by id. A new id stamps Created; an existing one keeps it and moves Updated.
func (*DB) Sessions ¶
func (d *DB) Sessions(ctx context.Context, q obsdb.SessionQuery) (_ obsdb.SessionPage, err error)
Sessions is a GROUP BY session_id over top-level, non-playground runs (S3.4): turns, first/last seen, summed usage, and the newest turn's derived status. A PublicID scopes the runs before the GROUP BY (RunQuery.PublicID's WHERE): every aggregate, the order, Total and the cursor are that public id's runs alone.
func (*DB) Sweep ¶ added in v0.14.0
Sweep deletes the runs not running whose last_seen is before the cutoff, with their children that are not running (obsdb.DB). A run's subtree is swept from its topmost not-running ancestor: a candidate whose parent is a stored run is reached through it, or — when the parent is running, or was never stored — is its own root. Each pass is one writer transaction over at most obsdb.SweepBatch roots, so a write waits on one pass, never on the whole sweep; a pass that found fewer roots than the batch was the last.
func (*DB) Traces ¶ added in v0.14.0
Traces groups the spans table by trace id (obsdb.DB.Traces): one scan of the primary key's order, the filters in HAVING over the group's aggregates — a trace's Start is MIN(start_ns), never one span's — and three keyed reads for the page's roots, services and run ids. The scope filter reads weft.public_id off every span's attrs, the attribute /api/traces/{id}'s scope check reads.
func (*DB) Transcript ¶
func (*DB) TranscriptBatches ¶ added in v0.10.0
func (*DB) Vacuum ¶ added in v0.14.0
Vacuum folds the search index's segments and runs VACUUM (obsdb.DB): the whole file is rewritten, every table's contents unchanged — thread's too, when it shares the file. VACUUM needs the file to itself for its duration; a writer in another process waits on the busy timeout.
func (*DB) Write ¶
Write stores one batch in a single transaction: spans and the event/messages records first (INSERT OR IGNORE on the transport idempotency keys), then one runs upsert per run the batch touched.
The rules (S3.4, D3): deltas are counted, never inserted unless KeepDeltas; heartbeats are never inserted, they only move last_seen; counts increment only for rows actually inserted, so retries don't inflate them; any record or span — heartbeats and deltas included — sets last_seen to its max, started_ns to its min (a span's start, the earliest winning: run_start's time, or the invoke_agent span's just before it) and fills empty identity columns; run_finish sets the terminal fields; the invoke_agent span sets trace, finish, usage, steps and, on error, failed and err. A run first seen through a later record (a reordered batch) gets a provisional started_ns, corrected when the earlier rows land. A stock GenAI span (no weft.run.id, obsdb.StockCandidate) re-derives its trace's derived runs (deriveStock, docs/record-contract.md §9).
After the commit, every frame the batch produced is published to the DB's hub before Write returns — an in-process subscriber sees a record within this call (setup A's live lane).
type Option ¶
type Option interface {
// contains filtered or unexported methods
}
Option configures Open.
func KeepDeltas ¶
func KeepDeltas() Option
KeepDeltas turns delta storage on for debugging (Q4, closed): deltas are counted but never stored by default, on their own counter, so storing them can never open a hole in the durable event sequence.
Source Files