// Package ingest is the 0001 §4 pipeline: a watcher validates the store
// prefix and publishes the new tail as events; a per-node indexer (a
// durable JetStream consumer) folds events into SQLite. The store is the
// event stream's source of truth — cursor loss (pack, shun reclaim) is
// detected, a reset event tells indexers to drop derived state, and the
// census republishes from zero. `quarry reindex` is the same path taken
// on purpose.
package ingest
import (
"context"
"crypto/sha1"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log"
"sync"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/artifact"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/zitcli"
)
// Event is one artifact on the stream: the hash plus its parsed form.
// Opaque blobs never ride the stream — content is served on demand
// through zit, only structure is indexed.
type Event struct {
Hash string `json:"hash"`
Art *artifact.Artifact `json:"artifact"`
}
// Stats is a snapshot of one store's ingest health. A fleet of eleven
// repositories makes the server log useless as a diagnostic — a store
// that has been failing its round every two seconds for an hour reads
// exactly like a healthy one on any page that shows only counts. This
// is the state that tells them apart.
type Stats struct {
// LastOK is when a round last completed without error; zero means
// none ever has.
LastOK time.Time
// LastRound is how long that round took.
LastRound time.Duration
// LastCount is the artifacts published in it, Total across all
// rounds since start.
LastCount int
Total int
// Fallbacks counts artifacts that failed to parse and were treated
// as opaque content. Nonzero is not fatal — most stores carry a few
// — but a jump is how a parser regression announces itself.
//
// Scope is THIS PROCESS, like Total: an artifact is only parsed when
// it is ingested, so a store whose index is already warm reports
// zero however many unparseable artifacts it holds. The number to
// watch is the one a `quarry reindex` produces, not the one a
// long-running server accumulates. Making it a property of the store
// rather than the run means counting fallbacks in the index, which
// is a schema change and a separate piece of work.
Fallbacks int
// LastErr is the most recent round's failure, empty once a round
// succeeds. Failures counts CONSECUTIVE failures, so a flapping
// store and a wedged one are distinguishable.
LastErr string
LastErrAt time.Time
Failures int
// Stalled means the pipeline has stopped retrying on its own
// schedule: stallAfter consecutive failures, each retry
// republishing the store's unfolded tail, is a wedge, not a blip.
// RetryAt is when it will try again — backoff, not a latch, so a
// fixed cause heals without operator ceremony. `quarry reindex`
// bypasses it deliberately.
Stalled bool
RetryAt time.Time
// Gaps counts DAG edges pointing at a check-in this index does
// not hold. Non-zero means artifacts were lost from the event
// window and a rebuild has not closed it — the one health signal
// that says the index is LYING rather than merely behind.
Gaps int
}
// Pipeline owns one store's watcher and indexer ends.
type Pipeline struct {
cli *zitcli.Client
ev *events.Server
ix *index.Index
code string // project code in subjects: zit.evt.<code>.<kind>
node string // this node's durable-consumer suffix (multi-node fan-out)
sub *nats.Subscription // durable pull consumer, created lazily
mu sync.Mutex // serializes Sync rounds
// lists resolves check-in file changes during fold (changes.go).
lists *fileLists
// healed records that a repair has been attempted, and healedGaps
// what it left behind, so a rebuild that cannot close a gap reports
// it once rather than replaying the whole store every few minutes
// forever. Written only from Sync, which is not concurrent with
// itself for a given pipeline.
healed bool
healedGaps int
// tipChecked: the first round of each process asks zit for the
// store's tip and rebuilds IMMEDIATELY if the index lacks it —
// before any bus draining. Repair outranks catch-up: a reader is
// looking at a wrong index NOW, and the rebuild is the fast path.
tipChecked bool
// statMu guards stats and is DELIBERATELY not mu: a round holds mu
// for its whole duration, and a store's first fold runs for minutes.
// Reading health through mu would block every fleet page render on
// the slowest store in the fleet.
statMu sync.Mutex
stats Stats
}
// Stats returns a snapshot of this store's ingest health.
func (p *Pipeline) Stats() Stats {
p.statMu.Lock()
defer p.statMu.Unlock()
return p.stats
}
// record folds one round's outcome into the stats.
func (p *Pipeline) record(n int, d time.Duration, err error) {
p.statMu.Lock()
defer p.statMu.Unlock()
p.stats.LastRound = d
if err != nil {
p.stats.LastErr = err.Error()
p.stats.LastErrAt = time.Now()
p.stats.Failures++
if p.stats.Failures >= stallAfter {
backoff := min(stallBase<<uint(p.stats.Failures-stallAfter), stallMax)
p.stats.Stalled = true
p.stats.RetryAt = time.Now().Add(backoff)
if p.stats.Failures == stallAfter {
log.Printf("ingest: %s: STALLED after %d consecutive failures (%v); backing off — the store is intact and the cursor is held, so nothing is lost, but this index stops advancing until the cause is fixed", p.code, p.stats.Failures, err)
}
}
return
}
p.stats.LastOK = time.Now()
p.stats.LastCount = n
p.stats.Total += n
p.stats.LastErr = ""
p.stats.Failures = 0
p.stats.Stalled = false
p.stats.RetryAt = time.Time{}
}
// fallback counts one artifact that would not parse.
func (p *Pipeline) fallback() {
p.statMu.Lock()
defer p.statMu.Unlock()
p.stats.Fallbacks++
}
// ErrStalled is Sync's answer while the pipeline is backing off after
// repeated failures. It is a sentinel so callers can tell "still broken,
// deliberately not retrying yet" from a fresh failure: the poller skips
// re-logging it, and a write op treats it as "the surface lags" rather
// than failing a write the store has already recorded.
var ErrStalled = errors.New("ingest: stalled; backing off")
// stallAfter is how many consecutive failures make a stall. One is a
// blip, two may be a race with a writer; three rounds of republishing a
// store's tail into a failing fold is a wedge.
const stallAfter = 3
// stallBase scales the backoff: base<<(failures-stallAfter), capped at
// stallMax. A var so tests can compress time.
var stallBase = 30 * time.Second
const stallMax = 10 * time.Minute
// ackWait is how long an unacked message stays invisible before the
// stream redelivers it. A round after a failed fold BLOCKS up to this
// long in the drain loop waiting for that redelivery — the wait is
// honest (breaking early would let the cursor save with messages still
// unfolded), so the knob is here rather than a shortcut there. A var so
// tests can compress it.
var ackWait = 30 * time.Second
// stallWait reports how long Sync must still hold off, zero when it may
// run.
func (p *Pipeline) stallWait() time.Duration {
p.statMu.Lock()
defer p.statMu.Unlock()
if !p.stats.Stalled {
return 0
}
return time.Until(p.stats.RetryAt)
}
// NewPipeline builds a store's pipeline for one node. node distinguishes
// this node's durable consumer from other nodes' on a shared stream, so
// each node folds the FULL stream (a shared durable name would
// load-balance one copy across nodes, leaving indexes partial). Empty
// node is the single-node default.
func NewPipeline(cli *zitcli.Client, ev *events.Server, ix *index.Index, code, node string) *Pipeline {
return &Pipeline{cli: cli, ev: ev, ix: ix, code: code, node: node}
}
// durable is this node's consumer name for the store — valid NATS token
// (the node id, which may be an addr with dots/colons, is hashed).
func (p *Pipeline) durable() string {
if p.node == "" {
return "indexer-" + p.code
}
h := sha1.Sum([]byte(p.node))
return "indexer-" + p.code + "-" + hex.EncodeToString(h[:6])
}
// Reindex is 0001 §5's full rebuild: drop the derived tables and the
// store cursor, then replay — the same convergence path cursor loss
// takes, invoked on purpose. Refolding events already in the stream is
// safe because folds are idempotent.
func (p *Pipeline) Reindex() (int, error) {
// A rebuild is LOCAL DERIVATION, not synchronization: it reads the
// census straight from zit into the fold, skipping the event bus
// entirely. The bus (0001 §4) exists so OTHER consumers can follow
// incrementally; routing a full replay through it cost JSON and
// JetStream persistence per artifact and took hours where fossil's
// in-process rebuild of the same store takes about a minute — and
// its bounded live tail (1GiB, one hour) silently dropped whatever
// out-published the fold. At the end the publisher cursor is set to
// the stream's own trailer, so incremental sync resumes exactly
// where the rebuild left the index. Folds stay idempotent, so any
// stale messages a durable consumer still holds are absorbed.
p.mu.Lock()
defer p.mu.Unlock()
return p.reindexLocked()
}
func (p *Pipeline) reindexLocked() (int, error) {
if err := p.ix.Reset(); err != nil {
return 0, err
}
kv, err := p.ev.Cursors()
if err != nil {
return 0, err
}
if err := kv.Delete(p.code); err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
return 0, err
}
const batchSize = 512
n := 0
start := time.Now()
if err := p.ix.SuspendSearchSync(); err != nil {
return 0, err
}
// The census reader starts FIRST and fills batches while the
// crosslink fold below owns the database.
framesBatches := make(chan []index.Parsed, 8)
var framesErr error
var framesCursor zitcli.Cursor
go func() {
defer close(framesBatches)
pending := make([]index.Parsed, 0, batchSize)
framesCursor, framesErr = p.cli.FramesEachStructural(zitcli.Cursor{}, func(a zitcli.Artifact) error {
if isM, ok := artifact.QuickIsManifest(a.Bytes); ok && isM {
return nil
}
art, perr := artifact.Parse(a.Bytes)
if perr != nil {
p.fallback()
log.Printf("ingest: %s: treating as content: %v", a.Hash, perr)
}
if art == nil || art.Type == artifact.Manifest {
return nil
}
pending = append(pending, index.Parsed{Hash: a.Hash, Art: art})
if len(pending) >= batchSize {
framesBatches <- pending
pending = make([]index.Parsed, 0, batchSize)
}
return nil
})
if len(pending) > 0 {
framesBatches <- pending
}
}()
// Check-in state comes from `zit crosslink` (zit#109): identity,
// author, comment, tags and CHANGED FILES, already derived and
// cache-served — 0.2s for tcl's 42K check-ins. This is what ended
// the era of quarry re-parsing and re-expanding every manifest:
// measured 85-163s of pass-2 work, deleted rather than optimised.
records := make([]index.CheckinRecord, 0, batchSize)
flushRecords := func() error {
if len(records) == 0 {
return nil
}
if err := p.ix.FoldCheckins(records); err != nil {
return err
}
records = records[:0]
return nil
}
if err := p.cli.Crosslink(func(r zitcli.CrosslinkRecord) error {
rec := index.CheckinRecord{
Hash: r.Hash, Date: crosslinkDate(r.MtimeMillis),
User: r.User, Comment: r.Comment,
Parents: r.Parents, Tags: r.Tags,
}
if r.FilesKnown {
rec.Changes = r.Changes
}
records = append(records, rec)
n++
if len(records) >= batchSize {
return flushRecords()
}
return nil
}); err != nil {
return n, err
}
if err := flushRecords(); err != nil {
return n, err
}
crosslinkDone := time.Now()
// Everything that is NOT a check-in still comes from the census:
// tickets, wiki, forum, technotes, attachments, control artifacts.
// Manifests are skipped outright — crosslink already carried their
// every derived fact. Streaming and parsing runs CONCURRENTLY with
// the crosslink fold above having warmed nothing — this goroutine
// was started before it, so the census read and the check-in SQL
// overlap; only the folds themselves serialise on the database.
for pb := range framesBatches {
n += len(pb)
if err := p.ix.FoldBulk(pb); err != nil {
return n, err
}
}
if framesErr != nil {
return n, framesErr
}
cur := framesCursor
streamDone := time.Now()
changesDone := streamDone
if err := p.ix.FinishBulk(); err != nil {
return n, err
}
if err := p.ix.ResumeSearchSync(); err != nil {
return n, err
}
if err := p.ix.RecomputeBranches(); err != nil {
return n, err
}
if err := saveCursor(kv, p.code, cur); err != nil {
return n, err
}
log.Printf("ingest: %s: rebuilt %d records in %.1fs (crosslink %.1fs, frames %.1fs, finish %.1fs)",
p.code, n, time.Since(start).Seconds(), crosslinkDone.Sub(start).Seconds(),
streamDone.Sub(crosslinkDone).Seconds(), time.Since(changesDone).Seconds())
return n, nil
}
// Sync runs one full round: watch (store → events) then fold (events →
// index). Returns the number of artifacts published this round. Safe
// for concurrent callers (the background poll and write ops both call
// it); rounds serialize.
func (p *Pipeline) Sync() (int, error) {
if wait := p.stallWait(); wait > 0 {
return 0, fmt.Errorf("%w for %s more", ErrStalled, wait.Round(time.Second))
}
n, err := p.round()
if err != nil {
return n, err
}
// A round can succeed and still leave the index short. The stream
// is a bounded live tail (1 GiB, one hour, discard-old): a fold
// that keeps failing lets its own messages age out unacked, and the
// publisher's cursor then advances past them on a later healthy
// round. Nothing errors at the moment of loss — the artifacts are
// simply never republished, and every surface built on the index
// renders the hole as if it were the truth. A missing check-in read
// as two branches on the timeline is how this was found.
//
// flushBytes already keeps a single burst inside the size cap; this
// covers the other dimension, where the backlog is small but the
// failure is long.
// Shape before completeness: an index written under a different
// derived shape cannot be repaired by folding forward, so there is
// no point asking whether it is complete first.
if p.ix.Stale() {
log.Printf("ingest: %s: index was written by a different schema; rebuilding", p.code)
if _, err := p.Reindex(); err != nil {
log.Printf("ingest: %s: rebuild failed: %v", p.code, err)
return n, nil
}
p.ix.ClearStale()
return n, nil
}
// A dangling parent is only damage if the STORE still holds that
// artifact. Fossil-born histories reference ancestors the store
// never had — shunned artifacts, or an import that began partway
// down the DAG; fossil, tcl and sqlite all carry some, and a
// rebuild cannot conjure them. Counting those as damage would make
// five of this corpus's repositories rebuild themselves on every
// poll, forever.
lost, gerr := p.lostArtifacts()
if gerr != nil {
log.Printf("ingest: %s: gap check: %v", p.code, gerr)
return n, nil
}
p.statMu.Lock()
p.stats.Gaps = lost
p.statMu.Unlock()
if lost == 0 {
return n, nil
}
// One attempt, then report. A rebuild that does not close the gap
// will not close it on the next poll either, and retrying forever
// costs a full store replay every few minutes — on tk that is
// 69,182 artifacts and about two minutes of work, permanently. An
// index that cannot repair itself is a thing an operator must be
// TOLD about, not something to grind at.
if p.healed && lost >= p.healedGaps {
return n, nil // already tried; no worse than last time
}
if p.healed {
log.Printf("ingest: %s: %d artifacts still missing (was %d) — rebuilding again as it is getting worse", p.code, lost, p.healedGaps)
} else {
log.Printf("ingest: %s: %d artifacts the store holds are missing from the index; rebuilding", p.code, lost)
}
if _, err := p.Reindex(); err != nil {
log.Printf("ingest: %s: rebuild failed: %v", p.code, err)
}
after, aerr := p.lostArtifacts()
if aerr == nil {
p.statMu.Lock()
p.stats.Gaps = after
p.statMu.Unlock()
if after > 0 {
log.Printf("ingest: %s: STILL %d artifacts missing after a full rebuild — the store holds them and the index does not; NOT retrying", p.code, after)
}
p.healedGaps = after
}
p.healed = true
return n, nil
}
// lostArtifacts counts dangling parents the STORE still holds AS
// CHECK-INS — the repairable kind. Holding the hash is not enough: tk's
// history carries a P card naming a file blob (tkMacOSXEvent.c, CVS-
// conversion residue from 2011), and fossil's own database records the
// same edge and draws the same orphan. Counting that as damage sent tk
// through a full 69,182-artifact rebuild that could not possibly close
// the gap — no rebuild conjures a check-in from a C file — and then
// reported the store permanently broken. An artifact that is not a
// manifest is the same category as a shunned ancestor: inherent
// history, not loss.
//
// Bounded: a handful of probes is a fair price per round, and a
// repository with more dangling edges than this is already telling us
// what we need to know.
func (p *Pipeline) lostArtifacts() (int, error) {
dangling, err := p.ix.DanglingParents(danglingProbe)
if err != nil || len(dangling) == 0 {
return 0, err
}
lost := 0
for _, h := range dangling {
raw, err := p.cli.ArtifactRaw(context.Background(), h, 1<<20)
if err != nil {
continue // the store does not hold it: inherent, not loss
}
if art, perr := artifact.Parse(raw); perr != nil || art == nil || art.Type != artifact.Manifest {
continue // held, but not a check-in: no rebuild can index it as one
}
lost++
}
return lost, nil
}
// danglingProbe bounds the per-round cost of the check above: each probe
// is a zit subprocess, and the point is to notice loss, not to audit.
const danglingProbe = 25
// round is one publish-then-fold pass. Rounds serialize; the gap check
// in Sync deliberately runs OUTSIDE this lock, because healing re-enters
// the pipeline.
func (p *Pipeline) round() (int, error) {
p.mu.Lock()
defer p.mu.Unlock()
// No cursor means bootstrap or repair, and both mean the same
// thing: derive the index from the census directly. Publishing the
// whole census through the bus "to catch up" is the hours-long
// path the rebuild was moved off of.
if kv, kerr := p.ev.Cursors(); kerr == nil {
if _, gerr := kv.Get(p.code); gerr != nil {
return p.reindexLocked()
}
}
// Once per process: if the store's own tip is not in the index, the
// index is wrong RIGHT NOW, and no amount of draining old bus
// traffic fixes that faster than deriving from the census does.
if !p.tipChecked {
p.tipChecked = true
if hash, _, err := p.cli.Info(context.Background(), "trunk"); err == nil {
if _, _, ok, ierr := p.ix.CheckinByPrefix(hash); ierr == nil && !ok {
log.Printf("ingest: %s: index lacks the store tip %.12s; rebuilding", p.code, hash)
return p.reindexLocked()
}
}
}
start := time.Now()
n, save, err := p.watch(p.fold)
if err == nil {
err = p.fold()
}
// The cursor is saved ONLY once everything published this round has
// also been folded. The stream is a bounded live tail: messages a
// failing fold leaves unacked age out of it, and a cursor that has
// already moved past them means nothing ever republishes those
// artifacts — the loss is silent and permanent (the 2026-08-19
// schema incident). Held back, a fold failure costs only the re-read
// and republish of the same tail next round, which idempotent folds
// absorb.
if err == nil && save != nil {
err = save()
}
p.record(n, time.Since(start), err)
return n, err
}
// flushBytes bounds how much a single publish burst may put on the
// stream before the caller is made to fold. ZITEVT is capped and
// discards OLD messages (events.go), so a census larger than the cap
// used to delete its own head while the indexer was still folding it —
// the index came out silently partial and then declared itself done.
// Keeping each burst far inside the window makes the stream apply
// backpressure instead of dropping. Well under the 1 GiB ceiling even
// with every repo ingesting at once.
var flushBytes = 64 << 20
// watch polls the store and publishes new structural artifacts. flush
// is called whenever the un-folded burst reaches flushBytes; it must
// drain the consumer (see Sync). The returned save advances the
// publisher cursor; round calls it only after the final fold succeeds.
func (p *Pipeline) watch(flush func() error) (int, func() error, error) {
kv, err := p.ev.Cursors()
if err != nil {
return 0, nil, err
}
cur := loadCursor(kv, p.code)
// Publish AS THE CENSUS IS READ. Collecting it first would hold the
// whole thing in memory — sqlite's stream is 13.2GB and tcl's is
// larger — and the peak is now one artifact plus the in-flight
// backlog, whatever the store's size.
published, unfolded := 0, 0
each := func(a zitcli.Artifact) error {
art, perr := artifact.Parse(a.Bytes)
if perr != nil {
p.fallback()
log.Printf("ingest: %s: treating as content: %v", a.Hash, perr)
}
if art == nil {
return nil
}
msg, err := json.Marshal(Event{Hash: a.Hash, Art: art})
if err != nil {
return err
}
if _, err := p.ev.JS().Publish(p.subject("artifact"), msg); err != nil {
return fmt.Errorf("publish %s (%d bytes): %w", a.Hash, len(msg), err)
}
published++
if unfolded += len(msg); unfolded >= flushBytes {
if err := flush(); err != nil {
return err
}
unfolded = 0
}
return nil
}
next, err := p.cli.FramesEach(cur, each)
if errors.Is(err, zitcli.ErrCursorInvalid) {
// Store rewritten: tell indexers to drop derived state, then
// republish the census from zero.
if _, err := p.ev.JS().Publish(p.subject("reset"), nil); err != nil {
return published, nil, err
}
next, err = p.cli.FramesEach(zitcli.Cursor{}, each)
}
if err != nil {
// The cursor is NOT saved on a failed round. Whatever was already
// published gets seen again next time, which idempotent folds
// absorb; saving it would skip the remainder for good.
return published, nil, err
}
return published, func() error { return saveCursor(kv, p.code, next) }, nil
}
// fold drains the durable consumer into the index, one transaction per
// fetched batch. At-least-once redelivery is safe: folds are idempotent
// upserts keyed by artifact hash.
func (p *Pipeline) fold() error {
if p.sub == nil {
sub, err := p.ev.JS().PullSubscribe(p.subject(">"), p.durable(),
nats.AckWait(ackWait))
if err != nil {
return err
}
p.sub = sub
}
branchy := false
for {
msgs, err := p.sub.Fetch(256, nats.MaxWait(50*time.Millisecond))
if errors.Is(err, nats.ErrTimeout) {
// An empty fetch is not proof of drained — delivery can lag
// the backlog. The consumer's pending count is authoritative.
info, ierr := p.sub.ConsumerInfo()
if ierr != nil {
return ierr
}
if info.NumPending > 0 || info.NumAckPending > 0 {
continue
}
break // drained
}
if err != nil {
return err
}
if p.lists == nil {
p.lists = newFileLists(p.cli)
}
var parsed []index.Parsed
for _, m := range msgs {
if m.Subject == p.subject("reset") {
// Order matters: fold what preceded the reset, drop,
// then continue with what follows.
p.lists.annotate(parsed)
if _, err := p.ix.Fold(parsed); err != nil {
return err
}
parsed = nil
branchy = false
if err := p.ix.Reset(); err != nil {
return err
}
continue
}
var e Event
if err := json.Unmarshal(m.Data, &e); err != nil {
return fmt.Errorf("ingest: bad event: %w", err)
}
parsed = append(parsed, index.Parsed{Hash: e.Hash, Art: e.Art})
}
p.lists.annotate(parsed)
b, err := p.ix.Fold(parsed)
if err != nil {
return err
}
branchy = branchy || b
for _, m := range msgs {
if err := m.Ack(); err != nil {
return err
}
}
}
if branchy {
return p.ix.RecomputeBranches()
}
return nil
}
func (p *Pipeline) subject(kind string) string {
return "zit.evt." + p.code + "." + kind
}
func loadCursor(kv nats.KeyValue, code string) zitcli.Cursor {
entry, err := kv.Get(code)
if err != nil {
return zitcli.Cursor{} // advisory: absent means bootstrap
}
var c zitcli.Cursor
if json.Unmarshal(entry.Value(), &c) != nil {
return zitcli.Cursor{}
}
return c
}
// crosslinkDate renders epoch milliseconds in the canonical D-card
// millisecond form, matching what fold() stores from artifacts.
func crosslinkDate(ms int64) string {
return time.UnixMilli(ms).UTC().Format("2006-01-02T15:04:05.000")
}
func saveCursor(kv nats.KeyValue, code string, c zitcli.Cursor) error {
raw, err := json.Marshal(c)
if err != nil {
return err
}
_, err = kv.Put(code, raw)
return err
}