diff --git a/eng/run-windows-chaos-with-dump.ps1 b/eng/run-windows-chaos-with-dump.ps1 index 2d28640f..8c52a69c 100644 --- a/eng/run-windows-chaos-with-dump.ps1 +++ b/eng/run-windows-chaos-with-dump.ps1 @@ -18,6 +18,7 @@ for ($attempt = 1; $attempt -le $Attempts; $attempt++) { $info.UseShellExecute = $false $info.RedirectStandardOutput = $true $info.RedirectStandardError = $true + $info.Environment['SHARPLINK_CHAOS_DUMP_TOOL'] = $collector foreach ($argument in @($dll, '--duration-seconds', '120', '--transport', 'sharedmemory', '--concurrency', '32', '--restart-interval-seconds', '10', '--json-output', $report)) { $info.ArgumentList.Add($argument) } @@ -28,16 +29,8 @@ for ($attempt = 1; $attempt -le $Attempts; $attempt++) { $failure = $null try { while (-not $process.WaitForExit(1000)) { - if (Test-Path $report) { - try { - $checkpoint = Get-Content $report -Raw | ConvertFrom-Json - if ($checkpoint.UnexpectedFailures -gt 0) { - $failure = "Unexpected chaos failure: $($checkpoint.TerminalFailure.Message)" - } - } catch { - # The application atomically rewrites the checkpoint; re-read on the next tick. - } - } + # The application captures unexpected failures before shutdown. This + # collector is the fallback for a process that cannot reach that path. if ($clock.Elapsed.TotalSeconds -gt 180) { $failure = 'Chaos exceeded its 120-second workload plus 60-second shutdown bound.' } diff --git a/src/SharpLink.Runtime/RpcSession.SendPump.cs b/src/SharpLink.Runtime/RpcSession.SendPump.cs index 00778c4b..666eb895 100644 --- a/src/SharpLink.Runtime/RpcSession.SendPump.cs +++ b/src/SharpLink.Runtime/RpcSession.SendPump.cs @@ -428,10 +428,9 @@ private async ValueTask WaitForMoreUntilFlushBoundaryAsync( long batchStartTimestamp, int bytesAccumulated) { - // Queue publication and policy publication share one wake authority. A policy - // generation change is the only wake that is consumed internally: it restarts the - // decision from the original batch start. An ordinary data wake keeps the static - // pump's established behavior and returns to the outer control loop immediately. + // Queue publication and policy publication share one wake authority. Every wake + // rechecks the queue, policy, stop state, and original batch deadline: a producer + // may signal only after the pump has already consumed its published frame. while (true) { if (HasProgressFrames() || HasNormalFrames()) @@ -470,11 +469,10 @@ private async ValueTask WaitForMoreUntilFlushBoundaryAsync( if (!ReferenceEquals(policy, _flushPolicyState.Capture())) continue; - // Preserve the pre-runtime static pump contract: a data wake returns to the outer - // loop. If the queue was already drained by the time it is observed, the outer - // queue check falls through to the same immediate flush behavior as before #590. + // A wake is a request to recheck state, not an independent flush boundary. + // Delayed signals for already consumed frames must not flush a timed batch. if (woke) - return true; + continue; if (remaining <= MaximumTimerDelay) return false; diff --git a/src/SharpLink.Runtime/Transport/SharedMemoryMapping.cs b/src/SharpLink.Runtime/Transport/SharedMemoryMapping.cs index 499fd738..79b4f326 100644 --- a/src/SharpLink.Runtime/Transport/SharedMemoryMapping.cs +++ b/src/SharpLink.Runtime/Transport/SharedMemoryMapping.cs @@ -251,6 +251,11 @@ private void Dispose() _view.SafeMemoryMappedViewHandle.ReleasePointer(); _pointer = null; } + // This mapping carries transient IPC data; peers observe the shared pages directly. + // Close the view handle before the accessor so its Dispose does not explicitly flush + // the backing file. On Windows that flush can block/retry for seconds and exhaust + // the server shutdown budget, even though this file is deleted when peers close. + _view.SafeMemoryMappedViewHandle.Dispose(); _view.Dispose(); _mappedFile.Dispose(); _file.Dispose(); diff --git a/test/SharpLink.ChaosTests/Program.cs b/test/SharpLink.ChaosTests/Program.cs index ae38f2ee..1d19a089 100644 --- a/test/SharpLink.ChaosTests/Program.cs +++ b/test/SharpLink.ChaosTests/Program.cs @@ -153,10 +153,21 @@ public static async Task Main(string[] args) await restarter.ConfigureAwait(false); await memorySampler.ConfigureAwait(false); + await AwaitDiagnosticCaptureAsync().ConfigureAwait(false); + phase = "StoppingClient"; await client.StopAsync().ConfigureAwait(false); phase = "StoppingServer"; - serverStops.Enqueue(await server.StopAsync("FinalStop").ConfigureAwait(false)); + try + { + serverStops.Enqueue(await server.StopAsync("FinalStop").ConfigureAwait(false)); + } + catch (Exception exception) + { + RecordUnexpectedFailure(exception); + await AwaitDiagnosticCaptureAsync().ConfigureAwait(false); + throw; + } phase = "DrainingMetrics"; var drain = await metrics.WaitForZeroAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false); if (options.InjectUnobservedTaskException) @@ -176,11 +187,7 @@ public static async Task Main(string[] args) clientLogs.InjectErrorForGateProbe("client"); if (options.InjectServerError) serverLogs.InjectErrorForGateProbe("server"); - Task? activeDiagnosticCapture; - lock (diagnosticGate) - activeDiagnosticCapture = diagnosticCaptureTask; - if (activeDiagnosticCapture is not null) - diagnosticArtifact = await activeDiagnosticCapture.ConfigureAwait(false); + await AwaitDiagnosticCaptureAsync().ConfigureAwait(false); var exitCode = 0; ChaosFailure? terminalFailure = null; if (!drain.Drained) @@ -282,6 +289,15 @@ public static async Task Main(string[] args) Console.WriteLine($"CHAOS_SERVER_ERROR {error}"); return exitCode; + async Task AwaitDiagnosticCaptureAsync() + { + Task? activeDiagnosticCapture; + lock (diagnosticGate) + activeDiagnosticCapture = diagnosticCaptureTask; + if (activeDiagnosticCapture is not null) + diagnosticArtifact = await activeDiagnosticCapture.ConfigureAwait(false); + } + async Task RestartLoopAsync() { try @@ -849,7 +865,10 @@ private static async Task CaptureProcessDumpAsync(strin : Path.ChangeExtension(Path.GetFullPath(reportPath), ".dmp"); Directory.CreateDirectory(Path.GetDirectoryName(dumpPath)!); var executableName = OperatingSystem.IsWindows() ? "createdump.exe" : "createdump"; - var toolPath = Path.Combine( + var externalTool = OperatingSystem.IsWindows() + ? Environment.GetEnvironmentVariable("SHARPLINK_CHAOS_DUMP_TOOL") + : null; + var toolPath = externalTool ?? Path.Combine( System.Runtime.InteropServices.RuntimeEnvironment.GetRuntimeDirectory(), executableName); if (!File.Exists(toolPath)) @@ -870,12 +889,18 @@ private static async Task CaptureProcessDumpAsync(strin UseShellExecute = false, CreateNoWindow = true }; - info.ArgumentList.Add("--withheap"); - info.ArgumentList.Add("--crashreport"); - info.ArgumentList.Add("--name"); - info.ArgumentList.Add(dumpPath); - info.ArgumentList.Add(Environment.ProcessId.ToString( - System.Globalization.CultureInfo.InvariantCulture)); + var processId = Environment.ProcessId.ToString( + System.Globalization.CultureInfo.InvariantCulture); + if (externalTool is not null) + { + foreach (var argument in new[] { "collect", "--process-id", processId, "--type", "Heap", "--output", dumpPath }) + info.ArgumentList.Add(argument); + } + else + { + foreach (var argument in new[] { "--withheap", "--crashreport", "--name", dumpPath, processId }) + info.ArgumentList.Add(argument); + } using var process = Process.Start(info); if (process is null) { diff --git a/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.ModuleLifecycle.cs b/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.ModuleLifecycle.cs index 9fa5cb08..98bd1dd9 100644 --- a/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.ModuleLifecycle.cs +++ b/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.ModuleLifecycle.cs @@ -77,6 +77,7 @@ public async Task RuntimeAssembliesShouldRegisterTransactionallyAndSupportEveryC Ensure(duplicate.Error?.IncomingLoadContext?.Contains("dynamic-call-shapes", StringComparison.Ordinal) == true, "duplicate diagnostics contain ALC identity"); + await WaitForRemoteContractManifestAsync(harness.Client, plugin.ContractType); object? proxy = GetProxy(harness.Client, plugin.ContractType); var unary = await InvokeValueTaskAsync(proxy, plugin.ContractType, "UnaryAsync", 7, CancellationToken.None); Ensure(unary == 8, "dynamic unary"); diff --git a/test/SharpLink.UnitTests/Hosting/SharpLinkClientAccessorTests.cs b/test/SharpLink.UnitTests/Hosting/SharpLinkClientAccessorTests.cs index 3758bc96..0e1f68a5 100644 --- a/test/SharpLink.UnitTests/Hosting/SharpLinkClientAccessorTests.cs +++ b/test/SharpLink.UnitTests/Hosting/SharpLinkClientAccessorTests.cs @@ -59,7 +59,10 @@ public async Task ConcurrentPublicationMustNotResurrectClientAfterStop() { const int attempts = 100_000; using var start = new Barrier(3); - using var workersCancellation = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + // Keep all 100,000 race rounds. Hosted macOS runners can spend over 30 seconds + // scheduling these barriers alongside the full suite; this bounds test workers, + // not any production lifecycle operation. + using var workersCancellation = new CancellationTokenSource(TimeSpan.FromMinutes(2)); var accessor = new SharpLinkClientAccessor(); var client = new FakeSharpLinkClient(); Exception? publicationFailure = null; diff --git a/test/SharpLink.UnitTests/Runtime/RpcSessionRuntimeFlushPolicyTests.cs b/test/SharpLink.UnitTests/Runtime/RpcSessionRuntimeFlushPolicyTests.cs index ca94f9c7..b194b1c3 100644 --- a/test/SharpLink.UnitTests/Runtime/RpcSessionRuntimeFlushPolicyTests.cs +++ b/test/SharpLink.UnitTests/Runtime/RpcSessionRuntimeFlushPolicyTests.cs @@ -1,11 +1,64 @@ using System.Diagnostics; using System.IO.Pipelines; +using System.Reflection; namespace SharpLink.UnitTests.Runtime; [NotInParallel] public sealed class RpcSessionRuntimeFlushPolicyTests { + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task TimedBatchShouldIgnoreDelayedDataWakeAfterQueueWasDrained(bool publishRuntimePolicy) + { + var clock = new ManualTimeProvider(); + var provider = new TimerCountingTimeProvider(clock); + var initial = new RpcSessionFlushOptions(1024, TimeSpan.FromSeconds(30)); + using var context = new SharpLinkRuntimeContextBuilder() + .UseTimeProvider(provider) + .Build(includeGeneratedAssemblyCatalog: false); + var owner = CompressionSendPolicyState.CreateInitial(new SharpLinkCompressionSendPolicy()); + var policy = owner.GetOrCreateSessionFlushPolicyState(initial, context.PerformanceProfile); + if (publishRuntimePolicy) + Ensure(policy.Publish(4096, TimeSpan.FromSeconds(30)), "runtime threshold publication"); + var input = new Pipe(); + var output = new Pipe(); + var session = CreateSession("delayed-data-wake", context, owner, initial, input, output); + try + { + var readTask = output.Reader.ReadAsync().AsTask(); + session.SendPacket(CreateFrame(session, 64, 1)); + await WaitUntilAsync(() => provider.TimerCount > 0); + var armedCount = provider.TimerCount; + var pump = typeof(RpcSession).GetField("_pump", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(session)!; + var wakeup = (WakeupSignal)pump.GetType() + .GetField("_wakeup", BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(pump)!; + + // Model an enqueuer delayed between queue publication and Signal: the + // pump already consumed its frame through a previous wake. + wakeup.Signal(); + await WaitUntilAsync(() => readTask.IsCompleted || provider.TimerCount > armedCount); + Ensure(!readTask.IsCompleted && session.QueuedSendBytes > 0, + "a delayed data wake must preserve the timed batch and retained frame"); + + clock.Advance(TimeSpan.FromSeconds(30)); + var read = await readTask.WaitAsync(TimeSpan.FromSeconds(2)); + Ensure(read.Buffer.Length == ProtocolV2Constants.HeaderBytes + 64, + "the original deadline must publish the complete retained frame"); + output.Reader.AdvanceTo(read.Buffer.End); + await session.FlushSendQueueAsync(); + Ensure(session.QueuedSendBytes == 0, "the deadline flush must release queued byte ownership"); + } + finally + { + await session.DisposeAsync(); + await output.Reader.CompleteAsync(); + await input.Writer.CompleteAsync(); + } + } + [Test] public async Task ThresholdDecreaseShouldWakeArmedExistingSessionAndFutureSessionShouldShareGeneration() { @@ -176,6 +229,20 @@ public async Task LatencyIncreaseShouldIgnoreOldTimerBoundary() } } + private sealed class TimerCountingTimeProvider(ManualTimeProvider clock) : TimeProvider + { + private int _timerCount; + internal int TimerCount => Volatile.Read(ref _timerCount); + public override long TimestampFrequency => clock.TimestampFrequency; + public override long GetTimestamp() => clock.GetTimestamp(); + public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period) + { + var timer = clock.CreateTimer(callback, state, dueTime, period); + Interlocked.Increment(ref _timerCount); + return timer; + } + } + private static RpcSession CreateSession( string id, SharpLinkRuntimeContext context, diff --git a/test/SharpLink.UnitTests/Runtime/SharedMemoryLayoutTests.cs b/test/SharpLink.UnitTests/Runtime/SharedMemoryLayoutTests.cs index 4f8c8858..041d0d9c 100644 --- a/test/SharpLink.UnitTests/Runtime/SharedMemoryLayoutTests.cs +++ b/test/SharpLink.UnitTests/Runtime/SharedMemoryLayoutTests.cs @@ -127,6 +127,47 @@ public async Task MappingFileShouldDisappearAfterBothSidesClose() } } + [Test] + [Arguments(true)] + [Arguments(false)] + [NotInParallel] + public async Task DisposingOneMappingShouldPreservePeerMemoryAndInvalidateOwnedMemory(bool closeServerFirst) + { + const int capacity = 64 * 1024; + var baseline = SharedMemoryMapping.ActiveMappingCount; + var nonce = RandomNumberGenerator.GetBytes(SharedMemoryLayout.NonceBytes); + var server = SharedMemoryMapping.CreateServer(capacity, nonce, out var path); + var client = SharedMemoryMapping.OpenClient(path, capacity, nonce); + try + { + server.UnlinkAfterClientOpened(); + var owner = closeServerFirst ? server : client; + var peer = closeServerFirst ? client : server; + var ownedMemory = owner.Memory.Slice(SharedMemoryLayout.HeaderBytes, 64); + var peerMemory = peer.Memory.Slice(SharedMemoryLayout.HeaderBytes, 64); + var expected = Enumerable.Range(1, 64).Select(value => (byte)value).ToArray(); + expected.CopyTo(ownedMemory); + + await owner.DisposeAsync(); + + await Assert.That(peerMemory.ToArray().SequenceEqual(expected)).IsTrue(); + peerMemory.Span[0] = 123; + await Assert.That(peerMemory.Span[0]).IsEqualTo((byte)123); + await Assert.That(() => ownedMemory.Span[0]).Throws(); + await Assert.That(SharedMemoryMapping.ActiveMappingCount).IsEqualTo(baseline + 1); + + await peer.DisposeAsync(); + await owner.DisposeAsync(); + await Assert.That(SharedMemoryMapping.ActiveMappingCount).IsEqualTo(baseline); + await Assert.That(File.Exists(path)).IsFalse(); + } + finally + { + await client.DisposeAsync(); + await server.DisposeAsync(); + } + } + [Test] public async Task MappingPathShouldRejectLocationsOutsideTransportDirectory() {