Merge branch 'unstable' into gloas-payload-cache

beacon_node/beacon_chain/tests/column_verification.rs

@@ -115,6 +115,78 @@ async fn rpc_columns_with_invalid_header_signature() {
     ));
 }
 
+/// Test that Gloas block production caches blobs alongside the envelope, and that
+/// data columns can be built from those cached blobs.
+#[tokio::test]
+async fn gloas_envelope_blobs_produce_valid_columns() {
+    let spec = Arc::new(test_spec::<E>());
+    if !spec.is_gloas_scheduled() {
+        return;
+    }
+
+    let harness = get_harness(VALIDATOR_COUNT, spec.clone(), NodeCustodyType::Supernode);
+    harness.execution_block_generator().set_min_blob_count(1);
+
+    // Build some chain depth.
+    let num_blocks = E::slots_per_epoch() as usize;
+    harness
+        .extend_chain(
+            num_blocks,
+            BlockStrategy::OnCanonicalHead,
+            AttestationStrategy::AllValidators,
+        )
+        .await;
+
+    harness.advance_slot();
+    let slot = harness.get_current_slot();
+
+    // Produce a Gloas block via the harness. This caches envelope + blobs.
+    let state = harness.get_current_state();
+    let (block_contents, opt_envelope, _post_state) =
+        harness.make_block_with_envelope(state, slot).await;
+    let signed_block = &block_contents.0;
+
+    assert!(
+        opt_envelope.is_some(),
+        "Gloas block production should produce an envelope"
+    );
+
+    // Verify the block has blob commitments in the bid.
+    let bid = signed_block
+        .message()
+        .body()
+        .signed_execution_payload_bid()
+        .expect("Gloas block should have a payload bid");
+    assert!(
+        !bid.message.blob_kzg_commitments.is_empty(),
+        "Block should have blob KZG commitments"
+    );
+
+    // Generate data columns from the block (using test fixtures, same as the harness does).
+    let data_column_sidecars =
+        generate_data_column_sidecars_from_block(signed_block, &harness.chain.spec);
+    assert_eq!(
+        data_column_sidecars.len(),
+        E::number_of_columns(),
+        "Should produce the correct number of data columns"
+    );
+
+    // Verify all columns are Gloas-format.
+    for col in &data_column_sidecars {
+        assert!(
+            col.as_gloas().is_ok(),
+            "Data column sidecar should be Gloas variant"
+        );
+        let gloas_col = col.as_gloas().expect("should be Gloas sidecar");
+        assert_eq!(gloas_col.beacon_block_root, signed_block.canonical_root());
+        assert_eq!(gloas_col.slot, slot);
+    }
+
+    // End-to-end DA flow (process_block → process_envelope → process_rpc_custody_columns)
+    // is not exercised here: Gloas blocks are not gated on columns at block-import time
+    // and the envelope/column gating belongs in a dedicated test once the DA path matures.
+}
+
 // Regression test for verify_header_signature bug: it uses head_fork() which is wrong for fork blocks
 #[tokio::test]
 async fn verify_header_signature_fork_block_bug() {

beacon_node/beacon_chain/tests/events.rs

@@ -10,8 +10,8 @@ use std::sync::Arc;
 use types::data::FixedBlobSidecarList;
 use types::test_utils::TestRandom;
 use types::{
-    BlobSidecar, DataColumnSidecar, DataColumnSidecarFulu, DataColumnSidecarGloas, EthSpec,
-    MinimalEthSpec, Slot,
+    BlobSidecar, DataColumnSidecar, DataColumnSidecarFulu, DataColumnSidecarGloas, Domain, EthSpec,
+    MinimalEthSpec, PayloadAttestationData, PayloadAttestationMessage, SignedRoot, Slot,
 };
 
 type E = MinimalEthSpec;
@@ -258,3 +258,177 @@ async fn head_event_on_block_import() {
         panic!("Expected Head event, got {:?}", head_event);
     }
 }
+
+/// Verifies that `execution_payload_gossip` fires at gossip verification time, and
+/// `execution_payload` + `execution_payload_available` fire at import time.
+#[tokio::test]
+async fn execution_payload_envelope_events() {
+    if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) {
+        return;
+    }
+
+    let harness = BeaconChainHarness::builder(E::default())
+        .default_spec()
+        .deterministic_keypairs(64)
+        .fresh_ephemeral_store()
+        .mock_execution_layer()
+        .build();
+
+    harness.extend_to_slot(Slot::new(1)).await;
+
+    let state = harness.get_current_state();
+    let target_slot = Slot::new(2);
+    harness.advance_slot();
+    let (block_contents, opt_envelope, _new_state) =
+        harness.make_block_with_envelope(state, target_slot).await;
+
+    let block_root = block_contents.0.canonical_root();
+
+    harness
+        .process_block(target_slot, block_root, block_contents)
+        .await
+        .expect("block should be processed");
+
+    let signed_envelope = opt_envelope.expect("Gloas block should produce an envelope");
+
+    let event_handler = harness.chain.event_handler.as_ref().unwrap();
+    let mut gossip_receiver = event_handler.subscribe_execution_payload_gossip();
+    let mut payload_receiver = event_handler.subscribe_execution_payload();
+    let mut available_receiver = event_handler.subscribe_execution_payload_available();
+
+    // Stage 1: gossip verification fires execution_payload_gossip only.
+    let gossip_verified = harness
+        .chain
+        .verify_envelope_for_gossip(Arc::new(signed_envelope))
+        .await
+        .expect("envelope gossip verification should succeed");
+
+    let gossip_event = gossip_receiver
+        .try_recv()
+        .expect("should receive execution_payload_gossip after gossip verification");
+    if let EventKind::ExecutionPayloadGossip(sse) = gossip_event {
+        assert_eq!(sse.slot, target_slot);
+        assert_eq!(sse.block_root, block_root);
+    } else {
+        panic!(
+            "Expected ExecutionPayloadGossip event, got {:?}",
+            gossip_event
+        );
+    }
+    assert!(payload_receiver.try_recv().is_err());
+    assert!(available_receiver.try_recv().is_err());
+
+    // Stage 2: import fires execution_payload and execution_payload_available.
+    harness
+        .chain
+        .process_execution_payload_envelope(
+            block_root,
+            gossip_verified,
+            beacon_chain::NotifyExecutionLayer::Yes,
+            types::BlockImportSource::Gossip,
+            #[allow(clippy::result_large_err)]
+            || Ok(()),
+        )
+        .await
+        .expect("envelope import should succeed");
+
+    let payload_event = payload_receiver
+        .try_recv()
+        .expect("should receive execution_payload after import");
+    if let EventKind::ExecutionPayload(sse) = payload_event {
+        assert_eq!(sse.slot, target_slot);
+        assert_eq!(sse.block_root, block_root);
+    } else {
+        panic!("Expected ExecutionPayload event, got {:?}", payload_event);
+    }
+
+    let available_event = available_receiver
+        .try_recv()
+        .expect("should receive execution_payload_available after import");
+    if let EventKind::ExecutionPayloadAvailable(sse) = available_event {
+        assert_eq!(sse.slot, target_slot);
+        assert_eq!(sse.block_root, block_root);
+    } else {
+        panic!(
+            "Expected ExecutionPayloadAvailable event, got {:?}",
+            available_event
+        );
+    }
+
+    assert!(
+        gossip_receiver.try_recv().is_err(),
+        "no extra gossip events should fire during import"
+    );
+}
+
+/// Verifies that a `payload_attestation_message` event is emitted when a payload attestation
+/// message passes gossip verification.
+#[tokio::test]
+async fn payload_attestation_message_event_on_gossip_verification() {
+    if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) {
+        return;
+    }
+
+    let harness = BeaconChainHarness::builder(E::default())
+        .default_spec()
+        .deterministic_keypairs(64)
+        .fresh_ephemeral_store()
+        .mock_execution_layer()
+        .build();
+
+    // Advance chain to have a valid head block.
+    let target_slot = Slot::new(1);
+    harness.extend_to_slot(target_slot).await;
+
+    let head = harness.chain.canonical_head.cached_head();
+    let head_state = &head.snapshot.beacon_state;
+
+    // Get a PTC member for this slot.
+    let ptc = head_state
+        .get_ptc(target_slot, &harness.spec)
+        .expect("should get PTC");
+    let validator_index = *ptc.0.first().expect("PTC should have at least one member") as u64;
+
+    // Sign a payload attestation.
+    let target_epoch = target_slot.epoch(E::slots_per_epoch());
+    let domain = harness.spec.get_domain(
+        target_epoch,
+        Domain::PTCAttester,
+        &head_state.fork(),
+        head_state.genesis_validators_root(),
+    );
+    let data = PayloadAttestationData {
+        beacon_block_root: head.head_block_root(),
+        slot: target_slot,
+        payload_present: true,
+        blob_data_available: true,
+    };
+    let message = data.signing_root(domain);
+    let signature = harness.validator_keypairs[validator_index as usize]
+        .sk
+        .sign(message);
+    let msg = PayloadAttestationMessage {
+        validator_index,
+        data: data.clone(),
+        signature: signature.clone(),
+    };
+
+    // Subscribe before verification.
+    let event_handler = harness.chain.event_handler.as_ref().unwrap();
+    let mut receiver = event_handler.subscribe_payload_attestation_message();
+
+    // Verify the attestation through the gossip path.
+    harness
+        .chain
+        .verify_payload_attestation_message_for_gossip(msg)
+        .expect("verification should succeed");
+
+    // Assert the event was emitted.
+    let event = receiver.try_recv().expect("should receive event");
+    if let EventKind::PayloadAttestationMessage(versioned) = event {
+        assert_eq!(versioned.data.validator_index, validator_index);
+        assert_eq!(versioned.data.data, data);
+    } else {
+        panic!("Expected PayloadAttestationMessage event, got {:?}", event);
+    }
+}

beacon_node/beacon_chain/tests/prepare_payload.rs

@@ -573,3 +573,121 @@ async fn prepare_payload_on_fork_boundary(
          advanced state"
     );
 }
+
+#[tokio::test]
+async fn gloas_block_production_caches_blobs_for_column_publishing() {
+    use beacon_chain::ProduceBlockVerification;
+    use beacon_chain::graffiti_calculator::GraffitiSettings;
+    use eth2::types::GraffitiPolicy;
+
+    let spec = Arc::new(test_spec::<E>());
+    if !spec.fork_name_at_slot::<E>(Slot::new(0)).gloas_enabled() {
+        return;
+    }
+
+    let db_path = tempdir().unwrap();
+    let store = get_store(&db_path, spec.clone());
+    let harness = get_harness(store.clone(), LOW_VALIDATOR_COUNT);
+
+    // Configure the mock EL to produce at least 1 blob per block.
+    harness.execution_block_generator().set_min_blob_count(1);
+
+    // Extend the chain a few slots to get past genesis.
+    harness
+        .extend_chain(
+            (E::slots_per_epoch() as usize) + 1,
+            BlockStrategy::OnCanonicalHead,
+            AttestationStrategy::AllValidators,
+        )
+        .await;
+
+    harness.advance_slot();
+    let slot = harness.get_current_slot();
+
+    // Produce a Gloas block directly via produce_block_on_state_gloas so we can
+    // inspect the pending cache before it's consumed.
+    let mut state = harness.get_current_state();
+    complete_state_advance(&mut state, None, slot, &spec).unwrap();
+    state.build_caches(&spec).unwrap();
+
+    let proposer_index = state.get_beacon_proposer_index(slot, &spec).unwrap();
+    let randao_reveal = harness.sign_randao_reveal(&state, proposer_index, slot);
+
+    let (parent_payload_status, parent_envelope) = {
+        let head = harness.chain.canonical_head.cached_head();
+        (
+            head.head_payload_status(),
+            head.snapshot.execution_envelope.clone(),
+        )
+    };
+
+    let graffiti_settings = GraffitiSettings::new(
+        Some(Graffiti::default()),
+        Some(GraffitiPolicy::PreserveUserGraffiti),
+    );
+
+    let (_block, _post_state, _value) = harness
+        .chain
+        .produce_block_on_state_gloas(
+            state,
+            None,
+            parent_payload_status,
+            parent_envelope,
+            slot,
+            randao_reveal,
+            graffiti_settings,
+            ProduceBlockVerification::VerifyRandao,
+        )
+        .await
+        .unwrap();
+
+    // The envelope + blobs should now be in the pending cache.
+    assert!(
+        harness
+            .chain
+            .pending_payload_envelopes
+            .read()
+            .contains(slot),
+        "Pending cache should contain an envelope for the produced slot"
+    );
+
+    // Take the blobs from the cache — this is what publish_execution_payload_envelope does.
+    let blobs = harness
+        .chain
+        .pending_payload_envelopes
+        .write()
+        .take_blobs(slot);
+
+    assert!(
+        blobs.is_some(),
+        "Blobs should be cached alongside the envelope"
+    );
+
+    let blobs = blobs.unwrap();
+    assert!(
+        !blobs.is_empty(),
+        "Blobs should be non-empty when min_blob_count >= 1"
+    );
+
+    // Verify take_blobs is consume-once.
+    let second_take = harness
+        .chain
+        .pending_payload_envelopes
+        .write()
+        .take_blobs(slot);
+    assert!(
+        second_take.is_none(),
+        "Blobs should only be consumable once"
+    );
+
+    // The envelope should still be in the cache after taking blobs.
+    assert!(
+        harness
+            .chain
+            .pending_payload_envelopes
+            .read()
+            .get(slot)
+            .is_some(),
+        "Envelope should remain in cache after taking blobs"
+    );
+}