using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.SiteEventLogging;
namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;
///
/// Core store-and-forward service.
///
/// Lifecycle:
/// 1. Caller attempts immediate delivery via IDeliveryHandler
/// 2. On transient failure → buffer in SQLite → retry loop
/// 3. On success → remove from buffer
/// 4. On reaching MaxRetries → park (a MaxRetries of 0 means "no limit" — the
/// message is retried until delivered and is never parked for retry exhaustion)
/// 5. Permanent failures are returned to caller immediately (never buffered)
///
/// Fixed retry interval (not exponential). Per-source-entity retry settings.
/// Background timer-based retry sweep.
///
/// Parked messages queryable, retryable, and discardable.
///
/// Buffer depth reported as health metric. Activity logged to site event log.
///
/// CachedCall idempotency is the caller's responsibility.
/// This service does not deduplicate — if the same message is enqueued twice,
/// it will be delivered twice. Callers using ExternalSystem.CachedCall() must
/// design their payloads to be idempotent (e.g., include unique request IDs
/// and handle duplicate detection on the remote end).
///
public class StoreAndForwardService
{
private readonly StoreAndForwardStorage _storage;
private readonly StoreAndForwardOptions _options;
private readonly ReplicationService? _replication;
private readonly ILogger _logger;
///
/// Site-side observer notified
/// after every cached-call delivery attempt. Optional — when null no
/// telemetry is emitted; the legacy retry loop behaviour is
/// preserved exactly.
///
private readonly ICachedCallLifecycleObserver? _cachedCallObserver;
///
/// Optional site operational-event log. When non-null the service maps
/// its own buffer/retry/park activity (the same activity that drives
/// ) onto site events — store_and_forward for the
/// cached-call categories and notification for the site's
/// forward-to-central notification path. Best-effort and fire-and-forget so a
/// failing logger never affects delivery bookkeeping.
///
private readonly ISiteEventLogger? _siteEventLogger;
///
/// Site id stamped onto the
/// cached-call attempt context so the audit bridge can build the
/// half of the telemetry packet.
///
private readonly string _siteId;
///
/// Distinctive marker stamped onto cached-call audit
/// telemetry when the host has not registered an
/// . Chosen with a leading $
/// so it cannot collide with a real site id (which is a configuration
/// identifier and never starts with $). Surfacing this in the
/// central audit log makes a missing site-context binding immediately
/// recognisable instead of an unattributable empty string.
///
public const string UnknownSiteSentinel = "$unknown-site";
///
/// The detail-string prefix written by
/// when an immediate forward attempt throws and the message is buffered for
/// the retry sweep. matches on this same prefix
/// to distinguish a forward failure (logged) from a routine
/// no-handler enqueue (not logged), so both the construction site and the
/// check reference this single constant rather than duplicating the
/// literal — keeping the two ends from drifting apart.
///
private const string BufferedForRetryDetailPrefix = "Buffered for retry";
private Timer? _retryTimer;
private int _retryInProgress;
///
/// Active-node delivery gate. When set, the retry sweep runs ONLY when the
/// gate reports true — the standby node applies replicated buffer operations
/// but must never deliver (Component-StoreAndForward.md "standby-passive"
/// model). Re-evaluated on every sweep tick so a failover resumes delivery
/// within one RetryTimerInterval without re-wiring. Null (tests, central
/// hosts) preserves the ungated legacy behaviour. A throwing gate is treated
/// as standby — safe-by-default, mirroring SiteCommunicationActor's
/// DefaultIsActiveCheck fallback.
///
private Func? _deliveryGate;
/// Installs the active-node delivery gate (see ).
public void SetDeliveryGate(Func gate) => _deliveryGate = gate;
///
/// The in-flight retry sweep , or
/// null when no sweep is currently running. Captured when the timer
/// callback starts a sweep so can wait for it to
/// finish before the host disposes downstream dependencies
/// (, ) that the sweep is
/// still touching. Written from the timer thread and from
/// , so reads are synchronised via the
/// APIs.
///
private Task? _sweepTask;
///
/// How long waits for an
/// in-flight retry sweep to finish before returning. The default — 10 s —
/// is generous enough to let a typical sweep over the buffered queue drain,
/// but bounded so a hung downstream call (a stuck SQLite write, a
/// long-running delivery handler) cannot block host shutdown indefinitely.
/// On timeout the wait is abandoned and the timer is still disposed; the
/// sweep keeps running but will throw on the next call into a disposed
/// dependency — preferred to blocking shutdown forever.
///
private static readonly TimeSpan SweepShutdownWaitTimeout = TimeSpan.FromSeconds(10);
///
/// Cached count of messages currently buffered for
/// forwarding — i.e. rows in ,
/// the live store-and-forward queue waiting to be delivered. This backs the
/// scadabridge.store_and_forward.queue.depth observable gauge.
///
/// The gauge's collection callback is synchronous and is invoked frequently by
/// the OpenTelemetry/Prometheus collector, so it must never run an async SQLite
/// COUNT(*). Instead this is seeded once from storage
/// in and then adjusted in-process on the existing
/// paths that change the Pending population: (+1),
/// successful-retry removal and Pending→Parked transitions in
/// (-1), and operator requeue in
/// (+1). The provider registered with
/// reads it via
/// — non-blocking and sync-safe. It is an
/// approximate, eventually-consistent gauge (concurrent failover replication
/// applies to the standby's own store, not this counter), which is exactly
/// what a queue-depth metric needs.
///
///
private long _bufferedCount;
///
/// Test seam: simulates a concurrent pre-seed
/// increment landing on
/// before seeds it, so a test can prove the seed uses
/// (additive) rather than Exchange (clobbering).
///
internal void TestOnly_IncrementBufferedCount() =>
Interlocked.Increment(ref _bufferedCount);
///
/// An instance field that guards against a single instance
/// registering the queue-depth provider (and re-seeding the counter) more than
/// once — e.g. a second on the same instance. It does NOT
/// coordinate across instances: the gauge slot in
/// is process-global, so in a multi-instance process the last
/// wins the global slot. 0 = not yet registered, 1 = done.
///
private int _queueDepthProviderRegistered;
///
/// The exact provider delegate this instance registered with
/// the process-global gauge slot, retained so
/// can deregister it by identity (compare-and-clear). Holding
/// the same reference both registers and clears the slot; the identity check ensures a
/// late stop of this instance cannot clear a newer instance's provider. Null until
/// registers.
///
private Func? _queueDepthProvider;
///
/// Delivery handler delegate. The return value / exception is interpreted
/// the same way on both the immediate-delivery path ()
/// and the background retry path (RetryMessageAsync):
///
/// - true — delivered successfully. The message is
/// removed from the buffer (or, on the immediate path, never buffered).
/// - false — permanent failure. On the immediate path
/// the message is NOT buffered; on a retry the message is already buffered and
/// is parked immediately (no further retries).
/// - throws — transient failure. On the immediate path the
/// message is buffered for retry; on a retry the retry count is incremented and
/// the message is parked once is
/// reached.
///
///
private readonly Dictionary>> _deliveryHandlers = new();
///
/// Event callback for logging S&F activity to site event log.
///
public event Action? OnActivity;
///
/// Initializes a new instance of the StoreAndForwardService.
///
/// The storage backend for buffered messages.
/// Configuration options.
/// Logger instance.
/// Optional replication service for standby synchronization.
/// Optional observer for cached call lifecycle events.
/// The site identifier this service belongs to.
///
/// Optional site operational-event log. When non-null, buffer/retry/park
/// activity is mirrored to site events (store_and_forward /
/// notification by category). Optional with a null default so the
/// many direct-construction tests still compile unchanged.
///
public StoreAndForwardService(
StoreAndForwardStorage storage,
StoreAndForwardOptions options,
ILogger logger,
ReplicationService? replication = null,
ICachedCallLifecycleObserver? cachedCallObserver = null,
string siteId = "",
ISiteEventLogger? siteEventLogger = null)
{
_storage = storage;
_options = options;
_logger = logger;
_replication = replication;
_cachedCallObserver = cachedCallObserver;
_siteId = string.IsNullOrWhiteSpace(siteId) ? UnknownSiteSentinel : siteId;
_siteEventLogger = siteEventLogger;
// Ride the existing activity hook to emit site operational events.
// RaiseActivity already isolates a throwing subscriber, so a failing
// event log can never be misclassified as a transient delivery failure.
// Only subscribe when a logger is wired so the
// legacy (test/central) construction path stays a no-op.
if (_siteEventLogger != null)
{
OnActivity += EmitSiteEvent;
}
}
///
/// Maps one store-and-forward activity to a site operational event,
/// following the Site Event Logging spec's per-category scope
/// (Component-SiteEventLogging.md §"Events Logged"):
///
/// - Cached-call categories
/// ( /
/// ) log under
/// store_and_forward for queued / retried / parked / retry-delivered
/// activity.
/// - The site's notification forward-to-central path
/// () logs under
/// notification ONLY on a forward FAILURE (buffered after the
/// immediate forward threw) or a park (long-buffered / retries exhausted).
/// Routine enqueue and forward-success are deliberately NOT logged — central's
/// Notifications table is the record of audit; the site only fills the
/// in-transit blind spot when central is unreachable.
///
/// A successful immediate cached-call Delivered is the normal hot path and
/// is not logged.
///
private void EmitSiteEvent(string action, StoreAndForwardCategory category, string detail)
{
var logger = _siteEventLogger;
if (logger == null)
{
return;
}
// An immediate-delivery success is the normal hot path, not an
// operational event. A retry-loop success (detail "Delivered to … after
// N retries") IS logged for cached calls — it records a recovery.
if (action == "Delivered" && detail.StartsWith("Immediate", StringComparison.Ordinal))
{
return;
}
if (category == StoreAndForwardCategory.Notification)
{
// Spec: log only forward-failure (the immediate forward threw and the
// notification was buffered for retry — detail prefixed
// BufferedForRetryDetailPrefix) and park. A routine "No handler
// registered, buffered" enqueue and a forward-success "Delivered"
// are deliberately NOT logged.
var isForwardFailure = action == "Queued"
&& detail.StartsWith(BufferedForRetryDetailPrefix, StringComparison.Ordinal);
if (!isForwardFailure && action != "Parked")
{
return;
}
var notifSeverity = action == "Parked" ? "Error" : "Warning";
_ = logger.LogEventAsync(
"notification", notifSeverity, instanceId: null,
source: "StoreAndForwardService",
message: $"Notification {action.ToLowerInvariant()}: {detail}");
return;
}
// Cached-call categories: queued / retried / parked / retry-delivered.
// Severity: parking is an Error (delivery abandoned for retry purposes);
// queue/retry/requeue are Warning; a retry-loop Delivered is Info.
var severity = action switch
{
"Parked" => "Error",
"Delivered" => "Info",
_ => "Warning",
};
_ = logger.LogEventAsync(
"store_and_forward", severity, instanceId: null,
source: "StoreAndForwardService",
message: $"Operation {action.ToLowerInvariant()}: {detail}");
}
///
/// Registers a delivery handler for a given message category. See the
/// _deliveryHandlers field documentation for the true/false/throws contract,
/// which applies identically on the immediate and retry paths.
///
/// The message category to handle.
/// The delivery handler function.
public void RegisterDeliveryHandler(
StoreAndForwardCategory category,
Func> handler)
{
_deliveryHandlers[category] = handler;
}
///
/// Initializes storage and starts the background retry timer.
///
/// A task representing the asynchronous start operation.
public async Task StartAsync()
{
await _storage.InitializeAsync();
var pending = await _storage.GetMessageCountByStatusAsync(
StoreAndForwardMessageStatus.Pending);
if (Interlocked.CompareExchange(ref _queueDepthProviderRegistered, 1, 0) == 0)
{
Interlocked.Add(ref _bufferedCount, pending);
var provider = (Func)(() => Interlocked.Read(ref _bufferedCount));
_queueDepthProvider = provider;
ScadaBridgeTelemetry.SetQueueDepthProvider(provider);
}
_retryTimer = new Timer(
_ => Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync()),
null,
_options.RetryTimerInterval,
_options.RetryTimerInterval);
_logger.LogInformation(
"Store-and-forward service started. Retry interval: {Interval}s",
_options.DefaultRetryInterval.TotalSeconds);
}
///
/// Stops the background retry timer and waits (bounded) for any in-flight
/// retry sweep to finish before returning.
///
/// A task representing the asynchronous stop operation.
public async Task StopAsync()
{
if (_retryTimer != null)
{
// Stop the periodic callback first so no new sweep starts while we
// are waiting for the in-flight one to drain.
await _retryTimer.DisposeAsync();
_retryTimer = null;
}
var inflight = Volatile.Read(ref _sweepTask);
if (inflight is not null && !inflight.IsCompleted)
{
try
{
// WaitAsync with a finite timeout: a hung delivery handler /
// storage call cannot block host shutdown indefinitely. On timeout
// the sweep keeps running but the host is free to proceed with
// disposal — preferred to never returning.
await inflight.WaitAsync(SweepShutdownWaitTimeout).ConfigureAwait(false);
}
catch (TimeoutException)
{
_logger.LogWarning(
"Store-and-forward retry sweep did not finish within {Timeout}; " +
"shutdown is proceeding while the sweep is still in-flight",
SweepShutdownWaitTimeout);
}
catch (Exception ex)
{
// The sweep itself already logs at Error on failure (see
// RetryPendingMessagesAsync's catch); we only log here so a
// surprise fault during shutdown is still visible. Swallow so the
// host's shutdown sequence can continue regardless.
_logger.LogWarning(ex,
"Store-and-forward retry sweep faulted during shutdown wait");
}
}
var provider = _queueDepthProvider;
if (provider is not null)
{
ScadaBridgeTelemetry.ClearQueueDepthProvider(provider);
_queueDepthProvider = null;
}
Interlocked.Exchange(ref _bufferedCount, 0);
Interlocked.Exchange(ref _queueDepthProviderRegistered, 0);
}
///
/// Enqueues a message for store-and-forward delivery.
/// Attempts immediate delivery first. On transient failure, buffers for retry.
/// On permanent failure (handler returns false), returns false immediately.
///
/// Retry-count lifecycle — the immediate (or caller-made) delivery attempt
/// is attempt 0 and is not counted; the background retry sweep increments
/// on each retry. A buffered
/// message is parked once RetryCount reaches
/// — but only when is greater than 0. A
/// of 0 means no limit: the message is
/// retried on every sweep until it is delivered and is never parked on a
/// retry-count basis. It is therefore not a "do not retry" value — callers
/// that want delivery abandoned after a bounded number of attempts must pass a
/// positive .
///
/// CachedCall idempotency note — this method does not deduplicate.
/// The caller (e.g., ExternalSystem.CachedCall()) is responsible for ensuring
/// that the remote system can handle duplicate deliveries safely.
///
/// Message category (selects the delivery handler).
/// Target system name (external system / notification list / DB connection).
/// JSON-serialized call payload, treated opaquely.
/// Instance that originated the message (survives instance deletion).
///
/// Maximum background retry-sweep attempts before the message is parked.
/// 0 = no limit — the message is retried on every sweep until
/// delivered and is never parked for exhausting retries; it is not a
/// "never retry" value. null uses .
/// Must be positive to bound delivery attempts. Mirrors the
/// contract.
///
/// Fixed interval between retry sweeps for this message; null uses the configured default.
///
/// When false, the caller has already made its own delivery attempt and the
/// message is buffered directly for the retry sweep (the handler is not invoked here).
///
///
/// An explicit, caller-supplied message id. null (the default) makes the
/// service mint a fresh GUID. The Notification Outbox enqueue path supplies its own
/// id so the script-generated NotificationId is the single idempotency key —
/// it is the buffered row's , it is carried
/// inside the payload, and it is the id the forwarder submits to central.
///
///
/// The originating script execution's
/// per-run correlation id. Threaded onto the buffered row so the retry-loop
/// cached-call audit rows carry it. null for callers (e.g. notifications)
/// that do not supply one.
///
///
/// The originating script identifier,
/// threaded onto the buffered row alongside
/// so the retry-loop audit rows carry the same provenance the script-side
/// cached rows do. null when not known.
///
///
/// The ExecutionId of the
/// inbound-API request that spawned the originating script execution.
/// Threaded onto the buffered row alongside
/// so the retry-loop cached-call audit rows carry it. null for a
/// non-routed run and for callers (e.g. notifications) that
/// do not supply one.
///
/// A task that resolves to a result indicating whether the message was delivered or buffered.
public async Task EnqueueAsync(
StoreAndForwardCategory category,
string target,
string payloadJson,
string? originInstanceName = null,
int? maxRetries = null,
TimeSpan? retryInterval = null,
bool attemptImmediateDelivery = true,
string? messageId = null,
Guid? executionId = null,
string? sourceScript = null,
Guid? parentExecutionId = null)
{
var message = new StoreAndForwardMessage
{
Id = messageId ?? Guid.NewGuid().ToString("N"),
Category = category,
Target = target,
PayloadJson = payloadJson,
RetryCount = 0,
MaxRetries = maxRetries ?? _options.DefaultMaxRetries,
RetryIntervalMs = (long)(retryInterval ?? _options.DefaultRetryInterval).TotalMilliseconds,
CreatedAt = DateTimeOffset.UtcNow,
Status = StoreAndForwardMessageStatus.Pending,
OriginInstanceName = originInstanceName,
ExecutionId = executionId,
SourceScript = sourceScript,
ParentExecutionId = parentExecutionId
};
// Attempt immediate delivery — unless the caller has already made a
// delivery attempt of its own (attemptImmediateDelivery: false). In that
// case re-invoking the handler here would dispatch the request twice.
if (attemptImmediateDelivery && _deliveryHandlers.TryGetValue(category, out var handler))
{
try
{
var success = await handler(message);
if (success)
{
RaiseActivity("Delivered", category, $"Immediate delivery to {target}");
return new StoreAndForwardResult(true, message.Id, false);
}
// Permanent failure — do not buffer
return new StoreAndForwardResult(false, message.Id, false);
}
catch (Exception ex)
{
// Transient failure — buffer for retry. The immediate attempt is
// attempt 0; RetryCount tracks only sweep retries, so it stays 0
// here.
_logger.LogWarning(ex,
"Immediate delivery to {Target} failed (transient), buffering for retry",
target);
message.LastAttemptAt = DateTimeOffset.UtcNow;
message.LastError = ex.Message;
await BufferAsync(message);
RaiseActivity("Queued", category, $"{BufferedForRetryDetailPrefix}: {target} ({ex.Message})");
return new StoreAndForwardResult(true, message.Id, true);
}
}
// Either no handler is registered yet, or the caller already attempted
// delivery itself — buffer for the background retry sweep to deliver.
// The initial attempt (caller-made, or skipped because no handler is
// registered) is attempt 0; RetryCount tracks only sweep retries and
// therefore stays 0 here.
if (!attemptImmediateDelivery)
{
message.LastAttemptAt = DateTimeOffset.UtcNow;
}
await BufferAsync(message);
RaiseActivity("Queued", category, attemptImmediateDelivery
? $"No handler registered, buffered: {target}"
: $"{BufferedForRetryDetailPrefix}: {target}");
return new StoreAndForwardResult(true, message.Id, true);
}
///
/// Persists a message to the local SQLite buffer and replicates the
/// add to the standby node so a failover does not lose the buffered message.
///
private async Task BufferAsync(StoreAndForwardMessage message)
{
await _storage.EnqueueAsync(message);
_replication?.ReplicateEnqueue(message);
Interlocked.Increment(ref _bufferedCount);
}
///
/// Background retry sweep. Processes all pending messages that are due for retry.
///
/// A task representing the asynchronous retry sweep.
internal async Task RetryPendingMessagesAsync()
{
// Prevent overlapping retry sweeps
if (Interlocked.CompareExchange(ref _retryInProgress, 1, 0) != 0)
return;
try
{
var gate = _deliveryGate;
if (gate != null)
{
bool isActive;
try { isActive = gate(); }
catch (Exception ex)
{
_logger.LogWarning(ex,
"S&F delivery gate threw; treating this node as standby for this sweep");
isActive = false;
}
if (!isActive)
{
_logger.LogDebug("S&F retry sweep skipped: this node is not the active site node");
return;
}
}
var messages = await _storage.GetMessagesForRetryAsync(_options.SweepBatchLimit);
if (messages.Count == 0) return;
_logger.LogDebug("Retry sweep: {Count} messages due for retry", messages.Count);
var failedTargets = new HashSet<(StoreAndForwardCategory, string)>();
foreach (var message in messages)
{
// One transient failure per (category, target) per sweep: the target
// is down — burning a full timeout per remaining message serializes
// the sweep into hours under backlog (arch review 02, Performance #1).
// Skipped rows keep their RetryCount/LastAttemptAt untouched.
if (failedTargets.Contains((message.Category, message.Target)))
continue;
var outcome = await RetryMessageAsync(message);
if (outcome == RetryOutcome.TransientFailure)
failedTargets.Add((message.Category, message.Target));
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Error during retry sweep");
}
finally
{
Interlocked.Exchange(ref _retryInProgress, 0);
}
}
///
/// Outcome of a single message's retry attempt, used by the sweep to
/// short-circuit a (category, target) lane after its first transient
/// failure. = no attempt was made (no handler, or the
/// row's status changed under a concurrent apply); its RetryCount is untouched.
///
internal enum RetryOutcome { Delivered, Parked, TransientFailure, Skipped }
private async Task RetryMessageAsync(StoreAndForwardMessage message)
{
if (!_deliveryHandlers.TryGetValue(message.Category, out var handler))
{
_logger.LogWarning("No delivery handler for category {Category}", message.Category);
return RetryOutcome.Skipped;
}
// Measure per-attempt
// duration so the audit row carries a meaningful DurationMs. Captured
// around the handler invocation only — storage / replication overhead
// is excluded.
var attemptStartUtc = DateTime.UtcNow;
var attemptStopwatch = System.Diagnostics.Stopwatch.StartNew();
try
{
var success = await handler(message);
attemptStopwatch.Stop();
if (success)
{
await _storage.RemoveMessageAsync(message.Id);
_replication?.ReplicateRemove(message.Id);
Interlocked.Decrement(ref _bufferedCount);
RaiseActivity("Delivered", message.Category,
$"Delivered to {message.Target} after {message.RetryCount} retries");
// Terminal Delivered observer notification — the audit
// bridge maps this to Attempted + CachedResolve(Delivered).
await NotifyCachedCallObserverAsync(
message,
CachedCallAttemptOutcome.Delivered,
lastError: null,
httpStatus: null,
occurredAtUtc: attemptStartUtc,
durationMs: (int)attemptStopwatch.ElapsedMilliseconds);
return RetryOutcome.Delivered;
}
// Permanent failure on retry — park immediately.
message.Status = StoreAndForwardMessageStatus.Parked;
message.LastAttemptAt = DateTimeOffset.UtcNow;
message.LastError = "Permanent failure (handler returned false)";
var parked = await _storage.UpdateMessageIfStatusAsync(
message, StoreAndForwardMessageStatus.Pending);
if (!parked)
{
_logger.LogDebug(
"Message {MessageId} changed status during delivery; sweep park skipped",
message.Id);
return RetryOutcome.Skipped;
}
Interlocked.Decrement(ref _bufferedCount);
_replication?.ReplicatePark(message);
RaiseActivity("Parked", message.Category,
$"Permanent failure for {message.Target}: handler returned false");
// Terminal PermanentFailure observer notification — the
// audit bridge maps this to Attempted(Failed) + CachedResolve(Parked).
await NotifyCachedCallObserverAsync(
message,
CachedCallAttemptOutcome.PermanentFailure,
lastError: message.LastError,
httpStatus: null,
occurredAtUtc: attemptStartUtc,
durationMs: (int)attemptStopwatch.ElapsedMilliseconds);
return RetryOutcome.Parked;
}
catch (Exception ex)
{
attemptStopwatch.Stop();
// Transient failure — increment retry, check max
message.RetryCount++;
message.LastAttemptAt = DateTimeOffset.UtcNow;
message.LastError = ex.Message;
if (message.MaxRetries > 0 && message.RetryCount >= message.MaxRetries)
{
message.Status = StoreAndForwardMessageStatus.Parked;
var parked = await _storage.UpdateMessageIfStatusAsync(
message, StoreAndForwardMessageStatus.Pending);
if (!parked)
{
_logger.LogDebug(
"Message {MessageId} changed status during delivery; sweep park skipped",
message.Id);
return RetryOutcome.Skipped;
}
Interlocked.Decrement(ref _bufferedCount);
_replication?.ReplicatePark(message);
RaiseActivity("Parked", message.Category,
$"Max retries ({message.MaxRetries}) reached for {message.Target}");
_logger.LogWarning(
"Message {MessageId} parked after {MaxRetries} retries to {Target}",
message.Id, message.MaxRetries, message.Target);
// Terminal ParkedMaxRetries observer notification — the
// audit bridge maps this to Attempted(Failed) + CachedResolve(Parked).
await NotifyCachedCallObserverAsync(
message,
CachedCallAttemptOutcome.ParkedMaxRetries,
lastError: ex.Message,
httpStatus: null,
occurredAtUtc: attemptStartUtc,
durationMs: (int)attemptStopwatch.ElapsedMilliseconds);
return RetryOutcome.Parked;
}
else
{
if (!await _storage.UpdateMessageIfStatusAsync(
message, StoreAndForwardMessageStatus.Pending))
{
_logger.LogDebug(
"Message {MessageId} changed status during delivery; sweep retry-count update skipped",
message.Id);
return RetryOutcome.Skipped;
}
RaiseActivity("Retried", message.Category,
$"Retry {message.RetryCount}/{message.MaxRetries} for {message.Target}: {ex.Message}");
// Per-attempt TransientFailure observer notification —
// the audit bridge maps this to Attempted(Failed).
await NotifyCachedCallObserverAsync(
message,
CachedCallAttemptOutcome.TransientFailure,
lastError: ex.Message,
httpStatus: null,
occurredAtUtc: attemptStartUtc,
durationMs: (int)attemptStopwatch.ElapsedMilliseconds);
return RetryOutcome.TransientFailure;
}
}
}
///
/// Notify the registered
/// of the just-completed
/// attempt. Only fires for cached-call categories
/// ( and
/// ); the
/// category has its
/// own central-side audit pipeline (Notification Outbox) and must
/// not surface on this hook.
///
///
/// Best-effort: an observer that throws is logged and swallowed so a
/// failing audit pipeline cannot corrupt S&F retry bookkeeping
/// (alog.md §7 contract). Messages whose ids are not valid GUIDs (callers
/// that didn't thread a TrackedOperationId in) are silently
/// skipped — the observer requires a parseable id by contract.
///
private async Task NotifyCachedCallObserverAsync(
StoreAndForwardMessage message,
CachedCallAttemptOutcome outcome,
string? lastError,
int? httpStatus,
DateTime occurredAtUtc,
int? durationMs)
{
if (_cachedCallObserver == null)
{
return;
}
// Only cached-call categories generate audit telemetry on this hook —
// notifications have their own outbox-side audit pipeline.
var channel = message.Category switch
{
StoreAndForwardCategory.ExternalSystem => "ApiOutbound",
StoreAndForwardCategory.CachedDbWrite => "DbOutbound",
_ => null,
};
if (channel is null)
{
return;
}
if (!TrackedOperationId.TryParse(message.Id, out var trackedId))
{
_logger.LogWarning(
"Cached-call audit observer skipped: message id {MessageId} is not a parseable TrackedOperationId (category {Category}, outcome {Outcome}). " +
"Audit lifecycle for this operation will have no rows.",
message.Id, message.Category, outcome);
return;
}
CachedCallAttemptContext context;
try
{
context = new CachedCallAttemptContext(
TrackedOperationId: trackedId,
Channel: channel,
Target: message.Target,
SourceSite: _siteId,
Outcome: outcome,
RetryCount: message.RetryCount,
LastError: lastError,
HttpStatus: httpStatus,
CreatedAtUtc: message.CreatedAt.UtcDateTime,
OccurredAtUtc: DateTime.SpecifyKind(occurredAtUtc, DateTimeKind.Utc),
DurationMs: durationMs,
SourceInstanceId: message.OriginInstanceName,
// The buffered message carries the originating script
// execution's ExecutionId + SourceScript; surface them on the
// context so the bridge can stamp the retry-loop cached audit
// rows. Null on rows buffered before this field existed
// (back-compat).
ExecutionId: message.ExecutionId,
SourceScript: message.SourceScript,
// The buffered message also carries the spawning inbound-API
// request's ExecutionId; surface it so the bridge stamps it
// onto the retry-loop cached rows. Null for a non-routed run
// and on rows buffered before this field existed (back-compat).
ParentExecutionId: message.ParentExecutionId);
}
catch (Exception buildEx)
{
// Defensive — record construction shouldn't throw, but the alog.md
// §7 contract requires this path be exception-safe regardless.
_logger.LogWarning(buildEx,
"Failed to build cached-call attempt context for {MessageId}; observer skipped",
message.Id);
return;
}
try
{
await _cachedCallObserver.OnAttemptCompletedAsync(context, CancellationToken.None)
.ConfigureAwait(false);
}
catch (Exception ex)
{
// alog.md §7 best-effort: an audit observer outage must NEVER be
// misclassified as a transient delivery failure or corrupt the
// S&F retry bookkeeping.
_logger.LogWarning(ex,
"ICachedCallLifecycleObserver threw for {MessageId} (Outcome {Outcome}); ignored",
message.Id, outcome);
}
}
///
/// Gets parked messages for central query (Pattern 8).
///
/// Optional category filter, or null for all categories.
/// The page number (1-based).
/// The page size.
/// A tuple of parked messages and the total count.
public async Task<(List Messages, int TotalCount)> GetParkedMessagesAsync(
StoreAndForwardCategory? category = null,
int pageNumber = 1,
int pageSize = 50)
{
return await _storage.GetParkedMessagesAsync(category, pageNumber, pageSize);
}
///
/// Retries a parked message (moves back to pending queue).
///
/// An operator requeue is a buffer state change and is
/// replicated to the standby (as a )
/// so a failover preserves the operator's retry intent.
/// The activity-log entry carries the message's true
/// category rather than a hard-coded one.
/// The parked row is captured before the local
/// requeue write rather than re-read after it, so a concurrent
/// RemoveMessageAsync or DiscardParkedMessageAsync running
/// between the two storage calls cannot leave the standby in Parked
/// while the active node has already requeued — we always have the row in
/// hand for the Requeue replication.
///
/// The identifier of the message to retry.
/// True if successfully retried, false otherwise.
public async Task RetryParkedMessageAsync(string messageId)
{
var captured = await _storage.GetMessageByIdAsync(messageId);
if (captured is null || captured.Status != StoreAndForwardMessageStatus.Parked)
{
return false;
}
var success = await _storage.RetryParkedMessageAsync(messageId);
if (!success)
{
return false;
}
Interlocked.Increment(ref _bufferedCount);
// The active node just rewrote this row to Pending with retry_count = 0
// and cleared last_error / last_attempt_at (see
// StoreAndForwardStorage.RetryParkedMessageAsync). Reconstruct the
// post-requeue state on the captured POCO so the standby applies the
// same mutations even if a concurrent writer has already deleted the
// row underneath us.
captured.Status = StoreAndForwardMessageStatus.Pending;
captured.RetryCount = 0;
captured.LastError = null;
captured.LastAttemptAt = null;
_replication?.ReplicateRequeue(captured);
RaiseActivity("Retry", captured.Category,
$"Parked message {messageId} moved back to queue");
return true;
}
///
/// Permanently discards a parked message.
///
/// An operator discard is a buffer removal and is replicated
/// to the standby (as a ) so the
/// discarded message does not reappear after a failover.
/// The activity-log entry carries the message's true
/// category rather than a hard-coded one.
///
/// The identifier of the message to discard.
/// True if successfully discarded, false otherwise.
public async Task DiscardParkedMessageAsync(string messageId)
{
// Capture the category before the row is deleted so the activity log is
// labelled correctly.
var message = await _storage.GetMessageByIdAsync(messageId);
var success = await _storage.DiscardParkedMessageAsync(messageId);
if (success)
{
_replication?.ReplicateRemove(messageId);
RaiseActivity("Discard", message?.Category ?? StoreAndForwardCategory.ExternalSystem,
$"Parked message {messageId} discarded");
}
return success;
}
///
/// Gets buffer depth by category for health reporting.
///
/// A dictionary of buffer depths by category.
public async Task> GetBufferDepthAsync()
{
return await _storage.GetBufferDepthByCategoryAsync();
}
///
/// Gets count of S&F messages for a given instance (for verifying survival on deletion).
///
/// The instance name to query.
/// The number of messages originating from the instance.
public async Task GetMessageCountForInstanceAsync(string instanceName)
{
return await _storage.GetMessageCountByOriginInstanceAsync(instanceName);
}
///
/// Notification Outbox: looks up a buffered message by its id, or null if it
/// is not (or no longer) in the buffer. Notify.Status uses this to detect a
/// notification still in transit at the site — central reports it not-found while
/// the S&F buffer still holds it, which is the site-local Forwarding state.
///
/// The message identifier.
/// The message, or null if not found.
public async Task GetMessageByIdAsync(string messageId)
{
return await _storage.GetMessageByIdAsync(messageId);
}
///
/// Raises the S&F activity notification. The
/// delegate is snapshotted (so a concurrent unsubscribe cannot NRE) and every
/// subscriber invocation is wrapped so a slow/throwing subscriber (e.g. the site
/// event log) cannot abort the caller. Crucially, a subscriber exception raised
/// from or RetryMessageAsync must NOT be
/// misclassified as a transient delivery failure — pre-fix it escaped into the
/// delivery try/catch and caused a successfully delivered message to be buffered
/// (or its retry count to be bumped). Activity logging is best-effort.
///
private void RaiseActivity(string action, StoreAndForwardCategory category, string detail)
{
var handlers = OnActivity;
if (handlers == null) return;
foreach (var handler in handlers.GetInvocationList().Cast>())
{
try
{
handler(action, category, detail);
}
catch (Exception ex)
{
_logger.LogWarning(ex,
"Store-and-forward activity subscriber threw for action {Action}; ignored",
action);
}
}
}
}
///
/// Result of an enqueue operation.
///
public record StoreAndForwardResult(
/// True if the message was accepted (either delivered immediately or buffered).
bool Accepted,
/// Unique message ID for tracking.
string MessageId,
/// True if the message was buffered (not delivered immediately).
bool WasBuffered);