loro_internal/
txn.rs

1use std::{
2    borrow::Cow,
3    mem::take,
4    sync::{Arc, Weak},
5};
6
7use enum_as_inner::EnumAsInner;
8use generic_btree::rle::{HasLength as RleHasLength, Mergeable as GBSliceable};
9use loro_common::{ContainerType, IdLp, IdSpan, LoroResult};
10use loro_delta::{array_vec::ArrayVec, DeltaRopeBuilder};
11use rle::{HasLength, Mergable, RleVec, Sliceable};
12use rustc_hash::FxHashMap;
13use smallvec::{smallvec, SmallVec};
14
15use crate::{
16    change::{Change, Lamport, Timestamp},
17    container::{
18        idx::ContainerIdx,
19        list::list_op::{DeleteSpan, InnerListOp},
20        richtext::Style,
21        IntoContainerId,
22    },
23    delta::{ResolvedMapDelta, ResolvedMapValue, StyleMeta, StyleMetaItem, TreeDiff, TreeDiffItem},
24    encoding::export_fast_updates_in_range,
25    event::{Diff, ListDeltaMeta, TextDiff},
26    handler::{Handler, ValueOrHandler},
27    id::{Counter, PeerID, ID},
28    lock::{LoroMutex, LoroMutexGuard},
29    loro::CommitOptions,
30    op::{Op, RawOp, RawOpContent},
31    pre_commit::{ChangeModifier, PreCommitCallbackPayload},
32    span::HasIdSpan,
33    version::Frontiers,
34    ChangeMeta, InternalString, LoroDoc, LoroDocInner, LoroError, LoroValue,
35};
36
37use super::{
38    arena::SharedArena,
39    event::{InternalContainerDiff, InternalDocDiff},
40    handler::{ListHandler, MapHandler, TextHandler, TreeHandler},
41    oplog::OpLog,
42    state::DocState,
43};
44
45impl crate::LoroDoc {
46    /// Create a new transaction.
47    /// Every ops created inside one transaction will be packed into a single
48    /// [Change].
49    ///
50    /// There can only be one active transaction at a time for a [LoroDoc].
51    #[inline(always)]
52    pub fn txn(&self) -> Result<Transaction, LoroError> {
53        self.txn_with_origin("")
54    }
55
56    /// Create a new transaction with specified origin.
57    ///
58    /// The origin will be propagated to the events.
59    /// There can only be one active transaction at a time for a [LoroDoc].
60    pub fn txn_with_origin(&self, origin: &str) -> Result<Transaction, LoroError> {
61        if !self.can_edit() {
62            return Err(LoroError::TransactionError(
63                String::from("LoroDoc is in readonly detached mode. To make it writable in detached mode, call `set_detached_editing(true)`.").into_boxed_str(),
64            ));
65        }
66
67        let mut txn = Transaction::new_with_origin(self.inner.clone(), origin.into())?;
68
69        let obs = self.observer.clone();
70        let local_update_subs_weak = self.local_update_subs.downgrade();
71        txn.set_on_commit(Box::new(move |state, oplog, id_span| {
72            let mut state = state.lock();
73            let events = state.take_events();
74            drop(state);
75            for event in events {
76                obs.emit(event);
77            }
78
79            if id_span.atom_len() == 0 {
80                return;
81            }
82
83            if let Some(local_update_subs) = local_update_subs_weak.upgrade() {
84                if !local_update_subs.inner().is_empty() {
85                    let bytes = { export_fast_updates_in_range(&oplog.lock(), &[id_span]) };
86                    local_update_subs.emit(&(), bytes);
87                }
88            }
89        }));
90
91        Ok(txn)
92    }
93
94    pub fn start_auto_commit(&self) {
95        self.auto_commit
96            .store(true, std::sync::atomic::Ordering::Release);
97        let mut self_txn = self.txn.lock();
98        if self_txn.is_some() || !self.can_edit() {
99            return;
100        }
101
102        let txn = self
103            .txn()
104            .expect("auto-commit should be able to create a transaction");
105        self_txn.replace(txn);
106    }
107
108    #[inline]
109    pub fn renew_txn_if_auto_commit(&self, options: Option<CommitOptions>) {
110        if self.auto_commit.load(std::sync::atomic::Ordering::Acquire) && self.can_edit() {
111            let mut self_txn = self.txn.lock();
112            if self_txn.is_some() {
113                return;
114            }
115
116            let mut txn = self
117                .txn()
118                .expect("auto-commit should be able to renew a transaction");
119            if let Some(options) = options {
120                txn.set_options(options);
121            }
122            self_txn.replace(txn);
123        }
124    }
125
126    #[inline]
127    pub(crate) fn _renew_txn_if_auto_commit_with_guard(
128        &self,
129        options: Option<CommitOptions>,
130        mut guard: LoroMutexGuard<Option<Transaction>>,
131    ) {
132        if self.auto_commit.load(std::sync::atomic::Ordering::Acquire) && self.can_edit() {
133            if guard.is_some() {
134                return;
135            }
136
137            let mut txn = self
138                .txn()
139                .expect("auto-commit should be able to renew a transaction");
140            if let Some(options) = options {
141                txn.set_options(options);
142            }
143            guard.replace(txn);
144        }
145    }
146}
147
148pub(crate) type OnCommitFn =
149    Box<dyn FnOnce(&Arc<LoroMutex<DocState>>, &Arc<LoroMutex<OpLog>>, IdSpan) + Sync + Send>;
150
151pub struct Transaction {
152    peer: PeerID,
153    origin: InternalString,
154    start_counter: Counter,
155    next_counter: Counter,
156    start_lamport: Lamport,
157    next_lamport: Lamport,
158    doc: Weak<LoroDocInner>,
159    frontiers: Frontiers,
160    local_ops: RleVec<[Op; 1]>, // TODO: use a more efficient data structure
161    event_hints: FxHashMap<ContainerIdx, Vec<EventHint>>,
162    pub(super) arena: SharedArena,
163    finished: bool,
164    on_commit: Option<OnCommitFn>,
165    timestamp: Option<Timestamp>,
166    msg: Option<Arc<str>>,
167    latest_timestamp: Timestamp,
168    pub(super) is_peer_first_appearance: bool,
169}
170
171impl std::fmt::Debug for Transaction {
172    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
173        f.debug_struct("Transaction")
174            .field("peer", &self.peer)
175            .field("origin", &self.origin)
176            .field("start_counter", &self.start_counter)
177            .field("next_counter", &self.next_counter)
178            .field("start_lamport", &self.start_lamport)
179            .field("next_lamport", &self.next_lamport)
180            .field("frontiers", &self.frontiers)
181            .field("local_ops", &self.local_ops)
182            .field("event_hints", &self.event_hints)
183            .field("arena", &self.arena)
184            .field("finished", &self.finished)
185            .field("on_commit", &self.on_commit.is_some())
186            .field("timestamp", &self.timestamp)
187            .finish()
188    }
189}
190
191/// We can infer local events directly from the local behavior. This enum is used to
192/// record them, so that we can avoid recalculate them when we commit the transaction.
193///
194/// For example, when we insert a text in wasm, users use the utf16 index to send the
195/// command. However, internally loro will convert it to unicode index. But the users
196/// still need events that are in utf16 index. To avoid the round trip, we record the
197/// events here.
198#[derive(Debug, Clone, EnumAsInner)]
199pub(super) enum EventHint {
200    Mark {
201        start: u32,
202        end: u32,
203        style: Style,
204    },
205    InsertText {
206        /// pos is a Unicode index. If wasm, it's a UTF-16 index.
207        pos: u32,
208        event_len: u32,
209        unicode_len: u32,
210        styles: StyleMeta,
211    },
212    /// pos is a Unicode index. If wasm, it's a UTF-16 index.
213    DeleteText {
214        span: DeleteSpan,
215        unicode_len: usize,
216    },
217    InsertList {
218        len: u32,
219        pos: usize,
220    },
221    SetList {
222        index: usize,
223        value: LoroValue,
224    },
225    Move {
226        value: LoroValue,
227        from: u32,
228        to: u32,
229    },
230    DeleteList(DeleteSpan),
231    Map {
232        key: InternalString,
233        value: Option<LoroValue>,
234    },
235    // use vec because we could bring back some node that has children
236    Tree(SmallVec<[TreeDiffItem; 1]>),
237    MarkEnd,
238    #[cfg(feature = "counter")]
239    Counter(f64),
240}
241
242impl generic_btree::rle::HasLength for EventHint {
243    fn rle_len(&self) -> usize {
244        match self {
245            EventHint::Mark { .. } => 1,
246            EventHint::InsertText {
247                unicode_len: len, ..
248            } => *len as usize,
249            EventHint::DeleteText { unicode_len, .. } => *unicode_len,
250            EventHint::InsertList { len, .. } => *len as usize,
251            EventHint::DeleteList(d) => d.len(),
252            EventHint::Map { .. } => 1,
253            EventHint::Tree(_) => 1,
254            EventHint::MarkEnd => 1,
255            EventHint::Move { .. } => 1,
256            EventHint::SetList { .. } => 1,
257            #[cfg(feature = "counter")]
258            EventHint::Counter(_) => 1,
259        }
260    }
261}
262
263impl generic_btree::rle::Mergeable for EventHint {
264    fn can_merge(&self, rhs: &Self) -> bool {
265        match (self, rhs) {
266            (
267                EventHint::InsertText {
268                    pos,
269                    unicode_len: _,
270                    event_len,
271                    styles,
272                },
273                EventHint::InsertText {
274                    pos: r_pos,
275                    styles: r_styles,
276                    ..
277                },
278            ) => *pos + *event_len == *r_pos && styles == r_styles,
279            (EventHint::InsertList { pos, len }, EventHint::InsertList { pos: pos_right, .. }) => {
280                pos + *len as usize == *pos_right
281            }
282            // We don't merge delete text because it's hard to infer the correct pos to split:
283            // `range` param is in unicode range, but the delete text event is in UTF-16 range.
284            // Without the original text, it's impossible to convert the range.
285            (EventHint::DeleteText { span, .. }, EventHint::DeleteText { span: r, .. }) => {
286                span.is_mergable(r, &())
287            }
288            (EventHint::DeleteList(l), EventHint::DeleteList(r)) => l.is_mergable(r, &()),
289            _ => false,
290        }
291    }
292
293    fn merge_right(&mut self, rhs: &Self) {
294        match (self, rhs) {
295            (
296                EventHint::InsertText {
297                    event_len,
298                    unicode_len: len,
299                    ..
300                },
301                EventHint::InsertText {
302                    event_len: r_event_len,
303                    unicode_len: r_len,
304                    ..
305                },
306            ) => {
307                *len += *r_len;
308                *event_len += *r_event_len;
309            }
310            (
311                EventHint::InsertList { len, pos: _ },
312                EventHint::InsertList { len: r_len, pos: _ },
313            ) => *len += *r_len,
314            (EventHint::DeleteList(l), EventHint::DeleteList(r)) => l.merge(r, &()),
315            (
316                EventHint::DeleteText { span, unicode_len },
317                EventHint::DeleteText {
318                    span: r_span,
319                    unicode_len: r_len,
320                },
321            ) => {
322                *unicode_len += *r_len;
323                span.merge(r_span, &());
324            }
325            _ => unreachable!(),
326        }
327    }
328
329    fn merge_left(&mut self, _: &Self) {
330        unreachable!()
331    }
332}
333
334impl Transaction {
335    #[inline]
336    pub fn new(doc: Arc<LoroDocInner>) -> LoroResult<Self> {
337        Self::new_with_origin(doc.clone(), "".into())
338    }
339
340    pub fn new_with_origin(doc: Arc<LoroDocInner>, origin: InternalString) -> LoroResult<Self> {
341        let oplog_lock = doc.oplog.lock();
342        let mut state_lock = doc.state.lock();
343        if state_lock.is_in_txn() {
344            return Err(LoroError::DuplicatedTransactionError);
345        }
346
347        state_lock.start_txn(origin, crate::event::EventTriggerKind::Local);
348        let arena = state_lock.arena.clone();
349        let frontiers = state_lock.frontiers.clone();
350        let peer = state_lock.peer.load(std::sync::atomic::Ordering::Relaxed);
351        let next_counter = oplog_lock.next_id(peer).counter;
352        let next_lamport = oplog_lock.dag.frontiers_to_next_lamport(&frontiers);
353        let latest_timestamp = oplog_lock.get_greatest_timestamp(&frontiers);
354        if let Err(err) =
355            oplog_lock.check_change_greater_than_last_peer_id(peer, next_counter, &frontiers)
356        {
357            state_lock.abort_txn();
358            return Err(err);
359        }
360        drop(state_lock);
361        drop(oplog_lock);
362        Ok(Self {
363            peer,
364            doc: Arc::downgrade(&doc),
365            arena,
366            frontiers,
367            timestamp: None,
368            next_counter,
369            next_lamport,
370            origin: Default::default(),
371            start_counter: next_counter,
372            start_lamport: next_lamport,
373            event_hints: Default::default(),
374            local_ops: RleVec::new(),
375            finished: false,
376            on_commit: None,
377            msg: None,
378            latest_timestamp,
379            is_peer_first_appearance: false,
380        })
381    }
382
383    pub fn set_origin(&mut self, origin: InternalString) {
384        self.origin = origin;
385    }
386
387    pub fn set_timestamp(&mut self, time: Timestamp) {
388        self.timestamp = Some(time);
389    }
390
391    pub fn set_msg(&mut self, msg: Option<Arc<str>>) {
392        self.msg = msg;
393    }
394
395    pub fn local_ops(&self) -> &RleVec<[Op; 1]> {
396        &self.local_ops
397    }
398
399    pub fn peer(&self) -> &PeerID {
400        &self.peer
401    }
402
403    pub fn timestamp(&self) -> &Option<Timestamp> {
404        &self.timestamp
405    }
406
407    pub fn frontiers(&self) -> &Frontiers {
408        &self.frontiers
409    }
410    pub fn msg(&self) -> &Option<Arc<str>> {
411        &self.msg
412    }
413
414    pub fn lamport(&self) -> &Lamport {
415        &self.start_lamport
416    }
417
418    pub(crate) fn set_on_commit(&mut self, f: OnCommitFn) {
419        self.on_commit = Some(f);
420    }
421
422    pub(crate) fn take_on_commit(&mut self) -> Option<OnCommitFn> {
423        self.on_commit.take()
424    }
425
426    pub fn commit(mut self) -> Result<Option<CommitOptions>, LoroError> {
427        self._commit()
428    }
429
430    #[tracing::instrument(level = "debug", skip(self))]
431    fn _commit(&mut self) -> Result<Option<CommitOptions>, LoroError> {
432        if self.finished {
433            return Ok(None);
434        }
435
436        let Some(doc) = self.doc.upgrade() else {
437            return Ok(None);
438        };
439        self.finished = true;
440        if self.local_ops.is_empty() {
441            let mut state = doc.state.lock();
442            state.abort_txn();
443            return Ok(Some(self.take_options()));
444        }
445
446        let ops = std::mem::take(&mut self.local_ops);
447        let deps = take(&mut self.frontiers);
448        let change = Change {
449            lamport: self.start_lamport,
450            ops,
451            deps,
452            id: ID::new(self.peer, self.start_counter),
453            timestamp: self.latest_timestamp.max(
454                self.timestamp
455                    .unwrap_or_else(|| doc.oplog.lock().get_timestamp_for_next_txn()),
456            ),
457            commit_msg: take(&mut self.msg),
458        };
459
460        let change_meta = ChangeMeta::from_change(&change);
461        {
462            // add change to uncommit field of oplog
463            let mut oplog = doc.oplog.lock();
464            oplog.set_uncommitted_change(change);
465        }
466
467        let modifier = ChangeModifier::default();
468        doc.pre_commit_subs.emit(
469            &(),
470            PreCommitCallbackPayload {
471                change_meta,
472                origin: self.origin.to_string(),
473                modifier: modifier.clone(),
474            },
475        );
476
477        let mut oplog = doc.oplog.lock();
478        let mut state = doc.state.lock();
479
480        let Some(mut change) = oplog.uncommitted_change.take() else {
481            state.abort_txn();
482            drop(state);
483            drop(oplog);
484            return Err(LoroError::internal(
485                "missing uncommitted change while committing transaction",
486            ));
487        };
488        modifier.modify_change(&mut change);
489        let diff = if state.is_recording() {
490            Some(change_to_diff(
491                &change,
492                doc.clone(),
493                std::mem::take(&mut self.event_hints),
494            ))
495        } else {
496            None
497        };
498
499        let last_id = change.id_last();
500        if let Err(err) = oplog.import_local_change(change) {
501            state.abort_txn();
502            drop(state);
503            drop(oplog);
504            return Err(err);
505        }
506
507        state.commit_txn(
508            Frontiers::from_id(last_id),
509            diff.map(|arr| InternalDocDiff {
510                by: crate::event::EventTriggerKind::Local,
511                origin: self.origin.clone(),
512                diff: Cow::Owned(
513                    arr.into_iter()
514                        .map(|x| InternalContainerDiff {
515                            idx: x.idx,
516                            bring_back: false,
517                            diff: (x.diff.into()),
518                            diff_mode: crate::diff_calc::DiffMode::Linear,
519                        })
520                        .collect(),
521                ),
522                new_version: Cow::Borrowed(oplog.frontiers()),
523            }),
524        );
525        drop(state);
526        drop(oplog);
527        if let Some(on_commit) = self.on_commit.take() {
528            assert!(!doc.txn.is_locked());
529            on_commit(&doc.state.clone(), &doc.oplog.clone(), self.id_span());
530        }
531        Ok(None)
532    }
533
534    fn take_options(&self) -> CommitOptions {
535        let mut options = CommitOptions::new();
536        if !self.origin.is_empty() {
537            options = options.origin(self.origin.as_str());
538        }
539        if let Some(msg) = self.msg.as_ref() {
540            options = options.commit_msg(msg);
541        }
542        if let Some(timestamp) = self.timestamp {
543            options = options.timestamp(timestamp);
544        }
545        options
546    }
547
548    pub(super) fn apply_local_op(
549        &mut self,
550        container: ContainerIdx,
551        content: RawOpContent,
552        event: EventHint,
553        // check whether context and txn are referring to the same state context
554        doc: &LoroDoc,
555    ) -> LoroResult<()> {
556        // A transaction must be applied to the document it was created for.
557        // `insert_with_txn`/`delete_with_txn` are public API, so this guards
558        // against misuse (passing one doc's txn to another doc's handler), which
559        // would otherwise stamp the target doc's state/oplog with the wrong
560        // peer+counter. Compare doc identity via the `Weak` pointer (no atomic
561        // upgrade) on the hot path; only upgrade to fill in the error on the
562        // rare mismatch.
563        if self.doc.as_ptr() != Arc::as_ptr(&doc.inner) {
564            let found = doc
565                .state
566                .lock()
567                .peer
568                .load(std::sync::atomic::Ordering::Relaxed);
569            let expected = self
570                .doc
571                .upgrade()
572                .map(|d| d.state.lock().peer.load(std::sync::atomic::Ordering::Relaxed))
573                .unwrap_or(found);
574            return Err(LoroError::UnmatchedContext { expected, found });
575        }
576
577        let len = content.content_len();
578        assert!(len > 0);
579        let raw_op = RawOp {
580            id: ID {
581                peer: self.peer,
582                counter: self.next_counter,
583            },
584            lamport: self.next_lamport,
585            container,
586            content,
587        };
588
589        let mut oplog = doc.oplog.lock();
590        let mut state = doc.state.lock();
591        if state.is_deleted(container) {
592            return Err(LoroError::ContainerDeleted {
593                container: Box::new(state.arena.idx_to_id(container).unwrap()),
594            });
595        }
596
597        let op = self.arena.convert_raw_op(&raw_op);
598        state.apply_local_op(&raw_op, &op)?;
599        {
600            if !self.is_peer_first_appearance && !oplog.dag.latest_vv_contains_peer(self.peer) {
601                self.is_peer_first_appearance = true;
602            }
603            // update version info
604            let dep_id = Frontiers::from_id(ID::new(self.peer, self.next_counter - 1));
605            let start_id = ID::new(self.peer, self.next_counter);
606            self.next_counter += len as Counter;
607            oplog.dag.update_version_on_new_local_op(
608                if self.local_ops.is_empty() {
609                    &self.frontiers
610                } else {
611                    &dep_id
612                },
613                start_id,
614                self.next_lamport,
615                len,
616            );
617            // Local ops are always visible; bump the cached count incrementally
618            // instead of recomputing it from the version vectors every op.
619            oplog.inc_visible_op_count(len);
620            self.next_lamport += len as Lamport;
621            // set frontiers to the last op id
622            let last_id = start_id.inc(len as Counter - 1);
623            state.frontiers = Frontiers::from_id(last_id);
624        };
625        drop(state);
626        drop(oplog);
627        debug_assert_eq!(
628            event.rle_len(),
629            op.atom_len(),
630            "event:{:#?} \nop:{:#?}",
631            &event,
632            &op
633        );
634
635        let container_hints = self.event_hints.entry(container).or_default();
636
637        match container_hints.last_mut() {
638            Some(last) if last.can_merge(&event) => {
639                last.merge_right(&event);
640            }
641            _ => {
642                container_hints.push(event);
643            }
644        }
645        self.local_ops.push(op);
646        Ok(())
647    }
648
649    /// id can be a str, ContainerID, or ContainerIdRaw.
650    /// if it's str it will use Root container, which will not be None
651    pub fn get_text<I: IntoContainerId>(&self, id: I) -> TextHandler {
652        let id = id.into_container_id(&self.arena, ContainerType::Text);
653        Handler::new_attached(id, LoroDoc::from_inner(self.doc.upgrade().unwrap()))
654            .into_text()
655            .unwrap()
656    }
657
658    /// id can be a str, ContainerID, or ContainerIdRaw.
659    /// if it's str it will use Root container, which will not be None
660    pub fn get_list<I: IntoContainerId>(&self, id: I) -> ListHandler {
661        let id = id.into_container_id(&self.arena, ContainerType::List);
662        Handler::new_attached(id, LoroDoc::from_inner(self.doc.upgrade().unwrap()))
663            .into_list()
664            .unwrap()
665    }
666
667    /// id can be a str, ContainerID, or ContainerIdRaw.
668    /// if it's str it will use Root container, which will not be None
669    pub fn get_map<I: IntoContainerId>(&self, id: I) -> MapHandler {
670        let id = id.into_container_id(&self.arena, ContainerType::Map);
671        Handler::new_attached(id, LoroDoc::from_inner(self.doc.upgrade().unwrap()))
672            .into_map()
673            .unwrap()
674    }
675
676    /// id can be a str, ContainerID, or ContainerIdRaw.
677    /// if it's str it will use Root container, which will not be None
678    pub fn get_tree<I: IntoContainerId>(&self, id: I) -> TreeHandler {
679        let id = id.into_container_id(&self.arena, ContainerType::Tree);
680        Handler::new_attached(id, LoroDoc::from_inner(self.doc.upgrade().unwrap()))
681            .into_tree()
682            .unwrap()
683    }
684    pub fn next_id(&self) -> ID {
685        ID {
686            peer: self.peer,
687            counter: self.next_counter,
688        }
689    }
690
691    #[inline]
692    pub fn id_span(&self) -> IdSpan {
693        IdSpan::new(self.peer, self.start_counter, self.next_counter)
694    }
695
696    pub fn next_idlp(&self) -> IdLp {
697        IdLp {
698            peer: self.peer,
699            lamport: self.next_lamport,
700        }
701    }
702
703    pub fn is_empty(&self) -> bool {
704        self.local_ops.is_empty()
705    }
706
707    pub(crate) fn len(&self) -> usize {
708        (self.next_counter - self.start_counter) as usize
709    }
710
711    pub(crate) fn set_options(&mut self, options: CommitOptions) {
712        self.origin = options.origin.unwrap_or_default();
713        self.msg = options.commit_msg;
714        self.timestamp = options.timestamp;
715    }
716
717    pub(crate) fn set_default_options(&mut self, default_options: crate::loro::CommitOptions) {
718        if self.origin.is_empty() {
719            self.origin = default_options.origin.unwrap_or_default();
720        }
721        if self.msg.is_none() {
722            self.msg = default_options.commit_msg;
723        }
724        if self.timestamp.is_none() {
725            self.timestamp = default_options.timestamp;
726        }
727    }
728}
729
730impl Drop for Transaction {
731    #[tracing::instrument(level = "debug", skip(self))]
732    fn drop(&mut self) {
733        if !self.finished {
734            // TODO: should we abort here or commit here?
735            // what if commit fails?
736            self._commit().unwrap();
737        }
738    }
739}
740
741#[derive(Debug, Clone)]
742pub(crate) struct TxnContainerDiff {
743    pub(crate) idx: ContainerIdx,
744    pub(crate) diff: Diff,
745}
746
747// PERF: could be compacter
748fn change_to_diff(
749    change: &Change,
750    doc: Arc<LoroDocInner>,
751    event_hints: FxHashMap<ContainerIdx, Vec<EventHint>>,
752) -> Vec<TxnContainerDiff> {
753    let mut ans: Vec<TxnContainerDiff> = Vec::with_capacity(change.ops.len());
754    let peer = change.id.peer;
755    let mut lamport = change.lamport;
756
757    // Group ops by container first to match our new structure
758    let mut ops_by_container: FxHashMap<ContainerIdx, Vec<Op>> = FxHashMap::default();
759    for op in change.ops.iter() {
760        ops_by_container
761            .entry(op.container)
762            .or_default()
763            .push(op.clone());
764    }
765
766    // Process each container's hints and ops together
767    for (container_idx, hints) in event_hints {
768        let Some(container_ops) = ops_by_container.get_mut(&container_idx) else {
769            continue;
770        };
771
772        let mut op_index = 0;
773        let mut hint_iter = hints.into_iter();
774        let mut current_hint = hint_iter.next();
775
776        while op_index < container_ops.len() {
777            let Some(hint) = current_hint.take() else {
778                unreachable!("Missing hint for op");
779            };
780
781            // Collect ops that belong to this hint
782            let mut ops_for_hint: SmallVec<[Op; 1]> = smallvec![container_ops[op_index].clone()];
783            let mut total_len = container_ops[op_index].atom_len();
784
785            // If hint spans multiple ops, collect them
786            while total_len < hint.rle_len() {
787                op_index += 1;
788                let next_op_len = container_ops[op_index].atom_len();
789                let op = if next_op_len + total_len > hint.rle_len() {
790                    let new_len = hint.rle_len() - total_len;
791                    let left = container_ops[op_index].slice(0, new_len);
792                    let right = container_ops[op_index].slice(new_len, next_op_len);
793                    container_ops[op_index] = right;
794                    op_index -= 1;
795                    left
796                } else {
797                    container_ops[op_index].clone()
798                };
799
800                total_len += op.atom_len();
801                ops_for_hint.push(op);
802            }
803
804            op_index += 1;
805            assert_eq!(total_len, hint.rle_len(), "Op/hint length mismatch");
806
807            // Move to next hint
808            current_hint = hint_iter.next();
809
810            // Generate diff based on hint type
811            match hint {
812                EventHint::Mark { start, end, style } => {
813                    let mut meta = StyleMeta::default();
814                    meta.insert(
815                        style.key.clone(),
816                        StyleMetaItem {
817                            lamport,
818                            peer: change.id.peer,
819                            value: style.data,
820                        },
821                    );
822                    let diff = DeltaRopeBuilder::new()
823                        .retain(start as usize, Default::default())
824                        .retain(
825                            (end - start) as usize,
826                            meta.to_option_map().unwrap_or_default().into(),
827                        )
828                        .build();
829                    ans.push(TxnContainerDiff {
830                        idx: container_idx,
831                        diff: Diff::Text(diff),
832                    });
833                }
834                EventHint::InsertText { styles, pos, .. } => {
835                    let mut delta: TextDiff = DeltaRopeBuilder::new()
836                        .retain(pos as usize, Default::default())
837                        .build();
838                    for op in ops_for_hint.iter() {
839                        let InnerListOp::InsertText { slice, .. } = op.content.as_list().unwrap()
840                        else {
841                            unreachable!()
842                        };
843
844                        delta.push_insert(
845                            slice.clone().into(),
846                            styles.to_option_map().unwrap_or_default().into(),
847                        );
848                    }
849                    ans.push(TxnContainerDiff {
850                        idx: container_idx,
851                        diff: Diff::Text(delta),
852                    })
853                }
854                EventHint::DeleteText {
855                    span,
856                    unicode_len: _,
857                    // we don't need to iter over ops here, because we already
858                    // know what the events should be
859                } => ans.push(TxnContainerDiff {
860                    idx: container_idx,
861                    diff: Diff::Text(
862                        DeltaRopeBuilder::new()
863                            .retain(span.start() as usize, Default::default())
864                            .delete(span.len())
865                            .build(),
866                    ),
867                }),
868                EventHint::InsertList { pos, .. } => {
869                    // We should use pos from event hint because index in op may
870                    // be using op index for the MovableList
871                    let mut index = pos;
872                    for op in ops_for_hint.iter() {
873                        let (range, _) = op.content.as_list().unwrap().as_insert().unwrap();
874                        let values = doc
875                            .arena
876                            .get_values(range.to_range())
877                            .into_iter()
878                            .map(|v| ValueOrHandler::from_value(v, &doc));
879                        let len = values.len();
880                        ans.push(TxnContainerDiff {
881                            idx: container_idx,
882                            diff: Diff::List(
883                                DeltaRopeBuilder::new()
884                                    .retain(index, Default::default())
885                                    .insert_many(values, Default::default())
886                                    .build(),
887                            ),
888                        });
889                        index += len;
890                    }
891                }
892                EventHint::DeleteList(s) => {
893                    ans.push(TxnContainerDiff {
894                        idx: container_idx,
895                        diff: Diff::List(
896                            DeltaRopeBuilder::new()
897                                .retain(s.start() as usize, Default::default())
898                                .delete(s.len())
899                                .build(),
900                        ),
901                    });
902                }
903                EventHint::Map { key, value } => {
904                    // Mirror `MapState::apply_diff_and_convert`: a marker write surfaces to
905                    // subscribers as the deterministic Container, not the raw binary value.
906                    let parent_id = doc.arena.idx_to_id(container_idx);
907                    let value = value.map(|v| match parent_id.as_ref() {
908                        Some(pid) => {
909                            loro_common::translate_mergeable_marker_value(pid, key.as_ref(), v)
910                        }
911                        None => v,
912                    });
913                    ans.push(TxnContainerDiff {
914                        idx: container_idx,
915                        diff: Diff::Map(ResolvedMapDelta::new().with_entry(
916                            key,
917                            ResolvedMapValue {
918                                value: value.map(|v| ValueOrHandler::from_value(v, &doc)),
919                                idlp: IdLp::new(peer, lamport),
920                            },
921                        )),
922                    });
923                }
924                EventHint::Tree(tree_diff) => {
925                    let mut diff = TreeDiff::default();
926                    diff.diff.extend(tree_diff);
927                    ans.push(TxnContainerDiff {
928                        idx: container_idx,
929                        diff: Diff::Tree(diff),
930                    });
931                }
932                EventHint::Move { from, to, value } => {
933                    let mut a = DeltaRopeBuilder::new()
934                        .retain(from as usize, Default::default())
935                        .delete(1)
936                        .build();
937                    a.compose(
938                        &DeltaRopeBuilder::new()
939                            .retain(to as usize, Default::default())
940                            .insert(
941                                ArrayVec::from([ValueOrHandler::from_value(value, &doc)]),
942                                ListDeltaMeta { from_move: true },
943                            )
944                            .build(),
945                    );
946                    ans.push(TxnContainerDiff {
947                        idx: container_idx,
948                        diff: Diff::List(a),
949                    });
950                }
951                EventHint::SetList { index, value } => {
952                    ans.push(TxnContainerDiff {
953                        idx: container_idx,
954                        diff: Diff::List(
955                            DeltaRopeBuilder::new()
956                                .retain(index, Default::default())
957                                .delete(1)
958                                .insert(
959                                    ArrayVec::from([ValueOrHandler::from_value(value, &doc)]),
960                                    Default::default(),
961                                )
962                                .build(),
963                        ),
964                    });
965                }
966                EventHint::MarkEnd => {
967                    // do nothing
968                }
969                #[cfg(feature = "counter")]
970                EventHint::Counter(diff) => {
971                    ans.push(TxnContainerDiff {
972                        idx: container_idx,
973                        diff: Diff::Counter(diff),
974                    });
975                }
976            }
977
978            // Update lamport for this hint's operations
979            lamport += ops_for_hint
980                .iter()
981                .map(|x| x.content_len() as Lamport)
982                .sum::<Lamport>();
983        }
984    }
985
986    ans
987}
988
989#[cfg(test)]
990mod tests {
991    use super::*;
992    use crate::{cursor::PosType, version::Frontiers};
993
994    #[test]
995    fn txn_creation_rolls_back_in_txn_after_peer_conflict() {
996        let doc = LoroDoc::new();
997        doc.set_detached_editing(true);
998        doc.set_peer_id(7).unwrap();
999
1000        let text = doc.get_text("text");
1001        let mut txn = doc.txn().unwrap();
1002        text.insert_with_txn(&mut txn, 0, "a", PosType::Unicode)
1003            .unwrap();
1004        txn.commit().unwrap();
1005
1006        doc.checkout(&Frontiers::default()).unwrap();
1007        doc.set_peer_id(7).unwrap();
1008
1009        let err = doc
1010            .txn()
1011            .expect_err("stale detached frontiers should reject reusing the same peer");
1012        assert!(matches!(
1013            err,
1014            LoroError::ConcurrentOpsWithSamePeerID { peer: 7, .. }
1015        ));
1016        assert!(!doc.app_state().lock().is_in_txn());
1017    }
1018}
loro_internal/txn.rs

loro_internal/
txn.rs
