Don't accept commits while clean in progress

server/recceiver/cf/model.py

@@ -1,6 +1,6 @@
 import enum
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 
 class PVStatus(enum.Enum):
@@ -57,6 +57,29 @@
     owner: str
     properties: List[CFProperty]
 
+    def has_property(self, prop: Union[CFProperty, CFPropertyName, str]) -> bool:
+        """Check for a property name or an exact CFProperty match.
+
+        Name-only checks ignore owner and value, including None or empty values.
+        A CFProperty checks name, owner, and value using dataclass equality.
+        """
+        if isinstance(prop, CFProperty):
+            return prop in self.properties
+        return self.property(prop) is not None
+
+    def property(self, name: Union[CFPropertyName, str]) -> Optional[CFProperty]:
+        """Return the first property matching an enum or string name, or None."""
+        property_name = name.value if isinstance(name, CFPropertyName) else name
+        return next((prop for prop in self.properties if prop.name == property_name), None)
+
+    def property_value(self, name: Union[CFPropertyName, str]) -> Optional[str]:
+        """Return the first matching property's value, or None if absent.
+
+        Accepts a canonical CFPropertyName or a string for custom properties.
+        """
+        prop = self.property(name)
+        return prop.value if prop is not None else None
+
     def as_dict(self) -> Dict[str, Any]:
         """Serialise to the dict shape expected by pyCFClient."""
         return {

server/recceiver/cf/processor.py

@@ -10,6 +10,7 @@
 from twisted.internet import defer, task
 from twisted.internet.defer import DeferredLock
 from twisted.internet.threads import deferToThread
+from twisted.python.failure import Failure
 from zope.interface import implementer
 
 from recceiver import interfaces, metrics
@@ -42,13 +43,29 @@
         self.name = name  # Override name from service.Service
         self.channel_ioc_ids: Dict[str, List[str]] = defaultdict(list)
         self.iocs: Dict[str, IOCInfo] = {}
-        self.client: Optional[ChannelFinderAdapter] = None
+        self.client: Optional[ChannelFinderAdapter] = PyCFClientAdapter(
+            ChannelFinderClient(
+                BaseURL=self.cf_config.base_url,
+                username=self.cf_config.cf_username,
+                password=self.cf_config.cf_password,
+                verify_ssl=self.cf_config.verify_ssl,
+            ),
+            size_limit=int(self.cf_config.cf_query_limit),
+        )
         self.current_time: Callable[[Optional[str]], str] = get_current_time
         self.lock: DeferredLock = DeferredLock()
+        self._startup_clean: Optional[defer.Deferred] = None
+        self._startup_commit_waiters: Set[defer.Deferred] = set()
+        self._stopping = False
+        self._shutdown: Optional[defer.Deferred] = None
         self._ioc_channels: Dict[str, Set[str]] = defaultdict(set)  # iocid → set of channel names
         self._statusLoop = None
 
     def startService(self):
+        if self.running or self._stopping:
+            raise RuntimeError("Cannot start CF Processor while running or stopping")
+        # Deferreds are one-shot; create a fresh gate for every service start.
+        self._startup_clean = defer.Deferred()
         service.Service.startService(self)
         # Returning a Deferred is not supported by startService(),
         # so instead attempt to acquire the lock synchonously!
@@ -60,8 +77,10 @@
 
         try:
             self._start_service_with_lock()
+            deferToThread(self.clean_on_start).chainDeferred(self._startup_clean)
         except:
             service.Service.stopService(self)
+            self._startup_clean.callback(None)
             raise
         finally:
             self.lock.release()
@@ -78,28 +97,12 @@
     def _start_service_with_lock(self):
         log.info("CF_START with configuration: %s", self.cf_config)
 
-        if self.client is None:  # For setting up mock test client
-            self.client = PyCFClientAdapter(
-                ChannelFinderClient(
-                    BaseURL=self.cf_config.base_url,
-                    username=self.cf_config.cf_username,
-                    password=self.cf_config.cf_password,
-                    verify_ssl=self.cf_config.verify_ssl,
-                ),
-                size_limit=int(self.cf_config.cf_query_limit),
-            )
-            try:
-                cf_properties = set(self.client.get_property_names())
-                self._setup_cf_properties(cf_properties)
-            except ConnectionError:
-                log.exception("Cannot connect to Channelfinder service")
-                raise
-            else:
-                if self.cf_config.clean_on_start:
-                    log.info("CF Clean: scheduling background startup sweep")
-                    from twisted.internet import reactor
-
-                    reactor.callLater(0, self._start_background_clean)
+        try:
+            cf_properties = set(self.client.get_property_names())
+            self._setup_cf_properties(cf_properties)
+        except ConnectionError:
+            log.exception("Cannot connect to Channelfinder service")
+            raise
 
     def _setup_cf_properties(self, cf_properties: Set[str]) -> None:
         """Compute required CF properties, register any missing ones, and cache state.
@@ -149,11 +152,41 @@
         log.debug("record_property_names_list = %s", self.record_property_names_list)
 
     def stopService(self):
+        """Wait for startup and commit workers before performing stop cleanup.
+
+        Startup-waiting transactions are cancelled, not drained. A restart is
+        allowed only after shutdown finishes, so old workers cannot cross into
+        the next startup cycle.
+        """
+        if self._stopping:
+            return self._shutdown
+        if not self.running:
+            return defer.succeed(None)
         log.info("CF_STOP")
         if self._statusLoop is not None and self._statusLoop.running:
             self._statusLoop.stop()
         service.Service.stopService(self)
-        return self.lock.run(self._stop_service_with_lock)
+        self._stopping = True
+        shutdown = defer.Deferred()
+        self._shutdown = shutdown
+        shutdown.addCallback(lambda _: self.lock.run(self._stop_service_with_lock))
+
+        def finished(result):
+            self._stopping = False
+            return result
+
+        shutdown.addBoth(finished)
+        for waiter in list(self._startup_commit_waiters):
+            waiter.cancel()
+
+        def startup_finished(_result):
+            shutdown.callback(None)
+            # Startup failures have already reached commit callers. Consume the
+            # shared gate on shutdown, when no further commits can observe it.
+            return None
+
+        self._startup_clean.addBoth(startup_finished)
+        return shutdown
 
     def _stop_service_with_lock(self):
         """Stop the CFProcessor service with lock held.
@@ -166,13 +199,29 @@
         if self.cf_config.clean_on_stop:
             return deferToThread(self.clean_service)
 
-    def _start_background_clean(self):
-        log.info("CF Clean: background startup sweep beginning")
-        deferToThread(self.clean_service).addErrback(lambda err: log.error("CF Clean background sweep failed: %s", err))
-
     def commit(self, transaction_record: interfaces.ITransaction) -> defer.Deferred:
-        """Commit a transaction to Channelfinder."""
-        return self.lock.run(self._commit_with_lock, transaction_record)
+        """Commit a transaction to Channelfinder, waiting for startup cleanup if enabled."""
+        if not self.running:
+            return defer.fail(defer.CancelledError("CF Processor is not running"))
+        result: defer.Deferred = defer.Deferred(lambda d: self._startup_commit_waiters.discard(d))
+        self._startup_commit_waiters.add(result)
+        # Returning the lock Deferred from a callback preserves cancellation,
+        # including waiting for a cancelled commit's worker to finish.
+        result.addCallback(lambda _: self.lock.run(self._commit_with_lock, transaction_record))
+
+        def start_commit(clean_result):
+            self._startup_commit_waiters.discard(result)
+            # Cancellation while waiting must not cancel the shared startup gate
+            # or allow this transaction to run once cleanup completes.
+            if not result.called:
+                if isinstance(clean_result, Failure):
+                    result.errback(clean_result)
+                else:
+                    result.callback(None)
+            return clean_result
+
+        self._startup_clean.addBoth(start_commit)
+        return result
 
     def _commit_with_lock(self, transaction: interfaces.ITransaction) -> defer.Deferred:
         """Bridge the blocking commit thread to a cancellable Deferred.
@@ -181,6 +230,8 @@
         Deferred (d) around it so that cancelling d (e.g. on service stop) sets
         self.cancelled=True, which _assert_not_cancelled picks up mid-push.
         """
+        if not self.running:
+            return defer.fail(defer.CancelledError("CF Processor stopped while waiting for commit lock"))
         self.cancelled = False
 
         t = deferToThread(self._commit_with_thread, transaction)
@@ -435,6 +486,11 @@
         if len(self.channel_ioc_ids[record_name]) == 0:
             del self.channel_ioc_ids[record_name]
 
+    def clean_on_start(self) -> None:
+        if self.cf_config.clean_on_start:
+            log.info("CF Clean: starting background clean")
+            self.clean_service()
+
     def clean_service(self) -> None:
         """Mark all channels belonging to this recceiver as 'Inactive'."""
         sleep = 1

server/tests/integration/test_single_ioc.py

@@ -1,11 +1,15 @@
 import logging
 import time
 from pathlib import Path
+from uuid import uuid4
 
 import pytest
 from channelfinder import ChannelFinderClient
 from testcontainers.compose import DockerCompose
 
+from recceiver.cf.adapter import PyCFClientAdapter
+from recceiver.cf.model import CFChannel, CFProperty, CFPropertyName, PVStatus
+
 from .cf_client import (
     BASE_IOC_CHANNEL_COUNT,
     DEFAULT_CHANNEL_NAME,
@@ -39,6 +43,11 @@
     return create_client_and_wait(setup_compose, expected_channel_count=BASE_IOC_CHANNEL_COUNT)
 
 
+@pytest.fixture(scope="class")
+def cf_adapter(setup_compose: DockerCompose) -> PyCFClientAdapter:  # noqa: F811
+    return PyCFClientAdapter(create_client_and_wait(setup_compose, expected_channel_count=BASE_IOC_CHANNEL_COUNT))
+
+
 class TestRestartIOC:
     def test_channels_same_after_restart(self, setup_compose: DockerCompose, cf_client: ChannelFinderClient) -> None:  # noqa: F811
         channels_begin = cf_client.find(name="*")
@@ -154,6 +163,66 @@
         assert all(INACTIVE_PROPERTY in ch["properties"] for ch in channels_inactive)
 
 
+class TestLiveIOCRestartRecceiver:
+    def test_restart_restores_live_channels_and_cleans_stale_channels(
+        self, setup_compose: DockerCompose, cf_adapter: PyCFClientAdapter
+    ) -> None:  # noqa: F811
+        baseline = cf_adapter.find_by_names(["*"])
+        assert len(baseline) == BASE_IOC_CHANNEL_COUNT
+        expected_names = {channel.name for channel in baseline}
+        live_names = sorted(expected_names)
+        receiver = baseline[0].property(CFPropertyName.RECCEIVER_ID)
+        assert receiver is not None and receiver.value is not None
+        active = CFProperty(CFPropertyName.PV_STATUS.value, "admin", PVStatus.ACTIVE.value)
+        inactive = CFProperty(CFPropertyName.PV_STATUS.value, "admin", PVStatus.INACTIVE.value)
+        stale_name = f"RECSYNC:STALE:{uuid4().hex}"
+        receiver_id = setup_compose.get_container("recc1").ID
+
+        def recovered(adapter: PyCFClientAdapter) -> bool:
+            channels = adapter.find_active_for_recceiver(receiver.value)
+            stale = adapter.find_by_names([stale_name])
+            return (
+                {channel.name for channel in channels} == expected_names
+                and all(
+                    channel.property_value(CFPropertyName.TIME) not in (None, previous_times[channel.name])
+                    for channel in channels
+                )
+                and len(stale) == 1
+                and stale[0].has_property(inactive)
+            )
+
+        assert wait_for_sync(
+            cf_adapter,
+            lambda adapter: {ch.name for ch in adapter.find_active_for_recceiver(receiver.value)} == expected_names,
+        ), "IOC channels did not become Active before restart"
+        receiver_stopped = False
+        try:
+            # The class-scoped Compose fixture removes the seeded channel along
+            # with the rest of the ChannelFinder data during teardown.
+            cf_adapter.set_channels([CFChannel(stale_name, "admin", [receiver, active])])
+            assert wait_for_sync(
+                cf_adapter,
+                lambda adapter: any(ch.has_property(active) for ch in adapter.find_by_names([stale_name])),
+            )
+            # Leave the IOC running so its upload competes with startup cleanup.
+            kill_container(setup_compose, "recc1")
+            receiver_stopped = True
+            # Startup cleanup changes only pvStatus. A changed timestamp proves
+            # each live channel was uploaded again, rather than left Active.
+            previous_times = {
+                channel.name: channel.property_value(CFPropertyName.TIME)
+                for channel in cf_adapter.find_by_names(live_names)
+            }
+            assert set(previous_times) == expected_names
+            assert all(value is not None for value in previous_times.values()), "Live channels are missing timestamps"
+            start_container(setup_compose, container_id=receiver_id)
+            receiver_stopped = False
+            assert wait_for_sync(cf_adapter, recovered), "Fresh IOC upload and stale-channel cleanup did not complete"
+        finally:
+            if receiver_stopped:
+                start_container(setup_compose, container_id=receiver_id)
+
+
 class TestMoveIocHost:
     def test_move_ioc_host(
         self,