dynamo_runtime/storage/
kv.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Interface to a traditional key-value store such as etcd.
5//! "key_value_store" spelt out because in AI land "KV" means something else.
6
7use std::borrow::Cow;
8use std::pin::Pin;
9use std::str::FromStr;
10use std::sync::Arc;
11use std::time::Duration;
12use std::{collections::HashMap, path::PathBuf};
13use std::{env, fmt};
14
15use crate::CancellationToken;
16use crate::transports::etcd as etcd_transport;
17use async_trait::async_trait;
18use futures::StreamExt;
19use percent_encoding::{NON_ALPHANUMERIC, percent_decode_str, percent_encode};
20use serde::{Deserialize, Serialize};
21
22mod mem;
23pub use mem::MemoryStore;
24mod nats;
25pub use nats::NATSStore;
26mod etcd;
27pub use etcd::EtcdStore;
28mod file;
29pub use file::FileStore;
30
31/// String we use as the Key in a key-value storage operation. Simple String wrapper
32/// that can encode / decode a string.
33#[derive(Debug, Clone, PartialEq, Eq, Hash)]
34pub struct Key(String);
35
36impl Key {
37    pub fn new(s: String) -> Key {
38        Key(s)
39    }
40
41    /// Takes a URL-safe percent-encoded string and creates a Key from it by decoding first.
42    /// dynamo%2Fbackend%2Fgenerate%2F17216e63492ef21f becomes dynamo/backend/generate/17216e63492ef21f
43    pub fn from_url_safe(s: &str) -> Key {
44        Key(percent_decode_str(s).decode_utf8_lossy().to_string())
45    }
46
47    /// A URL-safe percent-encoded representation of this key.
48    /// e.g.  dynamo/backend/generate/17216e63492ef21f becomes dynamo%2Fbackend%2Fgenerate%2F17216e63492ef21f
49    pub fn url_safe(&self) -> Cow<'_, str> {
50        percent_encode(self.0.as_bytes(), NON_ALPHANUMERIC).into()
51    }
52}
53
54impl From<&str> for Key {
55    fn from(s: &str) -> Key {
56        Key::new(s.to_string())
57    }
58}
59
60impl fmt::Display for Key {
61    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
62        write!(f, "{}", self.0)
63    }
64}
65
66impl AsRef<str> for Key {
67    fn as_ref(&self) -> &str {
68        &self.0
69    }
70}
71
72impl From<&Key> for String {
73    fn from(k: &Key) -> String {
74        k.0.clone()
75    }
76}
77
78#[derive(Debug, Clone, PartialEq)]
79pub struct KeyValue {
80    key: Key,
81    value: bytes::Bytes,
82}
83
84impl KeyValue {
85    pub fn new(key: Key, value: bytes::Bytes) -> Self {
86        KeyValue { key, value }
87    }
88
89    pub fn key(&self) -> String {
90        self.key.clone().to_string()
91    }
92
93    pub fn key_str(&self) -> &str {
94        self.key.as_ref()
95    }
96
97    pub fn value(&self) -> &[u8] {
98        &self.value
99    }
100
101    pub fn value_str(&self) -> anyhow::Result<&str> {
102        std::str::from_utf8(self.value()).map_err(From::from)
103    }
104}
105
106#[derive(Debug, Clone, PartialEq)]
107pub enum WatchEvent {
108    Put(KeyValue),
109    Delete(Key),
110    Resync(HashMap<Key, bytes::Bytes>),
111}
112
113#[async_trait]
114pub trait Store: Send + Sync {
115    type Bucket: Bucket + Send + Sync + 'static;
116
117    async fn get_or_create_bucket(
118        &self,
119        bucket_name: &str,
120        // auto-delete items older than this
121        ttl: Option<Duration>,
122    ) -> Result<Self::Bucket, StoreError>;
123
124    async fn get_bucket(&self, bucket_name: &str) -> Result<Option<Self::Bucket>, StoreError>;
125
126    fn connection_id(&self) -> u64;
127
128    fn shutdown(&self);
129}
130
131#[derive(Clone, Debug, Default)]
132pub enum Selector {
133    // Box it because it is significantly bigger than the other variants
134    Etcd(Box<etcd_transport::ClientOptions>),
135    File(PathBuf),
136    #[default]
137    Memory,
138    // Nats not listed because likely we want to remove that impl. It is not currently used and not well tested.
139}
140
141impl fmt::Display for Selector {
142    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
143        match self {
144            Selector::Etcd(opts) => {
145                let urls = opts.etcd_url.join(",");
146                write!(f, "Etcd({urls})")
147            }
148            Selector::File(path) => write!(f, "File({})", path.display()),
149            Selector::Memory => write!(f, "Memory"),
150        }
151    }
152}
153
154impl FromStr for Selector {
155    type Err = anyhow::Error;
156
157    fn from_str(s: &str) -> anyhow::Result<Selector> {
158        match s {
159            "etcd" => Ok(Self::Etcd(Box::default())),
160            "file" => {
161                let root = env::var("DYN_FILE_KV")
162                    .map(PathBuf::from)
163                    .unwrap_or_else(|_| env::temp_dir().join("dynamo_store_kv"));
164                Ok(Self::File(root))
165            }
166            "mem" => Ok(Self::Memory),
167            x => anyhow::bail!("Unknown key-value store type '{x}'"),
168        }
169    }
170}
171
172impl TryFrom<String> for Selector {
173    type Error = anyhow::Error;
174
175    fn try_from(s: String) -> anyhow::Result<Selector> {
176        s.parse()
177    }
178}
179
180#[allow(clippy::large_enum_variant)]
181enum KeyValueStoreEnum {
182    Memory(MemoryStore),
183    Nats(NATSStore),
184    Etcd(EtcdStore),
185    File(FileStore),
186}
187
188impl KeyValueStoreEnum {
189    async fn get_or_create_bucket(
190        &self,
191        bucket_name: &str,
192        // auto-delete items older than this
193        ttl: Option<Duration>,
194    ) -> Result<Box<dyn Bucket>, StoreError> {
195        use KeyValueStoreEnum::*;
196        Ok(match self {
197            Memory(x) => Box::new(x.get_or_create_bucket(bucket_name, ttl).await?),
198            Nats(x) => Box::new(x.get_or_create_bucket(bucket_name, ttl).await?),
199            Etcd(x) => Box::new(x.get_or_create_bucket(bucket_name, ttl).await?),
200            File(x) => Box::new(x.get_or_create_bucket(bucket_name, ttl).await?),
201        })
202    }
203
204    async fn get_bucket(&self, bucket_name: &str) -> Result<Option<Box<dyn Bucket>>, StoreError> {
205        use KeyValueStoreEnum::*;
206        let maybe_bucket: Option<Box<dyn Bucket>> = match self {
207            Memory(x) => x
208                .get_bucket(bucket_name)
209                .await?
210                .map(|b| Box::new(b) as Box<dyn Bucket>),
211            Nats(x) => x
212                .get_bucket(bucket_name)
213                .await?
214                .map(|b| Box::new(b) as Box<dyn Bucket>),
215            Etcd(x) => x
216                .get_bucket(bucket_name)
217                .await?
218                .map(|b| Box::new(b) as Box<dyn Bucket>),
219            File(x) => x
220                .get_bucket(bucket_name)
221                .await?
222                .map(|b| Box::new(b) as Box<dyn Bucket>),
223        };
224        Ok(maybe_bucket)
225    }
226
227    fn connection_id(&self) -> u64 {
228        use KeyValueStoreEnum::*;
229        match self {
230            Memory(x) => x.connection_id(),
231            Etcd(x) => x.connection_id(),
232            Nats(x) => x.connection_id(),
233            File(x) => x.connection_id(),
234        }
235    }
236
237    fn shutdown(&self) {
238        use KeyValueStoreEnum::*;
239        match self {
240            Memory(x) => x.shutdown(),
241            Etcd(x) => x.shutdown(),
242            Nats(x) => x.shutdown(),
243            File(x) => x.shutdown(),
244        }
245    }
246}
247
248#[derive(Clone)]
249pub struct Manager(Arc<KeyValueStoreEnum>);
250
251impl Default for Manager {
252    fn default() -> Self {
253        Manager::memory()
254    }
255}
256
257impl Manager {
258    /// In-memory KeyValueStoreManager for testing
259    pub fn memory() -> Self {
260        Self::new(KeyValueStoreEnum::Memory(MemoryStore::new()))
261    }
262
263    pub fn etcd(etcd_client: crate::transports::etcd::Client) -> Self {
264        Self::new(KeyValueStoreEnum::Etcd(EtcdStore::new(etcd_client)))
265    }
266
267    pub fn file<P: Into<PathBuf>>(cancel_token: CancellationToken, root: P) -> Self {
268        Self::new(KeyValueStoreEnum::File(FileStore::new(cancel_token, root)))
269    }
270
271    fn new(s: KeyValueStoreEnum) -> Manager {
272        Manager(Arc::new(s))
273    }
274
275    pub async fn get_or_create_bucket(
276        &self,
277        bucket_name: &str,
278        // auto-delete items older than this
279        ttl: Option<Duration>,
280    ) -> Result<Box<dyn Bucket>, StoreError> {
281        self.0.get_or_create_bucket(bucket_name, ttl).await
282    }
283
284    pub async fn get_bucket(
285        &self,
286        bucket_name: &str,
287    ) -> Result<Option<Box<dyn Bucket>>, StoreError> {
288        self.0.get_bucket(bucket_name).await
289    }
290
291    pub fn connection_id(&self) -> u64 {
292        self.0.connection_id()
293    }
294
295    pub async fn load<T: for<'a> Deserialize<'a>>(
296        &self,
297        bucket: &str,
298        key: &Key,
299    ) -> Result<Option<T>, StoreError> {
300        let Some(bucket) = self.0.get_bucket(bucket).await? else {
301            // No bucket means no cards
302            return Ok(None);
303        };
304        Ok(match bucket.get(key).await? {
305            Some(card_bytes) => {
306                let card: T = serde_json::from_slice(card_bytes.as_ref())?;
307                Some(card)
308            }
309            None => None,
310        })
311    }
312
313    async fn forward_watch_event(
314        tx: &tokio::sync::mpsc::Sender<WatchEvent>,
315        event: WatchEvent,
316        cancel_token: &CancellationToken,
317        bucket_name: &str,
318    ) -> bool {
319        tokio::select! {
320            _ = cancel_token.cancelled() => false,
321            result = tx.send(event) => {
322                if let Err(error) = result {
323                    tracing::error!(
324                        bucket_name,
325                        %error,
326                        "KeyValueStoreManager.watch receiver closed"
327                    );
328                    false
329                } else {
330                    true
331                }
332            }
333        }
334    }
335
336    /// Returns a receiver that will receive all the existing keys, and
337    /// then block and receive new keys as they are created.
338    /// Starts a task that runs forever, watches the store.
339    pub fn watch(
340        self: Arc<Self>,
341        bucket_name: &str,
342        bucket_ttl: Option<Duration>,
343        cancel_token: CancellationToken,
344    ) -> (
345        tokio::task::JoinHandle<Result<(), StoreError>>,
346        tokio::sync::mpsc::Receiver<WatchEvent>,
347    ) {
348        let bucket_name = bucket_name.to_string();
349        // Backpressure is intentional: discovery state events must never be dropped.
350        let (tx, rx) = tokio::sync::mpsc::channel(16384);
351        let watch_task = tokio::spawn(async move {
352            // Start listening for changes but don't poll this yet
353            let bucket = self
354                .0
355                .get_or_create_bucket(&bucket_name, bucket_ttl)
356                .await?;
357            // Bucket::watch atomically establishes its initial snapshot and incremental
358            // stream. A separate entries() read here could replay an older buffered update
359            // after a newer snapshot.
360            let mut stream = bucket.watch().await?;
361
362            loop {
363                let event = tokio::select! {
364                    _ = cancel_token.cancelled() => break,
365                    result = stream.next() => match result {
366                        Some(event) => event,
367                        None => break,
368                    }
369                };
370                if !Self::forward_watch_event(&tx, event, &cancel_token, &bucket_name).await {
371                    break;
372                }
373            }
374
375            Ok::<(), StoreError>(())
376        });
377        (watch_task, rx)
378    }
379
380    pub async fn publish<T: Serialize + Versioned + Send + Sync>(
381        &self,
382        bucket_name: &str,
383        bucket_ttl: Option<Duration>,
384        key: &Key,
385        obj: &mut T,
386    ) -> anyhow::Result<StoreOutcome> {
387        let obj_json = serde_json::to_vec(obj)?;
388        let bucket = self.0.get_or_create_bucket(bucket_name, bucket_ttl).await?;
389
390        let outcome = bucket.insert(key, obj_json.into(), obj.revision()).await?;
391
392        match outcome {
393            StoreOutcome::Created(revision) | StoreOutcome::Exists(revision) => {
394                obj.set_revision(revision);
395            }
396        }
397        Ok(outcome)
398    }
399
400    /// Cleanup any temporary state.
401    /// TODO: Should this be async? Take &mut self?
402    pub fn shutdown(&self) {
403        self.0.shutdown()
404    }
405}
406
407/// An online storage for key-value config values.
408#[async_trait]
409pub trait Bucket: Send + Sync {
410    /// A bucket is a collection of key/value pairs.
411    /// Insert a value into a bucket, if it doesn't exist already
412    /// The Key should be the name of the item, not including the bucket name.
413    async fn insert(
414        &self,
415        key: &Key,
416        value: bytes::Bytes,
417        revision: u64,
418    ) -> Result<StoreOutcome, StoreError>;
419
420    /// Fetch an item from the key-value storage
421    /// The Key should be the name of the item, not including the bucket name.
422    async fn get(&self, key: &Key) -> Result<Option<bytes::Bytes>, StoreError>;
423
424    /// Replace an existing item only if its current value matches `expected`.
425    ///
426    /// Implementations must perform the comparison and replacement atomically.
427    /// A missing key returns [`StoreError::MissingKey`] and must never be created;
428    /// a value changed by another writer returns [`StoreError::Retry`].
429    /// A successful [`StoreOutcome`] revision is backend-specific and must not be
430    /// compared across backends or treated as a globally monotonic version.
431    async fn compare_and_replace(
432        &self,
433        key: &Key,
434        expected: bytes::Bytes,
435        value: bytes::Bytes,
436    ) -> Result<StoreOutcome, StoreError>;
437
438    /// Delete an item from the bucket
439    /// The Key should be the name of the item, not including the bucket name.
440    async fn delete(&self, key: &Key) -> Result<(), StoreError>;
441
442    /// An atomic initial snapshot followed by changes newer than that snapshot.
443    ///
444    /// Implementations must establish the snapshot and incremental watch without a gap and must
445    /// never emit an incremental value older than a value already emitted in the initial snapshot.
446    /// Existing entries may be emitted as individual WatchEvent::Put events or as one
447    /// WatchEvent::Resync.
448    async fn watch(
449        &self,
450    ) -> Result<Pin<Box<dyn futures::Stream<Item = WatchEvent> + Send + '_>>, StoreError>;
451
452    /// The entries in this bucket.
453    /// The Key includes the full path including the bucket name.
454    /// That means you cannot directory get a Key from `entries` and pass it to `get` or `delete`.
455    async fn entries(&self) -> Result<HashMap<Key, bytes::Bytes>, StoreError>;
456}
457
458#[derive(Debug, Copy, Clone, Eq, PartialEq)]
459pub enum StoreOutcome {
460    /// The operation succeeded and created a new entry with this revision.
461    /// Note that "create" also means update, because each new revision is a "create".
462    Created(u64),
463    /// The operation did not do anything, the value was already present, with this revision.
464    Exists(u64),
465}
466impl fmt::Display for StoreOutcome {
467    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
468        match self {
469            StoreOutcome::Created(revision) => write!(f, "Created at {revision}"),
470            StoreOutcome::Exists(revision) => write!(f, "Exists at {revision}"),
471        }
472    }
473}
474
475#[derive(thiserror::Error, Debug)]
476pub enum StoreError {
477    #[error("Could not find bucket '{0}'")]
478    MissingBucket(String),
479
480    #[error("Could not find key '{0}'")]
481    MissingKey(String),
482
483    #[error("Internal storage error: '{0}'")]
484    ProviderError(String),
485
486    #[error("Internal NATS error: {0}")]
487    NATSError(String),
488
489    #[error("Internal etcd error: {0}")]
490    EtcdError(String),
491
492    #[error("Internal filesystem error: {0}")]
493    FilesystemError(String),
494
495    #[error("Key Value Error: {0} for bucket '{1}'")]
496    KeyValueError(String, String),
497
498    #[error("Error decoding bytes: {0}")]
499    JSONDecodeError(#[from] serde_json::error::Error),
500
501    #[error("Race condition, retry the call")]
502    Retry,
503}
504
505/// A trait allowing to get/set a revision on an object.
506/// NATS uses this to ensure atomic updates.
507pub trait Versioned {
508    fn revision(&self) -> u64;
509    fn set_revision(&mut self, r: u64);
510}
511
512#[cfg(test)]
513mod tests {
514    use std::sync::Arc;
515
516    use super::*;
517    use futures::{StreamExt, pin_mut};
518
519    const BUCKET_NAME: &str = "v1/mdc";
520
521    /// Convert the value returned by `watch()` into a broadcast stream that multiple
522    /// clients can listen to.
523    #[allow(dead_code)]
524    pub struct TappableStream {
525        tx: tokio::sync::broadcast::Sender<WatchEvent>,
526    }
527
528    #[allow(dead_code)]
529    impl TappableStream {
530        async fn new<T>(stream: T, max_size: usize) -> Self
531        where
532            T: futures::Stream<Item = WatchEvent> + Send + 'static,
533        {
534            let (tx, _) = tokio::sync::broadcast::channel(max_size);
535            let tx2 = tx.clone();
536            tokio::spawn(async move {
537                pin_mut!(stream);
538                while let Some(x) = stream.next().await {
539                    let _ = tx2.send(x);
540                }
541            });
542            TappableStream { tx }
543        }
544
545        fn subscribe(&self) -> tokio::sync::broadcast::Receiver<WatchEvent> {
546            self.tx.subscribe()
547        }
548    }
549
550    fn init() {
551        crate::logging::init();
552    }
553
554    #[tokio::test]
555    async fn manager_watch_emits_initial_snapshot_once_before_updates() {
556        let manager = Arc::new(Manager::memory());
557        let bucket = manager
558            .get_or_create_bucket(BUCKET_NAME, None)
559            .await
560            .unwrap();
561        let key = Key::new("ns/worker/generate/1".to_string());
562        bucket.insert(&key, "old".into(), 1).await.unwrap();
563
564        let cancel_token = CancellationToken::new();
565        let (watch_task, mut rx) = manager
566            .clone()
567            .watch(BUCKET_NAME, None, cancel_token.clone());
568
569        let first = tokio::time::timeout(Duration::from_secs(1), rx.recv())
570            .await
571            .unwrap()
572            .unwrap();
573        let WatchEvent::Put(first) = first else {
574            panic!("expected initial put");
575        };
576        assert_eq!(first.value(), b"old");
577
578        bucket.insert(&key, "new".into(), 2).await.unwrap();
579        let second = tokio::time::timeout(Duration::from_secs(1), rx.recv())
580            .await
581            .unwrap()
582            .unwrap();
583        let WatchEvent::Put(second) = second else {
584            panic!("expected updated put");
585        };
586        assert_eq!(
587            second.value(),
588            b"new",
589            "the initial value must not be replayed after the snapshot"
590        );
591
592        cancel_token.cancel();
593        watch_task.await.unwrap().unwrap();
594    }
595
596    #[tokio::test]
597    async fn saturated_watch_channel_delivers_final_taint_state() {
598        let cancel_token = CancellationToken::new();
599        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
600        let first = WatchEvent::Put(KeyValue::new(
601            Key::new("v1/mdc/ns/worker/generate/1".to_string()),
602            br#"{"runtime_config":{"taints":["slow"]}}"#[..].into(),
603        ));
604        assert!(Manager::forward_watch_event(&tx, first.clone(), &cancel_token, BUCKET_NAME).await);
605
606        let final_event = WatchEvent::Put(KeyValue::new(
607            Key::new("v1/mdc/ns/worker/generate/1".to_string()),
608            br#"{"runtime_config":{"taints":["fast"]}}"#[..].into(),
609        ));
610        let final_event_for_send = final_event.clone();
611        let tx_clone = tx.clone();
612        let cancel_clone = cancel_token.clone();
613        let send_task = tokio::spawn(async move {
614            Manager::forward_watch_event(
615                &tx_clone,
616                final_event_for_send,
617                &cancel_clone,
618                BUCKET_NAME,
619            )
620            .await
621        });
622
623        tokio::task::yield_now().await;
624        assert!(
625            !send_task.is_finished(),
626            "send must wait for channel capacity"
627        );
628        assert_eq!(rx.recv().await, Some(first));
629        assert!(send_task.await.unwrap());
630        assert_eq!(rx.recv().await, Some(final_event));
631    }
632
633    #[tokio::test]
634    async fn test_memory_storage() -> anyhow::Result<()> {
635        init();
636
637        let s = Arc::new(MemoryStore::new());
638        let s2 = Arc::clone(&s);
639
640        let bucket = s.get_or_create_bucket(BUCKET_NAME, None).await?;
641        let res = bucket.insert(&"test1".into(), "value1".into(), 0).await?;
642        assert_eq!(res, StoreOutcome::Created(0));
643
644        let expected = [
645            WatchEvent::Put(KeyValue::new(Key::new("test1".into()), "value1".into())),
646            WatchEvent::Put(KeyValue::new(Key::new("test2".into()), "value2".into())),
647            WatchEvent::Put(KeyValue::new(
648                Key::new("test2".into()),
649                "value2-updated".into(),
650            )),
651            WatchEvent::Put(KeyValue::new(Key::new("test3".into()), "value3".into())),
652        ];
653
654        let (got_first_tx, got_first_rx) = tokio::sync::oneshot::channel();
655        let ingress = tokio::spawn(async move {
656            let b2 = s2.get_or_create_bucket(BUCKET_NAME, None).await?;
657            let mut stream = b2.watch().await?;
658
659            // Put in before starting the watch-all
660            let v = stream.next().await.unwrap();
661            assert_eq!(v, expected[0]);
662
663            got_first_tx.send(()).unwrap();
664
665            // Put in after
666            let v = stream.next().await.unwrap();
667            assert_eq!(v, expected[1]);
668
669            let v = stream.next().await.unwrap();
670            assert_eq!(v, expected[2]);
671
672            let v = stream.next().await.unwrap();
673            assert_eq!(v, expected[3]);
674
675            Ok::<_, StoreError>(())
676        });
677
678        // MemoryStore uses a HashMap with no inherent ordering, so we must ensure test1 is
679        // fetched before test2 is inserted, otherwise they can come out in any order, and we
680        // wouldn't be testing the watch behavior.
681        got_first_rx.await?;
682
683        let res = bucket.insert(&"test2".into(), "value2".into(), 0).await?;
684        assert_eq!(res, StoreOutcome::Created(0));
685
686        // Repeat a key and revision. Ignored.
687        let res = bucket.insert(&"test2".into(), "value2".into(), 0).await?;
688        assert_eq!(res, StoreOutcome::Exists(0));
689
690        // Increment revision
691        let res = bucket
692            .insert(&"test2".into(), "value2-updated".into(), 1)
693            .await?;
694        assert_eq!(res, StoreOutcome::Created(1));
695
696        let res = bucket.insert(&"test3".into(), "value3".into(), 0).await?;
697        assert_eq!(res, StoreOutcome::Created(0));
698
699        // ingress exits once it has received all values
700        let _ = ingress.await?;
701
702        Ok(())
703    }
704
705    #[tokio::test]
706    async fn test_broadcast_stream() -> anyhow::Result<()> {
707        init();
708
709        let s: &'static _ = Box::leak(Box::new(MemoryStore::new()));
710        let bucket: &'static _ =
711            Box::leak(Box::new(s.get_or_create_bucket(BUCKET_NAME, None).await?));
712
713        let res = bucket.insert(&"test1".into(), "value1".into(), 0).await?;
714        assert_eq!(res, StoreOutcome::Created(0));
715
716        let stream = bucket.watch().await?;
717        let tap = TappableStream::new(stream, 10).await;
718
719        let mut rx1 = tap.subscribe();
720        let mut rx2 = tap.subscribe();
721
722        let item = WatchEvent::Put(KeyValue::new(Key::new("test1".to_string()), "GK".into()));
723        let item_clone = item.clone();
724        let handle1 = tokio::spawn(async move {
725            let b = rx1.recv().await.unwrap();
726            assert_eq!(b, item_clone);
727        });
728        let handle2 = tokio::spawn(async move {
729            let b = rx2.recv().await.unwrap();
730            assert_eq!(b, item);
731        });
732
733        bucket.insert(&"test1".into(), "GK".into(), 1).await?;
734
735        let _ = futures::join!(handle1, handle2);
736        Ok(())
737    }
738}
dynamo_runtime/storage/kv.rs

dynamo_runtime/storage/
kv.rs
