chore: (A-115) add reqresp effectiveness network test (#17905)

- test tx dropping at various percentages (ie. 10%, 30%, 50%) at high
TPS and log included/failed transactions and latency of tx inclusion
- add validator pod rolling with tx dropping flags via helm upgrade
function to utils

Note: Only logging for now as these tests won't be run outside of
testing networks

Author: PhilWindle

yarn-project/end-to-end/src/spartan/reqresp_effectiveness.test.ts

@@ -0,0 +1,164 @@
+import { SponsoredFeePaymentMethod } from '@aztec/aztec.js/fee';
+import type { AztecNode } from '@aztec/aztec.js/node';
+import { readFieldCompressedString } from '@aztec/aztec.js/utils';
+import { createLogger } from '@aztec/foundation/log';
+import { sleep } from '@aztec/foundation/sleep';
+import { ProvenTx, TestWallet, proveInteraction } from '@aztec/test-wallet/server';
+
+import { jest } from '@jest/globals';
+import type { ChildProcess } from 'child_process';
+
+import { getSponsoredFPCAddress } from '../fixtures/utils.js';
+import { createWalletAndAztecNodeClient, deploySponsoredTestAccounts } from './setup_test_wallets.js';
+import type { TestAccounts } from './setup_test_wallets.js';
+import { type TestConfig, setValidatorTxDrop, setupEnvironment, startPortForwardForRPC } from './utils.js';
+
+describe('reqresp effectiveness under tx drop', () => {
+  jest.setTimeout(60 * 60 * 1000);
+
+  const logger = createLogger(`e2e:spartan-test:reqresp-effectiveness`);
+
+  const config: TestConfig = { ...setupEnvironment(process.env) };
+  const TEST_DURATION_SECONDS = 10;
+  const TARGET_TPS = 10;
+  const TOTAL_TXS = TEST_DURATION_SECONDS * TARGET_TPS;
+  const MINT_AMOUNT = 10000n;
+
+  let wallet: TestWallet;
+  let aztecNode: AztecNode;
+  let cleanup: undefined | (() => Promise<void>);
+  let testAccounts: TestAccounts;
+  let recipient: any;
+  const forwardProcesses: ChildProcess[] = [];
+
+  afterAll(async () => {
+    // Reset validators to default (no tx drop)
+    try {
+      await setValidatorTxDrop({
+        namespace: config.NAMESPACE,
+        enabled: false,
+        probability: 0,
+        logger,
+      });
+    } catch (e) {
+      logger.warn(`Failed to reset validator tx drop flags: ${String(e)}`);
+    }
+    await cleanup?.();
+    forwardProcesses.forEach(p => p.kill());
+  });
+
+  beforeAll(async () => {
+    logger.info('Starting port forward for PXE');
+    const { process: aztecRpcProcess, port: aztecRpcPort } = await startPortForwardForRPC(config.NAMESPACE);
+    forwardProcesses.push(aztecRpcProcess);
+    const rpcUrl = `http://127.0.0.1:${aztecRpcPort}`;
+
+    const {
+      wallet: _wallet,
+      aztecNode: _aztecNode,
+      cleanup: _cleanup,
+    } = await createWalletAndAztecNodeClient(rpcUrl, config.REAL_VERIFIER, logger);
+    cleanup = _cleanup;
+    wallet = _wallet;
+    aztecNode = _aztecNode;
+    testAccounts = await deploySponsoredTestAccounts(wallet, aztecNode, MINT_AMOUNT, logger);
+    recipient = testAccounts.recipientAddress;
+    const name = readFieldCompressedString(
+      await testAccounts.tokenContract.methods.private_get_name().simulate({ from: testAccounts.tokenAdminAddress }),
+    );
+    expect(name).toBe(testAccounts.tokenName);
+  });
+
+  async function runLoadAndMeasure(probability: number) {
+    logger.info(`Applying tx drop: enabled=true, probability=${probability}`);
+
+    // Pre-prove load
+    const sponsor = new SponsoredFeePaymentMethod(await getSponsoredFPCAddress());
+    const transferAmount = 1n;
+    const txs: ProvenTx[] = await Promise.all(
+      Array.from({ length: TOTAL_TXS }, () =>
+        proveInteraction(
+          wallet,
+          testAccounts.tokenContract.methods.transfer_in_public(
+            testAccounts.tokenAdminAddress,
+            recipient,
+            transferAmount,
+            0,
+          ),
+          { from: testAccounts.tokenAdminAddress, fee: { paymentMethod: sponsor } },
+        ),
+      ),
+    );
+
+    if (!(probability == 0)) {
+      await setValidatorTxDrop({
+        namespace: config.NAMESPACE,
+        enabled: true,
+        probability,
+        logger,
+      });
+    }
+
+    const sends: Array<{ sentAt: number; promise: ReturnType<ProvenTx['send']> }[]> = [];
+    let sentSoFar = 0;
+    for (let sec = 0; sec < TEST_DURATION_SECONDS; sec++) {
+      const secondStart = Date.now();
+      const batch = txs.splice(0, TARGET_TPS);
+      const sentBatch = batch.map((tx, i) => {
+        const sent = tx.send();
+        logger.info(`p=${probability} sec ${sec + 1}: sent tx ${sentSoFar + i + 1}`);
+        return { sentAt: Date.now(), promise: sent };
+      });
+      sends.push(sentBatch);
+      sentSoFar += batch.length;
+      const elapsed = Date.now() - secondStart;
+      if (elapsed < 1000) {
+        await sleep(1000 - elapsed);
+      }
+    }
+
+    // Collect tx inclusion time
+    const latencies: number[] = [];
+    let included = 0;
+    let failed = 0;
+    await Promise.all(
+      sends.flat().map(async ({ sentAt, promise }, idx) => {
+        try {
+          await promise.wait({ timeout: 180, interval: 1, ignoreDroppedReceiptsFor: 2 });
+          const receipt = await promise.getReceipt();
+          if (receipt?.blockNumber !== undefined) {
+            included++;
+            const l = Date.now() - sentAt;
+            latencies.push(l);
+          } else {
+            failed++;
+          }
+        } catch (err) {
+          failed++;
+          logger.warn(`tx ${idx + 1} failed: ${String(err)}`);
+        }
+      }),
+    );
+
+    const pct = (p: number) => latencies[Math.floor((latencies.length - 1) * p)] ?? 0;
+    latencies.sort((a, b) => a - b);
+    const p50 = pct(0.5);
+    const p90 = pct(0.9);
+    const p99 = pct(0.99);
+
+    logger.info(
+      `Drop p=${probability}: included=${included}/${TOTAL_TXS}, failed=${failed}, latency(ms) p50=${p50}, p90=${p90}, p99=${p99}`,
+    );
+
+    expect(included + failed).toBe(TOTAL_TXS);
+    // Soft assertion: inclusion should remain reasonable even under drop
+    expect(included).toBeGreaterThan(0);
+  }
+
+  it('measures req/resp effectiveness across drop probabilities', async () => {
+    // Tx drop probabilities
+    for (const p of [0.1, 0.3, 0.5]) {
+      await runLoadAndMeasure(p);
+    }
+  });
+});

yarn-project/end-to-end/src/spartan/utils.ts

@@ -278,7 +278,7 @@ function createHelmCommand({
   namespace: string;
   valuesFile: string | undefined;
   timeout: string;
-  values: Record<string, string | number>;
+  values: Record<string, string | number | boolean>;
   reuseValues?: boolean;
 }) {
   const valuesFileArgs = valuesFile ? `--values ${helmChartDir}/values/${valuesFile}` : '';
@@ -633,7 +633,7 @@ export async function installTransferBot({
     namespace,
     valuesFile: undefined,
     timeout,
-    values: values as unknown as Record<string, string | number>,
+    values: values as unknown as Record<string, string | number | boolean>,
     reuseValues,
   });
 
@@ -661,6 +661,87 @@ export async function uninstallTransferBot(namespace: string, logger: Logger) {
   );
 }
 
+/**
+ * Enables or disables probabilistic transaction dropping on validators and waits for rollout.
+ * Wired to env vars P2P_DROP_TX and P2P_DROP_TX_CHANCE via Helm values.
+ */
+export async function setValidatorTxDrop({
+  namespace,
+  enabled,
+  probability,
+  logger,
+}: {
+  namespace: string;
+  enabled: boolean;
+  probability: number;
+  logger: Logger;
+}) {
+  const drop = enabled ? 'true' : 'false';
+  const prob = String(probability);
+
+  const selectors = ['app=validator', 'app.kubernetes.io/component=validator'];
+  let updated = false;
+  for (const selector of selectors) {
+    try {
+      const list = await execAsync(`kubectl get statefulset -l ${selector} -n ${namespace} --no-headers -o name | cat`);
+      const names = list.stdout
+        .split('\n')
+        .map(s => s.trim())
+        .filter(Boolean);
+      if (names.length === 0) {
+        continue;
+      }
+      const cmd = `kubectl set env statefulset -l ${selector} -n ${namespace} P2P_DROP_TX=${drop} P2P_DROP_TX_CHANCE=${prob}`;
+      logger.info(`command: ${cmd}`);
+      await execAsync(cmd);
+      updated = true;
+    } catch (e) {
+      logger.warn(`Failed to update validators with selector ${selector}: ${String(e)}`);
+    }
+  }
+
+  if (!updated) {
+    logger.warn(`No validator StatefulSets found in ${namespace}. Skipping tx drop toggle.`);
+    return;
+  }
+
+  // Restart validator pods to ensure env vars take effect and wait for readiness
+  await restartValidators(namespace, logger);
+}
+
+export async function restartValidators(namespace: string, logger: Logger) {
+  const selectors = ['app=validator', 'app.kubernetes.io/component=validator'];
+  let any = false;
+  for (const selector of selectors) {
+    try {
+      const { stdout } = await execAsync(`kubectl get pods -l ${selector} -n ${namespace} --no-headers -o name | cat`);
+      if (!stdout || stdout.trim().length === 0) {
+        continue;
+      }
+      any = true;
+      await deleteResourceByLabel({ resource: 'pods', namespace, label: selector });
+    } catch (e) {
+      logger.warn(`Error restarting validator pods with selector ${selector}: ${String(e)}`);
+    }
+  }
+
+  if (!any) {
+    logger.warn(`No validator pods found to restart in ${namespace}.`);
+    return;
+  }
+
+  // Wait for either label to be Ready
+  for (const selector of selectors) {
+    try {
+      await waitForResourceByLabel({ resource: 'pods', namespace, label: selector });
+      return;
+    } catch {
+      // try next
+    }
+  }
+  logger.warn(`Validator pods did not report Ready; continuing.`);
+}
+
 export async function enableValidatorDynamicBootNode(
   instanceName: string,
   namespace: string,