Retain and replay ServiceLevel across publisher binding and failures
v2-ci / unit-tests (pull_request) Waiting to run

This commit is contained in:
Joseph Doherty
2026-10-13 13:26:27 -04:00
parent 6f0674296b
commit c8e3e279b4
13 changed files with 545 additions and 30 deletions
@@ -1,21 +1,38 @@
namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
/// <summary>
/// Late-binding adapter that holds an inner <see cref="IServiceLevelPublisher"/> reference
/// swappable at runtime. Mirrors <see cref="DeferredAddressSpaceSink"/>: Akka actors resolve
/// the publisher at DI time, but the production <c>SdkServiceLevelPublisher</c> only exists
/// after <c>StandardServer.Start</c>. The Host's hosted service swaps the inner once the SDK
/// is up; until then writes route through <see cref="NullServiceLevelPublisher"/>.
/// Late-binding adapter that accepts and retains the latest requested ServiceLevel until
/// the SDK publisher is available. Binding synchronously replays that request, including zero.
/// </summary>
/// <remarks>
/// Publishing and binding are serialized through the synchronous inner call. Inner publishers
/// must not call back into this adapter. Acceptance while unbound is not an SDK write confirmation.
/// A failed call retains both the requested level and the installed publisher for later recovery.
/// </remarks>
public sealed class DeferredServiceLevelPublisher : IServiceLevelPublisher
{
private volatile IServiceLevelPublisher _inner = NullServiceLevelPublisher.Instance;
private readonly object _gate = new();
private IServiceLevelPublisher? _inner;
private byte? _latest;
/// <summary>Swap the underlying publisher. Pass null to revert to the Null no-op.</summary>
/// <param name="inner">The publisher implementation to use, or null to use the null publisher.</param>
public void SetInner(IServiceLevelPublisher? inner) =>
_inner = inner ?? NullServiceLevelPublisher.Instance;
/// <summary>Bind and replay the latest request, or unbind while retaining that request.</summary>
/// <param name="inner">The publisher implementation to use, or null to unbind.</param>
public void SetInner(IServiceLevelPublisher? inner)
{
lock (_gate)
{
_inner = inner;
if (_latest is { } latest) _inner?.Publish(latest);
}
}
/// <inheritdoc />
public void Publish(byte serviceLevel) => _inner.Publish(serviceLevel);
public void Publish(byte serviceLevel)
{
lock (_gate)
{
_latest = serviceLevel;
_inner?.Publish(serviceLevel);
}
}
}
@@ -0,0 +1,38 @@
using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa;
/// <summary>Coordinates endpoint cleanup after ServiceLevel binding fails during startup.</summary>
internal static class FailedServiceLevelBindingCleanup
{
/// <summary>Stops and disposes the endpoint, returning whether ownership can be released.</summary>
/// <param name="stop">The SDK stop operation.</param>
/// <param name="dispose">The application-host disposal operation.</param>
/// <param name="logger">The cleanup logger.</param>
/// <returns>Whether cleanup completed successfully.</returns>
internal static async Task<bool> StopAndDisposeAsync(
Func<Task> stop, Func<ValueTask> dispose, ILogger logger)
{
var completed = true;
try
{
await stop().ConfigureAwait(false);
}
catch (Exception ex)
{
completed = false;
logger.LogError(ex, "OtOpcUaServerHostedService: SDK stop failed after binding failure; retaining endpoint ownership for shutdown retry");
}
// Attempt application cleanup even when stopping faults; keep ownership on either failure.
try
{
await dispose().ConfigureAwait(false);
}
catch (Exception ex)
{
completed = false;
logger.LogError(ex, "OtOpcUaServerHostedService: endpoint disposal failed after binding failure; retaining endpoint ownership for shutdown retry");
}
return completed;
}
}
@@ -339,11 +339,36 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
_server.NodeManager.Clock = historyUpdate.Clock;
// ServiceLevel publisher needs IServerInternal — only available after Start.
if (_server.CurrentInstance is { } serverInternal)
try
{
_deferredServiceLevel.SetInner(new SdkServiceLevelPublisher(
serverInternal,
_loggerFactory.CreateLogger<SdkServiceLevelPublisher>()));
if (_server.CurrentInstance is { } serverInternal)
{
_deferredServiceLevel.SetInner(new SdkServiceLevelPublisher(
serverInternal,
_loggerFactory.CreateLogger<SdkServiceLevelPublisher>()));
}
}
catch (Exception ex)
{
_logger.LogError(ex,
"OtOpcUaServerHostedService: ServiceLevel binding failed; stopping the OPC UA endpoint while the admin host continues");
try
{
await StopAsync(cancellationToken).ConfigureAwait(false);
}
catch (Exception cleanupError)
{
_logger.LogError(cleanupError, "OtOpcUaServerHostedService: failed to detach OPC UA bindings after startup failure");
}
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
async () => await _server.StopAsync(CancellationToken.None).ConfigureAwait(false),
() => _appHost.DisposeAsync(), _logger).ConfigureAwait(false);
if (released)
{
_appHost = null;
_server = null;
}
return;
}
// arch-review 03/S19 — LAST, after the node manager is known non-null and after the ACL gate is
@@ -558,10 +558,15 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
/// <returns>A task that represents the asynchronous operation.</returns>
public async ValueTask DisposeAsync()
{
if (_impersonateHandler is not null && _server?.CurrentInstance?.SessionManager is { } sessionManager)
try
{
try { sessionManager.ImpersonateUser -= _impersonateHandler; }
catch (Exception ex) { _logger.LogWarning(ex, "OpcUaApplicationHost: detaching ImpersonateUser threw"); }
// An explicit SDK stop may already have halted CurrentInstance; still continue disposal.
if (_impersonateHandler is not null && _server?.CurrentInstance?.SessionManager is { } sessionManager)
sessionManager.ImpersonateUser -= _impersonateHandler;
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OpcUaApplicationHost: detaching ImpersonateUser threw");
}
_impersonateHandler = null;
@@ -38,8 +38,8 @@ public sealed class SdkServiceLevelPublisher : IServiceLevelPublisher
var node = _serverInternal.ServerObject?.ServiceLevel;
if (node is null)
{
_logger.LogWarning("SdkServiceLevelPublisher: ServerObject.ServiceLevel unavailable; skipping write");
return;
_logger.LogWarning("SdkServiceLevelPublisher: ServerObject.ServiceLevel unavailable");
throw new InvalidOperationException("Server.ServiceLevel is unavailable.");
}
try
@@ -55,6 +55,7 @@ public sealed class SdkServiceLevelPublisher : IServiceLevelPublisher
catch (Exception ex)
{
_logger.LogWarning(ex, "SdkServiceLevelPublisher: write to Server.ServiceLevel threw");
throw;
}
}
}
@@ -970,10 +970,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
// node starting Detached/role-less (first level = 0) would be dedup'd away, leaving the SDK's
// built-in default (255 = full service) standing — a degraded node wrongly advertising 255.
if (_publishedAtLeastOnce && msg.ServiceLevel == _lastServiceLevel) return;
_lastServiceLevel = msg.ServiceLevel;
try
{
_serviceLevel.Publish(msg.ServiceLevel);
_lastServiceLevel = msg.ServiceLevel;
_publishedAtLeastOnce = true;
OtOpcUaTelemetry.ServiceLevelChange.Add(1,
new KeyValuePair<string, object?>("level", msg.ServiceLevel));
@@ -981,6 +981,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
catch (Exception ex)
{
// A failed write may be partially applied; even a previous successful value must retry.
_publishedAtLeastOnce = false;
_log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
}
}