SEP-1734: Pass the previous snapshot to RefreshCallback so rebind callbacks can act on what changed

app/core/settings_override/__init__.py

@@ -27,6 +27,7 @@
     "SettingClassEnum",
     "SettingOverride",
     "SettingsOverrideManager",
+    "SnapshotChange",
     "build_snapshot",
     "coerce_field_value",
     "coerce_nested_field_value",
@@ -59,6 +60,7 @@
     refresh_all,
     RefreshCallback,
     settings_override_refresher,
+    SnapshotChange,
     start_refresh_task,
 )
 from app.core.settings_override.manager import SettingsOverrideManager

app/core/settings_override/lifecycle.py

@@ -22,6 +22,7 @@
     "ProxyEntry",
     "ProxyRegistry",
     "RefreshCallback",
+    "SnapshotChange",
     "fire_change_callbacks",
     "publish_snapshot",
     "refresh_all",
@@ -62,11 +63,31 @@ def _drain_cancelled_seed_task(task: asyncio.Task) -> None:
         task.exception()
 
 
+class SnapshotChange(NamedTuple):
+    """Represent the override snapshots on either side of a republish.
+
+    A snapshot holds active overrides only -- never an effective view -- so a
+    key may be absent from ``previous``, ``current``, or both. When a key is
+    absent the prior (or new) effective value is the YAML/env one reachable
+    through the proxy's wrapped instance.
+
+    :param previous: The snapshot in effect before the republish.
+    :type previous: Mapping[str, object]
+    :param current: The snapshot now in effect.
+    :type current: Mapping[str, object]
+    """
+
+    previous: Mapping[str, object]
+    current: Mapping[str, object]
+
+
 #: A rebind callback fired when a watched ``(setting_class, key)`` override
-#: changes value between refresh cycles. The callback receives the new effective
-#: snapshot mapping for its setting class; any exception it raises is caught and
-#: logged by :func:`refresh_all` so one failing callback cannot break the cycle.
-RefreshCallback = Callable[[Mapping[str, object]], Awaitable[None]]
+#: changes value between refresh cycles. The callback receives a
+#: :class:`SnapshotChange` carrying the override snapshots on either side of
+#: the republish (overrides-only; a key may be absent from either side). Any
+#: exception it raises is caught and logged by :func:`fire_change_callbacks`
+#: so one failing callback cannot break the cycle.
+RefreshCallback = Callable[[SnapshotChange], Awaitable[None]]
 CallbackRegistry = dict[tuple[SettingClassEnum, str], RefreshCallback]
 
 
@@ -124,8 +145,15 @@ async def fire_change_callbacks(
 
     Compares ``previous`` against ``current`` and, for each ``(setting_class,
     key)`` whose value differs and has a registered callback, awaits the
-    callback. Each callback runs inside its own ``try/except`` so one failure
-    neither aborts the cycle nor blocks the remaining callbacks.
+    callback with a :class:`SnapshotChange` carrying both mappings. Each
+    callback runs inside its own ``try/except`` so one failure neither aborts
+    the cycle nor blocks the remaining callbacks.
+
+    Both mappings are override snapshots only -- never an effective view -- so a
+    key may be absent from either side (for example on the override-delete path
+    the changed key is gone from ``current``). Callbacks that need the previous
+    effective value when the key is absent recover it through the proxy's
+    wrapped YAML/env instance.
 
     Shared by the background refresher (:func:`refresh_all`, diffing the snapshot
     it just rebuilt) and the settings-API PATCH/DELETE handlers (diffing the
@@ -137,19 +165,20 @@ async def fire_change_callbacks(
     :type callbacks: CallbackRegistry
     :param setting_class: The class whose snapshot was just republished.
     :type setting_class: SettingClassEnum
-    :param previous: The snapshot in effect before the republish.
+    :param previous: The override snapshot in effect before the republish.
     :type previous: Mapping[str, object]
-    :param current: The snapshot now in effect.
+    :param current: The override snapshot now in effect.
     :type current: Mapping[str, object]
     """
+    change = SnapshotChange(previous, current)
     for key in previous.keys() | current.keys():
         if previous.get(key) == current.get(key):
             continue
         callback = callbacks.get((setting_class, key))
         if callback is None:
             continue
         try:
-            await callback(current)
+            await callback(change)
         except Exception:
             logger.exception(
                 "Rebind callback for %s.%s failed; keeping previous binding",

app/sep/main.py

@@ -16,7 +16,7 @@
 """Define SEP routes."""
 
 import logging.config
-from collections.abc import AsyncGenerator, Callable, Mapping
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from copy import deepcopy
 from traceback import format_exception
@@ -28,7 +28,7 @@
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
 from fastapi.security import OAuth2PasswordRequestForm
-from pydantic import HttpUrl, ValidationError
+from pydantic import ValidationError
 from starlette.staticfiles import StaticFiles
 
 from app import __summary__, __version__
@@ -45,8 +45,10 @@
 from app.core.settings_override.lifecycle import (
     RefreshCallback,
     settings_override_refresher,
+    SnapshotChange,
 )
 from app.core.settings_override.models import SettingClassEnum
+from app.core.settings_override.proxy import OverridableSettingsProxy
 from app.core.utils import run_pydantic_type_validator
 from app.core.utils.fields import URIPath
 from app.inventory.config import inventory_settings
@@ -126,7 +128,8 @@ async def sep_startup() -> None:
 def _make_remote_api_rebinder(
     app: FastAPI,
     name: str,
-    endpoint_getter: Callable[[], HttpUrl],
+    proxy: OverridableSettingsProxy,
+    key: str,
     **ssl: Any,
 ) -> RefreshCallback:
     """Build a rebind callback for an ``app.state`` RemoteAPI endpoint override.
@@ -136,29 +139,39 @@ def _make_remote_api_rebinder(
     the new endpoint and the old one closed. Under the combined ``app.main:app``
     no ``app.state`` client exists -- ``get_*_client`` falls back to the
     registry-cached ``get_remote_api`` per request, which already key-misses to
-    the new HOT endpoint -- so the callback only evicts any stale client left on
-    the new endpoint.
+    the new HOT endpoint -- so the callback evicts the ordered de-duplicated set
+    of previous-and-current endpoints (covering endpoint moves as well as
+    same-endpoint credential/SSL changes). When ``key`` is absent from
+    ``change.previous`` (override created), the prior effective value is the
+    YAML/env one from the proxy's wrapped instance.
 
     :param app: The FastAPI application whose ``state`` holds the client.
     :type app: FastAPI
     :param name: The ``app.state`` attribute name (``inventory_api`` /
         ``tasks_api``).
     :type name: str
-    :param endpoint_getter: A zero-argument callable returning the current
-        (override-aware) endpoint.
-    :type endpoint_getter: Callable[[], HttpUrl]
+    :param proxy: The overridable settings proxy that owns the endpoint field.
+    :type proxy: OverridableSettingsProxy
+    :param key: The top-level snapshot key for the endpoint field.
+    :type key: str
     :param ssl: SSL keyword arguments forwarded to :class:`RemoteAPI` (not HOT,
         captured once at wiring time).
     :type ssl: Any
     :return: The rebind callback.
     :rtype: RefreshCallback
     """
 
-    async def _rebind(_: Mapping[str, object]) -> None:
-        new_endpoint = endpoint_getter()
+    async def _rebind(change: SnapshotChange) -> None:
+        new_endpoint = getattr(proxy, key)
         old = getattr(app.state, name, None)
         if old is None:
-            await settings.invalidate_client(str(new_endpoint))
+            previous_endpoint = change.previous.get(key)
+            if previous_endpoint is None:
+                previous_endpoint = getattr(proxy._resolve(), key)  # noqa: SLF001
+            for endpoint in dict.fromkeys(
+                str(ep) for ep in (previous_endpoint, new_endpoint) if ep is not None
+            ):
+                await settings.invalidate_client(endpoint)
             return
         try:
             new_api = await RemoteAPI(endpoint=new_endpoint, **ssl).open()
@@ -171,7 +184,7 @@ async def _rebind(_: Mapping[str, object]) -> None:
     return _rebind
 
 
-async def _apply_logging_dictconfig(_: Mapping[str, object]) -> None:
+async def _apply_logging_dictconfig(_: SnapshotChange) -> None:
     """Re-apply ``logging.config.dictConfig`` after a global ``LOGGING`` override.
 
     ``LOGGING`` is a HOT field, but ``LOGGING_CONFIG`` (the dict handed to
@@ -183,8 +196,8 @@ async def _apply_logging_dictconfig(_: Mapping[str, object]) -> None:
     process without a restart. Failures are logged and swallowed: a malformed
     config must not take the process down mid-request.
 
-    :param _: The new effective ``Settings`` snapshot mapping (unused -- the level
-        is re-read from the proxy).
+    :param _: The override snapshots on either side of the republish (unused --
+        the level is re-read from the proxy).
     """
     try:
         config = deepcopy(settings.LOGGING_CONFIG)
@@ -195,7 +208,7 @@ async def _apply_logging_dictconfig(_: Mapping[str, object]) -> None:
         logger.exception("Failed to re-apply logging config after LOGGING override")
 
 
-async def _reseed_system_periodic_tasks(_: Mapping[str, object]) -> None:
+async def _reseed_system_periodic_tasks(_: SnapshotChange) -> None:
     """Re-seed the SEP beat schedule after a hot interval override.
 
     Wired for both ``SnippetsSettings.SYNC_INTERVAL`` (``sep__sync_snippets``) and
@@ -211,8 +224,8 @@ async def _reseed_system_periodic_tasks(_: Mapping[str, object]) -> None:
     Updating the ``IntervalSchedule`` bumps ``PeriodicTaskChanged.last_update``, so
     Celery beat reloads the schedule on its next scheduler tick without a restart.
 
-    :param _: The new effective settings snapshot mapping (unused -- the interval is
-        re-read from the proxy by the task-set builder).
+    :param _: The override snapshots on either side of the republish (unused -- the
+        interval is re-read from the proxy by the task-set builder).
     """
     await init_periodic_tasks_db(get_system_periodic_tasks(), "sep__")
 
@@ -255,15 +268,17 @@ async def sep_overrides_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         ): _make_remote_api_rebinder(
             app,
             "inventory_api",
-            lambda: sep_settings.INVENTORY_ENDPOINT,
+            sep_settings,
+            "INVENTORY_ENDPOINT",
             ssl_cafile=settings.SSL_CAFILE,
             ssl_keyfile=inventory_settings.SSL_KEYFILE,
             ssl_certfile=inventory_settings.SSL_CERTFILE,
         ),
         (SettingClassEnum.SEP_SETTINGS, "TASKS_ENDPOINT"): _make_remote_api_rebinder(
             app,
             "tasks_api",
-            lambda: sep_settings.TASKS_ENDPOINT,
+            sep_settings,
+            "TASKS_ENDPOINT",
             ssl_cafile=settings.SSL_CAFILE,
             ssl_keyfile=tasks_settings.SSL_KEYFILE,
             ssl_certfile=tasks_settings.SSL_CERTFILE,

app/sep/settings_override.py

@@ -25,20 +25,20 @@
 ``celery.tasks``.
 """
 
-from collections.abc import Mapping
-from typing import Any
+from typing import Any, cast
 
 from celery.signals import worker_process_init, worker_process_shutdown
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.celery import celery
 from app.core.alerts.config import alert_settings, AlertSettings
-from app.core.config import Settings, settings
+from app.core.config import PMMSettings, Settings, settings
 from app.core.settings_override.lifecycle import (
     CallbackRegistry,
     ProxyEntry,
     ProxyRegistry,
     publish_snapshot,
+    SnapshotChange,
 )
 from app.core.settings_override.models import SettingClassEnum
 from app.core.settings_override.worker import WorkerRefresher
@@ -119,21 +119,28 @@ async def republish_sep_settings_snapshot(session: AsyncSession) -> None:
     await publish_snapshot(sep_settings, session, SEPSettings)
 
 
-async def invalidate_pmm_clients(_: Mapping[str, object]) -> None:
-    """Evict the cached PMM client on the current endpoint after a ``PMM`` override.
+async def invalidate_pmm_clients(change: SnapshotChange) -> None:
+    """Evict cached PMM clients on the previous and current endpoints after a ``PMM`` override.
 
-    A same-endpoint change (credentials, SSL) evicts the now-stale client so the
-    next :class:`PMMSyncer` key-misses to a fresh one via its ``default_factory``
-    PMM read. Known limitation: an endpoint change leaves the client keyed by the
-    old endpoint cached until ``close_all`` closes it at shutdown; syncers key on
-    the new endpoint, so nothing reads it in the meantime.
+    Evicts the ordered de-duplicated set of previous-and-current endpoints so a
+    same-endpoint change (credentials, SSL) collapses to a single eviction, while
+    an endpoint change also drops the client keyed by the endpoint no longer in
+    use. The next :class:`PMMSyncer` key-misses to a fresh client via its
+    ``default_factory`` PMM read. When ``PMM`` is absent from ``change.previous``
+    (override created), the prior effective value is the YAML/env one from the
+    proxy's wrapped instance.
 
-    :param _: The new effective ``Settings`` snapshot mapping (unused -- the
-        current PMM endpoint is re-read from the proxy).
+    :param change: The override snapshots on either side of the republish.
     """
-    endpoint = settings.PMM.endpoint
-    if endpoint is not None:
-        await settings.invalidate_client(str(endpoint))
+    previous_pmm = cast(PMMSettings | None, change.previous.get("PMM"))
+    if previous_pmm is None:
+        previous_pmm = settings._resolve().PMM  # noqa: SLF001
+    for endpoint in dict.fromkeys(
+        str(pmm.endpoint)
+        for pmm in (previous_pmm, settings.PMM)
+        if pmm is not None and pmm.endpoint is not None
+    ):
+        await settings.invalidate_client(endpoint)
 
 
 #: The callbacks the worker refresher registers -- deliberately a strict subset

app/tasks/main.py

@@ -17,7 +17,7 @@
 
 import json
 import logging.config
-from collections.abc import AsyncGenerator, Mapping
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from typing import Any
 
@@ -33,6 +33,7 @@
     CallbackRegistry,
     ProxyEntry,
     settings_override_refresher,
+    SnapshotChange,
 )
 from app.core.settings_override.models import SettingClassEnum
 from app.tasks.anonymizer.config import anonymizer_settings, AnonymizerSettings
@@ -53,7 +54,7 @@
 celery_logger = get_task_logger(__name__)
 
 
-async def _reconcile_nomad(_: Mapping[str, object]) -> None:
+async def _reconcile_nomad(_: SnapshotChange) -> None:
     """Rebind the live Nomad executor when its override changed.
 
     Registered as the ``(TASKS_SETTINGS, NOMAD)`` rebind callback by
@@ -62,9 +63,9 @@ async def _reconcile_nomad(_: Mapping[str, object]) -> None:
     shutdown can find ``tasks_app.state.nomad_lifecycle`` already gone; the
     rebind is skipped in that window rather than raising a noisy callback error.
 
-    :param _: The new effective ``TasksSettings`` snapshot mapping. Unused -- the
-        holder reads the live ``NOMAD`` config itself when reconciling.
-    :type _: Mapping[str, object]
+    :param _: The override snapshots on either side of the republish. Unused --
+        the holder reads the live ``NOMAD`` config itself when reconciling.
+    :type _: SnapshotChange
     """
     holder = getattr(tasks_app.state, "nomad_lifecycle", None)
     if holder is not None:

changelog.d/SEP-1734.fixed.md

@@ -0,0 +1 @@
+A runtime override that changes the PMM, Inventory, or Tasks endpoint now evicts the cached client on the endpoint no longer in use, instead of leaving it open until process shutdown.

tests/app/core/settings_override/test_lifecycle.py

@@ -29,8 +29,10 @@
 from app.core.db.utils import get_async_session_maker_from_engine
 from app.core.settings_override.cache import build_snapshot
 from app.core.settings_override.lifecycle import (
+    fire_change_callbacks,
     ProxyEntry,
     refresh_all,
+    SnapshotChange,
     start_refresh_task,
 )
 from app.core.settings_override.manager import SettingsOverrideManager
@@ -426,6 +428,32 @@ async def _callback(_: object) -> None:
     assert proxy.CONNECTIVITY_CHECK_DEFAULT is override_value
 
 
+@pytest.mark.asyncio
+async def test_fire_change_callbacks_delivers_snapshot_change_on_delete() -> None:
+    """Callbacks receive a SnapshotChange; on delete the key is absent from current."""
+    previous = {"CONNECTIVITY_CHECK_DEFAULT": True}
+    current: dict[str, object] = {}
+    received: list[object] = []
+
+    async def _callback(change: object) -> None:
+        received.append(change)
+
+    await fire_change_callbacks(
+        {_CALLBACK_KEY: _callback},
+        SettingClassEnum.SEP_SETTINGS,
+        previous,
+        current,
+    )
+
+    assert len(received) == 1
+    change = received[0]
+    assert isinstance(change, SnapshotChange)
+    assert change.previous == previous
+    assert change.current == current
+    assert "CONNECTIVITY_CHECK_DEFAULT" not in change.current
+    assert change.previous["CONNECTIVITY_CHECK_DEFAULT"] is True
+
+
 @pytest.mark.asyncio
 async def test_refresh_all_skips_callback_for_unchanged_key(
     session_maker: async_sessionmaker,