internal/infra/pg/observationstore.go
internal/infra/pg · 590 lines · 34 declarations · source
Declarations
type ObservationStore
type ObservationStore struct{ pool *pgxpool.Pool }
ObservationStore appends turns to the log.
func NewObservationStore
func NewObservationStore(pool *pgxpool.Pool) *ObservationStore
const KindTurn
const KindTurn = "turn"
KindTurn is the observation kind a conversation turn is stored under.
const ProjectionChunk
const ProjectionChunk = "chunk"
ProjectionChunk is the projection kind a chunk registers as. Erasure walks `projection_dependency` rather than a list of tables, so this string is what makes a chunk reachable from the observation it derived from.
method ObservationStore.Append
func (s *ObservationStore) Append(ctx context.Context, schema Schema, turn domain.Turn) (domain.Observation, error)
Append writes a turn: the observation, its messages, and one chunk per message, in one transaction.
Why the offset is claimed inside the transaction
The freshness watermark is the highest offset with NO GAP BELOW IT. A sequence issued outside the transaction leaves a permanent hole whenever a transaction rolls back, and one hole stalls the watermark forever — it would sit below the gap while the log grew above it, reporting a scope as hours behind when it is current.
`max(log_offset) + 1` under a row lock on the scope is the price of that guarantee: appends to one scope serialise. That is the right trade, because the alternative is a number that cannot mean what it says.
Why a chunk per message rather than per turn
Evidence is a byte span, and a span is only meaningful against the string it indexes. A chunk spanning a whole turn would need spans into a concatenation that exists nowhere, so a citation would resolve to the wrong words while looking correct.
func appendTurnTx
func appendTurnTx(ctx context.Context, tx pgx.Tx, schema Schema, turn domain.Turn) (domain.Observation, error)
appendTurnTx writes one new observation inside the transaction that also owns its retry receipt.
const stampRetentionSQL
const stampRetentionSQL = `
UPDATE {schema}.observation o
SET retention_until = now() + p.retention
FROM {schema}.project p
WHERE o.observation_id = $1 AND p.scope = $2 AND p.retention IS NOT NULL`
The project may not exist — a write to an unprovisioned scope still lands, deliberately, because losing memory to a missed provisioning step is the worse failure. No project row means no policy means keep, which is the same answer as a project with no policy.
func nullIfEmpty
func nullIfEmpty(s string) any
const claimOffsetSQL
const claimOffsetSQL = `
INSERT INTO {schema}.watermark (scope, log_offset, watermark_at)
VALUES ($1, 0, $2)
ON CONFLICT (scope) DO UPDATE
SET log_offset = {schema}.watermark.log_offset + 1,
watermark_at = greatest({schema}.watermark.watermark_at, excluded.watermark_at),
updated_at = now()
RETURNING log_offset`
claimOffsetSQL takes the scope's next offset and locks the scope for the rest of the transaction.
The watermark row doubles as the lock, which is why this is an upsert rather than a select: it serialises appends to one scope, and it advances the row that reports freshness in the same statement. A separate advisory lock would need its own key derivation and could disagree with the row it is meant to protect.
`watermark_at` is the TURN's time, not now(): the watermark answers "how far through the customer's history are we", and stamping it with wall-clock time makes a backfill of last year's conversations report as current.
const insertObservationSQL
const insertObservationSQL = `
INSERT INTO {schema}.observation
(observation_id, scope, log_offset, kind, occurred_at, ingested_at, source_role, data_subject_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`
const insertMessageSQL
const insertMessageSQL = `
INSERT INTO {schema}.turn_message
(observation_id, ordinal, role, content, group_ordinal, occurred_at)
VALUES ($1, $2, $3, $4, $5, $6)`
const insertChunkSQL
const insertChunkSQL = `
INSERT INTO {schema}.chunk
(chunk_id, scope, source_observation_id, source_message_ordinal, text, occurred_at, source_role, data_subject_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`
const registerProjectionSQL
const registerProjectionSQL = `
INSERT INTO {schema}.projection_dependency
(source_observation_id, scope, projection_kind, projection_id, data_subject_id)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (source_observation_id, projection_kind, projection_id) DO NOTHING`
method ObservationStore.Messages
func (s *ObservationStore) Messages(ctx context.Context, schema Schema, observationID string) ([]domain.Message, error)
Messages returns a turn's messages in the caller's original order.
const selectMessagesSQL
const selectMessagesSQL = `
SELECT ordinal, role, content, group_ordinal, occurred_at
FROM {schema}.turn_message
WHERE observation_id = $1
ORDER BY ordinal`
method ObservationStore.NextUnformed
func (s *ObservationStore) NextUnformed(ctx context.Context, schema Schema, scope string,
retryAfter time.Duration) (domain.Observation, bool, error)
NextUnformed returns the oldest turn in a scope that extraction has not run over.
Oldest first, because forming out of order would let the formed watermark stall behind a turn nothing is working on while newer turns complete past it — and a watermark that reports a scope as behind when it is nearly current is one people learn to ignore. NextUnformed returns the oldest turn in the scope's backlog that is due for an attempt.
The backlog is what is neither formed nor parked. `retryAfter` holds back a turn whose last attempt failed recently, so one turn that always fails cannot spin and starve the ones behind it — backoff belongs here rather than in a sleep, because the driver may have other scopes to serve while this one waits.
const selectNextUnformedSQL
const selectNextUnformedSQL = `
UPDATE {schema}.observation
SET formation_attempts = formation_attempts + 1,
formation_failed_at = now()
WHERE observation_id = (
SELECT observation_id
FROM {schema}.observation
WHERE scope = $1 AND formed_at IS NULL AND parked_at IS NULL
-- A turn that just failed is not retried on the very next pass, and the wait doubles with
-- each attempt. Backoff is expressed where the backlog is chosen rather than as a sleep, so a
-- scope with one failing turn neither spins on it nor blocks the driver from serving others.
--
-- It doubles because the failure that matters most is the one that is not this turn's fault:
-- a provider outage fails every turn, and a flat retry would burn the attempt budget of the
-- whole backlog inside a few minutes and park all of it. Doubling makes exhausting the budget
-- take long enough that an outage has to be a real one.
AND (formation_failed_at IS NULL
OR formation_failed_at
< now() - ($2::interval * power(2, least(formation_attempts - 1, 6))))
ORDER BY log_offset
FOR UPDATE SKIP LOCKED
LIMIT 1)
RETURNING observation_id::text, log_offset, scope, data_subject_id, occurred_at, ingested_at`
The turn is claimed, not merely read: the attempt is counted here, before the model is called.
Counted on failure instead, an attempt that kills the worker — an out-of-memory, a pod evicted mid-extraction, a machine losing power — was never counted at all, so the turn came back on the next pass with the same count and could be retried forever. Counting at the claim makes every try cost one, whatever happens next, so a turn that reliably kills its worker reaches the attempt bound and parks like any other turn that cannot be formed.
The stamp goes on with it, so the backoff below measures from when the turn was last tried rather than from when somebody last managed to write down why it failed.
method ObservationStore.MarkFormed
func (s *ObservationStore) MarkFormed(ctx context.Context, schema Schema, observationID string) error
MarkFormed records that extraction has run over a turn, and moves the formed watermark.
Why the watermark is derived rather than incremented
An increment per completion is correct only if completions happen in order. They do not: two turns forming concurrently finish in whichever order their model calls return, and an increment would let the number run past a turn that is still unformed. A caller told "current to 40" would then be missing turn 12, which is the one failure a freshness number must not have.
So it is recomputed from the backlog: the offset below the lowest unformed turn, or the highest stored offset when there is no backlog at all. Two index lookups against a partial index that holds the backlog rather than the history.
Why the scope's watermark row is locked first
The recompute reads the backlog and writes a number derived from it. Two of these interleaving would have one overwrite the other's read, and the loser's turn would be missing from a number that claims to cover it. The lock is the same row an append takes, so the two serialise — cheap, because this is two index probes and not a model call.
const markFormedSQL
const markFormedSQL = `
UPDATE {schema}.observation
SET formed_at = now()
WHERE observation_id = $1 AND formed_at IS NULL
RETURNING scope`
Set once and never cleared. Re-forming a turn is a rebuild, which derives from the log rather than repairing rows in place.
const lockScopeWatermarkSQL
const lockScopeWatermarkSQL = `SELECT 1 FROM {schema}.watermark WHERE scope = $1 FOR UPDATE`
const advanceFormedWatermarkSQL
const advanceFormedWatermarkSQL = `
UPDATE {schema}.watermark
SET formed_offset = (
SELECT CASE
WHEN backlog.lowest IS NULL THEN stored.highest
WHEN backlog.lowest = 0 THEN NULL
ELSE backlog.lowest - 1
END
FROM (SELECT min(log_offset) AS lowest FROM {schema}.observation
-- Parked turns are excluded, which is what lets the watermark advance past one
-- the driver gave up on rather than freezing the whole scope behind it. The
-- exception that creates is reported as a count, not hidden.
WHERE scope = $1 AND formed_at IS NULL AND parked_at IS NULL) backlog,
(SELECT max(log_offset) AS highest FROM {schema}.observation
WHERE scope = $1) stored)
WHERE scope = $1`
NULL when the scope's very first turn is still unformed. Not -1: offsets start at zero, so any sentinel is a value in the same domain as the data, and every comparison then carries a clause to exclude it that one query eventually forgets.
method ObservationStore.Freshness
func (s *ObservationStore) Freshness(ctx context.Context, schema Schema, scope string) (domain.Freshness, error)
Freshness reports what a scope has stored and what of it is in memory.
Two numbers, because they answer two questions. An append confirms the first, and the caller already knows it. A recall depends on the second, and it is the one being asked about.
const selectActiveRebuildSQL
const selectActiveRebuildSQL = `
SELECT after_offset, through_offset, sources_rebuilt, sources_skipped
FROM {schema}.fact_rebuild_job
WHERE scope = $1 AND status = 'active'`
Only `active`. A cancelled or completed job is not a reason to distrust a bundle, and reporting one would make the field mean "a rebuild happened here once", which is history rather than state. The partial unique index is what makes this at most one row without an ORDER BY to pick between candidates.
const selectFreshnessSQL
const selectFreshnessSQL = `
SELECT w.log_offset,
w.formed_offset,
(SELECT count(*) FROM {schema}.observation o
WHERE o.scope = w.scope AND o.parked_at IS NOT NULL)
FROM {schema}.watermark w
WHERE w.scope = $1`
The parked count is a subquery against a partial index holding only parked turns, which in a healthy scope is empty. It is read on every freshness check because a number with an unstated exception is worse than a number with a stated one.
method ObservationStore.RecordFormationFailure
func (s *ObservationStore) RecordFormationFailure(ctx context.Context, schema Schema,
observationID, reason string) (attempts int, err error)
RecordFormationFailure remembers why an attempt failed, and returns how many have been made.
The attempt itself was counted when the turn was claimed, so this writes the reason and the time and counts nothing: an attempt that never reaches here, because it took the worker with it, has still been counted. The error is what makes the failure legible without reading a log. Both live on the observation so that erasure takes them with the turn they describe — a provider's error text can contain the message that was sent to it.
const recordFormationFailureSQL
const recordFormationFailureSQL = `
UPDATE {schema}.observation
SET formation_failed_at = now(),
formation_error = $2
WHERE observation_id = $1 AND formed_at IS NULL AND parked_at IS NULL
RETURNING formation_attempts`
method ObservationStore.Park
func (s *ObservationStore) Park(ctx context.Context, schema Schema, observationID string) error
Park stops trying to form a turn, and lets the scope move on.
The observation and its messages are untouched, so this is a decision to stop rather than a deletion: unparking is one update and re-forming is a rebuild. The watermark is recomputed in the same transaction because parking is exactly what allows it to advance, and a park that did not advance it would leave the scope frozen with a row that says it should not be.
const parkSQL
const parkSQL = `
UPDATE {schema}.observation
SET parked_at = now()
WHERE observation_id = $1 AND formed_at IS NULL AND parked_at IS NULL
RETURNING scope`
method ObservationStore.Parked
func (s *ObservationStore) Parked(ctx context.Context, schema Schema, scope string) ([]domain.ParkedTurn, error)
Parked reports the turns a scope gave up on, oldest first.
This is what makes the watermark's exception readable rather than merely counted: an operator asking why a scope has parked turns gets the offsets, the attempts and the reason.
const selectParkedSQL
const selectParkedSQL = `
SELECT observation_id::text, log_offset, formation_attempts, formation_error, parked_at
FROM {schema}.observation
WHERE scope = $1 AND parked_at IS NOT NULL
ORDER BY log_offset`
method ObservationStore.Unpark
func (s *ObservationStore) Unpark(ctx context.Context, schema Schema, observationID string) error
Unpark returns a turn to the backlog, clearing what it learned from failing.
Kept beside Park because a decision to stop trying that cannot be reversed is not a decision, it is a deletion with extra steps.
const unparkSQL
const unparkSQL = `
UPDATE {schema}.observation
SET parked_at = NULL, formation_attempts = 0, formation_failed_at = NULL, formation_error = NULL
WHERE observation_id = $1 AND parked_at IS NOT NULL
RETURNING scope`
method ObservationStore.ScopesWithBacklog
func (s *ObservationStore) ScopesWithBacklog(ctx context.Context, schema Schema) ([]string, error)
ScopesWithBacklog names the scopes that have work waiting.
The driver asks this rather than being told which scopes exist, because a scope is created by the first turn written to it and nothing announces that. One index scan over the backlog index.
method ObservationStore.ScopesWithRetention
func (s *ObservationStore) ScopesWithRetention(ctx context.Context, schema Schema) ([]string, error)
ScopesWithRetention names projects with due stored deadlines. Current policy does not change deadlines already stamped on a source, and artifacts can expire under an indefinite project policy. One indexed existence check per project avoids rescanning a large expiry backlog merely to discover its project on every bounded sweep pass. Unused subject mappings also schedule expiry without any observations; source-backed mappings remain available while attributed bytes survive.