deadletter queue (#1180)

* deadletter queue

* fix

* ruff

* Update paths.py

Author: bpblanken

v03_pipeline/bin/pipeline_worker.py

@@ -19,6 +19,10 @@
     safe_post_to_slack_failure,
     safe_post_to_slack_success,
 )
+from v03_pipeline.lib.paths import (
+    loading_pipeline_deadletter_queue_dir,
+    loading_pipeline_deadletter_queue_path,
+)
 
 logger = get_logger(__name__)
 
@@ -73,6 +77,9 @@ def process_queue(local_scheduler=False):
                 prr,
                 e,
             )
+            os.makedirs(loading_pipeline_deadletter_queue_dir(), exist_ok=True)
+            with open(loading_pipeline_deadletter_queue_path(run_id), 'w') as f:
+                f.write(prr.model_dump_json())
     finally:
         if latest_queue_path is not None and os.path.exists(latest_queue_path):
             os.remove(latest_queue_path)

v03_pipeline/bin/pipeline_worker_test.py

@@ -6,7 +6,10 @@
 
 from v03_pipeline.bin.pipeline_worker import process_queue
 from v03_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType
-from v03_pipeline.lib.paths import loading_pipeline_queue_dir
+from v03_pipeline.lib.paths import (
+    loading_pipeline_deadletter_queue_dir,
+    loading_pipeline_queue_dir,
+)
 from v03_pipeline.lib.test.mock_complete_task import MockCompleteTask
 from v03_pipeline.lib.test.mocked_dataroot_testcase import MockedDatarootTestCase
 
@@ -92,3 +95,10 @@ def test_process_failure(
         mock_safe_post_to_slack.assert_called_once_with(
             ':failed: Pipeline Runner Request Failed :failed:\nRun ID: 20250918-200704-123456\n```{\n    "callset_path": "v03_pipeline/var/test/callsets/1kg_30variants.vcf",\n    "dataset_type": "SNV_INDEL",\n    "project_guids": [\n        "project_a"\n    ],\n    "reference_genome": "GRCh38",\n    "request_type": "LoadingPipelineRequest",\n    "sample_type": "WGS",\n    "skip_check_sex_and_relatedness": false,\n    "skip_expect_tdr_metrics": false,\n    "skip_validation": false\n}```\nReason: there were failed tasks',
         )
+        with open(
+            os.path.join(
+                loading_pipeline_deadletter_queue_dir(),
+                'request_20250918-200704-123456.json',
+            ),
+        ) as f:
+            self.assertEqual(json.load(f)['request_type'], 'LoadingPipelineRequest')

v03_pipeline/lib/paths.py

@@ -410,6 +410,13 @@ def loading_pipeline_queue_dir() -> str:
         'loading_pipeline_queue',
     )
 
+# https://en.wikipedia.org/wiki/Dead_letter_queue
+def loading_pipeline_deadletter_queue_dir() -> str:
+    return os.path.join(
+        Env.LOCAL_DISK_MOUNT_PATH,
+        'loading_pipeline_deadletter_queue',
+    )
+
 
 def loading_pipeline_queue_path(run_id: str) -> str:
     """
@@ -421,6 +428,16 @@ def loading_pipeline_queue_path(run_id: str) -> str:
     )
 
 
+def loading_pipeline_deadletter_queue_path(run_id: str) -> str:
+    """
+    Returns a new path for a loading pipeline queue request file.
+    """
+    return os.path.join(
+        loading_pipeline_deadletter_queue_dir(),
+        f'request_{run_id}.json',
+    )
+
+
 def pipeline_run_success_file_path(
     reference_genome: ReferenceGenome,
     dataset_type: DatasetType,