Merge pull request #153 from ChannelFinder/fix-recceiver-issues

Fix recCeiver issues

Author: anderslindho

server/recceiver/cfstore.py

@@ -6,7 +6,7 @@
 import socket
 import time
 from collections import defaultdict
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, fields
 from typing import Any, Callable, Dict, List, Optional, Set, Tuple
 
 from channelfinder import ChannelFinderClient
@@ -89,6 +89,15 @@ def loads(cls, conf: ConfigAdapter) -> "CFConfig":
             push_always_retry=conf.getboolean("pushAlwaysRetry", True),
         )
 
+    def __repr__(self) -> str:
+        parts = []
+        for f in fields(self):
+            value = getattr(self, f.name)
+            if f.name == "cf_password":
+                value = "***" if value else None
+            parts.append(f"{f.name}={value!r}")
+        return f"CFConfig({', '.join(parts)})"
+
 
 @dataclass
 class CFProperty:
@@ -555,26 +564,41 @@ def _commit_with_thread(self, transaction: CommitTransaction):
         Args:
             transaction: The transaction to commit.
         """
+        host = transaction.source_address.host
+        port = transaction.source_address.port
+
         if not self.running:
-            host = transaction.source_address.host
-            port = transaction.source_address.port
             raise defer.CancelledError(f"CF Processor is not running (transaction: {host}:{port})")
 
         _log.info("CF_COMMIT: %s", transaction)
         _log.debug("CF_COMMIT: transaction: %s", repr(transaction))
 
+        ioc_name = transaction.client_infos.get("IOCNAME")
+        if not ioc_name:
+            ioc_name = str(port)
+            _log.debug("IOC at %s:%d did not send IOCNAME; using port as iocName", host, port)
+
+        owner = (
+            transaction.client_infos.get(self.cf_config.env_owner_variable)
+            or transaction.client_infos.get("CF_USERNAME")
+            or self.cf_config.username
+        )
+        if owner == self.cf_config.username:
+            _log.debug(
+                "IOC at %s:%d did not send %s or CF_USERNAME; using service account as owner",
+                host,
+                port,
+                self.cf_config.env_owner_variable,
+            )
+
         ioc_info = IocInfo(
-            host=transaction.source_address.host,
-            hostname=transaction.client_infos.get("HOSTNAME") or transaction.source_address.host,
-            ioc_name=transaction.client_infos.get("IOCNAME") or str(transaction.source_address.port),
-            ioc_IP=transaction.source_address.host,
-            owner=(
-                transaction.client_infos.get(self.cf_config.env_owner_variable)
-                or transaction.client_infos.get("CF_USERNAME")
-                or self.cf_config.username
-            ),
+            host=host,
+            hostname=transaction.client_infos.get("HOSTNAME") or host,
+            ioc_name=ioc_name,
+            ioc_IP=host,
+            owner=owner,
             time=self.current_time(self.cf_config.timezone),
-            port=transaction.source_address.port,
+            port=port,
         )
 
         record_infos = self.transaction_to_record_infos(ioc_info, transaction)
@@ -596,13 +620,16 @@ def remove_channel(self, recordName: str, iocid: str) -> None:
             iocid: The IOC ID of the record to remove from.
         """
         self.channel_ioc_ids[recordName].remove(iocid)
-        if iocid in self.iocs:
-            self.iocs[iocid].channelcount -= 1
-        if self.iocs[iocid].channelcount == 0:
+        if iocid not in self.iocs:
+            if len(self.channel_ioc_ids[recordName]) == 0:
+                del self.channel_ioc_ids[recordName]
+            return
+        self.iocs[iocid].channelcount -= 1
+        if self.iocs[iocid].channelcount <= 0:
+            if self.iocs[iocid].channelcount < 0:
+                _log.error("Channel count negative: %s", iocid)
             self.iocs.pop(iocid)
-        elif self.iocs[iocid].channelcount < 0:
-            _log.error("Channel count negative: %s", iocid)
-        if len(self.channel_ioc_ids[recordName]) <= 0:  # case: channel has no more iocs
+        if len(self.channel_ioc_ids[recordName]) == 0:
             del self.channel_ioc_ids[recordName]
 
     def clean_service(self) -> None:
@@ -1095,9 +1122,9 @@ def _update_channelfinder(
 
     if iocid not in iocs:
         _log.warning(
-            "IOC %s did not send an initial transaction to join IOC list: %s",
+            "IOC %s did not send an initial transaction to join IOC list (%d IOCs known)",
             ioc_info,
-            [(ioc.ioc_name, ioc.ioc_id) for ioc in iocs.values()],
+            len(iocs),
         )
 
     if ioc_info.hostname is None or ioc_info.ioc_name is None:

server/recceiver/recast.py

@@ -255,7 +255,7 @@ def __repr__(self):
 
 class CollectionSession(object):
     timeout = 5.0
-    trlimit = 0
+    trlimit = 5000
 
     def __init__(self, proto, endpoint):
         from twisted.internet import reactor
@@ -272,15 +272,11 @@ def __init__(self, proto, endpoint):
     def close(self):
         _log.info("Close session from {ep}".format(ep=self.ep))
 
-        def suppressCancelled(err):
-            if not err.check(defer.CancelledError):
-                return err
-            _log.debug("Suppress the expected CancelledError")
-
-        self.C.addErrback(suppressCancelled).cancel()
-
-        # Clear the current transaction and
-        # commit an empty one for disconnect.
+        # Do not cancel self.C here. Any data commit that is still queued
+        # behind the global lock must be allowed to complete so that channels
+        # are registered as active in CF before the disconnect is processed.
+        # The disconnect transaction is chained after self.C and will execute
+        # once all preceding commits have finished.
         self.transaction = Transaction(self.ep, id(self))
         self.transaction.connected = False
         self.dirty = True
@@ -315,10 +311,12 @@ def abort(err):
     # between transactions. Only flush after Add or Del or Done message received.
     def flushSafely(self):
         if self.T and self.T <= time.time():
+            _log.debug("flushSafely: timeout elapsed for %s", self.ep)
             self.flush()
         elif self.trlimit and self.trlimit <= (
             len(self.transaction.records_to_add) + len(self.transaction.records_to_delete)
         ):
+            _log.debug("flushSafely: trlimit %d reached for %s", self.trlimit, self.ep)
             self.flush()
 
     def markDirty(self):

server/tests/test_cfstore.py

@@ -1,6 +1,6 @@
 from configparser import ConfigParser
 
-from recceiver.cfstore import CFConfig
+from recceiver.cfstore import CFConfig, CFProcessor, IocInfo
 from recceiver.processors import ConfigAdapter
 
 
@@ -50,3 +50,60 @@ def test_alias_enabled_from_config(self):
         adapter = make_adapter(values={"alias": "true"})
         config = CFConfig.loads(adapter)
         assert config.alias_enabled is True
+
+
+def make_processor() -> CFProcessor:
+    return CFProcessor("test", make_adapter())
+
+
+def make_ioc(channelcount: int = 1) -> IocInfo:
+    return IocInfo(
+        host="1.2.3.4",
+        hostname="ioc1.example.com",
+        ioc_name="IOC1",
+        ioc_IP="1.2.3.4",
+        owner="engineer",
+        time="2026-01-01T00:00:00",
+        port=5064,
+        channelcount=channelcount,
+    )
+
+
+class TestRemoveChannel:
+    def test_missing_iocid_does_not_raise(self):
+        proc = make_processor()
+        iocid = "1.2.3.4:5064"
+        proc.channel_ioc_ids["CHAN:1"].append(iocid)
+        # iocid deliberately absent from proc.iocs
+        proc.remove_channel("CHAN:1", iocid)
+        assert "CHAN:1" not in proc.channel_ioc_ids
+
+    def test_missing_iocid_preserves_channel_when_other_iocs_remain(self):
+        proc = make_processor()
+        iocid = "1.2.3.4:5064"
+        proc.channel_ioc_ids["CHAN:1"].append(iocid)
+        proc.channel_ioc_ids["CHAN:1"].append("9.9.9.9:5064")
+        proc.remove_channel("CHAN:1", iocid)
+        assert "CHAN:1" in proc.channel_ioc_ids
+        assert "9.9.9.9:5064" in proc.channel_ioc_ids["CHAN:1"]
+
+    def test_removes_ioc_when_channelcount_reaches_zero(self):
+        proc = make_processor()
+        ioc = make_ioc(channelcount=1)
+        iocid = ioc.ioc_id
+        proc.iocs[iocid] = ioc
+        proc.channel_ioc_ids["CHAN:1"].append(iocid)
+        proc.remove_channel("CHAN:1", iocid)
+        assert iocid not in proc.iocs
+        assert "CHAN:1" not in proc.channel_ioc_ids
+
+    def test_keeps_ioc_when_channelcount_still_positive(self):
+        proc = make_processor()
+        ioc = make_ioc(channelcount=2)
+        iocid = ioc.ioc_id
+        proc.iocs[iocid] = ioc
+        proc.channel_ioc_ids["CHAN:1"].append(iocid)
+        proc.channel_ioc_ids["CHAN:2"].append(iocid)
+        proc.remove_channel("CHAN:1", iocid)
+        assert iocid in proc.iocs
+        assert proc.iocs[iocid].channelcount == 1

server/tests/test_recast.py

@@ -0,0 +1,56 @@
+from unittest.mock import MagicMock
+
+from twisted.internet import defer
+
+from recceiver.recast import CollectionSession
+
+
+def make_session() -> CollectionSession:
+    proto = MagicMock()
+    proto.transport = MagicMock()
+    session = CollectionSession(proto, "test:1234")
+    session.factory = MagicMock()
+    return session
+
+
+class TestFlushSafely:
+    def test_trlimit_triggers_flush_when_reached(self):
+        session = make_session()
+        session.trlimit = 3
+        session.flush = MagicMock()
+        session.markDirty()
+        for i in range(3):
+            session.transaction.records_to_add[i] = (f"REC:{i}", "ai")
+        session.flushSafely()
+        session.flush.assert_called_once()
+
+    def test_trlimit_does_not_flush_below_limit(self):
+        session = make_session()
+        session.trlimit = 3
+        session.flush = MagicMock()
+        session.markDirty()
+        for i in range(2):
+            session.transaction.records_to_add[i] = (f"REC:{i}", "ai")
+        session.flushSafely()
+        session.flush.assert_not_called()
+
+    def test_trlimit_zero_never_triggers_flush(self):
+        session = make_session()
+        session.trlimit = 0
+        session.flush = MagicMock()
+        session.markDirty()
+        for i in range(10000):
+            session.transaction.records_to_add[i] = (f"REC:{i}", "ai")
+        session.flushSafely()
+        session.flush.assert_not_called()
+
+
+class TestCollectionSessionClose:
+    def test_close_does_not_cancel_pending_commit(self):
+        session = make_session()
+        pending = defer.Deferred()
+        cancelled_errors = []
+        pending.addErrback(lambda f: cancelled_errors.append(f.type) or f)
+        session.C = pending
+        session.close()
+        assert cancelled_errors == [], "close() must not cancel a queued data commit"