Eager Batch Derivation (#2882)

* Eager Batch Derivation

Does the following:
- Split the batch queue into a batch queue & payload attributes queue
- Change batch derivation rules to enable eager batch derivation
- Eagerly create payload attributes

* Cleanup inclusion block

* Batch Queue: Add tests

Only for the batch queue & not the attributes queue.

* Fix lint + go mod tidy

* backport fixes from attributes-cleanup

* Update op-node/rollup/derive/batch_queue.go
Co-authored-by: Diederik Loerakker <proto@protolambda.com>

* address PR comments

* revert testlog

* use timestamp as RNG seed

* explain anti overflow check

* fix lint

* better logging

* only adjust maxL2Time based on minL2Time when epoch is not already started (#2939)
Co-authored-by: Diederik Loerakker <proto@protolambda.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

op-batcher/go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 // indirect
 	github.com/btcsuite/btcutil v1.0.3-0.20201208143702-a53e38424cce // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
-	github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/deckarep/golang-set v1.8.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
 	github.com/ethereum-optimism/optimism/op-bindings v0.0.0 // indirect
@@ -48,8 +48,8 @@ require (
 	github.com/tklauser/numcpus v0.4.0 // indirect
 	github.com/tyler-smith/go-bip39 v1.1.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
-	golang.org/x/crypto v0.0.0-20220307211146-efcb8507fb70 // indirect
-	golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c // indirect
+	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/sys v0.0.0-20220701225701-179beb0bd1a1 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 )

op-e2e/go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/libp2p/go-libp2p v0.18.1
 	github.com/libp2p/go-libp2p-core v0.15.0
 	github.com/miguelmota/go-ethereum-hdwallet v0.1.1
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.8.0
 )
 
 require (
@@ -24,8 +24,8 @@ require (
 	github.com/btcsuite/btcutil v1.0.3-0.20201208143702-a53e38424cce // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/containerd/cgroups v0.0.0-20201119153540-4cbc285b3327 // indirect
-	github.com/coreos/go-systemd/v22 v22.1.0 // indirect
-	github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
+	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
 	github.com/deckarep/golang-set v1.8.0 // indirect
@@ -38,7 +38,7 @@ require (
 	github.com/gballet/go-libpcsclite v0.0.0-20191108122812-4678299bea08 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-stack/stack v1.8.1 // indirect
-	github.com/godbus/dbus/v5 v5.0.3 // indirect
+	github.com/godbus/dbus/v5 v5.0.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.3.0 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
@@ -46,9 +46,9 @@ require (
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
-	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-bexpr v0.1.11 // indirect
-	github.com/hashicorp/go-multierror v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
 	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
 	github.com/holiman/uint256 v1.2.0 // indirect
@@ -103,7 +103,7 @@ require (
 	github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc // indirect
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 // indirect
 	github.com/minio/sha256-simd v1.0.0 // indirect
-	github.com/mitchellh/mapstructure v1.4.3 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/pointerstructure v1.2.1 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
 	github.com/multiformats/go-base32 v0.0.3 // indirect
@@ -144,20 +144,20 @@ require (
 	github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
-	go.uber.org/multierr v1.7.0 // indirect
-	go.uber.org/zap v1.19.1 // indirect
-	golang.org/x/crypto v0.0.0-20220307211146-efcb8507fb70 // indirect
-	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c // indirect
+	go.uber.org/multierr v1.8.0 // indirect
+	go.uber.org/zap v1.21.0 // indirect
+	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/net v0.0.0-20220520000938-2e3eb7b945c2 // indirect
+	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f // indirect
+	golang.org/x/sys v0.0.0-20220701225701-179beb0bd1a1 // indirect
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65 // indirect
-	google.golang.org/grpc v1.40.0 // indirect
+	google.golang.org/grpc v1.46.2 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 	gopkg.in/urfave/cli.v1 v1.20.0 // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
 replace (

op-node/go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/ethereum/go-ethereum v1.10.17
 	github.com/golang/snappy v0.0.4
 	github.com/google/go-cmp v0.5.8
-	github.com/hashicorp/go-multierror v1.0.0
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d
 	github.com/holiman/uint256 v1.2.0
 	github.com/ipfs/go-datastore v0.5.0
@@ -26,7 +26,7 @@ require (
 	github.com/multiformats/go-multiaddr v0.5.0
 	github.com/multiformats/go-multiaddr-dns v0.3.1
 	github.com/prometheus/client_golang v1.12.2
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.8.0
 	github.com/urfave/cli v1.22.5
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
 )
@@ -41,8 +41,8 @@ require (
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/cheekybits/genny v1.0.0 // indirect
 	github.com/containerd/cgroups v0.0.0-20201119153540-4cbc285b3327 // indirect
-	github.com/coreos/go-systemd/v22 v22.1.0 // indirect
-	github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
+	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
 	github.com/deckarep/golang-set v1.8.0 // indirect
@@ -52,19 +52,19 @@ require (
 	github.com/fjl/memsize v0.0.1 // indirect
 	github.com/flynn/noise v1.0.0 // indirect
 	github.com/francoispqt/gojay v1.2.13 // indirect
-	github.com/fsnotify/fsnotify v1.4.9 // indirect
+	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/gballet/go-libpcsclite v0.0.0-20191108122812-4678299bea08 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-stack/stack v1.8.1 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
-	github.com/godbus/dbus/v5 v5.0.3 // indirect
+	github.com/godbus/dbus/v5 v5.0.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.3.0 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
-	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-bexpr v0.1.11 // indirect
 	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
 	github.com/huin/goupnp v1.0.3 // indirect
@@ -78,6 +78,7 @@ require (
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/koron/go-ssdp v0.0.2 // indirect
+	github.com/kr/pretty v0.3.0 // indirect
 	github.com/libp2p/go-buffer-pool v0.0.2 // indirect
 	github.com/libp2p/go-cidranger v1.1.0 // indirect
 	github.com/libp2p/go-conn-security-multistream v0.3.0 // indirect
@@ -113,7 +114,7 @@ require (
 	github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc // indirect
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 // indirect
 	github.com/minio/sha256-simd v1.0.0 // indirect
-	github.com/mitchellh/mapstructure v1.4.3 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/pointerstructure v1.2.1 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
 	github.com/multiformats/go-base32 v0.0.3 // indirect
@@ -127,7 +128,7 @@ require (
 	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/onsi/ginkgo v1.16.4 // indirect
-	github.com/onsi/gomega v1.16.0 // indirect
+	github.com/onsi/gomega v1.19.0 // indirect
 	github.com/opencontainers/runtime-spec v1.0.2 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
@@ -140,12 +141,13 @@ require (
 	github.com/raulk/go-watchdog v1.2.0 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/rjeczalik/notify v0.9.2 // indirect
+	github.com/rogpeppe/go-internal v1.8.1 // indirect
 	github.com/rs/cors v1.8.2 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
 	github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 // indirect
 	github.com/status-im/keycard-go v0.0.0-20211109104530-b0e0482ba91d // indirect
-	github.com/stretchr/objx v0.2.0 // indirect
+	github.com/stretchr/objx v0.4.0 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect
 	github.com/tklauser/go-sysconf v0.3.10 // indirect
 	github.com/tklauser/numcpus v0.4.0 // indirect
@@ -154,22 +156,21 @@ require (
 	github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
-	go.uber.org/multierr v1.7.0 // indirect
-	go.uber.org/zap v1.19.1 // indirect
-	golang.org/x/crypto v0.0.0-20220307211146-efcb8507fb70 // indirect
-	golang.org/x/mod v0.6.0-dev.0.20211013180041-c96bc1413d57 // indirect
-	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c // indirect
+	go.uber.org/multierr v1.8.0 // indirect
+	go.uber.org/zap v1.21.0 // indirect
+	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
+	golang.org/x/net v0.0.0-20220520000938-2e3eb7b945c2 // indirect
+	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f // indirect
+	golang.org/x/sys v0.0.0-20220701225701-179beb0bd1a1 // indirect
 	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65 // indirect
-	golang.org/x/tools v0.1.8-0.20211029000441-d6a9af8af023 // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	google.golang.org/grpc v1.40.0 // indirect
+	golang.org/x/tools v0.1.11 // indirect
+	google.golang.org/grpc v1.46.2 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/urfave/cli.v1 v1.20.0 // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
 replace github.com/ethereum-optimism/optimism/op-bindings v0.0.0 => ../op-bindings

op-node/rollup/derive/attributes_queue.go

+package derive
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type L1ReceiptsFetcher interface {
+	Fetch(ctx context.Context, blockHash common.Hash) (eth.L1Info, types.Transactions, types.Receipts, error)
+}
+
+type AttributesQueueOutput interface {
+	AddSafeAttributes(attributes *eth.PayloadAttributes)
+	SafeL2Head() eth.L2BlockRef
+	StageProgress
+}
+
+type AttributesQueue struct {
+	log      log.Logger
+	config   *rollup.Config
+	dl       L1ReceiptsFetcher
+	next     AttributesQueueOutput
+	progress Progress
+	batches  []*BatchData
+}
+
+func NewAttributesQueue(log log.Logger, cfg *rollup.Config, l1Fetcher L1ReceiptsFetcher, next AttributesQueueOutput) *AttributesQueue {
+	return &AttributesQueue{
+		log:    log,
+		config: cfg,
+		dl:     l1Fetcher,
+		next:   next,
+	}
+}
+
+func (aq *AttributesQueue) AddBatch(batch *BatchData) {
+	aq.log.Info("Received next batch", "batch_epoch", batch.EpochNum, "batch_timestamp", batch.Timestamp, "tx_count", len(batch.Transactions))
+	aq.batches = append(aq.batches, batch)
+}
+
+func (aq *AttributesQueue) Progress() Progress {
+	return aq.progress
+}
+
+func (aq *AttributesQueue) Step(ctx context.Context, outer Progress) error {
+	if changed, err := aq.progress.Update(outer); err != nil || changed {
+		return err
+	}
+	attr, err := aq.DeriveL2Inputs(ctx, aq.next.SafeL2Head())
+	if err != nil {
+		return err
+	}
+	aq.next.AddSafeAttributes(attr)
+	return nil
+}
+
+func (aq *AttributesQueue) ResetStep(ctx context.Context, l1Fetcher L1Fetcher) error {
+	aq.batches = aq.batches[:0]
+	aq.progress = aq.next.Progress()
+	return io.EOF
+}
+
+func (aq *AttributesQueue) SafeL2Head() eth.L2BlockRef {
+	return aq.next.SafeL2Head()
+}
+
+// DeriveL2Inputs turns the next L2 batch into an Payload Attributes that builds off of the safe head
+func (aq *AttributesQueue) DeriveL2Inputs(ctx context.Context, l2SafeHead eth.L2BlockRef) (*eth.PayloadAttributes, error) {
+	if len(aq.batches) == 0 {
+		return nil, io.EOF
+	}
+	batch := aq.batches[0]
+
+	seqNumber := l2SafeHead.SequenceNumber + 1
+	// Check if we need to advance an epoch & update local state
+	if l2SafeHead.L1Origin != batch.Epoch() {
+		aq.log.Info("advancing epoch in the attributes queue", "l2SafeHead", l2SafeHead, "l2SafeHead_origin", l2SafeHead.L1Origin, "batch_timestamp", batch.Timestamp, "batch_epoch", batch.Epoch())
+		seqNumber = 0
+	}
+
+	fetchCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
+	l1Info, _, receipts, err := aq.dl.Fetch(fetchCtx, batch.EpochHash)
+	if err != nil {
+		aq.log.Error("failed to fetch L1 block info", "l1Origin", batch.Epoch(), "err", err)
+		return nil, err
+	}
+
+	// Fill in deposits if we are the first block of the epoch
+	var deposits []hexutil.Bytes
+	if seqNumber == 0 {
+		var errs []error
+		deposits, errs = DeriveDeposits(receipts, aq.config.DepositContractAddress)
+		for _, err := range errs {
+			aq.log.Error("Failed to derive a deposit", "l1Origin", batch.Epoch(), "err", err)
+		}
+		if len(errs) != 0 {
+			// TODO: Multierror here
+			return nil, fmt.Errorf("failed to derive some deposits: %v", errs)
+		}
+	}
+
+	var txns []eth.Data
+	l1InfoTx, err := L1InfoDepositBytes(seqNumber, l1Info)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create l1InfoTx: %w", err)
+	}
+	txns = append(txns, l1InfoTx)
+	if seqNumber == 0 {
+		txns = append(txns, deposits...)
+	}
+	txns = append(txns, batch.Transactions...)
+	attrs := &eth.PayloadAttributes{
+		Timestamp:             hexutil.Uint64(batch.Timestamp),
+		PrevRandao:            eth.Bytes32(l1Info.MixDigest()),
+		SuggestedFeeRecipient: aq.config.FeeRecipientAddress,
+		Transactions:          txns,
+		// we are verifying, not sequencing, we've got all transactions and do not pull from the tx-pool
+		// (that would make the block derivation non-deterministic)
+		NoTxPool: true,
+	}
+	aq.log.Info("generated attributes in payload queue", "tx_count", len(txns), "timestamp", batch.Timestamp)
+
+	// Slice off the batch once we are guaranteed to succeed
+	aq.batches = aq.batches[1:]
+	return attrs, nil
+}

op-node/rollup/derive/attributes_queue_test.go

+package derive

op-node/rollup/derive/batch_queue_test.go

+package derive
+
+import (
+	"context"
+	"io"
+	"math/rand"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum-optimism/optimism/op-node/testutils"
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeBatchQueueOutput fakes the next stage (receive only) for the batch queue
+// It tracks the progress state of the next stage.
+// Upon receiving a batch, relevant characteristic of safeL2Head are immediately advanced.
+type fakeBatchQueueOutput struct {
+	progress   Progress
+	batches    []*BatchData
+	safeL2Head eth.L2BlockRef
+}
+
+var _ BatchQueueOutput = (*fakeBatchQueueOutput)(nil)
+
+func (f *fakeBatchQueueOutput) AddBatch(batch *BatchData) {
+	f.batches = append(f.batches, batch)
+	// Advance SafeL2Head
+	f.safeL2Head.Time = batch.Timestamp
+	f.safeL2Head.L1Origin.Number = uint64(batch.EpochNum)
+}
+
+func (f *fakeBatchQueueOutput) SafeL2Head() eth.L2BlockRef {
+	return f.safeL2Head
+}
+
+func (f *fakeBatchQueueOutput) Progress() Progress {
+	return f.progress
+}
+
+func b(timestamp uint64, epoch eth.L1BlockRef) *BatchData {
+	rng := rand.New(rand.NewSource(int64(timestamp)))
+	data := testutils.RandomData(rng, 20)
+	return &BatchData{BatchV1{
+		Timestamp:    timestamp,
+		EpochNum:     rollup.Epoch(epoch.Number),
+		EpochHash:    epoch.Hash,
+		Transactions: []hexutil.Bytes{data},
+	}}
+}
+
+func L1Chain(l1Times []uint64) []eth.L1BlockRef {
+	var out []eth.L1BlockRef
+	var parentHash [32]byte
+	for i, time := range l1Times {
+		hash := [32]byte{byte(i)}
+		out = append(out, eth.L1BlockRef{
+			Hash:       hash,
+			Number:     uint64(i),
+			ParentHash: parentHash,
+			Time:       time,
+		})
+		parentHash = hash
+	}
+	return out
+}
+
+type fakeL1Fetcher struct {
+	l1 []eth.L1BlockRef
+}
+
+func (f *fakeL1Fetcher) L1BlockRefByNumber(_ context.Context, n uint64) (eth.L1BlockRef, error) {
+	if n >= uint64(len(f.l1)) {
+		return eth.L1BlockRef{}, ethereum.NotFound
+	}
+	return f.l1[int(n)], nil
+}
+
+func TestBatchQueueEager(t *testing.T) {
+	log := testlog.Logger(t, log.LvlTrace)
+	next := &fakeBatchQueueOutput{
+		safeL2Head: eth.L2BlockRef{
+			Number:   0,
+			Time:     10,
+			L1Origin: eth.BlockID{Number: 0},
+		},
+	}
+	cfg := &rollup.Config{
+		Genesis: rollup.Genesis{
+			L2Time: 10,
+		},
+		BlockTime:         2,
+		MaxSequencerDrift: 600,
+		SeqWindowSize:     30,
+	}
+
+	l1 := L1Chain([]uint64{10, 20, 30})
+
+	fetcher := fakeL1Fetcher{l1: l1}
+	bq := NewBatchQueue(log, cfg, &fetcher, next)
+
+	prevProgress := Progress{
+		Origin: l1[0],
+		Closed: false,
+	}
+
+	// Setup progress
+	bq.progress.Closed = true
+	err := bq.Step(context.Background(), prevProgress)
+	require.Nil(t, err)
+
+	// Add batches
+	batches := []*BatchData{b(12, l1[0]), b(14, l1[0])}
+	for _, batch := range batches {
+		err := bq.AddBatch(batch)
+		require.Nil(t, err)
+	}
+	// Step
+	for {
+		if err := bq.Step(context.Background(), prevProgress); err == io.EOF {
+			break
+		} else {
+			require.Nil(t, err)
+		}
+	}
+	// Verify Output
+	require.Equal(t, batches, next.batches)
+}
+
+func TestBatchQueueFull(t *testing.T) {
+	log := testlog.Logger(t, log.LvlTrace)
+	next := &fakeBatchQueueOutput{
+		safeL2Head: eth.L2BlockRef{
+			Number:   0,
+			Time:     10,
+			L1Origin: eth.BlockID{Number: 0},
+		},
+	}
+	cfg := &rollup.Config{
+		Genesis: rollup.Genesis{
+			L2Time: 10,
+		},
+		BlockTime:         2,
+		MaxSequencerDrift: 600,
+		SeqWindowSize:     2,
+	}
+
+	l1 := L1Chain([]uint64{10, 15, 20})
+
+	fetcher := fakeL1Fetcher{l1: l1}
+	bq := NewBatchQueue(log, cfg, &fetcher, next)
+
+	// Start with open previous & closed self.
+	// Then this stage is opened at the first step.
+	bq.progress.Closed = true
+	prevProgress := Progress{
+		Origin: l1[0],
+		Closed: false,
+	}
+
+	// Do the bq open
+	err := bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, false)
+
+	// Add batches
+	batches := []*BatchData{b(14, l1[0]), b(16, l1[0]), b(18, l1[1])}
+	for _, batch := range batches {
+		err := bq.AddBatch(batch)
+		require.Nil(t, err)
+	}
+	// Missing first batch
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, io.EOF)
+
+	// Close previous to close bq
+	prevProgress.Closed = true
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, true)
+
+	// Open previous to open bq with the new inclusion block
+	prevProgress.Closed = false
+	prevProgress.Origin = l1[1]
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, false)
+
+	// Close previous to close bq (for epoch 2)
+	prevProgress.Closed = true
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, true)
+
+	// Open previous to open bq with the new inclusion block (epoch 2)
+	prevProgress.Closed = false
+	prevProgress.Origin = l1[2]
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, false)
+
+	// Finally add batch
+	firstBatch := b(12, l1[0])
+	err = bq.AddBatch(firstBatch)
+	require.Equal(t, err, nil)
+
+	// Close the origin
+	prevProgress.Closed = true
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, true)
+
+	// Step, but should have full epoch now
+	for {
+		if err := bq.Step(context.Background(), prevProgress); err == io.EOF {
+			break
+		} else {
+			require.Nil(t, err)
+		}
+	}
+	// Verify Output
+	var final []*BatchData
+	final = append(final, firstBatch)
+	final = append(final, batches...)
+	require.Equal(t, final, next.batches)
+}
+
+func TestBatchQueueMissing(t *testing.T) {
+	log := testlog.Logger(t, log.LvlTrace)
+	next := &fakeBatchQueueOutput{
+		safeL2Head: eth.L2BlockRef{
+			Number:   0,
+			Time:     10,
+			L1Origin: eth.BlockID{Number: 0},
+		},
+	}
+	cfg := &rollup.Config{
+		Genesis: rollup.Genesis{
+			L2Time: 10,
+		},
+		BlockTime:         2,
+		MaxSequencerDrift: 600,
+		SeqWindowSize:     2,
+	}
+
+	l1 := L1Chain([]uint64{10, 15, 20})
+
+	fetcher := fakeL1Fetcher{l1: l1}
+	bq := NewBatchQueue(log, cfg, &fetcher, next)
+
+	// Start with open previous & closed self.
+	// Then this stage is opened at the first step.
+	bq.progress.Closed = true
+	prevProgress := Progress{
+		Origin: l1[0],
+		Closed: false,
+	}
+
+	// Do the bq open
+	err := bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, false)
+
+	// Add batches
+	// NB: The batch at 18 is skipped to skip over the ability to
+	// do eager batch processing for that batch. This test checks
+	// that batch timestamp 12 & 14 is created & 16 is used.
+	batches := []*BatchData{b(16, l1[0]), b(20, l1[1])}
+	for _, batch := range batches {
+		err := bq.AddBatch(batch)
+		require.Nil(t, err)
+	}
+	// Missing first batch
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, io.EOF)
+
+	// Close previous to close bq
+	prevProgress.Closed = true
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, true)
+
+	// Open previous to open bq with the new inclusion block
+	prevProgress.Closed = false
+	prevProgress.Origin = l1[1]
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, false)
+
+	// Close previous to close bq (for epoch 2)
+	prevProgress.Closed = true
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, true)
+
+	// Open previous to open bq with the new inclusion block (epoch 2)
+	prevProgress.Closed = false
+	prevProgress.Origin = l1[2]
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, false)
+
+	// Close the origin
+	prevProgress.Closed = true
+	err = bq.Step(context.Background(), prevProgress)
+	require.Equal(t, err, nil)
+	require.Equal(t, bq.progress.Closed, true)
+
+	// Step, but should have full epoch now + fill missing
+	for {
+		if err := bq.Step(context.Background(), prevProgress); err == io.EOF {
+			break
+		} else {
+			require.Nil(t, err)
+		}
+	}
+	// TODO: Maybe check actuall batch validity better
+	require.Equal(t, 3, len(next.batches))
+}

op-node/rollup/derive/batches.go

@@ -80,7 +80,7 @@ func FillMissingBatches(batches []*BatchData, epoch eth.BlockID, blockTime, minL
 		newHeadL2Timestamp = nextL1Time - 1
 	}
 	for _, b := range batches {
-		m[b.BatchV1.Timestamp] = b
+		m[b.Timestamp] = b
 		if b.Timestamp > newHeadL2Timestamp {
 			newHeadL2Timestamp = b.Timestamp
 		}
@@ -91,14 +91,16 @@ func FillMissingBatches(batches []*BatchData, epoch eth.BlockID, blockTime, minL
 		if ok {
 			out = append(out, b)
 		} else {
-			out = append(out, &BatchData{
-				BatchV1{
-					EpochNum:  rollup.Epoch(epoch.Number),
-					EpochHash: epoch.Hash,
-					Timestamp: t,
-				},
-			})
+			out = append(out,
+				&BatchData{
+					BatchV1{
+						EpochNum:  rollup.Epoch(epoch.Number),
+						EpochHash: epoch.Hash,
+						Timestamp: t,
+					},
+				})
 		}
+
 	}
 	return out
 }

op-node/rollup/derive/engine_queue.go

@@ -49,7 +49,7 @@ type EngineQueue struct {
 	engine Engine
 }
 
-var _ BatchQueueOutput = (*EngineQueue)(nil)
+var _ AttributesQueueOutput = (*EngineQueue)(nil)
 
 // NewEngineQueue creates a new EngineQueue, which should be Reset(origin) before use.
 func NewEngineQueue(log log.Logger, cfg *rollup.Config, engine Engine) *EngineQueue {
@@ -72,6 +72,7 @@ func (eq *EngineQueue) AddUnsafePayload(payload *eth.ExecutionPayload) {
 }
 
 func (eq *EngineQueue) AddSafeAttributes(attributes *eth.PayloadAttributes) {
+	eq.log.Trace("received next safe attributes")
 	eq.safeAttributes = append(eq.safeAttributes, attributes)
 }
 

op-node/rollup/derive/l1_retrieval.go

@@ -65,7 +65,7 @@ func (l1r *L1Retrieval) Step(ctx context.Context, outer Progress) error {
 	if l1r.datas == nil {
 		datas, err := l1r.dataSrc.OpenData(ctx, l1r.progress.Origin.ID())
 		if err != nil {
-			l1r.log.Error("can't fetch L1 data", "origin", l1r.progress.Origin)
+			l1r.log.Error("can't fetch L1 data", "origin", l1r.progress.Origin, "err", err)
 			return nil
 		}
 		l1r.datas = datas

op-node/rollup/derive/pipeline.go

@@ -75,13 +75,14 @@ type DerivationPipeline struct {
 // NewDerivationPipeline creates a derivation pipeline, which should be reset before use.
 func NewDerivationPipeline(log log.Logger, cfg *rollup.Config, l1Fetcher L1Fetcher, engine Engine) *DerivationPipeline {
 	eng := NewEngineQueue(log, cfg, engine)
-	batchQueue := NewBatchQueue(log, cfg, l1Fetcher, eng)
+	attributesQueue := NewAttributesQueue(log, cfg, l1Fetcher, eng)
+	batchQueue := NewBatchQueue(log, cfg, l1Fetcher, attributesQueue)
 	chInReader := NewChannelInReader(log, batchQueue)
 	bank := NewChannelBank(log, cfg, chInReader)
 	dataSrc := NewCalldataSource(log, cfg, l1Fetcher)
 	l1Src := NewL1Retrieval(log, dataSrc, bank)
 	l1Traversal := NewL1Traversal(log, l1Fetcher, l1Src)
-	stages := []Stage{eng, batchQueue, chInReader, bank, l1Src, l1Traversal}
+	stages := []Stage{eng, attributesQueue, batchQueue, chInReader, bank, l1Src, l1Traversal}
 
 	return &DerivationPipeline{
 		log:       log,

op-node/rollup/derive/progress.go

@@ -23,7 +23,7 @@ func (pr *Progress) Update(outer Progress) (changed bool, err error) {
 	}
 	if pr.Closed {
 		if outer.Closed {
-			if pr.Origin != outer.Origin {
+			if pr.Origin.ID() != outer.Origin.ID() {
 				return true, fmt.Errorf("outer stage changed origin from %s to %s without opening it", pr.Origin, outer.Origin)
 			}
 			return false, nil
@@ -36,7 +36,7 @@ func (pr *Progress) Update(outer Progress) (changed bool, err error) {
 			return true, nil
 		}
 	} else {
-		if pr.Origin != outer.Origin {
+		if pr.Origin.ID() != outer.Origin.ID() {
 			return true, fmt.Errorf("outer stage changed origin from %s to %s before closing it", pr.Origin, outer.Origin)
 		}
 		if outer.Closed {

op-node/testlog/testlog.go

@@ -135,23 +135,17 @@ func (l *logger) SetHandler(h log.Handler) {
 	l.l.SetHandler(h)
 }
 
-// tracks the largest seen decoration length, to make alignment between all test loggers consistent.
-var logDecorationLength = 12
-
 // flush writes all buffered messages and clears the buffer.
 func (l *logger) flush() {
 	l.t.Helper()
 	// 2 frame skip for flush() + public logger fn
 	decorationLen := estimateInfoLen(2)
-	if decorationLen > 30 { // limit to a maximum size, to avoid huge padding
-		decorationLen = 30
-	}
-	// logDecorationLength is only increasing, should be safe even with inaccurate concurrent use.
-	if decorationLen > logDecorationLength { // increase padding if we encounter larger decoration
-		logDecorationLength = decorationLen
+	padding := 20
+	if decorationLen <= 25 {
+		padding = 25 - decorationLen
 	}
 	for _, r := range l.h.buf {
-		l.t.Logf("%*s%s", logDecorationLength-decorationLen, "", l.h.fmt.Format(r))
+		l.t.Logf("%*s%s", padding, "", l.h.fmt.Format(r))
 	}
 	l.h.buf = nil
 }

op-proposer/go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/ethereum-optimism/optimism/op-node v0.0.0
 	github.com/ethereum/go-ethereum v1.10.17
 	github.com/miguelmota/go-ethereum-hdwallet v0.1.1
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.8.0
 	github.com/urfave/cli v1.22.5
 )
 
@@ -19,7 +19,7 @@ require (
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 // indirect
 	github.com/btcsuite/btcutil v1.0.3-0.20201208143702-a53e38424cce // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
-	github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/deckarep/golang-set v1.8.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
@@ -53,11 +53,11 @@ require (
 	github.com/tklauser/numcpus v0.4.0 // indirect
 	github.com/tyler-smith/go-bip39 v1.1.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
-	golang.org/x/crypto v0.0.0-20220307211146-efcb8507fb70 // indirect
-	golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c // indirect
+	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/sys v0.0.0-20220701225701-179beb0bd1a1 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
 replace github.com/ethereum-optimism/optimism/op-node v0.0.0 => ../op-node