Add simple client discovery phase for pool configurations (#1737)

* Add simple client discovery phase

* fix

* Speed up client connections on startup if connection exists

* Drop bluebird import

* Use native allSettled now that we're on a recent Node version

* Better iteration

* While

* Cunning plan to parallelize room loads with proxy loads.

* Cleanup delay code

* Refactors to parallelize connections

* Check for undefined

* Skip checking if we're not syncing membership

* Add debug comment

* Drop extra awaits

* Preserve ordering in test

* Cleanup

* Cleaner

Author: Half-Shot

changelog.d/1737.bugfix

@@ -0,0 +1,2 @@
+Fix a case where a proxied client connection may get missed until they use the bridge on startup, leading to missed
+messages and logspam.

spec/unit/promiseutil.spec.js

@@ -8,7 +8,6 @@ import { MatrixClient } from "matrix-bot-sdk";
 import { TestIrcServer } from "matrix-org-irc";
 import { IrcConnectionPool } from "../../src/pool-service/IrcConnectionPool";
 import dns from 'node:dns';
-
 // Needed to make tests work on GitHub actions. Node 17+ defaults
 // to IPv6, and the homerunner domain resolves to IPv6, but the
 // runtime doesn't actually support IPv6 🤦
@@ -51,8 +50,12 @@ export class E2ETestMatrixClient extends MatrixClient {
 
             // Check only the keys we care about
             for (const [key, value] of Object.entries(expected)) {
-                console.log(key, value, "---", eventData.content[key]);
-                if (JSON.stringify(eventData.content[key], undefined, 0) !== JSON.stringify(value, undefined, 0)) {
+                const evValue = eventData.content[key] ?? undefined;
+                const sortOrder = value !== null && typeof value === "object" ? Object.keys(value).sort() : undefined;
+                const jsonLeft = JSON.stringify(evValue, sortOrder);
+                const jsonRight = JSON.stringify(value, sortOrder);
+                console.log(jsonLeft, "---", jsonRight);
+                if (jsonLeft !== jsonRight) {
                     return undefined;
                 }
             }

spec/util/e2e-test.ts

@@ -681,6 +681,9 @@ export class IrcBridge {
 
         this.clientPool = new ClientPool(this, this.dataStore, this.ircPoolClient);
 
+        // We can begin discovering clients from the pool immediately.
+        const discoveringClientsPromise = this.clientPool.discoverPoolConnectedClients();
+
         if (this.config.ircService.debugApi.enabled) {
             this.debugApi = new DebugApi(
                 this,
@@ -756,6 +759,9 @@ export class IrcBridge {
         log.info("Syncing relevant membership lists...");
         const memberlistPromises: Promise<void>[] = [];
 
+        // Note in the following section we will be waiting for discoveringClientsPromise
+        // to complete before we execute our first join, this is by design so we don't
+        // acidentally connect the same user twice by doing two mass client create loops.
         this.ircServers.forEach((server) => {
             //  If memberlist-syncing 100s of connections, the scheduler will cause massive
             //  waiting times for connections to be created.
@@ -765,16 +771,17 @@ export class IrcBridge {
 
             // TODO reduce deps required to make MemberListSyncers.
             // TODO Remove injectJoinFn bodge
-            this.memberListSyncers[server.domain] = new MemberListSyncer(
+            const syncer = this.memberListSyncers[server.domain] = new MemberListSyncer(
                 this, this.membershipQueue, this.bridge.getBot(), server, this.appServiceUserId,
-                (roomId: string, joiningUserId: string, displayName: string, isFrontier: boolean) => {
+                async (roomId: string, joiningUserId: string, displayName: string, isFrontier: boolean) => {
                     const req = new BridgeRequest(
                         this.bridge.getRequestFactory().newRequest()
                     );
+                    const isFresh = !this.clientPool.getBridgedClientByUserId(server, joiningUserId);
                     const target = new MatrixUser(joiningUserId);
                     // inject a fake join event which will do M->I connections and
                     // therefore sync the member list
-                    return this.matrixHandler.onJoin(req, {
+                    await this.matrixHandler.onJoin(req, {
                         room_id: roomId,
                         content: {
                             displayname: displayName,
@@ -786,10 +793,17 @@ export class IrcBridge {
                         event_id: "!injected",
                         _frontier: isFrontier
                     }, target);
+                    return isFresh;
                 }
             );
             memberlistPromises.push(
-                this.memberListSyncers[server.domain].sync()
+                // Before we can actually join Matrix users to channels, we need to ensure we've discovered
+                // all the clients already connected to avoid races.
+                syncer.sync().then(() =>
+                    discoveringClientsPromise
+                ).finally(() =>
+                    syncer.joinMatrixUsersToChannels()
+                )
             );
         });
 
@@ -813,14 +827,13 @@ export class IrcBridge {
         log.info("Connecting to IRC networks...");
         await this.connectToIrcNetworks();
 
-        await promiseutil.allSettled(this.ircServers.map((server) => {
+        await Promise.allSettled(this.ircServers.map((server) => {
             // Call MODE on all known channels to get modes of all channels
             return Bluebird.cast(this.publicitySyncer.initModes(server));
         })).catch((err) => {
             log.error('Could not init modes for publicity syncer');
             log.error(err.stack);
         });
-
         await Promise.all(memberlistPromises);
 
         // Reset reconnectIntervals
@@ -962,7 +975,7 @@ export class IrcBridge {
             }
             await this.bridge.getIntent().join(roomId);
         }).map(Bluebird.cast);
-        await promiseutil.allSettled(promises);
+        await Promise.allSettled(promises);
     }
 
     public async sendMatrixAction(room: MatrixRoom, from: MatrixUser|undefined, action: MatrixAction): Promise<void> {

src/bridge/IrcBridge.ts

@@ -1,7 +1,6 @@
 // Controls the logic for determining which membership lists should be synced and
 // handles the sequence of events until the lists are in sync.
 
-import Bluebird from "bluebird";
 import { IrcBridge } from "./IrcBridge";
 import { AppServiceBot, MembershipQueue } from "matrix-appservice-bridge";
 import { IrcServer } from "../irc/IrcServer";
@@ -35,8 +34,15 @@ interface LeaveQueueItem {
     userIds: string[];
 }
 
+interface MemberJoinEntry {
+    roomId: string;
+    displayName: string;
+    userId: string;
+    frontier: boolean
+}
+
 type InjectJoinFn = (roomId: string, joiningUserId: string,
-                     displayName: string, isFrontier: boolean) => PromiseLike<unknown>;
+                     displayName: string, isFrontier: boolean) => PromiseLike<boolean>;
 
 export class MemberListSyncer {
     private syncableRoomsPromise: Promise<RoomInfo[]>|null = null;
@@ -46,6 +52,7 @@ export class MemberListSyncer {
         irc: {[channel: string]: string[]};
         matrix: {[roomId: string]: RoomInfo};
     } = { irc: {}, matrix: {} };
+    private memberEntriesToSync?: MemberJoinEntry[];
 
     constructor(private ircBridge: IrcBridge, private memberQueue: MembershipQueue,
                 private appServiceBot: AppServiceBot, private server: IrcServer,
@@ -64,9 +71,9 @@ export class MemberListSyncer {
         log.info("Found %s syncable rooms (%sms)", rooms.length, Date.now() - start);
         this.leaveIrcUsersFromRooms(rooms);
         start = Date.now();
-        log.info("Joining Matrix users to IRC channels...");
-        await this.joinMatrixUsersToChannels(rooms, this.injectJoinFn);
-        log.info("Joined Matrix users to IRC channels. (%sms)", Date.now() - start);
+        log.info("Collecting all Matrix users in all channel rooms...");
+        await this.collectMatrixUsersToJoinToChannels(rooms);
+        log.info("Collected all Matrix users in all channel rooms. (%sms)", Date.now() - start);
         // NB: We do not need to explicitly join IRC users to Matrix rooms
         // because we get all of the NAMEs/JOINs as events when we connect to
         // the IRC server. This effectively "injects" the list for us.
@@ -234,8 +241,7 @@ export class MemberListSyncer {
         return this.syncableRoomsPromise;
     }
 
-    private async joinMatrixUsersToChannels(rooms: RoomInfo[], injectJoinFn: InjectJoinFn) {
-
+    private async collectMatrixUsersToJoinToChannels(rooms: RoomInfo[]) {
         // filter out rooms listed in the rules
         const filteredRooms: RoomInfo[] = [];
         rooms.forEach((roomInfo) => {
@@ -260,7 +266,7 @@ export class MemberListSyncer {
 
         // map the filtered rooms to a list of users to join
         // [Room:{reals:[uid,uid]}, ...] => [{uid,roomid}, ...]
-        const entries: { roomId: string; displayName: string; userId: string; frontier: boolean}[] = [];
+        const entries: MemberJoinEntry[] = [];
         const idleRegex = this.server.ignoreIdleUsersOnStartupExcludeRegex;
         for (const roomInfo of filteredRooms) {
             for (const uid of roomInfo.realJoinedUsers) {
@@ -300,37 +306,52 @@ export class MemberListSyncer {
         });
 
         log.debug("Got %s matrix join events to inject.", entries.length);
+        this.memberEntriesToSync = entries;
+    }
+
+    public async joinMatrixUsersToChannels() {
+        const start = Date.now();
+        log.info("Joining all Matrix users in all channel rooms");
+        const entries = this.memberEntriesToSync;
+        if (entries === undefined) {
+            // Can be expected if syncing is off.
+            log.info(`joinMatrixUsersToChannels: No entries collected for joining`);
+            return;
+        }
         this.usersToJoin = entries.length;
-        const d = promiseutil.defer();
-        // take the first entry and inject a join event
-        const joinNextUser = () => {
-            const entry = entries.shift();
-            if (!entry) {
-                d.resolve();
-                return;
-            }
+
+        let entry: MemberJoinEntry|undefined;
+        const floodDelayMs = this.server.getMemberListFloodDelayMs();
+        // eslint-disable-next-line no-cond-assign
+        while (entry = entries.shift()) {
             this.usersToJoin--;
             if (entry.userId.startsWith("@-")) {
-                joinNextUser();
-                return;
+                // Ignore guest users.
+                continue;
             }
             log.debug(
                 "Injecting join event for %s in %s (%s left) is_frontier=%s",
                 entry.userId, entry.roomId, entries.length, entry.frontier
             );
-            Bluebird.cast(injectJoinFn(entry.roomId, entry.userId, entry.displayName, entry.frontier)).timeout(
-                this.server.getMemberListFloodDelayMs()
-            ).then(() => {
-                joinNextUser();
-            }).catch(() => {
-                // discard error, this will be due to timeouts which we don't want to log
-                joinNextUser();
-            })
+            try {
+                // Inject a join to connect the user. We wait up til the delay time,
+                // and then just connect the next user.
+                // If the user connects *faster* than the delay time, then we need
+                // to delay for the remainder.
+                const delayPromise = promiseutil.delay(floodDelayMs);
+                await Promise.race([
+                    delayPromise,
+                    this.injectJoinFn(entry.roomId, entry.userId, entry.displayName, entry.frontier),
+                ]);
+                await delayPromise;
+            }
+            catch (ex) {
+                // injectJoinFn may fail due to failure to get a client (user may be banned)
+                // or any other reason, we should continue to iterate regardless
+                log.debug(`Failed to inject join for ${entry.userId} ${entry.roomId}`, ex);
+            }
         }
-
-        joinNextUser();
-
-        return d.promise;
+        log.info("Joining all Matrix users in all channel rooms. (%sms)", Date.now() - start);
     }
 
     public leaveIrcUsersFromRooms(rooms: RoomInfo[]) {

src/bridge/MemberListSyncer.ts

@@ -128,6 +128,33 @@ export class ClientPool {
         return this.botClients.get(server.domain);
     }
 
+    /**
+     * Discover any clients connected from a previous session when
+     * using the pool
+     */
+    public async discoverPoolConnectedClients() {
+        if (!this.redisPool) {
+            return;
+        }
+
+        for await (const connection of this.redisPool.getPreviouslyConnectedClients()) {
+            if (connection.clientId === 'bot') {
+                // The bot will be connected via the usual process.
+                continue;
+            }
+            // XXX: This is a safe assumption *for now* but when the proxy supports multiple
+            // servers this will break!
+            const server = this.ircBridge.getServers()[0];
+            try {
+                await this.getBridgedClient(server, connection.clientId);
+                log.info(`Connected previously connected user ${connection.clientId}`);
+            }
+            catch (ex) {
+                log.warn(`Failed to connect ${connection.clientId}, who is connected through the proxy`, ex);
+            }
+        }
+    }
+
     public async loginToServer(server: IrcServer): Promise<BridgedClient> {
         let bridgedClient = this.getBot(server);
         if (!bridgedClient) {

src/irc/ClientPool.ts

@@ -53,6 +53,25 @@ export class IrcPoolClient extends (EventEmitter as unknown as new () => TypedEm
         log.debug(`Sent command in ${type}: ${payload}`);
     }
 
+
+    public async *getPreviouslyConnectedClients(): AsyncGenerator<RedisIrcConnection> {
+        let count = 0;
+        for (const [clientId, clientAddressPair] of
+            Object.entries(await this.redis.hgetall(REDIS_IRC_POOL_CONNECTIONS))) {
+            const [, address, portStr] = /(.+):(\d+)/.exec(clientAddressPair) || [];
+
+            // Doing this here allows us to frontload the work that would be done in createOrGetIrcSocket
+            const state = await IrcClientRedisState.create(this.redis, clientId, false);
+            const connection = new RedisIrcConnection(this, clientId, state);
+            const port = parseInt(portStr);
+            connection.setConnectionInfo({ localPort: port, localIp: address, clientId });
+            this.connections.set(clientId, Promise.resolve(connection));
+            yield connection;
+            count++;
+        }
+        log.info(`Found ${count} previously connected clients`);
+    }
+
     public async createOrGetIrcSocket(clientId: string, netOpts: ConnectionCreateArgs): Promise<RedisIrcConnection> {
         const existingConnection = this.connections.get(clientId);
         if (existingConnection) {

src/pool-service/IrcPoolClient.ts

@@ -44,11 +44,3 @@ export function defer<T>(): Defer<T> {
 export function delay(ms: number): Promise<void> {
     return new Promise(resolve => setTimeout(resolve, ms));
 }
-
-/**
- * Waits for all Promises to be "settled". A Promise is settled when
- * it resolves or rejects.
- */
-export function allSettled<T>(promises: Bluebird<T>[]): Promise<Bluebird.Inspection<T>[]> {
-    return Promise.all(promises.map(p => p.reflect()));
-}