fix(warehouse): repair capability probing and the service-map rollup

apps/api/src/routes/v1/session-replay.schema.test.ts

@@ -20,6 +20,12 @@ const baseRow = {
 	durationMs: null,
 	status: "ended",
 	userId: null,
+	// `""` is the never-identified state (see SessionReplayListItem), which is
+	// what an anonymous `userId: null` row carries.
+	userName: "",
+	userEmail: "",
+	groupId: "",
+	groupName: "",
 	visitorId: "b0f2b0c6-8a9f-4a1f-9a0e-2b4f9e2f1a77",
 	utmSource: "",
 	entryPath: "/",

apps/api/src/services/dashboards/ServiceMapRollupService.ts

@@ -1,5 +1,5 @@
 import { RoleName, UserId as UserIdSchema, type OrgId } from "@maple/domain/http"
-import { orgIngestKeys } from "@maple/db"
+import { orgClickHouseSettings, orgIngestKeys } from "@maple/db"
 import * as CH from "@maple/query-engine/ch"
 import { Clock, Cause, Context, Effect, Layer, Schema } from "effect"
 import type { TenantContext } from "@/services/auth/AuthService"
@@ -15,12 +15,24 @@ const HOUR_MS = 3_600_000
 /**
  * How many completed hours back the rollup re-checks on every run. Bounds the
  * per-run cost to a constant and lets a few missed cron ticks catch up. Hours
- * already present in `service_map_edges_hourly` are skipped, so re-checking is
- * cheap; an hour with genuinely zero cross-service calls is re-attempted each
- * run until it ages out of this window — also cheap (an empty join).
+ * already present in `service_map_edges_hourly` are skipped; an hour with
+ * genuinely zero cross-service calls is re-attempted each run until it ages out
+ * of this window.
+ *
+ * That re-attempt is NOT free — the join is empty only *after* it has scanned
+ * the hour's spans. Keep the org set gated (see `resolveActiveOrgs`): when the
+ * rollup fanned out across every org that ever held an ingest key, this window
+ * turned ~260 orgs into ~6 executions each per tick, of two queries, one of
+ * them a raw-`traces` self-join.
  */
 const LOOKBACK_HOURS = 6
 
+/**
+ * Discovery window for the active-org scan. Must be a SUPERSET of the per-org
+ * lookback so no org that produced spans in a candidate hour is skipped.
+ */
+const ACTIVE_DISCOVERY_HOURS = LOOKBACK_HOURS + 2
+
 /** Concurrency for per-org rollup processing. */
 const ORG_CONCURRENCY = 4
 
@@ -149,8 +161,26 @@ export class ServiceMapRollupService extends Context.Service<
 			// ingest can fail independently. Recompute resolution rows for sealed
 			// hours in the bounded lookback as an idempotent repair pass
 			// (ReplacingMergeTree deduplicates the same mapping key).
+			//
+			// Only for hours that have no resolution rows yet: the repair join
+			// reads raw `traces`, so re-running it for every sealed hour on every
+			// tick was the single most expensive thing this service did.
+			const resolvedRows = yield* warehouse.compiledQuery(
+				tenant,
+				CH.serviceMapResolutionsExistingHoursSQL({
+					orgId,
+					startTime: formatWarehouseDateTime(oldestHourMs),
+					endTime: formatWarehouseDateTime(currentHourMs),
+				}),
+				{ context: "serviceMapResolutionsExistingHours" },
+			)
+			const resolved = new Set(resolvedRows.map((row) => Number(row.hourTs)))
+
 			yield* Effect.forEach(
-				candidates.filter((hourMs) => existing.has(Math.floor(hourMs / 1000))),
+				candidates.filter(
+					(hourMs) =>
+						existing.has(Math.floor(hourMs / 1000)) && !resolved.has(Math.floor(hourMs / 1000)),
+				),
 				(hourMs) =>
 					Effect.gen(function* () {
 						const resolutionsRows = yield* warehouse.compiledQuery(
@@ -182,17 +212,91 @@ export class ServiceMapRollupService extends Context.Service<
 			}
 		})
 
+		/**
+		 * Which orgs actually produced spans in the discovery window.
+		 *
+		 * One cross-org scan of `traces_aggregates_hourly` replaces per-org
+		 * guessing — the same gating the error-issue and anomaly ticks already
+		 * use (`packages/query-engine/src/ch/queries/activity.ts`). Idle orgs
+		 * have no spans to join, so every hour they are handed costs two
+		 * warehouse scans to prove a join is empty.
+		 *
+		 * BYO-ClickHouse orgs are invisible to this managed-workspace scan and
+		 * are always processed.
+		 *
+		 * Fails OPEN, unlike the anomaly tick: a skipped rollup hour is a
+		 * permanent gap in the service map once it ages past `LOOKBACK_HOURS`,
+		 * whereas a skipped anomaly evaluation is only a late alert. A discovery
+		 * outage costs one hour of the old fan-out; failing closed could cost
+		 * the edges outright.
+		 */
+		const resolveActiveOrgs = Effect.fn("ServiceMapRollupService.resolveActiveOrgs")(function* (
+			knownOrgs: ReadonlyArray<OrgId>,
+		) {
+			if (knownOrgs.length === 0) return undefined
+
+			const nowMs = yield* Clock.currentTimeMillis
+			const startTime = formatWarehouseDateTime(nowMs - ACTIVE_DISCOVERY_HOURS * HOUR_MS)
+			const byoRows = yield* database
+				.execute((db) =>
+					db.selectDistinct({ orgId: orgClickHouseSettings.orgId }).from(orgClickHouseSettings),
+				)
+				.pipe(Effect.orElseSucceed(() => [] as ReadonlyArray<{ orgId: string }>))
+
+			return yield* warehouse
+				.crossOrgQuery(
+					systemTenant(knownOrgs[0]!),
+					CH.compile(CH.activeOrgsByTracesQuery(), { startTime }),
+					{
+						profile: "discovery",
+						context: "serviceMapRollupActiveOrgs",
+						justification:
+							"enumerate orgs with recent span aggregates so the hourly service-map rollup skips idle orgs",
+					},
+				)
+				.pipe(
+					Effect.map((rows) => {
+						const active = new Set<string>(byoRows.map((row) => row.orgId))
+						for (const row of rows) {
+							const orgId = String((row as { orgId?: unknown }).orgId ?? "")
+							if (orgId) active.add(orgId)
+						}
+						return active as ReadonlySet<string>
+					}),
+					Effect.catchCause((cause) =>
+						Cause.hasInterruptsOnly(cause)
+							? Effect.interrupt
+							: Effect.as(
+									Effect.logWarning(
+										"Service map rollup active-org discovery failed; processing every known org",
+									).pipe(Effect.annotateLogs({ error: Cause.pretty(cause) })),
+									undefined,
+								),
+					),
+				)
+		})
+
 		const runRollupTick: ServiceMapRollupServiceShape["runRollupTick"] = Effect.fn(
 			"ServiceMapRollupService.runRollupTick",
 		)(function* () {
 			const orgRows = yield* database.execute((db) =>
 				db.selectDistinct({ orgId: orgIngestKeys.orgId }).from(orgIngestKeys),
 			)
 
+			const knownOrgs = orgRows.map((row) => row.orgId as OrgId)
+			const active = yield* resolveActiveOrgs(knownOrgs)
+			const targetOrgs =
+				active === undefined ? knownOrgs : knownOrgs.filter((orgId) => active.has(orgId))
+			yield* Effect.annotateCurrentSpan({
+				knownOrgs: knownOrgs.length,
+				targetOrgs: targetOrgs.length,
+				activeOrgDiscovery: active === undefined ? "failed" : "ok",
+			})
+
 			const results = yield* Effect.forEach(
-				orgRows,
-				(row) =>
-					processOrg(row.orgId as OrgId).pipe(
+				targetOrgs,
+				(orgId) =>
+					processOrg(orgId).pipe(
 						// Interrupts (isolate teardown) are NOT failures — re-raise them
 						// so the tick cancels promptly instead of logging phantom
 						// per-org failures.
@@ -202,7 +306,7 @@ export class ServiceMapRollupService extends Context.Service<
 								: Effect.as(
 										Effect.logError("Service map rollup failed for org").pipe(
 											Effect.annotateLogs({
-												orgId: row.orgId,
+												orgId,
 												error: Cause.pretty(cause),
 											}),
 										),
@@ -221,7 +325,7 @@ export class ServiceMapRollupService extends Context.Service<
 			)
 
 			return {
-				orgsProcessed: orgRows.length,
+				orgsProcessed: targetOrgs.length,
 				hoursRolledUp: results.reduce((sum, r) => sum + r.hoursRolledUp, 0),
 				edgesWritten: results.reduce((sum, r) => sum + r.edgesWritten, 0),
 				resolutionsWritten: results.reduce((sum, r) => sum + r.resolutionsWritten, 0),

apps/cli/src/server/schema/local-inserts.json

@@ -1,5 +1,5 @@
 {
-	"projectRevision": "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06",
+	"projectRevision": "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa",
 	"orgPlaceholder": "__ORG__",
 	"datasources": {
 		"traces": {

apps/cli/src/server/schema/local-schema.sql

@@ -1,6 +1,6 @@
 -- This file is generated by scripts/generate-clickhouse-schema-sql.ts
 -- Do not edit manually. Run `bun run clickhouse:schema` to regenerate.
--- projectRevision: 6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06
+-- projectRevision: 12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa
 -- localSchemaVersion: 1
 
 CREATE TABLE IF NOT EXISTS alert_checks (

apps/ingest/src/clickhouse_insert_mappings.rs

@@ -1,7 +1,7 @@
 // This file is generated by scripts/generate-clickhouse-insert-mappings.ts
 // Do not edit manually.
 
-pub const PROJECT_REVISION: &str = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06";
+pub const PROJECT_REVISION: &str = "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa";
 // Gate for BYO-ClickHouse ingest readiness — the migration version, NOT the
 // Tinybird-coupled PROJECT_REVISION. Compared against
 // org_clickhouse_settings.schema_version. See @maple/domain/clickhouse

packages/domain/src/generated/clickhouse-schema.ts

@@ -1,7 +1,7 @@
 // This file is generated by scripts/generate-clickhouse-schema.ts
 // Do not edit manually.
 
-export const projectRevision = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06" as const
+export const projectRevision = "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa" as const
 
 export const latestSnapshotStatements: ReadonlyArray<string> = [
 	"CREATE TABLE IF NOT EXISTS alert_checks (\n    OrgId LowCardinality(String),\n    RuleId String,\n    GroupKey String,\n    Timestamp DateTime64(3),\n    Status LowCardinality(String),\n    SignalType LowCardinality(String),\n    Comparator LowCardinality(String),\n    Threshold Float64,\n    ObservedValue Nullable(Float64),\n    SampleCount UInt32,\n    WindowMinutes UInt16,\n    WindowStart DateTime64(3),\n    WindowEnd DateTime64(3),\n    ConsecutiveBreaches UInt16,\n    ConsecutiveHealthy UInt16,\n    IncidentId Nullable(String),\n    IncidentTransition LowCardinality(String),\n    EvaluationDurationMs UInt32,\n    ErrorMessage Nullable(String),\n    ErrorCategory LowCardinality(String)\n)\nENGINE = MergeTree\nPARTITION BY toDate(Timestamp)\nORDER BY (OrgId, RuleId, GroupKey, Timestamp)\nTTL toDate(Timestamp) + INTERVAL 365 DAY",

packages/domain/src/generated/tinybird-project-manifest.ts

@@ -1,7 +1,7 @@
 // This file is generated by scripts/generate-tinybird-project-manifest.ts
 // Do not edit manually.
 
-export const projectRevision = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06" as const
+export const projectRevision = "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa" as const
 
 export const datasources = [
 	{
@@ -72,7 +72,7 @@ export const datasources = [
 	{
 		name: "service_address_resolutions_hourly",
 		content:
-			'DESCRIPTION >\n    Resolved (sourceService, parent.server.address) → resolved targetService facts emitted by the ServiceMapRollupService rollup. Used to anti-join internal-service overlap out of the external-edges query.\n\nSCHEMA >\n    OrgId LowCardinality(String),\n    Hour DateTime,\n    SourceService LowCardinality(String),\n    ParentServerAddress String,\n    ResolvedTargetService LowCardinality(String),\n    DeploymentEnv LowCardinality(String)\n\nENGINE "ReplacingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, ParentServerAddress, ResolvedTargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"',
+			'DESCRIPTION >\n    Resolved (sourceService, parent.server.address) → resolved targetService facts emitted by the ServiceMapRollupService rollup. Used to anti-join internal-service overlap out of the external-edges query.\n\nSCHEMA >\n    OrgId LowCardinality(String) `json:$.OrgId`,\n    Hour DateTime `json:$.Hour`,\n    SourceService LowCardinality(String) `json:$.SourceService`,\n    ParentServerAddress String `json:$.ParentServerAddress`,\n    ResolvedTargetService LowCardinality(String) `json:$.ResolvedTargetService`,\n    DeploymentEnv LowCardinality(String) `json:$.DeploymentEnv`\n\nENGINE "ReplacingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, ParentServerAddress, ResolvedTargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"',
 	},
 	{
 		name: "service_external_edges_hourly",
@@ -97,7 +97,7 @@ export const datasources = [
 	{
 		name: "service_map_edges_hourly",
 		content:
-			'DESCRIPTION >\n    Pre-aggregated hourly service-to-service edges for the service map. Uses AggregatingMergeTree for incremental aggregation. Populated by the scheduled ServiceMapRollupService rollup (one write per completed hour).\n\nSCHEMA >\n    OrgId LowCardinality(String),\n    Hour DateTime,\n    SourceService LowCardinality(String),\n    TargetService String,\n    DeploymentEnv LowCardinality(String),\n    CallCount SimpleAggregateFunction(sum, UInt64),\n    ErrorCount SimpleAggregateFunction(sum, UInt64),\n    DurationSumMs SimpleAggregateFunction(sum, Float64),\n    MaxDurationMs SimpleAggregateFunction(max, Float64),\n    SampledSpanCount SimpleAggregateFunction(sum, UInt64),\n    UnsampledSpanCount SimpleAggregateFunction(sum, UInt64),\n    SampleRateSum SimpleAggregateFunction(sum, Float64)\n\nENGINE "AggregatingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, TargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"',
+			'DESCRIPTION >\n    Pre-aggregated hourly service-to-service edges for the service map. Uses AggregatingMergeTree for incremental aggregation. Populated by the scheduled ServiceMapRollupService rollup (one write per completed hour).\n\nSCHEMA >\n    OrgId LowCardinality(String) `json:$.OrgId`,\n    Hour DateTime `json:$.Hour`,\n    SourceService LowCardinality(String) `json:$.SourceService`,\n    TargetService String `json:$.TargetService`,\n    DeploymentEnv LowCardinality(String) `json:$.DeploymentEnv`,\n    CallCount SimpleAggregateFunction(sum, UInt64) `json:$.CallCount`,\n    ErrorCount SimpleAggregateFunction(sum, UInt64) `json:$.ErrorCount`,\n    DurationSumMs SimpleAggregateFunction(sum, Float64) `json:$.DurationSumMs`,\n    MaxDurationMs SimpleAggregateFunction(max, Float64) `json:$.MaxDurationMs`,\n    SampledSpanCount SimpleAggregateFunction(sum, UInt64) `json:$.SampledSpanCount`,\n    UnsampledSpanCount SimpleAggregateFunction(sum, UInt64) `json:$.UnsampledSpanCount`,\n    SampleRateSum SimpleAggregateFunction(sum, Float64) `json:$.SampleRateSum`\n\nENGINE "AggregatingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, TargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"',
 	},
 	{
 		name: "service_map_spans",

packages/domain/src/tinybird/datasources.ts

@@ -413,7 +413,12 @@ export type ServiceMapChildrenRow = InferRow<typeof serviceMapChildren>
 export const serviceMapEdgesHourly = defineDatasource("service_map_edges_hourly", {
 	description:
 		"Pre-aggregated hourly service-to-service edges for the service map. Uses AggregatingMergeTree for incremental aggregation. Populated by the scheduled ServiceMapRollupService rollup (one write per completed hour).",
-	jsonPaths: false,
+	// jsonPaths enabled: this is ingested directly via POST /v0/events from
+	// ServiceMapRollupService, not by a materialized view. Declaring
+	// `jsonPaths: false` made Tinybird reject every write with "Data Source
+	// needs to have JSONPaths defined", so the table stopped filling — and
+	// because an hour is sealed only once its edge rows land, the rollup then
+	// re-ran all six lookback hours for every org on every tick, forever.
 	schema: {
 		OrgId: t.string().lowCardinality(),
 		Hour: t.dateTime(),
@@ -609,7 +614,8 @@ export type ServiceExternalEdgesHourlyRow = InferRow<typeof serviceExternalEdges
 export const serviceAddressResolutionsHourly = defineDatasource("service_address_resolutions_hourly", {
 	description:
 		"Resolved (sourceService, parent.server.address) → resolved targetService facts emitted by the ServiceMapRollupService rollup. Used to anti-join internal-service overlap out of the external-edges query.",
-	jsonPaths: false,
+	// jsonPaths enabled — same reason as `service_map_edges_hourly`: the rollup
+	// writes these rows directly via POST /v0/events, which requires them.
 	schema: {
 		OrgId: t.string().lowCardinality(),
 		Hour: t.dateTime(),

packages/query-engine/src/__sql_baseline__/catalog.sql

@@ -554,6 +554,16 @@ SELECT
         GROUP BY OrgId, Hour, SourceService, TargetService, DeploymentEnv
         FORMAT JSON
 
+-- builder:service-map-rollup:serviceMapResolutionsExistingHoursSQL:default  [ea1bee51]
+SELECT
+          toUnixTimestamp(Hour) AS hourTs
+        FROM service_address_resolutions_hourly
+        WHERE OrgId = 'org_sql_catalog'
+          AND Hour >= '2026-01-01 10:30:00'
+          AND Hour < '2026-01-03 14:15:00'
+        GROUP BY hourTs
+        FORMAT JSON
+
 -- builder:service-map:serviceDependenciesForServiceQuery:default  [32f3ee6f]
 SELECT
           sourceService AS sourceService,
@@ -4410,7 +4420,7 @@ SELECT
         LIMIT 5000
         FORMAT JSON
 
--- pipe:span_search:default:baseline  [d4e4a298]
+-- pipe:span_search:default:baseline  [bfce1ab3]
 SELECT
           TraceId AS traceId,
           SpanId AS spanId,
@@ -4427,11 +4437,20 @@ SELECT
           AND Timestamp >= '2026-01-01 10:30:00'
           AND Timestamp <= '2026-01-03 14:15:00'
           AND StatusCode != 'Error'
+          AND Timestamp >= (SELECT min(ts) FROM (SELECT
+          Timestamp AS ts
+        FROM traces
+        WHERE OrgId = 'org_sql_catalog'
+          AND Timestamp >= '2026-01-01 10:30:00'
+          AND Timestamp <= '2026-01-03 14:15:00'
+          AND StatusCode != 'Error'
+        ORDER BY ts DESC
+        LIMIT 20))
         ORDER BY timestamp DESC
         LIMIT 20
         FORMAT JSON
 
--- pipe:span_search:default:bloom  [d4e4a298]
+-- pipe:span_search:default:bloom  [bfce1ab3]
 SELECT
           TraceId AS traceId,
           SpanId AS spanId,
@@ -4448,11 +4467,20 @@ SELECT
           AND Timestamp >= '2026-01-01 10:30:00'
           AND Timestamp <= '2026-01-03 14:15:00'
           AND StatusCode != 'Error'
+          AND Timestamp >= (SELECT min(ts) FROM (SELECT
+          Timestamp AS ts
+        FROM traces
+        WHERE OrgId = 'org_sql_catalog'
+          AND Timestamp >= '2026-01-01 10:30:00'
+          AND Timestamp <= '2026-01-03 14:15:00'
+          AND StatusCode != 'Error'
+        ORDER BY ts DESC
+        LIMIT 20))
         ORDER BY timestamp DESC
         LIMIT 20
         FORMAT JSON
 
--- pipe:span_search:default:text  [d4e4a298]
+-- pipe:span_search:default:text  [bfce1ab3]
 SELECT
           TraceId AS traceId,
           SpanId AS spanId,
@@ -4469,6 +4497,15 @@ SELECT
           AND Timestamp >= '2026-01-01 10:30:00'
           AND Timestamp <= '2026-01-03 14:15:00'
           AND StatusCode != 'Error'
+          AND Timestamp >= (SELECT min(ts) FROM (SELECT
+          Timestamp AS ts
+        FROM traces
+        WHERE OrgId = 'org_sql_catalog'
+          AND Timestamp >= '2026-01-01 10:30:00'
+          AND Timestamp <= '2026-01-03 14:15:00'
+          AND StatusCode != 'Error'
+        ORDER BY ts DESC
+        LIMIT 20))
         ORDER BY timestamp DESC
         LIMIT 20
         FORMAT JSON

packages/query-engine/src/ch/builder-fixtures.ts

@@ -473,6 +473,12 @@ export const builderFixtures: ReadonlyArray<BuilderFixture> = [
 		label: "default",
 		compile: () => CH.serviceMapEdgesExistingHoursSQL(window),
 	},
+	{
+		module: "service-map-rollup",
+		name: "serviceMapResolutionsExistingHoursSQL",
+		label: "default",
+		compile: () => CH.serviceMapResolutionsExistingHoursSQL(window),
+	},
 	{
 		// Org-wide: hourly MV branch UNION ALL two partial-hour live joins.
 		module: "service-map",

packages/query-engine/src/ch/index.ts

@@ -293,6 +293,7 @@ export {
 export {
 	serviceMapEdgesRollupSQL,
 	serviceMapEdgesExistingHoursSQL,
+	serviceMapResolutionsExistingHoursSQL,
 	serviceMapResolutionsRollupSQL,
 	type ServiceMapEdgesRollupParams,
 	type ServiceMapEdgesHourlyOutput,