Rename things to progress code review

`*Storage` -> `*Store`
`storage_client` -> `store`
`etcd_root` -> `instance_root`

Add a comment explaining that although `store` is `dyn KeyValueStore`
right now, it has ambitions to be the one-ring of Dynamo storage.

Signed-off-by: Graham King <[email protected]>

Author: grahamking

lib/llm/src/http/service/health.rs

@@ -52,10 +52,10 @@ async fn live_handler(
 async fn health_handler(
     axum::extract::State(state): axum::extract::State<Arc<service_v2::State>>,
 ) -> impl IntoResponse {
-    let instances = match list_all_instances(state.storage_client()).await {
+    let instances = match list_all_instances(state.store()).await {
         Ok(instances) => instances,
         Err(err) => {
-            tracing::warn!(%err, "Failed to fetch instances from storage");
+            tracing::warn!(%err, "Failed to fetch instances from store");
             vec![]
         }
     };

lib/llm/src/http/service/service_v2.rs

@@ -19,9 +19,9 @@ use anyhow::Result;
 use axum_server::tls_rustls::RustlsConfig;
 use derive_builder::Builder;
 use dynamo_runtime::logging::make_request_span;
-use dynamo_runtime::storage::key_value_store::EtcdStorage;
+use dynamo_runtime::storage::key_value_store::EtcdStore;
 use dynamo_runtime::storage::key_value_store::KeyValueStore;
-use dynamo_runtime::storage::key_value_store::MemoryStorage;
+use dynamo_runtime::storage::key_value_store::MemoryStore;
 use dynamo_runtime::transports::etcd;
 use std::net::SocketAddr;
 use tokio::task::JoinHandle;
@@ -33,7 +33,7 @@ pub struct State {
     metrics: Arc<Metrics>,
     manager: Arc<ModelManager>,
     etcd_client: Option<etcd::Client>,
-    storage_client: Arc<dyn KeyValueStore>,
+    store: Arc<dyn KeyValueStore>,
     flags: StateFlags,
 }
 
@@ -79,7 +79,7 @@ impl State {
             manager,
             metrics: Arc::new(Metrics::default()),
             etcd_client: None,
-            storage_client: Arc::new(MemoryStorage::new()),
+            store: Arc::new(MemoryStore::new()),
             flags: StateFlags {
                 chat_endpoints_enabled: AtomicBool::new(false),
                 cmpl_endpoints_enabled: AtomicBool::new(false),
@@ -93,7 +93,7 @@ impl State {
         Self {
             manager,
             metrics: Arc::new(Metrics::default()),
-            storage_client: Arc::new(EtcdStorage::new(etcd_client.clone())),
+            store: Arc::new(EtcdStore::new(etcd_client.clone())),
             etcd_client: Some(etcd_client),
             flags: StateFlags {
                 chat_endpoints_enabled: AtomicBool::new(false),
@@ -120,8 +120,8 @@ impl State {
         self.etcd_client.as_ref()
     }
 
-    pub fn storage_client(&self) -> Arc<dyn KeyValueStore> {
-        self.storage_client.clone()
+    pub fn store(&self) -> Arc<dyn KeyValueStore> {
+        self.store.clone()
     }
 
     // TODO

lib/llm/src/local_model.rs

@@ -12,7 +12,7 @@ use dynamo_runtime::storage::key_value_store::Key;
 use dynamo_runtime::traits::DistributedRuntimeProvider;
 use dynamo_runtime::{
     component::Endpoint,
-    storage::key_value_store::{EtcdStorage, KeyValueStore, KeyValueStoreManager},
+    storage::key_value_store::{EtcdStore, KeyValueStore, KeyValueStoreManager},
 };
 
 use crate::entrypoint::RouterConfig;
@@ -409,7 +409,7 @@ impl LocalModel {
         self.card.move_to_nats(nats_client.clone()).await?;
 
         // Publish the Model Deployment Card to KV store
-        let kvstore: Box<dyn KeyValueStore> = Box::new(EtcdStorage::new(etcd_client.clone()));
+        let kvstore: Box<dyn KeyValueStore> = Box::new(EtcdStore::new(etcd_client.clone()));
         let card_store = Arc::new(KeyValueStoreManager::new(kvstore));
         let lease_id = endpoint.drt().primary_lease().map(|l| l.id()).unwrap_or(0);
         let key = Key::from_raw(endpoint.unique_path(lease_id));

lib/llm/src/model_card.rs

@@ -23,7 +23,7 @@ use anyhow::{Context, Result};
 use derive_builder::Builder;
 use dynamo_runtime::DistributedRuntime;
 use dynamo_runtime::storage::key_value_store::{
-    EtcdStorage, Key, KeyValueStore, KeyValueStoreManager,
+    EtcdStore, Key, KeyValueStore, KeyValueStoreManager,
 };
 use dynamo_runtime::{slug::Slug, storage::key_value_store::Versioned, transports::nats};
 use serde::{Deserialize, Serialize};
@@ -457,7 +457,7 @@ impl ModelDeploymentCard {
             // Should be impossible because we only get here on an etcd event
             anyhow::bail!("Missing etcd_client");
         };
-        let store: Box<dyn KeyValueStore> = Box::new(EtcdStorage::new(etcd_client));
+        let store: Box<dyn KeyValueStore> = Box::new(EtcdStore::new(etcd_client));
         let card_store = Arc::new(KeyValueStoreManager::new(store));
         let Some(mut card) = card_store
             .load::<ModelDeploymentCard>(ROOT_PATH, mdc_key)

lib/runtime/src/component.rs

@@ -72,7 +72,7 @@ pub mod service;
 
 pub use client::{Client, InstanceSource};
 
-/// The root etcd path where each instance registers itself in etcd.
+/// The root key-value path where each instance registers itself in.
 /// An instance is namespace+component+endpoint+lease_id and must be unique.
 pub const INSTANCE_ROOT_PATH: &str = "v1/instances";
 
@@ -223,8 +223,8 @@ impl MetricsRegistry for Component {
 }
 
 impl Component {
-    /// The component part of an instance path in etcd.
-    pub fn etcd_root(&self) -> String {
+    /// The component part of an instance path in key-value store.
+    pub fn instance_root(&self) -> String {
         let ns = self.namespace.name();
         let cp = &self.name;
         format!("{INSTANCE_ROOT_PATH}/{ns}/{cp}")
@@ -266,8 +266,8 @@ impl Component {
     }
 
     pub async fn list_instances(&self) -> anyhow::Result<Vec<Instance>> {
-        let client = self.drt.storage_client();
-        let Some(bucket) = client.get_bucket(&self.etcd_root()).await? else {
+        let client = self.drt.store();
+        let Some(bucket) = client.get_bucket(&self.instance_root()).await? else {
             return Ok(vec![]);
         };
         let entries = bucket.entries().await?;
@@ -467,7 +467,7 @@ impl Endpoint {
 
     /// The endpoint part of an instance path in etcd
     pub fn etcd_root(&self) -> String {
-        let component_path = self.component.etcd_root();
+        let component_path = self.component.instance_root();
         let endpoint_name = &self.name;
         format!("{component_path}/{endpoint_name}")
     }

lib/runtime/src/distributed.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub use crate::component::Component;
-use crate::storage::key_value_store::{EtcdStorage, KeyValueStore, MemoryStorage};
+use crate::storage::key_value_store::{EtcdStore, KeyValueStore, MemoryStore};
 use crate::transports::nats::DRTNatsClientPrometheusMetrics;
 use crate::{
     ErrorContext, RuntimeCallback,
@@ -45,15 +45,14 @@ impl DistributedRuntime {
 
         let runtime_clone = runtime.clone();
 
-        let (etcd_client, storage_client) = if is_static {
-            let storage_client: Arc<dyn KeyValueStore> = Arc::new(MemoryStorage::new());
-            (None, storage_client)
+        let (etcd_client, store) = if is_static {
+            let store: Arc<dyn KeyValueStore> = Arc::new(MemoryStore::new());
+            (None, store)
         } else {
             let etcd_client = etcd::Client::new(etcd_config.clone(), runtime_clone).await?;
-            let storage_client: Arc<dyn KeyValueStore> =
-                Arc::new(EtcdStorage::new(etcd_client.clone()));
+            let store: Arc<dyn KeyValueStore> = Arc::new(EtcdStore::new(etcd_client.clone()));
 
-            (Some(etcd_client), storage_client)
+            (Some(etcd_client), store)
         };
 
         let nats_client = nats_config.clone().connect().await?;
@@ -83,7 +82,7 @@ impl DistributedRuntime {
         let distributed_runtime = Self {
             runtime,
             etcd_client,
-            storage_client,
+            store,
             nats_client,
             tcp_server: Arc::new(OnceCell::new()),
             system_status_server: Arc::new(OnceLock::new()),
@@ -278,8 +277,9 @@ impl DistributedRuntime {
     }
 
     /// An interface to store things. Will eventually replace `etcd_client`.
-    pub fn storage_client(&self) -> Arc<dyn KeyValueStore> {
-        self.storage_client.clone()
+    /// Currently does key-value, but will grow to include whatever we need to store.
+    pub fn store(&self) -> Arc<dyn KeyValueStore> {
+        self.store.clone()
     }
 
     pub fn child_token(&self) -> CancellationToken {

lib/runtime/src/lib.rs

@@ -154,7 +154,7 @@ pub struct DistributedRuntime {
     // we might consider a unifed transport manager here
     etcd_client: Option<transports::etcd::Client>,
     nats_client: transports::nats::Client,
-    storage_client: Arc<dyn KeyValueStore>,
+    store: Arc<dyn KeyValueStore>,
     tcp_server: Arc<OnceCell<Arc<transports::tcp::server::TcpStreamServer>>>,
     system_status_server: Arc<OnceLock<Arc<system_status_server::SystemStatusServerInfo>>>,
 

lib/runtime/src/storage/key_value_store.rs

@@ -17,11 +17,11 @@ use futures::StreamExt;
 use serde::{Deserialize, Serialize};
 
 mod mem;
-pub use mem::MemoryStorage;
+pub use mem::MemoryStore;
 mod nats;
-pub use nats::NATSStorage;
+pub use nats::NATSStore;
 mod etcd;
-pub use etcd::EtcdStorage;
+pub use etcd::EtcdStore;
 
 /// A key that is safe to use directly in the KV store.
 #[derive(Debug, Clone, PartialEq)]
@@ -69,12 +69,12 @@ pub trait KeyValueStore: Send + Sync {
         bucket_name: &str,
         // auto-delete items older than this
         ttl: Option<Duration>,
-    ) -> Result<Box<dyn KeyValueBucket>, StorageError>;
+    ) -> Result<Box<dyn KeyValueBucket>, StoreError>;
 
     async fn get_bucket(
         &self,
         bucket_name: &str,
-    ) -> Result<Option<Box<dyn KeyValueBucket>>, StorageError>;
+    ) -> Result<Option<Box<dyn KeyValueBucket>>, StoreError>;
 
     fn connection_id(&self) -> u64;
 }
@@ -90,7 +90,7 @@ impl KeyValueStoreManager {
         &self,
         bucket: &str,
         key: &Key,
-    ) -> Result<Option<T>, StorageError> {
+    ) -> Result<Option<T>, StoreError> {
         let Some(bucket) = self.0.get_bucket(bucket).await? else {
             // No bucket means no cards
             return Ok(None);
@@ -103,7 +103,7 @@ impl KeyValueStoreManager {
             Ok(None) => Ok(None),
             Err(err) => {
                 // TODO look at what errors NATS can give us and make more specific wrappers
-                Err(StorageError::NATSError(err.to_string()))
+                Err(StoreError::NATSError(err.to_string()))
             }
         }
     }
@@ -116,7 +116,7 @@ impl KeyValueStoreManager {
         bucket_name: &str,
         bucket_ttl: Option<Duration>,
     ) -> (
-        tokio::task::JoinHandle<Result<(), StorageError>>,
+        tokio::task::JoinHandle<Result<(), StoreError>>,
         tokio::sync::mpsc::UnboundedReceiver<T>,
     ) {
         let bucket_name = bucket_name.to_string();
@@ -141,7 +141,7 @@ impl KeyValueStoreManager {
                 let _ = tx.send(card);
             }
 
-            Ok::<(), StorageError>(())
+            Ok::<(), StoreError>(())
         });
         (watch_task, rx)
     }
@@ -152,14 +152,14 @@ impl KeyValueStoreManager {
         bucket_ttl: Option<Duration>,
         key: &Key,
         obj: &mut T,
-    ) -> anyhow::Result<StorageOutcome> {
+    ) -> anyhow::Result<StoreOutcome> {
         let obj_json = serde_json::to_string(obj)?;
         let bucket = self.0.get_or_create_bucket(bucket_name, bucket_ttl).await?;
 
         let outcome = bucket.insert(key, &obj_json, obj.revision()).await?;
 
         match outcome {
-            StorageOutcome::Created(revision) | StorageOutcome::Exists(revision) => {
+            StoreOutcome::Created(revision) | StoreOutcome::Exists(revision) => {
                 obj.set_revision(revision);
             }
         }
@@ -178,43 +178,43 @@ pub trait KeyValueBucket: Send {
         key: &Key,
         value: &str,
         revision: u64,
-    ) -> Result<StorageOutcome, StorageError>;
+    ) -> Result<StoreOutcome, StoreError>;
 
     /// Fetch an item from the key-value storage
-    async fn get(&self, key: &Key) -> Result<Option<bytes::Bytes>, StorageError>;
+    async fn get(&self, key: &Key) -> Result<Option<bytes::Bytes>, StoreError>;
 
     /// Delete an item from the bucket
-    async fn delete(&self, key: &Key) -> Result<(), StorageError>;
+    async fn delete(&self, key: &Key) -> Result<(), StoreError>;
 
     /// A stream of items inserted into the bucket.
     /// Every time the stream is polled it will either return a newly created entry, or block until
     /// such time.
     async fn watch(
         &self,
-    ) -> Result<Pin<Box<dyn futures::Stream<Item = bytes::Bytes> + Send + 'life0>>, StorageError>;
+    ) -> Result<Pin<Box<dyn futures::Stream<Item = bytes::Bytes> + Send + 'life0>>, StoreError>;
 
-    async fn entries(&self) -> Result<HashMap<String, bytes::Bytes>, StorageError>;
+    async fn entries(&self) -> Result<HashMap<String, bytes::Bytes>, StoreError>;
 }
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
-pub enum StorageOutcome {
+pub enum StoreOutcome {
     /// The operation succeeded and created a new entry with this revision.
     /// Note that "create" also means update, because each new revision is a "create".
     Created(u64),
     /// The operation did not do anything, the value was already present, with this revision.
     Exists(u64),
 }
-impl fmt::Display for StorageOutcome {
+impl fmt::Display for StoreOutcome {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            StorageOutcome::Created(revision) => write!(f, "Created at {revision}"),
-            StorageOutcome::Exists(revision) => write!(f, "Exists at {revision}"),
+            StoreOutcome::Created(revision) => write!(f, "Created at {revision}"),
+            StoreOutcome::Exists(revision) => write!(f, "Exists at {revision}"),
         }
     }
 }
 
 #[derive(thiserror::Error, Debug)]
-pub enum StorageError {
+pub enum StoreError {
     #[error("Could not find bucket '{0}'")]
     MissingBucket(String),
 
@@ -293,12 +293,12 @@ mod tests {
     async fn test_memory_storage() -> anyhow::Result<()> {
         init();
 
-        let s = Arc::new(MemoryStorage::new());
+        let s = Arc::new(MemoryStore::new());
         let s2 = Arc::clone(&s);
 
         let bucket = s.get_or_create_bucket(BUCKET_NAME, None).await?;
         let res = bucket.insert(&"test1".into(), "value1", 0).await?;
-        assert_eq!(res, StorageOutcome::Created(0));
+        assert_eq!(res, StoreOutcome::Created(0));
 
         let (got_first_tx, got_first_rx) = tokio::sync::oneshot::channel();
         let ingress = tokio::spawn(async move {
@@ -317,27 +317,27 @@ mod tests {
             let v = stream.next().await.unwrap();
             assert_eq!(v, "value3".as_bytes());
 
-            Ok::<_, StorageError>(())
+            Ok::<_, StoreError>(())
         });
 
-        // MemoryStorage uses a HashMap with no inherent ordering, so we must ensure test1 is
+        // MemoryStore uses a HashMap with no inherent ordering, so we must ensure test1 is
         // fetched before test2 is inserted, otherwise they can come out in any order, and we
         // wouldn't be testing the watch behavior.
         got_first_rx.await?;
 
         let res = bucket.insert(&"test2".into(), "value2", 0).await?;
-        assert_eq!(res, StorageOutcome::Created(0));
+        assert_eq!(res, StoreOutcome::Created(0));
 
         // Repeat a key and revision. Ignored.
         let res = bucket.insert(&"test2".into(), "value2", 0).await?;
-        assert_eq!(res, StorageOutcome::Exists(0));
+        assert_eq!(res, StoreOutcome::Exists(0));
 
         // Increment revision
         let res = bucket.insert(&"test2".into(), "value2", 1).await?;
-        assert_eq!(res, StorageOutcome::Created(1));
+        assert_eq!(res, StoreOutcome::Created(1));
 
         let res = bucket.insert(&"test3".into(), "value3", 0).await?;
-        assert_eq!(res, StorageOutcome::Created(0));
+        assert_eq!(res, StoreOutcome::Created(0));
 
         // ingress exits once it has received all values
         let _ = ingress.await?;
@@ -349,12 +349,12 @@ mod tests {
     async fn test_broadcast_stream() -> anyhow::Result<()> {
         init();
 
-        let s: &'static _ = Box::leak(Box::new(MemoryStorage::new()));
+        let s: &'static _ = Box::leak(Box::new(MemoryStore::new()));
         let bucket: &'static _ =
             Box::leak(Box::new(s.get_or_create_bucket(BUCKET_NAME, None).await?));
 
         let res = bucket.insert(&"test1".into(), "value1", 0).await?;
-        assert_eq!(res, StorageOutcome::Created(0));
+        assert_eq!(res, StoreOutcome::Created(0));
 
         let stream = bucket.watch().await?;
         let tap = TappableStream::new(stream, 10).await;