Skip to content

Commit b06f40a

Browse files
steipetesteipete
authored
refactor(plugin-sdk): share ingress lifecycle fan-in (#113648)
* refactor(channels): share ingress lifecycle fan-in * chore(plugin-sdk): refresh ingress API baseline
1 parent 92e4041 commit b06f40a

15 files changed

Lines changed: 179 additions & 428 deletions

File tree

‎docs/.generated/plugin-sdk-api-baseline.sha256‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ fbf353eb38ae68d8ded3f2a60b432c7bb2c245d2ec7e7c9f53c6da19a0db0938 module/channel
3030
982f29a18e07228e3da82cae67d06ff38249592a29c2fd28f01f0d2016ff80d9 module/channel-feedback
3131
a18c29e77a6797a9712e52f9469062438f32c33209c54be23e5bbc50efa2fc7d module/channel-inbound
3232
2940039d5ebdb16c1d745914390db01c0f5f1e9cff33f95ea36c9e77d97c2bae module/channel-inbound-debounce
33-
91425ad75ddb58a3796f7bc88570cb2c81b3db0dd6521a7ce0e55f9148880a47 module/channel-ingress-runtime
33+
4a7ada095f0f483525dcbd848fbccab26473749eba87e6a6c6e5074fd04ee1d1 module/channel-ingress-runtime
3434
c97dd36cdf8f83c2893c33e9430a93cd131a03d855725783ca5b545de0cf84f8 module/channel-lifecycle
3535
159d034b431d113f3a6dc41ec0bcadba2d6664051f158330b0e3dd3da8b5d42f module/channel-logging
3636
b8a583252a6188a5c6e15b093e08407c691e8714ec51d4adadfac49c1e76a55d module/channel-message

‎extensions/discord/src/monitor/message-dispatcher.ts‎

Lines changed: 4 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import {
33
createChannelInboundDebouncer,
44
shouldDebounceTextInbound,
55
} from "openclaw/plugin-sdk/channel-inbound";
6+
import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime";
67
import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime";
78
import { danger } from "openclaw/plugin-sdk/runtime-env";
89
import { resolveOpenProviderRuntimeGroupPolicy } from "openclaw/plugin-sdk/runtime-group-policy";
@@ -58,82 +59,10 @@ type DiscordMessageDispatcherWithLifecycle = DiscordMessageDispatcher & {
5859
deactivate: () => Promise<void>;
5960
};
6061

61-
type DiscordFlushIngressSettlement = {
62-
lifecycle: DiscordIngressLifecycle | undefined;
63-
settle: () => Promise<void>;
64-
abandon: (error?: unknown) => Promise<void>;
65-
};
66-
6762
function isNonEmptyString(value: string | undefined): value is string {
6863
return typeof value === "string" && value.length > 0;
6964
}
7065

71-
function buildFlushIngressLifecycle(
72-
entries: Array<{ turnAdoptionLifecycle?: DiscordIngressLifecycle }>,
73-
): DiscordFlushIngressSettlement {
74-
const lifecycles = entries
75-
.map((entry) => entry.turnAdoptionLifecycle)
76-
.filter((lifecycle) => lifecycle !== undefined);
77-
const [firstLifecycle] = lifecycles;
78-
if (!firstLifecycle) {
79-
return {
80-
lifecycle: undefined,
81-
settle: async () => {},
82-
abandon: async () => {},
83-
};
84-
}
85-
let handedOff = false;
86-
const adoptAll = async () => {
87-
for (const lifecycle of lifecycles) {
88-
await lifecycle.onAdopted();
89-
}
90-
};
91-
return {
92-
lifecycle: {
93-
abortSignal:
94-
lifecycles.length === 1
95-
? firstLifecycle.abortSignal
96-
: AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal)),
97-
onAdopted: async () => {
98-
handedOff = true;
99-
await adoptAll();
100-
},
101-
onDeferred: () => {
102-
handedOff = true;
103-
for (const lifecycle of lifecycles) {
104-
lifecycle.onDeferred();
105-
}
106-
},
107-
onAdoptionFinalizing: () => {
108-
for (const lifecycle of lifecycles) {
109-
lifecycle.onAdoptionFinalizing();
110-
}
111-
},
112-
onAbandoned: async () => {
113-
handedOff = true;
114-
for (const lifecycle of lifecycles) {
115-
await lifecycle.onAbandoned();
116-
}
117-
},
118-
},
119-
// A gate or deliberate no-dispatch still consumes every merged queue row.
120-
settle: async () => {
121-
if (!handedOff) {
122-
await adoptAll();
123-
}
124-
},
125-
abandon: async () => {
126-
if (handedOff) {
127-
return;
128-
}
129-
handedOff = true;
130-
for (const lifecycle of lifecycles) {
131-
await lifecycle.onAbandoned();
132-
}
133-
},
134-
};
135-
}
136-
13766
export function createDiscordMessageDispatcher(
13867
params: DiscordMessageHandlerParams,
13968
): DiscordMessageDispatcherWithLifecycle {
@@ -209,7 +138,9 @@ export function createDiscordMessageDispatcher(
209138
if (!last) {
210139
return;
211140
}
212-
const ingress = buildFlushIngressLifecycle(entries);
141+
const ingress = fanInChannelIngressLifecycles(
142+
entries.map((entry) => entry.turnAdoptionLifecycle),
143+
);
213144
const abortSignal = last.abortSignal;
214145
if (abortSignal?.aborted) {
215146
await ingress.abandon(abortSignal.reason);

‎extensions/feishu/src/feishu-ingress.ts‎

Lines changed: 10 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
// Feishu plugin owns raw Lark event admission, replay, and turn adoption.
22
import * as Lark from "@larksuiteoapi/node-sdk";
3+
import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime";
34
import {
45
createChannelIngressMonitor,
56
DEFAULT_INGRESS_ADOPTION_STALL_MS,
@@ -219,8 +220,8 @@ export function buildFeishuFlushIngressLifecycle(
219220
const replayClaims = durableSources
220221
.map((source) => source.replayClaim)
221222
.filter((claim) => claim !== undefined);
222-
const [firstLifecycle] = lifecycles;
223-
if (!firstLifecycle) {
223+
const transportLifecycle = fanInChannelIngressLifecycles(lifecycles).lifecycle;
224+
if (!transportLifecycle) {
224225
return { lifecycle: undefined, settle: async () => {} };
225226
}
226227
let handedOff = false;
@@ -237,7 +238,7 @@ export function buildFeishuFlushIngressLifecycle(
237238
return;
238239
}
239240
releaseReplayClaims();
240-
await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onAbandoned()));
241+
await transportLifecycle.onAbandoned();
241242
terminal = "abandoned";
242243
};
243244
const ensureAbandoned = async () => {
@@ -280,9 +281,7 @@ export function buildFeishuFlushIngressLifecycle(
280281
adopting ??
281282
(async () => {
282283
try {
283-
for (const lifecycle of lifecycles) {
284-
await lifecycle.onAdopted();
285-
}
284+
await transportLifecycle.onAdopted();
286285
terminal = "adopted";
287286
// Queue adoption is authoritative. Logical twin guards commit only
288287
// afterward: partial best-effort guard writes may admit a duplicate,
@@ -315,24 +314,17 @@ export function buildFeishuFlushIngressLifecycle(
315314
};
316315
return {
317316
lifecycle: {
318-
abortSignal:
319-
lifecycles.length === 1
320-
? firstLifecycle.abortSignal
321-
: AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal)),
317+
abortSignal: transportLifecycle.abortSignal,
322318
onAdopted: async () => {
323319
handedOff = true;
324320
await adoptAll();
325321
},
326322
onDeferred: () => {
327323
handedOff = true;
328-
for (const lifecycle of lifecycles) {
329-
lifecycle.onDeferred();
330-
}
324+
transportLifecycle.onDeferred();
331325
},
332326
onAdoptionFinalizing: () => {
333-
for (const lifecycle of lifecycles) {
334-
lifecycle.onAdoptionFinalizing();
335-
}
327+
transportLifecycle.onAdoptionFinalizing();
336328
},
337329
onAbandoned: async () => {
338330
handedOff = true;
@@ -348,12 +340,8 @@ export function buildFeishuFlushIngressLifecycle(
348340
handedOff = true;
349341
releaseReplayClaims();
350342
try {
351-
for (const lifecycle of lifecycles) {
352-
lifecycle.onAdoptionFinalizing();
353-
}
354-
for (const lifecycle of lifecycles) {
355-
await lifecycle.onAdopted();
356-
}
343+
transportLifecycle.onAdoptionFinalizing();
344+
await transportLifecycle.onAdopted();
357345
terminal = "adopted";
358346
} catch (error) {
359347
await ensureAbandoned().catch(() => undefined);

‎extensions/imessage/src/monitor/ingress.test.ts‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,13 @@
22
import fs from "node:fs/promises";
33
import os from "node:os";
44
import path from "node:path";
5+
import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime";
56
import {
67
closeOpenClawStateDatabaseForTest,
78
createChannelIngressQueueForTests,
89
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
910
import { afterEach, describe, expect, it, vi } from "vitest";
10-
import {
11-
buildIMessageFlushIngressLifecycle,
12-
createIMessageDurableIngress,
13-
type IMessageIngressLifecycle,
14-
} from "./ingress.js";
11+
import { createIMessageDurableIngress, type IMessageIngressLifecycle } from "./ingress.js";
1512

1613
type IMessageIngressQueue = NonNullable<
1714
Parameters<typeof createIMessageDurableIngress>[0]["queue"]
@@ -399,7 +396,7 @@ describe("iMessage durable ingress", () => {
399396
it("fans merged adoption to every constituent claim", async () => {
400397
const first = lifecycle();
401398
const second = lifecycle();
402-
const merged = buildIMessageFlushIngressLifecycle([first, second]);
399+
const merged = fanInChannelIngressLifecycles([first, second]);
403400

404401
await merged.lifecycle?.onAdopted();
405402

@@ -410,7 +407,7 @@ describe("iMessage durable ingress", () => {
410407
it("completes every constituent claim when a flush has no dispatch", async () => {
411408
const first = lifecycle();
412409
const second = lifecycle();
413-
const merged = buildIMessageFlushIngressLifecycle([first, second]);
410+
const merged = fanInChannelIngressLifecycles([first, second]);
414411

415412
await merged.settle();
416413

‎extensions/imessage/src/monitor/ingress.ts‎

Lines changed: 0 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -167,70 +167,6 @@ type IMessageDurableIngress = {
167167
waitForIdle: () => Promise<void>;
168168
};
169169

170-
export function buildIMessageFlushIngressLifecycle(
171-
lifecycles: readonly IMessageIngressLifecycle[],
172-
): {
173-
lifecycle: IMessageIngressLifecycle | undefined;
174-
settle: () => Promise<void>;
175-
abandon: () => Promise<void>;
176-
} {
177-
const first = lifecycles[0];
178-
if (!first) {
179-
return { lifecycle: undefined, settle: async () => {}, abandon: async () => {} };
180-
}
181-
let handedOff = false;
182-
const adoptAll = async () => {
183-
for (const lifecycle of lifecycles) {
184-
await lifecycle.onAdopted();
185-
}
186-
};
187-
const abandonAll = async () => {
188-
for (const lifecycle of lifecycles) {
189-
await lifecycle.onAbandoned();
190-
}
191-
};
192-
return {
193-
lifecycle: {
194-
abortSignal:
195-
lifecycles.length === 1
196-
? first.abortSignal
197-
: AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal)),
198-
onAdopted: async () => {
199-
handedOff = true;
200-
await adoptAll();
201-
},
202-
onDeferred: () => {
203-
handedOff = true;
204-
for (const lifecycle of lifecycles) {
205-
lifecycle.onDeferred();
206-
}
207-
},
208-
onAdoptionFinalizing: () => {
209-
for (const lifecycle of lifecycles) {
210-
lifecycle.onAdoptionFinalizing();
211-
}
212-
},
213-
onAbandoned: async () => {
214-
handedOff = true;
215-
await abandonAll();
216-
},
217-
},
218-
// Gated/no-dispatch turns still consumed every raw row in the flush.
219-
settle: async () => {
220-
if (!handedOff) {
221-
handedOff = true;
222-
await adoptAll();
223-
}
224-
},
225-
abandon: async () => {
226-
if (!handedOff) {
227-
handedOff = true;
228-
await abandonAll();
229-
}
230-
},
231-
};
232-
}
233-
234170
export function createIMessageDurableIngress(options: {
235171
accountId: string;
236172
queue?: ChannelIngressQueue<IMessageIngressPayload>;

‎extensions/imessage/src/monitor/monitor-provider.ts‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import {
1414
type ChannelInboundTurnPlan,
1515
type ChannelInboundMediaInput,
1616
} from "openclaw/plugin-sdk/channel-inbound";
17+
import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime";
1718
import {
1819
bindIngressLifecycleToReplyOptions,
1920
createChannelMessageReplyPipeline,
@@ -96,11 +97,7 @@ import {
9697
resolveIMessageReactionContext,
9798
resolveIMessageInboundDecision,
9899
} from "./inbound-processing.js";
99-
import {
100-
buildIMessageFlushIngressLifecycle,
101-
createIMessageDurableIngress,
102-
type IMessageIngressLifecycle,
103-
} from "./ingress.js";
100+
import { createIMessageDurableIngress, type IMessageIngressLifecycle } from "./ingress.js";
104101
import { createLoopRateLimiter } from "./loop-rate-limiter.js";
105102
import { stageIMessageAttachments } from "./media-staging.js";
106103
import { createPollCommentFolder } from "./poll-comment.js";
@@ -620,7 +617,7 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P
620617
unitEntries: { message: IMessagePayload; ingressLifecycle?: IMessageIngressLifecycle }[],
621618
message: IMessagePayload,
622619
) => {
623-
const { lifecycle, settle, abandon } = buildIMessageFlushIngressLifecycle(
620+
const { lifecycle, settle, abandon } = fanInChannelIngressLifecycles(
624621
unitEntries.flatMap((entry) => (entry.ingressLifecycle ? [entry.ingressLifecycle] : [])),
625622
);
626623
try {

‎extensions/mattermost/src/mattermost/monitor-ingress.test.ts‎

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,13 @@
22
import fs from "node:fs/promises";
33
import os from "node:os";
44
import path from "node:path";
5+
import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime";
56
import {
67
closeOpenClawStateDatabaseForTest,
78
createChannelIngressQueueForTests,
89
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
910
import { afterEach, describe, expect, it, vi } from "vitest";
1011
import {
11-
buildMattermostFlushIngressLifecycle,
1212
createMattermostIngressMonitor,
1313
type MattermostIngressLifecycle,
1414
} from "./monitor-ingress.js";
@@ -493,10 +493,7 @@ describe("Mattermost merged ingress lifecycle", () => {
493493
it("fans adoption out to every constituent claim", async () => {
494494
const first = testLifecycle();
495495
const second = testLifecycle();
496-
const merged = buildMattermostFlushIngressLifecycle([
497-
{ turnAdoptionLifecycle: first.lifecycle },
498-
{ turnAdoptionLifecycle: second.lifecycle },
499-
]);
496+
const merged = fanInChannelIngressLifecycles([first.lifecycle, second.lifecycle]);
500497

501498
merged.lifecycle?.onDeferred();
502499
await merged.lifecycle?.onAdopted();
@@ -511,10 +508,7 @@ describe("Mattermost merged ingress lifecycle", () => {
511508
it("completes all claims when a gated flush never dispatches", async () => {
512509
const first = testLifecycle();
513510
const second = testLifecycle();
514-
const merged = buildMattermostFlushIngressLifecycle([
515-
{ turnAdoptionLifecycle: first.lifecycle },
516-
{ turnAdoptionLifecycle: second.lifecycle },
517-
]);
511+
const merged = fanInChannelIngressLifecycles([first.lifecycle, second.lifecycle]);
518512

519513
await merged.settle();
520514

0 commit comments

Comments
 (0)