fix(projectinstance): atomic plan save and correct completed_by attribution

- Add Store.SavePlanWithRevision: single-transaction planning upsert + revision bump (bump first with optimistic lock, conflict rolls back before any plan write), closing the SaveWorkflowPlan split-transaction race where plan content and revision could diverge - AdvanceStage now takes operatorID and writes completed_by explicitly instead of SELECT activated_by, so the completing operator is the actual advancing user, not the original activator - Service.SaveWorkflowPlan/AdvanceStage pass userID through; memStore implements SavePlanWithRevision under its single mutex Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-19 05:00:01 +00:00
parent c7eab1f7d1
commit e2b13406db
4 changed files with 102 additions and 12 deletions
@@ -182,10 +182,7 @@ func (s *Service) SaveWorkflowPlan(ctx context.Context, projectID int64, userID
 		ConfigSnapshot:    in.Plan,
 		CreatedBy:         &userID,
 	}
-	if _, err := s.store.UpsertPlanningInstance(ctx, inst); err != nil {
-		return nil, err
-	}
-	if _, err := s.store.BumpPlanRevision(ctx, projectID, planRev); err != nil {
+	if _, _, err := s.store.SavePlanWithRevision(ctx, inst, planRev); err != nil {
 		return nil, err
 	}
 	return s.GetSettings(ctx, projectID, userID, isCreator)
@@ -94,7 +94,7 @@ func (s *Service) AdvanceStage(ctx context.Context, instanceID int64, userID int
 		}
 	}

-	after, err := s.store.AdvanceStage(ctx, instanceID, in.StageKey, in.ExpectedConfigRevision)
+	after, err := s.store.AdvanceStage(ctx, instanceID, in.StageKey, in.ExpectedConfigRevision, userID)
 	if err != nil {
 		return nil, err
 	}
@@ -69,7 +69,7 @@ func (s *pgStore) loadInstanceDetail(ctx context.Context, inst *WorkflowInstance
 	return detail, rows.Err()
 }

-func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error) {
+func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int, operatorID int64) (*InstanceDetail, error) {
 	tx, err := s.pool.Begin(ctx)
 	if err != nil {
 		return nil, err
@@ -131,11 +131,9 @@ func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey s
 	if errors.Is(err, pgx.ErrNoRows) {
 		_, err = tx.Exec(ctx, `
 			UPDATE project_workflow_instances SET status = 'completed', config_revision = config_revision + 1,
-				completed_at = $2, completed_by = (
-					SELECT activated_by FROM project_workflow_instances WHERE id = $1
-				), updated_at = $2
+				completed_at = $2, completed_by = $3, updated_at = $2
 			WHERE id = $1
-		`, instanceID, now)
+		`, instanceID, now, operatorID)
 		if err != nil {
 			return nil, err
 		}
@@ -44,6 +44,9 @@ type Store interface {
 	GetGoverningInstanceForAccess(ctx context.Context, projectID int64) (*WorkflowInstance, error)
 	HasLiveInstance(ctx context.Context, projectID int64) (bool, error)
 	UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error)
+	// SavePlanWithRevision 单事务完成 planning upsert + revision bump，
+	// 避免 SaveWorkflowPlan 两条独立事务间的版本号与 plan 内容错位。
+	SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error)
 	ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error)
 	ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error)
 	CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error)
@@ -53,7 +56,7 @@ type Store interface {
 	ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error)

 	GetInstanceByID(ctx context.Context, instanceID int64) (*InstanceDetail, error)
-	AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error)
+	AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int, operatorID int64) (*InstanceDetail, error)
 	MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error
 }

@@ -260,6 +263,85 @@ func (s *pgStore) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInst
 	return inst, nil
 }

+// SavePlanWithRevision 在单事务内完成 planning 实例 upsert 与 plan revision bump。
+// bump 先执行（带期望版本乐观锁），失败则整个事务回滚，保证 plan 内容与版本号原子一致。
+func (s *pgStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) {
+	snapRaw, _ := json.Marshal(inst.ConfigSnapshot)
+	now := time.Now()
+	if inst.CreatedAt.IsZero() {
+		inst.CreatedAt = now
+	}
+	inst.UpdatedAt = now
+
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer tx.Rollback(ctx)
+
+	// 先 bump revision（乐观锁），冲突即回滚，不写 plan
+	var newRev int
+	err = tx.QueryRow(ctx, `
+		UPDATE projects SET workflow_plan_revision = workflow_plan_revision + 1, updated_at = now()
+		WHERE id = $1 AND workflow_plan_revision = $2
+		RETURNING workflow_plan_revision
+	`, inst.ProjectID, expectedPlanRevision).Scan(&newRev)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, 0, ErrPlanRevisionConflict
+	}
+	if err != nil {
+		return nil, 0, err
+	}
+
+	var existingID int64
+	err = tx.QueryRow(ctx, `
+		SELECT id FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning' LIMIT 1
+	`, inst.ProjectID).Scan(&existingID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		var nextNo int
+		if err := tx.QueryRow(ctx, `
+			SELECT COALESCE(MAX(instance_no), 0) + 1
+			FROM project_workflow_instances
+			WHERE project_id = $1
+		`, inst.ProjectID).Scan(&nextNo); err != nil {
+			return nil, 0, err
+		}
+		inst.InstanceNo = nextNo
+		err = tx.QueryRow(ctx, `
+			INSERT INTO project_workflow_instances (
+				project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
+				config_snapshot, created_by, created_at, updated_at
+			) VALUES ($1, $2, $3, 'planning', $4, $5, $6, $7, $8, $9)
+			RETURNING id
+		`, inst.ProjectID, inst.WorkflowVersionID, nextNo, inst.ConfigRevision, inst.PlanHash, snapRaw,
+			inst.CreatedBy, inst.CreatedAt, inst.UpdatedAt).Scan(&inst.ID)
+		if err != nil {
+			return nil, 0, err
+		}
+	} else if err != nil {
+		return nil, 0, err
+	} else {
+		err = tx.QueryRow(ctx, `
+			UPDATE project_workflow_instances SET
+				workflow_version_id = $2, config_revision = $3, plan_hash = $4, config_snapshot = $5,
+				updated_at = $6
+			WHERE id = $1 AND status = 'planning'
+			RETURNING id
+		`, existingID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw, inst.UpdatedAt).Scan(&inst.ID)
+		if err != nil {
+			return nil, 0, err
+		}
+	}
+
+	if err := s.replaceInstanceChildren(ctx, tx, inst); err != nil {
+		return nil, 0, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, 0, err
+	}
+	return inst, newRev, nil
+}
+
 func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst *WorkflowInstance) error {
 	_, err := tx.Exec(ctx, `
 		DELETE FROM document_duty_assignments WHERE doc_task_instance_id IN (
@@ -859,6 +941,19 @@ func (m *memStore) UpsertPlanningInstance(_ context.Context, inst *WorkflowInsta
 	return &copyInst, nil
 }

+// SavePlanWithRevision 内存实现：bump 与 upsert 在同一互斥锁内，天然原子。
+func (m *memStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) {
+	newRev, err := m.BumpPlanRevision(ctx, inst.ProjectID, expectedPlanRevision)
+	if err != nil {
+		return nil, 0, err
+	}
+	saved, err := m.UpsertPlanningInstance(ctx, inst)
+	if err != nil {
+		return nil, 0, err
+	}
+	return saved, newRev, nil
+}
+
 func (m *memStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) {
 	usesAuthorOnly := planUsesAuthorOnly(inst.ConfigSnapshot)
 	var currentPolicyRev int64
@@ -937,7 +1032,7 @@ func (m *memStore) GetInstanceByID(_ context.Context, instanceID int64) (*Instan
 	return nil, ErrInstanceNotFound
 }

-func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error) {
+func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey string, expectedRevision int, _ int64) (*InstanceDetail, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	var inst *WorkflowInstance