|
@@ -267,11 +267,20 @@ impl AccountStore for SqlStore {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
async fn create_account(&self, account: Account) -> Result<(), StoreError> {
|
|
async fn create_account(&self, account: Account) -> Result<(), StoreError> {
|
|
|
|
|
+ // Wrap the existence check and the insert in one transaction so two
|
|
|
|
|
+ // concurrent creates of the same (id, subaccount) cannot both pass the
|
|
|
|
|
+ // check; the loser sees the conflict rather than a torn read.
|
|
|
|
|
+ let mut tx = self
|
|
|
|
|
+ .pool
|
|
|
|
|
+ .begin()
|
|
|
|
|
+ .await
|
|
|
|
|
+ .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
|
|
+
|
|
|
let exists =
|
|
let exists =
|
|
|
sqlx::query("SELECT 1 FROM accounts WHERE id = $1 AND subaccount = $2 LIMIT 1")
|
|
sqlx::query("SELECT 1 FROM accounts WHERE id = $1 AND subaccount = $2 LIMIT 1")
|
|
|
.bind(account.id.id)
|
|
.bind(account.id.id)
|
|
|
.bind(account.id.sub as i64)
|
|
.bind(account.id.sub as i64)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
|
|
|
|
+ .fetch_optional(&mut *tx)
|
|
|
.await
|
|
.await
|
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
if exists.is_some() {
|
|
if exists.is_some() {
|
|
@@ -292,19 +301,32 @@ impl AccountStore for SqlStore {
|
|
|
.bind(account.book.0)
|
|
.bind(account.book.0)
|
|
|
.bind(serialize_json(&account.user_data)?)
|
|
.bind(serialize_json(&account.user_data)?)
|
|
|
.bind(serialize_json(&account.metadata)?)
|
|
.bind(serialize_json(&account.metadata)?)
|
|
|
- .execute(&self.pool)
|
|
|
|
|
|
|
+ .execute(&mut *tx)
|
|
|
|
|
+ .await
|
|
|
|
|
+ .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
|
|
+
|
|
|
|
|
+ tx.commit()
|
|
|
.await
|
|
.await
|
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
Ok(())
|
|
Ok(())
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
async fn append_account_version(&self, account: Account) -> Result<(), StoreError> {
|
|
async fn append_account_version(&self, account: Account) -> Result<(), StoreError> {
|
|
|
|
|
+ // Read the current version and insert the next one atomically, so two
|
|
|
|
|
+ // concurrent appends cannot both read the same version and race to insert
|
|
|
|
|
+ // `current + 1`.
|
|
|
|
|
+ let mut tx = self
|
|
|
|
|
+ .pool
|
|
|
|
|
+ .begin()
|
|
|
|
|
+ .await
|
|
|
|
|
+ .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
|
|
+
|
|
|
let current = sqlx::query(
|
|
let current = sqlx::query(
|
|
|
"SELECT version FROM accounts WHERE id = $1 AND subaccount = $2 ORDER BY version DESC LIMIT 1",
|
|
"SELECT version FROM accounts WHERE id = $1 AND subaccount = $2 ORDER BY version DESC LIMIT 1",
|
|
|
)
|
|
)
|
|
|
.bind(account.id.id)
|
|
.bind(account.id.id)
|
|
|
.bind(account.id.sub as i64)
|
|
.bind(account.id.sub as i64)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
|
|
|
|
+ .fetch_optional(&mut *tx)
|
|
|
.await
|
|
.await
|
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?
|
|
|
.ok_or_else(|| StoreError::NotFound(format!("account {:?}", account.id)))?;
|
|
.ok_or_else(|| StoreError::NotFound(format!("account {:?}", account.id)))?;
|
|
@@ -335,7 +357,11 @@ impl AccountStore for SqlStore {
|
|
|
.bind(account.book.0)
|
|
.bind(account.book.0)
|
|
|
.bind(serialize_json(&account.user_data)?)
|
|
.bind(serialize_json(&account.user_data)?)
|
|
|
.bind(serialize_json(&account.metadata)?)
|
|
.bind(serialize_json(&account.metadata)?)
|
|
|
- .execute(&self.pool)
|
|
|
|
|
|
|
+ .execute(&mut *tx)
|
|
|
|
|
+ .await
|
|
|
|
|
+ .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
|
|
+
|
|
|
|
|
+ tx.commit()
|
|
|
.await
|
|
.await
|
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
Ok(())
|
|
Ok(())
|
|
@@ -964,9 +990,17 @@ impl EventStore for SqlStore {
|
|
|
#[async_trait]
|
|
#[async_trait]
|
|
|
impl BookStore for SqlStore {
|
|
impl BookStore for SqlStore {
|
|
|
async fn create_book(&self, book: Book) -> Result<(), StoreError> {
|
|
async fn create_book(&self, book: Book) -> Result<(), StoreError> {
|
|
|
|
|
+ // Existence check and insert in one transaction so two concurrent creates
|
|
|
|
|
+ // of the same book id cannot both pass the check.
|
|
|
|
|
+ let mut tx = self
|
|
|
|
|
+ .pool
|
|
|
|
|
+ .begin()
|
|
|
|
|
+ .await
|
|
|
|
|
+ .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
|
|
+
|
|
|
let exists = sqlx::query("SELECT 1 FROM books WHERE id = $1 LIMIT 1")
|
|
let exists = sqlx::query("SELECT 1 FROM books WHERE id = $1 LIMIT 1")
|
|
|
.bind(book.id.0)
|
|
.bind(book.id.0)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
|
|
|
|
+ .fetch_optional(&mut *tx)
|
|
|
.await
|
|
.await
|
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
if exists.is_some() {
|
|
if exists.is_some() {
|
|
@@ -978,7 +1012,11 @@ impl BookStore for SqlStore {
|
|
|
.bind(book.id.0)
|
|
.bind(book.id.0)
|
|
|
.bind(&book.name)
|
|
.bind(&book.name)
|
|
|
.bind(&data)
|
|
.bind(&data)
|
|
|
- .execute(&self.pool)
|
|
|
|
|
|
|
+ .execute(&mut *tx)
|
|
|
|
|
+ .await
|
|
|
|
|
+ .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
|
|
+
|
|
|
|
|
+ tx.commit()
|
|
|
.await
|
|
.await
|
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
.map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
Ok(())
|
|
Ok(())
|