jb/shared schemas

[Maxim Manuylov (maxmanuylov)]

No description provided.

[Copilot]

Pull request overview

This PR introduces a “shared schema” capability that lets multiple tenants reference a single stored schema, while preserving existing per-tenant schema behavior. It extends the API surface (new SharedSchema service + request metadata fields) and updates storage implementations (Postgres + in-memory) plus supporting plumbing (servers, factories, proxies, docs).

Changes:

• Add SharedSchema gRPC/HTTP service for writing/reading/assigning/listing shared schemas.
• Extend multiple request metadata messages with shared_schema_id and update schema reader interfaces to route reads via tenant/shared-schema selection.
• Add Postgres migration + storage implementations and update in-memory storage + docs/OpenAPI artifacts.

Reviewed changes

Copilot reviewed 53 out of 54 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
proto/base/v1/service.proto	Adds shared_schema_id to request metadata and introduces SharedSchema service + messages.
proto/base/v1/openapi.proto	Bumps published OpenAPI version string.
pkg/pb/base/v1/service.pb.validate.go	Regenerates validation for new proto fields/messages.
pkg/pb/base/v1/service.pb.gw.go	Regenerates grpc-gateway handlers for new SharedSchema endpoints.
pkg/pb/base/v1/service_grpc.pb.go	Regenerates gRPC service stubs for SharedSchema.
pkg/pb/base/v1/openapi.pb.go	Regenerates embedded OpenAPI descriptor with new version.
pkg/development/development.go	Wires noop shared-schema reader/writer and updates schema-reader call signatures.
pkg/cmd/validate.go	Updates schema-reader call signatures in validation command.
pkg/cmd/serve.go	Wires SharedSchema reader/writer factories into server container.
internal/storage/storage.go	Extends SchemaReader interface and adds SharedSchemaReader/Writer interfaces + noops.
internal/storage/proxies/singleflight/schema_reader.go	Updates proxy to new HeadVersion return type and new schema-reader signatures.
internal/storage/proxies/singleflight/schema_reader_test.go	Updates tests for new schema-reader signatures/returns.
internal/storage/proxies/circuitbreaker/schema_reader.go	Updates circuitbreaker proxy for new schema-reader signatures/returns.
internal/storage/proxies/cache/schema_reader.go	Updates cache proxy signatures and introduces shared-schema-aware cache keys.
internal/storage/postgres/tenant_reader.go	Includes shared_schema_id column in tenant listing reads.
internal/storage/postgres/shared_schema_writer.go	Adds Postgres writer for shared schema definitions and tenant assignment.
internal/storage/postgres/shared_schema_reader.go	Adds Postgres reader for shared schema reads, head version, listing, and tenant mapping.
internal/storage/postgres/schema_writer.go	Clears tenant shared_schema_id when writing per-tenant schema.
internal/storage/postgres/schema_writer_test.go	Updates schema read calls for new signature.
internal/storage/postgres/schema_reader.go	Routes schema reads to shared vs per-tenant tables; HeadVersion now returns (sharedSchemaID, version).
internal/storage/postgres/schema_reader_test.go	Updates tests for new signature and behaviors.
internal/storage/postgres/migrations/20260707000000_shared_schemas.sql	Adds shared_schema_definitions table and tenants.shared_schema_id column.
internal/storage/postgres/consts.go	Adds constant for shared_schema_definitions table.
internal/storage/model.go	Adds SharedSchemaDefinition model and Tenant.SharedSchemaID field.
internal/storage/memory/shared_schema_writer.go	Adds in-memory shared schema writer + tenant assignment.
internal/storage/memory/shared_schema_writer_test.go	Adds tests for in-memory shared schema writer.
internal/storage/memory/shared_schema_reader.go	Adds in-memory shared schema reader + head version/listing/tenant mapping.
internal/storage/memory/shared_schema_reader_test.go	Adds tests for in-memory shared schema reader and shared-schema head-version behavior.
internal/storage/memory/schema_writer.go	Clears tenant shared schema assignment when writing per-tenant schema (memory).
internal/storage/memory/schema_writer_test.go	Updates schema read calls for new signature.
internal/storage/memory/schema_reader.go	Routes reads to shared vs per-tenant tables; HeadVersion now returns (sharedSchemaID, version).
internal/storage/memory/schema_reader_test.go	Updates tests for new signature/returns.
internal/storage/memory/migrations/schema.go	Adds memdb table/indexes for shared schema definitions.
internal/storage/memory/constants/constants.go	Adds constant for shared schema definitions table (memory).
internal/servers/shared_schema_server.go	Implements SharedSchema server (write/read/assign/list).
internal/servers/server.go	Adds shared schema reader/writer to container and registers gRPC + HTTP handlers.
internal/servers/schema_server.go	Updates schema server to use shared schema routing via updated schema reader interface.
internal/servers/data_server.go	Uses shared_schema_id from request metadata (or resolves via HeadVersion) for schema validation reads.
internal/invoke/invoke.go	Populates metadata.shared_schema_id alongside schema_version when auto-resolving head version.
internal/info.go	Bumps internal version constant.
internal/factories/storage.go	Adds factories for shared schema reader/writer for Postgres and memory.
internal/engines/subject_permission.go	Passes shared_schema_id into schema-reader calls.
internal/engines/subject_filter.go	Passes shared_schema_id into schema-reader calls.
internal/engines/lookup.go	Makes LookupEngine schema cache/shared reads aware of shared_schema_id.
internal/engines/lookup_test.go	Updates mock schema reader and tests for new signatures.
internal/engines/expand.go	Passes shared_schema_id into schema-reader calls.
internal/engines/check.go	Passes shared_schema_id into schema-reader calls.
internal/engines/cache/check.go	Passes shared_schema_id into schema-reader calls.
internal/engines/balancer/balancer.go	Passes shared_schema_id into schema-reader calls.
docs/api-reference/openapiv2/apidocs.swagger.json	Updates generated swagger (version + SharedSchema endpoints + metadata fields).
docs/api-reference/apidocs.swagger.json	Updates generated swagger (version + SharedSchema endpoints + metadata fields).
.github/workflows/release.yml	Updates release workflow input example tag format.

Suppressed comments (1)

internal/servers/schema_server.go:214

• SchemaServer.Read has the same issue as PartialWrite: when schema_version is provided, sharedSchemaID remains empty. For tenants using a shared schema, this will read from per-tenant schema tables instead of the shared schema tables. Consider always resolving sharedSchemaID via HeadVersion and only defaulting version when it’s empty.

	version := request.GetMetadata().GetSchemaVersion()
	var sharedSchemaID string
	if version == "" {
		var err error
		sharedSchemaID, version, err = r.sr.HeadVersion(ctx, request.GetTenantId())

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 55 out of 56 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

internal/storage/memory/shared_schema_reader.go:193

• Pagination token handling drops the first item of the next page. This method returns the token as the (page_size+1)th schema ID, but when a token is provided it skips entries until it equals the token and then continues, so the token entry is never returned and is effectively lost.

		if lowerBound != "" && !startPage {
			if e.id == lowerBound {
				startPage = true
			}
			continue

internal/servers/schema_server.go:214

• When schema_version is explicitly provided, sharedSchemaID stays empty, so tenants that are assigned a shared schema will read from per-tenant tables instead of the shared schema. Since SchemaReadRequestMetadata has no shared_schema_id, the server should still infer whether the tenant is in shared-schema mode even when a version is provided.

	version := request.GetMetadata().GetSchemaVersion()
	var sharedSchemaID string
	if version == "" {
		var err error
		sharedSchemaID, version, err = r.sr.HeadVersion(ctx, request.GetTenantId())

internal/storage/postgres/shared_schema_writer.go:36

• Postgres implementations for shared schemas are new (write + assign) but there are no postgres-focused tests covering key behaviors (write/read roundtrip, assignment, and pagination tokens). The memory implementation has tests, and other postgres storage components in this repo are covered, so adding analogous tests here would help prevent regressions.

[Copilot]

Pull request overview

Copilot reviewed 55 out of 56 changed files in this pull request and generated 2 comments.

Suppressed comments (6)

internal/servers/schema_server.go:214

• When schema_version is provided, sharedSchemaID remains empty and ReadSchema will always hit per-tenant schema tables even if the tenant is configured to use a shared schema. This can make reads fail or return the wrong schema unless the caller also explicitly provides shared_schema_id.

Consider always resolving the tenant’s shared schema assignment (via HeadVersion) and only defaulting the version when schema_version is omitted.

	version := request.GetMetadata().GetSchemaVersion()
	var sharedSchemaID string
	if version == "" {
		var err error
		sharedSchemaID, version, err = r.sr.HeadVersion(ctx, request.GetTenantId())

internal/servers/shared_schema_server.go:51

• This server method doesn’t run protobuf validation (request.Validate()), while other API servers do (e.g., permission_server.go, data_server.go). This means invalid shared_schema_id values (pattern/max_bytes) won’t be rejected consistently.

Add request.Validate() checks here (and similarly in Read/Assign/List) before parsing/using the request fields.

	ctx, span := internal.Tracer.Start(ctx, "shared-schemas.write")
	defer span.End()

	sch, err := parser.NewParser(request.GetSchema()).Parse()
	if err != nil {

internal/storage/postgres/shared_schema_writer.go:38

• WriteSharedSchema builds an INSERT statement even when definitions is empty. That can generate invalid SQL (INSERT with no VALUES) or silently do nothing depending on the SQL builder behavior.

Return a clear invalid-argument error when no definitions are provided (callers can validate empty schemas earlier).
internal/storage/memory/schema_writer.go:67

• This tenant update ignores the Insert error and doesn’t Abort the transaction on early returns. If the insert fails, the function will still return nil and may leave the tenant record unchanged (still pointing at the shared schema).

Handle the Insert error and use defer tenantTxn.Abort() for consistent transaction cleanup.

		t := raw.(storage.Tenant)
		if t.SharedSchemaID != "" {
			t.SharedSchemaID = ""
			_ = tenantTxn.Insert(constants.TenantsTable, t)
		}

internal/storage/postgres/shared_schema_reader.go:41

• New Postgres shared-schema storage logic is introduced here, but there are no Postgres-level tests exercising it (contrast with schema_reader_test.go / schema_writer_test.go). Adding tests for read/write/head/list and tenant assignment would help catch SQL/query regressions and ensure behavior matches memory implementation.
  pkg/pb/base/v1/service_grpc.pb.go:1570
• Typo in generated comment: "pancis" → "panics".

[Copilot]

Pull request overview

Copilot reviewed 59 out of 60 changed files in this pull request and generated no new comments.

Suppressed comments (4)

internal/servers/schema_server.go:220

• When schema_version is provided, sharedSchemaID stays empty, so tenants assigned to a shared schema will incorrectly read from per-tenant schema_definitions. Since SchemaReadRequest has no shared_schema_id field, this makes versioned reads for shared-schema tenants effectively impossible. Consider always resolving sharedSchemaID via HeadVersion and only defaulting version from the head when it is empty.

	version := request.GetMetadata().GetSchemaVersion()
	var sharedSchemaID string
	if version == "" {
		var err error
		sharedSchemaID, version, err = r.sr.HeadVersion(ctx, request.GetTenantId())
		if err != nil {
			return nil, status.Error(GetStatus(err), err.Error()) // Return version error
		}
	}

	response, err := r.sr.ReadSchema(ctx, request.GetTenantId(), sharedSchemaID, version)

internal/storage/postgres/shared_schema_writer.go:41

• WriteSharedSchema can be called with an empty definitions slice (e.g., when the parsed schema contains no statements). In that case the INSERT builder has no VALUES, which yields a SQL-builder/exec error and an unhelpful error code. Consider explicitly rejecting empty writes with ERROR_CODE_INVALID_ARGUMENT before building/executing the INSERT.
  internal/storage/postgres/shared_schema_reader.go:41
• New Postgres SharedSchemaReader logic (read/list/head-version mapping) is not covered by tests, while other Postgres storage components in this package have integration tests. Adding tests for at least WriteSharedSchema + ReadSharedSchema + ListSharedSchemas would help prevent regressions in SQL queries and pagination semantics.
  internal/storage/memory/shared_schema_writer.go:32
• WriteSharedSchema commits and updates sharedHeadVersion even when definitions is empty, which will write an entry for the empty sharedSchemaID ("") and version (""). This can corrupt in-memory state and break List/HeadVersion behavior. Add an early return for empty writes.

// WriteSharedSchema writes a shared schema to repository
func (w *SharedSchemaWriter) WriteSharedSchema(_ context.Context, definitions []storage.SharedSchemaDefinition) error {
	txn := w.database.DB.Txn(true)
	defer txn.Abort()