diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index 6219b58d6..b7b44040f 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -455,8 +455,17 @@ func (atsync *ATProtoSynchronizer) repoStreamCallbacks(ctx context.Context, rela if err != nil { return } - atsync.markEvent(aqt.Time()) - cursor.observeTime(aqt.Time()) + at := aqt.Time() + // A relay with a clock running ahead must not pin either mark into the + // future: max-wins would then blank out lag reporting (and make the + // replay cursor look eternally fresh) until that stamp catches up. + // Fold it to the receive time instead -- that is still evidence the + // relay is delivering, so a later, correctly dated event moves on. + if now := time.Now(); at.After(now) { + at = now + } + atsync.markEvent(at) + cursor.observeTime(at) } return &events.RepoStreamCallbacks{ RepoCommit: func(evt *indigoatproto.SyncSubscribeRepos_Commit) error { diff --git a/pkg/atproto/firehose_liveness_test.go b/pkg/atproto/firehose_liveness_test.go index 1ce013ea9..cab607e0a 100644 --- a/pkg/atproto/firehose_liveness_test.go +++ b/pkg/atproto/firehose_liveness_test.go @@ -49,4 +49,18 @@ func TestFirehoseLivenessCountsUnindexedCommits(t *testing.T) { // before dedup: a duplicate is still evidence of what the relay sent. require.NoError(t, rsc.RepoCommit(commit(evtTime.Add(-time.Minute)))) require.Equal(t, evtTime.UnixNano(), atsync.lastEvent.get()) + + // A relay with a clock running ahead must not pin the mark into the + // future: that would silence lag reporting until the bogus stamp caught + // up. The observation folds to the receive time, so a later, correctly + // dated event still advances it. + require.NoError(t, rsc.RepoCommit(commit(time.Now().Add(time.Hour)))) + require.LessOrEqual(t, atsync.lastEvent.get(), time.Now().UnixNano(), + "a future stamp must not push the position past the wall clock") + before := atsync.lastEvent.get() + time.Sleep(2 * time.Millisecond) + require.NoError(t, rsc.RepoCommit(commit(time.Now()))) + require.Greater(t, atsync.lastEvent.get(), before, + "correctly dated events keep advancing after a bogus future one") + require.Less(t, sinceNanos(atsync.lastEvent.get()), time.Minute) }