feat: disposable Deno resources

This commit implements Symbol.dispose and Symbol.asyncDispose for
the relevant resources.

ext/fs/30_fs.js

@@ -34,7 +34,7 @@ import {
   ReadableStreamPrototype,
   writableStreamForRid,
 } from "ext:deno_web/06_streams.js";
-import { pathFromURL } from "ext:deno_web/00_infra.js";
+import { pathFromURL, SymbolDispose } from "ext:deno_web/00_infra.js";
 
 function chmodSync(path, mode) {
   ops.op_fs_chmod_sync(pathFromURL(path), mode);
@@ -669,6 +669,10 @@ class FsFile {
     }
     return this.#writable;
   }
+
+  [SymbolDispose]() {
+    core.tryClose(this.rid);
+  }
 }
 
 function checkOpenOptions(options) {

ext/http/00_serve.js

@@ -36,6 +36,7 @@ import {
 } from "ext:deno_web/06_streams.js";
 import { listen, listenOptionApiName, TcpConn } from "ext:deno_net/01_net.js";
 import { listenTls } from "ext:deno_net/02_tls.js";
+import { SymbolAsyncDispose } from "ext:deno_web/00_infra.js";
 const {
   ArrayPrototypePush,
   ObjectHasOwn,
@@ -343,6 +344,7 @@ class CallbackContext {
   fallbackHost;
   serverRid;
   closed;
+  /** @type {Promise<void> | undefined} */
   closing;
   listener;
 
@@ -682,22 +684,25 @@ function serveHttpOn(context, callback) {
       PromisePrototypeCatch(callback(req), promiseErrorHandler);
     }
 
-    if (!context.closed && !context.closing) {
-      context.closed = true;
-      await op_http_close(rid, false);
+    if (!context.closing && !context.closed) {
+      context.closing = op_http_close(rid, false);
       context.close();
     }
+
+    await context.closing;
+    context.close();
+    context.closed = true;
   })();
 
   return {
     finished,
     async shutdown() {
-      if (!context.closed && !context.closing) {
+      if (!context.closing && !context.closed) {
         // Shut this HTTP server down gracefully
-        context.closing = true;
-        await op_http_close(context.serverRid, true);
-        context.closed = true;
+        context.closing = op_http_close(rid, true);
       }
+      await context.closing;
+      context.closed = true;
     },
     ref() {
       ref = true;
@@ -711,6 +716,9 @@ function serveHttpOn(context, callback) {
         core.unrefOp(currentPromise[promiseIdSymbol]);
       }
     },
+    [SymbolAsyncDispose]() {
+      return this.shutdown();
+    },
   };
 }
 

ext/http/01_http.js

@@ -44,6 +44,7 @@ import {
   ReadableStreamPrototype,
 } from "ext:deno_web/06_streams.js";
 import { serve } from "ext:deno_http/00_serve.js";
+import { SymbolDispose } from "ext:deno_web/00_infra.js";
 const {
   ArrayPrototypeIncludes,
   ArrayPrototypeMap,
@@ -69,6 +70,9 @@ const {
 const connErrorSymbol = Symbol("connError");
 const _deferred = Symbol("upgradeHttpDeferred");
 
+/** @type {(self: HttpConn, rid: number) => boolean} */
+let deleteManagedResource;
+
 class HttpConn {
   #rid = 0;
   #closed = false;
@@ -79,7 +83,12 @@ class HttpConn {
   // that were created during lifecycle of this request.
   // When the connection is closed these resources should be closed
   // as well.
-  managedResources = new SafeSet();
+  #managedResources = new SafeSet();
+
+  static {
+    deleteManagedResource = (self, rid) =>
+      SetPrototypeDelete(self.#managedResources, rid);
+  }
 
   constructor(rid, remoteAddr, localAddr) {
     this.#rid = rid;
@@ -123,7 +132,7 @@ class HttpConn {
     }
 
     const { 0: streamRid, 1: method, 2: url } = nextRequest;
-    SetPrototypeAdd(this.managedResources, streamRid);
+    SetPrototypeAdd(this.#managedResources, streamRid);
 
     /** @type {ReadableStream<Uint8Array> | undefined} */
     let body = null;
@@ -167,13 +176,21 @@ class HttpConn {
     if (!this.#closed) {
       this.#closed = true;
       core.close(this.#rid);
-      for (const rid of new SafeSetIterator(this.managedResources)) {
-        SetPrototypeDelete(this.managedResources, rid);
+      for (const rid of new SafeSetIterator(this.#managedResources)) {
+        SetPrototypeDelete(this.#managedResources, rid);
         core.close(rid);
       }
     }
   }
 
+  [SymbolDispose]() {
+    core.tryClose(this.#rid);
+    for (const rid of new SafeSetIterator(this.#managedResources)) {
+      SetPrototypeDelete(this.#managedResources, rid);
+      core.tryClose(rid);
+    }
+  }
+
   [SymbolAsyncIterator]() {
     // deno-lint-ignore no-this-alias
     const httpConn = this;
@@ -395,7 +412,7 @@ function createRespondWith(
       abortController.abort(error);
       throw error;
     } finally {
-      if (SetPrototypeDelete(httpConn.managedResources, streamRid)) {
+      if (deleteManagedResource(httpConn, streamRid)) {
         core.close(streamRid);
       }
     }

ext/io/12_io.js

@@ -11,6 +11,7 @@ import {
   readableStreamForRid,
   writableStreamForRid,
 } from "ext:deno_web/06_streams.js";
+import { SymbolDispose } from "ext:deno_web/00_infra.js";
 const {
   Uint8Array,
   ArrayPrototypePush,
@@ -255,6 +256,10 @@ class Stdin {
     const cbreak = !!(options.cbreak ?? false);
     ops.op_stdin_set_raw(mode, cbreak);
   }
+
+  [SymbolDispose]() {
+    core.tryClose(this.rid);
+  }
 }
 
 class Stdout {

ext/kv/01_db.ts

@@ -12,6 +12,7 @@ const {
   SymbolToStringTag,
   Uint8ArrayPrototype,
 } = globalThis.__bootstrap.primordials;
+import { SymbolDispose } from "ext:deno_web/00_infra.js";
 const core = Deno.core;
 const ops = core.ops;
 
@@ -299,6 +300,10 @@ class Kv {
   close() {
     core.close(this.#rid);
   }
+
+  [SymbolDispose]() {
+    core.tryClose(this.#rid);
+  }
 }
 
 class AtomicOperation {

ext/kv/lib.rs

@@ -45,7 +45,7 @@ const MAX_TOTAL_MUTATION_SIZE_BYTES: usize = 800 * 1024;
 const MAX_TOTAL_KEY_SIZE_BYTES: usize = 80 * 1024;
 
 deno_core::extension!(deno_kv,
-  deps = [ deno_console ],
+  deps = [ deno_console, deno_web ],
   parameters = [ DBH: DatabaseHandler ],
   ops = [
     op_kv_database_open<DBH>,

ext/net/01_net.js

@@ -9,6 +9,8 @@ import {
   writableStreamForRid,
 } from "ext:deno_web/06_streams.js";
 import * as abortSignal from "ext:deno_web/03_abort_signal.js";
+import { SymbolDispose } from "ext:deno_web/00_infra.js";
+
 const primordials = globalThis.__bootstrap.primordials;
 const {
   ArrayPrototypeFilter,
@@ -160,6 +162,10 @@ class Conn {
       (id) => core.unrefOp(id),
     );
   }
+
+  [SymbolDispose]() {
+    core.tryClose(this.#rid);
+  }
 }
 
 class TcpConn extends Conn {
@@ -249,6 +255,10 @@ class Listener {
     core.close(this.rid);
   }
 
+  [SymbolDispose]() {
+    core.tryClose(this.#rid);
+  }
+
   [SymbolAsyncIterator]() {
     return this;
   }

ext/web/00_infra.js

@@ -458,6 +458,11 @@ function pathFromURL(pathOrUrl) {
 // it in unit tests
 internals.pathFromURL = pathFromURL;
 
+// deno-lint-ignore prefer-primordials
+export const SymbolDispose = Symbol.dispose ?? Symbol("Symbol.dispose");
+// deno-lint-ignore prefer-primordials
+export const SymbolAsyncDispose = Symbol.asyncDispose ?? Symbol("Symbol.asyncDispose");
+
 export {
   ASCII_ALPHA,
   ASCII_ALPHANUMERIC,

runtime/js/40_fs_events.js

@@ -9,6 +9,8 @@ const {
   PromiseResolve,
   SymbolAsyncIterator,
 } = primordials;
+import { SymbolDispose } from "ext:deno_web/00_infra.js";
+
 class FsWatcher {
   #rid = 0;
 
@@ -51,6 +53,10 @@ class FsWatcher {
   [SymbolAsyncIterator]() {
     return this;
   }
+
+  [SymbolDispose]() {
+    core.tryClose(this.#rid);
+  }
 }
 
 function watchFs(

runtime/js/40_process.js

@@ -18,7 +18,7 @@ const {
 } = primordials;
 import { FsFile } from "ext:deno_fs/30_fs.js";
 import { readAll } from "ext:deno_io/12_io.js";
-import { assert, pathFromURL } from "ext:deno_web/00_infra.js";
+import { assert, pathFromURL, SymbolAsyncDispose } from "ext:deno_web/00_infra.js";
 import * as abortSignal from "ext:deno_web/03_abort_signal.js";
 import {
   readableStreamCollectIntoUint8Array,
@@ -201,7 +201,6 @@ class ChildProcess {
   #rid;
   #waitPromiseId;
   #waitComplete = false;
-  #unrefed = false;
 
   #pid;
   get pid() {
@@ -216,7 +215,6 @@ class ChildProcess {
     return this.#stdin;
   }
 
-  #stdoutRid;
   #stdout = null;
   get stdout() {
     if (this.#stdout == null) {
@@ -225,7 +223,6 @@ class ChildProcess {
     return this.#stdout;
   }
 
-  #stderrRid;
   #stderr = null;
   get stderr() {
     if (this.#stderr == null) {
@@ -254,12 +251,10 @@ class ChildProcess {
     }
 
     if (stdoutRid !== null) {
-      this.#stdoutRid = stdoutRid;
       this.#stdout = readableStreamForRidUnrefable(stdoutRid);
     }
 
     if (stderrRid !== null) {
-      this.#stderrRid = stderrRid;
       this.#stderr = readableStreamForRidUnrefable(stderrRid);
     }
 
@@ -324,15 +319,22 @@ class ChildProcess {
     ops.op_spawn_kill(this.#rid, signo);
   }
 
+  async [SymbolAsyncDispose]() {
+    try { 
+      ops.op_spawn_kill(this.#rid, "SIGTERM")
+    } catch {
+      // ignore errors from killing the process (such as ESRCH or BadResource)
+    }
+    await this.#status;
+  }
+
   ref() {
-    this.#unrefed = false;
     core.refOp(this.#waitPromiseId);
     if (this.#stdout) readableStreamForRidUnrefableRef(this.#stdout);
     if (this.#stderr) readableStreamForRidUnrefableRef(this.#stderr);
   }
 
   unref() {
-    this.#unrefed = true;
     core.unrefOp(this.#waitPromiseId);
     if (this.#stdout) readableStreamForRidUnrefableUnref(this.#stdout);
     if (this.#stderr) readableStreamForRidUnrefableUnref(this.#stderr);

runtime/js/99_main.js

@@ -70,11 +70,23 @@ import {
   windowOrWorkerGlobalScope,
   workerRuntimeGlobalProperties,
 } from "ext:runtime/98_global_scope.js";
-
-// deno-lint-ignore prefer-primordials
-Symbol.dispose ??= Symbol("Symbol.dispose");
-// deno-lint-ignore prefer-primordials
-Symbol.asyncDispose ??= Symbol("Symbol.asyncDispose");
+import { SymbolAsyncDispose, SymbolDispose } from "ext:deno_web/00_infra.js";
+
+if (Symbol.dispose) throw "V8 supports Symbol.dispose now, no need to shim it!";
+ObjectDefineProperties(Symbol, {
+  dispose: {
+    value: SymbolAsyncDispose,
+    enumerable: false,
+    writable: false,
+    configurable: false,
+  },
+  asyncDispose: {
+    value: SymbolDispose,
+    enumerable: false,
+    writable: false,
+    configurable: false,
+  },
+});
 
 let windowIsClosing = false;
 let globalThis_;