chore(pipeline): spartan config

spartan/.gitignore

@@ -31,4 +31,5 @@ environments/*
 !environments/kind-minimal.env
 !environments/kind-provers.env
 !environments/alpha-net.env
+!environments/mbps-pipeline.env
 *.tfvars

spartan/environments/mbps-pipeline.env

@@ -0,0 +1,65 @@
+CREATE_ETH_DEVNET=true
+GCP_REGION=us-west1-a
+CLUSTER=aztec-gke-private
+NETWORK=next-net
+NAMESPACE=mbps-pipe
+DESTROY_NAMESPACE=true
+ETHEREUM_CHAIN_ID=1337
+FUNDING_PRIVATE_KEY="0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"
+LABS_INFRA_MNEMONIC="test test test test test test test test test test test junk"
+OTEL_COLLECTOR_ENDPOINT=REPLACE_WITH_GCP_SECRET
+
+DEPLOY_INTERNAL_BOOTNODE=true
+TEST_ACCOUNTS=true
+SPONSORED_FPC=true
+SEQ_MIN_TX_PER_BLOCK=0
+SEQ_MAX_TX_PER_BLOCK=8
+AZTEC_EPOCH_DURATION=8
+REAL_VERIFIER=false
+PROVER_REAL_PROOFS=false
+
+CREATE_ROLLUP_CONTRACTS=true
+VERIFY_CONTRACTS=false
+DESTROY_AZTEC_INFRA=true
+
+SEQ_BUILD_CHECKPOINT_IF_EMPTY=true
+SEQ_BLOCK_DURATION_MS=6000
+SEQ_ENABLE_PROPOSER_PIPELINING=true
+SEQ_PER_BLOCK_ALLOCATION_MULTIPLIER=1.1
+LOG_LEVEL=verbose
+
+AZTEC_LAG_IN_EPOCHS_FOR_VALIDATOR_SET=2
+AZTEC_LAG_IN_EPOCHS_FOR_RANDAO=2
+
+VALIDATOR_REPLICAS=4
+VALIDATORS_PER_NODE=12
+VALIDATOR_PUBLISHERS_PER_REPLICA=4
+VALIDATOR_PUBLISHER_MNEMONIC_START_INDEX=5000
+
+PUBLISHERS_PER_PROVER=2
+PROVER_PUBLISHER_MNEMONIC_START_INDEX=8000
+
+BOT_TRANSFERS_REPLICAS=1
+BOT_TRANSFERS_TX_INTERVAL_SECONDS=4
+BOT_TRANSFERS_FOLLOW_CHAIN=PROPOSED
+BOT_TRANSFERS_PXE_SYNC_CHAIN_TIP=proposed
+
+BOT_SWAPS_REPLICAS=1
+BOT_SWAPS_TX_INTERVAL_SECONDS=4
+BOT_SWAPS_FOLLOW_CHAIN=PROPOSED
+BOT_SWAPS_PXE_SYNC_CHAIN_TIP=proposed
+
+BOT_CROSS_CHAIN_REPLICAS=1
+BOT_CROSS_CHAIN_TX_INTERVAL_SECONDS=8
+BOT_CROSS_CHAIN_FOLLOW_CHAIN=PROPOSED
+BOT_CROSS_CHAIN_PXE_SYNC_CHAIN_TIP=proposed
+
+REDEPLOY_ROLLUP_CONTRACTS=true
+
+DEBUG_P2P_INSTRUMENT_MESSAGES=true
+OTEL_COLLECT_INTERVAL_MS=10000
+OTEL_EXPORT_TIMEOUT_MS=5000
+
+VALIDATOR_HA_REPLICAS=1
+VALIDATOR_RESOURCE_PROFILE="prod-spot"
+

spartan/scripts/deploy_network.sh

@@ -111,6 +111,7 @@ SEQ_PER_BLOCK_ALLOCATION_MULTIPLIER=${SEQ_PER_BLOCK_ALLOCATION_MULTIPLIER:-null}
 SEQ_BLOCK_DURATION_MS=${SEQ_BLOCK_DURATION_MS:-}
 SEQ_L1_PUBLISHING_TIME_ALLOWANCE_IN_SLOT=${SEQ_L1_PUBLISHING_TIME_ALLOWANCE_IN_SLOT:-}
 SEQ_BUILD_CHECKPOINT_IF_EMPTY=${SEQ_BUILD_CHECKPOINT_IF_EMPTY:-}
+SEQ_ENABLE_PROPOSER_PIPELINING=${SEQ_ENABLE_PROPOSER_PIPELINING:-false}
 SEQ_ENFORCE_TIME_TABLE=${SEQ_ENFORCE_TIME_TABLE:-}
 SEQ_SKIP_CHECKPOINT_PUBLISH_PERCENT=${SEQ_SKIP_CHECKPOINT_PUBLISH_PERCENT:-0}
 PROVER_REPLICAS=${PROVER_REPLICAS:-4}
@@ -119,6 +120,8 @@ R2_ACCESS_KEY_ID=${R2_ACCESS_KEY_ID:-}
 R2_SECRET_ACCESS_KEY=${R2_SECRET_ACCESS_KEY:-}
 
 OTEL_COLLECTOR_ENDPOINT=${OTEL_COLLECTOR_ENDPOINT:-}
+OTEL_COLLECT_INTERVAL_MS=${OTEL_COLLECT_INTERVAL_MS:-}
+OTEL_EXPORT_TIMEOUT_MS=${OTEL_EXPORT_TIMEOUT_MS:-}
 DEPLOY_INTERNAL_BOOTNODE=${DEPLOY_INTERNAL_BOOTNODE:-}
 DEPLOY_ARCHIVAL_NODE=${DEPLOY_ARCHIVAL_NODE:-false}
 
@@ -537,6 +540,7 @@ SEQ_PER_BLOCK_ALLOCATION_MULTIPLIER = ${SEQ_PER_BLOCK_ALLOCATION_MULTIPLIER}
 SEQ_BLOCK_DURATION_MS = ${SEQ_BLOCK_DURATION_MS:-null}
 SEQ_L1_PUBLISHING_TIME_ALLOWANCE_IN_SLOT = ${SEQ_L1_PUBLISHING_TIME_ALLOWANCE_IN_SLOT:-null}
 SEQ_BUILD_CHECKPOINT_IF_EMPTY = ${SEQ_BUILD_CHECKPOINT_IF_EMPTY:-null}
+SEQ_ENABLE_PROPOSER_PIPELINING = ${SEQ_ENABLE_PROPOSER_PIPELINING}
 SEQ_ENFORCE_TIME_TABLE = ${SEQ_ENFORCE_TIME_TABLE:-null}
 SEQ_SKIP_CHECKPOINT_PUBLISH_PERCENT = ${SEQ_SKIP_CHECKPOINT_PUBLISH_PERCENT}
 PROVER_MNEMONIC = "${LABS_INFRA_MNEMONIC}"
@@ -558,6 +562,8 @@ SLASH_INVALID_BLOCK_PENALTY = ${SLASH_INVALID_BLOCK_PENALTY:-null}
 SLASH_OFFENSE_EXPIRATION_ROUNDS = ${SLASH_OFFENSE_EXPIRATION_ROUNDS:-null}
 SLASH_MAX_PAYLOAD_SIZE = ${SLASH_MAX_PAYLOAD_SIZE:-null}
 OTEL_COLLECTOR_ENDPOINT = "${OTEL_COLLECTOR_ENDPOINT}"
+OTEL_COLLECT_INTERVAL_MS = ${OTEL_COLLECT_INTERVAL_MS:-null}
+OTEL_EXPORT_TIMEOUT_MS = ${OTEL_EXPORT_TIMEOUT_MS:-null}
 DEPLOY_INTERNAL_BOOTNODE = ${DEPLOY_INTERNAL_BOOTNODE:-true}
 PROVER_REAL_PROOFS = ${PROVER_REAL_PROOFS}
 TRANSACTIONS_DISABLED = ${TRANSACTIONS_DISABLED:-null}

spartan/terraform/deploy-aztec-infra/main.tf

@@ -217,6 +217,7 @@ locals {
     "validator.node.env.SEQ_BLOCK_DURATION_MS"                 = var.SEQ_BLOCK_DURATION_MS
     "validator.node.env.SEQ_L1_PUBLISHING_TIME_ALLOWANCE_IN_SLOT" = var.SEQ_L1_PUBLISHING_TIME_ALLOWANCE_IN_SLOT
     "validator.node.env.SEQ_BUILD_CHECKPOINT_IF_EMPTY"         = var.SEQ_BUILD_CHECKPOINT_IF_EMPTY
+    "validator.node.env.SEQ_ENABLE_PROPOSER_PIPELINING"       = var.SEQ_ENABLE_PROPOSER_PIPELINING
     "validator.node.env.SEQ_ENFORCE_TIME_TABLE"                = var.SEQ_ENFORCE_TIME_TABLE
     "validator.node.env.P2P_TX_POOL_DELETE_TXS_AFTER_REORG"    = var.P2P_TX_POOL_DELETE_TXS_AFTER_REORG
     "validator.node.env.L1_PRIORITY_FEE_BUMP_PERCENTAGE"       = var.VALIDATOR_L1_PRIORITY_FEE_BUMP_PERCENTAGE
@@ -225,6 +226,8 @@ locals {
     "validator.node.env.P2P_MAX_TX_POOL_SIZE"                  = var.P2P_MAX_TX_POOL_SIZE
     "validator.node.env.PROVER_TEST_VERIFICATION_DELAY_MS"     = var.PROVER_TEST_VERIFICATION_DELAY_MS
     "validator.node.env.DEBUG_P2P_INSTRUMENT_MESSAGES"         = var.DEBUG_P2P_INSTRUMENT_MESSAGES
+    "validator.node.env.OTEL_COLLECT_INTERVAL_MS"              = var.OTEL_COLLECT_INTERVAL_MS
+    "validator.node.env.OTEL_EXPORT_TIMEOUT_MS"                = var.OTEL_EXPORT_TIMEOUT_MS
     "validator.node.secret.envEnabled"                         = true
     "validator.node.secret.mnemonic"                           = var.VALIDATOR_MNEMONIC
     "validator.node.secret.mnemonicIndex"                      = var.VALIDATOR_MNEMONIC_START_INDEX

spartan/terraform/deploy-aztec-infra/variables.tf

@@ -315,6 +315,20 @@ variable "OTEL_COLLECTOR_ENDPOINT" {
   nullable    = true
 }
 
+variable "OTEL_COLLECT_INTERVAL_MS" {
+  description = "Interval in ms at which OTEL metrics are exported from nodes"
+  type        = string
+  nullable    = true
+  default     = null
+}
+
+variable "OTEL_EXPORT_TIMEOUT_MS" {
+  description = "Timeout in ms for OTEL metric exports (must be <= OTEL_COLLECT_INTERVAL_MS)"
+  type        = string
+  nullable    = true
+  default     = null
+}
+
 variable "LOG_LEVEL" {
   description = "Log level for all nodes"
   type        = string
@@ -395,6 +409,12 @@ variable "SEQ_PER_BLOCK_ALLOCATION_MULTIPLIER" {
   default     = null
 }
 
+variable "SEQ_ENABLE_PROPOSER_PIPELINING" {
+  description = "Whether to enable build-ahead proposer pipelining"
+  type        = string
+  default     = "false"
+}
+
 variable "SENTINEL_ENABLED" {
   description = "Whether to enable sentinel"
   type        = string

yarn-project/archiver/src/archiver.ts

@@ -85,6 +85,8 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
 
   public readonly tracer: Tracer;
 
+  private readonly instrumentation: ArchiverInstrumentation;
+
   /**
    * Creates a new instance of the Archiver.
    * @param publicClient - A client for interacting with the Ethereum node.
@@ -130,6 +132,7 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
     super(dataStore, l1Constants);
 
     this.tracer = instrumentation.tracer;
+    this.instrumentation = instrumentation;
     this.initialSyncPromise = promiseWithResolvers();
     this.synchronizer = synchronizer;
     this.events = events;
@@ -250,6 +253,7 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
 
       try {
         await this.updater.addProposedBlock(block);
+        this.instrumentation.processNewProposedBlock(block);
         this.log.debug(`Added block ${block.number} to store`);
         resolve();
       } catch (err: any) {

yarn-project/archiver/src/modules/instrumentation.ts

@@ -113,17 +113,31 @@ export class ArchiverInstrumentation {
     return this.telemetry.isEnabled();
   }
 
+  public processNewProposedBlock(block: L2Block) {
+    this.blockHeight.record(block.number, { [Attributes.STATUS]: 'proposed' });
+    this.processNewBlock(block, 'proposed');
+  }
+
   public processNewBlocks(syncTimePerBlock: number, blocks: L2Block[]) {
     this.syncDurationPerBlock.record(Math.ceil(syncTimePerBlock));
     this.blockHeight.record(Math.max(...blocks.map(b => b.number)));
     this.checkpointHeight.record(Math.max(...blocks.map(b => b.checkpointNumber)));
     this.syncBlockCount.add(blocks.length);
 
     for (const block of blocks) {
-      this.txCount.add(block.body.txEffects.length);
-      this.txsPerBlock.record(block.body.txEffects.length);
-      this.manaPerBlock.record(block.header.totalManaUsed.toNumber() / 1e6);
+      this.processNewBlock(block);
+    }
+  }
+
+  /** Records per-block metrics tagged with a status (e.g. 'proposed'). */
+  private processNewBlock(block: L2Block, status?: string) {
+    let attrs = undefined;
+    if (status) {
+      attrs = { [Attributes.STATUS]: status };
     }
+    this.txCount.add(block.body.txEffects.length, attrs);
+    this.txsPerBlock.record(block.body.txEffects.length, attrs);
+    this.manaPerBlock.record(block.header.totalManaUsed.toNumber() / 1e6, attrs);
   }
 
   public processNewMessages(count: number, syncPerMessageMs: number) {

yarn-project/foundation/src/sleep/index.ts

@@ -1,4 +1,4 @@
-import { InterruptError } from '../error/index.js';
+import { AbortError, InterruptError } from '../error/index.js';
 
 /**
  * InterruptibleSleep is a utility class that allows you to create an interruptible sleep function.
@@ -28,11 +28,14 @@ export class InterruptibleSleep {
    * Sleep for a specified amount of time in milliseconds.
    * The sleep function will pause the execution of the current async function
    * for the given time period, allowing other tasks to run before resuming.
+   * If an AbortSignal is provided, the sleep can be cut short when the signal fires.
    *
    * @param ms - The number of milliseconds to sleep.
+   * @param signal - Optional AbortSignal to interrupt the sleep early.
+   * @param opts - Options controlling behaviour on abort. If `throwOnAbort` is true, the sleep throws `signal.reason`; otherwise it resolves silently.
    * @returns A Promise that resolves after the specified time has passed.
    */
-  public async sleep(ms: number): Promise<void> {
+  public async sleep(ms: number, signal?: AbortSignal, opts?: { throwOnAbort?: boolean }): Promise<void> {
     let interruptResolve: (shouldThrow: boolean) => void;
     const interruptPromise = new Promise<boolean>(resolve => {
       interruptResolve = resolve;
@@ -44,14 +47,29 @@ export class InterruptibleSleep {
       timeoutId = setTimeout(() => resolve(false), ms);
       this.timeoutIds.push(timeoutId);
     });
+
+    // Listen for AbortSignal if provided
+    let onAbort: (() => void) | undefined;
+    if (signal) {
+      if (signal.aborted) {
+        interruptResolve!(opts?.throwOnAbort ?? false);
+      } else {
+        onAbort = () => interruptResolve!(opts?.throwOnAbort ?? false);
+        signal.addEventListener('abort', onAbort, { once: true });
+      }
+    }
+
     const shouldThrow = await Promise.race([interruptPromise, timeoutPromise]);
 
     clearTimeout(timeoutId!);
     this.timeoutIds = this.timeoutIds.filter(id => id !== timeoutId);
     this.interrupts = this.interrupts.filter(res => res !== interruptResolve);
+    if (onAbort && signal) {
+      signal.removeEventListener('abort', onAbort);
+    }
 
     if (shouldThrow) {
-      throw new InterruptError('Interrupted.');
+      throw signal?.reason ?? new InterruptError('Interrupted.');
     }
   }
 
@@ -88,3 +106,28 @@ export function sleepUntil<T>(target: Date, now: Date, returnValue?: T): Promise
   const ms = target.getTime() - now.getTime();
   return sleep(ms, returnValue);
 }
+
+/**
+ * Sleeps for the given duration. If an AbortSignal is provided, the sleep
+ * rejects with `signal.reason` (or AbortError) when the signal fires.
+ * Without a signal, behaves identically to `sleep()`.
+ */
+export function abortableSleep(ms: number, signal?: AbortSignal): Promise<void> {
+  if (!signal) {
+    return sleep(ms);
+  }
+  if (signal.aborted) {
+    return Promise.reject(signal.reason ?? new AbortError('Aborted'));
+  }
+  return new Promise<void>((resolve, reject) => {
+    const onAbort = () => {
+      clearTimeout(timer);
+      reject(signal.reason ?? new AbortError('Aborted'));
+    };
+    const timer = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort);
+      resolve();
+    }, ms);
+    signal.addEventListener('abort', onAbort, { once: true });
+  });
+}

yarn-project/foundation/src/sleep/sleep.test.ts

@@ -1,7 +1,7 @@
 import { jest } from '@jest/globals';
 
 import { InterruptError } from '../error/index.js';
-import { InterruptibleSleep } from './index.js';
+import { InterruptibleSleep, abortableSleep } from './index.js';
 
 describe('InterruptibleSleep', () => {
   it('should sleep for 100ms', async () => {
@@ -37,4 +37,45 @@ describe('InterruptibleSleep', () => {
     expect(end1! - start).toBeGreaterThanOrEqual(90);
     expect(stub).not.toHaveBeenCalled();
   });
+
+  it('should resolve when signal is aborted (non-throwing)', async () => {
+    const sleeper = new InterruptibleSleep();
+    const controller = new AbortController();
+    const promise = sleeper.sleep(5000, controller.signal);
+    setTimeout(() => controller.abort(new Error('aborted')), 50);
+    await expect(promise).resolves.toBeUndefined();
+  });
+
+  it('should throw signal.reason when signal is aborted with throwOnAbort', async () => {
+    const sleeper = new InterruptibleSleep();
+    const controller = new AbortController();
+    const promise = sleeper.sleep(5000, controller.signal, { throwOnAbort: true });
+    setTimeout(() => controller.abort(new Error('test reason')), 50);
+    await expect(promise).rejects.toThrow('test reason');
+  });
+});
+
+describe('abortableSleep', () => {
+  it('resolves after the given time', async () => {
+    const start = Date.now();
+    await abortableSleep(50);
+    expect(Date.now() - start).toBeGreaterThanOrEqual(45);
+  });
+
+  it('rejects immediately if signal is already aborted', async () => {
+    const controller = new AbortController();
+    controller.abort(new Error('already aborted'));
+    await expect(abortableSleep(5000, controller.signal)).rejects.toThrow('already aborted');
+  });
+
+  it('rejects when signal is aborted during sleep', async () => {
+    const controller = new AbortController();
+    const promise = abortableSleep(5000, controller.signal);
+    setTimeout(() => controller.abort(new Error('interrupted')), 50);
+    await expect(promise).rejects.toThrow('interrupted');
+  });
+
+  it('resolves normally without a signal', async () => {
+    await expect(abortableSleep(10)).resolves.toBeUndefined();
+  });
 });

yarn-project/sequencer-client/src/sequencer/metrics.ts

@@ -68,6 +68,9 @@ export class SequencerMetrics {
   private fishermanMinedBlobTxPriorityFee: Histogram;
   private fishermanMinedBlobTxTotalCost: Histogram;
 
+  private blockInterBlockTime: Histogram;
+  private lastBlockBuiltTimestamp?: number;
+
   private lastSeenSlot?: SlotNumber;
 
   constructor(
@@ -86,6 +89,8 @@ export class SequencerMetrics {
 
     this.blockBuildManaPerSecond = this.meter.createGauge(Metrics.SEQUENCER_BLOCK_BUILD_MANA_PER_SECOND);
 
+    this.blockInterBlockTime = this.meter.createHistogram(Metrics.SEQUENCER_BLOCK_INTER_BLOCK_TIME);
+
     this.stateTransitionBufferDuration = this.meter.createHistogram(Metrics.SEQUENCER_STATE_TRANSITION_BUFFER_DURATION);
 
     this.checkpointAttestationDelay = this.meter.createHistogram(Metrics.SEQUENCER_CHECKPOINT_ATTESTATION_DELAY);
@@ -226,6 +231,12 @@ export class SequencerMetrics {
     });
     this.blockBuildDuration.record(Math.ceil(buildDurationMs));
     this.blockBuildManaPerSecond.record(Math.ceil((totalMana * 1000) / buildDurationMs));
+
+    const now = Date.now();
+    if (this.lastBlockBuiltTimestamp !== undefined) {
+      this.blockInterBlockTime.record(now - this.lastBlockBuiltTimestamp);
+    }
+    this.lastBlockBuiltTimestamp = now;
   }
 
   recordFailedBlock() {

yarn-project/telemetry-client/src/metrics.ts

@@ -405,6 +405,12 @@ export const SEQUENCER_BLOCK_COUNT: MetricDefinition = {
   description: 'Number of blocks built by this sequencer',
   valueType: ValueType.INT,
 };
+export const SEQUENCER_BLOCK_INTER_BLOCK_TIME: MetricDefinition = {
+  name: 'aztec.sequencer.block.inter_block_time',
+  description: 'Wall-clock time elapsed between consecutive blocks being built by this sequencer',
+  unit: 'ms',
+  valueType: ValueType.INT,
+};
 export const SEQUENCER_CURRENT_SLOT_REWARDS: MetricDefinition = {
   name: 'aztec.sequencer.current_slot_rewards',
   description: 'The rewards earned per filled slot',