fix(mtree): fail fast when no Merkle tree exists for a table

.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'
 

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 {

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",

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)
+}

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)
+	}
+}

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")
+}