Start9/patch-db
2
0
Fork 0
mirror of https://github.com/Start9Labs/patch-db.git synced 2026-03-26 02:11:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

atomic write of .bak

Browse Source
This commit is contained in:
Aiden McClelland
2021-03-22 17:22:50 -06:00
parent 9fd80fae58
commit 3fa224f2f6
1 changed files with 12 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
patch-db/src/lib.rs
View File

@@ -221,7 +221,13 @@ impl Store {
json_patch::patch(&mut data, &patch)?; json_patch::patch(&mut data, &patch)?;
revision += 1; revision += 1;
} }
serde_cbor::to_writer(std::fs::File::create(&bak)?, &data)?; let bak_tmp = bak.with_extension("bak.tmp");
let mut backup_file = std::fs::File::create(&bak_tmp)?;
serde_cbor::to_writer(&mut backup_file, &revision)?;
serde_cbor::to_writer(&mut backup_file, &data)?;
backup_file.flush()?;
backup_file.sync_all()?;
std::fs::rename(&bak_tmp, &bak)?;
nix::unistd::ftruncate(std::os::unix::io::AsRawFd::as_raw_fd(&*f), 0) nix::unistd::ftruncate(std::os::unix::io::AsRawFd::as_raw_fd(&*f), 0)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
serde_cbor::to_writer(&mut *f, &revision)?; serde_cbor::to_writer(&mut *f, &revision)?;
@@ -321,7 +327,7 @@ pub struct PatchDb {
} }
impl PatchDb { impl PatchDb {
pub async fn open<P: AsRef<Path> + Send + 'static>(path: P) -> Result<Self, Error> { pub async fn open<P: AsRef<Path> + Send + 'static>(path: P) -> Result<Self, Error> {
let (subscriber, _) = tokio::sync::broadcast::channel(16); let (subscriber, _) = tokio::sync::broadcast::channel(16); // TODO: make this unbounded
Ok(PatchDb { Ok(PatchDb {
store: Arc::new(RwLock::new(Store::open(path).await?)), store: Arc::new(RwLock::new(Store::open(path).await?)),
@@ -340,7 +346,10 @@ impl PatchDb {
ptr: &JsonPointer<S, V>, ptr: &JsonPointer<S, V>,
value: &T, value: &T,
) -> Result<Arc<Revision>, Error> { ) -> Result<Arc<Revision>, Error> {
self.store.write().await.put(ptr, value).await let mut store = self.store.write().await;
let rev = store.put(ptr, value).await?;
self.subscriber.send(rev.clone()).unwrap_or_default();
Ok(rev)
} }
pub async fn apply( pub async fn apply(
&self, &self,