diff --git a/docs/TEST_INTEGRITY.md b/docs/TEST_INTEGRITY.md index 9bb24fd..c17dc32 100644 --- a/docs/TEST_INTEGRITY.md +++ b/docs/TEST_INTEGRITY.md @@ -3176,8 +3176,12 @@ test_integrity_note: then: "The transition refuses with its structured boundary code and no integration queue record" - id: SCN-integration-adoption-request-replay given: "A finalized adoption that has created its one ordinary private-ref integration request" - when: "A concurrent exact-row insert or later request replay is attempted" - then: "Real SQLite rejects the duplicate exact source row, the journal replay refuses, and exactly one queue row remains" + when: "Two independently opened authenticated API callers request it concurrently" + then: "The existing transactional adoption journal serializes them, one caller refuses after the completed transition, and exactly one queue row remains" + - id: SCN-integration-adoption-legacy-retries + given: "An existing project database containing two historical ordinary-request retry rows with the same source branch and commit" + when: "IntegrationService starts" + then: "Both records remain readable; startup adds no global uniqueness constraint and does not reject historical evidence" observable_outcomes: - "Git ref values, structured refusal codes, durable journal state, and native integration-request fields" determinism_controls: @@ -3189,7 +3193,8 @@ test_integrity_note: - "Claiming crash recovery without preserving actual ref effects" - "Treating a private candidate as an ordinary specialist branch" - "Using an unrelated, source-only, stale, or mismatched approval as adoption authority" - - "Duplicating a private-ref queue record through a concurrent row or replay" + - "Duplicating a private-ref queue record through concurrent authenticated API requests or replay" + - "Rejecting historical ordinary-request retry evidence with an unqualified schema constraint" relaxation: did_relax_any_assertion: false if_true_explain_spec_basis: "" diff --git a/src/integration/adoption.mjs b/src/integration/adoption.mjs index 3a6c263..9d5eabd 100644 --- a/src/integration/adoption.mjs +++ b/src/integration/adoption.mjs @@ -236,23 +236,26 @@ export class AdoptionService { } #append(adoptionId, expectedRevision, record, operation) { - const now = this.clock().toISOString(); this.controlPlane.database.exec('BEGIN IMMEDIATE'); try { - const current = this.#find(adoptionId); - if (expectedRevision === null ? current : !current || current.revision !== expectedRevision) { - throw new TorchError('Adoption journal changed before this transition could be recorded', { - code: 'ADOPTION_RECORD_REVISION_STALE', details: { adoptionId, expectedRevision, actual: current?.revision ?? null }, - }); - } - this.controlPlane.database.prepare(`INSERT INTO audit_events - (id, project_id, actor_id, operation, entity_type, entity_id, details, created_at) - VALUES (?, ?, 'owner', ?, 'integration-adoption', ?, ?, ?)`) - .run(this.idFactory(), this.controlPlane.projectId, operation, adoptionId, JSON.stringify(record), now); + this.#appendInTransaction(adoptionId, expectedRevision, record, operation); this.controlPlane.database.exec('COMMIT'); } catch (error) { this.controlPlane.database.exec('ROLLBACK'); throw error; } } + #appendInTransaction(adoptionId, expectedRevision, record, operation) { + const current = this.#find(adoptionId); + if (expectedRevision === null ? current : !current || current.revision !== expectedRevision) { + throw new TorchError('Adoption journal changed before this transition could be recorded', { + code: 'ADOPTION_RECORD_REVISION_STALE', details: { adoptionId, expectedRevision, actual: current?.revision ?? null }, + }); + } + this.controlPlane.database.prepare(`INSERT INTO audit_events + (id, project_id, actor_id, operation, entity_type, entity_id, details, created_at) + VALUES (?, ?, 'owner', ?, 'integration-adoption', ?, ?, ?)`) + .run(this.idFactory(), this.controlPlane.projectId, operation, adoptionId, JSON.stringify(record), this.clock().toISOString()); + } + #fromRecord(record, input, action) { exactInput(record, input); return this.plan({ action, actorId: input.actorId, approvalId: input.approvalId, approvalRevision: input.approvalRevision, @@ -326,20 +329,37 @@ export class AdoptionService { } requestAdoptedCandidate(input = {}, { requestIntegration, findIntegrationRequest } = {}) { - const record = this.getPrepared(text(input.adoptionId, 'adoptionId')); - if (!['finalized', 'requesting'].includes(record.state)) throw new TorchError('Only a finalized adoption can request native qualification', { code: 'ADOPTION_REQUEST_STATE_INVALID', details: { state: record.state } }); - const plan = this.#fromRecord(record, input, 'request'); - if (!plan.canProceed) throw new TorchError('Adopted candidate request is blocked', { code: 'ADOPTION_REQUEST_BLOCKED', details: plan.blockers }); if (typeof requestIntegration !== 'function') throw new TorchError('Native integration request adapter is unavailable', { code: 'ADOPTION_NATIVE_REQUEST_UNAVAILABLE' }); - const requesting = record.state === 'requesting' ? record : { ...record, revision: record.revision + 1, state: 'requesting', - requestApprovalId: text(input.approvalId, 'approvalId'), requestApprovalRevision: input.approvalRevision, requestStartedAt: this.clock().toISOString() }; - if (record.state !== 'requesting') this.#append(record.id, record.revision, requesting, 'integration.adoption.request-started'); - const request = findIntegrationRequest?.({ candidateRef: requesting.candidateRef, commit: requesting.candidateSha }) - ?? requestIntegration({ adoptionId: requesting.id, areaId: requesting.contributorArea, + const adoptionId = text(input.adoptionId, 'adoptionId'); + this.controlPlane.database.exec('BEGIN IMMEDIATE'); + try { + const record = this.getPrepared(adoptionId); + if (!['finalized', 'requesting'].includes(record.state)) { + throw new TorchError('Only a finalized adoption can request native qualification', { + code: 'ADOPTION_REQUEST_STATE_INVALID', details: { state: record.state }, + }); + } + const plan = this.#fromRecord(record, input, 'request'); + if (!plan.canProceed) throw new TorchError('Adopted candidate request is blocked', { code: 'ADOPTION_REQUEST_BLOCKED', details: plan.blockers }); + const requesting = record.state === 'requesting' ? record : { ...record, revision: record.revision + 1, state: 'requesting', + requestApprovalId: text(input.approvalId, 'approvalId'), requestApprovalRevision: input.approvalRevision, requestStartedAt: this.clock().toISOString() }; + if (record.state !== 'requesting') this.#appendInTransaction(record.id, record.revision, requesting, 'integration.adoption.request-started'); + const existing = findIntegrationRequest?.({ candidateRef: requesting.candidateRef, commit: requesting.candidateSha }); + if (existing && (existing.sourceArea !== requesting.contributorArea || existing.targetBranch !== requesting.targetBranch)) { + throw new TorchError('Existing private-ref integration request does not match this adoption', { + code: 'ADOPTION_INTEGRATION_REQUEST_CONFLICT', details: { adoptionId: requesting.id, integrationRequestId: existing.id }, + }); + } + const request = existing ?? requestIntegration({ adoptionId: requesting.id, areaId: requesting.contributorArea, commit: requesting.candidateSha, candidateRef: requesting.candidateRef }); - const requested = { ...requesting, revision: requesting.revision + 1, state: 'requested', integrationRequestId: request.id, requestedAt: this.clock().toISOString() }; - this.#append(requesting.id, requesting.revision, requested, 'integration.adoption.request'); - return { ...requested, integrationRequest: request, mutationPerformed: true }; + const requested = { ...requesting, revision: requesting.revision + 1, state: 'requested', integrationRequestId: request.id, requestedAt: this.clock().toISOString() }; + this.#appendInTransaction(requesting.id, requesting.revision, requested, 'integration.adoption.request'); + this.controlPlane.database.exec('COMMIT'); + return { ...requested, integrationRequest: request, mutationPerformed: true }; + } catch (error) { + this.controlPlane.database.exec('ROLLBACK'); + throw error; + } } #find(adoptionId) { diff --git a/src/integration/service.mjs b/src/integration/service.mjs index a5e1f9d..e027dab 100644 --- a/src/integration/service.mjs +++ b/src/integration/service.mjs @@ -80,8 +80,6 @@ export class IntegrationService { ); CREATE INDEX IF NOT EXISTS integration_state_created ON integration_requests(state, created_at, id); - CREATE UNIQUE INDEX IF NOT EXISTS integration_request_exact_source - ON integration_requests(project_id, source_branch, source_commit); `); } diff --git a/test/fleet/integration-adoption.test.mjs b/test/fleet/integration-adoption.test.mjs index a692194..741149e 100644 --- a/test/fleet/integration-adoption.test.mjs +++ b/test/fleet/integration-adoption.test.mjs @@ -1,6 +1,6 @@ import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; -import { execFileSync } from 'node:child_process'; +import { execFileSync, spawn } from 'node:child_process'; import { mkdtempSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { basename, join } from 'node:path'; @@ -47,7 +47,43 @@ function fixture(name) { const control = openControlPlane({ repositoryRoot: root, env }); const service = new IntegrationService({ repositoryRoot: root, controlPlane: control, checkService: { exactPasses: () => false } }); const contributor = proposal.domains[0].id; - return { root, control, service, contributor, destination: 'review-area', base, candidate }; + return { root, env, control, service, contributor, destination: 'review-area', base, candidate }; +} + +function concurrentAdoptedRequest(context, requestInput) { + const childSource = ` + import { pathToFileURL } from 'node:url'; + import { resolve } from 'node:path'; + const { repositoryRoot, stateRoot, requestInput } = JSON.parse(process.argv[1]); + const sourceRoot = process.cwd(); + const { openControlPlane } = await import(pathToFileURL(resolve(sourceRoot, 'src/control-plane/service.mjs')).href); + const { IntegrationService } = await import(pathToFileURL(resolve(sourceRoot, 'src/integration/service.mjs')).href); + const control = openControlPlane({ repositoryRoot, env: { ...process.env, XDG_DATA_HOME: stateRoot } }); + const service = new IntegrationService({ repositoryRoot, controlPlane: control, checkService: { exactPasses: () => false } }); + try { + const result = service.requestAdoptedCandidate(requestInput); + console.log(JSON.stringify({ outcome: 'requested', integrationRequestId: result.integrationRequest.id })); + } catch (error) { + console.log(JSON.stringify({ outcome: 'refused', code: error.code ?? null })); + } finally { + control.close(); + } + `; + const encoded = JSON.stringify({ repositoryRoot: context.root, stateRoot: context.env.XDG_DATA_HOME, requestInput }); + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, ['--input-type=module', '-e', childSource, encoded], { + cwd: process.cwd(), env: process.env, stdio: ['ignore', 'pipe', 'pipe'], + }); + let stdout = ''; let stderr = ''; + child.stdout.on('data', chunk => { stdout += chunk; }); + child.stderr.on('data', chunk => { stderr += chunk; }); + child.on('error', reject); + child.on('close', (code) => { + if (code !== 0) return reject(new Error(`concurrent adopted request child failed: ${stderr}`)); + try { resolve(JSON.parse(stdout.trim().split('\n').at(-1))); } + catch (error) { reject(new Error(`invalid concurrent adopted request output: ${stdout}\n${stderr}\n${error.message}`)); } + }); + }); } function authority(context, { action, adoptionId }) { @@ -266,22 +302,37 @@ test('SCN-integration-adoption-approval-boundaries: unrelated or source-only app wrongProvenance.control.close(); }); -test('SCN-integration-adoption-request-replay: real SQLite rejects a duplicate exact private-ref queue row, and a completed adoption request cannot replay into a second row', () => { +test('SCN-integration-adoption-request-replay: concurrent authenticated API callers serialize on the adoption journal and create exactly one private-ref queue request', async () => { const context = fixture('torch-adoption-request-replay'); const adoptionId = 'fixture-adoption'; prepare(context, adoptionId); - const finalized = context.service.finalizeAdoption(lifecycleInput(context, 'finalize', adoptionId)); - const requested = context.service.requestAdoptedCandidate(lifecycleInput(context, 'request', adoptionId)); - const duplicate = context.control.database.prepare(`INSERT INTO integration_requests ( + context.service.finalizeAdoption(lifecycleInput(context, 'finalize', adoptionId)); + const requestInput = lifecycleInput(context, 'request', adoptionId); + const outcomes = await Promise.all([ + concurrentAdoptedRequest(context, requestInput), + concurrentAdoptedRequest(context, requestInput), + ]); + assert.deepEqual(outcomes.map((entry) => entry.outcome).sort(), ['refused', 'requested']); + assert.equal(outcomes.find((entry) => entry.outcome === 'refused').code, 'ADOPTION_REQUEST_STATE_INVALID'); + const requests = context.service.list(); + assert.equal(requests.length, 1); + assert.equal(context.service.getPreparedAdoption(adoptionId).state, 'requested'); + context.control.close(); +}); + +test('SCN-integration-adoption-legacy-retries: historical ordinary-request retries remain readable when integration service starts', () => { + const context = fixture('torch-adoption-legacy-retries'); + const insert = context.control.database.prepare(`INSERT INTO integration_requests ( id, project_id, source_area, source_branch, source_commit, target_branch, base_target_commit, state, reason, required_checks, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`); - assert.throws(() => duplicate.run( - 'duplicate-race-row', context.control.projectId, context.contributor, finalized.candidateRef, context.candidate, - 'main', context.base, 'testing', null, '[]', requested.requestedAt, requested.requestedAt, - ), /UNIQUE constraint failed: integration_requests.project_id, integration_requests.source_branch, integration_requests.source_commit/); - assert.throws(() => context.service.requestAdoptedCandidate(lifecycleInput(context, 'request', adoptionId)), - (error) => error.code === 'ADOPTION_REQUEST_STATE_INVALID'); - assert.deepEqual(context.service.list().map((record) => record.id), [requested.integrationRequest.id]); + for (const [id, state] of [['historical-failed', 'blocked'], ['historical-retry', 'superseded']]) { + insert.run(id, context.control.projectId, context.contributor, 'refs/heads/candidate', context.candidate, + 'main', context.base, state, 'historical-retry', '[]', '2026-10-03T00:00:00.000Z', '2026-10-03T00:00:00.000Z'); + } + assert.doesNotThrow(() => new IntegrationService({ + repositoryRoot: context.root, controlPlane: context.control, checkService: { exactPasses: () => false }, + })); + assert.equal(context.service.list().length, 2); context.control.close(); });