Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
11 changes: 10 additions & 1 deletion apps/API.md
Original file line number Diff line number Diff line change
Expand Up @@ -827,7 +827,16 @@ Commerce, which does not currently call them.

Admission applies the plan value to the organization's non-finalized Box total.
`STOPPED` and `UNKNOWN` Boxes count, while `ERROR`, `DESTROYING`, `DESTROYED`,
`ARCHIVING`, and `ARCHIVED` do not. The billing client is hand-written rather
`ARCHIVING`, and `ARCHIVED` do not. Concurrent creates for one organization
queue on a transaction-scoped advisory lock keyed by the organization, then
count and write under `READ COMMITTED`, so a burst waits instead of failing;
a wait longer than 5 s returns `503` with `Retry-After: 5`
([`box.repository.ts`](./api/src/box/repositories/box.repository.ts)).
A queued create holds a pooled database connection while it waits. At the
default pool of 10, a local 100-create burst finished in under 1 s, and other
queries on that pool waited up to 0.9 s. A stuck lock holder can pin the pool
for the full 5 s timeout.
The billing client is hand-written rather
than generated, so neither side detects its divergence. Five of its 20 routes —
the billing-email group — have no
implementation in Commerce, of which only `email/verify` has a live caller here
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

import { randomUUID } from 'node:crypto'
import { DataSource, EntityManager, Repository } from 'typeorm'
import { IsolationLevel } from 'typeorm/driver/types/IsolationLevel'
import { CustomNamingStrategy } from '../../common/utils/naming-strategy.util'
import { BOX_WARM_POOL_UNASSIGNED_ORGANIZATION } from '../constants/box.constants'
import { Box } from '../entities/box.entity'
Expand Down Expand Up @@ -34,28 +35,44 @@ function newBox(name: string, owner = organizationId): Box {
return box
}

// Waits until another admission for the organization is queued on its advisory
// lock. Gives up after a grace period, so an admission that never queues still
// races the paused one instead of hanging the test.
async function waitForQueuedAdmission(dataSource: DataSource, owner: string): Promise<void> {
const deadline = Date.now() + 2000
while (Date.now() < deadline) {
const [{ waiting }] = await dataSource.query(
`SELECT count(*)::int AS waiting FROM pg_locks
WHERE locktype = 'advisory' AND NOT granted AND objid::bigint = (hashtext($1)::bigint & 4294967295)`,
[owner],
)
if (waiting > 0) {
return
}
await new Promise((resolve) => setTimeout(resolve, 10))
}
}

// Pauses the first admission right after its count until a competing admission
// is waiting, so both compete for the final slot.
function installInitialCountBarrier(dataSource: DataSource): jest.SpyInstance {
const transaction = dataSource.transaction.bind(dataSource)
let initialCountCalls = 0
let releaseBarrier!: () => void
const barrier = new Promise<void>((resolve) => (releaseBarrier = resolve))
let paused = false

return jest.spyOn(dataSource, 'transaction').mockImplementation(async (...args: unknown[]) => {
if (args[0] !== 'SERIALIZABLE' || typeof args[1] !== 'function') {
if (typeof args[0] !== 'string' || typeof args[1] !== 'function') {
return transaction(args[0] as (entityManager: EntityManager) => Promise<unknown>)
}

const isolation = args[0] as IsolationLevel
const callback = args[1] as (entityManager: EntityManager) => Promise<unknown>
return transaction('SERIALIZABLE', async (entityManager) => {
return transaction(isolation, async (entityManager) => {
const count = entityManager.count.bind(entityManager)
entityManager.count = async (...countArgs: Parameters<EntityManager['count']>) => {
const result = await count(...countArgs)
initialCountCalls++
if (initialCountCalls <= 2) {
if (initialCountCalls === 2) {
releaseBarrier()
}
await barrier
if (!paused) {
paused = true
await waitForQueuedAdmission(dataSource, organizationId)
}
return result
}
Expand Down Expand Up @@ -161,6 +178,17 @@ describeIfDatabase('BoxRepository creation admission (integration, real Postgres
}
})

it('admits a same-organization burst up to the limit without contention failures', async () => {
const burst = 50
const results = await Promise.allSettled(
Array.from({ length: burst }, (_, index) => repository.insert(newBox(`burst-${index}`), burst)),
)

const failures = results.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
expect(failures.map((failure) => (failure.reason as Error).name)).toEqual([])
expect(await boxes.countBy({ organizationId })).toBe(burst)
})

it.each(Object.values(BoxState))('applies the creation-count policy to state %s', async (state) => {
const existing = newBox(`existing-${state}`)
existing.state = state
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ type TransactionCallback = (entityManager: ReturnType<typeof makeEntityManager>)

function makeEntityManager(currentCount = 0) {
return {
query: jest.fn().mockResolvedValue([]),
count: jest.fn().mockResolvedValue(currentCount),
insert: jest.fn().mockResolvedValue(undefined),
update: jest.fn().mockResolvedValue({ affected: 1 }),
Expand Down Expand Up @@ -59,13 +60,20 @@ function responseCode(error: BoxCreationLimitExceededError | BoxCreationAdmissio
}

describe('BoxRepository creation admission', () => {
it('counts and inserts with one SERIALIZABLE transactional manager', async () => {
it('counts and inserts under the organization advisory lock in one READ COMMITTED transaction', async () => {
const { boxRepository, entityManager, transaction, cacheInvalidation } = makeRepository(1)
const entity = box()

await expect(boxRepository.insert(entity, 2)).resolves.toBe(entity)

expect(transaction).toHaveBeenCalledWith('SERIALIZABLE', expect.any(Function))
expect(transaction).toHaveBeenCalledWith('READ COMMITTED', expect.any(Function))
expect(entityManager.query).toHaveBeenCalledWith('SELECT pg_advisory_xact_lock(hashtext($1))', [
entity.organizationId,
])
const lockCall = entityManager.query.mock.calls.findIndex(([sql]) => String(sql).includes('pg_advisory_xact_lock'))
expect(entityManager.query.mock.invocationCallOrder[lockCall]).toBeLessThan(
entityManager.count.mock.invocationCallOrder[0],
)
expect(entityManager.count).toHaveBeenCalledWith(
Box,
expect.objectContaining({ where: expect.objectContaining({ organizationId: entity.organizationId }) }),
Expand Down Expand Up @@ -115,7 +123,8 @@ describe('BoxRepository creation admission', () => {
maxCreatedBoxes: 1,
})

expect(transaction).toHaveBeenCalledWith('SERIALIZABLE', expect.any(Function))
expect(transaction).toHaveBeenCalledWith('READ COMMITTED', expect.any(Function))
expect(entityManager.query).toHaveBeenCalledWith('SELECT pg_advisory_xact_lock(hashtext($1))', [organizationId])
expect(entityManager.count).toHaveBeenCalledWith(
Box,
expect.objectContaining({ where: expect.objectContaining({ organizationId }) }),
Expand All @@ -124,31 +133,13 @@ describe('BoxRepository creation admission', () => {
expect(entityManager.upsert).toHaveBeenCalledTimes(1)
})

it('retries the complete count-and-write transaction after SQLSTATE 40001', async () => {
it('returns a retryable 503 when the admission lock wait times out', async () => {
const { boxRepository, entityManager, transaction } = makeRepository(0)
let attempts = 0
transaction.mockImplementation(async (_isolation: string, callback: TransactionCallback) => {
const result = await callback(entityManager)
attempts++
if (attempts < 3) {
throw Object.assign(new Error('could not serialize access'), { code: '40001' })
entityManager.query.mockImplementation(async (sql: string) => {
if (sql.includes('pg_advisory_xact_lock')) {
throw Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' })
}
return result
})

await boxRepository.insert(box(), 3)

expect(transaction).toHaveBeenCalledTimes(3)
expect(entityManager.count).toHaveBeenCalledTimes(3)
expect(entityManager.insert).toHaveBeenCalledTimes(3)
expect(entityManager.upsert).toHaveBeenCalledTimes(3)
})

it('returns a retryable 503 after five serialization failures', async () => {
const { boxRepository, entityManager, transaction } = makeRepository(0)
transaction.mockImplementation(async (_isolation: string, callback: TransactionCallback) => {
await callback(entityManager)
throw Object.assign(new Error('could not serialize access'), { code: '40001' })
return []
})

const error = await boxRepository.insert(box(), 3).catch((caught) => caught)
Expand All @@ -157,23 +148,25 @@ describe('BoxRepository creation admission', () => {
expect(responseCode(error)).toBe(BOX_CREATION_ADMISSION_UNAVAILABLE_CODE)
expect(error.getStatus()).toBe(503)
expect((error as unknown as { retryAfterSeconds?: number }).retryAfterSeconds).toBe(5)
expect(transaction).toHaveBeenCalledTimes(5)
expect(transaction).toHaveBeenCalledTimes(1)
expect(entityManager.insert).not.toHaveBeenCalled()
})

it('does not retry database errors other than serialization failures', async () => {
it('propagates other database errors without retrying', async () => {
const { boxRepository, transaction } = makeRepository(0)
transaction.mockRejectedValue(Object.assign(new Error('connection lost'), { code: '08006' }))

await expect(boxRepository.insert(box(), 3)).rejects.toThrow('connection lost')
expect(transaction).toHaveBeenCalledTimes(1)
})

it('preserves the existing non-serializable transaction when admission is disabled', async () => {
it('preserves the plain transaction without the admission lock when admission is disabled', async () => {
const { boxRepository, transaction, entityManager } = makeRepository()

await boxRepository.insert(box())

expect(transaction).toHaveBeenCalledWith(expect.any(Function))
expect(entityManager.query).not.toHaveBeenCalled()
expect(entityManager.count).not.toHaveBeenCalled()
})
})
83 changes: 28 additions & 55 deletions apps/api/src/box/repositories/box.repository.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,14 +42,12 @@ const PG_LOCK_TIMEOUT_CODE = '55P03'
// boxlite-proxy.controller.ts (PROXY_START_HINT_TIMEOUT_MS).
const PROXY_START_LOCK_TIMEOUT_MS = 2000

// SERIALIZABLE protects the count-and-write admission decision without a
// per-organization mutex. PostgreSQL aborts one conflicting transaction with
// SQLSTATE 40001; retry the whole decision, never only the write.
const PG_SERIALIZATION_FAILURE_CODE = '40001'
const CREATION_ADMISSION_MAX_ATTEMPTS = 5
const CREATION_ADMISSION_BASE_BACKOFF_MS = 5
const CREATION_ADMISSION_MAX_BACKOFF_MS = 40
const CREATION_ADMISSION_JITTER_MS = 5
// Same-organization admissions queue on a transaction-scoped advisory lock
// keyed by the organization, so a burst waits instead of aborting. The lock
// must come before the count, under READ COMMITTED: each statement then reads
// a fresh snapshot that includes the previous holder's committed Box. Cap the
// wait, since a queued admission pins a pooled connection.
const CREATION_ADMISSION_LOCK_TIMEOUT_MS = 5000
const CREATION_ADMISSION_RETRY_AFTER_SECONDS = 5

const BOX_CREATION_LIMIT_EXCLUDED_STATES = [
Expand Down Expand Up @@ -204,58 +202,33 @@ export class BoxRepository extends BaseRepository<Box> {
return this.dataSource.transaction(persist)
}

for (let attempt = 0; attempt < CREATION_ADMISSION_MAX_ATTEMPTS; attempt++) {
try {
return await this.dataSource.transaction('SERIALIZABLE', async (entityManager) => {
const currentCount = await entityManager.count(Box, {
where: {
organizationId,
state: Not(In(BOX_CREATION_LIMIT_EXCLUDED_STATES)),
},
})
if (currentCount >= maxCreatedBoxes) {
throw new BoxCreationLimitExceededError(currentCount, maxCreatedBoxes)
}

return persist(entityManager)
try {
return await this.dataSource.transaction('READ COMMITTED', async (entityManager) => {
// SET LOCAL takes a literal, not a bind parameter; the value is a constant.
await entityManager.query(`SET LOCAL lock_timeout = '${CREATION_ADMISSION_LOCK_TIMEOUT_MS}ms'`)
await entityManager.query('SELECT pg_advisory_xact_lock(hashtext($1))', [organizationId])

const currentCount = await entityManager.count(Box, {
where: {
organizationId,
state: Not(In(BOX_CREATION_LIMIT_EXCLUDED_STATES)),
},
})
} catch (error) {
if (!this.isSerializationFailure(error)) {
throw error
}
if (attempt === CREATION_ADMISSION_MAX_ATTEMPTS - 1) {
throw new BoxCreationAdmissionUnavailableError(
'Box creation admission remained contended; retry the request',
CREATION_ADMISSION_RETRY_AFTER_SECONDS,
)
if (currentCount >= maxCreatedBoxes) {
throw new BoxCreationLimitExceededError(currentCount, maxCreatedBoxes)
}

await this.waitBeforeAdmissionRetry(attempt)
return persist(entityManager)
})
} catch (error) {
if ((error as { code?: string }).code === PG_LOCK_TIMEOUT_CODE) {
throw new BoxCreationAdmissionUnavailableError(
'Box creation admission remained contended; retry the request',
CREATION_ADMISSION_RETRY_AFTER_SECONDS,
)
}
throw error
}

throw new BoxCreationAdmissionUnavailableError(
'Box creation admission remained contended; retry the request',
CREATION_ADMISSION_RETRY_AFTER_SECONDS,
)
}

private isSerializationFailure(error: unknown): boolean {
return (
typeof error === 'object' &&
error !== null &&
'code' in error &&
(error as { code?: unknown }).code === PG_SERIALIZATION_FAILURE_CODE
)
}

private async waitBeforeAdmissionRetry(attempt: number): Promise<void> {
const exponentialBackoff = Math.min(
CREATION_ADMISSION_BASE_BACKOFF_MS * 2 ** attempt,
CREATION_ADMISSION_MAX_BACKOFF_MS,
)
const jitter = Math.floor(Math.random() * (CREATION_ADMISSION_JITTER_MS + 1))
await new Promise((resolve) => setTimeout(resolve, exponentialBackoff + jitter))
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import { ConflictException } from '@nestjs/common'
import { randomUUID } from 'node:crypto'
import { DataSource, EntityManager } from 'typeorm'
import { IsolationLevel } from 'typeorm/driver/types/IsolationLevel'
import { Box } from '../../box/entities/box.entity'
import { BoxLastActivity } from '../../box/entities/box-last-activity.entity'
import { BoxMigration } from '../../box/entities/box-migration.entity'
Expand Down Expand Up @@ -218,7 +219,7 @@ describeIfDatabase('Registry credential deletion concurrency (integration, real
}
return persist(manager)
}
return args[0] === 'SERIALIZABLE' ? transaction('SERIALIZABLE', gated) : transaction(gated)
return typeof args[0] === 'string' ? transaction(args[0] as IsolationLevel, gated) : transaction(gated)
})
const creating = outcome(create(limit))
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,15 +4,15 @@ Serialize registry credential deletion with private-image box persistence so a c

## Scope

- Coordinate `RegistriesService.delete` and `BoxRepository.insert`, including generated-name retries and optional SERIALIZABLE box-limit admission.
- Coordinate `RegistriesService.delete` and `BoxRepository.insert`, including generated-name retries and optional box-limit admission.
- Preserve longest-prefix credential matching, curated/public image behavior, and the existing 409 refusal for credentials used by live boxes.
- Cover both operation orderings with real PostgreSQL transactions before changing production code.
- Secret-store/database crash atomicity and catalog-entry deletion are separate concerns.

## Related work and lessons

- At `7e41ef35`, `RegistriesService.delete` checks dependent boxes before destroying the password, without a transaction. `BoxService.create` resolves the proxy reference before placement and `BoxRepository.insert`; neither operation locks credentials. A create can resolve the existing login and persist after deletion checked for boxes.
- PostgreSQL [row locks](https://www.postgresql.org/docs/current/explicit-locking.html#LOCKING-ROWS) last until transaction end. Shared row locks permit concurrent creators but conflict with the exclusive lock required by deletion. Under SERIALIZABLE, locking a row deleted since the snapshot raises a serialization failure, which existing box admission retries.
- PostgreSQL [row locks](https://www.postgresql.org/docs/current/explicit-locking.html#LOCKING-ROWS) last until transaction end. Shared row locks permit concurrent creators but conflict with the exclusive lock required by deletion. Box-limit admission runs under READ COMMITTED (POL-822), where each statement reads a fresh snapshot, so the shared-lock read sees a credential deleted before it as missing, and creation returns 409.
- TypeORM [transactions](https://typeorm.io/docs/advanced-topics/transactions/) require transaction-scoped managers for participating queries. Dependency checks and credential deletion must use the same manager holding the lock.
- A lock only on deletion leaves creation uncoordinated. A process mutex fails across API replicas. Holding a lock from image resolution through runner placement unnecessarily keeps a database connection across unrelated work. Rechecking the proxy reference at the existing box-persistence transaction is the smallest durable boundary.

Expand All @@ -36,7 +36,7 @@ argument or an injection into the deletion service.
| --- | --- | --- |
| Deletion first | Pause secret destruction after the dependent-box query; attempt a real box create | Creation returns 409, no box is saved, deletion can finish |
| Creation first | Pause the box transaction after INSERT and before commit; delete the credential | Deletion returns 409 without destroying the secret; creation commits |
| Resolution before completed deletion | Pause runner placement after the real resolver returns a proxy ref; delete, then resume creation | Creation returns 409, including with SERIALIZABLE admission |
| Resolution before completed deletion | Pause runner placement after the real resolver returns a proxy ref; delete, then resume creation | Creation returns 409, including with box-limit admission |
| Existing dependent box | Create and commit, then delete | 409 identifies the dependent box |
| Curated images | Create without a proxy ref after credential deletion | Existing behavior is preserved |

Expand Down
Loading