deep-stack/ipld-eth-server
4
0
Fork 0
forked from cerc-io/ipld-eth-server
Code Pull Requests Activity

clean up db transaction rollback/commit handling

Browse Source
This commit is contained in:
Ian Norden
2020-05-01 17:00:00 -05:00
parent 44e7f0570f
commit 9b0ded692b
14 changed files with 294 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/super_node/btc/cid_retriever.go
+41 -23
View File
@@ -65,17 +65,26 @@ func (ecr *CIDRetriever) Retrieve(filter shared.SubscriptionSettings, blockNumbe
return nil, true, fmt.Errorf("btc retriever expected filter type %T got %T", &SubscriptionSettings{}, filter)
}
log.Debug("retrieving cids")
// Begin new db tx
tx, err := ecr.db.Beginx()
if err != nil {
return nil, true, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
// Retrieve cached header CIDs
headers, err := ecr.RetrieveHeaderCIDs(tx, blockNumber)
if err != nil {
if err := tx.Rollback(); err != nil {
log.Error(err)
}
log.Error("header cid retrieval error")
return nil, true, err
}
@@ -92,9 +101,6 @@ func (ecr *CIDRetriever) Retrieve(filter shared.SubscriptionSettings, blockNumbe
if !streamFilter.TxFilter.Off {
cw.Transactions, err = ecr.RetrieveTxCIDs(tx, streamFilter.TxFilter, header.ID)
if err != nil {
if err := tx.Rollback(); err != nil {
log.Error(err)
}
log.Error("transaction cid retrieval error")
return nil, true, err
}
@@ -105,7 +111,7 @@ func (ecr *CIDRetriever) Retrieve(filter shared.SubscriptionSettings, blockNumbe
cws[i] = cw
}
return cws, empty, tx.Commit()
return cws, empty, err
}
// RetrieveHeaderCIDs retrieves and returns all of the header cids at the provided blockheight
@@ -202,41 +208,57 @@ func (ecr *CIDRetriever) RetrieveGapsInData(validationLevel int) ([]shared.Gap,
// RetrieveBlockByHash returns all of the CIDs needed to compose an entire block, for a given block hash
func (ecr *CIDRetriever) RetrieveBlockByHash(blockHash common.Hash) (HeaderModel, []TxModel, error) {
log.Debug("retrieving block cids for block hash ", blockHash.String())
// Begin new db tx
tx, err := ecr.db.Beginx()
if err != nil {
return HeaderModel{}, nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
headerCID, err := ecr.RetrieveHeaderCIDByHash(tx, blockHash)
if err != nil {
if err := tx.Rollback(); err != nil {
log.Error(err)
}
log.Error("header cid retrieval error")
return HeaderModel{}, nil, err
}
txCIDs, err := ecr.RetrieveTxCIDsByHeaderID(tx, headerCID.ID)
if err != nil {
if err := tx.Rollback(); err != nil {
log.Error(err)
}
log.Error("tx cid retrieval error")
return HeaderModel{}, nil, err
}
return headerCID, txCIDs, tx.Commit()
return headerCID, txCIDs, err
}
// RetrieveBlockByNumber returns all of the CIDs needed to compose an entire block, for a given block number
func (ecr *CIDRetriever) RetrieveBlockByNumber(blockNumber int64) (HeaderModel, []TxModel, error) {
log.Debug("retrieving block cids for block number ", blockNumber)
// Begin new db tx
tx, err := ecr.db.Beginx()
if err != nil {
return HeaderModel{}, nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
headerCID, err := ecr.RetrieveHeaderCIDs(tx, blockNumber)
if err != nil {
if err := tx.Rollback(); err != nil {
log.Error(err)
}
log.Error("header cid retrieval error")
return HeaderModel{}, nil, err
}
@@ -245,13 +267,9 @@ func (ecr *CIDRetriever) RetrieveBlockByNumber(blockNumber int64) (HeaderModel,
}
txCIDs, err := ecr.RetrieveTxCIDsByHeaderID(tx, headerCID[0].ID)
if err != nil {
if err := tx.Rollback(); err != nil {
log.Error(err)
}
log.Error("tx cid retrieval error")
return HeaderModel{}, nil, err
}
return headerCID[0], txCIDs, tx.Commit()
return headerCID[0], txCIDs, err
}
// RetrieveHeaderCIDByHash returns the header for the given block hash
pkg/super_node/btc/cleaner.go
+2 -6
View File
@@ -50,9 +50,7 @@ func (c *Cleaner) ResetValidation(rngs [][2]uint64) error {
SET times_validated = 0
WHERE block_number BETWEEN $1 AND $2`
if _, err := tx.Exec(pgStr, rng[0], rng[1]); err != nil {
if err := tx.Rollback(); err != nil {
logrus.Error(err)
}
shared.Rollback(tx)
return err
}
}
@@ -68,9 +66,7 @@ func (c *Cleaner) Clean(rngs [][2]uint64, t shared.DataType) error {
for _, rng := range rngs {
logrus.Infof("btc db cleaner cleaning up block range %d to %d", rng[0], rng[1])
if err := c.clean(tx, rng, t); err != nil {
if err := tx.Rollback(); err != nil {
logrus.Error(err)
}
shared.Rollback(tx)
return err
}
}
pkg/super_node/btc/indexer.go
+16 -3
View File
@@ -43,20 +43,33 @@ func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
if !ok {
return fmt.Errorf("btc indexer expected cids type %T got %T", &CIDPayload{}, cids)
}
// Begin new db tx
tx, err := in.db.Beginx()
if err != nil {
return err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
headerID, err := in.indexHeaderCID(tx, cidWrapper.HeaderCID)
if err != nil {
logrus.Error("btc indexer error when indexing header")
return err
}
if err := in.indexTransactionCIDs(tx, cidWrapper.TransactionCIDs, headerID); err != nil {
err = in.indexTransactionCIDs(tx, cidWrapper.TransactionCIDs, headerID)
if err != nil {
logrus.Error("btc indexer error when indexing transactions")
return err
}
return tx.Commit()
return err
}
func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel) (int64, error) {
pkg/super_node/btc/ipld_pg_fetcher.go
+13 -4
View File
@@ -49,22 +49,31 @@ func (f *IPLDPGFetcher) Fetch(cids shared.CIDsForFetching) (shared.IPLDs, error)
log.Debug("fetching iplds")
iplds := IPLDs{}
iplds.BlockNumber = cidWrapper.BlockNumber
var err error
tx, err := f.db.Beginx()
if err != nil {
return nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
iplds.Header, err = f.FetchHeader(tx, cidWrapper.Header)
if err != nil {
shared.Rollback(tx)
return nil, fmt.Errorf("btc pg fetcher: header fetching error: %s", err.Error())
}
iplds.Transactions, err = f.FetchTrxs(tx, cidWrapper.Transactions)
if err != nil {
shared.Rollback(tx)
return nil, fmt.Errorf("btc pg fetcher: transaction fetching error: %s", err.Error())
}
return iplds, tx.Commit()
return iplds, err
}
// FetchHeaders fetches headers
pkg/super_node/btc/publishAndIndexer.go
+11 -8
View File
@@ -56,18 +56,26 @@ func (pub *IPLDPublisherAndIndexer) Publish(payload shared.ConvertedData) (share
if err != nil {
return nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
// Publish trie nodes
for _, node := range txTrieNodes {
if err := shared.PublishIPLD(tx, node); err != nil {
shared.Rollback(tx)
return nil, err
}
}
// Publish and index header
if err := shared.PublishIPLD(tx, headerNode); err != nil {
shared.Rollback(tx)
return nil, err
}
header := HeaderModel{
@@ -80,39 +88,34 @@ func (pub *IPLDPublisherAndIndexer) Publish(payload shared.ConvertedData) (share
}
headerID, err := pub.indexer.indexHeaderCID(tx, header)
if err != nil {
shared.Rollback(tx)
return nil, err
}
// Publish and index txs
for i, txNode := range txNodes {
if err := shared.PublishIPLD(tx, txNode); err != nil {
shared.Rollback(tx)
return nil, err
}
txModel := ipldPayload.TxMetaData[i]
txModel.CID = txNode.Cid().String()
txID, err := pub.indexer.indexTransactionCID(tx, txModel, headerID)
if err != nil {
shared.Rollback(tx)
return nil, err
}
for _, input := range txModel.TxInputs {
if err := pub.indexer.indexTxInput(tx, input, txID); err != nil {
shared.Rollback(tx)
return nil, err
}
}
for _, output := range txModel.TxOutputs {
if err := pub.indexer.indexTxOutput(tx, output, txID); err != nil {
shared.Rollback(tx)
return nil, err
}
}
}
// This IPLDPublisher does both publishing and indexing, we do not need to pass anything forward to the indexer
return nil, tx.Commit()
return nil, err
}
// Index satisfies the shared.CIDIndexer interface
pkg/super_node/btc/publishAndIndexer_test.go
-2
View File
@@ -18,7 +18,6 @@ package btc_test
import (
"bytes"
"fmt"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-ipfs-blockstore"
@@ -112,7 +111,6 @@ var _ = Describe("PublishAndIndexer", func() {
Expect(err).ToNot(HaveOccurred())
mhKey := dshelp.CidToDsKey(dc)
prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
fmt.Printf("mhKey: %s\r\n", prefixedKey)
var data []byte
err = db.Get(&data, ipfsPgGet, prefixedKey)
Expect(err).ToNot(HaveOccurred())