nodedb_lite/nodedb/core/ops/
peer_identity.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Recovering this replica's producer identity when Origin refuses it.
4
5use nodedb_types::error::{NodeDbError, NodeDbResult};
6
7use crate::nodedb::core::types::NodeDbLite;
8use crate::nodedb::lock_ext::LockExt;
9use crate::storage::engine::StorageEngine;
10
11impl<S: StorageEngine> NodeDbLite<S> {
12    /// The Loro peer id every local operation is authored under.
13    pub fn peer_id(&self) -> u64 {
14        self.identity.lock_or_recover().peer_id
15    }
16
17    /// This instance's durable producer identity, as sent in a sync handshake.
18    #[cfg(not(target_arch = "wasm32"))]
19    pub(crate) fn sync_identity(&self) -> crate::identity::LiteIdentity {
20        self.identity.lock_or_recover().clone()
21    }
22
23    /// Adopt a new Loro peer id and re-author every document under it.
24    ///
25    /// Called when Origin refuses a delta because its peer id belongs to
26    /// another replica. Continuing under the refused id is not an option: the
27    /// server refuses every subsequent write, and were it not to, the CRDT
28    /// merge would discard them as replays of the owning replica's operations.
29    ///
30    /// The new id is persisted before the documents adopt it, so a crash
31    /// mid-rotation cannot leave the store authoring under an id its own
32    /// identity record no longer claims. The rebuilt documents are then
33    /// flushed, because they exist only in memory until they are — and a
34    /// restart that reloaded the pre-rotation snapshots would resurrect the
35    /// operations the rotation exists to abandon.
36    ///
37    /// Every row is queued for re-push: the rebuilt documents share no history
38    /// with what Origin holds, so the rotation is a resync, not a resume.
39    pub async fn rotate_peer_id(&self) -> NodeDbResult<u64> {
40        let guard = self.fts_state.admit_mutation().await;
41        let result = self.rotate_peer_id_admitted(guard.permit()).await;
42        guard.finish(result)
43    }
44
45    async fn rotate_peer_id_admitted(
46        &self,
47        permit: &crate::engine::fts::coordinator::TextMutationPermit,
48    ) -> NodeDbResult<u64> {
49        let _change = self.identity_change.lock().await;
50
51        // The identity is mutated on a copy so the store-wide guard is never
52        // held across the persist; `identity_change` is what keeps two
53        // rotations from interleaving on the same starting value.
54        let mut identity = self.identity.lock_or_recover().clone();
55        let new_peer_id = identity
56            .rotate_peer_id(&*self.storage)
57            .await
58            .map_err(|e| NodeDbError::storage(format!("peer-id rotation failed: {e}")))?;
59        *self.identity.lock_or_recover() = identity;
60
61        {
62            let mut crdt = self.crdt.lock_or_recover();
63            crdt.rotate_peer_id(new_peer_id)
64                .map_err(|e| NodeDbError::storage(format!("peer-id rotation failed: {e}")))?;
65        }
66
67        self.flush_admitted(permit).await?;
68
69        tracing::warn!(
70            peer_id = new_peer_id,
71            "adopted a new Loro peer id and queued every row for resync"
72        );
73        Ok(new_peer_id)
74    }
75
76    /// Replace the whole producer identity and re-author every document.
77    ///
78    /// Called when Origin reports this instance as forked: the history it
79    /// holds under this `lite_id` diverged from the one here, so the instance
80    /// cannot resume the producer stream. A new `lite_id` and epoch make it a
81    /// distinct producer, and the peer id goes with them — keeping it would
82    /// collide with the history the fork was detected against, which is the
83    /// same refusal one step later.
84    pub async fn regenerate_identity(&self) -> NodeDbResult<()> {
85        let guard = self.fts_state.admit_mutation().await;
86        let result = self.regenerate_identity_admitted(guard.permit()).await;
87        guard.finish(result)
88    }
89
90    async fn regenerate_identity_admitted(
91        &self,
92        permit: &crate::engine::fts::coordinator::TextMutationPermit,
93    ) -> NodeDbResult<()> {
94        let _change = self.identity_change.lock().await;
95
96        let mut identity = self.identity.lock_or_recover().clone();
97        identity
98            .regenerate(&*self.storage)
99            .await
100            .map_err(|e| NodeDbError::storage(format!("identity regeneration failed: {e}")))?;
101        let (lite_id, peer_id) = (identity.lite_id.clone(), identity.peer_id);
102        *self.identity.lock_or_recover() = identity;
103
104        {
105            let mut crdt = self.crdt.lock_or_recover();
106            crdt.rotate_peer_id(peer_id)
107                .map_err(|e| NodeDbError::storage(format!("identity regeneration failed: {e}")))?;
108        }
109
110        self.flush_admitted(permit).await?;
111
112        tracing::warn!(
113            %lite_id,
114            peer_id,
115            "fork reported by Origin — adopted a new producer identity and queued every row \
116             for resync"
117        );
118        Ok(())
119    }
120}
nodedb_lite/nodedb/core/ops/peer_identity.rs

nodedb_lite/nodedb/core/ops/
peer_identity.rs
