fix(app): bound Android provider snapshot storage (#3234)
Android AsyncStorage's 6 MiB SQLite ceiling was exhausted by per-workspace provider snapshots with no eviction. Raise the native cap to 10 MiB and bound snapshots to 4 MiB with indexed eviction, clearing legacy unindexed data without scanning payloads on every save.
This commit is contained in:
Mohamed Boudra
@@ -1,6 +1,7 @@
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const pkg = require("./package.json");
|
||||
const withAndroidAsyncStorageSize = require("./plugins/with-android-async-storage-size");
|
||||
const withAndroidProfileable = require("./plugins/with-android-profileable");
|
||||
const withFdroidAutolinking = require("./plugins/with-fdroid-autolinking");
|
||||
const { getNativeReleaseVersion } = require("./native-release-version");
|
||||
@@ -138,6 +139,7 @@ export default {
|
||||
},
|
||||
plugins: [
|
||||
"expo-router",
|
||||
[withAndroidAsyncStorageSize, 10],
|
||||
...buildProfile.cameraPlugins,
|
||||
[
|
||||
"expo-splash-screen",
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
const { withGradleProperties } = require("expo/config-plugins");
|
||||
|
||||
const ASYNC_STORAGE_SIZE_PROPERTY = "AsyncStorage_db_size_in_MB";
|
||||
|
||||
function withAndroidAsyncStorageSize(config, sizeInMegabytes) {
|
||||
return withGradleProperties(config, (modConfig) => {
|
||||
const existingProperty = modConfig.modResults.find(
|
||||
(property) => property.type === "property" && property.key === ASYNC_STORAGE_SIZE_PROPERTY,
|
||||
);
|
||||
if (existingProperty?.type === "property") {
|
||||
existingProperty.value = String(sizeInMegabytes);
|
||||
return modConfig;
|
||||
}
|
||||
|
||||
modConfig.modResults.push({
|
||||
type: "property",
|
||||
key: ASYNC_STORAGE_SIZE_PROPERTY,
|
||||
value: String(sizeInMegabytes),
|
||||
});
|
||||
return modConfig;
|
||||
});
|
||||
}
|
||||
|
||||
module.exports = withAndroidAsyncStorageSize;
|
||||
@@ -0,0 +1,27 @@
|
||||
import { execFileSync } from "node:child_process";
|
||||
import path from "node:path";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
describe("Android AsyncStorage configuration", () => {
|
||||
it("writes a 10 MiB database limit into Gradle properties", () => {
|
||||
const appDirectory = path.resolve(import.meta.dirname, "..");
|
||||
const output = execFileSync(
|
||||
process.execPath,
|
||||
[
|
||||
"-e",
|
||||
`
|
||||
const plugin = require("./plugins/with-android-async-storage-size");
|
||||
const config = plugin({ name: "Paseo", slug: "paseo" }, 10);
|
||||
const mod = config.mods.android.gradleProperties;
|
||||
mod({ ...config, modResults: [], modRequest: {}, modRawConfig: config })
|
||||
.then((result) => process.stdout.write(JSON.stringify(result.modResults)));
|
||||
`,
|
||||
],
|
||||
{ cwd: appDirectory, encoding: "utf8" },
|
||||
);
|
||||
|
||||
expect(JSON.parse(output)).toEqual([
|
||||
{ type: "property", key: "AsyncStorage_db_size_in_MB", value: "10" },
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -1,24 +1,144 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { Buffer } from "buffer";
|
||||
import type { ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
|
||||
import { compactProviderSnapshot } from "@getpaseo/protocol/provider-snapshot-codec";
|
||||
import { createProviderSnapshotCache } from "./provider-snapshot-cache";
|
||||
import { z } from "zod";
|
||||
import { createProviderSnapshotCache, type ProviderSnapshotCache } from "./provider-snapshot-cache";
|
||||
|
||||
function createStorage() {
|
||||
const SNAPSHOT_KEY_PREFIX = "@paseo/provider-snapshot/v1:";
|
||||
const SNAPSHOT_INDEX_KEY = "@paseo/provider-snapshot-index/v1";
|
||||
const SnapshotIndexSchema = z.object({
|
||||
version: z.literal(1),
|
||||
entries: z.array(z.object({ key: z.string(), bytes: z.number(), writtenAt: z.string() })),
|
||||
});
|
||||
|
||||
function createStorage(maxSnapshotBytes = Number.POSITIVE_INFINITY) {
|
||||
const values = new Map<string, string>();
|
||||
const stats = { getAllKeysCalls: 0 };
|
||||
type Operation = "setItem" | "removeItem" | "multiRemove";
|
||||
let failure: { operation: Operation; timing: "before" | "after"; key?: string } | null = null;
|
||||
|
||||
function takeFailure(operation: Operation, timing: "before" | "after", key?: string): boolean {
|
||||
if (
|
||||
failure?.operation !== operation ||
|
||||
failure.timing !== timing ||
|
||||
(failure.key !== undefined && failure.key !== key)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
failure = null;
|
||||
return true;
|
||||
}
|
||||
|
||||
return {
|
||||
values,
|
||||
stats,
|
||||
failNext(operation: Operation, timing: "before" | "after", key?: string) {
|
||||
failure = { operation, timing, key };
|
||||
},
|
||||
async getItem(key: string) {
|
||||
return values.get(key) ?? null;
|
||||
},
|
||||
async setItem(key: string, value: string) {
|
||||
if (takeFailure("setItem", "before", key)) throw new Error("Injected setItem failure");
|
||||
if (key.startsWith(SNAPSHOT_KEY_PREFIX)) {
|
||||
const bytesWithoutReplacement = snapshotBytes(
|
||||
new Map([...values].filter(([storedKey]) => storedKey !== key)),
|
||||
);
|
||||
if (
|
||||
bytesWithoutReplacement +
|
||||
Buffer.byteLength(key, "utf8") +
|
||||
Buffer.byteLength(value, "utf8") >
|
||||
maxSnapshotBytes
|
||||
) {
|
||||
throw new Error("Injected storage capacity failure");
|
||||
}
|
||||
}
|
||||
values.set(key, value);
|
||||
if (takeFailure("setItem", "after", key)) throw new Error("Injected setItem failure");
|
||||
},
|
||||
async removeItem(key: string) {
|
||||
if (takeFailure("removeItem", "before", key)) throw new Error("Injected removeItem failure");
|
||||
values.delete(key);
|
||||
if (takeFailure("removeItem", "after", key)) throw new Error("Injected removeItem failure");
|
||||
},
|
||||
async getAllKeys() {
|
||||
stats.getAllKeysCalls += 1;
|
||||
return [...values.keys()];
|
||||
},
|
||||
async multiGet(keys: readonly string[]) {
|
||||
return keys.map((key) => [key, values.get(key) ?? null] as const);
|
||||
},
|
||||
async multiRemove(keys: readonly string[]) {
|
||||
if (takeFailure("multiRemove", "before")) throw new Error("Injected multiRemove failure");
|
||||
for (const [index, key] of keys.entries()) {
|
||||
values.delete(key);
|
||||
if (index === 0 && takeFailure("multiRemove", "after")) {
|
||||
throw new Error("Injected partial multiRemove failure");
|
||||
}
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function snapshotEntries(label: string): ProviderSnapshotEntry[] {
|
||||
return [
|
||||
{
|
||||
provider: "pi",
|
||||
status: "ready",
|
||||
enabled: true,
|
||||
models: [
|
||||
{
|
||||
provider: "pi",
|
||||
id: label,
|
||||
label: label.repeat(20),
|
||||
thinkingOptions: [],
|
||||
},
|
||||
],
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function snapshotBytes(values: Map<string, string>): number {
|
||||
return [...values]
|
||||
.filter(([key]) => key.startsWith(SNAPSHOT_KEY_PREFIX))
|
||||
.reduce(
|
||||
(total, [key, value]) =>
|
||||
total + Buffer.byteLength(key, "utf8") + Buffer.byteLength(value, "utf8"),
|
||||
0,
|
||||
);
|
||||
}
|
||||
|
||||
function expectIndexMatchesSnapshots(values: Map<string, string>): void {
|
||||
const storedIndex = values.get(SNAPSHOT_INDEX_KEY);
|
||||
if (!storedIndex) throw new Error("Expected provider snapshot index");
|
||||
const index = SnapshotIndexSchema.parse(JSON.parse(storedIndex));
|
||||
const storedSnapshots = [...values]
|
||||
.filter(([key]) => key.startsWith(SNAPSHOT_KEY_PREFIX))
|
||||
.map(([key, value]) => ({
|
||||
key,
|
||||
bytes: Buffer.byteLength(key, "utf8") + Buffer.byteLength(value, "utf8"),
|
||||
}))
|
||||
.toSorted((left, right) => left.key.localeCompare(right.key));
|
||||
const indexedSnapshots = index.entries
|
||||
.map(({ key, bytes }) => ({ key, bytes }))
|
||||
.toSorted((left, right) => left.key.localeCompare(right.key));
|
||||
expect(indexedSnapshots).toEqual(storedSnapshots);
|
||||
}
|
||||
|
||||
function writeSnapshot(
|
||||
cache: ProviderSnapshotCache,
|
||||
input: { cwd: string; label: string; generatedAt: string },
|
||||
): Promise<void> {
|
||||
return cache.write({
|
||||
serverId: "server-1",
|
||||
cwd: input.cwd,
|
||||
hash: input.label,
|
||||
generatedAt: input.generatedAt,
|
||||
compactSnapshot: compactProviderSnapshot(snapshotEntries(input.label)),
|
||||
});
|
||||
}
|
||||
|
||||
describe("provider snapshot cache", () => {
|
||||
it("round-trips compact snapshots with shared thinking option references", async () => {
|
||||
const thinkingOptions = [{ id: "high", label: "High", isDefault: true }];
|
||||
@@ -60,6 +180,356 @@ describe("provider snapshot cache", () => {
|
||||
storage.values.set('@paseo/provider-snapshot/v1:["server-1","/repo"]', "not json");
|
||||
|
||||
await expect(cache.read("server-1", "/repo")).resolves.toBeNull();
|
||||
expect(storage.values.size).toBe(0);
|
||||
expect([...storage.values.keys()].some((key) => key.startsWith(SNAPSHOT_KEY_PREFIX))).toBe(
|
||||
false,
|
||||
);
|
||||
});
|
||||
|
||||
it("evicts the oldest snapshots to keep the cache within its byte budget", async () => {
|
||||
const storage = createStorage();
|
||||
const maxBytes = 1_000;
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes });
|
||||
|
||||
await Promise.all([
|
||||
cache.write({
|
||||
serverId: "server-1",
|
||||
cwd: "/oldest",
|
||||
hash: "oldest",
|
||||
generatedAt: "2026-08-01T00:00:00.000Z",
|
||||
compactSnapshot: compactProviderSnapshot(snapshotEntries("oldest")),
|
||||
}),
|
||||
cache.write({
|
||||
serverId: "server-1",
|
||||
cwd: "/middle",
|
||||
hash: "middle",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
compactSnapshot: compactProviderSnapshot(snapshotEntries("middle")),
|
||||
}),
|
||||
cache.write({
|
||||
serverId: "server-1",
|
||||
cwd: "/newest",
|
||||
hash: "newest",
|
||||
generatedAt: "2026-08-03T00:00:00.000Z",
|
||||
compactSnapshot: compactProviderSnapshot(snapshotEntries("newest")),
|
||||
}),
|
||||
]);
|
||||
|
||||
await expect(cache.read("server-1", "/oldest")).resolves.toBeNull();
|
||||
await expect(cache.read("server-1", "/newest")).resolves.not.toBeNull();
|
||||
expect(snapshotBytes(storage.values)).toBeLessThanOrEqual(maxBytes);
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
expect(storage.stats.getAllKeysCalls).toBe(1);
|
||||
});
|
||||
|
||||
it("clears unindexed snapshots once when creating the cache index", async () => {
|
||||
const storage = createStorage();
|
||||
storage.values.set('@paseo/provider-snapshot/v1:["server-1","/legacy"]', "legacy");
|
||||
const cache = createProviderSnapshotCache(storage);
|
||||
|
||||
await cache.write({
|
||||
serverId: "server-1",
|
||||
cwd: "/current",
|
||||
hash: "current",
|
||||
generatedAt: "2026-08-03T00:00:00.000Z",
|
||||
compactSnapshot: compactProviderSnapshot(snapshotEntries("current")),
|
||||
});
|
||||
|
||||
expect(storage.values.has('@paseo/provider-snapshot/v1:["server-1","/legacy"]')).toBe(false);
|
||||
await expect(cache.read("server-1", "/current")).resolves.not.toBeNull();
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
});
|
||||
|
||||
it("clears unindexed snapshots on the first read", async () => {
|
||||
const storage = createStorage();
|
||||
storage.values.set('@paseo/provider-snapshot/v1:["server-1","/legacy"]', "legacy");
|
||||
const cache = createProviderSnapshotCache(storage);
|
||||
|
||||
await expect(cache.read("server-1", "/legacy")).resolves.toBeNull();
|
||||
|
||||
expect(storage.values.has('@paseo/provider-snapshot/v1:["server-1","/legacy"]')).toBe(false);
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
expect(storage.stats.getAllKeysCalls).toBe(1);
|
||||
});
|
||||
|
||||
it("initializes the cache index once when the first read and write race", async () => {
|
||||
const storage = createStorage();
|
||||
storage.values.set('@paseo/provider-snapshot/v1:["server-1","/legacy"]', "legacy");
|
||||
const cache = createProviderSnapshotCache(storage);
|
||||
|
||||
await Promise.all([
|
||||
cache.read("server-1", "/legacy"),
|
||||
writeSnapshot(cache, {
|
||||
cwd: "/current",
|
||||
label: "current",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
}),
|
||||
]);
|
||||
|
||||
await expect(cache.read("server-1", "/current")).resolves.not.toBeNull();
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
expect(storage.stats.getAllKeysCalls).toBe(1);
|
||||
});
|
||||
|
||||
it("replaces an indexed snapshot without counting both versions", async () => {
|
||||
const storage = createStorage();
|
||||
const maxBytes = 1_000;
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes });
|
||||
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/repo",
|
||||
label: "first",
|
||||
generatedAt: "2026-08-01T00:00:00.000Z",
|
||||
});
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/repo",
|
||||
label: "replacement-is-larger",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
|
||||
await expect(cache.read("server-1", "/repo")).resolves.toMatchObject({
|
||||
hash: "replacement-is-larger",
|
||||
});
|
||||
expect(snapshotBytes(storage.values)).toBeLessThanOrEqual(maxBytes);
|
||||
expect(
|
||||
[...storage.values.keys()].filter((key) => key.startsWith(SNAPSHOT_KEY_PREFIX)),
|
||||
).toHaveLength(1);
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
});
|
||||
|
||||
it("removes the previous value when a replacement exceeds the entire budget", async () => {
|
||||
const storage = createStorage();
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes: 1_000 });
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/repo",
|
||||
label: "cached",
|
||||
generatedAt: "2026-08-01T00:00:00.000Z",
|
||||
});
|
||||
|
||||
const tinyCache = createProviderSnapshotCache(storage, { maxBytes: 1 });
|
||||
await writeSnapshot(tinyCache, {
|
||||
cwd: "/repo",
|
||||
label: "too-large",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
|
||||
await expect(tinyCache.read("server-1", "/repo")).resolves.toBeNull();
|
||||
expect(snapshotBytes(storage.values)).toBe(0);
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
});
|
||||
|
||||
it("resets corrupted index state before accepting another snapshot", async () => {
|
||||
const storage = createStorage();
|
||||
storage.values.set(SNAPSHOT_INDEX_KEY, "corrupt");
|
||||
storage.values.set(`${SNAPSHOT_KEY_PREFIX}legacy`, "legacy");
|
||||
const cache = createProviderSnapshotCache(storage);
|
||||
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/current",
|
||||
label: "current",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
|
||||
expect(storage.values.has(`${SNAPSHOT_KEY_PREFIX}legacy`)).toBe(false);
|
||||
await expect(cache.read("server-1", "/current")).resolves.not.toBeNull();
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
expect(storage.stats.getAllKeysCalls).toBe(1);
|
||||
});
|
||||
|
||||
it("measures UTF-8 bytes rather than JavaScript string length", async () => {
|
||||
const stagingStorage = createStorage();
|
||||
const stagingCache = createProviderSnapshotCache(stagingStorage);
|
||||
await writeSnapshot(stagingCache, {
|
||||
cwd: "/emoji",
|
||||
label: "🧨".repeat(100),
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
const storedSnapshot = [...stagingStorage.values].find(([key]) =>
|
||||
key.startsWith(SNAPSHOT_KEY_PREFIX),
|
||||
);
|
||||
if (!storedSnapshot) throw new Error("Expected staged provider snapshot");
|
||||
const [key, value] = storedSnapshot;
|
||||
const codeUnitBytes = key.length + value.length;
|
||||
const utf8Bytes = Buffer.byteLength(key, "utf8") + Buffer.byteLength(value, "utf8");
|
||||
expect(utf8Bytes).toBeGreaterThan(codeUnitBytes);
|
||||
|
||||
const storage = createStorage();
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes: codeUnitBytes });
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/emoji",
|
||||
label: "🧨".repeat(100),
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
|
||||
await expect(cache.read("server-1", "/emoji")).resolves.toBeNull();
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
});
|
||||
|
||||
it("keeps hundreds of concurrent writes bounded without rescanning snapshots", async () => {
|
||||
const storage = createStorage();
|
||||
const maxBytes = 8_000;
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes });
|
||||
const writes = Array.from({ length: 250 }, (_, index) =>
|
||||
writeSnapshot(cache, {
|
||||
cwd: `/repo-${index}`,
|
||||
label: `snapshot-${index}-${"x".repeat(index % 40)}`,
|
||||
generatedAt: new Date(Date.UTC(2026, 0, 1, 0, 0, index)).toISOString(),
|
||||
}),
|
||||
);
|
||||
|
||||
await Promise.all(writes);
|
||||
|
||||
expect(snapshotBytes(storage.values)).toBeLessThanOrEqual(maxBytes);
|
||||
await expect(cache.read("server-1", "/repo-249")).resolves.not.toBeNull();
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
expect(storage.stats.getAllKeysCalls).toBe(1);
|
||||
});
|
||||
|
||||
it.each(["before", "after"] as const)(
|
||||
"recovers when a snapshot write fails %s its side effect",
|
||||
async (timing) => {
|
||||
const storage = createStorage();
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes: 2_000 });
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/stable",
|
||||
label: "stable",
|
||||
generatedAt: "2026-08-01T00:00:00.000Z",
|
||||
});
|
||||
const failedKey = `${SNAPSHOT_KEY_PREFIX}["server-1","/failed"]`;
|
||||
storage.failNext("setItem", timing, failedKey);
|
||||
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/failed",
|
||||
label: "failed",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
|
||||
const restarted = createProviderSnapshotCache(storage, { maxBytes: 2_000 });
|
||||
await writeSnapshot(restarted, {
|
||||
cwd: "/next",
|
||||
label: "next",
|
||||
generatedAt: "2026-08-03T00:00:00.000Z",
|
||||
});
|
||||
await expect(restarted.read("server-1", "/stable")).resolves.not.toBeNull();
|
||||
await expect(restarted.read("server-1", "/next")).resolves.not.toBeNull();
|
||||
expect(snapshotBytes(storage.values)).toBeLessThanOrEqual(2_000);
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["before", "after"] as const)(
|
||||
"reconciles snapshot records when the derived index write fails %s its side effect",
|
||||
async (timing) => {
|
||||
const storage = createStorage();
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes: 2_000 });
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/stable",
|
||||
label: "stable",
|
||||
generatedAt: "2026-08-01T00:00:00.000Z",
|
||||
});
|
||||
storage.failNext("setItem", timing, SNAPSHOT_INDEX_KEY);
|
||||
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/orphan-after-index-failure",
|
||||
label: "orphan",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
|
||||
const restarted = createProviderSnapshotCache(storage, { maxBytes: 2_000 });
|
||||
await expect(
|
||||
restarted.read("server-1", "/orphan-after-index-failure"),
|
||||
).resolves.not.toBeNull();
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
expect(storage.stats.getAllKeysCalls).toBe(2);
|
||||
},
|
||||
);
|
||||
|
||||
it("repairs partial eviction, stale index metadata, and preserves unrelated storage", async () => {
|
||||
const storage = createStorage();
|
||||
const maxBytes = 1_000;
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes });
|
||||
storage.values.set("@paseo/unrelated", "keep-me");
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/oldest",
|
||||
label: "oldest",
|
||||
generatedAt: "2026-08-01T00:00:00.000Z",
|
||||
});
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/middle",
|
||||
label: "middle",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
storage.failNext("multiRemove", "after");
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/large-newest",
|
||||
label: "newest",
|
||||
generatedAt: "2026-08-03T00:00:00.000Z",
|
||||
});
|
||||
storage.values.set(
|
||||
SNAPSHOT_INDEX_KEY,
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
entries: [{ key: `${SNAPSHOT_KEY_PREFIX}phantom`, bytes: 999_999, writtenAt: "bad" }],
|
||||
}),
|
||||
);
|
||||
|
||||
const restarted = createProviderSnapshotCache(storage, { maxBytes });
|
||||
await writeSnapshot(restarted, {
|
||||
cwd: "/final",
|
||||
label: "final",
|
||||
generatedAt: "2026-08-04T00:00:00.000Z",
|
||||
});
|
||||
|
||||
expect(storage.values.get("@paseo/unrelated")).toBe("keep-me");
|
||||
expect(snapshotBytes(storage.values)).toBeLessThanOrEqual(maxBytes);
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
await expect(restarted.read("server-1", "/final")).resolves.not.toBeNull();
|
||||
});
|
||||
|
||||
it("evicts before writing when storage has no temporary headroom", async () => {
|
||||
const maxBytes = 1_000;
|
||||
const storage = createStorage(maxBytes);
|
||||
const cache = createProviderSnapshotCache(storage, { maxBytes });
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/oldest",
|
||||
label: "oldest",
|
||||
generatedAt: "2026-08-01T00:00:00.000Z",
|
||||
});
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/middle",
|
||||
label: "middle",
|
||||
generatedAt: "2026-08-02T00:00:00.000Z",
|
||||
});
|
||||
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/newest",
|
||||
label: "newest",
|
||||
generatedAt: "2026-08-03T00:00:00.000Z",
|
||||
});
|
||||
|
||||
await expect(cache.read("server-1", "/newest")).resolves.not.toBeNull();
|
||||
expect(snapshotBytes(storage.values)).toBeLessThanOrEqual(maxBytes);
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
});
|
||||
|
||||
it.each(["before", "after"] as const)(
|
||||
"recovers when invalid-record cleanup fails %s deletion",
|
||||
async (timing) => {
|
||||
const storage = createStorage();
|
||||
const key = `${SNAPSHOT_KEY_PREFIX}["server-1","/invalid"]`;
|
||||
storage.values.set(key, "invalid");
|
||||
storage.failNext("multiRemove", timing);
|
||||
const cache = createProviderSnapshotCache(storage);
|
||||
|
||||
await expect(cache.read("server-1", "/invalid")).resolves.toBeNull();
|
||||
await writeSnapshot(cache, {
|
||||
cwd: "/recovered",
|
||||
label: "recovered",
|
||||
generatedAt: "2026-08-03T00:00:00.000Z",
|
||||
});
|
||||
|
||||
expect(storage.values.has(key)).toBe(false);
|
||||
await expect(cache.read("server-1", "/recovered")).resolves.not.toBeNull();
|
||||
expectIndexMatchesSnapshots(storage.values);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import AsyncStorage from "@react-native-async-storage/async-storage";
|
||||
import { Buffer } from "buffer";
|
||||
import type { ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
|
||||
import {
|
||||
expandProviderSnapshot,
|
||||
@@ -8,11 +9,21 @@ import { CompactProviderSnapshotSchema } from "@getpaseo/protocol/messages";
|
||||
|
||||
const CACHE_VERSION = 1;
|
||||
const CACHE_KEY_PREFIX = "@paseo/provider-snapshot/v1";
|
||||
const CACHE_INDEX_KEY = "@paseo/provider-snapshot-index/v1";
|
||||
const CACHE_INDEX_VERSION = 1;
|
||||
const DEFAULT_MAX_CACHE_BYTES = 4 * 1024 * 1024;
|
||||
|
||||
interface ProviderSnapshotStorage {
|
||||
getItem(key: string): Promise<string | null>;
|
||||
setItem(key: string, value: string): Promise<void>;
|
||||
removeItem(key: string): Promise<void>;
|
||||
getAllKeys(): Promise<readonly string[]>;
|
||||
multiGet(keys: readonly string[]): Promise<ReadonlyArray<readonly [string, string | null]>>;
|
||||
multiRemove(keys: readonly string[]): Promise<void>;
|
||||
}
|
||||
|
||||
interface ProviderSnapshotCacheOptions {
|
||||
maxBytes?: number;
|
||||
}
|
||||
|
||||
interface StoredProviderSnapshot {
|
||||
@@ -22,6 +33,17 @@ interface StoredProviderSnapshot {
|
||||
compactSnapshot: CompactProviderSnapshot;
|
||||
}
|
||||
|
||||
interface ProviderSnapshotIndexEntry {
|
||||
key: string;
|
||||
bytes: number;
|
||||
writtenAt: string;
|
||||
}
|
||||
|
||||
interface ProviderSnapshotIndex {
|
||||
version: typeof CACHE_INDEX_VERSION;
|
||||
entries: ProviderSnapshotIndexEntry[];
|
||||
}
|
||||
|
||||
export interface CachedProviderSnapshot extends StoredProviderSnapshot {
|
||||
entries: ProviderSnapshotEntry[];
|
||||
}
|
||||
@@ -48,6 +70,15 @@ function cacheKey(serverId: string, cwd: string | null): string {
|
||||
return `${CACHE_KEY_PREFIX}:${JSON.stringify([serverId, cwd])}`;
|
||||
}
|
||||
|
||||
function storedBytes(key: string, value: string): number {
|
||||
return Buffer.byteLength(key, "utf8") + Buffer.byteLength(value, "utf8");
|
||||
}
|
||||
|
||||
function oldestFirst(left: ProviderSnapshotIndexEntry, right: ProviderSnapshotIndexEntry): number {
|
||||
const writtenAtOrder = left.writtenAt.localeCompare(right.writtenAt);
|
||||
return writtenAtOrder !== 0 ? writtenAtOrder : left.key.localeCompare(right.key);
|
||||
}
|
||||
|
||||
function parseStoredProviderSnapshot(value: string): StoredProviderSnapshot | null {
|
||||
const parsed: unknown = JSON.parse(value);
|
||||
if (typeof parsed !== "object" || parsed === null) return null;
|
||||
@@ -65,60 +96,142 @@ function parseStoredProviderSnapshot(value: string): StoredProviderSnapshot | nu
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
version,
|
||||
hash,
|
||||
generatedAt,
|
||||
compactSnapshot: compactSnapshot.data,
|
||||
};
|
||||
}
|
||||
|
||||
async function discardCachedSnapshot(storage: ProviderSnapshotStorage, key: string): Promise<void> {
|
||||
try {
|
||||
await storage.removeItem(key);
|
||||
} catch (error) {
|
||||
if (!(error instanceof Error)) throw error;
|
||||
}
|
||||
return { version, hash, generatedAt, compactSnapshot: compactSnapshot.data };
|
||||
}
|
||||
|
||||
export function createProviderSnapshotCache(
|
||||
storage: ProviderSnapshotStorage,
|
||||
options: ProviderSnapshotCacheOptions = {},
|
||||
): ProviderSnapshotCache {
|
||||
return {
|
||||
async read(serverId, cwd) {
|
||||
const key = cacheKey(serverId, cwd);
|
||||
let value: string | null;
|
||||
try {
|
||||
value = await storage.getItem(key);
|
||||
} catch (error) {
|
||||
if (error instanceof Error) return null;
|
||||
throw error;
|
||||
const maxBytes = options.maxBytes ?? DEFAULT_MAX_CACHE_BYTES;
|
||||
let index: ProviderSnapshotIndex | null = null;
|
||||
let operationQueue = Promise.resolve();
|
||||
|
||||
async function persistIndex(entries: ProviderSnapshotIndexEntry[]): Promise<void> {
|
||||
const nextIndex: ProviderSnapshotIndex = { version: CACHE_INDEX_VERSION, entries };
|
||||
await storage.setItem(CACHE_INDEX_KEY, JSON.stringify(nextIndex));
|
||||
index = nextIndex;
|
||||
}
|
||||
if (value === null) return null;
|
||||
|
||||
async function reconcileCache(): Promise<void> {
|
||||
const keys = (await storage.getAllKeys()).filter((key) =>
|
||||
key.startsWith(`${CACHE_KEY_PREFIX}:`),
|
||||
);
|
||||
const values = await storage.multiGet(keys);
|
||||
const keysToRemove: string[] = [];
|
||||
const entries: ProviderSnapshotIndexEntry[] = [];
|
||||
for (const [key, value] of values) {
|
||||
if (value === null) continue;
|
||||
try {
|
||||
const stored = parseStoredProviderSnapshot(value);
|
||||
if (!stored) {
|
||||
await discardCachedSnapshot(storage, key);
|
||||
return null;
|
||||
keysToRemove.push(key);
|
||||
continue;
|
||||
}
|
||||
return { ...stored, entries: expandProviderSnapshot(stored.compactSnapshot) };
|
||||
entries.push({ key, bytes: storedBytes(key, value), writtenAt: stored.generatedAt });
|
||||
} catch (error) {
|
||||
if (!(error instanceof SyntaxError) && !(error instanceof RangeError)) {
|
||||
throw error;
|
||||
if (!(error instanceof SyntaxError) && !(error instanceof RangeError)) throw error;
|
||||
keysToRemove.push(key);
|
||||
}
|
||||
await discardCachedSnapshot(storage, key);
|
||||
return null;
|
||||
}
|
||||
|
||||
let totalBytes = entries.reduce((total, entry) => total + entry.bytes, 0);
|
||||
const retainedEntries = entries.toSorted(oldestFirst);
|
||||
while (totalBytes > maxBytes) {
|
||||
const evicted = retainedEntries.shift();
|
||||
if (!evicted) break;
|
||||
totalBytes -= evicted.bytes;
|
||||
keysToRemove.push(evicted.key);
|
||||
}
|
||||
if (keysToRemove.length > 0) await storage.multiRemove(keysToRemove);
|
||||
await persistIndex(retainedEntries);
|
||||
}
|
||||
|
||||
async function ensureCacheIndex(): Promise<ProviderSnapshotIndex> {
|
||||
if (index === null) await reconcileCache();
|
||||
if (index === null) throw new Error("Provider snapshot cache reconciliation failed");
|
||||
return index;
|
||||
}
|
||||
|
||||
async function runSerialized<T>(operation: () => Promise<T>): Promise<T> {
|
||||
const result = operationQueue.then(operation);
|
||||
operationQueue = result.then(
|
||||
() => undefined,
|
||||
() => {
|
||||
index = null;
|
||||
},
|
||||
async write(input) {
|
||||
const stored: StoredProviderSnapshot = {
|
||||
);
|
||||
return result;
|
||||
}
|
||||
|
||||
async function writeSnapshot(input: {
|
||||
serverId: string;
|
||||
cwd: string | null;
|
||||
hash: string;
|
||||
generatedAt: string;
|
||||
compactSnapshot: CompactProviderSnapshot;
|
||||
}): Promise<void> {
|
||||
const currentIndex = await ensureCacheIndex();
|
||||
const key = cacheKey(input.serverId, input.cwd);
|
||||
const value = JSON.stringify({
|
||||
version: CACHE_VERSION,
|
||||
hash: input.hash,
|
||||
generatedAt: input.generatedAt,
|
||||
compactSnapshot: input.compactSnapshot,
|
||||
};
|
||||
} satisfies StoredProviderSnapshot);
|
||||
const incomingBytes = storedBytes(key, value);
|
||||
const canStoreIncoming = incomingBytes <= maxBytes;
|
||||
const retainedEntries = currentIndex.entries.filter((entry) => entry.key !== key);
|
||||
let projectedBytes = retainedEntries.reduce((total, entry) => total + entry.bytes, 0);
|
||||
if (canStoreIncoming) projectedBytes += incomingBytes;
|
||||
|
||||
const keysToRemove: string[] = [];
|
||||
for (const candidate of retainedEntries.toSorted(oldestFirst)) {
|
||||
if (projectedBytes <= maxBytes) break;
|
||||
projectedBytes -= candidate.bytes;
|
||||
keysToRemove.push(candidate.key);
|
||||
}
|
||||
const removedKeys = new Set(keysToRemove);
|
||||
const nextEntries = retainedEntries.filter((entry) => !removedKeys.has(entry.key));
|
||||
|
||||
if (!canStoreIncoming && currentIndex.entries.some((entry) => entry.key === key)) {
|
||||
keysToRemove.push(key);
|
||||
}
|
||||
if (keysToRemove.length > 0) await storage.multiRemove(keysToRemove);
|
||||
if (canStoreIncoming) {
|
||||
await storage.setItem(key, value);
|
||||
nextEntries.push({ key, bytes: incomingBytes, writtenAt: input.generatedAt });
|
||||
}
|
||||
await persistIndex(nextEntries);
|
||||
}
|
||||
|
||||
return {
|
||||
async read(serverId, cwd) {
|
||||
try {
|
||||
await storage.setItem(cacheKey(input.serverId, input.cwd), JSON.stringify(stored));
|
||||
return await runSerialized(async () => {
|
||||
const currentIndex = await ensureCacheIndex();
|
||||
const key = cacheKey(serverId, cwd);
|
||||
const value = await storage.getItem(key);
|
||||
if (value === null) return null;
|
||||
try {
|
||||
const stored = parseStoredProviderSnapshot(value);
|
||||
if (!stored) throw new SyntaxError("Invalid provider snapshot");
|
||||
return { ...stored, entries: expandProviderSnapshot(stored.compactSnapshot) };
|
||||
} catch (error) {
|
||||
if (!(error instanceof SyntaxError) && !(error instanceof RangeError)) throw error;
|
||||
await storage.removeItem(key);
|
||||
await persistIndex(currentIndex.entries.filter((entry) => entry.key !== key));
|
||||
return null;
|
||||
}
|
||||
});
|
||||
} catch (error) {
|
||||
if (error instanceof Error) return null;
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
async write(input) {
|
||||
try {
|
||||
await runSerialized(() => writeSnapshot(input));
|
||||
} catch (error) {
|
||||
if (!(error instanceof Error)) throw error;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user