feat: drop newest on full queue, add configurable MaxQueueSize

.changeset/tidy-queues-drop-newest.md

@@ -0,0 +1,9 @@
+---
+"posthog-go": minor
+---
+
+Drop the newest event instead of blocking when the in-memory queue is full, and make the queue size configurable.
+
+- `Enqueue` now returns `ErrQueueFull` when the queue is full, dropping the newest message rather than blocking the caller until space frees up. This matches posthog-python and posthog-rs. The drop is reported only through the returned error, not through `Callback.Failure`, so it stays cheap and off the callback goroutines under sustained overload. **Backfill/bulk callers**: check the error returned by `Enqueue` for `ErrQueueFull` and throttle or retry (or raise `MaxQueueSize`); otherwise events that overflow the queue are dropped, not delayed.
+- Add `Config.MaxQueueSize` (default `DefaultMaxQueueSize` = 10000) to control the in-memory message queue capacity independently of `BatchSize`. It is clamped up to `BatchSize` so the queue always holds at least one full batch. This replaces the previous hardcoded `BatchSize * 10` sizing.
+- Change the default `BatchSize` (`DefaultBatchSize`) from 250 to 100, aligning with posthog-python, posthog-node, and posthog-rs. Callers that set `BatchSize` explicitly are unaffected.

api/public-api.txt

@@ -21,7 +21,7 @@ const (
 
 	DefaultFeatureFlagRequestTimeout = 3 * time.Second
 
-	DefaultBatchSize = 250
+	DefaultBatchSize = 100
 
 	DefaultMaxAttempts = 4
 
@@ -30,6 +30,8 @@ const (
 	DefaultBatchSubmitTimeout = 100 * time.Millisecond
 
 	DefaultMaxEnqueuedRequests = 1000
+
+	DefaultMaxQueueSize = 10000
 )
 const (
 	FeatureFlagErrorErrorsWhileComputing = "errors_while_computing_flags"
@@ -81,6 +83,8 @@ var (
 
 	ErrNoDistinctID = errors.New("no distinct_id provided")
 
+	ErrQueueFull = errors.New("the message queue is full, the message was dropped")
+
 	ErrSDKDisabled = errors.New("posthog SDK is disabled because project API key is missing")
 )
 var ErrFlagNotFound = errors.New("feature flag not found")
@@ -322,6 +326,8 @@ type Config struct {
 
 	BatchSize int
 
+	MaxQueueSize int
+
 	Verbose bool
 
 	RetryAfter func(int) time.Duration

batching_test.go

@@ -61,7 +61,7 @@ func TestBatching_SmallEventsBatchTogether(t *testing.T) {
 	// With 500KB batch limit, should fit ~100 events per batch
 	client, err := NewWithConfig("test-key", Config{
 		Endpoint:  server.URL,
-		BatchSize: 250,             // Use default
+		BatchSize: DefaultBatchSize,
 		Interval:  5 * time.Second, // Long interval - rely on Close() to flush
 	})
 	require.NoError(t, err)
@@ -103,7 +103,7 @@ func TestBatching_LargeEventsTriggerFlush(t *testing.T) {
 	// With 500KB batch limit, should fit ~5 events per batch
 	client, err := NewWithConfig("test-key", Config{
 		Endpoint:  server.URL,
-		BatchSize: 250, // Use default - byte limit should trigger before count
+		BatchSize: DefaultBatchSize, // byte limit should trigger before count
 		Interval:  50 * time.Millisecond,
 	})
 	require.NoError(t, err)

benchmark_test.go

@@ -46,7 +46,13 @@ func BenchmarkConcurrentEnqueue(b *testing.B) {
 			client, _ := NewWithConfig("test-key", Config{
 				Transport: NoOpTransport(),
 				Callback:  callback,
-				// Uses production defaults for BatchSize, MaxEnqueuedRequests
+				// Discard logs: under sustained overload the consumer sheds batches
+				// (sendBatch backpressure) and logs from its own goroutine, which would
+				// flood stderr and skew ns/op. This benchmark measures pure enqueue
+				// throughput; stage-1 drops are counted via the returned error and
+				// stage-2 backpressure via the callback.
+				Logger: testLogger{},
+				// Uses production defaults for BatchSize, MaxEnqueuedRequests, MaxQueueSize
 			})
 			defer client.Close()
 
@@ -61,9 +67,18 @@ func BenchmarkConcurrentEnqueue(b *testing.B) {
 			})
 			b.StopTimer()
 
-			// Enqueue errors indicate channel full - fail immediately
-			if enqueueErrors.Load() > 0 {
-				b.Fatalf("Benchmark invalid: %d enqueue errors (msgs channel full)", enqueueErrors.Load())
+			// Enqueue drops the newest message (rather than blocking) once the msgs
+			// channel is full, so a sustained high-concurrency producer burst can
+			// outrun the single consumer loop. This is the expected overload ceiling,
+			// not an invalid measurement, so report the drop rate as a tracked metric
+			// instead of failing. (Under the previous blocking Enqueue this manifested
+			// as hidden caller latency rather than drops.)
+			drops := enqueueErrors.Load()
+			if b.N > 0 {
+				b.ReportMetric(float64(drops)/float64(b.N)*100, "drop%")
+			}
+			if drops > 0 {
+				b.Logf("Note: %d enqueue drops (msgs channel full at %d goroutines)", drops, concurrency)
 			}
 			// Delivery failures indicate batches channel backpressure - report but don't fail
 			if failures := callback.FailureCount(); failures > 0 {
@@ -98,7 +113,13 @@ func BenchmarkConcurrentEnqueueWithCardinality(b *testing.B) {
 			client, _ := NewWithConfig("test-key", Config{
 				Transport: NoOpTransport(),
 				Callback:  callback,
-				// Uses production defaults for BatchSize, MaxEnqueuedRequests
+				// Discard logs: under sustained overload the consumer sheds batches
+				// (sendBatch backpressure) and logs from its own goroutine, which would
+				// flood stderr and skew ns/op. This benchmark measures pure enqueue
+				// throughput; stage-1 drops are counted via the returned error and
+				// stage-2 backpressure via the callback.
+				Logger: testLogger{},
+				// Uses production defaults for BatchSize, MaxEnqueuedRequests, MaxQueueSize
 			})
 			defer client.Close()
 
@@ -113,9 +134,18 @@ func BenchmarkConcurrentEnqueueWithCardinality(b *testing.B) {
 			})
 			b.StopTimer()
 
-			// Enqueue errors indicate channel full - fail immediately
-			if enqueueErrors.Load() > 0 {
-				b.Fatalf("Benchmark invalid: %d enqueue errors (msgs channel full)", enqueueErrors.Load())
+			// Enqueue drops the newest message (rather than blocking) once the msgs
+			// channel is full, so a sustained high-concurrency producer burst can
+			// outrun the single consumer loop. This is the expected overload ceiling,
+			// not an invalid measurement, so report the drop rate as a tracked metric
+			// instead of failing. (Under the previous blocking Enqueue this manifested
+			// as hidden caller latency rather than drops.)
+			drops := enqueueErrors.Load()
+			if b.N > 0 {
+				b.ReportMetric(float64(drops)/float64(b.N)*100, "drop%")
+			}
+			if drops > 0 {
+				b.Logf("Note: %d enqueue drops (msgs channel full at %d goroutines)", drops, concurrency)
 			}
 			// Delivery failures indicate batches channel backpressure - report but don't fail
 			if failures := callback.FailureCount(); failures > 0 {

config.go

@@ -157,6 +157,18 @@ type Config struct {
 	// it defaults to DefaultBatchSize. The API still enforces a 500KB request limit.
 	BatchSize int
 
+	// MaxQueueSize is the maximum number of messages buffered in memory waiting to
+	// be batched and sent. If zero, it defaults to DefaultMaxQueueSize. It is
+	// clamped up to BatchSize so the queue can always hold at least one full batch.
+	//
+	// When the queue is full, Enqueue drops the newest message rather than blocking
+	// the caller and returns ErrQueueFull (the drop is not reported via
+	// Callback.Failure). Bulk or backfill workloads that enqueue faster than the
+	// client can upload should check the error returned by Enqueue for ErrQueueFull
+	// and throttle or retry (or raise MaxQueueSize), otherwise events are dropped,
+	// not delayed.
+	MaxQueueSize int
+
 	// Verbose enables more frequent and detailed debug logging through Logger.
 	Verbose bool
 
@@ -236,7 +248,7 @@ const (
 	DefaultFeatureFlagRequestTimeout = 3 * time.Second
 
 	// DefaultBatchSize is the default batch size used when Config.BatchSize is zero.
-	DefaultBatchSize = 250
+	DefaultBatchSize = 100
 
 	// DefaultMaxAttempts is the total number of capture delivery attempts (1
 	// initial + retries) used when Config.MaxRetries is unset or out of range.
@@ -256,6 +268,11 @@ const (
 	// DefaultMaxEnqueuedRequests is the default maximum number of batches that
 	// can be queued for sending.
 	DefaultMaxEnqueuedRequests = 1000
+
+	// DefaultMaxQueueSize is the default in-memory message queue capacity used when
+	// Config.MaxQueueSize is zero. It matches the posthog-python, posthog-rs, and
+	// posthog-node defaults so backend SDKs behave consistently under bursty load.
+	DefaultMaxQueueSize = 10000
 )
 
 func (c *Config) normalize() {
@@ -293,6 +310,14 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if c.MaxQueueSize < 0 {
+		return ConfigError{
+			Reason: "negative queue sizes are not supported",
+			Field:  "MaxQueueSize",
+			Value:  c.MaxQueueSize,
+		}
+	}
+
 	if _, err := url.Parse(c.Endpoint); err != nil {
 		return ConfigError{
 			Reason: "invalid endpoint",
@@ -381,6 +406,15 @@ func makeConfig(c Config) Config {
 		c.BatchSize = DefaultBatchSize
 	}
 
+	if c.MaxQueueSize == 0 {
+		c.MaxQueueSize = DefaultMaxQueueSize
+	}
+	// The queue must be able to hold at least one full batch, otherwise a batch
+	// could never accumulate before the queue overflows.
+	if c.MaxQueueSize < c.BatchSize {
+		c.MaxQueueSize = c.BatchSize
+	}
+
 	if c.RetryAfter == nil {
 		c.RetryAfter = DefaultBackoff().Duration
 	}

config_test.go

@@ -110,6 +110,45 @@ func TestConfigInvalidBatchSize(t *testing.T) {
 	}
 }
 
+func TestMakeConfigBatchSizeDefault(t *testing.T) {
+	require.Equal(t, 100, DefaultBatchSize, "backend SDKs share a BatchSize default of 100")
+	require.Equal(t, DefaultBatchSize, makeConfig(Config{}).BatchSize)
+}
+
+func TestMakeConfigMaxQueueSize(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		batchSize    int
+		maxQueueSize int
+		want         int
+	}{
+		{"defaults", 0, 0, DefaultMaxQueueSize},
+		{"explicit override honored", 1, 42, 42},
+		{"clamped up to explicit BatchSize", 500, 10, 500},
+		{"clamped up to BatchSize larger than default", 20000, 0, 20000},
+		{"equal to BatchSize is left untouched", 100, 100, 100},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got := makeConfig(Config{BatchSize: tc.batchSize, MaxQueueSize: tc.maxQueueSize})
+			require.Equal(t, tc.want, got.MaxQueueSize)
+			require.GreaterOrEqual(t, got.MaxQueueSize, got.BatchSize,
+				"the queue must always hold at least one full batch")
+		})
+	}
+}
+
+func TestConfigInvalidMaxQueueSize(t *testing.T) {
+	c := Config{MaxQueueSize: -1}
+
+	err := c.Validate()
+	require.Error(t, err)
+
+	e, ok := err.(ConfigError)
+	require.True(t, ok, "expected a ConfigError, got %T", err)
+	require.Equal(t, "MaxQueueSize", e.Field)
+	require.Equal(t, -1, e.Value.(int))
+}
+
 func TestConfigBoolDefaults(t *testing.T) {
 	tests := []struct {
 		name string

error.go

@@ -89,6 +89,11 @@ var (
 	// operation but was not provided.
 	ErrNoDistinctID = errors.New("no distinct_id provided")
 
+	// ErrQueueFull is returned by Enqueue/EnqueueWithContext when the in-memory
+	// message queue is full. The message is dropped (not retried) and the drop is
+	// reported only through this returned error, not through Callback.Failure.
+	ErrQueueFull = errors.New("the message queue is full, the message was dropped")
+
 	// ErrSDKDisabled is returned when the SDK is disabled because the project API key is missing.
 	ErrSDKDisabled = errors.New("posthog SDK is disabled because project API key is missing")
 )

posthog.go

@@ -49,7 +49,14 @@ type EnqueueClient interface {
 	//	client.Close()
 	//
 	// Enqueue returns an error if the message could not be queued, which happens
-	// when the client is closed or the message is invalid.
+	// when the client is closed, the message is invalid, or the in-memory queue
+	// is full (ErrQueueFull) -- in the latter case the message is dropped rather
+	// than blocking the caller. A full-queue drop is reported only through this
+	// returned error, not through Callback.Failure.
+	//
+	// Bulk or backfill workloads that can enqueue faster than the client uploads
+	// should check the returned error for ErrQueueFull and throttle or retry (or
+	// raise Config.MaxQueueSize); otherwise events are dropped, not delayed.
 	Enqueue(Message) error
 }
 
@@ -245,10 +252,11 @@ func NewWithConfig(apiKey string, config Config) (cli Client, err error) {
 	}
 
 	// Channel sizing:
-	// - batches queue sized by MaxEnqueuedRequests (default 1000)
-	// - msgs queue sized to hold multiple batches worth of messages
+	// - msgs queue (incoming messages) sized by MaxQueueSize (default 10000)
+	// - batches queue (prepared batches awaiting upload) sized by MaxEnqueuedRequests (default 1000)
+	// Both defaults and the MaxQueueSize >= BatchSize clamp are applied in makeConfig.
 	batchesQueueSize := config.MaxEnqueuedRequests
-	msgQueueSize := max(100, config.BatchSize*10)
+	msgQueueSize := config.MaxQueueSize
 
 	ctx, cancel := context.WithCancel(context.Background())
 	c := &client{
@@ -555,7 +563,11 @@ func (c *client) EnqueueWithContext(ctx context.Context, msg Message) (err error
 
 	var ts = c.now()
 
-	// Helper to send prepared message with panic recovery
+	// Helper to send prepared message with panic recovery. Non-blocking: if the
+	// `msgs` queue is full, the new message is dropped and ErrQueueFull is returned
+	// rather than blocking the caller, matching the posthog-python and posthog-rs
+	// SDKs. The drop is reported only via the returned error -- no callback or log
+	// is invoked on the caller's goroutine, so a sustained overload stays cheap.
 	sendPrepared := func(prepared preparedMessage) {
 		defer func() {
 			// When the `msgs` channel is closed writing to it will trigger a panic.
@@ -566,7 +578,11 @@ func (c *client) EnqueueWithContext(ctx context.Context, msg Message) (err error
 				err = ErrClosed
 			}
 		}()
-		c.msgs <- prepared
+		select {
+		case c.msgs <- prepared:
+		default:
+			err = ErrQueueFull
+		}
 	}
 
 	switch m := msg.(type) {