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"

	"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
	Ptx    *blobTxForPool
}

// 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).
//
// TODO(karalabe): Currently updating the inclusion block of a blob needs a full db rewrite. Can we do without?
type limbo struct {
	store billy.Database // Persistent data store for limboed blobs

	index  map[common.Hash]uint64            // Mappings from tx hashes to datastore ids
	groups map[uint64]map[uint64]common.Hash // Set of txs included in past blocks
}

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

	// Create new slotter for pre-Osaka blob configuration.
	slotter := newSlotter(params.BlobTxMaxBlobs)

	// See if we need to migrate the limbo after fusaka.
	slotter, err := tryMigrate(config, slotter, datadir)
	if err != nil {
		return nil, err
	}

	// 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
}

// 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")
	}
	l.index[item.TxHash] = id

	if _, ok := l.groups[item.Block]; !ok {
		l.groups[item.Block] = make(map[uint64]common.Hash)
	}
	l.groups[item.Block][id] = item.TxHash

	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
	}
	for block, ids := range l.groups {
		if block > final.Number.Uint64() {
			continue
		}
		for id, owner := range ids {
			if err := l.store.Delete(id); err != nil {
				log.Error("Failed to drop finalized blob", "block", block, "id", id, "err", err)
			}
			delete(l.index, owner)
		}
		delete(l.groups, block)
	}
}

// push stores a new blob transaction into the limbo, waiting until finality for
// it to be automatically evicted.
func (l *limbo) push(ptx *blobTxForPool, block uint64) error {
	hash := ptx.Tx.Hash()
	if _, ok := l.index[hash]; ok {
		log.Error("Limbo cannot push already tracked blobs", "tx", hash)
		return errors.New("already tracked blob transaction")
	}
	if err := l.setAndIndex(ptx, block); err != nil {
		log.Error("Failed to set and index limboed blobs", "tx", 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) (*blobTxForPool, 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.
	id, ok := l.index[tx]
	if !ok {
		log.Trace("Limbo cannot pull non-tracked blobs", "tx", tx)
		return nil, errors.New("unseen blob transaction")
	}
	item, err := l.getAndDrop(id)
	if err != nil {
		log.Error("Failed to get and drop limboed blobs", "tx", tx, "id", id, "err", err)
		return nil, err
	}
	return item.Ptx, 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.
	id, 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 _, ok := l.groups[block][id]; ok {
		log.Trace("Blob transaction unchanged in limbo", "tx", txhash, "block", block)
		return
	}
	// Retrieve the old blobs from the data store and write them back with a new
	// block number. The new entry is written before the old one is dropped: blob
	// sidecars are not stored in chain state, so the limbo is the only place
	// from which a re-injection on a reorg-rollback can recover them. Dropping
	// the existing entry first (as the previous implementation did) would lose
	// the blob outright if the subsequent setAndIndex call failed.
	data, err := l.store.Get(id)
	if err != nil {
		log.Error("Failed to load limboed blobs", "tx", txhash, "id", id, "err", err)
		return
	}
	item := new(limboBlob)
	if err := rlp.DecodeBytes(data, item); err != nil {
		log.Error("Failed to decode limboed blobs", "tx", txhash, "id", id, "err", err)
		return
	}
	if err := l.setAndIndex(item.Ptx, block); err != nil {
		log.Error("Failed to set and index limboed blobs", "tx", txhash, "err", err)
		return
	}
	// The new entry is now durable in the store and indexed under the new
	// block; setAndIndex has overwritten l.index[txhash] with the new id, so
	// clean up the old store entry and group mapping. A failure to drop the
	// old store id only leaks a billy slot - the blob is still preserved
	// under the new id.
	if err := l.store.Delete(id); err != nil {
		log.Error("Failed to drop superseded limbo entry", "tx", txhash, "old-id", id, "err", err)
	}
	delete(l.groups[item.Block], id)
	if len(l.groups[item.Block]) == 0 {
		delete(l.groups, item.Block)
	}
	log.Trace("Blob transaction updated in limbo", "tx", txhash, "old-block", item.Block, "new-block", block)
}

// getAndDrop retrieves a blob item from the limbo store and deletes it both from
// the store and indices.
func (l *limbo) getAndDrop(id uint64) (*limboBlob, error) {
	data, err := l.store.Get(id)
	if err != nil {
		return nil, err
	}
	item := new(limboBlob)
	if err = rlp.DecodeBytes(data, item); err != nil {
		return nil, err
	}
	delete(l.index, item.TxHash)
	delete(l.groups[item.Block], id)
	if len(l.groups[item.Block]) == 0 {
		delete(l.groups, item.Block)
	}
	if err := l.store.Delete(id); err != nil {
		return nil, err
	}
	return item, nil
}

// setAndIndex assembles a limbo blob database entry and stores it, also updating
// the in-memory indices.
func (l *limbo) setAndIndex(ptx *blobTxForPool, block uint64) error {
	txhash := ptx.Tx.Hash()
	item := &limboBlob{
		TxHash: txhash,
		Block:  block,
		Ptx:    ptx,
	}
	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
	}
	l.index[txhash] = id
	if _, ok := l.groups[block]; !ok {
		l.groups[block] = make(map[uint64]common.Hash)
	}
	l.groups[block][id] = txhash
	return nil
}