Vector Clocks for Causality

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214

"""
Vector Clock Implementation
 
A complete implementation of vector clocks with
comparison operations and causal relationship detection.
"""
 
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from threading import Lock
from copy import deepcopy
 
 
@dataclass
class VectorClock:
    """
    A vector clock implementation using a dictionary for sparse storage.
    
    Uses dict instead of array to handle dynamic process sets
    and avoid wasting space when process counts are unknown.
    """
    
    process_id: str
    _clock: Dict[str, int] = field(default_factory=dict)
    _lock: Lock = field(default_factory=Lock, repr=False)
    
    def __post_init__(self):
        # Initialize own component to 0
        if self.process_id not in self._clock:
            self._clock[self.process_id] = 0
    
    @property
    def clock(self) -> Dict[str, int]:
        """Get a copy of the current clock state."""
        with self._lock:
            return deepcopy(self._clock)
    
    def tick(self) -> Dict[str, int]:
        """
        Local event: increment own component.
        Returns copy of new clock state.
        """
        with self._lock:
            self._clock[self.process_id] = self._clock.get(self.process_id, 0) + 1
            return deepcopy(self._clock)
    
    def send(self) -> Dict[str, int]:
        """
        Prepare to send: increment and return clock to attach to message.
        """
        return self.tick()  # Send increments before attaching
    
    def receive(self, received_clock: Dict[str, int]) -> Dict[str, int]:
        """
        Receive message: merge clocks, then increment.
        
        Args:
            received_clock: The vector clock from the received message.
        
        Returns:
            Copy of updated clock state.
        """
        with self._lock:
            # Merge: take component-wise maximum
            all_processes = set(self._clock.keys()) | set(received_clock.keys())
            for proc in all_processes:
                self._clock[proc] = max(
                    self._clock.get(proc, 0),
                    received_clock.get(proc, 0)
                )
            
            # Increment own component
            self._clock[self.process_id] = self._clock.get(self.process_id, 0) + 1
            
            return deepcopy(self._clock)
    
    def __str__(self):
        """Pretty print the clock."""
        items = sorted(self._clock.items())
        return "{" + ", ".join(f"{k}:{v}" for k, v in items) + "}"
 
 
def compare_vector_clocks(vc1: Dict[str, int], vc2: Dict[str, int]) -> str:
    """
    Compare two vector clocks to determine their causal relationship.
    
    Returns:
        'BEFORE': vc1 < vc2 (vc1 happened before vc2)
        'AFTER': vc1 > vc2 (vc1 happened after vc2)
        'EQUAL': vc1 = vc2 (same event or identical clocks)
        'CONCURRENT': Neither dominates (concurrent events)
    """
    all_processes = set(vc1.keys()) | set(vc2.keys())
    
    less_than_or_equal = True  # vc1 ≤ vc2 (all components)
    greater_than_or_equal = True  # vc1 ≥ vc2 (all components)
    
    for proc in all_processes:
        v1 = vc1.get(proc, 0)
        v2 = vc2.get(proc, 0)
        
        if v1 > v2:
            less_than_or_equal = False
        if v1 < v2:
            greater_than_or_equal = False
    
    if less_than_or_equal and greater_than_or_equal:
        return 'EQUAL'
    elif less_than_or_equal:
        return 'BEFORE'  # vc1 < vc2
    elif greater_than_or_equal:
        return 'AFTER'   # vc1 > vc2
    else:
        return 'CONCURRENT'
 
 
def happens_before(vc1: Dict[str, int], vc2: Dict[str, int]) -> bool:
    """Check if vc1 happened-before vc2 (vc1 < vc2)."""
    return compare_vector_clocks(vc1, vc2) == 'BEFORE'
 
 
def is_concurrent(vc1: Dict[str, int], vc2: Dict[str, int]) -> bool:
    """Check if two events are concurrent."""
    return compare_vector_clocks(vc1, vc2) == 'CONCURRENT'
 
 
# Demonstration
def demonstrate_vector_clocks():
    """
    Show how vector clocks capture causality and detect concurrency.
    """
    print("Vector Clock Demonstration")
    print("=" * 60)
    
    # Three processes
    p1 = VectorClock("P1")
    p2 = VectorClock("P2")
    p3 = VectorClock("P3")
    
    print("
Initial state:")
    print(f"  P1: {p1}")
    print(f"  P2: {p2}")
    print(f"  P3: {p3}")
    
    # P1 has local event 'a'
    vc_a = p1.tick()
    print(f"
Event a on P1: {vc_a}")
    
    # P1 sends message to P2
    msg1 = p1.send()
    print(f"P1 sends (message VC): {msg1}")
    
    # P2 receives message
    vc_b = p2.receive(msg1)
    print(f"Event b (P2 receives): {vc_b}")
    
    # P3 has local event 'c' (concurrent with everything above)
    vc_c = p3.tick()
    print(f"Event c on P3: {vc_c}")
    
    # P2 sends to P3
    msg2 = p2.send()
    print(f"P2 sends (message VC): {msg2}")
    
    # P3 receives from P2
    vc_d = p3.receive(msg2)
    print(f"Event d (P3 receives): {vc_d}")
    
    # P1 has another local event 'e'
    vc_e = p1.tick()
    print(f"Event e on P1: {vc_e}")
    
    print("
" + "=" * 60)
    print("Causal Relationship Analysis:")
    print("-" * 60)
    
    # Test various pairs
    test_pairs = [
        ("a", vc_a, "b", vc_b),
        ("b", vc_b, "d", vc_d),
        ("a", vc_a, "d", vc_d),
        ("c", vc_c, "b", vc_b),
        ("c", vc_c, "d", vc_d),
        ("e", vc_e, "d", vc_d),
    ]
    
    for name1, clock1, name2, clock2 in test_pairs:
        rel = compare_vector_clocks(clock1, clock2)
        symbol = {
            'BEFORE': '→',
            'AFTER': '←',
            'CONCURRENT': '||',
            'EQUAL': '='
        }[rel]
        print(f"  {name1} {symbol} {name2} ({rel})")
    
    print("
" + "=" * 60)
    print("Key observations:")
    print("1. a → b (message causality): VC(a) < VC(b)")
    print("2. c || b (concurrent): Neither VC dominates")
    print("3. c → d (c happened on P3 before receiving message)")
    print("4. Vector clocks DETECT concurrency - Lamport clocks cannot!")
    
    return {
        'a': vc_a, 'b': vc_b, 'c': vc_c, 'd': vc_d, 'e': vc_e
    }
 
 
if __name__ == "__main__":
    demonstrate_vector_clocks()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173

"""
Vector Clock Comparison: Mathematical Operations
 
Demonstrate all comparison operations and their
relationship to causality.
"""
 
from typing import Dict, Tuple, List
 
 
def leq(v: Dict[str, int], w: Dict[str, int]) -> bool:
    """
    Check if v ≤ w (v is less than or equal to w component-wise).
    
    Returns True if for all processes p: v[p] ≤ w[p]
    """
    all_processes = set(v.keys()) | set(w.keys())
    return all(v.get(p, 0) <= w.get(p, 0) for p in all_processes)
 
 
def lt(v: Dict[str, int], w: Dict[str, int]) -> bool:
    """
    Check if v < w (v strictly less than w).
    
    Returns True if v ≤ w AND v ≠ w
    """
    return leq(v, w) and v != w and not leq(w, v)
 
 
def eq(v: Dict[str, int], w: Dict[str, int]) -> bool:
    """Check if v = w (identical clocks)."""
    all_processes = set(v.keys()) | set(w.keys())
    return all(v.get(p, 0) == w.get(p, 0) for p in all_processes)
 
 
def concurrent(v: Dict[str, int], w: Dict[str, int]) -> bool:
    """
    Check if v || w (concurrent/incomparable).
    
    Returns True if neither v ≤ w nor w ≤ v
    """
    return not leq(v, w) and not leq(w, v)
 
 
def merge(v: Dict[str, int], w: Dict[str, int]) -> Dict[str, int]:
    """
    Compute the merge (join/least upper bound) of two vector clocks.
    
    merge(v, w)[p] = max(v[p], w[p]) for all processes p
    
    This represents the earliest point that "knows about" both v and w.
    """
    all_processes = set(v.keys()) | set(w.keys())
    return {p: max(v.get(p, 0), w.get(p, 0)) for p in all_processes}
 
 
def describe_relationship(v: Dict[str, int], w: Dict[str, int]) -> str:
    """
    Provide detailed description of the relationship between clocks.
    """
    if eq(v, w):
        return "EQUAL: v = w (same event or identical knowledge)"
    elif lt(v, w):
        return "BEFORE: v → w (v happened-before w)"
    elif lt(w, v):
        return "AFTER: w → v (w happened-before v)"
    else:
        return "CONCURRENT: v || w (neither caused the other)"
 
 
# Demonstration of all cases
def demonstrate_comparisons():
    """Demonstrate vector clock comparison with examples."""
    
    print("Vector Clock Comparison Examples")
    print("=" * 65)
    
    examples = [
        # (name1, vc1, name2, vc2, explanation)
        (
            "Simple before",
            {'P1': 1, 'P2': 0, 'P3': 0},
            {'P1': 2, 'P2': 1, 'P3': 0},
            "All v1 components ≤ v2, with at least one <"
        ),
        (
            "Message causality",
            {'P1': 3, 'P2': 2, 'P3': 1},
            {'P1': 3, 'P2': 3, 'P3': 1},
            "P2 received from P1, incremented own counter"
        ),
        (
            "Concurrent writes",
            {'P1': 2, 'P2': 1, 'P3': 0},
            {'P1': 1, 'P2': 2, 'P3': 0},
            "P1: 2>1, but P2: 1<2. Neither dominates."
        ),
        (
            "Independent events",
            {'P1': 5, 'P2': 0, 'P3': 0},
            {'P1': 0, 'P2': 0, 'P3': 7},
            "P1 and P3 haven't communicated at all"
        ),
        (
            "Equal clocks",
            {'P1': 2, 'P2': 3, 'P3': 1},
            {'P1': 2, 'P2': 3, 'P3': 1},
            "Identical - same event or same knowledge state"
        ),
        (
            "Complex causality",
            {'P1': 2, 'P2': 3, 'P3': 1},
            {'P1': 4, 'P2': 5, 'P3': 3},
            "All components increased in v2"
        ),
    ]
    
    for name, v1, v2, explanation in examples:
        print(f"
{name}:")
        print(f"  v1 = {v1}")
        print(f"  v2 = {v2}")
        rel = describe_relationship(v1, v2)
        print(f"  Result: {rel}")
        print(f"  Why: {explanation}")
        
        # Show detailed comparison
        all_procs = sorted(set(v1.keys()) | set(v2.keys()))
        comparisons = []
        for p in all_procs:
            c1, c2 = v1.get(p, 0), v2.get(p, 0)
            if c1 < c2:
                comparisons.append(f"{p}:{c1}<{c2}")
            elif c1 > c2:
                comparisons.append(f"{p}:{c1}>{c2}")
            else:
                comparisons.append(f"{p}:{c1}={c2}")
        print(f"  Component-wise: [{', '.join(comparisons)}]")
 
 
def demonstrate_merge():
    """Show how merge computes least upper bound."""
    
    print("
 
Vector Clock Merge (Least Upper Bound)")
    print("=" * 65)
    
    v1 = {'P1': 2, 'P2': 1, 'P3': 3}
    v2 = {'P1': 1, 'P2': 4, 'P3': 2}
    merged = merge(v1, v2)
    
    print(f"v1 = {v1}")
    print(f"v2 = {v2}")
    print(f"merge(v1, v2) = {merged}")
    print(f"
Note: merge takes max of each component")
    print(f"  P1: max(2,1) = 2")
    print(f"  P2: max(1,4) = 4")
    print(f"  P3: max(3,2) = 3")
    print(f"
The merged clock represents knowledge of both events.")
    print(f"Any event with this clock 'knows about' both v1 and v2.")
    
    # Verify v1 ≤ merged and v2 ≤ merged
    print(f"
v1 ≤ merged: {leq(v1, merged)}")
    print(f"v2 ≤ merged: {leq(v2, merged)}")
 
 
if __name__ == "__main__":
    demonstrate_comparisons()
    demonstrate_merge()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219

"""
Conflict Detection in a Replicated Data Store
 
Demonstrates how vector clocks enable detection of
concurrent writes (conflicts) vs sequential updates.
"""
 
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set, Any
from copy import deepcopy
 
 
@dataclass
class VersionedValue:
    """A value tagged with its causal history (vector clock)."""
    value: Any
    vector_clock: Dict[str, int]
    
    def __str__(self):
        return f"'{self.value}' @ {self.vector_clock}"
 
 
class ReplicatedStore:
    """
    A simple key-value store with vector clock versioning.
    Detects conflicts and maintains concurrent values (siblings).
    """
    
    def __init__(self, replica_id: str):
        self.replica_id = replica_id
        # Each key maps to a LIST of values (siblings for concurrent writes)
        self.data: Dict[str, List[VersionedValue]] = {}
        # Local vector clock tracking what we've seen
        self.clock: Dict[str, int] = {replica_id: 0}
    
    def _increment_clock(self):
        """Increment our local clock component."""
        self.clock[self.replica_id] = self.clock.get(self.replica_id, 0) + 1
        return deepcopy(self.clock)
    
    def _merge_clock(self, other_clock: Dict[str, int]):
        """Merge another clock into ours."""
        all_replicas = set(self.clock.keys()) | set(other_clock.keys())
        for r in all_replicas:
            self.clock[r] = max(self.clock.get(r, 0), other_clock.get(r, 0))
    
    def put(self, key: str, value: Any, context: Optional[List[Dict[str, int]]] = None) -> Dict[str, int]:
        """
        Write a value. Context is the vector clock(s) the client read.
        
        If context is provided, this write supersedes those versions.
        If context is absent, this is a blind write (may create conflict).
        """
        new_clock = self._increment_clock()
        new_version = VersionedValue(value, new_clock)
        
        if key not in self.data:
            # New key, simply store
            self.data[key] = [new_version]
        elif context:
            # Client has read previous version(s) and is updating based on them
            # Remove any versions that are superseded by the context
            self.data[key] = [
                v for v in self.data[key]
                if not self._is_superseded(v.vector_clock, context)
            ]
            self.data[key].append(new_version)
        else:
            # Blind write: may conflict with existing versions
            self.data[key].append(new_version)
            self._resolve_siblings(key)
        
        return new_clock
    
    def _is_superseded(self, vc: Dict[str, int], context: List[Dict[str, int]]) -> bool:
        """Check if a version is superseded by any clock in the context."""
        for ctx_vc in context:
            if self._vc_leq(vc, ctx_vc):
                return True
        return False
    
    def _vc_leq(self, v1: Dict[str, int], v2: Dict[str, int]) -> bool:
        """Check if v1 ≤ v2 (v1 happened-before or equal to v2)."""
        all_replicas = set(v1.keys()) | set(v2.keys())
        return all(v1.get(r, 0) <= v2.get(r, 0) for r in all_replicas)
    
    def _resolve_siblings(self, key: str):
        """
        Resolve siblings: remove any version that is superseded by another.
        Keep only incomparable (concurrent) versions.
        """
        if key not in self.data:
            return
        
        values = self.data[key]
        # Keep a value only if no other value supersedes it
        resolved = []
        for v in values:
            superseded = False
            for other in values:
                if v is not other and self._vc_leq(v.vector_clock, other.vector_clock):
                    if v.vector_clock != other.vector_clock:  # Strict less than
                        superseded = True
                        break
            if not superseded:
                resolved.append(v)
        
        self.data[key] = resolved
    
    def get(self, key: str) -> Optional[List[VersionedValue]]:
        """
        Read a key. Returns list of values (multiple if concurrent versions exist).
        """
        self._resolve_siblings(key)
        return self.data.get(key)
    
    def has_conflict(self, key: str) -> bool:
        """Check if a key has conflicting (concurrent) versions."""
        values = self.get(key)
        return values is not None and len(values) > 1
    
    def receive_write(self, key: str, versioned_value: VersionedValue):
        """
        Receive a write from another replica (replication).
        """
        self._merge_clock(versioned_value.vector_clock)
        
        if key not in self.data:
            self.data[key] = [versioned_value]
        else:
            self.data[key].append(versioned_value)
            self._resolve_siblings(key)
 
 
# Demonstration of conflict detection
def demonstrate_conflict_detection():
    """Show how concurrent writes create conflicts (siblings)."""
    
    print("Conflict Detection with Vector Clocks")
    print("=" * 65)
    
    # Two replicas
    replica_a = ReplicatedStore("A")
    replica_b = ReplicatedStore("B")
    
    print("
Scenario: Two users concurrently update 'username'")
    print("-" * 65)
    
    # User 1 writes to replica A
    vc_a = replica_a.put("username", "alice")
    print(f"Replica A: put('username', 'alice') -> {vc_a}")
    
    # User 2 writes to replica B (concurrently, before syncing)
    vc_b = replica_b.put("username", "bob")
    print(f"Replica B: put('username', 'bob') -> {vc_b}")
    
    # Now replicas sync
    print("
Replicas synchronize...")
    
    # A receives B's write
    b_value = replica_b.get("username")[0]
    replica_a.receive_write("username", b_value)
    
    # B receives A's write
    a_value_result = replica_a.get("username")
    for v in a_value_result:
        if v.value == "alice":  # Only sync Alice's write
            replica_b.receive_write("username", v)
            break
    
    # Check for conflicts
    print("
After sync:")
    
    a_values = replica_a.get("username")
    print(f"Replica A sees: {[str(v) for v in a_values]}")
    print(f"Replica A has conflict: {replica_a.has_conflict('username')}")
    
    b_values = replica_b.get("username")  
    print(f"Replica B sees: {[str(v) for v in b_values]}")
    print(f"Replica B has conflict: {replica_b.has_conflict('username')}")
    
    print("
" + "=" * 65)
    print("Analysis:")
    print("- Alice's write: A:1 (no knowledge of B)")
    print("- Bob's write: B:1 (no knowledge of A)")
    print("- Neither dominates: {A:1} || {B:1}")
    print("- Result: CONFLICT DETECTED - both values preserved as siblings")
    print("
The application must now resolve: use Alice, Bob, or merge?")
    
    # Demonstrate resolution
    print("
" + "=" * 65)
    print("Resolution: User reads both, chooses 'alicebob' as merge")
    print("-" * 65)
    
    # Client reads and sees conflict
    values = replica_a.get("username")
    context = [v.vector_clock for v in values]  # Read context
    
    # Client writes merge, providing context
    new_vc = replica_a.put("username", "alicebob", context=context)
    print(f"Replica A: put('username', 'alicebob', context) -> {new_vc}")
    
    # Check result
    resolved = replica_a.get("username")
    print(f"After resolution: {[str(v) for v in resolved]}")
    print(f"Has conflict: {replica_a.has_conflict('username')}")
    print("
Context tells the system which versions this write supersedes.")
    print("The new write's clock dominates both old clocks -> siblings removed.")
 
 
if __name__ == "__main__":
    demonstrate_conflict_detection()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

"""
Hybrid Logical Clock (HLC) Implementation
 
Combines physical time precision with logical clock guarantees.
Provides causality detection with smaller overhead than vector clocks.
"""
 
from dataclasses import dataclass
from typing import Tuple
import time
 
 
@dataclass(frozen=True, order=True)
class HLC:
    """
    Hybrid Logical Clock timestamp.
    
    Ordered by: (physical_component, logical_counter, node_id)
    
    Properties:
    - Causally consistent: if a → b, then HLC(a) < HLC(b)
    - Close to wall-clock: physical component tracks real time
    - Unique: (node_id, physical, logical) is globally unique
    """
    physical: int      # Wall-clock time in nanoseconds
    logical: int       # Logical counter for same-time events
    node_id: str       # Node identifier for tie-breaking
    
    def __str__(self):
        # Human-readable format
        time_ms = self.physical // 1_000_000
        return f"HLC({time_ms}.{self.logical:03d}@{self.node_id})"
 
 
class HLCClock:
    """
    HLC clock implementation following Kulkarni et al. (2014).
    
    Guarantees:
    1. Causality: If event a causes event b, HLC(a) < HLC(b)
    2. Bounded: |HLC.physical - physical_time| is bounded
    3. Monotonic: Timestamps on a node are strictly increasing
    """
    
    def __init__(self, node_id: str, max_drift_ns: int = 1_000_000_000):
        """
        Args:
            node_id: Unique identifier for this node
            max_drift_ns: Maximum allowed drift from physical time (default 1s)
        """
        self.node_id = node_id
        self.max_drift = max_drift_ns
        
        # Last known physical time (nanoseconds)
        self.last_physical = 0
        # Logical counter
        self.last_logical = 0
    
    def _physical_time_ns(self) -> int:
        """Get current physical time in nanoseconds."""
        return int(time.time() * 1_000_000_000)
    
    def now(self) -> HLC:
        """
        Generate a new HLC timestamp for a local event.
        
        Returns:
            New HLC timestamp
        """
        pt = self._physical_time_ns()
        
        if pt > self.last_physical:
            # Physical time advanced: reset logical counter
            self.last_physical = pt
            self.last_logical = 0
        else:
            # Physical time hasn't advanced (same or went backward)
            # Increment logical counter
            self.last_logical += 1
        
        return HLC(self.last_physical, self.last_logical, self.node_id)
    
    def send(self) -> HLC:
        """Generate timestamp for message send."""
        return self.now()
    
    def receive(self, received: HLC) -> HLC:
        """
        Update clock on receiving a message and generate receive timestamp.
        
        Args:
            received: HLC from received message
            
        Returns:
            HLC for the receive event
        """
        pt = self._physical_time_ns()
        
        # Check for excessive drift
        if received.physical > pt + self.max_drift:
            raise ValueError(
                f"Received timestamp too far in future: "
                f"received={received.physical}, local={pt}"
            )
        
        if pt > self.last_physical and pt > received.physical:
            # Physical time is ahead of both: reset logical
            self.last_physical = pt
            self.last_logical = 0
        elif self.last_physical == received.physical:
            # Same physical time as both last and received
            self.last_logical = max(self.last_logical, received.logical) + 1
        elif received.physical > self.last_physical:
            # Received is newer: adopt its physical, increment logical
            self.last_physical = received.physical
            self.last_logical = received.logical + 1
        else:
            # Our last is newest: increment our logical
            self.last_logical += 1
        
        return HLC(self.last_physical, self.last_logical, self.node_id)
 
 
def compare_hlc(a: HLC, b: HLC) -> str:
    """Compare two HLC timestamps."""
    if a < b:
        return f"{a} < {b} (a before b)"
    elif a > b:
        return f"{a} > {b} (a after b)"  
    else:
        return f"{a} = {b} (same timestamp)"
 
 
# Demonstration
def demonstrate_hlc():
    """Show HLC properties."""
    
    print("Hybrid Logical Clock Demonstration")
    print("=" * 60)
    
    node_a = HLCClock("A")
    node_b = HLCClock("B")
    
    # Local events on A
    ts1 = node_a.now()
    print(f"A local event 1: {ts1}")
    
    ts2 = node_a.now()
    print(f"A local event 2: {ts2}")
    
    # A sends to B
    msg_ts = node_a.send()
    print(f"A sends message: {msg_ts}")
    
    # B receives
    ts3 = node_b.receive(msg_ts)
    print(f"B receives: {ts3}")
    
    # B has local event
    ts4 = node_b.now()
    print(f"B local event: {ts4}")
    
    print("
Ordering verification:")
    print(f"  {compare_hlc(ts1, ts2)}")
    print(f"  {compare_hlc(msg_ts, ts3)}")
    print(f"  {compare_hlc(ts3, ts4)}")
    
    print("
HLC advantages over vector clocks:")
    print("  - Fixed size (doesn't grow with nodes)")
    print("  - Physical time component for rough ordering")
    print("  - Efficient comparison (lexicographic)")
 
 
if __name__ == "__main__":
    demonstrate_hlc()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

"""
Vector Clock Serialization for Production Use
 
Efficient encoding for storage and network transmission.
"""
 
import struct
import json
from typing import Dict
 
 
def to_json(vc: Dict[str, int]) -> str:
    """Serialize to JSON (for debugging/APIs)."""
    return json.dumps(vc, sort_keys=True)
 
 
def from_json(data: str) -> Dict[str, int]:
    """Deserialize from JSON."""
    return json.loads(data)
 
 
def to_binary(vc: Dict[str, int], max_node_len: int = 32) -> bytes:
    """
    Compact binary serialization.
    
    Format:
    - 2 bytes: entry count
    - For each entry:
      - 1 byte: node ID length
      - N bytes: node ID (UTF-8)
      - 8 bytes: counter (uint64)
    """
    entries = sorted(vc.items())
    result = struct.pack('>H', len(entries))
    
    for node_id, counter in entries:
        node_bytes = node_id.encode('utf-8')[:max_node_len]
        result += struct.pack('>B', len(node_bytes))
        result += node_bytes
        result += struct.pack('>Q', counter)
    
    return result
 
 
def from_binary(data: bytes) -> Dict[str, int]:
    """Deserialize from binary format."""
    offset = 0
    count = struct.unpack_from('>H', data, offset)[0]
    offset += 2
    
    result = {}
    for _ in range(count):
        node_len = struct.unpack_from('>B', data, offset)[0]
        offset += 1
        node_id = data[offset:offset + node_len].decode('utf-8')
        offset += node_len
        counter = struct.unpack_from('>Q', data, offset)[0]
        offset += 8
        result[node_id] = counter
    
    return result
 
 
# Size comparison
def compare_formats():
    """Compare serialization formats by size."""
    
    vc = {
        'node-us-east-1a-001': 12345,
        'node-us-west-2b-042': 67890,
        'node-eu-west-1c-003': 11111,
    }
    
    json_size = len(to_json(vc).encode('utf-8'))
    binary_size = len(to_binary(vc))
    
    print("Serialization Size Comparison")
    print("-" * 40)
    print(f"Vector clock: {vc}")
    print(f"JSON size: {json_size} bytes")
    print(f"Binary size: {binary_size} bytes")
    print(f"Binary saves: {json_size - binary_size} bytes ({100*(json_size-binary_size)/json_size:.1f}%)")
 
 
if __name__ == "__main__":
    compare_formats()