ref(transport): introduce transport boundary

internal/telemetry/interfaces.go

@@ -2,6 +2,7 @@ package telemetry
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	"github.com/getsentry/sentry-go/internal/ratelimit"
@@ -36,6 +37,9 @@ type EnvelopeConvertible interface {
 	ToEnvelope(*protocol.EnvelopeHeader) (*protocol.Envelope, error)
 }
 
+// ErrQueueFull reports that a transport rejected an envelope because its queue is full.
+var ErrQueueFull = errors.New("transport queue full")
+
 // Transport represents the envelope-first transport interface.
 // This interface is designed for the telemetry buffer system and provides
 // non-blocking sends with backpressure signals.
@@ -44,12 +48,6 @@ type Transport interface {
 	// backpressure error if the queue is full.
 	SendEnvelope(envelope *protocol.Envelope) error
 
-	// HasCapacity reports whether the transport has capacity to accept at least one more envelope.
-	HasCapacity() bool
-
-	// IsRateLimited checks if a specific category is currently rate limited
-	IsRateLimited(category ratelimit.Category) bool
-
 	// Flush waits for all pending envelopes to be sent, with timeout
 	Flush(timeout time.Duration) bool
 

internal/telemetry/scheduler.go

@@ -2,6 +2,7 @@ package telemetry
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 
@@ -242,19 +243,6 @@ func (s *Scheduler) processItems(buffer Buffer[Item], category ratelimit.Categor
 		return
 	}
 
-	if s.isRateLimited(category) {
-		for _, item := range items {
-			s.recorder.RecordItem(report.ReasonRateLimitBackoff, item)
-		}
-		return
-	}
-	if !s.transport.HasCapacity() {
-		for _, item := range items {
-			s.recorder.RecordItem(report.ReasonQueueOverflow, item)
-		}
-		return
-	}
-
 	for _, item := range s.envelopeConvertibles(category, items) {
 		s.sendItem(item)
 	}
@@ -301,6 +289,11 @@ func (s *Scheduler) sendItem(item EnvelopeConvertible) {
 	}
 	if err := s.transport.SendEnvelope(envelope); err != nil {
 		debuglog.Printf("error sending envelope: %v", err)
+		reason := report.ReasonSendError
+		if errors.Is(err, ErrQueueFull) {
+			reason = report.ReasonQueueOverflow
+		}
+		s.recorder.RecordForEnvelope(reason, envelope)
 	}
 }
 
@@ -322,7 +315,3 @@ func (s *Scheduler) flushBuffers(ctx context.Context) bool {
 	}
 	return ctx.Err() == nil
 }
-
-func (s *Scheduler) isRateLimited(category ratelimit.Category) bool {
-	return s.transport.IsRateLimited(category)
-}

internal/telemetry/scheduler_test.go

@@ -311,39 +311,6 @@ func TestTelemetrySchedulerFlushAlreadyCanceled(t *testing.T) {
 	require.True(t, scheduler.Flush(testutils.FlushTimeout()))
 }
 
-func TestTelemetrySchedulerRateLimiting(t *testing.T) {
-	transport := &testutils.MockTelemetryTransport{}
-	dsn := &protocol.Dsn{}
-
-	buffer := NewRingBuffer[Item](ratelimit.CategoryError, 10, OverflowPolicyDropOldest, 1, 0, nil)
-	buffers := map[ratelimit.Category]Buffer[Item]{
-		ratelimit.CategoryError: buffer,
-	}
-	// no log buffer used in simplified scheduler tests
-	sdkInfo := &protocol.SdkInfo{Name: "test-sdk", Version: "1.0.0"}
-
-	scheduler := NewScheduler(buffers, transport, dsn, func() *protocol.SdkInfo { return sdkInfo }, nil)
-
-	transport.SetRateLimited("error", true)
-
-	scheduler.Start()
-	defer scheduler.Stop(100 * time.Millisecond)
-
-	item := &testTelemetryItem{id: 1, data: "test"}
-	buffer.Offer(item)
-	scheduler.Signal()
-
-	time.Sleep(200 * time.Millisecond)
-
-	if transport.GetSendCount() > 0 {
-		t.Errorf("Expected 0 items to be processed due to rate limiting, got %d", transport.GetSendCount())
-	}
-
-	if transport.GetRateLimitedCalls() == 0 {
-		t.Error("Expected rate limit check to be called")
-	}
-}
-
 func TestTelemetrySchedulerStartStop(t *testing.T) {
 	transport := &testutils.MockTelemetryTransport{}
 	dsn := &protocol.Dsn{}
@@ -405,43 +372,68 @@ func TestTelemetrySchedulerContextCancellation(t *testing.T) {
 	}
 }
 
-func TestTelemetrySchedulerRecordsFullDiscardCountsOnEnvelopeError(t *testing.T) {
-	transport := &testutils.MockTelemetryTransport{}
-	dsn := &protocol.Dsn{}
-	recorder := reportpkg.NewAggregator()
+type rejectingTransport struct {
+	testutils.MockTelemetryTransport
+	err error
+}
 
-	buffer := NewRingBuffer[Item](ratelimit.CategoryTransaction, 10, OverflowPolicyDropOldest, 1, 0, nil)
-	buffers := map[ratelimit.Category]Buffer[Item]{
-		ratelimit.CategoryTransaction: buffer,
+func (t *rejectingTransport) SendEnvelope(*protocol.Envelope) error { return t.err }
+
+type transactionTelemetryItem struct{ testTelemetryItem }
+
+func (t *transactionTelemetryItem) ToEnvelope(header *protocol.EnvelopeHeader) (*protocol.Envelope, error) {
+	return protocol.NewEnvelope(header, protocol.NewTransactionItem(3, []byte(`{}`))), nil
+}
+
+func TestTelemetrySchedulerRecordsFullDiscardCountsOnEnvelopeError(t *testing.T) {
+	transaction := testTelemetryItem{data: "tx", category: ratelimit.CategoryTransaction}
+	tests := []struct {
+		name   string
+		item   Item
+		err    error
+		reason reportpkg.DiscardReason
+	}{
+		{"conversion error", &failingTransactionTelemetryItem{testTelemetryItem: transaction, spanCount: 3}, nil, reportpkg.ReasonInternalError},
+		{"queue full", &transactionTelemetryItem{transaction}, ErrQueueFull, reportpkg.ReasonQueueOverflow},
+		{"send error", &transactionTelemetryItem{transaction}, errors.New("send failed"), reportpkg.ReasonSendError},
 	}
-	sdkInfo := &protocol.SdkInfo{Name: "test-sdk", Version: "1.0.0"}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			transport := &rejectingTransport{err: tt.err}
+			dsn := &protocol.Dsn{}
+			recorder := reportpkg.NewAggregator()
 
-	scheduler := NewScheduler(buffers, transport, dsn, func() *protocol.SdkInfo { return sdkInfo }, recorder)
+			buffer := NewRingBuffer[Item](ratelimit.CategoryTransaction, 10, OverflowPolicyDropOldest, 1, 0, nil)
+			buffers := map[ratelimit.Category]Buffer[Item]{
+				ratelimit.CategoryTransaction: buffer,
+			}
+			sdkInfo := &protocol.SdkInfo{Name: "test-sdk", Version: "1.0.0"}
 
-	buffer.Offer(&failingTransactionTelemetryItem{
-		testTelemetryItem: testTelemetryItem{data: "tx", category: ratelimit.CategoryTransaction},
-		spanCount:         3,
-	})
+			scheduler := NewScheduler(buffers, transport, dsn, func() *protocol.SdkInfo { return sdkInfo }, recorder)
 
-	scheduler.Flush(time.Second)
+			buffer.Offer(tt.item)
 
-	clientReport := recorder.TakeReport()
-	if clientReport == nil {
-		t.Fatal("expected client report")
-	}
+			scheduler.Flush(time.Second)
 
-	outcomes := map[ratelimit.Category]int64{}
-	for _, discarded := range clientReport.DiscardedEvents {
-		if discarded.Reason != reportpkg.ReasonInternalError {
-			t.Fatalf("unexpected reason: %s", discarded.Reason)
-		}
-		outcomes[discarded.Category] += discarded.Quantity
-	}
+			clientReport := recorder.TakeReport()
+			if clientReport == nil {
+				t.Fatal("expected client report")
+			}
 
-	if outcomes[ratelimit.CategoryTransaction] != 1 {
-		t.Fatalf("expected one discarded transaction, got %d", outcomes[ratelimit.CategoryTransaction])
-	}
-	if outcomes[ratelimit.CategorySpan] != 3 {
-		t.Fatalf("expected discarded span count to be recorded, got %d", outcomes[ratelimit.CategorySpan])
+			outcomes := map[ratelimit.Category]int64{}
+			for _, discarded := range clientReport.DiscardedEvents {
+				if discarded.Reason != tt.reason {
+					t.Fatalf("unexpected reason: %s", discarded.Reason)
+				}
+				outcomes[discarded.Category] += discarded.Quantity
+			}
+
+			if outcomes[ratelimit.CategoryTransaction] != 1 {
+				t.Fatalf("expected one discarded transaction, got %d", outcomes[ratelimit.CategoryTransaction])
+			}
+			if outcomes[ratelimit.CategorySpan] != 3 {
+				t.Fatalf("expected discarded span count to be recorded, got %d", outcomes[ratelimit.CategorySpan])
+			}
+		})
 	}
 }

internal/testutils/mocks.go

@@ -6,18 +6,14 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/getsentry/sentry-go/internal/ratelimit"
 	"github.com/getsentry/sentry-go/protocol"
 )
 
 type MockTelemetryTransport struct {
-	sentEnvelopes    []*protocol.Envelope
-	rateLimited      map[string]bool
-	sendError        error
-	mu               sync.Mutex
-	sendCount        int64
-	rateLimitedCalls int64
-	capacity         int
+	sentEnvelopes []*protocol.Envelope
+	sendError     error
+	mu            sync.Mutex
+	sendCount     int64
 }
 
 func (m *MockTelemetryTransport) SendEnvelope(envelope *protocol.Envelope) error {
@@ -33,26 +29,6 @@ func (m *MockTelemetryTransport) SendEnvelope(envelope *protocol.Envelope) error
 	return nil
 }
 
-func (m *MockTelemetryTransport) IsRateLimited(category ratelimit.Category) bool {
-	atomic.AddInt64(&m.rateLimitedCalls, 1)
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if m.rateLimited == nil {
-		return false
-	}
-	return m.rateLimited[string(category)]
-}
-
-func (m *MockTelemetryTransport) HasCapacity() bool {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.capacity == 0 {
-		return true
-	}
-	return int(m.sendCount) < m.capacity
-}
-
 func (m *MockTelemetryTransport) Flush(_ time.Duration) bool {
 	return true
 }
@@ -76,29 +52,13 @@ func (m *MockTelemetryTransport) GetSentEnvelopes() []*protocol.Envelope {
 	return result
 }
 
-func (m *MockTelemetryTransport) SetRateLimited(category string, limited bool) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.rateLimited == nil {
-		m.rateLimited = make(map[string]bool)
-	}
-	m.rateLimited[category] = limited
-}
-
 func (m *MockTelemetryTransport) GetSendCount() int64 {
 	return atomic.LoadInt64(&m.sendCount)
 }
 
-func (m *MockTelemetryTransport) GetRateLimitedCalls() int64 {
-	return atomic.LoadInt64(&m.rateLimitedCalls)
-}
-
 func (m *MockTelemetryTransport) Reset() {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.sentEnvelopes = nil
-	m.rateLimited = nil
 	atomic.StoreInt64(&m.sendCount, 0)
-	atomic.StoreInt64(&m.rateLimitedCalls, 0)
-	m.capacity = 0
 }

report/aggregator.go

@@ -1,6 +1,7 @@
 package report
 
 import (
+	"encoding/json"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -115,10 +116,21 @@ func (a *Aggregator) RecordForEnvelope(reason DiscardReason, envelope *protocol.
 				a.Record(reason, ratelimit.CategoryLogByte, int64(*item.Header.Length))
 			}
 		case protocol.EnvelopeItemTypeTraceMetric:
-			a.RecordOne(reason, ratelimit.CategoryTraceMetric)
+			if item.Header.ItemCount != nil {
+				a.Record(reason, ratelimit.CategoryTraceMetric, int64(*item.Header.ItemCount))
+			}
 		case protocol.EnvelopeItemTypeCheckIn:
 			a.RecordOne(reason, ratelimit.CategoryMonitor)
-		case protocol.EnvelopeItemTypeAttachment, protocol.EnvelopeItemTypeClientReport:
+		case protocol.EnvelopeItemTypeClientReport:
+			var clientReport ClientReport
+			if err := json.Unmarshal(item.Payload, &clientReport); err != nil {
+				debuglog.Printf("failed to decode client report: %v", err)
+				continue
+			}
+			for _, event := range clientReport.DiscardedEvents {
+				a.Record(event.Reason, event.Category, event.Quantity)
+			}
+		case protocol.EnvelopeItemTypeAttachment:
 			// Skip — not reportable categories
 		}
 	}