refactor(storage): group MySQL schema into readmodel and pipeline packages by owning service

Makefile

@@ -386,7 +386,7 @@ local-submitqueue-gateway-stop: ## Stop Gateway service
 
 local-init-submitqueue-schemas: ## Manually apply all database schemas
 	@echo "Applying storage schema to mysql-app..."
-	@for file in submitqueue/extension/storage/mysql/schema/*.sql; do \
+	@for file in submitqueue/extension/storage/mysql/schema/*/*.sql; do \
 		echo "  - Applying $$(basename $$file)..."; \
 		docker exec -i $(SUBMITQUEUE_LOCAL_PROJECT)-mysql-app-1 mysql -uroot -proot submitqueue < $$file 2>&1 | grep -v "Using a password" || true; \
 	done

submitqueue/extension/storage/mysql/schema/BUILD.bazel

@@ -1,5 +1,11 @@
+# The whole schema. A deployment that colocates gateway and orchestrator on one
+# database uses this alone; the per-owner packages exist for deployments that
+# give each service its own database, and are not a statement that they must.
 filegroup(
     name = "schema",
-    srcs = glob(["*.sql"]),
+    srcs = [
+        "//submitqueue/extension/storage/mysql/schema/pipeline",
+        "//submitqueue/extension/storage/mysql/schema/readmodel",
+    ],
     visibility = ["//visibility:public"],
 )

submitqueue/extension/storage/mysql/schema/README.md

@@ -1,5 +1,14 @@
 # MySQL Schema
 
+## Layout
+
+Tables are grouped by the service that owns them, one Bazel package each:
+
+- `readmodel/` — the gateway-owned read model: the append-only request log and the three materialized projections behind request-summary retrieval and `List` (see [Gateway request read model](#gateway-request-read-model) below and [doc/rfc/submitqueue/status-list-api.md](../../../../../doc/rfc/submitqueue/status-list-api.md)).
+- `pipeline/` — orchestrator pipeline working state: requests, batches, builds, changes, and speculation. Different retention semantics from the read model, and never read by the gateway APIs.
+
+The parent `:schema` filegroup is the union of both and remains the default: a deployment that colocates gateway and orchestrator on one database depends on it alone. The per-owner packages exist so a deployment that gives each service its own database can provision exactly that service's tables; they are not a statement that it must. Adding a table means placing its `.sql` in the owning package — the grouping is the file's location, so there is no list to keep in sync.
+
 ## Queue-leading primary keys
 
 Every table leads its primary key with `queue`: `request` and `batch` on `(queue, id)`, `build` on `(queue, id)`, `batch_dependent` on `(queue, batch_id)`, `request_batch` on `(queue, request_id, batch_id)`, `change` on `(queue, uri, request_id)`, `queue_batch_state` on `(queue, state, batch_id)`, `speculation_path_set` on `(queue, head)`, `request_summary` on `(queue, request_id)`, `request_log` on `(queue, request_id, timestamp_ms, salt)`, `change_uri_request_mapping` on `(queue, change_uri, received_at_ms, request_id)`, and `request_summary_by_queue` on `(queue, received_at_ms, request_id)`. A queue-bound store instance prefixes every read and stamps every write with its bound queue, so one queue's rows are unreachable through another queue's binding and every table is shardable by queue. `//tool/linter/queueshard` enforces this, and also rejects any secondary index that does not itself lead with `queue`, since such an index would reintroduce a cross-queue access path.

submitqueue/extension/storage/mysql/schema/pipeline/BUILD.bazel

@@ -0,0 +1,5 @@
+filegroup(
+    name = "pipeline",
+    srcs = glob(["*.sql"]),
+    visibility = ["//visibility:public"],
+)

submitqueue/extension/storage/mysql/schema/readmodel/BUILD.bazel

@@ -0,0 +1,5 @@
+filegroup(
+    name = "readmodel",
+    srcs = glob(["*.sql"]),
+    visibility = ["//visibility:public"],
+)

test/testutil/schema.go

@@ -17,9 +17,11 @@ package testutil
 import (
 	"context"
 	"database/sql"
+	"io/fs"
 	"os"
 	"path/filepath"
 	"sort"
+	"strings"
 	"testing"
 
 	_ "github.com/go-sql-driver/mysql"
@@ -46,19 +48,33 @@ func SchemaDir(relativePath string) string {
 	return Runfile(relativePath)
 }
 
-// ApplySchema reads all .sql files from the schema directory and executes them on the database.
+// ApplySchema reads every .sql file under the schema directory, including
+// subdirectories, and executes them on the database. A schema may group its
+// tables into per-owner subpackages (see
+// submitqueue/extension/storage/mysql/schema), so passing the root applies the
+// whole schema regardless of how it is subdivided.
 func ApplySchema(t *testing.T, log *TestLogger, db *sql.DB, schemaDirectory string) {
 	t.Helper()
 
-	files, err := filepath.Glob(filepath.Join(schemaDirectory, "*.sql"))
-	require.NoError(t, err, "failed to glob schema files")
-	require.NotEmpty(t, files, "no .sql schema files found in %s", schemaDirectory)
+	var files []string
+	err := filepath.WalkDir(schemaDirectory, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if !d.IsDir() && strings.HasSuffix(d.Name(), ".sql") {
+			files = append(files, path)
+		}
+		return nil
+	})
+	require.NoError(t, err, "failed to walk schema files")
+	require.NotEmpty(t, files, "no .sql schema files found under %s", schemaDirectory)
 
 	// Sort files to ensure deterministic schema application order.
 	sort.Strings(files)
 
 	for _, f := range files {
-		name := filepath.Base(f)
+		name, relErr := filepath.Rel(schemaDirectory, f)
+		require.NoError(t, relErr, "failed to relativize schema file %s", f)
 		log.Logf("Applying schema: %s", name)
 
 		content, err := os.ReadFile(f)

tool/linter/queueshard/main.go

@@ -26,9 +26,11 @@ package main
 import (
 	"flag"
 	"fmt"
+	"io/fs"
 	"os"
 	"path/filepath"
 	"regexp"
+	"sort"
 	"strings"
 )
 
@@ -40,7 +42,8 @@ var schemaShardColumns = map[string]map[string]bool{
 	"platform/extension/messagequeue/mysql/schema": {"tenant": true},
 }
 
-// schemaRoots are the directories scanned for table definitions.
+// schemaRoots are the directory trees scanned for table definitions. A root's
+// tables may be grouped into per-owner subpackages; the whole tree is scanned.
 var schemaRoots = []string{
 	"submitqueue/extension/storage/mysql/schema",
 	"stovepipe/extension/storage/mysql/schema",
@@ -74,9 +77,9 @@ func main() {
 	var checked int
 	for _, schemaRoot := range schemaRoots {
 		shardColumns := schemaShardColumns[schemaRoot]
-		files, err := filepath.Glob(filepath.Join(root, schemaRoot, "*.sql"))
+		files, err := findSchemaFiles(filepath.Join(root, schemaRoot))
 		if err != nil {
-			fmt.Fprintf(os.Stderr, "error globbing %s: %v\n", schemaRoot, err)
+			fmt.Fprintf(os.Stderr, "error walking %s: %v\n", schemaRoot, err)
 			os.Exit(1)
 		}
 		if len(files) == 0 {
@@ -112,6 +115,27 @@ func main() {
 	fmt.Printf("All %d tables are shardable.\n", checked)
 }
 
+// findSchemaFiles returns every .sql file under dir, including subdirectories.
+// A schema root may group its tables into per-owner subpackages, so the whole
+// tree is scanned rather than only the root's own files.
+func findSchemaFiles(dir string) ([]string, error) {
+	var files []string
+	err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if !d.IsDir() && strings.HasSuffix(d.Name(), ".sql") {
+			files = append(files, path)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(files)
+	return files, nil
+}
+
 // check returns the number of tables found in content and any violations.
 func check(file, content string, shardColumns map[string]bool) (int, []violation) {
 	var violations []violation