go-ethereum/core/txpool/blobpool/limbo.go

// Copyright 2023 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library 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.
//
// The go-ethereum library 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 the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.

package blobpool

import (
	"errors"
	"sort"

	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/log"
	"github.com/ethereum/go-ethereum/params"
	"github.com/ethereum/go-ethereum/rlp"
	"github.com/holiman/billy"
)

// limboBlob is a wrapper around an opaque blobset that also contains the tx hash
// to which it belongs as well as the block number in which it was included for
// finality eviction.
type limboBlob struct {
	TxHash common.Hash        // Owner transaction's hash to support resurrecting reorged txs
	Block  uint64             // Block in which the blob transaction was included
	Tx     *types.Transaction `rlp:"nil"` // Optional full blob transaction (old storage style)
	TxMeta *blobTxMeta        // the blob transaction metadata.
	Raw    []byte             // Canonical raw blob transaction payload for reinjection
	id     uint64             // the billy id of limboBlob
}

// limbo is a light, indexed database to temporarily store recently included
// blobs until they are finalized. The purpose is to support small reorgs, which
// would require pulling back up old blobs (which aren't part of the chain).
type limbo struct {
	store billy.Database             // Persistent data store for limboed blobs
	index map[common.Hash]*limboBlob // Mappings from tx hashes to datastore ids
}

// newLimbo opens and indexes a set of limboed blob transactions.
func newLimbo(_ *params.ChainConfig, datadir string) (*limbo, error) {
	l := &limbo{
		index: make(map[common.Hash]*limboBlob),
	}

	slotter := newLimboSlotter()

	// Index all limboed blobs on disk and delete anything unprocessable
	var fails []uint64
	index := func(id uint64, size uint32, data []byte) {
		if l.parseBlob(id, data) != nil {
			fails = append(fails, id)
		}
	}
	store, err := billy.Open(billy.Options{Path: datadir, Repair: true}, slotter, index)
	if err != nil {
		return nil, err
	}
	l.store = store

	if len(fails) > 0 {
		log.Warn("Dropping invalidated limboed blobs", "ids", fails)
		for _, id := range fails {
			if err := l.store.Delete(id); err != nil {
				l.Close()
				return nil, err
			}
		}
	}

	return l, nil
}

// newLimboSlotter returns a shelf layout that can read the legacy limbo formats
// and also store full blob transaction payloads for reinjection.
func newLimboSlotter() billy.SlotSizeFn {
	var (
		sizes []uint32
		seen  = make(map[uint32]struct{})
	)
	addSlotter := func(slotter billy.SlotSizeFn) {
		for {
			size, done := slotter()
			if _, ok := seen[size]; !ok {
				seen[size] = struct{}{}
				sizes = append(sizes, size)
			}
			if done {
				break
			}
		}
	}
	// Preserve compatibility with the metadata-only limbo format introduced on
	// this branch while also supporting the legacy and restored full payloads.
	sizes = append(sizes, 1024)
	seen[1024] = struct{}{}

	addSlotter(newSlotter(params.BlobTxMaxBlobs))
	addSlotter(newSlotterEIP7594(params.BlobTxMaxBlobs))

	sort.Slice(sizes, func(i, j int) bool { return sizes[i] < sizes[j] })
	var idx int
	return func() (size uint32, done bool) {
		size = sizes[idx]
		idx++
		return size, idx == len(sizes)
	}
}

// Close closes down the underlying persistent store.
func (l *limbo) Close() error {
	return l.store.Close()
}

// parseBlob is a callback method on limbo creation that gets called for each
// limboed blob on disk to create the in-memory metadata index.
func (l *limbo) parseBlob(id uint64, data []byte) error {
	item := new(limboBlob)
	if err := rlp.DecodeBytes(data, item); err != nil {
		// This path is impossible unless the disk data representation changes
		// across restarts. For that ever improbable case, recover gracefully
		// by ignoring this data entry.
		log.Error("Failed to decode blob limbo entry", "id", id, "err", err)
		return err
	}
	if _, ok := l.index[item.TxHash]; ok {
		// This path is impossible, unless due to a programming error a blob gets
		// inserted into the limbo which was already part of if. Recover gracefully
		// by ignoring this data entry.
		log.Error("Dropping duplicate blob limbo entry", "owner", item.TxHash, "id", id)
		return errors.New("duplicate blob")
	}
	item.id = id
	l.index[item.TxHash] = item

	return nil
}

// finalize evicts all blobs belonging to a recently finalized block or older.
func (l *limbo) finalize(final *types.Header) {
	// Just in case there's no final block yet (network not yet merged, weird
	// restart, sethead, etc), fail gracefully.
	if final == nil {
		log.Warn("Nil finalized block cannot evict old blobs")
		return
	}
	// Note: deleting keys from a map during range is explicitly safe in Go.
	// Any key deleted mid-iteration may or may not be visited; entries missed
	// here will be cleaned up on the next finalize call.
	for _, item := range l.index {
		if item.Block > final.Number.Uint64() {
			continue
		}
		if err := l.drop(item.TxHash); err != nil {
			log.Error("Failed to drop finalized blob", "block", item.Block, "id", item.id, "err", err)
		}
	}
}

// push stores a new blob transaction into the limbo, waiting until finality for
// it to be automatically evicted.
func (l *limbo) push(raw []byte, meta *blobTxMeta, block uint64) error {
	// If the blobs are already tracked by the limbo, consider it a programming
	// error. There's not much to do against it, but be loud.
	if _, ok := l.index[meta.hash]; ok {
		log.Error("Limbo cannot push already tracked blobs", "tx", meta.hash)
		return errors.New("already tracked blob transaction")
	}
	if err := l.setAndIndex(raw, nil, meta, block); err != nil {
		log.Error("Failed to set and index limboed blobs", "tx", meta.hash, "err", err)
		return err
	}
	return nil
}

// pull retrieves a previously pushed set of blobs back from the limbo, removing
// it at the same time. This method should be used when a previously included blob
// transaction gets reorged out.
func (l *limbo) pull(tx common.Hash) (*limboBlob, error) {
	// If the blobs are not tracked by the limbo, there's not much to do. This
	// can happen for example if a blob transaction is mined without pushing it
	// into the network first.
	item, ok := l.index[tx]
	if !ok {
		log.Trace("Limbo cannot pull non-tracked blobs", "tx", tx)
		return nil, errors.New("unseen blob transaction")
	}
	if err := l.drop(item.TxHash); err != nil {
		return nil, err
	}
	return item, nil
}

// update changes the block number under which a blob transaction is tracked. This
// method should be used when a reorg changes a transaction's inclusion block.
//
// The method may log errors for various unexpected scenarios but will not return
// any of it since there's no clear error case. Some errors may be due to coding
// issues, others caused by signers mining MEV stuff or swapping transactions. In
// all cases, the pool needs to continue operating.
func (l *limbo) update(txhash common.Hash, block uint64) {
	// If the blobs are not tracked by the limbo, there's not much to do. This
	// can happen for example if a blob transaction is mined without pushing it
	// into the network first.
	item, ok := l.index[txhash]
	if !ok {
		log.Trace("Limbo cannot update non-tracked blobs", "tx", txhash)
		return
	}
	// If there was no change in the blob's inclusion block, don't mess around
	// with heavy database operations.
	if item.Block == block {
		log.Trace("Blob transaction unchanged in limbo", "tx", txhash, "block", block)
		return
	}
	if err := l.drop(txhash); err != nil {
		log.Error("Failed to drop old limboed metadata", "tx", txhash, "err", err)
		return
	}
	if err := l.setAndIndex(item.Raw, item.Tx, item.TxMeta, block); err != nil {
		log.Error("Failed to set and index limboed blobs", "tx", txhash, "err", err)
		return
	}
	log.Trace("Blob transaction updated in limbo", "tx", txhash, "old-block", item.Block, "new-block", block)
}

// drop removes the blob metadata from the limbo.
func (l *limbo) drop(txhash common.Hash) error {
	if item, ok := l.index[txhash]; ok {
		// Remove the blob metadata entry from the limbo store. If anything fails,
		// there's not much to do besides logging the error and returning.
		if err := l.store.Delete(item.id); err != nil {
			log.Error("Failed to drop old limboed blobs", "tx", txhash, "err", err)
			return err
		}
		delete(l.index, txhash)
	}
	return nil
}

// setAndIndex assembles a limbo blob database entry and stores it, also updating
// the in-memory indices.
func (l *limbo) setAndIndex(raw []byte, tx *types.Transaction, meta *blobTxMeta, block uint64) error {
	var txhash common.Hash
	switch {
	case meta != nil:
		txhash = meta.hash
	case tx != nil:
		txhash = tx.Hash()
	default:
		return errors.New("missing limbo payload")
	}
	item := &limboBlob{
		TxHash: txhash,
		Block:  block,
		Raw:    raw,
		TxMeta: meta,
	}
	data, err := rlp.EncodeToBytes(item)
	if err != nil {
		panic(err) // cannot happen runtime, dev error
	}
	id, err := l.store.Put(data)
	if err != nil {
		return err
	}
	// Set the in-memory index
	item.id = id
	l.index[txhash] = item

	return nil
}