File size: 6,424 Bytes
76289e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | import { ok } from "@openclaw/normalization-core/result";
import { afterEach, describe, expect, it } from "vitest";
import { trackSqliteStatementExecutions } from "../../test/helpers/sqlite-statement-execution-counter.js";
import {
isOpenClawStateDatabaseOpen,
openOpenClawStateDatabase,
} from "../state/openclaw-state-db.js";
import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js";
import {
createPluginStateKeyedStore,
createPluginStateSyncKeyedStore,
resetPluginStateStoreForTests,
} from "./plugin-state-store.js";
import { closePluginStateDatabase } from "./plugin-state-store.sqlite.js";
import { seedPluginStateEntriesForTests } from "./plugin-state-store.test-helpers.js";
import { PluginStateStoreError } from "./plugin-state-store.types.js";
afterEach(() => resetPluginStateStoreForTests());
describe("plugin state bulk reads", () => {
it("bulk reads exact keys positionally with sync/async parity across reopen", async () => {
await withOpenClawTestState({ label: "plugin-state-bulk-0" }, async () => {
const options = { namespace: "bulk", maxEntries: 20 };
const sync = createPluginStateSyncKeyedStore<{ index: number }>("discord", options);
const asyncStore = createPluginStateKeyedStore<{ index: number }>("discord", options);
const keys = [
"ten:10",
"two:2",
"nul\0tail",
"literal\\u0000",
"lone\ud800",
"__proto__",
] as const;
keys.forEach((key, index) => sync.register(key, { index }));
createPluginStateSyncKeyedStore("telegram", options).register(keys[0], { index: 99 });
createPluginStateSyncKeyedStore("discord", { ...options, namespace: "other" }).register(
keys[0],
{ index: 98 },
);
const now = Date.now();
seedPluginStateEntriesForTests([
{ pluginId: "discord", namespace: "bulk", key: "expired", value: {}, expiresAt: now },
]);
const request = [keys[3], "missing", keys[0], "expired", ...keys, ` ${keys[1]} `];
const expected = [
{ index: 3 },
undefined,
{ index: 0 },
undefined,
...keys.map((_, index) => ({ index })),
{ index: 1 },
];
for (let connection = 0; connection < 2; connection++) {
expect(sync.lookupMany(request)).toEqual(expected.map(ok));
await expect(asyncStore.lookupMany(request)).resolves.toEqual(expected.map(ok));
for (const duplicates of [
sync.lookupMany([keys[0], keys[0]]),
await asyncStore.lookupMany([keys[0], keys[0]]),
]) {
expect(duplicates[0]?.ok && duplicates[0].value).not.toBe(
duplicates[1]?.ok && duplicates[1].value,
);
}
if (connection > 0) {
expect(isOpenClawStateDatabaseOpen()).toBe(false);
}
closePluginStateDatabase();
}
expect(isOpenClawStateDatabaseOpen()).toBe(false);
});
});
it("bulk reads use fresh expiry and restore positional corrupt JSON errors", async () => {
await withOpenClawTestState({ label: "plugin-state-bulk-1" }, async () => {
const options = { namespace: "bulk-errors", maxEntries: 10 };
const sync = createPluginStateSyncKeyedStore<number>("discord", options);
const asyncStore = createPluginStateKeyedStore<number>("discord", options);
sync.register("short", 1, { ttlMs: 24 * 60 * 60_000 });
sync.register("long", 2);
sync.register("healthy", 3);
await expect(asyncStore.lookupMany(["short", "long"])).resolves.toEqual([ok(1), ok(2)]);
seedPluginStateEntriesForTests([
{
pluginId: "discord",
namespace: options.namespace,
key: "short",
value: 1,
expiresAt: Date.now() - 1,
},
]);
await expect(asyncStore.lookupMany(["short", "long"])).resolves.toEqual([
ok(undefined),
ok(2),
]);
const { db, path } = openOpenClawStateDatabase();
db.prepare(
"UPDATE plugin_state_entries SET value_json = ? WHERE namespace = ? AND entry_key = ?",
).run("invalid JSON", "bulk-errors", "long");
const corrupt = {
ok: false,
error: expect.objectContaining({ code: "PLUGIN_STATE_CORRUPT", operation: "lookup", path }),
};
const request = ["healthy", "long", "short", "long", "missing"];
const expected = [ok(3), corrupt, ok(undefined), corrupt, ok(undefined)];
expect(sync.lookupMany(request)).toEqual(expected);
const results = await asyncStore.lookupMany(request);
expect(results).toEqual(expected);
for (const result of results) {
if (!result.ok) {
expect(result.error).toBeInstanceOf(PluginStateStoreError);
expect(result.error.cause).toBeInstanceOf(SyntaxError);
}
}
expect(() => sync.lookup("long")).toThrowError(corrupt.error);
});
});
it("bounds and validates every bulk key before reading, with one native query", async () => {
await withOpenClawTestState({ label: "plugin-state-bulk-2" }, async () => {
const store = createPluginStateSyncKeyedStore<number>("discord", {
namespace: "bulk-bounds",
maxEntries: 10,
});
const asyncStore = createPluginStateKeyedStore<number>("discord", {
namespace: "bulk-bounds",
maxEntries: 10,
});
store.register("key", 1);
const { db } = openOpenClawStateDatabase();
const reads = trackSqliteStatementExecutions(db, ["reads"], (sql) =>
sql.startsWith("select ") && sql.includes('"plugin_state_entries"') ? "reads" : null,
);
try {
expect(store.lookupMany([])).toEqual([]);
expect(() => store.lookupMany(["key", " "])).toThrowError(
expect.objectContaining({ code: "PLUGIN_STATE_INVALID_INPUT", operation: "lookup" }),
);
await expect(
asyncStore.lookupMany(Array.from({ length: 10_001 }, () => "key")),
).rejects.toMatchObject({ code: "PLUGIN_STATE_INVALID_INPUT", operation: "lookup" });
expect(reads.counts.reads).toBe(0);
expect(store.lookupMany(Array.from({ length: 10_000 }, () => "key"))).toEqual(
Array.from({ length: 10_000 }, () => ok(1)),
);
expect(reads.counts.reads).toBe(1);
expect(reads.rowCounts.reads).toBe(1);
} finally {
reads.restore();
}
});
});
});
|