From b05224c54996cd580d6076a5cb60f5b56d167b10 Mon Sep 17 00:00:00 2001 From: Tim Beccue Date: Sat, 16 May 2026 19:21:03 -0400 Subject: [PATCH 1/2] Add adaptive smart-stack timeouts --- banzai/dbs.py | 19 ++- banzai/scheduling.py | 2 + banzai/stacking.py | 112 +++++++++++++++ banzai/tests/site_e2e/test_site_e2e.py | 66 ++++++++- banzai/tests/test_smart_stacking.py | 183 ++++++++++++++++++++++++- 5 files changed, 372 insertions(+), 10 deletions(-) diff --git a/banzai/dbs.py b/banzai/dbs.py index b4c6dc69..363ab06d 100755 --- a/banzai/dbs.py +++ b/banzai/dbs.py @@ -139,6 +139,7 @@ class Subframe(SiteBase): is_last = Column(Boolean, default=False) status = Column(String(20), default='active', nullable=False) dateobs = Column(DateTime, nullable=True) + exptime = Column(Float, nullable=True) created_at = Column(DateTime, default=datetime.datetime.utcnow) completed_at = Column(DateTime, nullable=True) __table_args__ = ( @@ -624,7 +625,7 @@ def replicate_instrument(instrument_record, db_address): db_session.commit() -def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, is_last, dateobs): +def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, is_last, dateobs, exptime=None): """Upsert a subframe record. On conflict (moluid, stack_num), reset the row to a fresh active state so a requeue behaves like a retry from scratch.""" if filepath is None: @@ -641,7 +642,7 @@ def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, i stmt = insert(Subframe).values( moluid=moluid, stack_num=stack_num, frmtotal=frmtotal, camera=camera, - filepath=filepath, is_last=is_last, dateobs=dateobs, + filepath=filepath, is_last=is_last, dateobs=dateobs, exptime=exptime, ) stmt = stmt.on_conflict_do_update( index_elements=['moluid', 'stack_num'], @@ -650,6 +651,7 @@ def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, i 'camera': stmt.excluded.camera, 'is_last': stmt.excluded.is_last, 'dateobs': stmt.excluded.dateobs, + 'exptime': stmt.excluded.exptime, 'filepath': stmt.excluded.filepath, 'status': 'active', 'created_at': datetime.datetime.utcnow(), @@ -667,6 +669,19 @@ def get_subframes(db_address, moluid): ).all() +def get_active_subframes_for_camera(db_address, camera): + """Get active subframe records for a camera ordered by stack and arrival time.""" + with get_session(db_address, site_deploy=True) as session: + return session.query(Subframe).filter( + Subframe.camera == camera, + Subframe.status == 'active', + ).order_by( + Subframe.moluid, + Subframe.created_at, + Subframe.stack_num, + ).all() + + def mark_stack_complete(db_address, moluid, status='complete'): """Mark all subframes for a moluid as complete (or timeout).""" now = datetime.datetime.utcnow() diff --git a/banzai/scheduling.py b/banzai/scheduling.py index 43a2b25a..43908cf1 100644 --- a/banzai/scheduling.py +++ b/banzai/scheduling.py @@ -324,6 +324,7 @@ def process_subframe(self, body: dict, runtime_context: dict): camera = header.get('INSTRUME', '').strip() dateobs_str = header.get('DATE-OBS', '') dateobs = parse_date_obs(dateobs_str) if dateobs_str else None + exptime = header.get('EXPTIME') # Phase 1: Run reduction pipeline images = stage_utils.run_pipeline_stages([{'path': filepath}], runtime_context) @@ -346,6 +347,7 @@ def process_subframe(self, body: dict, runtime_context: dict): filepath=reduced_path, is_last=body.get('last_frame', False), dateobs=dateobs, + exptime=exptime, ) # Phase 3: Notify the stack worker that a new subframe is available diff --git a/banzai/stacking.py b/banzai/stacking.py index 28a5af41..36ac695e 100644 --- a/banzai/stacking.py +++ b/banzai/stacking.py @@ -1,8 +1,10 @@ """Smart stacking: worker, supervisor, and helper functions.""" import argparse +import datetime import multiprocessing import signal import time +from collections import defaultdict import redis as redis_lib @@ -17,6 +19,9 @@ # --------------------------------------------------------------------------- REQUIRED_MESSAGE_FIELDS = ('fits_file', 'last_frame', 'instrument_enqueue_timestamp') +INITIAL_FRAME_TIMEOUT_BUFFER_SECONDS = 60.0 +INVALID_BASELINE_FALLBACK_SECONDS = 60.0 +ADAPTIVE_TIMEOUT_MULTIPLIER = 2.0 def validate_message(body): @@ -36,6 +41,91 @@ def check_stack_complete(subframes, frmtotal): return bool(subframes) and (all_arrived or has_last) +def _arrival_ordered_subframes(subframes): + return sorted( + subframes, + key=lambda s: ( + s.created_at or datetime.datetime.max, + s.stack_num or 0, + ), + ) + + +def _exptime_seconds(subframe): + try: + exptime = float(subframe.exptime or 0.0) + except (TypeError, ValueError): + return 0.0 + return max(exptime, 0.0) + + +def adaptive_timeout_baseline_seconds(subframes): + """Return the adjusted first-to-second-frame baseline or invalid-value fallback.""" + ordered_subframes = _arrival_ordered_subframes(subframes) + if len(ordered_subframes) < 2: + return None + + # The first two arrivals establish the expected post-exposure delivery cadence + # for this moluid; later timeout checks compare against this baseline. + exptime = _exptime_seconds(ordered_subframes[0]) + first_arrival = ordered_subframes[0].created_at + second_arrival = ordered_subframes[1].created_at + if first_arrival is None or second_arrival is None: + return INVALID_BASELINE_FALLBACK_SECONDS + + # Subtract exposure duration because it is unrelated to processing/transport + # delay, which is the signal this timeout is trying to detect. + adjusted_gap = (second_arrival - first_arrival).total_seconds() - exptime + # A zero or negative adjusted gap means the timing data is not usable, so + # fall back to a conservative positive baseline instead of timing out instantly. + if adjusted_gap <= 0: + return INVALID_BASELINE_FALLBACK_SECONDS + return adjusted_gap + + +def stack_has_timed_out(subframes, now=None): + """Return True when a stack has exceeded the adaptive arrival timeout.""" + ordered_subframes = _arrival_ordered_subframes(subframes) + if not ordered_subframes: + return False + + now = now or datetime.datetime.utcnow() + # Use the first frame's exposure time for the whole stack; smart-stack subframes + # are expected to have consistent exposure durations. + exptime = _exptime_seconds(ordered_subframes[0]) + first_arrival = ordered_subframes[0].created_at + if first_arrival is None: + return False + + # Until frame 2 arrives, there is no cadence measurement yet, so use the + # first-frame exposure plus a fixed post-exposure buffer. + if len(ordered_subframes) == 1: + first_frame_timeout = exptime + INITIAL_FRAME_TIMEOUT_BUFFER_SECONDS + return (now - first_arrival).total_seconds() > first_frame_timeout + + # After frame 2, the first-to-second adjusted gap defines the expected + # post-exposure arrival cadence; future gaps get twice that allowance. + baseline = adaptive_timeout_baseline_seconds(ordered_subframes) + timeout_threshold = baseline * ADAPTIVE_TIMEOUT_MULTIPLIER + + # Late frames should still cause a timeout even if they eventually arrived; + # the arrival timestamps preserve those delayed adjacent gaps. + for previous_frame, current_frame in zip(ordered_subframes, ordered_subframes[1:]): + if previous_frame.created_at is None or current_frame.created_at is None: + continue + adjusted_gap = (current_frame.created_at - previous_frame.created_at).total_seconds() - exptime + if adjusted_gap > timeout_threshold: + return True + + # If no already-observed gap was too long, check whether the next expected + # frame has now been missing for too long. + latest_arrival = ordered_subframes[-1].created_at + if latest_arrival is None: + return False + missing_frame_gap = (now - latest_arrival).total_seconds() - exptime + return missing_frame_gap > timeout_threshold + + # --------------------------------------------------------------------------- # Notifications # --------------------------------------------------------------------------- @@ -91,6 +181,7 @@ def run_worker_loop(camera, db_address, redis_url, retention_days=30, poll_inter while True: try: process_notifications(db_address, redis_client, camera) + check_timeouts(db_address, camera) dbs.cleanup_old_subframes(db_address, retention_days) time.sleep(poll_interval) except Exception as e: @@ -122,6 +213,27 @@ def finalize_stack(db_address, moluid, status='complete'): logger.debug(f'Mock ingester upload for {moluid}', extra_tags={'moluid': moluid}) +def check_timeouts(db_address, camera, now=None): + """Finalize active stacks that are complete or have exceeded adaptive timeout.""" + active_subframes = dbs.get_active_subframes_for_camera(db_address, camera) + # The DB scan is camera-wide, so regroup rows into independent stack states + # before making terminal-status decisions. + subframes_by_moluid = defaultdict(list) + for subframe in active_subframes: + subframes_by_moluid[subframe.moluid].append(subframe) + + for moluid, subframes in subframes_by_moluid.items(): + frmtotal = subframes[0].frmtotal + # Completion wins over timeout so a fully reduced stack cannot be marked + # partial just because its cadence also exceeded the adaptive threshold. + if check_stack_complete(subframes, frmtotal): + finalize_stack(db_address, moluid, status='complete') + # Incomplete active stacks are finalized only when their arrival cadence + # indicates that the next expected frame is no longer arriving normally. + elif stack_has_timed_out(subframes, now=now): + finalize_stack(db_address, moluid, status='timeout') + + # --------------------------------------------------------------------------- # Supervisor # --------------------------------------------------------------------------- diff --git a/banzai/tests/site_e2e/test_site_e2e.py b/banzai/tests/site_e2e/test_site_e2e.py index 15326081..6985a9c6 100644 --- a/banzai/tests/site_e2e/test_site_e2e.py +++ b/banzai/tests/site_e2e/test_site_e2e.py @@ -5,6 +5,7 @@ import subprocess import time import json +import datetime from pathlib import Path import pytest @@ -349,16 +350,17 @@ def check(): dbs.Subframe.moluid == 'mol-e2e-test' ).all() if subframes and all(s.status == 'complete' for s in subframes): - return [s.filepath for s in subframes] + return subframes return None - filepaths = poll_until(check, timeout=300) - assert filepaths, "Subframe stack did not complete within timeout" + subframes = poll_until(check, timeout=300) + assert subframes, "Subframe stack did not complete within timeout" + assert subframes[0].exptime is not None # Verify the reduced output file exists on disk. # Paths in the DB are now absolute host paths, so check directly. - assert filepaths[0], "Subframe has no filepath after completion" - expected_path = Path(filepaths[0]) + assert subframes[0].filepath, "Subframe has no filepath after completion" + expected_path = Path(subframes[0].filepath) found = poll_until( lambda p=expected_path: p.exists() and p.stat().st_size > 0, @@ -366,6 +368,60 @@ def check(): ) assert found, f"Reduced subframe output not found: {expected_path}" + @pytest.mark.e2e_site_startup + def test_13_subframe_stack_times_out(self, site_deployment): + """Verify the deployed stacking supervisor can mark an incomplete stack timeout.""" + + raw_dir = RAW_DIR + src_path = raw_dir / RAW_FRAME_FILENAME + subframe_path = raw_dir / 'subframe_timeout_test.fits.fz' + moluid = 'mol-e2e-timeout-test' + + assert src_path.exists(), f"Raw frame not found: {src_path}" + shutil.copy2(str(src_path), str(subframe_path)) + + with fits.open(str(subframe_path), mode='update') as hdul: + hdul['SCI'].header['MOLUID'] = moluid + hdul['SCI'].header['MOLFRNUM'] = 1 + hdul['SCI'].header['FRMTOTAL'] = 2 + hdul['SCI'].header['STACK'] = 'T' + + body = json.dumps({ + 'fits_file': str(subframe_path), + 'last_frame': False, + 'instrument_enqueue_timestamp': int(time.time() * 1000), + }) + stack_queue = os.environ.get('STACK_QUEUE_NAME', 'banzai_stack_queue') + publish_raw_string_to_queue(stack_queue, body) + + def row_exists(): + with dbs.get_session(LOCAL_DB_ADDRESS, site_deploy=True) as session: + return session.query(dbs.Subframe).filter( + dbs.Subframe.moluid == moluid + ).first() + + assert poll_until(row_exists, timeout=120, interval=2), \ + "Timed-out subframe row was not inserted" + + stale_created_at = datetime.datetime.utcnow() - datetime.timedelta(hours=2) + with dbs.get_session(LOCAL_DB_ADDRESS, site_deploy=True) as session: + session.execute( + text("UPDATE subframes SET created_at = :created_at WHERE moluid = :moluid"), + {'created_at': stale_created_at, 'moluid': moluid}, + ) + + def timed_out(): + with dbs.get_session(LOCAL_DB_ADDRESS, site_deploy=True) as session: + subframes = session.query(dbs.Subframe).filter( + dbs.Subframe.moluid == moluid + ).all() + if subframes and all(s.status == 'timeout' for s in subframes): + return subframes + return None + + assert poll_until(timed_out, timeout=120, interval=2), \ + "Incomplete subframe stack was not marked timeout" + @pytest.mark.e2e_site_startup def test_14_cache_init_reuses_existing_slot(self, site_deployment): """Verify cache-init succeeds when the replication slot already exists on the publisher.""" diff --git a/banzai/tests/test_smart_stacking.py b/banzai/tests/test_smart_stacking.py index 938581bb..c56b175d 100644 --- a/banzai/tests/test_smart_stacking.py +++ b/banzai/tests/test_smart_stacking.py @@ -13,8 +13,9 @@ from banzai.dbs import insert_subframe, get_subframes, mark_stack_complete, cleanup_old_subframes from banzai.stacking import (validate_message, check_stack_complete, push_notification, pop_notification, REDIS_KEY_PREFIX, - process_notifications, finalize_stack, run_worker_loop, - StackingSupervisor) + process_notifications, finalize_stack, check_timeouts, + stack_has_timed_out, adaptive_timeout_baseline_seconds, + run_worker_loop, StackingSupervisor) from banzai.scheduling import process_subframe from banzai.main import SubframeListener @@ -95,6 +96,18 @@ def test_insert_subframe_and_get_subframes(self, db_address): assert subframe.filepath == '/data/frame1.fits' assert subframe.is_last is False assert subframe.dateobs == dateobs + assert subframe.exptime is None + + def test_insert_subframe_stores_exptime(self, db_address): + dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) + insert_subframe( + db_address, moluid='mol-exp', stack_num=1, frmtotal=5, + camera='cam1', filepath='/data/frame1.fits', is_last=False, dateobs=dateobs, + exptime=30.0, + ) + subframes = get_subframes(db_address, moluid='mol-exp') + assert len(subframes) == 1 + assert subframes[0].exptime == 30.0 def test_insert_subframe_duplicate_resets_state_for_retry(self, db_address): dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) @@ -121,6 +134,23 @@ def test_insert_subframe_duplicate_resets_state_for_retry(self, db_address): assert subframe.is_last is True assert subframe.dateobs == new_dateobs + def test_insert_subframe_duplicate_refreshes_exptime(self, db_address): + dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) + insert_subframe( + db_address, moluid='mol-exp-dup', stack_num=1, frmtotal=3, + camera='cam1', filepath='/data/dup1.fits', is_last=False, dateobs=dateobs, + exptime=10.0, + ) + insert_subframe( + db_address, moluid='mol-exp-dup', stack_num=1, frmtotal=3, + camera='cam1', filepath='/data/dup2.fits', is_last=False, dateobs=dateobs, + exptime=45.0, + ) + subframes = get_subframes(db_address, 'mol-exp-dup') + assert len(subframes) == 1 + assert subframes[0].exptime == 45.0 + assert subframes[0].filepath == '/data/dup2.fits' + def test_insert_subframe_requires_filepath(self, db_address): dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) with pytest.raises(ValueError, match='filepath is required'): @@ -163,6 +193,128 @@ def test_mark_stack_complete_sets_timeout(self, db_address): assert s.completed_at is not None +# --------------------------------------------------------------------------- +# Adaptive timeout +# --------------------------------------------------------------------------- + +class TestAdaptiveTimeout: + + @staticmethod + def _created(base, seconds): + return base + datetime.timedelta(seconds=seconds) + + @staticmethod + def _insert_frame(db_address, moluid, stack_num, created_at, camera='cam1', + filepath='/data/reduced.fits', frmtotal=5, exptime=30.0, is_last=False): + dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) + insert_subframe( + db_address, moluid=moluid, stack_num=stack_num, frmtotal=frmtotal, + camera=camera, filepath=filepath, is_last=is_last, dateobs=dateobs, + exptime=exptime, + ) + with dbs.get_session(db_address, site_deploy=True) as session: + session.execute( + text("UPDATE subframes SET created_at = :created_at WHERE moluid = :moluid AND stack_num = :stack_num"), + {'created_at': created_at, 'moluid': moluid, 'stack_num': stack_num}, + ) + + def test_one_frame_stack_uses_exptime_plus_initial_buffer(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-one', 1, start, exptime=30.0) + + subframes = get_subframes(db_address, 'mol-one') + assert stack_has_timed_out(subframes, now=self._created(start, 90)) is False + assert stack_has_timed_out(subframes, now=self._created(start, 91)) is True + + def test_two_frame_stack_uses_first_to_second_baseline(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-two', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-two', 2, self._created(start, 120), exptime=30.0) + + subframes = get_subframes(db_address, 'mol-two') + assert adaptive_timeout_baseline_seconds(subframes) == 90.0 + assert stack_has_timed_out(subframes, now=self._created(start, 330)) is False + assert stack_has_timed_out(subframes, now=self._created(start, 331)) is True + + def test_positive_baseline_below_fallback_is_used_directly(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-fast', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-fast', 2, self._created(start, 40), exptime=30.0) + + subframes = get_subframes(db_address, 'mol-fast') + assert adaptive_timeout_baseline_seconds(subframes) == 10.0 + assert stack_has_timed_out(subframes, now=self._created(start, 89)) is False + assert stack_has_timed_out(subframes, now=self._created(start, 91)) is True + + def test_baseline_uses_fallback_when_adjusted_gap_is_nonpositive(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-clamp', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-clamp', 2, self._created(start, 20), exptime=30.0) + + subframes = get_subframes(db_address, 'mol-clamp') + assert adaptive_timeout_baseline_seconds(subframes) == 60.0 + assert stack_has_timed_out(subframes, now=self._created(start, 170)) is False + assert stack_has_timed_out(subframes, now=self._created(start, 171)) is True + + def test_late_observed_later_frame_gap_triggers_timeout(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-late-gap', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-late-gap', 2, self._created(start, 120), exptime=30.0) + self._insert_frame(db_address, 'mol-late-gap', 3, self._created(start, 331), exptime=30.0) + + subframes = get_subframes(db_address, 'mol-late-gap') + assert stack_has_timed_out(subframes, now=self._created(start, 340)) is True + + def test_missing_next_frame_beyond_threshold_triggers_timeout(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-missing', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-missing', 2, self._created(start, 120), exptime=30.0) + + subframes = get_subframes(db_address, 'mol-missing') + assert stack_has_timed_out(subframes, now=self._created(start, 331)) is True + + def test_incomplete_stack_inside_timeout_window_remains_active(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-active', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-active', 2, self._created(start, 120), exptime=30.0) + + check_timeouts(db_address, 'cam1', now=self._created(start, 200)) + + subframes = get_subframes(db_address, 'mol-active') + assert {s.status for s in subframes} == {'active'} + + def test_check_timeouts_marks_timeout_for_worker_camera_only(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-timeout', 1, start, camera='cam1', exptime=30.0) + self._insert_frame(db_address, 'mol-other-camera', 1, start, camera='cam2', exptime=30.0) + + check_timeouts(db_address, 'cam1', now=self._created(start, 91)) + + assert {s.status for s in get_subframes(db_address, 'mol-timeout')} == {'timeout'} + assert {s.status for s in get_subframes(db_address, 'mol-other-camera')} == {'active'} + + def test_check_timeouts_marks_complete_before_timeout(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + for stack_num in range(1, 4): + self._insert_frame( + db_address, 'mol-complete-wins', stack_num, self._created(start, stack_num * 300), + frmtotal=3, exptime=30.0, is_last=(stack_num == 3), + ) + + check_timeouts(db_address, 'cam1', now=self._created(start, 1000)) + + assert {s.status for s in get_subframes(db_address, 'mol-complete-wins')} == {'complete'} + + def test_check_timeouts_ignores_terminal_rows(self, db_address): + start = datetime.datetime(2024, 1, 1, 0, 0, 0) + self._insert_frame(db_address, 'mol-terminal', 1, start, exptime=30.0) + mark_stack_complete(db_address, 'mol-terminal', 'complete') + + check_timeouts(db_address, 'cam1', now=self._created(start, 91)) + + assert {s.status for s in get_subframes(db_address, 'mol-terminal')} == {'complete'} + + # --------------------------------------------------------------------------- # Redis notifications # --------------------------------------------------------------------------- @@ -520,6 +672,7 @@ def _make_fits_header(**overrides): h['MOLFRNUM'] = 1 h['FRMTOTAL'] = 5 h['MOLUID'] = 'mol-xyz' + h['EXPTIME'] = 30.0 for k, v in overrides.items(): h[k] = v return h @@ -562,6 +715,7 @@ def test_process_subframe(self, mock_run_stages, last_frame_val, expected_is_las assert subframes[0].frmtotal == 5 assert subframes[0].camera == 'cam1' assert subframes[0].is_last is expected_is_last + assert subframes[0].exptime == 30.0 assert subframes[0].filepath == '/data/processed/frame-e09.fits' mock_redis.eval.assert_called_once() @@ -656,8 +810,12 @@ class TestWorkerLoopResilience: @patch('banzai.stacking.redis_lib.Redis.from_url') @patch('banzai.stacking.time.sleep') + @patch('banzai.stacking.check_timeouts') + @patch('banzai.stacking.dbs.cleanup_old_subframes') @patch('banzai.stacking.process_notifications') - def test_run_worker_loop_continues_after_exception(self, mock_process, mock_sleep, mock_redis_cls): + def test_run_worker_loop_continues_after_exception( + self, mock_process, mock_cleanup, mock_check_timeouts, mock_sleep, mock_redis_cls + ): """run_worker_loop must not crash when process_notifications raises; it should log and continue.""" # First call raises, second call raises KeyboardInterrupt to escape the infinite loop. mock_process.side_effect = [Exception('boom'), KeyboardInterrupt] @@ -665,5 +823,24 @@ def test_run_worker_loop_continues_after_exception(self, mock_process, mock_slee run_worker_loop('cam1', 'sqlite:///fake.db', 'redis://localhost:6379', poll_interval=0) # process_notifications was called twice: once raised Exception, once raised KeyboardInterrupt. assert mock_process.call_count == 2 + mock_check_timeouts.assert_not_called() + mock_cleanup.assert_not_called() # sleep should have been called once (after the transient Exception, before continuing). mock_sleep.assert_called_once_with(0) + + @patch('banzai.stacking.redis_lib.Redis.from_url') + @patch('banzai.stacking.time.sleep') + @patch('banzai.stacking.check_timeouts') + @patch('banzai.stacking.dbs.cleanup_old_subframes') + @patch('banzai.stacking.process_notifications') + def test_run_worker_loop_invokes_timeout_sweep( + self, mock_process, mock_cleanup, mock_check_timeouts, mock_sleep, mock_redis_cls + ): + """run_worker_loop should sweep adaptive timeouts after notification processing.""" + mock_process.side_effect = [None, KeyboardInterrupt] + with pytest.raises(KeyboardInterrupt): + run_worker_loop('cam1', 'sqlite:///fake.db', 'redis://localhost:6379', poll_interval=0) + + mock_check_timeouts.assert_called_once_with('sqlite:///fake.db', 'cam1') + mock_cleanup.assert_called_once_with('sqlite:///fake.db', 30) + mock_sleep.assert_called_once_with(0) From 69d49a846b3d4392099534c4d84bd693d77824e8 Mon Sep 17 00:00:00 2001 From: Tim Beccue Date: Tue, 19 May 2026 19:57:58 -0400 Subject: [PATCH 2/2] Simplify smart stack timeout handling --- banzai/dbs.py | 8 +- banzai/scheduling.py | 2 - banzai/stacking.py | 100 +++------------- banzai/tests/site_e2e/test_site_e2e.py | 6 +- banzai/tests/test_smart_stacking.py | 151 +++++++++++-------------- docs/smartstacking_architecture.md | 46 ++++---- 6 files changed, 113 insertions(+), 200 deletions(-) diff --git a/banzai/dbs.py b/banzai/dbs.py index 363ab06d..2c969e0d 100755 --- a/banzai/dbs.py +++ b/banzai/dbs.py @@ -139,7 +139,6 @@ class Subframe(SiteBase): is_last = Column(Boolean, default=False) status = Column(String(20), default='active', nullable=False) dateobs = Column(DateTime, nullable=True) - exptime = Column(Float, nullable=True) created_at = Column(DateTime, default=datetime.datetime.utcnow) completed_at = Column(DateTime, nullable=True) __table_args__ = ( @@ -625,7 +624,7 @@ def replicate_instrument(instrument_record, db_address): db_session.commit() -def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, is_last, dateobs, exptime=None): +def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, is_last, dateobs): """Upsert a subframe record. On conflict (moluid, stack_num), reset the row to a fresh active state so a requeue behaves like a retry from scratch.""" if filepath is None: @@ -642,7 +641,7 @@ def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, i stmt = insert(Subframe).values( moluid=moluid, stack_num=stack_num, frmtotal=frmtotal, camera=camera, - filepath=filepath, is_last=is_last, dateobs=dateobs, exptime=exptime, + filepath=filepath, is_last=is_last, dateobs=dateobs, ) stmt = stmt.on_conflict_do_update( index_elements=['moluid', 'stack_num'], @@ -651,7 +650,6 @@ def insert_subframe(db_address, moluid, stack_num, frmtotal, camera, filepath, i 'camera': stmt.excluded.camera, 'is_last': stmt.excluded.is_last, 'dateobs': stmt.excluded.dateobs, - 'exptime': stmt.excluded.exptime, 'filepath': stmt.excluded.filepath, 'status': 'active', 'created_at': datetime.datetime.utcnow(), @@ -666,7 +664,7 @@ def get_subframes(db_address, moluid): with get_session(db_address, site_deploy=True) as session: return session.query(Subframe).filter( Subframe.moluid == moluid - ).all() + ).order_by(Subframe.stack_num).all() def get_active_subframes_for_camera(db_address, camera): diff --git a/banzai/scheduling.py b/banzai/scheduling.py index 43908cf1..43a2b25a 100644 --- a/banzai/scheduling.py +++ b/banzai/scheduling.py @@ -324,7 +324,6 @@ def process_subframe(self, body: dict, runtime_context: dict): camera = header.get('INSTRUME', '').strip() dateobs_str = header.get('DATE-OBS', '') dateobs = parse_date_obs(dateobs_str) if dateobs_str else None - exptime = header.get('EXPTIME') # Phase 1: Run reduction pipeline images = stage_utils.run_pipeline_stages([{'path': filepath}], runtime_context) @@ -347,7 +346,6 @@ def process_subframe(self, body: dict, runtime_context: dict): filepath=reduced_path, is_last=body.get('last_frame', False), dateobs=dateobs, - exptime=exptime, ) # Phase 3: Notify the stack worker that a new subframe is available diff --git a/banzai/stacking.py b/banzai/stacking.py index 36ac695e..12b44061 100644 --- a/banzai/stacking.py +++ b/banzai/stacking.py @@ -18,10 +18,9 @@ # Helpers # --------------------------------------------------------------------------- +# Time that may elapse after the latest reduced subframe before a stack times out. +STACK_TIMEOUT_SECONDS = 15 * 60 REQUIRED_MESSAGE_FIELDS = ('fits_file', 'last_frame', 'instrument_enqueue_timestamp') -INITIAL_FRAME_TIMEOUT_BUFFER_SECONDS = 60.0 -INVALID_BASELINE_FALLBACK_SECONDS = 60.0 -ADAPTIVE_TIMEOUT_MULTIPLIER = 2.0 def validate_message(body): @@ -41,90 +40,21 @@ def check_stack_complete(subframes, frmtotal): return bool(subframes) and (all_arrived or has_last) -def _arrival_ordered_subframes(subframes): - return sorted( - subframes, - key=lambda s: ( - s.created_at or datetime.datetime.max, - s.stack_num or 0, +def stack_has_timed_out(subframes, now=None): + """Return True when no newer reduced subframe has arrived before the timeout.""" + latest_reduced_time = max( + ( + subframe.created_at + for subframe in subframes + if subframe.created_at is not None ), + default=None, ) - - -def _exptime_seconds(subframe): - try: - exptime = float(subframe.exptime or 0.0) - except (TypeError, ValueError): - return 0.0 - return max(exptime, 0.0) - - -def adaptive_timeout_baseline_seconds(subframes): - """Return the adjusted first-to-second-frame baseline or invalid-value fallback.""" - ordered_subframes = _arrival_ordered_subframes(subframes) - if len(ordered_subframes) < 2: - return None - - # The first two arrivals establish the expected post-exposure delivery cadence - # for this moluid; later timeout checks compare against this baseline. - exptime = _exptime_seconds(ordered_subframes[0]) - first_arrival = ordered_subframes[0].created_at - second_arrival = ordered_subframes[1].created_at - if first_arrival is None or second_arrival is None: - return INVALID_BASELINE_FALLBACK_SECONDS - - # Subtract exposure duration because it is unrelated to processing/transport - # delay, which is the signal this timeout is trying to detect. - adjusted_gap = (second_arrival - first_arrival).total_seconds() - exptime - # A zero or negative adjusted gap means the timing data is not usable, so - # fall back to a conservative positive baseline instead of timing out instantly. - if adjusted_gap <= 0: - return INVALID_BASELINE_FALLBACK_SECONDS - return adjusted_gap - - -def stack_has_timed_out(subframes, now=None): - """Return True when a stack has exceeded the adaptive arrival timeout.""" - ordered_subframes = _arrival_ordered_subframes(subframes) - if not ordered_subframes: + if latest_reduced_time is None: return False now = now or datetime.datetime.utcnow() - # Use the first frame's exposure time for the whole stack; smart-stack subframes - # are expected to have consistent exposure durations. - exptime = _exptime_seconds(ordered_subframes[0]) - first_arrival = ordered_subframes[0].created_at - if first_arrival is None: - return False - - # Until frame 2 arrives, there is no cadence measurement yet, so use the - # first-frame exposure plus a fixed post-exposure buffer. - if len(ordered_subframes) == 1: - first_frame_timeout = exptime + INITIAL_FRAME_TIMEOUT_BUFFER_SECONDS - return (now - first_arrival).total_seconds() > first_frame_timeout - - # After frame 2, the first-to-second adjusted gap defines the expected - # post-exposure arrival cadence; future gaps get twice that allowance. - baseline = adaptive_timeout_baseline_seconds(ordered_subframes) - timeout_threshold = baseline * ADAPTIVE_TIMEOUT_MULTIPLIER - - # Late frames should still cause a timeout even if they eventually arrived; - # the arrival timestamps preserve those delayed adjacent gaps. - for previous_frame, current_frame in zip(ordered_subframes, ordered_subframes[1:]): - if previous_frame.created_at is None or current_frame.created_at is None: - continue - adjusted_gap = (current_frame.created_at - previous_frame.created_at).total_seconds() - exptime - if adjusted_gap > timeout_threshold: - return True - - # If no already-observed gap was too long, check whether the next expected - # frame has now been missing for too long. - latest_arrival = ordered_subframes[-1].created_at - if latest_arrival is None: - return False - missing_frame_gap = (now - latest_arrival).total_seconds() - exptime - return missing_frame_gap > timeout_threshold - + return (now - latest_reduced_time).total_seconds() > STACK_TIMEOUT_SECONDS # --------------------------------------------------------------------------- # Notifications @@ -214,7 +144,7 @@ def finalize_stack(db_address, moluid, status='complete'): def check_timeouts(db_address, camera, now=None): - """Finalize active stacks that are complete or have exceeded adaptive timeout.""" + """Finalize active stacks that are complete or have exceeded the fixed timeout.""" active_subframes = dbs.get_active_subframes_for_camera(db_address, camera) # The DB scan is camera-wide, so regroup rows into independent stack states # before making terminal-status decisions. @@ -225,10 +155,10 @@ def check_timeouts(db_address, camera, now=None): for moluid, subframes in subframes_by_moluid.items(): frmtotal = subframes[0].frmtotal # Completion wins over timeout so a fully reduced stack cannot be marked - # partial just because its cadence also exceeded the adaptive threshold. + # partial just because its cadence also exceeded the fixed threshold. if check_stack_complete(subframes, frmtotal): finalize_stack(db_address, moluid, status='complete') - # Incomplete active stacks are finalized only when their arrival cadence + # Incomplete active stacks are finalized only when reduced-row timing # indicates that the next expected frame is no longer arriving normally. elif stack_has_timed_out(subframes, now=now): finalize_stack(db_address, moluid, status='timeout') diff --git a/banzai/tests/site_e2e/test_site_e2e.py b/banzai/tests/site_e2e/test_site_e2e.py index 6985a9c6..d5bf9492 100644 --- a/banzai/tests/site_e2e/test_site_e2e.py +++ b/banzai/tests/site_e2e/test_site_e2e.py @@ -355,7 +355,6 @@ def check(): subframes = poll_until(check, timeout=300) assert subframes, "Subframe stack did not complete within timeout" - assert subframes[0].exptime is not None # Verify the reduced output file exists on disk. # Paths in the DB are now absolute host paths, so check directly. @@ -400,8 +399,11 @@ def row_exists(): dbs.Subframe.moluid == moluid ).first() - assert poll_until(row_exists, timeout=120, interval=2), \ + subframe = poll_until(row_exists, timeout=120, interval=2) + assert subframe, \ "Timed-out subframe row was not inserted" + assert subframe.filepath, \ + "Timed-out subframe row was inserted without a reduced filepath" stale_created_at = datetime.datetime.utcnow() - datetime.timedelta(hours=2) with dbs.get_session(LOCAL_DB_ADDRESS, site_deploy=True) as session: diff --git a/banzai/tests/test_smart_stacking.py b/banzai/tests/test_smart_stacking.py index c56b175d..140aa35f 100644 --- a/banzai/tests/test_smart_stacking.py +++ b/banzai/tests/test_smart_stacking.py @@ -14,7 +14,7 @@ from banzai.stacking import (validate_message, check_stack_complete, push_notification, pop_notification, REDIS_KEY_PREFIX, process_notifications, finalize_stack, check_timeouts, - stack_has_timed_out, adaptive_timeout_baseline_seconds, + stack_has_timed_out, STACK_TIMEOUT_SECONDS, run_worker_loop, StackingSupervisor) from banzai.scheduling import process_subframe from banzai.main import SubframeListener @@ -96,18 +96,18 @@ def test_insert_subframe_and_get_subframes(self, db_address): assert subframe.filepath == '/data/frame1.fits' assert subframe.is_last is False assert subframe.dateobs == dateobs - assert subframe.exptime is None - def test_insert_subframe_stores_exptime(self, db_address): + def test_get_subframes_orders_by_stack_num(self, db_address): dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) - insert_subframe( - db_address, moluid='mol-exp', stack_num=1, frmtotal=5, - camera='cam1', filepath='/data/frame1.fits', is_last=False, dateobs=dateobs, - exptime=30.0, - ) - subframes = get_subframes(db_address, moluid='mol-exp') - assert len(subframes) == 1 - assert subframes[0].exptime == 30.0 + for stack_num in (3, 1, 2): + insert_subframe( + db_address, moluid='mol-ordered', stack_num=stack_num, frmtotal=3, + camera='cam1', filepath=f'/data/frame{stack_num}.fits', is_last=False, dateobs=dateobs, + ) + + subframes = get_subframes(db_address, moluid='mol-ordered') + + assert [subframe.stack_num for subframe in subframes] == [1, 2, 3] def test_insert_subframe_duplicate_resets_state_for_retry(self, db_address): dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) @@ -134,23 +134,6 @@ def test_insert_subframe_duplicate_resets_state_for_retry(self, db_address): assert subframe.is_last is True assert subframe.dateobs == new_dateobs - def test_insert_subframe_duplicate_refreshes_exptime(self, db_address): - dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) - insert_subframe( - db_address, moluid='mol-exp-dup', stack_num=1, frmtotal=3, - camera='cam1', filepath='/data/dup1.fits', is_last=False, dateobs=dateobs, - exptime=10.0, - ) - insert_subframe( - db_address, moluid='mol-exp-dup', stack_num=1, frmtotal=3, - camera='cam1', filepath='/data/dup2.fits', is_last=False, dateobs=dateobs, - exptime=45.0, - ) - subframes = get_subframes(db_address, 'mol-exp-dup') - assert len(subframes) == 1 - assert subframes[0].exptime == 45.0 - assert subframes[0].filepath == '/data/dup2.fits' - def test_insert_subframe_requires_filepath(self, db_address): dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) with pytest.raises(ValueError, match='filepath is required'): @@ -194,10 +177,10 @@ def test_mark_stack_complete_sets_timeout(self, db_address): # --------------------------------------------------------------------------- -# Adaptive timeout +# Fixed reduced-subframe timeout # --------------------------------------------------------------------------- -class TestAdaptiveTimeout: +class TestFixedReducedSubframeTimeout: @staticmethod def _created(base, seconds): @@ -205,12 +188,11 @@ def _created(base, seconds): @staticmethod def _insert_frame(db_address, moluid, stack_num, created_at, camera='cam1', - filepath='/data/reduced.fits', frmtotal=5, exptime=30.0, is_last=False): + filepath='/data/reduced.fits', frmtotal=5, is_last=False): dateobs = datetime.datetime(2024, 6, 15, 12, 0, 0) insert_subframe( db_address, moluid=moluid, stack_num=stack_num, frmtotal=frmtotal, camera=camera, filepath=filepath, is_last=is_last, dateobs=dateobs, - exptime=exptime, ) with dbs.get_session(db_address, site_deploy=True) as session: session.execute( @@ -218,65 +200,63 @@ def _insert_frame(db_address, moluid, stack_num, created_at, camera='cam1', {'created_at': created_at, 'moluid': moluid, 'stack_num': stack_num}, ) - def test_one_frame_stack_uses_exptime_plus_initial_buffer(self, db_address): + def test_one_reduced_row_times_out_only_after_fixed_threshold(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-one', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-one', 1, start) - subframes = get_subframes(db_address, 'mol-one') - assert stack_has_timed_out(subframes, now=self._created(start, 90)) is False - assert stack_has_timed_out(subframes, now=self._created(start, 91)) is True + check_timeouts(db_address, 'cam1', now=self._created(start, STACK_TIMEOUT_SECONDS)) - def test_two_frame_stack_uses_first_to_second_baseline(self, db_address): - start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-two', 1, start, exptime=30.0) - self._insert_frame(db_address, 'mol-two', 2, self._created(start, 120), exptime=30.0) + assert {s.status for s in get_subframes(db_address, 'mol-one')} == {'active'} - subframes = get_subframes(db_address, 'mol-two') - assert adaptive_timeout_baseline_seconds(subframes) == 90.0 - assert stack_has_timed_out(subframes, now=self._created(start, 330)) is False - assert stack_has_timed_out(subframes, now=self._created(start, 331)) is True + check_timeouts(db_address, 'cam1', now=self._created(start, STACK_TIMEOUT_SECONDS + 1)) - def test_positive_baseline_below_fallback_is_used_directly(self, db_address): - start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-fast', 1, start, exptime=30.0) - self._insert_frame(db_address, 'mol-fast', 2, self._created(start, 40), exptime=30.0) + assert {s.status for s in get_subframes(db_address, 'mol-one')} == {'timeout'} - subframes = get_subframes(db_address, 'mol-fast') - assert adaptive_timeout_baseline_seconds(subframes) == 10.0 - assert stack_has_timed_out(subframes, now=self._created(start, 89)) is False - assert stack_has_timed_out(subframes, now=self._created(start, 91)) is True - - def test_baseline_uses_fallback_when_adjusted_gap_is_nonpositive(self, db_address): + def test_newer_reduced_row_resets_timeout(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-clamp', 1, start, exptime=30.0) - self._insert_frame(db_address, 'mol-clamp', 2, self._created(start, 20), exptime=30.0) + second_arrival = self._created(start, STACK_TIMEOUT_SECONDS + 1) + self._insert_frame(db_address, 'mol-reset', 1, start) + self._insert_frame(db_address, 'mol-reset', 2, second_arrival) + + check_timeouts( + db_address, + 'cam1', + now=self._created(second_arrival, STACK_TIMEOUT_SECONDS), + ) - subframes = get_subframes(db_address, 'mol-clamp') - assert adaptive_timeout_baseline_seconds(subframes) == 60.0 - assert stack_has_timed_out(subframes, now=self._created(start, 170)) is False - assert stack_has_timed_out(subframes, now=self._created(start, 171)) is True + subframes = get_subframes(db_address, 'mol-reset') + assert stack_has_timed_out(subframes, now=self._created(second_arrival, STACK_TIMEOUT_SECONDS)) is False + assert {s.status for s in subframes} == {'active'} - def test_late_observed_later_frame_gap_triggers_timeout(self, db_address): + def test_latest_reduced_row_over_threshold_times_out(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-late-gap', 1, start, exptime=30.0) - self._insert_frame(db_address, 'mol-late-gap', 2, self._created(start, 120), exptime=30.0) - self._insert_frame(db_address, 'mol-late-gap', 3, self._created(start, 331), exptime=30.0) + second_arrival = self._created(start, 60) + self._insert_frame(db_address, 'mol-latest-timeout', 1, start) + self._insert_frame(db_address, 'mol-latest-timeout', 2, second_arrival) - subframes = get_subframes(db_address, 'mol-late-gap') - assert stack_has_timed_out(subframes, now=self._created(start, 340)) is True + check_timeouts(db_address, 'cam1', now=self._created(second_arrival, STACK_TIMEOUT_SECONDS + 1)) - def test_missing_next_frame_beyond_threshold_triggers_timeout(self, db_address): + assert {s.status for s in get_subframes(db_address, 'mol-latest-timeout')} == {'timeout'} + + def test_later_newer_reduced_row_resets_timeout(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-missing', 1, start, exptime=30.0) - self._insert_frame(db_address, 'mol-missing', 2, self._created(start, 120), exptime=30.0) + third_arrival = self._created(start, 60 + STACK_TIMEOUT_SECONDS + 1) + self._insert_frame(db_address, 'mol-late-reset', 1, start) + self._insert_frame(db_address, 'mol-late-reset', 2, self._created(start, 60)) + self._insert_frame( + db_address, + 'mol-late-reset', + 3, + third_arrival, + ) - subframes = get_subframes(db_address, 'mol-missing') - assert stack_has_timed_out(subframes, now=self._created(start, 331)) is True + subframes = get_subframes(db_address, 'mol-late-reset') + assert stack_has_timed_out(subframes, now=self._created(third_arrival, 1)) is False def test_incomplete_stack_inside_timeout_window_remains_active(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-active', 1, start, exptime=30.0) - self._insert_frame(db_address, 'mol-active', 2, self._created(start, 120), exptime=30.0) + self._insert_frame(db_address, 'mol-active', 1, start) + self._insert_frame(db_address, 'mol-active', 2, self._created(start, 120)) check_timeouts(db_address, 'cam1', now=self._created(start, 200)) @@ -285,10 +265,10 @@ def test_incomplete_stack_inside_timeout_window_remains_active(self, db_address) def test_check_timeouts_marks_timeout_for_worker_camera_only(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-timeout', 1, start, camera='cam1', exptime=30.0) - self._insert_frame(db_address, 'mol-other-camera', 1, start, camera='cam2', exptime=30.0) + self._insert_frame(db_address, 'mol-timeout', 1, start, camera='cam1') + self._insert_frame(db_address, 'mol-other-camera', 1, start, camera='cam2') - check_timeouts(db_address, 'cam1', now=self._created(start, 91)) + check_timeouts(db_address, 'cam1', now=self._created(start, STACK_TIMEOUT_SECONDS + 1)) assert {s.status for s in get_subframes(db_address, 'mol-timeout')} == {'timeout'} assert {s.status for s in get_subframes(db_address, 'mol-other-camera')} == {'active'} @@ -297,20 +277,23 @@ def test_check_timeouts_marks_complete_before_timeout(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) for stack_num in range(1, 4): self._insert_frame( - db_address, 'mol-complete-wins', stack_num, self._created(start, stack_num * 300), - frmtotal=3, exptime=30.0, is_last=(stack_num == 3), + db_address, + 'mol-complete-wins', + stack_num, + self._created(start, (stack_num - 1) * (STACK_TIMEOUT_SECONDS + 1)), + frmtotal=3, is_last=(stack_num == 3), ) - check_timeouts(db_address, 'cam1', now=self._created(start, 1000)) + check_timeouts(db_address, 'cam1', now=self._created(start, 3 * (STACK_TIMEOUT_SECONDS + 1))) assert {s.status for s in get_subframes(db_address, 'mol-complete-wins')} == {'complete'} def test_check_timeouts_ignores_terminal_rows(self, db_address): start = datetime.datetime(2024, 1, 1, 0, 0, 0) - self._insert_frame(db_address, 'mol-terminal', 1, start, exptime=30.0) + self._insert_frame(db_address, 'mol-terminal', 1, start) mark_stack_complete(db_address, 'mol-terminal', 'complete') - check_timeouts(db_address, 'cam1', now=self._created(start, 91)) + check_timeouts(db_address, 'cam1', now=self._created(start, STACK_TIMEOUT_SECONDS + 1)) assert {s.status for s in get_subframes(db_address, 'mol-terminal')} == {'complete'} @@ -672,7 +655,6 @@ def _make_fits_header(**overrides): h['MOLFRNUM'] = 1 h['FRMTOTAL'] = 5 h['MOLUID'] = 'mol-xyz' - h['EXPTIME'] = 30.0 for k, v in overrides.items(): h[k] = v return h @@ -715,7 +697,6 @@ def test_process_subframe(self, mock_run_stages, last_frame_val, expected_is_las assert subframes[0].frmtotal == 5 assert subframes[0].camera == 'cam1' assert subframes[0].is_last is expected_is_last - assert subframes[0].exptime == 30.0 assert subframes[0].filepath == '/data/processed/frame-e09.fits' mock_redis.eval.assert_called_once() @@ -836,7 +817,7 @@ def test_run_worker_loop_continues_after_exception( def test_run_worker_loop_invokes_timeout_sweep( self, mock_process, mock_cleanup, mock_check_timeouts, mock_sleep, mock_redis_cls ): - """run_worker_loop should sweep adaptive timeouts after notification processing.""" + """run_worker_loop should sweep fixed timeouts after notification processing.""" mock_process.side_effect = [None, KeyboardInterrupt] with pytest.raises(KeyboardInterrupt): run_worker_loop('cam1', 'sqlite:///fake.db', 'redis://localhost:6379', poll_interval=0) diff --git a/docs/smartstacking_architecture.md b/docs/smartstacking_architecture.md index f80635df..23ba553f 100644 --- a/docs/smartstacking_architecture.md +++ b/docs/smartstacking_architecture.md @@ -140,14 +140,22 @@ database level. 4. Checks the completion rule when at least one row is active. 5. Marks a complete group as `complete`. 6. Repeats from step 1 until the sorted set is empty. -7. Deletes old terminal rows according to the retention setting. -8. Sleeps for the polling interval. +7. Sweeps active rows for its camera to recover missed notifications and apply timeouts. +8. Deletes old terminal rows according to the retention setting. +9. Sleeps for the polling interval. + +After the notification set is empty, the worker scans active rows for its +camera. This fallback sweep lets complete stacks reach terminal state even if +a notification was missed. It also marks an incomplete stack as `timeout` when +no newer reduced row has arrived within 15 minutes of the latest row's +`created_at`. A newer reduced row resets that window. Completion is checked +before timeout, so complete stacks become `complete` rather than `timeout`. The worker does not build stack state from Redis messages. Ten notifications for the same pending `MOLUID` become one sorted-set member and one database query. Multiple `MOLUID`s may be active for one camera. None exclusively owns -the worker: an incomplete group waits for another notification while older -pending work for other groups continues in FIFO order. +the worker: an incomplete group waits for another notification or the fallback +sweep while older pending work for other groups continues in FIFO order. ```mermaid flowchart TD @@ -158,6 +166,9 @@ flowchart TD active{"Any active rows?"} complete{"Completion rule true?"} mark["Mark every row complete"] + sweep["Scan active rows
for this camera"] + eligible{"Complete or timed out?"} + finalize["Mark complete or timeout"] cleanup["Delete old terminal rows"] sleep["Sleep"] @@ -167,7 +178,9 @@ flowchart TD active -->|"no"| pop complete -->|"yes"| mark --> pop complete -->|"no"| pop - notified -->|"no"| cleanup + notified -->|"no"| sweep --> eligible + eligible -->|"yes"| finalize --> cleanup + eligible -->|"no"| cleanup cleanup --> sleep --> tick ``` @@ -208,7 +221,7 @@ one or more raw e00 subframes | `banzai-subframe-listener` | `banzai.main:run_subframe_worker` | Consume and validate RabbitMQ messages; dispatch Celery tasks. Despite the CLI name, this is the listener process. | | `banzai-subframe-worker` | Celery worker consuming `SUBFRAME_TASK_QUEUE_NAME` | Reduce the FITS file, upsert the reduced row, and send a notification. | | `banzai-stacking-supervisor` | `banzai.stacking:run_supervisor` | Discover cameras at startup, start one child per camera, and restart children that exit. | -| `stacking-worker-{camera}` | `banzai.stacking:run_worker_loop` | Process unique notifications in FIFO order, query stack state, mark complete groups, and run retention cleanup. | +| `stacking-worker-{camera}` | `banzai.stacking:run_worker_loop` | Process unique notifications in FIFO order, sweep active stacks for completion or timeout, and run retention cleanup. | Camera discovery uses `dbs.get_instruments_at_site(site_id, db_address)` once when the supervisor starts. Adding a camera to the database does not create a @@ -264,7 +277,7 @@ state. | `is_last` | Copy of the queue message's `last_frame` flag. | | `status` | Expected values are `active`, `complete`, and `timeout`. | | `dateobs` | Observation time from `DATE-OBS`, when available. | -| `created_at` | Time the row was inserted or reset by an upsert. | +| `created_at` | Time the reduced row was inserted or reset; the fixed timeout is measured from the newest value in a stack. | | `completed_at` | Time the group was marked terminal. | Current transitions are: @@ -273,16 +286,17 @@ Current transitions are: stateDiagram-v2 [*] --> active: reduced row upserted active --> complete: completion rule passes + active --> timeout: no newer reduced row for 15 minutes complete --> active: same subframe is reprocessed timeout --> active: same subframe is reprocessed ``` -The model and database helper support `timeout`, but no production worker path -currently sets it. - Retention cleanup deletes rows whose status is not `active` and whose `completed_at` is older than `STACK_RETENTION_DAYS` (30 days by default). -Active rows are retained indefinitely. +Active rows remain until the completion rule passes or the fixed timeout +expires. `STACK_TIMEOUT_MINUTES` in `site-banzai-env.default` is a legacy, +unused setting; the worker timeout is the 15-minute constant in +`banzai.stacking`. ## Current limitations and failure behavior @@ -290,16 +304,6 @@ These details are important when debugging or extending the subsystem: - **No `e45` stack product is generated.** Completion currently means a database transition only; only the individual `e09` inputs exist. -- **No timeout finalization runs.** `STACK_TIMEOUT_MINUTES` remains in - `site-banzai-env.default`, but the current Python code does not read it. -- **There is no fallback database sweep.** An active stack is reconsidered only - after a Redis notification for its `MOLUID`. If notification is disabled, - lost, or missed, the stack remains active until another notification arrives. -- **The in-flight notification has a crash window.** `ZPOPMIN` removes one - `MOLUID` before its database work begins. A worker crash can lose that one - notification, but all other members remain ordered in the sorted set. - PostgreSQL remains correct, but there is currently no sweep to rediscover an - active stack whose final notification was lost. - **Cleanup is repeated per camera.** Every camera worker calls the same database-wide terminal-row cleanup each tick. This is correct but redundant. - **Camera discovery is startup-only.** Restart the supervisor after adding or