1use std::path::Path;
77use std::sync::Arc;
78
79use redb::ReadableTable;
80
81use crate::error::{Error, Result};
82
83const DOCUMENTS: redb::TableDefinition<'static, &str, &[u8]> =
85 redb::TableDefinition::new("documents");
86
87const BLOBS: redb::TableDefinition<'static, &[u8], &[u8]> = redb::TableDefinition::new("blobs");
90
91const DOC_HISTORY: redb::TableDefinition<'static, (&str, u64), &[u8]> =
99 redb::TableDefinition::new("document_history");
100
101const MARKS: redb::TableDefinition<'static, &str, u64> = redb::TableDefinition::new("box_marks");
104
105const PRINCIPAL_ACTIVITY: redb::TableDefinition<'static, &str, &[u8]> =
111 redb::TableDefinition::new("principal_activity");
112
113pub const TWIN_APPLIED: &str = "twin.applied";
120
121pub(crate) enum Txn<T> {
126 Commit(T),
128 CommitUnsynced(T),
131 Discard(T),
133}
134
135struct Group {
139 owner: std::thread::ThreadId,
140 txn: redb::WriteTransaction,
141 dirty: bool,
143 poisoned: bool,
145}
146
147pub const BLOB_ID_LEN: usize = 32;
149
150#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
158pub struct BlobId([u8; BLOB_ID_LEN]);
159
160impl BlobId {
161 pub fn of(bytes: &[u8]) -> Self {
163 use sha2::Digest;
164 let mut hasher = sha2::Sha256::new();
165 hasher.update(bytes);
166 BlobId(hasher.finalize().into())
167 }
168
169 pub fn as_bytes(&self) -> &[u8; BLOB_ID_LEN] {
171 &self.0
172 }
173
174 pub fn from_hex(text: &str) -> Option<Self> {
176 let mut out = [0u8; BLOB_ID_LEN];
177 hex::decode_to_slice(text, &mut out).ok()?;
178 Some(BlobId(out))
179 }
180
181 fn from_slice(bytes: &[u8]) -> Option<Self> {
183 Some(BlobId(bytes.try_into().ok()?))
184 }
185}
186
187impl std::fmt::Display for BlobId {
188 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
189 f.write_str(&hex::encode(self.0))
190 }
191}
192
193impl std::fmt::Debug for BlobId {
194 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
195 write!(f, "BlobId({self})")
196 }
197}
198
199#[derive(Clone, PartialEq, Eq, Debug)]
201pub struct DocVersion {
202 pub seq: u64,
204 pub at: crate::model::UnixMillis,
206 pub blob: BlobId,
209}
210
211pub trait ControlStore: Send + Sync + 'static {
221 fn put_blob(&self, bytes: &[u8]) -> Result<BlobId>;
224
225 fn put_blob_unsynced(&self, bytes: &[u8]) -> Result<BlobId> {
236 self.put_blob(bytes)
237 }
238
239 fn get_blob(&self, id: &BlobId) -> Result<Option<Vec<u8>>>;
241
242 fn read_doc(&self, name: &str) -> Result<Option<BlobId>>;
249
250 fn commit_doc(&self, name: &str, expected: Option<BlobId>, new: BlobId) -> Result<bool>;
259
260 fn history(&self, name: &str) -> Result<Vec<DocVersion>>;
262}
263
264#[derive(Debug, Clone)]
273pub struct RamCopy {
274 pub why: String,
276 docs: std::collections::BTreeMap<String, BlobId>,
277 blobs: std::collections::BTreeMap<BlobId, Vec<u8>>,
278 secrets: std::collections::BTreeMap<String, Vec<u8>>,
279 pub taken_unix_ms: crate::model::UnixMillis,
281}
282
283impl RamCopy {
284 pub fn bytes(&self) -> u64 {
286 self.blobs.values().map(|b| b.len() as u64).sum::<u64>()
287 + self.secrets.values().map(|b| b.len() as u64).sum::<u64>()
288 }
289 pub fn documents(&self) -> usize {
291 self.docs.len()
292 }
293 pub fn pointers(&self) -> Vec<(String, String)> {
296 self.docs
297 .iter()
298 .map(|(k, v)| (k.clone(), v.to_string()))
299 .collect()
300 }
301 pub(crate) fn secret_row(&self, key: &str) -> Option<&[u8]> {
303 self.secrets.get(key).map(|v| v.as_slice())
304 }
305}
306
307#[derive(Debug, Clone, Default)]
311pub struct ControlExport {
312 pub docs: std::collections::BTreeMap<String, BlobId>,
314 pub blobs: std::collections::BTreeMap<BlobId, Vec<u8>>,
316 pub secrets: std::collections::BTreeMap<String, Option<Vec<u8>>>,
318}
319
320impl ControlDb {
321 pub fn export(&self) -> Result<ControlExport> {
324 let raw = self.read(
325 "exporting the control database",
326 |txn| {
327 let mut docs = std::collections::BTreeMap::new();
328 let mut blobs = std::collections::BTreeMap::new();
329 let table = txn
330 .open_table(DOCUMENTS)
331 .map_err(|err| faulted("opening the documents table", err))?;
332 for entry in table
333 .iter()
334 .map_err(|err| faulted("scanning the documents table", err))?
335 {
336 let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
337 if let Some(id) = BlobId::from_slice(v.value()) {
338 docs.insert(k.value().to_owned(), id);
339 }
340 }
341 let table = txn
342 .open_table(BLOBS)
343 .map_err(|err| faulted("opening the blobs table", err))?;
344 for entry in table
345 .iter()
346 .map_err(|err| faulted("scanning the blobs table", err))?
347 {
348 let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
349 if let Some(id) = BlobId::from_slice(k.value()) {
350 blobs.insert(id, v.value().to_vec());
351 }
352 }
353 Ok((docs, blobs, crate::secrets::RedbSecretStore::rows(txn)?))
354 },
355 |ram| Ok((ram.docs.clone(), ram.blobs.clone(), ram.secrets.clone())),
356 )?;
357 let (docs, blobs, rows) = raw;
358 let mut secrets = std::collections::BTreeMap::new();
359 for (key, row) in rows {
360 let value = crate::secrets::RedbSecretStore::decode_row_of(&key, &row)
361 .map_err(|err| Error::ControlStore(err.to_string()))?;
362 secrets.insert(key, value);
363 }
364 Ok(ControlExport {
365 docs,
366 blobs,
367 secrets,
368 })
369 }
370}
371
372#[derive(Debug, Clone, Copy)]
375pub enum ExportRow<'a> {
376 Blob(BlobId, &'a [u8]),
378 Doc(&'a str, BlobId),
380 Secret(&'a str, Option<&'a [u8]>),
382}
383
384impl ControlDb {
385 pub fn export_each<E>(
390 &self,
391 row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
392 ) -> Result<std::result::Result<(), E>> {
393 let row = std::cell::RefCell::new(row);
394 self.read(
395 "exporting the control database row by row",
396 |txn| export_txn(txn, &mut **row.borrow_mut()),
397 |ram| export_ram(ram, &mut **row.borrow_mut()),
398 )
399 }
400
401 pub fn view(&self) -> Result<ControlView> {
409 if let Some(copy) = read_lock(&self.ram).as_ref() {
410 return Ok(ControlView(View::Ram(Box::new(copy.clone()))));
411 }
412 let guard = read_lock(&self.live);
413 let Some(db) = guard.as_ref() else {
414 return Err(self.off_stripe());
415 };
416 let txn = db
417 .begin_read()
418 .map_err(|err| faulted("beginning a view of the control database", err))?;
419 Ok(ControlView(View::Live(Box::new(txn))))
420 }
421}
422
423pub struct ControlView(View);
425
426enum View {
427 Live(Box<redb::ReadTransaction>),
428 Ram(Box<RamCopy>),
429}
430
431impl ControlView {
432 pub fn principal_activity(&self) -> Result<Vec<(String, Vec<u8>)>> {
436 match &self.0 {
437 View::Live(txn) => activity_rows(txn),
438 View::Ram(_) => Ok(Vec::new()),
439 }
440 }
441
442 pub fn export_each<E>(
444 &self,
445 row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
446 ) -> Result<std::result::Result<(), E>> {
447 match &self.0 {
448 View::Live(txn) => export_txn(txn, row),
449 View::Ram(ram) => export_ram(ram, row),
450 }
451 }
452}
453
454fn activity_rows(txn: &redb::ReadTransaction) -> Result<Vec<(String, Vec<u8>)>> {
456 let table = match txn.open_table(PRINCIPAL_ACTIVITY) {
457 Ok(t) => t,
458 Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
459 Err(err) => return Err(faulted("opening the last-seen table", err)),
460 };
461 let mut rows = Vec::new();
462 for entry in table
463 .iter()
464 .map_err(|err| faulted("scanning the last-seen table", err))?
465 {
466 let (k, v) = entry.map_err(|err| faulted("reading a last-seen row", err))?;
467 rows.push((k.value().to_owned(), v.value().to_vec()));
468 }
469 Ok(rows)
470}
471
472fn export_txn<E>(
473 txn: &redb::ReadTransaction,
474 row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
475) -> Result<std::result::Result<(), E>> {
476 let table = txn
477 .open_table(BLOBS)
478 .map_err(|err| faulted("opening the blobs table", err))?;
479 for entry in table
480 .iter()
481 .map_err(|err| faulted("scanning the blobs table", err))?
482 {
483 let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
484 if let Some(id) = BlobId::from_slice(k.value()) {
485 if let Err(e) = row(ExportRow::Blob(id, v.value())) {
486 return Ok(Err(e));
487 }
488 }
489 }
490 let table = txn
491 .open_table(DOCUMENTS)
492 .map_err(|err| faulted("opening the documents table", err))?;
493 for entry in table
494 .iter()
495 .map_err(|err| faulted("scanning the documents table", err))?
496 {
497 let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
498 if let Some(id) = BlobId::from_slice(v.value()) {
499 if let Err(e) = row(ExportRow::Doc(k.value(), id)) {
500 return Ok(Err(e));
501 }
502 }
503 }
504 for (key, raw) in crate::secrets::RedbSecretStore::rows(txn)? {
505 let value = crate::secrets::RedbSecretStore::decode_row_of(&key, &raw)
506 .map_err(|err| Error::ControlStore(err.to_string()))?;
507 if let Err(e) = row(ExportRow::Secret(&key, value.as_deref())) {
508 return Ok(Err(e));
509 }
510 }
511 Ok(Ok(()))
512}
513
514fn export_ram<E>(
515 ram: &RamCopy,
516 row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
517) -> Result<std::result::Result<(), E>> {
518 for (id, bytes) in &ram.blobs {
519 if let Err(e) = row(ExportRow::Blob(*id, bytes)) {
520 return Ok(Err(e));
521 }
522 }
523 for (name, id) in &ram.docs {
524 if let Err(e) = row(ExportRow::Doc(name, *id)) {
525 return Ok(Err(e));
526 }
527 }
528 for (key, raw) in &ram.secrets {
529 let value = crate::secrets::RedbSecretStore::decode_row_of(key, raw)
530 .map_err(|err| Error::ControlStore(err.to_string()))?;
531 if let Err(e) = row(ExportRow::Secret(key, value.as_deref())) {
532 return Ok(Err(e));
533 }
534 }
535 Ok(Ok(()))
536}
537
538pub const RAM_COPY_CEILING_BYTES: u64 = 512 << 20;
542
543#[derive(Debug, Clone, PartialEq, Eq)]
545pub struct ResumeReport {
546 pub documents_matched: usize,
548 pub drift: Vec<String>,
552 pub off_stripe_ms: i64,
554}
555
556pub struct ControlDb {
582 path: Option<std::path::PathBuf>,
583 live: std::sync::RwLock<Option<redb::Database>>,
584 ram: std::sync::RwLock<Option<RamCopy>>,
585 group: std::sync::Mutex<Option<Group>>,
587 durable_commits: std::sync::atomic::AtomicU64,
589}
590
591impl std::fmt::Debug for ControlDb {
592 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
593 f.debug_struct("ControlDb")
594 .field("path", &self.path)
595 .field("suspended", &self.is_suspended())
596 .finish()
597 }
598}
599
600fn read_lock<T>(lock: &std::sync::RwLock<T>) -> std::sync::RwLockReadGuard<'_, T> {
601 lock.read().unwrap_or_else(|poison| poison.into_inner())
602}
603
604fn write_lock<T>(lock: &std::sync::RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
605 lock.write().unwrap_or_else(|poison| poison.into_inner())
606}
607
608impl ControlDb {
609 fn new(path: Option<std::path::PathBuf>, db: redb::Database) -> Self {
610 ControlDb {
611 path,
612 live: std::sync::RwLock::new(Some(db)),
613 ram: std::sync::RwLock::new(None),
614 group: std::sync::Mutex::new(None),
615 durable_commits: std::sync::atomic::AtomicU64::new(0),
616 }
617 }
618
619 #[cfg(test)]
622 pub(crate) fn in_memory_for_tests(db: redb::Database) -> Self {
623 Self::new(None, db)
624 }
625
626 pub fn is_suspended(&self) -> bool {
628 read_lock(&self.ram).is_some()
629 }
630
631 pub fn ram_copy(&self) -> Option<RamCopy> {
633 read_lock(&self.ram).clone()
634 }
635
636 fn off_stripe(&self) -> Error {
638 Error::OffStripe {
639 why: read_lock(&self.ram)
640 .as_ref()
641 .map(|r| r.why.clone())
642 .unwrap_or_else(|| "suspended".to_owned()),
643 }
644 }
645
646 pub(crate) fn read<T>(
650 &self,
651 what: &str,
652 live: impl FnOnce(&redb::ReadTransaction) -> Result<T>,
653 ram: impl FnOnce(&RamCopy) -> Result<T>,
654 ) -> Result<T> {
655 if let Some(copy) = read_lock(&self.ram).as_ref() {
656 return ram(copy);
657 }
658 let guard = read_lock(&self.live);
659 let Some(db) = guard.as_ref() else {
660 return Err(self.off_stripe());
661 };
662 let txn = db
663 .begin_read()
664 .map_err(|err| faulted(&format!("beginning {what}"), err))?;
665 live(&txn)
666 }
667
668 pub(crate) fn write<T>(
673 &self,
674 what: &str,
675 f: impl FnOnce(&redb::WriteTransaction) -> Result<Txn<T>>,
676 ) -> Result<T> {
677 if self.is_suspended() {
678 return Err(self.off_stripe());
679 }
680 {
681 let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
682 if let Some(g) = group
683 .as_mut()
684 .filter(|g| g.owner == std::thread::current().id())
685 {
686 return match f(&g.txn) {
687 Ok(Txn::Commit(t) | Txn::CommitUnsynced(t)) => {
688 g.dirty = true;
689 Ok(t)
690 }
691 Ok(Txn::Discard(t)) => Ok(t),
692 Err(err) => {
693 g.poisoned = true;
694 Err(err)
695 }
696 };
697 }
698 }
699 let guard = read_lock(&self.live);
700 let Some(db) = guard.as_ref() else {
701 return Err(self.off_stripe());
702 };
703 let mut txn = db
704 .begin_write()
705 .map_err(|err| faulted(&format!("beginning {what}"), err))?;
706 match f(&txn)? {
707 Txn::CommitUnsynced(t) => {
708 txn.set_durability(redb::Durability::None);
709 txn.commit()
710 .map_err(|err| faulted(&format!("committing {what}"), err))?;
711 Ok(t)
712 }
713 Txn::Commit(t) => {
714 txn.commit()
715 .map_err(|err| faulted(&format!("committing {what}"), err))?;
716 self.durable_commits
717 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
718 Ok(t)
719 }
720 Txn::Discard(t) => Ok(t),
722 }
723 }
724
725 pub fn begin_group(&self) -> Result<()> {
731 if self.is_suspended() {
732 return Err(self.off_stripe());
733 }
734 let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
735 if group.is_some() {
736 return Err(Error::ControlStore(
737 "a group transaction is already open on this control store".to_owned(),
738 ));
739 }
740 let guard = read_lock(&self.live);
741 let Some(db) = guard.as_ref() else {
742 return Err(self.off_stripe());
743 };
744 let txn = db
745 .begin_write()
746 .map_err(|err| faulted("beginning a group of writes", err))?;
747 *group = Some(Group {
748 owner: std::thread::current().id(),
749 txn,
750 dirty: false,
751 poisoned: false,
752 });
753 Ok(())
754 }
755
756 pub fn commit_group(&self, key: &str, seq: u64, mark: bool) -> Result<bool> {
761 let g = self.take_group()?;
762 if g.poisoned {
763 drop(g.txn);
764 return Err(Error::ControlStore(
765 "a write inside the group failed; the whole group was rolled back".to_owned(),
766 ));
767 }
768 if !g.dirty && !mark {
769 return Ok(false);
770 }
771 {
772 let mut marks = g
773 .txn
774 .open_table(MARKS)
775 .map_err(|err| faulted("opening the marks table", err))?;
776 marks
777 .insert(key, seq)
778 .map_err(|err| faulted("moving a marker", err))?;
779 }
780 g.txn
781 .commit()
782 .map_err(|err| faulted("committing a group of writes", err))?;
783 self.durable_commits
784 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
785 Ok(true)
786 }
787
788 pub fn durable_commits(&self) -> u64 {
791 self.durable_commits
792 .load(std::sync::atomic::Ordering::Relaxed)
793 }
794
795 pub fn abort_group(&self) {
797 let _ = self.take_group();
798 }
799
800 fn take_group(&self) -> Result<Group> {
801 let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
802 match group.take() {
803 Some(g) if g.owner == std::thread::current().id() => Ok(g),
804 Some(g) => {
805 *group = Some(g);
806 Err(Error::ControlStore(
807 "the open group transaction belongs to another thread".to_owned(),
808 ))
809 }
810 None => Err(Error::ControlStore(
811 "no group transaction is open".to_owned(),
812 )),
813 }
814 }
815
816 pub fn set_mark(&self, key: &str, seq: u64) -> Result<()> {
820 self.write("a marker", |txn| {
821 let mut marks = txn
822 .open_table(MARKS)
823 .map_err(|err| faulted("opening the marks table", err))?;
824 marks
825 .insert(key, seq)
826 .map_err(|err| faulted("moving a marker", err))?;
827 Ok(Txn::Commit(()))
828 })
829 }
830
831 pub fn mark(&self, key: &str) -> Result<Option<u64>> {
833 self.read(
834 "a marker read",
835 |txn| {
836 let table = match txn.open_table(MARKS) {
837 Ok(t) => t,
838 Err(redb::TableError::TableDoesNotExist(_)) => return Ok(None),
839 Err(err) => return Err(faulted("opening the marks table", err)),
840 };
841 let found = table
842 .get(key)
843 .map_err(|err| faulted("reading a marker", err))?;
844 Ok(found.map(|v| v.value()))
845 },
846 |_ram| Ok(None),
847 )
848 }
849
850 pub fn principal_activity(&self) -> Result<Vec<(String, Vec<u8>)>> {
854 self.read("a last-seen read", activity_rows, |_ram| Ok(Vec::new()))
855 }
856
857 pub fn put_principal_activity(
862 &self,
863 rows: &[(String, Vec<u8>)],
864 mark: Option<&str>,
865 ) -> Result<()> {
866 self.write("last-seen rows", |txn| {
867 {
868 let mut table = txn
869 .open_table(PRINCIPAL_ACTIVITY)
870 .map_err(|err| faulted("opening the last-seen table", err))?;
871 for (id, value) in rows {
872 table
873 .insert(id.as_str(), value.as_slice())
874 .map_err(|err| faulted("writing a last-seen row", err))?;
875 }
876 }
877 if let Some(key) = mark {
878 let mut marks = txn
879 .open_table(MARKS)
880 .map_err(|err| faulted("opening the marks table", err))?;
881 marks
882 .insert(key, 1u64)
883 .map_err(|err| faulted("moving a marker", err))?;
884 }
885 Ok(Txn::Commit(()))
886 })
887 }
888
889 pub fn suspend(&self, why: &str) -> Result<RamCopy> {
894 if self
895 .group
896 .lock()
897 .unwrap_or_else(|p| p.into_inner())
898 .is_some()
899 {
900 return Err(Error::ControlStore(
901 "a group of writes is open on the control store; it is suspended after the group ends".to_owned(),
902 ));
903 }
904 let mut live = write_lock(&self.live);
905 if self.is_suspended() {
906 return Err(Error::ControlStore(
907 "the control store is already suspended".to_owned(),
908 ));
909 }
910 let Some(db) = live.as_ref() else {
911 return Err(Error::ControlStore(
912 "the control store has no database to suspend".to_owned(),
913 ));
914 };
915 let txn = db
916 .begin_read()
917 .map_err(|err| faulted("beginning the suspend snapshot", err))?;
918 let mut copy = RamCopy {
919 why: why.to_owned(),
920 docs: Default::default(),
921 blobs: Default::default(),
922 secrets: Default::default(),
923 taken_unix_ms: crate::model::now_ms(),
924 };
925 {
926 let table = txn
927 .open_table(DOCUMENTS)
928 .map_err(|err| faulted("opening the documents table", err))?;
929 for entry in table
930 .iter()
931 .map_err(|err| faulted("scanning the documents table", err))?
932 {
933 let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
934 if let Some(id) = BlobId::from_slice(v.value()) {
935 copy.docs.insert(k.value().to_owned(), id);
936 }
937 }
938 let table = txn
939 .open_table(BLOBS)
940 .map_err(|err| faulted("opening the blobs table", err))?;
941 for entry in table
942 .iter()
943 .map_err(|err| faulted("scanning the blobs table", err))?
944 {
945 let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
946 if let Some(id) = BlobId::from_slice(k.value()) {
947 copy.blobs.insert(id, v.value().to_vec());
948 }
949 }
950 copy.secrets = crate::secrets::RedbSecretStore::rows(&txn)?;
951 }
952 drop(txn);
953 let bytes = copy.bytes();
954 if bytes > RAM_COPY_CEILING_BYTES {
955 return Err(Error::ControlStore(format!(
956 "the control database would take {bytes} bytes of RAM to hold off the stripe, \
957 over the {RAM_COPY_CEILING_BYTES}-byte ceiling; not suspending"
958 )));
959 }
960 *write_lock(&self.ram) = Some(copy.clone());
965 if self.path.is_some() {
966 *live = None;
971 }
972 Ok(copy)
973 }
974
975 pub fn resume(&self) -> Result<ResumeReport> {
980 let mut live = write_lock(&self.live);
981 let Some(copy) = read_lock(&self.ram).clone() else {
982 return Err(Error::ControlStore(
983 "the control store is not suspended".to_owned(),
984 ));
985 };
986 if live.is_none() {
987 let path = self.path.as_ref().ok_or_else(|| {
988 Error::ControlStore("no path to reopen the control store from".to_owned())
989 })?;
990 let db = redb::Database::open(path).map_err(|err| {
991 faulted(
992 &format!("reopening the control store {}", path.display()),
993 err,
994 )
995 })?;
996 *live = Some(db);
997 }
998 let db = live.as_ref().expect("installed above");
999 let txn = db
1000 .begin_read()
1001 .map_err(|err| faulted("beginning the resume check", err))?;
1002 let table = txn
1003 .open_table(DOCUMENTS)
1004 .map_err(|err| faulted("opening the documents table", err))?;
1005 let mut matched = 0;
1006 let mut drift = Vec::new();
1007 for (name, ram_id) in ©.docs {
1008 let disk = table
1009 .get(name.as_str())
1010 .map_err(|err| faulted("reading a document pointer", err))?
1011 .and_then(|v| BlobId::from_slice(v.value()));
1012 match disk {
1013 Some(id) if id == *ram_id => matched += 1,
1014 other => drift.push(format!(
1015 "{name}: ram {ram_id} vs disk {}",
1016 other
1017 .map(|i| i.to_string())
1018 .unwrap_or_else(|| "absent".to_owned())
1019 )),
1020 }
1021 }
1022 for entry in table
1023 .iter()
1024 .map_err(|err| faulted("scanning the documents table", err))?
1025 {
1026 let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
1027 if !copy.docs.contains_key(k.value()) {
1028 drift.push(format!(
1029 "{}: ram absent vs disk {}",
1030 k.value(),
1031 BlobId::from_slice(v.value())
1032 .map(|i| i.to_string())
1033 .unwrap_or_else(|| "?".to_owned())
1034 ));
1035 }
1036 }
1037 drop(table);
1038 drop(txn);
1039 *write_lock(&self.ram) = None;
1040 Ok(ResumeReport {
1041 documents_matched: matched,
1042 drift,
1043 off_stripe_ms: crate::model::now_ms() - copy.taken_unix_ms,
1044 })
1045 }
1046}
1047
1048#[derive(Clone)]
1056pub struct RedbControlStore {
1057 db: Arc<ControlDb>,
1058}
1059
1060impl std::fmt::Debug for RedbControlStore {
1061 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1062 f.write_str("RedbControlStore")
1063 }
1064}
1065
1066fn faulted(what: &str, err: impl std::fmt::Display) -> Error {
1067 Error::ControlStore(format!("{what}: {err}"))
1068}
1069
1070impl RedbControlStore {
1071 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1073 let path = path.as_ref();
1074 let db = redb::Database::create(path).map_err(|err| {
1075 faulted(
1076 &format!("opening the control store {}", path.display()),
1077 err,
1078 )
1079 })?;
1080 Self::prepared(Some(path.to_path_buf()), db)
1081 }
1082
1083 pub fn control_db(&self) -> Arc<ControlDb> {
1086 Arc::clone(&self.db)
1087 }
1088
1089 pub fn in_memory() -> Result<Self> {
1092 let db = redb::Database::builder()
1093 .create_with_backend(redb::backends::InMemoryBackend::new())
1094 .map_err(|err| faulted("opening an in-memory control store", err))?;
1095 Self::prepared(None, db)
1096 }
1097
1098 fn prepared(path: Option<std::path::PathBuf>, db: redb::Database) -> Result<Self> {
1105 let txn = db
1106 .begin_write()
1107 .map_err(|err| faulted("beginning the control store's first transaction", err))?;
1108 {
1109 txn.open_table(DOCUMENTS)
1110 .map_err(|err| faulted("creating the documents table", err))?;
1111 txn.open_table(BLOBS)
1112 .map_err(|err| faulted("creating the blobs table", err))?;
1113 txn.open_table(DOC_HISTORY)
1114 .map_err(|err| faulted("creating the document history table", err))?;
1115 txn.open_table(MARKS)
1116 .map_err(|err| faulted("creating the marks table", err))?;
1117 }
1118 txn.commit()
1119 .map_err(|err| faulted("committing the control store's first transaction", err))?;
1120 crate::secrets::RedbSecretStore::prepare(&db)?;
1121 Ok(RedbControlStore {
1122 db: Arc::new(ControlDb::new(path, db)),
1123 })
1124 }
1125
1126 pub fn secrets(&self) -> crate::secrets::RedbSecretStore {
1141 crate::secrets::RedbSecretStore::new(Arc::clone(&self.db))
1142 }
1143
1144 fn next_seq(
1147 table: &impl redb::ReadableTable<(&'static str, u64), &'static [u8]>,
1148 name: &str,
1149 ) -> Result<u64> {
1150 let mut range = table
1151 .range((name, 0u64)..=(name, u64::MAX))
1152 .map_err(|err| faulted("reading a document's history", err))?;
1153 match range.next_back() {
1154 None => Ok(1),
1155 Some(entry) => {
1156 let (key, _) = entry.map_err(|err| faulted("reading a history row", err))?;
1157 Ok(key.value().1 + 1)
1158 }
1159 }
1160 }
1161}
1162
1163impl ControlStore for RedbControlStore {
1164 fn put_blob(&self, bytes: &[u8]) -> Result<BlobId> {
1165 let id = BlobId::of(bytes);
1166 self.db.write("a blob write", |txn| {
1167 {
1168 let mut table = txn
1169 .open_table(BLOBS)
1170 .map_err(|err| faulted("opening the blobs table", err))?;
1171 table
1172 .insert(id.0.as_slice(), bytes)
1173 .map_err(|err| faulted("storing a control blob", err))?;
1174 }
1175 Ok(Txn::Commit(id))
1176 })
1177 }
1178
1179 fn put_blob_unsynced(&self, bytes: &[u8]) -> Result<BlobId> {
1180 let id = BlobId::of(bytes);
1181 self.db.write("a blob write", |txn| {
1182 {
1183 let mut table = txn
1184 .open_table(BLOBS)
1185 .map_err(|err| faulted("opening the blobs table", err))?;
1186 table
1187 .insert(id.0.as_slice(), bytes)
1188 .map_err(|err| faulted("storing a control blob", err))?;
1189 }
1190 Ok(Txn::CommitUnsynced(id))
1191 })
1192 }
1193
1194 fn get_blob(&self, id: &BlobId) -> Result<Option<Vec<u8>>> {
1195 self.db.read(
1196 "a blob read",
1197 |txn| {
1198 let table = txn
1199 .open_table(BLOBS)
1200 .map_err(|err| faulted("opening the blobs table", err))?;
1201 let found = table
1202 .get(id.0.as_slice())
1203 .map_err(|err| faulted("reading a control blob", err))?;
1204 Ok(found.map(|value| value.value().to_vec()))
1205 },
1206 |ram| Ok(ram.blobs.get(id).cloned()),
1207 )
1208 }
1209
1210 fn read_doc(&self, name: &str) -> Result<Option<BlobId>> {
1211 self.db.read(
1212 "a document read",
1213 |txn| {
1214 let table = txn
1215 .open_table(DOCUMENTS)
1216 .map_err(|err| faulted("opening the documents table", err))?;
1217 let found = table
1218 .get(name)
1219 .map_err(|err| faulted("reading a control document pointer", err))?;
1220 match found {
1221 None => Ok(None),
1222 Some(value) => BlobId::from_slice(value.value())
1223 .map(Some)
1224 .ok_or_else(|| Error::CorruptDocument {
1225 ref_name: name.to_owned(),
1226 reason: format!(
1227 "its stored pointer is {} bytes, not the {BLOB_ID_LEN} of a blob id",
1228 value.value().len()
1229 ),
1230 }),
1231 }
1232 },
1233 |ram| Ok(ram.docs.get(name).copied()),
1234 )
1235 }
1236
1237 fn commit_doc(&self, name: &str, expected: Option<BlobId>, new: BlobId) -> Result<bool> {
1238 self.db.write("a control document commit", |txn| {
1239 {
1240 let mut documents = txn
1241 .open_table(DOCUMENTS)
1242 .map_err(|err| faulted("opening the documents table", err))?;
1243 let current = documents
1244 .get(name)
1245 .map_err(|err| faulted("reading a control document pointer", err))?
1246 .and_then(|value| BlobId::from_slice(value.value()));
1247 if current != expected {
1248 return Ok(Txn::Discard(false));
1251 }
1252 documents
1253 .insert(name, new.0.as_slice())
1254 .map_err(|err| faulted("moving a control document pointer", err))?;
1255
1256 let mut history = txn
1261 .open_table(DOC_HISTORY)
1262 .map_err(|err| faulted("opening the document history table", err))?;
1263 let seq = Self::next_seq(&history, name)?;
1264 let mut row = Vec::with_capacity(8 + BLOB_ID_LEN);
1265 row.extend_from_slice(&crate::model::now_ms().to_be_bytes());
1266 row.extend_from_slice(&new.0);
1267 history
1268 .insert((name, seq), row.as_slice())
1269 .map_err(|err| faulted("appending to the document history", err))?;
1270 }
1271 Ok(Txn::Commit(true))
1272 })
1273 }
1274
1275 fn history(&self, name: &str) -> Result<Vec<DocVersion>> {
1276 self.db.read(
1277 "a history read",
1278 |txn| {
1279 let table = txn
1280 .open_table(DOC_HISTORY)
1281 .map_err(|err| faulted("opening the document history table", err))?;
1282 let range = table
1283 .range((name, 0u64)..=(name, u64::MAX))
1284 .map_err(|err| faulted("scanning a document's history", err))?;
1285 let mut out = Vec::new();
1286 for entry in range {
1287 let (key, value) =
1288 entry.map_err(|err| faulted("reading a history row", err))?;
1289 let bytes = value.value();
1290 if bytes.len() != 8 + BLOB_ID_LEN {
1291 return Err(Error::CorruptDocument {
1292 ref_name: name.to_owned(),
1293 reason: format!(
1294 "a history row is {} bytes, not {}",
1295 bytes.len(),
1296 8 + BLOB_ID_LEN
1297 ),
1298 });
1299 }
1300 let mut millis = [0u8; 8];
1301 millis.copy_from_slice(&bytes[..8]);
1302 out.push(DocVersion {
1303 seq: key.value().1,
1304 at: i64::from_be_bytes(millis),
1305 blob: BlobId::from_slice(&bytes[8..]).expect("checked length"),
1306 });
1307 }
1308 Ok(out)
1309 },
1310 |ram| {
1314 Err(Error::OffStripe {
1315 why: format!("{}; the audit trail is not held in RAM", ram.why),
1316 })
1317 },
1318 )
1319 }
1320}
1321
1322pub type SharedControlStore = Arc<dyn ControlStore>;
1325
1326pub fn open_control_store(path: impl AsRef<Path>) -> Result<SharedControlStore> {
1328 Ok(Arc::new(RedbControlStore::open(path)?))
1329}
1330
1331pub fn open_control(
1338 path: impl AsRef<Path>,
1339) -> Result<(SharedControlStore, crate::secrets::RedbSecretStore)> {
1340 let (store, secrets, _) = open_control_with_db(path)?;
1341 Ok((store, secrets))
1342}
1343
1344pub fn open_control_with_db(
1349 path: impl AsRef<Path>,
1350) -> Result<(
1351 SharedControlStore,
1352 crate::secrets::RedbSecretStore,
1353 Arc<ControlDb>,
1354)> {
1355 let store = RedbControlStore::open(path)?;
1356 let secrets = store.secrets();
1357 let db = store.control_db();
1358 Ok((Arc::new(store), secrets, db))
1359}
1360
1361pub fn memory_control_store() -> Result<SharedControlStore> {
1363 Ok(Arc::new(RedbControlStore::in_memory()?))
1364}
1365
1366pub fn memory_control() -> Result<(SharedControlStore, crate::secrets::RedbSecretStore)> {
1369 let store = RedbControlStore::in_memory()?;
1370 let secrets = store.secrets();
1371 Ok((Arc::new(store), secrets))
1372}
1373
1374#[cfg(test)]
1375mod tests {
1376 use super::*;
1377
1378 #[test]
1379 fn a_blob_id_is_the_hash_of_the_bytes_and_round_trips_through_hex() {
1380 let id = BlobId::of(b"the identity table");
1381 assert_eq!(BlobId::from_hex(&id.to_string()), Some(id));
1382 assert_ne!(id, BlobId::of(b"the identity tabl3"));
1383 assert_eq!(BlobId::from_hex("not hex"), None);
1384 assert_eq!(BlobId::from_hex(""), None);
1385 }
1386
1387 #[test]
1388 fn an_unwritten_document_reads_as_absent_and_a_written_one_reads_back() {
1389 let store = RedbControlStore::in_memory().expect("open");
1390 assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), None);
1391 let id = store.put_blob(b"{}").expect("put");
1392 assert_eq!(
1393 store.get_blob(&id).expect("get").as_deref(),
1394 Some(&b"{}"[..])
1395 );
1396 assert!(store
1397 .commit_doc("refs/gunnar/auth", None, id)
1398 .expect("commit"));
1399 assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), Some(id));
1400 }
1401
1402 #[test]
1409 fn a_suspended_store_reads_from_ram_refuses_writes_and_resumes_without_drift() {
1410 use gunnar_store::SecretStore as _;
1411 let store = RedbControlStore::in_memory().expect("open");
1412 let secrets = store.secrets();
1413 let db = store.control_db();
1414 let id = store.put_blob(b"{\"principals\":1}").expect("put");
1415 assert!(store
1416 .commit_doc("refs/gunnar/auth", None, id)
1417 .expect("commit"));
1418 secrets
1419 .put("host-key", b"sealed bytes")
1420 .expect("put secret");
1421 assert!(!db.is_suspended());
1422
1423 let copy = db.suspend("draining for a growth").expect("suspend");
1424 assert!(db.is_suspended());
1425 assert_eq!(copy.documents(), 1);
1426 assert!(copy.bytes() > 0);
1427 assert_eq!(copy.why, "draining for a growth");
1428 assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), Some(id));
1430 assert_eq!(
1431 store.get_blob(&id).expect("get").as_deref(),
1432 Some(&b"{\"principals\":1}"[..])
1433 );
1434 assert_eq!(store.read_doc("never-written").expect("read"), None);
1435 assert_eq!(
1436 secrets.get("host-key").expect("get").as_deref(),
1437 Some(&b"sealed bytes"[..])
1438 );
1439 assert_eq!(secrets.get("absent").expect("get"), None);
1440 assert_eq!(
1441 secrets
1442 .get_or_create("host-key", &mut || panic!("must not mint"))
1443 .expect("held"),
1444 b"sealed bytes"
1445 );
1446 let refused = store.put_blob(b"new").unwrap_err();
1448 assert!(matches!(refused, Error::OffStripe { .. }), "{refused:?}");
1449 assert!(
1450 refused.to_string().contains("draining for a growth"),
1451 "{refused}"
1452 );
1453 assert!(refused.to_string().contains("retry"), "{refused}");
1454 let other = BlobId::of(b"other");
1455 assert!(matches!(
1456 store
1457 .commit_doc("refs/gunnar/auth", Some(id), other)
1458 .unwrap_err(),
1459 Error::OffStripe { .. }
1460 ));
1461 assert!(matches!(
1462 store.history("refs/gunnar/auth").unwrap_err(),
1463 Error::OffStripe { .. }
1464 ));
1465 let sec = secrets.put("host-key", b"x").unwrap_err().to_string();
1466 assert!(sec.contains("off the stripe"), "{sec}");
1467 let mint = secrets
1468 .get_or_create("fresh", &mut || Ok(b"minted".to_vec()))
1469 .unwrap_err()
1470 .to_string();
1471 assert!(mint.contains("will not be minted"), "{mint}");
1472 assert!(
1473 db.suspend("again").is_err(),
1474 "a second suspend is a refusal"
1475 );
1476
1477 let report = db.resume().expect("resume");
1478 assert!(!db.is_suspended());
1479 assert_eq!(report.documents_matched, 1);
1480 assert!(report.drift.is_empty(), "{:?}", report.drift);
1481 assert!(
1482 db.resume().is_err(),
1483 "resuming an unsuspended store is a refusal"
1484 );
1485 assert!(store
1487 .commit_doc("refs/gunnar/auth", Some(id), other)
1488 .expect("commit"));
1489 assert_eq!(store.history("refs/gunnar/auth").expect("history").len(), 2);
1490 assert_eq!(
1491 secrets
1492 .get_or_create("fresh", &mut || Ok(b"minted".to_vec()))
1493 .expect("mint"),
1494 b"minted"
1495 );
1496 }
1497
1498 #[cfg(target_os = "linux")]
1503 #[test]
1504 fn a_suspended_file_store_holds_no_descriptor_on_its_file() {
1505 let dir = tempfile::tempdir().expect("tempdir");
1506 let path = dir.path().join("control.redb");
1507 let store = RedbControlStore::open(&path).expect("open");
1508 let db = store.control_db();
1509 let id = store.put_blob(b"doc").expect("put");
1510 assert!(store.commit_doc("d", None, id).expect("commit"));
1511 let holds = |p: &Path| -> bool {
1512 std::fs::read_dir("/proc/self/fd")
1513 .expect("procfs")
1514 .flatten()
1515 .filter_map(|e| std::fs::read_link(e.path()).ok())
1516 .any(|t| t == p)
1517 };
1518 assert!(holds(&path), "the open store holds its file");
1519 db.suspend("unmounting").expect("suspend");
1520 assert!(
1521 !holds(&path),
1522 "a suspended store holds NO descriptor on its file"
1523 );
1524 assert_eq!(store.read_doc("d").expect("read from ram"), Some(id));
1525 db.resume().expect("resume");
1526 assert!(holds(&path), "a resumed store holds its file again");
1527 assert_eq!(store.read_doc("d").expect("read from disk"), Some(id));
1528 }
1529
1530 #[test]
1534 fn a_commit_against_the_wrong_expectation_is_refused_and_changes_nothing() {
1535 let store = RedbControlStore::in_memory().expect("open");
1536 let first = store.put_blob(b"one").expect("put");
1537 let second = store.put_blob(b"two").expect("put");
1538 assert!(store.commit_doc("doc", None, first).expect("commit"));
1539
1540 assert!(
1542 !store.commit_doc("doc", None, second).expect("commit"),
1543 "a commit that expected an absent document was accepted against a written one"
1544 );
1545 assert_eq!(
1546 store.read_doc("doc").expect("read"),
1547 Some(first),
1548 "a refused commit still moved the pointer"
1549 );
1550 assert_eq!(
1551 store.history("doc").expect("history").len(),
1552 1,
1553 "a refused commit still appended to the audit trail"
1554 );
1555
1556 assert!(store
1558 .commit_doc("doc", Some(first), second)
1559 .expect("commit"));
1560 assert_eq!(store.read_doc("doc").expect("read"), Some(second));
1561 }
1562
1563 #[test]
1565 fn the_history_is_append_only_and_one_document_does_not_see_anothers() {
1566 let store = RedbControlStore::in_memory().expect("open");
1567 let mut previous = None;
1568 let mut expected = Vec::new();
1569 for text in [b"a".as_slice(), b"bb", b"ccc"] {
1570 let id = store.put_blob(text).expect("put");
1571 assert!(store.commit_doc("auth", previous, id).expect("commit"));
1572 previous = Some(id);
1573 expected.push(id);
1574 }
1575 let other = store.put_blob(b"catalog").expect("put");
1576 assert!(store.commit_doc("catalog", None, other).expect("commit"));
1577
1578 let history = store.history("auth").expect("history");
1579 assert_eq!(
1580 history.iter().map(|v| v.blob).collect::<Vec<_>>(),
1581 expected,
1582 "the audit trail must list every version of the document, oldest first"
1583 );
1584 assert_eq!(
1585 history.iter().map(|v| v.seq).collect::<Vec<_>>(),
1586 vec![1, 2, 3]
1587 );
1588 assert_eq!(
1589 store.history("catalog").expect("history").len(),
1590 1,
1591 "one document's history picked up another document's versions"
1592 );
1593 assert_eq!(store.history("never-written").expect("history").len(), 0);
1594
1595 assert_eq!(
1598 store.get_blob(&history[0].blob).expect("get").as_deref(),
1599 Some(&b"a"[..])
1600 );
1601 }
1602}