Report every finished request to an optional observer

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})

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

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) {

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 <pthread.h>
+#include <stdlib.h>
+#include <string.h>
+
+#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();
+}