perf(runtime): O(1) attribute resolution, precomputed types, coalesced static writes, shared JSON options
This commit is contained in:
@@ -177,6 +177,15 @@ public class ScriptAnalysisService
|
|||||||
private const int SandboxMaxConsoleChars = 32_000;
|
private const int SandboxMaxConsoleChars = 32_000;
|
||||||
private const int SandboxMaxReturnJsonChars = 32_000;
|
private const int SandboxMaxReturnJsonChars = 32_000;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Shared options for the per-Test-Run return-value JSON serialize in
|
||||||
|
/// <see cref="SerializeReturn"/> (perf remediation, arch-review WP1.5) — avoids
|
||||||
|
/// allocating a new <see cref="JsonSerializerOptions"/> per run. Settings preserved
|
||||||
|
/// exactly.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly JsonSerializerOptions SandboxReturnJsonOptions =
|
||||||
|
new() { WriteIndented = true };
|
||||||
|
|
||||||
private const int SandboxMaxCallSharedDepth = 16;
|
private const int SandboxMaxCallSharedDepth = 16;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -520,7 +529,7 @@ public class ScriptAnalysisService
|
|||||||
var typeName = value.GetType().Name;
|
var typeName = value.GetType().Name;
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
var json = JsonSerializer.Serialize(value, new JsonSerializerOptions { WriteIndented = true });
|
var json = JsonSerializer.Serialize(value, SandboxReturnJsonOptions);
|
||||||
if (json.Length > SandboxMaxReturnJsonChars)
|
if (json.Length > SandboxMaxReturnJsonChars)
|
||||||
json = json[..SandboxMaxReturnJsonChars] + "\n… (truncated)";
|
json = json[..SandboxMaxReturnJsonChars] + "\n… (truncated)";
|
||||||
return (json, typeName);
|
return (json, typeName);
|
||||||
|
|||||||
@@ -27,6 +27,15 @@ public class ExternalSystemClient : IExternalSystemClient
|
|||||||
private readonly ILogger<ExternalSystemClient> _logger;
|
private readonly ILogger<ExternalSystemClient> _logger;
|
||||||
private readonly ExternalSystemGatewayOptions _options;
|
private readonly ExternalSystemGatewayOptions _options;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Shared options for the per-call AuthConfiguration JSON parse in
|
||||||
|
/// <see cref="TryParseJsonAuth"/> (perf remediation, arch-review WP1.5) — avoids
|
||||||
|
/// allocating a new <see cref="JsonSerializerOptions"/> (and its internal cached
|
||||||
|
/// metadata) on every outbound call. Settings preserved exactly.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly JsonSerializerOptions AuthJsonOptions =
|
||||||
|
new() { PropertyNameCaseInsensitive = true };
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the ExternalSystemClient.
|
/// Initializes a new instance of the ExternalSystemClient.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -755,7 +764,7 @@ public class ExternalSystemClient : IExternalSystemClient
|
|||||||
{
|
{
|
||||||
fields = JsonSerializer.Deserialize<Dictionary<string, string?>>(
|
fields = JsonSerializer.Deserialize<Dictionary<string, string?>>(
|
||||||
config,
|
config,
|
||||||
new JsonSerializerOptions { PropertyNameCaseInsensitive = true })
|
AuthJsonOptions)
|
||||||
?? new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase);
|
?? new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase);
|
||||||
|
|
||||||
// Normalize to case-insensitive so "Header"/"header" both resolve.
|
// Normalize to case-insensitive so "Header"/"header" both resolve.
|
||||||
|
|||||||
@@ -17,6 +17,15 @@ public static class ManagementEndpoints
|
|||||||
{
|
{
|
||||||
private static readonly TimeSpan DefaultAskTimeout = TimeSpan.FromSeconds(30);
|
private static readonly TimeSpan DefaultAskTimeout = TimeSpan.FromSeconds(30);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Shared options for the per-request command payload JSON parse in
|
||||||
|
/// <see cref="ParseCommand"/> (perf remediation, arch-review WP1.5) — avoids
|
||||||
|
/// allocating a new <see cref="JsonSerializerOptions"/> per inbound management
|
||||||
|
/// request. Settings preserved exactly.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly JsonSerializerOptions CommandPayloadJsonOptions =
|
||||||
|
new() { PropertyNameCaseInsensitive = true };
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Resolves the ManagementActor Ask timeout from configuration.
|
/// Resolves the ManagementActor Ask timeout from configuration.
|
||||||
/// Falls back to <see cref="DefaultAskTimeout"/>
|
/// Falls back to <see cref="DefaultAskTimeout"/>
|
||||||
@@ -197,7 +206,7 @@ public static class ManagementEndpoints
|
|||||||
? p.GetRawText()
|
? p.GetRawText()
|
||||||
: "{}";
|
: "{}";
|
||||||
var command = JsonSerializer.Deserialize(payloadJson, commandType,
|
var command = JsonSerializer.Deserialize(payloadJson, commandType,
|
||||||
new JsonSerializerOptions { PropertyNameCaseInsensitive = true })!;
|
CommandPayloadJsonOptions)!;
|
||||||
return CommandParseResult.Ok(command);
|
return CommandParseResult.Ok(command);
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
|
|||||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution;
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
||||||
@@ -92,8 +93,51 @@ public class InstanceActor : ReceiveActor
|
|||||||
// attribute lookup must be O(1) rather than a linear scan of
|
// attribute lookup must be O(1) rather than a linear scan of
|
||||||
// _configuration.Attributes. Built once in the constructor from the
|
// _configuration.Attributes. Built once in the constructor from the
|
||||||
// deserialized configuration (last-wins on duplicate canonical names,
|
// deserialized configuration (last-wins on duplicate canonical names,
|
||||||
// mirroring the rest of the actor's by-name dictionaries).
|
// mirroring the rest of the actor's by-name dictionaries). Each entry
|
||||||
private readonly Dictionary<string, ResolvedAttribute> _resolvedAttributeByName = new();
|
// additionally carries its DataType/ElementDataType parsed ONCE at
|
||||||
|
// config-apply time (perf remediation, arch-review WP1.5) so the hot
|
||||||
|
// TagValueUpdate path never re-Enum.TryParse's a string per update.
|
||||||
|
private readonly Dictionary<string, AttributeIndexEntry> _resolvedAttributeByName = new();
|
||||||
|
|
||||||
|
// Static-override write coalescing (perf remediation, arch-review WP1.5):
|
||||||
|
// mirrors NativeAlarmActor's MarkDirtyUpsert/FlushDirtyUpserts (P4) shape.
|
||||||
|
// A burst of rapid SetAttribute calls on the same or different attributes
|
||||||
|
// collapses into one batched SQLite transaction on a short single-shot
|
||||||
|
// timer instead of a write per call. Keyed by attribute name — only the
|
||||||
|
// LATEST value per attribute survives a coalescing window, which matches
|
||||||
|
// the "in-memory state is authoritative" fire-and-forget persistence model
|
||||||
|
// already documented on the single-write path this replaces.
|
||||||
|
private readonly Dictionary<string, string> _dirtyStaticOverrides = new();
|
||||||
|
private ICancelable? _staticOverrideFlushTimer;
|
||||||
|
private readonly TimeSpan _staticOverrideFlushInterval;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Cached string forms of <see cref="QualityCode"/> (perf remediation, arch-review
|
||||||
|
/// WP1.5) — the TagValueUpdate ingest path calls this once or twice per update, so a
|
||||||
|
/// static lookup replaces repeated <see cref="Enum.ToString()"/> reflection.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly Dictionary<QualityCode, string> QualityStrings = new()
|
||||||
|
{
|
||||||
|
[QualityCode.Good] = "Good",
|
||||||
|
[QualityCode.Bad] = "Bad",
|
||||||
|
[QualityCode.Uncertain] = "Uncertain",
|
||||||
|
};
|
||||||
|
|
||||||
|
private static string QualityToString(QualityCode quality) =>
|
||||||
|
QualityStrings.TryGetValue(quality, out var s) ? s : quality.ToString();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// O(1) attribute-resolution index entry (perf remediation, arch-review WP1.5): wraps a
|
||||||
|
/// <see cref="ResolvedAttribute"/> together with its <see cref="DataType"/> /
|
||||||
|
/// element <see cref="DataType"/> / IsList classification, parsed ONCE when the entry
|
||||||
|
/// is built (constructor) instead of re-parsed via <see cref="Enum.TryParse{DataType}"/>
|
||||||
|
/// on every access.
|
||||||
|
/// </summary>
|
||||||
|
private sealed record AttributeIndexEntry(
|
||||||
|
ResolvedAttribute Attribute,
|
||||||
|
DataType DataType,
|
||||||
|
DataType? ElementDataType,
|
||||||
|
bool IsList);
|
||||||
|
|
||||||
// WaitForAttribute (spec §4.2): one-shot waiter registry keyed by the
|
// WaitForAttribute (spec §4.2): one-shot waiter registry keyed by the
|
||||||
// request CorrelationId. Each entry holds the watched attribute name, the
|
// request CorrelationId. Each entry holds the watched attribute name, the
|
||||||
@@ -149,7 +193,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
ILogger logger,
|
ILogger logger,
|
||||||
IActorRef? dclManager = null,
|
IActorRef? dclManager = null,
|
||||||
ISiteHealthCollector? healthCollector = null,
|
ISiteHealthCollector? healthCollector = null,
|
||||||
IServiceProvider? serviceProvider = null)
|
IServiceProvider? serviceProvider = null,
|
||||||
|
TimeSpan? staticOverrideFlushInterval = null)
|
||||||
{
|
{
|
||||||
_instanceUniqueName = instanceUniqueName;
|
_instanceUniqueName = instanceUniqueName;
|
||||||
_storage = storage;
|
_storage = storage;
|
||||||
@@ -161,6 +206,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
_dclManager = dclManager;
|
_dclManager = dclManager;
|
||||||
_healthCollector = healthCollector;
|
_healthCollector = healthCollector;
|
||||||
_serviceProvider = serviceProvider;
|
_serviceProvider = serviceProvider;
|
||||||
|
// Defaults to 100 ms, mirroring NativeAlarmActor's persistFlushInterval (P4).
|
||||||
|
_staticOverrideFlushInterval = staticOverrideFlushInterval ?? TimeSpan.FromMilliseconds(100);
|
||||||
|
|
||||||
// Deserialize the flattened configuration
|
// Deserialize the flattened configuration
|
||||||
_configuration = JsonSerializer.Deserialize<FlattenedConfiguration>(configJson);
|
_configuration = JsonSerializer.Deserialize<FlattenedConfiguration>(configJson);
|
||||||
@@ -172,18 +219,23 @@ public class InstanceActor : ReceiveActor
|
|||||||
{
|
{
|
||||||
foreach (var attr in _configuration.Attributes)
|
foreach (var attr in _configuration.Attributes)
|
||||||
{
|
{
|
||||||
// Index resolved attributes for O(1) lookup on the hot
|
// Parse DataType/ElementDataType ONCE here (config-apply time) and index
|
||||||
// TagValueUpdate ingest path (last-wins on duplicate names).
|
// the wrapping entry for O(1) lookup on the hot TagValueUpdate ingest path
|
||||||
_resolvedAttributeByName[attr.CanonicalName] = attr;
|
// (last-wins on duplicate names) — perf remediation, arch-review WP1.5.
|
||||||
|
var dataType = ParseDataType(attr.DataType);
|
||||||
|
var elementDataType = ParseElementDataType(attr.ElementDataType);
|
||||||
|
var isList = dataType == DataType.List;
|
||||||
|
_resolvedAttributeByName[attr.CanonicalName] =
|
||||||
|
new AttributeIndexEntry(attr, dataType, elementDataType, isList);
|
||||||
|
|
||||||
// A STATIC List attribute's default is the canonical JSON
|
// A STATIC List attribute's default is the canonical JSON
|
||||||
// array string. Decode it to a typed List<T> for in-memory reads
|
// array string. Decode it to a typed List<T> for in-memory reads
|
||||||
// so scripts see a real collection. Scalars store their raw
|
// so scripts see a real collection. Scalars store their raw
|
||||||
// string unchanged. A malformed List default decodes to null and
|
// string unchanged. A malformed List default decodes to null and
|
||||||
// is marked Bad quality rather than crashing the actor.
|
// is marked Bad quality rather than crashing the actor.
|
||||||
if (IsListAttribute(attr))
|
if (isList)
|
||||||
{
|
{
|
||||||
var decoded = DecodeAttributeValue(attr, attr.Value);
|
var decoded = DecodeAttributeValue(dataType, elementDataType, attr.Value, attr.CanonicalName);
|
||||||
_attributes[attr.CanonicalName] = decoded;
|
_attributes[attr.CanonicalName] = decoded;
|
||||||
_attributeQualities[attr.CanonicalName] =
|
_attributeQualities[attr.CanonicalName] =
|
||||||
decoded is null && !string.IsNullOrEmpty(attr.Value) ? "Bad"
|
decoded is null && !string.IsNullOrEmpty(attr.Value) ? "Bad"
|
||||||
@@ -257,6 +309,10 @@ public class InstanceActor : ReceiveActor
|
|||||||
|
|
||||||
// Handle internal messages
|
// Handle internal messages
|
||||||
Receive<LoadOverridesResult>(HandleOverridesLoaded);
|
Receive<LoadOverridesResult>(HandleOverridesLoaded);
|
||||||
|
|
||||||
|
// Coalesced flush of buffered static-override writes (P4 pattern, see
|
||||||
|
// NativeAlarmActor.FlushDirtyUpserts).
|
||||||
|
Receive<FlushStaticOverrides>(_ => FlushDirtyStaticOverrides());
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
@@ -300,6 +356,13 @@ public class InstanceActor : ReceiveActor
|
|||||||
timer.Cancel();
|
timer.Cancel();
|
||||||
_tagSubscribeRetryTimers.Clear();
|
_tagSubscribeRetryTimers.Clear();
|
||||||
|
|
||||||
|
// Best-effort final flush of any coalesced static-override writes so a graceful
|
||||||
|
// stop does not lose the last <= flush-interval of writes (P4 pattern, mirrors
|
||||||
|
// NativeAlarmActor.PostStop). Fire-and-forget — the in-memory attribute value
|
||||||
|
// (already applied) remains authoritative even if this final persist is lost.
|
||||||
|
_staticOverrideFlushTimer?.Cancel();
|
||||||
|
FlushDirtyStaticOverrides();
|
||||||
|
|
||||||
// Operational `instance_lifecycle` event — instance stopped. An
|
// Operational `instance_lifecycle` event — instance stopped. An
|
||||||
// instance stops on disable, delete, redeployment, and graceful shutdown;
|
// instance stops on disable, delete, redeployment, and graceful shutdown;
|
||||||
// this single point covers them all.
|
// this single point covers them all.
|
||||||
@@ -369,9 +432,9 @@ public class InstanceActor : ReceiveActor
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private void HandleSetStaticAttribute(SetStaticAttributeCommand command)
|
private void HandleSetStaticAttribute(SetStaticAttributeCommand command)
|
||||||
{
|
{
|
||||||
// Resolve the target attribute's data binding from the flattened config.
|
// O(1) lookup via the existing hot-path index instead of a linear scan over
|
||||||
var resolved = _configuration?.Attributes
|
// _configuration.Attributes (perf remediation, arch-review WP1.5).
|
||||||
.FirstOrDefault(a => a.CanonicalName == command.AttributeName);
|
_resolvedAttributeByName.TryGetValue(command.AttributeName, out var resolved);
|
||||||
|
|
||||||
// Reject writes targeting an attribute that does not exist
|
// Reject writes targeting an attribute that does not exist
|
||||||
// on the deployed instance. Without this check, an inbound API
|
// on the deployed instance. Without this check, an inbound API
|
||||||
@@ -398,8 +461,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
}
|
}
|
||||||
|
|
||||||
var isDataSourced =
|
var isDataSourced =
|
||||||
!string.IsNullOrEmpty(resolved.DataSourceReference)
|
!string.IsNullOrEmpty(resolved.Attribute.DataSourceReference)
|
||||||
&& !string.IsNullOrEmpty(resolved.BoundDataConnectionName);
|
&& !string.IsNullOrEmpty(resolved.Attribute.BoundDataConnectionName);
|
||||||
|
|
||||||
if (isDataSourced)
|
if (isDataSourced)
|
||||||
{
|
{
|
||||||
@@ -423,7 +486,7 @@ public class InstanceActor : ReceiveActor
|
|||||||
// store the string verbatim. (HandleSetStaticAttribute already rejected
|
// store the string verbatim. (HandleSetStaticAttribute already rejected
|
||||||
// unknown attributes, so resolved is non-null here, but guard defensively.)
|
// unknown attributes, so resolved is non-null here, but guard defensively.)
|
||||||
if (_resolvedAttributeByName.TryGetValue(command.AttributeName, out var resolved)
|
if (_resolvedAttributeByName.TryGetValue(command.AttributeName, out var resolved)
|
||||||
&& IsListAttribute(resolved))
|
&& resolved.IsList)
|
||||||
{
|
{
|
||||||
// The script path pre-encodes valid canonical JSON via ScopeAccessors,
|
// The script path pre-encodes valid canonical JSON via ScopeAccessors,
|
||||||
// but the Inbound API / direct-command path can submit an arbitrary
|
// but the Inbound API / direct-command path can submit an arbitrary
|
||||||
@@ -435,7 +498,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
// (valid — clearing) AND a malformed non-empty input (invalid). Only the
|
// (valid — clearing) AND a malformed non-empty input (invalid). Only the
|
||||||
// latter is rejected, hence the explicit IsNullOrWhiteSpace guard. An empty
|
// latter is rejected, hence the explicit IsNullOrWhiteSpace guard. An empty
|
||||||
// list "[]" decodes to a non-null empty List, so it passes through.
|
// list "[]" decodes to a non-null empty List, so it passes through.
|
||||||
var decoded = DecodeAttributeValue(resolved, command.Value);
|
var decoded = DecodeAttributeValue(
|
||||||
|
resolved.DataType, resolved.ElementDataType, command.Value, command.AttributeName);
|
||||||
if (!string.IsNullOrWhiteSpace(command.Value) && decoded == null)
|
if (!string.IsNullOrWhiteSpace(command.Value) && decoded == null)
|
||||||
{
|
{
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
@@ -474,32 +538,67 @@ public class InstanceActor : ReceiveActor
|
|||||||
|
|
||||||
PublishAndNotifyChildren(changed);
|
PublishAndNotifyChildren(changed);
|
||||||
|
|
||||||
// Persist asynchronously -- fire and forget since the actor is the source of truth.
|
// Coalesce the persist: record the latest value and arm the short batched-flush
|
||||||
var instanceName = _instanceUniqueName;
|
// timer (P4 pattern) instead of firing a SQLite write per call — the actor's
|
||||||
var attributeName = command.AttributeName;
|
// in-memory state (already updated above) remains the source of truth regardless
|
||||||
var logger = _logger;
|
// of when the buffered write lands.
|
||||||
_storage.SetStaticOverrideAsync(_instanceUniqueName, command.AttributeName, command.Value)
|
MarkDirtyStaticOverride(command.AttributeName, command.Value);
|
||||||
.ContinueWith(t =>
|
|
||||||
{
|
|
||||||
logger.LogWarning(
|
|
||||||
t.Exception?.GetBaseException(),
|
|
||||||
"Failed to persist static override for {Instance}.{Attribute}; in-memory state is authoritative",
|
|
||||||
instanceName,
|
|
||||||
attributeName);
|
|
||||||
}, TaskContinuationOptions.OnlyOnFaulted);
|
|
||||||
|
|
||||||
Sender.Tell(new SetStaticAttributeResponse(
|
Sender.Tell(new SetStaticAttributeResponse(
|
||||||
command.CorrelationId, _instanceUniqueName, command.AttributeName,
|
command.CorrelationId, _instanceUniqueName, command.AttributeName,
|
||||||
true, null, DateTimeOffset.UtcNow));
|
true, null, DateTimeOffset.UtcNow));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Records the latest value per attribute name and arms the coalesced flush timer
|
||||||
|
/// (perf remediation, arch-review WP1.5) — mirrors NativeAlarmActor's
|
||||||
|
/// MarkDirtyUpsert/FlushDirtyUpserts (P4) shape. A burst of rapid static writes
|
||||||
|
/// collapses into one batched SQLite transaction instead of a write per call.
|
||||||
|
/// </summary>
|
||||||
|
private void MarkDirtyStaticOverride(string attributeName, string value)
|
||||||
|
{
|
||||||
|
_dirtyStaticOverrides[attributeName] = value;
|
||||||
|
if (_staticOverrideFlushTimer == null)
|
||||||
|
{
|
||||||
|
_staticOverrideFlushTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
|
||||||
|
_staticOverrideFlushInterval, Self, FlushStaticOverrides.Instance, Self);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Persists all coalesced static-override writes in one batched transaction (P4),
|
||||||
|
/// then disarms the flush timer. Fire-and-forget with OnlyOnFaulted logging — a
|
||||||
|
/// failed write never blocks the actor; the in-memory attribute value (already
|
||||||
|
/// applied) remains authoritative.
|
||||||
|
/// </summary>
|
||||||
|
private void FlushDirtyStaticOverrides()
|
||||||
|
{
|
||||||
|
_staticOverrideFlushTimer = null;
|
||||||
|
if (_dirtyStaticOverrides.Count == 0)
|
||||||
|
return;
|
||||||
|
|
||||||
|
var batch = _dirtyStaticOverrides
|
||||||
|
.Select(kv => (AttributeName: kv.Key, Value: kv.Value))
|
||||||
|
.ToList();
|
||||||
|
_dirtyStaticOverrides.Clear();
|
||||||
|
|
||||||
|
var instanceName = _instanceUniqueName;
|
||||||
|
var logger = _logger;
|
||||||
|
_storage.SetStaticOverridesAsync(instanceName, batch)
|
||||||
|
.ContinueWith(t => logger.LogWarning(
|
||||||
|
t.Exception?.GetBaseException(),
|
||||||
|
"Failed to persist {Count} static overrides for {Instance}; in-memory state is authoritative",
|
||||||
|
batch.Count, instanceName),
|
||||||
|
TaskContinuationOptions.OnlyOnFaulted);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Data-sourced attribute write: forwards a write request to the DCL and pipes
|
/// 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
|
/// 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);
|
/// untouched (it is refreshed by the subscription when the device confirms);
|
||||||
/// no static override is persisted for a data-sourced attribute.
|
/// no static override is persisted for a data-sourced attribute.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private void HandleSetDataAttribute(SetStaticAttributeCommand command, ResolvedAttribute resolved)
|
private void HandleSetDataAttribute(SetStaticAttributeCommand command, AttributeIndexEntry resolved)
|
||||||
{
|
{
|
||||||
var caller = Sender;
|
var caller = Sender;
|
||||||
var correlationId = command.CorrelationId;
|
var correlationId = command.CorrelationId;
|
||||||
@@ -526,9 +625,10 @@ public class InstanceActor : ReceiveActor
|
|||||||
// write rather than forward garbage to the device (mirrors the static-path
|
// write rather than forward garbage to the device (mirrors the static-path
|
||||||
// rejection in HandleSetStaticAttributeCore). Scalars are unchanged.
|
// rejection in HandleSetStaticAttributeCore). Scalars are unchanged.
|
||||||
object? writeValue = command.Value;
|
object? writeValue = command.Value;
|
||||||
if (IsListAttribute(resolved) && !string.IsNullOrWhiteSpace(command.Value))
|
if (resolved.IsList && !string.IsNullOrWhiteSpace(command.Value))
|
||||||
{
|
{
|
||||||
var decoded = DecodeAttributeValue(resolved, command.Value);
|
var decoded = DecodeAttributeValue(
|
||||||
|
resolved.DataType, resolved.ElementDataType, command.Value, attributeName);
|
||||||
if (decoded == null)
|
if (decoded == null)
|
||||||
{
|
{
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
@@ -545,8 +645,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
|
|
||||||
var writeRequest = new WriteTagRequest(
|
var writeRequest = new WriteTagRequest(
|
||||||
correlationId,
|
correlationId,
|
||||||
resolved.BoundDataConnectionName!,
|
resolved.Attribute.BoundDataConnectionName!,
|
||||||
resolved.DataSourceReference!,
|
resolved.Attribute.DataSourceReference!,
|
||||||
writeValue,
|
writeValue,
|
||||||
DateTimeOffset.UtcNow);
|
DateTimeOffset.UtcNow);
|
||||||
|
|
||||||
@@ -598,8 +698,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
foreach (var kv in request.AttributeEncodedValues)
|
foreach (var kv in request.AttributeEncodedValues)
|
||||||
{
|
{
|
||||||
if (!_resolvedAttributeByName.TryGetValue(kv.Key, out var resolved)
|
if (!_resolvedAttributeByName.TryGetValue(kv.Key, out var resolved)
|
||||||
|| string.IsNullOrEmpty(resolved.DataSourceReference)
|
|| string.IsNullOrEmpty(resolved.Attribute.DataSourceReference)
|
||||||
|| string.IsNullOrEmpty(resolved.BoundDataConnectionName))
|
|| string.IsNullOrEmpty(resolved.Attribute.BoundDataConnectionName))
|
||||||
{
|
{
|
||||||
caller.Tell(new WriteAttributeBatchResponse(
|
caller.Tell(new WriteAttributeBatchResponse(
|
||||||
cid, false, $"Attribute '{kv.Key}' is not a data-sourced attribute."));
|
cid, false, $"Attribute '{kv.Key}' is not a data-sourced attribute."));
|
||||||
@@ -607,8 +707,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (connName == null)
|
if (connName == null)
|
||||||
connName = resolved.BoundDataConnectionName;
|
connName = resolved.Attribute.BoundDataConnectionName;
|
||||||
else if (!string.Equals(connName, resolved.BoundDataConnectionName, StringComparison.Ordinal))
|
else if (!string.Equals(connName, resolved.Attribute.BoundDataConnectionName, StringComparison.Ordinal))
|
||||||
{
|
{
|
||||||
caller.Tell(new WriteAttributeBatchResponse(
|
caller.Tell(new WriteAttributeBatchResponse(
|
||||||
cid, false, "Batch write spans multiple data connections; not supported."));
|
cid, false, "Batch write spans multiple data connections; not supported."));
|
||||||
@@ -619,9 +719,10 @@ public class InstanceActor : ReceiveActor
|
|||||||
// MV: a data-sourced List attribute's encoded value is the canonical JSON
|
// MV: a data-sourced List attribute's encoded value is the canonical JSON
|
||||||
// array string — decode it to a typed List<T> so the DCL/Variant write
|
// array string — decode it to a typed List<T> so the DCL/Variant write
|
||||||
// produces a real array (same poison-rejection rule as HandleSetDataAttribute).
|
// produces a real array (same poison-rejection rule as HandleSetDataAttribute).
|
||||||
if (IsListAttribute(resolved) && !string.IsNullOrWhiteSpace(kv.Value))
|
if (resolved.IsList && !string.IsNullOrWhiteSpace(kv.Value))
|
||||||
{
|
{
|
||||||
var decoded = DecodeAttributeValue(resolved, kv.Value);
|
var decoded = DecodeAttributeValue(
|
||||||
|
resolved.DataType, resolved.ElementDataType, kv.Value, kv.Key);
|
||||||
if (decoded == null)
|
if (decoded == null)
|
||||||
{
|
{
|
||||||
caller.Tell(new WriteAttributeBatchResponse(
|
caller.Tell(new WriteAttributeBatchResponse(
|
||||||
@@ -631,7 +732,7 @@ public class InstanceActor : ReceiveActor
|
|||||||
wv = decoded;
|
wv = decoded;
|
||||||
}
|
}
|
||||||
|
|
||||||
values[resolved.DataSourceReference!] = wv;
|
values[resolved.Attribute.DataSourceReference!] = wv;
|
||||||
}
|
}
|
||||||
|
|
||||||
string? triggerPath = null;
|
string? triggerPath = null;
|
||||||
@@ -639,23 +740,23 @@ public class InstanceActor : ReceiveActor
|
|||||||
if (!string.IsNullOrEmpty(request.TriggerAttribute))
|
if (!string.IsNullOrEmpty(request.TriggerAttribute))
|
||||||
{
|
{
|
||||||
if (!_resolvedAttributeByName.TryGetValue(request.TriggerAttribute, out var tr)
|
if (!_resolvedAttributeByName.TryGetValue(request.TriggerAttribute, out var tr)
|
||||||
|| string.IsNullOrEmpty(tr.DataSourceReference)
|
|| string.IsNullOrEmpty(tr.Attribute.DataSourceReference)
|
||||||
|| string.IsNullOrEmpty(tr.BoundDataConnectionName))
|
|| string.IsNullOrEmpty(tr.Attribute.BoundDataConnectionName))
|
||||||
{
|
{
|
||||||
caller.Tell(new WriteAttributeBatchResponse(
|
caller.Tell(new WriteAttributeBatchResponse(
|
||||||
cid, false, $"Trigger attribute '{request.TriggerAttribute}' is not data-sourced."));
|
cid, false, $"Trigger attribute '{request.TriggerAttribute}' is not data-sourced."));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (connName != null && !string.Equals(connName, tr.BoundDataConnectionName, StringComparison.Ordinal))
|
if (connName != null && !string.Equals(connName, tr.Attribute.BoundDataConnectionName, StringComparison.Ordinal))
|
||||||
{
|
{
|
||||||
caller.Tell(new WriteAttributeBatchResponse(
|
caller.Tell(new WriteAttributeBatchResponse(
|
||||||
cid, false, "Trigger attribute is on a different data connection."));
|
cid, false, "Trigger attribute is on a different data connection."));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
connName ??= tr.BoundDataConnectionName;
|
connName ??= tr.Attribute.BoundDataConnectionName;
|
||||||
triggerPath = tr.DataSourceReference;
|
triggerPath = tr.Attribute.DataSourceReference;
|
||||||
triggerVal = request.TriggerEncodedValue;
|
triggerVal = request.TriggerEncodedValue;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -846,19 +947,19 @@ public class InstanceActor : ReceiveActor
|
|||||||
// (a CLR array/IEnumerable from the SDK) into a typed List<T>. On an
|
// (a CLR array/IEnumerable from the SDK) into a typed List<T>. On an
|
||||||
// element-type mismatch we set the attribute's quality to Bad, log a
|
// element-type mismatch we set the attribute's quality to Bad, log a
|
||||||
// warning, and skip storing a value rather than crashing the actor.
|
// warning, and skip storing a value rather than crashing the actor.
|
||||||
if (resolved != null && IsListAttribute(resolved))
|
if (resolved != null && resolved.IsList)
|
||||||
{
|
{
|
||||||
if (TryCoerceListValue(resolved, update.Value, out var typedList))
|
if (TryCoerceListValue(resolved.ElementDataType, update.Value, out var typedList))
|
||||||
{
|
{
|
||||||
HandleAttributeValueChanged(new AttributeValueChanged(
|
HandleAttributeValueChanged(new AttributeValueChanged(
|
||||||
_instanceUniqueName, update.TagPath, attrName,
|
_instanceUniqueName, update.TagPath, attrName,
|
||||||
typedList, update.Quality.ToString(), update.Timestamp));
|
typedList, QualityToString(update.Quality), update.Timestamp));
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
"List attribute {Instance}.{Attribute} received a value that could not be coerced to List<{Element}>; marking quality Bad",
|
"List attribute {Instance}.{Attribute} received a value that could not be coerced to List<{Element}>; marking quality Bad",
|
||||||
_instanceUniqueName, attrName, resolved.ElementDataType);
|
_instanceUniqueName, attrName, resolved.Attribute.ElementDataType);
|
||||||
_attributeQualities[attrName] = "Bad";
|
_attributeQualities[attrName] = "Bad";
|
||||||
_attributeTimestamps[attrName] = update.Timestamp;
|
_attributeTimestamps[attrName] = update.Timestamp;
|
||||||
var currentValue = _attributes.GetValueOrDefault(attrName);
|
var currentValue = _attributes.GetValueOrDefault(attrName);
|
||||||
@@ -883,14 +984,21 @@ public class InstanceActor : ReceiveActor
|
|||||||
|
|
||||||
HandleAttributeValueChanged(new AttributeValueChanged(
|
HandleAttributeValueChanged(new AttributeValueChanged(
|
||||||
_instanceUniqueName, update.TagPath, attrName,
|
_instanceUniqueName, update.TagPath, attrName,
|
||||||
value, update.Quality.ToString(), update.Timestamp));
|
value, QualityToString(update.Quality), update.Timestamp));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>True if the resolved attribute is declared as a <see cref="DataType.List"/>.</summary>
|
/// <summary>Parses a <see cref="ResolvedAttribute.DataType"/> string, defaulting to
|
||||||
private static bool IsListAttribute(ResolvedAttribute attr) =>
|
/// <see cref="DataType.String"/> on an unrecognized value (never throws).</summary>
|
||||||
Enum.TryParse<DataType>(attr.DataType, ignoreCase: true, out var dt)
|
private static DataType ParseDataType(string dataType) =>
|
||||||
&& dt == DataType.List;
|
Enum.TryParse<DataType>(dataType, ignoreCase: true, out var dt) ? dt : DataType.String;
|
||||||
|
|
||||||
|
/// <summary>Parses a <see cref="ResolvedAttribute.ElementDataType"/> string, returning
|
||||||
|
/// <see langword="null"/> when absent or unrecognized (never throws).</summary>
|
||||||
|
private static DataType? ParseElementDataType(string? elementDataType) =>
|
||||||
|
string.IsNullOrEmpty(elementDataType)
|
||||||
|
? null
|
||||||
|
: (Enum.TryParse<DataType>(elementDataType, ignoreCase: true, out var et) ? et : (DataType?)null);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Decodes a STATIC (authored / overridden) attribute's canonical value
|
/// Decodes a STATIC (authored / overridden) attribute's canonical value
|
||||||
@@ -903,16 +1011,15 @@ public class InstanceActor : ReceiveActor
|
|||||||
/// bad element, missing element type) degrades to <see langword="null"/> + a
|
/// bad element, missing element type) degrades to <see langword="null"/> + a
|
||||||
/// warning — the caller marks the attribute Bad quality. NEVER throws into the
|
/// warning — the caller marks the attribute Bad quality. NEVER throws into the
|
||||||
/// actor.
|
/// actor.
|
||||||
|
///
|
||||||
|
/// Takes the already-parsed <paramref name="dataType"/>/<paramref name="elementType"/>
|
||||||
|
/// (perf remediation, arch-review WP1.5) rather than re-parsing from the attribute's
|
||||||
|
/// raw DataType strings — every call site now resolves these once via
|
||||||
|
/// <see cref="AttributeIndexEntry"/> at config-apply time.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private object? DecodeAttributeValue(ResolvedAttribute attr, string? raw)
|
private object? DecodeAttributeValue(
|
||||||
|
DataType dataType, DataType? elementType, string? raw, string attributeCanonicalName)
|
||||||
{
|
{
|
||||||
DataType dataType = Enum.TryParse<DataType>(attr.DataType, ignoreCase: true, out var dt)
|
|
||||||
? dt
|
|
||||||
: DataType.String;
|
|
||||||
DataType? elementType = string.IsNullOrEmpty(attr.ElementDataType)
|
|
||||||
? null
|
|
||||||
: (Enum.TryParse<DataType>(attr.ElementDataType, ignoreCase: true, out var et) ? et : null);
|
|
||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
return AttributeValueCodec.Decode(raw, dataType, elementType);
|
return AttributeValueCodec.Decode(raw, dataType, elementType);
|
||||||
@@ -921,30 +1028,28 @@ public class InstanceActor : ReceiveActor
|
|||||||
{
|
{
|
||||||
_logger.LogWarning(ex,
|
_logger.LogWarning(ex,
|
||||||
"Attribute '{Attr}' on '{Instance}' has an undecodable List value; marking Bad quality",
|
"Attribute '{Attr}' on '{Instance}' has an undecodable List value; marking Bad quality",
|
||||||
attr.CanonicalName, _instanceUniqueName);
|
attributeCanonicalName, _instanceUniqueName);
|
||||||
return null; // caller marks quality Bad
|
return null; // caller marks quality Bad
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Coerces an incoming data-sourced value (an OPC UA array / IEnumerable)
|
/// Coerces an incoming data-sourced value (an OPC UA array / IEnumerable)
|
||||||
/// into a typed <c>List<elementClrType></c> matching the attribute's
|
/// into a typed <c>List<elementClrType></c> matching the attribute's element
|
||||||
/// <see cref="ResolvedAttribute.ElementDataType"/>. Each element is converted
|
/// type. Each element is converted with invariant culture (round-trip parse for
|
||||||
/// with invariant culture (round-trip parse for DateTime). Returns
|
/// DateTime). Returns <see langword="false"/> on a missing/invalid element type, a
|
||||||
/// <see langword="false"/> on a missing/invalid element type, a non-enumerable
|
/// non-enumerable value, or any element that cannot be coerced — the caller then
|
||||||
/// value, or any element that cannot be coerced — the caller then marks the
|
/// marks the attribute quality Bad. Never throws.
|
||||||
/// attribute quality Bad. Never throws.
|
///
|
||||||
|
/// Takes the already-parsed <paramref name="elementType"/> (perf remediation,
|
||||||
|
/// arch-review WP1.5) — see <see cref="DecodeAttributeValue"/>.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private bool TryCoerceListValue(ResolvedAttribute attr, object? incoming, out object? typedList)
|
private bool TryCoerceListValue(DataType? elementType, object? incoming, out object? typedList)
|
||||||
{
|
{
|
||||||
typedList = null;
|
typedList = null;
|
||||||
|
|
||||||
if (string.IsNullOrEmpty(attr.ElementDataType)
|
if (elementType is not { } et || !AttributeValueCodec.IsValidElementType(et))
|
||||||
|| !Enum.TryParse<DataType>(attr.ElementDataType, ignoreCase: true, out var elementType)
|
|
||||||
|| !AttributeValueCodec.IsValidElementType(elementType))
|
|
||||||
{
|
|
||||||
return false;
|
return false;
|
||||||
}
|
|
||||||
|
|
||||||
if (incoming is not System.Collections.IEnumerable enumerable || incoming is string)
|
if (incoming is not System.Collections.IEnumerable enumerable || incoming is string)
|
||||||
return false;
|
return false;
|
||||||
@@ -956,7 +1061,7 @@ public class InstanceActor : ReceiveActor
|
|||||||
// inside the guarded block means any future change that introduces a
|
// inside the guarded block means any future change that introduces a
|
||||||
// throw is caught and turned into a Bad-quality result rather than
|
// throw is caught and turned into a Bad-quality result rather than
|
||||||
// escaping into the actor and tripping supervision.
|
// escaping into the actor and tripping supervision.
|
||||||
typedList = AttributeValueCodec.CoerceEnumerable(enumerable, elementType);
|
typedList = AttributeValueCodec.CoerceEnumerable(enumerable, et);
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
@@ -964,7 +1069,7 @@ public class InstanceActor : ReceiveActor
|
|||||||
// Any coercion / construction failure → Bad quality, never a crash.
|
// Any coercion / construction failure → Bad quality, never a crash.
|
||||||
_logger.LogWarning(ex,
|
_logger.LogWarning(ex,
|
||||||
"Failed to coerce value to List<{Element}> for instance {Instance}; marking quality Bad",
|
"Failed to coerce value to List<{Element}> for instance {Instance}; marking quality Bad",
|
||||||
attr.ElementDataType, _instanceUniqueName);
|
et, _instanceUniqueName);
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1417,7 +1522,7 @@ public class InstanceActor : ReceiveActor
|
|||||||
// overrides to a typed list (matching the config-default load), set
|
// overrides to a typed list (matching the config-default load), set
|
||||||
// Bad quality on a malformed stored value, and never crash the actor.
|
// Bad quality on a malformed stored value, and never crash the actor.
|
||||||
if (_resolvedAttributeByName.TryGetValue(kvp.Key, out var resolved)
|
if (_resolvedAttributeByName.TryGetValue(kvp.Key, out var resolved)
|
||||||
&& IsListAttribute(resolved))
|
&& resolved.IsList)
|
||||||
{
|
{
|
||||||
// Decode the stored List override (both old array-of-strings
|
// Decode the stored List override (both old array-of-strings
|
||||||
// and native-typed forms decode) and re-persist the native form if
|
// and native-typed forms decode) and re-persist the native form if
|
||||||
@@ -1425,7 +1530,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
// list and comparing to the stored string detects old-form values
|
// list and comparing to the stored string detects old-form values
|
||||||
// (native → native is byte-identical, so a native value is a no-op).
|
// (native → native is byte-identical, so a native value is a no-op).
|
||||||
// The re-persist is fire-and-forget and never throws into the actor.
|
// The re-persist is fire-and-forget and never throws into the actor.
|
||||||
var decoded = DecodeAttributeValue(resolved, kvp.Value);
|
var decoded = DecodeAttributeValue(
|
||||||
|
resolved.DataType, resolved.ElementDataType, kvp.Value, kvp.Key);
|
||||||
_attributes[kvp.Key] = decoded;
|
_attributes[kvp.Key] = decoded;
|
||||||
if (decoded is null && !string.IsNullOrEmpty(kvp.Value))
|
if (decoded is null && !string.IsNullOrEmpty(kvp.Value))
|
||||||
{
|
{
|
||||||
@@ -1742,4 +1848,11 @@ public class InstanceActor : ReceiveActor
|
|||||||
/// connection whose previous attempt failed or whose response was lost.
|
/// connection whose previous attempt failed or whose response was lost.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private sealed record RetryTagSubscribe(string ConnectionName);
|
private sealed record RetryTagSubscribe(string ConnectionName);
|
||||||
|
|
||||||
|
/// <summary>Self-tell that fires the coalesced static-override persist flush (P4, arch-review WP1.5).</summary>
|
||||||
|
private sealed class FlushStaticOverrides
|
||||||
|
{
|
||||||
|
public static readonly FlushStaticOverrides Instance = new();
|
||||||
|
private FlushStaticOverrides() { }
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -345,6 +345,53 @@ public class SiteStorageService
|
|||||||
await command.ExecuteNonQueryAsync();
|
await command.ExecuteNonQueryAsync();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Batch counterpart of <see cref="SetStaticOverrideAsync"/> (perf remediation,
|
||||||
|
/// arch-review WP1.5): upserts a set of static attribute overrides for one instance
|
||||||
|
/// in a SINGLE connection + transaction with one prepared command re-bound per row.
|
||||||
|
/// Used by the InstanceActor's coalesced flush — mirrors NativeAlarmActor's
|
||||||
|
/// MarkDirtyUpsert/FlushDirtyUpserts (P4) pattern — so a burst of rapid SetAttribute
|
||||||
|
/// calls on an instance collapses into one batched write instead of a write per call.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="instanceName">The unique name of the instance.</param>
|
||||||
|
/// <param name="overrides">The attribute name/value pairs to upsert (latest value per name).</param>
|
||||||
|
/// <returns>A task that completes when all rows have been committed.</returns>
|
||||||
|
public async Task SetStaticOverridesAsync(
|
||||||
|
string instanceName, IReadOnlyList<(string AttributeName, string Value)> overrides)
|
||||||
|
{
|
||||||
|
if (overrides.Count == 0)
|
||||||
|
return;
|
||||||
|
|
||||||
|
await using var connection = OpenConnection();
|
||||||
|
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
|
||||||
|
|
||||||
|
await using var command = connection.CreateCommand();
|
||||||
|
command.Transaction = transaction;
|
||||||
|
command.CommandText = @"
|
||||||
|
INSERT INTO static_attribute_overrides (instance_unique_name, attribute_name, override_value, updated_at)
|
||||||
|
VALUES (@name, @attr, @val, @updatedAt)
|
||||||
|
ON CONFLICT(instance_unique_name, attribute_name) DO UPDATE SET
|
||||||
|
override_value = excluded.override_value,
|
||||||
|
updated_at = excluded.updated_at";
|
||||||
|
|
||||||
|
var pName = command.Parameters.Add("@name", SqliteType.Text);
|
||||||
|
var pAttr = command.Parameters.Add("@attr", SqliteType.Text);
|
||||||
|
var pVal = command.Parameters.Add("@val", SqliteType.Text);
|
||||||
|
var pUpdatedAt = command.Parameters.Add("@updatedAt", SqliteType.Text);
|
||||||
|
pName.Value = instanceName;
|
||||||
|
|
||||||
|
var updatedAt = DateTimeOffset.UtcNow.ToString("O");
|
||||||
|
foreach (var (attributeName, value) in overrides)
|
||||||
|
{
|
||||||
|
pAttr.Value = attributeName;
|
||||||
|
pVal.Value = value;
|
||||||
|
pUpdatedAt.Value = updatedAt;
|
||||||
|
await command.ExecuteNonQueryAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
await transaction.CommitAsync();
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Clears all static attribute overrides for an instance.
|
/// Clears all static attribute overrides for an instance.
|
||||||
/// Called on redeployment to reset overrides.
|
/// Called on redeployment to reset overrides.
|
||||||
|
|||||||
@@ -16,6 +16,15 @@ public class SiteExternalSystemRepository : IExternalSystemRepository
|
|||||||
{
|
{
|
||||||
private readonly SiteStorageService _storage;
|
private readonly SiteStorageService _storage;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Shared options for the per-row <c>MethodDefinitionDto</c> JSON parse in
|
||||||
|
/// <see cref="ParseMethodDefinitions"/> (perf remediation, arch-review WP1.5) —
|
||||||
|
/// avoids allocating a new <see cref="JsonSerializerOptions"/> per call. Settings
|
||||||
|
/// preserved exactly; the query shape itself is untouched (WP2.6).
|
||||||
|
/// </summary>
|
||||||
|
private static readonly JsonSerializerOptions MethodDefinitionJsonOptions =
|
||||||
|
new() { PropertyNameCaseInsensitive = true };
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new site-side external system repository.
|
/// Initializes a new site-side external system repository.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -279,7 +288,7 @@ public class SiteExternalSystemRepository : IExternalSystemRepository
|
|||||||
try
|
try
|
||||||
{
|
{
|
||||||
var methods = JsonSerializer.Deserialize<List<MethodDefinitionDto>>(json,
|
var methods = JsonSerializer.Deserialize<List<MethodDefinitionDto>>(json,
|
||||||
new JsonSerializerOptions { PropertyNameCaseInsensitive = true });
|
MethodDefinitionJsonOptions);
|
||||||
|
|
||||||
if (methods is null)
|
if (methods is null)
|
||||||
return Array.Empty<ExternalSystemMethod>();
|
return Array.Empty<ExternalSystemMethod>();
|
||||||
|
|||||||
@@ -242,6 +242,77 @@ public class InstanceActorTests : TestKit, IDisposable
|
|||||||
Assert.Equal("100.0", overrides["Temperature"]);
|
Assert.Equal("100.0", overrides["Temperature"]);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Perf remediation (arch-review WP1.5): a burst of rapid static-attribute writes must
|
||||||
|
/// coalesce into ONE batched SQLite transaction on a short single-shot timer — mirroring
|
||||||
|
/// NativeAlarmActor's MarkDirtyUpsert/FlushDirtyUpserts (P4) shape — rather than a write
|
||||||
|
/// per <see cref="SetStaticAttributeCommand"/>. Pinned two ways: (1) nothing is persisted
|
||||||
|
/// before the flush interval elapses, even though every write already replied
|
||||||
|
/// synchronously and updated in-memory state; (2) only the LATEST value per attribute
|
||||||
|
/// survives, across writes to two different attributes coalesced into the same flush.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task InstanceActor_RapidStaticWrites_CoalesceIntoSingleFlush()
|
||||||
|
{
|
||||||
|
var config = new FlattenedConfiguration
|
||||||
|
{
|
||||||
|
InstanceUniqueName = "PumpCoalesce1",
|
||||||
|
Attributes =
|
||||||
|
[
|
||||||
|
new ResolvedAttribute { CanonicalName = "Counter", Value = "0", DataType = "Int32" },
|
||||||
|
new ResolvedAttribute { CanonicalName = "Label", Value = "Idle", DataType = "String" }
|
||||||
|
]
|
||||||
|
};
|
||||||
|
|
||||||
|
var flushInterval = TimeSpan.FromSeconds(1);
|
||||||
|
var actor = ActorOf(Props.Create(() => new InstanceActor(
|
||||||
|
"PumpCoalesce1",
|
||||||
|
JsonSerializer.Serialize(config),
|
||||||
|
_storage,
|
||||||
|
_compilationService,
|
||||||
|
_sharedScriptLibrary,
|
||||||
|
null,
|
||||||
|
_options,
|
||||||
|
NullLogger<InstanceActor>.Instance,
|
||||||
|
null, // dclManager
|
||||||
|
null, // healthCollector
|
||||||
|
null, // serviceProvider
|
||||||
|
flushInterval)));
|
||||||
|
|
||||||
|
// Five rapid writes to "Counter" plus one to "Label", all sent back-to-back with no
|
||||||
|
// delay between them (well inside the 1s coalescing window).
|
||||||
|
for (var i = 1; i <= 5; i++)
|
||||||
|
{
|
||||||
|
actor.Tell(new SetStaticAttributeCommand(
|
||||||
|
$"corr-counter-{i}", "PumpCoalesce1", "Counter", i.ToString(), DateTimeOffset.UtcNow));
|
||||||
|
}
|
||||||
|
actor.Tell(new SetStaticAttributeCommand(
|
||||||
|
"corr-label", "PumpCoalesce1", "Label", "Running", DateTimeOffset.UtcNow));
|
||||||
|
|
||||||
|
// Every write replies synchronously (in-memory state is authoritative regardless of
|
||||||
|
// when the buffered persist lands) — six writes, six responses.
|
||||||
|
for (var i = 0; i < 6; i++)
|
||||||
|
{
|
||||||
|
var response = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||||||
|
Assert.True(response.Success);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Well before the flush interval elapses, NOTHING has reached SQLite yet — proves the
|
||||||
|
// writes did not each fire their own persist (the old per-write fire-and-forget shape
|
||||||
|
// would already show a row here).
|
||||||
|
await Task.Delay(200);
|
||||||
|
var beforeFlush = await _storage.GetStaticOverridesAsync("PumpCoalesce1");
|
||||||
|
Assert.Empty(beforeFlush);
|
||||||
|
|
||||||
|
// After the flush interval elapses, exactly one coalesced write has landed: only the
|
||||||
|
// LATEST value per attribute survives.
|
||||||
|
await Task.Delay(1500);
|
||||||
|
var afterFlush = await _storage.GetStaticOverridesAsync("PumpCoalesce1");
|
||||||
|
Assert.Equal(2, afterFlush.Count);
|
||||||
|
Assert.Equal("5", afterFlush["Counter"]);
|
||||||
|
Assert.Equal("Running", afterFlush["Label"]);
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task InstanceActor_LoadsStaticOverridesFromSQLite()
|
public async Task InstanceActor_LoadsStaticOverridesFromSQLite()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user