chore: upgrade interop tests

.github/workflows/ci.yml

@@ -122,14 +122,14 @@ jobs:
     uses: ./.github/workflows/test-node.yml
     secrets: inherit
     with:
-      nim_wakunode_image: ${{ inputs.nim_wakunode_image || 'wakuorg/nwaku:v0.31.0' }}
+      nim_wakunode_image: ${{ inputs.nim_wakunode_image || 'wakuorg/nwaku:v0.34.0' }}
       test_type: node
       allure_reports: true
 
   node_optional:
     uses: ./.github/workflows/test-node.yml
     with:
-      nim_wakunode_image: ${{ inputs.nim_wakunode_image || 'wakuorg/nwaku:v0.31.0' }}
+      nim_wakunode_image: ${{ inputs.nim_wakunode_image || 'wakuorg/nwaku:v0.34.0' }}
       test_type: node-optional
 
   node_with_nwaku_master:

.github/workflows/test-node.yml

@@ -33,6 +33,7 @@ env:
 jobs:
   node:
     runs-on: ubuntu-latest
+    timeout-minutes: 60  # Add a 1-hour timeout to fail faster
     env:
       WAKUNODE_IMAGE: ${{ inputs.nim_wakunode_image }}
       ALLURE_REPORTS: ${{ inputs.allure_reports }}

packages/tests/.mocharc.cjs

@@ -7,7 +7,8 @@ const config = {
     'loader=ts-node/esm'
   ],
   exit: true,
-  retries: 4
+  retries: 2,
+  timeout: 150_000 
 };
 
 if (process.env.CI) {

packages/tests/src/lib/dockerode.ts

@@ -18,7 +18,7 @@ export default class Dockerode {
   public containerId?: string;
 
   private static network: Docker.Network;
-  private containerIp: string;
+  public readonly containerIp: string;
 
   private constructor(imageName: string, containerIp: string) {
     this.docker = new Docker();
@@ -107,6 +107,7 @@ export default class Dockerode {
     const container = await this.docker.createContainer({
       Image: this.IMAGE_NAME,
       HostConfig: {
+        NetworkMode: NETWORK_NAME,
         AutoRemove: true,
         PortBindings: {
           [`${restPort}/tcp`]: [{ HostPort: restPort.toString() }],
@@ -116,6 +117,8 @@ export default class Dockerode {
             [`${discv5UdpPort}/udp`]: [{ HostPort: discv5UdpPort.toString() }]
           })
         },
+        Dns: ["8.8.8.8"],
+        Links: [],
         Mounts: args.rlnRelayEthClientAddress
           ? [
               {
@@ -135,18 +138,19 @@ export default class Dockerode {
           [`${discv5UdpPort}/udp`]: {}
         })
       },
-      Cmd: argsArrayWithIP
-    });
-    await container.start();
-
-    await Dockerode.network.connect({
-      Container: container.id,
-      EndpointConfig: {
-        IPAMConfig: {
-          IPv4Address: this.containerIp
+      Cmd: argsArrayWithIP,
+      NetworkingConfig: {
+        EndpointsConfig: {
+          [NETWORK_NAME]: {
+            IPAMConfig: {
+              IPv4Address: this.containerIp
+            }
+          }
         }
       }
     });
+    await container.start();
+
     const logStream = fs.createWriteStream(logPath);
 
     container.logs(

packages/tests/src/lib/index.ts

@@ -29,24 +29,29 @@ export class ServiceNodesFleet {
     _args?: Args,
     withoutFilter = false
   ): Promise<ServiceNodesFleet> {
-    const serviceNodePromises = Array.from(
-      { length: nodesToCreate },
-      async () => {
-        const node = new ServiceNode(
-          makeLogFileName(mochaContext) +
-            Math.random().toString(36).substring(7)
-        );
+    const nodes: ServiceNode[] = [];
+
+    for (let i = 0; i < nodesToCreate; i++) {
+      const node = new ServiceNode(
+        makeLogFileName(mochaContext) + Math.random().toString(36).substring(7)
+      );
 
-        const args = getArgs(networkConfig, _args);
-        await node.start(args, {
-          retries: 3
-        });
+      const args = getArgs(networkConfig, _args);
 
-        return node;
+      // If this is not the first node and previous node had a nodekey, use its multiaddr as static node
+      if (i > 0) {
+        const prevNode = nodes[i - 1];
+        const multiaddr = await prevNode.getExternalWebsocketMultiaddr();
+        args.staticnode = multiaddr;
       }
-    );
 
-    const nodes = await Promise.all(serviceNodePromises);
+      await node.start(args, {
+        retries: 3
+      });
+
+      nodes.push(node);
+    }
+
     return new ServiceNodesFleet(nodes, withoutFilter, strictChecking);
   }
 
@@ -95,7 +100,24 @@ export class ServiceNodesFleet {
     return relayMessages.every((message) => message);
   }
 
-  public async confirmMessageLength(numMessages: number): Promise<void> {
+  public async confirmMessageLength(
+    numMessages: number,
+    { encryptedPayload }: { encryptedPayload?: boolean } = {
+      encryptedPayload: false
+    }
+  ): Promise<void> {
+    if (encryptedPayload) {
+      const filteredMessageList = Array.from(
+        new Set(
+          this.messageCollector.messageList
+            .filter((msg) => msg.payload?.toString)
+            .map((msg) => msg.payload.toString())
+        )
+      );
+      expect(filteredMessageList.length).to.equal(numMessages);
+      return;
+    }
+
     if (this.strictChecking) {
       await Promise.all(
         this.nodes.map(async (node) =>
@@ -120,7 +142,7 @@ export class ServiceNodesFleet {
 
 class MultipleNodesMessageCollector {
   public callback: (msg: DecodedMessage) => void = () => {};
-  protected messageList: Array<DecodedMessage> = [];
+  public readonly messageList: Array<DecodedMessage> = [];
   public constructor(
     private messageCollectors: MessageCollector[],
     private relayNodes?: ServiceNode[],
@@ -170,21 +192,21 @@ class MultipleNodesMessageCollector {
     }
   ): boolean {
     if (this.strictChecking) {
-      return this.messageCollectors.every((collector) => {
+      return this.messageCollectors.every((collector, _i) => {
         try {
           collector.verifyReceivedMessage(index, options);
-          return true; // Verification successful
+          return true;
         } catch (error) {
-          return false; // Verification failed, continue with the next collector
+          return false;
         }
       });
     } else {
-      return this.messageCollectors.some((collector) => {
+      return this.messageCollectors.some((collector, _i) => {
         try {
           collector.verifyReceivedMessage(index, options);
-          return true; // Verification successful
+          return true;
         } catch (error) {
-          return false; // Verification failed, continue with the next collector
+          return false;
         }
       });
     }
@@ -204,34 +226,106 @@ class MultipleNodesMessageCollector {
     const startTime = Date.now();
     const pubsubTopic = options?.pubsubTopic || DefaultTestPubsubTopic;
     const timeoutDuration = options?.timeoutDuration || 400;
+    const maxTimeout = Math.min(timeoutDuration * numMessages, 30000);
+    const exact = options?.exact || false;
+
+    try {
+      while (Date.now() - startTime < maxTimeout) {
+        // Check if we already have enough messages
+        if (this.messageList.length >= numMessages) {
+          if (exact && this.messageList.length !== numMessages) {
+            log.warn(
+              `Was expecting exactly ${numMessages} messages. Received: ${this.messageList.length}`
+            );
+            return false;
+          }
+          return true;
+        }
+
+        // If we have relay nodes, check their messages
+        if (this.relayNodes) {
+          const nodeMessages = await Promise.all(
+            this.relayNodes.map(async (node) => node.messages(pubsubTopic))
+          );
+
+          // Check if we have enough messages according to strictness
+          const hasEnoughMessages = this.strictChecking
+            ? nodeMessages.every((msgs) => msgs.length >= numMessages)
+            : nodeMessages.some((msgs) => msgs.length >= numMessages);
+
+          if (hasEnoughMessages) {
+            return true;
+          }
+        }
+
+        await delay(10);
+      }
+
+      log.warn(`Timeout waiting for messages after ${maxTimeout}ms`);
+      return false;
+    } catch (error) {
+      log.error("Error in waitForMessages:", error);
+      return false;
+    }
+  }
+
+  /**
+   * Waits for a total number of messages across all nodes using autosharding.
+   */
+  public async waitForMessagesAutosharding(
+    numMessages: number,
+    options?: {
+      contentTopic: string;
+      timeoutDuration?: number;
+      exact?: boolean;
+    }
+  ): Promise<boolean> {
+    const startTime = Date.now();
+    const timeoutDuration = options?.timeoutDuration || 400;
     const exact = options?.exact || false;
 
     while (this.messageList.length < numMessages) {
       if (this.relayNodes) {
         if (this.strictChecking) {
+          // In strict mode, all nodes must have the messages
           const results = await Promise.all(
-            this.relayNodes.map(async (node) => {
-              const msgs = await node.messages(pubsubTopic);
-              return msgs.length >= numMessages;
+            this.messageCollectors.map(async (collector) => {
+              return collector.waitForMessagesAutosharding(
+                numMessages,
+                options
+              );
             })
           );
-          return results.every((result) => result);
+          if (results.every((result) => result)) {
+            return true;
+          }
         } else {
+          // In non-strict mode, at least one node must have the messages
           const results = await Promise.all(
-            this.relayNodes.map(async (node) => {
-              const msgs = await node.messages(pubsubTopic);
-              return msgs.length >= numMessages;
+            this.messageCollectors.map(async (collector) => {
+              return collector.waitForMessagesAutosharding(
+                numMessages,
+                options
+              );
             })
           );
-          return results.some((result) => result);
+          if (results.some((result) => result)) {
+            return true;
+          }
         }
-      }
 
-      if (Date.now() - startTime > timeoutDuration * numMessages) {
-        return false;
-      }
+        if (Date.now() - startTime > timeoutDuration * numMessages) {
+          return false;
+        }
 
-      await delay(10);
+        await delay(10);
+      } else {
+        // If no relay nodes, just wait for messages in the list
+        if (Date.now() - startTime > timeoutDuration * numMessages) {
+          return false;
+        }
+        await delay(10);
+      }
     }
 
     if (exact) {
@@ -241,7 +335,6 @@ class MultipleNodesMessageCollector {
         log.warn(
           `Was expecting exactly ${numMessages} messages. Received: ${this.messageList.length}`
         );
-
         return false;
       }
     } else {

packages/tests/src/lib/service_node.ts

@@ -27,7 +27,7 @@
 const NODE_READY_LOG_LINE = "Node setup complete";
 
 export const DOCKER_IMAGE_NAME =
-  process.env.WAKUNODE_IMAGE || "wakuorg/nwaku:v0.31.0";
+  process.env.WAKUNODE_IMAGE || "wakuorg/nwaku:v0.34.0";
 
 const LOG_DIR = "./log";
 
@@ -355,7 +355,7 @@
   public async restCall<T>(
     endpoint: string,
     method: "GET" | "POST",
     body: any = null,
     processResponse: (response: Response) => Promise<T>
   ): Promise<T> {
     this.checkProcess();
@@ -382,6 +382,15 @@
       throw `Container hasn't started`;
     }
   }
+
+  public async getExternalWebsocketMultiaddr(): Promise<string | undefined> {
+    if (!this.docker?.container) {
+      return undefined;
+    }
+    const containerIp = this.docker.containerIp;
+    const peerId = await this.getPeerId();
+    return `/ip4/${containerIp}/tcp/${this.websocketPort}/ws/p2p/${peerId}`;
+  }
 }
 
 export function defaultArgs(): Args {
@@ -402,3 +411,20 @@
   listenAddresses: string[];
   enrUri?: string;
 }
+
+export async function verifyServiceNodesConnected(
+  nodes: ServiceNode[]
+): Promise<void> {
+  for (const node of nodes) {
+    const peers = await node.peers();
+    log.info(`Service node ${node.containerName} peers:`, peers.length);
+    log.info(`Service node ${node.containerName} peers:`, peers);
+
+    if (nodes.length > 1 && peers.length === 0) {
+      log.error(`Service node ${node.containerName} has no peers connected`);
+      throw new Error(
+        `Service node ${node.containerName} has no peers connected`
+      );
+    }
+  }
+}

packages/tests/src/run-tests.js

@@ -3,7 +3,7 @@ import { promisify } from "util";
 
 const execAsync = promisify(exec);
 
-const WAKUNODE_IMAGE = process.env.WAKUNODE_IMAGE || "wakuorg/nwaku:v0.31.0";
+const WAKUNODE_IMAGE = process.env.WAKUNODE_IMAGE || "wakuorg/nwaku:v0.34.0";
 
 async function main() {
   try {
@@ -40,6 +40,15 @@ async function main() {
 
   mocha.on("exit", (code) => {
     console.log(`Mocha tests exited with code ${code}`);
+    try {
+      execAsync(
+        `docker ps -q -f "ancestor=${WAKUNODE_IMAGE}" | xargs -r docker stop`
+      ).catch((error) => {
+        console.error("Error cleaning up containers:", error);
+      });
+    } catch (error) {
+      console.error("Error cleaning up containers:", error);
+    }
     process.exit(code || 0);
   });
 }