refactor(compute): filter logs on subservice column, not source attribute

apps/cli/src/commands/experimental/compute/logs/SIDE_EFFECTS.md

@@ -49,9 +49,9 @@ Compute private-alpha allow-list — an unenrolled project answers 404.
 ### The query
 
 SQL in **ClickHouse dialect** against the project's unified `logs` table, filtered
-on `log_attributes['worker']` and `log_attributes['source']`. It does **not** filter
-the top-level `source` column: compute rows carry an empty string there, because the
-Compute Logflare source is not enrolled as a category in the generic logs path.
+on `log_attributes['worker']` and the top-level `subservice` column. It does **not**
+filter the top-level `source` column: compute rows carry an empty string there, because
+the Compute Logflare source is not enrolled as a category in the generic logs path.
 
 ### The window
 

apps/cli/src/commands/experimental/compute/logs/logs.integration.test.ts

@@ -184,7 +184,7 @@ describe("compute logs", () => {
     }).pipe(Effect.scoped, Effect.provide(BunServices.layer)),
   );
 
-  it.live("filters on log_attributes, never the empty source column", () =>
+  it.live("filters on the subservice column, never the empty source column", () =>
     Effect.gen(function* () {
       const repo = yield* project();
       const { layer, http } = setupCompute({
@@ -197,7 +197,8 @@ describe("compute logs", () => {
 
         const sql = sentQuery(http.requests[0]!).sql ?? "";
         expect(sql).toContain("log_attributes['worker'] = 'api'");
-        expect(sql).toContain("log_attributes['source'] in (");
+        expect(sql).toContain("and subservice in (");
+        expect(sql).not.toContain("log_attributes['source']");
         expect(sql).not.toMatch(/where source =/);
       }).pipe(Effect.provide(layer));
     }).pipe(Effect.scoped, Effect.provide(BunServices.layer)),

apps/cli/src/shared/compute/compute-logs.sql.ts

@@ -44,8 +44,14 @@ export const ALL_COMPUTE_LOG_STREAMS: ReadonlyArray<string> = Object.values(COMP
  */
 const COMPUTE_LOG_NAME_ATTRIBUTE = "worker";
 
-/** Which key carries the stream name. See {@link computeLogsQuery} for why. */
-const COMPUTE_LOG_STREAM_ATTRIBUTE = "source";
+/**
+ * Which top-level column carries the stream name.
+ *
+ * The writer publishes it beside `project` rather than inside `metadata`, so it is a column
+ * here rather than a `log_attributes` key. `metadata.source` carries the same value for
+ * readers that have not moved across; this is the one to filter on.
+ */
+const COMPUTE_LOG_STREAM_COLUMN = "subservice";
 
 /**
  * The server clamps a span of more than 24 hours, so the default window sits just under the
@@ -127,13 +133,11 @@ function quote(value: string): string {
 }
 
 /**
- * The logs query for one compute. Two things about the projection are load-bearing: the filter
- * is `log_attributes`, not the `source` column, since compute rows carry an empty top-level
- * `source` (the stream survives only in `log_attributes['source']`); and the `in (...)` list
- * is a tenancy guard, not a convenience — with `source` empty, it's the only thing keeping a
- * non-compute row with a `worker` attribute out of the result. `toUnixTimestamp64Milli` rather
- * than a formatter, since ClickHouse's `%M` is the month name and bare `toString(timestamp)`
- * has no zone.
+ * The logs query for one compute. Two things about the projection are load-bearing: the
+ * `in (...)` list is a tenancy guard, not a convenience — compute rows carry an empty
+ * top-level `source`, so the stream list is the only thing keeping a non-compute row with a
+ * `worker` attribute out of the result; and `toUnixTimestamp64Milli` rather than a formatter,
+ * since ClickHouse's `%M` is the month name and bare `toString(timestamp)` has no zone.
  */
 export function computeLogsQuery(options: {
   readonly name: string;
@@ -144,12 +148,12 @@ export function computeLogsQuery(options: {
   return (
     `select id, ` +
     `toUnixTimestamp64Milli(timestamp) as ts_ms, ` +
-    `log_attributes['${COMPUTE_LOG_STREAM_ATTRIBUTE}'] as stream, ` +
+    `${COMPUTE_LOG_STREAM_COLUMN} as stream, ` +
     `event_message, ` +
     `log_attributes ` +
     `from logs ` +
     `where log_attributes['${COMPUTE_LOG_NAME_ATTRIBUTE}'] = ${quote(options.name)} ` +
-    `and log_attributes['${COMPUTE_LOG_STREAM_ATTRIBUTE}'] in (${streams}) ` +
+    `and ${COMPUTE_LOG_STREAM_COLUMN} in (${streams}) ` +
     `order by timestamp desc ` +
     `limit ${options.tail}`
   );

apps/cli/src/shared/compute/compute-logs.sql.unit.test.ts

@@ -12,11 +12,13 @@ import {
 } from "./compute-logs.sql.ts";
 
 describe("computeLogsQuery", () => {
-  it("filters on log_attributes, never the source column", () => {
+  it("filters on the subservice column, never the source column", () => {
     const sql = computeLogsQuery({ name: "api", streams: ALL_COMPUTE_LOG_STREAMS, tail: 100 });
 
     expect(sql).toContain("log_attributes['worker'] = 'api'");
-    expect(sql).toContain("log_attributes['source'] in (");
+    expect(sql).toContain("subservice as stream");
+    expect(sql).toContain("and subservice in (");
+    expect(sql).not.toContain("log_attributes['source']");
     expect(sql).not.toMatch(/(?:^|\s)where source =/);
     expect(sql).not.toMatch(/(?:^|\s)and source =/);
   });