Retire workers after large CLI units.

This commit is contained in:
root committed 2026-09-04 20:19:01 +00:00
1 parent 9399a1b7ee
commit e0b1c420b0
5 files changed
+167 -8

No files matched your search

+20
View File
@@ -174,3 +174,23 @@ Each fault test must verify cleanup count, payload release, ARC count, and mappe
3. Add fault-injection coverage and the dense-DValue size regression gate from the patch.
4. Run the complete native, runtime, source-map, reproducibility, and golden suites.
5. Review and commit the change. Do not deploy it without separate approval.
## 2026-09-04 isolated acceptance
The DValue allocation helper was applied after review. This changed Wasm and source-map bytes as expected. The dense-DValue native gate, artifact golden gate, and cross-directory reproducibility gate pass.
The current Ludum CLI source produces a 2,571,192-byte Wasm artifact and a 17,459,328-byte serialized module. Standalone Wasmtime preparation peaked at 1,296,348 KiB. The native compiler used 30,948 KiB and completed in 0.40 seconds.
A runtime lifecycle change now serializes Capy CLI execution. A worker retires after it serves a CLI artifact larger than 1 MiB. The worker closes its listeners and unrelated clients. It uses an output-only drain for the completed CLI response. The CLI client retries only the explicit `BEARER_CLI_BUSY` response while the prior worker releases the admission lock.
The lifecycle passed these checks on isolated CT105:
1. A 1,200,000-byte response arrived without truncation.
2. A slow reader received the complete response.
3. A peer disconnect did not prevent worker replacement.
4. Two concurrent large CLI calls completed without concurrent module residency.
5. A normal HTTP page returned status 200 during the test.
6. The complete Ludum suite passed 681 checks across 34 groups.
7. The Bearer cgroup recorded no OOM event. Memory returned below 700 MiB after worker replacement.
The highest observed service memory peak was 1,993,117,696 bytes under the 3 GiB service limit. This was an isolated test deployment. Production remains unchanged. Production deployment still requires explicit approval.
+26 -3
View File
@@ -94,9 +94,32 @@ if [[ "$method" == "auto" ]]; then
fi
fi
curl_args=(-sS --fail-with-body --unix-socket "$socket_path")
run_curl() {
local attempt body_file error_file http_status status
for ((attempt = 0; attempt <= 10; ++attempt)); do
body_file=$(mktemp)
error_file=$(mktemp)
status=0
http_status=$(curl "$@" --output "$body_file" --write-out '%{http_code}' 2>"$error_file") || status=$?
if [[ "$http_status" == 503 ]] && grep -qx 'BEARER_CLI_BUSY: another Capy CLI unit is active' "$body_file" && ((attempt < 10)); then
rm -f "$body_file" "$error_file"
sleep 1
continue
fi
cat "$body_file"
if ((status != 0)); then
cat "$error_file" >&2
fi
rm -f "$body_file" "$error_file"
return "$status"
done
}
if [[ -n "$raw_json" ]]; then
printf '%s' "$raw_json" | python3 -m json.tool >/dev/null
curl -sS --fail-with-body --unix-socket "$socket_path" \
run_curl "${curl_args[@]}" \
-X POST \
-H 'Content-Type: application/json' \
--data-binary "$raw_json" \
@@ -120,7 +143,7 @@ PY
)
url="$url?$query"
fi
curl -sS --fail-with-body --unix-socket "$socket_path" "$url"
run_curl "${curl_args[@]}" "$url"
else
json=$(python3 - "$@" <<'PY'
import json, sys
@@ -148,7 +171,7 @@ for arg in sys.argv[1:]:
print(json.dumps(payload, separators=(',', ':')))
PY
)
curl -sS --fail-with-body --unix-socket "$socket_path" \
run_curl "${curl_args[@]}" \
-X POST \
-H 'Content-Type: application/json' \
--data-binary "$json" \
+39
View File
@@ -762,6 +762,45 @@ FastCGIServer::process(int timeout_ms)
open_client_connection(accepted.first, accepted.second);
}
void
FastCGIServer::drain_output(int timeout_ms)
{
poll_descriptors.clear();
poll_descriptors.reserve(client_sockets.size());
for(const auto& item : client_sockets)
poll_descriptors.push_back({item.first, (short)(item.second->output_buffer.empty() ? 0 : POLLOUT), 0});
if(poll_descriptors.empty())
return;
int poll_result = poll(poll_descriptors.data(), poll_descriptors.size(), timeout_ms);
if(poll_result == -1)
{
if(errno == EINTR)
return;
throw std::runtime_error("poll() failed during output drain");
}
size_t index = 0;
for(auto it = client_sockets.begin(); it != client_sockets.end();)
{
Connection* connection = it->second;
short events = poll_descriptors[index++].revents;
bool close_connection = connection->output_buffer.empty() || (events & (POLLNVAL | POLLHUP | POLLERR));
if(!close_connection && (events & POLLOUT))
close_connection = send_output_buffer(*connection) == -1 || connection->output_buffer.empty();
if(!close_connection)
{
++it;
continue;
}
printf("Closing socket %i\n", it->first);
close(it->first);
for(auto& request : connection->requests)
delete request.second;
delete connection;
it = client_sockets.erase(it);
}
}
bool
FastCGIServer::parse_http_message(FastCGIRequest& request, String& data)
{
+1
View File
@@ -61,6 +61,7 @@ struct FastCGIServer {
int adopt_listener(int socket_handle, char type = 'F');
void process(int timeout_ms = -1); // timeout_ms<0 blocks forever
void drain_output(int timeout_ms);
void process_forever();
int calls_until_termination = 8; // set this to -1 to never terminate
bool resolve_http_script_filename = true;
+81 -5
View File
@@ -12,7 +12,9 @@
#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <sys/file.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/syscall.h>
#include <sys/wait.h>
@@ -62,6 +64,8 @@ pid_t ws_broker_pid = 0;
std::map<int, WsBrokerOutbound> ws_broker_outbound;
std::map<String, u64> ws_broker_outbound_per_connection;
u64 ws_broker_outbound_bytes = 0;
volatile sig_atomic_t termination_signal_received = 0;
u64 worker_drain_client_socket = 0;
static sigjmp_buf request_fault_jmp;
static volatile sig_atomic_t request_fault_active = 0;
static volatile sig_atomic_t request_fault_signal = 0;
@@ -383,6 +387,30 @@ u64 request_seed_from_time(f64 time_value)
return(gen_noise64(bits));
}
static constexpr u64 CLI_RETAINED_WASM_MAX_BYTES = 1024 * 1024;
bool cli_unit_requires_disposable_worker(Request& request, const String& unit)
{
struct stat info = {};
String path = compiler_unit_wasm_path(&request, unit);
return(stat(path.c_str(), &info) == 0 && S_ISREG(info.st_mode) && info.st_size > 0 &&
(u64)info.st_size > CLI_RETAINED_WASM_MAX_BYTES);
}
int cli_unit_lock()
{
String path = path_join(server_state.config["BIN_DIRECTORY"], ".cli-unit.lock");
int fd = open(path.c_str(), O_CREAT | O_RDWR | O_CLOEXEC | O_NOFOLLOW, 0600);
if(fd < 0)
return(-1);
if(flock(fd, LOCK_EX | LOCK_NB) == 0)
return(fd);
int lock_error = errno;
close(fd);
errno = lock_error;
return(-1);
}
void prepare_request_body_maps(Request& request)
{
if(request.server)
@@ -412,6 +440,8 @@ void prepare_request_body_maps(Request& request)
int handle_cli_complete(FastCGIRequest& request)
{
int cli_unit_lock_fd = -1;
bool dispose_worker = false;
Request* previous_context = set_active_request(request);
server_state.request_count += 1;
request.server = &server_state;
@@ -450,7 +480,22 @@ int handle_cli_complete(FastCGIRequest& request)
}
else if(command.length() >= 5 && command.substr(command.length() - 5) == ".capy")
{
if(!cli_path_is_safe(command))
cli_unit_lock_fd = cli_unit_lock();
if(cli_unit_lock_fd < 0)
{
const int lock_error = errno;
if(lock_error == EWOULDBLOCK || lock_error == EAGAIN)
{
request.set_status(503, "Service Unavailable");
print("BEARER_CLI_BUSY: another Capy CLI unit is active\n");
}
else
{
request.set_status(500, "Internal Server Error");
print("BEARER_CLI_ADMISSION: could not lock Capy CLI execution: ", strerror(lock_error), "\n");
}
}
else if(!cli_path_is_safe(command))
{
request.set_status(400, "Bad Request");
print("invalid BEARER CLI unit path\n");
@@ -496,7 +541,9 @@ int handle_cli_complete(FastCGIRequest& request)
}
else if(cli_wasm_ready)
{
dispose_worker = cli_unit_requires_disposable_worker(request, cli_unit);
String wasm_error = wasm_backend_serve(request, cli_unit, "cli", wasm_invocation_remaining_ms(invocation_deadline));
dispose_worker = dispose_worker || cli_unit_requires_disposable_worker(request, cli_unit);
if(wasm_error != "")
{
request.set_status(500, "Internal Server Error");
@@ -536,6 +583,17 @@ int handle_cli_complete(FastCGIRequest& request)
file_unlink(f.tmp_name);
cleanup_mysql_connections();
cleanup_sqlite_connections();
if(cli_unit_lock_fd >= 0 && dispose_worker)
{
// Keep admission locked until shutdown releases the large Wasmtime module.
worker_drain_client_socket = request.resources.client_socket;
termination_signal_received = 1;
}
else if(cli_unit_lock_fd >= 0)
{
flock(cli_unit_lock_fd, LOCK_UN);
close(cli_unit_lock_fd);
}
restore_active_request(previous_context);
return(request.flags.status);
}
@@ -773,11 +831,9 @@ int handle_complete(FastCGIRequest& request) {
}
volatile bool termination_signal_received = false;
void on_terminate(int sig)
{
termination_signal_received = true;
termination_signal_received = 1;
}
void clear_shared_unit_cache(ServerState& state)
@@ -795,6 +851,24 @@ void close_inherited_server_sockets()
server.server_socket_types.clear();
}
void close_worker_clients_except(u64 retained_socket)
{
for(auto it = server.client_sockets.begin(); it != server.client_sockets.end();)
{
if((u64)it->first == retained_socket)
{
++it;
continue;
}
shutdown(it->first, SHUT_RDWR);
close(it->first);
for(auto& request : it->second->requests)
delete request.second;
delete it->second;
it = server.client_sockets.erase(it);
}
}
void install_process_fault_handlers()
{
signal(SIGSEGV, on_segfault);
@@ -2134,10 +2208,12 @@ void listen_for_connections()
reap_child_exits();
}
close_inherited_server_sockets();
if(worker_drain_client_socket != 0)
close_worker_clients_except(worker_drain_client_socket);
f64 drain_deadline = time_precise() + (f64)to_u64(server_state.config["WORKER_DRAIN_TIMEOUT_SECONDS"], 10);
while(!server.client_sockets.empty() && time_precise() < drain_deadline)
{
server.process(100);
server.drain_output(100);
reap_child_exits();
}
if(!server.client_sockets.empty())