libs/db: close batch (#3397)

ClevelDB requires closing when WriteBatch is no longer needed, https://godoc.org/github.com/jmhodges/levigo#WriteBatch.Close

Fixes the memory leak in https://github.com/cosmos/cosmos-sdk/issues/3842

libs/db/c_level_db.go

@@ -187,6 +187,11 @@ func (mBatch *cLevelDBBatch) WriteSync() {
 	}
 }
 
+// Implements Batch.
+func (mBatch *cLevelDBBatch) Close() {
+	mBatch.batch.Close()
+}
+
 //----------------------------------------
 // Iterator
 // NOTE This is almost identical to db/go_level_db.Iterator

libs/db/debug_db.go

@@ -250,3 +250,8 @@ func (dbch debugBatch) WriteSync() {
 	fmt.Printf("%v.batch.WriteSync()\n", dbch.label)
 	dbch.bch.WriteSync()
 }
+
+// Implements Batch.
+func (dbch debugBatch) Close() {
+	dbch.bch.Close()
+}

libs/db/go_level_db.go

@@ -184,6 +184,10 @@ func (mBatch *goLevelDBBatch) WriteSync() {
 	}
 }
 
+// Implements Batch.
+// Close is no-op for goLevelDBBatch.
+func (mBatch *goLevelDBBatch) Close() {}
+
 //----------------------------------------
 // Iterator
 // NOTE This is almost identical to db/c_level_db.Iterator

libs/db/mem_batch.go

@@ -46,6 +46,10 @@ func (mBatch *memBatch) WriteSync() {
 	mBatch.write(true)
 }
 
+func (mBatch *memBatch) Close() {
+	mBatch.ops = nil
+}
+
 func (mBatch *memBatch) write(doSync bool) {
 	if mtx := mBatch.db.Mutex(); mtx != nil {
 		mtx.Lock()

libs/db/prefix_db.go

@@ -248,6 +248,10 @@ func (pb prefixBatch) WriteSync() {
 	pb.source.WriteSync()
 }
 
+func (pb prefixBatch) Close() {
+	pb.source.Close()
+}
+
 //----------------------------------------
 // prefixIterator
 

libs/db/remotedb/grpcdb/server.go

@@ -180,6 +180,7 @@ func (s *server) BatchWriteSync(c context.Context, b *protodb.Batch) (*protodb.N
 
 func (s *server) batchWrite(c context.Context, b *protodb.Batch, sync bool) (*protodb.Nothing, error) {
 	bat := s.db.NewBatch()
+	defer bat.Close()
 	for _, op := range b.Ops {
 		switch op.Type {
 		case protodb.Operation_SET:

libs/db/remotedb/remotedb.go

@@ -260,3 +260,7 @@ func (bat *batch) WriteSync() {
 		panic(fmt.Sprintf("RemoteDB.BatchWriteSync: %v", err))
 	}
 }
+
+func (bat *batch) Close() {
+	bat.ops = nil
+}

libs/db/types.go

@@ -57,10 +57,12 @@ type DB interface {
 //----------------------------------------
 // Batch
 
+// Batch Close must be called when the program no longer needs the object.
 type Batch interface {
 	SetDeleter
 	Write()
 	WriteSync()
+	Close()
 }
 
 type SetDeleter interface {

lite/dbprovider.go

@@ -54,6 +54,7 @@ func (dbp *DBProvider) SaveFullCommit(fc FullCommit) error {
 
 	dbp.logger.Info("DBProvider.SaveFullCommit()...", "fc", fc)
 	batch := dbp.db.NewBatch()
+	defer batch.Close()
 
 	// Save the fc.validators.
 	// We might be overwriting what we already have, but

state/txindex/kv/kv.go

@@ -78,6 +78,7 @@ func (txi *TxIndex) Get(hash []byte) (*types.TxResult, error) {
 // AddBatch indexes a batch of transactions using the given list of tags.
 func (txi *TxIndex) AddBatch(b *txindex.Batch) error {
 	storeBatch := txi.store.NewBatch()
+	defer storeBatch.Close()
 
 	for _, result := range b.Ops {
 		hash := result.Tx.Hash()
@@ -109,6 +110,7 @@ func (txi *TxIndex) AddBatch(b *txindex.Batch) error {
 // Index indexes a single transaction using the given list of tags.
 func (txi *TxIndex) Index(result *types.TxResult) error {
 	b := txi.store.NewBatch()
+	defer b.Close()
 
 	hash := result.Tx.Hash()