Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/invalid-event-uuid-warning.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"posthog-go": patch
---

An invalid event `Uuid` is still replaced with a generated one, and the SDK now logs one warning when it does. The warning names the rule and not the value, because callers can put their own data in the field. An empty `Uuid` is generated without a warning. The same rule now applies to a `Uuid` set in `BeforeSend`. Before, an invalid or empty `Uuid` from the hook was sent as is, and capture rejected the whole batch.
5 changes: 5 additions & 0 deletions .changeset/session-id-lift.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"posthog-go": major
---

Capture v1 requires `$session_id` and `$window_id` to be strings. The SDK sends a value whose JSON form is a string, such as a `string` (including `""`), a named string type or a `uuid.UUID`. It drops any other value and logs a warning that names the key and the value's type, never the value. A `nil` value drops without a warning. In 1.x any value was sent as a property. See `docs/migration-v2.md`.
2 changes: 0 additions & 2 deletions api/public-api.txt
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,6 @@ const (

DefaultMaxEnqueuedRequests = 1000

DefaultMaxEventBytes = 500000

DefaultMaxBatchBytes = 500000

DefaultCaptureAIBatchUploadTimeout = 30 * time.Second
Expand Down
60 changes: 31 additions & 29 deletions batching_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -126,51 +126,53 @@ func TestBatching_LargeEventsTriggerFlush(t *testing.T) {
require.ElementsMatch(t, []int{4, 4, 2}, batchSizes)
}

// TestBatching_OversizedEventRejected verifies that events >500KB are rejected
func TestBatching_OversizedEventRejected(t *testing.T) {
// TestBatching_EventOverBatchTargetShipsAlone verifies that analytics has no
// per-event cap by default: an event larger than MaxBatchBytes is sent in a
// request of its own, and the events around it are not held back.
func TestBatching_EventOverBatchTargetShipsAlone(t *testing.T) {
t.Parallel()

var received atomic.Int64
var batchEvents [][]string
var mu sync.Mutex
var failureCount atomic.Int64

server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
var b eventBatch
json.Unmarshal(body, &b)
received.Add(int64(len(b.Batch)))
var b struct {
Batch []struct {
Event string `json:"event"`
} `json:"batch"`
}
require.NoError(t, json.Unmarshal(body, &b))
var names []string
for _, e := range b.Batch {
names = append(names, e.Event)
}
mu.Lock()
batchEvents = append(batchEvents, names)
mu.Unlock()
writeCaptureOK(w, body)
}))
defer server.Close()

callback := &testCallbackCounter{
onFailure: func() { failureCount.Add(1) },
}

client, err := NewWithConfig("test-key", Config{
Endpoint: server.URL,
Callback: callback,
Interval: time.Hour,
Callback: &testCallbackCounter{onFailure: func() { failureCount.Add(1) }},
})
require.NoError(t, err)

// Create an event with MANY properties to exceed 500KB
// Each property is ~50 bytes, need ~10000 properties to hit 500KB
oversizedProps := make(Properties, 15000)
for i := 0; i < 15000; i++ {
oversizedProps[generateDistinctId(i)] = generateDistinctId(i + 100000)
}

err = client.Enqueue(Capture{
DistinctId: "user_1",
Event: "oversized_event",
Properties: oversizedProps,
})
require.NoError(t, err) // Enqueue itself doesn't fail

client.Close()
require.NoError(t, client.Enqueue(Capture{DistinctId: "user", Event: "before"}))
require.NoError(t, client.Enqueue(Capture{
DistinctId: "user", Event: "oversized",
Properties: Properties{"payload": strings.Repeat("x", DefaultMaxBatchBytes+200000)},
}))
require.NoError(t, client.Enqueue(Capture{DistinctId: "user", Event: "after"}))
require.NoError(t, client.Close())

// The oversized event should be rejected via callback
require.Equal(t, int64(0), received.Load(), "Oversized event should not be delivered")
require.Equal(t, int64(1), failureCount.Load(), "Should have 1 failure callback for oversized event")
require.Zero(t, failureCount.Load())
// Batches upload concurrently, so only the split is fixed, not the arrival order.
require.ElementsMatch(t, [][]string{{"before"}, {"oversized"}, {"after"}}, batchEvents)
}

// TestBatching_MixedCardinalityBatching verifies correct batching with mixed event sizes
Expand Down
4 changes: 2 additions & 2 deletions benchmark_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ func BenchmarkAPIfy(b *testing.B) {
func BenchmarkCompressionOverhead(b *testing.B) {
for _, card := range []PropertyCardinality{CardinalityLow, CardinalityMedium, CardinalityHigh} {
capture := NewEventPoolWithCardinality(1, card).Next()
raw, _, _, err := prepareForSend(capture)
raw, _, _, err := prepareForSend(capture, nil)
if err != nil {
b.Fatal(err)
}
Expand Down Expand Up @@ -263,7 +263,7 @@ func BenchmarkCompressionRatio(b *testing.B) {
for _, card := range []PropertyCardinality{CardinalityLow, CardinalityMedium, CardinalityHigh} {
b.Run(CardinalityName(card), func(b *testing.B) {
capture := NewEventPoolWithCardinality(1, card).Next()
raw, _, _, err := prepareForSend(capture)
raw, _, _, err := prepareForSend(capture, nil)
if err != nil {
b.Fatal(err)
}
Expand Down
15 changes: 7 additions & 8 deletions capture_ai_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -231,11 +231,10 @@ func TestAILaneSizeLimits(t *testing.T) {
analytics := analyticsLaneConfig(makeConfig(Config{}))
ai := aiLaneConfig(makeConfig(Config{}))

require.Equal(t, DefaultMaxEventBytes, analytics.maxEventBytes)
require.Zero(t, analytics.maxEventBytes, "analytics has no per-event cap by default")
require.Equal(t, DefaultMaxBatchBytes, analytics.maxBatchBytes)
require.Equal(t, aiMaxEventBytes+aiEnvelopeHeadroom, ai.maxEventBytes)
require.Equal(t, aiBatchBytesTarget, ai.maxBatchBytes)
require.Greater(t, ai.maxEventBytes, analytics.maxEventBytes,
"an AI event well over the analytics cap must still be accepted")

require.Equal(t, DefaultCaptureAIMaxQueueSize, ai.maxQueueSize)
require.Equal(t, DefaultMaxQueueSize, analytics.maxQueueSize)
Expand All @@ -244,8 +243,8 @@ func TestAILaneSizeLimits(t *testing.T) {
}

// TestAILaneAcceptsEventOverAnalyticsCap is the behavioral half of the limits:
// a payload larger than the analytics per-event cap reaches the AI endpoint.
// Production AI events routinely exceed that cap.
// a payload larger than the analytics batch target reaches the AI endpoint.
// Production AI events routinely exceed it.
func TestAILaneAcceptsEventOverAnalyticsCap(t *testing.T) {
rec := newLaneRecorder()
srv := rec.server(t)
Expand All @@ -261,7 +260,7 @@ func TestAILaneAcceptsEventOverAnalyticsCap(t *testing.T) {
}}
})

big := NewProperties().Set("blob", strings.Repeat("x", DefaultMaxEventBytes+1000))
big := NewProperties().Set("blob", strings.Repeat("x", DefaultMaxBatchBytes+1000))
require.NoError(t, c.EnqueueAI(Capture{DistinctId: "d", Event: "$ai_generation", Properties: big}))
require.NoError(t, c.Close())

Expand Down Expand Up @@ -569,11 +568,11 @@ func TestAILaneSendsEventAtTheEndpointCeiling(t *testing.T) {
// Measure the non-blob overhead, then size the blob so the whole event lands
// just past the endpoint's ceiling -- the window the headroom covers.
probe, _, _, err := prepareForSend(
Capture{DistinctId: "d", Event: "$ai_generation", Properties: NewProperties().Set("blob", "")})
Capture{DistinctId: "d", Event: "$ai_generation", Properties: NewProperties().Set("blob", "")}, nil)
require.NoError(t, err)
props := NewProperties().Set("blob", strings.Repeat("x", aiMaxEventBytes-len(probe)+256))

data, _, _, err := prepareForSend(Capture{DistinctId: "d", Event: "$ai_generation", Properties: props})
data, _, _, err := prepareForSend(Capture{DistinctId: "d", Event: "$ai_generation", Properties: props}, nil)
require.NoError(t, err)
require.Greater(t, len(data), aiMaxEventBytes,
"precondition: the whole event must exceed the endpoint ceiling")
Expand Down
17 changes: 14 additions & 3 deletions capture_errors_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -104,21 +104,32 @@ func TestTransportErrorUnwraps(t *testing.T) {
}
}

// capturingLogger records Debugf format strings for assertions.
// capturingLogger records Debugf and Warnf messages for assertions.
type capturingLogger struct {
mu sync.Mutex
debugf []string
warnf []string
}

func (l *capturingLogger) Debugf(f string, a ...interface{}) {
l.mu.Lock()
l.debugf = append(l.debugf, fmt.Sprintf(f, a...))
l.mu.Unlock()
}
func (l *capturingLogger) Logf(string, ...interface{}) {}
func (l *capturingLogger) Warnf(string, ...interface{}) {}
func (l *capturingLogger) Logf(string, ...interface{}) {}
func (l *capturingLogger) Warnf(f string, a ...interface{}) {
l.mu.Lock()
l.warnf = append(l.warnf, fmt.Sprintf(f, a...))
l.mu.Unlock()
}
func (l *capturingLogger) Errorf(string, ...interface{}) {}

func (l *capturingLogger) warnings() []string {
l.mu.Lock()
defer l.mu.Unlock()
return append([]string(nil), l.warnf...)
}

func (l *capturingLogger) debugContains(sub string) bool {
l.mu.Lock()
defer l.mu.Unlock()
Expand Down
80 changes: 66 additions & 14 deletions capture_event.go
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,8 @@ type eventPayload struct {
Uuid string `json:"uuid"`
DistinctId string `json:"distinct_id"`
Timestamp time.Time `json:"timestamp"`
SessionId string `json:"session_id,omitempty"`
WindowId string `json:"window_id,omitempty"`
SessionId *string `json:"session_id,omitempty"`
WindowId *string `json:"window_id,omitempty"`
Options map[string]interface{} `json:"options"`
Properties Properties `json:"properties"`
}
Expand Down Expand Up @@ -103,7 +103,7 @@ type apiEvent struct {
// $window_id into top-level fields, and returns the wire payload. It mutates
// e.properties by deleting the moved keys; callers must ensure the properties
// map is not shared. e.options is copied, never mutated.
func buildEvent(e apiEvent) eventPayload {
func buildEvent(e apiEvent, logger Logger) eventPayload {
props := e.properties
if props == nil {
props = Properties{}
Expand All @@ -112,8 +112,8 @@ func buildEvent(e apiEvent) eventPayload {
for _, key := range sdkInfoProperties {
delete(props, key)
}
sessionId := liftStringProperty(props, propertySessionID)
windowId := liftStringProperty(props, propertyWindowID)
sessionId := liftStringProperty(props, propertySessionID, logger)
windowId := liftStringProperty(props, propertyWindowID, logger)
return eventPayload{
Event: e.event,
Uuid: e.uuid,
Expand Down Expand Up @@ -161,16 +161,63 @@ func isNilValue(v interface{}) bool {
return false
}

// liftStringProperty removes key from props and returns its value when it is a
// string, or "" otherwise.
func liftStringProperty(props Properties, key string) string {
// liftStringProperty removes key from props and returns its value when the
// value's JSON form is a string, such as a uuid.UUID or a named string type.
// Capture rejects the whole request when the field is not a string, so any
// other value is dropped with a warning that names its type, never the value.
// A nil value counts as unset and drops silently.
func liftStringProperty(props Properties, key string, logger Logger) *string {
v, ok := props[key]
if !ok {
return ""
return nil
}
delete(props, key)
s, _ := v.(string)
return s
if s, ok := v.(string); ok {
return &s
}
if v == nil {
return nil
}
data, err := json.Marshal(v)
if err != nil {
warnDroppedProperty(logger, key, fmt.Sprintf("%T", v))
return nil
}
// Unmarshaling "null" into a string succeeds and yields "", so check the
// JSON type first.
if data[0] == '"' {
var s string
if json.Unmarshal(data, &s) == nil {
return &s
}
}
valueType := jsonTypeName(data)
if valueType != "null" {
warnDroppedProperty(logger, key, valueType)
}
return nil
}

// jsonTypeName names the JSON type of an encoded value.
func jsonTypeName(data []byte) string {
switch data[0] {
case 'n':
return "null"
case 't', 'f':
return "bool"
case '{':
return "object"
case '[':
return "array"
default:
return "number"
}
}

func warnDroppedProperty(logger Logger, key, valueType string) {
if logger != nil {
logger.Warnf("dropping %s: a %s value is not a string", key, valueType)
}
}

// fillSDKProperties fills the values the SDK adds to every event: system
Expand Down Expand Up @@ -199,9 +246,9 @@ func boolProperty(props Properties, key string) bool {
// and returns the event uuid for per-event result correlation. The uuid is
// canonicalized because BeforeSend may set any form capture accepts, while
// capture keys its results by the canonical form.
func prepareForSend(msg Message) (json.RawMessage, APIMessage, string, error) {
func prepareForSend(msg Message, logger Logger) (json.RawMessage, APIMessage, string, error) {
apiMsg := msg.APIfy()
ev := buildEvent(msg.apifyEvent())
ev := buildEvent(msg.apifyEvent(), logger)
data, err := json.Marshal(ev)
if err != nil {
return nil, apiMsg, canonicalUUID(ev.Uuid), err
Expand Down Expand Up @@ -263,10 +310,15 @@ func (msg GroupIdentify) apifyEvent() apiEvent {
myProperties.Set("$group_set", mergeOverNested(myProperties["$group_set"], msg.Properties))
}

distinctID := msg.DistinctId
if distinctID == "" {
distinctID = groupDistinctID(msg.Type, msg.Key)
}

return apiEvent{
event: "$groupidentify",
uuid: msg.Uuid,
distinctId: fmt.Sprintf("$%s_%s", msg.Type, msg.Key),
distinctId: distinctID,
timestamp: msg.Timestamp,
properties: myProperties,
options: msg.Options,
Expand Down
Loading
Loading