Server-Sent Events thực chiến: Xây luồng realtime bền vững trên production

Một dashboard cần nhận trạng thái job, cảnh báo và số liệu mới theo thời gian thực, nhưng client không cần gửi dữ liệu ngược liên tục. Server-Sent Events (SSE) giải quyết đúng mô hình một chiều đó bằng một HTTP response mở lâu, định dạng text/event-stream và API EventSource có khả năng tự kết nối lại.

SSE trông đơn giản hơn WebSocket, nhưng production vẫn có những bẫy khó thấy: proxy gom buffer, idle timeout cắt kết nối, reconnect làm mất hoặc lặp event, consumer chậm ăn RAM, deploy đóng hàng nghìn socket cùng lúc. Bài này xây dựng luồng SSE có event ID, replay, heartbeat, backpressure và graceful shutdown dựa trên HTML Living Standard, Node.js HTTP API và Nginx proxy module.

1. Khi nào nên chọn SSE?

SSE phù hợp cho thông báo, tiến độ tác vụ, dashboard, log tail đã kiểm soát, feed trạng thái và cache invalidation tới UI. Nếu client cần gửi game state, collaborative editing hoặc binary frame hai chiều với tần suất cao, WebSocket thường hợp lý hơn.

2. Wire format nhỏ nhưng nghiêm ngặt

HTTP/1.1 200 OK
Content-Type: text/event-stream; charset=utf-8
Cache-Control: no-cache, no-transform
Connection: keep-alive

retry: 5000

id: 18452
event: job-progress
data: {"jobId":"a7","percent":72}

: heartbeat

id: 18453
data: a default message
data: with two lines

Mỗi event kết thúc bằng một dòng trống. Nhiều dòng data: được ghép bằng ký tự newline. event: đặt tên event, id: cập nhật last event ID, còn retry: gợi ý thời gian reconnect tính bằng mili giây. Dòng bắt đầu bằng dấu hai chấm là comment và hữu ích làm heartbeat.

Stream là UTF-8. Dữ liệu ứng dụng vẫn nên serialize JSON trên một hoặc nhiều dòng data:; không ghi JSON thô thiếu prefix và dòng trống, vì browser sẽ chưa dispatch event.

3. Client EventSource tối thiểu

const source = new EventSource('/api/events');

source.addEventListener('job-progress', (event) => {
  const payload = JSON.parse(event.data);
  updateProgress(payload.jobId, payload.percent);
});

source.onmessage = (event) => {
  console.log('default event', event.data, event.lastEventId);
};

source.onerror = () => {
  // EventSource thường tự reconnect; đây không nhất thiết là lỗi vĩnh viễn.
  showConnectionState(source.readyState);
};

// Khi rời màn hình hoặc không còn cần stream:
source.close();

Trạng thái CONNECTING, OPEN và CLOSED giúp UI biểu diễn tình trạng mà không tự mở thêm connection. Tránh tạo một EventSource cho từng widget; một connection có thể phân phối nhiều loại event bằng trường event:.

4. Reconnect không đồng nghĩa không mất dữ liệu

Browser tự kết nối lại và gửi Last-Event-ID khi đã nhận event có id. Nhưng server phải tự quyết định replay từ đâu. Nếu server chỉ broadcast dữ liệu đang sống trong RAM, event phát lúc client mất mạng sẽ biến mất.

Mô hình đáng tin cậy hơn:

1. Mỗi event có ID tăng đơn điệu hoặc cursor ổn định trong một stream.
2. Event được lưu trong log/stream có retention hữu hạn.
3. Khi reconnect, server đọc Last-Event-ID và replay phần sau cursor.
4. Sau khi bắt kịp, connection chuyển sang live tail mà không tạo khoảng trống.
5. Client xử lý idempotent và bỏ ID đã áp dụng vì duplicate vẫn có thể xảy ra.

SSE transport thường nên được thiết kế theo at-least-once, không hứa exactly-once. Event ID + deduplication dễ kiểm chứng hơn một lời hứa “không bao giờ lặp”.

5. Nếu cursor quá cũ thì làm gì?

Retention có giới hạn, nên một client offline nhiều ngày có thể yêu cầu ID đã bị xóa. Đừng âm thầm nhảy tới event mới nhất vì UI có thể thiếu trạng thái quan trọng. Endpoint nên có contract rõ:

• Gửi event resync-required rồi đóng stream để client tải snapshot mới.
• Hoặc trả một lỗi ứng dụng có thể nhận diện trước khi bắt đầu stream.
• Sau khi tải snapshot, client mở stream từ cursor gắn với snapshot đó.

Đừng dùng status 204 cho lỗi tạm thời: theo chuẩn EventSource, HTTP 204 yêu cầu user agent ngừng reconnect. Nó phù hợp khi server chủ động kết thúc vĩnh viễn subscription.

6. Server Node.js có heartbeat và backpressure

import http from 'node:http';

function encodeEvent({ id, event, data, retry }) {
  let output = '';
  if (retry !== undefined) output += `retry: ${retry}\n`;
  if (id !== undefined) output += `id: ${id}\n`;
  if (event) output += `event: ${event}\n`;
  for (const line of JSON.stringify(data).split('\n')) {
    output += `data: ${line}\n`;
  }
  return output + '\n';
}

http.createServer(async (req, res) => {
  if (req.url !== '/api/events') {
    res.writeHead(404).end();
    return;
  }

  const user = await authenticate(req);
  if (!user) {
    res.writeHead(401).end();
    return;
  }

  res.writeHead(200, {
    'Content-Type': 'text/event-stream; charset=utf-8',
    'Cache-Control': 'no-cache, no-transform',
    'X-Accel-Buffering': 'no'
  });
  res.flushHeaders();

  const cursor = req.headers['last-event-id'] ?? null;
  const subscription = await subscribeWithReplay(user.id, cursor);

  const send = async (event) => {
    const frame = encodeEvent(event);
    if (!res.write(frame)) {
      await new Promise((resolve) => res.once('drain', resolve));
    }
  };

  const heartbeat = setInterval(() => {
    res.write(': heartbeat\n\n');
  }, 15_000);

  let closed = false;
  const cleanup = async () => {
    if (closed) return;
    closed = true;
    clearInterval(heartbeat);
    await subscription.close();
  };
  req.on('close', () => void cleanup());

  try {
    for await (const event of subscription) {
      if (closed) break;
      await send(event);
    }
  } finally {
    await cleanup();
    if (!res.writableEnded) res.end();
  }
}).listen(8080);

Ví dụ minh họa các nguyên tắc, không phải framework hoàn chỉnh. Production cần bảo đảm cleanup chỉ chạy một lần, hủy iterator ngay khi client disconnect, giới hạn thời gian auth/subscription và xử lý lỗi sau khi headers đã được gửi.

7. Backpressure là ranh giới bảo vệ bộ nhớ

response.write() trả false khi buffer phía server đã đầy. Nếu tiếp tục ghi bất chấp tín hiệu này, một client chậm có thể tích lũy lượng lớn dữ liệu trong RAM. Chờ event drain, đặt giới hạn hàng đợi theo connection và có policy khi consumer tụt quá xa:

• Ngắt connection để browser reconnect và replay từ durable cursor.
• Gộp event dạng trạng thái mới nhất nếu domain cho phép.
• Không gộp event nghiệp vụ bắt buộc xử lý từng bản ghi.
• Đo queue depth, bytes pending và thời gian từ khi publish tới khi gửi.

Backpressure phải lan tới broker/consumer; chỉ chuyển unbounded queue từ socket sang application không giải quyết vấn đề.

8. Heartbeat giữ đường truyền và đo connection sống

Proxy, load balancer và NAT thường đóng connection im lặng quá lâu. Gửi comment : heartbeat\n\n theo chu kỳ ngắn hơn idle timeout thấp nhất trên toàn đường truyền. Heartbeat không tạo message event ở client và không nên mang dữ liệu nhạy cảm.

Chu kỳ 15–30 giây thường là điểm bắt đầu để test, không phải con số mặc định cho mọi hạ tầng. Heartbeat quá dày làm tăng byte và wake-up; quá thưa không ngăn được timeout. Theo dõi reconnect rate để hiệu chỉnh bằng dữ liệu thật.

9. Nginx phải stream thay vì gom buffer

location /api/events {
    proxy_pass http://sse_backend;
    proxy_http_version 1.1;

    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 75s;

    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
}

Khi proxy buffering bật, Nginx có thể đọc upstream nhanh rồi chỉ đẩy xuống khi buffer đủ, biến realtime thành từng cụm. Có thể cấu hình proxy_buffering off ở location hoặc để ứng dụng gửi X-Accel-Buffering: no khi hạ tầng cho phép. proxy_read_timeout phải dài hơn heartbeat và cần đối chiếu timeout ở CDN/load balancer phía trước.

Không nén một cách mù quáng: compressor hoặc middleware có thể buffer các frame nhỏ. Kiểm thử thời điểm từng event xuất hiện ở client, không chỉ kiểm tra response cuối có status 200.

10. Authentication, CORS và dữ liệu nhạy cảm

Native EventSource không cung cấp API tùy ý đặt Authorization header. Các lựa chọn phổ biến:

• Cookie phiên HttpOnly/Secure cho same-origin; endpoint GET không được tạo side effect.
• Cross-origin với withCredentials: true, CORS origin cụ thể và credential policy đúng.
• Token kết nối ngắn hạn, dùng một lần, đổi từ một request đã xác thực; tránh token dài hạn trong query vì URL có thể vào log/history.
• Fetch streaming tùy biến nếu bắt buộc custom headers, đổi lại phải tự implement parser và reconnect.

Authorization phải kiểm tra cả lúc mở connection và trong cách subscription được xây dựng. Khi quyền bị thu hồi, hệ thống cần cơ chế đóng stream hoặc lọc event ngay; không giữ authorization snapshot vô thời hạn.

11. Scale ngang mà không phụ thuộc RAM của một node

Với nhiều instance, event producer không biết client đang nối vào node nào. Dùng broker/pub-sub hoặc durable stream làm lớp phân phối, còn mỗi SSE node quản lý connection cục bộ. Replay store giữ event theo tenant/channel và cursor.

Sticky session có thể giảm churn nhưng không thay thế durability: deploy, crash hoặc rebalance vẫn chuyển client sang node khác. Mỗi connection tiêu tốn file descriptor, memory và scheduler work; cần capacity model cho concurrent connections, heartbeat rate, fan-out và reconnect storm.

Đừng đăng ký một broker consumer độc lập cho mỗi browser nếu broker không được thiết kế cho cardinality đó. Có thể multiplex theo channel/tenant tại node rồi fan-out có giới hạn tới các connection đã authorize.

12. Graceful shutdown khi deploy

1. Đánh dấu instance không ready để load balancer ngừng gửi connection mới.
2. Gửi event control tùy chọn báo client reconnect sớm, không chứa logic bắt buộc.
3. Dừng nhận event mới hoặc chốt cursor đã gửi cho từng connection.
4. Cho một cửa sổ drain ngắn, sau đó đóng các response còn lại.
5. Client reconnect sang instance khỏe và replay từ Last-Event-ID.

Random jitter ở client/retry giúp tránh hàng chục nghìn connection quay lại cùng mili giây. Deployment thành công phải được kiểm chứng bằng tỷ lệ reconnect, duplicate, gap và latency, không chỉ bằng việc process thoát sạch.

13. Kiểm thử bằng curl và browser

# -N/--no-buffer để nhìn frame ngay khi nhận
curl -N -H "Accept: text/event-stream" \
  https://example.com/api/events

# Mô phỏng reconnect từ cursor
curl -N -H "Last-Event-ID: 18452" \
  https://example.com/api/events

Kịch bản cần có:

• Event một dòng, nhiều dòng, named event, Unicode và payload lớn hợp lý.
• Mất mạng giữa event, reconnect với ID cuối và dedupe duplicate.
• Cursor còn retention, cursor quá cũ và snapshot/resync.
• Proxy/CDN buffering, idle timeout và heartbeat.
• Client chậm, buffer đầy, disconnect trước/sau drain.
• Token/cookie hết hạn, thu hồi quyền và cross-origin CORS.
• Rolling deploy nhiều instance và reconnect storm.

14. Observability cần theo connection và event

• Số connection đang mở theo instance, tenant và channel.
• Connection open rate, duration, close reason và reconnect rate.
• Event publish-to-send latency, bytes, queue depth và backpressure count.
• Replay count, replay lag, cursor-too-old và resync rate.
• Duplicate/gap được client hoặc integration test phát hiện.
• Heartbeat sent, proxy timeout, auth failure và broker lag.

Không log toàn payload nếu event chứa dữ liệu cá nhân. Log event ID, type, tenant/channel đã giảm nhạy cảm, connection ID và timing thường đủ để dựng timeline.

15. Checklist production

• Endpoint trả đúng text/event-stream, UTF-8 và mỗi event kết thúc bằng dòng trống.
• Mọi event cần replay có ID ổn định; client xử lý idempotent.
• Last-Event-ID, retention hết hạn và snapshot/resync có contract rõ.
• Heartbeat ngắn hơn idle timeout thấp nhất trên đường truyền.
• Proxy/CDN không buffer hoặc cache stream; compression đã được kiểm thử.
• Server tôn trọng backpressure và giới hạn queue mỗi connection.
• Auth, CORS, token/cookie và thu hồi quyền đã được threat-model.
• Broker/replay store cho phép reconnect sang instance khác.
• Graceful shutdown và jitter tránh reconnect storm khi deploy.
• Dashboard đo connection, replay, gap, duplicate, lag và memory/file descriptors.

SSE production không phải một vòng lặp gọi res.write(). Giá trị của nó đến từ một giao thức HTTP đơn giản kết hợp với discipline về cursor, replay, backpressure và lifecycle. Khi server có thể chứng minh client reconnect không mất trạng thái, proxy truyền từng frame ngay và consumer chậm không làm đầy RAM, SSE trở thành lựa chọn realtime rất hiệu quả.