Học từ Facebook #48: Snapshot + push, giữ event chờ snapshot và watchdog

Bạn bấm "Đăng" 50 quảng cáo trong Ads Manager. Server xử lý từng cái, mỗi cái xong lại đẩy một event qua kênh realtime để thanh tiến độ chạy. Rồi bạn tải lại trang khi đã xong 10 cái. Trang mới phải làm hai việc:

1. Subscribe kênh push để nhận các event tiếp theo.
2. Gọi REST lấy trạng thái hiện tại (snapshot): đã xong những cái nào.

Hai việc chạy song song, và đây là chỗ dễ sai. REST mất 900 ms mới về, nhưng server đọc trạng thái ngay lúc request tới (ví dụ ở mục thứ 11). Trong 900 ms đó, push đã kịp báo mục 12, 13… 25 xong. Nếu client áp các event đó ngay, rồi snapshot về ghi đè lên, thì 12–25 biến mất khỏi màn hình. Thanh tiến độ đứng ở 36/50 mãi mãi.

▶ Mở demo toàn màn hình ↗ · Sửa code trên StackBlitz ↗ · Xem source trên GitHub ↗

#Facebook làm thế nào?

Hai module trong 64.js: AdsDraftPublishRealtimeConnection (lớp kênh push) và AdsDraftPublishRealtimeClient (lớp nghe event, cập nhật store Flux của Ads Manager). Bên dưới là AdsDraftRealtimeSubscription: một Relay GraphQL subscription theo topic gqls/ad_draft_publish_subscribe/ad_draft_id_<id>, mỗi message là một chuỗi JSON.

#Hàng đợi nằm ở lớp kết nối

let pending = null;  // null = không buffer

function onMessage(message, accountID, draftID) {
  const { event, ...payload } = message;
  const emit = () => {
    connection.emit(event, payload, accountID, draftID);
    connection.emit('all', event, payload, accountID, draftID);
  };
  pending ? pending.push(emit) : emit();   // đang buffer thì giữ lại cả closure
}

function startBuffering() { pending = []; }
function stopBuffering()  { pending && pending.forEach((f) => f()); pending = null; }
function discardBuffer()  { if (pending) pending = []; }

Mỗi event được gói thành một closure. Phát lại là gọi lần lượt các closure, nên listener nhận đúng như thể event vừa tới.

#Lớp client quyết định khi nào buffer

export function bufferUntilPublishStatusFetched() {
  Connection.startBuffering();
  isBuffering = true;
}

function handleDispatch({ action }) {
  switch (action.type) {
    case CurrentDraftLoaded: case SetCurrentDraftSuccess:
      Connection.subscribe(action.accountID, action.draftID);
      break;
    case PublishInProgress:
      if (isBuffering && action.response) {    // snapshot đã về
        Connection.stopBuffering();
        isBuffering = false;
      }
      resetWatchdog();
      break;
    case PublishCompleted: case PublishFailed:
      clearWatchdog();
      break;
  }
}

Module export hàm bufferUntilPublishStatusFetched; nơi gọi nó không có trong bản dump, nhưng đúng như tên, nó phải được gọi trước khi gửi request lấy trạng thái publish. Khi action PublishInProgress mang theo response (tức snapshot), client mở van cho hàng đợi chảy ra.

#Áp event: idempotent theo ad_object_id

Event FRAGMENT_COMPLETED (một mục đã đăng xong) được trộn vào snapshot:

const statuses = getByPath(publishStatus, FRAGMENT_STATUSES.DATA);
const index = statuses.findIndex((s) => s.ad_object_id === event.fragment.ad_object_id);
index === -1 ? statuses.push(event.fragment) : Object.assign(statuses[index], event.fragment);

Tìm theo id rồi ghi đè, không có thì thêm. Áp cùng một event hai lần cho ra cùng kết quả. Đây là lý do phát lại hàng đợi an toàn: event nào đã có trong snapshot thì áp lại cũng không sao.

#Watchdog 30 giây

const STALL_MS = 3e4;
function resetWatchdog() {
  clearWatchdog();
  timer = setTimeout(() => {
    debugLog('Push might be stalled, getting status from ads API');
    const draftID = PublishUIStore.getState().publishingDraftID;
    draftID && RealtimePublishStalledAction.dispatch({ draftID, ancestorIDs });
  }, STALL_MS);
}

Mỗi action PublishInProgress đặt lại đồng hồ (event FRAGMENT_COMPLETED cũng dispatch một action tiến độ sau khi trộn vào snapshot). Kênh push có thể "chết êm": socket vẫn mở nhưng không còn message nào, hoặc proxy giữa đường cắt kết nối mà client không biết. 30 giây không nghe gì thì client tự hỏi REST.

#Gom event xoá

let deleted = new Map();
const flushDeleted = debounce(() => {
  FragmentBatchDeletedAction.dispatch({ fragments: deleted });
  deleted = new Map();
}, 100);

function handleFragmentDeleted(fragment) {
  deleted.set(fragment.id, fragment);
  flushDeleted();
}

Xoá 200 mục thì server đẩy 200 event gần như cùng lúc. Mỗi event một lần dispatch là 200 lần store đổi, 200 lần re-render bảng. Gom vào Map (theo id, nên event trùng tự gộp) và dispatch một lần sau 100 ms im lặng.

#Vài chi tiết phòng thủ

• Đúng thứ tự: nhận FRAGMENT_COMPLETED trước PUBLISH_STARTED hoặc trước FRAGMENT_STARTED thì log cảnh báo. Không chặn, chỉ ghi để biết server có gửi lệch thứ tự không.
• Tab khác đang đăng: event của một draft không phải draft tab này đang đăng (publishingDraftID !== draftID) thì chỉ đồng bộ phần xoá, không đụng UI tiến độ. Comment trong code: "This could happen if the publish is taking place in a different session / browser tab."
• Một subscription mỗi tài khoản: subscribe sang draft khác thì tự huỷ subscription cũ.

#Chỗ còn hở

stopBuffering chỉ chạy khi PublishInProgress có response. Mình không thấy nhánh nào xử lý trường hợp request lấy trạng thái bị lỗi: khi đó cờ isBuffering có lẽ bật mãi và event cứ dồn vào mảng. discardBuffer() có sẵn ở lớp kết nối nhưng không có chỗ gọi trong bản dump. Bản viết lại thêm nhánh lỗi: giữ hàng đợi, hẹn giờ thử lại REST.

#Bên trong: thứ tự sự kiện

sequenceDiagram
  participant C as Client
  participant P as Kênh push
  participant R as REST
  participant S as Server
  C->>P: subscribe
  C->>C: startBuffering
  C->>R: GET trạng thái
  R->>S: đọc ở mục 11
  S-->>P: event mục 12, 13, …, 25
  P-->>C: xếp hàng 14 event
  R-->>C: snapshot 11 mục, sau 900 ms
  C->>C: state = snapshot, phát lại 14 event
  S-->>P: event mục 26 …
  P-->>C: áp ngay
  Note over C: 30 s không có event → hỏi REST lại

Mấu chốt là subscribe trước, đọc snapshot sau. Làm ngược lại (đọc snapshot rồi mới subscribe) thì event phát ra giữa hai bước rơi mất, không có cách nào lấy lại. Subscribe trước thì event có thể tới sớm hơn snapshot, nhưng hàng đợi giữ được chúng.

Cách này giống mẫu "snapshot + replay" của các hệ thống sự kiện: đọc trạng thái tại một mốc, rồi phát lại các thay đổi sau mốc đó. Khác ở chỗ Facebook không biết chính xác mốc của snapshot, nên phát lại mọi event đã giữ và dựa vào tính idempotent.

#Viết lại

Bản viết lại gói toàn bộ vào một hàm createSnapshotSync({ fetchSnapshot, apply, onState, watchdogMs }), không phụ thuộc Flux hay GraphQL. Phần lõi: watchdog, áp event, lấy snapshot và phát lại.

  const armWatchdog = () => {
    if (watchdog) clearTimeout(watchdog);
    watchdog = null;
    if (!watchdogMs || stopped) return;
    watchdog = setTimeout(() => {
      onLog(`Không có push trong ${watchdogMs} ms, lấy trạng thái qua REST`);
      onStall?.();
      void refresh();
    }, watchdogMs);
  };

  const applyOne = (event: E) => {
    if (!state) return;
    // event cũ hơn snapshot thì đã nằm trong snapshot rồi. Chỉ so với mốc của SNAPSHOT,
    // không so với event vừa áp: push có thể tới lệch thứ tự (#12 tới sau #13) mà vẫn là event mới.
    if (event.seq <= snapshotSeq) {
      onLog(`Bỏ event #${event.seq} (snapshot đã có tới #${snapshotSeq})`);
      return;
    }
    state = apply(state, event);
    onState(state);
  };

  async function refresh() {
    const id = ++fetchId;
    buffer ??= [];
    try {
      const snapshot = await fetchSnapshot();
      if (id !== fetchId || stopped) return; // đã có lần refresh mới hơn
      state = snapshot;
      snapshotSeq = snapshot.seq;
      onState(state);
      const pending = buffer;
      buffer = null;
      onLog(`Snapshot #${snapshot.seq} về, phát lại ${pending.length} event đã giữ`);
      pending.forEach(applyOne);
      armWatchdog();
    } catch (error) {
      if (id !== fetchId || stopped) return;
      // bản gốc không có nhánh này: REST lỗi thì cờ buffering bật mãi, không ai lấy lại snapshot
      onLog(`Lấy snapshot lỗi: ${(error as Error).message}. Vẫn giữ ${buffer?.length ?? 0} event, thử lại sau ${retryMs} ms`);
      if (retry) clearTimeout(retry);
      retry = setTimeout(() => void refresh(), retryMs);
    }
  }

Gom event xoá là createBatchedMap(ms, flush): một debounce mang theo Map, xem trong file đầy đủ.

/**
 * Snapshot + delta qua kênh push (ý tưởng từ AdsDraftPublishRealtimeClient/Connection của
 * Facebook Ads Manager).
 *
 * Vấn đề: client vừa subscribe kênh push vừa gọi REST lấy trạng thái hiện tại (snapshot).
 * Event push nào tới TRƯỚC snapshot mà áp vào ngay thì sẽ bị snapshot ghi đè và mất.
 *
 * Cách làm:
 * - startBuffering() trước khi gọi REST: mọi event push đi vào hàng đợi.
 * - Snapshot về: lấy làm trạng thái, rồi phát lại hàng đợi theo thứ tự (stopBuffering).
 * - Bước áp event phải idempotent: event đã nằm trong snapshot thì áp lại không đổi gì.
 * - Watchdog: mỗi event đặt lại hẹn giờ; im lặng quá lâu thì coi như push bị kẹt và hỏi REST.
 */

export interface Versioned {
  /** số thứ tự tăng dần do server gán: snapshot mang seq của event cuối nó đã gộp vào */
  seq: number;
}

export interface SnapshotSyncOptions<S extends Versioned, E extends Versioned> {
  fetchSnapshot: () => Promise<S>;
  /** áp một event vào state, PHẢI idempotent */
  apply: (state: S, event: E) => S;
  onState: (state: S) => void;
  /** im lặng quá lâu thì gọi onStall rồi tự lấy lại snapshot qua REST; 0 = tắt */
  watchdogMs?: number;
  onStall?: () => void;
  /** REST lỗi thì thử lại sau bao lâu (event push vẫn được giữ trong lúc chờ) */
  retryMs?: number;
  onLog?: (message: string) => void;
}

export function createSnapshotSync<S extends Versioned, E extends Versioned>({
  fetchSnapshot,
  apply,
  onState,
  watchdogMs = 30_000,
  onStall,
  retryMs = 1000,
  onLog = () => {},
}: SnapshotSyncOptions<S, E>) {
  let state: S | null = null;
  /** seq của snapshot gần nhất: event có seq <= mốc này đã nằm sẵn trong snapshot */
  let snapshotSeq = -Infinity;
  let buffer: E[] | null = null;
  let watchdog: ReturnType<typeof setTimeout> | null = null;
  let retry: ReturnType<typeof setTimeout> | null = null;
  let fetchId = 0;
  let stopped = false;

  const armWatchdog = () => {
    if (watchdog) clearTimeout(watchdog);
    watchdog = null;
    if (!watchdogMs || stopped) return;
    watchdog = setTimeout(() => {
      onLog(`Không có push trong ${watchdogMs} ms, lấy trạng thái qua REST`);
      onStall?.();
      void refresh();
    }, watchdogMs);
  };

  const applyOne = (event: E) => {
    if (!state) return;
    // event cũ hơn snapshot thì đã nằm trong snapshot rồi. Chỉ so với mốc của SNAPSHOT,
    // không so với event vừa áp: push có thể tới lệch thứ tự (#12 tới sau #13) mà vẫn là event mới.
    if (event.seq <= snapshotSeq) {
      onLog(`Bỏ event #${event.seq} (snapshot đã có tới #${snapshotSeq})`);
      return;
    }
    state = apply(state, event);
    onState(state);
  };

  async function refresh() {
    const id = ++fetchId;
    buffer ??= [];
    try {
      const snapshot = await fetchSnapshot();
      if (id !== fetchId || stopped) return; // đã có lần refresh mới hơn
      state = snapshot;
      snapshotSeq = snapshot.seq;
      onState(state);
      const pending = buffer;
      buffer = null;
      onLog(`Snapshot #${snapshot.seq} về, phát lại ${pending.length} event đã giữ`);
      pending.forEach(applyOne);
      armWatchdog();
    } catch (error) {
      if (id !== fetchId || stopped) return;
      // bản gốc không có nhánh này: REST lỗi thì cờ buffering bật mãi, không ai lấy lại snapshot
      onLog(`Lấy snapshot lỗi: ${(error as Error).message}. Vẫn giữ ${buffer?.length ?? 0} event, thử lại sau ${retryMs} ms`);
      if (retry) clearTimeout(retry);
      retry = setTimeout(() => void refresh(), retryMs);
    }
  }

  return {
    /** gọi khi subscribe xong (hoặc cùng lúc): bắt đầu giữ event rồi lấy snapshot */
    start: () => refresh(),
    /** nơi kênh push đưa event vào */
    receive(event: E) {
      if (stopped) return;
      armWatchdog();
      if (buffer) {
        buffer.push(event);
        return;
      }
      applyOne(event);
    },
    refresh,
    getState: () => state,
    isBuffering: () => buffer !== null,
    stop() {
      stopped = true;
      if (watchdog) clearTimeout(watchdog);
      if (retry) clearTimeout(retry);
    },
  };
}

/**
 * Gom nhiều lần gọi trong `ms` thành một lần, dữ liệu dồn vào Map theo id (bản gốc dùng cho
 * các event "đã xoá fragment" để dispatch một lần thay vì mỗi event một lần re-render).
 */
export function createBatchedMap<K, V>(ms: number, flush: (items: Map<K, V>) => void) {
  let items = new Map<K, V>();
  let timer: ReturnType<typeof setTimeout> | null = null;
  return {
    add(key: K, value: V) {
      items.set(key, value);
      if (timer) clearTimeout(timer);
      timer = setTimeout(() => {
        timer = null;
        const batch = items;
        items = new Map();
        flush(batch);
      }, ms);
    },
    cancel() {
      if (timer) clearTimeout(timer);
      timer = null;
      items = new Map();
    },
  };
}

Khác bản gốc:

• Có số thứ tự (seq): snapshot mang seq của event cuối nó đã gộp. Event có seq nhỏ hơn hoặc bằng mốc đó bị bỏ, không áp lại. Vẫn cần apply idempotent, vì push có thể tới lệch thứ tự.
• REST lỗi thì giữ hàng đợi và thử lại sau retryMs, thay vì kẹt ở trạng thái buffer.
• Watchdog đặt lại theo event push, không theo action của store; tắt hẳn khi stop().
• Mỗi lần refresh có một id; response của lần refresh cũ về muộn thì bị bỏ.

#Đi qua code

• buffer ??= [] ở đầu refresh: lần đầu bật buffer; lần refresh do watchdog thì tiếp tục giữ event mới cho tới khi snapshot mới về.
• applyOne so event.seq <= snapshotSeq, không so với state.seq. Lần đầu viết mình so với state.seq và demo mất 2 mục: event #13 tới trước #12, áp #13 xong thì #12 bị coi là cũ. Chỉ mốc của snapshot mới cho biết "đã có sẵn".
• apply trong demo thêm id vào mảng nếu chưa có và lấy max cho seq, nên áp hai lần hay áp lệch thứ tự đều ra cùng state.
• createBatchedMap là debounce có dữ liệu: mỗi add đặt lại hẹn giờ, flush trả về cả Map rồi tạo Map mới.

#Tại sao không làm cách khác?

Không buffer, chỉ lấy snapshot rồi mới subscribe? Event giữa hai bước mất hẳn. Thử "Ngây thơ" trong demo: 12 ô đỏ là các mục client không bao giờ biết đã xong.

Bỏ event cho tới khi snapshot về? Cũng mất các event trong khoảng thời gian REST chạy, chỉ khác là mất trước thay vì bị ghi đè.

Chỉ poll REST, bỏ push? Đúng nhưng chậm và tốn: poll mỗi giây với 50 mục là 50 request cho một lần đăng. Push cho cập nhật gần như tức thì; REST chỉ dùng lúc mở trang và lúc push kẹt.

Server gửi kèm số thứ tự để client tự xin event còn thiếu? Tốt hơn nữa (Facebook làm vậy cho tin nhắn Messenger, với seqID), nhưng đòi server lưu lịch sử event. Cho thanh tiến độ vài phút thì snapshot + replay là đủ.

#Edge case và cạm bẫy

• Snapshot về chậm hơn watchdog: watchdog chỉ được bật sau khi có snapshot, nên không gọi REST chồng lên REST.
• Push kẹt hẳn: bật "push kẹt từ mục #31" và tắt watchdog, thanh tiến độ đứng ở 30/50. Bật watchdog lại thì sau 2 giây (bản gốc 30 giây) client lấy snapshot mới và đủ 50/50.
• Hàng đợi phình to: nếu REST lỗi kéo dài, hàng đợi tăng theo số event. Với tiến độ đăng thì vài trăm phần tử là cùng; với luồng event lớn hơn nên đặt giới hạn và bỏ hàng đợi (lấy snapshot mới là đủ).
• Đăng xong rồi vẫn còn watchdog: phải tắt khi nhận trạng thái hoàn tất, không thì cứ 30 giây lại hỏi REST cho một việc đã xong.
• Nhiều tab: tab không phải tab bấm "Đăng" vẫn nhận event. Quyết định tab nào được cập nhật UI nào là logic riêng của ứng dụng, bản gốc dựa vào publishingDraftID.

#Dùng trong dự án

const sync = createSnapshotSync<JobState, JobEvent>({
  fetchSnapshot: () => fetch(`/api/jobs/${id}`).then((r) => r.json()),
  apply: (state, e) => ({ seq: Math.max(state.seq, e.seq), done: union(state.done, [e.itemId]) }),
  onState: (state) => store.setJob(id, state),
  watchdogMs: 30_000,
});

const socket = new WebSocket(`wss://example.com/jobs/${id}`);
socket.addEventListener('message', (m) => sync.receive(JSON.parse(m.data)));
socket.addEventListener('open', () => void sync.start()); // subscribe xong mới lấy snapshot

Với Server-Sent Events thì header Last-Event-ID cho phép trình duyệt tự xin lại event bị lỡ sau khi kết nối lại, nếu server hỗ trợ. Nó giải quyết chuyện mất event khi rớt mạng, còn chuyện "event tới trước snapshot lúc mở trang" vẫn cần hàng đợi như trên.

#Kiểm thử

import { describe, expect, it, vi } from 'vitest';
import { createBatchedMap, createSnapshotSync } from './realtimeSnapshot';

type S = { seq: number; done: number[] };
type E = { seq: number; id: number };
const apply = (s: S, e: E): S => ({ seq: Math.max(s.seq, e.seq), done: s.done.includes(e.id) ? s.done : [...s.done, e.id] });

describe('createSnapshotSync', () => {
  it('giữ event tới trước snapshot rồi phát lại, bỏ event đã có', async () => {
    let resolve!: (s: S) => void;
    const states: S[] = [];
    const sync = createSnapshotSync<S, E>({
      fetchSnapshot: () => new Promise((r) => (resolve = r)),
      apply, onState: (s) => states.push(s), watchdogMs: 0,
    });
    const started = sync.start();
    sync.receive({ seq: 2, id: 2 });  // đã có trong snapshot
    sync.receive({ seq: 4, id: 4 });  // tới lệch thứ tự
    sync.receive({ seq: 3, id: 3 });
    expect(sync.isBuffering()).toBe(true);
    resolve({ seq: 2, done: [1, 2] });
    await started;
    expect(sync.getState()).toEqual({ seq: 4, done: [1, 2, 4, 3] });
  });

  it('REST lỗi thì giữ hàng đợi và thử lại', async () => {
    vi.useFakeTimers();
    let calls = 0;
    const sync = createSnapshotSync<S, E>({
      fetchSnapshot: async () => { if (++calls === 1) throw new Error('503'); return { seq: 0, done: [] }; },
      apply, onState: () => {}, watchdogMs: 0, retryMs: 1000,
    });
    await sync.start();
    sync.receive({ seq: 1, id: 1 });
    expect(sync.isBuffering()).toBe(true);
    await vi.advanceTimersByTimeAsync(1000);
    expect(sync.getState()).toEqual({ seq: 1, done: [1] });
    vi.useRealTimers();
  });

  it('watchdog hỏi lại REST khi im lặng', async () => {
    vi.useFakeTimers();
    const fetchSnapshot = vi.fn(async () => ({ seq: 0, done: [] }));
    const onStall = vi.fn();
    const sync = createSnapshotSync<S, E>({ fetchSnapshot, apply, onState: () => {}, watchdogMs: 30_000, onStall });
    await sync.start();
    await vi.advanceTimersByTimeAsync(29_000);
    sync.receive({ seq: 1, id: 1 });          // đặt lại đồng hồ
    await vi.advanceTimersByTimeAsync(29_000);
    expect(onStall).not.toHaveBeenCalled();
    await vi.advanceTimersByTimeAsync(1_000);
    expect(onStall).toHaveBeenCalledOnce();
    expect(fetchSnapshot).toHaveBeenCalledTimes(2);
    sync.stop();
    vi.useRealTimers();
  });
});

describe('createBatchedMap', () => {
  it('12 lần add sát nhau thành 1 lần flush', () => {
    vi.useFakeTimers();
    const flush = vi.fn();
    const batch = createBatchedMap<number, number>(100, flush);
    for (let i = 0; i < 12; i++) { batch.add(i, i); vi.advanceTimersByTime(8); }
    vi.advanceTimersByTime(100);
    expect(flush).toHaveBeenCalledOnce();
    expect(flush.mock.calls[0][0].size).toBe(12);
    vi.useRealTimers();
  });
});

#So với thư viện và chuẩn web

• Apollo Client / Relay subscriptions không tự giải quyết chuyện "event tới trước query": bạn tự quyết thứ tự query và subscribe. Mẫu subscribeToMore của Apollo áp event lên kết quả query đã có; event tới trước khi query xong thì tuỳ cách bạn viết updateQuery.
• Server-Sent Events có Last-Event-ID để nối lại đúng chỗ khi kết nối lại; WebSocket thì không có cơ chế tương tự, phải tự làm ở tầng ứng dụng.
• Firebase/Firestore onSnapshot gộp cả snapshot lẫn thay đổi vào một luồng, nên client không phải tự ghép. Nếu backend của bạn tự viết, hàm ở trên là phần ghép tối thiểu.

Bài liên quan: Học từ Facebook #38: đồng bộ tin nhắn, phát hiện lỗ seqID, Học từ Facebook #5: network status và heartbeat, Học từ Telegram #4: gom update realtime.

Bài này thuộc series Học từ mã nguồn Facebook (106 bài, đọc theo thứ tự).

Code + demo: examples/facebook-patterns · Mở trên StackBlitz