feat(ingest): enforce source tokens and stamp maple_source_id

[Makisuo]

Part 3 of 4 of Sources. Base: the integrations layer (the gateway reads sources.integration_id, so it must deploy after that migration).

The ingest gateway enforces source tokens and stamps every resource with its source.

• maple_src_ tokens resolve against sources by HMAC, cached like other keys. Expired: 401. Signal not allowed: 403 (gRPC PermissionDenied). Paused: 200 with the data dropped and not billed (gRPC skips decoding). Over the rate limit: 429 with Retry-After. Resource attribute overrides can never touch org_id or maple_*.
• Stamping: sender maple_source_id is stripped, then token traffic gets its token's source, and default-key traffic gets the org's default source (read in the key query itself, so a sources outage never fails auth). A default-key sender may keep a claimed id only if it is one of the org's integration sources; claims are capped at 8 per request and checked in one query. Google Cloud log pushes carry their connector's source.
• last_used_at is written at most once a minute per token source and every 5 minutes for the default source, and not for batches that carried a kept claim.

Verification

• cargo test --locked: 246 lib + 137 binary tests, 31 of them for sources.
• Local end to end against the dev stack: a source token is accepted (200) and a bad one rejected (401); with signals limited to traces, logs get 403 and traces 200; paused, both get 200 and last_used_at does not move; the UI flipped from "Waiting for first data" to "Receiving data" on the first request.

Stack: #1535 model → #1536 integrations → #1537 gateway → #1538 UI

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features
  • Source tokens can now be configured with signal access, rate limits, resource attributes, expiration, and pause periods.
  • Integration source IDs are included in logs when available.
• Improvements
  • Ingest rate limits now account for configured fleet scaling.
  • Ingest accepts additional source-related keys and records events dropped while a source is paused.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

• Configuration used: defaults
• Review profile: CHILL
• Plan: Advanced
• Run ID: e89158c2-d119-4b37-b42f-ad16227ecd9c

📥 Commits

Reviewing files that changed from the base of the PR and between 3fa8381 and 4ac57b0.

📒 Files selected for processing: 6

• apps/ingest/alchemy.run.ts
• apps/ingest/src/gcp_logging.rs
• apps/ingest/src/main.rs
• apps/ingest/src/metrics.rs
• apps/ingest/src/otel.rs
• apps/ingest/src/sources.rs

Included review availability: This review used your included allowance. Your plan provides up to 4 included reviews per hour; 3 remain after this review.

📝 Walkthrough

The ingest code adds source-policy parsing, public source IDs, and per-key rate limiting. The ingest task sets the fleet-size value used for rate limits. Logging, spans, and metrics gain source-related fields and measurements.

Changes: Ingest source controls and telemetry

Layer / File(s)	Summary
Source policies and public IDs apps/ingest/src/sources.rs	Source configuration rows are parsed into policies with signal allow-lists, resource overrides, expiry, and pause checks. The module adds versioned public source ID encoding and validation, with tests for policy parsing and ID conversion.
Per-key rate limiting and fleet configuration apps/ingest/src/sources.rs, apps/ingest/alchemy.run.ts	A token-bucket limiter divides fleet RPS across instances and expires idle buckets. The ingest task sets the fleet-size environment value from the scaling minimum or desired count. Blocked-key validation accepts maple_src_ keys.
Source identifiers and paused-drop telemetry apps/ingest/src/gcp_logging.rs, apps/ingest/src/otel.rs, apps/ingest/src/metrics.rs	Connector resource attributes include maple_source_id when an ID is present. Two span definitions declare maple.source.id. A counter records paused-source drops with organization and signal labels.

Priority: ➖ Normal

Estimated code review effort: 3 (Moderate) | ~25 minutes

Change: Feature

Suggested reviewers: jeremyfunk

Merge Risk: ⚪ Minimal · up to 4ac57

No actionable issue remains from the inspected source-control and stamping paths. Merge after normal checks.

Pre-merge checks: 4 · 1

Failed checks: 1 warning

Check name	Status	Explanation	Resolution
Docstring Coverage		Docstring coverage is 61.90% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 42 functions across 5 files.	Write docstrings for the functions missing them to satisfy the coverage threshold.

Passed checks: 4 passed

Check name	Status	Explanation
Description Check		Check skipped - CodeRabbit’s high-level summary is enabled.
Title check		The title clearly and concisely describes the main changes: enforcing source tokens and recording the source identifier on ingest data.
Linked Issues check		Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check		Check skipped because no linked issues were found for this pull request.

• Fix all pre-merge checks with AI

✨ Finishing Touches: 💡 1

📝 Generate docstrings: 💡

• Commit to this branch
• Create a new PR

🧪 Generate unit tests (beta)

• Commit to this branch
• Create a new PR

• Autofix · Keep fixing CodeRabbit findings and required CI, and resolving merge conflicts

Warning

Some tools did not complete. Review the errors below.

🔧 Clippy (1.98.1)

Clippy execution failed

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[maple-review-bot]

Maple review

🔴 Confidence 4/10 · risky as written
The source gate is auth/tenancy code with 31 unit tests, but the sender-supplied claim decode is unbounded and can hammer the fleet.
_{quality 75/100 · 1 critical · tests covered · risk high}

Adds maple_src_ token enforcement and maple_source_id stamping to the ingest gateway. The gating and stamping logic is sound and well tested, but the sender-supplied claim path decodes an unbounded attribute value.

• maple_src_ tokens resolve against sources and pass expiry, signal, pause and rate gates
• A paused source gets 200 with the payload dropped and unbilled
• Every resource is stamped with its token's, claimed, or default source
• last_used_at is written at most once a minute per source, off the request path

Before merge

• Infra · Check every environment has what this config now expects · apps/ingest/alchemy.run.ts
• Manual · Deploy after the sources migration lands: the default-key lookup now joins sources

Findings

🔴 Critical · F1 · source_uuid_from_public_id base58-decodes unbounded sender input

security · apps/ingest/src/sources.rs:224-225

A sender-supplied maple_source_id resource attribute is fed to base58_decode before its length is checked, and that decode is quadratic in its input: a 20 MB attribute (the body ceiling at apps/ingest/src/main.rs:4897) is ~10^14 byte operations, pinning an ingest task's CPU for hours and, with a few concurrent requests, wedging the shared fleet. Any default key that can reach the OTLP routes can send it — the claim path is reached from verify_source_claims (main.rs:6621) — and default-key traffic is not covered by the new source rate limit. Reject any body longer than one a valid src_ id can have (17 bytes encode to at most 24 base58 characters) before decoding.

    let encoded = public_id.strip_prefix(PUBLIC_ID_PREFIX)?;
    // 17 bytes encode to at most 24 base58 characters, and `base58_decode` is
    // quadratic, so a longer sender-supplied claim must never reach it.
    if encoded.len() > 24 {
        return None;
    }
    let bytes = base58_decode(encoded)?;

🤖 Prompt to fix this finding with an AI agent

Findings from an automated review of commit 3f9eb9ff59c739c560892cc2ef19e38912e49cfe. Verify each one against the current code before changing anything, fix only those that still apply, and keep each fix to the lines it names.

---

F1 · Critical · security · apps/ingest/src/sources.rs:224-225
`source_uuid_from_public_id` base58-decodes unbounded sender input
A sender-supplied `maple_source_id` resource attribute is fed to `base58_decode` before its length is checked, and that decode is quadratic in its input: a 20 MB attribute (the body ceiling at `apps/ingest/src/main.rs:4897`) is ~10^14 byte operations, pinning an ingest task's CPU for hours and, with a few concurrent requests, wedging the shared fleet. Any default key that can reach the OTLP routes can send it — the claim path is reached from `verify_source_claims` (`main.rs:6621`) — and default-key traffic is not covered by the new source rate limit. Reject any body longer than one a valid `src_` id can have (17 bytes encode to at most 24 base58 characters) before decoding.
Replace those lines with:
    let encoded = public_id.strip_prefix(PUBLIC_ID_PREFIX)?;
    // 17 bytes encode to at most 24 base58 characters, and `base58_decode` is
    // quadratic, so a longer sender-supplied claim must never reach it.
    if encoded.len() > 24 {
        return None;
    }
    let bytes = base58_decode(encoded)?;

Production impact

Production traffic of the changed files (last 7 days)

File	Calls/day	Busiest operations
apps/ingest/src/sources.rs	31k	POST /v1/sessionEvents 31k/day, 0.0% err POST /v1/events 15/day, 0.0% err POST /v1/events 10/day, 0.0% err

What was checked

• base58 encoding matches the API's: same alphabet and 0x01 mode, and src_aXunrkWtzvXSkw9Z2smcft matches packages/domain/src/http/v2/sources.ts:23
• All four new Postgres statements filter org_id and compare text ids (packages/db/src/tables/sources.ts:17)
• New DB work is wrapped in postgres_client_span; paused drops are counted and tagged maple.ingest.reject_reason

_{3f9eb9f · Updated on every push. Reply "won't fix" to dismiss a finding, or mention @maple-review-bot to ask about one.}

[maple-review-bot]

1 inline note from Maple's review. The score and summary are in the review comment above.

[maple-review-bot]

Note

⏳ Maple is reviewing this pull request at c53b2b3. This comment updates with the review when it finishes.

[maple-review-bot]

Maple review

🟢 Confidence 8/10 · likely safe to merge
The bounded-decode fix is pinned by a test and the only path to base58_decode; the wider sources gating in this PR is unchanged since the last review.
_{quality 100/100 · no findings · tests covered · risk medium}

Hardens the source-claim path: sender-supplied maple_source_id values longer than a valid public id are rejected before the quadratic base58 decode, in source_uuid_from_public_id and in claimed_source_ids. The earlier unbounded-decode issue is fixed; safe to merge.

• source_uuid_from_public_id returns None past MAX_PUBLIC_ID_LEN before decoding
• claimed_source_ids drops oversized claims before they are decoded or cached

Before merge

• Infra · Check every environment has what this config now expects · apps/ingest/alchemy.run.ts

Fixed since the last review

• ✅ F1 · source_uuid_from_public_id base58-decodes unbounded sender input

Production impact

Production traffic of the changed files (last 7 days)

File	Calls/day	Busiest operations
apps/ingest/src/sources.rs	31k	POST /v1/sessionEvents 31k/day, 0.0% err POST /v1/events 15/day, 0.0% err POST /v1/events 10/day, 0.0% err

What was checked

• MAX_PUBLIC_ID_LEN (28) covers the 24-char maximum of the 17-byte encoding, so no valid id is rejected (sources.rs:224)
• base58_decode is called only from source_uuid_from_public_id, so no other sender path reaches it
• The claims cache stays bounded at max_capacity(50_000) with a TTL (main.rs:821), so dropped claims cannot grow it

_{04d6bf6 · Updated on every push. Reply "won't fix" to dismiss a finding, or mention @maple-review-bot to ask about one.}

[maple-review-bot]

Maple review

🟢 Confidence 8/10 · likely safe to merge
Re-review of Rust unchanged since 04d6bf6: token auth, the 8-claim cap and the fleet rate-limit math are all pinned by the source tests.
_{quality 100/100 · no findings · tests covered · risk medium · 1/2 new units observable}

Enforces maple_src_ source tokens in the ingest gateway and stamps maple_source_id on every resource, with paused sources dropped and claims checked against the org's integration sources. The Rust is unchanged since the last review; the only files that moved are base-branch TS that is not part of this diff.

• authorize_source resolves maple_src_ tokens by HMAC and stamps maple_source_id
• Paused sources return 200 with data dropped and counted
• Sender claims verified against org integration sources, capped at 8
• Per-instance rate limit split by INGEST_RATE_LIMIT_FLEET_SIZE

Before merge

• Infra · Check every environment has what this config now expects · apps/ingest/alchemy.run.ts
• Manual · Deploy only after the sources/integrations migration from the base PR (feat(sources): integration sources, one health model and list_sources #1536) is applied
• Manual · Set INGEST_RATE_LIMIT_FLEET_SIZE per environment; alchemy derives it from scaling.min

Production impact

Production traffic of the changed files (last 7 days)

File	Calls/day	Busiest operations
apps/ingest/src/sources.rs	31k	POST /v1/sessionEvents 31k/day, 0.0% err POST /v1/events 15/day, 0.0% err POST /v1/events 10/day, 0.0% err

What was checked

• Claim lookup is org-scoped and integration-only (claimable_source_ids in apps/ingest/src/sources.rs)
• source_uuid_from_public_id length-guards before decoding and re-encodes, so non-canonical claims never match
• Fleet split math pinned by per_instance_limit tests (100/3→34, 10/4→3, 1/8→1)

Observability coverage: 1 of 2 changes observable

Change	Kind	Observable	Evidence
Source-token authorization and resource stamping in the ingest gateway (OTLP/HTTP, gRPC, replay, GCP/CF pushes)	server inbound ingest	yes	apps/ingest/src/otel.rs sets maple.source.id on the ingest server spans and Span::current().record("maple.ingest.reject_reason", ...) on every rejection; apps/ingest/src/metrics.rs adds the paused-drop counter
Throttled sources.last_used_at write and claim verification against Postgres	db write / query	no	No Client/DB span, matching how fetch_ingest_key already reaches MAPLE_DB untraced in this gateway; visible on the enclosing server span

_{f4ab22b · Updated on every push. Reply "won't fix" to dismiss a finding, or mention @maple-review-bot to ask about one.}

[maple-review-bot]

Maple review

🟢 Confidence 8/10 · likely safe to merge
Token auth, claim verification and stamping read end to end, each pinned by a test; the rest of main.rs's 2.1k-line diff read in part.
_{quality 100/100 · no findings · tests covered · risk high · 5/5 new units observable}

Ingest now authenticates maple_src_ tokens against sources, gates each signal (401 expired, 403 signal, paused drops with 200, 429 rate limit) and stamps maple_source_id, keeping only verified org-owned claims. No defects found; safe to merge once its base integration migration is deployed.

• authorize_source gates expiry, signals, pause and per-source rate limits before decode
• enrich_resource_attributes strips sender maple_source_id, then stamps verified claims or the token source
• last_used_at written off-path at 1 min (token) / 5 min (default), skipped for kept claims
• New pause counter, maple.source.id span field and INGEST_RATE_LIMIT_FLEET_SIZE env

Before merge

• Infra · Check every environment has what this config now expects · apps/ingest/alchemy.run.ts
• Manual · Land feat(sources): integration sources, one health model and list_sources #1536 first: ingest reads sources.integration_id, added by that migration

Production impact

Production traffic of the changed files (last 7 days)

File	Calls/day	Busiest operations
apps/ingest/src/sources.rs	31k	POST /v1/sessionEvents 31k/day, 0.0% err POST /v1/events 15/day, 0.0% err POST /v1/events 10/day, 0.0% err

What was checked

• Claim verification is org-scoped, deduped, capped at 8, oversized claims dropped before decode (main.rs:7838, sources.rs:232)
• Every new Postgres call has postgres_client_span (main.rs:7529, 7795, 7842); pause has a counter
• authorize_source runs once per request; gRPC reuses the middleware admission (main.rs:3033, 3065)

Observability coverage: 5 of 5 changes observable

Change	Kind	Observable	Evidence
source token auth (fetch_source_by_token_hash)	db query	yes	postgres_client_span("fetch_source_by_token_hash", "SELECT", "sources") main.rs:7529
claim verification (claimable_source_ids)	db query	yes	postgres_client_span("claimable_source_ids", …) main.rs:7842, cached per claim
last_used_at write (record_source_used)	db query	yes	postgres_client_span("record_source_used", "UPDATE", "sources") main.rs:7795, off the request path
HTTP and gRPC admission (authorize_source)	request gate	yes	server span records maple.source.id (main.rs:4859) and maple.ingest.reject_reason; ingest_source_paused_drops_* counters in metrics.rs
resource attribute stamping (enrich_resource_attributes)	payload rewrite	yes	stamped attributes reach the warehouse and are counted by source-usage.ts

_{4ac57b0 · Updated on every push. Reply "won't fix" to dismiss a finding, or mention @maple-review-bot to ask about one.}

[maple-review-bot]

Maple review

🟢 Confidence 9/10 · safe to merge
Every new gate (expiry, signals, pause, rate limit, claims) is pinned by unit and end-to-end tests; the real fleet-size value in production is the one thing tests cannot prove.
_{no findings · tests covered · risk high −1 · 4/4 new units observable}

The ingest gateway now enforces maple_src_ tokens (expiry, signal allow-list, pause, rate limit) and stamps maple_source_id on every resource, verifying claimed ids against the org's integration sources. Contained to apps/ingest, well covered; safe to merge after the migrations PR.

• authorize_source gates every ingest request: expired 401, disallowed signal 403, paused dropped, over limit 429
• enrich_resource_attributes strips sender maple_source_id, applies overrides, then stamps the resolved source
• Claimed source ids are verified in one org_id-scoped query and capped at 8 per request
• note_source_use throttles last_used_at per instance with its own root span

Before merge

• Infra · Check every environment has what this config now expects · apps/ingest/alchemy.run.ts
• Manual · Deploy after feat(sources): integration sources, one health model and list_sources #1536: fetch_ingest_key and fetch_source_by_token_hash read sources.integration_id and sources.is_default

Production impact

Production traffic of the changed files (last 7 days)

File	Calls/day	Busiest operations
apps/ingest/src/sources.rs	32k	POST /v1/sessionEvents 32k/day, 0.0% err POST /v1/events 16/day, 0.0% err POST /v1/events 10/day, 0.0% err

What was checked

• Paused path returns before decode, entitlement, org limiter and usage (main.rs:4874), so nothing is stored or billed
• The (response, item_count, org_id, decoded_bytes) order matches the caller and billing (main.rs:5013, 4669)
• GCP push labels map to fixed attribute names (gcp_logging.rs), so a push cannot spoof maple_source_id

Observability coverage: 4 of 4 changes observable

Change	Kind	Observable	Evidence
Source token auth and per-request policy checks	inbound request path	yes	maple.source.id on auth_internal_span and the server span; maple.ingest.reject_reason = source_paused; ingest_source_paused_drops_total
Postgres lookups for source token, claim verification and last_used_at	outbound database call	yes	each wrapped in postgres_client_span with db.system/peer.service
Per-source rate limiting	in-process throttle	yes	org_throttled(org_id, "key_rate_limit") on a rejected token
last_used_at write	background write	yes	root record_source_use span following from the request span

_{eea4570 · Updated on every push. Reply "won't fix" to dismiss a finding, or mention @maple-review-bot to ask about one.}

[maple-review-bot]

Merged with 2 steps from "Before merge" still open. If they are done, tick them on the review comment; if not, now is the time.

• Infra · Check every environment has what this config now expects
• Manual · Deploy after feat(sources): integration sources, one health model and list_sources #1536: fetch_ingest_key and fetch_source_by_token_hash read sources.integration_id and sources.is_default

[maple-review-bot]

Maple: after this shipped

✅ Shipped clean. Nothing this change touches got worse in the hour after it deployed.

Deployed to maple-api (pr-1585) at 2026-10-19 18:58 UTC as e297dce, the first version after the merge (the merge commit itself never reported). Compared the 60 minutes before and after.

Operation	Calls/h	Errors	p95
POST /v1/sessionEvents	2.4k → 2.3k	0.0% → 0.0%	0 → 1 ms
POST /v1/events	0 → 0	0.0% → 0.0%	0 → 0 ms
POST /v1/events	0 → 0	0.0% → 0.0%	0 → 0 ms

_{Maple looks once, an hour after the first deploy that carries the merge.}