chore: Revert HashCache DB changes causing E2E test failures (#892)

* chore: Revert "fix: Add backoff retry when opening HashDB. Also ensure each test is using isolated cache location. (#884)"

This reverts commit 08c1e89.

Signed-off-by: Jericho Tolentino <[email protected]>

* chore: Revert "fix: concurrent job bundle submissions fail due to sqlite3 table already exist (#862)"

This reverts commit b35aeb1.

Signed-off-by: Jericho Tolentino <[email protected]>

* chore: Revert "fix: Adding thread local s3 cache db connections to fix performance bottleneck during bundle submission (#848)"

This reverts commit 4914e61.

Signed-off-by: Jericho Tolentino <[email protected]>

---------

Signed-off-by: Jericho Tolentino <[email protected]>

Author: jericht

src/deadline/job_attachments/caches/cache_db.py

@@ -6,13 +6,11 @@
 
 import logging
 import os
-import threading
 from abc import ABC
 from threading import Lock
 from typing import Optional
 
 from ..exceptions import JobAttachmentsError
-from .._utils import _retry
 
 CONFIG_ROOT = ".deadline"
 COMPONENT_NAME = "job_attachments"
@@ -28,9 +26,6 @@ class CacheDB(ABC):
     close the connection to the cache database.
     """
 
-    # Number of retry attempts for SQLite operational errors (e.g., database locks)
-    RETRY_ATTEMPTS = 3
-
     def __init__(
         self, cache_name: str, table_name: str, create_query: str, cache_dir: Optional[str] = None
     ) -> None:
@@ -39,8 +34,6 @@ def __init__(
         self.cache_name: str = cache_name
         self.table_name: str = table_name
         self.create_query: str = create_query
-        self.local = threading.local()
-        self.local_connections: set = set()
 
         try:
             # SQLite is included in Python installers, but might not exist if building python from source.
@@ -67,85 +60,29 @@ def __enter__(self):
         if self.enabled:
             import sqlite3
 
-            @_retry(
-                ExceptionToCheck=sqlite3.OperationalError,
-                tries=self.RETRY_ATTEMPTS,
-                delay=(0.5, 1.5),  # Jitter between 0.5 and 1.5 seconds
-                backoff=1.0,
-                logger=logger.warning,
-            )
-            def _connect_to_db():
-                try:
-                    connection = sqlite3.connect(self.cache_dir, check_same_thread=False)
-                    try:
-                        # Test the connection by trying to query the table
-                        connection.execute(f"SELECT * FROM {self.table_name}")
-                    except Exception:
-                        # DB file doesn't have our table, so we need to create it
-                        logger.info(
-                            f"No cache entries for the current library version were found. Creating a new cache for {self.cache_name}"
-                        )
-                        connection.execute(self.create_query)
-                    return connection
-                except sqlite3.OperationalError as oe:
-                    logger.info("Error connecting to database, retrying.")
-                    raise oe
-
             try:
-                self.db_connection = _connect_to_db()
+                self.db_connection: sqlite3.Connection = sqlite3.connect(
+                    self.cache_dir, check_same_thread=False
+                )
             except sqlite3.OperationalError as oe:
                 raise JobAttachmentsError(
-                    f"Could not access cache file after {self.RETRY_ATTEMPTS} retry attempts: {self.cache_dir}"
+                    f"Could not access cache file in {self.cache_dir}"
                 ) from oe
+
+            try:
+                self.db_connection.execute(f"SELECT * FROM {self.table_name}")
+            except Exception:
+                # DB file doesn't have our table, so we need to create it
+                logger.info(
+                    f"No cache entries for the current library version were found. Creating a new cache for {self.cache_name}"
+                )
+                self.db_connection.execute(self.create_query)
         return self
 
     def __exit__(self, exc_type, exc_value, exc_traceback):
         """Called when exiting the context manager."""
-
         if self.enabled:
-            import sqlite3
-
             self.db_connection.close()
-            for conn in self.local_connections:
-                try:
-                    conn.close()
-                except sqlite3.Error as e:
-                    logger.warning(f"SQLite connection failed to close with error {e}")
-
-            self.local_connections.clear()
-
-    def get_local_connection(self):
-        """Create and/or returns a thread local connection to the SQLite database."""
-        if not self.enabled:
-            return None
-        import sqlite3
-
-        if not hasattr(self.local, "connection"):
-
-            @_retry(
-                ExceptionToCheck=sqlite3.OperationalError,
-                tries=self.RETRY_ATTEMPTS,
-                delay=(0.5, 1.5),  # Jitter between 0.5 and 1.5 seconds
-                backoff=1.0,
-                logger=logger.warning,
-            )
-            def _create_local_connection():
-                try:
-                    connection = sqlite3.connect(self.cache_dir, check_same_thread=False)
-                    return connection
-                except sqlite3.OperationalError as oe:
-                    logger.info("Error connecting to database, retrying.")
-                    raise oe
-
-            try:
-                self.local.connection = _create_local_connection()
-                self.local_connections.add(self.local.connection)
-            except sqlite3.OperationalError as oe:
-                raise JobAttachmentsError(
-                    f"Could not create connection to cache after {self.RETRY_ATTEMPTS} retry attempts: {self.cache_dir}"
-                ) from oe
-
-        return self.local.connection
 
     @classmethod
     def get_default_cache_db_file_dir(cls) -> Optional[str]:
@@ -162,23 +99,12 @@ def remove_cache(self) -> None:
         """
         Removes the underlying cache contents from the file system.
         """
-
         if self.enabled:
-            import sqlite3
-
             self.db_connection.close()
-            conn_list = list(self.local_connections)
-            for conn in conn_list:
-                try:
-                    conn.close()
-                    self.local_connections.remove(conn)
-                except sqlite3.Error as e:
-                    logger.warning(f"SQLite connection failed to close with error {e}")
 
         logger.debug(f"The cache {self.cache_dir} will be removed")
         try:
             os.remove(self.cache_dir)
         except Exception as e:
             logger.error(f"Error occurred while removing the cache file {self.cache_dir}: {e}")
-
             raise e

src/deadline/job_attachments/caches/hash_cache.py

@@ -51,40 +51,14 @@ class HashCache(CacheDB):
 
     def __init__(self, cache_dir: Optional[str] = None) -> None:
         table_name: str = f"hashesV{self.CACHE_DB_VERSION}"
-        create_query: str = f"CREATE TABLE IF NOT EXISTS hashesV{self.CACHE_DB_VERSION}(file_path blob primary key, hash_algorithm text secondary key, file_hash text, last_modified_time timestamp)"
+        create_query: str = f"CREATE TABLE hashesV{self.CACHE_DB_VERSION}(file_path blob primary key, hash_algorithm text secondary key, file_hash text, last_modified_time timestamp)"
         super().__init__(
             cache_name=self.CACHE_NAME,
             table_name=table_name,
             create_query=create_query,
             cache_dir=cache_dir,
         )
 
-    def get_connection_entry(
-        self, file_path_key: str, hash_algorithm: HashAlgorithm, connection
-    ) -> Optional[HashCacheEntry]:
-        """
-        Returns an entry from the hash cache, if it exists.
-        """
-        if not self.enabled:
-            return None
-
-        entry_vals = connection.execute(
-            f"SELECT * FROM {self.table_name} WHERE file_path=? AND hash_algorithm=?",
-            [
-                file_path_key.encode(encoding="utf-8", errors="surrogatepass"),
-                hash_algorithm.value,
-            ],
-        ).fetchone()
-        if entry_vals:
-            return HashCacheEntry(
-                file_path=str(entry_vals[0], encoding="utf-8", errors="surrogatepass"),
-                hash_algorithm=HashAlgorithm(entry_vals[1]),
-                file_hash=entry_vals[2],
-                last_modified_time=str(entry_vals[3]),
-            )
-        else:
-            return None
-
     def get_entry(
         self, file_path_key: str, hash_algorithm: HashAlgorithm
     ) -> Optional[HashCacheEntry]:
@@ -95,7 +69,22 @@ def get_entry(
             return None
 
         with self.db_lock, self.db_connection:
-            return self.get_connection_entry(file_path_key, hash_algorithm, self.db_connection)
+            entry_vals = self.db_connection.execute(
+                f"SELECT * FROM {self.table_name} WHERE file_path=? AND hash_algorithm=?",
+                [
+                    file_path_key.encode(encoding="utf-8", errors="surrogatepass"),
+                    hash_algorithm.value,
+                ],
+            ).fetchone()
+            if entry_vals:
+                return HashCacheEntry(
+                    file_path=str(entry_vals[0], encoding="utf-8", errors="surrogatepass"),
+                    hash_algorithm=HashAlgorithm(entry_vals[1]),
+                    file_hash=entry_vals[2],
+                    last_modified_time=str(entry_vals[3]),
+                )
+            else:
+                return None
 
     def put_entry(self, entry: HashCacheEntry) -> None:
         """Inserts or replaces an entry into the hash cache database after acquiring the lock."""

src/deadline/job_attachments/caches/s3_check_cache.py

@@ -48,37 +48,14 @@ class S3CheckCache(CacheDB):
 
     def __init__(self, cache_dir: Optional[str] = None) -> None:
         table_name: str = f"s3checkV{self.CACHE_DB_VERSION}"
-        create_query: str = f"CREATE TABLE IF NOT EXISTS s3checkV{self.CACHE_DB_VERSION}(s3_key text primary key, last_seen_time timestamp)"
+        create_query: str = f"CREATE TABLE s3checkV{self.CACHE_DB_VERSION}(s3_key text primary key, last_seen_time timestamp)"
         super().__init__(
             cache_name=self.CACHE_NAME,
             table_name=table_name,
             create_query=create_query,
             cache_dir=cache_dir,
         )
 
-    def get_connection_entry(self, s3_key: str, connection) -> Optional[S3CheckCacheEntry]:
-        """
-        Checks if an entry exists in the cache, and returns it if it hasn't expired.
-        """
-
-        entry_vals = connection.execute(
-            f"SELECT * FROM {self.table_name} WHERE s3_key=?",
-            [s3_key],
-        ).fetchone()
-        if entry_vals:
-            entry = S3CheckCacheEntry(
-                s3_key=entry_vals[0],
-                last_seen_time=str(entry_vals[1]),
-            )
-            try:
-                last_seen = datetime.fromtimestamp(float(entry.last_seen_time))
-                if (datetime.now() - last_seen).days < self.ENTRY_EXPIRY_DAYS:
-                    return entry
-            except ValueError:
-                logger.warning(f"Timestamp for S3 key {s3_key} is not valid. Ignoring.")
-
-        return None
-
     def get_entry(self, s3_key: str) -> Optional[S3CheckCacheEntry]:
         """
         Checks if an entry exists in the cache, and returns it if it hasn't expired.
@@ -87,7 +64,23 @@ def get_entry(self, s3_key: str) -> Optional[S3CheckCacheEntry]:
             return None
 
         with self.db_lock, self.db_connection:
-            return self.get_connection_entry(s3_key, self.db_connection)
+            entry_vals = self.db_connection.execute(
+                f"SELECT * FROM {self.table_name} WHERE s3_key=?",
+                [s3_key],
+            ).fetchone()
+            if entry_vals:
+                entry = S3CheckCacheEntry(
+                    s3_key=entry_vals[0],
+                    last_seen_time=str(entry_vals[1]),
+                )
+                try:
+                    last_seen = datetime.fromtimestamp(float(entry.last_seen_time))
+                    if (datetime.now() - last_seen).days < self.ENTRY_EXPIRY_DAYS:
+                        return entry
+                except ValueError:
+                    logger.warning(f"Timestamp for S3 key {s3_key} is not valid. Ignoring.")
+
+            return None
 
     def put_entry(self, entry: S3CheckCacheEntry) -> None:
         """Inserts or replaces an entry into the cache database."""

src/deadline/job_attachments/upload.py

@@ -593,11 +593,8 @@ def verify_hash_cache_integrity(
         random.shuffle(s3_upload_keys)
         sampled_cache_entries: List[S3CheckCacheEntry] = []
         with S3CheckCache(s3_check_cache_dir) as s3_cache:
-            local_connection = s3_cache.get_local_connection()
             for upload_key in s3_upload_keys:
-                this_entry = s3_cache.get_connection_entry(
-                    s3_key=f"{s3_bucket}/{upload_key}", connection=local_connection
-                )
+                this_entry = s3_cache.get_entry(s3_key=f"{s3_bucket}/{upload_key}")
                 if this_entry is not None:
                     sampled_cache_entries.append(this_entry)
                     if len(sampled_cache_entries) >= 30:
@@ -654,9 +651,7 @@ def upload_object_to_cas(
         s3_upload_key = self._generate_s3_upload_key(file, hash_algorithm, s3_cas_prefix)
         is_uploaded = False
 
-        if s3_check_cache.get_connection_entry(
-            s3_key=f"{s3_bucket}/{s3_upload_key}", connection=s3_check_cache.get_local_connection()
-        ):
+        if s3_check_cache.get_entry(s3_key=f"{s3_bucket}/{s3_upload_key}"):
             logger.debug(
                 f"skipping {local_path} because {s3_bucket}/{s3_upload_key} exists in the cache"
             )
@@ -1073,9 +1068,7 @@ def _process_input_path(
         file_status: FileStatus = FileStatus.UNCHANGED
         actual_modified_time = str(datetime.fromtimestamp(path.stat().st_mtime))
 
-        entry: Optional[HashCacheEntry] = hash_cache.get_connection_entry(
-            full_path, hash_alg, connection=hash_cache.get_local_connection()
-        )
+        entry: Optional[HashCacheEntry] = hash_cache.get_entry(full_path, hash_alg)
         if entry is not None:
             # If the file was modified, we need to rehash it
             if actual_modified_time != entry.last_modified_time:

test/unit/conftest.py

@@ -13,13 +13,10 @@
 
 
 @pytest.fixture(scope="function")
-def fresh_deadline_config(monkeypatch):
+def fresh_deadline_config():
     """
-    Fixture to start with a blank AWS Deadline Cloud config file and isolated cache directories.
+    Fixture to start with a blank AWS Deadline Cloud config file.
 
-    This fixture also overrides the HOME environment variable to ensure cache isolation
-    between tests. Both HashCache and S3CheckCache will use temporary directories under
-    the isolated HOME directory.
     """
 
     # Clear the session cache. Importing the cache invalidator at runtime is necessary
@@ -36,23 +33,6 @@ def fresh_deadline_config(monkeypatch):
         with open(temp_file_path, "w+t", encoding="utf8") as temp_file:
             temp_file.write("")
 
-        # Create a temporary HOME directory for cache isolation
-        temp_home_dir = tempfile.TemporaryDirectory()
-        temp_home_path = Path(temp_home_dir.name)
-
-        # Override HOME environment variable to isolate cache directories
-        # This affects both:
-        # - HashCache default: $HOME/.deadline/job_attachments/
-        # - S3CheckCache via config_file.get_cache_directory(): $HOME/.deadline/cache/
-        monkeypatch.setenv("HOME", str(temp_home_path))
-
-        # On Windows, os.path.expanduser("~") uses USERPROFILE instead of HOME
-        # So we need to override USERPROFILE as well for Windows compatibility
-        import sys
-
-        if sys.platform == "win32":
-            monkeypatch.setenv("USERPROFILE", str(temp_home_path))
-
         # Yield the temp file name with it patched in as the
         # AWS Deadline Cloud config file
         with patch.object(config_file, "CONFIG_FILE_PATH", str(temp_file_path)):
@@ -64,7 +44,6 @@ def fresh_deadline_config(monkeypatch):
             yield str(temp_file_path)
     finally:
         temp_dir.cleanup()
-        temp_home_dir.cleanup()
 
 
 @pytest.fixture(scope="function", autouse=True)