[keyserver] stall secondary nodes until database version is equal to …

…or greater than latest wrapped in transaction request blocking migration version.

Summary:
This blocks the secondary nodes from accepting any other traffic other than health checks while the current `dbVersion` is less than the latest migration specifying `wrap_in_transaction_and_block_requests`.

Depends on D13212

Test Plan:
I created two migrations and a new keyserver docker image.

Both migrations included `await sleep`. The first migration specified `wrap_in_transaction_and_block_requests` and the second migration specified `run_simultaneously_with_requests`.

While the first migration was running (and the primary node was unavailable due to running the migration), only health checks were available on the load balancer, meaning that the secondary nodes were only accepting health check traffic.

I also console logged to ensure the loop was running. On the second migration however, all endpoints were available for secondary nodes.

Reviewers: ashoat

Reviewed By: ashoat

Subscribers: ashoat, tomek

Differential Revision: https://phab.comm.dev/D13213

keyserver/src/database/migration-config.js

@@ -1058,6 +1058,17 @@ const versions: $ReadOnlyArray<number> = migrations.map(
 );
 const newDatabaseVersion: number = Math.max(...versions);
 
+const wrapInTransactionAndBlockRequestsVersions = migrations
+  .filter(
+    migration =>
+      !migration.migrationType ||
+      migration.migrationType === 'wrap_in_transaction_and_block_requests',
+  )
+  .map(migration => migration.version);
+const latestWrapInTransactionAndBlockRequestsVersion: number = Math.max(
+  ...wrapInTransactionAndBlockRequestsVersions,
+);
+
 async function writeJSONToFile(data: any, filePath: string): Promise<void> {
   console.warn(`updating ${filePath} to ${JSON.stringify(data)}`);
   const fileHandle = await fs.promises.open(filePath, 'w');
@@ -1190,6 +1201,7 @@ function isDockerEnvironment(): boolean {
 export {
   migrations,
   newDatabaseVersion,
+  latestWrapInTransactionAndBlockRequestsVersion,
   createOlmAccounts,
   saveNewOlmAccounts,
 };

keyserver/src/keyserver.js

@@ -17,7 +17,10 @@ import { qrCodeLinkURL } from 'lib/facts/links.js';
 import { identityDeviceTypes } from 'lib/types/identity-service-types.js';
 import { isDev } from 'lib/utils/dev-utils.js';
 import { ignorePromiseRejections } from 'lib/utils/promises.js';
+import sleep from 'lib/utils/sleep.js';
 
+import { fetchDBVersion } from './database/db-version.js';
+import { latestWrapInTransactionAndBlockRequestsVersion } from './database/migration-config.js';
 import { migrate } from './database/migrations.js';
 import { jsonEndpoints } from './endpoints.js';
 import { logEndpointMetrics } from './middleware/endpoint-profiling.js';
@@ -95,6 +98,16 @@ void (async () => {
   const areEndpointMetricsEnabled =
     process.env.KEYSERVER_ENDPOINT_METRICS_ENABLED;
 
+  const listenAddress = (() => {
+    if (process.env.COMM_LISTEN_ADDR) {
+      return process.env.COMM_LISTEN_ADDR;
+    } else if (process.env.NODE_ENV === 'development') {
+      return undefined;
+    } else {
+      return 'localhost';
+    }
+  })();
+
   if (cluster.isMaster) {
     if (isPrimaryNode) {
       const didMigrationsSucceed: boolean = await migrate();
@@ -192,6 +205,17 @@ void (async () => {
       res.send('OK');
     });
 
+    server.listen(parseInt(process.env.PORT, 10) || 3000, listenAddress);
+
+    if (isSecondaryNode) {
+      let dbVersion;
+      do {
+        await sleep(5000);
+
+        dbVersion = await fetchDBVersion();
+      } while (dbVersion < latestWrapInTransactionAndBlockRequestsVersion);
+    }
+
     // Note - the order of router declarations matters. On prod we have
     // keyserverBaseRoutePath configured to '/', which means it's a catch-all.
     // If we call server.use on keyserverRouter first, it will catch all
@@ -316,17 +340,5 @@ void (async () => {
         res.redirect(newURL);
       });
     }
-
-    const listenAddress = (() => {
-      if (process.env.COMM_LISTEN_ADDR) {
-        return process.env.COMM_LISTEN_ADDR;
-      } else if (process.env.NODE_ENV === 'development') {
-        return undefined;
-      } else {
-        return 'localhost';
-      }
-    })();
-
-    server.listen(parseInt(process.env.PORT, 10) || 3000, listenAddress);
   }
 })();