Merge pull request #38 from andrewwhitehead/fix/session-remove

Minor cleanups for session removal

.github/workflows/build.yml

@@ -205,7 +205,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, macos-11, windows-latest]
-        python-version: [3.6]
+        python-version: [3.7]
         include:
           - os: ubuntu-latest
             plat-name: manylinux2014_x86_64

Cargo.toml

@@ -30,8 +30,8 @@ any = []
 ffi = ["any", "ffi-support", "logger", "option-lock"]
 jemalloc = ["jemallocator"]
 logger = ["env_logger", "log"]
-postgres = ["sqlx", "sqlx-core", "sqlx/postgres", "sqlx/tls"]
-sqlite = ["num_cpus", "sqlx", "sqlx-core", "sqlx/sqlite"]
+postgres = ["sqlx", "sqlx/postgres", "sqlx/tls"]
+sqlite = ["num_cpus", "sqlx", "sqlx/sqlite"]
 pg_test = ["postgres"]
 
 [dev-dependencies]
@@ -72,16 +72,11 @@ path = "./askar-crypto"
 features = ["all_keys", "any_key", "argon2", "crypto_box", "std"]
 
 [dependencies.sqlx]
-version = "=0.5.9"
+version = "0.5.11"
 default-features = false
 features = ["chrono", "runtime-tokio-rustls"]
 optional = true
 
-[dependencies.sqlx-core]
-version = "=0.5.9"
-default-features = false
-optional = true
-
 [profile.release]
 lto = true
 codegen-units = 1

src/ffi/handle.rs

@@ -1,4 +1,4 @@
-use std::{marker::PhantomData, mem, sync::Arc};
+use std::{fmt::Display, marker::PhantomData, mem, sync::Arc};
 
 use crate::error::Error;
 
@@ -48,7 +48,7 @@ impl<T> std::fmt::Display for ArcHandle<T> {
     }
 }
 
-pub trait ResourceHandle: Copy + Ord + From<usize> {
+pub trait ResourceHandle: Copy + Ord + From<usize> + Display {
     fn invalid() -> Self {
         Self::from(0)
     }

src/ffi/store.rs

@@ -84,17 +84,12 @@ where
         handle
     }
 
-    pub async fn remove(&self, handle: K) -> Result<V, Error> {
-        Arc::try_unwrap(
-            self.map
-                .write()
-                .await
-                .remove(&handle)
-                .ok_or_else(|| err_msg!("Invalid resource handle"))?
-                .1,
-        )
-        .map(|item| item.into_inner().unwrap())
-        .map_err(|_| err_msg!(Busy, "Resource handle in use"))
+    pub async fn remove(&self, handle: K) -> Option<Result<V, Error>> {
+        self.map.write().await.remove(&handle).map(|(_s, v)| {
+            Arc::try_unwrap(v)
+                .map(|item| item.into_inner().unwrap())
+                .map_err(|_| err_msg!(Busy, "Resource handle in use"))
+        })
     }
 
     pub async fn borrow(&self, handle: K) -> Result<TryMutexGuard<V>, Error> {
@@ -502,8 +497,13 @@ pub extern "C" fn askar_scan_free(handle: ScanHandle) -> ErrorCode {
     catch_err! {
         trace!("Close scan");
         spawn_ok(async move {
-            FFI_SCANS.remove(handle).await.ok();
-            info!("Closed scan {}", handle);
+            // the Scan may have been removed due to the Store being closed
+            if let Some(scan) = FFI_SCANS.remove(handle).await {
+                scan.ok();
+                info!("Closed scan {}", handle);
+            } else {
+                info!("Scan not found for closing: {}", handle);
+            }
         });
         Ok(ErrorCode::Success)
     }
@@ -996,15 +996,20 @@ pub extern "C" fn askar_session_close(
         });
         spawn_ok(async move {
             let result = async {
-                let session = FFI_SESSIONS.remove(handle).await?;
-                if commit == 0 {
-                    // not necessary - rollback is automatic for txn,
-                    // and for regular session there is no action to perform
-                    // > session.rollback().await?;
+                // the Session may have been removed due to the Store being closed
+                if let Some(session) = FFI_SESSIONS.remove(handle).await {
+                    let session = session?;
+                    if commit == 0 {
+                        // not necessary - rollback is automatic for txn,
+                        // and for regular session there is no action to perform
+                        // > session.rollback().await?;
+                    } else {
+                        session.commit().await?;
+                    }
+                    info!("Closed session {}", handle);
                 } else {
-                    session.commit().await?;
+                    info!("Session not found for closing: {}", handle);
                 }
-                info!("Closed session {}", handle);
                 Ok(())
             }.await;
             if let Some(cb) = cb {

wrappers/python/aries_askar/bindings.py

@@ -67,7 +67,7 @@ def __repr__(self) -> str:
 
     async def close(self, commit: bool = False):
         """Close the session."""
-        if not getattr(self, "_closed", False):
+        if not getattr(self, "_closed", False) and self:
             await do_call_async(
                 "askar_session_close",
                 self,

wrappers/python/aries_askar/store.py

@@ -569,7 +569,7 @@ async def rollback(self):
 
     async def close(self):
         if self._handle:
-            await self.handle.close(commit=False)
+            await self._handle.close(commit=False)
             self._handle = None
 
     def __repr__(self) -> str:

wrappers/python/setup.cfg

@@ -10,3 +10,4 @@ per_file_ignores = */__init__.py:D104
 minversion = 5.0
 testpaths =
     tests
+asyncio_mode=strict

wrappers/python/tests/test_store.py

@@ -1,8 +1,10 @@
 import os
 
-from pytest import fixture, mark
+from pytest import mark, raises
+import pytest_asyncio
 
 from aries_askar import (
+    AskarError,
     KeyAlg,
     Key,
     Store,
@@ -22,7 +24,7 @@ def raw_key() -> str:
     return Store.generate_raw_key(b"00000000000000000000000000000My1")
 
 
-@fixture
+@pytest_asyncio.fixture
 async def store() -> Store:
     key = raw_key()
     store = await Store.provision(TEST_STORE_URI, "raw", key, recreate=True)
@@ -229,4 +231,38 @@ async def test_profile(store: Store):
             ) == 1
         await store_2.close()
 
+    with raises(AskarError, match="Duplicate"):
+        _ = await store.create_profile(profile)
+
+    # check profile is still usable
+    async with store.session(profile) as session:
+        assert (
+            await session.count(
+                TEST_ENTRY["category"], {"~plaintag": "a", "enctag": "b"}
+            )
+        ) == 1
+
     await store.remove_profile(profile)
+
+    # profile key is cached
+    async with store.session(profile) as session:
+        assert (
+            await session.count(
+                TEST_ENTRY["category"], {"~plaintag": "a", "enctag": "b"}
+            )
+        ) == 0
+
+    with raises(AskarError, match="not found"):
+        async with store.session("unknown profile") as session:
+            await session.count(
+                TEST_ENTRY["category"], {"~plaintag": "a", "enctag": "b"}
+            )
+
+    await store.create_profile(profile)
+
+    async with store.session(profile) as session:
+        assert (
+            await session.count(
+                TEST_ENTRY["category"], {"~plaintag": "a", "enctag": "b"}
+            )
+        ) == 0