objective categorical abstract machine language personal data server
Actually use transaction connection when importing car
+12
-8
+4
-4
ipld/lib/car.ml
+4
-4
ipld/lib/car.ml
···
95
95
let q : bytes option Lwt_mvar.t = Lwt_mvar.create_empty () in
96
96
let () =
97
97
Lwt.async (fun () ->
98
-
let%lwt () =
99
-
Lwt_seq.iter_s (fun chunk -> Lwt_mvar.put q (Some chunk)) stream
100
-
in
101
-
Lwt_mvar.put q None )
98
+
Lwt.finalize
99
+
(fun () ->
100
+
Lwt_seq.iter_s (fun chunk -> Lwt_mvar.put q (Some chunk)) stream )
101
+
(fun () -> Lwt_mvar.put q None) )
102
102
in
103
103
let buf = ref Bytes.empty in
104
104
let pos = ref 0 in
+8
-4
pegasus/lib/repository.ml
+8
-4
pegasus/lib/repository.ml
···
536
536
Lwt.return @@ Car.blocks_to_stream root all_blocks
537
537
538
538
let import_car t (stream : Car.stream) : (t, exn) Lwt_result.t =
539
+
let open Util.Syntax in
539
540
let%lwt roots, blocks_seq = Car.read_car_stream stream in
540
541
let root =
541
542
match roots with [root] -> root | _ -> failwith "invalid number of roots"
···
584
585
Util.use_pool t.db.db (fun conn ->
585
586
Util.transact conn (fun () ->
586
587
(* store commit *)
587
-
let%lwt _ = User_store.put_commit t.db commit in
588
+
let$! _ = User_store.Queries.put_commit root commit_bytes conn in
588
589
(* store mst nodes *)
589
590
let%lwt () =
590
591
Lwt_list.iter_s
591
592
(fun cid ->
592
593
match Block_map.get cid all_blocks with
593
594
| Some block ->
594
-
let%lwt _ = User_store.put_block t.db cid block in
595
+
let$! _ = User_store.Queries.put_block cid block conn in
595
596
Lwt.return_unit
596
597
| None ->
597
598
Lwt.return_unit )
···
603
604
(fun (path, cid) ->
604
605
match Block_map.get cid all_blocks with
605
606
| Some data ->
606
-
User_store.put_record_raw t.db ~path ~cid ~data
607
-
~since:(Tid.now ())
607
+
let$! _ =
608
+
User_store.Queries.put_record ~path ~cid ~data
609
+
~since:(Tid.now ()) conn
610
+
in
611
+
Lwt.return_unit
608
612
| None ->
609
613
failwith ("missing record block: " ^ Cid.to_string cid) )
610
614
leaves