Fix thread safety issues in WorkerNodeTelemetryData

src/Build.UnitTests/Telemetry/Telemetry_Tests.cs

@@ -6,11 +6,14 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Build.BackEnd.Logging;
 using Microsoft.Build.Execution;
 using Microsoft.Build.Framework;
 using Microsoft.Build.Framework.Telemetry;
 using Microsoft.Build.TelemetryInfra;
 using Microsoft.Build.UnitTests;
+using Microsoft.Build.UnitTests.BackEnd;
 using Shouldly;
 using Xunit;
 using Xunit.Abstractions;
@@ -470,5 +473,67 @@ public void BuildIncrementalityInfo_NoTargets_ClassifiedAsUnknown()
             incrementality.TotalTargetsCount.ShouldBe(0);
             incrementality.IncrementalityRatio.ShouldBe(0.0);
         }
+
+        [Fact]
+        public void IsEmpty_TrueForDefault_FalseAfterAdd()
+        {
+            var data = new WorkerNodeTelemetryData();
+            data.IsEmpty.ShouldBeTrue();
+
+            var targetKey = new TaskOrTargetTelemetryKey("Target1", isCustom: false, isFromNugetCache: false, isFromMetaProject: false);
+            data.AddTarget(targetKey, wasExecuted: true);
+            data.IsEmpty.ShouldBeFalse();
+        }
+
+        [Fact]
+        public void FinalizeProcessing_AfterMerge_ResetsState()
+        {
+            var forwarder = new TelemetryForwarderProvider.TelemetryForwarder();
+            var loggingService = new EventRecordingLoggingService();
+
+            var loggingContext = new MockLoggingContext(
+                loggingService,
+                new BuildEventContext(1, 2, BuildEventContext.InvalidProjectContextId, 4));
+
+            // Merge some data.
+            var localData = new WorkerNodeTelemetryData();
+            var key = new TaskOrTargetTelemetryKey("TestTarget", isCustom: true, isFromNugetCache: false, isFromMetaProject: false);
+            localData.AddTarget(key, wasExecuted: true);
+            forwarder.MergeWorkerData(localData);
+
+            // First FinalizeProcessing should emit a telemetry event.
+            forwarder.FinalizeProcessing(loggingContext);
+            var telemetryEvents = loggingService.RecordedEvents.OfType<WorkerNodeTelemetryEventArgs>().ToList();
+            telemetryEvents.Count.ShouldBe(1);
+            telemetryEvents[0].WorkerNodeTelemetryData.TargetsExecutionData.ShouldContainKey(key);
+
+            // Second FinalizeProcessing on an empty forwarder should be a no-op (state was reset).
+            forwarder.FinalizeProcessing(loggingContext);
+            loggingService.RecordedEvents.OfType<WorkerNodeTelemetryEventArgs>().Count().ShouldBe(1, "No new event should be emitted after reset");
+
+            // Merge new data after reset — forwarder should still work.
+            var localData2 = new WorkerNodeTelemetryData();
+            var key2 = new TaskOrTargetTelemetryKey("TestTarget2", isCustom: false, isFromNugetCache: false, isFromMetaProject: false);
+            localData2.AddTarget(key2, wasExecuted: false, skipReason: TargetSkipReason.ConditionWasFalse);
+            forwarder.MergeWorkerData(localData2);
+
+            // Third FinalizeProcessing should emit only the new data.
+            forwarder.FinalizeProcessing(loggingContext);
+            var allTelemetryEvents = loggingService.RecordedEvents.OfType<WorkerNodeTelemetryEventArgs>().ToList();
+            allTelemetryEvents.Count.ShouldBe(2);
+            allTelemetryEvents[1].WorkerNodeTelemetryData.TargetsExecutionData.ShouldContainKey(key2);
+            allTelemetryEvents[1].WorkerNodeTelemetryData.TargetsExecutionData.ShouldNotContainKey(key, "Old data should not appear after reset");
+        }
+
+        /// <summary>
+        /// <see cref="MockLoggingService"/> that records all <see cref="ILoggingService.LogBuildEvent"/> calls
+        /// so tests can inspect emitted build events.
+        /// </summary>
+        private sealed class EventRecordingLoggingService : MockLoggingService, ILoggingService
+        {
+            public List<BuildEventArgs> RecordedEvents { get; } = [];
+
+            void ILoggingService.LogBuildEvent(BuildEventArgs buildEvent) => RecordedEvents.Add(buildEvent);
+        }
     }
 }

src/Build/BackEnd/Components/RequestBuilder/RequestBuilder.cs

@@ -19,6 +19,7 @@
 using Microsoft.Build.Experimental.BuildCheck;
 using Microsoft.Build.Experimental.BuildCheck.Infrastructure;
 using Microsoft.Build.Framework;
+using Microsoft.Build.Framework.Telemetry;
 using Microsoft.Build.Internal;
 using Microsoft.Build.Shared;
 using Microsoft.Build.Shared.Debugging;
@@ -1285,6 +1286,9 @@ private void UpdateStatisticsPostBuild()
                 return;
             }
 
+            // Accumulate all telemetry into a local instance, then merge into the shared singleton once.
+            WorkerNodeTelemetryData telemetryData = new();
+
             foreach (var projectTargetInstance in _requestEntry.RequestConfiguration.Project.Targets)
             {
                 bool wasExecuted =
@@ -1315,20 +1319,16 @@ private void UpdateStatisticsPostBuild()
                                (isFromNuget && FileClassifier.Shared.IsMicrosoftPackageInNugetCache(projectTargetInstance.Value.FullPath));
                 }
 
-                telemetryForwarder.AddTarget(
-                    projectTargetInstance.Key,
-                    // would we want to distinguish targets that were executed only during this execution - we'd need
-                    //  to remember target names from ResultsByTarget from before execution
-                    wasExecuted,
-                    isCustom,
-                    isMetaprojTarget,
-                    isFromNuget,
-                    skipReason);
+                var key = new TaskOrTargetTelemetryKey(
+                    projectTargetInstance.Key, isCustom, isFromNuget, isMetaprojTarget);
+                telemetryData.AddTarget(key, wasExecuted, skipReason);
             }
 
             TaskRegistry taskReg = _requestEntry.RequestConfiguration.Project.TaskRegistry;
             CollectTasksStats(taskReg);
 
+            telemetryForwarder.MergeWorkerData(telemetryData);
+
             void CollectTasksStats(TaskRegistry taskRegistry)
             {
                 if (taskRegistry == null)
@@ -1338,13 +1338,16 @@ void CollectTasksStats(TaskRegistry taskRegistry)
 
                 foreach (TaskRegistry.RegisteredTaskRecord registeredTaskRecord in taskRegistry.TaskRegistrations.Values.SelectMany(record => record))
                 {
-                    telemetryForwarder.AddTask(
+                    var key = new TaskOrTargetTelemetryKey(
                         registeredTaskRecord.TaskIdentity.Name,
+                        registeredTaskRecord.ComputeIfCustom(),
+                        registeredTaskRecord.IsFromNugetCache,
+                        isFromMetaProject: false);
+                    telemetryData.AddTask(
+                        key,
                         registeredTaskRecord.Statistics.ExecutedTime,
                         registeredTaskRecord.Statistics.ExecutedCount,
                         registeredTaskRecord.Statistics.TotalMemoryConsumption,
-                        registeredTaskRecord.ComputeIfCustom(),
-                        registeredTaskRecord.IsFromNugetCache,
                         registeredTaskRecord.TaskFactoryAttributeName,
                         registeredTaskRecord.TaskFactoryParameters.Runtime);
 

src/Build/TelemetryInfra/ITelemetryForwarder.cs

@@ -1,9 +1,8 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
-using System;
 using Microsoft.Build.BackEnd.Logging;
-using Microsoft.Build.Framework;
+using Microsoft.Build.Framework.Telemetry;
 
 namespace Microsoft.Build.TelemetryInfra;
 
@@ -15,26 +14,13 @@ internal interface ITelemetryForwarder
 {
     bool IsTelemetryCollected { get; }
 
-    void AddTask(
-        string name,
-        TimeSpan cumulativeExecutionTime,
-        short executionsCount,
-        long totalMemoryConsumed,
-        bool isCustom,
-        bool isFromNugetCache,
-        string? taskFactoryName,
-        string? taskHostRuntime);
-
     /// <summary>
-    /// Add info about target execution to the telemetry.
+    /// Merges a batch of telemetry data into this forwarder's accumulated state.
     /// </summary>
-    /// <param name="name">The target name.</param>
-    /// <param name="wasExecuted">Whether the target was executed (not skipped).</param>
-    /// <param name="isCustom">Whether this is a custom target.</param>
-    /// <param name="isMetaproj">Whether the target is from a meta project.</param>
-    /// <param name="isFromNugetCache">Whether the target is from a NuGet package.</param>
-    /// <param name="skipReason">The reason the target was skipped, if applicable.</param>
-    void AddTarget(string name, bool wasExecuted, bool isCustom, bool isMetaproj, bool isFromNugetCache, TargetSkipReason skipReason = TargetSkipReason.None);
+    void MergeWorkerData(IWorkerNodeTelemetryData data);
 
+    /// <summary>
+    /// Sends accumulated telemetry and resets internal state.
+    /// </summary>
     void FinalizeProcessing(LoggingContext loggingContext);
 }

src/Build/TelemetryInfra/TelemetryForwarderProvider.cs

@@ -1,10 +1,8 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
-using System;
 using Microsoft.Build.BackEnd;
 using Microsoft.Build.BackEnd.Logging;
-using Microsoft.Build.Framework;
 using Microsoft.Build.Framework.Telemetry;
 using Microsoft.Build.Shared;
 
@@ -49,32 +47,48 @@ public void ShutdownComponent()
         _instance = null;
     }
 
+    /// <summary>
+    /// Active telemetry forwarder that accumulates worker node telemetry.
+    /// </summary>
+    /// <remarks>
+    /// Thread-safe: in /m /mt mode, multiple <see cref="BuildRequestEngine"/> instances share a single
+    /// <see cref="TelemetryForwarderProvider"/> singleton, so <see cref="MergeWorkerData"/> and
+    /// <see cref="FinalizeProcessing"/> may be called concurrently from different node threads.
+    /// </remarks>
     public class TelemetryForwarder : ITelemetryForwarder
     {
-        private readonly WorkerNodeTelemetryData _workerNodeTelemetryData = new();
+        private WorkerNodeTelemetryData _workerNodeTelemetryData = new();
+        private readonly LockType _lock = new();
 
         // in future, this might be per event type
         public bool IsTelemetryCollected => true;
 
-        public void AddTask(string name, TimeSpan cumulativeExecutionTime, short executionsCount, long totalMemoryConsumed, bool isCustom, bool isFromNugetCache, string? taskFactoryName, string? taskHostRuntime)
+        public void MergeWorkerData(IWorkerNodeTelemetryData data)
         {
-            var key = GetKey(name, isCustom, false, isFromNugetCache);
-            _workerNodeTelemetryData.AddTask(key, cumulativeExecutionTime, executionsCount, totalMemoryConsumed, taskFactoryName, taskHostRuntime);
-        }
-
-        public void AddTarget(string name, bool wasExecuted, bool isCustom, bool isMetaproj, bool isFromNugetCache, TargetSkipReason skipReason = TargetSkipReason.None)
-        {
-            var key = GetKey(name, isCustom, isMetaproj, isFromNugetCache);
-            _workerNodeTelemetryData.AddTarget(key, wasExecuted, skipReason);
+            lock (_lock)
+            {
+                _workerNodeTelemetryData.Add(data);
+            }
         }
 
-        private static TaskOrTargetTelemetryKey GetKey(string name, bool isCustom, bool isMetaproj,
-            bool isFromNugetCache)
-            => new TaskOrTargetTelemetryKey(name, isCustom, isFromNugetCache, isMetaproj);
 
         public void FinalizeProcessing(LoggingContext loggingContext)
         {
-            WorkerNodeTelemetryEventArgs telemetryArgs = new(_workerNodeTelemetryData)
+            WorkerNodeTelemetryData snapshot;
+
+            lock (_lock)
+            {
+                // Nothing accumulated since the last call — skip sending.
+                if (_workerNodeTelemetryData.IsEmpty)
+                {
+                    return;
+                }
+
+                snapshot = _workerNodeTelemetryData;
+                _workerNodeTelemetryData = new();
+            }
+
+            WorkerNodeTelemetryEventArgs telemetryArgs = new(snapshot)
             { BuildEventContext = loggingContext.BuildEventContext };
             loggingContext.LogBuildEvent(telemetryArgs);
         }
@@ -84,9 +98,7 @@ public class NullTelemetryForwarder : ITelemetryForwarder
     {
         public bool IsTelemetryCollected => false;
 
-        public void AddTask(string name, TimeSpan cumulativeExecutionTime, short executionsCount, long totalMemoryConsumed, bool isCustom, bool isFromNugetCache, string? taskFactoryName, string? taskHostRuntime) { }
-
-        public void AddTarget(string name, bool wasExecuted, bool isCustom, bool isMetaproj, bool isFromNugetCache, TargetSkipReason skipReason = TargetSkipReason.None) { }
+        public void MergeWorkerData(IWorkerNodeTelemetryData data) { }
 
         public void FinalizeProcessing(LoggingContext loggingContext) { }
     }

src/Framework/Telemetry/WorkerNodeTelemetryData.cs

@@ -14,6 +14,9 @@ public WorkerNodeTelemetryData(Dictionary<TaskOrTargetTelemetryKey, TaskExecutio
         TargetsExecutionData = targetsExecutionData;
     }
 
+    /// <summary>
+    /// Merges all data from another <see cref="IWorkerNodeTelemetryData"/> into this instance.
+    /// </summary>
     public void Add(IWorkerNodeTelemetryData other)
     {
         foreach (var task in other.TasksExecutionData)
@@ -27,10 +30,12 @@ public void Add(IWorkerNodeTelemetryData other)
         }
     }
 
+    /// <summary>
+    /// Adds or aggregates task execution data.
+    /// </summary>
     public void AddTask(TaskOrTargetTelemetryKey task, TimeSpan cumulativeExecutionTime, int executionsCount, long totalMemoryConsumption, string? factoryName, string? taskHostRuntime)
     {
-        TaskExecutionStats? taskExecutionStats;
-        if (!TasksExecutionData.TryGetValue(task, out taskExecutionStats))
+        if (!TasksExecutionData.TryGetValue(task, out TaskExecutionStats? taskExecutionStats))
         {
             taskExecutionStats = new(cumulativeExecutionTime, executionsCount, totalMemoryConsumption, factoryName, taskHostRuntime);
             TasksExecutionData[task] = taskExecutionStats;
@@ -45,6 +50,9 @@ public void AddTask(TaskOrTargetTelemetryKey task, TimeSpan cumulativeExecutionT
         }
     }
 
+    /// <summary>
+    /// Adds or updates target execution data.
+    /// </summary>
     public void AddTarget(TaskOrTargetTelemetryKey target, bool wasExecuted, TargetSkipReason skipReason = TargetSkipReason.None)
     {
         if (TargetsExecutionData.TryGetValue(target, out var existingStats))
@@ -71,6 +79,8 @@ public void AddTarget(TaskOrTargetTelemetryKey target, bool wasExecuted, TargetS
 
     public WorkerNodeTelemetryData() : this([], []) { }
 
+    public bool IsEmpty => TasksExecutionData.Count == 0 && TargetsExecutionData.Count == 0;
+
     public Dictionary<TaskOrTargetTelemetryKey, TaskExecutionStats> TasksExecutionData { get; }
 
     public Dictionary<TaskOrTargetTelemetryKey, TargetExecutionStats> TargetsExecutionData { get; }