aboutsummaryrefslogtreecommitdiffstats
path: root/lib/management/physical.ml
blob: b8963302489b230f8f4a6c554d21ef2c363b6f5c (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
(** Physical storage abstraction.

    This module defines a functor-based storage interface that can be backed by
    any key-value store (etcd, Redis, DynamoDB, PostgreSQL, filesystem, etc.).

    All data is content-addressed: keys are hashes, values are serialized
    objects. The storage is append-only - objects are never modified, only new
    versions are written with new hashes. *)

(** Backend module signature that any storage driver must implement *)
module type BACKEND = sig
  include Configuration.CONFIGURABLE

  type connection

  (** Connect to the storage backend *)
  val connect : configuration -> (connection, Condition.t) result

  (** Disconnect from the backend *)
  val disconnect : connection -> unit

  (** Get a value by its hash. Returns None if not found. *)
  val get : connection -> Conventions.Hash.t -> (bytes option, Condition.t) result

  (** Store a value at its hash *)
  val put : connection -> Conventions.Hash.t -> bytes -> (unit, Condition.t) result

  (** Check if a hash exists *)
  val exists : connection -> Conventions.Hash.t -> (bool, Condition.t) result

  (** Begin a transaction *)
  val begin_transaction : connection -> (unit, Condition.t) result

  (** Commit a transaction *)
  val commit : connection -> (unit, Condition.t) result

  (** Rollback a transaction *)
  val rollback : connection -> (unit, Condition.t) result

  (** Batch get multiple values *)
  val get_many : connection -> Conventions.Hash.t list -> (bytes option list, Condition.t) result

  (** Batch put multiple values *)
  val put_many : connection -> (Conventions.Hash.t * bytes) list -> (unit, Condition.t) result
end

(** Storage operations built on top of a backend *)
module type S = sig
  include Configuration.CONFIGURABLE

  type t

  (** Create a storage instance from a backend connection *)
  val create : configuration -> (t, Condition.t) result

  (** Close the storage *)
  val close : t -> unit

  (** Store an attribute value, returns its hash *)
  val store_attribute : t -> Conventions.AbstractValue.t -> (Conventions.Hash.t, Condition.t) result

  (** Load an attribute value by hash *)
  val load_attribute :
    t -> Conventions.Hash.t -> (Conventions.AbstractValue.t option, Condition.t) result

  (** Store raw bytes at a hash *)
  val store_raw : t -> Conventions.Hash.t -> bytes -> (unit, Condition.t) result

  (** Load raw bytes by hash *)
  val load_raw : t -> Conventions.Hash.t -> (bytes option, Condition.t) result

  (** Check if a hash exists *)
  val exists : t -> Conventions.Hash.t -> (bool, Condition.t) result

  (** Execute a function within a transaction *)
  val with_transaction : t -> (unit -> ('a, Condition.t) result) -> ('a, Condition.t) result
end

(** Functor to create storage operations from a backend *)
module Make (B : BACKEND) : S with type configuration = B.configuration = struct
  type configuration = B.configuration
  type t = B.connection

  let parse = B.parse
  let create config = B.connect config
  let close = B.disconnect

  let store_attribute conn value =
    let bytes = Marshal.to_bytes value [] in
    let hash = Conventions.AbstractValue.hash value in
    match B.put conn hash bytes with Ok () -> Ok hash | Error e -> Error e

  let load_attribute conn hash =
    match B.get conn hash with
    | Ok (Some bytes) -> Ok (Some (Marshal.from_bytes bytes 0))
    | Ok None -> Ok None
    | Error e -> Error e

  let store_raw conn hash bytes = B.put conn hash bytes
  let load_raw conn hash = B.get conn hash
  let exists conn hash = B.exists conn hash

  let with_transaction conn f =
    match B.begin_transaction conn with
    | Error e -> Error e
    | Ok () -> (
      match f () with
      | Ok result -> ( match B.commit conn with Ok () -> Ok result | Error e -> Error e )
      | Error e ->
          let _ = B.rollback conn in
          Error e )
end

(** In-memory backend for testing and development *)
module MemoryBackend : BACKEND with type configuration = unit = struct
  type configuration = unit

  type connection =
    { data: (string, bytes) Hashtbl.t;
      mutable in_transaction: bool;
      mutable transaction_buffer: (string * bytes) list }

  module Error = struct
    open Condition

    (* TODO: more structure *)
    let parse_error msg = condition "parse-error" msg empty

    let already_in_transaction =
      condition "already-in-transaction" "Already inside a transaction" empty

    let not_in_transaction = condition "not-in-transaction" "Not inside a transaction" empty
  end

  let parse sexp =
    match sexp with
    | Sexplib.Sexp.List [] -> Ok ()
    | _ ->
        Error
          (Error.parse_error
             (Printf.sprintf "memory backend takes no configuration, got: %s"
                (Sexplib.Sexp.to_string sexp) ) )

  let connect () = Ok {data= Hashtbl.create 1024; in_transaction= false; transaction_buffer= []}
  let disconnect _ = ()
  let get conn hash = Ok (Hashtbl.find_opt conn.data hash)

  let put conn hash value =
    if conn.in_transaction then begin
      conn.transaction_buffer <- (hash, value) :: conn.transaction_buffer;
      Ok ()
    end
    else begin
      Hashtbl.replace conn.data hash value;
      Ok ()
    end

  let exists conn hash = Ok (Hashtbl.mem conn.data hash)

  let begin_transaction conn =
    if conn.in_transaction then Error Error.already_in_transaction
    else begin
      conn.in_transaction <- true;
      conn.transaction_buffer <- [];
      Ok ()
    end

  let commit conn =
    if not conn.in_transaction then Error Error.not_in_transaction
    else begin
      List.iter (fun (hash, value) -> Hashtbl.replace conn.data hash value) conn.transaction_buffer;
      conn.in_transaction <- false;
      conn.transaction_buffer <- [];
      Ok ()
    end

  let rollback conn =
    if not conn.in_transaction then Error Error.not_in_transaction
    else begin
      conn.in_transaction <- false;
      conn.transaction_buffer <- [];
      Ok ()
    end

  let get_many conn hashes = Ok (List.map (fun h -> Hashtbl.find_opt conn.data h) hashes)

  let put_many conn pairs =
    List.iter
      (fun (hash, value) ->
        if conn.in_transaction then
          conn.transaction_buffer <- (hash, value) :: conn.transaction_buffer
        else Hashtbl.replace conn.data hash value )
      pairs;
    Ok ()
end

(** Default in-memory storage for development *)
module Memory = Make (MemoryBackend)