Revert "Temporarily turn on device sync and multiple client instances"

This reverts commit 76fcd9b.

Author: neekolas

chaos/db.ts

@@ -2,9 +2,9 @@ import type { ChaosProvider } from "@chaos/provider";
 import type { WorkerManager } from "@workers/manager";
 
 export type DbChaosConfig = {
-  minLockTime: number;
-  maxLockTime: number;
-  lockInterval: number;
+  minLockTime: number; // Minimum duration in milliseconds to lock the database
+  maxLockTime: number; // Maximum duration in milliseconds to lock the database
+  lockInterval: number; // Interval in milliseconds between lock attempts
   impactedWorkerPercentage: number; // number between 0 and 100 for what % of workers to lock on each run
 };
 

chaos/network.ts

@@ -11,8 +11,6 @@ export type NetworkChaosConfig = {
   interval: number; // How often to apply chaos in ms
 };
 
-// import type { Worker } from "@workers/manager";
-
 export class NetworkChaos implements ChaosProvider {
   config: NetworkChaosConfig;
   interval?: NodeJS.Timeout;

chaos/provider.ts

@@ -1,5 +1,9 @@
 import { type WorkerManager } from "@workers/manager";
 
+// Generic interface for the Chaos Provider.
+// A Chaos Provider is started after the test has been setup, but before we start performing actions
+// It is stopped after the core of the test has been completed, and should remove all chaos so that final
+// validations can be performed cleanly.
 export interface ChaosProvider {
   start(workers: WorkerManager): Promise<void>;
   stop(): Promise<void>;

chaos/streams.ts

@@ -3,7 +3,7 @@ import { typeofStream, type WorkerClient } from "@workers/main";
 import type { WorkerManager } from "@workers/manager";
 
 export type StreamsConfig = {
-  cloned: boolean;
+  cloned: boolean; // Should the stream be run against the workers used in the tests, or a cloned client instance?
 };
 
 export class StreamsChaos implements ChaosProvider {

forks/cli.ts

@@ -14,6 +14,7 @@ import {
   type RuntimeConfig,
 } from "./config";
 
+// CLI options for fork testing
 interface ForkOptions {
   count: number; // Number of times to run the process (default: 100)
   cleanAll: boolean; // Clean all raw logs before starting
@@ -223,7 +224,7 @@ async function main() {
       type: "string",
       choices: ["local", "dev", "production"] as const,
       default:
-        (process.env.XMTP_ENV as "local" | "dev" | "production") || "dev",
+        (process.env.XMTP_ENV as "local" | "dev" | "production") || "local",
       describe: "XMTP environment",
     })
     .option("network-chaos-level", {

forks/config.ts

@@ -12,11 +12,6 @@ export const workerNames = [
   "random3",
   "random4",
   "random5",
-  "random1",
-  "random2",
-  "random3",
-  "random4",
-  "random5",
 ] as string[];
 
 // Operations configuration - enable/disable specific operations
@@ -28,10 +23,11 @@ export const epochRotationOperations = {
 export const otherOperations = {
   createInstallation: true, // creates a new installation for a random worker
   sendMessage: true, // sends a message to the group
-  sync: false, // syncs the group
+  sync: true, // syncs the group
 };
 export const randomInboxIdsCount = 50; // How many inboxIds to use randomly in the add/remove operations
 export const installationCount = 2; // How many installations to use randomly in the createInstallation operations
+export const maxConsecutiveFailures = 5; // Maximum number of times we throw an error verifying the results of a batch of operations
 export const testName = "forks";
 
 // Network chaos configuration

forks/forks.test.ts

@@ -13,6 +13,7 @@ import {
   epochRotationOperations,
   getConfigFromEnv,
   installationCount,
+  maxConsecutiveFailures,
   multinodeContainers,
   NODE_VERSION,
   otherOperations,
@@ -149,7 +150,8 @@ describe(testName, () => {
           let currentEpoch = 0n;
           let numConsecutiveFailures = 0;
 
-          while (currentEpoch < targetEpoch && numConsecutiveFailures < 5) {
+          // Run until we reach the target epoch or we hit 5 consecutive failures.
+          while (currentEpoch < targetEpoch) {
             const parallelOperationsArray = Array.from(
               { length: parallelOperations },
               () =>
@@ -198,6 +200,8 @@ describe(testName, () => {
                   }
                 })(),
             );
+
+            // We want to wait for all operations to complete, but ignore any errors which may be caused by the chaos
             const results = await Promise.allSettled(parallelOperationsArray);
             for (const result of results) {
               if (result.status === "rejected") {
@@ -221,6 +225,9 @@ describe(testName, () => {
             } catch (e) {
               console.error(`Group ${groupIndex + 1} operation failed:`, e);
               numConsecutiveFailures++;
+              if (numConsecutiveFailures >= maxConsecutiveFailures) {
+                throw e;
+              }
             }
           }
 
@@ -232,6 +239,7 @@ describe(testName, () => {
     } catch (e: any) {
       const msg = `Error during fork testing: ${e}`;
       console.error(msg);
+      // This will fail the test if there were too many failures
       expect.fail(msg);
     } finally {
       clearInterval(verifyInterval);

helpers/versions.ts

@@ -275,8 +275,6 @@ export const regressionClient = async (
       loggingLevel,
       apiUrl,
       appVersion: APP_VERSION,
-      historySyncUrl: "http://localhost:5558",
-      disableDeviceSync: false,
       codecs: [new ReactionCodec(), new ReplyCodec()],
     });
   } catch (error) {

multinode/docker-compose.yml

@@ -113,16 +113,16 @@ services:
       - mlsdb
       - validation
 
-  validation:
-    image: ghcr.io/xmtp/mls-validation-service:main
-    platform: linux/amd64
-
   history-server:
     image: ghcr.io/xmtp/message-history-server:main
     platform: linux/amd64
     ports:
       - 5558:5558
 
+  validation:
+    image: ghcr.io/xmtp/mls-validation-service:main
+    platform: linux/amd64
+
   db:
     image: postgres:13
     environment:

workers/manager.ts

@@ -271,10 +271,6 @@ export class WorkerManager implements IWorkerManager {
       await group.addMembers(extraMembers);
     }
 
-    for (const member of memberList) {
-      await group.addAdmin(member);
-    }
-
     return group as Group;
   }
 
@@ -516,7 +512,7 @@ export async function getWorkers(
       (v) => v.nodeBindings,
     );
   }
-  let successfulResults: Worker[] = [];
+  let workerPromises: Promise<Worker>[] = [];
   let descriptors: string[] = [];
 
   // Handle different input types
@@ -528,75 +524,68 @@ export async function getWorkers(
           : getFixedNames(workers)
         : workers;
     descriptors = names;
-    for (const descriptor of descriptors) {
-      successfulResults.push(
-        await manager.createWorker(
-          descriptor,
-          sdkVersions[Math.floor(Math.random() * sdkVersions.length)],
-        ),
-      );
-    }
+    workerPromises = descriptors.map((descriptor) =>
+      manager.createWorker(
+        descriptor,
+        sdkVersions[Math.floor(Math.random() * sdkVersions.length)],
+      ),
+    );
   } else {
     // Record input - apply versioning if requested
     let entries = Object.entries(workers);
 
     descriptors = entries.map(([descriptor]) => descriptor);
-    for (const descriptor of descriptors) {
-      successfulResults.push(
-        await manager.createWorker(
-          descriptor,
-          sdkVersions[Math.floor(Math.random() * sdkVersions.length)],
-        ),
-      );
-    }
+    workerPromises = entries.map(([descriptor, apiUrl]) =>
+      manager.createWorker(descriptor, sdkVersions[0], apiUrl),
+    );
+  }
+
+  // Only use progress bar if there are more than 50 workers
+  const useProgressBar = workerPromises.length > 50;
+  let progressBar: ProgressBar | undefined;
+
+  if (useProgressBar) {
+    progressBar = new ProgressBar(
+      workerPromises.length,
+      `Initializing ${workerPromises.length} workers...`,
+    );
+    // Show initial progress immediately
+    progressBar.update(0);
   }
 
-  // // Only use progress bar if there are more than 50 workers
-  // const useProgressBar = workerPromises.length > 50;
-  // let progressBar: ProgressBar | undefined;
-
-  // if (useProgressBar) {
-  //   progressBar = new ProgressBar(
-  //     workerPromises.length,
-  //     `Initializing ${workerPromises.length} workers...`,
-  //   );
-  //   // Show initial progress immediately
-  //   progressBar.update(0);
-  // }
-
-  // // Track all workers in parallel and update progress as each completes
-  // let completedCount = 0;
-  // const results = await Promise.allSettled(
-  //   workerPromises.map(async (workerPromise) => {
-  //     try {
-  //       const worker = await workerPromise;
-  //       completedCount++;
-  //       if (useProgressBar && progressBar) {
-  //         progressBar.update(completedCount);
-  //       }
-  //       return worker;
-  //     } catch (error) {
-  //       completedCount++;
-  //       if (useProgressBar && progressBar) {
-  //         progressBar.update(completedCount);
-  //       }
-  //       throw error;
-  //     }
-  //   }),
-  // );
-
-  // // Check for any failures
-  // const failedResults = results.filter(
-  //   (result) => result.status === "rejected",
-  // );
-  // if (failedResults.length > 0) {
-  //   throw failedResults[0].reason;
-  // }
-
-  // // Extract successful results
-  // const successfulResults = results.map(
-  //   (result) => (result as PromiseFulfilledResult<Worker>).value,
-  // );
+  // Track all workers in parallel and update progress as each completes
+  let completedCount = 0;
+  const results = await Promise.allSettled(
+    workerPromises.map(async (workerPromise) => {
+      try {
+        const worker = await workerPromise;
+        completedCount++;
+        if (useProgressBar && progressBar) {
+          progressBar.update(completedCount);
+        }
+        return worker;
+      } catch (error) {
+        completedCount++;
+        if (useProgressBar && progressBar) {
+          progressBar.update(completedCount);
+        }
+        throw error;
+      }
+    }),
+  );
+
+  // Check for any failures
+  const failedResults = results.filter(
+    (result) => result.status === "rejected",
+  );
+  if (failedResults.length > 0) {
+    throw failedResults[0].reason;
+  }
+
+  // Extract successful results
+  const successfulResults = results.map(
+    (result) => (result as PromiseFulfilledResult<Worker>).value,
+  );
 
   // Add all successful workers to the manager
   for (const worker of successfulResults) {
@@ -625,11 +614,12 @@ function getNextFolderName(): string {
   const dataPath = path.resolve(process.cwd(), ".data");
   let folder = "a";
   if (fs.existsSync(dataPath)) {
-    const chars = "abcdefghijklmnopqrstuvwxyz0123456789";
-    folder = Array.from(
-      { length: 32 },
-      () => chars[Math.floor(Math.random() * chars.length)],
-    ).join("");
+    const existingFolders = fs
+      .readdirSync(dataPath)
+      .filter((f) => /^[a-z]$/.test(f));
+    folder = String.fromCharCode(
+      "a".charCodeAt(0) + (existingFolders.length % 26),
+    );
   }
   return folder;
 }