From dc0cc469a7098e2968185e279d168affb82f0a09 Mon Sep 17 00:00:00 2001
From: Abhishek Bhaskar <abhishekbhaskar@github.com>
Date: Fri, 11 Sep 2026 01:42:44 -0500
Subject: [PATCH 1/4] report hosts to backend endpoint to forward to splunk

---
 internal/apiclient/client.go               |  17 +++
 internal/apiclient/client_test.go          |  24 ++++
 internal/egress/collector.go               | 149 +++++++++++++++++++++
 internal/egress/collector_test.go          | 130 ++++++++++++++++++
 internal/handlers/egress_allowlist.go      |  43 ++++--
 internal/handlers/egress_allowlist_test.go |  40 +++++-
 internal/metrics/collector_client_test.go  |   6 +
 proxy.go                                   |   5 +-
 8 files changed, 398 insertions(+), 16 deletions(-)
 create mode 100644 internal/egress/collector.go
 create mode 100644 internal/egress/collector_test.go

diff --git a/internal/apiclient/client.go b/internal/apiclient/client.go
index 57dcb182..181577f7 100644
--- a/internal/apiclient/client.go
+++ b/internal/apiclient/client.go
@@ -42,6 +42,7 @@ type Client struct {
 
 type ClientInterface interface {
 	ReportMetrics(ctx context.Context, metricsData string) error
+	RecordEgressHosts(ctx context.Context, egressData string) error
 }
 
 // Ensure Client implements ClientInterface
@@ -165,6 +166,22 @@ func (c *Client) ReportMetrics(ctx context.Context, metricsData string) (err err
 	return nil
 }
 
+// RecordEgressHosts sends the outbound hosts observed during the job to the
+// server, which forwards them to Splunk/Kusto for egress-allowlist tuning.
+func (c *Client) RecordEgressHosts(ctx context.Context, egressData string) (err error) {
+	egressHostsURL := c.newURL("/update_jobs/%s/record_egress_hosts", c.jobID)
+
+	// Submit JSON:
+	rsp, err := c.doRequest(ctx, "POST", egressHostsURL, egressData)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		err = errors.Join(err, rsp.Body.Close())
+	}()
+	return nil
+}
+
 func (c *Client) newURL(path string, args ...any) string {
 	return c.baseURL + fmt.Sprintf(path, args...)
 }
diff --git a/internal/apiclient/client_test.go b/internal/apiclient/client_test.go
index 691b053c..00e78f76 100644
--- a/internal/apiclient/client_test.go
+++ b/internal/apiclient/client_test.go
@@ -79,6 +79,30 @@ func TestClient_ReportMetrics_Success(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestClient_RecordEgressHosts_Success(t *testing.T) {
+	egressData := []map[string]any{
+		{"host": "registry.npmjs.org", "allowlisted": true, "count": 3, "package_manager": "npm_and_yarn"},
+		{"host": "evil.com", "allowlisted": false, "count": 1, "package_manager": "npm_and_yarn"},
+	}
+	expectedBody, err := json.Marshal(map[string]any{"data": egressData})
+	require.NoError(t, err)
+
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "/update_jobs/1234/record_egress_hosts", r.URL.Path)
+		assert.Equal(t, "application/json", r.Header.Get("Content-Type"))
+
+		body, err := io.ReadAll(r.Body)
+		require.NoError(t, err)
+		assert.JSONEq(t, string(expectedBody), string(body))
+	}))
+	defer s.Close()
+
+	client := apiclient.New(s.URL, jobToken, jobID)
+
+	err = client.RecordEgressHosts(context.Background(), string(expectedBody))
+	require.NoError(t, err)
+}
+
 func TestClient_ReportMetrics_Error(t *testing.T) {
 	var requestCount int64
 	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
diff --git a/internal/egress/collector.go b/internal/egress/collector.go
new file mode 100644
index 00000000..075df6b2
--- /dev/null
+++ b/internal/egress/collector.go
@@ -0,0 +1,149 @@
+// Package egress buffers the outbound hosts a job's proxy contacts and reports
+// them to the Dependabot API, which forwards them to Splunk/Kusto for
+// egress-allowlist tuning.
+package egress
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/dependabot/proxy/internal/apiclient"
+	"github.com/dependabot/proxy/internal/config"
+)
+
+const (
+	// flushInterval is how often buffered egress hosts are reported.
+	flushInterval = 1 * time.Minute
+	// maxHosts caps the distinct host observations tracked per job to bound
+	// memory usage.
+	maxHosts = 10_000
+	// flushTimeout bounds a single report to the API.
+	flushTimeout = 30 * time.Second
+)
+
+// hostKey uniquely identifies a buffered host observation.
+type hostKey struct {
+	host        string
+	allowlisted bool
+}
+
+// Collector aggregates the distinct outbound hosts contacted during a job and
+// periodically reports them to the Dependabot API.
+type Collector struct {
+	apiClient      apiclient.ClientInterface
+	apiEndpoint    string
+	packageManager string
+
+	mutex sync.Mutex
+	hosts map[hostKey]int
+
+	flushTicker *time.Ticker
+	closeCh     chan struct{}
+	closeChOnce sync.Once
+}
+
+// New returns a running Collector. It starts a background goroutine that flushes
+// buffered hosts on an interval until StopBatchProcess is called.
+func New(envSettings config.ProxyEnvSettings, apiClient apiclient.ClientInterface) *Collector {
+	c := &Collector{
+		apiClient:      apiClient,
+		apiEndpoint:    envSettings.APIEndpoint,
+		packageManager: envSettings.PackageManager,
+		hosts:          make(map[hostKey]int),
+		flushTicker:    time.NewTicker(flushInterval),
+		closeCh:        make(chan struct{}),
+	}
+	go c.process()
+
+	return c
+}
+
+// RecordHost buffers a single outbound host observation.
+func (c *Collector) RecordHost(host string, allowlisted bool) {
+	if host == "" {
+		return
+	}
+
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	key := hostKey{host: host, allowlisted: allowlisted}
+	if _, seen := c.hosts[key]; !seen && len(c.hosts) >= maxHosts {
+		return
+	}
+	c.hosts[key]++
+}
+
+func (c *Collector) process() {
+	defer func() {
+		if r := recover(); r != nil {
+			logrus.Errorln("egress Collector process panicked:", r)
+		}
+	}()
+
+	for {
+		select {
+		case <-c.flushTicker.C:
+			c.flush()
+		case <-c.closeCh:
+			c.flush()
+			return
+		}
+	}
+}
+
+// StopBatchProcess stops the background flusher after a final flush.
+func (c *Collector) StopBatchProcess() {
+	c.closeChOnce.Do(func() {
+		close(c.closeCh)
+	})
+}
+
+// canReport avoids sending during smoke tests, where the api endpoint is empty.
+func (c *Collector) canReport() bool {
+	return c.apiEndpoint != ""
+}
+
+func (c *Collector) flush() {
+	if !c.canReport() {
+		logrus.Info("Skipping reporting egress hosts because api endpoint is empty")
+		return
+	}
+
+	c.mutex.Lock()
+	if len(c.hosts) == 0 {
+		c.mutex.Unlock()
+		return
+	}
+	records := make([]map[string]any, 0, len(c.hosts))
+	for key, count := range c.hosts {
+		records = append(records, map[string]any{
+			"host":            key.host,
+			"allowlisted":     key.allowlisted,
+			"count":           count,
+			"package_manager": c.packageManager,
+		})
+	}
+	c.hosts = make(map[hostKey]int)
+	c.mutex.Unlock()
+
+	jsonData, err := json.Marshal(map[string]any{"data": records})
+	if err != nil {
+		logrus.Errorln("Error marshaling egress hosts data:", err)
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), flushTimeout)
+	defer cancel()
+
+	logrus.Info("Posting egress hosts to remote API endpoint")
+	if err := c.apiClient.RecordEgressHosts(ctx, string(jsonData)); err != nil {
+		logrus.Errorln("Error posting egress hosts data via api client:", err)
+	} else {
+		logrus.Infoln("Successfully posted egress hosts data via api client")
+	}
+}
diff --git a/internal/egress/collector_test.go b/internal/egress/collector_test.go
new file mode 100644
index 00000000..930f9dde
--- /dev/null
+++ b/internal/egress/collector_test.go
@@ -0,0 +1,130 @@
+package egress
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/dependabot/proxy/internal/config"
+)
+
+// mockAPIClient captures the payloads passed to RecordEgressHosts.
+type mockAPIClient struct {
+	mutex    sync.Mutex
+	payloads []string
+}
+
+func (m *mockAPIClient) ReportMetrics(context.Context, string) error { return nil }
+
+func (m *mockAPIClient) RecordEgressHosts(_ context.Context, data string) error {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	m.payloads = append(m.payloads, data)
+	return nil
+}
+
+func (m *mockAPIClient) lastPayload() (string, bool) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	if len(m.payloads) == 0 {
+		return "", false
+	}
+	return m.payloads[len(m.payloads)-1], true
+}
+
+func newTestCollector(apiEndpoint string, apiClient *mockAPIClient) *Collector {
+	return New(config.ProxyEnvSettings{
+		APIEndpoint:    apiEndpoint,
+		PackageManager: "npm_and_yarn",
+	}, apiClient)
+}
+
+func TestCollectorFlushReportsAggregatedHosts(t *testing.T) {
+	apiClient := &mockAPIClient{}
+	c := newTestCollector("https://example.com", apiClient)
+	c.flushTicker.Stop()
+
+	c.RecordHost("registry.npmjs.org", true)
+	c.RecordHost("registry.npmjs.org", true)
+	c.RecordHost("evil.com", false)
+
+	c.flush()
+
+	payload, ok := apiClient.lastPayload()
+	require.True(t, ok, "expected a payload to be reported")
+
+	var parsed struct {
+		Data []map[string]any `json:"data"`
+	}
+	require.NoError(t, json.Unmarshal([]byte(payload), &parsed))
+
+	records := map[string]map[string]any{}
+	for _, record := range parsed.Data {
+		records[record["host"].(string)] = record
+	}
+
+	require.Contains(t, records, "registry.npmjs.org")
+	assert.Equal(t, true, records["registry.npmjs.org"]["allowlisted"])
+	assert.Equal(t, float64(2), records["registry.npmjs.org"]["count"])
+	assert.Equal(t, "npm_and_yarn", records["registry.npmjs.org"]["package_manager"])
+
+	require.Contains(t, records, "evil.com")
+	assert.Equal(t, false, records["evil.com"]["allowlisted"])
+	assert.Equal(t, float64(1), records["evil.com"]["count"])
+}
+
+func TestCollectorFlushClearsBuffer(t *testing.T) {
+	apiClient := &mockAPIClient{}
+	c := newTestCollector("https://example.com", apiClient)
+	c.flushTicker.Stop()
+
+	c.RecordHost("registry.npmjs.org", true)
+	c.flush()
+	c.flush() // second flush has nothing buffered
+
+	assert.Len(t, apiClient.payloads, 1, "empty buffer should not be reported")
+}
+
+func TestCollectorSkipsReportWhenEndpointEmpty(t *testing.T) {
+	apiClient := &mockAPIClient{}
+	c := newTestCollector("", apiClient)
+	c.flushTicker.Stop()
+
+	c.RecordHost("registry.npmjs.org", true)
+	c.flush()
+
+	_, ok := apiClient.lastPayload()
+	assert.False(t, ok, "no report should be sent when api endpoint is empty")
+}
+
+func TestCollectorIgnoresEmptyHost(t *testing.T) {
+	apiClient := &mockAPIClient{}
+	c := newTestCollector("https://example.com", apiClient)
+	c.flushTicker.Stop()
+
+	c.RecordHost("", true)
+	c.flush()
+
+	_, ok := apiClient.lastPayload()
+	assert.False(t, ok, "empty host should not produce a report")
+}
+
+func TestCollectorStopBatchProcessFlushes(t *testing.T) {
+	apiClient := &mockAPIClient{}
+	c := newTestCollector("https://example.com", apiClient)
+
+	c.RecordHost("registry.npmjs.org", true)
+	c.StopBatchProcess()
+	c.StopBatchProcess() // idempotent
+
+	// Give the background goroutine a moment to flush on close.
+	require.Eventually(t, func() bool {
+		_, ok := apiClient.lastPayload()
+		return ok
+	}, time.Second, 10*time.Millisecond)
+}
diff --git a/internal/handlers/egress_allowlist.go b/internal/handlers/egress_allowlist.go
index 30d4b924..305cde63 100644
--- a/internal/handlers/egress_allowlist.go
+++ b/internal/handlers/egress_allowlist.go
@@ -19,30 +19,39 @@ const (
 	egressEnforceExperiment = "proxy_egress_enforce"
 )
 
+// EgressHostRecorder buffers observed outbound hosts so they can be reported to
+// the backend for egress-allowlist tuning.
+type EgressHostRecorder interface {
+	RecordHost(host string, allowlisted bool)
+}
+
 // EgressAllowlistHandler filters outbound requests against a per-job allowlist
 // of non-hostile domains. In observe mode it only logs non-allowlisted hosts;
 // in enforce mode it drops them with a 403. When neither flag is set it allows
 // all traffic (fail-open).
 type EgressAllowlistHandler struct {
-	observe bool
-	enforce bool
-	allowed []string
+	observe  bool
+	enforce  bool
+	allowed  []string
+	recorder EgressHostRecorder
 }
 
 // NewEgressAllowlistHandler builds the allowlist from the always-allowed GitHub
 // infrastructure domains, the union of every ecosystem's default registry hosts,
 // and the job's dynamic hosts (configured registries and OIDC token-exchange
 // endpoints derived from cfg.Credentials). The observe/enforce toggles are
-// driven by job experiments.
-func NewEgressAllowlistHandler(cfg *config.Config, env config.ProxyEnvSettings) *EgressAllowlistHandler {
+// driven by job experiments. The recorder, when non-nil, receives every observed
+// host (with its allowlisted status) for reporting to the backend.
+func NewEgressAllowlistHandler(cfg *config.Config, env config.ProxyEnvSettings, recorder EgressHostRecorder) *EgressAllowlistHandler {
 	allowed := append([]string(nil), githubInfraDomains...)
 	allowed = append(allowed, allEcosystemDomains...)
 	allowed = append(allowed, dynamicHosts(cfg.Credentials)...)
 
 	return &EgressAllowlistHandler{
-		observe: cfg.Experiments.Enabled(egressObserveExperiment),
-		enforce: cfg.Experiments.Enabled(egressEnforceExperiment),
-		allowed: allowed,
+		observe:  cfg.Experiments.Enabled(egressObserveExperiment),
+		enforce:  cfg.Experiments.Enabled(egressEnforceExperiment),
+		allowed:  allowed,
+		recorder: recorder,
 	}
 }
 
@@ -53,15 +62,23 @@ func (h *EgressAllowlistHandler) HandleRequest(req *http.Request, proxyCtx *gopr
 	}
 
 	host := helpers.GetHost(req)
-	if host == "" || h.isAllowed(host) {
+	if host == "" {
 		return req, nil
 	}
 
-	if h.observe {
-		logging.RequestLogf(proxyCtx, "* egress not allowlisted %s", host)
+	allowed := h.isAllowed(host)
+
+	if h.recorder != nil {
+		h.recorder.RecordHost(host, allowed)
 	}
-	if h.enforce {
-		return req, goproxy.NewResponse(req, goproxy.ContentTypeText, http.StatusForbidden, "Forbidden")
+
+	if !allowed {
+		if h.observe {
+			logging.RequestLogf(proxyCtx, "* egress not allowlisted %s", host)
+		}
+		if h.enforce {
+			return req, goproxy.NewResponse(req, goproxy.ContentTypeText, http.StatusForbidden, "Forbidden")
+		}
 	}
 	return req, nil
 }
diff --git a/internal/handlers/egress_allowlist_test.go b/internal/handlers/egress_allowlist_test.go
index aac36615..8cc1e128 100644
--- a/internal/handlers/egress_allowlist_test.go
+++ b/internal/handlers/egress_allowlist_test.go
@@ -11,7 +11,7 @@ import (
 )
 
 func newEgressHandler(observe, enforce bool, packageManager string) *EgressAllowlistHandler {
-	return NewEgressAllowlistHandler(egressCfg(observe, enforce), config.ProxyEnvSettings{PackageManager: packageManager})
+	return NewEgressAllowlistHandler(egressCfg(observe, enforce), config.ProxyEnvSettings{PackageManager: packageManager}, nil)
 }
 
 // egressCfg builds a Config whose experiments toggle the egress observe/enforce
@@ -30,7 +30,7 @@ func egressCfg(observe, enforce bool) *config.Config {
 func newEgressHandlerWithCreds(creds config.Credentials) *EgressAllowlistHandler {
 	cfg := egressCfg(false, true)
 	cfg.Credentials = creds
-	return NewEgressAllowlistHandler(cfg, config.ProxyEnvSettings{})
+	return NewEgressAllowlistHandler(cfg, config.ProxyEnvSettings{}, nil)
 }
 
 // egressResult runs HandleRequest and returns the response (nil means allowed).
@@ -129,6 +129,42 @@ func TestEgressAllowlist_SuffixEntryAllowsSubdomain(t *testing.T) {
 	assert.Nil(t, egressResult(t, h, "https://europe-docker.pkg.dev/v2/project/image"), "artifact registry subdomain allowed")
 }
 
+// fakeRecorder records the hosts passed to RecordHost for assertions.
+type fakeRecorder struct {
+	hosts []recordedHost
+}
+
+type recordedHost struct {
+	host        string
+	allowlisted bool
+}
+
+func (r *fakeRecorder) RecordHost(host string, allowlisted bool) {
+	r.hosts = append(r.hosts, recordedHost{host: host, allowlisted: allowlisted})
+}
+
+func TestEgressAllowlist_RecordsObservedHosts(t *testing.T) {
+	recorder := &fakeRecorder{}
+	h := NewEgressAllowlistHandler(egressCfg(true, false), config.ProxyEnvSettings{}, recorder)
+
+	egressResult(t, h, "https://registry.npmjs.org/left-pad")
+	egressResult(t, h, "https://evil.com/steal")
+
+	assert.Equal(t, []recordedHost{
+		{host: "registry.npmjs.org", allowlisted: true},
+		{host: "evil.com", allowlisted: false},
+	}, recorder.hosts)
+}
+
+func TestEgressAllowlist_DoesNotRecordWhenDisabled(t *testing.T) {
+	recorder := &fakeRecorder{}
+	h := NewEgressAllowlistHandler(egressCfg(false, false), config.ProxyEnvSettings{}, recorder)
+
+	egressResult(t, h, "https://evil.com/steal")
+
+	assert.Empty(t, recorder.hosts, "fail-open mode records nothing")
+}
+
 func TestEgressAllowlist_UnknownOrEmptyPackageManagerStillGetsUnion(t *testing.T) {
 	// The allowlist does not depend on PACKAGE_MANAGER: an unknown or empty
 	// value still yields GitHub infra + the full ecosystem union.
diff --git a/internal/metrics/collector_client_test.go b/internal/metrics/collector_client_test.go
index dd6f19ac..80f93334 100644
--- a/internal/metrics/collector_client_test.go
+++ b/internal/metrics/collector_client_test.go
@@ -19,6 +19,12 @@ func (c *MockAPIClient) ReportMetrics(context.Context, string) error {
 	return nil
 }
 
+// Mock the RecordEgressHosts method
+func (c *MockAPIClient) RecordEgressHosts(context.Context, string) error {
+	// Mock logic or simply return nil to simulate success
+	return nil
+}
+
 func createTestClient() *CollectorClient {
 
 	envSettings := config.ProxyEnvSettings{
diff --git a/proxy.go b/proxy.go
index 07cb92b1..15d56276 100644
--- a/proxy.go
+++ b/proxy.go
@@ -17,6 +17,7 @@ import (
 	"github.com/dependabot/proxy/internal/cache"
 	"github.com/dependabot/proxy/internal/config"
 	"github.com/dependabot/proxy/internal/dialer"
+	"github.com/dependabot/proxy/internal/egress"
 	"github.com/dependabot/proxy/internal/handlers"
 	"github.com/dependabot/proxy/internal/metrics"
 )
@@ -59,6 +60,7 @@ func newProxyWithCacheDir(envSettings config.ProxyEnvSettings, cfg *config.Confi
 
 	apiClient := apiclient.New(envSettings.APIEndpoint, envSettings.JobToken, envSettings.JobID, apiclient.WithTransport(transport))
 	metricsClient := metrics.New(envSettings, apiClient)
+	egressCollector := egress.New(envSettings, apiClient)
 
 	proxy := goproxy.NewProxyHttpServer()
 	proxy.Tr = transport
@@ -73,7 +75,7 @@ func newProxyWithCacheDir(envSettings config.ProxyEnvSettings, cfg *config.Confi
 	proxy.OnRequest().DoFunc(logger.logRequest)
 	proxy.OnResponse().DoFunc(logger.logResponse)
 
-	egressAllowlistHandler := handlers.NewEgressAllowlistHandler(cfg, envSettings)
+	egressAllowlistHandler := handlers.NewEgressAllowlistHandler(cfg, envSettings, egressCollector)
 	proxy.OnRequest().DoFunc(egressAllowlistHandler.HandleRequest)
 
 	enableCache := os.Getenv("PROXY_CACHE") == "true"
@@ -156,6 +158,7 @@ func newProxyWithCacheDir(envSettings config.ProxyEnvSettings, cfg *config.Confi
 		metricsClient:   metricsClient,
 		Close: func() error {
 			metricsClient.StopBatchProcess()
+			egressCollector.StopBatchProcess()
 			if cacher != nil {
 				cacher.Statistics()
 				return cacher.WriteToDisk()

From 904964c517b461416e232c3b5c60d979ad93e885 Mon Sep 17 00:00:00 2001
From: Abhishek Bhaskar <abhishekbhaskar@github.com>
Date: Fri, 11 Sep 2026 14:56:53 -0500
Subject: [PATCH 2/4] handle stop batch process gracefully and retain batch on
 failure

---
 internal/egress/collector.go      | 44 +++++++++++++---
 internal/egress/collector_test.go | 85 ++++++++++++++++++++++++++++---
 2 files changed, 114 insertions(+), 15 deletions(-)

diff --git a/internal/egress/collector.go b/internal/egress/collector.go
index 075df6b2..8fcba3c0 100644
--- a/internal/egress/collector.go
+++ b/internal/egress/collector.go
@@ -43,6 +43,7 @@ type Collector struct {
 
 	flushTicker *time.Ticker
 	closeCh     chan struct{}
+	doneCh      chan struct{}
 	closeChOnce sync.Once
 }
 
@@ -56,6 +57,7 @@ func New(envSettings config.ProxyEnvSettings, apiClient apiclient.ClientInterfac
 		hosts:          make(map[hostKey]int),
 		flushTicker:    time.NewTicker(flushInterval),
 		closeCh:        make(chan struct{}),
+		doneCh:         make(chan struct{}),
 	}
 	go c.process()
 
@@ -79,6 +81,7 @@ func (c *Collector) RecordHost(host string, allowlisted bool) {
 }
 
 func (c *Collector) process() {
+	defer close(c.doneCh)
 	defer func() {
 		if r := recover(); r != nil {
 			logrus.Errorln("egress Collector process panicked:", r)
@@ -90,17 +93,21 @@ func (c *Collector) process() {
 		case <-c.flushTicker.C:
 			c.flush()
 		case <-c.closeCh:
+			c.flushTicker.Stop()
 			c.flush()
 			return
 		}
 	}
 }
 
-// StopBatchProcess stops the background flusher after a final flush.
+// StopBatchProcess stops the background flusher and blocks until it has
+// performed its final flush, so a shutting-down process does not exit before the
+// buffered hosts are posted. It is safe to call multiple times.
 func (c *Collector) StopBatchProcess() {
 	c.closeChOnce.Do(func() {
 		close(c.closeCh)
 	})
+	<-c.doneCh
 }
 
 // canReport avoids sending during smoke tests, where the api endpoint is empty.
@@ -114,13 +121,19 @@ func (c *Collector) flush() {
 		return
 	}
 
+	// Detach the current batch, but keep it so it can be requeued if the report
+	// fails. New observations recorded during the send accumulate in a fresh map.
 	c.mutex.Lock()
 	if len(c.hosts) == 0 {
 		c.mutex.Unlock()
 		return
 	}
-	records := make([]map[string]any, 0, len(c.hosts))
-	for key, count := range c.hosts {
+	batch := c.hosts
+	c.hosts = make(map[hostKey]int)
+	c.mutex.Unlock()
+
+	records := make([]map[string]any, 0, len(batch))
+	for key, count := range batch {
 		records = append(records, map[string]any{
 			"host":            key.host,
 			"allowlisted":     key.allowlisted,
@@ -128,11 +141,11 @@ func (c *Collector) flush() {
 			"package_manager": c.packageManager,
 		})
 	}
-	c.hosts = make(map[hostKey]int)
-	c.mutex.Unlock()
 
 	jsonData, err := json.Marshal(map[string]any{"data": records})
 	if err != nil {
+		// A marshaling failure is not transient, so dropping the batch avoids
+		// requeuing data that can never be sent.
 		logrus.Errorln("Error marshaling egress hosts data:", err)
 		return
 	}
@@ -143,7 +156,24 @@ func (c *Collector) flush() {
 	logrus.Info("Posting egress hosts to remote API endpoint")
 	if err := c.apiClient.RecordEgressHosts(ctx, string(jsonData)); err != nil {
 		logrus.Errorln("Error posting egress hosts data via api client:", err)
-	} else {
-		logrus.Infoln("Successfully posted egress hosts data via api client")
+		c.requeue(batch)
+		return
+	}
+	logrus.Infoln("Successfully posted egress hosts data via api client")
+}
+
+// requeue merges a failed batch back into the buffer so a transient backend
+// failure does not permanently discard observations. Counts are summed with any
+// observations recorded during the failed send, and the maxHosts cap still
+// bounds the number of distinct hosts retained.
+func (c *Collector) requeue(batch map[hostKey]int) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	for key, count := range batch {
+		if _, seen := c.hosts[key]; !seen && len(c.hosts) >= maxHosts {
+			continue
+		}
+		c.hosts[key] += count
 	}
 }
diff --git a/internal/egress/collector_test.go b/internal/egress/collector_test.go
index 930f9dde..ba38550f 100644
--- a/internal/egress/collector_test.go
+++ b/internal/egress/collector_test.go
@@ -3,9 +3,9 @@ package egress
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -13,10 +13,13 @@ import (
 	"github.com/dependabot/proxy/internal/config"
 )
 
-// mockAPIClient captures the payloads passed to RecordEgressHosts.
+// mockAPIClient captures the payloads passed to RecordEgressHosts. When err is
+// set, RecordEgressHosts fails (simulating an unavailable backend) but still
+// records the attempted payload.
 type mockAPIClient struct {
 	mutex    sync.Mutex
 	payloads []string
+	err      error
 }
 
 func (m *mockAPIClient) ReportMetrics(context.Context, string) error { return nil }
@@ -25,7 +28,19 @@ func (m *mockAPIClient) RecordEgressHosts(_ context.Context, data string) error
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 	m.payloads = append(m.payloads, data)
-	return nil
+	return m.err
+}
+
+func (m *mockAPIClient) setErr(err error) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	m.err = err
+}
+
+func (m *mockAPIClient) payloadCount() int {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	return len(m.payloads)
 }
 
 func (m *mockAPIClient) lastPayload() (string, bool) {
@@ -122,9 +137,63 @@ func TestCollectorStopBatchProcessFlushes(t *testing.T) {
 	c.StopBatchProcess()
 	c.StopBatchProcess() // idempotent
 
-	// Give the background goroutine a moment to flush on close.
-	require.Eventually(t, func() bool {
-		_, ok := apiClient.lastPayload()
-		return ok
-	}, time.Second, 10*time.Millisecond)
+	// StopBatchProcess blocks until the final flush completes, so the payload
+	// must already be present without any further waiting.
+	_, ok := apiClient.lastPayload()
+	assert.True(t, ok, "shutdown must wait for the final flush to post buffered hosts")
+}
+
+func TestCollectorRequeuesBatchOnFailure(t *testing.T) {
+	apiClient := &mockAPIClient{}
+	apiClient.setErr(errors.New("backend unavailable"))
+	c := newTestCollector("https://example.com", apiClient)
+	c.flushTicker.Stop()
+
+	c.RecordHost("registry.npmjs.org", true)
+	c.RecordHost("registry.npmjs.org", true)
+	c.flush() // fails, batch requeued
+
+	require.Equal(t, 1, apiClient.payloadCount(), "one failed attempt so far")
+
+	// Backend recovers; the retained observations are reported on the next flush.
+	apiClient.setErr(nil)
+	c.flush()
+
+	require.Equal(t, 2, apiClient.payloadCount(), "retained batch is retried after recovery")
+
+	payload, ok := apiClient.lastPayload()
+	require.True(t, ok)
+
+	var parsed struct {
+		Data []map[string]any `json:"data"`
+	}
+	require.NoError(t, json.Unmarshal([]byte(payload), &parsed))
+	require.Len(t, parsed.Data, 1)
+	assert.Equal(t, "registry.npmjs.org", parsed.Data[0]["host"])
+	assert.Equal(t, float64(2), parsed.Data[0]["count"], "counts are preserved across the failed attempt")
+}
+
+func TestCollectorRequeueMergesWithNewObservations(t *testing.T) {
+	apiClient := &mockAPIClient{}
+	apiClient.setErr(errors.New("backend unavailable"))
+	c := newTestCollector("https://example.com", apiClient)
+	c.flushTicker.Stop()
+
+	c.RecordHost("registry.npmjs.org", true)
+	c.flush() // fails, requeued
+
+	// A new observation of the same host recorded after the failed send should
+	// be summed with the requeued count.
+	c.RecordHost("registry.npmjs.org", true)
+
+	apiClient.setErr(nil)
+	c.flush()
+
+	payload, _ := apiClient.lastPayload()
+	var parsed struct {
+		Data []map[string]any `json:"data"`
+	}
+	require.NoError(t, json.Unmarshal([]byte(payload), &parsed))
+	require.Len(t, parsed.Data, 1)
+	assert.Equal(t, float64(2), parsed.Data[0]["count"])
 }

From 52919fef7423e9869c4f8150b304b54f1533de8e Mon Sep 17 00:00:00 2001
From: Abhishek Bhaskar <abhishekbhaskar@github.com>
Date: Tue, 15 Sep 2026 01:27:31 -0500
Subject: [PATCH 3/4] emit hosts from egress handler and add tag-aware
 aggregation

---
 internal/apiclient/client.go               |  17 --
 internal/apiclient/client_test.go          |  24 ---
 internal/egress/collector.go               | 179 ------------------
 internal/egress/collector_test.go          | 199 ---------------------
 internal/handlers/egress_allowlist.go      |  57 ++++--
 internal/handlers/egress_allowlist_test.go |  52 ++++--
 internal/metrics/collector_client.go       |  17 +-
 internal/metrics/collector_client_test.go  |  53 +++++-
 proxy.go                                   |   5 +-
 9 files changed, 137 insertions(+), 466 deletions(-)
 delete mode 100644 internal/egress/collector.go
 delete mode 100644 internal/egress/collector_test.go

diff --git a/internal/apiclient/client.go b/internal/apiclient/client.go
index 181577f7..57dcb182 100644
--- a/internal/apiclient/client.go
+++ b/internal/apiclient/client.go
@@ -42,7 +42,6 @@ type Client struct {
 
 type ClientInterface interface {
 	ReportMetrics(ctx context.Context, metricsData string) error
-	RecordEgressHosts(ctx context.Context, egressData string) error
 }
 
 // Ensure Client implements ClientInterface
@@ -166,22 +165,6 @@ func (c *Client) ReportMetrics(ctx context.Context, metricsData string) (err err
 	return nil
 }
 
-// RecordEgressHosts sends the outbound hosts observed during the job to the
-// server, which forwards them to Splunk/Kusto for egress-allowlist tuning.
-func (c *Client) RecordEgressHosts(ctx context.Context, egressData string) (err error) {
-	egressHostsURL := c.newURL("/update_jobs/%s/record_egress_hosts", c.jobID)
-
-	// Submit JSON:
-	rsp, err := c.doRequest(ctx, "POST", egressHostsURL, egressData)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		err = errors.Join(err, rsp.Body.Close())
-	}()
-	return nil
-}
-
 func (c *Client) newURL(path string, args ...any) string {
 	return c.baseURL + fmt.Sprintf(path, args...)
 }
diff --git a/internal/apiclient/client_test.go b/internal/apiclient/client_test.go
index 00e78f76..691b053c 100644
--- a/internal/apiclient/client_test.go
+++ b/internal/apiclient/client_test.go
@@ -79,30 +79,6 @@ func TestClient_ReportMetrics_Success(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestClient_RecordEgressHosts_Success(t *testing.T) {
-	egressData := []map[string]any{
-		{"host": "registry.npmjs.org", "allowlisted": true, "count": 3, "package_manager": "npm_and_yarn"},
-		{"host": "evil.com", "allowlisted": false, "count": 1, "package_manager": "npm_and_yarn"},
-	}
-	expectedBody, err := json.Marshal(map[string]any{"data": egressData})
-	require.NoError(t, err)
-
-	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		assert.Equal(t, "/update_jobs/1234/record_egress_hosts", r.URL.Path)
-		assert.Equal(t, "application/json", r.Header.Get("Content-Type"))
-
-		body, err := io.ReadAll(r.Body)
-		require.NoError(t, err)
-		assert.JSONEq(t, string(expectedBody), string(body))
-	}))
-	defer s.Close()
-
-	client := apiclient.New(s.URL, jobToken, jobID)
-
-	err = client.RecordEgressHosts(context.Background(), string(expectedBody))
-	require.NoError(t, err)
-}
-
 func TestClient_ReportMetrics_Error(t *testing.T) {
 	var requestCount int64
 	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
diff --git a/internal/egress/collector.go b/internal/egress/collector.go
deleted file mode 100644
index 8fcba3c0..00000000
--- a/internal/egress/collector.go
+++ /dev/null
@@ -1,179 +0,0 @@
-// Package egress buffers the outbound hosts a job's proxy contacts and reports
-// them to the Dependabot API, which forwards them to Splunk/Kusto for
-// egress-allowlist tuning.
-package egress
-
-import (
-	"context"
-	"encoding/json"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"github.com/dependabot/proxy/internal/apiclient"
-	"github.com/dependabot/proxy/internal/config"
-)
-
-const (
-	// flushInterval is how often buffered egress hosts are reported.
-	flushInterval = 1 * time.Minute
-	// maxHosts caps the distinct host observations tracked per job to bound
-	// memory usage.
-	maxHosts = 10_000
-	// flushTimeout bounds a single report to the API.
-	flushTimeout = 30 * time.Second
-)
-
-// hostKey uniquely identifies a buffered host observation.
-type hostKey struct {
-	host        string
-	allowlisted bool
-}
-
-// Collector aggregates the distinct outbound hosts contacted during a job and
-// periodically reports them to the Dependabot API.
-type Collector struct {
-	apiClient      apiclient.ClientInterface
-	apiEndpoint    string
-	packageManager string
-
-	mutex sync.Mutex
-	hosts map[hostKey]int
-
-	flushTicker *time.Ticker
-	closeCh     chan struct{}
-	doneCh      chan struct{}
-	closeChOnce sync.Once
-}
-
-// New returns a running Collector. It starts a background goroutine that flushes
-// buffered hosts on an interval until StopBatchProcess is called.
-func New(envSettings config.ProxyEnvSettings, apiClient apiclient.ClientInterface) *Collector {
-	c := &Collector{
-		apiClient:      apiClient,
-		apiEndpoint:    envSettings.APIEndpoint,
-		packageManager: envSettings.PackageManager,
-		hosts:          make(map[hostKey]int),
-		flushTicker:    time.NewTicker(flushInterval),
-		closeCh:        make(chan struct{}),
-		doneCh:         make(chan struct{}),
-	}
-	go c.process()
-
-	return c
-}
-
-// RecordHost buffers a single outbound host observation.
-func (c *Collector) RecordHost(host string, allowlisted bool) {
-	if host == "" {
-		return
-	}
-
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
-
-	key := hostKey{host: host, allowlisted: allowlisted}
-	if _, seen := c.hosts[key]; !seen && len(c.hosts) >= maxHosts {
-		return
-	}
-	c.hosts[key]++
-}
-
-func (c *Collector) process() {
-	defer close(c.doneCh)
-	defer func() {
-		if r := recover(); r != nil {
-			logrus.Errorln("egress Collector process panicked:", r)
-		}
-	}()
-
-	for {
-		select {
-		case <-c.flushTicker.C:
-			c.flush()
-		case <-c.closeCh:
-			c.flushTicker.Stop()
-			c.flush()
-			return
-		}
-	}
-}
-
-// StopBatchProcess stops the background flusher and blocks until it has
-// performed its final flush, so a shutting-down process does not exit before the
-// buffered hosts are posted. It is safe to call multiple times.
-func (c *Collector) StopBatchProcess() {
-	c.closeChOnce.Do(func() {
-		close(c.closeCh)
-	})
-	<-c.doneCh
-}
-
-// canReport avoids sending during smoke tests, where the api endpoint is empty.
-func (c *Collector) canReport() bool {
-	return c.apiEndpoint != ""
-}
-
-func (c *Collector) flush() {
-	if !c.canReport() {
-		logrus.Info("Skipping reporting egress hosts because api endpoint is empty")
-		return
-	}
-
-	// Detach the current batch, but keep it so it can be requeued if the report
-	// fails. New observations recorded during the send accumulate in a fresh map.
-	c.mutex.Lock()
-	if len(c.hosts) == 0 {
-		c.mutex.Unlock()
-		return
-	}
-	batch := c.hosts
-	c.hosts = make(map[hostKey]int)
-	c.mutex.Unlock()
-
-	records := make([]map[string]any, 0, len(batch))
-	for key, count := range batch {
-		records = append(records, map[string]any{
-			"host":            key.host,
-			"allowlisted":     key.allowlisted,
-			"count":           count,
-			"package_manager": c.packageManager,
-		})
-	}
-
-	jsonData, err := json.Marshal(map[string]any{"data": records})
-	if err != nil {
-		// A marshaling failure is not transient, so dropping the batch avoids
-		// requeuing data that can never be sent.
-		logrus.Errorln("Error marshaling egress hosts data:", err)
-		return
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), flushTimeout)
-	defer cancel()
-
-	logrus.Info("Posting egress hosts to remote API endpoint")
-	if err := c.apiClient.RecordEgressHosts(ctx, string(jsonData)); err != nil {
-		logrus.Errorln("Error posting egress hosts data via api client:", err)
-		c.requeue(batch)
-		return
-	}
-	logrus.Infoln("Successfully posted egress hosts data via api client")
-}
-
-// requeue merges a failed batch back into the buffer so a transient backend
-// failure does not permanently discard observations. Counts are summed with any
-// observations recorded during the failed send, and the maxHosts cap still
-// bounds the number of distinct hosts retained.
-func (c *Collector) requeue(batch map[hostKey]int) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
-
-	for key, count := range batch {
-		if _, seen := c.hosts[key]; !seen && len(c.hosts) >= maxHosts {
-			continue
-		}
-		c.hosts[key] += count
-	}
-}
diff --git a/internal/egress/collector_test.go b/internal/egress/collector_test.go
deleted file mode 100644
index ba38550f..00000000
--- a/internal/egress/collector_test.go
+++ /dev/null
@@ -1,199 +0,0 @@
-package egress
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"sync"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/dependabot/proxy/internal/config"
-)
-
-// mockAPIClient captures the payloads passed to RecordEgressHosts. When err is
-// set, RecordEgressHosts fails (simulating an unavailable backend) but still
-// records the attempted payload.
-type mockAPIClient struct {
-	mutex    sync.Mutex
-	payloads []string
-	err      error
-}
-
-func (m *mockAPIClient) ReportMetrics(context.Context, string) error { return nil }
-
-func (m *mockAPIClient) RecordEgressHosts(_ context.Context, data string) error {
-	m.mutex.Lock()
-	defer m.mutex.Unlock()
-	m.payloads = append(m.payloads, data)
-	return m.err
-}
-
-func (m *mockAPIClient) setErr(err error) {
-	m.mutex.Lock()
-	defer m.mutex.Unlock()
-	m.err = err
-}
-
-func (m *mockAPIClient) payloadCount() int {
-	m.mutex.Lock()
-	defer m.mutex.Unlock()
-	return len(m.payloads)
-}
-
-func (m *mockAPIClient) lastPayload() (string, bool) {
-	m.mutex.Lock()
-	defer m.mutex.Unlock()
-	if len(m.payloads) == 0 {
-		return "", false
-	}
-	return m.payloads[len(m.payloads)-1], true
-}
-
-func newTestCollector(apiEndpoint string, apiClient *mockAPIClient) *Collector {
-	return New(config.ProxyEnvSettings{
-		APIEndpoint:    apiEndpoint,
-		PackageManager: "npm_and_yarn",
-	}, apiClient)
-}
-
-func TestCollectorFlushReportsAggregatedHosts(t *testing.T) {
-	apiClient := &mockAPIClient{}
-	c := newTestCollector("https://example.com", apiClient)
-	c.flushTicker.Stop()
-
-	c.RecordHost("registry.npmjs.org", true)
-	c.RecordHost("registry.npmjs.org", true)
-	c.RecordHost("evil.com", false)
-
-	c.flush()
-
-	payload, ok := apiClient.lastPayload()
-	require.True(t, ok, "expected a payload to be reported")
-
-	var parsed struct {
-		Data []map[string]any `json:"data"`
-	}
-	require.NoError(t, json.Unmarshal([]byte(payload), &parsed))
-
-	records := map[string]map[string]any{}
-	for _, record := range parsed.Data {
-		records[record["host"].(string)] = record
-	}
-
-	require.Contains(t, records, "registry.npmjs.org")
-	assert.Equal(t, true, records["registry.npmjs.org"]["allowlisted"])
-	assert.Equal(t, float64(2), records["registry.npmjs.org"]["count"])
-	assert.Equal(t, "npm_and_yarn", records["registry.npmjs.org"]["package_manager"])
-
-	require.Contains(t, records, "evil.com")
-	assert.Equal(t, false, records["evil.com"]["allowlisted"])
-	assert.Equal(t, float64(1), records["evil.com"]["count"])
-}
-
-func TestCollectorFlushClearsBuffer(t *testing.T) {
-	apiClient := &mockAPIClient{}
-	c := newTestCollector("https://example.com", apiClient)
-	c.flushTicker.Stop()
-
-	c.RecordHost("registry.npmjs.org", true)
-	c.flush()
-	c.flush() // second flush has nothing buffered
-
-	assert.Len(t, apiClient.payloads, 1, "empty buffer should not be reported")
-}
-
-func TestCollectorSkipsReportWhenEndpointEmpty(t *testing.T) {
-	apiClient := &mockAPIClient{}
-	c := newTestCollector("", apiClient)
-	c.flushTicker.Stop()
-
-	c.RecordHost("registry.npmjs.org", true)
-	c.flush()
-
-	_, ok := apiClient.lastPayload()
-	assert.False(t, ok, "no report should be sent when api endpoint is empty")
-}
-
-func TestCollectorIgnoresEmptyHost(t *testing.T) {
-	apiClient := &mockAPIClient{}
-	c := newTestCollector("https://example.com", apiClient)
-	c.flushTicker.Stop()
-
-	c.RecordHost("", true)
-	c.flush()
-
-	_, ok := apiClient.lastPayload()
-	assert.False(t, ok, "empty host should not produce a report")
-}
-
-func TestCollectorStopBatchProcessFlushes(t *testing.T) {
-	apiClient := &mockAPIClient{}
-	c := newTestCollector("https://example.com", apiClient)
-
-	c.RecordHost("registry.npmjs.org", true)
-	c.StopBatchProcess()
-	c.StopBatchProcess() // idempotent
-
-	// StopBatchProcess blocks until the final flush completes, so the payload
-	// must already be present without any further waiting.
-	_, ok := apiClient.lastPayload()
-	assert.True(t, ok, "shutdown must wait for the final flush to post buffered hosts")
-}
-
-func TestCollectorRequeuesBatchOnFailure(t *testing.T) {
-	apiClient := &mockAPIClient{}
-	apiClient.setErr(errors.New("backend unavailable"))
-	c := newTestCollector("https://example.com", apiClient)
-	c.flushTicker.Stop()
-
-	c.RecordHost("registry.npmjs.org", true)
-	c.RecordHost("registry.npmjs.org", true)
-	c.flush() // fails, batch requeued
-
-	require.Equal(t, 1, apiClient.payloadCount(), "one failed attempt so far")
-
-	// Backend recovers; the retained observations are reported on the next flush.
-	apiClient.setErr(nil)
-	c.flush()
-
-	require.Equal(t, 2, apiClient.payloadCount(), "retained batch is retried after recovery")
-
-	payload, ok := apiClient.lastPayload()
-	require.True(t, ok)
-
-	var parsed struct {
-		Data []map[string]any `json:"data"`
-	}
-	require.NoError(t, json.Unmarshal([]byte(payload), &parsed))
-	require.Len(t, parsed.Data, 1)
-	assert.Equal(t, "registry.npmjs.org", parsed.Data[0]["host"])
-	assert.Equal(t, float64(2), parsed.Data[0]["count"], "counts are preserved across the failed attempt")
-}
-
-func TestCollectorRequeueMergesWithNewObservations(t *testing.T) {
-	apiClient := &mockAPIClient{}
-	apiClient.setErr(errors.New("backend unavailable"))
-	c := newTestCollector("https://example.com", apiClient)
-	c.flushTicker.Stop()
-
-	c.RecordHost("registry.npmjs.org", true)
-	c.flush() // fails, requeued
-
-	// A new observation of the same host recorded after the failed send should
-	// be summed with the requeued count.
-	c.RecordHost("registry.npmjs.org", true)
-
-	apiClient.setErr(nil)
-	c.flush()
-
-	payload, _ := apiClient.lastPayload()
-	var parsed struct {
-		Data []map[string]any `json:"data"`
-	}
-	require.NoError(t, json.Unmarshal([]byte(payload), &parsed))
-	require.Len(t, parsed.Data, 1)
-	assert.Equal(t, float64(2), parsed.Data[0]["count"])
-}
diff --git a/internal/handlers/egress_allowlist.go b/internal/handlers/egress_allowlist.go
index 305cde63..7b9dc396 100644
--- a/internal/handlers/egress_allowlist.go
+++ b/internal/handlers/egress_allowlist.go
@@ -2,6 +2,7 @@ package handlers
 
 import (
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/elazarl/goproxy"
@@ -19,10 +20,16 @@ const (
 	egressEnforceExperiment = "proxy_egress_enforce"
 )
 
-// EgressHostRecorder buffers observed outbound hosts so they can be reported to
-// the backend for egress-allowlist tuning.
-type EgressHostRecorder interface {
-	RecordHost(host string, allowlisted bool)
+// egressHostMetric is the metric emitted for every observed outbound host. The
+// backend logs its raw request_host tag for allowlist discovery and buckets the
+// host before forwarding to Datadog to keep tag cardinality low.
+const egressHostMetric = "egress_host"
+
+// MetricSender emits a metric for each observed outbound host, reusing the
+// proxy's existing metrics collector (buffering, flushing, retries, and
+// job-lifecycle handling) instead of a dedicated reporting pipeline.
+type MetricSender interface {
+	SendMetric(name string, metricType string, value float64, additionalTags map[string]string) error
 }
 
 // EgressAllowlistHandler filters outbound requests against a per-job allowlist
@@ -30,28 +37,29 @@ type EgressHostRecorder interface {
 // in enforce mode it drops them with a 403. When neither flag is set it allows
 // all traffic (fail-open).
 type EgressAllowlistHandler struct {
-	observe  bool
-	enforce  bool
-	allowed  []string
-	recorder EgressHostRecorder
+	observe bool
+	enforce bool
+	allowed []string
+	metrics MetricSender
 }
 
 // NewEgressAllowlistHandler builds the allowlist from the always-allowed GitHub
 // infrastructure domains, the union of every ecosystem's default registry hosts,
 // and the job's dynamic hosts (configured registries and OIDC token-exchange
 // endpoints derived from cfg.Credentials). The observe/enforce toggles are
-// driven by job experiments. The recorder, when non-nil, receives every observed
-// host (with its allowlisted status) for reporting to the backend.
-func NewEgressAllowlistHandler(cfg *config.Config, env config.ProxyEnvSettings, recorder EgressHostRecorder) *EgressAllowlistHandler {
+// driven by job experiments. The metric sender, when non-nil, receives an
+// observation for every host (with its allowlisted status) for reporting to the
+// backend.
+func NewEgressAllowlistHandler(cfg *config.Config, env config.ProxyEnvSettings, metricSender MetricSender) *EgressAllowlistHandler {
 	allowed := append([]string(nil), githubInfraDomains...)
 	allowed = append(allowed, allEcosystemDomains...)
 	allowed = append(allowed, dynamicHosts(cfg.Credentials)...)
 
 	return &EgressAllowlistHandler{
-		observe:  cfg.Experiments.Enabled(egressObserveExperiment),
-		enforce:  cfg.Experiments.Enabled(egressEnforceExperiment),
-		allowed:  allowed,
-		recorder: recorder,
+		observe: cfg.Experiments.Enabled(egressObserveExperiment),
+		enforce: cfg.Experiments.Enabled(egressEnforceExperiment),
+		allowed: allowed,
+		metrics: metricSender,
 	}
 }
 
@@ -68,9 +76,10 @@ func (h *EgressAllowlistHandler) HandleRequest(req *http.Request, proxyCtx *gopr
 
 	allowed := h.isAllowed(host)
 
-	if h.recorder != nil {
-		h.recorder.RecordHost(host, allowed)
-	}
+	// Record the observation here, at the point of the allowlist decision, so
+	// that enforce-blocked hosts are captured before the 403 short-circuits the
+	// request chain (the downstream metrics handler would never see them).
+	h.recordHost(host, allowed)
 
 	if !allowed {
 		if h.observe {
@@ -83,6 +92,18 @@ func (h *EgressAllowlistHandler) HandleRequest(req *http.Request, proxyCtx *gopr
 	return req, nil
 }
 
+func (h *EgressAllowlistHandler) recordHost(host string, allowed bool) {
+	if h.metrics == nil {
+		return
+	}
+	// package_manager is added by the collector's default tags. request_host is
+	// the raw host; the backend buckets it before emitting to Datadog.
+	_ = h.metrics.SendMetric(egressHostMetric, "increment", 1, map[string]string{
+		"request_host": host,
+		"allowlisted":  strconv.FormatBool(allowed),
+	})
+}
+
 func (h *EgressAllowlistHandler) isAllowed(host string) bool {
 	// Normalize an absolute DNS name (trailing dot) so exact matches treat
 	// "registry.npmjs.org." as equivalent to "registry.npmjs.org", consistent
diff --git a/internal/handlers/egress_allowlist_test.go b/internal/handlers/egress_allowlist_test.go
index 8cc1e128..e164916e 100644
--- a/internal/handlers/egress_allowlist_test.go
+++ b/internal/handlers/egress_allowlist_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/dependabot/proxy/internal/config"
 )
@@ -129,40 +130,57 @@ func TestEgressAllowlist_SuffixEntryAllowsSubdomain(t *testing.T) {
 	assert.Nil(t, egressResult(t, h, "https://europe-docker.pkg.dev/v2/project/image"), "artifact registry subdomain allowed")
 }
 
-// fakeRecorder records the hosts passed to RecordHost for assertions.
-type fakeRecorder struct {
-	hosts []recordedHost
+// fakeMetricSender captures the metrics emitted by the egress handler.
+type fakeMetricSender struct {
+	metrics []sentMetric
 }
 
-type recordedHost struct {
-	host        string
-	allowlisted bool
+type sentMetric struct {
+	name string
+	tags map[string]string
 }
 
-func (r *fakeRecorder) RecordHost(host string, allowlisted bool) {
-	r.hosts = append(r.hosts, recordedHost{host: host, allowlisted: allowlisted})
+func (s *fakeMetricSender) SendMetric(name string, _ string, _ float64, additionalTags map[string]string) error {
+	s.metrics = append(s.metrics, sentMetric{name: name, tags: additionalTags})
+	return nil
 }
 
 func TestEgressAllowlist_RecordsObservedHosts(t *testing.T) {
-	recorder := &fakeRecorder{}
-	h := NewEgressAllowlistHandler(egressCfg(true, false), config.ProxyEnvSettings{}, recorder)
+	sender := &fakeMetricSender{}
+	h := NewEgressAllowlistHandler(egressCfg(true, false), config.ProxyEnvSettings{}, sender)
 
 	egressResult(t, h, "https://registry.npmjs.org/left-pad")
 	egressResult(t, h, "https://evil.com/steal")
 
-	assert.Equal(t, []recordedHost{
-		{host: "registry.npmjs.org", allowlisted: true},
-		{host: "evil.com", allowlisted: false},
-	}, recorder.hosts)
+	assert.Equal(t, []sentMetric{
+		{name: egressHostMetric, tags: map[string]string{"request_host": "registry.npmjs.org", "allowlisted": "true"}},
+		{name: egressHostMetric, tags: map[string]string{"request_host": "evil.com", "allowlisted": "false"}},
+	}, sender.metrics)
+}
+
+// TestEgressAllowlist_RecordsEnforceBlockedHosts verifies that a host blocked in
+// enforce mode is still recorded, since the observation happens before the 403
+// short-circuits the request chain.
+func TestEgressAllowlist_RecordsEnforceBlockedHosts(t *testing.T) {
+	sender := &fakeMetricSender{}
+	h := NewEgressAllowlistHandler(egressCfg(false, true), config.ProxyEnvSettings{}, sender)
+
+	resp := egressResult(t, h, "https://evil.com/steal")
+	require.NotNil(t, resp, "enforce blocks the host")
+	assert.Equal(t, http.StatusForbidden, resp.StatusCode)
+
+	assert.Equal(t, []sentMetric{
+		{name: egressHostMetric, tags: map[string]string{"request_host": "evil.com", "allowlisted": "false"}},
+	}, sender.metrics, "blocked host is recorded despite the 403")
 }
 
 func TestEgressAllowlist_DoesNotRecordWhenDisabled(t *testing.T) {
-	recorder := &fakeRecorder{}
-	h := NewEgressAllowlistHandler(egressCfg(false, false), config.ProxyEnvSettings{}, recorder)
+	sender := &fakeMetricSender{}
+	h := NewEgressAllowlistHandler(egressCfg(false, false), config.ProxyEnvSettings{}, sender)
 
 	egressResult(t, h, "https://evil.com/steal")
 
-	assert.Empty(t, recorder.hosts, "fail-open mode records nothing")
+	assert.Empty(t, sender.metrics, "fail-open mode records nothing")
 }
 
 func TestEgressAllowlist_UnknownOrEmptyPackageManagerStillGetsUnion(t *testing.T) {
diff --git a/internal/metrics/collector_client.go b/internal/metrics/collector_client.go
index c1b663b1..a9ee6d3e 100644
--- a/internal/metrics/collector_client.go
+++ b/internal/metrics/collector_client.go
@@ -134,9 +134,13 @@ func (c *CollectorClient) SendMetric(name string, metricType string, value float
 	c.BufferMutex.Lock()
 	defer c.BufferMutex.Unlock()
 
-	// Check for existing metric and aggregate if possible
+	// Check for existing metric and aggregate if possible. Metrics are only
+	// aggregated when their name, type, AND tags all match, so distinct tag
+	// sets (e.g. different request_host values) are kept as separate series
+	// rather than being merged under whichever series was buffered first.
 	for i, existingMetric := range c.MetricsBuffer {
-		if existingMetric["metric"] == prefixedName && existingMetric["type"] == metricType {
+		existingTags, _ := existingMetric["tags"].(map[string]string)
+		if existingMetric["metric"] == prefixedName && existingMetric["type"] == metricType && maps.Equal(existingTags, combinedTags) {
 			if metricType == "increment" {
 				if existingValue, ok := existingMetric["value"].(float64); ok {
 					c.MetricsBuffer[i]["value"] = existingValue + value
@@ -156,6 +160,15 @@ func (c *CollectorClient) SendMetric(name string, metricType string, value float
 		}
 	}
 
+	// Bound the number of distinct series buffered between flushes. High
+	// cardinality tags (e.g. raw request_host from the egress handler) could
+	// otherwise grow the buffer without limit. Once the cap is reached, drop new
+	// series until the next flush clears the buffer; series already buffered
+	// continue to aggregate above.
+	if len(c.MetricsBuffer) >= c.MaxBufferSize {
+		return nil
+	}
+
 	// Create new metric data
 	metricData := map[string]any{
 		"metric": prefixedName,
diff --git a/internal/metrics/collector_client_test.go b/internal/metrics/collector_client_test.go
index 80f93334..8e886a31 100644
--- a/internal/metrics/collector_client_test.go
+++ b/internal/metrics/collector_client_test.go
@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/dependabot/proxy/internal/config"
 )
@@ -19,12 +20,6 @@ func (c *MockAPIClient) ReportMetrics(context.Context, string) error {
 	return nil
 }
 
-// Mock the RecordEgressHosts method
-func (c *MockAPIClient) RecordEgressHosts(context.Context, string) error {
-	// Mock logic or simply return nil to simulate success
-	return nil
-}
-
 func createTestClient() *CollectorClient {
 
 	envSettings := config.ProxyEnvSettings{
@@ -87,6 +82,52 @@ func TestSendResponseCountMetric(t *testing.T) {
 	assert.Equal(t, "example.com", tags["request_host"])
 }
 
+func TestSendMetricSeparatesDistinctTags(t *testing.T) {
+	// Metrics with the same name and type but different tags (e.g. different
+	// request_host values) must be kept as separate series, not merged under the
+	// first one buffered.
+	client := createTestClient()
+	client.MetricsBuffer = make([]map[string]any, 0)
+
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "a.example.com"}))
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "b.example.com"}))
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "a.example.com"}))
+
+	require.Len(t, client.MetricsBuffer, 2, "distinct hosts stay in separate series")
+
+	counts := map[string]float64{}
+	for _, metric := range client.MetricsBuffer {
+		tags := metric["tags"].(map[string]string)
+		counts[tags["request_host"]] = metric["value"].(float64)
+	}
+	assert.Equal(t, 2.0, counts["a.example.com"], "same host aggregates")
+	assert.Equal(t, 1.0, counts["b.example.com"], "other host not merged in")
+}
+
+func TestSendMetricCapsDistinctSeries(t *testing.T) {
+	// Once the buffer reaches MaxBufferSize distinct series, new series are
+	// dropped (bounding cardinality) while existing series keep aggregating.
+	client := createTestClient()
+	client.MetricsBuffer = make([]map[string]any, 0)
+	client.MaxBufferSize = 2
+
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "a"}))
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "b"}))
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "c"}))
+	// Existing series still aggregates past the cap.
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "a"}))
+
+	require.Len(t, client.MetricsBuffer, 2, "buffer is capped at MaxBufferSize distinct series")
+
+	counts := map[string]float64{}
+	for _, metric := range client.MetricsBuffer {
+		tags := metric["tags"].(map[string]string)
+		counts[tags["request_host"]] = metric["value"].(float64)
+	}
+	assert.Equal(t, 2.0, counts["a"], "existing series keeps aggregating after the cap")
+	assert.NotContains(t, counts, "c", "new series dropped once capped")
+}
+
 func TestFlushBuffer(t *testing.T) {
 	// Create a new CollectorClient instance for testing
 	client := createTestClient()
diff --git a/proxy.go b/proxy.go
index 15d56276..aee19867 100644
--- a/proxy.go
+++ b/proxy.go
@@ -17,7 +17,6 @@ import (
 	"github.com/dependabot/proxy/internal/cache"
 	"github.com/dependabot/proxy/internal/config"
 	"github.com/dependabot/proxy/internal/dialer"
-	"github.com/dependabot/proxy/internal/egress"
 	"github.com/dependabot/proxy/internal/handlers"
 	"github.com/dependabot/proxy/internal/metrics"
 )
@@ -60,7 +59,6 @@ func newProxyWithCacheDir(envSettings config.ProxyEnvSettings, cfg *config.Confi
 
 	apiClient := apiclient.New(envSettings.APIEndpoint, envSettings.JobToken, envSettings.JobID, apiclient.WithTransport(transport))
 	metricsClient := metrics.New(envSettings, apiClient)
-	egressCollector := egress.New(envSettings, apiClient)
 
 	proxy := goproxy.NewProxyHttpServer()
 	proxy.Tr = transport
@@ -75,7 +73,7 @@ func newProxyWithCacheDir(envSettings config.ProxyEnvSettings, cfg *config.Confi
 	proxy.OnRequest().DoFunc(logger.logRequest)
 	proxy.OnResponse().DoFunc(logger.logResponse)
 
-	egressAllowlistHandler := handlers.NewEgressAllowlistHandler(cfg, envSettings, egressCollector)
+	egressAllowlistHandler := handlers.NewEgressAllowlistHandler(cfg, envSettings, metricsClient)
 	proxy.OnRequest().DoFunc(egressAllowlistHandler.HandleRequest)
 
 	enableCache := os.Getenv("PROXY_CACHE") == "true"
@@ -158,7 +156,6 @@ func newProxyWithCacheDir(envSettings config.ProxyEnvSettings, cfg *config.Confi
 		metricsClient:   metricsClient,
 		Close: func() error {
 			metricsClient.StopBatchProcess()
-			egressCollector.StopBatchProcess()
 			if cacher != nil {
 				cacher.Statistics()
 				return cacher.WriteToDisk()

From 160e23bc90f404d329ee146ed9e8ddcc637115ea Mon Sep 17 00:00:00 2001
From: Abhishek Bhaskar <abhishekbhaskar@github.com>
Date: Tue, 15 Sep 2026 13:22:09 -0500
Subject: [PATCH 4/4] fix deadlock and estimate accumulation and egress
 starving ordinary metrics issue

---
 internal/metrics/collector_client.go      | 131 +++++++++++++++-------
 internal/metrics/collector_client_test.go |  91 ++++++++++++++-
 2 files changed, 175 insertions(+), 47 deletions(-)

diff --git a/internal/metrics/collector_client.go b/internal/metrics/collector_client.go
index a9ee6d3e..e48627c5 100644
--- a/internal/metrics/collector_client.go
+++ b/internal/metrics/collector_client.go
@@ -21,7 +21,7 @@ type CollectorClient struct {
 	JobID               string
 	MetricsBuffer       []map[string]any
 	BufferMutex         sync.Mutex
-	MaxBufferSize       int
+	MaxSeriesPerMetric  int
 	FlushTicker         *time.Ticker
 	estimatedBufferSize int
 	closeCh             chan struct{}
@@ -43,12 +43,19 @@ type Client interface {
 func New(envSettings config.ProxyEnvSettings, apiClient apiclient.ClientInterface) *CollectorClient {
 
 	instance = &CollectorClient{
-		APIClient:           apiClient,
-		APIEndpoint:         envSettings.APIEndpoint,
-		DefaultTags:         map[string]string{"package_manager": envSettings.PackageManager, "grouped_update": envSettings.GroupedUpdate},
-		JobID:               envSettings.JobID,
-		MetricsBuffer:       make([]map[string]any, 0),
-		MaxBufferSize:       1000,
+		APIClient:     apiClient,
+		APIEndpoint:   envSettings.APIEndpoint,
+		DefaultTags:   map[string]string{"package_manager": envSettings.PackageManager, "grouped_update": envSettings.GroupedUpdate},
+		JobID:         envSettings.JobID,
+		MetricsBuffer: make([]map[string]any, 0),
+		// MaxSeriesPerMetric bounds the number of distinct series buffered per
+		// metric name between flushes. It is applied per name (rather than as a
+		// single shared cap) so a high-cardinality metric such as egress_host,
+		// whose request_host tag is a raw hostname, cannot fill the buffer and
+		// starve the ordinary request/response metrics. 500 is comfortably above
+		// the distinct-host and bucketed-tag counts either metric produces in a
+		// one-minute flush window.
+		MaxSeriesPerMetric:  500,
 		FlushTicker:         time.NewTicker(1 * time.Minute),
 		estimatedBufferSize: 0,
 		closeCh:             make(chan struct{}),
@@ -88,22 +95,29 @@ func (c *CollectorClient) canSendMetrics() bool {
 	return c.APIEndpoint != ""
 }
 
-func (c *CollectorClient) flushBuffer() {
-	// To avoid sending metrics during smoke tests in CI build
-	if !c.canSendMetrics() {
-		logrus.Info("Skipping sending metrics because api endpoint is empty")
-		return
+// drainLocked snapshots the buffered metrics, clears the buffer, and resets the
+// running size estimate. The caller MUST already hold BufferMutex. Resetting the
+// estimate here (on every drain) is what keeps it in sync with the buffer;
+// previously flushBuffer cleared the slice but left estimatedBufferSize growing
+// forever, which eventually tripped the payload-size branch in SendMetric.
+func (c *CollectorClient) drainLocked() []map[string]any {
+	if len(c.MetricsBuffer) == 0 {
+		return nil
 	}
+	batch := c.MetricsBuffer
+	c.MetricsBuffer = make([]map[string]any, 0)
+	c.estimatedBufferSize = 0
+	return batch
+}
 
-	c.BufferMutex.Lock()
-	if len(c.MetricsBuffer) == 0 {
-		c.BufferMutex.Unlock()
+// sendBatch marshals a drained batch and posts it to the API. It performs
+// blocking network I/O and MUST be called without holding BufferMutex.
+func (c *CollectorClient) sendBatch(batch []map[string]any) {
+	if len(batch) == 0 {
 		return
 	}
-	jsonData, err := json.Marshal(map[string]any{"data": c.MetricsBuffer})
-	c.MetricsBuffer = c.MetricsBuffer[:0] // Reset buffer
-	c.BufferMutex.Unlock()
 
+	jsonData, err := json.Marshal(map[string]any{"data": batch})
 	if err != nil {
 		logrus.Errorln("Error marshaling metrics data:", err)
 		return
@@ -122,6 +136,21 @@ func (c *CollectorClient) flushBuffer() {
 	}
 }
 
+func (c *CollectorClient) flushBuffer() {
+	// To avoid sending metrics during smoke tests in CI build. Checked before
+	// draining so the buffer is preserved when there is no endpoint to post to.
+	if !c.canSendMetrics() {
+		logrus.Info("Skipping sending metrics because api endpoint is empty")
+		return
+	}
+
+	c.BufferMutex.Lock()
+	batch := c.drainLocked()
+	c.BufferMutex.Unlock()
+
+	c.sendBatch(batch)
+}
+
 func (c *CollectorClient) SendMetric(name string, metricType string, value float64, additionalTags map[string]string) error {
 	// This check is in place to prevent the transmission of metrics initiated by smoke tests
 	prefixedName := "dependabot.job_proxy." + name
@@ -131,41 +160,57 @@ func (c *CollectorClient) SendMetric(name string, metricType string, value float
 	maps.Copy(combinedTags, c.DefaultTags)
 	maps.Copy(combinedTags, additionalTags)
 
+	// A batch drained because the payload-size limit was reached is sent after
+	// the mutex is released. This defer is registered before the Unlock defer so
+	// that, LIFO, the unlock runs first: we never perform network I/O or re-lock
+	// BufferMutex while already holding it (which previously deadlocked when the
+	// size-triggered path called flushBuffer).
+	var toSend []map[string]any
+	defer func() { c.sendBatch(toSend) }()
+
 	c.BufferMutex.Lock()
 	defer c.BufferMutex.Unlock()
 
-	// Check for existing metric and aggregate if possible. Metrics are only
-	// aggregated when their name, type, AND tags all match, so distinct tag
-	// sets (e.g. different request_host values) are kept as separate series
-	// rather than being merged under whichever series was buffered first.
+	// Look for an existing series to aggregate into, and count how many series
+	// already exist for this metric name. Metrics are only aggregated when their
+	// name, type, AND tags all match, so distinct tag sets (e.g. different
+	// request_host values) are kept as separate series rather than being merged
+	// under whichever series was buffered first.
+	sameNameCount := 0
 	for i, existingMetric := range c.MetricsBuffer {
+		if existingMetric["metric"] != prefixedName {
+			continue
+		}
+		sameNameCount++
 		existingTags, _ := existingMetric["tags"].(map[string]string)
-		if existingMetric["metric"] == prefixedName && existingMetric["type"] == metricType && maps.Equal(existingTags, combinedTags) {
+		if existingMetric["type"] == metricType && maps.Equal(existingTags, combinedTags) {
 			if metricType == "increment" {
-				if existingValue, ok := existingMetric["value"].(float64); ok {
-					c.MetricsBuffer[i]["value"] = existingValue + value
-				} else {
+				existingValue, ok := existingMetric["value"].(float64)
+				if !ok {
 					return fmt.Errorf("type assertion failed for metric value")
 				}
+				c.MetricsBuffer[i]["value"] = existingValue + value
 				return nil
 			}
 			if metricType == "distribution" {
-				if existingValues, ok := existingMetric["values"].([]float64); ok {
-					c.MetricsBuffer[i]["values"] = append(existingValues, value)
-				} else {
+				existingValues, ok := existingMetric["values"].([]float64)
+				if !ok {
 					return fmt.Errorf("type assertion failed for metric values")
 				}
+				c.MetricsBuffer[i]["values"] = append(existingValues, value)
 				return nil
 			}
 		}
 	}
 
-	// Bound the number of distinct series buffered between flushes. High
-	// cardinality tags (e.g. raw request_host from the egress handler) could
-	// otherwise grow the buffer without limit. Once the cap is reached, drop new
-	// series until the next flush clears the buffer; series already buffered
-	// continue to aggregate above.
-	if len(c.MetricsBuffer) >= c.MaxBufferSize {
+	// Bound the number of distinct series per metric name. High cardinality tags
+	// (e.g. raw request_host from the egress handler) could otherwise grow the
+	// buffer without limit. The cap is applied per metric name so a flood of
+	// egress observations can never starve the ordinary request/response
+	// metrics. Once a metric reaches its cap, new series for it are dropped until
+	// the next flush clears the buffer; series already buffered continue to
+	// aggregate above.
+	if sameNameCount >= c.MaxSeriesPerMetric {
 		return nil
 	}
 
@@ -190,14 +235,14 @@ func (c *CollectorClient) SendMetric(name string, metricType string, value float
 	}
 	estimatedSize := len(data)
 
-	// Check if adding this metric exceeds the maximum payload size
-	if c.estimatedBufferSize+estimatedSize < MaxPayloadSize {
-		c.MetricsBuffer = append(c.MetricsBuffer, metricData)
-		c.estimatedBufferSize += estimatedSize
-	} else {
-		c.flushBuffer()
-		c.estimatedBufferSize = len(data)
-		c.MetricsBuffer = append(c.MetricsBuffer, metricData)
+	// If adding this series would exceed the maximum payload size, drain the
+	// current buffer first (posted after the mutex is released) so the new series
+	// starts a fresh payload. Only drain when there is an endpoint to post to, so
+	// buffered metrics are not discarded during smoke tests.
+	if c.canSendMetrics() && c.estimatedBufferSize+estimatedSize >= MaxPayloadSize {
+		toSend = c.drainLocked()
 	}
+	c.MetricsBuffer = append(c.MetricsBuffer, metricData)
+	c.estimatedBufferSize += estimatedSize
 	return nil
 }
diff --git a/internal/metrics/collector_client_test.go b/internal/metrics/collector_client_test.go
index 8e886a31..2e5e6198 100644
--- a/internal/metrics/collector_client_test.go
+++ b/internal/metrics/collector_client_test.go
@@ -2,9 +2,11 @@ package metrics
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -105,11 +107,12 @@ func TestSendMetricSeparatesDistinctTags(t *testing.T) {
 }
 
 func TestSendMetricCapsDistinctSeries(t *testing.T) {
-	// Once the buffer reaches MaxBufferSize distinct series, new series are
-	// dropped (bounding cardinality) while existing series keep aggregating.
+	// Once a metric name reaches MaxSeriesPerMetric distinct series, new series
+	// for it are dropped (bounding cardinality) while existing series keep
+	// aggregating.
 	client := createTestClient()
 	client.MetricsBuffer = make([]map[string]any, 0)
-	client.MaxBufferSize = 2
+	client.MaxSeriesPerMetric = 2
 
 	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "a"}))
 	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "b"}))
@@ -117,7 +120,7 @@ func TestSendMetricCapsDistinctSeries(t *testing.T) {
 	// Existing series still aggregates past the cap.
 	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "a"}))
 
-	require.Len(t, client.MetricsBuffer, 2, "buffer is capped at MaxBufferSize distinct series")
+	require.Len(t, client.MetricsBuffer, 2, "buffer is capped at MaxSeriesPerMetric distinct series")
 
 	counts := map[string]float64{}
 	for _, metric := range client.MetricsBuffer {
@@ -128,6 +131,86 @@ func TestSendMetricCapsDistinctSeries(t *testing.T) {
 	assert.NotContains(t, counts, "c", "new series dropped once capped")
 }
 
+func TestSendMetricCapDoesNotStarveOtherMetrics(t *testing.T) {
+	// The distinct-series cap is applied per metric name, so a flood of
+	// high-cardinality egress observations must not crowd ordinary
+	// request/response metrics out of the buffer.
+	client := createTestClient()
+	client.MetricsBuffer = make([]map[string]any, 0)
+	client.MaxSeriesPerMetric = 5
+
+	require.NoError(t, client.SendMetric("http_response_count", "increment", 1, map[string]string{"response_code": "200", "request_host": "api.github.com"}))
+
+	// Far more distinct egress hosts than the per-metric cap.
+	for i := 0; i < 50; i++ {
+		require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": fmt.Sprintf("host-%d.example.com", i)}))
+	}
+
+	// A later ordinary metric must still be recorded, not dropped.
+	require.NoError(t, client.SendMetric("http_response_count", "increment", 1, map[string]string{"response_code": "500", "request_host": "api.github.com"}))
+
+	egress, responses := 0, 0
+	for _, metric := range client.MetricsBuffer {
+		switch metric["metric"] {
+		case "dependabot.job_proxy.egress_host":
+			egress++
+		case "dependabot.job_proxy.http_response_count":
+			responses++
+		}
+	}
+	assert.Equal(t, 5, egress, "egress series capped at MaxSeriesPerMetric")
+	assert.Equal(t, 2, responses, "ordinary metrics are not starved by egress cardinality")
+}
+
+func TestFlushBufferResetsSizeEstimate(t *testing.T) {
+	// Draining the buffer must reset the running size estimate. Otherwise it
+	// accumulates across flushes and eventually trips the payload-size branch in
+	// SendMetric.
+	client := createTestClient()
+	client.MetricsBuffer = make([]map[string]any, 0)
+
+	for round := 0; round < 5; round++ {
+		for i := 0; i < 10; i++ {
+			require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": fmt.Sprintf("r%d-h%d.example.com", round, i)}))
+		}
+		require.Positive(t, client.estimatedBufferSize)
+
+		client.flushBuffer()
+
+		require.Empty(t, client.MetricsBuffer, "buffer drained on flush")
+		require.Equal(t, 0, client.estimatedBufferSize, "size estimate resets when the buffer drains")
+	}
+}
+
+func TestSendMetricFlushesAtPayloadLimitWithoutDeadlock(t *testing.T) {
+	// Reaching the payload-size limit must flush the buffer and start a fresh
+	// payload without deadlocking (the size-triggered flush must not re-acquire
+	// BufferMutex while SendMetric already holds it).
+	client := createTestClient()
+	client.MetricsBuffer = make([]map[string]any, 0)
+
+	require.NoError(t, client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "seed.example.com"}))
+	// Force the next series to exceed the payload-size limit.
+	client.estimatedBufferSize = MaxPayloadSize
+
+	done := make(chan error, 1)
+	go func() {
+		done <- client.SendMetric("egress_host", "increment", 1, map[string]string{"request_host": "overflow.example.com"})
+	}()
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("SendMetric deadlocked when the payload-size flush triggered")
+	}
+
+	require.Len(t, client.MetricsBuffer, 1, "oversize flush drained the buffer before adding the new series")
+	tags := client.MetricsBuffer[0]["tags"].(map[string]string)
+	require.Equal(t, "overflow.example.com", tags["request_host"])
+	require.Less(t, client.estimatedBufferSize, MaxPayloadSize, "size estimate reset after the flush")
+}
+
 func TestFlushBuffer(t *testing.T) {
 	// Create a new CollectorClient instance for testing
 	client := createTestClient()