Fold a push's follow-up requests together (#393)

A pushed mail change took three round trips: Email/changes beside a
Mailbox/get, then Email/get for what changed, then the list, the open
thread and a second Mailbox/get. Each page of changes now carries its own
Email/get calls by back-reference, and the one Mailbox/get goes out with
it, so a push settles in two. New mail fetched this way is not asked for
again by the notice.

A reply's sessionState that differs from the session is announced once
rather than on every reply, and session refreshes in flight are shared.
The mock's session state now matches the sessionState on its replies, as
Stalwart's does; tying it to the data counter made every reply trigger a
session refresh in development.

server/src/mock/index.ts

@@ -7,6 +7,8 @@ import { createServer, type IncomingMessage, type ServerResponse } from "node:ht
 import { parseOtpauthUrl, verifyTotp } from "../totp.js";
 import { gzipSync } from "node:zlib";
 import { ACCOUNT, MAX_DELAYED_SEND, MOCK_EDITION, MOCK_LOCALE, NO_REGISTRY, Obj, PASS, PERMISSION_SNAPSHOT, PORT, SHARED_ACCOUNT, SHARED_CAPS, USER, account, nextState, state } from "./config.js";
+
+const SESSION_STATE = "1";
 import { PING_FLOOR_SECONDS, addEmail, blobs, calendars, people, principals, putBlob, recount } from "./data.js";
 import { MAX_OBJECTS, MethodError, directory, enforceLimits, resolveRefs } from "./engine.js";
 import { handlers } from "./handlers.js";
@@ -63,7 +65,12 @@ const session = () => ({
   downloadUrl: `http://127.0.0.1:${PORT}/jmap/download/{accountId}/{blobId}/{name}?accept={type}`,
   uploadUrl: `http://127.0.0.1:${PORT}/jmap/upload/{accountId}/`,
   eventSourceUrl: `http://127.0.0.1:${PORT}/jmap/eventsource/?types={types}&closeafter={closeafter}&ping={ping}`,
-  state: String(state.n),
+  /*
+   * The session's own state, which the account's data changes do not move.
+   * It matches the sessionState on every JMAP reply below, as Stalwart's does;
+   * tying it to the data counter made every reply look like a session change.
+   */
+  state: SESSION_STATE,
 });
 
 
@@ -125,7 +132,7 @@ export const server = createServer(async (req, res) => {
     }
     if (touched.size) { nextState(); setTimeout(() => broadcast([...touched, ...(touched.has("Email") ? ["Mailbox", "Thread"] : [])]), 50); }
     res.writeHead(200, { "content-type": "application/json" });
-    return res.end(JSON.stringify({ methodResponses: responses, sessionState: "1" }));
+    return res.end(JSON.stringify({ methodResponses: responses, sessionState: SESSION_STATE }));
   }
   if (url.pathname.startsWith("/jmap/upload/") && req.method === "POST") {
     const data = await readBody(req);

web/src/jmap/client.ts

@@ -104,6 +104,8 @@ export class JmapClient {
   private callCounter = 0;
   private unauthHandlers = new Set<() => void>();
   private stateHandlers = new Set<(sessionState: string) => void>();
+  /** The last session state announced, so a burst of replies announces it once. */
+  private announcedState: string | null = null;
 
   get maxCallsInRequest(): number {
     const core = this.session?.capabilities[CAP.core] as { maxCallsInRequest?: number } | undefined;
@@ -255,7 +257,8 @@ export class JmapClient {
     const body: Record<string, unknown> = { using: this.supportedUsing(using), methodCalls };
     if (createdIds) body.createdIds = createdIds;
     const res = await apiFetch<JmapResponse>("/api/jmap", { method: "POST", body: JSON.stringify(body) });
-    if (res.sessionState && this.session && res.sessionState !== this.session.state) {
+    if (res.sessionState && this.session && res.sessionState !== this.session.state && res.sessionState !== this.announcedState) {
+      this.announcedState = res.sessionState;
       for (const fn of this.stateHandlers) fn(res.sessionState);
     }
     return res;

web/src/store/__tests__/push-changes.test.ts

@@ -0,0 +1,101 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { CAP, client } from "@/jmap/client";
+import type { JmapSession } from "@/jmap/types";
+import { useMail } from "@/store/mail";
+
+vi.mock("@/lib/notify/notify", () => ({ playNewMailSound: vi.fn(), showNotification: vi.fn() }));
+
+/**
+ * What a push costs. On a slow link every round trip is felt, and a push
+ * arrives after almost everything the reader does -- marking one message read
+ * is echoed back as a change.
+ */
+
+const INBOX = "mbInbox";
+type Call = [string, Record<string, unknown>, string];
+
+const email = (id: string, keywords: Record<string, boolean> = {}) => ({ id, threadId: `t${id}`, mailboxIds: { [INBOX]: true }, keywords, receivedAt: "2026-09-16T00:00:00Z" });
+
+function server(changes: { created?: string[]; updated?: string[]; destroyed?: string[] } | "cannotCalculateChanges") {
+  const requests: Call[][] = [];
+  const fetchMock = vi.fn(async (_url: string, init: RequestInit) => {
+    const { methodCalls } = JSON.parse(init.body as string) as { methodCalls: Call[] };
+    requests.push(methodCalls);
+    const responses: Call[] = [];
+    for (const [name, args, id] of methodCalls) {
+      const ref = args["#ids"] as { resultOf: string; path: string } | undefined;
+      if (name === "Email/changes") {
+        if (changes === "cannotCalculateChanges") responses.push(["error", { type: "cannotCalculateChanges" }, id]);
+        else responses.push([name, { accountId: "a1", oldState: "s1", newState: "s2", hasMoreChanges: false, created: changes.created ?? [], updated: changes.updated ?? [], destroyed: changes.destroyed ?? [] }, id]);
+      } else if (name === "Email/get") {
+        const from = ref ? responses.find((r) => r[2] === ref.resultOf)?.[1] : undefined;
+        const ids = ref ? ((from?.[ref.path.slice(1)] as string[] | undefined) ?? []) : (args.ids as string[]);
+        responses.push([name, { accountId: "a1", state: "s2", list: ids.map((x) => email(x, { $seen: true })), notFound: [] }, id]);
+      } else if (name === "Mailbox/get") {
+        responses.push([name, { accountId: "a1", state: "m2", list: [{ id: INBOX, role: "inbox", name: "Inbox" }], notFound: [] }, id]);
+      } else {
+        responses.push([name, { accountId: "a1", state: "s2", list: [], ids: [], total: 0, notFound: [] }, id]);
+      }
+    }
+    return { ok: true, status: 200, json: async () => ({ methodResponses: responses, sessionState: "x" }) } as Response;
+  });
+  vi.stubGlobal("fetch", fetchMock);
+  return { requests, names: () => requests.map((r) => r.map(([n]) => n)) };
+}
+
+beforeEach(() => {
+  client.session = { capabilities: { [CAP.core]: { maxObjectsInGet: 500 }, [CAP.mail]: {} }, accounts: {}, primaryAccounts: {}, state: "x" } as unknown as JmapSession;
+  useMail.setState({
+    accountId: "a1",
+    mailboxes: { [INBOX]: { id: INBOX, role: "inbox", name: "Inbox" } } as never,
+    list: null,
+    emails: { e1: email("e1"), e2: email("e2") } as never,
+    fullIds: {},
+    threads: {},
+    emailState: "s1",
+    openThreadId: null,
+  });
+});
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+});
+
+describe("a pushed mail change", () => {
+  it("fetches the changes and what they name in one request, beside one Mailbox/get", async () => {
+    const { names } = server({ created: ["e9"], updated: ["e1", "e5"], destroyed: ["e2"] });
+    await useMail.getState().applyChanges(new Set(["Email", "Mailbox", "Thread"]));
+    await vi.waitFor(() => expect(useMail.getState().mailboxState).toBe("m2"));
+    const all = names();
+    expect(all.find((r) => r.includes("Email/changes"))).toEqual(["Email/changes", "Email/get", "Email/get"]);
+    expect(all.flat().filter((n) => n === "Mailbox/get")).toHaveLength(1);
+    // Nothing named by the changes was asked for again.
+    expect(all.flat().filter((n) => n === "Email/get")).toHaveLength(2);
+    const { emails, emailState } = useMail.getState();
+    expect(emailState).toBe("s2");
+    expect(emails.e1?.keywords).toEqual({ $seen: true });
+    expect(emails.e2).toBeUndefined();
+    expect(emails.e9).toBeDefined();
+    // An update to a message not held is not taken in.
+    expect(emails.e5).toBeUndefined();
+  });
+
+  it("forgets its state when the server cannot say what changed", async () => {
+    server("cannotCalculateChanges");
+    await useMail.getState().applyChanges(new Set(["Email"]));
+    expect(useMail.getState().emailState).toBeNull();
+  });
+});
+
+describe("a new session state", () => {
+  it("is announced once, however many replies carry it", async () => {
+    server({});
+    client.session = { ...client.session!, state: "old" };
+    const seen = vi.fn();
+    const off = client.onSessionState(seen);
+    await Promise.all([client.request([["Core/echo", {}, "a"]]), client.request([["Core/echo", {}, "b"]]), client.request([["Core/echo", {}, "c"]])]);
+    off();
+    expect(seen).toHaveBeenCalledTimes(1);
+    expect(seen).toHaveBeenCalledWith("x");
+  });
+});

web/src/store/mail/index.ts

@@ -957,7 +957,11 @@ export const useMail = create<MailState>((set, get) => ({
   async applyChanges(types) {
     const accountId = get().accountId;
     if (!accountId) return;
-    if (types.has("Mailbox")) void get().loadMailboxes();
+    /*
+     * Folder counts move with mail, so one Mailbox/get serves both kinds of
+     * change. It goes out now, beside Email/changes, rather than again after.
+     */
+    if (types.has("Mailbox") || types.has("Email")) void get().loadMailboxes();
     if (types.has("Email")) {
       const state = get().emailState;
       if (state) {
@@ -967,12 +971,25 @@ export const useMail = create<MailState>((set, get) => ({
           const updated = new Set<Id>();
           const created = new Set<Id>();
           const destroyed = new Set<Id>();
-          // Page through Email/changes.
+          const fetched: Email[] = [];
+          /*
+           * Page through Email/changes, each page in one request with the
+           * list-level properties of what it names. Asking for those after the
+           * ids came back cost a second round trip on every push.
+           */
+          const maxChanges = Math.min(500, client.maxObjectsInGet);
           while (guard++ < 10) {
-            const ch = await client.call<ChangesResponse>("Email/changes", { accountId, sinceState: since, maxChanges: 500 });
+            const ref = (path: string) => ({ resultOf: "c", name: "Email/changes", path });
+            const res = await client.chain([
+              ["Email/changes", { accountId, sinceState: since, maxChanges }, "c"],
+              ["Email/get", { accountId, "#ids": ref("/updated"), properties: LIST_PROPS }, "u"],
+              ["Email/get", { accountId, "#ids": ref("/created"), properties: LIST_PROPS }, "n"],
+            ]);
+            const ch = res.get("c")![0] as unknown as ChangesResponse;
             ch.created.forEach((id) => created.add(id));
             ch.updated.forEach((id) => updated.add(id));
             ch.destroyed.forEach((id) => destroyed.add(id));
+            for (const key of ["u", "n"]) fetched.push(...((res.get(key)?.[0] as unknown as GetResponse<Email> | undefined)?.list ?? []));
             since = ch.newState;
             if (!ch.hasMoreChanges) break;
           }
@@ -983,6 +1000,12 @@ export const useMail = create<MailState>((set, get) => ({
               delete next[id];
               delete nextFull[id];
             }
+            // An update merges over what is held; a message not held stays out,
+            // except new mail, which the notice below and the list both want.
+            for (const e of fetched) {
+              if (destroyed.has(e.id)) continue;
+              if (next[e.id] || created.has(e.id)) next[e.id] = mergeEmail(next[e.id], e);
+            }
             /*
              * The full copy of an updated email is deliberately kept.
              *
@@ -1006,18 +1029,6 @@ export const useMail = create<MailState>((set, get) => ({
              */
             return { emails: next, fullIds: nextFull, emailState: since };
           });
-          // Refresh the list-level props of updated/cached emails.
-          const cached = [...updated].filter((id) => get().emails[id]);
-          if (cached.length) {
-            const results = await Promise.all(
-              chunk(cached, client.maxObjectsInGet).map((part) => client.call<GetResponse<Email>>("Email/get", { accountId, ids: part, properties: LIST_PROPS })),
-            );
-            set((s) => {
-              const next = { ...s.emails };
-              for (const r of results) for (const e of r.list) next[e.id] = mergeEmail(next[e.id], e);
-              return { emails: next };
-            });
-          }
           if (created.size) await notifyNewMail([...created], get);
         } catch (err) {
           if (err instanceof JmapMethodError && err.type === "cannotCalculateChanges") {
@@ -1026,7 +1037,6 @@ export const useMail = create<MailState>((set, get) => ({
         }
       }
       void get().refreshList();
-      void get().loadMailboxes();
     }
     if (types.has("Thread") || types.has("Email")) {
       const open = get().openThreadId;

web/src/store/session.ts

@@ -32,6 +32,8 @@ interface SessionState {
   ownAccountFor(cap: string): Id | null;
 }
 
+let refreshing: Promise<void> | null = null;
+
 export const useSession = create<SessionState>((set, get) => ({
   status: "loading",
   session: null,
@@ -100,15 +102,21 @@ export const useSession = create<SessionState>((set, get) => ({
     set({ status: "anonymous", session: null, accountId: null });
   },
 
-  async refresh() {
+  refresh() {
-    try {
+    // Callers arriving while a refresh is on its way share it.
-      const s = await apiFetch<JmapSession>("/api/auth/session?refresh=1");
+    refreshing ??= (async () => {
-      client.session = s;
+      try {
-      setServerLocale(s.ihasmail?.userLocale);
+        const s = await apiFetch<JmapSession>("/api/auth/session?refresh=1");
-      set({ session: s });
+        client.session = s;
-    } catch {
+        setServerLocale(s.ihasmail?.userLocale);
-      /* ignore */
+        set({ session: s });
-    }
+      } catch {
+        /* ignore */
+      } finally {
+        refreshing = null;
+      }
+    })();
+    return refreshing;
   },
 
   setAccount(id) {