A Channel Is Not a Topic

Open two browser tabs against a real-time feed and post something. Both tabs should show it. If one tab shows it and the other does not — and which one is arbitrary, and it alternates as you keep posting — you are not looking at a broadcast. You are looking at a queue, and every connected client is a competing consumer of it.

That is the behaviour of Trill.Pusher, the estate's real-time service, and the mechanism is eleven lines long. Part 8 finished with the gateway; this part starts the estate's push half, which is the best teaching artefact in the whole repository precisely because the bug is so small and so legible.

The transport, first, because it is a good decision

There is no SignalR anywhere in the estate — not a package reference, not a hub, not a line of source. The transport is gRPC server-streaming, exposed twice: native HTTP/2 on port 5011, and gRPC-Web over HTTP/1.1 on 5010, because the client is Blazor WebAssembly and browsers cannot speak raw h2c.

app.UseRouting();
app.UseGrpcWeb();
app.UseCors();

app.UseEndpoints(endpoints =>
{
    endpoints.MapGrpcService<NotifierService>().EnableGrpcWeb().RequireCors("AllowAll");

Trill.Pusher/src/Trill.Pusher/Startup.cs:75-81, with the client side using new GrpcWebHandler(GrpcWebMode.GrpcWebText, ...). For 2021, with a Blazor WASM front end, that is a well-judged choice. SignalR would have been the lazy answer and would have given no typed contract; gRPC-Web gives a generated client, a .proto that both ends compile against, and a genuinely typed stream. Credit where it is due.

The contract itself is where the first eyebrow goes up:

service Notifier {
  rpc StreamStories(SubscribeStories) returns (stream Story);
  rpc StreamRejectedActions(SubscribeRejectedActions) returns (stream ActionRejected);
}

message SubscribeStories {
}

message SubscribeRejectedActions {
}

src/Trill.Pusher/Protos/pusher.proto:7-16. Both subscribe messages are empty. No user id, no topic, no filter, no token. There is no per-user fan-out and no group concept in this design — every client subscribes to the same firehose. Whatever anybody posts is pushed to every listener, and every rejection event, which carries a reason string derived from a domain exception, is broadcast to everyone regardless of whose action was rejected.

That is a deliberate simplification for a demo, and it is fine as far as it goes. It is also the reason the next defect is invisible: with no per-subscriber routing to get wrong, the fan-out primitive is the only thing left to get wrong.

Eleven lines

public class StorySentChannels
{
    private readonly Channel<StorySent> _channel;

    public StorySentChannels()
    {
        _channel = Channel.CreateUnbounded<StorySent>();
    }

    public ChannelWriter<StorySent> Writer => _channel.Writer;
    public ChannelReader<StorySent> Reader => _channel.Reader;
}

src/Trill.Pusher/Channels/StorySentChannels.cs:6-17, registered as a singleton at Startup.cs:43. Its twin, ActionRejectedChannels, is byte-for-byte the same shape for the other stream.

The producer side is a RabbitMQ event handler:

public async Task HandleAsync(StorySent @event)
{
    if (@event.Visibility.From <= DateTime.UtcNow)
    {
        await _channels.Writer.WriteAsync(@event);
    }
}

Events/External/Handlers/StorySentHandler.cs:17-23 — the filter in there is part 11's subject.

And the consumer side is the gRPC streaming method:

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
        {
            Id = storyCreated.StoryId,
            Title = storyCreated.Title,
            CreatedAt = $"{storyCreated.CreatedAt:u}",
            ...

Services/NotifierService.cs:22-44.

Now put the three together. NotifierService is instantiated per gRPC call — one instance per connected client — and each instance resolves the same singleton StorySentChannels and calls ReadAllAsync() on the same ChannelReader.

A Channel<T> is a queue, not a broadcast. ReadAllAsync from N concurrent consumers is the Competing Consumers pattern: each item is dequeued exactly once and delivered to exactly one reader. With two browser tabs open, each story appears in one tab, arbitrarily. With ten users online, each sees roughly one story in ten. The system does not error, does not log anything unusual, and does not drop messages in aggregate — every message is delivered, to precisely the wrong number of people.

The class name knows

Here is the detail that makes this the estate's single best artefact rather than just a bug. The class is called StorySentChannels. Its twin is ActionRejectedChannels. Plural.

That naming is only sensible for a design where there is a channel per subscriber — a dictionary of writers keyed on connection, a channel created when a stream opens and completed when it closes. That is the correct implementation, it is maybe thirty lines, and somebody clearly had it in mind, because they named the type for it. Then a single Channel<T> field was written and the plural never got revisited.

Publish-Subscribe Channel, implemented as a Point-to-Point Channel, with the correct pattern name sitting in the class name. The intent is legible in the identifier and absent from the code. I have never found a cleaner example of the gap between what a developer meant and what a developer typed, and it survived to the final commit because with one browser tab open the behaviour is indistinguishable from correct.

The fix, and the package that is already referenced

The in-process fix is a dictionary of per-subscriber channels. Something like this, written fresh for the article:

public class StorySentChannels
{
    private readonly ConcurrentDictionary<Guid, Channel<StorySent>> _subscribers = new();

    public (Guid Id, ChannelReader<StorySent> Reader) Subscribe()
    {
        var id = Guid.NewGuid();
        var channel = Channel.CreateBounded<StorySent>(
            new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.DropOldest });
        _subscribers[id] = channel;
        return (id, channel.Reader);
    }

    public void Unsubscribe(Guid id)
    {
        if (_subscribers.TryRemove(id, out var channel))
        {
            channel.Writer.TryComplete();
        }
    }

    public void Publish(StorySent message)
    {
        foreach (var channel in _subscribers.Values)
        {
            channel.Writer.TryWrite(message);
        }
    }
}

Three things change and each earns its place. Subscribe/unsubscribe becomes explicit, which gives the streaming method a finally block to clean up in. CreateBounded with DropOldest means one stalled browser cannot grow the process heap without limit — a real-time feed is precisely the workload where shedding the oldest item is the right answer, because nobody wants a five-minute-old story replayed when their connection recovers. And TryWrite rather than WriteAsync means a slow subscriber cannot apply backpressure to the RabbitMQ handler, which would otherwise stall delivery for everybody.

That fixes one process. It does not fix two, and here the estate has already told us what the answer was meant to be. Trill.Pusher.csproj references Convey.Persistence.Redis. appsettings.json:121-125 configures "redis": { "connectionString": "localhost", "instance": "pusher:", "database": 0 }, and the docker profile configures it again against the redis host. Not one line of the Pusher touches Redis.

Redis pub/sub is exactly the backplane this design needs at more than one replica: each instance subscribes to a Redis channel, publishes locally received bus events to it, and fans them into its own per-subscriber channels. It is the standard shape, it is what SignalR's backplane does, and the package reference plus three environment files' worth of configuration say somebody knew. It is an absence with a receipt — the most instructive kind, because it tells you the design was understood and the implementation stopped.

A timestamp that round-trips through a culture

One more thing crosses this wire, and it is worth a paragraph because it is the kind of defect that only fires on somebody else's machine.

Story.createdAt in the proto is declared as a string — line 21 of pusher.proto — rather than google.protobuf.Timestamp, which is the type protobuf provides precisely for this. The server formats it:

CreatedAt = $"{storyCreated.CreatedAt:u}",

NotifierService.cs:37. The u format specifier is the universal sortable pattern, which is culture-invariant on the way out — so far so good. The Blazor client then reads it back:

CreatedAt = DateTime.Parse(story.CreatedAt),

Trill.Web/src/Trill.Web.UI/Services/PusherService.cs:51. DateTime.Parse with no IFormatProvider and no DateTimeStyles, which means the browser's ambient culture decides how to read a UTC instant, and the resulting DateTime has Kind of Local after an implicit conversion nobody asked for. A UTC timestamp goes out correctly and comes back subject to whatever locale the user's browser reports.

Using Timestamp in the proto removes the question entirely — it is an int64 seconds-plus-nanos pair with generated conversions on both ends. Failing that, DateTime.Parse(s, CultureInfo.InvariantCulture, DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeUniversal) is the correct incantation. The estate has no DateTimeOffset anywhere and never transmits a zone, so this is the one place the timezone question surfaces, and it is answered by ambient culture.

What scaling this actually does today

Worth spelling out, because it compounds unpleasantly. Each Pusher replica gets its own queue binding on the stories exchange — the queue template is pusher/{{exchange}}.{{message}}, and Convey declares it per connection — so with two replicas the broker will round-robin story_sent messages between them as competing consumers at the AMQP layer too. Then each replica hands its share to one of its connected clients. Two replicas with six clients means each story reaches one of six, and which one depends on two independent load-balancing decisions.

The Pusher is not horizontally scalable in any sense, and it is not obvious from any single file that it is not — its state is an in-process Channel<T>, which is exactly the kind of state that does not look like state.

To be fair to the design one last time: the architecture is right. A bus subscriber writing into an in-process channel, drained by a streaming endpoint, is the correct shape for this problem, and System.Threading.Channels is the correct primitive family. The bug is the choice of one channel instead of many, and it is eleven lines from being fixed.

Next, the second bug in the same await foreach, which is that cancellation is checked one statement too late: every disconnect eats a message.