Skip to content

Commit 52cf479

Browse files
AarononthewebAaronontheweb
authored
Added spec to demonstrate failed snapshot-only recovery (#6822)
* Added spec to demonstrate failed snapshot-only recovery Tests whether a persistent actor will properly fail during a failed snapshot recovery or during a timed out one. This is all without ever having saved an event to the journal, so the `MaxSeqNo` is set to `0`. * accelerated timeframes
1 parent 17a9c77 commit 52cf479

File tree

3 files changed

+152
-1
lines changed

3 files changed

+152
-1
lines changed

‎src/core/Akka.Persistence.TestKit/Akka.Persistence.TestKit.csproj

+1-1
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
<PropertyGroup>
44
<AssemblyTitle>Akka.Persistence.TestKit</AssemblyTitle>
55
<Description>TestKit for writing tests for Akka.NET Persistance module.</Description>
6-
<TargetFrameworks>$(NetStandardLibVersion)</TargetFrameworks>
6+
<TargetFrameworks>$(NetStandardLibVersion);$(NetLibVersion)</TargetFrameworks>
77
<PackageTags>$(AkkaPackageTags);testkit;persistance</PackageTags>
88
<GenerateDocumentationFile>true</GenerateDocumentationFile>
99
</PropertyGroup>

‎src/core/Akka.Persistence.Tests/Akka.Persistence.Tests.csproj

+1
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88

99
<ItemGroup>
1010
<ProjectReference Include="..\..\contrib\serializers\Akka.Serialization.Hyperion\Akka.Serialization.Hyperion.csproj" />
11+
<ProjectReference Include="..\Akka.Persistence.TestKit.Xunit2\Akka.Persistence.TestKit.Xunit2.csproj" />
1112
<ProjectReference Include="..\Akka.Persistence\Akka.Persistence.csproj" />
1213
<ProjectReference Include="..\Akka.Remote\Akka.Remote.csproj" />
1314
<ProjectReference Include="..\Akka.Tests.Shared.Internals\Akka.Tests.Shared.Internals.csproj" />

‎src/core/Akka.Persistence.Tests/FailedSnapshotStoreRecoverySpec.cs

+150
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
//-----------------------------------------------------------------------
2+
// <copyright file="FailedSnapshotStoreRecoverySpec.cs" company="Akka.NET Project">
3+
// Copyright (C) 2009-2023 Lightbend Inc. <http://www.lightbend.com>
4+
// Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
5+
// </copyright>
6+
//-----------------------------------------------------------------------
7+
8+
using System;
9+
using System.Threading.Tasks;
10+
using Akka.Actor;
11+
using Akka.Configuration;
12+
using Akka.Event;
13+
using Akka.Persistence.TestKit;
14+
using Xunit;
15+
using Xunit.Abstractions;
16+
17+
namespace Akka.Persistence.Tests;
18+
19+
/// <summary>
20+
/// Scenario: actor that uses ONLY the SnapshotStore fails to recover - what happens?
21+
/// </summary>
22+
public class FailedSnapshotStoreRecoverySpec : PersistenceTestKit
23+
{
24+
public enum FailureMode
25+
{
26+
Explicit,
27+
Timeout
28+
}
29+
30+
public static readonly Config Config = ConfigurationFactory.ParseString(@"
31+
# need to set recovery timeout to 1s
32+
akka.persistence.journal-plugin-fallback.recovery-event-timeout = 1s
33+
");
34+
35+
public FailedSnapshotStoreRecoverySpec(ITestOutputHelper output) : base(Config, output:output){}
36+
37+
private record Save(string Data);
38+
39+
private record Fetch();
40+
41+
private record SnapshotSaved();
42+
private record RecoveryCompleted();
43+
44+
public sealed class PersistentActor : UntypedPersistentActor
45+
{
46+
public PersistentActor(string persistenceId, IActorRef targetActor)
47+
{
48+
PersistenceId = persistenceId;
49+
_targetActor = targetActor;
50+
}
51+
52+
private readonly ILoggingAdapter _log = Context.GetLogger();
53+
private readonly IActorRef _targetActor;
54+
55+
public override string PersistenceId { get; }
56+
public string CurrentData { get; set; } = "none";
57+
protected override void OnCommand(object message)
58+
{
59+
switch (message)
60+
{
61+
case Save s:
62+
{
63+
CurrentData = s.Data;
64+
SaveSnapshot(CurrentData);
65+
Sender.Tell("ack");
66+
break;
67+
}
68+
case Fetch:
69+
{
70+
Sender.Tell(CurrentData);
71+
break;
72+
}
73+
case SaveSnapshotSuccess success:
74+
{
75+
_log.Info("Snapshot saved");
76+
_targetActor.Tell(new SnapshotSaved());
77+
break;
78+
}
79+
case SaveSnapshotFailure failure:
80+
{
81+
_log.Error(failure.Cause, "Snapshot failed");
82+
break;
83+
}
84+
}
85+
}
86+
87+
protected override void OnRecover(object message)
88+
{
89+
switch (message)
90+
{
91+
case SnapshotOffer { Snapshot: string str }:
92+
{
93+
CurrentData = str;
94+
break;
95+
}
96+
}
97+
}
98+
99+
protected override void OnReplaySuccess()
100+
{
101+
_targetActor.Tell(new RecoveryCompleted());
102+
}
103+
104+
protected override void OnRecoveryFailure(Exception reason, object message = null)
105+
{
106+
_log.Error(reason, "Recovery failed");
107+
base.OnRecoveryFailure(reason, message);
108+
}
109+
}
110+
111+
[Theory(DisplayName = "PersistentActor using Snapshots only must fail if snapshots are irrecoverable")]
112+
[InlineData(FailureMode.Explicit)]
113+
[InlineData(FailureMode.Timeout)]
114+
public async Task PersistentActor_using_Snapshots_only_must_fail_if_snapshots_irrecoverable(FailureMode mode)
115+
{
116+
// arrange
117+
var probe = CreateTestProbe();
118+
var actor = Sys.ActorOf(Props.Create(() => new PersistentActor("p1", probe.Ref)));
119+
await probe.ExpectMsgAsync<RecoveryCompleted>();
120+
actor.Tell(new Save("a"), probe);
121+
await probe.ExpectMsgAsync("ack");
122+
await probe.ExpectMsgAsync<SnapshotSaved>();
123+
await actor.GracefulStop(RemainingOrDefault);
124+
125+
Task SelectBehavior(SnapshotStoreLoadBehavior behavior)
126+
{
127+
switch (mode)
128+
{
129+
case FailureMode.Timeout:
130+
return behavior.FailWithDelay(TimeSpan.FromMinutes(1));
131+
case FailureMode.Explicit:
132+
default:
133+
return behavior.Fail();
134+
}
135+
}
136+
137+
// act
138+
await WithSnapshotLoad(SelectBehavior, async () =>
139+
{
140+
await WithinAsync(RemainingOrDefault, async () =>
141+
{
142+
var actor2 = Sys.ActorOf(Props.Create(() => new PersistentActor("p1", probe.Ref)));
143+
Watch(actor2);
144+
await probe.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(150));
145+
await ExpectTerminatedAsync(actor2);
146+
});
147+
});
148+
149+
}
150+
}

0 commit comments

Comments
 (0)