Học từ Facebook #41: Điều phối request bảy chế độ gộp và chống trùng

Trong một ứng dụng chat, người dùng bấm liên tục: lưu trữ năm cuộc trò chuyện liền nhau, đánh dấu đã đọc, mở thread rồi mở thread khác, kéo danh sách xuống. Mỗi thao tác là một lệnh gửi lên server. Gửi thẳng từng lệnh thì:

• "Lưu trữ thread 1", "lưu trữ thread 2"… thành năm request, trong khi server nhận được cả danh sách trong một lần.
• "Đánh dấu đã xem" bấm năm lần chỉ cần lần cuối cùng.
• "Tải danh sách thread trang 2" bấm hai lần (double click, effect chạy lại) thành hai request y hệt nhau.
• Lệnh lỗi vì mạng chập chờn thì mất luôn, trừ khi ai đó nhớ gửi lại.

Mỗi loại lệnh cần một cách gộp khác nhau. Messenger web gom chúng vào một chỗ: MercuryServerDispatcher.

▶ 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?

MercuryServerDispatcher nằm trong bundle 55.js. MercuryServerRequests đăng ký các endpoint một lần qua registerEndpoints({ [uri]: { mode, batch_function, handler, … } }). Một số cặp thật:

"Đã xem" giữ bản mới nhất ((a, b) => b). Lưu trữ thread trộn tập ID của các lệnh lại. Tải danh sách thread thì bỏ trùng. Tải thông tin thread gom mọi lệnh phát ra trong cùng một tick. Ba chế độ IDEMPOTENT, BATCH_CONDITIONAL, BATCH_SUCCESSIVE_PIGGYBACK_ON_ERROR có trong dispatcher nhưng mình không tìm thấy endpoint nào dùng chúng trong bản dump.

Endpoint được lưu theo ba tầng: URL → user ID → endpoint_id. Facebook cho một tab quản lý cả tài khoản cá nhân lẫn Page, nên cùng một URL có thể có nhiều hàng đợi riêng. Hàng đợi gửi tin (bài #31) cũng đăng ký qua đây, ở chế độ IMMEDIATE, với endpoint_id là ID hàng đợi.

#Một endpoint

Mỗi endpoint giữ _mode, _combineData (batch_function), _combineDataIf (batch_if), _inFlight, mảng _batches (thực tế tối đa 1 phần tử) và _deferredSend = debounce(_batchSend, 0). Phần quyết định:

trySend(data, mode = this._mode) {
  if (killedByServer) return;             // server trả kill_chat: tắt mọi request chat
  if (mode == IMMEDIATE) this._send(data);
  else if (mode == IDEMPOTENT) this._inFlight || this._send(data);
  else if (mode == BATCH_SUCCESSIVE || mode == BATCH_SUCCESSIVE_UNIQUE)
    this._inFlight ? this._batchData(data) : this._send(data);
  else if (mode == BATCH_CONDITIONAL) {
    const queued = this._batches[0]?.getData();
    this._inFlight && (this._combineDataIf(this._pendingRequestData, data) || this._combineDataIf(queued, data))
      ? this._batchData(data) : this._send(data);
  } else if (mode == BATCH_DEFERRED_MULTI) { this._batchData(data); this._deferredSend(); }
  else if (mode == BATCH_SUCCESSIVE_PIGGYBACK_ON_ERROR) { this._batchData(data); this._inFlight || this._batchSend(); }
}

_batchData trộn dữ liệu mới vào gói chờ cuối cùng bằng combineWith (hoặc combineWithOlder khi cần đặt dữ liệu cũ lên trước). Ở chế độ _UNIQUE, nó bỏ qua nếu dữ liệu mới giống hệt _pendingRequestData. Phép so sánh là JSON.stringify(a) == JSON.stringify(b) (module areJSONRepresentationsEqual), nên phụ thuộc thứ tự key.

#Khi request kết thúc

_handleResponse(response) {
  this._inFlight--;
  killedByServer = response.getPayload()?.kill_chat;
  payload?.error_payload ? this._errorHandler?.(response) : this._handler?.(payload, request);
  if (mode is one of BATCH_SUCCESSIVE, _UNIQUE, _PIGGYBACK_ON_ERROR, BATCH_CONDITIONAL) this._batchSend();
  delete this._pendingRequestData;          // ← sau _batchSend
}

_postErrorHandler(error) {                  // lỗi mạng, lỗi server, timeout
  this._inFlight--;
  recentErrors.push(record(error)); recentErrors = recentErrors.slice(-50);
  if (mode is BATCH_SUCCESSIVE, _UNIQUE, BATCH_CONDITIONAL) this._batchSend();
  else if (mode == BATCH_SUCCESSIVE_PIGGYBACK_ON_ERROR) {
    this._batchData(this._pendingRequestData, /* older */ true);
    if (had a queued batch) this._batchSend();       // không thì nằm chờ lệnh kế tiếp
  }
  delete this._pendingRequestData;          // ← cũng sau _batchSend
}

Chế độ "piggyback" có một ý hay: lệnh lỗi không tự thử lại (có thể lỗi tiếp ngay), mà được trộn trước dữ liệu của lệnh kế tiếp và đi cùng chuyến đó. Mạng hỏng hẳn thì không có vòng retry nào chạy vô ích; người dùng thao tác tiếp thì lệnh cũ được gửi lại kèm.

Mọi lỗi được giữ trong một mảng 50 phần tử gần nhất (getRequestErrors()), có lẽ để đính kèm vào báo cáo lỗi.

#Lỗi: xoá nhầm dữ liệu request đang bay

Nhìn hai dòng cuối của _handleResponse: _batchSend() gọi _send(gói chờ), mà _send đặt this._pendingRequestData = gói chờ. Ngay sau đó delete this._pendingRequestData xoá đúng giá trị vừa đặt. Request gói chờ đang bay mà endpoint "quên" nó chứa gì.

Hệ quả:

• BATCH_SUCCESSIVE_UNIQUE: chỉ bỏ trùng được với request đầu tiên của một chuỗi. Từ request thứ hai (gửi từ gói chờ) trở đi, yêu cầu trùng không còn bị bỏ. Demo: tải trang 1, xếp trang 2, trang 1 xong thì trang 2 đi, rồi yêu cầu trang 2 lần nữa. Kiểu Facebook gửi 3 request (trang 1, trang 2, trang 2), bản viết lại gửi 2.
• BATCH_CONDITIONAL: mất một nửa điều kiện so sánh với request đang bay.
• PIGGYBACK_ON_ERROR: nếu request đi từ gói chờ bị lỗi, this._pendingRequestData là undefined, và undefined bị trộn vào gói chờ.

Với endpoint thật đang dùng _UNIQUE (danh sách thread, thread chưa đọc), hậu quả chỉ là thỉnh thoảng có request thừa. Nhưng đây đúng loại lỗi khiến một cơ chế "bỏ trùng" trông như vẫn chạy trong test đơn giản và hỏng âm thầm trong thực tế.

sequenceDiagram
  participant U as UI
  participant E as Endpoint
  participant S as Server
  U->>E: trySend trang 1
  E->>S: gửi trang 1, pending = trang 1
  U->>E: trySend trang 2, đang bay nên vào gói chờ
  S-->>E: trang 1 xong
  E->>S: _batchSend, gửi trang 2, pending = trang 2
  Note over E: delete pending, pending = undefined
  U->>E: trySend trang 2
  Note over E: so với pending undefined, không trùng
  E->>E: vào gói chờ, sẽ gửi lại trang 2

#Viết lại

Phần quyết định theo chế độ:

    trySend(data: D) {
      switch (mode) {
        case 'immediate':
          return doSend(data);
        case 'idempotent':
          if (inFlight) return onEvent({ type: 'skipped', data, reason: 'in-flight' });
          return doSend(data);
        case 'successive':
          return inFlight ? addToBatch(data) : doSend(data);
        case 'successiveUnique':
          if (!inFlight) return doSend(data);
          if (pending !== undefined && isEqual(pending, data)) {
            return onEvent({ type: 'skipped', data, reason: 'duplicate' });
          }
          return addToBatch(data);
        case 'conditional':
          if (inFlight && ((pending !== undefined && combineIf(pending, data)) || (batch && combineIf(batch.data, data)))) {
            return addToBatch(data);
          }
          return doSend(data);
        case 'deferredMulti':
          addToBatch(data);
          clearTimeout(deferTimer);
          deferTimer = setTimeout(flushBatch, 0);
          return;
        case 'piggyback':
          addToBatch(data);
          if (!inFlight) flushBatch();
          return;
      }
    },

Phần kết thúc request, xoá pending của request vừa xong trước khi gửi gói chờ:

  function finish(data: D, failed: boolean) {
    inFlight--;
    onEvent({ type: failed ? 'error' : 'done', data });
    // Xoá pending của request VỪA XONG trước khi gửi gói kế tiếp.
    // Bản gốc làm ngược lại (gửi gói chờ rồi mới delete), nên xoá mất pending của request mới.
    if (!legacyClearPendingAfterFlush && pending === data) pending = undefined;
    if (failed && mode === 'piggyback') {
      // dữ liệu lỗi được xếp trước dữ liệu mới; có sẵn gói chờ thì gửi luôn,
      // không thì đợi lệnh kế tiếp rồi đi nhờ chuyến đó
      const hadBatch = batch !== undefined;
      addToBatch(data, true);
      if (hadBatch) flushBatch();
    } else if (mode === 'successive' || mode === 'successiveUnique' || mode === 'conditional' || mode === 'piggyback') {
      flushBatch();
    }
    if (legacyClearPendingAfterFlush) pending = undefined;
  }

/**
 * Bộ điều phối request theo endpoint với nhiều chế độ gộp / chống trùng
 * (ý tưởng từ MercuryServerDispatcher của Messenger web).
 *
 * Mỗi endpoint có: số request đang bay, dữ liệu của request đang bay (pending),
 * và tối đa MỘT gói chờ (batch) được trộn dần bằng hàm combine.
 *
 *  immediate        luôn gửi ngay
 *  idempotent       đang có request bay thì bỏ qua (lệnh lặp lại vô nghĩa)
 *  successive       đang bay thì gộp vào gói chờ, xong request cũ mới gửi gói chờ
 *  successiveUnique như successive nhưng bỏ dữ liệu y hệt request đang bay
 *  conditional      chỉ gộp khi combineIf cho phép, không thì gửi riêng ngay
 *  deferredMulti    luôn gộp, gửi ở cuối tick (gom mọi lệnh phát ra cùng lúc)
 *  piggyback        gộp; lỗi thì dữ liệu lỗi đi nhờ chuyến kế tiếp
 */

export type DispatchMode =
  | 'immediate'
  | 'idempotent'
  | 'successive'
  | 'successiveUnique'
  | 'conditional'
  | 'deferredMulti'
  | 'piggyback';

export interface EndpointConfig<D> {
  mode: DispatchMode;
  send: (data: D) => Promise<void>;
  /** trộn dữ liệu mới vào gói chờ: combine(cũ, mới) */
  combine?: (older: D, newer: D) => D;
  /** chế độ conditional: có được gộp `next` vào `current` không */
  combineIf?: (current: D, next: D) => boolean;
  /** so dữ liệu trùng cho successiveUnique (mặc định so JSON với key đã sắp) */
  isEqual?: (a: D, b: D) => boolean;
  onEvent?: (event: DispatchEvent<D>) => void;
  /** CHỈ để demo tái hiện lỗi của bản gốc: xoá pending SAU khi đã gửi gói chờ */
  legacyClearPendingAfterFlush?: boolean;
}

export type DispatchEvent<D> =
  | { type: 'send'; data: D }
  | { type: 'queued'; data: D }
  | { type: 'skipped'; data: D; reason: 'in-flight' | 'duplicate' }
  | { type: 'done'; data: D }
  | { type: 'error'; data: D };

/** JSON có sắp key, để {a,b} và {b,a} được coi là bằng nhau (bản gốc so JSON.stringify thô). */
export function stableJson(value: unknown): string {
  return JSON.stringify(value, (_key, v) =>
    v && typeof v === 'object' && !Array.isArray(v)
      ? Object.fromEntries(Object.keys(v).sort().map((k) => [k, (v as Record<string, unknown>)[k]]))
      : v,
  );
}

const defaultCombine = <D>(_older: D, newer: D) => newer;

export function createEndpoint<D>(config: EndpointConfig<D>) {
  const {
    mode,
    send,
    combine = defaultCombine,
    combineIf = () => true,
    isEqual = (a, b) => stableJson(a) === stableJson(b),
    onEvent = () => {},
    legacyClearPendingAfterFlush = false,
  } = config;
  let inFlight = 0;
  let pending: D | undefined;
  let batch: { data: D } | undefined;
  let deferTimer: ReturnType<typeof setTimeout> | undefined;

  function addToBatch(data: D, asOlder = false) {
    if (!batch) batch = { data };
    else batch.data = asOlder ? combine(data, batch.data) : combine(batch.data, data);
    onEvent({ type: 'queued', data });
  }

  function flushBatch() {
    if (!batch) return;
    const { data } = batch;
    batch = undefined;
    doSend(data);
  }

  function doSend(data: D) {
    inFlight++;
    pending = data;
    onEvent({ type: 'send', data });
    send(data).then(
      () => finish(data, false),
      () => finish(data, true),
    );
  }

  function finish(data: D, failed: boolean) {
    inFlight--;
    onEvent({ type: failed ? 'error' : 'done', data });
    // Xoá pending của request VỪA XONG trước khi gửi gói kế tiếp.
    // Bản gốc làm ngược lại (gửi gói chờ rồi mới delete), nên xoá mất pending của request mới.
    if (!legacyClearPendingAfterFlush && pending === data) pending = undefined;
    if (failed && mode === 'piggyback') {
      // dữ liệu lỗi được xếp trước dữ liệu mới; có sẵn gói chờ thì gửi luôn,
      // không thì đợi lệnh kế tiếp rồi đi nhờ chuyến đó
      const hadBatch = batch !== undefined;
      addToBatch(data, true);
      if (hadBatch) flushBatch();
    } else if (mode === 'successive' || mode === 'successiveUnique' || mode === 'conditional' || mode === 'piggyback') {
      flushBatch();
    }
    if (legacyClearPendingAfterFlush) pending = undefined;
  }

  return {
    trySend(data: D) {
      switch (mode) {
        case 'immediate':
          return doSend(data);
        case 'idempotent':
          if (inFlight) return onEvent({ type: 'skipped', data, reason: 'in-flight' });
          return doSend(data);
        case 'successive':
          return inFlight ? addToBatch(data) : doSend(data);
        case 'successiveUnique':
          if (!inFlight) return doSend(data);
          if (pending !== undefined && isEqual(pending, data)) {
            return onEvent({ type: 'skipped', data, reason: 'duplicate' });
          }
          return addToBatch(data);
        case 'conditional':
          if (inFlight && ((pending !== undefined && combineIf(pending, data)) || (batch && combineIf(batch.data, data)))) {
            return addToBatch(data);
          }
          return doSend(data);
        case 'deferredMulti':
          addToBatch(data);
          clearTimeout(deferTimer);
          deferTimer = setTimeout(flushBatch, 0);
          return;
        case 'piggyback':
          addToBatch(data);
          if (!inFlight) flushBatch();
          return;
      }
    },
    get inFlight() {
      return inFlight;
    },
    get hasBatch() {
      return batch !== undefined;
    },
  };
}

export type Endpoint<D> = ReturnType<typeof createEndpoint<D>>;

Khác bản gốc:

• Xoá pending đúng lúc.
• So trùng bằng JSON đã sắp key (stableJson).
• Trả Promise thay cho callback.
• Không có killswitch, sổ lỗi, hay đường riêng cho GraphQL (bản gốc có nhánh đặc biệt cho THREADLIST_INFO/THREAD_INFO chuyển request sang các fetcher GraphQL).
• Cờ legacyClearPendingAfterFlush chỉ để demo và test tái hiện lỗi.

#Đi qua code

• Một gói chờ duy nhất: dữ liệu tới khi đang có request bay được trộn vào gói đó bằng combine(cũ, mới). Chỉ khi request bay xong mới gửi gói chờ. Kết quả là luôn có tối đa một request đang bay và một gói chờ cho mỗi endpoint (trừ immediate).
• deferredMulti dùng setTimeout(flush, 0) đặt lại ở mỗi lệnh, giống debounce(_, 0) của bản gốc. Sáu lệnh trong cùng một tick thành một request. Cách nhau 70ms thì thành sáu request, đúng như bảng trong demo.
• conditional chỉ gộp khi combineIf trả true với request đang bay hoặc gói chờ. Demo dùng điều kiện "cùng folder": lệnh inbox:t4 chen giữa các lệnh archive được gửi riêng ngay.
• piggyback khi lỗi: dữ liệu lỗi được trộn trước dữ liệu mới (asOlder). Bật "request đầu lỗi 500" trong demo: chuyến sau mang archive:t1+t2+t3+t4+t5.

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

• Debounce ở từng nút bấm: mỗi nơi tự quyết khoảng chờ, không biết request trước đã xong chưa, và không trộn được payload của hai nơi khác nhau cùng gọi một endpoint. Dispatcher đặt quyết định ở cấp endpoint, nơi biết rõ ngữ nghĩa.
• Throttle theo thời gian: gửi đều đặn kể cả khi server đang chậm. Chế độ "successive" tự điều tiết theo tốc độ server: server chậm thì gói chờ lớn hơn, số request ít đi.
• Hàng đợi tuần tự gửi từng lệnh: đúng thứ tự nhưng chậm nhất. Chỉ cần cho lệnh mà thứ tự và từng lệnh đều quan trọng (gửi tin nhắn); với "đánh dấu đã xem" thì lãng phí.
• Gộp ở server (ví dụ GraphQL batching): giảm số kết nối nhưng không bỏ được lệnh vô nghĩa (năm lần "đã xem"). Hai cách bổ sung cho nhau.

#Edge case và cạm bẫy

• Hàm combine phải giữ ngữ nghĩa: "đã xem tới T" thì giữ bản mới nhất; "lưu trữ các ID" thì hợp tập; "chuyển thread sang folder" thì thread nào xuất hiện nhiều lần phải lấy folder cuối cùng. Bản gốc mergeFolders dựng map thread → folder rồi đảo lại, nên lệnh sau đè lệnh trước.
• Lỗi của gói đã gộp ảnh hưởng tới mọi lệnh trong gói. Nếu UI hiện trạng thái từng lệnh, phải trải lỗi ra cho từng phần.
• So trùng bằng JSON.stringify thô phụ thuộc thứ tự key: {a, b} và {b, a} bị coi là khác.
• IDEMPOTENT bỏ lệnh khi đang bay: chỉ dùng khi lệnh sau không mang thông tin mới. "Đã xem tới T2" sau "đã xem tới T1" thì mang thông tin mới; vì thế Messenger dùng BATCH_SUCCESSIVE + "giữ bản mới nhất" cho MARK_SEEN, không dùng IDEMPOTENT.
• Killswitch: bản gốc có cờ toàn cục kill_chat. Biến này được gán lại sau mỗi response, nên một response mang cờ sẽ làm mọi trySend sau đó bị bỏ. Vì không còn request mới nào được gửi, chỉ response của request còn đang bay dở mới có thể tắt cờ; nếu không có thì phải tải lại trang. Hữu ích khi server quá tải, nhưng nên có đường bật lại rõ ràng.

#Dùng trong dự án

import { createEndpoint } from './requestDispatcher';

const markSeen = createEndpoint<{ threadId: string; seenAt: number }>({
  mode: 'successive',
  combine: (_older, newer) => newer, // chỉ cần mốc mới nhất
  send: (d) => api.post('/threads/seen', d),
});

const archive = createEndpoint<{ ids: string[] }>({
  mode: 'successive',
  combine: (a, b) => ({ ids: [...new Set([...a.ids, ...b.ids])] }),
  send: (d) => api.post('/threads/archive', d),
});

const syncSettings = createEndpoint<Partial<Settings>>({
  mode: 'piggyback',
  combine: (older, newer) => ({ ...older, ...newer }), // lệnh mới đè lệnh cũ cùng key
  send: (d) => api.patch('/settings', d),
});

#Kiểm thử

import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { createEndpoint, type DispatchMode } from './requestDispatcher';

type P = { ids: string[] };
const combine = (a: P, b: P): P => ({ ids: [...a.ids, ...b.ids] });

function setup(mode: DispatchMode, extra: { fail?: (n: number) => boolean; legacy?: boolean } = {}) {
  const sent: string[] = [];
  let n = 0;
  const ep = createEndpoint<P>({
    mode,
    combine,
    combineIf: (a, b) => a.ids[0][0] === b.ids[0][0],
    legacyClearPendingAfterFlush: extra.legacy,
    send: (d) =>
      new Promise((resolve, reject) => {
        const i = ++n;
        sent.push(d.ids.join('+'));
        setTimeout(() => (extra.fail?.(i) ? reject(new Error('500')) : resolve()), 100);
      }),
  });
  return { ep, sent };
}

describe('createEndpoint', () => {
  beforeEach(() => vi.useFakeTimers());
  afterEach(() => vi.useRealTimers());

  it('successive: đang bay thì gộp, xong mới gửi gói chờ', async () => {
    const { ep, sent } = setup('successive');
    ['a', 'b', 'c'].forEach((x) => ep.trySend({ ids: [x] }));
    await vi.runAllTimersAsync();
    expect(sent).toEqual(['a', 'b+c']);
  });

  it('deferredMulti: gom mọi lệnh trong cùng tick thành một request', async () => {
    const { ep, sent } = setup('deferredMulti');
    ['a', 'b', 'c'].forEach((x) => ep.trySend({ ids: [x] }));
    await vi.runAllTimersAsync();
    expect(sent).toEqual(['a+b+c']);
  });

  it('conditional: chỉ gộp khi combineIf cho phép', async () => {
    const { ep, sent } = setup('conditional');
    ['a1', 'a2', 'b1'].forEach((x) => ep.trySend({ ids: [x] }));
    await vi.runAllTimersAsync();
    expect(sent).toEqual(['a1', 'b1', 'a2']);
  });

  it('piggyback: dữ liệu lỗi đi cùng chuyến sau, đứng trước dữ liệu mới', async () => {
    const { ep, sent } = setup('piggyback', { fail: (n) => n === 1 });
    ep.trySend({ ids: ['a'] });
    await vi.advanceTimersByTimeAsync(150);
    ep.trySend({ ids: ['b'] });
    await vi.runAllTimersAsync();
    expect(sent).toEqual(['a', 'a+b']);
  });

  it('successiveUnique: bỏ yêu cầu trùng request đang bay, kể cả request vừa đi từ gói chờ', async () => {
    for (const legacy of [true, false]) {
      const { ep, sent } = setup('successiveUnique', { legacy });
      ep.trySend({ ids: ['p1'] });
      ep.trySend({ ids: ['p2'] });
      await vi.advanceTimersByTimeAsync(120); // p1 xong → p2 bay
      ep.trySend({ ids: ['p2'] });
      await vi.runAllTimersAsync();
      expect(sent).toEqual(legacy ? ['p1', 'p2', 'p2'] : ['p1', 'p2']);
    }
  });
});

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

• TanStack Query / SWR bỏ trùng request đọc cùng khoá (dedupe khi đang bay), gần với successiveUnique cho GET. Với mutation, chúng không gộp; có scope để xếp mutation chạy tuần tự (TanStack Query v5), gần với successive nhưng không trộn payload.
• DataLoader gom theo tick như deferredMulti, dành cho đọc theo khoá; xem bài #32 về ApiBatcher, một biến thể của Facebook cho Graph API.
• Background Sync (service worker, sync event) cho phép thử lại request khi có mạng, kể cả sau khi đóng tab. Chỉ có trên trình duyệt dựa Chromium (tính tới 10/2026); không trộn payload.
• Không có API trình duyệt nào làm thay phần "gộp theo ngữ nghĩa endpoint". Đó là kiến thức nghiệp vụ, nên nó nằm trong code của bạn.

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