improve(device-pair): reduce notification status read cost (#146889)

Co-authored-by: Peter Steinberger <58493+steipete@users.noreply.github.com>
This commit is contained in:
Peter Steinberger 2026-09-13 02:08:05 -07:00 • committed by GitHub
parent b410624719
commit 89edff36f4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 103 additions and 3 deletions

View file

@ -10,6 +10,7 @@ import type {
} from "openclaw/plugin-sdk/plugin-state-runtime";
import {
createPluginStateKeyedStoreForTests,
openOpenClawStateDatabase,
resetPluginStateStoreForTests,
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
import { createTestPluginApi } from "openclaw/plugin-sdk/plugin-test-api";
@ -87,6 +88,103 @@ describe("device-pair notify persistence", () => {
});
}
it.each([true, false])(
"reports live subscriber counts with count support=%s",
async (supportsCount) => {
vi.useFakeTimers();
vi.setSystemTime(1_000);
const subscriber: NotifySubscription = {
to: "chat-123",
mode: "persistent",
addedAtMs: 1,
};
const subscriberStore = openSubscriberStore();
await subscriberStore.register(notifySubscriberStoreKey(subscriber), subscriber);
await subscriberStore.register("expired", subscriber, { ttlMs: 1 });
await openStore({ namespace: "other", maxEntries: 10 }).register("sibling", subscriber);
vi.setSystemTime(1_001);
const api = createApi(undefined, <T>(options: OpenKeyedStoreOptions) => {
const store = openStore<T>(options);
if (supportsCount) {
return store;
}
const { count: _count, ...olderStore } = store;
return olderStore;
});
for (const [senderId, mode] of [
["chat-123", "persistent"],
["missing", "off"],
]) {
const status = await handleNotifyCommand({
api,
ctx: { channel: "telegram", senderId },
action: "status",
});
expect(status.text).toContain(`Mode: ${mode}`);
expect(status.text).toContain("Subscribers: 1");
expect(status.text).toContain("Pending requests: 0");
}
},
);
it("counts unrelated corrupt subscriber rows but still validates the selected chat", async () => {
const subscriber: NotifySubscription = { to: "chat-123", mode: "once", addedAtMs: 1 };
const subscriberStore = openSubscriberStore();
await subscriberStore.register(notifySubscriberStoreKey(subscriber), subscriber);
await subscriberStore.register("unrelated", subscriber);
const { db } = openOpenClawStateDatabase({ env });
const corrupt = db.prepare(
"UPDATE plugin_state_entries SET value_json = ? WHERE plugin_id = ? AND namespace = ? AND entry_key = ?",
);
corrupt.run("{", "device-pair", DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, "unrelated");
const command = {
api: createApi(),
ctx: { channel: "telegram", senderId: "chat-123" },
action: "status",
};
const status = await handleNotifyCommand(command);
expect(status.text).toContain("Mode: once");
expect(status.text).toContain("Subscribers: 2");
const olderApi = createApi(undefined, <T>(options: OpenKeyedStoreOptions) => {
const { count: _count, ...store } = openStore<T>(options);
return store;
});
await expect(handleNotifyCommand({ ...command, api: olderApi })).rejects.toMatchObject({
code: "PLUGIN_STATE_CORRUPT",
operation: "entries",
});
corrupt.run(
"{",
"device-pair",
DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
notifySubscriberStoreKey(subscriber),
);
await expect(handleNotifyCommand(command)).rejects.toMatchObject({
code: "PLUGIN_STATE_CORRUPT",
operation: "lookup",
});
});
it("propagates count failures without retrying enumeration", async () => {
const failure = new Error("count unavailable");
const entries = vi.fn(async () => []);
const api = createApi(undefined, <T>(options: OpenKeyedStoreOptions) => ({
...openStore<T>(options),
count: async () => {
throw failure;
},
entries,
}));
await expect(
handleNotifyCommand({
api,
ctx: { channel: "telegram", senderId: "chat-123" },
action: "status",
}),
).rejects.toBe(failure);
expect(entries).not.toHaveBeenCalled();
});
it("defers the first notify poll and keeps one in flight across service recreation", async () => {
vi.useFakeTimers();
const firstPoll = createDeferred<Awaited<ReturnType<typeof listDevicePairingMock>>>();

View file

@ -427,9 +427,11 @@ export async function handleNotifyCommand(params: {
}
if (params.action === "status" || params.action === "") {
const [current, subscribers, pending] = await Promise.all([
const [current, subscriberCount, pending] = await Promise.all([
subscriberStore.lookup(targetStoreKey),
subscriberStore.entries(),
subscriberStore.count
? subscriberStore.count()
: subscriberStore.entries().then((entries) => entries.length),
listDevicePairing(),
]);
const enabled = Boolean(current);
@ -438,7 +440,7 @@ export async function handleNotifyCommand(params: {
text: [
`Pair request notifications: ${enabled ? "enabled" : "disabled"} for this chat.`,
`Mode: ${mode}`,
`Subscribers: ${subscribers.length}`,
`Subscribers: ${subscriberCount}`,
`Pending requests: ${pending.pending.length}`,
"",
"Use /pair notify on|off|once",