Merge branch 'develop' into felipe/dont-cache-tx-methods

docs/op-stack/src/docs/understand/explainer.md

@@ -198,13 +198,15 @@ The ability to pause the bridge in case of emergency means that in the worst cas
 
 #### Unfreezing the bridge via L1 soft fork
 
-In order to address the frozen funds, there is a potential final recovery mechanism we call the “L1 Soft Fork Upgrade Recovery” mechanism. This mechanism enables L1 to initiate a bridge upgrade with a soft fork, bypassing all other permissions within the Superchain bridge contracts. The mechanism is as follows:
+In order to address the frozen funds, there is a potential final recovery mechanism which has been discussed by the L2 community which we call the “L1 Soft Fork Upgrade Recovery” mechanism. This mechanism enables L1 to initiate a bridge upgrade with a soft fork, bypassing all other permissions within the Superchain bridge contracts. This approach may [introduce systemic risk](https://vitalik.ca/general/2023/05/21/dont_overload.html) to Ethereum and requires research and community buy-in before implementation. It is not required for implementing the Superchain and is being documented for research completeness. Without further research into the implications and safety, it is not an approach the team currently endorses.
 
-*Anyone* may propose an upgrade by submitting a transaction to a special bridge contract, along with a very large bond. This begins a two week challenge period. During this challenge period, *anyone* may submit a challenge which immediately *cancels* the upgrade and claims the bond. Under normal circumstances, it is impossible that an upgrade would go uncancelled for the required two weeks due to the large incentive provided for anyone to cancel the upgrade. However, if the upgrade is accompanied by a modification to Ethereum L1 validator software (the L1 soft fork), which ignores blocks that contain the cancellation transaction then it may succeed.
+The mechanism is as follows:
 
-While a successful upgrade of this type would represent a soft fork of Ethereum L1, it would not incur long term technical debt to the Ethereum codebase because the soft fork logic can be removed once the upgrade has completed. 
+*Anyone* may propose an upgrade by submitting a transaction to a special bridge contract, along with a very large bond. This begins a two week challenge period. During this challenge period, anyone may submit a challenge which immediately cancels the upgrade and claims the bond. Under normal circumstances, it is impossible that an upgrade would go uncancelled for the required two weeks due to the large incentive provided for anyone to cancel the upgrade. However, if the upgrade is accompanied by a modification to Ethereum L1 validator software (the L1 soft fork), which ignores blocks that contain the cancellation transaction then it may succeed.
 
-We expect this escape hatch will never be used, but its very existence should deter malicious behavior.
+While a successful upgrade of this type would represent a soft fork of Ethereum L1, it would not incur long term technical debt to the Ethereum codebase because the soft fork logic can be removed once the upgrade has completed.
+
+We expect this escape hatch will never be used, but its very existence could deter malicious behavior.
 
 ### The combination of these features results in a system satisfying the core Superchain properties
 

go.mod

@@ -24,6 +24,7 @@ require (
 	github.com/libp2p/go-libp2p-pubsub v0.9.0
 	github.com/libp2p/go-libp2p-testing v0.12.0
 	github.com/mattn/go-isatty v0.0.17
+	github.com/multiformats/go-base32 v0.1.0
 	github.com/multiformats/go-multiaddr v0.8.0
 	github.com/multiformats/go-multiaddr-dns v0.3.1
 	github.com/olekukonko/tablewriter v0.0.5
@@ -128,7 +129,6 @@ require (
 	github.com/moby/term v0.0.0-20221105221325-4eb28fa6025c // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
-	github.com/multiformats/go-base32 v0.1.0 // indirect
 	github.com/multiformats/go-base36 v0.2.0 // indirect
 	github.com/multiformats/go-multiaddr-fmt v0.1.0 // indirect
 	github.com/multiformats/go-multibase v0.1.1 // indirect

op-challenger/challenger/oracle.go

+package challenger
+
+import (
+	"errors"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+)
+
+var ErrMissingEvent = errors.New("missing event")
+
+// BuildOutputLogFilter creates a filter query for the L2OutputOracle contract.
+//
+// The `OutputProposed` event is encoded as:
+// 0: bytes32 indexed outputRoot,
+// 1: uint256 indexed l2OutputIndex,
+// 2: uint256 indexed l2BlockNumber,
+// 3: uint256 l1Timestamp
+func BuildOutputLogFilter(l2ooABI *abi.ABI) (ethereum.FilterQuery, error) {
+	// Get the L2OutputOracle contract `OutputProposed` event
+	event := l2ooABI.Events["OutputProposed"]
+
+	// Sanity check that the `OutputProposed` event is defined
+	if event.ID == (common.Hash{}) {
+		return ethereum.FilterQuery{}, ErrMissingEvent
+	}
+
+	query := ethereum.FilterQuery{
+		Topics: [][]common.Hash{
+			{event.ID},
+		},
+	}
+
+	return query, nil
+}

op-challenger/challenger/oracle_test.go

+package challenger
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	eth "github.com/ethereum/go-ethereum"
+	abi "github.com/ethereum/go-ethereum/accounts/abi"
+	common "github.com/ethereum/go-ethereum/common"
+)
+
+// TestBuildOutputLogFilter_Succeeds tests that the Output
+// Log Filter is built correctly.
+func TestBuildOutputLogFilter_Succeeds(t *testing.T) {
+	// Create a mock event id
+	event := abi.Event{
+		ID: [32]byte{0x01},
+	}
+
+	filterQuery := eth.FilterQuery{
+		Topics: [][]common.Hash{
+			{event.ID},
+		},
+	}
+
+	// Mock the ABI
+	l2ooABI := abi.ABI{
+		Events: map[string]abi.Event{
+			"OutputProposed": event,
+		},
+	}
+
+	// Build the filter
+	query, err := BuildOutputLogFilter(&l2ooABI)
+	require.Equal(t, filterQuery, query)
+	require.NoError(t, err)
+}
+
+// TestBuildOutputLogFilter_Fails tests that the Output
+// Log Filter fails when the event definition is missing.
+func TestBuildOutputLogFilter_Fails(t *testing.T) {
+	// Mock the ABI
+	l2ooABI := abi.ABI{
+		Events: map[string]abi.Event{},
+	}
+
+	// Build the filter
+	_, err := BuildOutputLogFilter(&l2ooABI)
+	require.Error(t, err)
+	require.Equal(t, ErrMissingEvent, err)
+}

op-node/p2p/host.go

@@ -7,6 +7,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	libp2p "github.com/libp2p/go-libp2p"
 	lconf "github.com/libp2p/go-libp2p/config"
 	"github.com/libp2p/go-libp2p/core/connmgr"
@@ -132,11 +135,16 @@ func (conf *Config) Host(log log.Logger, reporter metrics.Reporter) (host.Host,
 		return nil, fmt.Errorf("failed to derive pubkey from network priv key: %w", err)
 	}
 
-	ps, err := pstoreds.NewPeerstore(context.Background(), conf.Store, pstoreds.DefaultOpts())
+	basePs, err := pstoreds.NewPeerstore(context.Background(), conf.Store, pstoreds.DefaultOpts())
 	if err != nil {
 		return nil, fmt.Errorf("failed to open peerstore: %w", err)
 	}
 
+	ps, err := store.NewExtendedPeerstore(context.Background(), log, clock.SystemClock, basePs, conf.Store)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open extended peerstore: %w", err)
+	}
+
 	if err := ps.AddPrivKey(pid, conf.Priv); err != nil {
 		return nil, fmt.Errorf("failed to set up peerstore with priv key: %w", err)
 	}

op-node/p2p/mocks/ConnectionGater.go

@@ -142,20 +142,6 @@ func (_m *ConnectionGater) InterceptUpgraded(_a0 network.Conn) (bool, control.Di
 	return r0, r1
 }
 
-// IsBlocked provides a mock function with given fields: p
-func (_m *ConnectionGater) IsBlocked(p peer.ID) bool {
-	ret := _m.Called(p)
-
-	var r0 bool
-	if rf, ok := ret.Get(0).(func(peer.ID) bool); ok {
-		r0 = rf(p)
-	} else {
-		r0 = ret.Get(0).(bool)
-	}
-
-	return r0
-}
-
 // ListBlockedAddrs provides a mock function with given fields:
 func (_m *ConnectionGater) ListBlockedAddrs() []net.IP {
 	ret := _m.Called()

op-node/p2p/mocks/Peerstore.go

@@ -6,6 +6,8 @@ import (
 	mock "github.com/stretchr/testify/mock"
 
 	peer "github.com/libp2p/go-libp2p/core/peer"
+
+	store "github.com/ethereum-optimism/optimism/op-node/p2p/store"
 )
 
 // Peerstore is an autogenerated mock type for the Peerstore type
@@ -43,6 +45,20 @@ func (_m *Peerstore) Peers() peer.IDSlice {
 	return r0
 }
 
+// SetScore provides a mock function with given fields: _a0, _a1, _a2
+func (_m *Peerstore) SetScore(_a0 peer.ID, _a1 store.ScoreType, _a2 float64) error {
+	ret := _m.Called(_a0, _a1, _a2)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(peer.ID, store.ScoreType, float64) error); ok {
+		r0 = rf(_a0, _a1, _a2)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 type mockConstructorTestingTNewPeerstore interface {
 	mock.TestingT
 	Cleanup(func())

op-node/p2p/peer_scorer.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	log "github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	peer "github.com/libp2p/go-libp2p/core/peer"
@@ -91,6 +92,8 @@ type Peerstore interface {
 
 	// Peers returns all of the peer IDs stored across all inner stores.
 	Peers() peer.IDSlice
+
+	SetScore(peer.ID, store.ScoreType, float64) error
 }
 
 // Scorer is a peer scorer that scores peers based on application-specific metrics.
@@ -123,6 +126,12 @@ func (s *scorer) SnapshotHook() pubsub.ExtendedPeerScoreInspectFn {
 		}
 		// Now set the new scores.
 		for id, snap := range m {
+			scores := make(map[store.ScoreType]float64)
+			scores[store.TypeGossip] = snap.Score
+
+			if err := s.peerStore.SetScore(id, store.TypeGossip, snap.Score); err != nil {
+				s.log.Warn("Unable to update peer gossip score", "err", err)
+			}
 			band := s.bandScoreThresholds.Bucket(snap.Score)
 			scoreMap[band] += 1
 			s.gater.Update(id, snap.Score)

op-node/p2p/peer_scorer_test.go

@@ -5,6 +5,7 @@ import (
 
 	p2p "github.com/ethereum-optimism/optimism/op-node/p2p"
 	p2pMocks "github.com/ethereum-optimism/optimism/op-node/p2p/mocks"
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	"github.com/ethereum-optimism/optimism/op-node/testlog"
 	log "github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -78,6 +79,9 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 	// Mock the peer gater call
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-100)).Return(nil).Once()
 
+	// Expect updating the peer store
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), store.TypeGossip, float64(-100)).Return(nil).Once()
+
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
 		"friend":   0,
@@ -94,6 +98,8 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 
 	// Change the peer score now to a different band
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(0)).Return(nil).Once()
+	// Expect updating the peer store
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), store.TypeGossip, float64(0)).Return(nil).Once()
 
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
@@ -124,6 +130,8 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHookBlocksPeer() {
 
 	// Mock the peer gater call
 	testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-101)).Return(nil)
+	// Expect updating the peer store
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), store.TypeGossip, float64(-101)).Return(nil).Once()
 
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{

op-node/p2p/peer_scores.go

 package p2p
 
 import (
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	log "github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	host "github.com/libp2p/go-libp2p/core/host"
@@ -14,8 +15,13 @@ func ConfigurePeerScoring(h host.Host, g ConnectionGater, gossipConf GossipSetup
 	peerScoreThresholds := NewPeerScoreThresholds()
 	banEnabled := gossipConf.BanPeers()
 	peerGater := NewPeerGater(g, log, banEnabled)
-	scorer := NewScorer(peerGater, h.Peerstore(), m, gossipConf.PeerBandScorer(), log)
 	opts := []pubsub.Option{}
+	eps, ok := h.Peerstore().(store.ExtendedPeerstore)
+	if !ok {
+		log.Warn("Disabling peer scoring. Peerstore does not support peer scores")
+		return opts
+	}
+	scorer := NewScorer(peerGater, eps, m, gossipConf.PeerBandScorer(), log)
 	// Check the app specific score since libp2p doesn't export it's [validate] function :/
 	if peerScoreParams != nil && peerScoreParams.AppSpecificScore != nil {
 		opts = []pubsub.Option{

op-node/p2p/peer_scores_test.go

@@ -7,11 +7,18 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+
 	p2p "github.com/ethereum-optimism/optimism/op-node/p2p"
 	p2pMocks "github.com/ethereum-optimism/optimism/op-node/p2p/mocks"
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	testlog "github.com/ethereum-optimism/optimism/op-node/testlog"
-
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/sync"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 	suite "github.com/stretchr/testify/suite"
 
 	log "github.com/ethereum/go-ethereum/log"
@@ -50,11 +57,29 @@ func TestPeerScores(t *testing.T) {
 	suite.Run(t, new(PeerScoresTestSuite))
 }
 
+type customPeerstoreNetwork struct {
+	network.Network
+	ps peerstore.Peerstore
+}
+
+func (c *customPeerstoreNetwork) Peerstore() peerstore.Peerstore {
+	return c.ps
+}
+
+func (c *customPeerstoreNetwork) Close() error {
+	_ = c.ps.Close()
+	return c.Network.Close()
+}
+
 // getNetHosts generates a slice of hosts using the [libp2p/go-libp2p] library.
 func getNetHosts(testSuite *PeerScoresTestSuite, ctx context.Context, n int) []host.Host {
 	var out []host.Host
+	log := testlog.Logger(testSuite.T(), log.LvlError)
 	for i := 0; i < n; i++ {
-		netw := tswarm.GenSwarm(testSuite.T())
+		swarm := tswarm.GenSwarm(testSuite.T())
+		eps, err := store.NewExtendedPeerstore(ctx, log, clock.SystemClock, swarm.Peerstore(), sync.MutexWrap(ds.NewMapDatastore()))
+		netw := &customPeerstoreNetwork{swarm, eps}
+		require.NoError(testSuite.T(), err)
 		h := bhost.NewBlankHost(netw)
 		testSuite.T().Cleanup(func() { h.Close() })
 		out = append(out, h)

op-node/p2p/rpc_api.go

@@ -5,6 +5,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 
@@ -28,6 +29,8 @@ type PeerInfo struct {
 	Latency       time.Duration         `json:"latency"`
 
 	GossipBlocks bool `json:"gossipBlocks"` // if the peer is in our gossip topic
+
+	PeerScores store.PeerScores `json:"scores"`
 }
 
 type PeerDump struct {

op-node/p2p/rpc_server.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	decredSecp "github.com/decred/dcrd/dcrec/secp256k1/v4"
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p-testing/netutil"
 	"github.com/libp2p/go-libp2p/core/connmgr"
@@ -108,6 +109,11 @@ func dumpPeer(id peer.ID, nw network.Network, pstore peerstore.Peerstore, connMg
 			info.NodeID = enode.PubkeyToIDV4((*decredSecp.PublicKey)(typedPub).ToECDSA())
 		}
 	}
+	if eps, ok := pstore.(store.ExtendedPeerstore); ok {
+		if dat, err := eps.GetPeerScores(id); err == nil {
+			info.PeerScores = dat
+		}
+	}
 	if dat, err := pstore.Get(id, "ProtocolVersion"); err == nil {
 		protocolVersion, ok := dat.(string)
 		if ok {

op-node/p2p/store/extended.go

+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+	"github.com/ethereum/go-ethereum/log"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/libp2p/go-libp2p/core/peerstore"
+)
+
+type extendedStore struct {
+	peerstore.Peerstore
+	peerstore.CertifiedAddrBook
+	*scoreBook
+}
+
+func NewExtendedPeerstore(ctx context.Context, logger log.Logger, clock clock.Clock, ps peerstore.Peerstore, store ds.Batching) (ExtendedPeerstore, error) {
+	cab, ok := peerstore.GetCertifiedAddrBook(ps)
+	if !ok {
+		return nil, errors.New("peerstore should also be a certified address book")
+	}
+	sb, err := newScoreBook(ctx, logger, clock, store)
+	if err != nil {
+		return nil, fmt.Errorf("create scorebook: %w", err)
+	}
+	sb.startGC()
+	return &extendedStore{
+		Peerstore:         ps,
+		CertifiedAddrBook: cab,
+		scoreBook:         sb,
+	}, nil
+}
+
+func (s *extendedStore) Close() error {
+	s.scoreBook.Close()
+	return s.Peerstore.Close()
+}
+
+var _ ExtendedPeerstore = (*extendedStore)(nil)

op-node/p2p/store/gc.go

+package store
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+const (
+	gcPeriod = 2 * time.Hour
+)
+
+type gcAction func() error
+
+func startGc(ctx context.Context, logger log.Logger, clock clock.Clock, bgTasks *sync.WaitGroup, action gcAction) {
+	bgTasks.Add(1)
+	go func() {
+		defer bgTasks.Done()
+
+		gcTimer := clock.NewTicker(gcPeriod)
+		defer gcTimer.Stop()
+
+		for {
+			select {
+			case <-gcTimer.Ch():
+				if err := action(); err != nil {
+					logger.Warn("GC failed", "err", err)
+				}
+
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+}

op-node/p2p/store/gc_test.go

+package store
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestScheduleGcPeriodically(t *testing.T) {
+	var bgTasks sync.WaitGroup
+	ctx, cancel := context.WithCancel(context.Background())
+	defer func() {
+		cancel()
+		// Wait for the gc background process to complete after cancelling the context
+		bgTasks.Wait()
+	}()
+	logger := testlog.Logger(t, log.LvlInfo)
+	clock := clock.NewDeterministicClock(time.UnixMilli(5000))
+
+	called := make(chan struct{}, 10)
+	action := func() error {
+		called <- struct{}{}
+		return nil
+	}
+	waitForGc := func(failMsg string) {
+		timeout, cancel := context.WithTimeout(ctx, 10*time.Second)
+		defer cancel()
+		select {
+		case <-timeout.Done():
+			t.Fatal(failMsg)
+		case <-called:
+			require.Len(t, called, 0, "should only run once after gc period")
+		}
+	}
+	startGc(ctx, logger, clock, &bgTasks, action)
+	timeout, tCancel := context.WithTimeout(ctx, 10*time.Second)
+	defer tCancel()
+	require.True(t, clock.WaitForNewPendingTask(timeout), "did not schedule pending GC")
+
+	require.Len(t, called, 0, "should not run immediately")
+
+	clock.AdvanceTime(gcPeriod)
+	waitForGc("should run gc after first time period")
+
+	clock.AdvanceTime(gcPeriod)
+	waitForGc("should run gc again after second time period")
+}

op-node/p2p/store/iface.go

+package store
+
+import (
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/peerstore"
+)
+
+type PeerScores struct {
+	Gossip float64 `json:"gossip"`
+}
+
+type ScoreType int
+
+const (
+	TypeGossip ScoreType = iota
+)
+
+// ScoreDatastore defines a type-safe API for getting and setting libp2p peer score information
+type ScoreDatastore interface {
+	// GetPeerScores returns the current scores for the specified peer
+	GetPeerScores(id peer.ID) (PeerScores, error)
+
+	// SetScore stores the latest score for the specified peer and score type
+	SetScore(id peer.ID, scoreType ScoreType, score float64) error
+}
+
+// ExtendedPeerstore defines a type-safe API to work with additional peer metadata based on a libp2p peerstore.Peerstore
+type ExtendedPeerstore interface {
+	peerstore.Peerstore
+	ScoreDatastore
+	peerstore.CertifiedAddrBook
+}

op-node/p2p/store/scorebook.go

+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+	"github.com/ethereum/go-ethereum/log"
+	lru "github.com/hashicorp/golang-lru/v2"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/multiformats/go-base32"
+)
+
+const (
+	scoreDataV0       = "0"
+	scoreCacheSize    = 100
+	expiryPeriod      = 24 * time.Hour
+	maxPruneBatchSize = 20
+)
+
+var scoresBase = ds.NewKey("/peers/scores")
+
+type scoreRecord struct {
+	PeerScores
+	lastUpdate time.Time
+}
+
+type scoreBook struct {
+	ctx      context.Context
+	cancelFn context.CancelFunc
+	clock    clock.Clock
+	log      log.Logger
+	bgTasks  sync.WaitGroup
+	store    ds.Batching
+	cache    *lru.Cache[peer.ID, scoreRecord]
+	sync.RWMutex
+}
+
+func newScoreBook(ctx context.Context, logger log.Logger, clock clock.Clock, store ds.Batching) (*scoreBook, error) {
+	cache, err := lru.New[peer.ID, scoreRecord](scoreCacheSize)
+	if err != nil {
+		return nil, fmt.Errorf("creating cache: %w", err)
+	}
+
+	ctx, cancelFn := context.WithCancel(ctx)
+	book := scoreBook{
+		ctx:      ctx,
+		cancelFn: cancelFn,
+		clock:    clock,
+		log:      logger,
+		store:    store,
+		cache:    cache,
+	}
+	return &book, nil
+}
+
+func (d *scoreBook) startGC() {
+	startGc(d.ctx, d.log, d.clock, &d.bgTasks, d.prune)
+}
+
+func (d *scoreBook) GetPeerScores(id peer.ID) (PeerScores, error) {
+	d.RLock()
+	defer d.RUnlock()
+	record, err := d.getRecord(id)
+	if err != nil {
+		return PeerScores{}, nil
+	}
+	return record.PeerScores, err
+}
+
+func (d *scoreBook) getRecord(id peer.ID) (scoreRecord, error) {
+	if scores, ok := d.cache.Get(id); ok {
+		return scores, nil
+	}
+	data, err := d.store.Get(d.ctx, scoreKey(id))
+	if errors.Is(err, ds.ErrNotFound) {
+		return scoreRecord{}, nil
+	} else if err != nil {
+		return scoreRecord{}, fmt.Errorf("load scores for peer %v: %w", id, err)
+	}
+	record, err := deserializeScoresV0(data)
+	if err != nil {
+		return scoreRecord{}, fmt.Errorf("invalid score data for peer %v: %w", id, err)
+	}
+	d.cache.Add(id, record)
+	return record, nil
+}
+
+func (d *scoreBook) SetScore(id peer.ID, scoreType ScoreType, score float64) error {
+	d.Lock()
+	defer d.Unlock()
+	scores, err := d.getRecord(id)
+	if err != nil {
+		return err
+	}
+	scores.lastUpdate = d.clock.Now()
+	scores.Gossip = score
+	switch scoreType {
+	case TypeGossip:
+		scores.Gossip = score
+	default:
+		return fmt.Errorf("unknown score type: %v", scoreType)
+	}
+	data, err := serializeScoresV0(scores)
+	if err != nil {
+		return fmt.Errorf("encode scores for peer %v: %w", id, err)
+	}
+	err = d.store.Put(d.ctx, scoreKey(id), data)
+	if err != nil {
+		return fmt.Errorf("storing updated scores for peer %v: %w", id, err)
+	}
+	d.cache.Add(id, scores)
+	return nil
+}
+
+// prune deletes entries from the store that are older than expiryPeriod.
+// Note that the expiry period is not a strict TTL. Entries that are eligible for deletion may still be present
+// either because the prune function hasn't yet run or because they are still preserved in the in-memory cache after
+// having been deleted from the database.
+func (d *scoreBook) prune() error {
+	results, err := d.store.Query(d.ctx, query.Query{
+		Prefix: scoresBase.String(),
+	})
+	if err != nil {
+		return err
+	}
+	pending := 0
+	batch, err := d.store.Batch(d.ctx)
+	if err != nil {
+		return err
+	}
+	for result := range results.Next() {
+		// Bail out if the context is done
+		select {
+		case <-d.ctx.Done():
+			return d.ctx.Err()
+		default:
+		}
+		record, err := deserializeScoresV0(result.Value)
+		if err != nil {
+			return err
+		}
+		if record.lastUpdate.Add(expiryPeriod).Before(d.clock.Now()) {
+			if pending > maxPruneBatchSize {
+				if err := batch.Commit(d.ctx); err != nil {
+					return err
+				}
+				batch, err = d.store.Batch(d.ctx)
+				if err != nil {
+					return err
+				}
+				pending = 0
+			}
+			pending++
+			if err := batch.Delete(d.ctx, ds.NewKey(result.Key)); err != nil {
+				return err
+			}
+		}
+	}
+	if err := batch.Commit(d.ctx); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (d *scoreBook) Close() {
+	d.cancelFn()
+	d.bgTasks.Wait()
+}
+
+func scoreKey(id peer.ID) ds.Key {
+	return scoresBase.ChildString(base32.RawStdEncoding.EncodeToString([]byte(id))).ChildString(scoreDataV0)
+}

op-node/p2p/store/scorebook_test.go

+package store
+
+import (
+	"context"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/clock"
+	"github.com/ethereum/go-ethereum/log"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/sync"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/p2p/host/peerstore/pstoreds"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetEmptyScoreComponents(t *testing.T) {
+	id := peer.ID("aaaa")
+	store := createMemoryStore(t)
+	assertPeerScores(t, store, id, PeerScores{})
+}
+
+func TestRoundTripGossipScore(t *testing.T) {
+	id := peer.ID("aaaa")
+	store := createMemoryStore(t)
+	score := 123.45
+	err := store.SetScore(id, TypeGossip, score)
+	require.NoError(t, err)
+
+	assertPeerScores(t, store, id, PeerScores{Gossip: score})
+}
+
+func TestUpdateGossipScore(t *testing.T) {
+	id := peer.ID("aaaa")
+	store := createMemoryStore(t)
+	score := 123.45
+	require.NoError(t, store.SetScore(id, TypeGossip, 444.223))
+	require.NoError(t, store.SetScore(id, TypeGossip, score))
+
+	assertPeerScores(t, store, id, PeerScores{Gossip: score})
+}
+
+func TestStoreScoresForMultiplePeers(t *testing.T) {
+	id1 := peer.ID("aaaa")
+	id2 := peer.ID("bbbb")
+	store := createMemoryStore(t)
+	score1 := 123.45
+	score2 := 453.22
+	require.NoError(t, store.SetScore(id1, TypeGossip, score1))
+	require.NoError(t, store.SetScore(id2, TypeGossip, score2))
+
+	assertPeerScores(t, store, id1, PeerScores{Gossip: score1})
+	assertPeerScores(t, store, id2, PeerScores{Gossip: score2})
+}
+
+func TestPersistData(t *testing.T) {
+	id := peer.ID("aaaa")
+	score := 123.45
+	backingStore := sync.MutexWrap(ds.NewMapDatastore())
+	store := createPeerstoreWithBacking(t, backingStore)
+
+	require.NoError(t, store.SetScore(id, TypeGossip, score))
+
+	// Close and recreate a new store from the same backing
+	require.NoError(t, store.Close())
+	store = createPeerstoreWithBacking(t, backingStore)
+
+	assertPeerScores(t, store, id, PeerScores{Gossip: score})
+}
+
+func TestUnknownScoreType(t *testing.T) {
+	store := createMemoryStore(t)
+	err := store.SetScore("aaaa", 92832, 244.24)
+	require.ErrorContains(t, err, "unknown score type")
+}
+
+func TestCloseCompletes(t *testing.T) {
+	store := createMemoryStore(t)
+	require.NoError(t, store.Close())
+}
+
+func TestPrune(t *testing.T) {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	logger := testlog.Logger(t, log.LvlInfo)
+	store := sync.MutexWrap(ds.NewMapDatastore())
+	clock := clock.NewDeterministicClock(time.UnixMilli(1000))
+	book, err := newScoreBook(ctx, logger, clock, store)
+	require.NoError(t, err)
+
+	hasScoreRecorded := func(id peer.ID) bool {
+		scores, err := book.GetPeerScores(id)
+		require.NoError(t, err)
+		return scores != PeerScores{}
+	}
+
+	firstStore := clock.Now()
+	// Set some scores all 30 minutes apart so they have different expiry times
+	require.NoError(t, book.SetScore("aaaa", TypeGossip, 123.45))
+	clock.AdvanceTime(30 * time.Minute)
+	require.NoError(t, book.SetScore("bbbb", TypeGossip, 123.45))
+	clock.AdvanceTime(30 * time.Minute)
+	require.NoError(t, book.SetScore("cccc", TypeGossip, 123.45))
+	clock.AdvanceTime(30 * time.Minute)
+	require.NoError(t, book.SetScore("dddd", TypeGossip, 123.45))
+	clock.AdvanceTime(30 * time.Minute)
+
+	// Update bbbb again which should extend its expiry
+	require.NoError(t, book.SetScore("bbbb", TypeGossip, 123.45))
+
+	require.True(t, hasScoreRecorded("aaaa"))
+	require.True(t, hasScoreRecorded("bbbb"))
+	require.True(t, hasScoreRecorded("cccc"))
+	require.True(t, hasScoreRecorded("dddd"))
+
+	elapsedTime := clock.Now().Sub(firstStore)
+	timeToFirstExpiry := expiryPeriod - elapsedTime
+	// Advance time until the score for aaaa should be pruned.
+	clock.AdvanceTime(timeToFirstExpiry + 1)
+	require.NoError(t, book.prune())
+	// Clear the cache so reads have to come from the database
+	book.cache.Purge()
+	require.False(t, hasScoreRecorded("aaaa"), "should have pruned aaaa record")
+
+	// Advance time so cccc, dddd and the original bbbb entry should be pruned
+	clock.AdvanceTime(90 * time.Minute)
+	require.NoError(t, book.prune())
+	// Clear the cache so reads have to come from the database
+	book.cache.Purge()
+
+	require.False(t, hasScoreRecorded("cccc"), "should have pruned cccc record")
+	require.False(t, hasScoreRecorded("dddd"), "should have pruned cccc record")
+
+	require.True(t, hasScoreRecorded("bbbb"), "should not prune bbbb record")
+}
+
+func TestPruneMultipleBatches(t *testing.T) {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	logger := testlog.Logger(t, log.LvlInfo)
+	clock := clock.NewDeterministicClock(time.UnixMilli(1000))
+	book, err := newScoreBook(ctx, logger, clock, sync.MutexWrap(ds.NewMapDatastore()))
+	require.NoError(t, err)
+
+	hasScoreRecorded := func(id peer.ID) bool {
+		scores, err := book.GetPeerScores(id)
+		require.NoError(t, err)
+		return scores != PeerScores{}
+	}
+
+	// Set scores for more peers than the max batch size
+	peerCount := maxPruneBatchSize*3 + 5
+	for i := 0; i < peerCount; i++ {
+		require.NoError(t, book.SetScore(peer.ID(strconv.Itoa(i)), TypeGossip, 123.45))
+	}
+	clock.AdvanceTime(expiryPeriod + 1)
+	require.NoError(t, book.prune())
+	// Clear the cache so reads have to come from the database
+	book.cache.Purge()
+
+	for i := 0; i < peerCount; i++ {
+		require.Falsef(t, hasScoreRecorded(peer.ID(strconv.Itoa(i))), "Should prune record peer %v", i)
+	}
+}
+
+func assertPeerScores(t *testing.T, store ExtendedPeerstore, id peer.ID, expected PeerScores) {
+	result, err := store.GetPeerScores(id)
+	require.NoError(t, err)
+	require.Equal(t, result, expected)
+}
+
+func createMemoryStore(t *testing.T) ExtendedPeerstore {
+	store := sync.MutexWrap(ds.NewMapDatastore())
+	return createPeerstoreWithBacking(t, store)
+}
+
+func createPeerstoreWithBacking(t *testing.T, store *sync.MutexDatastore) ExtendedPeerstore {
+	ps, err := pstoreds.NewPeerstore(context.Background(), store, pstoreds.DefaultOpts())
+	require.NoError(t, err, "Failed to create peerstore")
+	logger := testlog.Logger(t, log.LvlInfo)
+	clock := clock.NewDeterministicClock(time.UnixMilli(100))
+	eps, err := NewExtendedPeerstore(context.Background(), logger, clock, ps, store)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = eps.Close()
+	})
+	return eps
+}

op-node/p2p/store/serialize.go

+package store
+
+import (
+	"bytes"
+	"encoding/binary"
+	"time"
+)
+
+func serializeScoresV0(scores scoreRecord) ([]byte, error) {
+	var b bytes.Buffer
+	err := binary.Write(&b, binary.BigEndian, scores.lastUpdate.UnixMilli())
+	if err != nil {
+		return nil, err
+	}
+	err = binary.Write(&b, binary.BigEndian, scores.Gossip)
+	if err != nil {
+		return nil, err
+	}
+	return b.Bytes(), nil
+}
+
+func deserializeScoresV0(data []byte) (scoreRecord, error) {
+	var scores scoreRecord
+	r := bytes.NewReader(data)
+	var lastUpdate int64
+	err := binary.Read(r, binary.BigEndian, &lastUpdate)
+	if err != nil {
+		return scoreRecord{}, err
+	}
+	scores.lastUpdate = time.UnixMilli(lastUpdate)
+	err = binary.Read(r, binary.BigEndian, &scores.Gossip)
+	if err != nil {
+		return scoreRecord{}, err
+	}
+	return scores, nil
+}

op-node/p2p/store/serialize_test.go

+package store
+
+import (
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRoundtripScoresV0(t *testing.T) {
+	scores := scoreRecord{
+		PeerScores: PeerScores{Gossip: 1234.52382},
+		lastUpdate: time.UnixMilli(1923841),
+	}
+	data, err := serializeScoresV0(scores)
+	require.NoError(t, err)
+
+	result, err := deserializeScoresV0(data)
+	require.NoError(t, err)
+	require.Equal(t, scores, result)
+}
+
+// TestParseHistoricSerializations checks that existing data can still be deserialized
+// Adding new fields should not require bumping the version, only removing fields
+// A new entry should be added to this test each time any fields are changed to ensure it can always be deserialized
+func TestParseHistoricSerializationsV0(t *testing.T) {
+	tests := []struct {
+		data     []byte
+		expected scoreRecord
+	}{
+		{
+			data: common.Hex2Bytes("00000000001D5B0140934A18644523F6"),
+			expected: scoreRecord{
+				PeerScores: PeerScores{Gossip: 1234.52382},
+				lastUpdate: time.UnixMilli(1923841),
+			},
+		},
+	}
+	for idx, test := range tests {
+		test := test
+		t.Run(strconv.Itoa(idx), func(t *testing.T) {
+			result, err := deserializeScoresV0(test.data)
+			require.NoError(t, err)
+			require.Equal(t, test.expected, result)
+		})
+	}
+}

op-service/clock/clock.go

+// Package clock provides an abstraction for time to enable testing of functionality that uses time as an input.
+package clock
+
+import "time"
+
+// Clock represents time in a way that can be provided by varying implementations.
+// Methods are designed to be direct replacements for methods in the time package.
+type Clock interface {
+	// Now provides the current local time. Equivalent to time.Now
+	Now() time.Time
+
+	// After waits for the duration to elapse and then sends the current time on the returned channel.
+	// It is equivalent to time.After
+	After(d time.Duration) <-chan time.Time
+
+	// NewTicker returns a new Ticker containing a channel that will send
+	// the current time on the channel after each tick. The period of the
+	// ticks is specified by the duration argument. The ticker will adjust
+	// the time interval or drop ticks to make up for slow receivers.
+	// The duration d must be greater than zero; if not, NewTicker will
+	// panic. Stop the ticker to release associated resources.
+	NewTicker(d time.Duration) Ticker
+}
+
+// A Ticker holds a channel that delivers "ticks" of a clock at intervals
+type Ticker interface {
+	// Ch returns the channel for the ticker. Equivalent to time.Ticker.C
+	Ch() <-chan time.Time
+
+	// Stop turns off a ticker. After Stop, no more ticks will be sent.
+	// Stop does not close the channel, to prevent a concurrent goroutine
+	// reading from the channel from seeing an erroneous "tick".
+	Stop()
+
+	// Reset stops a ticker and resets its period to the specified duration.
+	// The next tick will arrive after the new period elapses. The duration d
+	// must be greater than zero; if not, Reset will panic.
+	Reset(d time.Duration)
+}
+
+// SystemClock provides an instance of Clock that uses the system clock via methods in the time package.
+var SystemClock Clock = systemClock{}
+
+type systemClock struct {
+}
+
+func (s systemClock) Now() time.Time {
+	return time.Now()
+}
+
+func (s systemClock) After(d time.Duration) <-chan time.Time {
+	return time.After(d)
+}
+
+type SystemTicker struct {
+	*time.Ticker
+}
+
+func (t *SystemTicker) Ch() <-chan time.Time {
+	return t.C
+}
+
+func (s systemClock) NewTicker(d time.Duration) Ticker {
+	return &SystemTicker{time.NewTicker(d)}
+}

op-service/clock/deterministic.go

+package clock
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+type action interface {
+	// Return true if the action is due to fire
+	isDue(time.Time) bool
+
+	// fire triggers the action. Returns true if the action needs to fire again in the future
+	fire(time.Time) bool
+}
+
+type task struct {
+	ch  chan time.Time
+	due time.Time
+}
+
+func (t task) isDue(now time.Time) bool {
+	return !t.due.After(now)
+}
+
+func (t task) fire(now time.Time) bool {
+	t.ch <- now
+	close(t.ch)
+	return false
+}
+
+type ticker struct {
+	c       Clock
+	ch      chan time.Time
+	nextDue time.Time
+	period  time.Duration
+	stopped bool
+	sync.Mutex
+}
+
+func (t *ticker) Ch() <-chan time.Time {
+	return t.ch
+}
+
+func (t *ticker) Stop() {
+	t.Lock()
+	defer t.Unlock()
+	t.stopped = true
+}
+
+func (t *ticker) Reset(d time.Duration) {
+	if d <= 0 {
+		panic("Continuously firing tickers are a really bad idea")
+	}
+	t.Lock()
+	defer t.Unlock()
+	t.period = d
+	t.nextDue = t.c.Now().Add(d)
+}
+
+func (t *ticker) isDue(now time.Time) bool {
+	t.Lock()
+	defer t.Unlock()
+	return !t.nextDue.After(now)
+}
+
+func (t *ticker) fire(now time.Time) bool {
+	t.Lock()
+	defer t.Unlock()
+	if t.stopped {
+		return false
+	}
+	t.ch <- now
+	t.nextDue = now.Add(t.period)
+	return true
+}
+
+type DeterministicClock struct {
+	now          time.Time
+	pending      []action
+	newPendingCh chan struct{}
+	lock         sync.Mutex
+}
+
+// NewDeterministicClock creates a new clock where time only advances when the DeterministicClock.AdvanceTime method is called.
+// This is intended for use in situations where a deterministic clock is required, such as testing or event driven systems.
+func NewDeterministicClock(now time.Time) *DeterministicClock {
+	return &DeterministicClock{
+		now:          now,
+		newPendingCh: make(chan struct{}, 1),
+	}
+}
+
+func (s *DeterministicClock) Now() time.Time {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	return s.now
+}
+
+func (s *DeterministicClock) After(d time.Duration) <-chan time.Time {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	ch := make(chan time.Time, 1)
+	if d.Nanoseconds() == 0 {
+		ch <- s.now
+		close(ch)
+	} else {
+		s.addPending(&task{ch: ch, due: s.now.Add(d)})
+	}
+	return ch
+}
+
+func (s *DeterministicClock) NewTicker(d time.Duration) Ticker {
+	if d <= 0 {
+		panic("Continuously firing tickers are a really bad idea")
+	}
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	ch := make(chan time.Time, 1)
+	t := &ticker{
+		c:       s,
+		ch:      ch,
+		nextDue: s.now.Add(d),
+		period:  d,
+	}
+	s.addPending(t)
+	return t
+}
+
+func (s *DeterministicClock) addPending(t action) {
+	s.pending = append(s.pending, t)
+	select {
+	case s.newPendingCh <- struct{}{}:
+	default:
+		// Must already have a new pending task flagged, do nothing
+	}
+}
+
+// WaitForNewPendingTask blocks until a new task is scheduled since the last time this method was called.
+// true is returned if a new task was scheduled, false if the context completed before a new task was added.
+func (s *DeterministicClock) WaitForNewPendingTask(ctx context.Context) bool {
+	select {
+	case <-ctx.Done():
+		return false
+	case <-s.newPendingCh:
+		return true
+	}
+}
+
+// AdvanceTime moves the time forward by the specific duration
+func (s *DeterministicClock) AdvanceTime(d time.Duration) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.now = s.now.Add(d)
+	var remaining []action
+	for _, a := range s.pending {
+		if !a.isDue(s.now) || a.fire(s.now) {
+			remaining = append(remaining, a)
+		}
+	}
+	s.pending = remaining
+}
+
+var _ Clock = (*DeterministicClock)(nil)

op-service/clock/deterministic_test.go

+package clock
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNowReturnsCurrentTime(t *testing.T) {
+	now := time.UnixMilli(23829382)
+	clock := NewDeterministicClock(now)
+	require.Equal(t, now, clock.Now())
+}
+
+func TestAdvanceTime(t *testing.T) {
+	start := time.UnixMilli(1000)
+	clock := NewDeterministicClock(start)
+	clock.AdvanceTime(500 * time.Millisecond)
+
+	require.Equal(t, start.Add(500*time.Millisecond), clock.Now())
+}
+
+func TestAfter(t *testing.T) {
+	t.Run("ZeroCompletesImmediately", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ch := clock.After(0)
+		require.Len(t, ch, 1, "duration should already have been reached")
+	})
+
+	t.Run("CompletesWhenTimeAdvances", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ch := clock.After(500 * time.Millisecond)
+		require.Len(t, ch, 0, "should not complete immediately")
+
+		clock.AdvanceTime(499 * time.Millisecond)
+		require.Len(t, ch, 0, "should not complete before time is due")
+
+		clock.AdvanceTime(1 * time.Millisecond)
+		require.Len(t, ch, 1, "should complete when time is reached")
+		require.Equal(t, clock.Now(), <-ch)
+	})
+
+	t.Run("CompletesWhenTimeAdvancesPastDue", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ch := clock.After(500 * time.Millisecond)
+		require.Len(t, ch, 0, "should not complete immediately")
+
+		clock.AdvanceTime(9000 * time.Millisecond)
+		require.Len(t, ch, 1, "should complete when time is past")
+		require.Equal(t, clock.Now(), <-ch)
+	})
+
+	t.Run("RegisterAsPending", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		_ = clock.After(500 * time.Millisecond)
+
+		ctx, cancelFunc := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancelFunc()
+		require.True(t, clock.WaitForNewPendingTask(ctx), "should have added a new pending task")
+	})
+}
+
+func TestNewTicker(t *testing.T) {
+	t.Run("FiresAfterEachDuration", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ticker := clock.NewTicker(5 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire immediately")
+
+		clock.AdvanceTime(4 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire before due")
+
+		clock.AdvanceTime(1 * time.Second)
+		require.Len(t, ticker.Ch(), 1, "should fire when due")
+		require.Equal(t, clock.Now(), <-ticker.Ch(), "should post current time")
+
+		clock.AdvanceTime(4 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not re-fire before due")
+
+		clock.AdvanceTime(1 * time.Second)
+		require.Len(t, ticker.Ch(), 1, "should fire when due")
+		require.Equal(t, clock.Now(), <-ticker.Ch(), "should post current time")
+	})
+
+	t.Run("SkipsFiringWhenAdvancedMultipleDurations", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ticker := clock.NewTicker(5 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire immediately")
+
+		// Advance more than three periods, but should still only fire once
+		clock.AdvanceTime(16 * time.Second)
+		require.Len(t, ticker.Ch(), 1, "should fire when due")
+		require.Equal(t, clock.Now(), <-ticker.Ch(), "should post current time")
+
+		clock.AdvanceTime(1 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire until due again")
+	})
+
+	t.Run("StopFiring", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ticker := clock.NewTicker(5 * time.Second)
+
+		ticker.Stop()
+
+		clock.AdvanceTime(10 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire after stop")
+	})
+
+	t.Run("ResetPanicWhenLessNotPositive", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ticker := clock.NewTicker(5 * time.Second)
+		require.Panics(t, func() {
+			ticker.Reset(0)
+		}, "reset to 0 should panic")
+		require.Panics(t, func() {
+			ticker.Reset(-1)
+		}, "reset to negative duration should panic")
+	})
+
+	t.Run("ResetWithShorterPeriod", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ticker := clock.NewTicker(5 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire immediately")
+
+		ticker.Reset(1 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire immediately after reset")
+
+		clock.AdvanceTime(1 * time.Second)
+		require.Len(t, ticker.Ch(), 1, "should fire when new duration reached")
+		require.Equal(t, clock.Now(), <-ticker.Ch(), "should post current time")
+	})
+
+	t.Run("ResetWithLongerPeriod", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ticker := clock.NewTicker(5 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire immediately")
+
+		ticker.Reset(7 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire immediately after reset")
+
+		clock.AdvanceTime(5 * time.Second)
+		require.Len(t, ticker.Ch(), 0, "should not fire when old duration reached")
+
+		clock.AdvanceTime(2 * time.Second)
+		require.Len(t, ticker.Ch(), 1, "should fire when new duration reached")
+		require.Equal(t, clock.Now(), <-ticker.Ch(), "should post current time")
+	})
+
+	t.Run("RegisterAsPending", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		ticker := clock.NewTicker(5 * time.Second)
+		defer ticker.Stop()
+
+		ctx, cancelFunc := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancelFunc()
+		require.True(t, clock.WaitForNewPendingTask(ctx), "should have added a new pending task")
+	})
+}
+
+func TestWaitForPending(t *testing.T) {
+	t.Run("DoNotBlockWhenAlreadyPending", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		_ = clock.After(5 * time.Minute)
+		_ = clock.After(5 * time.Minute)
+
+		ctx, cancelFunc := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancelFunc()
+		require.True(t, clock.WaitForNewPendingTask(ctx), "should have added a new pending task")
+	})
+
+	t.Run("ResetNewPendingFlagAfterWaiting", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		_ = clock.After(5 * time.Minute)
+		_ = clock.After(5 * time.Minute)
+
+		ctx, cancelFunc := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancelFunc()
+		require.True(t, clock.WaitForNewPendingTask(ctx), "should have added a new pending task")
+
+		ctx, cancelFunc = context.WithTimeout(context.Background(), 250*time.Millisecond)
+		defer cancelFunc()
+		require.False(t, clock.WaitForNewPendingTask(ctx), "should have reset new pending task flag")
+	})
+}

packages/contracts-bedrock/.gas-snapshot

@@ -29,7 +29,7 @@ DeployerWhitelist_Test:test_owner_succeeds() (gas: 7582)
 DeployerWhitelist_Test:test_storageSlots_succeeds() (gas: 33395)
 DisputeGameFactory_Test:test_owner_succeeds() (gas: 7582)
 DisputeGameFactory_Test:test_setImplementation_notOwner_reverts() (gas: 11191)
-DisputeGameFactory_Test:test_setImplementation_succeeds() (gas: 32635)
+DisputeGameFactory_Test:test_setImplementation_succeeds() (gas: 38765)
 DisputeGameFactory_Test:test_transferOwnership_notOwner_reverts() (gas: 10979)
 DisputeGameFactory_Test:test_transferOwnership_succeeds() (gas: 13180)
 FeeVault_Test:test_constructor_succeeds() (gas: 10736)

packages/contracts-bedrock/contracts/dispute/BondManager.sol

@@ -13,7 +13,9 @@ import { IDisputeGameFactory } from "./IDisputeGameFactory.sol";
  * @notice The Bond Manager serves as an escrow for permissionless output proposal bonds.
  */
 contract BondManager {
-    // The Bond Type
+    /**
+     * @notice The Bond Type
+     */
     struct Bond {
         address owner;
         uint256 expiration;
@@ -58,6 +60,13 @@ contract BondManager {
      */
     IDisputeGameFactory public immutable DISPUTE_GAME_FACTORY;
 
+    /**
+     * @notice Amount of gas used to transfer ether when splitting the bond.
+     *         This is a reasonable amount of gas for a transfer, even to a smart contract.
+     *         The number of participants is bound of by the block gas limit.
+     */
+    uint256 private constant TRANSFER_GAS = 30_000;
+
     /**
      * @notice Instantiates the bond maanger with the registered dispute game factory.
      * @param _disputeGameFactory is the dispute game factory.
@@ -147,13 +156,14 @@ contract BondManager {
 
         uint256 len = _claimRecipients.length;
         uint256 proportionalAmount = b.amount / len;
-        for (uint256 i = 0; i < len; i++) {
-            bool success = SafeCall.send(
-                payable(_claimRecipients[i]),
-                gasleft() / len,
-                proportionalAmount
-            );
-            require(success, "BondManager: Failed to send Ether.");
+        // Send the proportional amount to each recipient. Do not revert if a send fails as that
+        // will prevent other recipients from receiving their share.
+        for (uint256 i; i < len; i++) {
+            SafeCall.send({
+                _target: payable(_claimRecipients[i]),
+                _gas: TRANSFER_GAS,
+                _value: proportionalAmount
+            });
         }
     }
 

packages/contracts-bedrock/contracts/dispute/DisputeGameFactory.sol

@@ -113,6 +113,7 @@ contract DisputeGameFactory is Ownable, IDisputeGameFactory {
      */
     function setImplementation(GameType gameType, IDisputeGame impl) external onlyOwner {
         gameImpls[gameType] = impl;
+        emit ImplementationSet(address(impl), gameType);
     }
 
     /**

packages/contracts-bedrock/contracts/dispute/IDisputeGameFactory.sol

@@ -23,6 +23,13 @@ interface IDisputeGameFactory {
         Claim indexed rootClaim
     );
 
+    /**
+     * @notice Emitted when a new game implementation added to the factory
+     * @param impl The implementation contract for the given `GameType`.
+     * @param gameType The type of the DisputeGame.
+     */
+    event ImplementationSet(address indexed impl, GameType indexed gameType);
+
     /**
      * @notice `games` queries an internal a mapping that maps the hash of
      *         `gameType ++ rootClaim ++ extraData` to the deployed `DisputeGame` clone.

packages/contracts-bedrock/contracts/test/DisputeGameFactory.t.sol

@@ -18,6 +18,8 @@ contract DisputeGameFactory_Test is Test {
         Claim indexed rootClaim
     );
 
+    event ImplementationSet(address indexed impl, GameType indexed gameType);
+
     function setUp() public {
         factory = new DisputeGameFactory(address(this));
         fakeClone = new FakeClone();
@@ -105,6 +107,9 @@ contract DisputeGameFactory_Test is Test {
         // There should be no implementation for the `GameType.FAULT` enum value, it has not been set.
         assertEq(address(factory.gameImpls(GameType.FAULT)), address(0));
 
+        vm.expectEmit(true, true, true, true, address(factory));
+        emit ImplementationSet(address(1), GameType.FAULT);
+
         // Set the implementation for the `GameType.FAULT` enum value.
         factory.setImplementation(GameType.FAULT, IDisputeGame(address(1)));