[DSM] Guard against future timestamps when extracting pathway context (#8672)

## Summary of changes

Guards against future timestamps when extracting pathwayContext and
calculating `pathwayLatencyNs`

## Reason for change

We decode the `dd-pathway-ctx` from inbound headers, and use the decoded
`PathwayStart` to compute `pathwayLatencyNs`. This is then used to
calculate an `originTimestamp` and is added to `_originBuckets`. We then
only flush `_originBuckets` for which the start time is <= the current
bucket window.

If a "future" timestamp ends up in `PathwayStart`, then we would end up
with that future bucket never being flushed and accumulating. This PR
guards against those future timestamps so we don't end up with unbounded
growth.

## Implementation details

Add two guards:
- In `SetCheckpoint`, guard against generating negative
`pathwayLatencyNs` or `edgeLatencyNs`
- In `PathwayContextEncoder.Decode`, reject data points with a "future"
`PathwayStart` time
- Using a tolerance here of 60s to make sure clock skew issues don't
bite us. I _think_ that number seems reasonable, but open to
suggestions.
- I was initially concerned about daylight savings etc, but given we're
using UTC for all of this, that shouldn't be an issue

## Test coverage

Added a bunch of additional unit tests around this.

Author: andrewlock

tracer/src/Datadog.Trace/DataStreamsMonitoring/DataStreamsManager.cs

@@ -325,15 +325,20 @@ public void InjectPathwayContextAsBase64String<TCarrier>(PathwayContext? context
             var parentHash = previousContext?.Hash ?? default;
             var pathwayHash = HashHelper.CalculatePathwayHash(nodeHash, parentHash);
 
+            // Prevent negative value latencies from a future-dated upstream timestamp
+            // (clock skew or a buggy/malicious peer)
+            var pathwayLatencyNs = Math.Max(0, nowNs - pathwayStartNs);
+            var edgeLatencyNs = Math.Max(0, nowNs - (previousContext?.EdgeStart ?? edgeStartNs));
+
             var writer = Volatile.Read(ref _writer);
             writer?.Add(
                 new StatsPoint(
                     edgeTags: edgeTags,
                     hash: pathwayHash,
                     parentHash: parentHash,
                     timestampNs: nowNs,
-                    pathwayLatencyNs: nowNs - pathwayStartNs,
-                    edgeLatencyNs: nowNs - (previousContext?.EdgeStart ?? edgeStartNs),
+                    pathwayLatencyNs: pathwayLatencyNs,
+                    edgeLatencyNs: edgeLatencyNs,
                     payloadSizeBytes));
 
             var pathway = new PathwayContext(

tracer/src/Datadog.Trace/DataStreamsMonitoring/PathwayContextEncoder.cs

@@ -21,6 +21,12 @@ internal static class PathwayContextEncoder
     /// <summary>Maximum byte length of the Base64 encoding of <see cref="MaxEncodedSize"/> bytes: ceil(26/3)*4 = 36.</summary>
     internal const int MaxBase64EncodedSize = 36;
 
+    /// <summary>
+    /// Tolerance (in nanoseconds) applied when validating that decoded pathway/edge timestamps
+    /// are not in the future. Allows for normal clock drift between hosts.
+    /// </summary>
+    internal const long MaxClockSkewNs = 60L * 1_000_000_000L;
+
     /// <summary>
     /// Encodes a <see cref="PathwayContext"/> as a series of bytes
     /// NOTE: the encoding is lossy, in that we convert <see cref="PathwayContext.PathwayStart"/>
@@ -108,6 +114,15 @@ public static int EncodeInto(PathwayContext pathway, Span<byte> buffer)
             return null;
         }
 
+        var maxAllowedNs = DateTimeOffset.UtcNow.ToUnixTimeNanoseconds() + MaxClockSkewNs;
+        if (pathwayStartNs > maxAllowedNs || edgeStartNs > maxAllowedNs)
+        {
+            Log.Warning(
+                "Error decoding Data Stream PathwayContext from bytes {Base64EncodedBytes}: pathway or edge start is in the future",
+                Convert.ToBase64String(bytes));
+            return null;
+        }
+
         // Pathway context values are in ns
         return new PathwayContext(new PathwayHash(hash), pathwayStartNs, edgeStartNs);
     }
@@ -152,6 +167,13 @@ public static int EncodeInto(PathwayContext pathway, Span<byte> buffer)
             return null;
         }
 
+        var maxAllowedNs = DateTimeOffset.UtcNow.ToUnixTimeNanoseconds() + MaxClockSkewNs;
+        if (pathwayStartNs > maxAllowedNs || edgeStartNs > maxAllowedNs)
+        {
+            Log.Warning("Error decoding Data Stream PathwayContext from bytes: pathway or edge start is in the future");
+            return null;
+        }
+
         return new PathwayContext(new PathwayHash(hash), pathwayStartNs, edgeStartNs);
     }
 #endif

tracer/test/Datadog.Trace.Tests/DataStreamsMonitoring/PathwayContextEncoderTests.cs

@@ -6,7 +6,9 @@
 using System;
 using Datadog.Trace.DataStreamsMonitoring;
 using Datadog.Trace.DataStreamsMonitoring.Hashes;
+using Datadog.Trace.ExtensionMethods;
 using FluentAssertions;
+using FluentAssertions.Extensions;
 using Xunit;
 
 namespace Datadog.Trace.Tests.DataStreamsMonitoring;
@@ -18,18 +20,34 @@ public class PathwayContextEncoderTests
     [Fact]
     public void TestRandomValues()
     {
+        var nowNs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000;
         for (var i = 0; i < 1000; i++)
         {
-            EncodeTest(unchecked((ulong)GetLong()), Math.Abs(GetLong()), Math.Abs(GetLong()));
+            // Mask off the sign bit to guarantee non-negative — Math.Abs(long.MinValue) returns long.MinValue
+            // (overflow), which would feed a negative timestamp into EncodeTest and fail the overflow guard.
+            EncodeTest(unchecked((ulong)GetLong()), GetNonNegativeLongInThePast(nowNs), GetNonNegativeLongInThePast(nowNs));
         }
+
+        static long GetNonNegativeLongInThePast(long nowNs) => (GetLong() & long.MaxValue) % nowNs;
     }
 
     [Theory]
     [InlineData(0, 0, 0)]
-    [InlineData(ulong.MaxValue, long.MaxValue, long.MaxValue)]
     public void TestEdgeCases(ulong hash, long pathwayStartNs, long edgeStartNs)
         => EncodeTest(hash, pathwayStartNs, edgeStartNs);
 
+    [Theory]
+    [InlineData(ulong.MaxValue, long.MaxValue, long.MaxValue)]
+    public void DecoderFailure_OverflowTimestamps(ulong hash, long pathwayStartNs, long edgeStartNs)
+    {
+        var pathway = new PathwayContext(new PathwayHash(hash), pathwayStartNs, edgeStartNs);
+
+        var bytes = PathwayContextEncoder.Encode(pathway);
+        var decoded = PathwayContextEncoder.Decode(bytes);
+
+        decoded.Should().BeNull();
+    }
+
     [Theory]
     [InlineData(0)]
     [InlineData(1)]
@@ -105,6 +123,37 @@ public void DecoderFailure_OutOfRangePathwayBytes()
         decoded.Should().BeNull();
     }
 
+    [Fact]
+    public void DecoderFailure_FutureTimestamps()
+    {
+        var futureNs = DateTimeOffset.UtcNow
+                                     .AddNanoseconds(PathwayContextEncoder.MaxClockSkewNs * 5)
+                                     .ToUnixTimeNanoseconds();
+        var pathway = new PathwayContext(new PathwayHash(123), pathwayStartNs: futureNs, edgeStartNs: futureNs);
+
+        var bytes = PathwayContextEncoder.Encode(pathway);
+        var decoded = PathwayContextEncoder.Decode(bytes);
+
+        decoded.Should().BeNull();
+    }
+
+    [Fact]
+    public void DecodeSucceeds_TimestampExactlyNow()
+    {
+        // The check uses strict >, so a timestamp equal to nowNs (within tolerance) must round-trip.
+        var nowNs = DateTimeOffset.UtcNow.ToUnixTimeNanoseconds();
+        EncodeTest(hash: 123, pathwayStartNs: nowNs, edgeStartNs: nowNs);
+    }
+
+    [Fact]
+    public void DecodeSucceeds_TimestampWithinSkewTolerance()
+    {
+        var futureNs = DateTimeOffset.UtcNow
+                                     .AddNanoseconds(PathwayContextEncoder.MaxClockSkewNs / 2)
+                                     .ToUnixTimeNanoseconds();
+        EncodeTest(hash: 123, pathwayStartNs: futureNs, edgeStartNs: futureNs);
+    }
+
 #if NETCOREAPP3_1_OR_GREATER
     [Theory]
     [InlineData(0, 0, 0)]
@@ -120,6 +169,22 @@ public void EncodeInto_ProducesSameBytesAsEncode(ulong hash, long pathwayStartNs
 
         buffer.Slice(0, bytesWritten).ToArray().Should().Equal(expected);
     }
+
+    [Fact]
+    public void DecoderFailure_FutureTimestamps_Span()
+    {
+        var futureNs = DateTimeOffset.UtcNow
+                                     .AddNanoseconds(PathwayContextEncoder.MaxClockSkewNs * 5)
+                                     .ToUnixTimeNanoseconds();
+        var pathway = new PathwayContext(new PathwayHash(123), pathwayStartNs: futureNs, edgeStartNs: futureNs);
+
+        Span<byte> buffer = stackalloc byte[PathwayContextEncoder.MaxEncodedSize];
+        var bytesWritten = PathwayContextEncoder.EncodeInto(pathway, buffer);
+
+        var decoded = PathwayContextEncoder.Decode(buffer.Slice(0, bytesWritten));
+
+        decoded.Should().BeNull();
+    }
 #endif
 
     private static void EncodeTest(ulong hash, long pathwayStartNs, long edgeStartNs)