diff --git a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.test.ts b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.test.ts index 1bdf1a37b7..c9d03837b5 100644 --- a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.test.ts +++ b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.test.ts @@ -1,6 +1,9 @@ import { afterEach, assert, describe, it } from "@effect/vitest" import * as Cloudflare from "@alchemy.run/cloudflare" -import { Cause, Effect, Exit, Layer, Schema, Array as Arr } from "effect" +import { Cause, Effect, Exit, Fiber, Layer, Schema, Array as Arr } from "effect" +import { TestClock } from "effect/testing" +import { Database } from "@maple/backend/platform/DatabaseLive" +import { OrgId } from "@maple/domain/http" import { FetchHttpClient } from "effect/http" import { @@ -12,7 +15,11 @@ import { } from "@maple/backend/platform/test-pglite" import { durableStep } from "@maple/backend/platform/durable-step" import { workerEnvLayer } from "@maple/infra/worker-runtime" -import { loadOptionalFeatureState, runClickHouseSchemaApply } from "./ClickHouseSchemaApplyWorkflow.run" +import { + loadOptionalFeatureState, + runClickHouseSchemaApply, + withRunHeartbeat, +} from "./ClickHouseSchemaApplyWorkflow.run" /** The step's Effect as `task` hands it over, with the body context already provided. */ const stepEffect = (options: Cloudflare.WorkflowTaskOptions): Effect.Effect => @@ -85,8 +92,8 @@ describe("runClickHouseSchemaApply failure bookkeeping", () => { const ENCRYPTION_KEY = Buffer.alloc(32, 5).toString("base64") /** The run as the alchemy class provides it (env as `ConfigProvider`), minus the connection scope and telemetry. */ - const runFor = (testDb: TestDb, env: Record, orgId: string) => - runClickHouseSchemaApply({ orgId }).pipe( + const runFor = (testDb: TestDb, env: Record, orgId: string, runId = "run-1") => + runClickHouseSchemaApply({ orgId, runId }).pipe( Effect.provide([ testDb.layer, Layer.succeed(Cloudflare.WorkflowStep, inlineStep), @@ -97,25 +104,90 @@ describe("runClickHouseSchemaApply failure bookkeeping", () => { Effect.exit, ) - const seedQueuedRun = (testDb: TestDb, orgId: string) => + const seedQueuedRun = (testDb: TestDb, orgId: string, runId = "run-1", status = "queued") => + Effect.promise(() => + executeSql( + testDb, + `INSERT INTO org_clickhouse_schema_apply_runs (org_id, workflow_instance_id, status, phase, created_at, updated_at) + VALUES ($1, $2, $3, 'queued', now(), now())`, + [orgId, runId, status], + ), + ) + + /** A settings row loadConfig accepts: no password, plain HTTP to a port nothing listens on. */ + const seedSettings = (testDb: TestDb, orgId: string) => Effect.promise(() => executeSql( testDb, - `INSERT INTO org_clickhouse_schema_apply_runs (org_id, status, phase, created_at, updated_at) - VALUES ($1, 'queued', 'queued', now(), now())`, + `INSERT INTO org_clickhouse_settings (org_id, ch_url, ch_user, ch_database, sync_status, created_at, updated_at, created_by, updated_by) + VALUES ($1, 'http://127.0.0.1:9', 'default', 'default', 'connected', now(), now(), 'u', 'u')`, [orgId], ), ) const readRun = (testDb: TestDb, orgId: string) => Effect.promise(() => - queryFirstRow<{ status: string; error_message: string | null }>( + queryFirstRow<{ + status: string + error_message: string | null + workflow_instance_id: string | null + phase: string | null + }>( testDb, - "SELECT status, error_message FROM org_clickhouse_schema_apply_runs WHERE org_id = $1", + "SELECT status, error_message, workflow_instance_id, phase FROM org_clickhouse_schema_apply_runs WHERE org_id = $1", [orgId], ), ) + const readSyncStatus = (testDb: TestDb, orgId: string) => + Effect.promise(() => + queryFirstRow<{ sync_status: string }>( + testDb, + "SELECT sync_status FROM org_clickhouse_settings WHERE org_id = $1", + [orgId], + ), + ) + + it.effect("a reclaimed instance stops before touching the cluster and leaves the new run alone", () => + Effect.gen(function* () { + const testDb = createTestDb(trackedDbs) + // A newer applySchema reclaimed the row for run-new; run-old wakes up afterwards. + yield* seedQueuedRun(testDb, "org_wf_stale", "run-new", "running") + yield* seedSettings(testDb, "org_wf_stale") + + const exit = yield* runFor( + testDb, + { MAPLE_INGEST_KEY_ENCRYPTION_KEY: ENCRYPTION_KEY }, + "org_wf_stale", + "run-old", + ) + assert.deepStrictEqual(exit, Exit.succeed({ status: "superseded", appliedVersions: [] })) + + const row = yield* readRun(testDb, "org_wf_stale") + assert.strictEqual(row?.workflow_instance_id, "run-new") + assert.strictEqual(row?.status, "running") + assert.strictEqual(row?.phase, "queued") + assert.strictEqual((yield* readSyncStatus(testDb, "org_wf_stale"))?.sync_status, "connected") + }), + ) + + it.effect("a reclaimed instance that fails cannot mark the new run failed", () => + Effect.gen(function* () { + const testDb = createTestDb(trackedDbs) + yield* seedQueuedRun(testDb, "org_wf_stale_fail", "run-new") + yield* seedSettings(testDb, "org_wf_stale_fail") + + // No encryption key: run-old fails before its first step and takes the failure path. + const exit = yield* runFor(testDb, {}, "org_wf_stale_fail", "run-old") + assert.include(failureMessage(exit), "MAPLE_INGEST_KEY_ENCRYPTION_KEY") + + const row = yield* readRun(testDb, "org_wf_stale_fail") + assert.strictEqual(row?.status, "queued") + assert.strictEqual(row?.error_message, null) + assert.strictEqual((yield* readSyncStatus(testDb, "org_wf_stale_fail"))?.sync_status, "connected") + }), + ) + /** A step that spent its retries and a typed failure outside the steps both stay in the error channel. */ const failureMessage = (exit: Exit.Exit): string => { assert.isTrue(Exit.isFailure(exit)) @@ -161,3 +233,53 @@ describe("runClickHouseSchemaApply failure bookkeeping", () => { }), ) }) + +describe("withRunHeartbeat", () => { + const trackedDbs: TestDb[] = [] + afterEach(() => cleanupTestDbs(trackedDbs)) + const ORG = Schema.decodeSync(OrgId)("org_wf_heartbeat") + + const updatedAt = (testDb: TestDb) => + Effect.promise(() => + queryFirstRow<{ updated_at: Date }>( + testDb, + "SELECT updated_at FROM org_clickhouse_schema_apply_runs WHERE org_id = $1", + [ORG], + ), + ).pipe(Effect.map((row) => row?.updated_at.getTime())) + + it.effect("keeps a long statement's run fresh, and abandons it once the run is reclaimed", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + executeSql( + testDb, + `INSERT INTO org_clickhouse_schema_apply_runs (org_id, workflow_instance_id, status, created_at, updated_at) + VALUES ($1, 'run-1', 'running', to_timestamp(0), to_timestamp(0))`, + [ORG], + ), + ) + const fence = { database: yield* Database, orgId: ORG, runId: "run-1" } + // A backfill chunk that outlasts the staleness window. + const fiber = yield* withRunHeartbeat(fence, Effect.never).pipe(Effect.forkChild) + + yield* TestClock.adjust("5 minutes") + assert.strictEqual(yield* updatedAt(testDb), 5 * 60_000) + + yield* Effect.promise(() => + executeSql( + testDb, + "UPDATE org_clickhouse_schema_apply_runs SET workflow_instance_id = 'run-2' WHERE org_id = $1", + [ORG], + ), + ) + yield* TestClock.adjust("5 minutes") + const exit = yield* Fiber.await(fiber) + assert.isTrue(Exit.isFailure(exit)) + if (Exit.isFailure(exit)) { + assert.include(String(Cause.squash(exit.cause)), "SchemaApplyRunSuperseded") + } + assert.strictEqual(yield* updatedAt(testDb), 5 * 60_000) + }).pipe(Effect.provide(testDb.layer)) + }) +}) diff --git a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts index 52dfa7dd89..750d2d4014 100644 --- a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts +++ b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts @@ -32,7 +32,18 @@ import { } from "@maple/domain/clickhouse" import { OrgId } from "@maple/domain/http" import * as Cloudflare from "@alchemy.run/cloudflare" -import { Cause, Clock, Config, Effect, Option, Redacted, Schema, Array as Arr, String as Str } from "effect" +import { + Cause, + Clock, + Config, + Effect, + Option, + Predicate, + Redacted, + Schema, + Array as Arr, + String as Str, +} from "effect" import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/http" import { EdgeCacheService } from "@maple/cache" import { EdgeCacheServiceLive } from "@maple/backend/platform/CacheBackendLive" @@ -42,6 +53,7 @@ import { ORG_CH_CONFIG_CACHE_BUCKET, } from "@maple/backend/services/org/OrgClickHouseSettingsService" import { durableStep, type DurableStepConfig } from "@maple/backend/platform/durable-step" +import { transitionRow } from "@maple/backend/platform/transition-row" /** * Bust the cached runtime config after the workflow writes to @@ -73,10 +85,12 @@ const bustRuntimeConfigCache = (orgId: OrgId): Effect.Effect => export interface SchemaApplyWorkflowPayload { readonly orgId: string + /** The claim's run id (`workflow_instance_id`); every progress write is fenced on it. */ + readonly runId: string } export interface SchemaApplyWorkflowResult { - readonly status: "succeeded" | "failed" + readonly status: "succeeded" | "superseded" readonly appliedVersions: ReadonlyArray } @@ -106,6 +120,16 @@ const causeMessage = (cause: Cause.Cause): string => { return error instanceof Error ? error.message : String(error) } +/** An optional step may be skipped, but a lost fence must still stop the run. */ +const isSkippable = (cause: Cause.Cause): boolean => + !Cause.hasInterrupts(cause) && + !Arr.some( + cause.reasons, + (reason) => + Cause.isFailReason(reason) && + Predicate.isTagged(reason.error, "@maple/api/workflows/SchemaApplyRunSupersededError"), + ) + /** * Optional feature metadata must never decide whether the correctness schema * is ready for ingest. Keeping this boundary explicit also makes the workflow @@ -300,6 +324,12 @@ export class SchemaApplyPayloadError extends Schema.TaggedError()( + "@maple/api/workflows/SchemaApplyRunSupersededError", + { message: Schema.String, orgId: OrgId, runId: Schema.String }, +) {} + /** A settings row that cannot be turned into a usable ClickHouse target. */ export class SchemaApplyConfigError extends Schema.TaggedError()( "@maple/api/workflows/SchemaApplyConfigError", @@ -393,18 +423,80 @@ type RunPatch = Partial<{ finishedAt: number | null }> -/** Stamps `updatedAt` from the clock, so it only ever runs inside a step (or on the failure path). */ -const updateRun = (database: DatabaseApi, orgId: OrgId, patch: RunPatch) => +type RunStatus = "queued" | "running" | "succeeded" | "failed" + +/** The run row this instance owns: the org's row while it still carries the claim's run id. */ +export interface RunFence { + readonly database: DatabaseApi + readonly orgId: OrgId + readonly runId: string +} + +/** + * Stamps `updatedAt` from the clock, so it only ever runs inside a step (or on the failure path). + * Fenced on the run id and a live status: a reclaimed or settled run fails it as superseded. + */ +const updateRun = (fence: RunFence, patch: RunPatch) => Effect.gen(function* () { const now = yield* Clock.currentTimeMillis - yield* database.execute((db) => - db.run( - PG.update(OrgClickHouseSchemaApplyRuns) - .set({ ...patch, updatedAt: now }) - .where(($) => [$.orgId.eq(orgId)]), - ), + yield* transitionRow( + { + table: "org_clickhouse_schema_apply_runs", + id: fence.orgId, + from: ["queued", "running"], + to: patch.status ?? "running", + }, + (from) => + fence.database.execute((db) => + db.run( + PG.update(OrgClickHouseSchemaApplyRuns) + .set({ ...patch, updatedAt: now }) + .where(($) => [ + $.orgId.eq(fence.orgId), + $.workflowInstanceId.eq(fence.runId), + $.status.in_(...from), + ]) + .returning("orgId"), + ), + ), ) - }) + }).pipe( + Effect.catchTag("@maple/backend/TransitionConflictError", (conflict) => + Effect.fail( + new SchemaApplyRunSupersededError({ + message: conflict.message, + orgId: fence.orgId, + runId: fence.runId, + }), + ), + ), + ) + +/** Under the 30 minute staleness window the service reclaims a silent run after. */ +const RUN_HEARTBEAT = "5 minutes" + +/** + * Runs one long statement (a backfill chunk) while touching the run row, so a live run never + * reads as stale. Losing the fence abandons the request (the server may still finish it). + */ +export const withRunHeartbeat = (fence: RunFence, statement: Effect.Effect) => + Effect.raceFirst( + statement, + Effect.sleep(RUN_HEARTBEAT).pipe( + Effect.andThen( + updateRun(fence, {}).pipe( + Effect.catchTag("@maple/backend/lib/DatabaseError", (error) => + Effect.logWarning("Schema apply heartbeat failed", error.message), + ), + ), + ), + Effect.forever, + ), + ) + +/** Claims the step for this run, then runs its statement under the heartbeat. */ +const fencedExec = (fence: RunFence, cfg: ChConfig, patch: RunPatch, sql: string) => + updateRun(fence, patch).pipe(Effect.andThen(withRunHeartbeat(fence, execClickHouse(cfg, sql)))) // --- snapshot-diff additive (mirror of OrgClickHouseSettingsService) -------- @@ -515,7 +607,14 @@ export const runClickHouseSchemaApply = ( (error) => new SchemaApplyPayloadError({ message: error.message, rawOrgId: payload.orgId }), ), ) - const database = yield* Database + if (typeof payload.runId !== "string" || payload.runId.length === 0) { + return yield* new SchemaApplyPayloadError({ + message: "Schema apply payload has no run id", + rawOrgId: payload.orgId, + }) + } + const fence: RunFence = { database: yield* Database, orgId, runId: payload.runId } + const { database } = fence // Inside the protected region below: a config-load failure (settings row // deleted, missing encryption key, decrypt failure, invalid URL) must still @@ -525,11 +624,23 @@ export const runClickHouseSchemaApply = ( Effect.gen(function* () { const finishedAt = yield* Clock.currentTimeMillis const message = causeMessage(cause) - yield* updateRun(database, orgId, { + // Only the run that still owns the row may report the org's sync error. + const owned = yield* updateRun(fence, { status: "failed", errorMessage: message, finishedAt, - }).pipe(Effect.ignore({ log: true, message: "Failed to mark schema apply run failed" })) + }).pipe( + Effect.as(true), + Effect.catchTag("@maple/api/workflows/SchemaApplyRunSupersededError", () => + Effect.succeed(false), + ), + Effect.catch((error) => + Effect.logWarning("Failed to mark schema apply run failed", error.message).pipe( + Effect.as(true), + ), + ), + ) + if (!owned) return yield* database .execute((db) => db.run( @@ -551,13 +662,18 @@ export const runClickHouseSchemaApply = ( yield* bustRuntimeConfigCache(orgId) }) - return yield* applySchema(database, orgId).pipe( + return yield* applySchema(fence).pipe( + // A reclaimed run ends quietly: the newer run owns the row and the org's sync status. + Effect.catchTag("@maple/api/workflows/SchemaApplyRunSupersededError", () => + Effect.succeed({ status: "superseded", appliedVersions: [] }), + ), Effect.catchCause((cause) => markFailed(cause).pipe(Effect.andThen(Effect.failCause(cause)))), ) }) -const applySchema = (database: DatabaseApi, orgId: OrgId) => +const applySchema = (fence: RunFence) => Effect.gen(function* () { + const { database, orgId } = fence const encryptionKeyText = yield* Config.Redacted(ENCRYPTION_KEY_CONFIG).pipe( Effect.mapError( (error) => @@ -581,7 +697,7 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => Effect.gen(function* () { const c = yield* loadConfig(database, orgId, encryptionKey) const startedAt = yield* Clock.currentTimeMillis - yield* updateRun(database, orgId, { + yield* updateRun(fence, { status: "running", phase: "connecting", errorMessage: null, @@ -615,12 +731,17 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => // would repeat, and a clock read there would differ per replay. durableStep( `m${migration.version}:${s.name}`, - updateRun(database, orgId, { - phase: `migration ${migration.version} · ${s.name}`, - currentMigration: migration.version, - stepsTotal: steps.length, - stepsDone: index, - }).pipe(Effect.andThen(execClickHouse(cfg, s.sql))), + fencedExec( + fence, + cfg, + { + phase: `migration ${migration.version} · ${s.name}`, + currentMigration: migration.version, + stepsTotal: steps.length, + stepsDone: index, + }, + s.sql, + ), STEP, ), { discard: true }, @@ -639,8 +760,8 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => // cluster + add missing columns (metadata-only, fits a step easily). yield* durableStep( "snapshot-diff", - updateRun(database, orgId, { phase: "reconciling schema snapshot" }).pipe( - Effect.andThen(reconcileSchemaSnapshot(cfg)), + updateRun(fence, { phase: "reconciling schema snapshot" }).pipe( + Effect.andThen(withRunHeartbeat(fence, reconcileSchemaSnapshot(cfg))), ), STEP, ) @@ -651,6 +772,8 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => yield* durableStep( "stamp-correctness", Effect.gen(function* () { + // Only the run that still owns the row speaks for the org's settings. + yield* updateRun(fence, { phase: "stamping schema version" }) const stampedAt = yield* Clock.currentTimeMillis yield* database.execute((db) => db.run( @@ -686,12 +809,17 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => (migrationStep, index) => durableStep( `performance-m${migration.version}:${migrationStep.name}`, - updateRun(database, orgId, { - phase: `performance migration ${migration.version} · ${index + 1}/${steps.length}`, - currentMigration: migration.version, - stepsTotal: steps.length, - stepsDone: index, - }).pipe(Effect.andThen(execClickHouse(cfg, migrationStep.sql))), + fencedExec( + fence, + cfg, + { + phase: `performance migration ${migration.version} · ${index + 1}/${steps.length}`, + currentMigration: migration.version, + stepsTotal: steps.length, + stepsDone: index, + }, + migrationStep.sql, + ), STEP, ), { discard: true }, @@ -706,7 +834,7 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => }) // A performance migration that cannot be installed is skipped, never fatal. return yield* applyPerformanceMigration.pipe( - Effect.catchCause((cause) => + Effect.catchCauseIf(isSkippable, (cause) => Effect.sync(() => { skippedFeatures.push({ id: `migration_${migration.version}`, @@ -779,9 +907,14 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => (statement, index) => durableStep( `feature-${feature.id}:${index + 1}`, - updateRun(database, orgId, { - phase: `feature ${feature.id} · ${index + 1}/${feature.statements.length}`, - }).pipe(Effect.andThen(execClickHouse(cfg, statement))), + fencedExec( + fence, + cfg, + { + phase: `feature ${feature.id} · ${index + 1}/${feature.statements.length}`, + }, + statement, + ), STEP, ), { discard: true }, @@ -794,7 +927,7 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => ) }) return yield* applyFeature.pipe( - Effect.catchCause((cause) => + Effect.catchCauseIf(isSkippable, (cause) => Effect.sync(() => { skippedFeatures.push({ id: feature.id, reason: causeMessage(cause) }) }), @@ -809,7 +942,7 @@ const applySchema = (database: DatabaseApi, orgId: OrgId) => "finalize", Effect.gen(function* () { const finishedAt = yield* Clock.currentTimeMillis - yield* updateRun(database, orgId, { + yield* updateRun(fence, { status: "succeeded", phase: "done", currentMigration: null, diff --git a/packages/backend/src/platform/bindings.ts b/packages/backend/src/platform/bindings.ts index 8d9836152c..c76c7653f1 100644 --- a/packages/backend/src/platform/bindings.ts +++ b/packages/backend/src/platform/bindings.ts @@ -147,7 +147,7 @@ export interface WorkflowStarter { /** The per-org ClickHouse schema apply, hosted by the api Worker. */ export class SchemaApplyWorkflow extends Context.Service< SchemaApplyWorkflow, - WorkflowStarter<{ readonly orgId: string }> + WorkflowStarter<{ readonly orgId: string; readonly runId: string }> >()("@maple/backend/platform/SchemaApplyWorkflow") {} export class WorkflowCancelError extends Schema.TaggedError()( diff --git a/packages/backend/src/services/dependency-scan/DependencyScanStore.ts b/packages/backend/src/services/dependency-scan/DependencyScanStore.ts index cf2d64636d..fdda390aec 100644 --- a/packages/backend/src/services/dependency-scan/DependencyScanStore.ts +++ b/packages/backend/src/services/dependency-scan/DependencyScanStore.ts @@ -106,6 +106,8 @@ export interface SetFindingStatusInput { readonly actor: UserId } +const ABANDONED_SCAN_ERROR = "The scan stopped before it finished" + export interface DependencyScanStoreApi { readonly startScan: ( input: StartScanInput, @@ -117,6 +119,15 @@ export interface DependencyScanStoreApi { DependencyScanRow, DependencyScanPersistenceError | DependencyScanNotFoundError | DependencyScanNotRunningError > + /** + * Settles a repository's scans still `running` past `startedBefore` as failed: their worker + * died. A scan that settled first is left as it is. Returns how many were failed. + */ + readonly failAbandonedScans: (input: { + readonly orgId: OrgId + readonly repositoryId: VcsRepositoryId + readonly startedBefore: number + }) => Effect.Effect readonly failScan: (input: { readonly orgId: OrgId readonly scanId: DependencyScanId @@ -589,6 +600,26 @@ export class DependencyScanStore extends Context.Service + db.run( + PG.update(DependencyScans) + .set({ status: "failed", error: ABANDONED_SCAN_ERROR, finishedAt: now }) + .where(($) => [ + $.orgId.eq(input.orgId), + $.repositoryId.eq(input.repositoryId), + $.status.eq("running"), + $.startedAt.lt(input.startedBefore), + ]) + .returning("id"), + ), + ) + return failed.length + }) + const failScan: DependencyScanStoreApi["failScan"] = Effect.fn("DependencyScanStore.failScan")( function* (input) { const now = yield* Clock.currentTimeMillis @@ -818,6 +849,7 @@ export class DependencyScanStore extends Context.Service { }).pipe(Effect.provide(scannerFor(state))) }) + it.effect("settles a scan abandoned in running as failed when the repository is scanned again", () => { + const state = world({ "package-lock.json": packageLock("4.17.15") }) + return Effect.gen(function* () { + const scanner = yield* DependencyScanner + const store = yield* DependencyScanStore + // A worker that died mid-scan: nothing ever settled its row. + const abandoned = yield* store.startScan({ + orgId: repository.orgId, + repositoryId: repository.id, + commitSha: sha("9"), + ref: "main", + trigger: "push", + defaultBranchHead: true, + }) + yield* TestClock.adjust(`${RESCAN_AFTER_MS + 1} millis`) + yield* scanner.scan(installation, repository, { trigger: "schedule" }) + + const scans = yield* store.listScans({ orgId: repository.orgId, limit: 10 }) + const settled = Arr.findFirst(scans, (scan) => scan.id === abandoned.id) + assert.strictEqual(Option.getOrUndefined(settled)?.status, "failed") + assert.isFalse(Arr.some(scans, (scan) => scan.status === "running")) + }).pipe(Effect.provide(scannerFor(state))) + }) + it.effect( "fails a scan whose tree listing was cut short instead of fixing the unlisted lockfiles' findings", () => { diff --git a/packages/backend/src/services/dependency-scan/DependencyScanner.ts b/packages/backend/src/services/dependency-scan/DependencyScanner.ts index d38e8ce0f1..16f1b3fcf5 100644 --- a/packages/backend/src/services/dependency-scan/DependencyScanner.ts +++ b/packages/backend/src/services/dependency-scan/DependencyScanner.ts @@ -306,6 +306,24 @@ export class DependencyScanner extends Context.Service (cause: unknown) => new DependencyScanError({ message, repositoryId: repository.id, cause }) + // A scan whose worker died stays `running`; each new scan of the repo settles those. + const startedAt = yield* Clock.currentTimeMillis + yield* store + .failAbandonedScans({ + orgId, + repositoryId: repository.id, + startedBefore: startedAt - RESCAN_AFTER_MS, + }) + .pipe( + Effect.catchTag( + "@maple/dependency-scan/DependencyScanPersistenceError", + (error) => + Effect.logWarning( + "Could not settle abandoned dependency scans", + error.message, + ), + ), + ) const provider = yield* sources .resolve(repository.provider) .pipe(Effect.mapError(scanError("Unknown VCS provider"))) diff --git a/packages/backend/src/services/errors/RecommendationIssueService.test.ts b/packages/backend/src/services/errors/RecommendationIssueService.test.ts new file mode 100644 index 0000000000..196993880f --- /dev/null +++ b/packages/backend/src/services/errors/RecommendationIssueService.test.ts @@ -0,0 +1,121 @@ +import { afterEach, assert, describe, it } from "@effect/vitest" +import { Effect, Layer, Schema } from "effect" +import { OrgId, RecommendationIssueId, UserId } from "@maple/domain/http" +import { Database } from "@maple/backend/platform/DatabaseLive" +import { + cleanupTestDbs, + createTestDb, + executeSql, + queryFirstRow, + type TestDb, +} from "@maple/backend/platform/test-pglite" +import type { TenantContext } from "@maple/backend/services/auth/AuthService" +import { WarehouseQueryService } from "@maple/backend/services/warehouse/WarehouseQueryService" +import { makeWarehouseServiceStub } from "@maple/backend/testing/warehouse-test-support" +import { RecommendationIssueService } from "./RecommendationIssueService" + +const trackedDbs: TestDb[] = [] +afterEach(() => cleanupTestDbs(trackedDbs)) + +const ORG = Schema.decodeSync(OrgId)("org_recommendations") +const ISSUE = Schema.decodeSync(RecommendationIssueId)("3f9a1b2c-0d4e-4f5a-8b6c-7d8e9f0a1b2c") +const tenant: TenantContext = { + orgId: ORG, + userId: Schema.decodeSync(UserId)("user_recommendations"), + roles: ["org:admin"], + authMode: "self_hosted", +} + +// The key is no longer seen in span telemetry, so reconcile plans the open issue to `resolved`. +const noSpanKeys = makeWarehouseServiceStub({ compiledQuery: () => Effect.succeed([]) }) + +/** + * The real PGlite database, except that `onCall` runs just before the given `execute` call: + * a write that lands between reconcile's read and its updates. + */ +const racingDatabase = (testDb: TestDb, call: number, onCall: Effect.Effect) => + Layer.effect( + Database, + Effect.gen(function* () { + const inner = yield* Database + let calls = 0 + return { + execute: (fn) => + Effect.suspend(() => { + calls += 1 + return calls === call + ? onCall.pipe(Effect.andThen(inner.execute(fn))) + : inner.execute(fn) + }), + } satisfies typeof inner + }), + ).pipe(Layer.provide(testDb.layer)) + +const layerFor = (testDb: TestDb, database: Layer.Layer) => + Layer.effect(RecommendationIssueService, RecommendationIssueService.make).pipe( + Layer.provide(database), + Layer.provide(Layer.succeed(WarehouseQueryService, noSpanKeys)), + ) + +const seedOpenIssue = (testDb: TestDb) => + Effect.promise(() => + executeSql( + testDb, + `INSERT INTO org_recommendation_issues (id, org_id, number, recommendation_key, kind, source_key, canonical_key, status, usage_count) + VALUES ($1, $2, 1, 'rename:http.status_code', 'rename', 'http.status_code', 'http.response.status_code', 'open', 10)`, + [ISSUE, ORG], + ), + ) + +const issueState = (testDb: TestDb) => + Effect.promise(() => + queryFirstRow<{ status: string; resolved_at: Date | null }>( + testDb, + "SELECT status, resolved_at FROM org_recommendation_issues WHERE id = $1", + [ISSUE], + ), + ) + +describe("RecommendationIssueService reconcile", () => { + it.effect("a dismissal that lands after reconcile read the issue stays dismissed", () => { + const testDb = createTestDb(trackedDbs) + // Reconcile has read the issue and not yet written it: it must still be open here. + const dismissMidReconcile = Effect.gen(function* () { + assert.strictEqual((yield* issueState(testDb))?.status, "open") + yield* Effect.promise(() => + executeSql( + testDb, + "UPDATE org_recommendation_issues SET status = 'dismissed' WHERE id = $1", + [ISSUE], + ), + ) + injected = true + }) + let injected = false + // Calls: the mappings read, the issues read, then the planned status update. + const database = racingDatabase(testDb, 3, dismissMidReconcile) + return Effect.gen(function* () { + yield* seedOpenIssue(testDb) + const list = yield* (yield* RecommendationIssueService).listReconciled(tenant) + assert.isTrue(injected) + assert.strictEqual(list.issues[0]?.status, "dismissed") + assert.strictEqual((yield* issueState(testDb))?.status, "dismissed") + }).pipe(Effect.provide(layerFor(testDb, database))) + }) + + it.effect("dismiss closes the issue with a resolvedAt, like reconcile's resolve", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* seedOpenIssue(testDb) + const service = yield* RecommendationIssueService + yield* service.dismiss(tenant, ISSUE) + assert.isNotNull((yield* issueState(testDb))?.resolved_at) + + yield* service.reopen(tenant, ISSUE) + yield* service.listReconciled(tenant) + const resolved = yield* issueState(testDb) + assert.strictEqual(resolved?.status, "resolved") + assert.isNotNull(resolved?.resolved_at) + }).pipe(Effect.provide(layerFor(testDb, testDb.layer))) + }) +}) diff --git a/packages/backend/src/services/errors/RecommendationIssueService.ts b/packages/backend/src/services/errors/RecommendationIssueService.ts index 0aa5663da4..f72800eb31 100644 --- a/packages/backend/src/services/errors/RecommendationIssueService.ts +++ b/packages/backend/src/services/errors/RecommendationIssueService.ts @@ -8,7 +8,6 @@ import { RecommendationIssueNotFoundError, RecommendationIssuePersistenceError, RecommendationIssuesListResponse, - RecommendationIssueStatus, } from "@maple/domain/http" import { canonicalKeysToProbe, @@ -54,7 +53,6 @@ export interface RecommendationIssueServiceApi { const decodeIssueIdSync = Schema.decodeUnknownSync(RecommendationIssueId) const decodeKindSync = Schema.decodeUnknownSync(RecommendationIssueKind) -const decodeStatusSync = Schema.decodeUnknownSync(RecommendationIssueStatus) const decodeIsoSync = Schema.decodeUnknownSync(IsoDateTimeString) const toPersistenceError = (error: DatabaseError) => @@ -68,7 +66,7 @@ const rowToIssue = (row: IssueRow): RecommendationIssue => kind: decodeKindSync(row.kind), sourceKey: row.sourceKey, ...(row.canonicalKey != null ? { canonicalKey: row.canonicalKey } : undefined), - status: decodeStatusSync(row.status), + status: row.status, usageCount: row.usageCount, openedAt: decodeIsoSync(new Date(row.openedAt).toISOString()), updatedAt: decodeIsoSync(new Date(row.updatedAt).toISOString()), @@ -185,14 +183,15 @@ export class RecommendationIssueService extends Context.Service< number: row.number, recommendationKey: row.recommendationKey, sourceKey: row.sourceKey, - status: decodeStatusSync(row.status), + status: row.status, })) const plan = planReconcileIssues(detected, existingLike, mappingSourceKeys) + const readStatus = new Map(existing.map((row) => [row.id, row.status] as const)) const now = yield* Clock.currentTimeMillis if (plan.inserts.length > 0) { - const rows = plan.inserts.map((insert) => ({ + const rows = plan.inserts.map((insert): OrgRecommendationIssueInsert => ({ id: decodeIssueIdSync(randomUUID()), orgId, number: insert.number, @@ -206,7 +205,10 @@ export class RecommendationIssueService extends Context.Service< updatedAt: now, resolvedAt: null, })) - yield* dbExecute((db) => db.run(PG.insertInto(OrgRecommendationIssues).values(rows))) + // A concurrent list already opened the same key; its row stands. + yield* dbExecute((db) => + db.run(PG.insertInto(OrgRecommendationIssues).values(rows).onConflictDoNothing()), + ) } yield* Effect.forEach( @@ -222,11 +224,18 @@ export class RecommendationIssueService extends Context.Service< } : undefined), } + // A status move applies only from the status the plan read: a dismissal or reopen + // that landed since is the user's, and stands. + const from = update.nextStatus === undefined ? undefined : readStatus.get(update.id) return dbExecute((db) => db.run( PG.update(OrgRecommendationIssues) .set(fields) - .where(($) => [$.orgId.eq(orgId), $.id.eq(update.id)]), + .where(($) => [ + $.orgId.eq(orgId), + $.id.eq(update.id), + ...(from === undefined ? [] : [$.status.eq(from)]), + ]), ), ) }, @@ -272,8 +281,11 @@ export class RecommendationIssueService extends Context.Service< return yield* listResponse(orgId) }) + // Dismissal closes the issue like reconcile's applied/resolved do, so it stamps resolvedAt too. const dismiss = (tenant: TenantContext, id: RecommendationIssueId) => - setStatus(tenant, id, { status: "dismissed" }) + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => setStatus(tenant, id, { status: "dismissed", resolvedAt: now })), + ) const reopen = (tenant: TenantContext, id: RecommendationIssueId) => setStatus(tenant, id, { status: "open", resolvedAt: null }) diff --git a/packages/backend/src/services/integrations/vcs/VcsRepository.ts b/packages/backend/src/services/integrations/vcs/VcsRepository.ts index 55be41cba6..42273c1c4b 100644 --- a/packages/backend/src/services/integrations/vcs/VcsRepository.ts +++ b/packages/backend/src/services/integrations/vcs/VcsRepository.ts @@ -44,6 +44,7 @@ import { import { Array as Arr, Clock, Context, Effect, Layer, Option, Schema } from "effect" import { Database, type DatabaseError } from "@maple/backend/platform/DatabaseLive" import { makeDbExecute } from "@maple/backend/platform/db-execute" +import { transitionRow } from "@maple/backend/platform/transition-row" import { deleteSourcesOf, upsertWithSource, @@ -215,10 +216,20 @@ interface UpsertInstallationInput { interface RepoSyncStatusUpdate { readonly status: VcsRepoSyncStatus + /** The states this writer expects; a repo another sync moved since is left alone. */ + readonly from: Arr.NonEmptyReadonlyArray readonly error?: string | null readonly syncedAt?: number | null } +/** Every sync status: a fresh sync start, or a provider verdict that holds whatever the walk state. */ +export const ANY_REPO_SYNC_STATUS: Arr.NonEmptyReadonlyArray = [ + "pending", + "backfilling", + "ready", + "error", +] + /** * Read scope for repository queries — a required argument so every caller must * consciously decide whether provider-removed repos are in scope. `"active"` @@ -784,51 +795,54 @@ export class VcsRepository extends Context.Service()( return true }) - // Write a sync-status transition. `last_synced_at` is touched ONLY when the - // caller passes `syncedAt` (i.e. a sync actually completed) — so marking a repo - // "backfilling" at the start of / mid a run never stamps a misleading sync time; - // it stays whatever the last *successful* sync left. + // Write a sync-status transition, guarded on the states the caller expects and on the + // repo still being active. `last_synced_at` is touched ONLY when the caller passes + // `syncedAt` (a sync actually completed). Reports whether the write applied. const updateRepoSyncStatus = Effect.fn("VcsRepository.updateRepoSyncStatus")(function* ( repositoryId: VcsRepositoryId, update: RepoSyncStatusUpdate, ) { const now = yield* Clock.currentTimeMillis - yield* dbExecute((db) => - db.run( - PG.update(VcsRepositories) - .set({ - syncStatus: update.status, - lastSyncError: update.error ?? null, - ...("syncedAt" in update - ? { lastSyncedAt: update.syncedAt ?? null } - : undefined), - updatedAt: now, - }) - .where(($) => [$.id.eq(repositoryId)]), - ), + const applied = yield* transitionRow( + { table: "vcs_repositories", id: repositoryId, from: update.from, to: update.status }, + (from) => + dbExecute((db) => + db.run( + PG.update(VcsRepositories) + .set({ + syncStatus: update.status, + lastSyncError: update.error ?? null, + ...("syncedAt" in update + ? { lastSyncedAt: update.syncedAt ?? null } + : undefined), + updatedAt: now, + }) + .where(($) => [ + $.id.eq(repositoryId), + $.status.eq("active"), + $.syncStatus.in_(...from), + ]) + .returning("id"), + ), + ), + ).pipe( + Effect.as(true), + Effect.catchTag("@maple/backend/TransitionConflictError", () => Effect.succeed(false)), ) // Only a settled sync speaks for the installation; a start or a backfill step does not. - if (update.status === "ready" || update.status === "error") { + if (applied && (update.status === "ready" || update.status === "error")) { yield* reportInstallationSync(repositoryId, now) } + return applied }) // Flag a single repo's sync as errored without touching its cursor / // last-synced time (a failed fetch must not wipe prior progress). - const markRepoSyncError = Effect.fn("VcsRepository.markRepoSyncError")(function* ( + const markRepoSyncError = ( repositoryId: VcsRepositoryId, message: string, - ) { - const now = yield* Clock.currentTimeMillis - yield* dbExecute((db) => - db.run( - PG.update(VcsRepositories) - .set({ syncStatus: "error", lastSyncError: message, updatedAt: now }) - .where(($) => [$.id.eq(repositoryId)]), - ), - ) - yield* reportInstallationSync(repositoryId, now) - }) + from: Arr.NonEmptyReadonlyArray, + ) => updateRepoSyncStatus(repositoryId, { status: "error", from, error: message }) // Persist commits for an already-resolved repository. Commits belong to the // repo only (no commit↔branch link — a repo tracks one branch at a time). diff --git a/packages/backend/src/services/integrations/vcs/VcsSyncService.ts b/packages/backend/src/services/integrations/vcs/VcsSyncService.ts index 06be1d8399..630162a8fa 100644 --- a/packages/backend/src/services/integrations/vcs/VcsSyncService.ts +++ b/packages/backend/src/services/integrations/vcs/VcsSyncService.ts @@ -38,7 +38,7 @@ import { PullRequestEventSink, } from "./PullRequestEventSink" import { VcsProviderRegistry } from "./VcsProviderRegistry" -import { VcsRepository } from "./VcsRepository" +import { ANY_REPO_SYNC_STATUS, VcsRepository } from "./VcsRepository" import { VcsSyncQueue } from "./VcsSyncQueue" import { WorkflowEventSink } from "./WorkflowEventSink" import { summarizeCause } from "@maple/backend/platform/describe-cause" @@ -197,7 +197,9 @@ export class VcsSyncService extends Context.Service Effect.logWarning("[VCS] Repository unavailable — backfill skipped").pipe( Effect.annotateLogs({ @@ -1277,6 +1292,7 @@ export class VcsSyncService extends Context.Service { const a = yield* repoFor(repo, orgId, "7") const b = yield* repoFor(repo, orgId, "8") - yield* repo.updateRepoSyncStatus(a.id, { status: "ready", syncedAt: 1 }) + yield* repo.updateRepoSyncStatus(a.id, { + status: "ready", + from: ANY_REPO_SYNC_STATUS, + syncedAt: 1, + }) const ok = yield* sourceState assert.isNotNull(ok.last_poll_ok_at) assert.isNull(ok.last_poll_error) assert.deepStrictEqual(ok.poll_resources, { total: 2, unhealthy: 0 }) - yield* repo.markRepoSyncError(b.id, "upstream unavailable") + yield* repo.markRepoSyncError(b.id, "upstream unavailable", ANY_REPO_SYNC_STATUS) const failed = yield* sourceState assert.strictEqual(failed.last_poll_error, "octo/b: upstream unavailable") assert.deepStrictEqual(failed.poll_resources, { total: 2, unhealthy: 1 }) @@ -1751,6 +1755,44 @@ describe("VcsRepository", () => { }).pipe(Effect.provide(repoLayer(testDb))) }) + it.effect("an overlapping sync cannot regress a settled repo or write a removed one", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const repo = yield* VcsRepository + const orgId = asOrgId("org_sync_race") + yield* repo.upsertInstallation({ orgId, ...installationSeed("42", "100") }) + yield* upsertReposFor(repo, "42", [ + repoFixture({ externalRepoId: "7", fullName: "octo/a", name: "a" }), + repoFixture({ externalRepoId: "8", fullName: "octo/b", name: "b" }), + ]) + const a = yield* repoFor(repo, orgId, "7") + const b = yield* repoFor(repo, orgId, "8") + + // Walks A and B both start; B finishes first. + yield* repo.updateRepoSyncStatus(a.id, { status: "backfilling", from: ANY_REPO_SYNC_STATUS }) + yield* repo.updateRepoSyncStatus(a.id, { status: "ready", from: ["backfilling"], syncedAt: 5 }) + // A's continuation checkpoint and its exhausted-retries error both find the repo settled. + const checkpoint = yield* repo.updateRepoSyncStatus(a.id, { + status: "backfilling", + from: ["backfilling"], + }) + yield* repo.markRepoSyncError(a.id, "backfill failed", ["pending", "backfilling"]) + assert.isFalse(checkpoint) + const settled = yield* repoFor(repo, orgId, "7") + assert.strictEqual(settled.syncStatus, "ready") + assert.strictEqual(settled.lastSyncError, null) + + yield* repo.markRepositoryRemoved(b.id) + const written = yield* repo.updateRepoSyncStatus(b.id, { + status: "backfilling", + from: ANY_REPO_SYNC_STATUS, + }) + assert.isFalse(written) + const removed = Option.getOrThrow(yield* repo.getRepositoryById(orgId, b.id)) + assert.strictEqual(removed.syncStatus, "pending") + }).pipe(Effect.provide(repoLayer(testDb))) + }) + // A SHA persisted in any case is normalized to lowercase (the GitCommitSha decode // lowercases on the write path), so an uppercase push and a lowercase lookup never // split into two rows. diff --git a/packages/backend/src/services/org/OnboardingChecklistService.ts b/packages/backend/src/services/org/OnboardingChecklistService.ts index a20dbc26da..9759e2aafa 100644 --- a/packages/backend/src/services/org/OnboardingChecklistService.ts +++ b/packages/backend/src/services/org/OnboardingChecklistService.ts @@ -280,7 +280,7 @@ const make = Effect.gen(function* () { "reserveRewardClaim", onboarding.reserveRewardClaim(orgId, CLAIM_LEASE_MS), ) - if (!reserved) { + if (Option.isNone(reserved)) { return yield* new OnboardingRewardNotClaimableError({ message: "Another admin is claiming the reward right now. Try again in a moment.", reason: "in_progress", @@ -288,7 +288,7 @@ const make = Effect.gen(function* () { } const release = onboarding - .releaseRewardClaim(orgId) + .releaseRewardClaim(orgId, reserved.value) .pipe( Effect.catchCause((cause) => Effect.logError("Onboarding reward lease could not be released").pipe( @@ -350,7 +350,7 @@ const make = Effect.gen(function* () { ).pipe(Effect.annotateLogs({ orgId, error }), Effect.andThen(Effect.fail(error))), }), ) - yield* unavailable("finalizeRewardClaim", onboarding.finalizeRewardClaim(orgId)) + yield* unavailable("finalizeRewardClaim", onboarding.finalizeRewardClaim(orgId, reserved.value)) yield* edgeCache.invalidate({ bucket: CUSTOMER_CACHE_BUCKET, key: orgId }) yield* Effect.annotateCurrentSpan({ orgId, "maple.onboarding.reward_claimed": true }) diff --git a/packages/backend/src/services/org/OnboardingService.test.ts b/packages/backend/src/services/org/OnboardingService.test.ts new file mode 100644 index 0000000000..e0687a9511 --- /dev/null +++ b/packages/backend/src/services/org/OnboardingService.test.ts @@ -0,0 +1,91 @@ +import { afterEach, assert, describe, it } from "@effect/vitest" +import { Effect, Layer, Option, Schema } from "effect" +import { TestClock } from "effect/testing" +import { OrgId } from "@maple/domain/http" +import { + cleanupTestDbs, + createTestDb, + executeSql, + queryFirstRow, + type TestDb, +} from "@maple/backend/platform/test-pglite" +import { OnboardingService } from "@maple/backend/services/org/OnboardingService" + +const trackedDbs: TestDb[] = [] +afterEach(() => cleanupTestDbs(trackedDbs)) + +const ORG = Schema.decodeSync(OrgId)("org_reward_lease") +const LEASE_MS = 60_000 + +const layerFor = (testDb: TestDb) => OnboardingService.layer.pipe(Layer.provideMerge(testDb.layer)) + +const seedRow = (testDb: TestDb) => + Effect.promise(() => + executeSql( + testDb, + `INSERT INTO org_onboarding_state (org_id, demo_data_requested, created_at, updated_at) + VALUES ($1, false, now(), now())`, + [ORG], + ), + ) + +const rewardState = (testDb: TestDb) => + Effect.promise(() => + queryFirstRow<{ reward_reserved_at: Date | null; reward_claimed_at: Date | null }>( + testDb, + "SELECT reward_reserved_at, reward_claimed_at FROM org_onboarding_state WHERE org_id = $1", + [ORG], + ), + ) + +/** Holder A's lease lapses and holder B takes it over; returns both reservations. */ +const lapseAndTakeOver = Effect.gen(function* () { + const onboarding = yield* OnboardingService + const first = yield* onboarding.reserveRewardClaim(ORG, LEASE_MS) + yield* TestClock.adjust(`${LEASE_MS + 1} millis`) + const second = yield* onboarding.reserveRewardClaim(ORG, LEASE_MS) + return { first: Option.getOrThrow(first), second: Option.getOrThrow(second) } +}) + +describe("OnboardingService reward lease", () => { + it.effect("a holder whose lease lapsed cannot release the next holder's reservation", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* seedRow(testDb) + const { first, second } = yield* lapseAndTakeOver + yield* (yield* OnboardingService).releaseRewardClaim(ORG, first) + + const row = yield* rewardState(testDb) + assert.strictEqual(row?.reward_reserved_at?.getTime(), second) + // The live reservation still blocks a third claimant. + assert.isTrue(Option.isNone(yield* (yield* OnboardingService).reserveRewardClaim(ORG, LEASE_MS))) + }).pipe(Effect.provide(layerFor(testDb))) + }) + + it.effect("a holder whose lease lapsed records its claim but keeps the next holder's lease", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* seedRow(testDb) + const { first, second } = yield* lapseAndTakeOver + yield* (yield* OnboardingService).finalizeRewardClaim(ORG, first) + + const row = yield* rewardState(testDb) + assert.strictEqual(row?.reward_reserved_at?.getTime(), second) + assert.isNotNull(row?.reward_claimed_at) + }).pipe(Effect.provide(layerFor(testDb))) + }) + + it.effect("the current holder finalizes and releases its own lease", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* seedRow(testDb) + const onboarding = yield* OnboardingService + const held = Option.getOrThrow(yield* onboarding.reserveRewardClaim(ORG, LEASE_MS)) + yield* onboarding.finalizeRewardClaim(ORG, held) + + const row = yield* rewardState(testDb) + assert.strictEqual(row?.reward_reserved_at, null) + assert.strictEqual(row?.reward_claimed_at?.getTime(), held) + }).pipe(Effect.provide(layerFor(testDb))) + }) +}) diff --git a/packages/backend/src/services/org/OnboardingService.ts b/packages/backend/src/services/org/OnboardingService.ts index 21b9ab0aa1..0820b08d67 100644 --- a/packages/backend/src/services/org/OnboardingService.ts +++ b/packages/backend/src/services/org/OnboardingService.ts @@ -204,7 +204,8 @@ export class OnboardingService extends Context.Service()( /** * Take the reward claim lease: stamps `rewardReservedAt` if the reward is not * claimed and no live reservation exists (a reservation older than `leaseMs` - * is a crashed attempt and may be taken over). Reports whether this call won. + * is a crashed attempt and may be taken over). Returns the reservation this call + * holds, or `None` when another holder has it. */ const reserveRewardClaim = Effect.fn("OnboardingService.reserveRewardClaim")(function* ( orgId: OrgId, @@ -223,39 +224,55 @@ export class OnboardingService extends Context.Service()( .returning(($) => ({ id: $.orgId })), ), ) - return result.length > 0 + return result.length > 0 ? Option.some(now) : Option.none() }) - /** Billing confirmed the credit: the claim is final and the lease is released. */ + /** Applies `set` only while the org's lease is still the reservation the caller took. */ + const whileReserved = (orgId: OrgId, reservedAt: number, set: Partial) => + dbExecute((db) => + db.run( + PG.update(OrgOnboardingState) + .set(set) + .where(($) => [$.orgId.eq(orgId), $.rewardReservedAt.eq(reservedAt)]) + .returning(($) => ({ id: $.orgId })), + ), + ).pipe(Effect.map((rows) => rows.length > 0)) + + /** + * Billing confirmed the credit: the claim is final and the caller's lease is released. + * A lease that lapsed under the call still records the claim, but leaves the next + * holder's reservation alone. + */ const finalizeRewardClaim = Effect.fn("OnboardingService.finalizeRewardClaim")(function* ( orgId: OrgId, + reservedAt: number, ) { const now = yield* Clock.currentTimeMillis + const held = yield* whileReserved(orgId, reservedAt, { + rewardClaimedAt: now, + rewardReservedAt: null, + updatedAt: now, + }) + if (held) return yield* dbExecute((db) => db.run( PG.update(OrgOnboardingState) - .set({ - rewardClaimedAt: now, - rewardReservedAt: null, - updatedAt: now, - }) - .where(($) => [$.orgId.eq(orgId)]), + .set({ rewardClaimedAt: now, updatedAt: now }) + .where(($) => [$.orgId.eq(orgId), $.rewardClaimedAt.isNull()]), ), ) }) - /** Billing refused before anything was applied: give the lease back so the org can retry now. */ + /** + * Billing refused before anything was applied: give the lease back so the org can retry + * now. A lease another holder already took over is theirs, and stays. + */ const releaseRewardClaim = Effect.fn("OnboardingService.releaseRewardClaim")(function* ( orgId: OrgId, + reservedAt: number, ) { const now = yield* Clock.currentTimeMillis - yield* dbExecute((db) => - db.run( - PG.update(OrgOnboardingState) - .set({ rewardReservedAt: null, updatedAt: now }) - .where(($) => [$.orgId.eq(orgId)]), - ), - ) + yield* whileReserved(orgId, reservedAt, { rewardReservedAt: null, updatedAt: now }) }) const markEmailSent = Effect.fn("OnboardingService.markEmailSent")(function* ( diff --git a/packages/backend/src/services/org/OrgClickHouseSettingsService.ts b/packages/backend/src/services/org/OrgClickHouseSettingsService.ts index 503cea21ad..ded2f4e9e9 100644 --- a/packages/backend/src/services/org/OrgClickHouseSettingsService.ts +++ b/packages/backend/src/services/org/OrgClickHouseSettingsService.ts @@ -1,5 +1,6 @@ // SAFETY-FILE: JSON rows here come from fixed internal formats and are validated before domain use. // BOUNDARY: This module intentionally carries opaque values; callers decode them before domain use. +import { randomUUID } from "node:crypto" import * as Policy from "@maple/domain/policy" import { IsoDateTimeString, @@ -1301,15 +1302,18 @@ export class OrgClickHouseSettingsService extends Context.Service< // concurrent applySchema calls wins (interleaved workflow instances // would race destructive DROP/TRUNCATE/backfill migrations). A // queued/running row that stopped updating for STALE_APPLY_RUN_MS is - // reclaimable — the workflow stamps progress on every step, so a silent - // stall that long means the instance died without reaching its catch. + // reclaimable: the workflow stamps progress on every step and heartbeats + // long ones, so a silent stall that long means the instance died. const now = yield* Clock.currentTimeMillis + // The run id fences every progress write: an instance reclaimed as stale finds it + // replaced and stops instead of writing over (or migrating under) the new run. + const runId = yield* Effect.sync(() => randomUUID()) const claimed = yield* dbExecute((db) => db.run( PG.insertInto(OrgClickHouseSchemaApplyRuns) .values({ orgId, - workflowInstanceId: null, + workflowInstanceId: runId, status: "queued", phase: "queued", currentMigration: null, @@ -1331,6 +1335,7 @@ export class OrgClickHouseSettingsService extends Context.Service< $.updatedAt.lt(now - STALE_APPLY_RUN_MS), ), set: { + workflowInstanceId: runId, status: "queued", phase: "queued", currentMigration: null, @@ -1351,7 +1356,7 @@ export class OrgClickHouseSettingsService extends Context.Service< return new OrgClickHouseApplySchemaStarted({ status: "already_running" }) } - yield* workflow.create({ orgId }).pipe( + yield* workflow.create({ orgId, runId }).pipe( Effect.mapError( (error) => new OrgClickHouseSettingsPersistenceError({ @@ -1372,7 +1377,11 @@ export class OrgClickHouseSettingsService extends Context.Service< finishedAt: now, updatedAt: now, }) - .where(($) => [$.orgId.eq(orgId), $.status.eq("queued")]), + .where(($) => [ + $.orgId.eq(orgId), + $.workflowInstanceId.eq(runId), + $.status.eq("queued"), + ]), ), ) .pipe(Effect.ignore), diff --git a/packages/backend/src/services/pr-review/PrReviewPostMergeService.test.ts b/packages/backend/src/services/pr-review/PrReviewPostMergeService.test.ts index 9b413dee28..bec1dbe958 100644 --- a/packages/backend/src/services/pr-review/PrReviewPostMergeService.test.ts +++ b/packages/backend/src/services/pr-review/PrReviewPostMergeService.test.ts @@ -306,3 +306,32 @@ describe("PrReviewPostMergeService.runTick, lease", () => { }).pipe(Effect.provide(layerFor(testDb, posted, { failOperations: false }, racing))) }) }) + +describe("PrReviewPostMergeService.runTick, settle", () => { + it.effect("does not overwrite a report a later tick made while this one read the deploys", () => { + const testDb = createTestDb(trackedDbs) + const posted: Array = [] + // A later tick, holding the row's next lease, reports while this one waits on the warehouse. + const reportElsewhere = Effect.gen(function* () { + const db = yield* Database + yield* db.execute((client) => + client.run( + PG.update(PrReviews) + .set({ postMergeStatus: "reported", postMergeAfter: null }) + .where(($) => [$.id.eq(reviewId)]), + ), + ) + }).pipe(Effect.provide(testDb.layer), Effect.orDie) + const noDeployYet: PrReviewTelemetryServiceApi = { + ...telemetryFake({ failOperations: false }), + deploymentsSince: () => reportElsewhere.pipe(Effect.as([])), + } + return Effect.gen(function* () { + yield* seedDueReview + yield* TestClock.setTime(DEPLOYED_AT + 70 * 60_000) + const result = yield* (yield* PrReviewPostMergeService).runTick + assert.equal(result.waiting, 0) + assert.deepStrictEqual(yield* rowState, { status: "reported", after: null }) + }).pipe(Effect.provide(layerFor(testDb, posted, { failOperations: false }, noDeployYet))) + }) +}) diff --git a/packages/backend/src/services/pr-review/PrReviewPostMergeService.ts b/packages/backend/src/services/pr-review/PrReviewPostMergeService.ts index 7b0a5eeca4..57b6f46fa6 100644 --- a/packages/backend/src/services/pr-review/PrReviewPostMergeService.ts +++ b/packages/backend/src/services/pr-review/PrReviewPostMergeService.ts @@ -21,6 +21,7 @@ import { PrReviews, type PrReviewRow } from "@maple/db/tables" import { Context, DateTime, Duration, Effect, Layer, Option, Schema } from "effect" import { Database } from "@maple/backend/platform/DatabaseLive" import { summarizeCause } from "@maple/backend/platform/describe-cause" +import { transitionRow } from "@maple/backend/platform/transition-row" import { VcsProviderRegistry } from "@maple/backend/services/integrations/vcs/VcsProviderRegistry" import { VcsRepository } from "@maple/backend/services/integrations/vcs/VcsRepository" import { PrReviewTelemetryService } from "./telemetry/PrReviewTelemetryService" @@ -78,23 +79,41 @@ export class PrReviewPostMergeService extends Context.Service< const repositories = yield* VcsRepository const providers = yield* VcsProviderRegistry + /** + * Applies only while the row is still `waiting` under the lease this examination took: a + * look that outlived it lost the row to a later tick, which may already have reported. + */ const settle = ( orgId: OrgId, id: PrReviewId, + leaseUntil: DateTime.Utc, status: PrReviewPostMergeStatus, postMergeAfter: number | null, now: DateTime.Utc, ) => - database.execute((db) => - db.run( - PG.update(PrReviews) - .set({ postMergeStatus: status, updatedAt: toMs(now), postMergeAfter }) - .where(($) => [$.orgId.eq(orgId), $.id.eq(id)]), + transitionRow({ table: "pr_reviews", id, from: ["waiting"], to: status }, (from) => + database.execute((db) => + db.run( + PG.update(PrReviews) + .set({ postMergeStatus: status, updatedAt: toMs(now), postMergeAfter }) + .where(($) => [ + $.orgId.eq(orgId), + $.id.eq(id), + $.postMergeStatus.in_(...from), + $.postMergeAfter.eq(toMs(leaseUntil)), + ]) + .returning("id"), + ), ), ) - const later = (orgId: OrgId, id: PrReviewId, at: DateTime.Utc, now: DateTime.Utc) => - settle(orgId, id, "waiting", toMs(at), now) + const later = ( + orgId: OrgId, + id: PrReviewId, + leaseUntil: DateTime.Utc, + at: DateTime.Utc, + now: DateTime.Utc, + ) => settle(orgId, id, leaseUntil, "waiting", toMs(at), now) const examine = Effect.fn("PrReviewPostMergeService.examine")(function* ( row: DueRow, @@ -112,7 +131,7 @@ export class PrReviewPostMergeService extends Context.Service< const breaksOnly = facts !== undefined && services.length === 0 && openContractBreaks(facts).length > 0 if (facts === undefined || (services.length === 0 && !breaksOnly)) { - yield* settle(orgId, row.id, "no_traffic", null, now) + yield* settle(orgId, row.id, leaseUntil, "no_traffic", null, now) return "gave_up" satisfies Outcome } const versions = yield* telemetry.deploymentsSince(orgId, services, mergedAt, now) @@ -131,10 +150,10 @@ export class PrReviewPostMergeService extends Context.Service< }) if (picked === undefined) { if (elapsed(mergedAt, now, POST_MERGE_GIVE_UP)) { - yield* settle(orgId, row.id, "no_deploy", null, now) + yield* settle(orgId, row.id, leaseUntil, "no_deploy", null, now) return "gave_up" satisfies Outcome } - yield* later(orgId, row.id, DateTime.addDuration(now, POST_MERGE_RETRY), now) + yield* later(orgId, row.id, leaseUntil, DateTime.addDuration(now, POST_MERGE_RETRY), now) return "waiting" satisfies Outcome } const deployAt = picked.deploy.firstSeen @@ -143,7 +162,7 @@ export class PrReviewPostMergeService extends Context.Service< // A full window of traffic after the deploy, and its rollups landed. const readable = DateTime.addDuration(windowEnd, ROLLUP_LAG) if (DateTime.isLessThan(now, readable)) { - yield* later(orgId, row.id, readable, now) + yield* later(orgId, row.id, leaseUntil, readable, now) return "waiting" satisfies Outcome } @@ -233,6 +252,7 @@ export class PrReviewPostMergeService extends Context.Service< updatedAt: toMs(now), }) .where(($) => [ + $.orgId.eq(orgId), $.id.eq(row.id), $.postMergeStatus.eq("waiting"), $.postMergeAfter.eq(toMs(leaseUntil)), @@ -289,9 +309,9 @@ export class PrReviewPostMergeService extends Context.Service< ) const outcomes = yield* Effect.forEach( due, - (row) => - Effect.gen(function* () { - const leaseUntil = DateTime.addDuration(now, CLAIM) + (row) => { + const leaseUntil = DateTime.addDuration(now, CLAIM) + return Effect.gen(function* () { // Claimed first: an overlapping tick that read the same row finds it leased and // skips it, so the follow-up is posted once. const claimed = yield* database.execute((db) => @@ -307,7 +327,12 @@ export class PrReviewPostMergeService extends Context.Service< ), ) if (claimed.length === 0) return "skipped" as const - return yield* examine(row, now, leaseUntil) + return yield* examine(row, now, leaseUntil).pipe( + // A later tick took the row over; it settles or reports it instead. + Effect.catchTag("@maple/backend/TransitionConflictError", () => + Effect.succeed("skipped" as const), + ), + ) }).pipe( Effect.map((outcome): Outcome | "failed" | "skipped" => outcome), Effect.catchCause((cause) => { @@ -324,10 +349,11 @@ export class PrReviewPostMergeService extends Context.Service< }), Effect.andThen( (giveUp - ? settle(row.orgId, row.id, "failed", null, now) + ? settle(row.orgId, row.id, leaseUntil, "failed", null, now) : later( row.orgId, row.id, + leaseUntil, DateTime.addDuration(now, POST_MERGE_RETRY), now, ) @@ -336,7 +362,8 @@ export class PrReviewPostMergeService extends Context.Service< Effect.as("failed" as const), ) }), - ), + ) + }, { concurrency: 2 }, ) const count = (outcome: Outcome | "failed" | "skipped") => diff --git a/packages/db/src/tables/org-recommendation-issues.ts b/packages/db/src/tables/org-recommendation-issues.ts index 479a043493..2fd0660a48 100644 --- a/packages/db/src/tables/org-recommendation-issues.ts +++ b/packages/db/src/tables/org-recommendation-issues.ts @@ -1,4 +1,5 @@ import * as PG from "@maple-dev/effect-orm/postgres" +import { RecommendationIssueStatus } from "@maple/domain/http" import { OrgId } from "@maple/domain/primitives" // Durable, numbered attribute-recommendation issues (PlanetScale-style). Recommendations are @@ -21,7 +22,7 @@ export const OrgRecommendationIssues = PG.table("org_recommendation_issues", { kind: PG.text, sourceKey: PG.column(PG.text, { name: "source_key" }), canonicalKey: PG.column(PG.nullable(PG.text), { name: "canonical_key" }), - status: PG.column(PG.text, { default: "open" }), + status: PG.column(PG.brand(PG.text, RecommendationIssueStatus), { default: "open" }), usageCount: PG.column(PG.int4, { name: "usage_count", default: 0 }), openedAt: PG.column(PG.timestamptzMillis, { name: "opened_at", defaultExpr: "now()" }), updatedAt: PG.column(PG.timestamptzMillis, { name: "updated_at", defaultExpr: "now()" }),