feat: add unified schema storage with ReadStoredSchema/WriteStoredSchema

Introduce the foundation for unified schema storage, where the entire
compiled schema (definitions + schema text + hash) is stored as a single
serialized proto blob, chunked across rows for SQL datastores.

Datastore interface changes:
- Add ReadStoredSchema(ctx, SchemaHash) to Reader interface
- Add WriteStoredSchema(ctx, *StoredSchema) to ReadWriteTransaction
- Add SchemaHash type with sentinel values for cache bypass

DataLayer interface changes:
- SnapshotReader now takes (Revision, SchemaHash) to thread cache keys
- OptimizedRevision/HeadRevision return SchemaHash alongside Revision
- Add SchemaMode configuration for migration path (legacy → dual → new)
- Add storedSchemaReaderAdapter for reading from StoredSchema protos
- Add writeSchemaViaStoredSchema for building and writing StoredSchema

Storage implementation:
- Add SQLByteChunker generic chunked blob storage for SQL datastores
- Add SQLSingleStoreSchemaReaderWriter for read/write of StoredSchema
- Add per-datastore ReadStoredSchema/WriteStoredSchema implementations
  for postgres, crdb, mysql, spanner, and memdb
- Add schema table migrations for all SQL datastores
- Add populate migrations to backfill from legacy namespace/caveat tables
- Rename MySQL schema table to stored_schema (schema is a reserved word)

Caching:
- Add SchemaHashCache with LRU + singleflight for schema-by-hash lookups

Proxy/middleware updates:
- Add ReadStoredSchema/WriteStoredSchema pass-through to all datastore
  proxies (observable, counting, readonly, singleflight, indexcheck,
  strictreplicated, checkingreplicated, relationshipintegrity, hashcache)
- Update consistency middleware for new HeadRevision/OptimizedRevision
  signatures

Proto changes:
- Add StoredSchema message to core.proto with V1StoredSchema containing
  schema_text, schema_hash, namespace_definitions, caveat_definitions

Author: josephschorr

internal/caveats/run_test.go

@@ -471,7 +471,7 @@ func TestRunCaveatExpressions(t *testing.T) {
 			req.NoError(err)
 
 			dl := datalayer.NewDataLayer(ds)
-			sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+			sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 			req.NoError(err)
 
 			for _, debugOption := range []RunCaveatExpressionDebugOption{
@@ -524,7 +524,7 @@ func TestRunCaveatWithMissingMap(t *testing.T) {
 	req.NoError(err)
 
 	dl := datalayer.NewDataLayer(ds)
-	sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+	sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 	req.NoError(err)
 
 	result, err := RunSingleCaveatExpression(
@@ -556,7 +556,7 @@ func TestRunCaveatWithEmptyMap(t *testing.T) {
 	req.NoError(err)
 
 	dl := datalayer.NewDataLayer(ds)
-	sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+	sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 	req.NoError(err)
 
 	_, err = RunSingleCaveatExpression(
@@ -594,7 +594,7 @@ func TestRunCaveatMultipleTimes(t *testing.T) {
 	req.NoError(err)
 
 	dl := datalayer.NewDataLayer(ds)
-	sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+	sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 	req.NoError(err)
 
 	runner := NewCaveatRunner(types.Default.TypeSet)
@@ -662,7 +662,7 @@ func TestRunCaveatWithMissingDefinition(t *testing.T) {
 	req.NoError(err)
 
 	dl := datalayer.NewDataLayer(ds)
-	sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+	sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 	req.NoError(err)
 
 	// Try to run a caveat that doesn't exist
@@ -697,7 +697,7 @@ func TestCaveatRunnerPopulateCaveatDefinitionsForExpr(t *testing.T) {
 	req.NoError(err)
 
 	dl := datalayer.NewDataLayer(ds)
-	sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+	sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 	req.NoError(err)
 
 	runner := NewCaveatRunner(types.Default.TypeSet)
@@ -742,7 +742,7 @@ func TestCaveatRunnerEmptyExpression(t *testing.T) {
 	req.NoError(err)
 
 	dl := datalayer.NewDataLayer(ds)
-	sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+	sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 	req.NoError(err)
 
 	runner := NewCaveatRunner(types.Default.TypeSet)
@@ -823,7 +823,7 @@ func TestUnknownCaveatOperation(t *testing.T) {
 	req.NoError(err)
 
 	dl := datalayer.NewDataLayer(ds)
-	sr, err := dl.SnapshotReader(headRevision).ReadSchema()
+	sr, err := dl.SnapshotReader(headRevision, datastore.NoSchemaHashForTesting).ReadSchema()
 	req.NoError(err)
 
 	runner := NewCaveatRunner(types.Default.TypeSet)

internal/datastore/common/chunkbytes.go

@@ -8,6 +8,9 @@ import (
 	sq "github.com/Masterminds/squirrel"
 )
 
+// ErrNoChunksFound is returned when no chunks are found for a given key.
+var ErrNoChunksFound = errors.New("no chunks found")
+
 // ChunkedBytesTransaction defines the interface for executing SQL queries within a transaction.
 type ChunkedBytesTransaction interface {
 	// ExecuteWrite executes an INSERT query.
@@ -81,48 +84,70 @@ type SQLByteChunkerConfig[T any] struct {
 	AliveValue T
 }
 
+// WithExecutor returns a copy of the config with the specified executor.
+func (c SQLByteChunkerConfig[T]) WithExecutor(executor ChunkedBytesExecutor) SQLByteChunkerConfig[T] {
+	c.Executor = executor
+	return c
+}
+
+// WithTableName returns a copy of the config with the specified table name.
+func (c SQLByteChunkerConfig[T]) WithTableName(tableName string) SQLByteChunkerConfig[T] {
+	c.TableName = tableName
+	return c
+}
+
 // SQLByteChunker provides methods for reading and writing byte data
 // that is chunked across multiple rows in a SQL table.
 type SQLByteChunker[T any] struct {
 	config SQLByteChunkerConfig[T]
 }
 
-// MustNewSQLByteChunker creates a new SQLByteChunker with the specified configuration.
-// Panics if the configuration is invalid.
-func MustNewSQLByteChunker[T any](config SQLByteChunkerConfig[T]) *SQLByteChunker[T] {
+// NewSQLByteChunker creates a new SQLByteChunker with the specified configuration.
+// Returns an error if the configuration is invalid.
+func NewSQLByteChunker[T any](config SQLByteChunkerConfig[T]) (*SQLByteChunker[T], error) {
 	if config.MaxChunkSize <= 0 {
-		panic("maxChunkSize must be greater than 0")
+		return nil, errors.New("maxChunkSize must be greater than 0")
 	}
 	if config.TableName == "" {
-		panic("tableName cannot be empty")
+		return nil, errors.New("tableName cannot be empty")
 	}
 	if config.NameColumn == "" {
-		panic("nameColumn cannot be empty")
+		return nil, errors.New("nameColumn cannot be empty")
 	}
 	if config.ChunkIndexColumn == "" {
-		panic("chunkIndexColumn cannot be empty")
+		return nil, errors.New("chunkIndexColumn cannot be empty")
 	}
 	if config.ChunkDataColumn == "" {
-		panic("chunkDataColumn cannot be empty")
+		return nil, errors.New("chunkDataColumn cannot be empty")
 	}
 	if config.PlaceholderFormat == nil {
-		panic("placeholderFormat cannot be nil")
+		return nil, errors.New("placeholderFormat cannot be nil")
 	}
 	if config.Executor == nil {
-		panic("executor cannot be nil")
+		return nil, errors.New("executor cannot be nil")
 	}
 	if config.WriteMode == WriteModeInsertWithTombstones {
 		if config.CreatedAtColumn == "" {
-			panic("createdAtColumn is required when using WriteModeInsertWithTombstones")
+			return nil, errors.New("createdAtColumn is required when using WriteModeInsertWithTombstones")
 		}
 		if config.DeletedAtColumn == "" {
-			panic("deletedAtColumn is required when using WriteModeInsertWithTombstones")
+			return nil, errors.New("deletedAtColumn is required when using WriteModeInsertWithTombstones")
 		}
 	}
 
 	return &SQLByteChunker[T]{
 		config: config,
+	}, nil
+}
+
+// MustNewSQLByteChunker creates a new SQLByteChunker with the specified configuration.
+// Panics if the configuration is invalid.
+func MustNewSQLByteChunker[T any](config SQLByteChunkerConfig[T]) *SQLByteChunker[T] {
+	chunker, err := NewSQLByteChunker(config)
+	if err != nil {
+		panic(err)
 	}
+	return chunker
 }
 
 // WriteChunkedBytes writes chunked byte data to the database within a transaction.
@@ -305,7 +330,7 @@ func (c *SQLByteChunker[T]) ReadChunkedBytes(
 // into the original byte array. It validates that all chunks are present and in order.
 func (c *SQLByteChunker[T]) reassembleChunks(chunks map[int][]byte) ([]byte, error) {
 	if len(chunks) == 0 {
-		return nil, errors.New("no chunks found")
+		return nil, ErrNoChunksFound
 	}
 
 	// Validate that we have all chunks from 0 to N-1 and calculate total size
@@ -338,7 +363,10 @@ func (c *SQLByteChunker[T]) chunkData(data []byte) [][]byte {
 	chunks := make([][]byte, 0, numChunks)
 
 	for i := 0; i < len(data); i += c.config.MaxChunkSize {
-		end := min(i+c.config.MaxChunkSize, len(data))
+		end := i + c.config.MaxChunkSize
+		if end > len(data) {
+			end = len(data)
+		}
 		chunks = append(chunks, data[i:end])
 	}
 

internal/datastore/common/chunkbytes_test.go

@@ -57,13 +57,17 @@ type fakeExecutor struct {
 	readResult  map[int][]byte
 	readErr     error
 	transaction *fakeTransaction
+	onRead      func() // Optional callback invoked on each read
 }
 
 func (m *fakeExecutor) BeginTransaction(ctx context.Context) (ChunkedBytesTransaction, error) {
 	return m.transaction, nil
 }
 
 func (m *fakeExecutor) ExecuteRead(ctx context.Context, builder sq.SelectBuilder) (map[int][]byte, error) {
+	if m.onRead != nil {
+		m.onRead()
+	}
 	if m.readErr != nil {
 		return nil, m.readErr
 	}
@@ -392,6 +396,7 @@ func TestReadChunkedBytes(t *testing.T) {
 		name          string
 		chunks        map[int][]byte
 		expectedData  []byte
+		expectedErr   error
 		expectedError string
 	}{
 		{
@@ -418,9 +423,9 @@ func TestReadChunkedBytes(t *testing.T) {
 			expectedData: []byte{},
 		},
 		{
-			name:          "no chunks",
-			chunks:        map[int][]byte{},
-			expectedError: "no chunks found",
+			name:        "no chunks",
+			chunks:      map[int][]byte{},
+			expectedErr: ErrNoChunksFound,
 		},
 		{
 			name: "missing chunk in sequence",
@@ -450,10 +455,13 @@ func TestReadChunkedBytes(t *testing.T) {
 
 			data, err := chunker.ReadChunkedBytes(context.Background(), "test-key")
 
-			if tt.expectedError != "" {
+			switch {
+			case tt.expectedErr != nil:
+				require.ErrorIs(t, err, tt.expectedErr)
+			case tt.expectedError != "":
 				require.Error(t, err)
 				require.Contains(t, err.Error(), tt.expectedError)
-			} else {
+			default:
 				require.NoError(t, err)
 				require.Equal(t, tt.expectedData, data)
 			}