mirror of
https://github.com/stablyai/orca.git
synced 2026-09-29 08:03:20 +00:00
fix(relay): one hold per proven hint, log it as held-previous, and reread before a self-heal delete
- a held incumbent is written without the fullCatalog marker, so the next expiry must re-earn it against a full catalog; holds cannot chain on partial evidence - the hold is logged as held-previous, not measured - self-heal rereads the cache before rmSync so a refresh that landed during its probes is not deleted on a stale verdict (new reason superseded-by-refresh)
This commit is contained in:
@@ -409,11 +409,29 @@ describe('Relay region preference', () => {
|
||||
logEvent: (event) => events.push(event)
|
||||
}).resolve()
|
||||
).resolves.toBe('asia-east2')
|
||||
expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({
|
||||
region: 'asia-east2',
|
||||
expiresAt: 1_000 + 24 * 60 * 60_000
|
||||
})
|
||||
expect(events).toEqual([expect.objectContaining({ chosenRegion: 'asia-east2' })])
|
||||
const held = JSON.parse(readFileSync(cachePath(path), 'utf8'))
|
||||
expect(held).toMatchObject({ region: 'asia-east2', expiresAt: 1_000 + 24 * 60 * 60_000 })
|
||||
// One hold per proven hint: the held entry carries no marker, so the next expiry
|
||||
// must re-earn the region against a full catalog rather than chain holds forever.
|
||||
expect(held.fullCatalog).toBeUndefined()
|
||||
expect(events).toEqual([
|
||||
expect.objectContaining({ chosenRegion: 'asia-east2', reason: 'held-previous' })
|
||||
])
|
||||
})
|
||||
|
||||
it('cannot chain holds: a held hint that expires during a second roll wave is not held again', async () => {
|
||||
const path = userDataPath()
|
||||
writeCache(path, 'asia-east2', 999, false)
|
||||
const healthy = sampledProbe({ [ASIA]: [90, 30, 32, 34] })
|
||||
await expect(
|
||||
new RelayRegionPreferenceResolver({
|
||||
directorUrl: DIRECTOR,
|
||||
userDataPath: path,
|
||||
fetch: catalogFetch([{ region: 'asia-east2', probeOrigins: [ASIA] }]),
|
||||
probe: healthy.probe,
|
||||
now: () => 1_000
|
||||
}).resolve()
|
||||
).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('does not hold an incumbent an older build cached without the full-catalog marker', async () => {
|
||||
@@ -712,6 +730,45 @@ describe('Relay region cache self-heal', () => {
|
||||
expect(existsSync(cachePath(path))).toBe(true)
|
||||
})
|
||||
|
||||
it('does not delete a cache a concurrent refresh rewrote while the self-heal probed', async () => {
|
||||
// Why: self-heal snapshots the cache before its probes. A refresh that finishes in
|
||||
// between may have written a different, correct region; the stale verdict must not
|
||||
// delete it and force yet another probe.
|
||||
const path = userDataPath()
|
||||
writeCache(path, 'asia-east2', 999)
|
||||
const events: unknown[] = []
|
||||
let release: () => void = () => {}
|
||||
const gate = new Promise<void>((resolve) => (release = resolve))
|
||||
const samples: Record<string, number[]> = {
|
||||
[US]: [300, 30, 32, 34],
|
||||
[ASIA]: [300, 200, 205, 210],
|
||||
[CELL]: [300, 200, 205, 210]
|
||||
}
|
||||
const probe = async (origin: string): Promise<number | null> => {
|
||||
await gate
|
||||
return samples[origin]?.shift() ?? null
|
||||
}
|
||||
const resolver = new RelayRegionPreferenceResolver({
|
||||
directorUrl: DIRECTOR,
|
||||
userDataPath: path,
|
||||
fetch: catalogFetch(BOTH_REGIONS),
|
||||
probe,
|
||||
now: () => 500,
|
||||
logEvent: (event) => events.push(event)
|
||||
})
|
||||
const heal = resolver.invalidateIfAssignedCellIsFar(CELL)
|
||||
// The refresh lands first and writes us-central1 against a full catalog.
|
||||
writeCache(path, 'us-central1', 999)
|
||||
release()
|
||||
await heal
|
||||
expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({
|
||||
region: 'us-central1'
|
||||
})
|
||||
expect(events).toContainEqual(
|
||||
expect.objectContaining({ decision: 'kept', reason: 'superseded-by-refresh' })
|
||||
)
|
||||
})
|
||||
|
||||
it('probes a given cell only once per process', async () => {
|
||||
const path = userDataPath()
|
||||
writeCache(path, 'asia-east2', LIVE_EXPIRY)
|
||||
|
||||
@@ -163,7 +163,15 @@ export class RelayRegionPreferenceResolver {
|
||||
outcome.decision = far ? 'deleted' : 'kept'
|
||||
outcome.reason = far ? 'assigned-cell-far' : 'assigned-cell-near'
|
||||
if (far) {
|
||||
rmSync(this.cachePath(), { force: true })
|
||||
// Reread first: a refresh that finished during the probes may have
|
||||
// written a different, correct region that this stale verdict must not delete.
|
||||
const latest = readRelayRegionCache(this.cachePath(), this.options.directorUrl, now)
|
||||
if (latest?.region === cache.region) {
|
||||
rmSync(this.cachePath(), { force: true })
|
||||
} else {
|
||||
outcome.decision = 'kept'
|
||||
outcome.reason = 'superseded-by-refresh'
|
||||
}
|
||||
}
|
||||
this.logSelfHeal(outcome)
|
||||
} catch {
|
||||
@@ -210,12 +218,21 @@ export class RelayRegionPreferenceResolver {
|
||||
reports,
|
||||
best: bestMeasurement(measurements),
|
||||
selected,
|
||||
held: !complete && selected !== null,
|
||||
ttlMs
|
||||
})
|
||||
)
|
||||
// A held incumbent is written without the marker: one roll wave may extend
|
||||
// a proven hint by a day, but the next expiry must re-earn it against a full
|
||||
// catalog, so a hint can never be renewed indefinitely on partial evidence.
|
||||
this.writeCache(
|
||||
selected
|
||||
? { region: selected.region, latencyMs: selected.latencyMs, ttlMs, fullCatalog: true }
|
||||
? {
|
||||
region: selected.region,
|
||||
latencyMs: selected.latencyMs,
|
||||
ttlMs,
|
||||
...(complete ? { fullCatalog: true as const } : {})
|
||||
}
|
||||
: { region: null, ttlMs },
|
||||
now
|
||||
)
|
||||
|
||||
@@ -48,6 +48,7 @@ export type RelayRegionSelfHealLogEvent = {
|
||||
| 'catalog-unavailable'
|
||||
| 'assigned-cell-near'
|
||||
| 'assigned-cell-far'
|
||||
| 'superseded-by-refresh'
|
||||
}
|
||||
|
||||
export type RelayRegionLogEvent = RelayRegionProbeLogEvent | RelayRegionSelfHealLogEvent
|
||||
@@ -107,6 +108,8 @@ export function relayRegionRefreshEvent(input: {
|
||||
reports: RelayRegionProbeReport[]
|
||||
best: RegionMeasurement | null
|
||||
selected: RegionMeasurement | null
|
||||
/** The incumbent was kept through an incomplete catalog rather than measured against a rival. */
|
||||
held?: boolean
|
||||
ttlMs: number
|
||||
}): RelayRegionProbeLogEvent {
|
||||
return {
|
||||
@@ -114,7 +117,7 @@ export function relayRegionRefreshEvent(input: {
|
||||
directorHost: relayDirectorHost(input.directorUrl),
|
||||
regions: input.reports,
|
||||
chosenRegion: input.selected?.region ?? 'no-hint',
|
||||
reason: refreshReason(input.reports, input.best, input.selected),
|
||||
reason: input.held ? 'held-previous' : refreshReason(input.reports, input.best, input.selected),
|
||||
cached: false,
|
||||
ttlMs: input.ttlMs
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user