From a6efa386f6d7ca052f87108bbb6f90b8aae62e42 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 21 Oct 2026 08:53:18 +0200 Subject: [PATCH] mason: preserve fallback admission during first identity publication A profile-triggered recovery can publish a fallback's first UUID between the host snapshot and the coordinator's pre-join read. Exact binding admission then dropped the fallback as row-ineligible, returning the main provider 400. A second window exposes a torn state/config row at that same read. Join recoverable same-epoch pool writes with the existing bounded budget, then adopt only a first identity whose store, row and epoch still match. Keep all admission exclusions and exact-version proof checks, and preserve the branded transient error when the writer cannot finish. Both race windows have red-first Core regressions and CAUGHT named mutation controls with Core broad settings. Existing history tests are unchanged. Pinned Bun 1.3.14 Linux loops: main originally had 2/40 single and 3/40 block status-400 misses; v2.1.2 had none; fixed loops have none in 40+40 processes, with one unrelated default-deadline timeout in each. Core owning file: 122 pass. Final OpenCode owning file: 304 pass, two claude-prime timeouts on the shared runner. Core/OpenCode typechecks, scripts typecheck, lint and formatting pass. --- mutations.toml | 32 +++ .../core/src/native-refresh-coordinator.ts | 47 +++- .../tests/native-refresh-coordinator.test.ts | 226 ++++++++++++++++++ 3 files changed, 301 insertions(+), 4 deletions(-) diff --git a/mutations.toml b/mutations.toml index ffe06260..18916c36 100644 --- a/mutations.toml +++ b/mutations.toml @@ -372,6 +372,38 @@ broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core broad_report = "tmp/mutations/core.xml" broad_id = "{name}" +[[control]] +id = "serve-adopts-concurrent-first-identity" +guards = "A serving caller captured before first identity publication must use the same row's validated credential without bypassing admission." +file = "packages/core/src/native-refresh-coordinator.ts" +old = "const learned = { ...request.binding, identity: row.identity }" +new = "const learned = request.binding // NON-VACUITY BREAK: retain the pre-publication identity-absent binding" +test_file = "packages/core/src/tests/native-refresh-coordinator.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-refresh-coordinator.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["serve follows first identity published after binding capture without a second exchange"] +expect_message = "error: expect(received).toMatchObject(expected)" +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] +broad_report = "tmp/mutations/core.xml" +broad_id = "{name}" + +[[control]] +id = "serve-joins-concurrent-first-identity-write" +guards = "A partial first-identity pool write seen after the host's snapshot must be joined rather than silently remove a fallback." +file = "packages/core/src/native-refresh-coordinator.ts" +old = "row.credentialEpoch === request.binding.credentialEpoch &&" +new = "row.credentialEpoch === -1 && // NON-VACUITY BREAK: refuse a partially saved first identity" +test_file = "packages/core/src/tests/native-refresh-coordinator.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-refresh-coordinator.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["serve joins a first identity write encountered after binding capture"] +expect_message = "error: expect(received).toMatchObject(expected)" +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] +broad_report = "tmp/mutations/core.xml" +broad_id = "{name}" + [[control]] id = "refresh-join-checks-cause" guards = "A proactive refresh must not delay a proven unexpired access token." diff --git a/packages/core/src/native-refresh-coordinator.ts b/packages/core/src/native-refresh-coordinator.ts index fbb0adbe..2dbeb3ed 100644 --- a/packages/core/src/native-refresh-coordinator.ts +++ b/packages/core/src/native-refresh-coordinator.ts @@ -26,14 +26,21 @@ import { CustodyTombstoneRefreshError, isCustodyTombstoneValue, } from './claustrum.ts' -import { nativeAccountCanServe } from './native-account-view.ts' +import { + nativeAccountCanServe, + nativeLocalRowAwaitsTornWrite, +} from './native-account-view.ts' import { type NativeLocalCredentialValidation, type NativeLocalCredentialVersion, nativeLocalCredentialValidationMatches, } from './native-credential-validation.ts' import { nativeLockObserver } from './native-lock-observer.ts' -import { NativeTransientAccountError } from './native-transient-account.ts' +import { + isNativeTransientAccountError, + joinNativePoolWrite, + NativeTransientAccountError, +} from './native-transient-account.ts' import { isTransientNetworkError } from './network-errors.ts' import { isNativeLocalPoolBinding, @@ -1325,7 +1332,7 @@ export function createNativeRefreshCoordinator( request = { ...request, binding: Object.freeze({ ...request.binding }) } if (request.signal?.aborted) return { status: 'refused', reason: 'cancelled', persisted: false } - const key = nativeLocalRefreshJobKey(request.binding) + let key = nativeLocalRefreshJobKey(request.binding) let job = jobs.get(key) const recordRejection = () => { // A 401 can arrive after proactive work started. Record its rejected token @@ -1338,7 +1345,38 @@ export function createNativeRefreshCoordinator( recordRejection() if (request.intent === 'serve') { try { - const row = await rowFor(request.binding) + let row = await rowFor(request.binding) + if ( + row && + row.credentialEpoch === request.binding.credentialEpoch && + nativeLocalRowAwaitsTornWrite(options.paths, row) + ) { + // The host's earlier read can precede a partial state/config write. + // Wait for that writer once before deciding this account is ineligible. + const joined = await joinNativePoolWrite( + options.paths, + request.signal ? [request.signal] : [], + () => request.signal?.throwIfAborted(), + onLockEvent, + ) + if (joined === 'deadline') throw new NativeTransientAccountError() + row = await rowFor(request.binding) + if (row && nativeLocalRowAwaitsTornWrite(options.paths, row)) + throw new NativeTransientAccountError() + } + if ( + request.binding.identity === undefined && + row?.identity !== undefined + ) { + const learned = { ...request.binding, identity: row.identity } + // First-identity publication can finish after the caller captures its + // binding. Use the saved identity only if the row, store and credential + // epoch still match; admission still requires permission and exact proof. + if (nativeLocalPoolBindingMatches(learned, options.paths, row)) { + request = { ...request, binding: Object.freeze(learned) } + key = nativeLocalRefreshJobKey(request.binding) + } + } job = jobs.get(key) ?? job recordRejection() // A pending recovery cannot override custody, quota or refresh-error @@ -1367,6 +1405,7 @@ export function createNativeRefreshCoordinator( return current } } catch (caught) { + if (isNativeTransientAccountError(caught)) throw caught return { status: 'failed', failure: failureOf(asError(caught)), diff --git a/packages/core/src/tests/native-refresh-coordinator.test.ts b/packages/core/src/tests/native-refresh-coordinator.test.ts index 8277a1ee..e72b07a5 100644 --- a/packages/core/src/tests/native-refresh-coordinator.test.ts +++ b/packages/core/src/tests/native-refresh-coordinator.test.ts @@ -61,6 +61,10 @@ import { type NativeRefreshSubject, nativeAccountProviderLock, } from '../native-refresh-coordinator.ts' +import { + isNativeTransientAccountError, + NativeTransientAccountError, +} from '../native-transient-account.ts' import { captureNativeLocalPoolBinding, nativeLocalPoolBindingMatches, @@ -1233,6 +1237,228 @@ test('callback-end identity alias joins during reconciliation and handoff enclos expect(active.size).toBe(0) }) +test('serve follows first identity published after binding capture without a second exchange', async () => { + const f = await fixture() + await f.add('a') + const before = await f.binding('a') + const publishing = barrier() + const finish = barrier() + let calls = 0 + const c = f.coordinator({ + refreshToken: async () => { + calls++ + return { ...credential('new'), expiresIn: 3600 } + }, + reconcile: async (event) => { + if (event.status === 'persisted') { + publishing.release() + await finish.promise + } + }, + }) + const first = c.authorize({ intent: 'serve', mode: 'local', binding: before }) + await publishing.promise + const learned = await f.binding('a') + expect(learned).toEqual({ ...before, identity: 'A' }) + // This binding was captured before the row had an identity. Saving its first + // identity must not make a serving caller treat the account as replaced. + const joined = c.authorize({ + intent: 'serve', + mode: 'local', + binding: before, + }) + finish.release() + expect(await joined).toMatchObject({ + status: 'usable', + binding: learned, + access: 'access-new', + }) + expect(await first).toMatchObject({ status: 'usable', binding: learned }) + expect( + await c.authorize({ intent: 'serve', mode: 'local', binding: before }), + ).toMatchObject({ status: 'usable', binding: learned, access: 'access-new' }) + expect(calls).toBe(1) +}) + +test('serve joins a first identity write encountered after binding capture', async () => { + const stateWritten = barrier() + const finishWrite = barrier() + const joining = barrier() + let held = false + let waitForJoin = false + const f = await fixture({ + onStep: async (step, info) => { + if ( + held || + step !== 'after-state-write' || + info.operation !== 'refresh' || + info.rowId !== 'a' + ) + return + held = true + stateWritten.release() + await finishWrite.promise + }, + }) + await f.add('a') + const before = await f.binding('a') + let calls = 0 + const c = f.coordinator({ + refreshToken: async () => { + calls++ + return { ...credential('new'), expiresIn: 3600 } + }, + onLockEvent: (event) => { + if ( + waitForJoin && + event.type === 'acquiring' && + event.name === 'pool-config' + ) + joining.release() + }, + }) + const first = c.authorize({ intent: 'serve', mode: 'local', binding: before }) + await stateWritten.promise + expect(await f.row('a')).toMatchObject({ + torn: true, + candidate: false, + identity: 'A', + }) + waitForJoin = true + const joined = c.authorize({ + intent: 'serve', + mode: 'local', + binding: before, + }) + // Release when the reader waits on the lock or returns early; never leave the + // writer held after a refusal. + await Promise.race([joining.promise, joined.then(() => {})]) + finishWrite.release() + expect(await joined).toMatchObject({ status: 'usable', access: 'access-new' }) + expect(await first).toMatchObject({ status: 'usable', access: 'access-new' }) + expect(calls).toBe(1) +}) + +test('a first identity write deadline keeps the transient account error brand', async () => { + const stateWritten = barrier() + const finishWrite = barrier() + let held = false + const f = await fixture({ + onStep: async (step, info) => { + if ( + held || + step !== 'after-state-write' || + info.operation !== 'refresh' || + info.rowId !== 'a' + ) + return + held = true + stateWritten.release() + await finishWrite.promise + }, + }) + await f.add('a') + const before = await f.binding('a') + const c = f.coordinator() + const first = c.authorize({ intent: 'serve', mode: 'local', binding: before }) + await stateWritten.promise + const refused = await c + .authorize({ intent: 'serve', mode: 'local', binding: before }) + .then( + (value) => ({ value }), + (error: unknown) => ({ error }), + ) + finishWrite.release() + expect(refused).toHaveProperty('error') + if (!('error' in refused)) + throw new Error('Expected a transient account error') + expect(isNativeTransientAccountError(refused.error)).toBe(true) + expect(refused.error).toBeInstanceOf(NativeTransientAccountError) + expect(await first).toMatchObject({ status: 'usable', access: 'access-new' }) +}) + +for (const reason of [ + 'quota-ineligible', + 'account-disabled', + 'local-mode-unavailable', + 'refresh-backoff', + 'invalid-grant', + 'validation-backoff', +] as const) { + test(`first identity publication preserves the ${reason} admission exclusion`, async () => { + const f = await fixture() + await f.add('a') + const before = await f.binding('a') + const publishing = barrier() + const finish = barrier() + let calls = 0 + const c = f.coordinator({ + refreshToken: async () => { + calls++ + return { ...credential('new'), expiresIn: 3600 } + }, + readAdmission: async () => ({ status: 'blocked', reason }), + reconcile: async (event) => { + if (event.status === 'persisted') { + publishing.release() + await finish.promise + } + }, + }) + const first = c.authorize({ + intent: 'serve', + mode: 'local', + binding: before, + }) + await publishing.promise + const refused = await c.authorize({ + intent: 'serve', + mode: 'local', + binding: before, + }) + finish.release() + expect(refused).toMatchObject({ status: 'refused', reason }) + expect(await first).toMatchObject({ status: 'refused', reason }) + expect(calls).toBe(1) + }) +} + +test('first identity adoption never follows a replaced epoch or a known different identity', async () => { + const f = await fixture() + await f.add('a') + const before = await f.binding('a') + let calls = 0 + const c = f.coordinator({ + refreshToken: async () => { + calls++ + return { ...credential('new'), expiresIn: 3600 } + }, + }) + expect( + await c.authorize({ intent: 'serve', mode: 'local', binding: before }), + ).toMatchObject({ status: 'usable', access: 'access-new' }) + expect( + await c.authorize({ + intent: 'serve', + mode: 'local', + binding: { ...before, identity: 'B' }, + }), + ).toMatchObject({ status: 'refused', reason: 'row-ineligible' }) + await f.store.replace('a', credential('replacement'), { identity: 'A' }) + await f.seedProof('a') + expect( + await c.authorize({ intent: 'serve', mode: 'local', binding: before }), + ).toMatchObject({ status: 'refused', reason: 'row-ineligible' }) + expect( + await c.authorize({ + intent: 'serve', + mode: 'local', + binding: await f.binding('a'), + }), + ).toMatchObject({ status: 'usable', access: 'access-replacement' }) + expect(calls).toBe(1) +}) + test('existing known job retains its alias while unknown handoff waits and both successors persist', async () => { const knownReconciling = barrier() const unknownBootstrapped = barrier()