diff --git a/shared/src/logseq/core/graph_runtime.ml b/shared/src/logseq/core/graph_runtime.ml index 159ad97a..6a09103b 100644 --- a/shared/src/logseq/core/graph_runtime.ml +++ b/shared/src/logseq/core/graph_runtime.ml @@ -17,6 +17,7 @@ type runtime_options = type read_caches = { blocks : (int * int option * int, Model.block list) Hashtbl.t + ; journal_selection : ((int * int option * int) * string list) option ; page_blocks : (string, Model.block list) Hashtbl.t ; page_readers : (string, Graph_read.page_reader) Hashtbl.t ; node_blocks : (string, Model.block list) Hashtbl.t @@ -71,6 +72,7 @@ let db runtime = (state runtime).snapshot.db let new_read_caches () = { blocks = Hashtbl.create 16; + journal_selection = None; page_blocks = Hashtbl.create 64; page_readers = Hashtbl.create 32; node_blocks = Hashtbl.create 64; @@ -266,10 +268,38 @@ let apply_local_properties runtime database = Some (Projection.property_tx uuid attr value)) in if operations = [] then database else Ds.db_with operations database -let rebase runtime server_t operation_ids changed_uuids = +let preserve_journal_selection runtime before operations changed_attributes changed_uuids = + match changed_attributes, (state runtime).read_cache, (state runtime).journal_block_limit with + | Some attributes, Some (cached, caches), Some limit + when cached == before && limit <= initial_journal_block_limit + && List.for_all (fun attr -> attr = "block/title" || attr = "block/updated-at") attributes + && List.for_all (fun (op : Ops.pending_operation) -> + match op.intent with Ops.Save_title _ -> true | _ -> false) operations -> + let after = db runtime in + let title_present database uuid = + Option.map (fun eid -> match Read.value database eid "block/title" with + | Some (Ds.String _) -> true | _ -> false) + (Ds.entid database "block/uuid" (Ds.Uuid uuid)) in + let affected = changed_uuids @ List.concat_map + (fun (op : Ops.pending_operation) -> Ops.affected_uuids before op.intent) operations in + if List.for_all (fun uuid -> title_present before uuid = title_present after uuid) affected then begin + let through_day = Model.journal_day_for_ms (int_of_float (Unix.gettimeofday () *. 1000.)) in + let key = (state runtime).journal_limit, Some limit, through_day in + let selected = match Hashtbl.find_opt caches.blocks key with + | Some blocks -> Some (List.map (fun (block : Model.block) -> block.uuid) blocks) + | None -> Option.bind caches.journal_selection (fun (selected_key, uuids) -> + if selected_key = key then Some uuids else None) in + Option.iter (fun uuids -> + let caches = { (new_read_caches ()) with journal_selection = Some (key, uuids) } in + runtime.state := { !(runtime.state) with read_cache = Some (after, caches) }) selected + end + | _ -> () + +let rebase ?changed_attributes runtime server_t operation_ids changed_uuids = let started = Unix.gettimeofday () in let before = db runtime in let authoritative = Ds.conn_db runtime.conn in + let pending = Ops.list runtime.path in let operations = List.filter (fun (operation : Ops.pending_operation) -> @@ -281,7 +311,7 @@ let rebase runtime server_t operation_ids changed_uuids = false end else true) - (Ops.list runtime.path) + pending in let projected = ref (Dev_sync.Sync_replay.display_db_from_server authoritative) in trace_stage "LOGSEQ_REBASE_METRIC" started "confirmed"; @@ -307,6 +337,7 @@ let rebase runtime server_t operation_ids changed_uuids = }; }; trace_stage "LOGSEQ_REBASE_METRIC" started "projected"; + preserve_journal_selection runtime before pending changed_attributes changed_uuids; refresh_search_after_rebase runtime before operations changed_uuids; trace_stage "LOGSEQ_REBASE_METRIC" started "complete" @@ -600,6 +631,7 @@ let stage runtime (operation : Ops.pending_operation) = if operation.base_t <> s.server_t then Error "operation was created against a stale server cursor" else + let visible_before = db runtime in let before = match previous with | None -> db runtime | Some _ -> db_before_operation runtime operation.operation_id @@ -621,6 +653,15 @@ let stage runtime (operation : Ops.pending_operation) = { !(runtime.state) with snapshot = { s.snapshot with Projection.db = projected; statuses = if local_only then s.snapshot.statuses else s.snapshot.statuses @ [ operation.operation_id, Ops.Applied ] } }); + (match normalized.intent with + | Ops.Save_title _ -> + let attributes = List.fold_left (fun attributes -> function + | Ds.Add (_, attr, _) | Ds.Retract (_, attr, _) -> + Option.map (fun attributes -> attr :: attributes) attributes + | _ -> None) (Some []) operations in + preserve_journal_selection runtime visible_before (normalized :: existing) attributes + (Ops.affected_uuids before normalized.intent) + | _ -> ()); refresh_search_affected runtime before normalized.intent operations; (match previous with None -> refresh_split_page_cache runtime before normalized.intent | Some _ -> ()); Ok () @@ -673,7 +714,14 @@ let blocks runtime = | Some blocks -> blocks | None -> let blocks = - Read.blocks ?block_limit ~through_day (fun v -> Ok v) limit (db runtime) + match caches.journal_selection with + | Some (key, uuids) when key = (limit, block_limit, through_day) -> + let database = db runtime in + List.filter_map (fun uuid -> + Option.bind (Ds.entid database "block/uuid" (Ds.Uuid uuid)) + (Read.block ~with_breadcrumbs:false (fun value -> Ok value) database)) uuids + |> List.sort Read.compare_journal_blocks + | _ -> Read.blocks ?block_limit ~through_day (fun v -> Ok v) limit (db runtime) in Hashtbl.replace caches.blocks (limit, block_limit, through_day) blocks; blocks diff --git a/shared/src/logseq/core/graph_runtime.mli b/shared/src/logseq/core/graph_runtime.mli index 506b6101..46d99dd0 100644 --- a/shared/src/logseq/core/graph_runtime.mli +++ b/shared/src/logseq/core/graph_runtime.mli @@ -10,6 +10,7 @@ type runtime_options = type read_caches = { blocks : (int * int option * int, Cache_model.block list) Hashtbl.t + ; journal_selection : ((int * int option * int) * string list) option ; page_blocks : (string, Cache_model.block list) Hashtbl.t ; page_readers : (string, Graph_read.page_reader) Hashtbl.t ; node_blocks : (string, Cache_model.block list) Hashtbl.t @@ -77,7 +78,7 @@ val rebase_operation : Pending_ops.pending_operation -> Pending_ops.pending_state val rebase : - graph_runtime -> int -> string list -> string list -> unit + ?changed_attributes:string list -> graph_runtime -> int -> string list -> string list -> unit val create_base : string -> int -> Datascript.conn -> runtime_options -> graph_runtime val fresh_uuid : unit -> string diff --git a/shared/src/logseq/core/journal_sync.ml b/shared/src/logseq/core/journal_sync.ml index aa217f9b..faaee48b 100644 --- a/shared/src/logseq/core/journal_sync.ml +++ b/shared/src/logseq/core/journal_sync.ml @@ -5,6 +5,11 @@ module Wire = Dev.Wire type entry = { t : int; tx_id : string option; tx : Wire.t list } type journal = { t : int; t_before : int option; entries : entry list; graph_id : string option } +type applied = + { operation_ids : string list + ; changed_uuids : string list + ; changed_attributes : string list + } let non_negative name fields = match List.assoc_opt name fields with @@ -116,10 +121,12 @@ let encode encrypt tx = (map_protected encrypt items) with error -> Error (Printexc.to_string error) -let changed_uuids before after datoms = +let changes operation_ids before after datoms = let seen = Hashtbl.create 16 in + let attributes = Hashtbl.create 16 in List.iter (fun (d : Ds.datom) -> + Hashtbl.replace attributes d.a (); List.iter (fun db -> match Ds.entity db (Ds.Entity_id d.e) with @@ -130,12 +137,16 @@ let changed_uuids before after datoms = | None -> ()) [ before; after ]) datoms; - Hashtbl.fold (fun uuid () acc -> uuid :: acc) seen [] + { operation_ids + ; changed_uuids = Hashtbl.fold (fun uuid () acc -> uuid :: acc) seen [] + ; changed_attributes = Hashtbl.fold (fun attr () acc -> attr :: acc) attributes [] + } let apply decrypt conn local_t journal = try let ( let* ) = Result.bind in - if journal.t <= local_t then Ok ([], []) + if journal.t <= local_t then + Ok { operation_ids = []; changed_uuids = []; changed_attributes = [] } else if Option.fold ~none:false ~some:(fun before -> local_t < before) journal.t_before then Error "journal prefix is missing; pull before applying ACK" else @@ -166,5 +177,5 @@ let apply decrypt conn local_t journal = ignore (Ds.apply_report conn { Ds.db_before = before; db_after = after; tx_data = datoms; tempids = []; tx_meta = [] }); - Ok (ids, changed_uuids before after datoms) + Ok (changes ids before after datoms) with error -> Error (Printexc.to_string error) diff --git a/shared/src/logseq/core/journal_sync.mli b/shared/src/logseq/core/journal_sync.mli index 32a2e298..a27fd8de 100644 --- a/shared/src/logseq/core/journal_sync.mli +++ b/shared/src/logseq/core/journal_sync.mli @@ -2,10 +2,15 @@ type entry = { t : int; tx_id : string option; tx : Dev_sync.Wire.t list } type journal = { t : int; t_before : int option; entries : entry list; graph_id : string option } +type applied = + { operation_ids : string list + ; changed_uuids : string list + ; changed_attributes : string list + } val decode : string -> (journal, string) result val capture : ?creator:Created_by.identity -> Datascript.db -> Datascript.tx_op list -> Datascript.db * string * string val replay : Datascript.db -> string -> string -> string -> Datascript.db val encode : (string -> (string, string) result) -> string -> (string, string) result val apply : (string -> (string, string) result) -> Datascript.conn -> int -> journal -> - (string list * string list, string) result + (applied, string) result diff --git a/shared/src/logseq/core/live_sync.ml b/shared/src/logseq/core/live_sync.ml index 659a2160..67064e8a 100644 --- a/shared/src/logseq/core/live_sync.ml +++ b/shared/src/logseq/core/live_sync.ml @@ -134,10 +134,11 @@ let has_title db decrypt expected = let apply_journal decrypt current body = let journal = require_ok "decode journal" (Journal_sync.decode body) in - let ids, changed = require_ok "apply canonical journal" + let applied = require_ok "apply canonical journal" (Journal_sync.apply decrypt current.Graph_runtime.conn (Graph_runtime.state current).server_t journal) in - Graph_runtime.rebase current journal.t ids changed + Graph_runtime.rebase ~changed_attributes:applied.changed_attributes + current journal.t applied.operation_ids applied.changed_uuids let rec submit_tx ?(decrypt = fun value -> Ok value) cfg current operation attempts = if attempts <= 0 then diff --git a/shared/src/logseq/core/mobile_graph.ml b/shared/src/logseq/core/mobile_graph.ml index d42ab0df..42b14eac 100644 --- a/shared/src/logseq/core/mobile_graph.ml +++ b/shared/src/logseq/core/mobile_graph.ml @@ -331,7 +331,7 @@ let apply_sync_event (host : mobile_graph) body = Error "journal belongs to another graph" | Some opened -> let local_t = Sync.applied_server_t opened.state in - let* ids, changed = Journal_sync.apply + let* applied = Journal_sync.apply (if opened.e2ee then host.crypto.decrypt_title opened.graph_id else fun value -> Ok value) opened.conn local_t journal in if journal.t <= local_t then Ok () @@ -339,7 +339,8 @@ let apply_sync_event (host : mobile_graph) body = let* () = Checkpoint.save_checkpoint_atomic opened.checkpoint_path (Checkpoint.create opened.graph_id opened.state.schema_version journal.t) in opened.state.applied_server_t := journal.t; - Runtime.rebase opened.read_runtime journal.t ids changed; + Runtime.rebase ~changed_attributes:applied.changed_attributes + opened.read_runtime journal.t applied.operation_ids applied.changed_uuids; Ok () let search_page host ~offset ~limit query = diff --git a/shared/test/logseq/autosave_test.ml b/shared/test/logseq/autosave_test.ml index f2bb7bf0..2f16e38b 100644 --- a/shared/test/logseq/autosave_test.ml +++ b/shared/test/logseq/autosave_test.ml @@ -276,8 +276,9 @@ let rapid_editing_and_reparenting_settles_after_the_last_save () = let apply_journal body = let ( let* ) = Result.bind in let* journal = Journal_sync.decode body in - let* ids, changed = Journal_sync.apply (fun value -> Ok value) conn (R.state runtime).server_t journal in - R.rebase runtime journal.t ids changed; + let* applied = Journal_sync.apply (fun value -> Ok value) conn (R.state runtime).server_t journal in + R.rebase ~changed_attributes:applied.changed_attributes runtime journal.t + applied.operation_ids applied.changed_uuids; Ok () in let session = Rpc_session.create_session { Rpc_session.default_options with graph_blocks = Some (fun () -> Some (R.blocks_for_page runtime page)); diff --git a/shared/test/logseq/composer_tag_test.ml b/shared/test/logseq/composer_tag_test.ml index f9cbfb39..a9cd7478 100644 --- a/shared/test/logseq/composer_tag_test.ml +++ b/shared/test/logseq/composer_tag_test.ml @@ -28,9 +28,10 @@ let sync_all runtime conn = "tx-id", `String operation.Pending_ops.operation_id; "tx", `String tx]]])) with | Ok value -> value | Error message -> fail message in check_ok (Graph_runtime.stage runtime {operation with Pending_ops.state = Pending_ops.Accepted (before + 1)}); - let ids, changed = match Journal_sync.apply (fun value -> Ok value) conn before journal with + let applied = match Journal_sync.apply (fun value -> Ok value) conn before journal with | Ok value -> value | Error message -> fail message in - Graph_runtime.rebase runtime journal.t ids changed) (Graph_runtime.pending_operations runtime); + Graph_runtime.rebase ~changed_attributes:applied.changed_attributes runtime journal.t + applied.operation_ids applied.changed_uuids) (Graph_runtime.pending_operations runtime); check_eq ~msg:"tagged captures round-trip through confirmed sync" (Graph_runtime.pending_operations runtime) [] let fresh_journal_captures_preserve_the_note () = R.with_runtime (fun session runtime _ -> diff --git a/shared/test/logseq/graph_runtime_test.ml b/shared/test/logseq/graph_runtime_test.ml index dcfdf8e6..a482aa65 100644 --- a/shared/test/logseq/graph_runtime_test.ml +++ b/shared/test/logseq/graph_runtime_test.ml @@ -1839,9 +1839,10 @@ let reopening_repairs_legacy_page_creations_and_unblocks_dependent_changes () = "txs", `List [`Assoc ["t", `Int (before + 1); "tx-id", `String op.operation_id; "tx", `String tx]]])) with | Ok value -> value | Error message -> fail message in check (is_ok (Runtime.stage current { op with state = Pending.Accepted (before + 1) })); - let ids, changed = match Journal_sync.apply (fun value -> Ok value) conn before journal with + let applied = match Journal_sync.apply (fun value -> Ok value) conn before journal with | Ok value -> value | Error message -> fail message in - Runtime.rebase current journal.t ids changed; + Runtime.rebase ~changed_attributes:applied.changed_attributes current journal.t + applied.operation_ids applied.changed_uuids; check_eq ~msg:"each acknowledgement advances the queue" (List.length (Runtime.pending_operations current)) (2 - index)) pending; check_eq ~msg:"the recovered edits reach the authoritative graph" (title (Ds.conn_db conn) block_uuid) saved_title; @@ -1893,9 +1894,10 @@ let reopening_repairs_only_unconfirmed_legacy_page_deletion_timestamps () = "t-before", `Int 42; "t", `Int 43; "txs", `List [`Assoc ["t", `Int 43; "tx-id", `String pending.operation_id; "tx", `String tx]]])) with | Ok value -> value | Error message -> fail message in - let ids, changed = match Journal_sync.apply (fun text -> Ok text) conn 42 journal with + let applied = match Journal_sync.apply (fun text -> Ok text) conn 42 journal with | Ok value -> value | Error message -> fail message in - Runtime.rebase current 43 ids changed; + Runtime.rebase ~changed_attributes:applied.changed_attributes current 43 + applied.operation_ids applied.changed_uuids; check_eq ~msg:"confirmation clears the recovered deletion" (Pending.list path) [] end else check_eq ~msg:"unrelated rejections and unconfirmed submissions retain their exact transaction" @@ -1904,8 +1906,56 @@ let reopening_repairs_only_unconfirmed_legacy_page_deletion_timestamps () = Pending.Conflicted "permission denied", false; Pending.Submitted, false; Pending.Accepted 43, false] +let synced_journal_windows_match_fresh_reads () = + with_runtime (fun path conn current -> + Fun.protect ~finally:(fun () -> cleanup_store path) (fun () -> + let uuid index = Printf.sprintf "00000000-0000-4000-8000-%012d" index in + ignore (Ds.transact_conn conn ([add 1 "block/uuid" (Ds.Uuid (uuid 1)); + add 10 "block/uuid" (Ds.Uuid (uuid 10))] @ List.concat_map (fun index -> + let eid = 100 + index in + [add eid "block/uuid" (Ds.Uuid (uuid eid)); + add eid "block/title" (Ds.String ("Row " ^ string_of_int index)); + add eid "block/page" (Ds.Ref 1); add eid "block/parent" (Ds.Ref 1); + add eid "block/order" (Ds.String (Printf.sprintf "b%03d" index))]) + (List.init 40 Fun.id))); + Runtime.rebase current 42 [] []; + let check_window () = + check_eq ~msg:"the cached window must match a fresh bounded journal read" + (Runtime.blocks current) + (Graph_read.blocks ~block_limit:32 (fun value -> Ok value) 1 (Runtime.db current)) in + let apply operations = + let before = (Runtime.state current).server_t in + let _, tx, _ = Journal_sync.capture (Ds.conn_db conn) operations in + let journal = match Journal_sync.decode (Json.to_string (`Assoc [ + "type", `String "pull/ok"; "t-before", `Int before; "t", `Int (before + 1); + "txs", `List [`Assoc ["t", `Int (before + 1); "tx", `String tx]]])) with + | Ok journal -> journal | Error message -> fail message in + let applied = match Journal_sync.apply (fun value -> Ok value) conn before journal with + | Ok applied -> applied | Error message -> fail message in + Runtime.rebase ~changed_attributes:applied.changed_attributes current journal.t + applied.operation_ids applied.changed_uuids; + check_window () in + check_window (); + apply [add 10 "block/title" (Ds.String "Remote")]; + check (is_ok (Runtime.stage current + { (operation "local-title" (Pending.Save_title + {uuid = uuid 10; expected_title = "Remote"; title = "Local"})) with + base_t = (Runtime.state current).server_t })); + check_window (); + apply [add 10 "block/title" (Ds.String "Another remote")]; + apply [Ds.Retract (Ds.Entity_id 100, "block/title", None)]; + apply [add 100 "block/title" (Ds.String "Restored")]; + apply [add 500 "block/uuid" (Ds.Uuid (uuid 500)); + add 500 "block/title" (Ds.String "Inserted"); + add 500 "block/page" (Ds.Ref 1); add 500 "block/parent" (Ds.Ref 1); + add 500 "block/order" (Ds.String "a-")]; + apply [add 120 "block/order" (Ds.String "a!")]; + apply [add 500 "logseq.property/deleted-at" (Ds.Int64 100L)]; + apply [add 1 "block/title" (Ds.String "Updated journal")])) + let cases = [ + case "synced journal windows match fresh reads" synced_journal_windows_match_fresh_reads; case "reopening repairs only unconfirmed legacy page deletion timestamps" reopening_repairs_only_unconfirmed_legacy_page_deletion_timestamps; case "new pages appear in the built-in Page collection" new_pages_appear_in_the_builtin_page_collection; case "reopening repairs legacy pages and unblocks dependent changes" reopening_repairs_legacy_page_creations_and_unblocks_dependent_changes; diff --git a/shared/test/logseq/node_property_test.ml b/shared/test/logseq/node_property_test.ml index 573df2f3..cf68a4d4 100644 --- a/shared/test/logseq/node_property_test.ml +++ b/shared/test/logseq/node_property_test.ml @@ -25,9 +25,10 @@ let sync_all runtime conn = "tx-id", `String operation.Pending_ops.operation_id; "tx", `String tx]]])) with | Ok value -> value | Error message -> fail message in check_ok (Graph_runtime.stage runtime {operation with Pending_ops.state = Pending_ops.Accepted (before + 1)}); - let ids, changed = match Journal_sync.apply (fun value -> Ok value) conn before journal with + let applied = match Journal_sync.apply (fun value -> Ok value) conn before journal with | Ok value -> value | Error message -> fail message in - Graph_runtime.rebase runtime journal.t ids changed) (Graph_runtime.pending_operations runtime) + Graph_runtime.rebase ~changed_attributes:applied.changed_attributes runtime journal.t + applied.operation_ids applied.changed_uuids) (Graph_runtime.pending_operations runtime) let projection_and_navigation () = let db = R.fixture () in @@ -106,9 +107,10 @@ let typed_edits_persist_and_sync () = R.with_runtime (fun session runtime conn - "tx-id", `String operation.Pending_ops.operation_id; "tx", `String tx]]])) with | Ok value -> value | Error message -> fail message in check_ok (Graph_runtime.stage restored {operation with Pending_ops.state = Pending_ops.Accepted (before + 1)}); - let ids, changed = match Journal_sync.apply (fun value -> Ok value) conn before journal with + let applied = match Journal_sync.apply (fun value -> Ok value) conn before journal with | Ok value -> value | Error message -> fail message in - Graph_runtime.rebase restored journal.t ids changed) (Graph_runtime.pending_operations restored); + Graph_runtime.rebase ~changed_attributes:applied.changed_attributes restored journal.t + applied.operation_ids applied.changed_uuids) (Graph_runtime.pending_operations restored); check_eq (Graph_runtime.pending_operations restored) []) let invalid_and_stale_edits_do_not_write () = R.with_runtime (fun session runtime _ -> diff --git a/shared/test/logseq/result_view_test.ml b/shared/test/logseq/result_view_test.ml index 2b526733..d2f9d711 100644 --- a/shared/test/logseq/result_view_test.ml +++ b/shared/test/logseq/result_view_test.ml @@ -586,9 +586,10 @@ let native_sort_settings_survive_save_and_refresh () = "t-before", `Int before; "t", `Int (before + 1); "txs", `List [`Assoc ["t", `Int (before + 1); "tx-id", `String operation.Pending_ops.operation_id; "tx", `String tx]]])) with | Ok value -> value | Error message -> fail message in - let ids, changed = match Journal_sync.apply (fun value -> Ok value) conn before journal with + let applied = match Journal_sync.apply (fun value -> Ok value) conn before journal with | Ok value -> value | Error message -> fail message in - Graph_runtime.rebase graph journal.t ids changed) (Graph_runtime.pending_operations graph); + Graph_runtime.rebase ~changed_attributes:applied.changed_attributes graph journal.t + applied.operation_ids applied.changed_uuids) (Graph_runtime.pending_operations graph); check_eq ~msg:"Sorts remain after the server acknowledges the transaction" (List.length (items "sorting" (find_view table_view (read session)))) 2; let restored = Graph_runtime.create graph.path (Graph_runtime.state graph).server_t conn @@ -686,9 +687,10 @@ let typed_edits_restart_and_sync_acknowledgements () = with_runtime (fun session "tx-id", `String operation.Pending_ops.operation_id; "tx", `String tx]]])) with | Ok value -> value | Error message -> fail message in check_ok (Graph_runtime.stage restored {operation with Pending_ops.state = Pending_ops.Accepted (before + 1)}); - let ids, changed = match Journal_sync.apply (fun value -> Ok value) conn before journal with + let applied = match Journal_sync.apply (fun value -> Ok value) conn before journal with | Ok value -> value | Error message -> fail message in - Graph_runtime.rebase restored journal.t ids changed; + Graph_runtime.rebase ~changed_attributes:applied.changed_attributes restored journal.t + applied.operation_ids applied.changed_uuids; check_eq ~msg:"server echoes confirm atomic property values and owner assignments together" (List.length (Graph_runtime.pending_operations restored)) (2 - index)) pending; check_eq (Graph_runtime.pending_operations restored) [];