BrighterCommand / BrighterCommand/Brighter

MQTT: Proactor deadlocks on first requeue — MqttMessagePublisher connects in its constructor

Open
#4,351 0 comments 0 reactions 1 assignee Claimed by @iancooper View on GitHub
Dominant language
C#
Stars
2.5k
Forks
296
Avg merge
1d 11h
Merged PRs (30d)
21

Description

# MQTT: Proactor deadlocks on first requeue — MqttMessagePublisher connects in its constructor

## Summary

`MqttMessagePublisher`'s constructor blocks on its own async connect —
`src/Paramore.Brighter.MessagingGateway.MQTT/MQTTMessagePublisher.cs:54`:

```csharp
ConnectAsync().GetAwaiter().GetResult();
```

`MqttMessageConsumer` creates its requeue producer **lazily, inside `RequeueAsync`**
(`EnsureRequeueProducer()`). A `Proactor` runs its event loop inside `BrighterAsyncContext.Run(...)`
(`Proactor.cs:99`), which is a **single-threaded** synchronization context. So the first time a
handler defers a message on MQTT, the publisher is constructed on the pump's only thread, that
thread blocks on `ConnectAsync()`, and MQTTnet's continuation is posted back to the context that
thread is blocking — **deadlock**.

This is not "deadlock risk". It is measured below.

## Measured

Two facts, same machine, same broker (`docker-compose-mqtt.yaml`, `efrecon/mosquitto`):

```csharp
// completes
var a = Task.Run(() => new MqttMessagePublisher(config));
Assert.True(a.Wait(TimeSpan.FromSeconds(15)));

// DEADLOCKS — times out at 15 s
var b = Task.Run(() => BrighterAsyncContext.Run(async () =>
{
var p = new MqttMessagePublisher(config);
await Task.Yield();
return p;
}));
Assert.True(b.Wait(TimeSpan.FromSeconds(15)));
```

> `Failed: 1, Passed: 1` — the plain thread-pool construction passes; the one under
> `BrighterAsyncContext` fails.

## How it shows up end to end

Running the conformance suite's FR-23 (`requeue budget exhausted to DLQ`) against MQTT — a real pump
over the channel, and a handler that always defers:

| variant | result |
|---|---|
| `Reactor` | **passes in 5 s** — the message reaches the DLQ |
| `Proactor` | **hangs indefinitely** — killed at 18 minutes, and again by `--blame-hang` at 180 s |

Instrumenting the Proactor run shows where it stops:

```
16:22:02.455 pump started
16:22:02.477 handler invocation 1 <- defers
16:23:02.495 poll returned MT_NONE <- nothing ever reaches the DLQ
16:23:02.997 enqueue quit
16:23:02.997 awaiting pump <- never returns
```

The handler is invoked **exactly once**. The deferral calls `RequeueAsync`, which constructs the
requeue producer on the pump thread, and nothing happens after that — no redelivery, no second
invocation, no dead-lettering, and the pump does not observe the quit message because its only
thread is blocked.

The `Reactor` variant survives the identical code path because `MqttMessageConsumer.Requeue` runs on
an ordinary thread-pool thread with no captured context, so `ConnectAsync`'s continuation has
somewhere to run.

## Impact

Any Brighter **`Proactor`** consumer on MQTT deadlocks the first time a handler defers a message.
That is a production path, not a test-harness one: nothing about the reproduction is specific to the
conformance suite beyond its being the first thing to drive a real pump over an MQTT channel.

It also means MQTT's delivery budget can never be exhausted on the async path, so a deferred message
is neither redelivered nor dead-lettered — it is lost along with the pump.

## Previously identified, and closed unfixed

[#4082](https://github.com/BrighterCommand/Brighter/issues/4082) named **this exact line** as
candidate 4 ("`MQTTMessagePublisher.cs:54` … deadlock-prone if constructed under a captured
context"), with a checklist item "MQTT publisher offers async construction". It was closed as
**COMPLETED** on 2026-04-27, but `MQTTMessagePublisher.cs` has not been touched since `b42887af4`,
which predates it. The item was missed.

## Suggested fix

#4082's own prescription still applies, and there are two independent halves:

1. **Do not connect in the constructor.** Either add
`public static async Task CreateAsync(...)`, or connect lazily on first
publish so the blocking call never sits in a constructor. At minimum route the blocking wait
through `BrighterAsyncContext.Run(...)` the way the rest of the codebase does, rather than a raw
`GetAwaiter().GetResult()`.
2. **Do not create the requeue producer on the pump thread.** `EnsureRequeueProducer()` fires inside
`RequeueAsync`; creating it eagerly when the consumer is built removes the pump thread from the
picture entirely, and would make (1) sufficient rather than necessary.

Either one alone unblocks the `Proactor` path; both are worth doing.

## Conformance ledger

`MQTT / MqttMessagingGateway` FR-23 stays `Deferred` against this issue in
`specs/0036-universal-transport-conformance-tests/conformance-status.md`. The `Reactor` half is
conformant; FR-14's both-variants rule is what holds the cell.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.