// 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
// 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
// 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
// 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++
return
}
p.stats.LastOK = time.Now()
p.stats.LastCount = n
p.stats.Total += n
p.stats.LastErr = ""
p.stats.Failures = 0
}
// fallback counts one artifact that would not parse.
func (p *Pipeline) fallback() {
p.statMu.Lock()
defer p.statMu.Unlock()
p.stats.Fallbacks++
}
// 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) {
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
}
return p.round()
}
// 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) {
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 — the
// repairable kind. 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 {
if _, err := p.cli.ArtifactRaw(context.Background(), h, 1<<20); err == nil {
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()
start := time.Now()
n, err := p.watch(p.fold)
if err == nil {
err = p.fold()
}
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).
func (p *Pipeline) watch(flush func() error) (int, error) {
kv, err := p.ev.Cursors()
if err != nil {
return 0, 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, 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, err
}
if err := saveCursor(kv, p.code, next); err != nil {
return published, err
}
return published, 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())
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
}
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.
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})
}
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
}
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
}