backend: add UnsafeForEach to BatchTx

Useful for efficiently iterating over an entire bucket.

storage/backend/batch_tx.go

@@ -31,6 +31,7 @@ type BatchTx interface {
 	UnsafeSeqPut(bucketName []byte, key []byte, value []byte)
 	UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte)
 	UnsafeDelete(bucketName []byte, key []byte)
+	UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error
 	Commit()
 	CommitAndStop()
 }
@@ -122,6 +123,11 @@ func (t *batchTx) UnsafeDelete(bucketName []byte, key []byte) {
 	t.pending++
 }
 
+// UnsafeForEach must be called holding the lock on the tx.
+func (t *batchTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
+	return t.tx.Bucket(bucketName).ForEach(visitor)
+}
+
 // Commit commits a previous tx and begins a new writable one.
 func (t *batchTx) Commit() {
 	t.Lock()

storage/kvstore_test.go

@@ -584,6 +584,9 @@ func (b *fakeBatchTx) UnsafeRange(bucketName []byte, key, endKey []byte, limit i
 func (b *fakeBatchTx) UnsafeDelete(bucketName []byte, key []byte) {
 	b.Recorder.Record(testutil.Action{Name: "delete", Params: []interface{}{bucketName, key}})
 }
+func (b *fakeBatchTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
+	return nil
+}
 func (b *fakeBatchTx) Commit()        {}
 func (b *fakeBatchTx) CommitAndStop() {}