Compile events.once with an AbortSignal and over streams, and combine Promise.race entries around it

docs/content/limitations/builtin-apis/eventemitter.mdx

@@ -1,4 +1,5 @@
 - **EventEmitter:** Literal event names use fixed argument tuples.
   Computed string names require a constant prefix or suffix that excludes `error`, meta events, and stream-internal events; their arguments must match the exact arity and support checked conversion.
   Unrestricted names, symbols, names that may reach internal events, and typed `listeners()`/`rawListeners()` results for names with possible computed registrations are unsupported.
+  `events.once` accepts EventEmitter and stream instances with a literal event name and an optional object-literal `{ signal }`; a stream's `data` event, other options values, and EventTarget emitters are unsupported.
   Reading `defaultMaxListeners` through a CommonJS destructured or member binding (`const { defaultMaxListeners } = require("events")`) is unsupported.

packages/compiler/ambient/scriptc-node-fallback.d.ts

@@ -6569,7 +6569,7 @@ declare module "events" {
     export { internal as EventEmitter };
     // The event's arguments, as checked-dynamic values.
     export function once(
-      emitter: EventEmitter,
+      emitter: EventEmitter | NodeJS.ReadableStream | NodeJS.WritableStream,
       eventName: string,
       options?: { signal?: AbortSignal },
     ): Promise<unknown[]>;

packages/compiler/src/backend/executable-features.ts

@@ -27,6 +27,7 @@ export function executableLinkFeatures(mod: IrModule): NativeLinkFeatures {
     dynAsync: features.dynAsync,
     ...(features.dynTier ? { dynTier: true } : {}),
     ...(features.streamDyn ? { streamDyn: true } : {}),
+    ...(features.eventsOnceSignal ? { eventsOnceSignal: true } : {}),
     events: features.processEvents,
     emitter: features.emitter,
     symbol: features.symbol,

packages/compiler/src/backend/llvm/emitter.ts

@@ -5599,8 +5599,11 @@ export class LlEmitter {
     return emitterFixedAdapter(this, cbT);
   }
 
-  emitterOnceAdapter(boxerT: IrType & { kind: "func" }): { fn: string; shim: string } {
-    return emitterOnceAdapter(this, boxerT);
+  emitterOnceAdapter(
+    boxerT: IrType & { kind: "func" },
+    fire?: string,
+  ): { fn: string; shim: string } {
+    return emitterOnceAdapter(this, boxerT, fire);
   }
 
   wrapEmitterListener(target: string, adapterFn: string): string {

packages/compiler/src/backend/llvm/expr-callbacks.ts

@@ -864,23 +864,27 @@ export function emitterFixedAdapter(
  * (the event's tuple to any[]): the runtime-built listener closure holds
  * the boxer in caps[0], so the adapter calls it with the event's typed
  * arguments, as emitterFixedAdapter calls a listener, and hands the array
- * to scr_events_once_fire. */
+ * to scr_events_once_fire (or `fire`, the signal form's entry). */
 export function emitterOnceAdapter(
   host: LlvmEmitterContext,
   boxerT: IrType & { kind: "func" },
+  fire = "scr_events_once_fire",
 ): { fn: string; shim: string } {
   const cursor = boxerT.params.length > 4;
   const shim = cursor ? "scr_ee_inv_args" : `scr_ee_inv_fixed${boxerT.params.length}`;
   if (cursor) host.declare("declare ptr @scr_ee_arg_next(ptr)");
   host.declare(`declare void @${shim}(ptr, ptr)`);
-  const key = `eeOnce:${typeKey(boxerT)}`;
+  const key =
+    fire === "scr_events_once_fire"
+      ? `eeOnce:${typeKey(boxerT)}`
+      : `eeOnce:${fire}:${typeKey(boxerT)}`;
   let sym = host.resolveThunks.get(key);
   if (sym) return { fn: sym, shim };
   sym = `sc_ee_once_${host.resolveThunks.size}`;
   host.resolveThunks.set(key, sym);
   host.declare(`declare ptr @scr_box_get_ref(ptr)`);
   host.declare(`declare void @scr_closure_release(ptr)`);
-  host.declare(`declare void @scr_events_once_fire(ptr, ptr)`);
+  host.declare(`declare void @${fire}(ptr, ptr)`);
   const params = cursor ? "ptr %args" : boxerT.params.map((_, i) => `ptr %a${i}`).join(", ");
   const d: string[] = [
     `define internal void @${sym}(ptr %cb${params ? ", " + params : ""}) ${FN_ATTRS} { ; events.once adapter ${typeKey(boxerT)}`,
@@ -911,7 +915,7 @@ export function emitterOnceAdapter(
     `  %fn = load ptr, ptr %fnp`,
     `  %arr = call ptr %fn(${passed.join(", ")}) ; the boxed arguments, +1`,
     `  call void @scr_closure_release(ptr %boxer)`,
-    `  call void @scr_events_once_fire(ptr %cb, ptr %arr) ; moves %arr`,
+    `  call void @${fire}(ptr %cb, ptr %arr) ; moves %arr`,
     `  ret void`,
     `}`,
     ``,

packages/compiler/src/backend/llvm/lib-dispatch.ts

@@ -755,7 +755,8 @@ export function emitLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue {
   }
   const prefix = e.fn.slice(0, e.fn.indexOf(".")) as LibCallPrefix;
   // events.once builds typed listener adapters beside the emitter calls.
-  if (e.fn === "events.once") return host.emitErrorsEventsLibCall(e);
+  if (e.fn === "events.once" || e.fn === "events.onceSignal")
+    return host.emitErrorsEventsLibCall(e);
   switch (prefix) {
     case "fetch":
     case "abort":

packages/compiler/src/backend/llvm/lib-process.ts

@@ -731,27 +731,34 @@ export function emitErrorsEventsLibCall(host: LlvmEmitterContext, e: LibCallExpr
     host.emitAlwaysPendingUnwind();
     return out;
   }
-  if (e.fn === "events.once") {
-    // (recv, name, boxer, errorBoxer, listenForError): scr_events_once
-    // registers one-shot typed listeners behind adapters that box the
-    // event's tuple (and the 'error' tuple) into the fulfilled array. The
-    // runtime retains the boxers; the frame's temporaries release as usual.
-    // May-throw ('newListener' meta listeners run inside).
+  if (e.fn === "events.once" || e.fn === "events.onceSignal") {
+    // (recv, name, boxer, errorBoxer, listenForError[, signal]):
+    // scr_events_once registers one-shot typed listeners behind adapters
+    // that box the event's tuple (and the 'error' tuple) into the fulfilled
+    // array. The runtime retains the boxers; the frame's temporaries
+    // release as usual. May-throw ('newListener' meta listeners run
+    // inside). The signal form's adapters end in its own fire entry.
     const boxerT = e.args[2]!.type;
     const errorT = e.args[3]!.type;
     if (boxerT.kind !== "func" || errorT.kind !== "func")
-      throw new InternalCompilerError("llvm emitter bug: events.once boxer not a func");
+      throw new InternalCompilerError(`llvm emitter bug: ${e.fn} boxer not a func`);
+    const signal = e.fn === "events.onceSignal";
     const args = e.args.map((a) => host.emitExpr(a));
-    const event = host.emitterOnceAdapter(boxerT);
-    const error = host.emitterOnceAdapter(errorT);
-    host.declare(`declare ptr @scr_events_once(ptr, ptr, ptr, ptr, ptr, ptr, ptr, ptr)`);
+    const fire = signal ? "scr_events_once_signal_fire" : undefined;
+    const event = host.emitterOnceAdapter(boxerT, fire);
+    const error = host.emitterOnceAdapter(errorT, fire);
+    const entry = signal ? "scr_events_once_signal" : "scr_events_once";
+    host.declare(
+      `declare ptr @${entry}(ptr, ptr, ptr, ptr, ptr, ptr, ptr, ptr${signal ? ", ptr" : ""})`,
+    );
     const errorBoxer = B.tmp();
     B.line(`${errorBoxer} = select i1 ${args[4]!.name}, ptr ${args[3]!.name}, ptr null`);
     const t = B.tmp();
     B.line(
-      `${t} = call ptr @scr_events_once(ptr ${args[0]!.name}, ptr ${args[1]!.name}, ` +
+      `${t} = call ptr @${entry}(ptr ${args[0]!.name}, ptr ${args[1]!.name}, ` +
         `ptr ${args[2]!.name}, ptr @${event.fn}, ptr @${event.shim}, ` +
-        `ptr ${errorBoxer}, ptr @${error.fn}, ptr @${error.shim})`,
+        `ptr ${errorBoxer}, ptr @${error.fn}, ptr @${error.shim}` +
+        `${signal ? `, ptr ${args[5]!.name}` : ""})`,
     );
     const out = host.own({ name: t, type: e.type });
     host.emitPendingCheck();

packages/compiler/src/backend/native-link-info-core.ts

@@ -46,6 +46,9 @@ export interface NativeLinkFeatures {
   /** Checked-dynamic stream slots: the scr_stream_dyn.c unit. Absent (not
    * false) for programs without them. */
   streamDyn?: boolean;
+  /** events.once with options: scr_events_once_signal.c. Absent (not
+   * false) for programs without it. */
+  eventsOnceSignal?: boolean;
   events: boolean;
   emitter: boolean;
   symbol: boolean;

packages/compiler/src/backend/native/contracts.ts

@@ -116,6 +116,9 @@ export interface CcOptions {
   /** Checked-dynamic stream slots (stream.dynNative / stream.pipeDyn):
    * compiles scr_stream_dyn.c. */
   streamDyn?: boolean;
+  /** events.once with options (events.onceSignal): compiles
+   * scr_events_once_signal.c. */
+  eventsOnceSignal?: boolean;
   /** The program constructs EventTarget/Event/CustomEvent values or
    * calls events.addAbortListener (the event.* libCalls), or uses the
    * native fetch AbortSignal built on the same engine: compiles

packages/compiler/src/backend/native/executable.ts

@@ -220,6 +220,9 @@
   // measures with.
   unit((u) => u.opts.events && u.regex, "scr_readline_tty.c"),
   unit((u) => u.opts.emitter || u.opts.workers, "scr_events_emitter.c"),
+  // events.once with options registers through the emitter and
+  // event-target units.
+  unit((u) => u.opts.eventsOnceSignal === true, "scr_events_once_signal.c"),
   // The checked-dynamic HANDLE support unit (listener gate + runtime
   // adapter closures): every referencing unit is one of the emitter or
   // net families (http implies net), so handle-free binaries keep
@@ -781,7 +784,7 @@
           ? "This build includes native FFI link inputs. Check that every symbol and system library exists, " +
             "that archive/object ordering is correct, and that each input matches the selected target."
           : `Check the supplied native source and the selected ${ccName} toolchain.`;
       throw new CcCompileError(
         ccName,
         stderr,
         `${ccName} failed compiling ${opts.cPath}.\n` + `${guidance}\n\n${stderr}`,

packages/compiler/src/backend/runtime-pack-core.ts

@@ -85,6 +85,7 @@ export interface RuntimeFeatureSet extends NativeLinkFeatures {
   wasm: boolean;
   dynTier: boolean;
   streamDyn: boolean;
+  eventsOnceSignal: boolean;
   warningReplace: boolean;
   stdioExtended: boolean;
   nativeFetch: boolean;
@@ -277,6 +278,7 @@ export function effectiveRuntimeFeatures(
     wasm: features.wasm === true,
     dynTier: features.dynTier === true,
     streamDyn: features.streamDyn === true,
+    eventsOnceSignal: features.eventsOnceSignal === true,
     warningReplace: features.warningReplace === true,
     stdioExtended: features.stdioExtended === true,
     nativeFetch,

packages/compiler/src/executable/executable-cache.ts

@@ -67,6 +67,7 @@ export interface EarlyExecutableNativeFeatures {
   dynAsync: boolean;
   dynTier?: boolean;
   streamDyn?: boolean;
+  eventsOnceSignal?: boolean;
   events: boolean;
   emitter: boolean;
   symbol: boolean;

packages/compiler/src/frontend/lowering/builtins/promises.ts

@@ -187,7 +187,11 @@ function lowerTuplePromiseAll(
  * type is the checker's combined promise, and each entry's inner type
  * must be that inner type, one of its union arms, or a sub-union of it
  * (the backend's interned adapters wrap/re-tag fulfillments; a wider
- * entry would need machinery that doesn't exist and fences).
+ * entry would need machinery that doesn't exist and fences). A
+ * Promise<never> entry fits any result, and an entry whose payload widens
+ * into the result's converts. When the checker's combined type has no
+ * representation, the result takes the type the entries that can fulfill
+ * share, or 'unknown'.
  * Promise.all/allSettled/any fence with the sequential-await hint;
  * resolve/reject and the rest fall to the member fence. Null for
  * non-Promise receivers. */
@@ -214,14 +218,47 @@ export function lowerPromiseStaticCall(
         "a non-empty array LITERAL of promises is the supported form: Promise.race([p, q])",
       );
     }
-    const resultT = lowerer.mapTypeOf(lowerer.typeOf(call));
-    if (resultT?.kind !== "promise") {
-      lowerer.noLowering(
-        "Promise.race with this combined result type",
-        call,
-        "the entries' value types must combine into a representable union",
-      );
+    // A Promise<never> entry (the deadline idiom: `new Promise<never>((_,
+    // reject) => setTimeout(reject, ms))`) can only reject, so it races
+    // under any result type; its inner type maps to void.
+    const neverEntry = (el: ts.Expression): boolean => {
+      const t = lowerer.typeOf(el);
+      const arg =
+        t.getSymbol()?.name === "Promise"
+          ? lowerer.checker.getTypeArguments(t as ts.TypeReference)[0]
+          : undefined;
+      return arg !== undefined && (arg.flags & ts.TypeFlags.Never) !== 0;
+    };
+    const elements = argNode.elements.nodes;
+    let combined = lowerer.mapTypeOf(lowerer.typeOf(call));
+    let lowered: IrExpr[] | null = null;
+    if (combined?.kind !== "promise") {
+      // The checker's combined type has no representation (an 'any'
+      // element array such as events.once's any[], or 'void | any[]'): the
+      // result takes the one type the entries that can fulfill share, or
+      // else 'unknown' when each of them widens into it.
+      lowered = elements.map((el) => lowerer.lowerExpr(el));
+      const fulfilling = lowered.filter((_, i) => !neverEntry(elements[i]!));
+      const first = fulfilling[0]?.type;
+      const dynResult: IrType & { kind: "promise" } = { kind: "promise", inner: DYN };
+      if (first?.kind === "promise" && fulfilling.every((e) => typeEquals(e.type, first))) {
+        combined = first;
+      } else if (
+        fulfilling.length > 0 &&
+        fulfilling.every(
+          (e) => e.type.kind === "promise" && lowerer.promisePayloadWidens(e.type, dynResult),
+        )
+      ) {
+        combined = dynResult;
+      } else {
+        lowerer.noLowering(
+          "Promise.race with this combined result type",
+          call,
+          "the entries' value types must combine into a representable union",
+        );
+      }
     }
+    const resultT = combined;
     const resultArms =
       resultT.inner.kind === "union"
         ? (lowerer.unions.get(resultT.inner.unionId)?.arms ?? [])
@@ -238,19 +275,8 @@ export function lowerPromiseStaticCall(
       }
       return resultArms.some((a) => typeEquals(a, inner));
     };
-    // A Promise<never> entry (the deadline idiom: `new Promise<never>((_,
-    // reject) => setTimeout(reject, ms))`) can only reject, so it races
-    // under any result type; its inner type maps to void.
-    const neverEntry = (el: ts.Expression): boolean => {
-      const t = lowerer.typeOf(el);
-      const arg =
-        t.getSymbol()?.name === "Promise"
-          ? lowerer.checker.getTypeArguments(t as ts.TypeReference)[0]
-          : undefined;
-      return arg !== undefined && (arg.flags & ts.TypeFlags.Never) !== 0;
-    };
-    const entries = argNode.elements.nodes.map((el) => {
-      const entry = lowerer.lowerExpr(el);
+    const entries = elements.map((el, i) => {
+      const entry = lowered?.[i] ?? lowerer.lowerExpr(el);
       if (entry.type.kind !== "promise") {
         lowerer.noLowering(
           "Promise.race over non-promise entries",
@@ -259,6 +285,10 @@ export function lowerPromiseStaticCall(
         );
       }
       if (!compatible(entry.type.inner) && !(entry.type.inner.kind === "void" && neverEntry(el))) {
+        // A payload that widens into the result's (into 'unknown', or a
+        // checked conversion) keeps its promise and settlement turn.
+        if (lowerer.promisePayloadWidens(entry.type, resultT))
+          return lowerer.coerceToExpected(entry, resultT);
         lowerer.unsupported(
           "SC1090",
           el,

packages/compiler/src/frontend/lowering/lower-classes.ts

@@ -116,6 +116,7 @@ import {
 import {
   STREAM_API_MEMBERS,
   STREAM_PROP_MEMBERS,
+  STREAM_COMPLETION_CALLBACK_INDEX,
   UNDERSCORE_METHODS,
   lowerStreamNew,
   lowerStreamSuperCall,
@@ -2698,6 +2699,24 @@ export function collectClassShapeInner(
               );
             }
           }
+          // A completion callback declared with the broad `Function` type
+          // (`cb: Function`) takes the checked-dynamic callable the runtime
+          // passes implicitly typed callbacks; calls through it are
+          // dynamic calls of that callable.
+          const callbackIndex = STREAM_COMPLETION_CALLBACK_INDEX.get(name);
+          const callbackParam =
+            ts.isMethodDeclaration(member) && callbackIndex !== undefined
+              ? member.parameters.nodes[callbackIndex]
+              : undefined;
+          if (
+            callbackParam !== undefined &&
+            ts.isIdentifier(callbackParam.name) &&
+            !callbackParam.dotDotDotToken &&
+            !callbackParam.questionToken &&
+            !callbackParam.initializer &&
+            lowerer.checker.typeToString(lowerer.typeOf(callbackParam.name)) === "Function"
+          )
+            lowerer.checkedCallbackParams.add(callbackParam);
         }
       }
       if (ts.isPropertyDeclaration(member)) {