From 25f67ddda060f698c5e79435970405965372cf28 Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 14 Jul 2026 17:03:35 -0700 Subject: [PATCH 1/3] feat: drop newest on full queue, add configurable MaxQueueSize Enqueue now drops the newest event and returns ErrQueueFull (invoking Callback.Failure) when the in-memory queue is full, instead of blocking the caller. This matches posthog-python and posthog-rs. - add Config.MaxQueueSize (default 10000), clamped up to BatchSize, replacing the hardcoded BatchSize*10 msgs channel sizing - change default BatchSize from 250 to 100 for cross-SDK consistency - document the drop-vs-block tradeoff for bulk/backfill callers on Enqueue and MaxQueueSize - benchmarks report a drop% metric instead of failing on overflow and discard logs to keep ns/op clean --- .changeset/tidy-queues-drop-newest.md | 9 +++ batching_test.go | 4 +- benchmark_test.go | 44 ++++++++++--- config.go | 35 ++++++++++- config_test.go | 39 ++++++++++++ error.go | 5 ++ posthog.go | 27 ++++++-- queue_overflow_test.go | 90 +++++++++++++++++++++++++++ 8 files changed, 236 insertions(+), 17 deletions(-) create mode 100644 .changeset/tidy-queues-drop-newest.md create mode 100644 queue_overflow_test.go diff --git a/.changeset/tidy-queues-drop-newest.md b/.changeset/tidy-queues-drop-newest.md new file mode 100644 index 00000000..d613c9c0 --- /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` (and invokes `Callback.Failure`) 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. **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/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..f1920586 100644 --- a/benchmark_test.go +++ b/benchmark_test.go @@ -46,7 +46,12 @@ func BenchmarkConcurrentEnqueue(b *testing.B) { client, _ := NewWithConfig("test-key", Config{ Transport: NoOpTransport(), Callback: callback, - // Uses production defaults for BatchSize, MaxEnqueuedRequests + // Discard logs: this benchmark measures pure enqueue throughput, and + // the drop-on-full path would otherwise flood stderr from inside the + // timed loop, polluting ns/op. Drops are counted via the returned + // error and callback; drop-log severity is covered by unit tests. + Logger: testLogger{}, + // Uses production defaults for BatchSize, MaxEnqueuedRequests, MaxQueueSize }) defer client.Close() @@ -61,9 +66,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 +112,12 @@ func BenchmarkConcurrentEnqueueWithCardinality(b *testing.B) { client, _ := NewWithConfig("test-key", Config{ Transport: NoOpTransport(), Callback: callback, - // Uses production defaults for BatchSize, MaxEnqueuedRequests + // Discard logs: this benchmark measures pure enqueue throughput, and + // the drop-on-full path would otherwise flood stderr from inside the + // timed loop, polluting ns/op. Drops are counted via the returned + // error and callback; drop-log severity is covered by unit tests. + Logger: testLogger{}, + // Uses production defaults for BatchSize, MaxEnqueuedRequests, MaxQueueSize }) defer client.Close() @@ -113,9 +132,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..e017c4eb 100644 --- a/config.go +++ b/config.go @@ -157,6 +157,17 @@ 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: it returns ErrQueueFull and invokes 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 +247,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 +267,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 +309,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 +405,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..0defe4da 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); Callback.Failure + // is invoked with this error if a Callback is configured. + 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..67012a36 100644 --- a/posthog.go +++ b/posthog.go @@ -49,7 +49,13 @@ 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, and Callback.Failure is invoked if configured. + // + // 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 +251,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 +562,9 @@ 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 rather than blocking the + // caller, matching the posthog-python and posthog-rs SDKs. sendPrepared := func(prepared preparedMessage) { defer func() { // When the `msgs` channel is closed writing to it will trigger a panic. @@ -566,7 +575,13 @@ func (c *client) EnqueueWithContext(ctx context.Context, msg Message) (err error err = ErrClosed } }() - c.msgs <- prepared + select { + case c.msgs <- prepared: + default: + err = ErrQueueFull + c.Warnf("message queue is full (capacity %d), dropping the newest message", cap(c.msgs)) + c.notifyFailure([]APIMessage{prepared.msg}, ErrQueueFull) + } } switch m := msg.(type) { diff --git a/queue_overflow_test.go b/queue_overflow_test.go new file mode 100644 index 00000000..d6521307 --- /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") + + require.Len(t, failures, 1, "Callback.Failure must fire exactly once for the dropped message") + require.ErrorIs(t, failures[0], ErrQueueFull) + + require.Len(t, warnings, 1, "exactly one warning should be logged for the drop") + require.Contains(t, warnings[0], "queue is full") +} + +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") +} From 73473a6bea8178cd18f94872120cb33f1bade95f Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 14 Jul 2026 17:28:14 -0700 Subject: [PATCH 2/3] chore: update public API snapshot for MaxQueueSize Regenerated api/public-api.txt via bin/api-diff --update to reflect the new Config.MaxQueueSize field, DefaultMaxQueueSize/ErrQueueFull additions, and the DefaultBatchSize 250->100 change. --- api/public-api.txt | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) 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 From 3e36a1e133da675eab4ead686905a55cfd83a05d Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 14 Jul 2026 18:29:35 -0700 Subject: [PATCH 3/3] fix: report full-queue drop only via ErrQueueFull, not callback or log Addresses PR #258 review: the Enqueue drop-on-full branch invoked Callback.Failure and Warnf synchronously on the caller's goroutine. - Callbacks are contractually run from the client's internal goroutines; firing one from Enqueue could re-enter or block the caller's request path. Drop the notifyFailure call; the caller learns via the returned ErrQueueFull. This also makes the closed-channel race drop (ErrClosed, no callback) and the full-queue drop consistent. - Remove the per-event Warnf: under sustained overload it re-created the caller latency the non-blocking drop path exists to avoid. Update the white-box test to assert the drop fires no callback and logs nothing on the caller goroutine, and refresh the now-stale benchmark log-silencing comment (the flood came from stage-2 sendBatch backpressure logging, not the drop path). --- .changeset/tidy-queues-drop-newest.md | 2 +- benchmark_test.go | 18 ++++++++++-------- config.go | 9 +++++---- error.go | 4 ++-- posthog.go | 11 ++++++----- queue_overflow_test.go | 10 +++++----- 6 files changed, 29 insertions(+), 25 deletions(-) diff --git a/.changeset/tidy-queues-drop-newest.md b/.changeset/tidy-queues-drop-newest.md index d613c9c0..705dd6d9 100644 --- a/.changeset/tidy-queues-drop-newest.md +++ b/.changeset/tidy-queues-drop-newest.md @@ -4,6 +4,6 @@ Drop the newest event instead of blocking when the in-memory queue is full, and make the queue size configurable. -- `Enqueue` now returns `ErrQueueFull` (and invokes `Callback.Failure`) 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. **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. +- `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/benchmark_test.go b/benchmark_test.go index f1920586..83908069 100644 --- a/benchmark_test.go +++ b/benchmark_test.go @@ -46,10 +46,11 @@ func BenchmarkConcurrentEnqueue(b *testing.B) { client, _ := NewWithConfig("test-key", Config{ Transport: NoOpTransport(), Callback: callback, - // Discard logs: this benchmark measures pure enqueue throughput, and - // the drop-on-full path would otherwise flood stderr from inside the - // timed loop, polluting ns/op. Drops are counted via the returned - // error and callback; drop-log severity is covered by unit tests. + // 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 }) @@ -112,10 +113,11 @@ func BenchmarkConcurrentEnqueueWithCardinality(b *testing.B) { client, _ := NewWithConfig("test-key", Config{ Transport: NoOpTransport(), Callback: callback, - // Discard logs: this benchmark measures pure enqueue throughput, and - // the drop-on-full path would otherwise flood stderr from inside the - // timed loop, polluting ns/op. Drops are counted via the returned - // error and callback; drop-log severity is covered by unit tests. + // 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 }) diff --git a/config.go b/config.go index e017c4eb..040010fd 100644 --- a/config.go +++ b/config.go @@ -162,10 +162,11 @@ type Config struct { // 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: it returns ErrQueueFull and invokes 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. + // 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. diff --git a/error.go b/error.go index 0defe4da..f763b15a 100644 --- a/error.go +++ b/error.go @@ -90,8 +90,8 @@ var ( 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); Callback.Failure - // is invoked with this error if a Callback is configured. + // 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. diff --git a/posthog.go b/posthog.go index 67012a36..b0831a20 100644 --- a/posthog.go +++ b/posthog.go @@ -51,7 +51,8 @@ type EnqueueClient interface { // Enqueue returns an error if the message could not be queued, which happens // 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, and Callback.Failure is invoked if configured. + // 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 @@ -563,8 +564,10 @@ func (c *client) EnqueueWithContext(ctx context.Context, msg Message) (err error var ts = c.now() // Helper to send prepared message with panic recovery. Non-blocking: if the - // `msgs` queue is full, the new message is dropped rather than blocking the - // caller, matching the posthog-python and posthog-rs SDKs. + // `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. @@ -579,8 +582,6 @@ func (c *client) EnqueueWithContext(ctx context.Context, msg Message) (err error case c.msgs <- prepared: default: err = ErrQueueFull - c.Warnf("message queue is full (capacity %d), dropping the newest message", cap(c.msgs)) - c.notifyFailure([]APIMessage{prepared.msg}, ErrQueueFull) } } diff --git a/queue_overflow_test.go b/queue_overflow_test.go index d6521307..1ddfa518 100644 --- a/queue_overflow_test.go +++ b/queue_overflow_test.go @@ -71,11 +71,11 @@ func TestEnqueue_DropsNewestWhenQueueFull(t *testing.T) { 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") - require.Len(t, failures, 1, "Callback.Failure must fire exactly once for the dropped message") - require.ErrorIs(t, failures[0], ErrQueueFull) - - require.Len(t, warnings, 1, "exactly one warning should be logged for the drop") - require.Contains(t, warnings[0], "queue is full") + // 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) {