[Tracing] Fix Hangfire baggage accumulation (#8900)

## Summary of changes

Fixes baggage accumulating across sequential Hangfire jobs processed by
the same worker. `DatadogHangfireServerFilter.OnPerforming` now replaces
`Baggage.Current` with the job's own extracted baggage instead of
merging into it, saving whatever baggage was ambient beforehand;
`OnPerformed` restores that saved baggage after the job completes.

## Reason for change

`Baggage.Current` is a mutable reference type stored in an `AsyncLocal`.
Hangfire workers process jobs sequentially on one persistent
thread/`ExecutionContext` (unlike, e.g., one `ExecutionContext` per HTTP
request), so that `AsyncLocal` is never naturally reset between jobs.
`OnPerforming` was merging each job's extracted baggage into the
existing `Baggage.Current` instance rather than replacing it, and
`OnPerformed` never reset it, so every job's baggage accumulated into
whatever the previous job on that worker had left behind. Over time this
produces oversized `baggage` headers on downstream HTTP calls, which can
be rejected by services enforcing header size limits.

## Implementation details

In `DatadogHangfireServerFilter.cs`:
- `OnPerforming`: save the current `Baggage.Current` into the job's
`Items` bag, then replace it with `propagationContext.Baggage ?? new
Baggage()` (the job's own extracted baggage) instead of merging into the
ambient instance.
- `OnPerformed`: restore the baggage saved in `OnPerforming` (in a
`finally` block, so it runs even if scope disposal throws), since the
worker's `ExecutionContext` is persistent across jobs. This mirrors the
save/replace/restore pattern used by OpenTelemetry's own Hangfire
instrumentation.

## Test coverage

Added `HangfireTests.BaggageDoesNotAccumulateAcrossSequentialJobs`,
which runs two sequential jobs on a single-worker Hangfire server, each
enqueued from its own isolated producer context with distinct baggage,
and asserts that the second job's `Baggage.Current` contains only its
own entry (not the first job's). The test also asserts both jobs ran on
the same worker thread, confirming the
single-worker/persistent-`ExecutionContext` precondition that makes the
bug reproducible.

## Other details
Fixes issue
[#8895](#8895)

---------

Co-authored-by: Steven Bouwkamp <steven.bouwkamp@datadoghq.com>
Co-authored-by: Zach Montoya <zach.montoya@datadoghq.com>

Author: 3 people

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Hangfire/DatadogHangfireServerFilter.cs

@@ -29,8 +29,14 @@ public void OnPerforming(object context)
                 return;
             }
 
+            // Hangfire workers reuse one ExecutionContext across sequential jobs, so save/restore
+            // baggage around each job instead of leaking it into the next one (see OnPerformed).
+            performingContext.Items[HangfireConstants.DatadogBaggageKey] = Baggage.Current;
+
             var spanContextData = performingContext.GetJobParameter<Dictionary<string, string?>?>(HangfireConstants.DatadogContextKey);
-            PropagationContext propagationContext = Tracer.Instance.TracerManager.SpanContextPropagator.Extract(spanContextData).MergeBaggageInto(Baggage.Current);
+            var propagationContext = Tracer.Instance.TracerManager.SpanContextPropagator.Extract(spanContextData);
+            Baggage.Current = propagationContext.Baggage ?? new Baggage();
+
             var parentContext = propagationContext.SpanContext;
             Scope? scope = HangfireCommon.CreateScope(Tracer.Instance, new HangfireTags(), performingContext, parentContext);
             ((Dictionary<string, object?>)performingContext.Items).Add(HangfireConstants.DatadogScopeKey, scope);
@@ -45,19 +51,29 @@ public void OnPerformed(object context)
         {
             if (context.TryDuckCast<IPerformedContextProxy>(out var performedContext))
             {
-                if (performedContext.Items.TryGetValue(HangfireConstants.DatadogScopeKey, out var scope))
+                try
                 {
-                    if (scope is not Scope typedScope)
+                    if (performedContext.Items.TryGetValue(HangfireConstants.DatadogScopeKey, out var scope))
                     {
-                        return;
-                    }
+                        if (scope is not Scope typedScope)
+                        {
+                            return;
+                        }
+
+                        if (performedContext.Exception is not null)
+                        {
+                            HangfireCommon.SetStatusAndRecordException(typedScope, performedContext.Exception);
+                        }
 
-                    if (performedContext.Exception is not null)
+                        typedScope.Dispose();
+                    }
+                }
+                finally
+                {
+                    if (performedContext.Items.TryGetValue(HangfireConstants.DatadogBaggageKey, out var previousBaggage))
                     {
-                        HangfireCommon.SetStatusAndRecordException(typedScope, performedContext.Exception);
+                        Baggage.Current = previousBaggage is Baggage baggage ? baggage : new Baggage();
                     }
-
-                    typedScope.Dispose();
                 }
             }
         }

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Hangfire/HangfireConstants.cs

@@ -14,6 +14,7 @@ internal static class HangfireConstants
     internal const string OnPerformOperation = "hangfire.perform";
     internal const string DatadogScopeKey = "datadog_scope_key";
     internal const string DatadogContextKey = "datadog_context_key";
+    internal const string DatadogBaggageKey = "datadog_baggage_key";
     internal const string JobIdTag = "job.id";
     internal const string JobCreatedAtTag = "job.createdat";
     internal const string ResourceNamePrefix = "job ";

tracer/test/Datadog.Trace.ClrProfiler.IntegrationTests/HangfireTests.cs

@@ -29,6 +29,25 @@ public HangfireTests(ITestOutputHelper output)
 
     public override Result ValidateIntegrationSpan(MockSpan span, string metadataSchemaVersion) => span.IsHangfire(metadataSchemaVersion);
 
+    [SkippableFact]
+    [Trait("Category", "EndToEnd")]
+    [Trait("RunOnWindows", "True")]
+    public async Task BaggageDoesNotAccumulateAcrossSequentialJobs()
+    {
+        using var agent = EnvironmentHelper.GetMockAgent();
+        using var process = await RunSampleAndWaitForExit(agent, arguments: "baggage");
+
+        var workerThreadIds = Regex.Matches(process.StandardOutput, @"Worker thread for job-(?:one|two): (\d+)")
+                                   .Cast<Match>()
+                                   .Select(x => x.Groups[1].Value)
+                                   .ToArray();
+
+        workerThreadIds.Should().HaveCount(2);
+        workerThreadIds.Distinct().Should().ContainSingle();
+        process.StandardOutput.Should().Contain("Baggage for job-one: [job-one=one]");
+        process.StandardOutput.Should().Contain("Baggage for job-two: [job-two=two]");
+    }
+
     [SkippableFact]
     [Trait("Category", "EndToEnd")]
     [Trait("RunOnWindows", "True")]

tracer/test/test-applications/integrations/Samples.Hangfire/Jobs/TestJob.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Linq;
 using System.Threading.Tasks;
 
 namespace Samples.Hangfire.Jobs;
@@ -11,6 +12,16 @@ public Task Execute()
         return Task.CompletedTask;
     }
 
+    public void PrintBaggage(string jobName)
+    {
+        var baggage = OpenTelemetry.Baggage.Current.GetBaggage()
+                                   .OrderBy(x => x.Key)
+                                   .Select(x => $"{x.Key}={x.Value}");
+
+        Console.WriteLine($"Worker thread for {jobName}: {Environment.CurrentManagedThreadId}");
+        Console.WriteLine($"Baggage for {jobName}: [{string.Join(", ", baggage)}]");
+    }
+
     public Task ThrowException()
     {
         Console.WriteLine("TestJob.ThrowException running...");

tracer/test/test-applications/integrations/Samples.Hangfire/Program.cs

@@ -30,7 +30,8 @@ public static async Task Main(string[] args)
         GlobalJobFilters.Filters.Add(new JobCompletionFilter());
 
         Console.WriteLine("Starting Hangfire server...");
-        using var server = new BackgroundJobServer();
+        var isBaggageScenario = args.Length > 0 && args[0] == "baggage";
+        using var server = CreateServer(isBaggageScenario);
 
         try
         {
@@ -43,6 +44,13 @@ public static async Task Main(string[] args)
                 kind: ActivityKind.Internal,
                 tags: tags
             );
+
+            if (isBaggageScenario)
+            {
+                await Should_Not_Accumulate_Baggage_Across_Jobs();
+                return;
+            }
+
             // run tests
             await Should_Create_Span();
             await Should_Create_Span_With_Status_Error_When_Job_Failed();
@@ -54,6 +62,50 @@ public static async Task Main(string[] args)
         }
     }
 
+    private static BackgroundJobServer CreateServer(bool suppressExecutionContext)
+    {
+        if (!suppressExecutionContext)
+        {
+            return new BackgroundJobServer();
+        }
+
+        // Keep the worker's mutable AsyncLocal baggage instance isolated from the in-process producers.
+        using (ExecutionContext.SuppressFlow())
+        {
+            return new BackgroundJobServer(new BackgroundJobServerOptions { WorkerCount = 1 });
+        }
+    }
+
+    private static async Task Should_Not_Accumulate_Baggage_Across_Jobs()
+    {
+        await Run_Baggage_Job("job-one", "one");
+        await Run_Baggage_Job("job-two", "two");
+    }
+
+    private static async Task Run_Baggage_Job(string baggageKey, string baggageValue)
+    {
+        // Model an independent producer request, with its own execution context and baggage instance.
+        Task<string> enqueueTask;
+        using (ExecutionContext.SuppressFlow())
+        {
+            enqueueTask = Task.Run(() =>
+            {
+                using var activity = AdditionalActivitySource.StartActivity($"Enqueue {baggageKey}");
+                OpenTelemetry.Baggage.Current = OpenTelemetry.Baggage.Create(new Dictionary<string, string>
+                {
+                    [baggageKey] = baggageValue,
+                });
+
+                return BackgroundJob.Enqueue<TestJob>(x => x.PrintBaggage(baggageKey));
+            });
+        }
+
+        var jobId = await enqueueTask;
+        var result = await JobCompletion.Register(jobId, new CancellationTokenSource(TimeSpan.FromSeconds(30)).Token);
+
+        Console.WriteLine($"Job {result.JobId} completed successfully = {result.Succeeded}");
+    }
+
     private static async Task Should_Create_Span()
     {
         var jobId = BackgroundJob.Enqueue<TestJob>(x => x.Execute());

tracer/test/test-applications/integrations/Samples.Hangfire/Samples.Hangfire.csproj

@@ -12,7 +12,8 @@
   <ItemGroup>
     <PackageReference Include="Hangfire" Version="$(ApiVersion)" />
     <PackageReference Include="Hangfire.MemoryStorage" Version="1.4.0" />
+    <PackageReference Include="OpenTelemetry.Api" Version="1.10.0" />
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="9.0.8" />
   </ItemGroup>
 
-</Project>
+</Project>