chore: augment reconciler test to validate that broken publish does n…

…ot affect behaviour Signed-off-by: Danny Kopping <dannykopping@gmail.com>
coder · dannykopping · May 5, 2025 · May 1, 2025 · May 2, 2025 · May 2, 2025
commit 1c84fac5616770b4ef5cfee1f4018aba749a56ff
diff --git a/coderd/database/dbgen/dbgen.go b/coderd/database/dbgen/dbgen.go
@@ -643,8 +643,8 @@ func ProvisionerJob(t testing.TB, db database.Store, ps pubsub.Pubsub, orig data
 	})
 	require.NoError(t, err, "insert job")
 	if ps != nil {
-		err = provisionerjobs.PostJob(ps, job)
-		require.NoError(t, err, "post job to pubsub")
+		// Advisory, but not essential since acquirer has a background poller to pick up missed jobs.
+		_ = provisionerjobs.PostJob(ps, job)
 	}
 	if !orig.StartedAt.Time.IsZero() {
 		job, err = db.AcquireProvisionerJob(genCtx, database.AcquireProvisionerJobParams{

diff --git a/enterprise/coderd/prebuilds/reconcile_test.go b/enterprise/coderd/prebuilds/reconcile_test.go
@@ -9,6 +9,7 @@
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/xerrors"
 
 	"github.com/coder/coder/v2/coderd/database/dbtime"
 	"github.com/coder/coder/v2/coderd/util/slice"
@@ -303,107 +304,125 @@
 			for _, prebuildLatestTransition := range tc.prebuildLatestTransitions {
 				for _, prebuildJobStatus := range tc.prebuildJobStatuses {
 					for _, templateDeleted := range tc.templateDeleted {
-						t.Run(fmt.Sprintf("%s - %s - %s", tc.name, prebuildLatestTransition, prebuildJobStatus), func(t *testing.T) {
-							t.Parallel()
-							t.Cleanup(func() {
-								if t.Failed() {
-									t.Logf("failed to run test: %s", tc.name)
-									t.Logf("templateVersionActive: %t", templateVersionActive)
-									t.Logf("prebuildLatestTransition: %s", prebuildLatestTransition)
-									t.Logf("prebuildJobStatus: %s", prebuildJobStatus)
+						for _, useBrokenPubsub := range []bool{true, false} {
+							t.Run(fmt.Sprintf("%s - %s - %s - pubsub_broken=%v", tc.name, prebuildLatestTransition, prebuildJobStatus, useBrokenPubsub), func(t *testing.T) {
+								t.Parallel()
+								t.Cleanup(func() {
+									if t.Failed() {
+										t.Logf("failed to run test: %s", tc.name)
+										t.Logf("templateVersionActive: %t", templateVersionActive)
+										t.Logf("prebuildLatestTransition: %s", prebuildLatestTransition)
+										t.Logf("prebuildJobStatus: %s", prebuildJobStatus)
+									}
+								})
+								clock := quartz.NewMock(t)
+								ctx := testutil.Context(t, testutil.WaitShort)
+								cfg := codersdk.PrebuildsConfig{}
+								logger := slogtest.Make(
+									t, &slogtest.Options{IgnoreErrors: true},
+								).Leveled(slog.LevelDebug)
+								db, pubSub := dbtestutil.NewDB(t)
+								if useBrokenPubsub {
+									pubSub = &brokenPublisher{Pubsub: pubSub}
 								}
-							})
-							clock := quartz.NewMock(t)
-							ctx := testutil.Context(t, testutil.WaitShort)
-							cfg := codersdk.PrebuildsConfig{}
-							logger := slogtest.Make(
-								t, &slogtest.Options{IgnoreErrors: true},
-							).Leveled(slog.LevelDebug)
-							db, pubSub := dbtestutil.NewDB(t)
-							controller := prebuilds.NewStoreReconciler(db, pubSub, cfg, logger, quartz.NewMock(t), prometheus.NewRegistry())
-
-							ownerID := uuid.New()
-							dbgen.User(t, db, database.User{
-								ID: ownerID,
-							})
-							org, template := setupTestDBTemplate(t, db, ownerID, templateDeleted)
-							templateVersionID := setupTestDBTemplateVersion(
-								ctx,
-								t,
-								clock,
-								db,
-								pubSub,
-								org.ID,
-								ownerID,
-								template.ID,
-							)
-							preset := setupTestDBPreset(
-								t,
-								db,
-								templateVersionID,
-								1,
-								uuid.New().String(),
-							)
-							prebuild := setupTestDBPrebuild(
-								t,
-								clock,
-								db,
-								pubSub,
-								prebuildLatestTransition,
-								prebuildJobStatus,
-								org.ID,
-								preset,
-								template.ID,
-								templateVersionID,
-							)
-
-							if !templateVersionActive {
-								// Create a new template version and mark it as active
-								// This marks the template version that we care about as inactive
-								setupTestDBTemplateVersion(ctx, t, clock, db, pubSub, org.ID, ownerID, template.ID)
-							}
-
-							// Run the reconciliation multiple times to ensure idempotency
-							// 8 was arbitrary, but large enough to reasonably trust the result
-							for i := 1; i <= 8; i++ {
-								require.NoErrorf(t, controller.ReconcileAll(ctx), "failed on iteration %d", i)
-
-								if tc.shouldCreateNewPrebuild != nil {
-									newPrebuildCount := 0
-									workspaces, err := db.GetWorkspacesByTemplateID(ctx, template.ID)
-									require.NoError(t, err)
-									for _, workspace := range workspaces {
-										if workspace.ID != prebuild.ID {
-											newPrebuildCount++
+
+								controller := prebuilds.NewStoreReconciler(db, pubSub, cfg, logger, quartz.NewMock(t), prometheus.NewRegistry())
+
+								ownerID := uuid.New()
+								dbgen.User(t, db, database.User{
+									ID: ownerID,
+								})
+								org, template := setupTestDBTemplate(t, db, ownerID, templateDeleted)
+								templateVersionID := setupTestDBTemplateVersion(
+									ctx,
+									t,
+									clock,
+									db,
+									pubSub,
+									org.ID,
+									ownerID,
+									template.ID,
+								)
+								preset := setupTestDBPreset(
+									t,
+									db,
+									templateVersionID,
+									1,
+									uuid.New().String(),
+								)
+								prebuild := setupTestDBPrebuild(
+									t,
+									clock,
+									db,
+									pubSub,
+									prebuildLatestTransition,
+									prebuildJobStatus,
+									org.ID,
+									preset,
+									template.ID,
+									templateVersionID,
+								)
+
+								if !templateVersionActive {
+									// Create a new template version and mark it as active
+									// This marks the template version that we care about as inactive
+									setupTestDBTemplateVersion(ctx, t, clock, db, pubSub, org.ID, ownerID, template.ID)
+								}
+
+								// Run the reconciliation multiple times to ensure idempotency
+								// 8 was arbitrary, but large enough to reasonably trust the result
+								for i := 1; i <= 8; i++ {
+									require.NoErrorf(t, controller.ReconcileAll(ctx), "failed on iteration %d", i)
+
+									if tc.shouldCreateNewPrebuild != nil {
+										newPrebuildCount := 0
+										workspaces, err := db.GetWorkspacesByTemplateID(ctx, template.ID)
+										require.NoError(t, err)
+										for _, workspace := range workspaces {
+											if workspace.ID != prebuild.ID {
+												newPrebuildCount++
+											}
 										}
+										// This test configures a preset that desires one prebuild.
+										// In cases where new prebuilds should be created, there should be exactly one.
+										require.Equal(t, *tc.shouldCreateNewPrebuild, newPrebuildCount == 1)
 									}
-									// This test configures a preset that desires one prebuild.
-									// In cases where new prebuilds should be created, there should be exactly one.
-									require.Equal(t, *tc.shouldCreateNewPrebuild, newPrebuildCount == 1)
-								}
 
-								if tc.shouldDeleteOldPrebuild != nil {
-									builds, err := db.GetWorkspaceBuildsByWorkspaceID(ctx, database.GetWorkspaceBuildsByWorkspaceIDParams{
-										WorkspaceID: prebuild.ID,
-									})
-									require.NoError(t, err)
-									if *tc.shouldDeleteOldPrebuild {
-										require.Equal(t, 2, len(builds))
-										require.Equal(t, database.WorkspaceTransitionDelete, builds[0].Transition)
-									} else {
-										require.Equal(t, 1, len(builds))
-										require.Equal(t, prebuildLatestTransition, builds[0].Transition)
+									if tc.shouldDeleteOldPrebuild != nil {
+										builds, err := db.GetWorkspaceBuildsByWorkspaceID(ctx, database.GetWorkspaceBuildsByWorkspaceIDParams{
+											WorkspaceID: prebuild.ID,
+										})
+										require.NoError(t, err)
+										if *tc.shouldDeleteOldPrebuild {
+											require.Equal(t, 2, len(builds))
+											require.Equal(t, database.WorkspaceTransitionDelete, builds[0].Transition)
+										} else {
+											require.Equal(t, 1, len(builds))
+											require.Equal(t, prebuildLatestTransition, builds[0].Transition)
+										}
 									}
 								}
-							}
-						})
+							})
+						}
 					}
 				}
 			}
 		}
 	}
 }
 
+// brokenPublisher is used to validate that Publish() calls which always fail do not affect the reconciler's behaviour,
+// since the messages published are not essential but merely advisory.
+type brokenPublisher struct {
+	pubsub.Pubsub
+}
+
+func (b *brokenPublisher) Publish(event string, _ []byte) error {
+	// I'm explicitly _not_ checking for EventJobPosted (coderd/database/provisionerjobs/provisionerjobs.go) since that
+	// required too much knowledge of the underlying implementation.
+	return xerrors.Errorf("refusing to publish %q", event)
+}
+
 func TestMultiplePresetsPerTemplateVersion(t *testing.T) {
 	t.Parallel()