Học từ Facebook #81: Flux waitFor tự viết và proxy có đệm

Ads Manager có hàng chục store: chiến dịch, ngân sách, tiền tệ, bảng tổng hợp. Người dùng sửa ngân sách của chiến dịch A. Store "tổng hợp" phải tính lại tổng, nhưng nó chỉ tính đúng nếu store "ngân sách" đã cập nhật trước. Thứ tự đăng ký callback không phải thứ tự phụ thuộc.

Flux giải bài này bằng waitFor: trong callback của mình, store tổng hợp nói "cho BudgetStore chạy trước đã". Facebook có Dispatcher của Flux công khai trên npm, nhưng Ads Manager dùng một bản riêng: AdsDispatchCycle.

Cùng nhóm hạ tầng là AdsFuncChannel: một proxy cho phép gọi một module trước khi nó tải xong. Lời gọi được đệm lại và chạy khi module thật có mặt.

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

#AdsDispatchCycle (bundle 99.js)

Lõi là một switch trên trạng thái của callback (bản đầy đủ trong spoiler):

invoke(id) {
  switch (this.state[id]) {
    case COMPLETE: break;
    case PENDING:  this.state[id] = STARTED; this.stack.push(id);
                   this.fns[id](this.payload);                       // có thể gọi waitFor bên trong
                   this.stack.pop(); this.state[id] = COMPLETE; break;
    case STARTED:  invariant(false, /* 15659 */ id, this.names[id], this.actionType,
                             this.stack.map((x) => this.names[x]).join(', '));
  }
}

const State = { PENDING: 'PENDING', STARTED: 'STARTED', COMPLETE: 'COMPLETE' };

class AdsDispatchCycle {
  constructor(callbacks /* Map id → fn */, payload, actionType, names /* id → tên store */) {
    this.ids = []; this.fns = {}; this.state = {}; this.stack = [];
    this.payload = payload; this.actionType = actionType; this.names = names;
    const report = AdsDataAtomInstrumentation.onMeasureDispatchCallbacks();
    if (report) { this.report = report; this.entries = []; }
    callbacks.forEach((fn, id) => this.register(id, fn));
  }

  start() {
    for (const id of this.ids) this.invoke(id);
    if (this.entries && this.report) this.report({ entries: this.entries, actionType: this.actionType });
  }

  waitFor(ids) { for (const id of ids) this.invoke(id); }

  invoke(id) {
    switch (this.state[id]) {
      case State.COMPLETE: break;
      case State.PENDING: {
        this.state[id] = State.STARTED;
        this.stack.push(id);
        const start = this.report ? performance.now() : null;
        this.fns[id](this.payload);
        if (this.entries && start != null)
          this.entries.push({ start, end: performance.now(), name: this.names[id] ?? 'unknown' });
        this.stack.pop();
        this.state[id] = State.COMPLETE;
        break;
      }
      case State.STARTED:
        invariant(false, /* mã lỗi 15659 */ id, this.names[id], this.actionType,
                  this.stack.map((x) => this.names[x]).join(', '));
    }
  }
}

So với Dispatcher của Flux:

• Chu kỳ là object riêng. Flux giữ trạng thái isPending/isHandled trên chính dispatcher và cấm dispatch lồng nhau ("Cannot dispatch in the middle of a dispatch"). Ở đây mỗi dispatch tạo một chu kỳ mới, trạng thái không dính sang lần sau.
• Lỗi vòng có ngữ cảnh. Flux báo "Circular dependency detected while waiting for X". Bản Ads báo tên store, loại action và cả chuỗi đang chạy dở, ví dụ SummaryStore, BudgetStore. Với hàng chục store, chuỗi này là thứ giúp tìm ra vòng.
• Đo thời gian tuỳ chọn. Chỉ khi onMeasureDispatchCallbacks() trả về hàm báo cáo thì mới gọi performance.now(). Tắt thì không tốn gì.

Chỗ thiếu: không có try/finally. Callback ném lỗi thì stack không được pop, trạng thái kẹt ở STARTED, và báo cáo thời gian của cả chu kỳ không được gửi. Vì chu kỳ là object dùng một lần, lần dispatch sau không bị ảnh hưởng (khác Flux, nơi thiếu finally sẽ kẹt cả dispatcher). Nhưng nếu một store bắt lỗi quanh waitFor rồi chạy tiếp, lần waitFor sau tới store vừa lỗi sẽ báo phụ thuộc vòng giả.

sequenceDiagram
  participant D as Chu kỳ
  participant S as SummaryStore
  participant B as BudgetStore
  participant C as CurrencyStore
  D->>S: invoke (đăng ký đầu tiên)
  S->>D: waitFor BudgetStore
  D->>B: invoke
  B->>D: waitFor CurrencyStore
  D->>C: invoke, COMPLETE
  B-->>D: COMPLETE
  S-->>D: tính tổng từ ngân sách mới, COMPLETE
  D->>B: đã COMPLETE, bỏ qua
  D->>C: đã COMPLETE, bỏ qua

#AdsFuncChannel (bundle 238.js)

Ý tưởng gói trong ba dòng (bản đầy đủ trong spoiler):

buffering[name] = (...args) => this.callMessageHandler({ type: 'call', method: name, thisArg: null, argList: args });
this.messageToCall = (msg) => this.proxyMethods[msg.method].apply(msg.thisArg, msg.argList);   // xả: gọi lại proxyMethods
setBackend(b) { if (this.backend !== b) { this.backend = b; this.flushBuffer(); } }

class AdsFuncChannel {
  constructor(factory) {
    this.buffer = [];
    this.callMessageHandler = (msg) => this.buffer.push(msg);               // mặc định: đệm
    this.messageToCall = (msg) => this.proxyMethods[msg.method].apply(msg.thisArg, msg.argList);
    const forward = {};
    const methods = factory(forward);       // { tên: lớp bọc } hoặc { tên: null }
    const buffering = {};
    for (const name of Object.keys(methods)) {
      buffering[name] = (...args) =>
        this.callMessageHandler({ type: 'call', method: name, thisArg: null, argList: args });
      forward[name] = (...args) => this.backend[name](...args);
      if (methods[name] === null) methods[name] = forward[name];
    }
    this.backend = buffering;               // chưa có backend thật: mọi lời gọi thành tin nhắn
    this.proxyMethods = methods;
  }
  flushBuffer() { const b = this.buffer; this.buffer = []; b.forEach(this.messageToCall); }
  setBackend(b) { if (this.backend !== b) { this.backend = b; this.flushBuffer(); } }
  setCallMessageHandler(h) { this.callMessageHandler = h; this.flushBuffer(); }
}

Người dùng gọi channel.proxyMethods.track('click'). Khi backend chưa có, lời gọi thành tin nhắn {type: 'call', method, argList} và nằm trong hàng đợi. setBackend xả hàng. setCallMessageHandler cho phép gửi tin nhắn đi nơi khác, ví dụ postMessage sang iframe hay worker: chính vì vậy nó tên là "channel".

Hai hành vi, kiểm chứng bằng cách chạy đúng module gốc trong Node:

1. Mất giá trị trả về. Trong lúc đệm, track('click') trả undefined (giá trị của hàm đệm, vốn không return). Người gọi không có cách nào biết kết quả, kể cả sau khi backend tải xong.
2. Lớp bọc chạy hai lần. Với factory = (fwd) => ({ log: (msg) => fwd.log('[ads] ' + msg) }), gọi log('hi') lúc đang đệm. Lớp bọc chạy, gọi fwd.log('[ads] hi'), tin nhắn đệm có đối số '[ads] hi'. Lúc xả, messageToCall gọi lại proxyMethods.log('[ads] hi'), tức lại qua lớp bọc, và backend nhận '[ads] [ads] hi'. Sau khi có backend thì không còn đệm nên không lặp. Có lẽ mọi chỗ dùng thật đều để null (không bọc) nên lỗi chưa lộ.

#Viết lại

Một chu kỳ dispatch, có try/finally:

    const run = (name: string) => {
      const current = state.get(name);
      if (current === undefined) throw new Error(`waitFor: không có callback "${name}"`);
      if (current === 'complete') return;
      if (current === 'started') throw new CycleError([...stack.slice(stack.indexOf(name)), name]);
      state.set(name, 'started');
      stack.push(name);
      const start = now();
      try {
        callbacks.get(name)!(payload, cycle);
      } finally {
        // lỗi hay không, chu kỳ cũng phải về trạng thái nhất quán
        timings.push({ name, ms: now() - start });
        stack.pop();
        state.set(name, 'complete');
      }
    };

Proxy có đệm trả Promise:

  const proxy = {} as BufferedProxy<T>;
  for (const method of methods) {
    (proxy as Record<PropertyKey, unknown>)[method] = (...args: unknown[]) => {
      if (backend) return invoke(method, args);
      if (buffer.length >= maxBuffer)
        return Promise.reject(new Error('bufferedProxy: hàng đợi đầy'));
      return new Promise((resolve, reject) => buffer.push({ method, args, resolve, reject }));
    };
  }

  return {
    proxy,
    get pending() {
      return buffer.length;
    },
    setBackend(next: T | null) {
      backend = next;
      if (!next) return;
      const calls = buffer;
      buffer = [];
      // theo đúng thứ tự gọi; một lời gọi lỗi không chặn các lời gọi sau
      for (const call of calls) invoke(call.method, call.args).then(call.resolve, call.reject);
    },

#Đi qua code

• run là bản invoke có try/finally. started gặp lại nghĩa là đang ở giữa chính nó: stack.slice(stack.indexOf(name)) cắt đúng phần vòng, bỏ các store không liên quan ở đầu chuỗi.
• runCycle (trong file đầy đủ) tạo state, stack, timings mới cho mỗi action, giống ý "một object cho mỗi chu kỳ" của bản gốc.
• dispatch đẩy action vào queue; chỉ lời gọi ngoài cùng mới chạy vòng while. Action phát ra từ trong callback được chạy sau khi chu kỳ hiện tại xong. Lỗi thì finally đặt lại dispatching và bỏ hàng đợi.
• Proxy: khi chưa có backend, mỗi lời gọi tạo một Promise và cất resolve/reject vào hàng đợi. setBackend gọi lần lượt và nối kết quả vào đúng Promise đó.
• invoke bọc lời gọi trong async, nên backend ném lỗi đồng bộ cũng thành Promise bị reject, cùng một kiểu xử lý lỗi cho cả hai trường hợp.

Khác bản gốc:

• try/finally quanh mỗi callback: lỗi hay không, thời gian vẫn được ghi, stack vẫn pop, trạng thái về complete.
• Lỗi vòng là CycleError có chain dạng mảng, chỉ gồm phần vòng: A → B → C → A.
• Dispatch gọi trong lúc đang dispatch được xếp hàng, chạy sau chu kỳ hiện tại: mọi store thấy action theo cùng một thứ tự.
• Proxy không có lớp bọc: muốn biến đổi đối số thì làm ở chỗ gọi. Mỗi lời gọi trả Promise kết quả thật, lỗi của backend thành reject; hàng đợi có giới hạn.

/**
 * Dispatcher kiểu Flux có waitFor (ý tưởng từ AdsDispatchCycle của Facebook Ads Manager):
 * - Mỗi lần dispatch là một "chu kỳ": mọi callback ở trạng thái pending → started → complete.
 * - Callback gọi waitFor(...) để bắt callback khác chạy trước (store B cần số liệu mới của store A).
 * - Gặp lại callback đang "started" nghĩa là phụ thuộc vòng: báo lỗi kèm cả chuỗi.
 * Khác bản gốc: try/finally nên callback ném lỗi không để chu kỳ kẹt; dispatch lồng nhau được
 * xếp hàng thay vì chạy chen; luôn đo thời gian từng callback.
 */

type State = 'pending' | 'started' | 'complete';

export interface DispatchCycle<P> {
  readonly payload: P;
  /** chạy (nếu chưa) các callback theo tên, ngay bây giờ, trước khi tiếp tục */
  waitFor(...names: string[]): void;
}

export type DispatchCallback<P> = (payload: P, cycle: DispatchCycle<P>) => void;

export interface DispatchTiming {
  name: string;
  ms: number;
}

export class CycleError extends Error {
  constructor(readonly chain: string[]) {
    super(`Phụ thuộc vòng trong dispatch: ${chain.join(' → ')}`);
    this.name = 'CycleError';
  }
}

export function createDispatcher<P>(
  options: { now?: () => number; onCycle?: (t: DispatchTiming[], payload: P) => void } = {},
) {
  const now = options.now ?? (() => performance.now());
  const callbacks = new Map<string, DispatchCallback<P>>();
  const queue: P[] = [];
  let dispatching = false;

  function runCycle(payload: P) {
    const state = new Map<string, State>();
    for (const name of callbacks.keys()) state.set(name, 'pending');
    const stack: string[] = []; // các callback đang chạy dở, để in chuỗi khi gặp vòng
    const timings: DispatchTiming[] = [];

    const run = (name: string) => {
      const current = state.get(name);
      if (current === undefined) throw new Error(`waitFor: không có callback "${name}"`);
      if (current === 'complete') return;
      if (current === 'started') throw new CycleError([...stack.slice(stack.indexOf(name)), name]);
      state.set(name, 'started');
      stack.push(name);
      const start = now();
      try {
        callbacks.get(name)!(payload, cycle);
      } finally {
        // lỗi hay không, chu kỳ cũng phải về trạng thái nhất quán
        timings.push({ name, ms: now() - start });
        stack.pop();
        state.set(name, 'complete');
      }
    };

    const cycle: DispatchCycle<P> = {
      payload,
      waitFor: (...names) => names.forEach(run),
    };
    for (const name of callbacks.keys()) run(name);
    options.onCycle?.(timings, payload);
    return timings;
  }

  return {
    register(name: string, callback: DispatchCallback<P>) {
      if (callbacks.has(name)) throw new Error(`callback "${name}" đã có`);
      callbacks.set(name, callback);
      return () => void callbacks.delete(name);
    },
    /**
     * Dispatch trong lúc đang dispatch (callback phát action mới) được xếp hàng, chạy sau khi
     * chu kỳ hiện tại xong, để mọi store thấy action theo cùng một thứ tự.
     */
    dispatch(payload: P) {
      queue.push(payload);
      if (dispatching) return;
      dispatching = true;
      try {
        while (queue.length) runCycle(queue.shift()!);
      } finally {
        dispatching = false;
        queue.length = 0;
      }
    },
    get isDispatching() {
      return dispatching;
    },
  };
}

/**
 * Proxy có đệm (ý tưởng từ AdsFuncChannel): gọi được ngay cả khi "backend" thật chưa tải xong;
 * lời gọi được xếp hàng và chạy theo đúng thứ tự khi có backend.
 * Khác bản gốc: mọi lời gọi trả Promise của kết quả thật (bản gốc trả undefined khi đang đệm),
 * và không có lớp bọc chạy lại lúc xả hàng (bản gốc chạy lớp bọc hai lần với đối số đã bọc).
 */

type AnyFn = (...args: never[]) => unknown;
type Methods<T> = { [K in keyof T]: T[K] extends AnyFn ? K : never }[keyof T];

export type BufferedProxy<T> = {
  [K in Methods<T>]: T[K] extends (...args: infer A) => infer R
    ? (...args: A) => Promise<Awaited<R>>
    : never;
};

interface Call {
  method: PropertyKey;
  args: unknown[];
  resolve: (v: unknown) => void;
  reject: (e: unknown) => void;
}

export function createBufferedProxy<T extends object>(
  methods: readonly Methods<T>[],
  options: { maxBuffer?: number } = {},
) {
  const maxBuffer = options.maxBuffer ?? 1000;
  let backend: T | null = null;
  let buffer: Call[] = [];

  const invoke = (method: PropertyKey, args: unknown[]) =>
    // async: lỗi đồng bộ của backend cũng thành Promise bị reject, như khi đang đệm
    (async () => (backend as Record<PropertyKey, (...a: unknown[]) => unknown>)[method](...args))();

  const proxy = {} as BufferedProxy<T>;
  for (const method of methods) {
    (proxy as Record<PropertyKey, unknown>)[method] = (...args: unknown[]) => {
      if (backend) return invoke(method, args);
      if (buffer.length >= maxBuffer)
        return Promise.reject(new Error('bufferedProxy: hàng đợi đầy'));
      return new Promise((resolve, reject) => buffer.push({ method, args, resolve, reject }));
    };
  }

  return {
    proxy,
    get pending() {
      return buffer.length;
    },
    setBackend(next: T | null) {
      backend = next;
      if (!next) return;
      const calls = buffer;
      buffer = [];
      // theo đúng thứ tự gọi; một lời gọi lỗi không chặn các lời gọi sau
      for (const call of calls) invoke(call.method, call.args).then(call.resolve, call.reject);
    },
  };
}

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

• Sắp xếp thứ tự đăng ký cho đúng: dễ vỡ. Thêm một store mới ở module khác là thứ tự đổi. waitFor khai báo phụ thuộc ngay chỗ dùng.
• Sắp xếp topo trước khi dispatch: cần biết phụ thuộc trước, mà phụ thuộc trong Flux có thể tuỳ theo action (chỉ action SET_BUDGET mới cần chờ). waitFor gọi lúc chạy nên linh hoạt hơn, và vẫn phát hiện vòng.
• Store tự subscribe nhau (kiểu observable): không có "một thời điểm mọi store đã cập nhật xong", dễ render với state nửa cũ nửa mới. Một chu kỳ dispatch đồng bộ đảm bảo React chỉ thấy state sau khi mọi store xong.
• Cấm dispatch lồng nhau (như Flux): an toàn nhưng làm người dùng phải setTimeout để dispatch tiếp. Xếp hàng giữ tính đồng bộ mà không chen ngang.
• Proxy trả undefined: hợp với lời gọi "bắn rồi quên" như log. Nhưng khi API có giá trị trả về (lấy ID sự kiện, kết quả xác thực), Promise là cách duy nhất để trả lời sau.

#Edge case và cạm bẫy

• waitFor ngoài chu kỳ: chỉ gọi được trong callback, vì nó cần cycle. Bản viết lại truyền cycle làm đối số thứ hai, không có cách nào gọi nhầm ngoài lúc dispatch.
• Lỗi giữa chu kỳ: các callback chưa chạy sẽ không chạy, và các action đang xếp hàng bị bỏ. Store có thể lệch nhau. Đây là hành vi của Flux; nếu muốn "chạy tiếp các store khác", phải bắt lỗi trong từng callback.
• Callback chậm: một dispatch đồng bộ chặn luồng chính. Timing cho biết store nào chậm; demo vẽ thanh thời gian từng store.
• Đệm vô hạn: backend không bao giờ tải (mạng lỗi) thì hàng đợi lớn mãi. Có maxBuffer; thêm timeout để reject các lời gọi đang chờ nếu cần.
• Thứ tự sau khi có backend: hàng đợi được xả đồng bộ trong setBackend, trước bất kỳ lời gọi mới nào, nên thứ tự được giữ. Nếu xả bất đồng bộ, một lời gọi mới có thể chen lên trước.

#Dùng trong dự án

import { createDispatcher } from './dispatchCycle';
import { createBufferedProxy } from './bufferedProxy';

type Action = { type: 'SET_BUDGET'; id: string; amount: number } | { type: 'RESET' };

const dispatcher = createDispatcher<Action>();

dispatcher.register('BudgetStore', (action) => {
  if (action.type === 'SET_BUDGET') budgets.set(action.id, action.amount);
});
dispatcher.register('SummaryStore', (action, cycle) => {
  cycle.waitFor('BudgetStore');
  summary.total = [...budgets.values()].reduce((a, b) => a + b, 0);
});

// analytics tải lười; code ở mọi nơi gọi ngay được
interface Analytics { track(name: string, data?: object): string }
const analytics = createBufferedProxy<Analytics>(['track']);
export const track = analytics.proxy.track;
import('./analytics-impl').then((m) => analytics.setBackend(m.createAnalytics()));

const eventId = await track('budget_changed', { id: 'A' }); // có ID thật dù gọi trước khi tải

Muốn bắt store chậm, truyền onCycle: (timings, action) => … khi tạo dispatcher và log những lần có callback quá 8ms.

#Kiểm thử

import { describe, expect, it } from 'vitest';
import { createDispatcher, CycleError } from './dispatchCycle';
import { createBufferedProxy } from './bufferedProxy';

describe('createDispatcher', () => {
  it('waitFor chạy phụ thuộc trước, mỗi callback một lần', () => {
    const order: string[] = [];
    const d = createDispatcher<string>();
    d.register('summary', (_, c) => { c.waitFor('campaign', 'budget'); order.push('summary'); });
    d.register('campaign', () => order.push('campaign'));
    d.register('budget', (_, c) => { c.waitFor('campaign'); order.push('budget'); });
    d.dispatch('x');
    expect(order).toEqual(['campaign', 'budget', 'summary']);
  });

  it('báo vòng kèm chuỗi', () => {
    const d = createDispatcher<number>();
    d.register('A', (_, c) => c.waitFor('B'));
    d.register('B', (_, c) => c.waitFor('C'));
    d.register('C', (_, c) => c.waitFor('A'));
    expect(() => d.dispatch(1)).toThrow(CycleError);
    try { d.dispatch(1); } catch (e) { expect((e as CycleError).chain).toEqual(['A', 'B', 'C', 'A']); }
  });

  it('dispatch lồng nhau chạy sau chu kỳ hiện tại', () => {
    const seen: string[] = [];
    const d = createDispatcher<string>();
    d.register('a', (p) => { seen.push(`a:${p}`); if (p === 'first') d.dispatch('second'); });
    d.register('b', (p) => seen.push(`b:${p}`));
    d.dispatch('first');
    expect(seen).toEqual(['a:first', 'b:first', 'a:second', 'b:second']);
  });

  it('callback lỗi không làm kẹt dispatcher', () => {
    const d = createDispatcher<number>();
    let ok = 0;
    d.register('x', (p) => { if (p === 1) throw new Error('boom'); ok++; });
    expect(() => d.dispatch(1)).toThrow('boom');
    expect(d.isDispatching).toBe(false);
    d.dispatch(2);
    expect(ok).toBe(1);
  });
});

describe('createBufferedProxy', () => {
  it('đệm theo thứ tự, trả kết quả thật, lỗi thành reject', async () => {
    const p = createBufferedProxy<{ track(e: string): string; fail(): void }>(['track', 'fail']);
    const a = p.proxy.track('click');
    const b = p.proxy.fail();
    expect(p.pending).toBe(2);
    p.setBackend({ track: (e) => `ok-${e}`, fail: () => { throw new Error('x'); } });
    await expect(a).resolves.toBe('ok-click');
    await expect(b).rejects.toThrow('x');
    await expect(p.proxy.track('now')).resolves.toBe('ok-now');
  });
});

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

• flux (npm, Facebook): Dispatcher với register, waitFor(tokens), isDispatching; cấm dispatch lồng nhau. Gói đã ngừng phát triển; README khuyên dùng thư viện khác.
• Redux: một store, reducer gộp lại bằng combineReducers; phụ thuộc giữa các phần state được giải bằng selector (tính lúc đọc) thay vì waitFor. Không có vấn đề thứ tự, nhưng mỗi selector phải được memo (xem bài #68).
• Zustand, Jotai, MobX: phụ thuộc giữa các store biểu diễn bằng derived state/computed, thư viện tự tính thứ tự.
• Comlink: proxy cho worker; mọi lời gọi trả Promise, giống createBufferedProxy nhưng qua postMessage. Đó cũng là ý tưởng của setCallMessageHandler trong bản gốc.
• Hàng đợi lệnh của snippet analytics (Google Analytics gtag, Segment analytics.js): đặt một mảng dataLayer/analytics tạm, đẩy lời gọi vào, script thật tải xong thì xả. Cùng mẫu, cũng mất giá trị trả về.

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