libanu/
apply.rs

1// org id g5piHODDo9cz2FFDRisMQw4h8fj1SqTfQ3I/i2p+EGc=
2//! Apply a change.
3use crate::change::{Atom, Change3, NewEdge, NewVertex};
4use crate::changestore::ChangeStore;
5use crate::missing_context::*;
6use crate::pristine::*;
7use crate::record::InodeUpdate;
8use crate::Error;
9use std::collections::{HashMap, HashSet};
10// org id Fn0IzU5GRCA4xymOneQzYPhm7xvcbR8viQt+xz5AzSU=
11/// Apply a change to a channel. This function does not update the
12/// inodes/tree tables, i.e. the correspondence between the pristine
13/// and the working copy. Therefore, this function must be used only
14/// on remote changes, or on "bare" repositories.
15pub fn apply_change_ws<T: MutTxnT, P: ChangeStore>(
16    changes: &P,
17    txn: &mut T,
18    channel: &mut ChannelRef<T>,
19    hash: Hash,
20    workspace: &mut Workspace,
21) -> Result<(u64, Merkle), anyhow::Error> {
22    debug!("apply_change {:?}", hash.to_base32());
23    workspace.clear();
24    let mut channel = channel.r.borrow_mut();
25    let change = changes.get_change(&hash)?;
26
27    for &hash in change.dependencies.iter() {
28        if let Hash::None = hash {
29            continue;
30        }
31        if let Some(int) = txn.get_internal(hash) {
32            if txn.get_changeset(&channel.changes, int, None).is_some() {
33                continue;
34            }
35        }
36        return Err((Error::DependencyMissing { hash }).into());
37    }
38
39    let internal = if let Some(p) = txn.get_internal(hash) {
40        p
41    } else {
42        let internal: ChangeId = txn.make_changeid(&hash);
43        txn.register_change(internal, hash, &change)?;
44        internal
45    };
46    debug!("internal = {:?}", internal);
47    assert!(workspace.is_empty());
48    apply_change_to_channel(txn, &mut channel, internal, &hash, &change, workspace)
49}
50
51pub fn apply_change_rec_ws<T: MutTxnT, P: ChangeStore>(
52    changes: &P,
53    txn: &mut T,
54    channel: &mut ChannelRef<T>,
55    hash: Hash,
56    workspace: &mut Workspace,
57    deps_only: bool,
58) -> Result<(), anyhow::Error> {
59    debug!("apply_change {:?}", hash.to_base32());
60    workspace.clear();
61    let mut channel = channel.r.borrow_mut();
62
63    let mut dep_stack = vec![(hash, true, !deps_only)];
64    while let Some((hash, first, actually_apply)) = dep_stack.pop() {
65        let change = changes.get_change(&hash)?;
66        if first {
67            dep_stack.push((hash, false, actually_apply));
68            for &hash in change.dependencies.iter() {
69                if let Hash::None = hash {
70                    continue;
71                }
72                dep_stack.push((hash, true, true))
73            }
74        } else if actually_apply {
75            let applied = if let Some(int) = txn.get_internal(hash) {
76                txn.get_changeset(&channel.changes, int, None).is_some()
77            } else {
78                false
79            };
80            if !applied {
81                let internal = if let Some(p) = txn.get_internal(hash) {
82                    p
83                } else {
84                    let internal: ChangeId = txn.make_changeid(&hash);
85                    txn.register_change(internal, hash, &change)?;
86                    internal
87                };
88                debug!("internal = {:?}", internal);
89                assert!(workspace.is_empty());
90                apply_change_to_channel(txn, &mut channel, internal, &hash, &change, workspace)?;
91            }
92        }
93    }
94    Ok(())
95}
96
97/// Same as [apply_change_ws], but allocates its own workspace.
98pub fn apply_change<T: MutTxnT, P: ChangeStore>(
99    changes: &P,
100    txn: &mut T,
101    channel: &mut ChannelRef<T>,
102    hash: Hash,
103) -> Result<(u64, Merkle), anyhow::Error> {
104    apply_change_ws(changes, txn, channel, hash, &mut Workspace::new())
105}
106
107/// Same as [apply_change_ws], but allocates its own workspace.
108pub fn apply_change_rec<T: MutTxnT, P: ChangeStore>(
109    changes: &P,
110    txn: &mut T,
111    channel: &mut ChannelRef<T>,
112    hash: Hash,
113    deps_only: bool,
114) -> Result<(), anyhow::Error> {
115    apply_change_rec_ws(
116        changes,
117        txn,
118        channel,
119        hash,
120        &mut Workspace::new(),
121        deps_only,
122    )
123}
124
125// org id T9nfH63JFCwYHwPZjZB+12xQEo2fUyo58K1R1FFRZaE=
126fn apply_change_to_channel<T: MutTxnT>(
127    txn: &mut T,
128    channel: &mut Channel<T>,
129    change_id: ChangeId,
130    hash: &Hash,
131    change: &Change3,
132    ws: &mut Workspace,
133) -> Result<(u64, Merkle), anyhow::Error> {
134    let n = channel.apply_counter;
135    let merkle =
136        if let Some(m) = txn.put_changes(channel, change_id, channel.apply_counter, hash)? {
137            m
138        } else {
139            return Err((Error::ChangeAlreadyOnChannel { hash: *hash }).into());
140        };
141    debug!("apply change to channel");
142    let now = std::time::Instant::now();
143    for change_ in change.changes.iter() {
144        debug!("Applying {:?} (1)", change_);
145        for change_ in change_.iter() {
146            match *change_ {
147                Atom::NewVertex(ref n) => put_newvertex(txn, channel, change, ws, change_id, n)?,
148                Atom::EdgeMap(ref n) if !n.flag.contains(EdgeFlags::DELETED) => {
149                    for edge in n.edges.iter() {
150                        put_newedge(
151                            txn,
152                            channel,
153                            ws,
154                            change_id,
155                            n.inode,
156                            n.previous,
157                            n.flag,
158                            edge,
159                            |_, _, _, _| Ok(true),
160                        )?
161                    }
162                }
163                _ => {}
164            }
165        }
166    }
167    for change_ in change.changes.iter() {
168        debug!("Applying {:?} (2)", change_);
169        for change_ in change_.iter() {
170            match *change_ {
171                Atom::EdgeMap(ref n) if n.flag.contains(EdgeFlags::DELETED) => {
172                    for edge in n.edges.iter() {
173                        put_newedge(
174                            txn,
175                            channel,
176                            ws,
177                            change_id,
178                            n.inode,
179                            n.previous,
180                            n.flag,
181                            edge,
182                            |_, _, _, _| Ok(true),
183                        )?;
184                        // org id aNVimxa4oSqbTYJ033w2QbZTCVMG/+tO2ymc1QSWPH0=
185                        crate::missing_context::collect_zombie_context(
186                            txn,
187                            channel,
188                            &mut ws.missing_context,
189                            n.inode,
190                            edge,
191                            n.flag,
192                            change_id,
193                            |h| change.knows(&h),
194                        )?
195                        // org id hXW9eN25ZM8EQ5B2Ew8NMra+xi7hFv6CN8no+qg8GpY=
196                    }
197                }
198                _ => {}
199            }
200        }
201    }
202    crate::TIMERS.lock().unwrap().apply += now.elapsed();
203
204    clean_obsolete_pseudo_edges(txn, channel, ws)?;
205
206    info!("repairing missing contexts");
207    repair_missing_contexts(txn, channel, ws, change_id, change)?;
208    repair_cyclic_paths(txn, channel, ws, change_id, change)?;
209    info!("done applying change");
210    channel.last_modified = std::time::SystemTime::now()
211        .duration_since(std::time::SystemTime::UNIX_EPOCH)?
212        .as_secs();
213    Ok((n, merkle))
214}
215// org id 7IPbKRayK1ey50vdV6FP2KOmzRGn2tCT3qapFICvlNc=
216/// Apply a change created locally: serialize it, compute its hash, and
217/// apply it. This function also registers changes in the filesystem
218/// introduced by the change (file additions, deletions and moves), to
219/// synchronise the pristine and the working copy after the
220/// application.
221pub fn apply_local_change_ws<T: MutTxnT>(
222    txn: &mut T,
223    channel: &mut ChannelRef<T>,
224    change: &Change3,
225    hash: Hash,
226    inode_updates: &HashMap<usize, InodeUpdate>,
227    workspace: &mut Workspace,
228) -> Result<(u64, Merkle), anyhow::Error> {
229    // org id GEylov+bdcaQUmngYlibUCe0K6dMYx9lmeEmNPGCFUQ=
230    let mut channel = channel.r.borrow_mut();
231    let internal: ChangeId = txn.make_changeid(&hash);
232    txn.register_change(internal, hash, &change)?;
233    // org id O9lNVqhAbRqj2tE6JW0Ax8U3WSetv22rYrSXHQlT78A=
234    let n = apply_change_to_channel(txn, &mut channel, internal, &hash, &change, workspace)?;
235    // org id w0Bu10TO1Kbn1WbRcFHyU/0xBR4Yi61jj66D/1uiicM=
236    for (_, update) in inode_updates.iter() {
237        info!("updating {:?}", update);
238        update_inode(txn, &channel, internal, update)?;
239    }
240    Ok(n)
241}
242
243/// Same as [apply_local_change_ws], but allocates its own workspace.
244pub fn apply_local_change<T: MutTxnT>(
245    txn: &mut T,
246    channel: &mut ChannelRef<T>,
247    change: &Change3,
248    hash: Hash,
249    inode_updates: &HashMap<usize, InodeUpdate>,
250) -> Result<(u64, Merkle), anyhow::Error> {
251    apply_local_change_ws(
252        txn,
253        channel,
254        change,
255        hash,
256        inode_updates,
257        &mut Workspace::new(),
258    )
259}
260// org id zzKI7nUCrGP6P9+0nk29zZnRY1qcdFJH2tXwOxTn/xM=
261fn update_inode<T: MutTxnT>(
262    txn: &mut T,
263    channel: &Channel<T>,
264    internal: ChangeId,
265    update: &InodeUpdate,
266) -> Result<(), anyhow::Error> {
267    debug!("update_inode {:?}", update);
268    match *update {
269        // org id 8cpuyZX4ja1VYT3lhtYBMXJenBtvUbVEGky3wZ6W/HM=
270        InodeUpdate::Add { inode, pos, .. } => {
271            let vertex = Position {
272                change: internal,
273                pos,
274            };
275            if txn
276                .get_graph(&channel.graph, vertex.inode_vertex(), None)
277                .is_some()
278            {
279                debug!("Adding inodes: {:?} {:?}", inode, vertex);
280                txn.put_inodes_with_rev(inode, vertex)?;
281            } else {
282                debug!("Not adding inodes: {:?} {:?}", inode, vertex);
283            }
284        }
285        // org id iTzVDtyzdcIEH3EDPbaq8PfGH9WgwLTd5XSxI5nK4Kc=
286        InodeUpdate::Deleted { inode } => {
287            if let Some(parent) = txn.get_revtree(inode, None).map(|x| x.to_owned()) {
288                txn.del_revtree(inode, Some(parent.as_file_id()))?;
289                txn.del_tree(parent.as_file_id(), Some(inode))?;
290            }
291            txn.del_tree(
292                (OwnedPathId {
293                    parent_inode: inode,
294                    basename: crate::small_string::SmallString::new(),
295                })
296                .as_file_id(),
297                Some(inode),
298            )?;
299            if let Some(vertex) = txn.get_inodes(inode, None) {
300                txn.del_inodes(inode, Some(vertex))?;
301                txn.del_revinodes(vertex, Some(inode))?;
302            }
303        }
304    }
305    Ok(())
306}
307// org id QIECdrRSNOnrMRRD9QO87SUCSg4dt9OUS5V6GZPPmeM=
308fn put_newvertex<T: MutTxnT>(
309    txn: &mut T,
310    channel: &mut Channel<T>,
311    ch: &Change3,
312    ws: &mut Workspace,
313    change: ChangeId,
314    n: &NewVertex<Option<Hash>>,
315) -> Result<(), anyhow::Error> {
316    let vertex = Vertex {
317        change,
318        start: n.start,
319        end: n.end,
320    };
321    debug!(
322        "put_newvertex {:?} {:?} {:?} {:?} {:?}",
323        vertex, n.up_context, n.down_context, n.flag, change
324    );
325    assert!(ws.deleted_by.is_empty());
326    for up in n.up_context.iter() {
327        let up = txn.internal_pos(up, change)?;
328        put_up_context(txn, channel, ch, n.inode, ws, up)?;
329    }
330    for down in n.down_context.iter() {
331        let down = txn.internal_pos(down, change)?;
332        put_down_context(txn, channel, ch, n.inode, ws, down)?;
333    }
334    debug!("deleted by: {:?}", ws.deleted_by);
335
336    for up in ws.up_context.drain(..) {
337        assert_ne!(up, vertex);
338        if !n.flag.contains(EdgeFlags::FOLDER) {
339            for (change, _) in ws.deleted_by.iter() {
340                let flag = n.flag | EdgeFlags::DELETED | EdgeFlags::BLOCK;
341                txn.put_graph_with_rev(channel, flag, up, vertex, *change)?;
342            }
343        }
344        txn.put_graph_with_rev(channel, n.flag | EdgeFlags::BLOCK, up, vertex, change)?;
345    }
346    debug!("down_context {:?}", ws.down_context);
347    for (down, alive_parent) in ws.down_context.drain(..) {
348        assert_ne!(down, vertex);
349        if !alive_parent && !n.flag.contains(EdgeFlags::FOLDER) {
350            for (change, _) in ws.deleted_by.iter() {
351                let flag = n.flag | EdgeFlags::DELETED | EdgeFlags::BLOCK;
352                txn.put_graph_with_rev(channel, flag, vertex, down, *change)?;
353            }
354        } else {
355            txn.put_graph_with_rev(channel, n.flag, vertex, down, change)?;
356        }
357    }
358    ws.deleted_by.clear();
359    Ok(())
360}
361// org id qJsZsuGOM0pVWr+gR/Q1oondyvfulu/VvSZEXxvzUww=
362fn put_up_context<T: MutTxnT>(
363    txn: &mut T,
364    channel: &mut Channel<T>,
365    ch: &Change3,
366    inode: Position<Option<Hash>>,
367    ws: &mut Workspace,
368    up: Position<ChangeId>,
369) -> Result<(), anyhow::Error> {
370    let up_vertex = if up.change.is_root() {
371        Vertex::ROOT
372    } else {
373        debug!("put_up_context {:?}", up);
374        let k = txn.find_block_end(channel, up)?;
375        assert_eq!(k.change, up.change);
376        assert!(k.start <= up.pos);
377        debug!("k = {:?}", k);
378        if k.start < up.pos && k.end > up.pos {
379            if let Some((_, vids)) = ws.missing_context.graphs.get_mut(&inode) {
380                if let Some(vid) = vids.remove(&k) {
381                    vids.insert(Vertex { end: up.pos, ..k }, vid);
382                    vids.insert(Vertex { start: up.pos, ..k }, vid);
383                }
384            }
385            txn.split_block(channel, k, up.pos)?
386        }
387        Vertex {
388            change: k.change,
389            start: k.start,
390            end: up.pos,
391        }
392    };
393    debug!("up_vertex {:?}", up_vertex);
394    let flag0 = EdgeFlags::PARENT | EdgeFlags::DELETED;
395    let flag1 = flag0 | EdgeFlags::FOLDER | EdgeFlags::BLOCK;
396    for parent in txn.iter_adjacent(&channel, up_vertex, flag0, flag1) {
397        // This unwrap is ok: `parent` is in the channel.
398        let introduced_by = txn.get_external(parent.introduced_by).unwrap();
399        if !ch.knows(&introduced_by) {
400            ws.deleted_by
401                .insert((parent.introduced_by, parent.flag - EdgeFlags::PARENT));
402        }
403    }
404    ws.up_context.push(up_vertex);
405    Ok(())
406}
407// org id wmLgN7MtIkCgY6DOEzzeaGOQtP0X7+eXRCRk23sR/F8=
408fn put_down_context<T: MutTxnT>(
409    txn: &mut T,
410    channel: &mut Channel<T>,
411    ch: &Change3,
412    inode: Position<Option<Hash>>,
413    ws: &mut Workspace,
414    down: Position<ChangeId>,
415) -> Result<Vertex<ChangeId>, anyhow::Error> {
416    let k = txn.find_block(&channel, down)?;
417    assert_eq!(k.change, down.change);
418    assert!(k.end >= down.pos);
419    if k.start < down.pos && k.end > down.pos {
420        if let Some((_, vids)) = ws.missing_context.graphs.get_mut(&inode) {
421            if let Some(vid) = vids.remove(&k) {
422                vids.insert(Vertex { end: down.pos, ..k }, vid);
423                vids.insert(
424                    Vertex {
425                        start: down.pos,
426                        ..k
427                    },
428                    vid,
429                );
430            }
431        }
432        txn.split_block(channel, k, down.pos)?
433    }
434    let down_vertex = Vertex {
435        change: k.change,
436        start: down.pos,
437        end: k.end,
438    };
439    debug!("down_vertex {:?}", down_vertex);
440    let flag0 = EdgeFlags::PARENT;
441    let flag1 = flag0 | EdgeFlags::FOLDER | EdgeFlags::BLOCK | EdgeFlags::DELETED;
442    let mut alive_parent = false;
443    for parent in txn.iter_adjacent(&channel, down_vertex, flag0, flag1) {
444        if parent.flag.contains(EdgeFlags::PARENT) {
445            if parent.flag.contains(EdgeFlags::DELETED) {
446                // This unwrap is ok: `parent` is in the channel.
447                let introduced_by = txn.get_external(parent.introduced_by).unwrap();
448                if !ch.knows(&introduced_by) {
449                    ws.deleted_by
450                        .insert((parent.introduced_by, parent.flag - EdgeFlags::PARENT));
451                }
452            } else {
453                alive_parent = true
454            }
455        }
456    }
457    ws.down_context.push((down_vertex, alive_parent));
458    Ok(down_vertex)
459}
460// org id Z9HKWXTmLYMKPH1tSnFV+QYAj3myOwZiyHaaocqlvvU=
461pub struct Workspace {
462    parents: HashSet<Vertex<ChangeId>>,
463    children: HashSet<Vertex<ChangeId>>,
464    pseudo: Vec<(Vertex<ChangeId>, Edge, Position<Option<Hash>>)>,
465    deleted_by: HashSet<(ChangeId, EdgeFlags)>,
466    up_context: Vec<Vertex<ChangeId>>,
467    down_context: Vec<(Vertex<ChangeId>, bool)>,
468    pub(crate) missing_context: crate::missing_context::Workspace,
469    rooted: HashMap<Vertex<ChangeId>, bool>,
470}
471
472impl Workspace {
473    pub fn new() -> Self {
474        Workspace {
475            children: HashSet::new(),
476            parents: HashSet::new(),
477            pseudo: Vec::new(),
478            deleted_by: HashSet::new(),
479            up_context: Vec::new(),
480            down_context: Vec::new(),
481            missing_context: crate::missing_context::Workspace::new(),
482            rooted: HashMap::new(),
483        }
484    }
485    fn clear(&mut self) {
486        self.children.clear();
487        self.parents.clear();
488        self.pseudo.clear();
489        self.deleted_by.clear();
490        self.up_context.clear();
491        self.down_context.clear();
492        self.missing_context.clear();
493        self.rooted.clear();
494    }
495    fn is_empty(&self) -> bool {
496        self.children.is_empty()
497            && self.parents.is_empty()
498            && self.pseudo.is_empty()
499            && self.deleted_by.is_empty()
500            && self.up_context.is_empty()
501            && self.down_context.is_empty()
502            && self.missing_context.is_empty()
503            && self.rooted.is_empty()
504    }
505}
506// org id wwZDqGk+qPaB9Oabmhr4ziOpXgWClHLLbxpCOAo1zpI=
507pub(crate) fn put_newedge<T, F>(
508    txn: &mut T,
509    channel: &mut Channel<T>,
510    ws: &mut Workspace,
511    change: ChangeId,
512    inode: Position<Option<Hash>>,
513    previous: EdgeFlags,
514    flag: EdgeFlags,
515    n: &NewEdge<Option<Hash>>,
516    apply_check: F,
517) -> Result<(), anyhow::Error>
518where
519    T: MutTxnT,
520    F: Fn(
521        &mut T,
522        &mut Channel<T>,
523        Vertex<ChangeId>,
524        Vertex<ChangeId>,
525    ) -> Result<bool, anyhow::Error>,
526{
527    if flag.contains(EdgeFlags::DELETED) {
528        ws.missing_context.load_graph(txn, channel, inode)?;
529    }
530
531    let flag = flag | EdgeFlags::BLOCK;
532    assert!(ws.children.is_empty());
533    assert!(ws.parents.is_empty());
534    debug!(
535        "put_newedge {:?} -> {:?}, {:?} {:?}",
536        previous, flag, n, change
537    );
538    let n_introduced_by = if let Some(n) = txn.internal(&n.introduced_by, change) {
539        n
540    } else {
541        return Err(crate::Error::InconsistentChange.into());
542    };
543    // org id kQMxNhCjAMxxuSOT4nBo7tPslBEr6iNzM4axiROL7og=
544    let mut source = find_source_vertex(txn, channel, &n.from, change, inode, flag, ws)?;
545    let mut target = find_target_vertex(txn, channel, &n.to, change, inode, flag, ws)?;
546    // org id a+LmZrQCNf3Bxa5QhlJyC2n53bkq9zZxc3HDUvPfCZM=
547    loop {
548        if target.end > n.to.end {
549            assert!(!flag.contains(EdgeFlags::FOLDER));
550            if let Some((_, vids)) = ws.missing_context.graphs.get_mut(&inode) {
551                if let Some(vid) = vids.remove(&target) {
552                    vids.insert(
553                        Vertex {
554                            end: n.to.end,
555                            ..target
556                        },
557                        vid,
558                    );
559                    vids.insert(
560                        Vertex {
561                            start: n.to.end,
562                            ..target
563                        },
564                        vid,
565                    );
566                }
567            }
568            txn.split_block(channel, target, n.to.end)?;
569            target.end = n.to.end
570        }
571        // org id gF9Z95RNa8wb5Hr5AMEANbEn6vKy0OtXkkl9g9Ly1xU=
572        if flag.contains(EdgeFlags::DELETED) {
573            collect_pseudo_edges(txn, channel, ws, inode, target)?;
574            if !flag.contains(EdgeFlags::FOLDER) {
575                reconnect_pseudo_edges(txn, channel, inode, ws, target)?;
576            }
577            ws.children.clear();
578        }
579        // org id oqVmCLqqb2bCD0DujhUVCet02c7KXRtRMXEn5rXc244=
580        debug!("deleting {:?} {:?}", previous, flag);
581        if !txn.del_graph_with_rev(channel, previous, source, target, n_introduced_by)? {
582            let del = txn.del_graph_with_rev(
583                channel,
584                previous | EdgeFlags::BLOCK,
585                source,
586                target,
587                n_introduced_by,
588            )?;
589            debug!("deleted: {:?}", del);
590        }
591        debug!("put_graph {:?} {:?}", source, target);
592        assert_ne!(source, target);
593        if apply_check(txn, channel, source, target)? {
594            txn.put_graph_with_rev(channel, flag, source, target, change)?;
595        }
596        // org id B5Ixh6yXfeQKLnmGIWVRXZJHWk6zDXJWjJu05jcsDgE=
597        if target.end >= n.to.end {
598            assert_eq!(target.end, n.to.end);
599            break;
600        }
601
602        source = target;
603        target = txn.find_block(channel, target.end_pos())?;
604        assert_ne!(source, target);
605    }
606    Ok(())
607}
608// org id 7zbypTSSsas1XfpHlx5O3tLiuSmZZ9+Eq05hBASSbn4=
609fn find_source_vertex<T: MutTxnT>(
610    txn: &mut T,
611    channel: &mut Channel<T>,
612    from: &Position<Option<Hash>>,
613    change: ChangeId,
614    inode: Position<Option<Hash>>,
615    flag: EdgeFlags,
616    ws: &mut Workspace,
617) -> Result<Vertex<ChangeId>, anyhow::Error> {
618    let mut source = txn.find_block_end(&channel, txn.internal_pos(&from, change)?)?;
619    if source.start < from.pos && source.end > from.pos {
620        assert!(!flag.contains(EdgeFlags::FOLDER));
621        if let Some((_, vids)) = ws.missing_context.graphs.get_mut(&inode) {
622            if let Some(vid) = vids.remove(&source) {
623                vids.insert(
624                    Vertex {
625                        end: from.pos,
626                        ..source
627                    },
628                    vid,
629                );
630                vids.insert(
631                    Vertex {
632                        start: from.pos,
633                        ..source
634                    },
635                    vid,
636                );
637            }
638        }
639        txn.split_block(channel, source, from.pos)?;
640        source.end = from.pos;
641    }
642    Ok(source)
643}
644
645fn find_target_vertex<T: MutTxnT>(
646    txn: &mut T,
647    channel: &mut Channel<T>,
648    to: &Vertex<Option<Hash>>,
649    change: ChangeId,
650    inode: Position<Option<Hash>>,
651    flag: EdgeFlags,
652    ws: &mut Workspace,
653) -> Result<Vertex<ChangeId>, anyhow::Error> {
654    let to_pos = txn.internal_pos(&to.start_pos(), change)?;
655    let mut target = txn.find_block(channel, to_pos)?;
656    if target.start < to.start {
657        assert!(!flag.contains(EdgeFlags::FOLDER));
658        if let Some((_, vids)) = ws.missing_context.graphs.get_mut(&inode) {
659            if let Some(vid) = vids.remove(&target) {
660                vids.insert(
661                    Vertex {
662                        end: to.start,
663                        ..target
664                    },
665                    vid,
666                );
667                vids.insert(
668                    Vertex {
669                        start: to.start,
670                        ..target
671                    },
672                    vid,
673                );
674            }
675        }
676        txn.split_block(channel, target, to.start)?;
677        target.start = to.start;
678    }
679    Ok(target)
680}
681// org id szeri9Xjwvhn/QRxGYNQRM76Obj5Mxd6Z6KYP1vDdik=
682fn collect_pseudo_edges<T: MutTxnT>(
683    txn: &mut T,
684    channel: &mut Channel<T>,
685    apply: &mut Workspace,
686    inode: Position<Option<Hash>>,
687    v: Vertex<ChangeId>,
688) -> Result<(), anyhow::Error> {
689    for e in txn.iter_adjacent(
690        &channel,
691        v,
692        EdgeFlags::empty(),
693        EdgeFlags::all() - EdgeFlags::DELETED,
694    ) {
695        debug!("collect_pseudo_edges {:?} {:?}", v, e);
696        if !e.flag.contains(EdgeFlags::FOLDER) {
697            if e.flag.contains(EdgeFlags::PARENT) {
698                let p = txn.find_block_end(channel, e.dest)?;
699                if txn.is_alive(channel, p) {
700                    apply.parents.insert(p);
701                }
702            } else {
703                let p = txn.find_block(channel, e.dest)?;
704                apply.children.insert(p);
705            }
706        }
707        if e.flag.contains(EdgeFlags::PSEUDO) {
708            apply.pseudo.push((v, e, inode));
709        }
710    }
711    Ok(())
712}
713// org id mkvErN7T/SkmZqqWYdWC3oNTXKi28ZS8zsmBdylcngA=
714fn reconnect_pseudo_edges<T: MutTxnT>(
715    txn: &mut T,
716    channel: &mut Channel<T>,
717    inode: Position<Option<Hash>>,
718    ws: &mut Workspace,
719    target: Vertex<ChangeId>,
720) -> Result<(), anyhow::Error> {
721    debug!(
722        "reconnect: parents.len() = {:?} to children.len() = {:?}",
723        ws.parents.len(),
724        ws.children.len()
725    );
726    debug!(
727        "reconnect: parents = {:#?} to children = {:#?}",
728        ws.parents, ws.children
729    );
730    if ws.parents.is_empty() {
731        ws.children.clear();
732        return Ok(());
733    }
734    if ws.children.is_empty() {
735        ws.parents.clear();
736        return Ok(());
737    }
738
739    let (graph, vids) = if let Some(n) = ws.missing_context.graphs.get(&inode) {
740        n
741    } else {
742        return Err(crate::Error::InconsistentChange.into());
743    };
744    crate::alive::remove_redundant_parents(
745        &graph,
746        &vids,
747        &mut ws.parents,
748        &mut ws.missing_context.covered_parents,
749        target,
750    );
751    for &p in ws.parents.iter() {
752        ws.missing_context.covered_parents.insert((p, target));
753    }
754
755    crate::alive::remove_redundant_children(&graph, &vids, &mut ws.children, target);
756    debug!(
757        "reconnect (nonredundant) {:?} to {:?}",
758        ws.parents, ws.children
759    );
760
761    for p in ws.parents.drain() {
762        for c in ws.children.iter() {
763            if p != *c && txn.is_alive(channel, p) && txn.is_alive(channel, *c) {
764                txn.put_graph_with_rev(channel, EdgeFlags::PSEUDO, p, *c, ChangeId::ROOT)?;
765            }
766        }
767    }
768    Ok(())
769}
770// org id 98Q84FaUOk/7navHiQObtyTKOkDVixV8OQF97nuKTPs=
771pub(crate) fn clean_obsolete_pseudo_edges<T: MutTxnT>(
772    txn: &mut T,
773    channel: &mut Channel<T>,
774    ws: &mut Workspace,
775) -> Result<(), anyhow::Error> {
776    debug!("pseudo = {:#?}", ws.pseudo);
777    for (next_vertex, p, inode) in ws.pseudo.drain(..) {
778        // Reorder next_vertex and p.dest to reformulate this as an "a -> b" edge.
779        debug!("{:?} {:?}", next_vertex, p);
780        let (a, b) = if p.flag.contains(EdgeFlags::PARENT) {
781            if let Ok(dest) = txn.find_block_end(channel, p.dest) {
782                (dest, next_vertex)
783            } else {
784                continue;
785            }
786        } else {
787            if let Ok(dest) = txn.find_block(channel, p.dest) {
788                (next_vertex, dest)
789            } else {
790                continue;
791            }
792        };
793
794        let a_is_alive = a.is_root()
795            || txn
796                .iter_adjacent(
797                    channel,
798                    a,
799                    EdgeFlags::PARENT,
800                    EdgeFlags::all() - EdgeFlags::DELETED,
801                )
802                .filter(|e| !e.flag.contains(EdgeFlags::PSEUDO))
803                .next()
804                .is_some();
805
806        let b_is_alive = b.is_root()
807            || txn
808                .iter_adjacent(
809                    channel,
810                    b,
811                    EdgeFlags::PARENT,
812                    EdgeFlags::all() - EdgeFlags::DELETED,
813                )
814                .filter(|e| !e.flag.contains(EdgeFlags::PSEUDO))
815                .next()
816                .is_some();
817
818        debug!("{:?} {:?}", a_is_alive, b_is_alive);
819        if a_is_alive {
820            if !b_is_alive {
821                debug!("deleting {:?} -> {:?}", a, b);
822                if txn.del_graph_with_rev(
823                    channel,
824                    p.flag - EdgeFlags::PARENT,
825                    a,
826                    b,
827                    p.introduced_by,
828                )? {
829                    // repair down context.
830                    crate::missing_context::repair_missing_down_context(
831                        txn,
832                        channel,
833                        &mut ws.missing_context,
834                        inode,
835                        b,
836                        &[a],
837                    )?
838                }
839            }
840        } else if b_is_alive {
841            debug!("deleting {:?} -> {:?}", a, b);
842            if txn.del_graph_with_rev(channel, p.flag - EdgeFlags::PARENT, a, b, p.introduced_by)? {
843                // repair up context.
844                crate::missing_context::repair_missing_up_context(
845                    txn,
846                    channel,
847                    &mut ws.missing_context,
848                    inode,
849                    a,
850                    &[b],
851                    p.flag.contains(EdgeFlags::FOLDER),
852                )?
853            }
854        } else {
855            txn.del_graph_with_rev(channel, p.flag - EdgeFlags::PARENT, a, b, p.introduced_by)?;
856        }
857    }
858    Ok(())
859}
860// org id aio1AtR3Nlw9db3yFD42mkoPf3Bx0O6CI+hHFGjyUO0=
861fn repair_missing_contexts<T: MutTxnT>(
862    txn: &mut T,
863    channel: &mut Channel<T>,
864    ws: &mut Workspace,
865    change_id: ChangeId,
866    change: &Change3,
867) -> Result<(), anyhow::Error> {
868    let now = std::time::Instant::now();
869    crate::missing_context::repair_parents_of_deleted(txn, channel, &mut ws.missing_context)?;
870    for atom in change.changes.iter().flat_map(|r| r.iter()) {
871        match atom {
872            Atom::NewVertex(ref n) => {
873                let vertex = Vertex {
874                    change: change_id,
875                    start: n.start,
876                    end: n.end,
877                };
878                debug!("repairing missing context for {:?}", vertex);
879                for up in n.up_context.iter() {
880                    let up = txn.find_block_end(channel, txn.internal_pos(&up, change_id)?)?;
881
882                    if !txn.is_alive(channel, up) {
883                        debug!("repairing missing up context {:?} {:?}", up, vertex);
884                        repair_missing_up_context(
885                            txn,
886                            channel,
887                            &mut ws.missing_context,
888                            n.inode,
889                            up,
890                            &[vertex],
891                            n.flag.contains(EdgeFlags::FOLDER),
892                        )?
893                    }
894                }
895                if !n.flag.contains(EdgeFlags::FOLDER) {
896                    for down in n.down_context.iter() {
897                        let down = txn.find_block(channel, txn.internal_pos(&down, change_id)?)?;
898                        if txn
899                            .iter_adjacent(
900                                channel,
901                                down,
902                                EdgeFlags::PARENT,
903                                EdgeFlags::all() - EdgeFlags::DELETED,
904                            )
905                            .filter(|e| e.introduced_by != change_id)
906                            .next()
907                            .is_none()
908                        {
909                            debug!("repairing missing down context {:?} {:?}", down, vertex);
910                            repair_missing_down_context(
911                                txn,
912                                channel,
913                                &mut ws.missing_context,
914                                n.inode,
915                                down,
916                                &[vertex],
917                            )?
918                        }
919                    }
920                }
921                debug!("done repairing contexts for {:?}", vertex);
922            }
923            Atom::EdgeMap(ref n) if !n.flag.contains(EdgeFlags::DELETED) => {
924                assert!(!n.flag.contains(EdgeFlags::PARENT));
925                for e in n.edges.iter() {
926                    trace!("repairing context nondeleted {:?}", e);
927                    repair_context_nondeleted(
928                        txn,
929                        channel,
930                        &mut ws.missing_context,
931                        n.inode,
932                        change_id,
933                        |h| change.knows(&h),
934                        e,
935                    )?
936                }
937            }
938            _ => {}
939        }
940    }
941    for atom in change.changes.iter().flat_map(|r| r.iter()) {
942        match atom {
943            Atom::EdgeMap(ref n) if n.flag.contains(EdgeFlags::DELETED) => {
944                for e in n.edges.iter() {
945                    trace!("repairing context deleted {:?}", e);
946                    repair_context_deleted(
947                        txn,
948                        channel,
949                        &mut ws.missing_context,
950                        n.inode,
951                        change_id,
952                        |h| change.knows(&h),
953                        e,
954                    )?
955                }
956            }
957            _ => {}
958        }
959    }
960    crate::missing_context::delete_pseudo_edges(txn, channel, &mut ws.missing_context)?;
961    crate::TIMERS.lock().unwrap().repair_context += now.elapsed();
962    Ok(())
963}
964// org id JQod1ELUh8e4H59MbZI504jS4RLdF1g3gcZQR16F44o=
965fn repair_cyclic_paths<T: MutTxnT>(
966    txn: &mut T,
967    channel: &mut Channel<T>,
968    ws: &mut Workspace,
969    change_id: ChangeId,
970    change: &Change3,
971) -> Result<(), anyhow::Error> {
972    let now = std::time::Instant::now();
973    for atom in change.changes.iter().flat_map(|r| r.iter()) {
974        match atom {
975            Atom::EdgeMap(ref n) if n.flag.contains(EdgeFlags::FOLDER | EdgeFlags::DELETED) => {
976                assert!(!n.flag.contains(EdgeFlags::PARENT));
977                for e in n.edges.iter() {
978                    if e.to.len() > 0 {
979                        repair_edge(txn, channel, e, change_id, ws)?;
980                    }
981                }
982            }
983            _ => {}
984        }
985    }
986    crate::TIMERS.lock().unwrap().check_cyclic_paths += now.elapsed();
987    Ok(())
988}
989
990fn repair_edge<T: MutTxnT>(
991    txn: &mut T,
992    channel: &mut Channel<T>,
993    e: &NewEdge<Option<Hash>>,
994    change_id: ChangeId,
995    ws: &mut Workspace,
996) -> Result<(), anyhow::Error> {
997    // If we're deleting a name and not a whole file.
998    let h = if let Some(h) = e.to.change {
999        h
1000    } else {
1001        return Ok(());
1002    };
1003    let to = Vertex {
1004        change: if let Some(h) = txn.get_internal(h) {
1005            h
1006        } else {
1007            return Err(crate::Error::InconsistentChange.into());
1008        },
1009        start: e.to.start,
1010        end: e.to.end,
1011    };
1012    // Check that the inode descendant of this name is rooted.
1013    let mut unrooted = false;
1014    let f0 = EdgeFlags::FOLDER;
1015    let f1 = EdgeFlags::FOLDER | EdgeFlags::BLOCK | EdgeFlags::PSEUDO;
1016    for ee in txn.iter_adjacent(channel, to, f0, f1) {
1017        if !is_rooted(txn, channel, ee.dest.inode_vertex(), ws)? {
1018            unrooted = true;
1019        }
1020    }
1021    // If not, repair.
1022    if unrooted {
1023        let from = txn.find_block_end(channel, txn.internal_pos(&e.from, change_id)?)?;
1024        debug!("repairing unrooted: {:?} {:?}", from, to);
1025        txn.put_graph_with_rev(
1026            channel,
1027            EdgeFlags::FOLDER | EdgeFlags::PSEUDO,
1028            from,
1029            to,
1030            ChangeId::ROOT,
1031        )?;
1032    }
1033    Ok(())
1034}
1035
1036fn is_rooted<T: TxnT>(
1037    txn: &T,
1038    channel: &Channel<T>,
1039    v: Vertex<ChangeId>,
1040    ws: &mut Workspace,
1041) -> Result<bool, crate::Error> {
1042    // Recycling ws.up_context and ws.parents as a stack and a
1043    // "visited" hashset, respectively.
1044    let ref mut stack = ws.up_context;
1045    stack.clear();
1046    stack.push(v);
1047    let ref mut visited = ws.parents;
1048    visited.clear();
1049
1050    while let Some(to) = stack.pop() {
1051        debug!("is_rooted, pop = {:?}", to);
1052        if to.is_root() {
1053            stack.clear();
1054            for v in visited.drain() {
1055                ws.rooted.insert(v, true);
1056            }
1057            return Ok(true);
1058        }
1059        if !visited.insert(to) {
1060            continue;
1061        }
1062        if let Some(&rooted) = ws.rooted.get(&to) {
1063            if rooted {
1064                for v in visited.drain() {
1065                    ws.rooted.insert(v, true);
1066                }
1067                return Ok(true);
1068            } else {
1069                continue;
1070            }
1071        }
1072        let f = EdgeFlags::PARENT | EdgeFlags::FOLDER;
1073        for parent in txn.iter_adjacent(channel, to, f, f | EdgeFlags::PSEUDO | EdgeFlags::BLOCK) {
1074            debug!("is_rooted, parent = {:?}", parent);
1075            let parent = txn.find_block_end(channel, parent.dest)?;
1076            stack.push(parent)
1077        }
1078    }
1079    for v in visited.drain() {
1080        ws.rooted.insert(v, false);
1081    }
1082    Ok(false)
1083}
libanu/apply.rs

libanu/
apply.rs
