diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f7732b..2b1abe7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,28 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.14.21] - 2026-04-07 + +### Added +- **Redis Cacher production-ready** — start()/stop() lifecycle, broker integration, + 16 integration tests (get/set/delete/clean/TTL/keys/broker) +- **seq/instanceID heartbeat checks** — detects remote node restart and service changes + via heartbeat (Node.js heartbeatReceived parity) + +### Fixed +- **instanceID persistence** — process_node_info now saves instanceID from INFO packets + (was causing infinite re-discovery loop on every heartbeat) +- **payload dict guard** — _handle_heartbeat validates payload is dict before access + (prevents crash on malformed packets) +- **seq type coercion** — int comparison for cross-language safety +- **instanceID=None guard** — skip comparison when node has no instanceID yet +- **_discover_pending cleanup** — stale entries evicted in check_remote_nodes + (prevents unbounded memory growth) +- **Redis cacher: logger fallback** — set in __init__ (was crashing if connect() before init()) +- **Redis cacher: prefix dedup** — removed duplicate namespace logic (BaseCacher handles it) +- **Redis cacher: TTL validation** — negative/zero TTL warns and stores without expiry +- **Redis cacher: ping loop** — pings first then sleeps (detects immediate connection drop) + ## [0.14.20] - 2026-04-06 ### Added diff --git a/moleculerpy/__init__.py b/moleculerpy/__init__.py index d5b5552..988acf2 100644 --- a/moleculerpy/__init__.py +++ b/moleculerpy/__init__.py @@ -54,7 +54,7 @@ try: __version__ = version("moleculerpy") except PackageNotFoundError: - __version__ = "0.14.20" + __version__ = "0.14.21" __all__ = [ # noqa: RUF022 # Core diff --git a/moleculerpy/cacher/redis.py b/moleculerpy/cacher/redis.py index 28e9cd4..44d2ed6 100644 --- a/moleculerpy/cacher/redis.py +++ b/moleculerpy/cacher/redis.py @@ -124,9 +124,12 @@ def __init__( # Default TTL in seconds self.default_ttl: int | None = self.opts.get("ttl") - # Serializer (initialized in init()) + # Serializer self.serializer = JSONSerializer() + # Logger fallback (overridden in init() with broker logger) + self.logger = logger + # Ping interval (optional periodic health check) self.ping_interval = self.opts.get("ping_interval") self._ping_task: asyncio.Task[None] | None = None @@ -139,16 +142,16 @@ def init(self, broker: ServiceBroker) -> None: """ super().init(broker) + # Redis is NOT connected until start() → connect() is called. + # BaseCacher.init() sets connected=True (fine for MemoryCacher), but + # for network cachers we override back to False until actual connection. + self.connected = False + # Create logger (use broker's logger factory) if hasattr(broker, "_create_logger"): self.logger = broker._create_logger("REDIS-CACHER") - else: - self.logger = logger - - # Add namespace to prefix if configured - if broker.namespace and broker.namespace != "": - self.prefix = f"MOL-{broker.namespace}-" + # Namespace prefix handled by BaseCacher.init() — no duplicate logic here. self.logger.info(f"Initializing Redis cacher with prefix '{self.prefix}'") async def connect(self) -> None: @@ -228,24 +231,25 @@ async def disconnect(self) -> None: self.connected = False async def _ping_loop(self) -> None: - """Periodic ping task for connection monitoring.""" + """Periodic ping task for connection monitoring. + + Pings first, then sleeps — detects immediate connection drop. + """ try: + interval = self.ping_interval if isinstance(self.ping_interval, (int, float)) else 10.0 while self.connected and self.client: - interval = ( - self.ping_interval if isinstance(self.ping_interval, (int, float)) else 0.0 - ) - await asyncio.sleep(interval) try: await self._await_maybe(self.client.ping()) except Exception as e: - self.logger.warning(f"Ping failed: {e}") + self.logger.error(f"Redis ping failed — connection may be broken: {e}") self.connected = False - # Broadcast error event if self.broker: await self.broker.broadcast_local( "$cacher.error", {"error": str(e), "module": "cacher", "type": "PING_FAILED"}, ) + break + await asyncio.sleep(interval) except asyncio.CancelledError: pass @@ -329,21 +333,25 @@ async def set( prefixed_key = self._get_prefixed_key(key) - # Use provided TTL or default - if ttl is None: - ttl = self.default_ttl + # Use provided TTL or default; validate + effective_ttl = ttl if ttl is not None else self.default_ttl + if effective_ttl is not None and effective_ttl <= 0: + self.logger.warning( + f"Invalid TTL {effective_ttl} for key {key}, storing without expiry" + ) + effective_ttl = None try: # Serialize data serialized = self.serializer.serialize(data) # SET with optional TTL (atomic operation) - if ttl: - await self.client.set(prefixed_key, serialized, ex=ttl) + if effective_ttl is not None: + await self.client.set(prefixed_key, serialized, ex=effective_ttl) else: await self.client.set(prefixed_key, serialized) - self.logger.debug(f"Cache SET: {key} (ttl={ttl}s)") + self.logger.debug(f"Cache SET: {key} (ttl={effective_ttl}s)") except Exception as e: self.logger.error(f"Redis SET error for {key}: {e}") @@ -517,9 +525,22 @@ async def get_cache_keys(self) -> list[dict[str, str]]: self.logger.error(f"Redis GET_CACHE_KEYS error: {e}") return [] - # Lock methods (TODO: Implement Redlock for distributed locking) - # For now, inherit from BaseCacher which provides in-memory lock fallback + async def start(self) -> None: + """Start cacher — connect to Redis. + + Called by broker during broker.start(). Matches Node.js init() behavior + where connection is established during broker lifecycle. + """ + await self.connect() + + async def stop(self) -> None: + """Stop cacher — disconnect from Redis. + + Called by broker during broker.stop(). + """ + await super().stop() # Clears in-memory locks + await self.disconnect() async def close(self) -> None: - """Close Redis connection gracefully.""" + """Close Redis connection gracefully (alias for disconnect).""" await self.disconnect() diff --git a/moleculerpy/discoverer.py b/moleculerpy/discoverer.py index f9b75f5..04148a7 100644 --- a/moleculerpy/discoverer.py +++ b/moleculerpy/discoverer.py @@ -203,6 +203,13 @@ def check_remote_nodes(self) -> None: ) node_catalog.disconnect_node(node_id, unexpected=True) + # Evict stale entries from _discover_pending (prevents unbounded growth) + expired = [ + k for k, ts in self._discover_pending.items() if now - ts > self._DISCOVER_COOLDOWN + ] + for k in expired: + self._discover_pending.pop(k, None) + def check_offline_nodes(self) -> None: """Check offline nodes. Remove which are older than 10 minutes. diff --git a/moleculerpy/node.py b/moleculerpy/node.py index ea7079d..8917883 100644 --- a/moleculerpy/node.py +++ b/moleculerpy/node.py @@ -415,6 +415,7 @@ def process_node_info(self, node_id: str, payload: dict[str, Any]) -> None: node.client = payload.get("client") node.metadata = payload.get("metadata", {}) node.seq = payload.get("seq", 0) + node.instanceID = payload.get("instanceID", node.instanceID) node.port = payload.get("port", 0) # Phase 5.1: Update extended metrics diff --git a/moleculerpy/transit.py b/moleculerpy/transit.py index f475034..f8af5d3 100644 --- a/moleculerpy/transit.py +++ b/moleculerpy/transit.py @@ -519,11 +519,16 @@ async def beat(self) -> None: local_node.hostname = static["hostname"] local_node.ipList = static["ip_list"] - heartbeat_data = { + heartbeat_data: dict[str, Any] = { "cpu": metrics["cpu"], "cpuSeq": metrics["cpuSeq"], "memory": metrics["memory"], # Python extension } + # Include seq and instanceID so remote nodes can detect service changes + # and restarts via heartbeat (Node.js checks these in heartbeatReceived). + if local_node: + heartbeat_data["seq"] = local_node.seq + heartbeat_data["instanceID"] = local_node.instanceID await self.publish(Packet(Topic.HEARTBEAT, None, heartbeat_data)) async def send_node_info(self) -> None: @@ -577,6 +582,8 @@ async def _handle_heartbeat(self, packet: Packet) -> None: """ if not packet.sender or packet.sender == self.node_id: return # Ignore own heartbeats (Node.js: sender === this.broker.nodeID) + if not isinstance(packet.payload, dict): + return # Malformed heartbeat node = self.node_catalog.get_node(packet.sender) if node is None: @@ -587,6 +594,28 @@ async def _handle_heartbeat(self, packet: Packet) -> None: await self._request_discovery(packet.sender, "offline") return + # Check seq mismatch — services changed on remote node (Node.js parity) + payload_seq = packet.payload.get("seq") + if payload_seq is not None and int(getattr(node, "seq", 0)) != int(payload_seq): + self.logger.debug( + f"Service seq changed on '{packet.sender}' " + f"({getattr(node, 'seq', 0)} → {payload_seq}), requesting INFO" + ) + await self._request_discovery(packet.sender, "seq-changed") + return + + # Check instanceID mismatch — node restarted (Node.js parity) + # Skip if node has no instanceID yet (first registration, no INFO received) + payload_iid = packet.payload.get("instanceID") + node_iid = getattr(node, "instanceID", None) + if payload_iid is not None and node_iid and not str(node_iid).startswith(str(payload_iid)): + self.logger.debug( + f"instanceID changed on '{packet.sender}' " + f"({node_iid} → {payload_iid}), requesting INFO" + ) + await self._request_discovery(packet.sender, "instance-restarted") + return + # Update Moleculer.js compatible metrics node.cpu = packet.payload.get("cpu", 0) node.cpuSeq = packet.payload.get("cpuSeq", 0) diff --git a/pyproject.toml b/pyproject.toml index eea822d..ba07571 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "moleculerpy" -version = "0.14.20" +version = "0.14.21" description = "Fast, modern microservices framework for Python - Port of Moleculer.js" authors = [ { name = "Eli Rum", email = "explosivebit@gmail.com" } diff --git a/tests/unit/cacher_redis_test.py b/tests/unit/cacher_redis_test.py new file mode 100644 index 0000000..65b31f3 --- /dev/null +++ b/tests/unit/cacher_redis_test.py @@ -0,0 +1,219 @@ +"""Unit + integration tests for Redis Cacher. + +Tests both mock-based unit tests and real Redis integration tests. +Integration tests require Redis on localhost:6381 (skip if unavailable). +""" + +import asyncio +import socket +from unittest.mock import AsyncMock, MagicMock + +import pytest +import pytest_asyncio + +from moleculerpy.cacher.redis import RedisCacher + + +def _redis_available() -> bool: + """Check if Redis is available for integration tests.""" + try: + with socket.create_connection(("localhost", 6381), timeout=1.0): + return True + except OSError: + return False + + +REDIS_URL = "redis://localhost:6381/15" # Use DB 15 for test isolation +skip_no_redis = pytest.mark.skipif(not _redis_available(), reason="Redis not available on 6381") + + +# --------------------------------------------------------------------------- +# Unit tests (no Redis needed) +# --------------------------------------------------------------------------- + + +class TestRedisCacherInit: + def test_from_url(self): + cacher = RedisCacher(REDIS_URL) + assert cacher.redis_url == REDIS_URL + assert cacher.prefix == "MOL-" + + def test_from_dict(self): + cacher = RedisCacher({"redis": {"host": "localhost", "port": 6379}, "prefix": "TEST-"}) + assert cacher.prefix == "TEST-" + assert cacher.redis_url is None + + def test_with_ttl(self): + cacher = RedisCacher({"redis": REDIS_URL, "ttl": 60}) + assert cacher.default_ttl == 60 + + def test_init_with_broker(self): + cacher = RedisCacher(REDIS_URL) + broker = MagicMock() + broker.namespace = "myapp" + broker.settings = MagicMock() + broker.settings.namespace = "myapp" + broker._create_logger = MagicMock(return_value=MagicMock()) + cacher.init(broker) + assert cacher.prefix == "MOL-myapp-" + + +# --------------------------------------------------------------------------- +# Integration tests (real Redis) +# --------------------------------------------------------------------------- + + +@skip_no_redis +class TestRedisCacherIntegration: + @pytest_asyncio.fixture + async def cacher(self): + c = RedisCacher({"redis": REDIS_URL, "ttl": 10, "prefix": "TEST-"}) + broker = MagicMock() + broker.namespace = None + broker._create_logger = MagicMock(return_value=MagicMock()) + broker.settings = MagicMock() + broker.settings.namespace = None + c.init(broker) + await c.start() + # Clean test prefix before test + await c.clean("*") + yield c + await c.clean("*") + await c.stop() + + @pytest.mark.asyncio + async def test_start_stop_lifecycle(self): + c = RedisCacher(REDIS_URL) + broker = MagicMock() + broker.namespace = None + broker._create_logger = MagicMock(return_value=MagicMock()) + broker.settings = MagicMock() + broker.settings.namespace = None + c.init(broker) + + assert not c.connected + await c.start() + assert c.connected + await c.stop() + assert not c.connected + + @pytest.mark.asyncio + async def test_get_set(self, cacher): + await cacher.set("user:1", {"name": "Alice", "age": 30}) + result = await cacher.get("user:1") + assert result == {"name": "Alice", "age": 30} + + @pytest.mark.asyncio + async def test_get_miss(self, cacher): + result = await cacher.get("nonexistent") + assert result is None + + @pytest.mark.asyncio + async def test_set_with_ttl(self, cacher): + await cacher.set("expiring", "value", ttl=1) + assert await cacher.get("expiring") == "value" + await asyncio.sleep(1.5) + assert await cacher.get("expiring") is None + + @pytest.mark.asyncio + async def test_delete_single(self, cacher): + await cacher.set("to-delete", "data") + await cacher.delete("to-delete") + assert await cacher.get("to-delete") is None + + @pytest.mark.asyncio + async def test_delete_multiple(self, cacher): + await cacher.set("a", 1) + await cacher.set("b", 2) + await cacher.delete(["a", "b"]) + assert await cacher.get("a") is None + assert await cacher.get("b") is None + + @pytest.mark.asyncio + async def test_clean_pattern(self, cacher): + await cacher.set("user:1", "alice") + await cacher.set("user:2", "bob") + await cacher.set("order:1", "pizza") + await cacher.clean("user:*") + assert await cacher.get("user:1") is None + assert await cacher.get("user:2") is None + assert await cacher.get("order:1") == "pizza" + + @pytest.mark.asyncio + async def test_get_with_ttl(self, cacher): + await cacher.set("ttl-test", "data", ttl=30) + data, ttl = await cacher.get_with_ttl("ttl-test") + assert data == "data" + assert ttl is not None + assert 0 < ttl <= 30 + + @pytest.mark.asyncio + async def test_get_cache_keys(self, cacher): + await cacher.set("key1", "v1") + await cacher.set("key2", "v2") + keys = await cacher.get_cache_keys() + key_names = [k["key"] for k in keys] + assert "key1" in key_names + assert "key2" in key_names + + @pytest.mark.asyncio + async def test_complex_values(self, cacher): + """Test caching complex nested structures.""" + data = { + "users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}], + "meta": {"total": 2, "page": 1}, + "flags": [True, False, None], + } + await cacher.set("complex", data) + result = await cacher.get("complex") + assert result == data + + @pytest.mark.asyncio + async def test_clean_all(self, cacher): + await cacher.set("x", 1) + await cacher.set("y", 2) + await cacher.clean("*") + assert await cacher.get("x") is None + assert await cacher.get("y") is None + + +@skip_no_redis +class TestRedisCacherWithBroker: + """Test Redis cacher integrated with actual ServiceBroker.""" + + @pytest.mark.asyncio + async def test_broker_with_redis_cacher(self): + """Full lifecycle: broker.start → cache action → broker.stop.""" + from moleculerpy import Service, ServiceBroker, Settings, action + + class UserService(Service): + name = "users" + + @action(cache=True) + async def get(self, ctx): + return {"id": ctx.params["id"], "name": f"User-{ctx.params['id']}"} + + cacher = RedisCacher({"redis": REDIS_URL, "ttl": 60, "prefix": "BROKER-TEST-"}) + broker = ServiceBroker( + id="cache-test", + settings=Settings(transporter="memory://", log_level="CRITICAL"), + cacher=cacher, + ) + await broker.register(UserService()) + await broker.start() + + try: + # First call — cache miss + r1 = await broker.call("users.get", {"id": 42}) + assert r1["name"] == "User-42" + + # Second call — should hit cache (same result) + r2 = await broker.call("users.get", {"id": 42}) + assert r2 == r1 + + # Verify key exists in Redis + keys = await cacher.get_cache_keys() + assert len(keys) >= 1 + finally: + await cacher.clean("*") + await broker.stop() diff --git a/tests/unit/transit_test.py b/tests/unit/transit_test.py index 9b551cd..c015cf6 100644 --- a/tests/unit/transit_test.py +++ b/tests/unit/transit_test.py @@ -1678,6 +1678,74 @@ async def test_heartbeat_no_sender_ignored(self, mock_dependencies, mock_transpo transit.discover_node.assert_not_awaited() transit.node_catalog.get_node.assert_not_called() + @pytest.mark.asyncio + async def test_heartbeat_seq_mismatch_triggers_discover( + self, mock_dependencies, mock_transporter + ): + """Heartbeat with different seq triggers re-discovery (services changed).""" + with patch("moleculerpy.transit.Transporter.get_by_name", return_value=mock_transporter): + transit = Transit(**mock_dependencies) + mock_node = MagicMock() + mock_node.available = True + mock_node.seq = 5 + mock_node.instanceID = "abc-123" + transit.node_catalog = MagicMock() + transit.node_catalog.get_node.return_value = mock_node + transit._request_discovery = AsyncMock() + + packet = Packet(Topic.HEARTBEAT, "remote", {"cpu": 50, "seq": 10}) + packet.sender = "remote" + await transit._handle_heartbeat(packet) + + transit._request_discovery.assert_awaited_once_with("remote", "seq-changed") + + @pytest.mark.asyncio + async def test_heartbeat_instanceid_mismatch_triggers_discover( + self, mock_dependencies, mock_transporter + ): + """Heartbeat with different instanceID triggers re-discovery (node restarted).""" + with patch("moleculerpy.transit.Transporter.get_by_name", return_value=mock_transporter): + transit = Transit(**mock_dependencies) + mock_node = MagicMock() + mock_node.available = True + mock_node.seq = 5 + mock_node.instanceID = "abc-123" + transit.node_catalog = MagicMock() + transit.node_catalog.get_node.return_value = mock_node + transit._request_discovery = AsyncMock() + + packet = Packet( + Topic.HEARTBEAT, "remote", {"cpu": 50, "seq": 5, "instanceID": "xyz-999"} + ) + packet.sender = "remote" + await transit._handle_heartbeat(packet) + + transit._request_discovery.assert_awaited_once_with("remote", "instance-restarted") + + @pytest.mark.asyncio + async def test_heartbeat_same_seq_instanceid_updates_metrics( + self, mock_dependencies, mock_transporter + ): + """Heartbeat with matching seq/instanceID just updates metrics (no discovery).""" + with patch("moleculerpy.transit.Transporter.get_by_name", return_value=mock_transporter): + transit = Transit(**mock_dependencies) + mock_node = MagicMock() + mock_node.available = True + mock_node.seq = 5 + mock_node.instanceID = "abc-123" + transit.node_catalog = MagicMock() + transit.node_catalog.get_node.return_value = mock_node + transit._request_discovery = AsyncMock() + + packet = Packet( + Topic.HEARTBEAT, "remote", {"cpu": 80, "seq": 5, "instanceID": "abc-123"} + ) + packet.sender = "remote" + await transit._handle_heartbeat(packet) + + transit._request_discovery.assert_not_awaited() + assert mock_node.cpu == 80 + class TestRequestDiscovery: """Tests for _request_discovery rate-limiting and _handle_discover targeted reply."""