diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f15e3e0..9df519eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,11 @@ loosely while pre-1.0 (breaking changes can land on minor bumps). ## [Unreleased] ### Fixed +- **SSE mailbox terminal after overflow.** `sse_mailbox_push(..., + force_deliver=true)` now evicts oldest events and delivers a terminal + envelope even when a prior push already set `overflowed`. Request-event + replay and A2A resubscribe close with the real `done` / task-status + frame instead of dropping it for `slow_consumer`. - **Event ingest hydrates the routed constitution.** `POST /v1/events` attaches the selected agent's `agent_def` on the synthetic orchestrate body. File-backed matches and tenant agents off the newest-200 catalog diff --git a/include/sse_mailbox.h b/include/sse_mailbox.h index a118cd7f..56b26c88 100644 --- a/include/sse_mailbox.h +++ b/include/sse_mailbox.h @@ -19,7 +19,9 @@ inline constexpr size_t kDefaultSseMailboxMaxDepth = 2048; // Push one item into a per-connection mailbox. Returns false when the // item was rejected because the client is too slow (overflowed is set). // When `force_deliver` is true (terminal SSE envelopes), drop the oldest -// buffered items until there is room so the stream can close cleanly. +// buffered items until there is room so the stream can close cleanly — +// including after a prior overflow, which would otherwise reject the +// terminal frame and leave the subscriber with only `slow_consumer`. template bool sse_mailbox_push(std::deque& mailbox, std::mutex& mu, @@ -28,7 +30,8 @@ bool sse_mailbox_push(std::deque& mailbox, T item, bool force_deliver = false) { std::lock_guard lk(mu); - if (overflowed.load(std::memory_order_relaxed)) return false; + if (overflowed.load(std::memory_order_relaxed) && !force_deliver) + return false; const size_t cap = kDefaultSseMailboxMaxDepth; if (mailbox.size() >= cap) { if (force_deliver) { @@ -42,6 +45,10 @@ bool sse_mailbox_push(std::deque& mailbox, } } mailbox.push_back(std::move(item)); + // Terminal delivery wins over a prior overflow so the drain loop + // writes the real done/status envelope instead of aborting. + if (force_deliver) + overflowed.store(false, std::memory_order_relaxed); cv.notify_one(); return true; } diff --git a/tests/test_sse_mailbox.cpp b/tests/test_sse_mailbox.cpp index ea366403..50575752 100644 --- a/tests/test_sse_mailbox.cpp +++ b/tests/test_sse_mailbox.cpp @@ -46,3 +46,27 @@ TEST_CASE("sse_mailbox_push force_deliver makes room for terminal item") { CHECK(mailbox.back() == "terminal"); CHECK(mailbox.front() == "1"); } + +TEST_CASE("sse_mailbox_push force_deliver still lands after overflow") { + std::mutex mu; + std::condition_variable cv; + std::deque mailbox; + std::atomic overflowed{false}; + + for (size_t i = 0; i < kDefaultSseMailboxMaxDepth; ++i) { + REQUIRE(sse_mailbox_push(mailbox, mu, cv, overflowed, + std::to_string(i))); + } + CHECK_FALSE(sse_mailbox_push(mailbox, mu, cv, overflowed, + std::string("dropped"))); + REQUIRE(overflowed.load()); + CHECK(mailbox.size() == kDefaultSseMailboxMaxDepth); + CHECK(mailbox.back() != "dropped"); + + REQUIRE(sse_mailbox_push(mailbox, mu, cv, overflowed, + std::string("terminal"), /*force_deliver=*/true)); + CHECK_FALSE(overflowed.load()); + CHECK(mailbox.size() == kDefaultSseMailboxMaxDepth); + CHECK(mailbox.back() == "terminal"); + CHECK(mailbox.front() == "1"); +}