rename RoomResource#loading to processing, and avoid TaskCancelation error

Author: lukemelia

packages/host/app/components/ai-assistant/panel.gts

@@ -7,7 +7,7 @@ import { service } from '@ember/service';
 import Component from '@glimmer/component';
 import { tracked, cached } from '@glimmer/tracking';
 
-import { restartableTask, timeout } from 'ember-concurrency';
+import { allSettled, restartableTask, timeout } from 'ember-concurrency';
 import { Velcro } from 'ember-velcro';
 import window from 'ember-window-mock';
 
@@ -446,9 +446,7 @@ export default class AiAssistantPanel extends Component<Signature> {
 
   private loadRoomsTask = restartableTask(async () => {
     await this.matrixService.flushAll;
-    await Promise.allSettled(
-      [...this.roomResources.values()].map((r) => r.loading),
-    );
+    await allSettled([...this.roomResources.values()].map((r) => r.processing));
     await this.enterRoomInitially();
   });
 

packages/host/app/components/matrix/room.gts

@@ -162,7 +162,7 @@ export default class Room extends Component<Signature> {
               />
               <LLMSelect
                 @selected={{@roomResource.activeLLM}}
-                @onChange={{@roomResource.activateLLM}}
+                @onChange={{perform @roomResource.activateLLMTask}}
                 @options={{this.supportedLLMs}}
                 @disabled={{@roomResource.isActivatingLLM}}
               />
@@ -555,7 +555,7 @@ export default class Room extends Component<Signature> {
   private doMatrixEventFlush = restartableTask(async () => {
     await this.matrixService.flushMembership;
     await this.matrixService.flushTimeline;
-    await this.args.roomResource.loading;
+    await this.args.roomResource.processing;
   });
 
   private get messages() {

packages/host/app/resources/room.ts

@@ -2,7 +2,7 @@ import { getOwner } from '@ember/owner';
 import { service } from '@ember/service';
 import { tracked, cached } from '@glimmer/tracking';
 
-import { restartableTask, timeout } from 'ember-concurrency';
+import { TaskInstance, restartableTask, timeout } from 'ember-concurrency';
 import { Resource } from 'ember-resources';
 
 import { TrackedMap } from 'tracked-built-ins';
@@ -74,7 +74,7 @@ export class RoomResource extends Resource<Args> {
   private _isDisplayingViewCodeMap: TrackedMap<string, boolean> =
     new TrackedMap();
   @tracked matrixRoom: Room | undefined;
-  @tracked loading: Promise<void> | undefined;
+  @tracked processing: TaskInstance<void> | undefined;
   @tracked roomId: string | undefined;
 
   // To avoid delay, instead of using `roomResource.activeLLM`, we use a tracked property
@@ -89,10 +89,17 @@ export class RoomResource extends Resource<Args> {
       return;
     }
     this.roomId = named.roomId;
-    this.loading = this.load.perform(named.roomId);
+    this.processing = this.processRoomTask.perform(named.roomId);
   }
 
-  private load = restartableTask(async (roomId: string) => {
+  get isProcessing() {
+    return this.processRoomTask.isRunning;
+  }
+
+  processingLastStarteAt = 0;
+
+  private processRoomTask = restartableTask(async (roomId: string) => {
+    this.processingLastStarteAt = Date.now();
     try {
       this.matrixRoom = roomId
         ? await this.matrixService.getRoomData(roomId)
@@ -271,15 +278,12 @@ export class RoomResource extends Resource<Args> {
     return this.llmBeingActivated ?? this.matrixRoom?.activeLLM ?? DEFAULT_LLM;
   }
 
-  activateLLM(model: string) {
-    this.activateLLMTask.perform(model);
-  }
-
   get isActivatingLLM() {
     return this.activateLLMTask.isRunning;
   }
 
-  private activateLLMTask = restartableTask(async (model: string) => {
+  activateLLMTask = restartableTask(async (model: string) => {
+    await this.processing;
     if (this.activeLLM === model) {
       return;
     }

packages/host/app/services/command-service.ts

@@ -1,4 +1,5 @@
 import { getOwner, setOwner } from '@ember/owner';
+import { debounce } from '@ember/runloop';
 import Service, { service } from '@ember/service';
 import { isTesting } from '@embroider/macros';
 
@@ -16,6 +17,7 @@ import {
   CommandContextStamp,
   getClass,
   identifyCard,
+  delay,
 } from '@cardstack/runtime-common';
 
 import type MatrixService from '@cardstack/host/services/matrix-service';
@@ -48,6 +50,7 @@ export default class CommandService extends Service {
   @service declare private realmServer: RealmServerService;
   currentlyExecutingCommandRequestIds = new TrackedSet<string>();
   private commandProcessingEventQueue: string[] = [];
+  private flushCommandProcessingQueue: Promise<void> | undefined;
 
   private commands: Map<
     string,
@@ -63,7 +66,7 @@ export default class CommandService extends Service {
     return name;
   }
 
-  public async queueEventForCommandProcessing(event: Partial<IEvent>) {
+  public queueEventForCommandProcessing(event: Partial<IEvent>) {
     let eventId = event.event_id;
     if (event.content?.['m.relates_to']?.rel_type === 'm.replace') {
       eventId = event.content?.['m.relates_to']!.event_id;
@@ -85,24 +88,52 @@ export default class CommandService extends Service {
     }
 
     this.commandProcessingEventQueue.push(compoundKey);
-    let roomResource = this.matrixService.roomResources.get(event.room_id!);
-    await roomResource?.loading;
 
-    this.drainCommandProcessingQueue();
+    debounce(this, this.drainCommandProcessingQueue, 100);
   }
 
   private async drainCommandProcessingQueue() {
-    while (this.commandProcessingEventQueue.length > 0) {
-      let [roomId, eventId] = this.commandProcessingEventQueue
-        .shift()!
-        .split('|');
+    await this.flushCommandProcessingQueue;
+
+    let finishedProcessingCommands: () => void;
+    this.flushCommandProcessingQueue = new Promise(
+      (res) => (finishedProcessingCommands = res),
+    );
+
+    let commandSpecs = [...this.commandProcessingEventQueue];
+    this.commandProcessingEventQueue = [];
+
+    while (commandSpecs.length > 0) {
+      let [roomId, eventId] = commandSpecs.shift()!.split('|');
 
       let roomResource = this.matrixService.roomResources.get(roomId!);
       if (!roomResource) {
         throw new Error(
           `Room resource not found for room id ${roomId}, this should not happen`,
         );
       }
+      let timeout = Date.now() + 60_000; // reset the timer to avoid a long wait if the room resource is processing
+      let currentRoomProcessingTimestamp = roomResource.processingLastStarteAt;
+      while (
+        roomResource.isProcessing &&
+        currentRoomProcessingTimestamp ===
+          roomResource.processingLastStarteAt &&
+        Date.now() < timeout
+      ) {
+        // wait for the room resource to finish processing
+        await delay(100);
+      }
+      if (
+        roomResource.isProcessing &&
+        currentRoomProcessingTimestamp === roomResource.processingLastStarteAt
+      ) {
+        // room seems to be stuck processing, so we will log and skip this event
+        console.error(
+          `Room resource for room ${roomId} seems to be stuck processing, skipping event ${eventId}`,
+        );
+        continue;
+      }
+
       let message = roomResource.messages.find((m) => m.eventId === eventId);
       if (!message) {
         continue;
@@ -127,6 +158,7 @@ export default class CommandService extends Service {
         }
       }
     }
+    finishedProcessingCommands!();
   }
 
   get commandContext(): CommandContext {

packages/host/app/services/matrix-service.ts

@@ -1529,11 +1529,11 @@ export default class MatrixService extends Service {
     });
   }
 
-  async setLLMForCodeMode() {
+  setLLMForCodeMode() {
     this.setLLMModel('anthropic/claude-3.5-sonnet');
   }
 
-  private async setLLMModel(model: string) {
+  private setLLMModel(model: string) {
     if (!DEFAULT_LLM_LIST.includes(model)) {
       throw new Error(`Cannot find LLM model: ${model}`);
     }
@@ -1544,8 +1544,7 @@ export default class MatrixService extends Service {
     if (!roomResource) {
       return;
     }
-    await roomResource.loading;
-    roomResource.activateLLM(model);
+    roomResource.activateLLMTask.perform(model);
   }
 }
 

packages/runtime-common/utils.ts

@@ -22,6 +22,10 @@ export async function retry<T>(
   return null;
 }
 
+export async function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
 /**
  * Encodes a string to web-safe base64 format.
  * Standard base64 uses '+', '/' and '=' which can cause issues in URLs and other web contexts.