| author |  Theo Browne <me@t3.gg> 2026-09-26 01:34:14-07:00 |
| committer |  GitHub <noreply@github.com> 2026-09-26 01:34:14-07:00 |
| log | 99efaeab58297c7e346ee91a4f28044faed03c59 |
| tree | 314058bc2126cc844e1c0056a9f17760f5c9cd46 |
| parent | daafcc4a972b25eb762962cb72575b1a2e583d61 |
| signature | Signed by PGP key B5690EEEBB952194 |
perf(observability): stop writing empty spans on spawns, projected events, and idle polls (#13756)
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>7 files changed, 102 insertions(+), 15 deletions(-)
apps
server
src
orchestration
- processRunner.ts
packages
shared
src
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+61
| ... | @@ -21,6 +21,7 @@ import * as FileSystem from "effect/FileSystem"; | ... | @@ -21,6 +21,7 @@ import * as FileSystem from "effect/FileSystem"; |
| 21 | import * as Layer from "effect/Layer"; | 21 | import * as Layer from "effect/Layer"; |
| 22 | import * as Path from "effect/Path"; | 22 | import * as Path from "effect/Path"; |
| 23 | import * as Schema from "effect/Schema"; | 23 | import * as Schema from "effect/Schema"; |
| 24 | import * as Tracer from "effect/Tracer"; | ||
| 24 | import * as SqlClient from "effect/unstable/sql/SqlClient"; | 25 | import * as SqlClient from "effect/unstable/sql/SqlClient"; |
| 25 | 26 | ||
| 26 | import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts"; | 27 | import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts"; |
| ... | @@ -114,6 +115,66 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-curs | ... | @@ -114,6 +115,66 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-curs |
| 114 | }, | 115 | }, |
| 115 | ); | 116 | ); |
| 116 | 117 | ||
| 118 | it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-cleanup-span-")))( | ||
| 119 | "OrchestrationProjectionPipeline attachment cleanup span", | ||
| 120 | (it) => { | ||
| 121 | it.effect("runs attachment cleanup only for events that remove attachments", () => | ||
| 122 | Effect.gen(function* () { | ||
| 123 | const projectionPipeline = yield* OrchestrationProjectionPipeline; | ||
| 124 | const eventStore = yield* OrchestrationEventStore; | ||
| 125 | let cleanupSpans = 0; | ||
| 126 | const tracer = Tracer.make({ | ||
| 127 | span: (options) => { | ||
| 128 | if (options.name === "applyAttachmentSideEffects") cleanupSpans += 1; | ||
| 129 | return new Tracer.NativeSpan(options); | ||
| 130 | }, | ||
| 131 | }); | ||
| 132 | const now = "2026-01-01T00:00:00.000Z"; | ||
| 133 | const projectId = ProjectId.make("project-cleanup-span"); | ||
| 134 | const threadId = ThreadId.make("thread-cleanup-span"); | ||
| 135 | |||
| 136 | const projectCreated = yield* eventStore.append({ | ||
| 137 | type: "project.created", | ||
| 138 | eventId: EventId.make("evt-cleanup-span-project"), | ||
| 139 | aggregateKind: "project", | ||
| 140 | aggregateId: projectId, | ||
| 141 | occurredAt: now, | ||
| 142 | commandId: CommandId.make("cmd-cleanup-span-project"), | ||
| 143 | causationEventId: null, | ||
| 144 | correlationId: null, | ||
| 145 | metadata: {}, | ||
| 146 | payload: { | ||
| 147 | projectId, | ||
| 148 | title: "Cleanup span project", | ||
| 149 | workspaceRoot: "/tmp/project-cleanup-span", | ||
| 150 | defaultModelSelection: null, | ||
| 151 | scripts: [], | ||
| 152 | createdAt: now, | ||
| 153 | updatedAt: now, | ||
| 154 | }, | ||
| 155 | }); | ||
| 156 | yield* projectionPipeline.projectEvent(projectCreated).pipe(Effect.withTracer(tracer)); | ||
| 157 | assert.strictEqual(cleanupSpans, 0); | ||
| 158 | |||
| 159 | const threadDeleted = yield* eventStore.append({ | ||
| 160 | type: "thread.deleted", | ||
| 161 | eventId: EventId.make("evt-cleanup-span-thread-delete"), | ||
| 162 | aggregateKind: "thread", | ||
| 163 | aggregateId: threadId, | ||
| 164 | occurredAt: now, | ||
| 165 | commandId: CommandId.make("cmd-cleanup-span-thread-delete"), | ||
| 166 | causationEventId: null, | ||
| 167 | correlationId: null, | ||
| 168 | metadata: {}, | ||
| 169 | payload: { threadId, deletedAt: now }, | ||
| 170 | }); | ||
| 171 | yield* projectionPipeline.projectEvent(threadDeleted).pipe(Effect.withTracer(tracer)); | ||
| 172 | assert.strictEqual(cleanupSpans, 1); | ||
| 173 | }), | ||
| 174 | ); | ||
| 175 | }, | ||
| 176 | ); | ||
| 177 | |||
| 117 | it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-import-shell-")))( | 178 | it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-import-shell-")))( |
| 118 | "imported thread shell projection", | 179 | "imported thread shell projection", |
| 119 | (it) => { | 180 | (it) => { |
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+7-8
| ... | @@ -1991,13 +1991,6 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti | ... | @@ -1991,13 +1991,6 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti |
| 1991 | 1991 | ||
| 1992 | const applyAttachmentSideEffects = Effect.fn("applyAttachmentSideEffects")( | 1992 | const applyAttachmentSideEffects = Effect.fn("applyAttachmentSideEffects")( |
| 1993 | function* (event: OrchestrationEvent, sideEffects: AttachmentSideEffects) { | 1993 | function* (event: OrchestrationEvent, sideEffects: AttachmentSideEffects) { |
| 1994 | if ( | ||
| 1995 | sideEffects.deletedThreadIds.size === 0 && | ||
| 1996 | sideEffects.prunedThreadRelativePaths.size === 0 | ||
| 1997 | ) { | ||
| 1998 | return; | ||
| 1999 | } | ||
| 2000 | |||
| 2001 | const deletedThreadIds = new Set<string>(); | 1994 | const deletedThreadIds = new Set<string>(); |
| 2002 | for (const threadId of sideEffects.deletedThreadIds) { | 1995 | for (const threadId of sideEffects.deletedThreadIds) { |
| 2003 | const recreatedLater = yield* eventStore.hasEventAfter({ | 1996 | const recreatedLater = yield* eventStore.hasEventAfter({ |
| ... | @@ -2113,9 +2106,15 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti | ... | @@ -2113,9 +2106,15 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti |
| 2113 | ); | 2106 | ); |
| 2114 | }), | 2107 | }), |
| 2115 | ); | 2108 | ); |
| 2109 | const hasCleanup = | ||
| 2110 | attachmentSideEffects.deletedThreadIds.size > 0 || | ||
| 2111 | attachmentSideEffects.prunedThreadRelativePaths.size > 0; | ||
| 2116 | // Return the cleanup effect so the caller runs it after the outer transaction commits. | 2112 | // Return the cleanup effect so the caller runs it after the outer transaction commits. |
| 2113 | // Most events have no cleanup, so they skip the call and write no cleanup span. | ||
| 2117 | // @effect-diagnostics-next-line returnEffectInGen:off | 2114 | // @effect-diagnostics-next-line returnEffectInGen:off |
| 2118 | return applyAttachmentSideEffects(event, attachmentSideEffects).pipe(Effect.asVoid); | 2115 | return hasCleanup |
| 2116 | ? applyAttachmentSideEffects(event, attachmentSideEffects).pipe(Effect.asVoid) | ||
| 2117 | : Effect.void; | ||
| 2119 | }, | 2118 | }, |
| 2120 | Effect.provideService(FileSystem.FileSystem, fileSystem), | 2119 | Effect.provideService(FileSystem.FileSystem, fileSystem), |
| 2121 | Effect.provideService(Path.Path, path), | 2120 | Effect.provideService(Path.Path, path), |
apps/server/src/preview/PortScanner.test.ts+21
| ... | @@ -18,6 +18,7 @@ import * as Layer from "effect/Layer"; | ... | @@ -18,6 +18,7 @@ import * as Layer from "effect/Layer"; |
| 18 | import * as PlatformError from "effect/PlatformError"; | 18 | import * as PlatformError from "effect/PlatformError"; |
| 19 | import * as Scope from "effect/Scope"; | 19 | import * as Scope from "effect/Scope"; |
| 20 | import * as TestClock from "effect/testing/TestClock"; | 20 | import * as TestClock from "effect/testing/TestClock"; |
| 21 | import * as Tracer from "effect/Tracer"; | ||
| 21 | import { expect } from "vite-plus/test"; | 22 | import { expect } from "vite-plus/test"; |
| 22 | import { FetchHttpClient } from "effect/unstable/http"; | 23 | import { FetchHttpClient } from "effect/unstable/http"; |
| 23 | 24 | ||
| ... | @@ -441,6 +442,26 @@ effectIt.effect("stops probing a subscriber's configured paths after its scope c | ... | @@ -441,6 +442,26 @@ effectIt.effect("stops probing a subscriber's configured paths after its scope c |
| 441 | }).pipe(Effect.scoped, Effect.provide(layer)); | 442 | }).pipe(Effect.scoped, Effect.provide(layer)); |
| 442 | }); | 443 | }); |
| 443 | 444 | ||
| 445 | effectIt.effect("writes no poll span while no client retains the scanner", () => { | ||
| 446 | let pollSpans = 0; | ||
| 447 | const tracer = Tracer.make({ | ||
| 448 | span: (options) => { | ||
| 449 | if (options.name === "PortDiscovery.pollTick") pollSpans += 1; | ||
| 450 | return new Tracer.NativeSpan(options); | ||
| 451 | }, | ||
| 452 | }); | ||
| 453 | const layer = makeProbeFailureLayer(processProbeFailure); | ||
| 454 | |||
| 455 | return Effect.gen(function* () { | ||
| 456 | const scanner = yield* PortScanner.PortDiscovery; | ||
| 457 | yield* TestClock.adjust(Duration.seconds(15)); | ||
| 458 | expect(pollSpans).toBe(0); | ||
| 459 | |||
| 460 | yield* scanner.retain; | ||
| 461 | expect(pollSpans).toBe(1); | ||
| 462 | }).pipe(Effect.scoped, Effect.provide(layer), Effect.withTracer(tracer)); | ||
| 463 | }); | ||
| 464 | |||
| 444 | effectIt.effect("uses the current configured fragment when readiness comes from cache", () => { | 465 | effectIt.effect("uses the current configured fragment when readiness comes from cache", () => { |
| 445 | const requests: string[] = []; | 466 | const requests: string[] = []; |
| 446 | const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { | 467 | const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { |
apps/server/src/preview/PortScanner.ts+6-4
| ... | @@ -550,7 +550,6 @@ export const make = Effect.gen(function* PortDiscoveryMake() { | ... | @@ -550,7 +550,6 @@ export const make = Effect.gen(function* PortDiscoveryMake() { |
| 550 | 550 | ||
| 551 | const pollTick = Effect.fn("PortDiscovery.pollTick")( | 551 | const pollTick = Effect.fn("PortDiscovery.pollTick")( |
| 552 | function* () { | 552 | function* () { |
| 553 | if ((yield* Ref.get(stateRef)).retainCount <= 0) return; | ||
| 554 | const configuredUrls = [ | 553 | const configuredUrls = [ |
| 555 | ...new Set( | 554 | ...new Set( |
| 556 | [...(yield* Ref.get(stateRef)).listeners.values()].flatMap( | 555 | [...(yield* Ref.get(stateRef)).listeners.values()].flatMap( |
| ... | @@ -579,9 +578,12 @@ export const make = Effect.gen(function* PortDiscoveryMake() { | ... | @@ -579,9 +578,12 @@ export const make = Effect.gen(function* PortDiscoveryMake() { |
| 579 | ), | 578 | ), |
| 580 | ); | 579 | ); |
| 581 | 580 | ||
| 582 | // Single layer-scoped polling fiber. Ticks are no-ops when no client is | 581 | // Single layer-scoped polling fiber. Ticks skip the scan and its span when no |
| 583 | // currently retained, so the cost is one Ref.get every POLL_INTERVAL. | 582 | // client is currently retained, so the cost is one Ref.get every POLL_INTERVAL. |
| 584 | yield* Effect.forkScoped(pollTick().pipe(Effect.repeat(Schedule.spaced(POLL_INTERVAL)))); | 583 | const pollIfRetained = Ref.get(stateRef).pipe( |
| 584 | Effect.flatMap((state) => (state.retainCount > 0 ? pollTick() : Effect.void)), | ||
| 585 | ); | ||
| 586 | yield* Effect.forkScoped(pollIfRetained.pipe(Effect.repeat(Schedule.spaced(POLL_INTERVAL)))); | ||
| 585 | 587 | ||
| 586 | const acquireRetention = Effect.fn("PortDiscovery.retain")(function* () { | 588 | const acquireRetention = Effect.fn("PortDiscovery.retain")(function* () { |
| 587 | const wasIdle = yield* Ref.modify(stateRef, (state) => [ | 589 | const wasIdle = yield* Ref.modify(stateRef, (state) => [ |
apps/server/src/processRunner.ts+2-1
| ... | @@ -171,7 +171,8 @@ export const isWindowsCommandNotFound = Effect.fn("processRunner.isWindowsComman | ... | @@ -171,7 +171,8 @@ export const isWindowsCommandNotFound = Effect.fn("processRunner.isWindowsComman |
| 171 | }, | 171 | }, |
| 172 | ); | 172 | ); |
| 173 | 173 | ||
| 174 | const collectText = Effect.fn("processRunner.collectText")(function* (input: { | 174 | // Untraced: no attributes, and its time is the runProcessCore span. Errors fail that span. |
| 175 | const collectText = Effect.fnUntraced(function* (input: { | ||
| 175 | readonly command: string; | 176 | readonly command: string; |
| 176 | readonly args: ReadonlyArray<string>; | 177 | readonly args: ReadonlyArray<string>; |
| 177 | readonly cwd?: string | undefined; | 178 | readonly cwd?: string | undefined; |
apps/server/src/vcs/GitVcsDriverCore.ts+3-1
| ... | @@ -549,7 +549,9 @@ function trace2ChildKey(record: Record<string, unknown>): string | null { | ... | @@ -549,7 +549,9 @@ function trace2ChildKey(record: Record<string, unknown>): string | null { |
| 549 | const Trace2Record = Schema.Record(Schema.String, Schema.Unknown); | 549 | const Trace2Record = Schema.Record(Schema.String, Schema.Unknown); |
| 550 | const decodeTrace2Record = decodeJsonResult(Trace2Record); | 550 | const decodeTrace2Record = decodeJsonResult(Trace2Record); |
| 551 | 551 | ||
| 552 | const createTrace2Monitor = Effect.fn("createTrace2Monitor")(function* ( | 552 | // Untraced because it runs on every git spawn and returns at once without hook |
| 553 | // callbacks. Its errors fail the runGitCommand span. | ||
| 554 | const createTrace2Monitor = Effect.fnUntraced(function* ( | ||
| 553 | input: Pick<GitVcsDriver.ExecuteGitInput, "operation" | "cwd" | "args">, | 555 | input: Pick<GitVcsDriver.ExecuteGitInput, "operation" | "cwd" | "args">, |
| 554 | progress: GitVcsDriver.ExecuteGitProgress | undefined, | 556 | progress: GitVcsDriver.ExecuteGitProgress | undefined, |
| 555 | ): Effect.fn.Return< | 557 | ): Effect.fn.Return< |
packages/shared/src/shell.ts+2-1
| ... | @@ -681,7 +681,8 @@ export const resolveCommandPath = Effect.fn("shell.resolveCommandPath")(function | ... | @@ -681,7 +681,8 @@ export const resolveCommandPath = Effect.fn("shell.resolveCommandPath")(function |
| 681 | }); | 681 | }); |
| 682 | }); | 682 | }); |
| 683 | 683 | ||
| 684 | export const resolveSpawnCommand = Effect.fn("shell.resolveSpawnCommand")(function* ( | 684 | // Untraced because it runs before most spawns and returns at once off Windows. |
| 685 | export const resolveSpawnCommand = Effect.fnUntraced(function* ( | ||
| 685 | command: string, | 686 | command: string, |
| 686 | args: ReadonlyArray<string>, | 687 | args: ReadonlyArray<string>, |
| 687 | options: CommandAvailabilityOptions = {}, | 688 | options: CommandAvailabilityOptions = {}, |