indexjobs

package
v1.97.2 Latest Latest
Warning

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

 Go to latest Published: Jul 4, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Package indexjobs is the Postgres-backed, source-kind-agnostic embedding-index job queue the platform's semantic-search consumers share. It is the generalization of the api-catalog embedding queue (formerly pkg/toolkits/apigateway/embedjobs): the queue mechanics (lease-based claim with FOR UPDATE SKIP LOCKED, exponential-backoff retry, reaper sweep, LISTEN/NOTIFY wake-ups, periodic gap reconciliation) carry over unchanged in shape, but the unit of work is keyed on an opaque (source_kind, source_id) pair instead of a catalog-specific (catalog_id, spec_name) pair.

Each consumer plugs in two small contracts:

  • Source declares "what text to embed for a given source_id" (LoadItems) and an optional post-embed hook (OnSucceeded).
  • Sink declares "where vectors for this kind live": the dedup read (ListExisting), the durable writes (Upsert / UpsertBatch), the expected-count breadcrumb (StampExpected), and the per-kind gap query the reconciler diffs against (FindGaps).

The framework owns everything between those contracts: the SHA-256 text-hash dedup, batched calls to the embedding provider, chunk-boundary progress publishing, and the claim/lease/retry/reaper/reconcile state machine. One worker pool, one reaper, and one reconciler serve every registered kind, routing by the source_kind stamped on each job row.

The package depends on Postgres-specific features (SKIP LOCKED, LISTEN/NOTIFY, partial unique indexes). There is no in-memory implementation of the queue because the contract those features encode does not collapse to a single-process map; tests that need a job store use sqlmock or a real Postgres instance, and the Source/Sink contracts are mockable independently.

Index

Constants

View Source 
const DefaultEmbedBatchSize = 32

DefaultEmbedBatchSize is the fallback chunk size when the worker has not been configured with an explicit batch size. 32 keeps a single timed-out batch's lost progress small while amortizing per-call overhead.

View Source 
const DefaultLeaseDuration = 10 * time.Minute

DefaultLeaseDuration is the fallback lease window when the worker / store has not been configured with an explicit duration. 10 minutes is long enough for a large unit against a typical provider and short enough that a genuinely crashed pod's work resumes within minutes. The worker heartbeat extends the active lease while a job is making progress, so this value gates "pod went silent", not "embed batch is slow".

View Source 
const DefaultRetentionDays = 14

DefaultRetentionDays is the fallback age past which the retainer purges terminal job rows (succeeded, or failed-and-resolved). The table accumulates one row per reconciler sweep per unit (every ReconcilerInterval, on every replica), so finished history grows unbounded without a sweep; 14 days keeps a useful window of recent throughput / latency / job-log history while bounding the table. Open failures (status='failed' with resolved_at NULL) and in-flight rows (pending / running) are never purged regardless of age.

View Source 
const MaxAttempts = 5

MaxAttempts caps retries before a job moves to StatusFailed. Set at the package level (not configurable) so the worker and the reconciler share one source of truth. Five attempts with exponential backoff covers ~30 minutes of transient outage; longer outages convert to failed and surface in the admin UI.

View Source 
const NotifyChannel = "index_jobs"

NotifyChannel is the Postgres LISTEN/NOTIFY channel producers and workers use to coordinate low-latency wake-ups. Workers LISTEN on this channel; producers issue pg_notify after a successful enqueue. The payload is intentionally empty because the worker re-queries the table on every wake.

View Source 
const ReaperInterval = 30 * time.Second

ReaperInterval is how often the reaper sweeps for expired leases. Mid-point between fast resumption after a crash and not hammering the DB on every tick.

View Source 
const ReconcilerInterval = 5 * time.Minute

ReconcilerInterval is the gap-detector tick. The reconciler also runs once on pod boot, so the periodic tick is a backstop for "vectors disappeared between boots". Five minutes is generous; the data path tolerates lexical fallback while a gap waits.

View Source 
const RetentionInterval = time.Hour

RetentionInterval is how often the retainer sweeps for expired terminal rows. The purge is cheap and the table is slow-growing relative to the lease/claim churn, so an hourly sweep keeps history bounded without competing with the worker for the DB.

Variables

View Source 
var ErrNoJob = errors.New("indexjobs: no pending job available")

ErrNoJob is returned by Claim when no pending job is available. Workers use it as the wait signal: receive this, block on LISTEN/NOTIFY or the poll tick, then call Claim again.

View Source 
var ErrNotFound = errors.New("indexjobs: job not found")

ErrNotFound is returned by Get when no job with the supplied id exists, and by Complete / Retry / Fail / RenewLease when the (id, worker_id, status) predicate matches no row (the lease has rotated). Distinct from ErrNoJob (a normal idle state) so admin handlers can translate it to a 404 and workers can abandon a rotated lease gracefully.

View Source 
var ErrUnknownKind = errors.New("indexjobs: unknown source kind")

ErrUnknownKind is returned by Reporter.Coverage and Reporter.Reindex when the requested source kind is not registered. Admin handlers translate it to a 404.

Functions

func ContentGap added in v1.75.1 

func ContentGap(items []Item, existing map[string]Vector) bool

ContentGap reports whether the live item set differs from the persisted vectors by membership or text: a unit was added or removed, or an existing item's embed text changed. It is the content-drift gap check a Sink whose corpus lives in the running process (not a countable DB table) uses in FindGaps, so the kind re-indexes on a real change and stays quiet otherwise.

existing is keyed by item id, as Sink.ListExisting returns it. Model and dimension drift (an embedding-provider swap) is intentionally out of scope here: that requires a configuration change and restart, where the boot re-index re-embeds against the new provider through the worker's model-aware dedup (planVectors). The per-sweep reconciler check is about content, so a steady-state corpus produces no jobs.

func TextHash added in v1.75.1 

func TextHash(text string) []byte

TextHash returns the canonical SHA-256 of an item's embed text: the exact hash planVectors stores in Vector.TextHash and dedups against. It is exported so a Sink's FindGaps can detect content drift with the identical hash the worker uses, rather than re-deriving it and risking the two definitions diverging.

Types

type Coverage added in v1.73.0 

type Coverage struct {
	Indexed       int
	Expected      int
	ExpectedKnown bool
}

Coverage reports a kind's indexed-vs-expected vector totals for the admin Indexing dashboard. Indexed is the number of vectors the kind's Sink currently persists. Expected is the number it should hold once fully indexed, and is meaningful only when ExpectedKnown is true: api_catalog stamps an operation_count per spec, so it reports a real ratio; the tools kind re-syncs continuously and stamps no expected count, so it reports ExpectedKnown=false and the dashboard shows a sync indicator instead of an indexed/expected ratio.

type CoverageReporter added in v1.73.0 

type CoverageReporter interface {
	Coverage(ctx context.Context) (Coverage, error)
}

CoverageReporter is an optional Sink capability. A Sink whose vector table can report indexed (and optionally expected) item totals implements it so the admin Indexing surface can render coverage for the kind. Sinks that cannot derive coverage simply do not implement it; the surface falls back to the job-state rollup from Counts.

type FailedUnit added in v1.75.0 

type FailedUnit struct {
	SourceKind string
	SourceID   string
	// LatestJobID is the id of the unit's most recent unresolved failed
	// job: the row the dashboard's drill-in links to for the full,
	// un-redacted error and the job timeline.
	LatestJobID int64
	// LastError is that latest failed job's error, un-redacted (the
	// triage cards group on a redacted signature but drill in to this).
	LastError string
	// Attempts is the latest failed job's worker-attempt count.
	Attempts int
	// Occurrences is how many open failed rows the unit has. A value >1
	// means the unit failed, was retried, and failed again without an
	// intervening success.
	Occurrences int
	// FirstFailedAt is the earliest open failure for the unit ("first
	// seen"); LastFailedAt is the most recent ("last seen").
	FirstFailedAt time.Time
	LastFailedAt  time.Time
	// LastSucceededAt is the unit's most recent successful completion,
	// or nil if it has never succeeded. When set, the dashboard shows
	// "last succeeded Xm ago" to distinguish a unit that used to work
	// from one that never has.
	LastSucceededAt *time.Time
}

FailedUnit is one unit (source_kind, source_id) whose index attempts left an unresolved failure, aggregated for the admin Indexing dashboard's failure-triage surface. Collapsing a unit's repeated failed rows into one entry keeps a unit that failed many times from flooding the panel, while still exposing what an operator needs to triage it: how long it has been failing, whether it ever succeeded, and which job to drill into for the un-redacted error.

type Item 

type Item struct {
	ItemID string
	Text   string
}

Item is one embeddable unit within a Source unit. A unit may yield many items (an api spec yields one per operation) or exactly one (a prompt, a tool, an insight). ItemID is unique within the (source_kind, source_id) pair and is the dedup key the Sink stores vectors against.

type Job 

type Job struct {
	ID             int64
	SourceKind     string
	SourceID       string
	Trigger        Trigger
	Status         Status
	Attempts       int
	LastError      string
	NextRunAt      time.Time
	WorkerID       string
	LeaseExpiresAt *time.Time
	CreatedAt      time.Time
	StartedAt      *time.Time
	CompletedAt    *time.Time
	// ItemsDone is the worker's in-flight progress counter, bumped
	// at every chunk boundary inside a long embed pass so a status
	// endpoint can render "running, N/M" while the final upsert
	// transaction is still pending. Reset to 0 only by Claim; a
	// terminal or reaper-recovered row may still carry a prior
	// attempt's value, so callers gate display on
	// Status == StatusRunning.
	ItemsDone int
}

Job is one row in index_jobs. The struct mirrors the SQL columns one-for-one so admin handlers and tests can construct it directly. LeaseExpiresAt / StartedAt / CompletedAt are pointers because they are nullable in the schema.

type Key 

type Key struct {
	SourceKind string
	SourceID   string
}

Key identifies one unit of indexing work: the corpus (SourceKind) and an opaque, consumer-defined identifier within that corpus (SourceID). For api_catalog the SourceID encodes (catalog_id, spec_name); for a 1:1 corpus like prompts it is the prompt id. The framework never parses SourceID; only the consumer's Source and Sink interpret it.

type KindCounts 

type KindCounts struct {
	SourceKind string
	Pending    int
	Running    int
	Succeeded  int
	Failed     int
	// LastActivity is the most recent moment any job for this kind
	// transitioned: MAX over the kind's rows of the greatest of
	// completed_at, started_at, and created_at. Nil when the kind has
	// no jobs. Computed as a true aggregate (not the newest-by-id row)
	// so an out-of-order completion of an older job is not missed.
	LastActivity *time.Time
	// UnresolvedFailures is the number of distinct units for this kind
	// with at least one open failed job (status='failed' AND
	// resolved_at IS NULL). It is the "is this kind degraded?" signal
	// the dashboard verdict keys on, and is deliberately distinct from
	// Failed: Failed is the per-unit latest-status rollup and still
	// counts a unit whose newest row is failed even after that failure
	// was dismissed (resolved) or superseded, whereas UnresolvedFailures
	// drops to zero the moment every failure is resolved.
	UnresolvedFailures int
}

KindCounts is the per-source-kind job-state roll-up the generic admin index-jobs surface renders. It is computed from index_jobs alone (no per-kind vector table needed), so the framework can produce it for every registered kind uniformly.

type ListFilter 

type ListFilter struct {
	SourceKind string
	SourceID   string
	// SourceIDPrefix matches every job whose source_id begins with
	// the given prefix. Consumers that pack structure into the
	// source_id (api_catalog encodes catalog_id then spec_name)
	// use this to list every unit under a parent without the
	// framework having to understand the encoding.
	SourceIDPrefix string
	Trigger        Trigger
	Status         Status
	Limit          int
}

ListFilter narrows the result set for admin queries against the job table. Zero-value fields are ignored.

type Listener 

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

Listener is the LISTEN side of the LISTEN/NOTIFY adapter. Producers issue NOTIFY in Store.Enqueue; this goroutine receives the notifications and calls Worker.Notify so the worker drops out of its poll wait immediately.

The listener is intentionally separate from the Worker: a pod running multiple Workers can share one Listener by registering multiple notifiers.

func NewListener 

func NewListener(dsn, channel string, notifiers ...notifier) *Listener

NewListener constructs a Listener for the supplied DSN. The listener does not connect until Start is called.

func (*Listener) Start 

func (l *Listener) Start(_ context.Context) error

Start opens the LISTEN connection and spawns the receive goroutine. Errors are returned because a missing notification path silently regresses indexing latency from immediate to the worker's poll interval.

func (*Listener) Stop 

func (l *Listener) Stop()

Stop closes the LISTEN connection and waits for the receive goroutine to drain.

type PostgresStore 

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

PostgresStore implements Store against PostgreSQL. The concrete type is exported so callers can inject the *sql.DB directly; the interface in types.go is the contract.

func NewPostgresStore 

func NewPostgresStore(db *sql.DB, opts ...PostgresStoreOption) *PostgresStore

NewPostgresStore returns a Store backed by db. The caller owns the connection lifecycle. Without options the store uses DefaultLeaseDuration for Claim and RenewLease windows.

func (*PostgresStore) ActiveFailures added in v1.75.0 

func (s *PostgresStore) ActiveFailures(ctx context.Context, sourceKind string, limit int) ([]FailedUnit, error)

ActiveFailures returns the units whose index attempts left an open failure, one entry per unit, most-recently-failed first. The CTE reduces a unit's repeated failed rows to a single row (rn = 1 picks the newest) while aggregating occurrence count and first/last-seen timestamps over the same window, then left-joins the unit's most recent success so the dashboard can show "last succeeded Xm ago". The $1 = ” branch lets one prepared statement serve both the cross-kind triage panel (empty kind) and a per-kind drill-down without string concatenation.

func (*PostgresStore) Claim 

func (s *PostgresStore) Claim(ctx context.Context, workerID string) (*Job, error)

Claim acquires the next runnable pending job across all kinds. Returns ErrNoJob when nothing is available. The query body is wrapped in a single transaction so the SELECT FOR UPDATE row lock persists across the UPDATE that flips status to running. SKIP LOCKED makes concurrent claims across pods non-blocking.

func (*PostgresStore) Complete 

func (s *PostgresStore) Complete(ctx context.Context, id int64, workerID string) error

Complete marks a running job succeeded. The worker_id predicate enforces lease ownership; a rotated worker gets ErrNotFound. On a successful flip it resolves any still-open failed rows for the same unit, so a failure superseded by this success self-clears from the admin Indexing dashboard's triage surface.

func (*PostgresStore) Counts 

func (s *PostgresStore) Counts(ctx context.Context, sourceKind string) (*KindCounts, error)

Counts returns the per-state job roll-up for one source kind, computed from index_jobs alone. The "last status per unit" subquery uses DISTINCT ON so a unit with a long history counts once, under its most recent job's state — matching how the admin surface presents a unit's current status.

func (*PostgresStore) Enqueue 

func (s *PostgresStore) Enqueue(ctx context.Context, key Key, trigger Trigger) (bool, error)

Enqueue inserts a new pending job row. The partial unique index index_jobs_open enforces "at most one pending or running job per (source_kind, source_id)" so a duplicate insert collapses to a no-op. Issues NOTIFY on the queue channel after a successful insert so workers wake immediately; NOTIFY is best-effort.

func (*PostgresStore) Fail 

func (s *PostgresStore) Fail(ctx context.Context, id int64, workerID, errMsg string) error

Fail moves the job to the terminal failed state.

func (*PostgresStore) Get 

func (s *PostgresStore) Get(ctx context.Context, id int64) (*Job, error)

Get returns one job by id.

func (*PostgresStore) LeaseDuration 

func (s *PostgresStore) LeaseDuration() time.Duration

LeaseDuration returns the configured lease window. Exposed so the worker can size its claim-context timeout against the same value the store stamps on Claim.

func (*PostgresStore) List 

func (s *PostgresStore) List(ctx context.Context, filter ListFilter) ([]Job, error)

List returns jobs matching the filter, newest first.

func (*PostgresStore) PurgeTerminal added in v1.77.0 

func (s *PostgresStore) PurgeTerminal(ctx context.Context, retentionDays int) (int, error)

PurgeTerminal deletes finished history older than retentionDays. The predicate matches only rows that are safe to forget: succeeded rows, and failed rows that have already been resolved (resolved_at set, i.e. superseded by a later success or dismissed). It never matches an open failure (status='failed' AND resolved_at IS NULL), so the triage surface keeps every unresolved failure regardless of age, nor a pending/running row (their completed_at is NULL). The cutoff is computed in Go rather than as NOW() - INTERVAL so the boundary is testable with a fixed clock. retentionDays <= 0 is a no-op so a misconfigured caller cannot wipe live history.

The DELETE drains in purgeBatchSize chunks (oldest-first, riding the index_jobs_retention partial index): it returns the running total and stops early if ctx is canceled mid-sweep, treating the deadline as a clean partial pass rather than an error since already-committed batches stand and the next tick resumes.

func (*PostgresStore) ReleaseExpiredLeases 

func (s *PostgresStore) ReleaseExpiredLeases(ctx context.Context) (int, error)

ReleaseExpiredLeases is the reaper's sweep. Flips status=running rows whose lease has elapsed back to pending so another worker can claim them. Does NOT increment attempts; the next Claim does that, preserving the "attempts means worker-tried" invariant.

func (*PostgresStore) RenewLease 

func (s *PostgresStore) RenewLease(ctx context.Context, id int64, workerID string, duration time.Duration) error

RenewLease extends the running job's lease window. The worker's heartbeat calls this at ~lease/3 cadence. The ownership predicate returns ErrNotFound on a rotated lease. duration <= 0 falls back to the store's configured lease so a misconfigured caller never stamps an instant-expire lease.

func (*PostgresStore) ResolveFailures added in v1.75.0 

func (s *PostgresStore) ResolveFailures(ctx context.Context, key Key) (int, error)

ResolveFailures stamps resolved_at on every open failed row for the unit, clearing it from the triage surface. Returns the number of rows resolved (zero when the unit had no open failures, which the dashboard treats as "already resolved", not an error).

func (*PostgresStore) Retry 

func (s *PostgresStore) Retry(ctx context.Context, id int64, workerID, errMsg string) error

Retry releases the lease and reschedules the job with exponential backoff. attempts is NOT incremented here (Claim already did it); the column reflects "how many times a worker started this job", which is the right thing for MaxAttempts comparisons in the caller.

func (*PostgresStore) UpdateProgress 

func (s *PostgresStore) UpdateProgress(ctx context.Context, id int64, workerID string, itemsDone int) error

UpdateProgress publishes the worker's chunk-boundary counter. The (id, worker_id, status='running') predicate enforces ownership; if the lease rotated the UPDATE matches zero rows and returns nil so the calling worker carries on. An error from the DB itself is returned for the worker to log but not retry.

type PostgresStoreOption 

type PostgresStoreOption func(*PostgresStore)

PostgresStoreOption configures a PostgresStore at construction time. Functional options keep NewPostgresStore backward-compatible while letting operators tune fields the constructor would otherwise grow positional arguments for.

func WithLeaseDuration 

func WithLeaseDuration(d time.Duration) PostgresStoreOption

WithLeaseDuration sets the duration the store stamps on a successful Claim and the renewal window for RenewLease. The worker's heartbeat re-stamps lease_expires_at = NOW() + d so the reaper does not release a job that is actively running.

d <= 0 falls back to DefaultLeaseDuration so a misconfigured caller never produces an instant-expire row.

type Reaper 

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

Reaper periodically releases expired leases so jobs whose holding workers crashed return to the queue. One Reaper per pod is fine (the UPDATE is idempotent and the cost is one query per interval) but multiple are also safe: each row's status=running predicate is checked atomically.

func NewReaper 

func NewReaper(store Store, interval time.Duration) *Reaper

NewReaper constructs a Reaper. interval=0 selects ReaperInterval.

func (*Reaper) Start 

func (r *Reaper) Start(_ context.Context)

Start begins the periodic sweep. Safe to call multiple times.

func (*Reaper) Stop 

func (r *Reaper) Stop()

Stop signals shutdown and waits for the goroutine.

type Reconciler 

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

Reconciler is the gap-detection backstop. The producer path (consumer write paths enqueue jobs) is the primary trigger for indexing work; the reconciler covers the cases the producer misses:

  • A source row was written before the embedding provider was configured, so no producer job ran.
  • A producer job ran, failed terminally, and the operator never noticed.
  • A backup/restore brought source rows back without vectors.
  • The kind's vector table was manually pruned for debugging.

Unlike the api-catalog precursor (one SQL statement against one pair of tables), gap detection here is per kind: the indexed count lives in each kind's own vector table, so each Sink owns its FindGaps query. The reconciler walks every registered Sink, asks for the source ids that need (re)indexing, and enqueues a reconciler job for each. The partial unique index on index_jobs makes the enqueue idempotent across pods running the sweep in lock-step.

func NewReconciler 

func NewReconciler(store Store, registry *Registry, interval time.Duration) *Reconciler

NewReconciler constructs a Reconciler. interval=0 selects ReconcilerInterval.

func (*Reconciler) Start 

func (r *Reconciler) Start(_ context.Context)

Start begins the periodic reconciliation loop. The first sweep runs immediately so a freshly-booted pod converges any gaps before its workers go idle.

func (*Reconciler) Stop 

func (r *Reconciler) Stop()

Stop signals shutdown and waits for the goroutine.

type Registry 

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

Registry maps a source_kind to its Source + Sink pair. The worker looks the pair up by the source_kind on each claimed job; the reconciler iterates every registered Sink to detect gaps. One Registry is shared across the worker pool, reaper, and reconciler so all three agree on the set of live kinds.

Registration happens at platform wiring time (before Start), so the map is effectively read-only once the queue is running. The mutex guards the wiring window and any future hot-registration path without forcing callers to reason about ordering.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry returns an empty Registry.

func (*Registry) Kinds 

func (r *Registry) Kinds() []string

Kinds returns every registered source kind, sorted. Exposed for the admin surface and for wiring tests that assert the expected set of consumers registered.

func (*Registry) Lookup 

func (r *Registry) Lookup(kind string) (source Source, sink Sink, ok bool)

Lookup returns the Source and Sink for a source kind. ok is false when the kind is not registered (a job for an unregistered kind, e.g. a leftover row from a removed consumer); the worker terminates such a job rather than spinning on it.

func (*Registry) Register 

func (r *Registry) Register(source Source, sink Sink) error

Register binds a Source and Sink to their shared source kind. Returns an error when source.Kind() and sink.Kind() disagree, or when the kind is already registered, so a wiring mistake fails loudly at startup rather than silently routing jobs to the wrong storage.

func (*Registry) Sinks 

func (r *Registry) Sinks() []Sink

Sinks returns every registered Sink, ordered by kind for stable iteration. The reconciler walks this set calling FindGaps on each.

type Reporter added in v1.73.0 

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

Reporter aggregates the cross-kind health the admin Indexing dashboard renders: the registered kinds, per-kind job-state counts, optional coverage, the job list / drill-down, and the operator-driven re-index command. It composes the queue Store with the Registry so it can dispatch coverage and gap enumeration to each kind's Sink. One Reporter serves every registered kind uniformly, so a new index_jobs consumer gets dashboard visibility for free the moment it registers.

func NewReporter added in v1.73.0 

func NewReporter(store Store, reg *Registry) *Reporter

NewReporter returns a Reporter over the shared queue store and the kind registry. Both must be non-nil; the platform only constructs a Reporter once the queue is wired (database + embedding provider both present), so a nil-guarded accessor returns nil rather than a half-built Reporter.

func (*Reporter) ActiveFailures added in v1.75.0 

func (r *Reporter) ActiveFailures(ctx context.Context, kind string, limit int) ([]FailedUnit, error)

ActiveFailures returns the units whose index attempts left an open failure, one entry per unit, most-recently-failed first, for the dashboard's failure-triage surface. An empty kind lists across every kind. limit bounds the result.

func (*Reporter) Counts added in v1.73.0 

func (r *Reporter) Counts(ctx context.Context, kind string) (*KindCounts, error)

Counts returns the per-state job rollup for one source kind. This is the wiring of indexjobs.Store.Counts that #438 added but never connected to a surface.

func (*Reporter) Coverage added in v1.73.0 

func (r *Reporter) Coverage(ctx context.Context, kind string) (*Coverage, error)

Coverage returns the indexed-vs-expected rollup for the kind, or nil when the kind's Sink does not implement CoverageReporter (the dashboard then renders job-state only for that kind). Returns ErrUnknownKind when the kind is not registered.

func (*Reporter) Kinds added in v1.73.0 

func (r *Reporter) Kinds() []string

Kinds returns every registered source kind, sorted.

func (*Reporter) List added in v1.73.0 

func (r *Reporter) List(ctx context.Context, filter ListFilter) ([]Job, error)

List returns jobs matching the filter, newest first. A zero-value SourceKind lists across every kind, which the dashboard's cross-kind failure triage relies on.

func (*Reporter) Reindex added in v1.73.0 

func (r *Reporter) Reindex(ctx context.Context, kind, sourceID string) ([]string, error)

Reindex enqueues manual-retry jobs for the kind. With a non-empty sourceID it targets exactly that unit (the failure-triage retry button and the per-spec re-embed path). With an empty sourceID it re-enqueues every unit the kind's Sink currently reports as out of sync via FindGaps, which is the one generic per-kind enumeration the framework exposes: for api_catalog that is every spec whose vector count disagrees with its operation_count; for tools it is the tool corpus when its descriptors have drifted from the persisted vectors. Both kinds report no gaps when fully in sync, so a kind-wide re-index of an already-synced kind enqueues nothing and returns an empty list (this matches the documented "re-enqueue every out-of-sync unit" contract; force-re-embedding an unchanged unit is the per-unit path). The manual-retry trigger makes the worker skip its text-hash dedup so every item of an enqueued unit is re-embedded. Returns the source ids enqueued, or ErrUnknownKind when the kind is not registered.

Enqueue is idempotent: a unit that already has an open (pending/running) job is collapsed by the partial unique index, so a re-index issued while a pass is in flight does not double-queue it.

func (*Reporter) Resolve added in v1.75.0 

func (r *Reporter) Resolve(ctx context.Context, kind, sourceID string) (int, error)

Resolve dismisses every open failure for the unit, backing the dashboard's explicit "this failure no longer reflects reality" action. It returns the number of failed rows resolved; zero is not an error (the unit may already be clean, e.g. a concurrent success already resolved it).

Resolve deliberately does NOT require the kind to be registered. The most important failures to dismiss are leftover rows for a kind no consumer registers any more (the "no consumer registered for source_kind" tombstones); requiring registration would make exactly those undismissable. The (kind, sourceID) pair only ever matches the operator's own failing units, so dismissing an unregistered kind is safe.

type Retainer added in v1.77.0 

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

Retainer periodically purges finished job history so index_jobs stays bounded. The table gains one row per reconciler sweep per unit (every ReconcilerInterval, on every replica), so succeeded history grows without limit; the reaper only releases leases and never deletes. The Retainer is that missing sweep: it deletes succeeded and failed-and-resolved rows older than the retention window, while leaving open failures and in-flight rows untouched (see Store.PurgeTerminal).

Like the Reaper, one Retainer per pod is fine and multiple are safe: the DELETE is idempotent (a row another replica already removed simply does not match), so no cross-replica coordination is needed.

func NewRetainer added in v1.77.0 

func NewRetainer(store Store, days int, interval time.Duration) *Retainer

NewRetainer constructs a Retainer. days <= 0 selects DefaultRetentionDays; interval <= 0 selects RetentionInterval. The caller decides whether to start it at all (a deployment that wants unbounded history simply never wires one); once started it always applies a positive window.

func (*Retainer) Start added in v1.77.0 

func (r *Retainer) Start(_ context.Context)

Start begins the periodic sweep. Safe to call multiple times; only the first call starts the goroutine.

func (*Retainer) Stop added in v1.77.0 

func (r *Retainer) Stop()

Stop signals shutdown and waits for the goroutine.

type Sink 

type Sink interface {
	// Kind returns the source_kind this Sink serves. Must match the
	// paired Source.
	Kind() string

	// ListExisting returns the persisted vectors for the unit keyed
	// by item id, for the worker's text-hash + model dedup pass.
	ListExisting(ctx context.Context, key Key) (map[string]Vector, error)

	// Upsert atomically replaces every vector for the unit with the
	// supplied set (delete-absent + insert/update), so a reindex
	// that drops items removes their stale vectors.
	Upsert(ctx context.Context, key Key, rows []Vector) error

	// UpsertBatch writes a single chunk's vectors in place without
	// deleting rows outside the batch. The worker calls this once
	// per chunk so a job that fails mid-pass leaves its prior
	// chunks visible to the next attempt's ListExisting dedup.
	UpsertBatch(ctx context.Context, key Key, rows []Vector) error

	// StampExpected records the unit's expected item count so the
	// reconciler's gap predicate has a target. Called after a
	// successful embed pass with len(items).
	StampExpected(ctx context.Context, key Key, count int) error

	// FindGaps returns the source ids for this kind whose indexed
	// vector count does not match the expected count. The reconciler
	// enqueues a job per returned id; the framework's Enqueue
	// idempotently suppresses any id that already has an open
	// (pending/running) job, so implementations need only diff the
	// kind's expected count against COUNT(*) in the kind's vector
	// table and must NOT consult index_jobs themselves (doing so would
	// re-couple the Sink to the queue's internals).
	FindGaps(ctx context.Context) ([]string, error)
}

Sink is a consumer's "where vectors live" contract. One Sink per source kind is registered with the framework. Sinks own their own physical storage (api_catalog keeps its existing api_catalog_operation_embeddings table; new kinds use the framework's generic per-kind tables), which is why gap detection (FindGaps) and the expected-count breadcrumb (StampExpected) are per-Sink rather than framework-global.

type Source 

type Source interface {
	// Kind returns the source_kind this Source handles. It must
	// match the source_kind producers stamp on jobs and the kind
	// the paired Sink serves.
	Kind() string

	// LoadItems returns every embeddable item for the unit. An
	// empty slice (with a nil error) means "nothing to index"
	// (e.g. the source row was deleted between enqueue and claim);
	// the worker treats that as a clean completion that writes zero
	// vectors. An error is treated as a unit-resolve failure and
	// terminates the job (the source is gone or unreadable; retry
	// will not help).
	LoadItems(ctx context.Context, sourceID string) ([]Item, error)

	// OnSucceeded is an optional post-embed hook called after a
	// successful job completes (e.g. reload live connections so
	// they pick up the new vectors). Implementations that need no
	// hook leave it a no-op.
	OnSucceeded(sourceID string)
}

Source is a consumer's "what to index" contract. One Source per source kind is registered with the framework; the worker calls LoadItems on every claimed job to fetch the current set of embeddable items for the unit (which may have changed since the job was enqueued).

type Status 

type Status string

Status enumerates the state machine an index job moves through. 

pending  -> running         (worker claim)
running  -> succeeded       (worker complete)
running  -> pending         (retryable failure with backoff)
running  -> failed          (attempts exhausted)
running  -> pending         (lease expired, reaper releases)

pending is the only state visible to claim queries; running is the only state visible to the reaper. 

const (
	// StatusPending is the worker-claimable state. The job is
	// visible to SELECT ... FOR UPDATE SKIP LOCKED and will be
	// taken by the next idle worker whose next_run_at <= NOW().
	StatusPending Status = "pending"

	// StatusRunning means a worker holds the lease.
	// lease_expires_at is set; if NOW() passes it, the reaper
	// flips the row back to pending so another worker can take
	// over.
	StatusRunning Status = "running"

	// StatusSucceeded is a terminal state: vectors for the unit
	// match its expected item count.
	StatusSucceeded Status = "succeeded"

	// StatusFailed is terminal for this job row: attempts are
	// exhausted (last_error explains) and the worker will not retry
	// it. It does NOT pin the unit permanently, though: the reconciler
	// enqueues a fresh job for the same unit on its next sweep if a
	// vector gap remains, because the partial unique index only
	// suppresses pending/running rows, not failed ones. So a transient
	// failure self-heals within one reconcile interval, while a
	// permanent failure re-fails each cycle until the operator fixes
	// the cause.
	StatusFailed Status = "failed"
)

Job state values. Stored as TEXT in index_jobs for readability; the column has no CHECK constraint so the migration stays portable to future states without an ALTER.

type Store 

type Store interface {
	// Enqueue inserts a pending job for the supplied key. Returns
	// created=true when a new row was written, created=false when
	// the partial unique index suppressed the insert (a pending or
	// running job for the same key already exists). Producers
	// fire-and-forget; the bool is for tests and metrics.
	Enqueue(ctx context.Context, key Key, trigger Trigger) (created bool, err error)

	// Claim acquires the next runnable job across all kinds via
	// SELECT ... FOR UPDATE SKIP LOCKED. The returned job's status
	// is set to running and lease_expires_at to NOW() + the store's
	// configured lease duration before Claim returns. Returns
	// ErrNoJob when no pending job is available.
	Claim(ctx context.Context, workerID string) (*Job, error)

	// Complete marks the job succeeded. The worker_id predicate
	// enforces lease ownership; a foreign worker whose lease has
	// rotated gets ErrNotFound.
	Complete(ctx context.Context, id int64, workerID string) error

	// UpdateProgress sets items_done on a running job's row.
	// Best-effort: a write that misses because the lease rotated is
	// silently dropped (returns nil). The final Complete is the
	// authoritative success signal.
	UpdateProgress(ctx context.Context, id int64, workerID string, itemsDone int) error

	// Retry releases the lease and reschedules the job with
	// exponential backoff. Used for retryable failures.
	Retry(ctx context.Context, id int64, workerID, errMsg string) error

	// Fail moves the job to the failed terminal state. Used after
	// attempts == MaxAttempts.
	Fail(ctx context.Context, id int64, workerID, errMsg string) error

	// ReleaseExpiredLeases is the reaper's sweep. Flips every
	// status=running row whose lease_expires_at <= NOW() back to
	// pending. Returns the number of rows released.
	ReleaseExpiredLeases(ctx context.Context) (released int, err error)

	// RenewLease extends a running job's lease window by duration.
	// The worker calls this on a timer during a long embed pass.
	// The (id, worker_id, status='running') predicate enforces
	// ownership; a renew from a rotated worker returns ErrNotFound.
	RenewLease(ctx context.Context, id int64, workerID string, duration time.Duration) error

	// Get returns a single job row by id. Returns ErrNotFound when
	// no such id.
	Get(ctx context.Context, id int64) (*Job, error)

	// List returns jobs matching the filter, newest first.
	List(ctx context.Context, filter ListFilter) ([]Job, error)

	// Counts returns the per-state job roll-up for one source kind.
	// Used by the generic admin index-jobs surface.
	Counts(ctx context.Context, sourceKind string) (*KindCounts, error)

	// ActiveFailures returns the units whose index attempts left an
	// open failure (status='failed' AND resolved_at IS NULL), one entry
	// per unit, most-recently-failed first. An empty sourceKind lists
	// across every kind, which the cross-kind triage panel relies on.
	// limit bounds the result; a non-positive or oversized limit falls
	// back to the store default.
	ActiveFailures(ctx context.Context, sourceKind string, limit int) ([]FailedUnit, error)

	// ResolveFailures stamps resolved_at on every open failed row for
	// the unit (status='failed' AND resolved_at IS NULL), clearing it
	// from the triage surface. Returns the number of rows resolved.
	// Backs the dashboard's explicit "dismiss"; Complete performs the
	// same resolution internally when a later job for the unit
	// succeeds, so a superseded failure self-clears.
	ResolveFailures(ctx context.Context, key Key) (resolved int, err error)

	// PurgeTerminal deletes finished job rows older than retentionDays:
	// succeeded rows and failed rows that have been resolved (superseded
	// by a later success or dismissed by an operator). Open failures
	// (status='failed' AND resolved_at IS NULL) and in-flight rows
	// (pending / running) are never deleted, so the failure-triage
	// surface and the active queue are unaffected by retention. Returns
	// the number of rows deleted. A non-positive retentionDays is a
	// no-op (retention disabled). Backs the retainer's periodic sweep.
	PurgeTerminal(ctx context.Context, retentionDays int) (deleted int, err error)
}

Store is the persistence interface for the job queue. The concrete implementation is Postgres-backed (see store_postgres.go); the interface is declared here so worker / reaper / reconciler unit tests can substitute mocks.

The Store deliberately knows nothing about vectors or expected counts: gap detection lives in Sink.FindGaps (per kind, because the indexed-count comes from each kind's own vector table) and the reconciler drives it via Enqueue.

type Trigger 

type Trigger string

Trigger identifies what enqueued a job. Kept on the row for audit (the admin job-history view filters by it) and for the worker's manual-retry handling (which skips the dedup pass).

Trigger is distinct from a job's source_kind: source_kind names the corpus (api_catalog, tools, prompts, ...); Trigger names the event that produced the job within that corpus. 

const (
	// TriggerWrite jobs are enqueued by a consumer's write path
	// (a spec upsert, a prompt save, an insight approval, ...).
	// Every source-row mutation produces one.
	TriggerWrite Trigger = "write"

	// TriggerReconciler jobs are enqueued by the periodic gap
	// detector for units whose indexed-vector count and expected
	// count disagree.
	TriggerReconciler Trigger = "reconciler"

	// TriggerManualRetry jobs are enqueued by an operator-driven
	// force-retry path, the escape hatch for when the embedding
	// model was swapped externally and vectors are stale even
	// though the text hash is unchanged. The worker treats this
	// trigger specially: it skips the dedup pass so every item is
	// re-embedded.
	TriggerManualRetry Trigger = "manual_retry"
)

Job triggers. The set is closed; producers always use one of these literals.

type Vector 

type Vector struct {
	ItemID    string
	TextHash  []byte
	Embedding []float32
	Model     string
	Dim       int
}

Vector is one item's embedding row. TextHash is the SHA-256 of the text fed to the provider, used to skip recomputation when an item's text is unchanged across reindexes. Model and Dim record the provider identity and dimensionality at write time so a model swap has a row-level breadcrumb that cached vectors no longer match the current provider's output.

The same type is used for both the dedup read (Sink.ListExisting) and the write (Sink.Upsert): the fields are identical, so unlike the api-catalog precursor there is no separate "existing" vs "computed" type. Embedding is empty on a freshly-planned row that still needs a provider call and is filled in by the embed loop.

type Verdict added in v1.75.0 

type Verdict string

Verdict is the plain-language health state the admin Indexing dashboard leads with for each kind. It collapses the three structurally different metric families the surface used to render side by side (vector coverage, per-unit job state, and open failures) into one of three words an operator reads at a glance. 

const (
	// VerdictIndexing means work is in flight: at least one unit is
	// running or pending. It takes priority over every other state so
	// an active pass never reads as degraded or healthy.
	VerdictIndexing Verdict = "indexing"

	// VerdictDegraded means the kind needs attention with no active
	// pass to fix it: an open (unresolved) failure, or a known coverage
	// shortfall (indexed < expected) that nothing is currently closing.
	VerdictDegraded Verdict = "degraded"

	// VerdictHealthy is the single resting state: the kind is fully
	// indexed (or in sync), nothing is running or pending, and there are
	// no open failures. It does NOT distinguish a queue-indexed kind from
	// one whose vectors were seeded outside the queue (no job history):
	// both are equally "done", and an earlier split on job-history
	// presence (issue #509's idle_complete) only confused operators with
	// two differently-named green states for the same situation. Recency
	// is conveyed by the response's last_activity, not by the verdict.
	VerdictHealthy Verdict = "healthy"
)

func DeriveVerdict added in v1.75.0 

func DeriveVerdict(c *KindCounts, cov *Coverage) Verdict

DeriveVerdict reduces a kind's job-state counts and optional coverage to a single Verdict. The branch order encodes the priority an operator cares about: active work first, then anything needing attention, else the single resting state. It reads only fields already computed elsewhere, so it is a pure function the admin handler and tests can exercise without a database.

A nil counts (no queue wired) is treated as fully quiescent. The "needs attention" guards use UnresolvedFailures (open failures), not Failed (the per-unit latest-status rollup), so a kind whose newest row is a dismissed or superseded failure is not painted degraded. A coverage shortfall counts only when ExpectedKnown: a kind with no expected target (ExpectedKnown=false) can never be "short".

type Worker 

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

Worker drains the job queue. One Worker instance per pod is the typical deployment; multiple workers in the same pod are supported and race for jobs the same way workers across pods do.

func NewWorker 

func NewWorker(cfg WorkerConfig) *Worker

NewWorker constructs a Worker from the supplied config, filling defaults for the optional fields. The returned Worker is idle until Start is called.

func (*Worker) Concurrency 

func (w *Worker) Concurrency() int

Concurrency reports the number of goroutines this worker spawns on Start. Exposed so wiring tests can assert the configured value flowed through without exporting cfg.

func (*Worker) Notify 

func (w *Worker) Notify()

Notify is the LISTEN/NOTIFY adapter's hook. The listener calls this when a NOTIFY arrives so the worker drops out of its poll wait. Buffered size 1 so a flurry coalesces into a single wake.

func (*Worker) Start 

func (w *Worker) Start(_ context.Context)

Start begins the worker loop. Safe to call multiple times; only the first call spawns goroutines. One run() goroutine per Concurrency unit; each shares the wakeup channel and stopCh and races for jobs through Claim's SKIP LOCKED predicate.

func (*Worker) Stop 

func (w *Worker) Stop()

Stop signals shutdown and waits for the goroutines to exit. Safe to call multiple times.

type WorkerConfig 

type WorkerConfig struct {
	Store    Store
	Registry *Registry

	// Embedder is the platform-wide embedding provider every kind's
	// embed pass uses. One provider serves the whole pool (#438:
	// per-kind model selection is out of scope).
	Embedder embedding.Provider

	WorkerID  string        // empty -> auto-generated
	PollEvery time.Duration // fallback poll interval; default 30s

	// Concurrency is the number of goroutines that share the queue.
	// Each independently calls Claim; the lease + SKIP LOCKED
	// machinery prevents two goroutines (same pod or across pods)
	// from picking the same job. Zero or negative falls back to 1.
	Concurrency int

	// LeaseDuration is the window each Claim stamps and the cadence
	// the heartbeat renews it. Should match the store's
	// WithLeaseDuration. Zero or negative falls back to
	// DefaultLeaseDuration.
	LeaseDuration time.Duration

	// BatchSize is the chunk size the embed pass uses per upstream
	// EmbedBatch call. Zero or negative falls back to
	// DefaultEmbedBatchSize.
	BatchSize int
}

WorkerConfig bundles the worker's dependencies. The worker is kind-agnostic: it resolves the Source and Sink for each claimed job from the Registry by the job's source_kind.

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.