4.x: stream-id and write-coalescer cleanup gaps left by #965

[nikagra]

Remaining problems after #965

#965 is merged (ee0d033f2). #1007 picks up the throttler-callback half of the fallout for
CqlRequestHandler. These are the remaining findings from a re-review of d29b529dcb, none of
which either PR touches. Verified against the blobs on scylla-4.x.

Ordered most severe first.

1. InFlightHandler.write() leaks a stream-id bit on the duplicate-callback guard

InFlightHandler.java:144-149

if (inFlight.containsKey(streamId)) {
  promise.setFailure(new IllegalStateException("Found pending callback for stream id " + streamId));
  streamIds.cancelPreAcquire();
  return;
}

streamIds.acquire() at L133 already did ids.set(streamId). cancelPreAcquire() only does
availableIds.incrementAndGet() — it never clears the bit. So the counter and the BitSet drift
apart: the channel keeps advertising one more free id than it really has, and the concrete id is
borrowed forever. After maxAvailableIds of these, acquire() returns -1 while
getAvailableIds() still reports capacity, and every subsequent write on the channel fails with
BusyConnectionException.

Should be release(streamId, ctx). The two sibling branches (L130 and L140) are correct, because
neither runs after a successful acquire().

Pre-existing, but it is the mirror image of the accounting bug #965 set out to fix, and it lives
in the function #965 rewrote.

2. CqlPrepareHandler.prepareOnOtherNode() strands the caller's prepareAsync future

CqlPrepareHandler.java:339-342

} catch (Throwable t) {
  channel.cancelPreAcquireId();
  throw t;
}

This contradicts the method's own contract, stated two lines above it: "Errors are not blocking,
the preparation will be retried later on that node. Simply warn and move on."

If toPrepareMessage(request) or Conversions.resolveRequestTimeout(...) throws for one node, the
rethrow escapes the for (Node node : queryPlan) loop in prepareOnOtherNodes() (L312-315), so
every remaining node is skipped, CompletableFutures.allDone(...) is never returned, and the
.thenRun(...) / .exceptionally(...) in onResponse never run. The user's prepareAsync future
is left hanging and the exception only surfaces as a netty-logged failure inside
InitialPrepareCallback.onResponse.

Cancel the reservation, log, and return CompletableFuture.completedFuture(null).

3. DefaultWriteCoalescer fails deliverable writes during the graceful quiet period

DefaultWriteCoalescer.java:183

if (eventLoop.isShuttingDown()) {
  failPendingWrites(rejectedExecutionFailure(new RejectedExecutionException("Event loop is shutting down")));
}

SingleThreadEventExecutor.isShuttingDown() is true from ST_SHUTTING_DOWN onward, but
offerTask() only rejects from ST_SHUTDOWN. For the whole quiet period the reschedule would
have succeeded, yet every queued Write is failed with ClosedConnectionException.

It also catches a queued GRACEFUL_CLOSE_MESSAGE / FORCEFUL_CLOSE_MESSAGE, which
DriverChannel.close() routes through the coalescer specifically so they aren't rejected — see
the comment at DriverChannel.java:320-322. Write.fail() only fails the promise (observable
solely through UncaughtExceptions::log), so InFlightHandler.startGracefulShutdown() never runs.

eventLoop.isShutdown() is the gate that matches offerTask().

4. ThrottledAdminRequestHandler.start() bypasses the throttler signal unconditionally

ThrottledAdminRequestHandler.java:141-147. Same shape as the bug #1007 fixes with its new
wasAdmitted parameter — raised there as review feedback, recorded here so it isn't lost:
holdsExternalReservation already distinguishes admitted from not-admitted, so capture it before
cancelExternalReservation() and branch on it, rather than always calling super.setFinalError(t).

5. DefaultWriteCoalescer's reschedule listener can release a flag it doesn't own

DefaultWriteCoalescer.java:199-205. Unlike the isCancelled() branch, the !isSuccess() branch
can fire after L166 released running and L180 re-took it for the next run. failPendingWrites
then (a) fails writes that run was about to deliver on a healthy event loop, and (b) does an
unconditional running.set(false) at L139, releasing the mutual-exclusion token the next run owns.
It is also the only branch of the four with no test.

Nit, same file: the t instanceof RejectedExecutionException ? rejectedExecutionFailure(t) : t
ternary is spelled out at L202-204 and L209, plus two direct wraps at L119 and L184-186. One
private static Throwable toWriteFailure(Throwable) would keep the wrapping policy in one place.

6. RejectionSafeEventExecutor forwards lifecycle calls to the shared event loop

RejectionSafeEventExecutor.java:59-72. This instance is handed out as a promise executor via
WriteCoalescer.listenerNotificationExecutor(channel) and DriverChannel.newFailedWriteFuture().
Anything that reaches future.executor() and calls shutdown() / shutdownGracefully() — or
shutdownNow(), which AbstractEventExecutor routes to shutdown() — tears down the shared Netty
event loop and every DriverChannel bound to it, not just this promise's notifier. Conversely
schedule() / scheduleAtFixedRate() fall through to AbstractEventExecutor's
UnsupportedOperationException, so the object claims to be a ScheduledExecutorService it isn't.
Both groups should throw UnsupportedOperationException explicitly.

7. Three implementations of "the rejection-safe executor for this event loop"

PassThroughWriteCoalescer.java:35/44 keeps its own
ConcurrentMap<EventLoop, RejectionSafeEventExecutor>; DefaultWriteCoalescer.Flusher gets the
same object off its per-event-loop field (L97); and WriteCoalescer's default method (L38-39)
returns new RejectionSafeEventExecutor(channel.eventLoop()) with no caching at all, so two
DriverChannels on the same loop get different executors.

A single cached helper — e.g. RejectionSafeEventExecutor.forChannel(channel) backed by one
static map, with the interface default delegating to it — collapses all three and makes item 6 a
one-line fix instead of three.