cl: preserve expanded Gloas lifecycle hardening

cl/das/peer_das.go

@@ -48,6 +48,7 @@ type gloasBlockData struct {
 
 //go:generate mockgen -typed=true -destination=mock_services/peer_das_mock.go -package=mock_services . PeerDas
 type PeerDas interface {
+	Start(ctx context.Context)
 	// [Modified in Gloas:EIP7732] Changed from []*SignedBlindedBeaconBlock to []ColumnSyncableSignedBlock
 	// to support both pre-GLOAS (blinded) and GLOAS (non-blinded) blocks
 	DownloadColumnsAndRecoverBlobs(ctx context.Context, blocks []cltypes.ColumnSyncableSignedBlock) error
@@ -88,10 +89,10 @@ type peerdas struct {
 	blockReader    freezeblocks.BeaconSnapshotReader
 	indiciesDB     kv.RoDB
 	gloasDataCache *lru.Cache[common.Hash, *gloasBlockData] // cache for GLOAS block data (~1KB per entry)
+	startOnce      sync.Once
 }
 
 func NewPeerDas(
-	ctx context.Context,
 	rpc *rpc.BeaconRpcP2P,
 	beaconConfig *clparams.BeaconChainConfig,
 	caplinConfig *clparams.CaplinConfig,
@@ -128,14 +129,19 @@ func NewPeerDas(
 		indiciesDB:     indiciesDB,
 		gloasDataCache: gloasDataCache,
 	}
-	p.resubscribeGossip()
-	for range numOfBlobRecoveryWorkers {
-		go p.blobsRecoverWorker(ctx)
-	}
-	go p.syncColumnDataWorker(ctx)
 	return p
 }
 
+func (d *peerdas) Start(ctx context.Context) {
+	d.startOnce.Do(func() {
+		d.resubscribeGossip()
+		for range numOfBlobRecoveryWorkers {
+			go d.blobsRecoverWorker(ctx)
+		}
+		go d.syncColumnDataWorker(ctx)
+	})
+}
+
 func (d *peerdas) StateReader() peerdasstate.PeerDasStateReader {
 	return d.state
 }

cl/das/peer_das_start_test.go

@@ -0,0 +1,45 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package das
+
+import (
+	"context"
+	"testing"
+
+	"go.uber.org/mock/gomock"
+
+	"github.com/erigontech/erigon/cl/clparams"
+	peerdasstate "github.com/erigontech/erigon/cl/das/state"
+	gossipmock "github.com/erigontech/erigon/cl/phase1/network/gossip/mock_services"
+)
+
+func TestPeerDasSubscribesOnlyAfterStart(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	gossipManager := gossipmock.NewMockGossip(ctrl)
+	beaconConfig := clparams.MainnetBeaconConfig
+	beaconConfig.DataColumnSidecarSubnetCount = 2
+	caplinConfig := clparams.CaplinConfig{ArchiveBlobs: true}
+	peerDasState := peerdasstate.NewPeerDasState(&beaconConfig, &clparams.NetworkConfig{})
+
+	peerDas := NewPeerDas(nil, &beaconConfig, &caplinConfig, nil, nil, nil, [32]byte{}, nil, peerDasState, gossipManager, nil, nil)
+
+	gossipManager.EXPECT().SubscribeWithExpiry(gomock.Any(), gomock.Any()).Times(2)
+	ctx, cancel := context.WithCancel(context.Background())
+	peerDas.Start(ctx)
+	peerDas.Start(ctx)
+	cancel()
+}

cl/phase1/execution_client/block_collector/encode_block_bench_test.go

@@ -52,9 +52,11 @@ func BenchmarkEncodeBlock(b *testing.B) {
 	} {
 		payload := benchPayload(tc.txCount, tc.txSize)
 		b.Run(tc.name, func(b *testing.B) {
-			p := &PersistentBlockCollector{}
-			p.mu.Lock()
-			defer p.mu.Unlock()
+			p := &PersistentBlockCollector{operationSlot: make(chan struct{}, 1)}
+			if err := p.acquire(b.Context()); err != nil {
+				b.Fatal(err)
+			}
+			defer p.release()
 			b.ReportAllocs()
 			for b.Loop() {
 				if _, err := p.encodeBlock(payload, parentRoot, nil); err != nil {

cl/phase1/execution_client/block_collector/encode_block_test.go

@@ -43,9 +43,12 @@ func signedTestTx(t *testing.T, nonce uint64) types.Transaction {
 // consecutive encodes on one collector reuse its scratch buffers, and each
 // result must decode back to the original execution block.
 func TestEncodeDecodeBlockRoundTrip(t *testing.T) {
-	c := &PersistentBlockCollector{beaconChainCfg: &clparams.MainnetBeaconConfig}
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	c := &PersistentBlockCollector{
+		beaconChainCfg: &clparams.MainnetBeaconConfig,
+		operationSlot:  make(chan struct{}, 1),
+	}
+	require.NoError(t, c.acquire(t.Context()))
+	defer c.release()
 
 	parent := common.HexToHash("0xaa")
 	tx0, tx1, tx2 := signedTestTx(t, 0), signedTestTx(t, 1), signedTestTx(t, 2)

cl/phase1/execution_client/block_collector/interface.go

@@ -28,7 +28,7 @@ var batchSize = 1000
 type BlockCollector interface {
 	AddBlock(block *cltypes.BeaconBlock) error
 	// AddGloasBlock adds a GLOAS (EIP-7732) FULL block using its execution payload envelope.
-	AddGloasBlock(block *cltypes.BeaconBlock, envelope *cltypes.SignedExecutionPayloadEnvelope) error
+	AddGloasBlock(ctx context.Context, block *cltypes.BeaconBlock, envelope *cltypes.SignedExecutionPayloadEnvelope) error
 	Flush(ctx context.Context) error
 	HasBlock(blockNumber uint64) bool
 }

cl/phase1/execution_client/block_collector/persistent_block_collector.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"encoding/binary"
 	"fmt"
-	"sync"
 
 	"github.com/c2h5oh/datasize"
 	"github.com/golang/snappy"
@@ -54,8 +53,8 @@ type PersistentBlockCollector struct {
 	logger         log.Logger
 	engine         execution_client.ExecutionEngine
 
-	mu sync.Mutex
-	// encodeBlock scratch buffers; guarded by mu.
+	operationSlot chan struct{}
+	// encodeBlock scratch buffers; guarded by operationSlot.
 	encodeBlockBuf   []byte
 	blockCompressBuf []byte
 }
@@ -101,13 +100,29 @@ func NewPersistentBlockCollector(
 		beaconChainCfg: beaconChainCfg,
 		logger:         logger,
 		engine:         engine,
+		operationSlot:  make(chan struct{}, 1),
 	}
 }
 
+func (p *PersistentBlockCollector) acquire(ctx context.Context) error {
+	select {
+	case p.operationSlot <- struct{}{}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (p *PersistentBlockCollector) release() {
+	<-p.operationSlot
+}
+
 // AddBlock adds a block to the collector, persisting it to the database
 func (p *PersistentBlockCollector) AddBlock(block *cltypes.BeaconBlock) error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+	if err := p.acquire(context.Background()); err != nil {
+		return err
+	}
+	defer p.release()
 
 	if p.db == nil {
 		return fmt.Errorf("database not initialized")
@@ -126,9 +141,11 @@ func (p *PersistentBlockCollector) AddBlock(block *cltypes.BeaconBlock) error {
 
 // AddGloasBlock adds a GLOAS (EIP-7732) FULL block with its execution payload envelope to the collector.
 // The execution payload is extracted from the envelope, not the beacon block body.
-func (p *PersistentBlockCollector) AddGloasBlock(block *cltypes.BeaconBlock, envelope *cltypes.SignedExecutionPayloadEnvelope) error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+func (p *PersistentBlockCollector) AddGloasBlock(ctx context.Context, block *cltypes.BeaconBlock, envelope *cltypes.SignedExecutionPayloadEnvelope) error {
+	if err := p.acquire(ctx); err != nil {
+		return err
+	}
+	defer p.release()
 
 	if p.db == nil {
 		return fmt.Errorf("database not initialized")
@@ -141,7 +158,7 @@ func (p *PersistentBlockCollector) AddGloasBlock(block *cltypes.BeaconBlock, env
 		return fmt.Errorf("failed to encode gloas block: %w", err)
 	}
 
-	return p.db.Update(context.Background(), func(tx kv.RwTx) error {
+	return p.db.Update(ctx, func(tx kv.RwTx) error {
 		return tx.Put(kv.Headers, payloadKey(payload), encodedBlock)
 	})
 }
@@ -156,7 +173,7 @@ const (
 // encodeBlock serializes the block value: snappy(version + parentRoot +
 // [requestsHash +] SSZ(payload)). The result aliases p.blockCompressBuf and is
 // valid only until the next call, so callers must copy it or fully consume it
-// before encoding again. Callers must hold p.mu.
+// before encoding again. Callers must own operationSlot.
 func (p *PersistentBlockCollector) encodeBlock(payload *cltypes.Eth1Block, parentRoot common.Hash, executionRequestsList []hexutil.Bytes) ([]byte, error) {
 	p.encodeBlockBuf = append(p.encodeBlockBuf[:0], byte(payload.Version()))
 	p.encodeBlockBuf = append(p.encodeBlockBuf, parentRoot[:]...)
@@ -196,8 +213,10 @@ func (p *PersistentBlockCollector) releaseOversizedScratch() {
 // If a real gap is detected, rows past the gap are kept so the next Flush can retry
 // once the missing range is re-downloaded.
 func (p *PersistentBlockCollector) Flush(ctx context.Context) error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+	if err := p.acquire(ctx); err != nil {
+		return err
+	}
+	defer p.release()
 	defer p.releaseOversizedScratch()
 
 	if p.db == nil {
@@ -517,8 +536,10 @@ func (p *PersistentBlockCollector) doForkChoiceUpdate(ctx context.Context, lastB
 
 // HasBlock checks if a block with the given number is already in the collector
 func (p *PersistentBlockCollector) HasBlock(blockNumber uint64) bool {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+	if err := p.acquire(context.Background()); err != nil {
+		return false
+	}
+	defer p.release()
 
 	if p.db == nil {
 		return false
@@ -550,8 +571,10 @@ func (p *PersistentBlockCollector) HasBlock(blockNumber uint64) bool {
 
 // Close closes the database
 func (p *PersistentBlockCollector) Close() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+	if err := p.acquire(context.Background()); err != nil {
+		return err
+	}
+	defer p.release()
 
 	if p.db != nil {
 		p.db.Close()

cl/phase1/execution_client/block_collector/persistent_block_collector_test.go

@@ -145,6 +145,48 @@ func countRowsAtOrAbove(t *testing.T, db kv.RoDB, minNumber uint64) int {
 	return count
 }
 
+func TestAddGloasBlockHonorsCanceledContext(t *testing.T) {
+	h := newFlushTestHarness(t, 0)
+	block := makeBeaconBlock(t, 1, 1, common.Hash{})
+	envelope := &cltypes.SignedExecutionPayloadEnvelope{Message: cltypes.NewExecutionPayloadEnvelope(&clparams.MainnetBeaconConfig)}
+	envelope.Message.Payload = block.Body.ExecutionPayload
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	require.ErrorIs(t, h.collector.AddGloasBlock(ctx, block, envelope), context.Canceled)
+	require.Equal(t, 0, countRowsAtOrAbove(t, h.collector.db, 0))
+}
+
+func TestAddGloasBlockCancellationWhileFlushOwnsCollector(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	engine := execution_client.NewMockExecutionEngine(ctrl)
+	flushStarted := make(chan struct{})
+	releaseFlush := make(chan struct{})
+	engine.EXPECT().FrozenBlocks(gomock.Any()).DoAndReturn(func(context.Context) uint64 {
+		close(flushStarted)
+		<-releaseFlush
+		return 0
+	})
+	collector := NewPersistentBlockCollector(log.New(), engine, &clparams.MainnetBeaconConfig, filepath.Join(t.TempDir(), "collector"))
+	require.NotNil(t, collector)
+	t.Cleanup(func() { _ = collector.Close() })
+
+	flushDone := make(chan error, 1)
+	go func() { flushDone <- collector.Flush(t.Context()) }()
+	<-flushStarted
+
+	block := makeBeaconBlock(t, 1, 1, common.Hash{})
+	envelope := &cltypes.SignedExecutionPayloadEnvelope{Message: cltypes.NewExecutionPayloadEnvelope(&clparams.MainnetBeaconConfig)}
+	envelope.Message.Payload = block.Body.ExecutionPayload
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	require.ErrorIs(t, collector.AddGloasBlock(ctx, block, envelope), context.Canceled)
+
+	close(releaseFlush)
+	require.NoError(t, <-flushDone)
+}
+
 func TestDecodeBlockRejectsShortPersistentValue(t *testing.T) {
 	c := &PersistentBlockCollector{}
 	for name, raw := range map[string][]byte{

cl/phase1/execution_client/execution_client_engine_test.go

@@ -114,6 +114,37 @@ func TestExecutionPayloadFromSSZBlock_BlockAccessListGloasOnly(t *testing.T) {
 	}
 }
 
+func TestExecutionPayloadFromSSZBlock_TransactionsAreJSONArray(t *testing.T) {
+	beaconCfg := clparams.MainnetBeaconConfig
+	tests := []struct {
+		name    string
+		version clparams.StateVersion
+		json    string
+		want    []any
+	}{
+		{name: "pre-Gloas empty", version: clparams.ElectraVersion, json: "[]", want: []any{}},
+		{name: "pre-Gloas non-empty", version: clparams.ElectraVersion, json: `["0x0102"]`, want: []any{"0x0102"}},
+		{name: "Gloas empty", version: clparams.GloasVersion, json: "[]", want: []any{}},
+		{name: "Gloas non-empty", version: clparams.GloasVersion, json: `["0x0102"]`, want: []any{"0x0102"}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			payload := cltypes.NewEth1Block(tt.version, &beaconCfg)
+			payload.Extra = solid.NewExtraData()
+			payload.Transactions = &solid.TransactionsSSZ{}
+			payload.Withdrawals = solid.NewStaticListSSZ[*cltypes.Withdrawal](int(beaconCfg.MaxWithdrawalsPerPayload), 44)
+			require.NoError(t, payload.Transactions.UnmarshalJSON([]byte(tt.json)))
+
+			raw, err := json.Marshal(engine_types.ExecutionPayloadFromSSZBlock(payload, tt.version))
+			require.NoError(t, err)
+
+			var decoded map[string]any
+			require.NoError(t, json.Unmarshal(raw, &decoded))
+			require.Equal(t, tt.want, decoded["transactions"])
+		})
+	}
+}
+
 func gloas(cfg *clparams.BeaconChainConfig, balData []byte) *cltypes.Eth1Block {
 	block := cltypes.NewEth1Block(clparams.GloasVersion, cfg)
 	block.Extra = solid.NewExtraData()