A few improvements to dagster

sds_data_manager/orchestration/config.py

@@ -11,22 +11,22 @@
 FIRST_MAP_START_DATE = datetime.datetime(2026, 1, 17, tzinfo=datetime.timezone.utc)
 
 sensor_schedules = {
-    "l0": 900,
-    "l1": 900,
-    "l1a": 900,
-    "l1b": 900,
-    "l1c": 900,
-    "l1d": 900,
-    "l2": 900,
-    "l2a": 900,
-    "l2b": 900,
-    "l2c": 900,
-    "l2d": 900,
-    "l3": 900,
-    "l3a": 900,
-    "l3b": 900,
-    "l3c": 900,
-    "l3d": 900,
+    "l0": 300,
+    "l1": 300,
+    "l1a": 300,
+    "l1b": 300,
+    "l1c": 300,
+    "l1d": 300,
+    "l2": 300,
+    "l2a": 300,
+    "l2b": 300,
+    "l2c": 300,
+    "l2d": 300,
+    "l3": 300,
+    "l3a": 300,
+    "l3b": 300,
+    "l3c": 300,
+    "l3d": 300,
 }
 
 

sds_data_manager/orchestration/imap_job.py

@@ -15,6 +15,7 @@
 from botocore.exceptions import ClientError
 from dagster import (
     AssetExecutionContext,
+    AssetObservation,
     AssetOut,
     DagsterEventType,
     DagsterRunStatus,
@@ -126,7 +127,7 @@ def __init__(self, job: ProcessingJobNode):
         job : ProcessingJobNode
             The job node to process.
         """
-        self.BATCH_JOB_TIMEOUT_SECONDS = 3600  # 1 hour, can be adjusted as needed.
+        self.BATCH_JOB_TIMEOUT_SECONDS = 18000  # 5 hours
         self.WAIT_TIME_AFTER_BATCH_SECONDS = (
             60  # time to wait after a batch job completes to search for files.
         )
@@ -221,6 +222,15 @@ def run_job(
                 )
             except MissingDependenciesError as e:
                 context.log.info(f"Skipping job: {e}")
+                for output in self.job_config.outputs:
+                    yield AssetObservation(
+                        asset_key=output.to_dagster_asset(),
+                        partition=context.partition_key,
+                        metadata={
+                            "status": "Skipped - Missing Dependencies",
+                            "missing_files": str(e),
+                        },
+                    )
                 return SkipReason(str(e))
 
             context.log.info(
@@ -1293,35 +1303,6 @@ def try_to_submit_job(
         if repoint is not None:
             batch_command.extend(["--repointing", f"repoint{repoint:05d}"])
 
-        # We will check here if this job has already failed with these
-        # exact dependencies
-        conditions = [
-            models.ProcessingJob.instrument == self.job_config.source,
-            models.ProcessingJob.data_level == self.job_config.data_type,
-            models.ProcessingJob.descriptor == self.job_config.descriptor,
-            models.ProcessingJob.dependency_hash == dep_hash,
-            models.ProcessingJob.start_date == start_date.date(),
-        ]
-        conditions.append(models.ProcessingJob.status.in_([models.Status.FAILED.value]))
-        if repoint is not None:
-            conditions.append(models.ProcessingJob.repointing == repoint)
-
-        already_failed_job = (
-            session.query(models.ProcessingJob)
-            .filter(*conditions)
-            .order_by(
-                models.ProcessingJob.major_version.desc(),
-                models.ProcessingJob.minor_version.desc(),
-            )
-            .first()
-        )
-        if already_failed_job:
-            return BatchJobSubmit(
-                status="failed",
-                message="""This exact job has been submitted previously,
-                           and has already failed. No need to run it again.""",
-            )
-
         # Get the necessary AWS information
         # NOTE: These are here for easier mocking in tests rather than at the
         # module level

tests/orchestration/test_spacecraft.py

@@ -14,6 +14,7 @@
 import imap_data_access
 import pytest
 from dagster import (
+    AssetObservation,
     Failure,
     build_asset_context,
     build_sensor_context,
@@ -136,9 +137,9 @@ def test_spacecraft_l1a_no_spice(mock_db_session, ephemeral_instance):
         )
     )
 
-    # Verify nothing has happened still, because SPICE is still missing.
+    # Verify that we have only returned an asset observation
     yielded_files = list(spacecraft_l1a_job.run_job(context, 1, 1))
-    assert len(yielded_files) == 0
+    assert isinstance(yielded_files[0], AssetObservation)
 
 
 def test_spacecraft_l1a_submits(