Fix native sync reconciliation latency

shared/src/logseq/core/cache_model.ml

@@ -739,16 +739,34 @@ 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")
+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
+    ["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) ())
+    |> blocks_from_sync_datoms model
+
 let unsynced_blocks model =
-  all_blocks model
-  |> List.filter (fun block -> block.sync_status <> "synced")
-  |> List.sort compare_recent
+  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 (

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) =

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

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);

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 ())

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"

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";

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 =

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

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")))

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);

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"

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 () =
@@ -4321,7 +4327,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