feat(protocol,lifecycle): 6-task sprint — hooks, proto schema, drain, tracking, tests

CHANGELOG.md

@@ -5,6 +5,54 @@ 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
+- **Kafka 2-node discovery** — 6 root causes fixed: batch make_subscriptions, heartbeat-driven
+  discover_node, targeted DISCOVER/INFO, broadcast PING subscription (Node.js parity)
+- **checkRemoteNodes timer** — marks nodes unavailable after heartbeat_timeout (Node.js base.js)
+- **checkOfflineNodes timer** — removes nodes silent >10 minutes (Node.js base.js)
+- **Auto-reconnect** — Transit.connect() retries with 5s backoff on transporter failure
+- **Demo matrix** — 28-cell smoke test (7 transports × 4 serializers)
+- **Comprehensive test suite** — 90 integration tests: lifecycle, actions, events, discovery,
+  errors, versioning, ping, multi-service across all 7 transports
+
+### Changed
+- **DRY Template Method** — receive()/publish()/get_topic_name() moved to base Transporter
+  (~250 lines eliminated). Subclasses only override send/connect/disconnect/_is_connected.
+- **Transit SRP** — discovery logic extracted to Discoverer (discover_all, discover_node,
+  request_discovery, _discover_pending). Transit delegates to broker.discoverer.
+- **SubscriptionTopic TypedDict** — replaces dict[str, Any] for type safety
+- **Typed attributes** — LatencyMonitor|None, logging.Logger, asyncio.Task[None]
+
+### Fixed
+- **AMQP broadcast consumer** — wrapped in try/except to prevent silent crash on unknown cmd
+- **PROTOCOL_VERSION dedup** — single source of truth in base.py (was duplicated 4x)
+- **NATS _is_connected** — checks nc.is_connected property, not just reference existence
+- **Discoverer stop()** — guards on _started flag, not _tasks (fixes zero-interval config)
+
 ## [0.14.19] - 2026-04-06
 
 ### Fixed (Audit-driven fixes for v0.14.18 serializers — PRD-019)

moleculerpy/__init__.py

@@ -39,7 +39,7 @@
 )
 from .serializers import BaseSerializer, JsonSerializer, MsgPackSerializer, resolve_serializer
 from .service import Service
-from .settings import Settings, SettingsValidationError
+from .settings import Settings, SettingsValidationError, TrackingConfig
 from .stream import AsyncStream, StreamError
 from .tracing import (
     BaseTraceExporter,
@@ -54,7 +54,7 @@
 try:
     __version__ = version("moleculerpy")
 except PackageNotFoundError:
-    __version__ = "0.14.19"
+    __version__ = "0.14.21"
 
 __all__ = [  # noqa: RUF022
     # Core
@@ -65,6 +65,7 @@
     "Lifecycle",
     "Settings",
     "SettingsValidationError",
+    "TrackingConfig",
     "NodeID",
     "ServiceName",
     "ActionName",

moleculerpy/broker.py

@@ -137,6 +137,16 @@ def __init__(
 
         self._validator = resolve_validator(getattr(self.settings, "validator", "default"))
 
+        # Auto-register ContextTracker middleware if tracking enabled
+        tracking_cfg = getattr(self.settings, "tracking", None)
+        if tracking_cfg is not None and getattr(tracking_cfg, "enabled", False):
+            from .middleware.context_tracker import ContextTrackerMiddleware  # noqa: PLC0415
+
+            # TrackingConfig.shutdown_timeout is float seconds;
+            # ContextTrackerMiddleware expects int milliseconds.
+            shutdown_timeout_ms = int(tracking_cfg.shutdown_timeout * 1000)
+            self.middlewares.append(ContextTrackerMiddleware(shutdown_timeout=shutdown_timeout_ms))
+
         # Wrapped event methods (set during start() by middleware)
         self._wrapped_emit: (
             Callable[[str, dict[str, Any], dict[str, Any]], Awaitable[Any]] | None
@@ -264,21 +274,50 @@ def _call_middleware_hooks(
         Returns:
             List of coroutines if is_async=True, None otherwise
         """
+        # Node.js Moleculer uses short hook names (starting/started/stopping/stopped)
+        # while MoleculerPy historically used broker_* names. To maintain backward
+        # compatibility AND Node.js ecosystem compatibility, we invoke both names.
+        # Note: "stopped" is intentionally NOT aliased because MoleculerPy's existing
+        # stopped() hook takes no arguments (middleware self-cleanup), which would
+        # collide with Node.js stopped(broker) signature.
+        from .middleware.base import Middleware as _BaseMiddleware  # noqa: PLC0415
+
+        _broker_hook_aliases = {
+            "broker_starting": "starting",
+            "broker_started": "started",
+            "broker_stopping": "stopping",
+        }
+        alias = _broker_hook_aliases.get(hook_name)
+
+        def _is_overridden(mw: Any, name: str) -> bool:
+            """True if mw class overrides the alias method (not base no-op)."""
+            mw_method = getattr(type(mw), name, None)
+            base_method = getattr(_BaseMiddleware, name, None)
+            return mw_method is not None and mw_method is not base_method
+
         if is_async:
             coroutines = []
             for middleware in self.middlewares:
-                hook = getattr(middleware, hook_name, None)
-                if hook and callable(hook):
-                    result = hook(*args)
-                    if asyncio.iscoroutine(result):
-                        coroutines.append(result)
+                names = [hook_name]
+                if alias and _is_overridden(middleware, alias):
+                    names.append(alias)
+                for name in names:
+                    hook = getattr(middleware, name, None)
+                    if hook and callable(hook):
+                        result = hook(*args)
+                        if asyncio.iscoroutine(result):
+                            coroutines.append(result)
             return coroutines if coroutines else None
         else:
             # Synchronous hooks
             for middleware in self.middlewares:
-                hook = getattr(middleware, hook_name, None)
-                if hook and callable(hook):
-                    hook(*args)
+                names = [hook_name]
+                if alias and _is_overridden(middleware, alias):
+                    names.append(alias)
+                for name in names:
+                    hook = getattr(middleware, name, None)
+                    if hook and callable(hook):
+                        hook(*args)
             return None
 
     async def _execute_middleware_hooks(
@@ -559,6 +598,13 @@ async def _stop_core() -> None:
             if self.cacher:
                 await self.cacher.stop()
 
+            # Drain: notify peers we're shutting down (empty services) so
+            # they stop routing new requests to us BEFORE we DISCONNECT.
+            try:
+                await self.transit.send_disconnect_info()
+            except Exception as e:
+                self.logger.warning(f"Error sending drain INFO: {e}")
+
             # Disconnect from the cluster
             await self.transit.disconnect()
 

moleculerpy/middleware/base.py

@@ -156,6 +156,27 @@ async def broker_stopped(self, broker: Any) -> None:
         """
         pass
 
+    # ==========================================================================
+    # Node.js Moleculer-compatible short aliases for broker lifecycle hooks.
+    # The broker invokes BOTH the long-form (broker_*) and the short alias name
+    # so middleware authored against the Node.js naming convention works as-is.
+    # Note: "stopped" is intentionally NOT aliased because MoleculerPy's existing
+    # stopped() hook (below) takes no arguments and is reserved for middleware
+    # self-cleanup. Use broker_stopped() if you need a broker reference.
+    # ==========================================================================
+
+    async def starting(self, broker: Any) -> None:
+        """Node.js-compatible alias for broker_starting. Called BEFORE connect."""
+        pass
+
+    async def started(self, broker: Any) -> None:
+        """Node.js-compatible alias for broker_started. Called AFTER connect."""
+        pass
+
+    async def stopping(self, broker: Any) -> None:
+        """Node.js-compatible alias for broker_stopping. Called BEFORE disconnect."""
+        pass
+
     async def service_creating(self, service: Any) -> None:
         """
         Hook called asynchronously BEFORE a service is registered.

moleculerpy/serializers/proto/packets.proto

@@ -92,9 +92,17 @@ message PacketDisconnect {
 }
 
 message PacketHeartbeat {
-	string ver 		= 1;
-	string sender 	= 2;
-	double cpu 		= 3;
+	string ver 			= 1;
+	string sender 		= 2;
+	double cpu 			= 3;
+	// MoleculerPy extension fields (4-7) for restart detection.
+	// Wire-compatible with Node.js: unknown fields are silently ignored
+	// by Node.js ProtoBuf parser. Field numbers 4-7 are RESERVED FOREVER.
+	// See: .forgeplan/adrs/ADR-heartbeat-schema.md
+	int32 seq 			= 4;
+	string instanceID 	= 5;
+	double memory 		= 6;
+	int32 cpuSeq 		= 7;
 }
 
 message PacketPing {

moleculerpy/serializers/protobuf.py

@@ -47,8 +47,9 @@
 # Separate from BaseSerializer.MAX_PAYLOAD_BYTES which guards the whole frame.
 MAX_NESTED_FIELD_BYTES: Final[int] = 1 * 1024 * 1024  # 1MB per nested field
 
-# Heuristic constant: HEARTBEAT packet has small field count (ver, sender, cpu, +1 optional)
-_HEARTBEAT_MAX_FIELDS: Final[int] = 4
+# Heuristic constant: HEARTBEAT packet field count.
+# Schema: ver, sender, cpu, seq, instanceID, memory, cpuSeq (+1 optional slack).
+_HEARTBEAT_MAX_FIELDS: Final[int] = 8
 
 
 def _check_json_depth(text: str, max_depth: int = MAX_JSON_DEPTH) -> bool: