Speedup startup by caching rooms with active connections

changelog.d/893.feature

@@ -0,0 +1 @@
+Improve startup time on redis configurations by caching the last known active rooms.

src/Bridge.ts

@@ -684,12 +684,20 @@
  const queue = new PQueue({
  concurrency: 2,
  });
+
  // Set up already joined rooms
- await queue.addAll(this.botUsersManager.joinedRooms.map((roomId) => async () => {
+ let allActiveJoinedRooms = await this.storage.getAllRoomsWithActiveConnections();
+ if (!allActiveJoinedRooms.length) {
+ allActiveJoinedRooms = this.botUsersManager.joinedRooms;
+ }
+ log.info(`Found ${allActiveJoinedRooms.length} active rooms`);
+
+ const loadRoom = async (roomId: string) => {
  log.debug("Fetching state for " + roomId);
 
  try {
  await connManager.createConnectionsForRoomId(roomId, false);
+ this.storage.addRoomHasActiveConnections(roomId);
  } catch (ex) {
  log.error(`Unable to create connection for ${roomId}`, ex);
  return;
@@ -735,13 +743,16 @@
  }
  }
  const adminRoom = await this.setUpAdminRoom(botUser.intent, roomId, accountData, notifContent || NotifFilter.getDefaultContent());
+ this.storage.addRoomHasActiveConnections(roomId);
  // Call this on startup to set the state
  await this.onAdminRoomSettingsChanged(adminRoom, accountData, { admin_user: accountData.admin_user });
  log.debug(`Room ${roomId} is connected to: ${adminRoom.toString()}`);
  } catch (ex) {
  log.error(`Failed to set up admin room ${roomId}:`, ex);
  }
- }));
+ }
+
+ await queue.addAll(allActiveJoinedRooms.map(roomId => () => loadRoom(roomId)));
 
  // Handle spaces
  for (const discussion of connManager.getAllConnectionsOfType(GitHubDiscussionSpace)) {
@@ -774,8 +785,9 @@
  if (this.config.metrics?.enabled) {
  this.listener.bindResource('metrics', Metrics.expressRouter);
  }
+ // This will load all the *active* connections
  await queue.onIdle();
- log.info(`All connections loaded`);
+ log.info(`All active connections loaded`);
 
  // Load feeds after connections, to limit the chances of us double
  // posting to rooms if a previous hookshot instance is being replaced.
@@ -794,6 +806,9 @@
  await this.as.begin();
  log.info(`Bridge is now ready. Found ${this.connectionManager.size} connections`);
  this.ready = true;
+ const inactiveRooms = this.botUsersManager.joinedRooms.filter(rId => !allActiveJoinedRooms.includes(rId));
+ log.info(`Checking ${inactiveRooms.length} rooms with previously inactive state in the background`);
+ await queue.addAll(inactiveRooms.map(rId => () => loadRoom(rId)));
  }
 
  private async handleHookshotEvent<EventType, ConnType extends IConnection>(msg: MessageQueueMessageOut<EventType>, connection: ConnType, handler: (c: ConnType, data: EventType) => Promise<unknown>|unknown) {
@@ -920,7 +935,7 @@
  }
  log.info(`Got message roomId=${roomId} type=${event.type} from=${event.sender}`);
  log.debug("Content:", JSON.stringify(event));
  let processedReply: any;
  let processedReplyMetadata: IRichReplyMetadata|undefined = undefined;
  try {
  processedReply = await this.replyProcessor.processEvent(event, this.as.botClient, EventKind.RoomEvent);
@@ -1384,6 +1399,7 @@
  const adminRoom = new AdminRoom(
  roomId, accountData, notifContent, intent, this.tokenStore, this.config, this.connectionManager,
  );
+ this.storage.addRoomHasActiveConnections(roomId);
 
  adminRoom.on("settings.changed", this.onAdminRoomSettingsChanged.bind(this));
  adminRoom.on("open.project", async (project: ProjectsGetResponseData) => {

src/ConnectionManager.ts

@@ -61,6 +61,7 @@
  }
  this.connections.push(connection);
  this.emit('new-connection', connection);
+ this.storage.addRoomHasActiveConnections(connection.roomId);
  }
  Metrics.connections.set(this.connections.length);
  // Already exists, noop.
@@ -173,7 +174,7 @@
  * @param rollbackBadState 
  * @returns 
  */
  public async createConnectionForState(roomId: string, state: StateEvent<any>, rollbackBadState: boolean): Promise<IConnection|undefined> {
  // Empty object == redacted
  if (state.content.disabled === true || Object.keys(state.content).length === 0) {
  log.debug(`${roomId} has disabled state for ${state.type}`);
@@ -394,6 +395,9 @@
  this.connections.splice(connectionIndex, 1);
  Metrics.connections.set(this.connections.length);
  this.emit('connection-removed', connection);
+ if (this.getAllConnectionsForRoom(roomId).length === 0) {
+ this.storage.removeRoomHasActiveConnections(roomId);
+ }
  }
 
  /**
@@ -403,8 +407,13 @@
  */
  public async removeConnectionsForRoom(roomId: string) {
  log.info(`Removing all connections from ${roomId}`);
- this.connections = this.connections.filter((c) => c.roomId !== roomId);
+ const removedConnections = this.connections.filter((c) => c.roomId === roomId);
+ this.connections = this.connections.filter((c) => !removedConnections.includes(c));
+ removedConnections.forEach(c => {
+ this.emit('connection-removed', c);
+ })
  Metrics.connections.set(this.connections.length);
+ this.storage.removeRoomHasActiveConnections(roomId);
  }
 
  public registerProvisioningConnection(connType: {getProvisionerDetails: (botUserId: string) => GetConnectionTypeResponseItem}) {

src/Stores/MemoryStorageProvider.ts

@@ -107,4 +107,14 @@ export class MemoryStorageProvider extends MSP implements IBridgeStorageProvider
  public async setGitlabDiscussionThreads(connectionId: string, value: SerializedGitlabDiscussionThreads): Promise<void> {
  this.gitlabDiscussionThreads.set(connectionId, value);
  }
+ public addRoomHasActiveConnections(): void {
+ // no-op: only used for startup speedups
+ }
+ public removeRoomHasActiveConnections(): void {
+ // no-op: only used for startup speedups
+ }
+ public async getAllRoomsWithActiveConnections(): Promise<string[]> {
+ // no-op: only used for startup speedups
+ return [];
+ }
 }

src/Stores/RedisStorageProvider.ts

@@ -18,19 +18,15 @@ const GH_ISSUES_REVIEW_DATA_KEY = "gh.issues.review_data";
 const FIGMA_EVENT_COMMENT_ID = "figma.comment_event_id";
 const STORED_FILES_KEY = "storedfiles.";
 const GL_DISCUSSIONTHREADS_KEY = "gl.discussion-threads";
+const ACTIVE_ROOMS = "cache.active_rooms";
 const STORED_FILES_EXPIRE_AFTER = 24 * 60 * 60; // 24 hours
 const COMPLETED_TRANSACTIONS_EXPIRE_AFTER = 24 * 60 * 60; // 24 hours
 const ISSUES_EXPIRE_AFTER = 7 * 24 * 60 * 60; // 7 days
 const ISSUES_LAST_COMMENT_EXPIRE_AFTER = 14 * 24 * 60 * 60; // 7 days
-
-
 const WIDGET_TOKENS = "widgets.tokens.";
 const WIDGET_USER_TOKENS = "widgets.user-tokens.";
-
 const FEED_GUIDS = "feeds.guids.";
 
-
-
 const log = new Logger("RedisASProvider");
 
 export class RedisStorageContextualProvider implements IStorageProvider {
@@ -229,4 +225,20 @@ export class RedisStorageProvider extends RedisStorageContextualProvider impleme
  public async hasSeenFeedGuid(url: string, guid: string): Promise<boolean> {
  return (await this.redis.lpos(`${FEED_GUIDS}${url}`, guid)) != null;
  }
+
+ public addRoomHasActiveConnections(roomId: string): void {
+ this.redis.sadd(ACTIVE_ROOMS, roomId).catch((ex) => {
+ log.warn(`Failed to add ${roomId} to active rooms`, ex);
+ });
+ }
+
+ public removeRoomHasActiveConnections(roomId: string): void {
+ this.redis.srem(ACTIVE_ROOMS, roomId).catch((ex) => {
+ log.warn(`Failed to remove ${roomId} from active rooms`, ex);
+ });
+ }
+
+ public getAllRoomsWithActiveConnections(): Promise<string[]> {
+ return this.redis.smembers(ACTIVE_ROOMS);
+ }
 }

src/Stores/StorageProvider.ts

@@ -28,4 +28,7 @@ export interface IBridgeStorageProvider extends IAppserviceStorageProvider, ISto
  storeFeedGuids(url: string, ...guid: string[]): Promise<void>;
  hasSeenFeed(url: string, ...guid: string[]): Promise<boolean>;
  hasSeenFeedGuid(url: string, guid: string): Promise<boolean>;
+ addRoomHasActiveConnections(roomId: string): void;
+ removeRoomHasActiveConnections(roomId: string): void;
+ getAllRoomsWithActiveConnections(): Promise<string[]>;
 }