Module 4 — The HTTP/TCP Subsystem & Ingestion Backpressure

• 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:

const socket = net.createConnection({ host, port });
 
let writesPending = 0;
const MAX_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, resume
      if (writesPending < MAX_PENDING / 2 && socket.isPaused()) {
        socket.resume();
      }
    });
  }
 
  // Pause if too many writes are in flight
  if (writesPending >= MAX_PENDING) {
    socket.pause();  // stops reading from TCP receive buffer
    // kernel receive buffer fills → window shrinks → sender slows
  }
});

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.

// Wrong: buffering the entire body before processing
app.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 loop
    processAll(transactions);
    res.sendStatus(200);
  });
});
 
// Correct: streaming processing with backpressure
app.post('/ingest', (req, res) => {
  const parser = createStreamingParser();   // Transform stream
  const 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 }));
});

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
• TCP socket paused → kernel receive buffer fills → sender slows

---

writable.writableLength and writable.writableHighWaterMark

Real-time monitoring of stream buffer state:

// Monitor backpressure state in production
const 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);

---

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:

1. 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.
2. 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.

import http2 from 'node:http2';
 
// HTTP/2 server for persistent block subscription connections
const 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());
});

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:

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 hanging
 
server.listen(3000);
 
// Monitor connection count
setInterval(() => {
  server.getConnections((err, count) => {
    if (!err) {
      console.log(`Active connections: ${count}/${server.maxConnections}`);
    }
  });
}, 5000);

socket.setTimeout(): Eliminating Zombie Connections

Connections that are established but never send data consume file descriptors indefinitely without a timeout:

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));
});

---

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:

// Token bucket rate limiter per IP address
class RateLimiter {
  #buckets = new Map();
  #rate;      // tokens per second
  #capacity;  // maximum bucket depth
 
  constructor({ 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 time
    const 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;
      return true;   // allowed
    }
    return false;    // rate limited
  }
}
 
const limiter = new RateLimiter({ 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 connection
      return;
    }
    processData(chunk);
  });
});

---

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:

// Original ingestion handler — no backpressure
socket.on('data', (chunk) => {
  const payments = parsePayments(chunk);
  payments.forEach(payment => {
    // db.write returns a Promise — not awaited
    db.write(payment).then(() => emitConfirmation(payment));
  });
});

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:

// Step 1: Create a write stream with appropriate HWM
const paymentWriteStream = new Writable({
  objectMode: true,
  highWaterMark: 500,  // buffer max 500 payments
  write(payment, _, callback) {
    db.write(payment)
      .then(() => {
        emitConfirmation(payment);
        callback();
      })
      .catch(callback);
  }
});
 
// Step 2: Create a parse transform stream
const parseStream = new Transform({
  readableObjectMode: true,
  transform(chunk, _, callback) {
    const payments = parsePayments(chunk);
    payments.forEach(p => this.push(p));
    callback();
  }
});
 
// Step 3: Wire with pipe — backpressure automatic
socket
  .pipe(parseStream)
  .pipe(paymentWriteStream);
 
// pipe() handles pause/resume automatically:
// paymentWriteStream full → parseStream pauses
// parseStream paused → socket pauses
// socket paused → TCP receive buffer fills
// TCP buffer fills → window = 0 → upstream aggregator slows

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

// Emit backpressure metrics for observability
const backpressureCounter = new Counter({
  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());
  }
});

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

---

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.

Next: Module 5 — Native Streams & Off-Heap Buffer Storage →

---