Files
UA-.NETStandard/tests/Opc.Ua.Subscriptions.Tests/SubscriptionManagerCoverageTests.cs
b409714731 Client: acknowledgement provider model and subscription acknowledgement fixes (#4899)
# Description

Fixes client subscription acknowledgement defects and replaces the
mutable event-based acknowledgement policy with an injectable,
asynchronous provider model. The latest changes in `7be912614` add
opt-in **whole-message delivery**, harden lifecycle/concurrency
behavior, and simplify acknowledgement state and wire-array
construction. Existing features and all deprecations remain in this PR.

The acknowledgement unit is one OPC UA `NotificationMessage`, not one
data change, event, downstream message, or group of Publish responses
indicated by `MoreNotifications`.

### Protocol and lifecycle fixes

- Ensure acknowledgements make progress when fewer acknowledgements are
queued than there are Publish workers.
- Honor legacy deferral on V2 and preserve deferred/in-flight
acknowledgements across retries. A throwing legacy callback restores the
original lists; a throwing provider defers the batch.
- Validate ownership before sending and after provider decisions,
including subscription removal, recreation, changed IDs and server ID
reuse. Reject detected responses from the previous owner before they
advance the replacement subscription's sequence cursor.
- Prune acknowledgements no longer retained by the server, including
servers without a retransmission queue.
- Retire one subscription's acknowledgement state without quiescing
publishing for the whole session.
- Reserve the classic engine's provider-decision slot before invoking
application code. Concurrent Publish calls cannot overwrite pending
decisions; asynchronous failures are observed and logged.
- Preserve the original outstanding lifecycle when a cancelled queue
attempt can still be queued. Retries do not create an additional
terminal outcome.
- Retire internal confirmation state synchronously, independently of
application observer progress. Observer callbacks remain asynchronous
and outside SDK locks.

### Public API and whole-message delivery

| API | Purpose |
|---|---|
| `AcknowledgementHandle` | Process-local acknowledgement identity
scoped to a subscription lifetime; avoids correlating messages solely by
subscription ID and sequence number. |
| `ISubscriptionAcknowledgementProvider` | Asynchronous, fail-closed
send/defer decisions over `ArrayOf<AcknowledgementHandle>`. |
| `ISubscriptionAcknowledgementObserver` | Terminal outcomes:
`Acknowledged`, `NotRetained`, `OwnerRetired`, or `Abandoned`, subject
to the cross-session limitation below. |
| `AcknowledgementMode.AfterConfirmation` /
`ISessionAcknowledgements.Confirm` | Do not acknowledge until the
application explicitly confirms processing of the complete message. |
| `ISubscriptionNotificationMessageHandler` /
`SubscriptionNotificationMessage` | Opt-in delivery of all typed
payloads in one envelope, preserving payload order, physical-partition
identity, publish time and diagnostic string table. |

- Provider configuration is available through `ManagedSessionOptions`,
fluent builder methods and DI registration.
- Whole-message handlers are registered through the V2 subscription APIs
or `Subscription.NotificationMessageHandler` on the classic engine.
Existing load/restore factories can return handlers implementing both
notification interfaces.
- Whole-message handling replaces data/event callbacks only for the
opted-in subscription; keep-alive and lifecycle handling remain
available. Returning from the handler does not implicitly confirm
delivery.
- Projection arrays do not borrow pooled wire notification containers.
Monitored-item references remain live SDK objects.
- Whole-message handlers receive shutdown cancellation before
subscription retirement waits for their dispatch to finish.
- Managed model-change tracking owns a dedicated whole-message
subscription instead of sharing application streaming acknowledgement
ownership. It confirms only after all events succeed; namespace-refresh,
cache-invalidation and subscriber failures remain logged but do not
produce a success confirmation.

### Compatibility and deprecations

The following deprecations remain, with compatibility behavior
preserved:

- `ISession.PublishSequenceNumbersToAcknowledge`: honored when no
provider is configured; not raised when a provider owns the policy.
- Classic `FastDataChangeCallback`, `FastEventCallback`,
`FastKeepAliveCallback` and their delegate types.
- `ISession.DefaultSubscription` and its implementations.

The lifetime proposal from the closed PR #4901 is incorporated through
the provider/handle model rather than additional lifecycle events.

### Simplification and performance scope

- Pending acknowledgements derive owner and generation from their scope
rather than storing redundant copies.
- Wire acknowledgements are projected directly into the final array,
eliminating the intermediate converted list and its extra copy.
- Whole-message envelopes are constructed only for opted-in handlers.
The existing payload-callback path remains available.
- Handles reuse cached subscription-lifetime scopes. Provider callbacks
execute outside SDK locks.

This is not a benchmarked throughput or allocation-parity claim. Broader
lock/contention work, provider-path allocations and benchmark evidence
remain tracked by #4912.

### Known limitations and non-goals

- **#4929 remains open:** classic subscriptions served by the default V2
compatibility bridge do not yet participate in normal acknowledgement
tracking. Supporting the classic engine is not the same as fixing that
bridge.
- **#4930 remains open:** transferring a classic subscription between
different `Session` objects does not yet move tracked
acknowledgements/confirmations between engines. Duplicate terminal
reports or lost confirmations remain possible on that path. In-place
managed-session recreation is a different, supported path.
- Delaying protocol acknowledgement is not itself downstream retry or
durable storage. Applications still own downstream retry/buffering;
server retransmission queues are finite.
- External observer delivery is buffered. Slow observers can grow their
notification backlog, although they no longer retain internal
confirmation state. Disposal waits up to five seconds for callbacks; an
unresponsive observer can outlive disposal.
- HA takeover/session-continuation redesign and server-side
acknowledgement defects are not addressed here.

Documentation: [Subscriptions](docs/Subscriptions.md), [Migration
guide](docs/MigrationGuide.md), and [Model-change
tracking](docs/ModelChangeTracking.md).

### Validation

Local validation of the changes included in `7be912614`:

| Check | Result |
|---|---|
| Expanded net10 client regression selection | **389 passed, 0 failed, 0
skipped** |
| Final API-registration and whole-message-routing selection | **3
passed**; two overlap the regression selection, one is additional (**390
distinct client cases across the runs**) |
| Acknowledgement integration tests | **4 passed**: classic/V2, each
with legacy payload callbacks and whole-message handling |
| `Opc.Ua.Client` net48 build | **0 warnings, 0 errors** |
| Scoped whitespace, style and analyzer verification | Passed, including
informational severity for style/analyzers |
| Negative/revert controls | Tests reproduced the provider race, missing
whole-message routing, observer-dependent cleanup, cancelled-queue
lifecycle loss, shutdown deadlock and false model-change confirmation;
corrections were restored and positive cases passed |

<details>
<summary>Exact local test/build commands and results</summary>

Run from the repository root. Framework-specific assets must be restored
before commands using `--no-restore`; perform the net48 build last
because changing `CustomTestTarget` replaces restored assets.

```powershell
dotnet test tests\Opc.Ua.Client.Tests\Opc.Ua.Client.Tests.csproj -p:CustomTestTarget=net10.0 --framework net10.0 --no-restore --filter "FullyQualifiedName~AcknowledgementProviderTests|FullyQualifiedName~SubscriptionManagerTests|FullyQualifiedName~ClassicSubscriptionEngineTests|FullyQualifiedName~ManagedSessionAcknowledgementForwardingTests|FullyQualifiedName~AcknowledgementOptionsTests|FullyQualifiedName~SubscriptionTests|FullyQualifiedName~ModelChangeTrackerTests|FullyQualifiedName~PartitionForwardingHandlerTests|FullyQualifiedName~SessionEngineContextTests|FullyQualifiedName~DefaultSubscriptionEngineTests|FullyQualifiedName~MonitoredItemManagerTests|FullyQualifiedName~MessageProcessorTests" --logger "console;verbosity=minimal" -v quiet -nologo
# Passed: 389; failed: 0; skipped: 0.

dotnet test tests\Opc.Ua.Client.Tests\Opc.Ua.Client.Tests.csproj -p:CustomTestTarget=net10.0 --framework net10.0 --no-restore --filter "FullyQualifiedName~WholeMessageRegistrationRejectsMissingInputs|FullyQualifiedName~WholeMessageDeliveryPreservesMixedPayloadsAsync" --logger "console;verbosity=minimal" -v quiet -nologo
# Passed: 3; failed: 0; skipped: 0. Two cases overlap the previous selection.

dotnet test tests\Opc.Ua.Subscriptions.Tests\Opc.Ua.Subscriptions.Tests.csproj -p:CustomTestTarget=net10.0 --framework net10.0 --no-restore --filter "FullyQualifiedName~AcknowledgementProviderIntegrationTests" --logger "console;verbosity=minimal" -v quiet -nologo
# Passed: 4; failed: 0; skipped: 0.

dotnet build src\Opc.Ua.Client\Opc.Ua.Client.csproj -p:CustomTestTarget=net48 -p:RestoreSources=https://www.nuget.org/api/v2/ -p:NuGetAudit=false -v quiet -nologo
# Build succeeded: 0 warnings, 0 errors.
```

The v2 feed and disabled NuGet audit above were local restore
workarounds only; repository settings were not changed. CI dependency
auditing remains required.

</details>

The full `UA.slnx` net48/net10 test matrix, NativeAOT publishing/runtime
validation, fresh coverage measurements and performance benchmarks were
**not run locally for this revision**. CI and maintainer API review
remain outstanding. Historical results on `a28c12252` are not presented
as validation of the latest head.

## Related Issues

_If this is a relatively large or complex change, a design must have
been discussed in the related tracking issue and signed off (which
becomes the Architectural Decision Record (ADR))._

- Fixes #4894: acknowledgement progress, deferral/retry behavior and
ownership validation.
- Fixes #4898: injectable fail-closed decision API, with legacy
precedence preserved.
- Fixes #4913: admission of detected old-owner responses after
subscription ID reuse.

Non-closing references:

| Issue | Title / scope | Disposition |
|---|---|---|
| [#4897](https://github.com/OPCFoundation/UA-.NETStandard/issues/4897)
| Stable notification identity and terminal acknowledgement state | Core
functionality is implemented via handles/providers, with whole-message
delivery added. Keep this link non-closing until the cross-session
exception in #4930 is resolved or the issue's accepted scope is
explicitly narrowed. |
| [#4912](https://github.com/OPCFoundation/UA-.NETStandard/issues/4912)
| Reduce acknowledgement-path lock contention and allocations |
Partially advanced by state/projection simplification; broader locking
changes and benchmarks remain. |
| [#4929](https://github.com/OPCFoundation/UA-.NETStandard/issues/4929)
| Classic subscriptions on the V2 compatibility bridge | Not fixed;
separate follow-up. |
| [#4930](https://github.com/OPCFoundation/UA-.NETStandard/issues/4930)
| Cross-session classic acknowledgement handover | Not fixed; separate
follow-up and explicit limitation on terminal-outcome guarantees. |

Design rationale: [maintainer request for injectable providers and
session
facets](https://github.com/OPCFoundation/UA-.NETStandard/pull/4899#discussion_r4366515418).
The new whole-message API is submitted for maintainer review; this
description does not claim that its final API shape or an ADR has been
signed off.

## Checklist

- [x] I have signed the
[CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf)
and read the
[CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md)
doc.
- [x] I have added tests that prove my fix is effective or that my
feature works and increased code coverage.
- [x] I have added all necessary documentation.
- [x] I have verified that my changes do not introduce (new) build or
analyzer warnings.
- [ ] I ran **all** tests locally using the **UA.slnx** solution against
at least .net **framework** and .net **10**, and all passed.
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [ ] I have addressed **all** PR feedback received.

Regression tests were added and passed; the combined tests/coverage
checkbox is left open because a fresh coverage increase has not yet been
verified.

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Marc Schier <marcschier@users.noreply.github.com>
Co-authored-by: Marc <marcschier@hotmail.com>
2026-10-28 13:55:20 +00:00

824 lines
34 KiB
C#

/* ========================================================================
* Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
*
* OPC Foundation MIT License 1.00
*
* Permission is hereby granted, free of charge, to any person
* obtaining a copy of this software and associated documentation
* files (the "Software"), to deal in the Software without
* restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the
* Software is furnished to do so, subject to the following
* conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
* OTHER DEALINGS IN THE SOFTWARE.
*
* The complete license agreement can be found here:
* http://opcfoundation.org/License/MIT/1.00/
* ======================================================================*/
// CA2000: test code; the hand-rolled managed-subscription stubs are
// IAsyncDisposable and their ownership is transferred to the manager under
// test (it disposes them on DisposeAsync). CA2000 cannot see through that
// transfer and would be noisy without a real leak risk. Disabled file-level.
#pragma warning disable CA2000
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Moq;
using NUnit.Framework;
using Opc.Ua.Client.Subscriptions;
using Opc.Ua.Client.Subscriptions.MonitoredItems;
namespace Opc.Ua.Subscriptions.Tests
{
/// <summary>
/// Deterministic, fully offline unit tests for the internal
/// <see cref="SubscriptionManager"/>. The background publish
/// controller launched in the constructor is kept quiescent: the
/// stub context never marks a subscription as <c>Created</c>, so
/// <c>GetDesiredPublishWorkerCount</c> returns 0 and no publish
/// worker is ever spawned. Every test asserts on concrete values
/// and never depends on wall-clock timing.
/// </summary>
[TestFixture]
[Category("Client")]
[Category("SubscriptionManager")]
[Category("SubscriptionManagerUnit")]
[SetCulture("en-us")]
[SetUICulture("en-us")]
public sealed class SubscriptionManagerCoverageTests
{
private static SubscriptionManager CreateManager(
StubSubscriptionManagerContext context,
DiagnosticsMasks diagnostics = DiagnosticsMasks.None)
{
return new SubscriptionManager(
context, NullLoggerFactory.Instance, diagnostics);
}
private static OptionsMonitor<SubscriptionOptions> SinglePartitionOptions()
{
// DisableUnboundedItemMode selects the inert single-partition
// wrapper path in Add (no placement policy, factory or idle
// timer), keeping the manager fully deterministic.
return new OptionsMonitor<SubscriptionOptions>(
new SubscriptionOptions { DisableUnboundedItemMode = true });
}
[Test]
public async Task ConstructorInitializesQuiescentDefaultsAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
Assert.That(manager.Count, Is.Zero);
Assert.That(manager.Items, Is.Empty);
Assert.That(manager.PublishWorkerCount, Is.Zero);
Assert.That(manager.GoodPublishRequestCount, Is.Zero);
Assert.That(manager.BadPublishRequestCount, Is.Zero);
Assert.That(manager.MissingMessageCount, Is.Zero);
Assert.That(manager.RepublishMessageCount, Is.Zero);
Assert.That(manager.CreatedCount, Is.Zero);
Assert.That(manager.Created, Is.Empty);
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(2));
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(15));
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.False);
Assert.That(manager.PoolNotifications, Is.False);
}
// The publish controller stayed parked the whole time.
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task ConstructorStoresReturnDiagnosticsAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context, DiagnosticsMasks.All);
await using (manager.ConfigureAwait(false))
{
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.All));
}
}
[Test]
public async Task MinPublishWorkerCountHonoursChangedAndEqualBranchesAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(2));
// Equal-value early-return branch: value must be unchanged.
manager.MinPublishWorkerCount = 2;
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(2));
// Changed-value branch: value updated and controller signalled.
manager.MinPublishWorkerCount = 7;
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(7));
// Equal-value branch again on the new value.
manager.MinPublishWorkerCount = 7;
Assert.That(manager.MinPublishWorkerCount, Is.EqualTo(7));
}
// No subscription is Created, so signalling never spawns a worker.
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task MaxPublishWorkerCountHonoursChangedAndEqualBranchesAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(15));
// Equal-value early-return branch.
manager.MaxPublishWorkerCount = 15;
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(15));
// Changed-value branch.
manager.MaxPublishWorkerCount = 30;
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(30));
// Equal-value branch again.
manager.MaxPublishWorkerCount = 30;
Assert.That(manager.MaxPublishWorkerCount, Is.EqualTo(30));
}
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task TransferSubscriptionsOnRecreateRoundTripsAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.False);
manager.TransferSubscriptionsOnRecreate = true;
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.True);
manager.TransferSubscriptionsOnRecreate = false;
Assert.That(manager.TransferSubscriptionsOnRecreate, Is.False);
}
}
[Test]
public async Task PoolNotificationsRoundTripsAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
// PoolNotifications is a single public auto-property that
// implicitly implements both ISubscriptionManager and
// IMessageAckQueue, so the manager surface is authoritative.
Assert.That(manager.PoolNotifications, Is.False);
manager.PoolNotifications = true;
Assert.That(manager.PoolNotifications, Is.True);
manager.PoolNotifications = false;
Assert.That(manager.PoolNotifications, Is.False);
}
}
[Test]
public async Task ReturnDiagnosticsRoundTripsAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context, DiagnosticsMasks.None);
await using (manager.ConfigureAwait(false))
{
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.None));
manager.ReturnDiagnostics = DiagnosticsMasks.All;
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.All));
manager.ReturnDiagnostics = DiagnosticsMasks.OperationAll;
Assert.That(manager.ReturnDiagnostics, Is.EqualTo(DiagnosticsMasks.OperationAll));
}
}
[Test]
public async Task QueueAsyncThenDropPendingIsIdempotentAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
var first = new StubManagedSubscription { Id = 1u };
var second = new StubManagedSubscription { Id = 2u };
await manager.QueueAsync(first, first.MessageGeneration, new SubscriptionAcknowledgement
{
SubscriptionId = 1u,
SequenceNumber = 10u
}, CancellationToken.None).ConfigureAwait(false);
await manager.QueueAsync(second, second.MessageGeneration, new SubscriptionAcknowledgement
{
SubscriptionId = 2u,
SequenceNumber = 11u
}, CancellationToken.None).ConfigureAwait(false);
// Queued entries are validated when sent; only deferred ones are removed here.
Assert.That(manager.DropPendingForSubscription(first, 1u), Is.Zero);
Assert.That(manager.DropPendingForSubscription(first, 1u), Is.Zero);
Assert.That(manager.QueuedAcknowledgementCount, Is.EqualTo(2));
}
}
[Test]
public async Task DropPendingForSubscriptionOnEmptyQueueReturnsZeroAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
Assert.That(manager.DropPendingForSubscription(new StubManagedSubscription { Id = 7u }, 7u),
Is.Zero);
}
}
[Test]
public async Task CompleteAsyncUnknownSubscriptionReturnsWithoutChangeAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
await manager.CompleteAsync(new StubManagedSubscription { Id = 999u }, 999u,
CancellationToken.None).ConfigureAwait(false);
Assert.That(manager.Count, Is.Zero);
}
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task UpdateSignalsControllerWithoutThrowingAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
manager.Update();
Assert.That(manager.Count, Is.Zero);
}
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task DrainAsyncCompletesImmediatelyWhenNoPublishActiveAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
Task drain = manager.DrainAsync(CancellationToken.None);
Assert.That(drain.IsCompleted, Is.True);
await drain.ConfigureAwait(false);
}
}
[Test]
public async Task ResumeWithoutSubscriptionsDoesNotThrowAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
manager.Resume();
Assert.That(manager.Count, Is.Zero);
}
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task PauseWithoutSubscriptionsDoesNotThrowAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
manager.Pause();
Assert.That(manager.Count, Is.Zero);
}
}
[Test]
public async Task RecreateSubscriptionsAsyncWithNoSubscriptionsReturnsImmediatelyAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
await manager.RecreateSubscriptionsAsync(null, CancellationToken.None)
.ConfigureAwait(false);
Assert.That(manager.Count, Is.Zero);
}
}
[Test]
public async Task DisposeAsyncIsIdempotentAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await manager.DisposeAsync().ConfigureAwait(false);
// Second dispose must be a no-op and must not throw.
await manager.DisposeAsync().ConfigureAwait(false);
Assert.That(manager.Count, Is.Zero);
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task AddRegistersSubscriptionAndReturnsWrapperAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscription = new StubManagedSubscription { Id = 1u };
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscription;
ISubscription result = manager.Add(handler, SinglePartitionOptions());
Assert.That(result, Is.Not.Null);
Assert.That(manager.Count, Is.EqualTo(1));
Assert.That(manager.Items, Has.Exactly(1).Items);
Assert.That(context.CreateSubscriptionCallCount, Is.EqualTo(1));
Assert.That(context.LastCreateQueue, Is.SameAs(manager));
Assert.That(manager.PublishWorkerCount, Is.Zero);
}
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task AddWithNullHandlerThrowsArgumentNullExceptionAsync()
{
var context = new StubSubscriptionManagerContext();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
ArgumentNullException ex = Assert.Throws<ArgumentNullException>(
() => manager.Add(null!, SinglePartitionOptions()));
Assert.That(ex.ParamName, Is.EqualTo("handler"));
}
}
[Test]
public async Task AddWithNullOptionsThrowsArgumentNullExceptionAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
ArgumentNullException ex = Assert.Throws<ArgumentNullException>(
() => manager.Add(handler, null!));
Assert.That(ex.ParamName, Is.EqualTo("options"));
}
}
[Test]
public async Task AddTwoSubscriptionsCountAndItemsReflectBothAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscriptions = new Queue<IManagedSubscription>(
[
new StubManagedSubscription { Id = 1u },
new StubManagedSubscription { Id = 2u }
]);
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscriptions.Dequeue();
manager.Add(handler, SinglePartitionOptions());
manager.Add(handler, SinglePartitionOptions());
Assert.That(manager.Count, Is.EqualTo(2));
Assert.That(manager.Items, Has.Exactly(2).Items);
Assert.That(context.CreateSubscriptionCallCount, Is.EqualTo(2));
}
}
[Test]
public async Task MissingMessageCountAggregatesAcrossSubscriptionsAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscriptions = new Queue<IManagedSubscription>(
[
new StubManagedSubscription { Id = 1u, MissingMessageCount = 3 },
new StubManagedSubscription { Id = 2u, MissingMessageCount = 4 }
]);
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscriptions.Dequeue();
manager.Add(handler, SinglePartitionOptions());
manager.Add(handler, SinglePartitionOptions());
Assert.That(manager.MissingMessageCount, Is.EqualTo(7));
}
}
[Test]
public async Task RepublishMessageCountAggregatesAcrossSubscriptionsAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscriptions = new Queue<IManagedSubscription>(
[
new StubManagedSubscription { Id = 1u, RepublishMessageCount = 2 },
new StubManagedSubscription { Id = 2u, RepublishMessageCount = 5 }
]);
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscriptions.Dequeue();
manager.Add(handler, SinglePartitionOptions());
manager.Add(handler, SinglePartitionOptions());
Assert.That(manager.RepublishMessageCount, Is.EqualTo(7));
}
}
[Test]
public async Task CreatedCountIsZeroWhenSubscriptionsAreNotCreatedAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscription = new StubManagedSubscription { Id = 1u, Created = false };
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscription;
manager.Add(handler, SinglePartitionOptions());
Assert.That(manager.Count, Is.EqualTo(1));
Assert.That(manager.CreatedCount, Is.Zero);
Assert.That(manager.Created, Is.Empty);
Assert.That(manager.PublishWorkerCount, Is.Zero);
}
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task CompleteAsyncRemovesKnownSubscriptionAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscription = new StubManagedSubscription { Id = 1u };
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscription;
manager.Add(handler, SinglePartitionOptions());
Assert.That(manager.Count, Is.EqualTo(1));
await manager.CompleteAsync(subscription, 1u, CancellationToken.None).ConfigureAwait(false);
Assert.That(manager.Count, Is.Zero);
Assert.That(manager.Items, Is.Empty);
// CompleteAsync only unregisters; it does not dispose the partition.
Assert.That(subscription.DisposeAsyncCalls, Is.Zero);
}
}
[Test]
public async Task DisposeAsyncDisposesRegisteredSubscriptionsAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscription = new StubManagedSubscription { Id = 1u };
SubscriptionManager manager = CreateManager(context);
context.CreateSubscriptionFactory = (_, _, _) => subscription;
manager.Add(handler, SinglePartitionOptions());
Assert.That(manager.Count, Is.EqualTo(1));
await manager.DisposeAsync().ConfigureAwait(false);
Assert.That(subscription.DisposeAsyncCalls, Is.EqualTo(1));
Assert.That(manager.Count, Is.Zero);
}
[Test]
public async Task RecreateSubscriptionsAsyncInvokesRecreateOnEachSubscriptionAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscription = new StubManagedSubscription { Id = 1u, Created = false };
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscription;
manager.Add(handler, SinglePartitionOptions());
// previousSessionId == null and TransferSubscriptionsOnRecreate
// defaults to false, so no server transfer is attempted; every
// subscription is recreated directly.
await manager.RecreateSubscriptionsAsync(null, CancellationToken.None)
.ConfigureAwait(false);
Assert.That(subscription.RecreateAsyncCalls, Is.EqualTo(1));
}
Assert.That(context.PublishCallCount, Is.Zero);
}
[Test]
public async Task ResumeAndPauseNotifyRegisteredSubscriptionsAsync()
{
var context = new StubSubscriptionManagerContext();
var handler = new NoopNotificationHandler();
var subscription = new StubManagedSubscription { Id = 1u };
SubscriptionManager manager = CreateManager(context);
await using (manager.ConfigureAwait(false))
{
context.CreateSubscriptionFactory = (_, _, _) => subscription;
manager.Add(handler, SinglePartitionOptions());
manager.Resume();
manager.Pause();
// Resume notifies with paused=false, then Pause with true.
Assert.That(subscription.PausedCalls, Has.Count.EqualTo(2));
Assert.That(subscription.PausedCalls[0], Is.False);
Assert.That(subscription.PausedCalls[1], Is.True);
}
Assert.That(context.PublishCallCount, Is.Zero);
}
/// <summary>
/// Hand-rolled stub for <see cref="ISubscriptionManagerContext"/>.
/// Keeps the publish controller quiescent: it never returns publish
/// work and records that <see cref="PublishAsync"/> was not reached.
/// </summary>
private sealed class StubSubscriptionManagerContext : ISubscriptionManagerContext
{
public int SessionSubscriptionCount => 0;
public int CreateSubscriptionCallCount { get; private set; }
public IMessageAckQueue? LastCreateQueue { get; private set; }
public int PublishCallCount => Volatile.Read(ref m_publishCallCount);
public Func<ISubscriptionNotificationHandler,
IOptionsMonitor<SubscriptionOptions>, IMessageAckQueue,
IManagedSubscription>? CreateSubscriptionFactory
{ get; set; }
public IManagedSubscription CreateSubscription(
ISubscriptionNotificationHandler handler,
IOptionsMonitor<SubscriptionOptions> options,
IMessageAckQueue queue,
SubscriptionLoadState? loadState = null,
Action<IManagedSubscription>? initialize = null)
{
CreateSubscriptionCallCount++;
LastCreateQueue = queue;
if (CreateSubscriptionFactory == null)
{
throw new InvalidOperationException(
"CreateSubscriptionFactory was not configured for this test.");
}
IManagedSubscription subscription = CreateSubscriptionFactory(handler, options, queue);
initialize?.Invoke(subscription);
return subscription;
}
/// <inheritdoc/>
public Client.ISubscriptionAcknowledgementProvider? AcknowledgementProvider => null;
/// <inheritdoc/>
public Client.ISubscriptionAcknowledgementObserver? AcknowledgementObserver => null;
/// <inheritdoc/>
public bool HasAcknowledgementPolicy => false;
/// <inheritdoc/>
public AcknowledgementSelection PrepareAcknowledgementsToSend(
List<SubscriptionAcknowledgement> acknowledgements)
{
return new AcknowledgementSelection(acknowledgements, []);
}
public ValueTask<PublishResponse> PublishAsync(
RequestHeader? requestHeader,
ArrayOf<SubscriptionAcknowledgement> subscriptionAcknowledgements,
CancellationToken ct = default)
{
// Never expected to run: no subscription is marked Created, so
// the controller spawns zero workers. Record the call so tests
// can prove the loop stayed quiescent.
Interlocked.Increment(ref m_publishCallCount);
return new ValueTask<PublishResponse>(new PublishResponse());
}
public ValueTask<TransferSubscriptionsResponse> TransferSubscriptionsAsync(
RequestHeader? requestHeader,
ArrayOf<uint> subscriptionIds,
bool sendInitialValues,
CancellationToken ct = default)
{
return new ValueTask<TransferSubscriptionsResponse>(
new TransferSubscriptionsResponse());
}
public ValueTask<DeleteSubscriptionsResponse> DeleteSubscriptionsAsync(
RequestHeader? requestHeader,
ArrayOf<uint> subscriptionIds,
CancellationToken ct = default)
{
return new ValueTask<DeleteSubscriptionsResponse>(
new DeleteSubscriptionsResponse());
}
/// <summary>
/// Reports that no subscription in this test is owned by the session
/// outside the manager's registry, so every identifier falls through
/// to the manager's own handling.
/// </summary>
public bool TryDispatchToSessionSubscription(
uint subscriptionId,
NotificationMessage message,
ArrayOf<uint> availableSequenceNumbers,
ArrayOf<string> stringTable,
bool moreNotifications)
{
return false;
}
private int m_publishCallCount;
}
/// <summary>
/// Minimal, inert notification handler. None of its callbacks are
/// invoked because no publish response is ever dispatched.
/// </summary>
private sealed class NoopNotificationHandler : ISubscriptionNotificationHandler
{
public ValueTask OnDataChangeNotificationAsync(ISubscription subscription,
uint sequenceNumber, DateTime publishTime,
ReadOnlyMemory<DataValueChange> notification,
PublishState publishStateMask, IReadOnlyList<string> stringTable)
{
throw new NotSupportedException();
}
public ValueTask OnEventDataNotificationAsync(ISubscription subscription,
uint sequenceNumber, DateTime publishTime,
ReadOnlyMemory<EventNotification> notification,
PublishState publishStateMask, IReadOnlyList<string> stringTable)
{
throw new NotSupportedException();
}
public ValueTask OnKeepAliveNotificationAsync(ISubscription subscription,
uint sequenceNumber, DateTime publishTime,
PublishState publishStateMask)
{
throw new NotSupportedException();
}
public ValueTask OnSubscriptionStateChangedAsync(ISubscription subscription,
SubscriptionState state, PublishState publishStateMask,
CancellationToken ct = default)
{
throw new NotSupportedException();
}
}
/// <summary>
/// Hand-rolled stub for the internal <see cref="IManagedSubscription"/>.
/// Exposes settable identity/counter state and records the small set of
/// lifecycle callbacks the tests assert on. All dispatch-driven members
/// throw because they are never reached in these quiescent tests.
/// </summary>
private sealed class StubManagedSubscription : IManagedSubscription
{
public uint Id { get; init; }
public long MessageGeneration => 0;
public Client.AcknowledgementScope GetAcknowledgementScope(long generation)
{
return new Client.AcknowledgementScope(this, generation, Id);
}
public bool Created { get; init; }
public bool IsCreationInProgress { get; init; }
public bool IsIntentionallyDeleted { get; init; }
public long MissingMessageCount { get; init; }
public long RepublishMessageCount { get; init; }
public int DisposeAsyncCalls { get; private set; }
public int RecreateAsyncCalls { get; private set; }
public List<bool> PausedCalls { get; } = [];
public TimeSpan CurrentPublishingInterval => TimeSpan.Zero;
public byte CurrentPriority => 0;
public uint CurrentLifetimeCount => 0;
public uint CurrentKeepAliveCount => 0;
public bool CurrentPublishingEnabled => false;
public uint CurrentMaxNotificationsPerPublish => 0;
public IMonitoredItemCollection MonitoredItems { get; } = Mock.Of<IMonitoredItemCollection>();
public ValueTask ConditionRefreshAsync(CancellationToken ct = default)
{
throw new NotSupportedException();
}
public ValueTask<TimeSpan> SetAsDurableAsync(
TimeSpan lifetime, CancellationToken ct = default)
{
throw new NotSupportedException();
}
public ValueTask<TimeSpan> SetAsDurableAsync(
Func<TimeSpan?> lifetime, CancellationToken ct = default)
{
throw new NotSupportedException();
}
public ValueTask<SetTriggeringResult> SetTriggeringAsync(
IMonitoredItem triggeringItem,
IReadOnlyCollection<IMonitoredItem>? linksToAdd = null,
IReadOnlyCollection<IMonitoredItem>? linksToRemove = null,
CancellationToken ct = default)
{
throw new NotSupportedException();
}
public ValueTask OnPublishReceivedAsync(NotificationMessage message,
IReadOnlyList<uint>? availableSequenceNumbers,
IReadOnlyList<string> stringTable)
{
throw new NotSupportedException();
}
public ValueTask<bool> TryCompleteTransferAsync(
IReadOnlyList<uint> availableSequenceNumbers,
CancellationToken ct = default)
{
throw new NotSupportedException();
}
public ValueTask RecreateAsync(CancellationToken ct = default)
{
RecreateAsyncCalls++;
return default;
}
public void NotifySubscriptionManagerPaused(bool paused)
{
PausedCalls.Add(paused);
}
public ValueTask DisposeAsync()
{
DisposeAsyncCalls++;
return default;
}
}
}
}