Skip to main content

pijul_core/pristine/
sanakirja.rs

1use super::*;
2use crate::HashMap;
3use ::sanakirja::*;
4use parking_lot::Mutex;
5use std::collections::hash_map::Entry;
6#[cfg(feature = "mmap")]
7use std::path::Path;
8use std::sync::Arc;
9
10/// A Sanakirja pristine.
11pub struct Pristine {
12    pub env: Arc<::sanakirja::Env>,
13}
14
15pub(crate) type P<K, V> = btree::page::Page<K, V>;
16pub type Db<K, V> = btree::Db<K, V>;
17pub(crate) type UP<K, V> = btree::page_unsized::Page<K, V>;
18pub type UDb<K, V> = btree::Db_<K, V, UP<K, V>>;
19
20#[derive(Debug, Error)]
21pub enum SanakirjaError {
22    #[error(transparent)]
23    Sanakirja(#[from] ::sanakirja::Error),
24    #[error("Pristine locked")]
25    PristineLocked,
26    #[error("Pristine corrupt")]
27    PristineCorrupt,
28    #[error(transparent)]
29    Borrow(#[from] std::cell::BorrowError),
30    #[error("Cannot dropped a borrowed channel: {:?}", c)]
31    ChannelRc { c: String },
32    #[error("Pristine version mismatch. Cloning over the network can fix this.")]
33    Version,
34}
35
36impl std::convert::From<::sanakirja::CRCError> for SanakirjaError {
37    fn from(_: ::sanakirja::CRCError) -> Self {
38        SanakirjaError::PristineCorrupt
39    }
40}
41
42impl std::convert::From<::sanakirja::CRCError> for TxnErr<SanakirjaError> {
43    fn from(_: ::sanakirja::CRCError) -> Self {
44        TxnErr(SanakirjaError::PristineCorrupt)
45    }
46}
47
48impl std::convert::From<::sanakirja::Error> for TxnErr<SanakirjaError> {
49    fn from(e: ::sanakirja::Error) -> Self {
50        TxnErr(e.into())
51    }
52}
53
54impl std::convert::From<::sanakirja::Error> for TreeErr<SanakirjaError> {
55    fn from(e: ::sanakirja::Error) -> Self {
56        TreeErr(e.into())
57    }
58}
59
60impl std::convert::From<TxnErr<::sanakirja::Error>> for TxnErr<SanakirjaError> {
61    fn from(e: TxnErr<::sanakirja::Error>) -> Self {
62        TxnErr(e.0.into())
63    }
64}
65
66impl std::convert::From<TreeErr<::sanakirja::Error>> for TreeErr<SanakirjaError> {
67    fn from(e: TreeErr<::sanakirja::Error>) -> Self {
68        TreeErr(e.0.into())
69    }
70}
71
72impl Pristine {
73    #[cfg(feature = "mmap")]
74    pub fn new<P: AsRef<Path>>(name: P) -> Result<Self, SanakirjaError> {
75        Self::new_with_size(name, 1 << 20)
76    }
77
78    /// # Safety
79    ///
80    /// Multiple processes might be writing to the same memory-mapped
81    /// file, leading to corruption. If you use this, make sure you
82    /// have another locking mechanism in place on the file.
83    #[cfg(feature = "mmap")]
84    pub unsafe fn new_nolock<P: AsRef<Path>>(name: P) -> Result<Self, SanakirjaError> {
85        unsafe { Self::new_with_size_nolock(name, 1 << 20) }
86    }
87
88    #[cfg(feature = "mmap")]
89    pub fn new_with_size<P: AsRef<Path>>(name: P, size: u64) -> Result<Self, SanakirjaError> {
90        let env = ::sanakirja::Env::new(name, size, 2);
91        match env {
92            Ok(env) => Ok(Pristine { env: Arc::new(env) }),
93            Err(::sanakirja::Error::IO(e)) => {
94                if let std::io::ErrorKind::WouldBlock = e.kind() {
95                    Err(SanakirjaError::PristineLocked)
96                } else {
97                    Err(SanakirjaError::Sanakirja(::sanakirja::Error::IO(e)))
98                }
99            }
100            Err(e) => Err(SanakirjaError::Sanakirja(e)),
101        }
102    }
103
104    /// # Safety
105    ///
106    /// Same as new_nolock:
107    ///
108    /// Multiple processes might be writing to the same memory-mapped
109    /// file, leading to corruption. If you use this, make sure you
110    /// have another locking mechanism in place on the file.
111    #[cfg(feature = "mmap")]
112    pub unsafe fn new_with_size_nolock<P: AsRef<Path>>(
113        name: P,
114        size: u64,
115    ) -> Result<Self, SanakirjaError> {
116        unsafe {
117            Ok(Pristine {
118                env: Arc::new(::sanakirja::Env::new_nolock(name, size, 2)?),
119            })
120        }
121    }
122    pub fn new_anon() -> Result<Self, SanakirjaError> {
123        Self::new_anon_with_size(1 << 20)
124    }
125    pub fn new_anon_with_size(size: u64) -> Result<Self, SanakirjaError> {
126        Ok(Pristine {
127            env: Arc::new(::sanakirja::Env::new_anon(size, 2)?),
128        })
129    }
130}
131
132#[derive(Debug, PartialEq, Clone, Copy)]
133#[repr(usize)]
134pub enum Root {
135    Version,
136    Tree,
137    RevTree,
138    Inodes,
139    RevInodes,
140    Internal,
141    External,
142    RevDep,
143    Channels,
144    TouchedFiles,
145    Dep,
146    RevTouchedFiles,
147    Partials,
148    Remotes,
149    Approvals,
150    Git0,
151    Git1,
152}
153
154const VERSION: u64 = 1u64;
155
156impl Pristine {
157    pub fn txn_begin(&self) -> Result<Txn, SanakirjaError> {
158        let txn = ::sanakirja::Env::txn_begin(self.env.clone())?;
159        if txn.root(Root::Version as usize) != VERSION {
160            return Err(SanakirjaError::Version);
161        }
162        debug!("txn_begin");
163        fn begin(txn: ::sanakirja::Txn<Arc<::sanakirja::Env>>) -> Option<Txn> {
164            Some(Txn {
165                channels: txn.root_db(Root::Channels as usize)?,
166                external: txn.root_db(Root::External as usize)?,
167                internal: txn.root_db(Root::Internal as usize)?,
168                inodes: txn.root_db(Root::Inodes as usize)?,
169                revinodes: txn.root_db(Root::RevInodes as usize)?,
170                tree: txn.root_db(Root::Tree as usize)?,
171                revtree: txn.root_db(Root::RevTree as usize)?,
172                revdep: txn.root_db(Root::RevDep as usize)?,
173                touched_files: txn.root_db(Root::TouchedFiles as usize)?,
174                rev_touched_files: txn.root_db(Root::RevTouchedFiles as usize)?,
175                partials: txn.root_db(Root::Partials as usize)?,
176                dep: txn.root_db(Root::Dep as usize)?,
177                remotes: txn.root_db(Root::Remotes as usize)?,
178                open_channels: Mutex::new(HashMap::default()),
179                open_remotes: Mutex::new(HashMap::default()),
180                txn,
181                counter: 0,
182                cur_channel: None,
183            })
184        }
185        debug!("txn begin done");
186        match begin(txn) {
187            Some(txn) => Ok(txn),
188            _ => Err(SanakirjaError::PristineCorrupt),
189        }
190    }
191
192    pub fn arc_txn_begin(
193        &self,
194    ) -> Result<ArcTxn<MutTxn<::sanakirja::MutTxn<Arc<::sanakirja::Env>>>>, SanakirjaError> {
195        Ok(ArcTxn(Arc::new(RwLock::new(self.mut_txn_begin()?))))
196    }
197
198    pub fn mut_txn_begin(
199        &self,
200    ) -> Result<MutTxn<::sanakirja::MutTxn<Arc<::sanakirja::Env>>>, SanakirjaError> {
201        unsafe {
202            let mut txn = ::sanakirja::Env::mut_txn_begin(self.env.clone()).unwrap();
203            if let Some(version) = txn.root(Root::Version as usize) {
204                if version != VERSION {
205                    return Err(SanakirjaError::Version);
206                }
207            } else {
208                txn.set_root(Root::Version as usize, VERSION);
209            }
210            Ok(MutTxn {
211                channels: if let Some(db) = txn.root_db(Root::Channels as usize) {
212                    db
213                } else {
214                    btree::create_db_(&mut txn)?
215                },
216                external: if let Some(db) = txn.root_db(Root::External as usize) {
217                    db
218                } else {
219                    btree::create_db_(&mut txn)?
220                },
221                internal: if let Some(db) = txn.root_db(Root::Internal as usize) {
222                    db
223                } else {
224                    btree::create_db_(&mut txn)?
225                },
226                inodes: if let Some(db) = txn.root_db(Root::Inodes as usize) {
227                    db
228                } else {
229                    btree::create_db_(&mut txn)?
230                },
231                revinodes: if let Some(db) = txn.root_db(Root::RevInodes as usize) {
232                    db
233                } else {
234                    btree::create_db_(&mut txn)?
235                },
236                tree: if let Some(db) = txn.root_db(Root::Tree as usize) {
237                    db
238                } else {
239                    btree::create_db_(&mut txn)?
240                },
241                revtree: if let Some(db) = txn.root_db(Root::RevTree as usize) {
242                    db
243                } else {
244                    btree::create_db_(&mut txn)?
245                },
246                revdep: if let Some(db) = txn.root_db(Root::RevDep as usize) {
247                    db
248                } else {
249                    btree::create_db_(&mut txn)?
250                },
251                dep: if let Some(db) = txn.root_db(Root::Dep as usize) {
252                    db
253                } else {
254                    btree::create_db_(&mut txn)?
255                },
256                touched_files: if let Some(db) = txn.root_db(Root::TouchedFiles as usize) {
257                    db
258                } else {
259                    btree::create_db_(&mut txn)?
260                },
261                rev_touched_files: if let Some(db) = txn.root_db(Root::RevTouchedFiles as usize) {
262                    db
263                } else {
264                    btree::create_db_(&mut txn)?
265                },
266                partials: if let Some(db) = txn.root_db(Root::Partials as usize) {
267                    db
268                } else {
269                    btree::create_db_(&mut txn)?
270                },
271                remotes: if let Some(db) = txn.root_db(Root::Remotes as usize) {
272                    db
273                } else {
274                    btree::create_db_(&mut txn)?
275                },
276                open_channels: Mutex::new(HashMap::default()),
277                open_remotes: Mutex::new(HashMap::default()),
278                txn,
279                counter: 0,
280                cur_channel: None,
281            })
282        }
283    }
284}
285
286pub type Txn = GenericTxn<::sanakirja::Txn<Arc<::sanakirja::Env>>>;
287pub type MutTxn<T> = GenericTxn<T>;
288pub type MutTxn0 = GenericTxn<::sanakirja::MutTxn<Arc<::sanakirja::Env>>>;
289pub trait RawMutTxnT:
290    ::sanakirja::AllocPage<Error = ::sanakirja::Error>
291    + ::sanakirja::RootPageMut
292    + ::sanakirja::Commit
293    + ::sanakirja::LoadPage<Error = ::sanakirja::Error>
294{
295}
296
297impl<
298    T: ::sanakirja::AllocPage<Error = ::sanakirja::Error>
299        + ::sanakirja::RootPageMut
300        + ::sanakirja::Commit
301        + ::sanakirja::LoadPage<Error = ::sanakirja::Error>,
302> RawMutTxnT for T
303{
304}
305
306/// A transaction, used both for mutable and immutable transactions,
307/// depending on type parameter `T`.
308///
309/// In Sanakirja, both `sanakirja::Txn` and `sanakirja::MutTxn`
310/// implement `sanakirja::Transaction`, explaining our implementation
311/// of `TxnT` for `Txn<T>` for all `T: sanakirja::Transaction`. This
312/// covers both mutable and immutable transactions in a single
313/// implementation.
314pub struct GenericTxn<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage>
315{
316    #[doc(hidden)]
317    pub txn: T,
318    #[doc(hidden)]
319    pub internal: UDb<SerializedHash, ChangeId>,
320    #[doc(hidden)]
321    pub external: UDb<ChangeId, SerializedHash>,
322    pub inodes: Db<Inode, Position<ChangeId>>,
323    pub revinodes: Db<Position<ChangeId>, Inode>,
324
325    pub tree: UDb<PathId, Inode>,
326    pub revtree: UDb<Inode, PathId>,
327
328    revdep: Db<ChangeId, ChangeId>,
329    dep: Db<ChangeId, ChangeId>,
330
331    touched_files: Db<Position<ChangeId>, ChangeId>,
332    rev_touched_files: Db<ChangeId, Position<ChangeId>>,
333
334    partials: UDb<SmallStr, Position<ChangeId>>,
335    channels: UDb<SmallStr, SerializedChannel>,
336    remotes: UDb<RemoteId, SerializedRemote>,
337
338    pub(crate) open_channels: Mutex<HashMap<SmallString, ChannelRef<Self>>>,
339    open_remotes: Mutex<HashMap<RemoteId, RemoteRef<Self>>>,
340    counter: usize,
341    cur_channel: Option<String>,
342}
343
344direct_repr!(SerializedPublicKey);
345
346/// This is actually safe because the only non-Send fields are
347/// `open_channels` and `open_remotes`, but we can't do anything with
348/// a `ChannelRef` whose transaction has been moved to another thread.
349unsafe impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> Send
350    for GenericTxn<T>
351{
352}
353
354impl Txn {
355    pub fn check_database(&self, refs: &mut std::collections::BTreeMap<u64, usize>) {
356        unsafe {
357            use ::sanakirja::debug::Check;
358            debug!("check: internal 0x{:x}", self.internal.db);
359            self.internal.add_refs(&self.txn, refs).unwrap();
360            debug!("check: external 0x{:x}", self.external.db);
361            self.external.add_refs(&self.txn, refs).unwrap();
362            debug!("check: inodes 0x{:x}", self.inodes.db);
363            self.inodes.add_refs(&self.txn, refs).unwrap();
364            debug!("check: revinodes 0x{:x}", self.revinodes.db);
365            self.revinodes.add_refs(&self.txn, refs).unwrap();
366            debug!("check: tree 0x{:x}", self.tree.db);
367            self.tree.add_refs(&self.txn, refs).unwrap();
368            debug!("check: revtree 0x{:x}", self.revtree.db);
369            self.revtree.add_refs(&self.txn, refs).unwrap();
370            debug!("check: revdep 0x{:x}", self.revdep.db);
371            self.revdep.add_refs(&self.txn, refs).unwrap();
372            debug!("check: dep 0x{:x}", self.dep.db);
373            self.dep.add_refs(&self.txn, refs).unwrap();
374            debug!("check: touched_files 0x{:x}", self.touched_files.db);
375            self.touched_files.add_refs(&self.txn, refs).unwrap();
376            debug!("check: rev_touched_files 0x{:x}", self.rev_touched_files.db);
377            self.rev_touched_files.add_refs(&self.txn, refs).unwrap();
378            debug!("check: partials 0x{:x}", self.partials.db);
379            self.partials.add_refs(&self.txn, refs).unwrap();
380            debug!("check: channels 0x{:x}", self.channels.db);
381            self.channels.add_refs(&self.txn, refs).unwrap();
382            for x in btree::iter(&self.txn, &self.channels, None).unwrap() {
383                let (name, tup) = x.unwrap();
384                debug!("check: channel name: {:?}", name.as_str());
385                let graph: Db<Vertex<ChangeId>, SerializedEdge> = Db::from_page(tup.graph.into());
386                let changes: Db<ChangeId, L64> = Db::from_page(tup.changes.into());
387                let revchanges: UDb<L64, Pair<ChangeId, SerializedMerkle>> =
388                    UDb::from_page(tup.revchanges.into());
389                let states: UDb<SerializedMerkle, L64> = UDb::from_page(tup.states.into());
390                let tags: Db<L64, Pair<SerializedMerkle, SerializedMerkle>> =
391                    Db::from_page(tup.tags.into());
392                debug!("check: graph 0x{:x}", graph.db);
393                graph.add_refs(&self.txn, refs).unwrap();
394                debug!("check: changes 0x{:x}", changes.db);
395                changes.add_refs(&self.txn, refs).unwrap();
396                debug!("check: revchanges 0x{:x}", revchanges.db);
397                revchanges.add_refs(&self.txn, refs).unwrap();
398                debug!("check: states 0x{:x}", states.db);
399                states.add_refs(&self.txn, refs).unwrap();
400                debug!("check: tags 0x{:x}", tags.db);
401                tags.add_refs(&self.txn, refs).unwrap();
402            }
403            debug!("check: remotes 0x{:x}", self.remotes.db);
404            self.remotes.add_refs(&self.txn, refs).unwrap();
405            for x in btree::iter(&self.txn, &self.remotes, None).unwrap() {
406                let (name, tup) = x.unwrap();
407                debug!("check: remote name: {:?}", name);
408                let remote: UDb<L64, Pair<SerializedHash, SerializedMerkle>> =
409                    UDb::from_page(tup.remote.into());
410
411                let rev: UDb<SerializedHash, L64> = UDb::from_page(tup.rev.into());
412                let states: UDb<SerializedMerkle, L64> = UDb::from_page(tup.states.into());
413                let tags: UDb<L64, Pair<SerializedMerkle, SerializedMerkle>> =
414                    UDb::from_page(tup.tags.into());
415                debug!("check: remote 0x{:x}", remote.db);
416                remote.add_refs(&self.txn, refs).unwrap();
417                debug!("check: rev 0x{:x}", rev.db);
418                rev.add_refs(&self.txn, refs).unwrap();
419                debug!("check: states 0x{:x}", states.db);
420                states.add_refs(&self.txn, refs).unwrap();
421                debug!("check: tags 0x{:x}", tags.db);
422                tags.add_refs(&self.txn, refs).unwrap();
423            }
424            ::sanakirja::debug::add_free_refs(&self.txn, refs).unwrap();
425            ::sanakirja::debug::check_free(&self.txn, refs);
426        }
427    }
428}
429
430impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> GraphTxnT
431    for GenericTxn<T>
432{
433    type Graph = Channel;
434    type GraphError = SanakirjaError;
435
436    fn get_graph<'txn>(
437        &'txn self,
438        db: &Self::Graph,
439        key: &Vertex<ChangeId>,
440        value: Option<&SerializedEdge>,
441    ) -> Result<Option<&'txn SerializedEdge>, TxnErr<Self::GraphError>> {
442        match ::sanakirja::btree::get(&self.txn, &db.graph, key, value) {
443            Ok(Some((k, v))) if k == key => Ok(Some(v)),
444            Ok(_) => Ok(None),
445            Err(e) => {
446                error!("{:?}", e);
447                Err(TxnErr(SanakirjaError::PristineCorrupt))
448            }
449        }
450    }
451
452    fn get_external(
453        &self,
454        p: &ChangeId,
455    ) -> Result<&SerializedHash, GetExternalError<Self::GraphError>> {
456        debug!("get_external {:?}", p);
457        if p.is_root() {
458            Ok(&HASH_NONE)
459        } else {
460            match btree::get(&self.txn, &self.external, p, None) {
461                Ok(Some((k, v))) if k == p => Ok(v),
462                Ok(_) => Err(GetExternalError::NotFound),
463                Err(e) => {
464                    error!("{:?}", e);
465                    Err(GetExternalError::Txn(SanakirjaError::PristineCorrupt))
466                }
467            }
468        }
469    }
470
471    fn get_internal(
472        &self,
473        p: &SerializedHash,
474    ) -> Result<Option<&ChangeId>, TxnErr<Self::GraphError>> {
475        if p.t == HashAlgorithm::None as u8 {
476            Ok(Some(&ChangeId::ROOT))
477        } else {
478            match btree::get(&self.txn, &self.internal, p, None) {
479                Ok(Some((k, v))) if k == p => Ok(Some(v)),
480                Ok(_) => Ok(None),
481                Err(e) => {
482                    error!("{:?}", e);
483                    Err(TxnErr(SanakirjaError::PristineCorrupt))
484                }
485            }
486        }
487    }
488
489    type Adj = Adj;
490
491    fn init_adj(
492        &self,
493        g: &Self::Graph,
494        key: Vertex<ChangeId>,
495        dest: Position<ChangeId>,
496        min_flag: EdgeFlags,
497        max_flag: EdgeFlags,
498    ) -> Result<Self::Adj, TxnErr<Self::GraphError>> {
499        let edge = SerializedEdge::new(min_flag, dest.change, dest.pos, ChangeId::ROOT);
500        let mut cursor = btree::cursor::Cursor::new(&self.txn, &g.graph).map_err(TxnErr)?;
501        cursor.set(&self.txn, &key, Some(&edge))?;
502        Ok(Adj {
503            cursor,
504            key,
505            min_flag,
506            max_flag,
507        })
508    }
509
510    fn next_adj<'a>(
511        &'a self,
512        _: &Self::Graph,
513        a: &mut Self::Adj,
514    ) -> Option<Result<&'a SerializedEdge, TxnErr<Self::GraphError>>> {
515        next_adj(&self.txn, a).map(|x| x.map_err(|x| TxnErr(x.into())))
516    }
517
518    fn find_block(
519        &self,
520        graph: &Self::Graph,
521        p: Position<ChangeId>,
522    ) -> Result<&Vertex<ChangeId>, BlockError<Self::GraphError>> {
523        Ok(find_block(&self.txn, &graph.graph, p)?)
524    }
525
526    fn find_block_end(
527        &self,
528        graph: &Self::Graph,
529        p: Position<ChangeId>,
530    ) -> Result<&Vertex<ChangeId>, BlockError<Self::GraphError>> {
531        Ok(find_block_end(&self.txn, &graph.graph, p)?)
532    }
533}
534
535impl std::convert::From<BlockError<::sanakirja::Error>> for BlockError<SanakirjaError> {
536    fn from(e: BlockError<::sanakirja::Error>) -> Self {
537        match e {
538            BlockError::Txn(t) => BlockError::Txn(t.into()),
539            BlockError::Block { block } => BlockError::Block { block },
540        }
541    }
542}
543
544#[doc(hidden)]
545pub fn next_adj<'a, T: ::sanakirja::LoadPage>(
546    txn: &'a T,
547    a: &mut Adj,
548) -> Option<Result<&'a SerializedEdge, T::Error>>
549where
550    T::Error: std::error::Error,
551{
552    loop {
553        let x: Result<Option<(&Vertex<ChangeId>, &SerializedEdge)>, _> = a.cursor.next(txn);
554        match x {
555            Ok(Some((v, e))) => {
556                if *v == a.key {
557                    if e.flag() >= a.min_flag {
558                        if e.flag() <= a.max_flag {
559                            return Some(Ok(e));
560                        } else {
561                            return None;
562                        }
563                    }
564                } else if *v > a.key {
565                    return None;
566                }
567            }
568            Err(e) => return Some(Err(e)),
569            Ok(None) => {
570                return None;
571            }
572        }
573    }
574}
575
576#[doc(hidden)]
577pub fn find_block<'a, T: ::sanakirja::LoadPage>(
578    txn: &'a T,
579    graph: &::sanakirja::btree::Db<Vertex<ChangeId>, SerializedEdge>,
580    p: Position<ChangeId>,
581) -> Result<&'a Vertex<ChangeId>, BlockError<T::Error>>
582where
583    T::Error: std::error::Error,
584{
585    if p.change.is_root() {
586        return Ok(&Vertex::ROOT);
587    }
588    let key = Vertex {
589        change: p.change,
590        start: p.pos,
591        end: p.pos,
592    };
593    let mut cursor = btree::cursor::Cursor::new(txn, graph).map_err(BlockError::Txn)?;
594    let mut k = if let Some((k, _)) = cursor.set(txn, &key, None).map_err(BlockError::Txn)? {
595        k
596    } else if let Some((k, _)) = cursor.prev(txn).map_err(BlockError::Txn)? {
597        k
598    } else {
599        debug!("find_block: BLOCK ERROR");
600        return Err(BlockError::Block { block: p });
601    };
602    // The only guarantee here is that k is either the first key >=
603    // `key`. We might need to rewind by one step if key is strictly
604    // larger than the result (i.e. if `p` is in the middle of the
605    // key).
606    while k.change > p.change || (k.change == p.change && k.start > p.pos) {
607        if let Some((k_, _)) = cursor.prev(txn).map_err(BlockError::Txn)? {
608            k = k_
609        } else {
610            break;
611        }
612    }
613    loop {
614        if k.change == p.change && k.start <= p.pos {
615            if k.end > p.pos || (k.start == k.end && k.end == p.pos) {
616                return Ok(k);
617            }
618        } else if k.change > p.change {
619            debug!("find_block: BLOCK ERROR");
620            return Err(BlockError::Block { block: p });
621        }
622        if let Some((k_, _)) = cursor.next(txn).map_err(BlockError::Txn)? {
623            k = k_
624        } else {
625            break;
626        }
627    }
628    debug!("find_block: BLOCK ERROR");
629    Err(BlockError::Block { block: p })
630}
631
632#[doc(hidden)]
633pub fn find_block_end<'a, T: ::sanakirja::LoadPage>(
634    txn: &'a T,
635    graph: &::sanakirja::btree::Db<Vertex<ChangeId>, SerializedEdge>,
636    p: Position<ChangeId>,
637) -> Result<&'a Vertex<ChangeId>, BlockError<T::Error>>
638where
639    T::Error: std::error::Error,
640{
641    if p.change.is_root() {
642        return Ok(&Vertex::ROOT);
643    }
644    let key = Vertex {
645        change: p.change,
646        start: p.pos,
647        end: p.pos,
648    };
649    let mut cursor = btree::cursor::Cursor::new(txn, graph).map_err(BlockError::Txn)?;
650    debug!("key {:?}", key);
651    let (mut k, v) = match cursor.set(txn, &key, None) {
652        Ok(Some((k, v))) => (k, v),
653        Ok(None) => {
654            if let Some((k, v)) = cursor.prev(txn).map_err(BlockError::Txn)? {
655                (k, v)
656            } else {
657                debug!("find_block_end, no prev");
658                return Err(BlockError::Block { block: p });
659            }
660        }
661        Err(e) => {
662            debug!("find_block_end: BLOCK ERROR 0 {:?}", e);
663            return Err(BlockError::Txn(e));
664        }
665    };
666    debug!("cursor {:?} {:?}", k, v);
667    loop {
668        debug!("find_block_end, loop, k = {:?}, p = {:?}", k, p);
669        if k.change < p.change {
670            break;
671        } else if k.change == p.change {
672            // Here we want to create an edge pointing between `p`
673            // and its successor. If k.start == p.pos, the only
674            // case where that's what we want is if k.start ==
675            // k.end.
676            if k.start == p.pos && k.end == p.pos {
677                return Ok(k);
678            } else if k.start < p.pos {
679                break;
680            }
681        }
682        if let Some((k_, _)) = cursor.prev(txn).map_err(BlockError::Txn)? {
683            k = k_
684        } else {
685            break;
686        }
687    }
688    // We also want k.end >= p.pos, so we just call next() until
689    // we have that.
690    debug!("find_block_end, {:?} {:?}", k, p);
691    while k.change < p.change || (k.change == p.change && p.pos > k.end) {
692        if let Some((k_, _)) = cursor.next(txn).map_err(BlockError::Txn)? {
693            k = k_
694        } else {
695            break;
696        }
697    }
698    debug!("find_block_end, {:?} {:?}", k, p);
699    if k.change == p.change
700        && ((k.start < p.pos && p.pos <= k.end) || (k.start == k.end && k.start == p.pos))
701    {
702        Ok(k)
703    } else {
704        debug!("find_block_end: BLOCK ERROR");
705        Err(BlockError::Block { block: p })
706    }
707}
708
709pub struct Adj {
710    pub cursor: ::sanakirja::btree::cursor::Cursor<
711        Vertex<ChangeId>,
712        SerializedEdge,
713        P<Vertex<ChangeId>, SerializedEdge>,
714    >,
715    pub key: Vertex<ChangeId>,
716    pub min_flag: EdgeFlags,
717    pub max_flag: EdgeFlags,
718}
719
720impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> GraphIter
721    for GenericTxn<T>
722{
723    type GraphCursor = ::sanakirja::btree::cursor::Cursor<
724        Vertex<ChangeId>,
725        SerializedEdge,
726        P<Vertex<ChangeId>, SerializedEdge>,
727    >;
728
729    fn graph_cursor(
730        &self,
731        g: &Self::Graph,
732        s: Option<&Vertex<ChangeId>>,
733    ) -> Result<Self::GraphCursor, TxnErr<Self::GraphError>> {
734        let mut c = ::sanakirja::btree::cursor::Cursor::new(&self.txn, &g.graph)?;
735        if let Some(s) = s {
736            c.set(&self.txn, s, None)?;
737        }
738        Ok(c)
739    }
740
741    fn next_graph<'txn>(
742        &'txn self,
743        _: &Self::Graph,
744        a: &mut Self::GraphCursor,
745    ) -> Option<Result<(&'txn Vertex<ChangeId>, &'txn SerializedEdge), TxnErr<Self::GraphError>>>
746    {
747        match a.next(&self.txn) {
748            Ok(Some(x)) => Some(Ok(x)),
749            Ok(None) => None,
750            Err(e) => {
751                error!("{:?}", e);
752                Some(Err(TxnErr(SanakirjaError::PristineCorrupt)))
753            }
754        }
755    }
756}
757
758// There is a choice here: the datastructure for `revchanges` is
759// intuitively a list. Moreover, when removing a change, we must
760// recompute the entire merkle tree after the removed change.
761//
762// This seems to indicate that a linked list could be an appropriate
763// structure (a growable array is excluded because amortised
764// complexity is not really acceptable here).
765//
766// However, we want to be able to answers queries such as "when was
767// change X introduced?" without having to read the entire database.
768//
769// Additionally, even though `SerializedMerkle` has only one
770// implementation, and is therefore sized in the current
771// implementation, we can't exclude that other algorithms may be
772// added, which means that the pages inside linked lists won't even be
773// randomly-accessible arrays.
774pub struct Channel {
775    pub graph: Db<Vertex<ChangeId>, SerializedEdge>,
776    pub changes: Db<ChangeId, L64>,
777    pub revchanges: UDb<L64, Pair<ChangeId, SerializedMerkle>>,
778    pub states: UDb<SerializedMerkle, L64>,
779    pub tags: Db<L64, Pair<SerializedMerkle, SerializedMerkle>>,
780    pub apply_counter: ApplyTimestamp,
781    pub name: SmallString,
782    pub last_modified: u64,
783    pub id: RemoteId,
784}
785
786impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> ChannelTxnT
787    for GenericTxn<T>
788{
789    type Channel = Channel;
790
791    fn graph<'a>(&self, c: &'a Self::Channel) -> &'a Channel {
792        c
793    }
794    fn name<'a>(&self, c: &'a Self::Channel) -> &'a str {
795        c.name.as_str()
796    }
797    fn id<'a>(&self, c: &'a Self::Channel) -> Option<&'a RemoteId> {
798        Some(&c.id)
799    }
800    fn apply_counter(&self, channel: &Self::Channel) -> u64 {
801        channel.apply_counter
802    }
803    fn last_modified(&self, channel: &Self::Channel) -> u64 {
804        channel.last_modified
805    }
806    fn changes<'a>(&self, channel: &'a Self::Channel) -> &'a Self::Changeset {
807        &channel.changes
808    }
809    fn rev_changes<'a>(&self, channel: &'a Self::Channel) -> &'a Self::RevChangeset {
810        &channel.revchanges
811    }
812    fn tags<'a>(&self, channel: &'a Self::Channel) -> &'a Self::Tags {
813        &channel.tags
814    }
815
816    type Changeset = Db<ChangeId, L64>;
817    type RevChangeset = UDb<L64, Pair<ChangeId, SerializedMerkle>>;
818
819    fn get_changeset(
820        &self,
821        channel: &Self::Changeset,
822        c: &ChangeId,
823    ) -> Result<Option<&L64>, TxnErr<Self::GraphError>> {
824        match btree::get(&self.txn, channel, c, None) {
825            Ok(Some((k, x))) if k == c => Ok(Some(x)),
826            Ok(x) => {
827                debug!("get_changeset = {:?}", x);
828                Ok(None)
829            }
830            Err(e) => {
831                error!("{:?}", e);
832                Err(TxnErr(SanakirjaError::PristineCorrupt))
833            }
834        }
835    }
836    fn get_revchangeset(
837        &self,
838        revchanges: &Self::RevChangeset,
839        c: &L64,
840    ) -> Result<Option<&Pair<ChangeId, SerializedMerkle>>, TxnErr<Self::GraphError>> {
841        match btree::get(&self.txn, revchanges, c, None) {
842            Ok(Some((k, x))) if k == c => Ok(Some(x)),
843            Ok(_) => Ok(None),
844            Err(e) => {
845                error!("{:?}", e);
846                Err(TxnErr(SanakirjaError::PristineCorrupt))
847            }
848        }
849    }
850
851    type ChangesetCursor = ::sanakirja::btree::cursor::Cursor<ChangeId, L64, P<ChangeId, L64>>;
852
853    fn cursor_changeset<'a>(
854        &'a self,
855        channel: &Self::Changeset,
856        pos: Option<ChangeId>,
857    ) -> Result<Cursor<Self, &'a Self, Self::ChangesetCursor, ChangeId, L64>, TxnErr<SanakirjaError>>
858    {
859        let mut cursor = btree::cursor::Cursor::new(&self.txn, channel)?;
860        if let Some(k) = pos {
861            cursor.set(&self.txn, &k, None)?;
862        }
863        Ok(Cursor {
864            cursor,
865            txn: self,
866            k: std::marker::PhantomData,
867            v: std::marker::PhantomData,
868            t: std::marker::PhantomData,
869        })
870    }
871
872    type RevchangesetCursor = ::sanakirja::btree::cursor::Cursor<
873        L64,
874        Pair<ChangeId, SerializedMerkle>,
875        UP<L64, Pair<ChangeId, SerializedMerkle>>,
876    >;
877
878    fn cursor_revchangeset_ref<'a, RT: std::ops::Deref<Target = Self>>(
879        txn: RT,
880        channel: &Self::RevChangeset,
881        pos: Option<L64>,
882    ) -> Result<
883        Cursor<Self, RT, Self::RevchangesetCursor, L64, Pair<ChangeId, SerializedMerkle>>,
884        TxnErr<SanakirjaError>,
885    > {
886        let mut cursor = btree::cursor::Cursor::new(&txn.txn, channel)?;
887        if let Some(k) = pos {
888            cursor.set(&txn.txn, &k, None)?;
889        }
890        Ok(Cursor {
891            cursor,
892            txn,
893            k: std::marker::PhantomData,
894            v: std::marker::PhantomData,
895            t: std::marker::PhantomData,
896        })
897    }
898
899    fn rev_cursor_revchangeset<'a>(
900        &'a self,
901        channel: &Self::RevChangeset,
902        pos: Option<L64>,
903    ) -> Result<
904        RevCursor<Self, &'a Self, Self::RevchangesetCursor, L64, Pair<ChangeId, SerializedMerkle>>,
905        TxnErr<SanakirjaError>,
906    > {
907        let mut cursor = btree::cursor::Cursor::new(&self.txn, channel)?;
908        if let Some(ref pos) = pos {
909            cursor.set(&self.txn, pos, None)?;
910        } else {
911            cursor.set_last(&self.txn)?;
912        };
913        Ok(RevCursor {
914            cursor,
915            txn: self,
916            k: std::marker::PhantomData,
917            v: std::marker::PhantomData,
918            t: std::marker::PhantomData,
919        })
920    }
921
922    fn cursor_revchangeset_next(
923        &self,
924        cursor: &mut Self::RevchangesetCursor,
925    ) -> Result<Option<(&L64, &Pair<ChangeId, SerializedMerkle>)>, TxnErr<SanakirjaError>> {
926        if let Ok(x) = cursor.next(&self.txn) {
927            Ok(x)
928        } else {
929            Err(TxnErr(SanakirjaError::PristineCorrupt))
930        }
931    }
932    fn cursor_revchangeset_prev(
933        &self,
934        cursor: &mut Self::RevchangesetCursor,
935    ) -> Result<Option<(&L64, &Pair<ChangeId, SerializedMerkle>)>, TxnErr<SanakirjaError>> {
936        if let Ok(x) = cursor.prev(&self.txn) {
937            Ok(x)
938        } else {
939            Err(TxnErr(SanakirjaError::PristineCorrupt))
940        }
941    }
942
943    fn cursor_changeset_next(
944        &self,
945        cursor: &mut Self::ChangesetCursor,
946    ) -> Result<Option<(&ChangeId, &L64)>, TxnErr<SanakirjaError>> {
947        if let Ok(x) = cursor.next(&self.txn) {
948            Ok(x)
949        } else {
950            Err(TxnErr(SanakirjaError::PristineCorrupt))
951        }
952    }
953    fn cursor_changeset_prev(
954        &self,
955        cursor: &mut Self::ChangesetCursor,
956    ) -> Result<Option<(&ChangeId, &L64)>, TxnErr<SanakirjaError>> {
957        if let Ok(x) = cursor.prev(&self.txn) {
958            Ok(x)
959        } else {
960            Err(TxnErr(SanakirjaError::PristineCorrupt))
961        }
962    }
963
964    type States = UDb<SerializedMerkle, L64>;
965    fn states<'a>(&self, channel: &'a Self::Channel) -> &'a Self::States {
966        &channel.states
967    }
968    fn channel_has_state(
969        &self,
970        channel: &Self::States,
971        m: &SerializedMerkle,
972    ) -> Result<Option<L64>, TxnErr<Self::GraphError>> {
973        match btree::get(&self.txn, channel, m, None)? {
974            Some((k, v)) if k == m => Ok(Some(*v)),
975            _ => Ok(None),
976        }
977    }
978
979    type Tags = Db<L64, Pair<SerializedMerkle, SerializedMerkle>>;
980
981    fn is_tagged(&self, tags: &Self::Tags, t: u64) -> Result<bool, TxnErr<Self::GraphError>> {
982        let t: L64 = t.into();
983        match btree::get(&self.txn, tags, &t, None)? {
984            Some((k, _)) => Ok(k == &t),
985            _ => Ok(false),
986        }
987    }
988
989    type TagsCursor = ::sanakirja::btree::cursor::Cursor<
990        L64,
991        Pair<SerializedMerkle, SerializedMerkle>,
992        P<L64, Pair<SerializedMerkle, SerializedMerkle>>,
993    >;
994    fn cursor_tags<'txn>(
995        &'txn self,
996        channel: &Self::Tags,
997        k: Option<L64>,
998    ) -> Result<
999        crate::pristine::Cursor<
1000            Self,
1001            &'txn Self,
1002            Self::TagsCursor,
1003            L64,
1004            Pair<SerializedMerkle, SerializedMerkle>,
1005        >,
1006        TxnErr<Self::GraphError>,
1007    > {
1008        let mut cursor = btree::cursor::Cursor::new(&self.txn, channel)?;
1009        if let Some(k) = k {
1010            cursor.set(&self.txn, &k, None)?;
1011        }
1012        Ok(Cursor {
1013            cursor,
1014            txn: self,
1015            k: std::marker::PhantomData,
1016            v: std::marker::PhantomData,
1017            t: std::marker::PhantomData,
1018        })
1019    }
1020    fn cursor_tags_next(
1021        &self,
1022        cursor: &mut Self::TagsCursor,
1023    ) -> Result<Option<(&L64, &Pair<SerializedMerkle, SerializedMerkle>)>, TxnErr<Self::GraphError>>
1024    {
1025        if let Ok(x) = cursor.next(&self.txn) {
1026            Ok(x)
1027        } else {
1028            Err(TxnErr(SanakirjaError::PristineCorrupt))
1029        }
1030    }
1031
1032    fn cursor_tags_prev(
1033        &self,
1034        cursor: &mut Self::TagsCursor,
1035    ) -> Result<Option<(&L64, &Pair<SerializedMerkle, SerializedMerkle>)>, TxnErr<Self::GraphError>>
1036    {
1037        if let Ok(x) = cursor.prev(&self.txn) {
1038            Ok(x)
1039        } else {
1040            Err(TxnErr(SanakirjaError::PristineCorrupt))
1041        }
1042    }
1043
1044    fn iter_tags(
1045        &self,
1046        channel: &Self::Tags,
1047        from: u64,
1048    ) -> Result<
1049        super::Cursor<Self, &Self, Self::TagsCursor, L64, Pair<SerializedMerkle, SerializedMerkle>>,
1050        TxnErr<Self::GraphError>,
1051    > {
1052        self.cursor_tags(channel, Some(from.into()))
1053    }
1054
1055    fn rev_iter_tags(
1056        &self,
1057        channel: &Self::Tags,
1058        from: Option<u64>,
1059    ) -> Result<
1060        super::RevCursor<
1061            Self,
1062            &Self,
1063            Self::TagsCursor,
1064            L64,
1065            Pair<SerializedMerkle, SerializedMerkle>,
1066        >,
1067        TxnErr<Self::GraphError>,
1068    > {
1069        let mut cursor = btree::cursor::Cursor::new(&self.txn, channel)?;
1070        if let Some(from) = from {
1071            cursor.set(&self.txn, &from.into(), None)?;
1072        } else {
1073            cursor.set_last(&self.txn)?;
1074        };
1075        Ok(RevCursor {
1076            cursor,
1077            txn: self,
1078            k: std::marker::PhantomData,
1079            v: std::marker::PhantomData,
1080            t: std::marker::PhantomData,
1081        })
1082    }
1083}
1084
1085impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> DepsTxnT
1086    for GenericTxn<T>
1087{
1088    type DepsError = SanakirjaError;
1089    type Dep = Db<ChangeId, ChangeId>;
1090    type Revdep = Db<ChangeId, ChangeId>;
1091
1092    sanakirja_table_get!(dep, ChangeId, ChangeId, DepsError);
1093    sanakirja_table_get!(revdep, ChangeId, ChangeId, DepsError);
1094    type DepCursor = ::sanakirja::btree::cursor::Cursor<ChangeId, ChangeId, P<ChangeId, ChangeId>>;
1095    sanakirja_cursor_ref!(dep, ChangeId, ChangeId);
1096    fn iter_dep_ref<RT: std::ops::Deref<Target = Self> + Clone>(
1097        txn: RT,
1098        p: &ChangeId,
1099    ) -> Result<super::Cursor<Self, RT, Self::DepCursor, ChangeId, ChangeId>, TxnErr<Self::DepsError>>
1100    {
1101        Self::cursor_dep_ref(txn.clone(), &txn.dep, Some((p, None)))
1102    }
1103
1104    sanakirja_table_get!(touched_files, Position<ChangeId>, ChangeId, DepsError);
1105    sanakirja_table_get!(rev_touched_files, ChangeId, Position<ChangeId>, DepsError);
1106
1107    type Touched_files = Db<Position<ChangeId>, ChangeId>;
1108
1109    type Rev_touched_files = Db<ChangeId, Position<ChangeId>>;
1110
1111    type Touched_filesCursor = ::sanakirja::btree::cursor::Cursor<
1112        Position<ChangeId>,
1113        ChangeId,
1114        P<Position<ChangeId>, ChangeId>,
1115    >;
1116    sanakirja_iter!(touched_files, Position<ChangeId>, ChangeId);
1117
1118    type Rev_touched_filesCursor = ::sanakirja::btree::cursor::Cursor<
1119        ChangeId,
1120        Position<ChangeId>,
1121        P<ChangeId, Position<ChangeId>>,
1122    >;
1123    sanakirja_iter!(rev_touched_files, ChangeId, Position<ChangeId>);
1124    fn iter_revdep(
1125        &self,
1126        k: &ChangeId,
1127    ) -> Result<
1128        super::Cursor<Self, &Self, Self::DepCursor, ChangeId, ChangeId>,
1129        TxnErr<Self::DepsError>,
1130    > {
1131        self.cursor_dep(&self.revdep, Some((k, None)))
1132    }
1133
1134    fn iter_dep(
1135        &self,
1136        k: &ChangeId,
1137    ) -> Result<
1138        super::Cursor<Self, &Self, Self::DepCursor, ChangeId, ChangeId>,
1139        TxnErr<Self::DepsError>,
1140    > {
1141        self.cursor_dep(&self.dep, Some((k, None)))
1142    }
1143
1144    fn iter_touched(
1145        &self,
1146        k: &Position<ChangeId>,
1147    ) -> Result<
1148        super::Cursor<Self, &Self, Self::Touched_filesCursor, Position<ChangeId>, ChangeId>,
1149        TxnErr<Self::DepsError>,
1150    > {
1151        self.cursor_touched_files(&self.touched_files, Some((k, None)))
1152    }
1153
1154    fn iter_rev_touched(
1155        &self,
1156        k: &ChangeId,
1157    ) -> Result<
1158        super::Cursor<Self, &Self, Self::Rev_touched_filesCursor, ChangeId, Position<ChangeId>>,
1159        TxnErr<Self::DepsError>,
1160    > {
1161        self.cursor_rev_touched_files(&self.rev_touched_files, Some((k, None)))
1162    }
1163}
1164
1165impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> TreeTxnT
1166    for GenericTxn<T>
1167{
1168    type TreeError = SanakirjaError;
1169    type Inodes = Db<Inode, Position<ChangeId>>;
1170    type Revinodes = Db<Position<ChangeId>, Inode>;
1171    sanakirja_table_get!(inodes, Inode, Position<ChangeId>, TreeError, TreeErr);
1172    sanakirja_table_get!(revinodes, Position<ChangeId>, Inode, TreeError, TreeErr);
1173    sanakirja_cursor!(inodes, Inode, Position<ChangeId>, TreeErr);
1174    // #[cfg(debug_assertions)]
1175    sanakirja_cursor!(revinodes, Position<ChangeId>, Inode, TreeErr);
1176
1177    type Tree = UDb<PathId, Inode>;
1178    sanakirja_table_get!(tree, PathId, Inode, TreeError, TreeErr);
1179    type TreeCursor = ::sanakirja::btree::cursor::Cursor<PathId, Inode, UP<PathId, Inode>>;
1180    sanakirja_iter!(tree, PathId, Inode, TreeErr);
1181    type RevtreeCursor = ::sanakirja::btree::cursor::Cursor<Inode, PathId, UP<Inode, PathId>>;
1182    sanakirja_iter!(revtree, Inode, PathId, TreeErr);
1183
1184    type Revtree = UDb<Inode, PathId>;
1185    sanakirja_table_get!(revtree, Inode, PathId, TreeError, TreeErr);
1186
1187    type Partials = UDb<SmallStr, Position<ChangeId>>;
1188    type PartialsCursor = ::sanakirja::btree::cursor::Cursor<
1189        SmallStr,
1190        Position<ChangeId>,
1191        UP<SmallStr, Position<ChangeId>>,
1192    >;
1193    sanakirja_cursor!(partials, SmallStr, Position<ChangeId>, TreeErr);
1194    type InodesCursor =
1195        ::sanakirja::btree::cursor::Cursor<Inode, Position<ChangeId>, P<Inode, Position<ChangeId>>>;
1196    fn iter_inodes(
1197        &self,
1198    ) -> Result<
1199        super::Cursor<Self, &Self, Self::InodesCursor, Inode, Position<ChangeId>>,
1200        TreeErr<Self::TreeError>,
1201    > {
1202        self.cursor_inodes(&self.inodes, None)
1203    }
1204
1205    // #[cfg(debug_assertions)]
1206    type RevinodesCursor =
1207        ::sanakirja::btree::cursor::Cursor<Position<ChangeId>, Inode, P<Position<ChangeId>, Inode>>;
1208    // #[cfg(debug_assertions)]
1209    fn iter_revinodes(
1210        &self,
1211    ) -> Result<
1212        super::Cursor<Self, &Self, Self::RevinodesCursor, Position<ChangeId>, Inode>,
1213        TreeErr<SanakirjaError>,
1214    > {
1215        self.cursor_revinodes(&self.revinodes, None)
1216    }
1217
1218    fn iter_partials<'txn>(
1219        &'txn self,
1220        k0: &crate::small_string::SmallStr,
1221    ) -> Result<
1222        super::Cursor<Self, &'txn Self, Self::PartialsCursor, SmallStr, Position<ChangeId>>,
1223        TreeErr<SanakirjaError>,
1224    > {
1225        self.cursor_partials(&self.partials, Some((k0, None)))
1226    }
1227}
1228
1229impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> GenericTxn<T> {
1230    #[doc(hidden)]
1231    pub unsafe fn unsafe_load_channel(
1232        &self,
1233        name: SmallString,
1234    ) -> Result<Option<Channel>, TxnErr<SanakirjaError>> {
1235        unsafe {
1236            debug!("unsafe load channel");
1237            match btree::get(&self.txn, &self.channels, &name, None)? {
1238                Some((name_, tup)) if name_ == name.as_ref() => {
1239                    debug!("load_channel: {:?} {:?}", name, tup);
1240                    Ok(Some(Channel {
1241                        graph: Db::from_page(tup.graph.into()),
1242                        changes: Db::from_page(tup.changes.into()),
1243                        revchanges: UDb::from_page(tup.revchanges.into()),
1244                        states: UDb::from_page(tup.states.into()),
1245                        tags: Db::from_page(tup.tags.into()),
1246                        apply_counter: tup.apply_counter.into(),
1247                        last_modified: tup.last_modified.into(),
1248                        id: tup.id,
1249                        name,
1250                    }))
1251                }
1252                _ => {
1253                    debug!("unsafe_load_channel: not found");
1254                    Ok(None)
1255                }
1256            }
1257        }
1258    }
1259}
1260
1261impl<T: ::sanakirja::LoadPage<Error = ::sanakirja::Error> + ::sanakirja::RootPage> TxnT
1262    for GenericTxn<T>
1263{
1264    fn hash_from_prefix(
1265        &self,
1266        s: &str,
1267    ) -> Result<(Hash, ChangeId), super::HashPrefixError<Self::GraphError>> {
1268        let mut result = None;
1269        for h in prefix_guesses(s) {
1270            let h: SerializedHash = (&h).into();
1271            debug!("h = {:?}", h);
1272            for x in btree::iter(&self.txn, &self.internal, Some((&h, None)))
1273                .map_err(|e| super::HashPrefixError::Txn(e.into()))?
1274            {
1275                let (e, i) = x.map_err(|e| super::HashPrefixError::Txn(e.into()))?;
1276                debug!("{:?}", e);
1277                if e < &h {
1278                    continue;
1279                } else {
1280                    let e: Hash = e.into();
1281                    let b32 = e.to_base32();
1282                    debug!("{:?}", b32);
1283                    let (b32, _) = b32.split_at(s.len().min(b32.len()));
1284                    if b32 != s {
1285                        break;
1286                    } else if result.is_none() {
1287                        result = Some((e, *i))
1288                    } else {
1289                        return Err(super::HashPrefixError::Ambiguous(s.to_string()));
1290                    }
1291                }
1292            }
1293        }
1294        if let Some(result) = result {
1295            return Ok(result);
1296        }
1297        Err(super::HashPrefixError::NotFound(s.to_string()))
1298    }
1299
1300    fn state_from_prefix(
1301        &self,
1302        channel: &Self::States,
1303        s: &str,
1304    ) -> Result<(Merkle, L64), super::HashPrefixError<Self::GraphError>> {
1305        let h = if let Some(h) = SerializedMerkle::from_prefix(s) {
1306            h
1307        } else {
1308            return Err(super::HashPrefixError::Parse(s.to_string()));
1309        };
1310        let mut result = None;
1311        debug!("h = {:?}", h);
1312        for x in btree::iter(&self.txn, channel, Some((&h, None)))
1313            .map_err(|e| super::HashPrefixError::Txn(e.into()))?
1314        {
1315            let (e, i) = x.map_err(|e| super::HashPrefixError::Txn(e.into()))?;
1316            debug!("{:?} {:?}", e, i);
1317            if e < &h {
1318                continue;
1319            } else {
1320                let e: Merkle = e.into();
1321                let b32 = e.to_base32();
1322                debug!("{:?}", b32);
1323                let (b32, _) = b32.split_at(s.len().min(b32.len()));
1324                if b32 != s {
1325                    break;
1326                } else if result.is_none() {
1327                    result = Some((e, *i))
1328                } else {
1329                    return Err(super::HashPrefixError::Ambiguous(s.to_string()));
1330                }
1331            }
1332        }
1333        if let Some(result) = result {
1334            Ok(result)
1335        } else {
1336            Err(super::HashPrefixError::NotFound(s.to_string()))
1337        }
1338    }
1339
1340    fn hash_from_prefix_remote(
1341        &self,
1342        remote: &RemoteRef<Self>,
1343        s: &str,
1344    ) -> Result<Hash, super::HashPrefixError<Self::GraphError>> {
1345        let remote = remote.db.lock();
1346
1347        let mut result = None;
1348
1349        for h in prefix_guesses(s) {
1350            let h: SerializedHash = (&h).into();
1351            debug!("h = {:?}", h);
1352            for x in btree::iter(&self.txn, &remote.rev, Some((&h, None)))
1353                .map_err(|e| super::HashPrefixError::Txn(e.into()))?
1354            {
1355                let (e, _) = x.map_err(|e| super::HashPrefixError::Txn(e.into()))?;
1356                debug!("{:?}", e);
1357                if e < &h {
1358                    continue;
1359                } else {
1360                    let e: Hash = e.into();
1361                    let b32 = e.to_base32();
1362                    debug!("{:?}", b32);
1363                    let (b32, _) = b32.split_at(s.len().min(b32.len()));
1364                    if b32 != s {
1365                        break;
1366                    } else if result.is_none() {
1367                        result = Some(e)
1368                    } else {
1369                        return Err(super::HashPrefixError::Ambiguous(s.to_string()));
1370                    }
1371                }
1372            }
1373        }
1374        if let Some(result) = result {
1375            return Ok(result);
1376        }
1377        Err(super::HashPrefixError::NotFound(s.to_string()))
1378    }
1379
1380    fn load_channel(
1381        &self,
1382        name: SmallString,
1383    ) -> Result<Option<ChannelRef<Self>>, TxnErr<Self::GraphError>> {
1384        match self.open_channels.lock().entry(name.clone()) {
1385            Entry::Vacant(v) => {
1386                if let Some(c) = unsafe { self.unsafe_load_channel(name)? } {
1387                    Ok(Some(
1388                        v.insert(ChannelRef {
1389                            r: Arc::new(RwLock::new(c)),
1390                        })
1391                        .clone(),
1392                    ))
1393                } else {
1394                    Ok(None)
1395                }
1396            }
1397            Entry::Occupied(occ) => Ok(Some(occ.get().clone())),
1398        }
1399    }
1400
1401    fn load_remote(
1402        &self,
1403        name: &RemoteId,
1404    ) -> Result<Option<RemoteRef<Self>>, TxnErr<Self::GraphError>> {
1405        unsafe {
1406            let name = name.to_owned();
1407            match self.open_remotes.lock().entry(name) {
1408                Entry::Vacant(v) => match btree::get(&self.txn, &self.remotes, &name, None)? {
1409                    Some((name_, remote)) if name == *name_ => {
1410                        debug!("load_remote: {:?} {:?}", name_, remote);
1411                        let r = Remote {
1412                            remote: UDb::from_page(remote.remote.into()),
1413                            rev: UDb::from_page(remote.rev.into()),
1414                            states: UDb::from_page(remote.states.into()),
1415                            id_rev: remote.id_rev,
1416                            tags: Db::from_page(remote.tags.into()),
1417                            path: remote.path.to_owned(),
1418                        };
1419                        for x in btree::iter(&self.txn, &r.remote, None).unwrap() {
1420                            debug!("remote -> {:?}", x);
1421                        }
1422                        for x in btree::iter(&self.txn, &r.rev, None).unwrap() {
1423                            debug!("rev -> {:?}", x);
1424                        }
1425                        for x in btree::iter(&self.txn, &r.states, None).unwrap() {
1426                            debug!("states -> {:?}", x);
1427                        }
1428
1429                        for x in self.iter_remote(&r.remote, 0).unwrap() {
1430                            debug!("ITER {:?}", x);
1431                        }
1432
1433                        let r = RemoteRef {
1434                            db: Arc::new(Mutex::new(r)),
1435                            id: name,
1436                        };
1437                        Ok(Some(v.insert(r).clone()))
1438                    }
1439                    _ => Ok(None),
1440                },
1441                Entry::Occupied(occ) => Ok(Some(occ.get().clone())),
1442            }
1443        }
1444    }
1445
1446    type Channels = UDb<SmallStr, SerializedChannel>;
1447    type ChannelsCursor = ::sanakirja::btree::cursor::Cursor<
1448        SmallStr,
1449        SerializedChannel,
1450        UP<SmallStr, SerializedChannel>,
1451    >;
1452    sanakirja_cursor!(channels, SmallStr, SerializedChannel);
1453    fn channels(
1454        &self,
1455        start: &SmallStr,
1456    ) -> Result<Vec<ChannelRef<Self>>, TxnErr<Self::GraphError>> {
1457        let name = start;
1458        let mut cursor = btree::cursor::Cursor::new(&self.txn, &self.channels)?;
1459        cursor.set(&self.txn, name, None)?;
1460        while let Ok(Some((name, _))) = self.cursor_channels_next(&mut cursor) {
1461            self.load_channel(name.to_owned())?;
1462        }
1463        Ok(self.open_channels.lock().values().cloned().collect())
1464    }
1465
1466    type Remotes = UDb<RemoteId, SerializedRemote>;
1467    type RemotesCursor = ::sanakirja::btree::cursor::Cursor<
1468        RemoteId,
1469        SerializedRemote,
1470        UP<RemoteId, SerializedRemote>,
1471    >;
1472    sanakirja_cursor!(remotes, RemoteId, SerializedRemote);
1473    fn iter_remotes<'txn>(
1474        &'txn self,
1475        start: &RemoteId,
1476    ) -> Result<RemotesIterator<'txn, Self>, TxnErr<Self::GraphError>> {
1477        let mut cursor = btree::cursor::Cursor::new(&self.txn, &self.remotes)?;
1478        cursor.set(&self.txn, start, None)?;
1479        Ok(RemotesIterator { cursor, txn: self })
1480    }
1481
1482    type Remote = UDb<L64, Pair<SerializedHash, SerializedMerkle>>;
1483    type Revremote = UDb<SerializedHash, L64>;
1484    type Remotestates = UDb<SerializedMerkle, L64>;
1485    type Remotetags = UDb<L64, Pair<SerializedMerkle, SerializedMerkle>>;
1486    type RemoteCursor = ::sanakirja::btree::cursor::Cursor<
1487        L64,
1488        Pair<SerializedHash, SerializedMerkle>,
1489        UP<L64, Pair<SerializedHash, SerializedMerkle>>,
1490    >;
1491    sanakirja_cursor!(remote, L64, Pair<SerializedHash, SerializedMerkle>);
1492    sanakirja_rev_cursor!(remote, L64, Pair<SerializedHash, SerializedMerkle>);
1493
1494    fn iter_remote<'txn>(
1495        &'txn self,
1496        remote: &Self::Remote,
1497        k: u64,
1498    ) -> Result<
1499        super::Cursor<
1500            Self,
1501            &'txn Self,
1502            Self::RemoteCursor,
1503            L64,
1504            Pair<SerializedHash, SerializedMerkle>,
1505        >,
1506        TxnErr<Self::GraphError>,
1507    > {
1508        self.cursor_remote(remote, Some((&k.into(), None)))
1509    }
1510
1511    fn iter_rev_remote<'txn>(
1512        &'txn self,
1513        remote: &Self::Remote,
1514        k: Option<L64>,
1515    ) -> Result<
1516        super::RevCursor<
1517            Self,
1518            &'txn Self,
1519            Self::RemoteCursor,
1520            L64,
1521            Pair<SerializedHash, SerializedMerkle>,
1522        >,
1523        TxnErr<Self::GraphError>,
1524    > {
1525        self.rev_cursor_remote(remote, k.as_ref().map(|k| (k, None)))
1526    }
1527
1528    fn get_remote(
1529        &mut self,
1530        name: RemoteId,
1531    ) -> Result<Option<RemoteRef<Self>>, TxnErr<Self::GraphError>> {
1532        unsafe {
1533            let name = name.to_owned();
1534            match self.open_remotes.lock().entry(name) {
1535                Entry::Vacant(v) => match btree::get(&self.txn, &self.remotes, &name, None)? {
1536                    Some((name_, remote)) if *name_ == name => {
1537                        let r = RemoteRef {
1538                            db: Arc::new(Mutex::new(Remote {
1539                                remote: UDb::from_page(remote.remote.into()),
1540                                rev: UDb::from_page(remote.rev.into()),
1541                                states: UDb::from_page(remote.states.into()),
1542                                id_rev: remote.id_rev,
1543                                tags: Db::from_page(remote.tags.into()),
1544                                path: remote.path.to_owned(),
1545                            })),
1546                            id: name,
1547                        };
1548                        v.insert(r);
1549                    }
1550                    _ => return Ok(None),
1551                },
1552                Entry::Occupied(_) => {}
1553            }
1554            Ok(self.open_remotes.lock().get(&name).cloned())
1555        }
1556    }
1557
1558    fn last_remote(
1559        &self,
1560        remote: &Self::Remote,
1561    ) -> Result<Option<(u64, &Pair<SerializedHash, SerializedMerkle>)>, TxnErr<Self::GraphError>>
1562    {
1563        debug!("last_remote: {:?}", remote);
1564        if let Some(x) = btree::rev_iter(&self.txn, remote, None)?.next() {
1565            let (&k, v) = x?;
1566            Ok(Some((k.into(), v)))
1567        } else {
1568            Ok(None)
1569        }
1570    }
1571
1572    fn last_remote_tag(
1573        &self,
1574        remote: &Self::Tags,
1575    ) -> Result<Option<(u64, &SerializedMerkle, &SerializedMerkle)>, TxnErr<Self::GraphError>> {
1576        if let Some(x) = btree::rev_iter(&self.txn, remote, None)?.next() {
1577            let (&k, v) = x?;
1578            Ok(Some((k.into(), &v.a, &v.b)))
1579        } else {
1580            Ok(None)
1581        }
1582    }
1583
1584    fn get_remote_state(
1585        &self,
1586        remote: &Self::Remote,
1587        n: u64,
1588    ) -> Result<Option<(u64, &Pair<SerializedHash, SerializedMerkle>)>, TxnErr<Self::GraphError>>
1589    {
1590        let n = n.into();
1591        for x in btree::iter(&self.txn, remote, Some((&n, None)))? {
1592            let (&k, m) = x?;
1593            if k >= n {
1594                return Ok(Some((k.into(), m)));
1595            }
1596        }
1597        Ok(None)
1598    }
1599
1600    fn get_remote_tag(
1601        &self,
1602        remote: &Self::Tags,
1603        n: u64,
1604    ) -> Result<Option<(u64, &Pair<SerializedMerkle, SerializedMerkle>)>, TxnErr<Self::GraphError>>
1605    {
1606        let n = n.into();
1607        if let Some(x) = btree::rev_iter(&self.txn, remote, Some((&n, None)))?.next() {
1608            let (&k, m) = x?;
1609            Ok(Some((k.into(), m)))
1610        } else {
1611            Ok(None)
1612        }
1613    }
1614
1615    fn remote_has_change(
1616        &self,
1617        remote: &RemoteRef<Self>,
1618        hash: &SerializedHash,
1619    ) -> Result<bool, TxnErr<Self::GraphError>> {
1620        match btree::get(&self.txn, &remote.db.lock().rev, hash, None)? {
1621            Some((k, _)) if k == hash => Ok(true),
1622            _ => Ok(false),
1623        }
1624    }
1625    fn remote_has_state(
1626        &self,
1627        remote: &RemoteRef<Self>,
1628        m: &SerializedMerkle,
1629    ) -> Result<Option<u64>, TxnErr<Self::GraphError>> {
1630        match btree::get(&self.txn, &remote.db.lock().states, m, None)? {
1631            Some((k, v)) if k == m => Ok(Some((*v).into())),
1632            _ => Ok(None),
1633        }
1634    }
1635    fn current_channel(&self) -> Result<&str, Self::GraphError> {
1636        if let Some(ref c) = self.cur_channel {
1637            Ok(c)
1638        } else {
1639            unsafe {
1640                let b = self.txn.root_page();
1641                let len = b[4096 - 256] as usize;
1642                if len == 0 {
1643                    Ok("main")
1644                } else {
1645                    let s = std::slice::from_raw_parts(b.as_ptr().add(4096 - 255), len);
1646                    Ok(std::str::from_utf8(s).unwrap_or("main"))
1647                }
1648            }
1649        }
1650    }
1651}
1652
1653impl<
1654    T: sanakirja::AllocPage<Error = ::sanakirja::Error>
1655        + sanakirja::RootPage
1656        + sanakirja::LoadPage<Error = ::sanakirja::Error>,
1657> GraphMutTxnT for MutTxn<T>
1658{
1659    fn put_graph(
1660        &mut self,
1661        graph: &mut Self::Graph,
1662        k: &Vertex<ChangeId>,
1663        e: &SerializedEdge,
1664    ) -> Result<bool, TxnErr<Self::GraphError>> {
1665        Ok(btree::put(&mut self.txn, &mut graph.graph, k, e)?)
1666    }
1667
1668    fn del_graph(
1669        &mut self,
1670        graph: &mut Self::Graph,
1671        k: &Vertex<ChangeId>,
1672        e: Option<&SerializedEdge>,
1673    ) -> Result<bool, TxnErr<Self::GraphError>> {
1674        Ok(btree::del(&mut self.txn, &mut graph.graph, k, e)?)
1675    }
1676
1677    fn debug(&mut self, graph: &mut Self::Graph, extra: &str) {
1678        ::sanakirja::debug::debug(
1679            &self.txn,
1680            &[&graph.graph],
1681            format!("debug{}{}", self.counter, extra),
1682            true,
1683        );
1684    }
1685
1686    sanakirja_put_del!(internal, SerializedHash, ChangeId, GraphError);
1687    sanakirja_put_del!(external, ChangeId, SerializedHash, GraphError);
1688
1689    fn split_block(
1690        &mut self,
1691        graph: &mut Self::Graph,
1692        key: &Vertex<ChangeId>,
1693        pos: ChangePosition,
1694        buf: &mut Vec<SerializedEdge>,
1695    ) -> Result<(), TxnErr<Self::GraphError>> {
1696        assert!(pos > key.start);
1697        assert!(pos < key.end);
1698        let mut cursor = btree::cursor::Cursor::new(&self.txn, &graph.graph)?;
1699        cursor.set(&self.txn, key, None)?;
1700        loop {
1701            match cursor.next(&self.txn) {
1702                Ok(Some((k, v))) => {
1703                    if k > key {
1704                        break;
1705                    } else if k < key {
1706                        continue;
1707                    }
1708                    buf.push(*v)
1709                }
1710                Ok(None) => break,
1711                Err(e) => {
1712                    error!("{:?}", e);
1713                    return Err(TxnErr(SanakirjaError::PristineCorrupt));
1714                }
1715            }
1716        }
1717        for chi in buf.drain(..) {
1718            assert!(
1719                chi.introduced_by() != ChangeId::ROOT || chi.flag().contains(EdgeFlags::PSEUDO)
1720            );
1721            if chi.flag().contains(EdgeFlags::PARENT | EdgeFlags::BLOCK) {
1722                put_graph_with_rev(
1723                    self,
1724                    graph,
1725                    chi.flag() - EdgeFlags::PARENT,
1726                    Vertex {
1727                        change: key.change,
1728                        start: key.start,
1729                        end: pos,
1730                    },
1731                    Vertex {
1732                        change: key.change,
1733                        start: pos,
1734                        end: key.end,
1735                    },
1736                    chi.introduced_by(),
1737                )?;
1738            }
1739
1740            self.del_graph(graph, key, Some(&chi))?;
1741            self.put_graph(
1742                graph,
1743                &if chi.flag().contains(EdgeFlags::PARENT) {
1744                    Vertex {
1745                        change: key.change,
1746                        start: key.start,
1747                        end: pos,
1748                    }
1749                } else {
1750                    Vertex {
1751                        change: key.change,
1752                        start: pos,
1753                        end: key.end,
1754                    }
1755                },
1756                &chi,
1757            )?;
1758        }
1759        Ok(())
1760    }
1761}
1762
1763impl<
1764    T: sanakirja::AllocPage<Error = ::sanakirja::Error>
1765        + sanakirja::RootPage
1766        + sanakirja::LoadPage<Error = ::sanakirja::Error>,
1767> ChannelMutTxnT for MutTxn<T>
1768{
1769    fn graph_mut(c: &mut Self::Channel) -> &mut Self::Graph {
1770        c
1771    }
1772    fn touch_channel(&mut self, channel: &mut Self::Channel, t: Option<u64>) {
1773        use std::time::SystemTime;
1774        debug!("touch_channel: {:?}", t);
1775        if let Some(t) = t {
1776            channel.last_modified = t
1777        } else if let Ok(duration) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
1778            debug!("touch {:?}", duration.as_secs() * 1000);
1779            channel.last_modified = duration.as_secs() * 1000
1780        }
1781    }
1782
1783    fn put_changes(
1784        &mut self,
1785        channel: &mut Self::Channel,
1786        p: ChangeId,
1787        t: ApplyTimestamp,
1788        h: &Hash,
1789    ) -> Result<Option<Merkle>, TxnErr<Self::GraphError>> {
1790        debug!("put_changes {:?} {:?}", p, h);
1791        if let Some(m) = self.get_changeset(&channel.changes, &p)? {
1792            debug!("found m = {:?}, p = {:?}", m, p);
1793            Ok(None)
1794        } else {
1795            channel.apply_counter += 1;
1796            debug!("put_changes {:?} {:?}", t, p);
1797            let m = if let Some(x) = btree::rev_iter(&self.txn, &channel.revchanges, None)?.next() {
1798                let (a, b) = x?;
1799                let a: u64 = (*a).into();
1800                assert!(a < t);
1801                (&b.b).into()
1802            } else {
1803                Merkle::zero()
1804            };
1805            let m = m.next(h);
1806            assert!(
1807                self.get_revchangeset(&channel.revchanges, &t.into())?
1808                    .is_none()
1809            );
1810            assert!(btree::put(
1811                &mut self.txn,
1812                &mut channel.changes,
1813                &p,
1814                &t.into()
1815            )?);
1816            assert!(btree::put(
1817                &mut self.txn,
1818                &mut channel.revchanges,
1819                &t.into(),
1820                &Pair { a: p, b: m.into() }
1821            )?);
1822            assert!(btree::put(
1823                &mut self.txn,
1824                &mut channel.states,
1825                &m.into(),
1826                &t.into(),
1827            )?);
1828            Ok(Some(m))
1829        }
1830    }
1831
1832    fn del_changes(
1833        &mut self,
1834        channel: &mut Self::Channel,
1835        p: ChangeId,
1836        t: ApplyTimestamp,
1837    ) -> Result<bool, TxnErr<Self::GraphError>> {
1838        let mut repl = Vec::new();
1839        let tl = t.into();
1840        for x in btree::iter(&self.txn, &channel.revchanges, Some((&tl, None)))? {
1841            let (t_, p) = x?;
1842            if *t_ >= tl {
1843                repl.push((*t_, p.a, p.b))
1844            }
1845        }
1846        let mut m = Merkle::zero();
1847        for x in btree::rev_iter(&self.txn, &channel.revchanges, Some((&tl, None)))? {
1848            let (t_, mm) = x?;
1849            if t_ < &tl {
1850                m = (&mm.b).into();
1851                break;
1852            }
1853        }
1854        for (t_, p, m0) in repl.iter() {
1855            debug!("del_changes {:?} {:?}", t_, p);
1856            btree::del(&mut self.txn, &mut channel.revchanges, t_, None)?;
1857            btree::del(&mut self.txn, &mut channel.states, m0, None)?;
1858            if *t_ > tl {
1859                m = m.next(self.get_external(p).optional()?.unwrap());
1860                btree::put(
1861                    &mut self.txn,
1862                    &mut channel.revchanges,
1863                    t_,
1864                    &Pair { a: *p, b: m.into() },
1865                )?;
1866                btree::put(&mut self.txn, &mut channel.states, &m.into(), t_)?;
1867            }
1868        }
1869        btree::del(&mut self.txn, &mut channel.tags, &t.into(), None)?;
1870        Ok(btree::del(
1871            &mut self.txn,
1872            &mut channel.changes,
1873            &p,
1874            Some(&t.into()),
1875        )?)
1876    }
1877
1878    fn tags_mut<'a>(&mut self, channel: &'a mut Self::Channel) -> &'a mut Self::Tags {
1879        &mut channel.tags
1880    }
1881
1882    fn put_tags(
1883        &mut self,
1884        channel: &mut Self::Tags,
1885        n: u64,
1886        m: &Merkle,
1887    ) -> Result<(), TxnErr<Self::GraphError>> {
1888        debug!("put_tags {:?}", m);
1889        let mm: SerializedMerkle = m.into();
1890        if btree::get(&self.txn, channel, &n.into(), None)?.is_some() {
1891            debug!("already tagged");
1892            Ok(())
1893        } else {
1894            let tl = n.into();
1895            let mut repl = vec![(tl, mm)];
1896            replay_tags(self, channel, tl, &mut repl)?;
1897            Ok(())
1898        }
1899    }
1900
1901    fn del_tags(
1902        &mut self,
1903        channel: &mut Self::Tags,
1904        t: u64,
1905    ) -> Result<(), TxnErr<Self::GraphError>> {
1906        replay_tags(self, channel, t.into(), &mut Vec::new())?;
1907        Ok(())
1908    }
1909
1910    fn move_change(
1911        &mut self,
1912        channel: &mut Self::Channel,
1913        change_id: ChangeId,
1914        old_pos: ApplyTimestamp,
1915        new_pos: ApplyTimestamp,
1916    ) -> Result<(), TxnErr<Self::GraphError>> {
1917        if old_pos == new_pos {
1918            return Ok(());
1919        }
1920        let new_l64 = L64::from(new_pos);
1921        let old_l64 = L64::from(old_pos);
1922        // Collect all entries in [new_pos, old_pos] — the range being reordered.
1923        let mut entries: Vec<(L64, ChangeId)> = Vec::new();
1924        for x in btree::iter(&self.txn, &channel.revchanges, Some((&new_l64, None)))? {
1925            let (t, p) = x?;
1926            let t_u64: u64 = (*t).into();
1927            if t_u64 > old_pos {
1928                break;
1929            }
1930            entries.push((*t, p.a));
1931        }
1932        // Remove them all from revchanges, states, and changes.
1933        let mut merkles: Vec<SerializedMerkle> = Vec::with_capacity(entries.len());
1934        for (t, _) in entries.iter() {
1935            let m = self.get_revchangeset(&channel.revchanges, t)?.unwrap().b;
1936            merkles.push(m);
1937        }
1938        for ((t, p), m) in entries.iter().zip(merkles.iter()) {
1939            btree::del(&mut self.txn, &mut channel.revchanges, t, None)?;
1940            btree::del(&mut self.txn, &mut channel.states, m, None)?;
1941            btree::del(&mut self.txn, &mut channel.changes, p, Some(t))?;
1942        }
1943        // Starting Merkle is the value just before new_pos.
1944        let mut m: Merkle = {
1945            let mut it = btree::rev_iter(&self.txn, &channel.revchanges, Some((&new_l64, None)))?;
1946            match it.next() {
1947                Some(x) => {
1948                    let (t_, mm) = x?;
1949                    if *t_ < new_l64 {
1950                        (&mm.b).into()
1951                    } else {
1952                        Merkle::zero()
1953                    }
1954                }
1955                None => Merkle::zero(),
1956            }
1957        };
1958        // New order: change_id first (it was at old_pos), then entries[0..last] in original order.
1959        let last = entries.len() - 1;
1960        let new_order: Vec<ChangeId> = std::iter::once(change_id)
1961            .chain(entries[..last].iter().map(|(_, p)| *p))
1962            .collect();
1963        for (i, p) in new_order.iter().enumerate() {
1964            let pos = L64::from(new_pos + i as u64);
1965            m = m.next(self.get_external(p).optional()?.unwrap());
1966            btree::put(
1967                &mut self.txn,
1968                &mut channel.revchanges,
1969                &pos,
1970                &Pair { a: *p, b: m.into() },
1971            )?;
1972            btree::put(&mut self.txn, &mut channel.states, &m.into(), &pos)?;
1973            btree::put(&mut self.txn, &mut channel.changes, p, &pos)?;
1974        }
1975        let _ = old_l64;
1976        Ok(())
1977    }
1978}
1979
1980fn replay_tags<
1981    T: sanakirja::AllocPage<Error = ::sanakirja::Error>
1982        + sanakirja::RootPage
1983        + sanakirja::LoadPage<Error = ::sanakirja::Error>,
1984>(
1985    txn: &mut MutTxn<T>,
1986    channel: &mut Db<L64, Pair<SerializedMerkle, SerializedMerkle>>,
1987    tl: L64,
1988    repl: &mut Vec<(L64, SerializedMerkle)>,
1989) -> Result<(), TxnErr<SanakirjaError>> {
1990    let del = repl.is_empty();
1991    for x in btree::iter(&txn.txn, channel, Some((&tl, None)))? {
1992        let (t_, p) = x?;
1993        if *t_ >= tl {
1994            repl.push((*t_, p.a))
1995        }
1996    }
1997    let mut m = Merkle::zero();
1998    for x in btree::rev_iter(&txn.txn, channel, Some((&tl, None)))? {
1999        let (t_, mm) = x?;
2000        if t_ < &tl {
2001            m = (&mm.b).into();
2002            break;
2003        }
2004    }
2005    for (t_, p) in repl.iter() {
2006        debug!("del_tags {:?} {:?}", t_, p);
2007        btree::del(&mut txn.txn, channel, t_, None)?;
2008        if *t_ > tl || !del {
2009            m = m.next(p);
2010            btree::put(&mut txn.txn, channel, t_, &Pair { a: *p, b: m.into() })?;
2011        }
2012    }
2013    Ok(())
2014}
2015
2016impl<
2017    T: sanakirja::AllocPage<Error = ::sanakirja::Error>
2018        + sanakirja::RootPage
2019        + sanakirja::LoadPage<Error = ::sanakirja::Error>,
2020> DepsMutTxnT for MutTxn<T>
2021{
2022    sanakirja_put_del!(dep, ChangeId, ChangeId, DepsError);
2023    sanakirja_put_del!(revdep, ChangeId, ChangeId, DepsError);
2024    sanakirja_put_del!(touched_files, Position<ChangeId>, ChangeId, DepsError);
2025    sanakirja_put_del!(rev_touched_files, ChangeId, Position<ChangeId>, DepsError);
2026}
2027
2028impl<
2029    T: sanakirja::AllocPage<Error = ::sanakirja::Error>
2030        + sanakirja::RootPageMut
2031        + sanakirja::LoadPage<Error = ::sanakirja::Error>,
2032> TreeMutTxnT for MutTxn<T>
2033{
2034    sanakirja_put_del!(inodes, Inode, Position<ChangeId>, TreeError, TreeErr);
2035    sanakirja_put_del!(revinodes, Position<ChangeId>, Inode, TreeError, TreeErr);
2036
2037    sanakirja_put_del!(tree, PathId, Inode, TreeError, TreeErr);
2038    sanakirja_put_del!(revtree, Inode, PathId, TreeError, TreeErr);
2039
2040    fn put_partials(
2041        &mut self,
2042        k: &SmallStr,
2043        e: Position<ChangeId>,
2044    ) -> Result<bool, TreeErr<Self::TreeError>> {
2045        btree::put(&mut self.txn, &mut self.partials, k, &e).map_err(|e| TreeErr(e.into()))
2046    }
2047
2048    fn del_partials(
2049        &mut self,
2050        k: &SmallStr,
2051        e: Option<Position<ChangeId>>,
2052    ) -> Result<bool, TreeErr<Self::TreeError>> {
2053        btree::del(&mut self.txn, &mut self.partials, k, e.as_ref()).map_err(|e| TreeErr(e.into()))
2054    }
2055}
2056
2057impl<T: RawMutTxnT> MutTxnT for MutTxn<T> {
2058    fn put_remote(
2059        &mut self,
2060        remote: &mut RemoteRef<Self>,
2061        k: u64,
2062        v: (Hash, Merkle),
2063    ) -> Result<bool, TxnErr<Self::GraphError>> {
2064        let mut remote = remote.db.lock();
2065        let h = (&v.0).into();
2066        let m: SerializedMerkle = (&v.1).into();
2067        btree::put(
2068            &mut self.txn,
2069            &mut remote.remote,
2070            &k.into(),
2071            &Pair { a: h, b: m },
2072        )?;
2073        debug!("remote.remote after put: {:?}", remote.remote);
2074        btree::put(&mut self.txn, &mut remote.states, &m, &k.into())?;
2075        // if v.2 {
2076        //     self.put_tags(&mut remote.tags, k, &v.1)?;
2077        // }
2078        Ok(btree::put(&mut self.txn, &mut remote.rev, &h, &k.into())?)
2079    }
2080
2081    fn del_remote(
2082        &mut self,
2083        remote: &mut RemoteRef<Self>,
2084        k: u64,
2085    ) -> Result<bool, TxnErr<Self::GraphError>> {
2086        let mut remote = remote.db.lock();
2087        let k = k.into();
2088        match btree::get(&self.txn, &remote.remote, &k, None)? {
2089            Some((k0, p)) if k0 == &k => {
2090                debug!("del_remote {:?} {:?}", k0, p);
2091                let p = *p;
2092                btree::del(&mut self.txn, &mut remote.rev, &p.a, None)?;
2093                btree::del(&mut self.txn, &mut remote.states, &p.b, None)?;
2094                Ok(btree::del(&mut self.txn, &mut remote.remote, &k, None)?)
2095            }
2096            x => {
2097                debug!("not found, {:?}", x);
2098                Ok(false)
2099            }
2100        }
2101    }
2102
2103    fn open_or_create_channel(
2104        &mut self,
2105        name: &SmallStr,
2106    ) -> Result<ChannelRef<Self>, Self::GraphError> {
2107        unsafe {
2108            let mut commit = None;
2109            let result = match self.open_channels.lock().entry(name.to_owned()) {
2110                Entry::Vacant(v) => {
2111                    let r = match btree::get(&self.txn, &self.channels, name, None)? {
2112                        Some((name_, b)) if name_ == name => ChannelRef {
2113                            r: Arc::new(RwLock::new(Channel {
2114                                graph: Db::from_page(b.graph.into()),
2115                                changes: Db::from_page(b.changes.into()),
2116                                revchanges: UDb::from_page(b.revchanges.into()),
2117                                states: UDb::from_page(b.states.into()),
2118                                tags: Db::from_page(b.tags.into()),
2119                                apply_counter: b.apply_counter.into(),
2120                                last_modified: b.last_modified.into(),
2121                                id: b.id,
2122                                name: name.to_owned(),
2123                            })),
2124                        },
2125                        _ => {
2126                            let br = ChannelRef {
2127                                r: Arc::new(RwLock::new(Channel {
2128                                    graph: btree::create_db_(&mut self.txn)?,
2129                                    changes: btree::create_db_(&mut self.txn)?,
2130                                    revchanges: btree::create_db_(&mut self.txn)?,
2131                                    states: btree::create_db_(&mut self.txn)?,
2132                                    tags: btree::create_db_(&mut self.txn)?,
2133                                    id: {
2134                                        let mut rng = rand::rng();
2135                                        use rand::RngExt;
2136                                        let mut x = RemoteId([0; 16]);
2137                                        for x in x.0.iter_mut() {
2138                                            *x = rng.random()
2139                                        }
2140                                        x
2141                                    },
2142                                    apply_counter: 0,
2143                                    last_modified: 0,
2144                                    name: name.to_owned(),
2145                                })),
2146                            };
2147                            commit = Some(br.clone());
2148                            br
2149                        }
2150                    };
2151                    v.insert(r).clone()
2152                }
2153                Entry::Occupied(occ) => occ.get().clone(),
2154            };
2155            if let Some(commit) = commit {
2156                self.put_channel(commit)?;
2157            }
2158            Ok(result)
2159        }
2160    }
2161
2162    fn fork(
2163        &mut self,
2164        channel: &ChannelRef<Self>,
2165        name: &SmallStr,
2166    ) -> Result<ChannelRef<Self>, ForkError<Self::GraphError>> {
2167        let channel = channel.r.read();
2168        match btree::get(&self.txn, &self.channels, name, None)
2169            .map_err(|e| ForkError::Txn(e.into()))?
2170        {
2171            Some((name_, _)) if name_ == name => {
2172                Err(super::ForkError::ChannelNameExists(name.to_string()))
2173            }
2174            _ => {
2175                let br = ChannelRef {
2176                    r: Arc::new(RwLock::new(Channel {
2177                        graph: btree::fork_db(&mut self.txn, &channel.graph)
2178                            .map_err(|e| ForkError::Txn(e.into()))?,
2179                        changes: btree::fork_db(&mut self.txn, &channel.changes)
2180                            .map_err(|e| ForkError::Txn(e.into()))?,
2181                        revchanges: btree::fork_db(&mut self.txn, &channel.revchanges)
2182                            .map_err(|e| ForkError::Txn(e.into()))?,
2183                        states: btree::fork_db(&mut self.txn, &channel.states)
2184                            .map_err(|e| ForkError::Txn(e.into()))?,
2185                        tags: btree::fork_db(&mut self.txn, &channel.tags)
2186                            .map_err(|e| ForkError::Txn(e.into()))?,
2187                        name: name.to_owned(),
2188                        apply_counter: channel.apply_counter,
2189                        last_modified: channel.last_modified,
2190                        id: {
2191                            let mut rng = rand::rng();
2192                            use rand::RngExt;
2193                            let mut x = RemoteId([0; 16]);
2194                            for x in x.0.iter_mut() {
2195                                *x = rng.random()
2196                            }
2197                            x
2198                        },
2199                    })),
2200                };
2201                self.open_channels
2202                    .lock()
2203                    .insert(name.to_owned(), br.clone());
2204                Ok(br)
2205            }
2206        }
2207    }
2208
2209    fn rename_channel(
2210        &mut self,
2211        channel: &mut ChannelRef<Self>,
2212        name: &SmallStr,
2213    ) -> Result<(), ForkError<Self::GraphError>> {
2214        match btree::get(&self.txn, &self.channels, name, None)
2215            .map_err(|e| ForkError::Txn(e.into()))?
2216        {
2217            Some((name_, _)) if name_ == name => {
2218                Err(super::ForkError::ChannelNameExists(name.to_string()))
2219            }
2220            _ => {
2221                btree::del(
2222                    &mut self.txn,
2223                    &mut self.channels,
2224                    &channel.r.read().name,
2225                    None,
2226                )
2227                .map_err(|e| ForkError::Txn(e.into()))?;
2228                std::mem::drop(
2229                    self.open_channels
2230                        .lock()
2231                        .remove(&channel.r.read().name)
2232                        .unwrap(),
2233                );
2234                channel.r.write().name = name.to_owned();
2235                self.open_channels
2236                    .lock()
2237                    .insert(name.to_owned(), channel.clone());
2238                Ok(())
2239            }
2240        }
2241    }
2242
2243    fn drop_channel(&mut self, name: &SmallStr) -> Result<bool, Self::GraphError> {
2244        unsafe {
2245            let name = name.to_owned();
2246            debug!(target: "drop_channel", "drop channel {:?}", name);
2247            let channel = if let Some(channel) = self.open_channels.lock().remove(&name) {
2248                let channel = Arc::try_unwrap(channel.r)
2249                    .map_err(|_| SanakirjaError::ChannelRc {
2250                        c: name.to_string(),
2251                    })?
2252                    .into_inner();
2253                Some((
2254                    channel.graph,
2255                    channel.changes,
2256                    channel.revchanges,
2257                    channel.states,
2258                    channel.tags,
2259                ))
2260            } else if let Some((name_, chan)) = btree::get(&self.txn, &self.channels, &name, None)?
2261            {
2262                if name_ == name.as_ref() {
2263                    Some((
2264                        Db::from_page(chan.graph.into()),
2265                        Db::from_page(chan.changes.into()),
2266                        UDb::from_page(chan.revchanges.into()),
2267                        UDb::from_page(chan.states.into()),
2268                        Db::from_page(chan.tags.into()),
2269                    ))
2270                } else {
2271                    None
2272                }
2273            } else {
2274                None
2275            };
2276            btree::del(&mut self.txn, &mut self.channels, &name, None)?;
2277            if let Some((a, b, c, d, e)) = channel {
2278                let mut unused_changes = Vec::new();
2279                'outer: for x in btree::rev_iter(&self.txn, &c, None)? {
2280                    let (_, p) = x?;
2281                    debug!(target: "drop_channel", "testing unused change: {:?}", p);
2282                    let empty = SmallString::new();
2283                    for chan in self.channels(&empty).map_err(|e| e.0)? {
2284                        debug!(target: "drop_channel", "channel: {:?}", name);
2285                        let chan = chan.read();
2286                        assert_ne!(chan.name, name);
2287                        if self
2288                            .channel_has_state(&chan.states, &p.b)
2289                            .map_err(|e| e.0)?
2290                            .is_some()
2291                        {
2292                            // This other channel is in the same state as
2293                            // our dropped channel is, so all subsequent
2294                            // patches are in use.
2295                            break 'outer;
2296                        }
2297                        if self
2298                            .get_changeset(&chan.changes, &p.a)
2299                            .map_err(|e| e.0)?
2300                            .is_some()
2301                        {
2302                            // This channel has a patch, move on.
2303                            continue 'outer;
2304                        }
2305                    }
2306
2307                    debug!(target: "drop_channel", "actually unused: {:?}", p);
2308                    unused_changes.push(p.a);
2309                }
2310                let mut deps = Vec::new();
2311                for ch in unused_changes.iter() {
2312                    for x in btree::iter(&self.txn, &self.dep, Some((ch, None)))? {
2313                        let (k, v) = x?;
2314                        if k > ch {
2315                            break;
2316                        }
2317                        deps.push((*k, *v));
2318                    }
2319                    for (k, v) in deps.drain(..) {
2320                        debug!(target: "drop_channel", "deleting from revdep: {:?} {:?}", k, v);
2321                        btree::del(&mut self.txn, &mut self.dep, &k, Some(&v))?;
2322                        btree::del(&mut self.txn, &mut self.revdep, &v, Some(&k))?;
2323                    }
2324                }
2325                btree::drop(&mut self.txn, a)?;
2326                btree::drop(&mut self.txn, b)?;
2327                btree::drop(&mut self.txn, c)?;
2328                btree::drop(&mut self.txn, d)?;
2329                btree::drop(&mut self.txn, e)?;
2330                Ok(true)
2331            } else {
2332                Ok(false)
2333            }
2334        }
2335    }
2336
2337    fn open_or_create_remote(
2338        &mut self,
2339        id: RemoteId,
2340        path: &SmallStr,
2341    ) -> Result<RemoteRef<Self>, Self::GraphError> {
2342        unsafe {
2343            let mut commit = None;
2344            match self.open_remotes.lock().entry(id) {
2345                Entry::Vacant(v) => {
2346                    let r = match btree::get(&self.txn, &self.remotes, &id, None)? {
2347                        Some((name_, remote)) if *name_ == id => RemoteRef {
2348                            db: Arc::new(Mutex::new(Remote {
2349                                remote: UDb::from_page(remote.remote.into()),
2350                                rev: UDb::from_page(remote.rev.into()),
2351                                states: UDb::from_page(remote.states.into()),
2352                                id_rev: remote.id_rev,
2353                                tags: Db::from_page(remote.tags.into()),
2354                                path: path.to_owned(),
2355                            })),
2356                            id,
2357                        },
2358                        _ => {
2359                            let br = RemoteRef {
2360                                db: Arc::new(Mutex::new(Remote {
2361                                    remote: btree::create_db_(&mut self.txn)?,
2362                                    rev: btree::create_db_(&mut self.txn)?,
2363                                    states: btree::create_db_(&mut self.txn)?,
2364                                    id_rev: 0u64.into(),
2365                                    tags: btree::create_db(&mut self.txn)?,
2366                                    path: path.to_owned(),
2367                                })),
2368                                id,
2369                            };
2370                            commit = Some(br.clone());
2371                            br
2372                        }
2373                    };
2374                    v.insert(r);
2375                }
2376                Entry::Occupied(_) => {}
2377            }
2378            if let Some(commit) = commit {
2379                self.put_remotes(commit)?;
2380            }
2381            Ok(self.open_remotes.lock().get(&id).unwrap().clone())
2382        }
2383    }
2384
2385    fn drop_remote(&mut self, remote: RemoteRef<Self>) -> Result<bool, Self::GraphError> {
2386        let r = self.open_remotes.lock().remove(&remote.id).unwrap();
2387        std::mem::drop(remote);
2388        assert_eq!(Arc::strong_count(&r.db), 1);
2389        Ok(btree::del(&mut self.txn, &mut self.remotes, &r.id, None)?)
2390    }
2391
2392    fn drop_named_remote(&mut self, id: RemoteId) -> Result<bool, Self::GraphError> {
2393        if let Some(r) = self.open_remotes.lock().remove(&id) {
2394            assert_eq!(Arc::strong_count(&r.db), 1);
2395        }
2396        Ok(btree::del(&mut self.txn, &mut self.remotes, &id, None)?)
2397    }
2398
2399    fn commit(mut self) -> Result<(), Self::GraphError> {
2400        use std::ops::DerefMut;
2401        {
2402            let open_channels = std::mem::take(self.open_channels.lock().deref_mut());
2403            for (name, channel) in open_channels {
2404                debug!("commit_channel {:?}", name);
2405                self.commit_channel(channel)?
2406            }
2407        }
2408        {
2409            let open_remotes = std::mem::take(self.open_remotes.lock().deref_mut());
2410            for (name, remote) in open_remotes {
2411                debug!("commit remote {:?}", name);
2412                self.commit_remote(remote)?
2413            }
2414        }
2415        if let Some(ref cur) = self.cur_channel {
2416            unsafe {
2417                assert!(cur.len() < 256);
2418                let b = self.txn.root_page_mut();
2419                b[4096 - 256] = cur.len() as u8;
2420                std::ptr::copy(cur.as_ptr(), b.as_mut_ptr().add(4096 - 255), cur.len())
2421            }
2422        }
2423        // No need to set `Root::Version`, it is set at init.
2424        debug!(
2425            "{:x} {:x} {:x} {:x} {:x} {:x} {:x} {:x} {:x} {:x} {:x} {:x} {:x}",
2426            self.tree.db,
2427            self.revtree.db,
2428            self.inodes.db,
2429            self.revinodes.db,
2430            self.internal.db,
2431            self.external.db,
2432            self.revdep.db,
2433            self.channels.db,
2434            self.remotes.db,
2435            self.touched_files.db,
2436            self.dep.db,
2437            self.rev_touched_files.db,
2438            self.partials.db,
2439        );
2440        self.txn
2441            .set_root(Root::Tree as usize, u64::from(self.tree.db));
2442        self.txn
2443            .set_root(Root::RevTree as usize, u64::from(self.revtree.db));
2444        self.txn
2445            .set_root(Root::Inodes as usize, u64::from(self.inodes.db));
2446        self.txn
2447            .set_root(Root::RevInodes as usize, self.revinodes.db.into());
2448        self.txn
2449            .set_root(Root::Internal as usize, self.internal.db.into());
2450        self.txn
2451            .set_root(Root::External as usize, self.external.db.into());
2452        self.txn
2453            .set_root(Root::RevDep as usize, self.revdep.db.into());
2454        self.txn
2455            .set_root(Root::Channels as usize, self.channels.db.into());
2456        self.txn
2457            .set_root(Root::Remotes as usize, self.remotes.db.into());
2458        self.txn
2459            .set_root(Root::TouchedFiles as usize, self.touched_files.db.into());
2460        self.txn.set_root(Root::Dep as usize, self.dep.db.into());
2461        self.txn.set_root(
2462            Root::RevTouchedFiles as usize,
2463            self.rev_touched_files.db.into(),
2464        );
2465        self.txn
2466            .set_root(Root::Partials as usize, self.partials.db.into());
2467        self.txn.commit()?;
2468        Ok(())
2469    }
2470
2471    fn set_current_channel(&mut self, cur: &str) -> Result<(), Self::GraphError> {
2472        self.cur_channel = Some(cur.to_string());
2473        Ok(())
2474    }
2475}
2476
2477impl Txn {
2478    pub fn load_const_channel(&self, name: &SmallStr) -> Result<Option<Channel>, SanakirjaError> {
2479        unsafe {
2480            let name = name.to_owned();
2481            match btree::get(&self.txn, &self.channels, &name, None)? {
2482                Some((name_, c)) if name.as_ref() == name_ => {
2483                    debug!("load_const_channel = {:?} {:?}", name_, c);
2484                    Ok(Some(Channel {
2485                        graph: Db::from_page(c.graph.into()),
2486                        changes: Db::from_page(c.changes.into()),
2487                        revchanges: UDb::from_page(c.revchanges.into()),
2488                        states: UDb::from_page(c.states.into()),
2489                        tags: Db::from_page(c.tags.into()),
2490                        apply_counter: c.apply_counter.into(),
2491                        last_modified: c.last_modified.into(),
2492                        id: c.id,
2493                        name,
2494                    }))
2495                }
2496                _ => Ok(None),
2497            }
2498        }
2499    }
2500}
2501
2502impl<
2503    T: sanakirja::AllocPage<Error = ::sanakirja::Error>
2504        + sanakirja::RootPage
2505        + sanakirja::LoadPage<Error = ::sanakirja::Error>,
2506> MutTxn<T>
2507{
2508    fn put_channel(&mut self, channel: ChannelRef<Self>) -> Result<(), SanakirjaError> {
2509        debug!("Commit_channel.");
2510        let channel = channel.r.read();
2511        debug!("Commit_channel, dbs_channels = {:?}", self.channels);
2512        btree::del(&mut self.txn, &mut self.channels, &channel.name, None)?;
2513        debug!(
2514            "channels: {:x} {:x} {:x} {:x} {:x}",
2515            channel.graph.db,
2516            channel.changes.db,
2517            channel.revchanges.db,
2518            channel.states.db,
2519            channel.tags.db,
2520        );
2521        let sc = SerializedChannel {
2522            graph: u64::from(channel.graph.db).into(),
2523            changes: u64::from(channel.changes.db).into(),
2524            revchanges: u64::from(channel.revchanges.db).into(),
2525            states: u64::from(channel.states.db).into(),
2526            tags: u64::from(channel.tags.db).into(),
2527            apply_counter: channel.apply_counter.into(),
2528            last_modified: channel.last_modified.into(),
2529            id: channel.id,
2530        };
2531        btree::put(&mut self.txn, &mut self.channels, &channel.name, &sc)?;
2532        debug!("Commit_channel, self.channels = {:?}", self.channels);
2533        Ok(())
2534    }
2535
2536    fn commit_channel(&mut self, channel: ChannelRef<Self>) -> Result<(), SanakirjaError> {
2537        std::mem::drop(self.open_channels.lock().remove(&channel.r.read().name));
2538        self.put_channel(channel)
2539    }
2540
2541    fn put_remotes(&mut self, remote: RemoteRef<Self>) -> Result<(), SanakirjaError> {
2542        btree::del(&mut self.txn, &mut self.remotes, &remote.id, None)?;
2543        debug!("Commit_remote, dbs_remotes = {:?}", self.remotes);
2544        let r = remote.db.lock();
2545        let rr = OwnedSerializedRemote {
2546            _remote: u64::from(r.remote.db).into(),
2547            _rev: u64::from(r.rev.db).into(),
2548            _states: u64::from(r.states.db).into(),
2549            _id_rev: r.id_rev,
2550            _tags: u64::from(r.tags.db).into(),
2551            _path: r.path.clone(),
2552        };
2553        debug!("put {:?}", rr);
2554        btree::put(&mut self.txn, &mut self.remotes, &remote.id, &rr)?;
2555        debug!("Commit_remote, self.dbs.remotes = {:?}", self.remotes);
2556        Ok(())
2557    }
2558
2559    fn commit_remote(&mut self, remote: RemoteRef<Self>) -> Result<(), SanakirjaError> {
2560        std::mem::drop(self.open_remotes.lock().remove(&remote.id));
2561        // assert_eq!(Rc::strong_count(&remote.db), 1);
2562        self.put_remotes(remote)
2563    }
2564}
2565
2566direct_repr!(ChangeId);
2567impl ::sanakirja::debug::Check for ChangeId {}
2568
2569direct_repr!(Vertex<ChangeId>);
2570impl ::sanakirja::debug::Check for Vertex<ChangeId> {}
2571
2572direct_repr!(Position<ChangeId>);
2573impl ::sanakirja::debug::Check for Position<ChangeId> {}
2574
2575direct_repr!(SerializedEdge);
2576impl ::sanakirja::debug::Check for SerializedEdge {}
2577
2578impl ::sanakirja::debug::Check for PathId {}
2579impl Storable for PathId {
2580    fn compare<T>(&self, _: &T, x: &Self) -> std::cmp::Ordering {
2581        self.cmp(x)
2582    }
2583    type PageReferences = std::iter::Empty<u64>;
2584    fn page_references(&self) -> Self::PageReferences {
2585        std::iter::empty()
2586    }
2587}
2588impl UnsizedStorable for PathId {
2589    const ALIGN: usize = 8;
2590    fn size(&self) -> usize {
2591        9 + self.basename.len()
2592    }
2593    unsafe fn onpage_size(p: *const u8) -> usize {
2594        unsafe {
2595            let len = *(p.add(8)) as usize;
2596            9 + len
2597        }
2598    }
2599    unsafe fn from_raw_ptr<'a, T>(_: &T, p: *const u8) -> &'a Self {
2600        unsafe { path_id_from_raw_ptr(p) }
2601    }
2602    unsafe fn write_to_page(&self, p: *mut u8) {
2603        unsafe {
2604            *(p as *mut u64) = (self.parent_inode.0).0;
2605            self.basename.write_to_page(p.add(8))
2606        }
2607    }
2608}
2609
2610unsafe fn path_id_from_raw_ptr<'a>(p: *const u8) -> &'a PathId {
2611    unsafe {
2612        let len = *(p.add(8)) as usize;
2613        std::mem::transmute(std::slice::from_raw_parts(p, 1 + len))
2614    }
2615}
2616
2617#[test]
2618fn pathid_repr() {
2619    let o = OwnedPathId {
2620        parent_inode: Inode::ROOT,
2621        basename: SmallString::from_str("blablabla"),
2622    };
2623    let mut x = vec![0u8; 200];
2624
2625    unsafe {
2626        o.write_to_page(x.as_mut_ptr());
2627        let p = path_id_from_raw_ptr(x.as_ptr());
2628        assert_eq!(p.basename.as_str(), "blablabla");
2629        assert_eq!(p.parent_inode, Inode::ROOT);
2630    }
2631}
2632
2633direct_repr!(Inode);
2634impl ::sanakirja::debug::Check for Inode {}
2635direct_repr!(SerializedMerkle);
2636impl ::sanakirja::debug::Check for SerializedMerkle {}
2637direct_repr!(SerializedHash);
2638impl ::sanakirja::debug::Check for SerializedHash {}
2639
2640impl<A: ::sanakirja::debug::Check, B: ::sanakirja::debug::Check> ::sanakirja::debug::Check
2641    for Pair<A, B>
2642{
2643    fn add_refs<T: LoadPage>(
2644        &self,
2645        txn: &T,
2646        pages: &mut std::collections::BTreeMap<u64, usize>,
2647    ) -> Result<(), T::Error>
2648    where
2649        T::Error: std::fmt::Debug,
2650    {
2651        self.a.add_refs(txn, pages)?;
2652        self.b.add_refs(txn, pages)
2653    }
2654}
2655impl<A: Storable, B: Storable> Storable for Pair<A, B> {
2656    type PageReferences = core::iter::Chain<A::PageReferences, B::PageReferences>;
2657    fn page_references(&self) -> Self::PageReferences {
2658        self.a.page_references().chain(self.b.page_references())
2659    }
2660    fn compare<T: LoadPage>(&self, t: &T, b: &Self) -> core::cmp::Ordering {
2661        match self.a.compare(t, &b.a) {
2662            core::cmp::Ordering::Equal => self.b.compare(t, &b.b),
2663            ord => ord,
2664        }
2665    }
2666}
2667
2668impl<A: Ord + UnsizedStorable, B: Ord + UnsizedStorable> UnsizedStorable for Pair<A, B> {
2669    const ALIGN: usize = std::mem::align_of::<(A, B)>();
2670
2671    fn size(&self) -> usize {
2672        let a = self.a.size();
2673        let b_off = (a + (B::ALIGN - 1)) & !(B::ALIGN - 1);
2674        (b_off + self.b.size() + (Self::ALIGN - 1)) & !(Self::ALIGN - 1)
2675    }
2676    unsafe fn onpage_size(p: *const u8) -> usize {
2677        unsafe {
2678            let a = A::onpage_size(p);
2679            let b_off = (a + (B::ALIGN - 1)) & !(B::ALIGN - 1);
2680            let b_size = B::onpage_size(p.add(b_off));
2681            (b_off + b_size + (Self::ALIGN - 1)) & !(Self::ALIGN - 1)
2682        }
2683    }
2684    unsafe fn from_raw_ptr<'a, T>(_: &T, p: *const u8) -> &'a Self {
2685        unsafe { &*(p as *const Self) }
2686    }
2687    unsafe fn write_to_page_alloc<T: sanakirja::AllocPage>(&self, t: &mut T, p: *mut u8) {
2688        unsafe {
2689            self.a.write_to_page_alloc(t, p);
2690            let off = (self.a.size() + (B::ALIGN - 1)) & !(B::ALIGN - 1);
2691            self.b.write_to_page_alloc(t, p.add(off));
2692        }
2693    }
2694}
2695
2696impl ::sanakirja::debug::Check for SerializedRemote {}
2697impl Storable for SerializedRemote {
2698    type PageReferences = std::iter::Empty<u64>;
2699    fn page_references(&self) -> Self::PageReferences {
2700        std::iter::empty()
2701    }
2702    fn compare<T: LoadPage>(&self, _t: &T, b: &Self) -> core::cmp::Ordering {
2703        self.cmp(b)
2704    }
2705}
2706
2707const REMOTE_LEN: usize = 40;
2708
2709impl UnsizedStorable for SerializedRemote {
2710    const ALIGN: usize = 8;
2711
2712    fn size(&self) -> usize {
2713        REMOTE_LEN + 1 + self.path.len()
2714    }
2715    unsafe fn onpage_size(p: *const u8) -> usize {
2716        unsafe { REMOTE_LEN + 1 + (*p.add(REMOTE_LEN)) as usize }
2717    }
2718    unsafe fn from_raw_ptr<'a, T>(_: &T, p: *const u8) -> &'a Self {
2719        unsafe {
2720            let len = *p.add(REMOTE_LEN) as usize;
2721            let m: &SerializedRemote = std::mem::transmute(std::slice::from_raw_parts(p, 1 + len));
2722            m
2723        }
2724    }
2725    unsafe fn write_to_page_alloc<T: sanakirja::AllocPage>(&self, _: &mut T, p: *mut u8) {
2726        unsafe {
2727            std::ptr::copy(
2728                &self.remote as *const L64 as *const u8,
2729                p,
2730                REMOTE_LEN + 1 + self.path.len(),
2731            );
2732            debug!(
2733                "write_to_page: {:?}",
2734                std::slice::from_raw_parts(p, REMOTE_LEN + 1 + self.path.len())
2735            );
2736        }
2737    }
2738}
2739
2740#[derive(Debug)]
2741#[repr(C)]
2742struct OwnedSerializedRemote {
2743    _remote: L64,
2744    _rev: L64,
2745    _states: L64,
2746    _id_rev: L64,
2747    _tags: L64,
2748    _path: SmallString,
2749}
2750
2751impl std::ops::Deref for OwnedSerializedRemote {
2752    type Target = SerializedRemote;
2753    fn deref(&self) -> &Self::Target {
2754        let len = REMOTE_LEN + 1 + self._path.len();
2755        unsafe {
2756            std::mem::transmute(std::slice::from_raw_parts(
2757                self as *const Self as *const u8,
2758                len,
2759            ))
2760        }
2761    }
2762}
2763
2764direct_repr!(SerializedChannel);
2765impl ::sanakirja::debug::Check for SerializedChannel {}
2766
2767direct_repr!(RemoteId);
2768impl ::sanakirja::debug::Check for RemoteId {}