indexer/database/sql/batch_tx.go

@@ -85,7 +85,6 @@ func (tx *BatchTx) BlockNumber() string {
 }
 
 func (tx *BatchTx) RollbackOnFailure(err error) {
-
 	if p := recover(); p != nil {
 		defer tx.close()
 		log.Info("panic detected before tx submission, rolling back the tx", "panic", p)

indexer/interfaces/interfaces.go

@@ -54,8 +54,11 @@ type StateDiffIndexer interface {
 
 // Batch required for indexing data atomically
 type Batch interface {
+	// Submit commits the batch transaction
 	Submit() error
+	// BlockNumber is the block number of the header this batch contains
 	BlockNumber() string
+	// RollbackOnFailure rolls back the batch transaction if the error is not nil
 	RollbackOnFailure(error)
 }
 

service.go

@@ -823,6 +823,7 @@ func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, p
 	if err != nil {
 		return err
 	}
+	defer tx.RollbackOnFailure(err)
 
 	// TODO: review/remove the need to sync here
 	var nodeMtx, ipldMtx sync.Mutex