feat(ext/kv): return versionstamp from set/commit (#18512)

This commit updates the `Deno.Kv` API to return the new commited
versionstamp for the mutated data from `db.set` and `ao.commit`. This is
returned in the form of a `Deno.KvCommitResult` object that has a
`versionstamp` property.

cli/tests/unit/kv_test.ts

@@ -64,7 +64,8 @@ dbTest("basic read-write-delete and versionstamps", async (db) => {
   assertEquals(result1.value, null);
   assertEquals(result1.versionstamp, null);
 
-  await db.set(["a"], "b");
+  const setRes = await db.set(["a"], "b");
+  assertEquals(setRes.versionstamp, "00000000000000010000");
   const result2 = await db.get(["a"]);
   assertEquals(result2.key, ["a"]);
   assertEquals(result2.value, "b");
@@ -177,21 +178,22 @@ dbTest("compare and mutate", async (db) => {
   const currentValue = await db.get(["t"]);
   assertEquals(currentValue.versionstamp, "00000000000000010000");
 
-  let ok = await db.atomic()
+  let res = await db.atomic()
     .check({ key: ["t"], versionstamp: currentValue.versionstamp })
     .set(currentValue.key, "2")
     .commit();
-  assertEquals(ok, true);
+  assert(res);
+  assertEquals(res.versionstamp, "00000000000000020000");
 
   const newValue = await db.get(["t"]);
   assertEquals(newValue.versionstamp, "00000000000000020000");
   assertEquals(newValue.value, "2");
 
-  ok = await db.atomic()
+  res = await db.atomic()
     .check({ key: ["t"], versionstamp: currentValue.versionstamp })
     .set(currentValue.key, "3")
     .commit();
-  assertEquals(ok, false);
+  assertEquals(res, null);
 
   const newValue2 = await db.get(["t"]);
   assertEquals(newValue2.versionstamp, "00000000000000020000");
@@ -199,21 +201,21 @@ dbTest("compare and mutate", async (db) => {
 });
 
 dbTest("compare and mutate not exists", async (db) => {
-  let ok = await db.atomic()
+  let res = await db.atomic()
     .check({ key: ["t"], versionstamp: null })
     .set(["t"], "1")
     .commit();
-  assertEquals(ok, true);
+  assert(res);
 
   const newValue = await db.get(["t"]);
   assertEquals(newValue.versionstamp, "00000000000000010000");
   assertEquals(newValue.value, "1");
 
-  ok = await db.atomic()
+  res = await db.atomic()
     .check({ key: ["t"], versionstamp: null })
     .set(["t"], "2")
     .commit();
-  assertEquals(ok, false);
+  assertEquals(res, null);
 });
 
 dbTest("compare multiple and mutate", async (db) => {
@@ -225,13 +227,13 @@ dbTest("compare multiple and mutate", async (db) => {
   const currentValue2 = await db.get(["t2"]);
   assertEquals(currentValue2.versionstamp, "00000000000000020000");
 
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .check({ key: ["t1"], versionstamp: currentValue1.versionstamp })
     .check({ key: ["t2"], versionstamp: currentValue2.versionstamp })
     .set(currentValue1.key, "3")
     .set(currentValue2.key, "4")
     .commit();
-  assertEquals(ok, true);
+  assert(res);
 
   const newValue1 = await db.get(["t1"]);
   assertEquals(newValue1.versionstamp, "00000000000000030000");
@@ -241,13 +243,13 @@ dbTest("compare multiple and mutate", async (db) => {
   assertEquals(newValue2.value, "4");
 
   // just one of the two checks failed
-  const ok2 = await db.atomic()
+  const res2 = await db.atomic()
     .check({ key: ["t1"], versionstamp: newValue1.versionstamp })
     .check({ key: ["t2"], versionstamp: null })
     .set(newValue1.key, "5")
     .set(newValue2.key, "6")
     .commit();
-  assertEquals(ok2, false);
+  assertEquals(res2, null);
 
   const newValue3 = await db.get(["t1"]);
   assertEquals(newValue3.versionstamp, "00000000000000030000");
@@ -259,79 +261,79 @@ dbTest("compare multiple and mutate", async (db) => {
 
 dbTest("atomic mutation ordering (set before delete)", async (db) => {
   await db.set(["a"], "1");
-  const ok1 = await db.atomic()
+  const res = await db.atomic()
     .set(["a"], "2")
     .delete(["a"])
     .commit();
-  assert(ok1);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, null);
 });
 
 dbTest("atomic mutation ordering (delete before set)", async (db) => {
   await db.set(["a"], "1");
-  const ok1 = await db.atomic()
+  const res = await db.atomic()
     .delete(["a"])
     .set(["a"], "2")
     .commit();
-  assert(ok1);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, "2");
 });
 
 dbTest("atomic mutation type=set", async (db) => {
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: "1", type: "set" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, "1");
 });
 
 dbTest("atomic mutation type=set overwrite", async (db) => {
   await db.set(["a"], "1");
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: "2", type: "set" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, "2");
 });
 
 dbTest("atomic mutation type=delete", async (db) => {
   await db.set(["a"], "1");
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], type: "delete" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, null);
 });
 
 dbTest("atomic mutation type=delete no exists", async (db) => {
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], type: "delete" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, null);
 });
 
 dbTest("atomic mutation type=sum", async (db) => {
   await db.set(["a"], new Deno.KvU64(10n));
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(1n), type: "sum" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, new Deno.KvU64(11n));
 });
 
 dbTest("atomic mutation type=sum no exists", async (db) => {
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(1n), type: "sum" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assert(result.value);
   assertEquals(result.value, new Deno.KvU64(1n));
@@ -339,21 +341,21 @@ dbTest("atomic mutation type=sum no exists", async (db) => {
 
 dbTest("atomic mutation type=sum wrap around", async (db) => {
   await db.set(["a"], new Deno.KvU64(0xffffffffffffffffn));
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(10n), type: "sum" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, new Deno.KvU64(9n));
 
-  const ok2 = await db.atomic()
+  const res2 = await db.atomic()
     .mutate({
       key: ["a"],
       value: new Deno.KvU64(0xffffffffffffffffn),
       type: "sum",
     })
     .commit();
-  assert(ok2);
+  assert(res2);
   const result2 = await db.get(["a"]);
   assertEquals(result2.value, new Deno.KvU64(8n));
 });
@@ -387,26 +389,26 @@ dbTest("atomic mutation type=sum wrong type in mutation", async (db) => {
 
 dbTest("atomic mutation type=min", async (db) => {
   await db.set(["a"], new Deno.KvU64(10n));
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(5n), type: "min" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, new Deno.KvU64(5n));
 
-  const ok2 = await db.atomic()
+  const res2 = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(15n), type: "min" })
     .commit();
-  assert(ok2);
+  assert(res2);
   const result2 = await db.get(["a"]);
   assertEquals(result2.value, new Deno.KvU64(5n));
 });
 
 dbTest("atomic mutation type=min no exists", async (db) => {
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(1n), type: "min" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assert(result.value);
   assertEquals(result.value, new Deno.KvU64(1n));
@@ -441,26 +443,26 @@ dbTest("atomic mutation type=min wrong type in mutation", async (db) => {
 
 dbTest("atomic mutation type=max", async (db) => {
   await db.set(["a"], new Deno.KvU64(10n));
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(5n), type: "max" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assertEquals(result.value, new Deno.KvU64(10n));
 
-  const ok2 = await db.atomic()
+  const res2 = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(15n), type: "max" })
     .commit();
-  assert(ok2);
+  assert(res2);
   const result2 = await db.get(["a"]);
   assertEquals(result2.value, new Deno.KvU64(15n));
 });
 
 dbTest("atomic mutation type=max no exists", async (db) => {
-  const ok = await db.atomic()
+  const res = await db.atomic()
     .mutate({ key: ["a"], value: new Deno.KvU64(1n), type: "max" })
     .commit();
-  assert(ok);
+  assert(res);
   const result = await db.get(["a"]);
   assert(result.value);
   assertEquals(result.value, new Deno.KvU64(1n));
@@ -1059,7 +1061,8 @@ dbTest("operation size limit", async (db) => {
     i,
   ) => ["a", i]);
 
-  assertEquals((await db.getMany(lastValidKeys)).length, 10);
+  const res = await db.getMany(lastValidKeys);
+  assertEquals(res.length, 10);
 
   await assertRejects(
     async () => await db.getMany(firstInvalidKeys),
@@ -1067,72 +1070,66 @@ dbTest("operation size limit", async (db) => {
     "too many ranges (max 10)",
   );
 
-  assertEquals(
+  const res2 = await collect(db.list({ prefix: ["a"] }, { batchSize: 1000 }));
-    (await collect(db.list({
+  assertEquals(res2.length, 0);
-      prefix: ["a"],
-    }, {
-      batchSize: 1000,
-    }))).length,
-    0,
-  );
 
   assertRejects(
-    async () =>
+    async () => await collect(db.list({ prefix: ["a"] }, { batchSize: 1001 })),
-      await collect(db.list({
-        prefix: ["a"],
-      }, {
-        batchSize: 1001,
-      })),
     TypeError,
     "too many entries (max 1000)",
   );
 
   // when batchSize is not specified, limit is used but is clamped to 500
   assertEquals(
-    (await collect(db.list({
+    (await collect(db.list({ prefix: ["a"] }, { limit: 1001 }))).length,
-      prefix: ["a"],
-    }, {
-      limit: 1001,
-    }))).length,
     0,
   );
 
-  assertEquals(
+  const res3 = await db.atomic()
-    await db.atomic().check(...lastValidKeys.map((key) => ({
+    .check(...lastValidKeys.map((key) => ({
       key,
       versionstamp: null,
-    }))).mutate(...lastValidKeys.map((key) => ({
+    })))
+    .mutate(...lastValidKeys.map((key) => ({
       key,
       type: "set",
       value: 1,
-    } satisfies Deno.KvMutation))).commit(),
+    } satisfies Deno.KvMutation)))
-    true,
+    .commit();
-  );
+  assert(res3);
 
   await assertRejects(
-    async () =>
+    async () => {
-      await db.atomic().check(...firstInvalidKeys.map((key) => ({
+      await db.atomic()
+        .check(...firstInvalidKeys.map((key) => ({
           key,
           versionstamp: null,
-      }))).mutate(...lastValidKeys.map((key) => ({
+        })))
+        .mutate(...lastValidKeys.map((key) => ({
           key,
           type: "set",
           value: 1,
-      } satisfies Deno.KvMutation))).commit(),
+        } satisfies Deno.KvMutation)))
+        .commit();
+    },
     TypeError,
     "too many checks (max 10)",
   );
 
   await assertRejects(
-    async () =>
+    async () => {
-      await db.atomic().check(...lastValidKeys.map((key) => ({
+      await db.atomic()
+        .check(...lastValidKeys.map((key) => ({
           key,
           versionstamp: null,
-      }))).mutate(...firstInvalidKeys.map((key) => ({
+        })))
+        .mutate(...firstInvalidKeys.map((key) => ({
           key,
           type: "set",
           value: 1,
-      } satisfies Deno.KvMutation))).commit(),
+        } satisfies Deno.KvMutation)))
+        .commit();
+    },
     TypeError,
     "too many mutations (max 10)",
   );

cli/tsc/dts/lib.deno.unstable.d.ts

@@ -1748,6 +1748,11 @@ declare namespace Deno {
     batchSize?: number;
   }
 
+  export interface KvCommitResult {
+    /** The versionstamp of the value committed to KV. */
+    versionstamp: string;
+  }
+
   /** **UNSTABLE**: New API, yet to be vetted.
    *
    * A check to perform as part of a {@linkcode Deno.AtomicOperation}. The check
@@ -1787,11 +1792,13 @@ declare namespace Deno {
    * performed and the operation will fail. One can then retry the read-modify-
    * write operation in a loop until it succeeds.
    *
-   * The `commit` method of an atomic operation returns a boolean indicating
+   * The `commit` method of an atomic operation returns a value indicating
    * whether checks passed and mutations were performed. If the operation failed
-   * because of a failed check, the return value will be `false`. If the
+   * because of a failed check, the return value will be `null`. If the
    * operation failed for any other reason (storage error, invalid value, etc.),
-   * an exception will be thrown.
+   * an exception will be thrown. If the operation succeeded, the return value
+   * will be a {@linkcode Deno.KvCommitResult} object containing the
+   * versionstamp of the value committed to KV.
    *
    * @category KV
    */
@@ -1821,17 +1828,19 @@ declare namespace Deno {
      */
     delete(key: KvKey): this;
     /**
-     * Commit the operation to the KV store. Returns a boolean indicating
+     * Commit the operation to the KV store. Returns a value indicating whether
-     * whether checks passed and mutations were performed. If the operation
+     * checks passed and mutations were performed. If the operation failed
-     * failed because of a failed check, the return value will be `false`. If
+     * because of a failed check, the return value will be `null`. If the
-     * the operation failed for any other reason (storage error, invalid value,
+     * operation failed for any other reason (storage error, invalid value,
-     * etc.), an exception will be thrown.
+     * etc.), an exception will be thrown. If the operation succeeded, the
+     * return value will be a {@linkcode Deno.KvCommitResult} object containing
+     * the versionstamp of the value committed to KV.
      *
-     * If the commit returns `false`, one may create a new atomic operation with
+     * If the commit returns `null`, one may create a new atomic operation with
      * updated checks and mutations and attempt to commit it again. See the note
      * on optimistic locking in the documentation for {@linkcode Deno.AtomicOperation}.
      */
-    commit(): Promise<boolean>;
+    commit(): Promise<KvCommitResult | null>;
   }
 
   /** **UNSTABLE**: New API, yet to be vetted.
@@ -1932,7 +1941,7 @@ declare namespace Deno {
      * await db.set(["foo"], "bar");
      * ```
      */
-    set(key: KvKey, value: unknown): Promise<void>;
+    set(key: KvKey, value: unknown): Promise<KvCommitResult>;
 
     /**
      * Delete the value for the given key from the database. If no value exists

ext/kv/01_db.ts

@@ -111,14 +111,15 @@ class Kv {
       [key, "set", value],
     ];
 
-    const result = await core.opAsync(
+    const versionstamp = await core.opAsync(
       "op_kv_atomic_write",
       this.#rid,
       checks,
       mutations,
       [],
     );
-    if (!result) throw new TypeError("Failed to set value");
+    if (versionstamp === null) throw new TypeError("Failed to set value");
+    return { versionstamp };
   }
 
   async delete(key: Deno.KvKey) {
@@ -255,15 +256,16 @@ class AtomicOperation {
     return this;
   }
 
-  async commit(): Promise<boolean> {
+  async commit(): Promise<Deno.KvCommitResult | null> {
-    const result = await core.opAsync(
+    const versionstamp = await core.opAsync(
       "op_kv_atomic_write",
       this.#rid,
       this.#checks,
       this.#mutations,
       [], // TODO(@losfair): enqueue
     );
-    return result;
+    if (versionstamp === null) return null;
+    return { versionstamp };
   }
 
   then() {

ext/kv/interface.rs

@@ -31,7 +31,10 @@ pub trait Database {
     options: SnapshotReadOptions,
   ) -> Result<Vec<ReadRangeOutput>, AnyError>;
 
-  async fn atomic_write(&self, write: AtomicWrite) -> Result<bool, AnyError>;
+  async fn atomic_write(
+    &self,
+    write: AtomicWrite,
+  ) -> Result<Option<CommitResult>, AnyError>;
 }
 
 /// Options for a snapshot read.
@@ -304,3 +307,9 @@ impl MutationKind {
     }
   }
 }
+
+/// The result of a successful commit of an atomic write operation.
+pub struct CommitResult {
+  /// The new versionstamp of the data that was committed.
+  pub versionstamp: Versionstamp,
+}

ext/kv/lib.rs

@@ -519,7 +519,7 @@ async fn op_kv_atomic_write<DBH>(
   checks: Vec<V8KvCheck>,
   mutations: Vec<V8KvMutation>,
   enqueues: Vec<V8Enqueue>,
-) -> Result<bool, AnyError>
+) -> Result<Option<String>, AnyError>
 where
   DBH: DatabaseHandler + 'static,
 {
@@ -585,7 +585,7 @@ where
 
   let result = db.atomic_write(atomic_write).await?;
 
-  Ok(result)
+  Ok(result.map(|res| hex::encode(res.versionstamp)))
 }
 
 // (prefix, start, end)

ext/kv/sqlite.rs

@@ -17,6 +17,7 @@ use rusqlite::OptionalExtension;
 use rusqlite::Transaction;
 
 use crate::AtomicWrite;
+use crate::CommitResult;
 use crate::Database;
 use crate::DatabaseHandler;
 use crate::KvEntry;
@@ -216,7 +217,10 @@ impl Database for SqliteDb {
     Ok(responses)
   }
 
-  async fn atomic_write(&self, write: AtomicWrite) -> Result<bool, AnyError> {
+  async fn atomic_write(
+    &self,
+    write: AtomicWrite,
+  ) -> Result<Option<CommitResult>, AnyError> {
     let mut db = self.0.borrow_mut();
 
     let tx = db.transaction()?;
@@ -228,7 +232,7 @@ impl Database for SqliteDb {
         .optional()?
         .map(version_to_versionstamp);
       if real_versionstamp != check.versionstamp {
-        return Ok(false);
+        return Ok(None);
       }
     }
 
@@ -273,7 +277,11 @@ impl Database for SqliteDb {
 
     tx.commit()?;
 
-    Ok(true)
+    let new_vesionstamp = version_to_versionstamp(version);
+
+    Ok(Some(CommitResult {
+      versionstamp: new_vesionstamp,
+    }))
   }
 }
 

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.68.0"
+channel = "1.68.2"
 components = ["rustfmt", "clippy"]