loro_internal/
oplog.rs

1mod change_store;
2pub(crate) mod loro_dag;
3mod pending_changes;
4
5use crate::sync::Mutex;
6use bytes::Bytes;
7use std::borrow::Cow;
8use std::cell::RefCell;
9use std::cmp::Ordering;
10use std::rc::Rc;
11use tracing::trace_span;
12
13use self::change_store::iter::MergedChangeIter;
14use self::pending_changes::PendingChanges;
15use super::arena::SharedArena;
16use crate::change::{get_sys_timestamp, Change, Lamport, Timestamp};
17use crate::configure::Configure;
18use crate::container::list::list_op;
19use crate::dag::{Dag, DagUtils};
20use crate::diff_calc::DiffMode;
21use crate::encoding::decode_oplog;
22use crate::encoding::{ImportStatus, ParsedHeaderAndBody};
23use crate::history_cache::ContainerHistoryCache;
24use crate::id::{Counter, PeerID, ID};
25use crate::op::{FutureInnerContent, ListSlice, RawOpContent, RemoteOp, RichOp};
26use crate::span::{HasCounterSpan, HasLamportSpan};
27use crate::version::{Frontiers, ImVersionVector, VersionVector};
28use crate::LoroError;
29use change_store::BlockOpRef;
30use loro_common::{HasIdSpan, IdLp, IdSpan};
31use rle::{HasLength, RleVec, Sliceable};
32use smallvec::SmallVec;
33
34pub use self::loro_dag::{AppDag, AppDagNode, FrontiersNotIncluded};
35pub use change_store::{BlockChangeRef, ChangeStore};
36
37/// [OpLog] store all the ops i.e. the history.
38/// It allows multiple [AppState] to attach to it.
39/// So you can derive different versions of the state from the [OpLog].
40/// It allows us to build a version control system.
41///
42/// The causal graph should always be a DAG and complete. So we can always find the LCA.
43/// If deps are missing, we can't import the change. It will be put into the `pending_changes`.
44pub struct OpLog {
45    pub(crate) dag: AppDag,
46    pub(crate) arena: SharedArena,
47    change_store: ChangeStore,
48    history_cache: Mutex<ContainerHistoryCache>,
49    /// Pending changes that haven't been applied to the dag.
50    /// A change can be imported only when all its deps are already imported.
51    /// Key is the ID of the missing dep
52    pub(crate) pending_changes: PendingChanges,
53    /// Whether we are importing a batch of changes.
54    /// If so the Dag's frontiers won't be updated until the batch is finished.
55    pub(crate) batch_importing: bool,
56    pub(crate) configure: Configure,
57    /// The uncommitted change, it's a placeholder for the change
58    /// that is being edited in pre-commit callback.
59    pub(crate) uncommitted_change: Option<Change>,
60}
61
62impl std::fmt::Debug for OpLog {
63    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64        f.debug_struct("OpLog")
65            .field("dag", &self.dag)
66            .field("pending_changes", &self.pending_changes)
67            .finish()
68    }
69}
70
71impl OpLog {
72    #[inline]
73    pub(crate) fn new() -> Self {
74        let arena = SharedArena::new();
75        let cfg = Configure::default();
76        let change_store = ChangeStore::new_mem(&arena, cfg.merge_interval_in_s.clone());
77        Self {
78            history_cache: Mutex::new(ContainerHistoryCache::new(change_store.clone(), None)),
79            dag: AppDag::new(change_store.clone()),
80            change_store,
81            arena,
82            pending_changes: Default::default(),
83            batch_importing: false,
84            configure: cfg,
85            uncommitted_change: None,
86        }
87    }
88
89    #[inline]
90    pub fn dag(&self) -> &AppDag {
91        &self.dag
92    }
93
94    pub fn change_store(&self) -> &ChangeStore {
95        &self.change_store
96    }
97
98    /// Get the change with the given peer and lamport.
99    ///
100    /// If not found, return the change with the greatest lamport that is smaller than the given lamport.
101    pub fn get_change_with_lamport_lte(
102        &self,
103        peer: PeerID,
104        lamport: Lamport,
105    ) -> Option<BlockChangeRef> {
106        let ans = self
107            .change_store
108            .get_change_by_lamport_lte(IdLp::new(peer, lamport))?;
109        debug_assert!(ans.lamport <= lamport);
110        Some(ans)
111    }
112
113    pub fn get_timestamp_of_version(&self, f: &Frontiers) -> Timestamp {
114        let mut timestamp = Timestamp::default();
115        for id in f.iter() {
116            if let Some(change) = self.lookup_change(id) {
117                timestamp = timestamp.max(change.timestamp);
118            }
119        }
120
121        timestamp
122    }
123
124    #[inline]
125    pub fn is_empty(&self) -> bool {
126        self.dag.is_empty() && self.arena.can_import_snapshot()
127    }
128
129    /// This is the **only** place to update the `OpLog.changes`
130    pub(crate) fn insert_new_change(&mut self, change: Change, from_local: bool) {
131        let s = trace_span!(
132            "insert_new_change",
133            id = ?change.id,
134            lamport = change.lamport,
135            deps = ?change.deps
136        );
137        let _enter = s.enter();
138        self.dag.handle_new_change(&change, from_local);
139        self.history_cache
140            .lock()
141            .unwrap()
142            .insert_by_new_change(&change, true, true);
143        self.register_container_and_parent_link(&change);
144        self.change_store.insert_change(change, true, from_local);
145    }
146
147    #[inline(always)]
148    pub(crate) fn with_history_cache<F, R>(&self, f: F) -> R
149    where
150        F: FnOnce(&mut ContainerHistoryCache) -> R,
151    {
152        let mut history_cache = self.history_cache.lock().unwrap();
153        f(&mut history_cache)
154    }
155
156    pub fn has_history_cache(&self) -> bool {
157        self.history_cache.lock().unwrap().has_cache()
158    }
159
160    pub fn free_history_cache(&self) {
161        let mut history_cache = self.history_cache.lock().unwrap();
162        history_cache.free();
163    }
164
165    /// Import a change.
166    ///
167    /// Pending changes that haven't been applied to the dag.
168    /// A change can be imported only when all its deps are already imported.
169    /// Key is the ID of the missing dep
170    ///
171    /// # Err
172    ///
173    /// - Return Err(LoroError::UsedOpID) when the change's id is occupied
174    /// - Return Err(LoroError::DecodeError) when the change's deps are missing
175    pub(crate) fn import_local_change(&mut self, change: Change) -> Result<(), LoroError> {
176        self.insert_new_change(change, true);
177        Ok(())
178    }
179
180    /// Trim the known part of change
181    pub(crate) fn trim_the_known_part_of_change(&self, change: Change) -> Option<Change> {
182        let Some(&end) = self.dag.vv().get(&change.id.peer) else {
183            return Some(change);
184        };
185
186        if change.id.counter >= end {
187            return Some(change);
188        }
189
190        if change.ctr_end() <= end {
191            return None;
192        }
193
194        let offset = (end - change.id.counter) as usize;
195        Some(change.slice(offset, change.atom_len()))
196    }
197
198    #[allow(unused)]
199    fn check_id_is_not_duplicated(&self, id: ID) -> Result<(), LoroError> {
200        let cur_end = self.dag.vv().get(&id.peer).cloned().unwrap_or(0);
201        if cur_end > id.counter {
202            return Err(LoroError::UsedOpID { id });
203        }
204
205        Ok(())
206    }
207
208    /// Ensure the new change is greater than the last peer's id and the counter is continuous.
209    ///
210    /// It can be false when users use detached editing mode and use a custom peer id.
211    // This method might be slow and can be optimized if needed in the future.
212    pub(crate) fn check_change_greater_than_last_peer_id(
213        &self,
214        peer: PeerID,
215        counter: Counter,
216        deps: &Frontiers,
217    ) -> Result<(), LoroError> {
218        if counter == 0 {
219            return Ok(());
220        }
221
222        if !self.configure.detached_editing() {
223            return Ok(());
224        }
225
226        let mut max_last_counter = -1;
227        for dep in deps.iter() {
228            let dep_vv = self.dag.get_vv(dep).unwrap();
229            max_last_counter = max_last_counter.max(dep_vv.get(&peer).cloned().unwrap_or(0) - 1);
230        }
231
232        if counter != max_last_counter + 1 {
233            return Err(LoroError::ConcurrentOpsWithSamePeerID {
234                peer,
235                last_counter: max_last_counter,
236                current: counter,
237            });
238        }
239
240        Ok(())
241    }
242
243    pub(crate) fn next_id(&self, peer: PeerID) -> ID {
244        let cnt = self.dag.vv().get(&peer).copied().unwrap_or(0);
245        ID::new(peer, cnt)
246    }
247
248    pub(crate) fn vv(&self) -> &VersionVector {
249        self.dag.vv()
250    }
251
252    pub(crate) fn frontiers(&self) -> &Frontiers {
253        self.dag.frontiers()
254    }
255
256    /// - Ordering::Less means self is less than target or parallel
257    /// - Ordering::Equal means versions equal
258    /// - Ordering::Greater means self's version is greater than target
259    pub fn cmp_with_frontiers(&self, other: &Frontiers) -> Ordering {
260        self.dag.cmp_with_frontiers(other)
261    }
262
263    /// Compare two [Frontiers] causally.
264    ///
265    /// If one of the [Frontiers] are not included, it will return [FrontiersNotIncluded].
266    #[inline]
267    pub fn cmp_frontiers(
268        &self,
269        a: &Frontiers,
270        b: &Frontiers,
271    ) -> Result<Option<Ordering>, FrontiersNotIncluded> {
272        self.dag.cmp_frontiers(a, b)
273    }
274
275    pub(crate) fn get_min_lamport_at(&self, id: ID) -> Lamport {
276        self.get_change_at(id).map(|c| c.lamport).unwrap_or(0)
277    }
278
279    pub(crate) fn get_lamport_at(&self, id: ID) -> Option<Lamport> {
280        self.get_change_at(id)
281            .map(|c| c.lamport + (id.counter - c.id.counter) as Lamport)
282    }
283
284    pub(crate) fn iter_ops(&self, id_span: IdSpan) -> impl Iterator<Item = RichOp<'static>> + '_ {
285        let change_iter = self.change_store.iter_changes(id_span);
286        change_iter.flat_map(move |c| RichOp::new_iter_by_cnt_range(c, id_span.counter))
287    }
288
289    pub(crate) fn get_max_lamport_at(&self, id: ID) -> Lamport {
290        self.get_change_at(id)
291            .map(|c| {
292                let change_counter = c.id.counter as u32;
293                c.lamport + c.ops().last().map(|op| op.counter).unwrap_or(0) as u32 - change_counter
294            })
295            .unwrap_or(Lamport::MAX)
296    }
297
298    pub fn get_change_at(&self, id: ID) -> Option<BlockChangeRef> {
299        self.change_store.get_change(id)
300    }
301
302    pub(crate) fn set_uncommitted_change(&mut self, change: Change) {
303        self.uncommitted_change = Some(change);
304    }
305
306    pub(crate) fn get_uncommitted_change_in_span(
307        &self,
308        id_span: IdSpan,
309    ) -> Option<Cow<'_, Change>> {
310        self.uncommitted_change.as_ref().and_then(|c| {
311            if c.id_span() == id_span {
312                Some(Cow::Borrowed(c))
313            } else if let Some((start, end)) = id_span.get_slice_range_on(&c.id_span()) {
314                Some(Cow::Owned(c.slice(start, end)))
315            } else {
316                None
317            }
318        })
319    }
320
321    pub fn get_deps_of(&self, id: ID) -> Option<Frontiers> {
322        self.get_change_at(id).map(|c| {
323            if c.id.counter == id.counter {
324                c.deps.clone()
325            } else {
326                Frontiers::from_id(id.inc(-1))
327            }
328        })
329    }
330
331    pub fn get_remote_change_at(&self, id: ID) -> Option<Change<RemoteOp<'static>>> {
332        let change = self.get_change_at(id)?;
333        Some(convert_change_to_remote(&self.arena, &change))
334    }
335
336    pub(crate) fn import_unknown_lamport_pending_changes(
337        &mut self,
338        remote_changes: Vec<Change>,
339    ) -> Result<(), LoroError> {
340        self.extend_pending_changes_with_unknown_lamport(remote_changes)
341    }
342
343    /// lookup change by id.
344    ///
345    /// if id does not included in this oplog, return None
346    pub(crate) fn lookup_change(&self, id: ID) -> Option<BlockChangeRef> {
347        self.change_store.get_change(id)
348    }
349
350    #[inline(always)]
351    pub(crate) fn export_change_store_from(&self, vv: &VersionVector, f: &Frontiers) -> Bytes {
352        self.change_store
353            .export_from(vv, f, self.vv(), self.frontiers())
354    }
355
356    #[inline(always)]
357    pub(crate) fn export_change_store_in_range(
358        &self,
359        vv: &VersionVector,
360        f: &Frontiers,
361        to_vv: &VersionVector,
362        to_frontiers: &Frontiers,
363    ) -> Bytes {
364        self.change_store.export_from(vv, f, to_vv, to_frontiers)
365    }
366
367    #[inline(always)]
368    pub(crate) fn export_blocks_from<W: std::io::Write>(&self, vv: &VersionVector, w: &mut W) {
369        self.change_store
370            .export_blocks_from(vv, self.shallow_since_vv(), self.vv(), w)
371    }
372
373    #[inline(always)]
374    pub(crate) fn export_blocks_in_range<W: std::io::Write>(&self, spans: &[IdSpan], w: &mut W) {
375        self.change_store.export_blocks_in_range(spans, w)
376    }
377
378    pub(crate) fn fork_changes_up_to(&self, frontiers: &Frontiers) -> Option<Bytes> {
379        let vv = self.dag.frontiers_to_vv(frontiers)?;
380        Some(
381            self.change_store
382                .fork_changes_up_to(self.dag.shallow_since_vv(), frontiers, &vv),
383        )
384    }
385
386    #[inline(always)]
387    pub(crate) fn decode(&mut self, data: ParsedHeaderAndBody) -> Result<ImportStatus, LoroError> {
388        decode_oplog(self, data)
389    }
390
391    /// iterates over all changes between LCA(common ancestors) to the merged version of (`from` and `to`) causally
392    ///
393    /// Tht iterator will include a version vector when the change is applied
394    ///
395    /// returns: (common_ancestor_vv, iterator)
396    ///
397    /// Note: the change returned by the iterator may include redundant ops at the beginning, you should trim it by yourself.
398    /// You can trim it by the provided counter value. It should start with the counter.
399    ///
400    /// If frontiers are provided, it will be faster (because we don't need to calculate it from version vector
401    #[allow(clippy::type_complexity)]
402    pub(crate) fn iter_from_lca_causally(
403        &self,
404        from: &VersionVector,
405        from_frontiers: &Frontiers,
406        to: &VersionVector,
407        to_frontiers: &Frontiers,
408    ) -> (
409        VersionVector,
410        DiffMode,
411        impl Iterator<
412                Item = (
413                    BlockChangeRef,
414                    (Counter, Counter),
415                    Rc<RefCell<VersionVector>>,
416                ),
417            > + '_,
418    ) {
419        let mut merged_vv = from.clone();
420        merged_vv.merge(to);
421        loro_common::debug!("to_frontiers={:?} vv={:?}", &to_frontiers, to);
422        let (common_ancestors, mut diff_mode) =
423            self.dag.find_common_ancestor(from_frontiers, to_frontiers);
424        if diff_mode == DiffMode::Checkout && to > from {
425            diff_mode = DiffMode::Import;
426        }
427
428        let common_ancestors_vv = self.dag.frontiers_to_vv(&common_ancestors).unwrap();
429        // go from lca to merged_vv
430        let diff = common_ancestors_vv.diff(&merged_vv).forward;
431        let mut iter = self.dag.iter_causal(common_ancestors, diff);
432        let mut node = iter.next();
433        let mut cur_cnt = 0;
434        let vv = Rc::new(RefCell::new(VersionVector::default()));
435        (
436            common_ancestors_vv.clone(),
437            diff_mode,
438            std::iter::from_fn(move || {
439                if let Some(inner) = &node {
440                    let mut inner_vv = vv.borrow_mut();
441                    // FIXME: PERF: it looks slow for large vv, like 10000+ entries
442                    inner_vv.clear();
443                    self.dag.ensure_vv_for(&inner.data);
444                    inner_vv.extend_to_include_vv(inner.data.vv.get().unwrap().iter());
445                    let peer = inner.data.peer;
446                    let cnt = inner
447                        .data
448                        .cnt
449                        .max(cur_cnt)
450                        .max(common_ancestors_vv.get(&peer).copied().unwrap_or(0));
451                    let dag_node_end = (inner.data.cnt + inner.data.len as Counter)
452                        .min(merged_vv.get(&peer).copied().unwrap_or(0));
453                    let change = self.change_store.get_change(ID::new(peer, cnt)).unwrap();
454
455                    if change.ctr_end() < dag_node_end {
456                        cur_cnt = change.ctr_end();
457                    } else {
458                        node = iter.next();
459                        cur_cnt = 0;
460                    }
461
462                    inner_vv.extend_to_include_end_id(change.id);
463
464                    Some((change, (cnt, dag_node_end), vv.clone()))
465                } else {
466                    None
467                }
468            }),
469        )
470    }
471
472    pub fn len_changes(&self) -> usize {
473        self.change_store.change_num()
474    }
475
476    pub fn diagnose_size(&self) -> SizeInfo {
477        let mut total_changes = 0;
478        let mut total_ops = 0;
479        let mut total_atom_ops = 0;
480        let total_dag_node = self.dag.total_parsed_dag_node();
481        self.change_store.visit_all_changes(&mut |change| {
482            total_changes += 1;
483            total_ops += change.ops.len();
484            total_atom_ops += change.atom_len();
485        });
486
487        println!("total changes: {}", total_changes);
488        println!("total ops: {}", total_ops);
489        println!("total atom ops: {}", total_atom_ops);
490        println!("total dag node: {}", total_dag_node);
491        SizeInfo {
492            total_changes,
493            total_ops,
494            total_atom_ops,
495            total_dag_node,
496        }
497    }
498
499    pub(crate) fn iter_changes_peer_by_peer<'a>(
500        &'a self,
501        from: &VersionVector,
502        to: &VersionVector,
503    ) -> impl Iterator<Item = BlockChangeRef> + 'a {
504        let spans: Vec<_> = from.diff_iter(to).1.collect();
505        spans
506            .into_iter()
507            .flat_map(move |span| self.change_store.iter_changes(span))
508    }
509
510    pub(crate) fn iter_changes_causally_rev<'a>(
511        &'a self,
512        from: &VersionVector,
513        to: &VersionVector,
514    ) -> impl Iterator<Item = BlockChangeRef> + 'a {
515        MergedChangeIter::new_change_iter_rev(self, from, to)
516    }
517
518    pub fn get_timestamp_for_next_txn(&self) -> Timestamp {
519        if self.configure.record_timestamp() {
520            get_timestamp_now_txn()
521        } else {
522            0
523        }
524    }
525
526    #[inline(never)]
527    pub(crate) fn idlp_to_id(&self, id: loro_common::IdLp) -> Option<ID> {
528        let change = self.change_store.get_change_by_lamport_lte(id)?;
529
530        if change.lamport > id.lamport || change.lamport_end() <= id.lamport {
531            return None;
532        }
533
534        Some(ID::new(
535            change.id.peer,
536            (id.lamport - change.lamport) as Counter + change.id.counter,
537        ))
538    }
539
540    #[allow(unused)]
541    pub(crate) fn id_to_idlp(&self, id_start: ID) -> IdLp {
542        let change = self.get_change_at(id_start).unwrap();
543        let lamport = change.lamport + (id_start.counter - change.id.counter) as Lamport;
544        let peer = id_start.peer;
545        loro_common::IdLp { peer, lamport }
546    }
547
548    /// NOTE: This may return a op that includes the given id, not necessarily start with the given id
549    pub(crate) fn get_op_that_includes(&self, id: ID) -> Option<BlockOpRef> {
550        let change = self.get_change_at(id)?;
551        change.get_op_with_counter(id.counter)
552    }
553
554    pub(crate) fn split_span_based_on_deps(&self, id_span: IdSpan) -> Vec<(IdSpan, Frontiers)> {
555        let peer = id_span.peer;
556        let mut counter = id_span.counter.min();
557        let span_end = id_span.counter.norm_end();
558        let mut ans = Vec::new();
559
560        while counter < span_end {
561            let id = ID::new(peer, counter);
562            let node = self.dag.get(id).unwrap();
563
564            let f = if node.cnt == counter {
565                node.deps.clone()
566            } else if counter > 0 {
567                id.inc(-1).into()
568            } else {
569                unreachable!()
570            };
571
572            let cur_end = node.cnt + node.len as Counter;
573            let len = cur_end.min(span_end) - counter;
574            ans.push((id.to_span(len as usize), f));
575            counter += len;
576        }
577
578        ans
579    }
580
581    #[inline]
582    pub fn compact_change_store(&mut self) {
583        self.change_store
584            .flush_and_compact(self.dag.vv(), self.dag.frontiers());
585    }
586
587    #[inline]
588    pub fn change_store_kv_size(&self) -> usize {
589        self.change_store.kv_size()
590    }
591
592    pub fn encode_change_store(&self) -> bytes::Bytes {
593        self.change_store
594            .encode_all(self.dag.vv(), self.dag.frontiers())
595    }
596
597    pub fn check_dag_correctness(&self) {
598        self.dag.check_dag_correctness();
599    }
600
601    pub fn shallow_since_vv(&self) -> &ImVersionVector {
602        self.dag.shallow_since_vv()
603    }
604
605    pub fn shallow_since_frontiers(&self) -> &Frontiers {
606        self.dag.shallow_since_frontiers()
607    }
608
609    pub fn is_shallow(&self) -> bool {
610        !self.dag.shallow_since_vv().is_empty()
611    }
612
613    pub fn get_greatest_timestamp(&self, frontiers: &Frontiers) -> Timestamp {
614        let mut max_timestamp = Timestamp::default();
615        for id in frontiers.iter() {
616            let change = self.get_change_at(id).unwrap();
617            if change.timestamp > max_timestamp {
618                max_timestamp = change.timestamp;
619            }
620        }
621
622        max_timestamp
623    }
624}
625
626#[derive(Debug)]
627pub struct SizeInfo {
628    pub total_changes: usize,
629    pub total_ops: usize,
630    pub total_atom_ops: usize,
631    pub total_dag_node: usize,
632}
633
634pub(crate) fn convert_change_to_remote(
635    arena: &SharedArena,
636    change: &Change,
637) -> Change<RemoteOp<'static>> {
638    let mut ops = RleVec::new();
639    for op in change.ops.iter() {
640        for op in local_op_to_remote(arena, op) {
641            ops.push(op);
642        }
643    }
644
645    Change {
646        ops,
647        id: change.id,
648        deps: change.deps.clone(),
649        lamport: change.lamport,
650        timestamp: change.timestamp,
651        commit_msg: change.commit_msg.clone(),
652    }
653}
654
655pub(crate) fn local_op_to_remote(
656    arena: &SharedArena,
657    op: &crate::op::Op,
658) -> SmallVec<[RemoteOp<'static>; 1]> {
659    let container = arena.get_container_id(op.container).unwrap();
660    let mut contents: SmallVec<[_; 1]> = SmallVec::new();
661    match &op.content {
662        crate::op::InnerContent::List(list) => match list {
663            list_op::InnerListOp::Insert { slice, pos } => match container.container_type() {
664                loro_common::ContainerType::Text => {
665                    let str = arena
666                        .slice_str_by_unicode_range(slice.0.start as usize..slice.0.end as usize);
667                    contents.push(RawOpContent::List(list_op::ListOp::Insert {
668                        slice: ListSlice::RawStr {
669                            unicode_len: str.chars().count(),
670                            str: Cow::Owned(str),
671                        },
672                        pos: *pos,
673                    }));
674                }
675                loro_common::ContainerType::List | loro_common::ContainerType::MovableList => {
676                    contents.push(RawOpContent::List(list_op::ListOp::Insert {
677                        slice: ListSlice::RawData(Cow::Owned(
678                            arena.get_values(slice.0.start as usize..slice.0.end as usize),
679                        )),
680                        pos: *pos,
681                    }))
682                }
683                _ => unreachable!(),
684            },
685            list_op::InnerListOp::InsertText {
686                slice,
687                unicode_len: len,
688                unicode_start: _,
689                pos,
690            } => match container.container_type() {
691                loro_common::ContainerType::Text => {
692                    contents.push(RawOpContent::List(list_op::ListOp::Insert {
693                        slice: ListSlice::RawStr {
694                            unicode_len: *len as usize,
695                            str: Cow::Owned(std::str::from_utf8(slice).unwrap().to_owned()),
696                        },
697                        pos: *pos as usize,
698                    }));
699                }
700                _ => unreachable!(),
701            },
702            list_op::InnerListOp::Delete(del) => {
703                contents.push(RawOpContent::List(list_op::ListOp::Delete(*del)))
704            }
705            list_op::InnerListOp::StyleStart {
706                start,
707                end,
708                key,
709                value,
710                info,
711            } => contents.push(RawOpContent::List(list_op::ListOp::StyleStart {
712                start: *start,
713                end: *end,
714                key: key.clone(),
715                value: value.clone(),
716                info: *info,
717            })),
718            list_op::InnerListOp::StyleEnd => {
719                contents.push(RawOpContent::List(list_op::ListOp::StyleEnd))
720            }
721            list_op::InnerListOp::Move {
722                from,
723                elem_id: from_id,
724                to,
725            } => contents.push(RawOpContent::List(list_op::ListOp::Move {
726                from: *from,
727                elem_id: *from_id,
728                to: *to,
729            })),
730            list_op::InnerListOp::Set { elem_id, value } => {
731                contents.push(RawOpContent::List(list_op::ListOp::Set {
732                    elem_id: *elem_id,
733                    value: value.clone(),
734                }))
735            }
736        },
737        crate::op::InnerContent::Map(map) => {
738            let value = map.value.clone();
739            contents.push(RawOpContent::Map(crate::container::map::MapSet {
740                key: map.key.clone(),
741                value,
742            }))
743        }
744        crate::op::InnerContent::Tree(tree) => contents.push(RawOpContent::Tree(tree.clone())),
745        crate::op::InnerContent::Future(f) => match f {
746            #[cfg(feature = "counter")]
747            crate::op::FutureInnerContent::Counter(c) => contents.push(RawOpContent::Counter(*c)),
748            FutureInnerContent::Unknown { prop, value } => {
749                contents.push(crate::op::RawOpContent::Unknown {
750                    prop: *prop,
751                    value: (**value).clone(),
752                })
753            }
754        },
755    };
756
757    let mut ans = SmallVec::with_capacity(contents.len());
758    for content in contents {
759        ans.push(RemoteOp {
760            container: container.clone(),
761            content,
762            counter: op.counter,
763        })
764    }
765    ans
766}
767
768pub(crate) fn get_timestamp_now_txn() -> Timestamp {
769    (get_sys_timestamp() as Timestamp + 500) / 1000
770}
loro_internal/oplog.rs

loro_internal/
oplog.rs
