OOM risk: put_persisted_messages bypasses max_memory_mb limit on startup

[juanherrera-jpy]

Summary

When a sink accumulates a large number of persisted events in the database (e.g., due to a delivery error), all events are loaded into memory unconditionally on startup/rehydration via put_persisted_messages, bypassing max_memory_mb and setting_max_messages limits. This can cause OOM kills on the pod.

Reproduction scenario

We experienced this in production with an Elasticsearch OpenSearch sink (jobs_sink) that had a strict_dynamic_mapping_exception delivery error. While the sink was erroring:

1. 140,416 events accumulated in sequin_streams.consumer_events (partition 68)
2. Each enriched Job message was ~18 KB in memory (enrichment joins ~12 related tables)
3. When the delivery error was fixed, the SlotMessageStore loaded all persisted messages at once
4. Memory jumped to 1,509 MB despite max_memory_mb being set to 512 MB
5. On an earlier run with max_memory_mb=128, the store oscillated between filling (hitting the limit for new WAL messages) and draining, but persisted messages were always loaded without limit

Root cause

In lib/sequin/runtime/slot_message_store_state.ex:

# Line 132-143
def put_persisted_messages(%State{} = state, messages) do
    persisted_message_groups =
      Enum.reduce(messages, state.persisted_message_groups, fn msg, acc ->
        Multiset.put(acc, group_id(msg), {msg.commit_lsn, msg.commit_idx})
      end)

    # This cannot fail because we `skip_limit_check?`
    {:ok, state} =
      put_messages(%{state | persisted_message_groups: persisted_message_groups}, messages, skip_limit_check?: true)

    state
end

Both put_persisted_messages (line 141) and put_table_reader_batch (line 160) call put_messages with skip_limit_check?: true, meaning:

• max_memory_mb is not enforced
• setting_max_messages (default 50,000) is not enforced
• There is no upper bound on memory consumption during rehydration

Impact

• OOM risk: A pod with 32 GB RAM could be killed if enough events accumulate. In our case, 140k events at ~18 KB each = ~2.5 GB. A table with larger payloads or more accumulated events could easily exceed pod memory limits.
• No recovery path: Once events are persisted, every restart will attempt to load them all, potentially causing repeated OOM crashes.
• Silent accumulation: Events pile up silently when a sink has delivery errors. There's no alert or limit on how many events can accumulate in the database.

Environment

• Sequin self-hosted on Kubernetes (EKS)
• Pod resources: 8 CPU / 32 GB RAM
• 25 Elasticsearch OpenSearch sinks, all using enrichment queries
• Source database: Aurora PostgreSQL 12
• Sink consumer config: batch_size=1, max_memory_mb=128 (later increased to 512), message_grouping=true

Suggested improvements

1. Batch-load persisted messages: Instead of loading all persisted messages at once, load them in configurable batches that respect max_memory_mb, processing each batch before loading the next.

2. Enforce a hard memory ceiling: Even for persisted messages, don't exceed a configurable absolute maximum (e.g., max_memory_mb * 2 or a separate hard_max_memory_mb setting).

3. Add a persisted event count limit or alert: When events accumulate beyond a threshold in the database (e.g., due to sustained delivery errors), surface a health check warning or automatically pause the sink before the backlog becomes dangerous.

4. Consider on-demand loading: Instead of eagerly loading all persisted messages into memory, load them on-demand as the sink has capacity to deliver, similar to how the ConsumerProducer pulls from the store.

Workarounds

Currently the only options to prevent OOM are:

• Fix delivery errors quickly before events accumulate
• Manually delete stuck events from sequin_streams.consumer_events
• Ensure pod memory limits are very generous relative to potential event backlogs

[mm-zacharydavison]

@dosu review this issue.

[dosubot]

Hi @mm-zacharydavison! I'm Dosu and I’m helping the sequin team.

Great issue report. I've verified the code behavior and this is a legitimate OOM vulnerability.

Confirmed findings:

• skip_limit_check?: true is intentional to prevent blocking on already-persisted messages
• However, the OOM problem is actually worse than described: stream_messages_into_state uses Enum.map to materialize ALL persisted messages at once before calling put_persisted_messages, despite the DB query using batched pagination (1000 messages)
• There's a misleading comment suggesting memory limits are respected during rehydration—they're not

Your suggested improvements are well-aligned with the existing infrastructure:

Suggestion	Feasibility	Notes
Batch-load respecting max_memory_mb	✅ High	Infrastructure exists—replace Enum.map with Enum.reduce_while, check limits per batch
Hard memory ceiling	✅ Feasible	80% allocation factor already exists but is bypassed
Persisted event count alert	✅ Complementary	Could use existing health event system
On-demand loading	⚠️ Complex	Previously rejected due to disk I/O throughput concerns

Edge cases to consider for the fix:

• Message ordering must be preserved (commit_lsn, commit_idx)
• Need a mechanism to resume loading as buffer drains
• Must coordinate with ongoing CDC ingestion to avoid starvation
• Partition distribution may require per-partition memory tracking

Related context: This issue is connected to #2129 (deadlock in MessageHandler.put_messages causing message accumulation and 50-100GB replication slot growth). Fixing rehydration OOM is important, but addressing the root cause of accumulation may also be needed for production stability.

To reply, just mention @dosu.

---

Share context across your team and agents. Try Dosu.