From 379b3165859a69f8bbedf40df366e7afef12ec06 Mon Sep 17 00:00:00 2001
From: Isaac Corbrey <isaac@isaaccorbrey.com>
Date: Wed, 23 Sep 2026 13:50:58 -0400
Subject: [PATCH] jetstream: backfill 24h on connect, harden replay against
 redelivery
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

goose.art joined at://.../boo.sky.go.game/3mw2mwccyqu2r at 01:19 today,
hours before any of today's join-handling fixes deployed at ~14:21.
cursor: None means live-tail only — that join can never surface on its
own. This is the second stuck-join incident from the same structural
gap (missed history during any downtime: deploy, crash, network blip),
so fix the gap, not just this instance.

Connect with a cursor 24h back instead of live-only. Safe now that
replay is idempotent: apply_move_to_game skips already-applied
move_numbers and uses try_apply_move instead of the panicking
apply_move; build_game_state does the same (skip-and-log on an
occupied cell, matching hydrate_from_network's existing replay_moves).

New test apply_move_ignores_already_applied_move locks in that a
redelivered/backfilled move is ignored rather than reapplied or
panicked on.
---
 src/jetstream.rs | 71 +++++++++++++++++++++++++++++++++++++++++-------
 1 file changed, 61 insertions(+), 10 deletions(-)

diff --git a/src/jetstream.rs b/src/jetstream.rs
index 36694e6..64eab1e 100644
--- a/src/jetstream.rs
+++ b/src/jetstream.rs
@@ -86,14 +86,21 @@ fn build_game_state(
     let mut last_move = None;
 
     for m in moves {
-        let caps = board.apply_move(m.x, m.y, m.color.as_u8());
-        move_number = m.move_number;
-        last_move = Some(LastMove {
-            x: m.x,
-            y: m.y,
-            color: m.color,
-            captures: caps,
-        });
+        match board.try_apply_move(m.x, m.y, m.color.as_u8()) {
+            Some(caps) => {
+                move_number = m.move_number;
+                last_move = Some(LastMove {
+                    x: m.x,
+                    y: m.y,
+                    color: m.color,
+                    captures: caps,
+                });
+            }
+            None => error!(
+                "Skipping corrupt move in {}: {},{} (move {}) lands on an occupied cell",
+                game_uri, m.x, m.y, m.move_number
+            ),
+        }
     }
 
     GameState {
@@ -151,9 +158,13 @@ fn apply_move_to_game(
     if parsed.player == bot_did {
         return None;
     }
+    // Already applied (e.g. redelivered by a Jetstream reconnect/backfill) — skip.
+    if parsed.move_number <= game.move_number {
+        return None;
+    }
     let captures = game
         .board
-        .apply_move(parsed.x, parsed.y, parsed.color.as_u8());
+        .try_apply_move(parsed.x, parsed.y, parsed.color.as_u8())?;
     game.move_number = parsed.move_number;
     game.last_move = Some(LastMove {
         x: parsed.x,
@@ -705,6 +716,15 @@ pub async fn run_event_loop(app: &App) -> Result<()> {
         .trim_end_matches("/subscribe")
         .to_string();
 
+    // Backfill the last 24h on connect instead of tailing live-only —
+    // without this, any downtime (deploy, crash, network blip) permanently
+    // drops game/move/action events that happened while disconnected (this
+    // is exactly how two different join actions sat unprocessed for hours).
+    // Safe to replay: handle_game/handle_action guard on already-tracked
+    // state and CAS, and apply_move_to_game/build_game_state/replay_moves
+    // all skip already-applied or occupied-cell moves instead of panicking.
+    let cursor = (chrono::Utc::now() - chrono::Duration::hours(24)).timestamp_micros();
+
     let config = ConsumerTaskConfig {
         user_agent: "jevgo/0.1".to_string(),
         compression: false,
@@ -717,7 +737,7 @@ pub async fn run_event_loop(app: &App) -> Result<()> {
         ],
         dids: vec![],
         max_message_size_bytes: None,
-        cursor: None,
+        cursor: Some(cursor),
         require_hello: true,
     };
 
@@ -1010,6 +1030,37 @@ mod proptests {
             }
         }
 
+        #[test]
+        fn apply_move_ignores_already_applied_move(
+            indices in move_indices(30),
+            bot_is_p1 in any::<bool>(),
+        ) {
+            // A Jetstream reconnect or backfill can redeliver an event
+            // already reflected in state — must not reapply (and must not
+            // panic trying to place a stone on an already-occupied cell).
+            let game_uri = "at://test/boo.sky.go.game/abc";
+            let record = game_record(bot_is_p1, "active");
+            let parsed = parse_game_record(&record).unwrap();
+
+            let initial_moves = generate_move_records(&indices, game_uri, "did:plc:opponent", BOT_DID);
+            if let Some(last) = initial_moves.last().cloned() {
+                let mut state = build_game_state(&parsed, game_uri, BOT_DID, &initial_moves);
+                let move_record = serde_json::json!({
+                    "x": last.x,
+                    "y": last.y,
+                    "color": last.color.name(),
+                    "game": game_uri,
+                    "player": "did:plc:opponent",
+                    "moveNumber": last.move_number,
+                });
+                let parsed_move = parse_move_record(&move_record).unwrap();
+                let result = apply_move_to_game(&mut state, &parsed_move, BOT_DID);
+
+                assert!(result.is_none(), "Already-applied move must be ignored, not reapplied");
+                assert_eq!(state.move_number, last.move_number);
+            }
+        }
+
         #[test]
         fn apply_move_ignores_wrong_game(
             indices in move_indices(20),
-- 
knotmirror 0.0.1