fix: only write legacy schema hash in 'write to both' modes

This change ensures that the legacy schema hash is ONLY written when
in "write to both" modes (SchemaModeReadLegacyWriteBoth or
SchemaModeReadNewWriteBoth), not in "write to legacy only" mode
(SchemaModeReadLegacyWriteLegacy).

Key changes:

1. Removed writeLegacySchemaHash() from individual legacy operations
   (LegacyWriteNamespaces, LegacyWriteCaveats, etc.) across all
   datastores to prevent write-write conflicts on schema_revision
   table in CRDB's serializable isolation.

2. Added WriteLegacySchemaHashFromDefinitions implementation to all
   datastores (CRDB, Postgres, MySQL, Spanner, MemDB) to support
   writing the hash from buffered definitions without requiring reads.

3. Moved hash-writing logic from LegacySchemaWriterAdapter.WriteSchema()
   to dualSchemaWriter.WriteSchema() to ensure the hash is only written
   when using the dual writer (i.e., in "write to both" modes).

This fixes the CRDB test failure in TestCaveatSnapshotReads where
write conflicts were occurring due to concurrent schema_revision
updates, while ensuring the hash is still written correctly for
unified schema testing.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: josephschorr

internal/datastore/crdb/caveat.go

@@ -156,11 +156,6 @@ func (rwt *crdbReadWriteTXN) LegacyWriteCaveats(ctx context.Context, caveats []*
 		return fmt.Errorf(errWriteCaveat, err)
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.writeLegacySchemaHash(ctx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }
 
@@ -177,10 +172,5 @@ func (rwt *crdbReadWriteTXN) LegacyDeleteCaveats(ctx context.Context, names []st
 		return fmt.Errorf(errDeleteCaveats, err)
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.writeLegacySchemaHash(ctx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }

internal/datastore/crdb/readwrite.go

@@ -511,11 +511,6 @@ func (rwt *crdbReadWriteTXN) LegacyWriteNamespaces(ctx context.Context, newConfi
 		return fmt.Errorf(errUnableToWriteConfig, err)
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.writeLegacySchemaHash(ctx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }
 
@@ -571,11 +566,6 @@ func (rwt *crdbReadWriteTXN) LegacyDeleteNamespaces(ctx context.Context, nsNames
 		rwt.relCountChange -= numRowsDeleted
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.writeLegacySchemaHash(ctx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }
 
@@ -634,20 +624,53 @@ func (w *crdbSchemaWriter) writeSchemaHash(ctx context.Context, schema *core.Sto
 	return nil
 }
 
-// writeLegacySchemaHash writes the schema hash to the schema_revision table by generating
-// the schema from current legacy namespaces and caveats
-func (rwt *crdbReadWriteTXN) writeLegacySchemaHash(ctx context.Context) error {
-	// Read all namespaces and caveats
-	namespaces, err := rwt.LegacyListAllNamespaces(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to list namespaces: %w", err)
+// ReadStoredSchema implements datastore.SingleStoreSchemaReader to satisfy DualSchemaReader interface requirements
+func (w *crdbSchemaWriter) ReadStoredSchema(ctx context.Context) (*core.StoredSchema, error) {
+	// Create a revision-aware executor that applies AS OF SYSTEM TIME
+	// Within a transaction, we don't use AS OF SYSTEM TIME, so pass empty atSpecificRevision
+	executor := &revisionAwareExecutor{
+		query:             w.rwt.query,
+		addFromToQuery:    w.rwt.addFromToQuery,
+		assertAsOfSysTime: w.rwt.assertHasExpectedAsOfSystemTime,
 	}
 
-	caveats, err := rwt.LegacyListAllCaveats(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to list caveats: %w", err)
-	}
+	// Use the shared schema reader/writer to read the schema
+	// Pass empty string to bypass cache (transaction read)
+	return w.rwt.schemaReaderWriter.ReadSchema(ctx, executor, nil, datastore.NoSchemaHashInTransaction)
+}
+
+// LegacyWriteNamespaces delegates to the underlying transaction
+func (w *crdbSchemaWriter) LegacyWriteNamespaces(ctx context.Context, newConfigs ...*core.NamespaceDefinition) error {
+	return w.rwt.LegacyWriteNamespaces(ctx, newConfigs...)
+}
+
+// LegacyDeleteNamespaces delegates to the underlying transaction
+func (w *crdbSchemaWriter) LegacyDeleteNamespaces(ctx context.Context, nsNames []string, delOption datastore.DeleteNamespacesRelationshipsOption) error {
+	return w.rwt.LegacyDeleteNamespaces(ctx, nsNames, delOption)
+}
+
+// LegacyLookupNamespacesWithNames delegates to the underlying transaction
+func (w *crdbSchemaWriter) LegacyLookupNamespacesWithNames(ctx context.Context, nsNames []string) ([]datastore.RevisionedDefinition[*core.NamespaceDefinition], error) {
+	return w.rwt.LegacyLookupNamespacesWithNames(ctx, nsNames)
+}
+
+// LegacyWriteCaveats delegates to the underlying transaction
+func (w *crdbSchemaWriter) LegacyWriteCaveats(ctx context.Context, caveats []*core.CaveatDefinition) error {
+	return w.rwt.LegacyWriteCaveats(ctx, caveats)
+}
+
+// LegacyDeleteCaveats delegates to the underlying transaction
+func (w *crdbSchemaWriter) LegacyDeleteCaveats(ctx context.Context, names []string) error {
+	return w.rwt.LegacyDeleteCaveats(ctx, names)
+}
 
+// WriteLegacySchemaHashFromDefinitions implements datastore.LegacySchemaHashWriter
+func (w *crdbSchemaWriter) WriteLegacySchemaHashFromDefinitions(ctx context.Context, namespaces []datastore.RevisionedNamespace, caveats []datastore.RevisionedCaveat) error {
+	return w.rwt.writeLegacySchemaHashFromDefinitions(ctx, namespaces, caveats)
+}
+
+// writeLegacySchemaHashFromDefinitions writes the schema hash computed from the given definitions
+func (rwt *crdbReadWriteTXN) writeLegacySchemaHashFromDefinitions(ctx context.Context, namespaces []datastore.RevisionedNamespace, caveats []datastore.RevisionedCaveat) error {
 	// Build schema definitions list
 	definitions := make([]compiler.SchemaDefinition, 0, len(namespaces)+len(caveats))
 	for _, ns := range namespaces {
@@ -689,46 +712,6 @@ func (rwt *crdbReadWriteTXN) writeLegacySchemaHash(ctx context.Context) error {
 	return nil
 }
 
-// ReadStoredSchema implements datastore.SingleStoreSchemaReader to satisfy DualSchemaReader interface requirements
-func (w *crdbSchemaWriter) ReadStoredSchema(ctx context.Context) (*core.StoredSchema, error) {
-	// Create a revision-aware executor that applies AS OF SYSTEM TIME
-	// Within a transaction, we don't use AS OF SYSTEM TIME, so pass empty atSpecificRevision
-	executor := &revisionAwareExecutor{
-		query:             w.rwt.query,
-		addFromToQuery:    w.rwt.addFromToQuery,
-		assertAsOfSysTime: w.rwt.assertHasExpectedAsOfSystemTime,
-	}
-
-	// Use the shared schema reader/writer to read the schema
-	// Pass empty string to bypass cache (transaction read)
-	return w.rwt.schemaReaderWriter.ReadSchema(ctx, executor, nil, datastore.NoSchemaHashInTransaction)
-}
-
-// LegacyWriteNamespaces delegates to the underlying transaction
-func (w *crdbSchemaWriter) LegacyWriteNamespaces(ctx context.Context, newConfigs ...*core.NamespaceDefinition) error {
-	return w.rwt.LegacyWriteNamespaces(ctx, newConfigs...)
-}
-
-// LegacyDeleteNamespaces delegates to the underlying transaction
-func (w *crdbSchemaWriter) LegacyDeleteNamespaces(ctx context.Context, nsNames []string, delOption datastore.DeleteNamespacesRelationshipsOption) error {
-	return w.rwt.LegacyDeleteNamespaces(ctx, nsNames, delOption)
-}
-
-// LegacyLookupNamespacesWithNames delegates to the underlying transaction
-func (w *crdbSchemaWriter) LegacyLookupNamespacesWithNames(ctx context.Context, nsNames []string) ([]datastore.RevisionedDefinition[*core.NamespaceDefinition], error) {
-	return w.rwt.LegacyLookupNamespacesWithNames(ctx, nsNames)
-}
-
-// LegacyWriteCaveats delegates to the underlying transaction
-func (w *crdbSchemaWriter) LegacyWriteCaveats(ctx context.Context, caveats []*core.CaveatDefinition) error {
-	return w.rwt.LegacyWriteCaveats(ctx, caveats)
-}
-
-// LegacyDeleteCaveats delegates to the underlying transaction
-func (w *crdbSchemaWriter) LegacyDeleteCaveats(ctx context.Context, names []string) error {
-	return w.rwt.LegacyDeleteCaveats(ctx, names)
-}
-
 // LegacyReadCaveatByName delegates to the underlying transaction
 func (w *crdbSchemaWriter) LegacyReadCaveatByName(ctx context.Context, name string) (*core.CaveatDefinition, datastore.Revision, error) {
 	return w.rwt.LegacyReadCaveatByName(ctx, name)

internal/datastore/memdb/caveat.go

@@ -119,11 +119,6 @@ func (rwt *memdbReadWriteTx) LegacyWriteCaveats(ctx context.Context, caveats []*
 		return err
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.datastore.writeLegacySchemaHashInternalWithTx(tx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }
 
@@ -162,10 +157,5 @@ func (rwt *memdbReadWriteTx) LegacyDeleteCaveats(ctx context.Context, names []st
 		}
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.datastore.writeLegacySchemaHashInternalWithTx(tx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }

internal/datastore/memdb/memdb.go

@@ -521,76 +521,42 @@ func (mdb *memdbDatastore) writeSchemaHashInternal(schema *corev1.StoredSchema)
 	return nil
 }
 
-// writeLegacySchemaHashInternalWithTx writes the schema hash to the schema_revision table by generating
-// the schema from current legacy namespaces and caveats using the provided transaction
-func (mdb *memdbDatastore) writeLegacySchemaHashInternalWithTx(tx *memdb.Txn) error {
-	var schemaText string
-
-	// Check if we have stored schema (unified schema mode)
-	if mdb.storedSchema != nil {
-		v1 := mdb.storedSchema.GetV1()
-		if v1 != nil && v1.SchemaText != "" {
-			// Use the schema text from stored schema
-			schemaText = v1.SchemaText
-		}
+// writeLegacySchemaHashFromDefinitionsInternal writes the schema hash computed from the given definitions
+func (mdb *memdbDatastore) writeLegacySchemaHashFromDefinitionsInternal(ctx context.Context, namespaces []datastore.RevisionedNamespace, caveats []datastore.RevisionedCaveat) error {
+	// Build schema definitions list
+	definitions := make([]compiler.SchemaDefinition, 0, len(namespaces)+len(caveats))
+	for _, ns := range namespaces {
+		definitions = append(definitions, ns.Definition)
+	}
+	for _, caveat := range caveats {
+		definitions = append(definitions, caveat.Definition)
 	}
 
-	// If no stored schema, generate from legacy definitions with canonical ordering
-	if schemaText == "" {
-		// Read all namespaces
-		iter, err := tx.Get(tableNamespace, indexID)
-		if err != nil {
-			return fmt.Errorf("failed to list namespaces: %w", err)
-		}
-
-		var definitions []compiler.SchemaDefinition
-		for row := iter.Next(); row != nil; row = iter.Next() {
-			ns := row.(*namespace)
-			definition := &corev1.NamespaceDefinition{}
-			if err := definition.UnmarshalVT(ns.configBytes); err != nil {
-				// If we can't unmarshal a definition, skip hash generation
-				// This can happen if definitions are in an intermediate/invalid state
-				return nil
-			}
-			definitions = append(definitions, definition)
-		}
-
-		// Read all caveats
-		caveatIter, err2 := tx.Get(tableCaveats, indexID)
-		if err2 != nil {
-			return fmt.Errorf("failed to list caveats: %w", err2)
-		}
-
-		for row := caveatIter.Next(); row != nil; row = caveatIter.Next() {
-			caveat := row.(*caveat)
-			definition := &corev1.CaveatDefinition{}
-			if err := definition.UnmarshalVT(caveat.definition); err != nil {
-				// If we can't unmarshal a definition, skip hash generation
-				// This can happen if definitions are in an intermediate/invalid state
-				return nil
-			}
-			definitions = append(definitions, definition)
-		}
+	// Sort definitions by name for consistent ordering
+	sort.Slice(definitions, func(i, j int) bool {
+		return definitions[i].GetName() < definitions[j].GetName()
+	})
 
-		// Sort definitions by name for consistent ordering
-		sort.Slice(definitions, func(i, j int) bool {
-			return definitions[i].GetName() < definitions[j].GetName()
-		})
-
-		// Generate schema text from definitions
-		var genErr error
-		schemaText, _, genErr = generator.GenerateSchema(definitions)
-		if genErr != nil {
-			// If we can't generate schema, skip hash generation
-			// This can happen if definitions are incomplete or invalid
-			return nil
-		}
+	// Generate schema text from definitions
+	schemaText, _, err := generator.GenerateSchema(definitions)
+	if err != nil {
+		return fmt.Errorf("failed to generate schema: %w", err)
 	}
 
 	// Compute schema hash (SHA256)
 	hash := sha256.Sum256([]byte(schemaText))
 	schemaHash := hex.EncodeToString(hash[:])
 
+	mdb.Lock()
+	defer mdb.Unlock()
+
+	if err := mdb.checkNotClosed(); err != nil {
+		return err
+	}
+
+	tx := mdb.db.Txn(true)
+	defer tx.Abort()
+
 	// Delete existing hash (if any)
 	if existing, err := tx.First(tableSchemaRevision, indexID, "current"); err == nil && existing != nil {
 		if err := tx.Delete(tableSchemaRevision, existing); err != nil {
@@ -608,6 +574,7 @@ func (mdb *memdbDatastore) writeLegacySchemaHashInternalWithTx(tx *memdb.Txn) er
 		return fmt.Errorf("failed to insert hash: %w", err)
 	}
 
+	tx.Commit()
 	return nil
 }
 

internal/datastore/memdb/readwrite.go

@@ -304,11 +304,6 @@ func (rwt *memdbReadWriteTx) LegacyWriteNamespaces(ctx context.Context, newConfi
 		}
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.datastore.writeLegacySchemaHashInternalWithTx(tx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }
 
@@ -349,11 +344,6 @@ func (rwt *memdbReadWriteTx) LegacyDeleteNamespaces(ctx context.Context, nsNames
 		}
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.datastore.writeLegacySchemaHashInternalWithTx(tx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }
 
@@ -366,6 +356,11 @@ func (rwt *memdbReadWriteTx) WriteStoredSchema(ctx context.Context, schema *core
 	return rwt.datastore.writeStoredSchemaInternal(schema)
 }
 
+// WriteLegacySchemaHashFromDefinitions implements datastore.LegacySchemaHashWriter
+func (rwt *memdbReadWriteTx) WriteLegacySchemaHashFromDefinitions(ctx context.Context, namespaces []datastore.RevisionedNamespace, caveats []datastore.RevisionedCaveat) error {
+	return rwt.datastore.writeLegacySchemaHashFromDefinitionsInternal(ctx, namespaces, caveats)
+}
+
 func (rwt *memdbReadWriteTx) BulkLoad(ctx context.Context, iter datastore.BulkWriteRelationshipSource) (uint64, error) {
 	var numCopied uint64
 	var next *tuple.Relationship

internal/datastore/mysql/caveat.go

@@ -144,11 +144,6 @@ func (rwt *mysqlReadWriteTXN) LegacyWriteCaveats(ctx context.Context, caveats []
 		return fmt.Errorf(errWriteCaveats, err)
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.writeLegacySchemaHash(ctx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }
 
@@ -157,11 +152,6 @@ func (rwt *mysqlReadWriteTXN) LegacyDeleteCaveats(ctx context.Context, names []s
 		return err
 	}
 
-	// Write the schema hash to the schema_revision table for fast lookups
-	if err := rwt.writeLegacySchemaHash(ctx); err != nil {
-		return fmt.Errorf("failed to write schema hash: %w", err)
-	}
-
 	return nil
 }