feat: 增加发布单版本冲突保护

- deploy-run: 将 version 纳入发布单摘要并在 run/step 更新时递增 - repository: 使用 expectedVersion 做乐观锁更新，冲突返回 CONFLICT - execution: Jenkins 同步、通知、取消路径按最新版本串行推进
2026-06-12 05:10:59 +08:00
parent f92ee66298
commit f883568250
5 changed files with 276 additions and 95 deletions
@@ -38,7 +38,11 @@ export class DeployExecutionService {

    let currentRun = this.requireRun(
      run.id,
-      await this.deployRunRepository.updateRun(run.id, { status: 'running' }),
+      await this.deployRunRepository.updateRun(
+        run.id,
+        { status: 'running' },
+        run.version,
+      ),
    );
    let currentStep: string | undefined;

@@ -58,7 +62,7 @@ export class DeployExecutionService {
    try {
      currentStep = 'gitea-check';
      currentRun = await this.completeStep(
-        currentRun.id,
+        currentRun,
        currentStep,
        'success',
      );
@@ -74,7 +78,7 @@ export class DeployExecutionService {

      currentStep = 'health-check';
      currentRun = await this.completeStep(
-        currentRun.id,
+        currentRun,
        currentStep,
        'success',
      );
@@ -84,15 +88,19 @@ export class DeployExecutionService {

      currentStep = 'finish-release';
      currentRun = await this.completeStep(
-        currentRun.id,
+        currentRun,
        currentStep,
        'success',
      );
      currentRun = this.requireRun(
        currentRun.id,
-        await this.deployRunRepository.updateRun(currentRun.id, {
+        await this.deployRunRepository.updateRun(
+          currentRun.id,
+          {
            status: 'success',
-        }),
+          },
+          currentRun.version,
+        ),
      );

      await this.auditService.record({
@@ -121,14 +129,19 @@ export class DeployExecutionService {
              message: undefined,
              errorSummary: failureSummary,
            },
+            currentRun.version,
          )) ?? currentRun;
      }

      currentRun =
-        (await this.deployRunRepository.updateRun(currentRun.id, {
+        (await this.deployRunRepository.updateRun(
+          currentRun.id,
+          {
            status: 'failed',
            failureSummary,
-        })) ?? currentRun;
+          },
+          currentRun.version,
+        )) ?? currentRun;

      currentRun = await this.advanceNotification(currentRun, 'failed');

@@ -172,26 +185,34 @@ export class DeployExecutionService {
      if (!queueItem.buildNumber) {
        return this.requireRun(
          currentRun.id,
-          await this.deployRunRepository.updateRun(currentRun.id, {
+          await this.deployRunRepository.updateRun(
+            currentRun.id,
+            {
              status: 'queued',
              jenkinsQueueId: queueItem.queueId,
              jenkinsBuildUrl: queueItem.executableUrl,
-          }),
+            },
+            currentRun.version,
+          ),
        );
      }

      buildNumber = queueItem.buildNumber;
      currentRun = this.requireRun(
        currentRun.id,
-        await this.deployRunRepository.updateRun(currentRun.id, {
+        await this.deployRunRepository.updateRun(
+          currentRun.id,
+          {
            status: 'running',
            jenkinsQueueId: queueItem.queueId,
            jenkinsBuildNumber: queueItem.buildNumber,
            jenkinsBuildUrl: queueItem.executableUrl,
-        }),
+          },
+          currentRun.version,
+        ),
      );
      currentRun = await this.markStep(
-        currentRun.id,
+        currentRun,
        'jenkins-build',
        'running',
        `Jenkins 构建 #${queueItem.buildNumber} 正在运行。`,
@@ -221,19 +242,23 @@ export class DeployExecutionService {

    currentRun = this.requireRun(
      currentRun.id,
-      await this.deployRunRepository.updateRun(currentRun.id, {
+      await this.deployRunRepository.updateRun(
+        currentRun.id,
+        {
          status:
            buildStatus.building || !buildStatus.result
              ? 'running'
              : currentRun.status,
          jenkinsBuildNumber: buildStatus.buildNumber,
          jenkinsBuildUrl: buildStatus.url,
-      }),
+        },
+        currentRun.version,
+      ),
    );

    if (buildStatus.building || !buildStatus.result) {
      return this.markStep(
-        currentRun.id,
+        currentRun,
        'jenkins-build',
        'running',
        `Jenkins 构建 #${buildStatus.buildNumber} 正在运行。`,
@@ -263,7 +288,7 @@ export class DeployExecutionService {
        health.message ??
        'Jenkins 尚未配置，已完成本地模拟构建。';
      const currentRun = await this.completeStep(
-        run.id,
+        run,
        'jenkins-build',
        'success',
      );
@@ -292,23 +317,27 @@ export class DeployExecutionService {
      );
    }

-    await this.markStep(run.id, 'jenkins-build', 'running');
+    let currentRun = await this.markStep(run, 'jenkins-build', 'running');
    const queueItem = await this.jenkinsClient.buildWithParameters(
      environment.jenkinsJobPath,
      this.buildParameters(run),
    );
    const nextStatus = queueItem.buildNumber ? 'running' : 'queued';
-    let currentRun = this.requireRun(
-      run.id,
-      await this.deployRunRepository.updateRun(run.id, {
+    currentRun = this.requireRun(
+      currentRun.id,
+      await this.deployRunRepository.updateRun(
+        currentRun.id,
+        {
          status: nextStatus,
          jenkinsQueueId: queueItem.queueId,
          jenkinsBuildNumber: queueItem.buildNumber,
          jenkinsBuildUrl: queueItem.executableUrl,
-      }),
+        },
+        currentRun.version,
+      ),
    );
    currentRun = await this.markStep(
-      run.id,
+      currentRun,
      'jenkins-build',
      nextStatus,
      queueItem.buildNumber
@@ -336,17 +365,20 @@ export class DeployExecutionService {
    run: DeployRunSummary,
    event: DeployNotificationEvent,
  ): Promise<DeployRunSummary> {
-    await this.markStep(run.id, 'notify-wecom', 'running');
+    let currentRun = await this.markStep(run, 'notify-wecom', 'running');
    const notification =
-      await this.deployNotificationService.sendDeployRunEvent(run, event);
+      await this.deployNotificationService.sendDeployRunEvent(
+        currentRun,
+        event,
+      );
    const stepStatus =
      notification.status === 'sent'
        ? 'success'
        : notification.status === 'failed'
          ? 'failed'
          : 'skipped';
-    const currentRun = await this.completeStep(
-      run.id,
+    currentRun = await this.completeStep(
+      currentRun,
      'notify-wecom',
      stepStatus,
      notification.status === 'sent' ? undefined : notification.reason,
@@ -372,26 +404,30 @@ export class DeployExecutionService {
    run: DeployRunSummary,
  ): Promise<DeployRunSummary> {
    let currentRun = await this.completeStep(
-      run.id,
+      run,
      'jenkins-build',
      'success',
    );
    currentRun = await this.completeStep(
-      currentRun.id,
+      currentRun,
      'health-check',
      'success',
    );
    currentRun = await this.advanceNotification(currentRun, 'succeeded');
    currentRun = await this.completeStep(
-      currentRun.id,
+      currentRun,
      'finish-release',
      'success',
    );
    currentRun = this.requireRun(
      currentRun.id,
-      await this.deployRunRepository.updateRun(currentRun.id, {
+      await this.deployRunRepository.updateRun(
+        currentRun.id,
+        {
          status: 'success',
-      }),
+        },
+        currentRun.version,
+      ),
    );

    await this.auditService.record({
@@ -417,17 +453,26 @@ export class DeployExecutionService {
    const logExcerpt = await this.readFailureLogExcerpt(run, jobPath);
    const failureSummary = `Jenkins 构建 #${run.jenkinsBuildNumber ?? '-'} ${result}`;
    let currentRun =
-      (await this.deployRunRepository.updateStep(run.id, 'jenkins-build', {
+      (await this.deployRunRepository.updateStep(
+        run.id,
+        'jenkins-build',
+        {
          status: 'failed',
          message: undefined,
          errorSummary: failureSummary,
          logExcerpt,
-      })) ?? run;
+        },
+        run.version,
+      )) ?? run;
    currentRun =
-      (await this.deployRunRepository.updateRun(currentRun.id, {
+      (await this.deployRunRepository.updateRun(
+        currentRun.id,
+        {
          status: 'failed',
          failureSummary,
-      })) ?? currentRun;
+        },
+        currentRun.version,
+      )) ?? currentRun;
    currentRun = await this.advanceNotification(currentRun, 'failed');

    await this.auditService.record({
@@ -502,33 +547,43 @@ export class DeployExecutionService {
  }

  private async completeStep(
-    runId: string,
+    run: DeployRunSummary,
    bpmnNodeId: string,
    status: Extract<DeployRunStepStatus, 'success' | 'skipped' | 'failed'>,
    errorSummary?: string,
  ): Promise<DeployRunSummary> {
    return this.requireRun(
-      runId,
-      await this.deployRunRepository.updateStep(runId, bpmnNodeId, {
+      run.id,
+      await this.deployRunRepository.updateStep(
+        run.id,
+        bpmnNodeId,
+        {
          status,
          message: undefined,
          errorSummary,
-      }),
+        },
+        run.version,
+      ),
    );
  }

  private async markStep(
-    runId: string,
+    run: DeployRunSummary,
    bpmnNodeId: string,
    status: Extract<DeployRunStepStatus, 'queued' | 'running'>,
    message?: string,
  ): Promise<DeployRunSummary> {
    return this.requireRun(
-      runId,
-      await this.deployRunRepository.updateStep(runId, bpmnNodeId, {
+      run.id,
+      await this.deployRunRepository.updateStep(
+        run.id,
+        bpmnNodeId,
+        {
          status,
          message,
-      }),
+        },
+        run.version,
+      ),
    );
  }

@@ -11,6 +11,7 @@ import {
  Prisma,
  RunStepStatus as PrismaRunStepStatus,
 } from '@prisma/client';
+import { AppError } from '../common/errors/app-error';
 import { EnvConfig } from '../config/env.schema';
 import { PrismaService } from '../prisma/prisma.service';
 import { seedDeployRuns } from './deploy-run.seed';
@@ -217,9 +218,10 @@ export class DeployRunRepository {
  async updateRun(
    id: string,
    patch: DeployRunMemoryPatch,
+    expectedVersion?: number,
  ): Promise<DeployRunSummary | null> {
    if (!this.useDatabase()) {
-      return this.updateMemoryRun(id, patch);
+      return this.updateMemoryRun(id, patch, expectedVersion);
    }

    const current = await this.prisma.deployRun.findUnique({
@@ -279,11 +281,16 @@ export class DeployRunRepository {
      data.finishedAt = this.toDate(patch.finishedAt);
    }

-    await this.prisma.deployRun.update({
-      where: { id },
+    const version = expectedVersion ?? current.version;
+    const updated = await this.prisma.deployRun.updateMany({
+      where: { id, version },
      data,
    });

+    if (updated.count === 0) {
+      await this.raiseRunVersionConflict(id, version);
+    }
+
    return this.findById(id);
  }

@@ -291,9 +298,15 @@ export class DeployRunRepository {
    runId: string,
    bpmnNodeId: string,
    patch: DeployRunStepMemoryPatch,
+    expectedRunVersion?: number,
  ): Promise<DeployRunSummary | null> {
    if (!this.useDatabase()) {
-      return this.updateMemoryStep(runId, bpmnNodeId, patch);
+      return this.updateMemoryStep(
+        runId,
+        bpmnNodeId,
+        patch,
+        expectedRunVersion,
+      );
    }

    const current = await this.prisma.runStep.findUnique({
@@ -303,6 +316,9 @@ export class DeployRunRepository {
          bpmnNodeId,
        },
      },
+      include: {
+        deployRun: true,
+      },
    });

    if (!current) {
@@ -355,7 +371,30 @@ export class DeployRunRepository {
      data.metadata = metadata;
    }

-    await this.prisma.runStep.update({
+    const runVersion = expectedRunVersion ?? current.deployRun.version;
+
+    await this.prisma.$transaction(async (tx) => {
+      const updatedRun = await tx.deployRun.updateMany({
+        where: {
+          id: runId,
+          version: runVersion,
+        },
+        data: {
+          version: {
+            increment: 1,
+          },
+        },
+      });
+
+      if (updatedRun.count === 0) {
+        throw this.createRunVersionConflict(runId, runVersion, {
+          version: current.deployRun.version,
+          status: current.deployRun.status,
+          updatedAt: current.deployRun.updatedAt,
+        });
+      }
+
+      await tx.runStep.update({
        where: {
          deployRunId_bpmnNodeId: {
            deployRunId: runId,
@@ -364,10 +403,57 @@ export class DeployRunRepository {
        },
        data,
      });
+    });

    return this.findById(runId);
  }

+  private async raiseRunVersionConflict(
+    id: string,
+    expectedVersion: number,
+  ): Promise<never> {
+    throw await this.buildRunVersionConflict(id, expectedVersion);
+  }
+
+  private async buildRunVersionConflict(
+    id: string,
+    expectedVersion: number,
+  ): Promise<AppError> {
+    const latest = await this.prisma.deployRun.findUnique({
+      where: { id },
+      select: {
+        status: true,
+        version: true,
+        updatedAt: true,
+      },
+    });
+
+    return this.createRunVersionConflict(id, expectedVersion, latest);
+  }
+
+  private createRunVersionConflict(
+    id: string,
+    expectedVersion: number,
+    latest?: {
+      status?: PrismaDeployRunStatus;
+      version?: number;
+      updatedAt?: Date;
+    } | null,
+  ): AppError {
+    return new AppError(
+      'CONFLICT',
+      `发布记录 ${id} 已被其他流程更新，请刷新后重试`,
+      409,
+      {
+        id,
+        expectedVersion,
+        latestVersion: latest?.version,
+        latestStatus: latest?.status?.toLowerCase(),
+        latestUpdatedAt: latest?.updatedAt?.toISOString(),
+      },
+    );
+  }
+
  private async findRecordById(id: string): Promise<DeployRunRecord | null> {
    return this.prisma.deployRun.findUnique({
      where: { id },
@@ -440,6 +526,7 @@ export class DeployRunRepository {

    return {
      id: idempotencyKey,
+      version: 1,
      projectKey: input.projectKey,
      environment: input.environment,
      ref: input.ref,
@@ -497,6 +584,7 @@ export class DeployRunRepository {
  private updateMemoryRun(
    id: string,
    patch: DeployRunMemoryPatch,
+    expectedVersion?: number,
  ): DeployRunSummary | null {
    const run = this.memoryRuns.find((item) => item.id === id);

@@ -504,6 +592,7 @@ export class DeployRunRepository {
      return null;
    }

+    this.assertMemoryVersion(run, expectedVersion);
    const now = new Date().toISOString();

    if (patch.status) {
@@ -542,6 +631,7 @@ export class DeployRunRepository {
      run.finishedAt = patch.finishedAt;
    }

+    run.version += 1;
    return run;
  }

@@ -549,6 +639,7 @@ export class DeployRunRepository {
    runId: string,
    bpmnNodeId: string,
    patch: DeployRunStepMemoryPatch,
+    expectedRunVersion?: number,
  ): DeployRunSummary | null {
    const run = this.memoryRuns.find((item) => item.id === runId);
    const step = run?.steps.find((item) => item.bpmnNodeId === bpmnNodeId);
@@ -557,6 +648,7 @@ export class DeployRunRepository {
      return null;
    }

+    this.assertMemoryVersion(run, expectedRunVersion);
    const now = new Date().toISOString();

    if (patch.status) {
@@ -591,15 +683,38 @@ export class DeployRunRepository {
      step.logExcerpt = patch.logExcerpt;
    }

+    run.version += 1;
    return run;
  }

+  private assertMemoryVersion(
+    run: DeployRunSummary,
+    expectedVersion?: number,
+  ): void {
+    if (expectedVersion === undefined || run.version === expectedVersion) {
+      return;
+    }
+
+    throw new AppError(
+      'CONFLICT',
+      `发布记录 ${run.id} 已被其他流程更新，请刷新后重试`,
+      409,
+      {
+        id: run.id,
+        expectedVersion,
+        latestVersion: run.version,
+        latestStatus: run.status,
+      },
+    );
+  }
+
  private toSummary(run: DeployRunRecord): DeployRunSummary {
    const metadata = this.objectMetadata<DeployRunMetadata>(run.metadata);
    const sortedSteps = [...run.steps].sort((left, right) => left.order - right.order);

    return {
      id: run.id,
+      version: run.version,
      projectKey: run.project.key,
      environment: run.environment.name === 'TEST' ? 'test' : 'production',
      ref: run.ref,
@@ -5,6 +5,7 @@ const now = new Date().toISOString();
 export const seedDeployRuns: DeployRunSummary[] = [
  {
    id: 'run_seed_access_manage_test_001',
+    version: 1,
    projectKey: 'access-manage',
    environment: 'test',
    ref: 'develop',
@@ -23,6 +23,7 @@ export type DeployRunStepSummary = {

 export type DeployRunSummary = {
  id: string;
+  version: number;
  projectKey: string;
  environment: DeployEnvironment;
  ref: string;
@@ -165,17 +165,26 @@ export class DeployRunsService {
    const jenkinsCancel = await this.cancelJenkinsExecution(source, project);
    let run = this.requireRun(
      id,
-      await this.deployRunRepository.updateRun(id, {
+      await this.deployRunRepository.updateRun(
+        id,
+        {
          status: 'canceled',
-      }),
+        },
+        source.version,
+      ),
    );
    run = this.requireRun(
      id,
-      await this.deployRunRepository.updateStep(id, 'jenkins-build', {
+      await this.deployRunRepository.updateStep(
+        id,
+        'jenkins-build',
+        {
          status: 'canceled',
          message: undefined,
          errorSummary: this.cancelSummary(jenkinsCancel),
-      }),
+        },
+        run.version,
+      ),
    );

    await this.auditService.record({