Partial download status (#597)

* Implement partial success

Author: oskarhurst

api/models/download.go

@@ -80,6 +80,7 @@ type Extent struct {
 // PackagerContentItem is a single item for Packager to include in output file
 // Note: Previously called DownloadContentItem
 type PackagerContentItem struct {
+	ProductID   string `json:"product_id" db:"product_id"`
 	Bucket      string `json:"bucket"`
 	Key         string `json:"key"`
 	DssDatatype string `json:"dss_datatype" db:"dss_datatype"`
@@ -146,6 +147,7 @@ func GetDownloadPackagerRequest(db *pgxpool.Pool, downloadID *uuid.UUID) (*Packa
 		context.Background(), db, &pr,
 		`WITH download_contents AS (
 			SELECT download_id,
+			       product_id,
 			       key,
 		           bucket,
 			       dss_datatype,
@@ -168,6 +170,7 @@ func GetDownloadPackagerRequest(db *pgxpool.Pool, downloadID *uuid.UUID) (*Packa
 		    	)
 		    UNION
 		    SELECT download_id,
+			       product_id,
 			       key,
 		           bucket,
 		           dss_datatype,
@@ -215,6 +218,7 @@ func GetDownloadPackagerRequest(db *pgxpool.Pool, downloadID *uuid.UUID) (*Packa
 			SELECT download_id,
 			       jsonb_agg(
 					   jsonb_build_object(
+						   'product_id',   product_id,
 						   'key',          key,
 						   'bucket',       bucket,
 						   'dss_datatype', dss_datatype,

async_packager/packager.py

@@ -34,6 +34,7 @@
 
 
 def handle_message(message):
+    package_file = None
     try:
         logger.info("%(spacer)s new message %(spacer)s" % {"spacer": "*" * 20})
 
@@ -69,11 +70,23 @@ def handle_message(message):
             # TODO: Add new package_status in database to represent EMPTY condition
             logger.info(f'Empty Contents: No products selected in the request for download ID "{download_id}"')
         else:
-            package_file = handler.handle_message(PayloadResp, dst.name)
+            writer_result = handler.handle_message(PayloadResp, dst.name)
+
+            if writer_result:
+                package_file = writer_result["file"]
+                product_stats = writer_result["product_stats"]
+
+                # Determine status from per-product stats
+                total_expected = sum(ps["expected"] for ps in product_stats.values())
+                total_successful = sum(ps["successful"] for ps in product_stats.values())
+
+                if total_successful == total_expected:
+                    status_key = "SUCCESS"
+                else:
+                    status_key = "PARTIAL_SUCCESS"
 
-            if package_file:
                 # Upload File to S3
-                logger.debug(f'Packaging successful for download ID "{download_id}"')
+                logger.debug(f'Packaging {status_key.lower()} for download ID "{download_id}" ({total_successful}/{total_expected} files)')
                 t1 = Timer(logger=None)
                 t1.start()
                 s3_upload_worked = s3_upload_file(
@@ -86,13 +99,15 @@ def handle_message(message):
                     )
                     handler.update_status(
                         download_id,
-                        handler.PACKAGE_STATUS["SUCCESS"],
+                        handler.PACKAGE_STATUS[status_key],
                         100,
                         PayloadResp.output_key,
-                        # Manifest JSON
+                        # Manifest JSON with per-product stats
                         {
                             "size_bytes": os.path.getsize(package_file),
-                            "filecount": len(PayloadResp.contents),
+                            "filecount": total_expected,
+                            "filecount_successful": total_successful,
+                            "product_stats": product_stats,
                         },
                     )
                 else:
@@ -101,7 +116,7 @@ def handle_message(message):
                     )
             else:
                 logger.critical(
-                    f'Failed to package or upload "{package_file}" to S3 download ID "{download_id}"'
+                    f'Failed to package or upload to S3 download ID "{download_id}"'
                 )
 
     except Exception as ex:

async_packager/src/cumulus_packager/packager/handler.py

@@ -26,6 +26,7 @@
 PACKAGE_STATUS = {
     "FAILED": "a553101e-8c51-4ddd-ac2e-b011ed54389b",
     "INITIATED": "94727878-7a50-41f8-99eb-a80eb82f737a",
+    "PARTIAL_SUCCESS": "b9e8fac1-0cd4-4645-91b2-36b83524549f",
     "SUCCESS": "3914f0bd-2290-42b1-bc24-41479b3a846f",
 }
 

async_packager/src/cumulus_packager/writers/dss7.py

@@ -6,7 +6,7 @@
 import os
 import sys
 import threading
-from collections import namedtuple
+from collections import defaultdict, namedtuple
 from pathlib import Path
 from queue import Queue, Empty
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -200,6 +200,9 @@ def process_single_tiff_gdal(args):
     (idx, tif, _bbox, cellsize, destination_srs, grid_type, grid_type_name,
      srs_definition, _extent_name, tz_name, tz_offset, is_interval) = args
 
+    # Extract product_id early for error reporting
+    product_id = tif.get('product_id', 'UNKNOWN')
+
     try:
         TifCfg = namedtuple("TifCfg", tif)(**tif)
         s3_path = f"/vsis3_streaming/{TifCfg.bucket}/{TifCfg.key}"
@@ -210,6 +213,7 @@ def process_single_tiff_gdal(args):
             return {
                 'success': False,
                 'index': idx,
+                'product_id': product_id,
                 'error': f"Failed to open {TifCfg.key}"
             }
 
@@ -232,6 +236,7 @@ def process_single_tiff_gdal(args):
             return {
                 'success': False,
                 'index': idx,
+                'product_id': product_id,
                 'error': f"Failed to warp {TifCfg.key}"
             }
 
@@ -309,6 +314,7 @@ def process_single_tiff_gdal(args):
         return {
             'success': True,
             'index': idx,
+            'product_id': product_id,
             'tif_key': TifCfg.key,
             'gd': gd,
             'compressed_data': compressed_data,
@@ -322,6 +328,7 @@ def process_single_tiff_gdal(args):
         return {
             'success': False,
             'index': idx,
+            'product_id': product_id,
             'error': str(e)
         }
 
@@ -373,12 +380,19 @@ def process_tiffs_with_bounded_queue(src, _bbox, cellsize, destination_srs, dss,
 
     Returns:
     --------
-    int : Number of successfully processed files
+    tuple : (processed_count, product_stats) where processed_count is total files
+            attempted and product_stats is {product_id: {"expected": N, "successful": M}}
     """
     # Auto-detect optimal number of workers if not provided
     if max_workers is None:
         max_workers = get_optimal_workers()
 
+    # Build expected counts per product
+    expected_counts = defaultdict(int)
+    for tif in src:
+        expected_counts[tif.get('product_id', 'UNKNOWN')] += 1
+    success_counts = defaultdict(int)
+
     try:
         # Estimate bbox dimensions for queue sizing
         first_tif = src[0]
@@ -455,6 +469,7 @@ def producer():
                         compressed_data = result['compressed_data']
                         compressed_size = result['compressed_size']
                         tif_key = result['tif_key']
+                        _product_id = result.get('product_id', 'UNKNOWN')
 
                         # Write precompressed data to DSS (GriddedData already created in worker)
                         t = Timer(name="accumuluated", logger=None)
@@ -464,8 +479,10 @@ def producer():
 
                         if dss_result != 0:
                             logger.warning(f'HEC-DSS-PY write record failed for "{tif_key}": {dss_result}')
-                        elif logger.isEnabledFor(logging.DEBUG):
-                            logger.debug(f'DSS writePrecompressedGrid processed "{tif_key}" in {elapsed_time:.4f}s')
+                        else:
+                            if logger.isEnabledFor(logging.DEBUG):
+                                logger.debug(f'DSS writePrecompressedGrid processed "{tif_key}" in {elapsed_time:.4f}s')
+                            success_counts[_product_id] += 1
 
                         processed_count += 1
 
@@ -489,24 +506,36 @@ def producer():
                         logger.error(f"Error writing to DSS for file {result['index']}: {e}")
                         import traceback
                         logger.error(traceback.format_exc())
+                        processed_count += 1
                         continue
                 else:
                     logger.error(f"Error processing file {result['index']}: {result.get('error', 'Unknown error')}")
+                    processed_count += 1
 
             except Empty:
                 logger.error("Timeout waiting for results from queue")
                 break
 
         producer_thread.join()
 
-        logger.info(f"Parallel GDAL with compression: Successfully processed {processed_count}/{len(src)} files")
-        return processed_count
+        total_successful = sum(success_counts.values())
+        logger.info(f"Parallel GDAL with compression: Successfully wrote {total_successful}/{len(src)} files")
+
+        product_stats = {
+            pid: {"expected": expected_counts[pid], "successful": success_counts.get(pid, 0)}
+            for pid in expected_counts
+        }
+        return processed_count, product_stats
 
     except Exception as e:
         logger.error(f"Parallel GDAL processing failed: {e}")
         import traceback
         logger.error(traceback.format_exc())
-        return 0
+        product_stats = {
+            pid: {"expected": expected_counts[pid], "successful": 0}
+            for pid in expected_counts
+        }
+        return 0, product_stats
 
 
 @pyplugs.register
@@ -537,8 +566,9 @@ def writer(
 
     Returns
     -------
-    str
-        FQPN to dss file
+    dict or None
+        {"file": FQPN to dss file, "product_stats": {product_id: {"expected": N, "successful": M}}}
+        or None if no files were processed
     """
 
     try:
@@ -579,11 +609,17 @@ def writer(
 
     dssfilename = Path(dst).joinpath(id).with_suffix(".dss").as_posix()
 
+    # Build expected counts per product for single-file path
+    expected_counts = defaultdict(int)
+    for tif in src:
+        expected_counts[tif.get('product_id', 'UNKNOWN')] += 1
+    success_counts = defaultdict(int)
+
     with HecDss(dssfilename) as dss:
         # Use parallel GDAL processing with compression and precompressed writes for multiple files
         if len(src) > 1:
             logger.info("Using parallel GDAL processing with compression and precompressed writes")
-            processed_count = process_tiffs_with_bounded_queue(
+            processed_count, product_stats = process_tiffs_with_bounded_queue(
                 src, _bbox, cellsize, destination_srs, dss,
                 grid_type, grid_type_name, srs_definition,
                 _extent_name, tz_name, tz_offset, is_interval,
@@ -662,14 +698,17 @@ def writer(
                     t.start()
                     result = dss.put(gd)
                     elapsed_time = t.stop()
-                    if logger.isEnabledFor(logging.DEBUG):
-                        logger.debug(
-                            f'DSS put Processed "{TifCfg.key}" in {elapsed_time:.4f} seconds'
-                        )
                     if result != 0:
                         logger.info(
                             f'HEC-DSS-PY write record failed for "{TifCfg.key}": {result}'
                         )
+                    else:
+                        if logger.isEnabledFor(logging.DEBUG):
+                            logger.debug(
+                                f'DSS put Processed "{TifCfg.key}" in {elapsed_time:.4f} seconds'
+                            )
+                        _product_id = tif.get('product_id', 'UNKNOWN')
+                        success_counts[_product_id] += 1
 
                     _progress = int(((idx + 1) / gridcount) * 100)
                     # Update progress at predefined interval
@@ -698,6 +737,13 @@ def writer(
                     warp_ds = None
                     ds = None
 
+    # Build product_stats for single-file path (multi-file path already has it)
+    if len(src) == 1:
+        product_stats = {
+            pid: {"expected": expected_counts[pid], "successful": success_counts.get(pid, 0)}
+            for pid in expected_counts
+        }
+
     # If no progress was made for any items in the payload (ex: all tifs could not be projected properly),
     # don't return a dssfilename
     if _progress == 0:
@@ -710,4 +756,4 @@ def writer(
         f'Total processing time for download ID "{id}" in {total_time:.4f} seconds'
     )
 
-    return dssfilename
+    return {"file": dssfilename, "product_stats": product_stats}

async_packager/src/cumulus_packager/writers/tgz-cog.py

@@ -4,7 +4,7 @@
 """
 import json
 import os
-from collections import namedtuple
+from collections import defaultdict, namedtuple
 from tarfile import TarFile
 
 import pyplugs
@@ -46,8 +46,9 @@ def writer(
 
     Returns
     -------
-    str
-        FQPN to dss file
+    dict or None
+        {"file": FQPN to tar.gz file, "product_stats": {product_id: {"expected": N, "successful": M}}}
+        or None if packaging failed
     """
     # convert the strings back to json objects; needed for pyplugs
     src = json.loads(src)
@@ -107,4 +108,13 @@ def writer(
         ds = None
         tar.close()
 
-    return tarfilename
+    # All-or-nothing: if we reached here, every file was processed
+    expected_counts = defaultdict(int)
+    for tif in src:
+        expected_counts[tif.get('product_id', 'UNKNOWN')] += 1
+    product_stats = {
+        pid: {"expected": count, "successful": count}
+        for pid, count in expected_counts.items()
+    }
+
+    return {"file": tarfilename, "product_stats": product_stats}

sql/common/V2.65.0__add_partial_success_download_status.sql

@@ -0,0 +1,2 @@
+INSERT INTO download_status (id, name) VALUES
+    ('b9e8fac1-0cd4-4645-91b2-36b83524549f', 'PARTIAL SUCCESS');