diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs index 7155c0fc..145c15dd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs @@ -137,6 +137,19 @@ public interface IOperationTrackingStore /// the caller advance the cursor monotonically across follow-up pulls. /// /// + /// Composite keyset (Task 15). When is + /// supplied the read uses a (UpdatedAtUtc, TrackedOperationId) keyset: + /// it returns rows strictly after that cursor — UpdatedAtUtc > sinceUtc, + /// or UpdatedAtUtc = sinceUtc with a TrackedOperationId greater + /// than . This un-pins a batch that would otherwise + /// stall: when more than rows share one + /// UpdatedAtUtc, the plain inclusive >= resume re-reads the + /// same page forever. When is null/empty the + /// legacy inclusive >= contract is preserved exactly (an older + /// central that does not send the cursor is unaffected). Ordering is always + /// deterministic — UpdatedAtUtc ASC, TrackedOperationId ASC. + /// + /// /// is left as the empty string: /// the site id is not a tracking-store column, and the central client re-stamps /// it from the siteId it dialed (the only authority that knows which @@ -148,10 +161,17 @@ public interface IOperationTrackingStore /// /// Inclusive lower bound on UpdatedAtUtc; reads from the start. /// Maximum number of rows to return (oldest first). + /// + /// Optional composite-keyset cursor — the TrackedOperationId of the last + /// row already consumed at . When supplied the read + /// returns only rows strictly after (sinceUtc, afterId); when + /// null/empty the legacy inclusive >= sinceUtc behaviour applies. + /// /// Cancellation token. /// The matching rows projected to , oldest-first, capped at . Task> ReadChangedSinceAsync( DateTime sinceUtc, int batchSize, + string? afterId = null, CancellationToken ct = default); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index fa35f99c..0207e450 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs @@ -561,19 +561,24 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase ? DateTime.SpecifyKind(request.SinceUtc.ToDateTime(), DateTimeKind.Utc) : DateTime.MinValue; + // Composite-keyset cursor (Task 15): proto3 defaults an unset string to + // "", which the store treats as "no cursor" (legacy inclusive >= behaviour) + // — so an older central that never sets after_id is unaffected. + var afterId = string.IsNullOrEmpty(request.AfterId) ? null : request.AfterId; + IReadOnlyList operationals; try { operationals = await store.ReadChangedSinceAsync( - since, request.BatchSize, context.CancellationToken); + since, request.BatchSize, afterId, context.CancellationToken); } catch (Exception ex) { // Best-effort, like PullAuditEvents: a read fault must never abort // the reconciliation tick — central retries on its next cycle. _logger.LogError(ex, - "ReadChangedSinceAsync failed for since={Since} batch={Batch}; returning empty response.", - since, request.BatchSize); + "ReadChangedSinceAsync failed for since={Since} batch={Batch} afterId={AfterId}; returning empty response.", + since, request.BatchSize, afterId); return new PullSiteCallsResponse(); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto index c76173b7..a7c9c478 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto @@ -171,6 +171,14 @@ message PullAuditEventsResponse { message PullSiteCallsRequest { google.protobuf.Timestamp since_utc = 1; int32 batch_size = 2; + // Composite-keyset cursor (Task 15): the TrackedOperationId ("D" GUID form) of + // the last row already consumed at since_utc. When set, the site returns only + // rows strictly after (since_utc, after_id) under a deterministic + // (UpdatedAtUtc, TrackedOperationId) order — un-pinning a batch that would + // otherwise stall when more than batch_size rows share one since_utc instant. + // Empty (the proto3 string default) preserves the legacy inclusive >= behaviour, + // so an older central that never sets it is unaffected. Additive-only. + string after_id = 3; } message PullSiteCallsResponse { diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs index f81bf8e2..ca300fa5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs @@ -83,29 +83,30 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { "GAEgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBISCgpiYXRjaF9z", "aXplGAIgASgFIlwKF1B1bGxBdWRpdEV2ZW50c1Jlc3BvbnNlEikKBmV2ZW50", "cxgBIAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVudER0bxIWCg5tb3JlX2F2", - "YWlsYWJsZRgCIAEoCCJZChRQdWxsU2l0ZUNhbGxzUmVxdWVzdBItCglzaW5j", + "YWlsYWJsZRgCIAEoCCJrChRQdWxsU2l0ZUNhbGxzUmVxdWVzdBItCglzaW5j", "ZV91dGMYASABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhIKCmJh", - "dGNoX3NpemUYAiABKAUiaQoVUHVsbFNpdGVDYWxsc1Jlc3BvbnNlEjgKDG9w", - "ZXJhdGlvbmFscxgBIAMoCzIiLnNpdGVzdHJlYW0uU2l0ZUNhbGxPcGVyYXRp", - "b25hbER0bxIWCg5tb3JlX2F2YWlsYWJsZRgCIAEoCCpcCgdRdWFsaXR5EhcK", - "E1FVQUxJVFlfVU5TUEVDSUZJRUQQABIQCgxRVUFMSVRZX0dPT0QQARIVChFR", - "VUFMSVRZX1VOQ0VSVEFJThACEg8KC1FVQUxJVFlfQkFEEAMqXQoOQWxhcm1T", - "dGF0ZUVudW0SGwoXQUxBUk1fU1RBVEVfVU5TUEVDSUZJRUQQABIWChJBTEFS", - "TV9TVEFURV9OT1JNQUwQARIWChJBTEFSTV9TVEFURV9BQ1RJVkUQAiqFAQoO", - "QWxhcm1MZXZlbEVudW0SFAoQQUxBUk1fTEVWRUxfTk9ORRAAEhMKD0FMQVJN", - "X0xFVkVMX0xPVxABEhcKE0FMQVJNX0xFVkVMX0xPV19MT1cQAhIUChBBTEFS", - "TV9MRVZFTF9ISUdIEAMSGQoVQUxBUk1fTEVWRUxfSElHSF9ISUdIEAQytwMK", - "EVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNjcmliZUluc3RhbmNlEiEuc2l0", - "ZXN0cmVhbS5JbnN0YW5jZVN0cmVhbVJlcXVlc3QaGy5zaXRlc3RyZWFtLlNp", - "dGVTdHJlYW1FdmVudDABEkcKEUluZ2VzdEF1ZGl0RXZlbnRzEhsuc2l0ZXN0", - "cmVhbS5BdWRpdEV2ZW50QmF0Y2gaFS5zaXRlc3RyZWFtLkluZ2VzdEFjaxJQ", - "ChVJbmdlc3RDYWNoZWRUZWxlbWV0cnkSIC5zaXRlc3RyZWFtLkNhY2hlZFRl", - "bGVtZXRyeUJhdGNoGhUuc2l0ZXN0cmVhbS5Jbmdlc3RBY2sSWgoPUHVsbEF1", - "ZGl0RXZlbnRzEiIuc2l0ZXN0cmVhbS5QdWxsQXVkaXRFdmVudHNSZXF1ZXN0", - "GiMuc2l0ZXN0cmVhbS5QdWxsQXVkaXRFdmVudHNSZXNwb25zZRJUCg1QdWxs", - "U2l0ZUNhbGxzEiAuc2l0ZXN0cmVhbS5QdWxsU2l0ZUNhbGxzUmVxdWVzdBoh", - "LnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jlc3BvbnNlQiuqAihaQi5NT00u", - "V1cuU2NhZGFCcmlkZ2UuQ29tbXVuaWNhdGlvbi5HcnBjYgZwcm90bzM=")); + "dGNoX3NpemUYAiABKAUSEAoIYWZ0ZXJfaWQYAyABKAkiaQoVUHVsbFNpdGVD", + "YWxsc1Jlc3BvbnNlEjgKDG9wZXJhdGlvbmFscxgBIAMoCzIiLnNpdGVzdHJl", + "YW0uU2l0ZUNhbGxPcGVyYXRpb25hbER0bxIWCg5tb3JlX2F2YWlsYWJsZRgC", + "IAEoCCpcCgdRdWFsaXR5EhcKE1FVQUxJVFlfVU5TUEVDSUZJRUQQABIQCgxR", + "VUFMSVRZX0dPT0QQARIVChFRVUFMSVRZX1VOQ0VSVEFJThACEg8KC1FVQUxJ", + "VFlfQkFEEAMqXQoOQWxhcm1TdGF0ZUVudW0SGwoXQUxBUk1fU1RBVEVfVU5T", + "UEVDSUZJRUQQABIWChJBTEFSTV9TVEFURV9OT1JNQUwQARIWChJBTEFSTV9T", + "VEFURV9BQ1RJVkUQAiqFAQoOQWxhcm1MZXZlbEVudW0SFAoQQUxBUk1fTEVW", + "RUxfTk9ORRAAEhMKD0FMQVJNX0xFVkVMX0xPVxABEhcKE0FMQVJNX0xFVkVM", + "X0xPV19MT1cQAhIUChBBTEFSTV9MRVZFTF9ISUdIEAMSGQoVQUxBUk1fTEVW", + "RUxfSElHSF9ISUdIEAQytwMKEVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNj", + "cmliZUluc3RhbmNlEiEuc2l0ZXN0cmVhbS5JbnN0YW5jZVN0cmVhbVJlcXVl", + "c3QaGy5zaXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEkcKEUluZ2VzdEF1", + "ZGl0RXZlbnRzEhsuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50QmF0Y2gaFS5zaXRl", + "c3RyZWFtLkluZ2VzdEFjaxJQChVJbmdlc3RDYWNoZWRUZWxlbWV0cnkSIC5z", + "aXRlc3RyZWFtLkNhY2hlZFRlbGVtZXRyeUJhdGNoGhUuc2l0ZXN0cmVhbS5J", + "bmdlc3RBY2sSWgoPUHVsbEF1ZGl0RXZlbnRzEiIuc2l0ZXN0cmVhbS5QdWxs", + "QXVkaXRFdmVudHNSZXF1ZXN0GiMuc2l0ZXN0cmVhbS5QdWxsQXVkaXRFdmVu", + "dHNSZXNwb25zZRJUCg1QdWxsU2l0ZUNhbGxzEiAuc2l0ZXN0cmVhbS5QdWxs", + "U2l0ZUNhbGxzUmVxdWVzdBohLnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jl", + "c3BvbnNlQiuqAihaQi5NT00uV1cuU2NhZGFCcmlkZ2UuQ29tbXVuaWNhdGlv", + "bi5HcnBjYgZwcm90bzM=")); descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData, new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.WrappersReflection.Descriptor, }, new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.Quality), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateEnum), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmLevelEnum), }, null, new pbr::GeneratedClrTypeInfo[] { @@ -121,7 +122,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch.Parser, new[]{ "Packets" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest.Parser, new[]{ "SinceUtc", "BatchSize" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse.Parser, new[]{ "Events", "MoreAvailable" }, null, null, null, null), - new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest.Parser, new[]{ "SinceUtc", "BatchSize" }, null, null, null, null), + new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest.Parser, new[]{ "SinceUtc", "BatchSize", "AfterId" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse.Parser, new[]{ "Operationals", "MoreAvailable" }, null, null, null, null) })); } @@ -5200,6 +5201,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { public PullSiteCallsRequest(PullSiteCallsRequest other) : this() { sinceUtc_ = other.sinceUtc_ != null ? other.sinceUtc_.Clone() : null; batchSize_ = other.batchSize_; + afterId_ = other.afterId_; _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); } @@ -5233,6 +5235,27 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } } + /// Field number for the "after_id" field. + public const int AfterIdFieldNumber = 3; + private string afterId_ = ""; + /// + /// Composite-keyset cursor (Task 15): the TrackedOperationId ("D" GUID form) of + /// the last row already consumed at since_utc. When set, the site returns only + /// rows strictly after (since_utc, after_id) under a deterministic + /// (UpdatedAtUtc, TrackedOperationId) order — un-pinning a batch that would + /// otherwise stall when more than batch_size rows share one since_utc instant. + /// Empty (the proto3 string default) preserves the legacy inclusive >= behaviour, + /// so an older central that never sets it is unaffected. Additive-only. + /// + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public string AfterId { + get { return afterId_; } + set { + afterId_ = pb::ProtoPreconditions.CheckNotNull(value, "value"); + } + } + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public override bool Equals(object other) { @@ -5250,6 +5273,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } if (!object.Equals(SinceUtc, other.SinceUtc)) return false; if (BatchSize != other.BatchSize) return false; + if (AfterId != other.AfterId) return false; return Equals(_unknownFields, other._unknownFields); } @@ -5259,6 +5283,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { int hash = 1; if (sinceUtc_ != null) hash ^= SinceUtc.GetHashCode(); if (BatchSize != 0) hash ^= BatchSize.GetHashCode(); + if (AfterId.Length != 0) hash ^= AfterId.GetHashCode(); if (_unknownFields != null) { hash ^= _unknownFields.GetHashCode(); } @@ -5285,6 +5310,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { output.WriteRawTag(16); output.WriteInt32(BatchSize); } + if (AfterId.Length != 0) { + output.WriteRawTag(26); + output.WriteString(AfterId); + } if (_unknownFields != null) { _unknownFields.WriteTo(output); } @@ -5303,6 +5332,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { output.WriteRawTag(16); output.WriteInt32(BatchSize); } + if (AfterId.Length != 0) { + output.WriteRawTag(26); + output.WriteString(AfterId); + } if (_unknownFields != null) { _unknownFields.WriteTo(ref output); } @@ -5319,6 +5352,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (BatchSize != 0) { size += 1 + pb::CodedOutputStream.ComputeInt32Size(BatchSize); } + if (AfterId.Length != 0) { + size += 1 + pb::CodedOutputStream.ComputeStringSize(AfterId); + } if (_unknownFields != null) { size += _unknownFields.CalculateSize(); } @@ -5340,6 +5376,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (other.BatchSize != 0) { BatchSize = other.BatchSize; } + if (other.AfterId.Length != 0) { + AfterId = other.AfterId; + } _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); } @@ -5370,6 +5409,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { BatchSize = input.ReadInt32(); break; } + case 26: { + AfterId = input.ReadString(); + break; + } } } #endif @@ -5400,6 +5443,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { BatchSize = input.ReadInt32(); break; } + case 26: { + AfterId = input.ReadString(); + break; + } } } } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs index d405bfcc..4d4d407d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs @@ -365,6 +365,7 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, public async Task> ReadChangedSinceAsync( DateTime sinceUtc, int batchSize, + string? afterId = null, CancellationToken ct = default) { ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this); @@ -379,18 +380,28 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, await readConnection.OpenAsync(ct).ConfigureAwait(false); await using var cmd = readConnection.CreateCommand(); - // Inclusive lower bound on UpdatedAtUtc (>=) so a caller resuming from - // the last returned timestamp does not skip a row sharing that instant; - // central ingest is insert-if-not-exists + upsert-on-newer, so the - // boundary row re-read is a no-op. ORDER BY ... ASC + LIMIT yields the - // OLDEST matching rows so the cursor advances monotonically. - cmd.CommandText = """ + + // Composite (UpdatedAtUtc, TrackedOperationId) keyset (Task 15). Ordering + // is ALWAYS deterministic (UpdatedAtUtc ASC, TrackedOperationId ASC) so the + // cursor is well-defined even when many rows share one UpdatedAtUtc. + // • afterId present → strict keyset: skip everything up to and including + // (sinceUtc, afterId). Without this a page fully inside one UpdatedAtUtc + // instant re-reads forever (the plain >= resume never advances past the + // tie). TrackedOperationId is the "D" GUID form on both sides, so the + // TEXT comparison is a stable total order. + // • afterId null/empty → legacy inclusive >= sinceUtc (an older central + // that does not send a cursor is byte-for-byte unaffected). + var useKeyset = !string.IsNullOrEmpty(afterId); + var predicate = useKeyset + ? "UpdatedAtUtc > $since OR (UpdatedAtUtc = $since AND TrackedOperationId > $afterId)" + : "UpdatedAtUtc >= $since"; + cmd.CommandText = $""" SELECT TrackedOperationId, Kind, TargetSummary, Status, RetryCount, LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, SourceNode FROM OperationTracking - WHERE UpdatedAtUtc >= $since - ORDER BY UpdatedAtUtc ASC + WHERE {predicate} + ORDER BY UpdatedAtUtc ASC, TrackedOperationId ASC LIMIT $batchSize; """; // Force UTC kind before formatting so the cursor's "o" text matches the @@ -403,6 +414,10 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, .ToString("o", CultureInfo.InvariantCulture); cmd.Parameters.AddWithValue("$since", sinceText); cmd.Parameters.AddWithValue("$batchSize", batchSize); + if (useKeyset) + { + cmd.Parameters.AddWithValue("$afterId", afterId!); + } var rows = new List(); await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullSiteCallsTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullSiteCallsTests.cs index 45104a84..022bbd2a 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullSiteCallsTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullSiteCallsTests.cs @@ -72,7 +72,7 @@ public class SiteStreamPullSiteCallsTests : TestKit { var store = Substitute.For(); var rows = Enumerable.Range(0, 5).Select(_ => NewOperational()).ToList(); - store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any()) + store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .Returns((IReadOnlyList)rows); var server = CreateServer(); @@ -97,7 +97,7 @@ public class SiteStreamPullSiteCallsTests : TestKit { var store = Substitute.For(); var capturedSince = DateTime.MinValue; - store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any()) + store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .Returns(call => { capturedSince = call.ArgAt(0); @@ -121,6 +121,65 @@ public class SiteStreamPullSiteCallsTests : TestKit Assert.Equal(since, capturedSince); } + [Fact] + public async Task PullSiteCalls_AfterIdSet_PassesCursorThroughToStore() + { + // Task 15: a non-empty after_id is forwarded verbatim to the store as the + // composite-keyset cursor. + var store = Substitute.For(); + string? capturedAfterId = "SENTINEL"; + store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => + { + capturedAfterId = call.ArgAt(2); + return (IReadOnlyList)Array.Empty(); + }); + + var server = CreateServer(); + server.SetOperationTrackingStore(store); + + var cursor = Guid.NewGuid().ToString("D"); + var request = new PullSiteCallsRequest + { + SinceUtc = Timestamp.FromDateTime(DateTime.UtcNow.AddHours(-1)), + BatchSize = 50, + AfterId = cursor, + }; + + await server.PullSiteCalls(request, NewContext()); + + Assert.Equal(cursor, capturedAfterId); + } + + [Fact] + public async Task PullSiteCalls_AfterIdUnset_PassesNullToStore() + { + // proto3 defaults an unset string to ""; the handler must normalise that to + // null so the store keeps its legacy inclusive >= behaviour (older central). + var store = Substitute.For(); + string? capturedAfterId = "SENTINEL"; + store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => + { + capturedAfterId = call.ArgAt(2); + return (IReadOnlyList)Array.Empty(); + }); + + var server = CreateServer(); + server.SetOperationTrackingStore(store); + + var request = new PullSiteCallsRequest + { + SinceUtc = Timestamp.FromDateTime(DateTime.UtcNow.AddHours(-1)), + BatchSize = 50, + // AfterId left unset → "" on the wire. + }; + + await server.PullSiteCalls(request, NewContext()); + + Assert.Null(capturedAfterId); + } + [Fact] public async Task PullSiteCalls_SinceUtcUnset_PassesDateTimeMinValue() { @@ -130,7 +189,7 @@ public class SiteStreamPullSiteCallsTests : TestKit // without a null-deref — this proves the very first cycle doesn't crash. var store = Substitute.For(); var captured = new DateTime(2099, 1, 1, 0, 0, 0, DateTimeKind.Utc); // sentinel - store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any()) + store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .Returns(call => { captured = call.ArgAt(0); @@ -158,7 +217,7 @@ public class SiteStreamPullSiteCallsTests : TestKit { var store = Substitute.For(); var rows = Enumerable.Range(0, 3).Select(_ => NewOperational()).ToList(); - store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any()) + store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .Returns((IReadOnlyList)rows); var server = CreateServer(); @@ -200,7 +259,7 @@ public class SiteStreamPullSiteCallsTests : TestKit { // Best-effort: a read fault must never abort the reconciliation tick. var store = Substitute.For(); - store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any()) + store.ReadChangedSinceAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .ThrowsAsync(new InvalidOperationException("SQLite disposed mid-call")); var server = CreateServer(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs index 9f6858d5..8ae7ebb4 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs @@ -466,7 +466,7 @@ public class OperationTrackingStoreTests // Cursor at the middle row's UpdatedAtUtc: inclusive lower bound, so // middle + newer come back, older is excluded. - var result = await store.ReadChangedSinceAsync(t0.AddMinutes(10), batchSize: 100, CancellationToken.None); + var result = await store.ReadChangedSinceAsync(t0.AddMinutes(10), batchSize: 100, ct: CancellationToken.None); Assert.Equal(2, result.Count); Assert.Equal(middle, result[0].TrackedOperationId); @@ -483,7 +483,7 @@ public class OperationTrackingStoreTests await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "A", null, null, null); await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "B", null, null, null); - var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, CancellationToken.None); + var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, ct: CancellationToken.None); Assert.Equal(2, result.Count); } @@ -504,7 +504,7 @@ public class OperationTrackingStoreTests SetUpdatedAt(dataSource, id, t0.AddMinutes(i)); } - var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 3, CancellationToken.None); + var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 3, ct: CancellationToken.None); // Capped to 3 — and the cap takes the OLDEST 3 (asc order) so the // caller can advance the cursor monotonically across follow-up pulls. @@ -527,7 +527,7 @@ public class OperationTrackingStoreTests await store.RecordAttemptAsync(apiId, nameof(AuditStatus.Attempted), 2, "HTTP 503", 503); await store.RecordTerminalAsync(dbId, nameof(AuditStatus.Parked), "max retries", null); - var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, CancellationToken.None); + var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, ct: CancellationToken.None); var api = result.Single(r => r.TrackedOperationId == apiId); var db = result.Single(r => r.TrackedOperationId == dbId); @@ -720,4 +720,92 @@ public class OperationTrackingStoreTests Assert.Equal(2, rows.Count); } + + // ── Task 15: composite (UpdatedAtUtc, TrackedOperationId) keyset ── + + /// + /// Seeds batchSize + 2 rows that all share ONE UpdatedAtUtc and + /// pages through them with the composite keyset. Page 1 (no cursor) returns the + /// first batchSize; page 2 resumes from (sinceUtc, lastId) and + /// returns the REMAINING rows with no overlap — the stall the plain inclusive + /// >= resume can never escape. + /// + [Fact] + public async Task ReadChangedSince_CompositeKeyset_AdvancesPastSharedTimestamp() + { + const int batchSize = 3; + var (store, dataSource) = CreateStore(nameof(ReadChangedSince_CompositeKeyset_AdvancesPastSharedTimestamp)); + await using var _store = store; + + for (var i = 0; i < batchSize + 2; i++) + { + await store.RecordEnqueueAsync(TrackedOperationId.New(), + kind: nameof(AuditKind.ApiCallCached), targetSummary: $"ERP.{i}", + sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a"); + } + + var boundary = ForceSharedUpdatedAt(dataSource); + + // Page 1: no cursor → the first batchSize rows, ordered by TrackedOperationId. + var page1 = await store.ReadChangedSinceAsync(boundary, batchSize); + Assert.Equal(batchSize, page1.Count); + + // Page 2: resume strictly after the last id of page 1. + var afterId = page1[^1].TrackedOperationId.ToString(); + var page2 = await store.ReadChangedSinceAsync(boundary, batchSize, afterId); + + // The remaining two rows, none seen on page 1, all strictly greater than the cursor. + Assert.Equal(2, page2.Count); + var page1Ids = page1.Select(r => r.TrackedOperationId).ToHashSet(); + Assert.All(page2, r => Assert.DoesNotContain(r.TrackedOperationId, page1Ids)); + Assert.All(page2, r => Assert.True( + string.CompareOrdinal(r.TrackedOperationId.ToString(), afterId) > 0)); + + // Union covers every seeded row exactly once — no gap, no dup. + Assert.Equal(batchSize + 2, page1Ids.Concat(page2.Select(r => r.TrackedOperationId)).Distinct().Count()); + } + + /// + /// Regression guard for the legacy contract: with NO cursor a resume at the + /// shared timestamp re-reads the same page (the exact stall the keyset fixes) — + /// an older central that never sends after_id is byte-for-byte unaffected. + /// + [Fact] + public async Task ReadChangedSince_WithoutKeyset_ReReadsSamePage_LegacyPreserved() + { + const int batchSize = 3; + var (store, dataSource) = CreateStore(nameof(ReadChangedSince_WithoutKeyset_ReReadsSamePage_LegacyPreserved)); + await using var _store = store; + + for (var i = 0; i < batchSize + 2; i++) + { + await store.RecordEnqueueAsync(TrackedOperationId.New(), + kind: nameof(AuditKind.ApiCallCached), targetSummary: $"ERP.{i}", + sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a"); + } + + var boundary = ForceSharedUpdatedAt(dataSource); + + var page1 = await store.ReadChangedSinceAsync(boundary, batchSize); + // Legacy resume: same sinceUtc, no cursor → identical page (the stall). + var pageAgain = await store.ReadChangedSinceAsync(boundary, batchSize); + + Assert.Equal( + page1.Select(r => r.TrackedOperationId).ToList(), + pageAgain.Select(r => r.TrackedOperationId).ToList()); + } + + /// Forces every OperationTracking row to one exact UpdatedAtUtc instant; returns it. + private DateTime ForceSharedUpdatedAt(string dataSource) + { + var boundary = DateTime.UtcNow; + var boundaryText = DateTime.SpecifyKind(boundary, DateTimeKind.Utc) + .ToString("o", System.Globalization.CultureInfo.InvariantCulture); + using var conn = OpenVerifierConnection(dataSource); + using var cmd = conn.CreateCommand(); + cmd.CommandText = "UPDATE OperationTracking SET UpdatedAtUtc = $ts"; + cmd.Parameters.AddWithValue("$ts", boundaryText); + cmd.ExecuteNonQuery(); + return boundary; + } }