mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-22 09:37:12 +08:00
romanett
9b034f8291
# Description Nullable reference types are now enabled once, in `common.props`, and every project in the repository builds warning-clean with them. **Configuration** - `<Nullable>enable</Nullable>` lives in `common.props`. All per-project `<Nullable>` settings are removed: 178 project files (enable, annotations, and the `disable` in the three Aot test projects). - 419 per-file `#nullable enable`/`restore` directives and the 4 `#nullable disable` directives are removed. - Kept on purpose: - The 5 `<auto-generated>` schema files (`UANodeSet.cs`, `UA Type Dictionary.cs`, `OPCBinarySchema.cs`, `SecuredApplication.cs`, `PInvokeHelper.cs`) keep `#nullable enable`. The compiler ignores the project setting in generated code, so the directive is required there. - `tests/Opc.Ua.Interop.LegacyPeer` keeps `disable`. It builds against the 1.5.378 packages and does not import the repository props. - `docs/DeveloperGuide.md` and `.github/copilot-instructions.md` now describe the central setting. **Migration** (the 43 unset and 13 annotations-only projects, roughly 10k diagnostics) - **Production code: annotated at the declarations.** Covers `tools/Opc.Ua.SourceGeneration.Core`, `src/Opc.Ua.Lds.Server` and the LDS sample. - Members, parameters and returns are nullable where null actually flows. - Null-check helpers carry `[NotNullWhen]` (`IsNull`, `IsValidXmlName`, `IsMethodTypeDesign`). - `CoreUtils.Format`, `Utils.Format`, `ITemplateWriter.WriteLine` take `params object?[]`. `Template.AddReplacement` takes `string?` (rendering already skips null replacements). - `IRegisteredServerStore.RegisterAsync` returns `Task<RegistrationEntry?>`: it returns null when a registration is removed. - Dereferences backed by an invariant (for example, the validated model) use `!`, so runtime behaviour is unchanged. - **Behavioural changes in production code** (small, deliberate): - `NodeSetToModelDesign` throws `InvalidOperationException` when the file is not a NodeSet. Before, it threw a NullReferenceException on the next line. - `ModelDesignValidator` starts its lookup tables empty instead of null. A redundant `if (m_nodes == null) return;` guard is gone. - **Test code: minimal annotations.** - `?` on values that can be null, `!` where the test asserts non-null. - `= null!` for fixture members assigned in SetUp. - `TryGetStructure<T>(out T? x)` where the inferred `T?` broke the `IEncodeable` constraint. - A few redundant `== null` checks on structs (`NodeId`, `ArrayOf<T>`) folded into the adjacent `IsNull`/`Count` checks. **Notes for reviewers** - The diff is large (~1,500 files) but almost entirely mechanical: `?`, `!` and directive removal. The non-annotation changes are listed above. - Known follow-up, not done here: `IModelDesign.Namespaces/TargetNamespace/NamespaceUris` stay non-nullable even though the validator returns null before it loads a dictionary. Making them nullable would ripple through every generator call site. ## Related Issues - Fixes #4439 ## Checklist - [ ] 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. - [ ] 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. The "no new warnings" box covers net10.0 and net48: full non-incremental builds of `UA.slnx`, `tools/SourceGeneration.slnx`, `tools/MigrationAnalyzer.slnx` and the Aot test projects finished with 0 warnings and 0 errors under the repository's warnings-as-errors settings. net9.0 and net8.0 also had no nullable diagnostics, but they were not rebuilt after the final fixes. Tests have not been run locally yet. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1996 lines
98 KiB
C#
1996 lines
98 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;
|
|
// there is no cross-test resource leak. Suppressed file-level for the suite.
|
|
#pragma warning disable CA2000 // Dispose objects before losing scope
|
|
|
|
// IDE0230: the byte[] fixtures here are arbitrary binary payloads (e.g. forged/rogue
|
|
// records), not text; rewriting them as UTF-8 (u8) string literals would misrepresent
|
|
// the intent. Suppressed file-level for the suite.
|
|
#pragma warning disable IDE0230 // Use UTF-8 string literal
|
|
|
|
// 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.Buffers.Binary;
|
|
using System.Collections.Generic;
|
|
using System.Globalization;
|
|
using System.IO;
|
|
using System.Runtime.CompilerServices;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Crdt;
|
|
using Crdt.Transport;
|
|
using Moq;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Redundancy;
|
|
using Opc.Ua.Redundancy.Server;
|
|
using Opc.Ua.Tests;
|
|
|
|
namespace Opc.Ua.Server.Tests.Redundancy
|
|
{
|
|
/// <summary>
|
|
/// Unit tests for <see cref="InMemoryNodeStateStore"/>, including a real
|
|
/// <c>NodeState</c> binary round-trip through the store.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Distributed")]
|
|
[Parallelizable(ParallelScope.All)]
|
|
public class InMemoryNodeStateStoreTests
|
|
{
|
|
private const ushort NamespaceIndex = 1;
|
|
private ServiceMessageContext m_messageContext;
|
|
private SystemContext m_systemContext;
|
|
|
|
[OneTimeSetUp]
|
|
public void OneTimeSetUp()
|
|
{
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
var messageContext = ServiceMessageContext.CreateEmpty(telemetry);
|
|
messageContext.NamespaceUris.GetIndexOrAppend("urn:test:distributed");
|
|
m_messageContext = messageContext;
|
|
m_systemContext = new SystemContext(telemetry)
|
|
{
|
|
NamespaceUris = messageContext.NamespaceUris,
|
|
ServerUris = messageContext.ServerUris,
|
|
EncodeableFactory = messageContext.Factory
|
|
};
|
|
}
|
|
|
|
[Test]
|
|
public async Task UpsertTryGetAndDeleteNodeAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var nodeId = new NodeId("var1", NamespaceIndex);
|
|
var payload = ByteString.From(new byte[] { 1, 2, 3, 4 });
|
|
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, payload)).ConfigureAwait(false);
|
|
IStoredNode? loaded = await store.TryGetNodeAsync(nodeId).ConfigureAwait(false);
|
|
bool deleted = await store.DeleteNodeAsync(nodeId).ConfigureAwait(false);
|
|
IStoredNode? afterDelete = await store.TryGetNodeAsync(nodeId).ConfigureAwait(false);
|
|
|
|
Assert.That(loaded, Is.Not.Null);
|
|
Assert.That(loaded!.NodeId, Is.EqualTo(nodeId));
|
|
Assert.That(loaded.Payload.ToArray(), Is.EqualTo(payload.ToArray()));
|
|
Assert.That(deleted, Is.True);
|
|
Assert.That(afterDelete, Is.Null);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EnumerateReturnsAllStoredNodesAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var a = new NodeId("a", NamespaceIndex);
|
|
var b = new NodeId("b", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(a, ByteString.From(new byte[] { 1 }))).ConfigureAwait(false);
|
|
await store.UpsertNodeAsync(new StoredNode(b, ByteString.From(new byte[] { 2 }))).ConfigureAwait(false);
|
|
|
|
var ids = new List<NodeId>();
|
|
await foreach (IStoredNode node in store.EnumerateAsync())
|
|
{
|
|
ids.Add(node.NodeId);
|
|
}
|
|
|
|
Assert.That(ids, Is.EquivalentTo([a, b]));
|
|
}
|
|
|
|
[Test]
|
|
public async Task WriteAndReadValueRoundTripsAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var nodeId = new NodeId("v", NamespaceIndex);
|
|
var original = new DataValue(new Variant(42.0), StatusCodes.Good, DateTimeUtc.Now);
|
|
|
|
await store.WriteValueAsync(nodeId, original).ConfigureAwait(false);
|
|
(bool found, DataValue read) = await store.TryReadValueAsync(nodeId).ConfigureAwait(false);
|
|
|
|
Assert.That(found, Is.True);
|
|
Assert.That(read.WrappedValue, Is.EqualTo(original.WrappedValue));
|
|
Assert.That(read.StatusCode, Is.EqualTo(original.StatusCode));
|
|
Assert.That(read.SourceTimestamp, Is.EqualTo(original.SourceTimestamp));
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryReadValueMissingReturnsFalseAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
|
|
(bool found, DataValue read) = await store.TryReadValueAsync(new NodeId("nope", NamespaceIndex)).ConfigureAwait(false);
|
|
|
|
Assert.That(found, Is.False);
|
|
Assert.That(read.IsNull, Is.True);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EnumerateValuesReturnsAllStoredValuesInOnePassAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var a = new NodeId("a", NamespaceIndex);
|
|
var b = new NodeId("b", NamespaceIndex);
|
|
await store.WriteValueAsync(a, new DataValue(new Variant(1.0), StatusCodes.Good, DateTimeUtc.Now)).ConfigureAwait(false);
|
|
await store.WriteValueAsync(b, new DataValue(new Variant(2.0), StatusCodes.Good, DateTimeUtc.Now)).ConfigureAwait(false);
|
|
|
|
var seen = new Dictionary<NodeId, DataValue>();
|
|
await foreach ((NodeId nodeId, DataValue value) in store.EnumerateValuesAsync())
|
|
{
|
|
seen[nodeId] = value;
|
|
}
|
|
|
|
Assert.That(seen.Keys, Is.EquivalentTo([a, b]));
|
|
Assert.That(seen[a].WrappedValue, Is.EqualTo(new Variant(1.0)));
|
|
Assert.That(seen[b].WrappedValue, Is.EqualTo(new Variant(2.0)));
|
|
}
|
|
|
|
[Test]
|
|
public async Task EnumerateValuesRoundTripsThroughRecordProtectorAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
byte[] key = new byte[32];
|
|
for (int i = 0; i < key.Length; i++)
|
|
{
|
|
key[i] = (byte)(i + 1);
|
|
}
|
|
using var protector = new AesCbcHmacRecordProtector(key);
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var nodeId = new NodeId("protected", NamespaceIndex);
|
|
await store.WriteValueAsync(nodeId, new DataValue(new Variant(7.0), StatusCodes.Good, DateTimeUtc.Now)).ConfigureAwait(false);
|
|
|
|
var seen = new Dictionary<NodeId, DataValue>();
|
|
await foreach ((NodeId id, DataValue value) in store.EnumerateValuesAsync())
|
|
{
|
|
seen[id] = value;
|
|
}
|
|
|
|
Assert.That(seen.Keys, Is.EquivalentTo([nodeId]));
|
|
Assert.That(seen[nodeId].WrappedValue, Is.EqualTo(new Variant(7.0)));
|
|
}
|
|
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task ProtectedPrimaryRecordsRejectDifferentKeysAsync(bool valueRecord)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(41));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var source = new NodeId("source", NamespaceIndex);
|
|
var target = new NodeId("target", NamespaceIndex);
|
|
var payload = ByteString.From(new byte[] { 1, 2, 3 });
|
|
if (valueRecord)
|
|
{
|
|
await store.WriteValueAsync(source, new DataValue(Variant.From(true))).ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
await store.UpsertNodeAsync(new StoredNode(source, payload)).ConfigureAwait(false);
|
|
}
|
|
string prefix = valueRecord ? "v/" : "n/";
|
|
(bool stored, ByteString record) = await kv.TryGetAsync(prefix + source).ConfigureAwait(false);
|
|
Assert.That(stored, Is.True);
|
|
await kv.SetAsync(prefix + target, record).ConfigureAwait(false);
|
|
|
|
if (valueRecord)
|
|
{
|
|
(bool found, DataValue value) = await store.TryReadValueAsync(target).ConfigureAwait(false);
|
|
Assert.That(found, Is.False);
|
|
Assert.That(value.IsNull, Is.True);
|
|
(found, value) = await store.TryReadValueAsync(source).ConfigureAwait(false);
|
|
Assert.That(found, Is.True);
|
|
Assert.That(value.WrappedValue, Is.EqualTo(Variant.From(true)));
|
|
}
|
|
else
|
|
{
|
|
Assert.That(await store.TryGetNodeAsync(target).ConfigureAwait(false), Is.Null);
|
|
IStoredNode? original = await store.TryGetNodeAsync(source).ConfigureAwait(false);
|
|
Assert.That(original, Is.Not.Null);
|
|
Assert.That(original!.Payload, Is.EqualTo(payload));
|
|
}
|
|
Assert.That(
|
|
async () => await store.WriteSnapshotAsync().ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode)).EqualTo(StatusCodes.BadDecodingError));
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProtectedPartitionMarkerCannotInitializeAnotherPartitionAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(42));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var partitions = (INodeStatePartitionStore)store;
|
|
const string source = "area/source";
|
|
const string target = "area/target";
|
|
await partitions.MarkPartitionInitializedAsync(source, default).ConfigureAwait(false);
|
|
(bool found, ByteString record) = await kv.TryGetAsync(
|
|
"partition/" + Uri.EscapeDataString(source)).ConfigureAwait(false);
|
|
Assert.That(found, Is.True);
|
|
await kv.SetAsync("partition/" + Uri.EscapeDataString(target), record).ConfigureAwait(false);
|
|
|
|
Assert.That(
|
|
async () => await partitions.IsPartitionInitializedAsync(target, default).ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode)).EqualTo(StatusCodes.BadDecodingError));
|
|
Assert.That(await partitions.IsPartitionInitializedAsync(source, default).ConfigureAwait(false), Is.True);
|
|
}
|
|
|
|
[TestCase("dlog/00000000000000000900")]
|
|
[TestCase("dlog/1")]
|
|
public async Task ProtectedDeltaRejectsDifferentSequenceKeyAsync(string targetKey)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(43));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var nodeId = new NodeId("delta", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(new byte[] { 1 })))
|
|
.ConfigureAwait(false);
|
|
List<NodeStateChange> changes = await ReadChangesAsync(store.ReadDeltaLogAsync(0)).ConfigureAwait(false);
|
|
Assert.That(changes, Has.Count.EqualTo(1));
|
|
Assert.That(changes[0].Sequence, Is.EqualTo(1));
|
|
Assert.That(changes[0].NodeId, Is.EqualTo(nodeId));
|
|
(bool found, ByteString record) = await kv.TryGetAsync("dlog/00000000000000000001")
|
|
.ConfigureAwait(false);
|
|
Assert.That(found, Is.True);
|
|
await kv.SetAsync(targetKey, record).ConfigureAwait(false);
|
|
|
|
Assert.That(
|
|
() => ReadChangesAsync(store.ReadDeltaLogAsync(0)),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode)).EqualTo(StatusCodes.BadDecodingError));
|
|
Assert.That(store.CurrentSequence, Is.EqualTo(1), "A forged key must not advance the local horizon.");
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProtectedSequenceRejectsPrimaryRecordAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(44));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var nodeId = new NodeId("sequence", NamespaceIndex);
|
|
byte[] payload = new byte[sizeof(ulong)];
|
|
BinaryPrimitives.WriteUInt64BigEndian(payload, 1);
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(payload))).ConfigureAwait(false);
|
|
(_, ByteString record) = await kv.TryGetAsync("n/" + nodeId).ConfigureAwait(false);
|
|
await kv.SetAsync(InMemoryNodeStateStore.SequenceKey, record).ConfigureAwait(false);
|
|
|
|
Assert.That(
|
|
async () => await store.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(payload)))
|
|
.ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode)).EqualTo(StatusCodes.BadDecodingError));
|
|
(_, ByteString retained) = await kv.TryGetAsync(InMemoryNodeStateStore.SequenceKey).ConfigureAwait(false);
|
|
Assert.That(retained, Is.EqualTo(record));
|
|
Assert.That(store.CurrentSequence, Is.EqualTo(1));
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProtectedSnapshotChunkRejectsAnotherGenerationAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(45));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var nodeId = new NodeId("snapshot", NamespaceIndex);
|
|
var originalPayload = ByteString.From(new byte[] { 1 });
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, originalPayload)).ConfigureAwait(false);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
NodeStateSnapshot? original = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
Assert.That(original, Is.Not.Null);
|
|
var chunks = new List<KeyValuePair<string, ByteString>>();
|
|
await foreach (KeyValuePair<string, ByteString> entry in kv.ScanAsync("snap/").ConfigureAwait(false))
|
|
{
|
|
chunks.Add(entry);
|
|
}
|
|
Assert.That(chunks, Has.Count.EqualTo(1));
|
|
KeyValuePair<string, ByteString> source = chunks[0];
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(new byte[] { 2 })))
|
|
.ConfigureAwait(false);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
chunks.Clear();
|
|
await foreach (KeyValuePair<string, ByteString> entry in kv.ScanAsync("snap/").ConfigureAwait(false))
|
|
{
|
|
if (entry.Key != source.Key)
|
|
{
|
|
chunks.Add(entry);
|
|
}
|
|
}
|
|
Assert.That(chunks, Has.Count.EqualTo(1));
|
|
await kv.SetAsync(chunks[0].Key, source.Value).ConfigureAwait(false);
|
|
NodeStateSnapshot? current = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
Assert.That(current, Is.Not.Null);
|
|
|
|
Assert.That(
|
|
() => ReadChangesAsync(current!.Entries),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode)).EqualTo(StatusCodes.BadDecodingError));
|
|
List<NodeStateChange> oldEntries = await ReadChangesAsync(original!.Entries).ConfigureAwait(false);
|
|
Assert.That(oldEntries, Has.Count.EqualTo(1));
|
|
Assert.That(oldEntries[0].Node!.Payload, Is.EqualTo(originalPayload));
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProtectedSnapshotChunkRejectsDifferentOrdinalAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(46));
|
|
var context = ServiceMessageContext.CreateEmpty(NUnitTelemetryContext.Create());
|
|
using var store = new InMemoryNodeStateStore(kv, context, protector);
|
|
var payload = ByteString.From(new byte[context.MaxByteStringLength]);
|
|
await store.UpsertNodeAsync(new StoredNode(new NodeId("first", 1), payload)).ConfigureAwait(false);
|
|
await store.UpsertNodeAsync(new StoredNode(new NodeId("second", 1), payload)).ConfigureAwait(false);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
var chunks = new List<KeyValuePair<string, ByteString>>();
|
|
await foreach (KeyValuePair<string, ByteString> entry in kv.ScanAsync("snap/").ConfigureAwait(false))
|
|
{
|
|
chunks.Add(entry);
|
|
}
|
|
Assert.That(chunks, Has.Count.EqualTo(2));
|
|
chunks.Sort(static (left, right) => StringComparer.Ordinal.Compare(left.Key, right.Key));
|
|
await kv.SetAsync(chunks[1].Key, chunks[0].Value).ConfigureAwait(false);
|
|
NodeStateSnapshot? snapshot = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
|
|
Assert.That(
|
|
() => ReadChangesAsync(snapshot!.Entries),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode)).EqualTo(StatusCodes.BadDecodingError));
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProtectedSnapshotManifestRejectsEmptyContextAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(47));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
(bool found, ByteString record) = await kv.TryGetAsync("snapmeta/manifest").ConfigureAwait(false);
|
|
Assert.That(found, Is.True);
|
|
Assert.That(protector.TryUnprotect(
|
|
RecordProtectionContext.Create("node-state-manifest", "snapmeta/manifest"),
|
|
record, out ByteString manifest), Is.True);
|
|
await kv.SetAsync("snapmeta/manifest", protector.Protect(default, manifest))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(await store.TryReadSnapshotAsync().ConfigureAwait(false), Is.Null);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EnumerateValuesOnEmptyStoreYieldsNothingAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
|
|
int count = 0;
|
|
await foreach ((NodeId _, DataValue _) in store.EnumerateValuesAsync())
|
|
{
|
|
count++;
|
|
}
|
|
|
|
Assert.That(count, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task WriteAndReadSnapshotRoundTripsEntriesAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var a = new NodeId("a", NamespaceIndex);
|
|
var b = new NodeId("b", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(a, ByteString.From(new byte[] { 1, 2 }))).ConfigureAwait(false);
|
|
await store.UpsertNodeAsync(new StoredNode(b, ByteString.From(new byte[] { 3 }))).ConfigureAwait(false);
|
|
await store.WriteValueAsync(a, new DataValue(new Variant(5.0), StatusCodes.Good, DateTimeUtc.Now)).ConfigureAwait(false);
|
|
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
NodeStateSnapshot? snapshot = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
var upserts = new HashSet<NodeId>();
|
|
var values = new Dictionary<NodeId, DataValue>();
|
|
await foreach (NodeStateChange entry in snapshot!.Entries)
|
|
{
|
|
if (entry.Kind == NodeStateChangeKind.Upsert)
|
|
{
|
|
upserts.Add(entry.NodeId);
|
|
}
|
|
else if (entry.Kind == NodeStateChangeKind.Value)
|
|
{
|
|
values[entry.NodeId] = entry.Value;
|
|
}
|
|
}
|
|
|
|
Assert.That(upserts, Is.EquivalentTo([a, b]));
|
|
Assert.That(values.Keys, Is.EquivalentTo([a]));
|
|
Assert.That(values[a].WrappedValue, Is.EqualTo(new Variant(5.0)));
|
|
Assert.That(snapshot.Sequence, Is.GreaterThanOrEqualTo(3));
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryReadSnapshotReturnsNullWhenNonePublishedAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
|
|
NodeStateSnapshot? snapshot = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(snapshot, Is.Null);
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryReadSnapshotRejectsInvalidGuidLengthsAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
using var encoder = new BinaryEncoder(m_messageContext);
|
|
encoder.WriteByteString(null, new ByteString(new byte[15]));
|
|
encoder.WriteInt32(null, 0);
|
|
encoder.WriteUInt64(null, 0);
|
|
encoder.WriteByteString(null, new ByteString(new byte[16]));
|
|
byte[]? manifest = encoder.CloseAndReturnBuffer();
|
|
Assert.That(manifest, Is.Not.Null);
|
|
await kv.SetAsync(
|
|
"snapmeta/manifest",
|
|
new ByteString(manifest!)).ConfigureAwait(false);
|
|
|
|
NodeStateSnapshot? snapshot = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(snapshot, Is.Null);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DeltaLogReplaysOnlyChangesAfterSequenceAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var a = new NodeId("a", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(a, ByteString.From(new byte[] { 1 }))).ConfigureAwait(false);
|
|
ulong afterUpsert = store.CurrentSequence;
|
|
await store.WriteValueAsync(a, new DataValue(new Variant(9.0), StatusCodes.Good, DateTimeUtc.Now)).ConfigureAwait(false);
|
|
|
|
var replayed = new List<NodeStateChange>();
|
|
await foreach (NodeStateChange change in store.ReadDeltaLogAsync(afterUpsert))
|
|
{
|
|
replayed.Add(change);
|
|
}
|
|
|
|
Assert.That(replayed, Has.Count.EqualTo(1));
|
|
Assert.That(replayed[0].Kind, Is.EqualTo(NodeStateChangeKind.Value));
|
|
Assert.That(replayed[0].NodeId, Is.EqualTo(a));
|
|
Assert.That(replayed[0].Sequence, Is.GreaterThan(afterUpsert));
|
|
}
|
|
|
|
[Test]
|
|
public async Task WriteSnapshotTrimsDeltaLogAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var a = new NodeId("a", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(a, ByteString.From(new byte[] { 1 }))).ConfigureAwait(false);
|
|
await store.WriteValueAsync(a, new DataValue(new Variant(2.0), StatusCodes.Good, DateTimeUtc.Now)).ConfigureAwait(false);
|
|
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
|
|
var replayed = new List<NodeStateChange>();
|
|
await foreach (NodeStateChange change in store.ReadDeltaLogAsync(0))
|
|
{
|
|
replayed.Add(change);
|
|
}
|
|
|
|
Assert.That(replayed, Is.Empty);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Prevents another store from snapshotting an unfinished write and prevents a delayed
|
|
/// older primary CAS from overwriting a newer completed row.
|
|
/// </summary>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task SnapshotRejectsUnfinishedPublicationAcrossStoreInstancesAsync(bool beforePrimary)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
var nodeId = new NodeId("overlap", NamespaceIndex);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
Mock<ISharedKeyValueStore> controlled = CreateStoreMock(kv);
|
|
if (beforePrimary)
|
|
{
|
|
controlled.Setup(s => s.CompareAndSwapAsync(
|
|
"n/" + nodeId,
|
|
It.IsAny<ByteString>(), It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns(WritePrimaryAsync);
|
|
}
|
|
else
|
|
{
|
|
controlled.Setup(s => s.SetAsync(
|
|
"dlog/00000000000000000001", It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns(WriteDeltaAsync);
|
|
}
|
|
|
|
async ValueTask<bool> WritePrimaryAsync(
|
|
string key,
|
|
ByteString expected,
|
|
ByteString value,
|
|
CancellationToken ct)
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
return await kv.CompareAndSwapAsync(key, expected, value, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
async ValueTask WriteDeltaAsync(string key, ByteString value, CancellationToken ct)
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
await kv.SetAsync(key, value, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
using var earlier = new InMemoryNodeStateStore(controlled.Object, m_messageContext);
|
|
using var later = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
using var snapshotWriter = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
Task publication = earlier.UpsertNodeAsync(
|
|
new StoredNode(nodeId, new ByteString(new byte[] { 1 })), cts.Token).AsTask();
|
|
try
|
|
{
|
|
await entered.Task.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
await later.UpsertNodeAsync(
|
|
new StoredNode(nodeId, new ByteString(new byte[] { 2 })), cts.Token).ConfigureAwait(false);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await snapshotWriter.WriteSnapshotAsync(cts.Token).ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("unfinished")).ConfigureAwait(false);
|
|
Assert.That(await snapshotWriter.TryReadSnapshotAsync(cts.Token).ConfigureAwait(false), Is.Null);
|
|
List<NodeStateChange> retained = await ReadChangesAsync(snapshotWriter.ReadDeltaLogAsync(0, cts.Token))
|
|
.ConfigureAwait(false);
|
|
Assert.That(retained, Has.Count.EqualTo(1));
|
|
Assert.That(retained[0].Sequence, Is.EqualTo(2));
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await publication.ConfigureAwait(false);
|
|
}
|
|
|
|
IStoredNode? newest = await snapshotWriter.TryGetNodeAsync(nodeId, cts.Token).ConfigureAwait(false);
|
|
Assert.That(newest, Is.Not.Null);
|
|
Assert.That(newest!.Payload, Is.EqualTo(new ByteString(new byte[] { 2 })));
|
|
await snapshotWriter.WriteSnapshotAsync(cts.Token).ConfigureAwait(false);
|
|
NodeStateSnapshot? snapshot = await snapshotWriter.TryReadSnapshotAsync(cts.Token).ConfigureAwait(false);
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
Assert.That(snapshot!.Sequence, Is.EqualTo(2));
|
|
Assert.That(await CountKeysAsync(kv, "dlog/").ConfigureAwait(false), Is.Zero);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Keeps failed or cancelled reservations in shared state so restart cannot trim later completed deltas.
|
|
/// </summary>
|
|
/// <exception cref="IOException">The test backend deliberately rejects a delta publication.</exception>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task UnfinishedPublicationSurvivesFailureAndRestartAsync(bool cancel)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var blocked = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
Mock<ISharedKeyValueStore> controlled = CreateStoreMock(kv);
|
|
controlled.Setup(s => s.SetAsync(
|
|
"dlog/00000000000000000001", It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns(FailPublicationAsync);
|
|
|
|
async ValueTask FailPublicationAsync(string key, ByteString value, CancellationToken ct)
|
|
{
|
|
entered.TrySetResult(true);
|
|
if (cancel)
|
|
{
|
|
await blocked.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
}
|
|
throw new IOException($"The delta '{key}' ({value.Length} bytes) could not be published.");
|
|
}
|
|
|
|
using (var writer = new InMemoryNodeStateStore(controlled.Object, m_messageContext))
|
|
{
|
|
Task publication = writer.UpsertNodeAsync(new StoredNode(
|
|
new NodeId("unfinished", NamespaceIndex),
|
|
new ByteString(new byte[] { 1 })), cts.Token).AsTask();
|
|
await entered.Task.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
if (cancel)
|
|
{
|
|
cts.Cancel();
|
|
await Assert.ThatAsync(
|
|
async () => await publication.ConfigureAwait(false),
|
|
Throws.InstanceOf<OperationCanceledException>()).ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
await Assert.ThatAsync(
|
|
async () => await publication.ConfigureAwait(false),
|
|
Throws.TypeOf<IOException>()).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
using var restarted = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
await restarted.UpsertNodeAsync(new StoredNode(
|
|
new NodeId("completed", NamespaceIndex),
|
|
new ByteString(new byte[] { 2 }))).ConfigureAwait(false);
|
|
await Assert.ThatAsync(
|
|
async () => await restarted.WriteSnapshotAsync().ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("publication 1 is unfinished"))
|
|
.ConfigureAwait(false);
|
|
|
|
List<NodeStateChange> retained = await ReadChangesAsync(restarted.ReadDeltaLogAsync(0))
|
|
.ConfigureAwait(false);
|
|
Assert.That(retained, Has.Count.EqualTo(1));
|
|
Assert.That(retained[0].Sequence, Is.EqualTo(2));
|
|
(bool found, ByteString coordinator) = await kv.TryGetAsync(InMemoryNodeStateStore.SequenceKey)
|
|
.ConfigureAwait(false);
|
|
Assert.That(found, Is.True);
|
|
Assert.That(coordinator.Length, Is.EqualTo(2 * sizeof(ulong)));
|
|
Assert.That(BinaryPrimitives.ReadUInt64BigEndian(coordinator.Span), Is.EqualTo(2));
|
|
Assert.That(BinaryPrimitives.ReadUInt64BigEndian(coordinator.Span[sizeof(ulong)..]), Is.EqualTo(1));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Revalidates shared coordination after both scans, preserving the previous manifest and all newer deltas.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task SnapshotRejectsConcurrentPublicationDuringScansAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var writer = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
var nodeId = new NodeId("baseline", NamespaceIndex);
|
|
await writer.UpsertNodeAsync(new StoredNode(nodeId, new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false);
|
|
await writer.WriteSnapshotAsync().ConfigureAwait(false);
|
|
(_, ByteString manifest) = await kv.TryGetAsync("snapmeta/manifest").ConfigureAwait(false);
|
|
await writer.UpsertNodeAsync(new StoredNode(nodeId, new ByteString(new byte[] { 2 })))
|
|
.ConfigureAwait(false);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
Mock<ISharedKeyValueStore> controlled = CreateStoreMock(kv);
|
|
controlled.Setup(s => s.ScanAsync("v/", It.IsAny<CancellationToken>()))
|
|
.Returns((string prefix, CancellationToken ct) => ScanValuesAsync(prefix, ct));
|
|
|
|
async IAsyncEnumerable<KeyValuePair<string, ByteString>> ScanValuesAsync(
|
|
string prefix,
|
|
[EnumeratorCancellation] CancellationToken ct)
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
await foreach (KeyValuePair<string, ByteString> entry in kv.ScanAsync(prefix, ct)
|
|
.ConfigureAwait(false))
|
|
{
|
|
yield return entry;
|
|
}
|
|
}
|
|
|
|
using var snapshotWriter = new InMemoryNodeStateStore(controlled.Object, m_messageContext);
|
|
Task snapshot = snapshotWriter.WriteSnapshotAsync(cts.Token).AsTask();
|
|
try
|
|
{
|
|
await entered.Task.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
await writer.UpsertNodeAsync(new StoredNode(
|
|
new NodeId("concurrent", NamespaceIndex),
|
|
new ByteString(new byte[] { 3 })), cts.Token).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
}
|
|
await Assert.ThatAsync(
|
|
async () => await snapshot.ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("changed")).ConfigureAwait(false);
|
|
|
|
(_, ByteString unchanged) = await kv.TryGetAsync("snapmeta/manifest").ConfigureAwait(false);
|
|
Assert.That(unchanged, Is.EqualTo(manifest));
|
|
Assert.That(await CountSnapshotGenerationsAsync(kv).ConfigureAwait(false), Is.EqualTo(1));
|
|
List<NodeStateChange> retained = await ReadChangesAsync(writer.ReadDeltaLogAsync(0)).ConfigureAwait(false);
|
|
Assert.That(retained, Has.Count.EqualTo(2));
|
|
Assert.That(retained[0].Sequence, Is.EqualTo(2));
|
|
Assert.That(retained[1].Sequence, Is.EqualTo(3));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Retains chunks when an indeterminate manifest CAS actually committed before its caller observed failure.
|
|
/// </summary>
|
|
/// <exception cref="OperationCanceledException">The backend injects cancellation after manifest commit.</exception>
|
|
/// <exception cref="IOException">The backend injects a lost response after manifest commit.</exception>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task IndeterminateManifestPublicationRetainsPossiblyPublishedChunksAsync(bool cancel)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var writer = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
var nodeId = new NodeId("manifest", NamespaceIndex);
|
|
await writer.UpsertNodeAsync(new StoredNode(nodeId, new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false);
|
|
await writer.WriteSnapshotAsync().ConfigureAwait(false);
|
|
await writer.UpsertNodeAsync(new StoredNode(nodeId, new ByteString(new byte[] { 2 })))
|
|
.ConfigureAwait(false);
|
|
Mock<ISharedKeyValueStore> controlled = CreateStoreMock(kv);
|
|
controlled.Setup(s => s.CompareAndSwapAsync(
|
|
"snapmeta/manifest",
|
|
It.IsAny<ByteString>(), It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns(PublishThenFailAsync);
|
|
|
|
async ValueTask<bool> PublishThenFailAsync(
|
|
string key,
|
|
ByteString expected,
|
|
ByteString value,
|
|
CancellationToken ct)
|
|
{
|
|
bool published = await kv.CompareAndSwapAsync(key, expected, value, ct).ConfigureAwait(false);
|
|
Assert.That(published, Is.True);
|
|
if (cancel)
|
|
{
|
|
cts.Cancel();
|
|
throw new OperationCanceledException(ct);
|
|
}
|
|
throw new IOException("The committed manifest response was lost.");
|
|
}
|
|
|
|
using var snapshotWriter = new InMemoryNodeStateStore(controlled.Object, m_messageContext);
|
|
Task publication = snapshotWriter.WriteSnapshotAsync(cts.Token).AsTask();
|
|
if (cancel)
|
|
{
|
|
await Assert.ThatAsync(
|
|
async () => await publication.ConfigureAwait(false),
|
|
Throws.InstanceOf<OperationCanceledException>()).ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
await Assert.ThatAsync(
|
|
async () => await publication.ConfigureAwait(false),
|
|
Throws.TypeOf<IOException>()).ConfigureAwait(false);
|
|
}
|
|
|
|
NodeStateSnapshot? snapshot = await writer.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
Assert.That(snapshot!.Sequence, Is.EqualTo(2));
|
|
List<NodeStateChange> entries = await ReadChangesAsync(snapshot.Entries).ConfigureAwait(false);
|
|
Assert.That(entries, Has.Count.EqualTo(1));
|
|
Assert.That(entries[0].Node!.Payload, Is.EqualTo(new ByteString(new byte[] { 2 })));
|
|
Assert.That(await CountSnapshotGenerationsAsync(kv).ConfigureAwait(false), Is.EqualTo(2));
|
|
Assert.That(await CountKeysAsync(kv, "dlog/").ConfigureAwait(false), Is.EqualTo(1));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Removes only the losing attempt's chunks when another instance publishes a snapshot first.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task ConcurrentSnapshotLoserCleansOnlyItsOwnGenerationAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var winner = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
await winner.UpsertNodeAsync(new StoredNode(
|
|
new NodeId("snapshot", NamespaceIndex),
|
|
new ByteString(new byte[] { 1 }))).ConfigureAwait(false);
|
|
await winner.WriteSnapshotAsync().ConfigureAwait(false);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
Mock<ISharedKeyValueStore> controlled = CreateStoreMock(kv);
|
|
controlled.Setup(s => s.CompareAndSwapAsync(
|
|
"snapmeta/manifest",
|
|
It.IsAny<ByteString>(), It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns(PublishLastAsync);
|
|
|
|
async ValueTask<bool> PublishLastAsync(
|
|
string key,
|
|
ByteString expected,
|
|
ByteString value,
|
|
CancellationToken ct)
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
return await kv.CompareAndSwapAsync(key, expected, value, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
using var loser = new InMemoryNodeStateStore(controlled.Object, m_messageContext);
|
|
Task publication = loser.WriteSnapshotAsync(cts.Token).AsTask();
|
|
ByteString manifest;
|
|
try
|
|
{
|
|
await entered.Task.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
await winner.WriteSnapshotAsync(cts.Token).ConfigureAwait(false);
|
|
(_, manifest) = await kv.TryGetAsync("snapmeta/manifest", cts.Token).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
}
|
|
await Assert.ThatAsync(
|
|
async () => await publication.ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("concurrently")).ConfigureAwait(false);
|
|
|
|
(_, ByteString unchanged) = await kv.TryGetAsync("snapmeta/manifest").ConfigureAwait(false);
|
|
Assert.That(unchanged, Is.EqualTo(manifest));
|
|
Assert.That(await CountSnapshotGenerationsAsync(kv).ConfigureAwait(false), Is.EqualTo(2));
|
|
NodeStateSnapshot? snapshot = await winner.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
Assert.That(await ReadChangesAsync(snapshot!.Entries).ConfigureAwait(false), Has.Count.EqualTo(1));
|
|
}
|
|
|
|
[Test]
|
|
public void ObserveSequenceRaisesHighWaterMarkOnly()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
|
|
store.ObserveSequence(50);
|
|
Assert.That(store.CurrentSequence, Is.EqualTo(50));
|
|
store.ObserveSequence(10);
|
|
Assert.That(store.CurrentSequence, Is.EqualTo(50));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Preserves the unsigned high-water mark across the signed interlocked storage boundary.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task ObserveSequencePreservesUnsignedHighWaterMarkAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
const ulong highWaterMark = (ulong)long.MaxValue + 2;
|
|
|
|
store.ObserveSequence(highWaterMark);
|
|
store.ObserveSequence(1);
|
|
await store.UpsertNodeAsync(
|
|
new StoredNode(new NodeId("unsigned", NamespaceIndex), new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(store.CurrentSequence, Is.EqualTo(highWaterMark + 1));
|
|
List<NodeStateChange> deltas = await ReadChangesAsync(store.ReadDeltaLogAsync(0)).ConfigureAwait(false);
|
|
Assert.That(deltas, Has.Count.EqualTo(1));
|
|
Assert.That(deltas[0].Sequence, Is.EqualTo(highWaterMark + 1));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Rejects exhausted shared or observed sequence numbers without wrapping or writing a node.
|
|
/// </summary>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task SequenceExhaustionRejectsBeforePrimaryMutationAsync(bool shared)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
if (shared)
|
|
{
|
|
await kv.SetAsync(
|
|
InMemoryNodeStateStore.SequenceKey,
|
|
MakeRecord(ulong.MaxValue, ByteString.Empty)).ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
store.ObserveSequence(ulong.MaxValue);
|
|
}
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await store.UpsertNodeAsync(
|
|
new StoredNode(new NodeId("exhausted", NamespaceIndex), new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadOutOfRange)).ConfigureAwait(false);
|
|
|
|
Assert.That(await CountKeysAsync(kv, "n/").ConfigureAwait(false), Is.Zero);
|
|
Assert.That(await CountKeysAsync(kv, "dlog/").ConfigureAwait(false), Is.Zero);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Assigns unique shared sequences across concurrent stores and a newly constructed writer.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task SharedCoordinatorAssignsUniqueSequencesAcrossStoreInstancesAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
var contendersReady = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
int contenders = 0;
|
|
Mock<ISharedKeyValueStore> controlled = CreateStoreMock(kv);
|
|
controlled.Setup(s => s.CompareAndSwapAsync(
|
|
InMemoryNodeStateStore.SequenceKey,
|
|
It.IsAny<ByteString>(), It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns(ReserveConcurrentlyAsync);
|
|
|
|
async ValueTask<bool> ReserveConcurrentlyAsync(
|
|
string key,
|
|
ByteString expected,
|
|
ByteString value,
|
|
CancellationToken ct)
|
|
{
|
|
int contender = Interlocked.Increment(ref contenders);
|
|
if (contender <= 32)
|
|
{
|
|
if (contender == 32)
|
|
{
|
|
contendersReady.TrySetResult(true);
|
|
}
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
}
|
|
return await kv.CompareAndSwapAsync(key, expected, value, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
using var first = new InMemoryNodeStateStore(controlled.Object, m_messageContext);
|
|
using var second = new InMemoryNodeStateStore(controlled.Object, m_messageContext);
|
|
var writes = new List<Task>();
|
|
for (int i = 0; i < 32; i++)
|
|
{
|
|
InMemoryNodeStateStore writer = i % 2 == 0 ? first : second;
|
|
writes.Add(writer.UpsertNodeAsync(new StoredNode(
|
|
new NodeId(i.ToString(CultureInfo.InvariantCulture), NamespaceIndex),
|
|
new ByteString(new byte[] { 1 })), cts.Token).AsTask());
|
|
}
|
|
try
|
|
{
|
|
await contendersReady.Task.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
}
|
|
await Task.WhenAll(writes).ConfigureAwait(false);
|
|
|
|
using var restarted = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
await restarted.UpsertNodeAsync(new StoredNode(
|
|
new NodeId("restarted", NamespaceIndex),
|
|
new ByteString(new byte[] { 2 }))).ConfigureAwait(false);
|
|
List<NodeStateChange> deltas = await ReadChangesAsync(restarted.ReadDeltaLogAsync(0))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(deltas, Has.Count.EqualTo(33));
|
|
for (int i = 0; i < deltas.Count; i++)
|
|
{
|
|
Assert.That(deltas[i].Sequence, Is.EqualTo((ulong)i + 1));
|
|
}
|
|
Assert.That(restarted.CurrentSequence, Is.EqualTo(33));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Leaves the coordinator and payload untouched when cancellation precedes sequence reservation.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task CancelledWriterDoesNotReserveASequenceAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
using var cts = new CancellationTokenSource();
|
|
cts.Cancel();
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await store.UpsertNodeAsync(
|
|
new StoredNode(new NodeId("cancelled", NamespaceIndex), new ByteString(new byte[] { 1 })),
|
|
cts.Token).ConfigureAwait(false),
|
|
Throws.InstanceOf<OperationCanceledException>()).ConfigureAwait(false);
|
|
|
|
Assert.That(await CountKeysAsync(kv, string.Empty).ConfigureAwait(false), Is.Zero);
|
|
Assert.That(store.CurrentSequence, Is.Zero);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Fails a bare CRDT writer before any coordinator, primary, or delta mutation.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task BareCrdtWriterRejectsBeforeAnyMutationAsync()
|
|
{
|
|
await using var network = new InMemoryNetwork();
|
|
await using var crdt = new ReplicatedSharedKeyValueStore(
|
|
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
|
|
using var store = new InMemoryNodeStateStore(crdt, m_messageContext);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await store.UpsertNodeAsync(
|
|
new StoredNode(new NodeId("unsupported", NamespaceIndex), new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains(InMemoryNodeStateStore.SequenceKey))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(await CountKeysAsync(crdt, string.Empty).ConfigureAwait(false), Is.Zero);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Does not mistake an undisclosed store's CAS method for a declared linearizable coordinator.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task UndeclaredCoordinatorRejectsBeforeCallingTheBackendAsync()
|
|
{
|
|
var kv = new Mock<ISharedKeyValueStore>(MockBehavior.Strict);
|
|
using var store = new InMemoryNodeStateStore(kv.Object, m_messageContext);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await store.UpsertNodeAsync(
|
|
new StoredNode(new NodeId("unsupported", NamespaceIndex), new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("declared linearizable"))
|
|
.ConfigureAwait(false);
|
|
|
|
kv.VerifyNoOtherCalls();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Rejects a private in-memory counter for replicated payloads while retaining ordinary local construction.
|
|
/// </summary>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task ReplicatedPayloadRejectsProcessLocalCoordinatorAsync(bool onlyDeltaReplicated)
|
|
{
|
|
await using var network = new InMemoryNetwork();
|
|
await using var crdt = new ReplicatedSharedKeyValueStore(
|
|
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
|
|
using var coordinator = new InMemorySharedKeyValueStore();
|
|
ArrayOf<string> prefixes = onlyDeltaReplicated ? ["election/", "n/", "v/"] : default;
|
|
await using var hybrid = new HybridSharedKeyValueStore(crdt, coordinator, prefixes);
|
|
using var store = new InMemoryNodeStateStore(hybrid, m_messageContext);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await store.UpsertNodeAsync(
|
|
new StoredNode(new NodeId("unsupported", NamespaceIndex), new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("process-local")).ConfigureAwait(false);
|
|
|
|
Assert.That(await CountKeysAsync(crdt, string.Empty).ConfigureAwait(false), Is.Zero);
|
|
Assert.That(await CountKeysAsync(coordinator, string.Empty).ConfigureAwait(false), Is.Zero);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Preserves a corrupt coordinator record and rejects publication before mutating primary state.
|
|
/// </summary>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task InvalidCoordinatorRejectsBeforePrimaryMutationAsync(bool invalidReservation)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
ByteString record = invalidReservation
|
|
? MakeRecord(1, MakeRecord(2, ByteString.Empty))
|
|
: new ByteString(new byte[] { 1, 2, 3 });
|
|
await kv.SetAsync(InMemoryNodeStateStore.SequenceKey, record).ConfigureAwait(false);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await store.UpsertNodeAsync(
|
|
new StoredNode(new NodeId("invalid", NamespaceIndex), new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
|
|
Assert.That(await CountKeysAsync(kv, "n/").ConfigureAwait(false), Is.Zero);
|
|
(bool found, ByteString unchanged) = await kv.TryGetAsync(InMemoryNodeStateStore.SequenceKey)
|
|
.ConfigureAwait(false);
|
|
Assert.That(found, Is.True);
|
|
Assert.That(unchanged, Is.EqualTo(record));
|
|
}
|
|
|
|
[Test]
|
|
public async Task SubscribeObservesTopologyAndValueChangesAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
using var cts = new CancellationTokenSource();
|
|
var nodeId = new NodeId("watched", NamespaceIndex);
|
|
|
|
await using IAsyncEnumerator<NodeStateChange> changes =
|
|
store.SubscribeChangesAsync(cts.Token).GetAsyncEnumerator();
|
|
|
|
ValueTask<bool> upsert = changes.MoveNextAsync();
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(new byte[] { 7 }))).ConfigureAwait(false);
|
|
Assert.That(await upsert.ConfigureAwait(false), Is.True);
|
|
Assert.That(changes.Current.Kind, Is.EqualTo(NodeStateChangeKind.Upsert));
|
|
Assert.That(changes.Current.NodeId, Is.EqualTo(nodeId));
|
|
Assert.That(changes.Current.Node, Is.Not.Null);
|
|
|
|
ValueTask<bool> valueChange = changes.MoveNextAsync();
|
|
await store.WriteValueAsync(nodeId, new DataValue(new Variant(1.0), StatusCodes.Good, DateTimeUtc.Now)).ConfigureAwait(false);
|
|
Assert.That(await valueChange.ConfigureAwait(false), Is.True);
|
|
Assert.That(changes.Current.Kind, Is.EqualTo(NodeStateChangeKind.Value));
|
|
Assert.That(changes.Current.NodeId, Is.EqualTo(nodeId));
|
|
|
|
ValueTask<bool> delete = changes.MoveNextAsync();
|
|
await store.DeleteNodeAsync(nodeId).ConfigureAwait(false);
|
|
Assert.That(await delete.ConfigureAwait(false), Is.True);
|
|
Assert.That(changes.Current.Kind, Is.EqualTo(NodeStateChangeKind.Delete));
|
|
Assert.That(changes.Current.NodeId, Is.EqualTo(nodeId));
|
|
|
|
cts.Cancel();
|
|
}
|
|
|
|
[Test]
|
|
public async Task SubscribePollsWhenStoreHasNoWatchAsync()
|
|
{
|
|
// A CRDT store has no change-feed (WatchAsync throws NotSupported),
|
|
// so the standby subscription must fall back to scan-polling instead
|
|
// of silently stopping.
|
|
await using var network = new InMemoryNetwork();
|
|
await using var crdt = new ReplicatedSharedKeyValueStore(
|
|
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
|
|
await using var coordinator = new RaftSharedKeyValueStore(
|
|
DefaultRaftConsensus.CreateSingleNode(), ownsConsensus: true);
|
|
await using var hybrid = new HybridSharedKeyValueStore(crdt, coordinator);
|
|
var baseline = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
Mock<ISharedKeyValueStore> observed = CreateStoreMock(hybrid);
|
|
observed.Setup(s => s.ScanAsync("dlog/", It.IsAny<CancellationToken>()))
|
|
.Returns((string prefix, CancellationToken ct) => ObserveBaselineAsync(prefix, ct));
|
|
|
|
async IAsyncEnumerable<KeyValuePair<string, ByteString>> ObserveBaselineAsync(
|
|
string prefix,
|
|
[EnumeratorCancellation] CancellationToken ct)
|
|
{
|
|
await foreach (KeyValuePair<string, ByteString> entry in hybrid.ScanAsync(prefix, ct)
|
|
.ConfigureAwait(false))
|
|
{
|
|
yield return entry;
|
|
}
|
|
baseline.TrySetResult(true);
|
|
}
|
|
|
|
using var store = new InMemoryNodeStateStore(
|
|
observed.Object, m_messageContext, null, TimeSpan.FromMilliseconds(50));
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
|
|
var nodeId = new NodeId("polled", NamespaceIndex);
|
|
|
|
await using IAsyncEnumerator<NodeStateChange> changes =
|
|
store.SubscribeChangesAsync(cts.Token).GetAsyncEnumerator();
|
|
|
|
Task<bool> pending = changes.MoveNextAsync().AsTask();
|
|
try
|
|
{
|
|
await baseline.Task.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
await store.UpsertNodeAsync(
|
|
new StoredNode(nodeId, ByteString.From(new byte[] { 9 })), cts.Token).ConfigureAwait(false);
|
|
|
|
Assert.That(await pending.ConfigureAwait(false), Is.True);
|
|
Assert.That(changes.Current.Kind, Is.EqualTo(NodeStateChangeKind.Upsert));
|
|
Assert.That(changes.Current.NodeId, Is.EqualTo(nodeId));
|
|
|
|
pending = changes.MoveNextAsync().AsTask();
|
|
await store.DeleteNodeAsync(nodeId, cts.Token).ConfigureAwait(false);
|
|
await Assert.ThatAsync(
|
|
async () => await store.WriteSnapshotAsync(cts.Token).ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("eventually consistent"))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(await pending.ConfigureAwait(false), Is.True);
|
|
Assert.That(changes.Current.Kind, Is.EqualTo(NodeStateChangeKind.Delete));
|
|
Assert.That(changes.Current.NodeId, Is.EqualTo(nodeId));
|
|
}
|
|
finally
|
|
{
|
|
cts.Cancel();
|
|
try
|
|
{
|
|
await pending.ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (cts.IsCancellationRequested)
|
|
{
|
|
// Drain an interrupted MoveNext before disposing the async iterator.
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task PreparedPollingSubscriptionEmitsCurrentStateEachTimeAsync()
|
|
{
|
|
await using var network = new InMemoryNetwork();
|
|
await using var crdt = new ReplicatedSharedKeyValueStore(
|
|
ReplicaId.New(),
|
|
network.CreateTransport(),
|
|
TimeProvider.System,
|
|
CrdtReaderOptions.Default);
|
|
await using var coordinator = new RaftSharedKeyValueStore(
|
|
DefaultRaftConsensus.CreateSingleNode(), ownsConsensus: true);
|
|
await using var hybrid = new HybridSharedKeyValueStore(crdt, coordinator);
|
|
using var store = new InMemoryNodeStateStore(
|
|
hybrid,
|
|
m_messageContext,
|
|
null,
|
|
TimeSpan.FromMilliseconds(25));
|
|
var nodeId = new NodeId("prepared", NamespaceIndex);
|
|
await store.UpsertNodeAsync(
|
|
new StoredNode(nodeId, ByteString.From(new byte[] { 9 }))).ConfigureAwait(false);
|
|
|
|
for (int attempt = 0; attempt < 2; attempt++)
|
|
{
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
((INodeStateSubscriptionPreparer)store).PrepareSubscription();
|
|
await using IAsyncEnumerator<NodeStateChange> changes =
|
|
store.SubscribeChangesAsync(cts.Token).GetAsyncEnumerator();
|
|
|
|
Assert.That(await changes.MoveNextAsync().ConfigureAwait(false), Is.True);
|
|
Assert.That(changes.Current.Kind, Is.EqualTo(NodeStateChangeKind.Upsert));
|
|
Assert.That(changes.Current.NodeId, Is.EqualTo(nodeId));
|
|
cts.Cancel();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Replays a retained higher-sequence delta when a delayed earlier CRDT write wins the raw row.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task HybridPollingRecoversNewerDeltaAfterDelayedPrimaryWriteAsync()
|
|
{
|
|
await using var network = new InMemoryNetwork();
|
|
await using var crdt = new ReplicatedSharedKeyValueStore(
|
|
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
|
|
await using var coordinator = new RaftSharedKeyValueStore(
|
|
DefaultRaftConsensus.CreateSingleNode(), ownsConsensus: true);
|
|
await using var hybrid = new HybridSharedKeyValueStore(crdt, coordinator);
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
|
|
var nodeId = new NodeId("delayed", NamespaceIndex);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
Mock<ISharedKeyValueStore> controlled = CreateStoreMock(hybrid);
|
|
controlled.Setup(s => s.SetAsync("n/" + nodeId, It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns(DelayPrimaryAsync);
|
|
|
|
async ValueTask DelayPrimaryAsync(string key, ByteString value, CancellationToken ct)
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
await hybrid.SetAsync(key, value, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
using var earlier = new InMemoryNodeStateStore(controlled.Object, m_messageContext);
|
|
using var later = new InMemoryNodeStateStore(hybrid, m_messageContext);
|
|
Task delayed = earlier.UpsertNodeAsync(
|
|
new StoredNode(nodeId, new ByteString(new byte[] { 1 })), cts.Token).AsTask();
|
|
try
|
|
{
|
|
await entered.Task.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
await later.UpsertNodeAsync(
|
|
new StoredNode(nodeId, new ByteString(new byte[] { 2 })), cts.Token).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await delayed.ConfigureAwait(false);
|
|
}
|
|
|
|
IStoredNode? raw = await later.TryGetNodeAsync(nodeId, cts.Token).ConfigureAwait(false);
|
|
Assert.That(raw, Is.Not.Null);
|
|
Assert.That(raw!.Payload, Is.EqualTo(new ByteString(new byte[] { 1 })));
|
|
Assert.That(((INodeStateStoreReadConsistency)later).HasAuthoritativeReads, Is.False);
|
|
Assert.That(((INodeStateStoreReadConsistency)later).SupportsSnapshots, Is.False);
|
|
await Assert.ThatAsync(
|
|
async () => await later.WriteSnapshotAsync(cts.Token).ConfigureAwait(false),
|
|
Throws.InvalidOperationException.With.Message.Contains("must be retained")).ConfigureAwait(false);
|
|
List<NodeStateChange> deltas = await ReadChangesAsync(later.ReadDeltaLogAsync(0, cts.Token))
|
|
.ConfigureAwait(false);
|
|
Assert.That(deltas, Has.Count.EqualTo(2));
|
|
Assert.That(deltas[1].Sequence, Is.EqualTo(2));
|
|
|
|
((INodeStateSubscriptionPreparer)later).PrepareSubscription();
|
|
await using IAsyncEnumerator<NodeStateChange> changes =
|
|
later.SubscribeChangesAsync(cts.Token).GetAsyncEnumerator();
|
|
do
|
|
{
|
|
Assert.That(await changes.MoveNextAsync().ConfigureAwait(false), Is.True);
|
|
}
|
|
while (changes.Current.Sequence < 2);
|
|
Assert.That(changes.Current.Sequence, Is.EqualTo(2));
|
|
Assert.That(changes.Current.NodeId, Is.EqualTo(nodeId));
|
|
Assert.That(changes.Current.Node!.Payload, Is.EqualTo(new ByteString(new byte[] { 2 })));
|
|
cts.Cancel();
|
|
|
|
(bool bulkHasCounter, _) = await crdt.TryGetAsync(InMemoryNodeStateStore.SequenceKey)
|
|
.ConfigureAwait(false);
|
|
Assert.That(bulkHasCounter, Is.False, "the real CRDT backend must not hold or CAS the coordinator key");
|
|
}
|
|
|
|
[Test]
|
|
public async Task NodeStateBinaryRoundTripThroughStoreAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var nodeId = new NodeId("sensor", NamespaceIndex);
|
|
|
|
var original = new BaseDataVariableState(null)
|
|
{
|
|
NodeId = nodeId,
|
|
BrowseName = new QualifiedName("Sensor", NamespaceIndex),
|
|
DisplayName = new LocalizedText("Sensor"),
|
|
DataType = DataTypeIds.Double,
|
|
ValueRank = ValueRanks.Scalar,
|
|
Value = new Variant(3.14)
|
|
};
|
|
|
|
using var stream = new MemoryStream();
|
|
original.SaveAsBinary(m_systemContext, stream);
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(stream.ToArray()))).ConfigureAwait(false);
|
|
|
|
IStoredNode? stored = await store.TryGetNodeAsync(nodeId).ConfigureAwait(false);
|
|
Assert.That(stored, Is.Not.Null);
|
|
|
|
var restored = new BaseDataVariableState(null);
|
|
using var loadStream = new MemoryStream(stored!.Payload.ToArray());
|
|
restored.LoadAsBinary(m_systemContext, loadStream);
|
|
|
|
Assert.That(restored.BrowseName, Is.EqualTo(original.BrowseName));
|
|
Assert.That(restored.Value, Is.EqualTo(original.Value));
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProtectedNodeAndValueRoundTripAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(11));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var nodeId = new NodeId("secret", NamespaceIndex);
|
|
var payload = ByteString.From(new byte[] { 9, 8, 7, 6 });
|
|
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, payload)).ConfigureAwait(false);
|
|
await store.WriteValueAsync(nodeId, new DataValue(new Variant(99.0), StatusCodes.Good)).ConfigureAwait(false);
|
|
|
|
IStoredNode? node = await store.TryGetNodeAsync(nodeId).ConfigureAwait(false);
|
|
(bool found, DataValue value) = await store.TryReadValueAsync(nodeId).ConfigureAwait(false);
|
|
|
|
// The bytes persisted in the backend must not be the plaintext.
|
|
(bool rawFound, ByteString raw) = await kv.TryGetAsync("n/" + nodeId).ConfigureAwait(false);
|
|
Assert.That(rawFound, Is.True);
|
|
Assert.That(raw.ToArray(), Is.Not.EqualTo(payload.ToArray()));
|
|
|
|
Assert.That(node, Is.Not.Null);
|
|
Assert.That(node!.Payload.ToArray(), Is.EqualTo(payload.ToArray()));
|
|
Assert.That(found, Is.True);
|
|
Assert.That(value.WrappedValue, Is.EqualTo(new Variant(99.0)));
|
|
}
|
|
|
|
[Test]
|
|
public async Task TamperedProtectedNodeIsRejectedFailClosedAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(12));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
var nodeId = new NodeId("tampered", NamespaceIndex);
|
|
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(new byte[] { 1, 2, 3 }))).ConfigureAwait(false);
|
|
|
|
// A compromised store / rogue replica forges the persisted record.
|
|
await kv.SetAsync("n/" + nodeId, ByteString.From(new byte[] { 66, 66, 66, 66, 66 })).ConfigureAwait(false);
|
|
|
|
IStoredNode? node = await store.TryGetNodeAsync(nodeId).ConfigureAwait(false);
|
|
|
|
Assert.That(node, Is.Null);
|
|
}
|
|
|
|
[Test]
|
|
public async Task NodeProtectedUnderDifferentKeyIsRejectedAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var writerProtector = new AesCbcHmacRecordProtector(MakeKey(13));
|
|
using var readerProtector = new AesCbcHmacRecordProtector(MakeKey(14));
|
|
var writer = new InMemoryNodeStateStore(kv, m_messageContext, writerProtector);
|
|
var reader = new InMemoryNodeStateStore(kv, m_messageContext, readerProtector);
|
|
var nodeId = new NodeId("crosskey", NamespaceIndex);
|
|
|
|
await writer.UpsertNodeAsync(new StoredNode(nodeId, ByteString.From(new byte[] { 5, 5 }))).ConfigureAwait(false);
|
|
|
|
IStoredNode? node = await reader.TryGetNodeAsync(nodeId).ConfigureAwait(false);
|
|
|
|
Assert.That(node, Is.Null);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Rejects an unauthenticated sequence coordinator without changing its protected record or stored nodes.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task ProtectedCoordinatorRejectsWriterWithDifferentKeyAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var originalProtector = new AesCbcHmacRecordProtector(MakeKey(23));
|
|
using var otherProtector = new AesCbcHmacRecordProtector(MakeKey(24));
|
|
using var original = new InMemoryNodeStateStore(kv, m_messageContext, originalProtector);
|
|
using var other = new InMemoryNodeStateStore(kv, m_messageContext, otherProtector);
|
|
await original.UpsertNodeAsync(new StoredNode(
|
|
new NodeId("original", NamespaceIndex),
|
|
new ByteString(new byte[] { 1 }))).ConfigureAwait(false);
|
|
(_, ByteString record) = await kv.TryGetAsync(InMemoryNodeStateStore.SequenceKey).ConfigureAwait(false);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await other.UpsertNodeAsync(new StoredNode(
|
|
new NodeId("unauthenticated", NamespaceIndex),
|
|
new ByteString(new byte[] { 2 }))).ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
|
|
(_, ByteString unchanged) = await kv.TryGetAsync(InMemoryNodeStateStore.SequenceKey).ConfigureAwait(false);
|
|
Assert.That(unchanged, Is.EqualTo(record));
|
|
Assert.That(await CountKeysAsync(kv, "n/").ConfigureAwait(false), Is.EqualTo(1));
|
|
Assert.That(await CountKeysAsync(kv, "dlog/").ConfigureAwait(false), Is.EqualTo(1));
|
|
}
|
|
|
|
[Test]
|
|
public void UpsertNodeRejectsNullArgument()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
|
|
Assert.That(
|
|
async () => await store.UpsertNodeAsync(null!).ConfigureAwait(false),
|
|
Throws.ArgumentNullException);
|
|
}
|
|
|
|
[Test]
|
|
public async Task DeltaLogReplaysUpsertAndDeleteChangesAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var a = new NodeId("a", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(a, ByteString.From(new byte[] { 1 }))).ConfigureAwait(false);
|
|
bool deleted = await store.DeleteNodeAsync(a).ConfigureAwait(false);
|
|
|
|
var replayed = new List<NodeStateChange>();
|
|
await foreach (NodeStateChange change in store.ReadDeltaLogAsync(0))
|
|
{
|
|
replayed.Add(change);
|
|
}
|
|
|
|
Assert.That(deleted, Is.True);
|
|
Assert.That(replayed, Has.Count.EqualTo(2));
|
|
Assert.That(replayed[0].Kind, Is.EqualTo(NodeStateChangeKind.Upsert));
|
|
Assert.That(replayed[0].NodeId, Is.EqualTo(a));
|
|
Assert.That(replayed[0].Node, Is.Not.Null);
|
|
Assert.That(replayed[1].Kind, Is.EqualTo(NodeStateChangeKind.Delete));
|
|
Assert.That(replayed[1].NodeId, Is.EqualTo(a));
|
|
}
|
|
|
|
[Test]
|
|
public async Task WriteSnapshotOnEmptyStoreProducesEmptySnapshotAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
NodeStateSnapshot? snapshot = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
int count = 0;
|
|
await foreach (NodeStateChange _ in snapshot!.Entries)
|
|
{
|
|
count++;
|
|
}
|
|
Assert.That(count, Is.Zero);
|
|
}
|
|
|
|
[Test]
|
|
public async Task RepeatedSnapshotsCollectSupersededGenerationAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var a = new NodeId("a", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(a, ByteString.From(new byte[] { 1 }))).ConfigureAwait(false);
|
|
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
|
|
int generations = await CountSnapshotGenerationsAsync(kv).ConfigureAwait(false);
|
|
NodeStateSnapshot? snapshot = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
|
|
// The predecessor-of-predecessor generation is garbage-collected once
|
|
// the third manifest is published, leaving the live and one retained.
|
|
Assert.That(generations, Is.EqualTo(2));
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
}
|
|
|
|
[Test]
|
|
public async Task EnumerateSkipsMalformedNodeKeysAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var valid = new NodeId("good", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(valid, ByteString.From(new byte[] { 1 }))).ConfigureAwait(false);
|
|
|
|
// A bare-prefix key and an unparseable NodeId suffix must both be skipped.
|
|
var record = ByteString.From(new byte[] { 0, 0, 0, 0, 0, 0, 0, 0, 1 });
|
|
await kv.SetAsync("n/", record).ConfigureAwait(false);
|
|
await kv.SetAsync("n/i=abc", record).ConfigureAwait(false);
|
|
|
|
var ids = new List<NodeId>();
|
|
await foreach (IStoredNode node in store.EnumerateAsync())
|
|
{
|
|
ids.Add(node.NodeId);
|
|
}
|
|
|
|
Assert.That(ids, Is.EquivalentTo([valid]));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Rejects a truncated record under a valid node key instead of treating
|
|
/// an incomplete, nonempty hydration scan as authoritative.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task EnumerateRejectsTruncatedRecordAlongsideValidNodeAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var valid = new NodeId("good", NamespaceIndex);
|
|
var corrupt = new NodeId("corrupt", NamespaceIndex);
|
|
await store.UpsertNodeAsync(
|
|
new StoredNode(valid, ByteString.From(new byte[] { 1 }))).ConfigureAwait(false);
|
|
|
|
var ids = new List<NodeId>();
|
|
await foreach (IStoredNode node in store.EnumerateAsync().ConfigureAwait(false))
|
|
{
|
|
ids.Add(node.NodeId);
|
|
}
|
|
Assert.That(ids, Is.EquivalentTo([valid]));
|
|
|
|
await kv.SetAsync(
|
|
"n/" + corrupt,
|
|
ByteString.From(new byte[] { 1, 2, 3 })).ConfigureAwait(false);
|
|
|
|
ids.Clear();
|
|
Assert.That(
|
|
async () =>
|
|
{
|
|
await foreach (IStoredNode node in store.EnumerateAsync().ConfigureAwait(false))
|
|
{
|
|
ids.Add(node.NodeId);
|
|
}
|
|
},
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Rejects a valid node key's corrupt frame before or after a good record, including failed authentication.
|
|
/// </summary>
|
|
[TestCase(false, false)]
|
|
[TestCase(false, true)]
|
|
[TestCase(true, false)]
|
|
[TestCase(true, true)]
|
|
public async Task EnumerateRejectsCorruptionAtEitherScanPositionAsync(bool corruptFirst, bool authenticated)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(17));
|
|
Mock<ISharedKeyValueStore> ordered = CreateStoreMock(kv);
|
|
ordered.Setup(s => s.ScanAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string prefix, CancellationToken ct) => ScanSortedAsync(kv, prefix, ct));
|
|
using var store = new InMemoryNodeStateStore(
|
|
ordered.Object,
|
|
m_messageContext,
|
|
authenticated ? protector : NullRecordProtector.Instance);
|
|
var valid = new NodeId(corruptFirst ? "z-valid" : "a-valid", NamespaceIndex);
|
|
var corrupt = new NodeId(corruptFirst ? "a-corrupt" : "z-corrupt", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(valid, new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false);
|
|
byte[] frame = authenticated
|
|
? protector.Protect(
|
|
RecordProtectionContext.Create("node-state-record", "n/" + corrupt),
|
|
MakeRecord(2, new ByteString(new byte[] { 2 }))).ToArray()
|
|
: [1, 2, 3];
|
|
if (authenticated)
|
|
{
|
|
frame[^1] ^= 1;
|
|
}
|
|
await kv.SetAsync("n/" + corrupt, new ByteString(frame)).ConfigureAwait(false);
|
|
var ids = new List<NodeId>();
|
|
|
|
await Assert.ThatAsync(
|
|
async () =>
|
|
{
|
|
await foreach (IStoredNode node in store.EnumerateAsync().ConfigureAwait(false))
|
|
{
|
|
ids.Add(node.NodeId);
|
|
}
|
|
},
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
|
|
Assert.That(ids, Has.Count.EqualTo(corruptFirst ? 0 : 1));
|
|
if (!corruptFirst)
|
|
{
|
|
Assert.That(ids[0], Is.EqualTo(valid));
|
|
}
|
|
Assert.That(await store.TryGetNodeAsync(corrupt).ConfigureAwait(false), Is.Null);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Retains the prior manifest, retained generations, and delta log after a mixed corrupt snapshot scan.
|
|
/// </summary>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task SnapshotCorruptionPreservesManifestLogAndOwnedGenerationsAsync(bool corruptValue)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(18));
|
|
Mock<ISharedKeyValueStore> ordered = CreateStoreMock(kv);
|
|
ordered.Setup(s => s.ScanAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string prefix, CancellationToken ct) => ScanSortedAsync(kv, prefix, ct));
|
|
using var store = new InMemoryNodeStateStore(ordered.Object, m_messageContext, protector);
|
|
var valid = new NodeId("a-valid", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(valid, new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
(_, ByteString manifest) = await kv.TryGetAsync("snapmeta/manifest").ConfigureAwait(false);
|
|
Assert.That(await CountSnapshotGenerationsAsync(kv).ConfigureAwait(false), Is.EqualTo(2));
|
|
|
|
var largePayload = new ByteString(new byte[m_messageContext.MaxByteStringLength]);
|
|
await store.UpsertNodeAsync(new StoredNode(valid, largePayload))
|
|
.ConfigureAwait(false);
|
|
string corruptKey = (corruptValue ? "v/" : "n/") + new NodeId("z-corrupt", NamespaceIndex);
|
|
await kv.SetAsync(
|
|
corruptKey,
|
|
protector.Protect(
|
|
RecordProtectionContext.Create("node-state-record", corruptKey),
|
|
new ByteString(new byte[] { 1, 2, 3 }))).ConfigureAwait(false);
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await store.WriteSnapshotAsync().ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
|
|
(_, ByteString unchanged) = await kv.TryGetAsync("snapmeta/manifest").ConfigureAwait(false);
|
|
Assert.That(unchanged, Is.EqualTo(manifest));
|
|
Assert.That(await CountSnapshotGenerationsAsync(kv).ConfigureAwait(false), Is.EqualTo(2));
|
|
List<NodeStateChange> retained = await ReadChangesAsync(store.ReadDeltaLogAsync(0)).ConfigureAwait(false);
|
|
Assert.That(retained, Has.Count.EqualTo(1));
|
|
Assert.That(retained[0].Sequence, Is.EqualTo(2));
|
|
NodeStateSnapshot? previous = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
Assert.That(previous, Is.Not.Null);
|
|
List<NodeStateChange> entries = await ReadChangesAsync(previous!.Entries).ConfigureAwait(false);
|
|
Assert.That(entries, Has.Count.EqualTo(1));
|
|
Assert.That(entries[0].Node!.Payload, Is.EqualTo(new ByteString(new byte[] { 1 })));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Treats valid topology tombstones as absence rather than corruption during scans and snapshot construction.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task AuthoritativeScansAcceptDeletionTombstonesAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var nodeId = new NodeId("deleted", NamespaceIndex);
|
|
await store.UpsertNodeAsync(new StoredNode(nodeId, new ByteString(new byte[] { 1 })))
|
|
.ConfigureAwait(false);
|
|
Assert.That(await store.DeleteNodeAsync(nodeId).ConfigureAwait(false), Is.True);
|
|
int count = 0;
|
|
await foreach (IStoredNode _ in store.EnumerateAsync().ConfigureAwait(false))
|
|
{
|
|
count++;
|
|
}
|
|
Assert.That(count, Is.Zero);
|
|
|
|
await store.WriteSnapshotAsync().ConfigureAwait(false);
|
|
NodeStateSnapshot? snapshot = await store.TryReadSnapshotAsync().ConfigureAwait(false);
|
|
Assert.That(snapshot, Is.Not.Null);
|
|
Assert.That(await ReadChangesAsync(snapshot!.Entries).ConfigureAwait(false), Is.Empty);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Rejects truncated value envelopes, empty values, trailing bytes, and unauthenticated value records.
|
|
/// </summary>
|
|
[TestCase("envelope")]
|
|
[TestCase("empty")]
|
|
[TestCase("trailing")]
|
|
[TestCase("authentication")]
|
|
public async Task EnumerateValuesRejectsInvalidRecordsAsync(string corruption)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(19));
|
|
using var store = new InMemoryNodeStateStore(
|
|
kv,
|
|
m_messageContext,
|
|
corruption == "authentication" ? protector : NullRecordProtector.Instance);
|
|
var valid = new NodeId("valid", NamespaceIndex);
|
|
var corrupt = new NodeId("corrupt", NamespaceIndex);
|
|
await store.WriteValueAsync(valid, new DataValue(new Variant(7))).ConfigureAwait(false);
|
|
ByteString frame = corruption switch
|
|
{
|
|
"empty" => MakeRecord(2, ByteString.Empty),
|
|
"trailing" => MakeRecord(2, new ByteString(new byte[] { 0, 1 })),
|
|
_ => new ByteString(new byte[] { 1, 2, 3 })
|
|
};
|
|
await kv.SetAsync("v/" + corrupt, frame).ConfigureAwait(false);
|
|
if (corruption == "empty")
|
|
{
|
|
(bool found, DataValue value) = await store.TryReadValueAsync(corrupt).ConfigureAwait(false);
|
|
Assert.That(found, Is.True);
|
|
Assert.That(value.IsNull, Is.True, "non-authoritative lookup compatibility is preserved");
|
|
}
|
|
|
|
await Assert.ThatAsync(
|
|
async () =>
|
|
{
|
|
await foreach ((NodeId _, DataValue _) in store.EnumerateValuesAsync().ConfigureAwait(false))
|
|
{
|
|
}
|
|
},
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Surfaces malformed delta envelopes, kinds, identifiers, payloads, and
|
|
/// trailing bytes instead of skipping them.
|
|
/// </summary>
|
|
[TestCase("envelope")]
|
|
[TestCase("kind")]
|
|
[TestCase("node")]
|
|
[TestCase("payload")]
|
|
[TestCase("trailing")]
|
|
[TestCase("authentication")]
|
|
public async Task DeltaReplayRejectsInvalidFramesAsync(string corruption)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(20));
|
|
using var store = new InMemoryNodeStateStore(
|
|
kv,
|
|
m_messageContext,
|
|
corruption == "authentication" ? protector : NullRecordProtector.Instance);
|
|
ByteString frame;
|
|
if (corruption is "envelope" or "authentication")
|
|
{
|
|
frame = new ByteString(new byte[] { 1, 2, 3 });
|
|
}
|
|
else
|
|
{
|
|
using var encoder = new BinaryEncoder(m_messageContext);
|
|
encoder.WriteByte(null, corruption == "kind" ? byte.MaxValue : (byte)NodeStateChangeKind.Upsert);
|
|
encoder.WriteNodeId(null, corruption == "node" ? NodeId.Null : new NodeId("delta", NamespaceIndex));
|
|
encoder.WriteByteString(
|
|
null,
|
|
corruption == "payload" ? ByteString.Empty : new ByteString(new byte[] { 1 }));
|
|
if (corruption == "trailing")
|
|
{
|
|
encoder.WriteByte(null, 0);
|
|
}
|
|
frame = new ByteString(encoder.CloseAndReturnBuffer()!);
|
|
}
|
|
await kv.SetAsync("dlog/00000000000000000001", frame).ConfigureAwait(false);
|
|
|
|
await Assert.ThatAsync(
|
|
() => ReadChangesAsync(store.ReadDeltaLogAsync(0)),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Does not wrap overflowing delta key suffixes into valid earlier sequence numbers.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task DeltaReplaySkipsOverflowedSequenceKeysAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
await kv.SetAsync("dlog/18446744073709551617", new ByteString(new byte[] { 1, 2, 3 }))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(await ReadChangesAsync(store.ReadDeltaLogAsync(0)).ConfigureAwait(false), Is.Empty);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reports corrupt node, value, and delta frames received on a live backend feed.
|
|
/// </summary>
|
|
[TestCase("n/")]
|
|
[TestCase("v/")]
|
|
[TestCase("dlog/")]
|
|
public async Task WatchRejectsCorruptLiveFramesAsync(string prefix)
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(21));
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext, protector);
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
await using IAsyncEnumerator<NodeStateChange> changes =
|
|
store.SubscribeChangesAsync(cts.Token).GetAsyncEnumerator();
|
|
Task<bool> pending = changes.MoveNextAsync().AsTask();
|
|
string key = prefix +
|
|
(prefix == "dlog/"
|
|
? "00000000000000000001"
|
|
: new NodeId("live", NamespaceIndex).ToString());
|
|
await kv.SetAsync(key, new ByteString(new byte[] { 1, 2, 3 }), cts.Token).ConfigureAwait(false);
|
|
|
|
await Assert.ThatAsync(
|
|
() => pending,
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
cts.Cancel();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reports corrupt CRDT rows even when they are already present in the polling baseline.
|
|
/// </summary>
|
|
[TestCase("n/")]
|
|
[TestCase("v/")]
|
|
[TestCase("dlog/")]
|
|
public async Task PollingRejectsCorruptBaselineAsync(string prefix)
|
|
{
|
|
await using var network = new InMemoryNetwork();
|
|
await using var crdt = new ReplicatedSharedKeyValueStore(
|
|
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
|
|
using var protector = new AesCbcHmacRecordProtector(MakeKey(22));
|
|
using var store = new InMemoryNodeStateStore(crdt, m_messageContext, protector);
|
|
string key = prefix +
|
|
(prefix == "dlog/"
|
|
? "00000000000000000001"
|
|
: new NodeId("baseline", NamespaceIndex).ToString());
|
|
await crdt.SetAsync(key, new ByteString(new byte[] { 1, 2, 3 })).ConfigureAwait(false);
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
await using IAsyncEnumerator<NodeStateChange> changes =
|
|
store.SubscribeChangesAsync(cts.Token).GetAsyncEnumerator();
|
|
|
|
await Assert.ThatAsync(
|
|
async () => await changes.MoveNextAsync().ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadDecodingError)).ConfigureAwait(false);
|
|
cts.Cancel();
|
|
}
|
|
|
|
[Test]
|
|
public async Task TryGetNodeReturnsNullForTruncatedRecordAsync()
|
|
{
|
|
using var kv = new InMemorySharedKeyValueStore();
|
|
using var store = new InMemoryNodeStateStore(kv, m_messageContext);
|
|
var nodeId = new NodeId("short", NamespaceIndex);
|
|
|
|
// A record shorter than the 8-byte sequence header cannot be split.
|
|
await kv.SetAsync("n/" + nodeId, ByteString.From(new byte[] { 1, 2, 3 })).ConfigureAwait(false);
|
|
|
|
IStoredNode? node = await store.TryGetNodeAsync(nodeId).ConfigureAwait(false);
|
|
|
|
Assert.That(node, Is.Null);
|
|
}
|
|
|
|
private static Mock<ISharedKeyValueStore> CreateStoreMock(ISharedKeyValueStore inner)
|
|
{
|
|
var store = new Mock<ISharedKeyValueStore>();
|
|
store.Setup(s => s.TryGetAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string key, CancellationToken ct) => inner.TryGetAsync(key, ct));
|
|
store.Setup(s => s.SetAsync(It.IsAny<string>(), It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string key, ByteString value, CancellationToken ct) => inner.SetAsync(key, value, ct));
|
|
store.Setup(s => s.CompareAndSwapAsync(
|
|
It.IsAny<string>(), It.IsAny<ByteString>(), It.IsAny<ByteString>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string key, ByteString expected, ByteString value, CancellationToken ct) =>
|
|
inner.CompareAndSwapAsync(key, expected, value, ct));
|
|
store.Setup(s => s.DeleteAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string key, CancellationToken ct) => inner.DeleteAsync(key, ct));
|
|
store.Setup(s => s.ScanAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string prefix, CancellationToken ct) => inner.ScanAsync(prefix, ct));
|
|
store.Setup(s => s.WatchAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
|
.Returns((string prefix, CancellationToken ct) => inner.WatchAsync(prefix, ct));
|
|
store.As<ISharedKeyValueStoreConsistency>()
|
|
.Setup(s => s.IsLinearizable(It.IsAny<string>()))
|
|
.Returns((string key) => inner is ISharedKeyValueStoreConsistency consistency &&
|
|
consistency.IsLinearizable(key));
|
|
store.As<ISharedKeyValueStoreConsistency>()
|
|
.Setup(s => s.IsProcessLocal(It.IsAny<string>()))
|
|
.Returns((string key) => inner is not ISharedKeyValueStoreConsistency consistency ||
|
|
consistency.IsProcessLocal(key));
|
|
return store;
|
|
}
|
|
|
|
private static ByteString MakeRecord(ulong sequence, ByteString payload)
|
|
{
|
|
byte[] record = new byte[sizeof(ulong) + payload.Length];
|
|
BinaryPrimitives.WriteUInt64BigEndian(record, sequence);
|
|
payload.Span.CopyTo(record.AsSpan(sizeof(ulong)));
|
|
return new ByteString(record);
|
|
}
|
|
|
|
private static async Task<List<NodeStateChange>> ReadChangesAsync(IAsyncEnumerable<NodeStateChange> source)
|
|
{
|
|
var changes = new List<NodeStateChange>();
|
|
await foreach (NodeStateChange change in source.ConfigureAwait(false))
|
|
{
|
|
changes.Add(change);
|
|
}
|
|
return changes;
|
|
}
|
|
|
|
private static async Task<int> CountKeysAsync(ISharedKeyValueStore store, string prefix)
|
|
{
|
|
int count = 0;
|
|
await foreach (KeyValuePair<string, ByteString> _ in store.ScanAsync(prefix).ConfigureAwait(false))
|
|
{
|
|
count++;
|
|
}
|
|
return count;
|
|
}
|
|
|
|
private static async IAsyncEnumerable<KeyValuePair<string, ByteString>> ScanSortedAsync(
|
|
InMemorySharedKeyValueStore store,
|
|
string prefix,
|
|
[EnumeratorCancellation] CancellationToken ct)
|
|
{
|
|
var entries = new List<KeyValuePair<string, ByteString>>();
|
|
await foreach (KeyValuePair<string, ByteString> entry in store.ScanAsync(prefix, ct).ConfigureAwait(false))
|
|
{
|
|
entries.Add(entry);
|
|
}
|
|
entries.Sort(static (left, right) => string.CompareOrdinal(left.Key, right.Key));
|
|
foreach (KeyValuePair<string, ByteString> entry in entries)
|
|
{
|
|
yield return entry;
|
|
}
|
|
}
|
|
|
|
private static async Task<int> CountSnapshotGenerationsAsync(InMemorySharedKeyValueStore kv)
|
|
{
|
|
var generations = new HashSet<string>(StringComparer.Ordinal);
|
|
await foreach (KeyValuePair<string, ByteString> entry in kv.ScanAsync("snap/"))
|
|
{
|
|
string remainder = entry.Key["snap/".Length..];
|
|
string[] parts = remainder.Split('/');
|
|
if (parts.Length > 1 && parts[0].Length > 0)
|
|
{
|
|
generations.Add(parts[0]);
|
|
}
|
|
}
|
|
return generations.Count;
|
|
}
|
|
|
|
private static byte[] MakeKey(byte seed)
|
|
{
|
|
byte[] key = new byte[32];
|
|
for (int i = 0; i < key.Length; i++)
|
|
{
|
|
key[i] = (byte)(seed + i);
|
|
}
|
|
return key;
|
|
}
|
|
}
|
|
}
|