From 388bb1208c88dad30305ca129d24d17c01af0fbe Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Mon, 20 Jul 2026 16:13:09 +0400 Subject: [PATCH 1/2] fix(ingest): silent success on quota block + stream-id metric label (JITSU-88) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two changes preparing ingest for the JITSU-88 usage-quota block, which works by setting throttle=100 on a workspace. 1. Throttled events now return a success response instead of HTTP 402. V1 blocking must be silent — the client should see no ingestion error (events are still preserved in backup and simply not delivered to destinations). sendToRotor writes {"ok":true} (or the tracking gif for the pixel endpoint) and returns the throttle marker, which still drives the SKIPPED events-log status, the `throttled` metric and the dead-letter copy. In the batch handler a throttled event counts toward okEvents instead of failing the batch. The classic handler's trailing success write is now guarded by !Written() — gin appends body writes rather than guarding them, so this also fixes a pre-existing double-body write on every sendToRotor error path there. 2. IngestHandlerRequests now labels `slug` with metricsId (the resolved stream id) instead of DefaultString(loc.Slug, loc.Domain), which was ambiguously a slug or a domain. Throttled counts are always post-resolution, so blocked-event volume can now be read per stream from bulkerapp_handler_ingest{status="throttled"} and mapped to a workspace. NOTE: this changes the `slug` label across all ingest statuses (success and error too) — Grafana panels keying on the old domain values need a look before deploy. Co-Authored-By: Claude Fable 5 --- bulker/ingest/router.go | 16 +++++++++++++++- bulker/ingest/router_batch_handler.go | 16 ++++++++++++---- bulker/ingest/router_classic_handler.go | 14 ++++++++++---- bulker/ingest/router_funcs_handler.go | 4 ++-- bulker/ingest/router_ingest_handler.go | 4 ++-- bulker/ingest/router_pixel_handler.go | 4 ++-- 6 files changed, 43 insertions(+), 15 deletions(-) diff --git a/bulker/ingest/router.go b/bulker/ingest/router.go index 8a9d6ec74..6f4bbc2a8 100644 --- a/bulker/ingest/router.go +++ b/bulker/ingest/router.go @@ -302,7 +302,21 @@ func (r *Router) sendToRotor(c *gin.Context, messageId string, ingestMessageByte if stream.Throttle > 0 { if stream.Throttle >= 100 || rand.Int31n(100) < int32(stream.Throttle) { - rError = r.ResponseError(c, http.StatusPaymentRequired, ErrThrottledType, false, fmt.Errorf(ErrThrottledDescription), sendResponse, false, true) + // Quota block (JITSU-88): the event is accepted at ingest and already + // preserved in backup above, but is not delivered to destinations. + // V1 blocking is silent — respond success so the client sees no + // ingestion error. The returned throttle marker still drives the + // SKIPPED events-log status, the `throttled` metric and the + // dead-letter copy in the caller (constructed with sendResponse=false + // so it doesn't write the error body). + rError = r.ResponseError(c, http.StatusOK, ErrThrottledType, false, fmt.Errorf(ErrThrottledDescription), false, false, true) + if sendResponse { + if c.FullPath() == "/api/px/:tp" { + c.Data(http.StatusOK, "image/gif", appbase.EmptyGif) + } else { + c.JSON(http.StatusOK, gin.H{"ok": true}) + } + } return } } diff --git a/bulker/ingest/router_batch_handler.go b/bulker/ingest/router_batch_handler.go index d1b21f90e..ce33f0c34 100644 --- a/bulker/ingest/router_batch_handler.go +++ b/bulker/ingest/router_batch_handler.go @@ -175,7 +175,7 @@ func (r *Router) BatchHandler(c *gin.Context) { defer func() { IngestedMessagesReceived(metricsId, "received").Add(float64(metricsBatchSize)) if rError != nil { - IngestHandlerRequests(domain, "error", rError.ErrorType).Inc() + IngestHandlerRequests(metricsId, "error", rError.ErrorType).Inc() IngestedMessagesReceived(metricsId, "errors").Add(float64(metricsBatchSize)) } }() @@ -311,9 +311,17 @@ func (r *Router) BatchHandler(c *gin.Context) { if rError != nil && rError.ErrorType != ErrNoDst { obj := map[string]any{"body": string(ingestMessageBytes), "error": rError.PublicError.Error(), "status": utils.Ternary(rError.ErrorType == ErrThrottledType, "SKIPPED", "FAILED")} r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelError, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() + IngestHandlerRequests(metricsId, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() _ = r.producer.ProduceAsync(r.config.KafkaDestinationsDeadLetterTopicName, uuid.New(), utils.TruncateBytes(ingestMessageBytes, r.config.MaxIngestPayloadSize), map[string]string{"error": rError.Error.Error()}, kafka2.PartitionAny, messageId, false, 0) - errors = append(errors, fmt.Sprintf("Message ID: %s: %v", messageId, rError.PublicError)) + if rError.ErrorType == ErrThrottledType { + // Quota block (JITSU-88) is a silent success for the client: the + // event is tracked as SKIPPED/throttled above but must not fail + // the batch. Count it toward okEvents so the batch response stays + // ok=true. + okEvents++ + } else { + errors = append(errors, fmt.Sprintf("Message ID: %s: %v", messageId, rError.PublicError)) + } } else { obj := map[string]any{"body": string(ingestMessageBytes), "asyncDestinations": asyncDestinations, "tags": tagsDestinations} if len(asyncDestinations) > 0 || len(tagsDestinations) > 0 { @@ -325,7 +333,7 @@ func (r *Router) BatchHandler(c *gin.Context) { errors = append(errors, fmt.Sprintf("Message ID: %s: %v", messageId, rError.PublicError)) } r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelInfo, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, "success", "").Inc() + IngestHandlerRequests(metricsId, "success", "").Inc() } } processedEvents := len(batch) diff --git a/bulker/ingest/router_classic_handler.go b/bulker/ingest/router_classic_handler.go index dbc76a565..3eb609618 100644 --- a/bulker/ingest/router_classic_handler.go +++ b/bulker/ingest/router_classic_handler.go @@ -86,7 +86,7 @@ func (r *Router) ClassicHandler(c *gin.Context) { defer func() { IngestedMessagesReceived(metricsId, "received").Inc() if rError != nil { - IngestHandlerRequests(domain, "error", rError.ErrorType).Inc() + IngestHandlerRequests(metricsId, "error", rError.ErrorType).Inc() IngestedMessagesReceived(metricsId, "errors").Inc() } }() @@ -186,7 +186,7 @@ func (r *Router) ClassicHandler(c *gin.Context) { if rError != nil && rError.ErrorType != ErrNoDst { obj := map[string]any{"body": string(ingestMessageBytes), "error": rError.PublicError.Error(), "status": utils.Ternary(rError.ErrorType == ErrThrottledType, "SKIPPED", "FAILED")} r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelError, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() + IngestHandlerRequests(metricsId, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() _ = r.producer.ProduceAsync(r.config.KafkaDestinationsDeadLetterTopicName, uuid.New(), utils.TruncateBytes(ingestMessageBytes, r.config.MaxIngestPayloadSize), map[string]string{"error": rError.Error.Error()}, kafka2.PartitionAny, messageId, false, 0) } else { obj := map[string]any{"body": string(ingestMessageBytes), "asyncDestinations": asyncDestinations} @@ -197,10 +197,16 @@ func (r *Router) ClassicHandler(c *gin.Context) { obj["error"] = ErrNoDst } r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelInfo, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, "success", "").Inc() + IngestHandlerRequests(metricsId, "success", "").Inc() } } - c.JSON(http.StatusOK, gin.H{"ok": true}) + // sendToRotor / ResponseError may already have written a response (e.g. the + // silent quota-block success, or a producer error). gin appends body writes + // rather than guarding them, so only write the default success when nothing + // has been written yet. + if !c.Writer.Written() { + c.JSON(http.StatusOK, gin.H{"ok": true}) + } return } diff --git a/bulker/ingest/router_funcs_handler.go b/bulker/ingest/router_funcs_handler.go index 1d4607334..116efa6a3 100644 --- a/bulker/ingest/router_funcs_handler.go +++ b/bulker/ingest/router_funcs_handler.go @@ -40,13 +40,13 @@ func (r *Router) FuncsHandler(c *gin.Context) { IngestedMessagesReceived(metricsId, "errors").Inc() obj := map[string]any{"body": string(ingestMessageBytes), "error": rError.PublicError.Error(), "status": utils.Ternary(rError.ErrorType == ErrThrottledType, "SKIPPED", "FAILED")} r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelError, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() + IngestHandlerRequests(metricsId, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() _ = r.producer.ProduceAsync(r.config.KafkaDestinationsDeadLetterTopicName, uuid.New(), utils.TruncateBytes(ingestMessageBytes, r.config.MaxIngestPayloadSize), map[string]string{"error": rError.Error.Error()}, kafka2.PartitionAny, messageId, false, 0) } else { obj := map[string]any{"body": string(ingestMessageBytes)} obj["status"] = "SUCCESS" r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelInfo, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, "success", "").Inc() + IngestHandlerRequests(metricsId, "success", "").Inc() } }() defer func() { diff --git a/bulker/ingest/router_ingest_handler.go b/bulker/ingest/router_ingest_handler.go index fad646fd9..039380e63 100644 --- a/bulker/ingest/router_ingest_handler.go +++ b/bulker/ingest/router_ingest_handler.go @@ -43,7 +43,7 @@ func (r *Router) IngestHandler(c *gin.Context) { IngestedMessagesReceived(metricsId, "errors").Inc() obj := map[string]any{"body": string(ingestMessageBytes), "error": rError.PublicError.Error(), "status": utils.Ternary(rError.ErrorType == ErrThrottledType, "SKIPPED", "FAILED")} r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelError, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() + IngestHandlerRequests(metricsId, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() _ = r.producer.ProduceAsync(r.config.KafkaDestinationsDeadLetterTopicName, uuid.New(), utils.TruncateBytes(ingestMessageBytes, r.config.MaxIngestPayloadSize), map[string]string{"error": rError.Error.Error()}, kafka2.PartitionAny, messageId, false, 0) } else { obj := map[string]any{"body": string(ingestMessageBytes), "asyncDestinations": asyncDestinations, "tags": tagsDestinations} @@ -54,7 +54,7 @@ func (r *Router) IngestHandler(c *gin.Context) { obj["error"] = ErrNoDst } r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelInfo, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, "success", "").Inc() + IngestHandlerRequests(metricsId, "success", "").Inc() } }() defer func() { diff --git a/bulker/ingest/router_pixel_handler.go b/bulker/ingest/router_pixel_handler.go index 8d6e0808c..982911053 100644 --- a/bulker/ingest/router_pixel_handler.go +++ b/bulker/ingest/router_pixel_handler.go @@ -52,7 +52,7 @@ func (r *Router) PixelHandler(c *gin.Context) { IngestedMessagesReceived(metricsId, "errors").Inc() obj := map[string]any{"body": string(ingestMessageBytes), "error": rError.PublicError.Error(), "status": utils.Ternary(rError.ErrorType == ErrThrottledType, "SKIPPED", "FAILED")} r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelError, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() + IngestHandlerRequests(metricsId, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() _ = r.producer.ProduceAsync(r.config.KafkaDestinationsDeadLetterTopicName, uuid.New(), utils.TruncateBytes(ingestMessageBytes, r.config.MaxIngestPayloadSize), map[string]string{"error": rError.Error.Error()}, kafka2.PartitionAny, messageId, false, 0) } else { obj := map[string]any{"body": string(ingestMessageBytes), "asyncDestinations": asyncDestinations} @@ -63,7 +63,7 @@ func (r *Router) PixelHandler(c *gin.Context) { obj["error"] = ErrNoDst } r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelInfo, ActorId: metricsId, Event: obj}) - IngestHandlerRequests(domain, "success", "").Inc() + IngestHandlerRequests(metricsId, "success", "").Inc() } }() defer func() { From f9d0b4e6062df3682b7b2347bdfd548a5c0b501c Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Mon, 20 Jul 2026 16:58:07 +0400 Subject: [PATCH 2/2] feat(ingest): ERROR_ON_THROTTLE config (default true) for quota-block response MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds ERROR_ON_THROTTLE, default true — preserves the pre-JITSU-88 behavior where a fully-throttled (quota-blocked) event is rejected with HTTP 402. Merging this PR therefore changes no default behavior; the silent block from the previous commit becomes opt-in (ERROR_ON_THROTTLE=false): the event is accepted (HTTP 200) and preserved in backup but not delivered to destinations. The flag gates both the single-event response (sendToRotor) and the batch aggregate (throttled events are errors when erroring, ok when silent). Internal accounting — SKIPPED events-log status, `throttled` metric, dead-letter copy, backup — is identical in both modes. Co-Authored-By: Claude Fable 5 --- bulker/ingest/config.go | 7 +++++++ bulker/ingest/router.go | 17 ++++++++++++----- bulker/ingest/router_batch_handler.go | 5 +++-- 3 files changed, 22 insertions(+), 7 deletions(-) diff --git a/bulker/ingest/config.go b/bulker/ingest/config.go index 079a65b8a..a4283fe11 100644 --- a/bulker/ingest/config.go +++ b/bulker/ingest/config.go @@ -64,6 +64,13 @@ type Config struct { MaxIngestPayloadSize int `mapstructure:"MAX_INGEST_PAYLOAD_SIZE" default:"1000000"` + // When a stream is fully throttled (quota block), reject the event with an + // HTTP 402 (default true — the pre-JITSU-88 behavior, visible to the client + // and its monitoring). Set false for a silent block: the event is accepted + // (HTTP 200) and preserved in backup but not delivered to destinations — + // invisible to the client. + ErrorOnThrottle bool `mapstructure:"ERROR_ON_THROTTLE" default:"true"` + WeightedPartitionSelectorLagThreshold int64 `mapstructure:"WEIGHTED_PARTITION_SELECTOR_LAG_THRESHOLD" default:"0"` // # GRACEFUL SHUTDOWN //Timeout that give running batch tasks time to finish during shutdown. diff --git a/bulker/ingest/router.go b/bulker/ingest/router.go index 6f4bbc2a8..26273a415 100644 --- a/bulker/ingest/router.go +++ b/bulker/ingest/router.go @@ -302,11 +302,18 @@ func (r *Router) sendToRotor(c *gin.Context, messageId string, ingestMessageByte if stream.Throttle > 0 { if stream.Throttle >= 100 || rand.Int31n(100) < int32(stream.Throttle) { - // Quota block (JITSU-88): the event is accepted at ingest and already - // preserved in backup above, but is not delivered to destinations. - // V1 blocking is silent — respond success so the client sees no - // ingestion error. The returned throttle marker still drives the - // SKIPPED events-log status, the `throttled` metric and the + if r.config.ErrorOnThrottle { + // Opt-in: surface the quota block as an HTTP 402 so the client + // sees the rejection (and it shows up in the client's own + // monitoring). ResponseError writes the error body itself. + rError = r.ResponseError(c, http.StatusPaymentRequired, ErrThrottledType, false, fmt.Errorf(ErrThrottledDescription), sendResponse, false, true) + return + } + // Quota block (JITSU-88), default: the event is accepted at ingest and + // already preserved in backup above, but is not delivered to + // destinations. Blocking is silent — respond success so the client + // sees no ingestion error. The returned throttle marker still drives + // the SKIPPED events-log status, the `throttled` metric and the // dead-letter copy in the caller (constructed with sendResponse=false // so it doesn't write the error body). rError = r.ResponseError(c, http.StatusOK, ErrThrottledType, false, fmt.Errorf(ErrThrottledDescription), false, false, true) diff --git a/bulker/ingest/router_batch_handler.go b/bulker/ingest/router_batch_handler.go index ce33f0c34..925deaa3c 100644 --- a/bulker/ingest/router_batch_handler.go +++ b/bulker/ingest/router_batch_handler.go @@ -313,11 +313,12 @@ func (r *Router) BatchHandler(c *gin.Context) { r.eventsLogService.PostAsync(&eventslog.ActorEvent{EventType: eventslog.EventTypeIncoming, Level: eventslog.LevelError, ActorId: metricsId, Event: obj}) IngestHandlerRequests(metricsId, utils.Ternary(rError.ErrorType == ErrThrottledType, "throttled", "error"), rError.ErrorType).Inc() _ = r.producer.ProduceAsync(r.config.KafkaDestinationsDeadLetterTopicName, uuid.New(), utils.TruncateBytes(ingestMessageBytes, r.config.MaxIngestPayloadSize), map[string]string{"error": rError.Error.Error()}, kafka2.PartitionAny, messageId, false, 0) - if rError.ErrorType == ErrThrottledType { + if rError.ErrorType == ErrThrottledType && !r.config.ErrorOnThrottle { // Quota block (JITSU-88) is a silent success for the client: the // event is tracked as SKIPPED/throttled above but must not fail // the batch. Count it toward okEvents so the batch response stays - // ok=true. + // ok=true. With ERROR_ON_THROTTLE the block is a client-visible + // error, so it falls through to the errors list below. okEvents++ } else { errors = append(errors, fmt.Sprintf("Message ID: %s: %v", messageId, rError.PublicError))