diff --git a/server/recceiver/cf/model.py b/server/recceiver/cf/model.py index 8dff6d26..12ae3416 100644 --- a/server/recceiver/cf/model.py +++ b/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 @@ class CFChannel: 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 { diff --git a/server/recceiver/cf/processor.py b/server/recceiver/cf/processor.py index 4ae97c75..329a4bb6 100755 --- a/server/recceiver/cf/processor.py +++ b/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 @@ def __init__(self, name: Optional[str], conf: ConfigAdapter): 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 @@ def startService(self): 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 _logStatus(self): 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 @@ def _setup_cf_properties(self, cf_properties: Set[str]) -> None: 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 @@ def _stop_service_with_lock(self): 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 @@ def _commit_with_lock(self, transaction: interfaces.ITransaction) -> defer.Defer 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 @@ def remove_channel(self, record_name: str, iocid: str) -> None: 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 diff --git a/server/tests/integration/test_single_ioc.py b/server/tests/integration/test_single_ioc.py index 5dacf812..0b42aeb5 100644 --- a/server/tests/integration/test_single_ioc.py +++ b/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 @@ def cf_client(setup_compose: DockerCompose) -> ChannelFinderClient: # noqa: F81 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 @@ def test_startup_sweep_marks_stale_channels_inactive( 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, diff --git a/server/tests/unit/cf/test_cleanup.py b/server/tests/unit/cf/test_cleanup.py new file mode 100644 index 00000000..bb36e907 --- /dev/null +++ b/server/tests/unit/cf/test_cleanup.py @@ -0,0 +1,136 @@ +"""Cleanup scope and retry tests, independent of Deferred lifecycle ordering.""" + +import pytest +from requests import RequestException + +from recceiver.cf.model import PVStatus +from recceiver.cf.processor import CFProcessor +from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_channel +from tests.unit.cf.mock_adapter import MockCFAdapter +from tests.unit.conftest import make_adapter + + +class CleanupAdapter(MockCFAdapter): + def __init__(self, batch_size=None): + super().__init__() + self.batch_size = batch_size + self.query_failures = 0 + self.update_failures = 0 + self.queries = 0 + self.update_attempts = 0 + self.updates = [] + + def find_active_for_recceiver(self, recceiverid): + self.queries += 1 + if self.query_failures: + self.query_failures -= 1 + raise RequestException("temporary query failure") + channels = super().find_active_for_recceiver(recceiverid) + return channels if self.batch_size is None else channels[: self.batch_size] + + def update_property(self, prop, channel_names): + self.update_attempts += 1 + if self.update_failures: + self.update_failures -= 1 + raise RequestException("temporary update failure") + super().update_property(prop, channel_names) + self.updates.append((prop, list(channel_names))) + + +@pytest.fixture +def cleanup(monkeypatch): + proc = CFProcessor("test", make_adapter(values={"recceiverId": DEFAULT_RECCEIVER_ID})) + adapter = CleanupAdapter() + proc.client = adapter + sleeps = [] + monkeypatch.setattr("recceiver.cf.processor.time.sleep", sleeps.append) + return proc, adapter, sleeps + + +def channel_status(adapter, name): + return next(p.value for p in adapter._channels[name].properties if p.name == "pvStatus") + + +def test_cleanup_only_updates_this_receivers_active_channels(cleanup): + proc, adapter, sleeps = cleanup + adapter.set_channels( + [ + make_channel("ACTIVE:PV"), + make_channel("INACTIVE:PV", active=False), + make_channel("FOREIGN:PV", recceiver_id="another-receiver"), + ] + ) + proc.clean_service() + + assert channel_status(adapter, "ACTIVE:PV") == PVStatus.INACTIVE.value + assert channel_status(adapter, "INACTIVE:PV") == PVStatus.INACTIVE.value + assert channel_status(adapter, "FOREIGN:PV") == PVStatus.ACTIVE.value + assert len(adapter.updates) == 1 + prop, names = adapter.updates[0] + assert names == ["ACTIVE:PV"] + assert prop.name == "pvStatus" + assert prop.value == PVStatus.INACTIVE.value + assert prop.owner == proc.cf_config.username + assert sleeps == [] + + +def test_empty_cleanup_makes_no_updates(cleanup): + proc, adapter, sleeps = cleanup + proc.clean_service() + assert adapter.queries == 1 + assert adapter.update_attempts == 0 + assert sleeps == [] + + +def test_cleanup_exhausts_all_query_batches(cleanup): + proc, adapter, sleeps = cleanup + adapter.batch_size = 2 + adapter.set_channels([make_channel(f"PV:{index}") for index in range(5)]) + proc.clean_service() + + assert [names for _, names in adapter.updates] == [["PV:0", "PV:1"], ["PV:2", "PV:3"], ["PV:4"]] + assert adapter.queries == 4 + assert all(channel_status(adapter, name) == PVStatus.INACTIVE.value for name in adapter._channels) + assert sleeps == [] + + +@pytest.mark.parametrize("failure_kind", ["query_failures", "update_failures"]) +def test_cleanup_recovers_from_transient_failure(cleanup, failure_kind): + proc, adapter, sleeps = cleanup + proc.running = True + setattr(adapter, failure_kind, 1) + adapter.set_channels([make_channel("PV:1")]) + proc.clean_service() + + assert sleeps == [1] + assert channel_status(adapter, "PV:1") == PVStatus.INACTIVE.value + assert len(adapter.updates) == 1 + assert adapter.queries == 3 + assert adapter.update_attempts == (2 if failure_kind == "update_failures" else 1) + + +def test_cleanup_retry_backoff_is_capped_and_recovers_while_running(cleanup): + proc, adapter, sleeps = cleanup + proc.running = True + adapter.query_failures = 14 + adapter.set_channels([make_channel("PV:1")]) + proc.clean_service() + + assert len(sleeps) == 14 + assert sleeps[:3] == [1, 1.5, 2.25] + assert max(sleeps) == 60 + assert sleeps[-1] == 60 + assert channel_status(adapter, "PV:1") == PVStatus.INACTIVE.value + + +def test_stopped_cleanup_abandons_persistent_failure_with_bounded_retries(cleanup): + proc, adapter, sleeps = cleanup + proc.running = False + adapter.query_failures = 100 + adapter.set_channels([make_channel("PV:1")]) + proc.clean_service() + + assert sleeps == [1, 1.5, 2.25, 3.375] + assert adapter.queries == 4 + assert adapter.update_attempts == 0 + assert channel_status(adapter, "PV:1") == PVStatus.ACTIVE.value diff --git a/server/tests/unit/cf/test_model.py b/server/tests/unit/cf/test_model.py index 5a917107..5cc298bc 100644 --- a/server/tests/unit/cf/test_model.py +++ b/server/tests/unit/cf/test_model.py @@ -1,3 +1,5 @@ +import pytest + from recceiver.cf.model import CFChannel, CFProperty, CFPropertyName, IOCInfo, PVStatus @@ -39,6 +41,91 @@ def test_from_dict_roundtrip(self): class TestCFChannel: + @pytest.mark.parametrize("name", [CFPropertyName.TIME, "time"]) + def test_property_returns_matching_object(self, name): + expected = CFProperty("time", "user", "timestamp") + channel = CFChannel("PV:1", "admin", [CFProperty("pvStatus", "admin", "Active"), expected]) + assert channel.property(name) is expected + + def test_property_accepts_custom_name_and_none_value(self): + expected = CFProperty("custom", "user", None) + channel = CFChannel("PV:1", "admin", [expected]) + assert channel.property("custom") is expected + + @pytest.mark.parametrize("properties", [[], [CFProperty("pvStatus", "admin", "Active")]]) + def test_property_returns_none_when_absent(self, properties): + channel = CFChannel("PV:1", "admin", properties) + assert channel.property(CFPropertyName.TIME) is None + + def test_property_returns_first_match(self): + first = CFProperty("time", "admin", "first") + channel = CFChannel("PV:1", "admin", [first, CFProperty("time", "admin", "second")]) + assert channel.property(CFPropertyName.TIME) is first + + @pytest.mark.parametrize("name", [CFPropertyName.TIME, "time"]) + @pytest.mark.parametrize("value", ["timestamp", None, ""]) + def test_has_property_accepts_enum_or_string_regardless_of_value(self, name, value): + channel = CFChannel("PV:1", "admin", [CFProperty("time", "admin", value)]) + assert channel.has_property(name) + + def test_has_property_accepts_custom_property_name(self): + channel = CFChannel("PV:1", "admin", [CFProperty("custom", "admin", "value")]) + assert channel.has_property("custom") + + @pytest.mark.parametrize("properties", [[], [CFProperty("pvStatus", "admin", "Active")]]) + def test_has_property_returns_false_when_name_absent(self, properties): + channel = CFChannel("PV:1", "admin", properties) + assert not channel.has_property(CFPropertyName.TIME) + + def test_has_property_matches_equal_property(self): + channel = CFChannel("PV:1", "admin", [CFProperty("pvStatus", "admin", "Inactive")]) + assert channel.has_property(CFProperty("pvStatus", "admin", "Inactive")) + + @pytest.mark.parametrize( + "prop", + [ + CFProperty("other", "admin", "Inactive"), + CFProperty("pvStatus", "other", "Inactive"), + CFProperty("pvStatus", "admin", "Active"), + ], + ) + def test_has_property_requires_matching_name_owner_and_value(self, prop): + channel = CFChannel("PV:1", "admin", [CFProperty("pvStatus", "admin", "Inactive")]) + assert not channel.has_property(prop) + + def test_has_property_returns_false_for_exact_match_on_empty_channel(self): + channel = CFChannel("PV:1", "admin", []) + assert not channel.has_property(CFProperty("pvStatus", "admin", "Inactive")) + + @pytest.mark.parametrize("name", [CFPropertyName.TIME, "time"]) + def test_property_value_accepts_enum_or_string(self, name): + channel = CFChannel( + "PV:1", + "admin", + [CFProperty("pvStatus", "admin", "Active"), CFProperty("time", "admin", "timestamp")], + ) + assert channel.property_value(name) == "timestamp" + + def test_property_value_accepts_custom_property(self): + channel = CFChannel("PV:1", "admin", [CFProperty("custom", "admin", "custom value")]) + assert channel.property_value("custom") == "custom value" + + @pytest.mark.parametrize("properties", [[], [CFProperty("pvStatus", "admin", "Active")]]) + def test_property_value_returns_none_when_absent(self, properties): + channel = CFChannel("PV:1", "admin", properties) + assert channel.property_value(CFPropertyName.TIME) is None + + @pytest.mark.parametrize("value", [None, ""]) + def test_property_value_preserves_none_or_empty_value(self, value): + channel = CFChannel("PV:1", "admin", [CFProperty("time", "admin", value)]) + assert channel.property_value(CFPropertyName.TIME) == value + + def test_property_value_returns_first_match(self): + channel = CFChannel( + "PV:1", "admin", [CFProperty("time", "admin", "first"), CFProperty("time", "admin", "second")] + ) + assert channel.property_value(CFPropertyName.TIME) == "first" + def test_from_dict_roundtrip(self): ch = CFChannel( name="PV:1", diff --git a/server/tests/unit/cf/test_processor.py b/server/tests/unit/cf/test_processor.py index 2d29f362..9cdd7739 100644 --- a/server/tests/unit/cf/test_processor.py +++ b/server/tests/unit/cf/test_processor.py @@ -8,7 +8,7 @@ from recceiver.cf.model import CFChannel, CFProperty, CFPropertyName, PVStatus, RecordInfo from recceiver.cf.processor import CFProcessor from recceiver.recast import Transaction -from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_channel, make_ioc +from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_ioc from tests.unit.cf.mock_adapter import MockCFAdapter from tests.unit.conftest import make_adapter @@ -74,18 +74,90 @@ def test_keeps_ioc_when_channelcount_still_positive(self): assert proc.iocs[iocid].channelcount == 1 -class TestCleanService: - def test_marks_active_channels_inactive(self): - proc, adapter = make_processor_with_mock() - adapter.set_channels([make_channel("PV:1"), make_channel("PV:2")]) - proc.clean_service() - for name in ("PV:1", "PV:2"): - status = next(p for p in adapter._channels[name].properties if p.name == CFPropertyName.PV_STATUS.value) - assert status.value == PVStatus.INACTIVE.value - - def test_is_no_op_when_no_active_channels(self): - proc, _ = make_processor_with_mock() - proc.clean_service() +class TestStartupCleanGate: + def test_commit_waits_for_startup_clean(self, monkeypatch): + proc = make_processor() + proc.running = True + transaction = make_transaction(_HOST_A, 5064) + startup_clean = defer.Deferred() + proc._startup_clean = startup_clean + committed = [] + monkeypatch.setattr(proc, "_commit_with_lock", lambda tx: committed.append(tx)) + + result = proc.commit(transaction) + completed = [] + result.addCallback(completed.append) + assert committed == [] + assert completed == [] + + startup_clean.callback(None) + + assert committed == [transaction] + assert completed == [None] + + def test_cancel_while_waiting_skips_commit_without_cancelling_startup(self, monkeypatch): + proc = make_processor() + proc.running = True + startup_clean = defer.Deferred() + proc._startup_clean = startup_clean + committed = [] + monkeypatch.setattr(proc, "_commit_with_lock", lambda tx: committed.append(tx)) + transaction = make_transaction(_HOST_A, 5064) + + cancelled = proc.commit(transaction) + errors = [] + cancelled.addErrback(errors.append) + cancelled.cancel() + assert errors[0].check(defer.CancelledError) + assert not startup_clean.called + + following = proc.commit(transaction) + completed = [] + following.addCallback(completed.append) + startup_clean.callback(None) + + assert committed == [transaction] + assert completed == [None] + + def test_cancel_while_waiting_for_lock_removes_queued_commit(self, monkeypatch): + proc = make_processor() + proc.running = True + proc._startup_clean = defer.succeed(None) + proc.lock.acquire() + committed = [] + monkeypatch.setattr(proc, "_commit_with_lock", lambda tx: committed.append(tx)) + + result = proc.commit(make_transaction(_HOST_A, 5064)) + errors = [] + result.addErrback(errors.append) + result.cancel() + proc.lock.release() + + assert errors[0].check(defer.CancelledError) + assert committed == [] + assert not proc.lock.locked + + def test_cancel_active_commit_waits_for_worker_and_releases_lock(self, monkeypatch): + proc = make_processor() + proc.running = True + proc._startup_clean = defer.succeed(None) + worker = defer.Deferred() + monkeypatch.setattr("recceiver.cf.processor.deferToThread", lambda *args: worker) + + result = proc.commit(make_transaction(_HOST_A, 5064)) + errors = [] + result.addErrback(errors.append) + result.cancel() + + assert proc.cancelled + assert proc.lock.locked + assert not worker.called + assert errors == [] + + worker.callback(None) + + assert errors[0].check(defer.CancelledError) + assert not proc.lock.locked class TestUpdateChannelFinder: diff --git a/server/tests/unit/cf/test_processor_lifecycle.py b/server/tests/unit/cf/test_processor_lifecycle.py new file mode 100644 index 00000000..51618117 --- /dev/null +++ b/server/tests/unit/cf/test_processor_lifecycle.py @@ -0,0 +1,451 @@ +"""Deterministic service tests: run real workers at explicitly chosen boundaries.""" + +from collections import deque +from dataclasses import dataclass, field + +import pytest +from twisted.internet import defer +from twisted.internet.address import IPv4Address + +from recceiver.cf.model import PVStatus +from recceiver.cf.processor import CFProcessor +from recceiver.recast import Transaction +from tests.unit.cf.conftest import DEFAULT_RECCEIVER_ID, make_channel +from tests.unit.cf.mock_adapter import MockCFAdapter +from tests.unit.conftest import make_adapter + + +@dataclass +class ThreadJob: + function: object + args: tuple + kwargs: dict + result: defer.Deferred = field(default_factory=defer.Deferred) + execution: object = field(default=None, init=False) + delivered: bool = field(default=False, init=False) + + def execute(self): + """Run worker code, but hold its completion until the test delivers it.""" + assert self.execution is None + self.execution = defer.maybeDeferred(self.function, *self.args, **self.kwargs) + + def finish(self): + assert self.execution is not None + assert not self.delivered + self.delivered = True + self.execution.chainDeferred(self.result) + + def run(self): + self.execute() + self.finish() + + +class ControlledThreads: + """Queue thread jobs without executing them or starting a reactor.""" + + def __init__(self): + self.pending = deque() + self.submitted = [] + + def submit(self, function, *args, **kwargs): + job = ThreadJob(function, args, kwargs) + self.pending.append(job) + self.submitted.append(job) + return job.result + + def take_next(self, expected_name): + job = self.pending.popleft() + assert job.function.__name__ == expected_name + return job + + def run_next(self, expected_name): + job = self.take_next(expected_name) + job.run() + return job + + +class Outcome: + """Observe completion, including called-but-paused Deferreds; consume failures.""" + + def __init__(self, deferred): + self.values = [] + self.failures = [] + deferred.addCallbacks(self.values.append, self.failures.append) + + @property + def completed(self): + return bool(self.values or self.failures) + + def assert_success(self): + assert self.values == [None] + assert self.failures == [] + + def assert_failure(self, exception): + assert self.values == [] + assert len(self.failures) == 1 + assert self.failures[0].check(exception) + + +def transaction(name="LIVE:PV", ioc_name="IOC1"): + tx = Transaction(IPv4Address("TCP", "1.2.3.4", 5064), id=0) + tx.initial = True + tx.client_infos.update({"IOCNAME": ioc_name, "HOSTNAME": "ioc.example.com"}) + tx.records_to_add[1] = (name, "ai") + return tx + + +def status(adapter, name): + return next(p.value for p in adapter._channels[name].properties if p.name == "pvStatus") + + +@pytest.fixture +def lifecycle(monkeypatch): + threads = ControlledThreads() + monkeypatch.setattr("recceiver.cf.processor.deferToThread", threads.submit) + proc = CFProcessor( + "test", + make_adapter(values={"recceiverId": DEFAULT_RECCEIVER_ID, "statusInterval": 0}), + ) + adapter = MockCFAdapter() + proc.client = adapter + yield proc, adapter, threads + # Tests must account for every submitted worker and release every lock. + assert not threads.pending + assert all(job.delivered for job in threads.submitted) + assert not proc.lock.locked + if proc.running: + stopped = Outcome(proc.stopService()) + if threads.pending: + threads.run_next("clean_service") + stopped.assert_success() + assert not threads.pending + assert all(job.delivered for job in threads.submitted) + assert not proc.lock.locked + assert not proc._startup_commit_waiters + assert not proc._stopping + + +def test_start_commit_stop_uses_real_workers(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + threads.run_next("clean_on_start") + + committed = Outcome(proc.commit(transaction())) + assert not committed.completed + threads.run_next("_commit_with_thread") + committed.assert_success() + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + + stopped = Outcome(proc.stopService()) + assert not stopped.completed + threads.run_next("clean_service") + stopped.assert_success() + assert status(adapter, "LIVE:PV") == PVStatus.INACTIVE.value + + +def test_executor_propagates_worker_exceptions(): + threads = ControlledThreads() + + def failing_worker(): + raise ValueError("worker failed") + + outcome = Outcome(threads.submit(failing_worker)) + assert not outcome.completed + threads.run_next("failing_worker") + outcome.assert_failure(ValueError) + assert not threads.pending + + +def test_cleanup_precedes_live_commit_and_leaves_stale_channels_inactive(lifecycle): + proc, adapter, threads = lifecycle + adapter.set_channels([make_channel("LIVE:PV"), make_channel("STALE:PV")]) + proc.startService() + committed = Outcome(proc.commit(transaction())) + + assert [job.function.__name__ for job in threads.submitted] == ["clean_on_start"] + assert not committed.completed + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + assert status(adapter, "STALE:PV") == PVStatus.ACTIVE.value + + threads.run_next("clean_on_start") + assert not committed.completed + assert status(adapter, "LIVE:PV") == PVStatus.INACTIVE.value + assert status(adapter, "STALE:PV") == PVStatus.INACTIVE.value + + threads.run_next("_commit_with_thread") + committed.assert_success() + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + assert status(adapter, "STALE:PV") == PVStatus.INACTIVE.value + + +def test_startup_waiters_commit_serially(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + first = Outcome(proc.commit(transaction("FIRST:PV", "FIRST"))) + second = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + assert len(threads.submitted) == 1 + + threads.run_next("clean_on_start") + assert len(threads.pending) == 1 + assert not first.completed + assert not second.completed + + threads.run_next("_commit_with_thread") + first.assert_success() + assert not second.completed + assert len(threads.pending) == 1 + assert "SECOND:PV" not in adapter._channels + + threads.run_next("_commit_with_thread") + second.assert_success() + assert status(adapter, "FIRST:PV") == PVStatus.ACTIVE.value + assert status(adapter, "SECOND:PV") == PVStatus.ACTIVE.value + + +@pytest.mark.parametrize("clean_on_start", [True, False]) +def test_commit_after_startup_does_not_repeat_cleanup(lifecycle, clean_on_start): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_start = clean_on_start + adapter.set_channels([make_channel("STALE:PV")]) + proc.startService() + threads.run_next("clean_on_start") + expected = PVStatus.INACTIVE if clean_on_start else PVStatus.ACTIVE + assert status(adapter, "STALE:PV") == expected.value + + committed = Outcome(proc.commit(transaction())) + threads.run_next("_commit_with_thread") + committed.assert_success() + assert [job.function.__name__ for job in threads.submitted] == ["clean_on_start", "_commit_with_thread"] + assert status(adapter, "STALE:PV") == expected.value + + +def test_cancelled_startup_waiter_does_not_affect_other_commits(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + cancelled = proc.commit(transaction("CANCELLED:PV", "CANCELLED")) + outcome = Outcome(cancelled) + following = Outcome(proc.commit(transaction())) + cancelled.cancel() + outcome.assert_failure(defer.CancelledError) + assert len(threads.submitted) == 1 + + threads.run_next("clean_on_start") + threads.run_next("_commit_with_thread") + following.assert_success() + assert "CANCELLED:PV" not in adapter._channels + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + + +def test_startup_failure_reaches_waiting_and_later_commits(lifecycle, monkeypatch): + proc, adapter, threads = lifecycle + + def fail_find(_recceiver_id): + raise ValueError("unexpected cleanup failure") + + monkeypatch.setattr(adapter, "find_active_for_recceiver", fail_find) + proc.startService() + first = Outcome(proc.commit(transaction())) + second = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + threads.run_next("clean_on_start") + first.assert_failure(ValueError) + second.assert_failure(ValueError) + later = Outcome(proc.commit(transaction("LATER:PV", "LATER"))) + later.assert_failure(ValueError) + assert len(threads.submitted) == 1 + assert adapter._channels == {} + # Shutdown consumes the shared gate, after all callers observed its failure. + # Avoid running shutdown's worker with the patched, failing adapter method. + proc.cf_config.clean_on_stop = False + + +@pytest.mark.parametrize("clean_on_stop", [True, False]) +def test_shutdown_waits_for_active_commit_before_stop_cleanup(lifecycle, clean_on_stop): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_stop = clean_on_stop + proc.startService() + threads.run_next("clean_on_start") + committed = Outcome(proc.commit(transaction())) + worker = threads.take_next("_commit_with_thread") + worker.execute() + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + assert not committed.completed + + stopped = Outcome(proc.stopService()) + assert not stopped.completed + assert not threads.pending + assert proc.lock.locked + + worker.finish() + committed.assert_success() + if clean_on_stop: + assert not stopped.completed + threads.run_next("clean_service") + stopped.assert_success() + expected = PVStatus.INACTIVE if clean_on_stop else PVStatus.ACTIVE + assert status(adapter, "LIVE:PV") == expected.value + + +@pytest.mark.parametrize("clean_on_stop", [True, False]) +def test_shutdown_during_startup_cancels_waiters_and_waits_for_cleanup(lifecycle, clean_on_stop): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_stop = clean_on_stop + adapter.set_channels([make_channel("STALE:PV")]) + proc.startService() + first = Outcome(proc.commit(transaction())) + second = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + + shutdown = proc.stopService() + assert proc.stopService() is shutdown + stopped = Outcome(shutdown) + first.assert_failure(defer.CancelledError) + second.assert_failure(defer.CancelledError) + assert not stopped.completed + assert len(threads.submitted) == 1 + with pytest.raises(RuntimeError, match="stopping"): + proc.startService() + + threads.run_next("clean_on_start") + if clean_on_stop: + assert not stopped.completed + threads.run_next("clean_service") + stopped.assert_success() + assert status(adapter, "STALE:PV") == PVStatus.INACTIVE.value + assert "LIVE:PV" not in adapter._channels + assert "SECOND:PV" not in adapter._channels + assert all(job.function.__name__ != "_commit_with_thread" for job in threads.submitted) + + +def test_restart_uses_fresh_cleanup_gate_without_old_transactions(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + old_gate = proc._startup_clean + old_commit = Outcome(proc.commit(transaction("OLD:PV", "OLD"))) + stopped = Outcome(proc.stopService()) + threads.run_next("clean_on_start") + threads.run_next("clean_service") + stopped.assert_success() + old_commit.assert_failure(defer.CancelledError) + + proc.startService() + assert proc._startup_clean is not old_gate + restarted = Outcome(proc.commit(transaction())) + assert not restarted.completed + assert len(threads.pending) == 1 + threads.run_next("clean_on_start") + assert not restarted.completed + threads.run_next("_commit_with_thread") + restarted.assert_success() + assert "OLD:PV" not in adapter._channels + assert status(adapter, "LIVE:PV") == PVStatus.ACTIVE.value + + +def test_initialization_failure_stops_service_without_launching_workers(lifecycle, monkeypatch): + proc, adapter, threads = lifecycle + + def fail_properties(): + raise ValueError("invalid property setup") + + monkeypatch.setattr(adapter, "get_property_names", fail_properties) + with pytest.raises(ValueError, match="invalid property setup"): + proc.startService() + assert not proc.running + assert not proc.lock.locked + assert threads.submitted == [] + Outcome(proc.commit(transaction())).assert_failure(defer.CancelledError) + Outcome(proc.stopService()).assert_success() + + +def test_commit_before_start_and_after_stop_is_cancelled(lifecycle): + proc, adapter, threads = lifecycle + Outcome(proc.commit(transaction())).assert_failure(defer.CancelledError) + proc.startService() + threads.run_next("clean_on_start") + stopped = Outcome(proc.stopService()) + threads.run_next("clean_service") + stopped.assert_success() + Outcome(proc.commit(transaction())).assert_failure(defer.CancelledError) + assert len(threads.submitted) == 2 + assert adapter._channels == {} + + +def test_active_cancellation_waits_for_completion_before_next_commit(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + threads.run_next("clean_on_start") + cancelled = proc.commit(transaction("FIRST:PV", "FIRST")) + first = Outcome(cancelled) + worker = threads.take_next("_commit_with_thread") + worker.execute() + following = Outcome(proc.commit(transaction("SECOND:PV", "SECOND"))) + + cancelled.cancel() + assert proc.cancelled + assert not first.completed + assert not following.completed + assert not threads.pending + assert proc.lock.locked + + worker.finish() + first.assert_failure(defer.CancelledError) + assert not following.completed + threads.run_next("_commit_with_thread") + following.assert_success() + assert not proc.lock.locked + assert status(adapter, "SECOND:PV") == PVStatus.ACTIVE.value + + +def test_shutdown_waits_for_startup_failure_without_stranding_waiters(lifecycle, monkeypatch): + proc, adapter, threads = lifecycle + proc.cf_config.clean_on_stop = False + + def fail_find(_recceiver_id): + raise ValueError("unexpected cleanup failure") + + monkeypatch.setattr(adapter, "find_active_for_recceiver", fail_find) + proc.startService() + committed = Outcome(proc.commit(transaction())) + stopped = Outcome(proc.stopService()) + committed.assert_failure(defer.CancelledError) + assert not stopped.completed + threads.run_next("clean_on_start") + stopped.assert_success() + assert not proc._stopping + + +def test_reentrant_stop_from_cancelled_waiter_returns_same_shutdown(lifecycle): + proc, _adapter, threads = lifecycle + proc.startService() + reentrant_shutdowns = [] + committed = proc.commit(transaction()) + + def cancelled(failure): + assert failure.check(defer.CancelledError) + reentrant_shutdowns.append(proc.stopService()) + + committed.addErrback(cancelled) + shutdown = proc.stopService() + assert reentrant_shutdowns == [shutdown] + stopped = Outcome(shutdown) + threads.run_next("clean_on_start") + threads.run_next("clean_service") + stopped.assert_success() + + +def test_shutdown_rejects_lock_queued_commit_without_starting_worker(lifecycle): + proc, adapter, threads = lifecycle + proc.startService() + threads.run_next("clean_on_start") + first = Outcome(proc.commit(transaction("FIRST:PV", "FIRST"))) + worker = threads.take_next("_commit_with_thread") + worker.execute() + queued = Outcome(proc.commit(transaction("QUEUED:PV", "QUEUED"))) + stopped = Outcome(proc.stopService()) + + worker.finish() + first.assert_success() + queued.assert_failure(defer.CancelledError) + assert [job.function.__name__ for job in threads.pending] == ["clean_service"] + assert "QUEUED:PV" not in adapter._channels + threads.run_next("clean_service") + stopped.assert_success() + assert status(adapter, "FIRST:PV") == PVStatus.INACTIVE.value diff --git a/server/tests/unit/test_integration_live_restart.py b/server/tests/unit/test_integration_live_restart.py new file mode 100644 index 00000000..429a6e03 --- /dev/null +++ b/server/tests/unit/test_integration_live_restart.py @@ -0,0 +1,37 @@ +from unittest.mock import Mock + +import pytest + +from recceiver.cf.adapter import PyCFClientAdapter +from recceiver.cf.model import CFChannel, CFProperty +from tests.integration import test_single_ioc + + +def test_live_restart_seeds_typed_properties_without_the_adapter_helper_migration(monkeypatch): + receiver = CFProperty("recceiverID", "admin", "receiver") + baseline = [CFChannel(name, "admin", [receiver]) for name in ["PV:1", "PV:1:alias"]] + adapter = Mock(spec=PyCFClientAdapter) + adapter.find_by_names.return_value = baseline + adapter.find_active_for_recceiver.return_value = baseline + compose = Mock() + compose.get_container.return_value.ID = "container-id" + monkeypatch.setattr(test_single_ioc, "wait_for_sync", lambda adapter, check: check(adapter)) + + class SeedValidated(Exception): + pass + + def validate_seed(channels): + assert len(channels) == 1 + assert all(isinstance(prop, CFProperty) for prop in channels[0].properties) + assert channels[0].as_dict()["properties"] == [ + receiver.as_dict(), + {"name": "pvStatus", "owner": "admin", "value": "Active"}, + ] + raise SeedValidated + + adapter.set_channels.side_effect = validate_seed + with pytest.raises(SeedValidated): + test_single_ioc.TestLiveIOCRestartRecceiver().test_restart_restores_live_channels_and_cleans_stale_channels( + compose, adapter + ) + adapter.set_channels.assert_called_once()