diff --git a/src/bin_dal_node/slot_headers_store.ml b/src/bin_dal_node/slot_headers_store.ml
index fd80adae6cb876fc9da60b93510ebc623865928d..74f54fb4ce8ed5425a5fa0e0f72f999beef51118 100644
--- a/src/bin_dal_node/slot_headers_store.ml
+++ b/src/bin_dal_node/slot_headers_store.ml
@@ -32,9 +32,11 @@ end)
 
 module Store = Store_utils.Make (IStore)
 
-type t = Store.t
+(* TODO: https://gitlab.com/tezos/tezos/-/issues/4154
+   Use type parameter to track effects. *)
+type t = [`Read | `Write] Store.t
 
-let load = IStore.load
+let load = IStore.load Read_write
 
 (* Published slot headers per block hash,
    stored as a list of bindings from `Dal_slot_index.t`
diff --git a/src/lib_layer2_store/delayed_write_monad.ml b/src/lib_layer2_store/delayed_write_monad.ml
new file mode 100644
index 0000000000000000000000000000000000000000..91a8bb7899c87b2dd678fb38652158c39af7ccbc
--- /dev/null
+++ b/src/lib_layer2_store/delayed_write_monad.ml
@@ -0,0 +1,96 @@
+(*****************************************************************************)
+(*                                                                           *)
+(* Open Source License                                                       *)
+(* Copyright (c) 2022 Nomadic Labs, <contact@nomadic-labs.com>               *)
+(*                                                                           *)
+(* Permission is hereby granted, free of charge, to any person obtaining a   *)
+(* copy of this software and associated documentation files (the "Software"),*)
+(* to deal in the Software without restriction, including without limitation *)
+(* the rights to use, copy, modify, merge, publish, distribute, sublicense,  *)
+(* and/or sell copies of the Software, and to permit persons to whom the     *)
+(* Software is furnished to do so, subject to the following conditions:      *)
+(*                                                                           *)
+(* The above copyright notice and this permission notice shall be included   *)
+(* in all copies or substantial portions of the Software.                    *)
+(*                                                                           *)
+(* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR*)
+(* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,  *)
+(* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL   *)
+(* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER*)
+(* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING   *)
+(* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER       *)
+(* DEALINGS IN THE SOFTWARE.                                                 *)
+(*                                                                           *)
+(*****************************************************************************)
+
+type ('a, 'store) t = {result : 'a; write : ('store -> unit Lwt.t) option}
+
+let no_write result = {result; write = None}
+
+let delay_write result write = {result; write = Some write}
+
+let map f {result; write} = {result = f result; write}
+
+let map_es f {result; write} =
+  let open Lwt_result_syntax in
+  let+ result = f result in
+  {result; write}
+
+let bind f {result; write = write1} =
+  let open Lwt_syntax in
+  let {result; write = write2} = f result in
+  let write =
+    match (write1, write2) with
+    | None, None -> None
+    | Some write, None | None, Some write -> Some write
+    | Some write1, Some write2 ->
+        Some
+          (fun node_ctxt ->
+            let* () = write1 node_ctxt in
+            write2 node_ctxt)
+  in
+  {result; write}
+
+let apply node_ctxt {result; write} =
+  let open Lwt_syntax in
+  let+ () =
+    match write with None -> return_unit | Some write -> write node_ctxt
+  in
+  result
+
+let ignore {result; _} = result
+
+module Lwt_result_syntax = struct
+  let bind f a =
+    let open Lwt_result_syntax in
+    let* a = a in
+    let* b = f a.result in
+    let write =
+      match (a.write, b.write) with
+      | None, None -> None
+      | Some write, None | None, Some write -> Some write
+      | Some write1, Some write2 ->
+          Some
+            (fun node_ctxt ->
+              let open Lwt_syntax in
+              let* () = write1 node_ctxt in
+              write2 node_ctxt)
+    in
+    return {result = b.result; write}
+
+  let map f a = Lwt_result.map (map f) a
+
+  let ( let>* ) a f = bind f a
+
+  let ( let>+ ) a f = map f a
+
+  let return x = Lwt_result.return (no_write x)
+
+  let list_fold_left_i_es f acc l =
+    List.fold_left_i_es
+      (fun i acc x ->
+        let>* acc = Lwt.return_ok acc in
+        f i acc x)
+      (no_write acc)
+      l
+end
diff --git a/src/lib_layer2_store/delayed_write_monad.mli b/src/lib_layer2_store/delayed_write_monad.mli
new file mode 100644
index 0000000000000000000000000000000000000000..19a96d2640e7fd264f822ee61d62605131f64983
--- /dev/null
+++ b/src/lib_layer2_store/delayed_write_monad.mli
@@ -0,0 +1,93 @@
+(*****************************************************************************)
+(*                                                                           *)
+(* Open Source License                                                       *)
+(* Copyright (c) 2022 Nomadic Labs, <contact@nomadic-labs.com>               *)
+(*                                                                           *)
+(* Permission is hereby granted, free of charge, to any person obtaining a   *)
+(* copy of this software and associated documentation files (the "Software"),*)
+(* to deal in the Software without restriction, including without limitation *)
+(* the rights to use, copy, modify, merge, publish, distribute, sublicense,  *)
+(* and/or sell copies of the Software, and to permit persons to whom the     *)
+(* Software is furnished to do so, subject to the following conditions:      *)
+(*                                                                           *)
+(* The above copyright notice and this permission notice shall be included   *)
+(* in all copies or substantial portions of the Software.                    *)
+(*                                                                           *)
+(* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR*)
+(* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,  *)
+(* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL   *)
+(* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER*)
+(* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING   *)
+(* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER       *)
+(* DEALINGS IN THE SOFTWARE.                                                 *)
+(*                                                                           *)
+(*****************************************************************************)
+
+(** {1 Monad for values with delayed write effects } *)
+
+(** Type to encapsulate values with a callback that should write to disk. This
+    allows to delay some writes to a location where we have write access to the
+    context. *)
+type ('a, 'store) t
+
+(** [no_write v] returns [v] together with a write function that does
+    nothing. *)
+val no_write : 'a -> ('a, _) t
+
+(** [delay_write v write] returns a delayed write version of [v] with the
+    [write] function to be called later. *)
+val delay_write : 'a -> ('store -> unit Lwt.t) -> ('a, 'store) t
+
+(** [map f dw] returns the delayed write [dw] where [f] is applied to the
+    encapsulated value. *)
+val map : ('a -> 'b) -> ('a, 'store) t -> ('b, 'store) t
+
+(** [map_es] is like {!map} in the Lwt result monad. *)
+val map_es :
+  ('a -> ('b, 'c) result Lwt.t) ->
+  ('a, 'store) t ->
+  (('b, 'store) t, 'c) result Lwt.t
+
+(** [bind f dw] returns a delayed write value where [f] is applied to the
+    encapsulated value and the write effects of [f] are added to the effects of
+    [dw]. *)
+val bind : ('a -> ('b, 'store) t) -> ('a, 'store) t -> ('b, 'store) t
+
+(** [apply node_ctxt dw] applies the write effects on the context [node_ctxt]
+    and returns the encapsulated value. *)
+val apply : 'store -> ('a, 'store) t -> 'a Lwt.t
+
+(** [ignore dw] ignores the write effects and returns the encapsulated value. *)
+val ignore : ('a, _) t -> 'a
+
+(** {2 Monad for values with delayed write effects on top of the Lwt_result
+    monad } *)
+module Lwt_result_syntax : sig
+  val bind :
+    ('a -> (('b, 'store) t, 'c) result Lwt.t) ->
+    (('a, 'store) t, 'c) result Lwt.t ->
+    (('b, 'store) t, 'c) result Lwt.t
+
+  val map :
+    ('a -> 'b) ->
+    (('a, 'store) t, 'c) Lwt_result.t ->
+    (('b, 'store) t, 'c) Lwt_result.t
+
+  val ( let>* ) :
+    (('a, 'store) t, 'b) result Lwt.t ->
+    ('a -> (('c, 'store) t, 'b) result Lwt.t) ->
+    (('c, 'store) t, 'b) result Lwt.t
+
+  val ( let>+ ) :
+    (('a, 'store) t, 'b) Lwt_result.t ->
+    ('a -> 'c) ->
+    (('c, 'store) t, 'b) Lwt_result.t
+
+  val return : 'a -> (('a, 'store) t, 'b) Lwt_result.t
+
+  val list_fold_left_i_es :
+    (int -> 'a -> 'b -> (('a, 'store) t, 'trace) result Lwt.t) ->
+    'a ->
+    'b list ->
+    (('a, 'store) t, 'trace) result Lwt.t
+end
diff --git a/src/lib_layer2_store/irmin_store.ml b/src/lib_layer2_store/irmin_store.ml
index c9dae03bd9c25250652acb3c74bf697b2524151b..ccc168b1b01cd696f8ebdc4db9c3ba0ce423f59f 100644
--- a/src/lib_layer2_store/irmin_store.ml
+++ b/src/lib_layer2_store/irmin_store.ml
@@ -23,6 +23,9 @@
 (* DEALINGS IN THE SOFTWARE.                                                 *)
 (*                                                                           *)
 (*****************************************************************************)
+
+open Store_sigs
+
 module Make (N : sig
   val name : string
 end) =
@@ -33,9 +36,13 @@ struct
 
   let make_key_path path key = path @ [key]
 
-  let load data_dir =
+  type nonrec +'a t = t
+
+  let load : type a. a mode -> string -> a t Lwt.t =
+   fun mode data_dir ->
     let open Lwt_syntax in
-    let* repo = Repo.v (Irmin_pack.config data_dir) in
+    let readonly = match mode with Read_only -> true | Read_write -> false in
+    let* repo = Repo.v (Irmin_pack.config ~readonly data_dir) in
     main repo
 
   let flush store = flush (repo store)
@@ -55,4 +62,6 @@ struct
     let full_path = path_to_string path in
     let info () = info full_path in
     set_exn ~info store path bytes
+
+  let readonly = Fun.id
 end
diff --git a/src/lib_layer2_store/store_sigs.ml b/src/lib_layer2_store/store_sigs.ml
index 41b938b88c663cf50448ab560b666e2050ef103c..6e2941baf29ca4f2f4a212862d98289e647fa25f 100644
--- a/src/lib_layer2_store/store_sigs.ml
+++ b/src/lib_layer2_store/store_sigs.ml
@@ -31,72 +31,82 @@ type path = string list
 (** Keys in Map-like storage functors are represented as strings. *)
 type key_path_representation = string
 
+type rw = [`Read | `Write]
+
+type ro = [`Read]
+
+type _ mode = Read_only : ro mode | Read_write : rw mode
+
 (** [BACKEND] is the module type defining the backend for persisting data to
     disk. It is used by the functors in [Store_utils] to create Storage
     modules that persist data structures on disk. *)
 module type BACKEND = sig
   (** The type for representing storage backends that can be accessed by the
-      module. *)
-  type t
+      module. The type parameter indicates whether the storage can be only read
+      or both read and written. *)
+  type +'a t
 
   (** [make_key_path path raw_key] constructs a new path from [path]
       and the [raw_key] key_path_representation. *)
   val make_key_path : path -> key_path_representation -> path
 
   (** [load location] loads the backend storage from [location]. *)
-  val load : string -> t Lwt.t
+  val load : 'a mode -> string -> 'a t Lwt.t
 
   (** [flush location] flushes to disk a sequence of changes to the data stored
       at [location]. *)
-  val flush : t -> unit
+  val flush : [> `Write] t -> unit
 
   (** [close t] closes the storage backend [t]. *)
-  val close : t -> unit Lwt.t
+  val close : _ t -> unit Lwt.t
 
   (** [set_exn t path b] sets the contents for the store [t] at [path] to the
       sequence of bytes [b]. The write operation can fail, in which case an
       exception is thrown. *)
-  val set_exn : t -> path -> bytes -> unit Lwt.t
+  val set_exn : [> `Write] t -> path -> bytes -> unit Lwt.t
 
   (** [get t path] returns the contents for the store [t] at the location
       indicated by [path]. It can fail if [t] does not have any content
       stored at [path]. *)
-  val get : t -> path -> bytes Lwt.t
+  val get : [> `Read] t -> path -> bytes Lwt.t
 
   (** [mem t path] returns whether the storage backend [t] contains any
       data at the location indicated by [path]. *)
-  val mem : t -> path -> bool Lwt.t
+  val mem : [> `Read] t -> path -> bool Lwt.t
 
   (** [find t path] is the same as [get t path], except that an optional
       value is returned. This value is [None] if the backend storage [t]
       does not have any content stored at location [path]. *)
-  val find : t -> path -> bytes option Lwt.t
+  val find : [> `Read] t -> path -> bytes option Lwt.t
 
   (** [path_to_string] converts a path to a string. *)
   val path_to_string : path -> string
+
+  (** [readonly t] returns a read only version of the storage [t]. *)
+  val readonly : [> `Read] t -> [`Read] t
 end
 
 (** Module type respresenting a [Mutable_value] that is persisted on store. *)
 module type Mutable_value = sig
   (** The type of the [store] that is used for persisting data on disk. *)
-  type store
+  type +'a store
 
   (** The type of the values that will be persisted. *)
   type value
 
   (** [set store value] persists [value] for this [Mutable_value] on [store]. *)
-  val set : store -> value -> unit Lwt.t
+  val set : [> `Write] store -> value -> unit Lwt.t
 
   (** [get store] retrieves the value persisted in [store] for this
       [Mutable_value]. If the underlying storage backend fails to retrieve the
       contents of the mutable value by throwing an exception, then the
       exception is propagated by [get store].  *)
-  val get : store -> value Lwt.t
+  val get : [> `Read] store -> value Lwt.t
 
   (** [find store] returns an optional value containing the value persisted
         in [store] for this [Mutable_value]. If no value is persisted for the
         [Mutable_value], [None] is returned. *)
-  val find : store -> value option Lwt.t
+  val find : [> `Read] store -> value option Lwt.t
 end
 
 (** This module contains information about where to store and retrieve contents
@@ -134,7 +144,7 @@ end
 (** Generic module type for maps to be persisted on disk. *)
 module type Map = sig
   (** The type of the [store] that is used for persisting data on disk. *)
-  type store
+  type +'a store
 
   (** The type of keys persisted by the map. *)
   type key
@@ -144,27 +154,37 @@ module type Map = sig
 
   (** [mem store key] checks whether there is a binding of the map for key [key]
       in [store]. *)
-  val mem : store -> key -> bool Lwt.t
+  val mem : [> `Read] store -> key -> bool Lwt.t
 
   (** [get store key] retrieves from [store] the value associated with [key] in
       the map. It raises an error if such a value does not exist. *)
-  val get : store -> key -> value Lwt.t
+  val get : [> `Read] store -> key -> value Lwt.t
 
   (** [find store key] retrieves from [store] the value associated with [key]
       in the map. If the value exists it is returned as an optional value.
       Otherwise, [None] is returned. *)
-  val find : store -> key -> value option Lwt.t
+  val find : [> `Read] store -> key -> value option Lwt.t
 
   (** [find_with_default ~on_default store key] retrieves from [store] the
-      value associated with [key] in the map. 
-      If the value exists it is returned as is. 
+      value associated with [key] in the map.
+      If the value exists it is returned as is.
       Otherwise, [on_default] is returned. *)
   val find_with_default :
-    store -> key -> on_default:(unit -> value) -> value Lwt.t
+    [> `Read] store -> key -> on_default:(unit -> value) -> value Lwt.t
 
-  (** [add store key value] adds a binding from [key] to [value] to
-      the map, and persists it to disk. *)
-  val add : store -> key -> value -> unit Lwt.t
+  (** [add store key value] adds a binding from [key] to [value] to the map, and
+      persists it to disk. *)
+  val add : [> `Write] store -> key -> value -> unit Lwt.t
+end
+
+(** Generic module type for append-only maps to be persisted on disk. *)
+module type Append_only_map = sig
+  include Map
+
+  (** [add store key value] adds a binding from [key] to [value] to the map, and
+      persists it to disk. If [key] already exists in the store, it must be
+      bound to [value]. *)
+  val add : rw store -> key -> value -> unit Lwt.t
 end
 
 (** [Nested_map] is a map where values are indexed by both a primary and
@@ -175,7 +195,7 @@ end
 *)
 module type Nested_map = sig
   (** The type of the [store] that is used for persisting data on disk. *)
-  type store
+  type +'a store
 
   (** [primary_key] is the type of primary keys for the [Nested_map]. *)
   type primary_key
@@ -197,7 +217,7 @@ module type Nested_map = sig
       value for the nested map persisted on [store] for the nested map,
       indexed by [primary_key] and then by [secondary_key]. *)
   val mem :
-    store ->
+    [> `Read] store ->
     primary_key:primary_key ->
     secondary_key:secondary_key ->
     bool Lwt.t
@@ -207,7 +227,7 @@ module type Nested_map = sig
       any. If such a value does not exist, it raises the exception
       [Get_failed {primary_key; secondary_key}]. *)
   val get :
-    store ->
+    [> `Read] store ->
     primary_key:primary_key ->
     secondary_key:secondary_key ->
     value Lwt.t
@@ -218,7 +238,7 @@ module type Nested_map = sig
       [None] if there is not a value bound to [primary_key] and [seconary_key]
       in the [store] for the [Nested_map]. *)
   val find :
-    store ->
+    [> `Read] store ->
     primary_key:primary_key ->
     secondary_key:secondary_key ->
     value option Lwt.t
@@ -227,24 +247,27 @@ module type Nested_map = sig
       of bindings of the nested map that share the same [~primary_key]. For
       each of these bindings, both the secondary_key and value are returned. *)
   val list_secondary_keys_with_values :
-    store -> primary_key:primary_key -> (secondary_key * value) list Lwt.t
+    [> `Read] store ->
+    primary_key:primary_key ->
+    (secondary_key * value) list Lwt.t
 
   (** [list_secondary_keys store ~primary_key] retrieves from [store]
       the list of secondary_keys for which a value indexed by both
       [primary_key] and secondary key is persisted on disk. *)
   val list_secondary_keys :
-    store -> primary_key:primary_key -> secondary_key list Lwt.t
+    [> `Read] store -> primary_key:primary_key -> secondary_key list Lwt.t
 
   (** [list_values store ~primary_key] retrieves from [store] the list of
       values for which a binding with primary key [primary_key] and
       arbitrary secondary key exists. *)
-  val list_values : store -> primary_key:primary_key -> value list Lwt.t
+  val list_values :
+    [> `Read] store -> primary_key:primary_key -> value list Lwt.t
 
   (** [add store ~primary_key ~secondary_key value] persists [value] to
       disk. The value is bound to the [primary_key] and [secondary_key].
   *)
   val add :
-    store ->
+    rw store ->
     primary_key:primary_key ->
     secondary_key:secondary_key ->
     value ->
@@ -268,55 +291,54 @@ module type COMPARABLE_KEY = sig
 end
 
 module type Store = sig
-  type t
-
-  (** [Make_updatable_map(S)(K)(V)] constructs a [Map] which can be persisted
-    on store. The module [B] defines the underlying store that will be used the
-    map on disk. The module [S] defines storage-dependent information about how
-    the map will be saved on and retrieved from the store (for example, it
-    defines the map location in the store). The module [K] defines the
-    information related to keys of the map, and the module [V] contains
-    information about how values will be stored to and retrieved from the
-    store. The resulting map allows to update the contents of an existing value
-    for a key.
+  type +'a t
+
+  (** [Make_updatable_map(S)(K)(V)] constructs a [Map] which can be persisted on
+      store. The module [S] defines storage-dependent information about how the
+      map will be saved on and retrieved from the store (for example, it defines
+      the map location in the store). The module [K] defines the information
+      related to keys of the map, and the module [V] contains information about
+      how values will be stored to and retrieved from the store. The resulting
+      map allows to update the contents of an existing value for a key.
 *)
   module Make_updatable_map (S : STORAGE_INFO) (K : KEY) (V : VALUE) :
-    Map with type store = t and type key = K.key and type value = V.value
-
-  (** [Make_append_only_map(B)(S)(K)(V)] constructs a [Map] which can be
-    persisted on store. The module [B] defines the underlying store that will
-    be used for the map on disk. The module [S] defines storage-dependent
-    information about how the map will be saved on and retrieved from the store
-    (for example, it defines the map location in the store). The module [K]
-    contains information related to keys of the map, and the module [V]
-    contains information about how values will be stored to and retrieved from
-    the store. The resulting map forbids updating the contents of an existing
-    value with a new value, different from the previous one.
+    Map with type 'a store = 'a t and type key = K.key and type value = V.value
+
+  (** [Make_append_only_map(S)(K)(V)] constructs an [Append_only_map] which can be
+      persisted on store. The module [S] defines storage-dependent information
+      about how the map will be saved on and retrieved from the store (for
+      example, it defines the map location in the store). The module [K]
+      contains information related to keys of the map, and the module [V]
+      contains information about how values will be stored to and retrieved from
+      the store. The resulting map forbids updating the contents of an existing
+      value with a new value, different from the previous one.
 *)
   module Make_append_only_map (S : STORAGE_INFO) (K : KEY) (V : VALUE) :
-    Map with type store = t and type key = K.key and type value = V.value
+    Append_only_map
+      with type 'a store = 'a t
+       and type key = K.key
+       and type value = V.value
 
   (** [Make_mutable_value(S)(V)] constructs a [Mutable_value] for persisting a
-    mutable value in a store. The underlying backend for the store is defined
-    by the module parameter [B]. The module parameter [S] defines the location
-    of the  mutable value in the store, and the module parameter [V] contains
-    information about the type of values that the constructed module will
-    persist in the underlying store. *)
+      mutable value in a store. The module parameter [S] defines the location of
+      the mutable value in the store, and the module parameter [V] contains
+      information about the type of values that the constructed module will
+      persist in the underlying store. *)
   module Make_mutable_value (S : STORAGE_INFO) (V : VALUE) :
-    Mutable_value with type store = t and type value = V.value
+    Mutable_value with type 'a store = 'a t and type value = V.value
 
-  (** Make_nested_map(B)(S)(K1)(K2)(V) constructs a [Nested_map]  module using
-    module parameter [B] to define the storage backend, module parameter [S]
-    to define where the map is going to be persisted on store, [K1] and [K2]
-    to define the primary and secondary key, respectively, and [V] to define
-    the values of the resulting [Nested_map]. *)
+  (** Make_nested_map(S)(K1)(K2)(V) constructs a [Nested_map] module using
+      module parameter [S] to define where the map is going to be persisted on
+      store, [K1] and [K2] to define the primary and secondary key,
+      respectively, and [V] to define the values of the resulting
+      [Nested_map]. *)
   module Make_nested_map
       (S : STORAGE_INFO)
       (K1 : KEY)
       (K2 : COMPARABLE_KEY)
       (V : VALUE) :
     Nested_map
-      with type store = t
+      with type 'a store = 'a t
        and type primary_key = K1.key
        and type secondary_key = K2.key
        and type value = V.value
diff --git a/src/lib_layer2_store/store_utils.ml b/src/lib_layer2_store/store_utils.ml
index 9c827644795c26273f14e9e2afa6fd0335aa993d..583f71fe18e3aa72e8f23bac9bfedc91312a33c7 100644
--- a/src/lib_layer2_store/store_utils.ml
+++ b/src/lib_layer2_store/store_utils.ml
@@ -27,12 +27,12 @@
 open Store_sigs
 
 module Make (B : BACKEND) = struct
-  type t = B.t
+  type 'a t = 'a B.t
 
   module Make_map (S : STORAGE_INFO) (K : KEY) (V : VALUE) = struct
     type key = K.key
 
-    type store = B.t
+    type 'a store = 'a B.t
 
     type value = V.value
 
@@ -94,7 +94,7 @@ module Make (B : BACKEND) = struct
   end
 
   module Make_mutable_value (S : STORAGE_INFO) (V : VALUE) = struct
-    type store = B.t
+    type 'a store = 'a B.t
 
     type value = V.value
 
@@ -124,7 +124,7 @@ module Make (B : BACKEND) = struct
       (K2 : COMPARABLE_KEY)
       (V : VALUE) =
   struct
-    type store = B.t
+    type 'a store = 'a B.t
 
     type primary_key = K1.key
 
diff --git a/src/lib_layer2_store/store_utils.mli b/src/lib_layer2_store/store_utils.mli
index ae43c9ad43fab1789aad111309216dba4ec1dc3a..60ab4b40e9e82faf02647539fc85c79ec3f351c1 100644
--- a/src/lib_layer2_store/store_utils.mli
+++ b/src/lib_layer2_store/store_utils.mli
@@ -26,4 +26,4 @@
 
 open Store_sigs
 
-module Make (B : BACKEND) : Store_sigs.Store with type t = B.t
+module Make (B : BACKEND) : Store_sigs.Store with type +'a t = 'a B.t
diff --git a/src/proto_alpha/bin_sc_rollup_client/RPC.ml b/src/proto_alpha/bin_sc_rollup_client/RPC.ml
index 602a7ccd75bda60b0bfea3fd7ba10f2505003802..98ad7f965b46660dda732b561bdb4f62009a0d25 100644
--- a/src/proto_alpha/bin_sc_rollup_client/RPC.ml
+++ b/src/proto_alpha/bin_sc_rollup_client/RPC.ml
@@ -30,7 +30,7 @@ let get_state_value_command cctxt block key =
   Sc_rollup_services.Global.Block.(make_call1 state_value) cctxt block {key} ()
 
 let get_outbox_proof cctxt serialized_output =
-  Sc_rollup_services.Global.(make_call outbox_proof)
+  Sc_rollup_services.Global.Helpers.(make_call outbox_proof)
     cctxt
     ()
     serialized_output
diff --git a/src/proto_alpha/bin_sc_rollup_node/RPC_server.ml b/src/proto_alpha/bin_sc_rollup_node/RPC_server.ml
index 229669e76b098bae4bb6d4d73db2c961c4931dbc..4ad0a4e9a253e82500dfa2335f6334bf9c86a819 100644
--- a/src/proto_alpha/bin_sc_rollup_node/RPC_server.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/RPC_server.ml
@@ -120,7 +120,7 @@ module type PARAM = sig
 
   type context
 
-  val context_of_prefix : Node_context.t -> prefix -> context tzresult Lwt.t
+  val context_of_prefix : Node_context.rw -> prefix -> context tzresult Lwt.t
 end
 
 module Make_directory (S : PARAM) = struct
@@ -148,7 +148,17 @@ end
 module Global_directory = Make_directory (struct
   include Sc_rollup_services.Global
 
-  type context = Node_context.t
+  type context = Node_context.ro
+
+  let context_of_prefix node_ctxt () = return (Node_context.readonly node_ctxt)
+end)
+
+module Proof_helpers_directory = Make_directory (struct
+  include Sc_rollup_services.Global.Helpers
+
+  (* The context needs to be accessed with write permissions because we need to
+     commit on disk to generate the proofs. *)
+  type context = Node_context.rw
 
   let context_of_prefix node_ctxt () = return node_ctxt
 end)
@@ -156,15 +166,15 @@ end)
 module Local_directory = Make_directory (struct
   include Sc_rollup_services.Local
 
-  type context = Node_context.t
+  type context = Node_context.ro
 
-  let context_of_prefix node_ctxt () = return node_ctxt
+  let context_of_prefix node_ctxt () = return (Node_context.readonly node_ctxt)
 end)
 
 module Block_directory = Make_directory (struct
   include Sc_rollup_services.Global.Block
 
-  type context = Node_context.t * Tezos_crypto.Block_hash.t
+  type context = Node_context.ro * Tezos_crypto.Block_hash.t
 
   let context_of_prefix node_ctxt (((), block) : prefix) =
     let open Lwt_result_syntax in
@@ -176,7 +186,7 @@ module Block_directory = Make_directory (struct
       | `Finalized -> get_finalized node_ctxt.Node_context.store
       | `Cemented -> get_last_cemented node_ctxt.Node_context.store
     in
-    (node_ctxt, block)
+    (Node_context.readonly node_ctxt, block)
 end)
 
 module Common = struct
@@ -223,7 +233,7 @@ module Make (PVM : Pvm.S) = struct
   module PVM = PVM
   module Outbox = Outbox.Make (PVM)
 
-  let get_state (node_ctxt : Node_context.t) block_hash =
+  let get_state (node_ctxt : _ Node_context.t) block_hash =
     let open Lwt_result_syntax in
     let* ctxt = Node_context.checkout_context node_ctxt block_hash in
     let*! state = PVM.State.find ctxt in
@@ -329,7 +339,8 @@ module Make (PVM : Pvm.S) = struct
     return outbox
 
   let () =
-    Global_directory.register0 Sc_rollup_services.Global.outbox_proof
+    Proof_helpers_directory.register0
+      Sc_rollup_services.Global.Helpers.outbox_proof
     @@ fun node_ctxt output () -> Outbox.proof_of_output node_ctxt output
 
   let register node_ctxt =
@@ -340,6 +351,7 @@ module Make (PVM : Pvm.S) = struct
         Global_directory.build_directory;
         Local_directory.build_directory;
         Block_directory.build_directory;
+        Proof_helpers_directory.build_directory;
       ]
 
   let start node_ctxt configuration =
diff --git a/src/proto_alpha/bin_sc_rollup_node/RPC_server.mli b/src/proto_alpha/bin_sc_rollup_node/RPC_server.mli
index e836b81e18134d5be945c6046b07ec254382fa55..2bb1a448b1ec896c1453947dd377292cc3378584 100644
--- a/src/proto_alpha/bin_sc_rollup_node/RPC_server.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/RPC_server.mli
@@ -31,7 +31,7 @@ module Make (PVM : Pvm.S) : sig
   (** [start node_ctxt config] starts an RPC server listening for requests on
       the port [config.rpc_port] and address [config.rpc_addr]. *)
   val start :
-    Node_context.t -> Configuration.t -> RPC_server.server tzresult Lwt.t
+    Node_context.rw -> Configuration.t -> RPC_server.server tzresult Lwt.t
 
   (** Shutdown a running RPC server. When this function is called, the rollup
       node will stop listening to incoming requests. *)
diff --git a/src/proto_alpha/bin_sc_rollup_node/commitment.ml b/src/proto_alpha/bin_sc_rollup_node/commitment.ml
index 339c605ab479e61356aeeee2810c7c33f496e4f7..a13a9d232cbcede8f0f823358d4b62bbe093d251 100644
--- a/src/proto_alpha/bin_sc_rollup_node/commitment.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/commitment.ml
@@ -47,7 +47,7 @@ open Alpha_context
 module type Mutable_level_store =
   Store_sigs.Mutable_value
     with type value = Raw_level.t
-     and type store = Store.t
+     and type 'a store = 'a Store.store
 
 (* We persist the number of ticks to be included in the
    next commitment on disk, in a map that is indexed by
@@ -56,8 +56,8 @@ module type Mutable_level_store =
    node, as only finalized heads are processed to build commitments.
 *)
 module Number_of_ticks :
-  Store_sigs.Map
-    with type store = Store.t
+  Store_sigs.Append_only_map
+    with type 'a store = 'a Store.store
      and type key = Raw_level.t
      and type value = Z.t =
   Store.Make_append_only_map
@@ -205,7 +205,7 @@ module Make (PVM : Pvm.S) : Commitment_sig.S with module PVM = PVM = struct
       return_unit
     else return_unit
 
-  let update_ticks (node_ctxt : Node_context.t) current_level block_hash =
+  let update_ticks (node_ctxt : _ Node_context.t) current_level block_hash =
     let open Lwt_result_syntax in
     let*! last_stored_commitment_level_opt =
       last_commitment_level
@@ -245,14 +245,14 @@ module Make (PVM : Pvm.S) : Commitment_sig.S with module PVM = PVM = struct
       current_level
       (Z.add previous_level_num_ticks num_ticks)
 
-  let process_head (node_ctxt : Node_context.t) Layer1.{level; hash} =
+  let process_head (node_ctxt : _ Node_context.t) Layer1.{level; hash} =
     let open Lwt_result_syntax in
     let current_level = Raw_level.of_int32_exn level in
     let*! () = update_ticks node_ctxt current_level hash in
     store_commitment_if_necessary node_ctxt current_level hash
 
   let sync_last_cemented_commitment_hash_with_level
-      ({cctxt; rollup_address; store; _} : Node_context.t) =
+      ({cctxt; rollup_address; store; _} : _ Node_context.t) =
     let open Lwt_result_syntax in
     let* hash, inbox_level =
       Plugin.RPC.Sc_rollup.last_cemented_commitment_hash_with_level
@@ -268,7 +268,7 @@ module Make (PVM : Pvm.S) : Commitment_sig.S with module PVM = PVM = struct
     return_unit
 
   let get_commitment_and_publish ~check_lcc_hash
-      ({store; _} as node_ctxt : Node_context.t) next_level_to_publish =
+      ({store; _} as node_ctxt : _ Node_context.t) next_level_to_publish =
     let open Lwt_result_syntax in
     let*! is_commitment_available =
       Store.Commitments.mem store next_level_to_publish
@@ -380,7 +380,7 @@ module Make (PVM : Pvm.S) : Commitment_sig.S with module PVM = PVM = struct
         ()
     else return_false
 
-  let cement_commitment (node_ctxt : Node_context.t) commitment_hash =
+  let cement_commitment (node_ctxt : _ Node_context.t) commitment_hash =
     let open Lwt_result_syntax in
     let operator = Node_context.get_operator node_ctxt Cement in
     match operator with
diff --git a/src/proto_alpha/bin_sc_rollup_node/commitment.mli b/src/proto_alpha/bin_sc_rollup_node/commitment.mli
index b95d5c827db9f73c632e16d50efdad7b60270fa8..2f4f72e075901eaf77050ae0031fc15990827303 100644
--- a/src/proto_alpha/bin_sc_rollup_node/commitment.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/commitment.mli
@@ -44,7 +44,7 @@ open Protocol.Alpha_context
 module type Mutable_level_store =
   Store_sigs.Mutable_value
     with type value = Raw_level.t
-     and type store = Store.t
+     and type 'a store = 'a Store.store
 
 (** [last_commitment_with_hash (module Last_level_module: Mutable_level_store) store]
       returns the last commitment and relative hash
@@ -58,7 +58,7 @@ module type Mutable_level_store =
 
 val last_commitment_with_hash :
   (module Mutable_level_store) ->
-  Store.t ->
+  _ Store.t ->
   (Sc_rollup.Commitment.t * Sc_rollup.Commitment.Hash.t) option Lwt.t
 
 module Make (PVM : Pvm.S) : Commitment_sig.S with module PVM = PVM
diff --git a/src/proto_alpha/bin_sc_rollup_node/commitment_sig.ml b/src/proto_alpha/bin_sc_rollup_node/commitment_sig.ml
index c5de6d7bda1b47e1958af00181f6ae8bc82198ec..3dabbe89f68e157c36005c0aa958f7dcedd3e9f6 100644
--- a/src/proto_alpha/bin_sc_rollup_node/commitment_sig.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/commitment_sig.ml
@@ -47,13 +47,13 @@ module type S = sig
       whether it is a multiple of `Commitment.sc_rollup_commitment_period`
       levels away from [node_ctxt.initial_level]. It uses the functionalities of
       [PVM] to compute the hash of to be included in the commitment.  *)
-  val process_head : Node_context.t -> Layer1.head -> unit tzresult Lwt.t
+  val process_head : Node_context.rw -> Layer1.head -> unit tzresult Lwt.t
 
   (** [sync_last_cemented_commitment_hash_with_level node_ctxt] fetches and
       stores information about the last cemeneted commitment in the layer1
       chain.  *)
   val sync_last_cemented_commitment_hash_with_level :
-    Node_context.t -> unit tzresult Lwt.t
+    Node_context.rw -> unit tzresult Lwt.t
 
   (** [publish_commitment node_ctxt] publishes the earliest commitment
       stored in [store] that has not been published yet, unless its inbox level
@@ -72,7 +72,7 @@ module type S = sig
         in particular, no commitment is published.}
     }
   *)
-  val publish_commitment : Node_context.t -> unit tzresult Lwt.t
+  val publish_commitment : Node_context.rw -> unit tzresult Lwt.t
 
   (** [cement_commitment_if_possible node_ctxt head] checks whether the next
       commitment to be cemented (i.e. whose inbox level is
@@ -83,7 +83,7 @@ module type S = sig
       [sc_rollup_challenge_period] levels have passed since when the commitment
       was originally published.  *)
   val cement_commitment_if_possible :
-    Node_context.t -> Layer1.head -> unit tzresult Lwt.t
+    _ Node_context.t -> Layer1.head -> unit tzresult Lwt.t
 
   (** [start ()] only emits the event that the commitment manager
       for the rollup node has started. *)
diff --git a/src/proto_alpha/bin_sc_rollup_node/context.ml b/src/proto_alpha/bin_sc_rollup_node/context.ml
index a3f257ddcc6c2ef05fd700cc499ffc3e5be707da..ea9ec24f8d691b6a850e5c5549c8b033c57b5aef 100644
--- a/src/proto_alpha/bin_sc_rollup_node/context.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/context.ml
@@ -25,6 +25,7 @@
 
 open Protocol
 open Alpha_context
+open Store_sigs
 module Maker = Irmin_pack_unix.Maker (Tezos_context_encoding.Context.Conf)
 
 module IStore = struct
@@ -39,9 +40,19 @@ module IStoreTree =
 
 type tree = IStore.tree
 
-type index = {path : string; repo : IStore.Repo.t}
+type 'a raw_index = {path : string; repo : IStore.Repo.t}
 
-type t = {index : index; tree : tree}
+type 'a index = 'a raw_index constraint 'a = [< `Read | `Write > `Read]
+
+type rw_index = [`Read | `Write] index
+
+type ro_index = [`Read] index
+
+type 'a t = {index : 'a index; tree : tree}
+
+type rw = [`Read | `Write] t
+
+type ro = [`Read] t
 
 type commit = IStore.commit
 
@@ -62,15 +73,19 @@ let pp_hash fmt h =
   IStore.Hash.to_raw_string h
   |> Hex.of_string |> Hex.show |> Format.pp_print_string fmt
 
-let load configuration =
+let load : type a. a mode -> Configuration.t -> a raw_index Lwt.t =
+ fun mode configuration ->
   let open Lwt_syntax in
   let open Configuration in
   let path = default_context_dir configuration.data_dir in
-  let+ repo = IStore.Repo.v (Irmin_pack.config path) in
+  let readonly = match mode with Read_only -> true | Read_write -> false in
+  let+ repo = IStore.Repo.v (Irmin_pack.config ~readonly path) in
   {path; repo}
 
 let close ctxt = IStore.Repo.close ctxt.repo
 
+let readonly (index : [> `Read] index) = (index :> [`Read] index)
+
 let raw_commit ?(message = "") index tree =
   let info = IStore.Info.v ~author:"Tezos" 0L ~message in
   IStore.Commit.v index.repo ~info ~parents:[] tree
@@ -112,7 +127,7 @@ struct
   module Tree = struct
     include IStoreTree
 
-    type nonrec t = index
+    type t = rw_index
 
     type tree = IStore.tree
 
@@ -189,7 +204,7 @@ module Inbox = struct
             .tree_proof_encoding
         end)
 
-    type t = index
+    type t = rw_index
 
     let commit_tree index _key tree =
       let open Lwt_syntax in
@@ -218,6 +233,8 @@ module PVMState = struct
 
   let key = ["pvm_state"]
 
+  let empty () = IStore.Tree.empty ()
+
   let find ctxt = IStore.Tree.find_tree ctxt.tree key
 
   let lookup tree path = IStore.Tree.find tree path
diff --git a/src/proto_alpha/bin_sc_rollup_node/context.mli b/src/proto_alpha/bin_sc_rollup_node/context.mli
index 516d3443f53295aeb1055b35a679d8ea707f748a..a437e86b166ff936d343c15c765cd8f9e948b35e 100644
--- a/src/proto_alpha/bin_sc_rollup_node/context.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/context.mli
@@ -25,15 +25,29 @@
 
 open Protocol
 open Alpha_context
+open Store_sigs
 
-(** The type of indexed repository for contexts  *)
-type index
+(** The type of indexed repository for contexts. The parameter indicates if the
+    index can be written or only read. *)
+type 'a index constraint 'a = [< `Read | `Write > `Read]
+
+(** Read/write {!index}. *)
+type rw_index = [`Read | `Write] index
+
+(** Read only {!index}. *)
+type ro_index = [`Read] index
 
 (** The type of trees stored in the context, i.e. the actual data. *)
 type tree
 
-(** The type of context with its content *)
-type t
+(** The type of context with its content. *)
+type 'a t constraint 'a = [< `Read | `Write > `Read]
+
+(** Read/write context {!t}. *)
+type rw = [`Read | `Write] t
+
+(** Read-only context {!t}. *)
+type ro = [`Read] t
 
 (** A context hash is the hash produced when the data of the context is
     committed to disk, i.e. the {!commit} hash. *)
@@ -54,34 +68,37 @@ val pp_hash : Format.formatter -> hash -> unit
 
 (** [load config] initializes from disk a context using the [data_dir]
     information contained in the configuration [config]. *)
-val load : Configuration.t -> index Lwt.t
+val load : 'a mode -> Configuration.t -> 'a index Lwt.t
 
 (** [index context] is the repository of the context [context]. *)
-val index : t -> index
+val index : 'a t -> 'a index
 
 (** [close ctxt] closes the context index [ctxt]. *)
-val close : index -> unit Lwt.t
+val close : _ index -> unit Lwt.t
+
+(** [readonly index] returns a read-only version of the index. *)
+val readonly : [> `Read] index -> [`Read] index
 
 (** [raw_commit ?message ctxt tree] commits the [tree] in the context repository
     [ctxt] on disk, and return the commit. *)
-val raw_commit : ?message:string -> index -> tree -> commit Lwt.t
+val raw_commit : ?message:string -> [> `Write] index -> tree -> commit Lwt.t
 
 (** [commit ?message context] commits content of the context [context] on disk,
     and return the commit hash. *)
-val commit : ?message:string -> t -> hash Lwt.t
+val commit : ?message:string -> [> `Write] t -> hash Lwt.t
 
 (** [checkout ctxt hash] checkouts the content that corresponds to the commit
     hash [hash] in the repository [ctxt] and returns the corresponding
     context. If there is no commit that corresponds to [hash], it returns
     [None].  *)
-val checkout : index -> hash -> t option Lwt.t
+val checkout : 'a index -> hash -> 'a t option Lwt.t
 
 (** [empty ctxt] is the context with an empty content for the repository [ctxt]. *)
-val empty : index -> t
+val empty : 'a index -> 'a t
 
 (** [is_empty context] returns [true] iff the context content of [context] is
     empty. *)
-val is_empty : t -> bool
+val is_empty : _ t -> bool
 
 (** Module for generating and verifying proofs for a context *)
 module Proof (Hash : sig
@@ -92,11 +109,17 @@ end) (Proof_encoding : sig
   val proof_encoding :
     Environment.Context.Proof.tree Environment.Context.Proof.t Data_encoding.t
 end) : sig
+  (** Tree representation for proof generation.
+
+      NOTE: The index needs to be accessed with write permissions because we
+      need to commit on disk to generate the proofs (e.g. in
+      {!Inbox.produce_proof}, {!PVM.produce_proof}. or
+      {!PVM.produce_output_proof}). *)
   module Tree :
     Tezos_context_sigs.Context.TREE
       with type key = string list
        and type value = bytes
-       and type t = index
+       and type t = rw_index
        and type tree = tree
 
   type tree = Tree.tree
@@ -120,7 +143,7 @@ end) : sig
   (** [produce_proof ctxt tree f] produces and returns a proof for the execution
       of [f] on the state [tree]. *)
   val produce_proof :
-    index -> tree -> (tree -> (tree * 'a) Lwt.t) -> (proof * 'a) option Lwt.t
+    rw_index -> tree -> (tree -> (tree * 'a) Lwt.t) -> (proof * 'a) option Lwt.t
 
   (** [verify_proof proof f] verifies that [f] produces the proof [proof] and
       returns the resulting [tree], or [None] if the proof cannot be
@@ -135,12 +158,12 @@ module MessageTrees : sig
   type value
 
   (** [find context] returns the messages tree stored in the [context], if any. *)
-  val find : t -> value option Lwt.t
+  val find : _ t -> value option Lwt.t
 
   (** [set context msg_tree] saves the messages tree [msg_tree] in the context
       and returns the updated context. Note: [set] does not perform any write on
       disk, this information must be committed using {!commit}. *)
-  val set : t -> value -> t Lwt.t
+  val set : 'a t -> value -> 'a t Lwt.t
 end
 
 (** L1 inboxes representation in the rollup node. This is a version of the
@@ -162,7 +185,7 @@ module Inbox : sig
   include
     Sc_rollup.Inbox.Merkelized_operations
       with type tree = MessageTrees.value
-       and type inbox_context = index
+       and type inbox_context = rw_index
 end
 
 (** State of the PVM that this rollup node deals with *)
@@ -170,8 +193,11 @@ module PVMState : sig
   (** The value of a PVM state *)
   type value = tree
 
+  (** [empty ()] is the empty PVM state. *)
+  val empty : unit -> value
+
   (** [find context] returns the PVM state stored in the [context], if any. *)
-  val find : t -> value option Lwt.t
+  val find : _ t -> value option Lwt.t
 
   (** [lookup state path] returns the data stored for the path [path] in the PVM
       state [state].  *)
@@ -180,5 +206,5 @@ module PVMState : sig
   (** [set context state] saves the PVM state [state] in the context and returns
       the updated context. Note: [set] does not perform any write on disk, this
       information must be committed using {!commit}. *)
-  val set : t -> value -> t Lwt.t
+  val set : 'a t -> value -> 'a t Lwt.t
 end
diff --git a/src/proto_alpha/bin_sc_rollup_node/daemon.ml b/src/proto_alpha/bin_sc_rollup_node/daemon.ml
index 595f5a7a7b4a44d6ebdc66b8c3cfc567faac07e5..f45ec5a80693e72351290fe76273f01b9baba7de 100644
--- a/src/proto_alpha/bin_sc_rollup_node/daemon.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/daemon.ml
@@ -33,7 +33,7 @@ module Make (PVM : Pvm.S) = struct
   (** Process an L1 SCORU operation (for the node's rollup) which is included
       for the first time. {b Note}: this function does not process inboxes for
       the rollup, which is done instead by {!Inbox.process_head}. *)
-  let process_included_l1_operation (type kind) (node_ctxt : Node_context.t)
+  let process_included_l1_operation (type kind) (node_ctxt : Node_context.rw)
       head ~source:_ (operation : kind manager_operation)
       (result : kind successful_manager_operation_result) =
     let open Lwt_result_syntax in
@@ -159,11 +159,11 @@ module Make (PVM : Pvm.S) = struct
     in
     return_unit
 
-  let before_origination (node_ctxt : Node_context.t) Layer1.{level; _} =
+  let before_origination (node_ctxt : _ Node_context.t) Layer1.{level; _} =
     let origination_level = Raw_level.to_int32 node_ctxt.genesis_info.level in
     level < origination_level
 
-  let rec processed_finalized_block (node_ctxt : Node_context.t)
+  let rec processed_finalized_block (node_ctxt : _ Node_context.t)
       Layer1.({hash; level} as block) =
     let open Lwt_result_syntax in
     let*! last_finalized = State.get_finalized_head_opt node_ctxt.store in
@@ -187,7 +187,8 @@ module Make (PVM : Pvm.S) = struct
     let*! () = State.mark_finalized_head node_ctxt.store block in
     return_unit
 
-  let process_head (node_ctxt : Node_context.t) Layer1.({hash; level} as head) =
+  let process_head (node_ctxt : _ Node_context.t) Layer1.({hash; level} as head)
+      =
     let open Lwt_result_syntax in
     let*! () = Daemon_event.head_processing hash level ~finalized:false in
     let* ctxt = Inbox.process_head node_ctxt head in
@@ -287,7 +288,7 @@ module Make (PVM : Pvm.S) = struct
 
   (* TODO: https://gitlab.com/tezos/tezos/-/issues/2895
      Use Lwt_stream.fold_es once it is exposed. *)
-  let daemonize configuration (node_ctxt : Node_context.t) =
+  let daemonize configuration (node_ctxt : _ Node_context.t) =
     let open Lwt_result_syntax in
     let rec loop (l1_ctxt : Layer1.t) =
       let*! () =
@@ -331,8 +332,7 @@ module Make (PVM : Pvm.S) = struct
     let* () = Event.shutdown_node exit_status in
     Tezos_base_unix.Internal_event_unix.close ()
 
-  let check_initial_state_hash {Node_context.cctxt; rollup_address; context; _}
-      =
+  let check_initial_state_hash {Node_context.cctxt; rollup_address; _} =
     let open Lwt_result_syntax in
     let* l1_reference_initial_state_hash =
       RPC.Sc_rollup.initial_pvm_state_hash
@@ -340,7 +340,7 @@ module Make (PVM : Pvm.S) = struct
         (cctxt#chain, cctxt#block)
         rollup_address
     in
-    let*! s = PVM.initial_state context in
+    let*! s = PVM.initial_state ~empty:(PVM.State.empty ()) in
     let*! l2_initial_state_hash = PVM.state_hash s in
     if
       not
@@ -388,7 +388,7 @@ module Make (PVM : Pvm.S) = struct
       let* () =
         Injector.init
           node_ctxt.cctxt
-          node_ctxt
+          (Node_context.readonly node_ctxt)
           ~data_dir:configuration.data_dir
           ~signers
       in
@@ -422,9 +422,11 @@ let run ~data_dir (configuration : Configuration.t)
       configuration.sc_rollup_node_operators
   in
   let*! store =
-    Store.load Configuration.(default_storage_dir configuration.data_dir)
+    Store.load
+      Read_write
+      Configuration.(default_storage_dir configuration.data_dir)
   in
-  let*! context = Context.load configuration in
+  let*! context = Context.load Read_write configuration in
   let* l1_ctxt, kind = Layer1.start configuration cctxt store in
   let* node_ctxt =
     Node_context.init
diff --git a/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.ml b/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.ml
index 804e9e4dcf021d1b4e5d3b9fc92297214a3dfc28..5c9e81e24c9a5bc63622fe19ed8a7a971697d2da 100644
--- a/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.ml
@@ -65,6 +65,29 @@ let store_entry_from_published_level ~dal_attestation_lag ~published_level store
   @@ Int32.(
        add (of_int dal_attestation_lag) (Raw_level.to_int32 published_level))
 
+(* The cache allows to not fetch pages on the DAL node more than necessary. *)
+module Pages_cache =
+  Ringo_lwt.Functors.Make
+    ((val Ringo.(
+            map_maker ~replacement:LRU ~overflow:Strong ~accounting:Precise))
+       (struct
+         include Cryptobox.Commitment
+
+         let hash commitment =
+           Data_encoding.Binary.to_string_exn
+             Cryptobox.Commitment.encoding
+             commitment
+           |> Hashtbl.hash
+       end))
+
+let get_slot_pages =
+  let pages_cache = Pages_cache.create 16 (* 130MB *) in
+  fun dal_cctxt commitment ->
+    Pages_cache.find_or_replace
+      pages_cache
+      commitment
+      (Dal_node_client.get_slot_pages dal_cctxt)
+
 let check_confirmation_status_and_download
     ({Node_context.store; dal_cctxt; _} as node_ctxt) ~confirmed_in_block_hash
     ~published_in_block_hash index =
@@ -85,23 +108,23 @@ let check_confirmation_status_and_download
         ~primary_key:published_in_block_hash
         ~secondary_key:index
     in
-    let* pages = Dal_node_client.get_slot_pages dal_cctxt commitment in
-    let*! () =
+    let* pages = get_slot_pages dal_cctxt commitment in
+    let save_pages node_ctxt =
       Dal_slots_tracker.save_confirmed_slot
         node_ctxt
         confirmed_in_block_hash
         index
         pages
     in
-    return_some pages
+    return (Delayed_write_monad.delay_write (Some pages) save_pages)
   else
-    let*! () =
+    let save_slot node_ctxt =
       Dal_slots_tracker.save_unconfirmed_slot
         node_ctxt
         confirmed_in_block_hash
         index
     in
-    return_none
+    return (Delayed_write_monad.delay_write None save_slot)
 
 let slot_pages ~dal_attestation_lag ({Node_context.store; _} as node_ctxt)
     Dal.Slot.Header.{published_level; index} =
@@ -125,21 +148,25 @@ let slot_pages ~dal_attestation_lag ({Node_context.store; _} as node_ctxt)
         ~published_in_block_hash
         ~confirmed_in_block_hash
         index
-  | Some `Unconfirmed -> return None
+  | Some `Unconfirmed -> return (Delayed_write_monad.no_write None)
   | Some `Confirmed ->
       let*! pages =
         Store.Dal_slot_pages.list_secondary_keys_with_values
           store
           ~primary_key:confirmed_in_block_hash
       in
-      List.filter
-        (fun ((slot_idx, _page_idx), _v) -> Dal.Slot_index.equal index slot_idx)
-        pages
-      |> List.map snd |> Option.some |> return
+      let pages =
+        List.filter_map
+          (fun ((slot_idx, _page_idx), v) ->
+            if Dal.Slot_index.equal index slot_idx then Some v else None)
+          pages
+      in
+      return (Delayed_write_monad.no_write (Some pages))
 
 let page_content ~dal_attestation_lag ({Node_context.store; _} as node_ctxt)
     page_id =
   let open Lwt_result_syntax in
+  let open Delayed_write_monad.Lwt_result_syntax in
   let Dal.Page.{slot_id; page_index} = page_id in
   let Dal.Slot.Header.{published_level; index} = slot_id in
   let*! confirmed_in_block_hash =
@@ -164,7 +191,7 @@ let page_content ~dal_attestation_lag ({Node_context.store; _} as node_ctxt)
       let*! published_in_block_hash =
         State.hash_of_level store (Raw_level.to_int32 published_level)
       in
-      let* pages =
+      let>* pages =
         check_confirmation_status_and_download
           node_ctxt
           ~published_in_block_hash
@@ -172,7 +199,7 @@ let page_content ~dal_attestation_lag ({Node_context.store; _} as node_ctxt)
           index
       in
       match pages with
-      | None -> (* Slot is not confirmed *) return_none
+      | None -> (* Slot is not confirmed *) return None
       | Some (* Slot is confirmed *) pages -> (
           match List.nth_opt pages page_index with
           | Some page -> return @@ Some page
diff --git a/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.mli b/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.mli
index 3ca07280600bdb44142c8a918a777d4a324a18e4..2a9a69a71b9512d5bd065aa109e8b9a7e9e6d287 100644
--- a/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/dal_pages_request.mli
@@ -55,9 +55,9 @@ type error += Dal_slot_not_found_in_store of Dal.Slot.Header.id
 *)
 val slot_pages :
   dal_attestation_lag:int ->
-  Node_context.t ->
+  _ Node_context.t ->
   Dal.slot_id ->
-  Dal.Page.content list option tzresult Lwt.t
+  Dal.Page.content list option Node_context.delayed_write tzresult Lwt.t
 
 (** Retrieve the content of the page identified by the given ID from the store.
 
@@ -71,6 +71,6 @@ val slot_pages :
 *)
 val page_content :
   dal_attestation_lag:int ->
-  Node_context.t ->
+  _ Node_context.t ->
   Dal.Page.t ->
-  Dal.Page.content option tzresult Lwt.t
+  Dal.Page.content option Node_context.delayed_write tzresult Lwt.t
diff --git a/src/proto_alpha/bin_sc_rollup_node/dal_slots_tracker.mli b/src/proto_alpha/bin_sc_rollup_node/dal_slots_tracker.mli
index 549a32c23abdbda9512195a79388fa5705c55c74..c5da7b2ebcb35598b1c2a9529ed457de57e4bfa2 100644
--- a/src/proto_alpha/bin_sc_rollup_node/dal_slots_tracker.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/dal_slots_tracker.mli
@@ -38,19 +38,19 @@ type error += Cannot_read_block_metadata of Tezos_crypto.Block_hash.t
 (** [is_slot_confirmed node_ctxt head slot_index] checks whether the slot
     with index [slot_index] has been confirmed in [head]. *)
 val is_slot_confirmed :
-  Node_context.t -> Layer1.head -> Dal.Slot_index.t -> bool tzresult Lwt.t
+  _ Node_context.t -> Layer1.head -> Dal.Slot_index.t -> bool tzresult Lwt.t
 
 (** [save_unconfirmed_slot node_ctxt hash slot_index] saves in [node_ctxt.store]
     that [slot_index] is unconfirmed in the block with hash in [node_ctxt.store].
 *)
 val save_unconfirmed_slot :
-  Node_context.t -> Tezos_crypto.Block_hash.t -> Dal.Slot_index.t -> unit Lwt.t
+  Node_context.rw -> Tezos_crypto.Block_hash.t -> Dal.Slot_index.t -> unit Lwt.t
 
 (** [save_confirmed_slot node_ctxt hash slot_index] saves in [node_ctxt.store]
     that [slot_index] is confirmed in the block with hashin [node_ctxt.store].
     The contents of the slot are set to [pages] in [node_ctxt.store]. *)
 val save_confirmed_slot :
-  Node_context.t ->
+  Node_context.rw ->
   Tezos_crypto.Block_hash.t ->
   Dal.Slot_index.t ->
   Dal.Page.content list ->
@@ -63,18 +63,18 @@ val save_confirmed_slot :
       ones the rollup node will download, and stores the results in
       [Store.Dal_confirmed_slots].}
     }  *)
-val process_head : Node_context.t -> Layer1.head -> unit tzresult Lwt.t
+val process_head : Node_context.rw -> Layer1.head -> unit tzresult Lwt.t
 
 (** [slots_history_of_hash node_ctxt block_hash] returns the DAL confirmed slots
    history at the end of the given [block_hash] validation. *)
 val slots_history_of_hash :
-  Node_context.t ->
+  _ Node_context.t ->
   Layer1.head ->
   Protocol.Alpha_context.Dal.Slots_history.t tzresult Lwt.t
 
 (** [slots_history_cache_of_hash node_ctxt block_hash] returns the DAL confirmed
    slots history cache at the end of the given [block_hash] validation. *)
 val slots_history_cache_of_hash :
-  Node_context.t ->
+  _ Node_context.t ->
   Layer1.head ->
   Protocol.Alpha_context.Dal.Slots_history.History_cache.t tzresult Lwt.t
diff --git a/src/proto_alpha/bin_sc_rollup_node/fueled_pvm.ml b/src/proto_alpha/bin_sc_rollup_node/fueled_pvm.ml
index adde7dea52873060d6881d69e1353e98b3a17698..a39c5144b496532978c3675d61d462b657555d10 100644
--- a/src/proto_alpha/bin_sc_rollup_node/fueled_pvm.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/fueled_pvm.ml
@@ -38,10 +38,10 @@ module type S = sig
     metadata:Sc_rollup.Metadata.t ->
     dal_attestation_lag:int ->
     fuel:fuel ->
-    Node_context.t ->
+    _ Node_context.t ->
     Tezos_crypto.Block_hash.t ->
     state ->
-    (state * Z.t * Raw_level.t * fuel, tztrace) result Lwt.t
+    (state * Z.t * Raw_level.t * fuel) Node_context.delayed_write tzresult Lwt.t
 end
 
 module Make
@@ -53,7 +53,7 @@ module Make
   type fuel = F.t
 
   let continue_with_fuel consumption initial_fuel state f =
-    let open Lwt_result_syntax in
+    let open Delayed_write_monad.Lwt_result_syntax in
     match F.consume consumption initial_fuel with
     | None -> return (state, 0L)
     | Some fuel_left -> f fuel_left state
@@ -70,6 +70,7 @@ module Make
   let eval_until_input ~metadata ~dal_attestation_lag data_dir store level
       message_index ~fuel start_tick failing_ticks state =
     let open Lwt_result_syntax in
+    let open Delayed_write_monad.Lwt_result_syntax in
     let module Builtins = struct
       let reveal_preimage hash =
         let hash =
@@ -127,7 +128,7 @@ module Make
       else
         match input_request with
         | No_input_required ->
-            let* next_state, executed_ticks, failing_ticks =
+            let>* next_state, executed_ticks, failing_ticks =
               eval_tick fuel current_tick failing_ticks state
             in
             go
@@ -151,7 +152,7 @@ module Make
             | Some fuel ->
                 go fuel (Int64.succ current_tick) failing_ticks next_state)
         | Needs_reveal (Request_dal_page page_id) -> (
-            let* content_opt =
+            let>* content_opt =
               Dal_pages_request.page_content ~dal_attestation_lag store page_id
             in
             let*! next_state =
@@ -180,7 +181,8 @@ module Make
   let feed_input ~metadata ~dal_attestation_lag data_dir store level
       message_index ~fuel ~failing_ticks state input =
     let open Lwt_result_syntax in
-    let* state, fuel, tick, failing_ticks =
+    let open Delayed_write_monad.Lwt_result_syntax in
+    let>* state, fuel, tick, failing_ticks =
       eval_until_input
         ~metadata
         ~dal_attestation_lag
@@ -195,7 +197,7 @@ module Make
     in
     let consumption = F.of_ticks tick in
     continue_with_fuel consumption fuel state @@ fun fuel state ->
-    let* input, failing_ticks =
+    let>* input, failing_ticks =
       match failing_ticks with
       | xtick :: failing_ticks' ->
           if xtick = tick then
@@ -211,7 +213,7 @@ module Make
       | _ -> return (input, failing_ticks)
     in
     let*! state = PVM.set_input (Inbox_message input) state in
-    let* state, _fuel, tick, _failing_ticks =
+    let>* state, _fuel, tick, _failing_ticks =
       eval_until_input
         ~metadata
         ~dal_attestation_lag
@@ -228,9 +230,11 @@ module Make
 
   let eval_block_inbox ~metadata ~dal_attestation_lag ~fuel
       (Node_context.{data_dir; store; loser_mode; _} as node_context) hash
-      (state : state) : (state * Z.t * Raw_level.t * fuel, tztrace) result Lwt.t
-      =
+      (state : state) :
+      (state * Z.t * Raw_level.t * fuel) Node_context.delayed_write tzresult
+      Lwt.t =
     let open Lwt_result_syntax in
+    let open Delayed_write_monad.Lwt_result_syntax in
     (* Obtain inbox and its messages for this block. *)
     let*! inbox = Store.Inboxes.find store hash in
     match inbox with
@@ -264,7 +268,7 @@ module Make
               ~level
               ~message_index:message_counter
           in
-          let* state, executed_ticks =
+          let>* state, executed_ticks =
             feed_input
               ~metadata
               ~dal_attestation_lag
@@ -280,8 +284,8 @@ module Make
           return (state, F.of_ticks executed_ticks)
         in
         (* Iterate the PVM state with all the messages for this level. *)
-        let* state, fuel =
-          List.fold_left_i_es feed_message (state, fuel) messages
+        let>* state, fuel =
+          list_fold_left_i_es feed_message (state, fuel) messages
         in
         return (state, num_messages, inbox_level, fuel)
 end
diff --git a/src/proto_alpha/bin_sc_rollup_node/inbox.ml b/src/proto_alpha/bin_sc_rollup_node/inbox.ml
index 5459cd00f5e6ed3e733350a0745fc4837475898b..45beefd267d9b2ae99567bb2850eb27d69f4659b 100644
--- a/src/proto_alpha/bin_sc_rollup_node/inbox.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/inbox.ml
@@ -159,7 +159,7 @@ let same_inbox_as_layer_1 node_ctxt head_hash inbox =
     (Sc_rollup.Inbox.equal layer1_inbox inbox)
     (Sc_rollup_node_errors.Inconsistent_inbox {layer1_inbox; inbox})
 
-let process_head (node_ctxt : Node_context.t)
+let process_head (node_ctxt : _ Node_context.t)
     Layer1.({level; hash = head_hash} as head) =
   let open Lwt_result_syntax in
   let first_inbox_level =
diff --git a/src/proto_alpha/bin_sc_rollup_node/inbox.mli b/src/proto_alpha/bin_sc_rollup_node/inbox.mli
index 7d93c29ffe10fa170806a4b7f8bc453ad6600d4c..24457aadba25e7f51e15ff1fa7c1598fcf968e9b 100644
--- a/src/proto_alpha/bin_sc_rollup_node/inbox.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/inbox.mli
@@ -38,19 +38,19 @@ open Protocol.Alpha_context
 (** [process_head node_ctxt head operations] changes the state of the inbox to
     react to [head]. In particular, this process filters the provided
     [operations] of the [head] block. *)
-val process_head : Node_context.t -> Layer1.head -> Context.t tzresult Lwt.t
+val process_head : Node_context.rw -> Layer1.head -> Context.rw tzresult Lwt.t
 
 (** [inbox_of_hash node_ctxt block_hash] returns the rollup inbox at the end of
     the given validation of [block_hash]. *)
 val inbox_of_hash :
-  Node_context.t ->
+  _ Node_context.t ->
   Tezos_crypto.Block_hash.t ->
   Sc_rollup.Inbox.t tzresult Lwt.t
 
 (** [history_of_hash node_ctxt block_hash] returns the rollup inbox history at
     the end of the given validation of [block_hash]. *)
 val history_of_hash :
-  Node_context.t ->
+  _ Node_context.t ->
   Tezos_crypto.Block_hash.t ->
   Sc_rollup.Inbox.History.t tzresult Lwt.t
 
diff --git a/src/proto_alpha/bin_sc_rollup_node/injector.ml b/src/proto_alpha/bin_sc_rollup_node/injector.ml
index 556ffbbc8fc78a4c6febee96ac96e50345c3cb31..87b3ef3cbdee3f9c36f877c64014bb5600be5806 100644
--- a/src/proto_alpha/bin_sc_rollup_node/injector.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/injector.ml
@@ -28,9 +28,9 @@ open Injector_sigs
 
 module Parameters :
   PARAMETERS
-    with type rollup_node_state = Node_context.t
+    with type rollup_node_state = Node_context.ro
      and type Tag.t = Configuration.purpose = struct
-  type rollup_node_state = Node_context.t
+  type rollup_node_state = Node_context.ro
 
   let events_section = ["sc_rollup.injector"]
 
diff --git a/src/proto_alpha/bin_sc_rollup_node/injector.mli b/src/proto_alpha/bin_sc_rollup_node/injector.mli
index c08bb06e06d7f07debf10827f5925e98764a73e8..d4b38f8ec62f5838eb9ee4fbb94ed7b1cfe92b06 100644
--- a/src/proto_alpha/bin_sc_rollup_node/injector.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/injector.mli
@@ -25,5 +25,5 @@
 
 include
   Injector_sigs.S
-    with type rollup_node_state := Node_context.t
+    with type rollup_node_state := Node_context.ro
      and type tag := Configuration.purpose
diff --git a/src/proto_alpha/bin_sc_rollup_node/interpreter.ml b/src/proto_alpha/bin_sc_rollup_node/interpreter.ml
index 15e1710caa82d3b5e26e0864d8cd44d8f0cac4fa..2421f68454b0a3805306d69b67d6995fb8dbec67 100644
--- a/src/proto_alpha/bin_sc_rollup_node/interpreter.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/interpreter.ml
@@ -29,19 +29,19 @@ open Alpha_context
 module type S = sig
   module PVM : Pvm.S
 
-  val metadata : Node_context.t -> Sc_rollup.Metadata.t
+  val metadata : _ Node_context.t -> Sc_rollup.Metadata.t
 
   (** [process_head node_ctxt head] interprets the messages associated
       with a [head] from a chain [event]. This requires the inbox to be updated
       beforehand. *)
   val process_head :
-    Node_context.t -> Context.t -> Layer1.head -> unit tzresult Lwt.t
+    Node_context.rw -> Context.rw -> Layer1.head -> unit tzresult Lwt.t
 
   (** [state_of_tick node_ctxt tick level] returns [Some (state, hash)]
       for a given [tick] if this [tick] happened before
       [level]. Otherwise, returns [None].*)
   val state_of_tick :
-    Node_context.t ->
+    _ Node_context.t ->
     Sc_rollup.Tick.t ->
     Raw_level.t ->
     (PVM.state * PVM.hash) option tzresult Lwt.t
@@ -59,7 +59,7 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
 
   (** [metadata node_ctxt] creates a {Sc_rollup.Metadata.t} using the information
       stored in [node_ctxt]. *)
-  let metadata (node_ctxt : Node_context.t) =
+  let metadata (node_ctxt : _ Node_context.t) =
     let address = node_ctxt.rollup_address in
     let origination_level = node_ctxt.genesis_info.Sc_rollup.Commitment.level in
     Sc_rollup.Metadata.{address; origination_level}
@@ -69,7 +69,7 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
       we're following.
       It must be called with [block_hash.level] = [node_ctxt.genesis_info.level].
   *)
-  let get_boot_sector block_hash (node_ctxt : Node_context.t) =
+  let get_boot_sector block_hash (node_ctxt : _ Node_context.t) =
     let open Lwt_result_syntax in
     let exception Found_boot_sector of string in
     let* block = Layer1.fetch_tezos_block node_ctxt.l1_ctxt block_hash in
@@ -111,10 +111,9 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
         | _ -> missing_boot_sector ())
 
   let genesis_state block_hash node_ctxt ctxt =
-    let open Node_context in
     let open Lwt_result_syntax in
     let* boot_sector = get_boot_sector block_hash node_ctxt in
-    let*! initial_state = PVM.initial_state node_ctxt.context in
+    let*! initial_state = PVM.initial_state ~empty:(PVM.State.empty ()) in
     let*! genesis_state = PVM.install_boot_sector initial_state boot_sector in
     let*! ctxt = PVM.State.set ctxt genesis_state in
     return (ctxt, genesis_state)
@@ -142,7 +141,7 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
     let dal_attestation_lag =
       node_ctxt.protocol_constants.parametric.dal.attestation_lag
     in
-    let* state, num_messages, inbox_level, _fuel =
+    let* eval_result =
       Free_pvm.eval_block_inbox
         ~metadata
         ~dal_attestation_lag
@@ -151,7 +150,9 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
         hash
         predecessor_state
     in
-
+    let*! state, num_messages, inbox_level, _fuel =
+      Delayed_write_monad.apply node_ctxt eval_result
+    in
     (* Write final state to store. *)
     let*! ctxt = PVM.State.set ctxt state in
     let*! context_hash = Context.commit ctxt in
@@ -195,7 +196,7 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
     return_unit
 
   (** [process_head node_ctxt head] runs the PVM for the given head. *)
-  let process_head (node_ctxt : Node_context.t) ctxt head =
+  let process_head (node_ctxt : _ Node_context.t) ctxt head =
     let open Lwt_result_syntax in
     let first_inbox_level =
       Raw_level.to_int32 node_ctxt.genesis_info.level |> Int32.succ
@@ -230,6 +231,7 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
       evaluation of the inbox at block [hash] for at most [tick_distance]. *)
   let run_for_ticks node_ctxt predecessor_hash hash level tick_distance =
     let open Lwt_result_syntax in
+    let open Delayed_write_monad.Lwt_result_syntax in
     let pred_level = Raw_level.to_int32 level |> Int32.pred in
     let* ctxt = Node_context.checkout_context node_ctxt predecessor_hash in
     let* _ctxt, state =
@@ -242,7 +244,7 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
     let dal_attestation_lag =
       node_ctxt.protocol_constants.parametric.dal.attestation_lag
     in
-    let* state, _counter, _level, _fuel =
+    let>* state, _counter, _level, _fuel =
       Accounted_pvm.eval_block_inbox
         ~metadata
         ~dal_attestation_lag
@@ -286,6 +288,7 @@ module Make (PVM : Pvm.S) : S with module PVM = PVM = struct
               event.level
               tick_distance
           in
+          let state = Delayed_write_monad.ignore state in
           let*! hash = PVM.state_hash state in
           return (Some (state, hash))
 end
diff --git a/src/proto_alpha/bin_sc_rollup_node/layer1.mli b/src/proto_alpha/bin_sc_rollup_node/layer1.mli
index 625050011d2a081ae8a84c23f8b1f515e3dfb2c5..441c413843f24b4d7788f386a0439be907504d2d 100644
--- a/src/proto_alpha/bin_sc_rollup_node/layer1.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/layer1.mli
@@ -54,13 +54,13 @@ type t = private {
 val start :
   Configuration.t ->
   Protocol_client_context.full ->
-  Store.t ->
+  _ Store.t ->
   (t * Protocol.Alpha_context.Sc_rollup.Kind.t) tzresult Lwt.t
 
 (** [reconnect cfg l1_ctxt store] reconnects (and retries with delay) to the
     Tezos node. The delay for each reconnection is increased with a randomized
     exponential backoff (capped to 1.5h) . *)
-val reconnect : Configuration.t -> t -> Store.t -> t tzresult Lwt.t
+val reconnect : Configuration.t -> t -> _ Store.t -> t tzresult Lwt.t
 
 (** [get_predecessor_opt state head] returns the predecessor of block [head],
     when [head] is not the genesis block. *)
diff --git a/src/proto_alpha/bin_sc_rollup_node/node_context.ml b/src/proto_alpha/bin_sc_rollup_node/node_context.ml
index 38b389105deecc7b9ab1d2946dc4995a1a2f5fb6..d666f9a2a93faa34445927ff578b191567614656 100644
--- a/src/proto_alpha/bin_sc_rollup_node/node_context.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/node_context.ml
@@ -26,7 +26,7 @@
 open Protocol
 open Alpha_context
 
-type t = {
+type 'a t = {
   cctxt : Protocol_client_context.full;
   dal_cctxt : Dal_node_client.cctxt;
   data_dir : string;
@@ -39,10 +39,14 @@ type t = {
   fee_parameters : Configuration.fee_parameters;
   protocol_constants : Constants.t;
   loser_mode : Loser_mode.t;
-  store : Store.t;
-  context : Context.index;
+  store : 'a Store.t;
+  context : 'a Context.index;
 }
 
+type rw = [`Read | `Write] t
+
+type ro = [`Read] t
+
 let get_operator node_ctxt purpose =
   Configuration.Operator_purpose_map.find purpose node_ctxt.operators
 
@@ -101,3 +105,12 @@ let checkout_context node_ctxt block_hash =
         (Sc_rollup_node_errors.Cannot_checkout_context
            (block_hash, Some (Context.hash_to_raw_string context_hash)))
   | Some ctxt -> return ctxt
+
+let readonly (node_ctxt : _ t) =
+  {
+    node_ctxt with
+    store = Store.readonly node_ctxt.store;
+    context = Context.readonly node_ctxt.context;
+  }
+
+type 'a delayed_write = ('a, rw) Delayed_write_monad.t
diff --git a/src/proto_alpha/bin_sc_rollup_node/node_context.mli b/src/proto_alpha/bin_sc_rollup_node/node_context.mli
index 5105e52f844d57210e68dac09973923e68a2b480..6ff286da6e4bab3853459aab5979f040d3be0132 100644
--- a/src/proto_alpha/bin_sc_rollup_node/node_context.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/node_context.mli
@@ -28,7 +28,7 @@
 open Protocol
 open Alpha_context
 
-type t = {
+type 'a t = {
   cctxt : Protocol_client_context.full;
       (** Client context used by the rollup node. *)
   dal_cctxt : Dal_node_client.cctxt;
@@ -52,26 +52,34 @@ type t = {
   loser_mode : Loser_mode.t;
       (** If different from [Loser_mode.no_failures], the rollup node
           issues wrong commitments (for tests). *)
-  store : Store.t;  (** The store for the persistent storage. *)
-  context : Context.index;  (** The persistent context for the rollup node. *)
+  store : 'a Store.t;  (** The store for the persistent storage. *)
+  context : 'a Context.index;  (** The persistent context for the rollup node. *)
 }
 
+(** Read/write node context {!t}. *)
+type rw = [`Read | `Write] t
+
+(** Read only node context {!t}. *)
+type ro = [`Read] t
+
 (** [get_operator cctxt purpose] returns the public key hash for the operator
     who has purpose [purpose], if any.
 *)
 val get_operator :
-  t -> Configuration.purpose -> Tezos_crypto.Signature.Public_key_hash.t option
+  _ t ->
+  Configuration.purpose ->
+  Tezos_crypto.Signature.Public_key_hash.t option
 
 (** [is_operator cctxt pkh] returns [true] if the public key hash [pkh] is an
     operator for the node (for any purpose). *)
-val is_operator : t -> Tezos_crypto.Signature.Public_key_hash.t -> bool
+val is_operator : _ t -> Tezos_crypto.Signature.Public_key_hash.t -> bool
 
 (** [get_fee_parameter cctxt purpose] returns the fee parameter to inject an
     operation for a given [purpose]. If no specific fee parameters were
     configured for this purpose, returns the default fee parameter for this
     purpose.
 *)
-val get_fee_parameter : t -> Configuration.purpose -> Injection.fee_parameter
+val get_fee_parameter : _ t -> Configuration.purpose -> Injection.fee_parameter
 
 (** [init cctxt dal_cctxt ~data_dir l1_ctxt sc_rollup genesis_info kind operators fees
     ~loser_mode store context] initialises the rollup representation. The rollup
@@ -88,11 +96,18 @@ val init :
   Configuration.operators ->
   Configuration.fee_parameters ->
   loser_mode:Loser_mode.t ->
-  Store.t ->
-  Context.index ->
-  t tzresult Lwt.t
+  'a Store.t ->
+  'a Context.index ->
+  'a t tzresult Lwt.t
 
 (** [checkout_context node_ctxt block_hash] returns the context at block
     [block_hash]. *)
 val checkout_context :
-  t -> Tezos_crypto.Block_hash.t -> Context.t tzresult Lwt.t
+  'a t -> Tezos_crypto.Block_hash.t -> 'a Context.t tzresult Lwt.t
+
+(** [readonly node_ctxt] returns a read only version of the node context
+    [node_ctxt].  *)
+val readonly : _ t -> ro
+
+(** Monad for values with delayed write effects in the node context. *)
+type 'a delayed_write = ('a, rw) Delayed_write_monad.t
diff --git a/src/proto_alpha/bin_sc_rollup_node/pvm.ml b/src/proto_alpha/bin_sc_rollup_node/pvm.ml
index 7cc342fbee6e949ae3fd55218f0c5254e919c6ca..f47b30948aa55b2ca2aa990ecff4bd96b79921ca 100644
--- a/src/proto_alpha/bin_sc_rollup_node/pvm.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/pvm.ml
@@ -31,7 +31,7 @@ open Alpha_context
 module type S = sig
   include
     Sc_rollup.PVM.S
-      with type context = Context.index
+      with type context = Context.rw_index
        and type hash = Sc_rollup.State_hash.t
 
   (** [get_tick state] gets the total tick counter for the given PVM state. *)
@@ -61,8 +61,11 @@ module type S = sig
 
   (** State storage for this PVM. *)
   module State : sig
+    (** [empty ()] is the empty state.  *)
+    val empty : unit -> state
+
     (** [find context] returns the PVM state stored in the [context], if any. *)
-    val find : Context.t -> state option Lwt.t
+    val find : _ Context.t -> state option Lwt.t
 
     (** [lookup state path] returns the data stored for the path [path] in the
         PVM state [state].  *)
@@ -71,6 +74,6 @@ module type S = sig
     (** [set context state] saves the PVM state [state] in the context and
         returns the updated context. Note: [set] does not perform any write on
         disk, this information must be committed using {!Context.commit}. *)
-    val set : Context.t -> state -> Context.t Lwt.t
+    val set : 'a Context.t -> state -> 'a Context.t Lwt.t
   end
 end
diff --git a/src/proto_alpha/bin_sc_rollup_node/refutation_game.ml b/src/proto_alpha/bin_sc_rollup_node/refutation_game.ml
index 418aace231780c4fc01b370452ef2865dfe3cff6..1a3836d4abedd0bfe57163882ec168023e7c6418 100644
--- a/src/proto_alpha/bin_sc_rollup_node/refutation_game.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/refutation_game.ml
@@ -49,7 +49,7 @@ open Alpha_context
 module type S = sig
   module PVM : Pvm.S
 
-  val process : Layer1.head -> Node_context.t -> unit tzresult Lwt.t
+  val process : Layer1.head -> Node_context.rw -> unit tzresult Lwt.t
 end
 
 module Make (Interpreter : Interpreter.S) :
@@ -76,8 +76,8 @@ module Make (Interpreter : Interpreter.S) :
   (** [inject_next_move node_ctxt source ~refutation ~opponent] submits an L1
       operation (signed by [source]) to issue the next move in the refutation
       game. *)
-  let inject_next_move (node_ctxt : Node_context.t) source ~refutation ~opponent
-      =
+  let inject_next_move (node_ctxt : _ Node_context.t) source ~refutation
+      ~opponent =
     let refute_operation =
       Sc_rollup_refute {rollup = node_ctxt.rollup_address; refutation; opponent}
     in
@@ -129,6 +129,7 @@ module Make (Interpreter : Interpreter.S) :
         let* pages =
           Dal_pages_request.slot_pages ~dal_attestation_lag node_ctxt slot_id
         in
+        let*! pages = Delayed_write_monad.apply node_ctxt pages in
         match pages with
         | None -> return_none (* The slot is not confirmed. *)
         | Some pages -> (
@@ -375,7 +376,7 @@ module Make (Interpreter : Interpreter.S) :
     let* refutation = next_move node_ctxt game in
     inject_next_move node_ctxt self ~refutation:(Some refutation) ~opponent
 
-  let play_timeout (node_ctxt : Node_context.t) self stakers =
+  let play_timeout (node_ctxt : _ Node_context.t) self stakers =
     let timeout_operation =
       Sc_rollup_timeout {rollup = node_ctxt.rollup_address; stakers}
     in
diff --git a/src/proto_alpha/bin_sc_rollup_node/refutation_game.mli b/src/proto_alpha/bin_sc_rollup_node/refutation_game.mli
index 359c5cd99e537242c221b8786c328c13d9d48a03..12f7125221b84c86d8ddbd6890e50c034999b5c3 100644
--- a/src/proto_alpha/bin_sc_rollup_node/refutation_game.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/refutation_game.mli
@@ -30,7 +30,7 @@ module type S = sig
 
   (** [process head node_ctxt] reacts to any operations of [head] related to
       refutation games. *)
-  val process : Layer1.head -> Node_context.t -> unit tzresult Lwt.t
+  val process : Layer1.head -> Node_context.rw -> unit tzresult Lwt.t
 end
 
 module Make (Interpreter : Interpreter.S) : S with module PVM = Interpreter.PVM
diff --git a/src/proto_alpha/bin_sc_rollup_node/state.mli b/src/proto_alpha/bin_sc_rollup_node/state.mli
index 0ad8de3582bdae63959207ffc8873bff9450881b..0e27e37dbaf8ad86e90805a46d853c1e4a67e73c 100644
--- a/src/proto_alpha/bin_sc_rollup_node/state.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/state.mli
@@ -25,33 +25,34 @@
 
 (** [is_processed store hash] returns [true] if the block with [hash] has
     already been processed by the daemon. *)
-val is_processed : Store.t -> Tezos_crypto.Block_hash.t -> bool Lwt.t
+val is_processed : _ Store.t -> Tezos_crypto.Block_hash.t -> bool Lwt.t
 
 (** [mark_processed_head store head] remembers that the [head] is processed. The
     system should not have to come back to it. *)
-val mark_processed_head : Store.t -> Layer1.head -> unit Lwt.t
+val mark_processed_head : Store.rw -> Layer1.head -> unit Lwt.t
 
 (** [last_processed_head_opt store] returns the last processed head if it
     exists. *)
-val last_processed_head_opt : Store.t -> Layer1.head option Lwt.t
+val last_processed_head_opt : _ Store.t -> Layer1.head option Lwt.t
 
 (** [mark_finalized_head store head] remembers that the [head] is finalized. By
     construction, every block whose level is smaller than [head]'s is also
     finalized. *)
-val mark_finalized_head : Store.t -> Layer1.head -> unit Lwt.t
+val mark_finalized_head : Store.rw -> Layer1.head -> unit Lwt.t
 
 (** [last_finalized_head_opt store] returns the last finalized head if it exists. *)
-val get_finalized_head_opt : Store.t -> Layer1.head option Lwt.t
+val get_finalized_head_opt : _ Store.t -> Layer1.head option Lwt.t
 
 (** [hash_of_level store level] returns the current block hash for a
    given [level]. Raise [Invalid_argument] if [hash] does not belong
    to [store]. *)
-val hash_of_level : Store.t -> int32 -> Tezos_crypto.Block_hash.t Lwt.t
+val hash_of_level : _ Store.t -> int32 -> Tezos_crypto.Block_hash.t Lwt.t
 
 (** [level_of_hash store hash] returns the level for Tezos block hash [hash] if
     it is known by the rollup node. *)
-val level_of_hash : Store.t -> Tezos_crypto.Block_hash.t -> int32 tzresult Lwt.t
+val level_of_hash :
+  _ Store.t -> Tezos_crypto.Block_hash.t -> int32 tzresult Lwt.t
 
 (** [set_block_level_and_has store head] registers the correspondences
     [head.level |-> head.hash] and [head.hash |-> head.level] in the store. *)
-val set_block_level_and_hash : Store.t -> Layer1.head -> unit Lwt.t
+val set_block_level_and_hash : Store.rw -> Layer1.head -> unit Lwt.t
diff --git a/src/proto_alpha/bin_sc_rollup_node/store.ml b/src/proto_alpha/bin_sc_rollup_node/store.ml
index 2acb7d7700dc960b393fb5366704a935d0c07bfb..ebfdb2f4753f324c39c8de3b178504142d36c380 100644
--- a/src/proto_alpha/bin_sc_rollup_node/store.ml
+++ b/src/proto_alpha/bin_sc_rollup_node/store.ml
@@ -36,10 +36,20 @@ end)
 
 include Store_utils.Make (IStore)
 
+type 'a store = 'a IStore.t
+
+type 'a t = ([< `Read | `Write > `Read] as 'a) store
+
+type rw = Store_sigs.rw t
+
+type ro = Store_sigs.ro t
+
 let close = IStore.close
 
 let load = IStore.load
 
+let readonly = IStore.readonly
+
 type state_info = {
   num_messages : Z.t;
   num_ticks : Z.t;
diff --git a/src/proto_alpha/bin_sc_rollup_node/store.mli b/src/proto_alpha/bin_sc_rollup_node/store.mli
index 0b5b7be85603c75cc2cc90142d3b941692db3ad2..9031a9921a5ff4e2459183513582e4014a27a3b5 100644
--- a/src/proto_alpha/bin_sc_rollup_node/store.mli
+++ b/src/proto_alpha/bin_sc_rollup_node/store.mli
@@ -33,7 +33,19 @@
 open Protocol
 open Alpha_context
 
-include Store_sigs.Store
+type +'a store
+
+include Store_sigs.Store with type 'a t = 'a store
+
+(** Type of store. The parameter indicates if the store can be written or only
+    read. *)
+type 'a t = ([< `Read | `Write > `Read] as 'a) store
+
+(** Read/write store {!t}. *)
+type rw = Store_sigs.rw t
+
+(** Read only store {!t}. *)
+type ro = Store_sigs.ro t
 
 type state_info = {
   num_messages : Z.t;
@@ -42,17 +54,20 @@ type state_info = {
 }
 
 (** [close store] closes the store. *)
-val close : t -> unit Lwt.t
+val close : _ t -> unit Lwt.t
 
-(** [load directory] loads a store from the data persisted in [directory].*)
-val load : string -> t Lwt.t
+(** [load mode directory] loads a store from the data persisted in [directory].*)
+val load : 'a Store_sigs.mode -> string -> 'a store Lwt.t
+
+(** [readonly store] returns a read-only version of [store]. *)
+val readonly : _ t -> ro
 
 (** Extraneous state information for the PVM *)
 module StateInfo :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Tezos_crypto.Block_hash.t
      and type value = state_info
-     and type store = t
+     and type 'a store = 'a store
 
 module StateHistoryRepr : sig
   type event = {
@@ -76,12 +91,12 @@ module StateHistory : sig
   include
     Store_sigs.Mutable_value
       with type value = StateHistoryRepr.value
-       and type store = t
+       and type 'a store = 'a store
 
-  val insert : t -> StateHistoryRepr.event -> unit Lwt.t
+  val insert : rw -> StateHistoryRepr.event -> unit Lwt.t
 
   val event_of_largest_tick_before :
-    t ->
+    _ t ->
     StateHistoryRepr.TickMap.key ->
     StateHistoryRepr.event option tzresult Lwt.t
 end
@@ -89,55 +104,61 @@ end
 (** Storage for persisting messages downloaded from the L1 node, indexed by
     [Tezos_crypto.Block_hash.t]. *)
 module Messages :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Tezos_crypto.Block_hash.t
      and type value = Sc_rollup.Inbox_message.t list
-     and type store = t
+     and type 'a store = 'a store
 
 (** Aggregated collection of messages from the L1 inbox *)
 module Inboxes :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Tezos_crypto.Block_hash.t
      and type value = Sc_rollup.Inbox.t
-     and type store = t
+     and type 'a store = 'a store
 
 (** Histories from the rollup node. **)
 module Histories :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Tezos_crypto.Block_hash.t
      and type value = Sc_rollup.Inbox.History.t
-     and type store = t
+     and type 'a store = 'a store
 
 (** Storage containing commitments and corresponding commitment hashes that the
     rollup node has knowledge of. *)
 module Commitments :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Raw_level.t
      and type value = Sc_rollup.Commitment.t * Sc_rollup.Commitment.Hash.t
-     and type store = t
+     and type 'a store = 'a store
 
 (** Storage containing the inbox level of the last commitment produced by the
     rollup node. *)
 module Last_stored_commitment_level :
-  Store_sigs.Mutable_value with type value = Raw_level.t and type store = t
+  Store_sigs.Mutable_value
+    with type value = Raw_level.t
+     and type 'a store = 'a store
 
 (** Storage contianing the inbox level of the last commitment published by the
     rollup node. *)
 module Last_published_commitment_level :
-  Store_sigs.Mutable_value with type value = Raw_level.t and type store = t
+  Store_sigs.Mutable_value
+    with type value = Raw_level.t
+     and type 'a store = 'a store
 
 (** Storage containing the inbox level of the last commitment cemented for the
     rollup. The commitment has not been necessarily generated by this rollup
     node. *)
 module Last_cemented_commitment_level :
-  Store_sigs.Mutable_value with type value = Raw_level.t and type store = t
+  Store_sigs.Mutable_value
+    with type value = Raw_level.t
+     and type 'a store = 'a store
 
 (** torage containing the hash  of the last commitment cemented for the rollup.
     The commitment has not been necessarily generated by this rollup node. *)
 module Last_cemented_commitment_hash :
   Store_sigs.Mutable_value
     with type value = Sc_rollup.Commitment.Hash.t
-     and type store = t
+     and type 'a store = 'a store
 
 (** Storage mapping commitment hashes to the level when they were published by
     the rollup node. It only contains hashes of commitments published by this
@@ -146,14 +167,14 @@ module Commitments_published_at_level :
   Store_sigs.Map
     with type key = Sc_rollup.Commitment.Hash.t
      and type value = Raw_level.t
-     and type store = t
+     and type 'a store = 'a store
 
 (** Storage containing the hashes of contexts retrieved from the L1 node. *)
 module Contexts :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Tezos_crypto.Block_hash.t
      and type value = Context.hash
-     and type store = t
+     and type 'a store = 'a store
 
 (** Published slot headers per block hash,
     stored as a list of bindings from [Dal_slot_index.t]
@@ -164,21 +185,21 @@ module Dal_slots_headers :
     with type primary_key = Tezos_crypto.Block_hash.t
      and type secondary_key = Dal.Slot_index.t
      and type value = Dal.Slot.Header.t
-     and type store = t
+     and type 'a store = 'a store
 
 module Dal_confirmed_slots_history :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Tezos_crypto.Block_hash.t
      and type value = Dal.Slots_history.t
-     and type store = t
+     and type 'a store = 'a store
 
 (** Confirmed DAL slots histories cache. See documentation of
     {Dal_slot_repr.Slots_history} for more details. *)
 module Dal_confirmed_slots_histories :
-  Store_sigs.Map
+  Store_sigs.Append_only_map
     with type key = Tezos_crypto.Block_hash.t
      and type value = Dal.Slots_history.History_cache.t
-     and type store = t
+     and type 'a store = 'a store
 
 (** [Dal_slot_pages] is a [Store_utils.Nested_map] used to store the contents
     of dal slots fetched by the rollup node, as a list of pages. The values of
@@ -191,7 +212,7 @@ module Dal_slot_pages :
     with type primary_key = Tezos_crypto.Block_hash.t
      and type secondary_key = Dal.Slot_index.t * Dal.Page.Index.t
      and type value = Dal.Page.content
-     and type store = t
+     and type 'a store = 'a store
 
 (** [Dal_processed_slots] is a [Store_utils.Nested_map] used to store the processing
     status of dal slots content fetched by the rollup node. The values of
@@ -205,4 +226,4 @@ module Dal_processed_slots :
     with type primary_key = Tezos_crypto.Block_hash.t
      and type secondary_key = Dal.Slot_index.t
      and type value = [`Confirmed | `Unconfirmed]
-     and type store = t
+     and type 'a store = 'a store
diff --git a/src/proto_alpha/lib_protocol/alpha_context.mli b/src/proto_alpha/lib_protocol/alpha_context.mli
index cbe118e0b754559d472c1d8bcbffd40cd408ec7a..40733c2cbc483320a86f40cfa3055e8c7bedf29f 100644
--- a/src/proto_alpha/lib_protocol/alpha_context.mli
+++ b/src/proto_alpha/lib_protocol/alpha_context.mli
@@ -3520,7 +3520,7 @@ module Sc_rollup : sig
 
       val state_hash : state -> hash Lwt.t
 
-      val initial_state : context -> state Lwt.t
+      val initial_state : empty:state -> state Lwt.t
 
       val install_boot_sector : state -> string -> state Lwt.t
 
diff --git a/src/proto_alpha/lib_protocol/sc_rollup_PVM_sig.ml b/src/proto_alpha/lib_protocol/sc_rollup_PVM_sig.ml
index d6081b5042ce581a515270fb27fc1df0759e5f38..c4a4d6ca4d7ab37c7d7618b1b44a6ecf0ef6bd27 100644
--- a/src/proto_alpha/lib_protocol/sc_rollup_PVM_sig.ml
+++ b/src/proto_alpha/lib_protocol/sc_rollup_PVM_sig.ml
@@ -424,10 +424,9 @@ module type S = sig
 
   val pp : state -> (Format.formatter -> unit -> unit) Lwt.t
 
-  (** A state is initialized in a given context. A [context]
-      represents the executable environment needed by the state to
-      exist. Typically, the rollup node storage can be part of this
-      context to allow the PVM state to be persistent. *)
+  (** A [context] represents the executable environment needed by the state to
+      exist. Typically, the rollup node storage can be part of this context to
+      allow the PVM state to be persistent. *)
   type context
 
   (** A [hash] characterizes the contents of a state. *)
@@ -496,12 +495,10 @@ module type S = sig
   (** [state_hash state] returns a compressed representation of [state]. *)
   val state_hash : state -> hash Lwt.t
 
-  (** [initial_state context] is the initial state of the PVM, before
-      its specialization with a given [boot_sector].
-
-      The [context] argument is required for technical reasons and does
-      not impact the result. *)
-  val initial_state : context -> state Lwt.t
+  (** [initial_state ~empty] is the initial state of the PVM, before its
+      specialization with a given [boot_sector]. The initial state is built on
+      the [empty] state which must be provided. *)
+  val initial_state : empty:state -> state Lwt.t
 
   (** [install_boot_sector state boot_sector] specializes the initial
       [state] of a PVM using a dedicated [boot_sector], submitted at
diff --git a/src/proto_alpha/lib_protocol/sc_rollup_arith.ml b/src/proto_alpha/lib_protocol/sc_rollup_arith.ml
index 2713ac99e2ca5ec5730de56eefa229e443e43a66..8b160e5183c31c0e0fb4e60926564eb85210fed1 100644
--- a/src/proto_alpha/lib_protocol/sc_rollup_arith.ml
+++ b/src/proto_alpha/lib_protocol/sc_rollup_arith.ml
@@ -45,7 +45,7 @@ module PS = Sc_rollup_PVM_sig
   Its value is the result of the following snippet
 
   {|
-  let*! state = Prover.initial_state context in
+  let*! state = Prover.initial_state ~empty in
   Prover.state_hash state
   |}
 *)
@@ -818,15 +818,14 @@ module Make (Context : P) :
 
   open Monad
 
-  let initial_state ctxt =
-    let state = Tree.empty ctxt in
+  let initial_state ~empty =
     let m =
       let open Monad.Syntax in
       let* () = Status.set Halted in
       return ()
     in
     let open Lwt_syntax in
-    let* state, _ = run m state in
+    let* state, _ = run m empty in
     return state
 
   let install_boot_sector state boot_sector =
@@ -1394,7 +1393,7 @@ module Make (Context : P) :
 
   let produce_origination_proof context boot_sector =
     let open Lwt_result_syntax in
-    let*! state = initial_state context in
+    let*! state = initial_state ~empty:(Tree.empty context) in
     let*! result =
       Context.produce_proof context state (fun state ->
           let open Lwt_syntax in
diff --git a/src/proto_alpha/lib_protocol/sc_rollup_wasm.ml b/src/proto_alpha/lib_protocol/sc_rollup_wasm.ml
index 573c7d4c3054f82cef7b7740db0101d923d89937..a8c4184ee4945cb6c072f6ecdb9ea7a91501ab1e 100644
--- a/src/proto_alpha/lib_protocol/sc_rollup_wasm.ml
+++ b/src/proto_alpha/lib_protocol/sc_rollup_wasm.ml
@@ -194,10 +194,9 @@ module V2_0_0 = struct
 
     open Monad
 
-    let initial_state ctxt =
+    let initial_state ~empty =
       let open Lwt_syntax in
-      let state = Tree.empty ctxt in
-      let* state = Tree.add state ["wasm-version"] (Bytes.of_string "2.0.0") in
+      let* state = Tree.add empty ["wasm-version"] (Bytes.of_string "2.0.0") in
       Lwt.return state
 
     let install_boot_sector state boot_sector =
@@ -375,7 +374,7 @@ module V2_0_0 = struct
 
     let produce_origination_proof context boot_sector =
       let open Lwt_result_syntax in
-      let*! state = initial_state context in
+      let*! state = initial_state ~empty:(Tree.empty context) in
       let*! result =
         Context.produce_proof context state (fun state ->
             let open Lwt_syntax in
diff --git a/src/proto_alpha/lib_protocol/test/integration/operations/test_sc_rollup.ml b/src/proto_alpha/lib_protocol/test/integration/operations/test_sc_rollup.ml
index c221cc126bbb5234b7979f47b6ca664c4d4b716e..a5f2586b4b5573e4f372d081ecf88be149a4ff39 100644
--- a/src/proto_alpha/lib_protocol/test/integration/operations/test_sc_rollup.ml
+++ b/src/proto_alpha/lib_protocol/test/integration/operations/test_sc_rollup.ml
@@ -1710,7 +1710,10 @@ module Arith_pvm = Sc_rollup_helpers.Arith_pvm
 let dumb_proof ~choice =
   let open Lwt_result_syntax in
   let context_arith_pvm = Tezos_context_memory.make_empty_context () in
-  let*! arith_state = Arith_pvm.initial_state context_arith_pvm in
+  let empty =
+    Sc_rollup_helpers.In_memory_context.Tree.empty context_arith_pvm
+  in
+  let*! arith_state = Arith_pvm.initial_state ~empty in
   let*! arith_state = Arith_pvm.install_boot_sector arith_state "" in
   let input = Sc_rollup_helpers.make_external_input "c4c4" in
   let* proof =
@@ -1875,7 +1878,8 @@ let test_dissection_during_final_move () =
 let init_arith_state ~boot_sector =
   let open Lwt_syntax in
   let context = Tezos_context_memory.make_empty_context () in
-  let* state = Arith_pvm.initial_state context in
+  let empty = Sc_rollup_helpers.In_memory_context.Tree.empty context in
+  let* state = Arith_pvm.initial_state ~empty in
   let* state = Arith_pvm.install_boot_sector state boot_sector in
   return (context, state)
 
diff --git a/src/proto_alpha/lib_protocol/test/integration/test_sc_rollup_wasm.ml b/src/proto_alpha/lib_protocol/test/integration/test_sc_rollup_wasm.ml
index a56dca847803c23d7be98836a5a3ac9ff9adb015..b971b3a70cd956f5b4fe3149f59f2d7b05e44f1b 100644
--- a/src/proto_alpha/lib_protocol/test/integration/test_sc_rollup_wasm.ml
+++ b/src/proto_alpha/lib_protocol/test/integration/test_sc_rollup_wasm.ml
@@ -203,7 +203,8 @@ let should_boot_computation_kernel () =
   let boot_sector = computation_kernel () in
   let*! index = Context_binary.init "/tmp" in
   let context = Context_binary.empty index in
-  let*! s = Prover.initial_state context in
+  let empty = Context_binary.Tree.empty context in
+  let*! s = Prover.initial_state ~empty in
   (* sets a reasonable nb-of-tick limit to limit test running time *)
   let*! s =
     Tree.add
diff --git a/src/proto_alpha/lib_protocol/test/pbt/test_refutation_game.ml b/src/proto_alpha/lib_protocol/test/pbt/test_refutation_game.ml
index acf07ed750b90805a79f2612c94bd050ee5ff211..908db413b6adb25d5b41adc9f3a83751dbe61a2a 100644
--- a/src/proto_alpha/lib_protocol/test/pbt/test_refutation_game.ml
+++ b/src/proto_alpha/lib_protocol/test/pbt/test_refutation_game.ml
@@ -789,17 +789,16 @@ module Store_inbox = Sc_rollup_helpers.Store_inbox
 module Arith_test_pvm = struct
   include ArithPVM
 
-  let init_context () = Tezos_context_memory.make_empty_context ()
-
-  let initial_state ctxt =
+  let initial_state () =
     let open Lwt_syntax in
-    let* state = initial_state ctxt in
+    let empty = Tezos_context_memory.make_empty_tree () in
+    let* state = initial_state ~empty in
     let* state = install_boot_sector state "" in
     return state
 
   let initial_hash =
     let open Lwt_syntax in
-    let* state = initial_state (init_context ()) in
+    let* state = initial_state () in
     state_hash state
 
   let consume_fuel = Option.map pred
@@ -869,9 +868,9 @@ module Arith_test_pvm = struct
       (state, fuel, tick, [])
       messages
 
-  let eval_levels_and_messages ~metadata ?fuel ctxt levels_and_messages =
+  let eval_levels_and_messages ~metadata ?fuel levels_and_messages =
     let open Lwt_result_syntax in
-    let*! state = initial_state ctxt in
+    let*! state = initial_state () in
     let*! state_hash = state_hash state in
     let tick = 0 in
     let our_states = [(tick, state_hash)] in
@@ -1050,7 +1049,7 @@ module Player_client = struct
   (** Generate [our_states] for [levels_and_messages] based on the strategy.
       It needs [start_level] and [max_level] in case it will need to generate
       new inputs. *)
-  let gen_our_states ~metadata ctxt strategy ~start_level ~max_level
+  let gen_our_states ~metadata strategy ~start_level ~max_level
       levels_and_messages =
     let open QCheck2.Gen in
     let eval_messages
@@ -1059,10 +1058,7 @@ module Player_client = struct
       @@
       let open Lwt_result_syntax in
       let*! r =
-        Arith_test_pvm.eval_levels_and_messages
-          ~metadata
-          ctxt
-          levels_and_messages
+        Arith_test_pvm.eval_levels_and_messages ~metadata levels_and_messages
       in
       Lwt.return @@ WithExceptions.Result.get_ok ~loc:__LOC__ r
     in
@@ -1168,7 +1164,6 @@ module Player_client = struct
     let* tick, our_states, levels_and_messages =
       gen_our_states
         ~metadata
-        ctxt
         player.strategy
         ~start_level
         ~max_level
@@ -1241,13 +1236,14 @@ let build_proof ~player_client start_tick (game : Game.t) =
     Arith_test_pvm.eval_levels_and_messages
       ~metadata
       ~fuel
-      (Arith_test_pvm.init_context ())
       player_client.levels_and_messages
   in
   let state, _, _ = WithExceptions.Result.get_ok ~loc:__LOC__ r in
   let module P = struct
     include Arith_test_pvm
 
+    let initial_state ~empty:_ = initial_state ()
+
     let context = inbox_context
 
     let state = state
diff --git a/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_arith.ml b/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_arith.ml
index 3afd1c0af6b06180377ea41a98ba5e76d0b9cfce..97102e99436b98884f6a8f731df6df6ca7961f1d 100644
--- a/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_arith.ml
+++ b/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_arith.ml
@@ -109,7 +109,8 @@ let setup boot_sector f =
   let open Lwt_syntax in
   let* index = Context_binary.init "/tmp" in
   let ctxt = Context_binary.empty index in
-  let* state = initial_state ctxt in
+  let empty = Context_binary.Tree.empty ctxt in
+  let* state = initial_state ~empty in
   let* state = install_boot_sector state boot_sector in
   f ctxt state
 
@@ -425,8 +426,8 @@ let test_invalid_outbox_level () =
 let test_initial_state_hash_arith_pvm () =
   let open Alpha_context in
   let open Lwt_result_syntax in
-  let context = Tezos_context_memory.make_empty_context () in
-  let*! state = Sc_rollup_helpers.Arith_pvm.initial_state context in
+  let empty = Tezos_context_memory.make_empty_tree () in
+  let*! state = Sc_rollup_helpers.Arith_pvm.initial_state ~empty in
   let*! hash = Sc_rollup_helpers.Arith_pvm.state_hash state in
   let expected = Sc_rollup.ArithPVM.reference_initial_state_hash in
   if Sc_rollup.State_hash.(hash = expected) then return_unit
diff --git a/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_game.ml b/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_game.ml
index 46377572eecaa69eabc27425b620783510dda769..297265697a2ea8f64adad42d32e1114f8a3c8e90 100644
--- a/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_game.ml
+++ b/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_game.ml
@@ -277,7 +277,8 @@ let test_invalid_serialized_inbox_proof () =
   let dal_snapshot = Dal.Slots_history.genesis in
   let dal_parameters = Default_parameters.constants_mainnet.dal in
   let ctxt = Tezos_context_memory.make_empty_context () in
-  let*! state = Arith_pvm.initial_state ctxt in
+  let empty = Tezos_context_memory.Context.Tree.empty ctxt in
+  let*! state = Arith_pvm.initial_state ~empty in
   (* We evaluate the boot sector, so the [input_requested] is a
      [First_after]. *)
   let*! state = Arith_pvm.eval state in
diff --git a/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_wasm.ml b/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_wasm.ml
index 9f17b6c05f68d6a95890af48ff95384615e9a301..ba0d396897f328092cf6f40110db2a40267b5c7a 100644
--- a/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_wasm.ml
+++ b/src/proto_alpha/lib_protocol/test/unit/test_sc_rollup_wasm.ml
@@ -93,8 +93,8 @@ module Full_Wasm =
 let test_initial_state_hash_wasm_pvm () =
   let open Alpha_context in
   let open Lwt_result_syntax in
-  let context = Tezos_context_memory.make_empty_context () in
-  let*! state = Sc_rollup_helpers.Wasm_pvm.initial_state context in
+  let empty = Tezos_context_memory.make_empty_tree () in
+  let*! state = Sc_rollup_helpers.Wasm_pvm.initial_state ~empty in
   let*! hash = Sc_rollup_helpers.Wasm_pvm.state_hash state in
   let expected = Sc_rollup.Wasm_2_0_0PVM.reference_initial_state_hash in
   if Sc_rollup.State_hash.(hash = expected) then return_unit
diff --git a/src/proto_alpha/lib_sc_rollup/sc_rollup_services.ml b/src/proto_alpha/lib_sc_rollup/sc_rollup_services.ml
index b8dbd535a02bed206bcc0f071ac9dd48fa2aae82..0dbbca4a226aa1247b553e9ef405de14876f03d9 100644
--- a/src/proto_alpha/lib_sc_rollup/sc_rollup_services.ml
+++ b/src/proto_alpha/lib_sc_rollup/sc_rollup_services.ml
@@ -155,57 +155,66 @@ module Global = struct
       ~output:(Data_encoding.option Encodings.commitment_with_hash_and_level)
       (path / "last_stored_commitment")
 
-  let outbox_proof_query =
-    let open Tezos_rpc.Query in
-    let open Sc_rollup in
-    let invalid_message e =
-      raise
-        (Invalid
-           (Format.asprintf
-              "Invalid message (%a)"
-              Environment.Error_monad.pp_trace
-              e))
-    in
-    query (fun outbox_level message_index serialized_outbox_message ->
-        let req name f = function
-          | None ->
-              raise
-                (Invalid (Format.sprintf "Query parameter %s is required" name))
-          | Some arg -> f arg
-        in
-        let outbox_level =
-          req "outbox_level" Raw_level.of_int32_exn outbox_level
-        in
-        let message_index = req "message_index" Z.of_int64 message_index in
-        let message =
-          req
-            "serialized_outbox_message"
-            (fun s -> Outbox.Message.(unsafe_of_string s |> deserialize))
-            serialized_outbox_message
-        in
-        match message with
-        | Error e -> invalid_message e
-        | Ok message -> {outbox_level; message_index; message})
-    |+ opt_field "outbox_level" Tezos_rpc.Arg.int32 (fun o ->
-           Some (Raw_level.to_int32 o.outbox_level))
-    |+ opt_field "message_index" Tezos_rpc.Arg.int64 (fun o ->
-           Some (Z.to_int64 o.message_index))
-    |+ opt_field "serialized_outbox_message" Tezos_rpc.Arg.string (fun o ->
-           match Outbox.Message.serialize o.message with
-           | Ok message -> Some (Outbox.Message.unsafe_to_string message)
-           | Error e -> invalid_message e)
-    |> seal
-
-  let outbox_proof =
-    Tezos_rpc.Service.get_service
-      ~description:"Generate serialized output proof for some outbox message"
-      ~query:outbox_proof_query
-      ~output:
-        Data_encoding.(
-          obj2
-            (req "commitment" Sc_rollup.Commitment.Hash.encoding)
-            (req "proof" Encodings.hex_string))
-      (path / "proofs" / "outbox")
+  module Helpers = struct
+    include Make_services (struct
+      type prefix = unit
+
+      let prefix = open_root / "helpers"
+    end)
+
+    let outbox_proof_query =
+      let open Tezos_rpc.Query in
+      let open Sc_rollup in
+      let invalid_message e =
+        raise
+          (Invalid
+             (Format.asprintf
+                "Invalid message (%a)"
+                Environment.Error_monad.pp_trace
+                e))
+      in
+      query (fun outbox_level message_index serialized_outbox_message ->
+          let req name f = function
+            | None ->
+                raise
+                  (Invalid
+                     (Format.sprintf "Query parameter %s is required" name))
+            | Some arg -> f arg
+          in
+          let outbox_level =
+            req "outbox_level" Raw_level.of_int32_exn outbox_level
+          in
+          let message_index = req "message_index" Z.of_int64 message_index in
+          let message =
+            req
+              "serialized_outbox_message"
+              (fun s -> Outbox.Message.(unsafe_of_string s |> deserialize))
+              serialized_outbox_message
+          in
+          match message with
+          | Error e -> invalid_message e
+          | Ok message -> {outbox_level; message_index; message})
+      |+ opt_field "outbox_level" Tezos_rpc.Arg.int32 (fun o ->
+             Some (Raw_level.to_int32 o.outbox_level))
+      |+ opt_field "message_index" Tezos_rpc.Arg.int64 (fun o ->
+             Some (Z.to_int64 o.message_index))
+      |+ opt_field "serialized_outbox_message" Tezos_rpc.Arg.string (fun o ->
+             match Outbox.Message.serialize o.message with
+             | Ok message -> Some (Outbox.Message.unsafe_to_string message)
+             | Error e -> invalid_message e)
+      |> seal
+
+    let outbox_proof =
+      Tezos_rpc.Service.get_service
+        ~description:"Generate serialized output proof for some outbox message"
+        ~query:outbox_proof_query
+        ~output:
+          Data_encoding.(
+            obj2
+              (req "commitment" Sc_rollup.Commitment.Hash.encoding)
+              (req "proof" Encodings.hex_string))
+        (path / "proofs" / "outbox")
+  end
 
   module Block = struct
     include Make_services (struct