fix connection leak (misplaced defer) and perform proper rollback on errs

Author: i-norden

params/version.go

@@ -24,7 +24,7 @@ const (
 	VersionMajor = 1                  // Major version component of the current release
 	VersionMinor = 10                 // Minor version component of the current release
 	VersionPatch = 2                  // Patch version component of the current release
-	VersionMeta  = "statediff-0.0.17" // Version metadata to append to the version string
+	VersionMeta  = "statediff-0.0.18" // Version metadata to append to the version string
 )
 
 // Version holds the textual version string.

statediff/indexer/indexer.go

@@ -73,8 +73,7 @@ type BlockTx struct {
 	dbtx        *sqlx.Tx
 	BlockNumber uint64
 	headerID    int64
-	err         error
-	Close       func() error
+	Close       func(err error) error
 }
 
 // Reporting function to run as goroutine
@@ -128,11 +127,12 @@ func (sdi *StateDiffIndexer) PushBlock(block *types.Block, receipts types.Receip
 	blocktx := BlockTx{
 		dbtx: tx,
 		// handle transaction commit or rollback for any return case
-		Close: func() error {
-			var err error
+		Close: func(err error) error {
 			if p := recover(); p != nil {
 				shared.Rollback(tx)
 				panic(p)
+			} else if err != nil {
+				shared.Rollback(tx)
 			} else {
 				tDiff := time.Since(t)
 				indexerMetrics.tStateStoreCodeProcessing.Update(tDiff)

statediff/indexer/indexer_test.go

@@ -117,7 +117,7 @@ func setup(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer tx.Close()
+	defer tx.Close(err)
 	for _, node := range mocks.StateDiffs {
 		err = ind.PushStateNode(tx, node)
 		if err != nil {

statediff/service.go

@@ -23,14 +23,13 @@ import (
 	"sync"
 	"sync/atomic"
 
-	"github.com/ethereum/go-ethereum/eth/ethconfig"
-
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/eth"
+	"github.com/ethereum/go-ethereum/eth/ethconfig"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
@@ -626,18 +625,20 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p
 	// log.Info("Writing state diff", "block height", block.Number().Uint64())
 	var totalDifficulty *big.Int
 	var receipts types.Receipts
+	var err error
+	var tx *ind.BlockTx
 	if params.IncludeTD {
 		totalDifficulty = sds.BlockChain.GetTdByHash(block.Hash())
 	}
 	if params.IncludeReceipts {
 		receipts = sds.BlockChain.GetReceiptsByHash(block.Hash())
 	}
-	tx, err := sds.indexer.PushBlock(block, receipts, totalDifficulty)
+	tx, err = sds.indexer.PushBlock(block, receipts, totalDifficulty)
+	// defer handling of commit/rollback for any return case
+	defer tx.Close(err)
 	if err != nil {
 		return err
 	}
-	// defer handling of commit/rollback for any return case
-	defer tx.Close()
 	output := func(node StateNode) error {
 		return sds.indexer.PushStateNode(tx, node)
 	}