diff --git a/internal/consistency/diff/repset_diff.go b/internal/consistency/diff/repset_diff.go index eb97dc6..4945854 100644 --- a/internal/consistency/diff/repset_diff.go +++ b/internal/consistency/diff/repset_diff.go @@ -45,6 +45,7 @@ type RepsetDiffCmd struct { SkipFile string skipTablesList []string tableList []string + coldFrontExcluded []string missingTables []MissingTableInfo nodeList []string clusterNodes []map[string]any @@ -226,10 +227,24 @@ func (c *RepsetDiffCmd) RunChecks(skipValidation bool) error { return missingTables[i].Table < missingTables[j].Table }) - c.tableList = allTables + // A repset can span several schemas, so drop anything belonging to the + // ColdFront schema. + coldFrontPrefix := coldFrontSchemaName + "." + var filteredTables []string + var coldFrontExcluded []string + for _, table := range allTables { + if strings.HasPrefix(table, coldFrontPrefix) { + coldFrontExcluded = append(coldFrontExcluded, table) + continue + } + filteredTables = append(filteredTables, table) + } + + c.tableList = filteredTables + c.coldFrontExcluded = coldFrontExcluded c.missingTables = missingTables - if len(c.tableList) == 0 { + if len(c.tableList) == 0 && len(c.coldFrontExcluded) == 0 { return fmt.Errorf("no tables found in repset %s", c.RepsetName) } @@ -361,6 +376,13 @@ func RepsetDiff(task *RepsetDiffCmd) (err error) { } }() + for _, tableName := range task.coldFrontExcluded { + if !task.Quiet { + logger.Info("Skipping table: %s (pgEdge ColdFront schema)", tableName) + } + skippedTables = append(skippedTables, fmt.Sprintf("%s (pgEdge ColdFront schema)", tableName)) + } + for _, tableName := range task.tableList { var skipped bool for _, skip := range task.skipTablesList { diff --git a/internal/consistency/diff/schema_diff.go b/internal/consistency/diff/schema_diff.go index ff52175..e2bd12a 100644 --- a/internal/consistency/diff/schema_diff.go +++ b/internal/consistency/diff/schema_diff.go @@ -157,6 +157,10 @@ func (c *SchemaDiffCmd) parseSkipList() error { return nil } +// coldFrontSchemaName holds the internal state of the pgEdge ColdFront +// extension, not user data. ACE never diffs or repairs it. +const coldFrontSchemaName = "coldfront" + func (c *SchemaDiffCmd) Validate() error { if c.ClusterName == "" { return fmt.Errorf("cluster name is required") @@ -164,6 +168,9 @@ func (c *SchemaDiffCmd) Validate() error { if c.SchemaName == "" { return fmt.Errorf("schema name is required") } + if c.SchemaName == coldFrontSchemaName { + return fmt.Errorf("schema %q is reserved for the pgEdge ColdFront extension; ACE does not diff it", coldFrontSchemaName) + } nodeList, err := utils.ParseNodes(c.Nodes) if err != nil { diff --git a/internal/consistency/diff/schema_diff_test.go b/internal/consistency/diff/schema_diff_test.go index f454a2a..4fcc696 100644 --- a/internal/consistency/diff/schema_diff_test.go +++ b/internal/consistency/diff/schema_diff_test.go @@ -18,6 +18,58 @@ import ( "testing" ) +// --------------------------------------------------------------------------- +// Validate +// --------------------------------------------------------------------------- + +func TestSchemaDiffCmd_Validate(t *testing.T) { + tests := []struct { + name string + cmd SchemaDiffCmd + wantErr bool + errContains string + }{ + { + name: "missing cluster name", + cmd: SchemaDiffCmd{SchemaName: "public"}, + wantErr: true, + errContains: "cluster name is required", + }, + { + name: "missing schema name", + cmd: SchemaDiffCmd{ClusterName: "test_cluster"}, + wantErr: true, + errContains: "schema name is required", + }, + { + name: "coldfront schema is rejected", + cmd: SchemaDiffCmd{ClusterName: "test_cluster", SchemaName: "coldfront"}, + wantErr: true, + errContains: "reserved for the pgEdge ColdFront extension", + }, + { + name: "valid", + cmd: SchemaDiffCmd{ClusterName: "test_cluster", SchemaName: "public", Nodes: "n1,n2"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := tc.cmd.Validate() + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if tc.errContains != "" && !strings.Contains(err.Error(), tc.errContains) { + t.Errorf("error = %q, want it to contain %q", err.Error(), tc.errContains) + } + } else if err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} + // writeSkipFile is a helper that writes lines to a temp file and returns its path. func writeSkipFile(t *testing.T, lines ...string) string { t.Helper() diff --git a/tests/integration/coldfront_exclusion_test.go b/tests/integration/coldfront_exclusion_test.go new file mode 100644 index 0000000..879c29d --- /dev/null +++ b/tests/integration/coldfront_exclusion_test.go @@ -0,0 +1,119 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # ACE - Active Consistency Engine +// +// Copyright (C) 2023 - 2026, pgEdge (https://www.pgedge.com/) +// +// This software is released under the PostgreSQL License: +// https://opensource.org/license/postgresql +// +// /////////////////////////////////////////////////////////////////////////// + +package integration + +import ( + "bytes" + "context" + "fmt" + "os" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/pgedge/ace/internal/consistency/diff" + "github.com/pgedge/ace/pkg/logger" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestSchemaDiff_ColdFrontSchemaRejected verifies that schema-diff refuses +// to diff the ColdFront schema. The rejection happens in Validate(), so no +// setup is required. +func TestSchemaDiff_ColdFrontSchemaRejected(t *testing.T) { + nodes := fmt.Sprintf("%s,%s", serviceN1, serviceN2) + + task := newTestSchemaDiffTask("coldfront", nodes) + err := task.SchemaTableDiff() + + require.Error(t, err, "schema-diff must refuse to diff the coldfront schema") + assert.Contains(t, err.Error(), "reserved for the pgEdge ColdFront extension") +} + +// TestRepsetDiff_ColdFrontSchemaExcluded verifies that repset-diff skips +// tables in the ColdFront schema even when they are part of the repset. +// The exclusion keys off the schema name only, so the extension itself +// does not need to be installed. +func TestRepsetDiff_ColdFrontSchemaExcluded(t *testing.T) { + ctx := context.Background() + const repsetName = "default" + const coldfrontTable = "claims_stub" + qualifiedColdfront := fmt.Sprintf("coldfront.%s", coldfrontTable) + + pools := []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} + + // Diverge the data between nodes: a broken exclusion would report it + // as a difference. + createSQL := fmt.Sprintf(` + CREATE SCHEMA IF NOT EXISTS coldfront; + CREATE TABLE IF NOT EXISTS %s ( + id INT PRIMARY KEY, + val TEXT + )`, qualifiedColdfront) + for _, pool := range pools { + _, err := pool.Exec(ctx, createSQL) + require.NoError(t, err) + } + _, err := pgCluster.Node1Pool.Exec(ctx, fmt.Sprintf( + `INSERT INTO %s (id, val) VALUES (1, 'from_n1') ON CONFLICT DO NOTHING`, qualifiedColdfront)) + require.NoError(t, err) + _, err = pgCluster.Node2Pool.Exec(ctx, fmt.Sprintf( + `INSERT INTO %s (id, val) VALUES (1, 'from_n2') ON CONFLICT DO NOTHING`, qualifiedColdfront)) + require.NoError(t, err) + + for _, pool := range pools { + _, err := pool.Exec(ctx, + fmt.Sprintf(`SELECT spock.repset_add_table('%s', '%s');`, repsetName, qualifiedColdfront)) + require.NoError(t, err) + } + t.Cleanup(func() { + for _, pool := range pools { + pool.Exec(ctx, fmt.Sprintf( + `SELECT spock.repset_remove_table('%s', '%s');`, repsetName, qualifiedColdfront)) + pool.Exec(ctx, fmt.Sprintf(`DROP TABLE IF EXISTS %s CASCADE`, qualifiedColdfront)) + } + }) + + // Control table: ordinary tables must still be diffed. + controlQualified := createRepsetDiffTable(t, "cf_excl_control_tbl", repsetName, false) + + r, w, err := os.Pipe() + require.NoError(t, err) + logger.SetOutput(w) + t.Cleanup(func() { logger.SetOutput(os.Stderr) }) + + task := newTestRepsetDiffTask(repsetName) + diffErr := diff.RepsetDiff(task) + + w.Close() + var buf bytes.Buffer + buf.ReadFrom(r) + logOutput := buf.String() + t.Logf("Captured log output:\n%s", logOutput) + + require.NoError(t, diffErr) + + summarySection := extractSummarySection(t, logOutput) + + skippedSection := extractBetween(summarySection, "table(s) were skipped:", "table(s)") + assert.Contains(t, skippedSection, qualifiedColdfront, + "the ColdFront table should be reported as skipped") + assert.Contains(t, skippedSection, "pgEdge ColdFront schema", + "the skipped entry should state why it was excluded") + + identicalSection := extractBetween(summarySection, "table(s) are identical:", "table(s)") + assert.Contains(t, identicalSection, controlQualified, + "the control table should still be diffed") + + diffSection := extractBetween(summarySection, "table(s) have differences:", "") + assert.NotContains(t, diffSection, coldfrontTable, + "the ColdFront table must not be reported as differing") +}