Don't make all writes wait on global sequencer mutex

pegasus/lib/repository.ml

···
 module String_map = Lex.String_map
 module Tid = Mist.Tid
 
+let write_locks : (string, Lwt_mutex.t) Hashtbl.t = Hashtbl.create 100
+
+let write_lock_mutex = Lwt_mutex.create ()
+
+let with_write_lock did f =
+  let%lwt lock =
+    Lwt_mutex.with_lock write_lock_mutex (fun () ->
+        match Hashtbl.find_opt write_locks did with
+        | Some l ->
+            Lwt.return l
+        | None ->
+            let l = Lwt_mutex.create () in
+            Hashtbl.add write_locks did l ;
+            Lwt.return l )
+  in
+  Lwt_mutex.with_lock lock f
+
 module Write_op = struct
   let create = "com.atproto.repo.applyWrites#create"
 
···
 
 let apply_writes (t : t) (writes : repo_write list) (swap_commit : Cid.t option)
     : write_result Lwt.t =
-  let open Sequencer.Types in
-  let module Inductive = Mist.Mst.Inductive (Mst) in
-  let%lwt prev_commit =
-    match%lwt User_store.get_commit t.db with
-    | Some (_, commit) ->
-        Lwt.return commit
-    | None ->
-        failwith ("failed to retrieve commit for " ^ t.did)
-  in
-  if swap_commit <> None && swap_commit <> Option.map fst t.commit then
-    Errors.invalid_request ~name:"InvalidSwap"
-      (Format.sprintf "swapCommit cid %s did not match last commit cid %s"
-         (Cid.to_string (Option.get swap_commit))
-         (match t.commit with Some (c, _) -> Cid.to_string c | None -> "null") ) ;
-  let%lwt block_map = Lwt.map ref (get_map t) in
-  let cached_store = Cached_store.create t.db in
-  let mst : Cached_mst.t ref =
-    ref (Cached_mst.create cached_store prev_commit.data)
-  in
-  (* ops to emit, built in loop because prev_data (previous cid) is otherwise inaccessible *)
-  let commit_ops : commit_evt_op list ref = ref [] in
-  let added_leaves = ref Block_map.empty in
-  let%lwt results =
-    Lwt_list.map_s
-      (fun (w : repo_write) ->
-        match w with
-        | Create {collection; rkey; value; _} ->
-            let rkey = Option.value rkey ~default:(Tid.now ()) in
-            let path = Format.sprintf "%s/%s" collection rkey in
-            let uri = Format.sprintf "at://%s/%s" t.did path in
-            let%lwt () =
-              match String_map.find_opt path !block_map with
-              | Some cid ->
-                  Errors.invalid_request ~name:"InvalidSwap"
-                    (Format.sprintf
-                       "attempted to write record %s that already exists with \
-                        cid %s"
-                       path (Cid.to_string cid) )
-              | None ->
-                  Lwt.return ()
-            in
-            let record_with_type : Lex.repo_record =
-              if String_map.mem "$type" value then value
-              else String_map.add "$type" (`String collection) value
-            in
-            let%lwt cid, block =
-              User_store.put_record t.db (`LexMap record_with_type) path
-            in
-            block_map := String_map.add path cid !block_map ;
-            added_leaves := Block_map.set cid block !added_leaves ;
-            commit_ops :=
-              !commit_ops @ [{action= `Create; path; cid= Some cid; prev= None}] ;
-            let%lwt new_mst = Cached_mst.add !mst path cid in
-            mst := new_mst ;
-            let refs =
-              Util.find_blob_refs value
-              |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
-            in
-            let%lwt () =
-              match%lwt User_store.put_blob_refs t.db path refs with
-              | Ok () ->
-                  Lwt.return ()
-              | Error err ->
-                  raise err
-            in
-            Lwt.return
-              (Create
-                 {type'= "com.atproto.repo.applyWrites#createResult"; uri; cid}
-              )
-        | Update {collection; rkey; value; swap_record; _} ->
-            let path = Format.sprintf "%s/%s" collection rkey in
-            let uri = Format.sprintf "at://%s/%s" t.did path in
-            let old_cid = String_map.find_opt path !block_map in
-            ( if
-                (swap_record <> None && swap_record <> old_cid)
-                || (swap_record = None && old_cid = None)
-              then
-                let cid_str =
-                  match old_cid with
+  with_write_lock t.did (fun () ->
+      let open Sequencer.Types in
+      let module Inductive = Mist.Mst.Inductive (Mst) in
+      let%lwt prev_commit =
+        match%lwt User_store.get_commit t.db with
+        | Some (_, commit) ->
+            Lwt.return commit
+        | None ->
+            failwith ("failed to retrieve commit for " ^ t.did)
+      in
+      if swap_commit <> None && swap_commit <> Option.map fst t.commit then
+        Errors.invalid_request ~name:"InvalidSwap"
+          (Format.sprintf "swapCommit cid %s did not match last commit cid %s"
+             (Cid.to_string (Option.get swap_commit))
+             ( match t.commit with
+             | Some (c, _) ->
+                 Cid.to_string c
+             | None ->
+                 "null" ) ) ;
+      let%lwt block_map = Lwt.map ref (get_map t) in
+      let cached_store = Cached_store.create t.db in
+      let mst : Cached_mst.t ref =
+        ref (Cached_mst.create cached_store prev_commit.data)
+      in
+      (* ops to emit, built in loop because prev_data (previous cid) is otherwise inaccessible *)
+      let commit_ops : commit_evt_op list ref = ref [] in
+      let added_leaves = ref Block_map.empty in
+      let%lwt results =
+        Lwt_list.map_s
+          (fun (w : repo_write) ->
+            match w with
+            | Create {collection; rkey; value; _} ->
+                let rkey = Option.value rkey ~default:(Tid.now ()) in
+                let path = Format.sprintf "%s/%s" collection rkey in
+                let uri = Format.sprintf "at://%s/%s" t.did path in
+                let%lwt () =
+                  match String_map.find_opt path !block_map with
                   | Some cid ->
-                      Cid.to_string cid
+                      Errors.invalid_request ~name:"InvalidSwap"
+                        (Format.sprintf
+                           "attempted to write record %s that already exists \
+                            with cid %s"
+                           path (Cid.to_string cid) )
                   | None ->
-                      "null"
+                      Lwt.return ()
+                in
+                let record_with_type : Lex.repo_record =
+                  if String_map.mem "$type" value then value
+                  else String_map.add "$type" (`String collection) value
+                in
+                let%lwt cid, block =
+                  User_store.put_record t.db (`LexMap record_with_type) path
+                in
+                block_map := String_map.add path cid !block_map ;
+                added_leaves := Block_map.set cid block !added_leaves ;
+                commit_ops :=
+                  !commit_ops
+                  @ [{action= `Create; path; cid= Some cid; prev= None}] ;
+                let%lwt new_mst = Cached_mst.add !mst path cid in
+                mst := new_mst ;
+                let refs =
+                  Util.find_blob_refs value
+                  |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
                 in
-                Errors.invalid_request ~name:"InvalidSwap"
-                  (Format.sprintf "attempted to update record %s with cid %s"
-                     path cid_str ) ) ;
-            let%lwt () =
-              match old_cid with
-              | Some _ -> (
-                match%lwt User_store.get_record t.db path with
-                | Some record ->
-                    let refs =
-                      Util.find_blob_refs record.value
-                      |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
+                let%lwt () =
+                  match%lwt User_store.put_blob_refs t.db path refs with
+                  | Ok () ->
+                      Lwt.return ()
+                  | Error err ->
+                      raise err
+                in
+                Lwt.return
+                  (Create
+                     { type'= "com.atproto.repo.applyWrites#createResult"
+                     ; uri
+                     ; cid } )
+            | Update {collection; rkey; value; swap_record; _} ->
+                let path = Format.sprintf "%s/%s" collection rkey in
+                let uri = Format.sprintf "at://%s/%s" t.did path in
+                let old_cid = String_map.find_opt path !block_map in
+                ( if
+                    (swap_record <> None && swap_record <> old_cid)
+                    || (swap_record = None && old_cid = None)
+                  then
+                    let cid_str =
+                      match old_cid with
+                      | Some cid ->
+                          Cid.to_string cid
+                      | None ->
+                          "null"
                     in
-                    if not (List.is_empty refs) then
-                      let%lwt _ =
-                        User_store.delete_orphaned_blobs_by_record_path t.db
-                          path
-                      in
-                      Lwt.return_unit
-                    else Lwt.return_unit
-                | None ->
-                    Lwt.return_unit )
-              | None ->
-                  Lwt.return_unit
-            in
-            let record_with_type : Lex.repo_record =
-              if String_map.mem "$type" value then value
-              else String_map.add "$type" (`String collection) value
-            in
-            let%lwt new_cid, new_block =
-              User_store.put_record t.db (`LexMap record_with_type) path
-            in
-            added_leaves := Block_map.set new_cid new_block !added_leaves ;
-            block_map := String_map.add path new_cid !block_map ;
-            commit_ops :=
-              !commit_ops
-              @ [{action= `Update; path; cid= Some new_cid; prev= old_cid}] ;
-            let%lwt new_mst = Cached_mst.add !mst path new_cid in
-            mst := new_mst ;
-            let refs =
-              Util.find_blob_refs value
-              |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
-            in
-            let%lwt () =
-              match%lwt User_store.put_blob_refs t.db path refs with
-              | Ok () ->
-                  Lwt.return ()
-              | Error err ->
-                  raise err
-            in
-            Lwt.return
-              (Update
-                 { type'= "com.atproto.repo.applyWrites#updateResult"
-                 ; uri
-                 ; cid= new_cid } )
-        | Delete {collection; rkey; swap_record; _} ->
-            let path = Format.sprintf "%s/%s" collection rkey in
-            let cid = String_map.find_opt path !block_map in
-            ( if cid = None || (swap_record <> None && swap_record <> cid) then
-                let cid_str =
-                  match cid with
-                  | Some cid ->
-                      Cid.to_string cid
+                    Errors.invalid_request ~name:"InvalidSwap"
+                      (Format.sprintf
+                         "attempted to update record %s with cid %s" path
+                         cid_str ) ) ;
+                let%lwt () =
+                  match old_cid with
+                  | Some _ -> (
+                    match%lwt User_store.get_record t.db path with
+                    | Some record ->
+                        let refs =
+                          Util.find_blob_refs record.value
+                          |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
+                        in
+                        if not (List.is_empty refs) then
+                          let%lwt _ =
+                            User_store.delete_orphaned_blobs_by_record_path t.db
+                              path
+                          in
+                          Lwt.return_unit
+                        else Lwt.return_unit
+                    | None ->
+                        Lwt.return_unit )
                   | None ->
-                      "null"
+                      Lwt.return_unit
+                in
+                let record_with_type : Lex.repo_record =
+                  if String_map.mem "$type" value then value
+                  else String_map.add "$type" (`String collection) value
+                in
+                let%lwt new_cid, new_block =
+                  User_store.put_record t.db (`LexMap record_with_type) path
+                in
+                added_leaves := Block_map.set new_cid new_block !added_leaves ;
+                block_map := String_map.add path new_cid !block_map ;
+                commit_ops :=
+                  !commit_ops
+                  @ [{action= `Update; path; cid= Some new_cid; prev= old_cid}] ;
+                let%lwt new_mst = Cached_mst.add !mst path new_cid in
+                mst := new_mst ;
+                let refs =
+                  Util.find_blob_refs value
+                  |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
+                in
+                let%lwt () =
+                  match%lwt User_store.put_blob_refs t.db path refs with
+                  | Ok () ->
+                      Lwt.return ()
+                  | Error err ->
+                      raise err
                 in
-                Errors.invalid_request ~name:"InvalidSwap"
-                  (Format.sprintf "attempted to delete record %s with cid %s"
-                     path cid_str ) ) ;
-            let%lwt () =
-              match%lwt User_store.get_record t.db path with
-              | Some record ->
-                  let refs =
-                    Util.find_blob_refs record.value
-                    |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
-                  in
-                  if not (List.is_empty refs) then
-                    let%lwt _ =
-                      User_store.delete_orphaned_blobs_by_record_path t.db path
+                Lwt.return
+                  (Update
+                     { type'= "com.atproto.repo.applyWrites#updateResult"
+                     ; uri
+                     ; cid= new_cid } )
+            | Delete {collection; rkey; swap_record; _} ->
+                let path = Format.sprintf "%s/%s" collection rkey in
+                let cid = String_map.find_opt path !block_map in
+                ( if cid = None || (swap_record <> None && swap_record <> cid)
+                  then
+                    let cid_str =
+                      match cid with
+                      | Some cid ->
+                          Cid.to_string cid
+                      | None ->
+                          "null"
                     in
-                    Lwt.return_unit
-                  else Lwt.return_unit
-              | None ->
-                  Lwt.return_unit
-            in
-            let%lwt () = User_store.delete_record t.db path in
-            block_map := String_map.remove path !block_map ;
-            commit_ops :=
-              !commit_ops @ [{action= `Delete; path; cid= None; prev= cid}] ;
-            let%lwt new_mst = Cached_mst.delete !mst path in
-            mst := new_mst ;
-            Lwt.return
-              (Delete {type'= "com.atproto.repo.applyWrites#deleteResult"}) )
-      writes
-  in
-  let new_mst = !mst in
-  let%lwt new_commit = put_commit t new_mst.root ~previous:(Some prev_commit) in
-  let new_commit_cid, new_commit_signed = new_commit in
-  let commit_block =
-    new_commit_signed |> signed_commit_to_yojson |> Dag_cbor.encode_yojson
-  in
-  let diff : Inductive.diff list =
-    List.fold_left
-      (fun (acc : Inductive.diff list)
-           ({action; path; cid; prev} : commit_evt_op) ->
-        match action with
-        | `Create ->
-            acc @ [Add {key= path; cid= Option.get cid}]
-        | `Update ->
-            acc @ [Update {key= path; cid= Option.get cid; prev}]
-        | `Delete ->
-            acc @ [Delete {key= path; prev= Option.get prev}] )
-      [] !commit_ops
-  in
-  let%lwt proof_blocks =
-    match%lwt
-      Inductive.generate_proof !block_map diff ~new_root:new_mst.root
-        ~prev_root:prev_commit.data
-    with
-    | Ok blocks ->
-        Lwt.return (Block_map.merge blocks !added_leaves)
-    | Error err ->
-        raise err
-  in
-  let block_stream =
-    proof_blocks |> Block_map.entries |> Lwt_seq.of_list
-    |> Lwt_seq.cons (new_commit_cid, commit_block)
-  in
-  let%lwt blocks =
-    Car.blocks_to_stream new_commit_cid block_stream |> Car.collect_stream
-  in
-  let%lwt ds = Data_store.connect () in
-  let%lwt _ =
-    Sequencer.sequence_commit ds ~did:t.did ~commit:new_commit_cid
-      ~rev:new_commit_signed.rev ~blocks ~ops:!commit_ops ~since:prev_commit.rev
-      ~prev_data:prev_commit.data ()
-  in
-  Lwt.return {commit= new_commit; results}
+                    Errors.invalid_request ~name:"InvalidSwap"
+                      (Format.sprintf
+                         "attempted to delete record %s with cid %s" path
+                         cid_str ) ) ;
+                let%lwt () =
+                  match%lwt User_store.get_record t.db path with
+                  | Some record ->
+                      let refs =
+                        Util.find_blob_refs record.value
+                        |> List.map (fun (r : Mist.Blob_ref.t) -> r.ref)
+                      in
+                      if not (List.is_empty refs) then
+                        let%lwt _ =
+                          User_store.delete_orphaned_blobs_by_record_path t.db
+                            path
+                        in
+                        Lwt.return_unit
+                      else Lwt.return_unit
+                  | None ->
+                      Lwt.return_unit
+                in
+                let%lwt () = User_store.delete_record t.db path in
+                block_map := String_map.remove path !block_map ;
+                commit_ops :=
+                  !commit_ops @ [{action= `Delete; path; cid= None; prev= cid}] ;
+                let%lwt new_mst = Cached_mst.delete !mst path in
+                mst := new_mst ;
+                Lwt.return
+                  (Delete {type'= "com.atproto.repo.applyWrites#deleteResult"}) )
+          writes
+      in
+      let new_mst = !mst in
+      let%lwt new_commit =
+        put_commit t new_mst.root ~previous:(Some prev_commit)
+      in
+      let new_commit_cid, new_commit_signed = new_commit in
+      let commit_block =
+        new_commit_signed |> signed_commit_to_yojson |> Dag_cbor.encode_yojson
+      in
+      let diff : Inductive.diff list =
+        List.fold_left
+          (fun (acc : Inductive.diff list)
+               ({action; path; cid; prev} : commit_evt_op) ->
+            match action with
+            | `Create ->
+                acc @ [Add {key= path; cid= Option.get cid}]
+            | `Update ->
+                acc @ [Update {key= path; cid= Option.get cid; prev}]
+            | `Delete ->
+                acc @ [Delete {key= path; prev= Option.get prev}] )
+          [] !commit_ops
+      in
+      let%lwt proof_blocks =
+        match%lwt
+          Inductive.generate_proof !block_map diff ~new_root:new_mst.root
+            ~prev_root:prev_commit.data
+        with
+        | Ok blocks ->
+            Lwt.return (Block_map.merge blocks !added_leaves)
+        | Error err ->
+            raise err
+      in
+      let block_stream =
+        proof_blocks |> Block_map.entries |> Lwt_seq.of_list
+        |> Lwt_seq.cons (new_commit_cid, commit_block)
+      in
+      let%lwt blocks =
+        Car.blocks_to_stream new_commit_cid block_stream |> Car.collect_stream
+      in
+      let%lwt ds = Data_store.connect () in
+      let%lwt _ =
+        Sequencer.sequence_commit ds ~did:t.did ~commit:new_commit_cid
+          ~rev:new_commit_signed.rev ~blocks ~ops:!commit_ops
+          ~since:prev_commit.rev ~prev_data:prev_commit.data ()
+      in
+      Lwt.return {commit= new_commit; results} )
 
 let load ?write ?create ?(ensure_active = false) ?ds did : t Lwt.t =
   let%lwt data_store_conn =

pegasus/lib/sequencer.ml

···
   let raw = Dag_cbor.encode_yojson @@ Encode.format_commit evt in
   let%lwt seq = DB.append_event conn ~t:`Commit ~time:time_ms ~data:raw in
   let frame = Frame.encode_message ~seq ~time:time_iso (Commit evt) in
-  let%lwt () = Bus.publish {seq; bytes= frame} in
+  Lwt.async (fun () -> Bus.publish {seq; bytes= frame}) ;
   Lwt.return seq
 
 let sequence_sync (conn : Data_store.t) ~(did : string) ~(rev : string)
···
   let raw = Dag_cbor.encode_yojson @@ Encode.format_sync evt in
   let%lwt seq = DB.append_event conn ~t:`Sync ~time:time_ms ~data:raw in
   let frame = Frame.encode_message ~seq ~time:time_iso (Sync evt) in
-  let%lwt () = Bus.publish {seq; bytes= frame} in
+  Lwt.async (fun () -> Bus.publish {seq; bytes= frame}) ;
   Lwt.return seq
 
 let sequence_identity (conn : Data_store.t) ~(did : string)
···
   let raw = Dag_cbor.encode_yojson @@ Encode.format_identity evt in
   let%lwt seq = DB.append_event conn ~t:`Identity ~time:time_ms ~data:raw in
   let frame = Frame.encode_message ~seq ~time:time_iso (Identity evt) in
-  let%lwt () = Bus.publish {seq; bytes= frame} in
+  Lwt.async (fun () -> Bus.publish {seq; bytes= frame}) ;
   Lwt.return seq
 
 let sequence_account (conn : Data_store.t) ~(did : string) ~(active : bool)
···
   let raw = Dag_cbor.encode_yojson @@ Encode.format_account evt in
   let%lwt seq = DB.append_event conn ~t:`Account ~time:time_ms ~data:raw in
   let frame = Frame.encode_message ~seq ~time:time_iso (Account evt) in
-  let%lwt () = Bus.publish {seq; bytes= frame} in
+  Lwt.async (fun () -> Bus.publish {seq; bytes= frame}) ;
   Lwt.return seq