Skip to content

fix(modbus): serialize gateway access to prevent slave ID mismatch #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Jonathan0823 merged 5 commits into from
Apr 16, 2026
Merged

fix(modbus): serialize gateway access to prevent slave ID mismatch #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c2b778c
chore(deps): update pymodbus to 3.13.0
Jonathan0823 Apr 16, 2026
286c5cc
fix(modbus): serialize gateway access to prevent slave ID mismatch
Jonathan0823 Apr 16, 2026
0274aa8
fix(tests): update mock assertions for pymodbus 3.13 API
Jonathan0823 Apr 16, 2026
399934b
fix(modbus): close stale gateways on config reload
Jonathan0823 Apr 16, 2026
9a9c980
fix(modbus): return None on invalid response after retries
Jonathan0823 Apr 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 129 additions & 17 deletions app/core/modbus_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,10 +223,11 @@ def _read_registers(
for attempt in range(num_attempts):
try:
response = read_method(
address=address, count=count, slave=slave_id
address=address, count=count, device_id=slave_id
)
last_response = response
# Only assign last_response after validation succeeds
if self._is_valid_response(response, op_name, slave_id):
last_response = response
if attempt > 0:
logger.info(
"modbus_read_success_after_retry",
Expand Down Expand Up @@ -314,7 +315,8 @@ def _read_registers(
# This prevents response mix-up when multiple slave IDs share a connection
self.close()

return last_response
# Return None - do not return potentially invalid last_response
return None
finally:
self._restore_timeout(orig_timeout)

Expand All @@ -340,11 +342,15 @@ def write_holding_register(self, slave_id: int, address: int, value: int):
for attempt in range(self.max_retries):
try:
response = self._client.write_register(
address=address, value=value, slave=slave_id
address=address, value=value, device_id=slave_id
)
last_response = response
# Only assign last_response after validation succeeds
if self._is_valid_response(response, "write_holding_register", slave_id):
last_response = response
return response
else:
# Invalid response - clear last_response
last_response = None
except (ModbusException, OSError) as exc:
logger.warning(
"modbus_write_exception",
Expand All @@ -370,7 +376,9 @@ def write_holding_register(self, slave_id: int, address: int, value: int):
max_attempts=self.max_retries,
)
time.sleep(self.retry_delay)
return last_response

# Return None - do not return potentially invalid last_response
return None

def is_connected(self) -> bool:
return bool(self._client and self._client.is_socket_open())
Expand Down Expand Up @@ -399,8 +407,10 @@ def __init__(self, device_configs: Iterable[DeviceConfig]) -> None:
cfg.device_id: cfg for cfg in device_configs
}
# Map (host, port) -> ModbusGateway
# IMPORTANT: One gateway per host:port, NOT per slave_id
# All devices sharing the same host:port share ONE connection
self._gateways: Dict[Tuple[str, int], ModbusGateway] = {}
# Map (host, port) -> Lock
# Map (host, port) -> Lock - ensures serialized access per gateway
self._locks: Dict[Tuple[str, int], asyncio.Lock] = {}
Comment on lines +410 to 414

@coderabbitai coderabbitai Bot Apr 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Avoid tearing down the shared socket for one device's Modbus-level failure.

Once (host, port) is the shared key, the existing self.close() paths in ModbusGateway._read_registers() / write_holding_register() become global to every slave behind that gateway. A bad frame from one device now drops the shared connection for all queued peers, which reintroduces the connection churn this PR is trying to remove.

Also applies to: 498-505

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@app/core/modbus_client.py` around lines 402 - 406, The shared-gateway design
is being broken by calls to ModbusGateway.close() inside
ModbusGateway._read_registers() and write_holding_register(), which tear down
the single (host, port) socket for all slaves; change the error handling so that
protocol/frame-level errors for a specific slave do NOT call self.close() on the
gateway—only close the gateway on transport-level failures (socket/connection
exceptions). Update ModbusGateway._read_registers and write_holding_register to
distinguish ModbusResponse/CRC/frame errors (raise or return a per-request error
without closing the gateway) from actual connection errors (where calling
self.close() is appropriate), and ensure caller code that uses self._gateways
and self._locks treats per-request failures as isolated to that slave rather
than triggering gateway removal.

self._manager_lock = asyncio.Lock()

Expand All @@ -411,8 +421,16 @@ def __init__(self, device_configs: Iterable[DeviceConfig]) -> None:
recovery_timeout=float(settings.CIRCUIT_BREAKER_RECOVERY_TIMEOUT),
)
)

# Validate for problematic configurations
self._validate_gateway_config()

def _create_gateway(self, config: DeviceConfig) -> ModbusGateway:
"""Create a gateway for a device.

Note: Gateway is shared across all devices on the same host:port,
regardless of slave_id. Access is serialized via locks.
"""
return ModbusGateway(
host=config.host,
port=config.port,
Expand All @@ -422,18 +440,77 @@ def _create_gateway(self, config: DeviceConfig) -> ModbusGateway:
retry_delay=config.retry_delay,
)

def _validate_gateway_config(self) -> None:
"""Check for problematic configurations across all devices.

Validates:
- Multiple devices on same host:port with different slave_ids (allowed but warned)
- Duplicate (host, port, slave_id) combinations (error)
"""
# Track (host, port, slave_id) -> list of device_ids
endpoint_map: Dict[Tuple[str, int, int], List[str]] = {}
# Track (host, port) -> list of device_ids for warnings
gateway_map: Dict[Tuple[str, int], List[str]] = {}

for device_id, config in self._configs.items():
endpoint_key = (config.host, config.port, config.slave_id)
gateway_key = (config.host, config.port)

if endpoint_key not in endpoint_map:
endpoint_map[endpoint_key] = []
endpoint_map[endpoint_key].append(device_id)

if gateway_key not in gateway_map:
gateway_map[gateway_key] = []
gateway_map[gateway_key].append(device_id)

# Check for duplicate endpoints (exact same host:port:slave_id)
duplicates = {key: devs for key, devs in endpoint_map.items() if len(devs) > 1}
if duplicates:
for (host, port, slave_id), device_ids in duplicates.items():
logger.error(
"modbus_duplicate_endpoint_error",
host=host,
port=port,
slave_id=slave_id,
device_ids=device_ids,
message=(
f"CRITICAL: Multiple devices with identical endpoint: "
f"{host}:{port} slave_id={slave_id} -> devices: {device_ids}"
),
)

# Warn about shared gateways (multiple devices on same host:port)
shared_gateways = {key: devs for key, devs in gateway_map.items() if len(devs) > 1}
if shared_gateways:
for (host, port), device_ids in shared_gateways.items():
slave_ids = [self._configs[did].slave_id for did in device_ids]
logger.warning(
"modbus_shared_gateway",
host=host,
port=port,
device_ids=device_ids,
slave_ids=slave_ids,
message=(
f"Multiple devices share gateway {host}:{port} - "
f"polling will be serialized to prevent response mix-up. "
f"Devices: {device_ids} (slave_ids: {slave_ids})"
),
)

async def _get_gateway_and_lock(self, device_id: str) -> Tuple[ModbusGateway, asyncio.Lock]:
config = self._configs.get(device_id)
if not config:
raise DeviceNotFoundError(f"Unknown device_id '{device_id}'")

key = (config.host, config.port)
# Key is (host, port) only - all devices on same gateway share connection
gateway_key = (config.host, config.port)

async with self._manager_lock:
if key not in self._gateways:
self._gateways[key] = self._create_gateway(config)
self._locks[key] = asyncio.Lock()
return self._gateways[key], self._locks[key]
if gateway_key not in self._gateways:
self._gateways[gateway_key] = self._create_gateway(config)
self._locks[gateway_key] = asyncio.Lock()
return self._gateways[gateway_key], self._locks[gateway_key]

async def _run_with_gateway(self, device_id: str, func_name: str, *args, **kwargs):
"""Execute a gateway method directly without circuit breaker.
Expand Down Expand Up @@ -551,19 +628,22 @@ async def reset_gateway(self, device_id: str) -> None:

This is useful when a connection becomes unresponsive.
The next request will create a fresh connection.

Note: Resets the entire gateway (host:port), not just one device.
"""
config = self._configs.get(device_id)
if not config:
raise DeviceNotFoundError(f"Unknown device_id '{device_id}'")

key = (config.host, config.port)
# Key is (host, port) - resets the shared gateway
gateway_key = (config.host, config.port)

async with self._manager_lock:
if key in self._gateways:
gateway = self._gateways[key]
if gateway_key in self._gateways:
gateway = self._gateways[gateway_key]
await asyncio.to_thread(gateway.close)
del self._gateways[key]
del self._locks[key]
del self._gateways[gateway_key]
del self._locks[gateway_key]
Comment on lines 641 to +646

@coderabbitai coderabbitai Bot Apr 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Coordinate reset_gateway() with the existing per-gateway lock.

Deleting _gateways[gateway_key] and _locks[gateway_key] under _manager_lock alone is not enough. A task that already captured the old (gateway, lock) can keep running while the next task creates a fresh pair for the same host:port, so two operations can overlap during a reset/reload window and defeat the serialization guarantee.

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@app/core/modbus_client.py` around lines 633 - 638, reset_gateway() currently
only uses _manager_lock when deleting entries, which allows a racing task
holding the old per-gateway lock to overlap with a new gateway creation; fix by
coordinating with the per-gateway lock: lookup and acquire the existing
per-gateway lock (self._locks[gateway_key]) before closing/deleting the gateway,
then while holding that per-gateway lock also hold _manager_lock to remove
entries (call await lock.acquire() on the existing lock, await
asyncio.to_thread(gateway.close), then acquire _manager_lock and del
self._gateways[gateway_key] and del self._locks[gateway_key]), and finally
release the per-gateway lock; ensure you handle the case where no per-gateway
lock exists (skip) and avoid deadlock by always acquiring the per-gateway lock
before _manager_lock in this code path.

logger.info(
"modbus_gateway_reset",
device_id=device_id,
Expand All @@ -576,6 +656,7 @@ async def reload_configs(self, new_configs: Iterable[DeviceConfig]) -> None:
"""Reload device configurations dynamically.

This closes connections for removed devices and updates the config map.
Also handles devices that moved to different host/port (stale gateway cleanup).
"""
old_device_ids = set(self._configs.keys())
new_device_ids = {cfg.device_id for cfg in new_configs}
Expand All @@ -588,8 +669,29 @@ async def reload_configs(self, new_configs: Iterable[DeviceConfig]) -> None:
except DeviceNotFoundError:
pass # Already removed

# Check for devices that changed host/port (moved endpoints)
moved_devices: List[str] = []
for new_cfg in new_configs:
if new_cfg.device_id in self._configs:
old_cfg = self._configs[new_cfg.device_id]
old_endpoint = (old_cfg.host, old_cfg.port)
new_endpoint = (new_cfg.host, new_cfg.port)
if old_endpoint != new_endpoint:
# Device moved to different endpoint - close old gateway
old_gateway_key = old_endpoint
if old_gateway_key in self._gateways:
try:
await self.reset_gateway(new_cfg.device_id)
moved_devices.append(new_cfg.device_id)
except DeviceNotFoundError:
pass

# Update configs
self._configs = {cfg.device_id: cfg for cfg in new_configs}

# Re-validate gateway config after reload
self._validate_gateway_config()

logger.info(
"modbus_configs_reloaded",
device_count=len(new_configs),
Expand All @@ -613,9 +715,19 @@ def get_gateways_status(self) -> List[dict]:
"""Return status of all active gateways."""
status_list = []
for (host, port), gateway in self._gateways.items():
# Get all devices using this gateway
device_ids = [
dev_id for dev_id, cfg in self._configs.items()
if cfg.host == host and cfg.port == port
]
slave_ids = [
self._configs[dev_id].slave_id for dev_id in device_ids
]
status_list.append({
"host": host,
"port": port,
"device_ids": device_ids,
"slave_ids": slave_ids,
"connected": gateway.is_connected(),
})
return status_list
Expand Down
Loading
Loading