[pull] main from hookdeck:main

.env.example

@@ -88,6 +88,8 @@
 # PUBLISH_RABBITMQ_SERVER_URL="amqp://guest:guest@rabbitmq:5672"
 # PUBLISH_RABBITMQ_QUEUE="publish"
 # PUBLISH_RABBITMQ_EXCHANGE="outpost"
+# Forward proxy for the publish queue connection (RabbitMQ only).
+# PUBLISH_PROXY_URL="http://user:pass@proxy:10000"
 
 ## AWS SQS
 # PUBLISH_AWS_SQS_ENDPOINT="http://aws:4566"

cmd/bench/metrics/README.md

@@ -60,4 +60,5 @@ Deterministic via modulo arithmetic (identical for both backends):
 - **3 topics** — `order.created`, `order.updated`, `payment.received`
 - **Time** — evenly spread across January 2000
 - **Attempts** — chained retries (1 event -> 1-4 attempts), 0.5% permanently failed
+- **Latency** — `20 + (n*7919) % 500` ms per attempt; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail for the percentile benchmarks)
 - 10M events -> ~12.6M attempts (22.6M total rows)

cmd/bench/metrics/bench_test.go

@@ -220,6 +220,103 @@ var attemptCases = []struct {
 			Filters:   tenant0(),
 		},
 	},
+	{
+		name: "AvgLatencyAll",
+		req: driver.MetricsRequest{
+			TimeRange: fullMonth,
+			Measures:  []string{"avg_latency"},
+			Filters:   tenant0(),
+		},
+	},
+	{
+		name: "LatencyPercentilesAll",
+		req: driver.MetricsRequest{
+			TimeRange: fullMonth,
+			Measures:  []string{"p50_latency", "p95_latency", "p99_latency"},
+			Filters:   tenant0(),
+		},
+	},
+	{
+		name: "LatencyPercentilesByDestinationHourly",
+		req: driver.MetricsRequest{
+			TimeRange:   oneDay,
+			Granularity: hourly(),
+			Measures:    []string{"count", "avg_latency", "p95_latency"},
+			Dimensions:  []string{"destination_id"},
+			Filters:     tenant0(),
+		},
+	},
+	{
+		name: "LatencyPercentilesSingleDestinationDaily",
+		req: driver.MetricsRequest{
+			TimeRange:   fullMonth,
+			Granularity: daily(),
+			Measures:    []string{"count", "error_rate", "avg_latency", "p50_latency", "p95_latency", "p99_latency"},
+			Filters:     withTenant0(map[string][]string{"destination_id": {"dest_42"}}),
+		},
+	},
+	// Marginal cost of latency on a query a dashboard already runs: the same
+	// count + error_rate by destination, with and without avg + p95 latency.
+	{
+		name: "Dashboard_Hourly_1Day",
+		req: driver.MetricsRequest{
+			TimeRange:   oneDay,
+			Granularity: hourly(),
+			Measures:    []string{"count", "error_rate"},
+			Dimensions:  []string{"destination_id"},
+			Filters:     tenant0(),
+		},
+	},
+	{
+		name: "DashboardWithLatency_Hourly_1Day",
+		req: driver.MetricsRequest{
+			TimeRange:   oneDay,
+			Granularity: hourly(),
+			Measures:    []string{"count", "error_rate", "avg_latency", "p95_latency"},
+			Dimensions:  []string{"destination_id"},
+			Filters:     tenant0(),
+		},
+	},
+	{
+		name: "Dashboard_Hourly_1Month",
+		req: driver.MetricsRequest{
+			TimeRange:   fullMonth,
+			Granularity: hourly(),
+			Measures:    []string{"count", "error_rate"},
+			Dimensions:  []string{"destination_id"},
+			Filters:     tenant0(),
+		},
+	},
+	{
+		name: "DashboardWithLatency_Hourly_1Month",
+		req: driver.MetricsRequest{
+			TimeRange:   fullMonth,
+			Granularity: hourly(),
+			Measures:    []string{"count", "error_rate", "avg_latency", "p95_latency"},
+			Dimensions:  []string{"destination_id"},
+			Filters:     tenant0(),
+		},
+	},
+	{
+		name: "Dashboard_Daily_1Month",
+		req: driver.MetricsRequest{
+			TimeRange:   fullMonth,
+			Granularity: daily(),
+			Measures:    []string{"count", "error_rate"},
+			Dimensions:  []string{"destination_id"},
+			Filters:     tenant0(),
+		},
+	},
+	{
+		name: "DashboardWithLatency_Daily_1Month",
+		req: driver.MetricsRequest{
+			TimeRange:   fullMonth,
+			Granularity: daily(),
+			Measures:    []string{"count", "error_rate", "avg_latency", "p95_latency"},
+			Dimensions:  []string{"destination_id"},
+			Filters:     tenant0(),
+		},
+	},
 	{
 		name: "FailedRate",
 		req: driver.MetricsRequest{

cmd/bench/metrics/ch/seed.sql

@@ -31,11 +31,11 @@ SELECT concat('Seeding ', toString({rows:UInt64}), ' events + chained attempts..
 
 SELECT '[1/7] Inserting events...' AS message;
 
-INSERT INTO events (event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data)
+INSERT INTO events (event_id, tenant_id, matched_destination_ids, topic, eligible_for_retry, event_time, metadata, data)
 SELECT
   concat('evt_', toString(number))                              AS event_id,
   if(number % 10 = 0, 'tenant_1', 'tenant_0')                  AS tenant_id,
-  concat('dest_', toString(number % 500))                       AS destination_id,
+  [concat('dest_', toString(number % 500))]                     AS matched_destination_ids,
   multiIf(
     number % 3 = 0, 'order.created',
     number % 3 = 1, 'order.updated',
@@ -62,7 +62,7 @@ SELECT '[2/7] Inserting attempt 1 (all events)...' AS message;
 
 INSERT INTO attempts (
   event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data,
-  attempt_id, status, attempt_time, code, response_data, manual, attempt_number
+  attempt_id, status, attempt_time, code, response_data, manual, attempt_number, latency_ms
 )
 SELECT
   concat('evt_', toString(number))                              AS event_id,
@@ -95,14 +95,16 @@ SELECT
   )                                                             AS code,
   ''                                                            AS response_data,
   false                                                         AS manual,
-  toUInt32(1)                                                   AS attempt_number
+  toUInt32(1)                                                   AS attempt_number,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  toUInt32(if(number % 200 = 0, 5000, 20 + (number * 7919) % 500))    AS latency_ms
 FROM numbers({rows:UInt64});
 
 SELECT '[3/7] Inserting attempt 2 (20% of events)...' AS message;
 
 INSERT INTO attempts (
   event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data,
-  attempt_id, status, attempt_time, code, response_data, manual, attempt_number
+  attempt_id, status, attempt_time, code, response_data, manual, attempt_number, latency_ms
 )
 SELECT
   concat('evt_', toString(number))                              AS event_id,
@@ -135,15 +137,17 @@ SELECT
   )                                                             AS code,
   ''                                                            AS response_data,
   false                                                         AS manual,
-  toUInt32(2)                                                   AS attempt_number
+  toUInt32(2)                                                   AS attempt_number,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  toUInt32(if(number % 200 = 0, 5000, 20 + (number * 7919) % 500))    AS latency_ms
 FROM numbers({rows:UInt64})
 WHERE number % 5 = 0;
 
 SELECT '[4/7] Inserting attempt 3 (5% of events)...' AS message;
 
 INSERT INTO attempts (
   event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data,
-  attempt_id, status, attempt_time, code, response_data, manual, attempt_number
+  attempt_id, status, attempt_time, code, response_data, manual, attempt_number, latency_ms
 )
 SELECT
   concat('evt_', toString(number))                              AS event_id,
@@ -176,15 +180,17 @@ SELECT
   )                                                             AS code,
   ''                                                            AS response_data,
   number % 10 = 9                                               AS manual,
-  toUInt32(3)                                                   AS attempt_number
+  toUInt32(3)                                                   AS attempt_number,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  toUInt32(if(number % 200 = 0, 5000, 20 + (number * 7919) % 500))    AS latency_ms
 FROM numbers({rows:UInt64})
 WHERE number % 20 = 0;
 
 SELECT '[5/7] Inserting attempt 4 (1% of events)...' AS message;
 
 INSERT INTO attempts (
   event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data,
-  attempt_id, status, attempt_time, code, response_data, manual, attempt_number
+  attempt_id, status, attempt_time, code, response_data, manual, attempt_number, latency_ms
 )
 SELECT
   concat('evt_', toString(number))                              AS event_id,
@@ -217,7 +223,9 @@ SELECT
   )                                                             AS code,
   ''                                                            AS response_data,
   number % 10 = 9                                               AS manual,
-  toUInt32(4)                                                   AS attempt_number
+  toUInt32(4)                                                   AS attempt_number,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  toUInt32(if(number % 200 = 0, 5000, 20 + (number * 7919) % 500))    AS latency_ms
 FROM numbers({rows:UInt64})
 WHERE number % 100 = 0;
 

cmd/bench/metrics/pg/seed.sql

@@ -38,14 +38,14 @@ SELECT COALESCE(:'ROWS', '10000000') AS rows_count \gset
 
 \echo [1/7] Inserting events...
 
-INSERT INTO events (id, tenant_id, destination_id, time, topic, eligible_for_retry, data, metadata)
+INSERT INTO events (id, tenant_id, matched_destination_ids, time, topic, eligible_for_retry, data, metadata)
 SELECT
   'evt_' || n                                           AS id,
   CASE WHEN n % 10 = 0
     THEN 'tenant_1'
     ELSE 'tenant_0'
   END                                                   AS tenant_id,
-  'dest_' || (n % 500)                                  AS destination_id,
+  ARRAY['dest_' || (n % 500)]                           AS matched_destination_ids,
   '2000-01-01'::timestamptz
     + (n::double precision / :'rows_count'::double precision)
     * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz)
@@ -74,7 +74,8 @@ FROM generate_series(0, :'rows_count'::int - 1) AS n;
 INSERT INTO attempts (
   id, event_id, tenant_id, destination_id, topic, status, time,
   attempt_number, manual, code, response_data,
-  event_time, eligible_for_retry, event_data, event_metadata
+  event_time, eligible_for_retry, event_data, event_metadata,
+  latency_ms
 )
 SELECT
   'att_' || n || '_0'                                   AS id,
@@ -106,15 +107,18 @@ SELECT
                                                         AS event_time,
   (n % 3 != 2)                                          AS eligible_for_retry,
   '{}'                                                  AS event_data,
-  '{}'::jsonb                                           AS event_metadata
+  '{}'::jsonb                                           AS event_metadata,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  CASE WHEN n % 200 = 0 THEN 5000 ELSE 20 + (n::bigint * 7919) % 500 END AS latency_ms
 FROM generate_series(0, :'rows_count'::int - 1) AS n;
 
 \echo [3/7] Inserting attempt 2 (20% of events)...
 
 INSERT INTO attempts (
   id, event_id, tenant_id, destination_id, topic, status, time,
   attempt_number, manual, code, response_data,
-  event_time, eligible_for_retry, event_data, event_metadata
+  event_time, eligible_for_retry, event_data, event_metadata,
+  latency_ms
 )
 SELECT
   'att_' || n || '_1'                                   AS id,
@@ -146,7 +150,9 @@ SELECT
                                                         AS event_time,
   (n % 3 != 2)                                          AS eligible_for_retry,
   '{}'                                                  AS event_data,
-  '{}'::jsonb                                           AS event_metadata
+  '{}'::jsonb                                           AS event_metadata,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  CASE WHEN n % 200 = 0 THEN 5000 ELSE 20 + (n::bigint * 7919) % 500 END AS latency_ms
 FROM generate_series(0, :'rows_count'::int - 1) AS n
 WHERE n % 5 = 0;
 
@@ -155,7 +161,8 @@ WHERE n % 5 = 0;
 INSERT INTO attempts (
   id, event_id, tenant_id, destination_id, topic, status, time,
   attempt_number, manual, code, response_data,
-  event_time, eligible_for_retry, event_data, event_metadata
+  event_time, eligible_for_retry, event_data, event_metadata,
+  latency_ms
 )
 SELECT
   'att_' || n || '_2'                                   AS id,
@@ -187,7 +194,9 @@ SELECT
                                                         AS event_time,
   (n % 3 != 2)                                          AS eligible_for_retry,
   '{}'                                                  AS event_data,
-  '{}'::jsonb                                           AS event_metadata
+  '{}'::jsonb                                           AS event_metadata,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  CASE WHEN n % 200 = 0 THEN 5000 ELSE 20 + (n::bigint * 7919) % 500 END AS latency_ms
 FROM generate_series(0, :'rows_count'::int - 1) AS n
 WHERE n % 20 = 0;
 
@@ -196,7 +205,8 @@ WHERE n % 20 = 0;
 INSERT INTO attempts (
   id, event_id, tenant_id, destination_id, topic, status, time,
   attempt_number, manual, code, response_data,
-  event_time, eligible_for_retry, event_data, event_metadata
+  event_time, eligible_for_retry, event_data, event_metadata,
+  latency_ms
 )
 SELECT
   'att_' || n || '_3'                                   AS id,
@@ -228,7 +238,9 @@ SELECT
                                                         AS event_time,
   (n % 3 != 2)                                          AS eligible_for_retry,
   '{}'                                                  AS event_data,
-  '{}'::jsonb                                           AS event_metadata
+  '{}'::jsonb                                           AS event_metadata,
+  -- latency: 20..520ms body; every 200th event's attempts time out at 5000ms (~1.6% of attempts, a p99 tail)
+  CASE WHEN n % 200 = 0 THEN 5000 ELSE 20 + (n::bigint * 7919) % 500 END AS latency_ms
 FROM generate_series(0, :'rows_count'::int - 1) AS n
 WHERE n % 100 = 0;