feat(siteruntime): Database.CachedWrite emits combined telemetry + S&F audit bridge (#23 M3)
Wire the M3 cached-call audit pipeline end-to-end for the database
channel and close the loop between the S&F lifecycle observer and the
site-side dual emitter.
* DatabaseCachedWriteEmissionTests covers Database.CachedWrite (set up
in Bundle E3): mints a TrackedOperationId, emits one CachedSubmit
packet on DbOutbound, threads the id into IDatabaseGateway, and is
best-effort on a thrown forwarder. Mirrors ExternalSystem.CachedCall
coverage from E3.
* CachedCallLifecycleBridge (new) implements ICachedCallLifecycleObserver
and lives alongside CachedCallTelemetryForwarder. The bridge ingests
per-attempt notifications from the S&F retry loop and fans them out
to the forwarder:
- TransientFailure -> 1 Attempted row
- Delivered -> Attempted + CachedResolve(Delivered)
- PermanentFailure -> Attempted + CachedResolve(Parked)
- ParkedMaxRetries -> Attempted + CachedResolve(Parked)
Channel string -> AuditKind mapping (ApiOutbound->ApiCallCached,
DbOutbound->DbWriteCached). Best-effort top-level catch swallows any
unexpected throw so the S&F retry bookkeeping is never disturbed.
* Bridge tests (7) cover all four outcomes, channel mapping, provenance
propagation, and the no-throw-on-forwarder-failure contract.
Bundle F (Host registration) will instantiate the bridge and inject it
into StoreAndForwardService.cachedCallObserver, closing the wiring path
end-to-end.
Bundle E task E6.
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,205 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ScadaLink.Commons.Entities.Audit;
|
||||
using ScadaLink.Commons.Interfaces.Services;
|
||||
using ScadaLink.Commons.Messages.Integration;
|
||||
using ScadaLink.Commons.Types;
|
||||
using ScadaLink.Commons.Types.Enums;
|
||||
|
||||
namespace ScadaLink.AuditLog.Site.Telemetry;
|
||||
|
||||
/// <summary>
|
||||
/// Audit Log #23 (M3 Bundle E — Tasks E4/E5): translates per-attempt
|
||||
/// notifications from the store-and-forward retry loop into one (or two)
|
||||
/// <see cref="CachedCallTelemetry"/> packets and pushes them through
|
||||
/// <see cref="ICachedCallTelemetryForwarder"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The S&F loop's <see cref="ICachedCallLifecycleObserver"/> reports a
|
||||
/// single coarse outcome per attempt; the audit pipeline however models the
|
||||
/// lifecycle as TWO rows on terminal outcomes — an <c>Attempted</c>
|
||||
/// (<see cref="AuditKind.ApiCallCached"/> / <see cref="AuditKind.DbWriteCached"/>)
|
||||
/// row capturing the per-attempt mechanics, plus a <see cref="AuditKind.CachedResolve"/>
|
||||
/// row marking the terminal state for downstream consumers. The bridge fans
|
||||
/// out per outcome:
|
||||
/// </para>
|
||||
/// <list type="bullet">
|
||||
/// <item><description><c>TransientFailure</c> -> one Attempted(Failed) row.</description></item>
|
||||
/// <item><description><c>Delivered</c> -> Attempted(Delivered) + CachedResolve(Delivered).</description></item>
|
||||
/// <item><description><c>PermanentFailure</c> -> Attempted(Failed) + CachedResolve(Parked).</description></item>
|
||||
/// <item><description><c>ParkedMaxRetries</c> -> Attempted(Failed) + CachedResolve(Parked).</description></item>
|
||||
/// </list>
|
||||
/// <para>
|
||||
/// <b>Best-effort emission (alog.md §7):</b> the bridge itself never throws;
|
||||
/// the underlying forwarder swallows + logs its own failures.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class CachedCallLifecycleBridge : ICachedCallLifecycleObserver
|
||||
{
|
||||
private readonly ICachedCallTelemetryForwarder _forwarder;
|
||||
private readonly ILogger<CachedCallLifecycleBridge> _logger;
|
||||
|
||||
public CachedCallLifecycleBridge(
|
||||
ICachedCallTelemetryForwarder forwarder,
|
||||
ILogger<CachedCallLifecycleBridge> logger)
|
||||
{
|
||||
_forwarder = forwarder ?? throw new ArgumentNullException(nameof(forwarder));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task OnAttemptCompletedAsync(
|
||||
CachedCallAttemptContext context, CancellationToken ct = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(context);
|
||||
|
||||
try
|
||||
{
|
||||
await EmitAttemptedAsync(context, ct).ConfigureAwait(false);
|
||||
|
||||
if (IsTerminal(context.Outcome))
|
||||
{
|
||||
await EmitResolveAsync(context, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Defensive — both EmitX paths call the forwarder which is itself
|
||||
// best-effort. A throw here is unexpected, but the alog.md §7
|
||||
// contract requires we never propagate.
|
||||
_logger.LogWarning(ex,
|
||||
"CachedCallLifecycleBridge: unexpected throw for {TrackedOperationId} (Outcome {Outcome})",
|
||||
context.TrackedOperationId, context.Outcome);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task EmitAttemptedAsync(CachedCallAttemptContext context, CancellationToken ct)
|
||||
{
|
||||
// Per-attempt row: kind discriminates channel, status carries the
|
||||
// per-attempt result.
|
||||
var kind = ChannelToAttemptKind(context.Channel);
|
||||
var status = context.Outcome == CachedCallAttemptOutcome.Delivered
|
||||
? AuditStatus.Attempted
|
||||
: AuditStatus.Attempted;
|
||||
// (Note: per the M3 brief and alog.md §4 the per-attempt row always
|
||||
// carries Status=Attempted; success vs. failure is captured by the
|
||||
// companion HttpStatus / ErrorMessage fields, NOT by flipping the
|
||||
// status. CachedResolve carries the terminal Status.)
|
||||
|
||||
var packet = BuildPacket(
|
||||
context,
|
||||
kind: kind,
|
||||
status: status,
|
||||
// Operational status mirror — for the per-attempt row the
|
||||
// operational state is the running status; the bridge always
|
||||
// writes "Attempted" so reconciliation can't roll back.
|
||||
operationalStatus: "Attempted",
|
||||
terminalAtUtc: null,
|
||||
lastError: context.LastError,
|
||||
httpStatus: context.HttpStatus);
|
||||
|
||||
await _forwarder.ForwardAsync(packet, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task EmitResolveAsync(CachedCallAttemptContext context, CancellationToken ct)
|
||||
{
|
||||
var (auditStatus, operationalStatus) = TerminalOutcomeToStatuses(context.Outcome);
|
||||
|
||||
var packet = BuildPacket(
|
||||
context,
|
||||
kind: AuditKind.CachedResolve,
|
||||
status: auditStatus,
|
||||
operationalStatus: operationalStatus,
|
||||
terminalAtUtc: context.OccurredAtUtc,
|
||||
lastError: context.LastError,
|
||||
httpStatus: context.HttpStatus);
|
||||
|
||||
await _forwarder.ForwardAsync(packet, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private static CachedCallTelemetry BuildPacket(
|
||||
CachedCallAttemptContext context,
|
||||
AuditKind kind,
|
||||
AuditStatus status,
|
||||
string operationalStatus,
|
||||
DateTime? terminalAtUtc,
|
||||
string? lastError,
|
||||
int? httpStatus)
|
||||
{
|
||||
var channel = ChannelStringToEnum(context.Channel);
|
||||
|
||||
return new CachedCallTelemetry(
|
||||
Audit: new AuditEvent
|
||||
{
|
||||
EventId = Guid.NewGuid(),
|
||||
OccurredAtUtc = DateTime.SpecifyKind(context.OccurredAtUtc, DateTimeKind.Utc),
|
||||
Channel = channel,
|
||||
Kind = kind,
|
||||
CorrelationId = context.TrackedOperationId.Value,
|
||||
SourceSiteId = string.IsNullOrEmpty(context.SourceSite) ? null : context.SourceSite,
|
||||
SourceInstanceId = context.SourceInstanceId,
|
||||
SourceScript = null, // Not threaded through S&F; left null on retry-loop rows.
|
||||
Target = context.Target,
|
||||
Status = status,
|
||||
HttpStatus = httpStatus,
|
||||
DurationMs = context.DurationMs,
|
||||
ErrorMessage = lastError,
|
||||
ForwardState = AuditForwardState.Pending,
|
||||
},
|
||||
Operational: new SiteCallOperational(
|
||||
TrackedOperationId: context.TrackedOperationId,
|
||||
Channel: context.Channel,
|
||||
Target: context.Target,
|
||||
SourceSite: context.SourceSite,
|
||||
Status: operationalStatus,
|
||||
RetryCount: context.RetryCount,
|
||||
LastError: lastError,
|
||||
HttpStatus: httpStatus,
|
||||
CreatedAtUtc: DateTime.SpecifyKind(context.CreatedAtUtc, DateTimeKind.Utc),
|
||||
UpdatedAtUtc: DateTime.SpecifyKind(context.OccurredAtUtc, DateTimeKind.Utc),
|
||||
TerminalAtUtc: terminalAtUtc is null
|
||||
? null
|
||||
: DateTime.SpecifyKind(terminalAtUtc.Value, DateTimeKind.Utc)));
|
||||
}
|
||||
|
||||
private static AuditKind ChannelToAttemptKind(string channel) => channel switch
|
||||
{
|
||||
"ApiOutbound" => AuditKind.ApiCallCached,
|
||||
"DbOutbound" => AuditKind.DbWriteCached,
|
||||
// Defensive default — the S&F observer is filtered to cached-call
|
||||
// categories so this branch shouldn't fire in practice.
|
||||
_ => AuditKind.ApiCallCached,
|
||||
};
|
||||
|
||||
private static AuditChannel ChannelStringToEnum(string channel) => channel switch
|
||||
{
|
||||
"ApiOutbound" => AuditChannel.ApiOutbound,
|
||||
"DbOutbound" => AuditChannel.DbOutbound,
|
||||
_ => AuditChannel.ApiOutbound,
|
||||
};
|
||||
|
||||
private static (AuditStatus auditStatus, string operationalStatus) TerminalOutcomeToStatuses(
|
||||
CachedCallAttemptOutcome outcome) => outcome switch
|
||||
{
|
||||
CachedCallAttemptOutcome.Delivered =>
|
||||
(AuditStatus.Delivered, "Delivered"),
|
||||
CachedCallAttemptOutcome.PermanentFailure =>
|
||||
(AuditStatus.Parked, "Parked"),
|
||||
CachedCallAttemptOutcome.ParkedMaxRetries =>
|
||||
(AuditStatus.Parked, "Parked"),
|
||||
// TransientFailure isn't terminal — see IsTerminal — but the switch
|
||||
// is exhaustive so we route it through Failed for safety.
|
||||
CachedCallAttemptOutcome.TransientFailure =>
|
||||
(AuditStatus.Failed, "Failed"),
|
||||
_ => (AuditStatus.Failed, "Failed"),
|
||||
};
|
||||
|
||||
private static bool IsTerminal(CachedCallAttemptOutcome outcome) => outcome switch
|
||||
{
|
||||
CachedCallAttemptOutcome.Delivered => true,
|
||||
CachedCallAttemptOutcome.PermanentFailure => true,
|
||||
CachedCallAttemptOutcome.ParkedMaxRetries => true,
|
||||
CachedCallAttemptOutcome.TransientFailure => false,
|
||||
_ => false,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user