add unit test

memiavl/db.go

@@ -80,6 +80,9 @@ type DB struct {
 	mtx sync.Mutex
 	// worker goroutine IdleTimeout = 5s
 	snapshotWriterPool *pond.WorkerPool
+
+	// reusable write batch
+	wbatch wal.Batch
 }
 
 type Options struct {
@@ -556,21 +559,18 @@ func (db *DB) Commit() (int64, error) {
 			// async wal writing
 			db.walChan <- &entry
 		} else {
-			bz, err := entry.data.Marshal()
+			lastIndex, err := db.wal.LastIndex()
 			if err != nil {
 				return 0, err
 			}
 
-			lastIndex, err := db.wal.LastIndex()
-			if err != nil {
+			db.wbatch.Clear()
+			if err := writeEntry(&db.wbatch, db.logger, lastIndex, &entry); err != nil {
 				return 0, err
 			}
-			if entry.index < lastIndex+1 {
-				db.logger.Error("commit old version idempotently", "expected", lastIndex+1, "actual", entry.index)
-			} else {
-				if err := db.wal.Write(entry.index, bz); err != nil {
-					return 0, err
-				}
+
+			if err := db.wal.WriteBatch(&db.wbatch); err != nil {
+				return 0, err
 			}
 		}
 	}
@@ -606,19 +606,11 @@ func (db *DB) initAsyncCommit() {
 				return
 			}
 
-			for i, entry := range entries {
-				bz, err := entry.data.Marshal()
-				if err != nil {
+			for _, entry := range entries {
+				if err := writeEntry(&batch, db.logger, lastIndex, entry); err != nil {
 					walQuit <- err
 					return
 				}
-
-				if entry.index < lastIndex+uint64(i+1) {
-					db.logger.Error("commit old version idempotently", "expected", lastIndex+uint64(i+1), "actual", entry.index)
-					continue
-				} else {
-					batch.Write(entry.index, bz)
-				}
 			}
 
 			if err := db.wal.WriteBatch(&batch); err != nil {
@@ -1114,3 +1106,17 @@ func channelBatchRecv[T any](ch <-chan *T) []*T {
 
 	return result
 }
+
+func writeEntry(batch *wal.Batch, logger Logger, lastIndex uint64, entry *walEntry) error {
+	bz, err := entry.data.Marshal()
+	if err != nil {
+		return err
+	}
+
+	if entry.index <= lastIndex {
+		logger.Error("commit old version idempotently", "lastIndex", lastIndex, "version", entry.index)
+	} else {
+		batch.Write(entry.index, bz)
+	}
+	return nil
+}

memiavl/db_test.go

@@ -3,6 +3,7 @@ package memiavl
 import (
 	"encoding/hex"
 	"errors"
+	fmt "fmt"
 	"os"
 	"path/filepath"
 	"runtime/debug"
@@ -497,3 +498,55 @@ func TestRepeatedApplyChangeSet(t *testing.T) {
 	})
 	require.NoError(t, err)
 }
+
+func TestIdempotentWrite(t *testing.T) {
+	dir := t.TempDir()
+	db, err := Load(dir, Options{CreateIfMissing: true, InitialStores: []string{"test1", "test2"}})
+	require.NoError(t, err)
+
+	// generate some data into db
+	var changes [][]*NamedChangeSet
+	for i := 0; i < 10; i++ {
+		cs := []*NamedChangeSet{
+			{
+				Name:      "test1",
+				Changeset: ChangeSet{Pairs: mockKVPairs("hello", fmt.Sprintf("world%d", i))},
+			},
+			{
+				Name:      "test2",
+				Changeset: ChangeSet{Pairs: mockKVPairs("hello", fmt.Sprintf("world%d", i))},
+			},
+		}
+		changes = append(changes, cs)
+	}
+
+	for _, cs := range changes {
+		require.NoError(t, db.ApplyChangeSets(cs))
+		_, err := db.Commit()
+		require.NoError(t, err)
+	}
+
+	commitInfo := *db.LastCommitInfo()
+	require.NoError(t, db.Close())
+
+	// reload db from disk at an intermidiate version
+	db, err = Load(dir, Options{TargetVersion: 5})
+	require.NoError(t, err)
+
+	// replay some random writes to reach same version
+	for i := 0; i < 5; i++ {
+		require.NoError(t, db.ApplyChangeSets(changes[i+5]))
+		_, err := db.Commit()
+		require.NoError(t, err)
+	}
+
+	// it should reach same result
+	require.Equal(t, commitInfo, *db.LastCommitInfo())
+
+	require.NoError(t, db.Close())
+
+	// reload db again, it should reach same result
+	db, err = Load(dir, Options{})
+	require.NoError(t, err)
+	require.Equal(t, commitInfo, *db.LastCommitInfo())
+}