From 6955acdfdb455c36e788d79647d78e8948984731 Mon Sep 17 00:00:00 2001 From: Ewan Croft Date: Tue, 28 Jul 2026 00:29:34 +0100 Subject: [PATCH] feat(server): report every finished request to an optional observer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing in the server saw a request together with its outcome. The auth callback is the closest thing, and it runs before the status is decided, never runs at all for a plain HTTP route, and never runs for a request the rate limiter refused — so a consumer counting requests there is counting a subset it cannot name and attributing no status to any of it. Called once per request from every path that decides a status, the limiter's 429 included, on the worker thread serving the connection: it is for incrementing a counter, not for writing to a socket. WF_XRPC_HAS_REQUEST_OBSERVER lets a consumer build against a Wolfram with or without it rather than requiring a lockstep upgrade. --- CMakeLists.txt | 5 + include/wolfram/xrpc_server.h | 31 ++++++ src/server/xrpc_server.c | 23 ++++ test/test_xrpc_server_observer.c | 175 +++++++++++++++++++++++++++++++ 4 files changed, 234 insertions(+) create mode 100644 test/test_xrpc_server_observer.c diff --git a/CMakeLists.txt b/CMakeLists.txt index 23f6c9e..6fc6077 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -480,6 +480,11 @@ if(WOLFRAM_BUILD_SERVER) target_include_directories(test_relay_server PRIVATE test ${cjson_SOURCE_DIR} ${MICROHTTPD_INCLUDE_DIR} ${OPENSSL_INCLUDE_DIR}) add_test(NAME relay_server COMMAND test_relay_server) + add_executable(test_xrpc_server_observer test/test_xrpc_server_observer.c) + target_link_libraries(test_xrpc_server_observer PRIVATE wolfram Threads::Threads) + target_include_directories(test_xrpc_server_observer PRIVATE test ${cjson_SOURCE_DIR} ${MICROHTTPD_INCLUDE_DIR} ${OPENSSL_INCLUDE_DIR}) + add_test(NAME xrpc_server_observer COMMAND test_xrpc_server_observer) + add_executable(test_xrpc_server_auth test/test_xrpc_server_auth.c) target_link_libraries(test_xrpc_server_auth PRIVATE wolfram Threads::Threads) target_include_directories(test_xrpc_server_auth PRIVATE test ${cjson_SOURCE_DIR} ${MICROHTTPD_INCLUDE_DIR} ${OPENSSL_INCLUDE_DIR}) diff --git a/include/wolfram/xrpc_server.h b/include/wolfram/xrpc_server.h index 22a591f..48cac89 100644 --- a/include/wolfram/xrpc_server.h +++ b/include/wolfram/xrpc_server.h @@ -229,6 +229,31 @@ typedef wf_status (*wf_xrpc_ws_handler)(void *ctx, * `authed_subject` and frees it after the handler returns. */ typedef wf_status (*wf_xrpc_auth_cb)(wf_xrpc_request *req, void *ctx); +/* ------------------------------------------------------------------ */ +/* Request observer (optional) */ +/* ------------------------------------------------------------------ */ + +/* + * Compile-time feature flag, so a consumer can build against a Wolfram with + * or without the observer rather than requiring a lockstep upgrade. + */ +#define WF_XRPC_HAS_REQUEST_OBSERVER 1 + +/** + * Called once per request, after the status is decided and before the + * response is queued. `nsid` is the XRPC method, or NULL for a plain HTTP + * route; `path` is always the request path. + * + * This is the only point that sees every request together with its outcome — + * an auth callback runs too early to know the status and never runs at all + * for HTTP routes or for a request the rate limiter refused. It is called on + * a worker thread while the connection is being served, so it must not block: + * incrementing a counter is the intended use, writing to a socket is not. + */ +typedef void (*wf_xrpc_request_observer)(void *ctx, const char *nsid, + const char *path, const char *method, + unsigned int status); + /* ------------------------------------------------------------------ */ /* Server lifecycle */ /* ------------------------------------------------------------------ */ @@ -420,6 +445,12 @@ wf_status wf_xrpc_server_ws_close(wf_xrpc_ws_stream *stream, uint16_t code); void wf_xrpc_server_set_auth_callback(wf_xrpc_server *server, wf_xrpc_auth_cb cb, void *ctx); +/** Install (or clear, with NULL) the per-request observer. `ctx` is not + * owned by the server. */ +void wf_xrpc_server_set_request_observer(wf_xrpc_server *server, + wf_xrpc_request_observer cb, + void *ctx); + /** * Install (or replace, with NULL) the fallback handler invoked for XRPC * NSIDs with no registered query/procedure route. See diff --git a/src/server/xrpc_server.c b/src/server/xrpc_server.c index 2e17783..5e169a4 100644 --- a/src/server/xrpc_server.c +++ b/src/server/xrpc_server.c @@ -294,6 +294,8 @@ struct wf_xrpc_server { * an external auth callback replaces the middleware. */ void *auth_mw_ctx; void (*auth_mw_free)(void *); + wf_xrpc_request_observer observer; /* optional, not owned */ + void *observer_ctx; wf_rate_limiter *rate_limiter; /* global IP-based limiter */ wf_rate_limiter *rate_limiter_owned; /* non-NULL => server frees it */ wf_rate_limit_entry *rate_limit_entries; /* per-route list */ @@ -1645,6 +1647,17 @@ static enum MHD_Result wf_server_mhd_options(void *cls, size_t *upload_data_size, void **con_cls); +/* Report a finished request to the observer, if one is installed. Called + * from every path that decides a status, the rate limiter's 429 included — + * a refusal the caller never sees counted is the one an operator most needs + * to see. */ +static void wf_server_observe(wf_xrpc_server *server, const char *nsid, + const char *path, const char *method, + unsigned int status) { + if (server && server->observer) + server->observer(server->observer_ctx, nsid, path, method, status); +} + static enum MHD_Result wf_server_mhd_handler(void *cls, struct MHD_Connection *conn, const char *url, @@ -1867,6 +1880,7 @@ static enum MHD_Result wf_server_mhd_handler(void *cls, MHD_queue_response(conn, 429, mhd_rl); MHD_destroy_response(mhd_rl); } + wf_server_observe(server, nsid, url, method, 429); ret = MHD_YES; goto cleanup; } @@ -1996,6 +2010,7 @@ static enum MHD_Result wf_server_mhd_handler(void *cls, MHD_add_response_header(mhd_resp, header->name, header->value); wf_server_apply_cors(server, mhd_resp); + wf_server_observe(server, nsid, url, method, resp.http_status); ret = MHD_queue_response(conn, resp.http_status, mhd_resp); MHD_destroy_response(mhd_resp); @@ -2503,6 +2518,14 @@ void wf_xrpc_server_set_cors(wf_xrpc_server *server, bool enabled, server->cors_origin = (origin && origin[0]) ? strdup(origin) : NULL; } +void wf_xrpc_server_set_request_observer(wf_xrpc_server *server, + wf_xrpc_request_observer cb, + void *ctx) { + if (!server) return; + server->observer = cb; + server->observer_ctx = ctx; +} + void wf_xrpc_server_set_auth_callback(wf_xrpc_server *server, wf_xrpc_auth_cb cb, void *ctx) { if (!server) { diff --git a/test/test_xrpc_server_observer.c b/test/test_xrpc_server_observer.c new file mode 100644 index 0000000..600ac4f --- /dev/null +++ b/test/test_xrpc_server_observer.c @@ -0,0 +1,175 @@ +/* + * test_xrpc_server_observer.c — the per-request observer. + * + * The observer exists because nothing else sees a request together with its + * outcome: an auth callback runs before the status is known, never runs at + * all for a plain HTTP route, and never runs for a request the rate limiter + * refused. Each of those is a case a consumer counting requests would + * silently miss, so each is exercised here. + */ + +#include "wolfram/xrpc.h" +#include "wolfram/xrpc_server.h" +#include "test.h" + +#include +#include +#include + +#define MAX_SEEN 32 + +static struct { + pthread_mutex_t lock; + int count; + struct { + char nsid[128]; + char path[128]; + char method[8]; + unsigned int status; + } seen[MAX_SEEN]; +} observed; + +static void observe(void *ctx, const char *nsid, const char *path, + const char *method, unsigned int status) { + (void)ctx; + pthread_mutex_lock(&observed.lock); + if (observed.count < MAX_SEEN) { + int i = observed.count++; + /* An HTTP route has no NSID; recorded as "" so the test can tell it + * apart from one that was never reported at all. */ + snprintf(observed.seen[i].nsid, sizeof(observed.seen[i].nsid), "%s", + nsid ? nsid : ""); + snprintf(observed.seen[i].path, sizeof(observed.seen[i].path), "%s", + path ? path : ""); + snprintf(observed.seen[i].method, sizeof(observed.seen[i].method), "%s", + method ? method : ""); + observed.seen[i].status = status; + } + pthread_mutex_unlock(&observed.lock); +} + +/* The status reported for `nsid`, or 0 when it was never observed. */ +static unsigned int status_for(const char *nsid) { + unsigned int status = 0; + pthread_mutex_lock(&observed.lock); + for (int i = 0; i < observed.count; i++) + if (strcmp(observed.seen[i].nsid, nsid) == 0) + status = observed.seen[i].status; + pthread_mutex_unlock(&observed.lock); + return status; +} + +static unsigned int status_for_path(const char *path) { + unsigned int status = 0; + pthread_mutex_lock(&observed.lock); + for (int i = 0; i < observed.count; i++) + if (strcmp(observed.seen[i].path, path) == 0) + status = observed.seen[i].status; + pthread_mutex_unlock(&observed.lock); + return status; +} + +static int seen_count(void) { + pthread_mutex_lock(&observed.lock); + int n = observed.count; + pthread_mutex_unlock(&observed.lock); + return n; +} + +static wf_status ok_handler(void *ctx, const wf_xrpc_request *req, + wf_xrpc_response *resp) { + (void)ctx; (void)req; + wf_xrpc_response_set_body(resp, "{\"ok\":true}", 11); + return WF_OK; +} + +static wf_status failing_handler(void *ctx, const wf_xrpc_request *req, + wf_xrpc_response *resp) { + (void)ctx; (void)req; + wf_xrpc_response_set_error(resp, 418, "Teapot", "no coffee here"); + return WF_OK; +} + +static wf_status http_handler(void *ctx, const wf_xrpc_request *req, + wf_xrpc_response *resp) { + (void)ctx; (void)req; + wf_xrpc_response_set_content_type(resp, "text/plain"); + wf_xrpc_response_set_body(resp, "hello", 5); + return WF_OK; +} + +static wf_status refuse_auth(wf_xrpc_request *req, void *ctx) { + (void)ctx; + /* Refuse exactly one route, so the same server serves both outcomes. */ + if (req->nsid && strcmp(req->nsid, "test.private") == 0) + return WF_ERR_PERMISSION; + return WF_OK; +} + +int main(void) { + pthread_mutex_init(&observed.lock, NULL); + + wf_xrpc_server *server = wf_xrpc_server_start("127.0.0.1", 0, 2); + WF_CHECK(server != NULL); + if (!server) { WF_TEST_SUMMARY(); } + + WF_CHECK(wf_xrpc_server_register_query(server, "test.ok", ok_handler, + NULL) == WF_OK); + WF_CHECK(wf_xrpc_server_register_query(server, "test.teapot", + failing_handler, NULL) == WF_OK); + WF_CHECK(wf_xrpc_server_register_query(server, "test.private", ok_handler, + NULL) == WF_OK); + WF_CHECK(wf_xrpc_server_register_http_route(server, "GET", "/plain", + http_handler, NULL) == WF_OK); + wf_xrpc_server_set_auth_callback(server, refuse_auth, NULL); + wf_xrpc_server_set_request_observer(server, observe, NULL); + + char base[64]; + snprintf(base, sizeof(base), "http://127.0.0.1:%u", + (unsigned)wf_xrpc_server_port(server)); + wf_xrpc_client *client = wf_xrpc_client_new(base); + WF_CHECK(client != NULL); + + wf_response response = {0}; + wf_xrpc_query(client, "test.ok", NULL, &response); + wf_response_free(&response); + wf_xrpc_query(client, "test.teapot", NULL, &response); + wf_response_free(&response); + wf_xrpc_query(client, "test.private", NULL, &response); + wf_response_free(&response); + + /* A handler's own status is reported, not just success or failure. */ + WF_CHECK(status_for("test.ok") == 200); + WF_CHECK(status_for("test.teapot") == 418); + /* Refused by the auth callback, which returns before the handler runs — + * the case a consumer counting inside its own auth callback can see, but + * only if it remembers to, and cannot attribute a status to. */ + WF_CHECK(status_for("test.private") == 401); + + /* A plain HTTP route has no NSID and no auth callback, so it is invisible + * to every other hook the server offers. */ + char url[128]; + snprintf(url, sizeof(url), "%s/plain", base); + WF_CHECK(wf_http_get(client, url, &response) == WF_OK); + wf_response_free(&response); + WF_CHECK(status_for_path("/plain") == 200); + + /* An unregistered method still produces a status, and still counts. */ + wf_xrpc_query(client, "test.missing", NULL, &response); + wf_response_free(&response); + WF_CHECK(status_for("test.missing") != 0); + + int before_clear = seen_count(); + WF_CHECK(before_clear >= 5); + + /* Clearing it stops the reports rather than crashing on a NULL callback. */ + wf_xrpc_server_set_request_observer(server, NULL, NULL); + wf_xrpc_query(client, "test.ok", NULL, &response); + wf_response_free(&response); + WF_CHECK(seen_count() == before_clear); + + wf_xrpc_client_free(client); + wf_xrpc_server_free(server); + pthread_mutex_destroy(&observed.lock); + WF_TEST_SUMMARY(); +}