internal/master/store/jobs_integration_test.go

@@ -339,3 +339,41 @@ func TestCreateJob_RefusesRevokedAgent(t *testing.T) {
 	_, err = s.CreateJob(ctx, nj)
 	assert.NoError(t, err, "a job that was never routed has no agent to check")
 }
+
+// CreateJob must wait for a revoke that is still in flight. Without the share
+// lock it would read the agent as active, insert the job, and the revoke's
+// job cleanup (which ran before the insert committed) would never fail it.
+func TestCreateJob_WaitsForConcurrentRevoke(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, svc := jobFixture(t, s, "a")
+
+	tx, err := s.pool.Begin(ctx)
+	require.NoError(t, err)
+	defer func() { _ = tx.Rollback(ctx) }()
+	_, err = tx.Exec(ctx, `UPDATE agents SET revoked_at = now() WHERE id = $1`, a)
+	require.NoError(t, err)
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := s.CreateJob(ctx, newJob(svc, a))
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		t.Fatalf("CreateJob returned (%v) while the revoke was uncommitted", err)
+	case <-time.After(500 * time.Millisecond):
+	}
+
+	require.NoError(t, tx.Commit(ctx))
+	select {
+	case err := <-done:
+		assert.ErrorIs(t, err, model.ErrAgentOffline)
+	case <-time.After(5 * time.Second):
+		t.Fatal("CreateJob did not return after the revoke committed")
+	}
+	list, err := s.ListJobs(ctx, model.JobFilter{})
+	require.NoError(t, err)
+	assert.Empty(t, list)
+}