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
// SPDX-License-Identifier: Apache-2.0
//! Recovering this replica's producer identity when Origin refuses it.
use nodedb_types::error::{NodeDbError, NodeDbResult};
use crate::nodedb::core::types::NodeDbLite;
use crate::nodedb::lock_ext::LockExt;
use crate::storage::engine::StorageEngine;
impl<S: StorageEngine> NodeDbLite<S> {
/// The Loro peer id every local operation is authored under.
pub fn peer_id(&self) -> u64 {
self.identity.lock_or_recover().peer_id
}
/// This instance's durable producer identity, as sent in a sync handshake.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn sync_identity(&self) -> crate::identity::LiteIdentity {
self.identity.lock_or_recover().clone()
}
/// Adopt a new Loro peer id and re-author every document under it.
///
/// Called when Origin refuses a delta because its peer id belongs to
/// another replica. Continuing under the refused id is not an option: the
/// server refuses every subsequent write, and were it not to, the CRDT
/// merge would discard them as replays of the owning replica's operations.
///
/// The new id is persisted before the documents adopt it, so a crash
/// mid-rotation cannot leave the store authoring under an id its own
/// identity record no longer claims. The rebuilt documents are then
/// flushed, because they exist only in memory until they are — and a
/// restart that reloaded the pre-rotation snapshots would resurrect the
/// operations the rotation exists to abandon.
///
/// Every row is queued for re-push: the rebuilt documents share no history
/// with what Origin holds, so the rotation is a resync, not a resume.
pub async fn rotate_peer_id(&self) -> NodeDbResult<u64> {
let guard = self.fts_state.admit_mutation().await;
let result = self.rotate_peer_id_admitted(guard.permit()).await;
guard.finish(result)
}
async fn rotate_peer_id_admitted(
&self,
permit: &crate::engine::fts::coordinator::TextMutationPermit,
) -> NodeDbResult<u64> {
let _change = self.identity_change.lock().await;
// The identity is mutated on a copy so the store-wide guard is never
// held across the persist; `identity_change` is what keeps two
// rotations from interleaving on the same starting value.
let mut identity = self.identity.lock_or_recover().clone();
let new_peer_id = identity
.rotate_peer_id(&*self.storage)
.await
.map_err(|e| NodeDbError::storage(format!("peer-id rotation failed: {e}")))?;
*self.identity.lock_or_recover() = identity;
{
let mut crdt = self.crdt.lock_or_recover();
crdt.rotate_peer_id(new_peer_id)
.map_err(|e| NodeDbError::storage(format!("peer-id rotation failed: {e}")))?;
}
self.flush_admitted(permit).await?;
tracing::warn!(
peer_id = new_peer_id,
"adopted a new Loro peer id and queued every row for resync"
);
Ok(new_peer_id)
}
/// Replace the whole producer identity and re-author every document.
///
/// Called when Origin reports this instance as forked: the history it
/// holds under this `lite_id` diverged from the one here, so the instance
/// cannot resume the producer stream. A new `lite_id` and epoch make it a
/// distinct producer, and the peer id goes with them — keeping it would
/// collide with the history the fork was detected against, which is the
/// same refusal one step later.
pub async fn regenerate_identity(&self) -> NodeDbResult<()> {
let guard = self.fts_state.admit_mutation().await;
let result = self.regenerate_identity_admitted(guard.permit()).await;
guard.finish(result)
}
async fn regenerate_identity_admitted(
&self,
permit: &crate::engine::fts::coordinator::TextMutationPermit,
) -> NodeDbResult<()> {
let _change = self.identity_change.lock().await;
let mut identity = self.identity.lock_or_recover().clone();
identity
.regenerate(&*self.storage)
.await
.map_err(|e| NodeDbError::storage(format!("identity regeneration failed: {e}")))?;
let (lite_id, peer_id) = (identity.lite_id.clone(), identity.peer_id);
*self.identity.lock_or_recover() = identity;
{
let mut crdt = self.crdt.lock_or_recover();
crdt.rotate_peer_id(peer_id)
.map_err(|e| NodeDbError::storage(format!("identity regeneration failed: {e}")))?;
}
self.flush_admitted(permit).await?;
tracing::warn!(
%lite_id,
peer_id,
"fork reported by Origin — adopted a new producer identity and queued every row \
for resync"
);
Ok(())
}
}