diff --git a/.changeset/schema-encode-slow-metrics.md b/.changeset/schema-encode-slow-metrics.md new file mode 100644 index 000000000..da4f0eb03 --- /dev/null +++ b/.changeset/schema-encode-slow-metrics.md @@ -0,0 +1,9 @@ +--- +"effect-app": patch +--- + +Improve repository schema encode/decode telemetry for event-loop tail analysis. + +- widen `app.schema.{encode,decode}.duration` histogram buckets into multi-second stalls +- count `app.schema.slow` (duration ≥ 100ms) with `app.entity` / operation attributes for alertable rates +- annotate spans with `app.schema.slow` and, on encode, `app.entity.state` from the first item's `_tag` diff --git a/packages/effect-app/src/Model/Repository/internal/internal.ts b/packages/effect-app/src/Model/Repository/internal/internal.ts index c53f3eadc..33db662b3 100644 --- a/packages/effect-app/src/Model/Repository/internal/internal.ts +++ b/packages/effect-app/src/Model/Repository/internal/internal.ts @@ -34,18 +34,53 @@ import { ValidationError, ValidationResult } from "../validation.ts" const dedupe = Array.dedupeWith(Equivalence.String) -const schemaDurationBoundaries = [0.1, 0.5, 1, 2, 5, 10, 25, 50, 100, 250, 500, 1_000] +// ms buckets: dense under 100ms (common path), then mid-tail and multi-second stalls. +// Rare 0.5–1s+ encodes (fat aggregates) must not collapse into a single overflow bin. +const schemaDurationBoundaries = [ + 0.1, + 0.5, + 1, + 2, + 5, + 10, + 25, + 50, + 100, + 150, + 200, + 300, + 500, + 750, + 1_000, + 1_500, + 2_000, + 5_000 +] +// Event-loop-relevant stall floor: encodes/decodes above this are rare but block Node. +const SCHEMA_SLOW_MS = 100 const schemaDecodeDuration = Metric.histogram("app.schema.decode.duration", { boundaries: schemaDurationBoundaries }) const schemaEncodeDuration = Metric.histogram("app.schema.encode.duration", { boundaries: schemaDurationBoundaries }) const schemaItemCount = Metric.histogram("app.schema.item_count", { boundaries: [0, 1, 2, 5, 10, 25, 50, 100, 250, 500, 1_000, 5_000] }) +const schemaSlow = Metric.counter("app.schema.slow", { + description: `Repository schema encode/decode slower than ${SCHEMA_SLOW_MS}ms`, + incremental: true +}) + +const entityStateFromItems = (items: readonly unknown[]): string | undefined => { + const first = items[0] + if (first === null || typeof first !== "object" || !("_tag" in first)) return undefined + const tag = (first as { readonly _tag: unknown })._tag + return typeof tag === "string" ? tag : undefined +} const timeSchema = ( operation: "decode" | "encode", entity: string, queryMode: "aggregate" | "collect" | "project" | "transform" | undefined, - itemCount: number + itemCount: number, + entityState?: string ) => (self: Effect.Effect): Effect.Effect => Effect.clockWith((clock) => { @@ -53,15 +88,19 @@ const timeSchema = ( const attributes = { "app.entity": entity, "app.schema.operation": operation, - ...(queryMode !== undefined && { "app.query.mode": queryMode }) + ...(queryMode !== undefined && { "app.query.mode": queryMode }), + ...(entityState !== undefined && { "app.entity.state": entityState }) } return Effect.onExit(self, () => { const durationMs = Number(clock.currentTimeNanosUnsafe() - startedAt) / 1_000_000 + const slow = durationMs >= SCHEMA_SLOW_MS return Effect.all([ Effect.annotateCurrentSpan({ [`app.schema.${operation}.duration_ms`]: durationMs, "app.schema.item_count": itemCount, - ...(queryMode !== undefined && { "app.query.mode": queryMode }) + "app.schema.slow": slow, + ...(queryMode !== undefined && { "app.query.mode": queryMode }), + ...(entityState !== undefined && { "app.entity.state": entityState }) }), Metric.update( Metric.withAttributes( @@ -70,7 +109,10 @@ const timeSchema = ( ), durationMs ), - Metric.update(Metric.withAttributes(schemaItemCount, attributes), itemCount) + Metric.update(Metric.withAttributes(schemaItemCount, attributes), itemCount), + ...(slow + ? [Metric.update(Metric.withAttributes(schemaSlow, attributes), 1)] as const + : []) ], { discard: true }) }) }) @@ -140,7 +182,7 @@ export function makeRepoInternal< const encodeMany = (items: readonly T[]) => S.encodeEffect(S.Array(schema))(items).pipe( provideRctx, - timeSchema("encode", name, undefined, items.length) + timeSchema("encode", name, undefined, items.length, entityStateFromItems(items)) ) const decode = flow(S.decodeEffectConcurrently(schema), provideRctx) const decodeMany = flow( @@ -756,7 +798,9 @@ export function makeRepoInternal< save: (...xes: any[]) => Effect .flatMap( - encMany(xes).pipe(timeSchema("encode", name, undefined, xes.length)), + encMany(xes).pipe( + timeSchema("encode", name, undefined, xes.length, entityStateFromItems(xes)) + ), (_) => saveAllE(_) ) .pipe( diff --git a/packages/infra/test/repository-ext.test.ts b/packages/infra/test/repository-ext.test.ts index ea7958bff..0b2fba7b0 100644 --- a/packages/infra/test/repository-ext.test.ts +++ b/packages/infra/test/repository-ext.test.ts @@ -414,13 +414,43 @@ describe("repository ext save/remove batching", () => { const allSpan = spans.find((_) => _.name === "Repository.all") expect(saveSpan?.attributes.get("app.schema.encode.duration_ms")).toEqual(expect.any(Number)) expect(saveSpan?.attributes.get("app.schema.item_count")).toBe(1) + expect(saveSpan?.attributes.get("app.schema.slow")).toBe(false) expect(saveSpan?.attributes.get("db.operation.duration_ms")).toEqual(expect.any(Number)) expect(allSpan?.attributes.get("app.schema.decode.duration_ms")).toEqual(expect.any(Number)) expect(allSpan?.attributes.get("app.schema.item_count")).toBe(1) + expect(allSpan?.attributes.get("app.schema.slow")).toBe(false) expect(allSpan?.attributes.get("db.operation.duration_ms")).toEqual(expect.any(Number)) }) .pipe( setupRequestContextFromCurrent(), Effect.provide(TestStoreLive) )) + + it.effect("annotates entity state tag on encode when items are tagged", () => + Effect + .gen(function*() { + const spans: Tracer.NativeSpan[] = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + } + }) + + yield* Effect + .gen(function*() { + const repo = yield* makeRepo("TaggedTelemetry", union, {}) + yield* repo.save({ _tag: "A", id: "1" }) + }) + .pipe(Effect.provideService(Tracer.Tracer, tracer)) + + const saveSpan = spans.find((_) => _.name === "Repository.saveAndPublish") + expect(saveSpan?.attributes.get("app.entity.state")).toBe("A") + expect(saveSpan?.attributes.get("app.schema.slow")).toBe(false) + }) + .pipe( + setupRequestContextFromCurrent(), + Effect.provide(TestStoreLive) + )) })