From 8a03bbdb9b55d51a46a524bfe41e18a3fb8444bc Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 22:29:29 -0700 Subject: [PATCH 1/3] [1/4 messagequeue] RFC: per-tenant Vitess sharding ## Summary ### Why? Vitess needs a stable vindex that is not the Kafka-style partition key. SubmitQueue already uses partition_key for ordering, so hashing it would scatter one queue across shards and mix tenants. ### What? - Add the per-tenant MQ sharding RFC: tenant column as the vindex, queueName mapped at wiring, partition_key remaining the in-tenant ordering unit. - Link the RFC from the index and note the consumer-gate interaction. ## Test Plan Docs only. Review the RFC against the follow-up implementation PRs. Co-authored-by: Cursor --- doc/rfc/consumer-gate.md | 8 +-- doc/rfc/index.md | 1 + doc/rfc/messagequeue-tenant-sharding.md | 96 +++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 4 deletions(-) create mode 100644 doc/rfc/messagequeue-tenant-sharding.md diff --git a/doc/rfc/consumer-gate.md b/doc/rfc/consumer-gate.md index 342aac93e..c62a91997 100644 --- a/doc/rfc/consumer-gate.md +++ b/doc/rfc/consumer-gate.md @@ -40,12 +40,12 @@ Keeping the contract separate from any backend is what lets the storage medium b The first implementation stores gate state as plain files under a configured directory. Presence of a gate file means the gate is closed; deleting the file opens it. Parked deliveries are recorded as JSON files. The layout: ``` -{dir}/gates/{consumer_group}/all # gates every partition of the controller -{dir}/gates/{consumer_group}/p-{urlenc(partition)} # gates one partition -{dir}/parked/{consumer_group}/{topic}/{urlenc(id)}.json # one parked delivery record +{dir}/gates/{consumer_group}/all # gates every partition of the controller +{dir}/gates/{consumer_group}/partitions/t-{urlenc(tenant)}/p-{urlenc(partition)} # gates one tenant-scoped partition +{dir}/parked/{consumer_group}/{topic}/t-{urlenc(tenant)}/p-{urlenc(partition)}/{urlenc(id)}.json ``` -Consumer groups and topics are already filesystem-safe by the repo's naming rules; partition keys and message IDs may contain `/` (request IDs like `queue/1`), so they are URL-encoded in file names. Gate files contain human-readable JSON metadata — `reason`, `created_by`, `created_at_ms` — so an operator finding a paused controller can tell why. Parked records carry the payload, attempt, and `parked_at_ms` while a delivery is blocked; each re-check of a still-closed gate refreshes the record, and the admit path removes it once the gate opens, so payloads are not retained after release. All writes go through temp-file-plus-rename so readers never see partial JSON. +Consumer groups and topics are already filesystem-safe by the repo's naming rules; tenant, partition keys, and message IDs are URL-encoded in path components. Gate files contain human-readable JSON metadata — `reason`, `created_by`, `created_at_ms` — so an operator finding a paused controller can tell why. Parked records carry the payload, attempt, and `parked_at_ms` while a delivery is blocked; each re-check of a still-closed gate refreshes the record, and the admit path removes it once the gate opens, so payloads are not retained after release. All writes go through temp-file-plus-rename so readers never see partial JSON. Files are the simplest medium for the E2E and single-host scope: diff --git a/doc/rfc/index.md b/doc/rfc/index.md index d3319e661..5fa4ec3f0 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -5,6 +5,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting ## Shared - [SQL-Based Distributed Queue](sql-queue-rfc.md) - MySQL-based distributed message queue with partition leasing and at-least-once delivery (used by SubmitQueue, Stovepipe, and other repo-local services) +- [Message Queue Tenant Sharding](messagequeue-tenant-sharding.md) - Per-tenant Vitess vindex on the platform MySQL message queue; SubmitQueue maps `queueName` to `tenant` at wiring - [Message Queue Contract](messagequeue-contract.md) - How queue payloads are defined (Protobuf, serialized as protobuf JSON), located by audience (external in `api/{domain}/messagequeue/`, internal in `{domain}/core/messagequeue/`), bound to topics (the `topics` proto option), and enforced by Bazel visibility - [Consumer Gate](consumer-gate.md) - Stopping and starting individual queue controllers at runtime via a consumer-side check: blocked deliveries are recorded as parked and postponed back to the queue (re-checked on redelivery), gate state as a separate extension with a file-based first implementation shared by tests and operators - [Consumer Hold](consumer-hold.md) - Fourth delivery outcome letting a controller postpone its delivery: the message becomes a partition barrier that pauses consumption for a chosen delay, redelivers in order, and does not count as a failure toward dead-lettering diff --git a/doc/rfc/messagequeue-tenant-sharding.md b/doc/rfc/messagequeue-tenant-sharding.md new file mode 100644 index 000000000..7b68928c8 --- /dev/null +++ b/doc/rfc/messagequeue-tenant-sharding.md @@ -0,0 +1,96 @@ +# RFC: Per-Tenant Vitess Sharding for the MySQL Message Queue + +## Metadata + +| Field | Value | +|-------|-------| +| **Author** | Preetam Dwivedi | +| **Status** | In Review | +| **Created** | 2026-09-03 | + +## Summary + +The platform MySQL message queue gains a domain-agnostic `tenant` column on every table. Vitess routes on `tenant`; every primary key and hot-path query leads with it. SubmitQueue, Stovepipe, and Runway map their `queueName` onto `tenant` at the wiring boundary. `partition_key` remains the ordering unit within a tenant and is not the vindex. + +## Background + +Domain storage (`request`, `batch`, `counter`, …) already shards by a leading `queue` column. The message queue backend was deliberately excluded from `tool/linter/queueshard` because it keyed rows by `(consumer_group, topic, partition_key)` with a global `AUTO_INCREMENT offset` — fine for a single MySQL instance, not for Vitess. + +SubmitQueue's business queue name already flows through publish metadata (`queue_name`) and consumer context (`WithQueueName`). Several pipeline stages use a different `partition_key` (build ID, request ID) so those deliveries serialize independently while still belonging to one business queue. Sharding on `partition_key` would split one queue across shards; the MQ needs a dedicated isolation column. + +## Naming + +| Layer | Column / field | Meaning | +|-------|----------------|---------| +| Platform MQ schema | `tenant` | Vitess vindex; opaque to the backend | +| Platform `Message` | `Tenant` | Persisted shard identity | +| SubmitQueue domain | `queue` / `queueName` | Same string as `tenant` at wiring | +| Platform MQ schema | `partition_key` | Ordering unit within `(tenant, topic)` | + +The MQ schema does not use `queue` — that word is overloaded (SubmitQueue domain, `Queue` interface, `queue_*` table prefix). + +## Schema + +Every table's primary key leads with `tenant`. Tenant, topic, consumer-group, and subscriber identifiers use `VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin`; partition keys and message IDs use explicit `VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`. This keeps operational identifiers readable, preserves unrestricted UTF-8 ordering keys and IDs, and keeps the largest composite key within InnoDB's 3072-byte limit. The backend validates each contract before database access. Secondary indexes that do not lead with `tenant` are removed except `queue_messages.idx_offset`, which InnoDB requires because the `AUTO_INCREMENT offset` column must be leftmost in an index. + +### `queue_messages` + +- PK: `(tenant, topic, partition_key, offset)` +- Unique: `(tenant, topic, partition_key, id)` +- Required InnoDB index: `idx_offset (offset)` for the `AUTO_INCREMENT` column +- `offset` is allocated from a shard-wide monotonic sequence and used as an ordering cursor within each partition; fetch is `WHERE tenant=? AND topic=? AND partition_key=? AND offset>? ORDER BY offset` + +### `queue_delivery_state` + +- PK: `(tenant, consumer_group, topic, partition_key, message_offset)` + +### `queue_offsets` + +- PK: `(tenant, topic, partition_key, consumer_group)` +- Drop `idx_topic` + +### `queue_partition_leases` + +- PK: `(tenant, consumer_group, topic, partition_key)` +- Drop `idx_lease_renewed`; purge is scoped to `(tenant, consumer_group, topic)` + +### `queue_subscriber_heartbeats` + +- PK: `(tenant, consumer_group, topic, subscriber_name)` + +DLQ moves rewrite `topic` to `original + suffix` and keep `tenant` + `partition_key` on the same shard. + +## Subscriber discovery + +Today partition discovery runs `SELECT DISTINCT partition_key FROM queue_messages WHERE topic=?`, which scatter-gathers across all Vitess shards. + +The subscriber takes an explicit configured tenant list from `MQ_TENANTS`. Consumer processes reject an empty list at startup; Stovepipe also rejects ingest requests for names outside the list. Discovery becomes: + +```sql +SELECT DISTINCT partition_key FROM queue_messages +WHERE tenant = ? AND topic = ? +ORDER BY partition_key +``` + +Fair-share, orphan sweep, and idle-lease release run per `(tenant, topic)`, not across all tenants on a topic. Discovery and shutdown attempt every configured tenant and aggregate errors so one unavailable shard does not block unrelated tenants. + +## Publish + +Every `platform/publish` call supplies tenant explicitly. The package stamps `Message.Tenant` and mirrors it into `queue_name` delivery metadata, rejecting empty tenants and conflicting caller metadata. `PartitionKey` is unchanged. + +## Wiring + +One `extqueue.Queue` and one VTGate DSN per service. `NewQueue` / subscriber `Params` carry `Tenants []string`. Consumer service wiring parses the authoritative comma-separated `MQ_TENANTS` list once and passes it to the backend and any ingress validation. + +## Operational tooling + +Fleet-wide admin reads must be explicit. The topic, offset, active-lease, and stale-lease listing commands require exactly one of a single tenant or all tenants; omitting both never falls back to a scatter query. Message inspection, deletion, and DLQ requeue identify a row by the complete unique identity `(tenant, topic, partition_key, id)`. + +## Out of scope + +Live migration of existing Stovepipe prod queue databases (expand/contract, backfill, dual-write). This RFC describes a breaking greenfield schema; prod cutover is a separate exercise. + +## Related + +- [SQL-Based Distributed Queue](sql-queue-rfc.md) +- [Modular Queue Wiring](submitqueue/modular-queue-wiring.md) From f7d04ef72d81a4bff5cd32263f992685c5c28019 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 23:03:20 -0700 Subject: [PATCH 2/3] docs(messagequeue): describe tenant sharding as sharded MySQL ## Summary ### Why? Vitess is the product, not the contract. The RFC should talk about sharded MySQL and a tenant shard key so the design does not read as Vitess-specific. ### What? - Drop the RFC metadata table. - Rename the RFC to per-tenant sharding for the MySQL message queue and replace vindex/Vitess wording with shard key and sharded MySQL. Co-authored-by: Cursor --- doc/rfc/index.md | 2 +- doc/rfc/messagequeue-tenant-sharding.md | 20 ++++++-------------- 2 files changed, 7 insertions(+), 15 deletions(-) diff --git a/doc/rfc/index.md b/doc/rfc/index.md index 5fa4ec3f0..1af941d9b 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -5,7 +5,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting ## Shared - [SQL-Based Distributed Queue](sql-queue-rfc.md) - MySQL-based distributed message queue with partition leasing and at-least-once delivery (used by SubmitQueue, Stovepipe, and other repo-local services) -- [Message Queue Tenant Sharding](messagequeue-tenant-sharding.md) - Per-tenant Vitess vindex on the platform MySQL message queue; SubmitQueue maps `queueName` to `tenant` at wiring +- [Message Queue Tenant Sharding](messagequeue-tenant-sharding.md) - Per-tenant shard key on the platform MySQL message queue; SubmitQueue maps `queueName` to `tenant` at wiring - [Message Queue Contract](messagequeue-contract.md) - How queue payloads are defined (Protobuf, serialized as protobuf JSON), located by audience (external in `api/{domain}/messagequeue/`, internal in `{domain}/core/messagequeue/`), bound to topics (the `topics` proto option), and enforced by Bazel visibility - [Consumer Gate](consumer-gate.md) - Stopping and starting individual queue controllers at runtime via a consumer-side check: blocked deliveries are recorded as parked and postponed back to the queue (re-checked on redelivery), gate state as a separate extension with a file-based first implementation shared by tests and operators - [Consumer Hold](consumer-hold.md) - Fourth delivery outcome letting a controller postpone its delivery: the message becomes a partition barrier that pauses consumption for a chosen delay, redelivers in order, and does not count as a failure toward dead-lettering diff --git a/doc/rfc/messagequeue-tenant-sharding.md b/doc/rfc/messagequeue-tenant-sharding.md index 7b68928c8..001d14a1e 100644 --- a/doc/rfc/messagequeue-tenant-sharding.md +++ b/doc/rfc/messagequeue-tenant-sharding.md @@ -1,20 +1,12 @@ -# RFC: Per-Tenant Vitess Sharding for the MySQL Message Queue - -## Metadata - -| Field | Value | -|-------|-------| -| **Author** | Preetam Dwivedi | -| **Status** | In Review | -| **Created** | 2026-09-03 | +# RFC: Per-Tenant Sharding for the MySQL Message Queue ## Summary -The platform MySQL message queue gains a domain-agnostic `tenant` column on every table. Vitess routes on `tenant`; every primary key and hot-path query leads with it. SubmitQueue, Stovepipe, and Runway map their `queueName` onto `tenant` at the wiring boundary. `partition_key` remains the ordering unit within a tenant and is not the vindex. +The platform MySQL message queue gains a domain-agnostic `tenant` column on every table. Sharded MySQL routes on `tenant`; every primary key and hot-path query leads with it. SubmitQueue, Stovepipe, and Runway map their `queueName` onto `tenant` at the wiring boundary. `partition_key` remains the ordering unit within a tenant and is not the shard key. ## Background -Domain storage (`request`, `batch`, `counter`, …) already shards by a leading `queue` column. The message queue backend was deliberately excluded from `tool/linter/queueshard` because it keyed rows by `(consumer_group, topic, partition_key)` with a global `AUTO_INCREMENT offset` — fine for a single MySQL instance, not for Vitess. +Domain storage (`request`, `batch`, `counter`, …) already shards by a leading `queue` column. The message queue backend was deliberately excluded from `tool/linter/queueshard` because it keyed rows by `(consumer_group, topic, partition_key)` with a global `AUTO_INCREMENT offset` — fine for a single MySQL instance, not for sharded MySQL. SubmitQueue's business queue name already flows through publish metadata (`queue_name`) and consumer context (`WithQueueName`). Several pipeline stages use a different `partition_key` (build ID, request ID) so those deliveries serialize independently while still belonging to one business queue. Sharding on `partition_key` would split one queue across shards; the MQ needs a dedicated isolation column. @@ -22,7 +14,7 @@ SubmitQueue's business queue name already flows through publish metadata (`queue | Layer | Column / field | Meaning | |-------|----------------|---------| -| Platform MQ schema | `tenant` | Vitess vindex; opaque to the backend | +| Platform MQ schema | `tenant` | Shard key; opaque to the backend | | Platform `Message` | `Tenant` | Persisted shard identity | | SubmitQueue domain | `queue` / `queueName` | Same string as `tenant` at wiring | | Platform MQ schema | `partition_key` | Ordering unit within `(tenant, topic)` | @@ -62,7 +54,7 @@ DLQ moves rewrite `topic` to `original + suffix` and keep `tenant` + `partition_ ## Subscriber discovery -Today partition discovery runs `SELECT DISTINCT partition_key FROM queue_messages WHERE topic=?`, which scatter-gathers across all Vitess shards. +Today partition discovery runs `SELECT DISTINCT partition_key FROM queue_messages WHERE topic=?`, which scatter-gathers across all shards. The subscriber takes an explicit configured tenant list from `MQ_TENANTS`. Consumer processes reject an empty list at startup; Stovepipe also rejects ingest requests for names outside the list. Discovery becomes: @@ -80,7 +72,7 @@ Every `platform/publish` call supplies tenant explicitly. The package stamps `Me ## Wiring -One `extqueue.Queue` and one VTGate DSN per service. `NewQueue` / subscriber `Params` carry `Tenants []string`. Consumer service wiring parses the authoritative comma-separated `MQ_TENANTS` list once and passes it to the backend and any ingress validation. +One `extqueue.Queue` and one sharded MySQL DSN per service. `NewQueue` / subscriber `Params` carry `Tenants []string`. Consumer service wiring parses the authoritative comma-separated `MQ_TENANTS` list once and passes it to the backend and any ingress validation. ## Operational tooling From f8b507a6417ab43a6737c11663df39839fba6c24 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 23:05:44 -0700 Subject: [PATCH 3/3] docs(messagequeue): document ascii vs utf8mb4 identifier limits ## Summary ### Why? VARCHAR(255) is not a single byte budget. Reviewers need to know why operational IDs are ascii_bin and partition keys are utf8mb4_bin, including the 255-byte vs 255-character distinction and InnoDB's 3072-byte key cap. ### What? - Explain CHARACTER SET ascii COLLATE ascii_bin vs utf8mb4/utf8mb4_bin, NOT NULL, and the per-column limits in the tenant-sharding RFC. Co-authored-by: Cursor --- doc/rfc/messagequeue-tenant-sharding.md | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/doc/rfc/messagequeue-tenant-sharding.md b/doc/rfc/messagequeue-tenant-sharding.md index 001d14a1e..dea5e5b85 100644 --- a/doc/rfc/messagequeue-tenant-sharding.md +++ b/doc/rfc/messagequeue-tenant-sharding.md @@ -23,7 +23,17 @@ The MQ schema does not use `queue` — that word is overloaded (SubmitQueue doma ## Schema -Every table's primary key leads with `tenant`. Tenant, topic, consumer-group, and subscriber identifiers use `VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin`; partition keys and message IDs use explicit `VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`. This keeps operational identifiers readable, preserves unrestricted UTF-8 ordering keys and IDs, and keeps the largest composite key within InnoDB's 3072-byte limit. The backend validates each contract before database access. Secondary indexes that do not lead with `tenant` are removed except `queue_messages.idx_offset`, which InnoDB requires because the `AUTO_INCREMENT offset` column must be leftmost in an index. +Every table's primary key leads with `tenant`. Secondary indexes that do not lead with `tenant` are removed except `queue_messages.idx_offset`, which InnoDB requires because the `AUTO_INCREMENT offset` column must be leftmost in an index. The Go backend validates each identifier before database access so a bad value fails in the process, not as a SQL error. + +### Column types and limits + +`VARCHAR(255)` is a *character* limit, not a universal byte limit. The character set decides how many bytes that is, what bytes are legal, and how equality and ordering work. + +`CHARACTER SET ascii COLLATE ascii_bin NOT NULL` is for operational identifiers the service owns: `tenant`, `topic`, `consumer_group`, `subscriber_name`, `leased_by`, `original_topic`. ASCII is bytes `0x00`–`0x7F` only (one byte per character), so `VARCHAR(255)` is **255 bytes**. `ascii_bin` compares those bytes as-is: case-sensitive (`Foo` ≠ `foo`), no Unicode folding, `ORDER BY` is byte order. `NOT NULL` rejects SQL `NULL`; the empty string is a different value and the backend still rejects empty tenant/topic/consumer-group before insert. The backend additionally rejects embedded `NUL` bytes. + +`CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL` is for caller-chosen keys that may be real Unicode: `partition_key` and message `id`. utf8mb4 is the full Unicode set (up to 4 bytes per character, including supplementary planes). `VARCHAR(255)` is **255 characters** (at most 1020 bytes). `utf8mb4_bin` compares by binary code points: case-sensitive, no accent folding. The table default `DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin` covers `payload`-adjacent text (`TEXT` / `JSON`) the same way. + +These two encodings keep composite primary keys inside InnoDB's **3072-byte** index limit. InnoDB counts utf8mb4 at 4 bytes per character, so a key of three ASCII `VARCHAR(255)` columns plus one utf8mb4 `VARCHAR(255)` is `255 + 255 + 255 + 1020 = 1785` bytes before the `BIGINT` offset; two utf8mb4 columns plus two ASCII columns stay under the cap as well. Putting `tenant` (the shard key) in utf8mb4 would spend four times the index budget on a value that is always an ASCII queue name. ### `queue_messages`