From f6b775f1af3bc391e9542e9b3e5147e38679d0a0 Mon Sep 17 00:00:00 2001 From: Tienson Qin Date: Mon, 9 Nov 2026 05:06:33 -0800 Subject: [PATCH 1/2] Fix native sync reconciliation latency Query only outstanding local UUIDs in the confirmed graph and skip empty reconciliation and pending pumps. Use indexed local sync statuses so settled captures cannot cause graph-sized work on the shared core executor. --- shared/src/logseq/core/cache_model.ml | 26 ++++++++---- shared/src/logseq/core/mobile_graph.ml | 8 +++- shared/src/logseq/core/mobile_graph.mli | 2 +- shared/src/logseq/core/mobile_session.ml | 2 +- shared/src/logseq/core/pending_pump.ml | 4 +- shared/src/logseq/core/rpc_ops.ml | 7 ++- shared/src/logseq/core/rpc_session.ml | 9 ++-- shared/src/logseq/core/session_types.ml | 4 +- shared/src/logseq/core/session_types.mli | 2 +- shared/src/logseq/entry/logseq_asset_e2e.ml | 2 +- shared/test/logseq/autosave_test.ml | 4 +- shared/test/logseq/mobile_graph_test.ml | 47 ++++++++++++++++++++- shared/test/logseq/rpc_test.ml | 2 +- 13 files changed, 88 insertions(+), 31 deletions(-) diff --git a/shared/src/logseq/core/cache_model.ml b/shared/src/logseq/core/cache_model.ml index be33c67b..60c71a56 100644 --- a/shared/src/logseq/core/cache_model.ml +++ b/shared/src/logseq/core/cache_model.ml @@ -739,16 +739,26 @@ let cache_local_child model uuid title parent_id now = Ok () | None -> Error ("unknown parent block: " ^ parent_id) -let pending_blocks model = - all_blocks model - |> List.filter (fun block -> - block.sync_status = "pending" || block.sync_status = "failed") - |> List.sort compare_recent +let blocks_with_sync_statuses model statuses = + if Seq.is_empty (Ds.Db.datoms model.db Ds.Aevt ~a:block_sync_status.attr_name ()) + then [] + else + List.concat_map (fun status -> + Ds.Db.datoms model.db Ds.Avet + ~a:block_sync_status.attr_name ~v:(Ds.String status) () + |> Seq.filter_map (fun (datom : Ds.datom) -> + Ds.Db.datoms model.db Ds.Eavt ~e:datom.e ~a:block_uuid.attr_name () + |> Seq.find_map (fun (datom : Ds.datom) -> + match datom.v with + | Ds.String uuid -> read_block model uuid + | _ -> None)) + |> List.of_seq) statuses + |> List.sort compare_recent + +let pending_blocks model = blocks_with_sync_statuses model ["pending"; "failed"] let unsynced_blocks model = - all_blocks model - |> List.filter (fun block -> block.sync_status <> "synced") - |> List.sort compare_recent + blocks_with_sync_statuses model ["pending"; "failed"; "submitted"] let update_sync_status model uuid sync_status = if block_exists model uuid then ( diff --git a/shared/src/logseq/core/mobile_graph.ml b/shared/src/logseq/core/mobile_graph.ml index 549e2289..d42ab0df 100644 --- a/shared/src/logseq/core/mobile_graph.ml +++ b/shared/src/logseq/core/mobile_graph.ml @@ -60,10 +60,14 @@ let empty_journals (host : mobile_graph) = | Some opened -> Runtime.empty_journals opened.read_runtime | None -> [] -let authoritative_blocks (host : mobile_graph) = +let authoritative_blocks (host : mobile_graph) uuids = match !(host.current) with | Some opened -> - Some (Read.blocks (fun v -> Ok v) 7 (Ds.conn_db opened.conn)) + let db = Ds.conn_db opened.conn in + Some (List.filter_map (fun uuid -> + Option.bind (Ds.entid db "block/uuid" (Ds.Uuid uuid)) (fun eid -> + if Read.page_is_hidden db eid then None + else Read.block ~with_breadcrumbs:false (fun value -> Ok value) db eid)) uuids) | None -> None let sidebar_pages (host : mobile_graph) = diff --git a/shared/src/logseq/core/mobile_graph.mli b/shared/src/logseq/core/mobile_graph.mli index 159a9a6c..3c9637d0 100644 --- a/shared/src/logseq/core/mobile_graph.mli +++ b/shared/src/logseq/core/mobile_graph.mli @@ -30,7 +30,7 @@ val resolve_asset_path : mobile_graph -> string -> string val sync_cursor : mobile_graph -> int option val blocks : mobile_graph -> Cache_model.block list option val projected_db : mobile_graph -> Datascript.db option -val authoritative_blocks : mobile_graph -> Cache_model.block list option +val authoritative_blocks : mobile_graph -> string list -> Cache_model.block list option val sidebar_pages : mobile_graph -> Graph_read.sidebar_pages option val tag_pages : mobile_graph -> Cache_model.entity_summary list option val node_is_tag : mobile_graph -> string -> bool diff --git a/shared/src/logseq/core/mobile_session.ml b/shared/src/logseq/core/mobile_session.ml index a390658c..d5f4ed58 100644 --- a/shared/src/logseq/core/mobile_session.ml +++ b/shared/src/logseq/core/mobile_session.ml @@ -67,7 +67,7 @@ let host_options (database : Database.mobile_database) ring catalog = empty_journals = Some (fun () -> Graph.empty_journals graph); graph_db = Some (fun () -> Graph.projected_db graph); authoritative_graph_blocks = - Some (fun () -> Graph.authoritative_blocks graph); + Some (Graph.authoritative_blocks graph); graph_sidebar_pages = Some (fun () -> Graph.sidebar_pages graph); graph_tag_pages = Some (fun () -> Graph.tag_pages graph); graph_node_is_tag = Some (fun uuid -> Graph.node_is_tag graph uuid); diff --git a/shared/src/logseq/core/pending_pump.ml b/shared/src/logseq/core/pending_pump.ml index d070b2e8..ba4c5b7c 100644 --- a/shared/src/logseq/core/pending_pump.ml +++ b/shared/src/logseq/core/pending_pump.ml @@ -348,11 +348,11 @@ let begin_pending_sync (session : Types.session) (config : Api.api_config) = let assets = List.filter (fun b -> b.Model.is_asset) pending in if assets = [] then activate_semantic_request session config None; let s = Types.state session in - if s.semantic_active = None && s.pending_sync = None then begin + if s.semantic_active = None && s.pending_sync = None && pending <> [] then begin let authoritative = Hashtbl.create 64 in (match (Types.host session).authoritative_graph_blocks with | Some load -> - (match load () with + (match load (List.map (fun (block : Model.block) -> block.uuid) pending) with | Some blocks -> List.iter (fun (b : Model.block) -> Hashtbl.replace authoritative b.uuid ()) diff --git a/shared/src/logseq/core/rpc_ops.ml b/shared/src/logseq/core/rpc_ops.ml index a1065bc1..15d43831 100644 --- a/shared/src/logseq/core/rpc_ops.ml +++ b/shared/src/logseq/core/rpc_ops.ml @@ -26,12 +26,11 @@ let same_pending_version (left : Model.block) (right : Model.block) = && left.local_path = right.local_path let reconcile_authoritative_blocks cache blocks = - let by_uuid = - List.map (fun (block : Model.block) -> (block.uuid, block)) blocks - in + let by_uuid = Hashtbl.create (List.length blocks) in + List.iter (fun (block : Model.block) -> Hashtbl.replace by_uuid block.uuid block) blocks; List.iter (fun (block : Model.block) -> - match List.assoc_opt block.uuid by_uuid with + match Hashtbl.find_opt by_uuid block.uuid with | Some (authoritative : Model.block) -> if block.sync_status = "submitted" diff --git a/shared/src/logseq/core/rpc_session.ml b/shared/src/logseq/core/rpc_session.ml index 0f67959b..8d00a908 100644 --- a/shared/src/logseq/core/rpc_session.ml +++ b/shared/src/logseq/core/rpc_session.ml @@ -634,13 +634,14 @@ let pending_sync_patch session = ]) let reconcile_authoritative_blocks session = - match (host session).Types.authoritative_graph_blocks with - | Some load -> - (match load () with + match (Model.unsynced_blocks (state session).model, + (host session).Types.authoritative_graph_blocks) with + | [], _ | _, None -> () + | pending, Some load -> + (match load (List.map (fun (block : Model.block) -> block.uuid) pending) with | Some blocks -> Rpc.reconcile_authoritative_blocks (state session).model blocks | None -> ()) - | None -> () let discover_graphs session (config : Api.api_config) = debug "graph discovery started"; diff --git a/shared/src/logseq/core/session_types.ml b/shared/src/logseq/core/session_types.ml index d65d025a..1ef46405 100644 --- a/shared/src/logseq/core/session_types.ml +++ b/shared/src/logseq/core/session_types.ml @@ -86,7 +86,7 @@ type host_options = ; sync_cursor : (unit -> int option) option ; graph_blocks : (unit -> Model.block list option) option ; graph_db : (unit -> Datascript.db option) option - ; authoritative_graph_blocks : (unit -> Model.block list option) option + ; authoritative_graph_blocks : (string list -> Model.block list option) option ; graph_sidebar_pages : (unit -> Graph.sidebar_pages option) option ; graph_tag_pages : (unit -> Model.entity_summary list option) option ; graph_node_is_tag : (string -> bool) option @@ -285,7 +285,7 @@ let create_session options = authoritative_graph_blocks = (match options.authoritative_graph_blocks with | Some _ -> options.authoritative_graph_blocks - | None -> options.graph_blocks); + | None -> Option.map (fun load _uuids -> load ()) options.graph_blocks); } in let catalog = diff --git a/shared/src/logseq/core/session_types.mli b/shared/src/logseq/core/session_types.mli index a8f17c21..a433de2e 100644 --- a/shared/src/logseq/core/session_types.mli +++ b/shared/src/logseq/core/session_types.mli @@ -79,7 +79,7 @@ type host_options = ; sync_cursor : (unit -> int option) option ; graph_blocks : (unit -> Cache_model.block list option) option ; graph_db : (unit -> Datascript.db option) option - ; authoritative_graph_blocks : (unit -> Cache_model.block list option) option + ; authoritative_graph_blocks : (string list -> Cache_model.block list option) option ; graph_sidebar_pages : (unit -> Graph_read.sidebar_pages option) option ; graph_tag_pages : (unit -> Cache_model.entity_summary list option) option ; graph_node_is_tag : (string -> bool) option diff --git a/shared/src/logseq/entry/logseq_asset_e2e.ml b/shared/src/logseq/entry/logseq_asset_e2e.ml index 8d293b7c..9cae1ab5 100644 --- a/shared/src/logseq/entry/logseq_asset_e2e.ml +++ b/shared/src/logseq/entry/logseq_asset_e2e.ml @@ -79,7 +79,7 @@ let () = let downloaded = Live.download_asset cfg uuid "png" in let actual = if encrypted then Live.require_ok "decrypt uploaded asset" (E2ee.decrypt_bytes crypto key downloaded.body) else downloaded.body in if actual <> bytes then failwith "Uploaded image bytes differ"; - let blocks = Option.get (Mobile_graph.authoritative_blocks host.database.graph) in + let blocks = Option.get (Mobile_graph.authoritative_blocks host.database.graph [uuid]) in if not (List.exists (fun (b : Cache_model.block) -> b.uuid = uuid && b.is_asset) blocks) then failwith "Asset never reached authoritative graph"; print_endline "PASS: real mobile capture, pending upload, datom sync and journal replay"))) diff --git a/shared/test/logseq/autosave_test.ml b/shared/test/logseq/autosave_test.ml index 0502f5af..f2bb7bf0 100644 --- a/shared/test/logseq/autosave_test.ml +++ b/shared/test/logseq/autosave_test.ml @@ -23,7 +23,7 @@ let with_editor run = let session = Rpc_session.create_session { Rpc_session.default_options with graph_blocks = Some (fun () -> Some [!saved]); - authoritative_graph_blocks = Some (fun () -> Some [!saved]); + authoritative_graph_blocks = Some (fun _uuids -> Some [!saved]); sync_cursor = Some (fun () -> Some 41); stage_operation = Some (fun operation -> (match operation.Pending_ops.intent with @@ -283,7 +283,7 @@ let rapid_editing_and_reparenting_settles_after_the_last_save () = graph_blocks = Some (fun () -> Some (R.blocks_for_page runtime page)); graph_page_blocks = Some (fun uuid -> Some (R.blocks_for_page runtime uuid)); graph_sidebar_pages = Some (fun () -> Some (R.sidebar_pages runtime)); - authoritative_graph_blocks = Some (fun () -> Some (Graph_read.blocks_for_page (fun value -> Ok value) (Ds.conn_db conn) page)); + authoritative_graph_blocks = Some (fun _uuids -> Some (Graph_read.blocks_for_page (fun value -> Ok value) (Ds.conn_db conn) page)); sync_cursor = Some (fun () -> Some (R.state runtime).server_t); stage_operation = Some (R.stage runtime); prepare_operation = Some (R.prepare_sync runtime); diff --git a/shared/test/logseq/mobile_graph_test.ml b/shared/test/logseq/mobile_graph_test.ml index fae071ec..8558cfef 100644 --- a/shared/test/logseq/mobile_graph_test.ml +++ b/shared/test/logseq/mobile_graph_test.ml @@ -99,7 +99,7 @@ let unopened_host_preserves_query_and_command_defaults () = let host = host () in check_eq (Mobile.sync_cursor host) None; check_eq (Mobile.blocks host) None; - check_eq (Mobile.authoritative_blocks host) None; + check_eq (Mobile.authoritative_blocks host ["missing"]) None; check_eq (Mobile.sidebar_pages host) None; check_eq (Mobile.tag_pages host) None; check_eq (Mobile.blocks_for_page host "missing") None; @@ -257,7 +257,7 @@ let opened_host_reads_live_runtime_and_observes_close () = let current = opened.read_runtime in check_eq (Mobile.sync_cursor host) (Some 7); check_eq (Mobile.blocks host) (Some (Runtime.blocks current)); - check_eq (Mobile.authoritative_blocks host) (Some []); + check_eq (Mobile.authoritative_blocks host ["missing"]) (Some []); check_eq (Mobile.sidebar_pages host) (Some (Runtime.sidebar_pages current)); @@ -438,6 +438,47 @@ let sync_events_persist_cursor_and_rebase_the_read_runtime () = check_eq (Sync.applied_server_t opened.state) 9; check_eq (Runtime.state opened.read_runtime).server_t 9) +let authoritative_confirmation_is_not_limited_to_journals () = + with_directory (fun dir -> + prepare_graph dir; + let host = host () in + expect_ok (Mobile.open_graph host (open_payload dir false)); + let opened = expect_opened !(host.current) in + let page = "00000000-0000-0000-0000-000000000001" in + let submitted = "00000000-0000-0000-0000-000000000002" in + let pending = "00000000-0000-0000-0000-000000000003" in + ignore (Ds.transact_conn opened.conn + [ Ds.Add (Ds.Entity_id 1, "block/uuid", Ds.Uuid page) + ; Ds.Add (Ds.Entity_id 1, "block/name", Ds.String "notes") + ; Ds.Add (Ds.Entity_id 1, "block/title", Ds.String "Notes") + ; Ds.Add (Ds.Entity_id 2, "block/uuid", Ds.Uuid submitted) + ; Ds.Add (Ds.Entity_id 2, "block/title", Ds.String "Remote title") + ; Ds.Add (Ds.Entity_id 2, "block/page", Ds.Ref 1) + ; Ds.Add (Ds.Entity_id 2, "block/parent", Ds.Ref 1) + ; Ds.Add (Ds.Entity_id 3, "block/uuid", Ds.Uuid pending) + ; Ds.Add (Ds.Entity_id 3, "block/title", Ds.String "Remote title") + ; Ds.Add (Ds.Entity_id 3, "block/page", Ds.Ref 1) + ; Ds.Add (Ds.Entity_id 3, "block/parent", Ds.Ref 1) ]); + let session = Rpc_session.create_session + { Rpc_session.default_options with + authoritative_graph_blocks = Some (Mobile.authoritative_blocks host); + apply_sync_event = Some (Mobile.apply_sync_event host) } in + let model = (Rpc_session.state session).model in + Model.upsert_blocks model + [ { (Model.local_block submitted "Local title" page (Some page) 1) with + sync_status = "submitted" } + ; Model.local_block pending "Unsent title" page (Some page) 1 ] 1; + let response = Rpc_session.call session (json_object + [ "apiVersion", `Int 1; "method", `String "dispatch"; + "params", `Assoc ["action", `String "applySyncEvent"; + "payload", `String (change_event 7 8)] ]) in + check_eq (Yojson.Basic.Util.member "ok" (Json.from_string response)) (`Bool true); + check_eq (Option.map (fun (block : Model.block) -> block.sync_status) + (Model.read_block model submitted)) (Some "synced"); + check_eq (Option.map (fun (block : Model.block) -> block.sync_status) + (Model.read_block model pending)) (Some "pending"); + check_eq (Mobile.sync_cursor host) (Some 8)) + let snapshot_import_activates_a_real_snapshot_and_reopens_it () = with_directory (fun dir -> let payload = snapshot_request dir false (database ()) in @@ -557,6 +598,8 @@ let cases = reset_and_invalid_events_do_not_require_an_open_graph; case "sync events persist cursor and rebase the read runtime" sync_events_persist_cursor_and_rebase_the_read_runtime; + case "authoritative confirmation is not limited to journals" + authoritative_confirmation_is_not_limited_to_journals; case "snapshot import activates a real snapshot and reopens it" snapshot_import_activates_a_real_snapshot_and_reopens_it; case "encrypted import decrypts before opening the local graph" diff --git a/shared/test/logseq/rpc_test.ml b/shared/test/logseq/rpc_test.ml index 4459f0bf..a33e4fd2 100644 --- a/shared/test/logseq/rpc_test.ml +++ b/shared/test/logseq/rpc_test.ml @@ -4321,7 +4321,7 @@ let plain_assets_project_before_upload_and_queue_stable_datom_transactions () = load_graph_catalog = Some (fun () -> Some plain_graph_catalog); sync_cursor = Some (fun () -> Some 41); graph_blocks = Some (fun () -> Some !projected); - authoritative_graph_blocks = Some (fun () -> Some [ target ]); + authoritative_graph_blocks = Some (fun _uuids -> Some [ target ]); journal_page_id = Some (fun _ -> Some "journal-page"); stage_operation = Some From 103cd0122d3c7bea19fb40a8c50129280cc09759 Mon Sep 17 00:00:00 2001 From: Tienson Qin Date: Mon, 9 Nov 2026 05:15:36 -0800 Subject: [PATCH 2/2] Preserve all unsynced states during reconciliation Filter indexed sync-status metadata before reading eligible blocks, retaining confirmation for arbitrary non-synced statuses without materializing settled records. Extend the reconciliation contract to cover retrying states. --- shared/src/logseq/core/cache_model.ml | 40 ++++++++++++++++----------- shared/test/logseq/rpc_test.ml | 8 +++++- 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/shared/src/logseq/core/cache_model.ml b/shared/src/logseq/core/cache_model.ml index 60c71a56..7dac3535 100644 --- a/shared/src/logseq/core/cache_model.ml +++ b/shared/src/logseq/core/cache_model.ml @@ -739,26 +739,34 @@ let cache_local_child model uuid title parent_id now = Ok () | None -> Error ("unknown parent block: " ^ parent_id) -let blocks_with_sync_statuses model statuses = - if Seq.is_empty (Ds.Db.datoms model.db Ds.Aevt ~a:block_sync_status.attr_name ()) - then [] +let blocks_from_sync_datoms model datoms = + datoms + |> Seq.filter_map (fun (datom : Ds.datom) -> + Ds.Db.datoms model.db Ds.Eavt ~e:datom.e ~a:block_uuid.attr_name () + |> Seq.find_map (fun (datom : Ds.datom) -> + match datom.v with + | Ds.String uuid -> read_block model uuid + | _ -> None)) + |> List.of_seq + |> List.sort compare_recent + +let sync_status_datoms model = + Ds.Db.datoms model.db Ds.Aevt ~a:block_sync_status.attr_name () + +let pending_blocks model = + if Seq.is_empty (sync_status_datoms model) then [] else - List.concat_map (fun status -> + ["pending"; "failed"] |> List.to_seq + |> Seq.flat_map (fun status -> Ds.Db.datoms model.db Ds.Avet - ~a:block_sync_status.attr_name ~v:(Ds.String status) () - |> Seq.filter_map (fun (datom : Ds.datom) -> - Ds.Db.datoms model.db Ds.Eavt ~e:datom.e ~a:block_uuid.attr_name () - |> Seq.find_map (fun (datom : Ds.datom) -> - match datom.v with - | Ds.String uuid -> read_block model uuid - | _ -> None)) - |> List.of_seq) statuses - |> List.sort compare_recent - -let pending_blocks model = blocks_with_sync_statuses model ["pending"; "failed"] + ~a:block_sync_status.attr_name ~v:(Ds.String status) ()) + |> blocks_from_sync_datoms model let unsynced_blocks model = - blocks_with_sync_statuses model ["pending"; "failed"; "submitted"] + sync_status_datoms model + |> Seq.filter (fun (datom : Ds.datom) -> + match datom.v with Ds.String status -> status <> "synced" | _ -> false) + |> blocks_from_sync_datoms model let update_sync_status model uuid sync_status = if block_exists model uuid then ( diff --git a/shared/test/logseq/rpc_test.ml b/shared/test/logseq/rpc_test.ml index a33e4fd2..83acb25e 100644 --- a/shared/test/logseq/rpc_test.ml +++ b/shared/test/logseq/rpc_test.ml @@ -390,12 +390,17 @@ let authoritative_reconciliation_preserves_newer_local_edits () = } in let missing = Model.local_block "missing" "Missing" "page" None 1 in - Model.upsert_blocks cache [ same; changed; submitted; missing ] 1; + let retrying = + { (Model.local_block "retrying" "Retrying" "page" None 1) with + Model.sync_status = "retrying" } + in + Model.upsert_blocks cache [ same; changed; submitted; missing; retrying ] 1; Rpc.reconcile_authoritative_blocks cache [ same; { changed with title = "Remote" }; { submitted with title = "Remote" }; + retrying; ]; List.iter (fun (uuid, expected) -> @@ -409,6 +414,7 @@ let authoritative_reconciliation_preserves_newer_local_edits () = ("changed", "pending"); ("submitted", "synced"); ("missing", "pending"); + ("retrying", "synced"); ] let node_navigation_preserves_independent_projections_and_editing () =