perf(acl): dedup subscription revalidation by (session, node)
This commit is contained in:
Joseph Doherty
+242
@@ -347,8 +347,102 @@ public sealed class NodeManagerStage6EnforcementTests : IDisposable
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Revalidation asks the authorizer once per distinct (session, node), not once per monitored item.
|
||||
/// A client that subscribes to the same tag from several items — one per attribute, or one per
|
||||
/// downstream consumer sharing a session — is the ordinary case, and every one of those items used
|
||||
/// to pay for its own grant walk on every deployment.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The observable is the authorizer's <see cref="TimeProvider"/>. <c>Evaluate</c> reads the clock
|
||||
/// exactly once per call, for the key revocation/expiry check that is deliberately outside the
|
||||
/// decision cache — so counting clock reads counts <c>IsAllowedCached</c> entries, which is the
|
||||
/// work this dedup removes. Counting the permission evaluator instead would be vacuous: the
|
||||
/// authorizer's own decision cache already memoizes the grant walk, so it would report one
|
||||
/// either way.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <see cref="OtOpcUaNodeManager.RevalidateMonitoredItems"/> is called directly rather than via
|
||||
/// an authority flip so the count covers exactly one pass — a flip would fire the pass through
|
||||
/// <c>OnAuthorityChanged</c> as well and the assertion would be about two.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task Revalidation_evaluates_one_verdict_per_session_and_node()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var clock = new CountingTimeProvider(new DateTime(2026, 8, 15, 12, 0, 0, DateTimeKind.Utc));
|
||||
nm.Authorizer = SubscribeAuthorizer(clock);
|
||||
|
||||
var nodeId = Variable(nm, "Plant/Modbus/dev1/Speed");
|
||||
var identity = Identity();
|
||||
var items = Enumerable.Range(0, 3)
|
||||
.Select(_ => TrackItem(nm, nodeId, identity))
|
||||
.ToList();
|
||||
|
||||
clock.Reset();
|
||||
nm.RevalidateMonitoredItems();
|
||||
|
||||
clock.Calls.ShouldBe(1, "three items on one node share one (session, node) verdict");
|
||||
|
||||
// Non-vacuity: the grant really permitted, so the memoized verdict is an allow rather than the
|
||||
// deny every unresolvable case produces.
|
||||
items.ShouldAllBe(i => i.Denials == 0);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The falsification control for <see cref="Revalidation_evaluates_one_verdict_per_session_and_node"/>:
|
||||
/// three items on three DISTINCT nodes still cost three evaluations, so the dedup is keyed on the
|
||||
/// node and not collapsing every item in the pass to a single verdict.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Revalidation_still_evaluates_each_distinct_node()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var clock = new CountingTimeProvider(new DateTime(2026, 8, 15, 12, 0, 0, DateTimeKind.Utc));
|
||||
nm.Authorizer = SubscribeAuthorizer(clock);
|
||||
|
||||
var identity = Identity();
|
||||
foreach (var tag in new[] { "Speed", "Torque", "Pressure" })
|
||||
TrackItem(nm, Variable(nm, $"Plant/Modbus/dev1/{tag}"), identity);
|
||||
|
||||
clock.Reset();
|
||||
nm.RevalidateMonitoredItems();
|
||||
|
||||
clock.Calls.ShouldBe(3);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ───────────────────────────── helpers ─────────────────────────────
|
||||
|
||||
/// <summary>A cluster-wide Subscribe grant — what <c>CreateMonitoredItems</c> resolves to.</summary>
|
||||
private static DataPlaneAuthorizer SubscribeAuthorizer(TimeProvider clock)
|
||||
{
|
||||
var state = new DataPlaneAuthorityState();
|
||||
state.Apply(Cluster, "rev-1", [], [GrantOn(NodePermissions.Subscribe)], Scopes());
|
||||
return new DataPlaneAuthorizer(
|
||||
state, new TriePermissionEvaluator(state.Tries), enabled: true, NullLogger.Instance, clock);
|
||||
}
|
||||
|
||||
/// <summary>Registers a stub data-change item against <paramref name="nodeId"/> as if the SDK had created it.</summary>
|
||||
private static StubSampledItem TrackItem(OtOpcUaNodeManager nm, NodeId nodeId, IUserIdentity identity)
|
||||
{
|
||||
var handle = (NodeHandle)nm.GetManagerHandle(nodeId);
|
||||
var item = new StubSampledItem(handle, identity);
|
||||
nm.TrackLiveMonitoredItem(handle, item);
|
||||
return item;
|
||||
}
|
||||
|
||||
private static NodeId Variable(OtOpcUaNodeManager nm, string rawPath)
|
||||
{
|
||||
nm.EnsureVariable(rawPath, parentFolderNodeId: null, displayName: rawPath, dataType: "Float",
|
||||
@@ -509,6 +603,154 @@ public sealed class NodeManagerStage6EnforcementTests : IDisposable
|
||||
public void Advance(TimeSpan by) => _now = _now.Add(by);
|
||||
}
|
||||
|
||||
/// <summary>A fixed clock that counts how many times it was read.</summary>
|
||||
/// <remarks>
|
||||
/// The authorizer reads its clock exactly once per <c>Evaluate</c>, so this counts authorization
|
||||
/// evaluations without needing a seam on the authorizer itself.
|
||||
/// </remarks>
|
||||
private sealed class CountingTimeProvider(DateTime utcNow) : TimeProvider
|
||||
{
|
||||
private readonly DateTimeOffset _now = new(utcNow, TimeSpan.Zero);
|
||||
|
||||
public int Calls { get; private set; }
|
||||
|
||||
public override DateTimeOffset GetUtcNow()
|
||||
{
|
||||
Calls++;
|
||||
return _now;
|
||||
}
|
||||
|
||||
public void Reset() => Calls = 0;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The minimum of a sampled data-change monitored item that revalidation touches: the identity that
|
||||
/// created it, and the denial queue it is told through.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Everything else throws rather than returning a plausible default, so a revalidation that started
|
||||
/// depending on more of the item surfaces here instead of being silently absorbed.
|
||||
/// </remarks>
|
||||
private sealed class StubSampledItem(NodeHandle handle, IUserIdentity identity)
|
||||
: ISampledDataChangeMonitoredItem
|
||||
{
|
||||
/// <summary>How many <c>BadUserAccessDenied</c> values revalidation queued on this item.</summary>
|
||||
public int Denials { get; private set; }
|
||||
|
||||
public IUserIdentity EffectiveIdentity => identity;
|
||||
|
||||
public object ManagerHandle => handle;
|
||||
|
||||
public NodeId NodeId => handle.NodeId;
|
||||
|
||||
public void QueueValue(DataValue value, ServiceResult error, bool ignoreFilters)
|
||||
{
|
||||
if (error?.StatusCode.Code == StatusCodes.BadUserAccessDenied) Denials++;
|
||||
}
|
||||
|
||||
public void QueueValue(DataValue value, ServiceResult error) => QueueValue(value, error, false);
|
||||
|
||||
public void Dispose() { }
|
||||
|
||||
// ── everything below is untouched by revalidation ──
|
||||
public INodeManager NodeManager => throw new NotSupportedException();
|
||||
|
||||
public ISession Session => throw new NotSupportedException();
|
||||
|
||||
public uint Id => throw new NotSupportedException();
|
||||
|
||||
public uint SubscriptionId => throw new NotSupportedException();
|
||||
|
||||
public bool IsDurable => throw new NotSupportedException();
|
||||
|
||||
public uint ClientHandle => throw new NotSupportedException();
|
||||
|
||||
public ISubscription SubscriptionCallback
|
||||
{
|
||||
get => throw new NotSupportedException();
|
||||
set => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
public int MonitoredItemType => throw new NotSupportedException();
|
||||
|
||||
public bool IsReadyToPublish => throw new NotSupportedException();
|
||||
|
||||
public bool IsReadyToTrigger
|
||||
{
|
||||
get => throw new NotSupportedException();
|
||||
set => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
public bool IsResendData => throw new NotSupportedException();
|
||||
|
||||
public MonitoringMode MonitoringMode => throw new NotSupportedException();
|
||||
|
||||
public double SamplingInterval => throw new NotSupportedException();
|
||||
|
||||
public DataChangeFilter DataChangeFilter => throw new NotSupportedException();
|
||||
|
||||
public uint AttributeId => throw new NotSupportedException();
|
||||
|
||||
public NumericRange IndexRange => throw new NotSupportedException();
|
||||
|
||||
public QualifiedName DataEncoding => throw new NotSupportedException();
|
||||
|
||||
public bool AlwaysReportUpdates
|
||||
{
|
||||
get => throw new NotSupportedException();
|
||||
set => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
public DiagnosticsMasks DiagnosticsMasks => throw new NotSupportedException();
|
||||
|
||||
public uint QueueSize => throw new NotSupportedException();
|
||||
|
||||
public double MinimumSamplingInterval => throw new NotSupportedException();
|
||||
|
||||
public int TimeToNextSample => throw new NotSupportedException();
|
||||
|
||||
public void SetupResendDataTrigger() => throw new NotSupportedException();
|
||||
|
||||
public ServiceResult GetCreateResult(out MonitoredItemCreateResult result) =>
|
||||
throw new NotSupportedException();
|
||||
|
||||
public ServiceResult GetModifyResult(out MonitoredItemModifyResult result) =>
|
||||
throw new NotSupportedException();
|
||||
|
||||
public IStoredMonitoredItem ToStorableMonitoredItem() => throw new NotSupportedException();
|
||||
|
||||
public MonitoringMode SetMonitoringMode(MonitoringMode monitoringMode) =>
|
||||
throw new NotSupportedException();
|
||||
|
||||
public bool Publish(
|
||||
OperationContext context,
|
||||
Queue<MonitoredItemNotification> notifications,
|
||||
Queue<DiagnosticInfo> diagnostics,
|
||||
uint maxNotificationsPerPublish,
|
||||
Microsoft.Extensions.Logging.ILogger logger) => throw new NotSupportedException();
|
||||
|
||||
public void SetSemanticsChanged() => throw new NotSupportedException();
|
||||
|
||||
public void SetStructureChanged() => throw new NotSupportedException();
|
||||
|
||||
public bool SamplingIntervalExpired() => throw new NotSupportedException();
|
||||
|
||||
public ReadValueId GetReadValueId() => throw new NotSupportedException();
|
||||
|
||||
public ServiceResult ModifyAttributes(
|
||||
DiagnosticsMasks diagnosticsMasks,
|
||||
TimestampsToReturn timestampsToReturn,
|
||||
uint clientHandle,
|
||||
MonitoringFilter originalFilter,
|
||||
MonitoringFilter filterToUse,
|
||||
Opc.Ua.Range range,
|
||||
double samplingInterval,
|
||||
uint queueSize,
|
||||
bool discardOldest) => throw new NotSupportedException();
|
||||
|
||||
public void SetSamplingInterval(double samplingInterval) => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
/// <summary>Cleans up the PKI root directory.</summary>
|
||||
public void Dispose()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user