feat(server): 委派回收并发防重 + 最小门禁闸门
- delegationRevokeJob 改用 FOR UPDATE SKIP LOCKED + UPDATE RETURNING 原子认领 - regression-workflow-actions 新增 RUN_DELEGATION_CONCURRENCY_CHECK 并发校验开关 - README 补充最小门禁与并发语义说明
这个提交包含在:
@@ -1,62 +1,78 @@
|
||||
const cron = require("node-cron");
|
||||
|
||||
async function claimAndRevokeExpiredDelegations(tx, batchSize) {
|
||||
return tx.$queryRaw`
|
||||
WITH picked AS (
|
||||
SELECT id
|
||||
FROM "delegation_grants"
|
||||
WHERE revoked_at IS NULL
|
||||
AND valid_to IS NOT NULL
|
||||
AND valid_to <= NOW()
|
||||
ORDER BY valid_to ASC
|
||||
LIMIT ${batchSize}
|
||||
FOR UPDATE SKIP LOCKED
|
||||
)
|
||||
UPDATE "delegation_grants" AS dg
|
||||
SET revoked_at = NOW()
|
||||
FROM picked
|
||||
WHERE dg.id = picked.id
|
||||
AND dg.revoked_at IS NULL
|
||||
RETURNING dg.id, dg.principal_employee_id, dg.delegate_employee_id, dg.revoked_at
|
||||
`;
|
||||
}
|
||||
|
||||
async function revokeExpiredDelegations(prisma, logger, options = {}) {
|
||||
const batchSize = options.batchSize ?? 200;
|
||||
const batchSize = Math.max(1, Number(options.batchSize ?? 200));
|
||||
let total = 0;
|
||||
let round = 0;
|
||||
|
||||
while (true) {
|
||||
const now = new Date();
|
||||
const expired = await prisma.delegationGrant.findMany({
|
||||
where: {
|
||||
revoked_at: null,
|
||||
valid_to: {
|
||||
not: null,
|
||||
lte: now,
|
||||
},
|
||||
},
|
||||
orderBy: { valid_to: "asc" },
|
||||
take: batchSize,
|
||||
select: {
|
||||
id: true,
|
||||
principal_employee_id: true,
|
||||
delegate_employee_id: true,
|
||||
},
|
||||
round += 1;
|
||||
|
||||
const revokedRows = await prisma.$transaction(async (tx) => {
|
||||
const claimed = await claimAndRevokeExpiredDelegations(tx, batchSize);
|
||||
if (!claimed.length) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const requestTime = Date.now();
|
||||
await tx.auditLog.createMany({
|
||||
data: claimed.map((item, index) => {
|
||||
const revokedAt = item.revoked_at ? new Date(item.revoked_at).getTime() : requestTime;
|
||||
return {
|
||||
request_id: `job:delegation-revoke:${revokedAt}:${item.id}:${index}`,
|
||||
actor_id: item.principal_employee_id,
|
||||
action: "DELEGATION_AUTO_REVOKED",
|
||||
target_type: "delegation_grant",
|
||||
target_id: item.id,
|
||||
ip_address: null,
|
||||
user_agent: "system/node-cron",
|
||||
extra: {
|
||||
principalEmployeeId: item.principal_employee_id,
|
||||
delegateEmployeeId: item.delegate_employee_id,
|
||||
},
|
||||
};
|
||||
}),
|
||||
});
|
||||
|
||||
return claimed;
|
||||
});
|
||||
|
||||
if (!expired.length) {
|
||||
if (!revokedRows.length) {
|
||||
break;
|
||||
}
|
||||
|
||||
await prisma.$transaction(async (tx) => {
|
||||
const ids = expired.map((item) => item.id);
|
||||
await tx.delegationGrant.updateMany({
|
||||
where: {
|
||||
id: { in: ids },
|
||||
revoked_at: null,
|
||||
},
|
||||
data: {
|
||||
revoked_at: now,
|
||||
},
|
||||
});
|
||||
total += revokedRows.length;
|
||||
|
||||
await tx.auditLog.createMany({
|
||||
data: expired.map((item) => ({
|
||||
request_id: `job:delegation-revoke:${now.getTime()}:${item.id}`,
|
||||
actor_id: item.principal_employee_id,
|
||||
action: "DELEGATION_AUTO_REVOKED",
|
||||
target_type: "delegation_grant",
|
||||
target_id: item.id,
|
||||
ip_address: null,
|
||||
user_agent: "system/node-cron",
|
||||
extra: {
|
||||
principalEmployeeId: item.principal_employee_id,
|
||||
delegateEmployeeId: item.delegate_employee_id,
|
||||
},
|
||||
})),
|
||||
});
|
||||
});
|
||||
|
||||
total += expired.length;
|
||||
logger.info(
|
||||
{
|
||||
round,
|
||||
claimed: revokedRows.length,
|
||||
updated: revokedRows.length,
|
||||
audited: revokedRows.length,
|
||||
},
|
||||
"Expired delegations batch processed",
|
||||
);
|
||||
}
|
||||
|
||||
if (total > 0) {
|
||||
|
||||
在新工单中引用
屏蔽一个用户