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

Issue:
Server store implementation



Discussion / hints:

Fix patch:
diff --git a/swim_common/src/model/time/mod.rs b/swim_common/src/model/time/mod.rs
index fb260cb06..c058dcd8e 100644
--- a/swim_common/src/model/time/mod.rs
+++ b/swim_common/src/model/time/mod.rs
@@ -16,7 +16,7 @@ use crate::form::{Form, FormErr, ValidatedForm};
 use crate::model::schema::StandardSchema;
 use crate::model::{Value, 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)]
@@ -28,6 +28,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 df5c9dd15..cc86f626a 100644
--- a/swim_server/Cargo.toml
+++ b/swim_server/Cargo.toml
@@ -46,6 +46,9 @@ 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 4c17f92a7..abc2666aa 100644
--- a/swim_server/agent_derive/src/agent.rs
+++ b/swim_server/agent_derive/src/agent.rs
@@ -131,25 +131,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<Store>>>>
             )
                 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<Store>>>> = HashMap::new();
 
                 #(#lifecycles_ast)*
 
@@ -269,7 +272,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,
@@ -285,7 +288,7 @@ 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,
@@ -305,7 +308,7 @@ 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))), Some(Box::new(LaneNoStore)))
                     );
                 },
                 model,
@@ -322,7 +325,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))), Some(Box::new(LaneNoStore)))
                     );
                 },
                 model,
@@ -438,7 +442,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 +475,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 41e4d8f5b..ab3b1acee 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::FormErr;
 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/store/error.rs b/swim_server/src/agent/lane/store/error.rs
new file mode 100644
index 000000000..a26744d94
--- /dev/null
+++ b/swim_server/src/agent/lane/store/error.rs
@@ -0,0 +1,109 @@
+// 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;
+
+        if is_operational(&error) {
+            errors.push(error);
+
+            let len = errors.len();
+            let errors: Vec<StoreTaskError> = errors.drain(0..len).collect();
+
+            Err(LaneStoreErrorReport { errors })
+        } else {
+            errors.push(error);
+
+            let len = errors.len();
+
+            if 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..5ca0fe25b
--- /dev/null
+++ b/swim_server/src/agent/lane/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::lane::store::error::{LaneStoreErrorReport, StoreErrorHandler};
+use futures::future::{ready, BoxFuture};
+
+#[cfg(test)]
+mod tests;
+
+use crate::agent::store::NodeStore;
+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<Store: NodeStore + Sized>: 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.
+    /// `lane_uri`: the URI of the lane.
+    /// `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,
+        store: Store,
+        lane_uri: String,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>>;
+
+    fn attach_boxed(
+        self: Box<Self>,
+        store: Store,
+        lane_uri: String,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>>;
+}
+
+/// A transient lane store which completes immediately.
+#[derive(Debug)]
+pub struct LaneNoStore;
+impl<Store> StoreIo<Store> for LaneNoStore
+where
+    Store: NodeStore,
+{
+    fn attach(
+        self,
+        _store: Store,
+        _lane_uri: String,
+        _error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        ready(Ok(())).boxed()
+    }
+
+    fn attach_boxed(
+        self: Box<Self>,
+        store: Store,
+        lane_uri: String,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        (*self).attach(store, lane_uri, 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..b5c8b4a05
--- /dev/null
+++ b/swim_server/src/agent/lane/store/task.rs
@@ -0,0 +1,164 @@
+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<Store>>>,
+        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 node_store = node_store.clone();
+
+            let task = async move {
+                let result = io
+                    .attach_boxed(node_store, lane_uri.clone(), store_error_handler)
+                    .await;
+                (lane_uri.clone(), 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 e06f35f00..a78265428 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;
@@ -76,32 +77,40 @@ use utilities::future::SwimStreamExt;
 use utilities::sync::{topic, trigger};
 use utilities::uri::RelativeUri;
 
+use crate::agent::lane::store::task::{NodeStoreErrors, NodeStoreTask};
+pub use crate::agent::lane::store::{LaneNoStore, StoreIo};
+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>,
+        DynamicAgentIo<Context, Store>,
     )
     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, Store> =
+    HashMap<String, IoPair<Box<dyn LaneIo<Context>>, Box<dyn StoreIo<Store>>>>;
 
 pub const COMMANDED: &str = "Command received";
 pub const ON_COMMAND: &str = "On command handler";
@@ -114,28 +123,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 +203,21 @@ impl<Config> AgentParameters<Config> {
     }
 }
 
+/// Lane IO pair consisting of routing IO and store IO.
+pub struct IoPair<Routing, Store> {
+    routing: Option<Routing>,
+    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 +229,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 +245,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 +256,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 +277,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,7 +290,13 @@ 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)
@@ -248,7 +312,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 +327,42 @@ where
             );
         }
 
-        let mut io_providers = io_providers
-            .into_iter()
-            .map(|(k, v)| (LaneIdentifier::agent(k), v))
-            .collect::<HashMap<_, _>>();
+        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);
 
-        io_providers.extend(meta_io);
+        let (store_result_tx, store_result_rx) = oneshot::channel();
+        let max_store_errors = execution_config.max_store_errors;
+        let store_task = async move {
+            let task = NodeStoreTask::new(stop_trigger.clone(), 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);
 
         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 +370,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 +384,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)
@@ -807,23 +899,28 @@ 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>(
+/// * `transient` - Whether to persist the lane's state
+pub fn make_value_lane<Agent, Context, T, L, Store, P>(
     name: impl Into<String>,
     is_public: bool,
     config: &AgentExecutionConfig,
     init: T,
     lifecycle: L,
-    projection: impl Fn(&Agent) -> &ValueLane<T> + Send + Sync + 'static,
+    projection: P,
+    transient: bool,
+    _store: Store,
 ) -> (
     ValueLane<T>,
     impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
+    IoPair<impl LaneIo<Context>, impl StoreIo<Store>>,
 )
 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);
 
@@ -840,7 +937,19 @@ where
         projection,
     });
 
-    (lane, tasks, lane_io)
+    // todo tk: this block will be removed in the next PRs. LaneNoStore is used just so the type isn't opaque
+    let store_io = if transient {
+        Some(LaneNoStore)
+    } else {
+        unimplemented!()
+    };
+
+    let io = IoPair {
+        routing: lane_io,
+        persistence: store_io,
+    };
+
+    (lane, tasks, io)
 }
 
 impl<L, S, P> Lane for MapLifecycleTasks<L, S, P> {
@@ -904,23 +1013,28 @@ 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,
+    projection: P,
+    transient: bool,
+    _store: Store,
 ) -> (
     MapLane<K, V>,
     impl LaneTasks<Agent, Context>,
-    Option<impl LaneIo<Context>>,
+    IoPair<impl LaneIo<Context>, impl StoreIo<Store>>,
 )
 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);
 
@@ -940,7 +1054,19 @@ where
         projection,
     });
 
-    (lane, tasks, lane_io)
+    // todo tk: this block will be removed in the next PRs. LaneNoStore is used just so the type isn't opaque
+    let store_io = if transient {
+        Some(LaneNoStore)
+    } else {
+        unimplemented!()
+    };
+
+    let io = IoPair {
+        routing: lane_io,
+        persistence: store_io,
+    };
+
+    (lane, tasks, io)
 }
 
 impl<L, S, P> Lane for ActionLifecycleTasks<L, S, P> {
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..045f114f5
--- /dev/null
+++ b/swim_server/src/agent/store/mock/mod.rs
@@ -0,0 +1,55 @@
+// 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(),
+        }
+    }
+}
+
+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..1318783ac
--- /dev/null
+++ b/swim_server/src/agent/store/mod.rs
@@ -0,0 +1,119 @@
+// 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;
+#[cfg(test)]
+mod tests;
+
+/// 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;
+}
+
+/// 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(),
+        }
+    }
+
+    /// 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.
+    pub 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)
+    }
+}
+
+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()
+    }
+}
diff --git a/swim_server/src/interface/mod.rs b/swim_server/src/interface/mod.rs
index b644e8c80..c34b0a205 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::transient_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::transient_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,34 @@ impl SwimServerBuilder {
             },
         ))
     }
+
+    pub fn transient_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 +281,7 @@ impl SwimServer {
             mut planes,
             stop_trigger_rx,
             address_tx,
+            _store,
         } = self;
 
         let SwimServerConfig {
@@ -301,6 +345,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 +355,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 +449,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 dd613c2c0..ebf4d87d7 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::*;
diff --git a/swim_server/src/plane/mod.rs b/swim_server/src/plane/mod.rs
index 113d38749..3298ae785 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,7 +58,7 @@ 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
@@ -75,9 +77,10 @@ trait AgentRoute<Clk, Envelopes, Router>: Debug + Send {
         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 +88,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 +269,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 +292,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 +318,7 @@ impl<Clk: Clock, DelegateFac: ServerRouterFactory> RouteResolver<Clk, DelegateFa
             clock,
             execution_config,
             routes,
+            store,
             router_fac,
             stop_trigger,
             active_routes,
@@ -314,6 +337,7 @@ impl<Clk: Clock, DelegateFac: ServerRouterFactory> RouteResolver<Clk, DelegateFa
             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 +375,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 +386,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 +402,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 +419,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 +536,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 +584,16 @@ 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>;
 
 /// 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..fc25c7aec 100644
--- a/swim_server/src/plane/provider/mod.rs
+++ b/swim_server/src/plane/provider/mod.rs
@@ -14,6 +14,7 @@
 
 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::routing::{ServerRouter, TaggedEnvelope};
@@ -62,7 +63,7 @@ where
         }
     }
 
-    pub(crate) fn run<Clk, Envelopes, Router>(
+    pub(crate) fn run<Clk, Envelopes, Router, Store>(
         &self,
         uri: RelativeUri,
         parameters: HashMap<String, String>,
@@ -70,11 +71,13 @@ where
         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,
@@ -91,13 +94,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,6 +109,7 @@ 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,
@@ -114,6 +119,7 @@ where
         clock: Clk,
         incoming_envelopes: Envelopes,
         router: Router,
+        store: Store,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>) {
         self.run(
             uri,
@@ -122,6 +128,7 @@ where
             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..fd571541c
--- /dev/null
+++ b/swim_server/src/plane/store/mock.rs
@@ -0,0 +1,87 @@
+// 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::keystore::{KeyRequest, KeystoreTask};
+use crate::store::{StoreEngine, StoreKey};
+use futures::future::BoxFuture;
+use futures::{FutureExt, Stream};
+use std::sync::Arc;
+use store::keyspaces::KeyspaceByteEngine;
+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 lane_id_of<I>(&self, _lane: I) -> BoxFuture<'static, u64>
+    where
+        I: Into<String>,
+    {
+        async { 0 }.boxed()
+    }
+}
+
+impl KeystoreTask for MockPlaneStore {
+    fn run<DB, S>(_db: Arc<DB>, _events: S) -> BoxFuture<'static, Result<(), StoreError>>
+    where
+        DB: KeyspaceByteEngine,
+        S: Stream<Item = KeyRequest> + Unpin + Send + 'static,
+    {
+        Box::pin(async { Ok(()) })
+    }
+}
+
+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..280f94ca6
--- /dev/null
+++ b/swim_server/src/plane/store/mod.rs
@@ -0,0 +1,244 @@
+// 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::num::NonZeroUsize;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use futures::future::BoxFuture;
+use futures::{FutureExt, Stream};
+
+use store::keyspaces::{KeyspaceByteEngine, Keyspaces};
+use store::{serialize, EngineInfo, Store, StoreBuilder, StoreError};
+use swim_common::model::text::Text;
+
+use crate::agent::store::{NodeStore, SwimNodeStore};
+use crate::store::keystore::{KeyRequest, KeyStore, KeystoreTask};
+use crate::store::{KeyspaceName, StoreEngine, StoreKey};
+
+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 + KeystoreTask + 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 lane_id_of<I>(&self, lane: I) -> BoxFuture<u64>
+    where
+        I: Into<String>;
+}
+
+/// A store engine for planes.
+///
+/// Backed by a value store and a map store, any operations on this store have their key variant
+/// checked and the operation is delegated to the corresponding store.
+pub struct SwimPlaneStore<D> {
+    /// The name of the plane.
+    plane_name: Text,
+    /// Delegate byte engine.
+    delegate: Arc<D>,
+    keystore: KeyStore,
+}
+
+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 lane_id_of<I>(&self, lane: I) -> BoxFuture<u64>
+    where
+        I: Into<String>,
+    {
+        self.keystore.id_for(lane.into()).boxed()
+    }
+}
+
+impl<D: Store> KeystoreTask for SwimPlaneStore<D> {
+    fn run<DB, S>(_db: Arc<DB>, _events: S) -> BoxFuture<'static, Result<(), StoreError>>
+    where
+        DB: KeyspaceByteEngine,
+        S: Stream<Item = KeyRequest> + Unpin + Send + 'static,
+    {
+        unimplemented!()
+    }
+}
+
+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,
+    D::Store: KeystoreTask,
+{
+    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);
+    // todo config
+    let keystore = KeyStore::new(arcd_delegate.clone(), NonZeroUsize::new(8).unwrap());
+
+    Ok(SwimPlaneStore::new(plane_name, arcd_delegate, keystore))
+}
+
+impl<D> SwimPlaneStore<D>
+where
+    D: Store + KeystoreTask,
+{
+    pub(crate) fn new<I: Into<Text>>(
+        plane_name: I,
+        delegate: Arc<D>,
+        keystore: KeyStore,
+    ) -> 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..cd6f7eff3
--- /dev/null
+++ b/swim_server/src/store/keystore.rs
@@ -0,0 +1,237 @@
+// 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 futures::future::BoxFuture;
+use futures::{Stream, StreamExt};
+use rocksdb::MergeOperands;
+use std::num::NonZeroUsize;
+use std::sync::Arc;
+use store::keyspaces::KeyspaceByteEngine;
+use store::{serialize, StoreError};
+use tokio::sync::{mpsc, oneshot};
+use tokio::task::JoinHandle;
+use tokio_stream::wrappers::ReceiverStream;
+use tracing::{event, Level};
+
+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 RESPONSE_FAILURE: &str = "Failed to send response";
+
+/// 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;
+
+pub trait KeystoreTask {
+    fn run<DB, S>(db: Arc<DB>, events: S) -> BoxFuture<'static, Result<(), StoreError>>
+    where
+        DB: KeyspaceByteEngine,
+        S: Stream<Item = KeyRequest> + Unpin + Send + 'static;
+}
+
+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())
+}
+
+/// A keystore for assigning unique identifiers to lane addresses.
+#[derive(Clone)]
+pub struct KeyStore {
+    tx: mpsc::Sender<KeyRequest>,
+    task: Arc<JoinHandle<()>>,
+}
+
+impl KeyStore {
+    /// Produces a new keystore which will delegate its operations to `db` and will have an internal
+    /// task communication buffer size of `buffer_size`.
+    pub fn new<S>(db: Arc<S>, buffer_size: NonZeroUsize) -> KeyStore
+    where
+        S: KeyspaceByteEngine + KeystoreTask,
+    {
+        let (tx, rx) = mpsc::channel(buffer_size.get());
+        let task = async move {
+            if let Err(e) = S::run(db, ReceiverStream::new(rx)).await {
+                event!(Level::ERROR, "Keystore failed with: {:?}", e);
+            }
+        };
+
+        KeyStore {
+            tx,
+            task: Arc::new(tokio::spawn(task)),
+        }
+    }
+
+    /// Returns a unique identifier that has been assigned to the `lane_id`. This ID must be a
+    /// well-formed String of `/node_uri/lane_uri` for this host.
+    pub async fn id_for<I>(&self, lane_id: I) -> u64
+    where
+        I: Into<String>,
+    {
+        let (tx, rx) = oneshot::channel();
+        self.tx
+            .send((lane_id.into(), tx))
+            .await
+            .expect("Failed to make lane ID request");
+        rx.await.expect("No response received for lane ID")
+    }
+}
+
+#[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))
+}
+
+pub async fn lane_key_task<DB, S>(db: Arc<DB>, events: S) -> Result<(), StoreError>
+where
+    DB: KeyspaceByteEngine,
+    S: Stream<Item = KeyRequest> + Unpin + Send + 'static,
+{
+    let mut requests = events.fuse();
+
+    while let Some((uri, responder)) = requests.next().await {
+        let prefixed = format_key(uri);
+
+        match db.get_keyspace(KeyspaceName::Lane, prefixed.as_bytes())? {
+            Some(bytes) => {
+                let id = deserialize_key(bytes)?;
+                responder.send(id).expect(RESPONSE_FAILURE);
+            }
+            None => {
+                db.merge_keyspace(KeyspaceName::Lane, COUNTER_BYTES, STEP)?;
+
+                let counter_bytes = db
+                    .get_keyspace(KeyspaceName::Lane, COUNTER_BYTES)?
+                    .expect(INCONSISTENT_KEYSPACE);
+                let id = deserialize_key(counter_bytes.as_slice())?;
+
+                db.put_keyspace(
+                    KeyspaceName::Lane,
+                    prefixed.as_bytes(),
+                    counter_bytes.as_slice(),
+                )?;
+
+                responder.send(id).expect(RESPONSE_FAILURE);
+            }
+        }
+    }
+
+    Ok(())
+}
+
+#[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::num::NonZeroUsize;
+    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(),
+        ]
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn lane_id() {
+        let delegate = Arc::new(MockStore::with_keyspaces(keyspaces()));
+        let store = KeyStore::new(delegate.clone(), NonZeroUsize::new(8).unwrap());
+
+        let lane_uri = "A";
+
+        assert_eq!(store.id_for(lane_uri).await, 1);
+        assert_eq!(store.id_for(lane_uri).await, 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);
+    }
+
+    #[tokio::test]
+    async fn multiple_lane_ids() {
+        let delegate = Arc::new(MockStore::with_keyspaces(keyspaces()));
+        let store = KeyStore::new(delegate.clone(), NonZeroUsize::new(8).unwrap());
+
+        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.as_str()).await, 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..a3f9ba61e
--- /dev/null
+++ b/swim_server/src/store/mock.rs
@@ -0,0 +1,128 @@
+// 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::{Arc, Mutex};
+
+use crate::store::keystore::{lane_key_task, KeyRequest, KeystoreTask, STEP};
+use futures::future::BoxFuture;
+use futures::{FutureExt, Stream};
+use store::keyspaces::{Keyspace, KeyspaceByteEngine};
+use store::{deserialize, serialize, StoreError};
+
+pub struct MockStore {
+    values: Mutex<HashMap<String, HashMap<Vec<u8>, Vec<u8>>>>,
+}
+
+impl KeystoreTask for MockStore {
+    fn run<DB, S>(db: Arc<DB>, events: S) -> BoxFuture<'static, Result<(), StoreError>>
+    where
+        DB: KeyspaceByteEngine,
+        S: Stream<Item = KeyRequest> + Unpin + Send + 'static,
+    {
+        lane_key_task(db, events).boxed()
+    }
+}
+
+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..4a62c65cf
--- /dev/null
+++ b/swim_server/src/store/mod.rs
@@ -0,0 +1,205 @@
+// 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::io;
+use std::path::PathBuf;
+
+use serde::{Deserialize, Serialize};
+
+use store::keyspaces::{Keyspace, Keyspaces};
+use store::{StoreBuilder, StoreError};
+use utilities::fs::Dir;
+
+use crate::plane::store::{open_plane, PlaneStore, SwimPlaneStore};
+
+#[cfg(test)]
+pub mod mock;
+
+pub mod keystore;
+mod nostore;
+mod rocks;
+
+use crate::store::keystore::KeystoreTask;
+pub use rocks::{default_db_opts, default_keyspaces, RocksDatabase, RocksOpts};
+
+/// 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,
+    D::Store: KeystoreTask,
+{
+    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,
+    },
+}
+
+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..c720c9a95
--- /dev/null
+++ b/swim_server/src/store/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::store::keystore::{KeyRequest, KeystoreTask};
+use futures::future::BoxFuture;
+use futures::{Stream, StreamExt};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use store::keyspaces::{Keyspace, KeyspaceByteEngine, KeyspaceResolver, Keyspaces};
+use store::{
+    EngineInfo, EngineIterOpts, EngineIterator, EnginePrefixIterator, EngineRefIterator,
+    IteratorKey, KvPair, Store, StoreBuilder, 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) -> KvPair {
+        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)
+    }
+}
+
+impl KeystoreTask for NoStore {
+    fn run<DB, S>(_: Arc<DB>, events: S) -> BoxFuture<'static, Result<(), StoreError>>
+    where
+        DB: KeyspaceByteEngine,
+        S: Stream<Item = KeyRequest> + Unpin + Send + 'static,
+    {
+        Box::pin(async move {
+            let _ = events.for_each(|(_, responder)| async {
+                let _ = responder.send(0);
+            });
+            Ok(())
+        })
+    }
+}
diff --git a/swim_server/src/store/rocks.rs b/swim_server/src/store/rocks.rs
new file mode 100644
index 000000000..fe96fcb22
--- /dev/null
+++ b/swim_server/src/store/rocks.rs
@@ -0,0 +1,192 @@
+// 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::keystore::{lane_key_task, KeyRequest, KeystoreTask};
+use crate::store::{LANE_KS, MAP_LANE_KS, VALUE_LANE_KS};
+use futures::future::BoxFuture;
+use futures::FutureExt;
+use futures::Stream;
+use rocksdb::{ColumnFamily, ColumnFamilyDescriptor, Options, SliceTransform, DB};
+use std::mem::size_of;
+use std::path::Path;
+use std::sync::Arc;
+use store::keyspaces::{Keyspace, KeyspaceByteEngine};
+use store::keyspaces::{KeyspaceDef, KeyspaceResolver, Keyspaces};
+use store::{
+    EngineInfo, EngineIterOpts, EngineRefIterator, RocksEngine, RocksIterator, RocksPrefixIterator,
+    Store, StoreBuilder, StoreError,
+};
+
+#[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)
+    }
+}
+
+impl KeystoreTask for RocksDatabase {
+    fn run<DB, S>(db: Arc<DB>, events: S) -> BoxFuture<'static, Result<(), StoreError>>
+    where
+        DB: KeyspaceByteEngine,
+        S: Stream<Item = KeyRequest> + Unpin + Send + 'static,
+    {
+        lane_key_task(db, events).boxed()
+    }
+}
+
+#[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
+                .into_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(0.2);
+
+    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/src/engines/mod.rs b/swim_server/store/src/engines/mod.rs
index 6ea987ede..fd9bfaeb2 100644
--- a/swim_server/store/src/engines/mod.rs
+++ b/swim_server/store/src/engines/mod.rs
@@ -15,7 +15,7 @@
 use std::path::Path;
 
 pub use nostore::{NoStore, NoStoreOpts};
-pub use rocks::{RocksDatabase, RocksOpts};
+pub use rocks::{RocksEngine, RocksIterator, RocksOpts, RocksPrefixIterator};
 
 use crate::keyspaces::Keyspaces;
 use crate::{Store, StoreError};
@@ -35,7 +35,7 @@ pub trait ByteEngine: 'static {
     fn delete(&self, key: &[u8]) -> Result<(), StoreError>;
 }
 
-pub trait StoreBuilder: Sized {
+pub trait StoreBuilder: Sized + Clone {
     type Store: Store;
 
     fn build<I>(self, path: I, keyspaces: &Keyspaces<Self>) -> Result<Self::Store, StoreError>
diff --git a/swim_server/store/src/engines/nostore.rs b/swim_server/store/src/engines/nostore.rs
index 68e0e1b03..ff3c560bb 100644
--- a/swim_server/store/src/engines/nostore.rs
+++ b/swim_server/store/src/engines/nostore.rs
@@ -12,17 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::engines::StoreBuilder;
+use crate::engines::{ByteEngine, StoreBuilder};
 use crate::iterator::{
     EngineIterOpts, EngineIterator, EnginePrefixIterator, EngineRefIterator, IteratorKey,
 };
 use crate::keyspaces::{Keyspace, KeyspaceByteEngine, KeyspaceResolver, Keyspaces};
-use crate::{ByteEngine, EngineInfo, Store, StoreError};
+use crate::{EngineInfo, KvPair, Store, StoreError};
 use std::borrow::Borrow;
 use std::path::{Path, PathBuf};
 
 /// A delegate store database that does nothing.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct NoStore {
     path: PathBuf,
 }
@@ -93,7 +93,7 @@ impl KeyspaceByteEngine for NoStore {
     }
 }
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct NoStoreOpts;
 impl StoreBuilder for NoStoreOpts {
     type Store = NoStore;
@@ -148,7 +148,7 @@ impl EngineIterator for NoStoreEngineIterator {
 
 pub struct NoStoreEnginePrefixIterator;
 impl EnginePrefixIterator for NoStoreEnginePrefixIterator {
-    fn next(&mut self) -> Option<(Box<[u8]>, Box<[u8]>)> {
+    fn next(&mut self) -> KvPair {
         None
     }
 
diff --git a/swim_server/store/src/engines/rocks/iterator.rs b/swim_server/store/src/engines/rocks/iterator.rs
index 79259eb5b..32ed75421 100644
--- a/swim_server/store/src/engines/rocks/iterator.rs
+++ b/swim_server/store/src/engines/rocks/iterator.rs
@@ -13,10 +13,10 @@
 // limitations under the License.
 
 use crate::iterator::{EngineIterOpts, EngineIterator, EnginePrefixIterator};
-use crate::{EngineRefIterator, IteratorKey, RocksDatabase, StoreError};
+use crate::{EngineRefIterator, IteratorKey, KvPair, RocksEngine, StoreError};
 use rocksdb::{DBIteratorWithThreadMode, DBRawIterator, DBWithThreadMode, SingleThreaded};
 
-impl<'a: 'b, 'b> EngineRefIterator<'a, 'b> for RocksDatabase {
+impl<'a: 'b, 'b> EngineRefIterator<'a, 'b> for RocksEngine {
     type EngineIterator = RocksIterator<'b>;
     type EnginePrefixIterator = RocksPrefixIterator<'b>;
 
@@ -47,7 +47,7 @@ pub struct RocksPrefixIterator<'p> {
 }
 
 impl<'d> EnginePrefixIterator for RocksPrefixIterator<'d> {
-    fn next(&mut self) -> Option<(Box<[u8]>, Box<[u8]>)> {
+    fn next(&mut self) -> KvPair {
         self.delegate.next()
     }
 
diff --git a/swim_server/store/src/engines/rocks/mod.rs b/swim_server/store/src/engines/rocks/mod.rs
index 7f5ab616b..a82fd9d57 100644
--- a/swim_server/store/src/engines/rocks/mod.rs
+++ b/swim_server/store/src/engines/rocks/mod.rs
@@ -17,6 +17,7 @@ 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};
@@ -35,20 +36,20 @@ impl From<rocksdb::Error> for StoreError {
 /// A Rocks database engine.
 ///
 /// See https://github.com/facebook/rocksdb/wiki for details about the features and limitations.
-#[derive(Debug)]
-pub struct RocksDatabase {
+#[derive(Debug, Clone)]
+pub struct RocksEngine {
     pub(crate) delegate: Arc<DB>,
 }
 
-impl RocksDatabase {
-    pub fn new(delegate: DB) -> RocksDatabase {
-        RocksDatabase {
+impl RocksEngine {
+    pub fn new(delegate: DB) -> RocksEngine {
+        RocksEngine {
             delegate: Arc::new(delegate),
         }
     }
 }
 
-impl Store for RocksDatabase {
+impl Store for RocksEngine {
     fn path(&self) -> &Path {
         &self.delegate.path()
     }
@@ -61,7 +62,7 @@ impl Store for RocksDatabase {
     }
 }
 
-impl KeyspaceResolver for RocksDatabase {
+impl KeyspaceResolver for RocksEngine {
     type ResolvedKeyspace = ColumnFamily;
 
     fn resolve_keyspace<K: Keyspace>(&self, space: &K) -> Option<&Self::ResolvedKeyspace> {
@@ -74,7 +75,7 @@ impl KeyspaceResolver for RocksDatabase {
 pub struct RocksOpts(pub Options);
 
 impl StoreBuilder for RocksOpts {
-    type Store = RocksDatabase;
+    type Store = RocksEngine;
 
     fn build<I>(self, path: I, keyspaces: &Keyspaces<Self>) -> Result<Self::Store, StoreError>
     where
@@ -83,7 +84,7 @@ impl StoreBuilder for RocksOpts {
         let Keyspaces { keyspaces } = keyspaces;
         let descriptors =
             keyspaces
-                .into_iter()
+                .iter()
                 .fold(Vec::with_capacity(keyspaces.len()), |mut vec, def| {
                     let cf_descriptor =
                         ColumnFamilyDescriptor::new(def.name.to_string(), def.opts.0.clone());
@@ -92,7 +93,7 @@ impl StoreBuilder for RocksOpts {
                 });
 
         let db = DB::open_cf_descriptors(&self.0, path, descriptors)?;
-        Ok(RocksDatabase::new(db))
+        Ok(RocksEngine::new(db))
     }
 }
 
@@ -117,7 +118,7 @@ where
     }
 }
 
-impl KeyspaceByteEngine for RocksDatabase {
+impl KeyspaceByteEngine for RocksEngine {
     fn put_keyspace<K: Keyspace>(
         &self,
         keyspace: K,
diff --git a/swim_server/store/src/iterator.rs b/swim_server/store/src/iterator.rs
index 7dfb85fc9..3480fcfb6 100644
--- a/swim_server/store/src/iterator.rs
+++ b/swim_server/store/src/iterator.rs
@@ -58,11 +58,13 @@ pub trait EngineIterator {
     fn valid(&self) -> Result<bool, StoreError>;
 }
 
+pub type KvPair = Option<(Box<[u8]>, Box<[u8]>)>;
+
 /// 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<(Box<[u8]>, Box<[u8]>)>;
+    fn next(&mut self) -> KvPair;
 
     /// 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
diff --git a/swim_server/store/src/keyspaces.rs b/swim_server/store/src/keyspaces.rs
index 911509bab..831b16650 100644
--- a/swim_server/store/src/keyspaces.rs
+++ b/swim_server/store/src/keyspaces.rs
@@ -14,7 +14,9 @@
 
 use crate::StoreError;
 
+/// A handle to a portion of logically partitioned data.
 pub trait Keyspace {
+    /// The name of the keyspace.
     fn name(&self) -> &str;
 }
 
@@ -23,11 +25,12 @@ pub trait Keyspace {
 /// 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(crate) name: &'static str,
+    pub name: &'static str,
     /// The configuration options that will be used to open the keyspace.
-    pub(crate) opts: O,
+    pub opts: O,
 }
 
 impl<O> KeyspaceDef<O> {
@@ -37,6 +40,7 @@ impl<O> KeyspaceDef<O> {
 }
 
 /// A list of keyspace definitions to initialise a store with.
+#[derive(Clone)]
 pub struct Keyspaces<O> {
     pub keyspaces: Vec<KeyspaceDef<O>>,
 }
diff --git a/swim_server/store/src/lib.rs b/swim_server/store/src/lib.rs
index 2118a54d4..779b7876b 100644
--- a/swim_server/store/src/lib.rs
+++ b/swim_server/store/src/lib.rs
@@ -25,12 +25,14 @@ use std::{error::Error as StdError, io};
 
 use thiserror::Error;
 
-pub use engines::{RocksDatabase, RocksOpts};
-pub use rocksdb::{MergeOperands, SliceTransform};
-
-use crate::engines::ByteEngine;
-pub use crate::iterator::{EngineRefIterator, IteratorKey, OwnedEngineRefIterator};
 use crate::keyspaces::{KeyspaceByteEngine, KeyspaceResolver};
+pub use engines::{RocksEngine, RocksIterator, RocksPrefixIterator, StoreBuilder};
+
+pub use crate::iterator::{
+    EngineIterOpts, EngineIterator, EnginePrefixIterator, EngineRefIterator, IteratorKey, KvPair,
+    OwnedEngineRefIterator,
+};
+
 use serde::{Deserialize, Serialize};
 
 /// Store errors.
@@ -88,6 +90,7 @@ impl PartialEq for StoreError {
             (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,
@@ -99,7 +102,14 @@ impl PartialEq for StoreError {
 ///
 /// This trait only serves to compose the multiple traits that are required for a store.
 pub trait Store:
-    KeyspaceByteEngine + KeyspaceResolver + Send + Sync + Debug + OwnedEngineRefIterator + 'static
+    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;
@@ -133,3 +143,7 @@ pub fn serialize<S: Serialize>(obj: &S) -> Result<Vec<u8>, StoreError> {
 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/utilities/Cargo.toml b/utilities/Cargo.toml
index a8f2f5785..819ca5bac 100644
--- a/utilities/Cargo.toml
+++ b/utilities/Cargo.toml
@@ -24,6 +24,7 @@ macro_helpers = { path = "macro_helpers" }
 stringify_attr = { path = "stringify_attr" }
 url = "2.1.1"
 num = "0.3.0"
+tempdir = "0.3.7"
 
 [dev-dependencies]
 tokio = { version = "1.1.1", features = [ "rt-multi-thread", "macros"] }
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 bc65e5103..f07cd849c 100644
--- a/utilities/src/lib.rs
+++ b/utilities/src/lib.rs
@@ -18,6 +18,7 @@ use std::error::Error;
 use std::fmt::{Display, Formatter};
 
 pub mod errors;
+pub mod fs;
 pub mod future;
 pub mod instant;
 pub mod iteratee;


Test patch:
diff --git a/swim_client/test_server/src/lib.rs b/swim_client/test_server/src/lib.rs
index 240b56735..ce7ce6a19 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::transient_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 e220826e7..6abf8440b 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::{UplinkActionObserver, 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,
@@ -398,6 +401,7 @@ impl ServerRouter for TestRouter {
 
 impl AgentExecutionContext for TestContext {
     type Router = TestRouter;
+    type Store = SwimNodeStore<MockPlaneStore>;
 
     fn router_handle(&self) -> Self::Router {
         let TestContext {
@@ -420,6 +424,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 {
@@ -1405,6 +1413,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())
@@ -1421,6 +1430,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 ebd8c1147..14c160f9c 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,
@@ -329,6 +332,7 @@ impl TestContext {
 
 impl AgentExecutionContext for TestContext {
     type Router = TestRouter;
+    type Store = SwimNodeStore<MockPlaneStore>;
 
     fn router_handle(&self) -> Self::Router {
         let TestContext {
@@ -351,6 +355,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 05c901bae..b3b5682f1 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, UplinkActionObserver, UplinkEventObserver,
     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/store/tests.rs b/swim_server/src/agent/lane/store/tests.rs
new file mode 100644
index 000000000..df20f6f7f
--- /dev/null
+++ b/swim_server/src/agent/lane/store/tests.rs
@@ -0,0 +1,248 @@
+// 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 crate::agent::store::SwimNodeStore;
+use crate::plane::store::mock::MockPlaneStore;
+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<SwimNodeStore<MockPlaneStore>> for TestStoreIo<T>
+where
+    T: Future<Output = Result<(), LaneStoreErrorReport>> + Send + 'static,
+{
+    fn attach(
+        self,
+        _store: SwimNodeStore<MockPlaneStore>,
+        _lane_uri: String,
+        _error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        Box::pin(async move { self.0.await })
+    }
+
+    fn attach_boxed(
+        self: Box<Self>,
+        store: SwimNodeStore<MockPlaneStore>,
+        lane_uri: String,
+        error_handler: StoreErrorHandler,
+    ) -> BoxFuture<'static, Result<(), LaneStoreErrorReport>> {
+        (*self).attach(store, lane_uri, error_handler)
+    }
+}
+
+#[tokio::test]
+async fn task_ok() {
+    let test_io: Box<dyn StoreIo<SwimNodeStore<MockPlaneStore>>> =
+        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<SwimNodeStore<MockPlaneStore>>> = 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<SwimNodeStore<MockPlaneStore>>> = 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<SwimNodeStore<MockPlaneStore>>> =
+        Box::new(TestStoreIo(pending()));
+    tasks.insert("pending_io".to_string(), pending_io);
+
+    let ok_io: Box<dyn StoreIo<SwimNodeStore<MockPlaneStore>>> =
+        Box::new(TestStoreIo(async { Ok(()) }));
+    tasks.insert("ok_io".to_string(), ok_io);
+
+    let err_io: Box<dyn StoreIo<SwimNodeStore<MockPlaneStore>>> = 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/store/tests.rs b/swim_server/src/agent/store/tests.rs
new file mode 100644
index 000000000..23c60d3cf
--- /dev/null
+++ b/swim_server/src/agent/store/tests.rs
@@ -0,0 +1,13 @@
+// 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.
diff --git a/swim_server/src/agent/tests/data_macro_agent.rs b/swim_server/src/agent/tests/data_macro_agent.rs
index ce50da7d2..cbaf9ecc9 100644
--- a/swim_server/src/agent/tests/data_macro_agent.rs
+++ b/swim_server/src/agent/tests/data_macro_agent.rs
@@ -21,6 +21,7 @@ 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;
@@ -538,6 +539,7 @@ async fn agent_loop() {
         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..b685cb3e4 100644
--- a/swim_server/src/agent/tests/mod.rs
+++ b/swim_server/src/agent/tests/mod.rs
@@ -29,6 +29,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;
@@ -764,6 +765,7 @@ pub async fn run_agent_test<Agent, Config, Lifecycle>(
         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..6b740ed18 100644
--- a/swim_server/src/agent/tests/reporting_agent.rs
+++ b/swim_server/src/agent/tests/reporting_agent.rs
@@ -25,8 +25,9 @@ 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, LaneTasks, SwimAgent};
 use futures::future::{ready, BoxFuture, Ready};
 use futures::FutureExt;
 use std::collections::HashMap;
@@ -343,33 +344,37 @@ 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, Store>,
     )
     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",
+            "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",
+            "total".to_string(),
             false,
             exec_conf,
             0,
@@ -377,6 +382,8 @@ impl SwimAgent<TestAgentConfig> for ReportingAgent {
                 inner: inner.clone(),
             },
             |agent: &ReportingAgent| &agent.total,
+            true,
+            store,
         );
 
         let (action, action_tasks, _) = agent::make_command_lane(
diff --git a/swim_server/src/meta/info/tests.rs b/swim_server/src/meta/info/tests.rs
index 6b5fc7aee..cf0dfdac0 100644
--- a/swim_server/src/meta/info/tests.rs
+++ b/swim_server/src/meta/info/tests.rs
@@ -15,6 +15,7 @@
 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;
@@ -176,6 +177,7 @@ async fn lane_info_sync() {
         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..5c7be6541 100644
--- a/swim_server/src/meta/log/tests.rs
+++ b/swim_server/src/meta/log/tests.rs
@@ -16,6 +16,7 @@ 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,
 };
@@ -155,6 +156,7 @@ async fn agent_log() {
         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..ed4294fb6 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,16 +86,18 @@ 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>,
-        DynamicAgentIo<Context>,
+        DynamicAgentIo<Context, Store>,
     )
     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..8f78bbb09 100644
--- a/swim_server/src/plane/tests/fixture.rs
+++ b/swim_server/src/plane/tests/fixture.rs
@@ -14,10 +14,12 @@
 
 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::store::mock::MockPlaneStore;
 use crate::plane::{AgentRoute, EnvChannel};
 use crate::routing::{ServerRouter, TaggedEnvelope};
 use futures::future::BoxFuture;
@@ -89,6 +91,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,7 +102,8 @@ 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,
@@ -109,6 +113,7 @@ impl<Clk: Clock, Delegate: ServerRouter + 'static>
         _clock: Clk,
         incoming_envelopes: EnvChannel,
         mut router: PlaneRouter<Delegate>,
+        _store: SwimNodeStore<MockPlaneStore>,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>) {
         let id = parameters[PARAM_NAME].clone();
         let target = self.0.clone();
@@ -133,8 +138,8 @@ impl<Clk: Clock, Delegate: ServerRouter + 'static>
             while incoming_envelopes.next().await.is_some() {}
             AgentResult {
                 route: uri,
-                dispatcher_errors: DispatcherErrors::default(),
-                failed: false,
+                dispatcher_task: Default::default(),
+                store_task: Default::default(),
             }
         }
         .boxed();
@@ -142,7 +147,8 @@ 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(
@@ -153,6 +159,7 @@ impl<Clk: Clock, Delegate> AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>>
         _clock: Clk,
         incoming_envelopes: EnvChannel,
         _router: PlaneRouter<Delegate>,
+        _store: SwimNodeStore<MockPlaneStore>,
     ) -> (Arc<dyn Any + Send + Sync>, BoxFuture<'static, AgentResult>) {
         let ReceiveAgentRoute { expected_id, done } = self;
         let mut done_sender = done.lock().take();
@@ -188,8 +195,11 @@ impl<Clk: Clock, Delegate> AgentRoute<Clk, EnvChannel, PlaneRouter<Delegate>>
             }
             AgentResult {
                 route: uri,
-                dispatcher_errors: DispatcherErrors::default(),
-                failed: times_seen != 1,
+                dispatcher_task: AgentTaskResult {
+                    errors: DispatcherErrors::default(),
+                    failed: times_seen != 1,
+                },
+                store_task: Default::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
index 0a6c2a23c..e744f85a6 100644
--- a/swim_server/store/src/engines/rocks/tests.rs
+++ b/swim_server/store/src/engines/rocks/tests.rs
@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::engines::StoreBuilder;
+use crate::engines::{RocksOpts, StoreBuilder};
 use crate::iterator::EngineIterator;
 use crate::keyspaces::{Keyspace, KeyspaceByteEngine, KeyspaceDef, KeyspaceResolver, Keyspaces};
-use crate::{deserialize, serialize};
+use crate::RocksEngine;
+use crate::{deserialize, deserialize_key, serialize};
 use crate::{EngineRefIterator, StoreError};
-use crate::{RocksDatabase, RocksOpts};
 use rocksdb::{MergeOperands, Options, SliceTransform};
 use std::collections::HashMap;
 use std::mem::size_of;
@@ -25,7 +25,7 @@ use std::ops::{Deref, Range};
 use tempdir::TempDir;
 
 impl Deref for TransientDatabase {
-    type Target = RocksDatabase;
+    type Target = RocksEngine;
 
     fn deref(&self) -> &Self::Target {
         &self.delegate
@@ -34,7 +34,7 @@ impl Deref for TransientDatabase {
 
 pub struct TransientDatabase {
     _dir: TempDir,
-    delegate: RocksDatabase,
+    delegate: RocksEngine,
 }
 
 impl TransientDatabase {
@@ -51,10 +51,6 @@ impl TransientDatabase {
     }
 }
 
-fn deserialize_key<B: AsRef<[u8]>>(bytes: B) -> Result<u64, ()> {
-    bincode::deserialize::<u64>(bytes.as_ref()).map_err(|_| ())
-}
-
 fn default_lane_opts() -> Options {
     let mut opts = rocksdb::Options::default();
     opts.set_prefix_extractor(SliceTransform::create_fixed_prefix(size_of::<u64>()));