Fix silent table exclusion in schema-diff and repset-diff

Tables that failed during per-table diff (e.g. missing primary key) were
silently dropped from the report — only a logger.Warn on stderr, with
overall status still COMPLETED. This made it possible for a user to see
"no differences" while an entire table was never compared.

Changes:
- Add end-of-run summary listing identical, skipped, differed, missing,
  and errored tables with error reasons (shared via DiffSummary)
- Set task status to FAILED when any table errors
- Query tables from all nodes (not just the first) and report tables
  not present on every node
- Support uni-directional repset-diff where the repset only exists on
  the publisher node
- Add integration tests covering all summary categories

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: mason-sharp

internal/consistency/diff/diff_summary.go

@@ -0,0 +1,88 @@
+// ///////////////////////////////////////////////////////////////////////////
+//
+// # 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 diff
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/pgedge/ace/pkg/logger"
+)
+
+// MissingTableInfo records a table that was not found on every node.
+type MissingTableInfo struct {
+	Table       string   // schema-qualified table name
+	PresentOn   []string // node names where the table exists
+	MissingFrom []string // node names where the table does not exist
+}
+
+// FailedTableInfo records a table whose diff failed along with the reason.
+type FailedTableInfo struct {
+	Table string // schema-qualified table name
+	Err   error  // the error that caused the failure
+}
+
+// DiffSummary collects per-table outcomes during a schema-diff or repset-diff
+// run and prints a human-readable summary at the end.
+type DiffSummary struct {
+	MatchedTables  []string
+	DifferedTables []string
+	FailedTables   []FailedTableInfo
+	MissingTables  []MissingTableInfo
+	SkippedTables  []string
+}
+
+// PrintAndFinalize logs the summary and returns an error if any tables failed.
+// The label identifies the scope (e.g. "schema public" or "repset default").
+func (s *DiffSummary) PrintAndFinalize(commandName, label string) error {
+	total := len(s.MatchedTables) + len(s.DifferedTables) + len(s.FailedTables) +
+		len(s.SkippedTables) + len(s.MissingTables)
+	logger.Info("%s summary: %d table(s) in %s", commandName, total, label)
+	if len(s.MatchedTables) > 0 {
+		logger.Info("  %d table(s) are identical:", len(s.MatchedTables))
+		for _, t := range s.MatchedTables {
+			logger.Info("    - %s", t)
+		}
+	}
+	if len(s.SkippedTables) > 0 {
+		logger.Info("  %d table(s) were skipped:", len(s.SkippedTables))
+		for _, t := range s.SkippedTables {
+			logger.Info("    - %s", t)
+		}
+	}
+	if len(s.DifferedTables) > 0 {
+		logger.Warn("  %d table(s) have differences:", len(s.DifferedTables))
+		for _, t := range s.DifferedTables {
+			logger.Warn("    - %s", t)
+		}
+	}
+	if len(s.MissingTables) > 0 {
+		logger.Warn("  %d table(s) not present on all nodes:", len(s.MissingTables))
+		for _, mt := range s.MissingTables {
+			logger.Warn("    - %s found on [%s] but missing from [%s]",
+				mt.Table,
+				strings.Join(mt.PresentOn, ", "),
+				strings.Join(mt.MissingFrom, ", "))
+		}
+	}
+	if len(s.FailedTables) > 0 {
+		logger.Error("  %d table(s) encountered errors and could not be compared:", len(s.FailedTables))
+		var names []string
+		for _, ft := range s.FailedTables {
+			logger.Error("    - %s: %v", ft.Table, ft.Err)
+			names = append(names, ft.Table)
+		}
+		return fmt.Errorf("%d table(s) failed during %s: %v", len(s.FailedTables), commandName, names)
+	}
+
+	return nil
+}

internal/consistency/diff/repset_diff.go

@@ -17,6 +17,7 @@ import (
 	"fmt"
 	"maps"
 	"os"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -43,6 +44,7 @@ type RepsetDiffCmd struct {
 	SkipFile          string
 	skipTablesList    []string
 	tableList         []string
+	missingTables     []MissingTableInfo
 	nodeList          []string
 	clusterNodes      []map[string]any
 	database          types.Database
@@ -129,38 +131,64 @@ func (c *RepsetDiffCmd) RunChecks(skipValidation bool) error {
 		return fmt.Errorf("no nodes found in cluster config")
 	}
 
-	firstNode := c.clusterNodes[0]
+	// Query repset tables from every node that has the repset and build a union.
+	// In a uni-directional setup the repset may only exist on the publisher, but
+	// the tables themselves exist on all nodes, so we still diff them across
+	// every node.
+	tableSet := make(map[string]bool)
+	var nodesWithRepset int
+
+	for _, nodeInfo := range c.clusterNodes {
+		nodeName := nodeInfo["Name"].(string)
+
+		nodeWithDBInfo := make(map[string]any)
+		maps.Copy(nodeWithDBInfo, nodeInfo)
+		utils.ApplyDatabaseCredentials(nodeWithDBInfo, c.database)
+		if portVal, ok := nodeWithDBInfo["Port"]; ok {
+			if portFloat, isFloat := portVal.(float64); isFloat {
+				nodeWithDBInfo["Port"] = strconv.Itoa(int(portFloat))
+			}
+		}
 
-	nodeWithDBInfo := make(map[string]any)
-	maps.Copy(nodeWithDBInfo, firstNode)
-	utils.ApplyDatabaseCredentials(nodeWithDBInfo, c.database)
+		pool, err := auth.GetClusterNodeConnection(c.Ctx, nodeWithDBInfo, c.connOpts())
+		if err != nil {
+			return fmt.Errorf("could not connect to node %s: %w", nodeName, err)
+		}
 
-	if portVal, ok := nodeWithDBInfo["Port"]; ok {
-		if portFloat, isFloat := portVal.(float64); isFloat {
-			nodeWithDBInfo["Port"] = strconv.Itoa(int(portFloat))
+		repsetExists, err := queries.CheckRepSetExists(c.Ctx, pool, c.RepsetName)
+		if err != nil {
+			pool.Close()
+			return fmt.Errorf("could not check if repset exists on node %s: %w", nodeName, err)
 		}
-	}
+		if !repsetExists {
+			pool.Close()
+			logger.Warn("repset %s not found on node %s, skipping for table discovery", c.RepsetName, nodeName)
+			continue
+		}
+		nodesWithRepset++
 
-	pool, err := auth.GetClusterNodeConnection(c.Ctx, nodeWithDBInfo, c.connOpts())
-	if err != nil {
-		return fmt.Errorf("could not connect to database: %w", err)
-	}
-	defer pool.Close()
+		tables, err := queries.GetTablesInRepSet(c.Ctx, pool, c.RepsetName)
+		pool.Close()
+		if err != nil {
+			return fmt.Errorf("could not get tables in repset on node %s: %w", nodeName, err)
+		}
 
-	repsetExists, err := queries.CheckRepSetExists(c.Ctx, pool, c.RepsetName)
-	if err != nil {
-		return fmt.Errorf("could not check if repset exists: %w", err)
+		for _, t := range tables {
+			tableSet[t] = true
+		}
 	}
-	if !repsetExists {
-		return fmt.Errorf("repset %s not found", c.RepsetName)
+
+	if nodesWithRepset == 0 {
+		return fmt.Errorf("repset %s not found on any node", c.RepsetName)
 	}
 
-	tables, err := queries.GetTablesInRepSet(c.Ctx, pool, c.RepsetName)
-	if err != nil {
-		return fmt.Errorf("could not get tables in repset: %w", err)
+	var allTables []string
+	for t := range tableSet {
+		allTables = append(allTables, t)
 	}
+	sort.Strings(allTables)
 
-	c.tableList = tables
+	c.tableList = allTables
 
 	if len(c.tableList) == 0 {
 		return fmt.Errorf("no tables found in repset %s", c.RepsetName)
@@ -230,8 +258,10 @@ func RepsetDiff(task *RepsetDiffCmd) (err error) {
 		}
 	}
 
-	var tablesProcessed, tablesFailed, tablesSkipped int
-	var failedTables []string
+	var tablesProcessed, tablesFailed int
+	var failedTables []FailedTableInfo
+	var skippedTables []string
+	var summary DiffSummary
 
 	defer func() {
 		finishedAt := time.Now()
@@ -249,10 +279,14 @@ func RepsetDiff(task *RepsetDiffCmd) (err error) {
 				"tables_total":   len(task.tableList),
 				"tables_diffed":  tablesProcessed,
 				"tables_failed":  tablesFailed,
-				"tables_skipped": tablesSkipped,
+				"tables_skipped": len(skippedTables),
 			}
 			if len(failedTables) > 0 {
-				ctx["failed_tables"] = failedTables
+				names := make([]string, len(failedTables))
+				for i, ft := range failedTables {
+					names[i] = ft.Table
+				}
+				ctx["failed_tables"] = names
 			}
 			if err != nil {
 				ctx["error"] = err.Error()
@@ -293,7 +327,7 @@ func RepsetDiff(task *RepsetDiffCmd) (err error) {
 			}
 		}
 		if skipped {
-			tablesSkipped++
+			skippedTables = append(skippedTables, tableName)
 			continue
 		}
 
@@ -320,27 +354,35 @@ func RepsetDiff(task *RepsetDiffCmd) (err error) {
 		if err := tdTask.Validate(); err != nil {
 			logger.Warn("validation for table %s failed: %v", tableName, err)
 			tablesFailed++
-			failedTables = append(failedTables, tableName)
+			failedTables = append(failedTables, FailedTableInfo{Table: tableName, Err: err})
 			continue
 		}
 
 		if err := tdTask.RunChecks(true); err != nil {
 			logger.Warn("checks for table %s failed: %v", tableName, err)
 			tablesFailed++
-			failedTables = append(failedTables, tableName)
+			failedTables = append(failedTables, FailedTableInfo{Table: tableName, Err: err})
 			continue
 		}
 		if err := tdTask.ExecuteTask(); err != nil {
 			logger.Warn("error during comparison for table %s: %v", tableName, err)
 			tablesFailed++
-			failedTables = append(failedTables, tableName)
+			failedTables = append(failedTables, FailedTableInfo{Table: tableName, Err: err})
 			continue
 		}
 
+		if len(tdTask.DiffResult.NodeDiffs) > 0 {
+			summary.DifferedTables = append(summary.DifferedTables, tableName)
+		} else {
+			summary.MatchedTables = append(summary.MatchedTables, tableName)
+		}
 		tablesProcessed++
 	}
 
-	return nil
+	summary.FailedTables = failedTables
+	summary.SkippedTables = skippedTables
+	summary.MissingTables = task.missingTables
+	return summary.PrintAndFinalize("Repset diff", "repset "+task.RepsetName)
 }
 
 func (task *RepsetDiffCmd) CloneForSchedule(ctx context.Context) *RepsetDiffCmd {