Add hash version tracking for automatic merkle tree migration

Store a hash_version in ace_mtree_metadata so that mtree-update can
detect trees built with an older hash algorithm and automatically
recompute all leaf hashes. On first mtree-update after upgrade, the
version mismatch marks all leaves dirty, triggering a full recompute
within the normal update flow. This is crash-safe (version is updated
in the same transaction as the hashes) and idempotent.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: mason-sharp

db/queries/queries.go

@@ -26,6 +26,11 @@ import (
 	"github.com/pgedge/ace/pkg/types"
 )
 
+// CurrentHashVersion is the version of the hash algorithm used by this build.
+// Increment when the SQL hash computation changes (e.g., switching from
+// whole-row ::text to per-column concat_ws with trim_scale).
+const CurrentHashVersion = 2
+
 type DBQuerier interface {
 	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
 	Query(context.Context, string, ...interface{}) (pgx.Rows, error)
@@ -1165,13 +1170,13 @@ func GetPkeyType(ctx context.Context, db DBQuerier, schema, table, pkey string)
 	return pkeyType, nil
 }
 
-func UpdateMetadata(ctx context.Context, db DBQuerier, schema, table string, totalRows int64, blockSize, numBlocks int, isComposite bool) error {
+func UpdateMetadata(ctx context.Context, db DBQuerier, schema, table string, totalRows int64, blockSize, numBlocks int, isComposite bool, hashVersion int) error {
 	sql, err := RenderSQL(SQLTemplates.UpdateMetadata, nil)
 	if err != nil {
 		return fmt.Errorf("failed to render UpdateMetadata SQL: %w", err)
 	}
 
-	_, err = db.Exec(ctx, sql, schema, table, totalRows, blockSize, numBlocks, isComposite)
+	_, err = db.Exec(ctx, sql, schema, table, totalRows, blockSize, numBlocks, isComposite, hashVersion)
 	if err != nil {
 		return fmt.Errorf("query to update metadata for '%s.%s' failed: %w", schema, table, err)
 	}
@@ -2326,6 +2331,63 @@ func GetBlockSizeFromMetadata(ctx context.Context, db DBQuerier, schema, table s
 	return blockSize, nil
 }
 
+func EnsureHashVersionColumn(ctx context.Context, db DBQuerier) error {
+	sql, err := RenderSQL(SQLTemplates.EnsureHashVersionColumn, nil)
+	if err != nil {
+		return fmt.Errorf("failed to render EnsureHashVersionColumn SQL: %w", err)
+	}
+
+	_, err = db.Exec(ctx, sql)
+	if err != nil {
+		return fmt.Errorf("failed to add hash_version column to metadata table: %w", err)
+	}
+	return nil
+}
+
+func GetHashVersion(ctx context.Context, db DBQuerier, schema, table string) (int, error) {
+	sql, err := RenderSQL(SQLTemplates.GetHashVersion, nil)
+	if err != nil {
+		return 0, fmt.Errorf("failed to render GetHashVersion SQL: %w", err)
+	}
+
+	var version int
+	err = db.QueryRow(ctx, sql, schema, table).Scan(&version)
+	if err != nil {
+		return 0, fmt.Errorf("query to get hash version for '%s.%s' failed: %w", schema, table, err)
+	}
+	return version, nil
+}
+
+func MarkAllLeavesDirty(ctx context.Context, db DBQuerier, mtreeTable string) (int64, error) {
+	data := map[string]interface{}{
+		"MtreeTable": mtreeTable,
+	}
+
+	sql, err := RenderSQL(SQLTemplates.MarkAllLeavesDirty, data)
+	if err != nil {
+		return 0, fmt.Errorf("failed to render MarkAllLeavesDirty SQL: %w", err)
+	}
+
+	tag, err := db.Exec(ctx, sql)
+	if err != nil {
+		return 0, fmt.Errorf("query to mark all leaves dirty for '%s' failed: %w", mtreeTable, err)
+	}
+	return tag.RowsAffected(), nil
+}
+
+func UpdateHashVersion(ctx context.Context, db DBQuerier, schema, table string, version int) error {
+	sql, err := RenderSQL(SQLTemplates.UpdateHashVersion, nil)
+	if err != nil {
+		return fmt.Errorf("failed to render UpdateHashVersion SQL: %w", err)
+	}
+
+	_, err = db.Exec(ctx, sql, version, schema, table)
+	if err != nil {
+		return fmt.Errorf("query to update hash version for '%s.%s' failed: %w", schema, table, err)
+	}
+	return nil
+}
+
 func GetMaxNodeLevel(ctx context.Context, db DBQuerier, mtreeTable string) (int, error) {
 	data := map[string]interface{}{
 		"MtreeTable": mtreeTable,

db/queries/templates.go

@@ -120,6 +120,10 @@ type Templates struct {
 	RemoveTableFromCDCMetadata       *template.Template
 	GetSpockOriginLSNForNode         *template.Template
 	GetSpockSlotLSNForNode           *template.Template
+	EnsureHashVersionColumn          *template.Template
+	GetHashVersion                   *template.Template
+	MarkAllLeavesDirty               *template.Template
+	UpdateHashVersion                *template.Template
 }
 
 var SQLTemplates = Templates{
@@ -132,6 +136,7 @@ var SQLTemplates = Templates{
 			block_size int,
 			num_blocks int,
 			is_composite boolean NOT NULL DEFAULT false,
+			hash_version int NOT NULL DEFAULT 2,
 			last_updated timestamptz,
 			PRIMARY KEY (schema_name, table_name)
 		)`),
@@ -839,6 +844,7 @@ var SQLTemplates = Templates{
 				block_size,
 				num_blocks,
 				is_composite,
+				hash_version,
 				last_updated
 			)
 		VALUES
@@ -849,6 +855,7 @@ var SQLTemplates = Templates{
 				$4,
 				$5,
 				$6,
+				$7,
 				current_timestamp
 			)
 		ON CONFLICT (schema_name, table_name) DO
@@ -858,6 +865,7 @@ var SQLTemplates = Templates{
 			block_size = EXCLUDED.block_size,
 			num_blocks = EXCLUDED.num_blocks,
 			is_composite = EXCLUDED.is_composite,
+			hash_version = EXCLUDED.hash_version,
 			last_updated = EXCLUDED.last_updated
 	`)),
 	DeleteMetadata: template.Must(template.New("deleteMetadata").Parse(`
@@ -1543,4 +1551,25 @@ var SQLTemplates = Templates{
 		ORDER BY rs.confirmed_flush_lsn DESC
 		LIMIT 1
 	`)),
+	EnsureHashVersionColumn: template.Must(template.New("ensureHashVersionColumn").Parse(`
+		ALTER TABLE spock.ace_mtree_metadata
+		ADD COLUMN IF NOT EXISTS hash_version int NOT NULL DEFAULT 1
+	`)),
+	GetHashVersion: template.Must(template.New("getHashVersion").Parse(`
+		SELECT COALESCE(
+			(SELECT hash_version FROM spock.ace_mtree_metadata
+			 WHERE schema_name = $1 AND table_name = $2),
+			1
+		)
+	`)),
+	MarkAllLeavesDirty: template.Must(template.New("markAllLeavesDirty").Parse(`
+		UPDATE {{.MtreeTable}}
+		SET dirty = true
+		WHERE node_level = 0
+	`)),
+	UpdateHashVersion: template.Must(template.New("updateHashVersion").Parse(`
+		UPDATE spock.ace_mtree_metadata
+		SET hash_version = $1, last_updated = current_timestamp
+		WHERE schema_name = $2 AND table_name = $3
+	`)),
 }

internal/consistency/mtree/merkle.go

@@ -1507,6 +1507,12 @@ func (m *MerkleTreeTask) UpdateMtree(skipAllChecks bool) (err error) {
 			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()
@@ -1556,12 +1562,34 @@ func (m *MerkleTreeTask) UpdateMtree(skipAllChecks bool) (err error) {
 		mtreeTableIdentifier := pgx.Identifier{aceSchema(), fmt.Sprintf("ace_mtree_%s_%s", m.Schema, m.Table)}
 		mtreeTableName := mtreeTableIdentifier.Sanitize()
 
+		// Check if stored hashes use an older algorithm and need full recomputation.
+		hashVersion, err := queries.GetHashVersion(m.Ctx, tx, m.Schema, m.Table)
+		if err != nil {
+			return fmt.Errorf("error getting hash version on node %s: %w", nodeInfo["Name"], err)
+		}
+		hashVersionUpgraded := false
+		if hashVersion < queries.CurrentHashVersion {
+			marked, err := queries.MarkAllLeavesDirty(m.Ctx, tx, mtreeTableName)
+			if err != nil {
+				return fmt.Errorf("error marking all leaves dirty for hash upgrade on node %s: %w", nodeInfo["Name"], err)
+			}
+			fmt.Printf("Hash algorithm upgraded (v%d -> v%d): marked %d blocks for recomputation on %s\n",
+				hashVersion, queries.CurrentHashVersion, marked, nodeInfo["Name"])
+			hashVersionUpgraded = true
+		}
+
 		blocksToUpdate, err := queries.GetDirtyAndNewBlocks(m.Ctx, tx, mtreeTableName, m.SimplePrimaryKey, m.Key)
 		if err != nil {
 			return fmt.Errorf("error getting dirty blocks on node %s: %w", nodeInfo["Name"], err)
 		}
 
 		if len(blocksToUpdate) == 0 {
+			if hashVersionUpgraded {
+				// No blocks exist yet, but still update the version marker.
+				if err := queries.UpdateHashVersion(m.Ctx, tx, m.Schema, m.Table, queries.CurrentHashVersion); err != nil {
+					return fmt.Errorf("error updating hash version on node %s: %w", nodeInfo["Name"], err)
+				}
+			}
 			fmt.Printf("No updates needed for %s\n", nodeInfo["Name"])
 			tx.Commit(m.Ctx)
 			continue
@@ -1642,6 +1670,12 @@ func (m *MerkleTreeTask) UpdateMtree(skipAllChecks bool) (err error) {
 			}
 		}
 
+		if hashVersionUpgraded {
+			if err := queries.UpdateHashVersion(m.Ctx, tx, m.Schema, m.Table, queries.CurrentHashVersion); err != nil {
+				return fmt.Errorf("error updating hash version on node %s: %w", nodeInfo["Name"], err)
+			}
+		}
+
 		if err := tx.Commit(m.Ctx); err != nil {
 			return fmt.Errorf("error committing transaction on node %s: %w", nodeInfo["Name"], err)
 		}
@@ -2440,7 +2474,7 @@ func (m *MerkleTreeTask) createMtreeObjects(tx pgx.Tx, totalRows int64, numBlock
 		return fmt.Errorf("failed to create metadata table: %w", err)
 	}
 
-	err = queries.UpdateMetadata(m.Ctx, tx, m.Schema, m.Table, totalRows, m.BlockSize, numBlocks, !m.SimplePrimaryKey)
+	err = queries.UpdateMetadata(m.Ctx, tx, m.Schema, m.Table, totalRows, m.BlockSize, numBlocks, !m.SimplePrimaryKey, queries.CurrentHashVersion)
 	if err != nil {
 		return fmt.Errorf("failed to update metadata: %w", err)
 	}