Handle duplicate tars

Author: forsyth2

zstash/extract.py

@@ -13,7 +13,7 @@
 import tarfile
 import traceback
 from datetime import datetime
-from typing import DefaultDict, List, Optional, Tuple
+from typing import DefaultDict, List, Optional, Set, Tuple
 
 import _hashlib
 import _io
@@ -390,25 +390,75 @@ def multiprocess_extract(
 
 
 def check_sizes_match(cur, tfname):
-    match: bool
     if cur and tars_table_exists(cur):
         logger.info(f"{tfname} exists. Checking expected size matches actual size.")
-        actual_size = os.path.getsize(tfname)
-        name_only = os.path.split(tfname)[1]
-        cur.execute(f"select size from tars where name is '{name_only}';")
-        expected_size: int = cur.fetchall()[0][0]
+        actual_size: int = os.path.getsize(tfname)
+        name_only: str = os.path.split(tfname)[1]
+
+        # Get ALL entries for this tar name
+        cur.execute("SELECT size FROM tars WHERE name = ?", (name_only,))
+        results = cur.fetchall()
+
+        if not results:
+            # Cannot access size information; assume the sizes match.
+            logger.error(f"No database entries found for {name_only}")
+            return True
+
+        # Check for multiple entries
+        if len(results) > 1:
+            # Extract just the size values
+            sizes: List[int] = [row[0] for row in results]
+            unique_sizes: Set[int] = set(sizes)
+
+            logger.error(
+                f"{name_only}: Found {len(results)} database entries for this tar"
+            )
+            logger.error(f"{name_only}: Database sizes: {sizes}")
+
+            if len(unique_sizes) > 1:
+                # Multiple entries with different sizes.
+                for unique_size in unique_sizes:
+                    if unique_size == actual_size:
+                        logger.info(
+                            f"{name_only}: There exists at least one entry with the same size as the actual file size: {unique_size}. "
+                        )
+                        break
+                error_msg = (
+                    f"{name_only}: Database corruption detected! "
+                    f"Found {len(results)} entries with {len(unique_sizes)} different sizes: {list(unique_sizes)}. "
+                    f"Actual file size: {actual_size}. "
+                    f"Database cleanup required before proceeding."
+                )
+                logger.error(error_msg)
+                raise ValueError(error_msg)
+            else:
+                # Multiple entries, but they all have the same size
+                logger.warning(
+                    f"{name_only}: Found {len(results)} duplicate database entries, "
+                    f"but all have the same size ({sizes[0]}). Consider cleaning up duplicates."
+                )
+                expected_size = sizes[0]
+        else:
+            # Single entry - normal case
+            expected_size = results[0][0]
+
+        # Now check if actual size matches expected size
         if expected_size != actual_size:
-            logger.info(
-                f"{name_only}: expected size={expected_size} != {actual_size}=actual_size"
+            error_msg = (
+                f"{name_only}: Size mismatch! "
+                f"Expected={expected_size} != {actual_size}=actual. "
+                f"Difference={actual_size - expected_size}."
             )
-            match = False
+            logger.error(error_msg)
+            return False
         else:
             # Sizes match
-            match = True
+            logger.info(f"{name_only}: Size check passed ({actual_size} bytes)")
+            return True
     else:
         # Cannot access size information; assume the sizes match.
-        match = True
-    return match
+        logger.debug("Cannot access tar size information; assuming sizes match")
+        return True
 
 
 # FIXME: C901 'extractFiles' is too complex (33)

zstash/hpss_utils.py

@@ -144,25 +144,16 @@ def add_files(
             tar_md5: Optional[str] = tarFileObject.md5()
             tarFileObject.close()
             logger.info(f"{ts_utc()}: (add_files): Completed archive file {tfname}")
-            if not skip_tars_md5:
-                tar_tuple: TupleTarsRowNoId = (tfname, tarsize, tar_md5)
-                logger.info("tar name={}, tar size={}, tar md5={}".format(*tar_tuple))
-                if not tars_table_exists(cur):
-                    # Need to create tars table
-                    create_tars_table(cur, con)
-                cur.execute("insert into tars values (NULL,?,?,?)", tar_tuple)
-                con.commit()
 
             # Transfer tar to HPSS
             if config.hpss is not None:
                 hpss: str = config.hpss
             else:
                 raise TypeError("Invalid config.hpss={}".format(config.hpss))
 
-            # NOTE: These lines could be added under an "if debug" condition
-            # logger.info(f"{ts_utc()}: CONTENTS of CACHE upon call to hpss_put:")
-            # process = subprocess.run(["ls", "-l", "zstash"], capture_output=True, text=True)
-            # print(process.stdout)
+            logger.info(
+                f"Contents of the cache prior to `hpss_put`: {os.listdir(os.path.join(cache, tfname))}"
+            )
 
             logger.info(
                 f"{ts_utc()}: DIVING: (add_files): Calling hpss_put to dispatch archive file {tfname} [keep, non_blocking] = [{keep}, {non_blocking}]"
@@ -172,7 +163,33 @@ def add_files(
                 f"{ts_utc()}: SURFACE (add_files): Called hpss_put to dispatch archive file {tfname}"
             )
 
-            # Update database with files that have been archived
+            if not skip_tars_md5:
+                tar_tuple: TupleTarsRowNoId = (tfname, tarsize, tar_md5)
+                logger.info("tar name={}, tar size={}, tar md5={}".format(*tar_tuple))
+                if not tars_table_exists(cur):
+                    # Need to create tars table
+                    create_tars_table(cur, con)
+                # We're done adding files to the tar.
+                # And we've transferred it to HPSS.
+                # Now we can insert the tar into the database.
+                cur.execute("SELECT COUNT(*) FROM tars WHERE name = ?", (tfname,))
+                if cur.fetchone()[0] == 0:
+                    # Typical case
+                    # Doesn't exist - insert new
+                    cur.execute("INSERT INTO tars VALUES (NULL,?,?,?)", tar_tuple)
+                else:
+                    # Unusual case
+                    # Exists - update with new size and md5
+                    logger.warning(
+                        f"Possible database corruption. Updated existing tar {tfname} with new size {tarsize}"
+                    )
+                    cur.execute(
+                        "UPDATE tars SET size = ?, md5 = ? WHERE name = ?",
+                        (tarsize, tar_md5, tfname),
+                    )
+                con.commit()
+
+            # Update database with the individual files that have been archived
             # Add a row to the "files" table,
             # the last 6 columns matching the values of `archived`
             cur.executemany("insert into files values (NULL,?,?,?,?,?,?)", archived)