Add cross-thread stress tests and fix benchmark setup race

Add three targeted concurrency tests that exercise the exact cross-thread
tag write pattern the JMH crossThread benchmark was measuring:
- crossThreadSustainedNoCrash: 8 threads × 10k setTag on same span
- ownerToSharedTransition: owner writes first, then 8 threads join
- manySpansCrossThread: 10k short-lived spans tagged from 8 threads

All pass, proving the production code handles cross-thread writes without
NPE or structural corruption.

Fix the crossThread benchmark: change SharedSpan @setup from
Level.Invocation to Level.Iteration. With Level.Invocation, 8 threads
raced to call setup() concurrently, causing NPE when state.span was
transiently null between invocations — a benchmark harness bug, not a
production code bug.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: bm1549

dd-trace-core/src/jmh/java/datadog/trace/core/SpanTagBenchmark.java

@@ -39,7 +39,7 @@ public void setup() {
   public static class SharedSpan {
     AgentSpan span;
 
-    @Setup(Level.Invocation)
+    @Setup(Level.Iteration)
     public void setup() {
       span = TRACER.startSpan("benchmark", "tag-benchmark-shared");
     }

dd-trace-core/src/test/java/datadog/trace/core/DDSpanContextConcurrencyTest.java

@@ -159,4 +159,151 @@ void mixedMetricsAndTags() {
 
     span.finish();
   }
+
+  /**
+   * Reproduces the exact benchmark pattern: one span created by thread A, 8 threads calling setTag
+   * concurrently. This is the pattern that the JMH crossThread benchmark uses, except the benchmark
+   * failed because of a race in the JMH harness (SharedSpan.setup with Level.Invocation + 8
+   * threads), not in the production code.
+   *
+   * <p>This test proves the production code handles this pattern without NPE or structural
+   * corruption.
+   */
+  @Test
+  @DisplayName(
+      "Cross-thread sustained: 8 threads setTag on same span for 10k iterations — no crash")
+  void crossThreadSustainedNoCrash() throws Exception {
+    int numThreads = 8;
+    int iterationsPerThread = 10_000;
+    ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+
+    // Create span on main thread, then hand it to 8 other threads
+    AgentSpan span = TRACER.startSpan("test", "crossSustained");
+    CyclicBarrier barrier = new CyclicBarrier(numThreads);
+    CopyOnWriteArrayList<Throwable> errors = new CopyOnWriteArrayList<>();
+
+    List<Future<?>> futures = new ArrayList<>();
+    for (int t = 0; t < numThreads; t++) {
+      futures.add(
+          executor.submit(
+              () -> {
+                try {
+                  barrier.await(5, TimeUnit.SECONDS);
+                  for (int i = 0; i < iterationsPerThread; i++) {
+                    span.setTag("key", "value");
+                  }
+                } catch (Throwable e) {
+                  errors.add(e);
+                }
+              }));
+    }
+
+    for (Future<?> f : futures) {
+      f.get(30, TimeUnit.SECONDS);
+    }
+    span.finish();
+    executor.shutdown();
+
+    assertEquals(0, errors.size(), "Unexpected errors during cross-thread setTag: " + errors);
+    assertEquals("value", span.getTag("key"));
+  }
+
+  /**
+   * Tests the transition from owner-thread to shared mode under concurrent writes: the creating
+   * thread sets tags first, then 8 other threads join in. This exercises the STATE_OWNER →
+   * STATE_SHARED transition while writes are in flight.
+   */
+  @Test
+  @DisplayName("Owner-to-shared transition under concurrent writes — no crash, all tags present")
+  void ownerToSharedTransition() throws Exception {
+    int numThreads = 8;
+    int tagsPerThread = 1_000;
+    ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+
+    AgentSpan span = TRACER.startSpan("test", "ownerToShared");
+
+    // Owner thread writes first batch — these are on the fast path (no lock)
+    for (int i = 0; i < 100; i++) {
+      span.setTag("owner_" + i, "val_" + i);
+    }
+
+    // Now launch 8 threads that also write — these trigger the transition to STATE_SHARED
+    CyclicBarrier barrier = new CyclicBarrier(numThreads);
+    CopyOnWriteArrayList<Throwable> errors = new CopyOnWriteArrayList<>();
+
+    List<Future<?>> futures = new ArrayList<>();
+    for (int t = 0; t < numThreads; t++) {
+      final int threadIdx = t;
+      futures.add(
+          executor.submit(
+              () -> {
+                try {
+                  barrier.await(5, TimeUnit.SECONDS);
+                  for (int i = 0; i < tagsPerThread; i++) {
+                    span.setTag("thread" + threadIdx + "_" + i, "v" + i);
+                  }
+                } catch (Throwable e) {
+                  errors.add(e);
+                }
+              }));
+    }
+
+    for (Future<?> f : futures) {
+      f.get(30, TimeUnit.SECONDS);
+    }
+    span.finish();
+    executor.shutdown();
+
+    assertEquals(0, errors.size(), "Unexpected errors during transition: " + errors);
+
+    // Owner-thread tags should all be present — they were written before any contention
+    for (int i = 0; i < 100; i++) {
+      assertEquals("val_" + i, span.getTag("owner_" + i), "Owner tag owner_" + i + " missing");
+    }
+
+    // Each thread's last write should be visible (earlier writes may be overwritten by races)
+    for (int t = 0; t < numThreads; t++) {
+      assertNotNull(
+          span.getTag("thread" + t + "_" + (tagsPerThread - 1)),
+          "Thread " + t + " last tag missing");
+    }
+  }
+
+  /**
+   * Exercises many short-lived spans created on one thread and tagged from another — the exact
+   * pattern the crossThread benchmark was trying to measure. Uses a stable handoff (CountDownLatch)
+   * instead of the racy JMH setup.
+   */
+  @Test
+  @DisplayName("Many short spans tagged cross-thread — no NPE or crash")
+  void manySpansCrossThread() throws Exception {
+    int numSpans = 10_000;
+    ExecutorService tagger = Executors.newFixedThreadPool(8);
+    CopyOnWriteArrayList<Throwable> errors = new CopyOnWriteArrayList<>();
+
+    for (int s = 0; s < numSpans; s++) {
+      AgentSpan span = TRACER.startSpan("test", "manyShort");
+      CountDownLatch tagged = new CountDownLatch(8);
+
+      for (int t = 0; t < 8; t++) {
+        tagger.submit(
+            () -> {
+              try {
+                span.setTag("key", "value");
+              } catch (Throwable e) {
+                errors.add(e);
+              } finally {
+                tagged.countDown();
+              }
+            });
+      }
+
+      tagged.await(5, TimeUnit.SECONDS);
+      span.finish();
+    }
+
+    tagger.shutdown();
+    tagger.awaitTermination(10, TimeUnit.SECONDS);
+    assertEquals(0, errors.size(), "Errors during cross-thread tagging of short spans: " + errors);
+  }
 }