Reuse bounded journal windows across title sync

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

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

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)

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

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

shared/src/logseq/core/mobile_graph.ml

@@ -331,15 +331,16 @@ 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 ()
     else
       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 =

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

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 _ ->

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;