From 6302edf6c3f5a601d4f5f16f37cd4d55a3e4776d Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Thu, 13 Aug 2026 13:30:37 -0400 Subject: [PATCH] feat: fail-fast preflight for schema upgrade path Before running ALTER EXTENSION documentdb UPDATE, query pg_extension_update_paths to confirm PostgreSQL can resolve a chain of update scripts from the installed schema to the target. The documentdb extension ships minors frequently, so a user may bump more than one minor in a single change; that only succeeds when the extension packages the intermediate documentdb--A--B.sql scripts. When the chain is broken, ALTER EXTENSION UPDATE would otherwise fail every reconcile with a raw "no update path" error. The preflight detects the missing path up front, skips the ALTER, and emits a SchemaUpgradePathMissing Warning event (mirroring the existing ExtensionRollback precedent) so the schema and data are left untouched and the operator surfaces an actionable message instead of looping. Adds unit coverage for the no-path skip case and for extensionUpdatePathExists (source==target short-circuit, has-path, no-path, and SQL error propagation). Existing ALTER-path mocks are converted from order-based to content-based branching to account for the extra preflight query. Refs: #439 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d53b8fec-0585-4888-89da-d9a6847597a5 Signed-off-by: Wenting Wu --- .../controller/documentdb_controller.go | 69 +++++++ .../controller/documentdb_controller_test.go | 193 ++++++++++++++++-- 2 files changed, 245 insertions(+), 17 deletions(-) diff --git a/operator/src/internal/controller/documentdb_controller.go b/operator/src/internal/controller/documentdb_controller.go index 88bf1d799..07a6fc00f 100644 --- a/operator/src/internal/controller/documentdb_controller.go +++ b/operator/src/internal/controller/documentdb_controller.go @@ -769,6 +769,46 @@ func parseExtensionVersionsFromOutput(output string) (defaultVersion, installedV return defaultVersion, installedVersion, true } +// updatePathSentinelYes / updatePathSentinelNo are the tokens the preflight +// query emits so the result can be parsed out of psql's aligned table output +// without depending on column layout. +const ( + updatePathSentinelYes = "HAS_UPDATE_PATH" + updatePathSentinelNo = "NO_UPDATE_PATH" +) + +// extensionUpdatePathExists reports whether PostgreSQL can resolve an +// ALTER EXTENSION update path for the documentdb extension from source to +// target. Both versions are in pg_available_extensions format (e.g. +// "0.109-0"). It queries pg_extension_update_paths, which exposes the same +// version graph ALTER EXTENSION UPDATE walks internally: a row exists for a +// resolvable (source, target) pair with a non-NULL path, and the path is NULL +// when no chain of update scripts connects them. +// +// The query returns a unique sentinel token rather than the raw path so the +// answer survives psql's aligned output formatting. +func (r *DocumentDBReconciler) extensionUpdatePathExists( + ctx context.Context, + cluster *cnpgv1.Cluster, + source, target string, +) (bool, error) { + // A no-op update (source == target) has no path row but is trivially safe. + if source == target { + return true, nil + } + + query := fmt.Sprintf( + "SELECT CASE WHEN EXISTS (SELECT 1 FROM pg_extension_update_paths('documentdb') "+ + "WHERE source = '%s' AND target = '%s' AND path IS NOT NULL) THEN '%s' ELSE '%s' END", + source, target, updatePathSentinelYes, updatePathSentinelNo) + + output, err := r.SQLExecutor(ctx, cluster, query) + if err != nil { + return false, fmt.Errorf("query pg_extension_update_paths: %w", err) + } + return strings.Contains(output, updatePathSentinelYes), nil +} + // handleExtensionUpgrade handles the ALTER EXTENSION lifecycle after images have been synced // by SyncCnpgCluster. It: // 1. Updates DocumentDB status with the current images from the CNPG cluster @@ -866,6 +906,35 @@ func (r *DocumentDBReconciler) handleExtensionUpgrade(ctx context.Context, curre return nil } + // Preflight: confirm PostgreSQL can resolve an update-script path from the + // installed schema to the target before attempting ALTER EXTENSION. The + // documentdb extension ships minors frequently, so a user may jump more + // than one minor in a single bump; that only works if the extension + // packages the intermediate documentdb--A--B.sql scripts so PostgreSQL can + // chain them. When the chain is broken, ALTER EXTENSION UPDATE fails at + // execution time with a raw "no update path" error every reconcile. Detect + // it here and surface an actionable event instead, leaving the schema and + // data untouched. + pathExists, err := r.extensionUpdatePathExists(ctx, currentCluster, installedVersion, schemaTarget) + if err != nil { + return fmt.Errorf("failed to check extension update path from %s to %s: %w", + installedVersion, schemaTarget, err) + } + if !pathExists { + msg := fmt.Sprintf( + "No supported schema upgrade path from %s to %s. The documentdb extension does not "+ + "ship the update scripts needed to bridge this gap in one step. Upgrade to an "+ + "intermediate version first, or report a missing migration script for this release. "+ + "ALTER EXTENSION UPDATE was skipped; the schema and data are unchanged.", + util.ExtensionVersionToSemver(installedVersion), + util.ExtensionVersionToSemver(schemaTarget)) + logger.Info(msg) + if r.Recorder != nil { + r.Recorder.Event(documentdb, corev1.EventTypeWarning, "SchemaUpgradePathMissing", msg) + } + return nil + } + // Run ALTER EXTENSION to upgrade logger.Info("Upgrading DocumentDB extension", "fromVersion", installedVersion, diff --git a/operator/src/internal/controller/documentdb_controller_test.go b/operator/src/internal/controller/documentdb_controller_test.go index 889cd7065..3d607413a 100644 --- a/operator/src/internal/controller/documentdb_controller_test.go +++ b/operator/src/internal/controller/documentdb_controller_test.go @@ -838,11 +838,15 @@ var _ = Describe("DocumentDB Controller", func() { Recorder: recorder, SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) { sqlCalls = append(sqlCalls, sql) - if len(sqlCalls) == 1 { - // First call: version check — installed 0.109-0, default 0.110-0 + if strings.Contains(sql, "pg_available_extensions") { + // Version check — installed 0.109-0, default 0.110-0 return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil } - // Second call: ALTER EXTENSION + if strings.Contains(sql, "pg_extension_update_paths") { + // Preflight — a valid update path exists + return "HAS_UPDATE_PATH", nil + } + // ALTER EXTENSION return "ALTER EXTENSION", nil }, } @@ -850,10 +854,11 @@ var _ = Describe("DocumentDB Controller", func() { err := reconciler.handleExtensionUpgrade(ctx, cluster, documentdb) Expect(err).ToNot(HaveOccurred()) - // Verify both SQL calls were made - Expect(sqlCalls).To(HaveLen(2)) + // Verify version check, update-path preflight, and ALTER were called + Expect(sqlCalls).To(HaveLen(3)) Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions")) - Expect(sqlCalls[1]).To(Equal("ALTER EXTENSION documentdb UPDATE")) + Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths")) + Expect(sqlCalls[2]).To(Equal("ALTER EXTENSION documentdb UPDATE")) // Status should reflect the upgraded version (default version as semver) updatedDB := &dbpreview.DocumentDB{} @@ -861,6 +866,88 @@ var _ = Describe("DocumentDB Controller", func() { Expect(updatedDB.Status.SchemaVersion).To(Equal("0.110.0")) }) + It("should skip ALTER EXTENSION and emit an event when no update path exists", func() { + cluster := &cnpgv1.Cluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: clusterName, + Namespace: clusterNamespace, + }, + Spec: cnpgv1.ClusterSpec{ + PostgresConfiguration: cnpgv1.PostgresConfiguration{ + Extensions: []cnpgv1.ExtensionConfiguration{ + { + Name: "documentdb", + ImageVolumeSource: corev1.ImageVolumeSource{ + Reference: "documentdb/documentdb:v1.0.0", + }, + }, + }, + }, + }, + Status: cnpgv1.ClusterStatus{ + CurrentPrimary: "test-cluster-1", + InstancesStatus: map[cnpgv1.PodStatus][]string{ + cnpgv1.PodHealthy: {"test-cluster-1"}, + }, + }, + } + + documentdb := &dbpreview.DocumentDB{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-documentdb", + Namespace: clusterNamespace, + }, + Spec: dbpreview.DocumentDBSpec{ + SchemaVersion: "auto", + }, + } + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(cluster, documentdb). + WithStatusSubresource(&dbpreview.DocumentDB{}). + Build() + + sqlCalls := []string{} + reconciler := &DocumentDBReconciler{ + Client: fakeClient, + Scheme: scheme, + Recorder: recorder, + SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) { + sqlCalls = append(sqlCalls, sql) + if strings.Contains(sql, "pg_available_extensions") { + // Version check — installed 0.109-0, default 0.110-0 + return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil + } + if strings.Contains(sql, "pg_extension_update_paths") { + // Preflight — no update path bridges installed → target + return "NO_UPDATE_PATH", nil + } + // ALTER EXTENSION must not be reached + return "ALTER EXTENSION", nil + }, + } + + err := reconciler.handleExtensionUpgrade(ctx, cluster, documentdb) + Expect(err).ToNot(HaveOccurred()) + + // Only version check and preflight ran — ALTER EXTENSION was skipped + Expect(sqlCalls).To(HaveLen(2)) + Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions")) + Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths")) + + // A Warning event should surface the missing upgrade path + Expect(recorder.Events).To(HaveLen(1)) + event := <-recorder.Events + Expect(event).To(ContainSubstring("SchemaUpgradePathMissing")) + + // Schema version must stay at the installed version (no premature + // bump to the target since the migration was skipped) + updatedDB := &dbpreview.DocumentDB{} + Expect(fakeClient.Get(ctx, types.NamespacedName{Name: "test-documentdb", Namespace: clusterNamespace}, updatedDB)).To(Succeed()) + Expect(updatedDB.Status.SchemaVersion).ToNot(Equal("0.110.0")) + }) + It("should return error when ALTER EXTENSION fails", func() { cluster := &cnpgv1.Cluster{ ObjectMeta: metav1.ObjectMeta{ @@ -903,17 +990,19 @@ var _ = Describe("DocumentDB Controller", func() { WithStatusSubresource(&dbpreview.DocumentDB{}). Build() - callCount := 0 reconciler := &DocumentDBReconciler{ Client: fakeClient, Scheme: scheme, Recorder: recorder, SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) { - callCount++ - if callCount == 1 { + if strings.Contains(sql, "pg_available_extensions") { // Version check: upgrade needed return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil } + if strings.Contains(sql, "pg_extension_update_paths") { + // Preflight — a valid update path exists, so ALTER is attempted + return "HAS_UPDATE_PATH", nil + } // ALTER EXTENSION fails return "", fmt.Errorf("permission denied") }, @@ -1212,10 +1301,13 @@ var _ = Describe("DocumentDB Controller", func() { Recorder: recorder, SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) { sqlCalls = append(sqlCalls, sql) - if len(sqlCalls) == 1 { + if strings.Contains(sql, "pg_available_extensions") { // default > installed → triggers ALTER EXTENSION return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil } + if strings.Contains(sql, "pg_extension_update_paths") { + return "HAS_UPDATE_PATH", nil + } return "ALTER EXTENSION", nil }, } @@ -1223,7 +1315,66 @@ var _ = Describe("DocumentDB Controller", func() { err := reconciler.handleExtensionUpgrade(ctx, cluster, documentdb) Expect(err).To(HaveOccurred()) Expect(err.Error()).To(ContainSubstring("failed to update DocumentDB status after schema upgrade")) - Expect(sqlCalls).To(HaveLen(2)) + Expect(sqlCalls).To(HaveLen(3)) + }) + }) + + Describe("extensionUpdatePathExists", func() { + It("returns true without querying when source equals target", func() { + called := false + reconciler := &DocumentDBReconciler{ + SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, _ string) (string, error) { + called = true + return "", nil + }, + } + + exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.110-0", "0.110-0") + Expect(err).ToNot(HaveOccurred()) + Expect(exists).To(BeTrue()) + Expect(called).To(BeFalse()) + }) + + It("returns true when the preflight query reports a path", func() { + var gotSQL string + reconciler := &DocumentDBReconciler{ + SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) { + gotSQL = sql + return " case \n--------\n HAS_UPDATE_PATH\n", nil + }, + } + + exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.109-0", "0.110-0") + Expect(err).ToNot(HaveOccurred()) + Expect(exists).To(BeTrue()) + Expect(gotSQL).To(ContainSubstring("pg_extension_update_paths")) + Expect(gotSQL).To(ContainSubstring("source = '0.109-0'")) + Expect(gotSQL).To(ContainSubstring("target = '0.110-0'")) + }) + + It("returns false when the preflight query reports no path", func() { + reconciler := &DocumentDBReconciler{ + SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, _ string) (string, error) { + return " case \n--------\n NO_UPDATE_PATH\n", nil + }, + } + + exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.109-0", "0.130-0") + Expect(err).ToNot(HaveOccurred()) + Expect(exists).To(BeFalse()) + }) + + It("propagates SQL execution errors", func() { + reconciler := &DocumentDBReconciler{ + SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, _ string) (string, error) { + return "", fmt.Errorf("connection refused") + }, + } + + exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.109-0", "0.110-0") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("query pg_extension_update_paths")) + Expect(exists).To(BeFalse()) }) }) @@ -1347,9 +1498,12 @@ var _ = Describe("DocumentDB Controller", func() { Recorder: recorder, SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) { sqlCalls = append(sqlCalls, sql) - if len(sqlCalls) == 1 { + if strings.Contains(sql, "pg_available_extensions") { return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil } + if strings.Contains(sql, "pg_extension_update_paths") { + return "HAS_UPDATE_PATH", nil + } return "ALTER EXTENSION", nil }, } @@ -1358,9 +1512,10 @@ var _ = Describe("DocumentDB Controller", func() { Expect(err).ToNot(HaveOccurred()) // Both version-check and ALTER EXTENSION should have been called - Expect(sqlCalls).To(HaveLen(2)) + Expect(sqlCalls).To(HaveLen(3)) Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions")) - Expect(sqlCalls[1]).To(Equal("ALTER EXTENSION documentdb UPDATE")) + Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths")) + Expect(sqlCalls[2]).To(Equal("ALTER EXTENSION documentdb UPDATE")) // Status should reflect the upgraded version updatedDB := &dbpreview.DocumentDB{} @@ -1418,10 +1573,13 @@ var _ = Describe("DocumentDB Controller", func() { Recorder: recorder, SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) { sqlCalls = append(sqlCalls, sql) - if len(sqlCalls) == 1 { + if strings.Contains(sql, "pg_available_extensions") { // Binary is 0.110-0, installed is 0.109-0 return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil } + if strings.Contains(sql, "pg_extension_update_paths") { + return "HAS_UPDATE_PATH", nil + } return "ALTER EXTENSION", nil }, } @@ -1430,9 +1588,10 @@ var _ = Describe("DocumentDB Controller", func() { Expect(err).ToNot(HaveOccurred()) // Should run ALTER EXTENSION UPDATE TO specific version - Expect(sqlCalls).To(HaveLen(2)) + Expect(sqlCalls).To(HaveLen(3)) Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions")) - Expect(sqlCalls[1]).To(Equal("ALTER EXTENSION documentdb UPDATE TO '0.110-0'")) + Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths")) + Expect(sqlCalls[2]).To(Equal("ALTER EXTENSION documentdb UPDATE TO '0.110-0'")) // Status should reflect the explicit version updatedDB := &dbpreview.DocumentDB{}