diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 865d6e5..7695412 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -37,6 +37,12 @@ jobs: - name: Run mtree regression tests run: go test -count=1 -v ./tests/integration -run 'TestBuildMtree|TestUpdateMtree|TestMtreeInit|TestACEConn' + - name: Run mtree unit tests + run: go test -count=1 -v ./internal/consistency/mtree + + - name: Run mtree missing-tree fail-fast test + run: go test -count=1 -v ./tests/integration -run 'TestMtreeDiffFailsFastWhenTreeNotBuilt' + - name: Run CDC regression tests run: go test -count=1 -v ./tests/integration -run 'CDC' diff --git a/internal/consistency/mtree/merkle.go b/internal/consistency/mtree/merkle.go index f7775b0..367ea67 100644 --- a/internal/consistency/mtree/merkle.go +++ b/internal/consistency/mtree/merkle.go @@ -88,8 +88,8 @@ type MerkleTreeTask struct { CDCTimeoutSec int // per-invocation CDC drain budget; 0 = use config/default // CDCSkippedNodes lists nodes whose CDC drain was skipped because a running `mtree listen` held the replication slot (best-effort mode). CDCSkippedNodes []string - SkipDBUpdate bool - Until string + SkipDBUpdate bool + Until string untilTime *time.Time @@ -1711,6 +1711,46 @@ func (m *MerkleTreeTask) UpdateMtree(skipAllChecks bool) (err error) { //nolint: }() } + // Pre-flight: read the tree's block size from metadata before any CDC + // work. Block size is fixed at build time, so reading it first is safe — + // and it doubles as an existence check: a table whose tree was never + // built fails fast here with an actionable message instead of after a + // full replication-stream drain. + var blockSize int + var foundBlockSize bool + for _, nodeInfo := range m.ClusterNodes { + pool, err := auth.GetClusterNodeConnection(m.Ctx, nodeInfo, m.connOpts()) + if err != nil { + return fmt.Errorf("error getting connection pool for node %s: %w", nodeInfo["Name"], err) + } + + // Ensure hash_version column exists (schema migration for upgrades). + if err := queries.EnsureHashVersionColumn(m.Ctx, pool); err != nil { + pool.Close() + if isMissingTreeErr(err) { + return m.missingTreeError(nodeInfo["Name"], err) + } + return fmt.Errorf("error migrating metadata schema on node %s: %w", nodeInfo["Name"], err) + } + + blockSize, err = queries.GetBlockSizeFromMetadata(m.Ctx, pool, m.Schema, m.Table) + if err != nil { + pool.Close() + if isMissingTreeErr(err) { + return m.missingTreeError(nodeInfo["Name"], err) + } + return fmt.Errorf("error getting block size from metadata on node %s: %w", nodeInfo["Name"], err) + } + + pool.Close() + foundBlockSize = true + } + + if !foundBlockSize { + return fmt.Errorf("could not determine block size from any node") + } + m.BlockSize = blockSize + if !m.NoCDC { cdcCfg := config.Get().MTree.CDC // Wall-clock budget for the whole CDC catch-up (all nodes drained @@ -1785,35 +1825,6 @@ func (m *MerkleTreeTask) UpdateMtree(skipAllChecks bool) (err error) { //nolint: } } - var blockSize int - var foundBlockSize bool - for _, nodeInfo := range m.ClusterNodes { - pool, err := auth.GetClusterNodeConnection(m.Ctx, nodeInfo, m.connOpts()) - if err != nil { - return fmt.Errorf("error getting connection pool for node %s: %w", nodeInfo["Name"], err) - } - - // Ensure hash_version column exists (schema migration for upgrades). - if err := queries.EnsureHashVersionColumn(m.Ctx, pool); err != nil { - pool.Close() - return fmt.Errorf("error migrating metadata schema on node %s: %w", nodeInfo["Name"], err) - } - - blockSize, err = queries.GetBlockSizeFromMetadata(m.Ctx, pool, m.Schema, m.Table) - if err != nil { - pool.Close() - return fmt.Errorf("error getting block size from metadata on node %s: %w", nodeInfo["Name"], err) - } - - pool.Close() - foundBlockSize = true - } - - if !foundBlockSize { - return fmt.Errorf("could not determine block size from any node") - } - m.BlockSize = blockSize - for _, nodeInfo := range m.ClusterNodes { fmt.Printf("\nUpdating Merkle tree on node: %s\n", nodeInfo["Name"]) pool, err := auth.GetClusterNodeConnection(m.Ctx, nodeInfo, m.connOpts()) @@ -2165,6 +2176,11 @@ func (m *MerkleTreeTask) DiffMtree() (err error) { } if err = m.UpdateMtree(true); err != nil { + // A missing tree already carries a complete, actionable message; + // prefixing it with update-failure context only buries the fix. + if errors.Is(err, ErrMtreeNotFound) { + return err + } return fmt.Errorf("failed to update merkle tree before diff: %w", err) } if len(m.CDCSkippedNodes) > 0 { diff --git a/internal/consistency/mtree/merkle_test.go b/internal/consistency/mtree/merkle_test.go index 9154764..8632df9 100644 --- a/internal/consistency/mtree/merkle_test.go +++ b/internal/consistency/mtree/merkle_test.go @@ -139,16 +139,16 @@ func TestIsNumericColType(t *testing.T) { func TestBuildRowHashQuery(t *testing.T) { tests := []struct { - name string - schema string - table string - key []string - cols []string - whereClause string - colTypes map[string]string - wantContains []string + name string + schema string + table string + key []string + cols []string + whereClause string + colTypes map[string]string + wantContains []string wantNotContain []string - wantOrderBy string + wantOrderBy string }{ { name: "nil colTypes - no trim_scale", diff --git a/internal/consistency/mtree/preflight.go b/internal/consistency/mtree/preflight.go new file mode 100644 index 0000000..7420065 --- /dev/null +++ b/internal/consistency/mtree/preflight.go @@ -0,0 +1,65 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # 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 mtree + +import ( + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +// ErrMtreeNotFound reports that no Merkle tree exists for the target table. +// Callers detect it with errors.Is to surface the message without extra +// wrapping. +var ErrMtreeNotFound = errors.New("no merkle tree found") + +// isMissingTreeErr reports whether err indicates the Merkle tree metadata for +// the table is absent: the metadata row is missing (tree never built), +// ace_mtree_metadata itself does not exist (mtree init never ran, SQLSTATE +// 42P01 undefined_table), or the ace schema itself is absent (mtree init +// never ran on a fresh cluster, SQLSTATE 3F000 invalid_schema_name). +func isMissingTreeErr(err error) bool { + if errors.Is(err, pgx.ErrNoRows) { + return true + } + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && (pgErr.Code == "42P01" || pgErr.Code == "3F000") +} + +// missingTreeError builds the user-facing fail-fast error for a missing tree, +// wrapping ErrMtreeNotFound and pointing at the exact command(s) to run. +// cause is the underlying error that isMissingTreeErr matched: when it is a +// SQLSTATE 3F000 (invalid_schema_name), the pgedge_ace schema itself is +// absent, meaning mtree init never ran, and "ace mtree build" alone would +// fail again since only init creates the schema. In that case the message +// also points at "ace mtree init" first. All other shapes (missing metadata +// row, undefined table) keep the build-only suggestion, since build creates +// the metadata table when the schema already exists. +func (m *MerkleTreeTask) missingTreeError(nodeName any, cause error) error { + dbnameFlag := "" + if m.DBName != "" { + dbnameFlag = " --dbname " + m.DBName + } + buildCmd := fmt.Sprintf("ace mtree build %s %s%s", m.ClusterName, m.QualifiedTableName, dbnameFlag) + + var pgErr *pgconn.PgError + if errors.As(cause, &pgErr) && pgErr.Code == "3F000" { + initCmd := fmt.Sprintf("ace mtree init %s%s", m.ClusterName, dbnameFlag) + return fmt.Errorf("%w for %s on node %v: run '%s' and then '%s' first", + ErrMtreeNotFound, m.QualifiedTableName, nodeName, initCmd, buildCmd) + } + + return fmt.Errorf("%w for %s on node %v: run '%s' first", + ErrMtreeNotFound, m.QualifiedTableName, nodeName, buildCmd) +} diff --git a/internal/consistency/mtree/preflight_test.go b/internal/consistency/mtree/preflight_test.go new file mode 100644 index 0000000..2005364 --- /dev/null +++ b/internal/consistency/mtree/preflight_test.go @@ -0,0 +1,107 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # 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 mtree + +import ( + "errors" + "fmt" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +func TestIsMissingTreeErr(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"no rows", pgx.ErrNoRows, true}, + {"wrapped no rows", fmt.Errorf("query failed: %w", pgx.ErrNoRows), true}, + {"undefined table", &pgconn.PgError{Code: "42P01"}, true}, + {"wrapped undefined table", fmt.Errorf("migrate: %w", &pgconn.PgError{Code: "42P01"}), true}, + {"invalid schema name", &pgconn.PgError{Code: "3F000"}, true}, + {"wrapped invalid schema name", fmt.Errorf("migrate: %w", &pgconn.PgError{Code: "3F000"}), true}, + {"other pg error", &pgconn.PgError{Code: "42501"}, false}, + {"unrelated error", errors.New("connection refused"), false}, + {"nil", nil, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := isMissingTreeErr(tc.err); got != tc.want { + t.Errorf("isMissingTreeErr(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +func TestMissingTreeErrorMessage(t *testing.T) { + m := &MerkleTreeTask{} + m.ClusterName = "demo" + m.QualifiedTableName = "public.time_repair_test2" + m.DBName = "postgres" + + err := m.missingTreeError("n1", pgx.ErrNoRows) + if !errors.Is(err, ErrMtreeNotFound) { + t.Fatalf("expected error to wrap ErrMtreeNotFound, got %v", err) + } + want := "no merkle tree found for public.time_repair_test2 on node n1: " + + "run 'ace mtree build demo public.time_repair_test2 --dbname postgres' first" + if err.Error() != want { + t.Errorf("message mismatch:\n got: %s\nwant: %s", err.Error(), want) + } +} + +func TestMissingTreeErrorMessageOmitsEmptyDBName(t *testing.T) { + m := &MerkleTreeTask{} + m.ClusterName = "demo" + m.QualifiedTableName = "public.t" + + err := m.missingTreeError("n2", pgx.ErrNoRows) + want := "no merkle tree found for public.t on node n2: " + + "run 'ace mtree build demo public.t' first" + if err.Error() != want { + t.Errorf("message mismatch:\n got: %s\nwant: %s", err.Error(), want) + } +} + +func TestMissingTreeErrorMessageSchemaMissingWithDBName(t *testing.T) { + m := &MerkleTreeTask{} + m.ClusterName = "demo" + m.QualifiedTableName = "public.time_repair_test2" + m.DBName = "postgres" + + err := m.missingTreeError("n1", &pgconn.PgError{Code: "3F000"}) + if !errors.Is(err, ErrMtreeNotFound) { + t.Fatalf("expected error to wrap ErrMtreeNotFound, got %v", err) + } + want := "no merkle tree found for public.time_repair_test2 on node n1: " + + "run 'ace mtree init demo --dbname postgres' and then " + + "'ace mtree build demo public.time_repair_test2 --dbname postgres' first" + if err.Error() != want { + t.Errorf("message mismatch:\n got: %s\nwant: %s", err.Error(), want) + } +} + +func TestMissingTreeErrorMessageSchemaMissingOmitsEmptyDBName(t *testing.T) { + m := &MerkleTreeTask{} + m.ClusterName = "demo" + m.QualifiedTableName = "public.t" + + err := m.missingTreeError("n2", &pgconn.PgError{Code: "3F000"}) + want := "no merkle tree found for public.t on node n2: " + + "run 'ace mtree init demo' and then 'ace mtree build demo public.t' first" + if err.Error() != want { + t.Errorf("message mismatch:\n got: %s\nwant: %s", err.Error(), want) + } +} diff --git a/tests/integration/mtree_missing_tree_test.go b/tests/integration/mtree_missing_tree_test.go new file mode 100644 index 0000000..7da8ace --- /dev/null +++ b/tests/integration/mtree_missing_tree_test.go @@ -0,0 +1,60 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # 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 ( + "context" + "fmt" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/pgedge/ace/internal/consistency/mtree" + "github.com/stretchr/testify/require" +) + +// mtree table-diff on a table whose tree was never built must fail +// fast with an actionable message pointing at 'ace mtree build', instead of +// draining the replication stream and then surfacing a raw metadata error. +func TestMtreeDiffFailsFastWhenTreeNotBuilt(t *testing.T) { + ctx := context.Background() + tableName := "mtree_missing_tree_test" + qualifiedTable := fmt.Sprintf("%s.%s", testSchema, tableName) + safeTable := pgx.Identifier{testSchema, tableName}.Sanitize() + nodes := []string{serviceN1, serviceN2} + + // Create the table on both nodes but deliberately skip mtree init/build. + for _, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} { + _, err := pool.Exec(ctx, "CREATE TABLE IF NOT EXISTS "+safeTable+" (id INT PRIMARY KEY, payload TEXT)") // nosemgrep + require.NoError(t, err) + } + t.Cleanup(func() { + for _, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} { + _, _ = pool.Exec(ctx, "DROP TABLE IF EXISTS "+safeTable) // nosemgrep + } + }) + + task := newTestMerkleTreeTask(t, qualifiedTable, nodes) + task.Mode = "diff" + task.Output = "json" + require.NoError(t, task.RunChecks(false)) + + err := task.DiffMtree() + require.Error(t, err, "diff on an unbuilt tree must fail") + require.ErrorIs(t, err, mtree.ErrMtreeNotFound) + require.Contains(t, err.Error(), "no merkle tree found for "+qualifiedTable) + require.Contains(t, err.Error(), "ace mtree build") + require.NotContains(t, err.Error(), "no rows in result set", + "raw driver error must not leak to the user") + require.NotContains(t, err.Error(), "failed to update merkle tree before diff", + "missing-tree error must not be buried under the update wrapper") +}