fix(site-runtime): resolve SiteRuntime-001/002/003 — route data-sourced writes to DCL, real per-attribute API results, race-free redeploy
This commit is contained in:
@@ -8,7 +8,7 @@
|
||||
| Last reviewed | 2026-05-16 |
|
||||
| Reviewer | claude-agent |
|
||||
| Commit reviewed | `9c60592` |
|
||||
| Open findings | 16 |
|
||||
| Open findings | 13 |
|
||||
|
||||
## Summary
|
||||
|
||||
@@ -51,7 +51,7 @@ actor, and the repositories are untested.
|
||||
|--|--|
|
||||
| Severity | High |
|
||||
| Category | Design-document adherence |
|
||||
| Status | Open |
|
||||
| Status | Resolved |
|
||||
| Location | `src/ScadaLink.SiteRuntime/Scripts/ScriptRuntimeContext.cs:106`, `src/ScadaLink.SiteRuntime/Actors/InstanceActor.cs:204` |
|
||||
|
||||
**Description**
|
||||
@@ -84,7 +84,15 @@ or branching inside the handler.
|
||||
|
||||
**Resolution**
|
||||
|
||||
_Unresolved._
|
||||
Resolved 2026-05-16 (`<pending>`): `InstanceActor.HandleSetStaticAttribute` now resolves
|
||||
the target attribute's data binding from `_configuration`. Data-sourced attributes are
|
||||
routed via a new `HandleSetDataAttribute` that Asks the DCL with a `WriteTagRequest` and
|
||||
pipes the device-write outcome back to the caller as a `SetStaticAttributeResponse` —
|
||||
no override is persisted and `_attributes` is not optimistically mutated. Static
|
||||
attributes keep the override path and now also reply with a `SetStaticAttributeResponse`.
|
||||
`ScriptRuntimeContext.SetAttribute` is now `async Task` and Asks the Instance Actor,
|
||||
throwing `InvalidOperationException` on a failed device write so scripts get the failure
|
||||
synchronously.
|
||||
|
||||
### SiteRuntime-002 — `RouteInboundApiSetAttributes` always treats writes as static overrides
|
||||
|
||||
@@ -92,7 +100,7 @@ _Unresolved._
|
||||
|--|--|
|
||||
| Severity | High |
|
||||
| Category | Correctness & logic bugs |
|
||||
| Status | Open |
|
||||
| Status | Resolved |
|
||||
| Location | `src/ScadaLink.SiteRuntime/Actors/DeploymentManagerActor.cs:632` |
|
||||
|
||||
**Description**
|
||||
@@ -115,7 +123,13 @@ response path.
|
||||
|
||||
**Resolution**
|
||||
|
||||
_Unresolved._
|
||||
Resolved 2026-05-16 (`<pending>`): `RouteInboundApiSetAttributes` now Asks the Instance
|
||||
Actor per attribute (instead of fire-and-forget Tell) and aggregates the
|
||||
`SetStaticAttributeResponse` results. Because the Instance Actor handler is the
|
||||
SiteRuntime-001 corrected handler, data-sourced attributes now reach the DCL and the
|
||||
`RouteToSetAttributesResponse` reflects the real per-attribute outcome — a non-existent
|
||||
attribute or a failed device write is reported as failure rather than an unconditional
|
||||
optimistic `true`.
|
||||
|
||||
### SiteRuntime-003 — Redeployment relies on a fixed 500 ms reschedule and can collide on the child actor name
|
||||
|
||||
@@ -123,7 +137,7 @@ _Unresolved._
|
||||
|--|--|
|
||||
| Severity | High |
|
||||
| Category | Akka.NET conventions |
|
||||
| Status | Open |
|
||||
| Status | Resolved |
|
||||
| Location | `src/ScadaLink.SiteRuntime/Actors/DeploymentManagerActor.cs:222` |
|
||||
|
||||
**Description**
|
||||
@@ -148,7 +162,15 @@ instance) until termination completes.
|
||||
|
||||
**Resolution**
|
||||
|
||||
_Unresolved._
|
||||
Resolved 2026-05-16 (`<pending>`): `HandleDeploy` no longer uses a fixed 500 ms
|
||||
reschedule. When a redeployment targets a running instance, the existing Instance Actor
|
||||
is `Context.Watch`-ed and stopped, and the in-flight `DeployInstanceCommand` is buffered
|
||||
in a `_pendingRedeploys` map keyed by the terminating actor ref. A new `Terminated`
|
||||
handler recreates the Instance Actor only after the predecessor (and its whole subtree)
|
||||
has fully stopped, eliminating the `InvalidActorNameException` race and the
|
||||
unconditional redeploy-latency penalty. The shared `ApplyDeployment` helper also skips
|
||||
the `_totalDeployedCount` increment for redeployments, so the deployed-instance count no
|
||||
longer drifts (this additionally addresses the root cause behind SiteRuntime-004).
|
||||
|
||||
### SiteRuntime-004 — `_totalDeployedCount` is incremented on redeployment of an existing instance
|
||||
|
||||
|
||||
@@ -39,6 +39,12 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
private readonly ISiteHealthCollector? _healthCollector;
|
||||
private readonly IServiceProvider? _serviceProvider;
|
||||
private readonly Dictionary<string, IActorRef> _instanceActors = new();
|
||||
/// <summary>
|
||||
/// Tracks Instance Actors that are terminating as part of a redeployment, keyed by
|
||||
/// the terminating actor ref. The buffered command is applied once <see cref="Terminated"/>
|
||||
/// confirms the child has fully stopped (SiteRuntime-003).
|
||||
/// </summary>
|
||||
private readonly Dictionary<IActorRef, PendingRedeploy> _pendingRedeploys = new();
|
||||
private int _totalDeployedCount;
|
||||
|
||||
public ITimerScheduler Timers { get; set; } = null!;
|
||||
@@ -94,6 +100,10 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
|
||||
// Internal deploy persistence result
|
||||
Receive<DeployPersistenceResult>(HandleDeployPersistenceResult);
|
||||
|
||||
// Terminated signal — drains a buffered redeployment once the previous
|
||||
// Instance Actor has fully stopped (SiteRuntime-003).
|
||||
Receive<Terminated>(HandleTerminated);
|
||||
}
|
||||
|
||||
protected override void PreStart()
|
||||
@@ -211,6 +221,13 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
/// <summary>
|
||||
/// Handles a new deployment: stores config in SQLite, clears previous static overrides,
|
||||
/// and creates or replaces the Instance Actor.
|
||||
///
|
||||
/// Redeployment of an already-running instance must wait for the previous Instance
|
||||
/// Actor to fully terminate (including PostStop on its descendants) before the
|
||||
/// replacement is created — otherwise <see cref="Context.ActorOf"/> can collide on
|
||||
/// the still-registered child name. Instead of guessing with a fixed timer, the
|
||||
/// terminating child is watched and the in-flight command is buffered until the
|
||||
/// <see cref="Terminated"/> signal arrives.
|
||||
/// </summary>
|
||||
private void HandleDeploy(DeployInstanceCommand command)
|
||||
{
|
||||
@@ -219,28 +236,54 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
"Deploying instance {Instance}, deploymentId={DeploymentId}",
|
||||
instanceName, command.DeploymentId);
|
||||
|
||||
// Stop existing actor if present (redeployment replaces)
|
||||
// Redeployment replaces a running instance. Watch + stop the existing actor
|
||||
// and buffer this command until its Terminated signal confirms the child
|
||||
// (and its whole subtree) has fully stopped and freed its actor name.
|
||||
if (_instanceActors.TryGetValue(instanceName, out var existing))
|
||||
{
|
||||
Context.Stop(existing);
|
||||
_instanceActors.Remove(instanceName);
|
||||
// Wait for the child to be removed from the children collection
|
||||
// by yielding and retrying — Context.Stop is processed before the next message
|
||||
Context.System.Scheduler.ScheduleTellOnce(
|
||||
TimeSpan.FromMilliseconds(500), Self, command, Sender);
|
||||
_pendingRedeploys[existing] = new PendingRedeploy(command, Sender);
|
||||
Context.Watch(existing);
|
||||
Context.Stop(existing);
|
||||
UpdateInstanceCounts();
|
||||
return;
|
||||
}
|
||||
|
||||
// Fresh deployment — no existing actor to replace.
|
||||
ApplyDeployment(command, Sender, isRedeploy: false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Recreates an Instance Actor once its predecessor has fully terminated during a
|
||||
/// redeployment, draining the buffered <see cref="DeployInstanceCommand"/>.
|
||||
/// </summary>
|
||||
private void HandleTerminated(Terminated terminated)
|
||||
{
|
||||
if (!_pendingRedeploys.Remove(terminated.ActorRef, out var pending))
|
||||
return;
|
||||
|
||||
ApplyDeployment(pending.Command, pending.OriginalSender, isRedeploy: true);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates the Instance Actor, persists the config, and replies to the deployer.
|
||||
/// A redeployment is an update of an existing instance, so the deployed-instance
|
||||
/// counter is only incremented for genuinely new deployments.
|
||||
/// </summary>
|
||||
private void ApplyDeployment(DeployInstanceCommand command, IActorRef sender, bool isRedeploy)
|
||||
{
|
||||
var instanceName = command.InstanceUniqueName;
|
||||
|
||||
// Ensure DCL connections exist for any data-sourced attributes
|
||||
EnsureDclConnections(command.FlattenedConfigurationJson);
|
||||
|
||||
// Create the Instance Actor immediately (no existing actor to replace)
|
||||
// Create the Instance Actor immediately
|
||||
CreateInstanceActor(instanceName, command.FlattenedConfigurationJson);
|
||||
if (!isRedeploy)
|
||||
_totalDeployedCount++;
|
||||
UpdateInstanceCounts();
|
||||
|
||||
// Persist to SQLite and clear static overrides asynchronously
|
||||
var sender = Sender;
|
||||
Task.Run(async () =>
|
||||
{
|
||||
await _storage.StoreDeployedConfigAsync(
|
||||
@@ -614,9 +657,11 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
|
||||
/// <summary>
|
||||
/// Writes attribute values on a deployed instance for a Route.To().SetAttribute(s)
|
||||
/// call (or a central Test Run bound to the instance). Writes are Tell'd to the
|
||||
/// Instance Actor — serialized through its mailbox — and acknowledged optimistically,
|
||||
/// matching the fire-and-forget semantics of Instance.SetAttribute.
|
||||
/// call (or a central Test Run bound to the instance). Each write is Ask'd to the
|
||||
/// Instance Actor, which routes data-sourced attributes through the DCL and static
|
||||
/// attributes to a persisted override. The response reflects the real per-attribute
|
||||
/// outcome (a non-existent attribute or a failed device write reports failure),
|
||||
/// rather than an unconditional optimistic ack.
|
||||
/// </summary>
|
||||
private void RouteInboundApiSetAttributes(RouteToSetAttributesRequest request)
|
||||
{
|
||||
@@ -629,14 +674,33 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (var (name, value) in request.AttributeValues)
|
||||
{
|
||||
instanceActor.Tell(new SetStaticAttributeCommand(
|
||||
request.CorrelationId, request.InstanceUniqueName, name, value, DateTimeOffset.UtcNow));
|
||||
}
|
||||
var sender = Sender;
|
||||
var correlationId = request.CorrelationId;
|
||||
var asks = request.AttributeValues
|
||||
.Select(kvp => instanceActor.Ask<SetStaticAttributeResponse>(
|
||||
new SetStaticAttributeCommand(
|
||||
correlationId, request.InstanceUniqueName, kvp.Key, kvp.Value, DateTimeOffset.UtcNow),
|
||||
TimeSpan.FromSeconds(30)))
|
||||
.ToArray();
|
||||
|
||||
Sender.Tell(new RouteToSetAttributesResponse(
|
||||
request.CorrelationId, true, null, DateTimeOffset.UtcNow));
|
||||
Task.WhenAll(asks).ContinueWith(t =>
|
||||
{
|
||||
if (!t.IsCompletedSuccessfully)
|
||||
return new RouteToSetAttributesResponse(
|
||||
correlationId, false,
|
||||
t.Exception?.GetBaseException().Message ?? "Attribute write timed out",
|
||||
DateTimeOffset.UtcNow);
|
||||
|
||||
var failures = t.Result
|
||||
.Where(r => !r.Success)
|
||||
.Select(r => $"{r.AttributeName}: {r.ErrorMessage}")
|
||||
.ToArray();
|
||||
|
||||
return failures.Length == 0
|
||||
? new RouteToSetAttributesResponse(correlationId, true, null, DateTimeOffset.UtcNow)
|
||||
: new RouteToSetAttributesResponse(
|
||||
correlationId, false, string.Join("; ", failures), DateTimeOffset.UtcNow);
|
||||
}).PipeTo(sender);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -789,4 +853,9 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
EnableInstanceCommand Command, DeployedInstance? Config, string? Error, IActorRef OriginalSender);
|
||||
internal record DeployPersistenceResult(
|
||||
string DeploymentId, string InstanceName, bool Success, string? Error, IActorRef OriginalSender);
|
||||
|
||||
/// <summary>
|
||||
/// A redeployment command buffered until the previous Instance Actor terminates.
|
||||
/// </summary>
|
||||
internal record PendingRedeploy(DeployInstanceCommand Command, IActorRef OriginalSender);
|
||||
}
|
||||
|
||||
@@ -198,10 +198,44 @@ public class InstanceActor : ReceiveActor
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Updates a static attribute in memory and persists the override to SQLite.
|
||||
/// Handles an attribute write (<c>Instance.SetAttribute</c> / Inbound API).
|
||||
/// WP-24: State mutation serialized through this actor's mailbox.
|
||||
///
|
||||
/// The write is routed by the attribute's data binding:
|
||||
/// * Data-sourced attribute → forwards a <see cref="WriteTagRequest"/> to the
|
||||
/// DCL, which writes the physical device. The in-memory value is NOT
|
||||
/// optimistically updated and NO static override is persisted — the
|
||||
/// confirmed device value arrives later via the subscription. Success or
|
||||
/// failure of the device write is returned to the caller.
|
||||
/// * Static attribute → updates the in-memory value and persists the override
|
||||
/// to SQLite.
|
||||
///
|
||||
/// Either way the caller receives a <see cref="SetStaticAttributeResponse"/>.
|
||||
/// </summary>
|
||||
private void HandleSetStaticAttribute(SetStaticAttributeCommand command)
|
||||
{
|
||||
// Resolve the target attribute's data binding from the flattened config.
|
||||
var resolved = _configuration?.Attributes
|
||||
.FirstOrDefault(a => a.CanonicalName == command.AttributeName);
|
||||
|
||||
var isDataSourced = resolved != null
|
||||
&& !string.IsNullOrEmpty(resolved.DataSourceReference)
|
||||
&& !string.IsNullOrEmpty(resolved.BoundDataConnectionName);
|
||||
|
||||
if (isDataSourced)
|
||||
{
|
||||
HandleSetDataAttribute(command, resolved!);
|
||||
return;
|
||||
}
|
||||
|
||||
HandleSetStaticAttributeCore(command);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Static attribute write: updates in-memory state, publishes the change,
|
||||
/// persists the override to SQLite, and replies with success.
|
||||
/// </summary>
|
||||
private void HandleSetStaticAttributeCore(SetStaticAttributeCommand command)
|
||||
{
|
||||
_attributes[command.AttributeName] = command.Value;
|
||||
|
||||
@@ -216,8 +250,7 @@ public class InstanceActor : ReceiveActor
|
||||
|
||||
PublishAndNotifyChildren(changed);
|
||||
|
||||
// Persist asynchronously -- fire and forget since the actor is the source of truth
|
||||
// and SetAttribute is called from scripts via Tell (no response consumer).
|
||||
// Persist asynchronously -- fire and forget since the actor is the source of truth.
|
||||
var instanceName = _instanceUniqueName;
|
||||
var attributeName = command.AttributeName;
|
||||
var logger = _logger;
|
||||
@@ -230,6 +263,58 @@ public class InstanceActor : ReceiveActor
|
||||
instanceName,
|
||||
attributeName);
|
||||
}, TaskContinuationOptions.OnlyOnFaulted);
|
||||
|
||||
Sender.Tell(new SetStaticAttributeResponse(
|
||||
command.CorrelationId, _instanceUniqueName, command.AttributeName,
|
||||
true, null, DateTimeOffset.UtcNow));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Data-sourced attribute write: forwards a write request to the DCL and pipes
|
||||
/// the device write result back to the caller. The in-memory value is left
|
||||
/// untouched (it is refreshed by the subscription when the device confirms);
|
||||
/// no static override is persisted for a data-sourced attribute.
|
||||
/// </summary>
|
||||
private void HandleSetDataAttribute(SetStaticAttributeCommand command, ResolvedAttribute resolved)
|
||||
{
|
||||
var caller = Sender;
|
||||
var correlationId = command.CorrelationId;
|
||||
var attributeName = command.AttributeName;
|
||||
var instanceName = _instanceUniqueName;
|
||||
|
||||
if (_dclManager == null)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"SetAttribute on data-sourced attribute {Instance}.{Attribute} cannot be routed — no DCL manager configured",
|
||||
instanceName, attributeName);
|
||||
caller.Tell(new SetStaticAttributeResponse(
|
||||
correlationId, instanceName, attributeName, false,
|
||||
"Data Connection Layer not available for write.", DateTimeOffset.UtcNow));
|
||||
return;
|
||||
}
|
||||
|
||||
var writeRequest = new WriteTagRequest(
|
||||
correlationId,
|
||||
resolved.BoundDataConnectionName!,
|
||||
resolved.DataSourceReference!,
|
||||
command.Value,
|
||||
DateTimeOffset.UtcNow);
|
||||
|
||||
// Ask the DCL and pipe the result back to the original caller. The DCL
|
||||
// returns the failure synchronously so the script can handle it.
|
||||
_dclManager.Ask<WriteTagResponse>(writeRequest, TimeSpan.FromSeconds(30))
|
||||
.ContinueWith(t =>
|
||||
{
|
||||
if (t.IsCompletedSuccessfully)
|
||||
return new SetStaticAttributeResponse(
|
||||
correlationId, instanceName, attributeName,
|
||||
t.Result.Success, t.Result.ErrorMessage, DateTimeOffset.UtcNow);
|
||||
|
||||
return new SetStaticAttributeResponse(
|
||||
correlationId, instanceName, attributeName, false,
|
||||
t.Exception?.GetBaseException().Message ?? "DCL write timed out",
|
||||
DateTimeOffset.UtcNow);
|
||||
}).PipeTo(caller);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -25,17 +25,17 @@ public class AttributeAccessor
|
||||
|
||||
public object? this[string key]
|
||||
{
|
||||
// Both reads and writes block on the actor Ask; the write also blocks
|
||||
// on the DCL round-trip for data-connected attributes. The async
|
||||
// variants (GetAsync/SetAsync) are preferred where awaiting is possible.
|
||||
get => _ctx.GetAttribute(Resolve(key)).GetAwaiter().GetResult();
|
||||
set => _ctx.SetAttribute(Resolve(key), value?.ToString() ?? string.Empty);
|
||||
set => _ctx.SetAttribute(Resolve(key), value?.ToString() ?? string.Empty).GetAwaiter().GetResult();
|
||||
}
|
||||
|
||||
public Task<object?> GetAsync(string key) => _ctx.GetAttribute(Resolve(key));
|
||||
|
||||
public Task SetAsync(string key, object? value)
|
||||
{
|
||||
_ctx.SetAttribute(Resolve(key), value?.ToString() ?? string.Empty);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
=> _ctx.SetAttribute(Resolve(key), value?.ToString() ?? string.Empty);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -99,18 +99,31 @@ public class ScriptRuntimeContext
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sets an attribute value. For data-connected attributes, forwards to DCL via Instance Actor.
|
||||
/// For static attributes, updates in-memory and persists to SQLite via Instance Actor.
|
||||
/// All mutations serialized through the Instance Actor mailbox.
|
||||
/// Sets an attribute value. For data-connected attributes the Instance Actor
|
||||
/// forwards the write to the DCL, which writes the physical device; the
|
||||
/// in-memory value is not optimistically updated. For static attributes the
|
||||
/// Instance Actor updates the in-memory value and persists the override to
|
||||
/// SQLite. All mutations are serialized through the Instance Actor mailbox.
|
||||
///
|
||||
/// The write is awaited so that a device-write failure on a data-connected
|
||||
/// attribute is surfaced synchronously to the calling script as an
|
||||
/// <see cref="InvalidOperationException"/>.
|
||||
/// </summary>
|
||||
public void SetAttribute(string attributeName, string value)
|
||||
public async Task SetAttribute(string attributeName, string value)
|
||||
{
|
||||
var correlationId = Guid.NewGuid().ToString();
|
||||
var command = new SetStaticAttributeCommand(
|
||||
correlationId, _instanceName, attributeName, value, DateTimeOffset.UtcNow);
|
||||
|
||||
// Tell (fire-and-forget) — mutation serialized through Instance Actor
|
||||
_instanceActor.Tell(command);
|
||||
// Ask — mutation serialized through the Instance Actor mailbox; the reply
|
||||
// carries the device-write outcome for data-connected attributes.
|
||||
var response = await _instanceActor.Ask<SetStaticAttributeResponse>(command, _askTimeout);
|
||||
|
||||
if (!response.Success)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"SetAttribute('{attributeName}') failed: {response.ErrorMessage}");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ScadaLink.Commons.Messages.Deployment;
|
||||
using ScadaLink.Commons.Messages.Health;
|
||||
using ScadaLink.Commons.Messages.Lifecycle;
|
||||
using ScadaLink.Commons.Types.Enums;
|
||||
using ScadaLink.Commons.Types.Flattening;
|
||||
using ScadaLink.HealthMonitoring;
|
||||
using ScadaLink.SiteRuntime.Actors;
|
||||
using ScadaLink.SiteRuntime.Persistence;
|
||||
using ScadaLink.SiteRuntime.Scripts;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace ScadaLink.SiteRuntime.Tests.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// Regression tests for SiteRuntime-003: redeployment of an existing instance must
|
||||
/// wait for the terminating Instance Actor before recreating the child, instead of
|
||||
/// relying on a fixed 500 ms reschedule that can collide on the child actor name.
|
||||
/// </summary>
|
||||
public class DeploymentManagerRedeployTests : TestKit, IDisposable
|
||||
{
|
||||
private readonly SiteStorageService _storage;
|
||||
private readonly ScriptCompilationService _compilationService;
|
||||
private readonly SharedScriptLibrary _sharedScriptLibrary;
|
||||
private readonly string _dbFile;
|
||||
|
||||
public DeploymentManagerRedeployTests()
|
||||
{
|
||||
_dbFile = Path.Combine(Path.GetTempPath(), $"dm-redeploy-test-{Guid.NewGuid():N}.db");
|
||||
_storage = new SiteStorageService(
|
||||
$"Data Source={_dbFile}",
|
||||
NullLogger<SiteStorageService>.Instance);
|
||||
_storage.InitializeAsync().GetAwaiter().GetResult();
|
||||
_compilationService = new ScriptCompilationService(
|
||||
NullLogger<ScriptCompilationService>.Instance);
|
||||
_sharedScriptLibrary = new SharedScriptLibrary(
|
||||
_compilationService, NullLogger<SharedScriptLibrary>.Instance);
|
||||
}
|
||||
|
||||
void IDisposable.Dispose()
|
||||
{
|
||||
Shutdown();
|
||||
try { File.Delete(_dbFile); } catch { /* cleanup */ }
|
||||
}
|
||||
|
||||
private IActorRef CreateDeploymentManager(ISiteHealthCollector? healthCollector = null)
|
||||
{
|
||||
return ActorOf(Props.Create(() => new DeploymentManagerActor(
|
||||
_storage,
|
||||
_compilationService,
|
||||
_sharedScriptLibrary,
|
||||
null,
|
||||
new SiteRuntimeOptions(),
|
||||
NullLogger<DeploymentManagerActor>.Instance,
|
||||
null,
|
||||
null,
|
||||
healthCollector,
|
||||
null)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Minimal fake that records the most recent deployed-instance count.
|
||||
/// </summary>
|
||||
private sealed class CountCapturingHealthCollector : ISiteHealthCollector
|
||||
{
|
||||
public int LastDeployedCount { get; private set; }
|
||||
public void IncrementScriptError() { }
|
||||
public void IncrementAlarmError() { }
|
||||
public void IncrementDeadLetter() { }
|
||||
public void UpdateConnectionHealth(string connectionName, ConnectionHealth health) { }
|
||||
public void RemoveConnection(string connectionName) { }
|
||||
public void UpdateTagResolution(string connectionName, int totalSubscribed, int successfullyResolved) { }
|
||||
public void UpdateConnectionEndpoint(string connectionName, string endpoint) { }
|
||||
public void UpdateTagQuality(string connectionName, int good, int bad, int uncertain) { }
|
||||
public void SetStoreAndForwardDepths(IReadOnlyDictionary<string, int> depths) { }
|
||||
public void SetInstanceCounts(int deployed, int enabled, int disabled) => LastDeployedCount = deployed;
|
||||
public void SetParkedMessageCount(int count) { }
|
||||
public void SetNodeHostname(string hostname) { }
|
||||
public void SetClusterNodes(IReadOnlyList<NodeStatus> nodes) { }
|
||||
public void SetActiveNode(bool isActive) { }
|
||||
public bool IsActiveNode => true;
|
||||
public SiteHealthReport CollectReport(string siteId) => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
private static string MakeConfigJson(string instanceName)
|
||||
{
|
||||
var config = new FlattenedConfiguration
|
||||
{
|
||||
InstanceUniqueName = instanceName,
|
||||
Attributes =
|
||||
[
|
||||
new ResolvedAttribute { CanonicalName = "TestAttr", Value = "1", DataType = "Int32" }
|
||||
]
|
||||
};
|
||||
return JsonSerializer.Serialize(config);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Redeploy_ExistingInstance_SucceedsWithoutNameCollision()
|
||||
{
|
||||
var actor = CreateDeploymentManager();
|
||||
await Task.Delay(500); // empty startup
|
||||
|
||||
// Initial deploy.
|
||||
actor.Tell(new DeployInstanceCommand(
|
||||
"dep-1", "RedeployPump", "h1", MakeConfigJson("RedeployPump"), "admin", DateTimeOffset.UtcNow));
|
||||
var first = ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal(DeploymentStatus.Success, first.Status);
|
||||
await Task.Delay(500);
|
||||
|
||||
// Redeploy the same instance — must replace the existing actor cleanly.
|
||||
actor.Tell(new DeployInstanceCommand(
|
||||
"dep-2", "RedeployPump", "h2", MakeConfigJson("RedeployPump"), "admin", DateTimeOffset.UtcNow));
|
||||
var second = ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(10));
|
||||
Assert.Equal(DeploymentStatus.Success, second.Status);
|
||||
|
||||
// The redeployed instance must still be operable (no orphaned/broken actor).
|
||||
actor.Tell(new DisableInstanceCommand("cmd-1", "RedeployPump", DateTimeOffset.UtcNow));
|
||||
var disable = ExpectMsg<InstanceLifecycleResponse>(TimeSpan.FromSeconds(5));
|
||||
Assert.True(disable.Success);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Redeploy_ExistingInstance_DoesNotOverCountDeployedInstances()
|
||||
{
|
||||
var health = new CountCapturingHealthCollector();
|
||||
var actor = CreateDeploymentManager(health);
|
||||
await Task.Delay(500);
|
||||
|
||||
// Deploy once.
|
||||
actor.Tell(new DeployInstanceCommand(
|
||||
"dep-1", "CountPump", "h1", MakeConfigJson("CountPump"), "admin", DateTimeOffset.UtcNow));
|
||||
ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(5));
|
||||
await Task.Delay(500);
|
||||
|
||||
// Redeploy several times.
|
||||
for (var i = 2; i <= 4; i++)
|
||||
{
|
||||
actor.Tell(new DeployInstanceCommand(
|
||||
$"dep-{i}", "CountPump", $"h{i}", MakeConfigJson("CountPump"), "admin", DateTimeOffset.UtcNow));
|
||||
ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(10));
|
||||
await Task.Delay(500);
|
||||
}
|
||||
|
||||
// Storage uses UPSERT — exactly one deployed config row should exist.
|
||||
var configs = await _storage.GetAllDeployedConfigsAsync();
|
||||
Assert.Single(configs, c => c.InstanceUniqueName == "CountPump");
|
||||
|
||||
// The reported deployed count must be exactly 1 — a redeploy is an update,
|
||||
// not a new instance, so the in-memory counter must not drift upward.
|
||||
Assert.Equal(1, health.LastDeployedCount);
|
||||
}
|
||||
}
|
||||
@@ -123,9 +123,12 @@ public class InstanceActorIntegrationTests : TestKit, IDisposable
|
||||
$"corr-{i}", "Pump1", "Temperature", $"{i}", DateTimeOffset.UtcNow));
|
||||
}
|
||||
|
||||
// SetStaticAttributeCommand is fire-and-forget; the GetAttributeRequest
|
||||
// round-trip below is the sync point — the FIFO mailbox guarantees all
|
||||
// 50 sets are processed before the get is.
|
||||
// Each static write replies with a SetStaticAttributeResponse; drain all
|
||||
// 50 — the FIFO mailbox guarantees they are processed in order.
|
||||
for (int i = 0; i < 50; i++)
|
||||
{
|
||||
ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
// The last value should be the final one
|
||||
actor.Tell(new GetAttributeRequest(
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Akka.TestKit;
|
||||
using ScadaLink.Commons.Messages.DataConnection;
|
||||
using ScadaLink.Commons.Messages.Instance;
|
||||
using ScadaLink.Commons.Types.Flattening;
|
||||
using ScadaLink.SiteRuntime.Actors;
|
||||
using ScadaLink.SiteRuntime.Persistence;
|
||||
using ScadaLink.SiteRuntime.Scripts;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace ScadaLink.SiteRuntime.Tests.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// Regression tests for SiteRuntime-001: Instance.SetAttribute must route writes
|
||||
/// to the Data Connection Layer for data-sourced attributes instead of persisting
|
||||
/// a local static override.
|
||||
/// </summary>
|
||||
public class InstanceActorSetAttributeTests : TestKit, IDisposable
|
||||
{
|
||||
private readonly SiteStorageService _storage;
|
||||
private readonly ScriptCompilationService _compilationService;
|
||||
private readonly SharedScriptLibrary _sharedScriptLibrary;
|
||||
private readonly SiteRuntimeOptions _options;
|
||||
private readonly string _dbFile;
|
||||
|
||||
public InstanceActorSetAttributeTests()
|
||||
{
|
||||
_dbFile = Path.Combine(Path.GetTempPath(), $"instance-setattr-test-{Guid.NewGuid():N}.db");
|
||||
_storage = new SiteStorageService(
|
||||
$"Data Source={_dbFile}",
|
||||
NullLogger<SiteStorageService>.Instance);
|
||||
_storage.InitializeAsync().GetAwaiter().GetResult();
|
||||
_compilationService = new ScriptCompilationService(
|
||||
NullLogger<ScriptCompilationService>.Instance);
|
||||
_sharedScriptLibrary = new SharedScriptLibrary(
|
||||
_compilationService, NullLogger<SharedScriptLibrary>.Instance);
|
||||
_options = new SiteRuntimeOptions();
|
||||
}
|
||||
|
||||
void IDisposable.Dispose()
|
||||
{
|
||||
Shutdown();
|
||||
try { File.Delete(_dbFile); } catch { /* cleanup */ }
|
||||
}
|
||||
|
||||
private IActorRef CreateInstanceActor(string instanceName, FlattenedConfiguration config, IActorRef? dclManager)
|
||||
{
|
||||
return ActorOf(Props.Create(() => new InstanceActor(
|
||||
instanceName,
|
||||
JsonSerializer.Serialize(config),
|
||||
_storage,
|
||||
_compilationService,
|
||||
_sharedScriptLibrary,
|
||||
null,
|
||||
_options,
|
||||
NullLogger<InstanceActor>.Instance,
|
||||
dclManager)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Drains the startup <see cref="SubscribeTagsRequest"/> the Instance Actor emits
|
||||
/// to the DCL in PreStart, then returns the next <see cref="WriteTagRequest"/>.
|
||||
/// </summary>
|
||||
private static WriteTagRequest ExpectWriteTag(TestProbe dclProbe)
|
||||
=> dclProbe.FishForMessage<WriteTagRequest>(_ => true, TimeSpan.FromSeconds(5));
|
||||
|
||||
private static FlattenedConfiguration DataSourcedConfig(string instanceName) => new()
|
||||
{
|
||||
InstanceUniqueName = instanceName,
|
||||
Attributes =
|
||||
[
|
||||
new ResolvedAttribute
|
||||
{
|
||||
CanonicalName = "Setpoint",
|
||||
Value = "10",
|
||||
DataType = "Double",
|
||||
DataSourceReference = "/Motor/Setpoint",
|
||||
BoundDataConnectionName = "OpcServer1"
|
||||
}
|
||||
]
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public async Task SetAttribute_DataSourcedAttribute_IssuesDclWriteAndDoesNotPersistOverride()
|
||||
{
|
||||
var config = DataSourcedConfig("PumpDcl1");
|
||||
var dclProbe = CreateTestProbe();
|
||||
var actor = CreateInstanceActor("PumpDcl1", config, dclProbe.Ref);
|
||||
|
||||
actor.Tell(new SetStaticAttributeCommand(
|
||||
"corr-dcl", "PumpDcl1", "Setpoint", "55", DateTimeOffset.UtcNow));
|
||||
|
||||
// The Instance Actor must forward a WriteTagRequest to the DCL manager.
|
||||
var write = ExpectWriteTag(dclProbe);
|
||||
Assert.Equal("OpcServer1", write.ConnectionName);
|
||||
Assert.Equal("/Motor/Setpoint", write.TagPath);
|
||||
Assert.Equal("55", write.Value);
|
||||
|
||||
// DCL confirms the write.
|
||||
dclProbe.Reply(new WriteTagResponse(write.CorrelationId, true, null, DateTimeOffset.UtcNow));
|
||||
var response = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
Assert.True(response.Success);
|
||||
|
||||
// No static override should be persisted for a data-sourced attribute.
|
||||
await Task.Delay(300);
|
||||
var overrides = await _storage.GetStaticOverridesAsync("PumpDcl1");
|
||||
Assert.Empty(overrides);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SetAttribute_DataSourcedAttribute_DoesNotOptimisticallyUpdateMemory()
|
||||
{
|
||||
var config = DataSourcedConfig("PumpDcl2");
|
||||
var dclProbe = CreateTestProbe();
|
||||
var actor = CreateInstanceActor("PumpDcl2", config, dclProbe.Ref);
|
||||
|
||||
actor.Tell(new SetStaticAttributeCommand(
|
||||
"corr-dcl2", "PumpDcl2", "Setpoint", "999", DateTimeOffset.UtcNow));
|
||||
|
||||
var write = ExpectWriteTag(dclProbe);
|
||||
dclProbe.Reply(new WriteTagResponse(write.CorrelationId, true, null, DateTimeOffset.UtcNow));
|
||||
ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
|
||||
// In-memory value must still be the original config value — it is only
|
||||
// updated when the subscription delivers the confirmed device value.
|
||||
actor.Tell(new GetAttributeRequest("corr-get", "PumpDcl2", "Setpoint", DateTimeOffset.UtcNow));
|
||||
var get = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("10", get.Value?.ToString());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SetAttribute_DataSourcedAttribute_DclWriteFailure_ReturnedToCaller()
|
||||
{
|
||||
var config = DataSourcedConfig("PumpDcl3");
|
||||
var dclProbe = CreateTestProbe();
|
||||
var actor = CreateInstanceActor("PumpDcl3", config, dclProbe.Ref);
|
||||
|
||||
actor.Tell(new SetStaticAttributeCommand(
|
||||
"corr-dcl3", "PumpDcl3", "Setpoint", "42", DateTimeOffset.UtcNow));
|
||||
|
||||
var write = ExpectWriteTag(dclProbe);
|
||||
dclProbe.Reply(new WriteTagResponse(write.CorrelationId, false, "device rejected write", DateTimeOffset.UtcNow));
|
||||
|
||||
var response = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
Assert.False(response.Success);
|
||||
Assert.Contains("device rejected write", response.ErrorMessage);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SetAttribute_StaticAttribute_StillPersistsOverrideAndDoesNotCallDcl()
|
||||
{
|
||||
var config = new FlattenedConfiguration
|
||||
{
|
||||
InstanceUniqueName = "PumpStatic1",
|
||||
Attributes =
|
||||
[
|
||||
new ResolvedAttribute { CanonicalName = "Label", Value = "Main", DataType = "String" }
|
||||
]
|
||||
};
|
||||
var dclProbe = CreateTestProbe();
|
||||
var actor = CreateInstanceActor("PumpStatic1", config, dclProbe.Ref);
|
||||
|
||||
actor.Tell(new SetStaticAttributeCommand(
|
||||
"corr-static", "PumpStatic1", "Label", "Backup", DateTimeOffset.UtcNow));
|
||||
|
||||
var response = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
Assert.True(response.Success);
|
||||
|
||||
// DCL must NOT receive a write for a static attribute.
|
||||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
|
||||
await Task.Delay(300);
|
||||
var overrides = await _storage.GetStaticOverridesAsync("PumpStatic1");
|
||||
Assert.Single(overrides);
|
||||
Assert.Equal("Backup", overrides["Label"]);
|
||||
}
|
||||
}
|
||||
@@ -113,11 +113,11 @@ public class InstanceActorTests : TestKit, IDisposable
|
||||
|
||||
var actor = CreateInstanceActor("Pump1", config);
|
||||
|
||||
// SetStaticAttributeCommand is fire-and-forget (no reply); the
|
||||
// GetAttributeRequest round-trip below confirms it was applied — the
|
||||
// actor mailbox is FIFO, so the set is processed before the get.
|
||||
// A static attribute write replies with SetStaticAttributeResponse.
|
||||
actor.Tell(new SetStaticAttributeCommand(
|
||||
"corr-3", "Pump1", "Temperature", "100.0", DateTimeOffset.UtcNow));
|
||||
var setResponse = ExpectMsg<SetStaticAttributeResponse>();
|
||||
Assert.True(setResponse.Success);
|
||||
|
||||
// Verify the value changed in memory
|
||||
actor.Tell(new GetAttributeRequest(
|
||||
@@ -145,12 +145,9 @@ public class InstanceActorTests : TestKit, IDisposable
|
||||
actor.Tell(new SetStaticAttributeCommand(
|
||||
"corr-persist", "PumpPersist1", "Temperature", "100.0", DateTimeOffset.UtcNow));
|
||||
|
||||
// SetStaticAttributeCommand is fire-and-forget; round-trip a
|
||||
// GetAttributeRequest to confirm the command was processed (FIFO
|
||||
// mailbox), then wait for the async SQLite persist to complete.
|
||||
actor.Tell(new GetAttributeRequest(
|
||||
"corr-persist-get", "PumpPersist1", "Temperature", DateTimeOffset.UtcNow));
|
||||
ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
// A static attribute write replies with SetStaticAttributeResponse once the
|
||||
// in-memory state is updated; then wait for the async SQLite persist.
|
||||
ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||
await Task.Delay(500);
|
||||
|
||||
// Verify it persisted to SQLite
|
||||
|
||||
Reference in New Issue
Block a user