mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-29 15:11:53 +08:00
# 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>
208 lines
8.4 KiB
C#
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);
|
|
}
|
|
}
|
|
}
|