[Prometheus Remote Write] The PRW emits partial histograms when buckets are splitted in requests

## Version
Observed with Elastic Agent 9.3.2. The behavior is also reproducible on the current Beats implementation.

## Configuration

```yaml
use_types: true
rate_counters: true
```

### Summary

The Prometheus remote_write typed [event generator](https://github.com/elastic/beats/blob/9b4045d9cc44dc2de5dbae127275e5bad4216fcb/metricbeat/module/prometheus/remote_write/data.go#L54) aggregates classic Prometheus histogram buckets only within a single remote_write HTTP request.

When bucket series for the same histogram, labels, and timestamp are split across requests, each request emits a separate partial Elasticsearch histogram. These events have the same timestamp, labels, and histogram metric name, making them potential TSDB document-identity collisions.

### Expected behavior

Histogram buckets for the same metric, labels, and timestamp should not produce multiple incomplete histogram documents that compete for the same TSDB identity.

Ideally, the resulting Elasticsearch histogram should contain entries derived from the complete Prometheus bucket set.

### Actual behavior

Each HTTP request creates a new local histogram map in data.go:
https://github.com/elastic/beats/blob/main/x-pack/metricbeat/module/prometheus/remote_write/data.go#L195

The histogram is processed and emitted at the end of that GenerateEvents call. No histogram state is retained for a later request.
The HTTP handler calls GenerateEvents once per decoded remote_write request:

https://github.com/elastic/beats/blob/main/metricbeat/module/prometheus/remote_write/remote_write.go

See
```yaml
samples := protoToSamples(&protoReq)
events := m.promEventsGen.GenerateEvents(samples)
```

Therefore, buckets received in a subsequent request cannot be merged with buckets from the previous request.

### Steps to reproduce

Use one histogram with identical labels and timestamp, split across two GenerateEvents calls:

Request 1:
```yaml
http_request_duration_seconds_bucket{handler="/checkout",le="0.1"} 10
http_request_duration_seconds_bucket{handler="/checkout",le="0.5"} 20
```

Request 2:
```yaml
http_request_duration_seconds_bucket{handler="/checkout",le="1"} 30
http_request_duration_seconds_bucket{handler="/checkout",le="+Inf"} 40
```

Sample test for 2 Requests with same metrics and diffrent buckets

```
func TestGenerateEventsHistogramSplitAcrossRequests(t *testing.T) {
g := remoteWriteTypedGenerator{
counterCache: xcollector.NewCounterCache(time.Minute),
}
timestamp := model.Time(424242)
labels := mapstr.M{
"handler": model.LabelValue("/checkout"),
}
eventKey := labels.String() + timestamp.Time().String()

firstRequest := model.Samples{
&model.Sample{
Metric: model.Metric{
"__name__": "http_request_duration_seconds_bucket",
"handler": "/checkout",
"le": "0.1",
},
Value: 10,
Timestamp: timestamp,
},
&model.Sample{
Metric: model.Metric{
"__name__": "http_request_duration_seconds_bucket",
"handler": "/checkout",
"le": "0.5",
},
Value: 20,
Timestamp: timestamp,
},
}
secondRequest := model.Samples{
&model.Sample{
Metric: model.Metric{
"__name__": "http_request_duration_seconds_bucket",
"handler": "/checkout",
"le": "1",
},
Value: 30,
Timestamp: timestamp,
},
&model.Sample{
Metric: model.Metric{
"__name__": "http_request_duration_seconds_bucket",
"handler": "/checkout",
"le": "+Inf",
},
Value: 40,
Timestamp: timestamp,
},
}

firstEvents := g.GenerateEvents(firstRequest)
secondEvents := g.GenerateEvents(secondRequest)

assert.Len(t, firstEvents, 1, "the first request should emit one event")
assert.Len(t, secondEvents, 1, "the second request should emit one event")
assert.Contains(t, firstEvents, eventKey, "the first event should use the shared labels and timestamp")
assert.Contains(t, secondEvents, eventKey, "the second event should use the same labels and timestamp")

firstHistogram := firstEvents[eventKey].ModuleFields["http_request_duration_seconds"]
secondHistogram := secondEvents[eventKey].ModuleFields["http_request_duration_seconds"]
assert.Equal(t, mapstr.M{
"histogram": mapstr.M{
"values": []float64{0.05, 0.30000000000000004},
"counts": []uint64{0, 0},
},
}, firstHistogram, "the first request should emit only its lower buckets")
assert.Equal(t, mapstr.M{
"histogram": mapstr.M{
"values": []float64{0.5, 1},
"counts": []uint64{0, 0},
},
}, secondHistogram, "the second request should emit only its upper buckets")
}

```

Both requests use the same timestamp.

Run:

go test -v -race \
-run '^TestGenerateEventsHistogramSplitAcrossRequests$' \
./x-pack/metricbeat/module/prometheus/remote_write

Reproduction result
Both calls emit an event with the same labels and timestamp.

The first event contains only the lower buckets:

{
"values": [0.05, 0.30000000000000004],
"counts": [0, 0]
}
The second event contains only the upper buckets:

{
"values": [0.5, 1],
"counts": [0, 0]
}

### Impact

- Elasticsearch receives incomplete histogram fields.
- Multiple partial events may share the same TSDB identity.
- This can lead to document conflicts or lost histogram buckets.
- The probability increases with histogram cardinality and remote_write request splitting.

### Possible direction

Introduce bounded cross-request histogram assembly keyed by:

Should follow the xisting dimensions: metric name + labels without le + sample timestamp

The design would need:

- Explicit memory and entry limits: If Agent buffers histogram buckets across requests, it must keep them in memory temporarily. So somethilng like:
```(metric + labels + timestamp) -> collected buckets```
In a high-cardinality environment, thousands or millions of incomplete histograms could accumulate. “Explicit memory and entry limits” means defining hard safeguards.

- Short assembly and hard timeouts
**Short assembly**
Starts or resets whenever another bucket arrives.

Example: 2 seconds.
```
bucket le=0.1 arrives
500ms later bucket le=0.5 arrives → reset timer
400ms later +Inf arrives → reset timer
2s with no more buckets → flush histogram
```
We should answer: “Has this histogram been quiet long enough that no more buckets are likely to arrive?”

**Hard timeout**
Starts when the first bucket arrives and never resets.

Example: 10 seconds.

```
first bucket arrives → hard deadline starts
more buckets keep arriving
10 seconds reached → flush or evict regardless
```
It prevents continuous or malformed traffic from keeping an entry in memory indefinitely.

- Late-bucket handling: defines what happens when a bucket arrives after its histogram was already flushed
Specifically useful in our case
- Concurrent request handling
- Metrics for timeout, eviction, and partial flushes: For intenral monitor usage

**Note:**
Bucket sorting by numeric le before PromHistogramToES should also be considered separately because the conversion assumes ascending cumulative buckets.