dohertj2/scadalink-design
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(store-and-forward): resolve StoreAndForward-006,007,008,009 — transactional parked reads, PipeTo, fault-isolated activity events; 002/011/012 deferred

Browse Source
This commit is contained in:
Joseph Doherty
2026-05-16 22:32:30 -04:00
parent dd7626da63
commit 9e2416b34c
6 changed files with 255 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
src/ScadaLink.StoreAndForward/ParkedMessageHandlerActor.cs
View File

@@ -31,12 +31,15 @@ public class ParkedMessageHandlerActor : ReceiveActor
var sender = Sender;
var siteId = _siteId;
// StoreAndForward-007: idiomatic PipeTo with explicit success/failure
// projections instead of ContinueWith. Both projections touch only locals
// (captured before the await), so they are safe to run off the actor thread.
_service.GetParkedMessagesAsync(category: null, msg.PageNumber, msg.PageSize)
.ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
.PipeTo(
sender,
success: result =>
{
var entries = t.Result.Messages
var entries = result.Messages
.Select(m => new ParkedMessageEntry(
MessageId: m.Id,
TargetSystem: m.Target,
@@ -51,14 +54,12 @@ public class ParkedMessageHandlerActor : ReceiveActor
.ToList();
return new ParkedMessageQueryResponse(
msg.CorrelationId, siteId, entries, t.Result.TotalCount,
msg.CorrelationId, siteId, entries, result.TotalCount,
msg.PageNumber, msg.PageSize, true, null, DateTimeOffset.UtcNow);
}
return new ParkedMessageQueryResponse(
},
failure: ex => new ParkedMessageQueryResponse(
msg.CorrelationId, siteId, [], 0, msg.PageNumber, msg.PageSize,
false, t.Exception?.GetBaseException().Message, DateTimeOffset.UtcNow);
}).PipeTo(sender);
false, ex.GetBaseException().Message, DateTimeOffset.UtcNow));
}
private void HandleRetry(ParkedMessageRetryRequest msg)
@@ -66,18 +67,13 @@ public class ParkedMessageHandlerActor : ReceiveActor
var sender = Sender;
_service.RetryParkedMessageAsync(msg.MessageId)
.ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
return new ParkedMessageRetryResponse(
msg.CorrelationId, t.Result,
t.Result ? null : "Message not found or no longer parked.");
}
return new ParkedMessageRetryResponse(
msg.CorrelationId, false, t.Exception?.GetBaseException().Message);
}).PipeTo(sender);
.PipeTo(
sender,
success: retried => new ParkedMessageRetryResponse(
msg.CorrelationId, retried,
retried ? null : "Message not found or no longer parked."),
failure: ex => new ParkedMessageRetryResponse(
msg.CorrelationId, false, ex.GetBaseException().Message));
}
private void HandleDiscard(ParkedMessageDiscardRequest msg)
@@ -85,18 +81,13 @@ public class ParkedMessageHandlerActor : ReceiveActor
var sender = Sender;
_service.DiscardParkedMessageAsync(msg.MessageId)
.ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
return new ParkedMessageDiscardResponse(
msg.CorrelationId, t.Result,
t.Result ? null : "Message not found or no longer parked.");
}
return new ParkedMessageDiscardResponse(
msg.CorrelationId, false, t.Exception?.GetBaseException().Message);
}).PipeTo(sender);
.PipeTo(
sender,
success: discarded => new ParkedMessageDiscardResponse(
msg.CorrelationId, discarded,
discarded ? null : "Message not found or no longer parked."),
failure: ex => new ParkedMessageDiscardResponse(
msg.CorrelationId, false, ex.GetBaseException().Message));
}
private static string ExtractMethodName(string payloadJson, Commons.Types.Enums.StoreAndForwardCategory category)

27
src/ScadaLink.StoreAndForward/StoreAndForwardService.cs
View File

@@ -377,9 +377,34 @@ public class StoreAndForwardService
return await _storage.GetMessageCountByOriginInstanceAsync(instanceName);
}
/// <summary>
/// WP-14: Raises the S&amp;F activity notification. StoreAndForward-009: 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 <see cref="EnqueueAsync"/> or <c>RetryMessageAsync</c> 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.
/// </summary>
private void RaiseActivity(string action, StoreAndForwardCategory category, string detail)
{
OnActivity?.Invoke(action, category, detail);
var handlers = OnActivity;
if (handlers == null) return;
foreach (var handler in handlers.GetInvocationList().Cast<Action<string, StoreAndForwardCategory, string>>())
{
try
{
handler(action, category, detail);
}
catch (Exception ex)
{
_logger.LogWarning(ex,
"Store-and-forward activity subscriber threw for action {Action}; ignored",
action);
}
}
}
}

25
src/ScadaLink.StoreAndForward/StoreAndForwardStorage.cs
View File

@@ -8,6 +8,19 @@ namespace ScadaLink.StoreAndForward;
/// WP-9: SQLite persistence layer for store-and-forward messages.
/// Uses direct Microsoft.Data.Sqlite (not EF Core) for lightweight site-side storage.
/// No max buffer size per design decision.
///
/// StoreAndForward-008: every method opens a fresh <see cref="SqliteConnection"/> for
/// the duration of the call rather than holding a long-lived connection. This is a
/// deliberate trade-off, not an oversight: Microsoft.Data.Sqlite maintains an internal
/// connection pool keyed on the connection string, so <c>OpenAsync</c> on a previously
/// used connection string reuses a pooled handle instead of performing a real file
/// open. The retry sweep therefore relies on that pool for acceptable performance —
/// it calls <see cref="RemoveMessageAsync"/> / <see cref="UpdateMessageIfStatusAsync"/>
/// once per due message, and with no max buffer size (by design) the buffer can grow
/// large. The connection-per-call style keeps each method self-contained and
/// transaction-scoped; if profiling ever shows the pooled open to be a bottleneck on
/// the hot retry path, the remedy is a batched sweep API that opens one connection (and
/// one transaction) per sweep.
/// </summary>
public class StoreAndForwardStorage
{
@@ -222,6 +235,12 @@ public class StoreAndForwardStorage
/// <summary>
/// WP-12: Gets all parked messages, optionally filtered by category, with pagination.
///
/// StoreAndForward-006: the COUNT(*) and the paged SELECT run inside a single
/// transaction so they observe one consistent snapshot. Without it, a concurrent
/// enqueue/park/discard arriving between the two statements yields a TotalCount
/// inconsistent with the returned page (flickering totals / off-by-one page math
/// in the paginated UI).
/// </summary>
public async Task<(List<StoreAndForwardMessage> Messages, int TotalCount)> GetParkedMessagesAsync(
StoreAndForwardCategory? category = null,
@@ -231,8 +250,11 @@ public class StoreAndForwardStorage
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
// Count
await using var countCmd = connection.CreateCommand();
countCmd.Transaction = transaction;
countCmd.CommandText = category.HasValue
? "SELECT COUNT(*) FROM sf_messages WHERE status = @parked AND category = @category"
: "SELECT COUNT(*) FROM sf_messages WHERE status = @parked";
@@ -242,6 +264,7 @@ public class StoreAndForwardStorage
// Page
await using var pageCmd = connection.CreateCommand();
pageCmd.Transaction = transaction;
var categoryFilter = category.HasValue ? " AND category = @category" : "";
pageCmd.CommandText = $@"
SELECT id, category, target, payload_json, retry_count, max_retries,
@@ -257,6 +280,8 @@ public class StoreAndForwardStorage
pageCmd.Parameters.AddWithValue("@offset", (pageNumber - 1) * pageSize);
var messages = await ReadMessagesAsync(pageCmd);
await transaction.CommitAsync();
return (messages, totalCount);
}