More bundle visibility (#42)

Use batch queries for upserting data on built blocks. Include more data on the built blocks - sealing time, orderflow cutoff time, and include data for all bundles considered.

Author: Ruteri

builder/builder.go

@@ -97,7 +97,7 @@ func (b *Builder) Stop() error {
 	return nil
 }
 
-func (b *Builder) onSealedBlock(block *types.Block, bundles []types.SimulatedBundle, proposerPubkey boostTypes.PublicKey, proposerFeeRecipient boostTypes.Address, attrs *BuilderPayloadAttributes) error {
+func (b *Builder) onSealedBlock(block *types.Block, ordersClosedAt time.Time, sealedAt time.Time, commitedBundles []types.SimulatedBundle, allBundles []types.SimulatedBundle, proposerPubkey boostTypes.PublicKey, proposerFeeRecipient boostTypes.Address, attrs *BuilderPayloadAttributes) error {
 	executableData := beacon.BlockToExecutableData(block)
 	payload, err := executableDataToExecutionPayload(executableData)
 	if err != nil {
@@ -142,15 +142,15 @@ func (b *Builder) onSealedBlock(block *types.Block, bundles []types.SimulatedBun
 			log.Error("could not validate block", "err", err)
 		}
 	} else {
-		go b.ds.ConsumeBuiltBlock(block, bundles, &blockBidMsg)
+		go b.ds.ConsumeBuiltBlock(block, ordersClosedAt, sealedAt, commitedBundles, allBundles, &blockBidMsg)
 		err = b.relay.SubmitBlock(&blockSubmitReq)
 		if err != nil {
-			log.Error("could not submit block", "err", err, "bundles", len(bundles))
+			log.Error("could not submit block", "err", err, "#commitedBundles", len(commitedBundles))
 			return err
 		}
 	}
 
-	log.Info("submitted block", "slot", blockBidMsg.Slot, "value", blockBidMsg.Value.String(), "parent", blockBidMsg.ParentHash, "hash", block.Hash(), "bundles", len(bundles))
+	log.Info("submitted block", "slot", blockBidMsg.Slot, "value", blockBidMsg.Value.String(), "parent", blockBidMsg.ParentHash, "hash", block.Hash(), "#commitedBundles", len(commitedBundles))
 
 	return nil
 }
@@ -212,6 +212,14 @@ func (b *Builder) OnPayloadAttribute(attrs *BuilderPayloadAttributes) error {
 	return nil
 }
 
+type blockQueueEntry struct {
+	block           *types.Block
+	ordersCloseTime time.Time
+	sealedAt        time.Time
+	commitedBundles []types.SimulatedBundle
+	allBundles      []types.SimulatedBundle
+}
+
 func (b *Builder) runBuildingJob(slotCtx context.Context, proposerPubkey boostTypes.PublicKey, feeRecipient boostTypes.Address, attrs *BuilderPayloadAttributes) {
 	ctx, cancel := context.WithTimeout(slotCtx, 12*time.Second)
 	defer cancel()
@@ -229,16 +237,16 @@ func (b *Builder) runBuildingJob(slotCtx context.Context, proposerPubkey boostTy
 		queueMu                  sync.Mutex
 		queueLastSubmittedProfit = new(big.Int)
 		queueBestProfit          = new(big.Int)
-		queueBestBlock           *types.Block
-		queueBestBundles         []types.SimulatedBundle
+		queueBestEntry           blockQueueEntry
 	)
 
 	log.Debug("runBuildingJob", "slot", attrs.Slot, "parent", attrs.HeadHash)
 
 	submitBestBlock := func() {
 		queueMu.Lock()
 		if queueLastSubmittedProfit.Cmp(queueBestProfit) < 0 {
-			err := b.onSealedBlock(queueBestBlock, queueBestBundles, proposerPubkey, feeRecipient, attrs)
+			err := b.onSealedBlock(queueBestEntry.block, queueBestEntry.ordersCloseTime, queueBestEntry.sealedAt, queueBestEntry.commitedBundles, queueBestEntry.allBundles, proposerPubkey, feeRecipient, attrs)
+
 			if err != nil {
 				log.Error("could not run sealed block hook", "err", err)
 			} else {
@@ -252,16 +260,23 @@ func (b *Builder) runBuildingJob(slotCtx context.Context, proposerPubkey boostTy
 	go runResubmitLoop(ctx, b.limiter, queueSignal, submitBestBlock)
 
 	// Populates queue with submissions that increase block profit
-	blockHook := func(block *types.Block, bundles []types.SimulatedBundle) {
+	blockHook := func(block *types.Block, ordersCloseTime time.Time, commitedBundles []types.SimulatedBundle, allBundles []types.SimulatedBundle) {
 		if ctx.Err() != nil {
 			return
 		}
 
+		sealedAt := time.Now()
+
 		queueMu.Lock()
 		defer queueMu.Unlock()
 		if block.Profit.Cmp(queueBestProfit) > 0 {
-			queueBestBlock = block
-			queueBestBundles = bundles
+			queueBestEntry = blockQueueEntry{
+				block:           block,
+				ordersCloseTime: ordersCloseTime,
+				sealedAt:        sealedAt,
+				commitedBundles: commitedBundles,
+				allBundles:      allBundles,
+			}
 			queueBestProfit.Set(block.Profit)
 
 			select {

builder/eth_service.go

@@ -9,10 +9,11 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/miner"
 )
 
 type IEthereumService interface {
-	BuildBlock(attrs *BuilderPayloadAttributes, sealedBlockCallback func(*types.Block, []types.SimulatedBundle)) error
+	BuildBlock(attrs *BuilderPayloadAttributes, sealedBlockCallback miner.BlockHookFn) error
 	GetBlockByHash(hash common.Hash) *types.Block
 	Synced() bool
 }
@@ -22,10 +23,11 @@ type testEthereumService struct {
 	testExecutableData *beacon.ExecutableDataV1
 	testBlock          *types.Block
 	testBundlesMerged  []types.SimulatedBundle
+	testAllBundles     []types.SimulatedBundle
 }
 
-func (t *testEthereumService) BuildBlock(attrs *BuilderPayloadAttributes, sealedBlockCallback func(*types.Block, []types.SimulatedBundle)) error {
-	sealedBlockCallback(t.testBlock, t.testBundlesMerged)
+func (t *testEthereumService) BuildBlock(attrs *BuilderPayloadAttributes, sealedBlockCallback miner.BlockHookFn) error {
+	sealedBlockCallback(t.testBlock, time.Now(), t.testBundlesMerged, t.testAllBundles)
 	return nil
 }
 
@@ -41,7 +43,7 @@ func NewEthereumService(eth *eth.Ethereum) *EthereumService {
 	return &EthereumService{eth: eth}
 }
 
-func (s *EthereumService) BuildBlock(attrs *BuilderPayloadAttributes, sealedBlockCallback func(*types.Block, []types.SimulatedBundle)) error {
+func (s *EthereumService) BuildBlock(attrs *BuilderPayloadAttributes, sealedBlockCallback miner.BlockHookFn) error {
 	// Send a request to generate a full block in the background.
 	// The result can be obtained via the returned channel.
 	resCh, err := s.eth.Miner().GetSealingBlockAsync(attrs.HeadHash, uint64(attrs.Timestamp), attrs.SuggestedFeeRecipient, attrs.GasLimit, attrs.Random, false, sealedBlockCallback)

builder/eth_service_test.go

@@ -93,7 +93,7 @@ func TestBuildBlock(t *testing.T) {
 	service := NewEthereumService(ethservice)
 	service.eth.APIBackend.Miner().SetEtherbase(common.Address{0x05, 0x11})
 
-	err := service.BuildBlock(testPayloadAttributes, func(block *types.Block, _ []types.SimulatedBundle) {
+	err := service.BuildBlock(testPayloadAttributes, func(block *types.Block, _ time.Time, _ []types.SimulatedBundle, _ []types.SimulatedBundle) {
 		executableData := beacon.BlockToExecutableData(block)
 		require.Equal(t, common.Address{0x05, 0x11}, executableData.FeeRecipient)
 		require.Equal(t, common.Hash{0x05, 0x10}, executableData.Random)