Skip to content

JITSU-88 fix(ingest): silent success on quota block + stream-id metric label #1413

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
absorbb wants to merge 2 commits into
base: newjitsu
Choose a base branch
from
Open

JITSU-88 fix(ingest): silent success on quota block + stream-id metric label #1413

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
388bb12
fix(ingest): silent success on quota block + stream-id metric label (…
absorbb Jul 20, 2026
f9d0b4e
feat(ingest): ERROR_ON_THROTTLE config (default true) for quota-block…
absorbb Jul 20, 2026
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
7 changes: 7 additions & 0 deletions bulker/ingest/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
23 changes: 22 additions & 1 deletion bulker/ingest/router.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,28 @@ 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)
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)
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})
Comment thread
absorbb marked this conversation as resolved.
Comment thread
absorbb marked this conversation as resolved.
}
}
return
}
}
Expand Down
17 changes: 13 additions & 4 deletions bulker/ingest/router_batch_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
}
}()
Expand Down Expand Up @@ -311,9 +311,18 @@ 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 && !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. 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))
}
} else {
obj := map[string]any{"body": string(ingestMessageBytes), "asyncDestinations": asyncDestinations, "tags": tagsDestinations}
if len(asyncDestinations) > 0 || len(tagsDestinations) > 0 {
Expand All @@ -325,7 +334,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)
Expand Down
14 changes: 10 additions & 4 deletions bulker/ingest/router_classic_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
}
}()
Expand Down Expand Up @@ -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}
Expand All @@ -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
}

Expand Down
4 changes: 2 additions & 2 deletions bulker/ingest/router_funcs_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
4 changes: 2 additions & 2 deletions bulker/ingest/router_ingest_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand All @@ -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() {
Expand Down
4 changes: 2 additions & 2 deletions bulker/ingest/router_pixel_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand All @@ -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() {
Expand Down
Loading