mirror of
https://github.com/block/buzz.git
synced 2026-11-11 20:01:09 +00:00
fix(desktop): diagnose and explicitly recover lifecycle receivers
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
@@ -308,7 +308,12 @@ with a TypeScript lookup table or an id comparison in a component.
|
||||
|
||||
Known Desktops exposes an owner-private, explicitly selected agent+Desktop Stop,
|
||||
not inferred agent location. The app-scoped receiver subscribes live only;
|
||||
reopening never replays commands. An explicit retry republishes the exact request;
|
||||
reopening never replays commands. Receiver initialization reports a safe failure
|
||||
stage without exposing raw transport/IPC exceptions. Its scope-owned notification
|
||||
can explicitly retry the **receiver** with a fresh live-only subscription; that
|
||||
must discard queued callbacks from the retired receiver, not retry an operation.
|
||||
A readiness timeout is unconfirmed delivery, not a failed initialization; late
|
||||
EOSE clears that warning after successful projection. An explicit operation retry republishes the exact request;
|
||||
the relay redelivers stored Stop duplicates without repeating relay side effects.
|
||||
The receiver returns saved results or Unknown, never repeats a consumed Stop.
|
||||
Native owner-delegation and community checks
|
||||
|
||||
@@ -238,3 +238,139 @@ test("receiver projects history without executing it and invalidates live work o
|
||||
assert.equal(f.errors.length, 1);
|
||||
close();
|
||||
});
|
||||
|
||||
for (const stage of [
|
||||
"subscription",
|
||||
"history",
|
||||
"projection",
|
||||
"reconciliation",
|
||||
]) {
|
||||
test(`receiver reports safe ${stage} failure and never admits queued work`, async () => {
|
||||
const f = fixture();
|
||||
const secret = "private key /home/private bearer secret";
|
||||
if (stage === "subscription")
|
||||
f.relay.subscribeLive = async () => {
|
||||
throw Error(secret);
|
||||
};
|
||||
if (stage === "history")
|
||||
f.relay.fetchEvents = async () => {
|
||||
throw Error(secret);
|
||||
};
|
||||
const ipc = async (command, args) => {
|
||||
if (
|
||||
command === "observe_desktop_placement" &&
|
||||
((stage === "projection" && !args.reconcile) ||
|
||||
(stage === "reconciliation" && args.reconcile))
|
||||
)
|
||||
throw Error(secret);
|
||||
return f.ipc(command, args);
|
||||
};
|
||||
const started = receiveLifecycle(
|
||||
scope,
|
||||
() => true,
|
||||
(e) => f.errors.push(e),
|
||||
ipc,
|
||||
f.relay,
|
||||
);
|
||||
f.deliver({ id: "queued", kind: 50182 });
|
||||
await assert.rejects(started, (error) => {
|
||||
assert.match(error.message, new RegExp(`${stage}: request failed`));
|
||||
assert.doesNotMatch(error.message, /private|bearer|secret/);
|
||||
return true;
|
||||
});
|
||||
await tick();
|
||||
assert.equal(
|
||||
f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length,
|
||||
0,
|
||||
);
|
||||
assert.deepEqual(
|
||||
f.errors,
|
||||
[],
|
||||
"discarded callbacks cannot replace the startup diagnosis",
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
test("explicit receiver recovery discards old queued work and projects history without replay", async () => {
|
||||
const f = fixture();
|
||||
let rejectHistory;
|
||||
f.relay.fetchEvents = () =>
|
||||
new Promise((_, reject) => {
|
||||
rejectHistory = reject;
|
||||
});
|
||||
const failed = receiveLifecycle(
|
||||
scope,
|
||||
() => true,
|
||||
() => {},
|
||||
f.ipc,
|
||||
f.relay,
|
||||
);
|
||||
await tick();
|
||||
f.deliver({ id: "old-live", kind: 50182 });
|
||||
rejectHistory(Error("Timed out while loading channel history."));
|
||||
await assert.rejects(failed, /history: history timed out/);
|
||||
f.relay.fetchEvents = async () => [{ id: "old-live", kind: 50182 }];
|
||||
const close = await receiveLifecycle(
|
||||
scope,
|
||||
() => true,
|
||||
() => {},
|
||||
f.ipc,
|
||||
f.relay,
|
||||
);
|
||||
await tick();
|
||||
assert.equal(
|
||||
f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length,
|
||||
0,
|
||||
);
|
||||
f.deliver({ id: "new-live", kind: 50182 });
|
||||
await tick();
|
||||
assert.equal(
|
||||
f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length,
|
||||
1,
|
||||
);
|
||||
f.deliver({ id: "closed-queue", kind: 50182 });
|
||||
close();
|
||||
await tick();
|
||||
assert.equal(
|
||||
f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length,
|
||||
1,
|
||||
);
|
||||
});
|
||||
|
||||
test("readiness timeout is distinct, late EOSE recovers, CLOSED retires old callbacks", async () => {
|
||||
const f = fixture();
|
||||
let notify, deliver;
|
||||
let closed = 0,
|
||||
ready = 0;
|
||||
f.relay.subscribeLive = async (_filter, event, onReady, timeout) => {
|
||||
assert.equal(timeout, 5000);
|
||||
deliver = event;
|
||||
notify = onReady;
|
||||
notify("timeout");
|
||||
return () => {
|
||||
closed++;
|
||||
};
|
||||
};
|
||||
const close = await receiveLifecycle(
|
||||
scope,
|
||||
() => true,
|
||||
(e) => f.errors.push(e),
|
||||
f.ipc,
|
||||
f.relay,
|
||||
() => ready++,
|
||||
);
|
||||
assert.match(f.errors[0], /readiness timed out/);
|
||||
assert.equal(ready, 0);
|
||||
notify("eose");
|
||||
assert.equal(ready, 1);
|
||||
notify("closed");
|
||||
deliver({ id: "late", kind: 50182 });
|
||||
await tick();
|
||||
assert.equal(closed, 1);
|
||||
assert.equal(
|
||||
f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length,
|
||||
0,
|
||||
);
|
||||
assert.match(f.errors.at(-1), /subscription closed/);
|
||||
close();
|
||||
});
|
||||
|
||||
@@ -9,6 +9,11 @@ import {
|
||||
sendStop,
|
||||
} from "./desktopStop";
|
||||
|
||||
import {
|
||||
LifecycleReceiverError,
|
||||
receiverStep,
|
||||
} from "./desktopLifecycleDiagnostics";
|
||||
|
||||
export const DESKTOP_LIFECYCLE = 50182;
|
||||
export const DESKTOP_LIFECYCLE_RESULT = 50183;
|
||||
export type LifecycleAction = "start" | "restart" | "status";
|
||||
@@ -99,27 +104,33 @@ export function lifecycleClient(
|
||||
let before_id: string | undefined;
|
||||
for (let page = 0; page < 64; page++) {
|
||||
check();
|
||||
const events = await relay.fetchEvents({
|
||||
kinds: [DESKTOP_STOP, DESKTOP_LIFECYCLE],
|
||||
authors: [scope.owner],
|
||||
limit: 256,
|
||||
until,
|
||||
before_id,
|
||||
});
|
||||
const events = await receiverStep("history", () =>
|
||||
relay.fetchEvents({
|
||||
kinds: [DESKTOP_STOP, DESKTOP_LIFECYCLE],
|
||||
authors: [scope.owner],
|
||||
limit: 256,
|
||||
until,
|
||||
before_id,
|
||||
}),
|
||||
);
|
||||
check();
|
||||
// No effects while a partial page could still hide a dominating Start.
|
||||
await ipc("observe_desktop_placement", {
|
||||
...scope,
|
||||
events,
|
||||
reconcile: false,
|
||||
});
|
||||
await receiverStep("projection", () =>
|
||||
ipc("observe_desktop_placement", {
|
||||
...scope,
|
||||
events,
|
||||
reconcile: false,
|
||||
}),
|
||||
);
|
||||
check();
|
||||
if (events.length < 256) {
|
||||
await ipc("observe_desktop_placement", {
|
||||
...scope,
|
||||
events: [],
|
||||
reconcile: true,
|
||||
});
|
||||
await receiverStep("reconciliation", () =>
|
||||
ipc("observe_desktop_placement", {
|
||||
...scope,
|
||||
events: [],
|
||||
reconcile: true,
|
||||
}),
|
||||
);
|
||||
check();
|
||||
return;
|
||||
}
|
||||
@@ -250,14 +261,21 @@ export function lifecycleClient(
|
||||
}
|
||||
|
||||
/** Subscribe first, then project history; live commands wait for complete
|
||||
* initialization. Reconnect gets a new client epoch and never replays history. */
|
||||
* initialization. Explicit receiver retry starts a fresh live-only subscription,
|
||||
* never retries an operation or executes historical commands. */
|
||||
export async function receiveLifecycle(
|
||||
scope: DesktopScope,
|
||||
active: () => boolean,
|
||||
onError: (message: string) => void,
|
||||
ipc = invoke,
|
||||
relay = relayClient,
|
||||
onReady: () => void = () => {},
|
||||
) {
|
||||
let stopped = false;
|
||||
let stopSubscription = () => {};
|
||||
let synced = false;
|
||||
let subscriptionReady = false;
|
||||
const valid = () => active() && !stopped;
|
||||
let client: ReturnType<typeof lifecycleClient>;
|
||||
let initialized: () => void = () => {};
|
||||
const ready = new Promise<void>((resolve) => {
|
||||
@@ -265,70 +283,99 @@ export async function receiveLifecycle(
|
||||
});
|
||||
let chain = Promise.resolve();
|
||||
let pending = 0;
|
||||
const close = await relay.subscribeLive(
|
||||
{
|
||||
kinds: [DESKTOP_LIFECYCLE, DESKTOP_STOP],
|
||||
authors: [scope.owner],
|
||||
limit: 0,
|
||||
},
|
||||
(event) => {
|
||||
if (!active()) return;
|
||||
if (pending >= 16) {
|
||||
onError("Desktop lifecycle receiver is busy; outcome is unconfirmed.");
|
||||
return;
|
||||
}
|
||||
pending++;
|
||||
chain = chain
|
||||
.then(async () => {
|
||||
await ready;
|
||||
client.check();
|
||||
await ipc("observe_desktop_placement", {
|
||||
...scope,
|
||||
events: [event],
|
||||
reconcile: true,
|
||||
});
|
||||
client.check();
|
||||
const result = await ipc<RelayEvent | null>(
|
||||
event.kind === DESKTOP_STOP
|
||||
? "receive_desktop_stop"
|
||||
: "receive_desktop_lifecycle",
|
||||
{ ...scope, event },
|
||||
const unsubscribe = await receiverStep("subscription", () =>
|
||||
relay.subscribeLive(
|
||||
{
|
||||
kinds: [DESKTOP_LIFECYCLE, DESKTOP_STOP],
|
||||
authors: [scope.owner],
|
||||
limit: 0,
|
||||
},
|
||||
(event) => {
|
||||
if (!valid()) return;
|
||||
if (pending >= 16) {
|
||||
onError(
|
||||
"Desktop lifecycle receiver is busy; outcome is unconfirmed.",
|
||||
);
|
||||
client.check();
|
||||
if (result)
|
||||
await relay.publishEvent(
|
||||
result,
|
||||
"Result delivery unconfirmed",
|
||||
"Result delivery failed",
|
||||
client.check,
|
||||
return;
|
||||
}
|
||||
pending++;
|
||||
chain = chain
|
||||
.then(async () => {
|
||||
await ready;
|
||||
if (!valid()) return;
|
||||
client.check();
|
||||
await ipc("observe_desktop_placement", {
|
||||
...scope,
|
||||
events: [event],
|
||||
reconcile: true,
|
||||
});
|
||||
client.check();
|
||||
const result = await ipc<RelayEvent | null>(
|
||||
event.kind === DESKTOP_STOP
|
||||
? "receive_desktop_stop"
|
||||
: "receive_desktop_lifecycle",
|
||||
{ ...scope, event },
|
||||
);
|
||||
})
|
||||
.catch(() => {
|
||||
if (active())
|
||||
onError(
|
||||
"Desktop lifecycle result is unconfirmed. No automatic operation retry.",
|
||||
);
|
||||
})
|
||||
.finally(() => {
|
||||
pending--;
|
||||
});
|
||||
},
|
||||
(readiness) => {
|
||||
if (active() && readiness !== "eose")
|
||||
onError("Desktop lifecycle receiver is unavailable.");
|
||||
},
|
||||
client.check();
|
||||
if (result)
|
||||
await relay.publishEvent(
|
||||
result,
|
||||
"Result delivery unconfirmed",
|
||||
"Result delivery failed",
|
||||
client.check,
|
||||
);
|
||||
})
|
||||
.catch(() => {
|
||||
if (valid())
|
||||
onError(
|
||||
"Desktop lifecycle result is unconfirmed. No automatic operation retry.",
|
||||
);
|
||||
})
|
||||
.finally(() => {
|
||||
pending--;
|
||||
});
|
||||
},
|
||||
(readiness) => {
|
||||
if (!valid()) return;
|
||||
subscriptionReady = readiness === "eose";
|
||||
if (readiness === "closed") {
|
||||
stopped = true;
|
||||
stopSubscription();
|
||||
onError(
|
||||
"Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests.",
|
||||
);
|
||||
} else if (readiness === "timeout") {
|
||||
onError(
|
||||
"Desktop lifecycle subscription readiness timed out. Delivery is unconfirmed.",
|
||||
);
|
||||
} else if (synced) onReady();
|
||||
},
|
||||
5000,
|
||||
),
|
||||
);
|
||||
client = lifecycleClient(scope, active, ipc, relay);
|
||||
const close = () => {
|
||||
stopped = true;
|
||||
// Unsubscribe may fail on a dead socket; it must not revive this receiver
|
||||
// or leave an unhandled promise. A retry always owns a new subscription.
|
||||
void Promise.resolve()
|
||||
.then(unsubscribe)
|
||||
.catch(() => {});
|
||||
};
|
||||
stopSubscription = close;
|
||||
client = lifecycleClient(scope, valid, ipc, relay);
|
||||
try {
|
||||
if (stopped) throw new LifecycleReceiverError("subscription", "closed");
|
||||
await client.sync();
|
||||
client.check();
|
||||
synced = true;
|
||||
initialized();
|
||||
if (subscriptionReady) onReady();
|
||||
} catch (error) {
|
||||
close();
|
||||
// Release queued callbacks into a permanently invalidated client.
|
||||
client = lifecycleClient(scope, () => false, ipc, relay);
|
||||
initialized();
|
||||
throw error;
|
||||
throw error instanceof LifecycleReceiverError
|
||||
? error
|
||||
: new LifecycleReceiverError("initialization", error);
|
||||
}
|
||||
return close;
|
||||
}
|
||||
|
||||
41
desktop/src/features/agents/desktopLifecycleDiagnostics.ts
Normal file
41
desktop/src/features/agents/desktopLifecycleDiagnostics.ts
Normal file
@@ -0,0 +1,41 @@
|
||||
/** Safe receiver diagnostics: raw IPC/transport errors can contain private data. */
|
||||
type Stage =
|
||||
| "initialization"
|
||||
| "subscription"
|
||||
| "history"
|
||||
| "projection"
|
||||
| "reconciliation";
|
||||
|
||||
export class LifecycleReceiverError extends Error {
|
||||
constructor(stage: Stage, error: unknown) {
|
||||
const value = error instanceof Error ? error.message : error;
|
||||
const reason =
|
||||
value === "closed"
|
||||
? "subscription closed"
|
||||
: value === "Desktop lifecycle scope changed"
|
||||
? "scope changed"
|
||||
: value === "Relay session is terminal; cannot reconnect."
|
||||
? "relay session requires reconnection"
|
||||
: value === "Timed out while loading channel history."
|
||||
? "history timed out"
|
||||
: "request failed";
|
||||
super(`Desktop lifecycle receiver is unavailable (${stage}: ${reason}).`);
|
||||
}
|
||||
}
|
||||
|
||||
export function receiverErrorMessage(error: unknown): string {
|
||||
return error instanceof LifecycleReceiverError
|
||||
? error.message
|
||||
: "Desktop lifecycle receiver is unavailable (initialization failed).";
|
||||
}
|
||||
|
||||
export async function receiverStep<T>(
|
||||
stage: Stage,
|
||||
action: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
try {
|
||||
return await action();
|
||||
} catch (error) {
|
||||
throw new LifecycleReceiverError(stage, error);
|
||||
}
|
||||
}
|
||||
@@ -157,6 +157,8 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy
|
||||
let closed = 0;
|
||||
let rejectLate;
|
||||
let delayed = false;
|
||||
let storageAvailable = false;
|
||||
let subscribed = 0;
|
||||
relayClient.fetchEvents = async () => {
|
||||
if (delayed)
|
||||
return new Promise((_, reject) => {
|
||||
@@ -165,13 +167,16 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy
|
||||
return [];
|
||||
};
|
||||
relayClient.subscribeLive = async (_filter, _event, onReadiness) => {
|
||||
subscribed++;
|
||||
readiness = onReadiness;
|
||||
onReadiness("eose");
|
||||
return () => {
|
||||
closed++;
|
||||
};
|
||||
};
|
||||
window.__TAURI_INTERNALS__ = {
|
||||
invoke: async () => {
|
||||
if (storageAvailable) return;
|
||||
throw new Error("fixture: storage unavailable");
|
||||
},
|
||||
};
|
||||
@@ -193,7 +198,7 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy
|
||||
assert.equal(warnings().length, 1);
|
||||
assert.equal(
|
||||
warnings()[0].title,
|
||||
"Desktop lifecycle receiver is unavailable.",
|
||||
"Desktop lifecycle receiver is unavailable (projection: request failed).",
|
||||
);
|
||||
assert.equal(warnings()[0].duration, Infinity);
|
||||
assert.equal(warnings()[0].closeButton, true);
|
||||
@@ -203,6 +208,16 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy
|
||||
1,
|
||||
"repeated failures update one notification",
|
||||
);
|
||||
const retryAction = warnings()[0].action;
|
||||
assert.equal(retryAction.label, "Retry receiver");
|
||||
storageAvailable = true;
|
||||
await React.act(async () => retryAction.onClick());
|
||||
assert.equal(subscribed, 2, "explicit recovery starts a new live receiver");
|
||||
assert.equal(
|
||||
warnings().length,
|
||||
0,
|
||||
"successful recovery removes its warning",
|
||||
);
|
||||
await React.act(async () =>
|
||||
root.render(
|
||||
React.createElement(DesktopLifecycleReceiver, { scope: null }),
|
||||
@@ -227,7 +242,11 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy
|
||||
0,
|
||||
"late startup rejection must not recreate the warning",
|
||||
);
|
||||
assert.equal(closed, 2, "both failed subscriptions are released");
|
||||
assert.equal(
|
||||
closed,
|
||||
3,
|
||||
"failed, recovered and retired subscriptions are released",
|
||||
);
|
||||
} finally {
|
||||
await React.act(async () => root.unmount());
|
||||
relayClient.fetchEvents = originals.fetch;
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
type LifecycleOutcome,
|
||||
} from "../desktopLifecycle";
|
||||
import { useRelayAgentsQuery } from "../hooks";
|
||||
import { receiverErrorMessage } from "../desktopLifecycleDiagnostics";
|
||||
|
||||
export function DesktopLifecycleReceiver({
|
||||
scope,
|
||||
@@ -16,6 +17,7 @@ export function DesktopLifecycleReceiver({
|
||||
scope: DesktopScope | null;
|
||||
}) {
|
||||
const { owner, community } = scope ?? {};
|
||||
const [attempt, retry] = useState(0);
|
||||
useEffect(() => {
|
||||
if (!owner || !community) return;
|
||||
let active = true;
|
||||
@@ -30,22 +32,40 @@ export function DesktopLifecycleReceiver({
|
||||
id: notification,
|
||||
duration: Infinity,
|
||||
closeButton: true,
|
||||
action: {
|
||||
label: "Retry receiver",
|
||||
onClick: () => {
|
||||
if (!active) return;
|
||||
active = false;
|
||||
close?.();
|
||||
retry(attempt + 1);
|
||||
},
|
||||
},
|
||||
});
|
||||
};
|
||||
void receiveLifecycle({ owner, community }, () => active, reportError)
|
||||
void receiveLifecycle(
|
||||
{ owner, community },
|
||||
() => active,
|
||||
reportError,
|
||||
undefined,
|
||||
undefined,
|
||||
() => {
|
||||
if (active && notification !== undefined) toast.dismiss(notification);
|
||||
},
|
||||
)
|
||||
.then((fn) => {
|
||||
if (active) close = fn;
|
||||
else fn();
|
||||
})
|
||||
.catch(() => {
|
||||
reportError("Desktop lifecycle receiver is unavailable.");
|
||||
.catch((error) => {
|
||||
reportError(receiverErrorMessage(error));
|
||||
});
|
||||
return () => {
|
||||
active = false;
|
||||
close?.();
|
||||
if (notification !== undefined) toast.dismiss(notification);
|
||||
};
|
||||
}, [owner, community]);
|
||||
}, [owner, community, attempt]);
|
||||
return null;
|
||||
}
|
||||
function message(outcome: LifecycleOutcome) {
|
||||
|
||||
@@ -118,9 +118,7 @@ test.describe("top chrome macOS traffic-light clearance under text zoom", () =>
|
||||
// A failed global receiver must remain visible without entering the shell's
|
||||
// layout flow. This also forces the error to settle before measuring chrome.
|
||||
await expect(
|
||||
page.getByText("Desktop lifecycle receiver is unavailable.", {
|
||||
exact: true,
|
||||
}),
|
||||
page.getByText(/Desktop lifecycle receiver is unavailable/),
|
||||
).toBeVisible();
|
||||
|
||||
// Lock the native and webview placements together: removing this explicit
|
||||
|
||||
Reference in New Issue
Block a user