Skip to content

feat(storage): queue-leading primary keys with row-level cross-queue isolation #537

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
behinddwalls merged 1 commit into from
Aug 6, 2026
Merged

feat(storage): queue-leading primary keys with row-level cross-queue isolation #537

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions submitqueue/extension/storage/mysql/batch_dependent_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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
Expand All @@ -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(
Expand Down
28 changes: 14 additions & 14 deletions submitqueue/extension/storage/mysql/batch_dependent_store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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{
Expand All @@ -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{
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -163,15 +163,15 @@ 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))
},
},
{
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,
Expand All @@ -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,
Expand Down Expand Up @@ -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))
},
},
Expand All @@ -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))
},
},
Expand All @@ -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))
},
},
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand Down
8 changes: 4 additions & 4 deletions submitqueue/extension/storage/mysql/batch_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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(
Expand Down
20 changes: 10 additions & 10 deletions submitqueue/extension/storage/mysql/batch_store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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))
},
},
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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))
},
},
Expand All @@ -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))
},
},
Expand Down
12 changes: 6 additions & 6 deletions submitqueue/extension/storage/mysql/build_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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
Expand All @@ -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)
Expand Down
20 changes: 10 additions & 10 deletions submitqueue/extension/storage/mysql/build_store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -129,15 +129,15 @@ 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))
},
},
{
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,
Expand All @@ -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,
Expand Down Expand Up @@ -192,15 +192,15 @@ 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))
},
},
{
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,
Expand All @@ -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,
Expand All @@ -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,
Expand Down
Loading
Loading