fix(worker): bound ShutdownAsync's envelope enqueue by the shutdown timeout
WorkerClient.ShutdownAsync marked the client Closing and then queued the WorkerShutdown envelope on the caller's token alone; the shutdown-timeout CTS was only created afterwards. The outbound channel is bounded, and a worker that has stopped reading the pipe parks the write loop in a pipe write it can only leave through _stopCts while the channel behind it fills with commands and their cancels. Once Closing, the heartbeat watchdog skips every tick, so nothing cancelled _stopCts and the enqueue waited for the caller's token - the lease sweep's stopping token, which fires at host stop and never before - with GatewaySession.CloseAsync holding its close lock throughout. Create the linked timeout CTS before the enqueue and run the enqueue, the loop waits and the process-exit wait on it, so the whole call is bounded by the shutdown timeout. A stalled queue now surfaces as the existing ShutdownTimeout fault, which cancels the loops, kills the worker process tree and hands GatewaySession its kill fallback. Test: WorkerClientTests.ShutdownAsync_WhenOutboundQueueIsFullBehindStalledPipeWrite_TimesOutAndKillsWorker stalls the gateway-side stream (a worker that stopped reading), fills the outbound channel with abandoned commands and their cancels, and asserts that ShutdownAsync throws ShutdownTimeout inside the test bound and kills the fake worker. It hung past the bound before the fix.
This commit is contained in:
+1
-1
@@ -267,7 +267,7 @@ The ReplayGap sentinel is emitted **only** on the `StreamEvents` server stream a
|
||||
|
||||
### Close
|
||||
|
||||
`GatewaySession.CloseAsync` is serialized by a per-session `SemaphoreSlim` (`_closeLock`) so only one close runs at a time, but every read/write of `_state` still passes through `_syncRoot` (via `TryBeginClose` and `MarkClosed`). The close path therefore obeys the same lock discipline as `TransitionTo` / `MarkFaulted`: it transitions to `Closing`, asks the worker client to shut down within `ShutdownTimeout`, and on success transitions to `Closed`. `DisposeAsync` waits on `_closeLock` once before disposing the semaphore so an in-flight close's `Release()` cannot race against the dispose. If `WorkerClient.ShutdownAsync` throws, the session falls back to `IWorkerClient.Kill` (forced close):
|
||||
`GatewaySession.CloseAsync` is serialized by a per-session `SemaphoreSlim` (`_closeLock`) so only one close runs at a time, but every read/write of `_state` still passes through `_syncRoot` (via `TryBeginClose` and `MarkClosed`). The close path therefore obeys the same lock discipline as `TransitionTo` / `MarkFaulted`: it transitions to `Closing`, asks the worker client to shut down within `ShutdownTimeout`, and on success transitions to `Closed`. `DisposeAsync` waits on `_closeLock` once before disposing the semaphore so an in-flight close's `Release()` cannot race against the dispose. `WorkerClient.ShutdownAsync` is bounded by `ShutdownTimeout` end to end — queuing the `WorkerShutdown` envelope included, since a worker that stopped reading the pipe can leave the bounded outbound channel full with no watchdog left to clear it once the client is `Closing` — so a close never holds `_closeLock` longer than the timeout before reaching the fallback. If `WorkerClient.ShutdownAsync` throws, the session falls back to `IWorkerClient.Kill` (forced close):
|
||||
|
||||
```csharp
|
||||
if (_workerClient is not null)
|
||||
|
||||
Reference in New Issue
Block a user