From 7ebd9aa63f6c7c2d95b0af4a2a11075f3423c8e5 Mon Sep 17 00:00:00 2001 From: RhenCloud Date: Sat, 15 Aug 2026 12:04:37 +0800 Subject: [PATCH] fix: prevent race conditions with KV locks in dispatch and group provisioning --- server/lib/core/dispatch.ts | 163 +++++++++++++++------ server/lib/drivers/discord/interactions.ts | 22 ++- server/lib/web/groups.ts | 77 ++++++---- server/lib/webhook.ts | 11 +- 4 files changed, 194 insertions(+), 79 deletions(-) diff --git a/server/lib/core/dispatch.ts b/server/lib/core/dispatch.ts index 0fab5bb..0b686ef 100644 --- a/server/lib/core/dispatch.ts +++ b/server/lib/core/dispatch.ts @@ -197,60 +197,131 @@ export async function dispatchEvent( if (message.updateKey) { const groupPrefix = route.groupId ? `${route.groupId}:` : ""; const kvKey = `msg:${groupPrefix}${route.id}:${message.updateKey}:${targetStr}`; - const existingId = await env.KV.get(kvKey); - if (existingId) { - result = await driver.edit(message, target, env, existingId); - if (result.ok) { - attempts.push({ - groupId: route.groupId, - routeId: route.id, - routeName: route.name, - target: targetStr, - ok: true, - }); - await recordSend(env.DB, { - ...base, - ok: true, - status: result.status, - messageId: existingId, - platform: driver.id, - attempts: result.attempts, - durationMs: Date.now() - started, - errorCode: result.errorCode, - }); + const lockKey = `msg:lock:${kvKey}`; + + // Acquire a short-lived lock so concurrent events for the same + // updateKey don't race (both read null, both send, the later put + // overwrites). KV is eventually consistent so the lock is + // best-effort; a retry loop further shrinks the window. + let locked = false; + for (let attempt = 0; attempt < 3; attempt++) { + const holder = await env.KV.get(lockKey); + if (holder) { + const existing = await env.KV.get(kvKey); + if (existing) { + result = await driver.edit(message, target, env, existing); + if (result.ok || /not modified/i.test(result.error ?? "")) { + const ok = result.ok || /not modified/i.test(result.error ?? ""); + attempts.push({ + groupId: route.groupId, + routeId: route.id, + routeName: route.name, + target: targetStr, + ok: true, + }); + await recordSend(env.DB, { + ...base, + ok: true, + status: result.status, + messageId: existing, + platform: driver.id, + attempts: result.attempts, + durationMs: Date.now() - started, + errorCode: result.errorCode, + }); + if (!ok) await env.KV.delete(kvKey); + continue; + } + await env.KV.delete(kvKey); + break; + } + await new Promise((r) => setTimeout(r, 50 * (attempt + 1))); continue; } - if (/not modified/i.test(result.error ?? "")) { - attempts.push({ - groupId: route.groupId, - routeId: route.id, - routeName: route.name, - target: targetStr, - ok: true, - }); - await recordSend(env.DB, { - ...base, - ok: true, - status: result.status, - messageId: existingId, - platform: driver.id, - attempts: result.attempts, - durationMs: Date.now() - started, - errorCode: result.errorCode, - }); - continue; - } - await env.KV.delete(kvKey); + await env.KV.put(lockKey, "1", { expirationTtl: 60 }); + locked = true; + break; } - result = await driver.send(message, target, env); - if (result.ok && result.messageId) { - await env.KV.put(kvKey, result.messageId, { expirationTtl: 604800 }); + + try { + const existingId = await env.KV.get(kvKey); + if (existingId) { + result = await driver.edit(message, target, env, existingId); + if (result.ok) { + attempts.push({ + groupId: route.groupId, + routeId: route.id, + routeName: route.name, + target: targetStr, + ok: true, + }); + await recordSend(env.DB, { + ...base, + ok: true, + status: result.status, + messageId: existingId, + platform: driver.id, + attempts: result.attempts, + durationMs: Date.now() - started, + errorCode: result.errorCode, + }); + continue; + } + if (/not modified/i.test(result.error ?? "")) { + attempts.push({ + groupId: route.groupId, + routeId: route.id, + routeName: route.name, + target: targetStr, + ok: true, + }); + await recordSend(env.DB, { + ...base, + ok: true, + status: result.status, + messageId: existingId, + platform: driver.id, + attempts: result.attempts, + durationMs: Date.now() - started, + errorCode: result.errorCode, + }); + continue; + } + await env.KV.delete(kvKey); + } + result = await driver.send(message, target, env); + if (result.ok && result.messageId) { + await env.KV.put(kvKey, result.messageId, { expirationTtl: 604800 }); + } + } finally { + if (locked) await env.KV.delete(lockKey); } } else { result = await driver.send(message, target, env); } const durationMs = Date.now() - started; - if (!result.ok) throw new Error(result.error ?? "Send failed"); + if (!result.ok) { + const error = result.error ?? "Send failed"; + attempts.push({ + groupId: route.groupId, + routeId: route.id, + routeName: route.name, + target: targetStr, + ok: false, + error, + }); + await recordSend(env.DB, { + ...base, + ok: false, + error, + durationMs, + status: result.status, + platform: driver.id, + attempts: result.attempts, + errorCode: result.errorCode, + }); + continue; + } attempts.push({ groupId: route.groupId, routeId: route.id, diff --git a/server/lib/drivers/discord/interactions.ts b/server/lib/drivers/discord/interactions.ts index 5b8f8d0..db8c523 100644 --- a/server/lib/drivers/discord/interactions.ts +++ b/server/lib/drivers/discord/interactions.ts @@ -33,6 +33,19 @@ const GITHUB_COMMENT_RE = /github\.com\/([^/\s]+)\/([^/\s]+)\/(?:issues|pull)\/\d+#issuecomment-(\d+)/; const GITHUB_ISSUE_RE = /github\.com\/([^/\s]+)\/([^/\s]+)\/(?:issues|pull)\/(\d+)/; +/** + * Split a pipe-delimited custom-id suffix into [owner, repo, number]. The + * number is always the last segment (arbitrary-length owner/repo can contain + * pipes on non-GitHub forges), so we split from the right. + */ +function splitThree(rest: string): [string, string, string] { + const parts = rest.split("|"); + const number = parts.pop(); + const repo = parts.pop(); + const owner = parts.join("|"); + return [owner ?? "", repo ?? "", number ?? ""]; +} + interface Interaction { id: string; token: string; @@ -101,7 +114,8 @@ export async function handleInteractionRequest(request: Request, env: Env): Prom ); return new Response("Invalid signature", { status: 401 }); } - if (Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp)) > TIMESTAMP_TOLERANCE_SECONDS) { + const ts = Number(timestamp); + if (!Number.isFinite(ts) || Math.abs(Math.floor(Date.now() / 1000) - ts) > TIMESTAMP_TOLERANCE_SECONDS) { return new Response("Invalid signature", { status: 401 }); } @@ -287,7 +301,7 @@ async function handleButton( return; } - const [owner, repo, number] = rest.split("|"); + const [owner, repo, number] = splitThree(rest); if (!owner || !repo || !number) return; // Acknowledge first (deferred, ephemeral) so the clicker sees a spinner @@ -494,7 +508,7 @@ async function modalSubmit( // ghc|add|owner|repo|issueNumber if (customId.startsWith(MODAL_ADD)) { - const [owner, repo, number] = customId.slice(MODAL_ADD.length).split("|"); + const [owner, repo, number] = splitThree(customId.slice(MODAL_ADD.length)); if (!owner || !repo || !number) return respond(env, id, token, "内部错误：无法解析目标 issue。"); try { @@ -514,7 +528,7 @@ async function modalSubmit( // ghc|edit|owner|repo|commentId if (customId.startsWith(MODAL_EDIT)) { - const [owner, repo, commentId] = customId.slice(MODAL_EDIT.length).split("|"); + const [owner, repo, commentId] = splitThree(customId.slice(MODAL_EDIT.length)); if (!owner || !repo || !commentId) return respond(env, id, token, "内部错误：无法解析目标评论。"); try { diff --git a/server/lib/web/groups.ts b/server/lib/web/groups.ts index 61e490b..80c2068 100644 --- a/server/lib/web/groups.ts +++ b/server/lib/web/groups.ts @@ -120,36 +120,57 @@ export async function ensureInstallationGroup( installationId: number, accountLogin: string, ): Promise { - const groups = await loadGroups(kv); - const existing = groups.find((g) => g.installationId === installationId); - if (existing) return existing; - - const login = accountLogin.trim().toLowerCase(); - const candidates = groups.filter( - (g) => - g.installationId == null && - login.length > 0 && - (g.owners ?? []).some((o) => o.trim().toLowerCase() === login), - ); - if (candidates.length > 0) { - const next = groups.map((g) => (candidates.includes(g) ? { ...g, installationId } : g)); - await saveGroups(kv, next); - return next.find((g) => g.id === candidates[0]!.id) ?? null; + // A per-installation lock serializes concurrent `installation.created` + // webhooks: without it two requests both pass the "no group bound" check, + // both create `inst-{id}`, and the second save overwrites the first. + const lockKey = `inst:lock:${installationId}`; + for (let attempt = 0; attempt < 5; attempt++) { + const holder = await kv.get(lockKey); + if (!holder) break; + const groups = await loadGroups(kv); + const existing = groups.find((g) => g.installationId === installationId); + if (existing) return existing; + await new Promise((r) => setTimeout(r, 40 * (attempt + 1))); } - const gid = `inst-${installationId}`; - const dedicated = groups.find((g) => g.id === gid); - const group: Group = { - id: gid, - name: accountLogin.trim() || `Installation ${installationId}`, - adminIds: [], - installationId, - }; - await saveGroups( - kv, - dedicated ? groups.map((g) => (g.id === gid ? { ...g, ...group } : g)) : [...groups, group], - ); - return group; + let locked = false; + try { + await kv.put(lockKey, "1", { expirationTtl: 60 }); + locked = true; + + const groups = await loadGroups(kv); + const existing = groups.find((g) => g.installationId === installationId); + if (existing) return existing; + + const login = accountLogin.trim().toLowerCase(); + const candidates = groups.filter( + (g) => + g.installationId == null && + login.length > 0 && + (g.owners ?? []).some((o) => o.trim().toLowerCase() === login), + ); + if (candidates.length > 0) { + const next = groups.map((g) => (candidates.includes(g) ? { ...g, installationId } : g)); + await saveGroups(kv, next); + return next.find((g) => g.id === candidates[0]!.id) ?? null; + } + + const gid = `inst-${installationId}`; + const dedicated = groups.find((g) => g.id === gid); + const group: Group = { + id: gid, + name: accountLogin.trim() || `Installation ${installationId}`, + adminIds: [], + installationId, + }; + await saveGroups( + kv, + dedicated ? groups.map((g) => (g.id === gid ? { ...g, ...group } : g)) : [...groups, group], + ); + return group; + } finally { + if (locked) await kv.delete(lockKey); + } } export interface AccessScope { diff --git a/server/lib/webhook.ts b/server/lib/webhook.ts index fc0b63f..2119150 100644 --- a/server/lib/webhook.ts +++ b/server/lib/webhook.ts @@ -50,7 +50,16 @@ export async function processWebhook( return { status: 400, body: { error: "Unknown webhook provider" } }; } - if (!(await provider.verify(body, headers, effectiveEnv))) { + let verified = false; + try { + verified = await provider.verify(body, headers, effectiveEnv); + } catch (err) { + // A malformed secret or an unavailable crypto implementation must fail as + // a clean 401, never an uncaught 500. + log.warn({ provider: provider.id, err: String(err) }, "Webhook signature verification failed"); + return { status: 401, body: { error: "Invalid signature" } }; + } + if (!verified) { // Log the actual cause: a missing provider secret is a deployment problem, // while a mismatched signature usually means the sender used the wrong secret. const secret =