diff --git a/apps/alerting/src/worker.ts b/apps/alerting/src/worker.ts index 03cecea26..6d81cee9c 100644 --- a/apps/alerting/src/worker.ts +++ b/apps/alerting/src/worker.ts @@ -1,5 +1,5 @@ import { - ANTICIPATED_ERROR_TAGS, + ANTICIPATED_ERROR_IDENTIFIERS, AlertsService, AnomalyDetectionService, BucketCacheService, @@ -38,7 +38,7 @@ const telemetry = MapleCloudflareSDK.make({ serviceName: "alerting", serviceNamespace: "backend", repositoryUrl: "https://github.com/Makisuo/maple", - anticipatedErrorTags: [...ANTICIPATED_ERROR_TAGS], + anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS], }) const buildLayer = (_env: Record) => { diff --git a/apps/api/src/alerting.ts b/apps/api/src/alerting.ts index 3f7fe7102..dc89cd89b 100644 --- a/apps/api/src/alerting.ts +++ b/apps/api/src/alerting.ts @@ -1,4 +1,4 @@ -export { ANTICIPATED_ERROR_TAGS } from "@maple/domain/anticipated-errors" +export { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" export { AlertRuntime, AlertsService } from "./services/AlertsService" export { AnomalyDetectionService } from "./services/AnomalyDetectionService" export { BucketCacheService } from "@maple/query-engine/caching" diff --git a/apps/api/src/internal-rpc.test.ts b/apps/api/src/internal-rpc.test.ts index 180cf49bf..37d894f71 100644 --- a/apps/api/src/internal-rpc.test.ts +++ b/apps/api/src/internal-rpc.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest" +import { describe, expect, it } from "@effect/vitest" import { Effect } from "effect" import type { InternalRpcInvalidInputError } from "@maple/domain/internal-rpc" import { callMcpToolRpc, submitDiagnosisRpc } from "./internal-rpc" @@ -31,60 +31,60 @@ const unusedInvestigationService: InvestigationServiceShape = { } describe("internal RPC boundary", () => { - it("rejects invalid org IDs before MCP dispatch", async () => { - const error = await Effect.runPromise( - Effect.flip( + it.effect("rejects invalid org IDs before MCP dispatch", () => + Effect.gen(function* () { + const error = yield* Effect.flip( callMcpToolRpc({ orgId: " ", name: "inspect_trace", input: {} }) as Effect.Effect< never, InternalRpcInvalidInputError, never >, - ), - ) - expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError") - expect(error.method).toBe("callMcpTool") - }) + ) + expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError") + expect(error.method).toBe("callMcpTool") + }), + ) - it("rejects invalid investigation IDs and model-produced reports", async () => { - for (const input of [ - { orgId: "org_1", investigationId: "not-a-uuid", report }, - { orgId: "org_1", investigationId, report: { summary: "incomplete" } }, - ]) { - const error = await Effect.runPromise( - Effect.flip( + it.effect("rejects invalid investigation IDs and model-produced reports", () => + Effect.gen(function* () { + for (const input of [ + { orgId: "org_1", investigationId: "not-a-uuid", report }, + { orgId: "org_1", investigationId, report: { summary: "incomplete" } }, + ]) { + const error = yield* Effect.flip( submitDiagnosisRpc(input).pipe( Effect.provideService(InvestigationService, unusedInvestigationService), ), - ), - ) - expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError") - if (error._tag !== "@maple/internal-rpc/InvalidInputError") { - throw new Error(`Expected invalid input, received ${error._tag}`) + ) + expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError") + if (error._tag !== "@maple/internal-rpc/InvalidInputError") { + throw new Error(`Expected invalid input, received ${error._tag}`) + } + expect(error.method).toBe("submitDiagnosis") } - expect(error.method).toBe("submitDiagnosis") - } - }) + }), + ) - it("submits a decoded diagnosis to the org-scoped service", async () => { - const calls: Array<{ orgId: string; investigationId: string; summary: string }> = [] - const expected = { id: investigationId, status: "diagnosed" } as never - const service: InvestigationServiceShape = { - ...unusedInvestigationService, - submitDiagnosis: (orgId, id, request) => - Effect.sync(() => { - calls.push({ orgId, investigationId: id, summary: request.report.summary }) - return expected - }), - } + it.effect("submits a decoded diagnosis to the org-scoped service", () => + Effect.gen(function* () { + const calls: Array<{ orgId: string; investigationId: string; summary: string }> = [] + const expected = { id: investigationId, status: "diagnosed" } as never + const service: InvestigationServiceShape = { + ...unusedInvestigationService, + submitDiagnosis: (orgId, id, request) => + Effect.sync(() => { + calls.push({ orgId, investigationId: id, summary: request.report.summary }) + return expected + }), + } - const result = await Effect.runPromise( - submitDiagnosisRpc({ orgId: "org_1", investigationId, report }).pipe( + const result = yield* submitDiagnosisRpc({ orgId: "org_1", investigationId, report }).pipe( Effect.provideService(InvestigationService, service), - ), - ) - expect(result).toBe(expected) - expect(calls).toEqual([ - { orgId: "org_1", investigationId, summary: "Checkout latency doubled after deploy." }, - ]) - }) + ) + expect(result).toBe(expected) + expect(calls).toEqual([ + { orgId: "org_1", investigationId, summary: "Checkout latency doubled after deploy." }, + ]) + }), + ) }) diff --git a/apps/api/src/mcp/dispatcher.test.ts b/apps/api/src/mcp/dispatcher.test.ts index 4d63ba995..e7eaad0b7 100644 --- a/apps/api/src/mcp/dispatcher.test.ts +++ b/apps/api/src/mcp/dispatcher.test.ts @@ -1,48 +1,50 @@ -import { describe, expect, it } from "vitest" +import { describe, expect, it } from "@effect/vitest" import { Effect } from "effect" import type { InternalRpcToolNotFoundError } from "@maple/domain/internal-rpc" import { callMcpTool, listMcpTools } from "./dispatcher" import { mapleToolDefinitions, toInputSchema } from "./tools/registry" describe("MCP dispatcher", () => { - it("publishes the same names, descriptions, and schemas used by HTTP MCP", async () => { - const descriptors = await Effect.runPromise(listMcpTools) - expect(descriptors).toEqual( - mapleToolDefinitions.map((definition) => ({ - name: definition.name, - description: definition.description, - inputSchema: toInputSchema(definition.schema), - })), - ) - }) + it.effect("publishes the same names, descriptions, and schemas used by HTTP MCP", () => + Effect.gen(function* () { + const descriptors = yield* listMcpTools + expect(descriptors).toEqual( + mapleToolDefinitions.map((definition) => ({ + name: definition.name, + description: definition.description, + inputSchema: toInputSchema(definition.schema), + })), + ) + }), + ) - it("returns MCP validation feedback for invalid model tool input", async () => { - const result = await Effect.runPromise( - callMcpTool("inspect_trace", {}) as unknown as Effect.Effect< + it.effect("returns MCP validation feedback for invalid model tool input", () => + Effect.gen(function* () { + const result = yield* callMcpTool("inspect_trace", {}) as unknown as Effect.Effect< { readonly content: ReadonlyArray<{ readonly type: "text"; readonly text: string }> readonly isError?: boolean }, never, never - >, - ) - expect(result.isError).toBe(true) - expect(result.content[0]?.text).toContain("Invalid parameters") - expect(result.content[0]?.text).toContain("inspect_trace") - }) + > + expect(result.isError).toBe(true) + expect(result.content[0]?.text).toContain("Invalid parameters") + expect(result.content[0]?.text).toContain("inspect_trace") + }), + ) - it("fails unknown RPC tool names with a typed error", async () => { - const error = await Effect.runPromise( - Effect.flip( + it.effect("fails unknown RPC tool names with a typed error", () => + Effect.gen(function* () { + const error = yield* Effect.flip( callMcpTool("not_a_maple_tool", {}) as Effect.Effect< never, InternalRpcToolNotFoundError, never >, - ), - ) - expect(error._tag).toBe("@maple/internal-rpc/ToolNotFoundError") - expect(error.name).toBe("not_a_maple_tool") - }) + ) + expect(error._tag).toBe("@maple/internal-rpc/ToolNotFoundError") + expect(error.name).toBe("not_a_maple_tool") + }), + ) }) diff --git a/apps/api/src/routes/scraper-internal.http.ts b/apps/api/src/routes/scraper-internal.http.ts index 6240be1eb..7c43c1042 100644 --- a/apps/api/src/routes/scraper-internal.http.ts +++ b/apps/api/src/routes/scraper-internal.http.ts @@ -117,7 +117,7 @@ export const ScraperInternalRouter = HttpRouter.use((router) => const denied = unauthorized(req) if (denied) return denied - const rows = yield* service.listAllEnabled().pipe(Effect.catch(() => Effect.succeed([]))) + const rows = yield* service.listAllEnabled() // One public ingest key per org (lazily created on first use, like // onboarding does). The scraper ingests with this key so scraped @@ -128,12 +128,9 @@ export const ScraperInternalRouter = HttpRouter.use((router) => Effect.gen(function* () { const cached = keyByOrg.get(orgId) if (cached !== undefined) return cached - const key: string | null = yield* ingestKeys + const key: string = yield* ingestKeys .getOrCreate(decodeOrgIdSync(orgId), SCRAPER_SYSTEM_USER) - .pipe( - Effect.map((keys): string | null => keys.publicKey), - Effect.catch(() => Effect.succeed(null)), - ) + .pipe(Effect.map((keys) => keys.publicKey)) keyByOrg.set(orgId, key) return key }) @@ -171,7 +168,9 @@ export const ScraperInternalRouter = HttpRouter.use((router) => if (Option.isSome(target)) { targets.push(target.value) } else { - yield* Effect.logWarning("Skipping scrape sub-target (invalid row)").pipe( + yield* Effect.logWarning( + "Skipping scrape sub-target (invalid row)", + ).pipe( Effect.annotateLogs({ scrapeTargetId: row.id, orgId: row.orgId, @@ -196,7 +195,15 @@ export const ScraperInternalRouter = HttpRouter.use((router) => ) return yield* HttpServerResponse.json(targets) - }).pipe(Effect.withSpan("ScraperInternal.listTargets")) + }).pipe( + Effect.catch((error) => + Effect.logError("Failed to build scraper target list").pipe( + Effect.annotateLogs({ error: error.message }), + Effect.as(errorText("Scraper target list unavailable", 503)), + ), + ), + Effect.withSpan("ScraperInternal.listTargets"), + ) const recordResults = (req: HttpServerRequest.HttpServerRequest) => Effect.gen(function* () { @@ -209,18 +216,18 @@ export const ScraperInternalRouter = HttpRouter.use((router) => const results = yield* decodeScrapeResultsEffect(body.value).pipe(Effect.option) if (Option.isNone(results)) return errorText("Invalid scrape results payload", 400) - yield* service - .recordScrapeResults(results.value) - .pipe( - Effect.catch((error) => - Effect.logWarning("Failed to persist scrape results").pipe( - Effect.annotateLogs({ error: error.message }), - ), - ), - ) + yield* service.recordScrapeResults(results.value) return yield* HttpServerResponse.json({ recorded: results.value.length }) - }).pipe(Effect.withSpan("ScraperInternal.recordResults")) + }).pipe( + Effect.catch((error) => + Effect.logError("Failed to persist scrape results").pipe( + Effect.annotateLogs({ error: error.message }), + Effect.as(errorText("Scrape result persistence unavailable", 503)), + ), + ), + Effect.withSpan("ScraperInternal.recordResults"), + ) yield* router.add("GET", "/api/internal/scrape-targets", listTargets) yield* router.add("POST", "/api/internal/scrape-results", recordResults) diff --git a/apps/api/src/services/AlertDeliveryDispatch.test.ts b/apps/api/src/services/AlertDeliveryDispatch.test.ts index e74de1403..69bfed125 100644 --- a/apps/api/src/services/AlertDeliveryDispatch.test.ts +++ b/apps/api/src/services/AlertDeliveryDispatch.test.ts @@ -1,7 +1,7 @@ import type { AlertDestinationRow } from "@maple/db" -import { AlertDeliveryError } from "@maple/domain/http" +import { AlertDeliveryError, AlertDestinationId } from "@maple/domain/http" import { assert, describe, it } from "@effect/vitest" -import { Effect } from "effect" +import { Effect, Schema } from "effect" import { buildAlertChatUrl, buildDiscordEmbedsFromTemplate, @@ -37,6 +37,7 @@ const baseContext: TemplateRenderContext = { const LINK = "https://web.localhost/alerts" const CHAT = "https://web.localhost/chat?mode=alert" +const DESTINATION_ID = Schema.decodeUnknownSync(AlertDestinationId)("7c6b5a49-3821-4e0f-9d8c-7b6a59483726") /** Dispatch deps for non-email destinations — email sends must not happen. */ const noEmailDeps: DispatchDeps = { @@ -138,7 +139,7 @@ describe("buildDiscordEmbedsFromTemplate", () => { describe("dispatchDelivery", () => { const destinationRow: AlertDestinationRow = { - id: "dest_1" as AlertDestinationRow["id"], + id: DESTINATION_ID, orgId: "org_1" as AlertDestinationRow["orgId"], name: "PagerDuty", type: "pagerduty", diff --git a/apps/api/src/services/AlertsService.ts b/apps/api/src/services/AlertsService.ts index 13ec465cd..255f1af07 100644 --- a/apps/api/src/services/AlertsService.ts +++ b/apps/api/src/services/AlertsService.ts @@ -27,7 +27,6 @@ import { AlertCheckDocument, AlertChecksListResponse, AlertCheckStatus as AlertCheckStatusSchema, - AlertEvaluationStatus as AlertEvaluationStatusSchema, AlertIncidentDocument, AlertIncidentsListResponse, AlertIncidentStatus, @@ -243,6 +242,8 @@ interface DeliveryAttemptFailure { } const MAX_DELIVERY_ATTEMPTS = 5 +const ALERT_TEST_DELIVERY_CONCURRENCY = 5 +const ALERT_CHECK_INGEST_CONCURRENCY = 4 // Storm fuse: cap issue-hub upserts per scheduler tick so a pathological // group-by rule opening hundreds of incidents can't stall the per-minute tick. const ISSUE_UPSERTS_PER_TICK = 50 @@ -300,7 +301,6 @@ const decodeAlertDestinationTypeSync = Schema.decodeUnknownSync(AlertDestination const decodeAlertSeveritySync = Schema.decodeUnknownSync(AlertSeveritySchema) const decodeAlertSignalTypeSync = Schema.decodeUnknownSync(AlertSignalTypeSchema) const decodeAlertComparatorSync = Schema.decodeUnknownSync(AlertComparatorSchema) -const decodeAlertEvaluationStatusSync = Schema.decodeUnknownSync(AlertEvaluationStatusSchema) const decodeAlertCheckStatusSync = Schema.decodeUnknownSync(AlertCheckStatusSchema) const decodeAlertIncidentTransitionSync = Schema.decodeUnknownSync(AlertIncidentTransitionSchema) const decodeAlertMetricTypeSync = Schema.decodeUnknownSync(AlertMetricTypeSchema) @@ -2826,7 +2826,7 @@ export class AlertsService extends Context.Service Effect.succeed("failed" as const), }), ), - { concurrency: "unbounded" }, + { concurrency: NOTIFICATION_DELIVERY_CONCURRENCY }, ) return { diff --git a/apps/api/src/services/OrgClickHouseSettingsService.test.ts b/apps/api/src/services/OrgClickHouseSettingsService.test.ts index c7838250e..ba1f1c94e 100644 --- a/apps/api/src/services/OrgClickHouseSettingsService.test.ts +++ b/apps/api/src/services/OrgClickHouseSettingsService.test.ts @@ -151,7 +151,7 @@ describe("isRetryableUpstream", () => { }) describe("execClickHouse", () => { - it.live("uses manual redirects and rejects every 3xx without following it", () => + it.effect("uses manual redirects and rejects every 3xx without following it", () => Effect.gen(function* () { let redirectMode: RequestRedirect | undefined let calls = 0 @@ -206,7 +206,7 @@ describe("execClickHouse", () => { }), ) - it.live("does NOT retry a 4xx rejection", () => + it.effect("does NOT retry a 4xx rejection", () => Effect.gen(function* () { const { state, fetchImpl } = makeFetch(() => Promise.resolve(mockResponse("Syntax error", 400))) @@ -218,7 +218,7 @@ describe("execClickHouse", () => { }), ) - it.live("does NOT retry a ClickHouse 500 SQL error (carries the DB::Exception text)", () => + it.effect("does NOT retry a ClickHouse 500 SQL error (carries the DB::Exception text)", () => Effect.gen(function* () { const { state, fetchImpl } = makeFetch(() => Promise.resolve(mockResponse("Code: 60. DB::Exception: UNKNOWN_TABLE", 500)), diff --git a/apps/api/src/services/vcs/__tests__/VcsCommitService.test.ts b/apps/api/src/services/vcs/__tests__/VcsCommitService.test.ts index 038eeb3e0..3f2f6afbe 100644 --- a/apps/api/src/services/vcs/__tests__/VcsCommitService.test.ts +++ b/apps/api/src/services/vcs/__tests__/VcsCommitService.test.ts @@ -23,6 +23,7 @@ import { testEnv, type VcsRepo, } from "./harness" +import { decodeGitCommitSha } from "./fixtures" const trackedDbs: TestDb[] = [] afterEach(() => cleanupTestDbs(trackedDbs)) @@ -195,7 +196,7 @@ describe("VcsCommitService.resolveCommitDetail", () => { // It was persisted: now stored, no further provider calls. const before = calls.commitGets - const stored = expectSome(yield* repo.findCommitBySha(orgId, SHA as never)) + const stored = expectSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA))) assert.strictEqual( stored.repositoryId, expectSome(yield* repo.resolveRepository(orgId, "github", "7")).id, diff --git a/apps/api/src/services/vcs/__tests__/fixtures.ts b/apps/api/src/services/vcs/__tests__/fixtures.ts new file mode 100644 index 000000000..c23c4212e --- /dev/null +++ b/apps/api/src/services/vcs/__tests__/fixtures.ts @@ -0,0 +1,5 @@ +import { GitCommitSha } from "@maple/domain/http" +import { Schema } from "effect" + +/** Decode commit SHA fixtures through the same branded schema used in production. */ +export const decodeGitCommitSha = Schema.decodeUnknownSync(GitCommitSha) diff --git a/apps/api/src/services/vcs/__tests__/vcs.test.ts b/apps/api/src/services/vcs/__tests__/vcs.test.ts index 1493e6565..eee71bef9 100644 --- a/apps/api/src/services/vcs/__tests__/vcs.test.ts +++ b/apps/api/src/services/vcs/__tests__/vcs.test.ts @@ -1,7 +1,6 @@ import { afterEach, assert, describe, it } from "@effect/vitest" import { randomUUID } from "node:crypto" import { - GitCommitSha, VcsInstallation, VcsInstallationGoneError, VcsProviderError, @@ -45,6 +44,7 @@ import { type VcsRepo, WEBHOOK_SECRET, } from "./harness" +import { decodeGitCommitSha } from "./fixtures" const trackedDbs: TestDb[] = [] afterEach(() => cleanupTestDbs(trackedDbs)) @@ -203,10 +203,7 @@ describe("GithubProvider.webhookToJobs", () => { // Push payloads carry no avatar URL — the provider derives one from the // committer login against the commit's own host (here github.com), so the // dashboard never has to patch a null avatar. - assert.strictEqual( - job.commits[0]!.authorAvatarUrl, - "https://github.com/octocat.png?size=64", - ) + assert.strictEqual(job.commits[0]!.authorAvatarUrl, "https://github.com/octocat.png?size=64") }).pipe(Effect.provide(providerLayer())), ) @@ -233,10 +230,7 @@ describe("GithubProvider.webhookToJobs", () => { }) const job = jobs[0]! if (job.kind !== "push") return assert.fail("expected a push job") - assert.strictEqual( - job.commits[0]!.authorAvatarUrl, - "https://github.acme.com/octocat.png?size=64", - ) + assert.strictEqual(job.commits[0]!.authorAvatarUrl, "https://github.acme.com/octocat.png?size=64") }).pipe(Effect.provide(providerLayer())), ) @@ -734,14 +728,14 @@ describe("VcsRepository", () => { assert.ok(branches.filter((b) => b.name !== "main").every((b) => !b.isDefault)) yield* repo.upsertCommits(r, [mk(SHA_X, 100), mk(SHA_Y, 200)]) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_X as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_X)))) // Changing the tracked branch wipes the repo's stored (old-branch) commits. yield* repo.changeTrackedBranch(orgId, r.id, "feature") r = yield* repoFor(repo, orgId, "7") assert.strictEqual(r.trackedBranch, "feature") - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_X as never))) - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_Y as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_X)))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_Y)))) // Reconcile: remote lacks "stale" → deleted; its name is returned. const deleted = yield* repo.reconcileBranchDeletions(r.id, new Set(["main", "feature"]), { @@ -824,7 +818,7 @@ describe("VcsRepository", () => { ]) assert.strictEqual(count, 1) - const commit = yield* repo.findCommitBySha(orgId, SHA as never) + const commit = yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA)) assert.ok(Option.isSome(commit)) assert.strictEqual(commit.value.authorLogin, "octocat") assert.strictEqual(commit.value.repositoryId, repoRow.value.id) @@ -846,18 +840,7 @@ describe("VcsRepository", () => { (id, org_id, provider, external_installation_id, account_login, account_type, external_account_id, repository_selection, status, installed_by_user_id, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10, now(), now())`, - [ - randomUUID(), - "org_x", - "github", - "55", - "octo", - "team", - "1", - "all", - "active", - "user_1", - ], + [randomUUID(), "org_x", "github", "55", "octo", "team", "1", "all", "active", "user_1"], ), ) const exit = yield* repo.resolveInstallation("github", "55").pipe(Effect.exit) @@ -926,12 +909,12 @@ describe("VcsRepository", () => { assert.ok(Option.isNone(yield* repo.resolveInstallation("github", "42"))) assert.strictEqual((yield* reposOfInstallation(repo, "42", "all")).length, 0) - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_42 as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_42)))) // Installation 99 is untouched (delete was scoped to 42's repo ids). assert.ok(Option.isSome(yield* repo.resolveInstallation("github", "99"))) assert.strictEqual((yield* reposOfInstallation(repo, "99", "all")).length, 1) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_99 as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_99)))) // Idempotent: purging again is a no-op, not an error. yield* purgeInstallationFor(repo, orgId, "42") @@ -1023,8 +1006,8 @@ describe("VcsRepository", () => { const repoA = yield* repoFor(repo, orgA, "7") const repoB = yield* repoFor(repo, orgB, "8") - const foundA = yield* repo.findCommitBySha(orgA, SHARED as never) - const foundB = yield* repo.findCommitBySha(orgB, SHARED as never) + const foundA = yield* repo.findCommitBySha(orgA, decodeGitCommitSha(SHARED)) + const foundB = yield* repo.findCommitBySha(orgB, decodeGitCommitSha(SHARED)) assert.strictEqual(expectSome(foundA).repositoryId, repoA.id) assert.strictEqual(expectSome(foundB).repositoryId, repoB.id) // And never the other org's row. @@ -1064,7 +1047,7 @@ describe("VcsRepository", () => { const after = yield* repoFor(repo, orgId, "7") assert.strictEqual(after.trackedBranch, "main") // untouched - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_X as never))) // not wiped + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_X)))) // not wiped }).pipe(Effect.provide(repoLayer(testDb))) }) @@ -1119,7 +1102,7 @@ describe("VcsRepository", () => { }, ]) // Stored lowercased → found by the lowercase form. - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, "a".repeat(40) as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha("a".repeat(40))))) }).pipe(Effect.provide(repoLayer(testDb))) }) @@ -1412,7 +1395,7 @@ describe("VcsSyncService orchestrator", () => { }), ) assert.ok(!(yield* repo.listBranchesByRepository(r.id)).some((b) => b.name === "feature/x")) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) assert.strictEqual(sent.length, 0) // branch events make no GitHub/queue calls }).pipe(Effect.provide(orchestratorLayer(testDb, { sent, repos: oneRepo }))) }) @@ -1433,7 +1416,7 @@ describe("VcsSyncService orchestrator", () => { ]) yield* repo.changeTrackedBranch(orgId, r.id, "release") yield* repo.upsertCommits(r, [commit(SHA_A, 1)]) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) yield* svc.processMessage( Schema.encodeSync(VcsSyncJob)({ @@ -1450,7 +1433,7 @@ describe("VcsSyncService orchestrator", () => { // backfill of the default enqueued. const updated = yield* repoFor(repo, orgId, "7") assert.strictEqual(updated.trackedBranch, "main") - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) const backfills = sent.filter((j) => j.kind === "sync-commits") assert.strictEqual(backfills.length, 1) assert.strictEqual(backfills[0]!.kind === "sync-commits" ? backfills[0]!.branch : "", "main") @@ -1489,13 +1472,17 @@ describe("VcsSyncService orchestrator", () => { // retarget to "main": commits wiped, exactly one backfill, for "main". const updated = yield* repoFor(repo, orgId, "7") assert.strictEqual(updated.trackedBranch, "main") - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) const backfills = sent.filter((j) => j.kind === "sync-commits") assert.strictEqual(backfills.length, 1) assert.strictEqual(backfills[0]!.kind === "sync-commits" ? backfills[0]!.branch : "", "main") }).pipe( Effect.provide( - orchestratorLayer(testDb, { sent, repos: oneRepo, branches: [{ name: "main", headSha: null }] }), + orchestratorLayer(testDb, { + sent, + repos: oneRepo, + branches: [{ name: "main", headSha: null }], + }), ), ) }) @@ -1527,7 +1514,7 @@ describe("VcsSyncService orchestrator", () => { "main", ) // Forced ⇒ the payload is discarded (we re-walk instead), so SHA_A is not stored. - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) }).pipe(Effect.provide(orchestratorLayer(testDb, { sent, repos: oneRepo }))) }) @@ -1547,7 +1534,7 @@ describe("VcsSyncService orchestrator", () => { commits: [commit(SHA_A, 1)], } yield* svc.processMessage(Schema.encodeSync(VcsSyncJob)(job)) // must not fail - const found = yield* repo.findCommitBySha(orgId, SHA_A as never) + const found = yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)) assert.ok(Option.isNone(found)) assert.strictEqual(sent.length, 0) }).pipe(Effect.provide(orchestratorLayer(testDb, { sent }))) @@ -1577,7 +1564,7 @@ describe("VcsSyncService orchestrator", () => { const branches = yield* repo.listBranchesByRepository(r.id) assert.ok(branches.some((b) => b.name === "feature/x")) // …and its commits are NOT stored — only the tracked branch's commits are. - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) }).pipe(Effect.provide(orchestratorLayer(testDb, { sent }))) }) @@ -1603,7 +1590,7 @@ describe("VcsSyncService orchestrator", () => { commits: [commit(SHA_A, 1)], }), ) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) }).pipe(Effect.provide(orchestratorLayer(testDb, { sent }))) }) @@ -1629,7 +1616,7 @@ describe("VcsSyncService orchestrator", () => { commits: [commit(SHA_A, 1)], }), ) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) // push landed… + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) // push landed… // …yet the sync status stays exactly as the backfill left it (untouched here). const stored = yield* reposOfInstallation(repo, "42", "all") assert.strictEqual(stored[0]!.syncStatus, "pending") @@ -1734,7 +1721,7 @@ describe("VcsSyncService orchestrator", () => { // The prior installation, its repos, and its commits are all hard-deleted… assert.ok(Option.isNone(yield* repo.resolveInstallation("github", "11"))) assert.strictEqual((yield* reposOfInstallation(repo, "11", "all")).length, 0) - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, STALE_SHA as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(STALE_SHA)))) // …leaving exactly the new installation, which synced its own repo. const remaining = (yield* repo.listInstallationsByOrg(orgId)).filter( @@ -1780,8 +1767,8 @@ describe("VcsSyncService orchestrator", () => { sinceMs: 0, } yield* svc.processMessage(Schema.encodeSync(VcsSyncJob)(job)) - const a = yield* repo.findCommitBySha(orgId, SHA_A as never) - const b = yield* repo.findCommitBySha(orgId, SHA_B as never) + const a = yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)) + const b = yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_B)) assert.ok(Option.isSome(a) && Option.isSome(b)) const stored = yield* reposOfInstallation(repo, "42", "all") assert.strictEqual(stored[0]!.syncStatus, "ready") @@ -1908,7 +1895,7 @@ describe("VcsSyncService orchestrator", () => { commits: [commit(SHA_A, 1)], } yield* svc.processMessage(Schema.encodeSync(VcsSyncJob)(job)) // gated → must not fail - const a = yield* repo.findCommitBySha(orgId, SHA_A as never) + const a = yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)) assert.ok(Option.isNone(a)) // gate short-circuits before the upsert const inst = yield* repo.resolveInstallation("github", "42") assert.ok(Option.isSome(inst)) @@ -1968,7 +1955,7 @@ describe("VcsSyncService orchestrator", () => { yield* seedRepo(repo) yield* svc.processMessage(Schema.encodeSync(VcsSyncJob)(backfillJob)) // must not fail // The fetched commit was persisted… - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) // …the repo is marked backfilling (in progress, not ready)… const stored = yield* reposOfInstallation(repo, "42", "all") assert.strictEqual(stored[0]!.syncStatus, "backfilling") @@ -2005,7 +1992,7 @@ describe("VcsSyncService orchestrator", () => { yield* seedRepo(repo) yield* svc.processMessage(Schema.encodeSync(VcsSyncJob)(backfillJob)) // must not fail // The fetched page was persisted and the repo marked backfilling… - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) const stored = yield* reposOfInstallation(repo, "42", "all") assert.strictEqual(stored[0]!.syncStatus, "backfilling") // …and a continuation was requeued from the watermark with NO delay… @@ -2127,7 +2114,7 @@ describe("VcsSyncService orchestrator", () => { assert.strictEqual(all.length, 1) assert.strictEqual(all[0]!.status, "removed") // Its commits are retained (soft delete, not a purge). - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) }).pipe(Effect.provide(orchestratorLayer(testDb, { sent, repos: [] }))) }) @@ -2192,7 +2179,7 @@ describe("VcsSyncService orchestrator", () => { commits: [commit(SHA_A, 1)], } yield* svc.processMessage(Schema.encodeSync(VcsSyncJob)(job)) // must not fail - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) }).pipe(Effect.provide(orchestratorLayer(testDb, { sent }))) }) @@ -2228,7 +2215,9 @@ describe("VcsSyncService orchestrator", () => { // …and reaches back exactly one window from the enqueue moment. assert.ok(backfill.sinceMs >= before - BACKFILL_WINDOW_MS) assert.ok(backfill.sinceMs <= after - BACKFILL_WINDOW_MS) - }).pipe(Effect.provide(orchestratorLayer(testDb, { sent, branches: [{ name: "main", headSha: null }] }))) + }).pipe( + Effect.provide(orchestratorLayer(testDb, { sent, branches: [{ name: "main", headSha: null }] })), + ) }) // A continuation must carry the original walk's `sinceMs` (and branch) unchanged — @@ -2485,7 +2474,7 @@ describe("VcsSyncService orchestrator", () => { }), ) assert.ok((yield* repo.listBranchesByRepository(r.id)).some((b) => b.name === "main")) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_A as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_A)))) }).pipe(Effect.provide(orchestratorLayer(testDb, { sent }))) }) @@ -2639,7 +2628,7 @@ describe("VcsSyncService orchestrator", () => { // validation fires at both the webhook decode boundary and on persistence. describe("git SHA validation (branded type)", () => { it("GitCommitSha accepts mixed-case input and normalizes it to lowercase", () => { - const decode = Schema.decodeUnknownSync(GitCommitSha) + const decode = decodeGitCommitSha // All-uppercase 40-hex is accepted and lowercased. assert.strictEqual(decode("A".repeat(40)), "a".repeat(40)) // Mixed case round-trips to its lowercase form (so case never splits a row). diff --git a/apps/api/src/services/vcs/vendor/github/__tests__/GithubConnectService.test.ts b/apps/api/src/services/vcs/vendor/github/__tests__/GithubConnectService.test.ts index 4d2eb9371..dec7c3a4d 100644 --- a/apps/api/src/services/vcs/vendor/github/__tests__/GithubConnectService.test.ts +++ b/apps/api/src/services/vcs/vendor/github/__tests__/GithubConnectService.test.ts @@ -25,6 +25,7 @@ import { upsertCommitsFor, upsertReposFor, } from "../../../__tests__/harness" +import { decodeGitCommitSha } from "../../../__tests__/fixtures" const trackedDbs: TestDb[] = [] afterEach(() => cleanupTestDbs(trackedDbs)) @@ -325,7 +326,7 @@ describe("GithubConnectService", () => { assert.strictEqual(result.disconnected, true) assert.ok(Option.isNone(yield* repo.resolveInstallation("github", "42"))) assert.strictEqual((yield* reposOfInstallation(repo, "42", "all")).length, 0) - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA)))) // A user-initiated disconnect fully removes the row, so status reverts to the // pristine "never connected" state; a second disconnect is a no-op. @@ -504,9 +505,9 @@ describe("GithubConnectService", () => { assert.strictEqual(result.deleted, true) assert.ok(Option.isNone(yield* repo.resolveRepository(orgId, "github", "7"))) - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA_7 as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_7)))) assert.ok(Option.isSome(yield* repo.resolveRepository(orgId, "github", "8"))) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA_8 as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA_8)))) // Deleting the same (now-absent) id again is a no-op (idempotent). const again = yield* svc.deleteRepository(orgId, repo7Id) @@ -563,7 +564,7 @@ describe("GithubConnectService", () => { const exit = yield* svc.deleteRepository(orgId, repo7.value.id).pipe(Effect.exit) assert.ok(findError(exit) instanceof IntegrationsValidationError) assert.ok(Option.isSome(yield* repo.resolveRepository(orgId, "github", "7"))) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA)))) }).pipe(Effect.provide(connectLayer(testDb, http, sent))) }) @@ -620,13 +621,13 @@ describe("GithubConnectService", () => { // An unknown branch is rejected; nothing changes. const bad = yield* svc.setTrackedBranch(orgId, r.id, "nope").pipe(Effect.exit) assert.ok(findError(bad) instanceof IntegrationsValidationError) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA)))) // Selecting the current branch is a no-op: no wipe, no backfill enqueued. sent.length = 0 const noop = yield* svc.setTrackedBranch(orgId, r.id, "main") assert.strictEqual(noop.backfillQueued, false) - assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, SHA as never))) + assert.ok(Option.isSome(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA)))) assert.strictEqual(sent.filter((j) => j.kind === "sync-commits").length, 0) // Changing the branch wipes commits and enqueues a backfill. @@ -634,7 +635,7 @@ describe("GithubConnectService", () => { const changed = yield* svc.setTrackedBranch(orgId, r.id, "release") assert.strictEqual(changed.trackedBranch, "release") assert.ok(changed.backfillQueued) - assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, SHA as never))) + assert.ok(Option.isNone(yield* repo.findCommitBySha(orgId, decodeGitCommitSha(SHA)))) assert.strictEqual((yield* repoFor(repo, orgId, "7")).trackedBranch, "release") const backfills = sent.filter((j) => j.kind === "sync-commits") assert.strictEqual(backfills.length, 1) diff --git a/apps/api/src/vcs-sync-runtime.ts b/apps/api/src/vcs-sync-runtime.ts index 1a2a504fc..fea7253df 100644 --- a/apps/api/src/vcs-sync-runtime.ts +++ b/apps/api/src/vcs-sync-runtime.ts @@ -1,6 +1,6 @@ import type { MessageBatch } from "@cloudflare/workers-types" import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" -import { ANTICIPATED_ERROR_TAGS } from "@maple/domain/anticipated-errors" +import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { WorkerConfigProviderLayer, WorkerEnvironment } from "@maple/effect-cloudflare" import { Cause, Effect, Layer, Option } from "effect" import { layerPg } from "./lib/DatabasePgLive" @@ -24,7 +24,7 @@ const telemetry = MapleCloudflareSDK.make({ serviceName: "maple-api", serviceNamespace: "backend", repositoryUrl: "https://github.com/Makisuo/maple", - anticipatedErrorTags: [...ANTICIPATED_ERROR_TAGS], + anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS], }) export const buildVcsSyncLayer = (_env: Record) => { diff --git a/apps/api/src/worker.ts b/apps/api/src/worker.ts index 1b0b07353..fdbac0fa8 100644 --- a/apps/api/src/worker.ts +++ b/apps/api/src/worker.ts @@ -1,6 +1,6 @@ import type { MessageBatch, ScheduledController } from "@cloudflare/workers-types" import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" -import { ANTICIPATED_ERROR_TAGS } from "@maple/domain/anticipated-errors" +import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { layerFromEnvRecord, runScheduledEffect, @@ -53,7 +53,7 @@ const telemetry = MapleCloudflareSDK.make({ dropSpanNames: ["McpServer/Notifications."], // Expected 4xx outcomes (validation, not-found, unauthorized, …) record as // Ok spans instead of errors — see @maple/domain/anticipated-errors. - anticipatedErrorTags: [...ANTICIPATED_ERROR_TAGS], + anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS], }) // `HttpMiddleware.tracer` ends the root server span on a deferred macrotask @@ -280,13 +280,23 @@ const handle = async ( return response } catch (err) { console.error("[worker] handler failed:", err) + const message = err instanceof Error ? err.message : String(err) + Effect.runFork( + Effect.logError("API worker handler failed").pipe( + Effect.annotateLogs({ + error: message, + method: request.method, + url: request.url, + }), + Effect.provide(telemetry.layer), + ), + ) if (isMcp && mcpFrame) { console.error( `[mcp-err] method=${mcpFrame.method} id=${mcpFrame.id}` + ` dur=${Date.now() - startedAt}ms`, ) } ctx.waitUntil(flushTelemetry(env)) - const message = err instanceof Error ? err.message : String(err) return new Response(`worker handler error: ${message}`, { status: 504 }) } } diff --git a/apps/api/src/workflows/AiTriageWorkflow.run.ts b/apps/api/src/workflows/AiTriageWorkflow.run.ts index 754d19928..8bb72dc54 100644 --- a/apps/api/src/workflows/AiTriageWorkflow.run.ts +++ b/apps/api/src/workflows/AiTriageWorkflow.run.ts @@ -22,7 +22,7 @@ import { createFlueClient } from "@flue/sdk" import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { aiTriageRuns, anomalyIncidents, errorIssueEvents } from "@maple/db" import { createMaplePgClient, type MaplePgClient } from "@maple/db/client" -import { ANTICIPATED_ERROR_TAGS } from "@maple/domain/anticipated-errors" +import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { AiTriageResult } from "@maple/domain/http" import { AiTriageRunId, @@ -177,7 +177,7 @@ const triageTelemetry = MapleCloudflareSDK.make({ serviceName: "maple-api", serviceNamespace: "backend", repositoryUrl: "https://github.com/Makisuo/maple", - anticipatedErrorTags: [...ANTICIPATED_ERROR_TAGS], + anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS], }) const GATE_STEP = { retries: { limit: 3, delay: "2 seconds", backoff: "exponential" } } diff --git a/apps/cli/src/bin.ts b/apps/cli/src/bin.ts index 833d39f1d..b9da65a45 100644 --- a/apps/cli/src/bin.ts +++ b/apps/cli/src/bin.ts @@ -3,6 +3,7 @@ import { BunRuntime } from "@effect/platform-bun" import * as BunServices from "@effect/platform-bun/BunServices" import { Effect, Layer } from "effect" import * as Command from "effect/unstable/cli/Command" +import { FetchHttpClient } from "effect/unstable/http" import { cli } from "./cli" import { MapleConfig } from "./core/config" import { Mode } from "./core/mode" @@ -20,6 +21,7 @@ const MainLayer = WarehouseExecutorFromMode.pipe( Layer.provideMerge(Mode.layer), Layer.provideMerge(MapleConfig.layer), Layer.provideMerge(BunServices.layer), + Layer.provideMerge(FetchHttpClient.layer), ) // Throttled, non-blocking "update available" notice before dispatching the diff --git a/apps/cli/src/commands/auth.ts b/apps/cli/src/commands/auth.ts index c42df0918..9e7bf7231 100644 --- a/apps/cli/src/commands/auth.ts +++ b/apps/cli/src/commands/auth.ts @@ -96,7 +96,7 @@ export const whoami = Command.make("whoami", {}).pipe( Effect.fnUntraced(function* () { const config = yield* MapleConfig const mode = yield* Mode - const defaultMode = config.defaultMode ?? "auto" + const defaultMode = Option.getOrElse(config.defaultMode, () => "auto" as const) const resolved = yield* mode.resolve.pipe( Effect.map((m) => ({ ok: true as const, m })), Effect.catch((e) => Effect.succeed({ ok: false as const, message: e.message })), diff --git a/apps/cli/src/core/config.ts b/apps/cli/src/core/config.ts index edb676e75..b1c3f5b61 100644 --- a/apps/cli/src/core/config.ts +++ b/apps/cli/src/core/config.ts @@ -1,4 +1,4 @@ -import { Clock, Context, Effect, Layer, type PlatformError, Schema } from "effect" +import { Clock, Context, Effect, Layer, Option, Redacted, type PlatformError, Schema } from "effect" import { FileSystem } from "effect/FileSystem" import * as os from "node:os" import * as path from "node:path" @@ -64,19 +64,19 @@ const writeMerged = ( export interface MapleConfigShape { /** Remote API base URL (env `MAPLE_API_URL` overrides the stored value). */ - readonly apiUrl: string | undefined + readonly apiUrl: Option.Option /** Remote bearer token (env `MAPLE_API_TOKEN` overrides the stored value). */ - readonly token: string | undefined - readonly orgId: string | undefined + readonly token: Option.Option> + readonly orgId: Option.Option /** Local binary base URL (env `MAPLE_LOCAL_URL`, else the default). */ readonly localUrl: string - readonly defaultMode: "local" | "remote" | undefined + readonly defaultMode: Option.Option<"local" | "remote"> /** API URL to use for `maple login` when none is passed. */ readonly defaultApiUrl: string - /** ISO timestamp of the last startup update check (undefined = never checked). */ - readonly lastUpdateCheck: string | undefined - /** Latest release tag seen by the last update check, or undefined. */ - readonly latestKnownVersion: string | undefined + /** ISO timestamp of the last startup update check (`None` = never checked). */ + readonly lastUpdateCheck: Option.Option + /** Latest release tag seen by the last update check, or `None`. */ + readonly latestKnownVersion: Option.Option /** Persist config fields (merged with existing). */ readonly write: (next: StoredConfig) => Effect.Effect /** Remove the stored token (used by `maple logout`). */ @@ -96,14 +96,14 @@ export class MapleConfig extends Context.Service( const stored = yield* readStored(fs) const env = process.env return { - apiUrl: env.MAPLE_API_URL ?? stored.apiUrl, - token: env.MAPLE_API_TOKEN ?? stored.token, - orgId: env.MAPLE_ORG_ID ?? stored.orgId, + apiUrl: Option.fromNullishOr(env.MAPLE_API_URL ?? stored.apiUrl), + token: Option.map(Option.fromNullishOr(env.MAPLE_API_TOKEN ?? stored.token), Redacted.make), + orgId: Option.fromNullishOr(env.MAPLE_ORG_ID ?? stored.orgId), localUrl: env.MAPLE_LOCAL_URL ?? defaultLocalUrl(env.MAPLE_LOCAL_BIND_HOST), - defaultMode: stored.defaultMode, + defaultMode: Option.fromNullishOr(stored.defaultMode), defaultApiUrl: env.MAPLE_API_URL ?? DEFAULT_API_URL, - lastUpdateCheck: stored.lastUpdateCheck, - latestKnownVersion: stored.latestKnownVersion, + lastUpdateCheck: Option.fromNullishOr(stored.lastUpdateCheck), + latestKnownVersion: Option.fromNullishOr(stored.latestKnownVersion), write: (next) => writeMerged(fs, (cur) => ({ ...cur, ...next })), clearToken: () => writeMerged(fs, (cur) => { diff --git a/apps/cli/src/core/mode.ts b/apps/cli/src/core/mode.ts index 5aa1ef019..8355b90d3 100644 --- a/apps/cli/src/core/mode.ts +++ b/apps/cli/src/core/mode.ts @@ -1,4 +1,5 @@ -import { Context, Effect, Layer, Schema } from "effect" +import { Context, Duration, Effect, Layer, Option, Redacted, Schema } from "effect" +import { HttpClient, HttpClientRequest } from "effect/unstable/http" import { MapleConfig } from "./config" /** @@ -28,17 +29,14 @@ const hasFlag = (name: string): boolean => typeof process !== "undefined" && Array.isArray(process.argv) && process.argv.includes(name) /** Fast, non-fatal liveness probe of the local binary's `/health` route. */ -const probeLocal = (baseUrl: string): Effect.Effect => - Effect.tryPromise(async () => { - const controller = new AbortController() - const timer = setTimeout(() => controller.abort(), 400) - try { - const res = await fetch(`${baseUrl.replace(/\/$/, "")}/health`, { signal: controller.signal }) - return res.ok - } finally { - clearTimeout(timer) - } - }).pipe(Effect.orElseSucceed(() => false)) +const probeLocal = (client: HttpClient.HttpClient, baseUrl: string): Effect.Effect => { + const request = HttpClientRequest.get(`${baseUrl.replace(/\/$/, "")}/health`) + return client.execute(request).pipe( + Effect.map((response) => response.status >= 200 && response.status < 300), + Effect.timeoutOrElse({ duration: Duration.millis(400), orElse: () => Effect.succeed(false) }), + Effect.orElseSucceed(() => false), + ) +} export interface ModeShape { /** Resolve the active backend. Fails with `ModeError` if none is available. */ @@ -48,14 +46,18 @@ export interface ModeShape { export class Mode extends Context.Service()("@maple/cli/Mode", { make: Effect.gen(function* () { const config = yield* MapleConfig + const client = yield* HttpClient.HttpClient - const hasRemoteCreds = !!config.token && !!config.apiUrl - const remote = (): ResolvedMode => ({ - _tag: "remote", - apiUrl: config.apiUrl!, - token: config.token!, - orgId: config.orgId, - }) + const remoteConfig = Option.map( + Option.all({ apiUrl: config.apiUrl, token: config.token }), + ({ apiUrl, token }): ResolvedMode => ({ + _tag: "remote", + apiUrl, + token: Redacted.value(token), + orgId: Option.getOrUndefined(config.orgId), + }), + ) + const remote = Option.getOrUndefined(remoteConfig) const local = (): ResolvedMode => ({ _tag: "local", baseUrl: config.localUrl }) const resolve: Effect.Effect = Effect.gen(function* () { @@ -66,24 +68,24 @@ export class Mode extends Context.Service()("@maple/cli/Mode", return yield* new ModeError({ message: "Cannot use --remote and --local together." }) } if (forceRemote) { - if (!hasRemoteCreds) { + if (remote === undefined) { return yield* new ModeError({ message: "Remote mode needs a workspace. Run `maple login`, or set MAPLE_API_URL + MAPLE_API_TOKEN.", }) } - return remote() + return remote } if (forceLocal) return local() // Stored preference. - if (config.defaultMode === "remote" && hasRemoteCreds) return remote() - if (config.defaultMode === "local") return local() + if (Option.contains(config.defaultMode, "remote") && remote !== undefined) return remote + if (Option.contains(config.defaultMode, "local")) return local() // Auto-detect: a configured token implies remote; otherwise probe for a // running local binary. - if (hasRemoteCreds) return remote() - if (yield* probeLocal(config.localUrl)) return local() + if (remote !== undefined) return remote + if (yield* probeLocal(client, config.localUrl)) return local() return yield* new ModeError({ message: diff --git a/apps/cli/src/core/remote-executor.test.ts b/apps/cli/src/core/remote-executor.test.ts index 7f36dcaad..a95b4329b 100644 --- a/apps/cli/src/core/remote-executor.test.ts +++ b/apps/cli/src/core/remote-executor.test.ts @@ -1,15 +1,10 @@ -import { afterEach, describe, it } from "@effect/vitest" +import { describe, it } from "@effect/vitest" import { strict as assert } from "node:assert" import { Effect, Tracer } from "effect" +import { FetchHttpClient, HttpClient } from "effect/unstable/http" import { WarehouseUpstreamError } from "@maple/domain/http/warehouse-errors" import { makeRemoteWarehouseExecutorShape } from "./remote-executor" -const realFetch = globalThis.fetch - -afterEach(() => { - globalThis.fetch = realFetch -}) - const makeRecordingTracer = () => { const spans: Array = [] const tracer = Tracer.make({ @@ -30,14 +25,24 @@ describe("remote warehouse executor", () => { pipeName: "list_services", upstreamStatus: 503, }) - globalThis.fetch = (async () => + const fetchStub = (async () => new Response(JSON.stringify(upstream), { status: 503, headers: { "content-type": "application/json" }, })) as unknown as typeof fetch - const shape = makeRemoteWarehouseExecutorShape("https://api.maple.dev", "test-token", "org_test") - - const error = yield* Effect.flip(shape.query("get_service_usage", {})) + const error = yield* Effect.gen(function* () { + const client = yield* HttpClient.HttpClient + const shape = makeRemoteWarehouseExecutorShape( + client, + "https://api.maple.dev", + "test-token", + "org_test", + ) + return yield* Effect.flip(shape.query("get_service_usage", {})) + }).pipe( + Effect.provide(FetchHttpClient.layer), + Effect.provideService(FetchHttpClient.Fetch, fetchStub), + ) assert.ok(error instanceof WarehouseUpstreamError) assert.strictEqual(error.upstreamStatus, 503) @@ -46,17 +51,27 @@ describe("remote warehouse executor", () => { it.effect("attributes the HTTP peer without inventing a database system", () => Effect.gen(function* () { - globalThis.fetch = (async () => + const fetchStub = (async () => new Response(JSON.stringify({ data: [] }), { status: 200, headers: { "content-type": "application/json" }, })) as unknown as typeof fetch const { spans, tracer } = makeRecordingTracer() - const shape = makeRemoteWarehouseExecutorShape("https://api.maple.dev", "test-token", "org_test") - - yield* shape - .query("get_service_usage", {}, { context: "remoteServices", profile: "list" }) - .pipe(Effect.withTracer(tracer)) + yield* Effect.gen(function* () { + const client = yield* HttpClient.HttpClient + const shape = makeRemoteWarehouseExecutorShape( + client, + "https://api.maple.dev", + "test-token", + "org_test", + ) + yield* shape + .query("get_service_usage", {}, { context: "remoteServices", profile: "list" }) + .pipe(Effect.withTracer(tracer)) + }).pipe( + Effect.provide(FetchHttpClient.layer), + Effect.provideService(FetchHttpClient.Fetch, fetchStub), + ) const span = spans.find((candidate) => candidate.name === "warehouse.query") assert.ok(span) diff --git a/apps/cli/src/core/remote-executor.ts b/apps/cli/src/core/remote-executor.ts index 227c4bc25..a8b3b6188 100644 --- a/apps/cli/src/core/remote-executor.ts +++ b/apps/cli/src/core/remote-executor.ts @@ -1,4 +1,5 @@ -import { Effect, Option, Schema, type Option as EffectOption } from "effect" +import { Duration, Effect, Option, Schema, type Option as EffectOption } from "effect" +import { HttpClient, HttpClientRequest } from "effect/unstable/http" import { type WarehouseExecutorShape, type SqlQueryOptions } from "@maple/query-engine/observability" import { type WarehouseError, @@ -45,6 +46,7 @@ const unsupported = (pipeName: string): Effect.Effect(pipe: string, params: Record, options?: SqlQueryOptions) => - Effect.tryPromise({ - try: async (): Promise<{ data: ReadonlyArray }> => { - const started = performance.now() - try { - const res = await fetch(endpoint, { - method: "POST", - headers: { - "content-type": "application/json", - authorization: `Bearer ${token}`, - }, - body: JSON.stringify({ pipe, params }), - }) - const text = await res.text() - if (!res.ok) { - const decoded = decodeRemoteWarehouseError(text) - if (decoded !== undefined) throw decoded - throw new Error(`HTTP ${res.status}${text ? `: ${text}` : ""}`) - } - const json = decodeRemoteQueryResponseJson(text) - return { data: (json.data ?? []) as ReadonlyArray } - } finally { - // Server-side SQL isn't returned; log the pipe + params instead. - debugLog( - `${pipe} · ${Math.round(performance.now() - started)}ms`, - JSON.stringify(params), + Effect.gen(function* () { + const started = performance.now() + const request = HttpClientRequest.post(endpoint, { + headers: { authorization: `Bearer ${token}` }, + }).pipe(HttpClientRequest.bodyText(JSON.stringify({ pipe, params }), "application/json")) + return yield* Effect.gen(function* () { + const response = yield* client + .execute(request) + .pipe( + Effect.mapError( + (error) => + new WarehouseQueryError({ + message: error.message, + pipeName: pipe, + cause: error, + }), + ), ) + const text = yield* response.text.pipe( + Effect.mapError( + (error) => + new WarehouseQueryError({ + message: error.message, + pipeName: pipe, + cause: error, + }), + ), + ) + if (response.status < 200 || response.status >= 300) { + const decoded = decodeRemoteWarehouseError(text) + if (decoded !== undefined) return yield* decoded + return yield* new WarehouseQueryError({ + message: `HTTP ${response.status}${text ? `: ${text}` : ""}`, + pipeName: pipe, + cause: text, + }) } - }, - catch: (error) => { - const decoded = Schema.decodeUnknownOption(WarehouseErrorSchema)(error) - return Option.isSome(decoded) - ? decoded.value - : new WarehouseQueryError({ + const json = yield* Effect.try({ + try: () => decodeRemoteQueryResponseJson(text), + catch: (error) => + new WarehouseQueryError({ message: error instanceof Error ? error.message : String(error), pipeName: pipe, cause: error, - }) - }, + }), + }) + return { data: (json.data ?? []) as ReadonlyArray } + }).pipe( + Effect.timeoutOrElse({ + duration: Duration.minutes(2), + orElse: () => + Effect.fail( + new WarehouseQueryError({ + message: "Remote warehouse query timed out after 2 minutes", + pipeName: pipe, + cause: "timeout", + }), + ), + }), + Effect.ensuring( + Effect.sync(() => + debugLog( + `${pipe} · ${Math.round(performance.now() - started)}ms`, + JSON.stringify(params), + ), + ), + ), + ) }).pipe( Effect.tap((result) => Effect.annotateCurrentSpan({ "result.rowCount": result.data.length })), Effect.withSpan("warehouse.query", { diff --git a/apps/cli/src/core/update.ts b/apps/cli/src/core/update.ts index 77e1a4364..d20ca086e 100644 --- a/apps/cli/src/core/update.ts +++ b/apps/cli/src/core/update.ts @@ -13,7 +13,8 @@ // rename swaps the directory entry, so the running process keeps its old inode // while new invocations pick up the new binary. Keep the triple/URL logic here // in sync with install.sh. -import { Clock, Effect, Option, Schema } from "effect" +import { Clock, Duration, Effect, Option, Schema } from "effect" +import { HttpClient, HttpClientRequest } from "effect/unstable/http" import { realpathSync } from "node:fs" import { chmod, mkdir, rename, rm } from "node:fs/promises" import { dirname, join } from "node:path" @@ -31,6 +32,8 @@ const REPO = "Makisuo/maple" const LATEST_API = `https://api.github.com/repos/${REPO}/releases/latest` /** Throttle window for the startup check — hit GitHub at most once per day. */ export const CHECK_TTL_MS = 24 * 60 * 60 * 1000 +const CHECKSUM_TIMEOUT = Duration.seconds(30) +const DOWNLOAD_TIMEOUT = Duration.minutes(2) // --- pure helpers (unit-tested) ---------------------------------------------- @@ -102,33 +105,51 @@ const resolveTarget: Effect.Effect = Effect.suspend(() => { ) }) -/** Fetch the latest release tag from the GitHub API (keeps the leading "v" for - * URL building). Bounded by `timeoutMs` via AbortController. */ -export const fetchLatestTag = (timeoutMs = 5000): Effect.Effect => - Effect.tryPromise({ - try: async () => { - const controller = new AbortController() - const timer = setTimeout(() => controller.abort(), timeoutMs) - try { - // GitHub rejects requests without a User-Agent; Accept pins the API version. - const res = await fetch(LATEST_API, { - headers: { "User-Agent": "maple-cli", Accept: "application/vnd.github+json" }, - signal: controller.signal, - }) - if (!res.ok) throw new Error(`GitHub API returned ${res.status} ${res.statusText}`) - const body = (await res.json()) as { tag_name?: string } - if (!body.tag_name) throw new Error("no published release found") - return body.tag_name - } finally { - clearTimeout(timer) - } - }, - catch: (e) => - new UpdateError({ - message: `could not check for updates: ${e instanceof Error ? e.message : String(e)}`, - }), +const toUpdateError = (prefix: string, error: unknown): UpdateError => + new UpdateError({ + message: `${prefix}: ${error instanceof Error ? error.message : String(error)}`, }) +/** Fetch the latest release tag from the GitHub API (keeps the leading "v" for URL building). */ +export const fetchLatestTag = (timeoutMs = 5000): Effect.Effect => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient + const request = HttpClientRequest.get(LATEST_API, { + headers: { "User-Agent": "maple-cli", Accept: "application/vnd.github+json" }, + }) + const response = yield* client + .execute(request) + .pipe(Effect.mapError((error) => toUpdateError("could not check for updates", error))) + if (response.status < 200 || response.status >= 300) { + return yield* new UpdateError({ + message: `could not check for updates: GitHub API returned ${response.status}`, + }) + } + const text = yield* response.text.pipe( + Effect.mapError((error) => toUpdateError("could not read GitHub release response", error)), + ) + const body = yield* Effect.try({ + try: () => JSON.parse(text) as { tag_name?: string }, + catch: (error) => toUpdateError("could not decode GitHub release response", error), + }) + if (!body.tag_name) { + return yield* new UpdateError({ + message: "could not check for updates: no published release found", + }) + } + return body.tag_name + }).pipe( + Effect.timeoutOrElse({ + duration: Duration.millis(timeoutMs), + orElse: () => + Effect.fail( + new UpdateError({ + message: `could not check for updates: timed out after ${timeoutMs}ms`, + }), + ), + }), + ) + /** Directory holding the running `maple` binary and its sibling `libchdb.so` * (the symlink on PATH resolves here). */ const resolveInstallDir = (): string => dirname(realpathSync(process.execPath)) @@ -143,29 +164,62 @@ const mapFsError = (e: unknown, installDir: string): UpdateError => { return new UpdateError({ message: e instanceof Error ? e.message : String(e) }) } -const downloadTo = (url: string, dest: string): Effect.Effect => - Effect.tryPromise({ - try: async () => { - const res = await fetch(url, { headers: { "User-Agent": "maple-cli" } }) - if (!res.ok) throw new Error(`download failed (${res.status} ${res.statusText}) for ${url}`) - // NB: `Bun.write(dest, res)` (writing the Response directly) hangs - // indefinitely on GitHub's redirect-backed release-asset streams — it - // never resolves. Buffer the body first, then write the bytes. - const body = await res.arrayBuffer() - await Bun.write(dest, body) - }, - catch: (e) => new UpdateError({ message: e instanceof Error ? e.message : String(e) }), - }) +const downloadTo = ( + url: string, + dest: string, + timeout: Duration.Input = DOWNLOAD_TIMEOUT, +): Effect.Effect => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient + const response = yield* client + .execute(HttpClientRequest.get(url, { headers: { "User-Agent": "maple-cli" } })) + .pipe(Effect.mapError((error) => toUpdateError("release download failed", error))) + if (response.status < 200 || response.status >= 300) { + return yield* new UpdateError({ + message: `download failed (${response.status}) for ${url}`, + }) + } + // Buffer the redirect-backed release asset before Bun.write; passing the + // response stream directly can hang indefinitely. + const body = yield* response.arrayBuffer.pipe( + Effect.mapError((error) => toUpdateError("release download body failed", error)), + ) + yield* Effect.tryPromise({ + try: () => Bun.write(dest, body).then(() => undefined), + catch: (error) => toUpdateError("release download write failed", error), + }) + }).pipe( + Effect.timeoutOrElse({ + duration: timeout, + orElse: () => + Effect.fail(new UpdateError({ message: "release download timed out after 2 minutes" })), + }), + ) -const fetchText = (url: string): Effect.Effect => - Effect.tryPromise({ - try: async () => { - const res = await fetch(url, { headers: { "User-Agent": "maple-cli" } }) - if (!res.ok) throw new Error(`could not fetch ${url} (${res.status} ${res.statusText})`) - return await res.text() - }, - catch: (e) => new UpdateError({ message: e instanceof Error ? e.message : String(e) }), - }) +const fetchText = ( + url: string, + timeout: Duration.Input = CHECKSUM_TIMEOUT, +): Effect.Effect => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient + const response = yield* client + .execute(HttpClientRequest.get(url, { headers: { "User-Agent": "maple-cli" } })) + .pipe(Effect.mapError((error) => toUpdateError("checksum request failed", error))) + if (response.status < 200 || response.status >= 300) { + return yield* new UpdateError({ message: `could not fetch ${url} (${response.status})` }) + } + return yield* response.text.pipe( + Effect.mapError((error) => toUpdateError("checksum response read failed", error)), + ) + }).pipe( + Effect.timeoutOrElse({ + duration: timeout, + orElse: () => + Effect.fail(new UpdateError({ message: "checksum request timed out after 30 seconds" })), + }), + ) + +export const __testables = { downloadTo, fetchText } const sha256File = (path: string): Effect.Effect => Effect.tryPromise({ @@ -218,7 +272,9 @@ export interface UpdateResult { } /** Download, verify, and atomically install a release bundle in place. */ -export const performUpdate = (opts: { tag?: string } = {}): Effect.Effect => +export const performUpdate = ( + opts: { tag?: string } = {}, +): Effect.Effect => Effect.gen(function* () { const target = yield* resolveTarget const tagRaw = opts.tag ?? (yield* fetchLatestTag(10_000)) @@ -314,28 +370,30 @@ const printNotice = (current: string, latest: string): void => { * notice to stderr when a newer release exists. Never blocks fatally and never * fails the command — every error path collapses to a no-op. */ -export const maybeNotifyUpdate: Effect.Effect = Effect.gen(function* () { - if (!shouldRunNotify(process.argv)) return - const config = yield* MapleConfig - const now = yield* Clock.currentTimeMillis - let latest = config.latestKnownVersion - - if (shouldCheck(config.lastUpdateCheck, now)) { - const fetched = yield* fetchLatestTag(1500).pipe( - Effect.map((tag) => Option.some(tag)), - Effect.catch(() => Effect.succeed(Option.none())), - ) - if (Option.isSome(fetched)) { - latest = fetched.value - yield* config.recordUpdateCheck(fetched.value).pipe(Effect.ignore) - } else { - // Record the timestamp even on failure so an offline machine backs off - // for the TTL instead of re-probing GitHub on every command. - yield* config.recordUpdateCheck().pipe(Effect.ignore) +export const maybeNotifyUpdate: Effect.Effect = Effect.gen( + function* () { + if (!shouldRunNotify(process.argv)) return + const config = yield* MapleConfig + const now = yield* Clock.currentTimeMillis + let latest = Option.getOrUndefined(config.latestKnownVersion) + + if (shouldCheck(Option.getOrUndefined(config.lastUpdateCheck), now)) { + const fetched = yield* fetchLatestTag(1500).pipe( + Effect.map((tag) => Option.some(tag)), + Effect.catch(() => Effect.succeed(Option.none())), + ) + if (Option.isSome(fetched)) { + latest = fetched.value + yield* config.recordUpdateCheck(fetched.value).pipe(Effect.ignore) + } else { + // Record the timestamp even on failure so an offline machine backs off + // for the TTL instead of re-probing GitHub on every command. + yield* config.recordUpdateCheck().pipe(Effect.ignore) + } } - } - if (latest && isNewer(MAPLE_VERSION, latest)) { - yield* Effect.sync(() => printNotice(MAPLE_VERSION, stripV(latest))) - } -}) + if (latest && isNewer(MAPLE_VERSION, latest)) { + yield* Effect.sync(() => printNotice(MAPLE_VERSION, stripV(latest))) + } + }, +) diff --git a/apps/cli/src/core/warehouse.ts b/apps/cli/src/core/warehouse.ts index 679746134..0227f3291 100644 --- a/apps/cli/src/core/warehouse.ts +++ b/apps/cli/src/core/warehouse.ts @@ -1,4 +1,5 @@ import { Effect, Layer } from "effect" +import { HttpClient } from "effect/unstable/http" import { WarehouseExecutor, type WarehouseExecutorShape, @@ -29,13 +30,14 @@ export const WarehouseExecutorFromMode = Layer.effect( WarehouseExecutor, Effect.gen(function* () { const mode = yield* Mode + const client = yield* HttpClient.HttpClient const getShape = yield* Effect.cached( mode.resolve.pipe( Effect.map( (m): WarehouseExecutorShape => m._tag === "local" ? makeLocalWarehouseExecutorShape(m.baseUrl) - : makeRemoteWarehouseExecutorShape(m.apiUrl, m.token, m.orgId ?? ""), + : makeRemoteWarehouseExecutorShape(client, m.apiUrl, m.token, m.orgId ?? ""), ), Effect.mapError((e) => new WarehouseConfigError({ message: e.message, pipeName: "mode" })), ), diff --git a/apps/cli/src/server/checkpoints.ts b/apps/cli/src/server/checkpoints.ts index 43894bed2..1610946e7 100644 --- a/apps/cli/src/server/checkpoints.ts +++ b/apps/cli/src/server/checkpoints.ts @@ -5,6 +5,7 @@ import { cp, lstat, mkdir, readFile, readdir, rm, stat } from "node:fs/promises" import { tmpdir } from "node:os" import { basename, dirname, isAbsolute, join, relative, resolve, sep } from "node:path" import { Effect, Schema } from "effect" +import { HttpClient, HttpClientRequest } from "effect/unstable/http" import { CHDB_VERSION, MAPLE_VERSION } from "../version" import { serverUrl } from "../lib/local-address" import { Chdb } from "./chdb" @@ -419,29 +420,32 @@ const localQueryError = (status: number, detail: string, cause = detail): LocalQ export const checkpointQueryUrl = (host: string, port: number): string => `${serverUrl(host, port)}/local/query` -const postLocalQuery = (host: string, port: number, sql: string): Effect.Effect => { +const postLocalQuery = ( + host: string, + port: number, + sql: string, +): Effect.Effect => { const url = checkpointQueryUrl(host, port) return Effect.gen(function* () { - const response = yield* Effect.tryPromise({ - try: (signal) => - fetch(url, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ sql }), - signal, - }), - catch: (error) => localQueryError(0, errorMessage(error), errorCause(error)), - }) + const client = yield* HttpClient.HttpClient + const request = HttpClientRequest.post(url).pipe( + HttpClientRequest.bodyText(JSON.stringify({ sql }), "application/json"), + ) + const response = yield* client + .execute(request) + .pipe(Effect.mapError((error) => localQueryError(0, errorMessage(error), errorCause(error)))) yield* Effect.annotateCurrentSpan("http.response.status_code", response.status) - if (!response.ok) { - const detail = yield* Effect.tryPromise({ - try: () => response.text(), - catch: (error) => localQueryError(response.status, errorMessage(error), errorCause(error)), - }) + const text = yield* response.text.pipe( + Effect.mapError((error) => + localQueryError(response.status, errorMessage(error), errorCause(error)), + ), + ) + if (response.status < 200 || response.status >= 300) { + const detail = text return yield* localQueryError(response.status, detail) } - return yield* Effect.tryPromise({ - try: () => response.json() as Promise, + return yield* Effect.try({ + try: () => JSON.parse(text) as unknown, catch: (error) => localQueryError(response.status, errorMessage(error), errorCause(error)), }) }).pipe( @@ -949,12 +953,12 @@ const acquireMaintenance = async ( } } -const withMaintenance = ( +const withMaintenance = ( dataDir: string, operationId: CheckpointOperationId, onAcquireError: (error: unknown) => E, - use: () => Effect.Effect, -): Effect.Effect => + use: () => Effect.Effect, +): Effect.Effect => Effect.acquireRelease( Effect.tryPromise({ try: () => acquireMaintenance(dataDir, operationId), diff --git a/apps/cli/test/update.test.ts b/apps/cli/test/update.test.ts index 52cae23e2..594755418 100644 --- a/apps/cli/test/update.test.ts +++ b/apps/cli/test/update.test.ts @@ -1,6 +1,25 @@ import { describe, it } from "@effect/vitest" -import { strictEqual } from "node:assert" -import { CHECK_TTL_MS, isNewer, shouldCheck, stripV, targetTripleFor } from "../src/core/update" +import { ok, strictEqual } from "node:assert" +import { mkdtemp, readFile, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { Duration, Effect } from "effect" +import { FetchHttpClient } from "effect/unstable/http" +import { + __testables, + CHECK_TTL_MS, + fetchLatestTag, + isNewer, + shouldCheck, + stripV, + targetTripleFor, +} from "../src/core/update" + +const withFetch = (effect: Effect.Effect, fetchStub: typeof fetch) => + effect.pipe( + Effect.provide(FetchHttpClient.layer), + Effect.provideService(FetchHttpClient.Fetch, fetchStub), + ) describe("stripV", () => { it("drops a leading v", () => { @@ -79,3 +98,87 @@ describe("shouldCheck", () => { strictEqual(shouldCheck(exactlyTtl, now), true) }) }) + +describe("update HTTP", () => { + it.effect("decodes the latest release tag without mutating global fetch", () => + Effect.gen(function* () { + const fetchStub = (async () => Response.json({ tag_name: "v1.2.3" })) as typeof fetch + const tag = yield* withFetch(fetchLatestTag(), fetchStub) + strictEqual(tag, "v1.2.3") + }), + ) + + it.effect("maps non-2xx and decode failures to UpdateError", () => + Effect.gen(function* () { + const statusError = yield* Effect.flip( + withFetch( + fetchLatestTag(), + (async () => new Response("unavailable", { status: 503 })) as typeof fetch, + ), + ) + strictEqual(statusError._tag, "@maple/cli/UpdateError") + ok(statusError.message.includes("503")) + + const decodeError = yield* Effect.flip( + withFetch(fetchLatestTag(), (async () => new Response("not-json")) as typeof fetch), + ) + strictEqual(decodeError._tag, "@maple/cli/UpdateError") + ok(decodeError.message.includes("decode")) + }), + ) + + it("aborts a tag request at its configured deadline", async () => { + let aborted = false + const fetchStub = ((input: string | URL | Request, init?: RequestInit) => { + const signal = input instanceof Request ? input.signal : init?.signal + return new Promise((_resolve, reject) => { + signal?.addEventListener("abort", () => { + aborted = true + reject(new DOMException("aborted", "AbortError")) + }) + }) + }) as typeof fetch + const error = await Effect.runPromise(Effect.flip(withFetch(fetchLatestTag(5), fetchStub))) + ok(error.message.includes("timed out")) + strictEqual(aborted, true) + }) + + it("downloads release bytes and reads checksums through the shared client", async () => { + const dir = await mkdtemp(join(tmpdir(), "maple-update-http-")) + const destination = join(dir, "release.tar.gz") + const fetchStub = (async (input: string | URL | Request) => { + const url = input instanceof Request ? input.url : String(input) + return url.endsWith(".sha256") + ? new Response("abc123 release.tar.gz\n") + : new Response("release-bytes") + }) as typeof fetch + + try { + await Effect.runPromise( + withFetch( + __testables.downloadTo( + "https://release.test/release.tar.gz", + destination, + Duration.seconds(1), + ), + fetchStub, + ), + ) + strictEqual(await readFile(destination, "utf8"), "release-bytes") + strictEqual( + await Effect.runPromise( + withFetch( + __testables.fetchText( + "https://release.test/release.tar.gz.sha256", + Duration.seconds(1), + ), + fetchStub, + ), + ), + "abc123 release.tar.gz\n", + ) + } finally { + await rm(dir, { recursive: true, force: true }) + } + }) +}) diff --git a/apps/electric-sync/src/worker.ts b/apps/electric-sync/src/worker.ts index c27702c21..3bbcdb16a 100644 --- a/apps/electric-sync/src/worker.ts +++ b/apps/electric-sync/src/worker.ts @@ -1,5 +1,5 @@ import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" -import { ANTICIPATED_ERROR_TAGS } from "@maple/domain/anticipated-errors" +import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { WorkerConfigProviderLayer } from "@maple/effect-cloudflare" import { Context, Effect, FileSystem, Layer, Path } from "effect" import { HttpMiddleware, HttpRouter } from "effect/unstable/http" @@ -42,7 +42,7 @@ const telemetry = MapleCloudflareSDK.make({ serviceName: "electric-sync", serviceNamespace: "backend", repositoryUrl: "https://github.com/Makisuo/maple", - anticipatedErrorTags: [...ANTICIPATED_ERROR_TAGS], + anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS], }) // `HttpMiddleware.tracer` ends the root server span on a deferred macrotask, but diff --git a/apps/scraper/src/ApiClient.test.ts b/apps/scraper/src/ApiClient.test.ts index 4da5a84cc..597872e90 100644 --- a/apps/scraper/src/ApiClient.test.ts +++ b/apps/scraper/src/ApiClient.test.ts @@ -121,7 +121,10 @@ describe("ApiClient", () => { ), ) - assert.strictEqual(recorded[0]?.url, "http://api.test/api/internal/prometheus-scrape?targetId=target-1") + assert.strictEqual( + recorded[0]?.url, + "http://api.test/api/internal/prometheus-scrape?targetId=target-1", + ) assert.strictEqual(recorded[0]?.headers.authorization, "Bearer internal-token") assert.strictEqual(response.status, 200) assert.include(response.body, "up 1") @@ -169,4 +172,24 @@ describe("ApiClient", () => { ]) }).pipe(Effect.provide(TestLayer)), ) + + it.effect("surfaces a 503 result-persistence response so the scheduler can re-buffer", () => + Effect.gen(function* () { + const client = yield* ApiClient + const error = yield* client + .reportResults([ + { targetId: VALID_TARGET.id, scrapedAt: 1750000000000, error: null } as never, + ]) + .pipe( + Effect.provideService( + FetchHttpClient.Fetch, + stubFetch([], () => new Response("persistence unavailable", { status: 503 })), + ), + Effect.flip, + ) + + assert.strictEqual(error._tag, "@maple/scraper/ApiRequestError") + assert.strictEqual(error.status, 503) + }).pipe(Effect.provide(TestLayer)), + ) }) diff --git a/apps/web/src/components/chat/chat-apply-payload.test.ts b/apps/web/src/components/chat/chat-apply-payload.test.ts new file mode 100644 index 000000000..edd7c6170 --- /dev/null +++ b/apps/web/src/components/chat/chat-apply-payload.test.ts @@ -0,0 +1,13 @@ +import { describe, expect, it } from "vitest" +import { Schema } from "effect" +import { ChatApplyRequest } from "@maple/domain/http" +import { makeChatApplyPayload } from "./chat-apply-payload" + +describe("makeChatApplyPayload", () => { + it("constructs an encodable ChatApplyRequest", () => { + const payload = makeChatApplyPayload("update_dashboard_widget", { title: "Latency" }) + + expect(payload).toBeInstanceOf(ChatApplyRequest) + expect(() => Schema.encodeUnknownSync(ChatApplyRequest)(payload)).not.toThrow() + }) +}) diff --git a/apps/web/src/components/chat/chat-apply-payload.ts b/apps/web/src/components/chat/chat-apply-payload.ts new file mode 100644 index 000000000..1735d8066 --- /dev/null +++ b/apps/web/src/components/chat/chat-apply-payload.ts @@ -0,0 +1,4 @@ +import { ChatApplyRequest } from "@maple/domain/http" + +export const makeChatApplyPayload = (tool: string, input: unknown): ChatApplyRequest => + new ChatApplyRequest({ tool, input }) diff --git a/apps/web/src/components/chat/chat-conversation.tsx b/apps/web/src/components/chat/chat-conversation.tsx index 56fad82e4..f3e60c04b 100644 --- a/apps/web/src/components/chat/chat-conversation.tsx +++ b/apps/web/src/components/chat/chat-conversation.tsx @@ -46,6 +46,7 @@ import { ThinkingIndicator } from "@/components/ai-elements/thinking-indicator" import { Tool, ToolRow, toolLabel } from "@/components/ai-elements/tool" import { ToolGroup } from "@/components/ai-elements/tool-group" import { ApprovalCard } from "./approval-card" +import { makeChatApplyPayload } from "./chat-apply-payload" import type { UIMessage } from "@/components/ai-elements/types" type ToolPart = { @@ -177,7 +178,7 @@ export function ChatConversation({ return next }) const handleApprove = async (toolCallId: string, tool: string, input: unknown) => { - const exit = await applyProposal({ payload: { tool, input } }) + const exit = await applyProposal({ payload: makeChatApplyPayload(tool, input) }) if (Exit.isSuccess(exit)) { if (exit.value.isError) { toast.error(exit.value.content || `Couldn't apply ${tool}`) @@ -262,8 +263,9 @@ export function ChatConversation({ Ready to investigate

- The {investigationNoun(investigationContext!.kind)} above is attached to every message - in this thread. Start with a suggestion or ask your own question. + The {investigationNoun(investigationContext!.kind)} above is attached to + every message in this thread. Start with a suggestion or ask your own + question.

) : isWidgetFixMode ? ( diff --git a/apps/web/src/components/integrations/hazel-integration-card.tsx b/apps/web/src/components/integrations/hazel-integration-card.tsx index 16cc68a48..96b85d9d5 100644 --- a/apps/web/src/components/integrations/hazel-integration-card.tsx +++ b/apps/web/src/components/integrations/hazel-integration-card.tsx @@ -1,22 +1,25 @@ -import { useEffect, useState } from "react" -import { Exit } from "effect" +import { useState } from "react" +import { Exit, Option } from "effect" import { HazelStartConnectRequest } from "@maple/domain/http" import { Badge } from "@maple/ui/components/ui/badge" import { Button } from "@maple/ui/components/ui/button" +import { Skeleton } from "@maple/ui/components/ui/skeleton" import { toast } from "sonner" +import { ErrorState } from "@/components/common/error-state" import { BellIcon, ConnectionIcon, HazelIcon, LoaderIcon, ShieldIcon } from "@/components/icons" -import { Result, useAtomSet, useAtomValue } from "@/lib/effect-atom" +import { useMountEffect } from "@/hooks/use-mount-effect" +import { Result, useAtomRefresh, useAtomSet, useAtomValue } from "@/lib/effect-atom" import { MapleApiAtomClient } from "@/lib/services/common/atom-client" import { HAZEL_ACCENT, IntegrationIconPlate } from "./integration-catalog" import { IntegrationEmptyState } from "./integration-empty-state" export function HazelIntegrationCard() { - const statusResult = useAtomValue( - MapleApiAtomClient.query("integrations", "hazelStatus", { - reactivityKeys: ["hazelIntegrationStatus"], - }), - ) + const statusAtom = MapleApiAtomClient.query("integrations", "hazelStatus", { + reactivityKeys: ["hazelIntegrationStatus"], + }) + const statusResult = useAtomValue(statusAtom) + const refreshStatus = useAtomRefresh(statusAtom) const startConnect = useAtomSet(MapleApiAtomClient.mutation("integrations", "hazelStart"), { mode: "promiseExit", @@ -27,7 +30,7 @@ export function HazelIntegrationCard() { const [busy, setBusy] = useState<"connect" | "disconnect" | null>(null) - useEffect(() => { + useMountEffect(() => { function onMessage(event: MessageEvent) { if (event.data?.type === "maple:integration:hazel") { if (event.data.status === "success") { @@ -39,11 +42,17 @@ export function HazelIntegrationCard() { } window.addEventListener("message", onMessage) return () => window.removeEventListener("message", onMessage) - }, []) + }) const status = Result.builder(statusResult) .onSuccess((s) => s) - .orElse(() => null) + .orElse(() => + Result.isFailure(statusResult) + ? Option.getOrNull(Option.map(statusResult.previousSuccess, (previous) => previous.value)) + : null, + ) + const isLoading = Result.isInitial(statusResult) && status === null + const loadFailed = Result.isFailure(statusResult) && status === null async function handleConnect() { const popup = window.open("", "maple-hazel-connect", "popup,width=520,height=640") @@ -77,6 +86,18 @@ export function HazelIntegrationCard() { } const isConnected = status?.connected === true + if (isLoading) { + return + } + if (loadFailed) { + return ( + + ) + } if (!isConnected) { return ( @@ -127,8 +148,8 @@ export function HazelIntegrationCard() { Connected

- Forward Maple alerts into a Hazel workspace via OAuth. Once connected, create a - Hazel destination to pick which workspace receives notifications. + Forward Maple alerts into a Hazel workspace via OAuth. Once connected, create a Hazel + destination to pick which workspace receives notifications.

@@ -152,9 +173,7 @@ export function HazelIntegrationCard() { Reconnect diff --git a/apps/web/src/lib/services/common/otel-layer.ts b/apps/web/src/lib/services/common/otel-layer.ts index 1cc677bd7..526d432d9 100644 --- a/apps/web/src/lib/services/common/otel-layer.ts +++ b/apps/web/src/lib/services/common/otel-layer.ts @@ -1,5 +1,5 @@ import { MapleFlush } from "@maple-dev/effect-sdk/client" -import { ANTICIPATED_ERROR_TAGS } from "@maple/domain/anticipated-errors" +import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { ingestUrl } from "./ingest-url" // Buffer-backed client telemetry with flush-on-unload. `Maple.layer` @@ -27,7 +27,7 @@ const telemetry = MapleFlush.make({ }, // Expected 4xx API responses (the maple-web → maple-api edge surfaces these // as client-span failures) record as Ok instead of errors. - anticipatedErrorTags: [...ANTICIPATED_ERROR_TAGS], + anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS], // Maple's dashboard prioritizes operator responsiveness over self-replay. // Keep lightweight session metadata and trace/log correlation, but do not // run rrweb DOM capture, compression, and upload loops in every open tab. diff --git a/apps/web/src/lib/services/common/v2-pagination.test.ts b/apps/web/src/lib/services/common/v2-pagination.test.ts index 53ebbb9ce..8d970687a 100644 --- a/apps/web/src/lib/services/common/v2-pagination.test.ts +++ b/apps/web/src/lib/services/common/v2-pagination.test.ts @@ -3,12 +3,12 @@ import { Effect } from "effect" import { collectV2Pages } from "./v2-pagination" describe("collectV2Pages", () => { - it("collects every page without an item cap", async () => { - const itemCount = 2_005 - const pageSize = 100 + it.effect("collects every page without an item cap", () => + Effect.gen(function* () { + const itemCount = 2_005 + const pageSize = 100 - const result = await Effect.runPromise( - collectV2Pages((cursor) => { + const result = yield* collectV2Pages((cursor) => { const offset = cursor === undefined ? 0 : Number(cursor) const data = Array.from( { length: Math.min(pageSize, itemCount - offset) }, @@ -20,22 +20,22 @@ describe("collectV2Pages", () => { has_more: nextOffset < itemCount, next_cursor: nextOffset < itemCount ? String(nextOffset) : null, }) - }), - ) + }) - expect(result).toHaveLength(itemCount) - expect(result[0]).toBe(0) - expect(result.at(-1)).toBe(itemCount - 1) - }) + expect(result).toHaveLength(itemCount) + expect(result[0]).toBe(0) + expect(result.at(-1)).toBe(itemCount - 1) + }), + ) - it("fails with a specific tag when a cursor repeats", async () => { - const error = await Effect.runPromise( - Effect.flip( + it.effect("fails with a specific tag when a cursor repeats", () => + Effect.gen(function* () { + const error = yield* Effect.flip( collectV2Pages(() => Effect.succeed({ data: [1], has_more: true, next_cursor: "off_1" })), - ), - ) + ) - expect(error._tag).toBe("@maple/web/services/V2PaginationCursorLoopError") - expect(error.cursor).toBe("off_1") - }) + expect(error._tag).toBe("@maple/web/services/V2PaginationCursorLoopError") + expect(error.cursor).toBe("off_1") + }), + ) }) diff --git a/apps/web/src/lib/services/common/v2-pagination.ts b/apps/web/src/lib/services/common/v2-pagination.ts index 3beb89894..6ec83d8a0 100644 --- a/apps/web/src/lib/services/common/v2-pagination.ts +++ b/apps/web/src/lib/services/common/v2-pagination.ts @@ -10,8 +10,16 @@ export class V2PaginationCursorLoopError extends Schema.TaggedErrorClass
Loading… )} - {!loading && todos.length === 0 && ( + {Result.isFailure(listResult) && ( +
  • + Failed to load todos. + +
    • + )} + {Result.isSuccess(listResult) && todos.length === 0 && (
  • Nothing yet — add one above.
    • diff --git a/lib/effect-sdk/README.md b/lib/effect-sdk/README.md index 2b37c4a4c..c5b710e20 100644 --- a/lib/effect-sdk/README.md +++ b/lib/effect-sdk/README.md @@ -48,6 +48,8 @@ const telemetry = MapleCloudflareSDK.make({ serviceName: "my-worker", // Optional: drop noisy spans before they hit OTLP (prefix match). // dropSpanNames: ["McpServer/Notifications."], + // Optional: classify expected 4xx failures as successful spans. + // anticipatedErrorIdentifiers: ["@my-app/http/NotFoundError"], }) const handler = HttpRouter.toWebHandler(Routes.pipe(Layer.provideMerge(telemetry.layer))) @@ -63,16 +65,17 @@ export default { `telemetry.layer` MUST live in the same runtime as your routes — provide it to the layer composition you hand to `HttpRouter.toWebHandler`, not a separate per-request runtime, or your spans won't pick up the Tracer reference. -When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, each signal sleeps 60s before retrying so a broken collector doesn't get hammered. +When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, the failed batch is restored ahead of newer telemetry and that signal sleeps 60s before retrying. Trace and log cooldowns are independent. ### Cloudflare-specific options -| Option | Description | -| ----------------- | ----------------------------------------------------------------------------------------------------------- | -| `dropSpanNames` | Span names whose prefix matches an entry are dropped before OTLP export (e.g. `"McpServer/Notifications."`) | -| `excludeLogSpans` | Skip Effect log spans in OTLP log attributes. Default `false` | -| `tracesPath` | OTLP traces path appended to `endpoint`. Default `/v1/traces` | -| `logsPath` | OTLP logs path appended to `endpoint`. Default `/v1/logs` | +| Option | Description | +| ----------------------------- | ----------------------------------------------------------------------------------------------------------- | +| `anticipatedErrorIdentifiers` | Stable `_tag` / `Error.name` identifiers for expected 4xx failures; exported as `Ok` without an exception | +| `dropSpanNames` | Span names whose prefix matches an entry are dropped before OTLP export (e.g. `"McpServer/Notifications."`) | +| `excludeLogSpans` | Skip Effect log spans in OTLP log attributes. Default `false` | +| `tracesPath` | OTLP traces path appended to `endpoint`. Default `/v1/traces` | +| `logsPath` | OTLP logs path appended to `endpoint`. Default `/v1/logs` | The same `MAPLE_ENDPOINT` / `MAPLE_INGEST_KEY` / `MAPLE_ENVIRONMENT` env vars apply, read from the Workers `env` binding. diff --git a/lib/effect-sdk/src/client/flushable.ts b/lib/effect-sdk/src/client/flushable.ts index 80e098de6..3c64a5cd4 100644 --- a/lib/effect-sdk/src/client/flushable.ts +++ b/lib/effect-sdk/src/client/flushable.ts @@ -30,6 +30,7 @@ import { Layer, Redacted } from "effect" import { buildResolved, type FlushTransport, + makeSerializedFlush, type Resolved, type ResourceInput, runFlush, @@ -61,10 +62,12 @@ export interface MapleClientFlushableConfig { /** Span name prefixes to drop before OTLP export. */ readonly dropSpanNames?: ReadonlyArray | undefined /** - * `_tag`s of *anticipated* failures (expected 4xx business errors). Spans + * Stable `_tag` / `Error.name` identifiers of anticipated 4xx failures. Spans * failing entirely with these export as status `Ok` (no `exception` event), * so they stay visible but never count as errors. */ + readonly anticipatedErrorIdentifiers?: ReadonlyArray | undefined + /** @deprecated Use `anticipatedErrorIdentifiers`. */ readonly anticipatedErrorTags?: ReadonlyArray | undefined /** OTLP traces path appended to `endpoint`. Default `/v1/traces`. */ readonly tracesPath?: string | undefined @@ -154,10 +157,12 @@ export const make = (config: MapleClientFlushableConfig): FlushableTelemetry => dropPrefixes !== undefined && dropPrefixes.length > 0 ? (name: string) => dropPrefixes.some((prefix) => name.startsWith(prefix)) : undefined - const anticipatedErrorTags = - config.anticipatedErrorTags !== undefined && config.anticipatedErrorTags.length > 0 - ? new Set(config.anticipatedErrorTags) - : undefined + const anticipatedErrorIdentifiers = [ + ...(config.anticipatedErrorIdentifiers ?? []), + ...(config.anticipatedErrorTags ?? []), + ] + const anticipatedIdentifiers = + anticipatedErrorIdentifiers.length > 0 ? new Set(anticipatedErrorIdentifiers) : undefined startClientSession({ endpoint: config.endpoint, ingestKey: config.ingestKey, @@ -168,7 +173,10 @@ export const make = (config: MapleClientFlushableConfig): FlushableTelemetry => emitSessionMeta: config.emitSessionMeta, }) - const spans: SpanBuffer = makeSpanBuffer({ dropSpan, anticipatedErrorTags }) + const spans: SpanBuffer = makeSpanBuffer({ + dropSpan, + anticipatedErrorIdentifiers: anticipatedIdentifiers, + }) const logs: LogBuffer = makeLogBuffer({ excludeLogSpans: config.excludeLogSpans }) // `withSessionLink` overrides only the Tracer reference, keeping the logger. const layer = withSessionLink(Layer.mergeAll(spans.tracerLayer, logs.loggerLayer)) @@ -195,7 +203,7 @@ export const make = (config: MapleClientFlushableConfig): FlushableTelemetry => const logsState: SignalState = { disabledUntil: 0 } let noOpLogged = false - const flush = async (): Promise => { + const flush = makeSerializedFlush(async (): Promise => { await runFlush({ resolved, spans, @@ -213,7 +221,7 @@ export const make = (config: MapleClientFlushableConfig): FlushableTelemetry => } }, }) - } + }) const intervalMs = config.autoFlushInterval === undefined diff --git a/lib/effect-sdk/src/cloudflare/index.ts b/lib/effect-sdk/src/cloudflare/index.ts index 090806cc2..ba41a6b8e 100644 --- a/lib/effect-sdk/src/cloudflare/index.ts +++ b/lib/effect-sdk/src/cloudflare/index.ts @@ -35,6 +35,7 @@ import { Layer } from "effect" import { buildResolved, fetchTransport, + makeSerializedFlush, type Resolved, runFlush, type SignalState, @@ -84,11 +85,13 @@ export interface Config { */ readonly dropSpanNames?: ReadonlyArray | undefined /** - * `_tag`s of *anticipated* failures (expected 4xx business errors). A span + * Stable `_tag` / `Error.name` identifiers of anticipated 4xx failures. A span * whose failure is caused entirely by these is exported with status `Ok` and * no `exception` event, so it stays visible as a trace but never counts as an - * error. See `SpanBufferOptions.anticipatedErrorTags`. + * error. */ + readonly anticipatedErrorIdentifiers?: ReadonlyArray | undefined + /** @deprecated Use `anticipatedErrorIdentifiers`. */ readonly anticipatedErrorTags?: ReadonlyArray | undefined /** OTLP traces path appended to `endpoint`. Default `/v1/traces`. */ readonly tracesPath?: string | undefined @@ -133,11 +136,16 @@ export const make = (config: Config = {}): Telemetry => { dropPrefixes !== undefined && dropPrefixes.length > 0 ? (name: string) => dropPrefixes.some((prefix) => name.startsWith(prefix)) : undefined - const anticipatedErrorTags = - config.anticipatedErrorTags !== undefined && config.anticipatedErrorTags.length > 0 - ? new Set(config.anticipatedErrorTags) - : undefined - const spans: SpanBuffer = makeSpanBuffer({ dropSpan, anticipatedErrorTags }) + const anticipatedErrorIdentifiers = [ + ...(config.anticipatedErrorIdentifiers ?? []), + ...(config.anticipatedErrorTags ?? []), + ] + const anticipatedIdentifiers = + anticipatedErrorIdentifiers.length > 0 ? new Set(anticipatedErrorIdentifiers) : undefined + const spans: SpanBuffer = makeSpanBuffer({ + dropSpan, + anticipatedErrorIdentifiers: anticipatedIdentifiers, + }) const logs: LogBuffer = makeLogBuffer({ excludeLogSpans: config.excludeLogSpans }) let resolved: Resolved | undefined = undefined @@ -147,7 +155,7 @@ export const make = (config: Config = {}): Telemetry => { const layer = Layer.mergeAll(spans.tracerLayer, logs.loggerLayer) - const flush = async (env: Record): Promise => { + const flush = makeSerializedFlush(async (env: Record): Promise => { if (resolved === undefined) { resolved = resolveOnce(env, config) } @@ -169,7 +177,7 @@ export const make = (config: Config = {}): Telemetry => { } }, }) - } + }) return { layer, flush } } diff --git a/lib/effect-sdk/src/server/flushable.ts b/lib/effect-sdk/src/server/flushable.ts index 8d6933f0d..25ca7ad49 100644 --- a/lib/effect-sdk/src/server/flushable.ts +++ b/lib/effect-sdk/src/server/flushable.ts @@ -21,6 +21,7 @@ import { Effect, Layer } from "effect" import { buildResolved, fetchTransport, + makeSerializedFlush, type Resolved, runFlush, type SignalState, @@ -61,10 +62,12 @@ export interface MapleFlushableConfig { /** Span name prefixes to drop before OTLP export. */ readonly dropSpanNames?: ReadonlyArray | undefined /** - * `_tag`s of *anticipated* failures (expected 4xx business errors). Spans + * Stable `_tag` / `Error.name` identifiers of anticipated 4xx failures. Spans * failing entirely with these export as status `Ok` (no `exception` event), * so they stay visible but never count as errors. */ + readonly anticipatedErrorIdentifiers?: ReadonlyArray | undefined + /** @deprecated Use `anticipatedErrorIdentifiers`. */ readonly anticipatedErrorTags?: ReadonlyArray | undefined /** OTLP traces path appended to `endpoint`. Default `/v1/traces`. */ readonly tracesPath?: string | undefined @@ -98,11 +101,16 @@ export const make = (config: MapleFlushableConfig = {}): FlushableTelemetry => { dropPrefixes !== undefined && dropPrefixes.length > 0 ? (name: string) => dropPrefixes.some((prefix) => name.startsWith(prefix)) : undefined - const anticipatedErrorTags = - config.anticipatedErrorTags !== undefined && config.anticipatedErrorTags.length > 0 - ? new Set(config.anticipatedErrorTags) - : undefined - const spans: SpanBuffer = makeSpanBuffer({ dropSpan, anticipatedErrorTags }) + const anticipatedErrorIdentifiers = [ + ...(config.anticipatedErrorIdentifiers ?? []), + ...(config.anticipatedErrorTags ?? []), + ] + const anticipatedIdentifiers = + anticipatedErrorIdentifiers.length > 0 ? new Set(anticipatedErrorIdentifiers) : undefined + const spans: SpanBuffer = makeSpanBuffer({ + dropSpan, + anticipatedErrorIdentifiers: anticipatedIdentifiers, + }) const logs: LogBuffer = makeLogBuffer({ excludeLogSpans: config.excludeLogSpans }) const layer = Layer.mergeAll(spans.tracerLayer, logs.loggerLayer) @@ -129,7 +137,7 @@ export const make = (config: MapleFlushableConfig = {}): FlushableTelemetry => { return resolvedPromise } - const flush = async (): Promise => { + const flush = makeSerializedFlush(async (): Promise => { const resolved = await ensureResolved() await runFlush({ resolved, @@ -148,7 +156,7 @@ export const make = (config: MapleFlushableConfig = {}): FlushableTelemetry => { } }, }) - } + }) const intervalMs = config.autoFlushInterval === undefined diff --git a/lib/effect-sdk/src/shared/flush-core.test.ts b/lib/effect-sdk/src/shared/flush-core.test.ts new file mode 100644 index 000000000..87f323fdb --- /dev/null +++ b/lib/effect-sdk/src/shared/flush-core.test.ts @@ -0,0 +1,113 @@ +import { afterEach, describe, expect, it, vi } from "vitest" +import { Effect, Redacted } from "effect" +import { + buildResolved, + makeSerializedFlush, + runFlush, + type FlushTransport, + type SignalState, +} from "./flush-core.js" +import { makeLogBuffer } from "./flushable-logger.js" +import { makeSpanBuffer } from "./flushable-tracer.js" + +const resolved = buildResolved( + { + endpoint: "https://collector.test", + ingestKey: Redacted.make("test-key"), + resource: { serviceName: "test", serviceVersion: undefined, attributes: {} }, + }, + { userAgent: "test" }, +) + +const recordSpan = (spans: ReturnType, name: string) => + Effect.runPromise(Effect.succeed(undefined).pipe(Effect.withSpan(name), Effect.provide(spans.tracerLayer))) + +const recordLog = (logs: ReturnType, message: string) => + Effect.runPromise(Effect.logInfo(message).pipe(Effect.provide(logs.loggerLayer))) + +describe("runFlush", () => { + afterEach(() => { + vi.useRealTimers() + }) + + it("retains failed and cooldown batches while allowing the other signal to succeed", async () => { + vi.useFakeTimers() + vi.setSystemTime(new Date("2026-01-01T00:00:00Z")) + const spans = makeSpanBuffer() + const logs = makeLogBuffer() + const tracesState: SignalState = { disabledUntil: 0 } + const logsState: SignalState = { disabledUntil: 0 } + const traceBodies: unknown[] = [] + const logBodies: unknown[] = [] + let failTraces = true + const transport: FlushTransport = { + post: async (url, _headers, body) => { + if (url.endsWith("/v1/traces")) { + if (failTraces) throw new Error("collector unavailable") + traceBodies.push(body) + } else { + logBodies.push(body) + } + }, + } + const flush = () => + runFlush({ + resolved, + spans, + logs, + tracesState, + logsState, + transport, + logPrefix: "[test]", + onNoOp: () => undefined, + }) + + await recordSpan(spans, "first") + await recordLog(logs, "first-log") + await flush() + expect(spans.size()).toBe(1) + expect(logs.size()).toBe(0) + expect(logBodies).toHaveLength(1) + + await recordSpan(spans, "second") + await flush() + expect(spans.size()).toBe(2) + expect(traceBodies).toHaveLength(0) + + failTraces = false + vi.advanceTimersByTime(60_000) + await flush() + expect(spans.size()).toBe(0) + expect(traceBodies).toHaveLength(1) + const body = traceBodies[0] as { + resourceSpans: Array<{ scopeSpans: Array<{ spans: Array<{ name: string }> }> }> + } + expect(body.resourceSpans[0]!.scopeSpans[0]!.spans.map((span) => span.name)).toEqual([ + "first", + "second", + ]) + }) + + it("serializes overlapping flush calls", async () => { + let active = 0 + let peak = 0 + let release: (() => void) | undefined + const gate = new Promise((resolve) => { + release = resolve + }) + const run = makeSerializedFlush(async () => { + active += 1 + peak = Math.max(peak, active) + await gate + active -= 1 + }) + + const first = run() + const second = run() + await Promise.resolve() + expect(peak).toBe(1) + release?.() + await Promise.all([first, second]) + expect(peak).toBe(1) + }) +}) diff --git a/lib/effect-sdk/src/shared/flush-core.ts b/lib/effect-sdk/src/shared/flush-core.ts index dfaed0809..06f86f300 100644 --- a/lib/effect-sdk/src/shared/flush-core.ts +++ b/lib/effect-sdk/src/shared/flush-core.ts @@ -136,17 +136,17 @@ const post = async (url: string, headers: Record, body: unknown) /** Default transport: plain `fetch` (server + Cloudflare). */ export const fetchTransport: FlushTransport = { post } -const flushSignal = async ( - url: string, - headers: Record, - body: () => unknown, - state: SignalState, - count: number, - signal: string, - transport: FlushTransport, - logPrefix: string, -): Promise => { - if (count === 0) return +const flushSignal = async     (args: { + readonly url: string + readonly headers: Record + readonly buffer: { readonly drain: () => Array; readonly restore: (items: ReadonlyArray) => void } + readonly body: (items: ReadonlyArray) => unknown + readonly state: SignalState + readonly signal: string + readonly transport: FlushTransport + readonly logPrefix: string +}): Promise => { + const { url, headers, buffer, body, state, signal, transport, logPrefix } = args if (state.disabledUntil && Date.now() < state.disabledUntil) { console.warn( `${logPrefix} ${signal} flush skipped (cooldown ${state.disabledUntil - Date.now()}ms remaining)`, @@ -154,14 +154,32 @@ const flushSignal = async ( return } state.disabledUntil = 0 + const batch = buffer.drain() + if (batch.length === 0) return try { - await transport.post(url, headers, body()) + await transport.post(url, headers, body(batch)) } catch (err) { + buffer.restore(batch) state.disabledUntil = Date.now() + COOLDOWN_MS console.error(`${logPrefix} ${signal} flush failed; cooldown 60s:`, err) } } +/** Serialize flush calls so concurrent timers/manual hooks cannot drain overlapping batches. */ +export const makeSerializedFlush = >( + run: (...args: Args) => Promise, +): ((...args: Args) => Promise) => { + let tail: Promise = Promise.resolve() + return (...args) => { + const next = tail.then( + () => run(...args), + () => run(...args), + ) + tail = next.catch(() => undefined) + return next + } +} + /** * Drain the span + log buffers and POST them. Errors are swallowed per signal * (see {@link flushSignal}); the returned promise never rejects. @@ -189,31 +207,27 @@ export const runFlush = async (args: { return } - const spanBatch = spans.drain() - const logBatch = logs.drain() - if (spanBatch.length === 0 && logBatch.length === 0) return - await Promise.all([ - flushSignal( - r.tracesUrl, - r.headers, - () => makeTracesBody(spanBatch, r), - tracesState, - spanBatch.length, - "traces", + flushSignal({ + url: r.tracesUrl, + headers: r.headers, + buffer: spans, + body: (items) => makeTracesBody(items, r), + state: tracesState, + signal: "traces", transport, logPrefix, - ), - flushSignal( - r.logsUrl, - r.headers, - () => makeLogsBody(logBatch, r), - logsState, - logBatch.length, - "logs", + }), + flushSignal({ + url: r.logsUrl, + headers: r.headers, + buffer: logs, + body: (items) => makeLogsBody(items, r), + state: logsState, + signal: "logs", transport, logPrefix, - ), + }), ]) } diff --git a/lib/effect-sdk/src/shared/flushable-logger.ts b/lib/effect-sdk/src/shared/flushable-logger.ts index e50274d40..26e31d045 100644 --- a/lib/effect-sdk/src/shared/flushable-logger.ts +++ b/lib/effect-sdk/src/shared/flushable-logger.ts @@ -11,6 +11,7 @@ import * as OtlpResource from "effect/unstable/observability/OtlpResource" export interface LogBuffer { readonly loggerLayer: Layer.Layer readonly drain: () => Array + readonly restore: (items: ReadonlyArray) => void readonly setDisabled: (value: boolean) => void readonly size: () => number } @@ -35,6 +36,10 @@ export const makeLogBuffer = (options: { readonly excludeLogSpans?: boolean } = buffer = [] return items }, + restore: (items) => { + if (disabled || items.length === 0) return + buffer = [...items, ...buffer].slice(0, MAX_BUFFER) + }, setDisabled: (value) => { disabled = value if (value) buffer = [] diff --git a/lib/effect-sdk/src/shared/flushable-tracer.test.ts b/lib/effect-sdk/src/shared/flushable-tracer.test.ts index be2e4aa48..0e543ff5d 100644 --- a/lib/effect-sdk/src/shared/flushable-tracer.test.ts +++ b/lib/effect-sdk/src/shared/flushable-tracer.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest" -import { Data, Effect } from "effect" +import { Data, Effect, Schema } from "effect" import * as ErrorReporter from "effect/ErrorReporter" import * as HttpServerError from "effect/unstable/http/HttpServerError" import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest" @@ -14,6 +14,9 @@ class ReportableError extends Data.TaggedError("ReportableError")<{}> {} // An anticipated 4xx business error (e.g. unauthorized / not-found). class UnauthorizedError extends Data.TaggedError("UnauthorizedError")<{}> {} +class V2InvalidRequestError extends Schema.ErrorClass( + "@maple/http/v2/InvalidRequestError", +)({ error: Schema.Struct({ type: Schema.Literal("invalid_request_error") }) }) {} const runSpan = (buffer: ReturnType, effect: Effect.Effect) => effect.pipe(Effect.withSpan("http.server GET"), Effect.provide(buffer.tracerLayer), Effect.exit) @@ -43,6 +46,14 @@ describe("makeSpanBuffer ignored-failure drop", () => { }), ) + it.effect("keeps a mixed ignored failure and defect", () => + Effect.gen(function* () { + const buffer = makeSpanBuffer() + yield* runSpan(buffer, Effect.fail(new BenignError()).pipe(Effect.ensuring(Effect.die("boom")))) + assert.strictEqual(buffer.size(), 1) + }), + ) + // Pins the upstream contract: the actual error HttpRouter raises for an // unmatched route must stay [ErrorReporter.ignore]-flagged, so the drop holds. it.effect("drops the real HttpServerError/RouteNotFound", () => @@ -75,6 +86,25 @@ describe("makeSpanBuffer anticipated-error classification", () => { }), ) + it.effect("classifies Schema.ErrorClass failures by Error.name", () => + Effect.gen(function* () { + const buffer = makeSpanBuffer({ + anticipatedErrorIdentifiers: new Set(["@maple/http/v2/InvalidRequestError"]), + }) + yield* runSpan( + buffer, + Effect.fail(new V2InvalidRequestError({ error: { type: "invalid_request_error" } })), + ) + const [span] = buffer.drain() + assert.isDefined(span) + assert.strictEqual(span!.status.code, 1 /* Ok */) + assert.strictEqual( + span!.events.some((event) => event.name === "exception"), + false, + ) + }), + ) + it.effect("still marks an unclassified failure as an Error span with an exception event", () => Effect.gen(function* () { const buffer = makeSpanBuffer({ anticipatedErrorTags: tags }) @@ -132,3 +162,26 @@ describe("makeSpanBuffer anticipated-error classification", () => { }), ) }) + +describe("makeSpanBuffer restore", () => { + it.effect("keeps older failed telemetry and discards newest overflow", () => + Effect.gen(function* () { + const buffer = makeSpanBuffer() + yield* runSpan(buffer, Effect.succeed("old")) + const [oldest] = buffer.drain() + yield* Effect.succeed(undefined).pipe( + Effect.withSpan("newest"), + Effect.provide(buffer.tracerLayer), + ) + const [newest] = buffer.drain() + assert.isDefined(oldest) + assert.isDefined(newest) + + buffer.restore([oldest!, ...Array.from({ length: 10_000 }, () => newest!)]) + const restored = buffer.drain() + assert.strictEqual(restored.length, 10_000) + assert.strictEqual(restored[0]?.name, "http.server GET") + assert.strictEqual(restored.at(-1)?.name, "newest") + }), + ) +}) diff --git a/lib/effect-sdk/src/shared/flushable-tracer.ts b/lib/effect-sdk/src/shared/flushable-tracer.ts index 21ebb7515..2a5e6b6c6 100644 --- a/lib/effect-sdk/src/shared/flushable-tracer.ts +++ b/lib/effect-sdk/src/shared/flushable-tracer.ts @@ -14,6 +14,7 @@ import type { ExtractTag } from "effect/Types" export interface SpanBuffer { readonly tracerLayer: Layer.Layer readonly drain: () => Array + readonly restore: (items: ReadonlyArray) => void readonly setDisabled: (value: boolean) => void readonly size: () => number } @@ -28,9 +29,10 @@ export interface SpanBufferOptions { */ readonly dropSpan?: ((name: string) => boolean) | undefined /** - * `_tag`s of failures that represent *anticipated* outcomes (expected 4xx - * business errors: validation, not-found, unauthorized, …). When a span's - * failure is caused *entirely* by errors with one of these tags, the span is + * Stable `_tag` or `Error.name` identifiers for failures that represent + * anticipated outcomes (expected 4xx business errors: validation, not-found, + * unauthorized, …). When a span's failure is caused *entirely* by errors with + * one of these identifiers, the span is * still exported (so latency / `http.response.status_code` stay visible) but * with OTLP status `Ok` and **no** `exception` event — so it never lands in * error tracking (`error_events_mv` keys off `StatusCode='Error'`). Mirrors @@ -39,6 +41,8 @@ export interface SpanBufferOptions { * Distinct from Effect's `ErrorReporter.ignore` flag, which *drops* the span * entirely (used for benign routing 404s). */ + readonly anticipatedErrorIdentifiers?: ReadonlySet | undefined + /** @deprecated Use `anticipatedErrorIdentifiers`. */ readonly anticipatedErrorTags?: ReadonlySet | undefined } @@ -56,15 +60,16 @@ const isIgnoredSpan = (span: SpanImpl): boolean => { if (status._tag !== "Ended") return false const exit = status.exit if (exit._tag !== "Failure") return false - // Cause is flat in effect v4: walk `reasons`, pick Fail reasons, inspect their error. - return exit.cause.reasons.some((reason) => Cause.isFailReason(reason) && isIgnoredFailure(reason.error)) + if (exit.cause.reasons.some(Cause.isDieReason)) return false + const failures = exit.cause.reasons.filter(Cause.isFailReason) + return failures.length > 0 && failures.every((reason) => isIgnoredFailure(reason.error)) } export const makeSpanBuffer = (options: SpanBufferOptions = {}): SpanBuffer => { let buffer: Array = [] let disabled = false const dropSpan = options.dropSpan - const anticipatedErrorTags = options.anticipatedErrorTags + const anticipatedErrorIdentifiers = options.anticipatedErrorIdentifiers ?? options.anticipatedErrorTags const exportFn = (span: SpanImpl) => { if (disabled) return @@ -72,7 +77,7 @@ export const makeSpanBuffer = (options: SpanBufferOptions = {}): SpanBuffer => { if (dropSpan !== undefined && dropSpan(span.name)) return if (isIgnoredSpan(span)) return if (buffer.length >= MAX_BUFFER) return - buffer.push(makeOtlpSpan(span, anticipatedErrorTags)) + buffer.push(makeOtlpSpan(span, anticipatedErrorIdentifiers)) } const tracer = Tracer.make({ @@ -93,6 +98,10 @@ export const makeSpanBuffer = (options: SpanBufferOptions = {}): SpanBuffer => { buffer = [] return items }, + restore: (items) => { + if (disabled || items.length === 0) return + buffer = [...items, ...buffer].slice(0, MAX_BUFFER) + }, setDisabled: (value) => { disabled = value if (value) buffer = [] @@ -163,20 +172,28 @@ const generateId = (len: number): string => { // A failure is "anticipated" when its `_tag` is in the configured set. A span // whose failure is caused *entirely* by anticipated errors (no defects/Die) // records OTLP status `Ok` and emits no `exception` event. -const isAnticipatedFailure = (error: unknown, tags: ReadonlySet): boolean => - Predicate.hasProperty(error, "_tag") && typeof error._tag === "string" && tags.has(error._tag) +const failureIdentifier = (error: unknown): string | undefined => { + if (Predicate.hasProperty(error, "_tag") && typeof error._tag === "string") return error._tag + if (Predicate.hasProperty(error, "name") && typeof error.name === "string") return error.name + return undefined +} + +const isAnticipatedFailure = (error: unknown, identifiers: ReadonlySet): boolean => { + const identifier = failureIdentifier(error) + return identifier !== undefined && identifiers.has(identifier) +} const isFullyAnticipated = ( cause: Cause.Cause, - tags: ReadonlySet | undefined, + identifiers: ReadonlySet | undefined, ): boolean => { - if (tags === undefined || tags.size === 0) return false + if (identifiers === undefined || identifiers.size === 0) return false if (cause.reasons.some(Cause.isDieReason)) return false const failErrors = cause.reasons.filter(Cause.isFailReason).map((reason) => reason.error) - return failErrors.length > 0 && failErrors.every((error) => isAnticipatedFailure(error, tags)) + return failErrors.length > 0 && failErrors.every((error) => isAnticipatedFailure(error, identifiers)) } -const makeOtlpSpan = (self: SpanImpl, anticipatedErrorTags?: ReadonlySet): OtlpSpan => { +const makeOtlpSpan = (self: SpanImpl, anticipatedErrorIdentifiers?: ReadonlySet): OtlpSpan => { const status = self.status as ExtractTag const attributes = OtlpResource.entriesToAttributes(self.attributes.entries()) const events = self.events.map(([name, startTime, attrs]) => ({ @@ -195,7 +212,7 @@ const makeOtlpSpan = (self: SpanImpl, anticipatedErrorTags?: ReadonlySet { key: "span.label", value: { stringValue: "⚠︎ Interrupted" } }, { key: "status.interrupted", value: { boolValue: true } }, ) - } else if (isFullyAnticipated(status.exit.cause, anticipatedErrorTags)) { + } else if (isFullyAnticipated(status.exit.cause, anticipatedErrorIdentifiers)) { // Expected business outcome (4xx). Keep the span (latency / status code // stay visible) but don't flag it as an error or fingerprint it. otelStatus = constOtelStatusSuccess diff --git a/packages/domain/src/anticipated-errors.test.ts b/packages/domain/src/anticipated-errors.test.ts index 7b10c7cb3..6b2fd100f 100644 --- a/packages/domain/src/anticipated-errors.test.ts +++ b/packages/domain/src/anticipated-errors.test.ts @@ -1,28 +1,33 @@ import { describe, expect, it } from "vitest" -import { ANTICIPATED_ERROR_TAGS, isAnticipatedErrorTag } from "./anticipated-errors" +import { ANTICIPATED_ERROR_IDENTIFIERS, isAnticipatedErrorIdentifier } from "./anticipated-errors" -describe("ANTICIPATED_ERROR_TAGS", () => { - it("includes the observed 4xx business-error tags", () => { - for (const tag of [ +describe("ANTICIPATED_ERROR_IDENTIFIERS", () => { + it("includes legacy tags and v2 ErrorClass names for 4xx business errors", () => { + for (const identifier of [ "@maple/http/errors/UnauthorizedError", "@maple/http/errors/RawSqlValidationError", "@maple/http/ai-triage/AiTriageNotFoundError", "@maple/http/errors/IntegrationsNotConnectedError", + "@maple/http/v2/InvalidRequestError", + "@maple/http/v2/AuthenticationError", + "@maple/http/v2/RateLimitError", ]) { - expect(isAnticipatedErrorTag(tag), tag).toBe(true) + expect(isAnticipatedErrorIdentifier(identifier), identifier).toBe(true) } }) it("excludes 5xx persistence / upstream failures", () => { - for (const tag of [ + for (const identifier of [ "@maple/http/errors/WarehouseQueryError", "@maple/http/errors/QueryEngineTimeoutError", + "@maple/http/v2/ApiError", + "@maple/http/v2/ServiceUnavailableError", ]) { - expect(isAnticipatedErrorTag(tag), tag).toBe(false) + expect(isAnticipatedErrorIdentifier(identifier), identifier).toBe(false) } }) it("derives a non-trivial set (reflection still works)", () => { - expect(ANTICIPATED_ERROR_TAGS.size).toBeGreaterThan(20) + expect(ANTICIPATED_ERROR_IDENTIFIERS.size).toBeGreaterThan(25) }) }) diff --git a/packages/domain/src/anticipated-errors.ts b/packages/domain/src/anticipated-errors.ts index 5995533c0..1675ba22c 100644 --- a/packages/domain/src/anticipated-errors.ts +++ b/packages/domain/src/anticipated-errors.ts @@ -1,8 +1,10 @@ // --------------------------------------------------------------------------- -// Anticipated error tags +// Anticipated error identifiers // -// The set of domain HTTP error `_tag`s that represent *expected* client-facing -// outcomes (4xx): validation, not-found, unauthorized, forbidden, conflict, … +// The set of stable domain HTTP error identifiers that represent *expected* +// client-facing outcomes (4xx): validation, not-found, unauthorized, forbidden, +// conflict, … Tagged errors contribute `_tag`; v2 ErrorClass values contribute +// their class identifier / `Error.name`. // // These are not bugs — they're normal business results. The telemetry SDK uses // this set to record spans that fail *entirely* with one of these errors as @@ -12,12 +14,13 @@ // rule (4xx → Ok, 5xx → Error). // // Derived (not hand-maintained) from the error classes themselves: every -// `Schema.TaggedErrorClass` carries its `_tag` literal and an `httpApiStatus` -// annotation on its AST, so a new 4xx error is picked up automatically. A 5xx -// error (persistence/upstream failures) is intentionally excluded and keeps -// tracing. +// Both `Schema.TaggedErrorClass` and `Schema.ErrorClass` carry a stable +// identifier plus an `httpApiStatus` annotation, so a new 4xx error is picked +// up automatically. A 5xx error (persistence/upstream failures) is intentionally +// excluded and keeps tracing. // --------------------------------------------------------------------------- import * as Http from "./http/index" +import * as HttpV2 from "./http/v2/index" /** Read `obj[key]` when `obj` is an object/function that has it; `undefined` otherwise. */ const prop = (obj: unknown, key: string): unknown => @@ -25,10 +28,12 @@ const prop = (obj: unknown, key: string): unknown => ? (obj as Record)[key] : undefined -/** The `_tag` literal of a `Schema.TaggedErrorClass` (`ast.fields._tag.schema.literal`). */ -const readTag = (value: unknown): string | undefined => { +/** Stable runtime identifier: tagged errors use `_tag`; ErrorClass uses its class identifier/name. */ +const readIdentifier = (value: unknown): string | undefined => { const literal = prop(prop(prop(prop(value, "fields"), "_tag"), "schema"), "literal") - return typeof literal === "string" ? literal : undefined + if (typeof literal === "string") return literal + const identifier = prop(value, "identifier") + return typeof identifier === "string" ? identifier : undefined } /** The `httpApiStatus` annotation on a schema's AST, when present. */ @@ -37,24 +42,30 @@ const readHttpStatus = (value: unknown): number | undefined => { return typeof status === "number" ? status : undefined } -const deriveAnticipatedTags = (): ReadonlySet => { - const tags = new Set() - for (const value of Object.values(Http)) { +const deriveAnticipatedIdentifiers = (): ReadonlySet => { + const identifiers = new Set() + for (const value of [...Object.values(Http), ...Object.values(HttpV2)]) { if (typeof value !== "function") continue - const tag = readTag(value) - if (tag === undefined) continue + const identifier = readIdentifier(value) + if (identifier === undefined) continue const status = readHttpStatus(value) if (status === undefined) continue - if (status >= 400 && status < 500) tags.add(tag) + if (status >= 400 && status < 500) identifiers.add(identifier) } - return tags + return identifiers } /** - * `_tag`s of all domain HTTP errors annotated with a 4xx `httpApiStatus`. - * Pass `[...ANTICIPATED_ERROR_TAGS]` to the telemetry SDK's `anticipatedErrorTags`. + * Stable identifiers of all domain HTTP errors annotated with a 4xx `httpApiStatus`. + * Tagged errors contribute `_tag`; v2 ErrorClass values contribute `Error.name`. */ -export const ANTICIPATED_ERROR_TAGS: ReadonlySet = deriveAnticipatedTags() +export const ANTICIPATED_ERROR_IDENTIFIERS: ReadonlySet = deriveAnticipatedIdentifiers() -/** True when `tag` is a known anticipated (4xx) domain error tag. */ -export const isAnticipatedErrorTag = (tag: string): boolean => ANTICIPATED_ERROR_TAGS.has(tag) +export const isAnticipatedErrorIdentifier = (identifier: string): boolean => + ANTICIPATED_ERROR_IDENTIFIERS.has(identifier) + +/** @deprecated Use `ANTICIPATED_ERROR_IDENTIFIERS`. */ +export const ANTICIPATED_ERROR_TAGS = ANTICIPATED_ERROR_IDENTIFIERS + +/** @deprecated Use `isAnticipatedErrorIdentifier`. */ +export const isAnticipatedErrorTag = isAnticipatedErrorIdentifier diff --git a/packages/domain/src/http/v2/v2-contract.test.ts b/packages/domain/src/http/v2/v2-contract.test.ts index 5f769755c..ee5d8bba0 100644 --- a/packages/domain/src/http/v2/v2-contract.test.ts +++ b/packages/domain/src/http/v2/v2-contract.test.ts @@ -480,10 +480,7 @@ describe("scopes", () => { describe("telemetry contracts", () => { it("round-trips the synthetic composite log ID", () => { - const internal = JSON.stringify([ - "2026-07-15 12:00:00.123", - "00112233445566778899AABBCCDDEEFF", - ]) + const internal = JSON.stringify(["2026-07-15 12:00:00.123", "00112233445566778899AABBCCDDEEFF"]) const wire = Schema.encodeSync(LogPublicId)(internal) expect(wire.startsWith("log_")).toBe(true) expect(Schema.decodeSync(LogPublicId)(wire)).toBe(internal) @@ -532,20 +529,22 @@ describe("telemetry contracts", () => { describe("list pagination", () => { const items = Array.from({ length: 45 }, (_, index) => index) - it("paginates with default limit and opaque cursors", () => { - const first = Effect.runSync(paginateArray(items, {})) - expect(first.data).toHaveLength(20) - expect(first.has_more).toBe(true) - expect(first.next_cursor).not.toBeNull() + it.effect("paginates with default limit and opaque cursors", () => + Effect.gen(function* () { + const first = yield* paginateArray(items, {}) + expect(first.data).toHaveLength(20) + expect(first.has_more).toBe(true) + expect(first.next_cursor).not.toBeNull() - const second = Effect.runSync(paginateArray(items, { cursor: first.next_cursor! })) - expect(second.data[0]).toBe(20) + const second = yield* paginateArray(items, { cursor: first.next_cursor! }) + expect(second.data[0]).toBe(20) - const third = Effect.runSync(paginateArray(items, { cursor: second.next_cursor!, limit: 20 })) - expect(third.data).toHaveLength(5) - expect(third.has_more).toBe(false) - expect(third.next_cursor).toBeNull() - }) + const third = yield* paginateArray(items, { cursor: second.next_cursor!, limit: 20 }) + expect(third.data).toHaveLength(5) + expect(third.has_more).toBe(false) + expect(third.next_cursor).toBeNull() + }), + ) it("cursor round-trips and rejects garbage", () => { expect(decodeOffsetCursor(encodeOffsetCursor(1234))).toBe(1234) @@ -553,29 +552,33 @@ describe("list pagination", () => { expect(decodeOffsetCursor("off_-1")).toBeNull() }) - it("fails invalid cursors instead of silently restarting at page one", () => { - const result = Effect.runSync(Effect.result(paginateArray(items, { cursor: "garbage" }))) - expect(Result.isFailure(result)).toBe(true) - if (Result.isFailure(result)) { - expect(result.failure.error.code).toBe("parameter_invalid") - expect(result.failure.error.param).toBe("cursor") - } - }) - - it("uses storage lookahead and returns each row exactly once", () => { - const calls: Array<{ limit: number; offset: number }> = [] - const fetch = ({ limit, offset }: { limit: number; offset: number }) => { - calls.push({ limit, offset }) - return Effect.succeed(items.slice(offset, offset + limit)) - } - const first = Effect.runSync(paginateOffsetQuery({ limit: 10 }, fetch)) - const second = Effect.runSync(paginateOffsetQuery({ limit: 10, cursor: first.next_cursor! }, fetch)) - expect(calls).toEqual([ - { limit: 11, offset: 0 }, - { limit: 11, offset: 10 }, - ]) - expect([...first.data, ...second.data]).toEqual(items.slice(0, 20)) - }) + it.effect("fails invalid cursors instead of silently restarting at page one", () => + Effect.gen(function* () { + const result = yield* Effect.result(paginateArray(items, { cursor: "garbage" })) + expect(Result.isFailure(result)).toBe(true) + if (Result.isFailure(result)) { + expect(result.failure.error.code).toBe("parameter_invalid") + expect(result.failure.error.param).toBe("cursor") + } + }), + ) + + it.effect("uses storage lookahead and returns each row exactly once", () => + Effect.gen(function* () { + const calls: Array<{ limit: number; offset: number }> = [] + const fetch = ({ limit, offset }: { limit: number; offset: number }) => { + calls.push({ limit, offset }) + return Effect.succeed(items.slice(offset, offset + limit)) + } + const first = yield* paginateOffsetQuery({ limit: 10 }, fetch) + const second = yield* paginateOffsetQuery({ limit: 10, cursor: first.next_cursor! }, fetch) + expect(calls).toEqual([ + { limit: 11, offset: 0 }, + { limit: 11, offset: 10 }, + ]) + expect([...first.data, ...second.data]).toEqual(items.slice(0, 20)) + }), + ) it("enforces the shared limit range", () => { const decode = Schema.decodeUnknownSync(ListQuery) diff --git a/packages/domain/src/tinybird/project-sync.test.ts b/packages/domain/src/tinybird/project-sync.test.ts index 19a052995..64bda1103 100644 --- a/packages/domain/src/tinybird/project-sync.test.ts +++ b/packages/domain/src/tinybird/project-sync.test.ts @@ -349,6 +349,49 @@ describe("Tinybird project sync", () => { expect(calls.filter((c) => c.method === "DELETE")).toHaveLength(0) }) + + it("bounds stale-deployment deletion concurrency and attempts every item", async () => { + const staleIds = Array.from({ length: 7 }, (_, index) => `failed-${index + 1}`) + const attempted: string[] = [] + let active = 0 + let peak = 0 + let releaseGate: (() => void) | undefined + const gate = new Promise((resolve) => { + releaseGate = resolve + }) + + globalThis.fetch = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url + const method = init?.method ?? "GET" + if (method === "GET") { + return jsonResponse({ + deployments: staleIds.map((id) => ({ id, status: "failed", live: false })), + }) + } + + const id = url.match(/\/v1\/deployments\/([^?]+)/)?.[1] + if (id === undefined) throw new Error(`Unexpected deletion URL: ${url}`) + attempted.push(id) + active += 1 + peak = Math.max(peak, active) + await gate + active -= 1 + return new Response("", { status: 200 }) + }) as unknown as typeof fetch + + const cleanup = cleanupStaleTinybirdDeployments({ + baseUrl: "https://customer.tinybird.co", + token: "token", + }) + await vi.waitFor(() => expect(attempted).toHaveLength(3)) + expect(peak).toBe(3) + releaseGate?.() + await cleanup + + expect(attempted.sort()).toEqual(staleIds) + expect(peak).toBe(3) + }) }) describe("cleanupOwnedTinybirdDeployment", () => { diff --git a/packages/domain/src/tinybird/project-sync.ts b/packages/domain/src/tinybird/project-sync.ts index 94134fc37..3d0670408 100644 --- a/packages/domain/src/tinybird/project-sync.ts +++ b/packages/domain/src/tinybird/project-sync.ts @@ -9,6 +9,7 @@ import { } from "./ttl-override" const REQUEST_TIMEOUT = Duration.seconds(30) +const STALE_DEPLOYMENT_DELETE_CONCURRENCY = 3 const FeedbackEntrySchema = Schema.Struct({ resource: Schema.NullOr(Schema.String), @@ -445,7 +446,7 @@ export class TinybirdProjectSync extends Context.Service Layer.succeed(EdgeCacheService, makeEdgeCacheService(backend, readTimeoutMs)) describe("EdgeCacheService.getOrCompute (no schema)", () => { - it.effect("fails open to computation when a backend read exceeds its deadline", () => { + it.live("fails open to computation when a backend read exceeds its deadline", () => { let computeCalls = 0 const backend: EdgeCacheBackend = { get: async () => await new Promise(() => {}), @@ -57,7 +57,7 @@ describe("EdgeCacheService.getOrCompute (no schema)", () => { }).pipe(Effect.provide(makeLayer(backend, 10)), Effect.timeout(200)) }) - it.effect("shares the complete slow read-or-compute operation across concurrent callers", () => { + it.live("shares the complete slow read-or-compute operation across concurrent callers", () => { let getCalls = 0 let computeCalls = 0 const backend: EdgeCacheBackend = { @@ -140,7 +140,7 @@ describe("EdgeCacheService.getOrCompute (no schema)", () => { }) describe("EdgeCacheService.rawGet", () => { - it.effect("reports hit, miss, and timeout outcomes without collapsing them", () => { + it.live("reports hit, miss, and timeout outcomes without collapsing them", () => { const backend: EdgeCacheBackend = { get: async (bucket) => { if (bucket === "hit") return { value: 42 } @@ -166,7 +166,7 @@ describe("EdgeCacheService.rawGet", () => { }).pipe(Effect.provide(makeLayer(backend, 10)), Effect.timeout(200)) }) - it.effect("treats a backend read timeout as a cache miss", () => { + it.live("treats a backend read timeout as a cache miss", () => { const backend: EdgeCacheBackend = { get: async () => await new Promise(() => {}), put: async () => {}, diff --git a/packages/query-engine/src/observability/session-events.ts b/packages/query-engine/src/observability/session-events.ts index 22c67ff9f..66c012ef3 100644 --- a/packages/query-engine/src/observability/session-events.ts +++ b/packages/query-engine/src/observability/session-events.ts @@ -68,7 +68,7 @@ export const searchSessions = Effect.fn("Observability.searchSessions")(function }), { orgId: executor.orgId, startTime: input.startTime, endTime: input.endTime }, ) - return yield* executor.compiledQuery(compiled, { profile: "list" }) + return yield* executor.compiledQuery(compiled, { profile: "list", context: "searchSessions" }) }) /** @@ -103,5 +103,5 @@ export const getSessionTranscript = Effect.fn("Observability.getSessionTranscrip sessionId: input.sessionId, }, ) - return yield* executor.compiledQuery(compiled, { profile: "list" }) + return yield* executor.compiledQuery(compiled, { profile: "list", context: "sessionTranscript" }) }) diff --git a/packages/query-engine/src/observability/session-replays.ts b/packages/query-engine/src/observability/session-replays.ts index b71126786..11c602265 100644 --- a/packages/query-engine/src/observability/session-replays.ts +++ b/packages/query-engine/src/observability/session-replays.ts @@ -75,8 +75,14 @@ export const getSessionTraces = Effect.fn("Observability.getSessionTraces")(func }) const [maybeSession, maybeActivity] = yield* Effect.all( [ - executor.compiledQueryFirst(detailCompiled, { profile: "discovery" }), - executor.compiledQueryFirst(activityCompiled, { profile: "discovery" }), + executor.compiledQueryFirst(detailCompiled, { + profile: "discovery", + context: "sessionReplayDetail", + }), + executor.compiledQueryFirst(activityCompiled, { + profile: "discovery", + context: "sessionActivity", + }), ], { concurrency: 2 }, ) @@ -106,7 +112,10 @@ export const getSessionTraces = Effect.fn("Observability.getSessionTraces")(func const summariesCompiled = CH.compile(CH.sessionTraceSummariesQuery({ traceIds }), { orgId: executor.orgId, }) - const traces = yield* executor.compiledQuery(summariesCompiled, { profile: "list" }) + const traces = yield* executor.compiledQuery(summariesCompiled, { + profile: "list", + context: "sessionTraceSummaries", + }) return { session, traces, totalTraceCount } satisfies SessionTracesOutput }) diff --git a/packages/query-engine/src/observability/span-detail.ts b/packages/query-engine/src/observability/span-detail.ts index 6f53693e7..773af0a1f 100644 --- a/packages/query-engine/src/observability/span-detail.ts +++ b/packages/query-engine/src/observability/span-detail.ts @@ -70,7 +70,10 @@ export const spanDetail = Effect.fn("Observability.spanDetail")(function* (input ? { orgId: executor.orgId, startTime: range.startTime, endTime: range.endTime } : { orgId: executor.orgId }, ) - const maybeRow = yield* executor.compiledQueryFirst(compiled, { profile: "discovery" }) + const maybeRow = yield* executor.compiledQueryFirst(compiled, { + profile: "discovery", + context: "spanDetail", + }) if (Option.isNone(maybeRow)) { return { diff --git a/packages/query-engine/src/runtime/raw-sql.test.ts b/packages/query-engine/src/runtime/raw-sql.test.ts index 6b286279b..9fbdc01f7 100644 --- a/packages/query-engine/src/runtime/raw-sql.test.ts +++ b/packages/query-engine/src/runtime/raw-sql.test.ts @@ -189,6 +189,8 @@ describe("makeExecuteRawSql", () => { const cellError = yield* Effect.flip( executeRows([{ value: "x".repeat(MAX_RAW_SQL_CELL_LENGTH + 1) }]), ) + assert.strictEqual(cellError._tag, "@maple/http/errors/RawSqlValidationError") + assert.strictEqual(cellError.code, "ResourceLimit") assert.include(cellError.message, "cells") const bytesError = yield* Effect.flip( @@ -199,9 +201,13 @@ describe("makeExecuteRawSql", () => { })), ), ) + assert.strictEqual(bytesError._tag, "@maple/http/errors/RawSqlValidationError") + assert.strictEqual(bytesError.code, "ResourceLimit") assert.include(bytesError.message, "bytes") const jsonError = yield* Effect.flip(executeRows([{ value: 1n }])) + assert.strictEqual(jsonError._tag, "@maple/http/errors/RawSqlValidationError") + assert.strictEqual(jsonError.code, "ResourceLimit") assert.include(jsonError.message, "JSON serializable") }), )