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"`
}
func (p *Pipeline) markKey() string { return "seen." + p.durable() }
// loadMark reads this node's mark once per process. An absent mark is
// the zero mark, which defers to the ack floor alone.
func (p *Pipeline) loadMark(kv nats.KeyValue) (windowMark, error) {
if p.mark != nil {
return *p.mark, nil
}
var m windowMark
entry, err := kv.Get(p.markKey())
switch {
case errors.Is(err, nats.ErrKeyNotFound):
case err != nil:
return m, fmt.Errorf("ingest: %s: read window mark: %w", p.code, err)
default:
if json.Unmarshal(entry.Value(), &m) != nil {
m = windowMark{}
}
}
p.mark = &m
return m, 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 == 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 reports 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() error {
stream, err := p.ev.JS().StreamInfo(events.StreamName)
if err != nil {
return err
}
consumer, err := p.sub.ConsumerInfo()
if err != nil {
return err
}
kv, err := p.cursorKV()
if err != nil {
return err
}
mark, err := p.loadMark(kv)
if err != nil {
return err
}
seen := consumer.AckFloor.Stream
if mark.Stream.Equal(stream.Created) && mark.Seq > seen {
seen = mark.Seq
}
if stream.State.FirstSeq <= seen+1 {
return 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 fmt.Errorf("%w: up to %d events aged out", errBehindWindow, aged)
}