diff --git a/src/core/marketplace.ts b/src/core/marketplace.ts index df94ed63..4f9a01e8 100644 --- a/src/core/marketplace.ts +++ b/src/core/marketplace.ts @@ -365,6 +365,51 @@ export async function saveRegistryToPath( } } +/** Tail of the in-process mutation queue for each resolved registry path. */ +const registryMutationTails = new Map>(); + +interface RegistryMutation { + result: T; + changed: boolean; +} + +/** + * Run one registry mutation against a fresh snapshot and save it when changed. + * + * This only protects against concurrent writers within one process; it is not + * a cross-process file lock. + */ +function mutateRegistry( + registryPath: string, + mutate: ( + registry: MarketplaceRegistry, + ) => RegistryMutation | Promise>, +): Promise { + const key = resolve(registryPath); + const previousTail = registryMutationTails.get(key) ?? Promise.resolve(); + const result = previousTail.then(async () => { + const registry = await loadRegistryFromPath(key); + const mutation = await mutate(registry); + if (mutation.changed) { + await saveRegistryToPath(registry, key); + } + return mutation.result; + }); + // Queue on a failure-neutral tail while returning the original result/error. + const tail = result.then( + () => undefined, + () => undefined, + ); + registryMutationTails.set(key, tail); + // A successor can replace this tail before its cleanup microtask runs. + void tail.then(() => { + if (registryMutationTails.get(key) === tail) { + registryMutationTails.delete(key); + } + }); + return result; +} + /** * Load marketplace registry from disk */ @@ -566,7 +611,9 @@ export async function addMarketplace( const registryPath = scopeOptions?.scope === 'project' && scopeOptions?.workspacePath ? getProjectRegistryPath(scopeOptions.workspacePath) : getRegistryPath(); - const registry = await loadRegistryFromPath(registryPath); + // Fail before cache work if the existing registry cannot be read. The + // mutation transaction reloads the authoritative snapshot before saving. + await loadRegistryFromPath(registryPath); // Check if already registered by source location (idempotent) // Use the full location including branch — each branch is a separate marketplace. @@ -580,8 +627,6 @@ export async function addMarketplace( } return parsed.location; })(); - const existingBySource = findBySourceLocation(registry, sourceLocation); - let alreadyRegistered = !!existingBySource; let marketplacePath: string; let clonedMarketplace = false; @@ -669,20 +714,11 @@ export async function addMarketplace( }; } if (manifestName !== name) { - // Track if the manifest name is already registered - if (getRegistryMarketplace(registry, manifestName)) { - alreadyRegistered = true; - } name = manifestName; } } } - // Check if already registered by name (after manifest parsing to use final name) - if (getRegistryMarketplace(registry, name)) { - alreadyRegistered = true; - } - // Build location: for GitHub, use owner/repo for default branch, owner/repo/branch for non-default let entryLocation: string; if (parsed.type === 'github') { @@ -705,15 +741,20 @@ export async function addMarketplace( lastUpdated: new Date().toISOString(), }; - // Save to registry - setRegistryMarketplace(registry, name, entry); - await saveRegistryToPath(registry, registryPath); - - return { - success: true, - marketplace: entry, - ...(alreadyRegistered && { replaced: true }), - }; + return mutateRegistry(registryPath, (registry) => { + const alreadyRegistered = + !!findBySourceLocation(registry, sourceLocation) || + !!getRegistryMarketplace(registry, name); + setRegistryMarketplace(registry, name, entry); + return { + changed: true, + result: { + success: true, + marketplace: entry, + ...(alreadyRegistered && { replaced: true }), + }, + }; + }); } /** @@ -757,12 +798,16 @@ export async function removeMarketplace( // Remove from user scope if (scope === 'user' || scope === 'all') { - const userRegistry = await loadRegistryFromPath(userRegPath); - const userEntry = getRegistryMarketplace(userRegistry, name); + const userEntry = await mutateRegistry(userRegPath, (registry) => { + const entry = getRegistryMarketplace(registry, name); + if (!entry) { + return { changed: false, result: undefined }; + } + deleteRegistryMarketplace(registry, name); + return { changed: true, result: entry }; + }); if (userEntry) { removedEntry = userEntry; - deleteRegistryMarketplace(userRegistry, name); - await saveRegistryToPath(userRegistry, userRegPath); if ( removedEntry.source.type !== 'local' && pathEntryExists(removedEntry.path) @@ -781,12 +826,16 @@ export async function removeMarketplace( // Remove from project scope if ((scope === 'project' || scope === 'all') && options.workspacePath) { const projectRegPath = getProjectRegistryPath(options.workspacePath); - const projectRegistry = await loadRegistryFromPath(projectRegPath); - const projectEntry = getRegistryMarketplace(projectRegistry, name); + const projectEntry = await mutateRegistry(projectRegPath, (registry) => { + const entry = getRegistryMarketplace(registry, name); + if (!entry) { + return { changed: false, result: undefined }; + } + deleteRegistryMarketplace(registry, name); + return { changed: true, result: entry }; + }); if (projectEntry) { removedEntry = projectEntry; - deleteRegistryMarketplace(projectRegistry, name); - await saveRegistryToPath(projectRegistry, projectRegPath); if ( removedEntry.source.type !== 'local' && pathEntryExists(removedEntry.path) @@ -926,6 +975,7 @@ interface MarketplaceUpdateGitClient { interface MarketplaceUpdateDeps { createGit(path: string): MarketplaceUpdateGitClient; pull(path: string): Promise; + now(): Date; } /** @@ -985,23 +1035,17 @@ export async function updateMarketplace( return [{ name, success: false, error: `Marketplace '${name}' not found` }]; } - const invalidRegistrations = new Set(); - const blockedSaveScopes = new Set(); - let userDirty = false; - let projectDirty = false; + const successfulUpdates: Array<{ + registration: MarketplaceRegistration; + lastUpdated: string; + resultIndex: number; + }> = []; for (const registration of toUpdateScoped) { - const { entry: marketplace, key, scope } = registration; + const { entry: marketplace } = registration; const accessError = getMarketplaceAccessError(marketplace); if (accessError) { - const registry = scope === 'user' ? userRegistry : projectRegistry; const removal = await removeInvalidMarketplaceRegistration(registration); - if (removal.removed && registry) { - deleteRegistryMarketplace(registry, key); - } else if (!removal.removed) { - blockedSaveScopes.add(scope); - } - invalidRegistrations.add(registration); results.push({ name: marketplace.name, success: false, @@ -1069,13 +1113,15 @@ export async function updateMarketplace( await git.checkout(targetBranch); await (deps.pull ?? pull)(marketplace.path); - // Update lastUpdated in the entry (mutates in place for scope tracking) - marketplace.lastUpdated = new Date().toISOString(); - + const lastUpdated = deps.now + ? deps.now().toISOString() + : new Date().toISOString(); + const resultIndex = results.length; results.push({ name: marketplace.name, success: true, }); + successfulUpdates.push({ registration, lastUpdated, resultIndex }); } catch (error) { results.push({ name: marketplace.name, @@ -1085,28 +1131,41 @@ export async function updateMarketplace( } } - // Save updated timestamps back to the appropriate registries - for (const registration of toUpdateScoped) { - const { entry, key, scope } = registration; - if (invalidRegistrations.has(registration)) continue; - if (scope === 'user') { - setRegistryMarketplace(userRegistry, key, entry); - userDirty = true; - } else if (projectRegistry) { - setRegistryMarketplace(projectRegistry, key, entry); - projectDirty = true; + const updatesByRegistry = new Map(); + for (const update of successfulUpdates) { + const registryPath = resolve(update.registration.registryPath); + const updates = updatesByRegistry.get(registryPath); + if (updates) { + updates.push(update); + } else { + updatesByRegistry.set(registryPath, [update]); } } - if (userDirty && !blockedSaveScopes.has('user')) { - await saveRegistry(userRegistry); - } - if ( - projectDirty && - !blockedSaveScopes.has('project') && - projectRegistry && - projectRegistryPath - ) { - await saveRegistryToPath(projectRegistry, projectRegistryPath); + + for (const [registryPath, updates] of updatesByRegistry) { + await mutateRegistry(registryPath, (registry) => { + let changed = false; + for (const { registration, lastUpdated, resultIndex } of updates) { + const currentEntry = getRegistryMarketplace(registry, registration.key); + if ( + !currentEntry || + !hasSameMarketplaceIdentity(currentEntry, registration.entry) + ) { + results[resultIndex] = { + name: registration.entry.name, + success: false, + error: `Marketplace '${registration.key}' changed during update. The registry was not overwritten; retry the command.`, + }; + continue; + } + if (currentEntry.lastUpdated !== registration.entry.lastUpdated) { + continue; + } + currentEntry.lastUpdated = lastUpdated; + changed = true; + } + return { changed, result: undefined }; + }); } return results; @@ -1448,25 +1507,37 @@ interface InvalidMarketplaceRemovalResult extends MarketplaceResult { async function removeInvalidMarketplaceRegistration( registration: MarketplaceRegistration, ): Promise { - const registry = await loadRegistryFromPath(registration.registryPath); - const currentEntry = getRegistryMarketplace(registry, registration.key); - if (!currentEntry || !hasSameMarketplaceIdentity(currentEntry, registration.entry)) { - return { - success: false, - removed: false, - error: `Marketplace registration '${registration.key}' changed before unsafe cleanup. No registry entry or filesystem path was removed; retry the command.`, - }; - } - deleteRegistryMarketplace(registry, registration.key); - await saveRegistryToPath(registry, registration.registryPath); - return { - success: false, - removed: true, - error: getInvalidMarketplaceRegistrationError( - registration.key, - registration.entry, - ), - }; + return mutateRegistry( + registration.registryPath, + (registry) => { + const currentEntry = getRegistryMarketplace(registry, registration.key); + if ( + !currentEntry || + !hasSameMarketplaceIdentity(currentEntry, registration.entry) + ) { + return { + changed: false, + result: { + success: false, + removed: false, + error: `Marketplace registration '${registration.key}' changed before unsafe cleanup. The registry was not overwritten and no filesystem path was removed; retry the command.`, + }, + }; + } + deleteRegistryMarketplace(registry, registration.key); + return { + changed: true, + result: { + success: false, + removed: true, + error: getInvalidMarketplaceRegistrationError( + registration.key, + registration.entry, + ), + }, + }; + }, + ); } function hasSameMarketplaceIdentity( @@ -1580,51 +1651,64 @@ async function refreshMarketplace( path: managedPath, lastUpdated: new Date().toISOString(), }; - const registry = await loadRegistryFromPath(registration.registryPath); - const currentEntry = getRegistryMarketplace(registry, registration.key); - if (!currentEntry || !hasSameMarketplaceIdentity(currentEntry, marketplace)) { - let cleanupError: unknown; - let restoreError: unknown; - try { - await rm(managedPath, { recursive: true, force: true }); - } catch (error) { - cleanupError = error; - } - if (!cleanupError && hadExistingCache && existsSync(backupPath)) { - try { - await rename(backupPath, managedPath); - } catch (error) { - restoreError = error; - } - } + const result = await mutateRegistry( + registration.registryPath, + async (registry) => { + const currentEntry = getRegistryMarketplace(registry, registration.key); + if (!currentEntry || !hasSameMarketplaceIdentity(currentEntry, marketplace)) { + let cleanupError: unknown; + let restoreError: unknown; + try { + await rm(managedPath, { recursive: true, force: true }); + } catch (error) { + cleanupError = error; + } + if (!cleanupError && hadExistingCache && existsSync(backupPath)) { + try { + await rename(backupPath, managedPath); + } catch (error) { + restoreError = error; + } + } - let recoveryMessage = ''; - if (cleanupError) { - const detail = cleanupError instanceof Error - ? cleanupError.message - : String(cleanupError); - recoveryMessage = hadExistingCache - ? ` Automatic recovery could not remove the replacement cache: ${detail}. The replacement remains at ${managedPath}, and the original cache remains at ${backupPath}.` - : ` Automatic cleanup failed: ${detail}. The replacement cache remains at ${managedPath}.`; - } else if (restoreError) { - const detail = restoreError instanceof Error - ? restoreError.message - : String(restoreError); - recoveryMessage = ` Automatic recovery failed: ${detail}. The original cache remains at ${backupPath}.`; - } - return { - success: false, - error: `Marketplace registration '${registration.key}' changed during refresh. The registry was not overwritten.${recoveryMessage}`, - }; - } - setRegistryMarketplace(registry, registration.key, refreshedMarketplace); - await saveRegistryToPath(registry, registration.registryPath); + let recoveryMessage = ''; + if (cleanupError) { + const detail = cleanupError instanceof Error + ? cleanupError.message + : String(cleanupError); + recoveryMessage = hadExistingCache + ? ` Automatic recovery could not remove the replacement cache: ${detail}. The replacement remains at ${managedPath}, and the original cache remains at ${backupPath}.` + : ` Automatic cleanup failed: ${detail}. The replacement cache remains at ${managedPath}.`; + } else if (restoreError) { + const detail = restoreError instanceof Error + ? restoreError.message + : String(restoreError); + recoveryMessage = ` Automatic recovery failed: ${detail}. The original cache remains at ${backupPath}.`; + } + return { + changed: false, + result: { + success: false, + error: `Marketplace registration '${registration.key}' changed during refresh. The registry was not overwritten; retry the command.${recoveryMessage}`, + }, + }; + } + setRegistryMarketplace(registry, registration.key, refreshedMarketplace); + return { + changed: true, + result: { + success: true, + marketplace: refreshedMarketplace, + replaced: true, + }, + }; + }, + ); - if (hadExistingCache) { + if (result.success && hadExistingCache) { await rm(backupPath, { recursive: true, force: true }).catch(() => {}); } - - return { success: true, marketplace: refreshedMarketplace, replaced: true }; + return result; } /** diff --git a/tests/unit/core/marketplace-add-branch.test.ts b/tests/unit/core/marketplace-add-branch.test.ts index bb0d14d6..bc47f6b3 100644 --- a/tests/unit/core/marketplace-add-branch.test.ts +++ b/tests/unit/core/marketplace-add-branch.test.ts @@ -99,6 +99,61 @@ describe('addMarketplace branch support', () => { expect(result.error).toContain('reserved for the default branch'); }); + it('fails before cloning when the registry is unreadable', async () => { + const registryPath = join(testHome, '.allagents', 'marketplaces.json'); + const cachePath = join( + testHome, + '.allagents', + 'plugins', + 'marketplaces', + 'repo', + ); + writeFileSync(registryPath, '{"version":1,"marketplaces":'); + + await expect(addMarketplace('owner/repo')).rejects.toThrow( + `Marketplace registry at ${registryPath} is unreadable`, + ); + + expect(cloneCalls).toHaveLength(0); + expect(existsSync(cachePath)).toBe(false); + }); + + it('preserves concurrent marketplace registrations', async () => { + let cloneCount = 0; + let signalBothClones!: () => void; + const bothClonesReached = new Promise((resolve) => { + signalBothClones = resolve; + }); + let releaseClones!: () => void; + const clonesReleased = new Promise((resolve) => { + releaseClones = resolve; + }); + cloneToMock.mockImplementation( + async (url: string, dest: string, ref?: string) => { + cloneCalls.push({ url, dest, ref }); + mkdirSync(dest, { recursive: true }); + cloneCount++; + if (cloneCount === 2) signalBothClones(); + await clonesReleased; + }, + ); + + const addA = addMarketplace('owner/repo-a', 'repo-a'); + const addB = addMarketplace('owner/repo-b', 'repo-b'); + await bothClonesReached; + releaseClones(); + + const [resultA, resultB] = await Promise.all([addA, addB]); + const registry = await loadRegistry(); + + expect(resultA.success).toBe(true); + expect(resultB.success).toBe(true); + expect(Object.keys(registry.marketplaces).sort()).toEqual([ + 'repo-a', + 'repo-b', + ]); + }); + it('should clone with branch when --name is provided', async () => { const result = await addMarketplace( 'https://github.com/owner/repo/tree/feat/v2', diff --git a/tests/unit/core/marketplace-update-concurrency.test.ts b/tests/unit/core/marketplace-update-concurrency.test.ts new file mode 100644 index 00000000..ddf67268 --- /dev/null +++ b/tests/unit/core/marketplace-update-concurrency.test.ts @@ -0,0 +1,255 @@ +import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; +import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { + removeMarketplace, + updateMarketplace, + type MarketplaceRegistry, +} from '../../../src/core/marketplace.js'; +import { stubHomeDir } from '../../helpers/env.js'; + +function createMockGit() { + return { + raw: async (args: string[]) => { + if (args[0] === 'symbolic-ref') return 'origin/main'; + return ''; + }, + checkout: async () => undefined, + }; +} + +const INITIAL_LAST_UPDATED = '2024-01-01T00:00:00.000Z'; + +function deferred() { + let resolve!: () => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +function readRegistry(path: string): MarketplaceRegistry { + return JSON.parse(readFileSync(path, 'utf-8')) as MarketplaceRegistry; +} + +describe('updateMarketplace concurrency', () => { + let restoreHomeDir: () => void; + let testHome: string; + let registryPath: string; + let marketplacePathA: string; + let marketplacePathB: string; + + beforeEach(() => { + testHome = join(tmpdir(), `marketplace-update-race-${Date.now()}`); + restoreHomeDir = stubHomeDir(testHome); + + marketplacePathA = join( + testHome, + '.allagents', + 'plugins', + 'marketplaces', + 'test-mp-a', + ); + marketplacePathB = join( + testHome, + '.allagents', + 'plugins', + 'marketplaces', + 'test-mp-b', + ); + mkdirSync(marketplacePathA, { recursive: true }); + mkdirSync(marketplacePathB, { recursive: true }); + + const registryDir = join(testHome, '.allagents'); + mkdirSync(registryDir, { recursive: true }); + registryPath = join(registryDir, 'marketplaces.json'); + writeFileSync( + registryPath, + JSON.stringify({ + version: 1, + marketplaces: { + 'test-mp-a': { + name: 'test-mp-a', + source: { type: 'github', location: 'owner/test-mp-a' }, + path: marketplacePathA, + lastUpdated: INITIAL_LAST_UPDATED, + }, + 'test-mp-b': { + name: 'test-mp-b', + source: { type: 'github', location: 'owner/test-mp-b' }, + path: marketplacePathB, + lastUpdated: INITIAL_LAST_UPDATED, + }, + }, + }), + ); + }); + + afterEach(() => { + restoreHomeDir(); + rmSync(testHome, { recursive: true, force: true }); + }); + + it('preserves a completed marketplace update when an earlier update finishes later', async () => { + const aPullReached = deferred(); + const resumeAPull = deferred(); + const callA = updateMarketplace('test-mp-a', undefined, { + createGit: () => createMockGit(), + pull: async () => { + aPullReached.resolve(); + await resumeAPull.promise; + }, + }); + + await aPullReached.promise; + + const resultB = await updateMarketplace('test-mp-b', undefined, { + createGit: () => createMockGit(), + pull: async () => undefined, + }); + const entryBAfterUpdate = readRegistry(registryPath).marketplaces['test-mp-b']; + + resumeAPull.resolve(); + const resultA = await callA; + const finalRegistry = readRegistry(registryPath); + + expect(resultA).toEqual([{ name: 'test-mp-a', success: true }]); + expect(resultB).toEqual([{ name: 'test-mp-b', success: true }]); + expect(entryBAfterUpdate?.lastUpdated).not.toBe(INITIAL_LAST_UPDATED); + expect(finalRegistry.marketplaces['test-mp-b']).toEqual(entryBAfterUpdate); + expect(finalRegistry.marketplaces['test-mp-a']?.lastUpdated).not.toBe( + INITIAL_LAST_UPDATED, + ); + }); + + it('preserves a successful named update when update-all previously failed that entry', async () => { + const registry = readRegistry(registryPath); + writeFileSync( + registryPath, + JSON.stringify({ + version: 1, + marketplaces: { + 'test-mp-b': registry.marketplaces['test-mp-b'], + 'test-mp-a': registry.marketplaces['test-mp-a'], + }, + }), + ); + + const aPullReached = deferred(); + const resumeAPull = deferred(); + const updateAll = updateMarketplace(undefined, undefined, { + createGit: () => createMockGit(), + pull: async (path) => { + if (path === marketplacePathB) { + throw new Error('B pull failed'); + } + expect(path).toBe(marketplacePathA); + aPullReached.resolve(); + await resumeAPull.promise; + }, + }); + + await aPullReached.promise; + + const namedBResult = await updateMarketplace('test-mp-b', undefined, { + createGit: () => createMockGit(), + pull: async () => undefined, + }); + const entryBAfterNamedUpdate = + readRegistry(registryPath).marketplaces['test-mp-b']; + + resumeAPull.resolve(); + const updateAllResult = await updateAll; + const finalRegistry = readRegistry(registryPath); + + expect(namedBResult).toEqual([{ name: 'test-mp-b', success: true }]); + expect(updateAllResult).toEqual([ + { name: 'test-mp-b', success: false, error: 'B pull failed' }, + { name: 'test-mp-a', success: true }, + ]); + expect(entryBAfterNamedUpdate?.lastUpdated).not.toBe(INITIAL_LAST_UPDATED); + expect(finalRegistry.marketplaces['test-mp-b']).toEqual(entryBAfterNamedUpdate); + expect(finalRegistry.marketplaces['test-mp-a']?.lastUpdated).not.toBe( + INITIAL_LAST_UPDATED, + ); + }); + + it('preserves a newer timestamp from a concurrent named update', async () => { + const olderUpdate = new Date('2024-02-01T00:00:00.000Z'); + const newerUpdate = new Date('2024-03-01T00:00:00.000Z'); + const bUpdate = new Date('2024-04-01T00:00:00.000Z'); + const bPullReached = deferred(); + const resumeBPull = deferred(); + const updateTimes = [olderUpdate, bUpdate]; + const updateAll = updateMarketplace(undefined, undefined, { + createGit: () => createMockGit(), + pull: async (path) => { + if (path === marketplacePathB) { + bPullReached.resolve(); + await resumeBPull.promise; + } + }, + now: () => updateTimes.shift()!, + }); + + await bPullReached.promise; + + const namedAResult = await updateMarketplace('test-mp-a', undefined, { + createGit: () => createMockGit(), + pull: async () => undefined, + now: () => newerUpdate, + }); + + resumeBPull.resolve(); + const updateAllResult = await updateAll; + const finalRegistry = readRegistry(registryPath); + + expect(namedAResult).toEqual([{ name: 'test-mp-a', success: true }]); + expect(updateAllResult).toEqual([ + { name: 'test-mp-a', success: true }, + { name: 'test-mp-b', success: true }, + ]); + expect(finalRegistry.marketplaces['test-mp-a']?.lastUpdated).toBe( + newerUpdate.toISOString(), + ); + expect(finalRegistry.marketplaces['test-mp-b']?.lastUpdated).toBe( + bUpdate.toISOString(), + ); + }); + + it('keeps removal authoritative when an in-flight named update finishes afterward', async () => { + const pullReached = deferred(); + const resumePull = deferred(); + const update = updateMarketplace('test-mp-a', undefined, { + createGit: () => createMockGit(), + pull: async () => { + pullReached.resolve(); + await resumePull.promise; + }, + }); + + await pullReached.promise; + + const removeResult = await removeMarketplace('test-mp-a'); + const registryAfterRemoval = readRegistry(registryPath); + expect(removeResult.success).toBe(true); + expect(registryAfterRemoval.marketplaces['test-mp-a']).toBeUndefined(); + expect(existsSync(marketplacePathA)).toBe(false); + + resumePull.resolve(); + const updateResult = await update; + const finalRegistry = readRegistry(registryPath); + + expect(updateResult).toEqual([ + { + name: 'test-mp-a', + success: false, + error: + "Marketplace 'test-mp-a' changed during update. The registry was not overwritten; retry the command.", + }, + ]); + expect(finalRegistry).toEqual(registryAfterRemoval); + expect(existsSync(marketplacePathA)).toBe(false); + }); +});