Make RecursiveOperationManager::sendResponse fully asynchronous (remove the last worker-pool park)

[ptesavol]

Summary

Follow-up to #81. That PR fixed the permanent worker-pool deadlock in RecursiveOperationManager::sendResponse (destroying the per-session ListeningRpcCommunicator on a pool thread parked the thread in the destructor's scope join, waiting for a send task that itself needed a pool thread) by retiring communicators into a pruned list. One bounded park remained: RecursiveOperationSessionRpcRemote::sendResponse wrapped the generated client notify in blockingWait(), so under pool saturation an RPC handler's shared worker thread was parked for up to sessionRpcTimeout (10 s). This PR removes that park: the response send never blocks any thread anywhere.

Design

• RecursiveOperationSessionRpcRemote::sendResponse is now a folly::coro::Task<void> coroutine that co_awaits the generated client notify in-frame (per the repo's lazy-task rule: the request locals live in the awaiting frame) and swallows failures — TS fire-and-forget parity, still bounded by sessionRpcTimeout.
• RecursiveOperationManager::sendResponse runs the whole send as a detached coroutine on a new manager-owned GuardedAsyncScope (sendResponseScope, on SharedExecutors::worker()). The task constructs the per-session ListeningRpcCommunicator and RpcRemote, co_awaits the notify (suspends instead of parking), then disposes of the communicator.
• Disposal keeps Fix Layer0 e2e teardown SIGSEGVs: drain RPC scopes before their targets die, roll back started on failed start() #81's deadlock rule: after the notify completes, the communicator's own client scope may still hold the send task's unfinished tail — and the coroutine may even have been resumed from inside that very task, so a non-empty-scope join there could self-deadlock. The task destroys the communicator in place only when pendingAsyncTaskCount() == 0 (an empty-scope join needs no pool thread, and a zero count also proves the coroutine is not inside one of the communicator's tasks; destroy() detaches the transport listeners first, so the zero observation is stable). Otherwise the communicator goes into the existing retiredSessionCommunicators backstop from Fix Layer0 e2e teardown SIGSEGVs: drain RPC scopes before their targets die, roll back started on failed start() #81, pruned exactly as before.
• The communicator is constructed inside the detached task, so a sendResponse racing stop() is dropped by the scope's close() gate before anything blocking-to-destroy exists: the dropped lambda releases only vectors and protobuf messages at the add() call site. A dropped response after stop is correct for a fire-and-forget send.
• stop() closes sendResponseScope (owner's thread; DhtNode::stop() calls it while the session transport is still alive) before clearing the retired list, because draining tasks may still retire communicators into it; member declaration order gives the same ordering in the destructor path.

Verification (macOS arm64, Debug)

Pool size forced with a temporary STREAMR_TEST_WORKER_THREADS override in SharedExecutors::worker() (removed before commit); DhtNodeExternalApiTest.ExternalStoreDataHappyPath with a 90 s watchdog per run:

pool size	runs	result
4 threads	15	15/15 pass (~0.2 s each)
2 threads	10	10/10 pass (~0.1 s each)
1 thread	5	5/5 pass (~0.1–0.4 s each)

Before this change the 1-thread configuration failed after ~10 s per run (the blockingWait park swallowed the only worker until the notify timed out). No hangs in any configuration.

Full suites: streamr-dht-test-unit 181/181, streamr-proto-rpc-test-unit 24/24, streamr-proto-rpc-test-integration 4/4, streamr-dht-test-end-to-end 7/7 four times (three consecutive runs plus once on the final clean rebuild), streamr-dht-test-integration 42/43 — the one failure was Layer1ScaleTest.SingleLayer1Dht, the known load-sensitive flake (#80): it exercises only joinDht, which never touches the recursive-operation path, and it passed when rerun in isolation.

Test updates: the manager unit test's transport.sendCount assertions moved after manager->stop() (which drains the send scope, making the observation deterministic now that the send is detached), and the session unit test blockingWaits the now-coroutine sendResponse on the test thread.

🤖 Generated with Claude Code