diff --git a/.changeset/tidy-queues-drop-newest.md b/.changeset/tidy-queues-drop-newest.md new file mode 100644 index 00000000..705dd6d9 --- /dev/null +++ b/.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. diff --git a/api/public-api.txt b/api/public-api.txt index 3c6baa3c..595f5aa5 100644 --- a/api/public-api.txt +++ b/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 diff --git a/batching_test.go b/batching_test.go index ac00432e..22b8b3ea 100644 --- a/batching_test.go +++ b/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) diff --git a/benchmark_test.go b/benchmark_test.go index 9ab0c702..83908069 100644 --- a/benchmark_test.go +++ b/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 { diff --git a/config.go b/config.go index ab863766..040010fd 100644 --- a/config.go +++ b/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 } diff --git a/config_test.go b/config_test.go index 17c152cb..da5f71bc 100644 --- a/config_test.go +++ b/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 diff --git a/error.go b/error.go index 6734d883..f763b15a 100644 --- a/error.go +++ b/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") ) diff --git a/posthog.go b/posthog.go index 392ce625..b0831a20 100644 --- a/posthog.go +++ b/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) { diff --git a/queue_overflow_test.go b/queue_overflow_test.go new file mode 100644 index 00000000..1ddfa518 --- /dev/null +++ b/queue_overflow_test.go @@ -0,0 +1,90 @@ +package posthog + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// newQueueTestClient builds a client whose consumer loop is deliberately NOT +// started. That lets a test fill the msgs channel to capacity and be certain +// nothing ever drains it, which is the only way to exercise the Enqueue +// drop-on-full path deterministically: a running consumer drains msgs into +// per-batch goroutines faster than a single-threaded producer can overflow it, +// so overflow is otherwise reachable only via a producer/consumer race. +func newQueueTestClient(capacity int, logger Logger, callback Callback) *client { + c := &client{ + Config: Config{ + now: time.Now, + Logger: logger, + Callback: callback, + }, + msgs: make(chan preparedMessage, capacity), + } + c.capture = legacyCapturer{c} + return c +} + +func captureOverflowEvent(event string) Capture { + return Capture{ + Event: event, + DistinctId: "user-1", + SendFeatureFlags: SendFeatureFlags(false), + } +} + +func TestEnqueue_DropsNewestWhenQueueFull(t *testing.T) { + const capacity = 4 + + var failures []error + var warnings []string + + logger := testLogger{logf: func(format string, args ...interface{}) { + warnings = append(warnings, format) + }} + callback := testCallback{failure: func(_ APIMessage, err error) { + failures = append(failures, err) + }} + + c := newQueueTestClient(capacity, logger, callback) + + // Saturate the queue so no further message can be buffered. + for i := 0; i < capacity; i++ { + c.msgs <- preparedMessage{} + } + + // Enqueue must not block even though the queue is full and has no consumer; + // a blocking-send implementation would hang here forever. + done := make(chan error, 1) + go func() { + done <- c.Enqueue(captureOverflowEvent("overflow")) + }() + + var err error + select { + case err = <-done: + case <-time.After(2 * time.Second): + t.Fatal("Enqueue blocked on a full queue; drop-on-full must be non-blocking") + } + + require.ErrorIs(t, err, ErrQueueFull, "Enqueue must return ErrQueueFull when the queue is full") + require.Len(t, c.msgs, capacity, "the dropped message must not be buffered") + + // The drop is reported only through the returned error. Doing callback or log + // work here would run on the caller's goroutine, re-introducing the latency and + // re-entrancy the non-blocking drop path exists to avoid. + require.Empty(t, failures, "Callback.Failure must not run on the caller's Enqueue goroutine") + require.Empty(t, warnings, "the drop path must not log synchronously on the caller's goroutine") +} + +func TestEnqueue_BuffersWhenQueueHasRoom(t *testing.T) { + var failures int + + callback := testCallback{failure: func(APIMessage, error) { failures++ }} + c := newQueueTestClient(2, testLogger{}, callback) + + require.NoError(t, c.Enqueue(captureOverflowEvent("buffered"))) + require.Len(t, c.msgs, 1, "a message enqueued with room to spare must be buffered, not dropped") + require.Zero(t, failures, "no failure callback should fire on the happy path") +}