Refactor chaos

Author: neekolas

chaos/db.ts

@@ -0,0 +1,67 @@
+import type { ChaosProvider } from "@chaos/provider";
+import type { WorkerManager } from "@workers/manager";
+
+export type DbChaosConfig = {
+  minLockTime: number;
+  maxLockTime: number;
+  lockInterval: number;
+};
+
+export class DbChaos implements ChaosProvider {
+  config: DbChaosConfig;
+  activeLocks = new Map<string, Promise<void>>();
+  interval?: NodeJS.Timeout;
+
+  constructor(config: DbChaosConfig) {
+    this.config = config;
+  }
+
+  start(workers: WorkerManager): Promise<void> {
+    const { minLockTime, maxLockTime, lockInterval } = 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()) {
+        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()));
+  }
+}

chaos/network.ts

@@ -0,0 +1,92 @@
+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
+};
+
+// import type { Worker } from "@workers/manager";
+
+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.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);
+    }
+  }
+
+  stop(): Promise<void> {
+    if (this.interval) {
+      clearInterval(this.interval);
+    }
+
+    for (const node of this.nodes) {
+      try {
+        node.clearLatency();
+      } catch (err) {
+        console.warn(`[chaos] Error clearing latency on ${node.name}:`, err);
+      }
+    }
+
+    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,6 @@
+import { type WorkerManager } from "@workers/manager";
+
+export interface ChaosProvider {
+  start(workers: WorkerManager): Promise<void>;
+  stop(): Promise<void>;
+}

chaos/streams.ts

@@ -0,0 +1,26 @@
+import type { ChaosProvider } from "@chaos/provider";
+import { typeofStream } from "@workers/main";
+import type { WorkerManager } from "@workers/manager";
+
+export class StreamsChaos implements ChaosProvider {
+  workers?: WorkerManager;
+  start(workers: WorkerManager) {
+    console.log("Starting StreamsChaos");
+    this.workers = workers;
+    for (const worker of workers.getAll()) {
+      worker.worker.startStream(typeofStream.Message);
+    }
+
+    return Promise.resolve();
+  }
+
+  stop() {
+    if (this.workers) {
+      for (const worker of this.workers.getAll()) {
+        worker.worker.stopStreams();
+      }
+    }
+
+    return Promise.resolve();
+  }
+}