fix: persist plan results before publishing success

server/events/command_runner_test.go

@@ -68,6 +68,7 @@ var postWorkflowHooksCommandRunner events.PostWorkflowHooksCommandRunner
 var cancellationTracker *mocks.MockCancellationTracker
 
 type TestConfig struct {
+	planRunnerWrapper          func(events.ProjectCommandRunner) events.ProjectPlanCommandRunner
 	parallelPoolSize           int
 	SilenceNoProjects          bool
 	silenceVCSStatusNoPlans    bool
@@ -177,6 +178,10 @@ func setup(t *testing.T, options ...func(testConfig *TestConfig)) *vcsmocks.Mock
 	if workingDirLocker == nil {
 		workingDirLocker = events.NewDefaultWorkingDirLocker()
 	}
+	var planRunner events.ProjectPlanCommandRunner = projectCommandRunner
+	if testConfig.planRunnerWrapper != nil {
+		planRunner = testConfig.planRunnerWrapper(projectCommandRunner)
+	}
 	planCommandRunner = events.NewPlanCommandRunner(
 		testConfig.silenceVCSStatusNoPlans,
 		testConfig.silenceVCSStatusNoProjects,
@@ -186,7 +191,7 @@ func setup(t *testing.T, options ...func(testConfig *TestConfig)) *vcsmocks.Mock
 		workingDirLocker,
 		commitUpdater,
 		projectCommandBuilder,
-		projectCommandRunner,
+		planRunner,
 		cancellationTracker,
 		dbUpdater,
 		pullUpdater,

server/events/instrumented_project_command_runner.go

@@ -103,3 +103,9 @@ func RunAndEmitStats(ctx command.ProjectContext, execute func(ctx command.Projec
 	return result
 
 }
+
+func (p *InstrumentedProjectCommandRunner) PublishDeferredPlanStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus) {
+	if publisher, ok := p.projectCommandRunner.(DeferredPlanStatusPublisher); ok {
+		publisher.PublishDeferredPlanStatuses(projectCmds, result, status)
+	}
+}

server/events/plan_command_runner.go

@@ -187,13 +187,14 @@ func (p *PlanCommandRunner) runAutoplan(ctx *command.Context) {
 		result.PlansDeleted = true
 	}
 
-	p.pullUpdater.updatePull(ctx, AutoplanCommand{}, result)
-
 	pullStatus, err := p.dbUpdater.updateDB(ctx, ctx.Pull, result.ProjectResults)
 	if err != nil {
-		ctx.Log.Err("writing results: %s", err)
+		p.planPersistenceFailed(ctx, AutoplanCommand{}, projectCmds, result, err)
+		return
 	}
 
+	p.publishPlanStatuses(projectCmds, result, models.SuccessCommitStatus)
+	p.pullUpdater.updatePull(ctx, AutoplanCommand{}, result)
 	p.updateCommitStatus(ctx, pullStatus, command.Plan)
 	p.updateCommitStatus(ctx, pullStatus, command.Apply)
 
@@ -334,11 +335,6 @@ func (p *PlanCommandRunner) run(ctx *command.Context, cmd *CommentCommand) {
 		result.PlansDeleted = true
 	}
 
-	p.pullUpdater.updatePull(
-		ctx,
-		cmd,
-		result)
-
 	var pullStatus models.PullStatus
 	if noProjectPullStatus != nil {
 		pullStatus = *noProjectPullStatus
@@ -348,10 +344,12 @@ func (p *PlanCommandRunner) run(ctx *command.Context, cmd *CommentCommand) {
 		pullStatus, err = p.dbUpdater.updateDB(ctx, pull, result.ProjectResults)
 	}
 	if err != nil {
-		ctx.Log.Err("writing results: %s", err)
+		p.planPersistenceFailed(ctx, cmd, projectCmds, result, err)
 		return
 	}
 
+	p.publishPlanStatuses(projectCmds, result, models.SuccessCommitStatus)
+	p.pullUpdater.updatePull(ctx, cmd, result)
 	p.updateCommitStatus(ctx, pullStatus, command.Plan)
 	p.updateCommitStatus(ctx, pullStatus, command.Apply)
 
@@ -604,3 +602,22 @@ func (p *PlanCommandRunner) partitionProjectCmds(
 func (p *PlanCommandRunner) isParallelEnabled(projectCmds []command.ProjectContext) bool {
 	return len(projectCmds) > 0 && projectCmds[0].ParallelPlanEnabled
 }
+
+// Successful project checks are deferred until the command's results are durable.
+func (p *PlanCommandRunner) publishPlanStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus) {
+	if publisher, ok := p.prjCmdRunner.(DeferredPlanStatusPublisher); ok {
+		publisher.PublishDeferredPlanStatuses(projectCmds, result, status)
+	}
+}
+
+func (p *PlanCommandRunner) planPersistenceFailed(ctx *command.Context, cmd PullCommand, projectCmds []command.ProjectContext, result command.Result, err error) {
+	ctx.CommandHasErrors = true
+	result.Error = fmt.Errorf("persisting plan results: %w; restore database connectivity and run `atlantis plan` again before applying", err)
+	p.publishPlanStatuses(projectCmds, result, models.FailedCommitStatus)
+	for _, name := range []command.Name{command.Plan, command.Apply} {
+		if statusErr := p.commitStatusUpdater.UpdateCombined(ctx.Log, ctx.Pull.BaseRepo, ctx.Pull, models.FailedCommitStatus, name); statusErr != nil {
+			ctx.Log.Warn("unable to update commit status: %s", statusErr)
+		}
+	}
+	p.pullUpdater.updatePull(ctx, cmd, result)
+}

server/events/plan_command_runner_test.go

@@ -5,8 +5,12 @@ package events_test
 
 import (
 	"errors"
+	"strings"
 	"testing"
 
+	"github.com/runatlantis/atlantis/server/core/db"
+	"github.com/runatlantis/atlantis/server/events/mocks"
+
 	"github.com/google/go-github/v88/github"
 	. "github.com/petergtz/pegomock/v4"
 	"github.com/runatlantis/atlantis/server/core/boltdb"
@@ -1147,3 +1151,88 @@ func TestPlanCommandRunner_PendingApplyStatus(t *testing.T) {
 		})
 	}
 }
+
+// A callback at the durable write boundary makes ordering deterministic without
+// racing a VCS poll against the plan command.
+type observingPlanDatabase struct {
+	db.Database
+	beforeWrite func()
+	writeErr    error
+	persisted   bool
+}
+
+func (d *observingPlanDatabase) UpdatePullWithResults(pull models.PullRequest, results []command.ProjectResult) (models.PullStatus, error) {
+	d.beforeWrite()
+	if d.writeErr != nil {
+		return models.PullStatus{}, d.writeErr
+	}
+	status, err := d.Database.UpdatePullWithResults(pull, results)
+	d.persisted = err == nil
+	return status, err
+}
+
+func TestPlanCommandRunner_PersistenceBeforePublication(t *testing.T) {
+	for _, auto := range []bool{false, true} {
+		for _, fail := range []bool{false, true} {
+			name := "manual"
+			if auto {
+				name = "autoplan"
+			}
+			if fail {
+				name += "_write_failure"
+			}
+			t.Run(name, func(t *testing.T) {
+				RegisterMockTestingT(t)
+				storage, err := boltdb.New(t.TempDir())
+				Ok(t, err)
+				t.Cleanup(func() { storage.Close() })
+				database := &observingPlanDatabase{Database: storage}
+				if fail {
+					database.writeErr = errors.New("disk full")
+				}
+				setter := mocks.NewMockJobURLSetter()
+				messages := mocks.NewMockJobMessageSender()
+				vcsClient := setup(t, func(tc *TestConfig) {
+					tc.database = database
+					tc.planRunnerWrapper = func(r events.ProjectCommandRunner) events.ProjectPlanCommandRunner {
+						return events.NewInstrumentedProjectCommandRunner(metricstest.NewLoggingScope(t, logging.NewNoopLogger(t), "atlantis"), &events.ProjectOutputWrapper{ProjectCommandRunner: r, JobURLSetter: setter, JobMessageSender: messages})
+					}
+				})
+				ctx := &command.Context{Log: logging.NewNoopLogger(t), Pull: testdata.Pull, HeadRepo: testdata.GithubRepo, Scope: metricstest.NewLoggingScope(t, logging.NewNoopLogger(t), "atlantis")}
+				if auto {
+					ctx.Trigger = command.AutoTrigger
+				} else {
+					ctx.Trigger = command.CommentTrigger
+				}
+				cmd := &events.CommentCommand{Name: command.Plan, RepoRelDir: "project"}
+				project := command.ProjectContext{CommandName: command.Plan, RepoRelDir: "project", Workspace: "default", Log: ctx.Log}
+				When(projectCommandBuilder.BuildPlanCommands(ctx, cmd)).ThenReturn([]command.ProjectContext{project}, nil)
+				When(projectCommandBuilder.BuildAutoplanCommands(ctx)).ThenReturn([]command.ProjectContext{project}, nil)
+				When(projectCommandRunner.Plan(Any[command.ProjectContext]())).ThenReturn(command.ProjectCommandOutput{PlanSuccess: &models.PlanSuccess{}})
+				database.beforeWrite = func() {
+					setter.VerifyWasCalled(Never()).SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]())
+					vcsClient.VerifyWasCalled(Never()).CreateComment(Any[logging.SimpleLogging](), Any[models.Repo](), Any[int](), Any[string](), Any[string]())
+					commitUpdater.VerifyWasCalled(Never()).UpdateCombinedCount(Any[logging.SimpleLogging](), Any[models.Repo](), Any[models.PullRequest](), Eq(models.SuccessCommitStatus), Eq(command.Plan), Any[models.ProjectCounts]())
+				}
+				When(setter.SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]())).Then(func([]Param) ReturnValues {
+					Assert(t, database.persisted, "project success must follow persistence")
+					return ReturnValues{nil}
+				})
+				planCommandRunner.Run(ctx, cmd)
+				Equals(t, !fail, database.persisted)
+				if fail {
+					Assert(t, ctx.CommandHasErrors, "persistence failure must fail the command")
+					setter.VerifyWasCalledOnce().SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.FailedCommitStatus), Any[*command.ProjectCommandOutput]())
+					setter.VerifyWasCalled(Never()).SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]())
+					for _, name := range []command.Name{command.Plan, command.Apply} {
+						commitUpdater.VerifyWasCalledOnce().UpdateCombined(Any[logging.SimpleLogging](), Any[models.Repo](), Any[models.PullRequest](), Eq(models.FailedCommitStatus), Eq(name))
+					}
+					_, _, _, comment, _ := vcsClient.VerifyWasCalledOnce().CreateComment(Any[logging.SimpleLogging](), Any[models.Repo](), Any[int](), Any[string](), Any[string]()).GetCapturedArguments()
+					Assert(t, strings.Contains(comment, "disk full") && strings.Contains(comment, "atlantis plan"), "expected actionable persistence error: %s", comment)
+				} else {
+					setter.VerifyWasCalledOnce().SetJobURLWithStatus(Any[command.ProjectContext](), Eq(command.Plan), Eq(models.SuccessCommitStatus), Any[*command.ProjectCommandOutput]())
+				}
+			})
+		}
+	}
+}

server/events/project_command_runner.go

@@ -162,6 +162,10 @@ type JobURLSetter interface {
 	SetJobURLWithStatus(ctx command.ProjectContext, cmdName command.Name, status models.CommitStatus, res *command.ProjectCommandOutput) error
 }
 
+type DeferredPlanStatusPublisher interface {
+	PublishDeferredPlanStatuses([]command.ProjectContext, command.Result, models.CommitStatus)
+}
+
 type DeferredApplyStatusPublisher interface {
 	PublishDeferredApplyStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus)
 }
@@ -221,7 +225,7 @@ func (p *ProjectOutputWrapper) updateProjectPRStatus(commandName command.Name, c
 		return result
 	}
 
-	if commandName == command.Apply {
+	if commandName == command.Apply || (commandName == command.Plan && !ctx.API) {
 		return result
 	}
 
@@ -1223,3 +1227,24 @@ func getMissingPolicySetNames(policySets []valid.PolicySet, receivedCount int) [
 func requiresManagedPlanFileForApply(ctx command.ProjectContext) bool {
 	return ctx.RequiresAtlantisManagedPlanFile || hasAtlantisManagedApplyStep(ctx.Steps)
 }
+
+func (p *ProjectOutputWrapper) PublishDeferredPlanStatuses(projectCmds []command.ProjectContext, result command.Result, status models.CommitStatus) {
+	for _, res := range result.ProjectResults {
+		if res.Command != command.Plan || res.PlanSuccess == nil || res.Error != nil || res.Failure != "" {
+			continue
+		}
+		for _, ctx := range projectCmds {
+			if ctx.CommandName != command.Plan || ctx.RepoRelDir != res.RepoRelDir || ctx.Workspace != res.Workspace || ctx.ProjectName != res.ProjectName || ctx.SuppressVCSStatus {
+				continue
+			}
+			output := res.ProjectCommandOutput
+			if result.Error != nil {
+				output = command.ProjectCommandOutput{Error: result.Error}
+			}
+			if err := p.JobURLSetter.SetJobURLWithStatus(ctx, command.Plan, status, &output); err != nil {
+				ctx.Log.Err("updating project PR status: %s", err)
+			}
+			break
+		}
+	}
+}

server/events/project_command_runner_test.go

@@ -356,7 +356,7 @@ func TestProjectOutputWrapper(t *testing.T) {
 			}
 
 			mockJobURLSetter.VerifyWasCalled(Once()).SetJobURLWithStatus(ctx, c.CommandName, models.PendingCommitStatus, nil)
-			if c.CommandName == command.Apply && c.Success {
+			if c.Success {
 				mockJobURLSetter.VerifyWasCalled(Never()).SetJobURLWithStatus(ctx, c.CommandName, models.SuccessCommitStatus, &prjResult)
 			} else {
 				mockJobURLSetter.VerifyWasCalled(Once()).SetJobURLWithStatus(ctx, c.CommandName, expCommitStatus, &prjResult)