12 Commits

Author SHA1 Message Date
Joseph Doherty dcdf79afdc fix(dcl): format ArrayValue objects as comma-separated strings for display
ArrayValue from LmxProxy client was showing as type name in debug views.
Added ValueFormatter utility and NormalizeValue in LmxProxyDataConnection
to convert arrays at the adapter boundary. DateTime arrays remain as
"System.DateTime[]" due to server-side v1 string serialization.
2026-03-22 14:46:15 -04:00
Joseph Doherty ea9c2857a7 fix(docker,cli): add LmxProxy.Client to Docker build, fix set-bindings JSON parsing
Docker: include lmxproxy/src/ZB.MOM.WW.LmxProxy.Client in build context
so the project reference resolves during container image build.

CLI: fix set-bindings JSON parsing — use JsonElement.GetString()/GetInt32()
instead of object.ToString() which returned null for deserialized elements.
2026-03-22 14:25:09 -04:00
Joseph Doherty 847302e297 test(dcl): add failover state machine tests for DataConnectionActor 2026-03-22 08:47:44 -04:00
Joseph Doherty 5de6c8d052 docs(dcl): document primary/backup endpoint redundancy across requirements and test infra 2026-03-22 08:43:59 -04:00
Joseph Doherty e8df71ea64 feat(cli): add --primary-config, --backup-config, --failover-retry-count to data connection commands
Thread backup data connection fields through management command messages,
ManagementActor handlers, SiteService, site-side SQLite storage, and
deployment/replication actors. The old --configuration CLI flag is kept
as a hidden alias for backwards compatibility.
2026-03-22 08:41:57 -04:00
Joseph Doherty ab4e88f17f feat(ui): add primary/backup endpoint fields to data connection form 2026-03-22 08:36:18 -04:00
Joseph Doherty 801c0c1df2 feat(dcl): add active endpoint to health reports and log failover events
Add ActiveEndpoint field to DataConnectionHealthReport showing which
endpoint is active (Primary, Backup, or Primary with no backup configured).
Log failover transitions and connection restoration events to the site
event log via ISiteEventLogger, passed as an optional parameter through
the actor hierarchy for backwards compatibility.
2026-03-22 08:34:05 -04:00
Joseph Doherty da290fa4f8 feat(dcl): add failover state machine to DataConnectionActor with round-robin endpoint switching 2026-03-22 08:30:03 -04:00
Joseph Doherty 46304678da feat(dcl): extend CreateConnectionCommand with backup config and failover retry count
Update CreateConnectionCommand to carry PrimaryConnectionDetails,
BackupConnectionDetails, and FailoverRetryCount. Update all callers:
DataConnectionManagerActor, DataConnectionActor, DeploymentManagerActor,
FlatteningService, and ConnectionConfig. The actor stores both configs
but continues using primary only — failover logic comes in Task 3.
2026-03-22 08:24:39 -04:00
Joseph Doherty 04af03980e feat(dcl): rename Configuration to PrimaryConfiguration, add BackupConfiguration and FailoverRetryCount 2026-03-22 08:18:31 -04:00
Joseph Doherty 5ca1be328c docs(dcl): add primary/backup data connections implementation plan
8 tasks with TDD steps, exact file paths, and code samples.
Covers entity model, failover state machine, health reporting,
UI, CLI, management API, deployment, and documentation.
2026-03-22 08:13:23 -04:00
Joseph Doherty 6267ff882c docs(dcl): add primary/backup data connection endpoints design
Covers entity model, failover state machine, health reporting,
UI/CLI changes, and deployment flow for optional backup endpoints
with automatic failover after configurable retry count.
2026-03-22 08:09:25 -04:00
40 changed files with 2257 additions and 64 deletions
+3 -1
View File
@@ -21,14 +21,16 @@ COPY src/ScadaLink.ClusterInfrastructure/ScadaLink.ClusterInfrastructure.csproj
COPY src/ScadaLink.InboundAPI/ScadaLink.InboundAPI.csproj src/ScadaLink.InboundAPI/
COPY src/ScadaLink.ConfigurationDatabase/ScadaLink.ConfigurationDatabase.csproj src/ScadaLink.ConfigurationDatabase/
COPY src/ScadaLink.ManagementService/ScadaLink.ManagementService.csproj src/ScadaLink.ManagementService/
COPY lmxproxy/src/ZB.MOM.WW.LmxProxy.Client/ZB.MOM.WW.LmxProxy.Client.csproj lmxproxy/src/ZB.MOM.WW.LmxProxy.Client/
# Restore NuGet packages via Host project (follows ProjectReferences to all 17 dependencies)
# Restore NuGet packages via Host project (follows ProjectReferences to all dependencies)
# This layer is cached until any .csproj changes — source-only changes skip restore entirely
RUN dotnet restore src/ScadaLink.Host/ScadaLink.Host.csproj
# Stage 2: Build + Publish
FROM restore AS build
COPY src/ src/
COPY lmxproxy/src/ZB.MOM.WW.LmxProxy.Client/ lmxproxy/src/ZB.MOM.WW.LmxProxy.Client/
RUN dotnet publish src/ScadaLink.Host/ScadaLink.Host.csproj \
-c Release -o /app/publish
@@ -0,0 +1,151 @@
# Primary/Backup Data Connection Endpoints — Design
**Date:** 2026-03-22
**Status:** Approved
## Problem
Data connections currently support a single endpoint. If that endpoint goes down, the connection retries indefinitely at 5s intervals against the same address. When redundant infrastructure exists (e.g., two LmxProxy instances, two OPC UA servers), there is no way to automatically fail over to a backup.
## Design Decisions
| Decision | Choice |
|----------|--------|
| Failover mode | Automatic after N failed retries |
| Failback | No auto-failback; stay on active until it fails (round-robin) |
| Backup required? | Optional — single-endpoint connections work unchanged |
| Failover trigger | After configurable retry count (default 3) |
| Entity model | Separate `PrimaryConfiguration` and `BackupConfiguration` columns |
| UI approach | Two JSON text areas; backup collapsible |
| Failover logic location | DataConnectionActor (adapters stay single-endpoint) |
| Observability | Health reports + site event log entries |
## Entity Model
**`DataConnection` changes:**
| Field | Type | Notes |
|-------|------|-------|
| `PrimaryConfiguration` | string? (max 4000) | Renamed from `Configuration` |
| `BackupConfiguration` | string? (max 4000) | New. Null = no backup |
| `FailoverRetryCount` | int (default 3) | New. Retries before switching |
Both endpoints use the same `Protocol`. EF Core migration renames `Configuration``PrimaryConfiguration` (data-preserving).
**`DataConnectionArtifact` changes:**
- `ConfigurationJson``PrimaryConfigurationJson` + `BackupConfigurationJson`
## Failover State Machine
The `DataConnectionActor` Reconnecting state is extended:
```
Connected
│ disconnect detected
Push bad quality to all subscribers
Retry active endpoint (5s interval)
│ failure
_consecutiveFailures++
├─ < FailoverRetryCount → retry same endpoint
├─ ≥ FailoverRetryCount AND backup exists
│ → dispose adapter, switch _activeEndpoint, reset counter
│ → create fresh adapter with other config
│ → attempt connect
└─ ≥ FailoverRetryCount AND no backup
→ keep retrying indefinitely (current behavior)
```
**On successful reconnect (either endpoint):**
1. Reset `_consecutiveFailures = 0`
2. `ReSubscribeAll()` — re-create all subscriptions on the new adapter
3. Transition to Connected
4. Log failover event if endpoint changed
5. Report active endpoint in health metrics
**Round-robin on failure:** primary → backup → primary → backup...
**Adapter lifecycle on failover:** Actor disposes current `IDataConnection` adapter and creates a fresh one via `DataConnectionFactory.Create()` with the other endpoint's config. Clean slate — no stale state.
## Actor State
New fields in `DataConnectionActor`:
- `IDictionary<string, string> _primaryConfig`
- `IDictionary<string, string>? _backupConfig`
- `ActiveEndpoint _activeEndpoint` (enum: Primary, Backup)
- `int _consecutiveFailures`
- `int _failoverRetryCount`
`CreateConnectionCommand` gains: `primaryConfig`, `backupConfig`, `failoverRetryCount`.
`DataConnectionFactory` is unchanged — still creates single-endpoint adapters.
## Health & Observability
**`DataConnectionHealthReport`** gains:
- `ActiveEndpoint` (string): `"Primary"`, `"Backup"`, or `"Primary (no backup)"`
**Site event log entries:**
- `DataConnectionFailover` — connection name, from-endpoint, to-endpoint, reason
- `DataConnectionRestored` — connection name, active endpoint
Uses existing `ISiteEventLogger`.
## Central UI
**List page:** Add `Active Endpoint` column from health reports.
**Form (Create/Edit):**
- "Primary Endpoint Configuration" label (renamed from "Configuration")
- "Add Backup Endpoint" button reveals second JSON text area
- "Remove Backup" button in edit mode when backup exists
- "Failover Retry Count" numeric input (default 3, min 1, max 20) — visible only when backup configured
- Vertical stacking, collapsible backup subsection
## CLI
- `--configuration` renamed to `--primary-config` (hidden alias for backwards compat)
- `--backup-config` (optional)
- `--failover-retry-count` (optional, default 3)
- `data-connection get` shows both configs and active endpoint
## Management API
- `CreateDataConnectionCommand` / `UpdateDataConnectionCommand` gain `PrimaryConfiguration`, `BackupConfiguration`, `FailoverRetryCount`
- Setting `BackupConfiguration` to null removes the backup
- `GetDataConnectionResponse` returns both configs
## Deployment Flow
`DataConnectionArtifact` carries `PrimaryConfigurationJson` and `BackupConfigurationJson`. Site-side deployment handler passes both to `CreateConnectionCommand`.
## Testing
**Unit tests:**
- Actor: failover after N failures, round-robin, single-endpoint retries forever, counter reset, ReSubscribeAll on failover
- Manager actor: updated CreateConnectionCommand
- Factory: unchanged registration
**Integration test (manual with test infra):**
1. Primary=`opc.tcp://localhost:50000`, backup=`opc.tcp://localhost:50010`
2. Subscribe to `Motor.Speed`
3. `docker compose stop opcua` → verify failover to opcua2 after 3 retries
4. `docker compose stop opcua2 && docker compose start opcua` → verify round-robin back
## Implementation Tasks
1. **#4** Entity model & database (foundation)
2. **#6** CreateConnectionCommand & DataConnectionManagerActor (blocked by #4)
3. **#5** DataConnectionActor failover state machine (blocked by #4, #6)
4. **#7** Health reporting & site event log (blocked by #5)
5. **#8** Central UI (blocked by #4)
6. **#9** CLI, Management API, deployment (blocked by #4)
7. **#10** Documentation (blocked by #5)
8. **#11** Tests (blocked by #5)
@@ -0,0 +1,695 @@
# Primary/Backup Data Connection Endpoints — Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.
**Goal:** Add optional backup endpoints to data connections with automatic failover after configurable retry count.
**Architecture:** The `DataConnectionActor` gains failover logic in its Reconnecting state — after N failed retries on the active endpoint, it disposes the adapter and creates a fresh one with the other endpoint's config. Adapters remain single-endpoint. Entity model splits `Configuration` into `PrimaryConfiguration` + `BackupConfiguration`.
**Tech Stack:** C# / .NET 10, Akka.NET, EF Core, Blazor Server, System.CommandLine
**Design doc:** `docs/plans/2026-03-22-primary-backup-data-connections-design.md`
---
## Task 1: Entity Model & Database Migration
**Files:**
- Modify: `src/ScadaLink.Commons/Entities/Sites/DataConnection.cs`
- Modify: `src/ScadaLink.ConfigurationDatabase/Configurations/SiteConfiguration.cs` (lines 32-56)
- Modify: `src/ScadaLink.Commons/Messages/Artifacts/DataConnectionArtifact.cs`
### Step 1: Update DataConnection entity
In `DataConnection.cs`, rename `Configuration` to `PrimaryConfiguration`, add `BackupConfiguration` and `FailoverRetryCount`:
```csharp
public class DataConnection
{
public int Id { get; set; }
public int SiteId { get; set; }
public string Name { get; set; }
public string Protocol { get; set; }
public string? PrimaryConfiguration { get; set; }
public string? BackupConfiguration { get; set; }
public int FailoverRetryCount { get; set; } = 3;
public DataConnection(int siteId, string name, string protocol)
{
SiteId = siteId;
Name = name ?? throw new ArgumentNullException(nameof(name));
Protocol = protocol ?? throw new ArgumentNullException(nameof(protocol));
}
}
```
### Step 2: Update EF Core mapping
In `SiteConfiguration.cs`, update the DataConnection mapping (around lines 46-47):
- Rename `Configuration` property mapping to `PrimaryConfiguration` (MaxLength 4000)
- Add `BackupConfiguration` property (optional, MaxLength 4000)
- Add `FailoverRetryCount` property (required, default 3)
```csharp
builder.Property(d => d.PrimaryConfiguration).HasMaxLength(4000);
builder.Property(d => d.BackupConfiguration).HasMaxLength(4000);
builder.Property(d => d.FailoverRetryCount).HasDefaultValue(3);
```
### Step 3: Create EF Core migration
Run:
```bash
cd src/ScadaLink.ConfigurationDatabase
dotnet ef migrations add AddDataConnectionBackupEndpoint \
--startup-project ../ScadaLink.Host
```
Verify the migration renames `Configuration``PrimaryConfiguration` (should use `RenameColumn`, not drop+add). If the scaffolded migration drops and recreates, manually fix it:
```csharp
migrationBuilder.RenameColumn(
name: "Configuration",
table: "DataConnections",
newName: "PrimaryConfiguration");
migrationBuilder.AddColumn<string>(
name: "BackupConfiguration",
table: "DataConnections",
maxLength: 4000,
nullable: true);
migrationBuilder.AddColumn<int>(
name: "FailoverRetryCount",
table: "DataConnections",
nullable: false,
defaultValue: 3);
```
### Step 4: Update DataConnectionArtifact
In `DataConnectionArtifact.cs`, replace single `ConfigurationJson` with both:
```csharp
public record DataConnectionArtifact(
string Name,
string Protocol,
string? PrimaryConfigurationJson,
string? BackupConfigurationJson,
int FailoverRetryCount = 3);
```
### Step 5: Build and fix compile errors
Run: `dotnet build ScadaLink.slnx`
This will surface all references to the old `Configuration` and `ConfigurationJson` fields across the codebase. Fix each one — this includes:
- ManagementActor handlers
- CLI commands
- UI pages
- Deployment/flattening code
- Tests
Fix only the field name renames in this step (use `PrimaryConfiguration` where `Configuration` was). Don't add backup logic yet — just make it compile.
### Step 6: Run tests, fix failures
Run: `dotnet test ScadaLink.slnx`
Fix any test failures caused by the rename.
### Step 7: Commit
```bash
git add -A
git commit -m "feat(dcl): rename Configuration to PrimaryConfiguration, add BackupConfiguration and FailoverRetryCount"
```
---
## Task 2: Update CreateConnectionCommand & Manager Actor
**Files:**
- Modify: `src/ScadaLink.Commons/Messages/DataConnection/CreateConnectionCommand.cs`
- Modify: `src/ScadaLink.DataConnectionLayer/Actors/DataConnectionManagerActor.cs` (lines 39-62)
### Step 1: Update CreateConnectionCommand message
```csharp
public record CreateConnectionCommand(
string ConnectionName,
string ProtocolType,
IDictionary<string, string> PrimaryConnectionDetails,
IDictionary<string, string>? BackupConnectionDetails = null,
int FailoverRetryCount = 3);
```
### Step 2: Update DataConnectionManagerActor.HandleCreateConnection
Update the handler (around line 39-62) to pass both configs to DataConnectionActor:
```csharp
private void HandleCreateConnection(CreateConnectionCommand command)
{
if (_connectionActors.ContainsKey(command.ConnectionName))
{
_log.Warning("Connection {0} already exists", command.ConnectionName);
return;
}
var adapter = _factory.Create(command.ProtocolType, command.PrimaryConnectionDetails);
var props = Props.Create(() => new DataConnectionActor(
command.ConnectionName,
adapter,
_options,
_healthCollector,
command.ProtocolType,
command.PrimaryConnectionDetails,
command.BackupConnectionDetails,
command.FailoverRetryCount));
var actorName = new string(command.ConnectionName
.Select(c => char.IsLetterOrDigit(c) || "-_.*$+:@&=,!~';()".Contains(c) ? c : '-')
.ToArray());
var actorRef = Context.ActorOf(props, actorName);
_connectionActors[command.ConnectionName] = actorRef;
_log.Info("Created DataConnectionActor for {0} (protocol={1}, backup={2})",
command.ConnectionName, command.ProtocolType, command.BackupConnectionDetails != null ? "yes" : "none");
}
```
### Step 3: Update all callers of CreateConnectionCommand
Search for all places that construct `CreateConnectionCommand` and update them to use the new signature. The primary caller is the site-side deployment handler.
### Step 4: Build and test
Run: `dotnet build ScadaLink.slnx && dotnet test tests/ScadaLink.DataConnectionLayer.Tests`
### Step 5: Commit
```bash
git add -A
git commit -m "feat(dcl): extend CreateConnectionCommand with backup config and failover retry count"
```
---
## Task 3: DataConnectionActor Failover State Machine
**Files:**
- Modify: `src/ScadaLink.DataConnectionLayer/Actors/DataConnectionActor.cs`
- Modify: `src/ScadaLink.DataConnectionLayer/DataConnectionFactory.cs`
This is the core change. The actor gains failover logic in its Reconnecting state.
### Step 1: Add new state fields to DataConnectionActor
Add these fields alongside the existing ones (around line 30):
```csharp
private readonly string _protocolType;
private readonly IDictionary<string, string> _primaryConfig;
private readonly IDictionary<string, string>? _backupConfig;
private readonly int _failoverRetryCount;
private readonly IDataConnectionFactory _factory;
private ActiveEndpoint _activeEndpoint = ActiveEndpoint.Primary;
private int _consecutiveFailures;
public enum ActiveEndpoint { Primary, Backup }
```
### Step 2: Update constructor
Extend the constructor to accept both configs and the factory:
```csharp
public DataConnectionActor(
string connectionName,
IDataConnection adapter,
DataConnectionOptions options,
ISiteHealthCollector healthCollector,
string protocolType,
IDictionary<string, string> primaryConfig,
IDictionary<string, string>? backupConfig = null,
int failoverRetryCount = 3)
{
_connectionName = connectionName;
_adapter = adapter;
_options = options;
_healthCollector = healthCollector;
_protocolType = protocolType;
_primaryConfig = primaryConfig;
_backupConfig = backupConfig;
_failoverRetryCount = failoverRetryCount;
_connectionDetails = primaryConfig; // start with primary
}
```
Note: The actor also needs `IDataConnectionFactory` injected to create new adapters on failover. Pass it through the constructor or resolve via DI. The `DataConnectionManagerActor` already has the factory — pass it through to the actor constructor.
### Step 3: Extend HandleReconnectResult with failover logic
Replace the reconnect failure handling (around lines 279-296) to include failover:
```csharp
private void HandleReconnectResult(ConnectResult result)
{
if (result.Success)
{
_consecutiveFailures = 0;
_log.Info("Reconnected {0} on {1} endpoint", _connectionName, _activeEndpoint);
ReSubscribeAll();
BecomeConnected();
return;
}
_consecutiveFailures++;
_log.Warning("Reconnect attempt {0}/{1} failed for {2} on {3}: {4}",
_consecutiveFailures, _failoverRetryCount, _connectionName, _activeEndpoint, result.Error);
if (_consecutiveFailures >= _failoverRetryCount && _backupConfig != null)
{
// Switch endpoint
var previousEndpoint = _activeEndpoint;
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
? ActiveEndpoint.Backup
: ActiveEndpoint.Primary;
_consecutiveFailures = 0;
var newConfig = _activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig;
_log.Warning("Failing over {0} from {1} to {2}", _connectionName, previousEndpoint, _activeEndpoint);
// Dispose old adapter, create new one
_ = _adapter.DisposeAsync();
_adapter = _factory.Create(_protocolType, newConfig);
_connectionDetails = newConfig;
// Wire up disconnect handler on new adapter
_adapter.Disconnected += () => _self.Tell(new AdapterDisconnected());
}
// Schedule next retry
Context.System.Scheduler.ScheduleTellOnce(
_options.ReconnectInterval, Self, AttemptConnect.Instance, ActorRefs.NoSender);
}
```
### Step 4: Pass IDataConnectionFactory to DataConnectionActor
Update `DataConnectionManagerActor.HandleCreateConnection` to pass the factory:
```csharp
var props = Props.Create(() => new DataConnectionActor(
command.ConnectionName, adapter, _options, _healthCollector,
_factory, // pass factory for failover adapter creation
command.ProtocolType, command.PrimaryConnectionDetails,
command.BackupConnectionDetails, command.FailoverRetryCount));
```
And update the DataConnectionActor constructor to store `_factory`.
### Step 5: Build and run existing tests
Run: `dotnet build ScadaLink.slnx && dotnet test tests/ScadaLink.DataConnectionLayer.Tests`
Existing tests must pass (they use single-endpoint configs, so no failover triggered).
### Step 6: Commit
```bash
git add -A
git commit -m "feat(dcl): add failover state machine to DataConnectionActor with round-robin endpoint switching"
```
---
## Task 4: Failover Tests
**Files:**
- Modify: `tests/ScadaLink.DataConnectionLayer.Tests/DataConnectionActorTests.cs`
### Step 1: Write test — failover after N retries
```csharp
[Fact]
public async Task Reconnecting_AfterFailoverRetryCount_SwitchesToBackup()
{
// Arrange: create actor with primary + backup, failoverRetryCount = 2
var primaryAdapter = Substitute.For<IDataConnection>();
var backupAdapter = Substitute.For<IDataConnection>();
var factory = Substitute.For<IDataConnectionFactory>();
factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["endpoint"] == "backup"))
.Returns(backupAdapter);
// Primary connects then disconnects
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
primaryAdapter.Status.Returns(ConnectionHealth.Connected);
var primaryConfig = new Dictionary<string, string> { ["endpoint"] = "primary" };
var backupConfig = new Dictionary<string, string> { ["endpoint"] = "backup" };
// Create actor, connect on primary
// ... (use test kit patterns from existing tests)
// Simulate disconnect, verify 2 failures then factory.Create called with backup config
}
```
### Step 2: Write test — single endpoint retries forever
```csharp
[Fact]
public async Task Reconnecting_NoBackup_RetriesIndefinitely()
{
// Arrange: create actor with primary only, no backup
// Simulate 10 reconnect failures
// Verify: factory.Create never called with backup, just keeps retrying
}
```
### Step 3: Write test — round-robin back to primary after backup fails
```csharp
[Fact]
public async Task Reconnecting_BackupFails_SwitchesBackToPrimary()
{
// Arrange: primary + backup, failoverRetryCount = 1
// Simulate: primary fails 1x → switch to backup → backup fails 1x → switch to primary
// Verify: round-robin pattern
}
```
### Step 4: Write test — successful reconnect resets counter
```csharp
[Fact]
public async Task Reconnecting_SuccessfulConnect_ResetsConsecutiveFailures()
{
// Arrange: failoverRetryCount = 3
// Simulate: 2 failures on primary, then success
// Verify: no failover, counter reset
}
```
### Step 5: Write test — ReSubscribeAll called after failover
```csharp
[Fact]
public async Task Failover_ReSubscribesAllTagsOnNewAdapter()
{
// Arrange: actor with subscriptions, then failover
// Verify: new adapter receives SubscribeAsync calls for all previously subscribed tags
}
```
### Step 6: Run all tests
Run: `dotnet test tests/ScadaLink.DataConnectionLayer.Tests -v`
### Step 7: Commit
```bash
git add -A
git commit -m "test(dcl): add failover state machine tests for DataConnectionActor"
```
---
## Task 5: Health Reporting & Site Event Logging
**Files:**
- Modify: `src/ScadaLink.Commons/Messages/DataConnection/DataConnectionHealthReport.cs`
- Modify: `src/ScadaLink.DataConnectionLayer/Actors/DataConnectionActor.cs` (ReplyWithHealthReport, HandleReconnectResult)
### Step 1: Add ActiveEndpoint to health report
```csharp
public record DataConnectionHealthReport(
string ConnectionName,
ConnectionHealth Status,
int TotalSubscribedTags,
int ResolvedTags,
string ActiveEndpoint,
DateTimeOffset Timestamp);
```
### Step 2: Update ReplyWithHealthReport in DataConnectionActor
Update the health report method (around line 516) to include the active endpoint:
```csharp
private void ReplyWithHealthReport()
{
var endpointLabel = _backupConfig == null
? "Primary (no backup)"
: _activeEndpoint.ToString();
Sender.Tell(new DataConnectionHealthReport(
_connectionName, _adapter.Status,
_subscriptionsByInstance.Values.Sum(s => s.Count),
_resolvedTags,
endpointLabel,
DateTimeOffset.UtcNow));
}
```
### Step 3: Add site event logging on failover
In `HandleReconnectResult`, after switching endpoints, log a site event:
```csharp
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Warning", null, _connectionName,
$"Failover from {previousEndpoint} to {_activeEndpoint}",
$"After {_failoverRetryCount} consecutive failures");
}
```
Note: The actor needs `ISiteEventLogger` injected. Add it as an optional constructor parameter.
### Step 4: Add site event logging on successful reconnect after failover
In `HandleReconnectResult` success path, if the endpoint changed from last known good:
```csharp
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Info", null, _connectionName,
$"Connection restored on {_activeEndpoint} endpoint", null);
}
```
### Step 5: Build and test
Run: `dotnet build ScadaLink.slnx && dotnet test tests/ScadaLink.DataConnectionLayer.Tests`
### Step 6: Commit
```bash
git add -A
git commit -m "feat(dcl): add active endpoint to health reports and log failover events"
```
---
## Task 6: Central UI Changes
**Files:**
- Modify: `src/ScadaLink.CentralUI/Components/Pages/Admin/DataConnections.razor`
- Modify: `src/ScadaLink.CentralUI/Components/Pages/Admin/DataConnectionForm.razor`
### Step 1: Update DataConnections list page
Add `Active Endpoint` column to the table (around line 28-64). Insert after the Protocol column:
```html
<th>Active Endpoint</th>
```
And in the row template:
```html
<td>@connection.ActiveEndpoint</td>
```
This requires the list page to fetch health data alongside the connection list. Add a health status lookup or include `ActiveEndpoint` in the data connection response.
### Step 2: Update DataConnectionForm — rename Configuration label
Change the "Configuration" label to "Primary Endpoint Configuration" (around line 44-61).
### Step 3: Add backup endpoint section
Below the primary config field, add:
```html
@if (!_showBackup)
{
<button type="button" class="btn btn-outline-secondary btn-sm mt-2"
@onclick="() => _showBackup = true">
Add Backup Endpoint
</button>
}
else
{
<div class="mt-3">
<div class="d-flex justify-content-between align-items-center">
<label class="form-label">Backup Endpoint Configuration</label>
<button type="button" class="btn btn-outline-danger btn-sm"
@onclick="RemoveBackup">
Remove Backup
</button>
</div>
<textarea class="form-control" rows="4"
@bind="_model.BackupConfiguration"
placeholder='{"Host": "backup-host", "Port": 50101}' />
</div>
<div class="mt-3">
<label class="form-label">Failover Retry Count</label>
<input type="number" class="form-control" min="1" max="20"
@bind="_model.FailoverRetryCount" />
<small class="text-muted">Retries before switching to backup (default: 3)</small>
</div>
}
```
### Step 4: Update form model and save logic
Add `BackupConfiguration` and `FailoverRetryCount` to the form model. Update the save method to pass both configs to the management API.
In edit mode, set `_showBackup = true` if `BackupConfiguration` is not null.
### Step 5: Build and verify visually
Run: `dotnet build ScadaLink.slnx`
Visual verification requires running the cluster — document as manual test.
### Step 6: Commit
```bash
git add -A
git commit -m "feat(ui): add primary/backup endpoint fields to data connection form"
```
---
## Task 7: CLI, Management API, and Deployment
**Files:**
- Modify: `src/ScadaLink.Commons/Messages/Management/DataConnectionCommands.cs`
- Modify: `src/ScadaLink.CLI/Commands/DataConnectionCommands.cs`
- Modify: `src/ScadaLink.ManagementService/ManagementActor.cs` (lines 689-711)
- Modify: Deployment/flattening code that creates DataConnectionArtifact
### Step 1: Update management command messages
```csharp
public record CreateDataConnectionCommand(
int SiteId, string Name, string Protocol,
string? PrimaryConfiguration,
string? BackupConfiguration = null,
int FailoverRetryCount = 3);
public record UpdateDataConnectionCommand(
int DataConnectionId, string Name, string Protocol,
string? PrimaryConfiguration,
string? BackupConfiguration = null,
int FailoverRetryCount = 3);
```
### Step 2: Update ManagementActor handlers
In `HandleCreateDataConnection` (around line 689): set `PrimaryConfiguration`, `BackupConfiguration`, `FailoverRetryCount` from command.
In `HandleUpdateDataConnection` (around line 699): same fields.
### Step 3: Update CLI commands
In `BuildCreate` (around line 75-98):
- Rename `--configuration` to `--primary-config`
- Add hidden alias `--configuration` pointing to same option
- Add `--backup-config` option (optional)
- Add `--failover-retry-count` option (optional, default 3)
In `BuildUpdate` (around line 36-59): same changes.
In `BuildGet` (around line 22-34): update output to show both configs.
### Step 4: Update deployment artifact creation
Find where `DataConnectionArtifact` is constructed (in deployment/flattening code). Update to pass `PrimaryConfigurationJson` and `BackupConfigurationJson` from the entity.
### Step 5: Build and test CLI
Run: `dotnet build ScadaLink.slnx`
Test CLI manually:
```bash
scadalink data-connection create --site-id 1 --name "Test" --protocol OpcUa \
--primary-config '{"endpoint":"opc.tcp://localhost:50000"}' \
--backup-config '{"endpoint":"opc.tcp://localhost:50010"}' \
--failover-retry-count 3
```
### Step 6: Commit
```bash
git add -A
git commit -m "feat(cli): add --primary-config, --backup-config, --failover-retry-count to data connection commands"
```
---
## Task 8: Documentation Updates
**Files:**
- Modify: `docs/requirements/Component-DataConnectionLayer.md`
- Modify: `docs/requirements/HighLevelReqs.md`
- Modify: `docs/requirements/Component-CentralUI.md`
- Modify: `docs/test_infra/test_infra.md`
### Step 1: Update Component-DataConnectionLayer.md
Add new section "Endpoint Redundancy" covering:
- Optional backup endpoints
- Failover state machine (include ASCII diagram from design doc)
- Configuration model (PrimaryConfiguration + BackupConfiguration)
- Failover retry count and round-robin behavior
- Subscription re-creation on failover
- Health reporting (ActiveEndpoint field)
- Site event logging (DataConnectionFailover, DataConnectionRestored)
Update the configuration reference tables to show the new entity fields.
### Step 2: Update HighLevelReqs.md
Add requirement: "Data connections support optional backup endpoints with automatic failover after configurable retry count. On failover, all subscriptions are transparently re-created on the new endpoint."
### Step 3: Update Component-CentralUI.md
Update the Data Connections workflow section to describe:
- Primary/backup config fields on the form
- Collapsible backup section
- Failover retry count field
- Active endpoint column on list page
### Step 4: Update test_infra.md
Add a note in the Remote Test Infrastructure section that the dual OPC UA servers (50000/50010) and dual LmxProxy instances (50100/50101) enable primary/backup testing.
### Step 5: Commit
```bash
git add -A
git commit -m "docs(dcl): document primary/backup endpoint redundancy across requirements and test infra"
```
@@ -0,0 +1,14 @@
{
"planPath": "docs/plans/2026-03-22-primary-backup-data-connections.md",
"tasks": [
{"id": 1, "subject": "Task 1: Entity Model & Database Migration", "status": "pending"},
{"id": 2, "subject": "Task 2: Update CreateConnectionCommand & Manager Actor", "status": "pending", "blockedBy": [1]},
{"id": 3, "subject": "Task 3: DataConnectionActor Failover State Machine", "status": "pending", "blockedBy": [1, 2]},
{"id": 4, "subject": "Task 4: Failover Tests", "status": "pending", "blockedBy": [3]},
{"id": 5, "subject": "Task 5: Health Reporting & Site Event Logging", "status": "pending", "blockedBy": [3]},
{"id": 6, "subject": "Task 6: Central UI Changes", "status": "pending", "blockedBy": [1]},
{"id": 7, "subject": "Task 7: CLI, Management API, and Deployment", "status": "pending", "blockedBy": [1]},
{"id": 8, "subject": "Task 8: Documentation Updates", "status": "pending", "blockedBy": [3]}
],
"lastUpdated": "2026-03-22T12:00:00Z"
}
+2
View File
@@ -68,6 +68,8 @@ Central cluster only. Sites have no user interface.
### Site & Data Connection Management (Admin Role)
- Create, edit, and delete site definitions, including Akka node addresses (NodeA/NodeB) and gRPC node addresses (GrpcNodeA/GrpcNodeB).
- Define data connections and assign them to sites (name, protocol type, connection details).
- **Data connection form**: "Primary Endpoint Configuration" (required JSON text area) and optional "Backup Endpoint Configuration" (collapsible section, hidden by default, revealed via "Add Backup Endpoint" button; "Remove Backup" button when editing an existing backup). "Failover Retry Count" numeric input (default 3, min 1, max 20) is visible only when a backup endpoint is configured.
- **Data connection list page**: Shows Primary Config and Backup Config columns. Active Endpoint column populated from health reports.
### Area Management (Admin Role)
- Define hierarchical area structures per site.
@@ -104,9 +104,46 @@ LmxProxy is a gRPC-based protocol for communicating with LMX data servers. The D
**Test Infrastructure**: The `infra/lmxfakeproxy/` project provides a fake LmxProxy server that bridges to the OPC UA test server. It implements the full `scada.ScadaService` proto, enabling end-to-end testing of `RealLmxProxyClient` without a Windows LmxProxy deployment. See [test_infra_lmxfakeproxy.md](../test_infra/test_infra_lmxfakeproxy.md) for setup.
## Endpoint Redundancy
Data connections support an optional backup endpoint for automatic failover when the active endpoint becomes unreachable. Both endpoints use the same protocol.
**Entity fields:**
| Field | Type | Notes |
|-------|------|-------|
| `PrimaryConfiguration` | string? (max 4000) | Required. Renamed from `Configuration` |
| `BackupConfiguration` | string? (max 4000) | Optional. Null = no backup |
| `FailoverRetryCount` | int (default 3) | Retries on active endpoint before switching |
**Failover state machine:**
```
Connected → disconnect → push bad quality → retry active endpoint (5s)
→ N failures (≥ FailoverRetryCount) → switch to other endpoint
→ dispose adapter, create fresh adapter with other config
→ reconnect → ReSubscribeAll → Connected
```
- **Round-robin**: primary → backup → primary → backup. No preferred endpoint after first failover — the connection stays on whichever endpoint is working.
- **No auto-failback**: The connection remains on the active endpoint until it fails.
- **Single-endpoint connections** (no backup): Retry indefinitely on the same endpoint, preserving existing behavior.
- **Adapter lifecycle on failover**: The actor disposes the current `IDataConnection` adapter and creates a fresh one via `DataConnectionFactory.Create()` with the other endpoint's configuration. Clean slate — no stale state.
**Health reporting:**
- `DataConnectionHealthReport` includes `ActiveEndpoint`: `"Primary"`, `"Backup"`, or `"Primary (no backup)"`.
**Site event log entries:**
- `DataConnectionFailover` (Warning) — connection name, from-endpoint, to-endpoint, failure count.
- `DataConnectionRestored` (Info) — connection name, active endpoint.
See [`2026-03-22-primary-backup-data-connections-design.md`](../plans/2026-03-22-primary-backup-data-connections-design.md) for the full design.
## Connection Configuration Reference
All settings are parsed from the data connection's `Configuration` JSON dictionary (stored as `IDictionary<string, string>` connection details). Invalid numeric values fall back to defaults silently.
All settings are parsed from the data connection's configuration JSON dictionaries (`PrimaryConfiguration` and optional `BackupConfiguration`, stored as `IDictionary<string, string>` connection details). Both endpoints use the same protocol-specific keys. Invalid numeric values fall back to defaults silently.
### OPC UA Settings
+1
View File
@@ -65,6 +65,7 @@
- Additional protocols can be added by implementing the common interface.
- The Data Connection Layer is a **clean data pipe** — it publishes tag value updates to Instance Actors but performs no evaluation of triggers or alarm conditions.
- **Initial attribute quality**: Attributes bound to a data connection start with **uncertain** quality when the Instance Actor initializes. The quality remains uncertain until the first value update is received from the Data Connection Layer. This distinguishes "never received a value" from "received a known-good value" or "connection lost" (bad quality).
- Data connections support optional **backup endpoints** with automatic failover after a configurable retry count. On failover, all subscriptions are transparently re-created on the new endpoint.
### 2.5 Scale
- Approximately **10 sites**.
+2
View File
@@ -64,6 +64,8 @@ API key (ReadWrite): `c4559c7c6acc60a997135c1381162e3c30f4572ece78dd933c1a626e6f
Full details: [`lmxproxy/instances_config.md`](../../lmxproxy/instances_config.md)
**Primary/backup testing**: The dual OPC UA test servers (ports 50000 and 50010) in local Docker and the dual LmxProxy v2 instances on windev (ports 50100 and 50101) provide primary/backup endpoint pairs for testing Data Connection Layer failover. Use `docker compose stop opcua` to simulate primary failure and verify automatic failover to the backup.
## Connection Strings
For use in `appsettings.Development.json`:
@@ -0,0 +1,673 @@
# Gap 1 & Gap 2: Active Health Probing + Subscription Handle Cleanup
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.
**Goal:** Fix two reconnect-related gaps: (1) the monitor loop cannot detect a silently-dead MxAccess connection, and (2) SubscriptionManager holds stale IAsyncDisposable handles after reconnect.
**Architecture:** Add a domain-level connection probe to `MxAccessClient` that classifies results as Healthy/TransportFailure/DataDegraded. The monitor loop uses this to decide reconnect vs degrade-and-backoff. Separately, remove `SubscriptionManager._mxAccessHandles` entirely and switch to address-based unsubscribe through `IScadaClient`, making `MxAccessClient` the sole owner of COM subscription lifecycle.
**Tech Stack:** .NET Framework 4.8, C#, MxAccess COM interop, Serilog
---
## Task 0: Add `ProbeResult` domain type
**Files:**
- Create: `src/ZB.MOM.WW.LmxProxy.Host/Domain/ProbeResult.cs`
**Step 1: Create the ProbeResult type**
```csharp
using System;
namespace ZB.MOM.WW.LmxProxy.Host.Domain
{
public enum ProbeStatus
{
Healthy,
TransportFailure,
DataDegraded
}
public sealed class ProbeResult
{
public ProbeStatus Status { get; }
public Quality? Quality { get; }
public DateTime? Timestamp { get; }
public string? Message { get; }
public Exception? Exception { get; }
private ProbeResult(ProbeStatus status, Quality? quality, DateTime? timestamp,
string? message, Exception? exception)
{
Status = status;
Quality = quality;
Timestamp = timestamp;
Message = message;
Exception = exception;
}
public static ProbeResult Healthy(Quality quality, DateTime timestamp)
=> new ProbeResult(ProbeStatus.Healthy, quality, timestamp, null, null);
public static ProbeResult Degraded(Quality quality, DateTime timestamp, string message)
=> new ProbeResult(ProbeStatus.DataDegraded, quality, timestamp, message, null);
public static ProbeResult TransportFailed(string message, Exception? ex = null)
=> new ProbeResult(ProbeStatus.TransportFailure, null, null, message, ex);
}
}
```
**Step 2: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/Domain/ProbeResult.cs
git commit -m "feat: add ProbeResult domain type for connection health classification"
```
---
## Task 1: Add `ProbeConnectionAsync` to `MxAccessClient`
**Files:**
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/Domain/IScadaClient.cs` — add `ProbeConnectionAsync` to interface
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.Connection.cs` — implement probe method
**Step 1: Add to IScadaClient interface**
In `IScadaClient.cs`, add after the `DisconnectAsync` method:
```csharp
/// <summary>
/// Probes connection health by reading a test tag.
/// Returns a classified result: Healthy, TransportFailure, or DataDegraded.
/// </summary>
Task<ProbeResult> ProbeConnectionAsync(string testTagAddress, int timeoutMs, CancellationToken ct = default);
```
**Step 2: Implement in MxAccessClient.Connection.cs**
Add before `MonitorConnectionAsync`:
```csharp
/// <summary>
/// Probes the connection by reading a test tag with a timeout.
/// Classifies the result as transport failure vs data degraded.
/// </summary>
public async Task<ProbeResult> ProbeConnectionAsync(string testTagAddress, int timeoutMs,
CancellationToken ct = default)
{
if (!IsConnected)
return ProbeResult.TransportFailed("Not connected");
try
{
using (var cts = CancellationTokenSource.CreateLinkedTokenSource(ct))
{
cts.CancelAfter(timeoutMs);
Vtq vtq;
try
{
vtq = await ReadAsync(testTagAddress, cts.Token);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
// Our timeout fired, not the caller's — treat as transport failure
return ProbeResult.TransportFailed("Probe read timed out after " + timeoutMs + "ms");
}
if (vtq.Quality == Domain.Quality.Bad_NotConnected ||
vtq.Quality == Domain.Quality.Bad_CommFailure)
{
return ProbeResult.TransportFailed("Probe returned " + vtq.Quality);
}
if (!vtq.Quality.IsGood())
{
return ProbeResult.Degraded(vtq.Quality, vtq.Timestamp,
"Probe quality: " + vtq.Quality);
}
if (DateTime.UtcNow - vtq.Timestamp > TimeSpan.FromMinutes(5))
{
return ProbeResult.Degraded(vtq.Quality, vtq.Timestamp,
"Probe data stale (>" + 5 + "min)");
}
return ProbeResult.Healthy(vtq.Quality, vtq.Timestamp);
}
}
catch (System.Runtime.InteropServices.COMException ex)
{
return ProbeResult.TransportFailed("COM exception: " + ex.Message, ex);
}
catch (InvalidOperationException ex) when (ex.Message.Contains("Not connected"))
{
return ProbeResult.TransportFailed(ex.Message, ex);
}
catch (Exception ex)
{
return ProbeResult.TransportFailed("Probe failed: " + ex.Message, ex);
}
}
```
**Step 3: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/Domain/IScadaClient.cs
git add src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.Connection.cs
git commit -m "feat: add ProbeConnectionAsync to MxAccessClient for active health probing"
```
---
## Task 2: Add health check configuration
**Files:**
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/Configuration/LmxProxyConfiguration.cs` — add HealthCheckConfiguration class and property
**Step 1: Add HealthCheckConfiguration**
Add a new class in the Configuration namespace (can be in the same file or a new file — keep it simple, same file):
```csharp
/// <summary>Health check / probe configuration.</summary>
public class HealthCheckConfiguration
{
/// <summary>Tag address to probe for connection liveness. Default: TestChildObject.TestBool.</summary>
public string TestTagAddress { get; set; } = "TestChildObject.TestBool";
/// <summary>Probe timeout in milliseconds. Default: 5000.</summary>
public int ProbeTimeoutMs { get; set; } = 5000;
/// <summary>Consecutive transport failures before forced reconnect. Default: 3.</summary>
public int MaxConsecutiveTransportFailures { get; set; } = 3;
/// <summary>Probe interval while in degraded state (ms). Default: 30000 (30s).</summary>
public int DegradedProbeIntervalMs { get; set; } = 30000;
}
```
Add to `LmxProxyConfiguration`:
```csharp
/// <summary>Health check / active probe settings.</summary>
public HealthCheckConfiguration HealthCheck { get; set; } = new HealthCheckConfiguration();
```
**Step 2: Add to appsettings.json**
In the existing `appsettings.json`, add the `HealthCheck` section:
```json
"HealthCheck": {
"TestTagAddress": "TestChildObject.TestBool",
"ProbeTimeoutMs": 5000,
"MaxConsecutiveTransportFailures": 3,
"DegradedProbeIntervalMs": 30000
}
```
**Step 3: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/Configuration/LmxProxyConfiguration.cs
git add src/ZB.MOM.WW.LmxProxy.Host/appsettings.json
git commit -m "feat: add HealthCheck configuration section for active connection probing"
```
---
## Task 3: Rewrite `MonitorConnectionAsync` with active probing
**Files:**
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.cs` — add probe state fields
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.Connection.cs` — rewrite monitor loop
The monitor needs configuration passed in. The simplest approach: add constructor parameters for the probe settings alongside the existing ones.
**Step 1: Add probe fields to MxAccessClient.cs**
Add fields after the existing reconnect fields (around line 42):
```csharp
// Probe configuration
private readonly string? _probeTestTagAddress;
private readonly int _probeTimeoutMs;
private readonly int _maxConsecutiveTransportFailures;
private readonly int _degradedProbeIntervalMs;
// Probe state
private int _consecutiveTransportFailures;
private bool _isDegraded;
```
Add constructor parameters and assignments. After the existing `_galaxyName = galaxyName;` line:
```csharp
public MxAccessClient(
int maxConcurrentOperations = 10,
int readTimeoutSeconds = 5,
int writeTimeoutSeconds = 5,
int monitorIntervalSeconds = 5,
bool autoReconnect = true,
string? nodeName = null,
string? galaxyName = null,
string? probeTestTagAddress = null,
int probeTimeoutMs = 5000,
int maxConsecutiveTransportFailures = 3,
int degradedProbeIntervalMs = 30000)
```
And in the body:
```csharp
_probeTestTagAddress = probeTestTagAddress;
_probeTimeoutMs = probeTimeoutMs;
_maxConsecutiveTransportFailures = maxConsecutiveTransportFailures;
_degradedProbeIntervalMs = degradedProbeIntervalMs;
```
**Step 2: Rewrite MonitorConnectionAsync in MxAccessClient.Connection.cs**
Replace the existing `MonitorConnectionAsync` (lines 177-213) with:
```csharp
/// <summary>
/// Auto-reconnect monitor loop with active health probing.
/// - If IsConnected is false: immediate reconnect (existing behavior).
/// - If IsConnected is true and probe configured: read test tag each interval.
/// - TransportFailure for N consecutive probes → forced disconnect + reconnect.
/// - DataDegraded → stay connected, back off probe interval, report degraded.
/// - Healthy → reset counters and resume normal interval.
/// </summary>
private async Task MonitorConnectionAsync(CancellationToken ct)
{
Log.Information("Connection monitor loop started (interval={IntervalMs}ms, probe={ProbeEnabled})",
_monitorIntervalMs, _probeTestTagAddress != null);
while (!ct.IsCancellationRequested)
{
var interval = _isDegraded ? _degradedProbeIntervalMs : _monitorIntervalMs;
try
{
await Task.Delay(interval, ct);
}
catch (OperationCanceledException)
{
break;
}
// ── Case 1: Already disconnected ──
if (!IsConnected)
{
_isDegraded = false;
_consecutiveTransportFailures = 0;
await AttemptReconnectAsync(ct);
continue;
}
// ── Case 2: Connected, no probe configured — legacy behavior ──
if (_probeTestTagAddress == null)
continue;
// ── Case 3: Connected, probe configured — active health check ──
var probe = await ProbeConnectionAsync(_probeTestTagAddress, _probeTimeoutMs, ct);
switch (probe.Status)
{
case ProbeStatus.Healthy:
if (_isDegraded)
{
Log.Information("Probe healthy — exiting degraded mode");
_isDegraded = false;
}
_consecutiveTransportFailures = 0;
break;
case ProbeStatus.DataDegraded:
_consecutiveTransportFailures = 0;
if (!_isDegraded)
{
Log.Warning("Probe degraded: {Message} — entering degraded mode (probe interval {IntervalMs}ms)",
probe.Message, _degradedProbeIntervalMs);
_isDegraded = true;
}
break;
case ProbeStatus.TransportFailure:
_isDegraded = false;
_consecutiveTransportFailures++;
Log.Warning("Probe transport failure ({Count}/{Max}): {Message}",
_consecutiveTransportFailures, _maxConsecutiveTransportFailures, probe.Message);
if (_consecutiveTransportFailures >= _maxConsecutiveTransportFailures)
{
Log.Warning("Max consecutive transport failures reached — forcing reconnect");
_consecutiveTransportFailures = 0;
try
{
await DisconnectAsync(ct);
}
catch (Exception ex)
{
Log.Warning(ex, "Error during forced disconnect before reconnect");
// DisconnectAsync already calls CleanupComObjectsAsync on error path
}
await AttemptReconnectAsync(ct);
}
break;
}
}
Log.Information("Connection monitor loop exited");
}
private async Task AttemptReconnectAsync(CancellationToken ct)
{
Log.Information("Attempting reconnect...");
SetState(ConnectionState.Reconnecting);
try
{
await ConnectAsync(ct);
Log.Information("Reconnected to MxAccess successfully");
}
catch (OperationCanceledException)
{
// Let the outer loop handle cancellation
}
catch (Exception ex)
{
Log.Warning(ex, "Reconnect attempt failed, will retry at next interval");
}
}
```
**Step 3: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.cs
git add src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.Connection.cs
git commit -m "feat: rewrite monitor loop with active probing, transport vs degraded classification"
```
---
## Task 4: Wire probe config through `LmxProxyService.Start()`
**Files:**
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/LmxProxyService.cs` — pass HealthCheck config to MxAccessClient constructor
**Step 1: Update MxAccessClient construction**
In `LmxProxyService.Start()`, update the MxAccessClient creation (around line 62) to pass the new parameters:
```csharp
_mxAccessClient = new MxAccessClient(
maxConcurrentOperations: _config.Connection.MaxConcurrentOperations,
readTimeoutSeconds: _config.Connection.ReadTimeoutSeconds,
writeTimeoutSeconds: _config.Connection.WriteTimeoutSeconds,
monitorIntervalSeconds: _config.Connection.MonitorIntervalSeconds,
autoReconnect: _config.Connection.AutoReconnect,
nodeName: _config.Connection.NodeName,
galaxyName: _config.Connection.GalaxyName,
probeTestTagAddress: _config.HealthCheck.TestTagAddress,
probeTimeoutMs: _config.HealthCheck.ProbeTimeoutMs,
maxConsecutiveTransportFailures: _config.HealthCheck.MaxConsecutiveTransportFailures,
degradedProbeIntervalMs: _config.HealthCheck.DegradedProbeIntervalMs);
```
**Step 2: Update DetailedHealthCheckService to use shared probe**
In `LmxProxyService.Start()`, update the DetailedHealthCheckService construction (around line 114) to pass the test tag address from config:
```csharp
_detailedHealthCheckService = new DetailedHealthCheckService(
_mxAccessClient, _config.HealthCheck.TestTagAddress);
```
**Step 3: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/LmxProxyService.cs
git commit -m "feat: wire HealthCheck config to MxAccessClient and DetailedHealthCheckService"
```
---
## Task 5: Add `UnsubscribeByAddressAsync` to `IScadaClient` and `MxAccessClient`
This is the foundation for removing handle-based unsubscribe from SubscriptionManager.
**Files:**
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/Domain/IScadaClient.cs` — add `UnsubscribeByAddressAsync`
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.Subscription.cs` — implement, change `UnsubscribeAsync` visibility
**Step 1: Add to IScadaClient**
After `SubscribeAsync`:
```csharp
/// <summary>
/// Unsubscribes specific tag addresses. Removes from stored subscriptions
/// and COM state. Safe to call after reconnect — uses current handle mappings.
/// </summary>
Task UnsubscribeByAddressAsync(IEnumerable<string> addresses);
```
**Step 2: Implement in MxAccessClient.Subscription.cs**
The existing `UnsubscribeAsync` (line 53) already does exactly this — it's just `internal`. Rename it or add a public wrapper:
```csharp
/// <summary>
/// Unsubscribes specific addresses by address name.
/// Removes from both COM state and stored subscriptions (no reconnect replay).
/// </summary>
public async Task UnsubscribeByAddressAsync(IEnumerable<string> addresses)
{
await UnsubscribeAsync(addresses);
}
```
This keeps the existing `internal UnsubscribeAsync` unchanged (it's still used by `SubscriptionHandle.DisposeAsync`).
**Step 3: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/Domain/IScadaClient.cs
git add src/ZB.MOM.WW.LmxProxy.Host/MxAccess/MxAccessClient.Subscription.cs
git commit -m "feat: add UnsubscribeByAddressAsync to IScadaClient for address-based unsubscribe"
```
---
## Task 6: Remove `_mxAccessHandles` from `SubscriptionManager`
**Files:**
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/Subscriptions/SubscriptionManager.cs`
**Step 1: Remove `_mxAccessHandles` field**
Delete line 34-35:
```csharp
// REMOVE:
private readonly ConcurrentDictionary<string, IAsyncDisposable> _mxAccessHandles
= new ConcurrentDictionary<string, IAsyncDisposable>(StringComparer.OrdinalIgnoreCase);
```
**Step 2: Rewrite `CreateMxAccessSubscriptionsAsync`**
The method no longer stores handles. It just calls `SubscribeAsync` to create the COM subscriptions. `MxAccessClient` stores them in `_storedSubscriptions` internally.
```csharp
private async Task CreateMxAccessSubscriptionsAsync(List<string> addresses)
{
try
{
await _scadaClient.SubscribeAsync(
addresses,
(address, vtq) => OnTagValueChanged(address, vtq));
}
catch (Exception ex)
{
Log.Error(ex, "Failed to create MxAccess subscriptions for {Count} tags", addresses.Count);
}
}
```
**Step 3: Rewrite unsubscribe logic in `UnsubscribeClient`**
Replace the handle disposal section (lines 198-212) with address-based unsubscribe:
```csharp
// Unsubscribe tags with no remaining clients via address-based API
if (tagsToDispose.Count > 0)
{
try
{
_scadaClient.UnsubscribeByAddressAsync(tagsToDispose).GetAwaiter().GetResult();
}
catch (Exception ex)
{
Log.Warning(ex, "Error unsubscribing {Count} tags from MxAccess", tagsToDispose.Count);
}
}
```
**Step 4: Verify build**
```bash
dotnet build src/ZB.MOM.WW.LmxProxy.Host
```
Expected: Build succeeds. No references to `_mxAccessHandles` remain.
**Step 5: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/Subscriptions/SubscriptionManager.cs
git commit -m "fix: remove _mxAccessHandles from SubscriptionManager, use address-based unsubscribe"
```
---
## Task 7: Wire `ConnectionStateChanged` for reconnect notification in `SubscriptionManager`
After reconnect, `RecreateStoredSubscriptionsAsync` recreates COM subscriptions, and `SubscriptionManager` continues to receive `OnTagValueChanged` callbacks because the callback references are preserved in `_storedSubscriptions`. However, we should notify subscribed clients that quality has been restored.
**Files:**
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/Subscriptions/SubscriptionManager.cs` — add `NotifyReconnection` method
- Modify: `src/ZB.MOM.WW.LmxProxy.Host/LmxProxyService.cs` — wire Connected state to SubscriptionManager
**Step 1: Add `NotifyReconnection` to SubscriptionManager**
```csharp
/// <summary>
/// Logs reconnection for observability. Data flow resumes automatically
/// via MxAccessClient.RecreateStoredSubscriptionsAsync callbacks.
/// </summary>
public void NotifyReconnection()
{
Log.Information("MxAccess reconnected — subscriptions recreated, " +
"data flow will resume via OnDataChange callbacks " +
"({ClientCount} clients, {TagCount} tags)",
_clientSubscriptions.Count, _tagSubscriptions.Count);
}
```
**Step 2: Wire in LmxProxyService.Start()**
Extend the existing `ConnectionStateChanged` handler (around line 97):
```csharp
_mxAccessClient.ConnectionStateChanged += (sender, e) =>
{
if (e.CurrentState == Domain.ConnectionState.Disconnected ||
e.CurrentState == Domain.ConnectionState.Error)
{
_subscriptionManager.NotifyDisconnection();
}
else if (e.CurrentState == Domain.ConnectionState.Connected &&
e.PreviousState == Domain.ConnectionState.Reconnecting)
{
_subscriptionManager.NotifyReconnection();
}
};
```
**Step 3: Commit**
```bash
git add src/ZB.MOM.WW.LmxProxy.Host/Subscriptions/SubscriptionManager.cs
git add src/ZB.MOM.WW.LmxProxy.Host/LmxProxyService.cs
git commit -m "feat: wire reconnection notification to SubscriptionManager for observability"
```
---
## Task 8: Build, deploy to windev, test
**Files:**
- No code changes — build and deployment task.
**Step 1: Build the solution**
```bash
dotnet build ZB.MOM.WW.LmxProxy.slnx
```
Expected: Clean build, no errors.
**Step 2: Deploy to windev**
Follow existing deployment procedure per `docker/README.md` or manual copy to windev.
**Step 3: Manual test — Gap 1 (active probing)**
1. Start the v2 service on windev. Verify logs show: `Connection monitor loop started (interval=5000ms, probe=True)`.
2. Verify probe runs: logs should show no warnings while platform is healthy.
3. Kill aaBootstrap on windev. Within 15-20s (3 probe failures at 5s intervals), logs should show:
- `Probe transport failure (1/3): Probe returned Bad_CommFailure` (or similar)
- `Probe transport failure (2/3): ...`
- `Probe transport failure (3/3): ...`
- `Max consecutive transport failures reached — forcing reconnect`
- `Attempting reconnect...`
4. After platform restart (but objects still stopped): Logs should show `Probe degraded` and `entering degraded mode`, then probe backs off to 30s interval. No reconnect churn.
5. After objects restart via SMC: Logs should show `Probe healthy — exiting degraded mode`.
**Step 4: Manual test — Gap 2 (subscription cleanup)**
1. Connect a gRPC client, subscribe to tags.
2. Kill aaBootstrap → client receives `Bad_NotConnected` quality.
3. Restart platform + objects. Verify client starts receiving Good quality updates again (via `RecreateStoredSubscriptionsAsync`).
4. Disconnect the client. Verify logs show `Unsubscribed from N tags` (address-based) with no handle disposal errors.
---
## Design Rationale
### Why two failure modes in the probe?
| Failure Mode | Cause | Correct Response |
|---|---|---|
| **Transport failure** | COM object dead, platform process crashed, MxAccess unreachable | Force disconnect + reconnect |
| **Data degraded** | COM session alive, AVEVA objects stopped, all reads return Bad quality | Stay connected, report degraded, back off probes |
Reconnecting on DataDegraded would churn COM objects with no benefit — the platform objects are stopped regardless of connection state. Observed: 40+ minutes of Bad quality after aaBootstrap crash until manual SMC restart.
### Why remove `_mxAccessHandles`?
1. **Batch handle bug**: `CreateMxAccessSubscriptionsAsync` stored the same `IAsyncDisposable` handle for every address in a batch. Disposing any one address disposed the entire batch, silently removing unrelated subscriptions from `_storedSubscriptions`.
2. **Stale after reconnect**: `RecreateStoredSubscriptionsAsync` recreates COM subscriptions but doesn't produce new `SubscriptionManager` handles. Old handles point to disposed COM state.
3. **Ownership violation**: `MxAccessClient` already owns subscription lifecycle via `_storedSubscriptions` and `_addressToHandle`. Duplicating ownership in `SubscriptionManager._mxAccessHandles` is a leaky abstraction.
The fix: `SubscriptionManager` owns client routing and ref counts only. `MxAccessClient` owns COM subscription lifecycle. Unsubscribe is by address, not by opaque handle.
@@ -0,0 +1,15 @@
{
"planPath": "lmxproxy/docs/plans/2026-03-22-gap1-gap2-reconnect-subscriptions.md",
"tasks": [
{"id": 0, "subject": "Task 0: Add ProbeResult domain type", "status": "pending"},
{"id": 1, "subject": "Task 1: Add ProbeConnectionAsync to MxAccessClient", "status": "pending", "blockedBy": [0]},
{"id": 2, "subject": "Task 2: Add health check configuration", "status": "pending"},
{"id": 3, "subject": "Task 3: Rewrite MonitorConnectionAsync with active probing", "status": "pending", "blockedBy": [1, 2]},
{"id": 4, "subject": "Task 4: Wire probe config through LmxProxyService.Start()", "status": "pending", "blockedBy": [2, 3]},
{"id": 5, "subject": "Task 5: Add UnsubscribeByAddressAsync to IScadaClient", "status": "pending"},
{"id": 6, "subject": "Task 6: Remove _mxAccessHandles from SubscriptionManager", "status": "pending", "blockedBy": [5]},
{"id": 7, "subject": "Task 7: Wire ConnectionStateChanged for reconnect notification", "status": "pending", "blockedBy": [6]},
{"id": 8, "subject": "Task 8: Build, deploy to windev, test", "status": "pending", "blockedBy": [4, 7]}
],
"lastUpdated": "2026-03-22T00:00:00Z"
}
@@ -38,22 +38,28 @@ public static class DataConnectionCommands
var idOption = new Option<int>("--id") { Description = "Data connection ID", Required = true };
var nameOption = new Option<string>("--name") { Description = "Connection name", Required = true };
var protocolOption = new Option<string>("--protocol") { Description = "Protocol", Required = true };
var configOption = new Option<string?>("--configuration") { Description = "Configuration JSON" };
var configOption = new Option<string?>("--primary-config", "--configuration") { Description = "Primary configuration JSON" };
var backupConfigOption = new Option<string?>("--backup-config") { Description = "Backup configuration JSON" };
var failoverRetryOption = new Option<int>("--failover-retry-count") { Description = "Number of retries before failover to backup", DefaultValueFactory = _ => 3 };
var cmd = new Command("update") { Description = "Update a data connection" };
cmd.Add(idOption);
cmd.Add(nameOption);
cmd.Add(protocolOption);
cmd.Add(configOption);
cmd.Add(backupConfigOption);
cmd.Add(failoverRetryOption);
cmd.SetAction(async (ParseResult result) =>
{
var id = result.GetValue(idOption);
var name = result.GetValue(nameOption)!;
var protocol = result.GetValue(protocolOption)!;
var config = result.GetValue(configOption);
var backupConfig = result.GetValue(backupConfigOption);
var failoverRetryCount = result.GetValue(failoverRetryOption);
return await CommandHelpers.ExecuteCommandAsync(
result, urlOption, formatOption, usernameOption, passwordOption,
new UpdateDataConnectionCommand(id, name, protocol, config));
new UpdateDataConnectionCommand(id, name, protocol, config, backupConfig, failoverRetryCount));
});
return cmd;
}
@@ -77,22 +83,28 @@ public static class DataConnectionCommands
var siteIdOption = new Option<int>("--site-id") { Description = "Site ID", Required = true };
var nameOption = new Option<string>("--name") { Description = "Connection name", Required = true };
var protocolOption = new Option<string>("--protocol") { Description = "Protocol (e.g. OpcUa)", Required = true };
var configOption = new Option<string?>("--configuration") { Description = "Connection configuration JSON" };
var configOption = new Option<string?>("--primary-config", "--configuration") { Description = "Primary configuration JSON" };
var backupConfigOption = new Option<string?>("--backup-config") { Description = "Backup configuration JSON" };
var failoverRetryOption = new Option<int>("--failover-retry-count") { Description = "Number of retries before failover to backup", DefaultValueFactory = _ => 3 };
var cmd = new Command("create") { Description = "Create a new data connection" };
cmd.Add(siteIdOption);
cmd.Add(nameOption);
cmd.Add(protocolOption);
cmd.Add(configOption);
cmd.Add(backupConfigOption);
cmd.Add(failoverRetryOption);
cmd.SetAction(async (ParseResult result) =>
{
var siteId = result.GetValue(siteIdOption);
var name = result.GetValue(nameOption)!;
var protocol = result.GetValue(protocolOption)!;
var config = result.GetValue(configOption);
var backupConfig = result.GetValue(backupConfigOption);
var failoverRetryCount = result.GetValue(failoverRetryOption);
return await CommandHelpers.ExecuteCommandAsync(
result, urlOption, formatOption, usernameOption, passwordOption,
new CreateDataConnectionCommand(siteId, name, protocol, config));
new CreateDataConnectionCommand(siteId, name, protocol, config, backupConfig, failoverRetryCount));
});
return cmd;
}
@@ -51,10 +51,10 @@ public static class InstanceCommands
{
var id = result.GetValue(idOption);
var bindingsJson = result.GetValue(bindingsOption)!;
var pairs = System.Text.Json.JsonSerializer.Deserialize<List<List<object>>>(bindingsJson)
var pairs = System.Text.Json.JsonSerializer.Deserialize<List<List<System.Text.Json.JsonElement>>>(bindingsJson)
?? throw new InvalidOperationException("Invalid bindings JSON");
var bindings = pairs.Select(p =>
(p[0].ToString()!, int.Parse(p[1].ToString()!))).ToList();
(p[0].GetString()!, p[1].GetInt32())).ToList();
return await CommandHelpers.ExecuteCommandAsync(
result, urlOption, formatOption, usernameOption, passwordOption,
new SetConnectionBindingsCommand(id, bindings));
@@ -55,10 +55,43 @@
</select>
</div>
<div class="mb-2">
<label class="form-label small">Configuration (JSON)</label>
<label class="form-label small">Primary Endpoint Configuration</label>
<input type="text" class="form-control form-control-sm" @bind="_formConfiguration"
placeholder='e.g. {"endpoint":"opc.tcp://..."}' />
</div>
@if (!_showBackup)
{
<div class="mb-3">
<button type="button" class="btn btn-outline-secondary btn-sm"
@onclick="() => _showBackup = true">
Add Backup Endpoint
</button>
</div>
}
else
{
<div class="mb-3">
<div class="d-flex justify-content-between align-items-center mb-1">
<label class="form-label small mb-0">Backup Endpoint Configuration</label>
<button type="button" class="btn btn-outline-danger btn-sm"
@onclick="RemoveBackup">
Remove Backup
</button>
</div>
<textarea class="form-control form-control-sm" rows="4"
@bind="_formBackupConfiguration"
placeholder='{"Host": "backup-host", "Port": 50101}' />
</div>
<div class="mb-3">
<label class="form-label small">Failover Retry Count</label>
<input type="number" class="form-control form-control-sm" style="max-width: 120px;"
min="1" max="20"
@bind="_formFailoverRetryCount" />
<div class="form-text">Retries on active endpoint before switching to backup (default: 3)</div>
</div>
}
@if (_formError != null)
{
<div class="text-danger small mt-2">@_formError</div>
@@ -83,6 +116,9 @@
private string _formName = string.Empty;
private string _formProtocol = string.Empty;
private string? _formConfiguration;
private bool _showBackup;
private string? _formBackupConfiguration;
private int _formFailoverRetryCount = 3;
private string? _formError;
protected override async Task OnInitializedAsync()
@@ -100,7 +136,10 @@
_siteName = _sites.FirstOrDefault(s => s.Id == _formSiteId)?.Name ?? $"Site {_formSiteId}";
_formName = _editingConnection.Name;
_formProtocol = _editingConnection.Protocol;
_formConfiguration = _editingConnection.Configuration;
_formConfiguration = _editingConnection.PrimaryConfiguration;
_formBackupConfiguration = _editingConnection.BackupConfiguration;
_formFailoverRetryCount = _editingConnection.FailoverRetryCount;
_showBackup = _editingConnection.BackupConfiguration != null;
}
}
catch (Exception ex)
@@ -124,14 +163,18 @@
{
_editingConnection.Name = _formName.Trim();
_editingConnection.Protocol = _formProtocol;
_editingConnection.Configuration = _formConfiguration?.Trim();
_editingConnection.PrimaryConfiguration = _formConfiguration?.Trim();
_editingConnection.BackupConfiguration = _showBackup ? _formBackupConfiguration?.Trim() : null;
_editingConnection.FailoverRetryCount = _showBackup ? _formFailoverRetryCount : 3;
await SiteRepository.UpdateDataConnectionAsync(_editingConnection);
}
else
{
var conn = new DataConnection(_formName.Trim(), _formProtocol, _formSiteId)
{
Configuration = _formConfiguration?.Trim()
PrimaryConfiguration = _formConfiguration?.Trim(),
BackupConfiguration = _showBackup ? _formBackupConfiguration?.Trim() : null,
FailoverRetryCount = _showBackup ? _formFailoverRetryCount : 3
};
await SiteRepository.AddDataConnectionAsync(conn);
}
@@ -144,6 +187,13 @@
}
}
private void RemoveBackup()
{
_showBackup = false;
_formBackupConfiguration = null;
_formFailoverRetryCount = 3;
}
private void GoBack()
{
NavigationManager.NavigateTo("/admin/data-connections");
@@ -32,7 +32,8 @@
<th>Name</th>
<th>Protocol</th>
<th>Site</th>
<th>Configuration</th>
<th>Primary Config</th>
<th>Backup Config</th>
<th style="width: 160px;">Actions</th>
</tr>
</thead>
@@ -40,7 +41,7 @@
@if (_connections.Count == 0)
{
<tr>
<td colspan="6" class="text-muted text-center">No data connections configured.</td>
<td colspan="7" class="text-muted text-center">No data connections configured.</td>
</tr>
}
@foreach (var conn in _connections)
@@ -50,7 +51,8 @@
<td>@conn.Name</td>
<td><span class="badge bg-secondary">@conn.Protocol</span></td>
<td>@(_siteLookup.GetValueOrDefault(conn.SiteId)?.Name ?? $"Site {conn.SiteId}")</td>
<td class="text-muted small text-truncate" style="max-width: 300px;">@(conn.Configuration ?? "—")</td>
<td class="text-muted small text-truncate" style="max-width: 300px;">@(conn.PrimaryConfiguration ?? "—")</td>
<td class="text-muted small text-truncate" style="max-width: 300px;">@(conn.BackupConfiguration ?? "—")</td>
<td>
<button class="btn btn-outline-primary btn-sm py-0 px-1 me-1"
@onclick='() => NavigationManager.NavigateTo($"/admin/data-connections/{conn.Id}/edit")'>Edit</button>
@@ -5,6 +5,7 @@
@using ScadaLink.Commons.Interfaces.Repositories
@using ScadaLink.Commons.Messages.DebugView
@using ScadaLink.Commons.Messages.Streaming
@using ScadaLink.Commons.Types
@using ScadaLink.Commons.Types.Enums
@using ScadaLink.Communication
@attribute [Authorize(Policy = AuthorizationPolicies.RequireDeployment)]
@@ -91,7 +92,7 @@
{
<tr>
<td class="small">@av.AttributeName</td>
<td class="small font-monospace"><strong>@av.Value</strong></td>
<td class="small font-monospace"><strong>@ValueFormatter.FormatDisplayValue(av.Value)</strong></td>
<td>
<span class="badge @(av.Quality == "Good" ? "bg-success" : "bg-warning text-dark")">@av.Quality</span>
</td>
@@ -6,7 +6,9 @@ public class DataConnection
public int SiteId { get; set; }
public string Name { get; set; }
public string Protocol { get; set; }
public string? Configuration { get; set; }
public string? PrimaryConfiguration { get; set; }
public string? BackupConfiguration { get; set; }
public int FailoverRetryCount { get; set; } = 3;
public DataConnection(string name, string protocol, int siteId)
{
@@ -3,4 +3,6 @@ namespace ScadaLink.Commons.Messages.Artifacts;
public record DataConnectionArtifact(
string Name,
string Protocol,
string? ConfigurationJson);
string? PrimaryConfigurationJson,
string? BackupConfigurationJson,
int FailoverRetryCount = 3);
@@ -7,4 +7,6 @@ namespace ScadaLink.Commons.Messages.DataConnection;
public record CreateConnectionCommand(
string ConnectionName,
string ProtocolType,
IDictionary<string, string> ConnectionDetails);
IDictionary<string, string> PrimaryConnectionDetails,
IDictionary<string, string>? BackupConnectionDetails = null,
int FailoverRetryCount = 3);
@@ -10,4 +10,5 @@ public record DataConnectionHealthReport(
ConnectionHealth Status,
int TotalSubscribedTags,
int ResolvedTags,
string ActiveEndpoint,
DateTimeOffset Timestamp);
@@ -2,6 +2,6 @@ namespace ScadaLink.Commons.Messages.Management;
public record ListDataConnectionsCommand(int? SiteId = null);
public record GetDataConnectionCommand(int DataConnectionId);
public record CreateDataConnectionCommand(int SiteId, string Name, string Protocol, string? Configuration);
public record UpdateDataConnectionCommand(int DataConnectionId, string Name, string Protocol, string? Configuration);
public record CreateDataConnectionCommand(int SiteId, string Name, string Protocol, string? PrimaryConfiguration, string? BackupConfiguration = null, int FailoverRetryCount = 3);
public record UpdateDataConnectionCommand(int DataConnectionId, string Name, string Protocol, string? PrimaryConfiguration, string? BackupConfiguration = null, int FailoverRetryCount = 3);
public record DeleteDataConnectionCommand(int DataConnectionId);
@@ -33,6 +33,8 @@ public sealed record ConnectionConfig
{
public string Protocol { get; init; } = string.Empty;
public string? ConfigurationJson { get; init; }
public string? BackupConfigurationJson { get; init; }
public int FailoverRetryCount { get; init; } = 3;
}
/// <summary>
@@ -0,0 +1,59 @@
using System.Collections;
using System.Reflection;
namespace ScadaLink.Commons.Types;
/// <summary>
/// Formats attribute values for display. Handles scalar types directly
/// and uses reflection to extract array contents from complex types
/// (e.g., LmxProxy ArrayValue) rather than showing the type name.
/// </summary>
public static class ValueFormatter
{
/// <summary>
/// Formats a value for display as a string. Returns the value's natural
/// string representation for scalars, and comma-separated elements for
/// array/collection types.
/// </summary>
public static string FormatDisplayValue(object? value)
{
if (value is null) return "";
if (value is string s) return s;
if (value is IFormattable) return value.ToString() ?? "";
// Check if it's an array-like container with typed sub-collections
// (e.g., LmxProxy ArrayValue with BoolValues, Int32Values, etc.)
var type = value.GetType();
if (type.Namespace?.Contains("LmxProxy") == true || type.Name == "ArrayValue")
{
return FormatArrayContainer(value, type);
}
// Fallback for IEnumerable (generic collections, arrays)
if (value is IEnumerable enumerable)
{
return string.Join(",", enumerable.Cast<object?>().Select(e => e?.ToString() ?? ""));
}
return value.ToString() ?? "";
}
private static string FormatArrayContainer(object container, Type type)
{
// Look for the first non-null property that has a Values list
foreach (var prop in type.GetProperties(BindingFlags.Public | BindingFlags.Instance))
{
var propValue = prop.GetValue(container);
if (propValue is null) continue;
// Check if this property has a Values sub-property (e.g., BoolArray.Values)
var valuesProp = propValue.GetType().GetProperty("Values");
if (valuesProp?.GetValue(propValue) is IEnumerable values)
{
return string.Join(",", values.Cast<object?>().Select(e => e?.ToString() ?? ""));
}
}
return "";
}
}
@@ -3,6 +3,7 @@ using Akka.Actor;
using Akka.Event;
using Google.Protobuf.WellKnownTypes;
using ScadaLink.Commons.Messages.Streaming;
using ScadaLink.Commons.Types;
using ScadaLink.Communication.Grpc;
using AlarmState = ScadaLink.Commons.Types.Enums.AlarmState;
@@ -38,7 +39,7 @@ public class StreamRelayActor : ReceiveActor
InstanceUniqueName = msg.InstanceUniqueName,
AttributePath = msg.AttributePath,
AttributeName = msg.AttributeName,
Value = msg.Value?.ToString() ?? "",
Value = ValueFormatter.FormatDisplayValue(msg.Value),
Quality = MapQuality(msg.Quality),
Timestamp = Timestamp.FromDateTimeOffset(msg.Timestamp)
}
@@ -43,9 +43,16 @@ public class DataConnectionConfiguration : IEntityTypeConfiguration<DataConnecti
.IsRequired()
.HasMaxLength(50);
builder.Property(d => d.Configuration)
builder.Property(d => d.PrimaryConfiguration)
.HasMaxLength(4000);
builder.Property(d => d.BackupConfiguration)
.HasMaxLength(4000);
builder.Property(d => d.FailoverRetryCount)
.IsRequired()
.HasDefaultValue(3);
builder.HasOne<Site>()
.WithMany()
.HasForeignKey(d => d.SiteId)
@@ -4,6 +4,7 @@ using ScadaLink.Commons.Interfaces.Protocol;
using ScadaLink.Commons.Messages.DataConnection;
using ScadaLink.Commons.Types.Enums;
using ScadaLink.HealthMonitoring;
using ScadaLink.SiteEventLogging;
namespace ScadaLink.DataConnectionLayer.Actors;
@@ -25,11 +26,16 @@ namespace ScadaLink.DataConnectionLayer.Actors;
/// </summary>
public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
{
public enum ActiveEndpoint { Primary, Backup }
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly string _connectionName;
private readonly IDataConnection _adapter;
private IDataConnection _adapter;
private readonly DataConnectionOptions _options;
private readonly ISiteHealthCollector _healthCollector;
private readonly IDataConnectionFactory _factory;
private readonly string _protocolType;
private readonly ISiteEventLogger? _siteEventLogger;
public IStash Stash { get; set; } = null!;
public ITimerScheduler Timers { get; set; } = null!;
@@ -60,7 +66,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
private int _totalSubscribed;
private int _resolvedTags;
private readonly IDictionary<string, string> _connectionDetails;
private IDictionary<string, string> _connectionDetails;
private readonly IDictionary<string, string> _primaryConfig;
private readonly IDictionary<string, string>? _backupConfig;
private readonly int _failoverRetryCount;
private ActiveEndpoint _activeEndpoint = ActiveEndpoint.Primary;
private int _consecutiveFailures;
/// <summary>
/// Captured Self reference for use from non-actor threads (event handlers, callbacks).
@@ -73,13 +84,24 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
IDataConnection adapter,
DataConnectionOptions options,
ISiteHealthCollector healthCollector,
IDictionary<string, string>? connectionDetails = null)
IDataConnectionFactory factory,
string protocolType,
IDictionary<string, string>? primaryConfig = null,
IDictionary<string, string>? backupConfig = null,
int failoverRetryCount = 3,
ISiteEventLogger? siteEventLogger = null)
{
_connectionName = connectionName;
_adapter = adapter;
_options = options;
_healthCollector = healthCollector;
_connectionDetails = connectionDetails ?? new Dictionary<string, string>();
_factory = factory;
_protocolType = protocolType;
_primaryConfig = primaryConfig ?? new Dictionary<string, string>();
_backupConfig = backupConfig;
_failoverRetryCount = failoverRetryCount;
_siteEventLogger = siteEventLogger;
_connectionDetails = _primaryConfig;
}
protected override void PreStart()
@@ -280,7 +302,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
{
if (result.Success)
{
_log.Info("[{0}] Reconnected successfully", _connectionName);
_log.Info("[{0}] Reconnected successfully on {1} endpoint", _connectionName, _activeEndpoint);
_consecutiveFailures = 0;
// Log restoration event to site event log
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Info", null, _connectionName,
$"Connection restored on {_activeEndpoint} endpoint", null);
}
// WP-10: Transparent re-subscribe — re-establish all active subscriptions
ReSubscribeAll();
@@ -289,8 +320,52 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
}
else
{
_log.Warning("[{0}] Reconnect failed: {1}. Retrying in {2}s",
_connectionName, result.Error, _options.ReconnectInterval.TotalSeconds);
_consecutiveFailures++;
// Failover: switch endpoint after exhausting retry count (only if backup is configured)
if (_backupConfig != null && _consecutiveFailures >= _failoverRetryCount)
{
var previousEndpoint = _activeEndpoint;
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
? ActiveEndpoint.Backup
: ActiveEndpoint.Primary;
_consecutiveFailures = 0;
var newConfig = _activeEndpoint == ActiveEndpoint.Primary
? _primaryConfig
: _backupConfig;
// Dispose old adapter (fire-and-forget — don't await in actor context)
_adapter.Disconnected -= OnAdapterDisconnected;
_ = _adapter.DisposeAsync().AsTask();
// Create new adapter for the target endpoint
_adapter = _factory.Create(_protocolType, newConfig);
_connectionDetails = newConfig;
// Wire disconnect handler on new adapter
_adapter.Disconnected += OnAdapterDisconnected;
_log.Warning("[{0}] Failing over from {1} to {2}",
_connectionName, previousEndpoint, _activeEndpoint);
// Log failover event to site event log
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Warning", null, _connectionName,
$"Failover from {previousEndpoint} to {_activeEndpoint}",
$"After {_failoverRetryCount} consecutive failures");
}
}
else
{
var retryLimit = _backupConfig != null ? _failoverRetryCount.ToString() : "∞";
_log.Warning("[{0}] Reconnect failed: {1}. Retrying in {2}s (attempt {3}/{4})",
_connectionName, result.Error, _options.ReconnectInterval.TotalSeconds,
_consecutiveFailures, retryLimit);
}
Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval);
}
}
@@ -516,8 +591,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
private void ReplyWithHealthReport()
{
var status = _adapter.Status;
var endpointLabel = _backupConfig == null
? "Primary (no backup)"
: _activeEndpoint.ToString();
Sender.Tell(new DataConnectionHealthReport(
_connectionName, status, _totalSubscribed, _resolvedTags, DateTimeOffset.UtcNow));
_connectionName, status, _totalSubscribed, _resolvedTags, endpointLabel, DateTimeOffset.UtcNow));
}
// ── Internal message handlers for piped async results ──
@@ -3,6 +3,7 @@ using Akka.Event;
using ScadaLink.Commons.Interfaces.Protocol;
using ScadaLink.Commons.Messages.DataConnection;
using ScadaLink.HealthMonitoring;
using ScadaLink.SiteEventLogging;
namespace ScadaLink.DataConnectionLayer.Actors;
@@ -17,16 +18,19 @@ public class DataConnectionManagerActor : ReceiveActor
private readonly IDataConnectionFactory _factory;
private readonly DataConnectionOptions _options;
private readonly ISiteHealthCollector _healthCollector;
private readonly ISiteEventLogger? _siteEventLogger;
private readonly Dictionary<string, IActorRef> _connectionActors = new();
public DataConnectionManagerActor(
IDataConnectionFactory factory,
DataConnectionOptions options,
ISiteHealthCollector healthCollector)
ISiteHealthCollector healthCollector,
ISiteEventLogger? siteEventLogger = null)
{
_factory = factory;
_options = options;
_healthCollector = healthCollector;
_siteEventLogger = siteEventLogger;
Receive<CreateConnectionCommand>(HandleCreateConnection);
Receive<SubscribeTagsRequest>(HandleRoute);
@@ -45,10 +49,15 @@ public class DataConnectionManagerActor : ReceiveActor
}
// WP-34: Factory creates the correct adapter based on protocol type
var adapter = _factory.Create(command.ProtocolType, command.ConnectionDetails);
var adapter = _factory.Create(command.ProtocolType, command.PrimaryConnectionDetails);
var props = Props.Create(() => new DataConnectionActor(
command.ConnectionName, adapter, _options, _healthCollector, command.ConnectionDetails));
command.ConnectionName, adapter, _options, _healthCollector,
_factory, command.ProtocolType,
command.PrimaryConnectionDetails,
command.BackupConnectionDetails,
command.FailoverRetryCount,
_siteEventLogger));
// Sanitize name for Akka actor path (replace spaces and invalid chars)
var actorName = new string(command.ConnectionName
@@ -2,6 +2,7 @@ using Microsoft.Extensions.Logging;
using ScadaLink.Commons.Interfaces.Protocol;
using ScadaLink.Commons.Types.Enums;
using ZB.MOM.WW.LmxProxy.Client.Domain;
using ScadaLink.Commons.Types;
using QualityCode = ScadaLink.Commons.Interfaces.Protocol.QualityCode;
using WriteResult = ScadaLink.Commons.Interfaces.Protocol.WriteResult;
@@ -76,7 +77,7 @@ public class LmxProxyDataConnection : IDataConnection
{
var vtq = await _client!.ReadAsync(tagPath, cancellationToken);
var quality = MapQuality(vtq.Quality);
var tagValue = new TagValue(vtq.Value, quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero));
var tagValue = new TagValue(NormalizeValue(vtq.Value), quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero));
return vtq.Quality.IsBad()
? new ReadResult(false, tagValue, "LmxProxy read returned bad quality")
@@ -100,7 +101,7 @@ public class LmxProxyDataConnection : IDataConnection
foreach (var (tag, vtq) in vtqs)
{
var quality = MapQuality(vtq.Quality);
var tagValue = new TagValue(vtq.Value, quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero));
var tagValue = new TagValue(NormalizeValue(vtq.Value), quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero));
results[tag] = vtq.Quality.IsBad()
? new ReadResult(false, tagValue, "LmxProxy read returned bad quality")
: new ReadResult(true, tagValue, null);
@@ -177,7 +178,7 @@ public class LmxProxyDataConnection : IDataConnection
(path, vtq) =>
{
var quality = MapQuality(vtq.Quality);
callback(path, new TagValue(vtq.Value, quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero)));
callback(path, new TagValue(NormalizeValue(vtq.Value), quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero)));
},
onStreamError: ex =>
{
@@ -231,6 +232,18 @@ public class LmxProxyDataConnection : IDataConnection
Disconnected?.Invoke();
}
/// <summary>
/// Normalizes a Vtq value for consumption by the rest of the system.
/// Converts LmxProxy ArrayValue objects to comma-separated strings
/// so downstream code doesn't need to know about LmxProxy domain types.
/// </summary>
private static object? NormalizeValue(object? value) => value switch
{
null or string => value,
IFormattable => value,
_ => ValueFormatter.FormatDisplayValue(value)
};
private static QualityCode MapQuality(Quality quality)
{
if (quality.IsGood()) return QualityCode.Good;
@@ -21,6 +21,7 @@
<ItemGroup>
<ProjectReference Include="../ScadaLink.Commons/ScadaLink.Commons.csproj" />
<ProjectReference Include="../ScadaLink.HealthMonitoring/ScadaLink.HealthMonitoring.csproj" />
<ProjectReference Include="../ScadaLink.SiteEventLogging/ScadaLink.SiteEventLogging.csproj" />
<ProjectReference Include="../../lmxproxy/src/ZB.MOM.WW.LmxProxy.Client/ZB.MOM.WW.LmxProxy.Client.csproj" />
</ItemGroup>
@@ -102,7 +102,7 @@ public class ArtifactDeploymentService
// Map data connections
var dataConnectionArtifacts = dataConnections.Select(dc =>
new DataConnectionArtifact(dc.Name, dc.Protocol, dc.Configuration)).ToList();
new DataConnectionArtifact(dc.Name, dc.Protocol, dc.PrimaryConfiguration, dc.BackupConfiguration, dc.FailoverRetryCount)).ToList();
// Map SMTP configurations — use Host as the artifact name (matches SQLite PK on site)
var smtpArtifacts = smtpConfigurations.Select(smtp =>
@@ -231,9 +231,10 @@ akka {{
if (dclFactory != null)
{
var healthCollector = _serviceProvider.GetRequiredService<ScadaLink.HealthMonitoring.ISiteHealthCollector>();
var siteEventLogger = _serviceProvider.GetService<ScadaLink.SiteEventLogging.ISiteEventLogger>();
dclManager = _actorSystem!.ActorOf(
Props.Create(() => new ScadaLink.DataConnectionLayer.Actors.DataConnectionManagerActor(
dclFactory, dclOptions, healthCollector)),
dclFactory, dclOptions, healthCollector, siteEventLogger)),
"dcl-manager");
_logger.LogInformation("Data Connection Layer manager actor created");
}
@@ -689,7 +689,12 @@ public class ManagementActor : ReceiveActor
private static async Task<object?> HandleCreateDataConnection(IServiceProvider sp, CreateDataConnectionCommand cmd, string user)
{
var repo = sp.GetRequiredService<ISiteRepository>();
var conn = new DataConnection(cmd.Name, cmd.Protocol, cmd.SiteId) { Configuration = cmd.Configuration };
var conn = new DataConnection(cmd.Name, cmd.Protocol, cmd.SiteId)
{
PrimaryConfiguration = cmd.PrimaryConfiguration,
BackupConfiguration = cmd.BackupConfiguration,
FailoverRetryCount = cmd.FailoverRetryCount
};
await repo.AddDataConnectionAsync(conn);
await repo.SaveChangesAsync();
await AuditAsync(sp, user, "Create", "DataConnection", conn.Id.ToString(), conn.Name, conn);
@@ -703,7 +708,9 @@ public class ManagementActor : ReceiveActor
?? throw new InvalidOperationException($"DataConnection with ID {cmd.DataConnectionId} not found.");
conn.Name = cmd.Name;
conn.Protocol = cmd.Protocol;
conn.Configuration = cmd.Configuration;
conn.PrimaryConfiguration = cmd.PrimaryConfiguration;
conn.BackupConfiguration = cmd.BackupConfiguration;
conn.FailoverRetryCount = cmd.FailoverRetryCount;
await repo.UpdateDataConnectionAsync(conn);
await repo.SaveChangesAsync();
await AuditAsync(sp, user, "Update", "DataConnection", conn.Id.ToString(), conn.Name, conn);
@@ -422,7 +422,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
if (_createdConnections.Contains(name))
continue;
var connectionDetails = new Dictionary<string, string>();
var primaryDetails = new Dictionary<string, string>();
if (!string.IsNullOrEmpty(connConfig.ConfigurationJson))
{
try
@@ -431,14 +431,29 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
using var doc = System.Text.Json.JsonDocument.Parse(connConfig.ConfigurationJson);
foreach (var prop in doc.RootElement.EnumerateObject())
{
connectionDetails[prop.Name] = prop.Value.ToString();
primaryDetails[prop.Name] = prop.Value.ToString();
}
}
catch { /* Ignore parse errors */ }
}
Dictionary<string, string>? backupDetails = null;
if (!string.IsNullOrEmpty(connConfig.BackupConfigurationJson))
{
try
{
backupDetails = new Dictionary<string, string>();
using var doc = System.Text.Json.JsonDocument.Parse(connConfig.BackupConfigurationJson);
foreach (var prop in doc.RootElement.EnumerateObject())
{
backupDetails[prop.Name] = prop.Value.ToString();
}
}
catch { backupDetails = null; /* Ignore parse errors */ }
}
_dclManager.Tell(new Commons.Messages.DataConnection.CreateConnectionCommand(
name, connConfig.Protocol, connectionDetails));
name, connConfig.Protocol, primaryDetails, backupDetails, connConfig.FailoverRetryCount));
_createdConnections.Add(name);
_logger.LogInformation(
@@ -615,7 +630,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
foreach (var dc in command.DataConnections)
{
await _storage.StoreDataConnectionDefinitionAsync(
dc.Name, dc.Protocol, dc.ConfigurationJson);
dc.Name, dc.Protocol, dc.PrimaryConfigurationJson,
dc.BackupConfigurationJson, dc.FailoverRetryCount);
}
}
@@ -185,7 +185,7 @@ public class SiteReplicationActor : ReceiveActor
if (command.DataConnections != null)
foreach (var dc in command.DataConnections)
await _storage.StoreDataConnectionDefinitionAsync(dc.Name, dc.Protocol, dc.ConfigurationJson);
await _storage.StoreDataConnectionDefinitionAsync(dc.Name, dc.Protocol, dc.PrimaryConfigurationJson, dc.BackupConfigurationJson, dc.FailoverRetryCount);
if (command.SmtpConfigurations != null)
foreach (var smtp in command.SmtpConfigurations)
@@ -79,10 +79,12 @@ public class SiteStorageService
);
CREATE TABLE IF NOT EXISTS data_connection_definitions (
name TEXT PRIMARY KEY,
protocol TEXT NOT NULL,
configuration TEXT,
updated_at TEXT NOT NULL
name TEXT PRIMARY KEY,
protocol TEXT NOT NULL,
configuration TEXT,
backup_configuration TEXT,
failover_retry_count INTEGER NOT NULL DEFAULT 3,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS smtp_configurations (
@@ -480,23 +482,28 @@ public class SiteStorageService
/// <summary>
/// Stores or updates a data connection definition (OPC UA endpoint, etc.).
/// </summary>
public async Task StoreDataConnectionDefinitionAsync(string name, string protocol, string? configJson)
public async Task StoreDataConnectionDefinitionAsync(
string name, string protocol, string? configJson, string? backupConfigJson = null, int failoverRetryCount = 3)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var command = connection.CreateCommand();
command.CommandText = @"
INSERT INTO data_connection_definitions (name, protocol, configuration, updated_at)
VALUES (@name, @protocol, @config, @updatedAt)
INSERT INTO data_connection_definitions (name, protocol, configuration, backup_configuration, failover_retry_count, updated_at)
VALUES (@name, @protocol, @config, @backupConfig, @failoverRetryCount, @updatedAt)
ON CONFLICT(name) DO UPDATE SET
protocol = excluded.protocol,
configuration = excluded.configuration,
backup_configuration = excluded.backup_configuration,
failover_retry_count = excluded.failover_retry_count,
updated_at = excluded.updated_at";
command.Parameters.AddWithValue("@name", name);
command.Parameters.AddWithValue("@protocol", protocol);
command.Parameters.AddWithValue("@config", (object?)configJson ?? DBNull.Value);
command.Parameters.AddWithValue("@backupConfig", (object?)backupConfigJson ?? DBNull.Value);
command.Parameters.AddWithValue("@failoverRetryCount", failoverRetryCount);
command.Parameters.AddWithValue("@updatedAt", DateTimeOffset.UtcNow.ToString("O"));
await command.ExecuteNonQueryAsync();
@@ -512,7 +519,7 @@ public class SiteStorageService
await connection.OpenAsync();
await using var command = connection.CreateCommand();
command.CommandText = "SELECT name, protocol, configuration FROM data_connection_definitions";
command.CommandText = "SELECT name, protocol, configuration, backup_configuration, failover_retry_count FROM data_connection_definitions";
var results = new List<StoredDataConnectionDefinition>();
await using var reader = await command.ExecuteReaderAsync();
@@ -522,7 +529,9 @@ public class SiteStorageService
{
Name = reader.GetString(0),
Protocol = reader.GetString(1),
ConfigurationJson = reader.IsDBNull(2) ? null : reader.GetString(2)
ConfigurationJson = reader.IsDBNull(2) ? null : reader.GetString(2),
BackupConfigurationJson = reader.IsDBNull(3) ? null : reader.GetString(3),
FailoverRetryCount = reader.GetInt32(4)
});
}
@@ -562,4 +571,6 @@ public class StoredDataConnectionDefinition
public string Name { get; init; } = string.Empty;
public string Protocol { get; init; } = string.Empty;
public string? ConfigurationJson { get; init; }
public string? BackupConfigurationJson { get; init; }
public int FailoverRetryCount { get; init; } = 3;
}
@@ -90,7 +90,9 @@ public class FlatteningService
connections[attr.BoundDataConnectionName] = new ConnectionConfig
{
Protocol = conn.Protocol,
ConfigurationJson = conn.Configuration
ConfigurationJson = conn.PrimaryConfiguration,
BackupConfigurationJson = conn.BackupConfiguration,
FailoverRetryCount = conn.FailoverRetryCount
};
}
}
@@ -96,7 +96,8 @@ public class SiteService
// --- Data Connection CRUD ---
public async Task<Result<DataConnection>> CreateDataConnectionAsync(
int siteId, string name, string protocol, string? configuration, string user,
int siteId, string name, string protocol, string? primaryConfiguration,
string? backupConfiguration, int failoverRetryCount, string user,
CancellationToken cancellationToken = default)
{
if (string.IsNullOrWhiteSpace(name))
@@ -104,7 +105,12 @@ public class SiteService
if (string.IsNullOrWhiteSpace(protocol))
return Result<DataConnection>.Failure("Protocol is required.");
var connection = new DataConnection(name, protocol, siteId) { Configuration = configuration };
var connection = new DataConnection(name, protocol, siteId)
{
PrimaryConfiguration = primaryConfiguration,
BackupConfiguration = backupConfiguration,
FailoverRetryCount = failoverRetryCount
};
await _repository.AddDataConnectionAsync(connection, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
@@ -115,7 +121,8 @@ public class SiteService
}
public async Task<Result<DataConnection>> UpdateDataConnectionAsync(
int connectionId, string name, string protocol, string? configuration, string user,
int connectionId, string name, string protocol, string? primaryConfiguration,
string? backupConfiguration, int failoverRetryCount, string user,
CancellationToken cancellationToken = default)
{
var connection = await _repository.GetDataConnectionByIdAsync(connectionId, cancellationToken);
@@ -124,7 +131,9 @@ public class SiteService
connection.Name = name;
connection.Protocol = protocol;
connection.Configuration = configuration;
connection.PrimaryConfiguration = primaryConfiguration;
connection.BackupConfiguration = backupConfiguration;
connection.FailoverRetryCount = failoverRetryCount;
await _repository.UpdateDataConnectionAsync(connection, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
@@ -1,6 +1,7 @@
using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using NSubstitute.Core;
using ScadaLink.Commons.Interfaces.Protocol;
using ScadaLink.Commons.Messages.DataConnection;
using ScadaLink.Commons.Types.Enums;
@@ -17,18 +18,21 @@ namespace ScadaLink.DataConnectionLayer.Tests;
/// WP-12: Tag path resolution with retry tests.
/// WP-13: Health reporting tests.
/// WP-14: Subscription lifecycle tests.
/// Task-4: Failover state machine tests.
/// </summary>
public class DataConnectionActorTests : TestKit
{
private readonly IDataConnection _mockAdapter;
private readonly DataConnectionOptions _options;
private readonly ISiteHealthCollector _mockHealthCollector;
private readonly IDataConnectionFactory _mockFactory;
public DataConnectionActorTests()
: base(@"akka.loglevel = DEBUG")
{
_mockAdapter = Substitute.For<IDataConnection>();
_mockHealthCollector = Substitute.For<ISiteHealthCollector>();
_mockFactory = Substitute.For<IDataConnectionFactory>();
_options = new DataConnectionOptions
{
ReconnectInterval = TimeSpan.FromMilliseconds(100),
@@ -40,7 +44,32 @@ public class DataConnectionActorTests : TestKit
private IActorRef CreateConnectionActor(string name = "test-conn")
{
return Sys.ActorOf(Props.Create(() =>
new DataConnectionActor(name, _mockAdapter, _options, _mockHealthCollector)), name);
new DataConnectionActor(name, _mockAdapter, _options, _mockHealthCollector,
_mockFactory, "OpcUa")), name);
}
/// <summary>
/// Creates a DataConnectionActor with primary/backup failover configuration.
/// </summary>
private IActorRef CreateFailoverActor(
IDataConnection adapter,
string name,
IDictionary<string, string> primaryConfig,
IDictionary<string, string>? backupConfig,
int failoverRetryCount)
{
return Sys.ActorOf(Props.Create(() =>
new DataConnectionActor(
name, adapter, _options, _mockHealthCollector, _mockFactory, "OpcUa",
primaryConfig, backupConfig, failoverRetryCount)), name);
}
/// <summary>
/// Raises the Disconnected event on a NSubstitute mock IDataConnection.
/// </summary>
private static void RaiseDisconnected(IDataConnection adapter)
{
adapter.Disconnected += Raise.Event<Action>();
}
[Fact]
@@ -144,4 +173,288 @@ public class DataConnectionActorTests : TestKit
Assert.Equal("health-test", report.ConnectionName);
Assert.Equal(ConnectionHealth.Connected, report.Status);
}
// ── Task-4: Failover state machine tests ──
[Fact]
public async Task Task4_FailoverAfterNRetries_SwitchesToBackup()
{
// Arrange: primary + backup, failoverRetryCount = 2
var primaryConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://primary:4840" };
var backupConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://backup:4840" };
var primaryAdapter = Substitute.For<IDataConnection>();
var backupAdapter = Substitute.For<IDataConnection>();
// Initial connect succeeds on primary
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
// Factory returns backup adapter when called with backup config
_mockFactory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == "opc.tcp://backup:4840"))
.Returns(backupAdapter);
// Backup adapter connect succeeds (so failover can complete)
backupAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
var actor = CreateFailoverActor(primaryAdapter, "failover-test", primaryConfig, backupConfig, failoverRetryCount: 2);
// Wait for initial connection on primary
AwaitCondition(() =>
primaryAdapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "ConnectAsync"),
TimeSpan.FromSeconds(2));
await Task.Delay(200); // State transition to Connected
// Now make primary reconnect fail
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.FromException(new Exception("Connection refused")));
// Trigger disconnect
RaiseDisconnected(primaryAdapter);
// Wait for failover: after 2 failures, factory should be called with backup config
AwaitCondition(() =>
_mockFactory.ReceivedCalls().Any(c =>
c.GetMethodInfo().Name == "Create" &&
c.GetArguments()[1] is IDictionary<string, string> d &&
d["Endpoint"] == "opc.tcp://backup:4840"),
TimeSpan.FromSeconds(5));
}
[Fact]
public async Task Task4_SingleEndpoint_RetriesIndefinitely_NoFailover()
{
// Arrange: primary only, no backup
var primaryConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://primary:4840" };
var primaryAdapter = Substitute.For<IDataConnection>();
var connectCount = 0;
// First connect succeeds, all subsequent fail
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(callInfo =>
{
var count = Interlocked.Increment(ref connectCount);
if (count == 1) return Task.CompletedTask;
return Task.FromException(new Exception("Connection refused"));
});
var actor = CreateFailoverActor(primaryAdapter, "no-backup-test", primaryConfig, backupConfig: null, failoverRetryCount: 3);
// Wait for initial connection
AwaitCondition(() => connectCount >= 1, TimeSpan.FromSeconds(2));
await Task.Delay(200);
// Trigger disconnect — starts reconnect attempts
RaiseDisconnected(primaryAdapter);
// Wait for many reconnect failures (well over the failoverRetryCount threshold)
AwaitCondition(() => connectCount >= 8, TimeSpan.FromSeconds(10));
// Factory should never be called — no backup to fail over to
_mockFactory.DidNotReceive().Create(Arg.Any<string>(), Arg.Any<IDictionary<string, string>>());
}
[Fact]
public async Task Task4_RoundRobin_BackToPrimary_AfterBackupFails()
{
// Arrange: primary + backup, failoverRetryCount = 1
var primaryConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://primary:4840" };
var backupConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://backup:4840" };
var primaryAdapter = Substitute.For<IDataConnection>();
var backupAdapter = Substitute.For<IDataConnection>();
var secondPrimaryAdapter = Substitute.For<IDataConnection>();
// Initial connect on primary succeeds
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
// After disconnect, primary reconnect fails (triggers failover to backup)
var primaryConnectCount = 0;
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(callInfo =>
{
var count = Interlocked.Increment(ref primaryConnectCount);
if (count == 1) return Task.CompletedTask; // Initial connect
return Task.FromException(new Exception("Primary down")); // Reconnect fails
});
// Factory: backup config → backupAdapter, primary config → secondPrimaryAdapter
_mockFactory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == "opc.tcp://backup:4840"))
.Returns(backupAdapter);
_mockFactory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == "opc.tcp://primary:4840"))
.Returns(secondPrimaryAdapter);
// Backup connect succeeds first time, then fails on reconnect
var backupConnectCount = 0;
backupAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(callInfo =>
{
var count = Interlocked.Increment(ref backupConnectCount);
if (count == 1) return Task.CompletedTask; // First backup connect succeeds
return Task.FromException(new Exception("Backup down")); // Backup reconnect fails
});
// Second primary adapter connects fine
secondPrimaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
var actor = CreateFailoverActor(primaryAdapter, "roundrobin-test", primaryConfig, backupConfig, failoverRetryCount: 1);
// Wait for initial primary connect
AwaitCondition(() => primaryConnectCount >= 1, TimeSpan.FromSeconds(2));
await Task.Delay(200);
// Disconnect primary → 1 failure → failover to backup
RaiseDisconnected(primaryAdapter);
// Wait for backup adapter creation
AwaitCondition(() =>
_mockFactory.ReceivedCalls().Any(c =>
c.GetMethodInfo().Name == "Create" &&
c.GetArguments()[1] is IDictionary<string, string> d &&
d["Endpoint"] == "opc.tcp://backup:4840"),
TimeSpan.FromSeconds(5));
// Wait for backup to connect successfully
AwaitCondition(() => backupConnectCount >= 1, TimeSpan.FromSeconds(2));
await Task.Delay(200);
// Now disconnect backup → 1 failure → failover back to primary
RaiseDisconnected(backupAdapter);
// Wait for primary adapter re-creation
AwaitCondition(() =>
_mockFactory.ReceivedCalls().Any(c =>
c.GetMethodInfo().Name == "Create" &&
c.GetArguments()[1] is IDictionary<string, string> d &&
d["Endpoint"] == "opc.tcp://primary:4840"),
TimeSpan.FromSeconds(5));
}
[Fact]
public async Task Task4_SuccessfulReconnect_ResetsFailureCounter()
{
// Arrange: primary + backup, failoverRetryCount = 3
var primaryConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://primary:4840" };
var backupConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://backup:4840" };
var primaryAdapter = Substitute.For<IDataConnection>();
var connectCount = 0;
// First connect succeeds, then 2 failures, then success, then 2 more failures, then success
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(callInfo =>
{
var count = Interlocked.Increment(ref connectCount);
// count 1: initial connect → success
// count 2,3: reconnect failures
// count 4: reconnect success (resets counter)
// count 5,6: reconnect failures again
// count 7: reconnect success again
return count switch
{
1 => Task.CompletedTask,
2 or 3 => Task.FromException(new Exception("Fail")),
4 => Task.CompletedTask,
5 or 6 => Task.FromException(new Exception("Fail")),
_ => Task.CompletedTask
};
});
var actor = CreateFailoverActor(primaryAdapter, "reset-counter-test", primaryConfig, backupConfig, failoverRetryCount: 3);
// Wait for initial connect
AwaitCondition(() => connectCount >= 1, TimeSpan.FromSeconds(2));
await Task.Delay(200);
// Disconnect: triggers 2 failures then success (count 2,3,4)
RaiseDisconnected(primaryAdapter);
// Wait for successful reconnect (count 4)
AwaitCondition(() => connectCount >= 4, TimeSpan.FromSeconds(5));
await Task.Delay(200);
// Disconnect again: triggers 2 more failures then success (count 5,6,7)
RaiseDisconnected(primaryAdapter);
// Wait for second successful reconnect (count 7)
AwaitCondition(() => connectCount >= 7, TimeSpan.FromSeconds(5));
await Task.Delay(200);
// Factory should never be called — counter reset each time before reaching 3
_mockFactory.DidNotReceive().Create(Arg.Any<string>(), Arg.Any<IDictionary<string, string>>());
}
[Fact]
public async Task Task4_ReSubscribeAll_CalledAfterFailoverReconnect()
{
// Arrange: primary + backup, failoverRetryCount = 1
var primaryConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://primary:4840" };
var backupConfig = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://backup:4840" };
var primaryAdapter = Substitute.For<IDataConnection>();
var backupAdapter = Substitute.For<IDataConnection>();
// Primary initial connect succeeds
var primaryConnectCount = 0;
primaryAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(callInfo =>
{
var count = Interlocked.Increment(ref primaryConnectCount);
if (count == 1) return Task.CompletedTask;
return Task.FromException(new Exception("Primary down"));
});
// Primary subscribe succeeds
primaryAdapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
.Returns("sub-primary-001");
// Primary read succeeds (for initial read after subscribe)
primaryAdapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.Returns(new ReadResult(true, new TagValue(42.0, QualityCode.Good, DateTimeOffset.UtcNow), null));
// Factory returns backup adapter
_mockFactory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == "opc.tcp://backup:4840"))
.Returns(backupAdapter);
// Backup connect succeeds
backupAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
.Returns(Task.CompletedTask);
// Backup subscribe succeeds (for re-subscribe after failover)
backupAdapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
.Returns("sub-backup-001");
var actor = CreateFailoverActor(primaryAdapter, "resub-test", primaryConfig, backupConfig, failoverRetryCount: 1);
// Wait for initial connect
AwaitCondition(() => primaryConnectCount >= 1, TimeSpan.FromSeconds(2));
await Task.Delay(200);
// Subscribe to tags while connected on primary
actor.Tell(new SubscribeTagsRequest("corr1", "inst1", "resub-test", ["sensor/temp"], DateTimeOffset.UtcNow));
ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(3));
// Verify primary adapter received subscribe call
await primaryAdapter.Received().SubscribeAsync(
"sensor/temp", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>());
// Disconnect primary → 1 failure → failover to backup
RaiseDisconnected(primaryAdapter);
// Wait for backup adapter creation and connect
AwaitCondition(() =>
_mockFactory.ReceivedCalls().Any(c =>
c.GetMethodInfo().Name == "Create" &&
c.GetArguments()[1] is IDictionary<string, string> d &&
d["Endpoint"] == "opc.tcp://backup:4840"),
TimeSpan.FromSeconds(5));
// Wait for ReSubscribeAll to fire on backup adapter
AwaitCondition(() =>
backupAdapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "SubscribeAsync"),
TimeSpan.FromSeconds(5));
// Verify backup adapter received SubscribeAsync for the same tag
await backupAdapter.Received().SubscribeAsync(
"sensor/temp", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>());
}
}
@@ -70,7 +70,7 @@ public class DataConnectionManagerActorTests : TestKit
new DataConnectionManagerActor(_mockFactory, _options, _mockHealthCollector)));
manager.Tell(new CreateConnectionCommand(
"conn1", "OpcUa", new Dictionary<string, string>()));
"conn1", "OpcUa", new Dictionary<string, string>(), null, 3));
// Factory should have been called
AwaitCondition(() =>
@@ -200,7 +200,7 @@ public class FlatteningServiceTests
var connections = new Dictionary<int, DataConnection>
{
[100] = new("OPC-Server1", "OpcUa", 1) { Id = 100, Configuration = "opc.tcp://localhost:4840" }
[100] = new("OPC-Server1", "OpcUa", 1) { Id = 100, PrimaryConfiguration = "opc.tcp://localhost:4840" }
};
var result = _sut.Flatten(
@@ -93,7 +93,7 @@ public class SiteServiceTests
_repoMock.Setup(r => r.SaveChangesAsync(It.IsAny<CancellationToken>()))
.ReturnsAsync(1);
var result = await _sut.CreateDataConnectionAsync(1, "OPC-Server1", "OpcUa", "{\"url\":\"opc.tcp://localhost\"}", "admin");
var result = await _sut.CreateDataConnectionAsync(1, "OPC-Server1", "OpcUa", "{\"url\":\"opc.tcp://localhost\"}", null, 3, "admin");
Assert.True(result.IsSuccess);
Assert.Equal("OPC-Server1", result.Value.Name);