feat: add proposer changes

proposer/op/proposer/service.go

@@ -56,6 +56,7 @@ type ProposerConfig struct {
 	UseCachedDb                bool
 	SlackToken                 string
 	BeaconRpc                  string
+	RollupRpc                  string
 	TxCacheOutDir              string
 	MaxBlockRangePerSpanProof  uint64
 	L2ChainID                  uint64
@@ -118,6 +119,7 @@ func (ps *ProposerService) initFromCLIConfig(ctx context.Context, version string
 	ps.UseCachedDb = cfg.UseCachedDb
 	ps.SlackToken = cfg.SlackToken
 	ps.BeaconRpc = cfg.BeaconRpc
+	ps.RollupRpc = cfg.RollupRpc
 	ps.TxCacheOutDir = cfg.TxCacheOutDir
 	ps.MaxBlockRangePerSpanProof = cfg.MaxBlockRangePerSpanProof
 	ps.OPSuccinctServerUrl = cfg.OPSuccinctServerUrl

proposer/op/proposer/span_batches.go

@@ -3,7 +3,10 @@ package proposer
 import (
 	"context"
 	"fmt"
+	"slices"
 
+	"github.com/ethereum-optimism/optimism/op-service/dial"
+	"github.com/ethereum-optimism/optimism/op-service/sources"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/succinctlabs/op-succinct-go/proposer/db/ent"
 	"github.com/succinctlabs/op-succinct-go/proposer/db/ent/proofrequest"
@@ -14,8 +17,113 @@ type Span struct {
 	End   uint64
 }
 
+// GetL1HeadForL2Block returns the L1 block from which the L2 block can be derived.
+func (l *L2OutputSubmitter) GetL1HeadForL2Block(ctx context.Context, rollupClient *sources.RollupClient, l2End uint64) (uint64, error) {
+	status, err := rollupClient.SyncStatus(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get sync status: %w", err)
+	}
+	latestL1Block := status.HeadL1.Number
+
+	// Get the L1 origin of the end block.
+	outputResponse, err := rollupClient.OutputAtBlock(ctx, l2End)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get l1 origin: %w", err)
+	}
+	L2EndL1Origin := outputResponse.BlockRef.L1Origin.Number
+
+	// Search forward from the L1 origin of the L2 end block until we find a safe head greater than the L2 end block.
+	for currentL1Block := L2EndL1Origin; currentL1Block <= latestL1Block; currentL1Block++ {
+		safeHead, err := rollupClient.SafeHeadAtL1Block(ctx, currentL1Block)
+		if err != nil {
+			return 0, fmt.Errorf("failed to get safe head: %w", err)
+		}
+		// If the safe head is greater than or equal to the L2 end block at this L1 block, then we can derive the L2 end block from this L1 block.
+		if safeHead.SafeHead.Number >= l2End {
+			return currentL1Block, nil
+		}
+	}
+
+	return 0, fmt.Errorf("could not find an L1 block with an L2 safe head greater than the L2 end block")
+}
+
+func (l *L2OutputSubmitter) isSafeDBActivated(ctx context.Context, rollupClient *sources.RollupClient) (bool, error) {
+	// Get the sync status of the rollup node.
+	status, err := rollupClient.SyncStatus(ctx)
+	if err != nil {
+		return false, fmt.Errorf("failed to get sync status: %w", err)
+	}
+
+	// Attempt querying the safe head at the latest L1 block.
+	_, err = rollupClient.SafeHeadAtL1Block(ctx, status.HeadL1.Number)
+	if err != nil {
+		return false, fmt.Errorf("failed to get safe head: %w", err)
+	}
+
+	return true, nil
+}
+
+// SplitRangeBasedOnSafeHeads splits a range into spans based on safe head boundaries.
+// This is useful when we want to ensure that each span aligns with L2 safe head boundaries.
+func (l *L2OutputSubmitter) SplitRangeBasedOnSafeHeads(ctx context.Context, l2Start, l2End uint64) ([]Span, error) {
+	spans := []Span{}
+	currentStart := l2Start
+
+	rollupClient, err := dial.DialRollupClientWithTimeout(ctx, dial.DefaultDialTimeout, l.Log, l.Cfg.RollupRpc)
+	if err != nil {
+		return nil, err
+	}
+
+	L1Head, err := l.GetL1HeadForL2Block(ctx, rollupClient, l2End)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get l1 head for l2 block: %w", err)
+	}
+
+	l2StartOutput, err := rollupClient.OutputAtBlock(ctx, l2Start)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get l2 start output: %w", err)
+	}
+	L2StartL1Origin := l2StartOutput.BlockRef.L1Origin.Number
+
+	// Get all the unique safe heads between l1_start and l1_head
+	safeHeads := make(map[uint64]struct{})
+	for currentL1Block := L2StartL1Origin; currentL1Block <= L1Head; currentL1Block++ {
+		safeHead, err := rollupClient.SafeHeadAtL1Block(ctx, currentL1Block)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get safe head: %w", err)
+		}
+		safeHeads[safeHead.SafeHead.Number] = struct{}{}
+	}
+	uniqueSafeHeads := make([]uint64, 0, len(safeHeads))
+	for safeHead := range safeHeads {
+		uniqueSafeHeads = append(uniqueSafeHeads, safeHead)
+	}
+	slices.Sort(uniqueSafeHeads)
+
+	// Loop over all of the safe heads and create spans.
+	for _, safeHead := range uniqueSafeHeads {
+		if safeHead > currentStart {
+			rangeStart := currentStart
+			for rangeStart+l.Cfg.MaxBlockRangePerSpanProof < min(l2End, safeHead) {
+				spans = append(spans, Span{
+					Start: rangeStart,
+					End:   rangeStart + l.Cfg.MaxBlockRangePerSpanProof,
+				})
+				rangeStart += l.Cfg.MaxBlockRangePerSpanProof
+			}
+			spans = append(spans, Span{
+				Start: rangeStart,
+				End:   min(l2End, safeHead),
+			})
+			currentStart = safeHead
+		}
+	}
+
+	return spans, nil
+}
+
 // CreateSpans creates a list of spans of size MaxBlockRangePerSpanProof from start to end. Note: The end of span i = start of span i+1.
-func (l *L2OutputSubmitter) CreateSpans(start, end uint64) []Span {
+func (l *L2OutputSubmitter) SplitRangeBasic(start, end uint64) []Span {
 	spans := []Span{}
 	// Create spans of size MaxBlockRangePerSpanProof from start to end.
 	// Each span starts where the previous one ended.
@@ -44,9 +152,9 @@ func (l *L2OutputSubmitter) DeriveNewSpanBatches(ctx context.Context) error {
 	}
 	newL2StartBlock := latestL2EndBlock
 
-	rollupClient, err := l.RollupProvider.RollupClient(ctx)
+	rollupClient, err := dial.DialRollupClientWithTimeout(ctx, dial.DefaultDialTimeout, l.Log, l.Cfg.RollupRpc)
 	if err != nil {
-		return fmt.Errorf("failed to get rollup client: %w", err)
+		return err
 	}
 
 	// Get the latest finalized L2 block.
@@ -58,8 +166,25 @@ func (l *L2OutputSubmitter) DeriveNewSpanBatches(ctx context.Context) error {
 	// Note: Originally, this used the L1 finalized block. However, to satisfy the new API, we now use the L2 finalized block.
 	newL2EndBlock := status.FinalizedL2.Number
 
-	// Create spans of size MaxBlockRangePerSpanProof from newL2StartBlock to newL2EndBlock.
-	spans := l.CreateSpans(newL2StartBlock, newL2EndBlock)
+	// Check if the safeDB is activated on the L2 node. If it is, we use the safeHead based range
+	// splitting algorithm. Otherwise, we use the simple range splitting algorithm.
+	safeDBActivated, err := l.isSafeDBActivated(ctx, rollupClient)
+	if err != nil {
+		return fmt.Errorf("failed to check if safeDB is activated: %w", err)
+	}
+
+	var spans []Span
+	// If the safeDB is activated, we use the safeHead based range splitting algorithm.
+	// Otherwise, we use the simple range splitting algorithm.
+	if safeDBActivated {
+		spans, err = l.SplitRangeBasedOnSafeHeads(ctx, newL2StartBlock, newL2EndBlock)
+		if err != nil {
+			return fmt.Errorf("failed to split range based on safe heads: %w", err)
+		}
+	} else {
+		spans = l.SplitRangeBasic(newL2StartBlock, newL2EndBlock)
+	}
+
 	// Add each span to the DB. If there are no spans, we will not create any proofs.
 	for _, span := range spans {
 		err := l.db.NewEntry(proofrequest.TypeSPAN, span.Start, span.End)

proposer/op/proposer/span_batches_test.go

@@ -59,7 +59,7 @@ func TestCreateSpans(t *testing.T) {
 			l := &L2OutputSubmitter{}
 			l.Cfg = ProposerConfig{MaxBlockRangePerSpanProof: tt.maxBlockRange}
 
-			spans := l.CreateSpans(tt.start, tt.end)
+			spans := l.SplitRangeBasic(tt.start, tt.end)
 
 			assert.Equal(t, tt.expectedSpansCount, len(spans), "Unexpected number of spans")
 

utils/host/src/fetcher.rs

@@ -838,7 +838,8 @@ impl OPSuccinctDataFetcher {
                 )
                 .await?;
             let l2_safe_head = result.safe_head.number;
-            if l2_safe_head > l2_end_block {
+            // If the safe head is greater than or equal to the L2 end block at this L1 block, then we can derive the L2 end block from this L1 block.
+            if l2_safe_head >= l2_end_block {
                 return Ok((result.l1_block.hash, result.l1_block.number));
             }