fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
KrisBraun merged 1 commit into from
Jun 22, 2026
Merged

fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade #217

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 89 additions & 0 deletions connectors/gmail/src/gmail.test.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
});
});
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
function setup(entries: Array<[string, unknown]>) {
const storeMap = new Map<string, unknown>(entries);
const store = {
get: vi.fn(async (k: string) =>
storeMap.has(k) ? storeMap.get(k) : null
),
set: vi.fn(async (k: string, v: unknown) => {
storeMap.set(k, v);
}),
clear: vi.fn(async (k: string) => {
storeMap.delete(k);
}),
list: vi.fn(async (p: string) =>
[...storeMap.keys()].filter((k) => k.startsWith(p))
),
};
const tools = { store, integrations: {}, network: {}, files: {} };
const gmail = new Gmail(
"twist-instance-1" as never,
{ getTools: () => tools } as never
) as any;
const spies = {
setupMailboxWebhook: vi
.spyOn(gmail, "setupMailboxWebhook")
.mockResolvedValue(undefined),
scheduleSelfHealCheck: vi
.spyOn(gmail, "scheduleSelfHealCheck")
.mockResolvedValue(undefined),
scheduleMailboxRenewal: vi
.spyOn(gmail, "scheduleMailboxRenewal")
.mockResolvedValue(undefined),
requeueInitialSync: vi
.spyOn(gmail, "requeueInitialSync")
.mockResolvedValue(undefined),
};
return { gmail, spies };
}

const webhook = (expiration: Date) => ({
topicName: "topic",
historyId: "1",
expiration,
created: "2026-01-01T00:00:00.000Z",
});
const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
await gmail.recoverMailboxDelivery();
expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
});

it("expired watch: treated as stranded — re-establishes and backfills", async () => {
const { gmail, spies } = setup([
["enabled_channels", ["INBOX"]],
["mailbox_webhook", webhook(PAST)],
]);
await gmail.recoverMailboxDelivery();
expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
});

it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
const { gmail, spies } = setup([
["enabled_channels", ["INBOX"]],
["mailbox_webhook", webhook(FUTURE)],
]);
await gmail.recoverMailboxDelivery();
expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
expect(spies.requeueInitialSync).not.toHaveBeenCalled();
});

it("no enabled channels: does nothing", async () => {
const { gmail, spies } = setup([["enabled_channels", []]]);
await gmail.recoverMailboxDelivery();
expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
expect(spies.requeueInitialSync).not.toHaveBeenCalled();
});
});
110 changes: 84 additions & 26 deletions connectors/gmail/src/gmail.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
await this.set("auth_actor_id", context.actor.id);
}

override async upgrade(): Promise<void> {
// `mailbox_webhook` is the canonical "already on the mailbox-wide watch
// model" sentinel. Instances that predate that change still carry old
// per-channel keys and need a one-time migration before recovery runs.
const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
if (!migrated) {
await this.migrateLegacyPerChannelState();
}

// Durable recovery backstop, run on every deploy. Re-asserts recurring
// maintenance for a healthy mailbox and re-establishes (plus backfills) a
// stranded one. See recoverMailboxDelivery for the stranded cases.
await this.recoverMailboxDelivery();
}

/**
* Migration from per-channel watches to a single mailbox-wide watch.
* Ensure live push delivery + recurring maintenance for any instance with
* enabled channels. Runs from upgrade() on every deploy.
*
* Old layout (per-channel):
* - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
* `sync_enabled_${id}`
* New layout (per-twist-instance):
* - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
* `initial_state_${id}` for in-flight backfills; recurring tasks are
* keyed "mailbox-watch-renewal" and "mailbox-self-heal"
* Two stranded states were previously unrecoverable — neither the old
* `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
* sweep could heal them, so the connection stayed silently dead until the
* user manually re-enabled a channel:
*
* The Twist runtime API doesn't let connectors enumerate stored keys, so we
* probe the system labels we know users can enable. Users with custom
* Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
* this upgrade — without `list()` we can't discover their IDs.
* 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
* before its `set()` (e.g. the Gmail "Only one user push notification
* client allowed per developer" 400). With no sentinel there was
* nothing to re-assert, and because `scheduleRecurring` never ran the
* maintenance sweep's `ever` marker was never set either; and
* 2. the watch expired while the self-heal/renewal chain was dead.
*
* A healthy watch (present and unexpired) only re-asserts the recurring
* tasks. A missing or expired watch is re-established AND every enabled label
* is re-walked, so mail that accumulated while delivery was dead is
* backfilled. The backfill upserts by `source` (no duplicates) and uses
* initial-sync semantics (read/unarchived), so it never spams notifications.
*/
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
if (already) {
// Re-assert durable recurring maintenance for any instance that had an
// active mailbox watch but whose pre-recurring self-heal/renewal chain
// died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
// Healthy watch — re-assert durable maintenance (idempotent, keyed).
try {
await this.scheduleSelfHealCheck();
await this.scheduleMailboxRenewal(new Date(already.expiration));
await this.scheduleMailboxRenewal(new Date(webhook.expiration));
} catch (error) {
console.error(
`Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
return;
}

// Stranded: watch missing or expired with nothing live renewing it.
try {
for (const channelId of enabled) {
await this.requeueInitialSync(channelId);
}
await this.setupMailboxWebhook();
} catch (error) {
console.error(
`Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
error
);
}
}

/**
* Re-queue a fresh full backfill of one label, dropping any stale cursor so
* the walk restarts from the newest thread. Used by recovery to re-import
* mail that arrived while push delivery was dead.
*/
private async requeueInitialSync(channelId: string): Promise<void> {
await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
const initial = await this.callback(this.initialSyncBatch, channelId, 1);
await this.runTask(initial);
}

/**
* One-time migration from per-channel watches to a single mailbox-wide watch.
*
* Old layout (per-channel):
* - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
* `sync_enabled_${id}`
* New layout (per-twist-instance):
* - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
* `initial_state_${id}` for in-flight backfills; recurring tasks are
* keyed "mailbox-watch-renewal" and "mailbox-self-heal"
*
* The Twist runtime API doesn't let connectors enumerate stored keys, so we
* probe the system labels we know users can enable. Users with custom
* Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
* this upgrade — without `list()` we can't discover their IDs. The mailbox
* watch itself is (re-)established afterward by recoverMailboxDelivery().
*/
private async migrateLegacyPerChannelState(): Promise<void> {
// Stop whatever per-channel watch was last active. Gmail allows only
// one watch per mailbox, so a single stopWatch() call covers all of
// them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
}

// Probe known system labels for old per-channel state and migrate.
let migratedAny = false;
for (const labelId of SYSTEM_LABEL_ORDER) {
const oldWebhook = await this.get<{ topicName?: string }>(
`channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
// Channel was enabled (presence of any old per-channel key counts).
if (oldEnabled || oldWebhook) {
await this.addEnabledChannel(labelId);
migratedAny = true;
}

// Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
await this.clear(`sync_state_${labelId}`);
await this.clear(`sync_enabled_${labelId}`);
}

if (migratedAny) {
await this.setupMailboxWebhook();
}
}

private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down
98 changes: 96 additions & 2 deletions connectors/outlook-mail/src/outlook-mail.test.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
OutlookMail,
pickChannelForConversation,
recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
).toEqual(["a@b.com"]);
});
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
function setup(entries: Array<[string, unknown]>) {
const storeMap = new Map<string, unknown>(entries);
const store = {
get: vi.fn(async (k: string) =>
storeMap.has(k) ? storeMap.get(k) : null
),
set: vi.fn(async (k: string, v: unknown) => {
storeMap.set(k, v);
}),
clear: vi.fn(async (k: string) => {
storeMap.delete(k);
}),
list: vi.fn(async (p: string) =>
[...storeMap.keys()].filter((k) => k.startsWith(p))
),
};
const tools = { store, integrations: {}, network: {}, files: {} };
const outlook = new OutlookMail(
"twist-instance-1" as never,
{ getTools: () => tools } as never
) as any;
const spies = {
setupMailboxSubscription: vi
.spyOn(outlook, "setupMailboxSubscription")
.mockResolvedValue(undefined),
scheduleSelfHealCheck: vi
.spyOn(outlook, "scheduleSelfHealCheck")
.mockResolvedValue(undefined),
scheduleMailboxRenewal: vi
.spyOn(outlook, "scheduleMailboxRenewal")
.mockResolvedValue(undefined),
requeueInitialSync: vi
.spyOn(outlook, "requeueInitialSync")
.mockResolvedValue(undefined),
};
return { outlook, spies };
}

const subscription = (expiration: Date) => ({
subscriptionId: "sub-1",
clientState: "secret",
webhookUrl: "https://example.com/hook",
expiration,
created: "2026-01-01T00:00:00.000Z",
});
const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
const PAST = new Date(Date.now() - 60 * 60 * 1000);

it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
await outlook.recoverMailboxDelivery();
expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
});

it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
const { outlook, spies } = setup([
["enabled_channels", ["f-inbox"]],
["mailbox_subscription", subscription(PAST)],
]);
await outlook.recoverMailboxDelivery();
expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
});

it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
const { outlook, spies } = setup([
["enabled_channels", ["f-inbox"]],
["mailbox_subscription", subscription(FUTURE)],
]);
await outlook.recoverMailboxDelivery();
expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
expect(spies.requeueInitialSync).not.toHaveBeenCalled();
});

it("no enabled channels: does nothing", async () => {
const { outlook, spies } = setup([["enabled_channels", []]]);
await outlook.recoverMailboxDelivery();
expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
expect(spies.requeueInitialSync).not.toHaveBeenCalled();
});
});
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

KrisBraun
merged 1 commit into

from

Jun 22, 2026
 Merged


fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade
#217
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


89 changes: 89 additions & 0 deletions
connectors/gmail/src/gmail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
 });
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const gmail = new Gmail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxWebhook: vi
 .spyOn(gmail, "setupMailboxWebhook")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(gmail, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(gmail, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(gmail, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { gmail, spies };
 }

 const webhook = (expiration: Date) => ({
 topicName: "topic",
 historyId: "1",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

 it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
 const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired watch: treated as stranded — re-establishes and backfills", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(PAST)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(FUTURE)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { gmail, spies } = setup([["enabled_channels", []]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});

110 changes: 84 additions & 26 deletions
connectors/gmail/src/gmail.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
 await this.set("auth_actor_id", context.actor.id);
 }

 override async upgrade(): Promise<void> {
 // `mailbox_webhook` is the canonical "already on the mailbox-wide watch
 // model" sentinel. Instances that predate that change still carry old
 // per-channel keys and need a one-time migration before recovery runs.
 const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (!migrated) {
 await this.migrateLegacyPerChannelState();
 }

 // Durable recovery backstop, run on every deploy. Re-asserts recurring
 // maintenance for a healthy mailbox and re-establishes (plus backfills) a
 // stranded one. See recoverMailboxDelivery for the stranded cases.
 await this.recoverMailboxDelivery();
 }

 /**
 * Migration from per-channel watches to a single mailbox-wide watch.
 * Ensure live push delivery + recurring maintenance for any instance with
 * enabled channels. Runs from upgrade() on every deploy.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 * Two stranded states were previously unrecoverable — neither the old
 * `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
 * sweep could heal them, so the connection stayed silently dead until the
 * user manually re-enabled a channel:
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs.
 * 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
 * before its `set()` (e.g. the Gmail "Only one user push notification
 * client allowed per developer" 400). With no sentinel there was
 * nothing to re-assert, and because `scheduleRecurring` never ran the
 * maintenance sweep's `ever` marker was never set either; and
 * 2. the watch expired while the self-heal/renewal chain was dead.
 *
 * A healthy watch (present and unexpired) only re-asserts the recurring
 * tasks. A missing or expired watch is re-established AND every enabled label
 * is re-walked, so mail that accumulated while delivery was dead is
 * backfilled. The backfill upserts by `source` (no duplicates) and uses
 * initial-sync semantics (read/unarchived), so it never spams notifications.
 */
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (already) {
 // Re-assert durable recurring maintenance for any instance that had an
 // active mailbox watch but whose pre-recurring self-heal/renewal chain
 // died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
 // Healthy watch — re-assert durable maintenance (idempotent, keyed).
 try {
 await this.scheduleSelfHealCheck();
 await this.scheduleMailboxRenewal(new Date(already.expiration));
 await this.scheduleMailboxRenewal(new Date(webhook.expiration));
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
 return;
 }

 // Stranded: watch missing or expired with nothing live renewing it.
 try {
 for (const channelId of enabled) {
 await this.requeueInitialSync(channelId);
 }
 await this.setupMailboxWebhook();
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
 error
 );
 }
 }

 /**
 * Re-queue a fresh full backfill of one label, dropping any stale cursor so
 * the walk restarts from the newest thread. Used by recovery to re-import
 * mail that arrived while push delivery was dead.
 */
 private async requeueInitialSync(channelId: string): Promise<void> {
 await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
 const initial = await this.callback(this.initialSyncBatch, channelId, 1);
 await this.runTask(initial);
 }

 /**
 * One-time migration from per-channel watches to a single mailbox-wide watch.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs. The mailbox
 * watch itself is (re-)established afterward by recoverMailboxDelivery().
 */
 private async migrateLegacyPerChannelState(): Promise<void> {
 // Stop whatever per-channel watch was last active. Gmail allows only
 // one watch per mailbox, so a single stopWatch() call covers all of
 // them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
 }

 // Probe known system labels for old per-channel state and migrate.
 let migratedAny = false;
 for (const labelId of SYSTEM_LABEL_ORDER) {
 const oldWebhook = await this.get<{ topicName?: string }>(
 `channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
 // Channel was enabled (presence of any old per-channel key counts).
 if (oldEnabled || oldWebhook) {
 await this.addEnabledChannel(labelId);
 migratedAny = true;
 }

 // Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
 await this.clear(`sync_state_${labelId}`);
 await this.clear(`sync_enabled_${labelId}`);
 }

 if (migratedAny) {
 await this.setupMailboxWebhook();
 }
 }

 private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down

98 changes: 96 additions & 2 deletions
connectors/outlook-mail/src/outlook-mail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
 OutlookMail,
 pickChannelForConversation,
 recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
 ).toEqual(["a@b.com"]);
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const outlook = new OutlookMail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxSubscription: vi
 .spyOn(outlook, "setupMailboxSubscription")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(outlook, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(outlook, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(outlook, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { outlook, spies };
 }

 const subscription = (expiration: Date) => ({
 subscriptionId: "sub-1",
 clientState: "secret",
 webhookUrl: "https://example.com/hook",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 60 * 60 * 1000);

 it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
 const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(PAST)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(FUTURE)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { outlook, spies } = setup([["enabled_channels", []]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

KrisBraun
merged 1 commit into

from

Jun 22, 2026
 Merged


fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade
#217
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


89 changes: 89 additions & 0 deletions
connectors/gmail/src/gmail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
 });
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const gmail = new Gmail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxWebhook: vi
 .spyOn(gmail, "setupMailboxWebhook")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(gmail, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(gmail, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(gmail, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { gmail, spies };
 }

 const webhook = (expiration: Date) => ({
 topicName: "topic",
 historyId: "1",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

 it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
 const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired watch: treated as stranded — re-establishes and backfills", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(PAST)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(FUTURE)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { gmail, spies } = setup([["enabled_channels", []]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});

110 changes: 84 additions & 26 deletions
connectors/gmail/src/gmail.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
 await this.set("auth_actor_id", context.actor.id);
 }

 override async upgrade(): Promise<void> {
 // `mailbox_webhook` is the canonical "already on the mailbox-wide watch
 // model" sentinel. Instances that predate that change still carry old
 // per-channel keys and need a one-time migration before recovery runs.
 const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (!migrated) {
 await this.migrateLegacyPerChannelState();
 }

 // Durable recovery backstop, run on every deploy. Re-asserts recurring
 // maintenance for a healthy mailbox and re-establishes (plus backfills) a
 // stranded one. See recoverMailboxDelivery for the stranded cases.
 await this.recoverMailboxDelivery();
 }

 /**
 * Migration from per-channel watches to a single mailbox-wide watch.
 * Ensure live push delivery + recurring maintenance for any instance with
 * enabled channels. Runs from upgrade() on every deploy.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 * Two stranded states were previously unrecoverable — neither the old
 * `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
 * sweep could heal them, so the connection stayed silently dead until the
 * user manually re-enabled a channel:
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs.
 * 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
 * before its `set()` (e.g. the Gmail "Only one user push notification
 * client allowed per developer" 400). With no sentinel there was
 * nothing to re-assert, and because `scheduleRecurring` never ran the
 * maintenance sweep's `ever` marker was never set either; and
 * 2. the watch expired while the self-heal/renewal chain was dead.
 *
 * A healthy watch (present and unexpired) only re-asserts the recurring
 * tasks. A missing or expired watch is re-established AND every enabled label
 * is re-walked, so mail that accumulated while delivery was dead is
 * backfilled. The backfill upserts by `source` (no duplicates) and uses
 * initial-sync semantics (read/unarchived), so it never spams notifications.
 */
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (already) {
 // Re-assert durable recurring maintenance for any instance that had an
 // active mailbox watch but whose pre-recurring self-heal/renewal chain
 // died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
 // Healthy watch — re-assert durable maintenance (idempotent, keyed).
 try {
 await this.scheduleSelfHealCheck();
 await this.scheduleMailboxRenewal(new Date(already.expiration));
 await this.scheduleMailboxRenewal(new Date(webhook.expiration));
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
 return;
 }

 // Stranded: watch missing or expired with nothing live renewing it.
 try {
 for (const channelId of enabled) {
 await this.requeueInitialSync(channelId);
 }
 await this.setupMailboxWebhook();
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
 error
 );
 }
 }

 /**
 * Re-queue a fresh full backfill of one label, dropping any stale cursor so
 * the walk restarts from the newest thread. Used by recovery to re-import
 * mail that arrived while push delivery was dead.
 */
 private async requeueInitialSync(channelId: string): Promise<void> {
 await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
 const initial = await this.callback(this.initialSyncBatch, channelId, 1);
 await this.runTask(initial);
 }

 /**
 * One-time migration from per-channel watches to a single mailbox-wide watch.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs. The mailbox
 * watch itself is (re-)established afterward by recoverMailboxDelivery().
 */
 private async migrateLegacyPerChannelState(): Promise<void> {
 // Stop whatever per-channel watch was last active. Gmail allows only
 // one watch per mailbox, so a single stopWatch() call covers all of
 // them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
 }

 // Probe known system labels for old per-channel state and migrate.
 let migratedAny = false;
 for (const labelId of SYSTEM_LABEL_ORDER) {
 const oldWebhook = await this.get<{ topicName?: string }>(
 `channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
 // Channel was enabled (presence of any old per-channel key counts).
 if (oldEnabled || oldWebhook) {
 await this.addEnabledChannel(labelId);
 migratedAny = true;
 }

 // Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
 await this.clear(`sync_state_${labelId}`);
 await this.clear(`sync_enabled_${labelId}`);
 }

 if (migratedAny) {
 await this.setupMailboxWebhook();
 }
 }

 private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down

98 changes: 96 additions & 2 deletions
connectors/outlook-mail/src/outlook-mail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
 OutlookMail,
 pickChannelForConversation,
 recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
 ).toEqual(["a@b.com"]);
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const outlook = new OutlookMail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxSubscription: vi
 .spyOn(outlook, "setupMailboxSubscription")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(outlook, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(outlook, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(outlook, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { outlook, spies };
 }

 const subscription = (expiration: Date) => ({
 subscriptionId: "sub-1",
 clientState: "secret",
 webhookUrl: "https://example.com/hook",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 60 * 60 * 1000);

 it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
 const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(PAST)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(FUTURE)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { outlook, spies } = setup([["enabled_channels", []]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

KrisBraun
merged 1 commit into

from

Jun 22, 2026
 Merged


fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade
#217
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


89 changes: 89 additions & 0 deletions
connectors/gmail/src/gmail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
 });
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const gmail = new Gmail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxWebhook: vi
 .spyOn(gmail, "setupMailboxWebhook")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(gmail, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(gmail, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(gmail, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { gmail, spies };
 }

 const webhook = (expiration: Date) => ({
 topicName: "topic",
 historyId: "1",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

 it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
 const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired watch: treated as stranded — re-establishes and backfills", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(PAST)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(FUTURE)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { gmail, spies } = setup([["enabled_channels", []]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});

110 changes: 84 additions & 26 deletions
connectors/gmail/src/gmail.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
 await this.set("auth_actor_id", context.actor.id);
 }

 override async upgrade(): Promise<void> {
 // `mailbox_webhook` is the canonical "already on the mailbox-wide watch
 // model" sentinel. Instances that predate that change still carry old
 // per-channel keys and need a one-time migration before recovery runs.
 const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (!migrated) {
 await this.migrateLegacyPerChannelState();
 }

 // Durable recovery backstop, run on every deploy. Re-asserts recurring
 // maintenance for a healthy mailbox and re-establishes (plus backfills) a
 // stranded one. See recoverMailboxDelivery for the stranded cases.
 await this.recoverMailboxDelivery();
 }

 /**
 * Migration from per-channel watches to a single mailbox-wide watch.
 * Ensure live push delivery + recurring maintenance for any instance with
 * enabled channels. Runs from upgrade() on every deploy.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 * Two stranded states were previously unrecoverable — neither the old
 * `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
 * sweep could heal them, so the connection stayed silently dead until the
 * user manually re-enabled a channel:
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs.
 * 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
 * before its `set()` (e.g. the Gmail "Only one user push notification
 * client allowed per developer" 400). With no sentinel there was
 * nothing to re-assert, and because `scheduleRecurring` never ran the
 * maintenance sweep's `ever` marker was never set either; and
 * 2. the watch expired while the self-heal/renewal chain was dead.
 *
 * A healthy watch (present and unexpired) only re-asserts the recurring
 * tasks. A missing or expired watch is re-established AND every enabled label
 * is re-walked, so mail that accumulated while delivery was dead is
 * backfilled. The backfill upserts by `source` (no duplicates) and uses
 * initial-sync semantics (read/unarchived), so it never spams notifications.
 */
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (already) {
 // Re-assert durable recurring maintenance for any instance that had an
 // active mailbox watch but whose pre-recurring self-heal/renewal chain
 // died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
 // Healthy watch — re-assert durable maintenance (idempotent, keyed).
 try {
 await this.scheduleSelfHealCheck();
 await this.scheduleMailboxRenewal(new Date(already.expiration));
 await this.scheduleMailboxRenewal(new Date(webhook.expiration));
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
 return;
 }

 // Stranded: watch missing or expired with nothing live renewing it.
 try {
 for (const channelId of enabled) {
 await this.requeueInitialSync(channelId);
 }
 await this.setupMailboxWebhook();
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
 error
 );
 }
 }

 /**
 * Re-queue a fresh full backfill of one label, dropping any stale cursor so
 * the walk restarts from the newest thread. Used by recovery to re-import
 * mail that arrived while push delivery was dead.
 */
 private async requeueInitialSync(channelId: string): Promise<void> {
 await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
 const initial = await this.callback(this.initialSyncBatch, channelId, 1);
 await this.runTask(initial);
 }

 /**
 * One-time migration from per-channel watches to a single mailbox-wide watch.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs. The mailbox
 * watch itself is (re-)established afterward by recoverMailboxDelivery().
 */
 private async migrateLegacyPerChannelState(): Promise<void> {
 // Stop whatever per-channel watch was last active. Gmail allows only
 // one watch per mailbox, so a single stopWatch() call covers all of
 // them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
 }

 // Probe known system labels for old per-channel state and migrate.
 let migratedAny = false;
 for (const labelId of SYSTEM_LABEL_ORDER) {
 const oldWebhook = await this.get<{ topicName?: string }>(
 `channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
 // Channel was enabled (presence of any old per-channel key counts).
 if (oldEnabled || oldWebhook) {
 await this.addEnabledChannel(labelId);
 migratedAny = true;
 }

 // Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
 await this.clear(`sync_state_${labelId}`);
 await this.clear(`sync_enabled_${labelId}`);
 }

 if (migratedAny) {
 await this.setupMailboxWebhook();
 }
 }

 private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down

98 changes: 96 additions & 2 deletions
connectors/outlook-mail/src/outlook-mail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
 OutlookMail,
 pickChannelForConversation,
 recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
 ).toEqual(["a@b.com"]);
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const outlook = new OutlookMail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxSubscription: vi
 .spyOn(outlook, "setupMailboxSubscription")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(outlook, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(outlook, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(outlook, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { outlook, spies };
 }

 const subscription = (expiration: Date) => ({
 subscriptionId: "sub-1",
 clientState: "secret",
 webhookUrl: "https://example.com/hook",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 60 * 60 * 1000);

 it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
 const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(PAST)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(FUTURE)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { outlook, spies } = setup([["enabled_channels", []]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

KrisBraun
merged 1 commit into

from

Jun 22, 2026
 Merged


fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade
#217
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


89 changes: 89 additions & 0 deletions
connectors/gmail/src/gmail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
 });
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const gmail = new Gmail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxWebhook: vi
 .spyOn(gmail, "setupMailboxWebhook")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(gmail, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(gmail, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(gmail, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { gmail, spies };
 }

 const webhook = (expiration: Date) => ({
 topicName: "topic",
 historyId: "1",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

 it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
 const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired watch: treated as stranded — re-establishes and backfills", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(PAST)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(FUTURE)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { gmail, spies } = setup([["enabled_channels", []]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});

110 changes: 84 additions & 26 deletions
connectors/gmail/src/gmail.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
 await this.set("auth_actor_id", context.actor.id);
 }

 override async upgrade(): Promise<void> {
 // `mailbox_webhook` is the canonical "already on the mailbox-wide watch
 // model" sentinel. Instances that predate that change still carry old
 // per-channel keys and need a one-time migration before recovery runs.
 const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (!migrated) {
 await this.migrateLegacyPerChannelState();
 }

 // Durable recovery backstop, run on every deploy. Re-asserts recurring
 // maintenance for a healthy mailbox and re-establishes (plus backfills) a
 // stranded one. See recoverMailboxDelivery for the stranded cases.
 await this.recoverMailboxDelivery();
 }

 /**
 * Migration from per-channel watches to a single mailbox-wide watch.
 * Ensure live push delivery + recurring maintenance for any instance with
 * enabled channels. Runs from upgrade() on every deploy.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 * Two stranded states were previously unrecoverable — neither the old
 * `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
 * sweep could heal them, so the connection stayed silently dead until the
 * user manually re-enabled a channel:
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs.
 * 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
 * before its `set()` (e.g. the Gmail "Only one user push notification
 * client allowed per developer" 400). With no sentinel there was
 * nothing to re-assert, and because `scheduleRecurring` never ran the
 * maintenance sweep's `ever` marker was never set either; and
 * 2. the watch expired while the self-heal/renewal chain was dead.
 *
 * A healthy watch (present and unexpired) only re-asserts the recurring
 * tasks. A missing or expired watch is re-established AND every enabled label
 * is re-walked, so mail that accumulated while delivery was dead is
 * backfilled. The backfill upserts by `source` (no duplicates) and uses
 * initial-sync semantics (read/unarchived), so it never spams notifications.
 */
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (already) {
 // Re-assert durable recurring maintenance for any instance that had an
 // active mailbox watch but whose pre-recurring self-heal/renewal chain
 // died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
 // Healthy watch — re-assert durable maintenance (idempotent, keyed).
 try {
 await this.scheduleSelfHealCheck();
 await this.scheduleMailboxRenewal(new Date(already.expiration));
 await this.scheduleMailboxRenewal(new Date(webhook.expiration));
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
 return;
 }

 // Stranded: watch missing or expired with nothing live renewing it.
 try {
 for (const channelId of enabled) {
 await this.requeueInitialSync(channelId);
 }
 await this.setupMailboxWebhook();
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
 error
 );
 }
 }

 /**
 * Re-queue a fresh full backfill of one label, dropping any stale cursor so
 * the walk restarts from the newest thread. Used by recovery to re-import
 * mail that arrived while push delivery was dead.
 */
 private async requeueInitialSync(channelId: string): Promise<void> {
 await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
 const initial = await this.callback(this.initialSyncBatch, channelId, 1);
 await this.runTask(initial);
 }

 /**
 * One-time migration from per-channel watches to a single mailbox-wide watch.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs. The mailbox
 * watch itself is (re-)established afterward by recoverMailboxDelivery().
 */
 private async migrateLegacyPerChannelState(): Promise<void> {
 // Stop whatever per-channel watch was last active. Gmail allows only
 // one watch per mailbox, so a single stopWatch() call covers all of
 // them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
 }

 // Probe known system labels for old per-channel state and migrate.
 let migratedAny = false;
 for (const labelId of SYSTEM_LABEL_ORDER) {
 const oldWebhook = await this.get<{ topicName?: string }>(
 `channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
 // Channel was enabled (presence of any old per-channel key counts).
 if (oldEnabled || oldWebhook) {
 await this.addEnabledChannel(labelId);
 migratedAny = true;
 }

 // Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
 await this.clear(`sync_state_${labelId}`);
 await this.clear(`sync_enabled_${labelId}`);
 }

 if (migratedAny) {
 await this.setupMailboxWebhook();
 }
 }

 private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down

98 changes: 96 additions & 2 deletions
connectors/outlook-mail/src/outlook-mail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
 OutlookMail,
 pickChannelForConversation,
 recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
 ).toEqual(["a@b.com"]);
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const outlook = new OutlookMail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxSubscription: vi
 .spyOn(outlook, "setupMailboxSubscription")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(outlook, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(outlook, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(outlook, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { outlook, spies };
 }

 const subscription = (expiration: Date) => ({
 subscriptionId: "sub-1",
 clientState: "secret",
 webhookUrl: "https://example.com/hook",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 60 * 60 * 1000);

 it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
 const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(PAST)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(FUTURE)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { outlook, spies } = setup([["enabled_channels", []]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

KrisBraun
merged 1 commit into

from

Jun 22, 2026
 Merged


fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade
#217
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


89 changes: 89 additions & 0 deletions
connectors/gmail/src/gmail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
 });
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const gmail = new Gmail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxWebhook: vi
 .spyOn(gmail, "setupMailboxWebhook")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(gmail, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(gmail, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(gmail, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { gmail, spies };
 }

 const webhook = (expiration: Date) => ({
 topicName: "topic",
 historyId: "1",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

 it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
 const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired watch: treated as stranded — re-establishes and backfills", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(PAST)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(FUTURE)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { gmail, spies } = setup([["enabled_channels", []]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});

110 changes: 84 additions & 26 deletions
connectors/gmail/src/gmail.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
 await this.set("auth_actor_id", context.actor.id);
 }

 override async upgrade(): Promise<void> {
 // `mailbox_webhook` is the canonical "already on the mailbox-wide watch
 // model" sentinel. Instances that predate that change still carry old
 // per-channel keys and need a one-time migration before recovery runs.
 const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (!migrated) {
 await this.migrateLegacyPerChannelState();
 }

 // Durable recovery backstop, run on every deploy. Re-asserts recurring
 // maintenance for a healthy mailbox and re-establishes (plus backfills) a
 // stranded one. See recoverMailboxDelivery for the stranded cases.
 await this.recoverMailboxDelivery();
 }

 /**
 * Migration from per-channel watches to a single mailbox-wide watch.
 * Ensure live push delivery + recurring maintenance for any instance with
 * enabled channels. Runs from upgrade() on every deploy.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 * Two stranded states were previously unrecoverable — neither the old
 * `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
 * sweep could heal them, so the connection stayed silently dead until the
 * user manually re-enabled a channel:
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs.
 * 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
 * before its `set()` (e.g. the Gmail "Only one user push notification
 * client allowed per developer" 400). With no sentinel there was
 * nothing to re-assert, and because `scheduleRecurring` never ran the
 * maintenance sweep's `ever` marker was never set either; and
 * 2. the watch expired while the self-heal/renewal chain was dead.
 *
 * A healthy watch (present and unexpired) only re-asserts the recurring
 * tasks. A missing or expired watch is re-established AND every enabled label
 * is re-walked, so mail that accumulated while delivery was dead is
 * backfilled. The backfill upserts by `source` (no duplicates) and uses
 * initial-sync semantics (read/unarchived), so it never spams notifications.
 */
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (already) {
 // Re-assert durable recurring maintenance for any instance that had an
 // active mailbox watch but whose pre-recurring self-heal/renewal chain
 // died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
 // Healthy watch — re-assert durable maintenance (idempotent, keyed).
 try {
 await this.scheduleSelfHealCheck();
 await this.scheduleMailboxRenewal(new Date(already.expiration));
 await this.scheduleMailboxRenewal(new Date(webhook.expiration));
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
 return;
 }

 // Stranded: watch missing or expired with nothing live renewing it.
 try {
 for (const channelId of enabled) {
 await this.requeueInitialSync(channelId);
 }
 await this.setupMailboxWebhook();
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
 error
 );
 }
 }

 /**
 * Re-queue a fresh full backfill of one label, dropping any stale cursor so
 * the walk restarts from the newest thread. Used by recovery to re-import
 * mail that arrived while push delivery was dead.
 */
 private async requeueInitialSync(channelId: string): Promise<void> {
 await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
 const initial = await this.callback(this.initialSyncBatch, channelId, 1);
 await this.runTask(initial);
 }

 /**
 * One-time migration from per-channel watches to a single mailbox-wide watch.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs. The mailbox
 * watch itself is (re-)established afterward by recoverMailboxDelivery().
 */
 private async migrateLegacyPerChannelState(): Promise<void> {
 // Stop whatever per-channel watch was last active. Gmail allows only
 // one watch per mailbox, so a single stopWatch() call covers all of
 // them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
 }

 // Probe known system labels for old per-channel state and migrate.
 let migratedAny = false;
 for (const labelId of SYSTEM_LABEL_ORDER) {
 const oldWebhook = await this.get<{ topicName?: string }>(
 `channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
 // Channel was enabled (presence of any old per-channel key counts).
 if (oldEnabled || oldWebhook) {
 await this.addEnabledChannel(labelId);
 migratedAny = true;
 }

 // Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
 await this.clear(`sync_state_${labelId}`);
 await this.clear(`sync_enabled_${labelId}`);
 }

 if (migratedAny) {
 await this.setupMailboxWebhook();
 }
 }

 private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down

98 changes: 96 additions & 2 deletions
connectors/outlook-mail/src/outlook-mail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
 OutlookMail,
 pickChannelForConversation,
 recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
 ).toEqual(["a@b.com"]);
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const outlook = new OutlookMail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxSubscription: vi
 .spyOn(outlook, "setupMailboxSubscription")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(outlook, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(outlook, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(outlook, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { outlook, spies };
 }

 const subscription = (expiration: Date) => ({
 subscriptionId: "sub-1",
 clientState: "secret",
 webhookUrl: "https://example.com/hook",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 60 * 60 * 1000);

 it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
 const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(PAST)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(FUTURE)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { outlook, spies } = setup([["enabled_channels", []]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

KrisBraun
merged 1 commit into

from

Jun 22, 2026
 Merged


fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade
#217
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


89 changes: 89 additions & 0 deletions
connectors/gmail/src/gmail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
 });
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const gmail = new Gmail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxWebhook: vi
 .spyOn(gmail, "setupMailboxWebhook")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(gmail, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(gmail, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(gmail, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { gmail, spies };
 }

 const webhook = (expiration: Date) => ({
 topicName: "topic",
 historyId: "1",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

 it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
 const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired watch: treated as stranded — re-establishes and backfills", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(PAST)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(FUTURE)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { gmail, spies } = setup([["enabled_channels", []]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});

110 changes: 84 additions & 26 deletions
connectors/gmail/src/gmail.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
 await this.set("auth_actor_id", context.actor.id);
 }

 override async upgrade(): Promise<void> {
 // `mailbox_webhook` is the canonical "already on the mailbox-wide watch
 // model" sentinel. Instances that predate that change still carry old
 // per-channel keys and need a one-time migration before recovery runs.
 const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (!migrated) {
 await this.migrateLegacyPerChannelState();
 }

 // Durable recovery backstop, run on every deploy. Re-asserts recurring
 // maintenance for a healthy mailbox and re-establishes (plus backfills) a
 // stranded one. See recoverMailboxDelivery for the stranded cases.
 await this.recoverMailboxDelivery();
 }

 /**
 * Migration from per-channel watches to a single mailbox-wide watch.
 * Ensure live push delivery + recurring maintenance for any instance with
 * enabled channels. Runs from upgrade() on every deploy.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 * Two stranded states were previously unrecoverable — neither the old
 * `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
 * sweep could heal them, so the connection stayed silently dead until the
 * user manually re-enabled a channel:
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs.
 * 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
 * before its `set()` (e.g. the Gmail "Only one user push notification
 * client allowed per developer" 400). With no sentinel there was
 * nothing to re-assert, and because `scheduleRecurring` never ran the
 * maintenance sweep's `ever` marker was never set either; and
 * 2. the watch expired while the self-heal/renewal chain was dead.
 *
 * A healthy watch (present and unexpired) only re-asserts the recurring
 * tasks. A missing or expired watch is re-established AND every enabled label
 * is re-walked, so mail that accumulated while delivery was dead is
 * backfilled. The backfill upserts by `source` (no duplicates) and uses
 * initial-sync semantics (read/unarchived), so it never spams notifications.
 */
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (already) {
 // Re-assert durable recurring maintenance for any instance that had an
 // active mailbox watch but whose pre-recurring self-heal/renewal chain
 // died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
 // Healthy watch — re-assert durable maintenance (idempotent, keyed).
 try {
 await this.scheduleSelfHealCheck();
 await this.scheduleMailboxRenewal(new Date(already.expiration));
 await this.scheduleMailboxRenewal(new Date(webhook.expiration));
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
 return;
 }

 // Stranded: watch missing or expired with nothing live renewing it.
 try {
 for (const channelId of enabled) {
 await this.requeueInitialSync(channelId);
 }
 await this.setupMailboxWebhook();
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
 error
 );
 }
 }

 /**
 * Re-queue a fresh full backfill of one label, dropping any stale cursor so
 * the walk restarts from the newest thread. Used by recovery to re-import
 * mail that arrived while push delivery was dead.
 */
 private async requeueInitialSync(channelId: string): Promise<void> {
 await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
 const initial = await this.callback(this.initialSyncBatch, channelId, 1);
 await this.runTask(initial);
 }

 /**
 * One-time migration from per-channel watches to a single mailbox-wide watch.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs. The mailbox
 * watch itself is (re-)established afterward by recoverMailboxDelivery().
 */
 private async migrateLegacyPerChannelState(): Promise<void> {
 // Stop whatever per-channel watch was last active. Gmail allows only
 // one watch per mailbox, so a single stopWatch() call covers all of
 // them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
 }

 // Probe known system labels for old per-channel state and migrate.
 let migratedAny = false;
 for (const labelId of SYSTEM_LABEL_ORDER) {
 const oldWebhook = await this.get<{ topicName?: string }>(
 `channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
 // Channel was enabled (presence of any old per-channel key counts).
 if (oldEnabled || oldWebhook) {
 await this.addEnabledChannel(labelId);
 migratedAny = true;
 }

 // Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
 await this.clear(`sync_state_${labelId}`);
 await this.clear(`sync_enabled_${labelId}`);
 }

 if (migratedAny) {
 await this.setupMailboxWebhook();
 }
 }

 private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down

98 changes: 96 additions & 2 deletions
connectors/outlook-mail/src/outlook-mail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
 OutlookMail,
 pickChannelForConversation,
 recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
 ).toEqual(["a@b.com"]);
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const outlook = new OutlookMail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxSubscription: vi
 .spyOn(outlook, "setupMailboxSubscription")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(outlook, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(outlook, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(outlook, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { outlook, spies };
 }

 const subscription = (expiration: Date) => ({
 subscriptionId: "sub-1",
 clientState: "secret",
 webhookUrl: "https://example.com/hook",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 60 * 60 * 1000);

 it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
 const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(PAST)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(FUTURE)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { outlook, spies } = setup([["enabled_channels", []]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade#217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

KrisBraun
merged 1 commit into

from

Jun 22, 2026
 Merged


fix(gmail,outlook-mail): durable recovery for stranded mailbox watches on upgrade
#217
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


89 changes: 89 additions & 0 deletions
connectors/gmail/src/gmail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -473,3 +473,92 @@ describe("processEmailThreads — no status set", () => {
 });
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const gmail = new Gmail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxWebhook: vi
 .spyOn(gmail, "setupMailboxWebhook")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(gmail, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(gmail, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(gmail, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { gmail, spies };
 }

 const webhook = (expiration: Date) => ({
 topicName: "topic",
 historyId: "1",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 24 * 60 * 60 * 1000);

 it("stranded (no mailbox_webhook): re-establishes the watch and backfills every enabled label", async () => {
 const { gmail, spies } = setup([["enabled_channels", ["INBOX", "SENT"]]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("SENT");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired watch: treated as stranded — re-establishes and backfills", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(PAST)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("INBOX");
 expect(spies.setupMailboxWebhook).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy watch: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { gmail, spies } = setup([
 ["enabled_channels", ["INBOX"]],
 ["mailbox_webhook", webhook(FUTURE)],
 ]);
 await gmail.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { gmail, spies } = setup([["enabled_channels", []]]);
 await gmail.recoverMailboxDelivery();
 expect(spies.setupMailboxWebhook).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});

110 changes: 84 additions & 26 deletions
connectors/gmail/src/gmail.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -257,32 +257,53 @@ export class Gmail extends Connector<Gmail> {
 await this.set("auth_actor_id", context.actor.id);
 }

 override async upgrade(): Promise<void> {
 // `mailbox_webhook` is the canonical "already on the mailbox-wide watch
 // model" sentinel. Instances that predate that change still carry old
 // per-channel keys and need a one-time migration before recovery runs.
 const migrated = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (!migrated) {
 await this.migrateLegacyPerChannelState();
 }

 // Durable recovery backstop, run on every deploy. Re-asserts recurring
 // maintenance for a healthy mailbox and re-establishes (plus backfills) a
 // stranded one. See recoverMailboxDelivery for the stranded cases.
 await this.recoverMailboxDelivery();
 }

 /**
 * Migration from per-channel watches to a single mailbox-wide watch.
 * Ensure live push delivery + recurring maintenance for any instance with
 * enabled channels. Runs from upgrade() on every deploy.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 * Two stranded states were previously unrecoverable — neither the old
 * `if (mailbox_webhook) re-assert` upgrade path nor the cron maintenance
 * sweep could heal them, so the connection stayed silently dead until the
 * user manually re-enabled a channel:
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs.
 * 1. `mailbox_webhook` never persisted — a prior `setupWatch()` threw
 * before its `set()` (e.g. the Gmail "Only one user push notification
 * client allowed per developer" 400). With no sentinel there was
 * nothing to re-assert, and because `scheduleRecurring` never ran the
 * maintenance sweep's `ever` marker was never set either; and
 * 2. the watch expired while the self-heal/renewal chain was dead.
 *
 * A healthy watch (present and unexpired) only re-asserts the recurring
 * tasks. A missing or expired watch is re-established AND every enabled label
 * is re-walked, so mail that accumulated while delivery was dead is
 * backfilled. The backfill upserts by `source` (no duplicates) and uses
 * initial-sync semantics (read/unarchived), so it never spams notifications.
 */
override async upgrade(): Promise<void> {
// Already migrated? `mailbox_webhook` is the canonical sentinel.
const already = await this.get<MailboxWebhookState>("mailbox_webhook");
 if (already) {
 // Re-assert durable recurring maintenance for any instance that had an
 // active mailbox watch but whose pre-recurring self-heal/renewal chain
 // died. scheduleRecurring is idempotent (keyed replace).
private async recoverMailboxDelivery(): Promise<void> {
const enabled = await this.getEnabledChannels();
if (enabled.size === 0) return;

const webhook = await this.get<MailboxWebhookState>("mailbox_webhook");
if (webhook && new Date(webhook.expiration).getTime() > Date.now()) {
 // Healthy watch — re-assert durable maintenance (idempotent, keyed).
 try {
 await this.scheduleSelfHealCheck();
 await this.scheduleMailboxRenewal(new Date(already.expiration));
 await this.scheduleMailboxRenewal(new Date(webhook.expiration));
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: failed to re-assert recurring tasks`,
Expand All@@ -292,6 +313,49 @@ export class Gmail extends Connector<Gmail> {
 return;
 }

 // Stranded: watch missing or expired with nothing live renewing it.
 try {
 for (const channelId of enabled) {
 await this.requeueInitialSync(channelId);
 }
 await this.setupMailboxWebhook();
 } catch (error) {
 console.error(
 `Gmail upgrade [${this.id}]: stranded-mailbox recovery failed`,
 error
 );
 }
 }

 /**
 * Re-queue a fresh full backfill of one label, dropping any stale cursor so
 * the walk restarts from the newest thread. Used by recovery to re-import
 * mail that arrived while push delivery was dead.
 */
 private async requeueInitialSync(channelId: string): Promise<void> {
 await this.set<InitialSyncState>(`initial_state_${channelId}`, {});
 const initial = await this.callback(this.initialSyncBatch, channelId, 1);
 await this.runTask(initial);
 }

 /**
 * One-time migration from per-channel watches to a single mailbox-wide watch.
 *
 * Old layout (per-channel):
 * - `channel_webhook_${id}`, `watch_renewal_task_${id}`, `sync_state_${id}`,
 * `sync_enabled_${id}`
 * New layout (per-twist-instance):
 * - `mailbox_webhook`, `incremental_state`, `enabled_channels`, plus
 * `initial_state_${id}` for in-flight backfills; recurring tasks are
 * keyed "mailbox-watch-renewal" and "mailbox-self-heal"
 *
 * The Twist runtime API doesn't let connectors enumerate stored keys, so we
 * probe the system labels we know users can enable. Users with custom
 * Gmail labels enabled (e.g. `Label_14`) will need to re-toggle them after
 * this upgrade — without `list()` we can't discover their IDs. The mailbox
 * watch itself is (re-)established afterward by recoverMailboxDelivery().
 */
 private async migrateLegacyPerChannelState(): Promise<void> {
 // Stop whatever per-channel watch was last active. Gmail allows only
 // one watch per mailbox, so a single stopWatch() call covers all of
 // them — but the call needs an authed API client.
Expand All@@ -305,7 +369,6 @@ export class Gmail extends Connector<Gmail> {
 }

 // Probe known system labels for old per-channel state and migrate.
 let migratedAny = false;
 for (const labelId of SYSTEM_LABEL_ORDER) {
 const oldWebhook = await this.get<{ topicName?: string }>(
 `channel_webhook_${labelId}`
Expand All@@ -328,7 +391,6 @@ export class Gmail extends Connector<Gmail> {
 // Channel was enabled (presence of any old per-channel key counts).
 if (oldEnabled || oldWebhook) {
 await this.addEnabledChannel(labelId);
 migratedAny = true;
 }

 // Carry over any in-flight initial-backfill cursor.
Expand DownExpand Up@@ -369,10 +431,6 @@ export class Gmail extends Connector<Gmail> {
 await this.clear(`sync_state_${labelId}`);
 await this.clear(`sync_enabled_${labelId}`);
 }

 if (migratedAny) {
 await this.setupMailboxWebhook();
 }
 }

 private async findAnyAuthApi(): Promise<GmailApi | null> {
Expand Down

98 changes: 96 additions & 2 deletions
connectors/outlook-mail/src/outlook-mail.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { pickChannelForConversation, recipientsFor } from "./outlook-mail";
import { describe, expect, it, vi } from "vitest";
import {
 OutlookMail,
 pickChannelForConversation,
 recipientsFor,
} from "./outlook-mail";
import type { GraphMessage, WellKnownFolders } from "./graph-mail-api";

const inFolder = (parentFolderId: string): GraphMessage =>
Expand DownExpand Up@@ -72,3 +76,93 @@ describe("recipientsFor", () => {
 ).toEqual(["a@b.com"]);
 });
});

describe("recoverMailboxDelivery — durable recovery on upgrade", () => {
 function setup(entries: Array<[string, unknown]>) {
 const storeMap = new Map<string, unknown>(entries);
 const store = {
 get: vi.fn(async (k: string) =>
 storeMap.has(k) ? storeMap.get(k) : null
 ),
 set: vi.fn(async (k: string, v: unknown) => {
 storeMap.set(k, v);
 }),
 clear: vi.fn(async (k: string) => {
 storeMap.delete(k);
 }),
 list: vi.fn(async (p: string) =>
 [...storeMap.keys()].filter((k) => k.startsWith(p))
 ),
 };
 const tools = { store, integrations: {}, network: {}, files: {} };
 const outlook = new OutlookMail(
 "twist-instance-1" as never,
 { getTools: () => tools } as never
 ) as any;
 const spies = {
 setupMailboxSubscription: vi
 .spyOn(outlook, "setupMailboxSubscription")
 .mockResolvedValue(undefined),
 scheduleSelfHealCheck: vi
 .spyOn(outlook, "scheduleSelfHealCheck")
 .mockResolvedValue(undefined),
 scheduleMailboxRenewal: vi
 .spyOn(outlook, "scheduleMailboxRenewal")
 .mockResolvedValue(undefined),
 requeueInitialSync: vi
 .spyOn(outlook, "requeueInitialSync")
 .mockResolvedValue(undefined),
 };
 return { outlook, spies };
 }

 const subscription = (expiration: Date) => ({
 subscriptionId: "sub-1",
 clientState: "secret",
 webhookUrl: "https://example.com/hook",
 expiration,
 created: "2026-01-01T00:00:00.000Z",
 });
 const FUTURE = new Date(Date.now() + 2 * 24 * 60 * 60 * 1000);
 const PAST = new Date(Date.now() - 60 * 60 * 1000);

 it("stranded (no mailbox_subscription): re-establishes and backfills every enabled folder", async () => {
 const { outlook, spies } = setup([["enabled_channels", ["f-inbox", "f-sent"]]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-sent");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 });

 it("expired subscription: treated as stranded — re-establishes and backfills", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(PAST)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.requeueInitialSync).toHaveBeenCalledWith("f-inbox");
 expect(spies.setupMailboxSubscription).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).not.toHaveBeenCalled();
 });

 it("healthy subscription: only re-asserts recurring tasks (no re-setup, no backfill)", async () => {
 const { outlook, spies } = setup([
 ["enabled_channels", ["f-inbox"]],
 ["mailbox_subscription", subscription(FUTURE)],
 ]);
 await outlook.recoverMailboxDelivery();
 expect(spies.scheduleSelfHealCheck).toHaveBeenCalledTimes(1);
 expect(spies.scheduleMailboxRenewal).toHaveBeenCalledTimes(1);
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });

 it("no enabled channels: does nothing", async () => {
 const { outlook, spies } = setup([["enabled_channels", []]]);
 await outlook.recoverMailboxDelivery();
 expect(spies.setupMailboxSubscription).not.toHaveBeenCalled();
 expect(spies.scheduleSelfHealCheck).not.toHaveBeenCalled();
 expect(spies.requeueInitialSync).not.toHaveBeenCalled();
 });
});
Loading
Loading