diff --git a/docs/design.md b/docs/design.md index 539fb64..f742a4f 100644 --- a/docs/design.md +++ b/docs/design.md @@ -169,6 +169,26 @@ threshold check, and the `Conn.storage_tail` accessor. The persisted `Storage_tail` payload keeps upstream's transaction-order-then-in-transaction datom order. +## Garbage Collection + +`Storage.collect_garbage` stays manual-only by default, matching upstream +DataScript: storage files grow until a caller collects unreachable addresses. +An opt-in automatic policy can be attached to a `storage` record through its +`storage_auto_gc` field — set it with the `?auto_gc` argument on +`memory_storage`, `file_storage`, or the SQLite storage constructors. + +- `auto_gc.min_garbage` — minimum number of unreachable storage addresses + before collection runs. +- `auto_gc.garbage_fraction` — minimum share of all stored addresses that must + be unreachable before collection runs. + +After a storage-tail compaction writes a fresh index snapshot, +`maybe_collect_garbage` re-evaluates the addresses unreachable from the new +root and deletes them when both thresholds trip. With no policy (`None`, the +default) nothing is deleted automatically; reclaim space by calling +`collect_garbage`, or `auto_collect_garbage` with an explicit policy. +`default_auto_gc` is `{ min_garbage = 1024; garbage_fraction = 0.5 }`. + ## Persistent Sorted Set Dependency The repo depends on `persistent_sorted_set_ocaml` through an opam pin to diff --git a/examples/logseq_sqlite_storage.ml b/examples/logseq_sqlite_storage.ml index 0c0fe9a..a07109e 100644 --- a/examples/logseq_sqlite_storage.ml +++ b/examples/logseq_sqlite_storage.ml @@ -1335,7 +1335,7 @@ let upsert_sql (address, payload) = (sql_quote content) addresses -let storage db_path = +let storage ?auto_gc db_path = create_kvs_table db_path; let store entries = let sql = String.concat "" (List.map upsert_sql entries) in @@ -1375,4 +1375,5 @@ let storage db_path = ; storage_restore = restore ; storage_list_addresses = list_addresses ; storage_delete = delete + ; storage_auto_gc = auto_gc } diff --git a/impl/conn.ml b/impl/conn.ml index 125fb81..82378a5 100644 --- a/impl/conn.ml +++ b/impl/conn.ml @@ -32,6 +32,7 @@ type transact_context = ; store_tail : storage -> datom list list -> unit ; storage_tail_datom_count : datom list list -> int ; storage_tail_compaction_threshold : db -> int + ; maybe_collect_garbage : storage -> unit ; transact : tx_meta:tx_meta -> db -> tx_op list -> tx_report } @@ -49,6 +50,7 @@ type context = ; restore_tail_groups : storage -> datom list list ; storage_tail_datom_count : datom list list -> int ; storage_tail_compaction_threshold : db -> int + ; maybe_collect_garbage : storage -> unit ; transact : tx_meta:tx_meta -> db -> tx_op list -> tx_report ; datoms : db -> datom list ; with_schema : db -> schema -> db @@ -151,7 +153,8 @@ let transact (context : transact_context) ?(tx_meta = []) conn tx_data = 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 <- [] + conn.storage_tail <- []; + context.maybe_collect_garbage storage end else begin conn.storage_tail <- tail; context.store_tail storage (List.rev conn.storage_tail) @@ -179,7 +182,8 @@ let apply_report (context : transact_context) conn (report : tx_report) = 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 <- [] + conn.storage_tail <- []; + context.maybe_collect_garbage storage end else begin conn.storage_tail <- tail; context.store_tail storage (List.rev conn.storage_tail) diff --git a/impl/conn.mli b/impl/conn.mli index bc45688..8a00b93 100644 --- a/impl/conn.mli +++ b/impl/conn.mli @@ -23,6 +23,7 @@ type transact_context = ; store_tail : storage -> datom list list -> unit ; storage_tail_datom_count : datom list list -> int ; storage_tail_compaction_threshold : db -> int + ; maybe_collect_garbage : storage -> unit ; transact : tx_meta:tx_meta -> db -> tx_op list -> tx_report } @@ -40,6 +41,7 @@ type context = ; restore_tail_groups : storage -> datom list list ; storage_tail_datom_count : datom list list -> int ; storage_tail_compaction_threshold : db -> int + ; maybe_collect_garbage : storage -> unit ; transact : tx_meta:tx_meta -> db -> tx_op list -> tx_report ; datoms : db -> datom list ; with_schema : db -> schema -> db diff --git a/impl/datascript.ml b/impl/datascript.ml index f2b522a..1916611 100644 --- a/impl/datascript.ml +++ b/impl/datascript.ml @@ -116,6 +116,7 @@ let store ?storage db = let memory_storage = Storage.memory_storage let file_storage = Storage.file_storage +let default_auto_gc = Storage.default_auto_gc let store_tail = Storage.store_tail let storage_tail_compaction_threshold = Storage.tail_compaction_threshold let storage_tail_datom_count = Storage.tail_datom_count @@ -125,6 +126,8 @@ let storage = Storage.storage let addresses = Storage.addresses let settings = Storage.settings let collect_garbage = Storage.collect_garbage +let auto_collect_garbage = Storage.auto_collect_garbage +let maybe_collect_garbage = Storage.maybe_collect_garbage let conn_creation_context : Conn.creation_context = { empty_db; init_db; store } @@ -907,9 +910,11 @@ let persist_transact_tail ~tx_meta db tx_data = if storage_tail_datom_count groups + List.length tx_data > storage_tail_compaction_threshold db - then - store ~storage db - else begin + then begin + let db = store ~storage db in + maybe_collect_garbage storage; + db + end else begin store_tail storage (groups @ [ tx_data ]); db end @@ -931,6 +936,7 @@ let transact_conn ?(tx_meta = []) conn tx_data = ; store_tail ; storage_tail_datom_count ; storage_tail_compaction_threshold + ; maybe_collect_garbage ; transact = (fun ~tx_meta db tx_data -> transact_report ~tx_meta db tx_data) } in @@ -942,6 +948,7 @@ let apply_report (conn : conn) (report : tx_report) : tx_report = ; store_tail ; storage_tail_datom_count ; storage_tail_compaction_threshold + ; maybe_collect_garbage ; transact = (fun ~tx_meta db tx_data -> transact_report ~tx_meta db tx_data) } in diff --git a/impl/datascript.mli b/impl/datascript.mli index ce80137..e6c0719 100644 --- a/impl/datascript.mli +++ b/impl/datascript.mli @@ -99,6 +99,7 @@ module Conn : sig ; store_tail : storage -> datom list list -> unit ; storage_tail_datom_count : datom list list -> int ; storage_tail_compaction_threshold : db -> int + ; maybe_collect_garbage : storage -> unit ; transact : tx_meta:tx_meta -> db -> tx_op list -> tx_report } @@ -247,8 +248,8 @@ module Storage : sig val root_address : storage_address val tail_address : storage_address - val memory_storage : unit -> storage - val file_storage : string -> storage + val memory_storage : ?auto_gc:auto_gc -> unit -> storage + val file_storage : ?auto_gc:auto_gc -> string -> storage val store : ?storage:storage -> db -> db val store_tail : storage -> datom list list -> unit val tail_compaction_threshold : db -> int @@ -262,6 +263,18 @@ module Storage : sig val addresses : db list -> storage_address list val settings : db -> (attr * value) list val collect_garbage : storage -> unit + (* Default opt-in GC policy: collect when at least 1024 addresses are + unreachable and they make up >= 50% of all stored addresses. *) + val default_auto_gc : auto_gc + (* Evaluate an auto-GC policy once: count the addresses no longer + reachable from the stored root and delete them when the garbage is + both >= [policy.min_garbage] and >= [policy.garbage_fraction] of all + stored addresses. *) + val auto_collect_garbage : auto_gc -> storage -> unit + (* [auto_collect_garbage] driven by [storage.storage_auto_gc]; a no-op + when the storage carries no policy (the default). Invoked + automatically after every storage-tail compaction. *) + val maybe_collect_garbage : storage -> unit end module Util : sig @@ -413,8 +426,12 @@ val unfiltered_db : db -> db val serializable : db -> serializable_db val from_serializable : serializable_db -> db val db_from_reader_string : string -> db -val memory_storage : unit -> storage -val file_storage : string -> storage +(* [auto_gc] opts a storage into automatic garbage collection after each + storage-tail compaction; omitted, [collect_garbage] stays manual-only + (upstream behaviour). *) +val memory_storage : ?auto_gc:auto_gc -> unit -> storage +val file_storage : ?auto_gc:auto_gc -> string -> storage +val default_auto_gc : auto_gc val store : ?storage:storage -> db -> db val store_tail : storage -> datom list list -> unit val restore : storage -> db option @@ -424,6 +441,9 @@ val addresses : db list -> storage_address list val settings : db -> (attr * value) list val storage_addresses : storage -> storage_address list val collect_garbage : storage -> unit +(* See [Storage.auto_collect_garbage] / [Storage.maybe_collect_garbage]. *) +val auto_collect_garbage : auto_gc -> storage -> unit +val maybe_collect_garbage : storage -> unit val db_hash : db -> int val db_hash_cache_size : unit -> int val diff : db -> db -> datom list * datom list * datom list diff --git a/impl/platform.mli b/impl/platform.mli index 09f5c74..70cc9cc 100644 --- a/impl/platform.mli +++ b/impl/platform.mli @@ -6,7 +6,7 @@ open Datascript_types val now_seconds : unit -> float (** Create a file-backed storage instance rooted at the given path. *) -val file_storage : string -> storage +val file_storage : ?auto_gc:auto_gc -> string -> storage (** Compile a platform-specific regular expression from a pattern string. *) val compile_regex : string -> regex diff --git a/impl/platform/jsoo/platform.ml b/impl/platform/jsoo/platform.ml index 67568e9..b5f4b58 100644 --- a/impl/platform/jsoo/platform.ml +++ b/impl/platform/jsoo/platform.ml @@ -6,7 +6,7 @@ let now_seconds () = let now_ms = Js.Unsafe.meth_call Js.date "now" [||] in Js.to_float now_ms /. 1000.0 -let file_storage _dir = +let file_storage ?auto_gc:_ _dir = invalid_arg "file_storage is not supported on js_of_ocaml" let compile_regex = Regexp.regexp diff --git a/impl/platform/melange/platform.ml b/impl/platform/melange/platform.ml index cde318e..31ffc7b 100644 --- a/impl/platform/melange/platform.ml +++ b/impl/platform/melange/platform.ml @@ -6,7 +6,7 @@ external set_last_index : Js.Re.t -> int -> unit = "lastIndex" [@@mel.set] let now_seconds () = date_now () /. 1000.0 -let file_storage _dir = +let file_storage ?auto_gc:_ _dir = invalid_arg "file_storage is not supported on Melange" let compile_regex pattern = pattern diff --git a/impl/platform/native/platform.ml b/impl/platform/native/platform.ml index 8c1bac8..5011ca7 100644 --- a/impl/platform/native/platform.ml +++ b/impl/platform/native/platform.ml @@ -40,7 +40,7 @@ let decode_storage_address encoded = let storage_payload_path dir address = Filename.concat dir (encode_storage_address address ^ ".bin") -let file_storage dir = +let file_storage ?auto_gc dir = ensure_storage_dir dir; let write_payload address payload = let channel = open_out_bin (storage_payload_path dir address) in @@ -81,6 +81,7 @@ let file_storage dir = ; storage_restore = read_payload ; storage_list_addresses = list_addresses ; storage_delete = delete + ; storage_auto_gc = auto_gc } let str_pattern_of_pattern pattern = diff --git a/impl/storage.ml b/impl/storage.ml index 8d93bbc..a3962e9 100644 --- a/impl/storage.ml +++ b/impl/storage.ml @@ -23,7 +23,7 @@ let next_storage_address () = let note_storage_root root = max_storage_addr := max !max_storage_addr root.storage_max_addr -let memory_storage () = +let memory_storage ?auto_gc () = let disk = ref [] in let store entries = disk := @@ -45,6 +45,7 @@ let memory_storage () = ; storage_restore = restore ; storage_list_addresses = list_addresses ; storage_delete = delete + ; storage_auto_gc = auto_gc } let file_storage = Platform.file_storage @@ -442,9 +443,29 @@ let settings (db : db) = ; "storage", Bool (Option.is_some db.storage_ref) ] -let collect_garbage storage = +let live_address_set storage = let live = Hashtbl.create 257 in List.iter (fun address -> Hashtbl.replace live address ()) (storage_root_addresses storage); - storage.storage_list_addresses () - |> List.filter (fun address -> not (Hashtbl.mem live address)) - |> storage.storage_delete + live + +let unreachable_addresses live storage = + List.filter (fun address -> not (Hashtbl.mem live address)) (storage.storage_list_addresses ()) + +let collect_garbage storage = + unreachable_addresses (live_address_set storage) storage |> storage.storage_delete + +let default_auto_gc = { min_garbage = 1024; garbage_fraction = 0.5 } + +let auto_collect_garbage policy storage = + let live = live_address_set storage in + let all = storage.storage_list_addresses () in + let garbage = List.filter (fun address -> not (Hashtbl.mem live address)) all in + if + List.length garbage >= policy.min_garbage + && float_of_int (List.length garbage) >= policy.garbage_fraction *. float_of_int (List.length all) + then storage.storage_delete garbage + +let maybe_collect_garbage storage = + match storage.storage_auto_gc with + | None -> () + | Some policy -> auto_collect_garbage policy storage diff --git a/impl/storage.mli b/impl/storage.mli index afc094e..f789fe8 100644 --- a/impl/storage.mli +++ b/impl/storage.mli @@ -11,8 +11,8 @@ type restore_context = val root_address : storage_address val tail_address : storage_address -val memory_storage : unit -> storage -val file_storage : string -> storage +val memory_storage : ?auto_gc:auto_gc -> unit -> storage +val file_storage : ?auto_gc:auto_gc -> string -> storage val store : ?storage:storage -> db -> db val store_tail : storage -> datom list list -> unit val normalize_stored_datom : schema -> datom -> datom @@ -27,3 +27,6 @@ val storage : db -> storage option val addresses : db list -> storage_address list val settings : db -> (attr * value) list val collect_garbage : storage -> unit +val default_auto_gc : auto_gc +val auto_collect_garbage : auto_gc -> storage -> unit +val maybe_collect_garbage : storage -> unit diff --git a/sqlite/datascript_sqlite.ml b/sqlite/datascript_sqlite.ml index 82b6e9c..c6401bd 100644 --- a/sqlite/datascript_sqlite.ml +++ b/sqlite/datascript_sqlite.ml @@ -24,7 +24,7 @@ let close session = sqlite_close session.path; session.closed <- true) -let storage session : Ds.storage = +let storage ?auto_gc session : Ds.storage = { storage_store = (fun entries -> ensure_open session; @@ -46,4 +46,5 @@ let storage session : Ds.storage = (fun addresses -> ensure_open session; sqlite_delete session.path addresses) + ; storage_auto_gc = auto_gc } diff --git a/test/test_storage.ml b/test/test_storage.ml index a2fd7ac..45104ea 100644 --- a/test/test_storage.ml +++ b/test/test_storage.ml @@ -443,6 +443,63 @@ let test_storage__test_tail_datom_count () = 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_auto_gc () = + let batches_of_txs conn tx_count batch_count = + let entity = ref 0 in + for _ = 1 to batch_count do + ignore + (transact_conn conn + (List.init tx_count (fun _ -> + incr entity; + Add (Entity_id !entity, "name", String (string_of_int !entity))))) + done + in + (* 35 > branching factor 32: every batch compacts, leaving orphaned index + nodes behind. Default storage keeps them; an always-tripping policy + collects after each compaction. *) + let manual_storage = memory_storage () in + let manual_conn = create_conn ~schema:[ "name", indexed ] ~storage:manual_storage () in + batches_of_txs manual_conn 35 4; + let auto_storage = memory_storage ~auto_gc:{ min_garbage = 1; garbage_fraction = 0.0 } () in + let auto_conn = create_conn ~schema:[ "name", indexed ] ~storage:auto_storage () in + batches_of_txs auto_conn 35 4; + let auto_live = List.length (addresses [ conn_db auto_conn ]) in + if List.length (storage_addresses auto_storage) <> auto_live then + failf + "auto-GC should leave only reachable addresses: stored=%d live=%d" + (List.length (storage_addresses auto_storage)) + auto_live; + if not (List.length (storage_addresses manual_storage) > auto_live) then + failwith "manual storage should accumulate unreachable addresses"; + (* min_garbage is a floor: a large one keeps the storage manual *) + let floored_storage = + memory_storage ~auto_gc:{ min_garbage = 1_000_000; garbage_fraction = 0.0 } () + in + let floored_conn = create_conn ~schema:[ "name", indexed ] ~storage:floored_storage () in + batches_of_txs floored_conn 35 4; + assert_equal_int + "unreachable floor keeps auto-GC from deleting" + (List.length (storage_addresses manual_storage)) + (List.length (storage_addresses floored_storage)); + (* maybe_collect_garbage is a no-op without a policy, and honours one *) + let probe = memory_storage () in + ignore (store ~storage:probe (small_db ())); + probe.storage_store [ "stale/addr", Storage_tail [] ]; + maybe_collect_garbage probe; + if not (List.mem "stale/addr" (storage_addresses probe)) then + failwith "maybe_collect_garbage should be a no-op without a policy"; + auto_collect_garbage { min_garbage = 1; garbage_fraction = 0.0 } probe; + if List.mem "stale/addr" (storage_addresses probe) then + failwith "auto_collect_garbage should delete unreachable addresses"; + let opted = memory_storage ~auto_gc:{ min_garbage = 1; garbage_fraction = 0.0 } () in + ignore (store ~storage:opted (small_db ())); + opted.storage_store [ "stale/addr", Storage_tail [] ]; + maybe_collect_garbage opted; + if List.mem "stale/addr" (storage_addresses opted) then + failwith "maybe_collect_garbage should honour storage_auto_gc"; + if default_auto_gc.min_garbage <> 1024 || default_auto_gc.garbage_fraction <> 0.5 then + failwith "default_auto_gc should keep the documented policy" + let test_storage__test_db_with_tail () = let db = empty_db ~schema:[ "block/updated-at", indexed; "block/uuid", unique_identity ] () @@ -591,4 +648,5 @@ let () = test_storage__test_conn (); test_storage__test_conn_tail_write_order (); test_storage__test_tail_datom_count (); + test_storage__test_auto_gc (); test_storage__test_db_with_tail (); diff --git a/type/datascript_types.ml b/type/datascript_types.ml index 244ac5c..c0f6e9f 100644 --- a/type/datascript_types.ml +++ b/type/datascript_types.ml @@ -108,11 +108,22 @@ type storage_payload = | Storage_node of datom Persistent_sorted_set.stored_node | Storage_tail of datom list list +(* Opt-in automatic garbage collection for a storage. After every + storage-tail compaction the unreachable-address count is re-evaluated + and the garbage is deleted when it is both >= [min_garbage] and + >= [garbage_fraction] of all stored addresses. [None] (the default) + keeps upstream's manual-only [collect_garbage]. *) +type auto_gc = + { min_garbage : int + ; garbage_fraction : float + } + type storage = { storage_store : (storage_address * storage_payload) list -> unit ; storage_restore : storage_address -> storage_payload option ; storage_list_addresses : unit -> storage_address list ; storage_delete : storage_address list -> unit + ; storage_auto_gc : auto_gc option } type tx_value =