fix(pi): serialize managed provider catalog writes
This commit is contained in:
1 parent
2ff2e4af79
commit
0d26d17cfc
3 files changed
+56
-11
No files matched your search
@@ -22,6 +22,7 @@ import { getImportedModelProfile } from '../../../shared/imported-model-profile'
|
|||||||
const PI_ENV_PREFIX = 'MAKELORE_PI';
|
const PI_ENV_PREFIX = 'MAKELORE_PI';
|
||||||
const WORKS_SQUARE_AI_GATEWAY_CREDENTIAL_MODE = 'works_square_ai_gateway';
|
const WORKS_SQUARE_AI_GATEWAY_CREDENTIAL_MODE = 'works_square_ai_gateway';
|
||||||
const WORKS_SQUARE_AI_GATEWAY_PROXY_CREDENTIAL_MODE = 'works_square_ai_gateway_proxy';
|
const WORKS_SQUARE_AI_GATEWAY_PROXY_CREDENTIAL_MODE = 'works_square_ai_gateway_proxy';
|
||||||
|
const providerCatalogWriteTails = new Map<string, Promise<void>>();
|
||||||
|
|
||||||
export const PI_PROVIDER_APIS = [
|
export const PI_PROVIDER_APIS = [
|
||||||
'openai-completions',
|
'openai-completions',
|
||||||
@@ -469,7 +470,18 @@ export async function writePiProviderCatalog(
|
|||||||
modelRef?: ProductModelRef,
|
modelRef?: ProductModelRef,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
if (modelRef) selectPiProviderModel(catalog, modelRef);
|
if (modelRef) selectPiProviderModel(catalog, modelRef);
|
||||||
await atomicWriteJson(filePath, catalog.modelsFile);
|
const previous = providerCatalogWriteTails.get(filePath) ?? Promise.resolve();
|
||||||
|
const write = previous
|
||||||
|
.catch(() => undefined)
|
||||||
|
.then(async () => await atomicWriteJson(filePath, catalog.modelsFile));
|
||||||
|
providerCatalogWriteTails.set(filePath, write);
|
||||||
|
try {
|
||||||
|
await write;
|
||||||
|
} finally {
|
||||||
|
if (providerCatalogWriteTails.get(filePath) === write) {
|
||||||
|
providerCatalogWriteTails.delete(filePath);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
function replaceCredentialEnvReferences(value: string, credential: string | null): string {
|
function replaceCredentialEnvReferences(value: string, credential: string | null): string {
|
||||||
|
|||||||
@@ -144,7 +144,7 @@ const EXPECTED_TURN_MILESTONES: readonly ProofMilestone[] = [
|
|||||||
let pressureRun: PressureRun | null = null;
|
let pressureRun: PressureRun | null = null;
|
||||||
|
|
||||||
async function waitFor(predicate: () => boolean, message: string): Promise<void> {
|
async function waitFor(predicate: () => boolean, message: string): Promise<void> {
|
||||||
const deadline = Date.now() + 15_000;
|
const deadline = Date.now() + 30_000;
|
||||||
while (Date.now() < deadline) {
|
while (Date.now() < deadline) {
|
||||||
if (predicate()) return;
|
if (predicate()) return;
|
||||||
await new Promise((resolveWait) => setTimeout(resolveWait, 20));
|
await new Promise((resolveWait) => setTimeout(resolveWait, 20));
|
||||||
@@ -350,7 +350,7 @@ async function createProofProjects(
|
|||||||
for (let index = 0; index < count; index += 1) {
|
for (let index = 0; index < count; index += 1) {
|
||||||
const number = index + 1;
|
const number = index + 1;
|
||||||
const projectPath = path.join(root, `project-${number}`);
|
const projectPath = path.join(root, `project-${number}`);
|
||||||
const project = await createLocalCodingProject({ projectPath, now }, projectStore);
|
const { project } = await createLocalCodingProject({ projectPath, now }, projectStore);
|
||||||
await createCodingProjectAgent(projectPath, {
|
await createCodingProjectAgent(projectPath, {
|
||||||
id: PROOF_AGENT_ID,
|
id: PROOF_AGENT_ID,
|
||||||
avatarId: 'avatar-01',
|
avatarId: 'avatar-01',
|
||||||
@@ -661,7 +661,7 @@ async function startPressureRun(): Promise<PressureRun> {
|
|||||||
`release-proof-write-project-${index + 1}`,
|
`release-proof-write-project-${index + 1}`,
|
||||||
`release-proof-holder-${index + 1}`,
|
`release-proof-holder-${index + 1}`,
|
||||||
));
|
));
|
||||||
dispatches.push(composition.scheduler.dispatch({
|
dispatches.push(composition.scheduler.dispatch({
|
||||||
conversationId: project.input.conversationId,
|
conversationId: project.input.conversationId,
|
||||||
workerGeneration: generation,
|
workerGeneration: generation,
|
||||||
runId,
|
runId,
|
||||||
@@ -676,13 +676,31 @@ async function startPressureRun(): Promise<PressureRun> {
|
|||||||
},
|
},
|
||||||
}));
|
}));
|
||||||
}
|
}
|
||||||
await waitFor(
|
try {
|
||||||
() => composition.processBudget.activeCount === 8
|
await waitFor(
|
||||||
&& composition.scheduler.getDiagnostics().activeChildPermits === 4
|
() => composition.processBudget.activeCount === 8
|
||||||
&& provider.activeCounts().child === 4
|
&& composition.scheduler.getDiagnostics().activeChildPermits === 4
|
||||||
&& processIds(composition.tracked, 'child', true).length === 4,
|
&& provider.activeCounts().child === 4
|
||||||
'Four real ephemeral Pi children did not become active',
|
&& processIds(composition.tracked, 'child', true).length === 4,
|
||||||
);
|
'Four real ephemeral Pi children did not become active',
|
||||||
|
);
|
||||||
|
} catch (error) {
|
||||||
|
const childDiagnostics = composition.tracked
|
||||||
|
.filter(({ role }) => role === 'child')
|
||||||
|
.map(({ process }) => ({
|
||||||
|
pid: process.processId ?? null,
|
||||||
|
running: process.isRunning,
|
||||||
|
diagnostic: process.stderrDiagnostic,
|
||||||
|
}));
|
||||||
|
throw new Error(
|
||||||
|
`${error instanceof Error ? error.message : String(error)}: ${JSON.stringify({
|
||||||
|
snapshot: pressureSnapshot(composition, provider, writeLeases),
|
||||||
|
scheduler: composition.scheduler.getDiagnostics(),
|
||||||
|
providerRequests: providerRequestCounts(provider),
|
||||||
|
childDiagnostics,
|
||||||
|
})}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
const active = pressureSnapshot(composition, provider, writeLeases);
|
const active = pressureSnapshot(composition, provider, writeLeases);
|
||||||
if (active.parentProcessIds.length !== 4
|
if (active.parentProcessIds.length !== 4
|
||||||
|| active.childProcessIds.length !== 4
|
|| active.childProcessIds.length !== 4
|
||||||
|
|||||||
@@ -194,6 +194,21 @@ describe('Pi Provider catalog', () => {
|
|||||||
expect(await readFile(filePath, 'utf8')).toBe('existing-catalog\n');
|
expect(await readFile(filePath, 'utf8')).toBe('existing-catalog\n');
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('serializes concurrent writes to the shared managed catalog', async () => {
|
||||||
|
const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-provider-concurrent-'));
|
||||||
|
temporaryRoots.push(root);
|
||||||
|
const filePath = path.join(root, 'models.json');
|
||||||
|
const catalog = buildPiProviderCatalog({ accounts: [account()] });
|
||||||
|
|
||||||
|
await Promise.all(Array.from({ length: 8 }, async () => await writePiProviderCatalog(
|
||||||
|
filePath,
|
||||||
|
catalog,
|
||||||
|
{ accountId: 'account-one', modelId: 'gpt-5.4', thinkingLevel: 'off' },
|
||||||
|
)));
|
||||||
|
|
||||||
|
expect(JSON.parse(await readFile(filePath, 'utf8'))).toEqual(catalog.modelsFile);
|
||||||
|
});
|
||||||
|
|
||||||
it('fails closed when a non-local credential is unavailable', async () => {
|
it('fails closed when a non-local credential is unavailable', async () => {
|
||||||
const provider = account();
|
const provider = account();
|
||||||
const descriptor = buildPiProviderCatalog({ accounts: [provider] }).descriptors[0]!;
|
const descriptor = buildPiProviderCatalog({ accounts: [provider] }).descriptors[0]!;
|
||||||
|
|||||||
Reference in new issue
Block a user