Logical Clocks (Lamport 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

"""
Happens-Before Relation Examples
 
Consider three processes P1, P2, P3 with events as follows:
 
    P1          P2          P3
    |           |           |
    a           |           |
    |\          |           |
    | \-------> b           |
    |           |\          |
    |           | \-------> d
    c           |           |
    |\          |           |
    | \-------> e           |
    |           |           |
 
Happens-before relationships:
- a → b (message from P1 to P2)
- b → d (message from P2 to P3)  
- a → d (transitivity: a → b → d)
- a → c (same process, sequential)
- c → e (message from P1 to P2)
 
Concurrent relationships:
- c || b (no causal path: c didn't cause b, b didn't cause c)
- c || d (no causal path)
- a || (no concurrent events with a - all others come after or before)
"""
 
from dataclasses import dataclass
from typing import List, Set, Optional, Tuple
from enum import Enum
 
class EventType(Enum):
    LOCAL = "local"      # Internal event on a process
    SEND = "send"        # Message send
    RECEIVE = "receive"  # Message receive
 
 
@dataclass
class Event:
    """Represents an event in a distributed system."""
    process_id: str
    event_id: str
    event_type: EventType
    sequence: int  # Order within the process
    
    # For send/receive events: the paired event
    paired_event_id: Optional[str] = None
    
    def __hash__(self):
        return hash(self.event_id)
    
    def __eq__(self, other):
        return self.event_id == other.event_id
 
 
class HappensBeforeAnalyzer:
    """
    Analyzes the happens-before relation in a distributed execution.
    """
    
    def __init__(self):
        self.events: dict[str, Event] = {}
        self.send_to_receive: dict[str, str] = {}  # send_id -> receive_id
        
    def add_event(self, event: Event):
        """Add an event to the execution."""
        self.events[event.event_id] = event
        if event.event_type == EventType.SEND and event.paired_event_id:
            self.send_to_receive[event.event_id] = event.paired_event_id
    
    def happens_before(self, a_id: str, b_id: str, visited: Optional[Set[str]] = None) -> bool:
        """
        Determine if event a happens-before event b.
        
        Uses depth-first search through the happens-before graph.
        """
        if visited is None:
            visited = set()
        
        if a_id in visited:
            return False
        visited.add(a_id)
        
        if a_id == b_id:
            return False  # Event doesn't happen before itself
            
        a = self.events[a_id]
        b = self.events[b_id]
        
        # Rule 1: Same process, sequential order
        if a.process_id == b.process_id and a.sequence < b.sequence:
            return True
        
        # Rule 2: Message send to receive
        if a.event_type == EventType.SEND and a.paired_event_id == b_id:
            return True
        
        # Rule 3: Transitivity - check all events this could lead to
        # Check later events on same process
        for event in self.events.values():
            if event.process_id == a.process_id and event.sequence > a.sequence:
                if self.happens_before(event.event_id, b_id, visited.copy()):
                    return True
        
        # Check message sends and their effects
        if a.event_type == EventType.SEND and a.paired_event_id:
            if self.happens_before(a.paired_event_id, b_id, visited.copy()):
                return True
        
        return False
    
    def are_concurrent(self, a_id: str, b_id: str) -> bool:
        """
        Determine if two events are concurrent (neither happens-before the other).
        """
        return not self.happens_before(a_id, b_id) and not self.happens_before(b_id, a_id)
    
    def classify_relationship(self, a_id: str, b_id: str) -> str:
        """
        Classify the relationship between two events.
        """
        if self.happens_before(a_id, b_id):
            return f"{a_id} → {b_id}"
        elif self.happens_before(b_id, a_id):
            return f"{b_id} → {a_id}"
        else:
            return f"{a_id} || {b_id}"  # Concurrent
 
 
# Example usage with the diagram above
analyzer = HappensBeforeAnalyzer()
 
# Add events from P1
analyzer.add_event(Event("P1", "a", EventType.SEND, 1, "b"))
analyzer.add_event(Event("P1", "c", EventType.SEND, 2, "e"))
 
# Add events from P2
analyzer.add_event(Event("P2", "b", EventType.RECEIVE, 1, "a"))
analyzer.add_event(Event("P2", "e", EventType.RECEIVE, 2, "c"))
 
# Add message from P2 to P3 (b sends, d receives)
analyzer.events["b"].event_type = EventType.SEND
analyzer.events["b"].paired_event_id = "d"
analyzer.add_event(Event("P3", "d", EventType.RECEIVE, 1, "b"))
 
# Verify relationships
print("Happens-Before Analysis:")
print("-" * 40)
pairs = [("a", "b"), ("b", "d"), ("a", "d"), ("c", "b"), ("c", "d")]
for x, y in pairs:
    print(f"  {analyzer.classify_relationship(x, y)}")

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

"""
Lamport Clock Implementation
 
A complete, production-quality implementation of Lamport clocks
with demonstration of correct ordering properties.
"""
 
from dataclasses import dataclass, field
from typing import List, Tuple, Optional
from threading import Lock
 
 
@dataclass
class LamportClock:
    """
    A Lamport logical clock for a single process.
    
    Thread-safe implementation suitable for concurrent use.
    """
    
    process_id: str
    _counter: int = field(default=0, repr=False)
    _lock: Lock = field(default_factory=Lock, repr=False)
    
    @property
    def value(self) -> int:
        """Current clock value (read-only)."""
        with self._lock:
            return self._counter
    
    def tick(self) -> int:
        """
        Increment clock for a local/internal event.
        
        Returns:
            The new timestamp for this event.
        """
        with self._lock:
            self._counter += 1
            return self._counter
    
    def send(self) -> Tuple[int, bytes]:
        """
        Prepare to send a message.
        
        Returns:
            Tuple of (timestamp, message_header) to include with message.
        """
        timestamp = self.tick()
        # In practice, encode timestamp in message header
        header = timestamp.to_bytes(8, 'big')
        return timestamp, header
    
    def receive(self, received_timestamp: int) -> int:
        """
        Process receipt of a message with a Lamport timestamp.
        
        Args:
            received_timestamp: The timestamp from the received message.
            
        Returns:
            The new timestamp for this receive event.
        """
        with self._lock:
            # max(local, received) + 1
            self._counter = max(self._counter, received_timestamp) + 1
            return self._counter
    
    def receive_from_header(self, header: bytes) -> int:
        """
        Process receipt of a message, extracting timestamp from header.
        
        Args:
            header: 8-byte timestamp header from received message.
            
        Returns:
            The new timestamp for this receive event.
        """
        received_timestamp = int.from_bytes(header, 'big')
        return self.receive(received_timestamp)
 
 
# Demonstration of the clock condition property
def demonstrate_clock_condition():
    """
    Show that Lamport clocks satisfy the clock condition:
    If a → b, then L(a) < L(b)
    """
    print("Lamport Clock Demonstration")
    print("=" * 50)
    
    # Create clocks for three processes
    p1 = LamportClock("P1")
    p2 = LamportClock("P2")
    p3 = LamportClock("P3")
    
    events = []
    
    # P1 has local event 'a'
    ts_a = p1.tick()
    events.append(("a", "P1", ts_a))
    print(f"Event a on P1: timestamp = {ts_a}")
    
    # P1 sends message to P2 (P2 receives as 'b')
    send_ts, header = p1.send()
    events.append(("send_a_to_b", "P1", send_ts))
    print(f"P1 sends to P2: send timestamp = {send_ts}")
    
    ts_b = p2.receive_from_header(header)
    events.append(("b", "P2", ts_b))
    print(f"Event b on P2 (receive): timestamp = {ts_b}")
    
    # P2 sends message to P3 (P3 receives as 'd')
    send_ts_2, header_2 = p2.send()
    events.append(("send_b_to_d", "P2", send_ts_2))
    print(f"P2 sends to P3: send timestamp = {send_ts_2}")
    
    ts_d = p3.receive_from_header(header_2)
    events.append(("d", "P3", ts_d))
    print(f"Event d on P3 (receive): timestamp = {ts_d}")
    
    # P1 has local event 'c'
    ts_c = p1.tick()
    events.append(("c", "P1", ts_c))
    print(f"Event c on P1: timestamp = {ts_c}")
    
    # P1 sends to P2 (P2 receives as 'e')
    send_ts_3, header_3 = p1.send()
    events.append(("send_c_to_e", "P1", send_ts_3))
    print(f"P1 sends to P2: send timestamp = {send_ts_3}")
    
    ts_e = p2.receive_from_header(header_3)
    events.append(("e", "P2", ts_e))
    print(f"Event e on P2 (receive): timestamp = {ts_e}")
    
    print("
" + "=" * 50)
    print("Clock Condition Verification:")
    print("-" * 50)
    
    # Verify: a → b implies L(a) < L(b)
    # a → b (send/receive)
    print(f"a → b: L(a)={ts_a} < L(b)={ts_b}? {ts_a < ts_b} ✓")
    
    # b → d (send/receive)  
    print(f"b → d: L(b)={ts_b} < L(d)={ts_d}? {ts_b < ts_d} ✓")
    
    # a → d (transitivity)
    print(f"a → d: L(a)={ts_a} < L(d)={ts_d}? {ts_a < ts_d} ✓")
    
    # c → e (send/receive)
    print(f"c → e: L(c)={ts_c} < L(e)={ts_e}? {ts_c < ts_e} ✓")
    
    print("
Note: L(b)={} and L(c)={} - but c || b (concurrent)!".format(ts_b, ts_c))
    print("Lamport clocks cannot detect concurrency (one-way implication only).")
    
    return events
 
 
if __name__ == "__main__":
    demonstrate_clock_condition()

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

"""
Total Ordering with Lamport Timestamps
 
Demonstrates how to create a total order from Lamport clocks
by breaking ties with process IDs.
"""
 
from dataclasses import dataclass
from typing import List
from functools import total_ordering
 
 
@total_ordering
@dataclass
class LamportTimestamp:
    """
    A Lamport timestamp that supports total ordering.
    
    Order is determined by:
    1. Clock value (primary)
    2. Process ID (tie-breaker)
    
    This creates a total order consistent with causality.
    """
    clock: int
    process_id: str
    
    # Optional: human-readable event name for debugging
    event_name: str = ""
    
    def __eq__(self, other):
        if not isinstance(other, LamportTimestamp):
            return NotImplemented
        return self.clock == other.clock and self.process_id == other.process_id
    
    def __lt__(self, other):
        if not isinstance(other, LamportTimestamp):
            return NotImplemented
        # Primary: compare clock values
        if self.clock != other.clock:
            return self.clock < other.clock
        # Tie-breaker: compare process IDs (lexicographic)
        return self.process_id < other.process_id
    
    def __hash__(self):
        return hash((self.clock, self.process_id))
    
    def __repr__(self):
        name = f" ({self.event_name})" if self.event_name else ""
        return f"({self.clock}, {self.process_id}){name}"
 
 
# Demonstration: sorting events in total order
def demonstrate_total_order():
    """
    Show how Lamport timestamps create a total order,
    including handling of concurrent events.
    """
    print("Total Ordering with Lamport Timestamps")
    print("=" * 50)
    
    # Create events with various timestamps
    # Some events are concurrent (same clock value, different processes)
    events = [
        LamportTimestamp(1, "P1", "a"),
        LamportTimestamp(2, "P1", "send"),
        LamportTimestamp(3, "P2", "b"),
        LamportTimestamp(3, "P1", "c"),      # Same clock as b, different process
        LamportTimestamp(4, "P2", "send_2"),
        LamportTimestamp(5, "P3", "d"),
        LamportTimestamp(4, "P1", "send_3"), # Same clock as send_2
        LamportTimestamp(6, "P2", "e"),
    ]
    
    print("
Unsorted events:")
    for e in events:
        print(f"  {e}")
    
    # Sort by total order
    sorted_events = sorted(events)
    
    print("
Sorted by total order:")
    for i, e in enumerate(sorted_events):
        print(f"  {i+1}. {e}")
    
    print("
" + "-" * 50)
    print("Key observations:")
    print("1. Event 'b' (3, P2) and 'c' (3, P1) have same clock")
    print("   But in total order: c < b (because P1 < P2)")
    print("
2. This is arbitrary but CONSISTENT across all processes")
    print("   Every process will sort these events the same way")
    print("
3. Causally related events maintain correct order:")
    print("   a → b: (1, P1) < (3, P2) ✓")
    
    return sorted_events
 
 
# Application: Mutual exclusion with Lamport timestamps
class LamportMutualExclusion:
    """
    Lamport's mutual exclusion algorithm using logical clocks.
    
    This demonstrates how total ordering enables distributed coordination.
    Each process maintains a request queue ordered by Lamport timestamps.
    A process can enter critical section when:
    1. Its request is at the head of its queue
    2. It has received acknowledgments (messages with higher timestamps)
       from all other processes
    """
    
    def __init__(self, process_id: str, all_processes: List[str]):
        self.process_id = process_id
        self.all_processes = all_processes
        self.clock = 0
        self.request_queue: List[LamportTimestamp] = []
        self.acks_received: set = set()
        
    def _tick(self) -> int:
        self.clock += 1
        return self.clock
    
    def _update_clock(self, received_ts: int):
        self.clock = max(self.clock, received_ts) + 1
    
    def request_lock(self) -> LamportTimestamp:
        """Request to enter critical section."""
        ts = LamportTimestamp(self._tick(), self.process_id, "request")
        self.request_queue.append(ts)
        self.request_queue.sort()  # Maintain total order
        self.acks_received = {self.process_id}  # Self-ack
        
        print(f"{self.process_id}: Requesting lock with timestamp {ts}")
        # In practice: broadcast REQUEST message to all other processes
        return ts
    
    def receive_request(self, request_ts: LamportTimestamp):
        """Receive a lock request from another process."""
        self._update_clock(request_ts.clock)
        self.request_queue.append(request_ts)
        self.request_queue.sort()
        
        print(f"{self.process_id}: Received request {request_ts}")
        # In practice: send ACK message back
        return LamportTimestamp(self._tick(), self.process_id, "ack")
    
    def receive_ack(self, from_process: str, ack_ts: int):
        """Receive acknowledgment from another process."""
        self._update_clock(ack_ts)
        self.acks_received.add(from_process)
        print(f"{self.process_id}: Received ack from {from_process}")
    
    def can_enter_critical_section(self) -> bool:
        """Check if this process can enter critical section."""
        if not self.request_queue:
            return False
            
        my_request = None
        for req in self.request_queue:
            if req.process_id == self.process_id:
                my_request = req
                break
        
        if my_request is None:
            return False
        
        # Condition 1: My request is at the head
        head_is_mine = self.request_queue[0].process_id == self.process_id
        
        # Condition 2: Received acks from all other processes
        all_acked = set(self.all_processes) == self.acks_received
        
        can_enter = head_is_mine and all_acked
        print(f"{self.process_id}: Can enter? head_is_mine={head_is_mine}, "
              f"all_acked={all_acked} -> {can_enter}")
        
        return can_enter
 
 
if __name__ == "__main__":
    demonstrate_total_order()

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

"""
Demonstration: Why Lamport Clocks Cannot Detect Concurrency
 
This example shows two concurrent events where Lamport timestamps
cannot distinguish from ordered events.
"""
 
def lamport_concurrency_problem():
    """
    Two processes have events that are concurrent,
    but Lamport timestamps don't reveal this.
    """
    
    print("Lamport Clock Concurrency Limitation")
    print("=" * 50)
    
    # Scenario 1: Ordered events
    print("
Scenario 1: P1 sends to P2, then P2 has local event")
    print("-" * 50)
    
    # P1: event a (timestamp 1), sends to P2
    # P2: receives (timestamp 2), then event b (timestamp 3)
    # a → b (definitely, via message)
    ts_a_scenario1 = 1
    ts_b_scenario1 = 3
    print(f"  Event a on P1: timestamp = {ts_a_scenario1}")
    print(f"  Event b on P2: timestamp = {ts_b_scenario1}")
    print(f"  L(a) < L(b): {ts_a_scenario1 < ts_b_scenario1}")
    print(f"  Relationship: a → b (a happened before b)")
    
    # Scenario 2: Concurrent events
    print("
Scenario 2: P1 and P2 have events with no message between them")
    print("-" * 50)
    
    # P1: local event a (timestamp 1)
    # P2: has done some work, local event b (timestamp 3)
    # a || b (concurrent - no messages exchanged)
    ts_a_scenario2 = 1
    ts_b_scenario2 = 3
    print(f"  Event a on P1: timestamp = {ts_a_scenario2}")
    print(f"  Event b on P2: timestamp = {ts_b_scenario2}")
    print(f"  L(a) < L(b): {ts_a_scenario2 < ts_b_scenario2}")
    print(f"  Relationship: a || b (CONCURRENT)")
    
    print("
" + "=" * 50)
    print("PROBLEM: Both scenarios have L(a)=1 and L(b)=3")
    print("Given ONLY timestamps (1, P1) and (3, P2), we CANNOT")
    print("distinguish ordered from concurrent events!")
    print("
This is the fundamental limitation of Lamport clocks.")
    print("Vector clocks solve this by tracking per-process knowledge.")
    
    
def lamport_datastore_problem():
    """
    Real-world problem: conflicts in a replicated data store.
    """
    
    print("
 
Replicated Data Store Conflict Scenario")
    print("=" * 50)
    
    # Two replicas of a key-value store
    # Two clients concurrently write different values
    
    print("
Timeline:")
    print("1. Replica A receives: SET x = 'alice' (Lamport ts = 10)")
    print("2. Replica B receives: SET x = 'bob' (Lamport ts = 12)")
    print("3. Replicas sync their logs...")
    
    print("
With Lamport clocks (last-write-wins):")
    print("  ts(bob) > ts(alice), so x = 'bob'")
    print("  Alice's write is discarded")
    
    print("
But wait! These writes might be concurrent:")
    print("  - Alice's client didn't know about Bob's write")
    print("  - Bob's client didn't know about Alice's write")
    print("  - Neither 'happened before' the other")
    print("
  Lamport clocks can't detect this - data loss may occur!")
    
    print("
With vector clocks:")
    print("  alice_vc = {A: 10, B: 5}  (Alice knows A's state up to 10, B's up to 5)")
    print("  bob_vc   = {A: 3, B: 12}  (Bob knows A's state up to 3, B's up to 12)")
    print("
  Neither vector dominates the other!")
    print("  {A:10,B:5} not ≤ {A:3,B:12} (10 > 3)")
    print("  {A:3,B:12} not ≤ {A:10,B:5} (12 > 5)")
    print("
  Therefore: CONCURRENT - preserve both or merge!")
 
 
if __name__ == "__main__":
    lamport_concurrency_problem()
    lamport_datastore_problem()

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

"""
Production-Quality Lamport Clock Implementation
 
Addresses practical concerns: thread safety, persistence,
overflow handling, and integration patterns.
"""
 
import struct
import json
from dataclasses import dataclass, field
from threading import Lock
from typing import Optional, Tuple
from pathlib import Path
 
 
@dataclass
class ProductionLamportClock:
    """
    Production-ready Lamport clock with:
    - Thread safety
    - Persistence support
    - Overflow detection
    - Metrics integration points
    """
    
    node_id: str
    _counter: int = field(default=0, repr=False)
    _lock: Lock = field(default_factory=Lock, repr=False)
    _high_water_mark: int = field(default=0, repr=False)  # For metrics
    
    # Overflow threshold (warn when approaching)
    MAX_SAFE_VALUE: int = (1 << 63) - 1  # Max 64-bit signed
    OVERFLOW_WARNING_THRESHOLD: int = (1 << 60)  # Warn at ~10% of max
    
    @property
    def value(self) -> int:
        with self._lock:
            return self._counter
    
    def tick(self) -> int:
        """
        Increment for local event.
        Returns new timestamp.
        """
        with self._lock:
            self._counter += 1
            self._check_overflow()
            self._update_high_water_mark()
            return self._counter
    
    def update(self, received: int) -> int:
        """
        Update clock on message receive.
        
        Args:
            received: Timestamp from received message
        
        Returns:
            New timestamp for receive event
        """
        with self._lock:
            self._counter = max(self._counter, received) + 1
            self._check_overflow()
            self._update_high_water_mark()
            return self._counter
    
    def _check_overflow(self):
        """Check for imminent overflow and handle appropriately."""
        if self._counter >= self.MAX_SAFE_VALUE:
            raise OverflowError(
                f"Lamport clock overflow on node {self.node_id}. "
                "Requires epoch reset procedure."
            )
        elif self._counter >= self.OVERFLOW_WARNING_THRESHOLD:
            # In production: emit warning metric
            # metrics.emit("lamport_clock.overflow_warning", node=self.node_id)
            pass
    
    def _update_high_water_mark(self):
        """Track high water mark for metrics."""
        if self._counter > self._high_water_mark:
            self._high_water_mark = self._counter
    
    def timestamp(self) -> 'LamportTimestampProduction':
        """Get current timestamp as comparable object."""
        return LamportTimestampProduction(self.value, self.node_id)
    
    # Persistence methods
    def save_state(self, path: Path):
        """Persist clock state for crash recovery."""
        state = {
            "node_id": self.node_id,
            "counter": self._counter,
            "high_water_mark": self._high_water_mark,
        }
        path.write_text(json.dumps(state))
    
    @classmethod
    def load_state(cls, path: Path) -> 'ProductionLamportClock':
        """Restore clock state after crash."""
        state = json.loads(path.read_text())
        clock = cls(node_id=state["node_id"])
        # Add safety margin after crash (we may have processed
        # events that weren't persisted)
        clock._counter = state["counter"] + 1000
        clock._high_water_mark = state["high_water_mark"]
        return clock
 
 
@dataclass(frozen=True)
class LamportTimestampProduction:
    """
    Immutable timestamp with comparison support.
    Suitable for use in sets, dicts, and sorting.
    """
    clock: int
    node_id: str
    
    def __lt__(self, other):
        if not isinstance(other, LamportTimestampProduction):
            return NotImplemented
        if self.clock != other.clock:
            return self.clock < other.clock
        return self.node_id < other.node_id
    
    def __le__(self, other):
        return self == other or self < other
    
    def __gt__(self, other):
        return other < self
    
    def __ge__(self, other):
        return self == other or self > other
    
    def to_bytes(self) -> bytes:
        """Serialize for network transmission."""
        # 8 bytes for clock + node_id as UTF-8 with length prefix
        node_bytes = self.node_id.encode('utf-8')
        return struct.pack('>Q', self.clock) + struct.pack('>H', len(node_bytes)) + node_bytes
    
    @classmethod
    def from_bytes(cls, data: bytes) -> Tuple['LamportTimestampProduction', int]:
        """
        Deserialize from bytes.
        Returns tuple of (timestamp, bytes_consumed).
        """
        clock = struct.unpack('>Q', data[:8])[0]
        node_len = struct.unpack('>H', data[8:10])[0]
        node_id = data[10:10+node_len].decode('utf-8')
        return cls(clock, node_id), 10 + node_len
 
 
# Integration example: Message with Lamport timestamp
@dataclass
class TimestampedMessage:
    """A message with embedded Lamport timestamp."""
    
    payload: bytes
    timestamp: LamportTimestampProduction
    
    def serialize(self) -> bytes:
        """Serialize message for network transmission."""
        ts_bytes = self.timestamp.to_bytes()
        payload_len = struct.pack('>I', len(self.payload))
        return ts_bytes + payload_len + self.payload
    
    @classmethod
    def deserialize(cls, data: bytes) -> 'TimestampedMessage':
        """Deserialize message from network."""
        ts, ts_len = LamportTimestampProduction.from_bytes(data)
        payload_len = struct.unpack('>I', data[ts_len:ts_len+4])[0]
        payload = data[ts_len+4:ts_len+4+payload_len]
        return cls(payload, ts)
 
 
# Usage pattern: sending and receiving messages
def message_send_receive_example():
    """Demonstrate proper message exchange pattern."""
    
    # Two nodes
    node_a = ProductionLamportClock("node-a")
    node_b = ProductionLamportClock("node-b")
    
    # Node A sends a message
    send_ts = node_a.tick()
    msg = TimestampedMessage(
        payload=b"Hello from A",
        timestamp=LamportTimestampProduction(send_ts, node_a.node_id)
    )
    
    # Serialize and "transmit"
    wire_format = msg.serialize()
    
    # Node B receives
    received_msg = TimestampedMessage.deserialize(wire_format)
    receive_ts = node_b.update(received_msg.timestamp.clock)
    
    print(f"A sent at: {send_ts}")
    print(f"B received at: {receive_ts}")
    print(f"B's clock properly advanced: {receive_ts > send_ts}")
 
 
if __name__ == "__main__":
    message_send_receive_example()