From e2b13406dbe5578468dc3d3dbed8b6d861f3aa0b Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 19 Sep 2026 05:00:01 +0000 Subject: [PATCH] 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 --- internal/projectinstance/service.go | 5 +- internal/projectinstance/stage_service.go | 2 +- internal/projectinstance/stage_store.go | 8 +- internal/projectinstance/store.go | 99 ++++++++++++++++++++++- 4 files changed, 102 insertions(+), 12 deletions(-) diff --git a/internal/projectinstance/service.go b/internal/projectinstance/service.go index 168bd3bf..2433ca6a 100644 --- a/internal/projectinstance/service.go +++ b/internal/projectinstance/service.go @@ -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) diff --git a/internal/projectinstance/stage_service.go b/internal/projectinstance/stage_service.go index de580e21..82b6317b 100644 --- a/internal/projectinstance/stage_service.go +++ b/internal/projectinstance/stage_service.go @@ -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 } diff --git a/internal/projectinstance/stage_store.go b/internal/projectinstance/stage_store.go index d82143f5..70b7dbad 100644 --- a/internal/projectinstance/stage_store.go +++ b/internal/projectinstance/stage_store.go @@ -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 } diff --git a/internal/projectinstance/store.go b/internal/projectinstance/store.go index e3555219..b43ba424 100644 --- a/internal/projectinstance/store.go +++ b/internal/projectinstance/store.go @@ -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 ©Inst, 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