diff --git a/docs/ci-performance.md b/docs/ci-performance.md index 747aa168ac..9bca5d4f38 100644 --- a/docs/ci-performance.md +++ b/docs/ci-performance.md @@ -10,7 +10,7 @@ when services moved out of API, leaving the longest suite unsharded. workspace patterns and tracked manifests. It sizes Vitest shards against a 140s estimated test budget and packs small suites into lanes with the same budget. Per-file rates for the large suites are measured from CI run 35725366619 -(backend 1.6s, api 2.3s, web 0.55s, …); unknown suites get 0.6s/file, fitted to +(api 2.3s, web 0.55s, …; backend was re-measured at 2.0s on run 42756486011); unknown suites get 0.6s/file, fitted to the slowest cold small-suite lanes, plus a per-suite startup cost. File counts include all tracked test/spec files, so a suite can be overestimated but its tests are never filtered. Each lane runs the original package scripts through diff --git a/docs/error-issue-lifecycle.md b/docs/error-issue-lifecycle.md index 2b765526b5..a1b08154e3 100644 --- a/docs/error-issue-lifecycle.md +++ b/docs/error-issue-lifecycle.md @@ -211,12 +211,25 @@ loop in verification without a human ever seeing it. ## Rules worth not re-deriving -- **Every state write is guarded on the state it was decided from.** `applyTransition`, the - errors tick's regression and the alert/integration reopen all go through `transitionRow` +- **Every state write is guarded on the state it was decided from.** `applyTransition` and the + alert/integration reopen go through `transitionRow` (`packages/backend/src/platform/transition-row.ts`): the UPDATE carries - `workflow_state IN ()`, so a concurrent move wins and the late writer gets a 409 - (or, for a tick, skips the row). A fixed alert or integration issue that fires again lands in - `regressed` too, never `triage`. + `workflow_state IN ()`, so a concurrent move wins and the late writer gets a 409. + The errors tick's regression carries the same guard in its own bulk UPDATE (`applyRegressions` + in `error-tick-persistence.ts`, `workflow_state = 'done'`) and skips the rows it lost. A fixed + alert or integration issue that fires again lands in `regressed` too, never `triage`. +- **A pull request link is verified once.** GitHub reports `merged: true` on every event after the + merge, so a later edit or a redelivery must not reopen a settled check. The guard is on the link + row, so unlinking a PR and linking it again is a new link and verifies again. +- **One verification row cannot starve the rest.** The tick reads due rows least recently + touched first. A row whose issue or link is gone is settled as `abandoned`. One whose + occurrences cannot be read is touched and goes to the back of the queue with its window + unchanged: it stays `waiting` for as long as its org's warehouse keeps failing, and is tried + again when its turn comes round, but it no longer sits ahead of rows that can be judged. +- **Retention deletes an issue's children by issue id.** There are no FK cascades. Incidents, + evaluator state, events, aliases, pull request links, verifications and escalations go with the + issue. Investigations stay: each keys a chat session whose storage only `apps/ai` can release. + Notification outbox rows in a terminal status are pruned after 30 days. - **Never make the claim a step an agent has to remember.** Any new tool that starts work on an issue takes the lease itself. See above for what happened when it did not. diff --git a/packages/backend/src/services/errors/ErrorIssueWorkflowService.ts b/packages/backend/src/services/errors/ErrorIssueWorkflowService.ts index 48d7c7d65b..70ecfaea52 100644 --- a/packages/backend/src/services/errors/ErrorIssueWorkflowService.ts +++ b/packages/backend/src/services/errors/ErrorIssueWorkflowService.ts @@ -59,6 +59,18 @@ const decodeIssueDateTimeSync = Schema.decodeUnknownSync(ErrorIssueDocument.fiel const DEFAULT_EVENTS_LIMIT = 100 const DEFAULT_LEASE_DURATION_MS = 30 * 60_000 +/** The 409 for an issue somebody else holds. `row` is the holder's row, when one was read. */ +export const leaseConflict = (issueId: ErrorIssueId, row: ErrorIssueRow | null) => + new ErrorIssueLeaseConflictError({ + message: "Issue is held by another actor", + issueId, + currentHolderActorId: row?.leaseHolderActorId ?? null, + leaseExpiresAt: + row?.leaseExpiresAt == null + ? null + : decodeIssueDateTimeSync(new Date(row.leaseExpiresAt).toISOString()), + }) + export interface ErrorIssueWorkflowPublicApi { readonly heartbeatIssue: ( orgId: OrgId, @@ -319,26 +331,33 @@ const make: Effect.Effect< issueIds, ) => { if (issueIds.length === 0) return Effect.succeed(new Set()) - return Effect.all([ - dbExecute((db) => - db.run( - PG.from(ErrorIncidents) - .select("issueId") - .where(($) => [$.orgId.eq(orgId), $.status.eq("open"), $.issueId.in_(...issueIds)]), + return Effect.all( + [ + dbExecute((db) => + db.run( + PG.from(ErrorIncidents) + .select("issueId") + .where(($) => [ + $.orgId.eq(orgId), + $.status.eq("open"), + $.issueId.in_(...issueIds), + ]), + ), ), - ), - dbExecute((db) => - db.run( - PG.from(AlertIncidents) - .select(($) => ({ issueId: $.errorIssueId })) - .where(($) => [ - $.orgId.eq(orgId), - $.status.eq("open"), - $.errorIssueId.in_(...issueIds), - ]), + dbExecute((db) => + db.run( + PG.from(AlertIncidents) + .select(($) => ({ issueId: $.errorIssueId })) + .where(($) => [ + $.orgId.eq(orgId), + $.status.eq("open"), + $.errorIssueId.in_(...issueIds), + ]), + ), ), - ), - ]).pipe( + ], + { concurrency: "unbounded" }, + ).pipe( Effect.map( ([errorRows, alertRows]) => new Set([ @@ -364,28 +383,31 @@ const make: Effect.Effect< if (issueIds.length === 0) return rollups const upsert = (issueId: ErrorIssueId, patch: Partial) => rollups.set(issueId, { ...(rollups.get(issueId) ?? EMPTY_ACTIVITY), ...patch }) - const [commentRows, prRows] = yield* Effect.all([ - dbExecute((db) => - db.run( - PG.from(ErrorIssueEvents) - .select(($) => ({ issueId: $.issueId, count: PG.count() })) - .where(($) => [ - $.orgId.eq(orgId), - $.issueId.in_(...issueIds), - $.type.in_("comment", "agent_note"), - ]) - .groupBy("issueId"), + const [commentRows, prRows] = yield* Effect.all( + [ + dbExecute((db) => + db.run( + PG.from(ErrorIssueEvents) + .select(($) => ({ issueId: $.issueId, count: PG.count() })) + .where(($) => [ + $.orgId.eq(orgId), + $.issueId.in_(...issueIds), + $.type.in_("comment", "agent_note"), + ]) + .groupBy("issueId"), + ), ), - ), - dbExecute((db) => - db.run( - PG.from(ErrorIssuePullRequests) - .select(($) => ({ issueId: $.issueId, state: $.state, count: PG.count() })) - .where(($) => [$.orgId.eq(orgId), $.issueId.in_(...issueIds)]) - .groupBy("issueId", "state"), + dbExecute((db) => + db.run( + PG.from(ErrorIssuePullRequests) + .select(($) => ({ issueId: $.issueId, state: $.state, count: PG.count() })) + .where(($) => [$.orgId.eq(orgId), $.issueId.in_(...issueIds)]) + .groupBy("issueId", "state"), + ), ), - ), - ]) + ], + { concurrency: "unbounded" }, + ) for (const row of commentRows) upsert(row.issueId, { commentCount: row.count }) for (const row of prRows) { if (row.state === "open") upsert(row.issueId, { openPullRequestCount: row.count }) @@ -397,12 +419,22 @@ const make: Effect.Effect< const hydrateNonEmpty = (orgId: OrgId, rows: Arr.NonEmptyReadonlyArray) => Effect.gen(function* () { const issueIds = Arr.map(rows, (row) => row.id) - const openSet = yield* issuesWithOpenIncidents(orgId, issueIds) - const activityMap = yield* issueActivityRollups(orgId, issueIds) - const hashesByIssue = yield* dbExecute((db) => issueFingerprintHashes(db, orgId, rows)) - const actorMap = yield* actorsService.collectActorDocs( - orgId, - Arr.flatMap(rows, (row) => [row.assignedActorId ?? null, row.leaseHolderActorId ?? null]), + // Independent reads, run on every list call and after every mutation. + // The invocation's pool bounds how many are really in flight. + const [openSet, activityMap, hashesByIssue, actorMap] = yield* Effect.all( + [ + issuesWithOpenIncidents(orgId, issueIds), + issueActivityRollups(orgId, issueIds), + dbExecute((db) => issueFingerprintHashes(db, orgId, rows)), + actorsService.collectActorDocs( + orgId, + Arr.flatMap(rows, (row) => [ + row.assignedActorId ?? null, + row.leaseHolderActorId ?? null, + ]), + ), + ], + { concurrency: "unbounded" }, ) return Arr.map(rows, (row) => rowToIssue( @@ -689,14 +721,6 @@ const make: Effect.Effect< return yield* requireIssue(orgId, row.id) }) - const leaseConflict = (issueId: ErrorIssueId, row: ErrorIssueRow | null) => - new ErrorIssueLeaseConflictError({ - message: "Issue is held by another actor", - issueId, - currentHolderActorId: row?.leaseHolderActorId ?? null, - leaseExpiresAt: row?.leaseExpiresAt == null ? null : isoFromMs(row.leaseExpiresAt), - }) - const heartbeatIssue: ErrorIssueWorkflowServiceApi["heartbeatIssue"] = Effect.fn( "ErrorIssueWorkflowService.heartbeatIssue", )(function* (orgId, actorId, issueId) { diff --git a/packages/backend/src/services/errors/ErrorsService.test.ts b/packages/backend/src/services/errors/ErrorsService.test.ts index 76b529453f..b5ee2a5fe9 100644 --- a/packages/backend/src/services/errors/ErrorsService.test.ts +++ b/packages/backend/src/services/errors/ErrorsService.test.ts @@ -17,12 +17,15 @@ import { } from "effect" import { TestClock } from "effect/testing" import { + ErrorIssuePullRequestId, + ErrorIssueVerificationId, ErrorPersistenceError, IssueEscalationPolicyRule, IssueEscalationPolicyUpsertRequest, IssueListCursor, IssueSeverityListCursor, OrgId, + type PullRequestSummary, UserId, WarehouseAuthError, WarehouseQueryError, @@ -33,6 +36,7 @@ import { ErrorIncidentId, ErrorIssueEventId, ErrorIssueId, + IssueEscalationId, } from "@maple/domain/primitives" import { FINGERPRINT_VERSION } from "@maple/domain/tinybird/fingerprint" import * as PG from "@maple-dev/effect-orm/postgres" @@ -44,7 +48,9 @@ import { ErrorNotificationDeliveries, ErrorIssues, ErrorIssueEvents, + ErrorIssuePullRequests, ErrorIssueStates, + ErrorIssueVerifications, ErrorNotificationPolicies, ErrorTickStates, IssueEscalations, @@ -69,7 +75,9 @@ import { ErrorActorsService } from "./ErrorActorsService" import { ErrorIssueReadModelsService } from "./ErrorIssueReadModelsService" import { ErrorIssueWorkflowService } from "./ErrorIssueWorkflowService" import { ErrorPolicyService } from "./ErrorPolicyService" -import { CANDIDATE_RETENTION_MS, ErrorsService } from "./ErrorsService" +import { CANDIDATE_RETENTION_MS, ErrorsService, NOTIFICATION_DELIVERY_PRUNE_BATCH } from "./ErrorsService" +import { IssueFixVerificationService } from "./IssueFixVerificationService" +import { PullRequestLookup } from "./PullRequestLookup" import { describeCause, makePersistenceError } from "./error-persistence" import { EarlyPromotionsPerDay, isErrorTickClaimLost, persistErrorTickWindow } from "./error-tick-persistence" import { NotificationDispatcher } from "@maple/backend/services/alerts/NotificationDispatcher" @@ -201,6 +209,8 @@ const makeErrorsLayer = ( fingerprintRows?: () => ReadonlyArray>, dispatcher?: (typeof NotificationDispatcher)["Service"], lineageRows?: () => ReadonlyArray>, + // What the provider reports for a linked PR. Present, it also wires fix verification in. + pullRequest?: () => PullRequestSummary, ) => { const testDb = createTestDb(createdDbs) const envLive = Env.layer.pipe(Layer.provide(testConfig())) @@ -236,6 +246,23 @@ const makeErrorsLayer = ( dispatch: () => Effect.succeed({ delivered: 0, failed: 0 }), }, ) + const fixVerificationLive = + pullRequest === undefined + ? Layer.empty + : Layer.effect(IssueFixVerificationService, IssueFixVerificationService.make).pipe( + Layer.provide( + Layer.mergeAll( + databaseLive, + envLive, + errorActorsLive, + errorIssueWorkflowLive, + Layer.succeed(PullRequestLookup, { + fetch: () => Effect.sync(() => Option.some(pullRequest())), + listRepositories: () => Effect.succeed([]), + }), + ), + ), + ) const errorsLive = Layer.effect(ErrorsService, ErrorsService.make).pipe( Layer.provide(OrgClickHouseSettingsService.layer), Layer.provide(envLive), @@ -249,6 +276,7 @@ const makeErrorsLayer = ( Layer.provide(errorIssueWorkflowLive), Layer.provide(errorPolicyLive), Layer.provide(investigationsLive), + Layer.provide(fixVerificationLive), ) return Layer.mergeAll( errorsLive, @@ -391,6 +419,9 @@ const asIssueId = Schema.decodeUnknownSync(ErrorIssueId) const asIncidentId = Schema.decodeUnknownSync(ErrorIncidentId) const asEventId = Schema.decodeUnknownSync(ErrorIssueEventId) const asDestinationId = Schema.decodeUnknownSync(AlertDestinationId) +const asPullRequestId = Schema.decodeUnknownSync(ErrorIssuePullRequestId) +const asVerificationId = Schema.decodeUnknownSync(ErrorIssueVerificationId) +const asEscalationId = Schema.decodeUnknownSync(IssueEscalationId) const asJsonRecord = Schema.decodeUnknownSync(Schema.Record(Schema.String, Schema.Unknown)) const ORG = asOrgId("org_errors_service_test") @@ -3008,6 +3039,184 @@ describe("ErrorsService.runTick", () => { }).pipe(Effect.provide(makeErrorsLayer())), ) + // No FK cascades: whatever retention does not delete by issue id stays forever. + it.effect("purging an issue takes its pull requests, verifications and escalations with it", () => + Effect.gen(function* () { + const errors = yield* ErrorsService + const database = yield* Database + yield* TestClock.setTime(RETENTION_TICK_MS) + const seededAt = RETENTION_TICK_MS - 120 * DAY_MS + + const purged = asIssueId(randomUUID()) + yield* seedIssue(purged, { + workflowState: "done", + resolvedAt: seededAt, + archivedAt: RETENTION_TICK_MS - 91 * DAY_MS, + lastSeenAt: seededAt, + }) + // Still live, so its own link must survive the purge of the other issue. + const kept = asIssueId(randomUUID()) + yield* seedIssue(kept, { workflowState: "in_review" }) + + const linkOf = (issueId: ErrorIssueId, number: number) => ({ + id: asPullRequestId(randomUUID()), + orgId: ORG, + issueId, + provider: "github" as const, + repoFullName: "MapleTechLabs/maple", + number, + url: `https://github.com/MapleTechLabs/maple/pull/${number}`, + linkSource: "agent" as const, + createdAt: seededAt, + updatedAt: seededAt, + }) + const purgedLink = linkOf(purged, 612) + yield* database.execute((db) => + db.run(PG.insertInto(ErrorIssuePullRequests).values([purgedLink, linkOf(kept, 613)])), + ) + yield* database.execute((db) => + db.run( + PG.insertInto(ErrorIssueVerifications).values({ + id: asVerificationId(randomUUID()), + orgId: ORG, + issueId: purged, + pullRequestId: purgedLink.id, + status: "verified", + mergedAt: seededAt, + verifyAfter: seededAt + DAY_MS, + createdAt: seededAt, + updatedAt: seededAt, + }), + ), + ) + yield* database.execute((db) => + db.run( + PG.insertInto(IssueEscalations).values({ + id: asEscalationId(randomUUID()), + orgId: ORG, + issueId: purged, + severity: "high", + source: "ai", + reason: "severity_set", + status: "sent", + dedupeKey: `esc:${ORG}:${purged}:high`, + createdAt: seededAt, + }), + ), + ) + + assert.strictEqual((yield* errors.runTick).issuesDeleted, 1) + + const links = yield* database.execute((db) => + db.run(PG.from(ErrorIssuePullRequests).select("issueId")), + ) + assert.deepStrictEqual( + Arr.map(links, (row) => row.issueId), + [kept], + ) + assert.lengthOf( + yield* database.execute((db) => db.run(PG.from(ErrorIssueVerifications).select("id"))), + 0, + ) + assert.lengthOf( + yield* database.execute((db) => db.run(PG.from(IssueEscalations).select("id"))), + 0, + ) + }).pipe(Effect.provide(makeErrorsLayer())), + ) + + it.effect("the retention phase prunes notification deliveries that settled over 30 days ago", () => + Effect.gen(function* () { + const errors = yield* ErrorsService + const database = yield* Database + yield* TestClock.setTime(RETENTION_TICK_MS) + // Retention runs per org with issue state, so the org needs an issue. + yield* seedIssue(asIssueId(randomUUID()), { workflowState: "in_review" }) + + const delivery = (key: string, status: "success" | "failed" | "queued", settledAt: number) => ({ + id: randomUUID(), + orgId: ORG, + destinationId: asDestinationId("7d31c9e1-0000-4000-8000-000000000001"), + deliveryKey: key, + payloadJson: {}, + status, + attemptCount: 1, + // Never due, so the outbox drain in the same tick leaves the row alone. + scheduledAt: RETENTION_TICK_MS + DAY_MS, + claimedAt: null, + claimExpiresAt: null, + claimedBy: null, + attemptedAt: settledAt, + errorMessage: null, + createdAt: settledAt, + updatedAt: settledAt, + }) + const old = RETENTION_TICK_MS - 31 * DAY_MS + yield* database.execute((db) => + db.run( + PG.insertInto(ErrorNotificationDeliveries).values([ + delivery("old-success", "success", old), + delivery("old-failed", "failed", old), + // Not terminal: still owed a delivery, however old. + delivery("old-queued", "queued", old), + delivery("recent-success", "success", RETENTION_TICK_MS - 29 * DAY_MS), + ]), + ), + ) + + assert.isTrue((yield* errors.runTick).retentionRan) + + const remaining = yield* database.execute((db) => + db.run(PG.from(ErrorNotificationDeliveries).select("deliveryKey")), + ) + expect(new Set(Arr.map(remaining, (row) => row.deliveryKey))).toEqual( + new Set(["old-queued", "recent-success"]), + ) + }).pipe(Effect.provide(makeErrorsLayer())), + ) + + it.effect("one retention run prunes a bounded batch of deliveries and leaves the rest for the next", () => + Effect.gen(function* () { + const errors = yield* ErrorsService + const database = yield* Database + yield* TestClock.setTime(RETENTION_TICK_MS) + yield* seedIssue(asIssueId(randomUUID()), { workflowState: "in_review" }) + + const old = RETENTION_TICK_MS - 31 * DAY_MS + const backlog = NOTIFICATION_DELIVERY_PRUNE_BATCH + 5 + yield* database.execute((db) => + db.run( + PG.insertInto(ErrorNotificationDeliveries).values( + Arr.makeBy(backlog, (index) => ({ + id: randomUUID(), + orgId: ORG, + destinationId: asDestinationId("7d31c9e1-0000-4000-8000-000000000001"), + deliveryKey: `backlog-${index}`, + payloadJson: {}, + status: "success" as const, + attemptCount: 1, + scheduledAt: RETENTION_TICK_MS + DAY_MS, + claimedAt: null, + claimExpiresAt: null, + claimedBy: null, + attemptedAt: old, + errorMessage: null, + createdAt: old, + updatedAt: old, + })), + ), + ), + ) + + assert.isTrue((yield* errors.runTick).retentionRan) + + const remaining = yield* database.execute((db) => + db.run(PG.from(ErrorNotificationDeliveries).select("id")), + ) + expect(remaining).toHaveLength(5) + }).pipe(Effect.provide(makeErrorsLayer())), + ) + it.effect("an archived fixed issue that old clients still hit is kept, not re-minted as new", () => { let rows: Array> = [] return Effect.gen(function* () { @@ -4549,6 +4758,46 @@ describe("ErrorsService.proposeFix claims the issue", () => { }).pipe(Effect.provide(makeErrorsLayer())), ) + // Attaching a PR that already merged opens the verification window and moves + // the issue to `verifying`. The walk to `in_review` then started from the row + // read before the link, matched nothing, and the caller got a transition + // error for a proposal whose event, link and window were all committed. + it.effect("reports success, in verifying, when the proposed PR had already merged", () => { + const mergedAtMs = 1_750_000_000_000 + const prUrl = "https://github.com/MapleTechLabs/maple/pull/612" + return Effect.gen(function* () { + const errors = yield* ErrorsService + const actorService = yield* ErrorActorsService + const actor = yield* actorService.ensureUserActor(ORG, USER) + const issueId = asIssueId(randomUUID()) + yield* seedIssue(issueId, { workflowState: "in_progress" }) + + const issue = yield* errors.proposeFix(ORG, actor.id, issueId, { + patchSummary: "Guard the null customer id", + prUrl, + }) + + assert.strictEqual(issue.workflowState, "verifying") + assert.strictEqual((yield* loadIssueById(issueId))?.workflowState, "verifying") + }).pipe( + Effect.provide( + makeErrorsLayer(undefined, undefined, undefined, undefined, undefined, undefined, () => ({ + number: 612, + title: "Guard the null customer id", + url: prUrl, + authorLogin: "octocat", + state: "merged", + headRef: "fix/null-customer", + baseRef: "main", + isDraft: false, + updatedAtMs: mergedAtMs, + mergedAtMs, + mergeCommitSha: Str.repeat(40)("a"), + })), + ), + ) + }) + it.effect("works from every state an issue can be worked from", () => Effect.gen(function* () { const errors = yield* ErrorsService diff --git a/packages/backend/src/services/errors/ErrorsService.ts b/packages/backend/src/services/errors/ErrorsService.ts index 9054bcf2b8..dddd455448 100644 --- a/packages/backend/src/services/errors/ErrorsService.ts +++ b/packages/backend/src/services/errors/ErrorsService.ts @@ -33,10 +33,13 @@ import { ErrorIssues, ErrorIssueEvents, ErrorIssueFingerprints, + ErrorIssuePullRequests, type ErrorIssueRow, ErrorIssueStates, + ErrorIssueVerifications, ErrorNotificationPolicies, ErrorTickStates, + IssueEscalations, OrgClickHouseSettings, OrgIngestKeys, } from "@maple/db/tables" @@ -80,7 +83,7 @@ import { quarantineOnConfigClassCause, } from "@maple/backend/services/warehouse/warehouse-org-quarantine" import { actorRowToDocument, ErrorActorsService } from "./ErrorActorsService" -import { ErrorIssueWorkflowService } from "./ErrorIssueWorkflowService" +import { ErrorIssueWorkflowService, leaseConflict } from "./ErrorIssueWorkflowService" import { IssueFixVerificationService } from "./IssueFixVerificationService" import { ErrorPolicyService } from "./ErrorPolicyService" import { makeErrorDatabaseExecute, makePersistenceError } from "./error-persistence" @@ -89,7 +92,6 @@ import { summarizeCause } from "@maple/backend/platform/describe-cause" const decodeErrorIssueIdSync = Schema.decodeUnknownSync(ErrorIssueDocument.fields.id) const decodeErrorIncidentIdSync = Schema.decodeUnknownSync(ErrorIncidentDocument.fields.id) const decodeEventIdSync = Schema.decodeUnknownSync(ErrorIssueEventIdSchema) -const decodeIsoDateTimeStringSync = Schema.decodeUnknownSync(ErrorIssueDocument.fields.firstSeenAt) const decodeRoleNameSync = Schema.decodeUnknownSync(RoleName) const decodeUserIdSync = Schema.decodeUnknownSync(UserIdSchema) @@ -166,6 +168,10 @@ const ACTIVE_ORGS_CACHE_TTL_S = 6 * 60 * 60 export const CANDIDATE_RETENTION_MS = 7 * 24 * 60 * 60 * 1000 const RESOLVED_RETENTION_DAYS = 14 const ARCHIVED_RETENTION_DAYS = 90 +/** How long a delivered, failed or silenced notification outbox row is kept for debugging. */ +const NOTIFICATION_DELIVERY_RETENTION_DAYS = 30 +/** Outbox rows pruned per org per retention run. */ +export const NOTIFICATION_DELIVERY_PRUNE_BATCH = 1000 /** * Retention runs one tick an hour. The phase is bucketed on the CRON period, * not on the evaluator's catch-up window: the alerting cron fires every minute, @@ -299,8 +305,6 @@ const make: Effect.Effect< const dbExecute = makeErrorDatabaseExecute(database, "ErrorsService") - const isoFromMs = (ms: number) => decodeIsoDateTimeStringSync(new Date(ms).toISOString()) - const systemTenant = (orgId: OrgId): TenantContext => ({ orgId, userId: decodeUserIdSync("system-errors"), @@ -523,14 +527,6 @@ const make: Effect.Effect< // Claim / lease - const leaseConflict = (issueId: ErrorIssueId, row: ErrorIssueRow | null) => - new ErrorIssueLeaseConflictError({ - message: "Issue is held by another actor", - issueId, - currentHolderActorId: row?.leaseHolderActorId ?? null, - leaseExpiresAt: row?.leaseExpiresAt == null ? null : isoFromMs(row.leaseExpiresAt), - }) - /** * Take (or renew) the lease on an issue and return the freshly-read row. * @@ -751,17 +747,26 @@ const make: Effect.Effect< ) } + // Linking can move the issue itself: a PR that had already merged opens the + // verification window and puts the issue in `verifying`. Walk from the row + // as it is now, or the first hop's guard matches nothing and the caller is + // told the proposal failed after the event, link and window were written. + const linked = request.prUrl === undefined ? row : yield* requireIssue(orgId, issueId) + // Usually `triage → in_progress → in_review`; one hop from a state the - // matrix lets straight through. Validated above, so no hop can fail here. - const next = yield* Effect.reduce( - fixProposalRoute(row.workflowState), - () => row, - (current, hop) => - applyTransition(orgId, actorId, current, hop, { - payload: { viaProposeFix: true }, - timestamp, - }), - ) + // matrix lets straight through. `verifying` is already past review. + const next = + linked.workflowState === "verifying" + ? linked + : yield* Effect.reduce( + fixProposalRoute(linked.workflowState), + () => linked, + (current, hop) => + applyTransition(orgId, actorId, current, hop, { + payload: { viaProposeFix: true }, + timestamp, + }), + ) yield* touchActor(orgId, actorId, timestamp) yield* maybeNotifyTransition(orgId, actorId, next, current.workflowState) return yield* hydrateIssue(orgId, next) @@ -1028,9 +1033,9 @@ const make: Effect.Effect< /** * The four unconditional reads at the head of every per-org tick, in ONE - * `Database.execute`. Under `DatabasePgLive` each execute dials and tears - * down its own postgres.js client, so the handshake count is what costs, not - * the statement count — same trade as `scrape-check-retention.ts`. + * `Database.execute`: one span and one retry boundary for the group. The + * invocation's pool is reused across executes (`docs/persistence.md`), so + * the batching no longer saves a handshake per read. * * The stale-incident sweep is deliberately NOT batched in here: it has to * observe the `last_triggered_at` writes the fingerprint loop makes, so @@ -1590,6 +1595,35 @@ const make: Effect.Effect< ), ) + // Outbox rows are only ever inserted and updated. One that has reached a + // terminal status is history; the delivery key it deduplicates on belongs + // to a tick window that committed long ago. + // A bounded batch, like the issue purge below: nothing pruned these before, so + // an org's first run can face a large backlog. Later runs drain the rest. + const staleDeliveries = yield* dbExecute((db) => + db.run( + PG.from(ErrorNotificationDeliveries) + .select("id") + .where(($) => [ + $.orgId.eq(orgId), + $.status.in_("success", "failed", "silenced"), + $.updatedAt.lt(nowMs - NOTIFICATION_DELIVERY_RETENTION_DAYS * DAY_MS), + ]) + .limit(NOTIFICATION_DELIVERY_PRUNE_BATCH), + ), + ) + if (staleDeliveries.length > 0) { + const deliveryIds = Arr.map(staleDeliveries, (row) => row.id) + yield* dbExecute((db) => + db.run( + PG.deleteFrom(ErrorNotificationDeliveries).where(($) => [ + $.orgId.eq(orgId), + $.id.in_(...deliveryIds), + ]), + ), + ) + } + const archivedCutoff = nowMs - ARCHIVED_RETENTION_DAYS * DAY_MS const toDelete = yield* dbExecute((db) => db.run( @@ -1632,6 +1666,33 @@ const make: Effect.Effect< ]), ), ) + // Rows that only mean something next to their issue. No FK cascades exist, + // so each is deleted here. Investigations are left: each keys a chat session + // whose storage only the AI worker can release. + yield* dbExecute((db) => + db.run( + PG.deleteFrom(ErrorIssueVerifications).where(($) => [ + $.orgId.eq(orgId), + $.issueId.in_(...ids), + ]), + ), + ) + yield* dbExecute((db) => + db.run( + PG.deleteFrom(ErrorIssuePullRequests).where(($) => [ + $.orgId.eq(orgId), + $.issueId.in_(...ids), + ]), + ), + ) + yield* dbExecute((db) => + db.run( + PG.deleteFrom(IssueEscalations).where(($) => [ + $.orgId.eq(orgId), + $.issueId.in_(...ids), + ]), + ), + ) // The issues merged into a purged one get back what they brought, as an unmerge // hands it back: their own hash is their primary again, the rest their aliases. const freed = yield* dbExecute((db) => @@ -1900,9 +1961,14 @@ const make: Effect.Effect< // rows a call — together ~36% of all database CPU. Walk the btree instead. // `org_ingest_keys` stays a plain DISTINCT on purpose: it is ~1 row per org, // where a loose index scan costs an index descent per row and wins nothing. - const stateOrgs = yield* dbExecute((db) => selectDistinctOrgIds(db, ErrorIssueStates)) - const issueOrgs = yield* dbExecute((db) => selectDistinctOrgIds(db, ErrorIssues)) - const ingestOrgs = yield* dbExecute((db) => db.run(PG.from(OrgIngestKeys).select("orgId").distinct())) + const [stateOrgs, issueOrgs, ingestOrgs] = yield* Effect.all( + [ + dbExecute((db) => selectDistinctOrgIds(db, ErrorIssueStates)), + dbExecute((db) => selectDistinctOrgIds(db, ErrorIssues)), + dbExecute((db) => db.run(PG.from(OrgIngestKeys).select("orgId").distinct())), + ], + { concurrency: "unbounded" }, + ) const knownOrgs = new Set([...stateOrgs, ...issueOrgs, ...Arr.map(ingestOrgs, (r) => r.orgId)]) const { active: activeOrgs, discovered } = yield* resolveActiveOrgs([...knownOrgs], nowMs) diff --git a/packages/backend/src/services/errors/FixVerificationTickService.test.ts b/packages/backend/src/services/errors/FixVerificationTickService.test.ts index ec7aa11e26..116969d764 100644 --- a/packages/backend/src/services/errors/FixVerificationTickService.test.ts +++ b/packages/backend/src/services/errors/FixVerificationTickService.test.ts @@ -1,8 +1,28 @@ -import { assert, describe, it } from "@effect/vitest" -import { verificationVerdictAutoCloses } from "@maple/domain/http" - -import { splitVersionRows, verdictFromInvestigationStatus } from "./FixVerificationTickService" -import { Array as Arr } from "effect" +import { randomUUID } from "node:crypto" +import { afterEach, assert, describe, it } from "@effect/vitest" +import { OrgId, verificationVerdictAutoCloses, WarehouseQueryError } from "@maple/domain/http" +import { ErrorIssueId } from "@maple/domain/primitives" +import * as PG from "@maple-dev/effect-orm/postgres" +import { ErrorIssues, ErrorIssueVerifications } from "@maple/db/tables" +import { Array as Arr, ConfigProvider, Effect, Layer, Schema } from "effect" +import { TestClock } from "effect/testing" +import { Database } from "@maple/backend/platform/DatabaseLive" +import { Env } from "@maple/backend/platform/Env" +import { cleanupTestDbs, createTestDb, type TestDb } from "@maple/backend/platform/test-pglite" +import { AuditLogService } from "@maple/backend/services/audit/AuditLogService" +import { + WarehouseQueryService, + type WarehouseQueryServiceApi, +} from "@maple/backend/services/warehouse/WarehouseQueryService" +import { ErrorActorsService } from "./ErrorActorsService" +import { ErrorIssueWorkflowService } from "./ErrorIssueWorkflowService" +import { + FixVerificationTickService, + splitVersionRows, + verdictFromInvestigationStatus, +} from "./FixVerificationTickService" +import { IssueFixVerificationService } from "./IssueFixVerificationService" +import { PullRequestLookup } from "./PullRequestLookup" /** * The verdict mapping is deliberately inverted relative to an incident @@ -91,3 +111,178 @@ describe("splitVersionRows", () => { assert.isAbove(split.unattributed, 0) }) }) + +const ORG = Schema.decodeSync(OrgId)("org_verification_tick") +const DAY = 24 * 60 * 60_000 +const NOW = 1_750_000_000_000 +const PR_URL = "https://github.com/MapleTechLabs/maple/pull/612" + +const createdDbs: TestDb[] = [] +afterEach(async () => { + await cleanupTestDbs(createdDbs) +}) + +const testConfig = () => + ConfigProvider.layer( + ConfigProvider.fromUnknown({ + PORT: "3478", + MCP_PORT: "3479", + TINYBIRD_HOST: "https://api.tinybird.co", + TINYBIRD_TOKEN: "test-token", + MAPLE_AUTH_MODE: "self_hosted", + MAPLE_ROOT_PASSWORD: "test-root-password", + MAPLE_DEFAULT_ORG_ID: "default", + MAPLE_INGEST_KEY_ENCRYPTION_KEY: Buffer.alloc(32, 1).toString("base64"), + MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: "maple-test-lookup-secret", + INTERNAL_SERVICE_TOKEN: "test-internal-token", + }), + ) + +const unexpected = () => Effect.die(new Error("unexpected warehouse call")) + +/** A warehouse whose every occurrence read fails, as a quarantined or broken org's does. */ +const failingWarehouse: WarehouseQueryServiceApi = { + query: unexpected, + rawSqlQuery: unexpected, + crossOrgQuery: unexpected, + compiledQuery: () => Effect.fail(new WarehouseQueryError({ message: "warehouse unavailable" })), + compiledQueryFirst: unexpected, + ingest: () => Effect.void, + asExecutor: () => { + throw new Error("asExecutor is not supported by this test stub") + }, +} + +const makeTickLayer = () => { + const databaseLive = createTestDb(createdDbs).layer + const actorsLive = ErrorActorsService.layer.pipe(Layer.provide(databaseLive)) + const workflowLive = Layer.effect(ErrorIssueWorkflowService, ErrorIssueWorkflowService.make).pipe( + Layer.provide(AuditLogService.layerMemory), + Layer.provide(databaseLive), + Layer.provide(actorsLive), + ) + const verificationLive = Layer.effect(IssueFixVerificationService, IssueFixVerificationService.make).pipe( + Layer.provide( + Layer.mergeAll( + databaseLive, + Env.layer.pipe(Layer.provide(testConfig())), + actorsLive, + workflowLive, + PullRequestLookup.none, + ), + ), + ) + const tickLive = Layer.effect(FixVerificationTickService, FixVerificationTickService.make).pipe( + Layer.provide( + Layer.mergeAll( + databaseLive, + verificationLive, + Layer.succeed(WarehouseQueryService, failingWarehouse), + ), + ), + ) + return Layer.mergeAll(tickLive, verificationLive, databaseLive) +} + +/** An issue in review whose linked PR merged 30 days ago, so its window has long closed. */ +const seedOverdueVerification = Effect.gen(function* () { + const database = yield* Database + const verification = yield* IssueFixVerificationService + const issueId = Schema.decodeSync(ErrorIssueId)(randomUUID()) + yield* database.execute((db) => + db.run( + PG.insertInto(ErrorIssues).values({ + id: issueId, + orgId: ORG, + kind: "error", + fingerprintHash: `fp-${issueId}`, + serviceName: "checkout", + exceptionType: "TypeError", + exceptionMessage: "undefined is not a function", + errorLabel: "", + topFrame: "src/checkout.ts:42", + workflowState: "in_review", + severity: "low", + firstSeenAt: NOW - 40 * DAY, + lastSeenAt: NOW - 31 * DAY, + occurrenceCount: 200, + seenVersionsJson: ["v1"], + createdAt: NOW - 40 * DAY, + updatedAt: NOW - 31 * DAY, + }), + ), + ) + yield* verification.linkPullRequest(ORG, null, issueId, PR_URL, "agent") + yield* verification.onPullRequestEvent({ + orgId: ORG, + provider: "github", + externalRepoId: "999", + repoFullName: "MapleTechLabs/maple", + number: 612, + action: "closed", + url: PR_URL, + title: "Fix the checkout crash", + body: null, + authorLogin: "octocat", + merged: true, + mergeCommitSha: "abc123", + mergedAtMs: NOW - 30 * DAY, + }) + return issueId +}) + +const verificationOf = (issueId: ErrorIssueId) => + Effect.gen(function* () { + const database = yield* Database + const rows = yield* database.execute((db) => + db.run( + PG.from(ErrorIssueVerifications) + .select("status", "verifyAfter", "updatedAt") + .where(($) => [$.issueId.eq(issueId)]), + ), + ) + return rows[0] + }) + +describe("FixVerificationTickService.runTick", () => { + // Twenty rows like this used to fill every tick's page and block every other org. + it.effect("settles a due row whose issue is gone, and does not examine it again", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW) + const tick = yield* FixVerificationTickService + const database = yield* Database + const issueId = yield* seedOverdueVerification + yield* database.execute((db) => + db.run(PG.deleteFrom(ErrorIssues).where(($) => [$.id.eq(issueId)])), + ) + + const first = yield* tick.runTick + assert.strictEqual(first.examined, 1) + assert.strictEqual(first.abandoned, 1) + assert.strictEqual((yield* verificationOf(issueId))?.status, "abandoned") + + assert.strictEqual((yield* tick.runTick).examined, 0) + }).pipe(Effect.provide(makeTickLayer())), + ) + + it.effect("sends a row whose occurrences cannot be read to the back, with its window intact", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW) + const tick = yield* FixVerificationTickService + const issueId = yield* seedOverdueVerification + const before = yield* verificationOf(issueId) + + yield* TestClock.setTime(NOW + 60_000) + const first = yield* tick.runTick + assert.strictEqual(first.skipped, 1) + // Not a verdict: still waiting, and the window an inconclusive verdict measures from is unchanged. + const row = yield* verificationOf(issueId) + assert.strictEqual(row?.status, "waiting") + assert.strictEqual(row?.verifyAfter, before?.verifyAfter) + // Touched, which is what puts it behind rows the tick has not reached yet. + assert.strictEqual(row?.updatedAt, NOW + 60_000) + + assert.strictEqual((yield* tick.runTick).examined, 1) + }).pipe(Effect.provide(makeTickLayer())), + ) +}) diff --git a/packages/backend/src/services/errors/FixVerificationTickService.ts b/packages/backend/src/services/errors/FixVerificationTickService.ts index 51417c2273..fc19c70638 100644 --- a/packages/backend/src/services/errors/FixVerificationTickService.ts +++ b/packages/backend/src/services/errors/FixVerificationTickService.ts @@ -1,4 +1,9 @@ -import { MAX_VERIFICATION_ATTEMPTS, type VerificationVerdict, type OrgId } from "@maple/domain/http" +import { + type ErrorPersistenceError, + MAX_VERIFICATION_ATTEMPTS, + type VerificationVerdict, + type OrgId, +} from "@maple/domain/http" import { RoleName, UserId } from "@maple/domain/primitives" import * as PG from "@maple-dev/effect-orm/postgres" import { ErrorIssuePullRequests, ErrorIssues, type ErrorIssueVerificationRow } from "@maple/db/tables" @@ -104,6 +109,8 @@ export interface FixVerificationTickResult { readonly investigationsStarted: number readonly verdictsApplied: number readonly skipped: number + /** Rows settled as `abandoned` because their issue or pull request link is gone. */ + readonly abandoned: number /** Rows whose verdict could not be written this tick; they are due again next minute. */ readonly failedRows: number } @@ -209,6 +216,29 @@ const make: Effect.Effect< ), ) + // The two writes that are not verdicts, guarded per row for the same reason. + const guardedWrite = ( + row: ErrorIssueVerificationRow, + action: "abandon" | "defer", + write: Effect.Effect, + ) => + write.pipe( + Effect.as(true), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logError("[FixVerification] could not update a verification").pipe( + Effect.annotateLogs({ + orgId: row.orgId, + verificationId: row.id, + action, + error: summarizeCause(cause), + }), + Effect.as(false), + ), + ), + ) + // Phase 1: settle runs that have finished since the last tick. Done first so // a verification that already has an answer is not competing for this // minute's budget with one that still needs an agent. @@ -239,7 +269,16 @@ const make: Effect.Effect< ), ) const subject = context[0] - if (subject === undefined) return "skipped" + if (subject === undefined) { + // The issue or its link is gone, so no later tick can judge this row. + // Left `waiting` it would hold a slot in every tick's page forever. + const abandoned = yield* guardedWrite( + row, + "abandon", + verification.abandonVerification(row, nowMs), + ) + return abandoned ? "abandoned" : "failed" + } const split = yield* occurrenceSplit(row, subject.fingerprintHash, nowMs).pipe( Effect.asSome, @@ -255,6 +294,15 @@ const make: Effect.Effect< error: summarizeCause(cause), }), // Leave the row waiting; the warehouse being down is not a verdict. + // It goes to the back of the due queue, so an org whose warehouse + // keeps failing does not sit ahead of every other org. + Effect.andThen( + guardedWrite( + row, + "defer", + verification.deferVerification(row, nowMs), + ), + ), Effect.as( Option.none<{ postMerge: number @@ -368,6 +416,7 @@ const make: Effect.Effect< const refuted = countOutcome("refuted") const investigationsStarted = countOutcome("started") const skipped = countOutcome("skipped") + const abandoned = countOutcome("abandoned") const verdictsApplied = Arr.filter(settledApplied, Boolean).length + countOutcome("verdict") const failedRows = Arr.filter(settledApplied, (applied) => !applied).length + countOutcome("failed") @@ -377,10 +426,19 @@ const make: Effect.Effect< "maple.verification.investigations_started": investigationsStarted, "maple.verification.verdicts_applied": verdictsApplied, "maple.verification.skipped": skipped, + "maple.verification.abandoned": abandoned, "maple.verification.failed_rows": failedRows, }) - return { examined: due.length, refuted, investigationsStarted, verdictsApplied, skipped, failedRows } + return { + examined: due.length, + refuted, + investigationsStarted, + verdictsApplied, + skipped, + abandoned, + failedRows, + } }).pipe(Effect.withSpan("FixVerificationTick.runTick")) const runTickSafely: FixVerificationTickServiceApi["runTick"] = runTick.pipe( @@ -399,6 +457,7 @@ const make: Effect.Effect< investigationsStarted: 0, verdictsApplied: 0, skipped: 0, + abandoned: 0, failedRows: 0, } satisfies FixVerificationTickResult), ), diff --git a/packages/backend/src/services/errors/IssueFixVerificationService.test.ts b/packages/backend/src/services/errors/IssueFixVerificationService.test.ts index f4f59f5116..e317611c7c 100644 --- a/packages/backend/src/services/errors/IssueFixVerificationService.test.ts +++ b/packages/backend/src/services/errors/IssueFixVerificationService.test.ts @@ -650,6 +650,62 @@ describe("onPullRequestEvent — merge", () => { }), ) + // GitHub reports `merged: true` on every action after the merge, so a title + // edit or a redelivery months later arrives looking exactly like the merge. + it.effect("does not reopen a settled verification when the merged PR is edited later", () => + Effect.gen(function* () { + const layer = makeLayer() + yield* Effect.gen(function* () { + const service = yield* IssueFixVerificationService + const issueId = yield* seedIssue({ severity: "low" }) + yield* service.linkPullRequest(ORG, null, issueId, PR_URL, "agent") + yield* service.onPullRequestEvent(mergeEvent()) + yield* service.applyVerdict( + yield* readVerification(issueId), + "verified", + "No occurrences.", + yield* Clock.currentTimeMillis, + ) + expect((yield* readIssueState(issueId))?.workflowState).toBe("done") + + const later = yield* service.onPullRequestEvent( + mergeEvent({ action: "edited", title: "Fix the checkout crash (typo)" }), + ) + expect(later.verificationsOpened).toBe(0) + expect(yield* readAllVerifications(issueId)).toHaveLength(1) + expect((yield* readIssueState(issueId))?.workflowState).toBe("done") + const merges = Arr.filter(yield* readEventTypes(issueId), (type) => type === "pr_merged") + expect(merges).toHaveLength(1) + }).pipe(Effect.provide(layer)) + }), + ) + + // The guard is per link, not per issue: detaching a PR and attaching it again + // is somebody asking for the check to run again. + it.effect("verifies again after the PR is unlinked and linked back", () => + Effect.gen(function* () { + const layer = makeLayer() + yield* Effect.gen(function* () { + const service = yield* IssueFixVerificationService + const issueId = yield* seedIssue() + const link = yield* service.linkPullRequest(ORG, null, issueId, PR_URL, "agent") + yield* service.onPullRequestEvent(mergeEvent()) + yield* service.applyVerdict( + yield* readVerification(issueId), + "not_fixed", + "Still firing.", + yield* Clock.currentTimeMillis, + ) + + yield* service.unlinkPullRequest(ORG, null, issueId, link.id) + yield* service.linkPullRequest(ORG, null, issueId, PR_URL, "user") + const again = yield* service.onPullRequestEvent(mergeEvent()) + expect(again.verificationsOpened).toBe(1) + expect((yield* readIssueState(issueId))?.workflowState).toBe("verifying") + }).pipe(Effect.provide(layer)) + }), + ) + it.effect("marks the link closed, and opens nothing, when a PR closes unmerged", () => Effect.gen(function* () { const layer = makeLayer() @@ -975,6 +1031,46 @@ describe("dueVerifications", () => { }).pipe(Effect.provide(layer)) }), ) + + // A page with no order lets the same rows fill every tick while others wait. + it.effect("returns the least recently touched rows first, so a deferred row goes to the back", () => + Effect.gen(function* () { + const layer = makeLayer() + yield* Effect.gen(function* () { + const service = yield* IssueFixVerificationService + const nowMs = yield* Clock.currentTimeMillis + const recent = yield* seedIssue() + const overdue = yield* seedIssue() + yield* service.linkPullRequest(ORG, null, recent, PR_URL, "agent") + yield* service.linkPullRequest( + ORG, + null, + overdue, + `https://github.com/${REPO}/pull/613`, + "agent", + ) + yield* service.onPullRequestEvent(mergeEvent({ mergedAtMs: nowMs })) + yield* service.onPullRequestEvent(mergeEvent({ number: 613, mergedAtMs: nowMs - 30 * DAY })) + const dueAt = nowMs + 365 * DAY + const rows = yield* service.dueVerifications(dueAt, 10) + const defer = (issueId: ErrorIssueId, atMs: number) => + Effect.forEach( + Arr.filter(rows, (row) => row.issueId === issueId), + (row) => service.deferVerification(row, atMs), + ) + + // Deferring one row puts the other ahead of it, whichever was opened first. + yield* defer(recent, dueAt) + expect(Arr.map(yield* service.dueVerifications(dueAt, 1), (row) => row.issueId)).toEqual([ + overdue, + ]) + yield* defer(overdue, dueAt + 1) + expect(Arr.map(yield* service.dueVerifications(dueAt + 1, 1), (row) => row.issueId)).toEqual([ + recent, + ]) + }).pipe(Effect.provide(layer)) + }), + ) }) /** An effect-orm insert into `table`, read off the builder's runtime shape. */ diff --git a/packages/backend/src/services/errors/IssueFixVerificationService.ts b/packages/backend/src/services/errors/IssueFixVerificationService.ts index 26c7e07c41..9f29022016 100644 --- a/packages/backend/src/services/errors/IssueFixVerificationService.ts +++ b/packages/backend/src/services/errors/IssueFixVerificationService.ts @@ -198,6 +198,22 @@ export interface IssueFixVerificationServiceApi { investigationId: InvestigationId | null, nowMs: number, ) => Effect.Effect + /** + * Settle a live row whose issue or pull request link no longer exists. There is + * nothing left to judge, and left `waiting` it would be re-examined every tick. + */ + readonly abandonVerification: ( + row: ErrorIssueVerificationRow, + nowMs: number, + ) => Effect.Effect + /** + * Send a row that cannot be read yet to the back of the due queue. Its window is + * left alone: an inconclusive verdict measures its retry from `verifyAfter`. + */ + readonly deferVerification: ( + row: ErrorIssueVerificationRow, + nowMs: number, + ) => Effect.Effect readonly applyVerdict: ( row: ErrorIssueVerificationRow, verdict: VerificationVerdict, @@ -752,22 +768,23 @@ const make: Effect.Effect< const openForLink = Effect.fn("IssueFixVerificationService.openForLink")(function* ( link: (typeof links)[number], ) { - // One live verification per issue. A `synchronize` after a merge, a - // redelivered webhook, or a second PR attached to the same issue must - // not open a second window. - const open = yield* dbExecute((db) => + // One live verification per issue, and one verification per link, ever. + // GitHub sends `merged: true` on every later action, so an edit to a + // merged PR or a redelivery must not reopen a window that already settled. + // Keyed on the link id: an unlink and re-link is a new link and verifies again. + const existing = yield* dbExecute((db) => db.run( PG.from(ErrorIssueVerifications) - .select() + .select("id") .where(($) => [ $.orgId.eq(orgId), $.issueId.eq(link.issueId), - $.status.in_("waiting", "running"), + PG.or($.status.in_("waiting", "running"), $.pullRequestId.eq(link.id)), ]) .limit(1), ), ) - if (Arr.isReadonlyArrayNonEmpty(open)) return false + if (Arr.isReadonlyArrayNonEmpty(existing)) return false const issueRows = yield* dbExecute((db) => db.run( @@ -1011,6 +1028,9 @@ const make: Effect.Effect< PG.from(ErrorIssueVerifications) .select() .where(($) => [$.status.eq("waiting"), $.verifyAfter.lte(nowMs)]) + // Least recently touched first, so a row the tick deferred goes to the + // back. Unordered, the same rows could fill every page while the rest never came up. + .orderBy(["updatedAt", "asc"]) .limit(limit), ), ) @@ -1037,6 +1057,30 @@ const make: Effect.Effect< ) }) + const abandonVerification: IssueFixVerificationServiceApi["abandonVerification"] = Effect.fn( + "IssueFixVerificationService.abandonVerification", + )(function* (row, nowMs) { + yield* dbExecute((db) => + db.run( + PG.update(ErrorIssueVerifications) + .set({ status: "abandoned", updatedAt: nowMs }) + .where(($) => [$.id.eq(row.id), $.status.eq(row.status), $.attempt.eq(row.attempt)]), + ), + ) + }) + + const deferVerification: IssueFixVerificationServiceApi["deferVerification"] = Effect.fn( + "IssueFixVerificationService.deferVerification", + )(function* (row, nowMs) { + yield* dbExecute((db) => + db.run( + PG.update(ErrorIssueVerifications) + .set({ updatedAt: nowMs }) + .where(($) => [$.id.eq(row.id), $.status.eq(row.status), $.attempt.eq(row.attempt)]), + ), + ) + }) + const applyVerdict: IssueFixVerificationServiceApi["applyVerdict"] = Effect.fn( "IssueFixVerificationService.applyVerdict", )(function* (row, verdict, note, nowMs) { @@ -1217,6 +1261,8 @@ const make: Effect.Effect< refuteOnPostMergeOccurrence, dueVerifications, markRunning, + abandonVerification, + deferVerification, applyVerdict, } satisfies IssueFixVerificationServiceApi }) diff --git a/scripts/ci/plan-tests.ts b/scripts/ci/plan-tests.ts index 7095810699..b7e63d4bf6 100644 --- a/scripts/ci/plan-tests.ts +++ b/scripts/ci/plan-tests.ts @@ -7,9 +7,11 @@ const ROOT = path.resolve(import.meta.dir, "../..") // Vitest seconds/file at two workers, measured on CI run 35725366619. Unknown suites // get the default, fitted to the slowest cold small-suite lanes of 2026-10. +// Backend was 1.6 then; by run 42756486011 its four shards took 185-239s against the +// 240s lane budget, about 2.0s/file once setup is taken out. const SECONDS_PER_FILE = new Map([ ["@maple/api", 2.3], - ["@maple/backend", 1.6], + ["@maple/backend", 2.0], ["@maple/ai", 1.1], ["@maple/query-engine", 0.45], ["@maple/ui", 0.45],