From 17e5f9748ed5237ec93bb50acea75b8de7cb316b Mon Sep 17 00:00:00 2001 From: "Nurul Huda (Apon)" Date: Sat, 11 Jul 2026 00:54:41 +0600 Subject: [PATCH] feat(wasm): add blocking fetch using jspi --- pkg/ziex/src/cloudflare/do.ts | 2 + pkg/ziex/src/fetch.ts | 100 +++++++++++++++++++++++++ pkg/ziex/src/runtime.ts | 2 + pkg/ziex/src/wasi.ts | 1 + pkg/ziex/src/wasm/core.ts | 22 ++++-- pkg/ziex/src/wasm/index.ts | 5 ++ pkg/ziex/src/wasm/wasi.ts | 9 ++- src/runtime/core/App/Wasm.zig | 5 +- src/runtime/core/Fetch.zig | 15 ++-- src/runtime/core/Fetch/Wasm.zig | 129 ++++++++++++++++++++++++++++++++ src/runtime/core/Http/Wasm.zig | 4 +- src/runtime/core/Kv/Wasm.zig | 21 +++++- src/runtime/core/Request.zig | 2 +- src/runtime/core/Response.zig | 2 +- 14 files changed, 295 insertions(+), 24 deletions(-) create mode 100644 pkg/ziex/src/fetch.ts create mode 100644 src/runtime/core/Fetch/Wasm.zig diff --git a/pkg/ziex/src/cloudflare/do.ts b/pkg/ziex/src/cloudflare/do.ts index 1aea71d7..92d349f1 100644 --- a/pkg/ziex/src/cloudflare/do.ts +++ b/pkg/ziex/src/cloudflare/do.ts @@ -1,5 +1,6 @@ import { ZxBridge } from "../wasm"; import { createKVImports, createMemoryKV } from "../kv"; +import { createFetchImports } from "../fetch"; import { createD1Imports } from "../db"; import { createWasiImports } from "../wasi"; import { buildWsImports, attachWebSocket } from "../runtime"; @@ -128,6 +129,7 @@ export function createWebSocketDO( __zx_ws: wsImports, __zx_kv: createKVImports(kvBindings ?? { default: createMemoryKV() }, mem), __zx_db: createD1Imports(dbBindings ?? {}, mem), + __zx_net: createFetchImports(mem), ...(options?.imports ? options.imports(mem) : {}), ...bridgeImports, } as WebAssembly.Imports); diff --git a/pkg/ziex/src/fetch.ts b/pkg/ziex/src/fetch.ts new file mode 100644 index 00000000..79e980d8 --- /dev/null +++ b/pkg/ziex/src/fetch.ts @@ -0,0 +1,100 @@ +/** + * Create a `fetch` import object for blocking WASM fetch via JSPI. + * When JSPI is unavailable the import always returns -1 (network error). + */ +export function createFetchImports( + getMemory: () => WebAssembly.Memory, +): Record { + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + + function readStr(ptr: number, len: number): string { + return decoder.decode(new Uint8Array(getMemory().buffer, ptr, len)); + } + + function writeBytes(buf_ptr: number, buf_max: number, data: Uint8Array): number { + if (data.length > buf_max) return -2; + new Uint8Array(getMemory().buffer, buf_ptr, data.length).set(data); + return data.length; + } + + function readBytes(ptr: number, len: number): Uint8Array { + return new Uint8Array(getMemory().buffer, ptr, len); + } + + function parseHeaders(headersJson: string): Record { + let headers: Record = {}; + try { + headers = JSON.parse(headersJson); + } catch { + for (const line of headersJson.split('\n')) { + const colonIdx = line.indexOf(':'); + if (colonIdx > 0) { + headers[line.slice(0, colonIdx)] = line.slice(colonIdx + 1); + } + } + } + return headers; + } + + async function doFetch( + url_ptr: number, + url_len: number, + method_ptr: number, + method_len: number, + headers_ptr: number, + headers_len: number, + body_ptr: number, + body_len: number, + timeout_ms: number, + status_out: number, + buf_ptr: number, + buf_max: number, + ): Promise { + const url = readStr(url_ptr, url_len); + const method = method_len > 0 ? readStr(method_ptr, method_len) : 'GET'; + const headersJson = headers_len > 0 ? readStr(headers_ptr, headers_len) : '{}'; + const body = body_len > 0 ? readBytes(body_ptr, body_len) : undefined; + const headers = parseHeaders(headersJson); + + const controller = new AbortController(); + const timeout = timeout_ms > 0 ? setTimeout(() => controller.abort(), timeout_ms) : null; + + try { + const response = await fetch(url, { + method, + headers: Object.keys(headers).length > 0 ? headers : undefined, + body: method !== 'GET' && method !== 'HEAD' ? body : undefined, + signal: controller.signal, + }); + if (timeout) clearTimeout(timeout); + new DataView(getMemory().buffer).setUint16(status_out, response.status, true); + const bytes = new Uint8Array(await response.arrayBuffer()); + return writeBytes(buf_ptr, buf_max, bytes); + } catch (err) { + if (timeout) clearTimeout(timeout); + const msg = err instanceof Error ? err.message : String(err); + console.error(`wasm fetch failed: ${url} (${msg})`); + return -1; + } + } + + const Suspending = (WebAssembly as any).Suspending; + if (typeof Suspending !== 'function') { + return { + fetch: ( + _url_ptr: number, _url_len: number, + _method_ptr: number, _method_len: number, + _headers_ptr: number, _headers_len: number, + _body_ptr: number, _body_len: number, + _timeout_ms: number, + _status_out: number, + _buf_ptr: number, _buf_max: number, + ): number => -1, + }; + } + + return { + fetch: new Suspending(doFetch), + }; +} diff --git a/pkg/ziex/src/runtime.ts b/pkg/ziex/src/runtime.ts index 1e6e55c5..31d20d76 100644 --- a/pkg/ziex/src/runtime.ts +++ b/pkg/ziex/src/runtime.ts @@ -1,5 +1,6 @@ import { ZxWasiBridge } from "./wasm/wasi"; import { createKVImports, createMemoryKV } from "./kv"; +import { createFetchImports } from "./fetch"; import { createD1Imports } from "./db"; import { createWasiImports, ProcExit, mergeUint8Arrays } from "./wasi"; import type { WASI } from "./wasi"; @@ -280,6 +281,7 @@ export async function run({ __zx_ws: buildWsImports(jspi ? Suspending : null, mem, new TextDecoder(), wsState), __zx_kv: createKVImports(kvBindings ?? { default: createMemoryKV() }, mem), __zx_db: createD1Imports(dbBindings ?? {}, mem), + __zx_net: createFetchImports(mem), ...(imports ? imports(mem) : {}), ...ZxWasiBridge.createImportObject(bridgeRef), } as WebAssembly.Imports); diff --git a/pkg/ziex/src/wasi.ts b/pkg/ziex/src/wasi.ts index bf294220..cccf6906 100644 --- a/pkg/ziex/src/wasi.ts +++ b/pkg/ziex/src/wasi.ts @@ -33,6 +33,7 @@ export function createWasiImports({ "--pathname", url.pathname, "--method", request.method, "--search", url.search, + "--url", url.href ]; request.headers.forEach((value, name) => { diff --git a/pkg/ziex/src/wasm/core.ts b/pkg/ziex/src/wasm/core.ts index 4ab6a0f1..5fe11b22 100644 --- a/pkg/ziex/src/wasm/core.ts +++ b/pkg/ziex/src/wasm/core.ts @@ -1,4 +1,5 @@ import { ZigJS } from "../../../../vendor/jsz/js/src"; +import { createFetchImports } from "../fetch"; /** * Core WASM bridge - environment-agnostic (browser + edge). @@ -156,7 +157,7 @@ export class ZxBridgeCore { const url = readString(urlPtr, urlLen); const method = methodLen > 0 ? readString(methodPtr, methodLen) : 'GET'; const headersJson = headersLen > 0 ? readString(headersPtr, headersLen) : '{}'; - const body = bodyLen > 0 ? readString(bodyPtr, bodyLen) : undefined; + const body = bodyLen > 0 ? getMemoryView().subarray(bodyPtr, bodyPtr + bodyLen) : undefined; let headers: Record = {}; try { @@ -183,8 +184,8 @@ export class ZxBridgeCore { fetch(url, fetchOptions) .then(async (response) => { if (timeout) clearTimeout(timeout); - const text = await response.text(); - this._notifyFetchComplete(fetchId, response.status, text, false); + const bytes = new Uint8Array(await response.arrayBuffer()); + this._notifyFetchComplete(fetchId, response.status, bytes, false); }) .catch((error) => { if (timeout) clearTimeout(timeout); @@ -194,10 +195,15 @@ export class ZxBridgeCore { }); } - /** Notify WASM that a fetch completed */ - protected _notifyFetchComplete(fetchId: bigint, statusCode: number, body: string, isError: boolean): void { + /** Notify WASM that a fetch completed. `body` is raw bytes on success, text on error. */ + protected _notifyFetchComplete( + fetchId: bigint, + statusCode: number, + body: string | Uint8Array, + isError: boolean, + ): void { const handler = this.#fetchCompleteHandler; - const encoded = textEncoder.encode(body); + const encoded = typeof body === 'string' ? textEncoder.encode(body) : body; const ptr = this._alloc(encoded.length); writeBytes(ptr, encoded); invokeWasmExport(handler, fetchId, statusCode, ptr, encoded.length, isError ? 1 : 0); @@ -265,6 +271,10 @@ export class ZxBridgeCore { static createImportObject(bridgeRef: { current: ZxBridgeCore | null }): WebAssembly.Imports { return { ...jsz.importObject(), + __zx_net: createFetchImports(() => { + if (!jsz.memory) throw new Error("WASM memory is not ready"); + return jsz.memory; + }) as WebAssembly.ModuleImports, __zx: { _log: (level: number, ptr: number, len: number) => ZxBridgeCore.log(level, ptr, len), _fetchAsync: ( diff --git a/pkg/ziex/src/wasm/index.ts b/pkg/ziex/src/wasm/index.ts index 8abde85f..1a9735e9 100644 --- a/pkg/ziex/src/wasm/index.ts +++ b/pkg/ziex/src/wasm/index.ts @@ -24,6 +24,7 @@ import { getMemoryView, } from "./core"; import { createKVImports, type KVNamespace } from "../kv"; +import { createFetchImports } from "../fetch"; import { createBrowserKVBindings } from "../browser/kv"; import type { WsOnOpenHandler, @@ -216,6 +217,10 @@ export class ZxBridge extends ZxBridgeCore { static override createImportObject(bridgeRef: { current: ZxBridge | null }): WebAssembly.Imports { return { ...jsz.importObject(), + __zx_net: createFetchImports(() => { + if (!jsz.memory) throw new Error("WASM memory is not ready"); + return jsz.memory; + }) as WebAssembly.ModuleImports, __zx: { _log: (level: number, ptr: number, len: number) => ZxBridgeCore.log(level, ptr, len), _setEventHandlerMode: (vnodeId: bigint, eventTypeId: number, maySuspend: number) => { diff --git a/pkg/ziex/src/wasm/wasi.ts b/pkg/ziex/src/wasm/wasi.ts index c6cddc07..1847b05a 100644 --- a/pkg/ziex/src/wasm/wasi.ts +++ b/pkg/ziex/src/wasm/wasi.ts @@ -68,7 +68,7 @@ export class ZxWasiBridge { const url = this.#readString(urlPtr, urlLen); const method = methodLen > 0 ? this.#readString(methodPtr, methodLen) : 'GET'; const headersJson = headersLen > 0 ? this.#readString(headersPtr, headersLen) : '{}'; - const body = bodyLen > 0 ? this.#readString(bodyPtr, bodyLen) : undefined; + const body = bodyLen > 0 ? this.#view().subarray(bodyPtr, bodyPtr + bodyLen) : undefined; let headers: Record = {}; try { @@ -91,7 +91,8 @@ export class ZxWasiBridge { }) .then(async (res) => { if (timeout) clearTimeout(timeout); - this.#notifyFetchComplete(fetchId, res.status, await res.text(), false); + const bytes = new Uint8Array(await res.arrayBuffer()); + this.#notifyFetchComplete(fetchId, res.status, bytes, false); }) .catch((err: Error) => { if (timeout) clearTimeout(timeout); @@ -100,8 +101,8 @@ export class ZxWasiBridge { }); } - #notifyFetchComplete(fetchId: bigint, status: number, body: string, isError: boolean): void { - const encoded = encoder.encode(body); + #notifyFetchComplete(fetchId: bigint, status: number, body: string | Uint8Array, isError: boolean): void { + const encoded = typeof body === 'string' ? encoder.encode(body) : body; const ptr = this.#alloc(encoded.length); this.#writeBytes(ptr, encoded); this.#fetchCompleteHandler(fetchId, status, ptr, encoded.length, isError ? 1 : 0); diff --git a/src/runtime/core/App/Wasm.zig b/src/runtime/core/App/Wasm.zig index 789c3f63..38396186 100644 --- a/src/runtime/core/App/Wasm.zig +++ b/src/runtime/core/App/Wasm.zig @@ -38,6 +38,7 @@ pub fn run(process_init: std.process.Init) !void { var pathname: []const u8 = "/"; var search: []const u8 = ""; + var url: []const u8 = ""; var method: zx.server.Request.Method = .GET; var header_entries = std.ArrayList(HeaderEntry).empty; defer header_entries.deinit(allocator); @@ -59,6 +60,8 @@ pub fn run(process_init: std.process.Init) !void { .value = std.mem.trimStart(u8, header_str[sep + 1 ..], " "), }); } + } else if (std.mem.eql(u8, arg, "--url")) { + url = args.next() orelse return error.MissingUrl; } } @@ -93,7 +96,7 @@ pub fn run(process_init: std.process.Init) !void { backend.cookie_header = cookie_header; backend.route_match = Router.matchRoute(pathname, .{ .match = .exact }); - const request = backend.request(method, pathname); + const request = backend.request(method, pathname, url); const response = backend.response(); const http = backend.http(); diff --git a/src/runtime/core/Fetch.zig b/src/runtime/core/Fetch.zig index 97d9dbe7..054ea36e 100644 --- a/src/runtime/core/Fetch.zig +++ b/src/runtime/core/Fetch.zig @@ -20,6 +20,7 @@ const common = @import("common.zig"); const server_impl = if (!is_wasm) @import("../server/fetch.zig") else struct {}; const client_impl = if (is_wasm) @import("../client/fetch.zig") else struct {}; +const wasm_impl = if (is_wasm) @import("Fetch/Wasm.zig") else struct {}; pub const Method = common.Method; pub const ContentType = common.ContentType; @@ -28,7 +29,7 @@ pub const is_wasm = builtin.cpu.arch == .wasm32 or builtin.cpu.arch == .wasm64; /// Io determines how fetch operations are executed. /// -/// - `blocking`: Blocks until complete (server-side) +/// - `blocking`: Blocks until complete (server-side native, WASM via JSPI) /// - `callback`: Calls a callback when complete (client-side WASM) pub const Io = struct { mode: Mode, @@ -57,7 +58,7 @@ pub const Io = struct { /// Check if this Io mode is supported on the current platform. pub fn isSupported(self: Io) bool { return switch (self.mode) { - .blocking => !is_wasm, // Blocking only works on server + .blocking => true, // Native on server, JSPI-suspended on WASM .callback => true, // Callback works everywhere }; } @@ -121,6 +122,11 @@ pub const Response = struct { return self._body; } + /// Returns the raw response body bytes. + pub fn bytes(self: *Response) ![]const u8 { + return self.text(); + } + /// Get an `std.Io.Reader` for streaming the response body. pub fn reader(self: *Response) std.Io.Reader { return std.Io.Reader.fixed(self._body); @@ -199,7 +205,7 @@ pub const ResponseCallbackCtx = *const fn (ctx: *anyopaque, response: ?*Response /// Perform an HTTP fetch request. /// /// The `io` parameter determines the execution model: -/// - `Io.blocking` - Blocks until complete, returns Response (server-only) +/// - `Io.blocking` - Blocks until complete, returns Response (server native, WASM via JSPI) /// - `Io.wasm(&callback)` - Calls callback when complete (WASM) /// /// **Server-side (blocking):** @@ -237,9 +243,8 @@ pub fn fetch( switch (io.mode) { .blocking => { - // Server-side: blocking fetch if (is_wasm) { - return error.UnsupportedIoMode; + return try wasm_impl.fetch(allocator, url, init); } const response = try server_impl.fetch(allocator, url, init); return response; diff --git a/src/runtime/core/Fetch/Wasm.zig b/src/runtime/core/Fetch/Wasm.zig new file mode 100644 index 00000000..7a583d69 --- /dev/null +++ b/src/runtime/core/Fetch/Wasm.zig @@ -0,0 +1,129 @@ +const std = @import("std"); +const Fetch = @import("../Fetch.zig"); + +const Response = Fetch.Response; +const Headers = Fetch.Headers; +const RequestInit = Fetch.RequestInit; +const FetchError = Fetch.FetchError; + +const max_response_bytes = 16 * 1024 * 1024; + +/// Perform a blocking HTTP fetch via the JS `fetch` import (JSPI). +pub fn fetch(allocator: std.mem.Allocator, url: []const u8, init: RequestInit) FetchError!Response { + const method_str = @tagName(init.method); + var headers_buf: [8192]u8 = undefined; + const headers_json = serializeHeadersJson(init.headers, &headers_buf); + const body = init.body orelse ""; + + var status: u16 = 0; + var capacity: usize = 65536; + var buf = try allocator.alloc(u8, capacity); + defer allocator.free(buf); + + while (true) { + const n = ext.fetch( + url.ptr, + url.len, + method_str.ptr, + method_str.len, + headers_json.ptr, + headers_json.len, + body.ptr, + body.len, + init.timeout_ms, + &status, + buf.ptr, + capacity, + ); + + if (n == -2) { + capacity *= 2; + if (capacity > max_response_bytes) return error.InvalidResponse; + buf = try allocator.realloc(buf, capacity); + continue; + } + + if (n < 0) return error.NetworkError; + + const body_data = try allocator.dupe(u8, buf[0..@intCast(n)]); + + return Response{ + .status = status, + .status_text = statusText(status), + .headers = Headers.init(allocator), + ._body = body_data, + ._body_used = false, + ._allocator = allocator, + ._owns_memory = true, + }; + } +} + +const ext = struct { + pub extern "__zx_net" fn fetch( + url_ptr: [*]const u8, + url_len: usize, + method_ptr: [*]const u8, + method_len: usize, + headers_ptr: [*]const u8, + headers_len: usize, + body_ptr: [*]const u8, + body_len: usize, + timeout_ms: u32, + status_out: *u16, + buf_ptr: [*]u8, + buf_max: usize, + ) i32; +}; + +fn serializeHeadersJson(headers: ?[]const RequestInit.Header, buf: []u8) []const u8 { + var len: usize = 0; + buf[len] = '{'; + len += 1; + + if (headers) |hdrs| { + for (hdrs, 0..) |h, i| { + if (i > 0) { + buf[len] = ','; + len += 1; + } + buf[len] = '"'; + len += 1; + const name_end = @min(len + h.name.len, buf.len - 10); + @memcpy(buf[len..name_end], h.name[0..@min(h.name.len, name_end - len)]); + len = name_end; + buf[len] = '"'; + len += 1; + buf[len] = ':'; + len += 1; + buf[len] = '"'; + len += 1; + const val_end = @min(len + h.value.len, buf.len - 2); + @memcpy(buf[len..val_end], h.value[0..@min(h.value.len, val_end - len)]); + len = val_end; + buf[len] = '"'; + len += 1; + } + } + + buf[len] = '}'; + len += 1; + return buf[0..len]; +} + +fn statusText(code: u16) []const u8 { + return switch (code) { + 200 => "OK", + 201 => "Created", + 204 => "No Content", + 301 => "Moved Permanently", + 302 => "Found", + 304 => "Not Modified", + 400 => "Bad Request", + 401 => "Unauthorized", + 403 => "Forbidden", + 404 => "Not Found", + 500 => "Internal Server Error", + else => "Unknown", + }; +} diff --git a/src/runtime/core/Http/Wasm.zig b/src/runtime/core/Http/Wasm.zig index 4dba8a15..821213d9 100644 --- a/src/runtime/core/Http/Wasm.zig +++ b/src/runtime/core/Http/Wasm.zig @@ -50,9 +50,9 @@ pub const Backend = struct { } /// Build an abstract `Request` backed by this instance. - pub fn request(self: *Backend, method: Request.Method, pathname: []const u8) Request { + pub fn request(self: *Backend, method: Request.Method, pathname: []const u8, url: []const u8) Request { return (Request.Builder{ - .url = "", + .url = url, .method = method, .method_str = @tagName(method), .pathname = pathname, diff --git a/src/runtime/core/Kv/Wasm.zig b/src/runtime/core/Kv/Wasm.zig index 00303a99..927b2f47 100644 --- a/src/runtime/core/Kv/Wasm.zig +++ b/src/runtime/core/Kv/Wasm.zig @@ -3,11 +3,24 @@ const Wasm = @This(); const std = @import("std"); const Kv = @import("../Kv.zig"); +const max_response_bytes = 16 * 1024 * 1024; + fn get(_: ?*anyopaque, ns: []const u8, allocator: std.mem.Allocator, key: []const u8) !?[]u8 { - var buf: [8192]u8 = undefined; - const n = ext.kv_get(ns.ptr, ns.len, key.ptr, key.len, &buf, buf.len); - if (n < 0) return null; - return try allocator.dupe(u8, buf[0..@intCast(n)]); + var capacity: usize = 65536; + var buf = try allocator.alloc(u8, capacity); + defer allocator.free(buf); + + while (true) { + const n = ext.kv_get(ns.ptr, ns.len, key.ptr, key.len, buf.ptr, capacity); + if (n == -2) { + capacity *= 2; + if (capacity > max_response_bytes) return error.InvalidResponse; + buf = try allocator.realloc(buf, capacity); + continue; + } + if (n < 0) return null; + return try allocator.dupe(u8, buf[0..@intCast(n)]); + } } fn put(_: ?*anyopaque, ns: []const u8, key: []const u8, value: []const u8, _: Kv.PutOptions) !void { diff --git a/src/runtime/core/Request.zig b/src/runtime/core/Request.zig index ee9e9f6d..bb1ba9b5 100644 --- a/src/runtime/core/Request.zig +++ b/src/runtime/core/Request.zig @@ -1,4 +1,4 @@ -pub const Request = @This(); +const Request = @This(); const std = @import("std"); const common = @import("common.zig"); diff --git a/src/runtime/core/Response.zig b/src/runtime/core/Response.zig index 339c7478..7c1c2e7f 100644 --- a/src/runtime/core/Response.zig +++ b/src/runtime/core/Response.zig @@ -1,4 +1,4 @@ -pub const Response = @This(); +const Response = @This(); const std = @import("std"); const common = @import("common.zig");