environmentd: bound HTTP and SUBSCRIBE output buffering to prevent OOM (SQL-428, SQL-423) by mtabebe · Pull Request #37905 · MaterializeInc/materialize

mtabebe · 2026-07-28T13:38:03Z

Problem:

A single query can drive environmentd to OOM, even when the client read as
fast as it could, because two paths buffered unbounded output in the shared
process.

HTTP SELECT. The peek path already streams results out of the replica in
bounded batches, but the HTTP SQL layer re-buffered every batch into a
serde_json::Value tree roughly 10-15x the wire size. A large SELECT over
/api/ws or /api/sql grew memory without limit. The /api/sql size guard
was a stale TODO and was never enforced.
SUBSCRIBE / COPY-TO. The replica -> controller -> coordinator response path
is unbounded at every hop, and the subscribe producer runs on the
non-blockable coordinator main loop, so it cannot apply backpressure to a slow
client. One slow or stopped SUBSCRIBE consumer could make the coordinator
buffer the entire subscribe output in memory.

Solution:

HTTP SQL layer (SQL-428).

WebSocket /api/ws streams rows straight through. SELECT now flows through a
new StatementResult::Rows variant whose sender emits Rows(desc) plus
per-row Row messages, flushing between stash batches so a slow client applies
backpressure and only one batch is resident. max_result_size is enforced
incrementally. Large results over WebSocket now get the same bounded memory
profile as pgwire SELECT.
Plain JSON /api/sql is a single buffered document. Rows are accumulated as
pre-serialized RawValues instead of a Value tree, and max_result_size is
enforced on the actual serialized byte count. An oversized result fails fast
with ResultSize.
MCP and prometheus consumers of SqlResult::Rows parse each RawValue on
demand.

SUBSCRIBE / COPY-TO buffer (SQL-423).
Bound the terminal buffer and retire the sink when it overflows. ActiveSubscribe
carries a shared byte counter and a per-sink budget. Each send adds the batch's
byte size and a receiver wrapper subtracts it as the client writer drains, so the
counter tracks the currently buffered depth. After each response the coordinator retires the subscribe if the depth exceeds the budget, surfacing AdapterError::SubscribeFellBehind
with a hint to use a non-buffering client or COPY (SUBSCRIBE ...) TO STDOUT.

The budget is a new dyncfg subscribe_max_buffered_bytes (default 128 MiB).
COPY (SUBSCRIBE ...) TO STDOUT shares the channel and is covered too.

Testing:

New integration tests in src/environmentd/tests/server.rs:

test_ws_select_streams_rows: a SELECT over /api/ws emits one Row message
per row before CommandComplete, confirming the non-buffering path.
test_http_sql_result_size_limit: with max_result_size lowered to 1 MB, an
oversized /api/sql result fails with a result-size error while a small one succeeds.
test_subscribe_buffer_bound: with subscribe_max_buffered_bytes set to 1 KiB,
a COPY-wrapped SUBSCRIBE whose client never reads surfaces the "fell behind"
error instead of growing without bound.

SangJunBak · 2026-08-13T20:13:06Z

+/// The backlog excludes the oldest in-flight message, so this bounds accumulated
+/// backlog, not the size of any single batch. A batch larger than this budget is
+/// still delivered, because the subscribe ships each frontier advance (and the
+/// initial snapshot) as one batch. So a client that keeps up never accumulates
+/// backlog and never approaches this cap, and a large snapshot to a keeping-up
+/// client is not retired.


Nit: This part confused me a bit on first read:

The backlog excludes the oldest in-flight message

Specifically I didn't know what "oldest in-flight" meant. Also, for:

A batch larger than this budget is /// still delivered, because the subscribe ships each frontier advance (and the /// initial snapshot) as one batch.

I think the reasoning following "because" is kind of extraneous and doesn't actually explain why the batch is delivered. The "backlog exclude the oldest in-flight message" does explain it though. Kind of feels like we can simplify this, something along the lines of:

The backlog excludes the message we have yet to send to the client. This is because max_result_size makes sure each individual message isn't too big but SUBSCRIBE_MAX_BUFFERED_BYTES controls the accumulation of messages.

SangJunBak · 2026-08-13T20:13:06Z

+
+    /// Bytes queued behind the oldest in-flight message. The oldest message is
+    /// tolerated, so only this counts against the budget.
+    pub fn backlog_behind_oldest(&self) -> usize {


Kind of feels like backlog is the word to describe messages behind the message being processed. I wonder if we should change Buffer to Backlog? So SubscribeBacklogAccounting, then pub fn backlog_size?

SangJunBak · 2026-08-13T20:13:07Z

+/// Overhead charged to every buffered message on top of its payload, so a flood
+/// of near-empty messages (frontier-only advances) still counts against the
+/// budget. Without it a stalled client could grow the buffer without bound.
+const SUBSCRIBE_MESSAGE_OVERHEAD_BYTES: usize = 1024;


Wouldn't the flood of empty messages still get accounted for in the budget? Is the worry that the channel itself / footprints would grow quite large?

Yes that is exactly it. The payload length is 0, so we need to account for some "empty" messages that fill the queue that are not being processed

SangJunBak · 2026-08-13T20:13:07Z

+    /// on a slow client, so instead of applying backpressure the coordinator
+    /// watches `backlog_behind_oldest` against `max_buffered_bytes` and retires
+    /// the subscribe once the backlog exceeds it.
+    pub buffer: Arc<Mutex<SubscribeBufferAccounting>>,


nit: buffer makes me think it's the buffer of rows and not accounting. Maybe rename to something more explicit like backlog_accounting, assuming the naming change above?

SangJunBak · 2026-08-13T20:13:08Z

+                let buffered = u64::try_from(*buffered_bytes).unwrap_or(u64::MAX);
+                let max = u64::try_from(*max_buffered_bytes).unwrap_or(u64::MAX);
+                write!(
+                    f,
+                    "SUBSCRIBE fell behind: the client did not read results fast enough, so its backlog reached {}, exceeding the {} budget",
+                    ByteSize::b(buffered),
+                    ByteSize::b(max),


Wondering if MAX is the right default here or if we should keep it nullish?

SangJunBak · 2026-08-13T20:13:08Z

+#[mz_ore::test]
+#[cfg_attr(miri, ignore)] // too slow
+#[allow(clippy::disallowed_methods)]
+fn test_subscribe_single_large_batch_not_retired() {


I would like to see more tests that assert we never error prematurely with BufferExceeded, since that'd be really annoying for users if it did! Maybe some edge cases like buffering right up to the budget and asserting we never saw an error? Then adding a batch and asserting that we do? I think we can control the rate of the client receiving the buffer via CURSOR and FETCH.

A test case involving the interaction between the new session variable and max_result_size would be nice too! Like a message that would fit under max_result_size but over subscribe_max_buffered_bytes would not error, but as soon as you push a message over subscribe_max_buffered_bytes again, then you get the error.

SangJunBak · 2026-08-13T20:13:09Z

        });

    for row in rows {
+        // Rows are stored as pre-serialized JSON arrays. Parse each one back to


One thing I'd really recommend doing is running a materialized with this dev/pr version, then opening up the Console on port 6874 and making sure the initial pages load correctly https://materialize.com/docs/get-started/quickstart/#step-0-open-the-sql-shell. The console relies heavily on the http/ws API so I just wanna make sure nothing breaks!

Ok, I did that by building the emulator and opening the console against it. Went through the quickstart and it all works (as well as the other initial pages).

I also looked at the size-limit edge case since (e.g., a SELECT that exceeds max_result_size) streams partial rows and then the error, and the Console renders that correctly.

(To hit the streaming path I raised the Console's own max_query_result_size above the result)

SangJunBak · 2026-08-13T20:13:09Z

+                }
+                let types = &desc.typ().column_types;
+                while let Some(row) = rows.next() {
+                    result_size += row.byte_len();


Maybe use saturating_add?

SangJunBak · 2026-08-13T20:13:09Z

+    let mut datum_vec = mz_repr::DatumVec::new();
+    let mut result_size: usize = 0;
+    let mut rows_returned: usize = 0;
+    let mut sent_rows_desc = false;


Why do we need to use this boolean and not just send WebSocketResponse::Rows here, similar to how we send WebSocketResponse::Rows for Subscribe?

Because the first batch runs verify_datum_desc and can fail. If it fails, the contract is to emit only an Error without a Rows header. The boolean defers the header until a batch passes the verification.

SangJunBak · 2026-08-13T20:13:10Z

+    // Every row arrives as its own `Row` message before `CommandComplete`.
+    let mut rows_seen = 0;
+    loop {
+        match read_msg() {
+            WebSocketResponse::Row(_) => rows_seen += 1,
+            WebSocketResponse::Notice(_) => continue,
+            WebSocketResponse::CommandComplete(tag) => {
+                assert_eq!(tag, "SELECT 1000");
+                break;
+            }
+            other => panic!("unexpected message before CommandComplete: {other:?}"),
+        }
+    }
+    assert_eq!(rows_seen, 1000);


Wasn't this already the case? i.e. SELECT via ws would still send each row as its own message, it'd just send them all at once rather than stream them out. I feel like maybe a better test is setting a max_result_size to equal like X messages. Then when you do the same test, you'd expect X messages to be sent back before the error.

On writing this however, we should probably ensure the Console handles partially returned rows preceding an error. I think it should. Although the prior method of reading everything out of a buffer was inefficient, what it did prevent was someone doing SELECT * from … in the Console's shell via the WS API, then hitting the error after 10,000 rows, but now the Console has 10,000 rows it may've already rendered.

Good catch, I reworked this based on your suggestion.

Re the console: I did verify that the console handles partially returned rows then an error (see earlier screenshot)!

SangJunBak

Moreso nits and naming suggestions, although I would highly recommend the suggestions regarding maintaining compatibility the Console.

Some things I noticed but probably not worth :

For the SUBSCRIBE buffer issue, I think the initial reproduction Dennis in the linear ticket found still exists since the query accumulates ~6GB in the snapshot rather than the backlog. I think max_result_size is meant to catch this, but this is more of a bug around max_result_size only accounting for the raw row data (e.g. uint4 rows) and not the pVTCs that wrap them.
Feels weird that the definition of max_result_size is inconsistent depending on http/ws/pgwire. e.g. for subscribes, clusterds use it before they send back each batch, while http actually looks at the full result set. I don't think it's important enough to act on, but it does feel weird, especially with these new session variables that now do memory accounting but for different forms of the same data.

…output (SQL-428) Problem: A single `SELECT` over the HTTP endpoints could make environmentd buffer the entire result in memory and OOM, even when the client reads as fast as possible. The peek path already streams results out of the replica in bounded batches, so the growth was entirely the HTTP layer re-buffering those batches as an inflated `serde_json::Value` tree (~10-15x the wire size). Solution: - WebSocket: stream rows straight through. `SELECT` now flows through a new `StatementResult::Rows` variant whose sender streams `Rows(desc)` + per-row `Row` messages, flushing between stash batches for backpressure and enforcing `max_result_size` incrementally. Only one batch is resident, so large results over WebSocket get the same bounded memory profile as pgwire `SELECT`. The row descriptor is sent lazily so a query that errors before producing rows still emits only an `Error`. - Plain JSON: the response is one buffered document, so it stays buffered. But its footprint is smaller and its cap is now accurate. Rows are accumulated as pre-serialized `Vec<Box<serde_json::value::RawValue>>` instead of a `Value` tree, and `max_result_size` is enforced on the real serialized byte count. A genuinely oversized result fails fast with `ResultSize` instead of OOMing. The MCP and prometheus consumers of `SqlResult::Rows` parse each `RawValue` on demand. A streamed `SELECT` that errors mid-peek now reports `has_rows: true` over WebSocket (it commits to being row-returning before streaming), matching the `SUBSCRIBE` arm and pgwire.

Problem: A single slow or stopped SUBSCRIBE client could make environmentd buffer the whole subscribe output in memory and OOM. The replica -> controller -> coordinator response path is unbounded at every hop, and the producer runs on the non-blockable coordinator main loop, so it cannot apply backpressure to a slow client. Solution: Bound the terminal buffer and retire the sink. `ActiveSubscribe` gains a shared byte counter and a per-sink budget. Each `send()` adds the batch's byte size and carries it on the channel. A thin receiver wrapper, installed where the subscribe `rx` is built, subtracts it as the client writer drains each batch, so the counter tracks the currently buffered depth. After each `process_response`, the coordinator retires the subscribe if the buffered depth exceeds the budget, surfacing a new `AdapterError::SubscribeFellBehind` ("SUBSCRIBE fell behind ...") with a hint to use a non-buffering client or `COPY (SUBSCRIBE ...) TO STDOUT`. The budget is a new dyncfg `subscribe_max_buffered_bytes`, default 128 MiB.

…ize error

SangJunBak

Thanks for applying all my requested changes / testing the Console! This looks good to me!

mtabebe force-pushed the ma/sql-428-oom-buffered-output branch 5 times, most recently from 13ce959 to 5854351 Compare July 28, 2026 20:24

mtabebe marked this pull request as ready for review July 28, 2026 21:09

mtabebe requested review from a team as code owners July 28, 2026 21:09

SangJunBak self-requested a review August 11, 2026 15:28

SangJunBak requested changes Aug 13, 2026

View reviewed changes

SangJunBak reviewed Aug 13, 2026

View reviewed changes

mtabebe added 2 commits August 18, 2026 17:01

mtabebe force-pushed the ma/sql-428-oom-buffered-output branch 2 times, most recently from 74cee1f to be0cfc4 Compare August 19, 2026 00:41

address feedback: pin that a WebSocket SELECT streams rows before a s…

282f3c1

…ize error

mtabebe force-pushed the ma/sql-428-oom-buffered-output branch from be0cfc4 to 282f3c1 Compare August 19, 2026 15:47

mtabebe requested a review from SangJunBak August 19, 2026 17:51

SangJunBak approved these changes Aug 19, 2026

View reviewed changes

mtabebe merged commit b0d4c75 into MaterializeInc:main Aug 19, 2026
83 checks passed

ggevay mentioned this pull request Oct 10, 2026

compute: allow SUBSCRIBE AS OF MAX UP TO MAX #37611

Merged

Conversation

mtabebe commented Jul 28, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SangJunBak left a comment

Choose a reason for hiding this comment

Uh oh!

SangJunBak left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

mtabebe commented Jul 28, 2026 •

edited by github-actions Bot

Loading