fix: prevent stale mutation execution

docs/wasm-runtime-architecture.md

@@ -136,15 +136,19 @@ refreshes the epoch deadline before its first guest call. Otherwise a component
 whose compilation outlasted the guest CPU budget would immediately trap in the
 following allocator/relocation call even though no guest loop consumed it.
 
-The proactive compiler and request workers coordinate through a per-unit file
+The proactive compiler and request workers coordinate through per-unit file
-lock. Unit compilation writes and validates a process-unique temporary wasm
+locks and a lock-protected demand-priority queue under `BIN_DIRECTORY`. Unit
-file, then publishes it with an atomic rename. When proactive compilation is
+compilation writes and validates a process-unique temporary wasm file, then
-enabled, HTTP and WebSocket requests keep using the last complete artifact
+publishes it with an atomic rename. When proactive compilation is enabled,
-while stale units rebuild in the background; CLI and explicit compile paths
+read-only HTTP requests keep using the last complete artifact while requesting
-remain synchronous. A failed rebuild removes availability and surfaces the
+that stale unit at the head of the compiler queue. Non-read requests never run
-compiler error rather than serving the old unit indefinitely. The per-unit lock
+a stale entry artifact: they return `503 Service Unavailable` with
-also keeps concurrent synchronous compilers from waiting across a transitive
+`Retry-After: 1`, allowing the client to retry after the priority rebuild.
-graph when a last complete artifact is available.
+CLI and explicit compile paths remain synchronous. A failed rebuild removes
+availability and surfaces the compiler error rather than serving the old unit
+indefinitely. The per-unit lock also keeps concurrent synchronous compilers
+from waiting across a transitive graph when a last complete artifact is
+available.
 
 Before preprocessing, the compiler verifies that the worker can actually read
 the source. An unreadable path is a compile failure with a persisted diagnostic;
@@ -367,8 +371,10 @@ header free-functions are `inline`. The wasm backend exposes only declarations
   parent request still renders it. The dependency-invalidation gate also holds
   parent and child compile locks across a transitive source edit, proves a
   warmed HTTP request returns the last atomically published result without
-  waiting, while a CLI request waits and returns only the current dependency
+  waiting, proves a POST returns a prompt retryable 503 without executing its
-  result after rebuild completion.
+  old mutation handler, and verifies demand-priority convergence. A CLI request
+  still waits and returns only the current dependency result after rebuild
+  completion.
 
 - **WebSocket end-to-end**: a headless client performs a raw WS handshake to
   `:HTTP_PORT` with path `/site/tests/websockets.ws.uce` (self-resolving

scripts/test_dependency_invalidation.sh

@@ -11,6 +11,9 @@ if [[ -z "$bin_directory" && -r /etc/uce/settings.cfg ]]; then
 fi
 bin_directory="${bin_directory:-/tmp/uce/work}"
 cache_dir=""
+mutation_file="/tmp/uce-dependency-mutation-$$"
+post_body="/tmp/uce-dependency-post-body-$$"
+post_headers="/tmp/uce-dependency-post-headers-$$"
 http_host="${UCE_TEST_HTTP_HOST:-uce.openfu.com}"
 
 cleanup() {
@@ -18,6 +21,7 @@ cleanup() {
 	if [[ -n "$cache_dir" ]]; then
 		rm -rf "$cache_dir"
 	fi
+	rm -f "$mutation_file" "$post_body" "$post_headers"
 }
 trap cleanup EXIT
 mkdir -p "$source_dir"
@@ -31,7 +35,7 @@ printf '%s\n' \
 printf '%s\n' \
 	'#load "child.uce"' \
 	'CLI(Request& context) { print(dependency_cache_marker(), ":", request_perf()["worker_pid"].to_string()); }' \
-	'RENDER(Request& context) { print(dependency_cache_marker(), ":", request_perf()["worker_pid"].to_string()); }' >"$source_dir/parent.uce"
+	"RENDER(Request& context) { if(context.params[\"REQUEST_METHOD\"] == \"POST\") file_put_contents(\"$mutation_file\", dependency_cache_marker()); print(dependency_cache_marker(), \":\", request_perf()[\"worker_pid\"].to_string()); }" >"$source_dir/parent.uce"
 
 assert_marker() {
 	local path="$1"
@@ -83,6 +87,15 @@ if (( elapsed_ms >= 2000 )); then
 	echo "HTTP request spent ${elapsed_ms}ms rebuilding a stale artifact" >&2
 	exit 1
 fi
+deadline=$((SECONDS + 15))
+while [[ "$http_during_rebuild" != *"dependency-marker-b"* && $SECONDS -lt $deadline ]]; do
+	sleep 0.2
+	http_during_rebuild=$(http_marker)
+done
+if [[ "$http_during_rebuild" != *"dependency-marker-b"* ]]; then
+	echo "requested stale HTTP unit did not receive a demand-priority rebuild" >&2
+	exit 1
+fi
 assert_marker parent dependency-marker-b
 
 # A proactive rebuild owns these same per-unit locks. While it publishes fresh
@@ -111,6 +124,26 @@ if (( elapsed_ms >= 2000 )); then
 	echo "HTTP request waited ${elapsed_ms}ms for an active transitive rebuild" >&2
 	exit 1
 fi
+rm -f "$mutation_file"
+started_at=$(date +%s%N)
+post_status=$(curl -sS -o "$post_body" -D "$post_headers" -w '%{http_code}' -X POST -H "Host: $http_host" "http://127.0.0.1/$test_name/parent.uce")
+elapsed_ms=$(( ($(date +%s%N) - started_at) / 1000000 ))
+if [[ "$post_status" != "503" ]]; then
+	echo "stale POST executed an application artifact instead of returning 503: status=$post_status body=$(cat "$post_body")" >&2
+	exit 1
+fi
+if ! grep -qi '^Retry-After: 1' "$post_headers"; then
+	echo "stale POST did not return a Retry-After header" >&2
+	exit 1
+fi
+if [[ -e "$mutation_file" ]]; then
+	echo "stale POST executed the old mutation handler: $(cat "$mutation_file")" >&2
+	exit 1
+fi
+if (( elapsed_ms >= 2000 )); then
+	echo "stale POST waited ${elapsed_ms}ms instead of failing closed promptly" >&2
+	exit 1
+fi
 started_at=$(date +%s%N)
 assert_marker parent dependency-marker-d
 elapsed_ms=$(( ($(date +%s%N) - started_at) / 1000000 ))

src/lib/compiler.cpp

@@ -312,6 +312,11 @@ String compiler_registry_lock_file_name(Request* context)
 	return(compiler_registry_file_name(context) + ".lock");
 }
 
+String compiler_priority_file_name(Request* context)
+{
+	return(context->server->config["BIN_DIRECTORY"] + "/proactive-priority.txt");
+}
+
 int compiler_open_lock_file(String file_name, String purpose, bool nonblocking = false)
 {
 	(void)purpose;
@@ -1015,6 +1020,58 @@ bool compiler_request_can_serve_stale_artifact(Request* context)
 		float_val(context->server->config["PROACTIVE_COMPILE_CHECK_INTERVAL"]) > 0);
 }
 
+void compiler_prioritize_unit(Request* context, String file_name)
+{
+	if(!compiler_request_can_serve_stale_artifact(context))
+		return;
+	file_name = compiler_normalize_unit_path(context, file_name);
+	if(file_name == "" || file_name.find('\n') != String::npos || file_name.find('\r') != String::npos)
+		return;
+
+	String queue_file = compiler_priority_file_name(context);
+	int fd = compiler_open_lock_file(queue_file + ".lock", "proactive-priority");
+	if(fd < 0)
+		return;
+	int queue_fd = open(queue_file.c_str(), O_WRONLY | O_CREAT | O_APPEND, 0666);
+	if(queue_fd >= 0)
+	{
+		String line = file_name + "\n";
+		ssize_t offset = 0;
+		while(offset < (ssize_t)line.size())
+		{
+			ssize_t written = write(queue_fd, line.data() + offset, line.size() - offset);
+			if(written <= 0)
+				break;
+			offset += written;
+		}
+		close(queue_fd);
+	}
+	compiler_close_lock_file(fd);
+}
+
+StringList compiler_take_priority_units(Request* context)
+{
+	StringList result;
+	if(!context || !context->server)
+		return(result);
+	String queue_file = compiler_priority_file_name(context);
+	int fd = compiler_open_lock_file(queue_file + ".lock", "proactive-priority");
+	if(fd < 0)
+		return(result);
+	if(file_exists(queue_file))
+	{
+		for(auto file_name : split(file_get_contents(queue_file), "\n"))
+		{
+			file_name = trim(file_name);
+			if(file_name != "" && std::find(result.begin(), result.end(), file_name) == result.end())
+				result.push_back(file_name);
+		}
+		file_put_contents(queue_file, "");
+	}
+	compiler_close_lock_file(fd);
+	return(result);
+}
+
 void unit_render(String file_name)
 {
 	unit_render(file_name, *context);

src/lib/compiler.h

@@ -19,6 +19,8 @@ String compiler_error_page_unit(Request* context, String config_key);
 bool compiler_unit_compile_pending(Request* context, String file_name);
 bool compiler_unit_compile_in_progress(Request* context, String file_name);
 bool compiler_request_can_serve_stale_artifact(Request* context);
+void compiler_prioritize_unit(Request* context, String file_name);
+StringList compiler_take_priority_units(Request* context);
 String compiler_site_directory(Request* context);
 StringList compiler_scan_site_units(Request* context);
 StringList compiler_list_known_units(Request* context);

src/linux_fastcgi.cpp

@@ -575,6 +575,17 @@ int handle_complete(FastCGIRequest& request) {
 			};
 
 			String entry_unit = compiler_normalize_unit_path(&request, request.params["SCRIPT_FILENAME"]);
+			String request_method = to_upper(trim(request.params["REQUEST_METHOD"]));
+			bool read_request = request_method == "GET" || request_method == "HEAD" || request_method == "OPTIONS";
+			bool stale_mutation = !request.resources.is_cli && !read_request && compiler_unit_needs_recompile(&request, entry_unit);
+			if(stale_mutation)
+			{
+				compiler_prioritize_unit(&request, entry_unit);
+				request.set_status(503, "Service Unavailable");
+				request.header["Content-Type"] = "text/plain; charset=utf-8";
+				request.header["Retry-After"] = "1";
+				print("The requested code is being updated. Retry this request shortly.\n");
+			}
 			// W7e: every unit runs on wasm. When the artifact is missing or stale
 			// (cold worker, or source edited since the last compile), compile the
 			// unit on demand — get_shared_unit() builds the .wasm side-module — and
@@ -593,7 +604,11 @@ int handle_complete(FastCGIRequest& request) {
 				failure_trace = "source: " + request.params["SCRIPT_FILENAME"];
 			};
 
-			if(request.params["UCE_WS"] == "1")
+			if(stale_mutation)
+			{
+				// The response above deliberately bypasses all stale application code.
+			}
+			else if(request.params["UCE_WS"] == "1")
 			{
 				// A WS message the broker forwarded here: rebuild the connection
 				// context the broker passed as params, then run __uce_websocket.
@@ -1243,7 +1258,13 @@ void run_proactive_compiler()
 	{
 		try
 		{
-		if(compile_queue.size() == 0 && time_precise() >= next_scan_at)
+			auto priority_units = compiler_take_priority_units(&background_context);
+			for(auto it = priority_units.rbegin(); it != priority_units.rend(); ++it)
+			{
+				compile_queue.erase(std::remove(compile_queue.begin(), compile_queue.end(), *it), compile_queue.end());
+				compile_queue.insert(compile_queue.begin(), *it);
+			}
+			if(compile_queue.size() == 0 && time_precise() >= next_scan_at)
 		{
 			auto tracked_units = compiler_list_known_units(&background_context);
 			StringList existing_units;

src/wasm/backend.cpp

@@ -100,7 +100,10 @@ static bool wasm_artifact_exists(Request* context, const String& entry_unit)
 	// metadata mismatches, which can leave stale wasm with old imports.
 	bool source_missing = false;
 	if(compiler_unit_needs_recompile(context, entry_unit, &source_missing))
+	{
+		compiler_prioritize_unit(context, entry_unit);
 		return(compiler_request_can_serve_stale_artifact(context));
+	}
 	if(source_missing)
 		return(false);
 	return(true);

src/wasm/worker.cpp

@@ -1606,7 +1606,10 @@ private:
 			return(1);
 		}
 
-		if(!file_exists_host(worker.unit_wasm_path(resolved)) || (compiler_unit_needs_recompile(context, resolved, 0) && !compiler_request_can_serve_stale_artifact(context) && !compiler_unit_compile_in_progress(context, resolved)))
+		bool stale = compiler_unit_needs_recompile(context, resolved, 0);
+		if(stale && compiler_request_can_serve_stale_artifact(context))
+			compiler_prioritize_unit(context, resolved);
+		if(!file_exists_host(worker.unit_wasm_path(resolved)) || (stale && !compiler_request_can_serve_stale_artifact(context) && !compiler_unit_compile_in_progress(context, resolved)))
 			get_shared_unit(context, resolved);
 
 		size_t unit_index = 0;