Help us shape the future of Go! Take the 2026 Go Developer Survey

storage

package
v0.3.19 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 4, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package storage provides durable per-request storage using an embedded SQLite database. Records flow in asynchronously from the hot path via a channel, so DB writes never block the proxy.

Index

Constants

This section is empty.

Variables

View Source 
var ErrQueryLimit = errors.New("query result exceeds its byte, row, or column limit")

ErrQueryLimit rejects the entire result; callers must never serve a prefix.

View Source 
var ErrStorageDisabled = errors.New("durable storage is disabled")

ErrStorageDisabled is the one sentinel answering "no durable store is available" (a nil *Store, or no db_path): every surface that refuses an operation for that reason returns or wraps it, so callers classify with errors.Is instead of matching drifted wordings. Each surface keeps its own status code; only the identity and the canonical message are shared.

Functions

func DecodeAttemptsColumn added in v0.3.10 

func DecodeAttemptsColumn(raw []byte) ([]metrics.RetryAttempt, error)

DecodeAttemptsColumn decodes the stored attempts JSON column through the shared decodeJSONColumn contract (external consumers live in internal/web).

func DecodeToolNamesColumn added in v0.3.10 

func DecodeToolNamesColumn(raw []byte) ([]string, error)

DecodeToolNamesColumn decodes the stored tool_names JSON column with the same empty/"null" contract as DecodeAttemptsColumn.

func PendingSnapshotPath added in v0.3.6 

func PendingSnapshotPath(path string) string

PendingSnapshotPath is the staged SQLite file Open admits for path.

func RequestParamColumns added in v0.3.17 

func RequestParamColumns() []string

RequestParamColumns lists, in durable bit order, the optional request parameter COLUMNS whose unset/zero distinction requestParamPresence preserves. It is derived from requestParamFields - the same owner as the bit mapping and the write/read presence - so a column added there cannot drift from what the MCP reference documents, and an external mirror (describe) can be pinned against it instead of a second hand-written list.

func StageSnapshot added in v0.3.6 

func StageSnapshot(path string, data []byte) error

StageSnapshot writes an integrity-checked SQLite snapshot that Open will admit in place of path. The live database is not touched.

Types

type LogCursor 

type LogCursor struct {
	StartMs int64
	ID      string
}

LogCursor is the complete durable ordering key. A timestamp alone cannot distinguish requests started in the same millisecond.

type Options 

type Options struct {
	// WriteChanCap bounds records buffered between the hot path and the
	// background writer; overflow drops records to protect the proxy.
	WriteChanCap int
	// BatchCap is the max number of records flushed per transaction.
	BatchCap int
	// FlushInterval is how often the writer flushes a partial batch.
	FlushInterval time.Duration
	// QueryTimeout bounds ad-hoc dashboard SELECT queries so a slow query
	// cannot hold a read connection indefinitely. Validate requires > 0;
	// withQueryTimeout is the single choke point (a 0 fail-closes).
	QueryTimeout time.Duration
	// QueryMaxBytes/Rows bound ad-hoc SELECT results, not analytical scans.
	// Both are required config values; zero fails closed, never defaults here.
	QueryMaxBytes int
	QueryMaxRows  int
	// WriteTrackCap bounds the in-memory set of recently written record ids.
	// Dashboard aggregates merge un-flushed ring records against stored rows;
	// the written set is the exact dedupe boundary. Sized 2x the ring
	// (history_size) so it always covers every record the ring can hold.
	WriteTrackCap int
	// BusyTimeout shortens the sqliteBusyTimeout guardrail for tests that
	// assert the contention window quickly (a 5s window costs 5s of wall
	// time per phase). Zero keeps the production constant; config has no
	// knob because the guardrail is deliberately not user-tunable.
	BusyTimeout time.Duration
}

Options holds the storage pipeline tunables. All values come from config (single source of truth); nothing is defaulted here.

type ProjectionCursor 

type ProjectionCursor struct {
	Epoch, RowID, Schema int64
	Ready                bool
}

ProjectionCursor pins one decoded analytical history to a consistent durable snapshot. RowID is a persistent high-water mark, not a request start time: late completions can carry arbitrarily old timestamps and still append.

type PurgeFilter 

type PurgeFilter struct {
	Provider       string `json:"provider"`        // exact match
	Model          string `json:"model"`           // exact match
	Client         string `json:"client"`          // exact match
	ConversationID string `json:"conversation_id"` // exact match
	ErrorType      string `json:"error_type"`      // exact match (final error type)
	StatusCode     int    `json:"status_code"`     // exact match (0 = ignore)
	HasError       bool   `json:"has_error"`       // when true, only rows with a final error
	HasDebug       bool   `json:"debug"`           // when true, only operator-debug captures
	BeforeMs       int64  `json:"before_ms"`       // started_at < BeforeMs (0 = ignore)
	AfterMs        int64  `json:"after_ms"`        // started_at >= AfterMs (0 = ignore)
}

PurgeFilter describes a filtered purge: rows matching all non-empty fields are deleted. Since every request is one self-contained row (retries live in the row's attempts JSON, conversation/turn are just fields), deleting rows can never leave orphans.

func (PurgeFilter) IsEmpty 

func (f PurgeFilter) IsEmpty() bool

IsEmpty reports whether the filter matches everything (no constraint set).

func (PurgeFilter) MatchRecord 

func (f PurgeFilter) MatchRecord(r *metrics.Record) bool

MatchRecord is the in-memory predicate equivalent of the SQL filter: it reports whether a buffered record matches every set constraint. Used to drop the same records from the live ring buffer that a filtered Clear deletes from the database, so the two never disagree.

func (*PurgeFilter) UnmarshalJSON 

func (f *PurgeFilter) UnmarshalJSON(data []byte) error

Null is not an omitted constraint: accepting it beside another valid field would silently broaden a destructive selection. Keep this rule at the filter type so count and deletion decode exactly the same way.

func (PurgeFilter) Validate 

func (f PurgeFilter) Validate() error

Validate is shared by count, export, and deletion, including non-HTTP callers.

type PurgeResult 

type PurgeResult struct {
	Deleted       int64
	BufferRemoved int
}

PurgeResult carries the store-level deletion counts Clear returns: durable rows deleted and live-ring records removed. Purely internal - the purge response dropped its wire keys ({"ok": true} only, pinned by the strict response decoder), and the fields are the store-level contract the storage tests assert against.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store is a durable SQLite-backed store. It is safe for concurrent use.

func Open 

func Open(path string, opts Options) (*Store, error)

Open opens (or creates) the SQLite database at path and starts the background writer. Call Close to shut down cleanly. opts carries the pipeline tunables (channel/batch sizes, flush cadence, query timeout).

func (*Store) Clear 

func (s *Store) Clear(ctx context.Context, filter *PurgeFilter, buffer *metrics.Buffer) (PurgeResult, error)

Clear deletes finalized history through a single ring/enqueue fence. A nil filter explicitly selects all history; a supplied empty filter is rejected. Admin callers serialize outside sendMu. The only work under the producer lock is capturing the in-memory boundary and publishing one command; SQL and waits are exclusively on the writer/admin goroutines.

func (*Store) Close 

func (s *Store) Close() error

func (*Store) CountDebugSession 

func (s *Store) CountDebugSession(ctx context.Context, sessionID string) (int64, error)

CountDebugSession returns how many sidecar rows belong to sessionID.

func (*Store) CountWhere 

func (s *Store) CountWhere(ctx context.Context, f PurgeFilter) (int64, error)

CountWhere returns how many stored rows match the filter (for the delete preview). It runs on the read pool like the other filter read paths (ExportWhere/QueryBefore): the HasError predicate's full-table json_each scan must never pin the sole writer connection up to QueryTimeout and stall insertBatch. With WAL the preview reads the committed snapshot, so it may miss the last un-flushed batch - exactly like the ExportWhere it previews, keeping the count==download contract.

func (*Store) DBPath added in v0.3.18 

func (s *Store) DBPath() string

DBPath is the running database's own file: the boot-time path the store opened, not the reloadable db_path config value (db_path is startup-bound, so a reload that changes it leaves the live database at its boot path until the process restarts). Its directory is the volume that owns database-sized transient staging; see stageDir.

func (*Store) DataVersion 

func (s *Store) DataVersion(ctx context.Context) (int64, error)

DataVersion identifies committed database content for aggregate memoization. SQLite versions are comparable only on the same connection, so never query this through the pool or replace a failed connection and reuse its version. Errors disable cache reuse; ordinary history queries can still proceed.

func (*Store) Dropped 

func (s *Store) Dropped() uint64

Dropped returns the number of records dropped due to a full write channel.

func (*Store) ExpireDebugCaptures 

func (s *Store) ExpireDebugCaptures(ctx context.Context, beforeMs int64) (int64, error)

ExpireDebugCaptures deletes sidecar rows whose expires_at is before beforeMs.

func (*Store) ExportWhere 

func (s *Store) ExportWhere(ctx context.Context, f PurgeFilter, w io.Writer) (int64, error)

ExportWhere streams every stored row matching the filter (an empty filter = all rows - export is read-only, so "everything" is safe) as a compact JSON array, one record per line, oldest-first, directly to the writer - memory stays bounded no matter how much history exists. Returns the number of records written.

func (*Store) FilterUnwritten 

func (s *Store) FilterUnwritten(ids []string) []string

FilterUnwritten takes a set of record ids and returns only those not in the written set - the exact dedupe boundary for ring/stored merges. Takes one aggMu hold instead of copying the full written map, so the caller pays O(ring) probes instead of O(written) allocation.

func (*Store) Flush 

func (s *Store) Flush() error

Flush drains every record buffered in the write channel and commits it, then returns - without closing the pipeline (records sent afterwards are written by the writer as usual). This is the restart-handoff primitive: the draining process makes all finalized records durable BEFORE the fresh process opens the database, so the child's totals scan and ring backfill see the complete history while the old process keeps a working store for a failed-handoff resume. Concurrent Close waits for the actual final drain, including its error; a sealed input channel alone does not prove durability.

func (*Store) HandleQuery 

func (s *Store) HandleQuery(w http.ResponseWriter, r *http.Request)

HandleQuery serves a JSON array of rows for a SELECT query passed as ?q=... (URL-decoded). Enforces SELECT-only, single-statement, and a bounded execution time. A nil store returns 503, never inference fallback. Same-origin dashboard endpoint; no CORS header.

func (*Store) ListClients 

func (s *Store) ListClients(ctx context.Context) ([]string, error)

ListClients returns distinct non-empty Record.Client values from history.

func (*Store) ListModels 

func (s *Store) ListModels(ctx context.Context) ([]string, error)

ListModels returns distinct non-empty Record.Model values from history.

func (*Store) ListProviders 

func (s *Store) ListProviders(ctx context.Context) ([]string, error)

ListProviders returns distinct non-empty Record.Provider values from history.

func (*Store) LoadDebugCapture 

func (s *Store) LoadDebugCapture(ctx context.Context, id string) ([]byte, error)

LoadDebugCapture returns the payload for id, or nil if absent or expired.

func (*Store) LoadDebugSessions 

func (s *Store) LoadDebugSessions(ctx context.Context) ([]byte, error)

LoadDebugSessions returns persisted operator-debug session JSON, or nil if none.

func (*Store) LoadMeta 

func (s *Store) LoadMeta(ctx context.Context, key string) (string, error)

LoadMeta returns the value for key, or "" if the key is absent.

func (*Store) LoadPause 

func (s *Store) LoadPause(ctx context.Context) ([]byte, error)

LoadPause returns persisted operator-pause JSON, or nil if none.

func (*Store) LoadRecent 

func (s *Store) LoadRecent(ctx context.Context, n int) ([]*metrics.Record, error)

LoadRecent returns up to n of the newest stored records, oldest-first. It powers the startup backfill (bounded by the configured history_size) so the dashboard survives restarts.

func (*Store) LoadThrottle 

func (s *Store) LoadThrottle(ctx context.Context) ([]byte, error)

LoadThrottle returns persisted provider-throttle JSON, or nil if none.

func (*Store) MarkWritten 

func (s *Store) MarkWritten(ids []string)

MarkWritten primes the written-id set with ids that were persisted BEFORE this process started (the startup ring backfill). Without it a fresh process would treat every backfilled record as "not yet written" and the dashboard aggregates would double-count the whole ring against the store.

func (*Store) MergeSnapshot added in v0.3.6 

func (s *Store) MergeSnapshot(ctx context.Context, data []byte) (inserted, skipped int64, err error)

MergeSnapshot inserts snapshot rows whose ids are not already in the live store (INSERT OR IGNORE). Existing live rows win. The live file is not replaced; no restart is required.

func (*Store) Query 

func (s *Store) Query(ctx context.Context, query string, args ...any) (out []map[string]any, err error)

Query owns ad-hoc SELECT execution time and result resource bounds. The SELECT-only HTTP restriction stays in HandleQuery. SQLite working memory (sorting/aggregate intermediates) is not a strict per-query heap budget.

func (*Store) QueryBefore 

func (s *Store) QueryBefore(ctx context.Context, before *LogCursor, limit int) ([]*metrics.Record, error)

QueryBefore returns a newest-first page below the exclusive cursor. A nil cursor starts at durable newest, independently of the arrival-ordered ring.

func (*Store) Record 

func (s *Store) Record(r *metrics.Record)

Record implements metrics.Recorder. It never blocks: if the channel is full, the record is dropped (metrics are best-effort, not lossless). Records arriving after Close are dropped too: a send on the closed channel panics, which we recover and count as a drop rather than crash the proxy.

func (*Store) Recorder 

func (s *Store) Recorder(buffer *metrics.Buffer) metrics.Recorder

Recorder couples the live ring and durable enqueue at their shared, short publication fence. It does not wait for storage: channel overflow still drops only the durable copy. Lifecycle events remain exclusively in the live buffer.

func (*Store) RenameProviders 

func (s *Store) RenameProviders(ctx context.Context, aliases map[string]string) (int64, error)

RenameProviders merges provider labels in stored history: every row whose provider matches an alias key is rewritten to its canonical label. Driven by config (provider_aliases) - a label split by a naming-scheme change is repaired generically, never with provider-specific code. Serialization with the writer goroutine comes from the single-connection write pool (the same shape a filtered Clear relies on); totals are label-independent, so no rebuild is needed. Malformed pairs are skipped (config Validate already rejects them - this is the last-chance deny), and each UPDATE is parameterized.

func (*Store) SaveDebugCapture 

func (s *Store) SaveDebugCapture(ctx context.Context, id, sessionID string, capturedAt, expiresAt int64, payload []byte) error

SaveDebugCapture writes one request's debug document (sidecar). Synchronous on the writer connection so a drawer fetch after the request finalizes cannot miss the row.

func (*Store) SaveDebugSessions 

func (s *Store) SaveDebugSessions(ctx context.Context, raw []byte) error

SaveDebugSessions persists operator-debug session JSON. Implements proxy.pausePersist.

func (*Store) SaveMeta 

func (s *Store) SaveMeta(ctx context.Context, key, value string) error

SaveMeta writes a durable key/value (operator state, not request rows). Synchronous on the writer connection so a reboot sees the latest value.

func (*Store) SavePause 

func (s *Store) SavePause(ctx context.Context, raw []byte) error

SavePause persists operator-pause JSON. Implements proxy.pausePersist.

func (*Store) SaveThrottle 

func (s *Store) SaveThrottle(ctx context.Context, raw []byte) error

SaveThrottle persists provider-throttle JSON. Implements proxy.pausePersist.

func (*Store) Snapshot added in v0.3.6 

func (s *Store) Snapshot(ctx context.Context) ([]byte, error)

Snapshot returns a packed, integrity-checked copy of the live database, including committed WAL pages. VACUUM INTO is the compact consistent snapshot (no free pages); it is not a file copy of an open DB.

func (*Store) SnapshotOverlap added in v0.3.6 

func (s *Store) SnapshotOverlap(ctx context.Context, data []byte) (int64, error)

SnapshotOverlap is how many request ids in a packed snapshot already exist in the live store. Used by restore inspect.

func (*Store) StreamProjection 

func (s *Store) StreamProjection(ctx context.Context, columns string, cursor ProjectionCursor, fn func(*sql.Rows) error) (ProjectionCursor, bool, error)

StreamProjection materializes either all analytical rows or only append deltas. The epoch, high-water mark, and rows share one read transaction, so another process committing during a scan cannot make its cursor outrun its rows. Rewrites/removals/schema changes require a replacement snapshot. columns is code-owned (the same list consumed by scanContrib), never input from an HTTP request. No prompts, responses, or full records are retained.

func (*Store) Totals 

func (s *Store) Totals() Totals

Totals returns the since-inception aggregate (durably committed rows). For ring-merge consumers prefer TotalsAndUnwritten (atomic pair).

func (*Store) TotalsAndUnwritten 

func (s *Store) TotalsAndUnwritten(ring []*metrics.Record) (Totals, []*metrics.Record)

TotalsAndUnwritten captures totals and filters the caller-owned ring slice under one accounting lock. A racing commit cannot remove a row from both sides of the merge. Only the bounded ring is probed; the written set is not copied, and record folding happens after the lock is released.

func (*Store) TotalsDegraded added in v0.3.10 

func (s *Store) TotalsDegraded() bool

TotalsDegraded reports whether the boot-time totals scan failed, leaving the aggregate a truthful subset (rows committed since the failed scan) instead of since-inception totals.

type Totals 

type Totals struct {
	Requests  int64   `json:"requests"`
	Errors    int64   `json:"errors"`
	Cost      float64 `json:"cost"`
	InputTok  int64   `json:"input_tokens"`
	OutputTok int64   `json:"output_tokens"`
	CacheRead int64   `json:"cache_read_tokens"`
	Reasoning int64   `json:"reasoning_tokens"`
	TTFTSum   int64   `json:"-"`
	TTFTN     int64   `json:"-"`
	TPSSum    float64 `json:"-"`
	TPSN      int64   `json:"-"`
	// Cost-bearing denominators for the KPI's blended unit economics: the
	// count of requests that reported cost > 0 and their in+out tokens.
	// Blended $/req and $/Mtok divide by these, so an unpriced provider
	// (no cost in its response → 0) can never dilute the rate.
	CostReqs  int64   `json:"-"`
	CostInOut float64 `json:"-"` // ratios are floating point; in+out need not fit int64
	// contains filtered or unexported fields
}

Totals is the since-inception cross-request aggregate the dashboard's KPI surface reads. The single writer goroutine maintains it exactly: only rows that have been durably committed are counted, so a dashboard aggregate that merges totals with the un-flushed ring records (deduped via the written-id set) matches the durable history byte-for-byte with no flush-lag window. IsError semantics come from the canonical metrics.Record.IsError.

func (*Totals) Add 

func (t *Totals) Add(r *metrics.Record)

Add is the shared durable/ring accounting owner. Arithmetic failure stays attached to the aggregate until a rebuild, never masquerading as zero.

func (Totals) AvgTPS 

func (t Totals) AvgTPS() *float64

AvgTPS returns the mean over captured throughput samples (tps > 0).

func (Totals) AvgTTFT 

func (t Totals) AvgTTFT() *float64

AvgTTFT returns the mean over captured TTFT samples (ttft_ms > 0), or null when no sample exists - mirrors the dashboard's "- not 0" rule.

func (Totals) Err 

func (t Totals) Err() error

Err reports invalid arithmetic without exposing partial or wrapped totals.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.