🐛 Bug Report — Runtime APIs: AsyncLocalStorage store is never collected when it references a pending promise with a reaction attached inside run() #7611

import { AsyncLocalStorage } from "node:async_hooks";

const als = new AsyncLocalStorage();
const PAYLOAD_BYTES = 100 * 1024;
const stats = {};
const registry = new FinalizationRegistry((route) => {
  stats[route].finalized++;
});

function makeStore(route) {
  stats[route] ??= { created: 0, finalized: 0 };
  stats[route].created++;

  // `interrupted` never settles unless store.interrupt() is called.
  let interrupt;
  const interrupted = new Promise((resolve) => {
    interrupt = () => resolve("interrupted");
  });
  const store = {
    payload: new Uint8Array(PAYLOAD_BYTES).fill(1),
    interrupt,
    pending: new Promise(() => {}),
  };
  registry.register(store, route);
  return { store, interrupted };
}

const work = () => Promise.resolve().then(() => "done");

const routes = {
  // ALS store + race against a never-settling promise the store can reach.
  async leak() {
    const { store, interrupted } = makeStore("leak");
    await als.run(store, async () => await Promise.race([work(), interrupted]));
  },
  // Smallest case: one .then on a never-settling promise the store holds, attached inside als.run.
  async minimal() {
    const { store } = makeStore("minimal");
    als.run(store, () => {
      store.pending.then(() => {});
    });
  },
  // Same, but `interrupted` is settled in finally.
  async settled() {
    const { store, interrupted } = makeStore("settled");
    await als.run(store, async () => {
      try {
        await Promise.race([work(), interrupted]);
      } finally {
        store.interrupt();
      }
    });
  },
  // Store holds the pending promise but nothing reacts to it.
  async "no-reaction"() {
    const { store } = makeStore("no-reaction");
    await als.run(store, async () => await work());
  },
  // Same race, no AsyncLocalStorage.
  async "no-als"() {
    const { interrupted } = makeStore("no-als");
    await (async () => await Promise.race([work(), interrupted]))();
  },
};

async function settle() {
  for (let i = 0; i < 5; i++) {
    gc();
    await new Promise((resolve) => setTimeout(resolve, 10));
  }
}

export default {
  async fetch(request) {
    if (typeof gc !== "function") {
      throw new Error("gc() is not exposed: run with --expose-gc / v8Flags");
    }
    const path = new URL(request.url).pathname.slice(1);
    if (path === "stats") {
      await settle();
      const out = {};
      for (const [route, s] of Object.entries(stats)) {
        out[route] = { ...s, retained: s.created - s.finalized };
      }
      return Response.json(out);
    }
    if (!Object.hasOwn(routes, path)) return new Response("not found", { status: 404 });
    await routes[path]();
    return new Response("ok");
  },
};

// Usage: node --expose-gc run.mjs node            (runs worker.js in-process on Node)
//        node run.mjs http://127.0.0.1:8799        (drives a running workerd)
import worker from "./worker.js";

const N = Number(process.env.N ?? 1000);
const ROUTES = (process.env.ROUTES ?? "leak,minimal,settled,no-reaction,no-als").split(",");
const target = process.argv[2];
if (!target) throw new Error("usage: run.mjs node | run.mjs <workerd base url>");

const call =
  target === "node"
    ? (p) => worker.fetch(new Request(`http://x/${p}`))
    : (p) => fetch(`${target}/${p}`);

async function ok(pending) {
  const res = await pending;
  if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`);
  return res;
}

for (const route of ROUTES) for (let i = 0; i < N; i++) await ok(call(route));
const stats = await (await ok(call("stats"))).json();
console.log(`${target === "node" ? process.version : "workerd"}, ${N} requests per route`);
console.table(stats);
for (const r of ROUTES) {
  if (stats[r].created !== N) throw new Error(`${r}: created ${stats[r].created} != ${N}`);
}