///| One contiguous counter interval requested from a peer, inclusive at both
///|
/// ends. Replica logs naturally map this range to dots.
pub(all) struct SyncRange {
replica : String
from_counter : Int
to_counter : Int
} derive(Eq, Debug)
///|
/// Summary exchanged before anti-entropy transfer.
pub(all) struct ReplicaDigest {
replica : String
frontier : VersionVector
} derive(Eq, Debug)
///|
/// Validation failures for anti-entropy metadata.
pub(all) enum SyncError {
EmptyDigestReplica
InvalidSyncRange(String, Int, Int)
} derive(Eq, Debug)
///|
/// Construct a digest for one replica.
pub fn ReplicaDigest::new(
replica : String,
frontier : VersionVector,
) -> Result[ReplicaDigest, SyncError] {
if replica.length() == 0 {
Err(EmptyDigestReplica)
} else {
Ok({ replica, frontier })
}
}
///|
/// Compute ranges present at the sender and missing at the receiver.
pub fn plan_sync(
sender : VersionVector,
receiver : VersionVector,
) -> Array[SyncRange] {
let output : Array[SyncRange] = []
for entry in sender.entries() {
let known = receiver.counter(entry.replica)
if entry.counter > known {
output.push({
replica: entry.replica,
from_counter: known + 1,
to_counter: entry.counter,
})
}
}
output
}
///|
/// Validate a range received from an untrusted peer.
pub fn SyncRange::new(
replica : String,
from_counter : Int,
to_counter : Int,
) -> Result[SyncRange, SyncError] {
if replica.length() == 0 {
Err(EmptyDigestReplica)
} else if from_counter <= 0 || to_counter < from_counter {
Err(InvalidSyncRange(replica, from_counter, to_counter))
} else {
Ok({ replica, from_counter, to_counter })
}
}
///|
/// Number of dots represented by a range.
pub fn SyncRange::length(self : SyncRange) -> Int {
self.to_counter - self.from_counter + 1
}
///|
/// Test whether a dot belongs to a requested range.
pub fn SyncRange::contains(self : SyncRange, dot : Dot) -> Bool {
dot.replica == self.replica &&
dot.counter >= self.from_counter &&
dot.counter <= self.to_counter
}
///| Select log entries requested by anti-entropy ranges while preserving the
///|
/// original causal release order.
pub fn select_sync_entries(
log : OperationLog,
ranges : Array[SyncRange],
) -> Array[LoggedOperation] {
let output : Array[LoggedOperation] = []
for entry in log.entries() {
let dot = Dot::new(entry.operation.replica, entry.operation.counter).unwrap()
for range in ranges {
if range.contains(dot) {
output.push(entry)
break
}
}
}
output
}
///|
/// Summarize a batch as a frontier. Useful after applying a complete response.
pub fn sync_batch_frontier(entries : Array[LoggedOperation]) -> VersionVector {
let vectors : Array[VectorEntry] = []
for entry in entries {
vectors.push({
replica: entry.operation.replica,
counter: entry.operation.counter,
})
}
VersionVector::from_entries(vectors).unwrap()
}
///|
/// Report gaps in a transferred batch relative to the requested ranges.
pub fn missing_sync_dots(
entries : Array[LoggedOperation],
ranges : Array[SyncRange],
) -> DotSet {
let mut missing = DotSet::new()
for range in ranges {
for counter in range.from_counter..<=range.to_counter {
let expected = Dot::new(range.replica, counter).unwrap()
let mut found = false
for entry in entries {
if entry.operation.replica == range.replica &&
entry.operation.counter == counter {
found = true
break
}
}
if !found {
missing = missing.add(expected)
}
}
}
missing
}
///|
/// Estimate transfer size as number of operations, without reading the log.
pub fn sync_operation_count(ranges : Array[SyncRange]) -> Int {
let mut total = 0
for range in ranges {
total = total + range.length()
}
total
}