Add --until support to mtree, fix frozen row handling

.github/workflows/test.yml

@@ -55,5 +55,11 @@ jobs:
     - name: Run catastrophic single-node failure recovery test
       run: go test -count=1 -v ./tests/integration -run 'TestCatastrophicSingleNodeFailure'
 
+    - name: Run table-diff --until filter tests
+      run: go test -count=1 -v ./tests/integration -run 'TestTableDiffUntilFilter'
+
+    - name: Run Merkle Tree --until filter tests
+      run: go test -count=1 -v ./tests/integration -run 'TestMerkleTreeUntilFilter'
+
     - name: Run timestamp comparison tests
       run: go test -count=1 -v ./tests/integration -run 'TestCompareTimestampsExact|TestPostgreSQLMicrosecondPrecision|TestOldVsNewComparison'

db/queries/queries.go

@@ -47,6 +47,17 @@ func SanitiseIdentifier(ident string) error {
 	return nil
 }
 
+// CommitTimestampFilter returns a SQL predicate that restricts rows to those
+// committed at or before the given timestamp. Frozen rows (where
+// pg_xact_commit_timestamp returns NULL after VACUUM FREEZE) are always
+// included. Returns an empty string when t is nil.
+func CommitTimestampFilter(t *time.Time) string {
+	if t == nil {
+		return ""
+	}
+	return fmt.Sprintf("(pg_xact_commit_timestamp(xmin) IS NULL OR pg_xact_commit_timestamp(xmin) <= '%s'::timestamptz)", t.Format(time.RFC3339Nano))
+}
+
 func RenderSQL(t *template.Template, data any) (string, error) {
 	var buf bytes.Buffer
 	if err := t.Execute(&buf, data); err != nil {

db/queries/queries_test.go

@@ -15,6 +15,7 @@ import (
 	"fmt"
 	"strings"
 	"testing"
+	"time"
 )
 
 type mockRow struct {
@@ -123,6 +124,29 @@ func TestSanitiseIdentifier(t *testing.T) {
 	}
 }
 
+func TestCommitTimestampFilter(t *testing.T) {
+	t.Run("nil returns empty string", func(t *testing.T) {
+		got := CommitTimestampFilter(nil)
+		if got != "" {
+			t.Errorf("expected empty string, got %q", got)
+		}
+	})
+
+	t.Run("non-nil returns predicate including frozen row handling", func(t *testing.T) {
+		ts := time.Date(2026, 4, 15, 12, 0, 0, 0, time.UTC)
+		got := CommitTimestampFilter(&ts)
+		if !strings.Contains(got, "IS NULL") {
+			t.Errorf("expected frozen-row NULL check, got %q", got)
+		}
+		if !strings.Contains(got, "2026-04-15T12:00:00Z") {
+			t.Errorf("expected formatted timestamp, got %q", got)
+		}
+		if !strings.Contains(got, "pg_xact_commit_timestamp(xmin) <=") {
+			t.Errorf("expected upper-bound comparison, got %q", got)
+		}
+	})
+}
+
 func normalizeSQLWhitespace(s string) string {
 	s = strings.Join(strings.Fields(s), " ")
 	s = strings.ReplaceAll(s, "( ", "(")

internal/cli/cli.go

@@ -364,6 +364,11 @@ func SetupCLI() *cli.Command {
 			Usage:   "Max CPU for parallel operations",
 			Value:   0.5,
 		},
+		&cli.StringFlag{
+			Name:  "until",
+			Usage: "Optional commit timestamp upper bound (RFC3339) for rows to include",
+			Value: "",
+		},
 		&cli.StringFlag{
 			Name:    "output",
 			Aliases: []string{"o"},
@@ -1145,6 +1150,7 @@ func MtreeDiffCLI(cmd *cli.Command) error {
 	task.MaxCpuRatio = cmd.Float64("max-cpu-ratio")
 	task.Output = cmd.String("output")
 	task.NoCDC = cmd.Bool("skip-cdc")
+	task.Until = cmd.String("until")
 	task.Mode = "diff"
 	task.Ctx = context.Background()
 

internal/consistency/diff/table_diff.go

@@ -238,7 +238,7 @@ func (t *TableDiffTask) resolveAgainstOrigin() error {
 
 func (t *TableDiffTask) buildEffectiveFilter() (string, error) {
 	if t.untilTime == nil && strings.TrimSpace(t.Until) != "" {
-		parsed, err := time.Parse(time.RFC3339, strings.TrimSpace(t.Until))
+		parsed, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(t.Until))
 		if err != nil {
 			return "", fmt.Errorf("invalid value for --until (expected RFC3339 timestamp): %w", err)
 		}
@@ -259,8 +259,8 @@ func (t *TableDiffTask) buildEffectiveFilter() (string, error) {
 		parts = append(parts, fmt.Sprintf("(to_json(spock.xact_commit_timestamp_origin(xmin))->>'roident' = '%d')", nodeID))
 	}
 
-	if t.untilTime != nil {
-		parts = append(parts, fmt.Sprintf("(pg_xact_commit_timestamp(xmin) <= '%s'::timestamptz)", t.untilTime.Format(time.RFC3339)))
+	if f := queries.CommitTimestampFilter(t.untilTime); f != "" {
+		parts = append(parts, f)
 	}
 
 	if len(parts) == 0 {
@@ -761,11 +761,13 @@ func (t *TableDiffTask) Validate() error {
 	}
 
 	if trimmed := strings.TrimSpace(t.Until); trimmed != "" {
-		parsed, err := time.Parse(time.RFC3339, trimmed)
+		parsed, err := time.Parse(time.RFC3339Nano, trimmed)
 		if err != nil {
 			return fmt.Errorf("invalid value for --until (expected RFC3339 timestamp): %w", err)
 		}
 		t.untilTime = &parsed
+	} else {
+		t.untilTime = nil
 	}
 
 	nodeList, err := utils.ParseNodes(t.Nodes)

internal/consistency/mtree/merkle.go

@@ -86,6 +86,9 @@ type MerkleTreeTask struct {
 	Mode              string
 	NoCDC             bool
 	SkipDBUpdate      bool
+	Until             string
+
+	untilTime *time.Time
 
 	TaskStore     *taskstore.Store
 	TaskStorePath string
@@ -373,6 +376,10 @@ func (m *MerkleTreeTask) processWorkItem(work CompareRangesWorkItem, pool1, pool
 		whereClause = "TRUE"
 	}
 
+	if f := queries.CommitTimestampFilter(m.untilTime); f != "" {
+		whereClause = fmt.Sprintf("(%s) AND %s", whereClause, f)
+	}
+
 	rowHashQuery, orderByStr := buildRowHashQuery(m.Schema, m.Table, m.Key, m.Cols, whereClause, m.ColTypes["_ref"])
 	logger.Debug("Row-hash Query: %s, Args: %v", rowHashQuery, args)
 
@@ -1121,6 +1128,16 @@ func (m *MerkleTreeTask) Validate() error {
 		return fmt.Errorf("invalid value range for max_cpu_ratio")
 	}
 
+	if trimmed := strings.TrimSpace(m.Until); trimmed != "" {
+		parsed, err := time.Parse(time.RFC3339Nano, trimmed)
+		if err != nil {
+			return fmt.Errorf("invalid value for --until (expected RFC3339 timestamp): %w", err)
+		}
+		m.untilTime = &parsed
+	} else {
+		m.untilTime = nil
+	}
+
 	if m.RangesFile != "" {
 		if _, err := os.Stat(m.RangesFile); os.IsNotExist(err) {
 			return fmt.Errorf("file %s does not exist", m.RangesFile)

tests/integration/merkle_tree_test.go

@@ -55,6 +55,123 @@ func TestMerkleTreeCompositePK(t *testing.T) {
 	runMerkleTreeTests(t, tableName)
 }
 
+func TestMerkleTreeUntilFilter(t *testing.T) {
+	ctx := context.Background()
+	// Use a dedicated small table so we can INSERT within the block range.
+	// The customers table has dense indices 1-10000; inserting outside that
+	// range falls beyond the last block's upper bound and is invisible to
+	// processWorkItem's range-bounded query.
+	untilTable := "mtree_until_test"
+	qualifiedTable := fmt.Sprintf("%s.%s", testSchema, untilTable)
+	nodes := []string{serviceN1, serviceN2}
+
+	// Create the table on both nodes and add to the default repset.
+	for i, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} {
+		nodeName := pgCluster.ClusterNodes[i]["Name"].(string)
+		_, err := pool.Exec(ctx, fmt.Sprintf(
+			"CREATE TABLE IF NOT EXISTS %s (id INT PRIMARY KEY, payload TEXT)", qualifiedTable))
+		require.NoError(t, err, "create table on %s", nodeName)
+		_, err = pool.Exec(ctx, fmt.Sprintf(
+			"SELECT spock.repset_add_table('default', '%s')", qualifiedTable))
+		require.NoError(t, err, "add to repset on %s", nodeName)
+	}
+	t.Cleanup(func() {
+		for _, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} {
+			pool.Exec(ctx, fmt.Sprintf("DROP TABLE IF EXISTS %s CASCADE", qualifiedTable))
+		}
+	})
+
+	// Seed identical rows on both nodes with a gap at id=5 (inside one block).
+	// The gap lets us INSERT id=5 later as a genuinely new row (not an update
+	// of an existing frozen tuple) so --until can cleanly exclude it.
+	for _, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} {
+		tx, err := pool.Begin(ctx)
+		require.NoError(t, err)
+		_, err = tx.Exec(ctx, "SELECT spock.repair_mode(true)")
+		require.NoError(t, err)
+		for id := 1; id <= 10; id++ {
+			if id == 5 {
+				continue // leave a gap
+			}
+			_, err = tx.Exec(ctx, fmt.Sprintf(
+				"INSERT INTO %s (id, payload) VALUES ($1, $2) ON CONFLICT DO NOTHING", qualifiedTable),
+				id, fmt.Sprintf("row-%d", id))
+			require.NoError(t, err)
+		}
+		_, err = tx.Exec(ctx, "SELECT spock.repair_mode(false)")
+		require.NoError(t, err)
+		require.NoError(t, tx.Commit(ctx))
+	}
+
+	// VACUUM FREEZE so baseline rows have NULL commit timestamps.
+	for _, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} {
+		_, err := pool.Exec(ctx, fmt.Sprintf("VACUUM FREEZE %s", qualifiedTable))
+		require.NoError(t, err)
+	}
+
+	// Init + build the merkle tree.
+	mtreeTask := newTestMerkleTreeTask(t, qualifiedTable, nodes)
+	mtreeTask.BlockSize = 1000 // single block covers all rows
+	require.NoError(t, mtreeTask.RunChecks(false))
+	require.NoError(t, mtreeTask.MtreeInit())
+	t.Cleanup(func() {
+		mtreeTask.MtreeTeardown()
+	})
+	require.NoError(t, mtreeTask.BuildMtree())
+
+	// Capture fence while both nodes are identical and frozen.
+	var fence time.Time
+	err := pgCluster.Node1Pool.QueryRow(ctx, "SELECT now()").Scan(&fence)
+	require.NoError(t, err)
+	time.Sleep(100 * time.Millisecond)
+
+	// INSERT id=5 on node1 only — fills the gap, within block range [1, 10].
+	// This is a new row (not an update), so the frozen baseline on both nodes
+	// is untouched. Node2 simply doesn't have id=5.
+	tx, err := pgCluster.Node1Pool.Begin(ctx)
+	require.NoError(t, err)
+	_, err = tx.Exec(ctx, "SELECT spock.repair_mode(true)")
+	require.NoError(t, err)
+	_, err = tx.Exec(ctx, fmt.Sprintf(
+		"INSERT INTO %s (id, payload) VALUES (5, 'divergent')", qualifiedTable))
+	require.NoError(t, err)
+	_, err = tx.Exec(ctx, "SELECT spock.repair_mode(false)")
+	require.NoError(t, err)
+	require.NoError(t, tx.Commit(ctx))
+
+	// Diff WITHOUT --until first: the divergent write should be visible.
+	// Running this first ensures UpdateMtree processes any CDC events.
+	taskWithout := newTestMerkleTreeTask(t, qualifiedTable, nodes)
+	taskWithout.Mode = "diff"
+	taskWithout.Output = "json"
+	taskWithout.BlockSize = 1000
+	require.NoError(t, taskWithout.RunChecks(false))
+	require.NoError(t, taskWithout.DiffMtree())
+
+	totalWithout := 0
+	for _, count := range taskWithout.DiffResult.Summary.DiffRowsCount {
+		totalWithout += count
+	}
+	require.Greater(t, totalWithout, 0, "without --until, expected at least 1 diff row")
+
+	// Diff WITH --until set to the fence: the divergent row's xmin is
+	// post-fence, so it is excluded. The frozen baseline (NULL timestamps)
+	// is included via IS NULL on both nodes and matches.
+	taskWithUntil := newTestMerkleTreeTask(t, qualifiedTable, nodes)
+	taskWithUntil.Until = fence.Format(time.RFC3339Nano)
+	taskWithUntil.Mode = "diff"
+	taskWithUntil.Output = "json"
+	taskWithUntil.BlockSize = 1000
+	require.NoError(t, taskWithUntil.RunChecks(false))
+	require.NoError(t, taskWithUntil.DiffMtree())
+
+	totalWithUntil := 0
+	for _, count := range taskWithUntil.DiffResult.Summary.DiffRowsCount {
+		totalWithUntil += count
+	}
+	require.Equal(t, 0, totalWithUntil, "with --until on frozen baseline, expected 0 diff rows")
+}
+
 func runMerkleTreeTests(t *testing.T, tableName string) {
 	if tableName == "customers" {
 		resetSharedTable(t, "customers")

tests/integration/table_diff_test.go

@@ -58,6 +58,74 @@ func newTestTableDiffTask(
 	return task
 }
 
+func TestTableDiffUntilFilter(t *testing.T) {
+	ctx := context.Background()
+	tableName := "customers"
+	qualifiedTableName := fmt.Sprintf("%s.%s", testSchema, tableName)
+	nodesToCompare := []string{serviceN1, serviceN2}
+
+	resetSharedTable(t, tableName)
+
+	// VACUUM FREEZE the baseline so all existing rows have NULL commit
+	// timestamps. This must happen before the fence so that the post-fence
+	// divergent row keeps its real timestamp and is correctly excluded.
+	for _, pool := range []*pgxpool.Pool{pgCluster.Node1Pool, pgCluster.Node2Pool} {
+		_, err := pool.Exec(ctx, fmt.Sprintf("VACUUM FREEZE %s", qualifiedTableName))
+		require.NoError(t, err, "VACUUM FREEZE should succeed")
+	}
+
+	// Capture a fence timestamp while both nodes are identical and frozen.
+	var fence time.Time
+	err := pgCluster.Node1Pool.QueryRow(ctx, "SELECT now()").Scan(&fence)
+	require.NoError(t, err)
+
+	// Wait briefly so that subsequent writes are strictly after the fence.
+	time.Sleep(100 * time.Millisecond)
+
+	// Create a local-only divergence on node1 by inserting a new row
+	// (repair_mode prevents replication). We INSERT rather than UPDATE so the
+	// existing frozen rows are untouched on both nodes — only the new row has
+	// a post-fence xmin and is excluded by --until.
+	tx, err := pgCluster.Node1Pool.Begin(ctx)
+	require.NoError(t, err)
+	_, err = tx.Exec(ctx, "SELECT spock.repair_mode(true)")
+	require.NoError(t, err)
+	_, err = tx.Exec(ctx, fmt.Sprintf("INSERT INTO %s (index, email) VALUES (99999, 'until_test@example.com')", qualifiedTableName))
+	require.NoError(t, err)
+	_, err = tx.Exec(ctx, "SELECT spock.repair_mode(false)")
+	require.NoError(t, err)
+	require.NoError(t, tx.Commit(ctx))
+
+	t.Cleanup(func() {
+		pgCluster.Node1Pool.Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE index = 99999", qualifiedTableName))
+		resetSharedTable(t, tableName)
+	})
+
+	// Diff WITH --until set to the fence: frozen baseline rows (NULL timestamps)
+	// are included via IS NULL, the post-fence divergent row is excluded.
+	taskWithUntil := newTestTableDiffTask(t, qualifiedTableName, nodesToCompare)
+	taskWithUntil.Until = fence.Format(time.RFC3339Nano)
+	require.NoError(t, taskWithUntil.RunChecks(false))
+	require.NoError(t, taskWithUntil.ExecuteTask())
+
+	totalWithUntil := 0
+	for _, count := range taskWithUntil.DiffResult.Summary.DiffRowsCount {
+		totalWithUntil += count
+	}
+	require.Equal(t, 0, totalWithUntil, "with --until on frozen baseline, expected 0 diff rows")
+
+	// Diff WITHOUT --until: the divergent write should be visible.
+	taskWithout := newTestTableDiffTask(t, qualifiedTableName, nodesToCompare)
+	require.NoError(t, taskWithout.RunChecks(false))
+	require.NoError(t, taskWithout.ExecuteTask())
+
+	totalWithout := 0
+	for _, count := range taskWithout.DiffResult.Summary.DiffRowsCount {
+		totalWithout += count
+	}
+	require.Greater(t, totalWithout, 0, "without --until, expected at least 1 diff row")
+}
+
 func TestTableDiffSimplePK(t *testing.T) {
 	t.Run("Customers", func(t *testing.T) {
 		runCustomerTableDiffTests(t)