From 8da0eb39f7faee0ba45fbabf07f840341b506077 Mon Sep 17 00:00:00 2001 From: 120318 <20685540+getupbuzz@users.noreply.github.com> Date: Tue, 7 Jul 2026 17:49:21 +0800 Subject: [PATCH 1/2] Reduce maintenance notification noise --- ...05_indexer_health_notification_cooldown.py | 39 +++++++++++ .../indexer_site_health_repository.py | 2 + backend/app/db/sql/models.py | 1 + .../schemas/runtime/indexer_site_health.py | 1 + .../media_server_sync/season_runner.py | 6 +- .../config/indexer_client_settings.py | 15 +++- .../tests/test_indexer_site_health_alerts.py | 69 +++++++++++++++++++ .../test_media_server_sync_boundaries.py | 63 ++++++++++++++++- 8 files changed, 190 insertions(+), 6 deletions(-) create mode 100644 backend/alembic/versions/0005_indexer_health_notification_cooldown.py diff --git a/backend/alembic/versions/0005_indexer_health_notification_cooldown.py b/backend/alembic/versions/0005_indexer_health_notification_cooldown.py new file mode 100644 index 0000000..bad5de4 --- /dev/null +++ b/backend/alembic/versions/0005_indexer_health_notification_cooldown.py @@ -0,0 +1,39 @@ +"""Track indexer health notification cooldown. + +Revision ID: 0005_indexer_health_notification_cooldown +Revises: 0004_subscription_search_cadence +Create Date: 2026-07-07 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "0005_indexer_health_notification_cooldown" +down_revision = "0004_subscription_search_cadence" +branch_labels = None +depends_on = None + + +def _has_table(table_name: str) -> bool: + bind = op.get_bind() + return sa.inspect(bind).has_table(table_name) + + +def _has_column(table_name: str, column_name: str) -> bool: + bind = op.get_bind() + inspector = sa.inspect(bind) + return any(column["name"] == column_name for column in inspector.get_columns(table_name)) + + +def upgrade() -> None: + if not _has_table("indexer_site_health") or _has_column("indexer_site_health", "last_notified_at"): + return + op.add_column("indexer_site_health", sa.Column("last_notified_at", sa.Text(), nullable=True)) + + +def downgrade() -> None: + if _has_table("indexer_site_health") and _has_column("indexer_site_health", "last_notified_at"): + op.drop_column("indexer_site_health", "last_notified_at") diff --git a/backend/app/db/repositories/indexer_site_health_repository.py b/backend/app/db/repositories/indexer_site_health_repository.py index 3279b55..3961926 100644 --- a/backend/app/db/repositories/indexer_site_health_repository.py +++ b/backend/app/db/repositories/indexer_site_health_repository.py @@ -23,6 +23,7 @@ def _to_model(row: IndexerSiteHealthORM) -> IndexerSiteHealthStatus: "consecutive_failures": row.consecutive_failures, "last_error_message": row.last_error_message, "notify_pending": bool(row.notify_pending), + "last_notified_at": row.last_notified_at, "client_type": row.client_type, } ) @@ -51,6 +52,7 @@ def upsert(self, status: IndexerSiteHealthStatus) -> IndexerSiteHealthStatus: row.consecutive_failures = status.consecutive_failures row.last_error_message = status.last_error_message row.notify_pending = bool(status.notify_pending) + row.last_notified_at = status.last_notified_at.isoformat() if status.last_notified_at else None row.client_type = status.client_type session.commit() return status diff --git a/backend/app/db/sql/models.py b/backend/app/db/sql/models.py index 5e456ba..6b5347d 100644 --- a/backend/app/db/sql/models.py +++ b/backend/app/db/sql/models.py @@ -440,6 +440,7 @@ class IndexerSiteHealthORM(Base): consecutive_failures: Mapped[int] = mapped_column(Integer, nullable=False, default=0) last_error_message: Mapped[str | None] = mapped_column(Text, nullable=True) notify_pending: Mapped[bool] = mapped_column(Integer, nullable=False, default=0) + last_notified_at: Mapped[str | None] = mapped_column(Text, nullable=True) client_type: Mapped[str] = mapped_column(Text, nullable=False, default="jackett") __table_args__ = ( diff --git a/backend/app/schemas/runtime/indexer_site_health.py b/backend/app/schemas/runtime/indexer_site_health.py index 3e3e627..3c0ad8d 100644 --- a/backend/app/schemas/runtime/indexer_site_health.py +++ b/backend/app/schemas/runtime/indexer_site_health.py @@ -16,6 +16,7 @@ class IndexerSiteHealthStatus(BaseModel): consecutive_failures: int = 0 last_error_message: Optional[str] = None notify_pending: bool = False + last_notified_at: Optional[datetime] = None client_type: str = "jackett" diff --git a/backend/app/services/application/workflows/media_server_sync/season_runner.py b/backend/app/services/application/workflows/media_server_sync/season_runner.py index 0422ccc..0693b7f 100644 --- a/backend/app/services/application/workflows/media_server_sync/season_runner.py +++ b/backend/app/services/application/workflows/media_server_sync/season_runner.py @@ -79,7 +79,7 @@ async def sync_one_season( error=str(exc), ) raise - if self._should_emit_scheduler_completed_event(needs.missing_flags): + if self._should_emit_scheduler_completed_event(state, needs.missing_flags): emit_media_server_sync_events( EventType.MEDIA_SERVER_SYNC_COMPLETED, media, @@ -129,8 +129,8 @@ async def apply_updates( ) @staticmethod - def _should_emit_scheduler_completed_event(missing_flags: list[str]) -> bool: - return set(missing_flags) != {"stale"} + def _should_emit_scheduler_completed_event(state: MediaServerSyncState, missing_flags: list[str]) -> bool: + return state.last_success_at is None and set(missing_flags) != {"stale"} media_server_sync_season_runner = MediaServerSyncSeasonRunner() diff --git a/backend/app/services/config/indexer_client_settings.py b/backend/app/services/config/indexer_client_settings.py index b9e2ed0..babac59 100644 --- a/backend/app/services/config/indexer_client_settings.py +++ b/backend/app/services/config/indexer_client_settings.py @@ -2,7 +2,7 @@ import logging from collections.abc import Mapping -from datetime import datetime +from datetime import datetime, timedelta from app.db.repositories.indexer_site_health_repository import IndexerSiteHealthRepository from app.db.repositories.settings_sqlite_repository import SettingsSqliteRepository @@ -16,6 +16,7 @@ INDEXER_SITE_FAILURE_NOTIFY_THRESHOLD = 3 +INDEXER_SITE_FAILURE_NOTIFY_COOLDOWN = timedelta(hours=24) logger = logging.getLogger("app.services.config.indexer_client_settings") @@ -103,6 +104,7 @@ def record_success( consecutive_failures=0, last_error_message=None, notify_pending=False, + last_notified_at=current.last_notified_at if current else None, client_type=client_type, ) return self._upsert(status) @@ -121,6 +123,10 @@ def record_failure( now = datetime.now() previous_failures = current.consecutive_failures if current else 0 consecutive_failures = previous_failures + 1 + should_emit = ( + consecutive_failures >= INDEXER_SITE_FAILURE_NOTIFY_THRESHOLD + and self._notify_cooldown_elapsed(current.last_notified_at if current else None, now) + ) status = IndexerSiteHealthStatus( indexer_id=indexer_id, indexer_name=indexer_name, @@ -133,13 +139,18 @@ def record_failure( consecutive_failures=consecutive_failures, last_error_message=error_message, notify_pending=consecutive_failures >= INDEXER_SITE_FAILURE_NOTIFY_THRESHOLD, + last_notified_at=now if should_emit else (current.last_notified_at if current else None), client_type=client_type, ) saved = self._upsert(status) - if previous_failures < INDEXER_SITE_FAILURE_NOTIFY_THRESHOLD <= consecutive_failures: + if should_emit: self._emit_unhealthy_event(saved) return saved + @staticmethod + def _notify_cooldown_elapsed(last_notified_at: datetime | None, now: datetime) -> bool: + return last_notified_at is None or (now - last_notified_at) >= INDEXER_SITE_FAILURE_NOTIFY_COOLDOWN + def list_by_indexer(self, indexer_id: str) -> list[IndexerSiteHealthStatus]: return self._repo.list_by_indexer(indexer_id) diff --git a/backend/tests/test_indexer_site_health_alerts.py b/backend/tests/test_indexer_site_health_alerts.py index 73375c5..8db6952 100644 --- a/backend/tests/test_indexer_site_health_alerts.py +++ b/backend/tests/test_indexer_site_health_alerts.py @@ -1,3 +1,5 @@ +from datetime import datetime, timedelta + from app.schemas.constants.event_types import EventTypes from app.schemas.domain.event import EventLevel from app.schemas.runtime.indexer_site_health import IndexerSiteHealthStatus @@ -103,3 +105,70 @@ def test_indexer_site_success_clears_notify_pending_and_allows_future_threshold( assert status is not None assert status.consecutive_failures == 3 assert status.notify_pending is True + + +def test_indexer_site_unhealthy_event_respects_notification_cooldown(monkeypatch): + emitted_events = [] + monkeypatch.setattr( + "app.services.config.indexer_client_settings.event_service.emit", + lambda event: emitted_events.append(event), + ) + state = IndexerSiteHealthState(repo=_FakeIndexerSiteHealthRepository()) + + for _ in range(3): + state.record_failure( + indexer_id="prowlarr", + indexer_name="Prowlarr", + site_id="audiences", + site_name="Audiences", + error_message="disabled", + ) + state.record_success( + indexer_id="prowlarr", + indexer_name="Prowlarr", + site_id="audiences", + site_name="Audiences", + ) + for _ in range(3): + state.record_failure( + indexer_id="prowlarr", + indexer_name="Prowlarr", + site_id="audiences", + site_name="Audiences", + error_message="disabled again", + ) + + assert len(emitted_events) == 1 + + +def test_indexer_site_unhealthy_event_repeats_after_notification_cooldown(monkeypatch): + emitted_events = [] + monkeypatch.setattr( + "app.services.config.indexer_client_settings.event_service.emit", + lambda event: emitted_events.append(event), + ) + repo = _FakeIndexerSiteHealthRepository() + state = IndexerSiteHealthState(repo=repo) + old_notification = datetime.now() - timedelta(hours=25) + repo.upsert( + IndexerSiteHealthStatus( + indexer_id="prowlarr", + indexer_name="Prowlarr", + site_id="audiences", + site_name="Audiences", + status="unhealthy", + consecutive_failures=7, + notify_pending=True, + last_notified_at=old_notification, + ) + ) + + state.record_failure( + indexer_id="prowlarr", + indexer_name="Prowlarr", + site_id="audiences", + site_name="Audiences", + error_message="still disabled", + ) + + assert len(emitted_events) == 1 diff --git a/backend/tests/test_media_server_sync_boundaries.py b/backend/tests/test_media_server_sync_boundaries.py index 03f93ab..2ea89d0 100644 --- a/backend/tests/test_media_server_sync_boundaries.py +++ b/backend/tests/test_media_server_sync_boundaries.py @@ -796,7 +796,7 @@ async def fake_apply_updates(*args, **kwargs): @pytest.mark.asyncio -async def test_media_server_sync_scheduler_emits_completed_event_for_missing_artifacts(monkeypatch, tmp_path: Path): +async def test_media_server_sync_scheduler_suppresses_completed_event_for_previously_synced_missing_artifacts(monkeypatch, tmp_path: Path): media = _tv() video = tmp_path / "Show" / "Season 01" / "Show.S01E01.mkv" video.parent.mkdir(parents=True) @@ -852,6 +852,67 @@ async def fake_apply_updates(*args, **kwargs): MediaServerSyncConfig(write_nfo=False, download_images=False), ) + assert result + assert events == [] + + +@pytest.mark.asyncio +async def test_media_server_sync_scheduler_emits_completed_event_for_first_scrape_missing_artifacts(monkeypatch, tmp_path: Path): + media = _tv() + video = tmp_path / "Show" / "Season 01" / "Show.S01E01.mkv" + video.parent.mkdir(parents=True) + video.write_text("video") + files = [ + LibraryFile( + id="file-1", + task_id="task-1", + directory_id="dir-1", + media_id=media.media_id, + path=str(video.parent), + file_name=video.name, + created_at=1.0, + ) + ] + target = MediaServerSyncTargetFile(destination_path=str(video), episode_number=1) + needs = MediaServerSyncDetectNeeds( + should_run=True, + missing_flags=["episode_nfo_missing"], + transfer_results=[target], + anchor_file=str(video), + media_root_dir=str(tmp_path / "Show"), + ) + runner = MediaServerSyncSeasonRunner() + events = [] + + async def fake_info(media_id, season_number=None): + return media + + async def fake_layout(media_id, library_files): + return LibraryMediaLayout(media_id=media_id, media_type=media.media_type, primary_anchor_file=str(video)) + + async def fake_detect(*args, **kwargs): + return needs + + async def fake_apply_updates(*args, **kwargs): + return None + + monkeypatch.setattr("app.services.application.workflows.media_server_sync.season_runner.media_service.info", fake_info) + monkeypatch.setattr("app.services.application.workflows.media_server_sync.season_runner.library_service.get_media_layout_for_files", fake_layout) + monkeypatch.setattr("app.services.application.workflows.media_server_sync.season_runner.media_server_sync_needs.detect", fake_detect) + monkeypatch.setattr("app.services.application.workflows.media_server_sync.season_runner.media_server_sync_state.record_success", lambda *args, **kwargs: None) + monkeypatch.setattr("app.services.application.workflows.media_server_sync.season_runner.emit_media_server_sync_events", lambda *args, **kwargs: events.append((args, kwargs))) + monkeypatch.setattr(runner, "apply_updates", fake_apply_updates) + + result = await runner.sync_one_season( + JellyfinConfig(id="jellyfin-1", name="Jellyfin", url="http://jellyfin", api_key="token"), + media.media_id, + 1, + files, + _state(media.media_id, last_success_at=None), + time.time(), + MediaServerSyncConfig(write_nfo=False, download_images=False), + ) + assert result assert len(events) == 1 assert events[0][1]["trigger"] == "scheduler" From 6b968d7b79850bbe84d0c0bfa76a967a0cec2163 Mon Sep 17 00:00:00 2001 From: 120318 <20685540+getupbuzz@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:04:59 +0800 Subject: [PATCH 2/2] Address notification noise review feedback --- ...05_indexer_health_notification_cooldown.py | 4 +- .../media_server_sync/season_runner.py | 11 ++- .../workflows/media_server_sync/service.py | 2 + .../config/indexer_client_settings.py | 11 +-- .../tests/test_indexer_site_health_alerts.py | 37 ++++++++++ .../test_media_server_sync_boundaries.py | 72 +++++++++++++++++++ 6 files changed, 128 insertions(+), 9 deletions(-) diff --git a/backend/alembic/versions/0005_indexer_health_notification_cooldown.py b/backend/alembic/versions/0005_indexer_health_notification_cooldown.py index bad5de4..b8232eb 100644 --- a/backend/alembic/versions/0005_indexer_health_notification_cooldown.py +++ b/backend/alembic/versions/0005_indexer_health_notification_cooldown.py @@ -1,6 +1,6 @@ """Track indexer health notification cooldown. -Revision ID: 0005_indexer_health_notification_cooldown +Revision ID: 0005_indexer_notify_cooldown Revises: 0004_subscription_search_cadence Create Date: 2026-07-07 """ @@ -11,7 +11,7 @@ import sqlalchemy as sa -revision = "0005_indexer_health_notification_cooldown" +revision = "0005_indexer_notify_cooldown" down_revision = "0004_subscription_search_cadence" branch_labels = None depends_on = None diff --git a/backend/app/services/application/workflows/media_server_sync/season_runner.py b/backend/app/services/application/workflows/media_server_sync/season_runner.py index 0693b7f..4c3e47d 100644 --- a/backend/app/services/application/workflows/media_server_sync/season_runner.py +++ b/backend/app/services/application/workflows/media_server_sync/season_runner.py @@ -23,6 +23,8 @@ async def sync_one_season( state: MediaServerSyncState, now: float, sync_cfg: MediaServerSyncConfig, + *, + initial_sync: bool | None = None, ) -> bool: media = await media_service.info(media_id, season_number=season_number) if not media: @@ -79,7 +81,10 @@ async def sync_one_season( error=str(exc), ) raise - if self._should_emit_scheduler_completed_event(state, needs.missing_flags): + if self._should_emit_scheduler_completed_event( + state.last_success_at is None if initial_sync is None else initial_sync, + needs.missing_flags, + ): emit_media_server_sync_events( EventType.MEDIA_SERVER_SYNC_COMPLETED, media, @@ -129,8 +134,8 @@ async def apply_updates( ) @staticmethod - def _should_emit_scheduler_completed_event(state: MediaServerSyncState, missing_flags: list[str]) -> bool: - return state.last_success_at is None and set(missing_flags) != {"stale"} + def _should_emit_scheduler_completed_event(initial_sync: bool, missing_flags: list[str]) -> bool: + return initial_sync and set(missing_flags) != {"stale"} media_server_sync_season_runner = MediaServerSyncSeasonRunner() diff --git a/backend/app/services/application/workflows/media_server_sync/service.py b/backend/app/services/application/workflows/media_server_sync/service.py index ee0905c..83a7c93 100644 --- a/backend/app/services/application/workflows/media_server_sync/service.py +++ b/backend/app/services/application/workflows/media_server_sync/service.py @@ -426,6 +426,7 @@ async def _sync_one( sync_cfg: MediaServerSyncConfig, ) -> MediaServerSyncItemResult: state = media_server_sync_state.get_or_create_state(media_server.id, media_id) + initial_sync = state.last_success_at is None out = MediaServerSyncItemResult(media_server_id=media_server.id, media_id=media_id) try: directory_ids = media_server_sync_config.directory_ids_for_media_server(media_server.id) @@ -443,6 +444,7 @@ async def _sync_one( state, now, sync_cfg, + initial_sync=initial_sync, ) out.updated = out.updated or season_updated return out diff --git a/backend/app/services/config/indexer_client_settings.py b/backend/app/services/config/indexer_client_settings.py index babac59..12c62c7 100644 --- a/backend/app/services/config/indexer_client_settings.py +++ b/backend/app/services/config/indexer_client_settings.py @@ -30,7 +30,7 @@ def _get_record(self, indexer_id: str, site_id: str) -> IndexerSiteHealthStatus def _upsert(self, status: IndexerSiteHealthStatus) -> IndexerSiteHealthStatus: return self._repo.upsert(status) - def _emit_unhealthy_event(self, status: IndexerSiteHealthStatus) -> None: + def _emit_unhealthy_event(self, status: IndexerSiteHealthStatus) -> bool: try: event_service.emit( EventCreate( @@ -53,6 +53,7 @@ def _emit_unhealthy_event(self, status: IndexerSiteHealthStatus) -> None: correlation_id=f"indexer:{status.indexer_id}:site:{status.site_id}:unhealthy", ) ) + return True except Exception as exc: logger.error( "Failed to emit indexer site unhealthy event for %s/%s: %s", @@ -60,6 +61,7 @@ def _emit_unhealthy_event(self, status: IndexerSiteHealthStatus) -> None: status.site_id, exc, ) + return False def record_outcomes(self, outcomes: list[IndexerSiteSearchOutcome]) -> None: for outcome in outcomes: @@ -139,12 +141,13 @@ def record_failure( consecutive_failures=consecutive_failures, last_error_message=error_message, notify_pending=consecutive_failures >= INDEXER_SITE_FAILURE_NOTIFY_THRESHOLD, - last_notified_at=now if should_emit else (current.last_notified_at if current else None), + last_notified_at=current.last_notified_at if current else None, client_type=client_type, ) saved = self._upsert(status) - if should_emit: - self._emit_unhealthy_event(saved) + if should_emit and self._emit_unhealthy_event(saved): + saved.last_notified_at = now + saved = self._upsert(saved) return saved @staticmethod diff --git a/backend/tests/test_indexer_site_health_alerts.py b/backend/tests/test_indexer_site_health_alerts.py index 8db6952..b933994 100644 --- a/backend/tests/test_indexer_site_health_alerts.py +++ b/backend/tests/test_indexer_site_health_alerts.py @@ -172,3 +172,40 @@ def test_indexer_site_unhealthy_event_repeats_after_notification_cooldown(monkey ) assert len(emitted_events) == 1 + + +def test_indexer_site_unhealthy_event_failure_does_not_start_cooldown(monkeypatch): + attempts = [] + + def fake_emit(event): + attempts.append(event) + if len(attempts) == 1: + raise RuntimeError("event store unavailable") + + monkeypatch.setattr( + "app.services.config.indexer_client_settings.event_service.emit", + fake_emit, + ) + state = IndexerSiteHealthState(repo=_FakeIndexerSiteHealthRepository()) + + for _ in range(3): + status = state.record_failure( + indexer_id="prowlarr", + indexer_name="Prowlarr", + site_id="audiences", + site_name="Audiences", + error_message="disabled", + ) + + assert status.last_notified_at is None + + status = state.record_failure( + indexer_id="prowlarr", + indexer_name="Prowlarr", + site_id="audiences", + site_name="Audiences", + error_message="still disabled", + ) + + assert len(attempts) == 2 + assert status.last_notified_at is not None diff --git a/backend/tests/test_media_server_sync_boundaries.py b/backend/tests/test_media_server_sync_boundaries.py index 2ea89d0..48a8aa6 100644 --- a/backend/tests/test_media_server_sync_boundaries.py +++ b/backend/tests/test_media_server_sync_boundaries.py @@ -14,6 +14,7 @@ from app.schemas.media_id import MediaID from app.services.application.workflows.media_server_sync.artifacts import media_server_sync_artifacts from app.services.application.workflows.media_server_sync.needs import media_server_sync_needs +from app.services.application.workflows.media_server_sync.service import MediaServerSyncService from app.services.application.workflows.media_server_sync.season_runner import MediaServerSyncSeasonRunner from app.services.audit.workflow_event_emitters import emit_media_server_sync_events from app.services.domain.library.sidecar_files import library_sidecar_files @@ -985,3 +986,74 @@ def fake_emit_media(event, meta=None): assert meta.file_path == str(video) assert meta.episode_number is None assert meta.episode_numbers == [] + + +@pytest.mark.asyncio +async def test_media_server_sync_initial_sync_snapshot_is_shared_across_tv_seasons(monkeypatch, tmp_path: Path): + media = _tv() + state = _state(media.media_id, last_success_at=None) + files = [ + LibraryFile( + id="file-1", + task_id="task-1", + directory_id="dir-1", + media_id=media.media_id, + path=str(tmp_path / "Show" / "Season 01"), + file_name="Show.S01E01.mkv", + created_at=1.0, + ) + ] + initial_sync_values = [] + + async def fake_get_files_by_media_and_directory_ids(media_id, directory_ids): + return files + + async def fake_sync_one_season( + media_server, + media_id, + season_number, + scoped_files, + current_state, + now, + sync_cfg, + *, + initial_sync=None, + ): + initial_sync_values.append(initial_sync) + current_state.last_success_at = now + return True + + monkeypatch.setattr( + "app.services.application.workflows.media_server_sync.service.media_server_sync_state.get_or_create_state", + lambda media_server_id, media_id: state, + ) + monkeypatch.setattr( + "app.services.application.workflows.media_server_sync.service.media_server_sync_config.directory_ids_for_media_server", + lambda media_server_id: ["dir-1"], + ) + monkeypatch.setattr( + "app.services.application.workflows.media_server_sync.service.library_service.get_files_by_media_and_directory_ids", + fake_get_files_by_media_and_directory_ids, + ) + monkeypatch.setattr( + "app.services.application.workflows.media_server_sync.service.library_files_season_numbers", + lambda library_files: [1, 2], + ) + monkeypatch.setattr( + "app.services.application.workflows.media_server_sync.service.library_files_for_season", + lambda library_files, season_number: library_files, + ) + monkeypatch.setattr( + "app.services.application.workflows.media_server_sync.service.media_server_sync_season_runner.sync_one_season", + fake_sync_one_season, + ) + + result = await MediaServerSyncService()._sync_one( + JellyfinConfig(id="jellyfin-1", name="Jellyfin", url="http://jellyfin", api_key="token"), + media.media_id, + time.time(), + MediaServerSyncConfig(write_nfo=False, download_images=False), + ) + + assert result.updated + assert initial_sync_values == [True, True]