using Grpc.Core; using Microsoft.Extensions.Logging; using ZB.MOM.WW.HistorianGateway.Client; using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; /// /// backed by the HistorianGateway SendEvent path. The /// drain worker behind SqliteStoreAndForwardSink calls /// and uses the returned per-event /// to decide retry vs. dead-letter, so this writer maps every /// gateway result — success ack, the published client's typed exception hierarchy, raw /// (defensive), and any unexpected error — onto exactly one outcome per /// event and never throws. /// /// /// /// Each event is sent individually so one poison event cannot fail the whole batch: a permanent /// failure on event N is dead-lettered while its siblings continue. Outcomes are returned in /// input order, one per event; an empty batch yields an empty list with no gateway call. /// /// /// Outcome mapping. Success (or store-forward-queued) ack ⇒ . /// Transient gRPC codes (Unavailable, DeadlineExceeded, ResourceExhausted, /// Aborted, Internal) and the auth codes (Unauthenticated, /// PermissionDenied) ⇒ — an auth fix /// re-enables the batch, so an auth blip never dead-letters. Permanent codes /// (InvalidArgument, FailedPrecondition, OutOfRange, Unimplemented) ⇒ /// (dead-letter poison — mirrors the Wonderware /// PerEventStatus==2 boundary). The typed client exceptions are classified by type, or by /// the they wrap; any other or unclassifiable error defaults to /// so the drain worker cannot loop a poison /// event forever. /// /// public sealed class GatewayAlarmHistorianWriter : IAlarmHistorianWriter, IDisposable { private readonly IHistorianGatewayClient _client; private readonly ILogger _logger; /// Creates the writer over a gateway client seam. /// The gateway client used for the SendEvent write path. /// Logger for per-event outcome diagnostics (never logs event content). public GatewayAlarmHistorianWriter(IHistorianGatewayClient client, ILogger logger) { _client = client ?? throw new ArgumentNullException(nameof(client)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); } /// public async Task> WriteBatchAsync( IReadOnlyList batch, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(batch); if (batch.Count == 0) { return Array.Empty(); } var outcomes = new HistorianWriteOutcome[batch.Count]; for (var i = 0; i < batch.Count; i++) { if (cancellationToken.IsCancellationRequested) { // Shutdown mid-drain: short-circuit the remaining events to RetryPlease rather than // calling the gateway with a cancelled token. They stay queued for retry next startup // — a cancellation must NEVER dead-letter an in-flight event (silent data loss). outcomes[i] = HistorianWriteOutcome.RetryPlease; continue; } outcomes[i] = await SendOneAsync(batch[i], cancellationToken).ConfigureAwait(false); } return outcomes; } private async Task SendOneAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken) { try { var ack = await _client.SendEventAsync(AlarmEventMapper.ToHistorianEvent(evt), cancellationToken) .ConfigureAwait(false); return MapAck(ack); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { // Cancellation mid-send at shutdown is NOT a poison event. Map to RetryPlease so the // event stays queued for next startup rather than being dead-lettered (data loss). _logger.LogDebug("Alarm SendEvent cancelled at shutdown; will retry."); return HistorianWriteOutcome.RetryPlease; } catch (Exception exception) { // NEVER throw out of the writer — the drain worker expects a per-event outcome. Classify // and log only the failure category (no event content, hostnames, or credentials). var outcome = Classify(exception); if (outcome == HistorianWriteOutcome.PermanentFail) { _logger.LogWarning( "Alarm SendEvent permanently failed ({Exception}); dead-lettering the event.", exception.GetType().Name); } else { _logger.LogDebug( "Alarm SendEvent transiently failed ({Exception}); will retry.", exception.GetType().Name); } return outcome; } } // A non-success ack that the gateway durably queued (store-forward) is still accepted — do not // re-drain it. A non-success, non-queued ack is a soft failure: retry rather than dead-letter. private static HistorianWriteOutcome MapAck(WriteAck ack) => ack.Success || ack.Queued ? HistorianWriteOutcome.Ack : HistorianWriteOutcome.RetryPlease; private static HistorianWriteOutcome Classify(Exception exception) => exception switch { // Published client's typed hierarchy (production reality). Unavailable + both auth kinds retry. HistorianGatewayUnavailableException => HistorianWriteOutcome.RetryPlease, HistorianGatewayAuthenticationException => HistorianWriteOutcome.RetryPlease, HistorianGatewayAuthorizationException => HistorianWriteOutcome.RetryPlease, // A base client exception wrapping a permanent/transient RpcException → classify by inner status. HistorianGatewayException { InnerException: RpcException inner } => ClassifyStatus(inner.StatusCode), // Defensive raw RpcException path (the seam type signature permits it). RpcException rpc => ClassifyStatus(rpc.StatusCode), // Anything else (incl. a bare base client exception we cannot classify) → dead-letter to avoid // an infinite drain loop on a poison event. _ => HistorianWriteOutcome.PermanentFail, }; private static HistorianWriteOutcome ClassifyStatus(StatusCode code) => code switch { StatusCode.Unavailable or StatusCode.DeadlineExceeded or StatusCode.ResourceExhausted or StatusCode.Aborted or StatusCode.Internal // An auth fix re-enables the whole batch — never dead-letter on an auth blip. or StatusCode.Unauthenticated or StatusCode.PermissionDenied => HistorianWriteOutcome.RetryPlease, StatusCode.InvalidArgument or StatusCode.FailedPrecondition or StatusCode.OutOfRange or StatusCode.Unimplemented => HistorianWriteOutcome.PermanentFail, // Unknown/unclassified gRPC code → dead-letter to avoid an infinite drain loop. _ => HistorianWriteOutcome.PermanentFail, }; /// /// Disposes the underlying gateway client and its gRPC channel. The concrete /// implements ; test doubles /// that only implement are safely no-opped by the cast guard. /// public void Dispose() => (_client as IDisposable)?.Dispose(); }