feat(ext/kv): key expiration (#20091)

Co-authored-by: Luca Casonato <hello@lcas.dev>

cli/tests/unit/kv_test.ts

@@ -1806,3 +1806,129 @@ Deno.test({
     }
   },
 });
+
+Deno.test({
+  name: "kv expiration",
+  async fn() {
+    const filename = await Deno.makeTempFile({ prefix: "kv_expiration_db" });
+    try {
+      await Deno.remove(filename);
+    } catch {
+      // pass
+    }
+    let db: Deno.Kv | null = null;
+
+    try {
+      db = await Deno.openKv(filename);
+
+      await db.set(["a"], 1, { expireIn: 1000 });
+      await db.set(["b"], 2, { expireIn: 1000 });
+      assertEquals((await db.get(["a"])).value, 1);
+      assertEquals((await db.get(["b"])).value, 2);
+
+      // Value overwrite should also reset expiration
+      await db.set(["b"], 2, { expireIn: 3600 * 1000 });
+
+      // Wait for expiration
+      await sleep(1000);
+
+      // Re-open to trigger immediate cleanup
+      db.close();
+      db = null;
+      db = await Deno.openKv(filename);
+
+      let ok = false;
+      for (let i = 0; i < 50; i++) {
+        await sleep(100);
+        if (
+          JSON.stringify(
+            (await db.getMany([["a"], ["b"]])).map((x) => x.value),
+          ) === "[null,2]"
+        ) {
+          ok = true;
+          break;
+        }
+      }
+
+      if (!ok) {
+        throw new Error("Values did not expire");
+      }
+    } finally {
+      if (db) {
+        try {
+          db.close();
+        } catch {
+          // pass
+        }
+      }
+      try {
+        await Deno.remove(filename);
+      } catch {
+        // pass
+      }
+    }
+  },
+});
+
+Deno.test({
+  name: "kv expiration with atomic",
+  async fn() {
+    const filename = await Deno.makeTempFile({ prefix: "kv_expiration_db" });
+    try {
+      await Deno.remove(filename);
+    } catch {
+      // pass
+    }
+    let db: Deno.Kv | null = null;
+
+    try {
+      db = await Deno.openKv(filename);
+
+      await db.atomic().set(["a"], 1, { expireIn: 1000 }).set(["b"], 2, {
+        expireIn: 1000,
+      }).commit();
+      assertEquals((await db.getMany([["a"], ["b"]])).map((x) => x.value), [
+        1,
+        2,
+      ]);
+
+      // Wait for expiration
+      await sleep(1000);
+
+      // Re-open to trigger immediate cleanup
+      db.close();
+      db = null;
+      db = await Deno.openKv(filename);
+
+      let ok = false;
+      for (let i = 0; i < 50; i++) {
+        await sleep(100);
+        if (
+          JSON.stringify(
+            (await db.getMany([["a"], ["b"]])).map((x) => x.value),
+          ) === "[null,null]"
+        ) {
+          ok = true;
+          break;
+        }
+      }
+
+      if (!ok) {
+        throw new Error("Values did not expire");
+      }
+    } finally {
+      if (db) {
+        try {
+          db.close();
+        } catch {
+          // pass
+        }
+      }
+      try {
+        await Deno.remove(filename);
+      } catch {
+        // pass
+      }
+    }
+  },
+});

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

@@ -1357,7 +1357,13 @@ declare namespace Deno {
    * mutation is applied to the key.
    *
    * - `set` - Sets the value of the key to the given value, overwriting any
-   *   existing value.
+   *   existing value. Optionally an `expireIn` option can be specified to
+   *   set a time-to-live (TTL) for the key. The TTL is specified in
+   *   milliseconds, and the key will be deleted from the database at earliest
+   *   after the specified number of milliseconds have elapsed. Once the
+   *   specified duration has passed, the key may still be visible for some
+   *   additional time. If the `expireIn` option is not specified, the key will
+   *   not expire.
    * - `delete` - Deletes the key from the database. The mutation is a no-op if
    *   the key does not exist.
    * - `sum` - Adds the given value to the existing value of the key. Both the
@@ -1379,7 +1385,7 @@ declare namespace Deno {
   export type KvMutation =
     & { key: KvKey }
     & (
-      | { type: "set"; value: unknown }
+      | { type: "set"; value: unknown; expireIn?: number }
       | { type: "delete" }
       | { type: "sum"; value: KvU64 }
       | { type: "max"; value: KvU64 }
@@ -1591,8 +1597,15 @@ declare namespace Deno {
     /**
      * Add to the operation a mutation that sets the value of the specified key
      * to the specified value if all checks pass during the commit.
+     *
+     * Optionally an `expireIn` option can be specified to set a time-to-live
+     * (TTL) for the key. The TTL is specified in milliseconds, and the key will
+     * be deleted from the database at earliest after the specified number of
+     * milliseconds have elapsed. Once the specified duration has passed, the
+     * key may still be visible for some additional time. If the `expireIn`
+     * option is not specified, the key will not expire.
      */
-    set(key: KvKey, value: unknown): this;
+    set(key: KvKey, value: unknown, options?: { expireIn?: number }): this;
     /**
      * Add to the operation a mutation that deletes the specified key if all
      * checks pass during the commit.
@@ -1721,8 +1734,19 @@ declare namespace Deno {
      * const db = await Deno.openKv();
      * await db.set(["foo"], "bar");
      * ```
+     *
+     * Optionally an `expireIn` option can be specified to set a time-to-live
+     * (TTL) for the key. The TTL is specified in milliseconds, and the key will
+     * be deleted from the database at earliest after the specified number of
+     * milliseconds have elapsed. Once the specified duration has passed, the
+     * key may still be visible for some additional time. If the `expireIn`
+     * option is not specified, the key will not expire.
      */
-    set(key: KvKey, value: unknown): Promise<KvCommitResult>;
+    set(
+      key: KvKey,
+      value: unknown,
+      options?: { expireIn?: number },
+    ): Promise<KvCommitResult>;
 
     /**
      * Delete the value for the given key from the database. If no value exists

ext/kv/01_db.ts

@@ -130,12 +130,15 @@ class Kv {
     });
   }
 
-  async set(key: Deno.KvKey, value: unknown) {
+  async set(key: Deno.KvKey, value: unknown, options?: { expireIn?: number }) {
     value = serializeValue(value);
 
     const checks: Deno.AtomicCheck[] = [];
+    const expireAt = typeof options?.expireIn === "number"
+      ? Date.now() + options.expireIn
+      : undefined;
     const mutations = [
-      [key, "set", value],
+      [key, "set", value, expireAt],
     ];
 
     const versionstamp = await core.opAsync(
@@ -152,7 +155,7 @@ class Kv {
   async delete(key: Deno.KvKey) {
     const checks: Deno.AtomicCheck[] = [];
     const mutations = [
-      [key, "delete", null],
+      [key, "delete", null, undefined],
     ];
 
     const result = await core.opAsync(
@@ -318,7 +321,7 @@ class AtomicOperation {
   #rid: number;
 
   #checks: [Deno.KvKey, string | null][] = [];
-  #mutations: [Deno.KvKey, string, RawValue | null][] = [];
+  #mutations: [Deno.KvKey, string, RawValue | null, number | undefined][] = [];
   #enqueues: [Uint8Array, number, Deno.KvKey[], number[] | null][] = [];
 
   constructor(rid: number) {
@@ -337,6 +340,7 @@ class AtomicOperation {
       const key = mutation.key;
       let type: string;
       let value: RawValue | null;
+      let expireAt: number | undefined = undefined;
       switch (mutation.type) {
         case "delete":
           type = "delete";
@@ -345,6 +349,10 @@ class AtomicOperation {
           }
           break;
         case "set":
+          if (typeof mutation.expireIn === "number") {
+            expireAt = Date.now() + mutation.expireIn;
+          }
+          /* falls through */
         case "sum":
         case "min":
         case "max":
@@ -357,33 +365,40 @@ class AtomicOperation {
         default:
           throw new TypeError("Invalid mutation type");
       }
-      this.#mutations.push([key, type, value]);
+      this.#mutations.push([key, type, value, expireAt]);
     }
     return this;
   }
 
   sum(key: Deno.KvKey, n: bigint): this {
-    this.#mutations.push([key, "sum", serializeValue(new KvU64(n))]);
+    this.#mutations.push([key, "sum", serializeValue(new KvU64(n)), undefined]);
     return this;
   }
 
   min(key: Deno.KvKey, n: bigint): this {
-    this.#mutations.push([key, "min", serializeValue(new KvU64(n))]);
+    this.#mutations.push([key, "min", serializeValue(new KvU64(n)), undefined]);
     return this;
   }
 
   max(key: Deno.KvKey, n: bigint): this {
-    this.#mutations.push([key, "max", serializeValue(new KvU64(n))]);
+    this.#mutations.push([key, "max", serializeValue(new KvU64(n)), undefined]);
     return this;
   }
 
-  set(key: Deno.KvKey, value: unknown): this {
-    this.#mutations.push([key, "set", serializeValue(value)]);
+  set(
+    key: Deno.KvKey,
+    value: unknown,
+    options?: { expireIn?: number },
+  ): this {
+    const expireAt = typeof options?.expireIn === "number"
+      ? Date.now() + options.expireIn
+      : undefined;
+    this.#mutations.push([key, "set", serializeValue(value), expireAt]);
     return this;
   }
 
   delete(key: Deno.KvKey): this {
-    this.#mutations.push([key, "delete", null]);
+    this.#mutations.push([key, "delete", null, undefined]);
     return this;
   }
 

ext/kv/interface.rs

@@ -237,6 +237,7 @@ pub struct KvCheck {
 pub struct KvMutation {
   pub key: Vec<u8>,
   pub kind: MutationKind,
+  pub expire_at: Option<u64>,
 }
 
 /// A request to enqueue a message to the database. This message is delivered

ext/kv/lib.rs

@@ -375,7 +375,7 @@ impl TryFrom<V8KvCheck> for KvCheck {
   }
 }
 
-type V8KvMutation = (KvKey, String, Option<FromV8Value>);
+type V8KvMutation = (KvKey, String, Option<FromV8Value>, Option<u64>);
 
 impl TryFrom<V8KvMutation> for KvMutation {
   type Error = AnyError;
@@ -396,7 +396,11 @@ impl TryFrom<V8KvMutation> for KvMutation {
         )))
       }
     };
-    Ok(KvMutation { key, kind })
+    Ok(KvMutation {
+      key,
+      kind,
+      expire_at: value.3,
+    })
   }
 }