mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(coding-agent): avoid custom store lock convoys
This commit is contained in:
parent
a79b22622a
commit
2c79ce4532
5 changed files with 82 additions and 37 deletions
|
|
@ -22,10 +22,16 @@ type LockResult<T> = {
|
|||
|
||||
const AUTH_FILE_WRITE_OPTIONS = { encoding: "utf-8", mode: 0o600 } as const;
|
||||
|
||||
type AuthFileReload = {
|
||||
controller: AbortController;
|
||||
promise: Promise<AuthStorageData>;
|
||||
readers: number;
|
||||
};
|
||||
|
||||
type AuthFileReadState = {
|
||||
data: AuthStorageData;
|
||||
revision?: string;
|
||||
reload?: Promise<AuthStorageData>;
|
||||
reload?: AuthFileReload;
|
||||
};
|
||||
|
||||
let sharedAuthFileReadState: { authPath: string; readState: AuthFileReadState } | undefined;
|
||||
|
|
@ -232,16 +238,14 @@ export class AuthStorage implements CredentialStore {
|
|||
|
||||
private constructor(storage: AuthStorageBackend, authPath?: string) {
|
||||
this.storage = storage;
|
||||
this.readState = { data: {} };
|
||||
this.authPath = authPath;
|
||||
this.readState =
|
||||
authPath && sharedAuthFileReadState?.authPath === authPath ? sharedAuthFileReadState.readState : { data: {} };
|
||||
if (authPath && !sharedAuthFileReadState) {
|
||||
this.authPath = authPath;
|
||||
sharedAuthFileReadState = { authPath, readState: this.readState };
|
||||
} else if (authPath && sharedAuthFileReadState?.authPath === authPath) {
|
||||
this.authPath = authPath;
|
||||
this.readState = sharedAuthFileReadState.readState;
|
||||
}
|
||||
if (this.authPath) {
|
||||
const revision = getFileRevision(this.authPath);
|
||||
if (authPath) {
|
||||
const revision = getFileRevision(authPath);
|
||||
if (revision !== undefined && revision === this.readState.revision) return;
|
||||
}
|
||||
this.reload();
|
||||
|
|
@ -301,25 +305,44 @@ export class AuthStorage implements CredentialStore {
|
|||
}, options);
|
||||
}
|
||||
|
||||
private readLatestData(options?: AuthOperationOptions): Promise<AuthStorageData> {
|
||||
private async readLatestData(options?: AuthOperationOptions): Promise<AuthStorageData> {
|
||||
options?.signal?.throwIfAborted();
|
||||
if (!this.authPath) {
|
||||
const reload = this.reloadFromStorageAsync(options);
|
||||
return options?.signal ? reload : reload.catch(() => this.readState.data);
|
||||
}
|
||||
const revision = getFileRevision(this.authPath);
|
||||
if (revision !== undefined && revision === this.readState.revision) {
|
||||
return Promise.resolve(this.readState.data);
|
||||
}
|
||||
if (options?.signal) return this.reloadFromStorageAsync(options);
|
||||
if (revision !== undefined && revision === this.readState.revision) return this.readState.data;
|
||||
if (!this.readState.reload) {
|
||||
const reload = this.reloadFromStorageAsync().catch(() => this.readState.data);
|
||||
const controller = new AbortController();
|
||||
const reload: AuthFileReload = {
|
||||
controller,
|
||||
promise: this.reloadFromStorageAsync({ signal: controller.signal }),
|
||||
readers: 0,
|
||||
};
|
||||
this.readState.reload = reload;
|
||||
void reload.then(() => {
|
||||
if (this.readState.reload === reload) this.readState.reload = undefined;
|
||||
});
|
||||
void reload.promise.then(
|
||||
() => {
|
||||
if (this.readState.reload === reload) this.readState.reload = undefined;
|
||||
},
|
||||
() => {
|
||||
if (this.readState.reload === reload) this.readState.reload = undefined;
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
const reload = this.readState.reload;
|
||||
reload.readers++;
|
||||
try {
|
||||
const result = raceWithAbortSignal(reload.promise, options?.signal);
|
||||
return options?.signal ? await result : await result.catch(() => this.readState.data);
|
||||
} finally {
|
||||
reload.readers--;
|
||||
if (reload.readers === 0 && this.readState.reload === reload) {
|
||||
this.readState.reload = undefined;
|
||||
reload.controller.abort();
|
||||
}
|
||||
}
|
||||
return this.readState.reload;
|
||||
}
|
||||
|
||||
async read(provider: string, options?: AuthOperationOptions): Promise<Credential | undefined> {
|
||||
|
|
|
|||
|
|
@ -46,16 +46,15 @@ export class InMemoryCodingAgentModelsStore implements ModelsStore {
|
|||
export class FileModelsStore implements ModelsStore {
|
||||
private readonly storage: AuthStorageBackend;
|
||||
private readonly path: string;
|
||||
private readonly readState: ModelsFileReadState | undefined;
|
||||
private readonly readState: ModelsFileReadState;
|
||||
|
||||
constructor(path: string = join(getAgentDir(), "models-store.json")) {
|
||||
this.path = normalizePath(path);
|
||||
this.storage = new FileAuthStorageBackend(this.path);
|
||||
this.readState =
|
||||
sharedModelsFileReadState?.path === this.path ? sharedModelsFileReadState.readState : { data: {} };
|
||||
if (!sharedModelsFileReadState) {
|
||||
this.readState = { data: {} };
|
||||
sharedModelsFileReadState = { path: this.path, readState: this.readState };
|
||||
} else if (sharedModelsFileReadState.path === this.path) {
|
||||
this.readState = sharedModelsFileReadState.readState;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -118,12 +117,6 @@ export class FileModelsStore implements ModelsStore {
|
|||
}
|
||||
|
||||
async read(providerId: string, options?: ModelsStoreOperationOptions): Promise<ModelsStoreEntry | undefined> {
|
||||
if (!this.readState) {
|
||||
return this.storage.withLockAsync(
|
||||
async (content) => ({ result: structuredClone(this.parse(content)[providerId]) }),
|
||||
options,
|
||||
);
|
||||
}
|
||||
const entry = (await this.readLatest(this.readState, options))[providerId];
|
||||
options?.signal?.throwIfAborted();
|
||||
return entry ? structuredClone(entry) : undefined;
|
||||
|
|
@ -137,7 +130,7 @@ export class FileModelsStore implements ModelsStore {
|
|||
latest = current;
|
||||
return { result: undefined, next: JSON.stringify(current, null, 2) };
|
||||
}, options);
|
||||
if (this.readState && latest) this.updateReadState(this.readState, latest);
|
||||
if (latest) this.updateReadState(this.readState, latest);
|
||||
}
|
||||
|
||||
async delete(providerId: string, options?: ModelsStoreOperationOptions): Promise<void> {
|
||||
|
|
@ -148,6 +141,6 @@ export class FileModelsStore implements ModelsStore {
|
|||
latest = current;
|
||||
return { result: undefined, next: JSON.stringify(current, null, 2) };
|
||||
}, options);
|
||||
if (this.readState && latest) this.updateReadState(this.readState, latest);
|
||||
if (latest) this.updateReadState(this.readState, latest);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -81,9 +81,9 @@ describe("AuthStorage", () => {
|
|||
});
|
||||
|
||||
const [anthropic, openai, credentials] = await Promise.all([
|
||||
first.read("anthropic"),
|
||||
second.read("openai"),
|
||||
first.list(),
|
||||
first.read("anthropic", { signal: new AbortController().signal }),
|
||||
second.read("openai", { signal: new AbortController().signal }),
|
||||
first.list({ signal: new AbortController().signal }),
|
||||
]);
|
||||
expect(anthropic).toEqual({ type: "api_key", key: "new" });
|
||||
expect(openai).toEqual({ type: "api_key", key: "openai-key" });
|
||||
|
|
@ -103,14 +103,40 @@ describe("AuthStorage", () => {
|
|||
await otherFirst.read("other");
|
||||
await otherSecond.read("other");
|
||||
await otherFirst.list();
|
||||
expect(lockSpy).toHaveBeenCalledTimes(4);
|
||||
expect(lockSpy).toHaveBeenCalledTimes(1);
|
||||
|
||||
const third = AuthStorage.create(authJsonPath);
|
||||
writeAuthJson({ anthropic: { type: "api_key", key: "newest" } });
|
||||
const [firstReload, thirdReload] = await Promise.all([first.read("anthropic"), third.read("anthropic")]);
|
||||
expect(firstReload).toEqual({ type: "api_key", key: "newest" });
|
||||
expect(thirdReload).toEqual({ type: "api_key", key: "newest" });
|
||||
expect(lockSpy).toHaveBeenCalledTimes(5);
|
||||
expect(lockSpy).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
test("keeps a coalesced reload alive while another credential reader is waiting", async () => {
|
||||
writeAuthJson({ anthropic: { type: "api_key", key: "old" } });
|
||||
const storage = AuthStorage.create(authJsonPath);
|
||||
writeAuthJson({ anthropic: { type: "api_key", key: "new" } });
|
||||
let grantLock: (() => void) | undefined;
|
||||
const lockGranted = new Promise<void>((resolve) => {
|
||||
grantLock = resolve;
|
||||
});
|
||||
const release = vi.fn(async () => {});
|
||||
const lockSpy = vi.spyOn(lockfile, "lock").mockImplementation(async () => {
|
||||
await lockGranted;
|
||||
return release;
|
||||
});
|
||||
const firstController = new AbortController();
|
||||
const secondController = new AbortController();
|
||||
const first = storage.read("anthropic", { signal: firstController.signal });
|
||||
const second = storage.read("anthropic", { signal: secondController.signal });
|
||||
|
||||
firstController.abort();
|
||||
await expect(first).rejects.toMatchObject({ name: "AbortError" });
|
||||
grantLock?.();
|
||||
await expect(second).resolves.toEqual({ type: "api_key", key: "new" });
|
||||
expect(lockSpy).toHaveBeenCalledTimes(1);
|
||||
expect(release).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test("modify persists a credential while preserving unrelated external edits", async () => {
|
||||
|
|
|
|||
|
|
@ -88,7 +88,7 @@ describe("FileModelsStore", () => {
|
|||
const [firstReload, thirdReload] = await Promise.all([first.read("one"), third.read("one")]);
|
||||
expect(firstReload).toMatchObject({ models: [{ id: "newest-model" }] });
|
||||
expect(thirdReload).toMatchObject({ models: [{ id: "newest-model" }] });
|
||||
expect(lockSpy).toHaveBeenCalledTimes(4);
|
||||
expect(lockSpy).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("keeps a coalesced reload alive while another reader is still waiting", async () => {
|
||||
|
|
|
|||
|
|
@ -89,7 +89,10 @@ describe("test harness", () => {
|
|||
});
|
||||
|
||||
it("turns a pending terminal response into an error", async () => {
|
||||
harness = await createHarness({ responses: [{ text: "partial", stopReason: "pending" }] });
|
||||
harness = await createHarness({
|
||||
responses: [{ text: "partial", stopReason: "pending" }],
|
||||
settings: { retry: { enabled: false } },
|
||||
});
|
||||
|
||||
await harness.session.prompt("hi");
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue