Repository: swimos/swim-rust (language: rust)

Issue:
Alter lanes to be able to persist their state

Store implementation



Discussion / hints:

Fix patch:
diff --git a/.travis.yml b/.travis.yml
index 02247aec2..4843db658 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -6,8 +6,6 @@ language: rust
 branches:
   only:
     - master
-    - server-prototype
-    - ws-move
 
 jobs:
   include:
@@ -27,6 +25,7 @@ jobs:
       script:
         # This can all be run on one machine to save time
         - rustup component add clippy
+        - rustup component list --installed
         - cargo clippy --version
         - rustup component add rustfmt
         - cargo fmt --version
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index 8d734fa1b..adcf50af3 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -1,5 +1,5 @@
-trigger: ["master", "server-prototype", "ws-move"]
-pr: ["master", "server-prototype", "ws-move"]
+trigger: ["master"]
+pr: ["master"]
 
 variables:
   RUSTFLAGS: -Dwarnings
diff --git a/ci/azure-install-rust.yml b/ci/azure-install-rust.yml
index 82e4116b6..13e72a55f 100644
--- a/ci/azure-install-rust.yml
+++ b/ci/azure-install-rust.yml
@@ -26,6 +26,8 @@ steps:
 
   # All platforms.
   - script: |
+      rustup component add rust-src
+      rustup component list --installed
       rustup toolchain list
       rustc -Vv
       cargo -V
diff --git a/demos b/demos
deleted file mode 160000
index 702437304..000000000
--- a/demos
+++ /dev/null
@@ -1,1 +0,0 @@
-Subproject commit 70243730446c4828c9214f8b621734e3c5d80179
diff --git a/swim_common/src/model/time/mod.rs b/swim_common/src/model/time/mod.rs
index 984c712f2..08cbe14dc 100644
--- a/swim_common/src/model/time/mod.rs
+++ b/swim_common/src/model/time/mod.rs
@@ -23,7 +23,7 @@ use crate::form::ValueSchema;
 use crate::model::schema::StandardSchema;
 use crate::model::ValueKind;
 use chrono::{DateTime, LocalResult, TimeZone, Utc};
-use std::fmt::Display;
+use std::fmt::{Display, Formatter};
 
 /// A structure representing the time that it was created.
 #[derive(Copy, Clone, PartialEq, Eq, Debug, Ord, PartialOrd, Hash)]
@@ -35,6 +35,12 @@ impl AsRef<DateTime<Utc>> for Timestamp {
     }
 }
 
+impl Display for Timestamp {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
 impl<TZ> From<DateTime<TZ>> for Timestamp
 where
     TZ: TimeZone,
diff --git a/swim_server/Cargo.toml b/swim_server/Cargo.toml
index 38513d2d6..48bf93e33 100644
--- a/swim_server/Cargo.toml
+++ b/swim_server/Cargo.toml
@@ -46,6 +46,10 @@ lazy_static = "1.4.0"
 percent-encoding = "2.1.0"
 thiserror = "1.0"
 crossbeam-queue = "0.3"
+store = { path = "store" }
+serde = { version = "1.0.124", features = ["derive", "rc"] }
+bincode = "1.3.2"
+rocksdb = "0.16.0"
 
 [dev-dependencies]
 tokio = { version = "1.1.1", features = ["rt-multi-thread", "macros", "test-util"] }
diff --git a/swim_server/agent_derive/src/agent.rs b/swim_server/agent_derive/src/agent.rs
index a8b05fbb1..44c9e3a10 100644
--- a/swim_server/agent_derive/src/agent.rs
+++ b/swim_server/agent_derive/src/agent.rs
@@ -14,6 +14,8 @@
 
 use crate::utils::{get_task_struct_name, validate_input_ast, Callback, InputAstType};
 use crate::utils::{parse_callback, CallbackKind};
+use darling::ast::Data;
+use darling::util::SpannedValue;
 use darling::{ast, FromDeriveInput, FromField, FromMeta};
 use macro_utilities::{as_const, string_to_ident, ungroup};
 use proc_macro::TokenStream;
@@ -24,6 +26,7 @@ use std::ops::Deref;
 use syn::{AttributeArgs, DeriveInput, Path, PathSegment, Type, TypePath, Visibility};
 
 type AgentName = Ident;
+type IdentifiedTokens = (proc_macro2::TokenStream, Ident);
 
 const COMMAND_LANE: &str = "CommandLane";
 const ACTION_LANE: &str = "ActionLane";
@@ -43,6 +46,8 @@ pub struct AgentAttrs {
 #[derive(Debug, FromField)]
 #[darling(attributes(lifecycle))]
 pub struct LifecycleAttrs {
+    #[darling(default)]
+    pub transient: SpannedValue<Option<bool>>,
     pub ident: Option<syn::Ident>,
     pub ty: syn::Type,
     pub vis: syn::Visibility,
@@ -83,7 +88,7 @@ impl LifecycleAttrs {
     }
 }
 
-#[derive(Debug)]
+#[derive(Copy, Clone, PartialEq, Debug)]
 pub enum LaneType {
     Command,
     Action,
@@ -115,7 +120,7 @@ pub fn derive_swim_agent(input: DeriveInput) -> Result<TokenStream, TokenStream>
         }
     };
 
-    let (agent_name, config_type, agent_fields) = get_agent_data(args);
+    let (agent_name, config_type, agent_fields) = get_agent_data(args)?;
 
     let lanes = agent_fields
         .iter()
@@ -131,25 +136,28 @@ pub fn derive_swim_agent(input: DeriveInput) -> Result<TokenStream, TokenStream>
         use std::collections::HashMap;
         use std::boxed::Box;
 
-        use swim_server::agent::{LaneTasks, SwimAgent, AgentContext, LaneIo};
+        use swim_server::agent::{LaneTasks, SwimAgent, AgentContext, IoPair, StoreIo, LaneIo};
         use swim_server::agent::lane::channels::AgentExecutionConfig;
         use swim_server::agent::context::AgentExecutionContext;
         use swim_server::agent::lane::lifecycle::LaneLifecycle;
+        use swim_server::agent::LaneNoStore;
 
         #[automatically_derived]
         impl SwimAgent<#config_type> for #agent_name {
-            fn instantiate<Context: AgentContext<Self> + AgentExecutionContext>(
+            fn instantiate<Context, Store>(
                 configuration: &#config_type,
                 exec_conf: &AgentExecutionConfig,
+                store: Store
             ) -> (
                 Self,
                 Vec<Box<dyn LaneTasks<Self, Context>>>,
-                HashMap<String, Box<dyn LaneIo<Context>>>,
+                HashMap<String, IoPair<Box<dyn LaneIo<Context>>, Box<dyn StoreIo>>>
             )
                 where
                     Context: AgentContext<Self> + AgentExecutionContext + Send + Sync + 'static,
+                    Store: swim_server::agent::store::NodeStore,
             {
-                let mut io_map: HashMap<String, Box<dyn LaneIo<Context>>> = HashMap::new();
+                let mut io_map: HashMap<String, IoPair<Box<dyn LaneIo<Context>>, Box<dyn StoreIo>>> = HashMap::new();
 
                 #(#lifecycles_ast)*
 
@@ -236,13 +244,30 @@ pub struct AgentField {
     pub lifecycle_ast: proc_macro2::TokenStream,
 }
 
+fn check_lane_persistence(
+    lane_type: LaneType,
+    transient: &SpannedValue<bool>,
+) -> Result<(), TokenStream> {
+    if !matches!(lane_type, LaneType::Map | LaneType::Value) && **transient {
+        Err(syn::Error::new(
+            transient.span(),
+            "Stateless lanes cannot persist their data",
+        )
+        .to_compile_error()
+        .into())
+    } else {
+        Ok(())
+    }
+}
+
 fn create_lane(
     lane_type: &LaneType,
     is_public: bool,
     agent_name: &Ident,
     lifecycle: &Ident,
     lane_name: &Ident,
-) -> (proc_macro2::TokenStream, Ident) {
+    transient: SpannedValue<bool>,
+) -> Result<IdentifiedTokens, TokenStream> {
     let lane_name_str = lane_name.to_string();
     let task_variable = get_task_var_name(&lane_name_str);
     let task_structure = get_task_struct_name(&lifecycle.to_string());
@@ -258,10 +283,11 @@ fn create_lane(
         lane_name_lit: &lane_name_lit,
     };
 
+    check_lane_persistence(*lane_type, &transient)?;
+
     let ts = match lane_type {
         LaneType::Command => {
             let model = quote!(let (#lane_name, event_stream) = swim_server::agent::lane::model::command::make_lane_model(exec_conf.action_buffer.clone()););
-
             build_lane_io(
                 lane_data,
                 quote! {
@@ -269,7 +295,7 @@ fn create_lane(
 
                     io_map.insert (
                         #lane_name_lit.to_string(),
-                        Box::new(swim_server::agent::CommandLaneIo::new(#lane_name.clone()))
+                        IoPair::new(Some(Box::new(swim_server::agent::CommandLaneIo::new(#lane_name.clone()))), Some(Box::new(LaneNoStore)))
                     );
                 },
                 model,
@@ -277,7 +303,6 @@ fn create_lane(
         }
         LaneType::Action => {
             let model = quote!(let (#lane_name, event_stream) = swim_server::agent::lane::model::action::make_lane_model(exec_conf.action_buffer.clone()););
-
             build_lane_io(
                 lane_data,
                 quote! {
@@ -285,17 +310,20 @@ fn create_lane(
 
                     io_map.insert (
                         #lane_name_lit.to_string(),
-                        Box::new(swim_server::agent::ActionLaneIo::new(#lane_name.clone()))
+                        IoPair::new(Some(Box::new(swim_server::agent::ActionLaneIo::new(#lane_name.clone()))), Some(Box::new(LaneNoStore)))
                     );
                 },
                 model,
             )
         }
         LaneType::Value => {
+            let is_transient = *transient;
+            let transient = quote!(#is_transient);
+
             let model = quote! {
-                let (#lane_name, observer) = swim_server::agent::lane::model::value::ValueLane::observable(Default::default(), exec_conf.observation_buffer);
+                let (#lane_name, observer, store_io) = swim_server::agent::lane::model::value::streamed_value_lane(#lane_name_lit, exec_conf.observation_buffer, #transient, store.clone());
                 let subscriber = observer.subscriber();
-                let event_stream = observer.into_stream();
+                let event_stream = observer.clone().into_stream();
             };
 
             build_lane_io(
@@ -305,15 +333,18 @@ fn create_lane(
 
                     io_map.insert (
                         #lane_name_lit.to_string(),
-                        Box::new(swim_server::agent::ValueLaneIo::new(#lane_name.clone(), subscriber))
+                        IoPair::new(Some(Box::new(swim_server::agent::ValueLaneIo::new(#lane_name.clone(), subscriber))), store_io)
                     );
                 },
                 model,
             )
         }
         LaneType::Map => {
+            let is_transient = *transient;
+            let transient = quote!(#is_transient);
+
             let model = quote! {
-                let (#lane_name, subscriber, event_stream) = swim_server::agent::lane::model::map::streamed_map_lane(exec_conf.observation_buffer);
+                let (#lane_name, subscriber, event_stream, store_io) = swim_server::agent::lane::model::map::streamed_map_lane(#lane_name_lit, exec_conf.observation_buffer, #transient, store.clone());
             };
 
             build_lane_io(
@@ -322,7 +353,8 @@ fn create_lane(
                     #model
 
                     io_map.insert (
-                        #lane_name_lit.to_string(), Box::new(swim_server::agent::MapLaneIo::new(#lane_name.clone(), subscriber))
+                        #lane_name_lit.to_string(),
+                        IoPair::new(Some(Box::new(swim_server::agent::MapLaneIo::new(#lane_name.clone(), subscriber))), store_io)
                     );
                 },
                 model,
@@ -332,10 +364,12 @@ fn create_lane(
         LaneType::DemandMap => build_demand_map_lane_io(lane_data),
     };
 
-    (ts, task_variable)
+    Ok((ts, task_variable))
 }
 
-pub fn get_agent_data(args: SwimAgentAttrs) -> (AgentName, ConfigType, Vec<AgentField>) {
+pub fn get_agent_data(
+    args: SwimAgentAttrs,
+) -> Result<(AgentName, ConfigType, Vec<AgentField>), TokenStream> {
     let SwimAgentAttrs {
         ident: agent_name,
         data: fields,
@@ -345,30 +379,43 @@ pub fn get_agent_data(args: SwimAgentAttrs) -> (AgentName, ConfigType, Vec<Agent
 
     let mut agent_fields = Vec::new();
 
-    fields.map_struct_fields(|field| {
-        if let (Some(lane_type), Some(lane_name), Some(lifecycle_name)) =
-            (field.get_lane_type(), field.ident, field.name)
-        {
-            let lifecycle_name = Ident::new(&lifecycle_name, Span::call_site());
-            let is_public = matches!(field.vis, Visibility::Public(_));
-
-            let (lifecycle_ast, task_name) = create_lane(
-                &lane_type,
-                is_public,
-                &agent_name,
-                &lifecycle_name,
-                &lane_name,
-            );
-
-            agent_fields.push(AgentField {
-                lane_name,
-                task_name,
-                lifecycle_ast,
-            });
+    match fields {
+        Data::Enum(_) => {
+            unimplemented!()
         }
-    });
+        Data::Struct(fields) => {
+            for field in fields {
+                if let (Some(lane_type), Some(lane_name), Some(lifecycle_name)) =
+                    (field.get_lane_type(), field.ident, field.name)
+                {
+                    let lifecycle_name = Ident::new(&lifecycle_name, Span::call_site());
+                    let is_public = matches!(field.vis, Visibility::Public(_));
+
+                    let span = field.transient.span();
+                    let is_transient = field.transient.unwrap_or_default();
+
+                    let create_result = create_lane(
+                        &lane_type,
+                        is_public,
+                        &agent_name,
+                        &lifecycle_name,
+                        &lane_name,
+                        SpannedValue::new(is_transient, span),
+                    );
+                    match create_result {
+                        Ok((lifecycle_ast, task_name)) => agent_fields.push(AgentField {
+                            lane_name,
+                            task_name,
+                            lifecycle_ast,
+                        }),
+                        Err(e) => return Err(e),
+                    }
+                }
+            }
+        }
+    }
 
-    (agent_name, config_type, agent_fields)
+    Ok((agent_name, config_type, agent_fields))
 }
 
 pub struct LaneData<'a> {
@@ -438,7 +485,7 @@ fn build_demand_lane_io(lane_data: LaneData) -> proc_macro2::TokenStream {
         };
 
         io_map.insert (
-            #lane_name_lit.to_string(), Box::new(swim_server::agent::DemandLaneIo::new(response_rx))
+            #lane_name_lit.to_string(), IoPair::new(Some(Box::new(swim_server::agent::DemandLaneIo::new(response_rx))), Some(Box::new(LaneNoStore)))
         );
     }
 }
@@ -471,7 +518,7 @@ fn build_demand_map_lane_io(lane_data: LaneData) -> proc_macro2::TokenStream {
 
         if #is_public {
             io_map.insert (
-                #lane_name_lit.to_string(), Box::new(swim_server::agent::DemandMapLaneIo::new(#lane_name.clone(), topic))
+                #lane_name_lit.to_string(), IoPair::new(Some(Box::new(swim_server::agent::DemandMapLaneIo::new(#lane_name.clone(), topic))), Some(Box::new(LaneNoStore)))
             );
         }
     }
diff --git a/swim_server/src/agent/context/mod.rs b/swim_server/src/agent/context/mod.rs
index b26631171..16427f0fe 100644
--- a/swim_server/src/agent/context/mod.rs
+++ b/swim_server/src/agent/context/mod.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::agent::store::NodeStore;
 use crate::agent::{AgentContext, Eff};
 use crate::meta::log::NodeLogger;
 use crate::meta::metric::NodeMetricAggregator;
@@ -41,12 +42,13 @@ mod tests;
 /// [`AgentContext`] implementation that dispatches effects to the scheduler through an MPSC
 /// channel.
 #[derive(Debug)]
-pub(super) struct ContextImpl<Agent, Clk, Router> {
+pub(super) struct ContextImpl<Agent, Clk, Router, Store> {
     agent_ref: Arc<Agent>,
     routing_context: RoutingContext<Router>,
     schedule_context: SchedulerContext<Clk>,
     meta_context: Arc<MetaContext>,
     pub(crate) uplinks_idle_since: Arc<AtomicInstant>,
+    store: Store,
 }
 
 const SCHEDULE: &str = "Schedule";
@@ -54,12 +56,13 @@ const SCHED_TRIGGERED: &str = "Schedule triggered";
 const SCHED_STOPPED: &str = "Scheduler unexpectedly stopped";
 const WAITING: &str = "Schedule waiting";
 
-impl<Agent, Clk, Router> ContextImpl<Agent, Clk, Router> {
+impl<Agent, Clk, Router, Store> ContextImpl<Agent, Clk, Router, Store> {
     pub(super) fn new(
         agent_ref: Arc<Agent>,
         routing_context: RoutingContext<Router>,
         schedule_context: SchedulerContext<Clk>,
         meta_context: MetaContext,
+        store: Store,
     ) -> Self {
         ContextImpl {
             agent_ref,
@@ -67,14 +70,16 @@ impl<Agent, Clk, Router> ContextImpl<Agent, Clk, Router> {
             schedule_context,
             meta_context: Arc::new(meta_context),
             uplinks_idle_since: Arc::new(AtomicInstant::new(Instant::now())),
+            store,
         }
     }
 }
 
-impl<Agent, Clk, Router> Clone for ContextImpl<Agent, Clk, Router>
+impl<Agent, Clk, Router, Store> Clone for ContextImpl<Agent, Clk, Router, Store>
 where
     Clk: Clone,
     Router: Clone,
+    Store: Clone,
 {
     fn clone(&self) -> Self {
         ContextImpl {
@@ -83,6 +88,7 @@ where
             schedule_context: self.schedule_context.clone(),
             meta_context: self.meta_context.clone(),
             uplinks_idle_since: self.uplinks_idle_since.clone(),
+            store: self.store.clone(),
         }
     }
 }
@@ -184,10 +190,11 @@ impl<Clk: Clone> Clone for SchedulerContext<Clk> {
     }
 }
 
-impl<Agent, Clk, Router> AgentContext<Agent> for ContextImpl<Agent, Clk, Router>
+impl<Agent, Clk, Router, Store> AgentContext<Agent> for ContextImpl<Agent, Clk, Router, Store>
 where
     Agent: Send + Sync + 'static,
     Clk: Clock,
+    Store: NodeStore,
 {
     fn schedule<Effect, Str, Sch>(&self, effects: Str, schedule: Sch) -> BoxFuture<()>
     where
@@ -226,6 +233,7 @@ where
 /// A context, scoped to an agent, to provide shared functionality to each of its lanes.
 pub trait AgentExecutionContext {
     type Router: ServerRouter + 'static;
+    type Store: NodeStore;
 
     /// Create a handle to the envelope router for the agent.
     fn router_handle(&self) -> Self::Router;
@@ -239,13 +247,19 @@ pub trait AgentExecutionContext {
 
     /// Return the time since the last outgoing message.
     fn uplinks_idle_since(&self) -> &Arc<AtomicInstant>;
+
+    /// Provides a handle to the store engine for this node.
+    fn store(&self) -> Self::Store;
 }
 
-impl<Agent, Clk, RouterInner> AgentExecutionContext for ContextImpl<Agent, Clk, RouterInner>
+impl<Agent, Clk, RouterInner, Store> AgentExecutionContext
+    for ContextImpl<Agent, Clk, RouterInner, Store>
 where
     RouterInner: ServerRouter + Clone + 'static,
+    Store: NodeStore,
 {
     type Router = RouterInner;
+    type Store = Store;
 
     fn router_handle(&self) -> Self::Router {
         self.routing_context.router.clone()
@@ -262,4 +276,8 @@ where
     fn uplinks_idle_since(&self) -> &Arc<AtomicInstant> {
         &self.uplinks_idle_since
     }
+
+    fn store(&self) -> Store {
+        self.store.clone()
+    }
 }
diff --git a/swim_server/src/agent/lane/channels/mod.rs b/swim_server/src/agent/lane/channels/mod.rs
index 63d609184..80cfd3c3f 100644
--- a/swim_server/src/agent/lane/channels/mod.rs
+++ b/swim_server/src/agent/lane/channels/mod.rs
@@ -69,6 +69,8 @@ pub struct AgentExecutionConfig {
     pub metrics: MetricAggregatorConfig,
     /// The maximum idle time before the agent is terminated.
     pub max_idle_time: Duration,
+    /// Maximum number of fatal store errors before the task running a lane will stop.
+    pub max_store_errors: usize,
 }
 
 const DEFAULT_YIELD_COUNT: NonZeroUsize = unsafe { NonZeroUsize::new_unchecked(2048) };
@@ -105,6 +107,7 @@ impl AgentExecutionConfig {
             node_log: LogConfig::default(),
             max_idle_time,
             metrics: Default::default(),
+            max_store_errors: error_threshold,
         }
     }
 }
@@ -133,6 +136,7 @@ impl Default for AgentExecutionConfig {
             node_log: LogConfig::default(),
             metrics: Default::default(),
             max_idle_time: Duration::from_secs(300),
+            max_store_errors: 0,
         }
     }
 }
diff --git a/swim_server/src/agent/lane/mod.rs b/swim_server/src/agent/lane/mod.rs
index f4db03e6e..01dbcdbcd 100644
--- a/swim_server/src/agent/lane/mod.rs
+++ b/swim_server/src/agent/lane/mod.rs
@@ -19,6 +19,7 @@ use swim_common::form::structural::read::ReadError;
 pub mod channels;
 pub mod lifecycle;
 pub mod model;
+pub mod store;
 
 #[cfg(test)]
 pub mod tests;
diff --git a/swim_server/src/agent/lane/model/map/map_store/io.rs b/swim_server/src/agent/lane/model/map/map_store/io.rs
new file mode 100644
index 000000000..7c9f15253
--- /dev/null
+++ b/swim_server/src/agent/lane/model/map/map_store/io.rs
@@ -0,0 +1,103 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::lane::model::map::map_store::MapDataModel;
+use crate::agent::lane::model::map::MapLaneEvent;
+use crate::agent::lane::store::error::{LaneStoreErrorReport, StoreErrorHandler, StoreTaskError};
+use crate::agent::lane::store::StoreIo;
+use crate::agent::store::NodeStore;
+use futures::future::BoxFuture;
+use futures::{Stream, StreamExt};
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::fmt::Debug;
+use store::StoreError;
+use swim_common::form::Form;
+use swim_common::model::time::Timestamp;
+
+/// Map lane store IO task.
+///
+/// This task loads the backing Map lane with a Map from the store if it exists and then
+/// stores all of the events into the delegate store. If any errors are produced, the task will fail
+/// after the store error handler's maximum allowed error policy has been reached.
+///
+/// # Type parameters:
+/// `K` - map lane key type.
+/// `V` - map lane value type.
+/// `Events` - map lane event stream.
+/// `Store` - delegate node store type.
+pub struct MapLaneStoreIo<K, V, Events, Store> {
+    events: Events,
+    model: MapDataModel<Store, K, V>,
+}
+
+impl<K, V, Events, Store> MapLaneStoreIo<K, V, Events, Store> {
+    pub fn new(
+        events: Events,
+        model: MapDataModel<Store, K, V>,
+    ) -> MapLaneStoreIo<K, V, Events, Store> {
+        MapLaneStoreIo { events, model }
+    }
+}
+
+impl<Store, Events, K, V> StoreIo for MapLaneStoreIo<K, V, Events, Store>
+where
+    Store: NodeStore,
+    Events: Stream<Item = MapLaneEvent<K, V>> + Unpin + Send + Sync + 'static,
+    K: Form + Debug + Send + Sync + Serialize + DeserializeOwned + 'static,
+    V: Debug + Send + Sync + Serialize + DeserializeOwned + 'static,
+{
+    fn attach(
+        self,
+        mut error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        Box::pin(async move {
+            let MapLaneStoreIo { mut events, model } = self;
+
+            while let Some(event) = events.next().await {
+                match event {
+                    MapLaneEvent::Checkpoint(_) => {}
+                    MapLaneEvent::Clear => on_event(&mut error_handler, || model.clear())?,
+                    MapLaneEvent::Update(key, value) => {
+                        on_event(&mut error_handler, || model.put(&key, &value))?
+                    }
+                    MapLaneEvent::Remove(key) => {
+                        on_event(&mut error_handler, || model.delete(&key))?
+                    }
+                }
+            }
+            Ok(())
+        })
+    }
+
+    fn attach_boxed(
+        self: Box<Self>,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        (*self).attach(error_handler)
+    }
+}
+
+fn on_event<F>(handler: &mut StoreErrorHandler, f: F) -> Result<(), LaneStoreErrorReport>
+where
+    F: Fn() -> Result<(), StoreError>,
+{
+    match f() {
+        Ok(()) => Ok(()),
+        Err(error) => handler.on_error(StoreTaskError {
+            timestamp: Timestamp::now(),
+            error,
+        }),
+    }
+}
diff --git a/swim_server/src/agent/lane/model/map/map_store/mod.rs b/swim_server/src/agent/lane/model/map/map_store/mod.rs
new file mode 100644
index 000000000..1813fdf08
--- /dev/null
+++ b/swim_server/src/agent/lane/model/map/map_store/mod.rs
@@ -0,0 +1,123 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::store::NodeStore;
+use crate::store::StoreKey;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::marker::PhantomData;
+use store::{deserialize, serialize, serialize_then, StoreError};
+
+#[cfg(test)]
+mod tests;
+
+mod io;
+pub use io::MapLaneStoreIo;
+
+const INCONSISTENT_DB: &str = "Missing key or value in store";
+
+pub struct MapDataModel<D, K, V> {
+    delegate: D,
+    lane_id: u64,
+    _key: PhantomData<K>,
+    _value: PhantomData<V>,
+}
+
+impl<D, K, V> MapDataModel<D, K, V> {
+    pub fn new(delegate: D, lane_id: u64) -> Self {
+        MapDataModel {
+            delegate,
+            lane_id,
+            _key: Default::default(),
+            _value: Default::default(),
+        }
+    }
+}
+
+impl<D, K, V> MapDataModel<D, K, V>
+where
+    D: NodeStore,
+    K: Serialize + DeserializeOwned,
+    V: Serialize + DeserializeOwned,
+{
+    pub fn clear(&self) -> Result<(), StoreError> {
+        unimplemented!()
+    }
+
+    pub fn put(&self, key: &K, value: &V) -> Result<(), StoreError> {
+        let MapDataModel {
+            delegate, lane_id, ..
+        } = self;
+
+        let key = serialize(key)?;
+        let value = serialize(value)?;
+        let k = StoreKey::Map {
+            lane_id: *lane_id,
+            key: Some(key),
+        };
+        delegate.put(k, value.as_slice())
+    }
+
+    pub fn get(&self, key: &K) -> Result<Option<V>, StoreError> {
+        let MapDataModel {
+            delegate, lane_id, ..
+        } = self;
+
+        let opt = serialize_then(delegate, key, |del, key| {
+            del.get(StoreKey::Map {
+                lane_id: *lane_id,
+                key: Some(key),
+            })
+        })?;
+
+        match opt {
+            Some(bytes) => deserialize(bytes.as_slice()).map(Some),
+            None => Ok(None),
+        }
+    }
+
+    pub fn delete(&self, key: &K) -> Result<(), StoreError> {
+        let MapDataModel {
+            delegate, lane_id, ..
+        } = self;
+
+        serialize_then(delegate, key, |del, key| {
+            del.delete(StoreKey::Map {
+                lane_id: *lane_id,
+                key: Some(key),
+            })
+        })
+    }
+
+    pub fn snapshot(&self) -> Result<Option<Vec<(K, V)>>, StoreError> {
+        let store_key = StoreKey::Map {
+            lane_id: self.lane_id,
+            key: None,
+        };
+
+        self.delegate.load_ranged_snapshot(store_key, |key, value| {
+            let store_key = deserialize::<StoreKey>(key)?;
+
+            match store_key {
+                StoreKey::Map { key, .. } => {
+                    let key = deserialize::<K>(&key.ok_or(StoreError::KeyNotFound)?)?;
+                    let value = deserialize::<V>(value)?;
+
+                    Ok((key, value))
+                }
+                StoreKey::Value { .. } => Err(StoreError::Decoding(INCONSISTENT_DB.to_string())),
+            }
+        })
+    }
+}
diff --git a/swim_server/src/agent/lane/model/map/mod.rs b/swim_server/src/agent/lane/model/map/mod.rs
index 98f25cfed..587e4f875 100644
--- a/swim_server/src/agent/lane/model/map/mod.rs
+++ b/swim_server/src/agent/lane/model/map/mod.rs
@@ -29,11 +29,17 @@ use summary::{clear_summary, remove_summary, update_summary};
 use swim_common::form::Form;
 use swim_common::model::Value;
 
+use crate::agent::lane::model::map::map_store::MapLaneStoreIo;
 use crate::agent::lane::model::map::summary::TransactionSummary;
 use crate::agent::lane::model::DeferredSubscription;
 use crate::agent::lane::{InvalidForm, LaneModel};
+use crate::agent::model::map::map_store::MapDataModel;
+use crate::agent::store::NodeStore;
+use crate::agent::StoreIo;
 use futures::stream::{iter, Iter};
 use futures::Stream;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::hash::Hash;
@@ -44,8 +50,16 @@ use swim_warp::model::map::MapUpdate;
 use tracing::{event, Level};
 use utilities::future::{FlatmapStream, SwimStreamExt, Transform};
 
+pub mod map_store;
 mod summary;
 
+pub type StreamedMapLane<K, V, S> = (
+    MapLane<K, V>,
+    MapSubscriber<K, V>,
+    S,
+    Option<Box<dyn StoreIo>>,
+);
+
 /// A lane consisting of a map from keys to values.
 #[derive(Debug)]
 pub struct MapLane<K, V> {
@@ -90,6 +104,39 @@ where
     }
 }
 
+impl<K, V> MapLane<K, V>
+where
+    K: Debug + Form + Send + Sync + Serialize + DeserializeOwned + 'static,
+    V: Debug + Send + Sync + Serialize + DeserializeOwned + 'static,
+{
+    pub fn store_observable<Store: NodeStore>(
+        model: &MapDataModel<Store, K, V>,
+        buffer_size: NonZeroUsize,
+    ) -> (Self, Observer<TransactionSummary<Value, V>>) {
+        let snapshot = model.snapshot().expect("Failed to load map lane state");
+        match snapshot {
+            Some(values) => {
+                let map_state: OrdMap<Value, TVar<V>> = values
+                    .into_iter()
+                    .map(|(k, v)| (k.into_value(), TVar::new(v)))
+                    .collect();
+
+                let (summary, observer) = TVar::new_with_observer(Default::default(), buffer_size);
+                (
+                    MapLane {
+                        map_state: TVar::new(map_state),
+                        summary,
+                        transaction_started: TLocal::new(false),
+                        _key_type: PhantomData,
+                    },
+                    observer,
+                )
+            }
+            None => Self::observable(buffer_size),
+        }
+    }
+}
+
 impl<K, V> Default for MapLane<K, V>
 where
     K: Form + Send + Sync + 'static,
@@ -822,19 +869,36 @@ where
 /// # Arguments
 ///
 /// * `buffer_size` - The size of the buffer for the observer.
-pub fn streamed_map_lane<K, V>(
+/// * `transient` - Whether to persist the lane's state
+/// * `store` - A node store which for persisting data, if the lane is not transient.
+pub fn streamed_map_lane<K, V, Store>(
+    name: impl Into<String>,
     buffer_size: NonZeroUsize,
-) -> (
-    MapLane<K, V>,
-    MapSubscriber<K, V>,
-    impl Stream<Item = MapLaneEvent<K, V>>,
-)
+    transient: bool,
+    store: Store,
+) -> StreamedMapLane<K, V, impl Stream<Item = MapLaneEvent<K, V>>>
 where
-    K: Form + Send + Sync + 'static,
-    V: Send + Sync + 'static,
+    K: Form + Send + Sync + Serialize + DeserializeOwned + Debug + 'static,
+    V: Send + Sync + Debug + Serialize + DeserializeOwned + 'static,
+    Store: NodeStore,
 {
     let (lane, observer) = MapLane::observable(buffer_size);
     let subscriber = MapSubscriber::new(observer.subscriber());
-    let stream = summaries_to_events::<K, V>(observer);
-    (lane, subscriber, stream)
+    let stream = summaries_to_events::<K, V>(observer.clone());
+
+    let store_io: Option<Box<dyn StoreIo>> = if transient {
+        None
+    } else {
+        let lane_id = store
+            .lane_id_of(&name.into())
+            .expect("Failed to fetch lane id");
+        let model = MapDataModel::<Store, K, V>::new(store, lane_id);
+
+        Some(Box::new(MapLaneStoreIo::new(
+            summaries_to_events(observer),
+            model,
+        )))
+    };
+
+    (lane, subscriber, stream, store_io)
 }
diff --git a/swim_server/src/agent/lane/model/value/mod.rs b/swim_server/src/agent/lane/model/value/mod.rs
index 5468da7c9..d7dcb41b7 100644
--- a/swim_server/src/agent/lane/model/value/mod.rs
+++ b/swim_server/src/agent/lane/model/value/mod.rs
@@ -22,6 +22,13 @@ use stm::var::TVar;
 
 #[cfg(test)]
 mod tests;
+mod value_store;
+
+use crate::agent::store::NodeStore;
+use crate::agent::StoreIo;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+pub use value_store::{ValueDataModel, ValueLaneStoreIo};
 
 /// A lane containing a single value.
 #[derive(Debug)]
@@ -50,6 +57,22 @@ impl<T: Any + Send + Sync> ValueLane<T> {
     }
 }
 
+impl<T> ValueLane<T>
+where
+    T: Any + Send + Sync + Serialize + DeserializeOwned + 'static,
+{
+    pub fn store_observable<Store: NodeStore>(
+        model: &ValueDataModel<Store, T>,
+        buffer_size: NonZeroUsize,
+        default: T,
+    ) -> (Self, Observer<T>) {
+        match model.load().expect("Failed to load value lane state") {
+            Some(value) => Self::observable(value, buffer_size),
+            None => Self::observable(default, buffer_size),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct ValueLaneEvent<T> {
     pub previous: Option<Arc<T>>,
@@ -95,3 +118,32 @@ impl<T: Any + Send + Sync> ValueLane<T> {
         self.value.lock().await
     }
 }
+
+pub fn streamed_value_lane<T, Store>(
+    name: impl Into<String>,
+    buffer_size: NonZeroUsize,
+    transient: bool,
+    store: Store,
+) -> (ValueLane<T>, Observer<T>, Option<Box<dyn StoreIo>>)
+where
+    Store: NodeStore,
+    T: Default + Send + Sync + Serialize + DeserializeOwned + 'static,
+{
+    let lane_id = store
+        .lane_id_of(&name.into())
+        .expect("Failed to fetch lane id");
+    let model = ValueDataModel::new(store, lane_id);
+
+    let (lane, observer) = ValueLane::store_observable(&model, buffer_size, Default::default());
+
+    let store_io: Option<Box<dyn StoreIo>> = if transient {
+        None
+    } else {
+        Some(Box::new(ValueLaneStoreIo::new(
+            observer.clone().into_stream(),
+            model,
+        )))
+    };
+
+    (lane, observer, store_io)
+}
diff --git a/swim_server/src/agent/lane/model/value/value_store/io.rs b/swim_server/src/agent/lane/model/value/value_store/io.rs
new file mode 100644
index 000000000..119754846
--- /dev/null
+++ b/swim_server/src/agent/lane/model/value/value_store/io.rs
@@ -0,0 +1,81 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::lane::store::error::{LaneStoreErrorReport, StoreErrorHandler, StoreTaskError};
+use crate::agent::lane::store::StoreIo;
+use crate::agent::model::value::value_store::ValueDataModel;
+use crate::agent::store::NodeStore;
+use futures::future::BoxFuture;
+use futures::{Stream, StreamExt};
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::sync::Arc;
+use swim_common::model::time::Timestamp;
+
+/// Value lane store IO task.
+///
+/// This task loads the backing value lane with a value from the store if it exists and then
+/// stores all of the events into the delegate store. If any errors are produced, the task will fail
+/// after the store error handler's maximum allowed error policy has been reached.
+///
+/// # Type parameters:
+/// `T` - the type of the value lane.
+/// `Events` - events produced by the lane.
+pub struct ValueLaneStoreIo<Store, T, Events> {
+    events: Events,
+    model: ValueDataModel<Store, T>,
+}
+
+impl<Store, T, Events> ValueLaneStoreIo<Store, T, Events> {
+    pub fn new(
+        events: Events,
+        model: ValueDataModel<Store, T>,
+    ) -> ValueLaneStoreIo<Store, T, Events> {
+        ValueLaneStoreIo { events, model }
+    }
+}
+
+impl<Store, Events, T> StoreIo for ValueLaneStoreIo<Store, T, Events>
+where
+    Store: NodeStore,
+    Events: Stream<Item = Arc<T>> + Unpin + Send + Sync + 'static,
+    T: Send + Sync + Serialize + DeserializeOwned + 'static,
+{
+    fn attach(
+        self,
+        mut error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        Box::pin(async move {
+            let ValueLaneStoreIo { mut events, model } = self;
+
+            while let Some(event) = events.next().await {
+                match model.store(&event) {
+                    Ok(()) => continue,
+                    Err(error) => error_handler.on_error(StoreTaskError {
+                        timestamp: Timestamp::now(),
+                        error,
+                    })?,
+                }
+            }
+            Ok(())
+        })
+    }
+
+    fn attach_boxed(
+        self: Box<Self>,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        (*self).attach(error_handler)
+    }
+}
diff --git a/swim_server/src/agent/lane/model/value/value_store/mod.rs b/swim_server/src/agent/lane/model/value/value_store/mod.rs
new file mode 100644
index 000000000..e2d53db2a
--- /dev/null
+++ b/swim_server/src/agent/lane/model/value/value_store/mod.rs
@@ -0,0 +1,77 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::store::NodeStore;
+use crate::store::StoreKey;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::marker::PhantomData;
+use store::{deserialize, serialize_then, StoreError};
+
+#[cfg(test)]
+mod tests;
+
+mod io;
+pub use io::ValueLaneStoreIo;
+
+pub struct ValueDataModel<D, V> {
+    /// The store to delegate this model's operations to.
+    delegate: D,
+    /// The lane URI that this store is operating on.
+    lane_id: u64,
+    _pd: PhantomData<V>,
+}
+
+impl<D: NodeStore, V> ValueDataModel<D, V> {
+    /// Constructs a new value data model.
+    ///
+    /// # Arguments
+    /// `delegate`: if this data model is *not* transient, then delegate operations to this store.
+    /// `lane_id`: the lane URI that this store represents.
+    pub fn new(delegate: D, lane_id: u64) -> Self {
+        ValueDataModel {
+            delegate,
+            lane_id,
+            _pd: Default::default(),
+        }
+    }
+
+    fn key(&self) -> StoreKey {
+        StoreKey::Value {
+            lane_id: self.lane_id,
+        }
+    }
+}
+
+impl<D, V> ValueDataModel<D, V>
+where
+    D: NodeStore,
+    V: DeserializeOwned + Serialize,
+{
+    /// Serialize and store `value`.
+    pub fn store(&self, value: &V) -> Result<(), StoreError> {
+        serialize_then(&self.delegate, value, |delegate, bytes| {
+            delegate.put(self.key(), bytes.as_slice())
+        })
+    }
+
+    /// Loads the value in the store if it exists.
+    pub fn load(&self) -> Result<Option<V>, StoreError> {
+        match self.delegate.get(self.key()) {
+            Ok(Some(bytes)) => deserialize::<V>(bytes.as_slice()).map(Some),
+            Ok(None) => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+}
diff --git a/swim_server/src/agent/lane/store/error.rs b/swim_server/src/agent/lane/store/error.rs
new file mode 100644
index 000000000..48728dd08
--- /dev/null
+++ b/swim_server/src/agent/lane/store/error.rs
@@ -0,0 +1,100 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt::{Display, Formatter};
+use store::StoreError;
+use swim_common::model::time::Timestamp;
+
+#[derive(Debug)]
+pub struct StoreTaskError {
+    pub timestamp: Timestamp,
+    pub error: StoreError,
+}
+
+/// A lane store error report.
+#[derive(Debug)]
+pub struct LaneStoreErrorReport {
+    /// A vector of the store errors and the time at which they were generated.
+    pub(crate) errors: Vec<StoreTaskError>,
+}
+
+impl LaneStoreErrorReport {
+    pub fn new(errors: Vec<StoreTaskError>) -> Self {
+        LaneStoreErrorReport { errors }
+    }
+
+    pub fn for_error(error: StoreTaskError) -> LaneStoreErrorReport {
+        LaneStoreErrorReport {
+            errors: vec![error],
+        }
+    }
+}
+
+impl Display for LaneStoreErrorReport {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let LaneStoreErrorReport { errors } = self;
+
+        writeln!(f, "Lane store error report: ")?;
+
+        writeln!(f, "\t- Errors: ")?;
+
+        for e in errors.iter() {
+            let StoreTaskError { timestamp, error } = e;
+            // todo display for store error
+            writeln!(f, "\t\t{}: `{:?}`", timestamp, error)?;
+        }
+
+        Ok(())
+    }
+}
+
+/// A store error handler which aggregates and timestamps any store errors which are provided.
+/// Upon reaching `max_errors`, the handler will return an error report.
+pub struct StoreErrorHandler {
+    /// The maximum number of errors to aggregate before returning a report.
+    max_errors: usize,
+    /// A vector of the store errors and the time at which they were generated.
+    errors: Vec<StoreTaskError>,
+}
+
+fn is_operational(task_error: &StoreTaskError) -> bool {
+    matches!(
+        task_error.error,
+        StoreError::InitialisationFailure(_) | StoreError::Io(_) | StoreError::Closing
+    )
+}
+
+impl StoreErrorHandler {
+    pub fn new(max_errors: usize) -> StoreErrorHandler {
+        StoreErrorHandler {
+            max_errors,
+            errors: Vec::new(),
+        }
+    }
+
+    pub fn on_error(&mut self, error: StoreTaskError) -> Result<(), LaneStoreErrorReport> {
+        let StoreErrorHandler { max_errors, errors } = self;
+        let is_operational_error = is_operational(&error);
+        errors.push(error);
+
+        let len = errors.len();
+
+        if is_operational_error || len >= *max_errors {
+            let errors: Vec<StoreTaskError> = errors.drain(0..len).collect();
+            Err(LaneStoreErrorReport { errors })
+        } else {
+            Ok(())
+        }
+    }
+}
diff --git a/swim_server/src/agent/lane/store/mod.rs b/swim_server/src/agent/lane/store/mod.rs
new file mode 100644
index 000000000..f278340c7
--- /dev/null
+++ b/swim_server/src/agent/lane/store/mod.rs
@@ -0,0 +1,64 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::lane::store::error::{LaneStoreErrorReport, StoreErrorHandler};
+use futures::future::{ready, BoxFuture};
+
+#[cfg(test)]
+mod tests;
+
+use futures::FutureExt;
+#[cfg(test)]
+pub use tests::store_err_partial_eq;
+
+pub mod error;
+pub mod task;
+
+/// Deferred lane store IO attachment task.
+pub trait StoreIo: Send + 'static {
+    /// Attempt to attach the lane event stream to the provided node store.
+    ///
+    /// # Arguments:
+    /// `store`: the node store that can be used to request a value or map lane store.
+    /// `error_handler`: a store error handler for reporting store error events to. The handler
+    /// is already initialised with an error handling strategy.
+    fn attach(
+        self,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>>;
+
+    fn attach_boxed(
+        self: Box<Self>,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>>;
+}
+
+/// A transient lane store which completes immediately.
+#[derive(Debug)]
+pub struct LaneNoStore;
+impl StoreIo for LaneNoStore {
+    fn attach(
+        self,
+        _error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        ready(Ok(())).boxed()
+    }
+
+    fn attach_boxed(
+        self: Box<Self>,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        (*self).attach(error_handler)
+    }
+}
diff --git a/swim_server/src/agent/lane/store/task.rs b/swim_server/src/agent/lane/store/task.rs
new file mode 100644
index 000000000..757f3117b
--- /dev/null
+++ b/swim_server/src/agent/lane/store/task.rs
@@ -0,0 +1,160 @@
+use crate::agent::lane::store::error::{LaneStoreErrorReport, StoreErrorHandler};
+use crate::agent::lane::store::StoreIo;
+use crate::agent::store::NodeStore;
+use futures::stream::FuturesUnordered;
+use futures::{FutureExt, StreamExt};
+use std::collections::HashMap;
+use std::fmt::{Display, Formatter};
+use store::EngineInfo;
+use utilities::sync::trigger;
+
+/// Aggregated node store errors.
+#[derive(Debug)]
+pub enum NodeStoreErrors {
+    None,
+    Failed {
+        /// Details about the store that generated this error report.
+        store_info: EngineInfo,
+        /// A vector of lane errors and the time at which they were generated.
+        errors: Vec<(String, LaneStoreErrorReport)>,
+    },
+}
+
+impl NodeStoreErrors {
+    #[cfg(test)]
+    pub fn expect_err(self) -> Vec<(String, LaneStoreErrorReport)> {
+        match self {
+            NodeStoreErrors::None => {
+                panic!("Expected errors")
+            }
+            NodeStoreErrors::Failed { errors, .. } => errors,
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        match self {
+            NodeStoreErrors::None => true,
+            NodeStoreErrors::Failed { errors, .. } => errors.is_empty(),
+        }
+    }
+
+    pub fn failed(&self) -> bool {
+        matches!(self, NodeStoreErrors::Failed { .. })
+    }
+
+    pub fn len(&self) -> usize {
+        match self {
+            NodeStoreErrors::None => 0,
+            NodeStoreErrors::Failed { errors, .. } => errors.len(),
+        }
+    }
+
+    pub fn push(&mut self, store_info: EngineInfo, lane_uri: String, report: LaneStoreErrorReport) {
+        match self {
+            NodeStoreErrors::None => {
+                *self = NodeStoreErrors::Failed {
+                    store_info,
+                    errors: vec![(lane_uri, report)],
+                }
+            }
+            NodeStoreErrors::Failed { errors, .. } => errors.push((lane_uri, report)),
+        }
+    }
+}
+
+impl Default for NodeStoreErrors {
+    fn default() -> Self {
+        NodeStoreErrors::None
+    }
+}
+
+impl Display for NodeStoreErrors {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            NodeStoreErrors::None => {
+                writeln!(f, "Node store errors:")?;
+                writeln!(f, "\tfailed: false")
+            }
+            NodeStoreErrors::Failed { store_info, errors } => {
+                writeln!(f, "Node store errors:")?;
+                writeln!(f, "\tfailed: true")?;
+                writeln!(f, "\tDelegate store:")?;
+                writeln!(f, "\t\tPath: `{}`", store_info.path)?;
+                writeln!(f, "\t\tKind: `{}`", store_info.kind)?;
+                writeln!(f, "\terror reports:")?;
+
+                for (lane_uri, report) in errors {
+                    writeln!(f, "\t\t lane URI: {}", lane_uri)?;
+                    writeln!(f, "\t\t report: {}", report)?;
+                }
+
+                Ok(())
+            }
+        }
+    }
+}
+
+/// Node store task manager which attaches lane store IO and aggregates any lane store IO errors.
+pub struct NodeStoreTask<Store> {
+    /// A stop trigger to run lane IO tasks until.
+    stop_rx: trigger::Receiver,
+    /// The node store which lane IO will be attached to.
+    node_store: Store,
+}
+
+impl<Store> NodeStoreTask<Store> {
+    pub fn new(stop_rx: trigger::Receiver, node_store: Store) -> Self {
+        NodeStoreTask {
+            stop_rx,
+            node_store,
+        }
+    }
+}
+
+impl<Store: NodeStore> NodeStoreTask<Store> {
+    /// Run the node store `tasks` and allow each lane to error `max_store_errors` times. Returns
+    /// `Ok(empty error report)` if no tasks failed or `Err(error report)` with reports of any
+    /// failed IO tasks.
+    pub async fn run(
+        self,
+        tasks: HashMap<String, Box<dyn StoreIo>>,
+        max_store_errors: usize,
+    ) -> Result<NodeStoreErrors, NodeStoreErrors> {
+        let NodeStoreTask {
+            stop_rx,
+            node_store,
+        } = self;
+
+        let pending = FuturesUnordered::new();
+
+        for (lane_uri, io) in tasks {
+            let store_error_handler = StoreErrorHandler::new(max_store_errors);
+
+            let task = async move {
+                let result = io.attach_boxed(store_error_handler).await;
+                (lane_uri, result)
+            };
+            pending.push(task.boxed());
+        }
+
+        let tasks = pending.take_until(stop_rx).fuse();
+        let info = node_store.engine_info();
+        let report = tasks
+            .fold(
+                NodeStoreErrors::default(),
+                |mut report, (lane_uri, result)| async {
+                    if let Err(err) = result {
+                        report.push(info.clone(), lane_uri, err);
+                    }
+                    report
+                },
+            )
+            .await;
+
+        if report.is_empty() {
+            Ok(report)
+        } else {
+            Err(report)
+        }
+    }
+}
diff --git a/swim_server/src/agent/mod.rs b/swim_server/src/agent/mod.rs
index 3a38ad3b6..320cb8934 100644
--- a/swim_server/src/agent/mod.rs
+++ b/swim_server/src/agent/mod.rs
@@ -16,6 +16,7 @@ pub mod context;
 pub(crate) mod dispatch;
 pub mod lane;
 pub mod lifecycle;
+pub mod store;
 
 #[cfg(test)]
 mod tests;
@@ -47,7 +48,9 @@ use crate::agent::lane::model::demand_map::{
 use crate::agent::lane::model::map::MapLane;
 use crate::agent::lane::model::map::{summaries_to_events, MapLaneEvent, MapSubscriber};
 use crate::agent::lane::model::supply::{make_lane_model, SupplyLane};
-use crate::agent::lane::model::value::{ValueLane, ValueLaneEvent};
+use crate::agent::lane::model::value::{
+    ValueDataModel, ValueLane, ValueLaneEvent, ValueLaneStoreIo,
+};
 use crate::agent::lane::model::DeferredSubscription;
 use crate::agent::lifecycle::AgentLifecycle;
 use crate::routing::{ServerRouter, TaggedClientEnvelope, TaggedEnvelope};
@@ -76,32 +79,42 @@ use utilities::future::SwimStreamExt;
 use utilities::sync::{topic, trigger};
 use utilities::uri::RelativeUri;
 
+use crate::agent::lane::model::map::map_store::MapDataModel;
+use crate::agent::lane::store::task::{NodeStoreErrors, NodeStoreTask};
+pub use crate::agent::lane::store::{LaneNoStore, StoreIo};
+use crate::agent::model::map::map_store::MapLaneStoreIo;
+use crate::agent::store::NodeStore;
 use crate::meta::info::{LaneInfo, LaneKind};
 use crate::meta::log::NodeLogger;
 use crate::meta::open_meta_lanes;
 #[doc(hidden)]
 #[allow(unused_imports)]
 pub use agent_derive::*;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
 use tokio_stream::wrappers::ReceiverStream;
 
 /// Trait that must be implemented for any agent. This is essentially just boilerplate and will
 /// eventually be implemented using a derive macro.
 pub trait SwimAgent<Config>: Any + Send + Sync + Sized {
     /// Create an instance of the agent and life-cycle handles for each of its lanes.
-    fn instantiate<Context>(
+    fn instantiate<Context, Store>(
         configuration: &Config,
         exec_conf: &AgentExecutionConfig,
+        store: Store,
     ) -> (
         Self,
         DynamicLaneTasks<Self, Context>,
         DynamicAgentIo<Context>,
     )
     where
-        Context: AgentContext<Self> + AgentExecutionContext + Send + Sync + 'static;
+        Context: AgentContext<Self> + AgentExecutionContext + Send + Sync + 'static,
+        Store: NodeStore;
 }
 
 pub type DynamicLaneTasks<Agent, Context> = Vec<Box<dyn LaneTasks<Agent, Context>>>;
-pub type DynamicAgentIo<Context> = HashMap<String, Box<dyn LaneIo<Context>>>;
+pub type DynamicAgentIo<Context> =
+    HashMap<String, IoPair<Box<dyn LaneIo<Context>>, Box<dyn StoreIo>>>;
 
 pub const COMMANDED: &str = "Command received";
 pub const ON_COMMAND: &str = "On command handler";
@@ -114,28 +127,58 @@ const LANE_START: &str = "Lane start";
 const SCHEDULER_TASK: &str = "Agent scheduler";
 const ROOT_DISPATCHER_TASK: &str = "Agent envelope dispatcher.";
 const LANE_EVENTS: &str = "Lane events";
+const STORE_TASK: &str = "Lane store task.";
+
+type TaskIoResult<Err> = Result<Result<Err, Err>, oneshot::error::RecvError>;
+type DispatchTaskResult = TaskIoResult<DispatcherErrors>;
+type StoreTaskResult = TaskIoResult<NodeStoreErrors>;
+
+#[derive(Debug, Default)]
+pub struct AgentTaskResult<Err: Debug> {
+    pub errors: Err,
+    pub failed: bool,
+}
 
 #[derive(Debug)]
 pub struct AgentResult {
     pub route: RelativeUri,
-    pub dispatcher_errors: DispatcherErrors,
-    pub failed: bool,
+    pub dispatcher_task: AgentTaskResult<DispatcherErrors>,
+    pub store_task: AgentTaskResult<NodeStoreErrors>,
 }
 
 impl AgentResult {
+    fn result_for<E>(result: TaskIoResult<E>) -> AgentTaskResult<E>
+    where
+        E: Default + Debug,
+    {
+        match result {
+            Ok(Ok(errs)) => AgentTaskResult {
+                errors: errs,
+                failed: false,
+            },
+            Ok(Err(errs)) => AgentTaskResult {
+                errors: errs,
+                failed: true,
+            },
+            _ => AgentTaskResult {
+                errors: E::default(),
+                failed: true,
+            },
+        }
+    }
+
     fn from(
         route: RelativeUri,
-        result: Result<Result<DispatcherErrors, DispatcherErrors>, oneshot::error::RecvError>,
+        dispatcher_result: DispatchTaskResult,
+        store_result: StoreTaskResult,
     ) -> Self {
-        let (errs, failed) = match result {
-            Ok(Ok(errs)) => (errs, false),
-            Ok(Err(errs)) => (errs, true),
-            _ => (Default::default(), true),
-        };
+        let dispatcher_task = Self::result_for(dispatcher_result);
+        let store_task = Self::result_for(store_result);
+
         AgentResult {
             route,
-            dispatcher_errors: errs,
-            failed,
+            dispatcher_task,
+            store_task,
         }
     }
 }
@@ -164,6 +207,21 @@ impl<Config> AgentParameters<Config> {
     }
 }
 
+/// Lane IO pair consisting of routing IO and store IO.
+pub struct IoPair<Routing, Store> {
+    pub routing: Option<Routing>,
+    pub persistence: Option<Store>,
+}
+
+impl<Routing, Store> IoPair<Routing, Store> {
+    pub fn new(routing: Option<Routing>, persistence: Option<Store>) -> Self {
+        IoPair {
+            routing,
+            persistence,
+        }
+    }
+}
+
 /// Creates a single, asynchronous task that manages the lifecycle of an agent, all of its lanes
 /// and any events that are scheduled within it.
 ///
@@ -175,12 +233,13 @@ impl<Config> AgentParameters<Config> {
 /// * `stop_trigger` - External trigger to cleanly stop the agent.
 /// * `parameters` - Parameters extracted from the agent node route pattern.
 /// * `incoming_envelopes` - The stream of envelopes routed to the agent.
-pub(crate) fn run_agent<Config, Clk, Agent, L, Router>(
+pub(crate) fn run_agent<Config, Clk, Agent, L, Router, Store>(
     lifecycle: L,
     clock: Clk,
     parameters: AgentParameters<Config>,
     incoming_envelopes: impl Stream<Item = TaggedEnvelope> + Send + 'static,
     router: Router,
+    store: Store,
 ) -> (
     Arc<Agent>,
     impl Future<Output = AgentResult> + Send + 'static,
@@ -190,6 +249,7 @@ where
     Agent: SwimAgent<Config> + Send + Sync + 'static,
     L: AgentLifecycle<Agent> + Send + Sync + 'static,
     Router: ServerRouter + Clone + 'static,
+    Store: NodeStore,
 {
     let AgentParameters {
         agent_config,
@@ -200,8 +260,10 @@ where
 
     let span = span!(Level::INFO, AGENT_TASK, %uri);
     let (tripwire, stop_trigger) = trigger::trigger();
-    let (agent, mut tasks, io_providers) =
-        Agent::instantiate::<ContextImpl<Agent, Clk, Router>>(&agent_config, &execution_config);
+    let (agent, mut tasks, io_providers) = Agent::instantiate::<
+        ContextImpl<Agent, Clk, Router, Store>,
+        Store,
+    >(&agent_config, &execution_config, store.clone());
     let agent_ref = Arc::new(agent);
     let agent_cpy = agent_ref.clone();
 
@@ -219,7 +281,7 @@ where
         let task_manager: FuturesUnordered<Instrumented<Eff>> = FuturesUnordered::new();
 
         let (meta_context, mut meta_tasks, meta_io) =
-            open_meta_lanes::<Config, Agent, ContextImpl<Agent, Clk, Router>>(
+            open_meta_lanes::<Config, Agent, ContextImpl<Agent, Clk, Router, Store>>(
                 uri.clone(),
                 &execution_config,
                 lane_summary,
@@ -232,13 +294,53 @@ where
         let (tx, rx) = mpsc::channel(execution_config.scheduler_buffer.get());
         let routing_context = RoutingContext::new(uri.clone(), router, parameters);
         let schedule_context = SchedulerContext::new(tx, clock, stop_trigger.clone());
-        let context = ContextImpl::new(agent_ref, routing_context, schedule_context, meta_context);
+        let context = ContextImpl::new(
+            agent_ref,
+            routing_context,
+            schedule_context,
+            meta_context,
+            store.clone(),
+        );
 
         lifecycle
             .starting(&context)
             .instrument(span!(Level::DEBUG, AGENT_START))
             .await;
 
+        let (mut routing_io, persistence_io) = io_providers.into_iter().fold(
+            (HashMap::new(), HashMap::new()),
+            |(mut routing_io, mut persistence_io), (lane_uri, lane_io)| {
+                let IoPair {
+                    routing,
+                    persistence,
+                } = lane_io;
+
+                if let Some(routing) = routing {
+                    let ident = LaneIdentifier::agent(lane_uri.clone());
+                    routing_io.insert(ident, routing);
+                }
+                if let Some(persistence) = persistence {
+                    persistence_io.insert(lane_uri, persistence);
+                }
+                (routing_io, persistence_io)
+            },
+        );
+
+        routing_io.extend(meta_io);
+
+        let (store_result_tx, store_result_rx) = oneshot::channel();
+        let max_store_errors = execution_config.max_store_errors;
+
+        let store_trigger = stop_trigger.clone();
+        let store_task = async move {
+            let task = NodeStoreTask::new(store_trigger, store);
+            let result = task.run(persistence_io, max_store_errors).await;
+            let _ = store_result_tx.send(result);
+        }
+        .boxed()
+        .instrument(span!(Level::INFO, STORE_TASK));
+        task_manager.push(store_task);
+
         for lane_task in tasks.iter() {
             let lane_name = lane_task.name();
             (**lane_task)
@@ -248,7 +350,7 @@ where
         }
 
         let scheduler_task = ReceiverStream::new(rx)
-            .take_until(stop_trigger)
+            .take_until(stop_trigger.clone())
             .for_each_concurrent(None, |eff| eff)
             .boxed()
             .instrument(span!(Level::TRACE, SCHEDULER_TASK));
@@ -263,17 +365,10 @@ where
             );
         }
 
-        let mut io_providers = io_providers
-            .into_iter()
-            .map(|(k, v)| (LaneIdentifier::agent(k), v))
-            .collect::<HashMap<_, _>>();
-
-        io_providers.extend(meta_io);
-
         let dispatcher =
-            AgentDispatcher::new(uri.clone(), execution_config, context.clone(), io_providers);
+            AgentDispatcher::new(uri.clone(), execution_config, context.clone(), routing_io);
 
-        let (result_tx, result_rx) = oneshot::channel();
+        let (dispatch_result_tx, dispatch_result_rx) = oneshot::channel();
 
         let uplinks_idle_since = context.uplinks_idle_since.clone();
 
@@ -281,7 +376,7 @@ where
             let tripwire = tripwire;
             let result = dispatcher.run(incoming_envelopes, uplinks_idle_since).await;
             tripwire.trigger();
-            let _ = result_tx.send(result);
+            let _ = dispatch_result_tx.send(result);
         }
         .boxed()
         .instrument(span!(Level::INFO, ROOT_DISPATCHER_TASK));
@@ -295,7 +390,10 @@ where
             .map(|_| ()) //Never is an empty type so we can discard the errors.
             .await;
 
-        AgentResult::from(uri, result_rx.await)
+        let dispatch_result = dispatch_result_rx.await;
+        let store_result = store_result_rx.await;
+
+        AgentResult::from(uri, dispatch_result, store_result)
     }
     .instrument(span);
     (agent_cpy, task)
@@ -797,6 +895,26 @@ where
     }
 }
 
+/// Configuration options used when constructing a new lane.
+pub struct LaneConfig {
+    /// The lane URI/name.
+    name: String,
+    /// Whether the lane is public (with respect to external message routing).
+    is_public: bool,
+    /// Whether the lane will persist its data.
+    transient: bool,
+}
+
+impl LaneConfig {
+    pub fn new(name: String, is_public: bool, transient: bool) -> Self {
+        LaneConfig {
+            name,
+            is_public,
+            transient,
+        }
+    }
+}
+
 /// Create a value lane instance along with its life-cycle.
 ///
 /// #Arguments
@@ -807,40 +925,66 @@ where
 /// * `init` - The initial value of the lane.
 /// * `lifecycle` - Life-cycle event handler for the lane.
 /// * `projection` - A projection from the agent type to this lane.
-pub fn make_value_lane<Agent, Context, T, L>(
-    name: impl Into<String>,
-    is_public: bool,
-    config: &AgentExecutionConfig,
+/// * `transient` - Whether to persist the lane's state
+pub fn make_value_lane<Agent, Context, T, L, Store, P>(
+    lane_config: LaneConfig,
+    exec_config: &AgentExecutionConfig,
     init: T,
     lifecycle: L,
-    projection: impl Fn(&Agent) -> &ValueLane<T> + Send + Sync + 'static,
-) -> (
-    ValueLane<T>,
-    impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
-)
+    projection: P,
+    store: Store,
+) -> LaneParts<ValueLane<T>, impl LaneTasks<Agent, Context>, Context>
 where
     Agent: 'static,
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
-    T: Any + Send + Sync + Form + Debug + Default,
+    T: Any + Send + Sync + Form + Default + Serialize + DeserializeOwned + Debug,
     L: for<'l> StatefulLaneLifecycle<'l, ValueLane<T>, Agent>,
+    Store: NodeStore,
+    P: Fn(&Agent) -> &ValueLane<T> + Send + Sync + 'static,
 {
-    let (lane, observer) = ValueLane::observable(init, config.observation_buffer);
+    let LaneConfig {
+        name,
+        is_public,
+        transient,
+    } = lane_config;
+
+    let lane_id = store.lane_id_of(&name).expect("Failed to fetch lane id");
+    let model = ValueDataModel::new(store, lane_id);
+
+    let (lane, observer) =
+        ValueLane::store_observable(&model, exec_config.observation_buffer, init);
 
-    let lane_io = if is_public {
-        Some(ValueLaneIo::new(lane.clone(), observer.subscriber()))
+    let lane_io: Option<Box<dyn LaneIo<Context>>> = if is_public {
+        Some(Box::new(ValueLaneIo::new(
+            lane.clone(),
+            observer.subscriber(),
+        )))
     } else {
         None
     };
 
     let tasks = ValueLifecycleTasks(LifecycleTasks {
-        name: name.into(),
+        name,
         lifecycle,
-        event_stream: observer.into_stream(),
+        event_stream: observer.clone().into_stream(),
         projection,
     });
 
-    (lane, tasks, lane_io)
+    let store_io: Option<Box<dyn StoreIo>> = if transient {
+        None
+    } else {
+        Some(Box::new(ValueLaneStoreIo::new(
+            observer.into_stream(),
+            model,
+        )))
+    };
+
+    let io = IoPair {
+        routing: lane_io,
+        persistence: store_io,
+    };
+
+    LaneParts { lane, tasks, io }
 }
 
 impl<L, S, P> Lane for MapLifecycleTasks<L, S, P> {
@@ -904,43 +1048,61 @@ where
 /// * `config` - Configuration parameters.
 /// * `lifecycle` - Life-cycle event handler for the lane.
 /// * `projection` - A projection from the agent type to this lane.
-pub fn make_map_lane<Agent, Context, K, V, L>(
+/// * `transient` - Whether to persist the lane's state
+pub fn make_map_lane<Agent, Context, K, V, L, P, Store>(
     name: impl Into<String>,
     is_public: bool,
     config: &AgentExecutionConfig,
     lifecycle: L,
-    projection: impl Fn(&Agent) -> &MapLane<K, V> + Send + Sync + 'static,
-) -> (
-    MapLane<K, V>,
-    impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
-)
+    projection: P,
+    transient: bool,
+    store: Store,
+) -> LaneParts<MapLane<K, V>, impl LaneTasks<Agent, Context>, Context>
 where
     Agent: 'static,
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
-    K: Any + Form + Send + Sync + Clone + Debug,
-    V: Any + Form + Send + Sync + Debug,
+    K: Any + Form + Send + Sync + Clone + Debug + Serialize + DeserializeOwned,
+    V: Any + Form + Send + Sync + Debug + Serialize + DeserializeOwned,
     L: for<'l> StatefulLaneLifecycle<'l, MapLane<K, V>, Agent>,
+    Store: NodeStore,
+    P: Fn(&Agent) -> &MapLane<K, V> + Send + Sync + 'static,
 {
-    let (lane, observer) = MapLane::observable(config.observation_buffer);
+    let name = name.into();
+    let lane_id = store.lane_id_of(&name).expect("Failed to fetch lane id");
+    let model = MapDataModel::new(store, lane_id);
+    let (lane, observer) = MapLane::store_observable(&model, config.observation_buffer);
 
-    let lane_io = if is_public {
-        Some(MapLaneIo::new(
+    let lane_io: Option<Box<dyn LaneIo<Context>>> = if is_public {
+        Some(Box::new(MapLaneIo::new(
             lane.clone(),
             MapSubscriber::new(observer.subscriber()),
-        ))
+        )))
     } else {
         None
     };
 
     let tasks = MapLifecycleTasks(LifecycleTasks {
-        name: name.into(),
+        name,
         lifecycle,
-        event_stream: summaries_to_events(observer),
+        event_stream: summaries_to_events(observer.clone()),
         projection,
     });
 
-    (lane, tasks, lane_io)
+    let store_io: Option<Box<dyn StoreIo>> = if transient {
+        None
+    } else {
+        Some(Box::new(MapLaneStoreIo::new(
+            summaries_to_events(observer),
+            model,
+        )))
+    };
+
+    let io = IoPair {
+        routing: lane_io,
+        persistence: store_io,
+    };
+
+    LaneParts { lane, tasks, io }
 }
 
 impl<L, S, P> Lane for ActionLifecycleTasks<L, S, P> {
@@ -1063,11 +1225,7 @@ pub fn make_action_lane<Agent, Context, Command, Response, L, S, P>(
     lifecycle: L,
     projection: impl Fn(&Agent) -> &ActionLane<Command, Response> + Send + Sync + 'static,
     buffer_size: NonZeroUsize,
-) -> (
-    ActionLane<Command, Response>,
-    impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
-)
+) -> LaneParts<ActionLane<Command, Response>, impl LaneTasks<Agent, Context>, Context>
 where
     Agent: 'static,
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
@@ -1086,12 +1244,20 @@ where
         projection,
     });
 
-    let lane_io = if is_public {
-        Some(ActionLaneIo::new(lane.clone()))
+    let lane_io: Option<Box<dyn LaneIo<Context>>> = if is_public {
+        Some(Box::new(ActionLaneIo::new(lane.clone())))
     } else {
         None
     };
-    (lane, tasks, lane_io)
+
+    LaneParts {
+        lane,
+        tasks,
+        io: IoPair {
+            routing: lane_io,
+            persistence: None,
+        },
+    }
 }
 
 /// Create a command lane from a lifecycle.
@@ -1109,11 +1275,7 @@ pub fn make_command_lane<Agent, Context, T, L>(
     lifecycle: L,
     projection: impl Fn(&Agent) -> &CommandLane<T> + Send + Sync + 'static,
     buffer_size: NonZeroUsize,
-) -> (
-    CommandLane<T>,
-    impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
-)
+) -> LaneParts<CommandLane<T>, impl LaneTasks<Agent, Context>, Context>
 where
     Agent: 'static,
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
@@ -1129,13 +1291,20 @@ where
         projection,
     });
 
-    let lane_io = if is_public {
-        Some(CommandLaneIo::new(lane.clone()))
+    let lane_io: Option<Box<dyn LaneIo<Context>>> = if is_public {
+        Some(Box::new(CommandLaneIo::new(lane.clone())))
     } else {
         None
     };
 
-    (lane, tasks, lane_io)
+    LaneParts {
+        lane,
+        tasks,
+        io: IoPair {
+            routing: lane_io,
+            persistence: None,
+        },
+    }
 }
 
 /// Create a new supply lane.
@@ -1149,11 +1318,7 @@ pub fn make_supply_lane<Agent, Context, T>(
     name: impl Into<String>,
     is_public: bool,
     buffer_size: NonZeroUsize,
-) -> (
-    SupplyLane<T>,
-    impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
-)
+) -> LaneParts<SupplyLane<T>, impl LaneTasks<Agent, Context>, Context>
 where
     Agent: 'static,
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
@@ -1166,13 +1331,20 @@ where
         kind: LaneKind::Supply,
     };
 
-    let lane_io = if is_public {
-        Some(SupplyLaneIo::new(view))
+    let lane_io: Option<Box<dyn LaneIo<Context>>> = if is_public {
+        Some(Box::new(SupplyLaneIo::new(view)))
     } else {
         None
     };
 
-    (lane, tasks, lane_io)
+    LaneParts {
+        lane,
+        tasks,
+        io: IoPair {
+            routing: lane_io,
+            persistence: None,
+        },
+    }
 }
 
 struct SupplyLaneIo<S> {
@@ -1221,6 +1393,12 @@ where
     }
 }
 
+pub struct LaneParts<L, T, C> {
+    pub lane: L,
+    pub tasks: T,
+    pub io: IoPair<Box<dyn LaneIo<C>>, Box<dyn StoreIo>>,
+}
+
 /// Create a new demand lane.
 ///
 /// # Arguments
@@ -1234,11 +1412,7 @@ pub fn make_demand_lane<Agent, Context, Event, L>(
     lifecycle: L,
     projection: impl Fn(&Agent) -> &DemandLane<Event> + Send + Sync + 'static,
     buffer_size: NonZeroUsize,
-) -> (
-    DemandLane<Event>,
-    impl LaneTasks<Agent, Context>,
-    impl LaneIo<Context>,
-)
+) -> LaneParts<DemandLane<Event>, impl LaneTasks<Agent, Context>, Context>
 where
     Agent: 'static,
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
@@ -1256,9 +1430,16 @@ where
         response_tx,
     };
 
-    let lane_io = DemandLaneIo::new(response_rx);
+    let lane_io: Box<dyn LaneIo<Context>> = Box::new(DemandLaneIo::new(response_rx));
 
-    (lane, tasks, lane_io)
+    LaneParts {
+        lane,
+        tasks,
+        io: IoPair {
+            routing: Some(lane_io),
+            persistence: None,
+        },
+    }
 }
 
 pub struct DemandLaneIo<Event> {
@@ -1373,11 +1554,7 @@ pub fn make_demand_map_lane<Agent, Context, Key, Value, L>(
     lifecycle: L,
     projection: impl Fn(&Agent) -> &DemandMapLane<Key, Value> + Send + Sync + 'static,
     buffer_size: NonZeroUsize,
-) -> (
-    DemandMapLane<Key, Value>,
-    impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
-)
+) -> LaneParts<DemandMapLane<Key, Value>, impl LaneTasks<Agent, Context>, Context>
 where
     Agent: 'static,
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
@@ -1395,13 +1572,20 @@ where
         projection,
     });
 
-    let lane_io = if is_public {
-        Some(DemandMapLaneIo::new(lane.clone(), topic))
+    let lane_io: Option<Box<dyn LaneIo<Context>>> = if is_public {
+        Some(Box::new(DemandMapLaneIo::new(lane.clone(), topic)))
     } else {
         None
     };
 
-    (lane, tasks, lane_io)
+    LaneParts {
+        lane,
+        tasks,
+        io: IoPair {
+            routing: lane_io,
+            persistence: None,
+        },
+    }
 }
 
 pub struct DemandMapLaneIo<Key, Value>
diff --git a/swim_server/src/agent/store/mock/mod.rs b/swim_server/src/agent/store/mock/mod.rs
new file mode 100644
index 000000000..c56e69ef6
--- /dev/null
+++ b/swim_server/src/agent/store/mock/mod.rs
@@ -0,0 +1,70 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::store::{NodeStore, SwimNodeStore};
+use crate::plane::store::mock::MockPlaneStore;
+use crate::store::{StoreEngine, StoreKey};
+use store::{EngineInfo, StoreError};
+
+#[derive(Clone, Debug)]
+pub struct MockNodeStore {
+    _priv: (),
+}
+
+impl MockNodeStore {
+    pub fn mock() -> SwimNodeStore<MockPlaneStore> {
+        let plane_store = MockPlaneStore;
+        SwimNodeStore::new(plane_store, "test_node")
+    }
+}
+
+impl NodeStore for MockNodeStore {
+    type Delegate = MockPlaneStore;
+
+    fn engine_info(&self) -> EngineInfo {
+        EngineInfo {
+            path: "".to_string(),
+            kind: "Mock".to_string(),
+        }
+    }
+
+    fn lane_id_of(&self, _lane: &str) -> Result<u64, StoreError> {
+        Ok(0)
+    }
+
+    fn load_ranged_snapshot<F, K, V>(
+        &self,
+        _prefix: StoreKey,
+        _map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+    {
+        Ok(None)
+    }
+}
+
+impl StoreEngine for MockNodeStore {
+    fn put(&self, _key: StoreKey, _value: &[u8]) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn get(&self, _key: StoreKey) -> Result<Option<Vec<u8>>, StoreError> {
+        Ok(None)
+    }
+
+    fn delete(&self, _key: StoreKey) -> Result<(), StoreError> {
+        Ok(())
+    }
+}
diff --git a/swim_server/src/agent/store/mod.rs b/swim_server/src/agent/store/mod.rs
new file mode 100644
index 000000000..43fdf6215
--- /dev/null
+++ b/swim_server/src/agent/store/mod.rs
@@ -0,0 +1,132 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt::{Debug, Formatter};
+use std::sync::Arc;
+
+use store::{EngineInfo, StoreError};
+use swim_common::model::text::Text;
+
+use crate::plane::store::PlaneStore;
+use crate::store::{StoreEngine, StoreKey};
+
+pub mod mock;
+
+/// A trait for defining store engines which open stores for nodes.
+///
+/// Node stores are responsible for ensuring that the data models that they open for their lanes
+/// correctly map their keys to their delegate store engines.
+///
+/// # Data models
+/// Data models may be either persistent or transient. Persistent data models are backed by a store
+/// in which operations on persistent data models are mapped to the correct node and delegated to
+/// the store; providing that the top-level server store is also persistent.
+///
+/// Transient data models will live in memory for the duration that a handle to the model exists.
+pub trait NodeStore: StoreEngine + Send + Sync + Clone + Debug + 'static {
+    type Delegate: PlaneStore;
+
+    /// Returns information about the delegate store
+    fn engine_info(&self) -> EngineInfo;
+
+    fn lane_id_of(&self, lane: &str) -> Result<u64, StoreError>;
+
+    /// Executes a ranged snapshot read prefixed by a lane key and deserialize each key-value pair
+    /// using `map_fn`.
+    ///
+    /// Returns an optional snapshot iterator if entries were found that will yield deserialized
+    /// key-value pairs.
+    fn load_ranged_snapshot<F, K, V>(
+        &self,
+        prefix: StoreKey,
+        map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>;
+}
+
+/// A node store which is used to open value and map lane data models.
+pub struct SwimNodeStore<D> {
+    /// The plane store that value and map data models will delegate their store engine operations
+    /// to.
+    delegate: Arc<D>,
+    /// The node URI that this store represents.
+    node_uri: Text,
+}
+
+impl<D> Debug for SwimNodeStore<D> {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        f.debug_struct("SwimNodeStore")
+            .field("node_uri", &self.node_uri)
+            .finish()
+    }
+}
+
+impl<D> Clone for SwimNodeStore<D> {
+    fn clone(&self) -> Self {
+        SwimNodeStore {
+            delegate: self.delegate.clone(),
+            node_uri: self.node_uri.clone(),
+        }
+    }
+}
+
+impl<D: PlaneStore> SwimNodeStore<D> {
+    /// Create a new Swim node store which will delegate its engine operations to `delegate` and
+    /// represents a node at `node_uri`.
+    pub fn new<I: Into<Text>>(delegate: D, node_uri: I) -> SwimNodeStore<D> {
+        SwimNodeStore {
+            delegate: Arc::new(delegate),
+            node_uri: node_uri.into(),
+        }
+    }
+}
+
+impl<D: PlaneStore> StoreEngine for SwimNodeStore<D> {
+    fn put(&self, key: StoreKey, value: &[u8]) -> Result<(), StoreError> {
+        self.delegate.put(key, value)
+    }
+
+    fn get(&self, key: StoreKey) -> Result<Option<Vec<u8>>, StoreError> {
+        self.delegate.get(key)
+    }
+
+    fn delete(&self, key: StoreKey) -> Result<(), StoreError> {
+        self.delegate.delete(key)
+    }
+}
+
+impl<D: PlaneStore> NodeStore for SwimNodeStore<D> {
+    type Delegate = D;
+
+    fn engine_info(&self) -> EngineInfo {
+        self.delegate.engine_info()
+    }
+
+    fn lane_id_of(&self, lane: &str) -> Result<u64, StoreError> {
+        let node_id = format!("{}/{}", self.node_uri, lane);
+        self.delegate.node_id_of(node_id)
+    }
+
+    fn load_ranged_snapshot<F, K, V>(
+        &self,
+        prefix: StoreKey,
+        map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+    {
+        self.delegate.get_prefix_range(prefix, map_fn)
+    }
+}
diff --git a/swim_server/src/interface/mod.rs b/swim_server/src/interface/mod.rs
index b644e8c80..62f363003 100644
--- a/swim_server/src/interface/mod.rs
+++ b/swim_server/src/interface/mod.rs
@@ -17,19 +17,24 @@
 //! The module provides methods and structures for creating and running Swim server instances.
 use crate::agent::lane::channels::AgentExecutionConfig;
 use crate::plane::router::PlaneRouter;
-use crate::plane::spec::PlaneSpec;
+use crate::plane::spec::{PlaneBuilder, PlaneSpec};
+use crate::plane::store::SwimPlaneStore;
 use crate::plane::{run_plane, EnvChannel};
 use crate::routing::remote::config::ConnectionConfig;
 use crate::routing::remote::net::dns::Resolver;
 use crate::routing::remote::net::plain::TokioPlainTextNetworking;
 use crate::routing::remote::{RemoteConnectionChannels, RemoteConnectionsTask};
 use crate::routing::{TopLevelRouter, TopLevelRouterFactory};
+use crate::store::RocksOpts;
+use crate::store::{default_keyspaces, RocksDatabase};
+use crate::store::{ServerStore, SwimStore};
 use either::Either;
 use futures::{io, join};
 use std::error::Error;
 use std::fmt::{Display, Formatter};
 use std::net::SocketAddr;
 use std::sync::Arc;
+use store::StoreError;
 use swim_common::routing::ws::tungstenite::TungsteniteWsConnections;
 use swim_runtime::task::TaskError;
 use swim_runtime::time::clock::RuntimeClock;
@@ -44,30 +49,36 @@ use utilities::sync::{promise, trigger};
 pub struct SwimServerBuilder {
     address: Option<SocketAddr>,
     config: SwimServerConfig,
-    planes: Vec<PlaneSpec<RuntimeClock, EnvChannel, PlaneRouter<TopLevelRouter>>>,
+    planes: Vec<
+        PlaneSpec<
+            RuntimeClock,
+            EnvChannel,
+            PlaneRouter<TopLevelRouter>,
+            SwimPlaneStore<RocksDatabase>,
+        >,
+    >,
+    store: ServerStore<RocksOpts>,
 }
 
-impl Default for SwimServerBuilder {
-    fn default() -> Self {
-        SwimServerBuilder {
-            address: None,
-            config: SwimServerConfig::default(),
-            planes: Vec::new(),
-        }
-    }
-}
+type ServerPlaneBuilder = PlaneBuilder<
+    RuntimeClock,
+    EnvChannel,
+    PlaneRouter<TopLevelRouter>,
+    SwimPlaneStore<RocksDatabase>,
+>;
 
 impl SwimServerBuilder {
     /// Create a new server builder with custom configuration.
     ///
     /// # Arguments
     /// * `config` - The custom configuration for the server.
-    pub fn new(config: SwimServerConfig) -> Self {
-        SwimServerBuilder {
+    pub fn new(config: SwimServerConfig) -> io::Result<Self> {
+        Ok(SwimServerBuilder {
             address: None,
             config,
             planes: Vec::new(),
-        }
+            store: ServerStore::new(RocksOpts::default(), default_keyspaces(), "target".into())?,
+        })
     }
 
     /// Set the address of the server.
@@ -89,13 +100,15 @@ impl SwimServerBuilder {
     /// # Example
     /// ```
     /// use std::net::{SocketAddr, IpAddr, Ipv4Addr};
-    /// use swim_server::interface::{SwimServer, SwimServerBuilder};
+    /// use swim_server::interface::{SwimServer, SwimServerBuilder, SwimServerConfig};
     /// use swim_server::RoutePattern;
     /// use swim_server::agent_lifecycle;
     /// use swim_server::agent::SwimAgent;
     /// use swim_server::agent::AgentContext;
     /// use swim_server::plane::spec::PlaneBuilder;
     ///
+    /// # #[tokio::main]
+    /// # async fn main() {
     /// #[derive(Debug, SwimAgent)]
     /// #[agent(config = "RustAgentConfig")]
     /// pub struct RustAgent;
@@ -115,9 +128,9 @@ impl SwimServerBuilder {
     ///     }
     /// }
     ///
-    /// let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
+    /// let mut swim_server_builder = SwimServerBuilder::temporary_store(SwimServerConfig::default(), "test").unwrap();
+    /// let mut plane_builder = swim_server_builder.plane_builder("test").unwrap();
     ///
-    /// let mut plane_builder = PlaneBuilder::new();
     /// plane_builder
     ///     .add_route(
     ///          RoutePattern::parse_str("/rust").unwrap(),
@@ -125,19 +138,27 @@ impl SwimServerBuilder {
     ///          RustAgentLifecycle {},
     ///     ).unwrap();
     ///
-    /// let mut swim_server_builder = SwimServerBuilder::default();
     /// swim_server_builder.add_plane(plane_builder.build());
+    /// # }
     /// ```
-    pub fn add_plane(
-        &mut self,
-        plane: PlaneSpec<RuntimeClock, EnvChannel, PlaneRouter<TopLevelRouter>>,
-    ) {
+    pub fn add_plane(&mut self, plane: PlaneDef) {
         if !self.planes.is_empty() {
             panic!("Multiple planes are not supported yet")
         }
         self.planes.push(plane)
     }
 
+    pub fn plane_builder<I: Into<String>>(
+        &mut self,
+        name: I,
+    ) -> Result<ServerPlaneBuilder, SwimServerBuilderError> {
+        let store = self
+            .store
+            .plane_store(name.into())
+            .map_err(SwimServerBuilderError::StoreError)?;
+        Ok(PlaneBuilder::new(store))
+    }
+
     /// Build the Swim Server.
     ///
     /// Returns an error if an address has not been provided for the server.
@@ -145,13 +166,15 @@ impl SwimServerBuilder {
     /// # Example
     /// ```
     /// use std::net::{SocketAddr, IpAddr, Ipv4Addr};
-    /// use swim_server::interface::{SwimServer, SwimServerBuilder};
+    /// use swim_server::interface::{SwimServer, SwimServerBuilder, SwimServerConfig};
     /// use swim_server::RoutePattern;
     /// use swim_server::agent_lifecycle;
     /// use swim_server::agent::SwimAgent;
     /// use swim_server::agent::AgentContext;
     /// use swim_server::plane::spec::PlaneBuilder;
     ///
+    /// # #[tokio::main]
+    /// # async fn main() {
     /// #[derive(Debug, SwimAgent)]
     /// #[agent(config = "RustAgentConfig")]
     /// pub struct RustAgent;
@@ -171,9 +194,9 @@ impl SwimServerBuilder {
     ///     }
     /// }
     ///
-    /// let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
+    /// let mut swim_server_builder = SwimServerBuilder::temporary_store(SwimServerConfig::default(), "test").unwrap();
+    /// let mut plane_builder = swim_server_builder.plane_builder("test").unwrap();
     ///
-    /// let mut plane_builder = PlaneBuilder::new();
     /// plane_builder
     ///     .add_route(
     ///          RoutePattern::parse_str("/rust").unwrap(),
@@ -181,16 +204,18 @@ impl SwimServerBuilder {
     ///          RustAgentLifecycle {},
     ///     ).unwrap();
     ///
-    /// let mut swim_server_builder = SwimServerBuilder::default();
     /// swim_server_builder.add_plane(plane_builder.build());
     ///
+    /// let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
     /// let (swim_server, server_handle) = swim_server_builder.bind_to(address).build().unwrap();
+    /// # }
     /// ```
     pub fn build(self) -> Result<(SwimServer, ServerHandle), SwimServerBuilderError> {
         let SwimServerBuilder {
             address,
             planes,
             config,
+            store,
         } = self;
 
         let (stop_trigger_tx, stop_trigger_rx) = trigger::trigger();
@@ -203,6 +228,7 @@ impl SwimServerBuilder {
                 address: address.ok_or(SwimServerBuilderError::MissingAddress)?,
                 stop_trigger_rx,
                 address_tx,
+                _store: store,
             },
             ServerHandle {
                 either_address: Either::Left(address_rx),
@@ -210,17 +236,42 @@ impl SwimServerBuilder {
             },
         ))
     }
+
+    /// Constructs a new `SwimServerBuilder` with the default configuration and is backed by a
+    /// temporary store.
+    ///
+    /// A temporary store uses a RocksDB engine that operates out of a temporary directory which is
+    /// deleted when the application terminates.
+    ///
+    /// # Panics
+    /// Panics if the directory cannot be created.
+    pub fn temporary_store(config: SwimServerConfig, prefix: &str) -> io::Result<Self> {
+        Ok(SwimServerBuilder {
+            address: None,
+            store: ServerStore::<RocksOpts>::transient(
+                RocksOpts::default(),
+                default_keyspaces(),
+                prefix,
+            )?,
+            config,
+            planes: vec![],
+        })
+    }
 }
 
+type PlaneDef =
+    PlaneSpec<RuntimeClock, EnvChannel, PlaneRouter<TopLevelRouter>, SwimPlaneStore<RocksDatabase>>;
+
 /// Swim server instance.
 ///
 /// The Swim server runs a plane and its agents and a task for remote connections asynchronously.
 pub struct SwimServer {
     address: SocketAddr,
     config: SwimServerConfig,
-    planes: Vec<PlaneSpec<RuntimeClock, EnvChannel, PlaneRouter<TopLevelRouter>>>,
+    planes: Vec<PlaneDef>,
     stop_trigger_rx: trigger::Receiver,
     address_tx: promise::Sender<SocketAddr>,
+    _store: ServerStore<RocksOpts>,
 }
 
 impl SwimServer {
@@ -238,6 +289,7 @@ impl SwimServer {
             mut planes,
             stop_trigger_rx,
             address_tx,
+            _store,
         } = self;
 
         let SwimServerConfig {
@@ -301,6 +353,8 @@ impl SwimServer {
 pub enum SwimServerBuilderError {
     /// An error that occurs when trying to create a server without providing the address first.
     MissingAddress,
+    /// An error occured when attempting to build the delegate store.
+    StoreError(StoreError),
 }
 
 impl Display for SwimServerBuilderError {
@@ -309,6 +363,9 @@ impl Display for SwimServerBuilderError {
             SwimServerBuilderError::MissingAddress => {
                 write!(f, "Cannot create a swim server without an address")
             }
+            SwimServerBuilderError::StoreError(e) => {
+                write!(f, "{}", e)
+            }
         }
     }
 }
@@ -400,7 +457,7 @@ impl ServerHandle {
     /// use swim_server::interface::{SwimServer, SwimServerBuilder};
     ///
     /// let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
-    /// let (mut swim_server, server_handle) = SwimServerBuilder::default().bind_to(address).build().unwrap();
+    /// let (mut swim_server, server_handle) = SwimServerBuilder::new(Default::default()).unwrap().bind_to(address).build().unwrap();
     ///
     /// let success = server_handle.stop();
     ///
diff --git a/swim_server/src/lib.rs b/swim_server/src/lib.rs
index f7fccc113..edc02eb83 100644
--- a/swim_server/src/lib.rs
+++ b/swim_server/src/lib.rs
@@ -21,6 +21,7 @@ pub mod macros;
 pub mod meta;
 pub mod plane;
 pub mod routing;
+pub mod store;
 
 #[allow(unused_imports)]
 pub use agent_derive::*;
@@ -29,3 +30,6 @@ pub use utilities::future::retryable::strategy::RetryStrategy;
 pub use utilities::future::SwimStreamExt;
 pub use utilities::route_pattern::RoutePattern;
 pub use utilities::uri;
+
+#[doc(hidden)]
+pub use agent::model::value::ValueLaneStoreIo;
diff --git a/swim_server/src/meta/log/mod.rs b/swim_server/src/meta/log/mod.rs
index 7960bc5ea..cfa67b394 100644
--- a/swim_server/src/meta/log/mod.rs
+++ b/swim_server/src/meta/log/mod.rs
@@ -20,9 +20,10 @@ pub mod config;
 use crate::agent::context::AgentExecutionContext;
 use crate::agent::dispatch::LaneIdentifier;
 use crate::agent::lane::model::supply::SupplyLane;
+use crate::agent::Eff;
+use crate::agent::LaneParts;
 use crate::agent::LaneTasks;
 use crate::agent::{make_supply_lane, AgentContext, DynamicLaneTasks, SwimAgent};
-use crate::agent::{Eff, LaneIo};
 use crate::meta::log::config::{FlushStrategy, LogConfig};
 use crate::meta::{IdentifiedAgentIo, MetaNodeAddressed};
 use either::Either;
@@ -33,18 +34,17 @@ use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::fmt::{Debug, Display, Formatter};
 use std::num::NonZeroUsize;
+use swim_common::form::structural::Tag;
 use swim_common::form::Form;
 use swim_common::model::time::Timestamp;
 use swim_common::model::Value;
+use swim_runtime::time::interval::interval;
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::error::SendError;
 use tokio::time::Duration;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{event, span, Level};
 use tracing_futures::{Instrument, Instrumented};
-
-use swim_common::form::structural::Tag;
-use swim_runtime::time::interval::interval;
 use utilities::sync::trigger;
 use utilities::uri::RelativeUri;
 
@@ -450,12 +450,13 @@ where
     let mut lane_ios = HashMap::with_capacity(lane_count);
 
     let mut make_log_lane = |level: LogLevel| {
-        let (lane, task, io) = make_supply_lane(level.uri_ref(), true, config.lane_buffer);
+        let LaneParts { lane, tasks, io } =
+            make_supply_lane(level.uri_ref(), true, config.lane_buffer);
 
-        lane_tasks.push(task.boxed());
+        lane_tasks.push(tasks.boxed());
         lane_ios.insert(
             LaneIdentifier::meta(MetaNodeAddressed::NodeLog(level)),
-            io.expect("Public lane didn't return any lane IO").boxed(),
+            io.routing.expect("Public lane didn't return any lane IO"),
         );
 
         lane
diff --git a/swim_server/src/meta/pulse.rs b/swim_server/src/meta/pulse.rs
index 76e351a1d..c42ccf51f 100644
--- a/swim_server/src/meta/pulse.rs
+++ b/swim_server/src/meta/pulse.rs
@@ -16,7 +16,7 @@ use crate::agent::context::AgentExecutionContext;
 use crate::agent::dispatch::LaneIdentifier;
 use crate::agent::lane::model::supply::SupplyLane;
 use crate::agent::{
-    make_supply_lane, AgentContext, DynamicLaneTasks, LaneIo, LaneTasks, SwimAgent,
+    make_supply_lane, AgentContext, DynamicLaneTasks, LaneIo, LaneParts, LaneTasks, SwimAgent,
 };
 use crate::meta::metric::lane::LanePulse;
 use crate::meta::metric::node::NodePulse;
@@ -58,11 +58,11 @@ where
     Context: AgentContext<Agent> + AgentExecutionContext + Send + Sync + 'static,
     V: Any + Clone + Send + Sync + Form + Debug + Unpin,
 {
-    let (lane, task, io) = make_supply_lane(lane_uri, true, buffer_size);
+    let LaneParts { lane, tasks, io } = make_supply_lane(lane_uri, true, buffer_size);
     (
         lane,
-        task.boxed(),
-        io.expect("Lane returned private IO").boxed(),
+        tasks.boxed(),
+        io.routing.expect("Lane returned private IO"),
     )
 }
 
diff --git a/swim_server/src/plane/mod.rs b/swim_server/src/plane/mod.rs
index 200ac974a..a85c667e1 100644
--- a/swim_server/src/plane/mod.rs
+++ b/swim_server/src/plane/mod.rs
@@ -18,6 +18,7 @@ pub mod lifecycle;
 pub(crate) mod provider;
 pub(crate) mod router;
 pub mod spec;
+pub mod store;
 #[cfg(test)]
 mod tests;
 
@@ -28,6 +29,7 @@ use crate::plane::context::PlaneContext;
 use crate::plane::error::NoAgentAtRoute;
 use crate::plane::router::{PlaneRouter, PlaneRouterFactory};
 use crate::plane::spec::{PlaneSpec, RouteSpec};
+use crate::plane::store::PlaneStore;
 use crate::routing::error::{RouterError, Unresolvable};
 use crate::routing::remote::RawRoute;
 use crate::routing::{ConnectionDropped, RoutingAddr, ServerRouterFactory, TaggedEnvelope};
@@ -56,28 +58,28 @@ use utilities::uri::RelativeUri;
 
 /// Trait for agent routes. An agent route can construct and run any number of instances of a
 /// [`SwimAgent`] type.
-trait AgentRoute<Clk, Envelopes, Router>: Debug + Send {
+trait AgentRoute<Clk, Envelopes, Router, Store>: Debug + Send {
     /// Run an instance of the agent.
     ///
     /// #Arguments
     ///
-    /// * `uri` The specific URI of the agent instance.
-    /// * `parameters` - Named parameters extracted from the agent URI with the route pattern.
+    /// * `route` - The route of the agent instance and named parameters extracted from the agent
+    /// URI with the route pattern.
     /// * `execution_config` - Configuration parameters controlling how the agent runs.
     /// * `clock` - Clock for scheduling events.
     /// * `incoming_envelopes`- The stream of envelopes routed to the agent.
     /// * `router` - The router by which the agent can send messages.
     fn run_agent(
         &self,
-        uri: RelativeUri,
-        parameters: HashMap<String, String>,
+        route: RouteAndParameters,
         execution_config: AgentExecutionConfig,
         clock: Clk,
         incoming_envelopes: Envelopes,
         router: Router,
+        store: Store,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>);
 
-    fn boxed(self) -> BoxAgentRoute<Clk, Envelopes, Router>
+    fn boxed(self) -> BoxAgentRoute<Clk, Envelopes, Router, Store>
     where
         Self: Sized + 'static,
     {
@@ -85,7 +87,8 @@ trait AgentRoute<Clk, Envelopes, Router>: Debug + Send {
     }
 }
 
-type BoxAgentRoute<Clk, Envelopes, Router> = Box<dyn AgentRoute<Clk, Envelopes, Router> + Send>;
+type BoxAgentRoute<Clk, Envelopes, Router, Store> =
+    Box<dyn AgentRoute<Clk, Envelopes, Router, Store> + Send>;
 
 /// Endpoint connected to an agent within the plane.
 #[derive(Debug)]
@@ -265,13 +268,19 @@ impl PlaneContext for ContextImpl {
 }
 /// Contains the specifications of all routes that are within a plane and maintains the map of
 /// currently active routes.
-struct RouteResolver<Clk, DelegateFac: ServerRouterFactory> {
+struct RouteResolver<Clk, DelegateFac, Store>
+where
+    DelegateFac: ServerRouterFactory,
+    Store: PlaneStore,
+{
     /// Clock for scheduling tasks.
     clock: Clk,
     /// The configuration for the agent routes that are opened.
     execution_config: AgentExecutionConfig,
     /// The routes for the plane.
-    routes: Vec<RouteSpec<Clk, EnvChannel, PlaneRouter<DelegateFac::Router>>>,
+    routes: Vec<PlaneRouteSpec<Clk, DelegateFac, Store>>,
+    /// The store that this plane uses to create node stores.
+    store: Store,
     /// Factory to create handles to the plane router when an agent is opened.
     router_fac: PlaneRouterFactory<DelegateFac>,
     /// External trigger that is fired when the plane should stop.
@@ -282,7 +291,19 @@ struct RouteResolver<Clk, DelegateFac: ServerRouterFactory> {
     counter: u32,
 }
 
-impl<Clk: Clock, DelegateFac: ServerRouterFactory> RouteResolver<Clk, DelegateFac> {
+type PlaneRouteSpec<Clk, DelegateFac, Store> = RouteSpec<
+    Clk,
+    EnvChannel,
+    PlaneRouter<<DelegateFac as ServerRouterFactory>::Router>,
+    <Store as PlaneStore>::NodeStore,
+>;
+
+impl<Clk, DelegateFac, Store> RouteResolver<Clk, DelegateFac, Store>
+where
+    Clk: Clock,
+    DelegateFac: ServerRouterFactory,
+    Store: PlaneStore,
+{
     /// Attempts to open an agent at a specified route.
     fn try_open_route<S>(
         &mut self,
@@ -296,6 +317,7 @@ impl<Clk: Clock, DelegateFac: ServerRouterFactory> RouteResolver<Clk, DelegateFa
             clock,
             execution_config,
             routes,
+            store,
             router_fac,
             stop_trigger,
             active_routes,
@@ -308,12 +330,12 @@ impl<Clk: Clock, DelegateFac: ServerRouterFactory> RouteResolver<Clk, DelegateFa
             .expect("Local endpoint counter overflow.");
         let addr = RoutingAddr::local(*counter);
         let (agent, task) = agent_route.run_agent(
-            route.clone(),
-            params,
+            RouteAndParameters::new(route.clone(), params),
             execution_config.clone(),
             clock.clone(),
             ReceiverStream::new(rx).take_until(stop_trigger.clone()),
             router_fac.create_for(addr),
+            store.node_store(route.path()),
         );
         active_routes.add_endpoint(addr, route, LocalEndpoint::new(Arc::downgrade(&agent), tx));
         if spawner.try_add(task).is_err() {
@@ -351,10 +373,10 @@ const PLANE_STOPPED: &str = "The plane has stopped.";
 /// * `spawner` - Spawns tasks to run the agents for the plane.
 /// * `context_channel` - Transmitter and receiver for plane requests.
 /// * `delegate_fac` - Factory for creating delegate routers.
-pub(crate) async fn run_plane<Clk, S, DelegateFac: ServerRouterFactory>(
+pub(crate) async fn run_plane<Clk, S, DelegateFac, Store>(
     execution_config: AgentExecutionConfig,
     clock: Clk,
-    spec: PlaneSpec<Clk, EnvChannel, PlaneRouter<DelegateFac::Router>>,
+    spec: PlaneSpec<Clk, EnvChannel, PlaneRouter<DelegateFac::Router>, Store>,
     stop_trigger: trigger::Receiver,
     spawner: S,
     context_channel: (mpsc::Sender<PlaneRequest>, mpsc::Receiver<PlaneRequest>),
@@ -362,6 +384,8 @@ pub(crate) async fn run_plane<Clk, S, DelegateFac: ServerRouterFactory>(
 ) where
     Clk: Clock,
     S: Spawner<BoxFuture<'static, AgentResult>>,
+    DelegateFac: ServerRouterFactory,
+    Store: PlaneStore,
 {
     event!(Level::DEBUG, STARTING);
     pin_mut!(spawner);
@@ -376,6 +400,7 @@ pub(crate) async fn run_plane<Clk, S, DelegateFac: ServerRouterFactory>(
     let PlaneSpec {
         routes,
         mut lifecycle,
+        store,
     } = spec;
 
     let start_task = async move {
@@ -392,6 +417,7 @@ pub(crate) async fn run_plane<Clk, S, DelegateFac: ServerRouterFactory>(
             clock,
             execution_config,
             routes,
+            store,
             router_fac: PlaneRouterFactory::new(context_tx, delegate_fac),
             stop_trigger,
             active_routes: PlaneActiveRoutes::default(),
@@ -508,27 +534,27 @@ pub(crate) async fn run_plane<Clk, S, DelegateFac: ServerRouterFactory>(
                 }
                 Either::Right(Some(AgentResult {
                     route,
-                    dispatcher_errors,
-                    failed,
+                    dispatcher_task,
+                    store_task,
                 })) => {
                     if let Some(LocalEndpoint { drop_tx, .. }) =
                         resolver.active_routes.remove_endpoint(&route)
                     {
-                        let _ = if failed {
+                        let _ = if dispatcher_task.failed {
                             drop_tx.provide(ConnectionDropped::AgentFailed)
                         } else {
                             drop_tx.provide(ConnectionDropped::Closed)
                         };
                     }
-                    if failed {
-                        event!(Level::ERROR, AGENT_TASK_FAILED, ?route, ?dispatcher_errors);
+                    if dispatcher_task.failed {
+                        event!(Level::ERROR, AGENT_TASK_FAILED, ?route, ?dispatcher_task);
+                    } else {
+                        event!(Level::DEBUG, AGENT_TASK_COMPLETE, ?route, ?dispatcher_task);
+                    }
+                    if store_task.failed {
+                        event!(Level::ERROR, AGENT_TASK_FAILED, ?route, ?store_task);
                     } else {
-                        event!(
-                            Level::DEBUG,
-                            AGENT_TASK_COMPLETE,
-                            ?route,
-                            ?dispatcher_errors
-                        );
+                        event!(Level::DEBUG, AGENT_TASK_COMPLETE, ?route, ?dispatcher_task);
                     }
                 }
                 Either::Left(None) => {
@@ -556,15 +582,27 @@ pub(crate) async fn run_plane<Clk, S, DelegateFac: ServerRouterFactory>(
     event!(Level::DEBUG, PLANE_STOPPED);
 }
 
-type PlaneAgentRoute<Clk, Delegate> = BoxAgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>>;
+type PlaneAgentRoute<Clk, Delegate, Store> =
+    BoxAgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>, Store>;
 type Params = HashMap<String, String>;
 
+pub struct RouteAndParameters {
+    pub route: RelativeUri,
+    pub parameters: HashMap<String, String>,
+}
+
+impl RouteAndParameters {
+    pub fn new(route: RelativeUri, parameters: HashMap<String, String>) -> RouteAndParameters {
+        RouteAndParameters { route, parameters }
+    }
+}
+
 /// Find the appropriate specification for a route along with any parameters derived from the
 /// route pattern.
-fn route_for<'a, Clk, Delegate>(
+fn route_for<'a, Clk, Delegate, Store>(
     route: &RelativeUri,
-    routes: &'a [RouteSpec<Clk, EnvChannel, PlaneRouter<Delegate>>],
-) -> Result<(&'a PlaneAgentRoute<Clk, Delegate>, Params), NoAgentAtRoute> {
+    routes: &'a [RouteSpec<Clk, EnvChannel, PlaneRouter<Delegate>, Store>],
+) -> Result<(&'a PlaneAgentRoute<Clk, Delegate, Store>, Params), NoAgentAtRoute> {
     //TODO This could be a lot more efficient though it would probably only matter for planes with a large number of routes.
     let matched = routes
         .iter()
diff --git a/swim_server/src/plane/provider/mod.rs b/swim_server/src/plane/provider/mod.rs
index 50d640688..e3a63c475 100644
--- a/swim_server/src/plane/provider/mod.rs
+++ b/swim_server/src/plane/provider/mod.rs
@@ -14,18 +14,17 @@
 
 use crate::agent::lane::channels::AgentExecutionConfig;
 use crate::agent::lifecycle::AgentLifecycle;
+use crate::agent::store::NodeStore;
 use crate::agent::{AgentParameters, AgentResult, SwimAgent};
-use crate::plane::AgentRoute;
+use crate::plane::{AgentRoute, RouteAndParameters};
 use crate::routing::{ServerRouter, TaggedEnvelope};
 use futures::future::BoxFuture;
 use futures::{FutureExt, Stream};
 use std::any::Any;
-use std::collections::HashMap;
 use std::fmt::{Debug, Formatter};
 use std::marker::PhantomData;
 use std::sync::Arc;
 use swim_runtime::time::clock::Clock;
-use utilities::uri::RelativeUri;
 
 /// [`AgentRoute`] implementation that spawns agents with a fixed configuration.
 pub struct AgentProvider<Agent, Config, Lifecycle> {
@@ -62,19 +61,20 @@ where
         }
     }
 
-    pub(crate) fn run<Clk, Envelopes, Router>(
+    pub(crate) fn run<Clk, Envelopes, Router, Store>(
         &self,
-        uri: RelativeUri,
-        parameters: HashMap<String, String>,
+        route: RouteAndParameters,
         execution_config: AgentExecutionConfig,
         clock: Clk,
         incoming_envelopes: Envelopes,
         router: Router,
+        store: Store,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>)
     where
         Clk: Clock,
         Envelopes: Stream<Item = TaggedEnvelope> + Send + 'static,
         Router: ServerRouter + Clone + 'static,
+        Store: NodeStore,
     {
         let AgentProvider {
             configuration,
@@ -82,8 +82,9 @@ where
             ..
         } = self;
 
+        let RouteAndParameters { route, parameters } = route;
         let parameters =
-            AgentParameters::new(configuration.clone(), execution_config, uri, parameters);
+            AgentParameters::new(configuration.clone(), execution_config, route, parameters);
 
         let (agent, task) = crate::agent::run_agent(
             lifecycle.clone(),
@@ -91,13 +92,14 @@ where
             parameters,
             incoming_envelopes,
             router,
+            store,
         );
         (agent, task.boxed())
     }
 }
 
-impl<Clk, Envelopes, Router, Agent, Config, Lifecycle> AgentRoute<Clk, Envelopes, Router>
-    for AgentProvider<Agent, Config, Lifecycle>
+impl<Clk, Envelopes, Router, Agent, Config, Lifecycle, Store>
+    AgentRoute<Clk, Envelopes, Router, Store> for AgentProvider<Agent, Config, Lifecycle>
 where
     Clk: Clock,
     Envelopes: Stream<Item = TaggedEnvelope> + Send + 'static,
@@ -105,23 +107,24 @@ where
     Agent: SwimAgent<Config> + Send + Sync + Debug + 'static,
     Config: Send + Sync + Clone + Debug + 'static,
     Lifecycle: AgentLifecycle<Agent> + Send + Sync + Clone + Debug + 'static,
+    Store: NodeStore,
 {
     fn run_agent(
         &self,
-        uri: RelativeUri,
-        parameters: HashMap<String, String>,
+        route: RouteAndParameters,
         execution_config: AgentExecutionConfig,
         clock: Clk,
         incoming_envelopes: Envelopes,
         router: Router,
+        store: Store,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>) {
         self.run(
-            uri,
-            parameters,
+            route,
             execution_config,
             clock,
             incoming_envelopes,
             router,
+            store,
         )
     }
 }
diff --git a/swim_server/src/plane/spec/mod.rs b/swim_server/src/plane/spec/mod.rs
index aed8e0755..7700f7f28 100644
--- a/swim_server/src/plane/spec/mod.rs
+++ b/swim_server/src/plane/spec/mod.rs
@@ -17,6 +17,7 @@ use crate::agent::SwimAgent;
 use crate::plane::error::AmbiguousRoutes;
 use crate::plane::lifecycle::PlaneLifecycle;
 use crate::plane::provider::AgentProvider;
+use crate::plane::store::PlaneStore;
 use crate::plane::{AgentRoute, BoxAgentRoute};
 use crate::routing::{ServerRouter, TaggedEnvelope};
 use futures::Stream;
@@ -30,15 +31,15 @@ mod tests;
 /// Specification of an agent route in a plane. The route pattern describes a parameterized
 /// family of agents, all of which share an implementation.
 #[derive(Debug)]
-pub(super) struct RouteSpec<Clk, Envelopes, Router> {
+pub(super) struct RouteSpec<Clk, Envelopes, Router, Store> {
     pub pattern: RoutePattern,
-    pub agent_route: BoxAgentRoute<Clk, Envelopes, Router>,
+    pub agent_route: BoxAgentRoute<Clk, Envelopes, Router, Store>,
 }
 
-impl<Clk, Envelopes, Router> RouteSpec<Clk, Envelopes, Router> {
+impl<Clk, Envelopes, Router, Store> RouteSpec<Clk, Envelopes, Router, Store> {
     pub(super) fn new(
         pattern: RoutePattern,
-        agent_route: BoxAgentRoute<Clk, Envelopes, Router>,
+        agent_route: BoxAgentRoute<Clk, Envelopes, Router, Store>,
     ) -> Self {
         RouteSpec {
             pattern,
@@ -50,12 +51,19 @@ impl<Clk, Envelopes, Router> RouteSpec<Clk, Envelopes, Router> {
 /// A specification of a plane, consisting of the defined routes and an optional custom lifecycle
 /// for the plane.
 #[derive(Debug)]
-pub struct PlaneSpec<Clk, Envelopes, Router> {
-    pub(super) routes: Vec<RouteSpec<Clk, Envelopes, Router>>,
+pub struct PlaneSpec<Clk, Envelopes, Router, Store>
+where
+    Store: PlaneStore,
+{
+    pub(super) routes: Vec<RouteSpec<Clk, Envelopes, Router, Store::NodeStore>>,
     pub(super) lifecycle: Option<Box<dyn PlaneLifecycle>>,
+    pub(super) store: Store,
 }
 
-impl<Clk, Envelopes, Router> PlaneSpec<Clk, Envelopes, Router> {
+impl<Clk, Envelopes, Router, Store> PlaneSpec<Clk, Envelopes, Router, Store>
+where
+    Store: PlaneStore,
+{
     pub fn routes(&self) -> Vec<RoutePattern> {
         self.routes
             .iter()
@@ -68,29 +76,25 @@ impl<Clk, Envelopes, Router> PlaneSpec<Clk, Envelopes, Router> {
 /// ambiguous routes generating an error. The specification can then be constructed with or
 /// without a place lifecycle.
 #[derive(Debug)]
-pub struct PlaneBuilder<Clk, Envelopes, Router>(PlaneSpec<Clk, Envelopes, Router>);
-
-impl<Clk, Envelopes, Router> PlaneBuilder<Clk, Envelopes, Router> {
-    pub fn new() -> Self {
-        Default::default()
-    }
-}
+pub struct PlaneBuilder<Clk, Envelopes, Router, Store>(PlaneSpec<Clk, Envelopes, Router, Store>)
+where
+    Store: PlaneStore;
 
-impl<Clk, Envelopes, Router> Default for PlaneBuilder<Clk, Envelopes, Router> {
-    fn default() -> Self {
+impl<Clk, Envelopes, Router, Store> PlaneBuilder<Clk, Envelopes, Router, Store>
+where
+    Clk: Clock,
+    Envelopes: Stream<Item = TaggedEnvelope> + Send + 'static,
+    Router: ServerRouter + Clone + 'static,
+    Store: PlaneStore,
+{
+    pub fn new(store: Store) -> Self {
         PlaneBuilder(PlaneSpec {
             routes: vec![],
             lifecycle: None,
+            store,
         })
     }
-}
 
-impl<Clk, Envelopes, Router> PlaneBuilder<Clk, Envelopes, Router>
-where
-    Clk: Clock,
-    Envelopes: Stream<Item = TaggedEnvelope> + Send + 'static,
-    Router: ServerRouter + Clone + 'static,
-{
     /// Attempt to add a new agent route to the plane.
     ///
     /// #Arguments
@@ -127,7 +131,7 @@ where
     }
 
     /// Construct the specification without adding a plane lifecycle.
-    pub fn build(self) -> PlaneSpec<Clk, Envelopes, Router> {
+    pub fn build(self) -> PlaneSpec<Clk, Envelopes, Router, Store> {
         self.0
     }
 
@@ -135,7 +139,7 @@ where
     pub fn build_with_lifecycle(
         mut self,
         custom_lc: Box<dyn PlaneLifecycle>,
-    ) -> PlaneSpec<Clk, Envelopes, Router> {
+    ) -> PlaneSpec<Clk, Envelopes, Router, Store> {
         self.0.lifecycle = Some(custom_lc);
         self.0
     }
diff --git a/swim_server/src/plane/store/mock.rs b/swim_server/src/plane/store/mock.rs
new file mode 100644
index 000000000..58827b42c
--- /dev/null
+++ b/swim_server/src/plane/store/mock.rs
@@ -0,0 +1,81 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::store::mock::MockNodeStore;
+use crate::agent::store::SwimNodeStore;
+use crate::plane::store::PlaneStore;
+use crate::store::{StoreEngine, StoreKey};
+use store::keyspaces::{Keyspace, KeyspaceResolver};
+use store::{EngineInfo, StoreError};
+use swim_common::model::text::Text;
+
+#[derive(Clone, Debug)]
+pub struct MockPlaneStore;
+impl PlaneStore for MockPlaneStore {
+    type NodeStore = SwimNodeStore<MockPlaneStore>;
+
+    fn node_store<I>(&self, _node: I) -> Self::NodeStore
+    where
+        I: Into<Text>,
+    {
+        MockNodeStore::mock()
+    }
+
+    fn get_prefix_range<F, K, V>(
+        &self,
+        _prefix: StoreKey,
+        _map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+    {
+        Ok(None)
+    }
+
+    fn engine_info(&self) -> EngineInfo {
+        EngineInfo {
+            path: "".to_string(),
+            kind: "Mock".to_string(),
+        }
+    }
+
+    fn node_id_of<I>(&self, _lane: I) -> Result<u64, StoreError>
+    where
+        I: Into<String>,
+    {
+        Ok(0)
+    }
+}
+
+impl KeyspaceResolver for MockPlaneStore {
+    type ResolvedKeyspace = ();
+
+    fn resolve_keyspace<K: Keyspace>(&self, _space: &K) -> Option<&Self::ResolvedKeyspace> {
+        None
+    }
+}
+
+impl StoreEngine for MockPlaneStore {
+    fn put(&self, _key: StoreKey, _value: &[u8]) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn get(&self, _key: StoreKey) -> Result<Option<Vec<u8>>, StoreError> {
+        Ok(None)
+    }
+
+    fn delete(&self, _key: StoreKey) -> Result<(), StoreError> {
+        Ok(())
+    }
+}
diff --git a/swim_server/src/plane/store/mod.rs b/swim_server/src/plane/store/mod.rs
new file mode 100644
index 000000000..fb6447a5a
--- /dev/null
+++ b/swim_server/src/plane/store/mod.rs
@@ -0,0 +1,226 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::ffi::OsStr;
+use std::fmt::{Debug, Formatter};
+use std::io;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use store::{serialize, EngineInfo, Store, StoreError};
+use swim_common::model::text::Text;
+
+use crate::agent::store::{NodeStore, SwimNodeStore};
+use crate::store::keystore::KeyStore;
+use crate::store::{KeyspaceName, StoreEngine, StoreKey};
+use store::engines::StoreBuilder;
+use store::keyspaces::Keyspaces;
+
+pub mod mock;
+
+const STORE_DIR: &str = "store";
+const PLANES_DIR: &str = "planes";
+
+/// Creates paths for both map and value stores with a base path of `base_path` and appended by
+/// `plane_name`.
+fn path_for<B, P>(base_path: &B, plane_name: &P) -> PathBuf
+where
+    B: AsRef<OsStr> + ?Sized,
+    P: AsRef<OsStr> + ?Sized,
+{
+    Path::new(base_path)
+        .join(STORE_DIR)
+        .join(PLANES_DIR)
+        .join(plane_name.as_ref())
+}
+
+/// A trait for defining plane stores which will create node stores.
+pub trait PlaneStore
+where
+    Self: StoreEngine + Sized + Debug + Send + Sync + Clone + 'static,
+{
+    /// The type of node stores which are created.
+    type NodeStore: NodeStore;
+
+    /// Create a node store for `node_uri`.
+    fn node_store<I>(&self, node_uri: I) -> Self::NodeStore
+    where
+        I: Into<Text>;
+
+    /// Executes a ranged snapshot read prefixed by a lane key and deserialize each key-value pair
+    /// using `map_fn`.
+    ///
+    /// Returns an optional snapshot iterator if entries were found that will yield deserialized
+    /// key-value pairs.
+    fn get_prefix_range<F, K, V>(
+        &self,
+        prefix: StoreKey,
+        map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>;
+
+    /// Returns information about the delegate store
+    fn engine_info(&self) -> EngineInfo;
+
+    fn node_id_of<I>(&self, node: I) -> Result<u64, StoreError>
+    where
+        I: Into<String>;
+}
+
+/// A store engine for planes.
+pub struct SwimPlaneStore<D> {
+    /// The name of the plane.
+    plane_name: Text,
+    /// Delegate byte engine.
+    delegate: Arc<D>,
+    keystore: KeyStore<D>,
+}
+
+impl<D> Clone for SwimPlaneStore<D> {
+    fn clone(&self) -> Self {
+        SwimPlaneStore {
+            plane_name: self.plane_name.clone(),
+            delegate: self.delegate.clone(),
+            keystore: self.keystore.clone(),
+        }
+    }
+}
+
+impl<D> Debug for SwimPlaneStore<D> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SwimPlaneStore")
+            .field("plane_name", &self.plane_name)
+            .finish()
+    }
+}
+
+fn exec_keyspace<F, O>(key: StoreKey, f: F) -> Result<O, StoreError>
+where
+    F: Fn(KeyspaceName, Vec<u8>) -> Result<O, StoreError>,
+{
+    match key {
+        s @ StoreKey::Map { .. } => f(KeyspaceName::Map, serialize(&s)?),
+        s @ StoreKey::Value { .. } => f(KeyspaceName::Value, serialize(&s)?),
+    }
+}
+
+impl<D> PlaneStore for SwimPlaneStore<D>
+where
+    D: Store,
+{
+    type NodeStore = SwimNodeStore<Self>;
+
+    fn node_store<I>(&self, node: I) -> Self::NodeStore
+    where
+        I: Into<Text>,
+    {
+        SwimNodeStore::new(self.clone(), node)
+    }
+
+    fn get_prefix_range<F, K, V>(
+        &self,
+        prefix: StoreKey,
+        map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+    {
+        let namespace = match &prefix {
+            StoreKey::Map { .. } => KeyspaceName::Map,
+            StoreKey::Value { .. } => KeyspaceName::Value,
+        };
+
+        self.delegate
+            .get_prefix_range(namespace, serialize(&prefix)?.as_slice(), map_fn)
+    }
+
+    fn engine_info(&self) -> EngineInfo {
+        self.delegate.engine_info()
+    }
+
+    fn node_id_of<I>(&self, lane: I) -> Result<u64, StoreError>
+    where
+        I: Into<String>,
+    {
+        self.keystore.id_for(lane.into())
+    }
+}
+
+impl<D: Store> StoreEngine for SwimPlaneStore<D> {
+    fn put(&self, key: StoreKey, value: &[u8]) -> Result<(), StoreError> {
+        exec_keyspace(key, |namespace, bytes| {
+            self.delegate
+                .put_keyspace(namespace, bytes.as_slice(), value)
+        })
+    }
+
+    fn get(&self, key: StoreKey) -> Result<Option<Vec<u8>>, StoreError> {
+        exec_keyspace(key, |namespace, bytes| {
+            self.delegate.get_keyspace(namespace, bytes.as_slice())
+        })
+    }
+
+    fn delete(&self, key: StoreKey) -> Result<(), StoreError> {
+        exec_keyspace(key, |namespace, bytes| {
+            self.delegate.delete_keyspace(namespace, bytes.as_slice())
+        })
+    }
+}
+
+pub(crate) fn open_plane<B, P, D>(
+    base_path: B,
+    plane_name: P,
+    builder: D,
+    keyspaces: Keyspaces<D>,
+) -> Result<SwimPlaneStore<D::Store>, StoreError>
+where
+    B: AsRef<Path>,
+    P: AsRef<Path>,
+    D: StoreBuilder,
+{
+    let path = path_for(base_path.as_ref(), plane_name.as_ref());
+    let delegate = builder.build(path, &keyspaces)?;
+    let plane_name = match plane_name.as_ref().to_str() {
+        Some(path) => path.to_string(),
+        None => {
+            return Err(StoreError::Io(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "Expected a valid UTF-8 path",
+            )));
+        }
+    };
+
+    let arcd_delegate = Arc::new(delegate);
+    let keystore = KeyStore::initialise_with(arcd_delegate.clone());
+
+    Ok(SwimPlaneStore::new(plane_name, arcd_delegate, keystore))
+}
+
+impl<D> SwimPlaneStore<D>
+where
+    D: Store,
+{
+    pub(crate) fn new<I: Into<Text>>(
+        plane_name: I,
+        delegate: Arc<D>,
+        keystore: KeyStore<D>,
+    ) -> SwimPlaneStore<D> {
+        SwimPlaneStore {
+            plane_name: plane_name.into(),
+            delegate,
+            keystore,
+        }
+    }
+}
diff --git a/swim_server/src/store/keystore.rs b/swim_server/src/store/keystore.rs
new file mode 100644
index 000000000..16330ff7c
--- /dev/null
+++ b/swim_server/src/store/keystore.rs
@@ -0,0 +1,195 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::store::KeyspaceName;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use store::keyspaces::KeyspaceByteEngine;
+use store::{deserialize, serialize, MergeOperands, StoreError};
+use tokio::sync::oneshot;
+
+pub type KeyRequest = (String, oneshot::Sender<u64>);
+
+/// The lane keyspace's counter key.
+pub const COUNTER_KEY: &str = "counter";
+pub const COUNTER_BYTES: &[u8] = COUNTER_KEY.as_bytes();
+const DESERIALIZATION_FAILURE: &str = "Failed to deserialize key";
+const SERIALIZATION_FAILURE: &str = "Failed to serialize key";
+pub const INCONSISTENT_KEYSPACE: &str = "Inconsistent keyspace";
+const INIT_FAILURE: &str = "Failed to initialise keystore";
+
+/// The prefix that all lane identifiers in the counter keyspace will be prefixed by.
+pub const LANE_PREFIX: &str = "lane";
+
+/// The initial value that the lane identifier keyspace will be initialised with if it doesn't
+/// already exist.
+pub const INITIAL: u64 = 0;
+pub const STEP: u64 = 1;
+
+#[derive(Debug)]
+pub struct KeyStore<D> {
+    delegate: Arc<D>,
+    count: Arc<AtomicU64>,
+}
+
+impl<D> Clone for KeyStore<D> {
+    fn clone(&self) -> Self {
+        KeyStore {
+            delegate: self.delegate.clone(),
+            count: self.count.clone(),
+        }
+    }
+}
+
+impl<D: KeyspaceByteEngine> KeyStore<D> {
+    pub fn initialise_with(delegate: Arc<D>) -> KeyStore<D> {
+        let count = match delegate.get_keyspace(KeyspaceName::Lane, COUNTER_BYTES) {
+            Ok(Some(counter)) => deserialize::<u64>(counter.as_slice()).expect(INIT_FAILURE),
+            Ok(None) => INITIAL,
+            Err(e) => {
+                panic!("{}: `{:?}`", INIT_FAILURE, e)
+            }
+        };
+
+        KeyStore {
+            delegate,
+            count: Arc::new(AtomicU64::new(count)),
+        }
+    }
+
+    pub fn id_for(&self, lane_id: String) -> Result<u64, StoreError> {
+        let KeyStore { delegate, count } = self;
+        let prefixed = format_key(lane_id);
+
+        match delegate.get_keyspace(KeyspaceName::Lane, prefixed.as_bytes())? {
+            Some(bytes) => deserialize_key(bytes),
+            None => {
+                let id = count.fetch_add(STEP, Ordering::Acquire) + 1;
+                delegate.merge_keyspace(KeyspaceName::Lane, COUNTER_BYTES, STEP)?;
+                delegate.put_keyspace(
+                    KeyspaceName::Lane,
+                    prefixed.as_bytes(),
+                    serialize(&id)?.as_slice(),
+                )?;
+
+                Ok(id)
+            }
+        }
+    }
+}
+
+fn deserialize_key<B: AsRef<[u8]>>(bytes: B) -> Result<u64, StoreError> {
+    bincode::deserialize::<u64>(bytes.as_ref()).map_err(|e| StoreError::Decoding(e.to_string()))
+}
+
+pub fn format_key<I: ToString>(uri: I) -> String {
+    format!("{}/{}", LANE_PREFIX, uri.to_string())
+}
+
+#[allow(clippy::unnecessary_wraps)]
+pub(crate) fn incrementing_merge_operator(
+    _new_key: &[u8],
+    existing_value: Option<&[u8]>,
+    operands: &mut MergeOperands,
+) -> Option<Vec<u8>> {
+    let mut value = match existing_value {
+        Some(bytes) => deserialize_key(bytes).expect(DESERIALIZATION_FAILURE),
+        None => INITIAL,
+    };
+
+    for op in operands {
+        let deserialized = deserialize_key(op).expect(DESERIALIZATION_FAILURE);
+        value += deserialized;
+    }
+    Some(serialize(&value).expect(SERIALIZATION_FAILURE))
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::store::keystore::{
+        deserialize_key, format_key, KeyStore, KeyspaceName, COUNTER_BYTES, COUNTER_KEY,
+        INCONSISTENT_KEYSPACE,
+    };
+    use crate::store::mock::MockStore;
+    use std::sync::Arc;
+    use store::deserialize;
+    use store::keyspaces::{Keyspace, KeyspaceByteEngine};
+
+    fn keyspaces() -> Vec<String> {
+        vec![
+            KeyspaceName::Lane.name().to_string(),
+            KeyspaceName::Value.name().to_string(),
+            KeyspaceName::Map.name().to_string(),
+        ]
+    }
+
+    #[test]
+    fn lane_id() {
+        let delegate = Arc::new(MockStore::with_keyspaces(keyspaces()));
+        let store = KeyStore::initialise_with(delegate.clone());
+
+        let lane_uri = "A";
+
+        assert_eq!(store.id_for(lane_uri.to_string()), Ok(1));
+        assert_eq!(store.id_for(lane_uri.to_string()), Ok(1));
+
+        let opt = delegate
+            .get_keyspace(KeyspaceName::Lane, COUNTER_BYTES)
+            .unwrap()
+            .expect(INCONSISTENT_KEYSPACE);
+
+        assert_eq!(deserialize_key(opt).unwrap(), 1);
+        assert_counters(&delegate, lane_uri, 1, 1);
+    }
+
+    #[test]
+    fn multiple_lane_ids() {
+        let delegate = Arc::new(MockStore::with_keyspaces(keyspaces()));
+        let store = KeyStore::initialise_with(delegate.clone());
+
+        let lane_prefix = "lane";
+        let mut lane_count = 0;
+
+        for lane_id in 1..=10 {
+            lane_count += 1;
+            let lane_uri = format!("{}/{}", lane_prefix, lane_id);
+            assert_eq!(store.id_for(lane_uri.to_string()), Ok(lane_count));
+
+            assert_counters(&delegate, lane_uri.as_str(), lane_count, lane_count);
+        }
+    }
+
+    fn assert_counters(
+        store: &Arc<MockStore>,
+        lane_uri: &str,
+        lane_count_at: u64,
+        counter_at: u64,
+    ) {
+        let key = format_key(lane_uri.to_string());
+        let lane_id = store
+            .get_keyspace(KeyspaceName::Lane, key.as_bytes())
+            .unwrap()
+            .expect("Missing key");
+        assert_eq!(
+            deserialize::<u64>(lane_id.as_slice()).unwrap(),
+            lane_count_at
+        );
+
+        let counter = store
+            .get_keyspace(KeyspaceName::Lane, COUNTER_KEY.as_bytes())
+            .unwrap()
+            .expect("Missing counter");
+        assert_eq!(deserialize::<u64>(counter.as_slice()).unwrap(), counter_at);
+    }
+}
diff --git a/swim_server/src/store/mock.rs b/swim_server/src/store/mock.rs
new file mode 100644
index 000000000..24f3a234c
--- /dev/null
+++ b/swim_server/src/store/mock.rs
@@ -0,0 +1,116 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::hash_map::Entry;
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+use crate::store::keystore::STEP;
+use store::keyspaces::{Keyspace, KeyspaceByteEngine};
+use store::{deserialize, serialize, StoreError};
+
+pub struct MockStore {
+    values: Mutex<HashMap<String, HashMap<Vec<u8>, Vec<u8>>>>,
+}
+
+impl MockStore {
+    pub fn with_keyspaces(keyspaces: Vec<String>) -> MockStore {
+        let keyspaces = keyspaces.into_iter().fold(HashMap::new(), |mut map, name| {
+            map.insert(name, HashMap::new());
+            map
+        });
+
+        MockStore {
+            values: Mutex::new(keyspaces),
+        }
+    }
+}
+
+impl KeyspaceByteEngine for MockStore {
+    fn put_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        value: &[u8],
+    ) -> Result<(), StoreError> {
+        let mut guard = self.values.lock().unwrap();
+        let keyspace = guard
+            .get_mut(keyspace.name())
+            .ok_or(StoreError::KeyspaceNotFound)?;
+        keyspace.insert(key.to_vec(), value.to_vec());
+
+        Ok(())
+    }
+
+    fn get_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StoreError> {
+        let mut guard = self.values.lock().unwrap();
+        let keyspace = guard
+            .get_mut(keyspace.name())
+            .ok_or(StoreError::KeyspaceNotFound)?;
+
+        Ok(keyspace.get(key).cloned())
+    }
+
+    fn delete_keyspace<K: Keyspace>(&self, keyspace: K, key: &[u8]) -> Result<(), StoreError> {
+        let mut guard = self.values.lock().unwrap();
+        let keyspace = guard
+            .get_mut(keyspace.name())
+            .ok_or(StoreError::KeyspaceNotFound)?;
+        keyspace.remove(key);
+
+        Ok(())
+    }
+
+    fn merge_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        step: u64,
+    ) -> Result<(), StoreError> {
+        let mut guard = self.values.lock().unwrap();
+        let keyspace = guard
+            .get_mut(keyspace.name())
+            .ok_or(StoreError::KeyspaceNotFound)?;
+
+        match keyspace.entry(key.to_vec()) {
+            Entry::Occupied(mut entry) => {
+                let mut value = deserialize::<u64>(entry.get()).unwrap();
+                value += step;
+                *entry.get_mut() = serialize(&value).unwrap();
+                Ok(())
+            }
+            Entry::Vacant(entry) => {
+                entry.insert(serialize(&STEP).unwrap());
+                Ok(())
+            }
+        }
+    }
+
+    fn get_prefix_range<F, K, V, S>(
+        &self,
+        _keyspace: S,
+        _prefix: &[u8],
+        _map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+        S: Keyspace,
+    {
+        todo!()
+    }
+}
diff --git a/swim_server/src/store/mod.rs b/swim_server/src/store/mod.rs
new file mode 100644
index 000000000..e159dd610
--- /dev/null
+++ b/swim_server/src/store/mod.rs
@@ -0,0 +1,212 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+pub mod mock;
+
+pub mod keystore;
+mod nostore;
+mod rocks;
+
+pub use rocks::{default_db_opts, default_keyspaces, RocksDatabase, RocksOpts};
+use std::fmt::{Debug, Formatter};
+use std::io;
+use std::path::PathBuf;
+use store::engines::StoreBuilder;
+use store::StoreError;
+
+use serde::{Deserialize, Serialize};
+
+use store::keyspaces::{Keyspace, Keyspaces};
+use utilities::fs::Dir;
+
+use crate::plane::store::{open_plane, PlaneStore, SwimPlaneStore};
+
+/// Unique lane identifier keyspace. The name is `default` as either the Rust RocksDB crate or
+/// Rocks DB itself has an issue in using merge operators under a non-default column family.
+///
+/// See: https://github.com/rust-rocksdb/rust-rocksdb/issues/29
+pub(crate) const LANE_KS: &str = "default";
+/// Value lane store keyspace.
+pub(crate) const VALUE_LANE_KS: &str = "value_lanes";
+/// Map lane store keyspace.
+pub(crate) const MAP_LANE_KS: &str = "map_lanes";
+
+/// An enumeration over the keyspaces that exist in a store.
+#[derive(Debug, Copy, Clone, PartialEq)]
+pub enum KeyspaceName {
+    Lane,
+    Value,
+    Map,
+}
+
+impl Keyspace for KeyspaceName {
+    fn name(&self) -> &str {
+        match self {
+            KeyspaceName::Lane => LANE_KS,
+            KeyspaceName::Value => VALUE_LANE_KS,
+            KeyspaceName::Map => MAP_LANE_KS,
+        }
+    }
+}
+
+/// A Swim server store which will create plane stores on demand.
+///
+/// When a new plane store is requested, then the implementor is expected to either load the plane
+/// from a delegate database or create a new database on demand.
+pub trait SwimStore {
+    /// The type of plane stores that are created.
+    type PlaneStore: PlaneStore;
+
+    /// Create a plane store with `plane_name`.
+    ///
+    /// # Errors
+    /// Errors if the delegate database could not be created.
+    fn plane_store<I>(&mut self, plane_name: I) -> Result<Self::PlaneStore, StoreError>
+    where
+        I: ToString;
+}
+
+/// A Swim server store that will open plane stores on request.
+pub struct ServerStore<D>
+where
+    D: StoreBuilder,
+{
+    /// The directory that this store is operating from.
+    dir: Dir,
+    /// Database environment open options
+    builder: D,
+    /// The keyspaces that all stores will be opened with.
+    keyspaces: Keyspaces<D>,
+}
+
+impl<D> Debug for ServerStore<D>
+where
+    D: StoreBuilder,
+{
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ServerStore")
+            .field("directory", &self.dir.path())
+            .finish()
+    }
+}
+
+impl<D> ServerStore<D>
+where
+    D: StoreBuilder,
+{
+    /// Constructs a new server store that will open stores using `opts` and will use the directory
+    /// `base_path` for opening all new stores.
+    ///
+    /// # Panics
+    /// Panics if the directory cannot be created.
+    pub fn new(
+        builder: D,
+        keyspaces: Keyspaces<D>,
+        base_path: PathBuf,
+    ) -> io::Result<ServerStore<D>> {
+        Ok(ServerStore {
+            dir: Dir::persistent(base_path)?,
+            builder,
+            keyspaces,
+        })
+    }
+
+    /// Constructs a new transient server store that will clear the directory (prefixed by `prefix`)
+    /// when dropped and open stores using `db_opts`.
+    ///
+    /// # Panics
+    /// Panics if the directory cannot be created.
+    pub fn transient(
+        builder: D,
+        keyspaces: Keyspaces<D>,
+        prefix: &str,
+    ) -> io::Result<ServerStore<D>> {
+        Ok(ServerStore {
+            dir: Dir::transient(prefix)?,
+            builder,
+            keyspaces,
+        })
+    }
+}
+
+impl ServerStore<RocksOpts> {
+    pub fn transient_default(prefix: &str) -> io::Result<ServerStore<RocksOpts>> {
+        let db_opts = default_db_opts();
+        let keyspaces = default_keyspaces();
+
+        Self::transient(db_opts, keyspaces, prefix)
+    }
+}
+
+impl<D> SwimStore for ServerStore<D>
+where
+    D: StoreBuilder,
+{
+    type PlaneStore = SwimPlaneStore<D::Store>;
+
+    fn plane_store<I: ToString>(&mut self, plane_name: I) -> Result<Self::PlaneStore, StoreError> {
+        let ServerStore {
+            builder,
+            keyspaces,
+            dir,
+            ..
+        } = self;
+        let plane_name = plane_name.to_string();
+
+        open_plane(dir.path(), &plane_name, builder.clone(), keyspaces.clone())
+    }
+}
+
+/// A lane key that is either a map lane key or a value lane key.
+#[derive(Serialize, Deserialize, Clone, Debug, PartialOrd, PartialEq)]
+pub enum StoreKey {
+    /// A map lane key.
+    ///
+    /// Within plane stores, map lane keys are defined in the format of `/lane_id/key` where `key`
+    /// is the key of a lane's map data structure.
+    Map {
+        /// The lane ID.
+        lane_id: u64,
+        /// An optional, serialized, key. This is optional as ranged snapshots to not require the
+        /// key.
+        #[serde(skip_serializing_if = "Option::is_none")]
+        key: Option<Vec<u8>>,
+    },
+    /// A value lane key.
+    Value {
+        /// The lane ID.
+        lane_id: u64,
+    },
+}
+
+impl StoreKey {
+    pub fn keyspace_name(&self) -> KeyspaceName {
+        match self {
+            StoreKey::Map { .. } => KeyspaceName::Map,
+            StoreKey::Value { .. } => KeyspaceName::Value,
+        }
+    }
+}
+
+pub trait StoreEngine {
+    /// Put a key-value pair into the delegate store.
+    fn put(&self, key: StoreKey, value: &[u8]) -> Result<(), StoreError>;
+
+    /// Get a value keyed by a store key from the delegate store.
+    fn get(&self, key: StoreKey) -> Result<Option<Vec<u8>>, StoreError>;
+
+    /// Delete a key-value pair by its store key from the delegate store.
+    fn delete(&self, key: StoreKey) -> Result<(), StoreError>;
+}
diff --git a/swim_server/src/store/nostore.rs b/swim_server/src/store/nostore.rs
new file mode 100644
index 000000000..c973b2e66
--- /dev/null
+++ b/swim_server/src/store/nostore.rs
@@ -0,0 +1,164 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::path::{Path, PathBuf};
+use store::engines::StoreBuilder;
+use store::iterator::{
+    EngineIterOpts, EngineIterator, EnginePrefixIterator, EngineRefIterator, IteratorKey,
+};
+use store::keyspaces::{Keyspace, KeyspaceByteEngine, KeyspaceResolver, Keyspaces};
+use store::{EngineInfo, KvBytes, Store, StoreError};
+
+/// A store which will persist no data and exists purely to uphold the minimum contract required
+/// between a lane and its store.
+#[derive(Debug, Clone)]
+pub struct NoStore {
+    path: PathBuf,
+    ks: (),
+}
+
+impl Store for NoStore {
+    fn path(&self) -> &Path {
+        &self.path
+    }
+
+    fn engine_info(&self) -> EngineInfo {
+        EngineInfo {
+            path: "no store".to_string(),
+            kind: "no store".to_string(),
+        }
+    }
+}
+
+pub struct NoStoreIterator;
+impl EngineIterator for NoStoreIterator {
+    fn seek_to(&mut self, _key: IteratorKey) -> Result<bool, StoreError> {
+        Ok(false)
+    }
+
+    fn seek_next(&mut self) {}
+
+    fn key(&self) -> Option<&[u8]> {
+        None
+    }
+
+    fn value(&self) -> Option<&[u8]> {
+        None
+    }
+
+    fn valid(&self) -> Result<bool, StoreError> {
+        Ok(false)
+    }
+}
+
+impl EnginePrefixIterator for NoStoreIterator {
+    fn next(&mut self) -> Option<KvBytes> {
+        None
+    }
+
+    fn valid(&self) -> Result<bool, StoreError> {
+        Ok(false)
+    }
+}
+
+impl<'a: 'b, 'b> EngineRefIterator<'a, 'b> for NoStore {
+    type EngineIterator = NoStoreIterator;
+    type EnginePrefixIterator = NoStoreIterator;
+
+    fn iterator_opt(
+        &'a self,
+        _space: &'b Self::ResolvedKeyspace,
+        _opts: EngineIterOpts,
+    ) -> Result<Self::EngineIterator, StoreError> {
+        Ok(NoStoreIterator)
+    }
+
+    fn prefix_iterator_opt(
+        &'a self,
+        _space: &'b Self::ResolvedKeyspace,
+        _opts: EngineIterOpts,
+        _prefix: &'b [u8],
+    ) -> Result<Self::EnginePrefixIterator, StoreError> {
+        Ok(NoStoreIterator)
+    }
+}
+
+#[derive(Default, Clone)]
+pub struct NoStoreOpts;
+impl StoreBuilder for NoStoreOpts {
+    type Store = NoStore;
+
+    fn build<I>(self, _path: I, _keyspaces: &Keyspaces<Self>) -> Result<Self::Store, StoreError>
+    where
+        I: AsRef<Path>,
+    {
+        Ok(NoStore {
+            path: PathBuf::from("Transient".to_string()),
+            ks: (),
+        })
+    }
+}
+
+impl KeyspaceByteEngine for NoStore {
+    fn put_keyspace<K: Keyspace>(
+        &self,
+        _keyspace: K,
+        _key: &[u8],
+        _value: &[u8],
+    ) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn get_keyspace<K: Keyspace>(
+        &self,
+        _keyspace: K,
+        _key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StoreError> {
+        Ok(None)
+    }
+
+    fn delete_keyspace<K: Keyspace>(&self, _keyspace: K, _key: &[u8]) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn merge_keyspace<K: Keyspace>(
+        &self,
+        _keyspace: K,
+        _key: &[u8],
+        _step: u64,
+    ) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn get_prefix_range<F, K, V, S>(
+        &self,
+        _keyspace: S,
+        _prefix: &[u8],
+        _map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+        S: Keyspace,
+    {
+        Ok(None)
+    }
+}
+
+impl KeyspaceResolver for NoStore {
+    type ResolvedKeyspace = ();
+
+    fn resolve_keyspace<K: Keyspace>(&self, _space: &K) -> Option<&Self::ResolvedKeyspace> {
+        Some(&self.ks)
+    }
+}
diff --git a/swim_server/src/store/rocks.rs b/swim_server/src/store/rocks.rs
new file mode 100644
index 000000000..3fbe781e0
--- /dev/null
+++ b/swim_server/src/store/rocks.rs
@@ -0,0 +1,179 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::store::keystore::{incrementing_merge_operator, COUNTER_KEY};
+use crate::store::{LANE_KS, MAP_LANE_KS, VALUE_LANE_KS};
+use rocksdb::{ColumnFamilyDescriptor, DB};
+use std::mem::size_of;
+use std::path::Path;
+use store::engines::{RocksEngine, RocksIterator, RocksPrefixIterator, StoreBuilder};
+use store::iterator::{EngineIterOpts, EngineRefIterator};
+use store::keyspaces::{Keyspace, KeyspaceByteEngine};
+use store::keyspaces::{KeyspaceDef, KeyspaceResolver, Keyspaces};
+use store::{ColumnFamily, Options, SliceTransform};
+use store::{EngineInfo, Store, StoreError};
+
+const PREFIX_BLOOM_RATIO: f64 = 0.2;
+
+#[derive(Debug, Clone)]
+pub struct RocksDatabase {
+    db: RocksEngine,
+}
+
+impl Store for RocksDatabase {
+    fn path(&self) -> &Path {
+        self.db.path()
+    }
+
+    fn engine_info(&self) -> EngineInfo {
+        self.db.engine_info()
+    }
+}
+
+impl KeyspaceByteEngine for RocksDatabase {
+    fn put_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        value: &[u8],
+    ) -> Result<(), StoreError> {
+        self.db.put_keyspace(keyspace, key, value)
+    }
+
+    fn get_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StoreError> {
+        self.db.get_keyspace(keyspace, key)
+    }
+
+    fn delete_keyspace<K: Keyspace>(&self, keyspace: K, key: &[u8]) -> Result<(), StoreError> {
+        self.db.delete_keyspace(keyspace, key)
+    }
+
+    fn merge_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        step: u64,
+    ) -> Result<(), StoreError> {
+        self.db.merge_keyspace(keyspace, key, step)
+    }
+
+    fn get_prefix_range<F, K, V, S>(
+        &self,
+        keyspace: S,
+        prefix: &[u8],
+        map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+        S: Keyspace,
+    {
+        self.db.get_prefix_range(keyspace, prefix, map_fn)
+    }
+}
+
+impl<'a: 'b, 'b> EngineRefIterator<'a, 'b> for RocksDatabase {
+    type EngineIterator = RocksIterator<'b>;
+    type EnginePrefixIterator = RocksPrefixIterator<'b>;
+
+    fn iterator_opt(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+        opts: EngineIterOpts,
+    ) -> Result<Self::EngineIterator, StoreError> {
+        self.db.iterator_opt(space, opts)
+    }
+
+    fn prefix_iterator_opt(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+        opts: EngineIterOpts,
+        prefix: &'b [u8],
+    ) -> Result<Self::EnginePrefixIterator, StoreError> {
+        self.db.prefix_iterator_opt(space, opts, prefix)
+    }
+}
+
+impl KeyspaceResolver for RocksDatabase {
+    type ResolvedKeyspace = ColumnFamily;
+
+    fn resolve_keyspace<K: Keyspace>(&self, space: &K) -> Option<&Self::ResolvedKeyspace> {
+        self.db.resolve_keyspace(space)
+    }
+}
+
+#[derive(Clone)]
+pub struct RocksOpts(Options);
+
+impl Default for RocksOpts {
+    fn default() -> Self {
+        let mut rock_opts = rocksdb::Options::default();
+        rock_opts.create_if_missing(true);
+        rock_opts.create_missing_column_families(true);
+
+        RocksOpts(rock_opts)
+    }
+}
+
+impl StoreBuilder for RocksOpts {
+    type Store = RocksDatabase;
+
+    fn build<I>(self, path: I, keyspaces: &Keyspaces<Self>) -> Result<Self::Store, StoreError>
+    where
+        I: AsRef<Path>,
+    {
+        let Keyspaces { keyspaces } = keyspaces;
+        let descriptors =
+            keyspaces
+                .iter()
+                .fold(Vec::with_capacity(keyspaces.len()), |mut vec, def| {
+                    let cf_descriptor =
+                        ColumnFamilyDescriptor::new(def.name.to_string(), def.opts.0.clone());
+                    vec.push(cf_descriptor);
+                    vec
+                });
+
+        let db = DB::open_cf_descriptors(&self.0, path, descriptors)?;
+        Ok(RocksDatabase {
+            db: RocksEngine::new(db),
+        })
+    }
+}
+
+pub fn default_keyspaces() -> Keyspaces<RocksOpts> {
+    let mut lane_counter_opts = Options::default();
+    lane_counter_opts.set_merge_operator_associative(COUNTER_KEY, incrementing_merge_operator);
+
+    let lane_def = KeyspaceDef::new(LANE_KS, RocksOpts(lane_counter_opts));
+    let value_def = KeyspaceDef::new(VALUE_LANE_KS, RocksOpts(Options::default()));
+
+    let mut map_opts = Options::default();
+    map_opts.set_prefix_extractor(SliceTransform::create_fixed_prefix(size_of::<u64>()));
+    map_opts.set_memtable_prefix_bloom_ratio(PREFIX_BLOOM_RATIO);
+
+    let map_def = KeyspaceDef::new(MAP_LANE_KS, RocksOpts(map_opts));
+
+    Keyspaces::new(vec![lane_def, value_def, map_def])
+}
+
+pub fn default_db_opts() -> RocksOpts {
+    let mut rock_opts = Options::default();
+    rock_opts.create_if_missing(true);
+    rock_opts.create_missing_column_families(true);
+
+    RocksOpts(rock_opts)
+}
diff --git a/swim_server/store/Cargo.toml b/swim_server/store/Cargo.toml
new file mode 100644
index 000000000..eee69c89e
--- /dev/null
+++ b/swim_server/store/Cargo.toml
@@ -0,0 +1,12 @@
+[package]
+name = "store"
+version = "0.1.0"
+authors = ["Swim.AI developers info@swim.ai"]
+edition = "2018"
+
+[dependencies]
+rocksdb = "0.16.0"
+bincode = "1.3.2"
+serde = { version = "1.0.124", features = ["derive", "rc"] }
+tempdir = "0.3.7"
+thiserror = "1.0"
diff --git a/swim_server/store/src/engines/mod.rs b/swim_server/store/src/engines/mod.rs
new file mode 100644
index 000000000..fd9bfaeb2
--- /dev/null
+++ b/swim_server/store/src/engines/mod.rs
@@ -0,0 +1,44 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::path::Path;
+
+pub use nostore::{NoStore, NoStoreOpts};
+pub use rocks::{RocksEngine, RocksIterator, RocksOpts, RocksPrefixIterator};
+
+use crate::keyspaces::Keyspaces;
+use crate::{Store, StoreError};
+
+mod nostore;
+mod rocks;
+
+/// A storage engine for server stores that handles byte arrays.
+pub trait ByteEngine: 'static {
+    /// Put a key-value pair into this store.
+    fn put(&self, key: &[u8], value: &[u8]) -> Result<(), StoreError>;
+
+    /// Get an entry from this store by its key.
+    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, StoreError>;
+
+    /// Delete a value from this store by its key.
+    fn delete(&self, key: &[u8]) -> Result<(), StoreError>;
+}
+
+pub trait StoreBuilder: Sized + Clone {
+    type Store: Store;
+
+    fn build<I>(self, path: I, keyspaces: &Keyspaces<Self>) -> Result<Self::Store, StoreError>
+    where
+        I: AsRef<Path>;
+}
diff --git a/swim_server/store/src/engines/nostore.rs b/swim_server/store/src/engines/nostore.rs
new file mode 100644
index 000000000..5b9161661
--- /dev/null
+++ b/swim_server/store/src/engines/nostore.rs
@@ -0,0 +1,182 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::engines::ByteEngine;
+use crate::engines::StoreBuilder;
+use crate::iterator::{
+    EngineIterOpts, EngineIterator, EnginePrefixIterator, EngineRefIterator, IteratorKey,
+};
+use crate::keyspaces::{Keyspace, KeyspaceByteEngine, KeyspaceResolver, Keyspaces};
+use crate::EngineInfo;
+use crate::{KvBytes, Store, StoreError};
+use std::borrow::Borrow;
+use std::path::{Path, PathBuf};
+
+/// A delegate store database that does nothing.
+#[derive(Debug, Clone)]
+pub struct NoStore {
+    path: PathBuf,
+}
+
+impl Store for NoStore {
+    fn path(&self) -> &Path {
+        self.path.borrow()
+    }
+
+    fn engine_info(&self) -> EngineInfo {
+        EngineInfo {
+            path: "Transient".to_string(),
+            kind: "NoStore".to_string(),
+        }
+    }
+}
+
+impl KeyspaceResolver for NoStore {
+    type ResolvedKeyspace = ();
+
+    fn resolve_keyspace<K: Keyspace>(&self, _space: &K) -> Option<&Self::ResolvedKeyspace> {
+        None
+    }
+}
+
+impl KeyspaceByteEngine for NoStore {
+    fn put_keyspace<K: Keyspace>(
+        &self,
+        _keyspace: K,
+        _key: &[u8],
+        _value: &[u8],
+    ) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn get_keyspace<K: Keyspace>(
+        &self,
+        _keyspace: K,
+        _key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StoreError> {
+        Ok(None)
+    }
+
+    fn delete_keyspace<K: Keyspace>(&self, _keyspace: K, _key: &[u8]) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn merge_keyspace<K: Keyspace>(
+        &self,
+        _keyspace: K,
+        _key: &[u8],
+        _value: u64,
+    ) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    fn get_prefix_range<F, K, V, S>(
+        &self,
+        _keyspace: S,
+        _prefix: &[u8],
+        _map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+        S: Keyspace,
+    {
+        Ok(None)
+    }
+}
+
+#[derive(Default, Clone)]
+pub struct NoStoreOpts;
+impl StoreBuilder for NoStoreOpts {
+    type Store = NoStore;
+
+    fn build<I>(self, _path: I, _keyspaces: &Keyspaces<Self>) -> Result<Self::Store, StoreError>
+    where
+        I: AsRef<Path>,
+    {
+        Ok(NoStore {
+            path: PathBuf::from("Transient".to_string()),
+        })
+    }
+}
+
+impl ByteEngine for NoStore {
+    /// Put operation does nothing and that always succeeds.
+    fn put(&self, _key: &[u8], _value: &[u8]) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    /// Get operation does nothing and always returns `Ok(None)`.
+    fn get(&self, _key: &[u8]) -> Result<Option<Vec<u8>>, StoreError> {
+        Ok(None)
+    }
+
+    /// Delete operation does nothing and always returns `Ok(())`.
+    fn delete(&self, _key: &[u8]) -> Result<(), StoreError> {
+        Ok(())
+    }
+}
+
+pub struct NoStoreEngineIterator;
+impl EngineIterator for NoStoreEngineIterator {
+    fn seek_to(&mut self, _key: IteratorKey) -> Result<bool, StoreError> {
+        Ok(true)
+    }
+
+    fn seek_next(&mut self) {}
+
+    fn key(&self) -> Option<&[u8]> {
+        None
+    }
+
+    fn value(&self) -> Option<&[u8]> {
+        None
+    }
+
+    fn valid(&self) -> Result<bool, StoreError> {
+        Ok(true)
+    }
+}
+
+pub struct NoStoreEnginePrefixIterator;
+impl EnginePrefixIterator for NoStoreEnginePrefixIterator {
+    fn next(&mut self) -> Option<KvBytes> {
+        None
+    }
+
+    fn valid(&self) -> Result<bool, StoreError> {
+        Ok(false)
+    }
+}
+
+impl<'a: 'b, 'b> EngineRefIterator<'a, 'b> for NoStore {
+    type EngineIterator = NoStoreEngineIterator;
+    type EnginePrefixIterator = NoStoreEnginePrefixIterator;
+
+    fn iterator_opt(
+        &'a self,
+        _space: &'b Self::ResolvedKeyspace,
+        _pts: EngineIterOpts,
+    ) -> Result<Self::EngineIterator, StoreError> {
+        Ok(NoStoreEngineIterator)
+    }
+
+    fn prefix_iterator_opt(
+        &'a self,
+        _space: &'b Self::ResolvedKeyspace,
+        _opts: EngineIterOpts,
+        _prefix: &'b [u8],
+    ) -> Result<Self::EnginePrefixIterator, StoreError> {
+        Ok(NoStoreEnginePrefixIterator)
+    }
+}
diff --git a/swim_server/store/src/engines/rocks/iterator.rs b/swim_server/store/src/engines/rocks/iterator.rs
new file mode 100644
index 000000000..10c71b9fd
--- /dev/null
+++ b/swim_server/store/src/engines/rocks/iterator.rs
@@ -0,0 +1,101 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::engines::RocksEngine;
+use crate::iterator::{
+    EngineIterOpts, EngineIterator, EnginePrefixIterator, EngineRefIterator, IteratorKey,
+};
+use crate::{KvBytes, StoreError};
+use rocksdb::{DBIteratorWithThreadMode, DBRawIterator, DBWithThreadMode, SingleThreaded};
+
+impl<'a: 'b, 'b> EngineRefIterator<'a, 'b> for RocksEngine {
+    type EngineIterator = RocksIterator<'b>;
+    type EnginePrefixIterator = RocksPrefixIterator<'b>;
+
+    fn iterator_opt(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+        _opts: EngineIterOpts,
+    ) -> Result<Self::EngineIterator, StoreError> {
+        let mut iter = self.delegate.raw_iterator_cf(space);
+        iter.seek_to_first();
+
+        Ok(RocksIterator { iter })
+    }
+
+    fn prefix_iterator_opt(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+        _opts: EngineIterOpts,
+        prefix: &'b [u8],
+    ) -> Result<Self::EnginePrefixIterator, StoreError> {
+        let it = self.delegate.prefix_iterator_cf(space, prefix);
+        Ok(RocksPrefixIterator { delegate: it })
+    }
+}
+
+pub struct RocksPrefixIterator<'p> {
+    delegate: DBIteratorWithThreadMode<'p, DBWithThreadMode<SingleThreaded>>,
+}
+
+impl<'d> EnginePrefixIterator for RocksPrefixIterator<'d> {
+    fn next(&mut self) -> Option<KvBytes> {
+        self.delegate.next()
+    }
+
+    fn valid(&self) -> Result<bool, StoreError> {
+        if self.delegate.valid() {
+            Ok(true)
+        } else {
+            match self.delegate.status() {
+                Ok(_) => Ok(false),
+                Err(e) => Err(StoreError::from(e)),
+            }
+        }
+    }
+}
+
+pub struct RocksIterator<'d> {
+    iter: DBRawIterator<'d>,
+}
+
+impl<'d> EngineIterator for RocksIterator<'d> {
+    fn seek_to(&mut self, key: IteratorKey) -> Result<bool, StoreError> {
+        match key {
+            IteratorKey::Start => self.iter.seek_to_first(),
+            IteratorKey::End => self.iter.seek_to_last(),
+            IteratorKey::ToKey(key) => self.iter.seek(key),
+        }
+        self.valid()
+    }
+
+    fn seek_next(&mut self) {
+        self.iter.next();
+    }
+
+    fn key(&self) -> Option<&[u8]> {
+        self.iter.key()
+    }
+
+    fn value(&self) -> Option<&[u8]> {
+        self.iter.value()
+    }
+
+    fn valid(&self) -> Result<bool, StoreError> {
+        if !self.iter.valid() {
+            self.iter.status().map_err(StoreError::from)?;
+        }
+        Ok(self.iter.valid())
+    }
+}
diff --git a/swim_server/store/src/engines/rocks/mod.rs b/swim_server/store/src/engines/rocks/mod.rs
new file mode 100644
index 000000000..3a0983103
--- /dev/null
+++ b/swim_server/store/src/engines/rocks/mod.rs
@@ -0,0 +1,197 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+mod tests;
+
+mod iterator;
+
+pub use crate::engines::rocks::iterator::{RocksIterator, RocksPrefixIterator};
+use crate::engines::StoreBuilder;
+use crate::iterator::{EnginePrefixIterator, EngineRefIterator};
+use crate::keyspaces::{Keyspace, KeyspaceByteEngine, KeyspaceResolver, Keyspaces};
+use crate::{serialize, EngineInfo, Store, StoreError};
+use rocksdb::{ColumnFamily, ColumnFamilyDescriptor};
+use rocksdb::{Error, Options, DB};
+use std::path::Path;
+use std::sync::Arc;
+
+impl From<rocksdb::Error> for StoreError {
+    fn from(e: Error) -> Self {
+        StoreError::Delegate(Box::new(e))
+    }
+}
+
+/// A Rocks database engine.
+///
+/// See <https://github.com/facebook/rocksdb/wiki> for details about the features and limitations.
+#[derive(Debug, Clone)]
+pub struct RocksEngine {
+    pub(crate) delegate: Arc<DB>,
+}
+
+impl RocksEngine {
+    pub fn new(delegate: DB) -> RocksEngine {
+        RocksEngine {
+            delegate: Arc::new(delegate),
+        }
+    }
+}
+
+impl Store for RocksEngine {
+    fn path(&self) -> &Path {
+        self.delegate.path()
+    }
+
+    fn engine_info(&self) -> EngineInfo {
+        EngineInfo {
+            path: self.path().to_string_lossy().to_string(),
+            kind: "RocksDB".to_string(),
+        }
+    }
+}
+
+impl KeyspaceResolver for RocksEngine {
+    type ResolvedKeyspace = ColumnFamily;
+
+    fn resolve_keyspace<K: Keyspace>(&self, space: &K) -> Option<&Self::ResolvedKeyspace> {
+        self.delegate.cf_handle(space.name())
+    }
+}
+
+/// Configuration wrapper for a Rocks database used by `FromOpts`.
+#[derive(Clone)]
+pub struct RocksOpts(pub Options);
+
+impl StoreBuilder for RocksOpts {
+    type Store = RocksEngine;
+
+    fn build<I>(self, path: I, keyspaces: &Keyspaces<Self>) -> Result<Self::Store, StoreError>
+    where
+        I: AsRef<Path>,
+    {
+        let Keyspaces { keyspaces } = keyspaces;
+        let descriptors =
+            keyspaces
+                .iter()
+                .fold(Vec::with_capacity(keyspaces.len()), |mut vec, def| {
+                    let cf_descriptor =
+                        ColumnFamilyDescriptor::new(def.name.to_string(), def.opts.0.clone());
+                    vec.push(cf_descriptor);
+                    vec
+                });
+
+        let db = DB::open_cf_descriptors(&self.0, path, descriptors)?;
+        Ok(RocksEngine::new(db))
+    }
+}
+
+impl Default for RocksOpts {
+    fn default() -> Self {
+        let mut rock_opts = rocksdb::Options::default();
+        rock_opts.create_if_missing(true);
+        rock_opts.create_missing_column_families(true);
+
+        RocksOpts(rock_opts)
+    }
+}
+
+fn exec_keyspace<F, O, K>(delegate: &Arc<DB>, keyspace: K, f: F) -> Result<O, StoreError>
+where
+    F: Fn(&Arc<DB>, &ColumnFamily) -> Result<O, rocksdb::Error>,
+    K: Keyspace,
+{
+    match delegate.cf_handle(keyspace.name()) {
+        Some(cf) => f(delegate, cf).map_err(Into::into),
+        None => Err(StoreError::KeyspaceNotFound),
+    }
+}
+
+impl KeyspaceByteEngine for RocksEngine {
+    fn put_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        value: &[u8],
+    ) -> Result<(), StoreError> {
+        exec_keyspace(&self.delegate, keyspace, |delegate, keyspace| {
+            delegate.put_cf(keyspace, key, value)
+        })
+    }
+
+    fn get_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StoreError> {
+        exec_keyspace(&self.delegate, keyspace, |delegate, keyspace| {
+            delegate.get_cf(keyspace, key)
+        })
+    }
+
+    fn delete_keyspace<K: Keyspace>(&self, keyspace: K, key: &[u8]) -> Result<(), StoreError> {
+        exec_keyspace(&self.delegate, keyspace, |delegate, keyspace| {
+            delegate.delete_cf(keyspace, key)
+        })
+    }
+
+    fn merge_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        value: u64,
+    ) -> Result<(), StoreError> {
+        let value = serialize(&value)?;
+        exec_keyspace(&self.delegate, keyspace, move |delegate, keyspace| {
+            delegate.merge_cf(keyspace, key, value.as_slice())
+        })
+    }
+
+    fn get_prefix_range<F, K, V, S>(
+        &self,
+        keyspace: S,
+        prefix: &[u8],
+        map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+        S: Keyspace,
+    {
+        let resolved = self
+            .resolve_keyspace(&keyspace)
+            .ok_or(StoreError::KeyspaceNotFound)?;
+        let mut it = self.prefix_iterator(resolved, prefix)?;
+        let mut data = Vec::new();
+
+        loop {
+            match it.valid() {
+                Ok(true) => match it.next() {
+                    Some((key, value)) => {
+                        let mapped = map_fn(key.as_ref(), value.as_ref())?;
+                        data.push(mapped);
+                    }
+                    None => break,
+                },
+                Ok(false) => break,
+                Err(e) => return Err(e),
+            }
+        }
+
+        if data.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(data))
+        }
+    }
+}
diff --git a/swim_server/store/src/iterator.rs b/swim_server/store/src/iterator.rs
new file mode 100644
index 000000000..ebdb67f0a
--- /dev/null
+++ b/swim_server/store/src/iterator.rs
@@ -0,0 +1,117 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::keyspaces::KeyspaceResolver;
+use crate::{KvBytes, StoreError};
+
+pub trait OwnedEngineRefIterator: for<'t> EngineRefIterator<'t, 't> {}
+impl<D> OwnedEngineRefIterator for D where D: for<'t> EngineRefIterator<'t, 't> {}
+
+/// An key representing a position that an iterator should seek to.
+pub enum IteratorKey<'p> {
+    /// Seek to the first key.
+    Start,
+    /// Seek to the last key.
+    End,
+    /// Seek to a key or the one that lexicographically follows it.
+    ToKey(&'p [u8]),
+}
+
+/// An iterator over a keyspace.
+pub trait EngineIterator {
+    /// Seek to the first key.
+    fn seek_first(&mut self) -> Result<bool, StoreError> {
+        self.seek_to(IteratorKey::Start)
+    }
+
+    /// Seek to the last key.
+    fn seek_end(&mut self) -> Result<bool, StoreError> {
+        self.seek_to(IteratorKey::End)
+    }
+
+    /// Seek to a key or the one that lexicographically follows it.
+    fn seek_to(&mut self, key: IteratorKey) -> Result<bool, StoreError>;
+
+    /// Seek to the next key that lexicographically follows the current.
+    fn seek_next(&mut self);
+
+    /// Returns the current key if the cursor points to a valid element.
+    fn key(&self) -> Option<&[u8]>;
+
+    /// Returns the current value if the cursor points to a valid element.
+    fn value(&self) -> Option<&[u8]>;
+
+    /// Returns `Ok(true)` if the iterator is currently valid and has not reached the end of its
+    /// range. `Ok(false)` if the iterator has not encountered any errors but has reached the end of
+    /// its range.  Or `Err` with a cause the error.
+    fn valid(&self) -> Result<bool, StoreError>;
+}
+
+/// An iterator over a range of keys in a keyspace that begin with a specified prefix.
+pub trait EnginePrefixIterator {
+    /// Returns a tuple containing an optional key and value for the lexicographically following
+    /// cursor. If `None` is returned, the iterator has reached the end of its range.
+    fn next(&mut self) -> Option<KvBytes>;
+
+    /// Returns `Ok(true)` if the iterator is currently valid and has not reached the end of its
+    /// range. `Ok(false)` if the iterator has not encountered any errors but has reached the end of
+    /// its range.  Or `Err` with a cause the error.
+    fn valid(&self) -> Result<bool, StoreError>;
+}
+
+// todo: add iterator direction
+#[derive(Default)]
+pub struct EngineIterOpts;
+
+/// A trait for creating iterators over a keyspace.
+pub trait EngineRefIterator<'a: 'b, 'b>: KeyspaceResolver {
+    type EngineIterator: EngineIterator;
+    type EnginePrefixIterator: EnginePrefixIterator;
+
+    /// Returns an iterator for all of the elements in the keyspace `space` with the default
+    /// iterator options.
+    fn iterator(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+    ) -> Result<Self::EngineIterator, StoreError> {
+        self.iterator_opt(space, EngineIterOpts::default())
+    }
+
+    /// Returns an iterator for all of the elements in the keyspace `space` with the provided
+    /// options.
+    fn iterator_opt(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+        opts: EngineIterOpts,
+    ) -> Result<Self::EngineIterator, StoreError>;
+
+    /// Returns an iterator for all of the elements in the keyspace `space` that have keys that are
+    /// prefixed by `prefix` and with the default iterator options.
+    fn prefix_iterator(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+        prefix: &'b [u8],
+    ) -> Result<Self::EnginePrefixIterator, StoreError> {
+        self.prefix_iterator_opt(space, EngineIterOpts::default(), prefix)
+    }
+
+    /// Returns an iterator for all of the elements in the keyspace `space` that have keys that are
+    /// prefixed by `prefix` and with the provided iterator options.
+    fn prefix_iterator_opt(
+        &'a self,
+        space: &'b Self::ResolvedKeyspace,
+        opts: EngineIterOpts,
+        prefix: &'b [u8],
+    ) -> Result<Self::EnginePrefixIterator, StoreError>;
+}
diff --git a/swim_server/store/src/keyspaces.rs b/swim_server/store/src/keyspaces.rs
new file mode 100644
index 000000000..831b16650
--- /dev/null
+++ b/swim_server/store/src/keyspaces.rs
@@ -0,0 +1,115 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::StoreError;
+
+/// A handle to a portion of logically partitioned data.
+pub trait Keyspace {
+    /// The name of the keyspace.
+    fn name(&self) -> &str;
+}
+
+/// A keyspace definition for persisting logically related data.
+///
+/// Definitions of a keyspace will depend on the underlying delegate store implementation used to
+/// run a store with. For a RocksDB engine this will correspond to a column family and for libmdbx
+/// this will correspond to a sub-database that is keyed by `name`.
+#[derive(Debug, Clone)]
+pub struct KeyspaceDef<O> {
+    /// The name of the keyspace.
+    pub name: &'static str,
+    /// The configuration options that will be used to open the keyspace.
+    pub opts: O,
+}
+
+impl<O> KeyspaceDef<O> {
+    pub fn new(name: &'static str, opts: O) -> Self {
+        KeyspaceDef { name, opts }
+    }
+}
+
+/// A list of keyspace definitions to initialise a store with.
+#[derive(Clone)]
+pub struct Keyspaces<O> {
+    pub keyspaces: Vec<KeyspaceDef<O>>,
+}
+
+impl<O> Keyspaces<O> {
+    pub fn new(keyspaces: Vec<KeyspaceDef<O>>) -> Self {
+        Keyspaces { keyspaces }
+    }
+}
+
+/// A trait for abstracting over database engines and partitioning data by a logical keyspace.
+pub trait KeyspaceByteEngine: Send + Sync + 'static {
+    /// Put a key-value pair into the specified keyspace.
+    fn put_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        value: &[u8],
+    ) -> Result<(), StoreError>;
+
+    /// Get an entry from the specified keyspace.
+    fn get_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StoreError>;
+
+    /// Delete a value from the specified keyspace.
+    fn delete_keyspace<K: Keyspace>(&self, keyspace: K, key: &[u8]) -> Result<(), StoreError>;
+
+    /// Perform a merge operation on the specified keyspace and key, incrementing by `step`.
+    fn merge_keyspace<K: Keyspace>(
+        &self,
+        keyspace: K,
+        key: &[u8],
+        step: u64,
+    ) -> Result<(), StoreError>;
+
+    /// Execute a ranged read on the store, seeking by `prefix` and deserializing results with
+    /// `map_fn`.
+    ///
+    /// Returns `Ok(None)` if no records matched `prefix` or `Ok(Some)` if matches were found.
+    ///
+    /// # Example:
+    /// Given a store engine that stores records for map lanes where the format of
+    /// `/node_uri/lane_uri/key` is used as the key. One could execute a ranged read on the store
+    /// engine with a prefix of `/node_1/lane_1/` to load all of the keys and values for that
+    /// lane.
+    ///
+    /// # Errors
+    /// Errors if an error is encountered when attempting to execute the ranged read on the
+    /// store engine or if the `map_fn` fails to deserialize a key or value.
+    fn get_prefix_range<F, K, V, S>(
+        &self,
+        _keyspace: S,
+        _prefix: &[u8],
+        _map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+        S: Keyspace;
+}
+
+/// A trait for converting an abstract keyspace name to a reference to a handle of one in a delegate
+/// engine; such as RocksDB's Column Families.
+pub trait KeyspaceResolver {
+    /// The concrete type of the keyspace.
+    type ResolvedKeyspace;
+
+    /// Resolve `space` in to a handle that can be used to make direct queries to a delegate engine.
+    fn resolve_keyspace<K: Keyspace>(&self, space: &K) -> Option<&Self::ResolvedKeyspace>;
+}
diff --git a/swim_server/store/src/lib.rs b/swim_server/store/src/lib.rs
new file mode 100644
index 000000000..dd7690ada
--- /dev/null
+++ b/swim_server/store/src/lib.rs
@@ -0,0 +1,146 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod engines;
+pub mod iterator;
+pub mod keyspaces;
+
+use std::error::Error;
+use std::fmt::Debug;
+use std::path::Path;
+use std::{error::Error as StdError, io};
+
+use thiserror::Error;
+
+use crate::iterator::OwnedEngineRefIterator;
+use crate::keyspaces::{KeyspaceByteEngine, KeyspaceResolver};
+
+pub use rocksdb::{ColumnFamily, MergeOperands, Options, SliceTransform};
+use serde::{Deserialize, Serialize};
+
+pub type KvBytes = (Box<[u8]>, Box<[u8]>);
+
+/// Store errors.
+#[derive(Debug, Error)]
+pub enum StoreError {
+    /// The provided key was not found in the store.
+    #[error("The specified key was not found")]
+    KeyNotFound,
+    /// The delegate byte engine failed to initialised.
+    #[error("The delegate store engine failed to initialise: {0}")]
+    InitialisationFailure(String),
+    /// An IO error produced by the delegate byte engine.
+    #[error("IO error: {0}")]
+    Io(io::Error),
+    /// An error produced when attempting to encode a value.
+    #[error("Encoding error: {0}")]
+    Encoding(String),
+    /// An error produced when attempting to decode a value.
+    #[error("Decoding error: {0}")]
+    Decoding(String),
+    /// A raw error produced by the delegate byte engine.
+    #[error("Delegate store error: {0}")]
+    Delegate(Box<dyn Error + Send + Sync>),
+    /// A raw error produced by the delegate byte engine that isnt send or sync
+    #[error("Delegate store error: {0}")]
+    DelegateMessage(String),
+    /// An operation was attempted on the byte engine when it was closing.
+    #[error("An operation was attempted on the delegate store engine when it was closing")]
+    Closing,
+    /// The requested keyspace was not found.
+    #[error("The requested keyspace was not found")]
+    KeyspaceNotFound,
+}
+
+impl StoreError {
+    pub fn downcast_ref<E: StdError + 'static>(&self) -> Option<&E> {
+        match self {
+            StoreError::Delegate(d) => {
+                if let Some(downcasted) = d.downcast_ref() {
+                    return Some(downcasted);
+                }
+                None
+            }
+            _ => None,
+        }
+    }
+}
+
+impl PartialEq for StoreError {
+    fn eq(&self, other: &Self) -> bool {
+        match (self, other) {
+            (StoreError::KeyNotFound, StoreError::KeyNotFound) => true,
+            (StoreError::InitialisationFailure(l), StoreError::InitialisationFailure(r)) => l.eq(r),
+            (StoreError::Io(l), StoreError::Io(r)) => l.kind().eq(&r.kind()),
+            (StoreError::Encoding(l), StoreError::Encoding(r)) => l.eq(r),
+            (StoreError::Decoding(l), StoreError::Decoding(r)) => l.eq(r),
+            (StoreError::Delegate(l), StoreError::Delegate(r)) => l.to_string().eq(&r.to_string()),
+            (StoreError::DelegateMessage(l), StoreError::DelegateMessage(r)) => l.eq(r),
+            (StoreError::Closing, StoreError::Closing) => true,
+            (StoreError::KeyspaceNotFound, StoreError::KeyspaceNotFound) => true,
+            _ => false,
+        }
+    }
+}
+
+/// A Swim server store.
+///
+/// This trait only serves to compose the multiple traits that are required for a store.
+pub trait Store:
+    KeyspaceByteEngine
+    + KeyspaceByteEngine
+    + KeyspaceResolver
+    + Send
+    + Sync
+    + Clone
+    + Debug
+    + OwnedEngineRefIterator
+    + 'static
+{
+    /// Returns a reference to the path that the delegate byte engine is operating from.
+    fn path(&self) -> &Path;
+
+    /// Returns information about the store engine.
+    fn engine_info(&self) -> EngineInfo;
+}
+
+/// Information regarding a delegate store engine that is useful for displaying along with debug
+/// information or an error report.
+#[derive(Debug, PartialEq, Clone)]
+pub struct EngineInfo {
+    /// The path that the store engine is operating from.
+    pub path: String,
+    /// The type of store engine: RocksDB/NoStore etc.
+    pub kind: String,
+}
+
+pub fn serialize_then<S, F, O, E>(engine: &E, obj: &S, f: F) -> Result<O, StoreError>
+where
+    S: Serialize,
+    F: Fn(&E, Vec<u8>) -> Result<O, StoreError>,
+{
+    f(engine, serialize(obj)?).map_err(Into::into)
+}
+
+pub fn serialize<S: Serialize>(obj: &S) -> Result<Vec<u8>, StoreError> {
+    bincode::serialize(obj).map_err(|e| StoreError::Encoding(e.to_string()))
+}
+
+pub fn deserialize<'de, D: Deserialize<'de>>(obj: &'de [u8]) -> Result<D, StoreError> {
+    bincode::deserialize(obj).map_err(|e| StoreError::Decoding(e.to_string()))
+}
+
+pub fn deserialize_key<B: AsRef<[u8]>>(bytes: B) -> Result<u64, StoreError> {
+    bincode::deserialize::<u64>(bytes.as_ref()).map_err(|e| StoreError::Decoding(e.to_string()))
+}
diff --git a/swim_server/store/src/transient.rs b/swim_server/store/src/transient.rs
new file mode 100644
index 000000000..8b1378917
--- /dev/null
+++ b/swim_server/store/src/transient.rs
@@ -0,0 +1,1 @@
+
diff --git a/utilities/Cargo.toml b/utilities/Cargo.toml
index 033a4f07f..0995d0166 100644
--- a/utilities/Cargo.toml
+++ b/utilities/Cargo.toml
@@ -22,6 +22,7 @@ swim_runtime = { path = "../swim_runtime" }
 slab = "0.4.2"
 url = "2.1.1"
 num = "0.3.0"
+tempdir = "0.3.7"
 
 [dev-dependencies]
 test_util = { path = "test_util" }
diff --git a/utilities/src/fs/mod.rs b/utilities/src/fs/mod.rs
new file mode 100644
index 000000000..6c4f9842c
--- /dev/null
+++ b/utilities/src/fs/mod.rs
@@ -0,0 +1,62 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io::ErrorKind;
+use std::path::{Path, PathBuf};
+use std::{fs, io};
+use tempdir::TempDir;
+
+/// A directory on the file system that is either persistence or transient.
+#[derive(Debug)]
+pub enum Dir {
+    /// A transient directory on the filesystem that is automatically deleted when it is dropped.
+    Transient(TempDir),
+    /// A persistent directory on the filesystem.
+    Persistent(PathBuf),
+}
+
+impl Dir {
+    /// Attempts to create a directory at the provided path. If the path already exists, then
+    /// this will return successfully.
+    ///
+    /// # Errors
+    /// Errors if the directory cannot be created.
+    pub fn persistent<I: AsRef<Path>>(path: I) -> io::Result<Dir> {
+        match fs::create_dir(&path) {
+            Ok(_) => Ok(Dir::Persistent(path.as_ref().to_path_buf())),
+            Err(e) if e.kind() == ErrorKind::AlreadyExists => {
+                Ok(Dir::Persistent(path.as_ref().to_path_buf()))
+            }
+            Err(e) => Err(e),
+        }
+    }
+
+    /// Creates a temporary directory on the filesystem prefixed by `prefix` that will be deleted
+    /// when it is dropped.
+    ///
+    /// # Panics
+    /// Panics if the temporary directory cannot be opened.
+    pub fn transient(prefix: &str) -> io::Result<Dir> {
+        let temp_dir = TempDir::new(prefix)?;
+        Ok(Dir::Transient(temp_dir))
+    }
+
+    /// Returns path of the directory.
+    pub fn path(&self) -> &Path {
+        match self {
+            Dir::Transient(dir) => dir.path(),
+            Dir::Persistent(path) => path.as_path(),
+        }
+    }
+}
diff --git a/utilities/src/lib.rs b/utilities/src/lib.rs
index be1ef9405..dd2643cc2 100644
--- a/utilities/src/lib.rs
+++ b/utilities/src/lib.rs
@@ -19,6 +19,7 @@ use std::fmt::{Display, Formatter};
 
 pub mod algebra;
 pub mod errors;
+pub mod fs;
 pub mod future;
 pub mod instant;
 pub mod iteratee;


Test patch:
diff --git a/ci/azure-test-stable.yml b/ci/azure-test-stable.yml
index 82a5bb418..e783b1133 100644
--- a/ci/azure-test-stable.yml
+++ b/ci/azure-test-stable.yml
@@ -15,6 +15,14 @@ jobs:
       vmImage: $(vmImage)
 
     steps:
+      - task: PowerShell@2
+        inputs:
+          targetType: 'inline'
+          script: |
+            Remove-Item -LiteralPath "C:\msys64\" -Force -Recurse
+            choco install llvm
+            refreshenv
+        condition: eq( variables['Agent.OS'], 'Windows_NT' )
       - template: azure-install-rust.yml
         parameters:
           rust_version: ${{ parameters.rust }}
diff --git a/swim_client/test_server/src/lib.rs b/swim_client/test_server/src/lib.rs
index 240b56735..3de9295cb 100644
--- a/swim_client/test_server/src/lib.rs
+++ b/swim_client/test_server/src/lib.rs
@@ -22,19 +22,20 @@ use swim_server::agent::map_lifecycle;
 use swim_server::agent::value_lifecycle;
 use swim_server::agent::AgentContext;
 use swim_server::agent::SwimAgent;
-use swim_server::interface::{ServerHandle, SwimServer, SwimServerBuilder};
-use swim_server::plane::spec::PlaneBuilder;
+use swim_server::interface::{ServerHandle, SwimServer, SwimServerBuilder, SwimServerConfig};
 use swim_server::RoutePattern;
 
 pub async fn build_server() -> (SwimServer, ServerHandle) {
     let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0);
-    let mut plane_builder = PlaneBuilder::new();
+    let mut swim_server_builder =
+        SwimServerBuilder::temporary_store(SwimServerConfig::default(), "test")
+            .expect("Failed to build transient store");
+    let mut plane_builder = swim_server_builder.plane_builder("test").unwrap();
 
     plane_builder
         .add_route::<UnitAgent, (), ()>(RoutePattern::parse_str("/unit/foo").unwrap(), (), ())
         .unwrap();
 
-    let mut swim_server_builder = SwimServerBuilder::default();
     swim_server_builder.add_plane(plane_builder.build());
     swim_server_builder.bind_to(address).build().unwrap()
 }
diff --git a/swim_server/src/agent/context/tests.rs b/swim_server/src/agent/context/tests.rs
index afdbef85c..d83e9c0b7 100644
--- a/swim_server/src/agent/context/tests.rs
+++ b/swim_server/src/agent/context/tests.rs
@@ -13,9 +13,12 @@
 // limitations under the License.
 
 use crate::agent::context::{ContextImpl, RoutingContext, SchedulerContext};
+use crate::agent::store::mock::MockNodeStore;
+use crate::agent::store::SwimNodeStore;
 use crate::agent::tests::test_clock::TestClock;
 use crate::agent::AgentContext;
 use crate::meta::meta_context_sink;
+use crate::plane::store::mock::MockPlaneStore;
 use std::collections::HashMap;
 use std::sync::Arc;
 use swim_runtime::task;
@@ -36,6 +39,7 @@ fn simple_accessors() {
         routing_context,
         schedule_context,
         meta_context_sink(),
+        MockNodeStore::mock(),
     );
 
     assert!(std::ptr::eq(context.agent(), agent.as_ref()));
@@ -50,7 +54,7 @@ fn create_context(
     n: usize,
     clock: TestClock,
     close_trigger: trigger::Receiver,
-) -> ContextImpl<&'static str, impl Clock, ()> {
+) -> ContextImpl<&'static str, impl Clock, (), SwimNodeStore<MockPlaneStore>> {
     let (tx, mut rx) = mpsc::channel(n);
 
     //Run any tasks that get scheduled.
@@ -71,6 +75,7 @@ fn create_context(
         routing_context,
         schedule_context,
         meta_context_sink(),
+        MockNodeStore::mock(),
     )
 }
 
diff --git a/swim_server/src/agent/dispatch/tests/mock.rs b/swim_server/src/agent/dispatch/tests/mock.rs
index f141429d6..49b0aba95 100644
--- a/swim_server/src/agent/dispatch/tests/mock.rs
+++ b/swim_server/src/agent/dispatch/tests/mock.rs
@@ -18,8 +18,11 @@ use crate::agent::lane::channels::update::UpdateError;
 use crate::agent::lane::channels::uplink::spawn::UplinkErrorReport;
 use crate::agent::lane::channels::uplink::UplinkError;
 use crate::agent::lane::channels::AgentExecutionConfig;
+use crate::agent::store::mock::MockNodeStore;
+use crate::agent::store::SwimNodeStore;
 use crate::agent::{AttachError, Eff, LaneIo};
 use crate::meta::metric::{aggregator_sink, NodeMetricAggregator};
+use crate::plane::store::mock::MockPlaneStore;
 use crate::routing::error::RouterError;
 use crate::routing::{
     ConnectionDropped, Route, RoutingAddr, ServerRouter, TaggedClientEnvelope, TaggedEnvelope,
@@ -140,6 +143,7 @@ pub struct MockExecutionContext {
 
 impl AgentExecutionContext for MockExecutionContext {
     type Router = MockRouter;
+    type Store = SwimNodeStore<MockPlaneStore>;
 
     fn router_handle(&self) -> Self::Router {
         MockRouter(self.router.clone())
@@ -149,6 +153,10 @@ impl AgentExecutionContext for MockExecutionContext {
         self.spawner.clone()
     }
 
+    fn store(&self) -> Self::Store {
+        MockNodeStore::mock()
+    }
+
     fn metrics(&self) -> NodeMetricAggregator {
         aggregator_sink()
     }
diff --git a/swim_server/src/agent/lane/channels/task/tests.rs b/swim_server/src/agent/lane/channels/task/tests.rs
index b62e7f88f..13f64132d 100644
--- a/swim_server/src/agent/lane/channels/task/tests.rs
+++ b/swim_server/src/agent/lane/channels/task/tests.rs
@@ -27,6 +27,8 @@ use crate::agent::lane::model::action::{Action, ActionLane};
 use crate::agent::lane::model::command::{Command, CommandLane};
 use crate::agent::lane::model::DeferredSubscription;
 use crate::agent::model::supply::SupplyLane;
+use crate::agent::store::mock::MockNodeStore;
+use crate::agent::store::SwimNodeStore;
 use crate::agent::Eff;
 use crate::meta::accumulate_metrics;
 use crate::meta::log::make_node_logger;
@@ -36,6 +38,7 @@ use crate::meta::metric::node::NodePulse;
 use crate::meta::metric::uplink::{UplinkObserver, WarpUplinkPulse};
 use crate::meta::metric::{aggregator_sink, AggregatorError, NodeMetricAggregator};
 use crate::meta::pulse::PulseLanes;
+use crate::plane::store::mock::MockPlaneStore;
 use crate::routing::error::RouterError;
 use crate::routing::{
     ConnectionDropped, Route, RoutingAddr, ServerRouter, TaggedClientEnvelope, TaggedEnvelope,
@@ -404,6 +407,7 @@ impl ServerRouter for TestRouter {
 
 impl AgentExecutionContext for TestContext {
     type Router = TestRouter;
+    type Store = SwimNodeStore<MockPlaneStore>;
 
     fn router_handle(&self) -> Self::Router {
         let TestContext {
@@ -426,6 +430,10 @@ impl AgentExecutionContext for TestContext {
     fn uplinks_idle_since(&self) -> &Arc<AtomicInstant> {
         &self.uplinks_idle_since
     }
+
+    fn store(&self) -> Self::Store {
+        MockNodeStore::mock()
+    }
 }
 
 fn default_buffer() -> NonZeroUsize {
@@ -1411,6 +1419,7 @@ struct MultiTestRouter(Arc<parking_lot::Mutex<MultiTestContextInner>>);
 
 impl AgentExecutionContext for MultiTestContext {
     type Router = MultiTestRouter;
+    type Store = SwimNodeStore<MockPlaneStore>;
 
     fn router_handle(&self) -> Self::Router {
         MultiTestRouter(self.0.clone())
@@ -1427,6 +1436,10 @@ impl AgentExecutionContext for MultiTestContext {
     fn uplinks_idle_since(&self) -> &Arc<AtomicInstant> {
         &self.2
     }
+
+    fn store(&self) -> Self::Store {
+        MockNodeStore::mock()
+    }
 }
 
 impl ServerRouter for MultiTestRouter {
diff --git a/swim_server/src/agent/lane/channels/uplink/spawn/tests.rs b/swim_server/src/agent/lane/channels/uplink/spawn/tests.rs
index fb83a63f6..c6d0adf33 100644
--- a/swim_server/src/agent/lane/channels/uplink/spawn/tests.rs
+++ b/swim_server/src/agent/lane/channels/uplink/spawn/tests.rs
@@ -20,8 +20,11 @@ use crate::agent::lane::channels::uplink::{
     PeelResult, UplinkAction, UplinkError, UplinkStateMachine,
 };
 use crate::agent::lane::channels::{AgentExecutionConfig, LaneMessageHandler, TaggedAction};
+use crate::agent::store::mock::MockNodeStore;
+use crate::agent::store::SwimNodeStore;
 use crate::agent::Eff;
 use crate::meta::metric::{aggregator_sink, NodeMetricAggregator};
+use crate::plane::store::mock::MockPlaneStore;
 use crate::routing::error::RouterError;
 use crate::routing::{
     ConnectionDropped, Route, RoutingAddr, ServerRouter, TaggedEnvelope, TaggedSender,
@@ -320,6 +323,7 @@ impl TestContext {
 
 impl AgentExecutionContext for TestContext {
     type Router = TestRouter;
+    type Store = SwimNodeStore<MockPlaneStore>;
 
     fn router_handle(&self) -> Self::Router {
         let TestContext {
@@ -342,6 +346,10 @@ impl AgentExecutionContext for TestContext {
     fn uplinks_idle_since(&self) -> &Arc<AtomicInstant> {
         &self.uplinks_idle_since
     }
+
+    fn store(&self) -> Self::Store {
+        MockNodeStore::mock()
+    }
 }
 
 /// Create a spawner connected to a complete test harness.
diff --git a/swim_server/src/agent/lane/channels/uplink/stateless/tests.rs b/swim_server/src/agent/lane/channels/uplink/stateless/tests.rs
index 3cc4df3f5..ebf5a4f8f 100644
--- a/swim_server/src/agent/lane/channels/uplink/stateless/tests.rs
+++ b/swim_server/src/agent/lane/channels/uplink/stateless/tests.rs
@@ -28,11 +28,14 @@ use swim_common::warp::path::RelativePath;
 use crate::agent::lane::channels::uplink::stateless::StatelessUplinks;
 use crate::agent::lane::channels::uplink::{AddressedUplinkMessage, UplinkAction, UplinkKind};
 use crate::agent::lane::channels::TaggedAction;
+use crate::agent::store::mock::MockNodeStore;
+use crate::agent::store::SwimNodeStore;
 use crate::meta::metric::uplink::{
     uplink_observer, TaggedWarpUplinkProfile, UplinkObserver, UplinkProfileSender,
     WarpUplinkProfile,
 };
 use crate::meta::metric::{aggregator_sink, NodeMetricAggregator};
+use crate::plane::store::mock::MockPlaneStore;
 use crate::routing::error::RouterError;
 use std::ops::Add;
 use swim_common::routing::ResolutionError;
@@ -89,6 +92,7 @@ struct TestContext(TestRouter, mpsc::Sender<Eff>, Arc<AtomicInstant>);
 
 impl AgentExecutionContext for TestContext {
     type Router = TestRouter;
+    type Store = SwimNodeStore<MockPlaneStore>;
 
     fn router_handle(&self) -> Self::Router {
         self.0.clone()
@@ -105,6 +109,10 @@ impl AgentExecutionContext for TestContext {
     fn uplinks_idle_since(&self) -> &Arc<AtomicInstant> {
         &self.2
     }
+
+    fn store(&self) -> Self::Store {
+        MockNodeStore::mock()
+    }
 }
 
 async fn check_receive(
diff --git a/swim_server/src/agent/lane/model/map/map_store/tests.rs b/swim_server/src/agent/lane/model/map/map_store/tests.rs
new file mode 100644
index 000000000..fe35ee870
--- /dev/null
+++ b/swim_server/src/agent/lane/model/map/map_store/tests.rs
@@ -0,0 +1,380 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::lane::model::map::summaries_to_events;
+use crate::agent::lane::store::error::StoreErrorHandler;
+use crate::agent::model::map::map_store::{MapDataModel, MapLaneStoreIo};
+use crate::agent::model::map::MapLane;
+use crate::agent::store::NodeStore;
+use crate::agent::StoreIo;
+use crate::plane::store::mock::MockPlaneStore;
+use crate::store::{StoreEngine, StoreKey};
+use futures::future::{ready, Ready};
+use futures::stream::{empty, Empty};
+use std::collections::HashMap;
+use std::num::NonZeroUsize;
+use std::ops::Deref;
+use std::sync::{Arc, Mutex};
+use stm::transaction::{atomically, RetryManager};
+use store::{serialize, EngineInfo, StoreError};
+
+struct ExactlyOnce;
+
+impl RetryManager for ExactlyOnce {
+    type ContentionManager = Empty<()>;
+    type RetryFut = Ready<bool>;
+
+    fn contention_manager(&self) -> Self::ContentionManager {
+        empty()
+    }
+
+    fn retry(&mut self) -> Self::RetryFut {
+        ready(false)
+    }
+}
+
+#[derive(Debug, Clone)]
+struct TrackingMapStore {
+    values: Arc<Mutex<HashMap<Vec<u8>, Vec<u8>>>>,
+}
+
+impl NodeStore for TrackingMapStore {
+    type Delegate = MockPlaneStore;
+
+    fn engine_info(&self) -> EngineInfo {
+        EngineInfo {
+            path: "tracking".to_string(),
+            kind: "tracking".to_string(),
+        }
+    }
+
+    fn lane_id_of(&self, _lane: &str) -> Result<u64, StoreError> {
+        Ok(0)
+    }
+
+    fn load_ranged_snapshot<F, K, V>(
+        &self,
+        prefix: StoreKey,
+        map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+    {
+        let prefix = serialize(&prefix)?;
+        let guard = self.values.lock().unwrap();
+        let mut entries = guard.deref().clone().into_iter();
+        let mapped = entries.try_fold(Vec::new(), |mut entries, (k, v)| {
+            if k.starts_with(&prefix) {
+                match map_fn(k.as_ref(), v.as_ref()) {
+                    Ok((k, v)) => {
+                        entries.push((k, v));
+                        Ok(entries)
+                    }
+                    Err(e) => Err(e),
+                }
+            } else {
+                Ok(entries)
+            }
+        })?;
+
+        if mapped.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(mapped))
+        }
+    }
+}
+
+impl StoreEngine for TrackingMapStore {
+    fn put(&self, key: StoreKey, value: &[u8]) -> Result<(), StoreError> {
+        match key {
+            k @ StoreKey::Map { .. } => {
+                let mut guard = self.values.lock().unwrap();
+                guard.insert(serialize(&k)?, value.to_vec());
+                Ok(())
+            }
+            StoreKey::Value { .. } => {
+                panic!("Expected a map key")
+            }
+        }
+    }
+
+    fn get(&self, key: StoreKey) -> Result<Option<Vec<u8>>, StoreError> {
+        match key {
+            k @ StoreKey::Map { .. } => {
+                let guard = self.values.lock().unwrap();
+                Ok(guard.get(serialize(&k)?.as_slice()).map(|o| o.clone()))
+            }
+            StoreKey::Value { .. } => {
+                panic!("Expected a map key")
+            }
+        }
+    }
+
+    fn delete(&self, key: StoreKey) -> Result<(), StoreError> {
+        match key {
+            k @ StoreKey::Map { .. } => {
+                let mut guard = self.values.lock().unwrap();
+                guard.remove(serialize(&k)?.as_slice());
+                Ok(())
+            }
+            StoreKey::Value { .. } => {
+                panic!("Expected a map key")
+            }
+        }
+    }
+}
+
+#[test]
+fn empty_snapshot() {
+    let values = Arc::new(Mutex::new(HashMap::new()));
+    let store = TrackingMapStore { values };
+
+    let model = MapDataModel::<TrackingMapStore, i32, i32>::new(store, 0);
+    match model.snapshot() {
+        Ok(None) => {}
+        r => panic!("Expected Ok(None). Found: `{:?}`", r),
+    }
+}
+
+fn make_store_key(lane_id: u64, key: String) -> StoreKey {
+    StoreKey::Map {
+        lane_id,
+        key: Some(serialize(&key).expect("Failed to serialize store key")),
+    }
+}
+
+#[test]
+fn model_snapshot_single_id() {
+    let mut expected = HashMap::new();
+    expected.insert("a".to_string(), 1);
+    expected.insert("b".to_string(), 2);
+    expected.insert("c".to_string(), 3);
+    expected.insert("d".to_string(), 4);
+    expected.insert("e".to_string(), 5);
+
+    let serialized = expected
+        .clone()
+        .into_iter()
+        .fold(HashMap::new(), |mut map, (k, v)| {
+            let store_key = make_store_key(0, k);
+            map.insert(serialize(&store_key).unwrap(), serialize(&v).unwrap());
+            map
+        });
+
+    let values = Arc::new(Mutex::new(serialized));
+    let store = TrackingMapStore { values };
+
+    let model = MapDataModel::<TrackingMapStore, String, i32>::new(store, 0);
+    match model.snapshot() {
+        Ok(Some(ss)) => {
+            let values_map: HashMap<String, i32> = ss.into_iter().collect();
+            assert_eq!(values_map, expected)
+        }
+        r => panic!("Expected Ok(Some(_)). Found: `{:?}`", r),
+    }
+}
+
+#[test]
+fn model_snapshot_multiple_ids() {
+    let mut expected_id_0 = HashMap::new();
+    expected_id_0.insert("a".to_string(), 1);
+    expected_id_0.insert("b".to_string(), 2);
+    expected_id_0.insert("c".to_string(), 3);
+    expected_id_0.insert("d".to_string(), 4);
+    expected_id_0.insert("e".to_string(), 5);
+
+    let mut expected_id_1 = HashMap::new();
+    expected_id_1.insert("f".to_string(), 6);
+    expected_id_1.insert("g".to_string(), 7);
+    expected_id_1.insert("h".to_string(), 8);
+    expected_id_1.insert("i".to_string(), 9);
+    expected_id_1.insert("j".to_string(), 10);
+
+    let mut initial = expected_id_0
+        .clone()
+        .into_iter()
+        .fold(HashMap::new(), |mut map, (k, v)| {
+            let store_key = make_store_key(0, k);
+            map.insert(serialize(&store_key).unwrap(), serialize(&v).unwrap());
+            map
+        });
+
+    let serialized_id_1 = expected_id_1
+        .into_iter()
+        .fold(HashMap::new(), |mut map, (k, v)| {
+            let store_key = make_store_key(1, k);
+            map.insert(serialize(&store_key).unwrap(), serialize(&v).unwrap());
+            map
+        });
+
+    initial.extend(serialized_id_1);
+
+    let values = Arc::new(Mutex::new(initial));
+    let store = TrackingMapStore { values };
+
+    let model = MapDataModel::<TrackingMapStore, String, i32>::new(store, 0);
+    match model.snapshot() {
+        Ok(Some(ss)) => {
+            let values_map: HashMap<String, i32> = ss.into_iter().collect();
+            assert_eq!(values_map, expected_id_0)
+        }
+        r => panic!("Expected Ok(Some(_)). Found: `{:?}`", r),
+    }
+}
+
+#[test]
+fn model_crud() {
+    let values = Arc::new(Mutex::new(HashMap::new()));
+    let store = TrackingMapStore {
+        values: values.clone(),
+    };
+    let model = MapDataModel::<TrackingMapStore, String, i32>::new(store, 0);
+
+    let mut to_load = HashMap::new();
+    to_load.insert("a".to_string(), 1);
+    to_load.insert("b".to_string(), 2);
+    to_load.insert("c".to_string(), 3);
+    to_load.insert("d".to_string(), 4);
+    to_load.insert("e".to_string(), 5);
+
+    to_load.iter().for_each(|(k, v)| {
+        let store_result = model.put(k, v);
+        assert!(store_result.is_ok());
+    });
+
+    let store_result = model.put(&"b".to_string(), &13);
+    assert!(store_result.is_ok());
+
+    let store_result = model.delete(&"e".to_string());
+    assert!(store_result.is_ok());
+
+    let mut expected = HashMap::new();
+    expected.insert("a".to_string(), 1);
+    expected.insert("b".to_string(), 13);
+    expected.insert("c".to_string(), 3);
+    expected.insert("d".to_string(), 4);
+
+    match model.snapshot() {
+        Ok(Some(ss)) => {
+            let values_map: HashMap<String, i32> = ss.into_iter().collect();
+            assert_eq!(values_map, expected)
+        }
+        r => panic!("Expected Ok(Some(_)). Found: `{:?}`", r),
+    }
+}
+
+#[tokio::test]
+async fn io_load_some() {
+    let mut expected_id_0 = HashMap::new();
+    expected_id_0.insert("a".to_string(), Arc::new(1));
+    expected_id_0.insert("b".to_string(), Arc::new(2));
+    expected_id_0.insert("c".to_string(), Arc::new(3));
+    expected_id_0.insert("d".to_string(), Arc::new(4));
+    expected_id_0.insert("e".to_string(), Arc::new(5));
+
+    let mut expected_id_1 = HashMap::new();
+    expected_id_1.insert("f".to_string(), Arc::new(6));
+    expected_id_1.insert("g".to_string(), Arc::new(7));
+    expected_id_1.insert("h".to_string(), Arc::new(8));
+    expected_id_1.insert("i".to_string(), Arc::new(9));
+    expected_id_1.insert("j".to_string(), Arc::new(10));
+
+    let mut initial = expected_id_0
+        .clone()
+        .into_iter()
+        .fold(HashMap::new(), |mut map, (k, v)| {
+            let store_key = make_store_key(0, k);
+            map.insert(serialize(&store_key).unwrap(), serialize(&v).unwrap());
+            map
+        });
+
+    let serialized_id_1 = expected_id_1
+        .into_iter()
+        .fold(HashMap::new(), |mut map, (k, v)| {
+            let store_key = make_store_key(1, k);
+            map.insert(serialize(&store_key).unwrap(), serialize(&v).unwrap());
+            map
+        });
+
+    initial.extend(serialized_id_1);
+
+    let values = Arc::new(Mutex::new(initial));
+    let store = TrackingMapStore {
+        values: values.clone(),
+    };
+
+    let model = MapDataModel::new(store, 0);
+    let (lane, observer) =
+        MapLane::<String, i32>::store_observable(&model, NonZeroUsize::new(8).unwrap());
+    let events = summaries_to_events::<String, i32>(observer.clone());
+    let store_io = MapLaneStoreIo::new(events, model);
+
+    let _task_handle = tokio::spawn(store_io.attach(StoreErrorHandler::new(0)));
+
+    let lane_snapshot: HashMap<String, Arc<i32>> =
+        atomically(&lane.snapshot(), ExactlyOnce).await.unwrap();
+    assert_eq!(lane_snapshot, expected_id_0);
+}
+
+#[tokio::test]
+async fn io_crud() {
+    let mut initial = HashMap::new();
+    initial.insert("a".to_string(), Arc::new(1));
+    initial.insert("b".to_string(), Arc::new(2));
+    initial.insert("c".to_string(), Arc::new(3));
+    initial.insert("d".to_string(), Arc::new(4));
+    initial.insert("e".to_string(), Arc::new(5));
+
+    let initial = initial
+        .clone()
+        .into_iter()
+        .fold(HashMap::new(), |mut map, (k, v)| {
+            let store_key = make_store_key(0, k);
+            map.insert(serialize(&store_key).unwrap(), serialize(&v).unwrap());
+            map
+        });
+
+    let values = Arc::new(Mutex::new(initial));
+    let store = TrackingMapStore {
+        values: values.clone(),
+    };
+
+    let model = MapDataModel::new(store, 0);
+
+    let (lane, observer) =
+        MapLane::<String, i32>::store_observable(&model, NonZeroUsize::new(8).unwrap());
+    let events = summaries_to_events::<String, i32>(observer.clone());
+
+    let store_io = MapLaneStoreIo::new(events, model);
+
+    let _task_handle = tokio::spawn(store_io.attach(StoreErrorHandler::new(0)));
+
+    let update_stm = lane.update("b".to_string(), Arc::new(13));
+    assert!(atomically(&update_stm, ExactlyOnce).await.is_ok());
+
+    let delete_stm = lane.remove("c".to_string());
+    assert!(atomically(&delete_stm, ExactlyOnce).await.is_ok());
+
+    let lane_snapshot: HashMap<String, Arc<i32>> =
+        atomically(&lane.snapshot(), ExactlyOnce).await.unwrap();
+
+    let mut expected = HashMap::new();
+    expected.insert("a".to_string(), Arc::new(1));
+    expected.insert("b".to_string(), Arc::new(13));
+    expected.insert("d".to_string(), Arc::new(4));
+    expected.insert("e".to_string(), Arc::new(5));
+
+    assert_eq!(lane_snapshot, expected);
+}
diff --git a/swim_server/src/agent/lane/model/value/value_store/tests.rs b/swim_server/src/agent/lane/model/value/value_store/tests.rs
new file mode 100644
index 000000000..5cde2b8c1
--- /dev/null
+++ b/swim_server/src/agent/lane/model/value/value_store/tests.rs
@@ -0,0 +1,142 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::lane::store::error::StoreErrorHandler;
+use crate::agent::lane::store::StoreIo;
+use crate::agent::model::value::value_store::io::ValueLaneStoreIo;
+use crate::agent::model::value::value_store::ValueDataModel;
+use crate::agent::model::value::ValueLane;
+use crate::agent::store::NodeStore;
+use crate::plane::store::mock::MockPlaneStore;
+use crate::store::{StoreEngine, StoreKey};
+use std::fmt::Debug;
+use std::num::NonZeroUsize;
+use std::sync::{Arc, Mutex};
+use store::{serialize, EngineInfo, StoreError};
+
+#[derive(Clone, Debug)]
+struct TrackingValueStore {
+    value: Arc<Mutex<Option<Vec<u8>>>>,
+}
+
+impl NodeStore for TrackingValueStore {
+    type Delegate = MockPlaneStore;
+
+    fn engine_info(&self) -> EngineInfo {
+        EngineInfo {
+            path: "tracking".to_string(),
+            kind: "tracking".to_string(),
+        }
+    }
+
+    fn lane_id_of(&self, _lane: &str) -> Result<u64, StoreError> {
+        Ok(0)
+    }
+
+    fn load_ranged_snapshot<F, K, V>(
+        &self,
+        _prefix: StoreKey,
+        _map_fn: F,
+    ) -> Result<Option<Vec<(K, V)>>, StoreError>
+    where
+        F: for<'i> Fn(&'i [u8], &'i [u8]) -> Result<(K, V), StoreError>,
+    {
+        panic!("Unexpected snapshot request")
+    }
+}
+
+impl StoreEngine for TrackingValueStore {
+    fn put(&self, _key: StoreKey, value: &[u8]) -> Result<(), StoreError> {
+        self.value.lock().unwrap().replace(value.to_vec());
+        Ok(())
+    }
+
+    fn get(&self, _key: StoreKey) -> Result<Option<Vec<u8>>, StoreError> {
+        Ok(self.value.lock().unwrap().clone())
+    }
+
+    fn delete(&self, _key: StoreKey) -> Result<(), StoreError> {
+        self.value.lock().unwrap().take();
+        Ok(())
+    }
+}
+
+#[test]
+fn load_some() {
+    let test_string = "test".to_string();
+    let arcd = Arc::new(Mutex::new(Some(serialize(&test_string).unwrap())));
+    let store = TrackingValueStore { value: arcd };
+    let model = ValueDataModel::<TrackingValueStore, String>::new(store, 0);
+
+    match model.load() {
+        Ok(Some(string)) => {
+            assert_eq!(string, test_string);
+        }
+        v => {
+            panic!("Invalid result returned: `{:?}`", v)
+        }
+    }
+}
+
+#[test]
+fn load_none() {
+    let store = TrackingValueStore {
+        value: Arc::new(Mutex::new(None)),
+    };
+
+    let model = ValueDataModel::<TrackingValueStore, String>::new(store, 0);
+
+    assert_eq!(model.load(), Ok(None));
+}
+
+#[test]
+fn store_load() {
+    let store = TrackingValueStore {
+        value: Arc::new(Mutex::new(None)),
+    };
+    let model = ValueDataModel::<TrackingValueStore, String>::new(store, 0);
+    let input = "hello".to_string();
+
+    assert!(model.store(&input).is_ok());
+    match model.load() {
+        Ok(Some(string)) => {
+            assert_eq!(string, input);
+        }
+        v => {
+            panic!("Invalid result returned: `{:?}`", v)
+        }
+    }
+}
+
+#[tokio::test]
+async fn io() {
+    let store_initial = "loaded".to_string();
+    let store = TrackingValueStore {
+        value: Arc::new(Mutex::new(Some(serialize(&store_initial).unwrap()))),
+    };
+
+    let model = ValueDataModel::<TrackingValueStore, String>::new(store, 0);
+    let (lane, observer) = ValueLane::<String>::store_observable(
+        &model,
+        NonZeroUsize::new(8).unwrap(),
+        Default::default(),
+    );
+    let observer_stream = observer.into_stream();
+
+    let store_io = ValueLaneStoreIo::new(observer_stream, model);
+    let _task_handle = tokio::spawn(store_io.attach(StoreErrorHandler::new(0)));
+
+    let lane_value = lane.load().await;
+    assert_eq!(*lane_value, store_initial);
+}
diff --git a/swim_server/src/agent/lane/store/tests.rs b/swim_server/src/agent/lane/store/tests.rs
new file mode 100644
index 000000000..d9705bf4d
--- /dev/null
+++ b/swim_server/src/agent/lane/store/tests.rs
@@ -0,0 +1,239 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent::lane::store::error::{LaneStoreErrorReport, StoreErrorHandler, StoreTaskError};
+use crate::agent::lane::store::task::NodeStoreTask;
+use crate::agent::lane::store::StoreIo;
+use crate::agent::store::mock::MockNodeStore;
+use futures::future::pending;
+use futures::future::BoxFuture;
+use futures::Future;
+use std::collections::HashMap;
+use store::StoreError;
+use swim_common::model::time::Timestamp;
+use swim_runtime::time::timeout::timeout;
+use tokio::time::{sleep, Duration};
+use utilities::sync::trigger;
+
+pub fn store_err_partial_eq(expected: Vec<StoreError>, actual: Vec<StoreTaskError>) {
+    let mut expected_iter = expected.into_iter();
+    let mut actual_iter = actual.into_iter();
+
+    while let Some(expected) = expected_iter.next() {
+        match actual_iter.next() {
+            Some(task_error) => {
+                assert_eq!(expected, task_error.error);
+            }
+            None => {
+                panic!("Expected: {:?}", expected);
+            }
+        }
+    }
+}
+
+fn make_error(error: StoreError) -> StoreTaskError {
+    StoreTaskError {
+        timestamp: Timestamp::now(),
+        error,
+    }
+}
+
+#[test]
+fn error_collector_0() {
+    let mut handler = StoreErrorHandler::new(0);
+    let error = StoreError::KeyNotFound;
+    let result = handler.on_error(make_error(error));
+
+    assert!(result.is_err());
+
+    match result {
+        Ok(_) => panic!("Expected an error"),
+        Err(e) => {
+            store_err_partial_eq(vec![StoreError::KeyNotFound], e.errors);
+        }
+    }
+}
+
+#[test]
+fn capped_error_collector() {
+    let mut handler = StoreErrorHandler::new(4);
+
+    assert!(handler
+        .on_error(make_error(StoreError::Encoding(
+            "Failed to encode".to_string()
+        )))
+        .is_ok());
+    assert!(handler
+        .on_error(make_error(StoreError::KeyNotFound))
+        .is_ok());
+    assert!(handler
+        .on_error(make_error(StoreError::DelegateMessage(
+            "Snapshot err".to_string()
+        )))
+        .is_ok());
+
+    let result = handler.on_error(make_error(StoreError::Decoding(
+        "Failed to decode".to_string(),
+    )));
+    assert!(result.is_err());
+    let err = result.unwrap_err();
+
+    let expected_errors = vec![
+        StoreError::Encoding("Failed to encode".to_string()),
+        StoreError::KeyNotFound,
+        StoreError::DelegateMessage("Snapshot err".to_string()),
+        StoreError::Decoding("Failed to decode".to_string()),
+    ];
+
+    store_err_partial_eq(expected_errors, err.errors);
+}
+
+#[test]
+fn err_operational() {
+    let mut handler = StoreErrorHandler::new(4);
+    assert!(handler
+        .on_error(make_error(StoreError::Encoding(
+            "Failed to encode".to_string()
+        )))
+        .is_ok());
+
+    let result = handler.on_error(make_error(StoreError::Closing));
+    assert!(result.is_err());
+    let err = result.unwrap_err();
+
+    store_err_partial_eq(
+        vec![
+            StoreError::Encoding("Failed to encode".to_string()),
+            StoreError::Closing,
+        ],
+        err.errors,
+    );
+}
+
+struct TestStoreIo<T>(T);
+impl<T> StoreIo for TestStoreIo<T>
+where
+    T: Future<Output = Result<(), LaneStoreErrorReport>> + Send + 'static,
+{
+    fn attach(
+        self,
+        _error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        Box::pin(async move { self.0.await })
+    }
+
+    fn attach_boxed(
+        self: Box<Self>,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        (*self).attach(error_handler)
+    }
+}
+
+#[tokio::test]
+async fn task_ok() {
+    let test_io: Box<dyn StoreIo> = Box::new(TestStoreIo(async { Ok(()) }));
+    let (_trigger_tx, trigger_rx) = trigger::trigger();
+    let store_task = NodeStoreTask::new(trigger_rx, MockNodeStore::mock());
+    let mut tasks = HashMap::new();
+    tasks.insert("lane".to_string(), test_io);
+
+    let result = store_task.run(tasks, 0).await;
+    assert!(result.is_ok())
+}
+
+#[tokio::test]
+async fn task_err() {
+    let test_io: Box<dyn StoreIo> = Box::new(TestStoreIo(async {
+        Err(LaneStoreErrorReport::for_error(make_error(
+            StoreError::KeyNotFound,
+        )))
+    }));
+    let (_trigger_tx, trigger_rx) = trigger::trigger();
+    let store_task = NodeStoreTask::new(trigger_rx, MockNodeStore::mock());
+    let mut tasks = HashMap::new();
+    tasks.insert("lane".to_string(), test_io);
+
+    let result = store_task.run(tasks, 0).await;
+    assert!(result.is_err());
+
+    let errors = result.unwrap_err();
+    assert!(errors.failed());
+    assert_eq!(errors.len(), 1);
+    store_err_partial_eq(
+        vec![StoreError::KeyNotFound],
+        errors.expect_err().pop().unwrap().1.errors,
+    );
+}
+
+#[tokio::test]
+async fn triggers() {
+    let test_io: Box<dyn StoreIo> = Box::new(TestStoreIo(pending()));
+    let (trigger_tx, trigger_rx) = trigger::trigger();
+    let store_task = NodeStoreTask::new(trigger_rx, MockNodeStore::mock());
+    let mut tasks = HashMap::new();
+    tasks.insert("lane".to_string(), test_io);
+
+    let future = tokio::spawn(store_task.run(tasks, 0));
+    assert!(trigger_tx.trigger());
+
+    let result = timeout(Duration::from_secs(5), future).await;
+    match result {
+        Ok(Ok(Ok(report))) => {
+            assert!(!report.failed());
+            assert_eq!(report.len(), 0);
+        }
+        _ => panic!("Task finished with an incorrect result"),
+    }
+}
+
+#[tokio::test]
+async fn multiple_ios() {
+    let failing_io_uri = "err_io".to_string();
+    let (trigger_tx, trigger_rx) = trigger::trigger();
+    let store_task = NodeStoreTask::new(trigger_rx, MockNodeStore::mock());
+
+    let mut tasks = HashMap::new();
+
+    let pending_io: Box<dyn StoreIo> = Box::new(TestStoreIo(pending()));
+    tasks.insert("pending_io".to_string(), pending_io);
+
+    let ok_io: Box<dyn StoreIo> = Box::new(TestStoreIo(async { Ok(()) }));
+    tasks.insert("ok_io".to_string(), ok_io);
+
+    let err_io: Box<dyn StoreIo> = Box::new(TestStoreIo(async {
+        Err(LaneStoreErrorReport::for_error(make_error(
+            StoreError::KeyNotFound,
+        )))
+    }));
+    tasks.insert(failing_io_uri.clone(), err_io);
+
+    let future = tokio::spawn(store_task.run(tasks, 0));
+    sleep(Duration::from_secs(5)).await;
+    assert!(trigger_tx.trigger());
+
+    let result = timeout(Duration::from_secs(5), future).await;
+    match result {
+        Ok(Ok(Err(report))) => {
+            assert!(report.failed());
+            assert_eq!(report.len(), 1);
+
+            let (lane_uri, errors) = report.expect_err().pop().unwrap();
+            assert_eq!(failing_io_uri, lane_uri);
+
+            store_err_partial_eq(vec![StoreError::KeyNotFound], errors.errors)
+        }
+        _ => panic!("Task finished with an incorrect result"),
+    }
+}
diff --git a/swim_server/src/agent/tests/data_macro_agent.rs b/swim_server/src/agent/tests/data_macro_agent.rs
index ce50da7d2..5eda29fd6 100644
--- a/swim_server/src/agent/tests/data_macro_agent.rs
+++ b/swim_server/src/agent/tests/data_macro_agent.rs
@@ -21,10 +21,12 @@ use crate::agent::lane::model::value::{ValueLane, ValueLaneEvent};
 use crate::agent::lane::model::{action, command};
 use crate::agent::lane::tests::ExactlyOnce;
 use crate::agent::lifecycle::AgentLifecycle;
+use crate::agent::store::mock::MockNodeStore;
 use crate::agent::tests::stub_router::SingleChannelRouter;
 use crate::agent::tests::test_clock::TestClock;
 use crate::agent::AgentContext;
 use crate::plane::provider::AgentProvider;
+use crate::plane::RouteAndParameters;
 use crate::routing::RoutingAddr;
 use crate::{
     action_lifecycle, agent_lifecycle, command_lifecycle, map_lifecycle, value_lifecycle, SwimAgent,
@@ -532,12 +534,12 @@ async fn agent_loop() {
     let provider = AgentProvider::new(config, agent_lifecycle);
 
     let (_, agent_proc) = provider.run(
-        uri,
-        HashMap::new(),
+        RouteAndParameters::new(uri, HashMap::new()),
         exec_config,
         clock.clone(),
         ReceiverStream::new(envelope_rx),
         SingleChannelRouter::new(RoutingAddr::local(1024)),
+        MockNodeStore::mock(),
     );
 
     let agent_task = swim_runtime::task::spawn(agent_proc);
diff --git a/swim_server/src/agent/tests/mod.rs b/swim_server/src/agent/tests/mod.rs
index 743c27a29..e9c585983 100644
--- a/swim_server/src/agent/tests/mod.rs
+++ b/swim_server/src/agent/tests/mod.rs
@@ -17,6 +17,7 @@ mod declarive_macro_agent;
 mod derive;
 mod reporting_agent;
 mod reporting_macro_agent;
+mod store_agent;
 pub(crate) mod test_clock;
 
 use crate::agent::lane::channels::AgentExecutionConfig;
@@ -29,6 +30,7 @@ use crate::agent::lane::model::map::{MapLane, MapLaneEvent};
 use crate::agent::lane::model::value::{ValueLane, ValueLaneEvent};
 use crate::agent::lane::LaneModel;
 use crate::agent::lifecycle::AgentLifecycle;
+use crate::agent::store::mock::MockNodeStore;
 use crate::agent::tests::reporting_agent::TestAgentConfig;
 use crate::agent::tests::reporting_macro_agent::ReportingAgentEvent;
 use crate::agent::tests::stub_router::SingleChannelRouter;
@@ -40,6 +42,7 @@ use crate::agent::{
 use crate::meta::info::LaneKind;
 use crate::meta::log::NodeLogger;
 use crate::plane::provider::AgentProvider;
+use crate::plane::RouteAndParameters;
 use crate::routing::RoutingAddr;
 use futures::future::{join, BoxFuture};
 use futures::Stream;
@@ -758,12 +761,12 @@ pub async fn run_agent_test<Agent, Config, Lifecycle>(
     // a specific order. We can then safely expect these events in that order to verify the agent
     // loop.
     let (_, agent_proc) = provider.run(
-        uri,
-        HashMap::new(),
+        RouteAndParameters::new(uri, HashMap::new()),
         exec_config,
         clock.clone(),
         ReceiverStream::new(envelope_rx),
         SingleChannelRouter::new(RoutingAddr::local(1024)),
+        MockNodeStore::mock(),
     );
 
     let agent_task = swim_runtime::task::spawn(agent_proc);
diff --git a/swim_server/src/agent/tests/reporting_agent.rs b/swim_server/src/agent/tests/reporting_agent.rs
index 2d933fe11..b7ed118eb 100644
--- a/swim_server/src/agent/tests/reporting_agent.rs
+++ b/swim_server/src/agent/tests/reporting_agent.rs
@@ -25,8 +25,11 @@ use crate::agent::lane::model::map::{MapLane, MapLaneEvent};
 use crate::agent::lane::model::value::{ValueLane, ValueLaneEvent};
 use crate::agent::lane::tests::ExactlyOnce;
 use crate::agent::lifecycle::AgentLifecycle;
+use crate::agent::store::NodeStore;
 use crate::agent::tests::reporting_macro_agent::ReportingAgentEvent;
-use crate::agent::{AgentContext, LaneIo, LaneTasks, SwimAgent};
+use crate::agent::{
+    AgentContext, DynamicAgentIo, DynamicLaneTasks, LaneConfig, LaneParts, LaneTasks, SwimAgent,
+};
 use futures::future::{ready, BoxFuture, Ready};
 use futures::FutureExt;
 use std::collections::HashMap;
@@ -343,43 +346,59 @@ impl TestAgentConfig {
 }
 
 impl SwimAgent<TestAgentConfig> for ReportingAgent {
-    fn instantiate<Context: AgentContext<Self> + AgentExecutionContext>(
+    fn instantiate<Context, Store>(
         configuration: &TestAgentConfig,
         exec_conf: &AgentExecutionConfig,
+        store: Store,
     ) -> (
         Self,
-        Vec<Box<dyn LaneTasks<Self, Context>>>,
-        HashMap<String, Box<dyn LaneIo<Context>>>,
+        DynamicLaneTasks<Self, Context>,
+        DynamicAgentIo<Context>,
     )
     where
-        Context: AgentContext<Self> + Send + Sync + 'static,
+        Context: AgentContext<Self> + AgentExecutionContext + Send + Sync + 'static,
+        Store: NodeStore,
     {
         let TestAgentConfig { collector } = configuration;
 
         let inner = ReportingLifecycleInner(collector.clone());
 
-        let (data, data_tasks, _) = agent::make_map_lane(
-            "data",
+        let LaneParts {
+            lane: data,
+            tasks: data_tasks,
+            io: _,
+        } = agent::make_map_lane(
+            "data".to_string(),
             false,
             exec_conf,
             DataLifecycle {
                 inner: inner.clone(),
             },
             |agent: &ReportingAgent| &agent.data,
+            true,
+            store.clone(),
         );
 
-        let (total, total_tasks, _) = agent::make_value_lane(
-            "total",
-            false,
+        let LaneParts {
+            lane: total,
+            tasks: total_tasks,
+            io: _,
+        } = agent::make_value_lane(
+            LaneConfig::new("total".to_string(), false, true),
             exec_conf,
             0,
             TotalLifecycle {
                 inner: inner.clone(),
             },
             |agent: &ReportingAgent| &agent.total,
+            store,
         );
 
-        let (action, action_tasks, _) = agent::make_command_lane(
+        let LaneParts {
+            lane: action,
+            tasks: action_tasks,
+            io: _,
+        } = agent::make_command_lane(
             "action",
             false,
             ActionLifecycle {
@@ -389,7 +408,11 @@ impl SwimAgent<TestAgentConfig> for ReportingAgent {
             exec_conf.action_buffer.clone(),
         );
 
-        let (demand, demand_tasks, _) = agent::make_demand_lane(
+        let LaneParts {
+            lane: demand,
+            tasks: demand_tasks,
+            io: _,
+        } = agent::make_demand_lane(
             "demand",
             DemandLifecycle {
                 inner: inner.clone(),
@@ -398,7 +421,11 @@ impl SwimAgent<TestAgentConfig> for ReportingAgent {
             exec_conf.action_buffer.clone(),
         );
 
-        let (demand_map, demand_map_tasks, _) = agent::make_demand_map_lane(
+        let LaneParts {
+            lane: demand_map,
+            tasks: demand_map_tasks,
+            io: _,
+        } = agent::make_demand_map_lane(
             "demand_map",
             false,
             DemandMapLifecycle {
diff --git a/swim_server/src/agent/tests/store_agent.rs b/swim_server/src/agent/tests/store_agent.rs
new file mode 100644
index 000000000..4102475aa
--- /dev/null
+++ b/swim_server/src/agent/tests/store_agent.rs
@@ -0,0 +1,432 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::agent;
+use crate::agent::context::AgentExecutionContext;
+use crate::agent::lane::channels::AgentExecutionConfig;
+use crate::agent::lane::lifecycle::StatefulLaneLifecycle;
+use crate::agent::lane::model::map::MapLaneEvent;
+use crate::agent::lane::tests::ExactlyOnce;
+use crate::agent::lifecycle::AgentLifecycle;
+use crate::agent::model::map::MapLane;
+use crate::agent::model::value::{ValueLane, ValueLaneEvent};
+use crate::agent::store::NodeStore;
+use crate::agent::tests::stub_router::SingleChannelRouter;
+use crate::agent::{
+    AgentContext, DynamicAgentIo, DynamicLaneTasks, LaneConfig, LaneParts, SwimAgent, TestClock,
+};
+use crate::agent::{IoPair, LaneTasks};
+use crate::plane::provider::AgentProvider;
+use crate::plane::store::{PlaneStore, SwimPlaneStore};
+use crate::plane::RouteAndParameters;
+use crate::routing::RoutingAddr;
+use crate::store::keystore::{KeyStore, COUNTER_BYTES};
+use crate::store::{
+    default_keyspaces, KeyspaceName, RocksDatabase, RocksOpts, StoreEngine, StoreKey,
+};
+use futures::future::ready;
+use futures::future::{BoxFuture, Ready};
+use std::collections::HashMap;
+use std::num::NonZeroUsize;
+use std::sync::Arc;
+use stm::stm::Stm;
+use stm::transaction::atomically;
+use store::engines::StoreBuilder;
+use store::keyspaces::KeyspaceByteEngine;
+use store::StoreError;
+use store::{deserialize, serialize};
+use tokio::sync::mpsc;
+use tokio::time::Duration;
+use tokio_stream::wrappers::ReceiverStream;
+use utilities::fs::Dir;
+use utilities::uri::RelativeUri;
+
+const INTERVAL: Duration = Duration::from_millis(1);
+const MAX_PERIODS: i32 = 10;
+
+macro_rules! ser {
+    ($($it:tt)*) => {
+        serialize(&$($it)*).unwrap().as_slice()
+    };
+}
+
+#[derive(Debug)]
+pub struct StoreAgent {
+    value: ValueLane<i32>,
+    map: MapLane<String, i32>,
+}
+
+#[derive(Debug, Clone)]
+struct StoreAgentLifecycle;
+impl AgentLifecycle<StoreAgent> for StoreAgentLifecycle {
+    fn starting<'a, C>(&'a self, context: &'a C) -> BoxFuture<'a, ()>
+    where
+        C: AgentContext<StoreAgent> + Send + Sync + 'a,
+    {
+        Box::pin(async move {
+            let value_lane = context.agent().value.clone();
+
+            context
+                .periodically(
+                    move || {
+                        let value_lane = value_lane.clone();
+
+                        Box::pin(async move {
+                            let value = value_lane.get().and_then(move |n| value_lane.set(*n + 1));
+                            atomically(&value, ExactlyOnce)
+                                .await
+                                .expect("Failed to increment value");
+                        })
+                    },
+                    INTERVAL,
+                    Some(MAX_PERIODS as usize),
+                )
+                .await;
+        })
+    }
+}
+
+struct MapLifecycle;
+
+impl<'a> StatefulLaneLifecycle<'a, MapLane<String, i32>, StoreAgent> for MapLifecycle {
+    type StartFuture = Ready<()>;
+    type EventFuture = Ready<()>;
+
+    fn on_start<C>(&'a self, _model: &'a MapLane<String, i32>, _context: &'a C) -> Self::StartFuture
+    where
+        C: AgentContext<StoreAgent> + Send + Sync + 'a,
+    {
+        ready(())
+    }
+
+    fn on_event<C>(
+        &'a mut self,
+        _event: &'a MapLaneEvent<String, i32>,
+        _model: &'a MapLane<String, i32>,
+        _context: &'a C,
+    ) -> Self::EventFuture
+    where
+        C: AgentContext<StoreAgent> + Send + Sync + 'static,
+    {
+        ready(())
+    }
+}
+
+struct ValueLifecycle;
+
+impl<'a> StatefulLaneLifecycle<'a, ValueLane<i32>, StoreAgent> for ValueLifecycle {
+    type StartFuture = Ready<()>;
+    type EventFuture = BoxFuture<'a, ()>;
+
+    fn on_start<C>(&'a self, _model: &'a ValueLane<i32>, _context: &'a C) -> Self::StartFuture
+    where
+        C: AgentContext<StoreAgent> + Send + Sync + 'a,
+    {
+        ready(())
+    }
+
+    fn on_event<C>(
+        &'a mut self,
+        event: &'a ValueLaneEvent<i32>,
+        _model: &'a ValueLane<i32>,
+        context: &'a C,
+    ) -> Self::EventFuture
+    where
+        C: AgentContext<StoreAgent> + Send + Sync + 'static,
+    {
+        Box::pin(async move {
+            let map_lane = &context.agent().map;
+            let stm = map_lane.update(event.current.to_string(), event.current.clone());
+
+            if let Err(e) = atomically(&stm, ExactlyOnce).await {
+                panic!("{:?}", e)
+            }
+        })
+    }
+}
+
+impl SwimAgent<AgentConfig> for StoreAgent {
+    fn instantiate<Context, Store>(
+        _config: &AgentConfig,
+        exec_conf: &AgentExecutionConfig,
+        store: Store,
+    ) -> (
+        Self,
+        DynamicLaneTasks<Self, Context>,
+        DynamicAgentIo<Context>,
+    )
+    where
+        Context: AgentContext<Self> + AgentExecutionContext + Send + Sync + 'static,
+        Store: NodeStore,
+    {
+        let LaneParts {
+            lane: value,
+            tasks: value_task,
+            io: value_lane_io,
+        } = agent::make_value_lane(
+            LaneConfig::new("value".to_string(), false, false),
+            exec_conf,
+            Default::default(),
+            ValueLifecycle,
+            |agent: &StoreAgent| &agent.value,
+            store.clone(),
+        );
+
+        let LaneParts {
+            lane: map,
+            tasks: map_task,
+            io: map_lane_io,
+        } = agent::make_map_lane(
+            "map",
+            false,
+            exec_conf,
+            MapLifecycle,
+            |agent: &StoreAgent| &agent.map,
+            false,
+            store,
+        );
+
+        let agent = StoreAgent { value, map };
+        let mut io = HashMap::new();
+        io.insert(
+            "value".to_string(),
+            IoPair {
+                routing: None,
+                persistence: value_lane_io.persistence,
+            },
+        );
+        io.insert(
+            "map".to_string(),
+            IoPair {
+                routing: None,
+                persistence: map_lane_io.persistence,
+            },
+        );
+
+        (agent, vec![value_task.boxed(), map_task.boxed()], io)
+    }
+}
+
+#[derive(Clone, Debug)]
+struct AgentConfig;
+
+struct TestStore {
+    _dir: Dir,
+    delegate: SwimPlaneStore<RocksDatabase>,
+    inner: Arc<RocksDatabase>,
+}
+
+impl Default for TestStore {
+    fn default() -> Self {
+        let temp_dir = Dir::transient("store_agent").unwrap();
+        let db = RocksOpts::default()
+            .build(temp_dir.path(), &default_keyspaces())
+            .unwrap();
+        let arcd_db = Arc::new(db);
+        let store = SwimPlaneStore::new(
+            "test",
+            arcd_db.clone(),
+            KeyStore::initialise_with(arcd_db.clone()),
+        );
+
+        TestStore {
+            _dir: temp_dir,
+            delegate: store,
+            inner: arcd_db,
+        }
+    }
+}
+
+#[tokio::test]
+async fn store_loads() {
+    let node_uri = "/test";
+
+    let store = TestStore::default();
+    let value_id = store
+        .delegate
+        .node_id_of(format!("{}/value", node_uri))
+        .unwrap();
+    let key = StoreKey::Value { lane_id: value_id };
+
+    store.delegate.put(key, ser!(i32::MAX)).unwrap();
+
+    let map_id = store
+        .delegate
+        .node_id_of(format!("{}/map", node_uri))
+        .unwrap();
+
+    let mut expected = HashMap::new();
+    for i in 0..10 {
+        expected.insert(i.to_string(), Arc::new(i));
+        let key = serialize(&i.to_string()).unwrap();
+        let key = StoreKey::Map {
+            lane_id: map_id,
+            key: Some(key),
+        };
+
+        store.delegate.put(key, ser!(i)).unwrap();
+    }
+
+    let provider = AgentProvider::new(AgentConfig, StoreAgentLifecycle);
+    let uri: RelativeUri = node_uri.parse().unwrap();
+    let buffer_size = NonZeroUsize::new(10).unwrap();
+    let clock = TestClock::default();
+
+    let exec_config = AgentExecutionConfig::with(
+        buffer_size,
+        1,
+        0,
+        Duration::from_secs(1),
+        None,
+        Duration::from_secs(1),
+    );
+    let (envelope_tx, envelope_rx) = mpsc::channel(buffer_size.get());
+    let (agent, agent_proc) = provider.run(
+        RouteAndParameters::new(uri.clone(), HashMap::new()),
+        exec_config,
+        clock.clone(),
+        ReceiverStream::new(envelope_rx),
+        SingleChannelRouter::new(RoutingAddr::local(1024)),
+        store.delegate.node_store(uri.to_string()),
+    );
+
+    let agent_task = swim_runtime::task::spawn(agent_proc);
+
+    match agent.downcast_ref::<StoreAgent>() {
+        Some(agent) => {
+            let value = agent.value.load().await;
+            assert_eq!(*value, i32::MAX);
+
+            let map = atomically(&agent.map.snapshot(), ExactlyOnce)
+                .await
+                .unwrap();
+
+            assert_eq!(expected, map);
+        }
+        None => {
+            panic!("Failed to get store agent");
+        }
+    }
+
+    drop(envelope_tx);
+    assert!(agent_task.await.is_ok());
+}
+
+#[tokio::test]
+async fn events() {
+    let provider = AgentProvider::new(AgentConfig, StoreAgentLifecycle);
+    let uri: RelativeUri = "/test".parse().unwrap();
+    let buffer_size = NonZeroUsize::new(10).unwrap();
+    let clock = TestClock::default();
+
+    let store = TestStore::default();
+
+    let exec_config = AgentExecutionConfig::with(
+        buffer_size,
+        1,
+        0,
+        Duration::from_secs(1),
+        None,
+        Duration::from_secs(1),
+    );
+    let (envelope_tx, envelope_rx) = mpsc::channel(buffer_size.get());
+    let (_, agent_proc) = provider.run(
+        RouteAndParameters::new(uri.clone(), HashMap::new()),
+        exec_config,
+        clock.clone(),
+        ReceiverStream::new(envelope_rx),
+        SingleChannelRouter::new(RoutingAddr::local(1024)),
+        store.delegate.node_store(uri.to_string()),
+    );
+
+    let agent_task = swim_runtime::task::spawn(agent_proc);
+
+    fn check_value_lane(store: &TestStore, i: i32) {
+        let delegate = &store.delegate;
+
+        let value_uri = "/test/value";
+        let value_id = delegate.node_id_of(value_uri).unwrap();
+        let value_key = StoreKey::Value { lane_id: value_id };
+
+        let value = delegate
+            .get(value_key)
+            .unwrap()
+            .expect("Missing value for value lane");
+        let deserialized = deserialize::<i32>(value.as_slice()).unwrap();
+        assert_eq!(i, deserialized);
+    }
+
+    fn check_map_lane(store: &TestStore, i: i32) {
+        let delegate = &store.delegate;
+
+        let map_uri = "/test/map";
+        let map_id = delegate.node_id_of(map_uri).unwrap();
+        let prefix = StoreKey::Map {
+            lane_id: map_id,
+            key: None,
+        };
+
+        let snapshot_opt = delegate
+            .get_prefix_range(prefix, |key, value| {
+                let store_key = deserialize::<StoreKey>(&key)?;
+
+                match store_key {
+                    StoreKey::Map { key, .. } => {
+                        let key = deserialize::<String>(&key.ok_or(StoreError::KeyNotFound)?)?;
+                        let value = deserialize::<i32>(&value)?;
+
+                        Ok((key, value))
+                    }
+                    StoreKey::Value { .. } => {
+                        Err(StoreError::Decoding("Expected a map key".to_string()))
+                    }
+                }
+            })
+            .expect("Snapshot failure");
+
+        match snapshot_opt {
+            Some(snapshot) => {
+                let actual: HashMap<String, i32> = snapshot.into_iter().collect();
+                let expected = (1..=i).into_iter().fold(HashMap::new(), |mut map, i| {
+                    map.insert(i.to_string(), i);
+                    map
+                });
+
+                assert_eq!(expected, actual);
+            }
+            None => {
+                panic!("Expected at least one entry in the snapshot")
+            }
+        }
+    }
+
+    clock.advance_when_blocked(INTERVAL).await;
+
+    for i in 1..MAX_PERIODS {
+        clock.advance_when_blocked(INTERVAL).await;
+        check_value_lane(&store, i);
+        check_map_lane(&store, i);
+    }
+
+    let counter = store
+        .inner
+        .get_keyspace(KeyspaceName::Lane, COUNTER_BYTES)
+        .unwrap()
+        .expect("Missing counter");
+    let counter_value = deserialize::<u64>(counter.as_slice()).unwrap();
+    assert_eq!(counter_value, 2);
+
+    drop(envelope_tx);
+    assert!(agent_task.await.is_ok());
+}
diff --git a/swim_server/src/meta/info/tests.rs b/swim_server/src/meta/info/tests.rs
index 4a226fde1..bc1d35e7c 100644
--- a/swim_server/src/meta/info/tests.rs
+++ b/swim_server/src/meta/info/tests.rs
@@ -15,9 +15,11 @@
 use crate::agent::lane::channels::AgentExecutionConfig;
 use crate::agent::lane::model::map::MapLane;
 use crate::agent::lane::model::value::ValueLane;
+use crate::agent::store::mock::MockNodeStore;
 use crate::agent::{agent_lifecycle, map_lifecycle, value_lifecycle, SwimAgent, TestClock};
 use crate::meta::info::{LaneInfo, LaneKind};
 use crate::plane::provider::AgentProvider;
+use crate::plane::RouteAndParameters;
 use crate::routing::error::RouterError;
 use crate::routing::{
     ConnectionDropped, Route, RoutingAddr, ServerRouter, TaggedEnvelope, TaggedSender,
@@ -182,12 +184,12 @@ async fn lane_info_sync() {
     let provider = AgentProvider::new(MockAgentConfig, MockAgentLifecycle);
 
     let (_a, agent_proc) = provider.run(
-        uri,
-        HashMap::new(),
+        RouteAndParameters::new(uri, HashMap::new()),
         exec_config,
         clock.clone(),
         ReceiverStream::new(envelope_rx),
         MockRouter::new(RoutingAddr::local(1024), tx),
+        MockNodeStore::mock(),
     );
 
     let _agent_task = swim_runtime::task::spawn(agent_proc);
diff --git a/swim_server/src/meta/log/tests.rs b/swim_server/src/meta/log/tests.rs
index 0962de869..417fed6f4 100644
--- a/swim_server/src/meta/log/tests.rs
+++ b/swim_server/src/meta/log/tests.rs
@@ -16,12 +16,14 @@ use crate::agent::lane::channels::AgentExecutionConfig;
 use crate::agent::lane::model::map::{MapLane, MapLaneEvent};
 use crate::agent::lane::model::supply::SupplyLane;
 use crate::agent::lane::model::value::ValueLane;
+use crate::agent::store::mock::MockNodeStore;
 use crate::agent::{
     agent_lifecycle, map_lifecycle, value_lifecycle, AgentContext, SwimAgent, TestClock,
 };
 use crate::meta::log::config::{FlushStrategy, LogConfig};
 use crate::meta::log::{LogBuffer, LogEntry, LogLanes, LogLevel, NodeLogger};
 use crate::plane::provider::AgentProvider;
+use crate::plane::RouteAndParameters;
 use crate::routing::error::RouterError;
 use crate::routing::{
     ConnectionDropped, Route, RoutingAddr, ServerRouter, TaggedEnvelope, TaggedSender,
@@ -149,12 +151,12 @@ async fn agent_log() {
     let provider = AgentProvider::new(MockAgentConfig, MockAgentLifecycle);
 
     let (_a, agent_proc) = provider.run(
-        uri,
-        HashMap::new(),
+        RouteAndParameters::new(uri, HashMap::new()),
         exec_config,
         clock.clone(),
         ReceiverStream::new(envelope_rx),
         MockRouter::new(RoutingAddr::local(1024), tx),
+        MockNodeStore::mock(),
     );
 
     let _agent_task = swim_runtime::task::spawn(agent_proc);
diff --git a/swim_server/src/plane/spec/tests.rs b/swim_server/src/plane/spec/tests.rs
index 6b5cdbba6..095bd11c0 100644
--- a/swim_server/src/plane/spec/tests.rs
+++ b/swim_server/src/plane/spec/tests.rs
@@ -15,12 +15,14 @@
 use crate::agent::context::AgentExecutionContext;
 use crate::agent::lane::channels::AgentExecutionConfig;
 use crate::agent::lifecycle::AgentLifecycle;
+use crate::agent::store::NodeStore;
 use crate::agent::{AgentContext, DynamicAgentIo, DynamicLaneTasks, SwimAgent};
 use crate::plane::context::PlaneContext;
 use crate::plane::error::AmbiguousRoutes;
 use crate::plane::lifecycle::PlaneLifecycle;
 use crate::plane::router::PlaneRouter;
 use crate::plane::spec::{PlaneBuilder, PlaneSpec, RouteSpec};
+use crate::plane::store::mock::MockPlaneStore;
 use crate::routing::error::RouterError;
 use crate::routing::{Route, RoutingAddr, ServerRouter, TaggedEnvelope};
 use futures::future::{ready, BoxFuture, Ready};
@@ -44,8 +46,12 @@ impl Clock for DummyClock {
     }
 }
 
-type BuilderType =
-    PlaneBuilder<DummyClock, ReceiverStream<TaggedEnvelope>, PlaneRouter<DummyDelegate>>;
+type BuilderType = PlaneBuilder<
+    DummyClock,
+    ReceiverStream<TaggedEnvelope>,
+    PlaneRouter<DummyDelegate>,
+    MockPlaneStore,
+>;
 
 #[derive(Debug)]
 struct DummyAgent;
@@ -80,9 +86,10 @@ impl ServerRouter for DummyDelegate {
 }
 
 impl SwimAgent<DummyConfig> for DummyAgent {
-    fn instantiate<Context>(
+    fn instantiate<Context, Store>(
         _configuration: &DummyConfig,
         _exec_conf: &AgentExecutionConfig,
+        _store: Store,
     ) -> (
         Self,
         DynamicLaneTasks<Self, Context>,
@@ -90,6 +97,7 @@ impl SwimAgent<DummyConfig> for DummyAgent {
     )
     where
         Context: AgentContext<Self> + AgentExecutionContext + Send + Sync + 'static,
+        Store: NodeStore,
     {
         panic!("Called unexpectedly.");
     }
@@ -118,12 +126,14 @@ impl PlaneLifecycle for DummyPlaneLifecycle {
 fn plane_builder_single_route() {
     let pat = RoutePattern::parse_str("/:id").unwrap();
 
-    let mut builder: BuilderType = PlaneBuilder::new();
+    let mut builder: BuilderType = PlaneBuilder::new(MockPlaneStore);
     assert!(builder
         .add_route(pat.clone(), DummyConfig(1), DummyLifecycle(1))
         .is_ok());
 
-    let PlaneSpec { routes, lifecycle } = builder.build();
+    let PlaneSpec {
+        routes, lifecycle, ..
+    } = builder.build();
     assert!(lifecycle.is_none());
     assert_eq!(routes.len(), 1);
     let RouteSpec {
@@ -143,7 +153,7 @@ fn plane_builder_two_routes() {
     let pat1 = RoutePattern::parse_str("/a").unwrap();
     let pat2 = RoutePattern::parse_str("/b").unwrap();
 
-    let mut builder: BuilderType = PlaneBuilder::new();
+    let mut builder: BuilderType = PlaneBuilder::new(MockPlaneStore);
     assert!(builder
         .add_route(pat1.clone(), DummyConfig(1), DummyLifecycle(1))
         .is_ok());
@@ -151,7 +161,9 @@ fn plane_builder_two_routes() {
         .add_route(pat2.clone(), DummyConfig(2), DummyLifecycle(2))
         .is_ok());
 
-    let PlaneSpec { routes, lifecycle } = builder.build();
+    let PlaneSpec {
+        routes, lifecycle, ..
+    } = builder.build();
     assert!(lifecycle.is_none());
     assert_eq!(routes.len(), 2);
     let RouteSpec {
@@ -182,7 +194,7 @@ fn plane_builder_route_collision() {
     let pat1 = RoutePattern::parse_str("/:id").unwrap();
     let pat2 = RoutePattern::parse_str("/b").unwrap();
 
-    let mut builder: BuilderType = PlaneBuilder::new();
+    let mut builder: BuilderType = PlaneBuilder::new(MockPlaneStore);
     assert!(builder
         .add_route(pat1.clone(), DummyConfig(1), DummyLifecycle(1))
         .is_ok());
@@ -199,14 +211,16 @@ fn plane_builder_route_collision() {
 fn add_plane_lifecycle() {
     let pat = RoutePattern::parse_str("/:id").unwrap();
 
-    let mut builder: BuilderType = PlaneBuilder::new();
+    let mut builder: BuilderType = PlaneBuilder::new(MockPlaneStore);
     assert!(builder
         .add_route(pat.clone(), DummyConfig(1), DummyLifecycle(1))
         .is_ok());
 
     let plane_lifecycle = DummyPlaneLifecycle(5);
 
-    let PlaneSpec { routes, lifecycle } = builder.build_with_lifecycle(plane_lifecycle.boxed());
+    let PlaneSpec {
+        routes, lifecycle, ..
+    } = builder.build_with_lifecycle(plane_lifecycle.boxed());
     assert!(lifecycle.is_some());
 
     let lc = lifecycle.unwrap();
diff --git a/swim_server/src/plane/tests/fixture.rs b/swim_server/src/plane/tests/fixture.rs
index 1d04da5c7..c84c88446 100644
--- a/swim_server/src/plane/tests/fixture.rs
+++ b/swim_server/src/plane/tests/fixture.rs
@@ -12,13 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::agent::dispatch::error::DispatcherErrors;
 use crate::agent::lane::channels::AgentExecutionConfig;
-use crate::agent::AgentResult;
+use crate::agent::store::SwimNodeStore;
+use crate::agent::{AgentResult, AgentTaskResult};
 use crate::plane::context::PlaneContext;
 use crate::plane::lifecycle::PlaneLifecycle;
 use crate::plane::router::PlaneRouter;
-use crate::plane::{AgentRoute, EnvChannel};
+use crate::plane::store::mock::MockPlaneStore;
+use crate::plane::{AgentRoute, EnvChannel, RouteAndParameters};
 use crate::routing::{ServerRouter, TaggedEnvelope};
 use futures::future::BoxFuture;
 use futures::{FutureExt, StreamExt};
@@ -26,7 +27,6 @@ use http::Uri;
 use parking_lot::Mutex;
 use pin_utils::pin_mut;
 use std::any::Any;
-use std::collections::HashMap;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::time::Duration;
@@ -89,6 +89,7 @@ pub fn make_config() -> AgentExecutionConfig {
         node_log: Default::default(),
         metrics: Default::default(),
         max_idle_time: Duration::from_secs(60),
+        max_store_errors: 0,
     }
 }
 
@@ -99,22 +100,25 @@ const LANE_NAME: &str = "receiver_lane";
 const MESSAGE: &str = "ping!";
 
 impl<Clk: Clock, Delegate: ServerRouter + 'static>
-    AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>> for SendAgentRoute
+    AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>, SwimNodeStore<MockPlaneStore>>
+    for SendAgentRoute
 {
     fn run_agent(
         &self,
-        uri: RelativeUri,
-        parameters: HashMap<String, String>,
+        route: RouteAndParameters,
         execution_config: AgentExecutionConfig,
         _clock: Clk,
         incoming_envelopes: EnvChannel,
         mut router: PlaneRouter<Delegate>,
+        _store: SwimNodeStore<MockPlaneStore>,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>) {
+        let RouteAndParameters { route, parameters } = route;
+
         let id = parameters[PARAM_NAME].clone();
         let target = self.0.clone();
         let agent = Arc::new(SendAgent(id.clone()));
         let expected_route: RelativeUri = format!("/{}/{}", SENDER_PREFIX, id).parse().unwrap();
-        assert_eq!(uri, expected_route);
+        assert_eq!(route, expected_route);
         assert_eq!(execution_config, make_config());
         let task = async move {
             let target_node: RelativeUri =
@@ -132,9 +136,9 @@ impl<Clk: Clock, Delegate: ServerRouter + 'static>
             pin_mut!(incoming_envelopes);
             while incoming_envelopes.next().await.is_some() {}
             AgentResult {
-                route: uri,
-                dispatcher_errors: DispatcherErrors::default(),
-                failed: false,
+                route,
+                dispatcher_task: AgentTaskResult::default(),
+                store_task: AgentTaskResult::default(),
             }
         }
         .boxed();
@@ -142,18 +146,21 @@ impl<Clk: Clock, Delegate: ServerRouter + 'static>
     }
 }
 
-impl<Clk: Clock, Delegate> AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>>
+impl<Clk: Clock, Delegate>
+    AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>, SwimNodeStore<MockPlaneStore>>
     for ReceiveAgentRoute
 {
     fn run_agent(
         &self,
-        uri: RelativeUri,
-        parameters: HashMap<String, String>,
+        route: RouteAndParameters,
         execution_config: AgentExecutionConfig,
         _clock: Clk,
         incoming_envelopes: EnvChannel,
         _router: PlaneRouter<Delegate>,
+        _store: SwimNodeStore<MockPlaneStore>,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>) {
+        let RouteAndParameters { route, parameters } = route;
+
         let ReceiveAgentRoute { expected_id, done } = self;
         let mut done_sender = done.lock().take();
         assert!(done_sender.is_some());
@@ -163,7 +170,7 @@ impl<Clk: Clock, Delegate> AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>>
         assert_eq!(id, expected_target);
         let agent = Arc::new(ReceiveAgent(id.clone()));
         let expected_route: Uri = format!("/{}/{}", RECEIVER_PREFIX, id).parse().unwrap();
-        assert_eq!(uri, expected_route);
+        assert_eq!(route, expected_route);
         assert_eq!(execution_config, make_config());
         let task = async move {
             pin_mut!(incoming_envelopes);
@@ -187,9 +194,12 @@ impl<Clk: Clock, Delegate> AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>>
                 }
             }
             AgentResult {
-                route: uri,
-                dispatcher_errors: DispatcherErrors::default(),
-                failed: times_seen != 1,
+                route,
+                dispatcher_task: AgentTaskResult {
+                    errors: Default::default(),
+                    failed: times_seen != 1,
+                },
+                store_task: AgentTaskResult::default(),
             }
         }
         .boxed();
diff --git a/swim_server/src/plane/tests/mod.rs b/swim_server/src/plane/tests/mod.rs
index 9543fda78..dca055d69 100644
--- a/swim_server/src/plane/tests/mod.rs
+++ b/swim_server/src/plane/tests/mod.rs
@@ -15,6 +15,7 @@
 use crate::plane::lifecycle::PlaneLifecycle;
 use crate::plane::router::PlaneRouter;
 use crate::plane::spec::{PlaneSpec, RouteSpec};
+use crate::plane::store::mock::MockPlaneStore;
 use crate::plane::tests::fixture::{ReceiveAgentRoute, SendAgentRoute, TestLifecycle};
 use crate::plane::{AgentRoute, EnvChannel};
 use crate::routing::{ServerRouter, TopLevelRouterFactory};
@@ -30,7 +31,7 @@ use utilities::sync::trigger;
 mod fixture;
 
 fn make_spec<Clk: Clock, Delegate: ServerRouter + 'static>() -> (
-    PlaneSpec<Clk, EnvChannel, PlaneRouter<Delegate>>,
+    PlaneSpec<Clk, EnvChannel, PlaneRouter<Delegate>, MockPlaneStore>,
     trigger::Receiver,
 ) {
     let send_pattern = RoutePattern::parse(
@@ -59,6 +60,7 @@ fn make_spec<Clk: Clock, Delegate: ServerRouter + 'static>() -> (
         PlaneSpec {
             routes: vec![sender, reciever],
             lifecycle: Some(lifecycle.boxed()),
+            store: MockPlaneStore,
         },
         rx,
     )
diff --git a/swim_server/store/src/engines/rocks/tests.rs b/swim_server/store/src/engines/rocks/tests.rs
new file mode 100644
index 000000000..b249a9819
--- /dev/null
+++ b/swim_server/store/src/engines/rocks/tests.rs
@@ -0,0 +1,307 @@
+// Copyright 2015-2021 SWIM.AI inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::engines::{RocksEngine, RocksOpts, StoreBuilder};
+use crate::iterator::{EngineIterator, EngineRefIterator};
+use crate::keyspaces::{Keyspace, KeyspaceByteEngine, KeyspaceDef, KeyspaceResolver, Keyspaces};
+
+use crate::{deserialize, deserialize_key, serialize, StoreError};
+use rocksdb::{MergeOperands, Options, SliceTransform};
+use std::collections::HashMap;
+use std::mem::size_of;
+use std::ops::{Deref, Range};
+use tempdir::TempDir;
+
+impl Deref for TransientDatabase {
+    type Target = RocksEngine;
+
+    fn deref(&self) -> &Self::Target {
+        &self.delegate
+    }
+}
+
+pub struct TransientDatabase {
+    _dir: TempDir,
+    delegate: RocksEngine,
+}
+
+impl TransientDatabase {
+    fn new(keyspaces: Keyspaces<RocksOpts>) -> TransientDatabase {
+        let dir = TempDir::new("test").expect("Failed to create temporary directory");
+        let delegate = RocksOpts::default()
+            .build(dir.path(), &keyspaces)
+            .expect("Failed to build delegate store");
+
+        TransientDatabase {
+            _dir: dir,
+            delegate,
+        }
+    }
+}
+
+fn default_lane_opts() -> Options {
+    let mut opts = rocksdb::Options::default();
+    opts.set_prefix_extractor(SliceTransform::create_fixed_prefix(size_of::<u64>()));
+    opts.set_memtable_prefix_bloom_ratio(0.2);
+
+    opts
+}
+
+pub(crate) fn incrementing_merge_operator(
+    _new_key: &[u8],
+    existing_value: Option<&[u8]>,
+    operands: &mut MergeOperands,
+) -> Option<Vec<u8>> {
+    let mut value = match existing_value {
+        Some(bytes) => deserialize_key(bytes).unwrap(),
+        None => 0,
+    };
+
+    for op in operands {
+        let deserialized = deserialize_key(op).unwrap();
+        value += deserialized;
+    }
+    Some(serialize(&value).unwrap())
+}
+
+fn default_db() -> TransientDatabase {
+    let mut lane_opts = rocksdb::Options::default();
+    lane_opts.set_merge_operator_associative("lane_id_counter", incrementing_merge_operator);
+
+    let keyspaces = vec![
+        KeyspaceDef::new(KeyspaceName::Value.name(), RocksOpts(default_lane_opts())),
+        KeyspaceDef::new(KeyspaceName::Map.name(), RocksOpts(default_lane_opts())),
+        KeyspaceDef::new(KeyspaceName::Lane.name(), RocksOpts(lane_opts)),
+    ];
+
+    TransientDatabase::new(Keyspaces::new(keyspaces))
+}
+
+fn assert_keyspaces_empty(db: &TransientDatabase, spaces: &[KeyspaceName]) {
+    for key_space in spaces {
+        let resolved = db.resolve_keyspace(key_space).unwrap();
+        let iter = db.iterator(resolved).unwrap();
+        assert_eq!(Ok(false), iter.valid())
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+enum KeyspaceName {
+    Value,
+    Map,
+    Lane,
+}
+
+impl Keyspace for KeyspaceName {
+    fn name(&self) -> &str {
+        match self {
+            KeyspaceName::Value => "value",
+            KeyspaceName::Map => "map",
+            KeyspaceName::Lane => "default",
+        }
+    }
+}
+
+#[test]
+fn get_keyspace() {
+    let db = default_db();
+
+    let key = b"test_key";
+    let value = b"test_value";
+
+    assert!(db.put_keyspace(KeyspaceName::Value, key, value).is_ok());
+    assert_keyspaces_empty(&db, &[KeyspaceName::Lane, KeyspaceName::Map]);
+}
+
+fn format_key(id: i32) -> String {
+    format!("key/{}", id)
+}
+
+fn populate_keyspace(
+    db: &TransientDatabase,
+    space: KeyspaceName,
+    range: Range<i32>,
+    clone_to: &mut HashMap<String, i32>,
+) {
+    for i in range {
+        let key = format_key(i);
+        clone_to.insert(key.clone(), i);
+        assert!(db
+            .put_keyspace(space, key.as_bytes(), serialize(&i).unwrap().as_slice())
+            .is_ok());
+    }
+}
+
+#[test]
+fn engine_iterator() {
+    let db = default_db();
+    let range = 0..100;
+    let mut expected = HashMap::new();
+
+    populate_keyspace(&db, KeyspaceName::Value, range.clone(), &mut expected);
+
+    let resolved = db.resolve_keyspace(&KeyspaceName::Value).unwrap();
+    let mut iter = db.iterator(resolved).unwrap();
+
+    assert_eq!(iter.seek_first(), Ok(true));
+
+    for _ in range {
+        let valid = iter.valid().expect("Invalid iterator");
+        if valid {
+            match (iter.key(), iter.value()) {
+                (Some(key), Some(value)) => {
+                    let key = String::from_utf8(key.to_vec()).unwrap();
+                    let value = deserialize::<i32>(value).unwrap();
+
+                    match expected.remove(&key) {
+                        Some(expected_value) => {
+                            assert_eq!(expected_value, value)
+                        }
+                        None => {
+                            panic!("Unexpected key: {:?}", key)
+                        }
+                    }
+                }
+                e => {
+                    panic!("Inconsistent state: {:?}", e);
+                }
+            }
+            iter.seek_next();
+        } else {
+            panic!("Invalid iterator");
+        }
+    }
+
+    assert!(expected.is_empty());
+    assert_keyspaces_empty(&db, &[KeyspaceName::Lane, KeyspaceName::Map]);
+}
+
+#[test]
+pub fn crud() {
+    let db = default_db();
+
+    let key = b"key_a";
+    let value_1 = b"value_a";
+    let value_2 = b"value_b";
+
+    assert!(db.put_keyspace(KeyspaceName::Value, key, value_1).is_ok());
+
+    let get_result = db.get_keyspace(KeyspaceName::Value, key);
+    assert!(matches!(get_result, Ok(Some(_))));
+    let get_value = get_result.unwrap().unwrap();
+    assert_eq!(value_1, String::from_utf8(get_value).unwrap().as_bytes());
+
+    let update_result = db.put_keyspace(KeyspaceName::Value, key, value_2);
+    assert!(update_result.is_ok());
+
+    let get_result = db.get_keyspace(KeyspaceName::Value, key);
+    assert!(matches!(get_result, Ok(Some(_))));
+    let get_value = get_result.unwrap().unwrap();
+    assert_eq!(value_2, String::from_utf8(get_value).unwrap().as_bytes());
+
+    let delete_result = db.delete_keyspace(KeyspaceName::Value, key);
+    assert!(matches!(delete_result, Ok(())));
+}
+
+#[test]
+pub fn get_missing() {
+    let db = default_db();
+    let get_result = db.get_keyspace(KeyspaceName::Value, b"key_a");
+    assert!(matches!(get_result, Ok(None)));
+}
+
+#[test]
+pub fn delete_missing() {
+    let db = default_db();
+    let get_result = db.delete_keyspace(KeyspaceName::Value, b"key_a");
+    assert!(matches!(get_result, Ok(())));
+}
+
+fn map_fn<'a>(key: &'a [u8], value: &'a [u8]) -> Result<(String, String), StoreError> {
+    let k = String::from_utf8(key.to_vec()).unwrap();
+    let v = String::from_utf8(value.to_vec()).unwrap();
+
+    Ok((k, v))
+}
+
+#[test]
+pub fn empty_range() {
+    let db = default_db();
+    let result = db.get_prefix_range(KeyspaceName::Value, b"prefix", map_fn);
+    match result {
+        Ok(ss) => {
+            assert!(ss.is_none());
+        }
+        Err(e) => panic!("{:?}", e),
+    }
+}
+
+#[test]
+pub fn prefix_range() {
+    let db = default_db();
+    let prefix = "/foo/bar";
+    let limit = 256;
+
+    let format = |i| format!("{}/{}", prefix, i);
+    let mut expected = HashMap::with_capacity(limit);
+
+    // In range records
+    for i in 0..limit {
+        let key = format(i);
+        let value = i.to_string();
+        let result = db.put_keyspace(
+            KeyspaceName::Value,
+            key.as_bytes(),
+            i.to_string().as_bytes(),
+        );
+
+        assert!(result.is_ok());
+
+        expected.insert(key, value);
+    }
+
+    // Out of range records
+    for i in 0..limit {
+        let key = format!("/foo/{}", i);
+        let value = i.to_string();
+        let result = db.put_keyspace(KeyspaceName::Value, key.as_bytes(), value.as_bytes());
+
+        assert!(result.is_ok());
+    }
+
+    let result = db.get_prefix_range(KeyspaceName::Value, prefix.as_bytes(), map_fn);
+    assert!(matches!(result, Ok(Some(_))));
+
+    let result = result.unwrap().unwrap();
+    let mut iter = result.into_iter();
+
+    for i in 0..limit {
+        match iter.next() {
+            Some((key, value)) => match expected.remove(&key) {
+                Some(expected_value) => {
+                    assert_eq!(expected_value, value);
+                }
+                None => {
+                    panic!("Missing key: `{}`", format(i));
+                }
+            },
+            None => {
+                panic!("Missing key: `{}`", format(i));
+            }
+        }
+    }
+
+    assert!(iter.next().is_none());
+    assert!(expected.is_empty());
+}