forked from relayooor/go-ethereum
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement pushing blocks data to the DB (ethereum#18)
Co-authored-by: Bhakiyaraj Kalimuthu <bhakiya.kalimuthu@gmail.com>
- Loading branch information
Showing
16 changed files
with
436 additions
and
108 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,136 @@ | ||
| package builder | ||
|
|
||
| import ( | ||
| "context" | ||
| "database/sql" | ||
| "math/big" | ||
| "time" | ||
|
|
||
| "github.com/ethereum/go-ethereum/core/types" | ||
| "github.com/ethereum/go-ethereum/log" | ||
| boostTypes "github.com/flashbots/go-boost-utils/types" | ||
| "github.com/jmoiron/sqlx" | ||
| _ "github.com/lib/pq" | ||
| ) | ||
|
|
||
| type IDatabaseService interface { | ||
| ConsumeBuiltBlock(block *types.Block, bundles []types.SimulatedBundle, bidTrace *boostTypes.BidTrace) | ||
| } | ||
|
|
||
| type NilDbService struct{} | ||
|
|
||
| func (NilDbService) ConsumeBuiltBlock(*types.Block, []types.SimulatedBundle, *boostTypes.BidTrace) {} | ||
|
|
||
| type DatabaseService struct { | ||
| db *sqlx.DB | ||
|
|
||
| insertBuiltBlockStmt *sqlx.NamedStmt | ||
| insertBlockBuiltBundleNoIdStmt *sqlx.NamedStmt | ||
| insertBlockBuiltBundleWithIdStmt *sqlx.NamedStmt | ||
| insertMissingBundleStmt *sqlx.NamedStmt | ||
| } | ||
|
|
||
| func NewDatabaseService(postgresDSN string) (*DatabaseService, error) { | ||
| db, err := sqlx.Connect("postgres", postgresDSN) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| insertBuiltBlockStmt, err := db.PrepareNamed("insert into built_blocks (block_number, profit, slot, hash, gas_limit, gas_used, base_fee, parent_hash, proposer_pubkey, proposer_fee_recipient, builder_pubkey, timestamp, timestamp_datetime) values (:block_number, :profit, :slot, :hash, :gas_limit, :gas_used, :base_fee, :parent_hash, :proposer_pubkey, :proposer_fee_recipient, :builder_pubkey, :timestamp, to_timestamp(:timestamp)) returning block_id") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| insertBlockBuiltBundleNoIdStmt, err := db.PrepareNamed("insert into built_blocks_bundles (block_id, bundle_id) select :block_id, id from bundles where bundle_hash = :bundle_hash and param_block_number = :block_number returning bundle_id") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| insertBlockBuiltBundleWithIdStmt, err := db.PrepareNamed("insert into built_blocks_bundles (block_id, bundle_id) select :block_id, :bundle_id returning bundle_id") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| insertMissingBundleStmt, err := db.PrepareNamed("insert into bundles (bundle_hash, param_signed_txs, param_block_number, param_timestamp, received_timestamp, param_reverting_tx_hashes, coinbase_diff, total_gas_used, state_block_number, gas_fees, eth_sent_to_coinbase) values (:bundle_hash, :param_signed_txs, :param_block_number, :param_timestamp, :received_timestamp, :param_reverting_tx_hashes, :coinbase_diff, :total_gas_used, :state_block_number, :gas_fees, :eth_sent_to_coinbase) on conflict (bundle_hash, param_block_number) do nothing returning id") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &DatabaseService{ | ||
| db: db, | ||
| insertBuiltBlockStmt: insertBuiltBlockStmt, | ||
| insertBlockBuiltBundleNoIdStmt: insertBlockBuiltBundleNoIdStmt, | ||
| insertBlockBuiltBundleWithIdStmt: insertBlockBuiltBundleWithIdStmt, | ||
| insertMissingBundleStmt: insertMissingBundleStmt, | ||
| }, nil | ||
| } | ||
|
|
||
| func (ds *DatabaseService) ConsumeBuiltBlock(block *types.Block, bundles []types.SimulatedBundle, bidTrace *boostTypes.BidTrace) { | ||
| tx, err := ds.db.Beginx() | ||
|
|
||
| blockData := BuiltBlock{ | ||
| BlockNumber: block.NumberU64(), | ||
| Profit: new(big.Rat).SetFrac(block.Profit, big.NewInt(1e18)).FloatString(18), | ||
| Slot: bidTrace.Slot, | ||
| Hash: block.Hash().String(), | ||
| GasLimit: block.GasLimit(), | ||
| GasUsed: block.GasUsed(), | ||
| BaseFee: block.BaseFee().Uint64(), | ||
| ParentHash: block.ParentHash().String(), | ||
| ProposerPubkey: bidTrace.ProposerPubkey.String(), | ||
| ProposerFeeRecipient: bidTrace.ProposerFeeRecipient.String(), | ||
| BuilderPubkey: bidTrace.BuilderPubkey.String(), | ||
| Timestamp: block.Time(), | ||
| } | ||
|
|
||
| ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second) | ||
| defer cancel() | ||
| var blockId uint64 | ||
| if err = tx.NamedStmtContext(ctx, ds.insertBuiltBlockStmt).GetContext(ctx, &blockId, blockData); err != nil { | ||
| log.Error("could not insert built block", "err", err) | ||
| tx.Rollback() | ||
| return | ||
| } | ||
|
|
||
| for _, bundle := range bundles { | ||
| bundleData := BuiltBlockBundle{ | ||
| BlockId: blockId, | ||
| BundleId: nil, | ||
| BlockNumber: blockData.BlockNumber, | ||
| BundleHash: bundle.OriginalBundle.Hash.String(), | ||
| } | ||
|
|
||
| var bundleId uint64 | ||
| err := tx.NamedStmtContext(ctx, ds.insertBlockBuiltBundleNoIdStmt).GetContext(ctx, &bundleId, bundleData) | ||
| if err == nil { | ||
| continue | ||
| } | ||
|
|
||
| if err != sql.ErrNoRows { | ||
| log.Error("could not insert bundle", "err", err) | ||
| // Try anyway | ||
| } | ||
|
|
||
| missingBundleData := SimulatedBundleToDbBundle(&bundle) | ||
| err = ds.insertMissingBundleStmt.GetContext(ctx, &bundleId, missingBundleData) // not using the tx as it relies on the unique constraint! | ||
| if err == nil { | ||
| bundleData.BundleId = &bundleId | ||
| _, err = tx.NamedStmtContext(ctx, ds.insertBlockBuiltBundleWithIdStmt).ExecContext(ctx, bundleData) | ||
| if err != nil { | ||
| log.Error("could not insert built block bundle after inserting missing bundle", "err", err) | ||
| } | ||
| } else if err == sql.ErrNoRows /* conflict, someone else inserted the bundle before we could */ { | ||
| if err := tx.NamedStmtContext(ctx, ds.insertBlockBuiltBundleNoIdStmt).GetContext(ctx, &bundleId, bundleData); err != nil { | ||
| log.Error("could not insert bundle on retry", "err", err) | ||
| continue | ||
| } | ||
| } else { | ||
| log.Error("could not insert missing bundle", "err", err) | ||
| } | ||
| } | ||
|
|
||
| err = tx.Commit() | ||
| if err != nil { | ||
| log.Error("could not commit DB trasnaction", "err", err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package builder | ||
|
|
||
| import ( | ||
| "math/big" | ||
| "os" | ||
| "testing" | ||
|
|
||
| "github.com/ethereum/go-ethereum/common" | ||
| "github.com/ethereum/go-ethereum/core/types" | ||
| boostTypes "github.com/flashbots/go-boost-utils/types" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestDatabaseBlockInsertion(t *testing.T) { | ||
| dsn := os.Getenv("FLASHBOTS_TEST_POSTGRES_DSN") | ||
| if dsn == "" { | ||
| return | ||
| } | ||
|
|
||
| ds, err := NewDatabaseService(dsn) | ||
| require.NoError(t, err) | ||
|
|
||
| _, err = ds.db.Exec("insert into bundles (id, param_block_number, bundle_hash) values (10, 20, '0x1078')") | ||
| require.NoError(t, err) | ||
|
|
||
| block := types.NewBlock( | ||
| &types.Header{ | ||
| ParentHash: common.HexToHash("0xafafafa"), | ||
| Number: big.NewInt(132), | ||
| GasLimit: uint64(10000), | ||
| GasUsed: uint64(1000), | ||
| Time: 16000000, | ||
| BaseFee: big.NewInt(7), | ||
| }, nil, nil, nil, nil) | ||
| block.Profit = big.NewInt(10) | ||
|
|
||
| simBundle1 := types.SimulatedBundle{ | ||
| MevGasPrice: big.NewInt(9), | ||
| TotalEth: big.NewInt(11), | ||
| EthSentToCoinbase: big.NewInt(10), | ||
| TotalGasUsed: uint64(100), | ||
| OriginalBundle: types.MevBundle{ | ||
| Txs: types.Transactions{types.NewTransaction(uint64(50), common.Address{0x60}, big.NewInt(19), uint64(67), big.NewInt(43), []byte{})}, | ||
| BlockNumber: big.NewInt(12), | ||
| MinTimestamp: uint64(1000000), | ||
| RevertingTxHashes: []common.Hash{common.Hash{0x10, 0x17}}, | ||
| Hash: common.Hash{0x09, 0x78}, | ||
| }, | ||
| } | ||
| simBundle2 := types.SimulatedBundle{ | ||
| MevGasPrice: big.NewInt(90), | ||
| TotalEth: big.NewInt(110), | ||
| EthSentToCoinbase: big.NewInt(100), | ||
| TotalGasUsed: uint64(1000), | ||
| OriginalBundle: types.MevBundle{ | ||
| Txs: types.Transactions{types.NewTransaction(uint64(51), common.Address{0x61}, big.NewInt(109), uint64(167), big.NewInt(433), []byte{})}, | ||
| BlockNumber: big.NewInt(20), | ||
| MinTimestamp: uint64(1000020), | ||
| RevertingTxHashes: []common.Hash{common.Hash{0x11, 0x17}}, | ||
| Hash: common.Hash{0x10, 0x78}, | ||
| }, | ||
| } | ||
|
|
||
| bidTrace := &boostTypes.BidTrace{} | ||
|
|
||
| ds.ConsumeBuiltBlock(block, []types.SimulatedBundle{simBundle1, simBundle2}, bidTrace) | ||
|
|
||
| var dbBlock BuiltBlock | ||
| ds.db.Get(&dbBlock, "select * from built_blocks where hash = '0x24e6998e4d2b4fd85f7f0d27ac4b87aaf0ec18e156e4b6e194ab5dadee0cd004'") | ||
| t.Logf("block %v", dbBlock) | ||
| } |
Oops, something went wrong.