From ecfcd55000418064203a149f1736c701b1a06f51 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 5 Aug 2026 23:06:54 -0700 Subject: [PATCH] feat(storage): queue-leading primary keys with row-level cross-queue isolation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The factory contract made queue-scoped storage instances the only way in, but the schema still let a bound instance reach another queue's rows: `request`, `batch`, `build`, `batch_dependent`, and `request_batch` were keyed by bare IDs. Leading every queue-scoped table's primary key with `queue` makes the binding real at the row level and leaves every table shardable by queue — cheap now (pre-prod, tables recreated), expensive to retrofit later. ### What? `request` and `batch` move to `(queue, id)`, `build` to `(queue, id)` with a new `queue` column, `batch_dependent` to `(queue, batch_id)`, and `request_batch` to `(queue, request_id, batch_id)`; `change`, `queue_batch_state`, and `request_summary_by_queue` were already queue-leading, and the global read-model tables stay queue-free. The bound MySQL stores now stamp their queue into every write and prefix every read with it, so a wrong-queue read misses (`ErrNotFound` / empty) instead of leaking. The `build` key change also removes a latent cross-queue uniqueness assumption on runner-minted build IDs — two queues sharing one CI pipeline may now mint the same identifier. The storage contract suite gains a queue-isolation test covering all five re-keyed stores, and the e2e harness resolves its white-box request reads per queue. ## Test Plan ✅ `go test ./...` ✅ storage integration contract suite via bazel, including the new cross-queue isolation cases ✅ `make fmt` ✅ `make lint` ✅ `make check-tidy` ✅ `make check-gazelle`. Schemas are pre-production: tables are recreated, no data migration. --- .../storage/mysql/batch_dependent_store.go | 12 ++--- .../mysql/batch_dependent_store_test.go | 28 ++++++------ .../extension/storage/mysql/batch_store.go | 8 ++-- .../storage/mysql/batch_store_test.go | 20 ++++----- .../extension/storage/mysql/build_store.go | 12 ++--- .../storage/mysql/build_store_test.go | 20 ++++----- .../storage/mysql/request_batch_store.go | 8 ++-- .../storage/mysql/request_batch_store_test.go | 16 +++---- .../extension/storage/mysql/request_store.go | 8 ++-- .../storage/mysql/request_store_test.go | 20 ++++----- .../extension/storage/mysql/schema/README.md | 8 +++- .../extension/storage/mysql/schema/batch.sql | 6 +-- .../storage/mysql/schema/batch_dependent.sql | 3 +- .../extension/storage/mysql/schema/build.sql | 3 +- .../storage/mysql/schema/request.sql | 4 +- .../storage/mysql/schema/request_batch.sql | 3 +- test/e2e/submitqueue/BUILD.bazel | 1 - test/e2e/submitqueue/harness_test.go | 6 ++- test/e2e/submitqueue/suite_test.go | 14 +++--- .../submitqueue/extension/storage/suite.go | 45 +++++++++++++++++++ 20 files changed, 149 insertions(+), 96 deletions(-) diff --git a/submitqueue/extension/storage/mysql/batch_dependent_store.go b/submitqueue/extension/storage/mysql/batch_dependent_store.go index 8e424079..b6463479 100644 --- a/submitqueue/extension/storage/mysql/batch_dependent_store.go +++ b/submitqueue/extension/storage/mysql/batch_dependent_store.go @@ -51,8 +51,8 @@ func (s *batchDependentStore) Get(ctx context.Context, batchID string) (ret enti var dependentsJSON []byte err := s.db.QueryRowContext(ctx, - "SELECT batch_id, dependents, version FROM batch_dependent WHERE batch_id = ?", - batchID, + "SELECT batch_id, dependents, version FROM batch_dependent WHERE queue = ? AND batch_id = ?", + s.queue, batchID, ).Scan(&bd.BatchID, &dependentsJSON, &bd.Version) if errors.Is(err, sql.ErrNoRows) { @@ -80,8 +80,8 @@ func (s *batchDependentStore) Create(ctx context.Context, batchDependent entity. } _, err = s.db.ExecContext(ctx, - "INSERT INTO batch_dependent (batch_id, dependents, version) VALUES (?, ?, ?)", - batchDependent.BatchID, dependentsJSON, batchDependent.Version, + "INSERT INTO batch_dependent (queue, batch_id, dependents, version) VALUES (?, ?, ?, ?)", + s.queue, batchDependent.BatchID, dependentsJSON, batchDependent.Version, ) if err != nil { var mysqlErr *mysql.MySQLError @@ -107,8 +107,8 @@ func (s *batchDependentStore) Update(ctx context.Context, batchDependent entity. } result, err := s.db.ExecContext(ctx, - "UPDATE batch_dependent SET dependents = ?, version = ? WHERE batch_id = ? AND version = ?", - dependentsJSON, newVersion, batchDependent.BatchID, oldVersion, + "UPDATE batch_dependent SET dependents = ?, version = ? WHERE queue = ? AND batch_id = ? AND version = ?", + dependentsJSON, newVersion, s.queue, batchDependent.BatchID, oldVersion, ) if err != nil { return fmt.Errorf( diff --git a/submitqueue/extension/storage/mysql/batch_dependent_store_test.go b/submitqueue/extension/storage/mysql/batch_dependent_store_test.go index 194dcebd..84393da3 100644 --- a/submitqueue/extension/storage/mysql/batch_dependent_store_test.go +++ b/submitqueue/extension/storage/mysql/batch_dependent_store_test.go @@ -65,7 +65,7 @@ func TestBatchDependentStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"batch_id", "dependents", "version"}). AddRow(want.BatchID, dependentsJSON, want.Version) mock.ExpectQuery("SELECT batch_id, dependents, version FROM batch_dependent"). - WithArgs(want.BatchID). + WithArgs("monorepo", want.BatchID). WillReturnRows(rows) }, want: want, @@ -77,7 +77,7 @@ func TestBatchDependentStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"batch_id", "dependents", "version"}). AddRow("monorepo/batch/nil", []byte("null"), int32(2)) mock.ExpectQuery("SELECT batch_id, dependents, version FROM batch_dependent"). - WithArgs("monorepo/batch/nil"). + WithArgs("monorepo", "monorepo/batch/nil"). WillReturnRows(rows) }, want: entity.BatchDependent{ @@ -92,7 +92,7 @@ func TestBatchDependentStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"batch_id", "dependents", "version"}). AddRow("monorepo/batch/empty", []byte("[]"), int32(3)) mock.ExpectQuery("SELECT batch_id, dependents, version FROM batch_dependent"). - WithArgs("monorepo/batch/empty"). + WithArgs("monorepo", "monorepo/batch/empty"). WillReturnRows(rows) }, want: entity.BatchDependent{ @@ -106,7 +106,7 @@ func TestBatchDependentStore_Get(t *testing.T) { batchID: "missing", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT batch_id, dependents, version FROM batch_dependent"). - WithArgs("missing"). + WithArgs("monorepo", "missing"). WillReturnError(sql.ErrNoRows) }, wantErr: true, @@ -117,7 +117,7 @@ func TestBatchDependentStore_Get(t *testing.T) { batchID: "bad", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT batch_id, dependents, version FROM batch_dependent"). - WithArgs("bad"). + WithArgs("monorepo", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -163,7 +163,7 @@ func TestBatchDependentStore_Create(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO batch_dependent"). - WithArgs(bd.BatchID, sqlmock.AnyArg(), bd.Version). + WithArgs("monorepo", bd.BatchID, sqlmock.AnyArg(), bd.Version). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -171,7 +171,7 @@ func TestBatchDependentStore_Create(t *testing.T) { name: "duplicate batch id returns ErrAlreadyExists", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO batch_dependent"). - WithArgs(bd.BatchID, sqlmock.AnyArg(), bd.Version). + WithArgs("monorepo", bd.BatchID, sqlmock.AnyArg(), bd.Version). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -181,7 +181,7 @@ func TestBatchDependentStore_Create(t *testing.T) { name: "other exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO batch_dependent"). - WithArgs(bd.BatchID, sqlmock.AnyArg(), bd.Version). + WithArgs("monorepo", bd.BatchID, sqlmock.AnyArg(), bd.Version). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -229,7 +229,7 @@ func TestBatchDependentStore_Update(t *testing.T) { entity: batchDependent, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch_dependent"). - WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, batchDependent.BatchID, oldVersion). + WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, "monorepo", batchDependent.BatchID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -241,7 +241,7 @@ func TestBatchDependentStore_Update(t *testing.T) { }, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch_dependent"). - WithArgs([]byte("null"), newVersion, "monorepo/batch/nil", oldVersion). + WithArgs([]byte("null"), newVersion, "monorepo", "monorepo/batch/nil", oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -254,7 +254,7 @@ func TestBatchDependentStore_Update(t *testing.T) { }, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch_dependent"). - WithArgs([]byte("[]"), newVersion, "monorepo/batch/empty", oldVersion). + WithArgs([]byte("[]"), newVersion, "monorepo", "monorepo/batch/empty", oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -263,7 +263,7 @@ func TestBatchDependentStore_Update(t *testing.T) { entity: batchDependent, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch_dependent"). - WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, batchDependent.BatchID, oldVersion). + WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, "monorepo", batchDependent.BatchID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -274,7 +274,7 @@ func TestBatchDependentStore_Update(t *testing.T) { entity: batchDependent, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch_dependent"). - WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, batchDependent.BatchID, oldVersion). + WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, "monorepo", batchDependent.BatchID, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -284,7 +284,7 @@ func TestBatchDependentStore_Update(t *testing.T) { entity: batchDependent, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch_dependent"). - WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, batchDependent.BatchID, oldVersion). + WithArgs([]byte(`["monorepo/batch/2","monorepo/batch/3"]`), newVersion, "monorepo", batchDependent.BatchID, oldVersion). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, diff --git a/submitqueue/extension/storage/mysql/batch_store.go b/submitqueue/extension/storage/mysql/batch_store.go index b3ad7fb2..2ac21a71 100644 --- a/submitqueue/extension/storage/mysql/batch_store.go +++ b/submitqueue/extension/storage/mysql/batch_store.go @@ -52,8 +52,8 @@ func (s *batchStore) Get(ctx context.Context, id string) (ret entity.Batch, retE var dependenciesJSON []byte err := s.db.QueryRowContext(ctx, - "SELECT id, queue, contains, dependencies, state, version FROM batch WHERE id = ?", - id, + "SELECT id, queue, contains, dependencies, state, version FROM batch WHERE queue = ? AND id = ?", + s.queue, id, ).Scan(&batch.ID, &batch.Queue, &containsJSON, &dependenciesJSON, &batch.State, &batch.Version) if errors.Is(err, sql.ErrNoRows) { @@ -130,8 +130,8 @@ func (s *batchStore) Update(ctx context.Context, batch entity.Batch, oldVersion, } result, err := s.db.ExecContext(ctx, - "UPDATE batch SET queue = ?, contains = ?, dependencies = ?, state = ?, version = ? WHERE id = ? AND version = ?", - batch.Queue, containsJSON, dependenciesJSON, batch.State, newVersion, batch.ID, oldVersion, + "UPDATE batch SET contains = ?, dependencies = ?, state = ?, version = ? WHERE queue = ? AND id = ? AND version = ?", + containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion, ) if err != nil { return fmt.Errorf( diff --git a/submitqueue/extension/storage/mysql/batch_store_test.go b/submitqueue/extension/storage/mysql/batch_store_test.go index 96182f8e..7cd69aef 100644 --- a/submitqueue/extension/storage/mysql/batch_store_test.go +++ b/submitqueue/extension/storage/mysql/batch_store_test.go @@ -70,7 +70,7 @@ func TestBatchStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"id", "queue", "contains", "dependencies", "state", "version"}). AddRow(want.ID, want.Queue, containsJSON, dependenciesJSON, string(want.State), want.Version) mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch"). - WithArgs(want.ID). + WithArgs("monorepo", want.ID). WillReturnRows(rows) }, want: want, @@ -80,7 +80,7 @@ func TestBatchStore_Get(t *testing.T) { id: "missing", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch"). - WithArgs("missing"). + WithArgs("monorepo", "missing"). WillReturnError(sql.ErrNoRows) }, wantErr: true, @@ -91,7 +91,7 @@ func TestBatchStore_Get(t *testing.T) { id: "bad", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch"). - WithArgs("bad"). + WithArgs("monorepo", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -103,7 +103,7 @@ func TestBatchStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"id", "queue", "contains", "dependencies", "state", "version"}). AddRow(want.ID, want.Queue, []byte("not json"), dependenciesJSON, string(want.State), want.Version) mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch"). - WithArgs("malformed"). + WithArgs("monorepo", "malformed"). WillReturnRows(rows) }, wantErr: true, @@ -225,7 +225,7 @@ func TestBatchStore_Update(t *testing.T) { batch: batch, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch"). - WithArgs(batch.Queue, containsJSON, dependenciesJSON, batch.State, newVersion, batch.ID, oldVersion). + WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -234,7 +234,7 @@ func TestBatchStore_Update(t *testing.T) { batch: batch, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch"). - WithArgs(batch.Queue, containsJSON, dependenciesJSON, batch.State, newVersion, batch.ID, oldVersion). + WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -245,7 +245,7 @@ func TestBatchStore_Update(t *testing.T) { batch: batch, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch"). - WithArgs(batch.Queue, containsJSON, dependenciesJSON, batch.State, newVersion, batch.ID, oldVersion). + WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -255,7 +255,7 @@ func TestBatchStore_Update(t *testing.T) { batch: batch, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch"). - WithArgs(batch.Queue, containsJSON, dependenciesJSON, batch.State, newVersion, batch.ID, oldVersion). + WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, @@ -270,7 +270,7 @@ func TestBatchStore_Update(t *testing.T) { }, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch"). - WithArgs(batch.Queue, []byte("null"), []byte("null"), batch.State, newVersion, batch.ID, oldVersion). + WithArgs([]byte("null"), []byte("null"), batch.State, newVersion, batch.Queue, batch.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -286,7 +286,7 @@ func TestBatchStore_Update(t *testing.T) { }, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE batch"). - WithArgs(batch.Queue, []byte("[]"), []byte("[]"), batch.State, newVersion, batch.ID, oldVersion). + WithArgs([]byte("[]"), []byte("[]"), batch.State, newVersion, batch.Queue, batch.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, diff --git a/submitqueue/extension/storage/mysql/build_store.go b/submitqueue/extension/storage/mysql/build_store.go index 65a6304f..40345946 100644 --- a/submitqueue/extension/storage/mysql/build_store.go +++ b/submitqueue/extension/storage/mysql/build_store.go @@ -49,8 +49,8 @@ func (s *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE var build entity.Build err := s.db.QueryRowContext(ctx, - "SELECT id, batch_id, status FROM build WHERE id = ?", - id, + "SELECT id, batch_id, status FROM build WHERE queue = ? AND id = ?", + s.queue, id, ).Scan(&build.ID, &build.BatchID, &build.Status) if errors.Is(err, sql.ErrNoRows) { @@ -69,8 +69,8 @@ func (s *buildStore) Create(ctx context.Context, build entity.Build) (retErr err defer func() { op.Complete(retErr) }() _, err := s.db.ExecContext(ctx, - "INSERT INTO build (id, batch_id, status) VALUES (?, ?, ?)", - build.ID, build.BatchID, build.Status, + "INSERT INTO build (queue, id, batch_id, status) VALUES (?, ?, ?, ?)", + s.queue, build.ID, build.BatchID, build.Status, ) if err != nil { var mysqlErr *mysql.MySQLError @@ -89,8 +89,8 @@ func (s *buildStore) Update(ctx context.Context, build entity.Build) (retErr err defer func() { op.Complete(retErr) }() result, err := s.db.ExecContext(ctx, - "UPDATE build SET batch_id = ?, status = ? WHERE id = ?", - build.BatchID, build.Status, build.ID, + "UPDATE build SET batch_id = ?, status = ? WHERE queue = ? AND id = ?", + build.BatchID, build.Status, s.queue, build.ID, ) if err != nil { return fmt.Errorf("failed to update build entity id=%q: %w", build.ID, err) diff --git a/submitqueue/extension/storage/mysql/build_store_test.go b/submitqueue/extension/storage/mysql/build_store_test.go index b1cbe163..2111f780 100644 --- a/submitqueue/extension/storage/mysql/build_store_test.go +++ b/submitqueue/extension/storage/mysql/build_store_test.go @@ -62,7 +62,7 @@ func TestBuildStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"id", "batch_id", "status"}). AddRow(want.ID, want.BatchID, string(want.Status)) mock.ExpectQuery("SELECT id, batch_id, status"). - WithArgs(want.ID). + WithArgs("monorepo", want.ID). WillReturnRows(rows) }, want: want, @@ -72,7 +72,7 @@ func TestBuildStore_Get(t *testing.T) { id: "missing", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT id, batch_id, status"). - WithArgs("missing"). + WithArgs("monorepo", "missing"). WillReturnError(sql.ErrNoRows) }, wantErr: true, @@ -83,7 +83,7 @@ func TestBuildStore_Get(t *testing.T) { id: "bad", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT id, batch_id, status"). - WithArgs("bad"). + WithArgs("monorepo", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -129,7 +129,7 @@ func TestBuildStore_Create(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs(build.ID, build.BatchID, build.Status). + WithArgs("monorepo", build.ID, build.BatchID, build.Status). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -137,7 +137,7 @@ func TestBuildStore_Create(t *testing.T) { name: "duplicate id returns ErrAlreadyExists", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs(build.ID, build.BatchID, build.Status). + WithArgs("monorepo", build.ID, build.BatchID, build.Status). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -147,7 +147,7 @@ func TestBuildStore_Create(t *testing.T) { name: "other exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs(build.ID, build.BatchID, build.Status). + WithArgs("monorepo", build.ID, build.BatchID, build.Status). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -192,7 +192,7 @@ func TestBuildStore_Update(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.Status, "monorepo", build.ID). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -200,7 +200,7 @@ func TestBuildStore_Update(t *testing.T) { name: "not found", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.Status, "monorepo", build.ID). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -210,7 +210,7 @@ func TestBuildStore_Update(t *testing.T) { name: "exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.Status, "monorepo", build.ID). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -219,7 +219,7 @@ func TestBuildStore_Update(t *testing.T) { name: "rows affected error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.Status, "monorepo", build.ID). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, diff --git a/submitqueue/extension/storage/mysql/request_batch_store.go b/submitqueue/extension/storage/mysql/request_batch_store.go index 4e4bb5cb..e96a0b10 100644 --- a/submitqueue/extension/storage/mysql/request_batch_store.go +++ b/submitqueue/extension/storage/mysql/request_batch_store.go @@ -46,8 +46,8 @@ func (s *requestBatchStore) GetByRequestID(ctx context.Context, requestID string defer func() { op.Complete(retErr) }() rows, err := s.db.QueryContext(ctx, - "SELECT request_id, batch_id, version FROM request_batch WHERE request_id = ? ORDER BY batch_id", - requestID, + "SELECT request_id, batch_id, version FROM request_batch WHERE queue = ? AND request_id = ? ORDER BY batch_id", + s.queue, requestID, ) if err != nil { return nil, fmt.Errorf("failed to get request batch associations requestID=%s: %w", requestID, err) @@ -73,8 +73,8 @@ func (s *requestBatchStore) Create(ctx context.Context, association entity.Reque defer func() { op.Complete(retErr) }() _, err := s.db.ExecContext(ctx, - "INSERT INTO request_batch (request_id, batch_id, version) VALUES (?, ?, ?)", - association.RequestID, association.BatchID, association.Version, + "INSERT INTO request_batch (queue, request_id, batch_id, version) VALUES (?, ?, ?, ?)", + s.queue, association.RequestID, association.BatchID, association.Version, ) if err != nil { var mysqlErr *mysql.MySQLError diff --git a/submitqueue/extension/storage/mysql/request_batch_store_test.go b/submitqueue/extension/storage/mysql/request_batch_store_test.go index d2c2f6ca..9496a3aa 100644 --- a/submitqueue/extension/storage/mysql/request_batch_store_test.go +++ b/submitqueue/extension/storage/mysql/request_batch_store_test.go @@ -58,7 +58,7 @@ func TestRequestBatchStore_GetByRequestID(t *testing.T) { "query fails": { setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT request_id, batch_id, version FROM request_batch"). - WithArgs(association1.RequestID). + WithArgs("monorepo", association1.RequestID). WillReturnError(fmt.Errorf("connection reset")) }, errMsg: "connection reset", @@ -66,7 +66,7 @@ func TestRequestBatchStore_GetByRequestID(t *testing.T) { "no associations": { setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT request_id, batch_id, version FROM request_batch"). - WithArgs(association1.RequestID). + WithArgs("monorepo", association1.RequestID). WillReturnRows(sqlmock.NewRows([]string{"request_id", "batch_id", "version"})) }, }, @@ -75,7 +75,7 @@ func TestRequestBatchStore_GetByRequestID(t *testing.T) { rows := sqlmock.NewRows([]string{"request_id", "batch_id", "version"}). AddRow(association1.RequestID, association1.BatchID, "invalid") mock.ExpectQuery("SELECT request_id, batch_id, version FROM request_batch"). - WithArgs(association1.RequestID). + WithArgs("monorepo", association1.RequestID). WillReturnRows(rows) }, errMsg: "failed to scan request batch association", @@ -86,7 +86,7 @@ func TestRequestBatchStore_GetByRequestID(t *testing.T) { AddRow(association1.RequestID, association1.BatchID, association1.Version). RowError(0, storeErr) mock.ExpectQuery("SELECT request_id, batch_id, version FROM request_batch"). - WithArgs(association1.RequestID). + WithArgs("monorepo", association1.RequestID). WillReturnRows(rows) }, errMsg: storeErr.Error(), @@ -97,7 +97,7 @@ func TestRequestBatchStore_GetByRequestID(t *testing.T) { AddRow(association1.RequestID, association1.BatchID, association1.Version). AddRow(association2.RequestID, association2.BatchID, association2.Version) mock.ExpectQuery("SELECT request_id, batch_id, version FROM request_batch"). - WithArgs(association1.RequestID). + WithArgs("monorepo", association1.RequestID). WillReturnRows(rows) }, want: []entity.RequestBatch{association1, association2}, @@ -135,7 +135,7 @@ func TestRequestBatchStore_Create(t *testing.T) { "duplicate association returns already exists": { setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request_batch"). - WithArgs(association.RequestID, association.BatchID, association.Version). + WithArgs("monorepo", association.RequestID, association.BatchID, association.Version). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, errMsg: storage.ErrAlreadyExists.Error(), @@ -143,7 +143,7 @@ func TestRequestBatchStore_Create(t *testing.T) { "insert fails": { setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request_batch"). - WithArgs(association.RequestID, association.BatchID, association.Version). + WithArgs("monorepo", association.RequestID, association.BatchID, association.Version). WillReturnError(fmt.Errorf("connection reset")) }, errMsg: "connection reset", @@ -151,7 +151,7 @@ func TestRequestBatchStore_Create(t *testing.T) { "success": { setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO request_batch"). - WithArgs(association.RequestID, association.BatchID, association.Version). + WithArgs("monorepo", association.RequestID, association.BatchID, association.Version). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, diff --git a/submitqueue/extension/storage/mysql/request_store.go b/submitqueue/extension/storage/mysql/request_store.go index 605dc29f..8671de95 100644 --- a/submitqueue/extension/storage/mysql/request_store.go +++ b/submitqueue/extension/storage/mysql/request_store.go @@ -51,8 +51,8 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request, var changeURIsJSON []byte err := r.db.QueryRowContext(ctx, - "SELECT id, queue, change_uri, land_strategy, state, version FROM request WHERE id = ?", - id, + "SELECT id, queue, change_uri, land_strategy, state, version FROM request WHERE queue = ? AND id = ?", + r.queue, id, ).Scan(&req.ID, &req.Queue, &changeURIsJSON, &req.LandStrategy, &req.State, &req.Version) if errors.Is(err, sql.ErrNoRows) { @@ -116,8 +116,8 @@ func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVe } result, err := r.db.ExecContext(ctx, - "UPDATE request SET queue = ?, change_uri = ?, land_strategy = ?, state = ?, version = ? WHERE id = ? AND version = ?", - request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion, + "UPDATE request SET change_uri = ?, land_strategy = ?, state = ?, version = ? WHERE queue = ? AND id = ? AND version = ?", + changeURIsJSON, request.LandStrategy, request.State, newVersion, request.Queue, request.ID, oldVersion, ) if err != nil { return fmt.Errorf( diff --git a/submitqueue/extension/storage/mysql/request_store_test.go b/submitqueue/extension/storage/mysql/request_store_test.go index 7f4ef711..80c05a59 100644 --- a/submitqueue/extension/storage/mysql/request_store_test.go +++ b/submitqueue/extension/storage/mysql/request_store_test.go @@ -70,7 +70,7 @@ func TestRequestStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"id", "queue", "change_uri", "land_strategy", "state", "version"}). AddRow(want.ID, want.Queue, changeURIsJSON, string(want.LandStrategy), string(want.State), want.Version) mock.ExpectQuery("SELECT id, queue, change_uri, land_strategy, state, version FROM request"). - WithArgs(want.ID). + WithArgs("monorepo", want.ID). WillReturnRows(rows) }, want: want, @@ -80,7 +80,7 @@ func TestRequestStore_Get(t *testing.T) { id: "missing", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT id, queue, change_uri, land_strategy, state, version FROM request"). - WithArgs("missing"). + WithArgs("monorepo", "missing"). WillReturnError(sql.ErrNoRows) }, wantErr: true, @@ -91,7 +91,7 @@ func TestRequestStore_Get(t *testing.T) { id: "bad", setup: func(mock sqlmock.Sqlmock) { mock.ExpectQuery("SELECT id, queue, change_uri, land_strategy, state, version FROM request"). - WithArgs("bad"). + WithArgs("monorepo", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -103,7 +103,7 @@ func TestRequestStore_Get(t *testing.T) { rows := sqlmock.NewRows([]string{"id", "queue", "change_uri", "land_strategy", "state", "version"}). AddRow(want.ID, want.Queue, []byte("not json"), string(want.LandStrategy), string(want.State), want.Version) mock.ExpectQuery("SELECT id, queue, change_uri, land_strategy, state, version FROM request"). - WithArgs("malformed"). + WithArgs("monorepo", "malformed"). WillReturnRows(rows) }, wantErr: true, @@ -223,7 +223,7 @@ func TestRequestStore_Update(t *testing.T) { request: request, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion). + WithArgs(changeURIsJSON, request.LandStrategy, request.State, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -232,7 +232,7 @@ func TestRequestStore_Update(t *testing.T) { request: request, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion). + WithArgs(changeURIsJSON, request.LandStrategy, request.State, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -243,7 +243,7 @@ func TestRequestStore_Update(t *testing.T) { request: request, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion). + WithArgs(changeURIsJSON, request.LandStrategy, request.State, newVersion, request.Queue, request.ID, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -253,7 +253,7 @@ func TestRequestStore_Update(t *testing.T) { request: request, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion). + WithArgs(changeURIsJSON, request.LandStrategy, request.State, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, @@ -263,7 +263,7 @@ func TestRequestStore_Update(t *testing.T) { request: entity.Request{ID: request.ID, Queue: request.Queue, LandStrategy: request.LandStrategy, State: request.State, Version: request.Version}, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.Queue, []byte("null"), request.LandStrategy, request.State, newVersion, request.ID, oldVersion). + WithArgs([]byte("null"), request.LandStrategy, request.State, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -279,7 +279,7 @@ func TestRequestStore_Update(t *testing.T) { }, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request"). - WithArgs(request.Queue, []byte("[]"), request.LandStrategy, request.State, newVersion, request.ID, oldVersion). + WithArgs([]byte("[]"), request.LandStrategy, request.State, newVersion, request.Queue, request.ID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, diff --git a/submitqueue/extension/storage/mysql/schema/README.md b/submitqueue/extension/storage/mysql/schema/README.md index 2e7ee204..2240a7e1 100644 --- a/submitqueue/extension/storage/mysql/schema/README.md +++ b/submitqueue/extension/storage/mysql/schema/README.md @@ -1,8 +1,14 @@ # MySQL Schema +## Queue-leading primary keys + +Every queue-scoped 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)`, 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 the tables are shardable by queue. The `build` key also removes a cross-queue uniqueness assumption: build IDs are runner-minted, so two queues sharing one CI pipeline may legitimately mint the same identifier. + +The global read-model tables (`request_summary`, `request_log`, `change_uri_request_mapping`) keep queue-free keys — their lookups start from identifiers that arrive without queue context. + ## batch table -The `batch` table is keyed by `id` alone and carries no secondary index. Listing a queue's batches by state goes through the `queue_batch_state` table instead, so batch reads and writes stay pure primary-key operations. +The `batch` table is keyed by `(queue, id)` and carries no secondary index. Listing a queue's batches by state goes through the `queue_batch_state` table instead, so batch reads and writes stay pure primary-key operations. ## queue_batch_state table diff --git a/submitqueue/extension/storage/mysql/schema/batch.sql b/submitqueue/extension/storage/mysql/schema/batch.sql index 0b12e792..9bf5d187 100644 --- a/submitqueue/extension/storage/mysql/schema/batch.sql +++ b/submitqueue/extension/storage/mysql/schema/batch.sql @@ -1,9 +1,9 @@ CREATE TABLE IF NOT EXISTS batch ( - id VARCHAR(255) NOT NULL, queue VARCHAR(255) NOT NULL, + id VARCHAR(255) NOT NULL, contains JSON NOT NULL, dependencies JSON NOT NULL, - state VARCHAR(255) NOT NUll, + state VARCHAR(255) NOT NULL, version INT NOT NULL, - PRIMARY KEY (id) + PRIMARY KEY (queue, id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/schema/batch_dependent.sql b/submitqueue/extension/storage/mysql/schema/batch_dependent.sql index c7023cfa..c1a6301c 100644 --- a/submitqueue/extension/storage/mysql/schema/batch_dependent.sql +++ b/submitqueue/extension/storage/mysql/schema/batch_dependent.sql @@ -1,6 +1,7 @@ CREATE TABLE IF NOT EXISTS batch_dependent ( + queue VARCHAR(255) NOT NULL, batch_id VARCHAR(255) NOT NULL, dependents JSON NOT NULL, version INT NOT NULL, - PRIMARY KEY (batch_id) + PRIMARY KEY (queue, batch_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/schema/build.sql b/submitqueue/extension/storage/mysql/schema/build.sql index f720397f..9cc0a83a 100644 --- a/submitqueue/extension/storage/mysql/schema/build.sql +++ b/submitqueue/extension/storage/mysql/schema/build.sql @@ -1,6 +1,7 @@ CREATE TABLE IF NOT EXISTS build ( + queue VARCHAR(255) NOT NULL, id VARCHAR(255) NOT NULL, batch_id VARCHAR(255) NOT NULL, status VARCHAR(64) NOT NULL, - PRIMARY KEY (id) + PRIMARY KEY (queue, id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/schema/request.sql b/submitqueue/extension/storage/mysql/schema/request.sql index e19a575f..937b6db3 100644 --- a/submitqueue/extension/storage/mysql/schema/request.sql +++ b/submitqueue/extension/storage/mysql/schema/request.sql @@ -1,9 +1,9 @@ CREATE TABLE IF NOT EXISTS request ( - id VARCHAR(255) NOT NULL, queue VARCHAR(255) NOT NULL, + id VARCHAR(255) NOT NULL, change_uri JSON NOT NULL, land_strategy VARCHAR(64) NOT NULL, state VARCHAR(64) NOT NULL, version INT NOT NULL, - PRIMARY KEY (id) + PRIMARY KEY (queue, id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/schema/request_batch.sql b/submitqueue/extension/storage/mysql/schema/request_batch.sql index f2d261ab..2fee7e03 100644 --- a/submitqueue/extension/storage/mysql/schema/request_batch.sql +++ b/submitqueue/extension/storage/mysql/schema/request_batch.sql @@ -1,6 +1,7 @@ CREATE TABLE IF NOT EXISTS request_batch ( + queue VARCHAR(255) NOT NULL, request_id VARCHAR(255) NOT NULL, batch_id VARCHAR(255) NOT NULL, version INT NOT NULL, - PRIMARY KEY (request_id, batch_id) + PRIMARY KEY (queue, request_id, batch_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel index 40586c31..a714d542 100644 --- a/test/e2e/submitqueue/BUILD.bazel +++ b/test/e2e/submitqueue/BUILD.bazel @@ -29,7 +29,6 @@ go_test( "//platform/extension/consumergate:go_default_library", "//platform/extension/consumergate/file:go_default_library", "//submitqueue/entity:go_default_library", - "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index d8fb81c2..1b6cde82 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/test/e2e/submitqueue/harness_test.go @@ -218,9 +218,11 @@ func (s *E2EIntegrationSuite) awaitUnparked(consumerGroup, topic, messageID stri // operating store (mysql-app). Unlike the status timeline, RequestState is // point-in-time — the Request entity is updated in place under optimistic // locking, so only the current (terminal, once settled) value is observable. -func (s *E2EIntegrationSuite) terminalState(sqid string) entity.RequestState { +func (s *E2EIntegrationSuite) terminalState(queue, sqid string) entity.RequestState { t := s.T() - req, err := s.requestStore.Get(s.ctx, sqid) + store, err := s.appStorage.For(queue) + require.NoError(t, err, "failed to resolve operating store for queue %s", queue) + req, err := store.GetRequestStore().Get(s.ctx, sqid) require.NoError(t, err, "failed to get request %s from operating store", sqid) return req.State } diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index e1b40cf8..f77ad84d 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -43,7 +43,6 @@ import ( orchestratorpb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb" consumergatefile "github.com/uber/submitqueue/platform/extension/consumergate/file" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" storagemysql "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" "google.golang.org/grpc" @@ -60,7 +59,7 @@ type E2EIntegrationSuite struct { orchestratorClient orchestratorpb.SubmitQueueOrchestratorClient db *sql.DB // App database queueDB *sql.DB // Queue database - requestStore storage.RequestStore // White-box view of the internal RequestState (app DB) + appStorage *storagemysql.Storage // White-box view of the operating store (app DB), resolved per queue gate *consumergatefile.Store // Consumer-gate control plane (shared dir bind-mounted into services) } @@ -135,9 +134,8 @@ func (s *E2EIntegrationSuite) SetupSuite() { s.log.Logf("Schemas applied successfully") // White-box handle on the operating store for point-in-time RequestState. - // Reads are not yet queue-filtered, so a single bound instance serves every - // e2e queue's point-in-time RequestState checks. - s.requestStore = storagemysql.NewRequestStore(s.db, tally.NoopScope, "e2e-test-queue") + s.appStorage, err = storagemysql.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err, "failed to create app storage backend") // Connect to Gateway gRPC service var gatewayConn *grpc.ClientConn @@ -250,7 +248,7 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() { // White-box (internal state): the operating store's authoritative // RequestState settled on landed. RequestState is point-in-time, so this is a // terminal check, not a sequence. - assert.Equal(s.T(), entity.RequestStateLanded, s.terminalState(sqid), + assert.Equal(s.T(), entity.RequestStateLanded, s.terminalState("e2e-test-queue", sqid), "operating store should show request %s in terminal state landed", sqid) } @@ -398,7 +396,7 @@ func (s *E2EIntegrationSuite) TestCancel_CaughtPreBatch_NeverLands() { entity.RequestStatusCancelling, entity.RequestStatusCancelled, ) - assert.Equal(t, entity.RequestStateCancelled, s.terminalState(sqid), + assert.Equal(t, entity.RequestStateCancelled, s.terminalState(queue, sqid), "operating store should show request %s terminal cancelled while its check is parked", sqid) // Start the controller again and prove the parked delivery cleared the gate. @@ -411,7 +409,7 @@ func (s *E2EIntegrationSuite) TestCancel_CaughtPreBatch_NeverLands() { s.awaitStatus(sentinel, entity.RequestStatusLanded) // The stale check answer was dropped: the cancelled request never advanced. - assert.Equal(t, entity.RequestStateCancelled, s.terminalState(sqid), + assert.Equal(t, entity.RequestStateCancelled, s.terminalState(queue, sqid), "request %s must stay terminal cancelled after its stale check signal is processed", sqid) s.assertStatusesNever(sqid, entity.RequestStatusBatched, entity.RequestStatusLanded) } diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go index 1d46afc9..0abe4751 100644 --- a/test/integration/submitqueue/extension/storage/suite.go +++ b/test/integration/submitqueue/extension/storage/suite.go @@ -411,6 +411,51 @@ func (s *StorageContractSuite) TestStorage_QueueBatchStateRecordLifecycle() { assert.ElementsMatch(t, []entity.QueueBatchState{otherQueue}, got) } +// TestStorage_QueueIsolation verifies a store aggregate bound to one queue can +// neither read nor write another queue's records: cross-queue reads miss, and +// writes whose entity queue disagrees with the binding are rejected. +func (s *StorageContractSuite) TestStorage_QueueIsolation() { + t := s.T() + ctx := s.ctx + storeA := s.forQueue("iso-queue-a") + storeB := s.forQueue("iso-queue-b") + + request := entity.Request{ID: "iso-a/1", Queue: "iso-queue-a", State: entity.RequestStateStarted, LandStrategy: mergestrategy.MergeStrategyMerge, Version: 1} + require.NoError(t, storeA.GetRequestStore().Create(ctx, request)) + _, err := storeB.GetRequestStore().Get(ctx, request.ID) + require.ErrorIs(t, err, storage.ErrNotFound, "a request must be invisible through another queue's binding") + require.Error(t, storeB.GetRequestStore().Create(ctx, request), "a mismatched-queue write must be rejected") + + batch := entity.Batch{ID: "iso-a/batch/1", Queue: "iso-queue-a", State: entity.BatchStateCreated, Version: 1} + require.NoError(t, storeA.GetBatchStore().Create(ctx, batch)) + _, err = storeB.GetBatchStore().Get(ctx, batch.ID) + require.ErrorIs(t, err, storage.ErrNotFound) + + // Builds are keyed by a runner-minted ID; the queue-leading key removes the + // cross-queue uniqueness assumption, so the same runner ID coexists per queue. + build := entity.Build{ID: "runner/iso/1", BatchID: batch.ID, Status: entity.BuildStatusRunning} + require.NoError(t, storeA.GetBuildStore().Create(ctx, build)) + _, err = storeB.GetBuildStore().Get(ctx, build.ID) + require.ErrorIs(t, err, storage.ErrNotFound) + require.NoError(t, storeB.GetBuildStore().Create(ctx, build), "the same runner-minted build ID must coexist across queues") + + dependent := entity.BatchDependent{BatchID: batch.ID, Dependents: []string{}, Version: 1} + require.NoError(t, storeA.GetBatchDependentStore().Create(ctx, dependent)) + _, err = storeB.GetBatchDependentStore().Get(ctx, batch.ID) + require.ErrorIs(t, err, storage.ErrNotFound) + + association := entity.RequestBatch{RequestID: request.ID, BatchID: batch.ID, Version: 1} + require.NoError(t, storeA.GetRequestBatchStore().Create(ctx, association)) + crossQueue, err := storeB.GetRequestBatchStore().GetByRequestID(ctx, request.ID) + require.NoError(t, err) + assert.Empty(t, crossQueue, "associations must be invisible through another queue's binding") + + // The owning queue still sees everything it wrote. + fromA, err := storeA.GetRequestStore().Get(ctx, request.ID) + require.NoError(t, err) + assert.Equal(t, request, fromA) +} + // TestStorage_NotFound tests getting a non-existent request func (s *StorageContractSuite) TestStorage_NotFound() { t := s.T()