[FEAT] implements the ability in KV to request a specific revision for a key (#302)

- [FEAT] implements the ability in KV to request a specific revision for a key

- [FEAT] added check when retrieving a key and specifying a sequence (new feature), that the key matched. If not, null is returned as the sequence is not referencing the expected key.

- [FIX] conversion from stored message captured the key as provided by the client, and returned an entry with the said key. This issue is only a problem when combined with the new feature to specify a revision for the key, as the entry returned could have been not related to the key.

FIX #301

Author: aricart

Diff for: nats-base-client/kv.ts

@@ -35,6 +35,7 @@ import {
   KvPutOptions,
   KvRemove,
   KvStatus,
+  MsgRequest,
   PurgeOpts,
   PurgeResponse,
   RetentionPolicy,
@@ -311,10 +312,10 @@ export class Bucket implements KV, KvRemove {
     return data.length;
   }
 
-  smToEntry(key: string, sm: StoredMsg): KvEntry {
+  smToEntry(sm: StoredMsg): KvEntry {
     return {
       bucket: this.bucket,
-      key: key,
+      key: sm.subject.substring(this.prefixLen),
       value: sm.data,
       delta: 0,
       created: sm.time,
@@ -324,7 +325,7 @@ export class Bucket implements KV, KvRemove {
     };
   }
 
-  jmToEntry(_k: string, jm: JsMsg): KvEntry {
+  jmToEntry(jm: JsMsg): KvEntry {
     const key = this.decodeKey(jm.subject.substring(this.prefixLen));
     return {
       bucket: this.bucket,
@@ -367,14 +368,25 @@ export class Bucket implements KV, KvRemove {
     return pa.seq;
   }
 
-  async get(k: string): Promise<KvEntry | null> {
+  async get(
+    k: string,
+    opts?: { revision: number },
+  ): Promise<KvEntry | null> {
     const ek = this.encodeKey(k);
     this.validateKey(ek);
+
+    let arg: MsgRequest = { last_by_subj: this.fullKeyName(ek) };
+    if (opts && opts.revision > 0) {
+      arg = { seq: opts.revision };
+    }
+
     try {
-      const sm = await this.jsm.streams.getMessage(this.bucketName(), {
-        last_by_subj: this.fullKeyName(ek),
-      });
-      return this.smToEntry(k, sm);
+      const sm = await this.jsm.streams.getMessage(this.bucketName(), arg);
+      const ke = this.smToEntry(sm);
+      if (ke.key !== ek) {
+        return null;
+      }
+      return ke;
     } catch (err) {
       if (err.message === "no message found") {
         return null;
@@ -468,7 +480,7 @@ export class Bucket implements KV, KvRemove {
         return;
       }
       if (jm) {
-        const e = this.jmToEntry(k, jm);
+        const e = this.jmToEntry(jm);
         qi.push(e);
         qi.received++;
         //@ts-ignore - function will be removed
@@ -545,7 +557,7 @@ export class Bucket implements KV, KvRemove {
         return;
       }
       if (jm) {
-        const e = this.jmToEntry(k, jm);
+        const e = this.jmToEntry(jm);
         qi.push(e);
         qi.received++;
 

Diff for: nats-base-client/types.ts

@@ -914,7 +914,7 @@ export interface KvRemove {
 }
 
 export interface RoKV {
-  get(k: string): Promise<KvEntry | null>;
+  get(k: string, opts?: { revision: number }): Promise<KvEntry | null>;
   history(opts?: { key?: string }): Promise<QueuedIterator<KvEntry>>;
   watch(
     opts?: { key?: string; headers_only?: boolean; initializedFn?: callbackFn },

Diff for: tests/jetstream_test.ts

@@ -39,8 +39,6 @@ import {
   JsMsg,
   JsMsgCallback,
   JSONCodec,
-  millis,
-  Nanos,
   nanos,
   NatsConnectionImpl,
   NatsError,
@@ -51,7 +49,6 @@ import {
   StringCodec,
 } from "../nats-base-client/internal_mod.ts";
 import {
-  assertArrayIncludes,
   assertEquals,
   assertRejects,
   assertThrows,

Diff for: tests/kv_test.ts

@@ -1271,3 +1271,39 @@ Deno.test("kv - watch init callback exceptions terminate the iterator", async ()
   assertEquals(err.message, "crash");
   await cleanup(ns, nc);
 });
+
+Deno.test("kv - get revision", async () => {
+  const { ns, nc } = await setup(
+    jetstreamServerConf({}, true),
+  );
+  const js = nc.jetstream();
+  const sc = StringCodec();
+
+  const b = await js.views.kv(nuid.next(), { history: 3 }) as Bucket;
+
+  async function check(key: string, value: string | null, revision = 0) {
+    const e = await b.get(key, { revision });
+    if (value === null) {
+      assertEquals(e, null);
+    } else {
+      assertEquals(sc.decode(e!.value), value);
+    }
+  }
+
+  await b.put("A", sc.encode("a"));
+  await b.put("A", sc.encode("b"));
+  await b.put("A", sc.encode("c"));
+
+  // expect null, as sequence 1, holds "A"
+  await check("B", null, 1);
+
+  await check("A", "c");
+  await check("A", "a", 1);
+  await check("A", "b", 2);
+
+  await b.put("A", sc.encode("d"));
+  await check("A", "d");
+  await check("A", null, 1);
+
+  await cleanup(ns, nc);
+});