fix(repair): make pick_freshest compare commit_ts from spock metadata

pick_freshest resolved its `key` only from the stripped diff row, but
commit_ts (and other spock metadata) are moved into n1Meta/n2Meta during
diff loading. So `pick_freshest: {key: commit_ts}` found nil on both sides
and fell back to `tie` every time, never actually comparing timestamps.

Resolve the key via lookupValue, which falls back to the metadata map, so
`key: commit_ts` compares commit timestamps while ordinary data columns
(e.g. updated_at) keep working.

Tests:
- unit: commit_ts-from-meta (both directions), one-side-null (the side with
  a timestamp wins), both-null (tie), and app-column-from-row.
- integration: one-pass latest-commit-wins end-to-end (newer commit_ts side
  wins) and pick_freshest by an app-level column (subscription_date).

Author: mason-sharp

internal/consistency/repair/executor.go

@@ -723,15 +723,11 @@ func unionKeys(maps ...map[string]any) map[string]struct{} {
 }
 
 func freshestSide(key string, tie string, row planDiffRow) string {
-	val1 := row.n1Row
-	val2 := row.n2Row
-	var k1, k2 any
-	if val1 != nil {
-		k1 = val1[key]
-	}
-	if val2 != nil {
-		k2 = val2[key]
-	}
+	// Resolve the key via lookupValue so it works for both ordinary data
+	// columns (in n1Row/n2Row) and spock metadata like commit_ts, which the
+	// executor strips from the row and keeps in n1Meta/n2Meta.
+	k1, _ := lookupValue("n1", key, row)
+	k2, _ := lookupValue("n2", key, row)
 	n1Ok := k1 != nil
 	n2Ok := k2 != nil
 	if !n1Ok && n2Ok {

internal/consistency/repair/pick_freshest_test.go

@@ -0,0 +1,89 @@
+// ///////////////////////////////////////////////////////////////////////////
+//
+// # 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 repair
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// commit_ts is carried in the spock metadata map (_spock_metadata_), which the
+// executor keeps in n1Meta/n2Meta while stripping it from n1Row/n2Row. These
+// tests ensure pick_freshest's freshestSide resolves the key from metadata too,
+// so `key: commit_ts` actually compares commit timestamps instead of always
+// falling through to the tie node.
+func TestFreshestSide_CommitTsFromMeta(t *testing.T) {
+	older := "2026-01-15T12:20:07.805412-08:00"
+	newer := "2026-01-15T12:21:39.662530-08:00"
+
+	t.Run("n2 newer", func(t *testing.T) {
+		row := planDiffRow{
+			n1Row:  map[string]any{"id": 1, "v": "a"},
+			n2Row:  map[string]any{"id": 1, "v": "b"},
+			n1Meta: map[string]any{"commit_ts": older},
+			n2Meta: map[string]any{"commit_ts": newer},
+		}
+		require.Equal(t, "n2", freshestSide("commit_ts", "n1", row))
+	})
+
+	t.Run("n1 newer", func(t *testing.T) {
+		row := planDiffRow{
+			n1Row:  map[string]any{"id": 1, "v": "a"},
+			n2Row:  map[string]any{"id": 1, "v": "b"},
+			n1Meta: map[string]any{"commit_ts": newer},
+			n2Meta: map[string]any{"commit_ts": older},
+		}
+		require.Equal(t, "n1", freshestSide("commit_ts", "n2", row))
+	})
+}
+
+// When only one side has a commit_ts (e.g. the other was written with
+// track_commit_timestamp off), the side that has one wins; when neither has
+// one, the tie node is chosen.
+func TestFreshestSide_CommitTsNullHandling(t *testing.T) {
+	ts := "2026-01-15T12:21:39.662530-08:00"
+
+	t.Run("only n2 has commit_ts -> n2", func(t *testing.T) {
+		row := planDiffRow{
+			n1Meta: map[string]any{},
+			n2Meta: map[string]any{"commit_ts": ts},
+		}
+		require.Equal(t, "n2", freshestSide("commit_ts", "n1", row))
+	})
+
+	t.Run("only n1 has commit_ts -> n1", func(t *testing.T) {
+		row := planDiffRow{
+			n1Meta: map[string]any{"commit_ts": ts},
+			n2Meta: map[string]any{},
+		}
+		require.Equal(t, "n1", freshestSide("commit_ts", "n2", row))
+	})
+
+	t.Run("neither has commit_ts -> tie", func(t *testing.T) {
+		row := planDiffRow{
+			n1Meta: map[string]any{},
+			n2Meta: map[string]any{},
+		}
+		require.Equal(t, "n2", freshestSide("commit_ts", "n2", row))
+	})
+}
+
+// An application-level timestamp column lives in the row itself and must keep
+// working after the metadata fallback is added.
+func TestFreshestSide_AppColumnFromRow(t *testing.T) {
+	row := planDiffRow{
+		n1Row: map[string]any{"updated_at": "2026-01-15T12:00:00-08:00"},
+		n2Row: map[string]any{"updated_at": "2026-01-15T13:00:00-08:00"},
+	}
+	require.Equal(t, "n2", freshestSide("updated_at", "n1", row))
+}

tests/integration/advanced_repair_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/stretchr/testify/require"
 )
 
@@ -212,6 +213,147 @@ tables:
 	require.True(t, hasStaleSkipEntry(entries, 2001, "delete", "stale_delete_missing_on_n1"))
 }
 
+// TestAdvancedRepairPlan_PickFreshestByCommitTS verifies the one-pass
+// latest-commit-wins plan: pick_freshest with key: commit_ts keeps the row from
+// the side with the newer Spock commit timestamp. It also guards the regression
+// where pick_freshest read commit_ts from the (stripped) row instead of the
+// spock metadata and therefore always fell back to `tie`.
+func TestAdvancedRepairPlan_PickFreshestByCommitTS(t *testing.T) {
+	ctx := context.Background()
+	qualifiedTableName := "public.customers"
+
+	setupDivergence(t, ctx, qualifiedTableName)
+	diffFile := runTableDiff(t, qualifiedTableName, []string{serviceN1, serviceN2})
+
+	// Precondition: setupDivergence UPDATEs index 1 & 2 on n2 after the initial
+	// inserts, so n2's commit_ts for those rows is strictly newer than n1's.
+	// That ordering is what "latest commit wins" must resolve to.
+	for _, idx := range []int{1, 2} {
+		var tsN1, tsN2 time.Time
+		require.NoError(t, pgCluster.Node1Pool.QueryRow(ctx,
+			fmt.Sprintf("SELECT pg_xact_commit_timestamp(xmin) FROM %s WHERE index = %d", qualifiedTableName, idx)).Scan(&tsN1))
+		require.NoError(t, pgCluster.Node2Pool.QueryRow(ctx,
+			fmt.Sprintf("SELECT pg_xact_commit_timestamp(xmin) FROM %s WHERE index = %d", qualifiedTableName, idx)).Scan(&tsN2))
+		require.Truef(t, tsN2.After(tsN1),
+			"precondition for index %d: n2 commit_ts %s should be newer than n1 %s", idx, tsN2, tsN1)
+	}
+
+	// One-pass plan: pick_freshest by commit_ts resolves the modified rows to
+	// the newer (n2) side; apply_from rules propagate the single-sided inserts
+	// so the tables fully converge.
+	plan := `
+version: 1
+tables:
+  public.customers:
+    rules:
+      - name: latest_commit_wins
+        diff_type: [row_mismatch]
+        action:
+          type: custom
+          helpers:
+            pick_freshest: { key: commit_ts, tie: n1 }
+      - name: insert_missing_on_n2
+        diff_type: [missing_on_n2]
+        action: { type: apply_from, from: n1, mode: insert }
+      - name: insert_missing_on_n1
+        diff_type: [missing_on_n1]
+        action: { type: apply_from, from: n2, mode: insert }
+`
+	planPath := filepath.Join(t.TempDir(), "repair.yaml")
+	require.NoError(t, os.WriteFile(planPath, []byte(plan), 0o644))
+
+	task := newTestTableRepairTask("", qualifiedTableName, diffFile)
+	task.RepairPlanPath = planPath
+	require.NoError(t, task.ValidateAndPrepare())
+	require.NoError(t, task.Run(true))
+
+	assertNoTableDiff(t, qualifiedTableName)
+
+	// The newer (n2) side won for the modified rows, on BOTH nodes.
+	for _, idx := range []int{1, 2} {
+		want := fmt.Sprintf("modified.email%d@example.com", idx)
+		var e1, e2 string
+		require.NoError(t, pgCluster.Node1Pool.QueryRow(ctx,
+			fmt.Sprintf("SELECT email FROM %s WHERE index = %d", qualifiedTableName, idx)).Scan(&e1))
+		require.NoError(t, pgCluster.Node2Pool.QueryRow(ctx,
+			fmt.Sprintf("SELECT email FROM %s WHERE index = %d", qualifiedTableName, idx)).Scan(&e2))
+		require.Equalf(t, want, e1, "index %d on n1 should be the newer (n2) value", idx)
+		require.Equalf(t, want, e2, "index %d on n2 should be the newer (n2) value", idx)
+	}
+}
+
+// TestAdvancedRepairPlan_PickFreshestByAppColumn verifies pick_freshest with an
+// application-level timestamp column (subscription_date). Unlike commit_ts this
+// is ordinary row data, so it is immune to freezing and works even where
+// commit_ts would be NULL. index 1 is newer on n2, index 2 is newer on n1, so
+// the winning side differs per row.
+func TestAdvancedRepairPlan_PickFreshestByAppColumn(t *testing.T) {
+	ctx := context.Background()
+	qualifiedTableName := "public.customers"
+	env := newSpockEnv()
+	t.Cleanup(func() { resetSharedTable(t, "customers") })
+
+	// Clean slate on both nodes.
+	for _, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} {
+		env.withRepairMode(t, ctx, pool, func(conn *pgxpool.Conn) {
+			_, err := conn.Exec(ctx, fmt.Sprintf("TRUNCATE TABLE %s CASCADE", qualifiedTableName))
+			require.NoError(t, err)
+		})
+	}
+
+	// Same PKs on both nodes but differing first_name and subscription_date, so
+	// each row is a row_mismatch resolvable by freshness of subscription_date.
+	insert := func(pool *pgxpool.Pool, idx int, firstName, subDate string) {
+		env.withRepairMode(t, ctx, pool, func(conn *pgxpool.Conn) {
+			_, err := conn.Exec(ctx,
+				fmt.Sprintf("INSERT INTO %s (index, customer_id, first_name, last_name, email, subscription_date) VALUES ($1,$2,$3,$4,$5,$6)", qualifiedTableName),
+				idx, fmt.Sprintf("CUST-%d", idx), firstName, "Last", fmt.Sprintf("e%d@example.com", idx), subDate)
+			require.NoError(t, err)
+		})
+	}
+	insert(pgCluster.Node1Pool, 1, "N1", "2020-01-01 00:00:00") // index 1: n2 newer -> n2 wins
+	insert(pgCluster.Node2Pool, 1, "N2", "2021-01-01 00:00:00")
+	insert(pgCluster.Node1Pool, 2, "N1", "2023-01-01 00:00:00") // index 2: n1 newer -> n1 wins
+	insert(pgCluster.Node2Pool, 2, "N2", "2022-01-01 00:00:00")
+
+	diffFile := runTableDiff(t, qualifiedTableName, []string{serviceN1, serviceN2})
+
+	plan := `
+version: 1
+tables:
+  public.customers:
+    rules:
+      - name: latest_by_subscription_date
+        diff_type: [row_mismatch]
+        action:
+          type: custom
+          helpers:
+            pick_freshest: { key: subscription_date, tie: n1 }
+`
+	planPath := filepath.Join(t.TempDir(), "repair.yaml")
+	require.NoError(t, os.WriteFile(planPath, []byte(plan), 0o644))
+
+	task := newTestTableRepairTask("", qualifiedTableName, diffFile)
+	task.RepairPlanPath = planPath
+	require.NoError(t, task.ValidateAndPrepare())
+	require.NoError(t, task.Run(true))
+
+	assertNoTableDiff(t, qualifiedTableName)
+
+	// index 1 -> n2 (newer subscription_date) won; index 2 -> n1 won. The whole
+	// winning row is applied, so first_name identifies the side that won.
+	want := map[int]string{1: "N2", 2: "N1"}
+	for idx, wantName := range want {
+		var f1, f2 string
+		require.NoError(t, pgCluster.Node1Pool.QueryRow(ctx,
+			fmt.Sprintf("SELECT first_name FROM %s WHERE index = %d", qualifiedTableName, idx)).Scan(&f1))
+		require.NoError(t, pgCluster.Node2Pool.QueryRow(ctx,
+			fmt.Sprintf("SELECT first_name FROM %s WHERE index = %d", qualifiedTableName, idx)).Scan(&f2))
+		require.Equalf(t, wantName, f1, "index %d on n1 should be the fresher side", idx)
+		require.Equalf(t, wantName, f2, "index %d on n2 should be the fresher side", idx)
+	}
+}
+
 func listStaleSkipLogs(t *testing.T) []string {
 	t.Helper()
 	paths, err := filepath.Glob(filepath.Join("reports", "*", "stale_repair_skips_*.json"))