From e00ffc42d7d78a4f9fb6f91b1f7ff78edf7b1565 Mon Sep 17 00:00:00 2001
From: githubsands <26752232+githubsands@users.noreply.github.com>
Date: Tue, 9 Feb 2021 00:09:02 -0700
Subject: [PATCH] store: use a batch instead of individual writes in SaveBlock
 (#6018)

---
 CHANGELOG_PENDING.md |  1 +
 store/store.go       | 27 ++++++++++++++++++---------
 2 files changed, 19 insertions(+), 9 deletions(-)

diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md
index 401d1d8b6..de3fe15c1 100644
--- a/CHANGELOG_PENDING.md
+++ b/CHANGELOG_PENDING.md
@@ -57,6 +57,7 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 - [cli] \#5772 `gen_node_key` output now contains node ID (`id` field) (@melekes)
 - [blockchain/v2] \#5774 Send status request when new peer joins (@melekes)
 - [consensus] \#5792 Deprecates the `time_iota_ms` consensus parameter, to reduce the bug surface. The parameter is no longer used. (@valardragon)
+- [store] \#5888 store.SaveBlock saves using batches instead of transactions for now to improve ACID properties. This is a quick fix for underlying issues around tm-db and ACID guarantees. (@githubsands)
 - [consensus] \#5987 Remove `time_iota_ms` from consensus params. Merge `tmproto.ConsensusParams` and `abci.ConsensusParams`. (@marbar3778)
 - [types] \#5994 Reduce the use of protobuf types in core logic. (@marbar3778)
   - `ConsensusParams`, `BlockParams`, `ValidatorParams`, `EvidenceParams`, `VersionParams`, `sm.Version` and `version.Consensus` have become native types. They still utilize protobuf when being sent over the wire or written to disk.
diff --git a/store/store.go b/store/store.go
index ed335c3c8..2d7c356e8 100644
--- a/store/store.go
+++ b/store/store.go
@@ -434,6 +434,8 @@ func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, s
 		panic("BlockStore can only save a non-nil block")
 	}
 
+	batch := bs.db.NewBatch()
+
 	height := block.Height
 	hash := block.Hash()
 
@@ -450,27 +452,27 @@ func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, s
 	// complete as soon as the block meta is written.
 	for i := 0; i < int(blockParts.Total()); i++ {
 		part := blockParts.GetPart(i)
-		bs.saveBlockPart(height, i, part)
+		bs.saveBlockPart(height, i, part, batch)
 	}
 
-	// Save block meta
 	blockMeta := types.NewBlockMeta(block, blockParts)
 	pbm := blockMeta.ToProto()
 	if pbm == nil {
 		panic("nil blockmeta")
 	}
+
 	metaBytes := mustEncode(pbm)
-	if err := bs.db.Set(blockMetaKey(height), metaBytes); err != nil {
+	if err := batch.Set(blockMetaKey(height), metaBytes); err != nil {
 		panic(err)
 	}
-	if err := bs.db.Set(blockHashKey(hash), []byte(fmt.Sprintf("%d", height))); err != nil {
+
+	if err := batch.Set(blockHashKey(hash), []byte(fmt.Sprintf("%d", height))); err != nil {
 		panic(err)
 	}
 
-	// Save block commit (duplicate and separate from the Block)
 	pbc := block.LastCommit.ToProto()
 	blockCommitBytes := mustEncode(pbc)
-	if err := bs.db.Set(blockCommitKey(height-1), blockCommitBytes); err != nil {
+	if err := batch.Set(blockCommitKey(height-1), blockCommitBytes); err != nil {
 		panic(err)
 	}
 
@@ -478,19 +480,26 @@ func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, s
 	// NOTE: we can delete this at a later height
 	pbsc := seenCommit.ToProto()
 	seenCommitBytes := mustEncode(pbsc)
-	if err := bs.db.SetSync(seenCommitKey(height), seenCommitBytes); err != nil {
+	if err := batch.Set(seenCommitKey(height), seenCommitBytes); err != nil {
+		panic(err)
+	}
+
+	if err := batch.WriteSync(); err != nil {
 		panic(err)
 	}
 
+	if err := batch.Close(); err != nil {
+		panic(err)
+	}
 }
 
-func (bs *BlockStore) saveBlockPart(height int64, index int, part *types.Part) {
+func (bs *BlockStore) saveBlockPart(height int64, index int, part *types.Part, batch dbm.Batch) {
 	pbp, err := part.ToProto()
 	if err != nil {
 		panic(fmt.Errorf("unable to make part into proto: %w", err))
 	}
 	partBytes := mustEncode(pbp)
-	if err := bs.db.Set(blockPartKey(height, index), partBytes); err != nil {
+	if err := batch.Set(blockPartKey(height, index), partBytes); err != nil {
 		panic(err)
 	}
 }