[CI Visibility] Fix message loss and abandoned mutexes in the IPC channel

tracer/build/_build/Build.Steps.cs

@@ -2722,15 +2722,10 @@ string NormalizedPath(AbsolutePath ap)
                new(@".*Timeout occurred when flushing spans.*", RegexOptions.Compiled),
                new(@".*TestOptimization: .*", RegexOptions.Compiled),
                new(@".*TestOptimizationClient: .*", RegexOptions.Compiled),
-               // TODO: for the CI Visibility team to fix. Under the .NET 11 SDK a sample process exits while
-               // holding the CircularChannel mutex. CircularChannel.Reader.InternalPollForMessage catches the
-               // resulting AbandonedMutexException and returns _without_ releasing - but an abandoned wait still
-               // acquires - so the channel is poisoned and every subsequent poll logs an error. The same
-               // WaitOne-outside-try shape in CircularChannel.Writer.TryWrite and the CircularChannel ctor lets
-               // the exception escape entirely, which produces the third pattern. The tests themselves pass.
-               new(@".*CircularChannel\.(Reader|Writer): Mutex was abandoned.*", RegexOptions.Compiled),
+               // With the .NET 11 RC SDK, a .NET 11 process and a .NET 8-10 process sharing the CircularChannel named mutex
+               // corrupt its ownership record (dotnet/runtime#134491), so ReleaseMutex can fail after a successful wait.
+               // Remove this once we build with .NET 11 GA.
                new(@".*CircularChannel\.Reader: Error while polling for messages.*Object synchronization method was called from an unsynchronized block of code.*", RegexOptions.Compiled | RegexOptions.Singleline),
-               new(@".*Error enabling IPC client and sending coverage data.*AbandonedMutexException.*", RegexOptions.Compiled | RegexOptions.Singleline),
                // This one is annoying but we _think_ due to a dodgy named pipes implementation, so ignoring for now
                new(@".*An error occurred while sending data to the agent at \\\\\.\\pipe\\trace-.*The operation has timed out.*", RegexOptions.Compiled),
                new(@".*An error occurred while sending data to the agent at \\\\\.\\pipe\\metrics-.*The operation has timed out.*", RegexOptions.Compiled),

tracer/src/Datadog.Trace/Ci/Ipc/CircularChannel.Reader.cs

@@ -35,8 +35,12 @@ public Reader(CircularChannel channel)
 
         private void PollForMessages(object? state)
         {
-            while (!_pollingThreadFinishEvent.IsSet)
+            // Poll once more after Dispose is requested, so messages written before it are not lost.
+            // Dispose joins this thread before the channel releases the memory mapped file.
+            var finishing = false;
+            while (!finishing)
             {
+                finishing = _pollingThreadFinishEvent.IsSet;
                 try
                 {
                     InternalPollForMessage();
@@ -46,20 +50,15 @@ private void PollForMessages(object? state)
                     Log.Error(ex, "CircularChannel.Reader: Error while polling for messages (InternalPollForMessage)");
                 }
 
-                if (!_pollingThreadFinishEvent.IsSet)
+                if (!finishing)
                 {
-                    Thread.Sleep(_channel._settings.PollingInterval);
+                    _pollingThreadFinishEvent.Wait(_channel._settings.PollingInterval);
                 }
             }
         }
 
         private void InternalPollForMessage()
         {
-            if (Interlocked.Read(ref _disposed) == 1 || Interlocked.Read(ref _channel._disposed) == 1)
-            {
-                return;
-            }
-
             if (_callback is null)
             {
                 // To avoid losing messages, we stop the polling if there are no subscribers
@@ -68,19 +67,12 @@ private void InternalPollForMessage()
 
             try
             {
-                var hasHandle = _channel._mutex.WaitOne(_channel._settings.MutexTimeout);
-                if (!hasHandle)
+                if (!_channel.TryAcquireMutex())
                 {
                     Log.Error("CircularChannel.Reader: Failed to acquire mutex within the time limit.");
                     return;
                 }
             }
-            catch (AbandonedMutexException ex)
-            {
-                // WaitOne grants ownership even when the previous owner abandoned the mutex.
-                // Continue through the finally below so other processes can send messages.
-                Log.Warning(ex, "CircularChannel.Reader: Mutex was abandoned. Continuing with ownership.");
-            }
             catch (ObjectDisposedException ex)
             {
                 // The mutex was disposed, nothing to do
@@ -96,10 +88,8 @@ private void InternalPollForMessage()
                 var readPos = accessor.ReadUInt16(2);
                 while (readPos != writePos)
                 {
-                    // Check if we had to use a virtual write position outside the buffer to avoid blocking the read position
-                    // condition for read is: writepos != readpos
-                    // So if we detect that we have a writepos > buffersize, we use the modulus to check if is the same to the readpos
-                    // and detect the buffer overflow.
+                    // Older versions marked a completely full buffer with a virtual write position past the end of the
+                    // buffer. The writer never does that anymore, but normalize it in case an older version shares this channel.
                     if (writePos >= _channel.BufferBodySize)
                     {
                         // This means that the read position moved to the next buffer, so we need to reset the write position

tracer/src/Datadog.Trace/Ci/Ipc/CircularChannel.Writer.cs

@@ -34,14 +34,15 @@ public bool TryWrite(in ArraySegment<byte> data)
             }
 
             var dataSize = GetMessageSize(in data);
-            if (dataSize > channel.BufferBodySize)
+
+            // One byte of the buffer always stays free (see the space check below)
+            if (dataSize >= channel.BufferBodySize)
             {
                 Log.Error("CircularChannel.Writer: Message size exceeds maximum allowed size.");
                 return false;
             }
 
-            var hasHandle = channel._mutex.WaitOne(_channel._settings.MutexTimeout);
-            if (!hasHandle)
+            if (!channel.TryAcquireMutex())
             {
                 Log.Error("CircularChannel.Writer: Failed to acquire mutex within the time limit.");
                 return false;
@@ -53,10 +54,10 @@ public bool TryWrite(in ArraySegment<byte> data)
                 var writePos = accessor.ReadUInt16(0);
                 var readPos = accessor.ReadUInt16(2);
 
-                // Check if we had to use a virtual write position outside the buffer to avoid blocking the read position
-                // condition for read is: writepos != readpos
-                // So if we detect that we have a writepos > buffersize, we use the modulus to check if is the same to the readpos
-                // and detect the buffer overflow.
+                // Older versions marked a completely full buffer with a virtual write position past the end of the buffer
+                // (BufferBodySize + writePos), because writePos == readPos means empty. That overflows the ushort once the
+                // body is larger than 32 KB, so we never write it anymore (see the space check below), but we still
+                // understand it in case an older version shares this channel.
                 if (writePos >= channel.BufferBodySize)
                 {
                     if (writePos % channel.BufferBodySize == readPos)
@@ -101,7 +102,10 @@ readPos          writePos
                 var spaceAvailable = writePos < readPos
                                          ? readPos - writePos
                                          : channel.BufferBodySize - (writePos - readPos);
-                if (spaceAvailable < dataSize)
+
+                // Always leave at least one byte free, so the write position never catches up with the read position.
+                // Otherwise a full buffer would look exactly like an empty one (writePos == readPos).
+                if (spaceAvailable <= dataSize)
                 {
                     Log.Warning("CircularChannel.Writer: Buffer overflow");
                     return false;
@@ -123,12 +127,6 @@ readPos          writePos
                     accessor.WriteArray(HeaderSize, data.Array!, firstPartLength, secondPartLength);
                 }
 
-                if (nextWritePos == readPos)
-                {
-                    // This means that we will overwrite the data in the next write, so we need to virtually use a position outside the buffer
-                    nextWritePos = (ushort)(channel.BufferBodySize + nextWritePos);
-                }
-
                 accessor.Write(0, nextWritePos); // Update write pointer
                 return true;
             }

tracer/src/Datadog.Trace/Ci/Ipc/CircularChannel.cs

@@ -76,8 +76,7 @@ public CircularChannel(string fileName, CircularChannelSettings settings)
             initiallyOwned: false,
             FrameworkDescription.Instance.IsWindows() ? @$"Global\{Path.GetFileNameWithoutExtension(fileName)}" : $"{Path.GetFileNameWithoutExtension(fileName)}");
 
-        var hasHandle = _mutex.WaitOne(_settings.MutexTimeout);
-        if (!hasHandle)
+        if (!TryAcquireMutex())
         {
             throw new TimeoutException("CircularChannel: Failed to acquire mutex within the time limit.");
         }
@@ -87,16 +86,24 @@ public CircularChannel(string fileName, CircularChannelSettings settings)
             // Let's open or create the file we want to map
             var stream = new FileStream(fileName, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.ReadWrite);
 
-            // Ensure we have the correct size
-            stream.SetLength(_settings.BufferSize);
+            // Only a file we have just created, or one with an unexpected size, needs initializing. The other side of
+            // the channel may already be using an existing one, and resetting its pointers drops every unread message.
+            var initialize = stream.Length != _settings.BufferSize;
+            if (initialize)
+            {
+                stream.SetLength(_settings.BufferSize);
+            }
 
             // Create the memory mapped file from the stream
             _mmf = MemoryMappedFile.CreateFromFile(stream, mapName: null, _settings.BufferSize, MemoryMappedFileAccess.ReadWrite, HandleInheritability.None, leaveOpen: false);
 
-            // Initialize the write and read pointer
-            using var accessor = _mmf.CreateViewAccessor();
-            accessor.Write(0, (ushort)0); // Write pointer
-            accessor.Write(2, (ushort)0); // Read pointer
+            if (initialize)
+            {
+                // Initialize the write and read pointer
+                using var accessor = _mmf.CreateViewAccessor();
+                accessor.Write(0, (ushort)0); // Write pointer
+                accessor.Write(2, (ushort)0); // Read pointer
+            }
         }
         finally
         {
@@ -130,4 +137,22 @@ public void Dispose()
         _mmf.Dispose();
         _mutex.Dispose();
     }
+
+    /// <summary>
+    /// Waits for the channel mutex. WaitOne grants ownership even when the previous owner abandoned the mutex,
+    /// so an abandoned mutex counts as acquired and the caller must release it as usual.
+    /// </summary>
+    /// <returns>False when the mutex wasn't acquired within the configured timeout.</returns>
+    private bool TryAcquireMutex()
+    {
+        try
+        {
+            return _mutex.WaitOne(_settings.MutexTimeout);
+        }
+        catch (AbandonedMutexException ex)
+        {
+            Log.Warning(ex, "CircularChannel: Mutex was abandoned. Continuing with ownership.");
+            return true;
+        }
+    }
 }