uce/src/wasm/backend.cpp

// W4 — FastCGI backend glue for the W3 wasm workspace runtime.
//
// Included into the native server TU (src/linux_fastcgi.cpp) after uce_lib.cpp,
// so it shares String/DValue/config and the UCEB1 codec. Provides a
// config-selectable page-render backend: when WASM_BACKEND_ENABLED and a wasm
// artifact exists for the entry unit, the request is served through a
// per-request wasm workspace instead of the native dlopen path.
//
// The seam is narrow on purpose — only the page render branch in
// handle_complete() changes. CLI, serve_http, and websocket stay native.

#include "../lib/wasm_trace.h"
// The native server TU has the real connectors (sqlite/mysql) compiled in, so
// the worker's host-side connector hostcalls are available here. The W3 CLI
// driver does not define this and gets a named-trap stub for those imports.
#define UCE_WASM_HOST_CONNECTORS 1
#include "worker.cpp"

#include <atomic>
#include <sys/stat.h>
#include <thread>
// Worker flushes ws_* dispatch batches to the broker via the FastCGI client.
#include "../lib/fcgi_forward.h"

// per forked worker process: one engine + compiled-core cache, one epoch ticker
static WasmWorker* g_wasm_worker = 0;
static std::thread* g_wasm_epoch_ticker = 0;
static std::atomic<bool> g_wasm_epoch_running(false);
static String g_wasm_init_error;
static bool g_wasm_init_attempted = false;

bool wasm_backend_configured(Request* context)
{
	if(!context || !context->server)
		return(false);
	return(config_bool("WASM_BACKEND_ENABLED", false));
}

// Lazily bring up the per-process worker on first use inside a forked child
// (the engine must not be inherited across fork). Returns "" on success.
static String wasm_backend_ensure_started(Request* context)
{
	if(g_wasm_init_attempted)
		return(g_wasm_init_error);
	g_wasm_init_attempted = true;

	StringMap& cfg = context->server->config;
	WasmWorkerConfig wc;
	wc.core_wasm_path = first(cfg["WASM_CORE_PATH"],
		path_join(cfg["COMPILER_SYS_PATH"], "bin/wasm/core.wasm"));
	wc.site_root = path_join(cfg["COMPILER_SYS_PATH"], cfg["SITE_DIRECTORY"]);
	wc.cache_root = cfg["BIN_DIRECTORY"];
	// write membrane allowlist: the site tree plus the runtime scratch dirs
	// pages legitimately write to (matches native reachable write targets).
	wc.write_roots = { wc.site_root, "/tmp" };
	for(const char* key : { "BIN_DIRECTORY", "SESSION_PATH", "TMP_UPLOAD_PATH" })
		if(cfg[key] != "")
			wc.write_roots.push_back(cfg[key]);
	wc.memory_limit = (int64_t)config_u64("WASM_MEMORY_LIMIT_BYTES", 512ull * 1024 * 1024);
	wc.epoch_deadline_ticks = config_u64("WASM_EPOCH_DEADLINE_TICKS", 200);
	wc.verbose = config_bool("WASM_BACKEND_VERBOSE", false);

	g_wasm_worker = new WasmWorker(wc);
	g_wasm_init_error = g_wasm_worker->init();
	if(g_wasm_init_error != "")
	{
		delete g_wasm_worker;
		g_wasm_worker = 0;
		return(g_wasm_init_error);
	}

	g_wasm_epoch_running.store(true);
	WasmWorker* worker = g_wasm_worker;
	u64 period_ms = config_u64("WASM_EPOCH_PERIOD_MS", 50);
	g_wasm_epoch_ticker = new std::thread([worker, period_ms] {
		while(g_wasm_epoch_running.load())
		{
			std::this_thread::sleep_for(std::chrono::milliseconds(period_ms));
			worker->engine.increment_epoch();
		}
	});
	return("");
}

static bool wasm_artifact_exists(Request* context, const String& entry_unit)
{
	if(entry_unit == "")
		return(false);
	String wasm_path = context->server->config["BIN_DIRECTORY"] + entry_unit + ".wasm";
	struct stat wasm_st;
	if(stat(wasm_path.c_str(), &wasm_st) != 0 || !S_ISREG(wasm_st.st_mode))
		return(false);
	// The wasm path bypasses the native JIT recompile, so a source edit would
	// otherwise be served from a stale .wasm. Require the artifact to be newer
	// than the source; if it is stale, fall back to native — which JIT-rebuilds
	// both .so and .wasm — so the next request gets a fresh artifact.
	struct stat src_st;
	if(stat(entry_unit.c_str(), &src_st) == 0 && wasm_st.st_mtime < src_st.st_mtime)
		return(false);
	return(true);
}

// Decide, by source inspection, whether a page must stay on the native
// backend for now (W5 surfaces not yet behind the membrane).
//
// Caveats this deliberately accepts:
//  - ENTRY-UNIT ONLY: a clean entry page that component()s into a native-only
//    unit is still served by wasm. With signals_based_traps off (see
//    make_engine) the worst case is a clean wasm error or a stubbed-empty
//    result, never a worker crash — so this is a best-effort routing hint,
//    not a guarantee. Full coverage is a W5 concern (un-stub the APIs).
//  - SUBSTRING match, intentionally conservative: a token like "xml_" also
//    matches an identifier or doc string containing it. That only ever
//    over-routes to native (always correct), never the reverse.
static bool wasm_backend_native_fallback_uncached(Request* context, const String& entry_unit)
{
	String source = file_get_contents(entry_unit);
	// Supported by the wasm core, intentionally NOT in this list:
	//  - regex_* (host PCRE2 hostcall), xml_*/yaml_*/markdown_* (compiled in),
	//  - unit_render()/component() (host resolver).
	// W7d: every former native-only surface now goes through the membrane —
	// zip_* (uce_host_zip), unit introspection (uce_host_units: unit_info/
	// unit_call/units_list), regex/xml/yaml/markdown, filesystem, sqlite,
	// background tasks, sleep, sockets/custom servers, memcache, mysql. No unit
	// source token forces native anymore, so there is nothing left to scan for.
	// (The native fallback now only covers cold/stale artifacts; W7e removes it.)
	(void)source;
	return(false);
}

static bool wasm_backend_native_fallback_needed(Request* context, const String& entry_unit)
{
	if(!context || !context->server || entry_unit == "")
		return(true);
	// Cache the verdict per process, keyed on path + mtime: the source scan is
	// otherwise a file read + 20 substring searches on every page request.
	struct CachedVerdict { time_t mtime; bool fallback; };
	static std::map<String, CachedVerdict> cache;
	struct stat st;
	time_t mtime = (stat(entry_unit.c_str(), &st) == 0) ? st.st_mtime : 0;
	auto it = cache.find(entry_unit);
	if(it != cache.end() && it->second.mtime == mtime)
		return(it->second.fallback);
	bool fallback = wasm_backend_native_fallback_uncached(context, entry_unit);
	cache[entry_unit] = { mtime, fallback };
	return(fallback);
}

// True if this request should be served by the wasm backend. Falls through to
// native when disabled, when init failed, or when the unit has no wasm artifact
// (e.g. units skip-listed for try/catch — automatic, graceful fallback). Mode-
// agnostic: the caller (page render / cli / serve_http) decides which entry to
// route here; this only gates on config, fallback tokens, artifact, and worker.
bool wasm_backend_should_handle(Request& request, const String& entry_unit)
{
	if(!wasm_backend_configured(&request))
		return(false);
	if(wasm_backend_native_fallback_needed(&request, entry_unit))
		return(false);
	if(!wasm_artifact_exists(&request, entry_unit))
		return(false);
	if(wasm_backend_ensure_started(&request) != "")
		return(false);
	return(true);
}

// Serve a request through a wasm workspace using the unit handler selected by
// `kind` (page render / cli / serve_http, with an optional named serve_http
// handler). Populates the native Request (status/headers/cookies/session/body)
// so the existing transport writes the response unchanged. Returns "" on
// success, or a collapsed error/trace string for the caller to route into the
// configured error page.
String wasm_backend_serve(Request& request, const String& entry_unit, const String& handler = "render")
{
	DValue ctx;
	auto copy_map = [&](const StringMap& source, const char* key) {
		for(auto& entry : source)
			ctx[key][entry.first] = entry.second;
	};
	copy_map(request.params, "params");
	copy_map(request.get, "get");
	copy_map(request.post, "post");
	copy_map(request.cookies, "cookies");
	copy_map(request.session, "session");
	ctx["entry_unit"] = entry_unit;
	// Raw request body: cli_input() parses a JSON CLI payload from context.in,
	// and serve_http handlers read it as req->in; carried into the workspace.
	ctx["in"] = request.in;
	// WebSocket event context: the workspace owns no connections, so the frame's
	// connection identity goes in and the handler's ws_send/ws_close dispatch
	// commands come back out (below) for the native broker to apply.
	if(handler == "websocket")
	{
		ctx["ws"]["connection_id"] = request.resources.websocket_connection_id;
		ctx["ws"]["scope"] = request.resources.websocket_scope;
		ctx["ws"]["opcode"] = (f64)request.resources.websocket_opcode;
		ctx["ws"]["binary"].set_bool(request.resources.websocket_is_binary);
		for(auto& id : request.resources.websocket_scope_connection_ids)
		{
			DValue v; v = id;
			ctx["ws"]["connections"].push(v);
		}
		ctx["ws"]["connection_state"] = request.connection;
	}

	WasmResponse response = wasm_worker_serve(*g_wasm_worker, ctx, entry_unit, handler);
	if(!response.ok)
		return(response.error == "" ? String("wasm workspace failed") : response.error);

	// Any handler may have called ws_send/ws_close (not just WS handlers). If it
	// left dispatch commands, flush the batch to the central WS broker, which owns
	// all connections and resolves each command's target (id/scope/broadcast)
	// against the full registry. This is the only path WS data takes out — the
	// workspace owns no connections.
	if(DValue* cmds = response.meta.key("ws_commands"))
	{
		DValue batch;
		batch["commands"] = *cmds;
		if(DValue* cstate = response.meta.key("ws_connection_state"))
		{
			batch["connection_id"] = request.resources.websocket_connection_id;
			batch["connection_state"] = *cstate;
		}
		StringMap dispatch_params;
		dispatch_params["UCE_WS_DISPATCH"] = "1";
		String broker_socket = first(request.server ? request.server->config["WS_BROKER_SOCKET_PATH"] : String(),
			"/run/uce/ws-broker.sock");
		fcgi_forward_request(broker_socket, dispatch_params, ucb_encode(batch), 5);
	}

	// A cli/serve_http unit that does not export the requested handler is a 404,
	// matching native compiler_invoke_cli. For page render a missing handler is
	// simply an empty body (native parity).
	if(!response.handler_present && handler != "render")
	{
		request.set_status(404, handler == "cli" ? "CLI Entry Point Not Found" : "Handler Not Found");
		return("");
	}

	// Diagnostic timing headers are opt-in: they leak workspace internals and
	// belong to the W5 benchmark harness, not public responses.
	if(config_bool("WASM_BACKEND_VERBOSE", false))
	{
		request.header["X-UCE-Backend"] = "wasm";
		request.header["X-UCE-Wasm-Workspace-Birth-Us"] = std::to_string(response.workspace_birth_us);
		request.header["X-UCE-Wasm-Component-Resolve-Count"] = std::to_string(response.component_resolve_count);
		request.header["X-UCE-Wasm-Component-Resolve-Total-Us"] = std::to_string(response.component_resolve_total_us);
		request.header["X-UCE-Wasm-Component-Resolve-Avg-Us"] = std::to_string(
			response.component_resolve_count ? response.component_resolve_total_us / response.component_resolve_count : 0);
	}

	// status line: keep the native default unless the unit set one
	String status = response.meta["status"].to_string();
	if(status != "")
		request.response_code = status;
	// merge headers over the native defaults (so Content-Type survives unless
	// the unit overrode it); replace cookies/session with the unit's view
	if(response.meta.key("headers"))
		response.meta["headers"].each([&](const DValue& value, String name) {
			request.header[name] = value.to_string();
		});
	if(response.meta.key("cookies"))
		response.meta["cookies"].each([&](const DValue& value, String) {
			request.set_cookies.push_back(value.to_string());
		});
	if(response.meta.key("session"))
	{
		request.session.clear();
		response.meta["session"].each([&](const DValue& value, String name) {
			request.session[name] = value.to_string();
		});
	}

	// body into the request's primary output stream (ob_stack[0]); the
	// transport's assemble_output_buffer concatenates the stack
	if(request.ob)
		request.ob->write(response.body.data(), response.body.size());
	return("");
}

// Stop the ticker before the worker process exits (best-effort; forked workers
// are usually killed, but a clean ager-out path should join the thread).
void wasm_backend_shutdown()
{
	if(g_wasm_epoch_ticker)
	{
		g_wasm_epoch_running.store(false);
		if(g_wasm_epoch_ticker->joinable())
			g_wasm_epoch_ticker->join();
		delete g_wasm_epoch_ticker;
		g_wasm_epoch_ticker = 0;
	}
}