Implement atomic batch operations API

batch_bench_test.go

@@ -0,0 +1,100 @@
+package clset_test
+
+import (
+	"fmt"
+	"testing"
+
+	"example.com/clset"
+	badgerds "github.com/ipfs/go-ds-badger4"
+)
+
+func BenchmarkPutSequential_10(b *testing.B)   { benchmarkPutSequential(b, 10) }
+func BenchmarkPutSequential_100(b *testing.B)  { benchmarkPutSequential(b, 100) }
+func BenchmarkPutSequential_1000(b *testing.B) { benchmarkPutSequential(b, 1000) }
+
+func benchmarkPutSequential(b *testing.B, numOps int) {
+	crdt := createBenchCRDT(b, "peer1")
+	ops := make(map[string][]byte, numOps)
+	for i := 0; i < numOps; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	for i := 0; i < b.N; i++ {
+		for key, value := range ops {
+			if err := crdt.Set(key, value); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+}
+
+func BenchmarkPutBatch_10(b *testing.B)   { benchmarkPutBatch(b, 10) }
+func BenchmarkPutBatch_100(b *testing.B)  { benchmarkPutBatch(b, 100) }
+func BenchmarkPutBatch_1000(b *testing.B) { benchmarkPutBatch(b, 1000) }
+
+func benchmarkPutBatch(b *testing.B, numOps int) {
+	crdt := createBenchCRDT(b, "peer1")
+	ops := make(map[string][]byte, numOps)
+	for i := 0; i < numOps; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	for i := 0; i < b.N; i++ {
+		if err := crdt.PutBatch(ops); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkChangeLogSize_Sequential(b *testing.B) {
+	crdt := createBenchCRDT(b, "peer1")
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < 10; j++ {
+			key := fmt.Sprintf("key%d", j)
+			if err := crdt.Set(key, []byte(fmt.Sprintf("value%d", j))); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+
+	b.StopTimer()
+	b.ReportMetric(float64(crdt.PeerSeq), "changes")
+}
+
+func BenchmarkChangeLogSize_Batch(b *testing.B) {
+	crdt := createBenchCRDT(b, "peer1")
+	ops := make(map[string][]byte, 10)
+	for j := 0; j < 10; j++ {
+		ops[fmt.Sprintf("key%d", j)] = []byte(fmt.Sprintf("value%d", j))
+	}
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		if err := crdt.PutBatch(ops); err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	b.StopTimer()
+	b.ReportMetric(float64(crdt.PeerSeq), "changes")
+}
+
+func createBenchCRDT(tb testing.TB, peerID string) *clset.CRDT {
+	opts := &badgerds.DefaultOptions
+	opts.InMemory = true
+	ds, err := badgerds.NewDatastore("", opts)
+	if err != nil {
+		tb.Fatalf("Failed to create datastore: %v", err)
+	}
+	tb.Cleanup(func() { ds.Close() })
+	return clset.New(peerID, ds)
+}

batch_test.go

@@ -0,0 +1,147 @@
+package clset_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPutBatch_Basic(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	ops := map[string][]byte{
+		"key1": []byte("value1"),
+		"key2": []byte("value2"),
+		"key3": []byte("value3"),
+	}
+
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// Verify all keys were set
+	for key, expectedValue := range ops {
+		val, exists, err := crdt.Get(key)
+		require.NoError(t, err)
+		assert.True(t, exists, "key %s should exist", key)
+		assert.Equal(t, expectedValue, val, "value for key %s should match", key)
+	}
+
+	// Verify only one sequence number was used
+	assert.Equal(t, uint64(1), crdt.PeerSeq, "PeerSeq should be 1 after batch")
+}
+
+func TestPutBatch_Empty(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	require.NoError(t, crdt.PutBatch(map[string][]byte{}))
+	assert.Equal(t, uint64(0), crdt.PeerSeq, "PeerSeq should remain 0 for empty batch")
+}
+
+func TestPutBatch_SingleSequenceNumber(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	// Put 10 keys in a batch
+	ops := make(map[string][]byte)
+	for i := 0; i < 10; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// All keys should have the same PeerSeq
+	for key := range ops {
+		meta, found, err := crdt.GetKeyMeta(key)
+		require.NoError(t, err)
+		require.True(t, found)
+		assert.Equal(t, uint64(1), meta.PeerSeq, "all keys should have PeerSeq=1")
+		assert.Equal(t, "peer1", meta.PeerID)
+	}
+
+	// Only one sequence number should be consumed
+	assert.Equal(t, uint64(1), crdt.PeerSeq)
+}
+
+func TestDeleteBatch_Basic(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	// First, create some keys
+	ops := map[string][]byte{
+		"key1": []byte("value1"),
+		"key2": []byte("value2"),
+		"key3": []byte("value3"),
+	}
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// Delete them in a batch
+	keys := []string{"key1", "key2", "key3"}
+	require.NoError(t, crdt.DeleteBatch(keys))
+
+	// Verify all keys were deleted
+	for _, key := range keys {
+		_, exists, err := crdt.Get(key)
+		require.NoError(t, err)
+		assert.False(t, exists, "key %s should not exist", key)
+	}
+
+	// Verify only one additional sequence number was used
+	assert.Equal(t, uint64(2), crdt.PeerSeq)
+}
+
+func TestBatch_LargeOperations(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	// Test with 1000 keys
+	ops := make(map[string][]byte)
+	for i := 0; i < 1000; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// Verify all keys exist
+	count, err := crdt.KeyCount()
+	require.NoError(t, err)
+	assert.Equal(t, uint64(1000), count)
+
+	// Delete half of them
+	keysToDelete := make([]string, 500)
+	for i := 0; i < 500; i++ {
+		keysToDelete[i] = fmt.Sprintf("key%d", i)
+	}
+	require.NoError(t, crdt.DeleteBatch(keysToDelete))
+
+	// Verify count is correct
+	count, err = crdt.KeyCount()
+	require.NoError(t, err)
+	assert.Equal(t, uint64(500), count)
+
+	// Only 2 sequence numbers should be used
+	assert.Equal(t, uint64(2), crdt.PeerSeq)
+}
+
+func TestBatch_P2PSync(t *testing.T) {
+	crdt1 := createTestCRDT(t, "peer1")
+	crdt2 := createTestCRDT(t, "peer2")
+
+	// Peer1 does a batch put
+	ops := map[string][]byte{
+		"key1": []byte("value1"),
+		"key2": []byte("value2"),
+		"key3": []byte("value3"),
+	}
+	require.NoError(t, crdt1.PutBatch(ops))
+
+	// Sync from peer1 to peer2
+	changes, tracked, err := crdt1.GetLatestChanges("peer2", crdt2.GetTrackedPeers())
+	require.NoError(t, err)
+	require.NoError(t, crdt2.MergeChanges("peer1", changes, tracked))
+
+	// Verify all keys made it to peer2
+	for key, expectedValue := range ops {
+		val, exists, err := crdt2.Get(key)
+		require.NoError(t, err)
+		assert.True(t, exists, "key %s should exist on peer2", key)
+		assert.Equal(t, expectedValue, val, "value for key %s should match", key)
+	}
+}