MAG L1C: deliver the previous day's L1C for day-boundary timeline continuity

[sapols]

Summary

Adds custom_behavior/mag.py: the MAG L1C jobs (norm-mago / norm-magi) fetch the previous day's L1C - the job's own output product - as an extra optional input at input-collection time. This is the delivery side of the day-boundary timeline continuity implemented in IMAP-Science-Operations-Center/imap_processing#3323 (the "time inheritance" cases from SDC Data Validation 4.3.3; tracked in imap_processing#3324 item 1). Implemented as a custom behavior override modeled on Hi's, per @tmplummer's and @bryan-harter's review direction.

How it works

MagL1CJob overrides get_science_files_inputs: the base method collects the current day's L1B inputs as usual, then the previous day's partition of the job's own output asset is queried (get_all_files_in_time_range over [target_start - 1 day, target_start)) and appended as an extra ScienceInput when it exists. Absent a previous-day L1C, the job runs with the current day alone - the processing side treats the input as optional.

Why the dependency is not declared in the yaml

Bryan's suggested experiment was run before writing this (details in the review thread): a declared self-dependency makes Dagster raise DagsterInvalidDefinitionError at import time - self-deps are only allowed for time-partitioned assets, and these assets use dynamic partitions. Probing further showed a tolerated edge would misbehave anyway: input selection does not consult Dagster edges, so a declared self-input would fetch the current day's own earlier L1C (wrong file, and a reprocessing hazard), and with trigger_job: true the job's own output would re-trigger its own partition indefinitely. The strict partition-overlap check in get_all_files_in_time_range cannot match the current day's own partition, and a test pins that. The previously staged date_range: ["-1d", "0d"] placeholder is removed (it was never applied to input selection, and the trigger sensor's range parser fails on negative values).

Known limitations (deliberate, documented in the code)

• Bounded waiting: if day N-1's L1C is in flight (sensor or backfill run) or expected (day N-1 has L1B data but no L1C and no finished L1C run), day N retries every 5 minutes - inheriting run_job's RetryRequested budget (10 retries) - and proceeds without it on the final retry. The wait can delay a run but never fail one, and a day whose previous day provably has nothing coming does not wait at all, so bootstrapping and data-gap days behave as before.
• Reprocessing: backfills run unordered, so a reprocessed day N can inherit the previous generation's N-1 L1C - acceptable per MAG when the earlier version was complete (see the review discussion). When day N-1's rerun is concurrently in flight, the wait above orders them.
• No retrigger: day N is not re-run when day N-1's L1C lands later (no declared edge). If wanted later, a build_sensor override can add it without a yaml edge - with the tradeoff that every day's rerun would invalidate the next.

Tests

tests/orchestration/test_mag.py, mirroring test_hi.py: registration for both sensors (regression against silent fallback to the generic handler), previous-day L1C delivered alongside current-day L1Bs while the current day's own earlier L1C is never selected, single-day processing when no previous-day L1C exists, and the wait behavior: in-flight sensor and backfill runs in all three run states, the expected-but-not-yet-requested window, the neighboring day N-2 partition, days with nothing coming, runs that must not count (other jobs', day N's own), the final-retry fallback, and the base ancestor check still applying.

Notes

• The earlier approval predates the implementation (the PR was a yaml placeholder then).
• History: staged as draft MAG L1C: depend on previous day's normal-mode L1B for time inheritance (#2925) sapols/sds-data-manager#1, then held while the orchestration approach was worked out on this PR's threads.

Remaining Tasks

• Retitle this PR when the past-day delivery design lands (done - was "previous day's normal-mode L1B", now reflects the previous-day L1C delivery).

[sapols]

The cdk-tests failure here is unrelated to this PR. Documenting what's actually failing since it'll probably bite other PRs too (@laspsandoval, tagging you since it's an I-ALiRT test):

tests/lambda_endpoints/test_ialirt_coverage.py::test_lambda_handler makes a live imap_data_access.download() call during CI and got a 404 for imap_ialirt_outages_20260708_v001.json (the next day's outages file). The pattern from the run history:

• Runs earlier in the day (~15:00 UTC, other PRs) passed — they requested the 20260707 file, which existed.
• My runs at 20:22, 20:54, and 20:59 UTC all failed — by then the test's date logic had rolled over to requesting 20260708, which didn't exist yet in the data store CI hits (sds-data-593025701104). Two reruns confirmed it's deterministic, not flaky.

So any PR whose CI runs in the late-UTC-day window fails on this until the next daily outages upload lands. I'll re-run the job tomorrow morning and expect it to go green; the PR itself is reviewable in the meantime.

Two things probably worth fixing in the test:

1. It isn't marked network, so pytest -m "not network" in the cdk-tests job still runs it despite the live download. Marking it (or mocking the download) would decouple CI from the daily upload schedule.
2. The date logic requests the next day's file partway through the current UTC day; possibly a timezone or +1 day buffer thing worth a look.

[laspsandoval]

@sapols I fixed the I-ALiRT test. Thanks for the heads up!

[sapols]

@laspsandoval You're welcome! Thanks for the quick fix 👍

[sapols]

@tech3371 @tmplummer @laspsandoval Can I ask at least one of you to review this PR please? Maxine told me she will not have time to before she leaves IMAP.

[tech3371]

The syntax looks good and helpful comments. LGTM!

[tmplummer]

This is the appropriate change that I had in mind, but I don't think that this has been implemented generically in the current orchestration code. @bryan-harter can you confirm?

If I am correct, that means that this ticket will need to also add custom handling similar to what has been done for Hi in https://github.com/IMAP-Science-Operations-Center/sds-data-manager/tree/dev/sds_data_manager/orchestration/custom_behavior.

[sapols]

Unfortunately I've confirmed this is correct. I added a big note to the top of this PR saying not to merge it. @tech3371 could you maybe please revoke your approval since merging this as-is would break things?

I'd still like confirmation (from @bryan-harter?) about how to move forward. Should I add custom handling similar to what was done for Hi? Or do you all have a more generic solution in mind?

[sapols]

Additional details from Claude:

It's a bit worse than that. I went through the orchestration code on dev:

• date_range is parsed into DependencyNode.dependency_query_time_range, but the base
  IMAPJobHandler.get_science_files_inputs only ever queries the job's own partition
  window - the range is never applied to input selection. Hi's wider queries work because
  custom_behavior/hi.py overrides that method.
• The one generic consumer is the trigger sensor (imap_job.py, int(date_range[0][0])),
  which parses just the first character - fine for "8np", but a negative value like our
  "-1d" would raise ValueError in the sensor.

So MAG would need a custom_behavior/mag.py along the lines of Hi's.

(There's also a wrinkle now that MAG wants the previous-day input to be L1C, i.e. the job consuming its own previous-day output; see Alastair's comment below and imap_processing#3324 item 1, where I've written up all of these findings).

[tmplummer]

I'd still like confirmation (from @bryan-harter?) about how to move forward. Should I add custom handling similar to what was done for Hi? Or do you all have a more generic solution in mind?

I know that is not me, but IMO, yes you will need to implement custom behavior for this.

[bryan-harter]

Yeah Tim is correct, you'd need to implement this date range behavior. At some point this was implemented briefly. However, the unfortunate issue is that we couldn't make the logic generic enough to actually do what we needed. If we tried to make a "generic" date_range handler, it would just be full of "if hi-goodtimes do this" and "if mag do that", which is something that we wanted to avoid. So it would be best to override the behavior of mag-l1c in a similar was that we do for hi.py in the custom_behavior area.

[alastairtree]

Rather than depending on the previous days L1B file I think this should depend on the previous days L1C file instead so that the current days L1C being generated can do data continuity in the L1C product.
In summary, to generate day N L1C MAGo you need

• Day N L1B burst MAGo
• Day N L1B norm MAGo
• Day N-1 L1C MAGo (if it exists)
  MAGi should follow the same pattern.

[sapols]

Thanks for giving MAG's perspective. I updated item 1 in #3324 to capture this.

Since this PR can't merge until the orchestration can deliver past-day inputs at all (see the thread above), I've left the yaml documenting the current L1B intent with a do-not-merge note; the switch to the previous day's L1C will come after we decide how to implement that orchestration.

UPDATE: the processing side has now made this switch. imap_processing#3323 (8f49ade5) takes the previous day's L1C as the timeline source. Only this yaml/delivery side still waits on the orchestration decision.

[tmplummer]

Oh, interesting. This will be a test for Dagster. I am not aware of any other products that depend on themselves in this way.

[bryan-harter]

Yeah I don't know what dagster would do if it was given itself as a dependency. I know it works when the assets are partitioned using the special time partitioning that dagster has, but our assets don't use those because you can't specify extremely custom time ranges.

You can try adding a dependency to itself, and running the unit tests to see if dagster throws an error. If it doesn't, I think we'll be fine, and you can add in something in custom_behavior.

If it does error though, then I'm not sure what the solution would be. The most straightforward solution would be to just leave it out of the YAML files and make sure the custom behavior can find it. But I'm hoping there could be some way to make it work by tinkering around with things.

[sapols]

@bryan-harter Thanks! I ran the experiment you suggested before writing anything, and it fails hard: Dagster rejects a declared self-dependency at import time (details below). So I went with your "most straightforward solution": nothing declared in the yaml, and a custom_behavior/mag.py modeled on hi.py that fetches the previous day's L1C at input-collection time. Absent a previous-day L1C, the job proceeds with the current day alone (MAG's processing side treats the input as optional).

Details, from Claude:

Adding mag/l1c/norm-mago as an input of its own job makes Dagster raise DagsterInvalidDefinitionError: Asset "mag_l1c_normmago" depends on itself at import time — the orchestration test suite can't even collect. Dagster only allows self-deps for time-partitioned assets with negative-offset partition mappings, which matches your memory: our dynamic partitions rule it out.

I also probed the tinkering routes: attaching a negative-offset TimeWindowPartitionMapping to an AssetDep sneaks past definition-time validation, but raises ParameterCheckError whenever anything resolves the mapping (reprocessing backfills do), so it just defers the failure. And even if the edge were tolerated, it wouldn't help: input selection doesn't consult Dagster edges, so a declared self-input would fetch the current day's own earlier L1C (wrong file, and a reprocessing hazard), and with trigger_job: true the job's own output materialization re-triggers its own partition — each rerun changes the dependency hash, so it resubmits and version-bumps forever.

The custom_behavior form avoids that hazard: the strict partition-overlap check in get_all_files_in_time_range can't match the job's own day, and there's a test pinning that.

Dismissing this early approval: it reviewed the yaml-placeholder version; the PR now contains the actual custom_behavior implementation and needs fresh review.

[sapols]

@alastairtree @bryan-harter @tmplummer This raises one behavior question from the delivery design, with a science half and an infra half. Reprocessing backfills currently run in no particular date order, so a reprocessed day N can re-run before day N−1's re-run finishes; the regenerated day N then inherits the previous generation's day N−1 L1C timeline (or none, if N−1 hasn't rerun yet). Forward daily processing is unaffected; N−1's L1C normally exists well before day N's L1Bs arrive.

For @alastairtree: is it acceptable for a reprocessed day to continue its timeline from an older version of the previous day's L1C, or should it only ever inherit from the freshly reprocessed one?

For @bryan-harter / @tmplummer: if it's not acceptable, I see two options:

1. Run MAG L1C backfills in date order. Is that feasible with the current reprocessing tooling?
2. Have the job wait when day N−1's L1C job is INPROGRESS (the pattern Hi already uses).

Which would you guys prefer?

[alastairtree]

In general it is easier if the jobs run in time order at least once (or in the first run) so that you always get a contiguous series of timestamps in the downsampled NM stream. However if the previous day has a full dataset (all the days data had been downlinked) in the earlier version then the timeline should already be complete/correct and day N can run with the earlier version and get the same results.

Not sure how you do the job triggering but if a given pass downlinks data for 3 days of MAG data, I would just run those sequentially as chances are that is the first run for those days. And if you are manually reprocessing a month of data because some dependency like SPICE changed then it probably does not matter and just do it any which way.

[tmplummer]

Just a couple of comments about some potential minor improvements.

[tmplummer]

I'm not sure what exactly a 'backfill' is. Is this for filling in missing data?

[sapols]

Yeah sorry, Dagster jargon like I mentioned below. I'll reword the docstring to avoid the term.

[tmplummer]

This comment is a bit hard to understand. Is it justifying passing the current day's target_start as the end_dt to get_all_files_in_time_range()?

[sapols]

Yes exactly. The window runs from target_start − 1 day up to but not including target_start, and the strict overlap check means the current day's own partition (which starts at target_start) can't match. So the job can't get its own output as input. I'll reword the comment to say that more plainly.

[tmplummer]

I'm not totally clear about why this is building a list and handling a list of files. Isn't it just getting a single previous day L1C file? If it returns more than one file, is that an error?

[sapols]

This mirrors hi.py (lines 84–90): get_all_files_in_time_range returns one metadata entry per materialization event, and file_names can be a string or a list. Usually that's one file, but the previous day can have multiple materializations (e.g. it was reprocessed), so more than one isn't an error. The set() below dedupes.

[tmplummer]

The hi code is retrieving across multiple partitions, so it makes sense to return a list. For this code, it seems like it could be simplified since it only wants a single asset/file.

• It looks to me like get_all_files_int_time_range is returning one metadata entry per partition: https://github.com/sapols/sds-data-manager/blob/dev/sds_data_manager/orchestration/types.py#L394
• The set() below is acting on filename strings right? So it won't de-dup since each filename will have a unique version.

[sapols]

Ah, crap you're totally right. I'm sorry and thank you for pushing back. This was a case where I trusted Claude's read of the situation, but after your last comment it gave me the whole spiel of "you're right, I was wrong, I generalized from Hi's calling pattern without reading enough."

I know I should verify more myself. It's just hard because I'm so out of my depth with this ticket.

Simplified this to extract the single file directly in commit 19470ce. I added a comment that's maybe longer than necessary, but it explains it at a level that makes sense to me. Look better now?

[tmplummer]

For @bryan-harter / @tmplummer: if it's not acceptable, I see two options:

1. Run MAG L1C backfills in date order. Is that feasible with the current reprocessing tooling?

2. Have the job wait when day N−1's L1C job is INPROGRESS (the pattern Hi already uses).

Which would you guys prefer?

I'm not sure what a "backfill" is. Is that MAG specific terminology?

I'm not sure if just prioritizing jobs would work. I don't know for sure, but does Dagster pre-fetch dependencies and then just run the jobs sequentially? It would need to wait until the previous day's job finishes and then get the dependencies for prioritizing by date order to work.

I don't know enough about Dagster to know if you could check for a day n-1 job that is INPROGRESS and somehow make day n's L1C job dependent on the output of that.

[sapols]

@tmplummer "backfill" is Dagster jargon: launching runs for a batch of dates at once (our reprocessing.py submits a PartitionBackfill). The point is just that those runs don't currently happen in date order.

Given Alastair's answer above, I think we can keep this PR as-is; use the latest previous-day L1C that exists, never wait. First-time processing is naturally in date order since data arrives day by day, and Alastair says out-of-order reprocessing is fine (as long as the earlier version of the previous day was complete).

To your Dagster question: my option 2 above wouldn't need Dagster support. hi.py already does it; at input-collection time it queries the processing-jobs database for in-progress neighbors and holds off rather than use a stale file. We could do the same for MAG's previous day, but given Alastair's answer I don't think we need to.

[tmplummer]

First-time processing is naturally in date order since data arrives day by day, and Alastair says out-of-order reprocessing is fine (as long as the earlier version of the previous day was complete).

That would be fine if we got files every day. But due to downlinks ~3-times/week, we actually receive batches of data that span 2-3 days. So, to say that data arrives day-by-day is a bit misleading. In fact, there will be a race condition when we receive a new batch of MAG data.

To your Dagster question: my option 2 above wouldn't need Dagster support. hi.py already does it; at input-collection time it queries the processing-jobs database for in-progress neighbors and holds off rather than use a stale file. We could do the same for MAG's previous day, but given Alastair's answer I don't think we need to.

The reason that this is different than Hi, is because Hi is set up to trigger processing across partitions. So, if the Hi check sees that partition n-1 is INPROGRESS, it is OK to just skip running the job since it knows that when that job finishes, it will re-trigger the partition n job again.

[bryan-harter]

Yeah this is a tricky one to implement!

The custom logic you have to grab the previous day seems great. Now I think it is just a matter of ensuring that processing runs in order, for the reasons that Tim said. There may be some custom logic that you need to add to the MAG l1c kickoff_sensor, to make it so that we only process a single file at a time (whichever one has the earliest timestamp). And then maybe have the asset code check and see if something is already running, and if so, wait?

[sapols]

@bryan-harter no kidding!!




@tmplummer oh man thanks again. I didn't realize downlinks are only ~3-times/week nor that we receive data in batches like that. So the race would happen like every downlink rather than being a rare edge case. Definitely changes things...

I went back to the drawing board and I think I have a clearer picture. I see why Hi can safely skip but MAG can't (because it has no re-trigger on its own L1C). I'm seeing two options to move forward now:

1. Wait-and-retry: day N's job retries every 5 minutes until day N-1’s L1C exists, then runs with it; or proceeds without it if day N-1 clearly has nothing coming (basically Bryan's "check if something is already running and wait").
2. Re-trigger: day N runs immediately, then re-runs once day N-1’s L1C lands (would need new sensor logic).

(or both, like Hi)

I strongly prefer option 1: it reuses the retry machinery every job already has, and the first file published would already be correct. With option 2, every raced day would get a second version, and each new version would re-trigger L1D/L2 downstream.

One question for you both: would a downlink ever deliver day N before day N-1 has arrived? That's the one case option 1 can't fix (day N would process alone until a reprocess). If that can actually happen, we could add option 2 on top later, like how Hi does both (at the cost of lots of new code).

On a personal note: The scope of this ticket has ballooned so much since it was first assigned to me (pitched as a simple one-liner like date_range: ["-1d", "0d"]) and I'm really struggling to keep up with the complexity and finally get this PR over the finish line. Your reviews have been invaluable, @tmplummer and @bryan-harter, so seriously thanks

[sapols]

Pushed the option 1 implementation: bdd1b3b. What you'll find:

• MagL1CJob now overrides _check_for_running_dependencies (the same check every job already runs before RetryRequested). The sensor is untouched.
• It reports a running dependency while day N-1’s L1C is in flight (sensor run or backfill run) or expected (day N-1 has L1B data but no L1C and no finished L1C run). That second case covers the window before the sensor has even requested day N-1’s job.
• Only on the final retry does it proceed without the previous day—the wait can delay a run but never fail one. And if day N-1 provably has nothing coming, there's no wait at all.


I unit tested the ever-living heck out of it (15 new tests) to try to prove it handles every edge case I could think of:

• in-flight runs in all three states,
• backfill runs (in flight and finished),
• day N-1's L1Bs existing before its L1C job is even requested,
• day N-1's L1C already existing (no wait),
• the day N-2 midnight neighbor,
• days with nothing coming (no data, no partitions at all, or a job that already ran and came up empty),
• runs that must not count (another job's, or day N's own),
• the give-up-on-last-retry path,
• and the base ancestor check still applying on top

One last question: the wait inherits the base class's 10 retries × 5 minutes (≤50 min total). Is that enough for a multi-day batch's L1B chain to clear, or should I bump it?

Ready for review again. Please let me know how it looks now.