diff --git a/docs/relay-queries.md b/docs/relay-queries.md index 4e756a40c4..73bd000dd6 100644 --- a/docs/relay-queries.md +++ b/docs/relay-queries.md @@ -379,6 +379,25 @@ roster, applied through shared discovery. That makes the channel a sidebar member and enables the composer, which then receives focus. An accepted but unconfirmed join reports that it may have taken effect and stays retryable. +On first run in a community, the session also joins the public starter channels +the community already has: `general`, `welcome-everyone` and `announcements` +(`starter-channels.ts`, mirroring the desktop's `ensure_starter_channels`). Once +the channel list is ready it looks each name up with `searchPublic(name, { exact })`, +which keeps only active public streams the viewer has not joined, and joins +each through the same lifecycle Join, one at a time. A bare `9021` takes the +channel's join role, so an announce channel (`guest`) makes the viewer a +read-only guest. It never creates a channel, and a missing starter is skipped. +Names match case-insensitively (the query is trimmed). The pass is silent and best +effort, skipped without a lifecycle writer (including cache-only sessions), +and ends with the connection. A `starter-channels-done` view-state marker per +viewer and community (`origin:viewer`) is saved only after a pass with no +failed join and no partial directory read (a lookup that hit the 20,000-event +bound cannot call a starter absent), so an incomplete pass retries next +session; leaving the community clears it with the rest of its view state. Starter +joins run as `background` lifecycle actions: the one-action-at-a-time rule makes +them wait for a viewer's own Join, Leave or Archive, and a viewer's action waits +for a background join already in flight, instead of failing as busy. + ### Finite read recovery A finite read that has no session recovery source retries transient failures diff --git a/src/features/relay/channel-lifecycle.test.ts b/src/features/relay/channel-lifecycle.test.ts index 320baff6db..00c3af2c1d 100644 --- a/src/features/relay/channel-lifecycle.test.ts +++ b/src/features/relay/channel-lifecycle.test.ts @@ -1826,6 +1826,126 @@ describe("join from a public preview", () => { }); }); +describe("background actions yield to foreground ones", () => { + /** A join whose publish is held until `gate` opens, so one action stays in flight. */ + function held() { + const h = harness("member"); + const metadata = h.record(39000, [ + ["d", id], + ["t", "stream"], + ["name", "fixture"], + ["public"], + ]); + const roster = (...members: string[]) => + h.record(39002, [["d", id], ...members.map((member) => ["p", member])]); + h.setEvents([metadata, roster(other)]); + h.deny(); + const gate = deferred(); + h.publish.mockImplementation(async (event: RelayEvent) => { + await gate.promise; + if (event.kind === 9021) h.setEvents([metadata, roster(other, viewer)]); + }); + const started = () => + vi.waitFor(() => expect(h.publish).toHaveBeenCalled()); + // Lets anything that could wrongly proceed run before a negative assertion. + const settle = () => new Promise((done) => setImmediate(done)); + return { ...h, gate, started, settle }; + } + const background = { background: true }; + + it("makes a foreground join wait for a background one, then succeed", async () => { + const h = held(); + const order: string[] = []; + try { + const first = h.owner.capability + .run("join", id, undefined, undefined, background) + .then(() => order.push("background")); + await h.started(); + const second = h.owner.capability + .run("join", id) + .then(() => order.push("foreground")); + await h.settle(); + expect(order).toEqual([]); + expect(h.sign).toHaveBeenCalledOnce(); + h.gate.resolve(); + await Promise.all([first, second]); + expect(order).toEqual(["background", "foreground"]); + // The foreground join found the roster the background one confirmed. + expect(h.publish).toHaveBeenCalledOnce(); + } finally { + h.gate.resolve(); + h.owner.dispose(); + } + }); + + it("makes a background join wait for a foreground action, then succeed", async () => { + const h = held(); + const order: string[] = []; + try { + const first = h.owner.capability + .run("join", id) + .then(() => order.push("foreground")); + await h.started(); + const second = h.owner.capability + .run("join", id, undefined, undefined, background) + .then(() => order.push("background")); + await h.settle(); + expect(order).toEqual([]); + h.gate.resolve(); + await Promise.all([first, second]); + expect(order).toEqual(["foreground", "background"]); + expect(h.publish).toHaveBeenCalledOnce(); + } finally { + h.gate.resolve(); + h.owner.dispose(); + } + }); + + it("still refuses a foreground action while another foreground one runs", async () => { + const h = held(); + try { + const first = h.owner.capability.run("join", id); + await h.started(); + await expect(h.owner.capability.run("join", id)).rejects.toThrow( + "Another channel action is still in progress", + ); + h.gate.resolve(); + await first; + } finally { + h.gate.resolve(); + h.owner.dispose(); + } + }); + + it("rejects an aborted foreground wait without publishing for it", async () => { + const h = held(); + const caller = new AbortController(); + try { + const first = h.owner.capability.run( + "join", + id, + undefined, + undefined, + background, + ); + await h.started(); + const waiting = h.owner.capability.run("join", id, caller.signal); + const rejected = expect(waiting).rejects.toMatchObject({ + name: "AbortError", + }); + caller.abort(new DOMException("Cancelled", "AbortError")); + await rejected; + h.gate.resolve(); + await first; + expect(h.publish).toHaveBeenCalledOnce(); + expect(h.sign).toHaveBeenCalledOnce(); + } finally { + h.gate.resolve(); + h.owner.dispose(); + } + }); +}); + it("retires queued entries and waits only after confirmed removal, including pending validation", async () => { const h = harness(); const deactivate = vi.fn(); diff --git a/src/features/relay/channel-lifecycle.ts b/src/features/relay/channel-lifecycle.ts index d1611a4add..73f3c059ad 100644 --- a/src/features/relay/channel-lifecycle.ts +++ b/src/features/relay/channel-lifecycle.ts @@ -45,6 +45,9 @@ export interface ChannelLifecycleCapability { signal?: AbortSignal, /** Checked synchronously right before publishing; false publishes nothing. */ active?: () => boolean, + /** Background work yields to every other action and waits its turn. A + * foreground action waits only for a background one. */ + options?: { background?: boolean }, ): Promise; snapshot(): DmVisibility; subscribe(listener: () => void): () => void; @@ -70,7 +73,8 @@ export function createChannelLifecycle({ removed(id: string): void; }) { let closed = false; - let busy = false; + // The one action in flight; `done` settles when it ends, however it ends. + let inflight: { background: boolean; done: Promise } | undefined; let epoch = 0; // Each in-flight operation and the channel it concerns; visibility reads have none. const controllers = new Map(); @@ -357,12 +361,38 @@ export function createChannelLifecycle({ value: string, caller?: AbortSignal, active?: () => boolean, + options?: { background?: boolean }, ) { if (!available || !writer) throw new Error("Channel actions are unavailable on this connection"); - if (busy) throw new Error("Another channel action is still in progress"); const id = lifecycleChannelId(value); - busy = true; + const background = options?.background === true; + // Wait out what yields, then look again: another caller may have started. + for (;;) { + const held = inflight; + if (!held) break; + if (!background && !held.background) + throw new Error("Another channel action is still in progress"); + caller?.throwIfAborted(); + let abort = () => {}; + try { + await Promise.race([ + held.done, + new Promise((_, reject) => { + abort = () => reject(caller?.reason); + caller?.addEventListener("abort", abort, { once: true }); + }), + ]); + } finally { + caller?.removeEventListener("abort", abort); + } + } + let release = () => {}; + const mine = { + background, + done: new Promise((resolve) => (release = resolve)), + }; + inflight = mine; let publicationStarted = false; try { await owned({ id, caller }, async (signal) => { @@ -501,7 +531,8 @@ export function createChannelLifecycle({ throw new ChannelLifecycleUnconfirmed(error); throw error; } finally { - busy = false; + inflight = undefined; + release(); } }, }); diff --git a/src/features/relay/service.ts b/src/features/relay/service.ts index 7391bfac0d..925676881b 100644 --- a/src/features/relay/service.ts +++ b/src/features/relay/service.ts @@ -14,6 +14,7 @@ import { import { isTransientFailure, ReadError, readErrorKind } from "./errors"; import { RECOVERY_DELAYS_MS } from "./recovery"; import type { ReadTransport } from "./transport"; +import { ensureStarterChannels } from "./starter-channels"; const WAKE_TICK_MS = 5_000; const WAKE_GAP_MS = 15_000; @@ -279,6 +280,13 @@ export function provideRelay( session: store.session, }); store.session.channels.ensureList(); + // First run in this community: join the starter channels it has. + // `signal` ends with the connection, so a replaced session stops. + void ensureStarterChannels( + store.session, + transportPartition(transport), + signal, + ); }) .catch(async (error) => { await restoring; diff --git a/src/features/relay/starter-channels.test.ts b/src/features/relay/starter-channels.test.ts new file mode 100644 index 0000000000..185383f66c --- /dev/null +++ b/src/features/relay/starter-channels.test.ts @@ -0,0 +1,249 @@ +// @vitest-environment jsdom +import { afterEach, describe, expect, it, vi } from "vitest"; +import { readView } from "../../shared/view-state"; +import type { ChannelSummary } from "./contracts"; +import type { RelaySession } from "./session"; +import { ensureStarterChannels } from "./starter-channels"; + +const scope = `https://example.test:${"ab".repeat(32)}`; +const marker = () => readView(scope, "starter-channels-done", false); +const stream = (name: string, extra: Partial = {}) => + ({ + id: `id-${name.toLowerCase()}`, + name, + channelType: "stream", + visibility: "public", + ...extra, + }) as ChannelSummary; + +/** The relay's unjoined public channels, found by exact name like the store. */ +function session( + directory: readonly ChannelSummary[], + run: (action: string, id: string) => Promise = async () => {}, + status: "loading" | "ready" = "ready", +) { + const listeners = new Set<() => void>(); + const state = { status }; + const searchPublic = vi.fn(async (query: string) => ({ + channels: directory.filter( + (channel) => channel.name.toLowerCase() === query, + ), + partial: false, + })); + const join = vi.fn( + ( + action: string, + id: string, + _signal?: AbortSignal, + _active?: () => boolean, + _options?: object, + ) => run(action, id), + ); + const fake = { + channels: { + list: () => ({ status: state.status, channels: [] }), + subscribeList(listener: () => void) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + searchPublic, + }, + channelLifecycle: { available: true, run: join }, + } as unknown as Pick; + return { + fake, + join, + searchPublic, + ready() { + state.status = "ready"; + for (const listener of listeners) listener(); + }, + listeners, + }; +} +const joined = (join: ReturnType["join"]) => + join.mock.calls.map(([, id]) => id); + +afterEach(() => localStorage.clear()); + +describe("ensureStarterChannels", () => { + it("joins existing starters one at a time and leaves other channels alone", async () => { + let open = 0; + let overlap = 0; + const { fake, join } = session( + [ + stream("random"), + stream("general"), + stream("welcome-everyone"), + stream("announcements", { joinRole: "guest" }), + ], + async () => { + overlap = Math.max(overlap, ++open); + await new Promise((resolve) => setTimeout(resolve, 0)); + open--; + }, + ); + await ensureStarterChannels(fake, scope, new AbortController().signal); + expect(joined(join)).toEqual([ + "id-general", + "id-welcome-everyone", + "id-announcements", + ]); + // A bare join: the lifecycle owner takes the channel's own join role. + expect(join.mock.calls.every(([action]) => action === "join")).toBe(true); + expect(overlap).toBe(1); + expect(marker()).toBe(true); + }); + + it("never creates a channel and skips a starter the community lacks", async () => { + const { fake, join } = session([stream("random"), stream("general")]); + await ensureStarterChannels(fake, scope, new AbortController().signal); + expect(joined(join)).toEqual(["id-general"]); + // Missing starters are not a failure, so the pass is complete. + expect(marker()).toBe(true); + }); + + it("leaves no marker after a partial lookup, but joins what it found", async () => { + const first = session([stream("general"), stream("announcements")]); + first.searchPublic.mockImplementation(async (query: string) => ({ + channels: query === "welcome-everyone" ? [] : [stream(query)], + partial: query === "welcome-everyone", + })); + await ensureStarterChannels( + first.fake, + scope, + new AbortController().signal, + ); + expect(joined(first.join)).toEqual(["id-general", "id-announcements"]); + expect(marker()).toBe(false); + // A missing starter is absent only when the whole directory was read. + const second = session([stream("welcome-everyone")]); + await ensureStarterChannels( + second.fake, + scope, + new AbortController().signal, + ); + expect(second.searchPublic).toHaveBeenCalledWith( + "welcome-everyone", + expect.anything(), + ); + expect(joined(second.join)).toEqual(["id-welcome-everyone"]); + expect(marker()).toBe(true); + }); + + it("joins in the background so a viewer's own action wins", async () => { + const { fake, join } = session([stream("general")]); + await ensureStarterChannels(fake, scope, new AbortController().signal); + expect(join.mock.calls[0]?.[4]).toEqual({ background: true }); + }); + + it("does nothing once the marker is set", async () => { + const first = session([stream("general")]); + await ensureStarterChannels( + first.fake, + scope, + new AbortController().signal, + ); + const second = session([stream("general"), stream("announcements")]); + await ensureStarterChannels( + second.fake, + scope, + new AbortController().signal, + ); + expect(second.searchPublic).not.toHaveBeenCalled(); + expect(second.join).not.toHaveBeenCalled(); + }); + + it("keeps the marker per viewer and community", async () => { + const { fake } = session([stream("general")]); + await ensureStarterChannels(fake, scope, new AbortController().signal); + const other = session([stream("general")]); + await ensureStarterChannels( + other.fake, + `https://other.test:${"ab".repeat(32)}`, + new AbortController().signal, + ); + expect(other.join).toHaveBeenCalledTimes(1); + }); + + it("leaves no marker after a failed join, so the next session retries", async () => { + const failing = session( + [stream("general"), stream("announcements")], + async (_, id) => { + if (id === "id-general") throw new Error("relay rejected the join"); + }, + ); + await ensureStarterChannels( + failing.fake, + scope, + new AbortController().signal, + ); + // One failure does not stop the others. + expect(joined(failing.join)).toEqual(["id-general", "id-announcements"]); + expect(marker()).toBe(false); + const retry = session([stream("general"), stream("announcements")]); + await ensureStarterChannels( + retry.fake, + scope, + new AbortController().signal, + ); + expect(retry.join).toHaveBeenCalledTimes(2); + expect(marker()).toBe(true); + }); + + it("matches names case-insensitively", async () => { + const { fake, join } = session([stream("Announcements")]); + await ensureStarterChannels(fake, scope, new AbortController().signal); + expect(joined(join)).toEqual(["id-announcements"]); + }); + + it("joins only active public streams", async () => { + const { fake, join } = session([ + stream("general", { channelType: "forum" }), + stream("welcome-everyone", { archived: true }), + stream("announcements"), + ]); + await ensureStarterChannels(fake, scope, new AbortController().signal); + expect(joined(join)).toEqual(["id-announcements"]); + }); + + it("skips sessions that cannot write", async () => { + const { fake, searchPublic } = session([stream("general")]); + await ensureStarterChannels( + { + channels: fake.channels, + channelLifecycle: { ...fake.channelLifecycle, available: false }, + }, + scope, + new AbortController().signal, + ); + expect(searchPublic).not.toHaveBeenCalled(); + expect(marker()).toBe(false); + }); + + it("waits for the channel list", async () => { + const waiting = session([stream("general")], undefined, "loading"); + const done = ensureStarterChannels( + waiting.fake, + scope, + new AbortController().signal, + ); + await Promise.resolve(); + expect(waiting.searchPublic).not.toHaveBeenCalled(); + waiting.ready(); + await done; + expect(waiting.join).toHaveBeenCalledTimes(1); + expect(waiting.listeners.size).toBe(0); + }); + + it("stops with the session without a marker", async () => { + const closed = new AbortController(); + const stopped = session([stream("general")], undefined, "loading"); + const pending = ensureStarterChannels(stopped.fake, scope, closed.signal); + closed.abort(); + await pending; + expect(stopped.listeners.size).toBe(0); + expect(stopped.join).not.toHaveBeenCalled(); + expect(marker()).toBe(false); + }); +}); diff --git a/src/features/relay/starter-channels.ts b/src/features/relay/starter-channels.ts new file mode 100644 index 0000000000..684231e311 --- /dev/null +++ b/src/features/relay/starter-channels.ts @@ -0,0 +1,83 @@ +import { readView, writeView } from "../../shared/view-state"; +import type { RelaySession } from "./session"; + +/** The public channels a community already has that every new member joins. + * Never created here: a community without one simply has fewer. */ +export const STARTER_CHANNELS = [ + "general", + "welcome-everyone", + "announcements", +] as const; +const DONE_KEY = "starter-channels-done"; + +/** Waits for the viewer's channel list, which public name search requires. */ +function listReady(channels: RelaySession["channels"], signal: AbortSignal) { + return new Promise((resolve, reject) => { + const check = () => { + if (!signal.aborted && channels.list().status !== "ready") return; + unsubscribe(); + signal.removeEventListener("abort", check); + if (signal.aborted) reject(signal.reason); + else resolve(); + }; + const unsubscribe = channels.subscribeList(check); + signal.addEventListener("abort", check, { once: true }); + check(); + }); +} + +/** Once per viewer and community (`scope` is `origin:viewer`), joins the + * starter channels that exist, are open and active, and that the viewer has + * not joined. A bare join takes the channel's own join role, so an announce + * channel makes the viewer a read-only guest. Best effort and silent: the + * marker is saved only after a complete pass (no failed join, no partial + * directory read), so an incomplete one retries next session. Joins run as + * background lifecycle actions: a viewer's own channel action waits for the + * one in flight instead of failing as busy. `signal` ends with the relay + * session; nothing is navigated. */ +export async function ensureStarterChannels( + session: Pick, + scope: string, + signal: AbortSignal, +): Promise { + const { channels, channelLifecycle: lifecycle } = session; + if (!lifecycle.available || !channels.searchPublic) return; + if (readView(scope, DONE_KEY, false)) return; + try { + await listReady(channels, signal); + let incomplete = false; + // The lifecycle owner runs one action at a time, so joins are sequential. + for (const name of STARTER_CHANNELS) { + const { channels: found, partial } = await channels.searchPublic(name, { + exact: true, + signal, + }); + // A directory read that hit its bound cannot say the starter is absent. + if (partial) incomplete = true; + for (const channel of found) { + if ( + channel.channelType !== "stream" || + channel.archived || + channel.name.toLowerCase() !== name + ) + continue; + try { + await lifecycle.run( + "join", + channel.id, + signal, + () => !signal.aborted, + { background: true }, + ); + } catch { + signal.throwIfAborted(); + incomplete = true; + } + } + } + signal.throwIfAborted(); + if (!incomplete) writeView(scope, DONE_KEY, true); + } catch { + /* Aborted with the session, or the relay was unavailable: retry next session. */ + } +}