Fix flaky data_streams bucket aggregation test

Fixes race condition in 'aggregates multiple checkpoints into DDSketch histograms' test.

Root Cause:
The test captured 'now = Time.now.to_f' at the start, but the checkpoints
created immediately after used their own Core::Utils::Time.now internally.
These timestamps could differ by milliseconds, causing checkpoints to be
bucketed into a different time window than the test expected.

When the test calculated the expected bucket:
  bucket_time_ns = now_ns - (now_ns % processor.bucket_size_ns)

And looked it up:
  bucket = processor.buckets[bucket_time_ns]

The bucket would be nil if the checkpoints used a slightly different
timestamp and landed in an adjacent bucket.

Solution:
Use Core::Utils::Time.now_provider to inject a fixed timestamp
(Time.utc(2000, 1, 1, 0, 0, 0)) for the entire test. This ensures
all checkpoints use the exact same deterministic timestamp,
eliminating the race condition completely.

The test logic remains unchanged - we're just controlling the time
source to make it deterministic rather than dependent on wall clock
timing.

Related to incident #46145
Seed that reproduced the flake: 61335

Author: ericfirth

spec/datadog/data_streams/processor_spec.rb

@@ -186,47 +186,54 @@
 
     describe 'internal bucket aggregation' do
       it 'aggregates multiple checkpoints into DDSketch histograms' do
-        now = Time.now.to_f
+        fixed_time = Time.utc(2000, 1, 1, 0, 0, 0)
+        allow(Datadog::Tracing).to receive(:active_span).and_return(nil)
 
-        # Create multiple checkpoints with the same tags to aggregate
-        processor.set_produce_checkpoint(type: 'kafka', destination: 'topicA', manual_checkpoint: false)
-        processor.set_produce_checkpoint(type: 'kafka', destination: 'topicA', manual_checkpoint: false)
-        processor.set_produce_checkpoint(type: 'kafka', destination: 'topicA', manual_checkpoint: false)
+        begin
+          Datadog::Core::Utils::Time.now_provider = -> { fixed_time }
 
-        # Flush the event buffer to process checkpoints
-        processor.send(:process_events)
+          # Create multiple checkpoints with the same tags to aggregate
+          processor.set_produce_checkpoint(type: 'kafka', destination: 'topicA', manual_checkpoint: false)
+          processor.set_produce_checkpoint(type: 'kafka', destination: 'topicA', manual_checkpoint: false)
+          processor.set_produce_checkpoint(type: 'kafka', destination: 'topicA', manual_checkpoint: false)
 
-        # Access internal buckets to verify aggregation
-        expect(processor.buckets).not_to be_empty
+          # Flush the event buffer to process checkpoints
+          processor.send(:process_events)
 
-        # Find the bucket for this time window
-        now_ns = (now * 1e9).to_i
-        bucket_time_ns = now_ns - (now_ns % processor.bucket_size_ns)
+          # Access internal buckets to verify aggregation
+          expect(processor.buckets).not_to be_empty
 
-        bucket = processor.buckets[bucket_time_ns]
-        expect(bucket).not_to be_nil
+          # Find the bucket for this time window
+          now_ns = (fixed_time.to_f * 1e9).to_i
+          bucket_time_ns = now_ns - (now_ns % processor.bucket_size_ns)
 
-        # Verify stats were aggregated for this pathway
-        pathway_stats = bucket[:pathway_stats]
-        expect(pathway_stats).not_to be_empty
+          bucket = processor.buckets[bucket_time_ns]
+          expect(bucket).not_to be_nil
 
-        # At least one aggregation key should exist
-        aggr_key = pathway_stats.keys.first
-        stats = pathway_stats[aggr_key]
+          # Verify stats were aggregated for this pathway
+          pathway_stats = bucket[:pathway_stats]
+          expect(pathway_stats).not_to be_empty
 
-        # Verify DDSketch objects were populated
-        expect(stats[:edge_latency]).to be_a(Datadog::Core::DDSketch)
-        expect(stats[:full_pathway_latency]).to be_a(Datadog::Core::DDSketch)
+          # At least one aggregation key should exist
+          aggr_key = pathway_stats.keys.first
+          stats = pathway_stats[aggr_key]
 
-        # Verify exactly 3 samples were recorded (matching Python test)
-        expect(stats[:edge_latency].count).to eq(3)
-        expect(stats[:full_pathway_latency].count).to eq(3)
+          # Verify DDSketch objects were populated
+          expect(stats[:edge_latency]).to be_a(Datadog::Core::DDSketch)
+          expect(stats[:full_pathway_latency]).to be_a(Datadog::Core::DDSketch)
 
-        # Verify sketches can be encoded for serialization
-        expect(stats[:edge_latency].encode).to be_a(String)
-        expect(stats[:edge_latency].encode).not_to be_empty
-        expect(stats[:full_pathway_latency].encode).to be_a(String)
-        expect(stats[:full_pathway_latency].encode).not_to be_empty
+          # Verify exactly 3 samples were recorded (matching Python test)
+          expect(stats[:edge_latency].count).to eq(3)
+          expect(stats[:full_pathway_latency].count).to eq(3)
+
+          # Verify sketches can be encoded for serialization
+          expect(stats[:edge_latency].encode).to be_a(String)
+          expect(stats[:edge_latency].encode).not_to be_empty
+          expect(stats[:full_pathway_latency].encode).to be_a(String)
+          expect(stats[:full_pathway_latency].encode).not_to be_empty
+        ensure
+          Datadog::Core::Utils::Time.now_provider = -> { ::Time.now }
+        end
       end
     end
   end