fix: per-node origin name translation in table diff and merkle tree

NodeOriginNames was loaded from one random pool and used to translate
roidents from all nodes. In native PG, roidents are node-local, so
the same roident maps to different subscription names on different
nodes. This caused intermittent preserve-origin repair failures when
the diff happened to load origin names from the wrong node.

Change NodeOriginNames to map[string]map[string]string (nodeName →
roident → name). Load from every pool. Pass the originating node name
into withSpockMetadata so each row's roident is resolved correctly.

Author: mason-sharp

internal/consistency/diff/table_diff.go

@@ -102,7 +102,7 @@ type TableDiffTask struct {
 	blockHashSQLCache map[hashBoundsKey]string
 	blockHashSQLMu    sync.Mutex
 
-	NodeOriginNames map[string]string
+	NodeOriginNames map[string]map[string]string
 
 	CompareUnitSize int
 	MaxDiffRows     int64
@@ -204,53 +204,66 @@ func (t *TableDiffTask) loadNodeOriginNames() error {
 		return nil
 	}
 
-	var firstPool *pgxpool.Pool
-	for _, pool := range t.Pools {
-		firstPool = pool
-		break
-	}
+	t.NodeOriginNames = make(map[string]map[string]string)
 
-	if firstPool == nil {
-		t.NodeOriginNames = make(map[string]string)
-		return fmt.Errorf("no connection pool available to load node origin names")
+	var lastErr error
+	for name, pool := range t.Pools {
+		names, err := queries.GetNodeOriginNames(t.Ctx, pool)
+		if err != nil {
+			lastErr = err
+			continue
+		}
+		t.NodeOriginNames[name] = names
 	}
 
-	names, err := queries.GetNodeOriginNames(t.Ctx, firstPool)
-	if err != nil {
-		t.NodeOriginNames = make(map[string]string)
-		return err
+	if len(t.NodeOriginNames) == 0 && lastErr != nil {
+		return lastErr
 	}
-
-	t.NodeOriginNames = names
 	return nil
 }
 
+// flatNodeOriginNames merges all per-node origin maps into a single map for
+// lookup purposes (e.g. resolving --against-origin). If the same roident
+// appears on multiple nodes, the last one wins — this is acceptable because
+// resolveAgainstOrigin is only used with spock (where roidents are global)
+// or for user-facing name resolution where any match suffices.
+func (t *TableDiffTask) flatNodeOriginNames() map[string]string {
+	flat := make(map[string]string)
+	for _, nodeMap := range t.NodeOriginNames {
+		for id, name := range nodeMap {
+			flat[id] = name
+		}
+	}
+	return flat
+}
+
 func (t *TableDiffTask) resolveAgainstOrigin() error {
 	if strings.TrimSpace(t.AgainstOrigin) == "" {
 		return nil
 	}
-	if len(t.NodeOriginNames) == 0 {
+	flat := t.flatNodeOriginNames()
+	if len(flat) == 0 {
 		return fmt.Errorf("unable to resolve --against-origin: no node origin names available")
 	}
 
 	orig := strings.TrimSpace(t.AgainstOrigin)
 	// direct match on origin id
-	if _, ok := t.NodeOriginNames[orig]; ok {
+	if _, ok := flat[orig]; ok {
 		t.resolvedAgainstOrigin = orig
 		return nil
 	}
 
 	// match on origin name
-	for id, name := range t.NodeOriginNames {
+	for id, name := range flat {
 		if name == orig {
 			t.resolvedAgainstOrigin = id
 			return nil
 		}
 	}
 
 	// build a list of available origins for the error message
-	available := make([]string, 0, len(t.NodeOriginNames))
-	for id, name := range t.NodeOriginNames {
+	available := make([]string, 0, len(flat))
+	for id, name := range flat {
 		if id != name {
 			available = append(available, fmt.Sprintf("%s (%s)", id, name))
 		} else {
@@ -294,8 +307,8 @@ func (t *TableDiffTask) buildEffectiveFilter() (string, error) {
 	return strings.Join(parts, " AND "), nil
 }
 
-func (t *TableDiffTask) withSpockMetadata(row map[string]any) map[string]any {
-	row["node_origin"] = utils.TranslateNodeOrigin(row["node_origin"], t.NodeOriginNames)
+func (t *TableDiffTask) withSpockMetadata(row map[string]any, nodeName string) map[string]any {
+	row["node_origin"] = utils.TranslateNodeOrigin(row["node_origin"], t.NodeOriginNames[nodeName])
 	return utils.AddSpockMetadata(row)
 }
 
@@ -1375,8 +1388,9 @@ func (t *TableDiffTask) ExecuteTask() (err error) {
 			DiffRowsCount:     make(map[string]int),
 			AgainstOrigin:     t.AgainstOrigin,
 			AgainstOriginResolved: func() string {
-				if t.resolvedAgainstOrigin != "" && t.NodeOriginNames != nil {
-					if name, ok := t.NodeOriginNames[t.resolvedAgainstOrigin]; ok {
+				if t.resolvedAgainstOrigin != "" {
+					flat := t.flatNodeOriginNames()
+					if name, ok := flat[t.resolvedAgainstOrigin]; ok {
 						return name
 					}
 				}
@@ -1995,7 +2009,7 @@ func (t *TableDiffTask) recursiveDiff(
 					break
 				}
 				rowAsMap := utils.OrderedMapToMap(row)
-				rowWithMeta := t.withSpockMetadata(rowAsMap)
+				rowWithMeta := t.withSpockMetadata(rowAsMap, node1Name)
 				rowAsOrderedMap := utils.MapToOrderedMap(rowWithMeta, t.Cols)
 				t.DiffResult.NodeDiffs[pairKey].Rows[node1Name] = append(t.DiffResult.NodeDiffs[pairKey].Rows[node1Name], rowAsOrderedMap)
 				currentDiffRowsForPair++
@@ -2012,7 +2026,7 @@ func (t *TableDiffTask) recursiveDiff(
 						break
 					}
 					rowAsMap := utils.OrderedMapToMap(row)
-					rowWithMeta := t.withSpockMetadata(rowAsMap)
+					rowWithMeta := t.withSpockMetadata(rowAsMap, node2Name)
 					rowAsOrderedMap := utils.MapToOrderedMap(rowWithMeta, t.Cols)
 					t.DiffResult.NodeDiffs[pairKey].Rows[node2Name] = append(t.DiffResult.NodeDiffs[pairKey].Rows[node2Name], rowAsOrderedMap)
 					currentDiffRowsForPair++
@@ -2030,12 +2044,12 @@ func (t *TableDiffTask) recursiveDiff(
 						break
 					}
 					node1DataAsMap := utils.OrderedMapToMap(modRow.Node1Data)
-					node1DataWithMeta := t.withSpockMetadata(node1DataAsMap)
+					node1DataWithMeta := t.withSpockMetadata(node1DataAsMap, node1Name)
 					node1DataAsOrderedMap := utils.MapToOrderedMap(node1DataWithMeta, t.Cols)
 					t.DiffResult.NodeDiffs[pairKey].Rows[node1Name] = append(t.DiffResult.NodeDiffs[pairKey].Rows[node1Name], node1DataAsOrderedMap)
 
 					node2DataAsMap := utils.OrderedMapToMap(modRow.Node2Data)
-					node2DataWithMeta := t.withSpockMetadata(node2DataAsMap)
+					node2DataWithMeta := t.withSpockMetadata(node2DataAsMap, node2Name)
 					node2DataAsOrderedMap := utils.MapToOrderedMap(node2DataWithMeta, t.Cols)
 					t.DiffResult.NodeDiffs[pairKey].Rows[node2Name] = append(t.DiffResult.NodeDiffs[pairKey].Rows[node2Name], node2DataAsOrderedMap)
 					currentDiffRowsForPair++

internal/consistency/diff/table_diff_origin_test.go

@@ -40,7 +40,7 @@ func TestResolveAgainstOrigin_NoNodeOriginNames(t *testing.T) {
 func TestResolveAgainstOrigin_MatchByID(t *testing.T) {
 	task := &TableDiffTask{
 		AgainstOrigin:   "3",
-		NodeOriginNames: map[string]string{"3": "n1", "4": "n2"},
+		NodeOriginNames: map[string]map[string]string{"node1": {"3": "n1", "4": "n2"}},
 	}
 	if err := task.resolveAgainstOrigin(); err != nil {
 		t.Fatalf("unexpected error: %v", err)
@@ -53,7 +53,7 @@ func TestResolveAgainstOrigin_MatchByID(t *testing.T) {
 func TestResolveAgainstOrigin_MatchByName_SpockNodeName(t *testing.T) {
 	task := &TableDiffTask{
 		AgainstOrigin:   "n1",
-		NodeOriginNames: map[string]string{"3": "n1", "4": "n2"},
+		NodeOriginNames: map[string]map[string]string{"node1": {"3": "n1", "4": "n2"}},
 	}
 	if err := task.resolveAgainstOrigin(); err != nil {
 		t.Fatalf("unexpected error: %v", err)
@@ -64,10 +64,12 @@ func TestResolveAgainstOrigin_MatchByName_SpockNodeName(t *testing.T) {
 }
 
 func TestResolveAgainstOrigin_MatchByName_SubscriptionName(t *testing.T) {
-	// Native PG: NodeOriginNames maps roident -> subscription name
+	// Native PG: NodeOriginNames maps roident -> subscription name, per node
 	task := &TableDiffTask{
-		AgainstOrigin:   "sub_n1_to_n2",
-		NodeOriginNames: map[string]string{"5": "sub_n1_to_n2", "6": "sub_n3_to_n2"},
+		AgainstOrigin: "sub_n1_to_n2",
+		NodeOriginNames: map[string]map[string]string{
+			"node1": {"5": "sub_n1_to_n2", "6": "sub_n3_to_n2"},
+		},
 	}
 	if err := task.resolveAgainstOrigin(); err != nil {
 		t.Fatalf("unexpected error: %v", err)
@@ -80,7 +82,7 @@ func TestResolveAgainstOrigin_MatchByName_SubscriptionName(t *testing.T) {
 func TestResolveAgainstOrigin_NoMatch(t *testing.T) {
 	task := &TableDiffTask{
 		AgainstOrigin:   "nonexistent",
-		NodeOriginNames: map[string]string{"3": "n1", "4": "n2"},
+		NodeOriginNames: map[string]map[string]string{"node1": {"3": "n1", "4": "n2"}},
 	}
 	err := task.resolveAgainstOrigin()
 	if err == nil {
@@ -133,3 +135,45 @@ func TestBuildEffectiveFilter_Empty(t *testing.T) {
 		t.Fatalf("expected empty filter, got: %s", filter)
 	}
 }
+
+func TestWithSpockMetadata_PerNodeTranslation(t *testing.T) {
+	// Simulate native PG: same roident "1" maps to different names on different nodes
+	task := &TableDiffTask{
+		NodeOriginNames: map[string]map[string]string{
+			"n1": {"1": "sub_n3_to_n1"},
+			"n2": {"1": "sub_n3_to_n2"},
+		},
+	}
+
+	row1 := map[string]any{"node_origin": "1", "id": 1}
+	result1 := task.withSpockMetadata(row1, "n1")
+	meta1 := result1["_spock_metadata_"].(map[string]any)
+	if meta1["node_origin"] != "sub_n3_to_n1" {
+		t.Fatalf("n1 row: expected origin sub_n3_to_n1, got %v", meta1["node_origin"])
+	}
+
+	row2 := map[string]any{"node_origin": "1", "id": 2}
+	result2 := task.withSpockMetadata(row2, "n2")
+	meta2 := result2["_spock_metadata_"].(map[string]any)
+	if meta2["node_origin"] != "sub_n3_to_n2" {
+		t.Fatalf("n2 row: expected origin sub_n3_to_n2, got %v", meta2["node_origin"])
+	}
+}
+
+func TestFlatNodeOriginNames_MergesAllNodes(t *testing.T) {
+	task := &TableDiffTask{
+		NodeOriginNames: map[string]map[string]string{
+			"n1": {"1": "sub_n3_to_n1"},
+			"n2": {"1": "sub_n3_to_n2", "2": "sub_n4_to_n2"},
+		},
+	}
+	flat := task.flatNodeOriginNames()
+	// roident "2" should always be present
+	if flat["2"] != "sub_n4_to_n2" {
+		t.Fatalf("expected flat[2]=sub_n4_to_n2, got %q", flat["2"])
+	}
+	// roident "1" will be one of the two — either is valid for flattened lookup
+	if flat["1"] != "sub_n3_to_n1" && flat["1"] != "sub_n3_to_n2" {
+		t.Fatalf("expected flat[1] to be one of the subscription names, got %q", flat["1"])
+	}
+}

internal/consistency/diff/table_rerun.go

@@ -374,12 +374,12 @@ func (t *TableDiffTask) reCompareDiffs(fetchedRowsByNode map[string]map[string]t
 				persistentDiffCount++
 				if nowOnNode1 {
 					rowAsMap := utils.OrderedMapToMap(newRow1)
-					rowWithMeta := t.withSpockMetadata(rowAsMap)
+					rowWithMeta := t.withSpockMetadata(rowAsMap, node1)
 					newDiffsForPair.Rows[node1] = append(newDiffsForPair.Rows[node1], utils.MapToOrderedMap(rowWithMeta, t.Cols))
 				}
 				if nowOnNode2 {
 					rowAsMap := utils.OrderedMapToMap(newRow2)
-					rowWithMeta := t.withSpockMetadata(rowAsMap)
+					rowWithMeta := t.withSpockMetadata(rowAsMap, node2)
 					newDiffsForPair.Rows[node2] = append(newDiffsForPair.Rows[node2], utils.MapToOrderedMap(rowWithMeta, t.Cols))
 				}
 			}

internal/consistency/mtree/merkle.go

@@ -99,7 +99,7 @@ type MerkleTreeTask struct {
 	diffMutex   sync.Mutex
 	diffRowKeySets map[string]map[string]map[string]struct{}
 	StartTime      time.Time
-	NodeOriginNames map[string]string
+	NodeOriginNames map[string]map[string]string
 
 	Ctx context.Context
 }
@@ -510,8 +510,14 @@ func (m *MerkleTreeTask) loadNodeOriginNames() error {
 		return nil
 	}
 
+	m.NodeOriginNames = make(map[string]map[string]string)
+
 	var lastErr error
 	for _, nodeInfo := range m.ClusterNodes {
+		nodeName, _ := nodeInfo["Name"].(string)
+		if nodeName == "" {
+			continue
+		}
 		pool, err := auth.GetClusterNodeConnection(m.Ctx, nodeInfo, m.connOpts())
 		if err != nil {
 			lastErr = err
@@ -523,15 +529,13 @@ func (m *MerkleTreeTask) loadNodeOriginNames() error {
 			lastErr = err
 			continue
 		}
-		m.NodeOriginNames = names
-		return nil
+		m.NodeOriginNames[nodeName] = names
 	}
 
-	m.NodeOriginNames = make(map[string]string)
-	if lastErr != nil {
+	if len(m.NodeOriginNames) == 0 && lastErr != nil {
 		return lastErr
 	}
-	return fmt.Errorf("no nodes available to load node origin names")
+	return nil
 }
 
 func (m *MerkleTreeTask) appendDiffs(nodePairKey string, work CompareRangesWorkItem, pr1, pr2 []types.OrderedMap) error {
@@ -603,7 +607,7 @@ func (m *MerkleTreeTask) addRowToDiff(nodePairKey, nodeName string, row types.Or
 	}
 
 	rowMap := utils.OrderedMapToMap(row)
-	rowMap["node_origin"] = utils.TranslateNodeOrigin(rowMap["node_origin"], m.NodeOriginNames)
+	rowMap["node_origin"] = utils.TranslateNodeOrigin(rowMap["node_origin"], m.NodeOriginNames[nodeName])
 	rowWithMeta := utils.AddSpockMetadata(rowMap)
 	orderedRow := utils.MapToOrderedMap(rowWithMeta, m.Cols)