Files
UA-.NETStandard/tests/Opc.Ua.Redundancy.Client.Tests/ClientRedundancyServiceCollectionExtensionsTests.cs
b72eb44a31 Redundancy: bound the Raft log with the engine's snapshots (#4836)
# Description

**Draft: builds on #4835 and waits for a RaftCs release** containing
marcschier/raft-cs#9, marcschier/raft-cs#10 and marcschier/raft-cs#11,
anticipated as 1.1.4. It contains the commit of #4835; review only the
second commit.

`RaftSharedKeyValueStore` can capture and restore its state (#4794), but
`IRaftConsensus` carries no log indices, so the store cannot bound the
RaftCs log, which grows without limit. This change uses the engine's
snapshots:

- `IRaftSnapshotConsensus` (`CommittedItems` with log indices and
snapshots, `CompactAsync`) and `RaftCommittedItem`.
`DefaultRaftConsensus` implements it over `RaftNode.CommittedItems`,
`CreateSnapshotAsync` and `CompactAsync`.
- `RaftSharedKeyValueStore` gains a constructor overload with a
compaction interval and an optional logger (the existing constructors
are unchanged) and `AppliedIndex`. Over an `IRaftSnapshotConsensus` it
applies the indexed stream. A snapshot on that stream (a lagging replica
installing the leader's snapshot, or the first item after a restart)
replaces the state and publishes the differences to watchers. Every
compaction interval it snapshots its state into the engine, which
discards the log behind it. A compaction that fails is logged and
retried at the next interval.
- The engine snapshot is a format marker followed by the snapshot
encoding of #4794, so it carries the command position: the commands
after it keep the positions they have on every replica, and
`CreateSnapshotAsync`/`RestoreSnapshotAsync` keep working over it. No
compaction runs while a store restored with `FromFirstCommand` still
skips the replayed commands its snapshot holds.
- A snapshot in another format stops the applier. Pending and later
proposals and watches then fail with its error instead of waiting.
- `AppliedIndex` is published once the state holds an item, before
watchers and the proposer learn of it.
- Registration: `RedundancyConsistencyOptions.RaftCompactionInterval`
(server), and overloads of `AddRaftClientSharedStore` and
`AddRedundantClientSharedStore` that take the interval (client). Zero,
the default, never compacts. The store's logger comes from the
registered `ITelemetryContext`, else `ILoggerFactory`.
- `HighAvailability.md` describes the compaction and warns that a
replica must not be rolled back to a release without engine snapshots
once the cluster compacted its log.

No public member is removed or changed.

## Related Issues

- #4794 (store snapshots), #4835 (RaftCs 1.1.4)
- marcschier/raft-cs#9, marcschier/raft-cs#10, marcschier/raft-cs#11

## 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.
(`RaftSharedKeyValueStoreCompactionTests`; registration tests in Server
and Client.)
- [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. (Redundancy
test projects on .NET 10 against a RaftCs build of the three pull
requests: Server 935, Client 135, Kubernetes 103, PubSub 45, Samples 17,
all pass; the Server and Client test projects build for net48.)
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings. (CI cannot restore RaftCs 1.1.4 until it is
released.)
- [x] I have addressed **all** PR feedback received.

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Marc Schier <marcschier@users.noreply.github.com>
2026-10-20 14:48:20 +00:00

208 lines
8.4 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: system-under-test disposables are created per test and released at teardown
// or via the resolving container; there is no cross-test resource leak.
#pragma warning disable CA2000 // Dispose objects before losing scope
// CA2007: tests run without a SynchronizationContext; ConfigureAwait(false)
// adds noise without a behavioural benefit. Disabled file-level for the suite.
#pragma warning disable CA2007
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Crdt;
using Crdt.Transport;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using NUnit.Framework;
using Opc.Ua.Redundancy;
using Opc.Ua.Redundancy.Client;
using Opc.Ua.Tests;
namespace Opc.Ua.Client.Redundancy.Tests
{
/// <summary>
/// Coverage for the client-redundancy dependency-injection registration helpers, focusing on the
/// argument guards, the CRDT store registration, and the hosted-service lifecycle wiring.
/// </summary>
[TestFixture]
[Category("ClientRedundancy")]
public sealed class ClientRedundancyServiceCollectionExtensionsTests
{
private ITelemetryContext m_telemetry = null!;
[SetUp]
public void SetUp()
{
m_telemetry = NUnitTelemetryContext.Create();
}
[Test]
public async Task AddCrdtClientSharedStoreRegistersReplicatedStoreAsync()
{
await using var network = new InMemoryNetwork();
var services = new ServiceCollection();
services.AddCrdtClientSharedStore(ReplicaId.New(), _ => network.CreateTransport());
await using ServiceProvider provider = services.BuildServiceProvider();
ISharedKeyValueStore store = provider.GetRequiredService<ISharedKeyValueStore>();
Assert.That(store, Is.InstanceOf<ReplicatedSharedKeyValueStore>());
}
[Test]
public void AddCrdtClientSharedStoreThrowsWhenServicesNull()
{
Assert.That(
() => ClientRedundancyServiceCollectionExtensions.AddCrdtClientSharedStore(
null!,
ReplicaId.New(),
_ => null!),
Throws.TypeOf<ArgumentNullException>());
}
[Test]
public void AddCrdtClientSharedStoreThrowsWhenTransportFactoryNull()
{
var services = new ServiceCollection();
Assert.That(
() => services.AddCrdtClientSharedStore(ReplicaId.New(), null!),
Throws.TypeOf<ArgumentNullException>());
}
[Test]
public async Task CompactionIntervalReachesTheClientRaftStoreAsync()
{
var services = new ServiceCollection();
services.AddRaftClientSharedStore(null, compactionInterval: 5);
await using (ServiceProvider provider = services.BuildServiceProvider())
{
ISharedKeyValueStore store = provider.GetRequiredService<ISharedKeyValueStore>();
bool created = await store.CompareAndSwapAsync("k", default, ByteString.From(new byte[] { 1 })).ConfigureAwait(false);
Assert.That(created, Is.True);
}
// A consensus without snapshots cannot compact: the interval reaches the store, which refuses it.
var raft = new ServiceCollection();
raft.AddRaftClientSharedStore(_ => new InProcessRaftConsensus(), compactionInterval: 5);
var redundant = new ServiceCollection();
redundant.AddRedundantClientSharedStore(
RedundancyConsistencyMode.Strong,
ReplicaId.New(),
null,
_ => new InProcessRaftConsensus(),
raftCompactionInterval: 5);
await using ServiceProvider raftProvider = raft.BuildServiceProvider();
await using ServiceProvider redundantProvider = redundant.BuildServiceProvider();
Assert.Multiple(() =>
{
Assert.That(() => raftProvider.GetRequiredService<ISharedKeyValueStore>(), Throws.ArgumentException);
Assert.That(() => redundantProvider.GetRequiredService<ISharedKeyValueStore>(), Throws.ArgumentException);
});
}
[Test]
public void AddRaftClientSharedStoreThrowsWhenServicesNull()
{
Assert.That(
() => ClientRedundancyServiceCollectionExtensions.AddRaftClientSharedStore(null!),
Throws.TypeOf<ArgumentNullException>());
}
[Test]
public void AddRedundantClientSharedStoreThrowsWhenServicesNull()
{
Assert.That(
() => ClientRedundancyServiceCollectionExtensions.AddRedundantClientSharedStore(
null!,
RedundancyConsistencyMode.Strong,
ReplicaId.New()),
Throws.TypeOf<ArgumentNullException>());
}
[Test]
public void AddRedundantClientSessionThrowsWhenServicesNull()
{
Assert.That(
() => ClientRedundancyServiceCollectionExtensions.AddRedundantClientSession(
null!,
_ => { }),
Throws.TypeOf<ArgumentNullException>());
}
[Test]
public void AddRedundantClientSessionThrowsWhenConfigureNull()
{
var services = new ServiceCollection();
Assert.That(
() => services.AddRedundantClientSession(null!),
Throws.TypeOf<ArgumentNullException>());
}
[Test]
public void AddRedundantClientSessionThrowsWhenCreateSessionMissing()
{
var services = new ServiceCollection();
Assert.That(
() => services.AddRedundantClientSession(options => options.NodeId = "replica-a"),
Throws.TypeOf<ArgumentException>());
}
[Test]
public async Task RedundantClientSessionHostedServiceStartsAndStopsSessionAsync()
{
var services = new ServiceCollection();
services.AddSingleton(m_telemetry);
services.AddSingleton<ISharedKeyValueStore>(_ => new InMemorySharedKeyValueStore());
services.AddSingleton<ILeaderElection>(new StaticLeaderElection(false));
services.AddRedundantClientSession(options =>
{
options.NodeId = "replica-a";
options.CreateSessionAsync = _ => default;
});
await using ServiceProvider provider = services.BuildServiceProvider();
IHostedService hosted = provider.GetServices<IHostedService>().Single();
RedundantClientSession session = provider.GetRequiredService<RedundantClientSession>();
await hosted.StartAsync(CancellationToken.None).ConfigureAwait(false);
Assert.That(session.Disposed, Is.False);
await hosted.StopAsync(CancellationToken.None).ConfigureAwait(false);
Assert.That(session.Disposed, Is.True);
}
}
}