From b3cd739b98e5aff80a4866dfeefdef1082bec0cc Mon Sep 17 00:00:00 2001
From: Eli Mallon <eli@iame.li>
Date: Mon, 12 Oct 2026 01:00:09 +0000
Subject: [PATCH] Repair sparse subscription gaps and verify restored upstream
 sessions

---
 .maestro/logged-in/upstream-station.yaml      | 10 +++++
 .maestro/setup/server-setup.yaml              | 18 +++++----
 go.mod                                        |  2 +-
 go.sum                                        |  4 +-
 hack/e2e-local.sh                             |  8 +++-
 hack/lib/e2e-harness.sh                       |  2 +-
 .../src/components/chat/chat-box.tsx          |  5 ++-
 js/desktop/README.md                          |  3 ++
 js/e2e-web/README.md                          |  3 ++
 .../flows/desktop-upstream-station.spec.ts    |  1 +
 js/e2e-web/flows/upstream-station.spec.ts     | 10 +++++
 pkg/atproto/firehose.go                       |  2 +-
 pkg/atproto/firehose_moq.go                   | 13 +++----
 pkg/atproto/headcheck_test.go                 | 39 ++++++++++++++++++-
 pkg/atproto/sweep.go                          | 19 ++++++++-
 pkg/cmd/e2e.go                                |  4 +-
 16 files changed, 113 insertions(+), 30 deletions(-)

diff --git a/.maestro/logged-in/upstream-station.yaml b/.maestro/logged-in/upstream-station.yaml
index 0c13a69b0..e71a88801 100644
--- a/.maestro/logged-in/upstream-station.yaml
+++ b/.maestro/logged-in/upstream-station.yaml
@@ -22,6 +22,11 @@ tags:
 - tapOn:
     id: "chat-input"
 - inputText: "upstream-station native write"
+- extendedWaitUntil:
+    visible:
+      text: "Send"
+      enabled: true
+    timeout: 30000
 - pressKey: Enter
 - extendedWaitUntil:
     visible:
@@ -46,6 +51,11 @@ tags:
 - tapOn:
     id: "chat-input"
 - inputText: "upstream-station native write after relaunch"
+- extendedWaitUntil:
+    visible:
+      text: "Send"
+      enabled: true
+    timeout: 30000
 - pressKey: Enter
 - extendedWaitUntil:
     visible:
diff --git a/.maestro/setup/server-setup.yaml b/.maestro/setup/server-setup.yaml
index 5bc6bd897..eba437e8d 100644
--- a/.maestro/setup/server-setup.yaml
+++ b/.maestro/setup/server-setup.yaml
@@ -48,12 +48,14 @@ appId: ${APP_ID}
 - takeScreenshot: server-setup-saved
 - tapOn:
     text: "Home"
-# The harness's own stream in the feed proves the app is talking to the
-# custom node; any card would also match production's feed, which is what
-# the app shows until the new node takes effect. Matched as a substring
-# because iOS collapses the card into one element with a concatenated label.
-- extendedWaitUntil:
-    visible:
-      text: ".*e2e test stream.*"
-    timeout: 30000
+# Standalone nodes already index the fixture stream. Upstream local nodes
+# intentionally have no indexed users until the following OAuth login flow.
+- runFlow:
+    when:
+      true: ${E2E_UPSTREAM_STATION != "true"}
+    commands:
+      - extendedWaitUntil:
+          visible:
+            text: ".*e2e test stream.*"
+          timeout: 30000
 - takeScreenshot: server-setup-home
diff --git a/go.mod b/go.mod
index 7bbf62d98..5294f2db7 100644
--- a/go.mod
+++ b/go.mod
@@ -79,7 +79,7 @@ require (
 	github.com/streamplace/atproto-oauth-golang v0.0.0-20260908211910-e9bac3356167
 	github.com/streamplace/glex v0.0.0-20260820164827-814f46540f22
 	github.com/streamplace/muxl/go v0.3.6
-	github.com/streamplace/oatproxy v0.0.0-20261012002509-80a9af303a32
+	github.com/streamplace/oatproxy v0.0.0-20261012003324-43c7ac5473f7
 	github.com/stretchr/testify v1.11.1
 	github.com/tdewolff/canvas v0.0.0-20250728095813-50d4cb1eee71
 	github.com/tus/tusd/v2 v2.8.0
diff --git a/go.sum b/go.sum
index 33b80c46a..d8e94a12c 100644
--- a/go.sum
+++ b/go.sum
@@ -1397,8 +1397,8 @@ github.com/streamplace/indigo v0.0.0-20260218231908-939cdaf0c507 h1:e8M3qPLr37Nx
 github.com/streamplace/indigo v0.0.0-20260218231908-939cdaf0c507/go.mod h1:Pm2I1+iDXn/hLbF7XCg/DsZi6uDCiOo7hZGWprSM7k0=
 github.com/streamplace/muxl/go v0.3.6 h1:lyWnY15nft/0GK3LBA4OkD2aHrLV9RRqI+YcDRPFk3g=
 github.com/streamplace/muxl/go v0.3.6/go.mod h1:aCyYTW3o6c1Kush9UJ/Yv6EYMUbj8l8GTD7cHKcSxw8=
-github.com/streamplace/oatproxy v0.0.0-20261012002509-80a9af303a32 h1:rZpc/Z3xCx2ooa6ypoUGEPUxifZRBprqmOt5kP0lhJk=
-github.com/streamplace/oatproxy v0.0.0-20261012002509-80a9af303a32/go.mod h1:WRKRYf1OcBCa/vWJSJOxabqMIPelwktWcT6WkTj56Hk=
+github.com/streamplace/oatproxy v0.0.0-20261012003324-43c7ac5473f7 h1:4cFVSUVeQSX2iAgK4z3PnIqDIZRho/fIwvAbPimvWUU=
+github.com/streamplace/oatproxy v0.0.0-20261012003324-43c7ac5473f7/go.mod h1:WRKRYf1OcBCa/vWJSJOxabqMIPelwktWcT6WkTj56Hk=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
diff --git a/hack/e2e-local.sh b/hack/e2e-local.sh
index 06e743e08..0db11ef2a 100755
--- a/hack/e2e-local.sh
+++ b/hack/e2e-local.sh
@@ -98,7 +98,10 @@ adb_ shell settings put secure stylus_handwriting_enabled 0 || true
 # otherwise float over the app and eat maestro's taps
 adb_ shell settings put global hide_error_dialogs 1 || true
 
-if [ -z "${SERVER_HTTPS_URL:-}" ]; then
+if [ "${E2E_UPSTREAM_STATION:-}" = true ]; then
+  : "${UPSTREAM_STATION_URL:?upstream mode requires the station HTTPS URL}"
+  : "${PDS_HTTPS_URL:?upstream mode requires the PDS HTTPS URL}"
+elif [ -z "${SERVER_HTTPS_URL:-}" ]; then
   echo "the Android suite needs the harness's HTTPS mode (release builds refuse cleartext); unset E2E_HTTPS_PDS_HOSTNAME"
   exit 1
 fi
@@ -143,13 +146,14 @@ adb_ shell pm grant "$APP_ID" android.permission.POST_NOTIFICATIONS 2>/dev/null
 adb_ shell am force-stop com.android.chrome || true
 
 
-APP_SERVER_URL="$SERVER_HTTPS_URL"
 if [ "${E2E_UPSTREAM_STATION:-}" = true ]; then
   # Keep the issuer and callback loopback names identical on host and device.
   # Merely substituting 10.0.2.2 would change the DPoP-bound request target.
   local_port="${SERVER_URL##*:}"
   adb_ reverse "tcp:$local_port" "tcp:$local_port"
   APP_SERVER_URL="$SERVER_URL"
+else
+  APP_SERVER_URL="$SERVER_HTTPS_URL"
 fi
 if [ "${E2E_UPSTREAM_STATION:-}" = true ]; then
   e2e_android_stop() {
diff --git a/hack/lib/e2e-harness.sh b/hack/lib/e2e-harness.sh
index c6811b652..1f4825468 100644
--- a/hack/lib/e2e-harness.sh
+++ b/hack/lib/e2e-harness.sh
@@ -104,7 +104,7 @@ e2e_harness_start() {
   export SERVER_HTTPS_URL PDS_HTTPS_URL E2E_PROXY_URL E2E_TLS_SPKI E2E_TLS_CA
   export E2E_UPSTREAM_STATION UPSTREAM_STATION_URL ATMOQ_RELAY_URL E2E_ATMOQ_LOG
   export E2E_LOCAL_DATA_DIR E2E_LOCAL_PDS_AUDIT E2E_STATION_DATA_DIR
-  export UNRELATED_ACCOUNT_DID UNRELATED_ACCOUNT_HANDLE UNRELATED_ACCESS_JWT E2E_PDS_INTERNAL_URL
+  export UNRELATED_ACCOUNT_DID UNRELATED_ACCOUNT_HANDLE UNRELATED_ACCESS_JWT E2E_PDS_INTERNAL_URL E2E_PLC_URL
   if [ "${E2E_UPSTREAM_STATION:-}" = true ]; then
     echo "upstream topology: local=$SERVER_URL station=$UPSTREAM_STATION_URL relay=$ATMOQ_RELAY_URL"
   fi
diff --git a/js/components/src/components/chat/chat-box.tsx b/js/components/src/components/chat/chat-box.tsx
index 71ff8cfd4..dc89169b9 100644
--- a/js/components/src/components/chat/chat-box.tsx
+++ b/js/components/src/components/chat/chat-box.tsx
@@ -114,6 +114,7 @@ export function ChatBox({
 
   const pdsAgent = usePDSAgent();
   const userDID = useDID();
+  const canSend = !submitting && !!pdsAgent && !!userDID && !!profile;
   const setActiveTeleportUri = useLivestreamStore(
     (state) => state.setActiveTeleportUri,
   );
@@ -329,7 +330,7 @@ export function ChatBox({
   };
 
   const submit = async () => {
-    if (!message.trim()) return;
+    if (!canSend || !message.trim()) return;
     if (graphemer.countGraphemes(message) > 300) {
       toast.show(
         "Message too long",
@@ -580,7 +581,7 @@ export function ChatBox({
         </View>
         <View>
           <Button
-            disabled={submitting}
+            disabled={!canSend}
             variant="secondary"
             width="min"
             style={{ borderRadius: theme.borderRadius.md, height: 43 }}
diff --git a/js/desktop/README.md b/js/desktop/README.md
index 8b6075fb8..a2f4a72cd 100644
--- a/js/desktop/README.md
+++ b/js/desktop/README.md
@@ -26,6 +26,9 @@ local users. Startup, sweeps, history backfill and event handling use the same
 persisted OAuth eligibility; previously indexed repos do not authorize new
 scraping. Logout removes eligibility when the last local grant ends. Existing
 indexed records remain on disk, but no longer receive updates.
+After subscribing or reconnecting, the node queues a head check and shallow
+repair for that DID behind any active sweep. Sparse tracks do not depend on
+firehose replay or wait for the next periodic sweep to recover missed writes.
 
 OAuth has three independent DPoP sessions: frontend → loopback node → HTTPS
 station → PDS. Only the station needs a public HTTPS callback and retains the
diff --git a/js/e2e-web/README.md b/js/e2e-web/README.md
index 052b476d3..59dc0ccd6 100644
--- a/js/e2e-web/README.md
+++ b/js/e2e-web/README.md
@@ -172,6 +172,9 @@ real Go node with its on-disk sessions and acknowledges readiness in a private
 file. Authenticated writes and indexed history must survive that restart.
 Real browser logout revokes the local downstream session and stops the final
 local DID subscription without destroying the station's confidential PDS grant.
+Chat submission waits for the restored OAuth session and initial streamer
+profile; Enter during initialization leaves the draft intact rather than
+clearing a message that cannot yet be sent.
 
 Prerequisites additionally include Python 3 with SQLite support (for fixture
 inspection), a POSIX shell with `mkfifo`, free loopback HTTPS/TCP/UDP ports, and
diff --git a/js/e2e-web/flows/desktop-upstream-station.spec.ts b/js/e2e-web/flows/desktop-upstream-station.spec.ts
index 7680ce761..a8416efa4 100644
--- a/js/e2e-web/flows/desktop-upstream-station.spec.ts
+++ b/js/e2e-web/flows/desktop-upstream-station.spec.ts
@@ -59,6 +59,7 @@ test("desktop-upstream-station: login and writes survive closing Desktop", async
     SP_DEV_FRONTEND_PROXY: "",
     SP_DEV_PUBLIC_OAUTH: "false",
     SP_TRUST_PRIVATE_NETWORK: "true",
+    SP_PLC_URL: process.env.E2E_PLC_URL!,
     SSL_CERT_FILE: process.env.E2E_TLS_CA!,
     HTTPS_PROXY: process.env.E2E_PROXY_URL!,
   });
diff --git a/js/e2e-web/flows/upstream-station.spec.ts b/js/e2e-web/flows/upstream-station.spec.ts
index 003167f0d..6edb65878 100644
--- a/js/e2e-web/flows/upstream-station.spec.ts
+++ b/js/e2e-web/flows/upstream-station.spec.ts
@@ -113,6 +113,9 @@ async function sendChat(
   );
   const input = page.getByPlaceholder("Type a message...");
   await input.fill(message);
+  await expect(
+    page.getByRole("button", { name: "Send", exact: true }),
+  ).toBeEnabled();
   await input.press("Enter");
   const response = await created;
   expect(
@@ -187,11 +190,18 @@ async function refreshAtLocal(page: Page): Promise<void> {
         true &&
       response.ok(),
   );
+  const restoredProfile = page.waitForResponse(
+    (response) =>
+      new URL(response.url()).origin === LOCAL &&
+      new URL(response.url()).pathname === "/xrpc/app.bsky.actor.getProfile" &&
+      response.ok(),
+  );
   await page.reload();
   const body = await (await refreshed).json();
   expect(body.sub).toBe(DID);
   expect(body.token_type).toBe("DPoP");
   expect(body.access_token).toBeTruthy();
+  expect((await (await restoredProfile).json()).did).toBe(DID);
 }
 
 test("upstream-station: local login, isolated DPoP hops, MoQ indexing, refresh and revoke", async ({
diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go
index 2b3cee655..fedb90fc4 100644
--- a/pkg/atproto/firehose.go
+++ b/pkg/atproto/firehose.go
@@ -74,7 +74,7 @@ type ATProtoSynchronizer struct {
 
 	// sweeping is held for the length of a sweep, so the periodic ticker
 	// cannot start a second one on top of the first.
-	sweeping atomic.Bool
+	sweeping sync.Mutex
 }
 
 func (atsync *ATProtoSynchronizer) markSeen() {
diff --git a/pkg/atproto/firehose_moq.go b/pkg/atproto/firehose_moq.go
index e4554cb65..ef9807576 100644
--- a/pkg/atproto/firehose_moq.go
+++ b/pkg/atproto/firehose_moq.go
@@ -77,14 +77,13 @@ func (atsync *ATProtoSynchronizer) connectRelayMoq(ctx context.Context, relay, t
 
 	log.Log(ctx, "connected to relay firehose (moq)", "version", sess.Version(), "track", track)
 	if track != atmoq.DefaultTrack {
-		// Reconcile after subscribing, so changes made while this sparse track
-		// was absent cannot fall between the PDS snapshot and live delivery.
-		// The consumer context survives reconnects; a dropped connection must
-		// not abort the gap repair it triggered.
+		// Repair after any in-flight sweep, which may have checked this DID
+		// before the gap. The consumer context survives reconnects; revocation
+		// still cancels the queued repair.
 		go func() {
-			atsync.sweepOnce(ctx, func(sweepCtx context.Context) error {
-				return atsync.sweep(sweepCtx, false)
-			})
+			if err := atsync.repairSubscribedRepo(ctx, track); err != nil && ctx.Err() == nil {
+				log.Error(ctx, "failed to repair upstream repo after subscribing", "did", track, "err", err)
+			}
 		}()
 	}
 	for {
diff --git a/pkg/atproto/headcheck_test.go b/pkg/atproto/headcheck_test.go
index a4715a893..a41d0cd94 100644
--- a/pkg/atproto/headcheck_test.go
+++ b/pkg/atproto/headcheck_test.go
@@ -56,6 +56,27 @@ func TestHeadCheckHealsSilentGap(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, messages, 1)
 
+	// A periodic sweep has checked this repo but is still busy elsewhere
+	// when the sparse subscription drops. Its snapshot predates the gap.
+	checked, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	sweepErr := make(chan error, 1)
+	go func() {
+		defer close(finished)
+		atsync.sweepOnce(ctx, func(ctx context.Context) error {
+			err := atsync.sweep(ctx, false)
+			sweepErr <- err
+			close(checked)
+			<-release
+			return err
+		})
+	}()
+	t.Cleanup(func() {
+		close(release)
+		<-finished
+	})
+	<-checked
+	require.NoError(t, <-sweepErr)
+
 	// Behind our back: no firehose is running in this test, so nothing at all
 	// tells the index that this happened.
 	createBackfillRecord(t, user, "place.stream.chat.message", "", chatMessageRecord(user.DID, "after the gap"))
@@ -78,8 +99,22 @@ func TestHeadCheckHealsSilentGap(t *testing.T) {
 	require.NoError(t, err)
 	require.NotEqual(t, stale.Version, hostRev, "the repo really did move")
 
-	// The sweep's head-check pass finds the drift and repairs it.
-	require.NoError(t, atsync.Sweep(ctx))
+	// The reconnect repair queues behind that pass rather than disappearing.
+	repaired := make(chan error, 1)
+	go func() { repaired <- atsync.repairSubscribedRepo(ctx, user.DID) }()
+	select {
+	case err := <-repaired:
+		t.Fatalf("reconnect repair returned before the active sweep finished: %v", err)
+	case <-time.After(100 * time.Millisecond):
+	}
+	// Release the sweep without closing its channel twice in cleanup.
+	release <- struct{}{}
+	select {
+	case err := <-repaired:
+		require.NoError(t, err)
+	case <-time.After(30 * time.Second):
+		t.Fatal("reconnect repair did not finish after the periodic sweep")
+	}
 
 	healed, err := mod.GetRepo(user.DID)
 	require.NoError(t, err)
diff --git a/pkg/atproto/sweep.go b/pkg/atproto/sweep.go
index 2987617cc..c9de2b81a 100644
--- a/pkg/atproto/sweep.go
+++ b/pkg/atproto/sweep.go
@@ -520,11 +520,11 @@ func (atsync *ATProtoSynchronizer) sweepLoop(ctx context.Context, bootDelay, int
 // queued: the next one is another interval away, which is exactly when a sweep
 // that just finished should run again.
 func (atsync *ATProtoSynchronizer) sweepOnce(ctx context.Context, sweep func(context.Context) error) {
-	if !atsync.sweeping.CompareAndSwap(false, true) {
+	if !atsync.sweeping.TryLock() {
 		log.Log(ctx, "skipping scheduled sweep; the previous one is still running")
 		return
 	}
-	defer atsync.sweeping.Store(false)
+	defer atsync.sweeping.Unlock()
 	if err := sweep(ctx); err != nil && ctx.Err() == nil {
 		log.Error(ctx, "backfill sweep failed", "err", err)
 	}
@@ -585,6 +585,21 @@ func (atsync *ATProtoSynchronizer) sweep(ctx context.Context, ladders bool) erro
 	if err != nil {
 		return err
 	}
+	return atsync.sweepRepos(ctx, dids, ladders)
+}
+
+// repairSubscribedRepo must run after any active head-check pass, not be
+// discarded like a periodic tick: sparse tracks cannot replay a reconnect gap.
+func (atsync *ATProtoSynchronizer) repairSubscribedRepo(ctx context.Context, did string) error {
+	atsync.sweeping.Lock()
+	defer atsync.sweeping.Unlock()
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	return atsync.sweepRepos(ctx, []string{did}, false)
+}
+
+func (atsync *ATProtoSynchronizer) sweepRepos(ctx context.Context, dids []string, ladders bool) error {
 	log.Log(ctx, "starting backfill sweep", "totalRepos", len(dids),
 		"concurrency", atsync.sweepConcurrency(), "deepening", ladders)
 
diff --git a/pkg/cmd/e2e.go b/pkg/cmd/e2e.go
index 87cd3f594..0bbbb3102 100644
--- a/pkg/cmd/e2e.go
+++ b/pkg/cmd/e2e.go
@@ -604,8 +604,8 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri
 		vars += fmt.Sprintf("PDS_HTTPS_URL=%s\nE2E_PROXY_URL=%s\nE2E_TLS_SPKI=%s\nE2E_TLS_CA=%s\n",
 			tlsEnv.PDSURL(), tlsEnv.ProxyURL(), tlsEnv.spki, tlsEnv.caPath)
 		if local != nil {
-			vars += fmt.Sprintf("E2E_UPSTREAM_STATION=true\nUPSTREAM_STATION_URL=%s\nATMOQ_RELAY_URL=%s\nE2E_ATMOQ_LOG=%s\nE2E_LOCAL_DATA_DIR=%s\nE2E_STATION_DATA_DIR=%s\nE2E_LOCAL_PDS_AUDIT=%s\nUNRELATED_ACCOUNT_DID=%s\nUNRELATED_ACCOUNT_HANDLE=%s\nUNRELATED_ACCESS_JWT=%s\nE2E_PDS_INTERNAL_URL=%s\n",
-				tlsEnv.StationURL(), bridgeURL, filepath.Join(tlsEnv.dir, "atmoq.log"), local.dataDir, dataDir, tlsEnv.auditPath, unrelated.did, unrelated.handle, unrelated.accessJWT, env.PDSURL)
+			vars += fmt.Sprintf("E2E_UPSTREAM_STATION=true\nUPSTREAM_STATION_URL=%s\nATMOQ_RELAY_URL=%s\nE2E_ATMOQ_LOG=%s\nE2E_LOCAL_DATA_DIR=%s\nE2E_STATION_DATA_DIR=%s\nE2E_LOCAL_PDS_AUDIT=%s\nUNRELATED_ACCOUNT_DID=%s\nUNRELATED_ACCOUNT_HANDLE=%s\nUNRELATED_ACCESS_JWT=%s\nE2E_PDS_INTERNAL_URL=%s\nE2E_PLC_URL=%s\n",
+				tlsEnv.StationURL(), bridgeURL, filepath.Join(tlsEnv.dir, "atmoq.log"), local.dataDir, dataDir, tlsEnv.auditPath, unrelated.did, unrelated.handle, unrelated.accessJWT, env.PDSURL, env.PLCURL)
 		} else {
 			vars += fmt.Sprintf("SERVER_HTTPS_URL=%s\n", tlsEnv.StationURL())
 		}
-- 
2.51.2