Coalesce webhook-triggered incremental sync in MS Teams and Slack

connectors/ms-teams/src/ms-teams.ts

@@ -27,6 +27,13 @@ import {
 
 const DM_CHANNEL_ID = "__direct_messages__";
 const MAX_SYNC_BATCHES = 50;
+
+/**
+ * Delay before a webhook-triggered incremental sync runs. The pass is
+ * scheduled as a keyed coalescing task, so a burst of Graph notifications
+ * collapses into one pass that fires at most this long after the first one.
+ */
+const INCREMENTAL_SYNC_COALESCE_MS = 10_000;
 /** Graph subscriptions for Teams channel messages max out at ~60 minutes. */
 const SUBSCRIPTION_EXPIRY_MINUTES = 55;
 
@@ -459,14 +466,24 @@ export class MsTeams extends Connector<MsTeams> {
  };
  await this.set(`sync_state_${channelId}`, incrementalState);
 
+ // Coalesced: Graph sends one change notification per channel message, so
+ // enqueueing an immediate task per notification floods the queue during
+ // active conversation and the duplicate passes (each re-fetching the same
+ // 1-hour window) stack concurrently into one worker. A notification burst
+ // collapses into a single pass; one arriving mid-pass schedules exactly
+ // one follow-up. The 1-hour window means the delayed pass still covers
+ // every notified change.
  const syncCallback = await this.callback(
  this.syncBatch,
  1,
  "incremental",
  channelId,
  false
  );
- await this.runTask(syncCallback);
+ await this.scheduleTask(`incremental-sync:${channelId}`, syncCallback, {
+ runAt: new Date(Date.now() + INCREMENTAL_SYNC_COALESCE_MS),
+ coalesce: true,
+ });
  }
 
  // ---- Subscription renewal ----

connectors/slack/src/slack.test.ts

@@ -423,3 +423,40 @@ describe("onNoteReactionChanged (custom emoji outbound)", () => {
  expect(fetchMock).not.toHaveBeenCalled();
  });
 });
+
+describe("startIncrementalSync — coalesced scheduling", () => {
+ it("schedules a keyed coalescing task instead of enqueueing per event", async () => {
+ const store = makeStore({ channel_webhook_C123: { url: "https://x" } });
+ const scheduleTask = vi.fn(async () => "cancel-token");
+ const runTask = vi.fn(async () => {});
+ const tools = {
+ store,
+ callbacks: { create: vi.fn(async () => "cb-token") },
+ tasks: { scheduleTask, runTask },
+ integrations: {},
+ network: {},
+ files: {},
+ };
+ const slack = new Slack(
+ "twist-instance-1" as never,
+ { getTools: () => tools } as never
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
+ ) as any;
+
+ await slack.startIncrementalSync("C123");
+
+ // One Slack event per message must NOT become one queued task per event —
+ // the pass is scheduled under a stable per-channel key with coalesce so
+ // bursts collapse into a single pending pass.
+ expect(runTask).not.toHaveBeenCalled();
+ expect(scheduleTask).toHaveBeenCalledTimes(1);
+ const [key, , options] = scheduleTask.mock.calls[0] as unknown as [
+ string,
+ unknown,
+ { runAt: Date; coalesce?: boolean },
+ ];
+ expect(key).toBe("incremental-sync:C123");
+ expect(options.coalesce).toBe(true);
+ expect(options.runAt.getTime()).toBeGreaterThan(Date.now());
+ });
+});

connectors/slack/src/slack.ts

@@ -73,6 +73,14 @@ import { slackFacets } from "./slack-facets";
  * - `mpim:history` - Read group direct messages
  * - `stars:read` / `stars:write` - Read and manage the user's saved items
  */
+
+/**
+ * Delay before an event-triggered incremental sync runs. The pass is
+ * scheduled as a keyed coalescing task, so a burst of message events
+ * collapses into one pass that fires at most this long after the first one.
+ */
+const INCREMENTAL_SYNC_COALESCE_MS = 10_000;
+
 export class Slack extends Connector<Slack> {
  static readonly PROVIDER = AuthProvider.Slack;
  static readonly handleReplies = true;
@@ -1021,14 +1029,25 @@ export class Slack extends Connector<Slack> {
  };
 
  await this.set(`sync_state_${channelId}`, incrementalState);
+
+ // Coalesced: Slack delivers one event per message, so enqueueing an
+ // immediate task per event turns a busy channel into a flood of duplicate
+ // passes (each re-fetching the same 15-minute window) that run
+ // concurrently in one worker. A burst of events collapses into a single
+ // pass; an event arriving mid-pass schedules exactly one follow-up. The
+ // 15-minute window (with the coalesce delay well inside it) means the
+ // delayed pass still covers every notified message.
  const syncCallback = await this.callback(
  this.syncBatch,
  1,
  "incremental",
  channelId,
  false
  );
- await this.runTask(syncCallback);
+ await this.scheduleTask(`incremental-sync:${channelId}`, syncCallback, {
+ runAt: new Date(Date.now() + INCREMENTAL_SYNC_COALESCE_MS),
+ coalesce: true,
+ });
  }
 
  private starredKey(channelId: string, threadTs: string): string {