feat(ingest): mirror Tinybird writes into a second workspace

.env.example

@@ -8,6 +8,16 @@ TINYBIRD_TOKEN=your-tinybird-token
 # Optional Tinybird-side ceiling shared by raw SQL from the API and alerting,
 # with an independent rate-limit bucket for each org. Unset means no JWT RPS limit.
 # TINYBIRD_RAW_SQL_JWT_RPS_LIMIT=100
+# A second Tinybird workspace the ingest gateway mirrors writes into, for the
+# duration of a workspace migration. Best-effort: the mirror has its own WAL
+# lane and can never slow or fail ingest. Set BOTH or neither — the gateway
+# refuses to start on half a config. Reads always stay on TINYBIRD_HOST.
+# TINYBIRD_MIRROR_HOST=https://api.us-east.aws.tinybird.co
+# TINYBIRD_MIRROR_TOKEN=your-mirror-workspace-token
+# Ramp knob: percentage of orgs to mirror, by org hash (default 100). Start at 1.
+# INGEST_TINYBIRD_MIRROR_SAMPLE_PERCENT=1
+# INGEST_TINYBIRD_MIRROR_MAX_ATTEMPTS=5
+# INGEST_TINYBIRD_MIRROR_TIMEOUT_MS=3000
 
 # ClickHouse
 CLICKHOUSE_URL=http://localhost:9000

.github/workflows/tinybird-cd.yml

@@ -36,6 +36,13 @@ jobs:
                       environment: tinybird-cd
                     - label: staging
                       environment: tinybird-cd-stg
+                    # The us-east-1 workspace being migrated to. Both production
+                    # legs must receive every schema deploy for the whole
+                    # dual-emit window — a month of drift between the two
+                    # workspaces is the quiet way that migration fails. No-ops
+                    # until the environment's secrets exist.
+                    - label: production-us
+                      environment: tinybird-cd-us
         environment: ${{ matrix.target.environment }}
         steps:
             - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6
@@ -57,12 +64,18 @@ jobs:
                     echo "::notice::TINYBIRD_TOKEN not set for the '${{ matrix.target.environment }}' environment; skipping Tinybird deploy."
                     exit 0
                   fi
-                  HOST_FLAG=()
+                  # `tinybird deploy` has NO --host flag (verified against the pinned
+                  # @tinybirdco/sdk 0.0.78: it exits `unknown option '--host'`). The
+                  # host comes from the config file's `baseUrl`, and the CLI resolves
+                  # config files in priority order, `tinybird.config.json` ahead of
+                  # `tinybird.json` — so writing one here overrides the repo default
+                  # without making TINYBIRD_HOST mandatory for `tb build` locally.
                   if [ -n "${TINYBIRD_HOST:-}" ]; then
-                    HOST_FLAG=(--host "$TINYBIRD_HOST")
+                    jq --arg host "$TINYBIRD_HOST" '.baseUrl = $host' tinybird.json > tinybird.config.json
+                    echo "::notice::Deploying to $TINYBIRD_HOST"
                   fi
                   if [ "$ALLOW_DESTRUCTIVE" = "true" ]; then
-                    bunx tinybird deploy "${HOST_FLAG[@]}" --allow-destructive-operations
+                    bunx tinybird deploy --allow-destructive-operations
                   else
-                    bunx tinybird deploy "${HOST_FLAG[@]}"
+                    bunx tinybird deploy
                   fi

.gitignore

@@ -1,4 +1,7 @@
 .tinyb
+# Written by the Tinybird CD job to point `deploy` at a specific workspace host
+# (the CLI has no --host flag); takes priority over the committed tinybird.json.
+tinybird.config.json
 .tinybird-generated
 .tinybird-entities-*.mjs
 # D1→Postgres migration dumps — contain prod data, never commit

CONTRIBUTING.md

@@ -238,6 +238,7 @@ port and keep `VITE_*` / `MAPLE_INGEST_PUBLIC_URL` consistent.
 | `MAPLE_ORG_ID_OVERRIDE`            | with static    | Must match `MAPLE_DEFAULT_ORG_ID`                                       |
 | `MAPLE_PG_URL`                     | postgres store | `postgres://maple:maple@localhost:5499/maple` if not using static store |
 | `TINYBIRD_HOST` / `TINYBIRD_TOKEN` | tinybird mode  | When `INGEST_WRITE_MODE=tinybird` or `dual`                             |
+| `TINYBIRD_MIRROR_HOST` / `_TOKEN`  | migration only | Mirrors writes into a second workspace; best-effort, set both or neither |
 | `INGEST_PORT`                      | optional       | Default from port / env                                                 |
 | `INGEST_REQUIRE_TLS`               | optional       | `false` locally                                                         |
 

apps/ingest/alchemy.run.ts

@@ -62,14 +62,25 @@ const INGEST_PORT = 3474
  * default (`INGEST_QUEUE_MAX_BYTES` = 20 GiB, `apps/ingest/src/main.rs`) would
  * sit exactly on the line. 8 GiB still buys hours of buffering at current
  * volume; raising it means paying for ephemeral storage beyond the free tier.
+ *
+ * The per-lane budget is `INGEST_QUEUE_MAX_BYTES / (WAL_SHARDS * lanes)`, so
+ * the Tinybird mirror's third lane would have cut every lane's share from 1 GiB
+ * to 683 MiB at exactly the moment Tinybird-bound traffic doubled. 12 GiB over
+ * 12 lanes restores the 1 GiB per lane that 8 GiB gave across 8, and still
+ * leaves ~8 GB of the ephemeral allowance for the image and the OS.
+ *
+ * Drop back to 8 GiB once the mirror is removed, or every lane silently gains
+ * headroom nobody sized for.
  */
-const WAL_MAX_BYTES = 8 * 1024 * 1024 * 1024
+const WAL_MAX_BYTES = 12 * 1024 * 1024 * 1024
 
 /**
  * Pinned rather than derived. The gateway defaults to `num_cpus * 2`, which
  * makes on-disk layout and fd count a function of task size — so a cpu bump, or
- * a move to another capacity provider, would silently reshape the WAL. Two lanes per shard
- * (Tinybird + ClickHouse) means this is 8 open WAL files.
+ * a move to another capacity provider, would silently reshape the WAL. Three
+ * lanes per shard (Tinybird + ClickHouse + Tinybird mirror) means this is 12
+ * open WAL files; the mirror lane is present but idle unless
+ * `TINYBIRD_MIRROR_HOST`/`TINYBIRD_MIRROR_TOKEN` are set.
  */
 const WAL_SHARDS = 4
 
@@ -229,6 +240,22 @@ export const createMapleIngest = ({ stage, domains, region, replayBlobs }: Creat
 		const autumnKey = process.env.AUTUMN_SECRET_KEY?.trim()
 		const autumnSecret = autumnKey ? yield* secret("autumn-secret-key", autumnKey) : undefined
 
+		// Second Tinybird workspace to mirror writes into during a workspace
+		// migration. Both halves must be set together; the gateway rejects a
+		// half-configured mirror at startup rather than 401-ing its lane forever.
+		//
+		// Resolved through `optionalPlain`, not `process.env`: alchemy reads
+		// `--env-file`/`.env` through its own ConfigProvider and never copies
+		// those values into `process.env`, so a bare read would see the var in
+		// CI and miss it locally. The host is plain (not a secret); the token is
+		// resolved here only to mint the Secrets Manager entry below, exactly as
+		// TINYBIRD_TOKEN is, and never reaches `env`.
+		const tinybirdMirrorHostEntry = yield* optionalPlain("TINYBIRD_MIRROR_HOST")
+		const tinybirdMirrorTokenValue = (yield* optionalPlain("TINYBIRD_MIRROR_TOKEN")).TINYBIRD_MIRROR_TOKEN
+		const tinybirdMirrorToken = tinybirdMirrorTokenValue
+			? yield* secret("tinybird-mirror-token", tinybirdMirrorTokenValue)
+			: undefined
+
 		// Both halves are stack-minted (`createReplayBlobStore`), so there is no
 		// half-set config left to guard against. The access key id is not secret,
 		// but it only exists once the token does and `env` takes plan-time strings
@@ -431,6 +458,9 @@ export const createMapleIngest = ({ stage, domains, region, replayBlobs }: Creat
 				MAPLE_INGEST_KEY_ENCRYPTION_KEY: keyEncryptionKey.secretArn,
 				MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: keyLookupHmacKey.secretArn,
 				...(autumnSecret ? { AUTUMN_SECRET_KEY: autumnSecret.secretArn } : undefined),
+				...(tinybirdMirrorToken
+					? { TINYBIRD_MIRROR_TOKEN: tinybirdMirrorToken.secretArn }
+					: undefined),
 				...(replayR2Secret && replayR2AccessKeyId
 					? {
 							INGEST_REPLAY_R2_SECRET_ACCESS_KEY: replayR2Secret.secretArn,
@@ -496,6 +526,14 @@ export const createMapleIngest = ({ stage, domains, region, replayBlobs }: Creat
 				...(yield* optionalPlain("INGEST_MAX_REQUEST_BODY_BYTES")),
 				...(yield* optionalPlain("INGEST_EXPORT_MAX_ATTEMPTS")),
 				...(yield* optionalPlain("INGEST_TINYBIRD_CONCURRENCY_PER_SHARD")),
+				// Tinybird mirror. The host is plain (it is not a secret); the token
+				// goes through Secrets Manager above. Ramp with SAMPLE_PERCENT: start
+				// at 1, confirm rows land and `ingest_tinybird_mirror_dropped_total`
+				// stays flat, then climb to 100.
+				...tinybirdMirrorHostEntry,
+				...(yield* optionalPlain("INGEST_TINYBIRD_MIRROR_SAMPLE_PERCENT")),
+				...(yield* optionalPlain("INGEST_TINYBIRD_MIRROR_MAX_ATTEMPTS")),
+				...(yield* optionalPlain("INGEST_TINYBIRD_MIRROR_TIMEOUT_MS")),
 				...(yield* optionalPlain("INGEST_REPLAY_MAX_SESSION_BYTES")),
 				// The org Maple's own telemetry is filed under. Required here and in
 				// the gateway (`AppConfig::from_env`), with no fallback on either

apps/ingest/benches/ingest_bench.rs

@@ -105,6 +105,7 @@ impl BenchFixture {
             TinybirdConfig {
                 endpoint: format!("http://{addr}"),
                 token: "bench-token".to_owned(),
+                mirror: None,
                 queue_dir: queue_dir.clone(),
                 // Effectively uncapped: this benchmark measures accept latency
                 // (encode + WAL append + ack), not back-pressure. A single org

apps/ingest/src/main.rs

@@ -56,6 +56,7 @@ use maple_ingest::telemetry::{
     AttributeMappingRule, ClickHouseBreakerConfig, ClickHouseTarget, ClickHouseTargetProvider,
     DatasourceNames, ExportDestination, HttpClient, MappingOperation, MappingSourceContext,
     PipelineError, SamplingPolicy, TelemetryPipeline, TelemetrySignal, TinybirdConfig,
+    TinybirdMirrorConfig,
 };
 use maple_ingest::usage_metrics::{billable_gb, usage_cardinality_view, UsageMetrics};
 use moka::future::Cache;
@@ -262,6 +263,51 @@ impl AppConfig {
             10_000,
         )?;
 
+        // A second Tinybird workspace to mirror writes into during a workspace
+        // migration. Present only when both halves are set, so the feature is
+        // off everywhere that does not opt in; setting exactly one half is a
+        // deploy mistake and `validate_for_pipeline` rejects it rather than
+        // letting the lane 401 in silence.
+        let mirror_host = std::env::var("TINYBIRD_MIRROR_HOST")
+            .unwrap_or_default()
+            .trim()
+            .trim_end_matches('/')
+            .to_owned();
+        let mirror_token = std::env::var("TINYBIRD_MIRROR_TOKEN")
+            .unwrap_or_default()
+            .trim()
+            .to_owned();
+        let mirror = if mirror_host.is_empty() && mirror_token.is_empty() {
+            None
+        } else {
+            Some(TinybirdMirrorConfig {
+                endpoint: mirror_host,
+                token: mirror_token,
+                // Deliberately far below the primary's 20: the mirror holds a
+                // lane worker for the whole budget, and losing a mirror batch is
+                // cheaper than stalling the lane behind a sick workspace.
+                max_attempts: parse_u32(
+                    "INGEST_TINYBIRD_MIRROR_MAX_ATTEMPTS",
+                    std::env::var("INGEST_TINYBIRD_MIRROR_MAX_ATTEMPTS").ok(),
+                    5,
+                )?,
+                export_timeout: Duration::from_millis(parse_u64(
+                    "INGEST_TINYBIRD_MIRROR_TIMEOUT_MS",
+                    std::env::var("INGEST_TINYBIRD_MIRROR_TIMEOUT_MS").ok(),
+                    3_000,
+                )?),
+                // The ramp knob: start at 1, confirm rows land and nothing is
+                // shed, then climb. Sampling is by org hash, so an org is
+                // consistently in or out and its stream is never half-mirrored.
+                sample_percent: u8::try_from(parse_u32(
+                    "INGEST_TINYBIRD_MIRROR_SAMPLE_PERCENT",
+                    std::env::var("INGEST_TINYBIRD_MIRROR_SAMPLE_PERCENT").ok(),
+                    100,
+                )?)
+                .map_err(|_| "INGEST_TINYBIRD_MIRROR_SAMPLE_PERCENT must be 0..=100".to_owned())?,
+            })
+        };
+
         let tinybird = TinybirdConfig {
             endpoint: std::env::var("TINYBIRD_HOST")
                 .unwrap_or_default()
@@ -272,6 +318,7 @@ impl AppConfig {
                 .unwrap_or_default()
                 .trim()
                 .to_owned(),
+            mirror,
             queue_dir: PathBuf::from(
                 std::env::var("INGEST_QUEUE_DIR")
                     .unwrap_or_else(|_| "/var/lib/maple-ingest/wal".to_owned()),
@@ -353,6 +400,10 @@ impl AppConfig {
         };
         if write_mode.uses_tinybird() {
             tinybird.validate()?;
+        } else {
+            // The mirror is a Tinybird destination, so a half-configured one is
+            // still a deploy mistake in forward-only mode.
+            tinybird.validate_for_pipeline(false)?;
         }
 
         let max_request_body_bytes = parse_usize(
@@ -7022,6 +7073,7 @@ mod tests {
         TinybirdConfig {
             endpoint: String::new(),
             token: String::new(),
+            mirror: None,
             queue_dir,
             queue_max_bytes: 1024 * 1024,
             org_queue_max_bytes: 1024 * 1024,

apps/ingest/src/metrics.rs

@@ -175,6 +175,13 @@ static TINYBIRD_EXPORT_RETRIES_TOTAL: LazyLock<Counter<u64>> = LazyLock::new(||
         .build()
 });
 
+static TINYBIRD_MIRROR_DROPPED_TOTAL: LazyLock<Counter<u64>> = LazyLock::new(|| {
+    METER
+        .u64_counter("ingest_tinybird_mirror_dropped_total")
+        .with_description("Rows dropped before reaching the Tinybird mirror lane")
+        .build()
+});
+
 static CLICKHOUSE_EXPORT_ROWS_TOTAL: LazyLock<Counter<u64>> = LazyLock::new(|| {
     METER
         .u64_counter("ingest_clickhouse_export_rows_total")
@@ -553,14 +560,30 @@ pub fn org_queue_bytes(org_id: &str, bytes: u64) {
 }
 
 /// Latency and exported-byte size of a completed WAL export batch.
-pub fn export_batch_completed(shard: usize, signal: &str, duration_secs: f64, exported_bytes: u64) {
-    EXPORT_BATCH_DURATION_SECONDS
-        .record(duration_secs, &[KeyValue::new("shard", shard.to_string())]);
+///
+/// `destination` matters here: without it a mirror lane's drain is
+/// indistinguishable from the primary's in the same shard, which is exactly the
+/// comparison a workspace migration needs.
+pub fn export_batch_completed(
+    shard: usize,
+    destination: &str,
+    signal: &str,
+    duration_secs: f64,
+    exported_bytes: u64,
+) {
+    EXPORT_BATCH_DURATION_SECONDS.record(
+        duration_secs,
+        &[
+            KeyValue::new("shard", shard.to_string()),
+            KeyValue::new("destination", destination.to_owned()),
+        ],
+    );
     WAL_EXPORTED_BYTES.record(
         exported_bytes,
         &[
             KeyValue::new("signal", signal.to_owned()),
             KeyValue::new("shard", shard.to_string()),
+            KeyValue::new("destination", destination.to_owned()),
         ],
     );
 }
@@ -605,39 +628,73 @@ pub fn native_sampled_dropped(signal: &str, count: u64) {
 }
 
 /// A successful Tinybird export: latency and exported row count.
-pub fn tinybird_export_succeeded(datasource: &str, duration_secs: f64, rows: u64) {
+///
+/// `destination` is `tinybird` or `tinybird_mirror`. Comparing the two row
+/// counters per datasource is how a mirrored workspace is proved complete.
+pub fn tinybird_export_succeeded(
+    destination: &str,
+    datasource: &str,
+    duration_secs: f64,
+    rows: u64,
+) {
     TINYBIRD_EXPORT_DURATION_SECONDS.record(
         duration_secs,
         &[
+            KeyValue::new("destination", destination.to_owned()),
             KeyValue::new("datasource", datasource.to_owned()),
             KeyValue::new("status", "2xx"),
         ],
     );
-    TINYBIRD_EXPORT_ROWS_TOTAL.add(rows, &[KeyValue::new("datasource", datasource.to_owned())]);
+    TINYBIRD_EXPORT_ROWS_TOTAL.add(
+        rows,
+        &[
+            KeyValue::new("destination", destination.to_owned()),
+            KeyValue::new("datasource", datasource.to_owned()),
+        ],
+    );
 }
 
 /// Rows dropped while exporting to Tinybird (`status` is an HTTP code or `retries_exhausted`).
-pub fn tinybird_export_dropped(datasource: &str, status: &str, rows: u64) {
+pub fn tinybird_export_dropped(destination: &str, datasource: &str, status: &str, rows: u64) {
     TINYBIRD_EXPORT_DROPPED_TOTAL.add(
         rows,
         &[
+            KeyValue::new("destination", destination.to_owned()),
             KeyValue::new("datasource", datasource.to_owned()),
             KeyValue::new("status", status.to_owned()),
         ],
     );
 }
 
 /// A Tinybird export attempt was retried (`status` is an HTTP code or `transport`).
-pub fn tinybird_export_retry(datasource: &str, status: &str) {
+pub fn tinybird_export_retry(destination: &str, datasource: &str, status: &str) {
     TINYBIRD_EXPORT_RETRIES_TOTAL.add(
         1,
         &[
+            KeyValue::new("destination", destination.to_owned()),
             KeyValue::new("datasource", datasource.to_owned()),
             KeyValue::new("status", status.to_owned()),
         ],
     );
 }
 
+/// Rows shed on the commit path before they ever reached the mirror lane
+/// (`reason` is `lane_full`, `org_quota`, `wal_error`, or `no_target`).
+///
+/// The mirror is best-effort, so these drops are invisible to clients and to
+/// every other counter — and nothing backfills the mirrored workspace, so a
+/// non-zero value here is permanent loss, not a gap to be repaired later. It is
+/// the one number that has to stay at zero for the whole migration window.
+pub fn tinybird_mirror_dropped(datasource: &str, reason: &str, rows: u64) {
+    TINYBIRD_MIRROR_DROPPED_TOTAL.add(
+        rows,
+        &[
+            KeyValue::new("datasource", datasource.to_owned()),
+            KeyValue::new("reason", reason.to_owned()),
+        ],
+    );
+}
+
 /// A successful ClickHouse export: latency and exported row count.
 pub fn clickhouse_export_succeeded(datasource: &str, status: &str, duration_secs: f64, rows: u64) {
     let attrs = [