EloAdaptor: waitForReachingServerVersion() never resolves on a one-shot backfill (compare runs before the un-awaited imports land) #66

// packages/loro-adaptors/src/elo-adaptor.ts (main, unchanged since 6570a2ee)
applyUpdate(updates: Uint8Array[]): void {
  if (this.destroyed || !updates?.length) return;
  for (const containerBytes of updates) {
    try {
      this.importEloContainer(containerBytes).catch(err => { ... });   // async, not awaited
    } catch (error) { ... }
  }
  if (this.initServerVersion && !this.hasReachedServerVersion) {
    const cmp = this.doc.version().compare(this.initServerVersion);     // pre-import version
    if (cmp != null && cmp >= 0) this.reachServerVersionPromise.resolve();
  }
}

// Minimal repro: EloAdaptor.waitForReachingServerVersion() never resolves on a
// one-shot backfill. No application code: loro-adaptors@0.6.1 + loro-crdt only.
import { LoroDoc } from "loro-crdt";
import { EloAdaptor, LoroAdaptor } from "loro-adaptors/loro";

const key = new Uint8Array(32).fill(7);
const config = { getPrivateKey: async () => ({ keyId: "k1", key }) };
const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));
const ctx = (sink: Uint8Array[][]) =>
  ({
    send: (u: Uint8Array[]) => void sink.push(u),
    onImportError: (e: Error) => { throw e; },
    onJoinFailed: (reason: string) => { throw new Error(reason); },
  }) as any;
const race = (p: Promise<void>, ms: number) =>
  Promise.race([p.then(() => "resolved"), sleep(ms).then(() => `still pending after ${ms} ms`)]);

async function scenario(label: string, make: (doc: LoroDoc) => EloAdaptor | LoroAdaptor) {
  // 1. A peer with content joins an empty room: its adaptor sends the snapshot.
  const docA = new LoroDoc();
  docA.setPeerId(1n);
  docA.getMap("m").set("hello", "world");
  docA.commit();
  const sentByA: Uint8Array[][] = [];
  const a = make(docA);
  a.setCtx(ctx(sentByA));
  await a.handleJoinOk({ version: new Uint8Array(0) } as any);
  const backfill = sentByA.flat();
  const serverVersion = docA.version().encode();

  // 2. A fresh peer joins the room whose server version is A's version, then
  //    receives the whole backfill in ONE applyUpdate call (what a join does).
  const docB = new LoroDoc();
  docB.setPeerId(2n);
  const b = make(docB);
  b.setCtx(ctx([]));
  await b.handleJoinOk({ version: serverVersion } as any);
  const waiting = b.waitForReachingServerVersion();
  b.applyUpdate(backfill);

  const outcome = await race(waiting, 2000);
  const cmp = docB.version().compare(docA.version());
  console.log(`[${label}] backfill containers: ${backfill.length}`);
  console.log(`[${label}] waitForReachingServerVersion(): ${outcome}`);
  console.log(`[${label}] docB has the data: ${docB.getMap("m").get("hello") === "world"}; docB.version() vs server: ${cmp} (0 = equal)`);
  if (outcome !== "resolved") {
    // Proof that only the timing of the compare is wrong: a second, redundant
    // applyUpdate re-runs the compare after the first import landed.
    b.applyUpdate(backfill);
    console.log(`[${label}] after a second (redundant) applyUpdate: ${await race(waiting, 500)}`);
  }
  console.log("");
}

await scenario("EloAdaptor (%ELO)", (doc) => new EloAdaptor(doc, config));
await scenario("LoroAdaptor (%LOR, control)", (doc) => new LoroAdaptor(doc));

applyUpdate(updates: Uint8Array[]): void {
  if (this.destroyed || !updates?.length) return;
  const imports = updates.map(bytes =>
    this.importEloContainer(bytes).catch(err => this.ctx?.onImportError(toError(err), [bytes]))
  );
  void Promise.allSettled(imports).then(() => {
    if (this.destroyed) return;
    if (this.initServerVersion && !this.hasReachedServerVersion) {
      const cmp = this.doc.version().compare(this.initServerVersion);
      if (cmp != null && cmp >= 0) this.reachServerVersionPromise.resolve();
    }
  });
}