293 lines
12 KiB
C#
293 lines
12 KiB
C#
using System;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Xunit;
|
|
using ZB.MOM.WW.CBDDC.Core;
|
|
using ZB.MOM.WW.CBDDC.Core.Network;
|
|
using ZB.MOM.WW.CBDDC.Core.Storage;
|
|
|
|
namespace ZB.MOM.WW.CBDDC.Network.Tests;
|
|
|
|
public class SyncOrchestratorMaintenancePruningTests
|
|
{
|
|
/// <summary>
|
|
/// Verifies that mixed peer confirmations produce the safest effective cutoff across peers and sources.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task CalculateEffectiveCutoffAsync_MixedPeerStates_ShouldUseSafestConfirmationAcrossPeers()
|
|
{
|
|
var oplogStore = Substitute.For<IOplogStore>();
|
|
var confirmationStore = Substitute.For<IPeerOplogConfirmationStore>();
|
|
var calculator = new OplogPruneCutoffCalculator(oplogStore, confirmationStore);
|
|
|
|
var vectorClock = new VectorClock();
|
|
vectorClock.SetTimestamp("node-local", new HlcTimestamp(500, 0, "node-local"));
|
|
vectorClock.SetTimestamp("node-secondary", new HlcTimestamp(450, 0, "node-secondary"));
|
|
oplogStore.GetVectorClockAsync(Arg.Any<CancellationToken>())
|
|
.Returns(vectorClock);
|
|
|
|
confirmationStore.GetActiveTrackedPeersAsync(Arg.Any<CancellationToken>())
|
|
.Returns(new[] { "peer-a", "peer-b", " " });
|
|
|
|
confirmationStore.GetConfirmationsForPeerAsync("peer-a", Arg.Any<CancellationToken>())
|
|
.Returns(new[]
|
|
{
|
|
CreateConfirmation("peer-a", "node-local", wall: 300, logic: 0, isActive: true),
|
|
CreateConfirmation("peer-a", "node-secondary", wall: 120, logic: 1, isActive: true),
|
|
CreateConfirmation("peer-a", "node-secondary", wall: 500, logic: 0, isActive: false)
|
|
});
|
|
|
|
confirmationStore.GetConfirmationsForPeerAsync("peer-b", Arg.Any<CancellationToken>())
|
|
.Returns(new[]
|
|
{
|
|
CreateConfirmation("peer-b", "node-local", wall: 250, logic: 0, isActive: true),
|
|
CreateConfirmation("peer-b", "node-secondary", wall: 180, logic: 0, isActive: true)
|
|
});
|
|
|
|
var decision = await calculator.CalculateEffectiveCutoffAsync(
|
|
new PeerNodeConfiguration
|
|
{
|
|
NodeId = "node-local",
|
|
OplogRetentionHours = 24
|
|
},
|
|
CancellationToken.None);
|
|
|
|
decision.HasCutoff.ShouldBeTrue();
|
|
decision.ConfirmationCutoff.HasValue.ShouldBeTrue();
|
|
decision.EffectiveCutoff.HasValue.ShouldBeTrue();
|
|
decision.ConfirmationCutoff.Value.PhysicalTime.ShouldBe(120);
|
|
decision.ConfirmationCutoff.Value.LogicalCounter.ShouldBe(1);
|
|
decision.ConfirmationCutoff.Value.NodeId.ShouldBe("node-secondary");
|
|
decision.EffectiveCutoff.Value.ShouldBe(decision.ConfirmationCutoff.Value);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that removing a peer from tracking immediately restores pruning eligibility.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task CalculateEffectiveCutoffAsync_RemovingPeerFromTracking_ShouldImmediatelyRestoreEligibility()
|
|
{
|
|
var oplogStore = Substitute.For<IOplogStore>();
|
|
var confirmationStore = Substitute.For<IPeerOplogConfirmationStore>();
|
|
var calculator = new OplogPruneCutoffCalculator(oplogStore, confirmationStore);
|
|
|
|
var vectorClock = new VectorClock();
|
|
vectorClock.SetTimestamp("node-local", new HlcTimestamp(200, 0, "node-local"));
|
|
oplogStore.GetVectorClockAsync(Arg.Any<CancellationToken>())
|
|
.Returns(vectorClock);
|
|
|
|
confirmationStore.GetActiveTrackedPeersAsync(Arg.Any<CancellationToken>())
|
|
.Returns(
|
|
new[] { "peer-active", "peer-deprecated" },
|
|
new[] { "peer-active" });
|
|
|
|
confirmationStore.GetConfirmationsForPeerAsync("peer-active", Arg.Any<CancellationToken>())
|
|
.Returns(new[]
|
|
{
|
|
CreateConfirmation("peer-active", "node-local", wall: 150, logic: 0, isActive: true)
|
|
});
|
|
confirmationStore.GetConfirmationsForPeerAsync("peer-deprecated", Arg.Any<CancellationToken>())
|
|
.Returns(Array.Empty<PeerOplogConfirmation>());
|
|
|
|
var configuration = new PeerNodeConfiguration
|
|
{
|
|
NodeId = "node-local",
|
|
OplogRetentionHours = 24
|
|
};
|
|
|
|
var blockedDecision = await calculator.CalculateEffectiveCutoffAsync(configuration, CancellationToken.None);
|
|
blockedDecision.HasCutoff.ShouldBeFalse();
|
|
|
|
confirmationStore.ClearReceivedCalls();
|
|
|
|
var unblockedDecision = await calculator.CalculateEffectiveCutoffAsync(configuration, CancellationToken.None);
|
|
unblockedDecision.HasCutoff.ShouldBeTrue();
|
|
unblockedDecision.EffectiveCutoff.HasValue.ShouldBeTrue();
|
|
unblockedDecision.EffectiveCutoff.Value.PhysicalTime.ShouldBe(150);
|
|
unblockedDecision.EffectiveCutoff.Value.NodeId.ShouldBe("node-local");
|
|
|
|
await confirmationStore.Received(1).GetConfirmationsForPeerAsync("peer-active", Arg.Any<CancellationToken>());
|
|
await confirmationStore.DidNotReceive().GetConfirmationsForPeerAsync("peer-deprecated", Arg.Any<CancellationToken>());
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that maintenance does not prune when peer confirmation is missing in a two-node topology.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task RunMaintenanceIfDueAsync_TwoNode_ShouldNotPruneBeforePeerConfirmation()
|
|
{
|
|
var oplogStore = Substitute.For<IOplogStore>();
|
|
var confirmationStore = Substitute.For<IPeerOplogConfirmationStore>();
|
|
var calculator = new OplogPruneCutoffCalculator(oplogStore, confirmationStore);
|
|
var orchestrator = CreateOrchestrator(oplogStore, confirmationStore, calculator);
|
|
|
|
var vectorClock = new VectorClock();
|
|
vectorClock.SetTimestamp("node-local", new HlcTimestamp(200, 0, "node-local"));
|
|
oplogStore.GetVectorClockAsync(Arg.Any<CancellationToken>())
|
|
.Returns(vectorClock);
|
|
|
|
confirmationStore.GetActiveTrackedPeersAsync(Arg.Any<CancellationToken>())
|
|
.Returns(new[] { "node-peer" });
|
|
confirmationStore.GetConfirmationsForPeerAsync("node-peer", Arg.Any<CancellationToken>())
|
|
.Returns(Array.Empty<PeerOplogConfirmation>());
|
|
|
|
var config = new PeerNodeConfiguration
|
|
{
|
|
NodeId = "node-local",
|
|
MaintenanceIntervalMinutes = 1,
|
|
OplogRetentionHours = 24
|
|
};
|
|
|
|
await orchestrator.RunMaintenanceIfDueAsync(config, DateTime.UtcNow, CancellationToken.None);
|
|
|
|
await oplogStore.DidNotReceive().PruneOplogAsync(Arg.Any<HlcTimestamp>(), Arg.Any<CancellationToken>());
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that maintenance prunes after peer confirmation is available in a two-node topology.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task RunMaintenanceIfDueAsync_TwoNode_ShouldPruneAfterPeerConfirmation()
|
|
{
|
|
var oplogStore = Substitute.For<IOplogStore>();
|
|
var confirmationStore = Substitute.For<IPeerOplogConfirmationStore>();
|
|
var calculator = new OplogPruneCutoffCalculator(oplogStore, confirmationStore);
|
|
var orchestrator = CreateOrchestrator(oplogStore, confirmationStore, calculator);
|
|
|
|
var vectorClock = new VectorClock();
|
|
vectorClock.SetTimestamp("node-local", new HlcTimestamp(200, 0, "node-local"));
|
|
oplogStore.GetVectorClockAsync(Arg.Any<CancellationToken>())
|
|
.Returns(vectorClock);
|
|
|
|
confirmationStore.GetActiveTrackedPeersAsync(Arg.Any<CancellationToken>())
|
|
.Returns(new[] { "node-peer" });
|
|
confirmationStore.GetConfirmationsForPeerAsync("node-peer", Arg.Any<CancellationToken>())
|
|
.Returns(new[]
|
|
{
|
|
new PeerOplogConfirmation
|
|
{
|
|
PeerNodeId = "node-peer",
|
|
SourceNodeId = "node-local",
|
|
ConfirmedWall = 100,
|
|
ConfirmedLogic = 0,
|
|
ConfirmedHash = "hash-100",
|
|
IsActive = true
|
|
}
|
|
});
|
|
|
|
var config = new PeerNodeConfiguration
|
|
{
|
|
NodeId = "node-local",
|
|
MaintenanceIntervalMinutes = 1,
|
|
OplogRetentionHours = 24
|
|
};
|
|
|
|
await orchestrator.RunMaintenanceIfDueAsync(config, DateTime.UtcNow, CancellationToken.None);
|
|
|
|
await oplogStore.Received(1).PruneOplogAsync(
|
|
Arg.Is<HlcTimestamp>(timestamp =>
|
|
timestamp.PhysicalTime == 100 &&
|
|
timestamp.LogicalCounter == 0 &&
|
|
string.Equals(timestamp.NodeId, "node-local", StringComparison.Ordinal)),
|
|
Arg.Any<CancellationToken>());
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that deprecated-node removal unblocks pruning on a subsequent maintenance run.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task RunMaintenanceIfDueAsync_DeprecatedNodeRemoval_ShouldUnblockPruning()
|
|
{
|
|
var oplogStore = Substitute.For<IOplogStore>();
|
|
var confirmationStore = Substitute.For<IPeerOplogConfirmationStore>();
|
|
var calculator = new OplogPruneCutoffCalculator(oplogStore, confirmationStore);
|
|
var orchestrator = CreateOrchestrator(oplogStore, confirmationStore, calculator);
|
|
|
|
var vectorClock = new VectorClock();
|
|
vectorClock.SetTimestamp("node-local", new HlcTimestamp(220, 0, "node-local"));
|
|
oplogStore.GetVectorClockAsync(Arg.Any<CancellationToken>())
|
|
.Returns(vectorClock);
|
|
|
|
confirmationStore.GetActiveTrackedPeersAsync(Arg.Any<CancellationToken>())
|
|
.Returns(
|
|
new[] { "node-active", "node-deprecated" },
|
|
new[] { "node-active" });
|
|
|
|
confirmationStore.GetConfirmationsForPeerAsync("node-active", Arg.Any<CancellationToken>())
|
|
.Returns(new[]
|
|
{
|
|
CreateConfirmation("node-active", "node-local", wall: 100, logic: 0, isActive: true)
|
|
});
|
|
confirmationStore.GetConfirmationsForPeerAsync("node-deprecated", Arg.Any<CancellationToken>())
|
|
.Returns(Array.Empty<PeerOplogConfirmation>());
|
|
|
|
var config = new PeerNodeConfiguration
|
|
{
|
|
NodeId = "node-local",
|
|
MaintenanceIntervalMinutes = 1,
|
|
OplogRetentionHours = 24
|
|
};
|
|
|
|
var now = DateTime.UtcNow;
|
|
await orchestrator.RunMaintenanceIfDueAsync(config, now, CancellationToken.None);
|
|
await oplogStore.DidNotReceive().PruneOplogAsync(Arg.Any<HlcTimestamp>(), Arg.Any<CancellationToken>());
|
|
|
|
await orchestrator.RunMaintenanceIfDueAsync(config, now.AddMinutes(2), CancellationToken.None);
|
|
|
|
await oplogStore.Received(1).PruneOplogAsync(
|
|
Arg.Is<HlcTimestamp>(timestamp =>
|
|
timestamp.PhysicalTime == 100 &&
|
|
timestamp.LogicalCounter == 0 &&
|
|
string.Equals(timestamp.NodeId, "node-local", StringComparison.Ordinal)),
|
|
Arg.Any<CancellationToken>());
|
|
}
|
|
|
|
private static SyncOrchestrator CreateOrchestrator(
|
|
IOplogStore oplogStore,
|
|
IPeerOplogConfirmationStore confirmationStore,
|
|
IOplogPruneCutoffCalculator cutoffCalculator)
|
|
{
|
|
var discovery = Substitute.For<IDiscoveryService>();
|
|
discovery.GetActivePeers().Returns(Array.Empty<PeerNode>());
|
|
|
|
var documentStore = Substitute.For<IDocumentStore>();
|
|
documentStore.InterestedCollection.Returns(Array.Empty<string>());
|
|
|
|
var snapshotMetadataStore = Substitute.For<ISnapshotMetadataStore>();
|
|
var snapshotService = Substitute.For<ISnapshotService>();
|
|
var configProvider = Substitute.For<IPeerNodeConfigurationProvider>();
|
|
configProvider.GetConfiguration().Returns(new PeerNodeConfiguration { NodeId = "node-local" });
|
|
|
|
return new SyncOrchestrator(
|
|
discovery,
|
|
oplogStore,
|
|
documentStore,
|
|
snapshotMetadataStore,
|
|
snapshotService,
|
|
configProvider,
|
|
NullLoggerFactory.Instance,
|
|
confirmationStore,
|
|
telemetry: null,
|
|
oplogPruneCutoffCalculator: cutoffCalculator);
|
|
}
|
|
|
|
private static PeerOplogConfirmation CreateConfirmation(
|
|
string peerNodeId,
|
|
string sourceNodeId,
|
|
long wall,
|
|
int logic,
|
|
bool isActive)
|
|
{
|
|
return new PeerOplogConfirmation
|
|
{
|
|
PeerNodeId = peerNodeId,
|
|
SourceNodeId = sourceNodeId,
|
|
ConfirmedWall = wall,
|
|
ConfirmedLogic = logic,
|
|
ConfirmedHash = $"hash-{wall}-{logic}",
|
|
IsActive = isActive
|
|
};
|
|
}
|
|
}
|