Propagate community leave across windows

docs/communities.md

@@ -115,6 +115,19 @@ empty state there. Try drafting in A, switching to B, then returning to A.
 Selected channels, drafts and reading offsets are partitioned by the canonical
 community origin and viewer; channel IDs alone are not sufficient keys.
 
+Windows on the same app origin share the saved memberships and selection.
+Each snapshot and command reads a changed device record before it uses membership.
+Storage events also notify readers and dispose sessions for removed memberships.
+A new local join has a generation value. Artifact handles capture this value, so
+leave and rejoin cannot restore an old handle. Records without a generation remain
+readable. Explicit rejoin creates a new handle and session. Account sync also assigns a new
+generation when it adds a membership. Snapshot reads notify subscribers after the
+read returns, so a sync owner cannot start a second drain from inside its snapshot.
+
+Leave takes the artifact partition lock before it removes membership. It holds the
+lock through device purge. New artifact collections wait for that purge before
+they initialize, so a concurrent rejoin keeps its fresh records.
+
 ## Import from the previous desktop app
 
 After native startup, Buzz offers the previous desktop app's saved communities

docs/local-artifact-storage.md

@@ -132,5 +132,5 @@ and indexed pages use all four stores. Metadata operations use only metadata.
 Large row batches can still delay other collections' reads or writes. A changed row rewrites
 its indexes. Index upgrades scan the collection. Delta completion does not scan records.
 Ancestry comparison and separate source-membership writes are absent. The database upgrade removes the obsolete membership store, builds the host
-channel index, and preserves records and device preferences. Leaving a community purges only its saved partition. If database
+channel index, and preserves records and device preferences. Leaving a community purges only its saved partition. The community service propagates membership and selection changes to other windows. Handles capture the local membership generation and close when it changes. Opens read the current device record, so a window cannot reopen a left partition before its storage event arrives. Explicit rejoin permits new handles; old handles stay closed. Leave holds an exclusive Web Lock for the partition before membership removal, through purge. Collection initialization takes the same lock in shared mode. New collections wait for an in-flight purge. Queued initialization and writes also validate the captured membership. If database
 opening fails, close that handle and open another after the cause clears.

src/bundled/builderlab/known-communities/sync.test.ts

@@ -1,5 +1,7 @@
 // @vitest-environment jsdom
 import { afterEach, beforeEach, expect, it, vi } from "vitest";
+import { Context } from "@deepseek-ai/cordis";
+import { createCommunities } from "../../../features/communities/service";
 import {
   emptySync,
   enqueue,
@@ -804,3 +806,67 @@ it("disposal stops listening", async () => {
   await vi.advanceTimersByTimeAsync(120_000);
   expect(h.client.add).not.toHaveBeenCalled();
 });
+
+it("keeps one upload in flight when a wake reads a changed device record", async () => {
+  const root = new Context();
+  const record = `buzz-client.v1:${viewer}`;
+  localStorage.removeItem(record);
+  const communities = createCommunities(
+    root,
+    false,
+    undefined,
+    "",
+    undefined,
+    Promise.resolve(viewer),
+  );
+  await until(() => expect(communities.snapshot().status).toBe("ready"));
+  const client = fakeClient();
+  const upload = deferred<UpdateResult>();
+  client.add.mockReturnValue(upload.promise);
+  const session = createOAuthSession(async () => ({
+    value: "fixture",
+    account: {
+      subject: "user",
+      email: "fixture@example.test",
+      canUseRemoteAgents: false,
+    },
+  }));
+  const stop = startKnownCommunitiesSync({
+    client,
+    session,
+    knownCommunities: {
+      snapshot: communities.snapshot,
+      subscribe: communities.subscribe,
+      pending: communities.pendingSync,
+      apply: communities.applySync,
+    },
+  });
+  try {
+    await session.signIn();
+    await until(() => expect(communities.snapshot().sync.account).toBe("user"));
+    localStorage.setItem(
+      record,
+      JSON.stringify({
+        ...communities.snapshot(),
+        memberships: [{ id: "https://primary.example", name: "Primary" }],
+        sync: {
+          known: [],
+          outbox: [{ url: primary, removed: false }],
+          account: "user",
+        },
+      }),
+    );
+    window.dispatchEvent(new Event("online"));
+    await until(() => expect(client.add).toHaveBeenCalled());
+    expect(client.add).toHaveBeenCalledTimes(1);
+    upload.resolve(accepted);
+    await until(() => expect(communities.pendingSync()).toEqual([]));
+    expect(client.add).toHaveBeenCalledTimes(1);
+  } finally {
+    upload.resolve(accepted);
+    stop();
+    session.dispose();
+    await root.fiber.dispose();
+    localStorage.removeItem(record);
+  }
+});

src/features/artifact-storage/service.test.ts

@@ -4,7 +4,11 @@ import { ArtifactStorageService } from "./service";
 
 it("binds collection ownership to the calling plugin and fences scope changes", async () => {
   const root = new Context();
-  let scope = { viewer: "viewer", selected: "https://relay.test" };
+  let scope = {
+    viewer: "viewer",
+    selected: "https://relay.test",
+    memberships: [{ id: "https://relay.test" }],
+  };
   const subscribers = new Set<() => void>();
   root.provide("communityReader", {
     snapshot: () => scope,
@@ -53,7 +57,11 @@ it("binds collection ownership to the calling plugin and fences scope changes",
   }
   listener.mockClear();
   const handle = plugin.artifactStorage.open(definition);
-  expect(factory).toHaveBeenCalledWith("test.plugin", definition);
+  expect(factory).toHaveBeenCalledWith(
+    "test.plugin",
+    definition,
+    expect.any(Function),
+  );
   expect(listener.mock.calls[0]?.[0]).toMatchObject({
     owner: "test.plugin",
     viewer: "viewer",
@@ -66,10 +74,20 @@ it("binds collection ownership to the calling plugin and fences scope changes",
   expect(refreshError).toHaveBeenCalledOnce();
   await handle.get("task");
   expect(get).toHaveBeenCalledWith("task");
-  scope = { viewer: "other", selected: "https://relay.test" };
+  let finishRead!: () => void;
+  get.mockImplementationOnce(
+    () =>
+      new Promise<undefined>((resolve) => {
+        finishRead = () => resolve(undefined);
+      }),
+  );
+  const pending = handle.get("task");
+  scope = { ...scope, viewer: "other" };
   for (const listener of subscribers) listener();
   expect(listener.mock.calls[0]?.[0].signal.aborted).toBe(true);
   expect(close).toHaveBeenCalled();
   expect(() => handle.get("task")).toThrow("scope changed");
+  finishRead();
+  await expect(pending).rejects.toThrow("scope changed");
   await root.fiber.dispose();
 });

src/features/artifact-storage/service.ts

@@ -47,13 +47,19 @@ export class ArtifactStorageService extends Service implements ArtifactStorage {
   open(definition: ArtifactCollectionDefinition): ArtifactCollection {
     const owner = this.ctx.pluginOwner?.id;
     const capture = () => {
-      const { viewer, selected } = this.ctx.communityReader.snapshot();
+      const { viewer, selected, memberships } =
+        this.ctx.communityReader.snapshot();
+      const membership = memberships.find((entry) => entry.id === selected);
       return {
         viewer,
-        origin: selected ? communityDestination(selected).url : undefined,
+        generation: membership?.generation,
+        origin:
+          selected && membership
+            ? communityDestination(selected).url
+            : undefined,
       };
     };
-    const { viewer, origin } = capture();
+    const { viewer, origin, generation } = capture();
     if (!owner || !viewer || !origin)
       throw new Error("Artifact storage needs a plugin, identity and relay");
     const scope = JSON.stringify([viewer, origin]);
@@ -62,7 +68,7 @@ export class ArtifactStorageService extends Service implements ArtifactStorage {
       store = createArtifactStore(viewer, origin);
       this.stores.set(scope, store);
     }
-    const collection = store.collection(owner, definition);
+    const collection = store.collection(owner, definition, valid);
     const controller = new AbortController();
     let unsubscribe: (() => void) | undefined;
     let removeEffect: (() => void) | undefined;
@@ -73,18 +79,26 @@ export class ArtifactStorageService extends Service implements ArtifactStorage {
       collection.close();
       removeEffect?.();
     };
-    const valid = () => {
+    function valid() {
       const now = capture();
       if (
         controller.signal.aborted ||
         now.viewer !== viewer ||
-        now.origin !== origin
-      )
+        now.origin !== origin ||
+        now.generation !== generation
+      ) {
+        close();
         throw new Error("Artifact storage scope changed");
-    };
+      }
+    }
     unsubscribe = this.ctx.communityReader.subscribe(() => {
       const now = capture();
-      if (now.viewer !== viewer || now.origin !== origin) close();
+      if (
+        now.viewer !== viewer ||
+        now.origin !== origin ||
+        now.generation !== generation
+      )
+        close();
     });
     const checked = async <T>(operation: Promise<T>) => {
       const result = await operation;

src/features/artifact-storage/store.ts

@@ -29,6 +29,21 @@ function sameRefresh(
   return left?.source === right.source && left.generation === right.generation;
 }
 
+/** Leave holds this lock before it removes membership, through the whole purge. */
+export function withArtifactPartitionLock<T>(
+  viewer: string,
+  relay: string,
+  mode: LockMode,
+  work: () => Promise<T>,
+  signal?: AbortSignal,
+): Promise<T> {
+  return navigator.locks.request(
+    `${DB}:${relayPartition(relay, viewer)}`,
+    { mode, ...(signal ? { signal } : {}) },
+    work,
+  );
+}
+
 /** Same IndexedDB transaction boundary as the relay journals. No head-cache eviction. */
 export function createArtifactStore(viewer: string, relay: string) {
   if (!viewer || !relay)
@@ -174,6 +189,7 @@ export function createArtifactStore(viewer: string, relay: string) {
   function collection(
     owner: string,
     definition: ArtifactCollectionDefinition,
+    validate?: () => void,
   ): HostArtifactCollection {
     if (closed) throw new Error("Artifact storage closed");
     if (
@@ -234,46 +250,56 @@ export function createArtifactStore(viewer: string, relay: string) {
         },
       );
     }
-    const ready = transact<void>(
-      "readwrite",
-      STORES,
-      (tx, then, done) => {
-        const metadata = tx.objectStore("metadata");
-        then(metadata.get(key("schema")), (schema) => {
-          if (schema?.type && schema.type !== definition.type)
-            throw new Error("Artifact collection type changed");
-          if (schema?.version === definition.version) {
-            done();
-            return;
-          }
-          if (schema?.version > definition.version)
-            throw new Error("Artifact collection requires a newer plugin");
-          metadata.delete(key("source"));
-          metadata.delete(key("refresh"));
-          metadata.delete(key("checkpoint"));
-          const request = tx.objectStore("records").openCursor(range(prefix));
-          then(request, (cursor) => {
-            if (!cursor) {
-              metadata.put({
-                scope,
-                collection,
-                key: "schema",
-                version: definition.version,
-                type: definition.type,
+    const ready = withArtifactPartitionLock(
+      viewer,
+      relay,
+      "shared",
+      () =>
+        transact<void>(
+          "readwrite",
+          STORES,
+          (tx, then, done) => {
+            validate?.();
+            const metadata = tx.objectStore("metadata");
+            then(metadata.get(key("schema")), (schema) => {
+              if (schema?.type && schema.type !== definition.type)
+                throw new Error("Artifact collection type changed");
+              if (schema?.version === definition.version) {
+                done();
+                return;
+              }
+              if (schema?.version > definition.version)
+                throw new Error("Artifact collection requires a newer plugin");
+              metadata.delete(key("source"));
+              metadata.delete(key("refresh"));
+              metadata.delete(key("checkpoint"));
+              const request = tx
+                .objectStore("records")
+                .openCursor(range(prefix));
+              then(request, (cursor) => {
+                if (!cursor) {
+                  metadata.put({
+                    scope,
+                    collection,
+                    key: "schema",
+                    version: definition.version,
+                    type: definition.type,
+                  });
+                  done();
+                  return;
+                }
+                then(
+                  tx.objectStore("preferences").get(key(cursor.value.id)),
+                  (pref: StoredPreference | undefined) =>
+                    reindex(tx, then, cursor.value, pref?.value, () =>
+                      cursor.continue(),
+                    ),
+                );
               });
-              done();
-              return;
-            }
-            then(
-              tx.objectStore("preferences").get(key(cursor.value.id)),
-              (pref: StoredPreference | undefined) =>
-                reindex(tx, then, cursor.value, pref?.value, () =>
-                  cursor.continue(),
-                ),
-            );
-          });
-        });
-      },
+            });
+          },
+          lifetime.signal,
+        ),
       lifetime.signal,
     );
     // Keep an unopened handle's initialization rejection observable by its first operation.
@@ -291,6 +317,7 @@ export function createArtifactStore(viewer: string, relay: string) {
         stores,
         (tx, then, done) => {
           then(tx.objectStore("metadata").get(key("schema")), (schema) => {
+            validate?.();
             if (schema?.version !== definition.version)
               throw new Error("Artifact collection version changed");
             work(tx, then, done);

src/features/communities/legacy-communities.test.ts

@@ -37,7 +37,10 @@ it("imports once and only after consent", async () => {
     answer(true);
     await flush();
     expect(communities.snapshot().memberships).toEqual([
-      { id: "https://old.example", name: "old.example" },
+      expect.objectContaining({
+        id: "https://old.example",
+        name: "old.example",
+      }),
     ]);
     expect(communities.snapshot().sync.outbox).toEqual([
       { url: "wss://old.example", removed: false },