Add network chaos to fork tests (#1570)

* Add network chaos to fork tests

* Set back to 20 epochs

* Add streams to fork test

* Add DB chaos by locking DB file

* Refactor chaos

* Clone workers in streams

* Temporarily turn on device sync and multiple client instances

* Properly catch missing groups

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

This reverts commit 76fcd9b.

Author: neekolas

chaos/db.ts

@@ -0,0 +1,96 @@
+import type { ChaosProvider } from "@chaos/provider";
+import type { WorkerManager } from "@workers/manager";
+
+export type DbChaosConfig = {
+  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
+};
+
+export class DbChaos implements ChaosProvider {
+  config: DbChaosConfig;
+  activeLocks = new Map<string, Promise<void>>();
+  interval?: NodeJS.Timeout;
+
+  constructor(config: DbChaosConfig) {
+    validateConfig(config);
+    this.config = config;
+  }
+
+  start(workers: WorkerManager): Promise<void> {
+    const { minLockTime, maxLockTime, lockInterval, impactedWorkerPercentage } =
+      this.config;
+    console.log(
+      `Starting DB Chaos:
+      Locking for ${minLockTime}ms - ${maxLockTime}ms
+      Interval: ${lockInterval}ms`,
+    );
+    this.interval = setInterval(() => {
+      for (const worker of workers.getAll()) {
+        if (Math.random() * 100 > impactedWorkerPercentage) {
+          continue;
+        }
+        const duration = Math.floor(
+          minLockTime + Math.random() * (maxLockTime - minLockTime),
+        );
+
+        const lockKey = `${worker.name}-${worker.installationId}`;
+
+        // Only lock if not already locked
+        if (!this.activeLocks.has(lockKey)) {
+          console.log(
+            `[db-chaos] Locking ${worker.name} database for ${duration}ms`,
+          );
+
+          // Call the lockDB method on the worker and track it
+          const lockPromise = worker.worker
+            .lockDB(duration)
+            .catch((err: unknown) => {
+              console.warn(err);
+            })
+            .finally(() => {
+              this.activeLocks.delete(lockKey);
+            });
+
+          this.activeLocks.set(lockKey, lockPromise);
+        }
+      }
+    }, lockInterval);
+
+    return Promise.resolve();
+  }
+
+  async stop() {
+    console.log("Stopping DB Chaos");
+    if (this.interval) {
+      clearInterval(this.interval);
+    }
+
+    // Wait for all the existing locks to complete
+    await Promise.allSettled(Array.from(this.activeLocks.values()));
+  }
+}
+
+function validateConfig(config: DbChaosConfig): void {
+  if (config.minLockTime > config.maxLockTime) {
+    throw new Error(
+      "Minimum lock time cannot be greater than maximum lock time",
+    );
+  }
+
+  if (
+    config.impactedWorkerPercentage < 0 ||
+    config.impactedWorkerPercentage > 100
+  ) {
+    throw new Error("Impacted worker percentage must be between 0 and 100");
+  }
+
+  if (!config.lockInterval) {
+    throw new Error("Lock interval must be defined");
+  }
+
+  if (config.impactedWorkerPercentage === undefined) {
+    throw new Error("Impacted worker percentage must be defined");
+  }
+}

chaos/network.ts

@@ -0,0 +1,95 @@
+import type { ChaosProvider } from "@chaos/provider";
+import type { DockerContainer } from "network-stability/container";
+
+export type NetworkChaosConfig = {
+  delayMin: number; // Minimum delay in ms
+  delayMax: number; // Maximum delay in ms
+  jitterMin: number; // Minimum jitter in ms
+  jitterMax: number; // Maximum jitter in ms
+  lossMin: number; // Minimum packet loss percentage (0-100)
+  lossMax: number; // Maximum packet loss percentage (0-100)
+  interval: number; // How often to apply chaos in ms
+};
+
+export class NetworkChaos implements ChaosProvider {
+  config: NetworkChaosConfig;
+  interval?: NodeJS.Timeout;
+  nodes: DockerContainer[];
+
+  constructor(config: NetworkChaosConfig, nodes: DockerContainer[]) {
+    this.config = config;
+    this.nodes = nodes;
+  }
+
+  start(): Promise<void> {
+    console.log(`Starting network chaos:
+      Nodes: ${this.nodes.map((node) => node.name).join(", ")}
+      Delay: ${this.config.delayMin}ms - ${this.config.delayMax}ms
+      Jitter: ${this.config.jitterMin}ms - ${this.config.jitterMax}ms
+      Loss: ${this.config.lossMin}% - ${this.config.lossMax}%
+      Interval: ${this.config.interval}ms`);
+
+    validateContainers(this.nodes);
+    this.clearAll();
+
+    this.interval = setInterval(() => {
+      for (const node of this.nodes) {
+        this.applyToNode(node);
+      }
+    }, this.config.interval);
+
+    return Promise.resolve();
+  }
+
+  private applyToNode(node: DockerContainer) {
+    const { delayMin, delayMax, jitterMin, jitterMax, lossMin, lossMax } =
+      this.config;
+    const delay = Math.floor(delayMin + Math.random() * (delayMax - delayMin));
+    const jitter = Math.floor(
+      jitterMin + Math.random() * (jitterMax - jitterMin),
+    );
+    const loss = lossMin + Math.random() * (lossMax - lossMin);
+
+    try {
+      node.addJitter(delay, jitter);
+      node.addLoss(loss);
+    } catch (err) {
+      console.warn(`[chaos] Error applying netem on ${node.name}:`, err);
+    }
+  }
+
+  clearAll() {
+    for (const node of this.nodes) {
+      try {
+        node.clearLatency();
+      } catch (err) {
+        console.warn(`[chaos] Error clearing latency on ${node.name}:`, err);
+      }
+    }
+  }
+
+  stop(): Promise<void> {
+    if (this.interval) {
+      clearInterval(this.interval);
+    }
+
+    this.clearAll();
+
+    return Promise.resolve();
+  }
+}
+
+const validateContainers = (allNodes: DockerContainer[]) => {
+  for (const node of allNodes) {
+    try {
+      // Test if container exists by trying to get its IP
+      if (!node.ip || !node.veth) {
+        throw new Error(`Container ${node.name} has no IP address`);
+      }
+    } catch {
+      throw new Error(
+        `Docker container ${node.name} is not running. Network chaos requires local multinode setup (./dev/up).`,
+      );
+    }
+  }
+};

chaos/provider.ts

@@ -0,0 +1,10 @@
+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

@@ -0,0 +1,41 @@
+import type { ChaosProvider } from "@chaos/provider";
+import { typeofStream, type WorkerClient } from "@workers/main";
+import type { WorkerManager } from "@workers/manager";
+
+export type StreamsConfig = {
+  cloned: boolean; // Should the stream be run against the workers used in the tests, or a cloned client instance?
+};
+
+export class StreamsChaos implements ChaosProvider {
+  workers?: WorkerClient[];
+  config: StreamsConfig;
+
+  constructor(config: StreamsConfig) {
+    this.config = config;
+  }
+
+  async start(workers: WorkerManager) {
+    console.log("Starting StreamsChaos");
+    let allWorkers = workers.getAll().map((w) => w.worker);
+    if (this.config.cloned) {
+      allWorkers = await Promise.all(allWorkers.map((w) => w.clone()));
+    }
+
+    this.workers = allWorkers;
+    for (const worker of allWorkers) {
+      worker.startStream(typeofStream.Message);
+    }
+
+    return Promise.resolve();
+  }
+
+  stop() {
+    if (this.workers) {
+      for (const worker of this.workers) {
+        worker.stopStreams();
+      }
+    }
+
+    return Promise.resolve();
+  }
+}

cli/gen.ts

@@ -62,9 +62,9 @@ EXAMPLES:
   yarn gen --help
 
 PRESET COMMANDS:
-  yarn update:local      Update 500 inboxes for local testing
-  yarn update:prod       Update inboxes for production testing
-  yarn restart:prod      Restart production installations (force recreate)
+  yarn gen update:local      Update 500 inboxes for local testing
+  yarn gen update:prod       Update inboxes for production testing
+  yarn gen restart:prod      Restart production installations (force recreate)
 
 For more information, see: cli/readme.md
 `);