Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

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

Merged
merged 2 commits into from
May 17, 2022
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4af5a1b
[FEAT] implements the ability in KV to request a specific revision fo…
aricart May 16, 2022
6bb7332
[FEAT] added check when retrieving a key, but also specifying a seque…
aricart May 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 22 additions & 10 deletions nats-base-client/kv.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import {
KvPutOptions,
KvRemove,
KvStatus,
MsgRequest,
PurgeOpts,
PurgeResponse,
RetentionPolicy,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is a nuance in the Go client: when asking for a revision (sequence number), it is possible that a message is found at that sequence, but that does not mean that it will match the subject (the key). So if you had "A"/seq 1, "B"/seq 2 and you ask Get("A", 2), a message will be returned by the low level "getMessage(2)", but that does not match the key/subject "A", so you should return "key not found" in that case.

const ke = this.smToEntry(sm);
if (ke.key !== ek) {
return null;
}
return ke;
} catch (err) {
if (err.message === "no message found") {
return null;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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++;

Expand Down
2 changes: 1 addition & 1 deletion nats-base-client/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 },
Expand Down
3 changes: 0 additions & 3 deletions tests/jetstream_test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,6 @@ import {
JsMsg,
JsMsgCallback,
JSONCodec,
millis,
Nanos,
nanos,
NatsConnectionImpl,
NatsError,
Expand All @@ -51,7 +49,6 @@ import {
StringCodec,
} from "../nats-base-client/internal_mod.ts";
import {
assertArrayIncludes,
assertEquals,
assertRejects,
assertThrows,
Expand Down
36 changes: 36 additions & 0 deletions tests/kv_test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should extend the test to demonstrate the issue I was referring to. Do a "put("B", someValue)" and then a "get("A", rev==5)", and you should have a failure (and not whatever value was stored in "B").

await cleanup(ns, nc);
});