Merge the consumption metric pushes (#9939)

#8564

## Problem

The main and backup consumption metric pushes are completely
independent,
resulting in different event time windows and different idempotency
keys.

## Summary of changes

* Merge the push tasks, but keep chunks the same size.

libs/consumption_metrics/src/lib.rs

@@ -103,11 +103,12 @@ impl<'a> IdempotencyKey<'a> {
     }
 }
 
+/// Split into chunks of 1000 metrics to avoid exceeding the max request size.
 pub const CHUNK_SIZE: usize = 1000;
 
 // Just a wrapper around a slice of events
 // to serialize it as `{"events" : [ ] }
-#[derive(serde::Serialize, Deserialize)]
-pub struct EventChunk<'a, T: Clone> {
+#[derive(Debug, serde::Serialize, serde::Deserialize, PartialEq)]
+pub struct EventChunk<'a, T: Clone + PartialEq> {
     pub events: std::borrow::Cow<'a, [T]>,
 }

proxy/Cargo.toml

@@ -112,6 +112,7 @@ workspace_hack.workspace = true
 [dev-dependencies]
 camino-tempfile.workspace = true
 fallible-iterator.workspace = true
+flate2.workspace = true
 tokio-tungstenite.workspace = true
 pbkdf2 = { workspace = true, features = ["simple", "std"] }
 rcgen.workspace = true

proxy/src/bin/proxy.rs

@@ -517,10 +517,6 @@ async fn main() -> anyhow::Result<()> {
     if let Some(metrics_config) = &config.metric_collection {
         // TODO: Add gc regardles of the metric collection being enabled.
         maintenance_tasks.spawn(usage_metrics::task_main(metrics_config));
-        client_tasks.spawn(usage_metrics::task_backup(
-            &metrics_config.backup_metric_collection_config,
-            cancellation_token.clone(),
-        ));
     }
 
     if let Either::Left(auth::Backend::ControlPlane(api, _)) = &auth_backend {