diff --git a/backend/internal/repository/channel_monitor_v2_aggregation.go b/backend/internal/repository/channel_monitor_v2_aggregation.go index cd98d0f1a..abf82f3a2 100644 --- a/backend/internal/repository/channel_monitor_v2_aggregation.go +++ b/backend/internal/repository/channel_monitor_v2_aggregation.go @@ -240,6 +240,11 @@ ELSE 2147483647 END` const channelMonitorV2ErrorAggregationSQL = ` WITH dedup AS ( + WITH candidate_ids AS MATERIALIZED ( + SELECT DISTINCT request_id + FROM ops_error_logs + WHERE created_at >= $1 AND created_at < $2 AND NULLIF(request_id, '') IS NOT NULL + ) SELECT DISTINCT ON (COALESCE(NULLIF(request_id, ''), 'error:' || id::text)) date_trunc('minute', created_at) AS bucket_start, lower(COALESCE(NULLIF(TRIM(platform), ''), 'unknown')) AS platform, @@ -252,16 +257,12 @@ WITH dedup AS ( OR error_owner = 'provider' OR upstream_status_code IS NOT NULL) AS upstream_affected, CASE WHEN jsonb_typeof(upstream_errors) = 'array' THEN jsonb_array_length(upstream_errors) ELSE 0 END AS upstream_attempts FROM ops_error_logs current_error - WHERE current_error.created_at >= $1 AND current_error.created_at < $2 AND NOT current_error.is_count_tokens + WHERE ( + (NULLIF(current_error.request_id, '') IS NULL AND current_error.created_at >= $1 AND current_error.created_at < $2) + OR current_error.request_id IN (SELECT request_id FROM candidate_ids) + ) + AND NOT current_error.is_count_tokens AND (COALESCE(current_error.status_code, 0) >= 400 OR current_error.error_type = 'cyber_policy') - AND (NULLIF(current_error.request_id, '') IS NULL OR NOT EXISTS ( - SELECT 1 FROM ops_error_logs newer - WHERE newer.request_id = current_error.request_id - AND NOT newer.is_count_tokens - AND (COALESCE(newer.status_code, 0) >= 400 OR newer.error_type = 'cyber_policy') - AND newer.created_at < $2 - AND (newer.created_at, newer.id) > (current_error.created_at, current_error.id) - )) ORDER BY COALESCE(NULLIF(request_id, ''), 'error:' || id::text), created_at DESC, id DESC ), classified AS ( SELECT *, CASE @@ -284,6 +285,7 @@ WITH dedup AS ( WHEN status_code >= 500 OR error_type = 'internal' OR error_owner = 'system' THEN 'internal' ELSE 'other' END AS category FROM dedup + WHERE bucket_start >= $1 AND bucket_start < $2 ), metric_rows AS ( INSERT INTO channel_monitor_v2_metrics_1m (bucket_start, platform, group_id, model, error_requests, upstream_affected_requests, upstream_attempt_count, computed_at) SELECT bucket_start, platform, group_id, model, COUNT(*), COUNT(*) FILTER (WHERE upstream_affected), SUM(upstream_attempts), NOW() diff --git a/backend/internal/repository/channel_monitor_v2_repo_test.go b/backend/internal/repository/channel_monitor_v2_repo_test.go index 3dff580fe..e51ec4569 100644 --- a/backend/internal/repository/channel_monitor_v2_repo_test.go +++ b/backend/internal/repository/channel_monitor_v2_repo_test.go @@ -98,7 +98,8 @@ func TestChannelMonitorV2ErrorAggregationCountsFinalUserErrorsOnly(t *testing.T) require.Contains(t, query, "not current_error.is_count_tokens") require.Contains(t, query, "error_type = 'cyber_policy'") require.Contains(t, query, "distinct on") - require.Contains(t, query, "not exists") + require.Contains(t, query, "candidate_ids") + require.Contains(t, query, "where bucket_start >= $1 and bucket_start < $2") require.Contains(t, query, "upstream_affected_requests") require.Contains(t, query, "jsonb_array_length(upstream_errors) > 0") }