Deadlock in MessageHandler.put_messages causes unbounded replication slot growth

[g14a]

Summary:

When SlotMessageStore buffer is full, MessageHandler.put_messages/3 blocks SlotProcessorServer with a Process.sleep retry loop (15 retries, exponential backoff, ~1.6s). This creates a deadlock because ConsumerProducer needs SlotProcessorServer to be responsive to drain the buffer, but it's sleeping. The buffer never drains, retries always fail, ReorderBuffer reschedules the same batch, and the cycle repeats forever. No LSN acks reach Postgres, so the replication slot grows unboundedly (we've seen 50-73GB).

Reproduction:

High-throughput WAL ingestion that fills SlotMessageStore beyond max_memory_bytes. Once payload_size_limit_exceeded fires, the system never recovers without a manual restart.

Workaround:

Clicking "Restart" on the database page in the UI breaks the deadlock by restarting the pipeline with a fresh empty buffer but that again gets filled up quite quickly.

I personally come from Go background and I haven't understood Elixir code really well since this project is a little complex for me. So I used AI to help me navigate the code and debug this issue. But the issue we are facing is real. We have production systems with more than 100GB lag since the past 3 months.

[dosubot]

Your analysis is spot-on. The deadlock is confirmed in the codebase:

Root cause: MessageHandler.put_messages blocks SlotProcessorServer with Process.sleep() for up to ~1.6s during retries when the buffer is full. Meanwhile, ReorderBuffer retries failed batches every 250ms indefinitely without any circuit breaker—creating the infinite loop you described.

This is a known issue documented in several related issues:

• #881 - OOM crashes from this pattern
• #872 - GenServer timeout with 222k accumulated messages
• #1004 - CDC hung and failed to recover

Potential workarounds:

1. Increase max_memory_mb on your consumer to give more buffer headroom (note: a 0.8 safety factor is applied, so actual limit is 80% of configured value)
2. Use load_shedding_policy: discard_on_full (#1358) - this drops messages for overloaded consumers instead of blocking, preventing slot growth but causing data loss
3. Reduce throughput on the source if possible to stay under buffer limits

Unfortunately, this appears to be an architectural issue requiring changes to the synchronous blocking behavior in put_messages or adding proper circuit breaking to the ReorderBuffer retry loop.

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[g14a]

Pasting a deadlock call trace for reference if it helps:

Deadlock Call Trace

① Postgres WAL change arrives
↓
② ReorderBuffer.handle_info(:flush_batch, state)
reorder_buffer.ex:115
├── Pops the next ready batch by idx
└── Calls state.flush_batch_fn.(state.id, batch) — line 124
↓
③ SlotProcessorServer.handle_batch(id, batch)
slot_processor_server.ex:144 (GenServer.call)
→ handle_call({:handle_batch, batch}, ...) — line 226
└── flush_messages(state, batch) — line 227
↓
④ SlotProcessorServer.flush_messages(state, batch)
slot_processor_server.ex:559
├── Calls state.message_handler_module.handle_messages(ctx, messages) — line 570
│ This is a SYNCHRONOUS call — SlotProcessorServer is BLOCKED here
│ ↓
⑤ MessageHandler.handle_messages(ctx, messages)
message_handler.ex — groups messages by consumer, then calls:
└── put_messages(consumer, messages) — line 190
↓
⑥ SlotMessageStore.put_messages(consumer, messages)
slot_message_store.ex:588 (GenServer.call)
└── State.validate_put_messages(state, messages) — line 590
↓
⑦ SlotMessageStoreState.validate_put_messages(state, messages)
slot_message_store_state.ex:79
└── payload_size_bytes + incoming > max_memory_bytes?
├── YES → {:error, :payload_size_limit_exceeded} — line 85
└── NO → {:ok, size} — line 91

When buffer is full (the deadlock path in UPSTREAM code):

⑦ returns {:error, :payload_size_limit_exceeded}
↓
⑥ SlotMessageStore replies with error
↓
⑤ MessageHandler.put_messages — OLD CODE had Process.sleep retry loop here
├── Retry 1: sleep(X ms), call SlotMessageStore again → still full
├── Retry 2: sleep(2X ms), call again → still full
├── ... (15 retries, exponential backoff, ~1.6s total)
└── All retries fail → returns {:error, ...}
↓
⚠️ DURING THIS ENTIRE 1.6s, SlotProcessorServer (step ④) is BLOCKED
⚠️ It cannot respond to ANY other GenServer.call
↓
④ SlotProcessorServer returns {:error, :payload_size_limit_exceeded}
↓
③ handle_call replies {:reply, {:error, error}, state} — line 231-234
↓
② ReorderBuffer.handle_info gets {:error, reason} — line 140
└── schedule_flush_timer_retry(state) — line 142
Schedules :flush_batch after setting_retry_flush_batch_interval — line 297-300
↓
... interval passes ...
↓
② :flush_batch fires again → SAME BATCH → SAME DEADLOCK → ♻️ FOREVER

Why it never recovers:

ConsumerProducer (the ONLY thing that drains SlotMessageStore)
needs to call SlotMessageStore.produce()
BUT it is triggered by :messages_maybe_available signals
AND those signals aren't published when the buffer is full
→ Buffer stays full → SlotProcessorServer keeps sleeping → No drain → ♻️

Meanwhile:
No LSN acks sent to Postgres (SlotProcessorServer is stuck)
→ Replication slot retains ALL WAL
→ Slot grows unboundedly (50-73GB observed)

Suggested fix:

In MessageHandler.put_messages/3, remove the Process.sleep retry loop. Instead:

1. Return {:error, payload_size_limit_exceeded} immediately so SlotProcessorServer is unblocked
2. Publish {:messages_maybe_available, consumer_id} via :syn to wake ConsumerProducer to drain
3. Let ReorderBuffer retry the batch after its configured interval (~250ms), by which time the buffer has drained