Conflict Resolution: Handle Concurrent Updates with Last-Write-Wins CRDT

Problem

In distributed systems, multiple clients can update the same data simultaneously across different nodes. Without proper conflict resolution:

• Data inconsistency: Different nodes hold different values
• Lost updates: Some changes are silently overwritten
• Merge conflicts: Manual intervention required to resolve conflicts
• Coordination overhead: Expensive consensus protocols needed
• Latency issues: Waiting for central authority slows operations

Example: Two users edit a document field simultaneously in different regions. Which value wins? How do we ensure eventual consistency?

Solution

Last-Write-Wins (LWW) CRDT is a Conflict-free Replicated Data Type that automatically resolves conflicts by keeping the value with the latest timestamp. This approach:

• Eliminates coordination: No need for locks or consensus
• Ensures convergence: All replicas eventually reach the same state
• Provides causality: Timestamps establish ordering
• Enables offline-first: Works without network connectivity
• Scales horizontally: Each node operates independently

How LWW Works

1. Timestamp assignment: Each update includes a logical or physical timestamp
2. Comparison: When conflicts occur, compare timestamps
3. Resolution: Keep the value with the highest timestamp
4. Propagation: Broadcast the winning value to all replicas

Trade-offs

• ✅ Simple, fast, no coordination
• ✅ Works in highly distributed environments
• ❌ May lose recent updates if clocks are skewed
• ❌ Not suitable for all use cases (e.g., financial transactions)

Code

Basic LWW Register Implementation

from dataclasses import dataclass
from typing import Any, Dict, Tuple
import time
from collections import defaultdict

@dataclass
class TimestampedValue:
    """Represents a value with its timestamp and metadata"""
    value: Any
    timestamp: float
    node_id: str

    def __lt__(self, other: 'TimestampedValue') -> bool:
        """Compare by timestamp, then by node_id for tie-breaking"""
        if self.timestamp != other.timestamp:
            return self.timestamp < other.timestamp
        return self.node_id < other.node_id

class LWWRegister:
    """Last-Write-Wins CRDT for a single value"""

    def __init__(self, node_id: str):
        self.node_id = node_id
        self.value: TimestampedValue = None

    def set(self, value: Any, timestamp: float = None) -> None:
        """Set value with timestamp (defaults to current time)"""
        if timestamp is None:
            timestamp = time.time()

        new_value = TimestampedValue(value, timestamp, self.node_id)

        # Update if no value exists or new timestamp is greater
        if self.value is None or new_value > self.value:
            self.value = new_value

    def get(self) -> Any:
        """Get current value"""
        return self.value.value if self.value else None

    def merge(self, other: 'LWWRegister') -> None:
        """Merge with another register (for replication)"""
        if other.value and (self.value is None or other.value > self.value):
            self.value = other.value

    def to_dict(self) -> Dict:
        """Serialize for transmission"""
        if self.value is None:
            return None
        return {
            'value': self.value.value,
            'timestamp': self.value.timestamp,
            'node_id': self.value.node_id
        }

    @classmethod
    def from_dict(cls, data: Dict, node_id: str) -> 'LWWRegister':
        """Deserialize from transmission"""
        register = cls(node_id)
        if data:
            register.value = TimestampedValue(
                data['value'],
                data['timestamp'],
                data['node_id']
            )
        return register


class LWWMap:
    """Last-Write-Wins CRDT for a map/dictionary"""

    def __init__(self, node_id: str):
        self.node_id = node_id
        self.data: Dict[str, TimestampedValue] = {}

    def set(self, key: str, value: Any, timestamp: float = None) -> None:
        """Set key-value pair with timestamp"""
        if timestamp is None:
            timestamp = time.time()

        new_value = TimestampedValue(value, timestamp, self.node_id)

        # Update if key doesn't exist or new timestamp is greater
        if key not in self.data or new_value > self.data[key]:
            self.data[key] = new_value

    def get(self, key: str) -> Any:
        """Get value by key"""
        if key in self.data:
            return self.data[key].value
        return None

    def delete(self, key: str, timestamp: float = None) -> None:
        """Delete by setting a tombstone"""
        if timestamp is None:
            timestamp = time.time()

        # Use None as tombstone marker
        self.set(key, None, timestamp)

    def merge(self, other: 'LWWMap') -> None:
        """Merge with another map"""
        for key, other_value in other.data.items():
            if key not in self.data or other_value > self.data[key]:
                self.data[key] = other_value

    def items(self) -> Dict[str, Any]:
        """Get all non-deleted items"""
        return {
            k: v.value for k, v in self.data.items()
            if v.value is not None
        }

    def to_dict(self) -> Dict:
        """Serialize for transmission"""
        return {
            k: {
                'value': v.value,
                'timestamp': v.timestamp,
                'node_id': v.node_id
            }
            for k, v in self.data.items()
        }

    @classmethod
    def from_dict(cls, data: Dict, node_id: str) -> 'LWWMap':
        """Deserialize from transmission"""
        lww_map = cls(node_id)
        for key, item in data.items():
            lww_map.data[key] = TimestampedValue(
                item['value'],
                item['timestamp'],
                item['node_id']
            )
        return lww_map


class DistributedLWWSystem:
    """Simulates a distributed system with multiple nodes"""

    def __init__(self):
        self.nodes: Dict[str, LWWMap] = {}

    def add_node(self, node_id: str) -> LWWMap:
        """Add a new node to the system"""
        self.nodes[node_id] = LWWMap(node_id)
        return self.nodes[node_id]

    def update(self, node_id: str, key: str, value: Any, 
               timestamp: float = None) -> None:
        """Update on a specific node"""
        if node_id in self.nodes:
            self.nodes[node_id].set(key, value, timestamp)

    def replicate(self, from_node: str, to_node: str) -> None:
        """Replicate state from one node to another"""
        if from_node in self.nodes and to_node in self.nodes:
            self.nodes[to_node].merge(self.nodes[from_node])

    def broadcast(self) -> None:
        """Broadcast all updates to all nodes"""
        all_data = {}
        for node in self.nodes.values():
            node.merge(LWWMap.from_dict(all_data, node.node_id))
            all_data.update(node.to_dict())

        for node in self.nodes.values():
            node.merge(LWWMap.from_dict(all_data, node.node_id))

    def get_state(self, node_id: str) -> Dict:
        """Get current state of a node"""
        if node_id in self.nodes:
            return self.nodes[node_id].items()
        return {}

    def verify_consistency(self) -> bool:
        """Verify all nodes have converged to same state"""
        if not self.nodes:
            return True

        reference_state = self.nodes[list(self.nodes.keys())[0]].items()

        for node in self.nodes.values():
            if node.items() != reference_state:
                return False

        return True


# Example Usage
if __name__ == "__main__":
    # Create distributed system
    system = DistributedLWWSystem()

    # Add three nodes
    node_a = system.add_node("node_a")
    node_b = system.add_node("node_b")
    node_c = system.add_node("node_c")

    print("=== Scenario: Concurrent Updates ===\n")

    # Simulate concurrent updates at different times
    t1 = 1000.0
    t2 = 1001.0
    t3 = 1002.0

    # Node A updates at t1
    system.update("node_a", "user_name", "Alice", t1)
    print(f"Node A: Set user_name='Alice' at t={t1}")

    # Node B updates at t2 (later)
    system.update("node_b", "user_name", "Bob", t2)
    print(f"Node B: Set user_name='Bob' at t={t2}")

    # Node C updates at t3 (latest)
    system.update("node_c", "user_name", "Charlie", t3)
    print(f"Node C: Set user_name='Charlie' at t={t3}")

    print(f"\nBefore replication:")
    print(f"  Node A: {system.get_state('node_a')}")
    print(f"  Node B: {system.get_state('node_b')}")
    print(f"  Node C: {system.get_state('node_c')}")

    # Broadcast updates
    system.broadcast()

    print(f"\nAfter replication:")
    print(f"  Node A: {system.get_state('node_a')}")
    print(f"  Node B: {system.get_state('node_b')}")
    print(f"  Node C: {system.get_state('node_c')}")

    print(f"\nConsistent? {system.verify_consistency()}")
    print(f"Winner: {system.get_state('node_a')['user_name']} (latest timestamp)")

    print("\n=== Scenario: Multiple Fields ===\n")

    # Reset system
    system = DistributedLWWSystem()
    node_x = system.add_node("node_x")
    node_y = system.add_node("node_y")

    # Concurrent updates on different fields
    system.update("node_x", "email", "alice@example.com", 2000.0)
    system.update("node_x", "age", 30, 2000.0)

    system.update("node_y", "email", "alice.new@example.com", 2001.0)
    system.update("node_y", "phone", "555-1234", 2001.0)

    print("Before merge:")
    print(f"  Node X: {system.get_state('node_x')}")
    print(f"  Node Y: {system.get_state('node_y')}")

    system.broadcast()

    print("\nAfter merge:")
    print(f"  Node X: {system.get_state('node_x')}")
    print(f"  Node Y: {system.get_state('node_y')}")
    print(f"  Consistent? {system.verify_consistency()}")

Advanced: Vector Clock Enhancement

from typing import List

class VectorClock:
    """Vector clock for causal ordering"""

    def __init__(self, node_id: str, nodes: List[str]):
        self.node_id = node_id
        self.clock = {node: 0 for node in nodes}

    def increment(self) -> None:
        """Increment local clock"""
        self.clock[self.node_id] += 1

    def update(self, other: 'VectorClock') -> None:
        """Update with received vector clock"""
        for node in self.clock:
            self.clock[node] = max(self.clock[node], other.clock[node])
        self.increment()

    def happens_before(self, other: 'VectorClock') -> bool:
        """Check if this clock happens before other"""
        less_or_equal = all(
            self.clock[n] <= other.clock[n]
            for n in self.clock
        )
        strictly_less = any(
            self.clock[n] < other.clock[n]
            for n in self.clock
        )
        return less_or_equal and strictly_less

    def concurrent_with(self, other: 'VectorClock') -> bool:
        """Check if concurrent (neither happens before)"""
        return (not self.happens_before(other) and 
                not other.happens_before(self))

    def __repr__(self) -> str:
        return str(self.clock)

Tips

1. Clock Synchronization

# Use NTP or similar for physical clocks
# Or use logical clocks (Lamport, Vector clocks)
# Hybrid logical clocks combine both approaches

2. Tie-Breaking Strategy

# When timestamps are equal, use:
# - Node ID (lexicographic)
# - Hash of value
# - Random selection
# Ensures deterministic resolution

3. Tombstones for Deletions

# Don't remove deleted entries immediately
# Mark with tombstone (None value, high timestamp)
# Prevents resurrection after merge
# Periodically garbage collect old tombstones

4. Causality Preservation

# LWW alone doesn't preserve causality
# Combine with vector clocks for stronger guarantees
# Use hybrid logical clocks for efficiency

5. Monitoring & Observability

# Track:
# - Conflict frequency
# - Data loss rate
# - Replication lag
# - Clock skew
# Helps identify issues early

6. Testing Strategies

# Test scenarios:
# - Concurrent writes to same key
# - Network partitions
# - Clock skew (simulate with offset)
# - Delayed message delivery
# - Node failures and recovery

7. Production Considerations

# - Use monotonic clocks (not wall-clock time)
# - Implement clock skew detection
# - Set reasonable timestamp bounds
# - Log conflicts for audit trail
# - Consider hybrid approaches for critical data

8. When NOT to Use LWW

# ❌ Financial transactions (need strong consistency)
# ❌ Inventory management (risk of overselling)
# ❌ Voting systems (need consensus)
# ✅ User profiles, cache, session data
# ✅ Collaborative editing (with operational transforms)
# ✅ Distributed logs, event streams

---

Key Takeaway: Last-Write-Wins CRDT provides a simple, scalable solution for conflict resolution in distributed systems. While it trades consistency for availability, it's ideal for scenarios where eventual consistency is acceptable and coordination overhead must be minimized.