From 0bd27f89877b1613166c8cf4974015cc40624e30 Mon Sep 17 00:00:00 2001 From: Vic Wang Date: Mon, 14 Sep 2026 09:14:41 +0800 Subject: [PATCH 1/2] fix(marketplace): serialize registry writes to prevent lost updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit validateAllPlugins validates every configured plugin in parallel. Each plugin backed by a remote marketplace independently calls updateMarketplace(), which loads the shared marketplaces.json, mutates its own in-memory copy, and saves — with no coordination between concurrent callers. Two calls racing on the same file let the one that saves last silently overwrite the other's already-persisted change (a lost update), and on Windows the concurrent temp-file renames onto the same destination can also throw EPERM, failing `allagents update` outright. Add an in-process, per-registry-path async lock and route every load-mutate-save sequence through it (updateMarketplace's final save, refreshMarketplace, removeInvalidMarketplaceRegistration). Each writer now reloads the latest saved state before merging its own change in, instead of overwriting with a stale snapshot loaded before the queue. --- src/core/marketplace.ts | 194 ++++++++++++------ .../marketplace-update-concurrency.test.ts | 108 ++++++++++ 2 files changed, 236 insertions(+), 66 deletions(-) create mode 100644 tests/unit/core/marketplace-update-concurrency.test.ts diff --git a/src/core/marketplace.ts b/src/core/marketplace.ts index df94ed63..e94aa4a6 100644 --- a/src/core/marketplace.ts +++ b/src/core/marketplace.ts @@ -365,6 +365,41 @@ export async function saveRegistryToPath( } } +/** Tail of the in-process write queue for each registry path, keyed by resolved path. */ +const registryLockTails = new Map>(); + +/** + * Serialize read-modify-write access to one registry file within this process. + * + * Every caller that loads a registry, changes it, and saves it back + * (updateMarketplace, refreshMarketplace, removeInvalidMarketplaceRegistration) + * runs its whole load-mutate-save sequence through this queue. Without it, + * two calls racing on the same shared `marketplaces.json` — e.g. validating + * several plugins in parallel, each auto-updating its own marketplace — can + * each load a snapshot, save independently, and have the second save silently + * discard the first save's change. On Windows the concurrent renames onto the + * same destination can also throw EPERM instead of just losing data. + * + * This only protects against concurrent writers within one process; it is + * not a cross-process file lock. + */ +function withRegistryLock( + registryPath: string, + fn: () => Promise, +): Promise { + const key = resolve(registryPath); + const previousTail = registryLockTails.get(key) ?? Promise.resolve(); + const result = previousTail.then(fn); + registryLockTails.set( + key, + result.then( + () => undefined, + () => undefined, + ), + ); + return result; +} + /** * Load marketplace registry from disk */ @@ -991,14 +1026,14 @@ export async function updateMarketplace( let projectDirty = false; for (const registration of toUpdateScoped) { - const { entry: marketplace, key, scope } = registration; + const { entry: marketplace, scope } = registration; const accessError = getMarketplaceAccessError(marketplace); if (accessError) { - const registry = scope === 'user' ? userRegistry : projectRegistry; + // removeInvalidMarketplaceRegistration reloads and saves the registry + // itself under withRegistryLock, so it already persisted the removal — + // the final save below reloads fresh and won't resurrect this entry. const removal = await removeInvalidMarketplaceRegistration(registration); - if (removal.removed && registry) { - deleteRegistryMarketplace(registry, key); - } else if (!removal.removed) { + if (!removal.removed) { blockedSaveScopes.add(scope); } invalidRegistrations.add(registration); @@ -1085,20 +1120,32 @@ export async function updateMarketplace( } } - // Save updated timestamps back to the appropriate registries + // Save updated timestamps back to the appropriate registries. + // + // Reload each registry fresh right before merging, under withRegistryLock: + // this call and any other updateMarketplace()/refreshMarketplace() call + // racing on the same shared registry path (e.g. one per plugin, validated + // in parallel — see validateAllPlugins) each apply only their own entries + // on top of the latest saved state, instead of overwriting each other with + // a stale in-memory snapshot loaded at the top of this function. for (const registration of toUpdateScoped) { - const { entry, key, scope } = registration; if (invalidRegistrations.has(registration)) continue; - if (scope === 'user') { - setRegistryMarketplace(userRegistry, key, entry); + if (registration.scope === 'user') { userDirty = true; } else if (projectRegistry) { - setRegistryMarketplace(projectRegistry, key, entry); projectDirty = true; } } if (userDirty && !blockedSaveScopes.has('user')) { - await saveRegistry(userRegistry); + await withRegistryLock(userRegistryPath, async () => { + const currentRegistry = await loadRegistryFromPath(userRegistryPath); + for (const registration of toUpdateScoped) { + if (registration.scope !== 'user' || invalidRegistrations.has(registration)) + continue; + setRegistryMarketplace(currentRegistry, registration.key, registration.entry); + } + await saveRegistryToPath(currentRegistry, userRegistryPath); + }); } if ( projectDirty && @@ -1106,7 +1153,18 @@ export async function updateMarketplace( projectRegistry && projectRegistryPath ) { - await saveRegistryToPath(projectRegistry, projectRegistryPath); + await withRegistryLock(projectRegistryPath, async () => { + const currentRegistry = await loadRegistryFromPath(projectRegistryPath); + for (const registration of toUpdateScoped) { + if ( + registration.scope !== 'project' || + invalidRegistrations.has(registration) + ) + continue; + setRegistryMarketplace(currentRegistry, registration.key, registration.entry); + } + await saveRegistryToPath(currentRegistry, projectRegistryPath); + }); } return results; @@ -1448,25 +1506,27 @@ 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 withRegistryLock(registration.registryPath, async () => { + 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: false, - error: `Marketplace registration '${registration.key}' changed before unsafe cleanup. No registry entry or filesystem path was removed; retry the command.`, + removed: true, + error: getInvalidMarketplaceRegistrationError( + registration.key, + registration.entry, + ), }; - } - deleteRegistryMarketplace(registry, registration.key); - await saveRegistryToPath(registry, registration.registryPath); - return { - success: false, - removed: true, - error: getInvalidMarketplaceRegistrationError( - registration.key, - registration.entry, - ), - }; + }); } function hasSameMarketplaceIdentity( @@ -1580,51 +1640,53 @@ 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)) { + return withRegistryLock(registration.registryPath, async () => { + 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 rename(backupPath, managedPath); + await rm(managedPath, { recursive: true, force: true }); } catch (error) { - restoreError = 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}.`; + 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}`, + }; } - 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); + setRegistryMarketplace(registry, registration.key, refreshedMarketplace); + await saveRegistryToPath(registry, registration.registryPath); - if (hadExistingCache) { - await rm(backupPath, { recursive: true, force: true }).catch(() => {}); - } + if (hadExistingCache) { + await rm(backupPath, { recursive: true, force: true }).catch(() => {}); + } - return { success: true, marketplace: refreshedMarketplace, replaced: true }; + return { success: true, marketplace: refreshedMarketplace, replaced: true }; + }); } /** 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..dcefb3f1 --- /dev/null +++ b/tests/unit/core/marketplace-update-concurrency.test.ts @@ -0,0 +1,108 @@ +import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; +import { mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { updateMarketplace } 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, + }; +} + +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: '2024-01-01T00:00:00.000Z', + }, + 'test-mp-b': { + name: 'test-mp-b', + source: { type: 'github', location: 'owner/test-mp-b' }, + path: marketplacePathB, + lastUpdated: '2024-01-01T00:00:00.000Z', + }, + }, + }), + ); + }); + + afterEach(() => { + restoreHomeDir(); + rmSync(testHome, { recursive: true, force: true }); + }); + + it('persists both updates when two updateMarketplace calls race on the same shared registry file', async () => { + // Simulates `allagents update` validating two plugins in parallel, each + // backed by a different marketplace (validateAllPlugins uses + // Promise.all). Marketplace A's git pull is slower, so its + // updateMarketplace() call loads the registry before B's call has saved + // its own update, then finishes (and saves) after B has already + // persisted. A naive load-mutate-save must not let A's save silently + // discard B's already-saved update. + const callA = updateMarketplace('test-mp-a', undefined, { + createGit: () => createMockGit(), + pull: async () => { + await new Promise((resolve) => setTimeout(resolve, 30)); + }, + }); + const callB = updateMarketplace('test-mp-b', undefined, { + createGit: () => createMockGit(), + pull: async () => undefined, + }); + + const [resultA, resultB] = await Promise.all([callA, callB]); + + expect(resultA[0]?.success).toBe(true); + expect(resultB[0]?.success).toBe(true); + + const registry = JSON.parse(readFileSync(registryPath, 'utf-8')); + expect(registry.marketplaces['test-mp-a'].lastUpdated).not.toBe( + '2024-01-01T00:00:00.000Z', + ); + expect(registry.marketplaces['test-mp-b'].lastUpdated).not.toBe( + '2024-01-01T00:00:00.000Z', + ); + }); +}); From 334215d3f8f8cbad5a3fdf4fed36316412196699 Mon Sep 17 00:00:00 2001 From: Christopher Tso Date: Mon, 14 Sep 2026 17:48:25 +1000 Subject: [PATCH 2/2] fix(marketplace): complete registry transaction locking --- src/core/marketplace.ts | 352 ++++++++++-------- .../unit/core/marketplace-add-branch.test.ts | 55 +++ .../marketplace-update-concurrency.test.ts | 191 ++++++++-- 3 files changed, 411 insertions(+), 187 deletions(-) diff --git a/src/core/marketplace.ts b/src/core/marketplace.ts index e94aa4a6..4f9a01e8 100644 --- a/src/core/marketplace.ts +++ b/src/core/marketplace.ts @@ -365,38 +365,48 @@ export async function saveRegistryToPath( } } -/** Tail of the in-process write queue for each registry path, keyed by resolved path. */ -const registryLockTails = new Map>(); +/** Tail of the in-process mutation queue for each resolved registry path. */ +const registryMutationTails = new Map>(); + +interface RegistryMutation { + result: T; + changed: boolean; +} /** - * Serialize read-modify-write access to one registry file within this process. - * - * Every caller that loads a registry, changes it, and saves it back - * (updateMarketplace, refreshMarketplace, removeInvalidMarketplaceRegistration) - * runs its whole load-mutate-save sequence through this queue. Without it, - * two calls racing on the same shared `marketplaces.json` — e.g. validating - * several plugins in parallel, each auto-updating its own marketplace — can - * each load a snapshot, save independently, and have the second save silently - * discard the first save's change. On Windows the concurrent renames onto the - * same destination can also throw EPERM instead of just losing data. + * 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. + * This only protects against concurrent writers within one process; it is not + * a cross-process file lock. */ -function withRegistryLock( +function mutateRegistry( registryPath: string, - fn: () => Promise, + mutate: ( + registry: MarketplaceRegistry, + ) => RegistryMutation | Promise>, ): Promise { const key = resolve(registryPath); - const previousTail = registryLockTails.get(key) ?? Promise.resolve(); - const result = previousTail.then(fn); - registryLockTails.set( - key, - result.then( - () => undefined, - () => undefined, - ), + 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; } @@ -601,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. @@ -615,8 +627,6 @@ export async function addMarketplace( } return parsed.location; })(); - const existingBySource = findBySourceLocation(registry, sourceLocation); - let alreadyRegistered = !!existingBySource; let marketplacePath: string; let clonedMarketplace = false; @@ -704,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') { @@ -740,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 }), + }, + }; + }); } /** @@ -792,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) @@ -816,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) @@ -961,6 +975,7 @@ interface MarketplaceUpdateGitClient { interface MarketplaceUpdateDeps { createGit(path: string): MarketplaceUpdateGitClient; pull(path: string): Promise; + now(): Date; } /** @@ -1020,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, scope } = registration; + const { entry: marketplace } = registration; const accessError = getMarketplaceAccessError(marketplace); if (accessError) { - // removeInvalidMarketplaceRegistration reloads and saves the registry - // itself under withRegistryLock, so it already persisted the removal — - // the final save below reloads fresh and won't resurrect this entry. const removal = await removeInvalidMarketplaceRegistration(registration); - if (!removal.removed) { - blockedSaveScopes.add(scope); - } - invalidRegistrations.add(registration); results.push({ name: marketplace.name, success: false, @@ -1104,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, @@ -1120,50 +1131,40 @@ export async function updateMarketplace( } } - // Save updated timestamps back to the appropriate registries. - // - // Reload each registry fresh right before merging, under withRegistryLock: - // this call and any other updateMarketplace()/refreshMarketplace() call - // racing on the same shared registry path (e.g. one per plugin, validated - // in parallel — see validateAllPlugins) each apply only their own entries - // on top of the latest saved state, instead of overwriting each other with - // a stale in-memory snapshot loaded at the top of this function. - for (const registration of toUpdateScoped) { - if (invalidRegistrations.has(registration)) continue; - if (registration.scope === 'user') { - userDirty = true; - } else if (projectRegistry) { - 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 withRegistryLock(userRegistryPath, async () => { - const currentRegistry = await loadRegistryFromPath(userRegistryPath); - for (const registration of toUpdateScoped) { - if (registration.scope !== 'user' || invalidRegistrations.has(registration)) - continue; - setRegistryMarketplace(currentRegistry, registration.key, registration.entry); - } - await saveRegistryToPath(currentRegistry, userRegistryPath); - }); - } - if ( - projectDirty && - !blockedSaveScopes.has('project') && - projectRegistry && - projectRegistryPath - ) { - await withRegistryLock(projectRegistryPath, async () => { - const currentRegistry = await loadRegistryFromPath(projectRegistryPath); - for (const registration of toUpdateScoped) { + + 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 ( - registration.scope !== 'project' || - invalidRegistrations.has(registration) - ) + !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; - setRegistryMarketplace(currentRegistry, registration.key, registration.entry); + } + if (currentEntry.lastUpdated !== registration.entry.lastUpdated) { + continue; + } + currentEntry.lastUpdated = lastUpdated; + changed = true; } - await saveRegistryToPath(currentRegistry, projectRegistryPath); + return { changed, result: undefined }; }); } @@ -1506,27 +1507,37 @@ interface InvalidMarketplaceRemovalResult extends MarketplaceResult { async function removeInvalidMarketplaceRegistration( registration: MarketplaceRegistration, ): Promise { - return withRegistryLock(registration.registryPath, async () => { - const registry = await loadRegistryFromPath(registration.registryPath); - const currentEntry = getRegistryMarketplace(registry, registration.key); - if (!currentEntry || !hasSameMarketplaceIdentity(currentEntry, 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 { - success: false, - removed: false, - error: `Marketplace registration '${registration.key}' changed before unsafe cleanup. No registry entry or filesystem path was removed; retry the command.`, + changed: true, + result: { + success: false, + removed: true, + error: getInvalidMarketplaceRegistrationError( + registration.key, + registration.entry, + ), + }, }; - } - deleteRegistryMarketplace(registry, registration.key); - await saveRegistryToPath(registry, registration.registryPath); - return { - success: false, - removed: true, - error: getInvalidMarketplaceRegistrationError( - registration.key, - registration.entry, - ), - }; - }); + }, + ); } function hasSameMarketplaceIdentity( @@ -1640,53 +1651,64 @@ async function refreshMarketplace( path: managedPath, lastUpdated: new Date().toISOString(), }; - return withRegistryLock(registration.registryPath, async () => { - 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)) { + 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 rename(backupPath, managedPath); + await rm(managedPath, { recursive: true, force: true }); } catch (error) { - restoreError = 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}.`; + 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 { - success: false, - error: `Marketplace registration '${registration.key}' changed during refresh. The registry was not overwritten.${recoveryMessage}`, + changed: true, + result: { + success: true, + marketplace: refreshedMarketplace, + replaced: true, + }, }; - } - setRegistryMarketplace(registry, registration.key, refreshedMarketplace); - await saveRegistryToPath(registry, registration.registryPath); - - if (hadExistingCache) { - await rm(backupPath, { recursive: true, force: true }).catch(() => {}); - } + }, + ); - return { success: true, marketplace: refreshedMarketplace, replaced: true }; - }); + if (result.success && hadExistingCache) { + await rm(backupPath, { recursive: true, force: true }).catch(() => {}); + } + 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 index dcefb3f1..ddf67268 100644 --- a/tests/unit/core/marketplace-update-concurrency.test.ts +++ b/tests/unit/core/marketplace-update-concurrency.test.ts @@ -1,8 +1,12 @@ import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; -import { mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; -import { updateMarketplace } from '../../../src/core/marketplace.js'; +import { + removeMarketplace, + updateMarketplace, + type MarketplaceRegistry, +} from '../../../src/core/marketplace.js'; import { stubHomeDir } from '../../helpers/env.js'; function createMockGit() { @@ -15,6 +19,20 @@ function createMockGit() { }; } +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; @@ -55,13 +73,13 @@ describe('updateMarketplace concurrency', () => { name: 'test-mp-a', source: { type: 'github', location: 'owner/test-mp-a' }, path: marketplacePathA, - lastUpdated: '2024-01-01T00:00:00.000Z', + lastUpdated: INITIAL_LAST_UPDATED, }, 'test-mp-b': { name: 'test-mp-b', source: { type: 'github', location: 'owner/test-mp-b' }, path: marketplacePathB, - lastUpdated: '2024-01-01T00:00:00.000Z', + lastUpdated: INITIAL_LAST_UPDATED, }, }, }), @@ -73,36 +91,165 @@ describe('updateMarketplace concurrency', () => { rmSync(testHome, { recursive: true, force: true }); }); - it('persists both updates when two updateMarketplace calls race on the same shared registry file', async () => { - // Simulates `allagents update` validating two plugins in parallel, each - // backed by a different marketplace (validateAllPlugins uses - // Promise.all). Marketplace A's git pull is slower, so its - // updateMarketplace() call loads the registry before B's call has saved - // its own update, then finishes (and saves) after B has already - // persisted. A naive load-mutate-save must not let A's save silently - // discard B's already-saved update. + 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 () => { - await new Promise((resolve) => setTimeout(resolve, 30)); + aPullReached.resolve(); + await resumeAPull.promise; }, }); - const callB = updateMarketplace('test-mp-b', undefined, { + + await aPullReached.promise; + + const resultB = await updateMarketplace('test-mp-b', undefined, { createGit: () => createMockGit(), pull: async () => undefined, }); + const entryBAfterUpdate = readRegistry(registryPath).marketplaces['test-mp-b']; - const [resultA, resultB] = await Promise.all([callA, callB]); + resumeAPull.resolve(); + const resultA = await callA; + const finalRegistry = readRegistry(registryPath); - expect(resultA[0]?.success).toBe(true); - expect(resultB[0]?.success).toBe(true); + 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, + ); + }); - const registry = JSON.parse(readFileSync(registryPath, 'utf-8')); - expect(registry.marketplaces['test-mp-a'].lastUpdated).not.toBe( - '2024-01-01T00:00:00.000Z', + 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'], + }, + }), ); - expect(registry.marketplaces['test-mp-b'].lastUpdated).not.toBe( - '2024-01-01T00:00:00.000Z', + + 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); + }); });