test(relay): load generator can send two concurrent assigns per host (--duplicate-assign-probes) (#27107)

Rebased onto main as one commit; kept both sides of the package.json test-list conflict with #27096.
This commit is contained in:
Jinwoo Hong
2026-10-10 20:06:42 -04:00
committed by GitHub
parent c2da6337b4
commit 6404b7e83d
8 changed files with 309 additions and 7 deletions
+47 -1
View File
@@ -20,6 +20,7 @@ import {
waitForRelayLoadRebindGate
} from './relay-load-rebind-boundary.mjs'
import { proveRelayLoadRegionBehavior } from './relay-load-region-behavior.mjs'
import { proveRelayLoadDuplicateAssign } from './relay-load-duplicate-assign.mjs'
import {
openRelayLoadInviteOffers,
proveRelayLoadRequestUnitBoundary
@@ -95,6 +96,7 @@ function report(state, final = false) {
regionalFallbacksProved: state.regionalFallbacksProved,
oldClientUsFirstProved: state.oldClientUsFirstProved,
stickyAssignmentProved: state.stickyAssignmentProved,
duplicateAssign: state.duplicateAssign,
requestUnitInvitesOpened: state.requestUnitInvitesOpened,
requestUnitPrincipalCount: state.requestUnitPrincipalCount,
relayAsiaLoadPrincipalCount: state.relayAsiaLoadPrincipalCount,
@@ -231,6 +233,7 @@ const state = {
regionalFallbacksProved: 0,
oldClientUsFirstProved: 0,
stickyAssignmentProved: 0,
duplicateAssign: null,
requestUnitInvitesOpened: 0,
requestUnitOverflowReason: null,
requestUnitCleanupProved: 0,
@@ -255,6 +258,8 @@ const state = {
}
const peers = new Map()
const reconnectTimers = new Set()
// Duplicate-assign probe hosts: outside `peers`, so they never count toward the run's controls.
const probePeers = []
async function readRuntimeQueuedBytes(origin) {
const response = await fetch(`${origin}/v1/admin/runtime-status`, {
@@ -396,6 +401,47 @@ progressTimer.unref()
await runRelayLoadWithShutdown(async () => {
await Promise.all(initialConnections)
assertRelayLoadRampAccepted(state.rampConnectionFailures, config.maxRampConnectionFailures)
if (config.duplicateAssignProbes > 0) {
const reconnects = { closes: 0, closesByCode: {}, reconnectFailures: 0 }
// A probe host closed by a demotion re-assigns through the director, as a desktop would.
const observeProbe = (peer) => (type, detail) => {
if (type !== 'closed' || detail.stopped) return
reconnects.closes++
const code = String(detail.code)
reconnects.closesByCode[code] = (reconnects.closesByCode[code] ?? 0) + 1
if (state.stopping) return
const timeout = setTimeout(() => {
reconnectTimers.delete(timeout)
void peer.connect().catch(() => {
reconnects.reconnectFailures++
})
}, 1_000 + Math.floor(Math.random() * 5_000))
reconnectTimers.add(timeout)
}
const base = config.controls * config.shardCount + 20_000
const pairs = Array.from({ length: config.duplicateAssignProbes }, (_, probe) => {
const index = base + probe * config.shardCount + config.shardIndex
const primary = new RelayLoadControlPeer(index, peerOptions(index), (type, detail) =>
observeProbe(primary)(type, detail)
)
const twin = new RelayLoadControlPeer(
index,
peerOptions(index, { keys: primary.keys }),
() => undefined
)
probePeers.push(primary, twin)
return { primary, twin }
})
state.duplicateAssign = {
...(await proveRelayLoadDuplicateAssign({
pairs,
holdMs: config.duplicateAssignHoldMs,
delay,
failureReason: relayLoadFailureReason
})),
probeReconnects: reconnects
}
}
if (
config.rebindProbes > 0 || config.placementOverflowProbes > 0 ||
config.regionalFallbackProbes > 0
@@ -527,7 +573,7 @@ await runRelayLoadWithShutdown(async () => {
clearInterval(progressTimer)
for (const timeout of reconnectTimers) clearTimeout(timeout)
reconnectTimers.clear()
await Promise.all([...peers.values()].map((peer) => peer.shutdown()))
await Promise.all([...peers.values(), ...probePeers].map((peer) => peer.shutdown()))
eventLoopDelay.disable()
state.shutdownEvidence = {
peerShutdowns: state.peerShutdowns,
@@ -188,7 +188,8 @@ export class RelayLoadControlPeer {
this.index = index
this.options = options
this.observe = observe
this.keys = nacl.box.keyPair()
// A twin shares a host's keys, so it can seat the same relay host on a second cell.
this.keys = options.keys ?? nacl.box.keyPair()
this.relayHostId = createHash('sha256')
.update(this.keys.publicKey)
.digest('base64url')
@@ -210,7 +211,8 @@ export class RelayLoadControlPeer {
this.spliceSequence = 0
}
connect() {
// `assignment`, when given, is used instead of asking the director.
connect(assignment) {
if (
this.stopped ||
this.connecting ||
@@ -219,7 +221,7 @@ export class RelayLoadControlPeer {
return Promise.resolve()
}
this.connecting = true
const operation = this.connectOnce()
const operation = this.connectOnce(assignment)
this.inFlight.add(operation)
const finish = () => {
this.connecting = false
@@ -233,12 +235,12 @@ export class RelayLoadControlPeer {
return this.lastAssignment?.cellUrl
}
async connectOnce() {
async connectOnce(givenAssignment) {
let socket = null
try {
const relayToken = await this.relayToken()
if (this.stopped) return
const assignment = await this.assignment(relayToken)
const assignment = givenAssignment ?? (await this.assignment(relayToken))
if (this.stopped) return
this.lastAssignment = assignment
socket = this.createSocket(assignment, relayToken)
@@ -1003,3 +1003,34 @@ for (const outcome of ['success', 'failure']) {
})
}
}
test('a twin shares its host keys and connects with a given assignment, not the director', async (context) => {
const originalFetch = global.fetch
context.after(() => {
global.fetch = originalFetch
})
const fetched = []
global.fetch = async (url) => {
fetched.push(String(url))
throw new Error('the director must not be asked')
}
const primary = new RelayLoadControlPeer(0, peerOptions(), () => undefined)
const twin = new RelayLoadControlPeer(0, peerOptions({ keys: primary.keys }), () => undefined)
assert.equal(twin.relayHostId, primary.relayHostId)
const socket = fakeHandshakeSocket()
let socketCell
twin.createSocket = (assignment) => {
socketCell = assignment.cellUrl
openOnNextTurn(socket)
return socket
}
const connecting = twin.connect({ cellUrl: 'https://c3.relay.test', assignmentEpoch: 7 })
await new Promise((resolve) => setTimeout(resolve, 10))
await twin.shutdown()
await connecting.catch(() => undefined)
assert.equal(socketCell, 'https://c3.relay.test')
assert.equal(JSON.parse(socket.sent[0]).assignmentEpoch, 7)
assert.equal(fetched.some((url) => url.endsWith('/v1/assign')), false)
})
@@ -0,0 +1,66 @@
// Step 5's equal-epoch case: two fresh assigns for one host that reach two director instances
// can each book a cell at the same epoch. The host connects with the first answer, and a twin
// with the same keys holds the other seat half-open, as a desktop that joined both would.
export async function proveRelayLoadDuplicateAssign({ pairs, holdMs, delay, failureReason }) {
const result = {
probes: pairs.length,
bothAnswered: 0,
splitCells: 0,
sameEpochSplits: 0,
joinedBoth: 0,
refusalsByReason: {},
connectFailuresByReason: {},
// Not secrets: synthetic load hosts, listed so the ledger rows can be checked afterwards.
splits: []
}
const count = (bucket, error) => {
const reason = failureReason(error)
bucket[reason] = (bucket[reason] ?? 0) + 1
}
const twins = []
await Promise.all(
pairs.map(async ({ primary, twin }) => {
const answers = await Promise.allSettled([
primary.requestAssignment(),
primary.requestAssignment()
])
for (const answer of answers) {
if (answer.status === 'rejected') count(result.refusalsByReason, answer.reason)
}
const [first, second] = answers
.filter((answer) => answer.status === 'fulfilled')
.map((answer) => answer.value)
if (!first) return
if (second) result.bothAnswered++
const split = second !== undefined && second.cellUrl !== first.cellUrl
if (split) {
result.splitCells++
if (second.assignmentEpoch === first.assignmentEpoch) result.sameEpochSplits++
result.splits.push({
relayHostId: primary.relayHostId,
first: { cellUrl: first.cellUrl, assignmentEpoch: first.assignmentEpoch },
second: { cellUrl: second.cellUrl, assignmentEpoch: second.assignmentEpoch }
})
}
try {
await primary.connect(first)
} catch (error) {
count(result.connectFailuresByReason, error)
return
}
if (!split) return
try {
await twin.connect(second)
result.joinedBoth++
twins.push(twin)
} catch (error) {
count(result.connectFailuresByReason, error)
}
})
)
if (twins.length > 0) {
await delay(holdMs)
await Promise.all(twins.map((twin) => twin.shutdown()))
}
return result
}
@@ -0,0 +1,127 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { proveRelayLoadDuplicateAssign } from './relay-load-duplicate-assign.mjs'
const c2 = 'https://c2.relay-staging.onorca.dev'
const c3 = 'https://c3.relay-staging.onorca.dev'
function pair(relayHostId, answers, { twinFails = false } = {}) {
const queue = [...answers]
const log = { primary: [], twin: [], twinShutdowns: 0 }
const primary = {
relayHostId,
requestAssignment: async () => {
const answer = queue.shift()
if (answer instanceof Error) throw answer
return answer
},
connect: async (assignment) => {
log.primary.push(assignment)
}
}
const twin = {
connect: async (assignment) => {
if (twinFails) throw new Error('control closed: 4409 wrong cell')
log.twin.push(assignment)
},
shutdown: async () => {
log.twinShutdowns++
}
}
return { pair: { primary, twin }, log }
}
const failureReason = (error) => (error.message.startsWith('control closed: 4409') ? 'control_close_4409' : 'other')
test('counts same-epoch splits, joins the second cell with a twin, then lets it go', async () => {
const split = pair('aaaaaaaaaaaaaaaa', [
{ cellUrl: c2, assignmentEpoch: 5 },
{ cellUrl: c3, assignmentEpoch: 5 }
])
const same = pair('bbbbbbbbbbbbbbbb', [
{ cellUrl: c2, assignmentEpoch: 3 },
{ cellUrl: c2, assignmentEpoch: 3 }
])
const delays = []
const result = await proveRelayLoadDuplicateAssign({
pairs: [split.pair, same.pair],
holdMs: 20_000,
delay: async (ms) => {
delays.push(ms)
},
failureReason
})
assert.equal(result.probes, 2)
assert.equal(result.bothAnswered, 2)
assert.equal(result.splitCells, 1)
assert.equal(result.sameEpochSplits, 1)
assert.equal(result.joinedBoth, 1)
assert.deepEqual(result.splits, [
{
relayHostId: 'aaaaaaaaaaaaaaaa',
first: { cellUrl: c2, assignmentEpoch: 5 },
second: { cellUrl: c3, assignmentEpoch: 5 }
}
])
assert.deepEqual(split.log.primary, [{ cellUrl: c2, assignmentEpoch: 5 }])
assert.deepEqual(split.log.twin, [{ cellUrl: c3, assignmentEpoch: 5 }])
assert.equal(split.log.twinShutdowns, 1)
assert.deepEqual(same.log.twin, [])
assert.deepEqual(delays, [20_000])
})
test('a split at different epochs is not a same-epoch split', async () => {
const split = pair('cccccccccccccccc', [
{ cellUrl: c2, assignmentEpoch: 5 },
{ cellUrl: c3, assignmentEpoch: 6 }
])
const result = await proveRelayLoadDuplicateAssign({
pairs: [split.pair],
holdMs: 1_000,
delay: async () => undefined,
failureReason
})
assert.equal(result.splitCells, 1)
assert.equal(result.sameEpochSplits, 0)
})
test('a refused second assign still connects the host once and holds nothing', async () => {
const refused = pair('dddddddddddddddd', [
{ cellUrl: c2, assignmentEpoch: 2 },
new Error('relay assignment failed: 503')
])
const delays = []
const result = await proveRelayLoadDuplicateAssign({
pairs: [refused.pair],
holdMs: 20_000,
delay: async (ms) => {
delays.push(ms)
},
failureReason
})
assert.equal(result.bothAnswered, 0)
assert.deepEqual(result.refusalsByReason, { other: 1 })
assert.deepEqual(refused.log.primary, [{ cellUrl: c2, assignmentEpoch: 2 }])
assert.deepEqual(delays, [])
})
test('a twin the second cell refuses is counted, not joined', async () => {
const split = pair(
'eeeeeeeeeeeeeeee',
[
{ cellUrl: c2, assignmentEpoch: 5 },
{ cellUrl: c3, assignmentEpoch: 5 }
],
{ twinFails: true }
)
const result = await proveRelayLoadDuplicateAssign({
pairs: [split.pair],
holdMs: 20_000,
delay: async () => undefined,
failureReason
})
assert.equal(result.joinedBoth, 0)
assert.deepEqual(result.connectFailuresByReason, { control_close_4409: 1 })
assert.equal(split.log.twinShutdowns, 0)
})
+13
View File
@@ -49,6 +49,8 @@ export function parseRelayLoadArguments(argv) {
const placementOverflowProbes = integer('--placement-overflow-probes', 0)
const regionalFallbackProbes = integer('--regional-fallback-probes', 0)
const regionBehaviorProbes = integer('--region-behavior-probes', 0)
const duplicateAssignProbes = integer('--duplicate-assign-probes', 0)
const duplicateAssignHoldMs = integer('--duplicate-assign-hold-ms', 20_000)
const requestUnitInvites = integer('--request-unit-invites', 0)
const requestUnitInvitesPerSecond = integer('--request-unit-invites-per-second', 0)
const requestUnitPrincipalCount = integer('--request-unit-principals', 0)
@@ -218,6 +220,15 @@ export function parseRelayLoadArguments(argv) {
!directorOrigin || !capacityCellOrigin || preferredRegionValue(values) !== 'asia-east2' ||
(shardCount > 1 && shardIndex !== 0)
)) throw new Error('regional behavior proof requires the coordinating Asia shard')
if (duplicateAssignProbes > 1_000) {
throw new Error('--duplicate-assign-probes cannot exceed 1000')
}
if (duplicateAssignProbes > 0 && !directorOrigin) {
throw new Error('duplicate assign probes require --director-origin')
}
if (duplicateAssignHoldMs < 1_000 || duplicateAssignHoldMs > 120_000) {
throw new Error('--duplicate-assign-hold-ms must be between 1000 and 120000')
}
if (phaseBarrierDir && shardCount < 2) {
throw new Error('phase barrier requires multiple shards')
}
@@ -276,6 +287,8 @@ export function parseRelayLoadArguments(argv) {
placementOverflowProbes,
regionalFallbackProbes,
regionBehaviorProbes,
duplicateAssignProbes,
duplicateAssignHoldMs,
requestUnitInvites,
requestUnitInvitesPerSecond,
requestUnitPrincipalCount,
@@ -393,3 +393,20 @@ test('requires queue, memory, and expected close evidence', () => {
/did not close with 4429/
)
})
test('bounds duplicate-assign probes to a director run', () => {
const config = parseRelayLoadArguments([...required, '--duplicate-assign-probes', '100'])
assert.equal(config.duplicateAssignProbes, 100)
assert.equal(config.duplicateAssignHoldMs, 20_000)
assert.equal(parseRelayLoadArguments(required).duplicateAssignProbes, 0)
assert.throws(() => parseRelayLoadArguments([
'--auth-origin', 'https://auth.test', '--target-origin', 'https://cell.test',
'--duplicate-assign-probes', '1'
]), /require --director-origin/)
assert.throws(() => parseRelayLoadArguments([
...required, '--duplicate-assign-probes', '1001'
]), /cannot exceed 1000/)
assert.throws(() => parseRelayLoadArguments([
...required, '--duplicate-assign-probes', '1', '--duplicate-assign-hold-ms', '500'
]), /between 1000 and 120000/)
})
+1 -1
View File
@@ -22,7 +22,7 @@
"load:relay:recovery-gate": "node dev/scripts/run-relay-recovery-wave-gate.mjs",
"ops:relay": "pnpm --filter @orca-cloud/relay-ops dev",
"pretest": "node --test dev/scripts/capture-terraform-plan-baseline.test.mjs dev/scripts/operate-relay-asia-admission.test.mjs dev/scripts/prepare-relay-asia-director-cells.test.mjs dev/scripts/prepare-relay-asia-topology-input.test.mjs dev/scripts/production-cloud-sql-rollout-lock.test.mjs dev/scripts/read-relay-serving-regional-placement-version.test.mjs dev/scripts/relay-asia-rollout-evidence.test.mjs dev/scripts/relay-asia-topology-workflow.test.mjs dev/scripts/relay-cloud-sql-connection-budget.test.mjs dev/scripts/relay-load-reader-evidence.test.mjs dev/scripts/relay-control-flag-alerts.test.mjs dev/scripts/relay-db-fault-alerts.test.mjs dev/scripts/relay-lock-contention-alerts.test.mjs dev/scripts/relay-reconnect-herd-alerts.test.mjs dev/scripts/relay-reserve-dead-man-alert.test.mjs dev/scripts/relay-region-hint-metrics.test.mjs dev/scripts/relay-staging-deploy-identity.test.mjs dev/scripts/sanitize-relay-asia-admission-result.test.mjs dev/scripts/terraform-root-partition.test.mjs dev/scripts/validate-relay-asia-topology-plan.test.mjs ../.github/actions/cloud-sql-rollout-lease/action-contract.test.mjs ../.github/actions/cloud-sql-rollout-lease/storage-lease.test.mjs",
"test": "pnpm -r test && node --test dev/scripts/check-relay-same-cap-headroom.test.mjs dev/scripts/classify-relay-production-capacity-director.test.mjs dev/scripts/classify-relay-staging-bootstrap.test.mjs dev/scripts/deploy-relay-blue-green.test.mjs dev/scripts/deploy-relay-gce-candidate.test.mjs dev/scripts/deploy-relay-gce-multi-target.test.mjs dev/scripts/drive-relay-director-deploy.test.mjs dev/scripts/github-smoke-token.test.mjs dev/scripts/infra.test.mjs dev/scripts/measure-relay-drain-disconnect-gap.test.mjs dev/scripts/operate-relay-cell-flags.test.mjs dev/scripts/operate-relay-regional-rehome.test.mjs dev/scripts/power-staging-relay.test.mjs dev/scripts/prepare-relay-capacity-canary.test.mjs dev/scripts/prepare-relay-production-capacity-canary.test.mjs dev/scripts/probe-relay-legacy-admission.test.mjs dev/scripts/probe-relay-rehome-trust.test.mjs dev/scripts/production-cell-image-digest-consistency.test.mjs dev/scripts/push-gateway-workflow.test.mjs dev/scripts/push-gateway-recovery.test.mjs dev/scripts/read-relay-production-capacity-identity.test.mjs dev/scripts/relay-admin-endpoint-retry-workflow.test.mjs dev/scripts/relay-admin-transient-retry.test.mjs dev/scripts/relay-admission-selector.test.mjs dev/scripts/relay-gce-terraform-fence.test.mjs dev/scripts/relay-load-connection-failure.test.mjs dev/scripts/relay-load-control-peer.test.mjs dev/scripts/relay-load-director-capacity-gate.test.mjs dev/scripts/relay-load-model.test.mjs dev/scripts/relay-load-phase-barrier.test.mjs dev/scripts/relay-load-placement-boundary.test.mjs dev/scripts/relay-load-profile.test.mjs dev/scripts/relay-load-rebind-boundary.test.mjs dev/scripts/relay-load-region-behavior.test.mjs dev/scripts/relay-load-request-unit-boundary.test.mjs dev/scripts/relay-load-run-lifecycle.test.mjs dev/scripts/relay-monitor-evidence.test.mjs dev/scripts/relay-production-capacity-wave.test.mjs dev/scripts/relay-production-capacity-workflow.test.mjs dev/scripts/relay-production-identity-boundaries.test.mjs dev/scripts/relay-production-same-cap-wave.test.mjs dev/scripts/relay-public-workflow-contract.test.mjs dev/scripts/relay-recovery-wave-gate.test.mjs dev/scripts/relay-region-observation-evidence.test.mjs dev/scripts/relay-rehome-aggregate-evidence.test.mjs dev/scripts/relay-repository.test.mjs dev/scripts/relay-same-cap-job-mode-conditions.test.mjs dev/scripts/relay-same-cap-script-census.test.mjs dev/scripts/relay-same-cap-shadow-gate.test.mjs dev/scripts/relay-staging-c4-refresh-workflow.test.mjs dev/scripts/relay-staging-capacity-identity.test.mjs dev/scripts/staging-relay-apply-guard.test.mjs dev/scripts/validate-relay-capacity-plan.test.mjs dev/scripts/verify-relay-capacity-transition.test.mjs dev/scripts/verify-relay-legacy-bootstrap.test.mjs dev/scripts/workload-identity-attribute-conditions.test.mjs",
"test": "pnpm -r test && node --test dev/scripts/check-relay-same-cap-headroom.test.mjs dev/scripts/classify-relay-production-capacity-director.test.mjs dev/scripts/classify-relay-staging-bootstrap.test.mjs dev/scripts/deploy-relay-blue-green.test.mjs dev/scripts/deploy-relay-gce-candidate.test.mjs dev/scripts/deploy-relay-gce-multi-target.test.mjs dev/scripts/drive-relay-director-deploy.test.mjs dev/scripts/github-smoke-token.test.mjs dev/scripts/infra.test.mjs dev/scripts/measure-relay-drain-disconnect-gap.test.mjs dev/scripts/operate-relay-cell-flags.test.mjs dev/scripts/operate-relay-regional-rehome.test.mjs dev/scripts/power-staging-relay.test.mjs dev/scripts/prepare-relay-capacity-canary.test.mjs dev/scripts/prepare-relay-production-capacity-canary.test.mjs dev/scripts/probe-relay-legacy-admission.test.mjs dev/scripts/probe-relay-rehome-trust.test.mjs dev/scripts/production-cell-image-digest-consistency.test.mjs dev/scripts/push-gateway-workflow.test.mjs dev/scripts/push-gateway-recovery.test.mjs dev/scripts/read-relay-production-capacity-identity.test.mjs dev/scripts/relay-admin-endpoint-retry-workflow.test.mjs dev/scripts/relay-admin-transient-retry.test.mjs dev/scripts/relay-admission-selector.test.mjs dev/scripts/relay-gce-terraform-fence.test.mjs dev/scripts/relay-load-connection-failure.test.mjs dev/scripts/relay-load-control-peer.test.mjs dev/scripts/relay-load-director-capacity-gate.test.mjs dev/scripts/relay-load-duplicate-assign.test.mjs dev/scripts/relay-load-model.test.mjs dev/scripts/relay-load-phase-barrier.test.mjs dev/scripts/relay-load-placement-boundary.test.mjs dev/scripts/relay-load-profile.test.mjs dev/scripts/relay-load-rebind-boundary.test.mjs dev/scripts/relay-load-region-behavior.test.mjs dev/scripts/relay-load-request-unit-boundary.test.mjs dev/scripts/relay-load-run-lifecycle.test.mjs dev/scripts/relay-monitor-evidence.test.mjs dev/scripts/relay-production-capacity-wave.test.mjs dev/scripts/relay-production-capacity-workflow.test.mjs dev/scripts/relay-production-identity-boundaries.test.mjs dev/scripts/relay-production-same-cap-wave.test.mjs dev/scripts/relay-public-workflow-contract.test.mjs dev/scripts/relay-recovery-wave-gate.test.mjs dev/scripts/relay-region-observation-evidence.test.mjs dev/scripts/relay-rehome-aggregate-evidence.test.mjs dev/scripts/relay-repository.test.mjs dev/scripts/relay-same-cap-job-mode-conditions.test.mjs dev/scripts/relay-same-cap-script-census.test.mjs dev/scripts/relay-same-cap-shadow-gate.test.mjs dev/scripts/relay-staging-c4-refresh-workflow.test.mjs dev/scripts/relay-staging-capacity-identity.test.mjs dev/scripts/staging-relay-apply-guard.test.mjs dev/scripts/validate-relay-capacity-plan.test.mjs dev/scripts/verify-relay-capacity-transition.test.mjs dev/scripts/verify-relay-legacy-bootstrap.test.mjs dev/scripts/workload-identity-attribute-conditions.test.mjs",
"typecheck": "pnpm -r typecheck"
},
"devDependencies": {