Module 4 — The HTTP/TCP Subsystem & Ingestion Backpressure
What this module covers: Backpressure is the mechanism that prevents a fast producer from overwhelming a slow consumer. In a blockchain indexer, the producer is the network delivering transaction events at 50K/sec. The consumer is your PostgreSQL database accepting 5K writes/sec. Without backpressure, your process accumulates an unbounded in-memory queue and eventually OOMs. This module covers how TCP flow control works at the kernel level, how Node.js stream backpressure mirrors it at the application level, and the precise implementation of backpressure for high-throughput ingestion pipelines.
The Problem: Producers Outpacing Consumers
A blockchain indexer at peak load has an inherent mismatch:
Producer: blockchain full node pushing 50,000 transaction events/second over TCP
The fix requires two layers of backpressure working together:
TCP flow control — the kernel tells the sender to slow down when the receive buffer is full
Stream backpressure — Node.js pauses the socket read when the downstream consumer is busy
Understanding how these two layers interact is the key to building a pipeline that never OOMs under load.
TCP Flow Control: The Kernel's Backpressure
TCP has backpressure built in via the receive window mechanism.
Every TCP ACK includes a window field — the number of bytes the receiver is willing to accept. When the kernel's receive buffer fills up, it advertises a smaller window to the sender. When the buffer is full, it advertises zero window — the sender must stop completely.
text
The critical insight: TCP flow control automatically propagates backpressure upstream. When Node.js stops reading from the socket (because downstream is slow), the kernel receive buffer fills, the window shrinks to zero, and the blockchain full node is forced to stop sending. The backpressure is communicated all the way back to the data source — no data is dropped, it is simply slowed.
The mailbox analogy: A zero TCP window is the postal equivalent of a mailbox so full the carrier physically can't fit another envelope through the slot — she doesn't lose your letter, she just waits on the porch until you empty the box.
Node.js Streams: Application-Level Backpressure
Node.js streams implement the same backpressure mechanism at the JavaScript level, mirroring TCP flow control.
The highWaterMark (HWM)
Every writable stream has a highWaterMark — the maximum number of bytes (or objects) it is willing to buffer before signalling that it is full.
javascript
The write() Return Value: The Backpressure Signal
When you write to a writable stream, it returns a boolean:
true — buffer is below HWM, safe to continue writing
false — buffer has reached HWM, you should stop writing
javascript
Production war story: A UPI settlement reconciliation service ignored the boolean return of write() entirely — every call site treated stream.write(data) as fire-and-forget. Normally this was harmless because the downstream consumer kept pace. Then a partner bank's API started responding in a degraded 2 seconds instead of the usual 50ms, at the same time festival load pushed reconciliation volume to several times normal. With no code checking the return value and no drain listener anywhere, the service's internal write buffer grew unbounded for 40 straight minutes — invisible in dashboards that only tracked request rate, not buffer depth — until the process hit its memory ceiling and was OOM-killed at 3 AM. The fix was mechanical (route every write through the writeWithBackpressure helper above) but the incident illustrated the core risk: write()'s return value isn't optional telemetry, it's the only signal standing between a slow downstream and an unbounded buffer.
The drain Event: The Resume Signal
When the stream's internal buffer empties below HWM after being full, it emits drain. This is your signal to resume writing.
javascript
cork() / uncork(): Micro-Batching Small Writes
When an ingestion pipeline emits many small writes in a tight loop — one write() per parsed transaction, for example — each call can incur its own overhead in the underlying writable (a syscall for a socket, a query round-trip for some DB drivers). writable.cork() tells the stream to buffer everything written until a matching uncork(), so those small writes get coalesced into fewer, larger underlying operations.
javascript
Two things to keep in mind:
cork()/uncork() calls nest. Each cork() increments an internal counter; the stream only actually flushes once uncork() has been called the same number of times. Mismatched calls (or forgetting the matching uncork()) leave the stream corked forever — nothing gets written and the pipeline silently stalls.
Corking does not bypass backpressure.write() still returns false once the buffered (corked) data exceeds highWaterMark, and drain still fires the same way. cork() changes how the underlying writes are batched, not the HWM threshold that governs when you should stop calling write() at all.
For an ingestion pipeline parsing thousands of small transactions per chunk, wrapping the per-transaction writes in cork()/uncork() around each _transform call is a standard way to cut per-write overhead without changing the backpressure contract.
readable.pause() and readable.resume()
For explicit backpressure control on a readable stream:
javascript
The Danger of Unpaused Readable Streams
A net.Socket is a Readable stream. In Node.js, a readable stream that is not being consumed operates in two modes:
Flowing mode (default when a data event listener is attached): data is emitted as fast as it arrives. If you can't process it, it accumulates in memory.
Paused mode (after socket.pause()): data accumulates in the kernel receive buffer. No JavaScript memory consumption. TCP flow control does its job.
The correct pattern for a high-throughput ingestion socket: monitor write queue depth and pause/resume the socket accordingly.
for await...of vs. Manual pause()/resume(): Don't Mix Them
The processIncoming example earlier in this module drives a socket with for await (const chunk of socket). That works because the async iterator protocol has its own internal backpressure handling: it pulls chunks on demand, respecting the stream's highWaterMark, and the await inside the loop body is the only pause/resume mechanism at play — when the awaited write is slow, the iterator simply doesn't request the next chunk yet, which lets the stream's own internal buffering (and, transitively, TCP flow control) take effect.
The danger is combining that with manual.pause()/.resume() calls on the same stream, driven by separate logic — for example, layering the writesPending counter pattern shown above on top of a for await...of loop over the same socket. The two mechanisms don't coordinate:
The async iterator's internal reader assumes it has exclusive control over the stream's flowing/paused state between iterations.
An external pause() call that lands mid-iteration can leave the stream paused when the iterator's next pull expects it to be readable, stalling the loop indefinitely.
An external resume() call can likewise flip the stream back into flowing mode while the iterator is mid-await, causing data to be emitted (and potentially dropped) outside the iterator's read path.
Rule of thumb: pick one backpressure mechanism per stream. Use for await...of (with await inside the loop as the throttle) for cleanly composed pipelines, or use .on('data') with manual pause()/resume() for fine-grained flood control like the token-bucket rate limiter later in this module — but never both on the same stream at the same time.
HTTP Ingestion: IncomingMessage as a Stream
http.IncomingMessage (the req object in an HTTP server) is a Readable stream backed by the underlying TCP socket. The same backpressure rules apply.
javascript
With .pipe(), backpressure is handled automatically:
If dbWriter returns false from write(), parser pauses
If parser is paused, req pauses
If req is paused, the underlying TCP socket is paused
writable.writableLength and writable.writableHighWaterMark
Real-time monitoring of stream buffer state:
javascript
HTTP/2: Multiplexed Streams for Persistent Ingestion Connections
For blockchain indexers that maintain persistent connections to full nodes, HTTP/2 provides stream multiplexing over a single TCP connection.
HTTP/2 flow control (RFC 7540 §5.2) operates at two independent layers, and both must have available window before a DATA frame can be sent:
Per-stream window — each HTTP/2 stream (e.g., one full node's block subscription) has its own flow-control window. Pausing or slowing one stream does not, by itself, affect the others multiplexed over the same connection.
Per-connection window — the underlying TCP connection has a single, shared flow-control window that all streams on it draw from. Even if an individual stream's window has plenty of room, no data for any stream can be sent once the connection-level window is exhausted.
This means a single slow-draining stream can starve the shared connection window and stall every other multiplexed stream on that connection — including ones whose consumers are keeping up fine. Node's HTTP/2 implementation manages connection-level WINDOW_UPDATE frames automatically, but when diagnosing head-of-line stalls in a multiplexed ingestion pipeline, per-stream throughput alone won't tell the full story — watch connection-level window exhaustion too.
javascript
Benefit for blockchain indexers: instead of 1,000 separate TCP connections from 1,000 full nodes, use 10 HTTP/2 connections each multiplexing 100 streams. Fewer file descriptors, fewer TCP handshakes, same data throughput. Each stream has independent backpressure at the stream level — but distribute full nodes across enough connections that one slow producer's connection-level stall doesn't drag down unrelated streams sharing its connection.
Handling Socket Floods: server.maxConnections
When a blockchain network goes through a major upgrade, every node may attempt to reconnect simultaneously (similar to the thundering herd from Module 3). server.maxConnections provides a hard cap:
Connections that are established but never send data consume file descriptors indefinitely without a timeout:
javascript
Rate Limiting at the TCP Layer
Before requests reach your application routing logic, you can enforce rate limits at the connection level using token bucket algorithms:
javascript
The Production Incident: OOM from Missing Backpressure During Airdrop
Context: A UPI payment processing gateway ingesting payment events from multiple upstream aggregators. Normal throughput: 3,000 events/second. During a major e-commerce sale, 35,000 events/second arrived simultaneously.
The broken pipeline:
javascript
At 35,000 events/sec with 8,000 db writes/sec capacity:
Second 1: 27,000 writes queued in memory
Second 5: 135,000 writes queued → ~270MB
Second 12: ~380,000 writes queued → ~760MB
Second 14: Process OOM-killed by the OS
The Kubernetes pod restarted. On restart, the TCP connections reconnected, and the same flood resumed. The pod crashed again in 14 seconds. Kubernetes kept restarting it. This cycle continued for 8 minutes until the upstream aggregators hit their own connection retry limits.
The fix — stream pipeline with backpressure:
javascript
Result: Under 35,000 events/sec, the pipeline automatically throttled to 8,000/sec — the database's write capacity. Memory stabilized at ~12MB (the 500-object buffer × ~24KB per object). No OOM. The upstream aggregator experienced slightly higher latency during the peak, but no data was lost.
Backpressure Monitoring
javascript
Tracking nodejs_stream_backpressure_total over time tells you exactly how often your pipeline is capacity-constrained — invaluable for right-sizing your database write pool and scaling decisions.
Summary
Concept
Key Takeaway
TCP flow control
Zero window = sender pauses. Kernel-level backpressure built into TCP.
highWaterMark
Buffer threshold. When exceeded, write() returns false.
write() return value
false means pause upstream. true means continue.
drain event
Buffer drained below HWM. Safe to resume writing.
cork() / uncork()
Micro-batch small writes into fewer underlying operations. Doesn't change HWM/backpressure behavior.
readable.pause()
Stops consuming from kernel receive buffer. TCP backpressure propagates upstream.
for await...of streams
Has its own internal HWM-driven backpressure. Never combine with manual pause()/resume() on the same stream.
.pipe()
Handles pause/resume automatically. Correct default for most pipelines.
writable.writableLength
Current bytes/objects buffered. Monitor this for capacity planning.
server.maxConnections
Hard cap on simultaneous connections. Refuse rather than hang.
socket.setTimeout()
Kill zombie connections. Essential for preventing fd exhaustion.
HTTP/2 multiplexing
Fewer TCP connections, per-stream backpressure, better fd utilization. Flow control is per-stream AND per-connection (RFC 7540 §5.2) — a shared connection window can stall all its streams.
Token bucket rate limit
Enforce per-IP limits before requests reach application logic.
Backpressure prevents memory exhaustion from the outside in. Module 5 goes inside — how to process gigabytes of data without touching the V8 heap at all, using off-heap Buffers, Transform streams, and streaming pipelines for blockchain transaction log processing.
What kernel-level mechanism causes an upstream sender to slow down when a Node.js application calls socket.pause() on a fast readable socket?
Why is piping streams using req.pipe(parser).pipe(dbWriter) superior to concatenating the payload via req.on('data', ...) and processing it synchronously?
What happens to incoming TCP data immediately after socket.pause() is executed in a Node.js application?
Test your knowledge with more question sets
Sign in to access a wider variety of questions and get notified when new practice sets are added to this module.
// What happens without backpressure:socket.on('data',(chunk)=>{const transactions =parse(chunk);for(const tx of transactions){ db.write(tx);// returns a Promise, does NOT wait for it}});// At 50K tx/sec ingest, 8K/sec db write:// After 1 second: 42K tx queued in memory// After 10 seconds: 420K tx queued → ~500MB memory// After 30 seconds: OOM kill
// Default HWMs:// Byte streams: 16KB (16384 bytes)// Object mode streams: 16 objects// Custom HWM for a database write stream:const dbWriteStream =newWritable({objectMode:true,highWaterMark:1000,// buffer up to 1000 objects before signalling fullwrite(transaction, encoding, callback){ db.write(transaction).then(()=>callback()).catch(callback);}});
// The contract:const canContinue = writable.write(chunk);if(!canContinue){// STOP writing. The stream has too much buffered.// Wait for the 'drain' event before writing more.}
// Correct backpressure implementation for a DB write pipelinefunctionwriteWithBackpressure(stream, data){const canContinue = stream.write(data);if(!canContinue){returnnewPromise(resolve=> stream.once('drain', resolve));}returnPromise.resolve();}// Usage in an ingestion loopasyncfunctionprocessIncoming(socket, dbStream){forawait(const chunk of socket){const transactions =parseChunk(chunk);for(const tx of transactions){awaitwriteWithBackpressure(dbStream, tx);// If dbStream is full, this await suspends here// The for-await loop pauses, which pauses the socket read// Socket pause triggers TCP receive buffer to fill// TCP receive buffer full → zero window → sender pauses// Backpressure propagated all the way to the data source}}}
// Micro-batch a burst of small transaction writes into one flushfunctionwriteBatchCorked(stream, transactions){ stream.cork();for(const tx of transactions){ stream.write(tx);}// Uncork on the next tick — everything written since cork() flushes together process.nextTick(()=> stream.uncork());}
const socket = net.createConnection({ host, port });let writesPending =0;constMAX_PENDING=5000;socket.on('data',(chunk)=>{const transactions =parse(chunk);for(const tx of transactions){ writesPending++; db.write(tx).finally(()=>{ writesPending--;// If socket was paused due to queue depth, resumeif(writesPending <MAX_PENDING/2&& socket.isPaused()){ socket.resume();}});}// Pause if too many writes are in flightif(writesPending >=MAX_PENDING){ socket.pause();// stops reading from TCP receive buffer// kernel receive buffer fills → window shrinks → sender slows}});
// Wrong: buffering the entire body before processingapp.post('/ingest',(req, res)=>{let body =''; req.on('data',chunk=> body += chunk);// accumulates entire body in memory req.on('end',()=>{const transactions =JSON.parse(body);// synchronous parse blocks event loopprocessAll(transactions); res.sendStatus(200);});});// Correct: streaming processing with backpressureapp.post('/ingest',(req, res)=>{const parser =createStreamingParser();// Transform streamconst dbWriter =createDbWriteStream();// Writable stream with HWM
req
.pipe(parser)// parse chunks as they arrive.pipe(dbWriter)// write parsed transactions, apply backpressure.on('finish',()=> res.sendStatus(200)).on('error',(err)=> res.status(500).json({error: err.message}));});
// Monitor backpressure state in productionconst dbStream =createDbWriteStream({highWaterMark:1000});setInterval(()=>{const utilization = dbStream.writableLength/ dbStream.writableHighWaterMark;console.log(`DB write buffer: ${(utilization *100).toFixed(0)}% full`);if(utilization >0.9){console.warn('DB write stream near capacity — upstream should pause');}},1000);
importhttp2from'node:http2';// HTTP/2 server for persistent block subscription connectionsconst server = http2.createSecureServer({ key, cert });server.on('stream',(stream, headers)=>{// Each HTTP/2 stream has its own flow-control window (per-stream), but all// streams on this connection share one connection-level window (RFC 7540 §5.2).// A stream only receives data when BOTH its own window and the shared// connection window have room. stream.on('data',(chunk)=>{const canContinue =processChunk(chunk);if(!canContinue){ stream.pause();// pause this stream's flow}}); stream.on('drain',()=> stream.resume());});
const server = net.createServer(handleConnection);server.maxConnections=5000;// refuse connections beyond this// When maxConnections is reached:// - New TCP SYN packets are rejected// - Clients receive connection refused immediately// - Cleaner than accepting and then hangingserver.listen(3000);// Monitor connection countsetInterval(()=>{ server.getConnections((err, count)=>{if(!err){console.log(`Active connections: ${count}/${server.maxConnections}`);}});},5000);
server.on('connection',(socket)=>{// Kill connections that have been idle for 30 seconds socket.setTimeout(30_000); socket.on('timeout',()=>{ socket.destroy();// force close — no graceful shutdown for idle sockets});// Reset timeout on activity socket.on('data',()=> socket.setTimeout(30_000));});
// Token bucket rate limiter per IP addressclassRateLimiter{ #buckets =newMap(); #rate;// tokens per second #capacity;// maximum bucket depthconstructor({ rate, capacity }){this.#rate= rate;this.#capacity= capacity;}consume(ip){const now =Date.now();let bucket =this.#buckets.get(ip);if(!bucket){ bucket ={tokens:this.#capacity,lastRefill: now };this.#buckets.set(ip, bucket);}// Refill based on elapsed timeconst elapsed =(now - bucket.lastRefill)/1000; bucket.tokens=Math.min(this.#capacity, bucket.tokens+ elapsed *this.#rate); bucket.lastRefill= now;if(bucket.tokens>=1){ bucket.tokens-=1;returntrue;// allowed}returnfalse;// rate limited}}const limiter =newRateLimiter({rate:1000,capacity:5000});server.on('connection',(socket)=>{const ip = socket.remoteAddress; socket.on('data',(chunk)=>{if(!limiter.consume(ip)){ socket.destroy();// hard limit exceeded — drop connectionreturn;}processData(chunk);});});
// Original ingestion handler — no backpressuresocket.on('data',(chunk)=>{const payments =parsePayments(chunk); payments.forEach(payment=>{// db.write returns a Promise — not awaited db.write(payment).then(()=>emitConfirmation(payment));});});
// Emit backpressure metrics for observabilityconst backpressureCounter =newCounter({name:'nodejs_stream_backpressure_total',help:'Number of times backpressure was applied to incoming socket'});socket.on('data',(chunk)=>{const canContinue = parseStream.write(chunk);if(!canContinue){ backpressureCounter.inc(); socket.pause(); parseStream.once('drain',()=> socket.resume());}});