always rollback (#52)

* always rollback

* cleanup

* cleanup

cacholote/cache.py

@@ -37,7 +37,7 @@
 _LOCKER = "__locked__"
 
 
-def _update_last_primary_keys_and_return(
+def _update_last_primary_keys(
     session: sqlalchemy.orm.Session, cache_entry: Any, tag: Optional[str]
 ) -> Any:
     # Wait until unlocked
@@ -55,12 +55,15 @@ def _update_last_primary_keys_and_return(
     cache_entry.counter += 1
     if tag is not None:
         cache_entry.tag = tag
-    session.commit()
+    try:
+        session.commit()
+    finally:
+        session.rollback()
     LAST_PRIMARY_KEYS.set(cache_entry._primary_keys)
     return result
 
 
-def _clear_last_primary_keys_and_return(result: Any) -> Any:
+def _clear_last_primary_keys(result: Any) -> Any:
     LAST_PRIMARY_KEYS.set({})
     return result
 
@@ -69,7 +72,10 @@ def _delete_cache_entry(
     session: sqlalchemy.orm.Session, cache_entry: config.CacheEntry
 ) -> None:
     session.delete(cache_entry)
-    session.commit()
+    try:
+        session.commit()
+    finally:
+        session.rollback()
     # Delete cache file
     json.loads(cache_entry._result_as_string, object_hook=clean._delete_cache_file)
 
@@ -122,14 +128,14 @@ def wrapper(
 
         # Cache opt-out
         if not settings.use_cache:
-            return _clear_last_primary_keys_and_return(func(*args, **kwargs))
+            return _clear_last_primary_keys(func(*args, **kwargs))
 
         try:
             # Get key
             hexdigest = hexdigestify_python_call(func, *args, **kwargs)
         except encode.EncodeError as ex:
             warnings.warn(f"can NOT encode python call: {ex!r}", UserWarning)
-            return _clear_last_primary_keys_and_return(func(*args, **kwargs))
+            return _clear_last_primary_keys(func(*args, **kwargs))
 
         # Filters for the database query
         filters = [
@@ -147,53 +153,48 @@ def wrapper(
             ):
                 # Attempt all valid cache entries
                 try:
-                    return _update_last_primary_keys_and_return(
-                        session, cache_entry, tag
-                    )
+                    return _update_last_primary_keys(session, cache_entry, tag)
                 except decode.DecodeError as ex:
                     # Something wrong, e.g. cached files are corrupted
                     warnings.warn(str(ex), UserWarning)
                     _delete_cache_entry(session, cache_entry)
-                finally:
-                    session.rollback()
 
             # Not in the cache
             cache_entry = None
             try:
-                # Lock cache entry
+                # Acquire lock
                 cache_entry = config.CacheEntry(
                     key=hexdigest,
                     expiration=expiration,
                     result=_LOCKER,
                     tag=settings.tag,
                 )
                 session.add(cache_entry)
-                session.commit()
+                try:
+                    session.commit()
+                finally:
+                    session.rollback()
             except sqlalchemy.exc.IntegrityError:
                 # Concurrent job: This cache entry already exists.
                 filters = [
                     config.CacheEntry.key == cache_entry.key,
                     config.CacheEntry.expiration == cache_entry.expiration,
                 ]
-                session.rollback()
                 cache_entry = session.query(config.CacheEntry).filter(*filters).one()
-                return _update_last_primary_keys_and_return(session, cache_entry, tag)
+                return _update_last_primary_keys(session, cache_entry, tag)
             else:
                 # Compute result from scratch
                 result = func(*args, **kwargs)
                 try:
                     # Update cache
                     cache_entry.result = json.loads(encode.dumps(result))
-                    return _update_last_primary_keys_and_return(
-                        session, cache_entry, tag
-                    )
+                    return _update_last_primary_keys(session, cache_entry, tag)
                 except encode.EncodeError as ex:
                     # Enconding error, return result without caching
                     warnings.warn(f"can NOT encode output: {ex!r}", UserWarning)
-                    return _clear_last_primary_keys_and_return(result)
+                    return _clear_last_primary_keys(result)
             finally:
-                session.rollback()
-                # Unlock
+                # Release lock
                 if cache_entry and cache_entry.result == _LOCKER:
                     _delete_cache_entry(session, cache_entry)
 

cacholote/clean.py

@@ -49,7 +49,10 @@ def _delete_cache_file(
             if session and cache_entry and not dry_run:
                 logging.info("Deleting cache entry: %r", cache_entry)
                 session.delete(cache_entry)
-                session.commit()
+                try:
+                    session.commit()
+                finally:
+                    session.rollback()
             if not dry_run:
                 with utils._Locker(fs, urlpath) as file_exists:
                     if file_exists: