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() }
// published is the last stream position at which this store's own
// subject was published by a round that went on to save the publisher
// cursor, on any node. It is what the window-gap rule asks when the
// stream's front has passed a store's folded position. The stream, shared
// by every store, cannot say whose events it dropped: a message aged out
// leaves no trace, per subject or otherwise. A
// publish whose round never saved the cursor needs no record: the next
// round reads the store from the unsaved cursor and publishes it again.
type published struct {
Seq uint64 `json:"seq"`
Stream time.Time `json:"stream"`
}
func (p *Pipeline) pubKey() string { return "pub." + p.code }
// recordPublished raises the store's publish record to seq. It runs
// before the publisher cursor is saved, so a saved cursor always has its
// publishes on record; a failure fails the round, which then republishes.
func (p *Pipeline) recordPublished(kv nats.KeyValue, seq uint64) error {
stream, err := p.ev.JS().StreamInfo(events.StreamName)
if err != nil {
return err
}
return putPublished(kv, p.pubKey(), published{Seq: seq, Stream: stream.Created}, true)
}
// putPublished writes rec unless the record already covers it. Within
// one stream the record only rises (raise) or is only seeded (not
// raise); a record from an earlier stream is always replaced. Nodes
// sharing the bucket race on it, so the write is compare-and-set.
func putPublished(kv nats.KeyValue, key string, rec published, raise bool) error {
raw, err := json.Marshal(rec)
if err != nil {
return err
}
for try := 0; ; try++ {
entry, err := kv.Get(key)
switch {
case errors.Is(err, nats.ErrKeyNotFound):
_, err = kv.Create(key, raw)
case err != nil:
return err
default:
var old published
if json.Unmarshal(entry.Value(), &old) == nil && old.Stream.Equal(rec.Stream) && (!raise || old.Seq >= rec.Seq) {
return nil
}
_, err = kv.Update(key, raw, entry.Revision())
}
if err == nil || try == 4 {
return err
}
}
}
// ownEventsKept reports whether none of this store's events can be among
// those the stream dropped below its first sequence: everything it ever
// published and committed on this stream sits at or before seen, which
// this node has folded. A missing or unreadable record, or one from an
// earlier stream, cannot vouch for that, and the answer is no.
func (p *Pipeline) ownEventsKept(kv nats.KeyValue, stream *nats.StreamInfo, seen uint64) bool {
entry, err := kv.Get(p.pubKey())
if err != nil {
return false
}
var rec published
if json.Unmarshal(entry.Value(), &rec) != nil || !rec.Stream.Equal(stream.Created) {
return false
}
return rec.Seq <= seen
}
// 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)
}
// markDrained is markFolded for a fold that emptied this store's
// consumer, and it seeds a missing publish record: with nothing of ours
// pending, every event of ours up to the stream's last sequence has been
// folded. A store idle since before the record existed gets it on its
// boot round, so its next restart is judged exactly.
func (p *Pipeline) markDrained(kv nats.KeyValue, stream *nats.StreamInfo) {
p.markFolded(kv, stream)
if err := putPublished(kv, p.pubKey(), published{Seq: stream.State.LastSeq, Stream: stream.Created}, false); err != nil {
log.Printf("ingest: %s: seed publish record: %v", p.code, err)
}
}
// markRebuilt records that a rebuild covered everything on the stream up
// to its last sequence, resets included.
//
// It also seeds the store's publish record: the rebuild read the store
// itself, so nothing published up to here can be missing from the index.
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})
if err := putPublished(kv, p.pubKey(), published{Seq: stream.State.LastSeq, Stream: stream.Created}, false); err != nil {
log.Printf("ingest: %s: seed publish record: %v", p.code, err)
}
}
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
}
// The stream is every store's, so its front moving past this store's
// position is usually other stores' events aging out: a quiet store
// sits still while the fleet publishes. Only the store's own events
// matter, and only its mark can vouch for having folded them.
if mark.Stream.Equal(stream.Created) && p.ownEventsKept(kv, stream, seen) {
p.putMark(kv, windowMark{Seq: stream.State.FirstSeq - 1, Stream: stream.Created, Rebuilt: mark.Rebuilt})
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)
}