postgres

package
v0.38.0 Latest Latest
Warning

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

 Go to latest Published: Aug 25, 2026 License: Apache-2.0, MIT Imports: 39 Imported by: 0

Documentation

Overview

Package postgres

Index

Constants

View Source 
const (
	// FidelityUnbacked marks a row with no raw turn behind it — synthetic or
	// transcript-derived. Not a gap: there were never wire bytes to keep.
	FidelityUnbacked = ""

	// FidelityRaw marks a row whose raw turn holds verbatim response bytes,
	// so it can be re-derived from what the upstream actually sent.
	FidelityRaw = "raw"

	// FidelityReduced marks a row whose raw turn holds only an adapter's
	// reduction. Re-derivation is bounded by what that adapter chose to keep.
	FidelityReduced = "reduced"

	// FidelityDegraded marks a row whose verbatim bytes existed and are gone:
	// either they arrived and exceeded the ingest cap, or the producer
	// captured them and withheld them to keep the envelope under the
	// transport limit. Distinct from reduced: this is a capture path that
	// HAD the bytes and a limit that took them, which is a tuning signal
	// rather than a deployment fact.
	//
	// Both causes share this tier on purpose. Fidelity answers what can be
	// re-derived from a row, and neither can — the reason a limit bit does
	// not change the answer. Which limit bit is an operations question,
	// answered by the ingest logs that record each cause separately; giving
	// it a tier would put a distinction nobody re-derives differently into
	// every rollup, and rollups take the worst tier, so the split would be
	// lost at the trace level anyway.
	FidelityDegraded = "degraded"
)

Provenance tiers stamped on the projection. See the 1781470000 migration for what each one means and why 'reduced' and 'degraded' stay distinct.

Variables

This section is empty.

Functions

func Open added in v0.5.0 

func Open(ctx context.Context, dsn string, opts ...PoolOption) (*pgxpool.Pool, error)

Types

type Driver 

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

func NewDriver 

func NewDriver(ctx context.Context, connStr string, opts ...PoolOption) (*Driver, error)

func (*Driver) AcquireDeriveSessionLock added in v0.25.0 

func (d *Driver) AcquireDeriveSessionLock(ctx context.Context, orgID, harnessID, harnessSessionID string) (release func(), err error)

AcquireDeriveSessionLock is the BLOCKING sibling of TryDeriveSessionLock: it waits until the per-session advisory lock is held rather than returning acquired=false. This is the right semantics for a manual/one-off re-derive that must serialize BEHIND the derive worker (finish deriving after the worker releases) instead of skipping. The worker itself keeps using the non-blocking probe, so it never waits on a manual re-derive — only the reverse — which keeps the two free of a lock cycle.

Like TryDeriveSessionLock the lock is connection-scoped: the pooled connection is pinned until release unlocks it (on a fresh background context, since the caller's ctx may already be canceled).

func (*Driver) AggregateSpanStats added in v0.16.0 

func (d *Driver) AggregateSpanStats(ctx context.Context, orgID string, since, until *time.Time, authSubject string) (storage.SpanStats, error)

AggregateSpanStats sums the trace-grain rollups over a time window — the span-layer aggregate behind /v1/stats. A non-empty authSubject narrows every total to that subject's sessions. Implements storage.SpanStatsReader.

func (*Driver) ClearDeriveDirty added in v0.16.0 

func (d *Driver) ClearDeriveDirty(ctx context.Context, e storage.DeriveQueueEntry) (bool, error)

ClearDeriveDirty implements storage.DeriveQueue. The DELETE is guarded on dirtied_at equality so a raw turn landing mid-derive (which bumps dirtied_at) keeps the session queued.

func (*Driver) Close added in v0.5.0 

func (d *Driver) Close() error

func (*Driver) CountRawTurns added in v0.16.0 

func (d *Driver) CountRawTurns(ctx context.Context) (int64, error)

CountRawTurns implements storage.RawTurnStore.

func (*Driver) DB added in v0.5.0 

func (d *Driver) DB() *pgxpool.Pool

func (*Driver) DeleteSession added in v0.22.0 

func (d *Driver) DeleteSession(ctx context.Context, orgID, id string) (bool, error)

DeleteSession removes a session by its org-scoped id and returns whether a row was actually deleted (false when the id was absent). The session_id ON DELETE CASCADE foreign keys tear down the rest of the subtree in the same statement: subagent child sessions (parent_session_id), the session's derived nodes, and its spans/span_turns/span_links. A malformed id is treated as a no-op delete, matching DeleteSkill.

func (*Driver) DeriveQueueStats added in v0.16.0 

func (d *Driver) DeriveQueueStats(ctx context.Context) (storage.DeriveQueueStats, error)

DeriveQueueStats implements storage.DeriveQueue.

func (*Driver) GetDeriveDirty added in v0.16.0 

func (d *Driver) GetDeriveDirty(ctx context.Context, orgID, harnessID, harnessSessionID string) (*storage.DeriveQueueEntry, error)

GetDeriveDirty implements storage.DeriveQueue.

func (*Driver) GetSessionRecord added in v0.12.0 

func (d *Driver) GetSessionRecord(ctx context.Context, orgID, id string) (*storage.SessionRecord, error)

GetSessionRecord returns a single session by its UUID, or nil if not found.

func (*Driver) GetSessionRecordByHarness added in v0.15.0 

func (d *Driver) GetSessionRecordByHarness(
	ctx context.Context,
	orgID string,
	harnessID string,
	harnessSessionID string,
) (*storage.SessionRecord, error)

GetSessionRecordByHarness returns the single session matching the org-scoped natural key (org_id, harness_id, harness_session_id), or nil if no row matches. The lookup is an exact-match point read on the sessions_harness_uq unique index, mirroring the GetSessionRecord nil-on-no-rows contract.

func (*Driver) GetSpanRecord added in v0.16.0 

func (d *Driver) GetSpanRecord(ctx context.Context, orgID, traceID, spanID string) (*storage.SpanRecord, error)

GetSpanRecord returns one span with full payloads. Implements storage.SpanModelReader.

func (*Driver) GetTraceDetail added in v0.16.0 

func (d *Driver) GetTraceDetail(ctx context.Context, orgID, traceID string) (*storage.SpanTurnRecord, []storage.SpanRecord, []storage.SpanLinkRecord, error)

GetTraceDetail returns one turn with its spans and links. Implements storage.SpanModelReader.

func (*Driver) IngestTurn added in v0.10.0 

func (d *Driver) IngestTurn(ctx context.Context, req storage.IngestTurnRequest) (storage.IngestTurnResult, error)

IngestTurn implements storage.SessionIngester for the Postgres driver. The session-tracking flow runs in a single transaction: resolve / UPSERT a sessions row (keyed by the envelope's natural key or a synthetic harness_session_id from the turn's Merkle root), resolve the optional fork-parent FK (placeholder-inserting the parent when its own first turn hasn't landed yet), and fold the derived title. It writes session IDENTITY only: nodes are not persisted (the merkle layer is in-memory), and every derived rollup — counters, model_usage, tasks, kind_counts, and the chain-aware status/git/tool signals — is owned by the derive-time span fold.

func (*Driver) ListDeriveDirty added in v0.16.0 

func (d *Driver) ListDeriveDirty(ctx context.Context, dirtiedBefore, firstDirtiedBefore time.Time, limit int32) ([]storage.DeriveQueueEntry, error)

ListDeriveDirty implements storage.DeriveQueue.

func (*Driver) ListRawTurnHeaders added in v0.16.0 

func (d *Driver) ListRawTurnHeaders(ctx context.Context, orgID, harnessID, harnessSessionID string) ([]storage.RawTurnHeader, error)

ListRawTurnHeaders returns the wire log for one session: capture identity and payload sizes, no blobs. Implements storage.SpanModelReader.

func (*Driver) ListRawTurns added in v0.16.0 

func (d *Driver) ListRawTurns(ctx context.Context, afterID int64, pageSize int32) ([]storage.RawTurnRecord, error)

ListRawTurns implements storage.RawTurnStore. 

func (d *Driver) ListSessionLinks(ctx context.Context, sessionID string) ([]storage.SpanLinkRecord, error)

ListSessionLinks returns a session's dataflow links alone, in the same deterministic key order ListSessionSpanModel serves them. It is the payload-free half of that read, for the per-trace streaming export. Implements storage.SpanModelReader.

func (*Driver) ListSessionRecords added in v0.12.0 

func (d *Driver) ListSessionRecords(
	ctx context.Context,
	orgID string,
	opts storage.SessionListOpts,
) ([]storage.SessionRecord, error)

ListSessionRecords returns a page of sessions for an org ordered by the requested sort column (default last_seen_at DESC), optionally windowed by activity (a turn started in the window, matching /v1/stats) and narrowed to one gateway-stamped JWT subject (exact match on the indexed column; empty lists every user's sessions). Pass zero-value opts to start from the beginning, unwindowed and unfiltered.

func (*Driver) ListSessionRecordsByHarnessSessionID added in v0.36.0 

func (d *Driver) ListSessionRecordsByHarnessSessionID(
	ctx context.Context,
	orgID string,
	harnessSessionID string,
) ([]storage.SessionRecord, error)

ListSessionRecordsByHarnessSessionID returns every session in the org whose harness_session_id exactly matches, across all harnesses. The id is unique within a harness (sessions_harness_uq), so this returns at most one row per harness that has seen it — in practice zero or one. It serves the lone-harness_session_id form of the /v1/sessions filter; the paired form stays on GetSessionRecordByHarness's point lookup. No match is an empty slice, not an error.

func (*Driver) ListSessionSpanModel added in v0.16.0 

func (d *Driver) ListSessionSpanModel(ctx context.Context, sessionID string) ([]storage.SpanTurnRecord, []storage.SpanRecord, []storage.SpanLinkRecord, error)

ListSessionSpanModel returns the stored span projection for one session: turns, spans, and links, each in stable presentation order. Implements storage.SpanModelReader.

func (*Driver) ListTraceSpans added in v0.25.0 

func (d *Driver) ListTraceSpans(ctx context.Context, orgID, traceID string) ([]storage.SpanRecord, error)

ListTraceSpans returns one trace's spans in presentation order (seq ASC, matching ListSessionSpanModel restricted to the trace) — the same per-trace read GetTraceDetail performs, without the turn/link round-trips. Implements storage.SpanModelReader.

func (*Driver) ListTraceSummaries added in v0.16.0 

func (d *Driver) ListTraceSummaries(ctx context.Context, sessionID string) ([]storage.TraceSummaryRecord, error)

ListTraceSummaries returns a session's turn headers with span counts — the lazy session-detail rows. Implements storage.SpanModelReader.

func (*Driver) MarkDeriveDirty added in v0.16.0 

func (d *Driver) MarkDeriveDirty(ctx context.Context, orgID, harnessID, harnessSessionID string) error

MarkDeriveDirty implements storage.DeriveQueue.

func (*Driver) Open added in v0.5.0 

func (d *Driver) Open(ctx context.Context) error

func (*Driver) PutRawTurn added in v0.16.0 

func (d *Driver) PutRawTurn(ctx context.Context, rec storage.RawTurnRecord) (bool, error)

PutRawTurn implements storage.RawTurnStore. The row is appended verbatim; a retried POST with the same (org_id, request_id) is a no-op per the partial unique index.

Session-keyed rows also mark the session dirty in derive_queue, in the same transaction, so the derive worker picks the session up. Marking happens even when the row deduped: a re-POST of an existing turn is the natural "re-derive this session" signal, and a redundant mark only costs one idempotent derive.

func (*Driver) RederiveFromRaw added in v0.16.0 

func (d *Driver) RederiveFromRaw(ctx context.Context, project string) (map[string]*derive.RederiveReport, error)

RederiveFromRaw rebuilds every persisted session from its effective raw turns. Sessions are enumerated from the read model rather than from current raw attribution so a repair source that has become empty is still covered and pruned.

Each session's full read-derive-write pass runs under the same advisory lock used by the derive worker and attribution repair. Holding one lock at a time avoids both stale-read races and lock cycles with repair's ordered two-lock acquisition. The pass is atomic per session, not per org; reports retain the existing per-org shape by aggregating session reports.

func (*Driver) RederiveSession added in v0.16.0 

func (d *Driver) RederiveSession(ctx context.Context, project, orgID, harnessID, harnessSessionID string) (*derive.RederiveReport, error)

RederiveSession is the session-scoped sibling of RederiveFromRaw: re-derive ONE harness session from its raw turns and apply the result transactionally (upsert + prune scoped to that session). This is the derive worker's unit of work — memory stays bounded by one session's unique content, and the full rows stream through the deriver one at a time exactly like the full-org pass.

Same idempotence contract: re-running against an unchanged raw layer upserts the same set and prunes nothing.

func (*Driver) RederiveSessionLocked added in v0.25.0 

func (d *Driver) RederiveSessionLocked(ctx context.Context, project, orgID, harnessID, harnessSessionID string) (*derive.RederiveReport, error)

RederiveSessionLocked is the externally-safe entry point for a session-scoped re-derive that may run WHILE the derive worker is live: it holds the per-session advisory lock across the whole read-derive-write pass, so it cannot interleave with the worker's derive of the same session and prune a turn the worker just wrote. The worker's own path (RederiveSession) is already called under this lock — it takes it in processEntry — so RederiveSession stays lock-free and this wrapper is the one non-worker callers use. Blocking: it waits out a concurrent worker derive rather than skipping, since a manual re-derive must actually run.

func (*Driver) RepairRawTurnAttribution added in v0.30.0 

func (d *Driver) RepairRawTurnAttribution(
	ctx context.Context,
	project string,
	req storage.RawTurnAttributionRepairRequest,
) (storage.RawTurnAttributionRepairResult, error)

RepairRawTurnAttribution appends an attribution correction and rebuilds both affected projections. raw_turns remains byte-for-byte untouched.

func (*Driver) SweepDeriveDirty added in v0.16.0 

func (d *Driver) SweepDeriveDirty(ctx context.Context, activeSince time.Time) (int64, error)

SweepDeriveDirty implements storage.DeriveQueue.

func (*Driver) TryDeriveSessionLock added in v0.16.0 

func (d *Driver) TryDeriveSessionLock(ctx context.Context, orgID, harnessID, harnessSessionID string) (release func(), acquired bool, err error)

TryDeriveSessionLock takes a session-scoped Postgres advisory lock so concurrent workers never double-derive one session. The lock is connection-scoped: the pooled connection is pinned for the lock's lifetime and returned by release. A false return with nil error means another holder has the session — skip it, this is not a failure.

The key hashes the harness triple to 64 bits (FNV-1a). A cross-triple collision only serializes two unrelated sessions' derives — safe, merely slower — so probabilistic keying is fine here.

func (*Driver) UpdateSessionDisplayName added in v0.27.0 

func (d *Driver) UpdateSessionDisplayName(ctx context.Context, orgID, id string, name *string) (int64, error)

UpdateSessionDisplayName sets (or clears, when name is nil) the user-facing `display_name` column for a single session, scoped to the caller's org. It targets display_name — not name — so a Console rename survives ingest re-sending the harness slug every turn (PCC-970). The org_id predicate lives in the SQL (UpdateSessionDisplayName query), never just checked beforehand, so a cross-org id can never be updated. Returns the number of rows affected: 0 means no row matched (unknown id, or an id that exists but belongs to a different org), which the caller (the sessions handler) maps to a 404 rather than treating as success.

type PoolOption added in v0.16.0 

type PoolOption func(*pgxpool.Config)

PoolOption tunes the pgx pool configuration built from the DSN. Options are applied after the DSN is parsed, so they take precedence over pool parameters embedded in the connection string.

func WithConnectTimeout added in v0.16.0 

func WithConnectTimeout(d time.Duration) PoolOption

WithConnectTimeout bounds each connection attempt. pgx has no built-in connect timeout, so an unreachable-but-blackholed host can otherwise hang a startup for the OS TCP timeout (minutes).

func WithMaxConns added in v0.16.0 

func WithMaxConns(n int32) PoolOption

WithMaxConns caps the pool size. Single-purpose services (e.g. the derive worker, which processes one session at a time) should set a small cap instead of inheriting pgx's NumCPU-based default.

Source Files

View all Source files

Directories

Path Synopsis

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.