From 43ef697aea9bf0980e2b3fc8db93d4b0e5d656d3 Mon Sep 17 00:00:00 2001
From: Eli Mallon <eli@iame.li>
Date: Wed, 21 Oct 2026 15:36:52 -0700
Subject: [PATCH] atproto: fold future event stamps to receive time

Both liveness marks are max-wins, so a relay with a clock running ahead
would pin the position into the future: lag reporting would be silenced
and the replay cursor would look eternally fresh until the bogus stamp
caught up. Require an event stamp to be no later than its receive time
before observing it; a later, correctly dated event then moves on
normally.
---
 pkg/atproto/firehose.go               | 13 +++++++++++--
 pkg/atproto/firehose_liveness_test.go | 14 ++++++++++++++
 2 files changed, 25 insertions(+), 2 deletions(-)

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)
 }
-- 
knotmirror 0.0.1