From 497c227569a0762156ef1389eb6c0c50b229f6ce Mon Sep 17 00:00:00 2001 From: Nazar Hussain Date: Tue, 18 Aug 2026 22:53:27 +0500 Subject: [PATCH] feat(js): add js.spawn async task DSL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker-thread async previously meant hand-rolling napi.AsyncWork plus Deferred at every call site, and DSL values were unusable in the completion callback: js.env() panics there because only the sync wrappers establish the thread-local env context. js.spawn takes a comptime duck-typed task (compute/resolve/deinit, with optional errorMessage/reject) and returns a Promise that settles on the JS thread. Its completion callback sets the DSL env context, so resolve can return DSL types, and composes with OwnedTypedArray to hand results to JS without copying. compute deliberately does NOT get the env context — napi calls are illegal on the worker thread, so a panic there is the guard rail. Lifted from lodestar-z bindings/napi/async_task.zig, which was written to be upstreamed and is deleted once this ships. --- README.md | 42 +++++++- examples/js_dsl/mod.test.ts | 42 ++++++++ examples/js_dsl/mod.zig | 103 +++++++++++++++++++ src/js.zig | 1 + src/js/async_task.zig | 197 ++++++++++++++++++++++++++++++++++++ src/js/context.zig | 7 +- src/js/promise.zig | 8 +- 7 files changed, 394 insertions(+), 6 deletions(-) create mode 100644 src/js/async_task.zig diff --git a/README.md b/README.md index dfe8961..a9f41a1 100644 --- a/README.md +++ b/README.md @@ -405,7 +405,47 @@ pub fn asyncOp(val: Number) !Promise(Number) { } ``` -`Promise(T)` in this DSL path is synchronous-only: resolve or reject it before the exported function returns. For truly asynchronous completion, keep the `Deferred` handle in lower-level N-API code and bridge back with `napi.AsyncWork` or `napi.ThreadSafeFunction`. +`Promise(T)` in this DSL path is synchronous-only: resolve or reject it before the exported function returns. For work that must run off the JS thread, use `js.spawn` below. + +### Async Tasks + +`js.spawn` runs a task on the libuv worker pool and returns a JS Promise that settles when it finishes. A task is any struct with `compute`, `resolve`, and `deinit`: + +```zig +const ScaleTask = struct { + data: []u32, + factor: u32, + + // Worker thread — must not touch napi or DSL values. + pub fn compute(self: *ScaleTask) !void { + for (self.data) |*value| value.* *= self.factor; + } + + // JS thread — the DSL env context is established, so DSL types work here. + pub fn resolve(self: *ScaleTask, _: napi.Env) !js.OwnedUint32Array { + const owned: js.OwnedUint32Array = .fromOwnedSlice(js.allocator(), self.data); + self.data = &.{}; // ownership handed to JS, no copy + return owned; + } + + // Safe to call after resolve transferred ownership. + pub fn deinit(self: *ScaleTask) void { + js.allocator().free(self.data); + } +}; + +pub fn asyncScale(data: js.Uint32Array, factor: Number) !Value { + const copy = try js.allocator().dupe(u32, try data.toSlice()); + errdefer js.allocator().free(copy); + return js.spawn(ScaleTask, .{ .data = copy, .factor = @intCast(factor.assertI32()) }, "asyncScale"); +} +``` + +`resolve` may return a DSL type (`js.Number`), an owned typed array (transferred without copying), `napi.Value`, or `void`. + +If `compute` returns an error the promise rejects with an `Error`. Add an optional `errorMessage(err: anyerror) [:0]const u8` to control the message, or an optional `reject(self: *Task, env: napi.Env, err: anyerror) !napi.Value` to build the rejection value yourself; otherwise the message is `@errorName(err)`. + +Ownership: if `spawn` fails the task is not consumed, so the caller's `errdefer`s must free it (as above). Once `spawn` succeeds the helper owns the task and calls `deinit` after the promise settles. ### Callbacks diff --git a/examples/js_dsl/mod.test.ts b/examples/js_dsl/mod.test.ts index 47471ff..db9cd28 100644 --- a/examples/js_dsl/mod.test.ts +++ b/examples/js_dsl/mod.test.ts @@ -840,3 +840,45 @@ describe("enum export", () => { expect(Object.isFrozen(mod.BlsPublicKey.Encoding)).toBe(true); }); }); + +// Section 18: Async Tasks +describe("async tasks", () => { + it("resolves with a DSL value built in the complete callback", async () => { + await expect(mod.asyncDouble(21)).resolves.toEqual(42); + }); + + it("returns a real Promise", () => { + const promise = mod.asyncDouble(1); + expect(promise).toBeInstanceOf(Promise); + return promise; + }); + + it("runs concurrent tasks independently", async () => { + const results = await Promise.all([1, 2, 3, 4, 5].map((n) => mod.asyncDouble(n))); + expect(results).toEqual([2, 4, 6, 8, 10]); + }); + + it("transfers an owned typed array without copying", async () => { + const result = await mod.asyncScale(new Uint32Array([1, 2, 3]), 3); + expect(result).toBeInstanceOf(Uint32Array); + expect(Array.from(result)).toEqual([3, 6, 9]); + }); + + it("handles empty typed array transfer", async () => { + const result = await mod.asyncScale(new Uint32Array(0), 2); + expect(result).toBeInstanceOf(Uint32Array); + expect(result.length).toEqual(0); + }); + + it("rejects with the task-supplied error message", async () => { + await expect(mod.asyncFail()).rejects.toThrow("worker could not finish the job"); + }); + + it("rejects with the error name when errorMessage is absent", async () => { + await expect(mod.asyncFailBare()).rejects.toThrow("Unlucky"); + }); + + it("rejects with an Error instance", async () => { + await expect(mod.asyncFail()).rejects.toBeInstanceOf(Error); + }); +}); diff --git a/examples/js_dsl/mod.zig b/examples/js_dsl/mod.zig index d2de845..56a1a5e 100644 --- a/examples/js_dsl/mod.zig +++ b/examples/js_dsl/mod.zig @@ -2,6 +2,7 @@ const std = @import("std"); const js = @import("zapi").js; +const napi = @import("zapi").napi; const Number = js.Number; const String = js.String; const Boolean = js.Boolean; @@ -722,6 +723,108 @@ pub var mutable_counter: u32 = 5; pub const IDENTITY_MATRIX = [_]u32{ 1, 0, 0, 1 }; pub const VERSION_INFO = .{ .major = 3, .minor = 1 }; +// ============================================================================ +// Section 18: Async Tasks +// ============================================================================ + +/// Doubles a number on the libuv worker pool. +/// +/// `resolve` returns a DSL `js.Number` rather than a raw `napi.Value`, which +/// works because the async complete callback establishes the DSL env context. +const DoubleTask = struct { + value: i32, + + pub fn compute(self: *DoubleTask) !void { + self.value *= 2; + } + + pub fn resolve(self: *DoubleTask, _: napi.Env) !Number { + return Number.from(self.value); + } + + pub fn deinit(_: *DoubleTask) void {} +}; + +/// JS: asyncDouble(n): Promise +pub fn asyncDouble(n: Number) !Value { + return js.spawn(DoubleTask, .{ .value = n.assertI32() }, "asyncDouble"); +} + +/// Scales a Uint32Array on the worker pool and hands the result back without +/// copying, composing `js.AsyncTask` with `js.OwnedUint32Array`. +const ScaleTask = struct { + data: []u32, + factor: u32, + + pub fn compute(self: *ScaleTask) !void { + for (self.data) |*value| value.* *= self.factor; + } + + pub fn resolve(self: *ScaleTask, _: napi.Env) !js.OwnedUint32Array { + const owned: js.OwnedUint32Array = .fromOwnedSlice(js.allocator(), self.data); + self.data = &.{}; + return owned; + } + + pub fn deinit(self: *ScaleTask) void { + // Empty (a no-op free) once resolve transferred ownership to JS. + js.allocator().free(self.data); + } +}; + +/// JS: asyncScale(data, factor): Promise +pub fn asyncScale(data: js.Uint32Array, factor: Number) !Value { + const copy = try js.allocator().dupe(u32, try data.toSlice()); + errdefer js.allocator().free(copy); + return js.spawn(ScaleTask, .{ + .data = copy, + .factor = @intCast(factor.assertI32()), + }, "asyncScale"); +} + +/// Rejects with a task-supplied message via the optional `errorMessage` decl. +const FailTask = struct { + pub fn compute(_: *FailTask) !void { + return error.ComputeFailed; + } + + pub fn resolve(_: *FailTask, _: napi.Env) !Number { + return Number.from(0); + } + + pub fn errorMessage(err: anyerror) [:0]const u8 { + return switch (err) { + error.ComputeFailed => "worker could not finish the job", + else => @errorName(err), + }; + } + + pub fn deinit(_: *FailTask) void {} +}; + +/// JS: asyncFail(): Promise +pub fn asyncFail() !Value { + return js.spawn(FailTask, .{}, "asyncFail"); +} + +/// Without `errorMessage`, the rejection message defaults to `@errorName`. +const BareFailTask = struct { + pub fn compute(_: *BareFailTask) !void { + return error.Unlucky; + } + + pub fn resolve(_: *BareFailTask, _: napi.Env) !Number { + return Number.from(0); + } + + pub fn deinit(_: *BareFailTask) void {} +}; + +/// JS: asyncFailBare(): Promise +pub fn asyncFailBare() !Value { + return js.spawn(BareFailTask, .{}, "asyncFailBare"); +} + comptime { js.exportModule(@This(), .{ .identity = @import("zapi_addon_identity"), diff --git a/src/js.zig b/src/js.zig index 1f6be93..28f6771 100644 --- a/src/js.zig +++ b/src/js.zig @@ -50,6 +50,7 @@ pub const OwnedBigInt64Array = typed_arrays.OwnedBigInt64Array; pub const OwnedBigUint64Array = typed_arrays.OwnedBigUint64Array; pub const Promise = @import("js/promise.zig").Promise; +pub const spawn = @import("js/async_task.zig").spawn; pub const createPromise = @import("js/promise.zig").createPromise; pub const NoAddonIdentity = @import("js/class_runtime.zig").NoAddonIdentity; diff --git a/src/js/async_task.zig b/src/js/async_task.zig new file mode 100644 index 0000000..960abc4 --- /dev/null +++ b/src/js/async_task.zig @@ -0,0 +1,197 @@ +//! Worker-thread async DSL: run Zig work on the libuv pool and settle a JS +//! Promise with a DSL value. +//! +//! `js.Promise` alone cannot express this — its deferred handle is not +//! preserved across the JS boundary, so async resolution otherwise means +//! hand-rolling `napi.AsyncWork` + `napi.Deferred` at every call site. +//! +//! A Task is any struct providing: +//! +//! Required: +//! - `compute(self: *Task) !void` — libuv worker thread; MUST NOT call napi +//! APIs or construct DSL values +//! - `resolve(self: *Task, env: napi.Env) !T` — JS thread; builds the +//! fulfillment value. `T` may be a DSL type (`js.Number`), an owned typed +//! array (`js.OwnedUint32Array`, transferred without copying), `napi.Value`, +//! or `void` +//! - `deinit(self: *Task) void` — frees task-owned memory; must be safe after +//! `resolve` transferred ownership to JS +//! +//! Optional: +//! - `reject(self: *Task, env: napi.Env, err: anyerror) !napi.Value` — builds +//! the rejection value for a failed `compute`. Takes precedence over +//! `errorMessage`. +//! - `errorMessage(err: anyerror) [:0]const u8` — maps a failed `compute`'s +//! error to the rejection Error's message. Defaults to `@errorName(err)`. +//! +//! Ownership: on `spawn` error the task is NOT consumed, so the caller's +//! errdefers must free its resources. Once `spawn` returns successfully the +//! helper owns the task and calls `deinit` after the promise settles. + +const std = @import("std"); +const napi = @import("../napi.zig"); +const context = @import("context.zig"); +const typed_arrays = @import("typed_arrays.zig"); +const wrap_function = @import("wrap_function.zig"); +const Value = @import("value.zig").Value; + +/// Runs `task.compute` on the libuv worker pool and returns a JS Promise that +/// settles on the JS thread: rejected if `compute` failed, otherwise resolved +/// with `task.resolve(env)`. +/// +/// `resource_name` labels the async resource for diagnostics and async hooks. +pub fn spawn(comptime Task: type, task: Task, comptime resource_name: []const u8) !Value { + comptime validateTask(Task); + + const Context = struct { + task: Task, + err: ?anyerror, + deferred: napi.Deferred, + work: napi.c.napi_async_work, + + const Self = @This(); + + /// Worker thread. Deliberately does NOT establish the DSL env context: + /// napi calls are illegal here, and `js.env()` panicking is the + /// intended guard rail. + fn execute(_: napi.Env, ctx: *Self) void { + ctx.task.compute() catch |err| { + ctx.err = err; + }; + } + + /// JS thread, after the worker finished. Establishes the DSL env + /// context so `resolve` can build DSL values, then always settles the + /// promise — if settling itself fails we fall back to a bare reject so + /// callers never see a dangling Promise. + fn complete(env: napi.Env, status: napi.status.Status, ctx: *Self) void { + const prev = context.setEnv(env); + defer context.restoreEnv(prev); + + defer { + napi.status.check(napi.c.napi_delete_async_work(env.env, ctx.work)) catch {}; + ctx.task.deinit(); + context.allocator().destroy(ctx); + } + + settle(env, status, ctx) catch { + rejectWithMessage(env, ctx.deferred, "InternalError") catch {}; + }; + } + + fn settle(env: napi.Env, status: napi.status.Status, ctx: *Self) !void { + if (status != .ok) { + // libuv's async work itself failed (e.g. cancelled), not compute. + return rejectWithMessage(env, ctx.deferred, @tagName(status)); + } + if (ctx.err) |err| { + if (comptime @hasDecl(Task, "reject")) { + return ctx.deferred.reject(try ctx.task.reject(env, err)); + } + const message = if (comptime @hasDecl(Task, "errorMessage")) + Task.errorMessage(err) + else + @errorName(err); + return rejectWithMessage(env, ctx.deferred, message); + } + try ctx.deferred.resolve(try resolveValue(&ctx.task, env)); + } + }; + + const env = context.env(); + const allocator = context.allocator(); + + const ctx = try allocator.create(Context); + errdefer allocator.destroy(ctx); + + ctx.* = .{ + .task = task, + .err = null, + .deferred = undefined, + .work = undefined, + }; + + const resource = try env.createStringUtf8(resource_name); + const cleanup_value = try env.getUndefined(); + + // Until queue succeeds, this function owns the unqueued work handle. + const work = try env.createAsyncWork( + Context, + null, + resource, + Context.execute, + Context.complete, + ctx, + ); + errdefer work.delete() catch |err| { + std.log.err("zapi: failed to delete unqueued async work ({s}): {s}", .{ resource_name, @errorName(err) }); + }; + ctx.work = work.work; + + ctx.deferred = try env.createPromise(); + // Settle the unreturned Promise so Node can release its deferred handle. + errdefer ctx.deferred.resolve(cleanup_value) catch |err| { + std.log.err("zapi: failed to settle unreturned async promise ({s}): {s}", .{ resource_name, @errorName(err) }); + }; + + try work.queue(); + + return .{ .val = ctx.deferred.getPromise() }; +} + +/// Calls `Task.resolve` and converts its result to a `napi.Value`, accepting +/// either an error union or a plain return type. +fn resolveValue(task: anytype, env: napi.Env) !napi.Value { + const result = @TypeOf(task.*).resolve(task, env); + const value = if (comptime @typeInfo(@TypeOf(result)) == .error_union) try result else result; + return toNapiValue(@TypeOf(value), value, env); +} + +fn toNapiValue(comptime T: type, value: T, env: napi.Env) !napi.Value { + if (T == napi.Value) return value; + if (T == void) return env.getUndefined(); + if (comptime typed_arrays.isOwnedTypedArray(T)) { + var owned = value; + defer owned.deinit(); + return owned.intoValue(env); + } + if (comptime wrap_function.isDslType(T)) return value.val; + @compileError("zapi: `resolve` cannot return " ++ @typeName(T) ++ + " — return a DSL type (e.g. `js.Number`), an owned typed array, `napi.Value`, or `void`"); +} + +/// Reject `deferred` with `new Error(message)` so JS callers can match on +/// `err.message`. +fn rejectWithMessage(env: napi.Env, deferred: napi.Deferred, message: []const u8) !void { + const msg_val = try env.createStringUtf8(message); + const err_val = try env.createError(napi.Value{ .env = env.env, .value = null }, msg_val); + try deferred.reject(err_val); +} + +fn validateTask(comptime Task: type) void { + if (@typeInfo(Task) != .@"struct") { + @compileError("zapi: async task `" ++ @typeName(Task) ++ "` must be a struct"); + } + for ([_][]const u8{ "compute", "resolve", "deinit" }) |decl| { + if (!@hasDecl(Task, decl)) { + @compileError("zapi: async task `" ++ @typeName(Task) ++ + "` is missing the required `" ++ decl ++ "` declaration"); + } + } +} + +test "validateTask accepts a well-formed task" { + const Task = struct { + pub fn compute(_: *@This()) !void {} + pub fn resolve(_: *@This(), _: napi.Env) !void {} + pub fn deinit(_: *@This()) void {} + }; + comptime validateTask(Task); + try std.testing.expect(@hasDecl(Task, "compute")); +} + +test "spawn requires a JS callback context" { + // spawn resolves the env from the DSL context, so calling it off a JS + // callback is a programming error rather than a silent no-op. + try std.testing.expect(@TypeOf(context.env) == fn () napi.Env); +} diff --git a/src/js/context.zig b/src/js/context.zig index e2b5b10..00e9033 100644 --- a/src/js/context.zig +++ b/src/js/context.zig @@ -19,8 +19,11 @@ threadlocal var current_env: ?napi.Env = null; /// execution scope of a JavaScript function, method, getter, or setter that /// was exposed to JS via the ZAPI DSL. Calling it outside such a context /// (e.g., from a background thread or a Zig-initiated function call) will -/// result in a panic. For asynchronous N-API work, use `napi.AsyncWork` or -/// `napi.ThreadSafeFunction` which provide explicit `napi_env` parameters. +/// result in a panic. `js.spawn`'s completion callback establishes this +/// context, so a task's `resolve` may build DSL values; its `compute` runs on +/// a worker thread and deliberately does not. For raw asynchronous N-API work, +/// use `napi.AsyncWork` or `napi.ThreadSafeFunction`, which provide explicit +/// `napi_env` parameters. pub fn env() napi.Env { return current_env orelse @panic("js.env() called outside of a JS callback context"); } diff --git a/src/js/promise.zig b/src/js/promise.zig index c4bdcbc..5352060 100644 --- a/src/js/promise.zig +++ b/src/js/promise.zig @@ -12,9 +12,11 @@ const String = @import("string.zig").String; /// IMPORTANT: When returning `js.Promise(T)` from a DSL function, the promise /// must be resolved or rejected *before* the function returns. The `deferred` /// handle is not preserved across the JS boundary — only the `.val` (the JS -/// promise object) is returned to the caller. For async resolution (e.g., from -/// a worker thread), store the `Deferred` handle separately and use `napi.AsyncWork` -/// or `napi.ThreadSafeFunction` from the low-level N-API layer. +/// promise object) is returned to the caller. For async resolution from a +/// worker thread, use `js.spawn` (see `js/async_task.zig`), which owns the +/// `Deferred` handle and settles it after the work completes. Dropping to +/// `napi.AsyncWork` or `napi.ThreadSafeFunction` directly remains an option +/// when you need control `js.spawn` does not expose. pub fn Promise(comptime T: type) type { return struct { /// The underlying `napi.Value` representing the JavaScript Promise object.