From f9fb383e7b09a6ee4791196a9f860126c6a82c81 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 02:17:15 +0100 Subject: [PATCH 1/4] fix(ingest): keep WAL backlog and org queue counters exact, stop billing refused batches - WAL backlog no longer drifts up. A sync counted every frame in its batch as committed, including frames whose caller had been dropped (request timeout, client disconnect). Those are never delivered to export, so they were never marked exported and the backlog never returned to 0: every shutdown waited the full drain window and scale-in protection could stick. A skipped delivery now takes its bytes back out of the lane's committed count. - A failed cursor write no longer leaks org queue bytes. The batch leaves memory either way, so the org's bytes are released before the error is returned. The comment that claimed the batch "will be replayed" is corrected. - A collector 4xx relayed in forward mode is no longer metered or billed, on both the HTTP and gRPC paths (it already did not count as key use on HTTP). - read_wal_frame no longer allocates the header's payload length (up to 4 GiB) before the checksum: the buffer grows with the bytes actually read, and a short read is a corrupt frame. - INGEST_QUEUE_CHANNEL_CAPACITY=0 is refused at validation instead of panicking in mpsc::channel(0) at boot. - Remove INGEST_TINYBIRD_CONCURRENCY_PER_SHARD: it was parsed, validated and forwarded but nothing read it (there is one export worker per lane). - Remove accept_rows, which had no callers. - Comments: the WAL lives on the EC2 instance store, not Fargate ephemeral storage, and the prebuilt image compiles in rust:1.94-bookworm. --- apps/ingest/Dockerfile.prebuilt | 2 +- apps/ingest/alchemy.run.ts | 1 - apps/ingest/benches/ingest_bench.rs | 1 - apps/ingest/src/main.rs | 22 ++-- apps/ingest/src/task_protection.rs | 13 +- apps/ingest/src/telemetry.rs | 191 ++++++++++++++++++++++------ apps/ingest/src/wal_store.rs | 7 +- 7 files changed, 178 insertions(+), 59 deletions(-) diff --git a/apps/ingest/Dockerfile.prebuilt b/apps/ingest/Dockerfile.prebuilt index ac45b59c2d..e6392ea8e4 100644 --- a/apps/ingest/Dockerfile.prebuilt +++ b/apps/ingest/Dockerfile.prebuilt @@ -19,7 +19,7 @@ # The compile MUST happen against bookworm's glibc (2.36) — building on the # ubuntu-24.04 runner (2.39) instead produces a binary that dies at startup with # `version 'GLIBC_2.39' not found`. The workflow enforces this by compiling -# inside rust:1.88-bookworm rather than on the host. +# inside rust:1.94-bookworm rather than on the host. # # `Dockerfile` (the self-contained source build) stays the default and is what a # local `alchemy deploy` uses; alchemy.run.ts picks this one only when diff --git a/apps/ingest/alchemy.run.ts b/apps/ingest/alchemy.run.ts index b7e8b91245..57db62b136 100644 --- a/apps/ingest/alchemy.run.ts +++ b/apps/ingest/alchemy.run.ts @@ -699,7 +699,6 @@ export const createMapleIngest = ({ stage, region, domains, profile, dbLogin }: // keys again after normalizing the header, so a respelled one is still blocked. ...(yield* Config.map(blockedIngestKeys, blockedIngestKeysEnv)), ...(yield* optionalPlain("INGEST_EXPORT_MAX_ATTEMPTS")), - ...(yield* optionalPlain("INGEST_TINYBIRD_CONCURRENCY_PER_SHARD")), ...(yield* optionalPlain("INGEST_REPLAY_MAX_SESSION_BYTES")), // The org Maple's own telemetry is filed under. Required, no fallback. MAPLE_INTERNAL_ORG_ID: yield* requiredPlain("MAPLE_INTERNAL_ORG_ID"), diff --git a/apps/ingest/benches/ingest_bench.rs b/apps/ingest/benches/ingest_bench.rs index b6b665705e..5ec9323b40 100644 --- a/apps/ingest/benches/ingest_bench.rs +++ b/apps/ingest/benches/ingest_bench.rs @@ -122,7 +122,6 @@ impl BenchFixture { batch_max_bytes: 4 * 1024 * 1024, queue_max_age: Duration::ZERO, batch_max_wait: Duration::from_millis(10), - export_concurrency_per_shard: 1, export_max_attempts: 20, clickhouse_export_timeout: Duration::from_secs(5), clickhouse_breaker: ClickHouseBreakerConfig::default(), diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index 0ab806f4ae..85a4b8cda1 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -411,11 +411,6 @@ impl AppConfig { std::env::var("INGEST_BATCH_MAX_WAIT_MS").ok(), 100, )?), - export_concurrency_per_shard: parse_usize( - "INGEST_TINYBIRD_CONCURRENCY_PER_SHARD", - std::env::var("INGEST_TINYBIRD_CONCURRENCY_PER_SHARD").ok(), - 1, - )?, export_max_attempts: parse_u32( "INGEST_EXPORT_MAX_ATTEMPTS", std::env::var("INGEST_EXPORT_MAX_ATTEMPTS").ok(), @@ -3295,8 +3290,9 @@ async fn main() { // Even a clean drain runs this: it marks the owner retired, so a // successor claims anything left instead of waiting out the staleness // window. With a backlog it also seals and ships the tail, which is the - // difference between "replays if this task's storage survives" (it does - // not — Fargate ephemeral storage dies with the task) and "replays". + // difference between "replays if this task's storage survives" (the + // instance store under `/mnt/wal` is wiped when the host is stopped or + // replaced) and "replays". let shipped = pipeline.flush_wal_to_object_store().await; if shipped > 0 { info!( @@ -3868,6 +3864,8 @@ async fn accept_grpc_decoded( .await; match result { + // A collector refusal stored nothing, so it is neither key use nor billed. + Ok(response) if !response.status().is_success() => Ok(()), Ok(_) => { state.resolver.note_source_use(resolved, claims.kept_claim); state @@ -5664,7 +5662,9 @@ async fn handle_signal_inner( let kind = error.error_kind(); (error, if kind == INGEST_RULES_SCAN_BUDGET.error_kind { kind } else { "forward" }) })?; - if response.status().is_success() { + // A relayed collector refusal stored nothing: not key use, and 0 bytes to meter. + let stored = response.status().is_success(); + if stored { state.resolver.note_source_use(&resolved_key, kept_claim); state .resolver @@ -5675,7 +5675,7 @@ async fn handle_signal_inner( response, item_count, resolved_key.org_id.clone(), - decoded_bytes, + if stored { decoded_bytes } else { 0 }, )) } @@ -11714,7 +11714,6 @@ mod tests { batch_max_bytes: 1024 * 1024, queue_max_age: Duration::ZERO, batch_max_wait: Duration::from_millis(1), - export_concurrency_per_shard: 1, export_max_attempts: 1, clickhouse_export_timeout: Duration::from_secs(5), clickhouse_breaker: ClickHouseBreakerConfig::default(), @@ -16190,11 +16189,12 @@ mod tests { .await; let body = Bytes::from(test_log_request("refused").encode_to_vec()); - let (response, ..) = + let (response, _items, _org_id, metered_bytes) = handle_signal_inner(&state, &test_headers(raw_key), body, Signal::Logs) .await .expect("a collector 4xx is relayed, not raised"); assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(metered_bytes, 0, "a refused batch is not billed"); assert!(recorded_key_uses(&store).await.is_empty()); drop(std::fs::remove_dir_all(&queue_dir)); diff --git a/apps/ingest/src/task_protection.rs b/apps/ingest/src/task_protection.rs index e5e347d530..28932bfc89 100644 --- a/apps/ingest/src/task_protection.rs +++ b/apps/ingest/src/task_protection.rs @@ -1,11 +1,12 @@ //! ECS task scale-in protection driven by WAL backlog. //! -//! The WAL lives on Fargate ephemeral storage, which is destroyed with the -//! task — so a scale-in that lands on a task holding unexported frames loses -//! them. Backlog is largest exactly when the autoscaler wants to scale in -//! (after a burst or a warehouse outage), so this loop marks the task -//! protected via the ECS agent's task-protection endpoint while the primary -//! lanes hold meaningful backlog, and releases it once they drain. +//! The WAL lives on the EC2 host's instance store (`/mnt/wal`), which is wiped +//! when the instance stops or is replaced, so a scale-in that lands on a task +//! holding unexported frames puts them at risk. Backlog is largest exactly +//! when the autoscaler wants to scale in (after a burst or a warehouse +//! outage), so this loop marks the task protected via the ECS agent's +//! task-protection endpoint while the primary lanes hold meaningful backlog, +//! and releases it once they drain. //! //! The endpoint is signed by the agent itself; the task role only needs //! `ecs:UpdateTaskProtection` (granted in `apps/ingest/alchemy.run.ts`). diff --git a/apps/ingest/src/telemetry.rs b/apps/ingest/src/telemetry.rs index 654475e21e..1bc45c2a80 100644 --- a/apps/ingest/src/telemetry.rs +++ b/apps/ingest/src/telemetry.rs @@ -58,6 +58,9 @@ const WAL_MAGIC: &[u8; 4] = b"MTW1"; const WAL_V1_HEADER_LEN: usize = 20; const WAL_V2_HEADER_LEN: usize = 22; const WAL_V3_HEADER_LEN: usize = 23; +/// Most a frame read reserves before any payload byte arrives. The declared +/// length is unverified until the checksum, so a corrupt one must not size it. +const WAL_PAYLOAD_PREALLOC_MAX: usize = 1024 * 1024; #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum ExportDestination { @@ -628,7 +631,6 @@ pub struct TinybirdConfig { pub batch_max_rows: usize, pub batch_max_bytes: usize, pub batch_max_wait: Duration, - pub export_concurrency_per_shard: usize, pub export_max_attempts: u32, /// Per-attempt timeout for a direct ClickHouse export POST. Kept short and /// separate from `forward_timeout` so a hanging BYO-ClickHouse target fails @@ -698,8 +700,9 @@ impl TinybirdConfig { if self.org_queue_max_bytes == 0 { return Err("INGEST_ORG_QUEUE_MAX_BYTES must be greater than 0".to_owned()); } - if self.export_concurrency_per_shard == 0 { - return Err("INGEST_TINYBIRD_CONCURRENCY_PER_SHARD must be greater than 0".to_owned()); + // `mpsc::channel(0)` panics, so a zero capacity is refused at boot instead. + if self.queue_channel_capacity == 0 { + return Err("INGEST_QUEUE_CHANNEL_CAPACITY must be greater than 0".to_owned()); } if self.export_max_attempts == 0 { return Err("INGEST_EXPORT_MAX_ATTEMPTS must be greater than 0".to_owned()); @@ -1166,23 +1169,6 @@ impl TelemetryPipeline { /// the session-replay ingest path, whose rows are built directly by the /// gateway (not derived from OTLP). Routes by org so a session's metadata /// and chunk-index rows land on the same shard in order. - pub async fn accept_rows( - &self, - org_id: &str, - datasource: String, - rows: Vec>, - signal: TelemetrySignal, - ) -> Result { - self.accept_rows_to( - org_id, - datasource, - rows, - signal, - ExportDestination::Tinybird, - ) - .await - } - #[hotpath::measure] pub async fn accept_rows_to( &self, @@ -1456,9 +1442,9 @@ struct WalLane { /// Bytes held by every segment on disk, the exported prefix included. Drives /// the lane cap and the `wal_shard_bytes` gauge. live_bytes: AtomicU64, - /// Cumulative bytes appended and marked exported since this lane was opened. - /// Their difference is the backlog, and it stays exact across rotation and - /// deletion — which nothing derived from file sizes does. + /// Cumulative bytes handed to export and marked exported since this lane + /// was opened. Their difference is the backlog, and it stays exact across + /// rotation and deletion, which nothing derived from file sizes does. committed_bytes: AtomicU64, exported_bytes: AtomicU64, append: Mutex, @@ -1492,13 +1478,16 @@ impl Drop for WaitingCaller { } /// `deliver`, run only while the returned guard is alive. Skipped, the inner -/// delivery is dropped, which releases its org reservation. -fn while_caller_waits(deliver: Delivery) -> (WaitingCaller, Delivery) { +/// delivery is dropped, which releases its org reservation, and the frame's +/// bytes leave `lane`'s backlog: nothing in this process will export them. +fn while_caller_waits(deliver: Delivery, lane: Arc) -> (WaitingCaller, Delivery) { let waiting = WaitingCaller(Arc::new(AtomicBool::new(true))); let still_waiting = Arc::clone(&waiting.0); let guarded: Delivery = Box::new(move |segment, start, end| { if still_waiting.load(Ordering::Acquire) { deliver(segment, start, end); + } else { + lane.forget_committed(end.saturating_sub(start)); } }); (waiting, guarded) @@ -2041,6 +2030,12 @@ impl WalLane { .load(Ordering::Relaxed) .saturating_sub(self.exported_bytes.load(Ordering::Relaxed)) } + + /// Take a synced frame back out of the backlog. Its sync already counted + /// it, but it never reaches export, so it would never be marked exported. + fn forget_committed(&self, bytes: u64) { + self.committed_bytes.fetch_sub(bytes, Ordering::Relaxed); + } } /// What WAL orphan recovery did with one segment's source object. @@ -2573,7 +2568,7 @@ impl ShardedWal { let lane_ref = self.lane(lane)?; // A caller dropped mid-append (a request timeout) has already answered // with a retryable error, so its frame must not reach export as well. - let (waiting, deliver) = while_caller_waits(deliver); + let (waiting, deliver) = while_caller_waits(deliver, Arc::clone(&lane_ref)); let appended = tokio::task::spawn_blocking(move || { lane_ref.append_blocking(&encoded, enforce_cap, deliver) }) @@ -3022,9 +3017,17 @@ fn read_wal_frame(file: &mut impl Read, start: u64) -> Result Delivery { let ran = Arc::clone(&ran); @@ -7944,13 +8034,13 @@ mod tests { }) }; - let (gone, deliver) = while_caller_waits(counting()); + let (gone, deliver) = while_caller_waits(counting(), Arc::clone(&lane)); drop(gone); lane.append_blocking(&encoded_test_frame(64), true, deliver) .expect("the frame is still made durable"); assert_eq!(ran.load(Ordering::SeqCst), 0, "but never reaches export"); - let (waiting, deliver) = while_caller_waits(counting()); + let (waiting, deliver) = while_caller_waits(counting(), Arc::clone(&lane)); lane.append_blocking(&encoded_test_frame(64), true, deliver) .expect("append"); drop(waiting); @@ -7962,6 +8052,35 @@ mod tests { drop(std::fs::remove_dir_all(queue_dir)); } + #[test] + fn a_frame_whose_caller_gave_up_is_not_backlog() { + let (lane, _probe, queue_dir) = slow_sync_lane("wal-caller-gone-backlog", Duration::ZERO); + let lane = Arc::new(lane); + let nothing = || -> Delivery { Box::new(|_, _, _| {}) }; + + let (gone, deliver) = while_caller_waits(nothing(), Arc::clone(&lane)); + drop(gone); + lane.append_blocking(&encoded_test_frame(64), true, deliver) + .expect("append"); + assert_eq!( + lane.backlog_bytes(), + 0, + "nothing will export a skipped frame, so it must not hold the drain open" + ); + + let delivered = encoded_test_frame(64); + let (waiting, deliver) = while_caller_waits(nothing(), Arc::clone(&lane)); + lane.append_blocking(&delivered, true, deliver) + .expect("append"); + drop(waiting); + assert_eq!( + lane.backlog_bytes(), + delivered.len() as u64, + "a delivered frame is backlog until it exports" + ); + drop(std::fs::remove_dir_all(queue_dir)); + } + #[test] fn a_failed_sync_refuses_its_batch_and_the_lane_recovers() { let (lane, probe, queue_dir) = slow_sync_lane("wal-group-commit-fail", Duration::ZERO); diff --git a/apps/ingest/src/wal_store.rs b/apps/ingest/src/wal_store.rs index f379737caf..e17de5f9f0 100644 --- a/apps/ingest/src/wal_store.rs +++ b/apps/ingest/src/wal_store.rs @@ -1,9 +1,10 @@ //! The WAL's durability tier: sealed segments shipped to S3, and the claiming //! protocol that lets a booting task recover a dead one's. //! -//! The local WAL lives on Fargate ephemeral storage, which is destroyed with the -//! task. A drain on SIGTERM covers a graceful stop; this covers everything else -//! — a crash, a spot reclaim, an AZ event — by keeping every sealed, unexported +//! The local WAL lives on the EC2 host's instance store (`/mnt/wal`), which +//! survives a reboot but is wiped when the instance stops or is replaced. A +//! drain on SIGTERM covers a graceful stop; this covers everything else (a +//! crash, a spot reclaim, an AZ event) by keeping every sealed, unexported //! segment in an object store too. //! //! Cost is small because segments are transient: a segment is deleted from the From e89817da6838ea2e354f86f55aa12c48ee9b4008 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 02:17:15 +0100 Subject: [PATCH 2/4] fix(otel-exporter): encode unknown severity and out-of-range timestamps like the gateway - An unknown severity number is now the empty string, not its decimal form, matching the gateway's and Maple Local's encoders. - A timestamp above the int64 maximum is written as the epoch. It used to wrap negative, and a value outside DateTime64(9) fails the whole insert. - Drop the leftover unused-import guard in exporter_traces.go. - README: org_id is stamped on every record; a resource attribute overrides it only when org_id_from_resource_attribute is set, as config.go documents. --- packages/otel-collector-maple-exporter/README.md | 6 +++--- .../exporter_traces.go | 4 ---- .../internal/encoding.go | 14 ++++++++------ .../internal/encoding_test.go | 10 ++++++++-- 4 files changed, 19 insertions(+), 15 deletions(-) diff --git a/packages/otel-collector-maple-exporter/README.md b/packages/otel-collector-maple-exporter/README.md index 6fb44f118e..e327a28fee 100644 --- a/packages/otel-collector-maple-exporter/README.md +++ b/packages/otel-collector-maple-exporter/README.md @@ -26,9 +26,9 @@ exporters: database: default username: maple password: ${env:MAPLE_CLICKHOUSE_PASSWORD} - org_id: org_3AuiNCIuD1XCbbzcjkzE3s5HoQj # static fallback; per-record - # `maple_org_id` resource attr - # wins over this + org_id: org_3AuiNCIuD1XCbbzcjkzE3s5HoQj # stamped on every record; a + # resource attribute overrides it only when + # `org_id_from_resource_attribute` is set timeout: 30s retry_on_failure: enabled: true diff --git a/packages/otel-collector-maple-exporter/exporter_traces.go b/packages/otel-collector-maple-exporter/exporter_traces.go index 538c4bf2a5..f89a7c7405 100644 --- a/packages/otel-collector-maple-exporter/exporter_traces.go +++ b/packages/otel-collector-maple-exporter/exporter_traces.go @@ -4,7 +4,6 @@ import ( "context" "go.opentelemetry.io/collector/component" - "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/ptrace" "go.uber.org/zap" @@ -181,6 +180,3 @@ func encodeTraceRow( "LinksAttributes": linksAttributes, } } - -// silence unused import warning during incremental build -var _ pcommon.Map diff --git a/packages/otel-collector-maple-exporter/internal/encoding.go b/packages/otel-collector-maple-exporter/internal/encoding.go index 9c9ac6f877..cfa7dde4a0 100644 --- a/packages/otel-collector-maple-exporter/internal/encoding.go +++ b/packages/otel-collector-maple-exporter/internal/encoding.go @@ -3,7 +3,7 @@ package internal import ( "encoding/json" "fmt" - "strconv" + "math" "time" "go.opentelemetry.io/collector/pdata/pcommon" @@ -72,7 +72,9 @@ func ServiceName(resource pcommon.Map) string { // the layout — but ClickHouse parses the no-zone form and stores at the // configured precision, so we omit the zone to match its native rendering. func FormatTimestampNano(unixNano uint64) string { - if unixNano == 0 { + // DateTime64(9) is an Int64 of nanoseconds: a value past MaxInt64 would wrap + // negative below and fail the whole insert, so it maps to the epoch like 0. + if unixNano == 0 || unixNano > math.MaxInt64 { return "1970-01-01 00:00:00.000000000" } t := time.Unix(0, int64(unixNano)).UTC() @@ -82,7 +84,7 @@ func FormatTimestampNano(unixNano uint64) string { // FormatDateTime formats a UnixNano integer as the second-precision DateTime // form used by Maple's `Timestamp DateTime` and `Hour DateTime` columns. func FormatDateTime(unixNano uint64) string { - if unixNano == 0 { + if unixNano == 0 || unixNano > math.MaxInt64 { return "1970-01-01 00:00:00" } return time.Unix(0, int64(unixNano)).UTC().Format("2006-01-02 15:04:05") @@ -123,8 +125,8 @@ func StatusCodeString(code int32) string { } // SeverityNumberToText returns the canonical OTel severity text for a -// numeric severity (1-24). Falls back to the numeric form for unknown -// values. +// numeric severity (1-24). Unknown values are the empty string, as in the +// gateway's and Maple Local's encoders, so all three write the same rows. func SeverityNumberToText(n int32) string { switch { case n >= 1 && n <= 4: @@ -140,7 +142,7 @@ func SeverityNumberToText(n int32) string { case n >= 21 && n <= 24: return "FATAL" default: - return strconv.Itoa(int(n)) + return "" } } diff --git a/packages/otel-collector-maple-exporter/internal/encoding_test.go b/packages/otel-collector-maple-exporter/internal/encoding_test.go index af0468f1e0..7e1a353cd6 100644 --- a/packages/otel-collector-maple-exporter/internal/encoding_test.go +++ b/packages/otel-collector-maple-exporter/internal/encoding_test.go @@ -1,6 +1,7 @@ package internal import ( + "math" "testing" "go.opentelemetry.io/collector/pdata/pcommon" @@ -16,6 +17,10 @@ func TestFormatTimestampNano(t *testing.T) { {"epoch_one_ns", 1, "1970-01-01 00:00:00.000000001"}, {"epoch_one_sec", 1_000_000_000, "1970-01-01 00:00:01.000000000"}, {"specific_moment", 1_704_067_200_123_456_789, "2024-01-01 00:00:00.123456789"}, + {"int64_max", math.MaxInt64, "2262-04-11 23:47:16.854775807"}, + // Past Int64 the value would wrap negative and fail the whole insert. + {"past_int64_max", math.MaxInt64 + 1, "1970-01-01 00:00:00.000000000"}, + {"uint64_max", math.MaxUint64, "1970-01-01 00:00:00.000000000"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -39,6 +44,7 @@ func TestFormatDateTime(t *testing.T) { {"zero", 0, "1970-01-01 00:00:00"}, {"end_of_year", 1_704_067_199_999_999_999, "2023-12-31 23:59:59"}, {"start_of_year", 1_704_067_200_000_000_000, "2024-01-01 00:00:00"}, + {"past_int64_max", math.MaxInt64 + 1, "1970-01-01 00:00:00"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -108,8 +114,8 @@ func TestSeverityNumberToText(t *testing.T) { {20, "ERROR"}, {21, "FATAL"}, {24, "FATAL"}, - {0, "0"}, // out-of-band → numeric form - {25, "25"}, + {0, ""}, // out-of-band is empty, as in the gateway's encoder + {25, ""}, } for _, tc := range cases { got := SeverityNumberToText(tc.n) From be774bd15a9ffbfed3925ecc0f4ec7b1e3eac6a3 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 04:30:33 +0100 Subject: [PATCH 3/4] fix(ingest): meter a refused relay when the native pipeline already stored the batch In dual mode the batch is accepted into the WAL before it is forwarded, so a collector refusal does not mean nothing was stored. Only a refusal that left nothing behind skips metering and key use, on the HTTP and the gRPC path. --- apps/ingest/src/main.rs | 50 +++++++++++++++++++++++++++++++++++++---- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index 85a4b8cda1..c2b1874d67 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -276,6 +276,16 @@ fn uses_forward_path_for(write_mode: WriteMode, destination: ExportDestination) write_mode.uses_forward() && destination != ExportDestination::ClickHouse } +/// Whether a request left rows behind: the collector took it, or the native +/// pipeline had already accepted it before the forward (dual mode). +fn batch_was_stored( + forwarded_ok: bool, + write_mode: WriteMode, + destination: ExportDestination, +) -> bool { + forwarded_ok || uses_native_pipeline_for(write_mode, destination) +} + #[derive(Clone)] enum KeyStoreBackend { // No-DB local backend: every well-formed ingest key resolves to a single @@ -3864,8 +3874,16 @@ async fn accept_grpc_decoded( .await; match result { - // A collector refusal stored nothing, so it is neither key use nor billed. - Ok(response) if !response.status().is_success() => Ok(()), + // A collector refusal that left nothing stored is neither key use nor billed. + Ok(response) + if !batch_was_stored( + response.status().is_success(), + state.config.write_mode, + native_destination_for(resolved), + ) => + { + Ok(()) + } Ok(_) => { state.resolver.note_source_use(resolved, claims.kept_claim); state @@ -5662,8 +5680,12 @@ async fn handle_signal_inner( let kind = error.error_kind(); (error, if kind == INGEST_RULES_SCAN_BUDGET.error_kind { kind } else { "forward" }) })?; - // A relayed collector refusal stored nothing: not key use, and 0 bytes to meter. - let stored = response.status().is_success(); + // A collector refusal that left nothing stored is not key use, and 0 bytes to meter. + let stored = batch_was_stored( + response.status().is_success(), + state.config.write_mode, + native_destination_for(&resolved_key), + ); if stored { state.resolver.note_source_use(&resolved_key, kept_claim); state @@ -11217,6 +11239,26 @@ mod tests { )); } + #[test] + fn a_refused_relay_still_counts_when_the_native_pipeline_took_the_batch() { + let refused = false; + assert!(!batch_was_stored( + refused, + WriteMode::Forward, + ExportDestination::Tinybird + )); + assert!(batch_was_stored( + refused, + WriteMode::Dual, + ExportDestination::Tinybird + )); + assert!(batch_was_stored( + true, + WriteMode::Forward, + ExportDestination::Tinybird + )); + } + /// In-memory KeyStore used to exercise the resolver's behavior (caching, /// key-type inference, ResolvedIngestKey construction) without a database. /// Keyed on the same `(hash, column)` shape the real key store sees. From c12917f6295524ac766875918fd860351af97e6b Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 05:02:14 +0100 Subject: [PATCH 4/4] test(ingest): a refused relay over gRPC is answered OK and is not key use Only the HTTP path of the metering skip was pinned. --- apps/ingest/src/main.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index c2b1874d67..94765134ee 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -16239,6 +16239,28 @@ mod tests { assert_eq!(metered_bytes, 0, "a refused batch is not billed"); assert!(recorded_key_uses(&store).await.is_empty()); + // Over gRPC the refusal is answered OK, and is not key use either. + let resolved = state + .resolver + .resolve_ingest_key(raw_key) + .await + .expect("resolution should not fail") + .expect("the test key resolves"); + let payload = test_log_request("refused over grpc"); + let decoded_bytes = payload.encoded_len(); + accept_grpc_decoded( + &state, + Signal::Logs, + DecodedPayload::Logs(payload), + &resolved, + KeyAdmission::Ingest, + decoded_bytes, + &ClaimOutcome::default(), + ) + .await + .expect("a collector refusal over gRPC is answered OK"); + assert!(recorded_key_uses(&store).await.is_empty()); + drop(std::fs::remove_dir_all(&queue_dir)); }