fix(kv): Fixes key listing for async client to match go client

async-nats/src/jetstream/kv/mod.rs

@@ -636,7 +636,7 @@ impl Store {
         })
     }
 
-    /// Returns a [futures::Stream] that allows iterating over all keys in the bucket.
+    /// Returns an iterator of `String` over all keys in the bucket.
     ///
     /// # Examples
     ///
@@ -651,8 +651,8 @@ impl Store {
     ///     history: 10,
     ///     ..Default::default()
     /// }).await?;
-    /// let mut entries = kv.keys().await?;
-    /// while let Some(key) = entries.next() {
+    /// let mut keys = kv.keys().await?;
+    /// for key in keys {
     ///     println!("key: {:?}", key);
     /// }
     /// # Ok(())
@@ -669,6 +669,8 @@ impl Store {
                 filter_subject: subject,
                 headers_only: true,
                 replay_policy: super::consumer::ReplayPolicy::Instant,
+                // We only need to know the latest state for each key, not the whole history
+                deliver_policy: DeliverPolicy::LastPerSubject,
                 ..Default::default()
             })
             .await?;
@@ -682,7 +684,10 @@ impl Store {
 
         let mut keys = HashSet::new();
         while let Some(entry) = entries.try_next().await? {
-            keys.insert(entry.key);
+            // Filter out deleted keys
+            if !matches!(entry.operation, Operation::Purge | Operation::Delete) {
+                keys.insert(entry.key);
+            }
         }
         Ok(keys.into_iter())
     }

async-nats/tests/kv_tests.rs

@@ -472,6 +472,19 @@ mod kv {
         keys.sort();
         assert_eq!(vec!["bar", "foo"], keys);
 
+        // Delete a key and make sure it doesn't show up in the keys list
+        kv.delete("bar").await.unwrap();
+        let keys = kv.keys().await.unwrap().collect::<Vec<String>>();
+        assert_eq!(vec!["foo"], keys, "Deleted key shouldn't appear in list");
+
+        // Put the key back, and then purge and make sure the key doesn't show up
+        for i in 0..10 {
+            kv.put("bar", i.to_string().into()).await.unwrap();
+        }
+        kv.purge("foo").await.unwrap();
+        let keys = kv.keys().await.unwrap().collect::<Vec<String>>();
+        assert_eq!(vec!["bar"], keys, "Purged key shouldn't appear in the list");
+
         let kv = context
             .create_key_value(async_nats::jetstream::kv::Config {
                 bucket: "history2".to_string(),