Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
13 kB · 353 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354package atproto
import ( "context" "testing" "time"
indigoatproto "github.com/bluesky-social/indigo/api/atproto" lexutil "github.com/bluesky-social/indigo/lex/util" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/model")
func TestRelayCursorResume(t *testing.T) { mod, err := model.MakeDB(":memory:") require.NoError(t, err) atsync := &ATProtoSynchronizer{Model: mod} ctx := context.Background() const host = "wss://relay.example"
// A fresh relay has no stored cursor, so we dial with none and tail live // rather than backfilling the relay's whole history. rc := atsync.newRelayCursor(ctx, host) if _, ok := rc.param(); ok { t.Fatal("fresh relay should not send a cursor") }
// observe tracks the high-water mark and never regresses on out-of-order // frames (the parallel scheduler can surface them out of sequence). rc.observe(100) rc.observe(50) rc.observe(120) if v, ok := rc.param(); !ok || v != 120 { t.Fatalf("expected cursor 120, got %d (ok=%v)", v, ok) }
// flush persists the high-water mark to the index DB. rc.flush(ctx) stored, err := mod.GetRelayCursor(host) require.NoError(t, err) require.NotNil(t, stored) require.Equal(t, int64(120), stored.Cursor)
// A second flush with no advance is a no-op, and the stored value is stable. rc.observe(120) rc.flush(ctx) stored, err = mod.GetRelayCursor(host) require.NoError(t, err) require.Equal(t, int64(120), stored.Cursor)
// A new cursor (as if the process restarted) resumes from the stored value. resumed := atsync.newRelayCursor(ctx, host) v, ok := resumed.param() require.True(t, ok) require.Equal(t, int64(120), v)
// Cursors are independent per relay. other := atsync.newRelayCursor(ctx, "wss://other.example") if _, ok := other.param(); ok { t.Fatal("a different relay must not inherit another relay's cursor") }}
func TestRelayCursorGroupResume(t *testing.T) { mod, err := model.MakeDB(":memory:") require.NoError(t, err) atsync := &ATProtoSynchronizer{Model: mod} ctx := context.Background() const host = "moqt://relay.example"
// A fresh moqt:// relay has no stored group, so we tail the live edge rather // than requesting replay. rc := atsync.newRelayCursor(ctx, host) if _, ok := rc.groupStart(); ok { t.Fatal("fresh moq relay should not request replay") }
// observeGroup tracks the high-water group and never regresses on // out-of-order delivery. rc.observeGroup(500) rc.observeGroup(200) rc.observeGroup(640) g, ok := rc.groupStart() require.True(t, ok) require.Equal(t, uint64(640), g)
// flush persists the high-water group as the relay's cursor. rc.flush(ctx) stored, err := mod.GetRelayCursor(host) require.NoError(t, err) require.NotNil(t, stored) require.Equal(t, int64(640), stored.Cursor)
// As if the process restarted: a new cursor resumes replay from the stored // group (connectRelayMoq calls SubscribeFrom with it). resumed := atsync.newRelayCursor(ctx, host) g, ok = resumed.groupStart() require.True(t, ok) require.Equal(t, uint64(640), g)}
// Regression: connectRelayMoq shares repoStreamCallbacks with the WebSocket// path, and those callbacks observe the upstream at-sequence on every event.// On a moq relay that seq (billions) must stay out of the group cursor// (thousands): observe() is max-wins, so one poisoned observation would bury// the group cursor and the next SubscribeFrom would wait forever, silently, on// a group the relay will never produce.func TestMoqCallbacksDontPoisonGroupCursor(t *testing.T) { atsync := &ATProtoSynchronizer{ commitDedup: newFirehoseDeduper(dedupWindow), identityDedup: newFirehoseDeduper(dedupWindow), } commit := &indigoatproto.SyncSubscribeRepos_Commit{ Seq: 5_000_000_000, // an upstream at-seq, far above any group seq Repo: "did:plc:abc123", Commit: lexutil.LexLink(mustCID(t)), } // Pre-claim the commit in the deduper so the callback stops after its // cursor/metrics bookkeeping instead of spawning the indexing handler. atsync.commitDedup.seen(commit.Commit.String())
// moq relay: the group cursor must survive the callback untouched. moqCursor := &relayCursor{host: "moqt://relay.example"} moqCursor.observeGroup(640) rsc := atsync.repoStreamCallbacks(context.Background(), "moqt://relay.example", moqCursor, func() {}) require.NoError(t, rsc.RepoCommit(commit)) g, ok := moqCursor.groupStart() require.True(t, ok) require.Equal(t, uint64(640), g, "upstream at-seq must not bury the group cursor")
// websocket relay: the same callback is what advances the cursor. wsCursor := &relayCursor{host: "wss://relay.example"} rsc = atsync.repoStreamCallbacks(context.Background(), "wss://relay.example", wsCursor, func() {}) require.NoError(t, rsc.RepoCommit(commit)) v, ok := wsCursor.param() require.True(t, ok) require.Equal(t, int64(5_000_000_000), v)}
func TestRelayCursorReset(t *testing.T) { mod, err := model.MakeDB(":memory:") require.NoError(t, err) atsync := &ATProtoSynchronizer{Model: mod} ctx := context.Background() const host = "moqt://relay.example"
rc := atsync.newRelayCursor(ctx, host) rc.observeGroup(640) rc.flush(ctx)
// A cursor a resume proved untrustworthy is abandoned wholesale: reset // drops it below max-wins observe()'s reach, and the next flush heals the // stored row too. rc.reset() if _, ok := rc.groupStart(); ok { t.Fatal("reset cursor should tail live") } rc.flush(ctx) stored, err := mod.GetRelayCursor(host) require.NoError(t, err) require.NotNil(t, stored) require.Equal(t, int64(0), stored.Cursor)
// A restart after the reset flush also tails live. resumed := atsync.newRelayCursor(ctx, host) if _, ok := resumed.groupStart(); ok { t.Fatal("restarted cursor should tail live after reset") }}
// TestRelayCursorEventTime covers the second half of the cursor: the newest// event time is what makes the cursor's age knowable, so it has to high-water,// persist and reload alongside the sequence.func TestRelayCursorEventTime(t *testing.T) { mod, err := model.MakeDB(":memory:") require.NoError(t, err) atsync := &ATProtoSynchronizer{Model: mod} ctx := context.Background() const host = "wss://relay.example"
rc := atsync.newRelayCursor(ctx, host) require.Equal(t, int64(0), rc.lastEventTime(), "a fresh relay has seen no events")
// Max-wins, like the seq: relays stamp events from several workers, so a // slightly older stamp arriving late must not walk the mark backwards. base := time.Unix(1700000000, 0) rc.observeTime(base) rc.observeTime(base.Add(-30 * time.Second)) rc.observeTime(base.Add(90 * time.Second)) require.Equal(t, base.Add(90*time.Second).Unix(), rc.lastEventTime())
// flush writes both columns, and a flush with nothing new is a no-op. rc.observe(120) rc.flush(ctx) stored, err := mod.GetRelayCursor(host) require.NoError(t, err) require.NotNil(t, stored) require.Equal(t, int64(120), stored.Cursor) require.Equal(t, base.Add(90*time.Second).Unix(), stored.LastEventTime)
// The event time advancing on its own is enough to trigger a write -- a moq // relay whose group cursor is unchanged still gets its freshness recorded. rc.observeTime(base.Add(5 * time.Minute)) rc.flush(ctx) stored, err = mod.GetRelayCursor(host) require.NoError(t, err) require.Equal(t, int64(120), stored.Cursor) require.Equal(t, base.Add(5*time.Minute).Unix(), stored.LastEventTime)
// A restart reloads both, so the staleness check has something to judge. resumed := atsync.newRelayCursor(ctx, host) v, ok := resumed.param() require.True(t, ok) require.Equal(t, int64(120), v) require.Equal(t, base.Add(5*time.Minute).Unix(), resumed.lastEventTime())
// reset drops the event time with the cursor: a fresh timestamp next to a // zero cursor would claim we are current with a relay we just abandoned. resumed.reset() require.Equal(t, int64(0), resumed.lastEventTime()) _, ok = resumed.param() require.False(t, ok)}
// TestRelayCursorDropIfStale is the production incident in a unit test: a// cursor whose newest event is hours old must not be replayed from, because// asking a relay to re-send hours of the whole network is the thing that buried// the node. Fresh cursors are left alone, since replay is genuinely cheaper// across an ordinary restart.func TestRelayCursorDropIfStale(t *testing.T) { ctx := context.Background() const window = 15 * time.Minute
newCursor := func(seq int64, lastEvent time.Time) *relayCursor { rc := &relayCursor{host: "wss://relay.example"} rc.latest.set(seq) if !lastEvent.IsZero() { rc.observeTime(lastEvent) } return rc }
// Inside the window: keep the cursor and replay from it. rc := newCursor(120, time.Now().Add(-time.Minute)) require.False(t, rc.dropIfStale(ctx, window)) v, ok := rc.param() require.True(t, ok) require.Equal(t, int64(120), v)
// Older than the window: drop it and tail live, leaving the gap for the // sweep's head check. rc = newCursor(120, time.Now().Add(-2*time.Hour)) require.True(t, rc.dropIfStale(ctx, window)) _, ok = rc.param() require.False(t, ok, "a dropped cursor must not be sent to the relay") require.Equal(t, int64(0), rc.lastEventTime())
// A nonzero cursor with NO event time is a row written before the column // existed -- exactly the state production was poisoned with. Unknown age // means unbounded replay, so it counts as stale. rc = newCursor(120, time.Time{}) require.Equal(t, int64(0), rc.lastEventTime()) require.True(t, rc.dropIfStale(ctx, window)) _, ok = rc.param() require.False(t, ok)
// A zero cursor is already tailing live; there is nothing to drop, so // nothing is reported (and no sweep gets kicked). rc = newCursor(0, time.Time{}) require.False(t, rc.dropIfStale(ctx, window))
// Window 0 disables the cap entirely: however old, replay from it. rc = newCursor(120, time.Now().Add(-72*time.Hour)) require.False(t, rc.dropIfStale(ctx, 0)) v, ok = rc.param() require.True(t, ok) require.Equal(t, int64(120), v)
// The moq flavour of the same cursor is dropped on the same terms, so // SubscribeFrom asks for the live edge rather than a two-hour replay. rc = &relayCursor{host: "moqt://relay.example"} rc.observeGroup(640) rc.observeTime(time.Now().Add(-2 * time.Hour)) require.True(t, rc.dropIfStale(ctx, window)) _, ok = rc.groupStart() require.False(t, ok)}
// TestFirehoseReplayWindow pins the flag plumbing: unset CLI means the default,// and an explicit zero means "never cap", which is how the escape hatch is// spelled everywhere else in the sweep config.func TestFirehoseReplayWindow(t *testing.T) { require.Equal(t, config.DefaultFirehoseReplayWindow, (&ATProtoSynchronizer{}).firehoseReplayWindow()) require.Equal(t, 30*time.Second, (&ATProtoSynchronizer{CLI: &config.CLI{FirehoseReplayWindow: 30 * time.Second}}).firehoseReplayWindow()) require.Equal(t, time.Duration(0), (&ATProtoSynchronizer{CLI: &config.CLI{}}).firehoseReplayWindow())}
// TestRelayCursorObserveTimeFromCallbacks checks the wiring the staleness cap// depends on: event times are recorded for both commit and identity events, on// every transport (unlike the seq, which stays out of a moq relay's group// cursor), and BEFORE dedup -- a duplicate is still evidence of how current// this relay is.func TestRelayCursorObserveTimeFromCallbacks(t *testing.T) { atsync := &ATProtoSynchronizer{ commitDedup: newFirehoseDeduper(dedupWindow), identityDedup: newFirehoseDeduper(dedupWindow), } evtTime := time.Now().UTC().Add(-time.Minute).Format(time.RFC3339Nano) commit := &indigoatproto.SyncSubscribeRepos_Commit{ Seq: 5_000_000_000, Repo: "did:plc:abc123", Time: evtTime, Commit: lexutil.LexLink(mustCID(t)), } // Pre-claim it so the callback stops after its bookkeeping instead of // running the indexing handler, which would want a whole model. atsync.commitDedup.seen(commit.Commit.String())
want, err := time.Parse(time.RFC3339Nano, evtTime) require.NoError(t, err)
moqCursor := &relayCursor{host: "moqt://relay.example"} rsc := atsync.repoStreamCallbacks(context.Background(), "moqt://relay.example", moqCursor, func() {}) require.NoError(t, rsc.RepoCommit(commit)) require.Equal(t, want.Unix(), moqCursor.lastEventTime(), "event time is transport-independent, unlike the seq")
identity := &indigoatproto.SyncSubscribeRepos_Identity{ Seq: 5_000_000_001, Did: "did:plc:abc123", Time: evtTime, } atsync.identityDedup.seen(identityDedupKey(identity)) wsCursor := &relayCursor{host: "wss://relay.example"} rsc = atsync.repoStreamCallbacks(context.Background(), "wss://relay.example", wsCursor, func() {}) require.NoError(t, rsc.RepoIdentity(identity)) require.Equal(t, want.Unix(), wsCursor.lastEventTime())
// An unparsable stamp is simply not an observation, never a crash. wsCursor2 := &relayCursor{host: "wss://relay.example"} rsc = atsync.repoStreamCallbacks(context.Background(), "wss://relay.example", wsCursor2, func() {}) require.NoError(t, rsc.RepoCommit(&indigoatproto.SyncSubscribeRepos_Commit{ Seq: 7, Repo: "did:plc:abc123", Time: "not a timestamp", Commit: lexutil.LexLink(mustCID(t)), })) require.Equal(t, int64(0), wsCursor2.lastEventTime())}