Skip to content

feat:ephemeral header put and state for findcontent an offer #182

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
GrapeBaBa wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat:ephemeral header put and state for findcontent an offer #182

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions beacon/storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ func NewBeaconStorage(config storage.PortalStorageConfig, db *pebble.DB) (storag
log: log.New("beacon_storage"),
spec: config.Spec,
cache: &beaconStorageCache{},
writeOptions: &pebble.WriteOptions{Sync: false},
bytePool: sync.Pool{
New: func() interface{} {
out := make([]byte, 8)
Expand Down
278 changes: 188 additions & 90 deletions history/ephemeral_storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,131 +41,229 @@ func NewEphemeralStorage(config storage.PortalStorageConfig, db *pebble.DB) *Eph
// The resulting value is a serialized `EphemeralHeaderPayload`.
// The contentId is not used in this implementation.
func (s *EphemeralStorage) Get(contentKey []byte, _ []byte) ([]byte, error) {
parsedKey := &history.FindContentEphemeralHeadersKey{}
err := parsedKey.UnmarshalSSZ(contentKey[1:])
if err != nil {
return nil, fmt.Errorf("failed to unmarshal contentKey: %w", err)
}

if parsedKey.AncestorCount > maxAncestorCount {
return nil, errInvalidAncestorCount
}

startBlockNumKeyBytes, numCloser, err := s.db.Get(parsedKey.BlockHash[:])
if err != nil {
if errors.Is(err, pebble.ErrNotFound) {
return nil, fmt.Errorf("block number key not found for hash %x: %w", parsedKey.BlockHash, err)
switch history.ContentType(contentKey[0]) {
case history.FindContentEphemeralType:
parsedKey := &history.FindContentEphemeralHeadersKey{}
err := parsedKey.UnmarshalSSZ(contentKey[1:])
if err != nil {
return nil, fmt.Errorf("failed to unmarshal contentKey: %w", err)
}
return nil, fmt.Errorf("failed to get block number key for hash %x: %w", parsedKey.BlockHash, err)
}

startBlockNumKey := make([]byte, len(startBlockNumKeyBytes))
copy(startBlockNumKey, startBlockNumKeyBytes)
if numCloser != nil {
if numCloseErr := numCloser.Close(); numCloseErr != nil {
return nil, fmt.Errorf("failed to close numCloser: %w", numCloseErr)
if parsedKey.AncestorCount > maxAncestorCount {
return nil, errInvalidAncestorCount
}
}

initialActualBlockNum, err := decodeBlockNumber(startBlockNumKey)
if err != nil {
return nil, fmt.Errorf("failed to decode initial block number from key %x: %w", startBlockNumKey, err)
}

initialHeaderBytes, headerCloser, err := s.db.Get(startBlockNumKey)
if err != nil {
if errors.Is(err, pebble.ErrNotFound) {
return nil, fmt.Errorf("header not found for start block number key %x (decoded: %d): %w", startBlockNumKey, initialActualBlockNum, err)
startBlockNumKeyBytes, numCloser, err := s.db.Get(parsedKey.BlockHash[:])
if err != nil {
if errors.Is(err, pebble.ErrNotFound) {
return nil, fmt.Errorf("block number key not found for hash %x: %w", parsedKey.BlockHash, err)
}
return nil, fmt.Errorf("failed to get block number key for hash %x: %w", parsedKey.BlockHash, err)
}
return nil, fmt.Errorf("failed to get header for start block number key %x: %w", startBlockNumKey, err)
}

initialHeader := make([]byte, len(initialHeaderBytes))
copy(initialHeader, initialHeaderBytes)
if headerCloser != nil {
if headCloseErr := headerCloser.Close(); headCloseErr != nil {
return nil, fmt.Errorf("failed to close headerCloser: %w", headCloseErr)
startBlockNumKey := make([]byte, len(startBlockNumKeyBytes))
copy(startBlockNumKey, startBlockNumKeyBytes)
if numCloser != nil {
if numCloseErr := numCloser.Close(); numCloseErr != nil {
return nil, fmt.Errorf("failed to close numCloser: %w", numCloseErr)
}
}
}

collectedHeaders := make([][]byte, 0, 1+int(parsedKey.AncestorCount))
collectedHeaders = append(collectedHeaders, initialHeader)
lastSuccessfullyAddedBlockNum := initialActualBlockNum

if parsedKey.AncestorCount > 0 {
iterOpts := &pebble.IterOptions{}
iter, err := s.db.NewIter(iterOpts)
initialActualBlockNum, err := decodeBlockNumber(startBlockNumKey)
if err != nil {
return nil, fmt.Errorf("failed to create iterator: %w", err)
return nil, fmt.Errorf("failed to decode initial block number from key %x: %w", startBlockNumKey, err)
}
defer iter.Close()

if !iter.SeekGE(startBlockNumKey) {
if iterErr := iter.Error(); iterErr != nil {
return nil, fmt.Errorf("iterator error on SeekGE for known key %x: %w", startBlockNumKey, iterErr)
initialHeaderBytes, headerCloser, err := s.db.Get(startBlockNumKey)
if err != nil {
if errors.Is(err, pebble.ErrNotFound) {
return nil, fmt.Errorf("header not found for start block number key %x (decoded: %d): %w", startBlockNumKey, initialActualBlockNum, err)
}
return nil, fmt.Errorf("SeekGE failed to find known key %x (iterator valid: %t)", startBlockNumKey, iter.Valid())
return nil, fmt.Errorf("failed to get header for start block number key %x: %w", startBlockNumKey, err)
}

if !bytes.Equal(iter.Key(), startBlockNumKey) {
if iterErr := iter.Error(); iterErr != nil {
return nil, fmt.Errorf("iterator error after SeekGE found %x instead of known key %x: %w", iter.Key(), startBlockNumKey, iterErr)
initialHeader := make([]byte, len(initialHeaderBytes))
copy(initialHeader, initialHeaderBytes)
if headerCloser != nil {
if headCloseErr := headerCloser.Close(); headCloseErr != nil {
return nil, fmt.Errorf("failed to close headerCloser: %w", headCloseErr)
}
return nil, fmt.Errorf("SeekGE for known key %x landed on a different key %x", startBlockNumKey, iter.Key())
}

for i := range int(parsedKey.AncestorCount) {
if !iter.Prev() {
collectedHeaders := make([][]byte, 0, 1+int(parsedKey.AncestorCount))
collectedHeaders = append(collectedHeaders, initialHeader)
lastSuccessfullyAddedBlockNum := initialActualBlockNum

if parsedKey.AncestorCount > 0 {
iterOpts := &pebble.IterOptions{}
iter, err := s.db.NewIter(iterOpts)
if err != nil {
return nil, fmt.Errorf("failed to create iterator: %w", err)
}
defer iter.Close()

if !iter.SeekGE(startBlockNumKey) {
if iterErr := iter.Error(); iterErr != nil {
return nil, fmt.Errorf("iterator error on Prev(): %w", iterErr)
return nil, fmt.Errorf("iterator error on SeekGE for known key %x: %w", startBlockNumKey, iterErr)
}
s.log.Debug("Iterator exhausted during Prev() for ancestors", "retrieved_count", i)
break
return nil, fmt.Errorf("SeekGE failed to find known key %x (iterator valid: %t)", startBlockNumKey, iter.Valid())
}

currentIterKey := iter.Key()
currentIterActualBlockNum, err := decodeBlockNumber(currentIterKey)
if err != nil {
s.log.Error("Failed to decode block number from iterator key", "key", currentIterKey, "err", err)
break
if !bytes.Equal(iter.Key(), startBlockNumKey) {
if iterErr := iter.Error(); iterErr != nil {
return nil, fmt.Errorf("iterator error after SeekGE found %x instead of known key %x: %w", iter.Key(), startBlockNumKey, iterErr)
}
return nil, fmt.Errorf("SeekGE for known key %x landed on a different key %x", startBlockNumKey, iter.Key())
}

if currentIterActualBlockNum != lastSuccessfullyAddedBlockNum-1 {
s.log.Warn("Discontinuity found in ancestor chain.",
"last_good_block", lastSuccessfullyAddedBlockNum,
"found_block", currentIterActualBlockNum,
"expected_block", lastSuccessfullyAddedBlockNum-1)
break
}
for i := range int(parsedKey.AncestorCount) {
if !iter.Prev() {
if iterErr := iter.Error(); iterErr != nil {
return nil, fmt.Errorf("iterator error on Prev(): %w", iterErr)
}
s.log.Debug("Iterator exhausted during Prev() for ancestors", "retrieved_count", i)
break
}

ancestorHeader := make([]byte, len(iter.Value()))
copy(ancestorHeader, iter.Value())
collectedHeaders = append(collectedHeaders, ancestorHeader)
lastSuccessfullyAddedBlockNum = currentIterActualBlockNum
currentIterKey := iter.Key()
currentIterActualBlockNum, err := decodeBlockNumber(currentIterKey)
if err != nil {
s.log.Error("Failed to decode block number from iterator key", "key", currentIterKey, "err", err)
break
}

if currentIterActualBlockNum != lastSuccessfullyAddedBlockNum-1 {
s.log.Warn("Discontinuity found in ancestor chain.",
"last_good_block", lastSuccessfullyAddedBlockNum,
"found_block", currentIterActualBlockNum,
"expected_block", lastSuccessfullyAddedBlockNum-1)
break
}

ancestorHeader := make([]byte, len(iter.Value()))
copy(ancestorHeader, iter.Value())
collectedHeaders = append(collectedHeaders, ancestorHeader)
lastSuccessfullyAddedBlockNum = currentIterActualBlockNum
}
}
}

payload := &history.EphemeralHeaderPayload{
Payload: collectedHeaders,
}
payload := &history.EphemeralHeaderPayload{
Payload: collectedHeaders,
}

payloadBytes, err := payload.MarshalSSZ()
if err != nil {
return nil, fmt.Errorf("failed to marshal payload: %w", err)
payloadBytes, err := payload.MarshalSSZ()
if err != nil {
return nil, fmt.Errorf("failed to marshal payload: %w", err)
}
return payloadBytes, nil
case history.OfferEphemeralType:
return nil, fmt.Errorf("OfferEphemeralType is not supported")

Copilot AI May 22, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Clarify in-code or in documentation why OfferEphemeralType is unsupported in the Get function, to avoid confusion from API consumers regarding its handling contrasted with the Put implementation.

Copilot uses AI. Check for mistakes.
default:
return nil, fmt.Errorf("unsupported content type: %d", contentKey[0])
}
return payloadBytes, nil
}

// Put implements storage.ContentStorage.
// It stores the value associated with the given contentKey in the database.
// The contentId is not used in this implementation.
func (s *EphemeralStorage) Put(contentKey []byte, _ []byte, value []byte) error {
batch := s.db.NewBatch()
defer batch.Close()
if err := batch.Set(contentKey, value, s.writeOptions); err != nil {
return err
switch history.ContentType(contentKey[0]) {
case history.FindContentEphemeralType:
ephemeralHeaderPayload, err := history.DecodeEphemeralHeaderPayload(value)
if err != nil {
return err
}
if len(ephemeralHeaderPayload.Payload) == 0 {
return nil
}
firstHeader, err := history.DecodeBlockHeader(ephemeralHeaderPayload.Payload[0])
if err != nil {
return err
}
decodedKey := &history.FindContentEphemeralHeadersKey{}
err = decodedKey.UnmarshalSSZ(contentKey[1:])
if err != nil {
return fmt.Errorf("failed to unmarshal contentKey: %w", err)
}

if !bytes.Equal(firstHeader.Hash().Bytes(), decodedKey.BlockHash[:]) {
return fmt.Errorf("block hash mismatch: %x != %x", firstHeader.Hash().Bytes(), decodedKey.BlockHash[:])
}

batch := s.db.NewBatch()
defer batch.Close()
firstBlockNumBytes := make([]byte, 8)
binary.BigEndian.PutUint64(firstBlockNumBytes, firstHeader.Number.Uint64())
if err := batch.Set(decodedKey.BlockHash[:], firstBlockNumBytes, s.writeOptions); err != nil {
return err
}
if err := batch.Set(firstBlockNumBytes, ephemeralHeaderPayload.Payload[0], s.writeOptions); err != nil {
return err
}
prevHeader := firstHeader
for _, header := range ephemeralHeaderPayload.Payload[1:] {
decodedHeader, err := history.DecodeBlockHeader(header)
if err != nil {
return err
}
if !bytes.Equal(prevHeader.ParentHash.Bytes(), decodedHeader.Hash().Bytes()) {
return fmt.Errorf("parent hash mismatch: %x != %x", prevHeader.ParentHash.Bytes(), decodedHeader.Hash().Bytes())
}
if prevHeader.Number.Uint64() != decodedHeader.Number.Uint64()+1 {
return fmt.Errorf("block number mismatch in ancestor chain: expected %d (parent of %d), but got %d", decodedHeader.Number.Uint64()+1, decodedHeader.Number.Uint64(), prevHeader.Number.Uint64())
}
blockNumBytes := make([]byte, 8)
binary.BigEndian.PutUint64(blockNumBytes, decodedHeader.Number.Uint64())
if err := batch.Set(decodedHeader.Hash().Bytes(), blockNumBytes, s.writeOptions); err != nil {
return err
}
if err := batch.Set(blockNumBytes, header, s.writeOptions); err != nil {
return err
}
prevHeader = decodedHeader
}
if err := batch.Commit(s.writeOptions); err != nil {
return fmt.Errorf("failed to commit batch: %w", err)
}
return nil
case history.OfferEphemeralType:

Copilot AI May 22, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[nitpick] The OfferEphemeralType branch in the Put function includes multiple batch operations and error checks; consider refactoring this block into smaller helper functions to improve readability and long-term maintainability.

Copilot uses AI. Check for mistakes.
offerEphemeralHeader, err := history.DecodeOfferEphemeralHeader(value)
if err != nil {
return err
}

header, err := history.DecodeBlockHeader(offerEphemeralHeader.Header)
if err != nil {
return err
}
_, closer, err := s.db.Get(header.Hash().Bytes())
if closer != nil {
if closeErr := closer.Close(); closeErr != nil {
return fmt.Errorf("failed to close closer: %w", closeErr)
}
}
if err != nil {
if errors.Is(err, pebble.ErrNotFound) {
batch := s.db.NewBatch()
defer batch.Close()
blockNumBytes := make([]byte, 8)
binary.BigEndian.PutUint64(blockNumBytes, header.Number.Uint64())
if err := batch.Set(header.Hash().Bytes(), blockNumBytes, s.writeOptions); err != nil {
return err
}
if err := batch.Set(blockNumBytes, offerEphemeralHeader.Header, s.writeOptions); err != nil {
return err
}
if err := batch.Commit(s.writeOptions); err != nil {
return fmt.Errorf("failed to commit batch: %w", err)
}
}
return fmt.Errorf("failed to get block number key for hash %x: %w", header.Hash(), err)
}
return nil
default:
return fmt.Errorf("unsupported content type: %d", contentKey[0])
}
return batch.Commit(s.writeOptions)
}

// Radius returns the storage radius.
Expand Down
6 changes: 6 additions & 0 deletions history/validation.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,12 @@ func (h *HistoryValidator) ValidateContent(contentKey []byte, content []byte) er
}
_, err = ValidatePortalReceiptsBytes(content, header.ReceiptHash.Bytes())
return err
case history.FindContentEphemeralType:

Copilot AI May 22, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consider adding a brief comment explaining why validation for ephemeral content is postponed, ensuring future maintainers understand that this is an intentional design decision.

Copilot uses AI. Check for mistakes.
// postpone the validation of ephemeral content to storage
return nil
case history.OfferEphemeralType:
// postpone the validation of ephemeral content to storage
return nil
}
return errors.New("unknown content type in validation")
}
12 changes: 12 additions & 0 deletions types/history/types_helper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,18 @@ func DecodeBlockHeader(headerBytes []byte) (*types.Header, error) {
return header, nil
}

func DecodeEphemeralHeaderPayload(payload []byte) (*EphemeralHeaderPayload, error) {
ephemeralHeaderPayload := new(EphemeralHeaderPayload)
err := ephemeralHeaderPayload.UnmarshalSSZ(payload)
return ephemeralHeaderPayload, err
}

func DecodeOfferEphemeralHeader(payload []byte) (*OfferEphemeralHeader, error) {
offerEphemeralHeader := new(OfferEphemeralHeader)
err := offerEphemeralHeader.UnmarshalSSZ(payload)
return offerEphemeralHeader, err
}

func GetEpochIndex(blockNumber uint64) uint64 {
return blockNumber / EpochSize
}
Expand Down