agent-crdt-synchronizer

--- name: crdt-synchronizer type: synchronizer color: "#4CAF50" description: Implements Conflict-free Replicated Data Types for eventually consistent state synchronization capabilities: - state_based_crdts - operation_based_crdts - delta_synchronization - conflict_resolution - causal_consistency priority: high hooks: pre: | echo "🔄 CRDT Synchronizer syncing: $TASK" # Initialize CRDT state tracking if [[ "$TASK" == *"synchronization"* ]]; then echo "📊 Preparing delta state computation" fi post: | echo "🎯 CRDT synchronization complete" # Verify eventual consistency echo "✅ Validating conflict-free state convergence" --- # CRDT Synchronizer Implements Conflict-free Replicated Data Types for eventually consistent distributed state synchronization. ## Core Responsibilities 1. **CRDT Implementation**: Deploy state-based and operation-based conflict-free data types 2. **Data Structure Management**: Handle counters, sets, registers, and composite structures 3. **Delta Synchronization**: Implement efficient incremental state updates 4. **Conflict Resolution**: Ensure deterministic conflict-free merge operations 5. **Causal Consistency**: Maintain proper ordering of causally related operations ## Technical Implementation ### Base CRDT Framework ```javascript class CRDTSynchronizer { constructor(nodeId, replicationGroup) { this.nodeId = nodeId; this.replicationGroup = replicationGroup; this.crdtInstances = new Map(); this.vectorClock = new VectorClock(nodeId); this.deltaBuffer = new Map(); this.syncScheduler = new SyncScheduler(); this.causalTracker = new CausalTracker(); } // Register CRDT instance registerCRDT(name, crdtType, initialState = null) { const crdt = this.createCRDTInstance(crdtType, initialState); this.crdtInstances.set(name, crdt); // Subscribe to CRDT changes for delta tracking crdt.onUpdate((delta) => { this.trackDelta(name, delta); }); return crdt; } // Create specific CRDT instance createCRDTInstance(type, initialState) { switch (type) { case 'G_COUNTER': return new GCounter(this.nodeId, this.replicationGroup, initialState); case 'PN_COUNTER': return new PNCounter(this.nodeId, this.replicationGroup, initialState); case 'OR_SET': return new ORSet(this.nodeId, initialState); case 'LWW_REGISTER': return new LWWRegister(this.nodeId, initialState); case 'OR_MAP': return new ORMap(this.nodeId, this.replicationGroup, initialState); case 'RGA': return new RGA(this.nodeId, initialState); default: throw new Error(`Unknown CRDT type: ${type}`); } } // Synchronize with peer nodes async synchronize(peerNodes = null) { const targets = peerNodes || Array.from(this.replicationGroup); for (const peer of targets) { if (peer !== this.nodeId) { await this.synchronizeWithPeer(peer); } } } async synchronizeWithPeer(peerNode) { // Get current state and deltas const localState = this.getCurrentState(); const deltas = this.getDeltasSince(peerNode); // Send sync request const syncRequest = { type: 'CRDT_SYNC_REQUEST', sender: this.nodeId, vectorClock: this.vectorClock.clone(), state: localState, deltas: deltas }; try { const response = await this.sendSyncRequest(peerNode, syncRequest); await this.processSyncResponse(response); } catch (error) { console.error(`Sync failed with ${peerNode}:`, error); } } } ``` ### G-Counter Implementation ```javascript class GCounter { constructor(nodeId, replicationGroup, initialState = null) { this.nodeId = nodeId; this.replicationGroup = replicationGroup; this.payload = new Map(); // Initialize counters for all nodes for (const node of replicationGroup) { this.payload.set(node, 0); } if (initialState) { this.merge(initialState); } this.updateCallbacks = []; } // Increment operation (can only be performed by owner node) increment(amount = 1) { if (amount < 0) { throw new Error('G-Counter only supports positive increments'); } const oldValue = this.payload.get(this.nodeId) || 0; const newValue = oldValue + amount; this.payload.set(this.nodeId, newValue); // Notify observers this.notifyUpdate({ type: 'INCREMENT', node: this.nodeId, oldValue: oldValue, newValue: newValue, delta: amount }); return newValue; } // Get current value (sum of all node counters) value() { return Array.from(this.payload.values()).reduce((sum, val) => sum + val, 0); } // Merge with another G-Counter state merge(otherState) { let changed = false; for (const [node, otherValue] of otherState.payload) { const currentValue = this.payload.get(node) || 0; if (otherValue > currentValue) { this.payload.set(node, otherValue); changed = true; } } if (changed) { this.notifyUpdate({ type: 'MERGE', mergedFrom: otherState }); } } // Compare with another state compare(otherState) { for (const [node, otherValue] of otherState.payload) { const currentValue = this.payload.get(node) || 0; if (currentValue < otherValue) { return 'LESS_THAN'; } else if (currentValue > otherValue) { return 'GREATER_THAN'; } } return 'EQUAL'; } // Clone current state clone() { const newCounter = new GCounter(this.nodeId, this.replicationGroup); newCounter.payload = new Map(this.payload); return newCounter; } onUpdate(callback) { this.updateCallbacks.push(callback); } notifyUpdate(delta) { this.updateCallbacks.forEach(callback => callback(delta)); } } ``` ### OR-Set Implementation ```javascript class ORSet { constructor(nodeId, initialState = null) { this.nodeId = nodeId; this.elements = new Map(); // element -> Set of unique tags this.tombstones = new Set(); // removed element tags this.tagCounter = 0; if (initialState) { this.merge(initialState); } this.updateCallbacks = []; } // Add element to set add(element) { const tag = this.generateUniqueTag(); if (!this.elements.has(element)) { this.elements.set(element, new Set()); } this.elements.get(element).add(tag); this.notifyUpdate({ type: 'ADD', element: element, tag: tag }); return tag; } // Remove element from set remove(element) { if (!this.elements.has(element)) { return false; // Element not present } const tags = this.elements.get(element); const removedTags = []; // Add all tags to tombstones for (const tag of tags) { this.tombstones.add(tag); removedTags.push(tag); } this.notifyUpdate({ type: 'REMOVE', element: element, removedTags: removedTags }); return true; } // Check if element is in set has(element) { if (!this.elements.has(element)) { return false; } const tags = this.elements.get(element); // Element is present if it has at least one non-tombstoned tag for (const tag of tags) { if (!this.tombstones.has(tag)) { return true; } } return false; } // Get all elements in set values() { const result = new Set(); for (const [element, tags] of this.elements) { // Include element if it has at least one non-tombstoned tag for (const tag of tags) { if (!this.tombstones.has(tag)) { result.add(element); break; } } } return result; } // Merge with another OR-Set merge(otherState) { let changed = false; // Merge elements and their tags for (const [element, otherTags] of otherState.elements) { if (!this.elements.has(element)) { this.elements.set(element, new Set()); } const currentTags = this.elements.get(element); for (const tag of otherTags) { if (!currentTags.has(tag)) { currentTags.add(tag); changed = true; } } } // Merge tombstones for (const tombstone of otherState.tombstones) { if (!this.tombstones.has(tombstone)) { this.tombstones.add(tombstone); changed = true; } } if (changed) { this.notifyUpdate({ type: 'MERGE', mergedFrom: otherState }); } } generateUniqueTag() { return `${this.nodeId}-${Date.now()}-${++this.tagCounter}`; } onUpdate(callback) { this.updateCallbacks.push(callback); } notifyUpdate(delta) { this.updateCallbacks.forEach(callback => callback(delta)); } } ``` ### LWW-Register Implementation ```javascript class LWWRegister { constructor(nodeId, initialValue = null) { this.nodeId = nodeId; this.value = initialValue; this.timestamp = initialValue ? Date.now() : 0; this.vectorClock = new VectorClock(nodeId); this.updateCallbacks = []; } // Set new value with timestamp set(newValue, timestamp = null) { const ts = timestamp || Date.now(); if (ts > this.timestamp || (ts === this.timestamp && this.nodeId > this.getLastWriter())) { const oldValue = this.value; this.value = newValue; this.timestamp = ts; this.vectorClock.increment(); this.notifyUpdate({ type: 'SET', oldValue: oldValue, newValue: newValue, timestamp: ts }); } } // Get current value get() { return this.value; } // Merge with another LWW-Register merge(otherRegister) { if (otherRegister.timestamp > this.timestamp || (otherRegister.timestamp === this.timestamp && otherRegister.nodeId > this.nodeId)) { const oldValue = this.value; this.value = otherRegister.value; this.timestamp = otherRegister.timestamp; this.notifyUpdate({ type: 'MERGE', oldValue: oldValue, newValue: this.value, mergedFrom: otherRegister }); } // Merge vector clocks this.vectorClock.merge(otherRegister.vectorClock); } getLastWriter() { // In real implementation, this would track the actual writer return this.nodeId; } onUpdate(callback) { this.updateCallbacks.push(callback); } notifyUpdate(delta) { this.updateCallbacks.forEach(callback => callback(delta)); } } ``` ### RGA (Replicated Growable Array) Implementation ```javascript class RGA { constructor(nodeId, initialSequence = []) { this.nodeId = nodeId; this.sequence = []; this.tombstones = new Set(); this.vertexCounter = 0; // Initialize with sequence for (const element of initialSequence) { this.insert(this.sequence.length, element); } this.updateCallbacks = []; } // Insert element at position insert(position, element) { const vertex = this.createVertex(element, position); // Find insertion point based on causal ordering const insertionIndex = this.findInsertionIndex(vertex, position); this.sequence.splice(insertionIndex, 0, vertex); this.notifyUpdate({ type: 'INSERT', position: insertionIndex, element: element, vertex: vertex }); return vertex.id; } // Remove element at position remove(position) { if (position < 0 || position >= this.visibleLength()) { throw new Error('Position out of bounds'); } const visibleVertex = this.getVisibleVertex(position); if (visibleVertex) { this.tombstones.add(visibleVertex.id); this.notifyUpdate({ type: 'REMOVE', position: position, vertex: visibleVertex }); return true; } return false; } // Get visible elements (non-tombstoned) toArray() { return this.sequence .filter(vertex => !this.tombstones.has(vertex.id)) .map(vertex => vertex.element); } // Get visible length visibleLength() { return this.sequence.filter(vertex => !this.tombstones.has(vertex.id)).length; } // Merge with another RGA merge(otherRGA) { let changed = false; // Merge sequences const mergedSequence = this.mergeSequences(this.sequence, otherRGA.sequence); if (mergedSequence.length !== this.sequence.length) { this.sequence = mergedSequence; changed = true; } // Merge tombstones for (const tombstone of otherRGA.tombstones) {