Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions apps/server/src/provider/ProviderRegistry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2121,6 +2121,44 @@ it.layer(
[latestSkills],
);

// With no cached snapshot, an older scan must leave publication and
// request deduplication to the forced scan even when it finishes first.
const olderStarted = yield* Deferred.make<void>();
const releaseOlder = yield* Deferred.make<void>();
yield* Ref.set(scanGate, { started: olderStarted, release: releaseOlder });
yield* Ref.set(scopedResult, scopedProvider);
const olderScan = yield* registry
.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace2" })
.pipe(Effect.forkChild);
yield* Deferred.await(olderStarted);
const newerStarted = yield* Deferred.make<void>();
const releaseNewer = yield* Deferred.make<void>();
yield* Ref.set(scanGate, { started: newerStarted, release: releaseNewer });
yield* Ref.set(scopedResult, { ...scopedProvider, skills: latestSkills });
const newerScan = yield* registry
.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace2", fresh: true })
.pipe(Effect.forkChild);
yield* Deferred.await(newerStarted);
yield* Deferred.succeed(releaseOlder, undefined);
yield* Fiber.join(olderScan);
assert.strictEqual(
(yield* registry.getProviders)[0]?.workspaceSnapshots?.find(
(snapshot) => snapshot.cwd === "/workspace2",
),
undefined,
);
const callsWhileForced = yield* Ref.get(snapshotCalls);
yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace2" });
assert.strictEqual(yield* Ref.get(snapshotCalls), callsWhileForced);
yield* Deferred.succeed(releaseNewer, undefined);
yield* Fiber.join(newerScan);
assert.deepStrictEqual(
(yield* registry.getProviders)[0]?.workspaceSnapshots?.find(
(snapshot) => snapshot.cwd === "/workspace2",
)?.skills,
latestSkills,
);

yield* Ref.set(instancesRef, [rebuiltInstance]);
yield* PubSub.publish(registryChanges, undefined);
let rebuilt = yield* registry.getProviders;
Expand Down
41 changes: 23 additions & 18 deletions apps/server/src/provider/ProviderRegistry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -437,7 +437,7 @@ export const layer = Layer.effect(
cachedProviders.map((provider) => classifyCompatibility(provider, initialManifest)),
);
const workspaceRefreshesRef = yield* Ref.make<
ReadonlyMap<ProviderInstance, ReadonlySet<string>>
ReadonlyMap<ProviderInstance, ReadonlyMap<string, symbol>>
>(new Map());
const maintenanceActionStatesRef = yield* Ref.make<
ReadonlyMap<ProviderInstanceId, { readonly update?: ServerProviderUpdateState | undefined }>
Expand Down Expand Up @@ -897,14 +897,15 @@ export const layer = Layer.effect(
const scannedAt = DateTime.formatIso(now);
const instance = yield* instanceRegistry.getInstance(input.instanceId);
if (!instance?.snapshotForCwd) return providers;
const token = Symbol();
const claimed = yield* Ref.modify(workspaceRefreshesRef, (refreshes) => {
const current = refreshes.get(instance);
if (current?.has(input.cwd)) return [false, refreshes] as const;
if (!input.fresh && current?.has(input.cwd)) return [false, refreshes] as const;
const next = new Map(refreshes);
next.set(instance, new Set(current).add(input.cwd));
next.set(instance, new Map(current).set(input.cwd, token));
return [true, next] as const;
});
if (!claimed && !input.fresh) return yield* Ref.get(providersRef);
if (!claimed) return yield* Ref.get(providersRef);
const refreshMachineSnapshot = input.fresh
? (instance.invalidateCaches ?? Effect.void).pipe(
Effect.andThen(refreshInstance(input.instanceId)),
Expand All @@ -918,8 +919,12 @@ export const layer = Layer.effect(
instanceRegistry.getInstance(input.instanceId).pipe(
Effect.flatMap((currentInstance) => {
if (currentInstance !== instance) return Ref.get(providersRef);
return updateProviders((currentProviders) =>
currentProviders.map((candidate) =>
return updateProviders((currentProviders) => {
// Read ownership inside the update so a newer claim cannot race publication.
if (Ref.getUnsafe(workspaceRefreshesRef).get(instance)?.get(input.cwd) !== token) {
return currentProviders;
}
return currentProviders.map((candidate) =>
candidate.instanceId === input.instanceId &&
Equal.equals(workspaceSnapshotOf(candidate), scannedFrom)
? upsertProviderWorkspaceSnapshot(candidate, input.cwd, {
Expand All @@ -932,22 +937,22 @@ export const layer = Layer.effect(
scopedSnapshot.slashCommandsPending ?? scopedSnapshot.status === "error",
})
: candidate,
),
);
);
});
}),
),
),
Effect.ensuring(
claimed
? Ref.update(workspaceRefreshesRef, (refreshes) => {
const next = new Map(refreshes);
const current = new Set(next.get(instance));
current.delete(input.cwd);
if (current.size) next.set(instance, current);
else next.delete(instance);
return next;
})
: Effect.void,
Ref.update(workspaceRefreshesRef, (refreshes) => {
const current = refreshes.get(instance);
if (current?.get(input.cwd) !== token) return refreshes;
const next = new Map(refreshes);
const nextCwds = new Map(current);
nextCwds.delete(input.cwd);
if (nextCwds.size) next.set(instance, nextCwds);
else next.delete(instance);
return next;
}),
),
);
});
Expand Down
Loading