From 103273939b7979b0a517fc89b8570cf7cb18e53c Mon Sep 17 00:00:00 2001 From: faheelsattar Date: Wed, 2 Oct 2024 20:47:06 +0500 Subject: [PATCH 1/3] add tx inclusion --- internal/data/metrics.sql.go | 17 + internal/data/models.sqlc.gen.go | 9 +- internal/data/sql/queries/metrics.sql | 4 + internal/metrics/tx_mapper_db.go | 311 ++++++++++-------- internal/metrics/types.go | 2 +- internal/scheduler/scheduler.go | 2 + internal/watcher/watcher.go | 2 + .../20241002011333_add_tx_status_val_type.sql | 12 + .../20241002075029_update_tx_status.sql | 11 + 9 files changed, 226 insertions(+), 144 deletions(-) create mode 100644 migrations/20241002011333_add_tx_status_val_type.sql create mode 100644 migrations/20241002075029_update_tx_status.sql diff --git a/internal/data/metrics.sql.go b/internal/data/metrics.sql.go index 77b461a..f1f4ccd 100644 --- a/internal/data/metrics.sql.go +++ b/internal/data/metrics.sql.go @@ -568,3 +568,20 @@ func (q *Queries) QueryValidatorStatuses(ctx context.Context, arg QueryValidator } return items, nil } + +const updateDecryptedTX = `-- name: UpdateDecryptedTX :exec +UPDATE decrypted_tx +SET tx_status = $1 +WHERE slot = $2 AND tx_index = $3 +` + +type UpdateDecryptedTXParams struct { + TxStatus TxStatusVal + Slot int64 + TxIndex int64 +} + +func (q *Queries) UpdateDecryptedTX(ctx context.Context, arg UpdateDecryptedTXParams) error { + _, err := q.db.Exec(ctx, updateDecryptedTX, arg.TxStatus, arg.Slot, arg.TxIndex) + return err +} diff --git a/internal/data/models.sqlc.gen.go b/internal/data/models.sqlc.gen.go index 71b1645..67c7f3d 100644 --- a/internal/data/models.sqlc.gen.go +++ b/internal/data/models.sqlc.gen.go @@ -14,8 +14,13 @@ import ( type TxStatusVal string const ( - TxStatusValIncluded TxStatusVal = "included" - TxStatusValNotincluded TxStatusVal = "not included" + TxStatusValIncluded TxStatusVal = "included" + TxStatusValNotincluded TxStatusVal = "not included" + TxStatusValNotdecrypted TxStatusVal = "not decrypted" + TxStatusValInvalid TxStatusVal = "invalid" + TxStatusValPending TxStatusVal = "pending" + TxStatusValShieldedinclusion TxStatusVal = "shielded inclusion" + TxStatusValUnshieldedinclusion TxStatusVal = "unshielded inclusion" ) func (e *TxStatusVal) Scan(src interface{}) error { diff --git a/internal/data/sql/queries/metrics.sql b/internal/data/sql/queries/metrics.sql index 0b82da8..5e2ca13 100644 --- a/internal/data/sql/queries/metrics.sql +++ b/internal/data/sql/queries/metrics.sql @@ -159,3 +159,7 @@ INSERT INTO proposer_duties (public_key, validator_index, slot) SELECT * FROM data ON CONFLICT DO NOTHING; +-- name: UpdateDecryptedTX :exec +UPDATE decrypted_tx +SET tx_status = $1 +WHERE slot = $2 AND tx_index = $3; \ No newline at end of file diff --git a/internal/metrics/tx_mapper_db.go b/internal/metrics/tx_mapper_db.go index 233429c..cf1ea82 100644 --- a/internal/metrics/tx_mapper_db.go +++ b/internal/metrics/tx_mapper_db.go @@ -5,10 +5,10 @@ import ( "encoding/hex" "fmt" "math" - "math/big" "sync" "time" + "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" @@ -20,6 +20,7 @@ import ( validatorRegistryBindings "github.com/shutter-network/gnosh-contracts/gnoshcontracts/validatorregistry" metricsCommon "github.com/shutter-network/gnosh-metrics/common" dbTypes "github.com/shutter-network/gnosh-metrics/common/database" + "github.com/shutter-network/gnosh-metrics/common/utils" "github.com/shutter-network/gnosh-metrics/internal/data" gnosis "github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/beaconapiclient" @@ -28,13 +29,17 @@ import ( blst "github.com/supranational/blst/bindings/go" ) +const ReceiptWaitTimeout = 1 * time.Hour + type TxMapperDB struct { - db *pgxpool.Pool - dbQuery *data.Queries - config *metricsCommon.Config - ethClient *ethclient.Client - beaconAPIClient *beaconapiclient.Client - chainID int64 + db *pgxpool.Pool + dbQuery *data.Queries + config *metricsCommon.Config + ethClient *ethclient.Client + beaconAPIClient *beaconapiclient.Client + chainID int64 + genesisTimestamp uint64 + slotDuration uint64 } func NewTxMapperDB( @@ -44,14 +49,18 @@ func NewTxMapperDB( ethClient *ethclient.Client, beaconAPIClient *beaconapiclient.Client, chainID int64, + genesisTimestamp uint64, + slotDuration uint64, ) TxMapper { return &TxMapperDB{ - db: db, - dbQuery: data.New(db), - config: config, - ethClient: ethClient, - beaconAPIClient: beaconAPIClient, - chainID: chainID, + db: db, + dbQuery: data.New(db), + config: config, + ethClient: ethClient, + beaconAPIClient: beaconAPIClient, + chainID: chainID, + genesisTimestamp: genesisTimestamp, + slotDuration: slotDuration, } } @@ -120,20 +129,16 @@ func (tm *TxMapperDB) AddDecryptionKeysAndMessages( if err != nil { return err } - totalDecKeysAndMessages := len(decKeysAndMessages.Keys) - block, err := qtx.QueryBlockFromSlot(ctx, decKeysAndMessages.Slot) + totalDecKeysAndMessages := len(decKeysAndMessages.Keys) + err = tx.Commit(ctx) if err != nil { - log.Debug().Int64("slot", decKeysAndMessages.Slot).Msg("block not found in AddDecryptedTxFromDecryptionKeys") - err = tx.Commit(ctx) - if err != nil { - log.Err(err).Msg("unable to commit db transaction") - return err - } - for i := 0; i < totalDecKeysAndMessages; i++ { - metricsDecKeyReceived.Inc() - } - return nil + log.Err(err).Msg("unable to commit db transaction") + return err + } + + for i := 0; i < totalDecKeysAndMessages; i++ { + metricsDecKeyReceived.Inc() } dkam := make([]*DecKeyAndMessage, totalDecKeysAndMessages) @@ -155,22 +160,11 @@ func (tm *TxMapperDB) AddDecryptionKeysAndMessages( } err = tm.processTransactionExecution(ctx, &TxExecution{ DecKeysAndMessages: dkam, - BlockNumber: block.BlockNumber, }) if err != nil { log.Err(err).Int64("slot", decKeysAndMessages.Slot).Msg("failed to process transaction execution") return err } - - err = tx.Commit(ctx) - if err != nil { - log.Err(err).Msg("unable to commit db transaction") - return err - } - - for i := 0; i < totalDecKeysAndMessages; i++ { - metricsDecKeyReceived.Inc() - } return nil } @@ -193,57 +187,13 @@ func (tm *TxMapperDB) AddBlock( ctx context.Context, b *data.Block, ) error { - tx, err := tm.db.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - qtx := tm.dbQuery.WithTx(tx) - err = qtx.CreateBlock(ctx, data.CreateBlockParams{ + err := tm.dbQuery.CreateBlock(ctx, data.CreateBlockParams{ BlockHash: b.BlockHash, BlockNumber: b.BlockNumber, BlockTimestamp: b.BlockTimestamp, Slot: b.Slot, }) - if err != nil { - return err - } - decKeysAndMessages, err := qtx.QueryDecryptionKeysAndMessage(ctx, b.Slot) - if err != nil { - return err - } - - totalDecKeysAndMessages := len(decKeysAndMessages) - - if totalDecKeysAndMessages == 0 { - log.Debug().Int64("slot", b.Slot).Msg("no decryption keys received yet") - return tx.Commit(ctx) - } - - dkam := make([]*DecKeyAndMessage, totalDecKeysAndMessages) - for index, elem := range decKeysAndMessages { - dkam[index] = &DecKeyAndMessage{ - Slot: elem.Slot.Int64, - TxPointer: elem.TxPointer.Int64, - Eon: elem.Eon.Int64, - Key: elem.Key, - IdentityPreimage: elem.IdentityPreimage, - KeyIndex: elem.KeyIndex, - DecryptionKeyID: elem.DecryptionKeyID, - } - } - if len(dkam) > 0 { - dkam = dkam[1:] - } - err = tm.processTransactionExecution(ctx, &TxExecution{ - DecKeysAndMessages: dkam, - BlockNumber: b.BlockNumber, - }) - if err != nil { - log.Err(err).Int64("slot", b.Slot).Msg("failed to process transaction execution") - return err - } - return tx.Commit(ctx) + return err } func (tm *TxMapperDB) QueryBlockNumberFromValidatorRegistryEventsSyncedUntil(ctx context.Context) (int64, error) { @@ -439,78 +389,120 @@ func (tm *TxMapperDB) processTransactionExecution( } slot := te.DecKeysAndMessages[0].Slot - detailedBlock, err := tm.ethClient.BlockByNumber(ctx, big.NewInt(te.BlockNumber)) - if err != nil { - return err - } - - var blockTxHashes []common.Hash - for _, tx := range detailedBlock.Transactions() { - blockTxHashes = append(blockTxHashes, tx.Hash()) - } - - log.Debug().Int64("block number", detailedBlock.Number().Int64()). - Int64("slot", slot). - Msg("block info while processing decrypted transactions") for index, txSubEvent := range txSubEvents { - decryptedTxHash, err := getDecryptedTXHash(txSubEvent, identityPreimageToDecKeyAndMsg) - if err != nil { - log.Err(err).Msg("error while trying to get decrypted tx hash") - } decryptionKeyID, err := getDecryptionKeyID(txSubEvent, identityPreimageToDecKeyAndMsg) if err != nil { log.Err(err).Msg("error while trying to retrieve decryption key ID") + continue } - if index < len(blockTxHashes) { - log.Debug(). - Str("decryptedTXHash", decryptedTxHash.Hex()). - Str("blockTxHash", blockTxHashes[index].Hex()). - Bool("matches", decryptedTxHash.Cmp(blockTxHashes[index]) == 0). - Msg("comparing tx hash") - if decryptedTxHash.Cmp(blockTxHashes[index]) == 0 { - // it means we have it in correct order and the transaction is correct - err := tm.dbQuery.CreateDecryptedTX(ctx, data.CreateDecryptedTXParams{ - Slot: slot, - TxIndex: txSubEvent.TxIndex, - TxHash: decryptedTxHash.Bytes(), - TxStatus: data.TxStatusValIncluded, - DecryptionKeyID: decryptionKeyID, - TransactionSubmittedEventID: txSubEvent.ID, - }) - if err != nil { - return err - } - } else { - // something went wrong case - err := tm.dbQuery.CreateDecryptedTX(ctx, data.CreateDecryptedTXParams{ - Slot: slot, - TxIndex: txSubEvent.TxIndex, - TxHash: decryptedTxHash.Bytes(), - TxStatus: data.TxStatusValNotincluded, - DecryptionKeyID: decryptionKeyID, - TransactionSubmittedEventID: txSubEvent.ID, - }) - if err != nil { - return err - } + decryptedTx, err := getDecryptedTXHash(txSubEvent, identityPreimageToDecKeyAndMsg) + if err != nil { + log.Err(err).Msg("error while trying to get decrypted tx hash") + err := tm.dbQuery.CreateDecryptedTX(ctx, data.CreateDecryptedTXParams{ + Slot: slot, + TxIndex: txSubEvent.TxIndex, + TxHash: common.Hash{}.Bytes(), + TxStatus: data.TxStatusValNotdecrypted, + DecryptionKeyID: decryptionKeyID, + TransactionSubmittedEventID: txSubEvent.ID, + }) + if err != nil { + log.Err(err).Msg("failed to create decrypted tx") } + continue + } + // send tx to public mempool since keys are already public, increases inclusion time + err = tm.ethClient.SendTransaction(context.Background(), decryptedTx) + if err != nil { + log.Err(err).Msg("failed to send transaction") + err := tm.dbQuery.CreateDecryptedTX(ctx, data.CreateDecryptedTXParams{ + Slot: slot, + TxIndex: txSubEvent.TxIndex, + TxHash: decryptedTx.Hash().Bytes(), + TxStatus: data.TxStatusValInvalid, + DecryptionKeyID: decryptionKeyID, + TransactionSubmittedEventID: txSubEvent.ID, + }) + if err != nil { + log.Err(err).Msg("failed to create decrypted tx") + } + continue } else { - // Mark remaining txSubEvents as missing - log.Debug().Str("txHash", decryptedTxHash.Hex()). - Msg("decryptedTXHash (missing block transaction)") + log.Info().Hex("tx-hash", decryptedTx.Hash().Bytes()).Msg("transaction sent") err := tm.dbQuery.CreateDecryptedTX(ctx, data.CreateDecryptedTXParams{ Slot: slot, TxIndex: txSubEvent.TxIndex, - TxHash: decryptedTxHash.Bytes(), - TxStatus: data.TxStatusValNotincluded, + TxHash: decryptedTx.Hash().Bytes(), + TxStatus: data.TxStatusValPending, DecryptionKeyID: decryptionKeyID, TransactionSubmittedEventID: txSubEvent.ID, }) if err != nil { - return err + log.Err(err).Msg("failed to create decrypted tx") + continue } } + + // Fire off a goroutine to wait for the transaction receipt + go func(ctx context.Context, index int, txHash common.Hash, txIndex int64, slot int64, decryptionKeyID int64, eventID int64) { + // Wait for the receipt with a timeout + receipt, err := tm.waitForReceiptWithTimeout(ctx, txHash, ReceiptWaitTimeout) + if err != nil { + log.Err(err).Msgf("failed to get receipt for transaction %s", txHash.Hex()) + // update status to not included + err := tm.dbQuery.UpdateDecryptedTX(ctx, data.UpdateDecryptedTXParams{ + TxStatus: data.TxStatusValNotincluded, + Slot: slot, + TxIndex: txIndex, + }) + if err != nil { + log.Err(err).Msg("failed to update decrypted tx") + return + } + } else { + // receipt found + log.Info().Hex("tx-hash", receipt.TxHash.Bytes()). + Uint64("receipt-status", receipt.Status). + Msg("transaction receipt found") + + block, err := tm.ethClient.BlockByNumber(ctx, receipt.BlockNumber) + if err != nil { + log.Err(err).Uint64("block-number", receipt.BlockNumber.Uint64()).Msg("failed to retrieve block") + return + } + + inclusionSlot := utils.GetSlotForBlock(block.Header().Time, tm.genesisTimestamp, tm.slotDuration) + txStatus := data.TxStatusValShieldedinclusion + + log.Info().Uint("tx-index", receipt.TransactionIndex). + Uint64("inclusion-slot", inclusionSlot). + Msg("receipt data") + + log.Info().Int("index", index). + Int64("inclusion-slot", slot). + Msg("local data") + + if receipt.TransactionIndex != uint(index) { + log.Info().Uint("tx-index", receipt.TransactionIndex).Msg("transaction index mismatch") + txStatus = data.TxStatusValUnshieldedinclusion + } + if inclusionSlot != uint64(slot) { + log.Info().Int64("slot", slot).Msg("transaction slot mismatch") + txStatus = data.TxStatusValUnshieldedinclusion + } + + err = tm.dbQuery.UpdateDecryptedTX(ctx, data.UpdateDecryptedTXParams{ + TxStatus: txStatus, + TxIndex: txIndex, + Slot: slot, + }) + if err != nil { + log.Err(err).Msg("failed to update decrypted tx") + return + } + } + }(ctx, index, decryptedTx.Hash(), txSubEvent.TxIndex, slot, decryptionKeyID, txSubEvent.ID) } return nil } @@ -644,17 +636,17 @@ func computeIdentity(prefix []byte, sender common.Address) []byte { func getDecryptedTXHash( txSubEvent data.TransactionSubmittedEvent, identityPreimageToDecKeyAndMsg map[string]*DecKeyAndMessage, -) (common.Hash, error) { +) (*types.Transaction, error) { identityPreimage := computeIdentity(txSubEvent.IdentityPrefix, common.BytesToAddress(txSubEvent.Sender)) dkam, ok := identityPreimageToDecKeyAndMsg[hex.EncodeToString(identityPreimage)] if !ok { - return common.Hash{}, fmt.Errorf("identity preimage not found %s", hex.EncodeToString(identityPreimage)) + return nil, fmt.Errorf("identity preimage not found %s", hex.EncodeToString(identityPreimage)) } tx, err := decryptTransaction(dkam.Key, txSubEvent.EncryptedTransaction) if err != nil { - return common.Hash{}, err + return nil, err } - return tx.Hash(), nil + return tx, nil } func getDecryptionKeyID( @@ -692,3 +684,40 @@ func decryptTransaction(key []byte, encrypted []byte) (*types.Transaction, error } return tx, nil } + +// waitForReceiptWithTimeout waits for a transaction receipt with a provided timeout. +func (tm *TxMapperDB) waitForReceiptWithTimeout(ctx context.Context, txHash common.Hash, receiptWaitTimeout time.Duration) (*types.Receipt, error) { + ctx, cancel := context.WithTimeout(ctx, receiptWaitTimeout) + defer cancel() + + // wait for the transaction receipt + receipt, err := tm.waitForReceipt(ctx, txHash) + if err != nil { + return nil, fmt.Errorf("failed to get receipt for transaction %s: %w", txHash.Hex(), err) + } + return receipt, nil +} + +// waitForReceipt polls the Ethereum network for the transaction receipt until it's available or the context is canceled. +func (tm *TxMapperDB) waitForReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error) { + for { + // check if the context has been canceled or timed out + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + + // query for the transaction receipt + receipt, err := tm.ethClient.TransactionReceipt(ctx, txHash) + if err == ethereum.NotFound { + // If the receipt is not found, continue polling + time.Sleep(3 * time.Second) + continue + } else if err != nil { + return nil, err + } + + return receipt, nil + } +} diff --git a/internal/metrics/types.go b/internal/metrics/types.go index ed78606..7863368 100644 --- a/internal/metrics/types.go +++ b/internal/metrics/types.go @@ -44,7 +44,7 @@ type DecKeyAndMessage struct { } type TxExecution struct { - BlockNumber int64 + // BlockNumber int64 DecKeysAndMessages []*DecKeyAndMessage } diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 805444c..36eaf89 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -97,6 +97,8 @@ func (s *Scheduler) Start(ctx context.Context, runner service.Runner) error { ethClient, beaconAPIClient, chainID.Int64(), + GenesisTimestamp, + SlotDuration, ) validatorStatusScheduler := NewValidatorStatusScheduler(txMapper) diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 612c7f1..65a5b46 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -98,6 +98,8 @@ func (w *Watcher) Start(ctx context.Context, runner service.Runner) error { ethClient, beaconAPIClient, chainID.Int64(), + GenesisTimestamp, + SlotDuration, ) blocksWatcher := NewBlocksWatcher(w.config, blocksChannel, ethClient) diff --git a/migrations/20241002011333_add_tx_status_val_type.sql b/migrations/20241002011333_add_tx_status_val_type.sql new file mode 100644 index 0000000..b12a869 --- /dev/null +++ b/migrations/20241002011333_add_tx_status_val_type.sql @@ -0,0 +1,12 @@ +-- +goose Up +-- +goose StatementBegin +ALTER TYPE tx_status_val ADD VALUE 'not decrypted'; +ALTER TYPE tx_status_val ADD VALUE 'invalid'; +ALTER TYPE tx_status_val ADD VALUE 'pending'; +ALTER TYPE tx_status_val ADD VALUE 'shielded inclusion'; +ALTER TYPE tx_status_val ADD VALUE 'unshielded inclusion'; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +-- +goose StatementEnd diff --git a/migrations/20241002075029_update_tx_status.sql b/migrations/20241002075029_update_tx_status.sql new file mode 100644 index 0000000..0816876 --- /dev/null +++ b/migrations/20241002075029_update_tx_status.sql @@ -0,0 +1,11 @@ +-- +goose Up +-- +goose StatementBegin +UPDATE decrypted_tx +SET tx_status = 'shielded inclusion' +WHERE tx_status = 'included'; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +SELECT 'down SQL query'; +-- +goose StatementEnd From fe7d1d8dbbc3ae31b81d3e079adb33cdb9e8a419 Mon Sep 17 00:00:00 2001 From: faheelsattar Date: Thu, 3 Oct 2024 21:44:31 +0500 Subject: [PATCH 2/3] add vals in test --- tests/init_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/init_test.go b/tests/init_test.go index 55ef684..33db0e7 100644 --- a/tests/init_test.go +++ b/tests/init_test.go @@ -2,6 +2,7 @@ package tests import ( "context" + "math/rand" "path" "runtime" "testing" @@ -40,6 +41,6 @@ func (s *TestMetricsSuite) SetupSuite() { migrationsPath := curDir + "/../migrations" s.testDB = common.SetupTestDatabase(migrationsPath) - s.txMapperDB = metrics.NewTxMapperDB(ctx, s.testDB.DbInstance, &common.Config{}, ðclient.Client{}, &beaconapiclient.Client{}, 1) + s.txMapperDB = metrics.NewTxMapperDB(ctx, s.testDB.DbInstance, &common.Config{}, ðclient.Client{}, &beaconapiclient.Client{}, 1, rand.Uint64(), rand.Uint64()) s.dbQuery = data.New(s.testDB.DbInstance) } From bb77c9b19fa8f9feb1f57790e5a464cab4353572 Mon Sep 17 00:00:00 2001 From: faheelsattar Date: Fri, 4 Oct 2024 21:06:13 +0500 Subject: [PATCH 3/3] rename func --- internal/metrics/tx_mapper_db.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/metrics/tx_mapper_db.go b/internal/metrics/tx_mapper_db.go index cf1ea82..0df892c 100644 --- a/internal/metrics/tx_mapper_db.go +++ b/internal/metrics/tx_mapper_db.go @@ -396,7 +396,7 @@ func (tm *TxMapperDB) processTransactionExecution( log.Err(err).Msg("error while trying to retrieve decryption key ID") continue } - decryptedTx, err := getDecryptedTXHash(txSubEvent, identityPreimageToDecKeyAndMsg) + decryptedTx, err := getDecryptedTX(txSubEvent, identityPreimageToDecKeyAndMsg) if err != nil { log.Err(err).Msg("error while trying to get decrypted tx hash") err := tm.dbQuery.CreateDecryptedTX(ctx, data.CreateDecryptedTXParams{ @@ -633,7 +633,7 @@ func computeIdentity(prefix []byte, sender common.Address) []byte { return imageBytes } -func getDecryptedTXHash( +func getDecryptedTX( txSubEvent data.TransactionSubmittedEvent, identityPreimageToDecKeyAndMsg map[string]*DecKeyAndMessage, ) (*types.Transaction, error) {