// 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"
"sync/atomic"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/artifact"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/evt"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/obs"
"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
// WindowGaps counts the times this node found the event stream had
// moved past everything it had folded — it was away longer than the
// stream's window, and events it never saw had aged out. Each one
// forces a rebuild, so a rising count is a node that keeps falling
// behind, not an index that is wrong.
WindowGaps 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
// cursors opens the cursor bucket. Nil means the events server's own;
// tests set it to stand a NATS outage in front of the real bucket.
cursors func() (nats.KeyValue, error)
// mark caches this node's folded-through position (window.go), nil
// until first read from the cursor bucket.
mark *windowMark
// 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}
}
// cursorKV opens the bucket that holds the publisher cursor.
func (p *Pipeline) cursorKV() (nats.KeyValue, error) {
if p.cursors != nil {
return p.cursors()
}
return p.ev.Cursors()
}
// 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) {
kv, err := p.cursorKV()
if err != nil {
return 0, err
}
// The cursor goes BEFORE the index is cleared. A rebuild that stops
// part-way must leave a state the next round rebuilds from: no cursor
// means "rebuild", whatever the index holds. The other order let a
// NATS hiccup empty the index and then fail the delete, and later
// rounds carried on from the surviving cursor into an empty index,
// never republishing what it had lost.
if err := kv.Delete(p.code); err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
return 0, err
}
if err := p.ix.Reset(); err != nil {
return 0, err
}
// Everything on the stream NOW was published from the store, so the
// census read below covers it. Taken before the read, the snapshot
// can only understate what the rebuild holds.
covered, coveredErr := p.ev.JS().StreamInfo(events.StreamName)
batchSize := reindexBatch
n := 0
start := time.Now()
// Every return from here on must put back what is suspended here,
// failure and panic included: left dropped, the store has no search
// and no secondary indexes until some later rebuild succeeds, and the
// stall backoff can put that many minutes off. Registered before the
// suspend, because a suspend that fails part-way has dropped some.
resumed := false
defer func() {
if resumed {
return
}
if err := p.ix.ResumeDerivedMaintenance(); err != nil {
log.Printf("ingest: %s: restoring search triggers and indexes after a failed rebuild: %v", p.code, err)
}
}()
if err := p.ix.SuspendDerivedMaintenance(); 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
// Content probes for this round: files offered to the parser that
// were never artifacts. Counted rather than logged one by one, and
// said once at the end where the number is a fact about the store
// rather than a wall of complaints about ordinary files.
var probes atomic.Int64
// Every return below this point must stop the census reader: a
// failed fold leaves nobody draining framesBatches, and a reader
// blocked on it keeps its `zit frames` child running for good.
// Cancelling makes the reader's callback fail, which kills and reaps
// the child; waiting means the rebuild does not return until it has.
readCtx, stopReader := context.WithCancel(context.Background())
var reader sync.WaitGroup
defer func() {
stopReader()
reader.Wait()
}()
send := func(batch []index.Parsed) error {
select {
case framesBatches <- batch:
return nil
case <-readCtx.Done():
return readCtx.Err()
}
}
reader.Add(1)
go func() {
defer reader.Done()
defer close(framesBatches)
pending := make([]index.Parsed, 0, batchSize)
framesCursor, framesErr = p.cli.FramesEachStructural(zitcli.Cursor{}, func(a zitcli.Artifact) error {
if err := readCtx.Err(); err != nil {
return err
}
if isM, ok := artifact.QuickIsManifest(a.Bytes); ok && isM {
return nil
}
art, perr := artifact.Parse(a.Bytes)
if perr != nil {
p.fallback()
// Probing is how quarry decides what a blob IS, so a
// refusal is the normal answer for most files and
// logging every one buries the failures that matter.
// Only an artifact that framed and carries its Z card
// was MEANT to parse; the rest are content, counted
// and reported once for the whole round.
if artifact.PlausiblyStructural(a.Bytes) {
log.Printf("ingest: %s: treating as content: %v", a.Hash, perr)
} else {
probes.Add(1)
}
}
if art == nil || art.Type == artifact.Manifest {
return nil
}
pending = append(pending, index.Parsed{Hash: a.Hash, Art: art})
if len(pending) >= batchSize {
if err := send(pending); err != nil {
return err
}
pending = make([]index.Parsed, 0, batchSize)
}
return nil
})
if framesErr == nil && len(pending) > 0 {
_ = send(pending) // fails only when the rebuild has already given up
}
}()
// 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.
if p.lists == nil {
p.lists = newFileLists(p.cli)
}
for pb := range framesBatches {
// A rebuild must learn the same things the live fold learns:
// an A card whose blob is a stub is an uploaded file, not a
// text file that happens to say so.
p.lists.annotateStubs(pb)
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.ResumeDerivedMaintenance(); err != nil {
return n, err
}
resumed = true
if err := p.ix.FinishBulk(); err != nil {
return n, err
}
if err := p.ix.RebuildSearch(); 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
}
// The durable consumer's ack floor is still wherever it was before
// the rebuild; without this mark the next fold would find the same
// gap and rebuild again, forever.
if coveredErr != nil {
log.Printf("ingest: %s: stream info for the window mark: %v", p.code, coveredErr)
} else {
p.markRebuilt(kv, covered)
}
probed := ""
if v := probes.Load(); v > 0 {
probed = fmt.Sprintf(", %d stored as content", v)
}
log.Printf("ingest: %s: rebuilt %d records in %.1fs (crosslink %.1fs, frames %.1fs, finish %.1fs)%s",
p.code, n, time.Since(start).Seconds(), crosslinkDone.Sub(start).Seconds(),
streamDone.Sub(crosslinkDone).Seconds(), time.Since(changesDone).Seconds(), probed)
return n, nil
}
// reindexBatch is how many records a rebuild folds per transaction. A
// var so tests can make the census reader outrun its channel.
var reindexBatch = 512
// 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(obs.WithReason(context.Background(), obs.ReasonIngest), 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.
//
// Only a cursor that is genuinely ABSENT means that. Any other
// failure to read it is NATS being briefly unreachable, and a rebuild
// started on that is a rebuild of a healthy index during an outage:
// the error goes to the round's backoff and nothing changes.
kv, err := p.cursorKV()
if err != nil {
p.record(0, 0, err)
return 0, err
}
if _, gerr := kv.Get(p.code); gerr != nil {
if !errors.Is(gerr, nats.ErrKeyNotFound) {
err := fmt.Errorf("ingest: %s: read cursor: %w", p.code, gerr)
p.record(0, 0, err)
return 0, err
}
return p.rebuildRound()
}
// 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(obs.WithReason(context.Background(), obs.ReasonIngest), "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.rebuildRound()
}
}
}
start := time.Now()
n, save, err := p.watch(p.fold)
if err == nil {
err = p.fold()
}
if errors.Is(err, errBehindWindow) {
// The publisher cursor is NOT saved: the rebuild sets it from the
// census it reads.
return p.rebuildRound()
}
// 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
}
// rebuildRound is a round whose work is a rebuild. Its outcome is
// recorded like any other round's, so a rebuild that keeps failing
// reaches the stall backoff instead of repeating on every tick.
func (p *Pipeline) rebuildRound() (int, error) {
start := time.Now()
n, err := p.reindexLocked()
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
// offered parses one artifact the census read, nil for content. Every
// byte that enters the store — a push, a mirror pull, an upload, a
// rebuild's replay — passes through here, and content is never folded,
// so this is where the index hears that a blob attachment.head
// remembered as missing may now be present.
func (p *Pipeline) offered(a zitcli.Artifact) *artifact.Artifact {
art, perr := artifact.Parse(a.Bytes)
if perr != nil {
p.fallback()
// Same rule as the fold: only an artifact that framed and
// carries its Z card was meant to parse. The counter is
// p.stats.Fallbacks, which the fleet page already shows.
if artifact.PlausiblyStructural(a.Bytes) {
log.Printf("ingest: %s: treating as content: %v", a.Hash, perr)
}
}
if art == nil && p.ix != nil {
p.ix.ForgetAbsent()
}
return art
}
// 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.cursorKV()
if err != nil {
return 0, nil, err
}
cur, err := loadCursor(kv, p.code)
if err != nil {
return 0, nil, err
}
// 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 := p.offered(a)
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.
//
// Recreating the stream deletes its consumers, and a handle to the
// deleted one fails every call until the process restarts. So a
// consumer found gone is dropped and made again on the new stream, once
// per fold. Whatever the old stream held that was never folded is not
// lost: the cursor bucket goes with the stream, and the round that finds
// no cursor rebuilds from the store.
func (p *Pipeline) fold() error {
err := p.drain()
if !errors.Is(err, nats.ErrConsumerNotFound) && !errors.Is(err, nats.ErrConsumerDeleted) {
return err
}
log.Printf("ingest: %s: consumer %s is gone (the event stream was recreated); subscribing again", p.code, p.durable())
_ = p.sub.Unsubscribe()
p.sub = nil
return p.drain()
}
func (p *Pipeline) drain() 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
}
stream, err := p.checkWindow()
if err != nil {
return err
}
// Resets at or below this position were published before the last
// rebuild read the store; that rebuild already reflects them.
var staleResets uint64
if p.mark != nil && p.mark.Stream.Equal(stream.Created) {
staleResets = p.mark.Rebuilt
}
branchy := false
// Check-ins new to this index in this drain. branch.moved cannot be
// raised inline with them: the branch is computed, not carried, and
// the answer only exists after RecomputeBranches below.
var freshCheckins []string
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.
//
// The stream is read FIRST: once the consumer then reports
// nothing pending, every message of ours up to that stream
// position has been folded, which is what the window mark
// records.
stream, serr := p.ev.JS().StreamInfo(events.StreamName)
if serr != nil {
return serr
}
info, ierr := p.sub.ConsumerInfo()
if ierr != nil {
return ierr
}
if info.NumPending > 0 || info.NumAckPending > 0 {
continue
}
if kv, kerr := p.cursorKV(); kerr == nil {
p.markFolded(kv, stream)
}
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") {
if meta, merr := m.Metadata(); merr == nil && meta.Sequence.Stream <= staleResets {
continue // acked below with the batch
}
// Order matters: fold what preceded the reset, drop,
// then continue with what follows.
p.lists.annotate(parsed)
if _, err := p.foldAndRaise(parsed, &freshCheckins); 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.foldAndRaise(parsed, &freshCheckins)
if err != nil {
return err
}
branchy = branchy || b
for _, m := range msgs {
if err := m.Ack(); err != nil {
return err
}
}
}
if branchy {
if err := p.ix.RecomputeBranches(); err != nil {
return err
}
}
p.raiseBranchMoved(freshCheckins)
return nil
}
// raiseBranchMoved fires one branch.moved per new check-in, after the
// branches are computed.
//
// Per check-in, not per push, and not deduplicated per branch. RFC 0005
// §2.3.2 settled this: dagnats triggers carry a debounce, so a pipeline
// that wants one build per ten-commit push configures one, and a
// pipeline that wants ten gets ten. Coalescing here would take the
// choice away from both.
func (p *Pipeline) raiseBranchMoved(hashes []string) {
if p.ev == nil || len(hashes) == 0 {
return
}
branches, err := p.ix.BranchOf(hashes)
if err != nil {
log.Printf("ingest: %s: branch lookup: %v", p.code, err)
return
}
for _, h := range hashes {
b, ok := branches[h]
if !ok {
continue
}
p.raise(raised{evt.BranchMoved, map[string]any{"branch": b, "hash": h}})
}
}
// foldAndRaise folds a batch and then raises RFC 0005's derived events
// for what was actually new.
//
// Two orderings matter and neither is arbitrary. Newness is asked BEFORE
// the fold, because the fold is what makes an artifact seen — asking
// afterwards would find everything already known and raise nothing. The
// events go out AFTER the commit, because an event says a thing is now
// true and a subscriber that reacts by querying must find it; raising
// first would offer a window where the forge says one thing and answers
// another.
//
// Only the DERIVED events come from here (§2.1). The operational ones —
// and the ticket, wiki and forum events that have an op behind them —
// are raised at that op, which knows more than the artifact does: an
// artifact records a ticket's new status, but only the op knows the old
// one, and without that there is no telling ticket.closed from
// ticket.changed. An artifact arriving from a clone's push has no op,
// and so today raises nothing beyond the check-in it rides on; that gap
// is real and named in the RFC rather than papered over here.
//
// A raise that fails is logged, never returned: the index is the record
// and it is already committed. Failing the fold over an undelivered
// notification would turn a missed reaction into a stalled store.
func (p *Pipeline) foldAndRaise(parsed []index.Parsed, freshCheckins *[]string) (bool, error) {
var (
fresh map[string]bool
prior index.Prior
)
if p.ev != nil && len(parsed) > 0 {
hashes := make([]string, 0, len(parsed))
for _, x := range parsed {
hashes = append(hashes, x.Hash)
}
var err error
if fresh, err = p.ix.Unseen(hashes); err != nil {
return false, err
}
if prior, err = p.ix.Prior(parsed); err != nil {
return false, err
}
}
branchy, err := p.ix.Fold(parsed)
if err != nil {
return branchy, err
}
for _, x := range parsed {
if !fresh[x.Hash] {
continue
}
if x.Art != nil && x.Art.Type == artifact.Manifest {
*freshCheckins = append(*freshCheckins, x.Hash)
}
for _, r := range derive(x.Hash, x.Art, prior) {
p.raise(r)
}
}
p.raiseTicketEvents(parsed, fresh, prior)
p.raiseLandingVerdicts(parsed, fresh, prior)
return branchy, nil
}
// raiseLandingVerdicts names a landing's settling by what the fold
// computed, not by what a reply claimed.
//
// A post can say `zit-landing-state: accepted` and be wrong — it might
// be from someone without the capability, or superseded by a later
// reply. LandingTransition folds the whole thread into one state, so
// comparing that state before and after is the only reading that cannot
// be talked into a lie.
func (p *Pipeline) raiseLandingVerdicts(parsed []index.Parsed, fresh map[string]bool, prior index.Prior) {
if p.ev == nil {
return
}
touched := map[string]index.Parsed{}
for _, x := range parsed {
if x.Art == nil || x.Art.Type != artifact.Forum || !fresh[x.Hash] {
continue
}
root := x.Art.ThreadRoot
if root == "" || !prior.LandingThreads[root] {
continue
}
if cur, ok := touched[root]; !ok || x.Art.Date > cur.Art.Date {
touched[root] = x
}
}
if len(touched) == 0 {
return
}
ids := make([]string, 0, len(touched))
for id := range touched {
ids = append(ids, id)
}
after, err := p.ix.LandingStateOf(ids)
if err != nil {
log.Printf("ingest: %s: landing state: %v", p.code, err)
return
}
for id, x := range touched {
before, now := prior.LandingState[id], after[id]
if before == now {
continue // commented, not settled — already raised
}
kind, ok := landingVerdict(now)
if !ok {
continue
}
p.raise(raised{kind, map[string]any{
"landing": id, "hash": x.Hash, "actor": x.Art.User,
"date": x.Art.Date, "state": now,
}})
}
}
// landingVerdict maps a folded landing state to its event. An unknown
// state raises nothing rather than something approximate.
func landingVerdict(state string) (evt.Kind, bool) {
switch state {
case "accepted":
return evt.LandingAccepted, true
case "rejected":
return evt.LandingRejected, true
default:
return "", false
}
}
// raiseTicketEvents names one transition per ticket a batch touched.
//
// Per ticket rather than per artifact: a batch may carry three changes
// to one ticket, and a subscriber told "changed, changed, closed" learns
// nothing the single "closed" did not say. The transition of the FOLDED
// state is the true one, and it is what the forge op reported too.
func (p *Pipeline) raiseTicketEvents(parsed []index.Parsed, fresh map[string]bool, prior index.Prior) {
if p.ev == nil {
return
}
// Newest change per ticket carries the payload; the status comes
// from the folded state, not from the change.
latest := map[string]index.Parsed{}
for _, x := range parsed {
if x.Art == nil || x.Art.Type != artifact.Ticket || x.Art.TicketID == "" {
continue
}
if !fresh[x.Hash] {
continue
}
if cur, ok := latest[x.Art.TicketID]; !ok || x.Art.Date > cur.Art.Date {
latest[x.Art.TicketID] = x
}
}
if len(latest) == 0 {
return
}
ids := make([]string, 0, len(latest))
for id := range latest {
ids = append(ids, id)
}
after, err := p.ix.TicketStatus(ids)
if err != nil {
log.Printf("ingest: %s: ticket status: %v", p.code, err)
return
}
for id, x := range latest {
before, seen := prior.TicketStatus[id]
payload := map[string]any{
"hash": x.Hash, "actor": x.Art.User, "date": x.Art.Date,
"summary": firstLine(x.Art.Comment),
}
for _, r := range ticketEvents(id, before, after[id], seen, payload) {
p.raise(r)
}
}
}
// raise publishes one derived event, logging what it cannot deliver. A
// failed raise never fails the fold: the index is the record and it is
// already committed, so failing here would turn a missed reaction into a
// stalled store.
func (p *Pipeline) raise(r raised) {
if err := p.ev.Raise(r.Kind, p.code, r.Payload); err != nil {
log.Printf("ingest: %s: raise %s: %v", p.code, r.Kind, err)
}
}
func (p *Pipeline) subject(kind string) string {
return "zit.evt." + p.code + "." + kind
}
// loadCursor reads the publisher cursor. Absent means bootstrap and
// reads as the zero cursor; a failed read is returned, because treating
// it as absent would republish the whole census over one NATS hiccup.
func loadCursor(kv nats.KeyValue, code string) (zitcli.Cursor, error) {
entry, err := kv.Get(code)
if errors.Is(err, nats.ErrKeyNotFound) {
return zitcli.Cursor{}, nil
}
if err != nil {
return zitcli.Cursor{}, fmt.Errorf("ingest: %s: read cursor: %w", code, err)
}
var c zitcli.Cursor
if json.Unmarshal(entry.Value(), &c) != nil {
return zitcli.Cursor{}, nil
}
return c, nil
}
// 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
}