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 8a9d6ec74..26273a415 100644
--- a/bulker/ingest/router.go
+++ b/bulker/ingest/router.go
@@ -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})
+				}
+			}
 			return
 		}
 	}
diff --git a/bulker/ingest/router_batch_handler.go b/bulker/ingest/router_batch_handler.go
index d1b21f90e..925deaa3c 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,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 {
@@ -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)
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() {