test: add E2E test infrastructure for Redis maintenance scenarios

packages/client/lib/client/enterprise-maintenance-manager.ts

@@ -6,6 +6,7 @@ import { lookup } from "dns/promises";
 import assert from "node:assert";
 import { setTimeout } from "node:timers/promises";
 import RedisSocket from "./socket";
+import diagnostics_channel from "node:diagnostics_channel";
 
 export const MAINTENANCE_EVENTS = {
   PAUSE_WRITING: "pause-writing",
@@ -21,11 +22,24 @@ const PN = {
   FAILED_OVER: "FAILED_OVER",
 };
 
+export type DiagnosticsEvent = {
+  type: string;
+  timestamp: number;
+  data?: Object;
+};
+
 export const dbgMaintenance = (...args: any[]) => {
   if (!process.env.DEBUG_MAINTENANCE) return;
   return console.log("[MNT]", ...args);
 };
 
+export const emitDiagnostics = (event: DiagnosticsEvent) => {
+  if (!process.env.EMIT_DIAGNOSTICS) return;
+
+  const channel = diagnostics_channel.channel("redis.maintenance");
+  channel.publish(event);
+};
+
 export interface MaintenanceUpdate {
   relaxedCommandTimeout?: number;
   relaxedSocketTimeout?: number;
@@ -106,7 +120,21 @@ export default class EnterpriseMaintenanceManager {
 
   #onPush = (push: Array<any>): boolean => {
     dbgMaintenance("ONPUSH:", push.map(String));
-    switch (push[0].toString()) {
+
+    if (!Array.isArray(push) || !["MOVING", "MIGRATING", "MIGRATED", "FAILING_OVER", "FAILED_OVER"].includes(String(push[0]))) {
+      return false;
+    }
+
+    const type = String(push[0]);
+
+    emitDiagnostics({
+          type,
+          timestamp: Date.now(),
+          data: {
+            push: push.map(String),
+          },
+        });
+    switch (type) {
       case PN.MOVING: {
         // [ 'MOVING', '17', '15', '54.78.247.156:12075' ]
         //             ^seq   ^after    ^new ip

packages/client/lib/tests/test-scenario/fault-injector-client.ts

@@ -0,0 +1,152 @@
+import { setTimeout } from "node:timers/promises";
+
+export type ActionType =
+  | "dmc_restart"
+  | "failover"
+  | "reshard"
+  | "sequence_of_actions"
+  | "network_failure"
+  | "execute_rlutil_command"
+  | "execute_rladmin_command"
+  | "migrate"
+  | "bind";
+
+export interface ActionRequest {
+  type: ActionType;
+  parameters?: {
+    bdb_id?: string;
+    [key: string]: unknown;
+  };
+}
+
+export interface ActionStatus {
+  status: string;
+  error: unknown;
+  output: string;
+}
+
+export class FaultInjectorClient {
+  private baseUrl: string;
+  #fetch: typeof fetch;
+
+  constructor(baseUrl: string, fetchImpl: typeof fetch = fetch) {
+    this.baseUrl = baseUrl.replace(/\/+$/, ""); // trim trailing slash
+    this.#fetch = fetchImpl;
+  }
+
+  /**
+   * Lists all available actions.
+   * @throws {Error} When the HTTP request fails or response cannot be parsed as JSON
+   */
+  public listActions<T = unknown>(): Promise<T> {
+    return this.#request<T>("GET", "/action");
+  }
+
+  /**
+   * Triggers a specific action.
+   * @param action The action request to trigger
+   * @throws {Error} When the HTTP request fails or response cannot be parsed as JSON
+   */
+  public triggerAction<T = unknown>(action: ActionRequest): Promise<T> {
+    return this.#request<T>("POST", "/action", action);
+  }
+
+  /**
+   * Gets the status of a specific action.
+   * @param actionId The ID of the action to check
+   * @throws {Error} When the HTTP request fails or response cannot be parsed as JSON
+   */
+  public getActionStatus<T = ActionStatus>(actionId: string): Promise<T> {
+    return this.#request<T>("GET", `/action/${actionId}`);
+  }
+
+  /**
+   * Executes an rladmin command.
+   * @param command The rladmin command to execute
+   * @param bdbId Optional database ID to target
+   * @throws {Error} When the HTTP request fails or response cannot be parsed as JSON
+   */
+  public executeRladminCommand<T = unknown>(
+    command: string,
+    bdbId?: string
+  ): Promise<T> {
+    const cmd = bdbId ? `rladmin -b ${bdbId} ${command}` : `rladmin ${command}`;
+    return this.#request<T>("POST", "/rladmin", cmd);
+  }
+
+  /**
+   * Waits for an action to complete.
+   * @param actionId The ID of the action to wait for
+   * @param options Optional timeout and max wait time
+   * @throws {Error} When the action does not complete within the max wait time
+   */
+  public async waitForAction(
+    actionId: string,
+    {
+      timeoutMs,
+      maxWaitTimeMs,
+    }: {
+      timeoutMs?: number;
+      maxWaitTimeMs?: number;
+    } = {}
+  ): Promise<ActionStatus> {
+    const timeout = timeoutMs || 1000;
+    const maxWaitTime = maxWaitTimeMs || 60000;
+
+    const startTime = Date.now();
+
+    while (Date.now() - startTime < maxWaitTime) {
+      const action = await this.getActionStatus<ActionStatus>(actionId);
+
+      if (["finished", "failed", "success"].includes(action.status)) {
+        return action;
+      }
+
+      await setTimeout(timeout);
+    }
+
+    throw new Error(`Timeout waiting for action ${actionId}`);
+  }
+
+  async #request<T>(
+    method: string,
+    path: string,
+    body?: Object | string
+  ): Promise<T> {
+    const url = `${this.baseUrl}${path}`;
+    const headers: Record<string, string> = {
+      "Content-Type": "application/json",
+    };
+
+    let payload: string | undefined;
+
+    if (body) {
+      if (typeof body === "string") {
+        headers["Content-Type"] = "text/plain";
+        payload = body;
+      } else {
+        headers["Content-Type"] = "application/json";
+        payload = JSON.stringify(body);
+      }
+    }
+
+    const response = await this.#fetch(url, { method, headers, body: payload });
+
+    if (!response.ok) {
+      try {
+        const text = await response.text();
+        throw new Error(`HTTP ${response.status} - ${text}`);
+      } catch {
+        throw new Error(`HTTP ${response.status}`);
+      }
+    }
+
+    try {
+      return (await response.json()) as T;
+    } catch {
+      throw new Error(
+        `HTTP ${response.status} - Unable to parse response as JSON`
+      );
+    }
+  }
+}

packages/client/lib/tests/test-scenario/push-notification.e2e.ts

@@ -0,0 +1,94 @@
+import assert from "node:assert";
+import diagnostics_channel from "node:diagnostics_channel";
+import { FaultInjectorClient } from "./fault-injector-client";
+import {
+  getDatabaseConfig,
+  getDatabaseConfigFromEnv,
+  getEnvConfig,
+  RedisConnectionConfig,
+} from "./test-scenario.util";
+import { createClient } from "../../..";
+import { DiagnosticsEvent } from "../../client/enterprise-maintenance-manager";
+import { before } from "mocha";
+
+describe("Push Notifications", () => {
+  const diagnosticsLog: DiagnosticsEvent[] = [];
+
+  const onMessageHandler = (message: unknown) => {
+    diagnosticsLog.push(message as DiagnosticsEvent);
+  };
+
+  let clientConfig: RedisConnectionConfig;
+  let client: ReturnType<typeof createClient<any, any, any, 3>>;
+  let faultInjectorClient: FaultInjectorClient;
+
+  before(() => {
+    const envConfig = getEnvConfig();
+    const redisConfig = getDatabaseConfigFromEnv(
+      envConfig.redisEndpointsConfigPath
+    );
+
+    faultInjectorClient = new FaultInjectorClient(envConfig.faultInjectorUrl);
+    clientConfig = getDatabaseConfig(redisConfig);
+  });
+
+  beforeEach(async () => {
+    diagnosticsLog.length = 0;
+    diagnostics_channel.subscribe("redis.maintenance", onMessageHandler);
+
+    client = createClient({
+      socket: {
+        host: clientConfig.host,
+        port: clientConfig.port,
+        ...(clientConfig.tls === true ? { tls: true } : {}),
+      },
+      password: clientConfig.password,
+      username: clientConfig.username,
+      RESP: 3,
+      maintPushNotifications: "auto",
+      maintMovingEndpointType: "external-ip",
+      maintRelaxedCommandTimeout: 10000,
+      maintRelaxedSocketTimeout: 10000,
+    });
+
+    client.on("error", (err: Error) => {
+      throw new Error(`Client error: ${err.message}`);
+    });
+
+    await client.connect();
+  });
+
+  afterEach(() => {
+    diagnostics_channel.unsubscribe("redis.maintenance", onMessageHandler);
+    client.destroy();
+  });
+
+  it("should receive MOVING, MIGRATING, and MIGRATED push notifications", async () => {
+    const { action_id: migrateActionId } =
+      await faultInjectorClient.triggerAction<{ action_id: string }>({
+        type: "migrate",
+        parameters: {
+          cluster_index: "0",
+        },
+      });
+
+    await faultInjectorClient.waitForAction(migrateActionId);
+
+    const { action_id: bindActionId } =
+      await faultInjectorClient.triggerAction<{ action_id: string }>({
+        type: "bind",
+        parameters: {
+          cluster_index: "0",
+          bdb_id: `${clientConfig.bdbId}`,
+        },
+      });
+
+    await faultInjectorClient.waitForAction(bindActionId);
+
+    const pushNotificationLogs = diagnosticsLog.filter((log) => {
+      return ["MOVING", "MIGRATING", "MIGRATED"].includes(log?.type);
+    });
+
+    assert.strictEqual(pushNotificationLogs.length, 3);
+  });
+});