From 1d3c7adbdfdff5ce43a15bd4eac6f40d562543fe Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 10 Jul 2026 20:14:40 +0000 Subject: [PATCH 1/2] Akka.Cluster: make Cluster extension startup non-blocking The Cluster constructor blocked on GetClusterCoreRef().Result (an Ask to /system/cluster bounded by akka.actor.creation-timeout), and the internal ClusterCore getter re-issued that blocking ask whenever it observed an unresolved core ref. Actors spawned during construction (cluster event subscribers, the SBR downing provider) hit the getter from PreStart and parked dispatcher threads; on small pools the parked waiters starved the very daemon that had to reply, the redundant asks timed out, and the timeout path called Shutdown() on an already-healthy node. This is the root cause of the StressSpec nondeterminism on 2-core Windows CI agents and reproduces as a hard deadlock with a 1-thread default dispatcher. - Replace the constructor ask with fire-and-forget InternalClusterAction.Init(cluster); the constructor is now straight-line code whose completion depends on no other thread - ClusterDaemon and ClusterCoreSupervisor become explicit Uninitialized/Initialized state machines (Become + unbounded stash): pre-Init traffic is buffered, post-Init unmatched traffic forwards to the core daemon; ordering is guaranteed by mailbox FIFO, not blocking - ClusterCoreSupervisor hands the core ref back via Cluster.SetClusterCoreRef(); the ClusterCore getter is now unconditionally non-blocking (_clusterCore ?? _clusterDaemons) - Delete GetClusterCoreRef() and its Shutdown()-on-timeout path; akka.actor.creation-timeout is no longer consulted by Akka.Cluster; genuine core-startup failures still shut the node down via the existing ClusterCoreSupervisor supervision strategy - Add ClusterStartupFuzzSpec: randomized startup fuzz under pinned 1-2-thread dispatcher pools; killed 63-91% of iterations pre-fix and gates this class of regression at zero kills - Record the behavioral change in BREAKING_CHANGES_V1.6.md - dotnet format applied to the touched files (includes pre-existing whitespace cleanup; diff is unchanged with whitespace ignored) Verification: fuzz spec 1,980/1,980 iterations reached Up (0 kills, 0 soft-timeouts) across baseline / 2-core / 1-core / CPU-saturated / combined conditions; StressSpec passes 3/3 under taskset -c 0,1 where it previously failed in ~90s with the CI signature; Akka.Cluster.Tests 379/379; Cluster.Tools 98/98; Cluster.Metrics 44/44; Sharding 191/192 (the 1 failure is pre-existing on dev, unrelated); no public API changes (Akka.API.Tests 18/18). --- BREAKING_CHANGES_V1.6.md | 1 + .../ClusterStartupFuzzSpec.cs | 653 ++++++++++++++++++ src/core/Akka.Cluster/Cluster.cs | 95 +-- src/core/Akka.Cluster/ClusterDaemon.cs | 211 ++++-- 4 files changed, 843 insertions(+), 117 deletions(-) create mode 100644 src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs diff --git a/BREAKING_CHANGES_V1.6.md b/BREAKING_CHANGES_V1.6.md index 71b2bfc36f1..e7a2a03edfc 100644 --- a/BREAKING_CHANGES_V1.6.md +++ b/BREAKING_CHANGES_V1.6.md @@ -40,6 +40,7 @@ of `Behavior`, `Wire`, `API` (combine with `+`). | Status | PR / Branch | Component | Type | Change | Migration | |--------|-------------|-----------|------|--------|-----------| +| Planned | `fix/cluster-nonblocking-startup` | `Akka.Cluster` | Behavior | Cluster extension startup no longer blocks on an internal 20s ask bounded by `akka.actor.creation-timeout`; `Cluster.Get()` returns immediately and core initialization completes asynchronously. The "Failed to startup Cluster" ask-timeout failure mode (which could terminate a healthy node under thread starvation) is removed; genuine core-startup failures still shut the node down via supervision. | None for typical code — all public APIs are message-based or awaitable. Code that used `Cluster.Get()` as a "core is started" synchronization point should await `JoinAsync`/`RegisterOnMemberUp` or observe cluster state. | | Planned | `fix/artery-pipe-watermarks` | `Akka.Remote` (Artery) | Behavior | Artery TCP connections now raise the pause/resume watermarks of their internal read pipe to 1 MiB (pause 2 MiB), via a new opt-in `Akka.IO.Inet.SO.PipeBufferSize` socket option consulted by `TcpIncomingConnection`/`TcpOutgoingConnection`, configurable via `akka.remote.artery.advanced.tcp.pipe-buffer-size` (default `1m`). Akka.IO's own default (~16 KiB, derived from `akka.io.tcp.receive-buffer-size`) is unchanged for every other TCP connection. | No action required. Artery connections use somewhat more memory per connection and see higher throughput under high-in-flight/one-way traffic; tune via `akka.remote.artery.advanced.tcp.pipe-buffer-size` if needed. | | Merged | [#8317](https://github.com/akkadotnet/akka.net/pull/8317) | `Akka.Remote` / `Akka.Cluster` | API | System/address UID widened `int` → `long` across `AddressUid.Uid`, `AddressUidExtension.Uid()`, `UniqueAddress(Address, int)`/`.Uid`, `QuarantinedEvent(Address, int)`/`.Uid`, `RemoteWatcher.HeartbeatRsp(int)`/`.AddressUid`, and all `Quarantine(Address, int? uid)` members (`IRemoteActorRefProvider` / `RemoteTransport` / `RemoteWatcher`). Hard re-type (v1.6). Prerequisite for Artery. | Recompile against the `long` members; replace `int` uid locals with `long`. Default UID generation stays in 32-bit range; full 64-bit generation is opt-in via `akka.remote.use-64bit-system-uids = on`. | | Merged | [#8317](https://github.com/akkadotnet/akka.net/pull/8317) | `Akka.Cluster` | Wire | Cluster gossip `UniqueAddress.uid` widened `uint32` → `uint64` (`ClusterMessages.proto`). Same varint wire type → binary-compatible for uids ≤ `uint32.MaxValue`. | Rolling upgrade safe provided uid generation stays in 32-bit range (the default); enable `akka.remote.use-64bit-system-uids` only after the whole cluster + all remote peers are on v1.6. | diff --git a/src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs b/src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs new file mode 100644 index 00000000000..59695443a58 --- /dev/null +++ b/src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs @@ -0,0 +1,653 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2022 Lightbend Inc. +// Copyright (C) 2013-2026 .NET Foundation +// +//----------------------------------------------------------------------- + +#nullable enable + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Globalization; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Configuration; +using Akka.Dispatch; +using Akka.Event; +using Akka.TestKit; +using Xunit; + +namespace Akka.Cluster.Tests +{ + /// + /// Thread-safe sink that the uses to hand back captured + /// "Failed to startup Cluster" error text to the test, keyed by (unique) actor-system name. + /// + internal static class FuzzFailureSink + { + private static readonly ConcurrentDictionary> ByName = new(); + + public static void Record(string systemName, string text) + => ByName.GetOrAdd(systemName, _ => new ConcurrentQueue()).Enqueue(text); + + public static bool TryGet(string systemName, out string text) + { + if (ByName.TryGetValue(systemName, out var q) && q.TryPeek(out var first)) + { + text = first; + return true; + } + + text = string.Empty; + return false; + } + + public static void Clear(string systemName) => ByName.TryRemove(systemName, out _); + } + + /// + /// A minimal logger installed via akka.loggers so that it is subscribed to the + /// before the Cluster extension is constructed (loggers are + /// started in ActorRefProvider.Init prior to Cluster.Get). This is the only reliable + /// way to observe the "Failed to startup Cluster" error that the startup race emits from the + /// Cluster constructor during . + /// + public sealed class ClusterStartupFuzzLogger : ReceiveActor, IRequiresMessageQueue + { + public const string FailureMarker = "Failed to startup Cluster"; + + public ClusterStartupFuzzLogger() + { + Receive(_ => Sender.Tell(new LoggerInitialized())); + Receive(error => + { + var text = error.Message?.ToString() ?? string.Empty; + if (text.IndexOf(FailureMarker, StringComparison.Ordinal) >= 0) + FuzzFailureSink.Record(Context.System.Name, error.ToString() ?? text); + }); + // ignore everything else (Warning/Info/Debug) to keep the logger mailbox light + ReceiveAny(_ => { }); + } + } + + /// + /// Randomized "startup fuzz" regression spec for the Cluster-extension startup race in + /// Cluster.cs. + /// + /// + /// The bug: cluster actors spawned during the extension constructor — the read-view event-bus + /// listener (internal dispatcher) and the SBR downing-provider actor (default dispatcher, a child of + /// ClusterCoreDaemon) — reach Cluster.ClusterCore re-entrantly from their PreStart + /// while the core is still being resolved, and block a dispatcher thread waiting for it (in + /// the original code by issuing their own redundant blocking ask). On a single-thread default + /// dispatcher this is a hard self-deadlock: the blocked thread is the only one the daemon that must + /// make progress needs, so the constructor's GetClusterCoreRef ask cannot be answered and + /// times out at akka.actor.creation-timeout. The catch block then calls Shutdown() — + /// terminating the cluster — and a subsequent JoinAsync throws + /// ("Cluster has already been terminated"). This is the exact + /// signature that killed StressSpec on 2-core Windows CI agents. + /// + /// + /// + /// Design notes: + /// + /// Single-thread default dispatcher (parallelism-min/max = 1) is the shape under + /// which a startup that blocks a dispatcher thread waiting for ClusterCore is a guaranteed + /// deadlock. Iterations that draw a 2-thread default dispatcher generally start healthily and reach + /// Up — the harness reports those passes, demonstrating it is not trivially always-failing. + /// Generous, not tiny, creation-timeout. A correct startup resolves the core in well + /// under a second even on one thread, so any value in the 4-7s range leaves ample head-room; the + /// deadlock never completes and trips regardless. The value only bounds how long each killed + /// iteration blocks. This keeps the harness targeted at the deadlock rather than at generic CPU + /// starvation (confirmed: it reproduces on an essentially idle box, loadavg ~2 on 8 cores). + /// Concurrent creation + light hammering. Systems are created concurrently and their + /// ClusterCore-dependent surface (Subscribe / SendCurrentClusterState, plus + /// actors that subscribe from PreStart) is driven from several tasks. This raises throughput/coverage + /// and jitter; the deadlock itself needs no CPU burn to trip. + /// A fix must make startup non-blocking to pass. The harness fails any implementation + /// that blocks a dispatcher thread on ClusterCore during construction — including the rejected + /// single-flight/TaskCompletionSource variant, which still blocks (just on the shared task). + /// + /// + /// + /// + /// Kills are classified immediate (the cluster was already terminated when Create + /// returned — the deadlock blocked the constructor's own ask) or delayed (Create + /// returned healthy and the cluster was terminated within the watch window by a later blocked ask). + /// Both share the same root cause and both fail the spec; the split is reported for diagnostics. + /// + /// + public sealed class ClusterStartupFuzzSpec : AkkaSpec + { + // ---- tunable knobs (consts so they are easy to sweep during tuning) ---- + private const int HardDeadlineSeconds = 52; // absolute guard vs. the 60s xunit session timeout + private const int PerIterationCapSeconds = 20; // hard cap on any single system's lifecycle + private const double WatchWindowExtraSeconds = 4.0; // watch = creation-timeout + this + + private static readonly Config BaseConfig = ConfigurationFactory.ParseString(@" + akka.loglevel = WARNING + akka.stdout-loglevel = WARNING"); + + private readonly ITestOutputHelper _output; + private readonly int _seed; + + public ClusterStartupFuzzSpec(ITestOutputHelper output) + : base(BaseConfig, output) + { + _output = output; + _seed = ResolveSeed(); + } + + private static int ResolveSeed() + { + var env = Environment.GetEnvironmentVariable("AKKA_FUZZ_SEED"); + if (!string.IsNullOrWhiteSpace(env) && + int.TryParse(env, NumberStyles.Integer, CultureInfo.InvariantCulture, out var s)) + return s; + return new Random().Next(1, int.MaxValue); + } + + // Optional env overrides used only while tuning the harness (unset in CI). + private static int? EnvInt(string key) + => int.TryParse(Environment.GetEnvironmentVariable(key), NumberStyles.Integer, + CultureInfo.InvariantCulture, out var v) ? v : (int?)null; + + private static double? EnvDouble(string key) + => double.TryParse(Environment.GetEnvironmentVariable(key), NumberStyles.Float, + CultureInfo.InvariantCulture, out var v) ? v : (double?)null; + + [Fact(DisplayName = "Should_not_terminate_a_healthy_cluster_during_concurrent_startup_hammering")] + public Task Should_not_terminate_a_healthy_cluster_during_concurrent_startup_hammering() + => RunFuzzAsync(new Profile( + name: "moderate", + aggressive: false, + concurrency: 3, + launchBudgetSeconds: 22, + maxIterations: 60)); + + [Fact(DisplayName = "Should_survive_aggressive_pinned_pool_startup_fuzz")] + public Task Should_survive_aggressive_pinned_pool_startup_fuzz() + => RunFuzzAsync(new Profile( + name: "aggressive", + aggressive: true, + concurrency: 4, + launchBudgetSeconds: 24, + maxIterations: 72)); + + private sealed record Profile( + string name, + bool aggressive, + int concurrency, + int launchBudgetSeconds, + int maxIterations); + + private sealed record FuzzParams( + int PoolDefault, + int PoolInternal, + double CreationTimeoutSeconds, + int HammerThreads, + int PreStartSubscribers, + int SpinBudgetBeforeYield, + bool MicroDelayBeforeCreate); + + private enum Outcome + { + Pass, + ReachedUp, + KillImmediate, // already terminated when Create returned (deadlock blocked the ctor's own ask) + KillDelayed, // Create returned healthy, then terminated within the watch window by a later blocked ask + CreateThrew, + SoftNoConfirm // never confirmed Up but was never killed either (machine load) + } + + private readonly record struct IterationResult(Outcome Outcome, string Detail); + + private async Task RunFuzzAsync(Profile profile) + { + var concurrency = EnvInt("AKKA_FUZZ_CONCURRENCY") ?? profile.concurrency; + var launchBudgetSeconds = EnvInt("AKKA_FUZZ_LAUNCH") ?? profile.launchBudgetSeconds; + var maxIterations = EnvInt("AKKA_FUZZ_MAXITER") ?? profile.maxIterations; + + _output.WriteLine( + $"[ClusterStartupFuzz:{profile.name}] seed={_seed} " + + $"(override with AKKA_FUZZ_SEED) concurrency={concurrency} " + + $"maxIterations={maxIterations} launchBudget={launchBudgetSeconds}s"); + + var master = new Random(_seed); + var failures = new ConcurrentQueue(); + var killImmediate = 0; + var killDelayed = 0; + var reachedUp = 0; + var passed = 0; + var soft = 0; + var completed = 0; + + using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(HardDeadlineSeconds)); + using var throttle = new SemaphoreSlim(concurrency); + var inflight = new List(); + var sw = Stopwatch.StartNew(); + + for (var iter = 0; + iter < maxIterations && + sw.Elapsed < TimeSpan.FromSeconds(launchBudgetSeconds) && + !deadline.IsCancellationRequested; + iter++) + { + try + { + await throttle.WaitAsync(deadline.Token); + } + catch (OperationCanceledException) + { + break; + } + + var iterationSeed = master.Next(); + var thisIter = iter; + inflight.Add(Task.Run(async () => + { + try + { + var result = await RunSingleSystemAsync(thisIter, iterationSeed, profile, deadline.Token) + .WaitAsync(TimeSpan.FromSeconds(PerIterationCapSeconds)); + + Interlocked.Increment(ref completed); + switch (result.Outcome) + { + case Outcome.KillImmediate: + Interlocked.Increment(ref killImmediate); + failures.Enqueue(result.Detail); + break; + case Outcome.KillDelayed: + Interlocked.Increment(ref killDelayed); + failures.Enqueue(result.Detail); + break; + case Outcome.CreateThrew: + failures.Enqueue(result.Detail); + break; + case Outcome.ReachedUp: + Interlocked.Increment(ref reachedUp); + Interlocked.Increment(ref passed); + break; + case Outcome.Pass: + Interlocked.Increment(ref passed); + break; + case Outcome.SoftNoConfirm: + Interlocked.Increment(ref soft); + break; + } + } + catch (TimeoutException) + { + Interlocked.Increment(ref soft); + } + catch (Exception ex) when (ex is OperationCanceledException or TaskCanceledException) + { + // deadline hit while tearing down — ignore + } + finally + { + throttle.Release(); + } + }, CancellationToken.None)); + } + + await Task.WhenAll(inflight); + + _output.WriteLine( + $"[ClusterStartupFuzz:{profile.name}] seed={_seed} completed={completed} in {sw.Elapsed.TotalSeconds:F1}s | " + + $"kills(delayed={killDelayed}, immediate={killImmediate}) | pass={passed} (reachedUp={reachedUp}) | soft={soft}"); + + foreach (var f in failures) + _output.WriteLine($"[ClusterStartupFuzz:{profile.name}] KILL: {f}"); + + if (!failures.IsEmpty) + { + Assert.Fail( + $"Cluster startup race reproduced: {failures.Count} healthy-cluster termination(s) " + + $"(delayed={killDelayed}, immediate={killImmediate}) across {completed} iterations. " + + $"Reproduce with AKKA_FUZZ_SEED={_seed}."); + } + + Assert.True(completed > 0, + $"Fuzz harness did not complete any iteration (seed={_seed}); machine likely overloaded."); + } + + private FuzzParams BuildParams(Random rnd, bool aggressive) + { + // default pool == 1 is the deadlock-guaranteeing shape; keep it dominant. + var poolDefault = EnvInt("AKKA_FUZZ_POOLD") ?? (rnd.Next(5) == 0 ? 2 : 1); // ~80% -> 1 + var poolInternal = EnvInt("AKKA_FUZZ_POOLI") ?? (rnd.Next(2) == 0 ? 1 : 2); + + // Generous relative to a correct startup (a single ask round-trip is sub-second even on one + // thread), yet short enough to keep throughput high. The failure is a hard deadlock that + // never completes, so it trips regardless of the exact value; the length only bounds how + // long each killed iteration blocks. + var creationTimeout = EnvDouble("AKKA_FUZZ_TIMEOUT") ?? (aggressive + ? 4.0 + rnd.NextDouble() * 2.0 // 4 - 6s + : 5.0 + rnd.NextDouble() * 2.0); // 5 - 7s + + var hammerThreads = EnvInt("AKKA_FUZZ_HAMMER") ?? + (aggressive ? 3 + rnd.Next(4) : 2 + rnd.Next(3)); // 3-6 / 2-4 + var preStartSubs = EnvInt("AKKA_FUZZ_SUBS") ?? (2 + rnd.Next(aggressive ? 5 : 3)); // 2-6 / 2-4 + // Light Phase-1 jitter only: the deadlock does not need CPU starvation to trip, and heavy + // spinning only slows throughput. + var spinBudget = EnvInt("AKKA_FUZZ_SPIN") ?? (20 + rnd.Next(160)); + var microDelay = rnd.Next(2) == 0; + + return new FuzzParams( + poolDefault, poolInternal, creationTimeout, hammerThreads, + preStartSubs, spinBudget, microDelay); + } + + private static Config BuildConfig(FuzzParams p) + { + string Fj(int parallelism) => $@"{{ + executor = fork-join-executor + fork-join-executor {{ + parallelism-min = {parallelism} + parallelism-max = {parallelism} + parallelism-factor = 1.0 + }} + }}"; + + var timeout = p.CreationTimeoutSeconds.ToString("F2", CultureInfo.InvariantCulture); + + return ConfigurationFactory.ParseString($@" + akka.loglevel = ERROR + akka.stdout-loglevel = ERROR + akka.loggers = [""Akka.Cluster.Tests.ClusterStartupFuzzLogger, Akka.Cluster.Tests""] + + akka.actor.provider = cluster + akka.actor.creation-timeout = {timeout}s + akka.actor.default-dispatcher {Fj(p.PoolDefault)} + akka.actor.internal-dispatcher {Fj(p.PoolInternal)} + + akka.remote.dot-netty.tcp.hostname = ""127.0.0.1"" + akka.remote.dot-netty.tcp.port = 0 + + akka.cluster.downing-provider-class = ""Akka.Cluster.SBR.SplitBrainResolverProvider, Akka.Cluster"" + akka.cluster.split-brain-resolver.active-strategy = keep-majority + akka.cluster.periodic-tasks-initial-delay = 100ms + akka.cluster.run-coordinated-shutdown-when-down = off + akka.coordinated-shutdown.terminate-actor-system = off + akka.coordinated-shutdown.run-by-actor-system-terminate = off + "); + } + + private async Task RunSingleSystemAsync( + int iter, int iterationSeed, Profile profile, CancellationToken ct) + { + var rnd = new Random(iterationSeed); + var p = BuildParams(rnd, profile.aggressive); + var name = $"fuzz-{_seed}-{profile.name}-{iter}"; + var config = BuildConfig(p); + var tag = + $"iter#{iter} (seed={_seed}, pools d={p.PoolDefault}/i={p.PoolInternal}, " + + $"creation-timeout={p.CreationTimeoutSeconds:F2}s, hammer={p.HammerThreads})"; + + FuzzFailureSink.Clear(name); + + ActorSystem? sys = null; + var burnCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + var sysReady = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var burners = new List(); + + try + { + // CPU burners contend for cores while ActorSystem.Create runs the Cluster construction, + // widening the window in which the constructor thread is descheduled right after the + // daemon replies but before it records the resolved core. + for (var h = 0; h < p.HammerThreads; h++) + burners.Add(Task.Run(() => BurnThenHammerAsync(sysReady.Task, p, iterationSeed + h, burnCts.Token))); + + if (p.MicroDelayBeforeCreate) + await Task.Yield(); + + try + { + sys = await Task.Run(() => ActorSystem.Create(name, config), ct); + sysReady.TrySetResult(sys); + } + catch (Exception ex) + { + sysReady.TrySetException(ex); + return new IterationResult(Outcome.CreateThrew, + $"{tag}: ActorSystem.Create threw {ex.GetType().Name}: {ex.Message}"); + } + + var cluster = Cluster.Get(sys); + var terminatedAtCreate = cluster.IsTerminated; + + // Subscriber actors on the pinned dispatchers whose PreStart subscribes to cluster + // events — mimicking SBR / the remote watcher — plus direct Subscribe / state pulls. + var sink = sys.ActorOf(Props.Create(() => new SinkActor()), "fuzz-sink"); + for (var s = 0; s < p.PreStartSubscribers; s++) + { + var dispatcher = (s % 2 == 0) + ? "akka.actor.internal-dispatcher" + : "akka.actor.default-dispatcher"; + sys.ActorOf( + Props.Create(() => new SubscriberActor()).WithDispatcher(dispatcher), + $"fuzz-sub-{s}"); + } + + for (var k = 0; k < 4; k++) + { + try + { + cluster.Subscribe(sink, typeof(ClusterEvent.IClusterDomainEvent)); + cluster.SendCurrentClusterState(sink); + } + catch + { + /* system may be shutting down — expected under the race */ + } + } + + // Brief hammer window, then quiesce burners so the (single-thread) system has room to + // either reach Up (healthy) or let the redundant-ask deadlock time out (raced). + await Task.Delay(80 + rnd.Next(160), ct); + burnCts.Cancel(); + + var watch = TimeSpan.FromSeconds(p.CreationTimeoutSeconds + WatchWindowExtraSeconds); + + using var joinCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + joinCts.CancelAfter(watch); + Task joinTask; + try + { + joinTask = cluster.JoinAsync(cluster.SelfAddress, joinCts.Token); + } + catch (ClusterJoinFailedException) + { + // Synchronous throw => already terminated when we tried to join. + joinTask = Task.CompletedTask; // don't observe it again + } + + var sw = Stopwatch.StartNew(); + Outcome outcome = Outcome.SoftNoConfirm; + string? marker = null; + + while (sw.Elapsed < watch + TimeSpan.FromSeconds(1)) + { + if (cluster.IsTerminated) + { + outcome = terminatedAtCreate ? Outcome.KillImmediate : Outcome.KillDelayed; + break; + } + + if (FuzzFailureSink.TryGet(name, out var m)) + { + marker = m; + outcome = terminatedAtCreate ? Outcome.KillImmediate : Outcome.KillDelayed; + break; + } + + if (joinTask.IsCompletedSuccessfully) + { + outcome = Outcome.ReachedUp; + break; + } + + try + { + await Task.Delay(75, ct); + } + catch (OperationCanceledException) + { + break; + } + } + + // final re-check (covers the marker/termination landing right at the window edge) + if (outcome is Outcome.SoftNoConfirm) + { + if (cluster.IsTerminated || FuzzFailureSink.TryGet(name, out marker)) + outcome = terminatedAtCreate ? Outcome.KillImmediate : Outcome.KillDelayed; + else if (joinTask.IsCompletedSuccessfully) + outcome = Outcome.ReachedUp; + else + outcome = Outcome.Pass; // survived the whole window without termination + } + + FuzzFailureSink.TryGet(name, out var finalMarker); + var markerText = marker ?? (string.IsNullOrEmpty(finalMarker) ? null : finalMarker); + + if (outcome is Outcome.KillImmediate or Outcome.KillDelayed) + { + var kind = outcome == Outcome.KillDelayed + ? "DELAYED (healthy Create, then redundant-ask deadlock Shutdown)" + : "IMMEDIATE (terminated during Create)"; + var detail = + $"{tag}: {kind}; IsTerminated={cluster.IsTerminated}; " + + $"marker={(markerText is not null)}" + + (markerText is not null ? $" [{Truncate(markerText, 220)}]" : string.Empty); + return new IterationResult(outcome, detail); + } + + return new IterationResult(outcome, tag); + } + finally + { + if (!burnCts.IsCancellationRequested) + burnCts.Cancel(); + + try + { + await Task.WhenAll(burners).WaitAsync(TimeSpan.FromSeconds(5)); + } + catch + { + /* burners are best-effort */ + } + + if (sys is not null) + { + try + { + await sys.Terminate().WaitAsync(TimeSpan.FromSeconds(10)); + } + catch + { + /* bounded teardown; a wedged system is GC'd with the test process */ + } + } + + burnCts.Dispose(); + FuzzFailureSink.Clear(name); + } + } + + private static async Task BurnThenHammerAsync( + Task sysReady, FuzzParams p, int seed, CancellationToken ct) + { + var rnd = new Random(seed); + + // Phase 1: contend for CPU while Create is running (until the system is ready or told to stop). + // SpinBudget == 0 => cooperative jitter only (yields/tiny delays), which provides scheduling + // pressure without hard-starving the single-thread pools. + while (!sysReady.IsCompleted && !ct.IsCancellationRequested) + { + if (p.SpinBudgetBeforeYield <= 0) + { + await Task.Yield(); + continue; + } + + var spin = new SpinWait(); + for (var i = 0; i < p.SpinBudgetBeforeYield && !sysReady.IsCompleted; i++) + spin.SpinOnce(); + await Task.Yield(); + } + + ActorSystem sys; + try + { + sys = await sysReady; + } + catch + { + return; + } + + IActorRef? sink = null; + try + { + sink = sys.ActorOf(Props.Create(() => new SinkActor())); + } + catch + { + // system already terminating + } + + // Phase 2: keep touching the ClusterCore-dependent surface under load until quiesced. + while (!ct.IsCancellationRequested) + { + try + { + var cluster = Cluster.Get(sys); + var target = sink ?? sys.DeadLetters; + cluster.Subscribe(target, typeof(ClusterEvent.IClusterDomainEvent)); + cluster.SendCurrentClusterState(target); + } + catch + { + // expected while the system is shutting down + } + + if (rnd.Next(3) == 0) + { + var spin = new SpinWait(); + for (var i = 0; i < 64; i++) + spin.SpinOnce(); + } + + await Task.Yield(); + } + } + + private static string Truncate(string s, int max) + => s.Length <= max ? s : s.Substring(0, max) + "..."; + + /// Subscribes to cluster events from PreStart, like SBR / the remote watcher do. + private sealed class SubscriberActor : ReceiveActor + { + public SubscriberActor() => ReceiveAny(_ => { }); + + protected override void PreStart() + => Cluster.Get(Context.System).Subscribe(Self, typeof(ClusterEvent.IClusterDomainEvent)); + } + + /// Inert message sink for Subscribe / SendCurrentClusterState targets. + private sealed class SinkActor : ReceiveActor + { + public SinkActor() => ReceiveAny(_ => { }); + } + } +} diff --git a/src/core/Akka.Cluster/Cluster.cs b/src/core/Akka.Cluster/Cluster.cs index 39b703eef00..3ab98452601 100644 --- a/src/core/Akka.Cluster/Cluster.cs +++ b/src/core/Akka.Cluster/Cluster.cs @@ -14,11 +14,11 @@ using System.Threading.Tasks; using Akka.Actor; using Akka.Actor.Internal; +using Akka.Configuration; using Akka.Event; using Akka.Remote; using Akka.Util; using Akka.Util.Internal; -using Akka.Configuration; namespace Akka.Cluster { @@ -118,7 +118,7 @@ public Cluster(ActorSystemImpl system) SelfUniqueAddress = new UniqueAddress(provider.Transport.DefaultAddress, AddressUidExtension.Uid(system)); _log = Logging.GetLogger(system, "Cluster"); - + // log a warning if the user has set auto-down-unreachable-after to any value other than "off" // obsolete setting, so suppress obsolete warning #pragma warning disable CS0618 @@ -129,7 +129,7 @@ public Cluster(ActorSystemImpl system) "The `auto-down-unreachable-after` feature has been deprecated as of Akka.NET v1.5.2 and will be removed in a future version of Akka.NET. " + "The `keep-majority` split brain resolver will be used instead. See https://getakka.net/articles/cluster/split-brain-resolver.html for more details."); } - + CurrentInfoLogger = new InfoLogger(_log, Settings, SelfAddress); @@ -146,30 +146,16 @@ public Cluster(ActorSystemImpl system) //create supervisor for daemons under path "/system/cluster" _clusterDaemons = system.SystemActorOf(Props.Create(() => new ClusterDaemon(Settings)).WithDeploy(Deploy.Local), "cluster"); - _readView = new ClusterReadView(this); + // Fire-and-forget: kick off non-blocking initialization of the cluster daemon tree. The + // daemon creates its children on receipt of this Init and hands the resolved core-daemon + // ref back via SetClusterCoreRef (which logs "Started up successfully"). The constructor + // must NEVER wait on that round-trip - see the XML doc on ClusterCore; a blocking ask + // here deadlocked small dispatcher pools for a decade. + _clusterDaemons.Tell(new InternalClusterAction.Init(this)); - // force the underlying system to start - _clusterCore = GetClusterCoreRef().Result; + _readView = new ClusterReadView(this); system.RegisterOnTermination(Shutdown); - - LogInfo("Started up successfully"); - } - - private async Task GetClusterCoreRef() - { - var timeout = System.Settings.CreationTimeout; - try - { - return await _clusterDaemons.Ask(new InternalClusterAction.GetClusterCoreRef(this), timeout).ConfigureAwait(false); - } - catch (Exception ex) - { - _log.Error(ex, "Failed to startup Cluster. You can try to increase 'akka.actor.creation-timeout'."); - Shutdown(); - System.DeadLetters.Tell(ex); //don't re-throw the error. Just log it. - return System.DeadLetters; - } } /// @@ -276,7 +262,7 @@ public Task JoinAsync(Address address, CancellationToken token = default) { if (_isTerminated.Value) throw new ClusterJoinFailedException("Cluster has already been terminated"); - + if (IsUp) return Task.CompletedTask; @@ -290,12 +276,12 @@ public Task JoinAsync(Address address, CancellationToken token = default) $"Node has not managed to join the cluster using provided address: {address}")); }); } - + RegisterOnMemberUp(() => { completion.TrySetResult(NotUsed.Instance); }); - + Join(address); return completion.Task; @@ -351,10 +337,10 @@ public Task JoinSeedNodesAsync(IEnumerable
seedNodes, CancellationToken { if (_isTerminated.Value) throw new ClusterJoinFailedException("Cluster has already been terminated"); - + if (IsUp) return Task.CompletedTask; - + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var nodes = seedNodes.ToList(); @@ -366,12 +352,12 @@ public Task JoinSeedNodesAsync(IEnumerable
seedNodes, CancellationToken $"Node has not managed to join the cluster using provided addresses: [{string.Join(",", nodes)}]")); }); } - + RegisterOnMemberUp(() => { completion.TrySetResult(NotUsed.Instance); }); - + JoinSeedNodes(nodes); return completion.Task; @@ -554,7 +540,7 @@ public ImmutableHashSet SelfRoles /// Determine whether the cluster is in the UP state. ///
public bool IsUp => SelfMember.Status is MemberStatus.Up or MemberStatus.WeaklyUp; - + /// /// The underlying supported by this plugin. /// @@ -616,23 +602,44 @@ internal void Shutdown() } private readonly IActorRef _clusterDaemons; - private IActorRef _clusterCore; + private volatile IActorRef _clusterCore; /// - /// TBD + /// INTERNAL API. + /// + /// Records the resolved cluster core-daemon once the + /// has created it. Called (once) from the daemon tree in + /// response to the message the constructor sent, + /// and emits the "Started up successfully" log line so it keeps meaning "core confirmed". /// - internal IActorRef ClusterCore + /// The resolved cluster core daemon. + internal void SetClusterCoreRef(IActorRef clusterCore) { - get - { - if (_clusterCore == null) - { - _clusterCore = GetClusterCoreRef().Result; - } - return _clusterCore; - } + _clusterCore = clusterCore; + LogInfo("Started up successfully"); } + /// + /// INTERNAL API. + /// + /// The actor that all cluster commands (Subscribe, Join, Leave, Down, ...) are sent to. + /// + /// This getter is unconditionally non-blocking and never returns null: until the daemon + /// hands the resolved core-daemon ref back via , it returns + /// the /system/cluster daemon itself. Pre- + /// sends are buffered by that daemon and, post-Init, forwarded on to the core daemon. It + /// never issues an Ask. + /// + /// + /// A blocking implementation here — an Ask bounded by + /// akka.actor.creation-timeout — deadlocked small dispatcher pools for a decade: an + /// actor spawned during startup that reached this getter parked the only thread the daemon + /// needed in order to reply, the ask timed out, and the timeout handler shut a healthy node + /// down. ClusterStartupFuzzSpec is the regression gate for that class of failure. + /// + /// + internal IActorRef ClusterCore => _clusterCore ?? _clusterDaemons; + /// /// INTERNAL API. /// @@ -678,7 +685,7 @@ internal void LogDebug(string template, object arg1) /// The message being logged. internal void LogInfo(string message) { - if(_settings.LogInfo) + if (_settings.LogInfo) _log.Info("Cluster Node [{0}] - {1}", _selfAddress, message); } diff --git a/src/core/Akka.Cluster/ClusterDaemon.cs b/src/core/Akka.Cluster/ClusterDaemon.cs index e61ec3c7def..48461fe191c 100644 --- a/src/core/Akka.Cluster/ClusterDaemon.cs +++ b/src/core/Akka.Cluster/ClusterDaemon.cs @@ -457,7 +457,7 @@ private GossipSpeedupTick() { } internal class ReapUnreachableTick : ITick { private ReapUnreachableTick() { } - public static ReapUnreachableTick Instance { get; } = new(); + public static ReapUnreachableTick Instance { get; } = new(); } /// @@ -519,11 +519,20 @@ public override int GetHashCode() } /// - /// Gets a reference to the cluster core daemon. + /// INTERNAL API. + /// + /// Fire-and-forget message sent by the extension constructor to the + /// to kick off non-blocking initialization of the cluster daemon + /// tree. Carries the (still-constructing) instance so the daemons can + /// create their children and hand the resolved core-daemon ref back to the extension via + /// . + /// + /// Local-only (), like the + /// GetClusterCoreRef ask message it replaces. /// - internal class GetClusterCoreRef: INoSerializationVerificationNeeded + internal sealed class Init : INoSerializationVerificationNeeded { - public GetClusterCoreRef(Cluster cluster) + public Init(Cluster cluster) { Cluster = cluster; } @@ -752,7 +761,7 @@ public ClusterEvent.IClusterDomainEvent Event /// /// Supervisor managing the different Cluster daemons. /// - internal sealed class ClusterDaemon : ReceiveActor, IRequiresMessageQueue + internal sealed class ClusterDaemon : ReceiveActor, IWithUnboundedStash { private Cluster _cluster; @@ -763,28 +772,47 @@ internal sealed class ClusterDaemon : ReceiveActor, IRequiresMessageQueue _clusterPromise = new(); + public IStash Stash { get; set; } + /// /// Creates a new instance of the ClusterDaemon /// /// The settings that will be used for the . public ClusterDaemon(ClusterSettings settings) { - // Important - don't use Cluster(context.system) in constructor because that would - // cause deadlock. The Cluster extension is currently being created and is waiting - // for response from GetClusterCoreRef in its constructor. - // Child actors are therefore created when GetClusterCoreRef is received + // Important - don't use Cluster(context.system) in constructor because the Cluster + // extension is still being constructed when this actor is created. Construction is now + // non-blocking: the extension sends a fire-and-forget InternalClusterAction.Init as its + // first message, and the child actors are created only when that Init is received. Any + // cluster traffic that arrives before Init (Subscribe, JoinTo, listener registrations, + // ...) is stashed by the Uninitialized behavior and replayed once initialized. _coreSupervisor = null; _settings = settings; AddCoordinatedLeave(); - Receive(msg => + Uninitialized(); + } + + private void Uninitialized() + { + Receive(msg => { - if (_coreSupervisor == null) - CreateChildren(msg.Cluster); + // assign _cluster FIRST so PreRestart can re-init after a failed initialization + _cluster = msg.Cluster; + CreateChildren(msg.Cluster); _coreSupervisor.Forward(msg); + Become(Initialized); + Stash.UnstashAll(); }); + // Buffer everything else (Subscribe / JoinTo / listener registrations, ...) until Init + // has been processed and the child tree exists. + ReceiveAny(_ => Stash.Stash()); + } + + private void Initialized() + { Receive(msg => { Context.ActorOf( @@ -806,6 +834,22 @@ public ClusterDaemon(ClusterSettings settings) // forward the Ask request so the shutdown task gets completed actor.Forward(leave); }); + + // Everything else (Subscribe, JoinTo, cluster state queries, ...) is forwarded to the + // core supervisor, which forwards it on to the core daemon. Registration order matters + // in a ReceiveActor: this catch-all MUST be registered last. + ReceiveAny(msg => _coreSupervisor.Forward(msg)); + } + + protected override void PreRestart(Exception reason, object message) + { + // Default restart semantics stop the children, so the fresh instance lands back in the + // Uninitialized behavior. The Init message that drove initialization has already been + // consumed and will not be re-delivered, so re-send it to Self to re-create the child + // tree; the Uninitialized stash catches any traffic that arrives in the meantime. + base.PreRestart(reason, message); + if (_cluster != null) + Self.Tell(new InternalClusterAction.Init(_cluster)); } private void AddCoordinatedLeave() @@ -851,7 +895,7 @@ protected override void PostStop() /// ClusterCoreDaemon and ClusterDomainEventPublisher can't be restarted because the state /// would be obsolete. Shutdown the member if any those actors crashed. /// - internal class ClusterCoreSupervisor : ReceiveActor, IRequiresMessageQueue + internal class ClusterCoreSupervisor : ReceiveActor, IWithUnboundedStash { private Cluster _cluster; @@ -860,22 +904,43 @@ internal class ClusterCoreSupervisor : ReceiveActor, IRequiresMessageQueue /// Creates a new instance of the ClusterCoreSupervisor /// public ClusterCoreSupervisor() { - // Important - don't use Cluster(Context.System) in constructor because that would - // cause deadlock. The Cluster extension is currently being created and is waiting - // for response from GetClusterCoreRef in its constructor. - // Child actors are therefore created when GetClusterCoreRef is received + // Important - don't use Cluster(Context.System) in constructor because the Cluster + // extension is still being constructed when this actor is created. Construction is now + // non-blocking: the child actors are created when InternalClusterAction.Init is received + // (forwarded from the ClusterDaemon), at which point the resolved core-daemon ref is + // handed back to the extension via Cluster.SetClusterCoreRef (replacing the old + // Sender.Tell reply to the constructor's ask). Any traffic that arrives before Init is + // stashed by the Uninitialized behavior and replayed once initialized. + + Uninitialized(); + } - Receive(cr => + private void Uninitialized() + { + Receive(msg => { - if (_coreDaemon == null) - CreateChildren(cr.Cluster); - Sender.Tell(_coreDaemon); + _cluster = msg.Cluster; + CreateChildren(msg.Cluster); + msg.Cluster.SetClusterCoreRef(_coreDaemon); + Become(Initialized); + Stash.UnstashAll(); }); + + // Defensive: the ClusterDaemon forwards Init before unstashing, so Init is normally the + // first message here, but buffer anything that races ahead of it just in case. + ReceiveAny(_ => Stash.Stash()); + } + + private void Initialized() + { + ReceiveAny(msg => _coreDaemon.Forward(msg)); } /// @@ -1628,7 +1693,7 @@ public void Welcome(Address joinWith, UniqueAddress from, Gossip gossip) public void Leaving(Address address) { // only try to update if the node is available (in the member ring) - foreach(var mem in LatestGossip.Members.Where(m => m.Address.Equals(address))) + foreach (var mem in LatestGossip.Members.Where(m => m.Address.Equals(address))) { if (mem.Status is MemberStatus.Joining or MemberStatus.WeaklyUp or MemberStatus.Up) { @@ -1636,7 +1701,7 @@ public void Leaving(Address address) var newMembers = LatestGossip.Members .Remove(mem).Add(mem.Copy(status: MemberStatus.Leaving)); var newGossip = LatestGossip.Copy(members: newMembers); - + UpdateLatestGossip(newGossip); _cluster.LogInfo("Marked address [{0}] as [{1}]", address, MemberStatus.Leaving); @@ -1698,11 +1763,11 @@ public void Downing(Address address) GossipTo(member.UniqueAddress); } } - + // if the previous statement did not evaluate to true, then this node is already being downed - + } - + if (!found) { _cluster.LogInfo("Ignoring down of unknown node [{0}]", address); @@ -1735,7 +1800,7 @@ public void Quarantined(UniqueAddress node) public void ReceiveGossipStatus(GossipStatus status) { var from = status.From; - if(!LatestGossip.HasMember(from)) + if (!LatestGossip.HasMember(from)) _cluster.LogInfo("Ignoring received gossip status from unknown [{0}]", from); else if (!LatestGossip.IsReachable(_selfUniqueAddress, from)) _cluster.LogInfo("Ignoring received gossip status from unreachable [{0}]", from); @@ -1805,9 +1870,9 @@ public ReceiveGossipType ReceiveGossip(GossipEnvelope envelope) { _cluster.LogInfo("Ignoring received gossip intended for someone else, from [{0}] to [{1}]. Our full address is [{2}]", from.Address, envelope.To, _selfUniqueAddress); - + // TODO: if the gossip is received for a version of ourselves with a different UID, do we issue a Down command? - + return ReceiveGossipType.Ignored; } if (!localGossip.HasMember(from)) @@ -1984,12 +2049,12 @@ public void GossipTick() #pragma warning restore AK1004 } } - + public void GossipSpeedupTick() { if (IsGossipSpeedupNeeded()) SendGossip(); } - + public bool IsGossipSpeedupNeeded() { return LatestGossip.Members.Any(m => m.Status == MemberStatus.Down) || @@ -2674,16 +2739,16 @@ protected override void OnReceive(object message) switch (message) { case InternalClusterAction.JoinSeenNode _: - { - //send InitJoin to all seed nodes (except myself) - foreach (var path in _otherSeeds - .Select(y => Context.ActorSelection(Context.Parent.Path.ToStringWithAddress(y)))) { - path.Tell(new InternalClusterAction.InitJoin()); + //send InitJoin to all seed nodes (except myself) + foreach (var path in _otherSeeds + .Select(y => Context.ActorSelection(Context.Parent.Path.ToStringWithAddress(y)))) + { + path.Tell(new InternalClusterAction.InitJoin()); + } + _attempts++; + break; } - _attempts++; - break; - } case InternalClusterAction.InitJoinAck initJoinAck: //first InitJoinAck reply Context.Parent.Tell(new ClusterUserAction.JoinTo(initJoinAck.Address)); @@ -2692,15 +2757,15 @@ protected override void OnReceive(object message) case InternalClusterAction.InitJoinNack _: break; //that seed was uninitialized case ReceiveTimeout _: - { - if (_attempts >= 2) - _log.Warning( - "Couldn't join seed nodes after [{0}] attempts, will try again. seed-nodes=[{1}]", - _attempts, string.Join(",", _seeds.Where(x => !x.Equals(_selfAddress)))); - //no InitJoinAck received - try again - Self.Tell(new InternalClusterAction.JoinSeenNode()); - break; - } + { + if (_attempts >= 2) + _log.Warning( + "Couldn't join seed nodes after [{0}] attempts, will try again. seed-nodes=[{1}]", + _attempts, string.Join(",", _seeds.Where(x => !x.Equals(_selfAddress)))); + //no InitJoinAck received - try again + Self.Tell(new InternalClusterAction.JoinSeenNode()); + break; + } default: Unhandled(message); break; @@ -2778,24 +2843,24 @@ protected override void OnReceive(object message) switch (message) { case InternalClusterAction.JoinSeenNode _ when _timeout.HasTimeLeft: - { - // send InitJoin to remaining seed nodes (except myself) - foreach (var seed in _remainingSeeds.Select( - x => Context.ActorSelection(Context.Parent.Path.ToStringWithAddress(x)))) - seed.Tell(new InternalClusterAction.InitJoin()); - break; - } + { + // send InitJoin to remaining seed nodes (except myself) + foreach (var seed in _remainingSeeds.Select( + x => Context.ActorSelection(Context.Parent.Path.ToStringWithAddress(x)))) + seed.Tell(new InternalClusterAction.InitJoin()); + break; + } case InternalClusterAction.JoinSeenNode _: - { - if (_log.IsDebugEnabled) { - _log.Debug("Couldn't join other seed nodes, will join myself. seed-nodes=[{0}]", string.Join(",", _seeds)); + if (_log.IsDebugEnabled) + { + _log.Debug("Couldn't join other seed nodes, will join myself. seed-nodes=[{0}]", string.Join(",", _seeds)); + } + // no InitJoinAck received, initialize new cluster by joining myself + Context.Parent.Tell(new ClusterUserAction.JoinTo(_selfAddress)); + Context.Stop(Self); + break; } - // no InitJoinAck received, initialize new cluster by joining myself - Context.Parent.Tell(new ClusterUserAction.JoinTo(_selfAddress)); - Context.Stop(Self); - break; - } case InternalClusterAction.InitJoinAck initJoinAck: _log.Info("Received InitJoinAck message from [{0}] to [{1}]", initJoinAck.Address, _selfAddress); // first InitJoinAck reply, join existing cluster @@ -2803,18 +2868,18 @@ protected override void OnReceive(object message) Context.Stop(Self); break; case InternalClusterAction.InitJoinNack initJoinNack: - { - _log.Info("Received InitJoinNack message from [{0}] to [{1}]", initJoinNack.Address, _selfAddress); - _remainingSeeds = _remainingSeeds.Remove(initJoinNack.Address); - if (_remainingSeeds.IsEmpty) { - // initialize new cluster by joining myself when nacks from all other seed nodes - Context.Parent.Tell(new ClusterUserAction.JoinTo(_selfAddress)); - Context.Stop(Self); - } + _log.Info("Received InitJoinNack message from [{0}] to [{1}]", initJoinNack.Address, _selfAddress); + _remainingSeeds = _remainingSeeds.Remove(initJoinNack.Address); + if (_remainingSeeds.IsEmpty) + { + // initialize new cluster by joining myself when nacks from all other seed nodes + Context.Parent.Tell(new ClusterUserAction.JoinTo(_selfAddress)); + Context.Stop(Self); + } - break; - } + break; + } default: Unhandled(message); break; From d8fa2e6d85df32e3513cdfcc67019d57cd501207 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 10 Jul 2026 20:34:30 +0000 Subject: [PATCH 2/2] Address slopwatch SW003 findings in ClusterStartupFuzzSpec Suppress six SW003 (empty catch block) findings in the fuzz spec's teardown/best-effort paths via slopwatch-ignore comments with justifications. These catches are intentional: cancellation on the hard test deadline, best-effort probing of a cluster mid-race, and bounded/best-effort teardown of burner tasks and the actor system. No behavior change. --- src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs b/src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs index 59695443a58..3d24b56d8db 100644 --- a/src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs +++ b/src/core/Akka.Cluster.Tests/ClusterStartupFuzzSpec.cs @@ -286,6 +286,7 @@ private async Task RunFuzzAsync(Profile profile) { Interlocked.Increment(ref soft); } + // slopwatch-ignore: SW003 Hard-deadline cancellation while an iteration is tearing down; the outcome was already recorded before this fires, so there is nothing left to handle. catch (Exception ex) when (ex is OperationCanceledException or TaskCanceledException) { // deadline hit while tearing down — ignore @@ -444,6 +445,7 @@ private async Task RunSingleSystemAsync( cluster.Subscribe(sink, typeof(ClusterEvent.IClusterDomainEvent)); cluster.SendCurrentClusterState(sink); } + // slopwatch-ignore: SW003 The cluster may already be terminated by the startup race under test; this retry loop is best-effort probing and the actual outcome is captured via FuzzFailureSink/IsTerminated below. catch { /* system may be shutting down — expected under the race */ @@ -542,6 +544,7 @@ private async Task RunSingleSystemAsync( { await Task.WhenAll(burners).WaitAsync(TimeSpan.FromSeconds(5)); } + // slopwatch-ignore: SW003 Draining the CPU-burner tasks is best-effort cleanup; a burner exception or the 5s WaitAsync timeout must never fail the fuzz iteration itself. catch { /* burners are best-effort */ @@ -553,6 +556,7 @@ private async Task RunSingleSystemAsync( { await sys.Terminate().WaitAsync(TimeSpan.FromSeconds(10)); } + // slopwatch-ignore: SW003 Teardown is intentionally bounded and best-effort; a system wedged by the deadlock under test is abandoned here and GC'd with the test process rather than failing the spec. catch { /* bounded teardown; a wedged system is GC'd with the test process */ @@ -601,6 +605,7 @@ private static async Task BurnThenHammerAsync( { sink = sys.ActorOf(Props.Create(() => new SinkActor())); } + // slopwatch-ignore: SW003 The target system may already be terminating from the deadlock under test; failing to create the sink actor here is expected and the hammer loop below falls back to DeadLetters. catch { // system already terminating @@ -616,6 +621,7 @@ private static async Task BurnThenHammerAsync( cluster.Subscribe(target, typeof(ClusterEvent.IClusterDomainEvent)); cluster.SendCurrentClusterState(target); } + // slopwatch-ignore: SW003 This loop hammers ClusterCore concurrently with the startup race under test, so the target system may be mid-shutdown; failures here are expected and the loop simply retries. catch { // expected while the system is shutting down