fix(cli): decode fixed32 OTLP span flags

.github/workflows/deploy-pr-preview.yml

@@ -59,6 +59,13 @@ jobs:
             - name: Install dependencies
               run: bun install --frozen-lockfile
 
+            # Tinybird's TypeScript resource generator loads @maple/domain from source,
+            # whose span-display helpers import this package through its dist exports.
+            # A clean runner has no dist/ until the workspace dependency is built.
+            - name: Build Tinybird TypeScript workspace dependencies
+              if: ${{ github.event.action != 'closed' }}
+              run: bun turbo build --filter=@maple-dev/clickhouse-builder
+
             # Needed for both branch create (deploy) and rm (close). `tb` is the
             # real Tinybird CLI (branch commands live there, not in `bunx tinybird`).
             - name: Install Tinybird CLI

apps/api/src/lib/WarehouseQueryService.test.ts

@@ -267,30 +267,23 @@ describe("WarehouseQueryService raw-SQL provider routing", () => {
 
 describe("bounded Tinybird response fetch", () => {
 	it("accepts an exact-boundary response and aborts one byte over", async () => {
-		const realFetch = globalThis.fetch
+		const exact = await __testables.boundedResponseFetch(
+			MAX_RAW_SQL_RESULT_BYTES,
+			(async () => new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES))) as typeof fetch,
+		)("https://api.tinybird.example/v0/sql")
+		assert.strictEqual((await exact.arrayBuffer()).byteLength, MAX_RAW_SQL_RESULT_BYTES)
+
+		let thrown: unknown
 		try {
-			globalThis.fetch = (async () =>
-				new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES))) as typeof fetch
-			const exact = await __testables.boundedResponseFetch(MAX_RAW_SQL_RESULT_BYTES)(
-				"https://api.tinybird.example/v0/sql",
-			)
-			assert.strictEqual((await exact.arrayBuffer()).byteLength, MAX_RAW_SQL_RESULT_BYTES)
-
-			globalThis.fetch = (async () =>
-				new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES + 1))) as typeof fetch
-			let thrown: unknown
-			try {
-				await __testables.boundedResponseFetch(MAX_RAW_SQL_RESULT_BYTES)(
-					"https://api.tinybird.example/v0/sql",
-				)
-			} catch (error) {
-				thrown = error
-			}
-			assert.instanceOf(thrown, Error)
-			assert.match((thrown as Error).message, /5000000 encoded bytes/)
-		} finally {
-			globalThis.fetch = realFetch
+			await __testables.boundedResponseFetch(
+				MAX_RAW_SQL_RESULT_BYTES,
+				(async () => new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES + 1))) as typeof fetch,
+			)("https://api.tinybird.example/v0/sql")
+		} catch (error) {
+			thrown = error
 		}
+		assert.instanceOf(thrown, Error)
+		assert.match((thrown as Error).message, /5000000 encoded bytes/)
 	})
 })
 
@@ -652,8 +645,7 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert
 			auth?: string
 			body: string
 		}> = []
-		const realFetch = globalThis.fetch
-		globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
+		const requestFetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
 			const headers = (init?.headers ?? {}) as Record<string, string>
 			captured.push({
 				url: String(input),
@@ -665,12 +657,8 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert
 			return new Response("", { status: 202 })
 		}) as typeof fetch
 
-		try {
-			const client = __testables.createTinybirdSdkSqlClient(tbConfig)
-			await client.insert("traces", [{ trace_id: "a" }, { trace_id: "b" }])
-		} finally {
-			globalThis.fetch = realFetch
-		}
+		const client = __testables.createTinybirdSdkSqlClient(tbConfig, requestFetch)
+		await client.insert("traces", [{ trace_id: "a" }, { trace_id: "b" }])
 
 		assert.strictEqual(captured.length, 1)
 		const req = captured[0]!
@@ -685,18 +673,13 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert
 
 	it("no-ops on an empty row set (no request issued)", async () => {
 		let calls = 0
-		const realFetch = globalThis.fetch
-		globalThis.fetch = (async () => {
+		const requestFetch = (async () => {
 			calls++
 			return new Response("", { status: 202 })
 		}) as typeof fetch
 
-		try {
-			const client = __testables.createTinybirdSdkSqlClient(tbConfig)
-			await client.insert("traces", [])
-		} finally {
-			globalThis.fetch = realFetch
-		}
+		const client = __testables.createTinybirdSdkSqlClient(tbConfig, requestFetch)
+		await client.insert("traces", [])
 
 		assert.strictEqual(calls, 0)
 	})
@@ -715,24 +698,21 @@ describe("createTinybirdSdkSqlClient.sql FORMAT normalization", () => {
 
 	const captureSql = async (sql: string): Promise<string> => {
 		const sent: string[] = []
-		const realFetch = globalThis.fetch
-		globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
-			const url = String(input)
-			const fromParam = new URL(url).searchParams.get("q")
-			const body = typeof init?.body === "string" ? init.body : ""
-			sent.push(fromParam ?? body)
+		const requestFetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
+			const url = new URL(String(input))
+			if (url.pathname.endsWith("/v0/sql")) {
+				const fromParam = url.searchParams.get("q")
+				const body = typeof init?.body === "string" ? init.body : ""
+				sent.push(fromParam ?? body)
+			}
 			return new Response(JSON.stringify({ meta: [], data: [], rows: 0 }), {
 				status: 200,
 				headers: { "Content-Type": "application/json" },
 			})
 		}) as typeof fetch
 
-		try {
-			const client = __testables.createTinybirdSdkSqlClient(tbConfig)
-			await client.sql(sql, undefined)
-		} finally {
-			globalThis.fetch = realFetch
-		}
+		const client = __testables.createTinybirdSdkSqlClient(tbConfig, requestFetch)
+		await client.sql(sql, undefined)
 
 		assert.strictEqual(sent.length, 1)
 		return sent[0]!

apps/api/src/lib/WarehouseQueryService.ts

@@ -106,9 +106,9 @@ const isEmptyJsonBodyError = (error: unknown): boolean =>
 	error instanceof SyntaxError && /unexpected end of json input/i.test(error.message)
 
 const boundedResponseFetch =
-	(maxBytes: number): typeof fetch =>
+	(maxBytes: number, requestFetch: typeof fetch = fetch): typeof fetch =>
 	async (input, init) => {
-		const response = await fetch(input, init)
+		const response = await requestFetch(input, init)
 		if (response.body === null) return response
 
 		const reader = response.body.getReader()
@@ -145,15 +145,18 @@ const boundedResponseFetch =
 		})
 	}
 
-const createTinybirdSdkSqlClient = (config: TinybirdBackendConfig): WarehouseSqlClient => {
-	const makeClient = (fetchAdapter?: typeof fetch) =>
+const createTinybirdSdkSqlClient = (
+	config: TinybirdBackendConfig,
+	requestFetch: typeof fetch = fetch,
+): WarehouseSqlClient => {
+	const makeClient = (fetchAdapter: typeof fetch = requestFetch) =>
 		new Tinybird({
 			baseUrl: config.host,
 			token: config.token,
 			datasources: {},
 			pipes: {},
 			devMode: false,
-			...(fetchAdapter === undefined ? {} : { fetch: fetchAdapter }),
+			fetch: fetchAdapter,
 		})
 	const client = makeClient()
 	const boundedClients = new Map<number, typeof client>()
@@ -176,7 +179,7 @@ const createTinybirdSdkSqlClient = (config: TinybirdBackendConfig): WarehouseSql
 				if (limits !== undefined) {
 					queryClient =
 						boundedClients.get(limits.maxBytes) ??
-						makeClient(boundedResponseFetch(limits.maxBytes))
+						makeClient(boundedResponseFetch(limits.maxBytes, requestFetch))
 					boundedClients.set(limits.maxBytes, queryClient)
 				}
 				const result = await queryClient.sql<Record<string, unknown>>(jsonSql)
@@ -198,7 +201,7 @@ const createTinybirdSdkSqlClient = (config: TinybirdBackendConfig): WarehouseSql
 			if (rows.length === 0) return
 			const ndjson = rows.map((row) => JSON.stringify(row)).join("\n")
 			const url = `${config.host.replace(/\/$/, "")}/v0/events?name=${encodeURIComponent(datasource)}&wait=false`
-			const response = await fetch(url, {
+			const response = await requestFetch(url, {
 				method: "POST",
 				headers: {
 					"Content-Type": "application/x-ndjson",
@@ -261,7 +264,8 @@ export class WarehouseQueryService extends Context.Service<
 						url: clickhouseUrl.toString().replace(/\/$/, ""),
 						username: env.CLICKHOUSE_USER,
 						password: Option.match(env.CLICKHOUSE_PASSWORD, {
-							onNone: () => (kind === "tinybird-gateway" ? Redacted.value(env.TINYBIRD_TOKEN) : ""),
+							onNone: () =>
+								kind === "tinybird-gateway" ? Redacted.value(env.TINYBIRD_TOKEN) : "",
 							onSome: Redacted.value,
 						}),
 						database: env.CLICKHOUSE_DATABASE,
@@ -350,8 +354,7 @@ export class WarehouseQueryService extends Context.Service<
 						password: override.value.password,
 						database: override.value.database,
 					},
-					clientCacheKey:
-						purpose === "raw" ? `raw:${tenant.orgId}` : `read:${tenant.orgId}`,
+					clientCacheKey: purpose === "raw" ? `raw:${tenant.orgId}` : `read:${tenant.orgId}`,
 				}
 			}
 

apps/cli/src/server/otlp/encode.test.ts

@@ -41,6 +41,11 @@ const b64 = (h: string): string => Buffer.from(h, "hex").toString("base64")
 const javaAgentTracePayloadB64 =
 	"CvULCqUICiUKG2RlcGxveW1lbnQuZW52aXJvbm1lbnQubmFtZRIGCgR0ZXN0ChYKCWhvc3QuYXJjaBIJCgdhYXJjaDY0CisKCWhvc3QubmFtZRIeChx6aGFvamlydWlkZU1hY0Jvb2stQWlyLmxvY2FsCiMKDm9zLmRlc2NyaXB0aW9uEhEKD01hYyBPUyBYIDI2LjQuMQoTCgdvcy50eXBlEggKBmRhcndpbgoWCgpvcy52ZXJzaW9uEggKBjI2LjQuMQqgAQoUcHJvY2Vzcy5jb21tYW5kX2FyZ3MShwEqhAEKVQpTL1VzZXJzL3poYW9qaXJ1aS9MaWJyYXJ5L0phdmEvSmF2YVZpcnR1YWxNYWNoaW5lcy9vcGVuamRrLTIzL0NvbnRlbnRzL0hvbWUvYmluL2phdmEKKwopL3RtcC9tYXBsZS1qYXZhLWFnZW50LTdLMHcvT3RlbFJlcHJvLmphdmEKcAoXcHJvY2Vzcy5leGVjdXRhYmxlLnBhdGgSVQpTL1VzZXJzL3poYW9qaXJ1aS9MaWJyYXJ5L0phdmEvSmF2YVZpcnR1YWxNYWNoaW5lcy9vcGVuamRrLTIzL0NvbnRlbnRzL0hvbWUvYmluL2phdmEKEwoLcHJvY2Vzcy5waWQSBBjtpQIKVwobcHJvY2Vzcy5ydW50aW1lLmRlc2NyaXB0aW9uEjgKNk9yYWNsZSBDb3Jwb3JhdGlvbiBPcGVuSkRLIDY0LUJpdCBTZXJ2ZXIgVk0gMjMrMzctMjM2OQo1ChRwcm9jZXNzLnJ1bnRpbWUubmFtZRIdChtPcGVuSkRLIFJ1bnRpbWUgRW52aXJvbm1lbnQKJwoXcHJvY2Vzcy5ydW50aW1lLnZlcnNpb24SDAoKMjMrMzctMjM2OQo9ChNzZXJ2aWNlLmluc3RhbmNlLmlkEiYKJGQxZTI5NjA0LTI5ODYtNDlmZC1iNWJiLWU0ZjFhN2Q0ZjkyYwohCgxzZXJ2aWNlLm5hbWUSEQoPd2lzY2hvaWNlci11c2VyCj0KFXRlbGVtZXRyeS5kaXN0cm8ubmFtZRIkCiJvcGVudGVsZW1ldHJ5LWphdmEtaW5zdHJ1bWVudGF0aW9uCiQKGHRlbGVtZXRyeS5kaXN0cm8udmVyc2lvbhIICgYyLjI4LjEKIAoWdGVsZW1ldHJ5LnNkay5sYW5ndWFnZRIGCgRqYXZhCiUKEnRlbGVtZXRyeS5zZGsubmFtZRIPCg1vcGVudGVsZW1ldHJ5CiEKFXRlbGVtZXRyeS5zZGsudmVyc2lvbhIICgYxLjYyLjAKTwoXdmNzLnJlcG9zaXRvcnkudXJsLmZ1bGwSNAoyaHR0cHM6Ly9naXRodWIuY29tL1dpc2Nob2ljZXItWGlhbi93aXNjaG9pY2VyLXVzZXISoQMKMQohaW8ub3BlbnRlbGVtZXRyeS5qYXZhLWh0dHAtY2xpZW50EgwyLjI4LjEtYWxwaGESwgIKEGmLshO9k6DTrmM7fx2w70cSCD/XzWWx87T5KgNHRVQwAzkgsMwtvzK3GEEJ4tgtvzK3GEoVCgt0aHJlYWQubmFtZRIGCgRtYWluSi0KCHVybC5mdWxsEiEKH2h0dHA6Ly8xMjcuMC4wLjE6MTgwODEvcGluZz9pPTJKEwoLc2VydmVyLnBvcnQSBBihjQFKEwoKZXJyb3IudHlwZRIFCgM0MDRKHQoOc2VydmVyLmFkZHJlc3MSCwoJMTI3LjAuMC4xSiEKGG5ldHdvcmsucHJvdG9jb2wudmVyc2lvbhIFCgMxLjFKIAoZaHR0cC5yZXNwb25zZS5zdGF0dXNfY29kZRIDGJQDShwKE2h0dHAucmVxdWVzdC5tZXRob2QSBQoDR0VUSg8KCXRocmVhZC5pZBICGAF6AhgChQEDAQAAGidodHRwczovL29wZW50ZWxlbWV0cnkuaW8vc2NoZW1hcy8xLjM3LjAaJ2h0dHBzOi8vb3BlbnRlbGVtZXRyeS5pby9zY2hlbWFzLzEuMjQuMA=="
 
+// Minimal ExportTraceServiceRequest encoded with the upstream OTLP schema.
+// Span.flags is field 16, so its fixed32 key is the two-byte sequence 0x85 0x01.
+const pythonSpanFlagsPayloadHex =
+	"0a480a00124412420a1000000000000000000000000000000000120800000000000000002a0a776974682d666c6167733001390100000000000000410200000000000000850101000000"
+
 const attr = (key: string, str: string) => ({ key, value: { stringValue: str } })
 
 const sampleTraceReq = () => ({
@@ -272,6 +277,17 @@ describe("protobuf round-trip (proves vendored .proto field numbers)", () => {
 })
 
 describe("external OTLP protobuf compatibility", () => {
+	it("decodes a non-zero fixed32 span flags field emitted by the Python SDK", () => {
+		const decoded = decodeTraceRequest(Buffer.from(pythonSpanFlagsPayloadHex, "hex")) as {
+			resourceSpans: Array<{ scopeSpans: Array<{ spans: Array<{ name: string; flags: number }> }> }>
+		}
+		const span = decoded.resourceSpans[0]?.scopeSpans[0]?.spans[0]
+		expect(span).toMatchObject({ name: "with-flags", flags: 1 })
+
+		const [batch] = encodeTraces(decoded)
+		expect(batch).toMatchObject({ datasource: "traces", rowCount: 1 })
+	})
+
 	it("decodes a Java agent OTLP/HTTP traces payload captured from the wire", () => {
 		const decoded = decodeTraceRequest(Buffer.from(javaAgentTracePayloadB64, "base64"))
 		const [batch] = encodeTraces(decoded)

apps/cli/src/server/otlp/proto.ts

@@ -119,13 +119,13 @@ message Span {
     string trace_state = 3;
     repeated KeyValue attributes = 4;
     uint32 dropped_attributes_count = 5;
-    uint32 flags = 6;
+    fixed32 flags = 6;
   }
 
   repeated Link links = 13;
   uint32 dropped_links_count = 14;
   Status status = 15;
-  uint32 flags = 16;
+  fixed32 flags = 16;
 }
 
 message Status {