chore: fix race condition on aggregating terraform logs (#21067)

I noticed we have a defer to make sure all log output is captured by the
async log handling routine.

```golang
	defer func() {
		_ = outWriter.Close()
		_ = errWriter.Close()
		<-doneOut
		<-doneErr
	}()
```

But `e.timings.aggregate()` compiles the **current** parsed logs. I was
getting some race conditions in testing if the logs did not completely
parse before `aggregate` was called.

So now `aggregate` happens outside the cmd exec function, to make sure
it is completed before logs are accumulated.

provisioner/terraform/executor.go

@@ -325,9 +325,7 @@ func (e *executor) plan(ctx, killCtx context.Context, env, vars []string, logr l
 		<-doneErr
 	}()
 
-	endStage := e.timings.startStage(database.ProvisionerJobTimingStagePlan)
 	err := e.execWriteOutput(ctx, killCtx, args, env, outWriter, errWriter)
-	endStage(err)
 	if err != nil {
 		return nil, xerrors.Errorf("terraform plan: %w", err)
 	}
@@ -390,7 +388,7 @@ func (e *executor) plan(ctx, killCtx context.Context, env, vars []string, logr l
 		Parameters:            state.Parameters,
 		Resources:             state.Resources,
 		ExternalAuthProviders: state.ExternalAuthProviders,
-		Timings:               append(e.timings.aggregate(), graphTimings.aggregate()...),
+		Timings:               graphTimings.aggregate(),
 		Presets:               state.Presets,
 		Plan:                  planJSON,
 		ResourceReplacements:  resReps,
@@ -599,9 +597,7 @@ func (e *executor) apply(
 	}()
 
 	// `terraform apply`
-	endStage := e.timings.startStage(database.ProvisionerJobTimingStageApply)
 	err := e.execWriteOutput(ctx, killCtx, args, env, outWriter, errWriter)
-	endStage(err)
 	if err != nil {
 		return nil, xerrors.Errorf("terraform apply: %w", err)
 	}
@@ -617,13 +613,11 @@ func (e *executor) apply(
 		return nil, xerrors.Errorf("read statefile %q: %w", statefilePath, err)
 	}
 
-	agg := e.timings.aggregate()
 	return &proto.ApplyComplete{
 		Parameters:            state.Parameters,
 		Resources:             state.Resources,
 		ExternalAuthProviders: state.ExternalAuthProviders,
 		State:                 stateContent,
-		Timings:               agg,
 		AiTasks:               state.AITasks,
 	}, nil
 }

provisioner/terraform/provision.go

@@ -160,14 +160,16 @@ func (s *server) Plan(
 		return provisionersdk.PlanErrorf("plan vars: %s", err)
 	}
 
+	endPlanStage := e.timings.startStage(database.ProvisionerJobTimingStagePlan)
 	resp, err := e.plan(ctx, killCtx, env, vars, sess, request)
+	endPlanStage(err)
 	if err != nil {
 		return provisionersdk.PlanErrorf("%s", err.Error())
 	}
 
 	// Prepend init timings since they occur prior to plan timings.
 	// Order is irrelevant; this is merely indicative.
-	resp.Timings = append(initTimings.aggregate(), resp.Timings...) // mergeInitTimings(initTimings.aggregate(), resp.Timings)
+	resp.Timings = append(resp.Timings, append(initTimings.aggregate(), e.timings.aggregate()...)...)
 	resp.Modules = modules
 	return resp
 }
@@ -204,9 +206,11 @@ func (s *server) Apply(
 		return provisionersdk.ApplyErrorf("provision env: %s", err)
 	}
 	env = otelEnvInject(ctx, env)
+	endStage := e.timings.startStage(database.ProvisionerJobTimingStageApply)
 	resp, err := e.apply(
 		ctx, killCtx, env, sess,
 	)
+	endStage(err)
 	if err != nil {
 		errorMessage := err.Error()
 		// Terraform can fail and apply and still need to store it's state.
@@ -217,6 +221,7 @@ func (s *server) Apply(
 			Error: errorMessage,
 		}
 	}
+	resp.Timings = e.timings.aggregate()
 	return resp
 }