package ingest
import (
"encoding/json"
"errors"
"fmt"
"log"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/events"
)
// errBehindWindow is fold's answer when the event stream no longer holds
// everything after this node's position. round answers it with a
// rebuild; nothing on the stream can close the gap.
var errBehindWindow = errors.New("ingest: behind the event stream's window")
// windowMark is the stream position this node's index is known to be
// folded through, kept in the cursor bucket so it survives a restart.
//
// The durable consumer's ack floor is not enough on its own. The stream
// is shared by every repository, and a consumer's ack floor moves only
// when its OWN messages are acked, so a repository with nothing new sits
// behind the stream's first sequence while the rest of the fleet
// publishes. The mark moves on every drained fold, whatever the other
// repositories are doing.
type windowMark struct {
Seq uint64 `json:"seq"`
// Stream is the stream's creation time. Sequences restart when the
// stream is recreated, and a mark from the old one would hide a real
// gap in the new one until the numbers caught up.
Stream time.Time `json:"stream"`
// Rebuilt is the stream position the last rebuild covered. A reset
// at or below it was published before that rebuild read the store,
// so the rebuild already reflects it, and folding it would clear the
// fresh index with nothing behind it to fill it again.
Rebuilt uint64 `json:"rebuilt,omitempty"`
}
func (p *Pipeline) markKey() string { return "seen." + p.durable() }
// loadMark reads this node's mark once per process and reports whether
// one exists.
func (p *Pipeline) loadMark(kv nats.KeyValue) (windowMark, bool, error) {
if p.mark != nil {
return *p.mark, true, nil
}
var m windowMark
entry, err := kv.Get(p.markKey())
switch {
case errors.Is(err, nats.ErrKeyNotFound):
return m, false, nil
case err != nil:
return m, false, fmt.Errorf("ingest: %s: read window mark: %w", p.code, err)
}
if json.Unmarshal(entry.Value(), &m) != nil {
m = windowMark{}
}
p.mark = &m
return m, true, nil
}
// markFolded records that everything on the stream up to its last
// sequence has been folded. The mark is advisory: a failed write is
// logged, and costs at worst one unnecessary rebuild later.
func (p *Pipeline) markFolded(kv nats.KeyValue, stream *nats.StreamInfo) {
m := windowMark{Seq: stream.State.LastSeq, Stream: stream.Created}
if p.mark != nil && p.mark.Stream.Equal(stream.Created) {
m.Rebuilt = p.mark.Rebuilt
}
p.putMark(kv, m)
}
// markRebuilt records that a rebuild covered everything on the stream up
// to its last sequence, resets included.
func (p *Pipeline) markRebuilt(kv nats.KeyValue, stream *nats.StreamInfo) {
p.putMark(kv, windowMark{Seq: stream.State.LastSeq, Stream: stream.Created, Rebuilt: stream.State.LastSeq})
}
func (p *Pipeline) putMark(kv nats.KeyValue, m windowMark) {
if p.mark != nil && *p.mark == m {
return
}
raw, err := json.Marshal(m)
if err != nil {
return
}
if _, err := kv.Put(p.markKey(), raw); err != nil {
log.Printf("ingest: %s: save window mark: %v", p.code, err)
return
}
p.mark = &m
}
// checkWindow returns the stream as it stands, or errBehindWindow when
// the stream's oldest message is past the next one this node expected,
// which means events it never folded have aged out. events/gap_test.go shows why this must be asked:
// the fetch after a gap succeeds and returns newer messages, so a fold
// that only drains carries on with a permanent hole.
func (p *Pipeline) checkWindow() (*nats.StreamInfo, error) {
stream, err := p.ev.JS().StreamInfo(events.StreamName)
if err != nil {
return nil, err
}
consumer, err := p.sub.ConsumerInfo()
if err != nil {
return nil, err
}
kv, err := p.cursorKV()
if err != nil {
return nil, err
}
mark, found, err := p.loadMark(kv)
if err != nil {
return nil, err
}
if !found {
// A missing mark is the first run after this check shipped, or a
// fresh bucket — not evidence of a gap. On a busy shared stream
// every idle store's ack floor is behind the first sequence, so
// reading its absence as a gap would rebuild the whole fleet at
// once. Seed it from the stream as it stands; only a node that
// later falls behind a mark it has is rebuilt.
p.markFolded(kv, stream)
return stream, nil
}
seen := consumer.AckFloor.Stream
// A mark from an earlier incarnation of the stream is ignored on
// purpose, not re-seeded: its sequences number a different stream,
// and events the old one held may never have been folded, so it
// vouches for nothing. The seed rule is for a mark that was never
// written, which says nothing either way; a stale one must not be
// quietly replaced as if it were fine. The ack floor alone decides
// here. Recreating the stream deletes its consumers, and a new one
// starts just below the first sequence, so this comparison does not
// itself rebuild. The fleet-wide rebuild a recreation needs comes
// from events.Start deleting the cursor bucket along with the
// stream: every store then finds no cursor and rebuilds.
if mark.Stream.Equal(stream.Created) && mark.Seq > seen {
seen = mark.Seq
}
if stream.State.FirstSeq <= seen+1 {
return stream, nil
}
aged := stream.State.FirstSeq - seen - 1
p.statMu.Lock()
p.stats.WindowGaps++
p.statMu.Unlock()
log.Printf("ingest: %s: behind the event stream's window — folded through %d, the stream now starts at %d, so up to %d events aged out unread; rebuilding",
p.code, seen, stream.State.FirstSeq, aged)
return nil, fmt.Errorf("%w: up to %d events aged out", errBehindWindow, aged)
}