internal/search/indexer.go
Rama main · réplica de lectura del repositorio original.
package search
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"strings"
"sync"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/stump-wtf/cairn/internal/artifact"
"github.com/stump-wtf/cairn/internal/objectstore"
)
// ErrStopped reports that a drain stopped because its context was cancelled. A
// caller running a reindex treats it as "interrupted, re-run me", not as a
// broken index.
var ErrStopped = errors.New("search: indexer stopped")
// Config tunes the indexer worker pool. Zero values fall back to the defaults
// in withDefaults, which mirror the SPEC-0022 design's configuration table.
type Config struct {
// Workers is the pool size (CAIRN_SEARCH_INDEX_WORKERS). It is bounded on
// purpose: indexing reads bodies out of object storage and writes a
// tsvector, and an unbounded pool turns a bulk import into a stampede on
// both (SPEC-0022 "Concurrency Safety").
Workers int
// Batch is how many queue rows one worker claims per pass.
Batch int
// PollInterval is how long an idle worker waits before looking for due
// work again. It only ever applies when the queue is empty: a worker
// that finds work loops immediately, so a backlog drains at full speed.
PollInterval time.Duration
// MaxAttempts bounds a document's retries. Past it the queue row is dropped
// and the failure logged: a permanently poison artifact must not hold the
// pool's attention forever, and `cairn search reindex` is the repair path.
MaxAttempts int
// Backoff and BackoffMax give the retry delay: Backoff doubled per attempt,
// capped at BackoffMax.
Backoff time.Duration
BackoffMax time.Duration
// Language is the Postgres text-search configuration for to_tsvector
// (CAIRN_SEARCH_LANGUAGE, default english).
Language string
// MaxBodyBytes caps one body, and one bundle member, in the index
// (CAIRN_SEARCH_MAX_BODY_BYTES, default 256 KiB).
MaxBodyBytes int
// Logger receives worker lifecycle and failure lines. Nil means
// slog.Default().
Logger *slog.Logger
}
func (c Config) withDefaults() Config {
if c.Workers <= 0 {
c.Workers = 2
}
if c.Batch <= 0 {
c.Batch = 50
}
if c.PollInterval <= 0 {
c.PollInterval = 250 * time.Millisecond
}
if c.MaxAttempts <= 0 {
c.MaxAttempts = 5
}
if c.Backoff <= 0 {
c.Backoff = time.Second
}
if c.BackoffMax <= 0 {
c.BackoffMax = 10 * time.Minute
}
if c.Language == "" {
c.Language = "english"
}
if c.MaxBodyBytes <= 0 {
c.MaxBodyBytes = 256 << 10
}
if c.Logger == nil {
c.Logger = slog.Default()
}
return c
}
// Blobs is the indexer's read surface on content-addressed storage.
// objectstore.ObjectStore satisfies it; the shape is what the indexer needs and
// nothing more, so tests hand it the in-memory store.
type Blobs interface {
Get(ctx context.Context, key string) (io.ReadCloser, error)
}
// Indexer drains search_queue into search_documents with a bounded worker pool.
// It is stateless: all state is the two tables, so any number of Cairn
// processes may run an Indexer against one database and drain it together
// (FOR UPDATE SKIP LOCKED is the only coordination).
type Indexer struct {
pool *pgxpool.Pool
blobs Blobs
cfg Config
log *slog.Logger
}
// New builds an indexer over the same pool the server serves from. blobs may be
// nil, in which case bodies are simply not read (title, tags and comments are
// still indexed) — that is a degenerate configuration, not a supported one, and
// nothing in the server does it.
func New(pool *pgxpool.Pool, blobs Blobs, cfg Config) *Indexer {
cfg = cfg.withDefaults()
return &Indexer{pool: pool, blobs: blobs, cfg: cfg, log: cfg.Logger.With("component", "search-indexer")}
}
// Run starts the worker pool and returns when ctx is cancelled and every
// worker has finished its current document. It is the reaper's contract
// (SPEC-0009 REQ "Concurrency Safety (Expiry Reaper)", SPEC-0012 REQ "Graceful
// Lifecycle"): no goroutine outlives Run, a cancellation mid-document finishes
// that document rather than abandoning it, and nothing is queued that the
// database does not already know about.
func (x *Indexer) Run(ctx context.Context) {
x.log.Info("search indexer started", "workers", x.cfg.Workers, "batch", x.cfg.Batch, "language", x.cfg.Language)
done := make(chan struct{})
var wg sync.WaitGroup
for i := 0; i < x.cfg.Workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
x.workerLoop(ctx)
}()
}
go func() {
wg.Wait()
close(done)
}()
<-done
x.log.Info("search indexer stopped")
}
// workerLoop is one worker: claim, index, repeat; idle only when nothing is
// due. Cancelling ctx stops it between documents, never in the middle of one
// statement's transaction.
func (x *Indexer) workerLoop(ctx context.Context) {
for {
if ctx.Err() != nil {
return
}
items, err := x.claim(ctx)
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return
}
if err != nil {
// A claim failure is a database problem. Back off like a failed
// document rather than spinning, and stay quiet about shutdown.
x.log.Error("search indexer claim failed", "error", err)
if !sleepCtx(ctx, x.cfg.PollInterval) {
return
}
continue
}
if len(items) == 0 {
if !sleepCtx(ctx, x.cfg.PollInterval) {
return
}
continue
}
for _, item := range items {
if ctx.Err() != nil {
return
}
if err := x.indexOne(ctx, item); err != nil {
x.reportFailure(ctx, item, err)
}
}
}
}
// reportFailure re-queues a document with a longer delay, or drops it once it
// has used its retries. The dropped case is logged at Error: the artifact is
// live and searchable only by its absence, and `cairn search reindex` is how an
// operator fixes it.
func (x *Indexer) reportFailure(ctx context.Context, item queuedItem, err error) {
attempts := item.attempts + 1
if attempts >= x.cfg.MaxAttempts {
x.log.Error("search indexer gave up on artifact",
"artifact_id", item.artifactID, "attempts", attempts, "error", err)
return
}
delay := x.backoff(attempts)
if qerr := requeue(ctx, x.pool, item.artifactID, item.reason, attempts, int(delay.Seconds()), err.Error()); qerr != nil {
x.log.Error("search indexer could not requeue artifact",
"artifact_id", item.artifactID, "error", qerr)
return
}
x.log.Warn("search indexer will retry artifact",
"artifact_id", item.artifactID, "attempts", attempts,
"retry_in", delay.String(), "error", err)
}
// backoff doubles per attempt and caps at BackoffMax.
func (x *Indexer) backoff(attempts int) time.Duration {
d := x.cfg.Backoff
for i := 1; i < attempts && d < x.cfg.BackoffMax; i++ {
d *= 2
}
if d > x.cfg.BackoffMax {
d = x.cfg.BackoffMax
}
return d
}
// sleepCtx waits d unless ctx is cancelled first; false means "shut down".
func sleepCtx(ctx context.Context, d time.Duration) bool {
t := time.NewTimer(d)
defer t.Stop()
select {
case <-ctx.Done():
return false
case <-t.C:
return true
}
}
// DrainStats reports one drain for the reindex command and tests (never any
// content).
type DrainStats struct {
Indexed int
// Missing counts documents skipped because their artifact was deleted or
// expired while queued — the delete-during-index no-op, counted.
Missing int
// Failed counts documents whose index attempt errored and were re-queued or
// dropped.
Failed int
}
// Drain processes the queue to empty (or until maxBatches passes, or ctx ends)
// without a worker pool. It is what `cairn search reindex` runs after enqueueing
// so the command's exit means the index is rebuilt, and what tests use instead
// of racing a ticker. maxBatches <= 0 drains everything.
func (x *Indexer) Drain(ctx context.Context, maxBatches int) (DrainStats, error) {
var stats DrainStats
for pass := 0; maxBatches <= 0 || pass < maxBatches; pass++ {
if ctx.Err() != nil {
return stats, ErrStopped
}
items, err := x.claim(ctx)
if err != nil {
return stats, err
}
if len(items) == 0 {
return stats, nil
}
for _, item := range items {
switch err := x.indexOne(ctx, item); {
case errors.Is(err, errArtifactGone):
stats.Missing++
case err != nil:
stats.Failed++
x.reportFailure(ctx, item, err)
default:
stats.Indexed++
}
}
}
return stats, nil
}
// queuedItem is one claimed unit of work. attempts is kept so a retry keeps
// backing off.
type queuedItem struct {
artifactID int64
reason string
attempts int
}
// claimSQL takes a bounded batch of due rows and removes them from the queue in
// the same transaction that selected them, under FOR UPDATE SKIP LOCKED.
//
// Deleting on claim, rather than marking a row locked, is what lets the queue
// stay one row per artifact without a lock column: a worker holds the work, and
// a second event for the same artifact re-inserts a fresh row only after the
// first worker finishes. A worker that dies before its claim commits leaves the
// row for the next pass; one that dies after the claim has committed drops the
// work, which the artifact's next write event or `cairn search reindex`
// repairs. Another worker never sees a claimed row, and never waits on one.
const claimSQL = `
WITH claim AS (
SELECT artifact_id, reason, attempts
FROM search_queue
WHERE not_before <= now()
ORDER BY not_before, artifact_id
LIMIT $1
FOR UPDATE SKIP LOCKED
), taken AS (
DELETE FROM search_queue
WHERE (artifact_id) IN (SELECT artifact_id FROM claim)
RETURNING artifact_id, reason, attempts
)
SELECT artifact_id, reason, attempts FROM taken
ORDER BY artifact_id`
// claim removes and returns up to Batch due rows.
func (x *Indexer) claim(ctx context.Context) ([]queuedItem, error) {
tx, err := x.pool.Begin(ctx)
if err != nil {
return nil, fmt.Errorf("search: begin claim: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
rows, err := tx.Query(ctx, claimSQL, x.cfg.Batch)
if err != nil {
return nil, fmt.Errorf("search: claim work: %w", err)
}
var items []queuedItem
for rows.Next() {
var item queuedItem
if err := rows.Scan(&item.artifactID, &item.reason, &item.attempts); err != nil {
rows.Close()
return nil, fmt.Errorf("search: scan claimed work: %w", err)
}
items = append(items, item)
}
rows.Close()
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("search: claim rows: %w", err)
}
if err := tx.Commit(ctx); err != nil {
return nil, fmt.Errorf("search: commit claim: %w", err)
}
return items, nil
}
// errArtifactGone marks the delete-during-index no-op: the artifact row was
// already gone, or had expired, so there is nothing to index and nothing to
// fail over.
var errArtifactGone = errors.New("search: artifact no longer indexable")
// IndexOne builds and writes one artifact's document now. Tests and the
// reindex path use it directly; the pool reaches it through Drain.
func (x *Indexer) IndexOne(ctx context.Context, artifactID int64) error {
doc, err := x.load(ctx, artifactID)
if err != nil {
return err
}
if doc == nil {
return errArtifactGone
}
return x.write(ctx, doc)
}
// indexOne is IndexOne for a claimed queue row.
func (x *Indexer) indexOne(ctx context.Context, item queuedItem) error {
return x.IndexOne(ctx, item.artifactID)
}
// document is everything one search_documents row needs.
type document struct {
artifactID int64
bodyText string
commentText string
content string
}
// upsertSQL writes the document in one statement, in one transaction: the
// artifact row is the source of the denormalized filter columns (so the copy can
// never disagree with the artifact it indexes), and the weighted tsvector is
// built by Postgres so tokenization stays the database's job.
//
// Two properties matter for SPEC-0022:
//
// - the SELECT reads artifacts, so an artifact deleted or expired since the
// work was claimed produces no rows and therefore neither an insert nor an
// update — delete-during-index is a no-op, and REQ-10's cascade has already
// removed any earlier row;
// - everything user-influenced is a bind parameter, including the text-search
// configuration, so no artifact text or language name reaches the statement
// as SQL ("Database Operation Standards").
const upsertSQL = `
INSERT INTO search_documents
(artifact_id, owner_user_id, owner_team_id, share_type, tags, "schema",
created_at, expires_at, doc, content, indexed_at)
SELECT a.id, a.owner_user_id, a.owner_team_id, a.share_type, a.tags, $2::text,
a.created_at, a.expires_at,
setweight(to_tsvector($3::regconfig, coalesce(a.title, '')), 'A')
|| setweight(to_tsvector($3::regconfig, array_to_string(a.tags, ' ')), 'B')
|| setweight(to_tsvector($3::regconfig, coalesce($4, '')), 'C')
|| setweight(to_tsvector($3::regconfig, coalesce($5, '')), 'D'),
$6, now()
FROM artifacts a
WHERE a.id = $1 AND a.expires_at > now()
ON CONFLICT (artifact_id) DO UPDATE SET
owner_user_id = EXCLUDED.owner_user_id,
owner_team_id = EXCLUDED.owner_team_id,
share_type = EXCLUDED.share_type,
tags = EXCLUDED.tags,
"schema" = EXCLUDED."schema",
created_at = EXCLUDED.created_at,
expires_at = EXCLUDED.expires_at,
doc = EXCLUDED.doc,
content = EXCLUDED.content,
indexed_at = EXCLUDED.indexed_at`
func (x *Indexer) write(ctx context.Context, doc *document) error {
_, err := x.pool.Exec(ctx, upsertSQL,
doc.artifactID, nil, x.cfg.Language, doc.bodyText, doc.commentText, doc.content)
if err != nil {
return fmt.Errorf("search: write document for artifact %d: %w", doc.artifactID, err)
}
return nil
}
// load gathers one artifact's indexed text: its body (or, for a bundle, its
// text members) and its live comments. The title and tags need no work here:
// upsertSQL reads them off the artifact row itself.
func (x *Indexer) load(ctx context.Context, artifactID int64) (*document, error) {
var (
shareType string
mediaType string
bodySHA *string
expiresAt time.Time
)
err := x.pool.QueryRow(ctx, `
SELECT share_type, media_type, body_sha256, expires_at
FROM artifacts WHERE id = $1`, artifactID).
Scan(&shareType, &mediaType, &bodySHA, &expiresAt)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("search: load artifact %d: %w", artifactID, err)
}
if time.Now().After(expiresAt) {
// Expired: invisible to every search anyway, and the reaper is on its
// way. Counting it as "gone" keeps the drain stats honest.
return nil, nil
}
var body strings.Builder
if artifact.ShareType(shareType) == artifact.TypeBundle {
if err := x.appendBundleMembers(ctx, &body, artifactID); err != nil {
return nil, err
}
} else if bodySHA != nil && indexesBody(artifact.ShareType(shareType), mediaType) {
text, err := x.readBlob(ctx, *bodySHA)
if err != nil {
return nil, err
}
body.WriteString(text)
}
comments, err := x.commentText(ctx, artifactID)
if err != nil {
return nil, err
}
content := body.String()
if comments != "" {
if content != "" {
content += "\n\n"
}
content += comments
}
return &document{
artifactID: artifactID,
bodyText: body.String(),
commentText: comments,
content: cappedText(content, x.cfg.MaxBodyBytes+maxCommentTextBytes),
}, nil
}
// readBlob fetches one content-addressed body and returns at most MaxBodyBytes
// of it as text. A blob row or object that has vanished (a concurrent reaper)
// is not an error: the artifact's metadata is still indexed.
func (x *Indexer) readBlob(ctx context.Context, sha string) (string, error) {
if x.blobs == nil {
return "", nil
}
var key string
err := x.pool.QueryRow(ctx, `SELECT storage_key FROM blobs WHERE sha256 = $1`, sha).Scan(&key)
if errors.Is(err, pgx.ErrNoRows) {
return "", nil
}
if err != nil {
return "", fmt.Errorf("search: blob %s: %w", sha, err)
}
rc, err := x.blobs.Get(ctx, key)
if errors.Is(err, objectstore.ErrNotExist) {
return "", nil
}
if err != nil {
return "", fmt.Errorf("search: open blob %s: %w", sha, err)
}
defer func() { _ = rc.Close() }()
buf, err := io.ReadAll(io.LimitReader(rc, int64(x.cfg.MaxBodyBytes)))
if err != nil && !errors.Is(err, io.ErrUnexpectedEOF) {
return "", fmt.Errorf("search: read blob %s: %w", sha, err)
}
return cappedText(string(buf), x.cfg.MaxBodyBytes), nil
}
// appendBundleMembers concatenates a bundle's text members, each under the same
// per-body cap (SPEC-0022 REQ-2). Member order is the bundle's own ordinal
// order, so re-indexing the same bundle produces the same document.
func (x *Indexer) appendBundleMembers(ctx context.Context, out *strings.Builder, bundleID int64) error {
rows, err := x.pool.Query(ctx, `
SELECT m.blob_sha256, m.name, m.media_type
FROM bundle_members m
WHERE m.bundle_id = $1
ORDER BY m.ordinal`, bundleID)
if err != nil {
return fmt.Errorf("search: list bundle %d members: %w", bundleID, err)
}
type member struct {
sha string
name string
mediaType string
}
var members []member
for rows.Next() {
var m member
if err := rows.Scan(&m.sha, &m.name, &m.mediaType); err != nil {
rows.Close()
return fmt.Errorf("search: scan bundle %d members: %w", bundleID, err)
}
members = append(members, m)
}
rows.Close()
if err := rows.Err(); err != nil {
return fmt.Errorf("search: bundle %d member rows: %w", bundleID, err)
}
for _, m := range members {
if !memberIsText(m.mediaType) {
continue
}
text, err := x.readBlob(ctx, m.sha)
if err != nil {
return err
}
if out.Len() > 0 {
out.WriteString("\n\n")
}
out.WriteString(m.name)
out.WriteString("\n")
out.WriteString(text)
}
return nil
}
// commentText concatenates the live comments on an artifact. Soft-deleted
// comments are excluded: SPEC-0022 REQ-2 indexes "the bodies of live,
// non-deleted comments", which is why a delete re-indexes rather than appending.
func (x *Indexer) commentText(ctx context.Context, artifactID int64) (string, error) {
rows, err := x.pool.Query(ctx, `
SELECT body FROM comments
WHERE artifact_id = $1 AND deleted_at IS NULL
ORDER BY id`, artifactID)
if err != nil {
return "", fmt.Errorf("search: list comments for artifact %d: %w", artifactID, err)
}
var out strings.Builder
for rows.Next() {
var body string
if err := rows.Scan(&body); err != nil {
rows.Close()
return "", fmt.Errorf("search: scan comments for artifact %d: %w", artifactID, err)
}
if out.Len() >= maxCommentTextBytes {
break
}
if out.Len() > 0 {
out.WriteString("\n")
}
out.WriteString(body)
}
rows.Close()
if err := rows.Err(); err != nil {
return "", fmt.Errorf("search: comment rows for artifact %d: %w", artifactID, err)
}
return cappedText(out.String(), maxCommentTextBytes), nil
}