fix(alerts): защитить ручной повтор очереди проверкой версии
This commit is contained in:
1 parent
89c31289bd
commit
7d0608bbce
16 files changed
+86
-17
No files matched your search
@@ -1,3 +1,4 @@
|
||||
import {executeOutboxRetry} from "../alerts-outbox-retry.js"
|
||||
import {readOutboxOverview} from "../alerts-outbox-overview.js"
|
||||
import {and,asc,desc,eq,inArray,lte,sql} from "drizzle-orm"
|
||||
import {db} from "../../db/index.js"
|
||||
@@ -44,6 +45,13 @@ export async function getAlertsOutboxOverview(){
|
||||
return readOutboxOverview<Parameters<Parameters<typeof db.transaction>[0]>[0]>({
|
||||
transaction:(work,options)=>db.transaction(work,options),
|
||||
counts:async context=>{const result=await context.execute(sql`SELECT status, COUNT(*)::bigint AS count FROM alert_outbox GROUP BY status`);return result.rows},
|
||||
items:async context=>{const rows=await context.select().from(alertOutbox).where(inArray(alertOutbox.status,["processing","unknown"])).orderBy(desc(alertOutbox.nextAttemptAt),desc(alertOutbox.id)).limit(20);return rows.map(row=>{const payload=row.payloadJson;const history=typeof payload==="object"&&payload!==null&&"history" in payload?payload.history:null;const ruleName=typeof history==="object"&&history!==null&&"ruleName" in history&&typeof history.ruleName==="string"?history.ruleName:null;return {id:row.id,status:row.status,ruleName,createdAt:row.createdAt,attemptedAt:row.nextAttemptAt}})},
|
||||
items:async context=>{const rows=await context.select().from(alertOutbox).where(inArray(alertOutbox.status,["processing","unknown"])).orderBy(desc(alertOutbox.nextAttemptAt),desc(alertOutbox.id)).limit(20);return rows.map(row=>{const payload=row.payloadJson;const history=typeof payload==="object"&&payload!==null&&"history" in payload?payload.history:null;const ruleName=typeof history==="object"&&history!==null&&"ruleName" in history&&typeof history.ruleName==="string"?history.ruleName:null;return {id:row.id,resolutionVersion:row.resolutionVersion,status:row.status,ruleName,createdAt:row.createdAt,attemptedAt:row.nextAttemptAt}})},
|
||||
})
|
||||
}
|
||||
|
||||
export async function retryUnknownOutbox(id:string,input:unknown){
|
||||
return executeOutboxRetry<Parameters<Parameters<typeof db.transaction>[0]>[0]>(id,input,{
|
||||
transaction:work=>db.transaction(work),
|
||||
update:async(tx,messageId,expectedVersion)=>{const [receipt]=await tx.update(alertOutbox).set({status:"pending",resolutionVersion:sql`resolution_version + 1`,nextAttemptAt:new Date().toISOString(),lastError:null}).where(and(eq(alertOutbox.id,messageId),eq(alertOutbox.status,"unknown"),eq(alertOutbox.channel,"telegram"),eq(alertOutbox.resolutionVersion,expectedVersion))).returning({id:alertOutbox.id,status:alertOutbox.status,resolutionVersion:alertOutbox.resolutionVersion});return receipt},
|
||||
})
|
||||
}
|
||||
@@ -10,6 +10,6 @@ test("queue counts keep unknown database states, reject unsafe counts and read o
|
||||
assert.equal(reads,2);assert.equal(result.counts.pending,3)
|
||||
})
|
||||
test("queue incomplete held list or failed read never produces empty success",async()=>{
|
||||
for(const counts of [[{status:"unknown",count:1}],[]])await assert.rejects(readOutboxOverview({transaction:async work=>work({}),counts:async()=>counts,items:async()=>counts.length?[]:[{id:"a",status:"unknown",ruleName:null,createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T00:00:00Z"}]}),/не подтверждено/)
|
||||
for(const counts of [[{status:"unknown",count:1}],[]])await assert.rejects(readOutboxOverview({transaction:async work=>work({}),counts:async()=>counts,items:async()=>counts.length?[]:[{id:"a",resolutionVersion:0,status:"unknown",ruleName:null,createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T00:00:00Z"}]}),/не подтверждено/)
|
||||
await assert.rejects(readOutboxOverview({transaction:async work=>work({}),counts:async()=>{throw Error("read failed")},items:async()=>[]}),/read failed/)
|
||||
})
|
||||
@@ -0,0 +1,16 @@
|
||||
import assert from "node:assert/strict"
|
||||
import {test} from "node:test"
|
||||
import {executeOutboxRetry,OutboxRetryConflict} from "./alerts-outbox-retry.js"
|
||||
const input={expectedVersion:0,acknowledgedDuplicateRisk:true}
|
||||
test("manual queue retry requires acknowledgement and CAS claims unknown once; processing is held",async()=>{
|
||||
let status="unknown",version=0,updates=0;const context={snapshot:true}
|
||||
const deps={transaction:async(work:(ctx:typeof context)=>Promise<Awaited<ReturnType<typeof executeOutboxRetry>>>)=>work(context),update:async(ctx:typeof context,id:string,expected:number)=>{assert.equal(ctx,context);if(status!=="unknown"||version!==expected)return undefined;updates++;status="pending";version++;return {id,status,resolutionVersion:version}}}
|
||||
const results=await Promise.allSettled([executeOutboxRetry("a",input,deps),executeOutboxRetry("a",input,deps)]);assert.equal(results.filter(x=>x.status==="fulfilled").length,1);assert.equal(updates,1);assert.equal(version,1)
|
||||
status="processing";await assert.rejects(executeOutboxRetry("a",{...input,expectedVersion:1},deps),OutboxRetryConflict);assert.equal(updates,1)
|
||||
status="unknown";await assert.rejects(executeOutboxRetry("a",input,deps),OutboxRetryConflict);assert.equal(updates,1)
|
||||
for(const raw of [{},{...input,acknowledgedDuplicateRisk:false},{...input,expectedVersion:2147483647}])await assert.rejects(executeOutboxRetry("a",raw,deps));assert.equal(updates,1)
|
||||
})
|
||||
test("manual queue retry rejects failed or mismatched returning inside transaction without sending Telegram",async()=>{
|
||||
for(const row of [{id:"other",status:"pending",resolutionVersion:1},{id:"a",status:"sent",resolutionVersion:1},{id:"a",status:"pending",resolutionVersion:2},{}]){let rolledBack=false;await assert.rejects(executeOutboxRetry("a",input,{transaction:async work=>{try{return await work({})}catch(e){rolledBack=true;throw e}},update:async()=>row}),/не подтверждена/);assert.equal(rolledBack,true)}
|
||||
await assert.rejects(executeOutboxRetry("a",input,{transaction:async work=>work({}),update:async()=>{throw Error("DB failed")}}),/DB failed/)
|
||||
})
|
||||
@@ -0,0 +1,13 @@
|
||||
import {alertsOutboxRetrySchema,alertsOutboxRetryReceiptSchema} from "@mmapp/contracts/alerts"
|
||||
export class OutboxRetryConflict extends Error{constructor(){super("Запись очереди изменилась или не имеет неизвестного результата. Обновите снимок.")}}
|
||||
export async function executeOutboxRetry<C>(id:string,raw:unknown,deps:{transaction:(work:(context:C)=>Promise<ReturnType<typeof alertsOutboxRetryReceiptSchema.parse>>)=>Promise<ReturnType<typeof alertsOutboxRetryReceiptSchema.parse>>;update:(context:C,id:string,expectedVersion:number)=>Promise<unknown>}){
|
||||
const input=alertsOutboxRetrySchema.parse(raw)
|
||||
if(!id.trim()||id.length>200)throw Error("Некорректный идентификатор сообщения")
|
||||
return deps.transaction(async context=>{
|
||||
const row=await deps.update(context,id,input.expectedVersion)
|
||||
if(row===undefined)throw new OutboxRetryConflict()
|
||||
const parsed=alertsOutboxRetryReceiptSchema.safeParse({ok:true,...(typeof row==="object"&&row!==null?row:{})})
|
||||
if(!parsed.success||parsed.data.id!==id||parsed.data.resolutionVersion!==input.expectedVersion+1)throw Error("Постановка сообщения в очередь не подтверждена")
|
||||
return parsed.data
|
||||
})
|
||||
}
|
||||
Reference in new issue
Block a user