From ca94588bd57fda9655e47d1a73a0501b831cf56f Mon Sep 17 00:00:00 2001 From: Susana Ferreira Date: Wed, 12 Nov 2025 10:36:39 +0000 Subject: [PATCH] fix: send prebuild job notification after job build db commit (#20693) ## Problem Fix race condition in prebuilds reconciler. Previously, a job notification event was sent to a Go channel before the provisioning database transaction completed. The notification is consumed by a separate goroutine that publishes to PostgreSQL's LISTEN/NOTIFY, using a separate database connection. This creates a potential race: if a provisioner daemon receives the notification and queries for the job before the provisioning transaction commits, it won't find the job in the database. This manifested as a flaky test failure in `TestReinitializeAgent`, where provisioners would occasionally miss newly created jobs. The test uses a 25-second timeout context, while the acquirer's backup polling mechanism checks for jobs every 30 seconds. This made the race condition visible in tests, though in production the backup polling would eventually pick up the job. The solution presented here guarantees that a job notification is only sent after the provisioning database transaction commits. ## Changes * The `provision()` and `provisionDelete()` functions now return the provisioner job instead of sending notifications internally. * A new `publishProvisionerJob()` helper centralizes the notification logic and is called after each transaction completes. Closes: https://github.com/coder/internal/issues/963 --- enterprise/coderd/prebuilds/reconcile.go | 97 +++++++++++++++++------ enterprise/coderd/workspaceagents_test.go | 2 +- 2 files changed, 72 insertions(+), 27 deletions(-) diff --git a/enterprise/coderd/prebuilds/reconcile.go b/enterprise/coderd/prebuilds/reconcile.go index f280436ea9..17a56d484c 100644 --- a/enterprise/coderd/prebuilds/reconcile.go +++ b/enterprise/coderd/prebuilds/reconcile.go @@ -697,7 +697,8 @@ func (c *StoreReconciler) createPrebuiltWorkspace(ctx context.Context, prebuiltW return xerrors.Errorf("failed to generate unique prebuild ID: %w", err) } - return c.store.InTx(func(db database.Store) error { + var provisionerJob *database.ProvisionerJob + err = c.store.InTx(func(db database.Store) error { template, err := db.GetTemplateByID(ctx, templateID) if err != nil { return xerrors.Errorf("failed to get template: %w", err) @@ -732,11 +733,20 @@ func (c *StoreReconciler) createPrebuiltWorkspace(ctx context.Context, prebuiltW c.logger.Info(ctx, "attempting to create prebuild", slog.F("name", name), slog.F("workspace_id", prebuiltWorkspaceID.String()), slog.F("preset_id", presetID.String())) - return c.provision(ctx, db, prebuiltWorkspaceID, template, presetID, database.WorkspaceTransitionStart, workspace, DeprovisionModeNormal) + provisionerJob, err = c.provision(ctx, db, prebuiltWorkspaceID, template, presetID, database.WorkspaceTransitionStart, workspace, DeprovisionModeNormal) + return err }, &database.TxOptions{ Isolation: sql.LevelRepeatableRead, ReadOnly: false, }) + if err != nil { + return err + } + + // Publish provisioner job event to notify the acquirer that a new job was posted + c.publishProvisionerJob(ctx, provisionerJob, prebuiltWorkspaceID) + + return nil } // provisionDelete provisions a delete transition for a prebuilt workspace. @@ -748,26 +758,25 @@ func (c *StoreReconciler) createPrebuiltWorkspace(ctx context.Context, prebuiltW // // IMPORTANT: This function must be called within a database transaction. It does not create its own transaction. // The caller is responsible for managing the transaction boundary via db.InTx(). -func (c *StoreReconciler) provisionDelete(ctx context.Context, db database.Store, workspaceID uuid.UUID, templateID uuid.UUID, presetID uuid.UUID, mode DeprovisionMode) error { +func (c *StoreReconciler) provisionDelete(ctx context.Context, db database.Store, workspaceID uuid.UUID, templateID uuid.UUID, presetID uuid.UUID, mode DeprovisionMode) (*database.ProvisionerJob, error) { workspace, err := db.GetWorkspaceByID(ctx, workspaceID) if err != nil { - return xerrors.Errorf("get workspace by ID: %w", err) + return nil, xerrors.Errorf("get workspace by ID: %w", err) } template, err := db.GetTemplateByID(ctx, templateID) if err != nil { - return xerrors.Errorf("failed to get template: %w", err) + return nil, xerrors.Errorf("failed to get template: %w", err) } if workspace.OwnerID != database.PrebuildsSystemUserID { - return xerrors.Errorf("prebuilt workspace is not owned by prebuild user anymore, probably it was claimed") + return nil, xerrors.Errorf("prebuilt workspace is not owned by prebuild user anymore, probably it was claimed") } c.logger.Info(ctx, "attempting to delete prebuild", slog.F("orphan", mode.String()), slog.F("name", workspace.Name), slog.F("workspace_id", workspaceID.String()), slog.F("preset_id", presetID.String())) - return c.provision(ctx, db, workspaceID, template, presetID, - database.WorkspaceTransitionDelete, workspace, mode) + return c.provision(ctx, db, workspaceID, template, presetID, database.WorkspaceTransitionDelete, workspace, mode) } // cancelAndOrphanDeletePendingPrebuilds cancels pending prebuild jobs from inactive template versions @@ -779,7 +788,9 @@ func (c *StoreReconciler) provisionDelete(ctx context.Context, db database.Store // Since these jobs were never processed by a provisioner, no Terraform resources were created, // making it safe to orphan-delete the workspaces (skipping Terraform destroy). func (c *StoreReconciler) cancelAndOrphanDeletePendingPrebuilds(ctx context.Context, templateID uuid.UUID, templateVersionID uuid.UUID, presetID uuid.UUID) error { - return c.store.InTx(func(db database.Store) error { + var canceledProvisionerJob *database.ProvisionerJob + var canceledWorkspaceID uuid.UUID + err := c.store.InTx(func(db database.Store) error { canceledJobs, err := db.UpdatePrebuildProvisionerJobWithCancel( ctx, database.UpdatePrebuildProvisionerJobWithCancelParams{ @@ -808,11 +819,14 @@ func (c *StoreReconciler) cancelAndOrphanDeletePendingPrebuilds(ctx context.Cont var multiErr multierror.Error for _, job := range canceledJobs { - err = c.provisionDelete(ctx, db, job.WorkspaceID, job.TemplateID, presetID, DeprovisionModeOrphan) + provisionerJob, err := c.provisionDelete(ctx, db, job.WorkspaceID, job.TemplateID, presetID, DeprovisionModeOrphan) if err != nil { c.logger.Error(ctx, "failed to orphan delete canceled prebuild", slog.F("workspace_id", job.WorkspaceID.String()), slog.Error(err)) multiErr.Errors = append(multiErr.Errors, err) + } else if canceledProvisionerJob == nil { + canceledProvisionerJob = provisionerJob + canceledWorkspaceID = job.WorkspaceID } } @@ -821,15 +835,38 @@ func (c *StoreReconciler) cancelAndOrphanDeletePendingPrebuilds(ctx context.Cont Isolation: sql.LevelRepeatableRead, ReadOnly: false, }) + if err != nil { + return err + } + + // Job event notifications contain organization, provisioner type, and tags. + // Since all canceled jobs have the same values, we only send one notification + // for the first successfully canceled job, which is sufficient to trigger the + // provisioner chain that processes all remaining jobs. + if canceledProvisionerJob != nil { + c.publishProvisionerJob(ctx, canceledProvisionerJob, canceledWorkspaceID) + } + + return nil } func (c *StoreReconciler) deletePrebuiltWorkspace(ctx context.Context, prebuiltWorkspaceID uuid.UUID, templateID uuid.UUID, presetID uuid.UUID) error { - return c.store.InTx(func(db database.Store) error { - return c.provisionDelete(ctx, db, prebuiltWorkspaceID, templateID, presetID, DeprovisionModeNormal) + var provisionerJob *database.ProvisionerJob + err := c.store.InTx(func(db database.Store) (err error) { + provisionerJob, err = c.provisionDelete(ctx, db, prebuiltWorkspaceID, templateID, presetID, DeprovisionModeNormal) + return err }, &database.TxOptions{ Isolation: sql.LevelRepeatableRead, ReadOnly: false, }) + if err != nil { + return err + } + + // Publish provisioner job event to notify the acquirer that a new job was posted + c.publishProvisionerJob(ctx, provisionerJob, prebuiltWorkspaceID) + + return nil } func (c *StoreReconciler) provision( @@ -841,10 +878,10 @@ func (c *StoreReconciler) provision( transition database.WorkspaceTransition, workspace database.Workspace, mode DeprovisionMode, -) error { +) (*database.ProvisionerJob, error) { tvp, err := db.GetPresetParametersByTemplateVersionID(ctx, template.ActiveVersionID) if err != nil { - return xerrors.Errorf("fetch preset details: %w", err) + return nil, xerrors.Errorf("fetch preset details: %w", err) } var params []codersdk.WorkspaceBuildParameter @@ -893,26 +930,34 @@ func (c *StoreReconciler) provision( audit.WorkspaceBuildBaggage{}, ) if err != nil { - return xerrors.Errorf("provision workspace: %w", err) + return nil, xerrors.Errorf("provision workspace: %w", err) } - if provisionerJob == nil { - return nil - } - - // Publish provisioner job event outside of transaction. - select { - case c.provisionNotifyCh <- *provisionerJob: - default: // channel full, drop the message; provisioner will pick this job up later with its periodic check, though. - c.logger.Warn(ctx, "provisioner job notification queue full, dropping", - slog.F("job_id", provisionerJob.ID), slog.F("prebuild_id", prebuildID.String())) + // This should not happen, builder.Build() should either return a job or an error. + // Returning an error to fail fast if we hit this unexpected case. + return nil, xerrors.Errorf("provision succeeded but returned no job") } c.logger.Info(ctx, "prebuild job scheduled", slog.F("transition", transition), slog.F("prebuild_id", prebuildID.String()), slog.F("preset_id", presetID.String()), slog.F("job_id", provisionerJob.ID)) - return nil + return provisionerJob, nil +} + +// publishProvisionerJob publishes a provisioner job event to notify the acquirer that a new job has been created. +// This must be called after the database transaction that creates the job has committed to ensure +// the job is visible to provisioners when they query the database. +func (c *StoreReconciler) publishProvisionerJob(ctx context.Context, provisionerJob *database.ProvisionerJob, workspaceID uuid.UUID) { + if provisionerJob == nil { + return + } + select { + case c.provisionNotifyCh <- *provisionerJob: + default: // channel full, drop the message; provisioner will pick this job up later with its periodic check + c.logger.Warn(ctx, "provisioner job notification queue full, dropping", + slog.F("job_id", provisionerJob.ID), slog.F("prebuild_id", workspaceID.String())) + } } // ForceMetricsUpdate forces the metrics collector, if defined, to update its state (we cache the metrics state to diff --git a/enterprise/coderd/workspaceagents_test.go b/enterprise/coderd/workspaceagents_test.go index 2e4690bc96..a150c0cdc0 100644 --- a/enterprise/coderd/workspaceagents_test.go +++ b/enterprise/coderd/workspaceagents_test.go @@ -95,7 +95,7 @@ func TestReinitializeAgent(t *testing.T) { // Ensure that workspace agents can reinitialize against claimed prebuilds in non-default organizations: for _, useDefaultOrg := range []bool{true, false} { - t.Run("", func(t *testing.T) { + t.Run(fmt.Sprintf("useDefaultOrg=%t", useDefaultOrg), func(t *testing.T) { t.Parallel() tempAgentLog := testutil.CreateTemp(t, "", "testReinitializeAgent")