mirror of
https://github.com/Start9Labs/patch-db.git
synced 2026-03-26 02:11:54 +00:00
lock api now returns errors
This commit is contained in:
@@ -1,16 +1,16 @@
|
||||
use std::collections::BTreeSet;
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use json_ptr::{JsonPointer, SegList};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
use std::collections::BTreeSet;
|
||||
use tokio::sync::RwLockWriteGuard;
|
||||
use tokio::sync::{broadcast::Receiver, RwLock, RwLockReadGuard};
|
||||
use tokio::sync::broadcast::Receiver;
|
||||
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
|
||||
|
||||
use crate::locker::LockType;
|
||||
use crate::{locker::Guard, Locker, PatchDb, Revision, Store, Transaction};
|
||||
use crate::{patch::DiffPatch, Error};
|
||||
use crate::locker::{Guard, LockError, LockType};
|
||||
use crate::patch::DiffPatch;
|
||||
use crate::{Error, Locker, PatchDb, Revision, Store, Transaction};
|
||||
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct HandleId {
|
||||
@@ -68,7 +68,7 @@ pub trait DbHandle: Send + Sync {
|
||||
patch: DiffPatch,
|
||||
store_write_lock: Option<RwLockWriteGuard<'_, Store>>,
|
||||
) -> Result<Option<Arc<Revision>>, Error>;
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) -> ();
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) -> Result<(), LockError>;
|
||||
async fn get<
|
||||
T: for<'de> Deserialize<'de>,
|
||||
S: AsRef<str> + Send + Sync,
|
||||
@@ -154,7 +154,7 @@ impl<Handle: DbHandle + ?Sized> DbHandle for &mut Handle {
|
||||
) -> Result<Option<Arc<Revision>>, Error> {
|
||||
(*self).apply(patch, store_write_lock).await
|
||||
}
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) {
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) -> Result<(), LockError> {
|
||||
(*self).lock(ptr, lock_type).await
|
||||
}
|
||||
async fn get<
|
||||
@@ -266,9 +266,10 @@ impl DbHandle for PatchDbHandle {
|
||||
) -> Result<Option<Arc<Revision>>, Error> {
|
||||
self.db.apply(patch, None, store_write_lock).await
|
||||
}
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) {
|
||||
self.locks
|
||||
.push(self.db.locker.lock(self.id.clone(), ptr, lock_type).await);
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) -> Result<(), LockError> {
|
||||
Ok(self
|
||||
.locks
|
||||
.push(self.db.locker.lock(self.id.clone(), ptr, lock_type).await?))
|
||||
}
|
||||
async fn get<
|
||||
T: for<'de> Deserialize<'de>,
|
||||
|
||||
@@ -2,6 +2,7 @@ use std::io::Error as IOError;
|
||||
use std::sync::Arc;
|
||||
|
||||
use json_ptr::JsonPointer;
|
||||
use locker::LockError;
|
||||
use thiserror::Error;
|
||||
use tokio::sync::broadcast::error::TryRecvError;
|
||||
|
||||
@@ -20,8 +21,6 @@ mod proptest;
|
||||
mod test;
|
||||
|
||||
pub use handle::{DbHandle, PatchDbHandle};
|
||||
pub use json_patch;
|
||||
pub use json_ptr;
|
||||
pub use locker::{LockType, Locker};
|
||||
pub use model::{
|
||||
BoxModel, HasModel, Map, MapModel, Model, ModelData, ModelDataMut, OptionModel, VecModel,
|
||||
@@ -30,6 +29,7 @@ pub use patch::{DiffPatch, Dump, Revision};
|
||||
pub use patch_db_macro::HasModel;
|
||||
pub use store::{PatchDb, Store};
|
||||
pub use transaction::Transaction;
|
||||
pub use {json_patch, json_ptr};
|
||||
|
||||
#[derive(Error, Debug)]
|
||||
pub enum Error {
|
||||
@@ -53,4 +53,6 @@ pub enum Error {
|
||||
Subscriber(#[from] TryRecvError),
|
||||
#[error("Node Does Not Exist: {0}")]
|
||||
NodeDoesNotExist(JsonPointer),
|
||||
#[error("Invalid Lock Request: {0}")]
|
||||
LockError(#[from] LockError),
|
||||
}
|
||||
|
||||
@@ -209,7 +209,12 @@ impl Locker {
|
||||
});
|
||||
Locker { sender }
|
||||
}
|
||||
pub async fn lock(&self, handle_id: HandleId, ptr: JsonPointer, lock_type: LockType) -> Guard {
|
||||
pub async fn lock(
|
||||
&self,
|
||||
handle_id: HandleId,
|
||||
ptr: JsonPointer,
|
||||
lock_type: LockType,
|
||||
) -> Result<Guard, LockError> {
|
||||
struct CancelGuard {
|
||||
lock_info: Option<LockInfo>,
|
||||
channel: Option<oneshot::Sender<LockInfo>>,
|
||||
@@ -246,7 +251,7 @@ impl Locker {
|
||||
.unwrap();
|
||||
let res = (&mut cancel_guard.recv).await.unwrap();
|
||||
cancel_guard.channel.take();
|
||||
res
|
||||
Ok(res)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -910,6 +915,31 @@ impl std::fmt::Display for LockType {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, thiserror::Error)]
|
||||
pub enum LockError {
|
||||
#[error("Lock Taxonomy Escalation: Session = {session:?}, First = {first}, Second = {second}")]
|
||||
LockTaxonomyEscalation {
|
||||
session: HandleId,
|
||||
first: JsonPointer,
|
||||
second: JsonPointer,
|
||||
},
|
||||
#[error("Lock Type Escalation: Session = {session:?}, Pointer = {ptr}, First = {first}, Second = {second}")]
|
||||
LockTypeEscalation {
|
||||
session: HandleId,
|
||||
ptr: JsonPointer,
|
||||
first: LockType,
|
||||
second: LockType,
|
||||
},
|
||||
#[error(
|
||||
"Non-Canonical Lock Ordering: Session = {session:?}, First = {first}, Second = {second}"
|
||||
)]
|
||||
NonCanonicalOrdering {
|
||||
session: HandleId,
|
||||
first: JsonPointer,
|
||||
second: JsonPointer,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Request {
|
||||
lock_info: LockInfo,
|
||||
|
||||
@@ -9,7 +9,7 @@ use json_ptr::JsonPointer;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::locker::LockType;
|
||||
use crate::locker::{LockError, LockType};
|
||||
use crate::{DbHandle, DiffPatch, Error, Revision};
|
||||
|
||||
#[derive(Debug)]
|
||||
@@ -65,19 +65,23 @@ impl<T> Model<T>
|
||||
where
|
||||
T: Serialize + for<'de> Deserialize<'de>,
|
||||
{
|
||||
pub async fn lock<Db: DbHandle>(&self, db: &mut Db, lock_type: LockType) {
|
||||
pub async fn lock<Db: DbHandle>(
|
||||
&self,
|
||||
db: &mut Db,
|
||||
lock_type: LockType,
|
||||
) -> Result<(), LockError> {
|
||||
db.lock(self.ptr.clone(), lock_type).await
|
||||
}
|
||||
|
||||
pub async fn get<Db: DbHandle>(&self, db: &mut Db, lock: bool) -> Result<ModelData<T>, Error> {
|
||||
if lock {
|
||||
self.lock(db, LockType::Read).await;
|
||||
self.lock(db, LockType::Read).await?;
|
||||
}
|
||||
Ok(ModelData(db.get(&self.ptr).await?))
|
||||
}
|
||||
|
||||
pub async fn get_mut<Db: DbHandle>(&self, db: &mut Db) -> Result<ModelDataMut<T>, Error> {
|
||||
self.lock(db, LockType::Write).await;
|
||||
self.lock(db, LockType::Write).await?;
|
||||
let original = db.get_value(&self.ptr, None).await?;
|
||||
let current = serde_json::from_value(original.clone())?;
|
||||
Ok(ModelDataMut {
|
||||
@@ -105,7 +109,7 @@ where
|
||||
db: &mut Db,
|
||||
value: &T,
|
||||
) -> Result<Option<Arc<Revision>>, Error> {
|
||||
self.lock(db, LockType::Write).await;
|
||||
self.lock(db, LockType::Write).await?;
|
||||
db.put(&self.ptr, value).await
|
||||
}
|
||||
}
|
||||
@@ -228,7 +232,11 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> HasModel for Box<T> {
|
||||
#[derive(Debug)]
|
||||
pub struct OptionModel<T: HasModel + Serialize + for<'de> Deserialize<'de>>(T::Model);
|
||||
impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
|
||||
pub async fn lock<Db: DbHandle>(&self, db: &mut Db, lock_type: LockType) {
|
||||
pub async fn lock<Db: DbHandle>(
|
||||
&self,
|
||||
db: &mut Db,
|
||||
lock_type: LockType,
|
||||
) -> Result<(), LockError> {
|
||||
db.lock(self.0.as_ref().clone(), lock_type).await
|
||||
}
|
||||
|
||||
@@ -238,7 +246,7 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
|
||||
lock: bool,
|
||||
) -> Result<ModelData<Option<T>>, Error> {
|
||||
if lock {
|
||||
self.lock(db, LockType::Read).await;
|
||||
self.lock(db, LockType::Read).await?;
|
||||
}
|
||||
Ok(ModelData(db.get(self.0.as_ref()).await?))
|
||||
}
|
||||
@@ -247,7 +255,7 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
|
||||
&self,
|
||||
db: &mut Db,
|
||||
) -> Result<ModelDataMut<Option<T>>, Error> {
|
||||
self.lock(db, LockType::Write).await;
|
||||
self.lock(db, LockType::Write).await?;
|
||||
let original = db.get_value(self.0.as_ref(), None).await?;
|
||||
let current = serde_json::from_value(original.clone())?;
|
||||
Ok(ModelDataMut {
|
||||
@@ -259,7 +267,7 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
|
||||
|
||||
pub async fn exists<Db: DbHandle>(&self, db: &mut Db, lock: bool) -> Result<bool, Error> {
|
||||
if lock {
|
||||
db.lock(self.0.as_ref().clone(), LockType::Exist).await;
|
||||
db.lock(self.0.as_ref().clone(), LockType::Exist).await?;
|
||||
}
|
||||
Ok(db.exists(&self.as_ref(), None).await?)
|
||||
}
|
||||
@@ -285,7 +293,7 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
|
||||
}
|
||||
|
||||
pub async fn delete<Db: DbHandle>(&self, db: &mut Db) -> Result<Option<Arc<Revision>>, Error> {
|
||||
db.lock(self.as_ref().clone(), LockType::Write).await;
|
||||
db.lock(self.as_ref().clone(), LockType::Write).await?;
|
||||
db.put(self.as_ref(), &Value::Null).await
|
||||
}
|
||||
}
|
||||
@@ -319,7 +327,7 @@ where
|
||||
db: &mut Db,
|
||||
value: &T,
|
||||
) -> Result<Option<Arc<Revision>>, Error> {
|
||||
db.lock(self.as_ref().clone(), LockType::Write).await;
|
||||
db.lock(self.as_ref().clone(), LockType::Write).await?;
|
||||
db.put(self.as_ref(), value).await
|
||||
}
|
||||
}
|
||||
@@ -490,7 +498,7 @@ where
|
||||
lock: bool,
|
||||
) -> Result<BTreeSet<T::Key>, Error> {
|
||||
if lock {
|
||||
db.lock(self.as_ref().clone(), LockType::Exist).await;
|
||||
db.lock(self.as_ref().clone(), LockType::Exist).await?;
|
||||
}
|
||||
let set = db.keys(self.as_ref(), None).await?;
|
||||
Ok(set
|
||||
@@ -499,7 +507,7 @@ where
|
||||
.collect::<Result<_, _>>()?)
|
||||
}
|
||||
pub async fn remove<Db: DbHandle>(&self, db: &mut Db, key: &T::Key) -> Result<(), Error> {
|
||||
db.lock(self.as_ref().clone(), LockType::Write).await;
|
||||
db.lock(self.as_ref().clone(), LockType::Write).await?;
|
||||
if db.exists(self.clone().idx(key).as_ref(), None).await? {
|
||||
db.apply(
|
||||
DiffPatch(Patch(vec![PatchOperation::Remove(RemoveOperation {
|
||||
|
||||
@@ -10,7 +10,7 @@ use tokio::sync::broadcast::Receiver;
|
||||
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
|
||||
|
||||
use crate::handle::HandleId;
|
||||
use crate::locker::{Guard, LockType, Locker};
|
||||
use crate::locker::{Guard, LockError, LockType, Locker};
|
||||
use crate::patch::{DiffPatch, Revision};
|
||||
use crate::store::Store;
|
||||
use crate::{DbHandle, Error, PatchDbHandle};
|
||||
@@ -165,13 +165,13 @@ impl<Parent: DbHandle + Send + Sync> DbHandle for Transaction<Parent> {
|
||||
self.updates.append(patch);
|
||||
Ok(None)
|
||||
}
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) {
|
||||
self.locks.push(
|
||||
async fn lock(&mut self, ptr: JsonPointer, lock_type: LockType) -> Result<(), LockError> {
|
||||
Ok(self.locks.push(
|
||||
self.parent
|
||||
.locker()
|
||||
.lock(self.id.clone(), ptr, lock_type)
|
||||
.await,
|
||||
)
|
||||
.await?,
|
||||
))
|
||||
}
|
||||
async fn get<
|
||||
T: for<'de> Deserialize<'de>,
|
||||
|
||||
Reference in New Issue
Block a user