Every Disconnect Eats a Message

Cancellation bugs are the ones that survive code review, because the code reads correctly at human speed. You check the token, you break out of the loop, you clean up. What the reviewer's eye does not do is ask when the token was checked relative to the side effect — and in an await foreach over a shared queue, one statement of ordering is the difference between a clean shutdown and destroying a message on every disconnect.

Part 9 showed that every connected browser competes for items from a single Channel<StorySent>. This part is about what happens to the browsers that are no longer connected, and the answer is that they keep competing.

The loop, again, with the ordering visible

public override async Task StreamStories(SubscribeStories request, IServerStreamWriter<Story> responseStream,
    ServerCallContext context)
{
    await foreach (var storyCreated in _storySentChannels.Reader.ReadAllAsync())
    {
        if (context.CancellationToken.IsCancellationRequested)
        {
            break;
        }

        _logger.LogInformation($"Channel has received story created {storyCreated.StoryId}");
        await responseStream.WriteAsync(new Story { ... });
    }

    _logger.LogInformation("Closing the stories stream.");
}

Trill.Pusher/src/Trill.Pusher/Services/NotifierService.cs:22-48. The second streaming method, StreamRejectedActions, has the identical shape at lines 50–70.

ReadAllAsync() is called with no argument. It has a CancellationToken overload — ReadAllAsync(context.CancellationToken) — and it is not used. Without it, the async enumerator's MoveNextAsync parks indefinitely inside WaitToReadAsync, waiting for the next item, with no knowledge that the client has gone.

The cancellation check runs after an item has been dequeued. By the time IsCancellationRequested is evaluated, storyCreated has already left the channel. ChannelReader<T> has no un-read, no peek-and-commit, no nack. The item exists only in the local variable of a loop that is about to break.

Put those two facts together with the competing-consumer behaviour from part 9 and the failure is exact. A client closes its tab. Its ServerCallContext.CancellationToken fires. Its loop is parked inside ReadAllAsync, which does not observe the token, so it stays parked — still an active consumer of the shared channel. The next story_sent arrives, the dead loop wins the race for it, dequeues it, notices cancellation, and breaks. The message is gone. It was not written to any stream and it was not returned to the queue.

Every disconnect eats exactly one event, and the two bugs compound: dead streams keep stealing from live ones for as long as it takes each of them to win one message.

Why it never showed up

Three reasons, and they are all instructive.

With one client, the failure is invisible — there is nothing to steal from. With zero clients, which is the normal state of a demo, there is no reader at all and the symptom becomes the next section's problem instead. And the final log line, "Closing the stories stream.", does fire on this path, so the logs look like a clean shutdown; they just fire one message later than the message they consumed.

There is also nothing to catch it with. The Pusher has no test project at all — it is one of two repositories in the estate without a tests/ directory — and, as part 5 noted, no committed artefact anywhere in the estate invokes a gRPC method. Trill.Pusher.rest is a single GET against https://localhost:5010, which is both the wrong scheme (the port binds plain HTTP) and the wrong protocol (it can only ever hit the app-name MapGet("/")). The streaming API has never been exercised by anything committed to this repository.

The fix is one argument and one reordering:

await foreach (var storyCreated in _storySentChannels.Reader.ReadAllAsync(context.CancellationToken))
{
    await responseStream.WriteAsync(new Story { ... });
}

Pass the token to ReadAllAsync and the enumerator throws OperationCanceledException while waiting, before any item is taken. The explicit IsCancellationRequested check then becomes unnecessary — which is the tell that it was standing in for the token it should have been given.

Unbounded, with no consumer at all

The other half of this is what happens when nobody is watching, and it follows directly from Channel.CreateUnbounded<StorySent>().

The writer is driven by the RabbitMQ subscription. The reader is driven only by a connected gRPC client. With zero clients connected — which is the normal state of a demo system, and the state it is in every night — every StorySent event ever published accumulates in process memory forever. There is no BoundedChannelOptions, no FullMode, no drop policy, no TryWrite-and-discard, and no metric counting depth. The queue is not drained by time; it is drained only by someone opening a browser, and even then only at whatever rate that one client's stream can absorb.

Two things make it worse than the usual unbounded-channel story. The messages retained are StorySent events carrying the full story title, author and tag list — this estate has no Claim Check, so the payload is the content, not a reference to it. And restarting the process is the only reclamation available, which silently discards everything queued: the same event that was retained for a week because nobody was watching is destroyed the moment someone deploys.

There is no backpressure anywhere in this path either. The AMQP layer has no prefetch configured — grep prefetch across all eleven repositories returns nothing — and the channel is unbounded, so a publisher that outruns delivery is regulated by exactly one thing, which is the size of the machine.

The client that never comes back

The Pusher's counterpart in Trill.Web completes the picture, and it is worth reading because the estate does have a real end-to-end path here: bus, to channel, to gRPC stream, to Blazor component, to a UI notification. Pages/Index.razor subscribes to StoryCreated; Shared/MainLayout.razor subscribes to ActionRejected and surfaces it as a toast. That path works, and building it at all is more than most samples manage.

Here is how it starts:

public Task InitAsync()
{
    Task.Run(SubscribeStoriesAsync);
    Task.Run(SubscribeRejectedActionsAsync);

    return Task.CompletedTask;
}

Trill.Web/src/Trill.Web.UI/Services/PusherService.cs:27-33.

Two un-awaited fire-and-forget tasks. Any exception in either — a server restart, a network blip, the stream simply ending — faults a Task that nobody holds a reference to and nobody observes. The stream dies silently. The UI shows a permanently frozen feed with no error, no spinner, no connection-state indicator, and no reason for the user to suspect anything other than that nobody has posted lately.

There is no reconnect and no backoff. The subscribe loops are not cancellable either:

var stream = _client.StreamStories(new SubscribeStories());
while (await stream.ResponseStream.MoveNext(CancellationToken.None))

PusherService.cs:43-44, and the same at line 65. CancellationToken.None on both, which means even a deliberate shutdown cannot stop them.

And the shutdown path is unreachable anyway:

var host = builder.Build();
...
await Task.Run(() => pusherService.InitAsync());
await host.RunAsync();
await pusherService.CloseAsync();

Trill.Web/src/Trill.Web.UI/Program.cs:38-44. In Blazor WebAssembly, RunAsync never returns. CloseAsync — which logs “Closing gRPC channel…” and calls _channel.ShutdownAsync() — is dead code, and the log line it writes has never been printed.

One line above, at Program.cs:17, is AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true), copy-pasted from a server-side gRPC sample. SocketsHttpHandler does not exist in the browser; in WASM the handler is the fetch API. The switch is inert.

The three rules this leaves

Give the token to the thing that waits, not to the thing that acts. A cancellation check inside the loop body can only cancel work you have already committed to. If the wait is where the time goes, the wait is where the token belongs — ReadAllAsync(ct), WaitToReadAsync(ct), ReceiveAsync(ct).

An unbounded queue is a decision to be regulated by memory. That is occasionally the right decision and it is never the right default. CreateBounded with an explicit FullMode forces the interesting question — block the producer, or shed the message? — that unbounded lets you defer until an incident asks it for you.

A long-lived stream needs a connection state, and the client owns it. Not a Task.Run and a hope: an observed task, a retry with backoff, and something on screen that distinguishes “quiet” from “disconnected”. Trill's client has a working push pipeline and no way for a user to tell whether it is alive.

Next, the four-line handler sitting between the bus and that channel, and the feature it half-implements: the filter that drops the future.