diff --git a/spec/connectors.md b/spec/connectors.md index b6f7e25..af1cad7 100644 --- a/spec/connectors.md +++ b/spec/connectors.md @@ -51,9 +51,9 @@ The receiver handles provider CRC `GET` requests and signature-verified event `P The live `fastmail-jmap` source is one serial polling producer for one exact source id. It resolves Fastmail's primary mail account through the pinned `api.fastmail.com` session endpoint and accepts API URLs only on Fastmail's unported `api.fastmail.com`, `*.api.fastmail.com`, or `jmap.fastmail.com` HTTPS hosts. It sends only `Email/get`, `Email/query`, `Email/changes`, and `Email/queryChanges` method calls. Its environment compartment contains only the token named by the manifest. The source cannot invoke a model, send mail, create drafts, mutate keywords or mailboxes, or reach Telegram. -First activation uses `replay: now`: the producer reads current email and query state, appends cursor and lifecycle evidence, and emits no historical email observations. Later polls collect bounded change pages from both JMAP state machines, fetch metadata for changed messages in bounded chunks, collapse repeated ids to one final operation, and append observations plus both new states in one transaction. A crash before settlement replays the same source state; deterministic event identity absorbs duplicates. Poll cycles never overlap. +First activation uses `replay: now`: the producer reads current email and query state, appends cursor and lifecycle evidence, and emits no historical email observations. Later polls collect bounded `Email/changes` pages and one bounded `Email/queryChanges` response, fetch metadata for changed messages in bounded chunks, collapse repeated ids to one final operation, and append observations plus both new states in one transaction. A crash before settlement replays the same source state; deterministic event identity absorbs duplicates. Poll cycles never overlap. -If either JMAP changes method returns `cannotCalculateChanges`, the producer performs one bounded current-mailbox resnapshot, emits at most the configured number of metadata-only `updated` observations, and advances both states in the same producer transaction. Other JMAP method errors fail the cycle without advancing either cursor. Response bytes, pages, changed ids, resnapshot ids, request time, and poll interval are bounded before activation. +If either JMAP changes method returns `cannotCalculateChanges`, or `Email/queryChanges` returns `tooManyChanges`, the producer performs one bounded current-mailbox resnapshot, emits at most the configured number of metadata-only `updated` observations, and advances both states in the same producer transaction. Other JMAP method errors fail the cycle without advancing either cursor. Response bytes, email-change pages, changed ids, resnapshot ids, request time, and poll interval are bounded before activation. Continuous polling backs off exponentially after consecutive failures, capped at 15 minutes, and returns to the configured interval after a successful cycle. Query membership deltas are validated only to advance the unfiltered query state; `Email/changes` owns operations for this all-mail source. Adding a mailbox or keyword filter requires merging query additions and removals into the event operations. diff --git a/src/connectors/fastmail-jmap.ts b/src/connectors/fastmail-jmap.ts index 885e1f0..15d0cdc 100644 --- a/src/connectors/fastmail-jmap.ts +++ b/src/connectors/fastmail-jmap.ts @@ -63,7 +63,6 @@ const emailQueryChangesSchema = z.object({ accountId: z.string().min(1), oldQueryState: z.string().min(1), newQueryState: z.string().min(1), - hasMoreChanges: z.boolean(), removed: z.array(z.string().min(1)), added: z.array(z.object({ id: z.string().min(1), @@ -271,7 +270,7 @@ export class FastmailJmapClient { const pageSize = boundedPositive(maxChanges, 1_000, "Fastmail max changes"); const pageLimit = boundedPositive(maxPages, 100, "Fastmail max pages"); const email = await this.collectEmailChanges(session, emailState, pageSize, pageLimit, signal); - const query = await this.collectQueryChanges(session, queryState, pageSize, pageLimit, signal); + const query = await this.collectQueryChanges(session, queryState, pageSize, signal); const operations = collapseOperations(email.pages); const changedIds = [...operations.entries()].filter(([, operation]) => operation !== "destroyed").map(([id]) => id).sort(); const details = await this.getEmails(session, changedIds, pageSize, signal); @@ -326,35 +325,28 @@ export class FastmailJmapClient { session: FastmailSession, initialState: string, maxChanges: number, - maxPages: number, signal?: AbortSignal, - ): Promise<{ state: string; pages: EmailQueryChanges[] }> { - let state = required(initialState, "Fastmail query state"); - const pages: EmailQueryChanges[] = []; - for (let page = 0; page < maxPages; page += 1) { - const id = `query-changes-${page}`; - const response = await this.request(session, [{ - name: "Email/queryChanges", - id, - arguments: { - accountId: session.accountId, - sinceQueryState: state, - maxChanges, - calculateTotal: false, - sort: [{ property: "receivedAt", isAscending: false }], - }, - }], signal); - const changes = methodResult(response, id, "Email/queryChanges", emailQueryChangesSchema); - assertAccount(session.accountId, changes.accountId); - if (changes.oldQueryState !== state) throw new FastmailJmapError("query-state-gap", "Fastmail Email/queryChanges returned a different old state"); - if (changes.removed.length + changes.added.length > maxChanges) { - throw new FastmailJmapError("query-change-bound", "Fastmail Email/queryChanges exceeded the requested change bound"); - } - pages.push(changes); - state = changes.newQueryState; - if (!changes.hasMoreChanges) return { state, pages }; + ): Promise<{ state: string; response: EmailQueryChanges }> { + const state = required(initialState, "Fastmail query state"); + const id = "query-changes"; + const response = await this.request(session, [{ + name: "Email/queryChanges", + id, + arguments: { + accountId: session.accountId, + sinceQueryState: state, + maxChanges, + calculateTotal: false, + sort: [{ property: "receivedAt", isAscending: false }], + }, + }], signal); + const changes = methodResult(response, id, "Email/queryChanges", emailQueryChangesSchema); + assertAccount(session.accountId, changes.accountId); + if (changes.oldQueryState !== state) throw new FastmailJmapError("query-state-gap", "Fastmail Email/queryChanges returned a different old state"); + if (changes.removed.length + changes.added.length > maxChanges) { + throw new FastmailJmapError("query-change-bound", "Fastmail Email/queryChanges exceeded the requested change bound"); } - throw new FastmailJmapError("query-page-bound", "Fastmail Email/queryChanges exceeded the configured page bound"); + return { state: changes.newQueryState, response: changes }; } private async getEmails( @@ -601,7 +593,7 @@ function methodResult(response: JmapResponse, id: string, name: string, schem const [actualName, payload] = matches[0]!; if (actualName === "error") { const type = typeof payload.type === "string" ? payload.type : "unknown"; - if (type === "cannotCalculateChanges") throw new FastmailJmapCannotCalculateChanges(); + if (type === "cannotCalculateChanges" || type === "tooManyChanges") throw new FastmailJmapCannotCalculateChanges(); throw new FastmailJmapError(`method-${safeCode(type)}`, `Fastmail ${name} returned ${safeCode(type)}`); } if (actualName !== name) throw new FastmailJmapError("method-name-mismatch", `Fastmail returned ${actualName} for ${id}`); diff --git a/test/fastmail-jmap.test.ts b/test/fastmail-jmap.test.ts index 53f68c8..ff23b5d 100644 --- a/test/fastmail-jmap.test.ts +++ b/test/fastmail-jmap.test.ts @@ -136,7 +136,7 @@ describe("live Fastmail JMAP polling", () => { updated: [], destroyed: [], }); - fixture.jmap.enqueueError("query-changes-0", "rateLimit"); + fixture.jmap.enqueueError("query-changes", "rateLimit"); await expect(fixture.connector.poll(fixture.store)).rejects.toMatchObject({ code: "method-ratelimit" }); @@ -204,7 +204,7 @@ describe("live Fastmail JMAP polling", () => { }); }); - test("repairs cannotCalculateChanges from query state without retaining the partial email state", async () => { + test("repairs query tooManyChanges without retaining the partial email state", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); @@ -216,7 +216,7 @@ describe("live Fastmail JMAP polling", () => { updated: [], destroyed: [], }); - fixture.jmap.enqueueError("query-changes-0", "cannotCalculateChanges"); + fixture.jmap.enqueueError("query-changes", "tooManyChanges"); fixture.jmap.enqueueSnapshot("email-9", "query-9", [syntheticEmail("resnapshot")]); const poll = await fixture.connector.poll(fixture.store); @@ -230,7 +230,7 @@ describe("live Fastmail JMAP polling", () => { .some((event) => event.externalId === "discarded-partial")).toBe(false); }); - test("paginates query state, chunks metadata gets to the server capability, and converts notFound to destroyed", async () => { + test("validates query state, chunks metadata gets to the server capability, and converts notFound to destroyed", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); @@ -250,8 +250,7 @@ describe("live Fastmail JMAP polling", () => { updated: ["missing"], destroyed: [], }); - fixture.jmap.enqueueQueryChanges("query-1", "query-1b", true); - fixture.jmap.enqueueQueryChanges("query-1b", "query-2", false); + fixture.jmap.enqueueQueryChanges("query-1", "query-2"); fixture.jmap.enqueueGet("email-2", [syntheticEmail("one"), syntheticEmail("two")]); fixture.jmap.enqueueGet("email-2", [], ["missing"]); @@ -262,7 +261,6 @@ describe("live Fastmail JMAP polling", () => { ["Email/changes"], ["Email/changes"], ["Email/queryChanges"], - ["Email/queryChanges"], ["Email/get"], ["Email/get"], ]); @@ -381,16 +379,14 @@ class SyntheticJmap { }, `email-changes-${this.emailChangesQueued()}`]])); } - enqueueQueryChanges(oldQueryState: string, newQueryState: string, hasMoreChanges = false) { - const id = `query-changes-${this.queryChangesQueued()}`; + enqueueQueryChanges(oldQueryState: string, newQueryState: string) { this.responses.push(jmapResponse([["Email/queryChanges", { accountId: "account-fixture", oldQueryState, newQueryState, - hasMoreChanges, removed: [], added: [], - }, id]])); + }, "query-changes"]])); } enqueueGet(state: string, emails: unknown[], notFound: string[] = []) { @@ -411,10 +407,6 @@ class SyntheticJmap { return this.responses.filter((response) => JSON.stringify(response).includes("Email/changes")).length; } - private queryChangesQueued(): number { - return this.responses.filter((response) => JSON.stringify(response).includes("Email/queryChanges")).length; - } - private emailGetQueued(): number { return this.responses.filter((response) => JSON.stringify(response).includes("email-get-")).length; }