mleku/next.orly.dev
1
0
Fork 1
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Add foundational resources for elliptic curve operations and distributed systems

Browse Source 
Added detailed pseudocode for elliptic curve algorithms covering modular arithmetic, point operations, scalar multiplication, and coordinate conversions. Also introduced a comprehensive knowledge base for distributed systems, including CAP theorem, consistency models, consensus protocols (e.g., Paxos, Raft, PBFT, Nakamoto), and fault-tolerant design principles.
This commit is contained in:
mleku
2025-12-02 19:14:39 +00:00
parent feae79af1a
commit 3c17e975df
8 changed files with 3705 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1115
.claude/skills/distributed-systems/SKILL.md Normal file
View File

File diff suppressed because it is too large Load Diff

610
.claude/skills/distributed-systems/references/consensus-protocols.md Normal file
View File

@@ -0,0 +1,610 @@
# Consensus Protocols - Detailed Reference
Complete specifications and implementation details for major consensus protocols.
## Paxos Complete Specification
### Proposal Numbers
Proposal numbers must be:
- **Unique**: No two proposers use the same number
- **Totally ordered**: Any two can be compared
**Implementation**: `(round_number, proposer_id)` where proposer_id breaks ties.
### Single-Decree Paxos State
**Proposer state**:
```
proposal_number: int
value: any
```
**Acceptor state (persistent)**:
```
highest_promised: int # Highest proposal number promised
accepted_proposal: int # Number of accepted proposal (0 if none)
accepted_value: any # Value of accepted proposal (null if none)
```
### Message Format
**Prepare** (Phase 1a):
```
{
type: "PREPARE",
proposal_number: n
}
```
**Promise** (Phase 1b):
```
{
type: "PROMISE",
proposal_number: n,
accepted_proposal: m, # null if nothing accepted
accepted_value: v # null if nothing accepted
}
```
**Accept** (Phase 2a):
```
{
type: "ACCEPT",
proposal_number: n,
value: v
}
```
**Accepted** (Phase 2b):
```
{
type: "ACCEPTED",
proposal_number: n,
value: v
}
```
### Proposer Algorithm
```
function propose(value):
n = generate_proposal_number()
# Phase 1: Prepare
promises = []
for acceptor in acceptors:
send PREPARE(n) to acceptor
wait until |promises| > |acceptors|/2 or timeout
if timeout:
return FAILED
# Choose value
highest = max(promises, key=p.accepted_proposal)
if highest.accepted_value is not null:
value = highest.accepted_value
# Phase 2: Accept
accepts = []
for acceptor in acceptors:
send ACCEPT(n, value) to acceptor
wait until |accepts| > |acceptors|/2 or timeout
if timeout:
return FAILED
return SUCCESS(value)
```
### Acceptor Algorithm
```
on receive PREPARE(n):
if n > highest_promised:
highest_promised = n
persist(highest_promised)
reply PROMISE(n, accepted_proposal, accepted_value)
else:
# Optionally reply NACK(highest_promised)
ignore or reject
on receive ACCEPT(n, v):
if n >= highest_promised:
highest_promised = n
accepted_proposal = n
accepted_value = v
persist(highest_promised, accepted_proposal, accepted_value)
reply ACCEPTED(n, v)
else:
ignore or reject
```
### Multi-Paxos Optimization
**Stable leader**:
```
# Leader election (using Paxos or other method)
leader = elect_leader()
# Leader's Phase 1 for all future instances
leader sends PREPARE(n) for instance range [i, ∞)
# For each command:
function propose_as_leader(value, instance):
# Skip Phase 1 if already leader
for acceptor in acceptors:
send ACCEPT(n, value, instance) to acceptor
wait for majority ACCEPTED
return SUCCESS
```
### Paxos Safety Proof Sketch
**Invariant**: If a value v is chosen for instance i, no other value can be chosen.
**Proof**:
1. Value chosen → accepted by majority with proposal n
2. Any higher proposal n' must contact majority
3. Majorities intersect → at least one acceptor has accepted v
4. New proposer adopts v (or higher already-accepted value)
5. By induction, all future proposals use v
## Raft Complete Specification
### State
**All servers (persistent)**:
```
currentTerm: int # Latest term seen
votedFor: ServerId # Candidate voted for in current term (null if none)
log[]: LogEntry # Log entries
```
**All servers (volatile)**:
```
commitIndex: int # Highest log index known to be committed
lastApplied: int # Highest log index applied to state machine
```
**Leader (volatile, reinitialized after election)**:
```
nextIndex[]: int # For each server, next log index to send
matchIndex[]: int # For each server, highest log index replicated
```
**LogEntry**:
```
{
term: int,
command: any
}
```
### RequestVote RPC
**Request**:
```
{
term: int, # Candidate's term
candidateId: ServerId, # Candidate requesting vote
lastLogIndex: int, # Index of candidate's last log entry
lastLogTerm: int # Term of candidate's last log entry
}
```
**Response**:
```
{
term: int, # currentTerm, for candidate to update itself
voteGranted: bool # True if candidate received vote
}
```
**Receiver implementation**:
```
on receive RequestVote(term, candidateId, lastLogIndex, lastLogTerm):
if term < currentTerm:
return {term: currentTerm, voteGranted: false}
if term > currentTerm:
currentTerm = term
votedFor = null
convert to follower
# Check if candidate's log is at least as up-to-date as ours
ourLastTerm = log[len(log)-1].term if log else 0
ourLastIndex = len(log) - 1
logOK = (lastLogTerm > ourLastTerm) or
(lastLogTerm == ourLastTerm and lastLogIndex >= ourLastIndex)
if (votedFor is null or votedFor == candidateId) and logOK:
votedFor = candidateId
persist(currentTerm, votedFor)
reset election timer
return {term: currentTerm, voteGranted: true}
return {term: currentTerm, voteGranted: false}
```
### AppendEntries RPC
**Request**:
```
{
term: int, # Leader's term
leaderId: ServerId, # For follower to redirect clients
prevLogIndex: int, # Index of log entry preceding new ones
prevLogTerm: int, # Term of prevLogIndex entry
entries[]: LogEntry, # Log entries to store (empty for heartbeat)
leaderCommit: int # Leader's commitIndex
}
```
**Response**:
```
{
term: int, # currentTerm, for leader to update itself
success: bool # True if follower had matching prevLog entry
}
```
**Receiver implementation**:
```
on receive AppendEntries(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit):
if term < currentTerm:
return {term: currentTerm, success: false}
reset election timer
if term > currentTerm:
currentTerm = term
votedFor = null
convert to follower
# Check log consistency
if prevLogIndex >= len(log) or
(prevLogIndex >= 0 and log[prevLogIndex].term != prevLogTerm):
return {term: currentTerm, success: false}
# Append new entries (handling conflicts)
for i, entry in enumerate(entries):
index = prevLogIndex + 1 + i
if index < len(log):
if log[index].term != entry.term:
# Delete conflicting entry and all following
log = log[:index]
log.append(entry)
else:
log.append(entry)
persist(currentTerm, votedFor, log)
# Update commit index
if leaderCommit > commitIndex:
commitIndex = min(leaderCommit, len(log) - 1)
return {term: currentTerm, success: true}
```
### Leader Behavior
```
on becoming leader:
for each server:
nextIndex[server] = len(log)
matchIndex[server] = 0
start sending heartbeats
on receiving client command:
append entry to local log
persist log
send AppendEntries to all followers
on receiving AppendEntries response from server:
if response.success:
matchIndex[server] = prevLogIndex + len(entries)
nextIndex[server] = matchIndex[server] + 1
# Update commit index
for N from commitIndex+1 to len(log)-1:
if log[N].term == currentTerm and
|{s : matchIndex[s] >= N}| > |servers|/2:
commitIndex = N
else:
nextIndex[server] = max(1, nextIndex[server] - 1)
retry AppendEntries with lower prevLogIndex
on commitIndex update:
while lastApplied < commitIndex:
lastApplied++
apply log[lastApplied].command to state machine
```
### Election Timeout
```
on election timeout (follower or candidate):
currentTerm++
convert to candidate
votedFor = self
persist(currentTerm, votedFor)
reset election timer
votes = 1 # Vote for self
for each server except self:
send RequestVote(currentTerm, self, lastLogIndex, lastLogTerm)
wait for responses or timeout:
if received votes > |servers|/2:
become leader
if received AppendEntries from valid leader:
become follower
if timeout:
start new election
```
## PBFT Complete Specification
### Message Types
**REQUEST**:
```
{
type: "REQUEST",
operation: o, # Operation to execute
timestamp: t, # Client timestamp (for reply matching)
client: c # Client identifier
}
```
**PRE-PREPARE**:
```
{
type: "PRE-PREPARE",
view: v, # Current view number
sequence: n, # Sequence number
digest: d, # Hash of request
request: m # The request message
}
signature(primary)
```
**PREPARE**:
```
{
type: "PREPARE",
view: v,
sequence: n,
digest: d,
replica: i # Sending replica
}
signature(replica_i)
```
**COMMIT**:
```
{
type: "COMMIT",
view: v,
sequence: n,
digest: d,
replica: i
}
signature(replica_i)
```
**REPLY**:
```
{
type: "REPLY",
view: v,
timestamp: t,
client: c,
replica: i,
result: r # Execution result
}
signature(replica_i)
```
### Replica State
```
view: int # Current view
sequence: int # Last assigned sequence number (primary)
log[]: {request, prepares, commits, state} # Log of requests
prepared_certificates: {} # Prepared certificates (2f+1 prepares)
committed_certificates: {} # Committed certificates (2f+1 commits)
h: int # Low water mark
H: int # High water mark (h + L)
```
### Normal Operation Protocol
**Primary (replica p = v mod n)**:
```
on receive REQUEST(m) from client:
if not primary for current view:
forward to primary
return
n = assign_sequence_number()
d = hash(m)
broadcast PRE-PREPARE(v, n, d, m) to all replicas
add to log
```
**All replicas**:
```
on receive PRE-PREPARE(v, n, d, m) from primary:
if v != current_view:
ignore
if already accepted pre-prepare for (v, n) with different digest:
ignore
if not in_view_as_backup(v):
ignore
if not h < n <= H:
ignore # Outside sequence window
# Valid pre-prepare
add to log
broadcast PREPARE(v, n, d, i) to all replicas
on receive PREPARE(v, n, d, j) from replica j:
if v != current_view:
ignore
add to log[n].prepares
if |log[n].prepares| >= 2f and not already_prepared(v, n, d):
# Prepared certificate complete
mark as prepared
broadcast COMMIT(v, n, d, i) to all replicas
on receive COMMIT(v, n, d, j) from replica j:
if v != current_view:
ignore
add to log[n].commits
if |log[n].commits| >= 2f + 1 and prepared(v, n, d):
# Committed certificate complete
if all entries < n are committed:
execute(m)
send REPLY(v, t, c, i, result) to client
```
### View Change Protocol
**Timeout trigger**:
```
on request timeout (no progress):
view_change_timeout++
broadcast VIEW-CHANGE(v+1, n, C, P, i)
where:
n = last stable checkpoint sequence number
C = checkpoint certificate (2f+1 checkpoint messages)
P = set of prepared certificates for messages after n
```
**VIEW-CHANGE**:
```
{
type: "VIEW-CHANGE",
view: v, # New view number
sequence: n, # Checkpoint sequence
checkpoints: C, # Checkpoint certificate
prepared: P, # Set of prepared certificates
replica: i
}
signature(replica_i)
```
**New primary (p' = v mod n)**:
```
on receive 2f VIEW-CHANGE for view v:
V = set of valid view-change messages
# Compute O: set of requests to re-propose
O = {}
for seq in max_checkpoint_seq(V) to max_seq(V):
if exists prepared certificate for seq in V:
O[seq] = request from certificate
else:
O[seq] = null-request # No-op
broadcast NEW-VIEW(v, V, O)
# Re-run protocol for requests in O
for seq, request in O:
if request != null:
send PRE-PREPARE(v, seq, hash(request), request)
```
**NEW-VIEW**:
```
{
type: "NEW-VIEW",
view: v,
view_changes: V, # 2f+1 view-change messages
pre_prepares: O # Set of pre-prepare messages
}
signature(primary)
```
### Checkpointing
Periodic stable checkpoints to garbage collect logs:
```
every K requests:
state_hash = hash(state_machine_state)
broadcast CHECKPOINT(n, state_hash, i)
on receive 2f+1 CHECKPOINT for (n, d):
if all digests match:
create stable checkpoint
h = n # Move low water mark
garbage_collect(entries < n)
```
## HotStuff Protocol
Linear complexity BFT using threshold signatures.
### Key Innovation
- **Three-phase**: prepare → pre-commit → commit → decide
- **Pipelining**: Next proposal starts before current finishes
- **Threshold signatures**: O(n) total messages instead of O(n²)
### Message Flow
```
Phase 1 (Prepare):
Leader: broadcast PREPARE(v, node)
Replicas: sign and send partial signature to leader
Leader: aggregate into prepare certificate QC
Phase 2 (Pre-commit):
Leader: broadcast PRE-COMMIT(v, QC_prepare)
Replicas: sign and send partial signature
Leader: aggregate into pre-commit certificate
Phase 3 (Commit):
Leader: broadcast COMMIT(v, QC_precommit)
Replicas: sign and send partial signature
Leader: aggregate into commit certificate
Phase 4 (Decide):
Leader: broadcast DECIDE(v, QC_commit)
Replicas: execute and commit
```
### Pipelining
```
Block k: [prepare] [pre-commit] [commit] [decide]
Block k+1: [prepare] [pre-commit] [commit] [decide]
Block k+2: [prepare] [pre-commit] [commit] [decide]
```
Each phase of block k+1 piggybacks on messages for block k.
## Protocol Comparison Matrix
| Feature | Paxos | Raft | PBFT | HotStuff |
|---------|-------|------|------|----------|
| Fault model | Crash | Crash | Byzantine | Byzantine |
| Fault tolerance | f with 2f+1 | f with 2f+1 | f with 3f+1 | f with 3f+1 |
| Message complexity | O(n) | O(n) | O(n²) | O(n) |
| Leader required | No (helps) | Yes | Yes | Yes |
| Phases | 2 | 2 | 3 | 3 |
| View change | Complex | Simple | Complex | Simple |

610
.claude/skills/distributed-systems/references/logical-clocks.md Normal file
View File

@@ -0,0 +1,610 @@
# Logical Clocks - Implementation Reference
Detailed implementations and algorithms for causality tracking.
## Lamport Clock Implementation
### Data Structure
```go
type LamportClock struct {
counter uint64
mu sync.Mutex
}
func NewLamportClock() *LamportClock {
return &LamportClock{counter: 0}
}
```
### Operations
```go
// Tick increments clock for local event
func (c *LamportClock) Tick() uint64 {
c.mu.Lock()
defer c.mu.Unlock()
c.counter++
return c.counter
}
// Send returns timestamp for outgoing message
func (c *LamportClock) Send() uint64 {
return c.Tick()
}
// Receive updates clock based on incoming message timestamp
func (c *LamportClock) Receive(msgTime uint64) uint64 {
c.mu.Lock()
defer c.mu.Unlock()
if msgTime > c.counter {
c.counter = msgTime
}
c.counter++
return c.counter
}
// Time returns current clock value without incrementing
func (c *LamportClock) Time() uint64 {
c.mu.Lock()
defer c.mu.Unlock()
return c.counter
}
```
### Usage Example
```go
// Process A
clockA := NewLamportClock()
e1 := clockA.Tick() // Event 1: time=1
msgTime := clockA.Send() // Send: time=2
// Process B
clockB := NewLamportClock()
e2 := clockB.Tick() // Event 2: time=1
e3 := clockB.Receive(msgTime) // Receive: time=3 (max(1,2)+1)
```
## Vector Clock Implementation
### Data Structure
```go
type VectorClock struct {
clocks map[string]uint64 // processID -> logical time
self string // this process's ID
mu sync.RWMutex
}
func NewVectorClock(processID string, allProcesses []string) *VectorClock {
clocks := make(map[string]uint64)
for _, p := range allProcesses {
clocks[p] = 0
}
return &VectorClock{
clocks: clocks,
self: processID,
}
}
```
### Operations
```go
// Tick increments own clock
func (vc *VectorClock) Tick() map[string]uint64 {
vc.mu.Lock()
defer vc.mu.Unlock()
vc.clocks[vc.self]++
return vc.copy()
}
// Send returns copy of vector for message
func (vc *VectorClock) Send() map[string]uint64 {
return vc.Tick()
}
// Receive merges incoming vector and increments
func (vc *VectorClock) Receive(incoming map[string]uint64) map[string]uint64 {
vc.mu.Lock()
defer vc.mu.Unlock()
// Merge: take max of each component
for pid, time := range incoming {
if time > vc.clocks[pid] {
vc.clocks[pid] = time
}
}
// Increment own clock
vc.clocks[vc.self]++
return vc.copy()
}
// copy returns a copy of the vector
func (vc *VectorClock) copy() map[string]uint64 {
result := make(map[string]uint64)
for k, v := range vc.clocks {
result[k] = v
}
return result
}
```
### Comparison Functions
```go
// Compare returns ordering relationship between two vectors
type Ordering int
const (
Equal Ordering = iota // V1 == V2
HappenedBefore // V1 < V2
HappenedAfter // V1 > V2
Concurrent // V1 || V2
)
func Compare(v1, v2 map[string]uint64) Ordering {
less := false
greater := false
// Get all keys
allKeys := make(map[string]bool)
for k := range v1 {
allKeys[k] = true
}
for k := range v2 {
allKeys[k] = true
}
for k := range allKeys {
t1 := v1[k] // 0 if not present
t2 := v2[k]
if t1 < t2 {
less = true
}
if t1 > t2 {
greater = true
}
}
if !less && !greater {
return Equal
}
if less && !greater {
return HappenedBefore
}
if greater && !less {
return HappenedAfter
}
return Concurrent
}
// IsConcurrent checks if two events are concurrent
func IsConcurrent(v1, v2 map[string]uint64) bool {
return Compare(v1, v2) == Concurrent
}
// HappenedBefore checks if v1 -> v2 (v1 causally precedes v2)
func HappenedBefore(v1, v2 map[string]uint64) bool {
return Compare(v1, v2) == HappenedBefore
}
```
## Interval Tree Clock Implementation
### Data Structures
```go
// ID represents the identity tree
type ID struct {
IsLeaf bool
Value int // 0 or 1 for leaves
Left *ID // nil for leaves
Right *ID
}
// Stamp represents the event tree
type Stamp struct {
Base int
Left *Stamp // nil for leaf stamps
Right *Stamp
}
// ITC combines ID and Stamp
type ITC struct {
ID *ID
Stamp *Stamp
}
```
### ID Operations
```go
// NewSeedID creates initial full ID (1)
func NewSeedID() *ID {
return &ID{IsLeaf: true, Value: 1}
}
// Fork splits an ID into two
func (id *ID) Fork() (*ID, *ID) {
if id.IsLeaf {
if id.Value == 0 {
// Cannot fork zero ID
return &ID{IsLeaf: true, Value: 0},
&ID{IsLeaf: true, Value: 0}
}
// Split full ID into left and right halves
return &ID{
IsLeaf: false,
Left: &ID{IsLeaf: true, Value: 1},
Right: &ID{IsLeaf: true, Value: 0},
},
&ID{
IsLeaf: false,
Left: &ID{IsLeaf: true, Value: 0},
Right: &ID{IsLeaf: true, Value: 1},
}
}
// Fork from non-leaf: give half to each
if id.Left.IsLeaf && id.Left.Value == 0 {
// Left is zero, fork right
newRight1, newRight2 := id.Right.Fork()
return &ID{IsLeaf: false, Left: id.Left, Right: newRight1},
&ID{IsLeaf: false, Left: &ID{IsLeaf: true, Value: 0}, Right: newRight2}
}
if id.Right.IsLeaf && id.Right.Value == 0 {
// Right is zero, fork left
newLeft1, newLeft2 := id.Left.Fork()
return &ID{IsLeaf: false, Left: newLeft1, Right: id.Right},
&ID{IsLeaf: false, Left: newLeft2, Right: &ID{IsLeaf: true, Value: 0}}
}
// Both have IDs, split
return &ID{IsLeaf: false, Left: id.Left, Right: &ID{IsLeaf: true, Value: 0}},
&ID{IsLeaf: false, Left: &ID{IsLeaf: true, Value: 0}, Right: id.Right}
}
// Join merges two IDs
func Join(id1, id2 *ID) *ID {
if id1.IsLeaf && id1.Value == 0 {
return id2
}
if id2.IsLeaf && id2.Value == 0 {
return id1
}
if id1.IsLeaf && id2.IsLeaf && id1.Value == 1 && id2.Value == 1 {
return &ID{IsLeaf: true, Value: 1}
}
// Normalize to non-leaf
left1 := id1.Left
right1 := id1.Right
left2 := id2.Left
right2 := id2.Right
if id1.IsLeaf {
left1 = id1
right1 = id1
}
if id2.IsLeaf {
left2 = id2
right2 = id2
}
newLeft := Join(left1, left2)
newRight := Join(right1, right2)
return normalize(&ID{IsLeaf: false, Left: newLeft, Right: newRight})
}
func normalize(id *ID) *ID {
if !id.IsLeaf {
if id.Left.IsLeaf && id.Right.IsLeaf &&
id.Left.Value == id.Right.Value {
return &ID{IsLeaf: true, Value: id.Left.Value}
}
}
return id
}
```
### Stamp Operations
```go
// NewStamp creates initial stamp (0)
func NewStamp() *Stamp {
return &Stamp{Base: 0}
}
// Event increments the stamp for the given ID
func Event(id *ID, stamp *Stamp) *Stamp {
if id.IsLeaf {
if id.Value == 1 {
return &Stamp{Base: stamp.Base + 1}
}
return stamp // Cannot increment with zero ID
}
// Non-leaf ID: fill where we have ID
if id.Left.IsLeaf && id.Left.Value == 1 {
// Have left ID, increment left
newLeft := Event(&ID{IsLeaf: true, Value: 1}, getLeft(stamp))
return normalizeStamp(&Stamp{
Base: stamp.Base,
Left: newLeft,
Right: getRight(stamp),
})
}
if id.Right.IsLeaf && id.Right.Value == 1 {
newRight := Event(&ID{IsLeaf: true, Value: 1}, getRight(stamp))
return normalizeStamp(&Stamp{
Base: stamp.Base,
Left: getLeft(stamp),
Right: newRight,
})
}
// Both non-zero, choose lower side
leftMax := maxStamp(getLeft(stamp))
rightMax := maxStamp(getRight(stamp))
if leftMax <= rightMax {
return normalizeStamp(&Stamp{
Base: stamp.Base,
Left: Event(id.Left, getLeft(stamp)),
Right: getRight(stamp),
})
}
return normalizeStamp(&Stamp{
Base: stamp.Base,
Left: getLeft(stamp),
Right: Event(id.Right, getRight(stamp)),
})
}
func getLeft(s *Stamp) *Stamp {
if s.Left == nil {
return &Stamp{Base: 0}
}
return s.Left
}
func getRight(s *Stamp) *Stamp {
if s.Right == nil {
return &Stamp{Base: 0}
}
return s.Right
}
func maxStamp(s *Stamp) int {
if s.Left == nil && s.Right == nil {
return s.Base
}
left := 0
right := 0
if s.Left != nil {
left = maxStamp(s.Left)
}
if s.Right != nil {
right = maxStamp(s.Right)
}
max := left
if right > max {
max = right
}
return s.Base + max
}
// JoinStamps merges two stamps
func JoinStamps(s1, s2 *Stamp) *Stamp {
// Take max at each level
base := s1.Base
if s2.Base > base {
base = s2.Base
}
// Adjust for base difference
adj1 := s1.Base
adj2 := s2.Base
return normalizeStamp(&Stamp{
Base: base,
Left: joinStampsRecursive(s1.Left, s2.Left, adj1-base, adj2-base),
Right: joinStampsRecursive(s1.Right, s2.Right, adj1-base, adj2-base),
})
}
func normalizeStamp(s *Stamp) *Stamp {
if s.Left == nil && s.Right == nil {
return s
}
if s.Left != nil && s.Right != nil {
if s.Left.Base > 0 && s.Right.Base > 0 {
min := s.Left.Base
if s.Right.Base < min {
min = s.Right.Base
}
return &Stamp{
Base: s.Base + min,
Left: &Stamp{Base: s.Left.Base - min, Left: s.Left.Left, Right: s.Left.Right},
Right: &Stamp{Base: s.Right.Base - min, Left: s.Right.Left, Right: s.Right.Right},
}
}
}
return s
}
```
## Hybrid Logical Clock Implementation
```go
type HLC struct {
l int64 // logical component (physical time)
c int64 // counter
mu sync.Mutex
}
func NewHLC() *HLC {
return &HLC{l: 0, c: 0}
}
type HLCTimestamp struct {
L int64
C int64
}
func (hlc *HLC) physicalTime() int64 {
return time.Now().UnixNano()
}
// Now returns current HLC timestamp for local/send event
func (hlc *HLC) Now() HLCTimestamp {
hlc.mu.Lock()
defer hlc.mu.Unlock()
pt := hlc.physicalTime()
if pt > hlc.l {
hlc.l = pt
hlc.c = 0
} else {
hlc.c++
}
return HLCTimestamp{L: hlc.l, C: hlc.c}
}
// Update updates HLC based on received timestamp
func (hlc *HLC) Update(received HLCTimestamp) HLCTimestamp {
hlc.mu.Lock()
defer hlc.mu.Unlock()
pt := hlc.physicalTime()
if pt > hlc.l && pt > received.L {
hlc.l = pt
hlc.c = 0
} else if received.L > hlc.l {
hlc.l = received.L
hlc.c = received.C + 1
} else if hlc.l > received.L {
hlc.c++
} else { // hlc.l == received.L
if received.C > hlc.c {
hlc.c = received.C + 1
} else {
hlc.c++
}
}
return HLCTimestamp{L: hlc.l, C: hlc.c}
}
// Compare compares two HLC timestamps
func (t1 HLCTimestamp) Compare(t2 HLCTimestamp) int {
if t1.L < t2.L {
return -1
}
if t1.L > t2.L {
return 1
}
if t1.C < t2.C {
return -1
}
if t1.C > t2.C {
return 1
}
return 0
}
```
## Causal Broadcast Implementation
```go
type CausalBroadcast struct {
vc *VectorClock
pending []PendingMessage
deliver func(Message)
mu sync.Mutex
}
type PendingMessage struct {
Msg Message
Timestamp map[string]uint64
}
func NewCausalBroadcast(processID string, processes []string, deliver func(Message)) *CausalBroadcast {
return &CausalBroadcast{
vc: NewVectorClock(processID, processes),
pending: make([]PendingMessage, 0),
deliver: deliver,
}
}
// Broadcast sends a message to all processes
func (cb *CausalBroadcast) Broadcast(msg Message) map[string]uint64 {
cb.mu.Lock()
defer cb.mu.Unlock()
timestamp := cb.vc.Send()
// Actual network broadcast would happen here
return timestamp
}
// Receive handles an incoming message
func (cb *CausalBroadcast) Receive(msg Message, sender string, timestamp map[string]uint64) {
cb.mu.Lock()
defer cb.mu.Unlock()
// Add to pending
cb.pending = append(cb.pending, PendingMessage{Msg: msg, Timestamp: timestamp})
// Try to deliver pending messages
cb.tryDeliver()
}
func (cb *CausalBroadcast) tryDeliver() {
changed := true
for changed {
changed = false
for i, pending := range cb.pending {
if cb.canDeliver(pending.Timestamp) {
// Deliver message
cb.vc.Receive(pending.Timestamp)
cb.deliver(pending.Msg)
// Remove from pending
cb.pending = append(cb.pending[:i], cb.pending[i+1:]...)
changed = true
break
}
}
}
}
func (cb *CausalBroadcast) canDeliver(msgVC map[string]uint64) bool {
currentVC := cb.vc.clocks
for pid, msgTime := range msgVC {
if pid == cb.vc.self {
// Must be next expected from sender
if msgTime != currentVC[pid]+1 {
return false
}
} else {
// All other dependencies must be satisfied
if msgTime > currentVC[pid] {
return false
}
}
}
return true
}
```