Compare commits
4 Commits
uns-tab-dr
...
alarm-invo
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
404bfbe7e4 | ||
| 006af636a0 | |||
|
|
c0751fdda5 | ||
| 80e080ecec |
@@ -2,6 +2,7 @@ using Microsoft.EntityFrameworkCore;
|
|||||||
using ZB.MOM.WW.OtOpcUa.Admin.Services;
|
using ZB.MOM.WW.OtOpcUa.Admin.Services;
|
||||||
using ZB.MOM.WW.OtOpcUa.Configuration;
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
||||||
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||||
|
|
||||||
namespace ZB.MOM.WW.OtOpcUa.Admin.Services;
|
namespace ZB.MOM.WW.OtOpcUa.Admin.Services;
|
||||||
|
|
||||||
@@ -152,14 +153,37 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
|
|||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
foreach (var row in batch.Rows.Where(r => r.IsAccepted))
|
// Snapshot active reservations that overlap this batch's ZTag + SAPID set — one
|
||||||
|
// round-trip instead of N. Released rows (ReleasedAt IS NOT NULL) are ignored so
|
||||||
|
// an explicitly-released value can be reused.
|
||||||
|
var accepted = batch.Rows.Where(r => r.IsAccepted).ToList();
|
||||||
|
var zTags = accepted.Where(r => !string.IsNullOrWhiteSpace(r.ZTag))
|
||||||
|
.Select(r => r.ZTag).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
|
||||||
|
var sapIds = accepted.Where(r => !string.IsNullOrWhiteSpace(r.SAPID))
|
||||||
|
.Select(r => r.SAPID).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
|
||||||
|
|
||||||
|
var existingReservations = await db.ExternalIdReservations
|
||||||
|
.Where(r => r.ReleasedAt == null &&
|
||||||
|
((r.Kind == ReservationKind.ZTag && zTags.Contains(r.Value)) ||
|
||||||
|
(r.Kind == ReservationKind.SAPID && sapIds.Contains(r.Value))))
|
||||||
|
.ToListAsync(ct).ConfigureAwait(false);
|
||||||
|
var resByKey = existingReservations.ToDictionary(
|
||||||
|
r => (r.Kind, r.Value.ToLowerInvariant()),
|
||||||
|
r => r);
|
||||||
|
|
||||||
|
var nowUtc = DateTime.UtcNow;
|
||||||
|
var firstPublishedBy = batch.CreatedBy;
|
||||||
|
|
||||||
|
foreach (var row in accepted)
|
||||||
{
|
{
|
||||||
|
var equipmentUuid = Guid.TryParse(row.EquipmentUuid, out var u) ? u : Guid.NewGuid();
|
||||||
|
|
||||||
db.Equipment.Add(new Equipment
|
db.Equipment.Add(new Equipment
|
||||||
{
|
{
|
||||||
EquipmentRowId = Guid.NewGuid(),
|
EquipmentRowId = Guid.NewGuid(),
|
||||||
GenerationId = generationId,
|
GenerationId = generationId,
|
||||||
EquipmentId = row.EquipmentId,
|
EquipmentId = row.EquipmentId,
|
||||||
EquipmentUuid = Guid.TryParse(row.EquipmentUuid, out var u) ? u : Guid.NewGuid(),
|
EquipmentUuid = equipmentUuid,
|
||||||
DriverInstanceId = driverInstanceIdForRows,
|
DriverInstanceId = driverInstanceIdForRows,
|
||||||
UnsLineId = unsLineIdForRows,
|
UnsLineId = unsLineIdForRows,
|
||||||
Name = row.Name,
|
Name = row.Name,
|
||||||
@@ -176,10 +200,25 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
|
|||||||
ManufacturerUri = row.ManufacturerUri,
|
ManufacturerUri = row.ManufacturerUri,
|
||||||
DeviceManualUri = row.DeviceManualUri,
|
DeviceManualUri = row.DeviceManualUri,
|
||||||
});
|
});
|
||||||
|
|
||||||
|
MergeReservation(row.ZTag, ReservationKind.ZTag, equipmentUuid, batch.ClusterId,
|
||||||
|
firstPublishedBy, nowUtc, resByKey);
|
||||||
|
MergeReservation(row.SAPID, ReservationKind.SAPID, equipmentUuid, batch.ClusterId,
|
||||||
|
firstPublishedBy, nowUtc, resByKey);
|
||||||
}
|
}
|
||||||
|
|
||||||
batch.FinalisedAtUtc = DateTime.UtcNow;
|
batch.FinalisedAtUtc = nowUtc;
|
||||||
await db.SaveChangesAsync(ct).ConfigureAwait(false);
|
try
|
||||||
|
{
|
||||||
|
await db.SaveChangesAsync(ct).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (DbUpdateException ex) when (IsReservationUniquenessViolation(ex))
|
||||||
|
{
|
||||||
|
throw new ExternalIdReservationConflictException(
|
||||||
|
"Finalise rejected: one or more ZTag/SAPID values were reserved by another operator " +
|
||||||
|
"between batch preview and commit. Inspect active reservations + retry after resolving the conflict.",
|
||||||
|
ex);
|
||||||
|
}
|
||||||
if (tx is not null) await tx.CommitAsync(ct).ConfigureAwait(false);
|
if (tx is not null) await tx.CommitAsync(ct).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch
|
catch
|
||||||
@@ -193,6 +232,71 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Merge one external-ID reservation for an equipment row. Three outcomes:
|
||||||
|
/// (1) value is empty → skip; (2) reservation exists for same <paramref name="equipmentUuid"/>
|
||||||
|
/// → bump <c>LastPublishedAt</c>; (3) reservation exists for a different EquipmentUuid
|
||||||
|
/// → throw <see cref="ExternalIdReservationConflictException"/> with the conflicting UUID
|
||||||
|
/// so the caller sees which equipment already owns the value; (4) no reservation → create new.
|
||||||
|
/// </summary>
|
||||||
|
private void MergeReservation(
|
||||||
|
string? value,
|
||||||
|
ReservationKind kind,
|
||||||
|
Guid equipmentUuid,
|
||||||
|
string clusterId,
|
||||||
|
string firstPublishedBy,
|
||||||
|
DateTime nowUtc,
|
||||||
|
Dictionary<(ReservationKind, string), ExternalIdReservation> cache)
|
||||||
|
{
|
||||||
|
if (string.IsNullOrWhiteSpace(value)) return;
|
||||||
|
|
||||||
|
var key = (kind, value.ToLowerInvariant());
|
||||||
|
if (cache.TryGetValue(key, out var existing))
|
||||||
|
{
|
||||||
|
if (existing.EquipmentUuid != equipmentUuid)
|
||||||
|
throw new ExternalIdReservationConflictException(
|
||||||
|
$"{kind} '{value}' is already reserved by EquipmentUuid {existing.EquipmentUuid} " +
|
||||||
|
$"(first published {existing.FirstPublishedAt:u} on cluster '{existing.ClusterId}'). " +
|
||||||
|
$"Refusing to re-assign to {equipmentUuid}.");
|
||||||
|
|
||||||
|
existing.LastPublishedAt = nowUtc;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
var fresh = new ExternalIdReservation
|
||||||
|
{
|
||||||
|
ReservationId = Guid.NewGuid(),
|
||||||
|
Kind = kind,
|
||||||
|
Value = value,
|
||||||
|
EquipmentUuid = equipmentUuid,
|
||||||
|
ClusterId = clusterId,
|
||||||
|
FirstPublishedAt = nowUtc,
|
||||||
|
FirstPublishedBy = firstPublishedBy,
|
||||||
|
LastPublishedAt = nowUtc,
|
||||||
|
};
|
||||||
|
db.ExternalIdReservations.Add(fresh);
|
||||||
|
cache[key] = fresh;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// True when the <see cref="DbUpdateException"/> root-cause was the filtered-unique
|
||||||
|
/// index <c>UX_ExternalIdReservation_KindValue_Active</c> — i.e. another transaction
|
||||||
|
/// won the race between our cache-load + commit. SQL Server surfaces this as 2601 / 2627.
|
||||||
|
/// </summary>
|
||||||
|
private static bool IsReservationUniquenessViolation(DbUpdateException ex)
|
||||||
|
{
|
||||||
|
for (Exception? inner = ex; inner is not null; inner = inner.InnerException)
|
||||||
|
{
|
||||||
|
if (inner is Microsoft.Data.SqlClient.SqlException sql &&
|
||||||
|
(sql.Number == 2601 || sql.Number == 2627) &&
|
||||||
|
sql.Message.Contains("UX_ExternalIdReservation_KindValue_Active", StringComparison.OrdinalIgnoreCase))
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>List batches created by the given user. Finalised batches are archived; include them on demand.</summary>
|
/// <summary>List batches created by the given user. Finalised batches are archived; include them on demand.</summary>
|
||||||
public async Task<IReadOnlyList<EquipmentImportBatch>> ListByUserAsync(string createdBy, bool includeFinalised, CancellationToken ct)
|
public async Task<IReadOnlyList<EquipmentImportBatch>> ListByUserAsync(string createdBy, bool includeFinalised, CancellationToken ct)
|
||||||
{
|
{
|
||||||
@@ -205,3 +309,16 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
|
|||||||
|
|
||||||
public sealed class ImportBatchNotFoundException(string message) : Exception(message);
|
public sealed class ImportBatchNotFoundException(string message) : Exception(message);
|
||||||
public sealed class ImportBatchAlreadyFinalisedException(string message) : Exception(message);
|
public sealed class ImportBatchAlreadyFinalisedException(string message) : Exception(message);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Thrown when a <c>FinaliseBatchAsync</c> call detects that one of its ZTag/SAPID values is
|
||||||
|
/// already reserved by a different EquipmentUuid — either from a prior published generation
|
||||||
|
/// or a concurrent finalise that won the race. The operator sees the message + the conflicting
|
||||||
|
/// equipment ownership so they can resolve the conflict (pick a new ZTag, release the existing
|
||||||
|
/// reservation via <c>sp_ReleaseExternalIdReservation</c>, etc.) and retry the finalise.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class ExternalIdReservationConflictException : Exception
|
||||||
|
{
|
||||||
|
public ExternalIdReservationConflictException(string message) : base(message) { }
|
||||||
|
public ExternalIdReservationConflictException(string message, Exception inner) : base(message, inner) { }
|
||||||
|
}
|
||||||
|
|||||||
129
src/ZB.MOM.WW.OtOpcUa.Core/Resilience/AlarmSurfaceInvoker.cs
Normal file
129
src/ZB.MOM.WW.OtOpcUa.Core/Resilience/AlarmSurfaceInvoker.cs
Normal file
@@ -0,0 +1,129 @@
|
|||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Core.Resilience;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Wraps the three mutating surfaces of <see cref="IAlarmSource"/>
|
||||||
|
/// (<see cref="IAlarmSource.SubscribeAlarmsAsync"/>, <see cref="IAlarmSource.UnsubscribeAlarmsAsync"/>,
|
||||||
|
/// <see cref="IAlarmSource.AcknowledgeAsync"/>) through <see cref="CapabilityInvoker"/> so the
|
||||||
|
/// Phase 6.1 resilience pipeline runs — retry semantics match
|
||||||
|
/// <see cref="DriverCapability.AlarmSubscribe"/> (retries by default) and
|
||||||
|
/// <see cref="DriverCapability.AlarmAcknowledge"/> (does NOT retry per decision #143).
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>Multi-host dispatch: when the driver implements <see cref="IPerCallHostResolver"/>,
|
||||||
|
/// each source-node-id is resolved individually + grouped by host so a dead PLC inside a
|
||||||
|
/// multi-device driver doesn't poison the sibling hosts' breakers. Drivers with a single
|
||||||
|
/// host fall back to <see cref="IDriver.DriverInstanceId"/> as the single-host key.</para>
|
||||||
|
///
|
||||||
|
/// <para>Why this lives here + not on <see cref="CapabilityInvoker"/>: alarm surfaces have a
|
||||||
|
/// handle-returning shape (SubscribeAlarmsAsync returns <see cref="IAlarmSubscriptionHandle"/>)
|
||||||
|
/// + a per-call fan-out (AcknowledgeAsync gets a batch of
|
||||||
|
/// <see cref="AlarmAcknowledgeRequest"/>s that may span multiple hosts). Keeping the fan-out
|
||||||
|
/// logic here keeps the invoker's execute-overloads narrow.</para>
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class AlarmSurfaceInvoker
|
||||||
|
{
|
||||||
|
private readonly CapabilityInvoker _invoker;
|
||||||
|
private readonly IAlarmSource _alarmSource;
|
||||||
|
private readonly IPerCallHostResolver? _hostResolver;
|
||||||
|
private readonly string _defaultHost;
|
||||||
|
|
||||||
|
public AlarmSurfaceInvoker(
|
||||||
|
CapabilityInvoker invoker,
|
||||||
|
IAlarmSource alarmSource,
|
||||||
|
string defaultHost,
|
||||||
|
IPerCallHostResolver? hostResolver = null)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(invoker);
|
||||||
|
ArgumentNullException.ThrowIfNull(alarmSource);
|
||||||
|
ArgumentException.ThrowIfNullOrWhiteSpace(defaultHost);
|
||||||
|
|
||||||
|
_invoker = invoker;
|
||||||
|
_alarmSource = alarmSource;
|
||||||
|
_defaultHost = defaultHost;
|
||||||
|
_hostResolver = hostResolver;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Subscribe to alarm events for a set of source node ids, fanning out by resolved host
|
||||||
|
/// so per-host breakers / bulkheads apply. Returns one handle per host — callers that
|
||||||
|
/// don't care about per-host separation may concatenate them.
|
||||||
|
/// </summary>
|
||||||
|
public async Task<IReadOnlyList<IAlarmSubscriptionHandle>> SubscribeAsync(
|
||||||
|
IReadOnlyList<string> sourceNodeIds,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(sourceNodeIds);
|
||||||
|
if (sourceNodeIds.Count == 0) return [];
|
||||||
|
|
||||||
|
var byHost = GroupByHost(sourceNodeIds);
|
||||||
|
var handles = new List<IAlarmSubscriptionHandle>(byHost.Count);
|
||||||
|
foreach (var (host, ids) in byHost)
|
||||||
|
{
|
||||||
|
var handle = await _invoker.ExecuteAsync(
|
||||||
|
DriverCapability.AlarmSubscribe,
|
||||||
|
host,
|
||||||
|
async ct => await _alarmSource.SubscribeAlarmsAsync(ids, ct).ConfigureAwait(false),
|
||||||
|
cancellationToken).ConfigureAwait(false);
|
||||||
|
handles.Add(handle);
|
||||||
|
}
|
||||||
|
return handles;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Cancel an alarm subscription. Routes through the AlarmSubscribe pipeline for parity.</summary>
|
||||||
|
public ValueTask UnsubscribeAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(handle);
|
||||||
|
return _invoker.ExecuteAsync(
|
||||||
|
DriverCapability.AlarmSubscribe,
|
||||||
|
_defaultHost,
|
||||||
|
async ct => await _alarmSource.UnsubscribeAlarmsAsync(handle, ct).ConfigureAwait(false),
|
||||||
|
cancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Acknowledge alarms. Fans out by resolved host; each host's batch runs through the
|
||||||
|
/// AlarmAcknowledge pipeline (no-retry per decision #143 — an alarm-ack is not idempotent
|
||||||
|
/// at the plant-floor acknowledgement level even if the OPC UA spec permits re-issue).
|
||||||
|
/// </summary>
|
||||||
|
public async Task AcknowledgeAsync(
|
||||||
|
IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(acknowledgements);
|
||||||
|
if (acknowledgements.Count == 0) return;
|
||||||
|
|
||||||
|
var byHost = _hostResolver is null
|
||||||
|
? new Dictionary<string, List<AlarmAcknowledgeRequest>> { [_defaultHost] = acknowledgements.ToList() }
|
||||||
|
: acknowledgements
|
||||||
|
.GroupBy(a => _hostResolver.ResolveHost(a.SourceNodeId))
|
||||||
|
.ToDictionary(g => g.Key, g => g.ToList());
|
||||||
|
|
||||||
|
foreach (var (host, batch) in byHost)
|
||||||
|
{
|
||||||
|
var batchSnapshot = batch; // capture for the lambda
|
||||||
|
await _invoker.ExecuteAsync(
|
||||||
|
DriverCapability.AlarmAcknowledge,
|
||||||
|
host,
|
||||||
|
async ct => await _alarmSource.AcknowledgeAsync(batchSnapshot, ct).ConfigureAwait(false),
|
||||||
|
cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private Dictionary<string, List<string>> GroupByHost(IReadOnlyList<string> sourceNodeIds)
|
||||||
|
{
|
||||||
|
if (_hostResolver is null)
|
||||||
|
return new Dictionary<string, List<string>> { [_defaultHost] = sourceNodeIds.ToList() };
|
||||||
|
|
||||||
|
var result = new Dictionary<string, List<string>>(StringComparer.Ordinal);
|
||||||
|
foreach (var id in sourceNodeIds)
|
||||||
|
{
|
||||||
|
var host = _hostResolver.ResolveHost(id);
|
||||||
|
if (!result.TryGetValue(host, out var list))
|
||||||
|
result[host] = list = new List<string>();
|
||||||
|
list.Add(id);
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -23,11 +23,13 @@ public sealed class EquipmentImportBatchServiceTests : IDisposable
|
|||||||
|
|
||||||
public void Dispose() => _db.Dispose();
|
public void Dispose() => _db.Dispose();
|
||||||
|
|
||||||
|
// Unique SAPID per row — FinaliseBatch reserves ZTag + SAPID via filtered-unique index, so
|
||||||
|
// two rows sharing a SAPID under different EquipmentUuids collide as intended.
|
||||||
private static EquipmentCsvRow Row(string zTag, string name = "eq-1") => new()
|
private static EquipmentCsvRow Row(string zTag, string name = "eq-1") => new()
|
||||||
{
|
{
|
||||||
ZTag = zTag,
|
ZTag = zTag,
|
||||||
MachineCode = "mc",
|
MachineCode = "mc",
|
||||||
SAPID = "sap",
|
SAPID = $"sap-{zTag}",
|
||||||
EquipmentId = "eq-id",
|
EquipmentId = "eq-id",
|
||||||
EquipmentUuid = Guid.NewGuid().ToString(),
|
EquipmentUuid = Guid.NewGuid().ToString(),
|
||||||
Name = name,
|
Name = name,
|
||||||
@@ -162,4 +164,93 @@ public sealed class EquipmentImportBatchServiceTests : IDisposable
|
|||||||
await _svc.DropBatchAsync(Guid.NewGuid(), CancellationToken.None);
|
await _svc.DropBatchAsync(Guid.NewGuid(), CancellationToken.None);
|
||||||
// no throw
|
// no throw
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task FinaliseBatch_Creates_ExternalIdReservations_ForZTagAndSAPID()
|
||||||
|
{
|
||||||
|
var batch = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
|
||||||
|
await _svc.StageRowsAsync(batch.Id, [Row("z-new-1")], [], CancellationToken.None);
|
||||||
|
|
||||||
|
await _svc.FinaliseBatchAsync(batch.Id, 1, "drv", "line", CancellationToken.None);
|
||||||
|
|
||||||
|
var active = await _db.ExternalIdReservations.AsNoTracking()
|
||||||
|
.Where(r => r.ReleasedAt == null)
|
||||||
|
.ToListAsync();
|
||||||
|
active.Count.ShouldBe(2);
|
||||||
|
active.ShouldContain(r => r.Kind == ZB.MOM.WW.OtOpcUa.Configuration.Enums.ReservationKind.ZTag && r.Value == "z-new-1");
|
||||||
|
active.ShouldContain(r => r.Kind == ZB.MOM.WW.OtOpcUa.Configuration.Enums.ReservationKind.SAPID && r.Value == "sap-z-new-1");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task FinaliseBatch_SameEquipmentUuid_ReusesExistingReservation()
|
||||||
|
{
|
||||||
|
var batch1 = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
|
||||||
|
var sharedUuid = Guid.NewGuid();
|
||||||
|
var row = new EquipmentCsvRow
|
||||||
|
{
|
||||||
|
ZTag = "z-shared", MachineCode = "mc", SAPID = "sap-shared",
|
||||||
|
EquipmentId = "eq-1", EquipmentUuid = sharedUuid.ToString(),
|
||||||
|
Name = "eq-1", UnsAreaName = "a", UnsLineName = "l",
|
||||||
|
};
|
||||||
|
await _svc.StageRowsAsync(batch1.Id, [row], [], CancellationToken.None);
|
||||||
|
await _svc.FinaliseBatchAsync(batch1.Id, 1, "drv", "line", CancellationToken.None);
|
||||||
|
|
||||||
|
var countAfterFirst = _db.ExternalIdReservations.Count(r => r.ReleasedAt == null);
|
||||||
|
|
||||||
|
// Second finalise with same EquipmentUuid + same ZTag — should NOT create a duplicate.
|
||||||
|
var batch2 = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
|
||||||
|
await _svc.StageRowsAsync(batch2.Id, [row], [], CancellationToken.None);
|
||||||
|
await _svc.FinaliseBatchAsync(batch2.Id, 2, "drv", "line", CancellationToken.None);
|
||||||
|
|
||||||
|
_db.ExternalIdReservations.Count(r => r.ReleasedAt == null).ShouldBe(countAfterFirst);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task FinaliseBatch_DifferentEquipmentUuid_SameZTag_Throws_Conflict()
|
||||||
|
{
|
||||||
|
var batchA = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
|
||||||
|
var rowA = new EquipmentCsvRow
|
||||||
|
{
|
||||||
|
ZTag = "z-collide", MachineCode = "mc-a", SAPID = "sap-a",
|
||||||
|
EquipmentId = "eq-a", EquipmentUuid = Guid.NewGuid().ToString(),
|
||||||
|
Name = "a", UnsAreaName = "ar", UnsLineName = "ln",
|
||||||
|
};
|
||||||
|
await _svc.StageRowsAsync(batchA.Id, [rowA], [], CancellationToken.None);
|
||||||
|
await _svc.FinaliseBatchAsync(batchA.Id, 1, "drv", "line", CancellationToken.None);
|
||||||
|
|
||||||
|
var batchB = await _svc.CreateBatchAsync("c1", "bob", CancellationToken.None);
|
||||||
|
var rowB = new EquipmentCsvRow
|
||||||
|
{
|
||||||
|
ZTag = "z-collide", MachineCode = "mc-b", SAPID = "sap-b", // same ZTag, different EquipmentUuid
|
||||||
|
EquipmentId = "eq-b", EquipmentUuid = Guid.NewGuid().ToString(),
|
||||||
|
Name = "b", UnsAreaName = "ar", UnsLineName = "ln",
|
||||||
|
};
|
||||||
|
await _svc.StageRowsAsync(batchB.Id, [rowB], [], CancellationToken.None);
|
||||||
|
|
||||||
|
var ex = await Should.ThrowAsync<ExternalIdReservationConflictException>(() =>
|
||||||
|
_svc.FinaliseBatchAsync(batchB.Id, 2, "drv", "line", CancellationToken.None));
|
||||||
|
ex.Message.ShouldContain("z-collide");
|
||||||
|
|
||||||
|
// Second finalise must have rolled back — no partial Equipment row for batch B.
|
||||||
|
var equipmentB = await _db.Equipment.AsNoTracking()
|
||||||
|
.Where(e => e.EquipmentId == "eq-b")
|
||||||
|
.ToListAsync();
|
||||||
|
equipmentB.ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task FinaliseBatch_EmptyZTagAndSAPID_SkipsReservation()
|
||||||
|
{
|
||||||
|
var batch = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
|
||||||
|
var row = new EquipmentCsvRow
|
||||||
|
{
|
||||||
|
ZTag = "", MachineCode = "mc", SAPID = "",
|
||||||
|
EquipmentId = "eq-nil", EquipmentUuid = Guid.NewGuid().ToString(),
|
||||||
|
Name = "nil", UnsAreaName = "ar", UnsLineName = "ln",
|
||||||
|
};
|
||||||
|
await _svc.StageRowsAsync(batch.Id, [row], [], CancellationToken.None);
|
||||||
|
await _svc.FinaliseBatchAsync(batch.Id, 1, "drv", "line", CancellationToken.None);
|
||||||
|
|
||||||
|
_db.ExternalIdReservations.Count().ShouldBe(0);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,127 @@
|
|||||||
|
using Shouldly;
|
||||||
|
using Xunit;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Core.Resilience;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.OtOpcUa.Core.Tests.Resilience;
|
||||||
|
|
||||||
|
[Trait("Category", "Unit")]
|
||||||
|
public sealed class AlarmSurfaceInvokerTests
|
||||||
|
{
|
||||||
|
private static readonly DriverResilienceOptions TierAOptions = new() { Tier = DriverTier.A };
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task SubscribeAsync_EmptyList_ReturnsEmpty_WithoutDriverCall()
|
||||||
|
{
|
||||||
|
var driver = new FakeAlarmSource();
|
||||||
|
var surface = NewSurface(driver, defaultHost: "h");
|
||||||
|
|
||||||
|
var handles = await surface.SubscribeAsync([], CancellationToken.None);
|
||||||
|
|
||||||
|
handles.Count.ShouldBe(0);
|
||||||
|
driver.SubscribeCallCount.ShouldBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task SubscribeAsync_SingleHost_RoutesThroughDefaultHost()
|
||||||
|
{
|
||||||
|
var driver = new FakeAlarmSource();
|
||||||
|
var surface = NewSurface(driver, defaultHost: "h1");
|
||||||
|
|
||||||
|
var handles = await surface.SubscribeAsync(["src-1", "src-2"], CancellationToken.None);
|
||||||
|
|
||||||
|
handles.Count.ShouldBe(1);
|
||||||
|
driver.SubscribeCallCount.ShouldBe(1);
|
||||||
|
driver.LastSubscribedIds.ShouldBe(["src-1", "src-2"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task SubscribeAsync_MultiHost_FansOutByResolvedHost()
|
||||||
|
{
|
||||||
|
var driver = new FakeAlarmSource();
|
||||||
|
var resolver = new StubResolver(new Dictionary<string, string>
|
||||||
|
{
|
||||||
|
["src-1"] = "plc-a",
|
||||||
|
["src-2"] = "plc-b",
|
||||||
|
["src-3"] = "plc-a",
|
||||||
|
});
|
||||||
|
var surface = NewSurface(driver, defaultHost: "default-ignored", resolver: resolver);
|
||||||
|
|
||||||
|
var handles = await surface.SubscribeAsync(["src-1", "src-2", "src-3"], CancellationToken.None);
|
||||||
|
|
||||||
|
handles.Count.ShouldBe(2); // one per distinct host
|
||||||
|
driver.SubscribeCallCount.ShouldBe(2); // one driver call per host
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task AcknowledgeAsync_DoesNotRetry_OnFailure()
|
||||||
|
{
|
||||||
|
var driver = new FakeAlarmSource { AcknowledgeShouldThrow = true };
|
||||||
|
var surface = NewSurface(driver, defaultHost: "h1");
|
||||||
|
|
||||||
|
await Should.ThrowAsync<InvalidOperationException>(() =>
|
||||||
|
surface.AcknowledgeAsync([new AlarmAcknowledgeRequest("s", "c", null)], CancellationToken.None));
|
||||||
|
|
||||||
|
driver.AcknowledgeCallCount.ShouldBe(1, "AlarmAcknowledge must not retry — decision #143");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task SubscribeAsync_Retries_Transient_Failures()
|
||||||
|
{
|
||||||
|
var driver = new FakeAlarmSource { SubscribeFailuresBeforeSuccess = 2 };
|
||||||
|
var surface = NewSurface(driver, defaultHost: "h1");
|
||||||
|
|
||||||
|
await surface.SubscribeAsync(["src"], CancellationToken.None);
|
||||||
|
|
||||||
|
driver.SubscribeCallCount.ShouldBe(3, "AlarmSubscribe retries by default — decision #143");
|
||||||
|
}
|
||||||
|
|
||||||
|
private static AlarmSurfaceInvoker NewSurface(
|
||||||
|
IAlarmSource driver,
|
||||||
|
string defaultHost,
|
||||||
|
IPerCallHostResolver? resolver = null)
|
||||||
|
{
|
||||||
|
var builder = new DriverResiliencePipelineBuilder();
|
||||||
|
var invoker = new CapabilityInvoker(builder, "drv-1", () => TierAOptions);
|
||||||
|
return new AlarmSurfaceInvoker(invoker, driver, defaultHost, resolver);
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed class FakeAlarmSource : IAlarmSource
|
||||||
|
{
|
||||||
|
public int SubscribeCallCount { get; private set; }
|
||||||
|
public int AcknowledgeCallCount { get; private set; }
|
||||||
|
public int SubscribeFailuresBeforeSuccess { get; set; }
|
||||||
|
public bool AcknowledgeShouldThrow { get; set; }
|
||||||
|
public IReadOnlyList<string> LastSubscribedIds { get; private set; } = [];
|
||||||
|
|
||||||
|
public Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
|
||||||
|
IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
SubscribeCallCount++;
|
||||||
|
LastSubscribedIds = sourceNodeIds;
|
||||||
|
if (SubscribeCallCount <= SubscribeFailuresBeforeSuccess)
|
||||||
|
throw new InvalidOperationException("transient");
|
||||||
|
return Task.FromResult<IAlarmSubscriptionHandle>(new StubHandle($"h-{SubscribeCallCount}"));
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
|
||||||
|
=> Task.CompletedTask;
|
||||||
|
|
||||||
|
public Task AcknowledgeAsync(
|
||||||
|
IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
AcknowledgeCallCount++;
|
||||||
|
if (AcknowledgeShouldThrow) throw new InvalidOperationException("ack boom");
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
public event EventHandler<AlarmEventArgs>? OnAlarmEvent { add { } remove { } }
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed record StubHandle(string DiagnosticId) : IAlarmSubscriptionHandle;
|
||||||
|
|
||||||
|
private sealed class StubResolver(Dictionary<string, string> map) : IPerCallHostResolver
|
||||||
|
{
|
||||||
|
public string ResolveHost(string fullReference) => map[fullReference];
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user