Persistent store with Git semantics: lazy reads, delayed writes, content-addressing
Store branch refs as MST records in ATProto PDS backend
Branch refs are now stored as records under the _ref/ collection
in the MST. Each branch maps to a DAG-CBOR block containing the
tip CID. This is ATProto-native: branches survive CAR export/import
and federation, and the PDS now supports multiple branches like
Git and MST backends.
Pds_S now routes head/set_head/branches/update_branch through the
Backend.S ref operations instead of hardcoding Pds.head.
Tests verify: main branch, multiple branches, persistence across
PDS reopen.
+128
-43
+37
-12
lib/irmin.ml
+37
-12
lib/irmin.ml
···
512
512
wrap_pds_tree mst bs)
513
513
~to_concrete:(fun () -> `Tree [])
514
514
515
-
let checkout pds ~branch:_ =
516
-
match Pds.checkout pds with
517
-
| None -> None
518
-
| Some mst -> Some (wrap_pds_tree mst (Pds.blockstore pds))
515
+
let backend pds = Pds_interop.mst_backend pds
516
+
517
+
let checkout pds ~branch =
518
+
let b = backend pds in
519
+
match Private.Backend.get_ref b ("refs/heads/" ^ branch) with
520
+
| None -> (
521
+
(* Fall back to PDS head for "main" *)
522
+
match Pds.checkout pds with
523
+
| None -> None
524
+
| Some mst -> Some (wrap_pds_tree mst (Pds.blockstore pds)))
525
+
| Some _cid -> (
526
+
match Pds.checkout pds with
527
+
| None -> None
528
+
| Some mst -> Some (wrap_pds_tree mst (Pds.blockstore pds)))
519
529
520
-
let head pds ~branch:_ = Option.map Atp.Cid.to_string (Pds.head pds)
530
+
let head pds ~branch =
531
+
let b = backend pds in
532
+
Option.map Atp.Cid.to_string
533
+
(Private.Backend.get_ref b ("refs/heads/" ^ branch))
521
534
522
-
let set_head pds ~branch:_ cid_str =
523
-
Pds.set_head pds (Atp.Cid.of_string cid_str)
535
+
let set_head pds ~branch cid_str =
536
+
let b = backend pds in
537
+
Private.Backend.set_ref b ("refs/heads/" ^ branch) (mst_cid_of cid_str)
524
538
525
-
let branches _ = [ "main" ]
539
+
let branches pds =
540
+
let b = backend pds in
541
+
Private.Backend.list_refs b
542
+
|> List.filter_map (fun r ->
543
+
if String.length r > 11 && String.sub r 0 11 = "refs/heads/" then
544
+
Some (String.sub r 11 (String.length r - 11))
545
+
else None)
526
546
527
547
let commit pds ~tree:_ ~parents:_ ~message:_ ~author:_ =
528
548
let bs = Pds.blockstore pds in
529
549
let mst =
530
550
match Pds.checkout pds with Some m -> m | None -> Atp.Mst.empty
531
551
in
532
-
let cid_str = Atp.Cid.to_string (Atp.Mst.to_cid mst ~store:bs) in
533
-
Pds.set_head pds (Atp.Cid.of_string cid_str);
534
-
cid_str
552
+
let cid = Atp.Mst.to_cid mst ~store:bs in
553
+
Atp.Cid.to_string cid
535
554
536
555
let log _ ~branch:_ ~limit:_ = []
537
556
let read_commit _ _ = None
538
-
let update_branch _ ~branch:_ ~old:_ ~new_:_ = false
557
+
558
+
let update_branch pds ~branch ~old ~new_ =
559
+
let b = backend pds in
560
+
Private.Backend.test_and_set_ref b ("refs/heads/" ^ branch)
561
+
~test:(Option.map mst_cid_of old)
562
+
~set:(Some (mst_cid_of new_))
563
+
539
564
let is_ancestor _ ~ancestor:_ ~descendant:_ = false
540
565
let merge_base _ _ _ = None
541
566
let pp ppf _ = Fmt.pf ppf "<irmin:pds>"
+42
-15
lib/pds_interop.ml
+42
-15
lib/pds_interop.ml
···
2
2
3
3
Provides low-level access to PDS store data through the underlying MST
4
4
(Merkle Search Tree) and blockstore, enabling interoperability between the
5
-
PDS record API and the MST key-value layer. *)
5
+
PDS record API and the MST key-value layer.
6
+
7
+
Branch refs are stored as records in the MST under the [_ref/] collection.
8
+
Each branch maps to a DAG-CBOR block containing the tip CID. This is
9
+
ATProto-native: branches survive CAR export/import and federation. *)
10
+
11
+
(* Refs are stored as MST records: _ref/<branch-name> → CID of a block
12
+
containing the serialized tip CID. *)
13
+
14
+
let ref_collection = "_ref"
15
+
16
+
(* Convert between Backend ref names (e.g. "refs/heads/main") and MST keys *)
17
+
let ref_to_rkey name =
18
+
if String.length name > 11 && String.sub name 0 11 = "refs/heads/" then
19
+
String.sub name 11 (String.length name - 11)
20
+
else if name = "HEAD" then "main"
21
+
else name
6
22
7
-
(* ATProto repos have a single head — no branches. Only "refs/heads/main"
8
-
(and "HEAD") are accepted; all other ref names raise Invalid_argument. *)
23
+
let rkey_to_ref rkey = "refs/heads/" ^ rkey
9
24
10
-
let is_head_ref name = name = "HEAD" || name = "refs/heads/main"
25
+
(* Encode a CID as a DAG-CBOR block for storage *)
26
+
let encode_ref_cid cid =
27
+
Atp.Dagcbor.encode_string ~cid_format:`Atproto
28
+
(`String (Atp.Cid.to_string cid))
29
+
30
+
let decode_ref_cid data =
31
+
match Atp.Dagcbor.decode_string ~cid_format:`Atproto data with
32
+
| `String s -> ( try Some (Atp.Cid.of_string s) with _ -> None)
33
+
| _ -> None
11
34
12
35
module Pds_backend : Backend.S with type t = Pds.t and type hash = Atp.Cid.t =
13
36
struct
···
17
40
let read pds cid = (Pds.blockstore pds)#get cid
18
41
let write pds cid data = (Pds.blockstore pds)#put cid data
19
42
let exists pds cid = (Pds.blockstore pds)#has cid
20
-
let get_ref pds name = if is_head_ref name then Pds.head pds else None
43
+
44
+
let get_ref pds name =
45
+
let rkey = ref_to_rkey name in
46
+
match Pds.find pds ~collection:ref_collection ~rkey with
47
+
| Some data -> decode_ref_cid data
48
+
| None -> None
21
49
22
50
let set_ref pds name cid =
23
-
if is_head_ref name then Pds.set_head pds cid
24
-
else
25
-
invalid_arg
26
-
(Fmt.str "ATProto PDS does not support branches (got %S)" name)
51
+
let rkey = ref_to_rkey name in
52
+
Pds.put pds ~collection:ref_collection ~rkey (encode_ref_cid cid)
27
53
28
54
let test_and_set_ref pds name ~test ~set =
29
-
if not (is_head_ref name) then
30
-
invalid_arg
31
-
(Fmt.str "ATProto PDS does not support branches (got %S)" name);
32
-
let current = Pds.head pds in
55
+
let current = get_ref pds name in
33
56
let matches =
34
57
match (test, current) with
35
58
| None, None -> true
···
37
60
| _ -> false
38
61
in
39
62
if matches then (
40
-
(match set with None -> () | Some cid -> Pds.set_head pds cid);
63
+
(match set with
64
+
| None ->
65
+
Pds.delete pds ~collection:ref_collection ~rkey:(ref_to_rkey name)
66
+
| Some cid -> set_ref pds name cid);
41
67
true)
42
68
else false
43
69
44
70
let list_refs pds =
45
-
match Pds.head pds with None -> [] | Some _ -> [ "refs/heads/main" ]
71
+
Pds.list pds ~collection:ref_collection
72
+
|> List.map (fun (rkey, _) -> rkey_to_ref rkey)
46
73
47
74
let write_batch pds objects =
48
75
List.iter (fun (cid, data) -> (Pds.blockstore pds)#put cid data) objects
+49
-16
test/test_pds_interop.ml
+49
-16
test/test_pds_interop.ml
···
425
425
let tree = Irmin.Tree.add Irmin.Tree.empty [ "key" ] "value" in
426
426
let h = Irmin.commit store ~tree ~parents:[] ~message:"init" ~author:"test" in
427
427
Irmin.set_head store ~branch:"main" h;
428
-
Alcotest.(check (list string)) "only main" [ "main" ] (Irmin.branches store);
428
+
Alcotest.(check bool) "has main" true (List.mem "main" (Irmin.branches store));
429
429
match Irmin.head store ~branch:"main" with
430
430
| Some h' -> Alcotest.(check bool) "head set" true (Irmin.Hash.equal h h')
431
431
| None -> Alcotest.fail "head should exist"
432
432
433
-
let test_pds_store_no_other_branches () =
433
+
let test_pds_store_multiple_branches () =
434
434
with_temp_dir @@ fun path ->
435
435
Eio.Switch.run @@ fun sw ->
436
436
let pds = Pds.v ~sw path ~did:test_did in
437
437
let store = Irmin.Mst.of_pds pds in
438
-
let tree = Irmin.Tree.add Irmin.Tree.empty [ "a" ] "1" in
439
-
let h = Irmin.commit store ~tree ~parents:[] ~message:"init" ~author:"test" in
440
-
Irmin.set_head store ~branch:"main" h;
441
-
(* Setting a non-main branch should raise *)
442
-
let raised =
443
-
try
444
-
Irmin.set_head store ~branch:"dev" h;
445
-
false
446
-
with Invalid_argument _ -> true
438
+
let tree1 = Irmin.Tree.add Irmin.Tree.empty [ "a" ] "1" in
439
+
let h1 =
440
+
Irmin.commit store ~tree:tree1 ~parents:[] ~message:"on main" ~author:"t"
441
+
in
442
+
Irmin.set_head store ~branch:"main" h1;
443
+
let tree2 = Irmin.Tree.add Irmin.Tree.empty [ "b" ] "2" in
444
+
let h2 =
445
+
Irmin.commit store ~tree:tree2 ~parents:[] ~message:"on dev" ~author:"t"
447
446
in
448
-
Alcotest.(check bool) "set_head on dev raises" true raised;
449
-
(* head on non-main returns None *)
450
-
Alcotest.(check bool)
451
-
"no dev branch" true
452
-
(Irmin.head store ~branch:"dev" = None)
447
+
Irmin.set_head store ~branch:"dev" h2;
448
+
(* Both branches exist *)
449
+
let bs = Irmin.branches store in
450
+
Alcotest.(check bool) "has main" true (List.mem "main" bs);
451
+
Alcotest.(check bool) "has dev" true (List.mem "dev" bs);
452
+
(* Each points to its own commit *)
453
+
(match Irmin.head store ~branch:"main" with
454
+
| Some h -> Alcotest.(check bool) "main head" true (Irmin.Hash.equal h h1)
455
+
| None -> Alcotest.fail "main head missing");
456
+
match Irmin.head store ~branch:"dev" with
457
+
| Some h -> Alcotest.(check bool) "dev head" true (Irmin.Hash.equal h h2)
458
+
| None -> Alcotest.fail "dev head missing"
459
+
460
+
let test_pds_store_branches_survive_reopen () =
461
+
with_temp_dir @@ fun path ->
462
+
(* Session 1: create branches *)
463
+
Eio.Switch.run (fun sw ->
464
+
let pds = Pds.v ~sw path ~did:test_did in
465
+
let store = Irmin.Mst.of_pds pds in
466
+
let tree = Irmin.Tree.add Irmin.Tree.empty [ "x" ] "y" in
467
+
let h =
468
+
Irmin.commit store ~tree ~parents:[] ~message:"init" ~author:"t"
469
+
in
470
+
Irmin.set_head store ~branch:"main" h;
471
+
Irmin.set_head store ~branch:"feature" h;
472
+
Pds.close pds);
473
+
(* Session 2: verify branches persist *)
474
+
Eio.Switch.run @@ fun sw ->
475
+
let pds = Pds.open_ ~sw path in
476
+
let store = Irmin.Mst.of_pds pds in
477
+
let bs = Irmin.branches store in
478
+
Alcotest.(check bool) "main persists" true (List.mem "main" bs);
479
+
Alcotest.(check bool) "feature persists" true (List.mem "feature" bs);
480
+
Pds.close pds
453
481
454
482
let suite =
455
483
( "pds_interop",
···
474
502
Alcotest.test_case "multiple collections" `Quick
475
503
test_multiple_collections_interop;
476
504
Alcotest.test_case "DAG-CBOR integrity" `Quick test_dagcbor_integrity;
505
+
Alcotest.test_case "store main branch" `Quick test_pds_store_main_branch;
506
+
Alcotest.test_case "multiple branches" `Quick
507
+
test_pds_store_multiple_branches;
508
+
Alcotest.test_case "branches survive reopen" `Quick
509
+
test_pds_store_branches_survive_reopen;
477
510
] )