Coalesce client loader/store rebuilds during incremental event bursts

packages/host/app/services/client-telemetry.ts

@@ -105,6 +105,11 @@ export interface RebuildEvent extends BaseEvent {
   trigger_module: string;
   modules_refetched: number;
   cards_reloaded: number;
+  // How many incremental events collapsed into this single rebuild. 1 for an
+  // isolated change; >1 when a write burst arrived faster than a rebuild
+  // completes and the events coalesced into one in-flight plus one pending
+  // rebuild.
+  coalesced_events: number;
 }
 
 export interface RealmEvent extends BaseEvent {

packages/host/app/services/store.ts

@@ -11,7 +11,7 @@ import { isTesting } from '@embroider/macros';
 import { tracked } from '@glimmer/tracking';
 
 import { formatDistanceToNow } from 'date-fns';
-import { task } from 'ember-concurrency';
+import { keepLatestTask, task } from 'ember-concurrency';
 
 import { cloneDeep } from 'lodash-es';
 import { isEqual } from 'lodash-es';
@@ -131,7 +131,10 @@ import type { SearchResource } from '../resources/search';
 import type * as CardAPI from '@cardstack/base/card-api';
 import type { CardDef, BaseDef } from '@cardstack/base/card-api';
 import type { FileDef } from '@cardstack/base/file-api';
-import type { RealmEventContent } from '@cardstack/base/matrix-event';
+import type {
+  IncrementalIndexEventContent,
+  RealmEventContent,
+} from '@cardstack/base/matrix-event';
 
 export { CardErrorJSONAPI, CardSaveSubscriber };
 
@@ -1884,49 +1887,75 @@ export default class StoreService extends Service implements StoreInterface {
       return;
     }
     let invalidations = event.invalidations as string[];
-    // Count reloads triggered while handling this event, shared by the
-    // realm-event and (when a loader reset fires) rebuild telemetry.
-    let reloadsTriggered = 0;
     let ownWrite = event.clientRequestId
       ? this.cardService.clientRequestIds.has(event.clientRequestId)
       : false;
-    let rebuildStart: number | undefined;
-    let rebuildTriggerModules: string[] = [];
-    let rebuildModulesRefetched = 0;
-    let rebuildTask: { then: (...a: unknown[]) => unknown } | undefined;
 
-    if (
-      invalidations.find(
-        (i) =>
-          hasExecutableExtension(i) &&
+    // The invalidation triggers a rebuild when it touches an already-loaded
+    // executable module: the loader must be flushed so the updated code is
+    // picked up before the open card graph re-runs. Net-new modules that were
+    // never loaded don't need one. Once a rebuild is in flight, fold every
+    // further executable invalidation into it regardless of `isModuleLoaded` —
+    // the freshly reset loader reports those modules as not-loaded, so the
+    // probe alone would drop a mid-burst code change.
+    let executableInvalidations = invalidations.filter(hasExecutableExtension);
+    let needsRebuild =
+      executableInvalidations.length > 0 &&
+      (this.rebuildForCodeChange.isRunning ||
+        executableInvalidations.some((i) =>
           this.loaderService.loader.isModuleLoaded(i),
-      )
-    ) {
-      // the invalidation included code changes to modules that are already
-      // loaded. in this case we need to flush the loader so that we can pick
-      // up the updated code before re-running the card. net-new modules that
-      // have never been loaded don't require a loader reset.
+        ));
+
+    let reloadsTriggered = 0;
+    if (needsRebuild) {
+      // Coalesce the rebuild. `keepLatestTask` keeps at most one rebuild in
+      // flight and one pending, so a burst of executable invalidations arriving
+      // faster than a rebuild completes collapses into at most 2 rebuilds; the
+      // final rebuild re-fetches current server state, so the end result
+      // reflects the latest generation. This is scheduling only and does not
+      // touch reactivity: an isolated change triggers a single reset and
+      // re-render.
       if (telemetry?.isEnabled) {
-        rebuildStart = performance.now();
-        rebuildTriggerModules = invalidations.filter((i) =>
-          hasExecutableExtension(i),
-        );
-        rebuildModulesRefetched = invalidations.filter(
-          (i) =>
-            hasExecutableExtension(i) &&
-            this.loaderService.loader.isModuleLoaded(i),
-        ).length;
+        this.#accumulatePendingRebuild(event.realmURL, executableInvalidations);
       }
-      this.loaderService.resetLoader();
-      this.store.reset();
-      rebuildTask = this.reestablishReferences.perform() as unknown as {
-        then: (...a: unknown[]) => unknown;
-      };
+      this.rebuildForCodeChange.perform();
+      // The rebuild subsumes the per-invalidation reloads below: store.reset
+      // empties the graph and reestablishReferences re-fetches every live
+      // reference, so running that loop too would only re-fetch cards the
+      // rebuild is about to discard.
+    } else {
+      reloadsTriggered = this.#reloadInvalidatedInstances(event, invalidations);
     }
 
+    if (telemetry?.isEnabled) {
+      telemetry.recordEvent({
+        event_type: 'realm-event',
+        realm: event.realmURL,
+        index_type: 'incremental',
+        invalidations_count: invalidations.length,
+        invalidated_ids: invalidations.slice(0, 50),
+        reloads_triggered: reloadsTriggered,
+        own_write: ownWrite,
+        processing_ms:
+          processingStart !== undefined
+            ? Math.round(performance.now() - processingStart)
+            : 0,
+      });
+    }
+  };
+
+  // Reload the individual cards / file-meta resources named by an incremental
+  // invalidation that did not trigger a full rebuild. Returns the number of
+  // reloads kicked off (for realm-event telemetry).
+  #reloadInvalidatedInstances(
+    event: IncrementalIndexEventContent,
+    invalidations: string[],
+  ): number {
+    let reloadsTriggered = 0;
     for (let invalidation of invalidations) {
       if (hasExecutableExtension(invalidation)) {
-        // we already dealt with this
+        // Executable modules have no card instance to reload here; when an
+        // already-loaded one changed, the coalesced rebuild handled it.
         continue;
       }
       let fileMetaInstance =
@@ -2038,47 +2067,8 @@ export default class StoreService extends Service implements StoreInterface {
       }
     }
 
-    if (telemetry?.isEnabled) {
-      telemetry.recordEvent({
-        event_type: 'realm-event',
-        realm: event.realmURL,
-        index_type: 'incremental',
-        invalidations_count: invalidations.length,
-        invalidated_ids: invalidations.slice(0, 50),
-        reloads_triggered: reloadsTriggered,
-        own_write: ownWrite,
-        processing_ms:
-          processingStart !== undefined
-            ? Math.round(performance.now() - processingStart)
-            : 0,
-      });
-
-      // When a loader reset was triggered, report the rebuild once the
-      // reference re-establishment settles so its duration spans the whole
-      // reset → reload cascade. cards_reloaded reflects the reloads counted
-      // above (this handler runs to completion before the task settles).
-      if (rebuildStart !== undefined && rebuildTask) {
-        let reporter = telemetry;
-        let startedAt = rebuildStart;
-        let triggerModules = rebuildTriggerModules.slice(0, 20);
-        let modulesRefetched = rebuildModulesRefetched;
-        let realmURL = event.realmURL;
-        Promise.resolve(rebuildTask as unknown as Promise<unknown>)
-          .then(() => {
-            reporter.recordEvent({
-              event_type: 'rebuild',
-              realm: realmURL,
-              duration_ms: Math.round(performance.now() - startedAt),
-              trigger_modules: triggerModules,
-              trigger_module: triggerModules[0] ?? '',
-              modules_refetched: modulesRefetched,
-              cards_reloaded: reloadsTriggered,
-            });
-          })
-          .catch(() => {});
-      }
-    }
-  };
+    return reloadsTriggered;
+  }
 
   private loadInstanceTask = task(
     async (idOrDoc: string | LooseSingleCardDocument) => {
@@ -2125,6 +2115,74 @@ export default class StoreService extends Service implements StoreInterface {
     await Promise.all(
       [...remoteIds].map((id) => this.getCardInstance({ idOrDoc: id })),
     );
+    return remoteIds.size;
+  });
+
+  // Telemetry metadata for the pending/in-flight coalesced rebuild, merged
+  // across every executable invalidation that collapses into it. Drained when
+  // the rebuild task begins so the emitted `rebuild` event describes exactly
+  // the code changes that rebuild picked up. Only populated when telemetry is
+  // enabled; the coalescing itself does not depend on it.
+  #pendingRebuild:
+    | {
+        realm: string;
+        triggerModules: Set<string>;
+        modulesRefetched: Set<string>;
+        events: number;
+      }
+    | undefined = undefined;
+
+  #accumulatePendingRebuild(realm: string, executableInvalidations: string[]) {
+    let pending = (this.#pendingRebuild ??= {
+      realm,
+      triggerModules: new Set<string>(),
+      modulesRefetched: new Set<string>(),
+      events: 0,
+    });
+    // The most recent event's realm labels the rebuild; a burst is
+    // overwhelmingly single-realm, and the final generation wins regardless.
+    pending.realm = realm;
+    pending.events++;
+    for (let module of executableInvalidations) {
+      pending.triggerModules.add(module);
+      if (this.loaderService.loader.isModuleLoaded(module)) {
+        pending.modulesRefetched.add(module);
+      }
+    }
+  }
+
+  // Coalesced client rebuild: flush the loader, reset the store, and re-fetch
+  // every live card reference. `keepLatest` bounds a write burst to one
+  // in-flight rebuild plus one pending — intermediate events collapse into the
+  // pending slot — so a burst of rapid executable invalidations costs at most 2
+  // rebuilds regardless of its length. The final rebuild re-fetches current
+  // server state, so the end state reflects the latest generation. This is
+  // scheduling only: each rebuild is the full load-bearing reset, not a partial
+  // one.
+  private rebuildForCodeChange = keepLatestTask(async () => {
+    let telemetry = this.#clientTelemetry();
+    let pending = this.#pendingRebuild;
+    this.#pendingRebuild = undefined;
+    let rebuildStart =
+      telemetry?.isEnabled && pending ? performance.now() : undefined;
+
+    this.loaderService.resetLoader();
+    this.store.reset();
+    let cardsReloaded = await this.reestablishReferences.perform();
+
+    if (telemetry?.isEnabled && pending && rebuildStart !== undefined) {
+      let triggerModules = [...pending.triggerModules].slice(0, 20);
+      telemetry.recordEvent({
+        event_type: 'rebuild',
+        realm: pending.realm,
+        duration_ms: Math.round(performance.now() - rebuildStart),
+        trigger_modules: triggerModules,
+        trigger_module: triggerModules[0] ?? '',
+        modules_refetched: pending.modulesRefetched.size,
+        cards_reloaded: cardsReloaded ?? 0,
+        coalesced_events: pending.events,
+      });
+    }
   });
 
   private reloadTask = task(async (instance: CardDef) => {