perf(storage): O(1) storage_tail append via newest-first batch list

docs/design.md

@@ -159,6 +159,16 @@ Not every value near the DB should become an ordered persistent index.
 Moving these to PSS would not improve parity with upstream DataScript and would
 make serialization and equality behavior more complex.
 
+## Storage Tail
+
+`conn.storage_tail` holds one `datom list` batch per committed transaction,
+stored newest-first so appending a transaction's `tx_data` is O(1). The
+batches are flattened back to transaction order only where the flat tail is
+consumed: `store_tail` writes, `storage_tail_datom_count`, the compaction
+threshold check, and the `Conn.storage_tail` accessor. The persisted
+`Storage_tail` payload keeps upstream's transaction-order-then-in-transaction
+datom order.
+
 ## Persistent Sorted Set Dependency
 
 The repo depends on `persistent_sorted_set_ocaml` through an opam pin to

impl/conn.ml

@@ -5,7 +5,10 @@ type t =
   ; mutable listeners : (string * (tx_report -> unit)) list
   ; mutable next_listener_id : int
   ; storage : storage option
-  ; mutable storage_tail : datom list list
+  ; (* per-transaction datom batches, newest first: consing each report's
+       tx_data is O(1); reversed back to on-disk tx order only when the
+       tail is written or read out *)
+    mutable storage_tail : datom list list
   }
 
 type creation_context =
@@ -70,7 +73,7 @@ let make ?storage ?(storage_tail = []) db =
     | None -> db
     | Some _ -> { db with storage_ref = storage }
   in
-  { db; listeners = []; next_listener_id = 0; storage; storage_tail }
+  { db; listeners = []; next_listener_id = 0; storage; storage_tail = List.rev storage_tail }
 
 let create (context : creation_context) ?schema ?storage () =
   let db = context.empty_db ?schema ?storage () in
@@ -97,7 +100,7 @@ let db conn = conn.db
    store/notify side effects, for bookkeeping fields like :max-tx. *)
 let update_db conn f = conn.db <- f conn.db
 
-let storage_tail conn = conn.storage_tail
+let storage_tail conn = List.rev conn.storage_tail
 
 let is_conn (_ : t) = true
 
@@ -145,13 +148,13 @@ let transact (context : transact_context) ?(tx_meta = []) conn tx_data =
      | None -> ()
      | Some storage ->
        if report.tx_data <> [] then begin
-         let tail = conn.storage_tail @ [ report.tx_data ] in
+         let tail = report.tx_data :: conn.storage_tail in
          if context.storage_tail_datom_count tail > context.storage_tail_compaction_threshold report.db_after then begin
            conn.db <- context.store ~storage report.db_after;
            conn.storage_tail <- []
          end else begin
            conn.storage_tail <- tail;
-           context.store_tail storage conn.storage_tail
+           context.store_tail storage (List.rev conn.storage_tail)
          end
        end);
   notify_listeners conn report;
@@ -173,13 +176,13 @@ let apply_report (context : transact_context) conn (report : tx_report) =
    | None -> ()
    | Some storage ->
      if report.tx_data <> [] then begin
-       let tail = conn.storage_tail @ [ report.tx_data ] in
+       let tail = report.tx_data :: conn.storage_tail in
        if context.storage_tail_datom_count tail > context.storage_tail_compaction_threshold db_after then begin
          conn.db <- context.store ~storage db_after;
          conn.storage_tail <- []
        end else begin
          conn.storage_tail <- tail;
-         context.store_tail storage conn.storage_tail
+         context.store_tail storage (List.rev conn.storage_tail)
        end
      end);
   let report = { report with db_after } in

impl/datascript.ml

@@ -903,11 +903,14 @@ let persist_transact_tail ~tx_meta db tx_data =
     match db.storage_ref with
     | None -> db
     | Some storage ->
-      let tail = restore_tail_groups storage @ [ tx_data ] in
-      if storage_tail_datom_count tail > storage_tail_compaction_threshold db then
+      let groups = restore_tail_groups storage in
+      if
+        storage_tail_datom_count groups + List.length tx_data
+        > storage_tail_compaction_threshold db
+      then
         store ~storage db
       else begin
-        store_tail storage tail;
+        store_tail storage (groups @ [ tx_data ]);
         db
       end
 

impl/storage.ml

@@ -280,7 +280,7 @@ let tail_compaction_threshold (db : db) =
   (PSet.settings db.eavt_index).branching_factor
 
 let tail_datom_count tail =
-  tail |> List.concat |> List.length
+  List.fold_left (fun count group -> count + List.length group) 0 tail
 
 let restore_root_snapshot storage =
   match storage.storage_restore root_address with

test/test_storage.ml

@@ -379,6 +379,70 @@ let test_storage__test_conn () =
       (datoms restored_db Eavt ())
   | None -> failwith "conn_from_datoms should store attached datoms"
 
+let tail_groups storage =
+  match storage.storage_restore Storage.tail_address with
+  | Some (Storage_tail groups) -> groups
+  | _ -> failwith "storage tail address missing"
+
+let flatten_tail_groups groups =
+  List.map (fun group -> List.map (fun d -> d.e, d.a, d.v) group) groups
+
+let test_storage__test_conn_tail_write_order () =
+  let storage = memory_storage () in
+  let conn = create_conn ~storage () in
+  let assert_tail label conn expected =
+    let stored = flatten_tail_groups (tail_groups storage) in
+    if stored <> expected then
+      failf "%s: stored tail order mismatch: %s" label
+        (String.concat ";" (List.map (fun g -> string_of_int (List.length g)) stored));
+    if flatten_tail_groups (Conn.storage_tail conn) <> expected then
+      failwith (label ^ ": conn storage_tail accessor order mismatch")
+  in
+  ignore (transact_conn conn [ Add (Entity_id 1, "a", String "1") ]);
+  ignore
+    (transact_conn conn
+       [ Add (Entity_id 2, "a", String "2"); Add (Entity_id 2, "b", String "3") ]);
+  assert_tail "two txs" conn
+    [ [ 1, "a", String "1" ]
+    ; [ 2, "a", String "2"; 2, "b", String "3" ]
+    ];
+  (* a restored conn seeds its tail from disk and keeps appending in tx
+     order; the on-disk order must survive the internal reversal *)
+  let conn =
+    match restore_conn storage with
+    | Some conn -> conn
+    | None -> failwith "restore_conn should return a conn"
+  in
+  ignore (transact_conn conn [ Add (Entity_id 3, "a", String "4") ]);
+  assert_tail "restored conn" conn
+    [ [ 1, "a", String "1" ]
+    ; [ 2, "a", String "2"; 2, "b", String "3" ]
+    ; [ 3, "a", String "4" ]
+    ];
+  (* crossing the compaction threshold clears the stored tail *)
+  ignore
+    (transact_conn conn
+       (List.init 40 (fun i -> Add (Entity_id (i + 10), "a", String (string_of_int i)))));
+  (match tail_groups storage with
+   | [] -> ()
+   | _ -> failwith "compaction should clear the stored tail");
+  ignore (transact_conn conn [ Add (Entity_id 99, "a", String "9") ]);
+  assert_tail "post-compaction" conn [ [ 99, "a", String "9" ] ];
+  assert_equal_triples
+    "all datoms queryable after compaction"
+    [ 1, "a", String "1"; 2, "a", String "2"; 2, "b", String "3"
+    ; 3, "a", String "4"; 99, "a", String "9" ]
+    (datoms (conn_db conn) Eavt ~e:1 ()
+     @ datoms (conn_db conn) Eavt ~e:2 ()
+     @ datoms (conn_db conn) Eavt ~e:3 ()
+     @ datoms (conn_db conn) Eavt ~e:99 ()
+     |> List.sort compare)
+
+let test_storage__test_tail_datom_count () =
+  let d tx e = datom ~tx ~e ~a:"a" ~v:(String "v") () in
+  assert_equal_int "tail_datom_count sums group sizes" 4
+    (Storage.tail_datom_count [ [ d 1 1 ]; [ d 2 2; d 2 3; d 2 4 ]; [] ])
+
 let test_storage__test_db_with_tail () =
   let db =
     empty_db ~schema:[ "block/updated-at", indexed; "block/uuid", unique_identity ] ()
@@ -525,4 +589,6 @@ let () =
   test_storage__test_transact_after_restore_uses_index_slices ();
   test_storage__test_conn_repeated_transacts_store_incrementally ();
   test_storage__test_conn ();
+  test_storage__test_conn_tail_write_order ();
+  test_storage__test_tail_datom_count ();
   test_storage__test_db_with_tail ();