indexjobs

package
v1.131.2 Latest Latest
Warning

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

 Go to latest Published: Sep 12, 2026 License: Apache-2.0 Imports: 20 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. The queue mechanics — lease-based claim with FOR UPDATE SKIP LOCKED, exponential-backoff retry, reaper sweep, LISTEN/NOTIFY wake-ups, and periodic gap reconciliation — key each unit of work on an opaque (source_kind, source_id) 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 (
	ParkBaseDelay = 30 * time.Minute
	ParkMaxDelay  = 6 * time.Hour
)

ParkBaseDelay and ParkMaxDelay bound the deferral a parked unit waits out. The delay doubles with each further failure past ParkThreshold, starting at ParkBaseDelay and capped at ParkMaxDelay.

Against the failure that motivated this (one unit, 835 identical failures over three days), the sequence 30m, 1h, 2h, 4h, 6h, 6h... cuts three days of re-queues from ~860 to ~14.

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 ParkThreshold = 3

ParkThreshold is how many open failed jobs one unit accumulates before the reconciler starts deferring its re-queue. Each open failed row is MaxAttempts exhausted worker attempts, so the default is 15 attempts spread over at least two reconcile sweeps before anything is deferred: far more than a transient provider blip produces.

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 ErrSourceGone = errors.New("indexjobs: source gone")

ErrSourceGone is the Source's "this unit no longer exists" signal. A LoadItems error wrapping this sentinel tells the worker the source row was deleted on purpose (not that loading failed), so the worker clears the unit's vectors, completes the job, and resolves any open failures for the key instead of recording a terminal failure. Without this distinction a job orphaned by a source delete lands in an open failure that no later success can ever supersede — permanent Degraded residue only an operator dismiss could clear (#998).

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 Enqueuer added in v1.121.0 

type Enqueuer interface {
	Enqueue(ctx context.Context, key Key, trigger Trigger) (created bool, err error)
}

Enqueuer is the write-path subset of Store: the one call a consumer's write path makes after a successful row mutation. Declared separately from Store so a Producer can be bound to a two-line fake in tests without standing up the whole queue.

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.

func (FailedUnit) ParkedUntil added in v1.122.0 

func (u FailedUnit) ParkedUntil() (time.Time, bool)

ParkedUntil reports the unit's deferral for the admin failure-triage surface. It delegates so the sweep that acts on the deferral and the panel that reports it can never disagree about what it is.

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, Running and Succeeded are per-unit latest-status counts:
	// how many units rest on a job in that state.
	Pending   int
	Running   int
	Succeeded int
	// Failed is the number of distinct units carrying at least one open
	// failed job (status='failed' AND resolved_at IS NULL): the same
	// population ActiveFailures enumerates, so the number and the
	// failure list can never disagree about what is open.
	//
	// It is deliberately NOT the latest-status rollup its three
	// siblings are. A failing unit is re-queued, so its newest row is
	// pending rather than failed, and a latest-status Failed reads zero
	// for a kind that is failing every unit continuously — the one
	// situation the count exists to report. A unit is therefore counted
	// under both Pending and Failed while a retry of a failed unit is
	// queued, which is the truth an operator needs: work is queued, and
	// it is queued because it failed.
	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
}

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 ParkCandidate added in v1.122.0 

type ParkCandidate struct {
	Key          Key
	Occurrences  int
	LastFailedAt time.Time
}

ParkCandidate is one unit whose repeated failures make it eligible for the reconciler's deferral: how many open failed jobs it carries and when the last of them landed, which is everything ParkedUntil needs. It is a narrower read than FailedUnit because the sweep needs no error text, attempt counts, or last-success context to decide whether to wait.

func (ParkCandidate) ParkedUntil added in v1.122.0 

func (c ParkCandidate) ParkedUntil() (time.Time, bool)

ParkedUntil reports when the reconciler may re-queue this unit again, and whether it is deferred at all right now.

A unit that has failed ParkThreshold times with no intervening success is failing deterministically far more often than not: every open failed row is MaxAttempts exhausted attempts, and the reconciler enqueues a fresh job every ReconcilerInterval regardless of how the last one went. Left alone that is a job every five minutes forever, which is what produced 835 identical failures on one unit over three days (#1350).

The deferral grows and is capped rather than being permanent, because this counter cannot tell a deterministic input-shape failure from a provider outage long enough to exhaust every unit in the corpus. A permanent park would freeze the whole index behind one outage and wait for a human; a growing one drops the retry rate by two orders of magnitude and still converges on its own once the cause clears.

Only the reconciler consults this. A write to the source and an operator reindex both enqueue as they always did, so the two events most likely to resolve the failure are never delayed by it.

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) CancelPending added in v1.108.0 

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

CancelPending deletes the unit's pending job rows. Running rows are deliberately not touched (a worker holds their lease; the ErrSourceGone path resolves them when LoadItems notices the source is gone).

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.

Failed is the one count that does not come from that subquery. It counts the units holding an open failure, which is the same population ActiveFailures lists; see KindCounts.Failed for why the latest-status reading of it reports zero exactly when a kind is failing continuously.

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) ParkCandidates added in v1.122.0 

func (s *PostgresStore) ParkCandidates(ctx context.Context, minOccurrences, limit int) ([]ParkCandidate, error)

ParkCandidates returns the units whose open failed rows have reached minOccurrences, newest-independent: the ORDER BY is the unit key rather than a timestamp, because a deferred unit stops failing and any recency ordering would sort it out of a bounded window and undo the deferral (see the Store contract).

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 Producer added in v1.121.0 

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

Producer is the write-path enqueue seam for one source kind: a consumer's store calls NotifyWrite after a row mutation commits, and the row enters ranked search in roughly the time one embed takes instead of up to a full ReconcilerInterval. It is the TriggerWrite path for the DB-backed consumers, the counterpart to the api-catalog consumer's EnqueueBestEffort.

It is a distinct object from the queue store because the two ends are constructed in the opposite order from how they are used: a consumer's store is built while the platform assembles its subsystems, and the queue that serves it is assembled afterwards (it needs those subsystems to know which consumers to register). A store therefore takes an unbound Producer at construction and the queue Binds itself once it exists. Until then — and forever, on a deployment with no worker to run the jobs — NotifyWrite is a no-op and the reconciler remains the only path to the index.

Every method is safe on a nil *Producer, so a store built without one needs no nil checks around its notify calls.

One write, one job: the enqueue is idempotent through the index_jobs_open partial unique index, so a write landing while a job for the same row is pending collapses onto that job, which reads the row's committed text when it runs. A write landing while a job for the same row is RUNNING collapses the same way and the in-flight job's snapshot of the row wins; that unit's next convergence signal is its next write, a provider model swap, or an operator re-index, and until one lands the row still matches lexically. That is the framework's pre-existing one-open-job-per-unit contract, not a property of the write path.

func NewProducer added in v1.121.0 

func NewProducer(kind string) *Producer

NewProducer returns an unbound Producer for the given source kind. Callers pass the kind's own SourceKind constant so the kind string keeps one definition per consumer.

func (*Producer) Bind added in v1.121.0 

func (p *Producer) Bind(e Enqueuer)

Bind attaches the queue's job store. Calling it again replaces the binding; passing nil unbinds, which is how a queue that failed to assemble leaves the write paths as they were.

func (*Producer) Kind added in v1.121.0 

func (p *Producer) Kind() string

Kind reports the source kind this Producer enqueues for. The queue reads it to bind only the producers whose consumer actually registered.

func (*Producer) NotifyWrite added in v1.121.0 

func (p *Producer) NotifyWrite(ctx context.Context, sourceID string)

NotifyWrite enqueues a TriggerWrite job for sourceID. Best-effort by contract: a nil receiver, an unbound producer, or a failed insert logs (at most) and returns, leaving the row for the reconciler to converge on its next sweep. The originating write has already committed and must never be failed by an indexing concern.

The enqueue runs on a context detached from the caller's cancellation (bounded by enqueueTimeout) because the write it follows has committed: a client that disconnects the moment its write returns would otherwise cancel the enqueue and leave the row indexed only on the next sweep, which is the latency this path exists to remove.

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 "the unit exists but has
	// nothing to index"; the worker treats that as a clean completion
	// that writes zero vectors. An error wrapping ErrSourceGone means
	// the source row no longer exists (deleted between enqueue and
	// claim); the worker clears the unit's vectors, completes the
	// job, and resolves any open failures for the key. Any other
	// error is a unit-resolve failure and terminates the job into an
	// open failure (the source is 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.
	//
	// A unit that keeps failing does not re-fail every cycle forever:
	// past ParkThreshold open failures the reconciler defers it on a
	// growing delay (see FailedUnit.ParkedUntil) so a deterministic
	// failure is retried on the order of hours rather than minutes,
	// while staying on the admin failure-triage surface.
	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)

	// ParkCandidates returns the units carrying at least minOccurrences
	// open failed jobs: the population the reconciler's deferral applies
	// to. limit bounds the result.
	//
	// It is deliberately not ActiveFailures with a filter. That query
	// orders by last-failure descending, and a deferred unit's last
	// failure is frozen by construction: it stops being re-queued, so it
	// stops failing, while an un-deferred one refreshes on every sweep.
	// Reusing it would make the row limit evict precisely the units that
	// are deferred, re-queue them, and let their fresh timestamps displace
	// others in turn, so on a corpus with more open failures than the
	// limit the deferral would never take hold. This query selects the
	// population by predicate and orders on the key, which no amount of
	// failing can move.
	ParkCandidates(ctx context.Context, minOccurrences, limit int) ([]ParkCandidate, 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)

	// CancelPending deletes the unit's pending job rows, the producer's
	// counterpart to Enqueue for a source delete: work queued for a unit
	// that no longer exists is dropped before a worker wastes a claim on
	// it. Running rows are left alone (a worker holds the lease); their
	// LoadItems surfaces ErrSourceGone and the worker resolves them.
	// Returns the number of rows deleted (zero when nothing was queued,
	// which a delete path treats as success, not an error).
	CancelPending(ctx context.Context, key Key) (canceled 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 StoreOption added in v1.121.0 

type StoreOption func(*StoreOptions)

StoreOption configures a consumer store's optional dependencies at construction. One definition serves every kind: each DB-backed consumer's store constructor takes `opts ...indexjobs.StoreOption`, so a caller with no queue keeps the single-argument form and that store's write path stays reconciler-only.

func WithProducer added in v1.121.0 

func WithProducer(p *Producer) StoreOption

WithProducer sets the Producer a store's write path notifies after a successful mutation. A nil Producer is accepted and leaves the store's write path a no-op, so callers can pass an accessor's result without a nil check.

type StoreOptions added in v1.121.0 

type StoreOptions struct {
	// Producer receives a write-path enqueue after each row mutation the
	// store commits. Nil means no queue is wired.
	Producer *Producer
}

StoreOptions is the resolved set of optional dependencies a consumer store was constructed with.

func ResolveStoreOptions added in v1.121.0 

func ResolveStoreOptions(opts []StoreOption) StoreOptions

ResolveStoreOptions applies opts in order and returns the result. Store constructors call it on their variadic parameter.

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 and that work is
	// evidence of progress: at least one unit is running or pending,
	// and the kind has something to show for the passes that already
	// ran. It takes priority over the other two states so an active
	// pass never reads as degraded or healthy.
	VerdictIndexing Verdict = "indexing"

	// VerdictDegraded means the kind needs attention: an open
	// (unresolved) failure, or a known coverage shortfall (indexed <
	// expected). Queued work does not clear it when that work has
	// produced nothing — a kind failing every unit is re-queued
	// forever, and reporting it as an active pass is how a permanent,
	// total failure hides behind a pending count.
	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 Failed (units holding an open failure), so a kind whose failures have all been dismissed or superseded 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.