From 65a08322cbb99b3406fcac142e8f8817da66b726 Mon Sep 17 00:00:00 2001 From: shantichanal <158101918+shantichanal@users.noreply.github.com> Date: Wed, 9 Jul 2025 15:48:00 +0200 Subject: [PATCH] refactored and updated e2store framing for objects --- internal/era2/builder2.go | 94 +++++++------- internal/era2/era2.go | 266 +++++++++++++++----------------------- internal/era2/era_test.go | 28 ++-- 3 files changed, 165 insertions(+), 223 deletions(-) diff --git a/internal/era2/builder2.go b/internal/era2/builder2.go index 6fcfdfa5b4..14f3ce1b2e 100644 --- a/internal/era2/builder2.go +++ b/internal/era2/builder2.go @@ -58,7 +58,6 @@ const ( headerSize uint64 = 8 ) -// ProofVariant is an idiomatic “enum”. type proofvar uint16 const ( @@ -102,7 +101,6 @@ type Builder struct { writtenBytes uint64 } -// NewBuilder returns a new EraE Builder writing into the given io.Writer. func NewBuilder(w io.Writer) *Builder { buf := bytes.NewBuffer(nil) return &Builder{ @@ -117,27 +115,27 @@ func (b *Builder) Add(header types.Header, body types.Body, receipts types.Recei return fmt.Errorf("exceeds MaxEraESize %d", MaxEraESize) } + var pv proofvar = ProofNone + var pData []byte if proof != nil { - if proof.Variant == ProofNone || len(proof.Data) == 0 { - return fmt.Errorf("invalid proof: variant=%d len=%d", proof.Variant, len(proof.Data)) + pv = proof.Variant + pData = proof.Data + if pv == ProofNone || len(pData) == 0 { + return fmt.Errorf("invalid proof: variant=%d len=%d", pv, len(pData)) } } if len(b.headersRLP) == 0 { - if proof == nil { - b.prooftype = ProofNone - } else { - b.prooftype = proof.Variant - } - } else if proof.Variant != b.prooftype { - return fmt.Errorf("cannot mix proof variants: have %d want %d", b.prooftype, proof.Variant) + b.prooftype = pv + } else if pv != b.prooftype { + return fmt.Errorf("cannot mix proof variants: have %d want %d", b.prooftype, pv) } - hdr, err := rlp.EncodeToBytes(header) + hdr, err := rlp.EncodeToBytes(&header) if err != nil { return fmt.Errorf("error encoding block header: %w", err) } - bod, err := rlp.EncodeToBytes(body) + bod, err := rlp.EncodeToBytes(&body) if err != nil { return fmt.Errorf("error encoding block header: %w", err) } @@ -174,30 +172,24 @@ func (b *Builder) Finalize() (common.Hash, error) { return common.Hash{}, errors.New("no blocks added, cannot finalize") } var err error - b.headeroffsets, err = b.writeSection(TypeCompressedHeader, b.headersRLP, true) - if err != nil { - return common.Hash{}, fmt.Errorf("error writing compressed headers: %w", err) + if err := b.flushKind(TypeCompressedHeader, b.headersRLP, true, &b.headeroffsets); err != nil { + return common.Hash{}, err } - b.bodyoffsets, err = b.writeSection(TypeCompressedBody, b.bodiesRLP, true) - if err != nil { - return common.Hash{}, fmt.Errorf("error writing compressed bodies: %w", err) + if err := b.flushKind(TypeCompressedBody, b.bodiesRLP, true, &b.bodyoffsets); err != nil { + return common.Hash{}, err } - b.receiptoffsets, err = b.writeSection(TypeCompressedReceipts, b.receiptsRLP, true) - if err != nil { - return common.Hash{}, fmt.Errorf("error writing compressed receipts: %w", err) + if err := b.flushKind(TypeCompressedReceipts, b.receiptsRLP, true, &b.receiptoffsets); err != nil { + return common.Hash{}, err } if len(b.tds) > 0 { - b.tdoff, err = b.writeSection(TypeTotalDifficulty, b.tds, false) - if err != nil { - return common.Hash{}, fmt.Errorf("error writing total difficulties: %w", err) + if err := b.flushKind(TypeTotalDifficulty, b.tds, false, &b.tdoff); err != nil { + return common.Hash{}, err } } - - if b.prooftype != 0 { - b.proofoffsets, err = b.writeSection(uint16(b.prooftype), b.proofsRLP, true) - if err != nil { - return common.Hash{}, fmt.Errorf("error writing proofs: %w", err) + if b.prooftype != ProofNone { + if err := b.flushKind(uint16(b.prooftype), b.proofsRLP, true, &b.proofoffsets); err != nil { + return common.Hash{}, err } } @@ -236,7 +228,6 @@ func decodeBigs(raw [][]byte) []*big.Int { return out } -// snappyWrite is a small helper to take care snappy encoding and writing an e2store entry. func (b *Builder) snappyWrite(typ uint16, in []byte) error { var ( buf = b.buf @@ -258,29 +249,32 @@ func (b *Builder) snappyWrite(typ uint16, in []byte) error { return nil } -func (b *Builder) writeSection(typ uint16, list [][]byte, useSnappy bool) ([]uint64, error) { - if len(list) == 0 { - return nil, errors.New("cannot write empty section") - } - - buf := bytes.NewBuffer(nil) - offs := make([]uint64, len(list)) - for i, data := range list { - offs[i] = b.writtenBytes + headerSize + uint64(buf.Len()) - if useSnappy { - buf.Write(snappy.Encode(nil, data)) - } else { - buf.Write(data) +func (b *Builder) addEntry(typ uint16, payload []byte, snappyIt bool) (uint64, error) { + offset := b.writtenBytes + var err error + if snappyIt { + if err = b.snappyWrite(typ, payload); err != nil { + return 0, err } - } - - if n, err := b.w.Write(typ, buf.Bytes()); err != nil { - return nil, fmt.Errorf("error writing section %d: %w", typ, err) } else { + var n int + if n, err = b.w.Write(typ, payload); err != nil { + return 0, err + } b.writtenBytes += uint64(n) } + return offset, nil +} - return offs, nil +func (b *Builder) flushKind(typ uint16, list [][]byte, useSnappy bool, dst *[]uint64) error { + for _, data := range list { + off, err := b.addEntry(typ, data, useSnappy) + if err != nil { + return fmt.Errorf("entry type %d: %w", typ, err) + } + *dst = append(*dst, off) + } + return nil } func (b *Builder) writeIndex() error { @@ -306,7 +300,7 @@ func (b *Builder) writeIndex() error { binary.LittleEndian.PutUint64(idx[pos:], rel(b.receiptoffsets[i])) pos += 8 if len(b.tds) > 0 { - binary.LittleEndian.PutUint64(idx[pos+24:], rel(b.tdoff[i])) + binary.LittleEndian.PutUint64(idx[pos:], rel(b.tdoff[i])) pos += 8 } if len(b.proofsRLP) > 0 { diff --git a/internal/era2/era2.go b/internal/era2/era2.go index e01fd0980c..030c2400e1 100644 --- a/internal/era2/era2.go +++ b/internal/era2/era2.go @@ -57,23 +57,22 @@ type ReadAtSeekCloser interface { } type Era2 struct { - f ReadAtSeekCloser - s *e2store.Reader - m meta // metadata for the era2 file - mu *sync.Mutex - buf [8]byte // buffer reading entry offset - + f ReadAtSeekCloser + s *e2store.Reader + m meta // metadata for the era2 file + mu *sync.Mutex headeroff, bodyoff, receiptsoff, tdoff, proofsoff []uint64 // offsets for each entry type - rootheader uint64 // offset of the root header in the file if present indstart int64 + rootheader uint64 // offset of the root header in the file if present + prooftype uint16 } -func Open(filename string) (*Era2, error) { - f, err := os.Open(filename) +func Open(path string) (*Era2, error) { + f, err := os.Open(path) if err != nil { return nil, err } - e := &Era2{f: f, s: e2store.NewReader(f), mu: new(sync.Mutex)} + e := &Era2{f: f, s: e2store.NewReader(f)} if err := e.loadIndex(); err != nil { f.Close() return nil, err @@ -82,8 +81,6 @@ func Open(filename string) (*Era2, error) { } func (e *Era2) Close() error { - e.mu.Lock() - defer e.mu.Unlock() if e.f == nil { return nil } @@ -93,14 +90,10 @@ func (e *Era2) Close() error { } func (e *Era2) Start() uint64 { - e.mu.Lock() - defer e.mu.Unlock() return e.m.start } func (e *Era2) Count() uint64 { - e.mu.Lock() - defer e.mu.Unlock() return e.m.count } @@ -117,29 +110,29 @@ func (e *Era2) GetBlockByNumber(blockNum uint64) (*types.Block, error) { } func (e *Era2) getHeader(blockNum uint64) (*types.Header, error) { - e.mu.Lock() - defer e.mu.Unlock() if blockNum < e.m.start || blockNum >= e.m.start+e.m.count { return nil, fmt.Errorf("block number %d out of range [%d, %d)", blockNum, e.m.start, e.m.start+e.m.count) } - r, err := e.snappyPayload(e.headeroff[blockNum-e.m.start]) + r, _, err := e.s.ReaderAt(TypeCompressedHeader, int64(e.headeroff[blockNum-e.m.start])) if err != nil { - return nil, fmt.Errorf("error reading header for block %d: %w", blockNum, err) + return nil, err } + r = snappy.NewReader(r) + var h types.Header return &h, rlp.Decode(r, &h) } func (e *Era2) getBody(blockNum uint64) (*types.Body, error) { - e.mu.Lock() - defer e.mu.Unlock() if blockNum < e.m.start || blockNum >= e.m.start+e.m.count { return nil, fmt.Errorf("block number %d out of range [%d, %d)", blockNum, e.m.start, e.m.start+e.m.count) } - r, err := e.snappyPayload(e.bodyoff[blockNum-e.m.start]) + r, _, err := e.s.ReaderAt(TypeCompressedBody, int64(e.bodyoff[blockNum-e.m.start])) if err != nil { - return nil, fmt.Errorf("error reading body for block %d: %w", blockNum, err) + return nil, err } + r = snappy.NewReader(r) + var b types.Body return &b, rlp.Decode(r, &b) } @@ -153,57 +146,49 @@ func (e *Era2) getTD(blockNum uint64) (*big.Int, error) { return nil, fmt.Errorf("total-difficulty section not present") } - start := e.tdoff[blockNum-e.m.start] - buf := make([]byte, 32) - - if _, err := e.f.ReadAt(buf, int64(start)); err != nil { - return nil, fmt.Errorf("error reading total difficulty for block %d: %w", - blockNum, err) + r, _, err := e.s.ReaderAt(TypeTotalDifficulty, int64(e.tdoff[blockNum-e.m.start])) + if err != nil { + return nil, err } + buf, _ := io.ReadAll(r) td := new(big.Int).SetBytes(reverseOrder(buf)) return td, nil } -func (e *Era2) GetRawBodyFrameByNumber(n uint64) ([]byte, error) { - if n < e.m.start || n >= e.m.start+e.m.count { - return nil, fmt.Errorf("block number %d out of range [%d, %d)", n, e.m.start, e.m.start+e.m.count) +func (e *Era2) GetRawBodyFrameByNumber(blockNum uint64) ([]byte, error) { + if blockNum < e.m.start || blockNum >= e.m.start+e.m.count { + return nil, fmt.Errorf("block number %d out of range [%d, %d)", blockNum, e.m.start, e.m.start+e.m.count) } - start := e.bodyoff[n-e.m.start] - end := e.nextBoundary(n, e.bodyoff, e.receiptsoff) // receipts section is the next safest fallback - length := end - start - sr := io.NewSectionReader(e.f, int64(start), int64(length)) - return io.ReadAll(sr) + r, _, err := e.s.ReaderAt(TypeCompressedBody, int64(e.bodyoff[blockNum-e.m.start])) + if err != nil { + return nil, err + } + return io.ReadAll(r) } -func (e *Era2) GetRawReceiptsFrameByNumber(n uint64) ([]byte, error) { - if n < e.m.start || n >= e.m.start+e.m.count { - return nil, fmt.Errorf("block number %d out of range [%d, %d)", n, e.m.start, e.m.start+e.m.count) +func (e *Era2) GetRawReceiptsFrameByNumber(blockNum uint64) ([]byte, error) { + if blockNum < e.m.start || blockNum >= e.m.start+e.m.count { + return nil, fmt.Errorf("block number %d out of range [%d, %d)", blockNum, e.m.start, e.m.start+e.m.count) } - start := e.receiptsoff[n-e.m.start] - end := e.nextBoundary(n, e.receiptsoff, e.tdoff) // TD sec. is next fallback - length := end - start - sr := io.NewSectionReader(e.f, int64(start), int64(length)) - return io.ReadAll(sr) + r, _, err := e.s.ReaderAt(TypeCompressedReceipts, int64(e.receiptsoff[blockNum-e.m.start])) + if err != nil { + return nil, err + } + return io.ReadAll(r) } -func (e *Era2) GetRawProofFrameByNumber(n uint64) ([]byte, error) { +func (e *Era2) GetRawProofFrameByNumber(blockNum uint64) ([]byte, error) { if len(e.proofsoff) == 0 { return nil, fmt.Errorf("proofs section not present") } - if n < e.m.start || n >= e.m.start+e.m.count { - return nil, fmt.Errorf("block number %d out of range [%d, %d)", n, e.m.start, e.m.start+e.m.count) + if blockNum < e.m.start || blockNum >= e.m.start+e.m.count { + return nil, fmt.Errorf("block number %d out of range [%d, %d)", blockNum, e.m.start, e.m.start+e.m.count) } - start := e.proofsoff[n-e.m.start] - - // fallback next proof frame → AccRoot header → BlockIndex - var nextSec []uint64 - if e.rootheader != 0 { - nextSec = []uint64{e.rootheader} + r, _, err := e.s.ReaderAt(e.prooftype, int64(e.proofsoff[blockNum-e.m.start])) + if err != nil { + return nil, err } - end := e.nextBoundary(n, e.proofsoff, nextSec) - length := end - start - sr := io.NewSectionReader(e.f, int64(start), int64(length)) - return io.ReadAll(sr) + return io.ReadAll(r) } func (e *Era2) rawPayload(abs uint64) ([]byte, error) { @@ -216,20 +201,6 @@ func (e *Era2) snappyPayload(abs uint64) (io.Reader, error) { return snappy.NewReader(sr), nil } -func (e *Era2) nextBoundary(idx uint64, sameSec []uint64, nextSec []uint64) uint64 { - local := idx - e.m.start - // next frame in the same section - if local+1 < uint64(len(sameSec)) { - return sameSec[local+1] - } - // first frame of the NEXT section (if present) - if len(nextSec) > 0 { - return nextSec[0] - } - // otherwise clamp to start of BlockIndex - return uint64(e.indstart) -} - func (e *Era2) loadIndex() error { var err error e.m.filelen, err = e.f.Seek(0, io.SeekEnd) @@ -291,6 +262,14 @@ func (e *Era2) loadIndex() error { } } + if len(e.proofsoff) > 0 { + typ, _, perr := e.s.ReadMetadataAt(int64(e.proofsoff[0])) + if perr != nil { + return fmt.Errorf("read proof header: %w", perr) + } + e.prooftype = typ + } + var off int64 = 0 // start at byte-0 of file for off < e.indstart { // never enter the Block-Index TLV @@ -307,116 +286,79 @@ func (e *Era2) loadIndex() error { return nil } -func (e *Era2) BatchRange(first, count uint64, wantHeaders, wantBodies, wantReceipts, wantProofs bool) (hdrs []*types.Header, bods []*types.Body, recs []types.Receipts, prfs [][]byte, err error) { +func (e *Era2) BatchRange(first, count uint64, wantHdr, wantBody, wantRec, wantPrf bool) (hdrs []*types.Header, bods []*types.Body, recs []types.Receipts, prfs [][]byte, err error) { if count == 0 { - err = fmt.Errorf("count must be greater than 0") + err = fmt.Errorf("count must be > 0") return } - if first < e.m.start || first+count > e.m.start+e.m.count { - err = fmt.Errorf("range [%d,%d) out of bounds [%d,%d)", first, first+count, e.m.start, e.m.start+e.m.count) + err = fmt.Errorf("range [%d,%d) out of bounds", first, first+count) return } - id0 := first - e.m.start - if wantHeaders { + idx := first - e.m.start + if wantHdr { hdrs = make([]*types.Header, count) } - if wantBodies { + if wantBody { bods = make([]*types.Body, count) } - if wantReceipts { + if wantRec { recs = make([]types.Receipts, count) } - if wantProofs { + if wantPrf { prfs = make([][]byte, count) } - stream := func(startOff uint64, endOff uint64, decode func(io.Reader, uint64) error) error { - length := int64(endOff) - int64(startOff) - r := snappy.NewReader(io.NewSectionReader(e.f, int64(startOff), length)) - for i := uint64(0); i < count; i++ { - if err := decode(r, i); err != nil { - return err - } - } - return nil - } + for i := uint64(0); i < count; i++ { + id := idx + i - if wantHeaders { - err = stream(e.headeroff[id0], e.bodyoff[0], func(r io.Reader, i uint64) error { - var hdr types.Header - if err := rlp.Decode(r, &hdr); err != nil { - return fmt.Errorf("error decoding header for block %d: %w", first+i, err) - } - hdrs[i] = &hdr - return nil - }) - if err != nil { - return - } - } - - if wantBodies { - err = stream(e.bodyoff[id0], e.receiptsoff[0], func(r io.Reader, i uint64) error { - var body types.Body - if err := rlp.Decode(r, &body); err != nil { - return fmt.Errorf("error decoding body for block %d: %w", first+i, err) - } - bods[i] = &body - return nil - }) - if err != nil { - return - } - } - - if wantReceipts { - var receiptsEnd uint64 - if len(e.tdoff) > 0 { - receiptsEnd = e.tdoff[0] - } else { - receiptsEnd = uint64(e.indstart) - } - err = stream(e.receiptsoff[id0], receiptsEnd, func(r io.Reader, i uint64) error { - var rct types.Receipts - if err := rlp.Decode(r, &rct); err != nil { - return fmt.Errorf("error decoding receipts for block %d: %w", first+i, err) - } - recs[i] = rct - return nil - }) - if err != nil { - return - } - } - - if wantProofs { - if len(prfs) == 0 { - err = fmt.Errorf("proofs section is not present") - return - } - - for i := uint64(0); i < count; i++ { - start := e.proofsoff[id0+i] - var end uint64 - if id0+i+1 < uint64(len(e.proofsoff)) { - end = e.proofsoff[id0+i+1] // next proof frame - } else if e.rootheader != 0 { - end = e.rootheader // AccRoot header - } else { - end = uint64(e.indstart) // BlockIndex header - } - length := int64(end - start) - frame, readErr := io.ReadAll(io.NewSectionReader(e.f, int64(start), length)) - if readErr != nil { - err = fmt.Errorf("proof read %d: %w", first+i, readErr) + if wantHdr { + r, _, er := e.s.ReaderAt(TypeCompressedHeader, int64(e.headeroff[id])) + if er != nil { + err = er return } - prfs[i] = frame + if er = rlp.Decode(snappy.NewReader(r), &hdrs[i]); er != nil { + err = er + return + } + } + if wantBody { + r, _, er := e.s.ReaderAt(TypeCompressedBody, int64(e.bodyoff[id])) + if er != nil { + err = er + return + } + if er = rlp.Decode(snappy.NewReader(r), &bods[i]); er != nil { + err = er + return + } + } + if wantRec { + r, _, er := e.s.ReaderAt(TypeCompressedReceipts, int64(e.receiptsoff[id])) + if er != nil { + err = er + return + } + if er = rlp.Decode(snappy.NewReader(r), &recs[i]); er != nil { + err = er + return + } + } + if wantPrf { + if len(e.proofsoff) == 0 { + err = fmt.Errorf("proofs section not present") + return + } + r, _, er := e.s.ReaderAt(e.prooftype, int64(e.proofsoff[id])) // type already checked when writing + if er != nil { + err = er + return + } + prfs[i], _ = io.ReadAll(r) } } - return } diff --git a/internal/era2/era_test.go b/internal/era2/era_test.go index 6978c02ba3..1b49e033cb 100644 --- a/internal/era2/era_test.go +++ b/internal/era2/era_test.go @@ -19,6 +19,7 @@ package era2 import ( "bytes" "fmt" + "io" "math/big" "os" "testing" @@ -26,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/rlp" + "github.com/klauspost/compress/snappy" ) type testchain struct { @@ -37,7 +39,6 @@ type testchain struct { func TestEra2Builder(t *testing.T) { t.Parallel() - // Get temp directory. f, err := os.CreateTemp(t.TempDir(), "era2-test") if err != nil { @@ -76,22 +77,19 @@ func TestEra2Builder(t *testing.T) { } // 3. open reader + t.Logf("filename: %s", f.Name()) era, err := Open(f.Name()) if err != nil { t.Fatalf("open era: %v", err) } defer era.Close() - // 4. verify every block for i := 0; i < 128; i++ { bn := uint64(i) - - // -- header + body via GetBlockByNumber ------------------------------ gotBlock, err := era.GetBlockByNumber(bn) if err != nil { t.Fatalf("get block %d: %v", i, err) } - if chain.headers[i].Hash() != gotBlock.Header().Hash() { t.Fatalf("header %d mismatch", i) } @@ -99,23 +97,31 @@ func TestEra2Builder(t *testing.T) { t.Fatalf("body %d mismatch", i) } - // -- raw body frame -------------------------------------------------- rawBody, err := era.GetRawBodyFrameByNumber(bn) if err != nil { t.Fatalf("raw body %d: %v", i, err) } - expectBody := mustEncode(chain.bodies[i]) - if !bytes.Contains(rawBody, expectBody) { // frame may include next + decBody, err := io.ReadAll( + snappy.NewReader(bytes.NewReader(rawBody)), + ) + if err != nil { + t.Fatalf("snappy decode body %d: %v", i, err) + } + if !bytes.Equal(decBody, mustEncode(chain.bodies[i])) { t.Fatalf("body frame %d mismatch", i) } - // -- raw receipts frame --------------------------------------------- rawRcpt, err := era.GetRawReceiptsFrameByNumber(bn) if err != nil { t.Fatalf("raw receipts %d: %v", i, err) } - expectRcpt := mustEncode(chain.receipts[i]) - if !bytes.Contains(rawRcpt, expectRcpt) { + decRcpt, err := io.ReadAll( + snappy.NewReader(bytes.NewReader(rawRcpt)), + ) + if err != nil { + t.Fatalf("snappy decode receipts %d: %v", i, err) + } + if !bytes.Equal(decRcpt, mustEncode(chain.receipts[i])) { t.Fatalf("receipts frame %d mismatch", i) } }