diff --git a/shared/src/logseq/core/cache_model.ml b/shared/src/logseq/core/cache_model.ml index be33c67b..7dac3535 100644 --- a/shared/src/logseq/core/cache_model.ml +++ b/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 ( 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..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 () = @@ -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