wip

Author: jordigh

app/common/ConfigAPI.ts

@@ -26,7 +26,7 @@ export class ConfigAPI extends BaseAPI {
   }
 
   public async healthcheck(): Promise<void> {
-    const resp = await this.request(`${this._homeUrl}/status?ready=1`);
+    const resp = await this.request(`${this._homeUrl}/status?allInstancesReady=1`);
     if (!resp.ok) {
       throw new Error(await resp.text());
     }

app/server/lib/FlexServer.ts

@@ -101,6 +101,7 @@ import {AddressInfo} from 'net';
 import fetch from 'node-fetch';
 import * as path from 'path';
 import * as serveStatic from 'serve-static';
+import {v4 as uuidv4} from 'uuid';
 
 // Health checks are a little noisy in the logs, so we don't show them all.
 // We show the first N health checks:
@@ -127,6 +128,29 @@ export interface FlexServerOptions {
   settings?: IGristCoreConfig;
 }
 
+// Not to be confused with health checks from the frontend, these
+// request/response pairs are internal checks between Grist instances
+// in multi-server environments
+interface ServerHealthcheckRequest {
+  id: string;
+  checkReady: boolean;
+}
+interface ServerHealthcheckResponse {
+  instanceId: string;
+  requestId: string;
+  healthy: boolean;
+}
+
+// For keeping track of pending health checks for all other servers
+// for each request that was broadcast to all of them.
+interface PendingServerHealthCheck {
+  expectedCount: number;
+  responses: Record<string, boolean>;
+  resolve: (res: boolean) => void;
+  reject: (err: Error) => void;
+  timeout: NodeJS.Timeout;
+}
+
 const noop: express.RequestHandler = (req, res, next) => next();
 
 export class FlexServer implements GristServer {
@@ -213,6 +237,8 @@ export class FlexServer implements GristServer {
   private _emitNotifier = new EmitNotifier();
   private _testPendingNotifications: number = 0;
   private _latestVersionAvailable?: LatestVersionAvailable;
+  private _instanceId: string;
+  private _pendingServerHealthChecks: Map<string, PendingServerHealthCheck>;
 
   constructor(public port: number, public name: string = 'flexServer',
               public readonly options: FlexServerOptions = {}) {
@@ -274,6 +300,9 @@ export class FlexServer implements GristServer {
       this.setLatestVersionAvailable(latestVersionAvailable);
     });
 
+    this._pendingServerHealthChecks = new Map<string, PendingServerHealthCheck>();
+    this._registerGristInstance();
+
     // The electron build is not supported at this time, but this stub
     // implementation of electronServerMethods is present to allow kicking
     // its tires.
@@ -600,6 +629,34 @@ export class FlexServer implements GristServer {
       if (isParameterOn(req.query.ready)) {
         checks.set('ready', this._isReady);
       }
+      if (isParameterOn(req.query.allInstancesReady)) {
+        const requestId = uuidv4();
+        const client = this._pubSubManager.getClient();
+
+        // If there is no redis, then our current instance is the only instance
+        const allInstances = await client?.smembers('grist-instances') || [this._instanceId];
+
+        const allInstancesPromise: Promise<boolean> = new Promise((resolve, reject) => {
+          const allInstancesTimeout = setTimeout(() => {
+            reject(new Error('Timeout waiting for health responses'));
+            this._pendingServerHealthChecks.delete(requestId);
+          }, timeout);
+
+          this._pendingServerHealthChecks.set(requestId, {
+            responses: {},
+            expectedCount: allInstances.length,
+            resolve,
+            reject,
+            timeout: allInstancesTimeout,
+          });
+        });
+        const request: ServerHealthcheckRequest = {
+          id: requestId,
+          checkReady: true
+        };
+        await this._pubSubManager.publish('healthcheck:requests', JSON.stringify(request));
+        checks.set('allInstancesReady', allInstancesPromise);
+      }
       let extra = '';
       let ok = true;
       // If we had any extra check, collect their status to report them.
@@ -1061,6 +1118,7 @@ export class FlexServer implements GristServer {
     if (this.httpsServer) { this.httpsServer.close(); }
     if (this.housekeeper) { await this.housekeeper.stop(); }
     if (this._jobs)       { await this._jobs.stop(); }
+    await this._pubSubManager.getClient()?.srem('grist-instances', [this._instanceId]);
     await this._shutdown();
     if (this._accessTokens) { await this._accessTokens.close(); }
     // Do this after _shutdown, since DocWorkerMap is used during shutdown.
@@ -2693,6 +2751,40 @@ export class FlexServer implements GristServer {
       },
     });
   }
+
+  private _registerGristInstance() {
+    this._instanceId = process.env.GRIST_INSTANCE_ID || `testInsanceId_${this.port}`;
+    this._pubSubManager.getClient()?.sadd('grist-instances', this._instanceId).catch((err) => {
+      log.error('Failed to contact redis', err);
+    });
+    this._pubSubManager.subscribe('healthcheck:requests', async (message) => {
+      const request: ServerHealthcheckRequest = JSON.parse(message);
+      const response: ServerHealthcheckResponse = {
+        instanceId: this._instanceId || '',
+        requestId: request.id,
+        healthy: !request.checkReady || this._isReady,
+      };
+      log.debug(`Healthcheck request`, response);
+      await this._pubSubManager.publish('healthcheck:responses', JSON.stringify(response));
+    });
+
+    this._pubSubManager.subscribe('healthcheck:responses', (message) => {
+      const response: ServerHealthcheckResponse = JSON.parse(message);
+      const pending = this._pendingServerHealthChecks.get(response.requestId);
+      if (!pending) {
+        return;
+      }
+
+      pending.responses[response.instanceId] = response.healthy;
+
+      if (Object.keys(pending.responses).length === pending.expectedCount) {
+        // All servers have replied. Make it known and clean up.
+        clearTimeout(pending.timeout);
+        pending.resolve(Object.values(pending.responses).every(e => e));
+        this._pendingServerHealthChecks.delete(response.requestId);
+      }
+    });
+  }
 }
 
 /**

app/server/lib/PubSubManager.ts

@@ -29,6 +29,7 @@ export interface IPubSubManager {
   close(): Promise<void>;
   subscribe(channel: string, callback: Callback): UnsubscribeCallback;
   publish(channel: string, message: string): Promise<void>;
+  getClient(): IORedis|undefined;
 }
 
 export type Callback = (message: string) => void;
@@ -72,6 +73,7 @@ abstract class PubSubManagerBase implements IPubSubManager {
    * - In Redis, the channel gets prefixed with getPubSubPrefix() to scope it to the current Redis DB.
    */
   public abstract publish(channel: string, message: string): Promise<void>;
+  public abstract getClient(): IORedis|undefined;
 
   protected abstract _redisSubscribe(channel: string): Promise<void>;
   protected abstract _redisUnsubscribe(channel: string): Promise<void>;
@@ -96,6 +98,7 @@ abstract class PubSubManagerBase implements IPubSubManager {
 
 class PubSubManagerNoRedis extends PubSubManagerBase {
   public async publish(channel: string, message: string) { this._deliverMessage(channel, message); }
+  public getClient(): IORedis|undefined { return; }
   protected async _redisSubscribe(channel: string): Promise<void> {}
   protected async _redisUnsubscribe(channel: string): Promise<void> {}
 }
@@ -131,6 +134,12 @@ class PubSubManagerRedis extends PubSubManagerBase {
     await this._redisPub.publish(this._prefixChannel(channel), message);
   }
 
+  public getClient(): IORedis|undefined {
+    // The redisSub client is already tied listening to a channel, but
+    // the redisPub is "free" for the client to mess around with.
+    return this._redisPub;
+  }
+
   protected async _redisSubscribe(channel: string): Promise<void> {
     await this._redisSub.subscribe(this._prefixChannel(channel));
   }