diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 414541b6ef..9b282ed961 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -191,8 +191,9 @@ func (ptx *pooledBlobTx) convert() (*types.Transaction, error) { } // `removeParity` removes parity cells of cellSidecar -// This is needed to reduce the size of transaction -// Otherwise, pooledBlobTx will about 2 times bigger than the original tx +// But in fact, the result of this is just a split of blobs +// e.g. Blob = [a, b, c] -> Cells = [a], [b], [c] +// Why do we need to apply EC then? func (ptx *pooledBlobTx) removeParity() error { sc := ptx.Sidecar if sc == nil { @@ -478,9 +479,13 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser } // Index all transactions on disk and delete anything unprocessable var fails []uint64 + var convertTxs []*types.Transaction index := func(id uint64, size uint32, blob []byte) { - if p.parseTransaction(id, size, blob) != nil { + if tx, err := p.parseTransaction(id, size, blob); err != nil { fails = append(fails, id) + } else if tx != nil { + fails = append(fails, id) + convertTxs = append(convertTxs, tx) } } store, err := billy.Open(billy.Options{Path: queuedir, Repair: true}, slotter, index) @@ -489,6 +494,40 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser } p.store = store + if len(convertTxs) > 0 { + log.Trace("Convert plain transaction to pooled transaction", "len", len(convertTxs)) + for _, tx := range convertTxs { + // Note that we skip errors here + // Just like parseTransaction failure does not abort the blobpool creation, + // conversion process also cannot abort the entire process. + pooledTx, err := newPooledBlobTx(tx) + if err != nil { + continue + } + if err := pooledTx.removeParity(); err != nil { + return err + } + blob, err := rlp.EncodeToBytes(pooledTx) + if err != nil { + continue + } + id, err := p.store.Put(blob) + if err != nil { + continue + } + meta := newBlobTxMeta(id, pooledTx.Size, p.store.Size(id), pooledTx) + + sender, err := types.Sender(p.signer, pooledTx.Transaction) + if err != nil { + fails = append(fails, id) + continue + } + if err := p.trackTransaction(meta, sender); err != nil { + fails = append(fails, id) + continue + } + } + } if len(fails) > 0 { log.Warn("Dropping invalidated blob transactions", "ids", fails) dropInvalidMeter.Mark(int64(len(fails))) @@ -561,43 +600,50 @@ func (p *BlobPool) Close() error { // parseTransaction is a callback method on pool creation that gets called for // each transaction on disk to create the in-memory metadata index. -func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error { +// return: +// - types.Transaction : plain tx if the type of 'blob' is plain tx +func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) (*types.Transaction, error) { tx := new(types.Transaction) - pooledTx := new(pooledBlobTx) + if err := rlp.DecodeBytes(blob, pooledTx); 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. if err := rlp.DecodeBytes(blob, tx); err != nil { log.Error("Failed to decode blob pool entry", "id", id, "err", err) - return err + return nil, err } if tx.BlobTxSidecar() == nil { log.Error("Missing sidecar in blob pool entry", "id", id, "hash", tx.Hash()) - return errors.New("missing blob sidecar") - } - pooledTx, err = newPooledBlobTx(tx) - if err != nil { - return err + return nil, errors.New("missing blob sidecar") } + return tx, nil + } + if pooledTx.Sidecar == nil { + log.Error("Missing sidecar in blob pool entry", "id", id, "hash", tx.Hash()) + return nil, errors.New("missing blob sidecar") } meta := newBlobTxMeta(id, pooledTx.Size, size, pooledTx) - if p.lookup.exists(meta.hash) { - // This path is only possible after a crash, where deleted items are not - // removed via the normal shutdown-startup procedure and thus may get - // partially resurrected. - log.Error("Rejecting duplicate blob pool entry", "id", id, "hash", tx.Hash()) - return errors.New("duplicate blob entry") - } sender, err := types.Sender(p.signer, pooledTx.Transaction) if err != nil { // This path is impossible unless the signature validity changes across // restarts. For that ever improbable case, recover gracefully by ignoring // this data entry. log.Error("Failed to recover blob tx sender", "id", id, "hash", tx.Hash(), "err", err) - return err + return nil, err + } + return nil, p.trackTransaction(meta, sender) +} + +func (p *BlobPool) trackTransaction(meta *blobTxMeta, sender common.Address) error { + if p.lookup.exists(meta.hash) { + // This path is only possible after a crash, where deleted items are not + // removed via the normal shutdown-startup procedure and thus may get + // partially resurrected. + log.Error("Rejecting duplicate blob pool entry", "id", meta.id, "hash", meta.hash) + return fmt.Errorf("duplicate blob entry %d, %s", meta.id, meta.hash) } if _, ok := p.index[sender]; !ok { if err := p.reserver.Hold(sender); err != nil { @@ -1356,30 +1402,10 @@ func (p *BlobPool) getRLP(hash common.Hash) []byte { } data, err := p.store.Get(id) if err != nil { - log.Error("Failed to get transaction in blobpool", "hash", hash, "id", id, "err", err) + log.Error("Tracked blob transaction missing from store", "hash", hash, "id", id, "err", err) return nil } - tx := new(types.Transaction) - pooledTx := new(pooledBlobTx) - if err := rlp.DecodeBytes(data, pooledTx); err != nil { - if err := rlp.DecodeBytes(data, tx); err != nil { - log.Error("Failed to decode transaction in blobpool", "hash", hash, "id", id, "err", err) - return nil - } - return data - } - tx, err = pooledTx.convert() - if err != nil { - log.Error("Failed to convert transaction in blobpool", "hash", hash, "id", id, "err", err) - return nil - } - encoded, err := rlp.EncodeToBytes(tx) - if err != nil { - log.Error("Failed to encode transaction in blobpool", "hash", hash, "id", id, "err", err) - return nil - } - - return encoded + return data } // Get returns a transaction if it is contained in the pool, or nil otherwise. @@ -1388,28 +1414,43 @@ func (p *BlobPool) Get(hash common.Hash) *types.Transaction { if len(data) == 0 { return nil } - tx := new(types.Transaction) pooledTx := new(pooledBlobTx) if err := rlp.DecodeBytes(data, pooledTx); err != nil { - if err := rlp.DecodeBytes(data, tx); err != nil { - id, _ := p.lookup.storeidOfTx(hash) + id, _ := p.lookup.storeidOfTx(hash) - log.Error("Blobs corrupted for traced transaction", - "hash", hash, "id", id, "err", err) - return nil - } + log.Error("Blobs corrupted for traced transaction", + "hash", hash, "id", id, "err", err) + return nil + } + if tx, err := pooledTx.convert(); err != nil { + return nil + } else { return tx } - tx, err := pooledTx.convert() - if err == nil { - return tx - } - return nil } // GetRLP returns a RLP-encoded transaction if it is contained in the pool. func (p *BlobPool) GetRLP(hash common.Hash) []byte { - return p.getRLP(hash) + data := p.getRLP(hash) + pooledTx := new(pooledBlobTx) + if err := rlp.DecodeBytes(data, pooledTx); err != nil { + id, _ := p.lookup.storeidOfTx(hash) + + log.Error("Blobs corrupted for traced transaction", + "hash", hash, "id", id, "err", err) + return nil + } + tx, err := pooledTx.convert() + if err != nil { + log.Error("Failed to convert pooledTx", "err", err) + return nil + } + blob, err := rlp.EncodeToBytes(tx) + if err != nil { + log.Error("Failed to encode", "err", err) + return nil + } + return blob } // GetMetadata returns the transaction type and transaction size with the @@ -1476,10 +1517,8 @@ func (p *BlobPool) GetBlobs(vhashes []common.Hash, version byte) ([]*kzg4844.Blo tx := new(types.Transaction) pooledTx := new(pooledBlobTx) if err := rlp.DecodeBytes(data, pooledTx); err != nil { - if err := rlp.DecodeBytes(data, tx); err != nil { - log.Error("Blobs corrupted for traced transaction", "id", txID, "err", err) - continue - } + log.Error("Blobs corrupted for traced transaction", "id", txID, "err", err) + continue } else { tx, err = pooledTx.convert() if err != nil { diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index 1b50781e70..ab877e09fd 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/core/txpool/blobpool/blobpool_test.go @@ -488,7 +488,7 @@ func verifyBlobRetrievals(t *testing.T, pool *BlobPool) { // - 8. Fully duplicate transactions (matching hash) must be dropped // - 9. Duplicate nonces from the same account must be dropped func TestOpenDrops(t *testing.T) { - //log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelTrace, true))) + // log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelTrace, true))) // Create a temporary folder for the persistent backend storage := t.TempDir() @@ -515,75 +515,76 @@ func TestOpenDrops(t *testing.T) { S: new(uint256.Int), }) blob, _ := rlp.EncodeToBytes(tx) - badsig, _ := store.Put(blob) + badsig := tx.Hash() + store.Put(blob) // Insert a sequence of transactions with a nonce gap in between to verify // that anything gapped will get evicted (case 3). var ( gapper, _ = crypto.GenerateKey() - valids = make(map[uint64]struct{}) - gapped = make(map[uint64]struct{}) + valids = make(map[common.Hash]struct{}) + gapped = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{0, 1, 3, 4, 6, 7} { // first gap at #2, another at #5 tx := makeTx(nonce, 1, 1, 1, gapper) blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) + store.Put(blob) if nonce < 2 { - valids[id] = struct{}{} + valids[tx.Hash()] = struct{}{} } else { - gapped[id] = struct{}{} + gapped[tx.Hash()] = struct{}{} } } // Insert a sequence of transactions with a gapped starting nonce to verify // that the entire set will get dropped (case 3). var ( dangler, _ = crypto.GenerateKey() - dangling = make(map[uint64]struct{}) + dangling = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{1, 2, 3} { // first gap at #0, all set dangling tx := makeTx(nonce, 1, 1, 1, dangler) blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) - dangling[id] = struct{}{} + store.Put(blob) + dangling[tx.Hash()] = struct{}{} } // Insert a sequence of transactions with already passed nonces to veirfy // that the entire set will get dropped (case 4). var ( filler, _ = crypto.GenerateKey() - filled = make(map[uint64]struct{}) + filled = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{0, 1, 2} { // account nonce at 3, all set filled tx := makeTx(nonce, 1, 1, 1, filler) blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) - filled[id] = struct{}{} + store.Put(blob) + filled[tx.Hash()] = struct{}{} } // Insert a sequence of transactions with partially passed nonces to verify // that the included part of the set will get dropped (case 4). var ( overlapper, _ = crypto.GenerateKey() - overlapped = make(map[uint64]struct{}) + overlapped = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{0, 1, 2, 3} { // account nonce at 2, half filled tx := makeTx(nonce, 1, 1, 1, overlapper) blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) + store.Put(blob) if nonce >= 2 { - valids[id] = struct{}{} + valids[tx.Hash()] = struct{}{} } else { - overlapped[id] = struct{}{} + overlapped[tx.Hash()] = struct{}{} } } // Insert a sequence of transactions with an underpriced first to verify that // the entire set will get dropped (case 5). var ( underpayer, _ = crypto.GenerateKey() - underpaid = make(map[uint64]struct{}) + underpaid = make(map[common.Hash]struct{}) ) for i := 0; i < 5; i++ { // make #0 underpriced var tx *types.Transaction @@ -594,15 +595,15 @@ func TestOpenDrops(t *testing.T) { } blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) - underpaid[id] = struct{}{} + store.Put(blob) + underpaid[tx.Hash()] = struct{}{} } // Insert a sequence of transactions with an underpriced in between to verify // that it and anything newly gapped will get evicted (case 5). var ( outpricer, _ = crypto.GenerateKey() - outpriced = make(map[uint64]struct{}) + outpriced = make(map[common.Hash]struct{}) ) for i := 0; i < 5; i++ { // make #2 underpriced var tx *types.Transaction @@ -613,18 +614,18 @@ func TestOpenDrops(t *testing.T) { } blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) + store.Put(blob) if i < 2 { - valids[id] = struct{}{} + valids[tx.Hash()] = struct{}{} } else { - outpriced[id] = struct{}{} + outpriced[tx.Hash()] = struct{}{} } } // Insert a sequence of transactions fully overdrafted to verify that the // entire set will get invalidated (case 6). var ( exceeder, _ = crypto.GenerateKey() - exceeded = make(map[uint64]struct{}) + exceeded = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{0, 1, 2} { // nonce 0 overdrafts the account var tx *types.Transaction @@ -635,14 +636,14 @@ func TestOpenDrops(t *testing.T) { } blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) - exceeded[id] = struct{}{} + store.Put(blob) + exceeded[tx.Hash()] = struct{}{} } // Insert a sequence of transactions partially overdrafted to verify that part // of the set will get invalidated (case 6). var ( overdrafter, _ = crypto.GenerateKey() - overdrafted = make(map[uint64]struct{}) + overdrafted = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{0, 1, 2} { // nonce 1 overdrafts the account var tx *types.Transaction @@ -653,44 +654,46 @@ func TestOpenDrops(t *testing.T) { } blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) + store.Put(blob) if nonce < 1 { - valids[id] = struct{}{} + valids[tx.Hash()] = struct{}{} } else { - overdrafted[id] = struct{}{} + overdrafted[tx.Hash()] = struct{}{} } } // Insert a sequence of transactions overflowing the account cap to verify // that part of the set will get invalidated (case 7). var ( overcapper, _ = crypto.GenerateKey() - overcapped = make(map[uint64]struct{}) + overcapped = make(map[common.Hash]struct{}) ) for nonce := uint64(0); nonce < maxTxsPerAccount+3; nonce++ { - blob, _ := rlp.EncodeToBytes(makeTx(nonce, 1, 1, 1, overcapper)) + tx := makeTx(nonce, 1, 1, 1, overcapper) + blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) + store.Put(blob) if nonce < maxTxsPerAccount { - valids[id] = struct{}{} + valids[tx.Hash()] = struct{}{} } else { - overcapped[id] = struct{}{} + overcapped[tx.Hash()] = struct{}{} } } // Insert a batch of duplicated transactions to verify that only one of each // version will remain (case 8). var ( duplicater, _ = crypto.GenerateKey() - duplicated = make(map[uint64]struct{}) + duplicated = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{0, 1, 2} { - blob, _ := rlp.EncodeToBytes(makeTx(nonce, 1, 1, 1, duplicater)) + tx := makeTx(nonce, 1, 1, 1, duplicater) + blob, _ := rlp.EncodeToBytes(tx) for i := 0; i < int(nonce)+1; i++ { - id, _ := store.Put(blob) + store.Put(blob) if i == 0 { - valids[id] = struct{}{} + valids[tx.Hash()] = struct{}{} } else { - duplicated[id] = struct{}{} + duplicated[tx.Hash()] = struct{}{} } } } @@ -698,17 +701,18 @@ func TestOpenDrops(t *testing.T) { // remain (case 9). var ( repeater, _ = crypto.GenerateKey() - repeated = make(map[uint64]struct{}) + repeated = make(map[common.Hash]struct{}) ) for _, nonce := range []uint64{0, 1, 2} { for i := 0; i < int(nonce)+1; i++ { - blob, _ := rlp.EncodeToBytes(makeTx(nonce, 1, uint64(i)+1 /* unique hashes */, 1, repeater)) + tx := makeTx(nonce, 1, uint64(i)+1 /* unique hashes */, 1, repeater) + blob, _ := rlp.EncodeToBytes(tx) - id, _ := store.Put(blob) + store.Put(blob) if i == 0 { - valids[id] = struct{}{} + valids[tx.Hash()] = struct{}{} } else { - repeated[id] = struct{}{} + repeated[tx.Hash()] = struct{}{} } } } @@ -745,39 +749,41 @@ func TestOpenDrops(t *testing.T) { // Verify that the malformed (case 1), badly signed (case 2) and gapped (case // 3) txs have been deleted from the pool - alive := make(map[uint64]struct{}) + alive := make(map[common.Hash]struct{}) for _, txs := range pool.index { for _, tx := range txs { switch tx.id { case malformed: t.Errorf("malformed RLP transaction remained in storage") - case badsig: - t.Errorf("invalidly signed transaction remained in storage") default: - if _, ok := dangling[tx.id]; ok { + if badsig == tx.hash { + t.Errorf("invalidly signed transaction remained in storage") + } + if _, ok := dangling[tx.hash]; ok { t.Errorf("dangling transaction remained in storage: %d", tx.id) - } else if _, ok := filled[tx.id]; ok { + } else if _, ok := filled[tx.hash]; ok { t.Errorf("filled transaction remained in storage: %d", tx.id) - } else if _, ok := overlapped[tx.id]; ok { + } else if _, ok := overlapped[tx.hash]; ok { t.Errorf("overlapped transaction remained in storage: %d", tx.id) - } else if _, ok := gapped[tx.id]; ok { + } else if _, ok := gapped[tx.hash]; ok { t.Errorf("gapped transaction remained in storage: %d", tx.id) - } else if _, ok := underpaid[tx.id]; ok { + } else if _, ok := underpaid[tx.hash]; ok { t.Errorf("underpaid transaction remained in storage: %d", tx.id) - } else if _, ok := outpriced[tx.id]; ok { + } else if _, ok := outpriced[tx.hash]; ok { t.Errorf("outpriced transaction remained in storage: %d", tx.id) - } else if _, ok := exceeded[tx.id]; ok { + } else if _, ok := exceeded[tx.hash]; ok { t.Errorf("fully overdrafted transaction remained in storage: %d", tx.id) - } else if _, ok := overdrafted[tx.id]; ok { + } else if _, ok := overdrafted[tx.hash]; ok { t.Errorf("partially overdrafted transaction remained in storage: %d", tx.id) - } else if _, ok := overcapped[tx.id]; ok { + } else if _, ok := overcapped[tx.hash]; ok { t.Errorf("overcapped transaction remained in storage: %d", tx.id) - } else if _, ok := duplicated[tx.id]; ok { - t.Errorf("duplicated transaction remained in storage: %d", tx.id) - } else if _, ok := repeated[tx.id]; ok { + } else if _, ok := repeated[tx.hash]; ok { t.Errorf("repeated nonce transaction remained in storage: %d", tx.id) } else { - alive[tx.id] = struct{}{} + if _, ok := alive[tx.hash]; ok { + t.Errorf("duplicated transaction remained in storage: %d", tx.id) + } + alive[tx.hash] = struct{}{} } } } @@ -786,14 +792,14 @@ func TestOpenDrops(t *testing.T) { if len(alive) != len(valids) { t.Errorf("valid transaction count mismatch: have %d, want %d", len(alive), len(valids)) } - for id := range alive { - if _, ok := valids[id]; !ok { - t.Errorf("extra transaction %d", id) + for hash := range alive { + if _, ok := valids[hash]; !ok { + t.Errorf("extra transaction %s", hash) } } - for id := range valids { - if _, ok := alive[id]; !ok { - t.Errorf("missing transaction %d", id) + for hash := range valids { + if _, ok := alive[hash]; !ok { + t.Errorf("missing transaction %s", hash) } } // Verify all the calculated pool internals. Interestingly, this is **not**