fix(subscribe): unwind streams on a shutdown signal, and make the writer safe to race

v2.11.2 bounded Shutdown() with a deadline so a subscriber could not hang the
stop. The stream still only noticed the stop when that deadline expired and gRPC
cancelled it, so every stop paid the grace period. beginShutdown() now sets a
flag and notifies a condition variable, and stopGrpcServer() calls it before
Shutdown(), so the streams are gone before gRPC is asked to stop. Measured 0ms
between 'Stopping database service...' and the handler reporting its close; the
deadline is now a backstop rather than the mechanism.

The 100ms timeout stays inside the wait: the sync API offers no way to be woken
when a CLIENT disconnects, so that half is still polled.

This required fixing a writer-lifetime bug first, and would have made it worse
otherwise. EventManager::publish() copies matching callbacks out under its lock
and invokes them outside it, so unsubscribe() does not stop a dispatch already
running - a callback could write through 'writer' after the handler returned and
gRPC freed it. Narrow before; unwinding every subscriber at once is exactly what
widens it. A per-stream shared state {mutex, open, writer} closes it in both
directions, and the same mutex serialises concurrent writes from several
publishing threads, which grpc::ServerWriter does not permit and nothing
prevented before.

Handlers now return UNAVAILABLE on shutdown instead of OK, which is what a gRPC
client already retries on.

Not fixed, measured and recorded in the roadmap: ~4s inside Shutdown() draining a
still-connected client's transport (1ms with the client killed first) and ~4s
joining component threads after persistence stops. The test therefore asserts
the handler's own unwind latency from the log rather than process wall-clock,
which would have been asserting gRPC's drain by accident.

tests: ctest 23/23; test_shutdown_with_subscriber asserts 0ms unwind and fails
with 'the Subscribe handler never reported a shutdown close' when the
beginShutdown() call is removed.

VERSION

@@ -1 +1 @@
-2.11.2
+2.11.3

docs/ROADMAP.md

@@ -1,6 +1,6 @@
 # Smartbotic Database - Status and Roadmap
 
-**Current version: 2.11.2** (see `VERSION`). Last reviewed: 2026-08-10.
+**Current version: 2.11.3** (see `VERSION`). Last reviewed: 2026-08-10.
 
 This is the single authoritative statement of what exists and what does not.
 If any other document in this repository disagrees with this one, this one is
@@ -53,7 +53,7 @@ Consequences of that substitution, which trip up readers:
 
 ## Shipped
 
-Every item below is in the installed product as of 2.11.2. `CLAUDE.md` has the
+Every item below is in the installed product as of 2.11.3. `CLAUDE.md` has the
 detail and the failure modes.
 
 - JSON document store: collections, version history, field-level encryption, TTL
@@ -190,6 +190,23 @@ reason each was left:
   becomes readable. Deliberate (a row predicate selects rows, and the row's
   identity is its current document) but worth restating before security is armed.
 
+### 5c. Shutdown latency after v2.11.2
+
+The Subscribe unwind is instant (0 ms measured). What remains in a stop, neither
+caused nor worsened by v2.11.2, both bounded, neither yet investigated:
+
+- **~4 s inside `grpc::Server::Shutdown()` when a client is still connected**,
+  draining that connection's transport. With the client killed first, Shutdown
+  returns in 1 ms. Lowering the 5 s grace would cut the tail, but it would also
+  cancel legitimately long streaming calls sooner (`DownloadFile` chunks a whole
+  file), so it is not a free change.
+- **~4 s between "Persistence manager stopped" and "Database service stopped"**,
+  i.e. joining component threads. Unmeasured per-thread; likely a poll interval
+  somewhere that could become a condition variable.
+- `stop()` appears to run some component stops twice ("FileManager stopped" is
+  logged again after "Database service exited cleanly"). Harmless, noisy, worth
+  a look when either of the above is picked up.
+
 ### 6. Smaller known gaps
 
 - Policy management has no dedicated RPCs; `_policies` is edited through the

service/src/database_grpc_impl.cpp

@@ -15,7 +15,10 @@
 #include <nlohmann/json.hpp>
 
 #include <algorithm>
+#include <chrono>
 #include <functional>
+#include <memory>
+#include <mutex>
 #include <queue>
 #include <utility>
 #include <vector>
@@ -2430,13 +2433,45 @@ grpc::Status DatabaseGrpcImpl::Subscribe(
     std::vector<std::string> patterns(request->patterns().begin(), request->patterns().end());
     bool includeData = request->include_data();
 
+    // v2.11.2 — WRITER LIFETIME. `writer` and `context` belong to this call and
+    // die when this function returns, but the event callback runs on whatever
+    // thread published the event: EventManager::publish() copies the matching
+    // callbacks out under its lock and then invokes them OUTSIDE it, so
+    // unsubscribing does not stop a dispatch already in progress. A callback
+    // could therefore still be writing through `writer` after this handler had
+    // returned and gRPC had freed it.
+    //
+    // That window was always there (a client disconnecting while an event was
+    // being published) but it stayed narrow. Making every subscriber unwind at
+    // once on shutdown - the point of this change - is exactly the situation
+    // that widens it, so the lifetime has to be made safe first.
+    //
+    // This shared state closes it in both directions: the callback takes the
+    // mutex and gives up if the stream is closed, and the handler closes it
+    // under the same mutex, so on return either the write finished or it never
+    // started. The mutex also serialises concurrent writes from several
+    // publishing threads, which grpc::ServerWriter does not permit and which
+    // nothing prevented before.
+    struct StreamState {
+        std::mutex mu;
+        bool open = true;
+        grpc::ServerWriter<pb::DatabaseEvent>* writer = nullptr;
+    };
+    auto state = std::make_shared<StreamState>();
+    state->writer = writer;
+
     // v2.8.0 — Subscribe is gated PER EVENT, not once at entry. An empty
     // `collections` list means "every collection" and `patterns` accepts globs,
     // so there is no single name to authorise up front. Filtering in the
     // callback is also what makes a partial grant work: a principal subscribed
     // to everything receives only the collections it may read.
     uint64_t subId = events_.subscribe(collections, patterns,
-                                        [this, writer, includeData, context](const DatabaseEvent& event) {
+                                        [this, state, includeData, context](const DatabaseEvent& event) {
+        // Held for the whole callback: the writer must not be touched once the
+        // handler has closed the stream, and two publishers must not write
+        // concurrently.
+        std::lock_guard<std::mutex> streamLock(state->mu);
+        if (!state->open) return;
         if (context->IsCancelled()) return;
 
         smartbotic::database::Decision edec;
@@ -2467,19 +2502,68 @@ grpc::Status DatabaseGrpcImpl::Subscribe(
             }
         }
 
-        writer->Write(protoEvent);
+        state->writer->Write(protoEvent);
     });
 
-    // Block until cancelled
-    while (!context->IsCancelled()) {
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    // v2.11.2 — park until the client goes away OR the service starts stopping.
+    //
+    // This used to be a 100 ms sleep loop over context->IsCancelled() alone,
+    // which meant a stopping server was noticed only once the shutdown deadline
+    // expired and gRPC cancelled the call - so every stop waited out the full
+    // grace period with nothing to show for it. Waiting on the condition
+    // variable returns the moment beginShutdown() fires; the 100 ms timeout
+    // remains because the sync API offers no way to be woken when a client
+    // disconnects, so that half still has to be polled.
+    bool stopping = false;
+    {
+        std::unique_lock<std::mutex> lock(shutdownMutex_);
+        while (!context->IsCancelled()) {
+            if (shutdownCv_.wait_for(lock, std::chrono::milliseconds(100), [this] {
+                    return shuttingDown_.load(std::memory_order_acquire);
+                })) {
+                stopping = true;
+                break;
+            }
+        }
     }
 
+    // Order matters. unsubscribe() first so no NEW dispatch picks up this
+    // callback, then close the stream under its own mutex so any dispatch
+    // already running has either finished or will see a closed stream. Only
+    // after both is it safe to return and let gRPC free `writer`.
     events_.unsubscribe(subId);
-
+    {
+        std::lock_guard<std::mutex> streamLock(state->mu);
+        state->open = false;
+    }
+
+    // A client that sees OK would reasonably conclude its subscription ended
+    // normally. It did not - the server is going away - and UNAVAILABLE is the
+    // status a gRPC client already treats as "retry against a new connection".
+    if (stopping) {
+        // Logged at INFO deliberately. It is bounded (at most
+        // maxSubscribeStreams_ lines, 50 by default, once per process
+        // lifetime), it tells an operator why consumers saw UNAVAILABLE, and it
+        // is the only externally visible evidence that the stream unwound on
+        // the shutdown SIGNAL rather than by waiting out the Shutdown()
+        // deadline - which is what tests/load_test/test_shutdown_with_subscriber.sh
+        // measures.
+        spdlog::info("Subscribe: closing stream, server is shutting down");
+        return grpc::Status(grpc::StatusCode::UNAVAILABLE, "server is shutting down");
+    }
     return grpc::Status::OK;
 }
 
+void DatabaseGrpcImpl::beginShutdown() {
+    spdlog::info("Signalling shutdown to {} active Subscribe stream(s)",
+                 activeSubscribeStreams_.load());
+    {
+        std::lock_guard<std::mutex> lock(shutdownMutex_);
+        shuttingDown_.store(true, std::memory_order_release);
+    }
+    shutdownCv_.notify_all();
+}
+
 // ===== Health and Stats =====
 
 grpc::Status DatabaseGrpcImpl::HealthCheck(

service/src/database_grpc_impl.hpp

@@ -17,8 +17,10 @@
 #include <grpcpp/grpcpp.h>
 #include <atomic>
 #include <chrono>
+#include <condition_variable>
 #include <cstdint>
 #include <memory>
+#include <mutex>
 #include <string>
 
 namespace smartbotic::database {
@@ -56,6 +58,11 @@ public:
         principals_.setKeys(std::move(keys));
     }
 
+    // v2.11.2 — tell every parked streaming handler to return now. Called by
+    // DatabaseService::stopGrpcServer() BEFORE grpc::Server::Shutdown(), so the
+    // streams are gone before Shutdown has to wait for them. Idempotent.
+    void beginShutdown();
+
     // ===== Document Operations =====
 
     grpc::Status Insert(
@@ -595,6 +602,19 @@ private:
 
     // v1.6.2 — per-RPC-type streaming concurrency limits (set by DatabaseService
     // from GrpcConfig). Excess streams get RESOURCE_EXHAUSTED.
+    // v2.11.2 — shutdown signal for long-lived streaming handlers.
+    //
+    // The Subscribe handler has no work of its own: it parks until the stream
+    // ends. Parking on a 100 ms poll of context->IsCancelled() meant it only
+    // noticed a server stop when the shutdown DEADLINE expired and gRPC
+    // cancelled it - so every stop paid the full grace period. Waiting on this
+    // condition variable instead lets it unwind the moment the service says it
+    // is stopping, and still wakes every 100 ms to notice a client that went
+    // away (the sync API gives no callback for that).
+    std::atomic<bool> shuttingDown_{false};
+    std::mutex shutdownMutex_;
+    std::condition_variable shutdownCv_;
+
     std::atomic<uint32_t> activeSubscribeStreams_{0};
     std::atomic<uint32_t> activeFileStreams_{0};
     uint32_t maxSubscribeStreams_ = 50;

service/src/database_service.cpp

@@ -1920,6 +1920,17 @@ void DatabaseService::stopGrpcServer() {
     // (they are milliseconds), short enough that it is invisible next to the
     // final snapshot, which is the part of shutdown that legitimately takes
     // time (~70 s on the largest known dataset).
+    // v2.11.2 — tell the parked streaming handlers to return BEFORE asking gRPC
+    // to shut down. Subscribe has no work of its own; it waits for the stream to
+    // end. Without this signal it learned of the stop only when the deadline
+    // below expired and gRPC cancelled it, so every stop paid the full grace
+    // period. With it, the streams are usually gone before Shutdown is even
+    // called and the deadline is what it should be: a backstop, not the
+    // mechanism.
+    if (storageImpl_) {
+        storageImpl_->beginShutdown();
+    }
+
     constexpr auto kShutdownGrace = std::chrono::seconds(5);
     const auto deadline = std::chrono::system_clock::now() + kShutdownGrace;
     for (auto& server : grpcServers_) {

tests/load_test/test_shutdown_with_subscriber.sh

@@ -31,10 +31,23 @@ DRIVER="$WORK/subscriber"
 SERVER_PID=""
 SUB_PID=""
 
-# How long we allow the whole stop to take. The grace period is 5 s and the
-# dataset here is tiny, so anything near this means the shutdown is stuck.
+# How long we allow the whole stop to take. The 5 s deadline in
+# stopGrpcServer() is the backstop; since v2.11.2 the Subscribe handlers are
+# signalled directly and return before Shutdown() is even called, so a healthy
+# stop is dominated by the final snapshot and finishes well inside this.
 STOP_BUDGET_SEC=30
 
+# v2.11.2 — the stream must unwind on the SIGNAL, not by waiting out the
+# Shutdown() deadline. Wall-clock for the whole process cannot show this: a stop
+# also spends ~4 s in gRPC draining a still-connected client's transport (proven
+# by isolation - with the client killed first, Shutdown() returns in 1 ms) plus a
+# few seconds joining component threads, and neither is affected by this change.
+#
+# So measure the thing itself: the gap between the service announcing it is
+# stopping and the Subscribe handler reporting that it closed. That is the
+# handler's own latency and nothing else's.
+MAX_UNWIND_LATENCY_SEC=1
+
 cleanup() {
     [[ -n "$SUB_PID" ]] && kill -9 "$SUB_PID" 2>/dev/null
     [[ -n "$SERVER_PID" ]] && kill -9 "$SERVER_PID" 2>/dev/null
@@ -129,8 +142,31 @@ if [[ $STOPPED -ne 1 ]]; then
     fail "server still running ${STOP_BUDGET_SEC}s after SIGTERM - Shutdown() is blocking on the open Subscribe stream"
 fi
 echo "  exited ${ELAPSED}s after SIGTERM (budget ${STOP_BUDGET_SEC}s)"
+WITH_SUB_SEC=$ELAPSED
 SERVER_PID=""
 
+# The handler must SAY it closed because of the shutdown. Absent, it exited some
+# other way and the rest of this measurement would be meaningless.
+grep -q "Subscribe: closing stream, server is shutting down" "$WORK/boot1.log" \
+    || { grep -iE "subscribe|shutting" "$WORK/boot1.log" | tail -5
+         fail "the Subscribe handler never reported a shutdown close - it did not unwind on the signal"; }
+
+# Both lines carry [YYYY-MM-DD HH:MM:SS.mmm]; compare them in milliseconds.
+ts_ms() {
+    grep -m1 -F "$2" "$1" | sed -n 's/^\[[0-9-]* \([0-9:]*\)\.\([0-9]*\)\].*/\1 \2/p' \
+        | awk '{split($1,t,":"); print ((t[1]*3600)+(t[2]*60)+t[3])*1000 + $2}'
+}
+T_STOP=$(ts_ms "$WORK/boot1.log" "Stopping database service")
+T_CLOSE=$(ts_ms "$WORK/boot1.log" "Subscribe: closing stream")
+if [[ -z "$T_STOP" || -z "$T_CLOSE" ]]; then
+    fail "could not read the shutdown timestamps out of the log"
+fi
+UNWIND_MS=$(( T_CLOSE - T_STOP ))
+echo "  the Subscribe stream unwound ${UNWIND_MS}ms after the stop began"
+if (( UNWIND_MS > MAX_UNWIND_LATENCY_SEC * 1000 )); then
+    fail "the stream took ${UNWIND_MS}ms to unwind (limit $(( MAX_UNWIND_LATENCY_SEC * 1000 ))ms) - it is waiting out the Shutdown() deadline instead of observing beginShutdown()"
+fi
+
 kill -9 "$SUB_PID" 2>/dev/null; SUB_PID=""
 
 # The point of unblocking shutdown: the code AFTER stopGrpcServer() gets to run.
@@ -150,7 +186,20 @@ if grep -qi "read-only mode" "$WORK/boot2.log"; then
     fail "the service came up read-only after a graceful stop"
 fi
 echo "  recovery: TrivialSuccess, not read-only"
-kill "$SERVER_PID" 2>/dev/null; SERVER_PID=""
+
+echo
+echo "=== for reference: the same stop with NO subscriber attached ==="
+START=$(date +%s)
+kill -TERM "$SERVER_PID"
+for _ in $(seq 1 "$STOP_BUDGET_SEC"); do
+    kill -0 "$SERVER_PID" 2>/dev/null || break
+    sleep 1
+done
+BASELINE_SEC=$(( $(date +%s) - START ))
+kill -9 "$SERVER_PID" 2>/dev/null; SERVER_PID=""
+# Reported, not asserted. The difference is gRPC draining a live client
+# connection, which this change does not touch and which the deadline bounds.
+echo "  no subscriber: ${BASELINE_SEC}s | with subscriber: ${WITH_SUB_SEC}s"
 
 echo
 echo "ALL SHUTDOWN CHECKS PASSED"