Network Programming · Part 8 — Streams, Backpressure & Message Framing

Đây là Phần 8 của series 10 bài về lập trình mạng với Node.js + TypeScript. Phần 2 đã hứa một sự thật khó: TCP là luồng byte — giữ thứ tự và tin cậy, nhưng không giữ ranh giới message. Hôm nay ta đối mặt sự thật đó, thêm framing để message logic sống sót qua việc cắt chunk tùy ý, và xử lý backpressure để bên gửi nhanh không làm tràn bộ nhớ.

Sự thật cốt lõi: không có ranh giới message

Khi bạn socket.write('HELLO') rồi socket.write('WORLD'), TCP đảm bảo bên nhận nhận byte đúng thứ tự — nhưng không đảm bảo cách chúng được gom thành sự kiện 'data'.

Hai điều này xảy ra liên tục trên production:

• Một message, nhiều chunk — một payload JSON đến bị cắt qua nhiều callback 'data'.
• Nhiều message, một chunk — hai lần write liên tiếp nằm chung một sự kiện 'data'.

OS, NIC, và thuật toán Nagle quyết định kích thước chunk — không phải ứng dụng của bạn. Giao thức của bạn phải dựng lại message logic từ luồng byte.

Ý chính: 'data' báo “thêm byte đã đến” — không phải “đây là một message hoàn chỉnh”.

Server ngây thơ bị vỡ

Server này giả định một sự kiện 'data' = một message JSON. Chạy được trên localhost với payload nhỏ — rồi hỏng khi traffic thật.

import { createServer } from 'node:net';

const server = createServer((socket) => {
  socket.on('data', (chunk: Buffer) => {
    // ❌ WRONG: treats each chunk as a complete message
    const msg = JSON.parse(chunk.toString('utf8'));
    console.log('got:', msg);
    socket.write(JSON.stringify({ ok: true, echo: msg }) + '\n');
  });
});

server.listen(3000);

Client gửi hai message nhỏ liên tiếp:

import { connect } from 'node:net';

const socket = connect(3000, () => {
  socket.write(JSON.stringify({ id: 1, cmd: 'ping' }));
  socket.write(JSON.stringify({ id: 2, cmd: 'ping' }));
});

Điều gì hỏng:

• Gộp — một chunk có thể là {"id":1,...}{"id":2,...} → JSON.parse ném SyntaxError.
• Cắt — chunk 1 có thể là {"id":1,"cmd":"pi và chunk 2 ng"} → parse thất bại trên cả hai nửa.

Cách sửa không bao giờ là “hy vọng chunk khớp” — bạn đóng khung message trên dây và buffer phía đọc cho đến khi có frame đủ.

Chiến lược framing A: delimiter

Framing đơn giản nhất: chọn delimiter mà byte payload không thể chứa (hoặc escape chúng). Giao thức text thường dùng xuống dòng — một message mỗi dòng, hay NDJSON khi mỗi dòng là JSON.

// Encode: payload + '\n'
function encodeLine(obj: unknown): Buffer {
  return Buffer.from(JSON.stringify(obj) + '\n', 'utf8');
}

// Decode: accumulate bytes, split on '\n', emit complete lines
class LineDecoder {
  private buf = Buffer.alloc(0);

  push(chunk: Buffer): string[] {
    this.buf = Buffer.concat([this.buf, chunk]);
    const lines: string[] = [];
    let start = 0;

    for (let i = 0; i < this.buf.length; i++) {
      if (this.buf[i] === 0x0a) {
        // byte before newline is one complete message
        lines.push(this.buf.subarray(start, i).toString('utf8'));
        start = i + 1;
      }
    }

    // keep incomplete tail for next chunk
    this.buf = this.buf.subarray(start);
    return lines;
  }
}

Server dùng line decoder:

import { createServer } from 'node:net';

const decoder = new LineDecoder();

const server = createServer((socket) => {
  const perSocket = new LineDecoder();

  socket.on('data', (chunk) => {
    for (const line of perSocket.push(chunk)) {
      const msg = JSON.parse(line) as { id: number; cmd: string };
      console.log('frame:', msg);
      socket.write(encodeLine({ ok: true, id: msg.id }));
    }
  });
});

server.listen(3000);

Framing delimiter tốt cho log, chat, và giao thức debug được bằng mắt. Khó khi payload là binary hoặc phải chứa byte delimiter.

Chiến lược framing B: length-prefix

Length-prefix framing thêm header cố định chứa độ dài payload, rồi byte thô. Lựa chọn phổ biến: số nguyên 4 byte big-endian + payload.

 on the wire:
┌──────────┬─────────────────────┐
│ len = 5  │  H E L L O          │
│ 4 bytes  │  5 bytes payload    │
└──────────┴─────────────────────┘

Encoder

const HEADER_SIZE = 4;

function encodeFrame(payload: Buffer): Buffer {
  const header = Buffer.alloc(HEADER_SIZE);
  header.writeUInt32BE(payload.length, 0);
  return Buffer.concat([header, payload]);
}

function encodeJsonFrame(obj: unknown): Buffer {
  return encodeFrame(Buffer.from(JSON.stringify(obj), 'utf8'));
}

Stateful decoder (buffers partial reads)

class LengthPrefixDecoder {
  private buf = Buffer.alloc(0);

  push(chunk: Buffer): Buffer[] {
    this.buf = Buffer.concat([this.buf, chunk]);
    const frames: Buffer[] = [];

    while (this.buf.length >= HEADER_SIZE) {
      const len = this.buf.readUInt32BE(0);

      if (this.buf.length < HEADER_SIZE + len) {
        // header says N bytes, but we haven't received all of them yet
        break;
      }

      const payload = this.buf.subarray(HEADER_SIZE, HEADER_SIZE + len);
      frames.push(payload);
      this.buf = this.buf.subarray(HEADER_SIZE + len);
    }

    return frames;
  }
}

Server + client with length-prefix

import { connect, createServer } from 'node:net';

// --- server ---
const server = createServer((socket) => {
  const decoder = new LengthPrefixDecoder();

  socket.on('data', (chunk) => {
    for (const payload of decoder.push(chunk)) {
      const msg = JSON.parse(payload.toString('utf8')) as {
        id: number;
        cmd: string;
      };
      console.log('frame:', msg);
      socket.write(encodeJsonFrame({ ok: true, id: msg.id }));
    }
  });
});

server.listen(3000);

// --- client ---
const client = connect(3000, () => {
  client.write(encodeJsonFrame({ id: 1, cmd: 'ping' }));
  client.write(encodeJsonFrame({ id: 2, cmd: 'ping' }));
});

client.on('data', (chunk) => {
  const decoder = new LengthPrefixDecoder();
  for (const payload of decoder.push(chunk)) {
    console.log('reply:', JSON.parse(payload.toString('utf8')));
  }
});

Lưu ý: ví dụ client trên tạo decoder mới mỗi 'data' — trong code thật, giữ một decoder mỗi socket suốt đời kết nối, giống server. Vòng while là cốt lõi: không parse cho đến khi có header + đủ payload.

So sánh delimiter và length-prefix

Chọn delimiter khi người và nc quan trọng; chọn length-prefix khi message có thể binary, lớn, hoặc dày.

Backpressure: khi write() bảo “chậm lại”

Framing giải quyết chunking phía đọc. Backpressure giải quyết quá tải phía ghi.

socket.write(data) copy vào buffer gửi kernel và trả về ngay hầu hết thời gian. Khi buffer đầy, write() trả về false. Nếu bạn vẫn gọi write(), Node xếp hàng dữ liệu trong bộ nhớ userland — tăng không giới hạn đến khi OOM.

import type { Socket } from 'node:net';

// Correct: respect backpressure
function writeWithBackpressure(
  socket: Socket,
  data: Buffer,
): Promise<void> {
  return new Promise((resolve, reject) => {
    const ok = socket.write(data, (err) => {
      if (err) reject(err);
    });

    if (ok) {
      // kernel accepted everything — safe to send more
      resolve();
    } else {
      // buffer full — wait until it drains
      socket.once('drain', resolve);
    }
  });
}

// Send many frames without blowing memory
async function sendMany(
  socket: Socket,
  frames: Buffer[],
): Promise<void> {
  for (const frame of frames) {
    await writeWithBackpressure(socket, frame);
  }
}

Sự kiện 'drain' bắn khi buffer kernel có chỗ trở lại. writableHighWaterMark (mặc định 16 KiB trên nhiều socket) gợi ý khi write() bắt đầu trả false.

Khi copy file→socket hoặc socket→socket, ưu tiên built-in đã xử lý sẵn:

import { createReadStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import type { Socket } from 'node:net';

async function pumpFileToSocket(
  filePath: string,
  socket: Socket,
): Promise<void> {
  const rs = createReadStream(filePath);
  await pipeline(rs, socket);
  // pipeline() pauses the source on backpressure and resumes on drain
}

socket.pipe() làm tương tự cho cặp readable→writable, nhưng stream.pipeline() còn chuyển lỗi và dọn dẹp — ưu tiên pipeline() trong code mới.

Gộp lại: echo có framing và backpressure

import { createServer, type Socket } from 'node:net';

const PORT = 3000;

function writeWithBackpressure(socket: Socket, data: Buffer): Promise<void> {
  return new Promise((resolve, reject) => {
    const ok = socket.write(data, (err) => {
      if (err) reject(err);
    });
    if (ok) resolve();
    else socket.once('drain', resolve);
  });
}

const server = createServer((socket) => {
  const decoder = new LengthPrefixDecoder();

  socket.on('data', async (chunk) => {
    for (const payload of decoder.push(chunk)) {
      const text = payload.toString('utf8');
      const reply = encodeFrame(Buffer.from(`echo:${text}`, 'utf8'));
      await writeWithBackpressure(socket, reply);
    }
  });
});

server.listen(PORT, () => {
  console.log(`framed echo on :${PORT}`);
});

Server này chịu được write bị cắt, bị gộp, và khối lượng lớn vì framing và backpressure đều được xử lý rõ ràng.

Lỗi người mới hay mắc

• Giả định một 'data' = một message — lỗi TCP số 1; luôn buffer và frame.
• Bỏ qua giá trị trả về của write() — xếp hàng mãi khi false gây tăng bộ nhớ không giới hạn.
• Quên buffer đọc một phần trong decoder — parse ngay khi byte đến chắc chắn JSON.parse lỗi lúc này lúc kia.
• Gọi JSON.parse trực tiếp trên chunk — demo chạy, production hỏng.
• Tạo decoder mới mỗi 'data' thay vì một decoder mỗi kết nối — mất state giữa các chunk.

Bài tập

Thử từng bài trước khi mở lời giải.

1. Chạy server JSON ngây thơ và client gửi hai write() liên tiếp. Log độ dài chunk và chứng minh khi JSON.parse ném lỗi.
2. Mở rộng LineDecoder từ chối dòng dài hơn 64 KiB và emit lỗi thay vì buffer mãi.
3. Viết client gửi 10 000 frame length-prefix nhanh nhất có thể không xử lý backpressure, rồi lặp lại với writeWithBackpressure. So sánh process.memoryUsage().heapUsed.

import { connect, createServer } from 'node:net';

// --- Exercise 1: demonstrate coalescing ---
const naive = createServer((socket) => {
  socket.on('data', (chunk) => {
    console.log('chunk length:', chunk.length);
    try {
      console.log('parsed:', JSON.parse(chunk.toString('utf8')));
    } catch (e) {
      console.log('JSON.parse failed:', (e as Error).message);
    }
  });
});
naive.listen(3001);

const c1 = connect(3001, () => {
  c1.write(JSON.stringify({ n: 1 }));
  c1.write(JSON.stringify({ n: 2 }));
});
// Often prints one chunk with length ~26 and SyntaxError — two objects, one chunk

// --- Exercise 2: LineDecoder with max line size ---
class SafeLineDecoder extends EventTarget {
  private buf = Buffer.alloc(0);
  private readonly max: number;

  constructor(maxBytes = 64 * 1024) {
    super();
    this.max = maxBytes;
  }

  push(chunk: Buffer): string[] {
    this.buf = Buffer.concat([this.buf, chunk]);
    if (this.buf.length > this.max) {
      this.dispatchEvent(new ErrorEvent('error', {
        error: new Error(`line exceeds ${this.max} bytes`),
      }));
      this.buf = Buffer.alloc(0);
      return [];
    }
    const lines: string[] = [];
    let start = 0;
    for (let i = 0; i < this.buf.length; i++) {
      if (this.buf[i] === 0x0a) {
        lines.push(this.buf.subarray(start, i).toString('utf8'));
        start = i + 1;
      }
    }
    this.buf = this.buf.subarray(start);
    return lines;
  }
}

// --- Exercise 3: backpressure memory comparison ---
import type { Socket } from 'node:net';

function writeWithBackpressure(socket: Socket, data: Buffer): Promise<void> {
  return new Promise((resolve) => {
    if (socket.write(data)) resolve();
    else socket.once('drain', resolve);
  });
}

async function flood(useBackpressure: boolean): Promise<void> {
  const socket = connect(3000);
  await new Promise<void>((r) => socket.once('connect', r));
  const frames = Array.from({ length: 10_000 }, (_, i) =>
    encodeJsonFrame({ i }),
  );
  const before = process.memoryUsage().heapUsed;
  if (useBackpressure) {
    for (const f of frames) await writeWithBackpressure(socket, f);
  } else {
    for (const f of frames) socket.write(f);
  }
  const after = process.memoryUsage().heapUsed;
  console.log(useBackpressure ? 'with BP' : 'no BP', {
    deltaMB: ((after - before) / 1024 / 1024).toFixed(1),
  });
  socket.end();
}

Điều cốt lõi

TCP cho bạn luồng byte có thứ tự, tin cậy — không phải message. 'data' là thông báo chunk, không phải ranh giới message; bạn frame trên dây (delimiter hoặc length-prefix) và buffer đến khi frame đủ. Phía gửi, tôn trọng write() trả false và chờ 'drain' (hoặc dùng pipeline()) để producer nhanh không làm cạn bộ nhớ. Mọi giao thức bạn từng dùng — HTTP, WebSocket, Redis — đều làm vậy bên trong; giờ bạn có thể tự xây.