fix(sync): initialize downloads from snapshot metadata

docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md

@@ -0,0 +1,41 @@
+# Snapshot Download Cursor
+
+## Problem
+
+Journal downloads issue GET /sync/{graph-id}/pull (without query; the server defaults since to 0) only to obtain the initial journal cursor. Current Logseq obtains it from /snapshot/download metadata.
+
+## Decision
+
+Start bootstrap with snapshot metadata, validate its non-negative integer t before downloading, and pass that cursor unchanged to snapshot activation. Keep normal incremental pulls and WebSocket catchup. Metadata is sampled before a live stream, not a point-in-time snapshot. Preserve existing metadata fields and optional canonical acknowledgement compatibility.
+
+Production ownership is Core's public reducer: graph selection, metadata completion, download completion, activation and graph attachment determine request order and the cursor. Regression coverage belongs there. Existing transport fixtures will follow the new request sequence without duplicating regression tests.
+
+Sources: logseq/logseq_journal main 2d64cce; logseq/logseq master 69da7e2215; upstream PR #13814; frontend/worker/sync/download.cljs; db_sync/worker/handler/sync.cljs. The current upstream schema requires t and its client fails when absent. Do not add a legacy pull fallback.
+
+## Alternatives considered
+
+### Retain an initial pull for old servers
+
+Rejected: explicitly outside the requested protocol and not current upstream behavior.
+
+### Advance the cursor to the stream's later journal position
+
+Rejected: live rows may include later writes without representing every intervening transaction. Preserve the metadata cursor and let normal catchup replay the journal.
+
+## Acceptance criteria
+
+- Bootstrap emits metadata then download then activation, with no preliminary pull.
+- Metadata t reaches activation and initial WebSocket catchup unchanged.
+- Missing, negative, fractional or malformed t fails before download; zero is accepted.
+- Replaced-graph metadata and stream completions (including failures) are inert; current graph downloads retain their own URL and cursor.
+- Stale completions, download/activation failure and retry cannot publish an uncommitted cursor.
+- Existing incremental synchronization and acknowledgement compatibility tests remain green.
+- No real server writes or graph/account resets. Simulator verification uses an isolated synthetic fixture; release artifacts remain untouched.
+
+## Consequences
+
+- Old servers without metadata t cannot bootstrap, matching current upstream's explicit contract. Existing public legacy request types stay available because changing spec interfaces is outside this task.
+
+## Questions
+
+None outstanding. The user authorized code alignment and verification, prohibited a preliminary pull, and later authorized isolated Simulator testing and a formal PR, while excluding merge, deployment and unrelated repairs.

logseq_db_worker/test/test_effect_runner.ml

@@ -861,9 +861,9 @@ let test_db_shutdown_accepts_late_staging_completion () =
                   (Core.Sync_event
                      (Sync.Runner_completed (Sync.Completion (ticket, Ok [ graph ]))));
                 Eio.Promise.resolve catalog_resolve ()
-              | Sync.Request (_, Sync.Fetch_snapshot_baseline scope) ->
+              | Sync.Request (_, Sync.Fetch_snapshot_metadata scope) ->
                 Eio.Promise.resolve scope_resolve scope
-              | Sync.Request (_, Sync.Fetch_snapshot_metadata _)
+              | Sync.Request (_, Sync.Fetch_snapshot_baseline _)
               | Sync.Request (_, Sync.Download_snapshot _) -> ()
               | Sync.Asset_io (_, request) as runnable ->
                 (match request.action with

logseq_db_worker/test/test_pure_reducer.ml

@@ -1016,16 +1016,11 @@ let snapshot_activation () =
           { event = Some (Sync.Mirror_inspected (Mirror_absent request.scope))
           ; lifecycle = Lifecycle_unchanged }))), request.scope)
     | _ -> None) selected.effects |> Option.get in
-  let baseline = List.find_map (function
-    | Core.Run_sync (Sync.Request (ticket, Sync.Fetch_snapshot_baseline _)) ->
-      Some (Core.step missing.next (Sync_event (Sync.Runner_completed
-        (Sync.Completion (ticket, Ok {|{"type":"pull/ok","t":42}|})))))
-    | _ -> None) missing.effects |> Option.get in
   let download = List.find_map (function
     | Core.Run_sync (Sync.Request (ticket, Sync.Fetch_snapshot_metadata _)) ->
-      Some (Core.step baseline.next (Sync_event (Sync.Runner_completed
-        (Sync.Completion (ticket, Ok {|{"ok":true,"url":"https://example.invalid/snapshot"}|})))))
-    | _ -> None) baseline.effects |> Option.get in
+      Some (Core.step missing.next (Sync_event (Sync.Runner_completed
+        (Sync.Completion (ticket, Ok {|{"ok":true,"t":42,"url":"https://example.invalid/snapshot"}|})))))
+    | _ -> None) missing.effects |> Option.get in
   let activation = List.find_map (function
     | Core.Run_sync (Sync.Request (ticket, Sync.Download_snapshot _)) ->
       let artifact = Sync.staged_artifact ~id:"synthetic" ~scope

logseq_sync/lib/pure_reducer/core.ml

@@ -1074,7 +1074,7 @@ let issue_request : type a. t -> a runner_request -> t * runner_effect =
 let begin_snapshot_bootstrap core scope =
   if not (network_allowed core) then local_unavailable core Local_mirror_unavailable "This graph is not downloaded. Connect to download it." else
   let core = { core with snapshot_scope = Some scope; snapshot_server_t = None } in
-  let next, request = issue_request core (Fetch_snapshot_baseline scope) in
+  let next, request = issue_request core (Fetch_snapshot_metadata scope) in
   { next; effects = [ Run request ] }
 ;;
 
@@ -1134,6 +1134,7 @@ let decode_snapshot_baseline source =
 type snapshot_uri_response =
   { ok : bool
   ; url : string
+  ; t : int64
   }
 [@@deriving of_yojson { strict = false }]
 
@@ -1144,11 +1145,15 @@ let decode_snapshot_uri source =
         snapshot_uri_response_of_yojson
         (Yojson.Safe.from_string source)
     with
-    | Ok { ok = true; url } ->
+    | Ok { ok = true; url; t } ->
       let uri = Uri.of_string url in
-      (match Uri.scheme uri, Uri.host uri, Uri.userinfo uri, Uri.fragment uri with
-       | Some "https", Some host, None, None when String.length host > 0 -> Ok uri
-       | _ -> Error "snapshot download URL is invalid")
+      (match Server_cursor.of_int64 t with
+       | Error `Negative_cursor -> Error "snapshot metadata cursor is invalid"
+       | Ok server_t ->
+         (match Uri.scheme uri, Uri.host uri, Uri.userinfo uri, Uri.fragment uri with
+          | Some "https", Some host, None, None when String.length host > 0 ->
+            Ok (uri, server_t)
+          | _ -> Error "snapshot download URL is invalid"))
     | Ok _ | Error _ -> Error "snapshot metadata is invalid"
   with
   | Yojson.Json_error _ -> Error "snapshot metadata is not valid JSON"
@@ -2368,6 +2373,18 @@ let consume_ticket : type a. t -> a effect_ticket -> t option =
   loop [] core.pending_effects
 ;;
 
+let snapshot_ticket_is_current : type a. t -> a effect_ticket -> bool =
+  fun core ticket ->
+  match ticket.kind with
+  | Fetch_snapshot_baseline_kind | Fetch_snapshot_metadata_kind | Download_snapshot_kind ->
+    (match core.snapshot_scope with
+     | Some scope ->
+       graph_scope_is_current core scope
+       && ticket.scope = effect_scope_of_graph scope
+     | None -> false)
+  | _ -> true
+;;
+
 let graph_key_loaded core key =
   match core.current_graph_scope with
   | Some scope when graph_key_handle_scope key = scope ->
@@ -2598,7 +2615,7 @@ let complete_runner : type a. t -> a effect_ticket -> a -> transition =
      | Ok _, _, _ -> unchanged core)
   | Fetch_snapshot_metadata_kind ->
     (match decode_snapshot_uri value, core.snapshot_scope with
-     | Ok uri, Some scope ->
+     | Ok (uri, server_t), Some scope ->
        let request =
          { scope
          ; uri
@@ -2607,7 +2624,12 @@ let complete_runner : type a. t -> a effect_ticket -> a -> transition =
          }
        in
        let next, request_effect = issue_request core (Download_snapshot request) in
-       { next = { next with active_snapshot_download = Some scope }
+       (* Metadata precedes a live stream that may contain later writes. Keep
+          this cursor so catchup replays all intervening journal entries. *)
+       { next =
+           { next with active_snapshot_download = Some scope
+           ; snapshot_server_t = Some server_t
+           }
        ; effects = [ Run request_effect ]
        }
      | Error message, _ -> fail core During_bootstrap message
@@ -2688,6 +2710,7 @@ let consume_completion core = function
   | Completion (ticket, result) ->
     (match consume_ticket core ticket with
      | None -> unchanged core
+     | Some core when not (snapshot_ticket_is_current core ticket) -> unchanged core
      | Some core ->
        (match result with
         | Ok value -> complete_runner core ticket value