perf(site-runtime): batch Route.To().GetAttributes/SetAttributes into one Instance Actor message (review 2026-10-21 01/SR-P4)

One Ask per attribute became one site-local InstanceGet/SetAttributesRequest
answered in a single pass, keeping the per-attribute GetAttributeResponse /
SetStaticAttributeResponse outcomes. The single-write path is refactored to a
Task<SetStaticAttributeResponse> outcome shared by both paths; static writes
and rejections still reply synchronously.
This commit is contained in:
Joseph Doherty
2026-10-22 06:37:06 -04:00
parent 35ffd025c3
commit c782744b07
5 changed files with 319 additions and 62 deletions
@@ -3108,8 +3108,9 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// <summary>
/// Reads attribute values from a deployed instance for a Route.To().GetAttribute(s)
/// call (or a central Test Run bound to the instance). Asks the Instance Actor
/// per attribute and combines the results.
/// call (or a central Test Run bound to the instance). One batched Ask to the Instance
/// Actor answers every attribute in a single pass (review 2026-10-21 01/SR-P4) — this used
/// to be one Ask (temp actor + mailbox round trip + 30 s timer) per attribute.
/// </summary>
private void RouteInboundApiGetAttributes(RouteToGetAttributesRequest request)
{
@@ -3124,17 +3125,14 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
var sender = Sender;
var names = request.AttributeNames;
var asks = names
.Select(name => instanceActor.Ask<GetAttributeResponse>(
new GetAttributeRequest(
request.CorrelationId, request.InstanceUniqueName, name, DateTimeOffset.UtcNow),
TimeSpan.FromSeconds(30)))
.ToArray();
Task.WhenAll(asks).ContinueWith(t =>
instanceActor.Ask<InstanceGetAttributesResponse>(
new InstanceGetAttributesRequest(request.CorrelationId, request.InstanceUniqueName, names),
TimeSpan.FromSeconds(30))
.ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
var results = t.Result.Results;
var values = new Dictionary<string, object?>();
for (var i = 0; i < names.Count; i++)
// Each attribute value crosses the Central↔Site PROCESS boundary inside
@@ -3146,8 +3144,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// through the same normalizer so the wire carries a plain CLR graph
// (Dictionary/List/primitive). Keys are preserved; scalars/strings/null pass
// through unchanged.
values[names[i]] = t.Result[i].Found
? NormalizeRoutedReturnValue(t.Result[i].Value)
values[names[i]] = results[i].Found
? NormalizeRoutedReturnValue(results[i].Value)
: null;
return new RouteToGetAttributesResponse(
request.CorrelationId, values, true, null, DateTimeOffset.UtcNow);
@@ -3214,11 +3212,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). 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.
/// call (or a central Test Run bound to the instance). The writes go to the Instance Actor
/// in ONE batched Ask (review 2026-10-21 01/SR-P4), 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)
{
@@ -3233,9 +3231,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
var sender = Sender;
var correlationId = request.CorrelationId;
var asks = request.AttributeValues
.Select(kvp => instanceActor.Ask<SetStaticAttributeResponse>(
new SetStaticAttributeCommand(
var writes = request.AttributeValues
.Select(kvp => new SetStaticAttributeCommand(
correlationId, request.InstanceUniqueName, kvp.Key, kvp.Value, DateTimeOffset.UtcNow,
// (ParentExecutionId tag-cascade): the routed
// write's originating execution is the inbound API request
@@ -3245,11 +3242,16 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// script run under the inbound execution.
SourceExecutionId: request.ParentExecutionId,
// Flow Debugger (design §5.3): a traced routed write arms the echo stamp.
Trace: request.Trace),
TimeSpan.FromSeconds(30)))
.ToArray();
Trace: request.Trace))
.ToList();
Task.WhenAll(asks).ContinueWith(t =>
// Each data-sourced write inside the batch is bounded by its own 30 s DCL Ask, and the
// writes run concurrently; the slack keeps a device timeout surfacing as that attribute's
// own failure rather than as a whole-batch Ask timeout.
instanceActor.Ask<InstanceSetAttributesResponse>(
new InstanceSetAttributesRequest(correlationId, writes),
TimeSpan.FromSeconds(35))
.ContinueWith(t =>
{
if (!t.IsCompletedSuccessfully)
return new RouteToSetAttributesResponse(
@@ -3257,7 +3259,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
t.Exception?.GetBaseException().Message ?? "Attribute write timed out",
DateTimeOffset.UtcNow);
var failures = t.Result
var failures = t.Result.Results
.Where(r => !r.Success)
.Select(r => $"{r.AttributeName}: {r.ErrorMessage}")
.ToArray();
@@ -321,6 +321,11 @@ public class InstanceActor : ReceiveActor
// Handle attribute queries (Tell pattern -- sender gets response)
Receive<GetAttributeRequest>(HandleGetAttribute);
// Review 2026-10-21 01/SR-P4: batched reads/writes for Route.To().GetAttributes() /
// SetAttributes() — one message per call instead of one Ask per attribute.
Receive<InstanceGetAttributesRequest>(HandleGetAttributes);
Receive<InstanceSetAttributesRequest>(HandleSetAttributes);
// Handle static attribute writes
Receive<SetStaticAttributeCommand>(HandleSetStaticAttribute);
@@ -509,18 +514,75 @@ public class InstanceActor : ReceiveActor
/// <summary>
/// Returns the current attribute value. Uses Tell pattern; sender gets the response.
/// </summary>
private void HandleGetAttribute(GetAttributeRequest request)
private void HandleGetAttribute(GetAttributeRequest request) =>
Sender.Tell(ReadAttribute(request.CorrelationId, request.AttributeName));
/// <summary>The single-attribute read shared by the single and batched paths.</summary>
private GetAttributeResponse ReadAttribute(string correlationId, string attributeName)
{
var found = _attributes.TryGetValue(request.AttributeName, out var value);
_attributeQualities.TryGetValue(request.AttributeName, out var quality);
Sender.Tell(new GetAttributeResponse(
request.CorrelationId,
var found = _attributes.TryGetValue(attributeName, out var value);
_attributeQualities.TryGetValue(attributeName, out var quality);
return new GetAttributeResponse(
correlationId,
_instanceUniqueName,
request.AttributeName,
attributeName,
value,
found,
quality ?? "Good",
DateTimeOffset.UtcNow));
DateTimeOffset.UtcNow);
}
/// <summary>
/// Review 2026-10-21 01/SR-P4: answers every requested attribute in one pass, one
/// <see cref="GetAttributeResponse"/> per name (index-aligned), from a single consistent
/// point in this actor's mailbox.
/// </summary>
private void HandleGetAttributes(InstanceGetAttributesRequest request)
{
var results = new GetAttributeResponse[request.AttributeNames.Count];
for (var i = 0; i < results.Length; i++)
results[i] = ReadAttribute(request.CorrelationId, request.AttributeNames[i]);
Sender.Tell(new InstanceGetAttributesResponse(request.CorrelationId, results));
}
/// <summary>
/// Review 2026-10-21 01/SR-P4: applies each write in order exactly as the equivalent single
/// <see cref="SetStaticAttributeCommand"/> would be applied (static writes take effect
/// synchronously here, data-sourced writes are issued to the DCL concurrently), then replies
/// once with every per-write outcome — immediately when nothing is outstanding, otherwise
/// when the last DCL write answers. Each data write keeps its own 30 s DCL bound.
/// </summary>
private void HandleSetAttributes(InstanceSetAttributesRequest request)
{
var caller = Sender;
var outcomes = new Task<SetStaticAttributeResponse>[request.Writes.Count];
for (var i = 0; i < outcomes.Length; i++)
outcomes[i] = ApplySetAttribute(request.Writes[i]);
if (outcomes.All(t => t.IsCompleted))
{
caller.Tell(new InstanceSetAttributesResponse(
request.CorrelationId, outcomes.Select(t => t.Result).ToArray()));
return;
}
// Every outcome task is a never-faulting ContinueWith projection (see
// ApplySetDataAttribute), so WhenAll always yields the full result array.
Task.WhenAll(outcomes)
.ContinueWith(t => new InstanceSetAttributesResponse(request.CorrelationId, t.Result))
.PipeTo(caller);
}
/// <summary>
/// Replies with a write outcome: synchronously when it is already known (every static write
/// and every rejection — the exact pre-batching timing), otherwise when the DCL answers.
/// </summary>
private static void ReplyWhenDone(Task<SetStaticAttributeResponse> outcome, IActorRef caller)
{
if (outcome.IsCompleted)
caller.Tell(outcome.Result);
else
outcome.PipeTo(caller);
}
/// <summary>
@@ -538,7 +600,15 @@ public class InstanceActor : ReceiveActor
///
/// Either way the caller receives a <see cref="SetStaticAttributeResponse"/>.
/// </summary>
private void HandleSetStaticAttribute(SetStaticAttributeCommand command)
private void HandleSetStaticAttribute(SetStaticAttributeCommand command) =>
ReplyWhenDone(ApplySetAttribute(command), Sender);
/// <summary>
/// Applies one attribute write (see <see cref="HandleSetStaticAttribute"/>) and returns its
/// outcome — completed for a static write or a rejection, pending for a data-sourced write
/// until the DCL answers. Shared by the single and batched write paths (01/SR-P4).
/// </summary>
private Task<SetStaticAttributeResponse> ApplySetAttribute(SetStaticAttributeCommand command)
{
// O(1) lookup via the existing hot-path index instead of a linear scan over
// _configuration.Attributes.
@@ -558,14 +628,13 @@ public class InstanceActor : ReceiveActor
_logger.LogWarning(
"SetAttribute rejected — attribute '{Attribute}' is not defined on instance '{Instance}'",
command.AttributeName, _instanceUniqueName);
Sender.Tell(new SetStaticAttributeResponse(
return Task.FromResult(new SetStaticAttributeResponse(
command.CorrelationId,
_instanceUniqueName,
command.AttributeName,
false,
$"Unknown attribute '{command.AttributeName}'",
DateTimeOffset.UtcNow));
return;
}
var isDataSourced =
@@ -573,19 +642,16 @@ public class InstanceActor : ReceiveActor
&& !string.IsNullOrEmpty(resolved.Attribute.BoundDataConnectionName);
if (isDataSourced)
{
HandleSetDataAttribute(command, resolved);
return;
}
return ApplySetDataAttribute(command, resolved);
HandleSetStaticAttributeCore(command);
return Task.FromResult(ApplySetStaticAttributeCore(command));
}
/// <summary>
/// Static attribute write: updates in-memory state, publishes the change,
/// persists the override to SQLite, and replies with success.
/// persists the override to SQLite, and returns the outcome to reply with.
/// </summary>
private void HandleSetStaticAttributeCore(SetStaticAttributeCommand command)
private SetStaticAttributeResponse ApplySetStaticAttributeCore(SetStaticAttributeCommand command)
{
// command.Value is the canonical form — a plain string for scalars,
// a JSON array string for List attributes. For a List attribute we store
@@ -613,14 +679,13 @@ public class InstanceActor : ReceiveActor
_logger.LogWarning(
"SetAttribute rejected — value for List attribute '{Attribute}' on instance '{Instance}' is not a valid list",
command.AttributeName, _instanceUniqueName);
Sender.Tell(new SetStaticAttributeResponse(
return new SetStaticAttributeResponse(
command.CorrelationId,
_instanceUniqueName,
command.AttributeName,
false,
$"Invalid list value for attribute '{command.AttributeName}'",
DateTimeOffset.UtcNow));
return;
DateTimeOffset.UtcNow);
}
_attributes[command.AttributeName] = decoded;
@@ -658,9 +723,9 @@ public class InstanceActor : ReceiveActor
// of when the buffered write lands.
MarkDirtyStaticOverride(command.AttributeName, command.Value);
Sender.Tell(new SetStaticAttributeResponse(
return new SetStaticAttributeResponse(
command.CorrelationId, _instanceUniqueName, command.AttributeName,
true, null, DateTimeOffset.UtcNow));
true, null, DateTimeOffset.UtcNow);
}
/// <summary>
@@ -728,14 +793,14 @@ public class InstanceActor : ReceiveActor
}
/// <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
/// Data-sourced attribute write: forwards a write request to the DCL and returns
/// the device write outcome (a task that never faults). 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, AttributeIndexEntry resolved)
private Task<SetStaticAttributeResponse> ApplySetDataAttribute(
SetStaticAttributeCommand command, AttributeIndexEntry resolved)
{
var caller = Sender;
var correlationId = command.CorrelationId;
var attributeName = command.AttributeName;
var instanceName = _instanceUniqueName;
@@ -745,10 +810,9 @@ public class InstanceActor : ReceiveActor
_logger.LogWarning(
"SetAttribute on data-sourced attribute {Instance}.{Attribute} cannot be routed — no DCL manager configured",
instanceName, attributeName);
caller.Tell(new SetStaticAttributeResponse(
return Task.FromResult(new SetStaticAttributeResponse(
correlationId, instanceName, attributeName, false,
"Data Connection Layer not available for write.", DateTimeOffset.UtcNow));
return;
}
// MV: for a data-sourced List attribute the incoming command.Value is
@@ -758,7 +822,7 @@ public class InstanceActor : ReceiveActor
// List<T> so the DCL/Variant write produces a real array. A non-empty value
// that fails to decode (malformed JSON / bad element) is poison — reject the
// write rather than forward garbage to the device (mirrors the static-path
// rejection in HandleSetStaticAttributeCore). Scalars are unchanged.
// rejection in ApplySetStaticAttributeCore). Scalars are unchanged.
object? writeValue = command.Value;
if (resolved.IsList && !string.IsNullOrWhiteSpace(command.Value))
{
@@ -769,10 +833,9 @@ public class InstanceActor : ReceiveActor
_logger.LogWarning(
"SetAttribute rejected — value for data-sourced List attribute '{Attribute}' on instance '{Instance}' is not a valid list",
attributeName, instanceName);
caller.Tell(new SetStaticAttributeResponse(
return Task.FromResult(new SetStaticAttributeResponse(
correlationId, instanceName, attributeName, false,
$"Invalid list value for attribute '{attributeName}'", DateTimeOffset.UtcNow));
return;
}
writeValue = decoded;
@@ -801,9 +864,9 @@ public class InstanceActor : ReceiveActor
// touch _traceArms directly — it Tells self instead.
var self = Self;
// Ask the DCL and pipe the result back to the original caller. The DCL
// Ask the DCL; the outcome is replied to the original caller. The DCL
// returns the failure synchronously so the script can handle it.
_dclManager.Ask<WriteTagResponse>(writeRequest, TimeSpan.FromSeconds(30))
return _dclManager.Ask<WriteTagResponse>(writeRequest, TimeSpan.FromSeconds(30))
.ContinueWith(t =>
{
var succeeded = t.IsCompletedSuccessfully && t.Result.Success;
@@ -819,13 +882,13 @@ public class InstanceActor : ReceiveActor
correlationId, instanceName, attributeName, false,
t.Exception?.GetBaseException().Message ?? "DCL write timed out",
DateTimeOffset.UtcNow);
}).PipeTo(caller);
});
}
/// <summary>
/// Batch write: resolves a SET of data-sourced attributes (and an optional trigger
/// attribute) to their device bindings, enforces a single data connection across the
/// whole batch, decodes List values (same rule as <see cref="HandleSetDataAttribute"/>),
/// whole batch, decodes List values (same rule as <see cref="ApplySetDataAttribute"/>),
/// and forwards ONE <see cref="WriteTagBatchRequest"/> to the DCL — replacing N
/// sequential per-attribute writes with a single gateway round-trip. The write outcome
/// is returned synchronously to the caller; the EVENT-DRIVEN wait for the response
@@ -873,7 +936,7 @@ public class InstanceActor : ReceiveActor
object? wv = kv.Value;
// 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
// produces a real array (same poison-rejection rule as HandleSetDataAttribute).
// produces a real array (same poison-rejection rule as ApplySetDataAttribute).
if (resolved.IsList && !string.IsNullOrWhiteSpace(kv.Value))
{
var decoded = DecodeAttributeValue(
@@ -1749,7 +1812,7 @@ public class InstanceActor : ReceiveActor
// WaitForAttribute (spec §4.2): re-evaluate any waiters on THIS attribute —
// but ONLY when this publish reflects a real value change (evaluateWaiters).
// The genuine value-change paths (HandleAttributeValueChanged, the scalar
// DCL update path, HandleSetStaticAttributeCore) call it AFTER assigning
// DCL update path, ApplySetStaticAttributeCore) call it AFTER assigning
// _attributes[name], so changed.Value is the just-applied current value.
// Republish/quality-only paths (List-coerce-failure Bad-quality, which
// publishes the OLD value) pass evaluateWaiters:false so an "any-change" or
@@ -0,0 +1,59 @@
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages;
/// <summary>
/// Site-local batch read of several attributes from ONE Instance Actor, answered in a single
/// pass with <see cref="InstanceGetAttributesResponse"/> (review 2026-10-21 01/SR-P4). Sent by
/// the Deployment Manager for <c>Route.To().GetAttributes()</c> instead of one Ask — temp actor,
/// mailbox round trip and 30 s timer — per attribute.
///
/// <para>
/// Never crosses a process boundary: the Deployment Manager and its Instance Actors share a
/// node, so this is a plain in-memory record with no wire-compatibility contract. (The unused
/// Commons <c>GetAttributesBatchRequest</c> was not reused because it has no per-attribute
/// found/quality outcome.)
/// </para>
/// </summary>
/// <param name="CorrelationId">Echoed on the response and on every per-attribute result.</param>
/// <param name="InstanceUniqueName">The instance being read.</param>
/// <param name="AttributeNames">Attributes to read, in the order results are returned.</param>
public sealed record InstanceGetAttributesRequest(
string CorrelationId,
string InstanceUniqueName,
IReadOnlyList<string> AttributeNames);
/// <summary>
/// Reply to <see cref="InstanceGetAttributesRequest"/>: exactly one
/// <see cref="GetAttributeResponse"/> per requested name, index-aligned with the request, each
/// carrying its own found/quality outcome — the same records a single read returns.
/// </summary>
/// <param name="CorrelationId">The request's correlation id.</param>
/// <param name="Results">Per-attribute results, index-aligned with the request.</param>
public sealed record InstanceGetAttributesResponse(
string CorrelationId,
IReadOnlyList<GetAttributeResponse> Results);
/// <summary>
/// Site-local batch write of several attributes on ONE Instance Actor (review 2026-10-21
/// 01/SR-P4), backing <c>Route.To().SetAttributes()</c>. Each write is applied in order exactly
/// as the equivalent single <see cref="SetStaticAttributeCommand"/> would be (static → in-memory +
/// coalesced persist; data-sourced → DCL write), and the reply waits only for the data-sourced
/// writes still in flight.
/// </summary>
/// <param name="CorrelationId">Echoed on the response.</param>
/// <param name="Writes">The writes, in application order.</param>
public sealed record InstanceSetAttributesRequest(
string CorrelationId,
IReadOnlyList<SetStaticAttributeCommand> Writes);
/// <summary>
/// Reply to <see cref="InstanceSetAttributesRequest"/>: one <see cref="SetStaticAttributeResponse"/>
/// per write, index-aligned, so a failed device write or an unknown attribute is reported per
/// attribute just as a single write reports it.
/// </summary>
/// <param name="CorrelationId">The request's correlation id.</param>
/// <param name="Results">Per-write results, index-aligned with the request.</param>
public sealed record InstanceSetAttributesResponse(
string CorrelationId,
IReadOnlyList<SetStaticAttributeResponse> Results);