Học từ Facebook #89: Chỉ mục subscription và hàng đợi theo chiều sâu

Một trang Facebook có thể có hàng nghìn fragment Relay đang sống: mỗi bài trong feed, mỗi bình luận, mỗi avatar trong thanh chat là một subscription vào store. Khi một payload nhỏ tới (một lượt thích), store phải trả lời câu hỏi: những subscription nào cần đọc lại?

Cách đơn giản là duyệt hết. Relay có sẵn cách đó, và đang thử một cách khác sau một feature flag: chỉ mục theo ID record. Bài này đọc cả hai, rồi thêm hai mảnh hạ tầng nhỏ khác trong cùng bundle: hàng đợi task của Relay classic và cache response của Relay.

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

#Cách cũ: duyệt mọi subscription

Store của Relay (relay-runtime/store/RelayModernStore) chọn một trong hai lớp quản lý subscription, tuỳ feature flag:

this._storeSubscriptions = RelayFeatureFlags.ENABLE_STORE_SUBSCRIPTIONS_REFACTOR === true
  ? new RelayStoreSubscriptionsUsingMapByID()
  : new RelayStoreSubscriptions();

Trên facebook.com, flag này được bật theo gatekeeper trong configureRelayFeatureFlagsForWWW (ENABLE_STORE_SUBSCRIPTIONS_REFACTOR = gkx("1839774")), tức là một thử nghiệm A/B. Lớp mặc định RelayStoreSubscriptions giữ một Set mọi subscription và duyệt hết mỗi lần store notify:

updateSubscriptions(source, updatedRecordIDs, owners) {
  const hasUpdatedRecords = !isEmptyObject(updatedRecordIDs);
  this._subscriptions.forEach((sub) => {
    const owner = this._updateSubscription(source, sub, updatedRecordIDs, hasUpdatedRecords);
    if (owner != null) owners.push(owner);
  });
}
// trong _updateSubscription:
const overlaps = hasUpdatedRecords && hasOverlappingIDs(snapshot.seenRecords, updatedRecordIDs);
if (!stale && !overlaps) return;

hasOverlappingIDs là một vòng for…in trên seenRecords của snapshot, kiểm tra từng ID có trong tập vừa đổi không. Mỗi lần store đổi, chi phí là tổng số record đã đọc của mọi subscription, kể cả khi chỉ một record đổi.

#Cách mới: chỉ mục theo record

relay-runtime/store/RelayStoreSubscriptionsUsingMapByID (trong 99 ChatComposer/all.js) đảo chiều: thay vì hỏi từng subscription "bạn có dính record nào đổi không", nó hỏi từng record đổi "ai đang đọc bạn":

updateSubscriptions(source, updatedRecordIDs, owners) {
  this._notifiedRevision++;
  Object.keys(updatedRecordIDs).forEach((id) => {
    const subs = this._subscriptionsByDataId.get(id);
    if (subs == null) return;
    subs.forEach((sub) => {
      if (sub.notifiedRevision === this._notifiedRevision) return; // đã đọc lại trong vòng này
      const owner = this._updateSubscription(source, sub, false);
      if (owner != null) owners.push(owner);
    });
  });
  this._staleSubscriptions.forEach(/* subscription bị đánh dấu cũ sau restore optimistic */);
  this._staleSubscriptions.clear();
}

Ba chi tiết đáng học:

1. Revision thay cho Set "đã xử lý". Một subscription đọc 5 record, cả 5 đều đổi thì nó nằm trong 5 tập. Thay vì tạo một Set mới mỗi vòng để nhớ ai đã được đọc lại, mỗi subscription mang một số notifiedRevision. Bằng số vòng hiện tại thì bỏ qua. Không cấp phát gì cả.
2. Dựng lại chỉ mục sau mỗi lần đọc. Đọc lại có thể chạm vào record khác (ví dụ bestFriend vừa trỏ sang người khác). Hàm _updateSubscriptionIndex gỡ subscription khỏi mọi ID cũ rồi thêm vào mọi ID mới.
3. Chỉ báo khi khác. Dữ liệu mới đi qua recycleNodesInto (bài #87). Nếu kết quả === dữ liệu cũ, callback không được gọi. Đổi một field không nằm trong fragment thì subscription bị đọc lại nhưng component không render.

Phần optimistic (bài #88) có thêm hai hàm: snapshotSubscriptions lưu backup cho từng subscription trước khi đặt lớp phủ, và restoreSubscriptions so backup với dữ liệu hiện tại để đưa subscription nào lệch vào tập _staleSubscriptions. Một bộ đếm thứ hai (snapshotRevision) chống xử lý một subscription hai lần khi nó nằm dưới nhiều ID.

#Bug: subscription sống lại sau dispose

Hàm subscribe:

subscribe(snapshot, callback) {
  const subscription = { backup: null, callback, notifiedRevision: this._notifiedRevision,
                         snapshotRevision: this._snapshotRevision, snapshot };
  const dispose = () => {
    for (const id in snapshot.seenRecords) {        // ← snapshot của lúc subscribe
      const subs = this._subscriptionsByDataId.get(id);
      if (subs != null) { subs.delete(subscription); if (subs.size === 0) this._subscriptionsByDataId.delete(id); }
    }
  };
  for (const id in snapshot.seenRecords) { /* thêm vào chỉ mục */ }
  return { dispose };
}

dispose đóng gói biến snapshot của lúc gọi subscribe. Sau mỗi lần đọc lại, subscription.snapshot được thay bằng snapshot mới và chỉ mục được dựng theo seenRecords mới, nhưng dispose vẫn duyệt seenRecords cũ.

Mình nạp đúng đoạn module trong bundle vào node (giả lập __d, RelayReader là hàm đọc đơn giản) và chạy kịch bản:

1. Subscribe một selector đọc record A.
2. A đổi và trỏ sang B. Đọc lại, giờ subscription nằm dưới A và B. Callback: 1 lần.
3. dispose(). Chỉ gỡ khỏi A.
4. B đổi. Subscription vẫn nằm dưới B, bị đọc lại, callback chạy lần 2. Tệ hơn, lần đọc lại này thêm nó trở lại dưới A.

Kết quả node: calls after dispose + B update 2, chỉ mục còn cả hai khoá A và B. Trong React, callback này gọi setState lên component đã unmount, giữ component đó trong bộ nhớ, và đọc lại dữ liệu vô ích mỗi lần B đổi. Phần "Subscription ma" trong demo chạy cùng kịch bản. Cách sửa: dispose duyệt subscription.snapshot.seenRecords (tập hiện tại), và đánh dấu subscription đã huỷ để vòng update đang chạy bỏ qua nó.

#RelayTaskQueue: một bước mỗi tick, task con chen lên đầu

react-relay-deprecated/classic/tools/RelayTaskQueue (trong 155.js) là di sản của Relay classic. enqueue(f1, f2, f3) trả về Promise. Mỗi bước là một mục riêng trong hàng đợi, và bước sau chỉ được đẩy vào khi bước trước chạy xong:

enqueue(...steps) {
  const promise = new Promise((resolve, reject) => {
    let i = 0, error = null;
    const next = (value) => {
      if (error) { reject(error); return; }
      if (i >= steps.length) resolve(value);
      else this._queue.push(() => {
        next((() => { try { return steps[i++](value); } catch (e) { error = e; } })());
      });
    };
    next(undefined);
  });
  this._scheduleIfNecessary();
  return promise;
}

Mẹo nằm ở hàm chạy một mục (tên gốc $6):

_invokeWithinScopedQueue(item) {
  const outer = this._queue;
  this._queue = [];                          // mục mới sinh ra trong lúc chạy vào mảng rỗng này
  try { item(); }
  finally {
    Array.prototype.unshift.apply(outer, this._queue); // rồi được đặt lên ĐẦU hàng cũ
    this._queue = outer;
  }
}

Bước tiếp theo của chính task đó, và mọi task con mà bước này enqueue, đều chạy trước các task đã xếp hàng từ trước. Chạy bằng node với task A (bước A1 sinh task con 2 bước, sau đó A2), task B: thứ tự là A1 child1 child2 A2 B1. Đó là duyệt theo chiều sâu. Có lẽ Relay classic cần điều này để một lần ghi dữ liệu (gồm các bước con) hoàn tất trước khi lần khác bắt đầu, trong khi vẫn nhường trình duyệt giữa các bước nhờ scheduler.

Scheduler được inject (injectScheduler). Callback scheduler nhận có guard: gọi lại lần hai thì invariant ném lỗi. Không có scheduler thì mọi thứ chạy đồng bộ, nhưng theo kiểu đệ quy: chạy xong một mục thì gọi lại hàm lên lịch, hàm này chạy mục tiếp theo. Chạy bằng node, một chuỗi 20.000 bước không có scheduler báo Maximum call stack size exceeded. Thực tế không ai tạo chuỗi dài như vậy, nhưng đó là lý do bản viết lại dùng vòng lặp.

#RelayQueryResponseCache: "LRU" là FIFO

relay-runtime/network/RelayQueryResponseCache dùng để cache response của network layer:

get(queryID, variables) {
  const key = getCacheKey(queryID, variables); // JSON.stringify(stableCopy({queryID, variables}))
  this._responses.forEach((resp, k) => {        // quét toàn bộ để xoá mục hết hạn
    if (!isCurrent(resp.fetchTime, this._ttl)) this._responses.delete(k);
  });
  const resp = this._responses.get(key);
  return resp != null ? { ...resp.payload, extensions: { ...resp.payload.extensions, cacheTimestamp: resp.fetchTime } } : null;
}
set(queryID, variables, payload) {
  const key = getCacheKey(queryID, variables);
  this._responses.delete(key);                 // đưa xuống cuối thứ tự chèn
  this._responses.set(key, { fetchTime: Date.now(), payload });
  if (this._responses.size > this._size) {
    const first = this._responses.keys().next();
    if (!first.done) this._responses.delete(first.value); // đuổi mục đầu
  }
}

set dùng mẹo delete + set của Map để đưa khoá xuống cuối, đúng kiểu LRU. Nhưng get không làm vậy. Chạy bằng node với cache 2 mục: đặt A, đặt B, đọc A hai lần, đặt C. Kết quả: A bị đuổi, còn lại [B, C]. Mục được đọc nhiều nhất bị đuổi theo thứ tự chèn. Ngoài ra mỗi get tốn O(n) vì vòng quét TTL.

Có lẽ đây là lựa chọn có chủ ý: cache này thường nhỏ (vài chục mục), và việc dọn TTL mỗi lần đọc giữ code đơn giản. Nhưng tên gọi và cách dùng delete+set dễ làm người đọc nghĩ đây là LRU.

Khoá cache dùng stableCopy: sao chép object với khoá được sắp xếp đệ quy, để {a: 1, b: 2} và {b: 2, a: 1} cho cùng một chuỗi JSON.

#Bên trong: hai cách tìm subscription bị ảnh hưởng

flowchart LR
  subgraph Cu["Cách cũ: duyệt mọi subscription"]
    U1["record u4 đổi"] --> S1["sub 0: seen có u4?"]
    U1 --> S2["sub 1: seen có u4?"]
    U1 --> S3["… sub N: seen có u4?"]
  end
  subgraph Moi["Chỉ mục theo record"]
    U2["record u4 đổi"] --> M["Map: u4 → {sub 1, sub 4}"]
    M --> R1["đọc lại sub 1"]
    M --> R2["đọc lại sub 4"]
    R1 --> RE["dựng lại chỉ mục theo seen mới"]
  end

Chi phí cách cũ tỉ lệ với tổng số record mà mọi subscription đã đọc. Chi phí chỉ mục tỉ lệ với số subscription thật sự dính tới record đổi, cộng chi phí cập nhật chỉ mục cho các subscription đó. Đánh đổi: bộ nhớ cho Map và các Set, cùng một ít việc mỗi lần đọc lại.

#Viết lại

Ba file. src/lib/subscriptionIndex.ts dùng recycleNodesInto của bài #87. Selector nhận một hàm read(id). Index tự ghi lại mọi ID đã đọc, nên không cần reader riêng như RelayReader:

  update(changedIds: Iterable<string>): UpdateStats {
    this.revision++;
    const stats: UpdateStats = { reads: 0, notified: 0 };
    for (const id of changedIds) {
      const subs = this.byRecord.get(id);
      if (!subs) continue;
      // sao chép: callback có thể subscribe/dispose làm đổi Set đang duyệt
      for (const sub of [...subs]) {
        if (sub.disposed || sub.notifiedRevision === this.revision) continue;
        sub.notifiedRevision = this.revision;
        stats.reads++;
        const { data, seen } = this.read(sub.selector);
        const next = recycleNodesInto(sub.data, data);
        const oldSeen = sub.seen;
        sub.seen = seen;
        this.index(sub, oldSeen, seen);
        if (next !== sub.data) {
          sub.data = next;
          stats.notified++;
          sub.callback(next);
        }
      }
    }
    return stats;
  }

  private index(sub: Subscription<unknown>, oldSeen: Set<string>, newSeen: Set<string>) {
    oldSeen.forEach((id) => {
      if (newSeen.has(id)) return;
      const set = this.byRecord.get(id);
      if (!set) return;
      set.delete(sub);
      if (set.size === 0) this.byRecord.delete(id);
    });
    newSeen.forEach((id) => {
      if (oldSeen.has(id)) return;
      let set = this.byRecord.get(id);
      if (!set) this.byRecord.set(id, (set = new Set()));
      set.add(sub);
    });
  }

src/lib/relayTaskQueue.ts: vòng lặp thay cho đệ quy, và trong lúc một mục đang chạy thì scheduled vẫn là true để enqueue lồng không hẹn thêm lần nữa:

  /** Chạy một mục: mục mới sinh ra trong lúc chạy được đặt lên ĐẦU hàng. */
  private runOne(item: Item) {
    const rest = this.queue;
    this.queue = [];
    try {
      item();
    } finally {
      this.queue = this.queue.concat(rest);
    }
  }

  private schedule() {
    if (this.scheduled || this.draining || this.queue.length === 0) return;
    if (!this.scheduler) {
      // không có scheduler: chạy hết ngay bằng vòng lặp, không đệ quy
      this.draining = true;
      try {
        while (this.queue.length) this.runOne(this.queue.shift()!);
      } finally {
        this.draining = false;
      }
      return;
    }
    this.scheduled = true;
    let called = false;
    this.scheduler(() => {
      if (called) throw new Error('Scheduler gọi lại cùng một callback hai lần');
      called = true;
      const item = this.queue.shift();
      if (item) this.runOne(item); // trong lúc chạy, scheduled vẫn true nên enqueue lồng không hẹn thêm
      this.scheduled = false;
      this.schedule();
    });
  }
}

/**
 * Chỉ mục subscription theo record, ý tưởng từ RelayStoreSubscriptionsUsingMapByID (viết lại):
 * - mỗi subscription nhớ tập record nó đã đọc (seen); chỉ mục là Map recordId → Set<subscription>,
 * - store đổi một số record thì chỉ đọc lại subscription dính tới chúng, mỗi subscription tối đa một lần/vòng,
 * - đọc lại xong thì dựng lại chỉ mục theo tập record mới, và chỉ báo khi dữ liệu thật sự khác (structural sharing).
 */
import { recycleNodesInto } from './recycleNodes';

export type Reader = (id: string) => unknown;
export type Selector<T> = (read: Reader) => T;

interface Subscription<T> {
  selector: Selector<T>;
  callback: (data: T) => void;
  data: T;
  seen: Set<string>;
  notifiedRevision: number;
  disposed: boolean;
}

export interface UpdateStats {
  /** số subscription bị đọc lại */
  reads: number;
  /** số callback được gọi (dữ liệu thật sự khác) */
  notified: number;
}

export class SubscriptionIndex {
  private byRecord = new Map<string, Set<Subscription<unknown>>>();
  private revision = 0;
  totalReads = 0;

  constructor(private getRecord: (id: string) => unknown) {}

  /** Chạy selector qua một reader có ghi lại mọi ID đã đọc. */
  private read<T>(selector: Selector<T>) {
    const seen = new Set<string>();
    const data = selector((id) => {
      seen.add(id);
      return this.getRecord(id);
    });
    this.totalReads++;
    return { data, seen };
  }

  subscribe<T>(
    selector: Selector<T>,
    callback: (data: T) => void,
  ): { data: T; dispose: () => void } {
    const { data, seen } = this.read(selector);
    const sub: Subscription<T> = {
      selector,
      callback,
      data,
      seen,
      notifiedRevision: this.revision,
      disposed: false,
    };
    this.index(sub as Subscription<unknown>, new Set(), seen);
    return {
      data,
      dispose: () => {
        if (sub.disposed) return;
        sub.disposed = true;
        // gỡ theo tập seen HIỆN TẠI, không phải tập lúc subscribe
        this.index(sub as Subscription<unknown>, sub.seen, new Set());
      },
    };
  }

  /** Báo cho index biết các record vừa đổi. */
  update(changedIds: Iterable<string>): UpdateStats {
    this.revision++;
    const stats: UpdateStats = { reads: 0, notified: 0 };
    for (const id of changedIds) {
      const subs = this.byRecord.get(id);
      if (!subs) continue;
      // sao chép: callback có thể subscribe/dispose làm đổi Set đang duyệt
      for (const sub of [...subs]) {
        if (sub.disposed || sub.notifiedRevision === this.revision) continue;
        sub.notifiedRevision = this.revision;
        stats.reads++;
        const { data, seen } = this.read(sub.selector);
        const next = recycleNodesInto(sub.data, data);
        const oldSeen = sub.seen;
        sub.seen = seen;
        this.index(sub, oldSeen, seen);
        if (next !== sub.data) {
          sub.data = next;
          stats.notified++;
          sub.callback(next);
        }
      }
    }
    return stats;
  }

  private index(sub: Subscription<unknown>, oldSeen: Set<string>, newSeen: Set<string>) {
    oldSeen.forEach((id) => {
      if (newSeen.has(id)) return;
      const set = this.byRecord.get(id);
      if (!set) return;
      set.delete(sub);
      if (set.size === 0) this.byRecord.delete(id);
    });
    newSeen.forEach((id) => {
      if (oldSeen.has(id)) return;
      let set = this.byRecord.get(id);
      if (!set) this.byRecord.set(id, (set = new Set()));
      set.add(sub);
    });
  }

  /** Số subscription đang được chỉ mục dưới một record (cho demo/test). */
  countFor(id: string) {
    return this.byRecord.get(id)?.size ?? 0;
  }
  get indexedRecordCount() {
    return this.byRecord.size;
  }
}

/**
 * Hàng đợi task chạy từng bước, task con chen lên đầu (theo chiều sâu).
 * Ý tưởng từ RelayTaskQueue (react-relay classic), viết lại: vòng lặp thay cho đệ quy, scheduler tuỳ chọn.
 */

export type Step = (input: any) => any;
/** Scheduler nhận một hàm "chạy một bước" và quyết định khi nào gọi nó (rAF, setTimeout, idle…). */
export type Scheduler = (runOne: () => void) => void;

type Item = () => void;

export class TaskQueue {
  private queue: Item[] = [];
  private scheduled = false;
  private draining = false;

  constructor(private scheduler: Scheduler | null = null) {}

  injectScheduler(scheduler: Scheduler | null) {
    this.scheduler = scheduler;
  }

  get size() {
    return this.queue.length;
  }

  /** Mỗi bước nhận kết quả bước trước; bước sau chỉ được xếp hàng khi bước trước xong. */
  enqueue(...steps: Step[]): Promise<unknown> {
    const promise = new Promise<unknown>((resolve, reject) => {
      let i = 0;
      const pushNext = (value: unknown) => {
        if (i >= steps.length) return resolve(value);
        this.queue.push(() => {
          let out: unknown;
          try {
            out = steps[i++](value);
          } catch (e) {
            return reject(e); // bước lỗi: bỏ các bước còn lại
          }
          pushNext(out);
        });
      };
      pushNext(undefined);
    });
    this.schedule();
    return promise;
  }

  /** Chạy một mục: mục mới sinh ra trong lúc chạy được đặt lên ĐẦU hàng. */
  private runOne(item: Item) {
    const rest = this.queue;
    this.queue = [];
    try {
      item();
    } finally {
      this.queue = this.queue.concat(rest);
    }
  }

  private schedule() {
    if (this.scheduled || this.draining || this.queue.length === 0) return;
    if (!this.scheduler) {
      // không có scheduler: chạy hết ngay bằng vòng lặp, không đệ quy
      this.draining = true;
      try {
        while (this.queue.length) this.runOne(this.queue.shift()!);
      } finally {
        this.draining = false;
      }
      return;
    }
    this.scheduled = true;
    let called = false;
    this.scheduler(() => {
      if (called) throw new Error('Scheduler gọi lại cùng một callback hai lần');
      called = true;
      const item = this.queue.shift();
      if (item) this.runOne(item); // trong lúc chạy, scheduled vẫn true nên enqueue lồng không hẹn thêm
      this.scheduled = false;
      this.schedule();
    });
  }
}

/**
 * Cache kết quả query theo (queryId, biến), có TTL và giới hạn kích thước.
 * Ý tưởng từ RelayQueryResponseCache, viết lại thành LRU thật: get cũng làm mới thứ tự,
 * và chỉ kiểm tra hạn của đúng mục đang đọc thay vì quét cả Map.
 */

/** JSON với khoá object được sắp xếp, để {a,b} và {b,a} cho cùng một chuỗi. */
export function stableStringify(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,
  );
}

interface Entry<T> {
  fetchTime: number;
  payload: T;
}

export interface CacheOptions {
  size: number;
  ttl: number;
  /** đổi được để test/demo */
  now?: () => number;
}

export class ResponseCache<T> {
  private map = new Map<string, Entry<T>>();
  private size: number;
  private ttl: number;
  private now: () => number;

  constructor({ size, ttl, now = Date.now }: CacheOptions) {
    if (!(size > 0) || !(ttl > 0)) throw new Error('size và ttl phải > 0');
    this.size = size;
    this.ttl = ttl;
    this.now = now;
  }

  static key(queryId: string, variables: Record<string, unknown>) {
    return stableStringify({ queryId, variables });
  }

  get(
    queryId: string,
    variables: Record<string, unknown>,
  ): (T & { cacheTimestamp: number }) | null {
    const key = ResponseCache.key(queryId, variables);
    const entry = this.map.get(key);
    if (!entry) return null;
    if (entry.fetchTime + this.ttl < this.now()) {
      this.map.delete(key);
      return null;
    }
    // LRU: đưa xuống cuối thứ tự chèn
    this.map.delete(key);
    this.map.set(key, entry);
    return { ...entry.payload, cacheTimestamp: entry.fetchTime };
  }

  set(queryId: string, variables: Record<string, unknown>, payload: T) {
    const key = ResponseCache.key(queryId, variables);
    this.map.delete(key);
    this.map.set(key, { fetchTime: this.now(), payload });
    while (this.map.size > this.size) {
      // mục cũ nhất (ít được dùng gần đây nhất) nằm đầu Map
      this.map.delete(this.map.keys().next().value!);
    }
  }

  /** Dọn mục hết hạn, gọi định kỳ nếu cần (không phải mỗi lần get). */
  prune() {
    const now = this.now();
    for (const [k, e] of this.map) if (e.fetchTime + this.ttl < now) this.map.delete(k);
  }

  clear() {
    this.map.clear();
  }

  keys() {
    return [...this.map.keys()];
  }
}

Khác bản gốc:

• dispose gỡ theo sub.seen hiện tại và đặt cờ disposed. Vòng update duyệt một bản sao của Set, nên callback nào subscribe hoặc dispose giữa chừng cũng không làm hỏng vòng lặp.
• index(sub, oldSeen, newSeen) chỉ đụng tới ID có trong một tập mà không có trong tập kia, thay vì gỡ hết rồi thêm lại.
• Không có phần optimistic (backup, stale). Bài #88 đã có store theo lớp. Ghép hai phần thì thêm snapshot/restore như bản gốc.
• TaskQueue không đệ quy, và một bước ném lỗi thì Promise reject ngay, các bước còn lại không được xếp hàng.
• ResponseCache.get làm mới thứ tự (LRU thật) và chỉ kiểm tra hạn của mục đang đọc. Dọn toàn bộ mục hết hạn là việc của prune(), gọi khi cần.

#Đi qua code

• SubscriptionIndex.read bọc getRecord bằng một hàm ghi ID vào seen. Selector đọc record nào thì record đó vào chỉ mục, không cần khai báo trước.
• update tăng revision một lần. Với mỗi ID đổi, lấy tập subscription, bỏ qua cái đã đọc trong vòng này, đọc lại, recycleNodesInto với dữ liệu cũ, dựng lại chỉ mục, rồi gọi callback nếu tham chiếu khác.
• TaskQueue.enqueue không đẩy mọi bước vào hàng ngay. pushNext đẩy bước đầu. Bước chạy xong mới gọi pushNext cho bước sau, và vì lúc đó this.queue đang là mảng tạm của runOne, bước sau sẽ được nối lên đầu.
• runOne tráo this.queue sang mảng rỗng, chạy mục, rồi this.queue = mới.concat(cũ).
• ResponseCache.set dùng vòng while để đuổi mục đầu cho tới khi đủ kích thước, phòng trường hợp ai đó giảm size.

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

• Mỗi component tự so dữ liệu khi store đổi (kiểu useSelector của Redux): mọi selector đều chạy lại mỗi lần dispatch. Ổn với vài trăm component, nhưng Relay có store chuẩn hoá nên biết chính xác record nào đổi, và nên tận dụng.
• Đăng ký theo field thay vì theo record: chính xác hơn nhưng chỉ mục lớn hơn nhiều lần. Đọc lại theo record rồi lọc bằng recycleNodesInto là điểm cân bằng tốt.
• Proxy theo dõi truy cập (kiểu MobX, Valtio): theo dõi ở mức thuộc tính một cách tự động. Relay không cần vì đã có seenRecords từ reader.
• queueMicrotask cho hàng đợi task: không nhường trình duyệt render giữa các bước, và microtask lồng nhau có thể chặn khung hình. Scheduler inject được cho phép chọn requestIdleCallback, setTimeout hay scheduler.postTask.

#Edge case và cạm bẫy

• Subscription dưới nhiều ID đổi cùng lúc chỉ đọc lại một lần nhờ revision. Nếu bỏ cơ chế này, một bình luận đổi cả author lẫn body sẽ làm component đọc lại hai lần.
• Callback gọi update lồng nhau: revision tăng ở vòng trong, nên vòng ngoài có thể đọc lại một subscription lần nữa. Nên gom thay đổi và gọi update một lần sau khi callback xong, giống notify của Relay chạy sau publish.
• Thứ tự callback theo thứ tự ID trong danh sách đổi và thứ tự chèn trong Set, không theo thứ tự cây React. React gom setState trong cùng một tick nên thường không thành vấn đề.
• Task lỗi giữa chừng: bản gốc ghi lỗi vào biến và reject khi bước tiếp theo được gọi. Bản viết lại reject ngay. Cả hai đều không chạy bước còn lại.
• Khoá cache: stableStringify không phân biệt undefined với khoá thiếu, và Date thành chuỗi ISO. Biến của GraphQL là JSON nên không sao.

#Dùng trong dự án

Store nhỏ có subscription theo ID:

import { SubscriptionIndex } from './lib/subscriptionIndex';

const records = new Map<string, unknown>();
const index = new SubscriptionIndex((id) => records.get(id));

export function useRecordSelector<T>(selector: Selector<T>) {
  const [data, setData] = useState(() => selector((id) => records.get(id)) as T);
  useEffect(() => {
    const sub = index.subscribe(selector, setData);
    setData(sub.data);
    return sub.dispose;
  }, [selector]);
  return data;
}

export function write(changes: Record<string, unknown>) {
  Object.entries(changes).forEach(([id, rec]) => records.set(id, rec));
  index.update(Object.keys(changes));
}

Hàng đợi chạy việc nền theo từng khung hình:

const queue = new TaskQueue((run) => requestAnimationFrame(() => run()));
await queue.enqueue(parsePayload, normalize, (records) => write(records));

#Kiểm thử