///|
// Pending changes wait for missing dependencies.
pub enum PendingChange {
Unknown(Change)
Known(Change)
} derive(Show)
///|
pub struct PendingChanges {
changes : Map[@types.PeerID, Map[@types.Counter, Array[PendingChange]]]
}
///|
pub fn PendingChanges::new() -> PendingChanges {
PendingChanges::{ changes: Map::new() }
}
///|
///|
fn remote_change_apply_state(
vv : @types.VersionVector,
change : Change,
) -> ChangeState {
let peer = change.id.peer
let span = change.ctr_span()
let vv_latest_ctr = vv.get(peer)
if vv_latest_ctr >= span.end {
return ChangeState::Applied
}
if vv_latest_ctr < span.start {
return ChangeState::AwaitingMissingDependency(change.id.inc(-1))
}
for dep in change.deps.iter() {
let dep_vv_latest_ctr = vv.get(dep.peer)
if dep_vv_latest_ctr - 1 < dep.counter {
return ChangeState::AwaitingMissingDependency(dep)
}
}
ChangeState::CanApplyDirectly
}
///|
fn push_pending(
pending : PendingChanges,
miss_dep : @types.ID,
change : PendingChange,
) -> Unit {
let peer = miss_dep.peer
let counter = miss_dep.counter
let per_peer = match pending.changes.get(peer) {
Some(map) => map
None => Map::new()
}
let bucket = match per_peer.get(counter) {
Some(existing) => existing
None => []
}
bucket.push(change)
per_peer[counter] = bucket
pending.changes[peer] = per_peer
}
///|
pub enum ChangeState {
Applied
CanApplyDirectly
AwaitingMissingDependency(@types.ID)
} derive(Show)
///|
pub fn OpLog::extend_pending_changes_with_unknown_lamport(
self : OpLog,
remote_changes : Array[Change],
) -> Result[Unit, @types.LoroError] {
for change in remote_changes {
match remote_change_apply_state(self.vv(), change) {
ChangeState::AwaitingMissingDependency(miss_dep) =>
push_pending(
self.pending_changes,
miss_dep,
PendingChange::Unknown(change),
)
ChangeState::Applied =>
return Err(@types.LoroError::InternalError("already applied"))
ChangeState::CanApplyDirectly =>
return Err(@types.LoroError::InternalError("can apply"))
}
}
Ok(())
}
///|
pub fn OpLog::try_apply_pending(
self : OpLog,
new_ids : Array[@types.ID],
) -> Unit {
let queue = new_ids
while queue.length() > 0 {
let id = match queue.pop() {
Some(value) => value
None => break
}
match self.pending_changes.changes.get(id.peer) {
Some(tree) => {
let to_remove : Array[@types.Counter] = []
for cnt, _ in tree {
if cnt <= id.counter {
to_remove.push(cnt)
}
}
let pending_sets : Array[Array[PendingChange]] = []
for cnt in to_remove {
match tree.get(cnt) {
Some(items) => {
pending_sets.push(items)
tree.remove(cnt)
}
None => ()
}
}
if tree.length() == 0 {
self.pending_changes.changes.remove(id.peer)
} else {
self.pending_changes.changes[id.peer] = tree
}
for pending_set in pending_sets {
for pending_change in pending_set {
match pending_change {
PendingChange::Unknown(change) =>
match remote_change_apply_state(self.dag.vv(), change) {
ChangeState::CanApplyDirectly => {
queue.push(change.id_last())
self.apply_change_from_remote(
PendingChange::Unknown(change),
)
}
ChangeState::Applied => ()
ChangeState::AwaitingMissingDependency(miss_dep) =>
push_pending(
self.pending_changes,
miss_dep,
PendingChange::Unknown(change),
)
}
PendingChange::Known(change) =>
match remote_change_apply_state(self.dag.vv(), change) {
ChangeState::CanApplyDirectly => {
queue.push(change.id_last())
self.apply_change_from_remote(PendingChange::Known(change))
}
ChangeState::Applied => ()
ChangeState::AwaitingMissingDependency(miss_dep) =>
push_pending(
self.pending_changes,
miss_dep,
PendingChange::Known(change),
)
}
}
}
}
}
None => ()
}
}
}
///|
pub fn OpLog::apply_change_from_remote(
self : OpLog,
pending_change : PendingChange,
) -> Unit {
let change = match pending_change {
PendingChange::Unknown(c) => c
PendingChange::Known(c) => c
}
let change = match self.dag.calc_unknown_lamport_change(change) {
Ok(c) => c
Err(_) => return
}
match self.trim_the_known_part_of_change(change) {
Some(trimmed) => self.insert_new_change(trimmed, false)
None => ()
}
}