From c2255853d3216a03b30aba43277f51d5c413906f Mon Sep 17 00:00:00 2001 From: joelteply Date: Wed, 11 Nov 2026 14:21:28 -0600 Subject: [PATCH] server: fan ticket results out to each event subscriber --- tools/server/server-context.cpp | 14 +++++----- tools/server/server-queue.cpp | 41 +++++++++++++++++------------ tools/server/server-queue.h | 26 +++++++++++------- tools/server/tests/learn-tickets.py | 6 +++++ 4 files changed, 52 insertions(+), 35 deletions(-) diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 6c065a19bb4..24755b56df7 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -6605,12 +6605,12 @@ void server_routes::init_routes() { this->get_learn_events = [this](const server_http_req & req) { struct ticket_stream : server_res_generator { server_response & results; + const int subscriber; ticket_stream(server_queue & tasks, server_response & results, int sleep_idle_seconds) - : server_res_generator(tasks, results, sleep_idle_seconds, true), results(results) { - results.subscribe_tickets(); - } + : server_res_generator(tasks, results, sleep_idle_seconds, true), results(results), + subscriber(results.subscribe_tickets()) {} void on_complete() override { - results.unsubscribe_tickets(); + results.unsubscribe_tickets(subscriber); server_res_generator::on_complete(); } }; @@ -6629,12 +6629,10 @@ void server_routes::init_routes() { res_this->data.clear(); return true; } - auto result = res_this->results.next_ticket(keepalive); - if (!result) { + output = res_this->results.next_ticket(res_this->subscriber, keepalive); + if (output.empty()) { output = ":\n\n"; // keepalive - return true; } - output = "data: " + safe_json_to_str({{"ticket", result->id}, {"error", result->is_error()}, {"result", result->to_json()}}) + "\n\n"; return true; }); return res; diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index 8d3ae9350be..e736e53ed54 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -482,9 +482,13 @@ void server_response::send(server_task_result_ptr && result) { std::unique_lock lock(mutex_results); if (ticketed_ids.erase(result->id) > 0) { - // stored before the notify, under the same lock the stream waits with (no lost wakeup) - if (ticket_subscribers > 0) { - ticket_results.emplace_back(std::move(result)); + // stored before the notify, under the same lock the streams wait with (no lost wakeup) + if (!ticket_queues.empty()) { + const std::string event = "data: " + safe_json_to_str({ + {"ticket", result->id}, {"error", result->is_error()}, {"result", result->to_json()}}) + "\n\n"; + for (auto & [subscriber, queue] : ticket_queues) { + queue.push_back(event); + } condition_results.notify_all(); } return; @@ -633,26 +637,29 @@ void server_response::add_ticket(int id_task) { ticketed_ids.insert(id_task); } -void server_response::subscribe_tickets() { +int server_response::subscribe_tickets() { std::unique_lock lock(mutex_results); - ++ticket_subscribers; + const int subscriber = next_ticket_subscriber++; + ticket_queues[subscriber]; + return subscriber; } -void server_response::unsubscribe_tickets() { +void server_response::unsubscribe_tickets(int subscriber) { std::unique_lock lock(mutex_results); - if (--ticket_subscribers <= 0) { - ticket_subscribers = 0; - ticket_results.clear(); // nobody is listening: undelivered results are gone - } + ticket_queues.erase(subscriber); // what it never read is gone with it } -server_task_result_ptr server_response::next_ticket(int timeout) { +std::string server_response::next_ticket(int subscriber, int timeout) { std::unique_lock lock(mutex_results); - condition_results.wait_for(lock, std::chrono::seconds(timeout), [&] { return !running || !ticket_results.empty(); }); - if (!running || ticket_results.empty()) { - return nullptr; + condition_results.wait_for(lock, std::chrono::seconds(timeout), [&] { + auto it = ticket_queues.find(subscriber); + return !running || it == ticket_queues.end() || !it->second.empty(); + }); + auto it = ticket_queues.find(subscriber); + if (!running || it == ticket_queues.end() || it->second.empty()) { + return {}; } - server_task_result_ptr res = std::move(ticket_results.front()); - ticket_results.erase(ticket_results.begin()); - return res; + std::string event = std::move(it->second.front()); + it->second.pop_front(); + return event; } diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index 16426ebc591..dd48aff6c0a 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -8,6 +8,8 @@ #include #include #include +#include +#include #include // struct for managing server tasks @@ -164,23 +166,27 @@ struct server_response { std::mutex mutex_results; std::condition_variable condition_results; - // Continuum fork (card 7fc7e2ce): ticketed loop mutations. A ticketed task's result goes to the - // learning event stream's subscriber, then it is gone (delivery is the drop); with no subscriber - // it is dropped, and a core that (re)connects reads the physical state, never a replay. + // Continuum fork (card 7fc7e2ce): ticketed loop mutations. A ticketed task's result goes to EVERY + // open learning event stream, each through its own queue, then it is gone (delivery is the drop); + // with no subscriber it is dropped, and a core that (re)connects reads the physical state, never a + // replay. One shared queue let a stale stream (its client gone, its socket not yet failed) take the + // result a live core was waiting for, and that core waited forever (the IntelMac, 2026-11-11 19:58Z). + // The core matches by ticket id, so a copy on a stale stream costs nothing. std::unordered_set ticketed_ids; - std::vector ticket_results; - int ticket_subscribers = 0; + std::unordered_map> ticket_queues; // per subscriber: formatted events + int next_ticket_subscriber = 0; public: // register `id_task` as ticketed: its result is delivered on the event stream, not to a waiter void add_ticket(int id_task); - // the learning event stream opens / closes; the last close drops anything undelivered - void subscribe_tickets(); - void unsubscribe_tickets(); + // a learning event stream opens (returns its subscriber id) / closes (drops what it never read) + int subscribe_tickets(); + void unsubscribe_tickets(int subscriber); - // blocks until a ticketed result arrives (returned) or `timeout` seconds pass (nullptr: keepalive) - server_task_result_ptr next_ticket(int timeout); + // blocks until an event for `subscriber` arrives (returned, SSE-ready) or `timeout` seconds pass + // (empty: keepalive) + std::string next_ticket(int subscriber, int timeout); // add the id_task to the list of tasks waiting for response void add_waiting_task_id(int id_task); diff --git a/tools/server/tests/learn-tickets.py b/tools/server/tests/learn-tickets.py index bbe181ce0ef..52e87d17d44 100644 --- a/tools/server/tests/learn-tickets.py +++ b/tools/server/tests/learn-tickets.py @@ -54,6 +54,9 @@ def ticketed(route, body): unseen = ticketed("/learn/grants/revoke", {"id": gid_unseen}) stream, events = subscribe() +# a second open stream, as a reconnecting core leaves its predecessor's until its socket fails: each +# stream gets every result, so a stale one can never take the result a live core waits for +other, other_events = subscribe() # a ticketed grant revoke: its terminal result arrives on the stream under its ticket gid = engine.register_grant("-revoked") @@ -63,6 +66,9 @@ def ticketed(route, body): assert event["ticket"] != unseen, ("a result made while nobody listened is never replayed", event) assert event["ticket"] == ticket and not event["error"], event assert event["result"]["revoked"] is True, event +copy = other_events.get(timeout=engine.timeout) +assert copy["ticket"] == ticket, ("every open stream gets the result, never just one of them", copy) +other.close() # a ticketed unload: the adapter's retirement arrives the same way shadow = engine.fresh()