From 59c12a9672cbbfa6bcaee82582583f102c4cd3b5 Mon Sep 17 00:00:00 2001 From: rcmerci Date: Wed, 21 Oct 2026 16:49:32 +0800 Subject: [PATCH 1/2] fix(sync): initialize downloads from snapshot metadata --- .../2026-10-21-snapshot-download-cursor.md | 40 ++ logseq_sync/lib/pure_reducer/core.ml | 24 +- logseq_sync/test/core_contract.ml | 351 +++++++++++++----- logseq_sync/test/transport_contract.ml | 23 +- 4 files changed, 311 insertions(+), 127 deletions(-) create mode 100644 docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md diff --git a/docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md b/docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md new file mode 100644 index 0000000..749aaea --- /dev/null +++ b/docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md @@ -0,0 +1,40 @@ +# Snapshot Download Cursor + +## Problem + +Journal downloads issue GET /pull?since=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. +- Stale completions, download/activation failure and retry cannot publish an uncommitted cursor. +- Existing incremental synchronization and acknowledgement compatibility tests remain green. +- No server writes, graph resets, signing, Simulator or release artifact mutations. + +## 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 scoped out publishing and unrelated repairs. diff --git a/logseq_sync/lib/pure_reducer/core.ml b/logseq_sync/lib/pure_reducer/core.ml index 4ed15c6..927224a 100644 --- a/logseq_sync/lib/pure_reducer/core.ml +++ b/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" @@ -2598,7 +2603,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 +2612,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 diff --git a/logseq_sync/test/core_contract.ml b/logseq_sync/test/core_contract.ml index 4a1dbb9..d56f108 100644 --- a/logseq_sync/test/core_contract.ml +++ b/logseq_sync/test/core_contract.ml @@ -696,7 +696,7 @@ let warm_encrypted_graph_online_recovery_attaches_existing_mirror () = false (List.exists (function - | Core.Run (Core.Request (_, Core.Fetch_snapshot_baseline _)) -> true + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) -> true | Run _ | Delegate _ | Publish _ -> false) recovered.effects) ;; @@ -849,7 +849,7 @@ let e2ee_password_retry_recovers_after_failed_unlock () = true (List.exists (function - | Core.Run (Core.Request (_, Core.Fetch_snapshot_baseline _)) -> true + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) -> true | Run _ | Delegate _ | Publish _ -> false) resumed.effects); let picker = Core.step rejected.next Core.Graph_picker_requested in @@ -977,7 +977,7 @@ let encrypted_graph_recovery_fetches_and_unlocks_key_before_bootstrap () = true (List.exists (function - | Core.Run (Core.Request (_, Core.Fetch_snapshot_baseline _)) -> true + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) -> true | Run _ | Delegate _ | Publish _ -> false) bootstrapping.effects) ;; @@ -1013,34 +1013,27 @@ let begin_snapshot_bootstrap () = missing, scope ;; -let cold_start_absent_mirror_reaches_pull_after_snapshot_activation () = +let snapshot_catchup_fixture () = let requested, scope = begin_snapshot_bootstrap () in - let metadata = - List.find_map - (function - | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> - Some - (Core.step - requested.next - (Core.Runner_completed - (Core.Completion (ticket, Ok {|{"type":"pull/ok","t":42}|})))) - | Run _ | Delegate _ | Publish _ -> None) - requested.effects - |> Option.get - in + Alcotest.(check bool) + "bootstrap starts with exactly one metadata request and no preliminary pull" + true + (match List.filter (function Core.Publish _ -> false | _ -> true) requested.effects with + | [ Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) ] -> true + | _ -> false); let download = List.find_map (function | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> Some (Core.step - metadata.next + requested.next (Core.Runner_completed (Core.Completion ( ticket - , Ok {|{"ok":true,"url":"https://example.com/snapshot.sqlite"}|} )))) + , Ok {|{"ok":true,"t":42,"url":"https://example.com/snapshot.sqlite"}|} )))) | Run _ | Delegate _ | Publish _ -> None) - metadata.effects + requested.effects |> Option.get in let activated = @@ -1075,13 +1068,16 @@ let cold_start_absent_mirror_reaches_pull_after_snapshot_activation () = |> Option.get in Alcotest.(check cursor_testable) - "activation uses baseline cursor" + "activation uses metadata cursor" (server_cursor_of_int 42) activation.applied_server_t; Alcotest.(check bool) "plain snapshot has no graph key" true (Option.is_none activation.key); + Alcotest.(check (option cursor_testable)) + "download completion cannot publish an uncommitted cursor" + None (Core.state activated.next).snapshot.applied_server_t; let activated_snapshot = Core.step activated.next (Core.Snapshot_activated { scope }) in let mirror = inspect_mirror_request activated_snapshot.effects in let attaching = @@ -1131,11 +1127,15 @@ let cold_start_absent_mirror_reaches_pull_after_snapshot_activation () = { scope; message = Logseq_sync_pure_reducer.Sync_protocol.Client.Pull _ }) -> scope = connection | Run _ | Delegate _ | Publish _ -> false) - pulling.effects) + pulling.effects); + pulling, connection, sync ;; -let snapshot_baseline_above_websocket_budget_bootstraps () = - let requested, _scope = begin_snapshot_bootstrap () in +let cold_start_absent_mirror_reaches_pull_after_snapshot_activation () = + ignore (snapshot_catchup_fixture ()) +;; + +let snapshot_baseline_decoder_retains_http_budget () = (* Ordinary cursor values and small valid Transit transactions form a valid HTTP pull response larger than a WebSocket frame, within the endpoint budget. *) let transaction = @@ -1168,50 +1168,30 @@ let snapshot_baseline_above_websocket_budget_bootstraps () = true (String.length body > Logseq_db_types.Limits.maximum_response_bytes && String.length body <= 64 * 1024 * 1024); - let metadata = - List.find_map - (function - | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> - Some - (Core.step - requested.next - (Core.Runner_completed (Core.Completion (ticket, Ok body)))) - | Run _ | Delegate _ | Publish _ -> None) - requested.effects - |> Option.get - in Alcotest.(check bool) - "valid HTTP baseline proceeds to snapshot metadata" + "legacy HTTP baseline decoder retains its endpoint budget" true - (List.exists - (function - | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) -> true - | Run _ | Delegate _ | Publish _ -> false) - metadata.effects); - Alcotest.(check bool) - "valid HTTP baseline does not fail bootstrap" - false - ((Core.state metadata.next).snapshot.sync_phase = Core.Failed) + (Result.is_ok (Logseq_sync_pure_reducer.Sync_protocol.decode_snapshot_baseline body)) ;; -let invalid_snapshot_baseline_fails_bootstrap () = +let invalid_snapshot_metadata_fails_bootstrap () = let requested, _scope = begin_snapshot_bootstrap () in let failed = List.find_map (function - | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> + | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> Some (Core.step requested.next (Core.Runner_completed - (Core.Completion (ticket, Ok {|{"type":"pull/ok","t":-1}|})))) + (Core.Completion (ticket, Ok {|{"ok":true,"t":-1,"url":"https://example.com/snapshot"}|})))) | Run _ | Delegate _ | Publish _ -> None) requested.effects |> Option.get in let snapshot = (Core.state failed.next).snapshot in Alcotest.(check bool) - "invalid baseline enters failed state" + "invalid metadata cursor enters failed state" true (snapshot.sync_phase = Core.Failed && snapshot.startup.failure = Some Core.During_bootstrap) @@ -2337,32 +2317,19 @@ let local_cache_deletion_rejects_unopened_graphs () = let absent = Core.step selected.next (Core.Mirror_inspected (Core.Mirror_absent scope)) in - let metadata = - List.find_map - (function - | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> - Some - (Core.step - absent.next - (Core.Runner_completed - (Core.Completion (ticket, Ok {|{"type":"pull/ok","t":42}|})))) - | _ -> None) - absent.effects - |> Option.get - in let downloading = List.find_map (function | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> Some (Core.step - metadata.next + absent.next (Core.Runner_completed (Core.Completion ( ticket - , Ok {|{"ok":true,"url":"https://example.com/snapshot.sqlite"}|} )))) + , Ok {|{"ok":true,"t":42,"url":"https://example.com/snapshot.sqlite"}|} )))) | _ -> None) - metadata.effects + absent.effects |> Option.get in let activating = @@ -2409,7 +2376,6 @@ let local_cache_deletion_rejects_unopened_graphs () = ; authenticated.next ; selected.next ; absent.next - ; metadata.next ; downloading.next ; activating.next ; activated.next @@ -2577,7 +2543,7 @@ let local_deletion_orders_cleanup encrypted () = let absent = Core.step requested.next (Core.Mirror_inspected (Core.Mirror_absent mirror.scope)) in Alcotest.(check bool) "explicit download of deleted graph uses normal absent-mirror flow" true (List.exists (function - | Core.Run (Core.Request (_, Core.Fetch_snapshot_baseline _)) -> not encrypted + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) -> not encrypted | Core.Run (Core.Request (_, Core.Load_and_unlock_graph_key _)) -> encrypted | _ -> false) absent.effects); let _ = sync in @@ -4044,7 +4010,7 @@ let scenarios = [ Alcotest.test_case "snapshot baseline retains HTTP response budget" `Quick - snapshot_baseline_above_websocket_budget_bootstraps + snapshot_baseline_decoder_retains_http_budget ; Alcotest.test_case "numeric cursor decimal boundaries" `Quick @@ -4142,9 +4108,9 @@ let scenarios = `Quick cold_start_absent_mirror_reaches_pull_after_snapshot_activation ; Alcotest.test_case - "invalid snapshot baseline fails bootstrap" + "invalid snapshot metadata fails bootstrap" `Quick - invalid_snapshot_baseline_fails_bootstrap + invalid_snapshot_metadata_fails_bootstrap ; Alcotest.test_case "E2EE password retry after failed unlock" `Quick @@ -5160,31 +5126,16 @@ let snapshot_metadata_open_fields_keep_declared_constraints () = List.iter (fun ((body : string), success) -> let requested, _ = begin_snapshot_bootstrap () in - let metadata = - List.find_map - (function - | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> - Some - (Core.step - requested.next - (Core.Runner_completed - (Core.Completion - ( ticket - , Ok {|{"type":"pull/ok","t":1,"future":{"nested":true}}|} )))) - | _ -> None) - requested.effects - |> Option.get - in let downloaded = List.find_map (function | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> Some (Core.step - metadata.next + requested.next (Core.Runner_completed (Core.Completion (ticket, Ok body)))) | _ -> None) - metadata.effects + requested.effects |> Option.get in Alcotest.(check bool) @@ -5195,12 +5146,20 @@ let snapshot_metadata_open_fields_keep_declared_constraints () = | Core.Run (Core.Request (_, Core.Download_snapshot _)) -> true | _ -> false) downloaded.effects)) - [ ( {|{"ok":true,"url":"https://example.com/snapshot","key":"synthetic","future":{"nested":true}}|} + [ ( {|{"ok":true,"t":0,"url":"https://example.com/snapshot","key":"synthetic","future":{"nested":true}}|} , true ) - ; {|{"ok":true,"future":true}|}, false - ; {|{"ok":true,"url":null,"future":true}|}, false - ; {|{"ok":"true","url":"https://example.com/snapshot","future":true}|}, false - ; {|{"ok":true,"url":"http://example.com/snapshot","future":true}|}, false + ; {|{"ok":true,"url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":null,"url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":-1,"url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":1.5,"url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":"42","url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":true,"url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":0,"t":1,"url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":9223372036854775808,"url":"https://example.com/snapshot"}|}, false + ; {|{"ok":true,"t":0,"future":true}|}, false + ; {|{"ok":true,"t":0,"url":null,"future":true}|}, false + ; {|{"ok":"true","t":0,"url":"https://example.com/snapshot","future":true}|}, false + ; {|{"ok":true,"t":0,"url":"http://example.com/snapshot","future":true}|}, false ] ;; @@ -5482,12 +5441,13 @@ let cancellation_fences_deferred_catalog_and_snapshot () = Alcotest.(check bool) "local completion cannot restore online authentication" false (Core.state loaded.next).snapshot.startup.authenticated; let requested, _ = begin_snapshot_bootstrap () in - let metadata = List.find_map (function - | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> - Some (Core.step requested.next (Core.Runner_completed - (Core.Completion (ticket, Ok {|{"type":"pull/ok","t":42}|})))) - | _ -> None) requested.effects |> Option.get in - let cancelled = Core.step metadata.next (Core.Online_cancelled (startup_ticket metadata.next)) in + let cancelled = Core.step requested.next (Core.Online_cancelled (startup_ticket requested.next)) in + List.iter (function + | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> + let late = Core.step cancelled.next (Core.Runner_completed + (Core.Completion (ticket, Ok {|{"ok":true,"t":42,"url":"https://example.com/snapshot"}|}))) in + check_instructions "cancelled snapshot metadata completion" [] late.effects + | _ -> ()) requested.effects; let facts = (Core.state cancelled.next).snapshot.startup in Alcotest.(check bool) "cancelled metadata has a terminal local result" true (match facts.local_access with Core.Local_unavailable _ -> true | _ -> false) @@ -5563,10 +5523,10 @@ let user_selected_missing_mirror () = && facts.local_access = Core.Local_pending && not facts.authenticated); let confirmed = Core.step missing.next (Core.Account_authenticated {user_id=Some "user"}) in let downloading = confirmed in - Alcotest.(check bool) "verified selection starts one baseline request" true - (List.exists (function Core.Run(Core.Request(_,Core.Fetch_snapshot_baseline _))->true|_->false) downloading.effects); + Alcotest.(check bool) "verified selection starts one metadata request" true + (List.exists (function Core.Run(Core.Request(_,Core.Fetch_snapshot_metadata _))->true|_->false) downloading.effects); let failure = List.find_map (function - | Core.Run(Core.Request(ticket,Core.Fetch_snapshot_baseline _)) -> + | Core.Run(Core.Request(ticket,Core.Fetch_snapshot_metadata _)) -> Some (Core.Runner_completed(Core.Completion(ticket,Error(Core.Effect_failed "synthetic offline")))) | _ -> None) downloading.effects |> Option.get in let failed = Core.step downloading.next failure in @@ -5649,3 +5609,190 @@ let selected_encrypted_download_retains_crypto_and_failure () = ;; let scenarios=scenarios @ [Alcotest.test_case "selected encrypted download retains key validation and bounded failure" `Quick selected_encrypted_download_retains_crypto_and_failure] ;; + +(* The stream is live: a write after metadata may already be present in the + artifact. Core must still replay every journal position after metadata t, + and only durable application may advance that cursor. *) +let snapshot_live_stream_catchup_keeps_metadata_cursor () = + let opened, connection, sync = snapshot_catchup_fixture () in + Alcotest.(check bool) + "initial catchup starts at metadata t" + true + (pull_sends opened.effects = [ connection, Some (server_cursor_of_int 42) ]); + let hello = + pull_message + opened.next + connection + (Protocol.Server.Hello { t = server_cursor_of_int 44; checksum = None }) + in + let changed = + pull_message + hello.next + connection + (Protocol.Server.Changed { t = server_cursor_of_int 45 }) + in + check_pull_count "head changes join the in-flight catchup" 0 changed.effects; + Alcotest.(check (option cursor_testable)) + "later live stream writes cannot skip journal positions" + (Some (server_cursor_of_int 42)) + (Core.state changed.next).snapshot.applied_server_t; + let received = pull_response changed.next connection ~after:42 44 in + let batches = + List.filter_map + (function + | Core.Delegate (Core.Apply_authoritative_batch batch) -> Some batch + | _ -> None) + received.effects + in + let batch = + match batches with + | [ batch ] -> batch + | _ -> Alcotest.fail "one catchup batch" + in + Alcotest.(check (list int64)) + "every post-metadata transaction reaches the owner in order" + [ 43L; 44L ] + (List.map + (fun tx -> + Overlay.authoritative_transaction_cursor tx + |> Logseq_db_types.Server_cursor.to_int64) + (Overlay.authoritative_batch_transactions batch.input)); + Alcotest.(check (option cursor_testable)) + "delegation is not a durable commit" + (Some (server_cursor_of_int 42)) + (Core.state received.next).snapshot.applied_server_t; + let committed = + recovery_apply received.next connection (recovery_sync sync ~checkpoint:44 1 []) + in + Alcotest.(check bool) + "next catchup starts at committed t, including concurrent write" + true + (pull_sends committed.effects = [ connection, Some (server_cursor_of_int 44) ]); + let final = pull_response committed.next connection ~after:44 45 in + let committed = + recovery_apply final.next connection (recovery_sync sync ~checkpoint:45 2 []) + in + Alcotest.(check (option cursor_testable)) + "all concurrent writes are durably acknowledged" + (Some (server_cursor_of_int 45)) + (Core.state committed.next).snapshot.applied_server_t; + check_pull_count "caught up has no redundant pull" 0 committed.effects; + let repeated = pull_response committed.next connection ~after:42 45 in + Alcotest.(check bool) + "repeated catchup cannot reapply committed transactions" + false + (List.exists + (function + | Core.Delegate (Core.Apply_authoritative_batch _) -> true + | _ -> false) + repeated.effects) +;; + +let snapshot_failure_and_retry_keep_cursor_uncommitted () = + List.iter + (fun activation_failure -> + let requested, scope = begin_snapshot_bootstrap () in + let download = + List.find_map + (function + | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> + Some + (Core.step + requested.next + (Core.Runner_completed + (Core.Completion + ( ticket + , Ok + {|{"ok":true,"t":42,"key":"stream/graph.snapshot","url":"https://example.com/snapshot","content-encoding":"gzip"}|} + )))) + | _ -> None) + requested.effects + |> Option.get + in + let failed, late_completion = + List.find_map + (function + | Core.Run (Core.Request (ticket, Core.Download_snapshot _)) -> + let artifact = + Core.staged_artifact + ~id:"synthetic-failure" + ~scope + ~path:"/tmp/synthetic-snapshot" + ~expected_rows:1 + in + let completion = + Core.Runner_completed (Core.Completion (ticket, Ok artifact)) + in + let failed = + if activation_failure + then ( + let activating = Core.step download.next completion in + Core.step + activating.next + (Core.Snapshot_activation_failed + { scope = Core.effect_scope_of_graph scope + ; message = "synthetic activation failure" + })) + else + Core.step + download.next + (Core.Runner_completed + (Core.Completion + (ticket, Error (Core.Effect_failed "synthetic stream failure")))) + in + Some (failed, completion) + | _ -> None) + download.effects + |> Option.get + in + Alcotest.(check (option cursor_testable)) + "failed bootstrap cannot publish a cursor" + None + (Core.state failed.next).snapshot.applied_server_t; + Alcotest.(check bool) + "failure stays visible" + true + ((Core.state failed.next).snapshot.startup.failure = Some Core.During_bootstrap); + let stale = Core.step failed.next late_completion in + check_instructions + "completed download ticket cannot be consumed twice" + [] + stale.effects; + let retry = + Core.step + stale.next + (Core.Online_recovery_requested (timeline_ticket stale.next)) + in + let mirror = inspect_mirror_request retry.effects in + let missing = + Core.step retry.next (Core.Mirror_inspected (Core.Mirror_absent mirror.scope)) + in + Alcotest.(check bool) + "retry obtains fresh metadata without a preliminary pull" + true + (List.exists + (function + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) -> true + | _ -> false) + missing.effects + && not + (List.exists + (function + | Core.Run (Core.Request (_, Core.Fetch_snapshot_baseline _)) -> true + | _ -> false) + missing.effects))) + [ false; true ] +;; + +let scenarios = + scenarios + @ [ Alcotest.test_case + "live snapshot catchup preserves metadata cursor" + `Quick + snapshot_live_stream_catchup_keeps_metadata_cursor + ; Alcotest.test_case + "snapshot stream and activation failure retry" + `Quick + snapshot_failure_and_retry_keep_cursor_uncommitted + ] +;; diff --git a/logseq_sync/test/transport_contract.ml b/logseq_sync/test/transport_contract.ml index e3de61d..0d25aa5 100644 --- a/logseq_sync/test/transport_contract.ml +++ b/logseq_sync/test/transport_contract.ml @@ -237,31 +237,18 @@ let operation ?artifact_uri ~download origin = then List.find_map (function - | Core.Run (Core.Request (_, Core.Fetch_snapshot_baseline _) as op) -> Some op + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _) as op) -> Some op | _ -> None) transition.effects |> Option.get else ( - let metadata = - List.find_map - (function - | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> - Some - (Core.step - transition.next - (Core.Runner_completed - (Core.Completion (ticket, Ok {|{"type":"pull/ok","t":42}|})))) - | _ -> None) - transition.effects - |> Option.get - in let artifact_uri = Option.value artifact_uri ~default:(Uri.with_path origin "/artifact" |> Uri.to_string) in let response = - Yojson.Basic.to_string (`Assoc [ "ok", `Bool true; "url", `String artifact_uri ]) + Yojson.Basic.to_string (`Assoc [ "ok", `Bool true; "t", `Int 42; "url", `String artifact_uri ]) in let downloading = List.find_map @@ -269,10 +256,10 @@ let operation ?artifact_uri ~download origin = | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> Some (Core.step - metadata.next + transition.next (Core.Runner_completed (Core.Completion (ticket, Ok response)))) | _ -> None) - metadata.effects + transition.effects |> Option.get in List.find_map @@ -2453,7 +2440,7 @@ let submitted_authentication_policy let runnable = List.find_map (function - | Core.Run (Core.Request (_, Core.Fetch_snapshot_baseline _) as runnable) + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _) as runnable) -> Some runnable | _ -> None) requested.effects From 9105e9755cc5a8bb6ca8cd6ba762d3af5c65778b Mon Sep 17 00:00:00 2001 From: rcmerci Date: Wed, 21 Oct 2026 17:27:22 +0800 Subject: [PATCH 2/2] fix(sync): fence replaced snapshot completions --- .../2026-10-21-snapshot-download-cursor.md | 7 +- logseq_db_worker/test/test_effect_runner.ml | 4 +- logseq_db_worker/test/test_pure_reducer.ml | 11 +--- logseq_sync/lib/pure_reducer/core.ml | 13 ++++ logseq_sync/test/core_contract.ml | 66 ++++++++++++++++++- 5 files changed, 86 insertions(+), 15 deletions(-) diff --git a/docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md b/docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md index 749aaea..e55a061 100644 --- a/docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md +++ b/docs/agent-guide/implemented/bugfix/2026-10-21-snapshot-download-cursor.md @@ -2,7 +2,7 @@ ## Problem -Journal downloads issue GET /pull?since=0 only to obtain the initial journal cursor. Current Logseq obtains it from /snapshot/download metadata. +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 @@ -27,9 +27,10 @@ Rejected: live rows may include later writes without representing every interven - 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 server writes, graph resets, signing, Simulator or release artifact mutations. +- No real server writes or graph/account resets. Simulator verification uses an isolated synthetic fixture; release artifacts remain untouched. ## Consequences @@ -37,4 +38,4 @@ Rejected: live rows may include later writes without representing every interven ## Questions -None outstanding. The user authorized code alignment and verification, prohibited a preliminary pull, and scoped out publishing and unrelated repairs. +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. diff --git a/logseq_db_worker/test/test_effect_runner.ml b/logseq_db_worker/test/test_effect_runner.ml index ca8015a..e699c84 100644 --- a/logseq_db_worker/test/test_effect_runner.ml +++ b/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 diff --git a/logseq_db_worker/test/test_pure_reducer.ml b/logseq_db_worker/test/test_pure_reducer.ml index 5fb0c4a..81be3ad 100644 --- a/logseq_db_worker/test/test_pure_reducer.ml +++ b/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 diff --git a/logseq_sync/lib/pure_reducer/core.ml b/logseq_sync/lib/pure_reducer/core.ml index 927224a..a3748b8 100644 --- a/logseq_sync/lib/pure_reducer/core.ml +++ b/logseq_sync/lib/pure_reducer/core.ml @@ -2373,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 -> @@ -2698,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 diff --git a/logseq_sync/test/core_contract.ml b/logseq_sync/test/core_contract.ml index d56f108..301a4f7 100644 --- a/logseq_sync/test/core_contract.ml +++ b/logseq_sync/test/core_contract.ml @@ -982,7 +982,7 @@ let encrypted_graph_recovery_fetches_and_unlocks_key_before_bootstrap () = bootstrapping.effects) ;; -let begin_snapshot_bootstrap () = +let begin_snapshot_bootstrap ?(graphs = [ graph ]) () = let authenticated = Core.step (initial ()) (Core.Account_authenticated { user_id = Some "user" }) in @@ -993,7 +993,7 @@ let begin_snapshot_bootstrap () = Some (Core.step authenticated.next - (Core.Runner_completed (Core.Completion (ticket, Ok [ graph ])))) + (Core.Runner_completed (Core.Completion (ticket, Ok graphs)))) | Run _ | Delegate _ | Publish _ -> None) authenticated.effects |> Option.get @@ -5796,3 +5796,65 @@ let scenarios = snapshot_failure_and_retry_keep_cursor_uncommitted ] ;; + +let replaced_snapshot_completions_are_inert () = + let other = { graph with graph_id = + Logseq_db_types.Graph_types.Uuid.of_string "22222222-2222-4222-8222-222222222222" + |> Result.get_ok } in + List.iter (fun download_pending -> + let requested, old_scope = begin_snapshot_bootstrap ~graphs:[graph; other] () in + let metadata (result : (Core.snapshot_metadata, Core.effect_error) result) = List.find_map (function + | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> + Some (Core.Runner_completed (Core.Completion (ticket, result))) + | _ -> None) requested.effects |> Option.get in + let old_owner, successes, failure = + if download_pending then ( + let downloading = Core.step requested.next + (metadata (Ok {|{"ok":true,"t":42,"url":"https://example.com/old"}|})) in + List.find_map (function + | Core.Run (Core.Request (ticket, Core.Download_snapshot _)) -> + let artifact = Core.staged_artifact ~id:"old" ~scope:old_scope + ~path:"/tmp/synthetic-old" ~expected_rows:1 in + Some (downloading.next, + [Core.Runner_completed (Core.Completion (ticket, Ok artifact))], + Core.Runner_completed (Core.Completion + (ticket, Error (Core.Effect_failed "obsolete stream")))) + | _ -> None) downloading.effects |> Option.get) + else requested.next, + [metadata (Ok {|{"ok":true,"t":42,"url":"https://example.com/old"}|}); + metadata (Ok {|{"ok":true,"url":"https://example.com/invalid"}|})], + metadata (Error (Core.Effect_failed "obsolete metadata")) in + let selected = Core.step old_owner (Core.Graph_selected other.graph_id) in + let inspection = inspect_mirror_request selected.effects in + let current = Core.step selected.next + (Core.Mirror_inspected (Core.Mirror_absent inspection.scope)) in + List.iter (fun completion -> + let late = Core.step current.next completion in + check_instructions "replaced snapshot completion has no effects" [] late.effects; + check_observed "replaced snapshot" "current owner is unchanged" + (observe current.next) (observe late.next); + let fresh = List.find_map (function + | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_metadata _)) -> + Some (Core.step late.next (Core.Runner_completed (Core.Completion + (ticket, Ok {|{"ok":true,"t":84,"url":"https://example.com/current"}|})))) + | _ -> None) current.effects |> Option.get in + List.find_map (function + | Core.Run (Core.Request (ticket, Core.Download_snapshot request)) -> + Alcotest.(check bool) "fresh download retains graph and URL" true + (request.scope = inspection.scope && request.uri = Uri.of_string "https://example.com/current"); + let artifact = Core.staged_artifact ~id:"current" ~scope:inspection.scope + ~path:"/tmp/synthetic-current" ~expected_rows:1 in + Some (Core.step fresh.next (Core.Runner_completed + (Core.Completion (ticket, Ok artifact)))) + | _ -> None) fresh.effects |> Option.get |> fun activation -> + Alcotest.(check bool) "fresh activation keeps its own graph and cursor" true + (List.exists (function + | Core.Delegate (Core.Activate_snapshot request) -> + request.scope = inspection.scope && request.applied_server_t = server_cursor_of_int 84 + | _ -> false) activation.effects)) (successes @ [failure])) [false; true] +;; + +let scenarios = scenarios @ [ Alcotest.test_case + "replaced snapshot metadata and download completions are inert" `Quick + replaced_snapshot_completions_are_inert ] +;;