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
63 changes: 57 additions & 6 deletions apps/server/src/provider/ProviderRegistry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1677,10 +1677,12 @@ it.layer(
slashCommandsPending: true,
});
const cacheInvalidations = yield* Ref.make(0);
const scanGate = yield* Ref.make<{
readonly started: Deferred.Deferred<void>;
readonly release: Deferred.Deferred<void>;
} | null>(null);
const scanGate = yield* Ref.make<
ReadonlyArray<{
readonly started: Deferred.Deferred<void>;
readonly release: Deferred.Deferred<void>;
}>
>([]);
const returnPendingSnapshot = yield* Ref.make(true);
const probeStarted = yield* Deferred.make<void>();
const releaseProbe = yield* Deferred.make<void>();
Expand Down Expand Up @@ -1721,7 +1723,10 @@ it.layer(
yield* Deferred.succeed(probeStarted, undefined);
yield* Deferred.await(releaseProbe);
const result = yield* Ref.get(scopedResult);
const gate = yield* Ref.getAndSet(scanGate, null);
const gate = yield* Ref.modify(
scanGate,
(gates) => [gates[0], gates.slice(1)] as const,
);
if (gate) {
yield* Deferred.succeed(gate.started, undefined);
yield* Deferred.await(gate.release);
Expand Down Expand Up @@ -1849,7 +1854,7 @@ it.layer(
// newer scan that finished first.
const slowStarted = yield* Deferred.make<void>();
const releaseSlow = yield* Deferred.make<void>();
yield* Ref.set(scanGate, { started: slowStarted, release: releaseSlow });
yield* Ref.set(scanGate, [{ started: slowStarted, release: releaseSlow }]);
yield* Ref.set(scopedResult, scopedProvider);
const slowScan = yield* registry
.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace", fresh: true })
Expand All @@ -1872,6 +1877,52 @@ it.layer(
[latestSkills],
);

// When no cwd snapshot exists yet, an older ordinary scan can be in
// flight when a newer forced scan starts. The forced scan must own
// the cwd before the older result is allowed to publish.
const olderStarted = yield* Deferred.make<void>();
const releaseOlder = yield* Deferred.make<void>();
const newerStarted = yield* Deferred.make<void>();
const releaseNewer = yield* Deferred.make<void>();
yield* Ref.set(scanGate, [
{ started: olderStarted, release: releaseOlder },
{ started: newerStarted, release: releaseNewer },
]);
const olderSkills = [
...scopedProvider.skills,
{ name: "older", path: "/workspace2/older/SKILL.md", enabled: true },
];
const newerSkills = [
...scopedProvider.skills,
{ name: "forced", path: "/workspace2/forced/SKILL.md", enabled: true },
];
yield* Ref.set(scopedResult, { ...scopedProvider, skills: olderSkills });
const olderScan = yield* registry
.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace2" })
.pipe(Effect.forkChild);
yield* Deferred.await(olderStarted);
yield* Ref.set(scopedResult, { ...scopedProvider, skills: newerSkills });
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.deepStrictEqual(
(yield* registry.getProviders)[0]?.workspaceSnapshots?.find(
(snapshot) => snapshot.cwd === "/workspace2",
)?.skills,
undefined,
);
yield* Deferred.succeed(releaseNewer, undefined);
yield* Fiber.join(newerScan);
assert.deepStrictEqual(
(yield* registry.getProviders)[0]?.workspaceSnapshots?.find(
(snapshot) => snapshot.cwd === "/workspace2",
)?.skills,
newerSkills,
);

yield* Ref.set(instancesRef, [rebuiltInstance]);
yield* PubSub.publish(registryChanges, undefined);
let rebuilt = yield* registry.getProviders;
Expand Down
49 changes: 29 additions & 20 deletions apps/server/src/provider/ProviderRegistry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -522,8 +522,12 @@ export const layer = Layer.effect(
const providersRef = yield* Ref.make<ReadonlyArray<ServerProvider>>(
cachedProviders.map((provider) => classifyCompatibility(provider, initialManifest)),
);
// A fresh scan supersedes an older scan for the same instance/cwd. Keep
// an ownership token rather than only a running set so an older scan
// cannot publish after the newer scan has taken over. The existing
// snapshot-equality guard keeps session writes ahead of a scan.
const workspaceRefreshesRef = yield* Ref.make<
ReadonlyMap<ProviderInstance, ReadonlySet<string>>
ReadonlyMap<ProviderInstance, ReadonlyMap<string, object>>
>(new Map());
const maintenanceActionStatesRef = yield* Ref.make<
ReadonlyMap<ProviderInstanceId, { readonly update?: ServerProviderUpdateState | undefined }>
Expand Down Expand Up @@ -1075,15 +1079,16 @@ export const layer = Layer.effect(
const scannedAt = DateTime.formatIso(now);
const instance = yield* instanceRegistry.getInstance(input.instanceId);
if (!instance?.snapshotForCwd) return providers;
const token = {};
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;
});
// A fresh scan never joins a running one, which may predate the change.
if (!claimed && !input.fresh) return yield* Ref.get(providersRef);
if (!claimed) return yield* Ref.get(providersRef);
// Fresh scans also re-read the machine snapshot after invalidating caches.
const refreshMachineSnapshot = input.fresh
? (instance.invalidateCaches ?? Effect.void).pipe(
Expand All @@ -1098,33 +1103,37 @@ export const layer = Layer.effect(
: instanceRegistry.getInstance(input.instanceId).pipe(
Effect.flatMap((currentInstance) => {
if (currentInstance !== instance) return Ref.get(providersRef);
// Write only if the cwd's snapshot did not change during the
// scan. A session event or another scan that landed first is newer.
return updateProviders((currentProviders) =>
currentProviders.map((candidate) =>
// Check ownership in the same atomic provider update so
// an older scan cannot publish after a newer claim.
return updateProviders((currentProviders) => {
const refreshes = Ref.getUnsafe(workspaceRefreshesRef);
if (refreshes.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, {
...scopedSnapshot,
checkedAt: scannedAt,
})
: 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