mirror of
https://github.com/block/buzz.git
synced 2026-10-29 13:31:44 +00:00
fix(desktop): refresh channels after access-revoked closure (#7784)
Pinky is updating this PR on Wes’s behalf. ## Summary - Treat exact `CLOSED restricted: channel access revoked` on a known channel-scoped live subscription as a hint to refresh authoritative channel state. - Reuse `useLiveChannelUpdates`’ existing debounced, idle-aware query refresh. Keep terminal retirement, explicit membership denial, and polling unchanged. - Cover archived, unchanged, removed, and failed authoritative results; burst coalescing; irrelevant/stale frames; exact-reason rejection; and unmount cleanup with seven production-path regression cases. The hint does **not** establish archive or membership state. This closes a missed-update recovery gap without changing the relay or blindly retrying subscriptions. Related symptom report: https://github.com/block/buzz/issues/7026 (not claimed fully resolved; archived-channel delete policy is separate). Related new-client fix: https://github.com/block/buzz-app/pull/106 Originating conversation: buzz://message?channel=cb38cd5a-c920-40a1-b770-e9b89c289476&id=d22c1eed473016cf1f73c889a4cc88f1491ee092cfe3794575f8fea8003aa19e ## Review follow-up and main integration At **`51ec7482ad1e600bd02ea6b9ed265db8a70e7266`**, based on main **`c213d90d5c450e2e579505da4a4145f528ecbcf2`**: - Addressed Jude’s P2: a known channel-scoped live subscription closed with `restricted: not a channel member` stays terminal but must not refresh channels. - Mutation verification: replacing the exact reason predicate with `startsWith("restricted:")` produced **6,588 passes / 1 expected failure**, specifically the new no-refresh assertion (`actual: 1`, `expected: 0`). Restored the exact predicate; the final full suite passed **6,589/6,589**. - Resolved both conflicts with #7790, retaining main’s paced/cancellable subscriptions and this PR’s hint after terminal cleanup/drain cancellation. The fixture advances the production drain with a consistent virtual clock; it does not stub or bypass admission. - Integration crossed the session file’s 1,200-line ceiling (1,207). Extracted only this PR’s exact predicate/listener set into `relayChannelAccessRevocations.ts`, with unchanged session ownership and cleanup. No limit change, new timer, retry policy, or unrelated refactor. Final feature diff: 46 production additions across three files, plus the 333-line regression fixture. - Independent read-only review of the integration and final extraction found no blocking findings. This is not GitHub approval. ## Validation At exact pushed head **`51ec7482ad1e600bd02ea6b9ed265db8a70e7266`**: - Mandatory hooks passed without bypasses: desktop lint/policy checks, TypeScript, repository file-size gate, branch-skew check, and full desktop suite **6,589/6,589**, zero failed/cancelled/skipped. - Working tree clean after push. DCO passed on GitHub. - Fresh hosted CI and Desktop Release Candidate checks are running; prior-head green checks are not evidence for this head. Re-review requested from `jedwards27`; the changes-requested review and required approval remain for the reviewer to update. ## Evidence boundaries - Native IPC is stubbed; the fixture exercises actual subscription creation, raw CLOSED dispatch, mounted hook, authoritative query refresh, and IPC result conversion. It does not exercise native discovery or rendered sidebar behavior. Existing in-flight refresh retention remains source/helper-test coverage rather than a new integrated case. - No native app launch, live archive mutation, or release certification. This fixes a demonstrated recovery mechanism, not proof of the exact original incident ordering. - The initial broad local `just ci` attempt hit a 300-second timeout compiling `buzz-voice`; full local `just ci` remains incomplete. Current delivery uses mandatory path-scoped hooks and fresh hosted CI; no unrelated native/mobile lane was bypassed or claimed rerun locally. --------- Signed-off-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz> Co-authored-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz>
This commit is contained in:
333
desktop/src/features/channels/channelAccessRevocation.test.mjs
Normal file
333
desktop/src/features/channels/channelAccessRevocation.test.mjs
Normal file
@@ -0,0 +1,333 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { after, before, test } from "node:test";
|
||||
import { JSDOM } from "jsdom";
|
||||
|
||||
const dom = new JSDOM("<!doctype html><html><body></body></html>", {
|
||||
url: "http://localhost",
|
||||
});
|
||||
before(() => {
|
||||
Object.assign(globalThis, {
|
||||
document: dom.window.document,
|
||||
HTMLElement: dom.window.HTMLElement,
|
||||
IS_REACT_ACT_ENVIRONMENT: true,
|
||||
window: dom.window,
|
||||
localStorage: dom.window.localStorage,
|
||||
});
|
||||
});
|
||||
after(() => dom.window.close());
|
||||
|
||||
const rawChannels = ["a", "b"].map((id) => ({
|
||||
id,
|
||||
name: id,
|
||||
channel_type: "stream",
|
||||
visibility: "private",
|
||||
description: "",
|
||||
member_count: 1,
|
||||
member_pubkeys: ["viewer"],
|
||||
archived_at: null,
|
||||
last_message_at: null,
|
||||
participants: [],
|
||||
participant_pubkeys: [],
|
||||
is_member: true,
|
||||
}));
|
||||
const revoked = "restricted: channel access revoked";
|
||||
|
||||
async function mount() {
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
const React = await import("react");
|
||||
const { QueryClient, QueryClientProvider, useQuery } = await import(
|
||||
"@tanstack/react-query"
|
||||
);
|
||||
const { relayClient } = await import("@/shared/api/relayClient");
|
||||
const { fromRawChannel } = await import("@/shared/api/tauriChannels");
|
||||
const { channelsQueryKey, refreshChannelsQuery } = await import("./hooks.ts");
|
||||
const { useLiveChannelUpdates } = await import("./useLiveChannelUpdates.ts");
|
||||
const original = {
|
||||
setTimeout: window.setTimeout,
|
||||
clearTimeout: window.clearTimeout,
|
||||
internals: window.__TAURI_INTERNALS__,
|
||||
now: Date.now,
|
||||
};
|
||||
const timers = new Map();
|
||||
let now = Date.now();
|
||||
Date.now = () => now;
|
||||
let nextTimer = 0;
|
||||
window.setTimeout = (fn, ms) => {
|
||||
timers.set(++nextTimer, { fn, at: now + ms });
|
||||
return nextTimer;
|
||||
};
|
||||
window.clearTimeout = (id) => timers.delete(id);
|
||||
const sent = [];
|
||||
const reads = [];
|
||||
window.__TAURI_INTERNALS__ = {
|
||||
invoke: async (command, args) => {
|
||||
if (command === "plugin:websocket|send") {
|
||||
const frame = JSON.parse(args.message.data);
|
||||
sent.push(frame);
|
||||
if (frame[0] === "REQ")
|
||||
queueMicrotask(() => deliver(["EOSE", frame[1]]));
|
||||
return;
|
||||
}
|
||||
if (command === "get_channels") {
|
||||
return new Promise((resolve, reject) =>
|
||||
reads.push({ args, resolve, reject }),
|
||||
);
|
||||
}
|
||||
assert.fail(`Unexpected IPC command: ${command}`);
|
||||
},
|
||||
};
|
||||
// Stub only native transport, not subscription creation or CLOSED dispatch.
|
||||
relayClient.wsId = 7;
|
||||
const deliver = (frame, generation = relayClient.connectionGeneration) =>
|
||||
relayClient.handleWsMessage(JSON.stringify(frame), generation);
|
||||
const queryClient = new QueryClient({
|
||||
defaultOptions: { queries: { retry: false, gcTime: Infinity } },
|
||||
});
|
||||
const hook = renderHook(
|
||||
() => {
|
||||
const query = useQuery({
|
||||
queryKey: channelsQueryKey,
|
||||
queryFn: () =>
|
||||
refreshChannelsQuery({
|
||||
queryClient,
|
||||
initialSnapshotPair: null,
|
||||
relayUrl: null,
|
||||
ownerPubkey: null,
|
||||
}),
|
||||
initialData: rawChannels.map(fromRawChannel),
|
||||
staleTime: Infinity,
|
||||
});
|
||||
useLiveChannelUpdates(query.data, null);
|
||||
return query;
|
||||
},
|
||||
{
|
||||
wrapper: ({ children }) =>
|
||||
React.createElement(
|
||||
QueryClientProvider,
|
||||
{ client: queryClient },
|
||||
children,
|
||||
),
|
||||
},
|
||||
);
|
||||
const settle = () =>
|
||||
act(async () => {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
});
|
||||
const tick = async (ms) => {
|
||||
await act(async () => {
|
||||
const target = now + ms;
|
||||
let fired = 0;
|
||||
for (;;) {
|
||||
const next = [...timers]
|
||||
.filter(([, timer]) => timer.at <= target)
|
||||
.sort((a, b) => a[1].at - b[1].at || a[0] - b[0])[0];
|
||||
if (!next) break;
|
||||
assert.ok(++fired < 1000, "timer loop must make progress");
|
||||
now = next[1].at;
|
||||
timers.delete(next[0]);
|
||||
next[1].fn();
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
now = target;
|
||||
});
|
||||
await settle();
|
||||
};
|
||||
await settle();
|
||||
// Admit both subscriptions through the production drain before testing CLOSED.
|
||||
await tick(500);
|
||||
assert.equal(sent.filter(([type]) => type === "REQ").length, 2);
|
||||
const ids = sent.filter(([type]) => type === "REQ").map(([, id]) => id);
|
||||
return {
|
||||
reads,
|
||||
sent,
|
||||
ids,
|
||||
timers,
|
||||
relayClient,
|
||||
queryClient,
|
||||
channelsQueryKey,
|
||||
state: () => queryClient.getQueryState(channelsQueryKey),
|
||||
data: () => queryClient.getQueryData(channelsQueryKey),
|
||||
close: (id = ids[0], reason = revoked, generation) =>
|
||||
act(async () => {
|
||||
await deliver(["CLOSED", id, reason], generation);
|
||||
}),
|
||||
tick,
|
||||
async respond(channels = rawChannels) {
|
||||
assert.equal(
|
||||
reads.length,
|
||||
1,
|
||||
"one authoritative refresh must be pending",
|
||||
);
|
||||
await act(async () =>
|
||||
reads[0].resolve({ hash: "fresh", channels, last_messages: {} }),
|
||||
);
|
||||
await settle();
|
||||
},
|
||||
async fail() {
|
||||
assert.equal(
|
||||
reads.length,
|
||||
1,
|
||||
"one authoritative refresh must be pending",
|
||||
);
|
||||
await act(async () => reads[0].reject(new Error("fixture offline")));
|
||||
await settle();
|
||||
},
|
||||
unmount: hook.unmount,
|
||||
restore() {
|
||||
hook.unmount();
|
||||
cleanup();
|
||||
queryClient.clear();
|
||||
// Avoid a native disconnect command; clear the real session's listeners/state.
|
||||
relayClient.wsId = null;
|
||||
relayClient.disconnect();
|
||||
window.setTimeout = original.setTimeout;
|
||||
window.clearTimeout = original.clearTimeout;
|
||||
window.__TAURI_INTERNALS__ = original.internals;
|
||||
Date.now = original.now;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
for (const outcome of ["archived", "unchanged", "removed", "failed"]) {
|
||||
test(`CLOSED refreshes authoritative channel state: ${outcome}`, async () => {
|
||||
const h = await mount();
|
||||
try {
|
||||
await h.close();
|
||||
await h.close(); // duplicate frame is not a second hint
|
||||
await h.close(h.ids[1]); // a second channel shares the debounce
|
||||
assert.equal(
|
||||
h.relayClient.subscriptions.size,
|
||||
0,
|
||||
"terminal subscriptions stay removed",
|
||||
);
|
||||
assert.equal(h.reads.length, 0);
|
||||
assert.ok(
|
||||
h
|
||||
.data()
|
||||
.every((channel) => channel.archivedAt === null && channel.isMember),
|
||||
);
|
||||
await h.tick(499);
|
||||
assert.equal(h.reads.length, 0);
|
||||
await h.tick(1);
|
||||
assert.equal(
|
||||
h.reads.length,
|
||||
1,
|
||||
"burst coalesces into one query without waiting for polling",
|
||||
);
|
||||
assert.equal(h.data().length, 2, "CLOSED alone cannot remove membership");
|
||||
if (outcome === "failed") {
|
||||
await h.fail();
|
||||
assert.equal(h.state().status, "error");
|
||||
assert.equal(h.data().length, 2);
|
||||
assert.ok(
|
||||
h
|
||||
.data()
|
||||
.every(
|
||||
(channel) => channel.archivedAt === null && channel.isMember,
|
||||
),
|
||||
);
|
||||
} else {
|
||||
await h.respond(
|
||||
outcome === "archived"
|
||||
? [
|
||||
{ ...rawChannels[0], archived_at: "2026-09-21T15:36:31Z" },
|
||||
rawChannels[1],
|
||||
]
|
||||
: outcome === "removed"
|
||||
? rawChannels.slice(1)
|
||||
: rawChannels,
|
||||
);
|
||||
assert.deepEqual(
|
||||
h.data().map((channel) => channel.id),
|
||||
outcome === "removed" ? ["b"] : ["a", "b"],
|
||||
);
|
||||
assert.equal(
|
||||
h.data().find((channel) => channel.id === "a")?.archivedAt,
|
||||
outcome === "archived"
|
||||
? "2026-09-21T15:36:31Z"
|
||||
: outcome === "removed"
|
||||
? undefined
|
||||
: null,
|
||||
);
|
||||
}
|
||||
await h.tick(30_000);
|
||||
assert.equal(
|
||||
h.reads.length,
|
||||
1,
|
||||
"refresh failure or unchanged state must not loop",
|
||||
);
|
||||
assert.equal(
|
||||
h.sent.filter(([type]) => type === "REQ").length,
|
||||
2,
|
||||
"no terminal resubscribe",
|
||||
);
|
||||
} finally {
|
||||
h.restore();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
test("unknown, non-channel, unrelated and old-generation CLOSED do not refresh channels", async () => {
|
||||
const h = await mount();
|
||||
try {
|
||||
await h.close("unknown");
|
||||
await h.close(h.ids[0], revoked, h.relayClient.connectionGeneration - 1);
|
||||
await h.close(h.ids[0], "invalid: bad filter");
|
||||
const dispose = await h.relayClient.subscribeLive(
|
||||
{ kinds: [0], limit: 1 },
|
||||
() => {},
|
||||
);
|
||||
const globalId = h.sent.at(-1)[1];
|
||||
await h.close(globalId);
|
||||
await dispose();
|
||||
await h.tick(500);
|
||||
assert.equal(h.reads.length, 0);
|
||||
assert.equal(h.relayClient.subscriptions.size, 1);
|
||||
} finally {
|
||||
h.restore();
|
||||
}
|
||||
});
|
||||
|
||||
test("a different restricted CLOSED reason does not refresh channels", async () => {
|
||||
const h = await mount();
|
||||
try {
|
||||
await h.close(h.ids[0], "restricted: not a channel member");
|
||||
assert.equal(
|
||||
h.relayClient.subscriptions.has(h.ids[0]),
|
||||
false,
|
||||
"the known channel subscription is still terminal",
|
||||
);
|
||||
await h.tick(500);
|
||||
assert.equal(
|
||||
h.reads.length,
|
||||
0,
|
||||
"only the exact revocation reason refreshes",
|
||||
);
|
||||
} finally {
|
||||
h.restore();
|
||||
}
|
||||
});
|
||||
|
||||
test("unmount cancels queued invalidation and removes its session listener", async () => {
|
||||
const h = await mount();
|
||||
try {
|
||||
await h.close();
|
||||
h.unmount();
|
||||
await h.tick(500);
|
||||
assert.equal(h.reads.length, 0);
|
||||
const dispose = await h.relayClient.subscribeLive(
|
||||
{ kinds: [9], "#h": ["later"], limit: 1 },
|
||||
() => {},
|
||||
);
|
||||
await h.close(h.sent.at(-1)[1]);
|
||||
await h.tick(500);
|
||||
assert.equal(
|
||||
h.timers.size,
|
||||
0,
|
||||
"unmounted listener must not schedule another invalidation",
|
||||
);
|
||||
await dispose();
|
||||
} finally {
|
||||
h.restore();
|
||||
}
|
||||
});
|
||||
@@ -372,6 +372,14 @@ export function useLiveChannelUpdates(
|
||||
});
|
||||
}, [queryClient]);
|
||||
|
||||
React.useEffect(
|
||||
() =>
|
||||
relayClient.subscribeToChannelAccessRevocations(
|
||||
invalidateChannelsDebounced,
|
||||
),
|
||||
[invalidateChannelsDebounced],
|
||||
);
|
||||
|
||||
const liveSubsRef = React.useRef(new Map<string, LiveChannelEntry>());
|
||||
|
||||
React.useEffect(() => {
|
||||
|
||||
28
desktop/src/shared/api/relayChannelAccessRevocations.ts
Normal file
28
desktop/src/shared/api/relayChannelAccessRevocations.ts
Normal file
@@ -0,0 +1,28 @@
|
||||
import type { RelaySubscription } from "./relayClientShared";
|
||||
|
||||
/** Session-owned hints; CLOSED never establishes archive or membership state. */
|
||||
export class RelayChannelAccessRevocations {
|
||||
private listeners = new Set<() => void>();
|
||||
|
||||
subscribe(listener: () => void) {
|
||||
this.listeners.add(listener);
|
||||
return () => {
|
||||
this.listeners.delete(listener);
|
||||
};
|
||||
}
|
||||
|
||||
clear() {
|
||||
this.listeners.clear();
|
||||
}
|
||||
|
||||
/** Call after terminal cleanup, using the subscription captured before removal. */
|
||||
notify(subscription: RelaySubscription | undefined, reason: unknown) {
|
||||
if (
|
||||
subscription?.mode === "live" &&
|
||||
(subscription.filter["#h"]?.length ?? 0) > 0 &&
|
||||
reason === "restricted: channel access revoked"
|
||||
) {
|
||||
for (const listener of this.listeners) listener();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -38,6 +38,7 @@ import {
|
||||
import { getChannelReconnectRepairEvents } from "@/shared/api/channelReconnectRepair";
|
||||
import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay";
|
||||
import { RelayLiveReqDrain } from "./relayLiveReqDrain";
|
||||
import { RelayChannelAccessRevocations } from "./relayChannelAccessRevocations";
|
||||
import { publishSessionEvent } from "@/shared/api/relayEventPublisher";
|
||||
import { activateRateLimitIfSignalled } from "@/shared/api/relayRateLimitGate";
|
||||
import {
|
||||
@@ -90,6 +91,7 @@ export class RelayClient {
|
||||
private eventBuffer: SubscriptionEventBufferItem[] = [];
|
||||
private flushTimeout: number | null = null;
|
||||
private reconnectListeners = new Set<() => void>();
|
||||
private channelAccessRevocations = new RelayChannelAccessRevocations();
|
||||
private hasConnectedOnce = false;
|
||||
private notifyReconnectListeners = false;
|
||||
private onMessageChannel: Channel<unknown> | null = null;
|
||||
@@ -176,6 +178,7 @@ export class RelayClient {
|
||||
}
|
||||
this.eventBuffer = [];
|
||||
this.reconnectListeners.clear();
|
||||
this.channelAccessRevocations.clear();
|
||||
this.connectionStateEmitter.clear();
|
||||
this.onMessageChannel = null;
|
||||
this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS;
|
||||
@@ -478,6 +481,11 @@ export class RelayClient {
|
||||
};
|
||||
}
|
||||
|
||||
/** Listen for channel access-revocation hints; authoritative state needs a refresh. */
|
||||
subscribeToChannelAccessRevocations(listener: () => void) {
|
||||
return this.channelAccessRevocations.subscribe(listener);
|
||||
}
|
||||
|
||||
/** Current connection state — synchronous read. */
|
||||
getConnectionState(): ConnectionState {
|
||||
return this.connectionStateEmitter.get();
|
||||
@@ -913,6 +921,7 @@ export class RelayClient {
|
||||
return;
|
||||
}
|
||||
if (type === "CLOSED" && typeof rest[0] === "string") {
|
||||
const subscription = this.subscriptions.get(rest[0]);
|
||||
handleRelayClosed({
|
||||
subscriptions: this.subscriptions,
|
||||
subId: rest[0],
|
||||
@@ -925,6 +934,7 @@ export class RelayClient {
|
||||
closeSubscription: (subId) => this.closeSubscription(subId),
|
||||
});
|
||||
if (!this.subscriptions.has(rest[0])) this.liveReqDrain.cancel(rest[0]);
|
||||
this.channelAccessRevocations.notify(subscription, rest[1]);
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user