@@ -7,19 +7,28 @@ import { resolveStateDir } from "openclaw/plugin-sdk/state-paths";
7788const SPOOL_VERSION = 1;
99export const TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS = 6 * 60 * 60 * 1000;
10+const TELEGRAM_SPOOLED_UPDATE_PROCESS_ID = `${process.pid}:${randomUUID()}`;
11+12+type TelegramSpooledUpdateClaimOwner = {
13+processId: string;
14+processPid: number;
15+claimedAt: number;
16+};
10171118type TelegramSpooledUpdatePayload = {
1219version: number;
1320updateId: number;
1421receivedAt: number;
1522update: unknown;
23+claim?: TelegramSpooledUpdateClaimOwner;
1624};
17251826export type TelegramSpooledUpdate = {
1927updateId: number;
2028path: string;
2129update: unknown;
2230receivedAt: number;
31+claim?: TelegramSpooledUpdateClaimOwner;
2332};
24332534export type ClaimedTelegramSpooledUpdate = TelegramSpooledUpdate & {
@@ -105,12 +114,59 @@ function parseSpooledUpdate(value: unknown, filePath: string): TelegramSpooledUp
105114if (payload.version !== SPOOL_VERSION || !isValidUpdateId(payload.updateId)) {
106115return null;
107116}
108-return {
117+const update: TelegramSpooledUpdate = {
109118updateId: payload.updateId,
110119path: filePath,
111120update: payload.update,
112121receivedAt: typeof payload.receivedAt === "number" ? payload.receivedAt : 0,
113122};
123+if (
124+payload.claim &&
125+typeof payload.claim.processId === "string" &&
126+isValidUpdateId(payload.claim.processPid) &&
127+typeof payload.claim.claimedAt === "number"
128+) {
129+update.claim = payload.claim;
130+}
131+return update;
132+}
133+134+function buildClaimedPayload(update: TelegramSpooledUpdate): TelegramSpooledUpdatePayload {
135+return {
136+version: SPOOL_VERSION,
137+updateId: update.updateId,
138+receivedAt: update.receivedAt,
139+update: update.update,
140+claim: {
141+processId: TELEGRAM_SPOOLED_UPDATE_PROCESS_ID,
142+processPid: process.pid,
143+claimedAt: Date.now(),
144+},
145+};
146+}
147+148+function processExists(pid: number): boolean {
149+try {
150+process.kill(pid, 0);
151+return true;
152+} catch (err) {
153+return (err as { code?: string }).code !== "ESRCH";
154+}
155+}
156+157+function isFreshClaimOwner(claim: TelegramSpooledUpdateClaimOwner): boolean {
158+return Date.now() - claim.claimedAt < TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS;
159+}
160+161+export function isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess(
162+claim: ClaimedTelegramSpooledUpdate,
163+): boolean {
164+return Boolean(
165+claim.claim &&
166+claim.claim.processId !== TELEGRAM_SPOOLED_UPDATE_PROCESS_ID &&
167+isFreshClaimOwner(claim.claim) &&
168+processExists(claim.claim.processPid),
169+);
114170}
115171116172export async function writeTelegramSpooledUpdate(params: {
@@ -183,26 +239,32 @@ export async function claimTelegramSpooledUpdate(
183239update: TelegramSpooledUpdate,
184240): Promise<ClaimedTelegramSpooledUpdate | null> {
185241const claimedPath = processingPath(path.dirname(update.path), update.updateId);
186-try {
187-// A hard link is an atomic non-overwriting claim in the same spool directory.
188-await fs.link(update.path, claimedPath);
189-} catch (err) {
190-const code = (err as { code?: string }).code;
191-if (code === "ENOENT") {
192-return null;
193-}
194-if (code === "EEXIST") {
195-await unlinkIfPresent(update.path);
196-return null;
197-}
198-throw err;
199-}
242+const holdPath = path.join(
243+path.dirname(update.path),
244+`${spoolFileName(update.updateId)}.${randomUUID()}.claim`,
245+);
246+const tempPath = path.join(
247+path.dirname(update.path),
248+`${processingFileName(update.updateId)}.${randomUUID()}.tmp`,
249+);
200250try {
201251const claimedAt = new Date();
252+await fs.writeFile(tempPath, `${JSON.stringify(buildClaimedPayload(update))}\n`, {
253+mode: 0o600,
254+});
255+await fs.link(update.path, holdPath);
256+await fs.link(tempPath, claimedPath);
257+await unlinkIfPresent(tempPath);
258+await unlinkIfPresent(holdPath);
202259await fs.utimes(claimedPath, claimedAt, claimedAt);
203260await unlinkIfPresent(update.path);
204261} catch (err) {
205-await unlinkIfPresent(claimedPath);
262+const code = (err as { code?: string }).code;
263+await unlinkIfPresent(tempPath);
264+await unlinkIfPresent(holdPath);
265+if (code === "ENOENT" || code === "EEXIST") {
266+return null;
267+}
206268throw err;
207269}
208270return {
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。