Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
9.3 kB · 228 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229package atproto
import ( "context" "sync/atomic" "time"
"stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/model")
// cursorFlushInterval bounds how often a relay's progress is written to the// index DB. The firehose is high-volume, so we persist on a timer rather than// per-event; between flushes the latest seq lives in memory and already covers// in-process reconnects.const cursorFlushInterval = 5 * time.Second
// highWater is a concurrency-safe monotonically-increasing maximum.type highWater struct{ v atomic.Int64 }
// observe raises the mark to seq if it is higher. Safe for concurrent callers.func (h *highWater) observe(seq int64) { for { cur := h.v.Load() if seq <= cur { return } if h.v.CompareAndSwap(cur, seq) { return } }}
func (h *highWater) get() int64 { return h.v.Load() }
// set overwrites the mark unconditionally, including downward. Only for// initialization and reset; everything else goes through observe.func (h *highWater) set(seq int64) { h.v.Store(seq) }
// relayCursor tracks how far we've consumed one relay's firehose so we can// resume after a disconnect or restart instead of re-tailing from live (which// would leave a gap). It keeps the high-water sequence number in memory,// updated on every frame, and persists it periodically and once on shutdown.//// Because the parallel scheduler may surface frames slightly out of sequence// order, the persisted cursor is the highest seq observed — on an unclean crash// a handful of in-flight frames just below it can be skipped on resume. That is// safe here: downstream handlers are idempotent, and with several relays plus a// cold deduper after restart, those commits get re-delivered and re-indexed.//// Alongside the sequence it tracks the time of the newest event seen, which is// what makes the cursor's age knowable and so gives [relayCursor.dropIfStale]// something to judge. A sequence number on its own says nothing about how much// history resuming from it would ask the relay to re-send.type relayCursor struct { host string model model.Model
// latest is the high-water cursor: the upstream at-sequence for a WebSocket // relay, or the high-water MoQ group sequence for a moqt:// relay (used to // resume replay via SubscribeFrom — see connectRelayMoq). A host is one // transport or the other, so a single value covers both — but the two // sequence spaces must never mix: upstream at-seqs are orders of magnitude // larger than group sequences, so on a moq host only observeGroup may feed // this (repoStreamCallbacks keeps the upstream seq out of it and in the // metrics gauge only). 0 = nothing seen yet (tail from live). Persisted // periodically so a Streamplace restart resumes from here — the relay // assigns durable ids across its own restarts, so a stored cursor stays // valid (it just ages out of the relay's replay window if we are down too // long, which is the gap PDS re-sync covers). latest highWater // lastEvent is the unix-seconds stamp of the newest event seen on this // relay, high-watered the same way and for the same reason: event times // jitter slightly out of order across a relay's own workers, and the newest // one is the only one that says anything about how current we are. Unlike // latest it is transport-independent — every relay stamps its events with a // wall-clock time — so it is fed on both the WebSocket and moq paths. lastEvent highWater
flushed int64 // last persisted cursor; only the flush loop touches it flushedEvent int64 // last persisted event time; likewise}
func (atsync *ATProtoSynchronizer) newRelayCursor(ctx context.Context, host string) *relayCursor { rc := &relayCursor{host: host, model: atsync.Model} stored, err := atsync.Model.GetRelayCursor(host) if err != nil { log.Error(ctx, "failed to load relay cursor; tailing from live", "err", err) return rc } if stored != nil { rc.latest.set(stored.Cursor) rc.flushed = stored.Cursor rc.lastEvent.set(stored.LastEventTime) rc.flushedEvent = stored.LastEventTime log.Log(ctx, "resuming relay from stored cursor", "cursor", stored.Cursor, "lastEventTime", stored.LastEventTime) } return rc}
// observe advances the high-water mark. Safe for concurrent callers (the// scheduler runs several event workers).func (rc *relayCursor) observe(seq int64) { rc.latest.observe(seq)}
// observeTime records the time stamped on an event we just received, which is// what [relayCursor.dropIfStale] later measures the cursor's age against. Safe// for concurrent callers; max-wins, so a slightly out-of-order stamp cannot// walk the mark backwards.func (rc *relayCursor) observeTime(t time.Time) { rc.lastEvent.observe(t.Unix())}
// lastEventTime is the unix-seconds stamp of the newest event seen so far, or 0// if we have never seen one (and never loaded one from the index).func (rc *relayCursor) lastEventTime() int64 { return rc.lastEvent.get() }
// stale reports whether resuming from this cursor would ask the relay to replay// more than window of history.//// A zero cursor is never stale: there is nothing to replay from, so the connect// tails live regardless. A nonzero cursor with NO event time is always stale —// that is a row written before this column existed, and it is exactly the state// that melted production: unknown age, unbounded replay. A non-positive window// disables the whole check.func (rc *relayCursor) stale(window time.Duration, now time.Time) bool { if window <= 0 { return false } if rc.latest.get() == 0 { return false } last := rc.lastEvent.get() if last == 0 { return true } return now.Sub(time.Unix(last, 0)) > window}
// dropIfStale abandons the cursor when its last observed event is older than// window, so the connect tails live instead of replaying a backlog the sweep// can heal more cheaply. Returns true if it dropped.//// Called at the top of every connect attempt rather than once at load, because// a relay also goes stale mid-process: a long disconnect-and-backoff stretch// leaves a cursor that was fine when we loaded it pointing hours into the past// by the time we get back in.func (rc *relayCursor) dropIfStale(ctx context.Context, window time.Duration) bool { if !rc.stale(window, time.Now()) { return false } seq := rc.latest.get() age := "unknown" if last := rc.lastEvent.get(); last != 0 { age = time.Since(time.Unix(last, 0)).String() } log.Warn(ctx, "abandoning stale relay cursor; tailing live and leaving the gap to the sweep", "relay", rc.host, "cursor", seq, "lastEventAge", age, "replayWindow", window) rc.reset() return true}
// param returns the cursor to dial with and whether to send one at all. With no// progress yet we send none, so a fresh external relay tails from live instead// of backfilling its entire history.func (rc *relayCursor) param() (int64, bool) { v := rc.latest.get() return v, v > 0}
// highSeq returns the high-water upstream sequence number observed so far.func (rc *relayCursor) highSeq() int64 { return rc.latest.get() }
// observeGroup advances the high-water cursor from a MoQ group sequence (the// moqt:// transport's flavour of a cursor). Called on every frame received from// a moqt:// relay (concurrency-safe).func (rc *relayCursor) observeGroup(seq uint64) { rc.observe(int64(seq))}
// groupStart returns the MoQ group to resume replay from and whether to request// replay at all. Before any frame is seen we tail the live edge; after a// reconnect we resume from the last group seen so the relay replays from there// (already-seen frames in that group are deduped downstream). Group 0 (a// brand-new relay's very first group) reads as "none" and tails live — harmless// and self-healing, and in practice relay group ids are large (seeded for// durability across restarts).func (rc *relayCursor) groupStart() (uint64, bool) { v := rc.latest.get() if v <= 0 { return 0, false } return uint64(v), true}
// reset abandons the cursor so the next connect tails the live edge. For when// a resume proves the stored value can't be trusted — e.g. a row poisoned by// upstream at-seqs before the group/at-seq split in repoStreamCallbacks, or one// so old that replaying it would cost more than the sweep — which observe()'s// max-wins semantics could otherwise never walk back. The flush loop persists// the zeroes, healing the stored row too.//// The event time goes with it: keeping a fresh timestamp next to a zero cursor// would describe a relay we are current with, which is the opposite of what a// reset means.func (rc *relayCursor) reset() { rc.latest.set(0) rc.lastEvent.set(0)}
// flush persists the high-water marks if either has advanced since the last// write. Only ever called from the single flush goroutine, so the flushed// values are unsynchronized.func (rc *relayCursor) flush(ctx context.Context) { v := rc.latest.get() t := rc.lastEvent.get() if v == rc.flushed && t == rc.flushedEvent { return } if err := rc.model.UpsertRelayCursor(rc.host, v, t); err != nil { log.Error(ctx, "failed to persist relay cursor", "err", err, "cursor", v, "lastEventTime", t) return } rc.flushed = v rc.flushedEvent = t}