erigon-pulse/cl/persistence/block_saver.go

package persistence

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"

	"github.com/ledgerwatch/erigon/cl/sentinel/communication/ssz_snappy"
	"go.uber.org/zap/buffer"

	libcommon "github.com/ledgerwatch/erigon-lib/common"
	"github.com/ledgerwatch/erigon-lib/kv"
	"github.com/ledgerwatch/erigon/cl/clparams"
	"github.com/ledgerwatch/erigon/cl/cltypes"
	"github.com/ledgerwatch/erigon/cl/persistence/beacon_indicies"
	"github.com/ledgerwatch/erigon/cl/phase1/execution_client"
	"github.com/ledgerwatch/erigon/cl/sentinel/peers"
	"github.com/spf13/afero"
)

const subDivisionFolderSize = 10_000

type beaconChainDatabaseFilesystem struct {
	rawDB RawBeaconBlockChain
	cfg   *clparams.BeaconChainConfig

	executionEngine execution_client.ExecutionEngine
}

func NewBeaconChainDatabaseFilesystem(rawDB RawBeaconBlockChain, executionEngine execution_client.ExecutionEngine, cfg *clparams.BeaconChainConfig) BeaconChainDatabase {
	return beaconChainDatabaseFilesystem{
		rawDB:           rawDB,
		cfg:             cfg,
		executionEngine: executionEngine,
	}
}

func (b beaconChainDatabaseFilesystem) GetBlock(ctx context.Context, tx kv.Tx, slot uint64) (*peers.PeeredObject[*cltypes.SignedBeaconBlock], error) {
	blockRoot, err := beacon_indicies.ReadCanonicalBlockRoot(tx, slot)
	if err != nil {
		return nil, err
	}
	if blockRoot == (libcommon.Hash{}) {
		return nil, nil
	}

	r, err := b.rawDB.BlockReader(ctx, slot, blockRoot)
	if err != nil {
		return nil, err
	}
	defer r.Close()

	block := cltypes.NewSignedBeaconBlock(b.cfg)
	version := b.cfg.GetCurrentStateVersion(slot / b.cfg.SlotsPerEpoch)
	if err := ssz_snappy.DecodeAndReadNoForkDigest(r, block, version); err != nil {
		return nil, err
	}

	return &peers.PeeredObject[*cltypes.SignedBeaconBlock]{Data: block}, nil
}

func (b beaconChainDatabaseFilesystem) GetRange(ctx context.Context, tx kv.Tx, from uint64, count uint64) (*peers.PeeredObject[[]*cltypes.SignedBeaconBlock], error) {
	// Retrieve block roots for each ranged slot
	beaconBlockRooots, slots, err := beacon_indicies.ReadBeaconBlockRootsInSlotRange(ctx, tx, from, count)
	if err != nil {
		return nil, err
	}

	if len(beaconBlockRooots) == 0 {
		return &peers.PeeredObject[[]*cltypes.SignedBeaconBlock]{}, nil
	}

	blocks := []*cltypes.SignedBeaconBlock{}
	for idx, blockRoot := range beaconBlockRooots {
		slot := slots[idx]

		r, err := b.rawDB.BlockReader(ctx, slot, blockRoot)
		if errors.Is(err, afero.ErrFileNotFound) {
			continue
		}
		if err != nil {
			return nil, err
		}
		defer r.Close()

		block := cltypes.NewSignedBeaconBlock(b.cfg)
		version := b.cfg.GetCurrentStateVersion(slot / b.cfg.SlotsPerEpoch)
		if err := ssz_snappy.DecodeAndReadNoForkDigest(r, block, version); err != nil {
			return nil, err
		}

		blocks = append(blocks, block)
	}
	return &peers.PeeredObject[[]*cltypes.SignedBeaconBlock]{Data: blocks}, nil

}

func (b beaconChainDatabaseFilesystem) PurgeRange(ctx context.Context, tx kv.Tx, from uint64, count uint64) error {
	if err := beacon_indicies.RangeBlockRoots(ctx, tx, from, from+count, func(slot uint64, beaconBlockRoot libcommon.Hash) bool {
		b.rawDB.DeleteBlock(ctx, slot, beaconBlockRoot)
		return true
	}); err != nil {
		return err
	}

	return nil
}

func (b beaconChainDatabaseFilesystem) WriteBlock(ctx context.Context, tx kv.RwTx, block *cltypes.SignedBeaconBlock, canonical bool) error {
	blockRoot, err := block.Block.HashSSZ()
	if err != nil {
		return err
	}

	w, err := b.rawDB.BlockWriter(ctx, block.Block.Slot, blockRoot)
	if err != nil {
		return err
	}
	defer w.Close()

	if fp, ok := w.(afero.File); ok {
		err = fp.Truncate(0)
		if err != nil {
			return err
		}
		_, err = fp.Seek(0, io.SeekStart)
		if err != nil {
			return err
		}
	}

	err = ssz_snappy.EncodeAndWrite(w, block)
	if err != nil {
		return err
	}

	if fp, ok := w.(afero.File); ok {
		err = fp.Sync()
		if err != nil {
			return err
		}
	}
	bodyRoot, err := block.Block.Body.HashSSZ()
	if err != nil {
		return err
	}
	if block.Version() >= clparams.BellatrixVersion {
		if err := beacon_indicies.WriteExecutionBlockNumber(tx, blockRoot, block.Block.Body.ExecutionPayload.BlockNumber); err != nil {
			return err
		}
		if err := beacon_indicies.WriteExecutionBlockHash(tx, blockRoot, block.Block.Body.ExecutionPayload.BlockHash); err != nil {
			return err
		}
	}

	if err := beacon_indicies.WriteBeaconBlockHeaderAndIndicies(ctx, tx, &cltypes.SignedBeaconBlockHeader{
		Signature: block.Signature,
		Header: &cltypes.BeaconBlockHeader{
			Slot:          block.Block.Slot,
			ParentRoot:    block.Block.ParentRoot,
			ProposerIndex: block.Block.ProposerIndex,
			Root:          block.Block.StateRoot,
			BodyRoot:      bodyRoot,
		},
	}, canonical); err != nil {
		return err
	}
	return nil
}

// SlotToPaths define the file structure to store a block
//
// "/signedBeaconBlock/{slot/10_000}/{root}.ssz_snappy"
func rootToPaths(slot uint64, root libcommon.Hash, config *clparams.BeaconChainConfig) (folderPath string, filePath string) {
	// bufio
	buffer := bPool.Get().(*buffer.Buffer)
	defer bPool.Put(buffer)
	buffer.Reset()

	fmt.Fprintf(buffer, "%d/%x.sz", slot/subDivisionFolderSize, root)
	split := strings.Split(buffer.String(), "/")
	return split[0], buffer.String()
}