Implement Batch Operations for Atomic Multi-Key Updates

[mgazza]

Problem

CLSet currently only supports single-key operations (Set, Delete). Applications that need to update multiple keys must do so sequentially:

crdt.Set("key1", value1)  // Separate CRDT operation
crdt.Set("key2", value2)  // Separate CRDT operation
crdt.Set("key3", value3)  // Separate CRDT operation

Issues:

1. No atomicity: If one operation fails, others may have succeeded (partial state)
2. Poor performance: Each operation increments sequence number and triggers sync
3. High network traffic: Multiple gossip broadcasts for related changes
4. Datastore incompatibility: go-datastore Batching interface not properly supported

Current Workaround

Applications using CLSet implement a workaround that executes operations sequentially:

type clsetBatch struct {
    ctx  context.Context
    crdt *clset.CRDT
    ops  []batchOp
}

func (b *clsetBatch) Commit(ctx context.Context) error {
    // Execute all operations sequentially
    // Note: This isn't truly atomic
    for _, op := range b.ops {
        if op.delete {
            if err := b.crdt.Delete(op.key.String()); err != nil {
                return err
            }
        } else {
            if err := b.crdt.Set(op.key.String(), op.value); err != nil {
                return err
            }
        }
    }
    return nil
}

Limitations:

• Not atomic (partial commits possible)
• Each operation still separate in CRDT log
• Multiple sequence numbers consumed
• Multiple gossip broadcasts

Solution

Implement native batch operations in CLSet with true atomicity.

API Design

// Batch represents a collection of operations to be applied atomically
type Batch interface {
    Set(key string, value []byte)
    Delete(key string)
    Commit(ctx context.Context) error
}

// NewBatch creates a new batch for atomic operations
func (c *CRDT) NewBatch() Batch {
    return &batch{
        crdt: c,
        ops:  make([]batchOp, 0),
    }
}

Usage

// Create batch
batch := crdt.NewBatch()

// Queue operations
batch.Set("key1", value1)
batch.Set("key2", value2)
batch.Delete("key3")

// Commit atomically - all or nothing
if err := batch.Commit(ctx); err != nil {
    // All operations rolled back
    return err
}
// All operations applied together

Implementation

1. Batch Structure

type batchOp struct {
    key    string
    value  []byte
    delete bool
}

type batch struct {
    crdt *CRDT
    ops  []batchOp
    mu   sync.Mutex
}

func (c *CRDT) NewBatch() *batch {
    return &batch{
        crdt: c,
        ops:  make([]batchOp, 0),
    }
}

func (b *batch) Set(key string, value []byte) {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    b.ops = append(b.ops, batchOp{
        key:    key,
        value:  value,
        delete: false,
    })
}

func (b *batch) Delete(key string) {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    b.ops = append(b.ops, batchOp{
        key:    key,
        delete: true,
    })
}

2. Atomic Commit

func (b *batch) Commit(ctx context.Context) error {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    if len(b.ops) == 0 {
        return nil
    }
    
    b.crdt.mu.Lock()
    defer b.crdt.mu.Unlock()
    
    // Use underlying datastore transaction for atomicity
    txn, err := b.crdt.ds.Batch(ctx)
    if err != nil {
        return fmt.Errorf("creating batch transaction: %w", err)
    }
    
    // Single sequence number for entire batch
    b.crdt.PeerSeq++
    batchSeq := b.crdt.PeerSeq
    
    changes := make([]KeyEntry, 0, len(b.ops))
    
    // Apply all operations in transaction
    for _, op := range b.ops {
        if op.delete {
            // Delete operation
            oldVal, exists, err := b.crdt.getInternal(op.key)
            if err != nil {
                return err
            }
            if !exists {
                continue
            }
            
            // Create tombstone
            tombstone := KeyEntry{
                Key:      op.key,
                Value:    nil,
                PeerID:   b.crdt.PeerID,
                Sequence: batchSeq,
            }
            
            encoded, err := encodeTombstone(tombstone)
            if err != nil {
                return err
            }
            
            // Write to transaction
            if err := txn.Put(ctx, b.crdt.tombstoneKey(op.key), encoded); err != nil {
                return err
            }
            if err := txn.Delete(ctx, b.crdt.dataKey(op.key)); err != nil {
                return err
            }
            
            changes = append(changes, tombstone)
            
        } else {
            // Set operation
            entry := KeyEntry{
                Key:      op.key,
                Value:    op.value,
                PeerID:   b.crdt.PeerID,
                Sequence: batchSeq,
            }
            
            encoded, err := encodeEntry(entry)
            if err != nil {
                return err
            }
            
            // Write to transaction
            if err := txn.Put(ctx, b.crdt.dataKey(op.key), encoded); err != nil {
                return err
            }
            
            changes = append(changes, entry)
        }
    }
    
    // Write single change record for batch
    changeEntry := ChangeEntry{
        PeerID:   b.crdt.PeerID,
        Sequence: batchSeq,
        Changes:  changes,
    }
    
    changeEncoded, err := encodeChangeEntry(changeEntry)
    if err != nil {
        return err
    }
    
    if err := txn.Put(ctx, b.crdt.changeKey(b.crdt.PeerID, batchSeq), changeEncoded); err != nil {
        return err
    }
    
    // Commit transaction atomically
    if err := txn.Commit(ctx); err != nil {
        return fmt.Errorf("committing batch: %w", err)
    }
    
    // Trigger hooks after successful commit
    for _, op := range b.ops {
        if op.delete {
            b.crdt.triggerDeleteHooks(op.key, nil)
        } else {
            b.crdt.triggerInsertHooks(op.key, op.value)
        }
    }
    
    // Broadcast batch as single gossip message
    b.crdt.broadcastChanges(changes)
    
    return nil
}

3. Datastore Batch Interface

Implement datastore.Batching interface properly:

// Batch implements datastore.Batching interface
func (c *CRDT) Batch(ctx context.Context) (datastore.Batch, error) {
    return &datastoreBatch{
        crdt:   c,
        ops:    make([]batchOp, 0),
        ctx:    ctx,
    }, nil
}

type datastoreBatch struct {
    crdt *CRDT
    ops  []batchOp
    ctx  context.Context
    mu   sync.Mutex
}

func (b *datastoreBatch) Put(ctx context.Context, key datastore.Key, value []byte) error {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    b.ops = append(b.ops, batchOp{
        key:    key.String(),
        value:  value,
        delete: false,
    })
    return nil
}

func (b *datastoreBatch) Delete(ctx context.Context, key datastore.Key) error {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    b.ops = append(b.ops, batchOp{
        key:    key.String(),
        delete: true,
    })
    return nil
}

func (b *datastoreBatch) Commit(ctx context.Context) error {
    batch := &batch{
        crdt: b.crdt,
        ops:  b.ops,
    }
    return batch.Commit(ctx)
}

Benefits

1. Atomicity: All operations succeed or all fail together
2. Performance: Single sequence number, single transaction
3. Reduced network traffic: One gossip broadcast for multiple changes
4. Better API: Matches standard datastore Batching interface
5. Cleaner application code: No manual rollback logic needed

Use Cases

1. IP Allocation Metadata

// Allocate IP and update metadata atomically
batch := crdt.NewBatch()
batch.Set("/ipservice/allocations/10.0.0.1", allocationData)
batch.Set("/allocation/metadata/customer-123", customerData)
batch.Set("/allocation/timestamp/10.0.0.1", timestampData)
batch.Commit(ctx)

2. Multi-Key Updates

// Update related keys together
batch := crdt.NewBatch()
batch.Set("/pool/vpc-prod/status", []byte("active"))
batch.Set("/pool/vpc-prod/size", []byte("1024"))
batch.Delete("/pool/vpc-prod/pending")
batch.Commit(ctx)

3. Cleanup Operations

// Delete multiple keys atomically
batch := crdt.NewBatch()
for _, key := range keysToDelete {
    batch.Delete(key)
}
batch.Commit(ctx)

Testing

Unit Tests

• Test batch with multiple Set operations
• Test batch with multiple Delete operations
• Test batch with mixed Set/Delete operations
• Test batch atomicity (rollback on error)
• Test empty batch commit
• Test concurrent batches

Integration Tests

• Test batch with CRDT merge
• Test batch with P2P sync
• Verify single sequence number consumed
• Verify single gossip broadcast
• Test batch with hooks (all triggered after commit)

Performance Tests

• Benchmark batch vs sequential operations
• Measure network traffic reduction
• Test large batches (100+ operations)

Tasks

• Design Batch interface
• Implement NewBatch() method
• Implement batch.Set() and batch.Delete()
• Implement atomic batch.Commit()
• Use single sequence number for batch
• Use single datastore transaction
• Trigger hooks after commit
• Implement datastore.Batch interface
• Add unit tests for batch operations
• Add integration tests with P2P sync
• Document batch API in README
• Add examples

Related

• go-ds-crdt has batch support via datastore transactions
• Would enable better atomicity guarantees for multi-key updates