Skip to content

Commit e6e014d

Browse files
AarononthewebAaronontheweb
authored
De-flake ClusterShardingSpec family: re-send Gets that race shard hand-off (#8500)
`ClusterShardingSpec` (base class for the Persistent/DData/WithEntityRecovery variants) failed in CI on both transports with the same shape: one node timed out waiting for an `Int32` counter reply and every other node then failed the next barrier. Mechanism. `HandOffStopper` replies `ShardStopped` the moment the last entity terminates - before the `Shard` actor stops, and before the `ShardRegion` observes `Terminated` and drops the shard from `_shards`/`_regionByShard`. The spec treats `ShardStopped` as "the shard is gone" and immediately sends a `Get` through the region, so the region forwards it into the dying shard: [.../RememberCounterEntitiesRegion/1] DeadLetter from [.../system/testActor1] to [.../RememberCounterEntitiesRegion/1]: Get { CounterId = 13 } Sharding is at-most-once, so the request has to be re-sent, not merely re-awaited. Two sites did not do that: * `recover_entities_upon_restart` wrapped `Get(13)` in `AwaitAssertAsync(5s)` but the inner `ExpectMsgAsync(0)` carried no bound, so it inherited `akka.test.single-expect-default` (5s) - the entire budget. The loop made exactly one attempt: AwaitAssert failed, timeout [00:00:05] is over after [1] attempts and [00:00:05.0055266] elapsed time * `permanently_stop_entities_which_passivate` sent `Get(25)` as a bare one-shot, so a single dropped message was an unconditional failure. Both now re-send with a fresh probe per attempt and a 1s per-attempt bound, so the 5s budget admits four sends instead of one. A fresh probe per attempt also keeps a late reply from a timed-out attempt out of the next attempt's queue; the same defect made the `Identify(4)` retry loop in the second phase a no-op, and it is bounded now too. The second phase also ran its three barriers inside `WithinAsync(15s)`. `EnterBarrierAsync` derives its timeout from `RemainingOr(barrier-timeout)`, so the rendezvous got the Within remainder instead of the configured 70s: EnterBarrier(Name: after-13, Role: [RoleName(sixth)], Timeout:00:00:13.8157551) timeout while waiting for barrier 'after-13' The Within is removed, matching the treatment already applied to `recover_entities_upon_restart`; its 15s is re-attached to the two waits it was actually protecting, so no wait ends up with a smaller budget than before. Verified by widening the hand-off window in a local throwaway build (delaying the Shard's stop after `ShardStopped`), which reproduces both CI signatures exactly - `Timeout 00:00:05 ... System.Int32` with `barrier failed:after-shard-restart`, and `Timeout 00:00:13.9 ... System.Int32` with `barrier failed:after-13` - and passes with these changes. Test-only change.
1 parent 81289e8 commit e6e014d

1 file changed

Lines changed: 91 additions & 57 deletions

File tree

‎src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingSpec.cs‎

Lines changed: 91 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -1004,15 +1004,29 @@ await RunOnAsync(async () =>
10041004
await ExpectMsgAsync(new ShardStopped("1"), TimeSpan.FromSeconds(10), "ShardStopped not received");
10051005

10061006
//Get the path to where the shard now resides
1007+
// HandOffStopper answers ShardStopped as soon as the last entity terminates,
1008+
// which is strictly before the Shard actor stops and before the ShardRegion
1009+
// observes Terminated and drops the shard from its routing table. A Get sent
1010+
// in that window is forwarded to the dying Shard and dead-lettered - sharding
1011+
// is at-most-once, so the request has to be re-sent, not merely re-awaited.
1012+
// Each attempt gets a fresh probe (a late reply from a previous attempt must
1013+
// not be mistaken for this attempt's) and a 1s bound: that covers one region
1014+
// retry-interval (1s) plus the shard restart, so the 5s budget admits four
1015+
// sends instead of the single one an unbounded inner expect allowed - the
1016+
// inner expect inherited akka.test.single-expect-default (5 s), the whole
1017+
// budget, so the loop never retried.
1018+
IActorRef entity13 = null;
10071019
await AwaitAssertAsync(async () =>
10081020
{
1009-
_persistentEntitiesRegion.Value.Tell(new Get(13));
1010-
await ExpectMsgAsync(0);
1021+
var probe0 = CreateTestProbe();
1022+
_persistentEntitiesRegion.Value.Tell(new Get(13), probe0.Ref);
1023+
await probe0.ExpectMsgAsync(0, TimeSpan.FromSeconds(1));
1024+
entity13 = probe0.LastSender;
10111025
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
10121026

10131027
//Check that counter 1 is now alive again, even though we have
10141028
// not sent a message to it via the ShardRegion
1015-
var counter1 = Sys.ActorSelection(LastSender.Path.Parent / "1");
1029+
var counter1 = Sys.ActorSelection(entity13.Path.Parent / "1");
10161030
await WithinAsync(TimeSpan.FromSeconds(5), async () =>
10171031
{
10181032
await AwaitAssertAsync(async () =>
@@ -1055,78 +1069,98 @@ await RunOnAsync(async () =>
10551069

10561070
private async Task PersistentClusterSharding_should_permanently_stop_entities_which_passivate()
10571071
{
1058-
await WithinAsync(TimeSpan.FromSeconds(15), async () =>
1072+
// No outer Within, for the reason documented on
1073+
// PersistentClusterSharding_should_recover_entities_upon_restart: barriers derive
1074+
// their timeout from RemainingOr(barrier-timeout), so a Within here hands the
1075+
// rendezvous whatever is left of the phase budget instead of the configured 70s
1076+
// testconductor barrier-timeout. The 15s the removed Within granted this phase is
1077+
// re-attached below to the two waits it was actually protecting - the first Get
1078+
// against each region, which has to allocate the shard, recover the
1079+
// remember-entities store and persist through the shared journal - so no wait ends
1080+
// up with a smaller budget than it had, and the barriers get the full 70s.
1081+
RunOn(() =>
10591082
{
1060-
RunOn(() =>
1061-
{
1062-
_ = _persistentRegion.Value;
1063-
}, Config.Third, Config.Fourth, Config.Fifth);
1064-
await EnterBarrierAsync("cluster-started-12");
1083+
_ = _persistentRegion.Value;
1084+
}, Config.Third, Config.Fourth, Config.Fifth);
1085+
await EnterBarrierAsync("cluster-started-12");
10651086

1066-
await RunOnAsync(async () =>
1067-
{
1068-
//create and increment counter 1
1069-
_persistentRegion.Value.Tell(new EntityEnvelope(1, Increment.Instance));
1070-
_persistentRegion.Value.Tell(new Get(1));
1071-
await ExpectMsgAsync(1);
1087+
await RunOnAsync(async () =>
1088+
{
1089+
//create and increment counter 1
1090+
_persistentRegion.Value.Tell(new EntityEnvelope(1, Increment.Instance));
1091+
_persistentRegion.Value.Tell(new Get(1));
1092+
await ExpectMsgAsync(1, TimeSpan.FromSeconds(15));
10721093

1073-
var counter1 = LastSender;
1074-
var shard = Sys.ActorSelection(counter1.Path.Parent);
1075-
var region = Sys.ActorSelection(counter1.Path.Parent.Parent);
1094+
var counter1 = LastSender;
1095+
var shard = Sys.ActorSelection(counter1.Path.Parent);
1096+
var region = Sys.ActorSelection(counter1.Path.Parent.Parent);
10761097

1077-
//create and increment counter 13
1078-
_persistentRegion.Value.Tell(new EntityEnvelope(13, Increment.Instance));
1079-
_persistentRegion.Value.Tell(new Get(13));
1080-
await ExpectMsgAsync(1);
1098+
//create and increment counter 13
1099+
_persistentRegion.Value.Tell(new EntityEnvelope(13, Increment.Instance));
1100+
_persistentRegion.Value.Tell(new Get(13));
1101+
await ExpectMsgAsync(1, TimeSpan.FromSeconds(15));
10811102

1082-
var counter13 = LastSender;
1103+
var counter13 = LastSender;
10831104

1084-
counter13.Path.Parent.Should().Be(counter1.Path.Parent);
1105+
counter13.Path.Parent.Should().Be(counter1.Path.Parent);
10851106

1086-
//Send the shard the passivate message from the counter
1087-
await WatchAsync(counter1);
1088-
shard.Tell(new Passivate(Stop.Instance), counter1);
1107+
//Send the shard the passivate message from the counter
1108+
await WatchAsync(counter1);
1109+
shard.Tell(new Passivate(Stop.Instance), counter1);
10891110

1090-
// watch for the Terminated message
1091-
await ExpectTerminatedAsync(counter1, TimeSpan.FromSeconds(5));
1111+
// watch for the Terminated message
1112+
await ExpectTerminatedAsync(counter1, TimeSpan.FromSeconds(5));
10921113

1114+
await AwaitAssertAsync(async () =>
1115+
{
1116+
// check counter 1 is dead
10931117
var probe1 = CreateTestProbe();
1094-
await AwaitAssertAsync(async () =>
1095-
{
1096-
// check counter 1 is dead
1097-
counter1.Tell(new Identify(1), probe1.Ref);
1098-
await probe1.ExpectMsgAsync(new ActorIdentity(1, null), TimeSpan.FromSeconds(1), "Entity 1 was still around");
1099-
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
1118+
counter1.Tell(new Identify(1), probe1.Ref);
1119+
await probe1.ExpectMsgAsync(new ActorIdentity(1, null), TimeSpan.FromSeconds(1), "Entity 1 was still around");
1120+
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
11001121

1101-
// stop shard cleanly
1102-
region.Tell(new HandOff("1"));
1103-
await ExpectMsgAsync(new ShardStopped("1"), TimeSpan.FromSeconds(10), "ShardStopped not received");
1122+
// stop shard cleanly
1123+
region.Tell(new HandOff("1"));
1124+
await ExpectMsgAsync(new ShardStopped("1"), TimeSpan.FromSeconds(10), "ShardStopped not received");
11041125

1105-
}, Config.Third);
1106-
await EnterBarrierAsync("shard-shutdonw-12");
1126+
}, Config.Third);
1127+
await EnterBarrierAsync("shard-shutdonw-12");
11071128

1108-
await RunOnAsync(async () =>
1129+
await RunOnAsync(async () =>
1130+
{
1131+
// force shard backup
1132+
// ShardStopped - which released the barrier above - is answered by the
1133+
// HandOffStopper the moment the last entity terminates, before the Shard
1134+
// actor stops and before its ShardRegion drops it from the routing table.
1135+
// Shard "1" may be hosted by this node, so this Get can be forwarded to the
1136+
// dying Shard and dead-lettered. Sharding is at-most-once: re-send until the
1137+
// shard has been re-allocated. 1s per attempt covers one region
1138+
// retry-interval (1s) plus the restart, so the 5s budget admits four sends.
1139+
IActorRef entity25 = null;
1140+
await AwaitAssertAsync(async () =>
11091141
{
1110-
// force shard backup
1111-
_persistentRegion.Value.Tell(new Get(25));
1112-
await ExpectMsgAsync(0);
1142+
var probe0 = CreateTestProbe();
1143+
_persistentRegion.Value.Tell(new Get(25), probe0.Ref);
1144+
await probe0.ExpectMsgAsync(0, TimeSpan.FromSeconds(1));
1145+
entity25 = probe0.LastSender;
1146+
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
11131147

1114-
var shard = LastSender.Path.Parent;
1148+
var shard = entity25.Path.Parent;
11151149

1116-
// check counter 1 is still dead
1117-
Sys.ActorSelection(shard / "1").Tell(new Identify(3));
1118-
await ExpectMsgAsync(new ActorIdentity(3, null));
1150+
// check counter 1 is still dead
1151+
Sys.ActorSelection(shard / "1").Tell(new Identify(3));
1152+
await ExpectMsgAsync(new ActorIdentity(3, null), TimeSpan.FromSeconds(5));
11191153

1120-
// check counter 13 is alive again
1154+
// check counter 13 is alive again
1155+
await AwaitAssertAsync(async () =>
1156+
{
11211157
var probe3 = CreateTestProbe();
1122-
await AwaitAssertAsync(async () =>
1123-
{
1124-
Sys.ActorSelection(shard / "13").Tell(new Identify(4), probe3.Ref);
1125-
await probe3.ExpectMsgAsync<ActorIdentity>(i => i.MessageId.Equals(4) && i.Subject != null);
1126-
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
1127-
}, Config.Fourth);
1128-
await EnterBarrierAsync("after-13");
1129-
});
1158+
Sys.ActorSelection(shard / "13").Tell(new Identify(4), probe3.Ref);
1159+
await probe3.ExpectMsgAsync<ActorIdentity>(
1160+
i => i.MessageId.Equals(4) && i.Subject != null, TimeSpan.FromSeconds(1));
1161+
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
1162+
}, Config.Fourth);
1163+
await EnterBarrierAsync("after-13");
11301164
}
11311165

11321166
private async Task PersistentClusterSharding_should_restart_entities_which_stop_without_passivation()

0 commit comments

Comments
 (0)