diff --git a/backend/drizzle/0009_alert_outbox_resolution.sql b/backend/drizzle/0009_alert_outbox_resolution.sql new file mode 100644 index 00000000..fbedccf1 --- /dev/null +++ b/backend/drizzle/0009_alert_outbox_resolution.sql @@ -0,0 +1,4 @@ +BEGIN; +ALTER TABLE alert_outbox ADD COLUMN IF NOT EXISTS resolution_version integer NOT NULL DEFAULT 0; +ALTER TABLE alert_outbox ADD CONSTRAINT alert_outbox_resolution_version_check CHECK (resolution_version >= 0); +COMMIT; diff --git a/backend/package.json b/backend/package.json index 4fce8ec8..9d711781 100644 --- a/backend/package.json +++ b/backend/package.json @@ -6,7 +6,7 @@ "scripts": { "dev": "tsx watch src/index.ts", "build": "npm run build -w @mmapp/contracts && tsc", - "test:alerts-telegram": "tsx src/services/alerts-outbox-overview.test.ts && tsx src/services/alert-engine/outbox-execution.test.ts && tsx src/services/alerts-telegram-delivery.test.ts && tsx src/services/alerts-revision.test.ts && tsx src/services/alerts-overview.test.ts && tsx ../packages/contracts/src/alerts-normalization.test.ts && tsx src/services/alerts-config-execution.test.ts && tsx src/services/alerts-telegram-settings.test.ts && tsx src/services/alerts-telegram-send.test.ts && tsx src/routes/alerts.test.ts && tsx src/plugins/auth.alerts.test.ts", + "test:alerts-telegram": "tsx src/services/alerts-outbox-retry.test.ts && tsx src/services/alerts-outbox-overview.test.ts && tsx src/services/alert-engine/outbox-execution.test.ts && tsx src/services/alerts-telegram-delivery.test.ts && tsx src/services/alerts-revision.test.ts && tsx src/services/alerts-overview.test.ts && tsx ../packages/contracts/src/alerts-normalization.test.ts && tsx src/services/alerts-config-execution.test.ts && tsx src/services/alerts-telegram-settings.test.ts && tsx src/services/alerts-telegram-send.test.ts && tsx src/routes/alerts.test.ts && tsx src/plugins/auth.alerts.test.ts", "test:alert-engine": "tsx src/services/alert-engine/groups.test.ts && tsx src/services/alert-engine/evaluate-rule.test.ts && tsx src/services/alert-engine/confirm-stability.test.ts", "start": "node dist/index.js", "db:generate": "drizzle-kit generate", diff --git a/backend/src/db/schema.ts b/backend/src/db/schema.ts index 62d5f4a8..9b7f86fa 100644 --- a/backend/src/db/schema.ts +++ b/backend/src/db/schema.ts @@ -607,6 +607,7 @@ export const alertOutbox = pgTable("alert_outbox", { dedupeKey: text("dedupe_key").notNull(), channel: text("channel", { enum: ["telegram"] }).notNull().default("telegram"), status: text("status", { enum: ["pending", "processing", "unknown", "sent", "failed"] }).notNull().default("pending"), + resolutionVersion: integer("resolution_version").notNull().default(0), retryCount: integer("retry_count").notNull().default(0), maxRetries: integer("max_retries").notNull().default(3), nextAttemptAt: ts("next_attempt_at").notNull(), diff --git a/backend/src/plugins/auth.alerts.test.ts b/backend/src/plugins/auth.alerts.test.ts index 67512c21..32b1a97a 100644 --- a/backend/src/plugins/auth.alerts.test.ts +++ b/backend/src/plugins/auth.alerts.test.ts @@ -7,12 +7,12 @@ process.env.AUTH_REQUIRED="true";process.env.AUTH_JWT_SECRET="evolution-fixture- const {default:authPlugin,requireAuth}=await import("./auth.js") const {createAlertsRoutes,defaultAlertsDependencies}=await import("../routes/alerts.js") const app=Fastify();await app.register(authPlugin);app.addHook("preHandler",requireAuth) -let writes=0,sends=0,reads=0 -await app.register(createAlertsRoutes({...defaultAlertsDependencies,outbox:async()=>({counts:{total:0,pending:0,processing:0,unknown:0,sent:0,failed:0,unrecognized:0},items:[],itemsLimit:20}),overview:async()=>({revision,telegram:{chatId:"123",tokenConfigured:true,connected:true,messageThreadId:null},groups:[],rules:[],history:[],meta:{servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}}),writeTelegram:async()=>{writes++;return {chatId:"123",tokenConfigured:true,messageThreadId:null}},readTelegram:async()=>{reads++;return {botToken:"fixture-only",chatId:"123",messageThreadId:null}},send:async()=>{sends++;return {ok:true}},replace:async()=>{writes++;return {ok:true as const,revision,groups:[],rules:[]}}}),{prefix:"/api"}) +let writes=0,sends=0,reads=0,retries=0 +await app.register(createAlertsRoutes({...defaultAlertsDependencies,retryOutbox:async(id)=>{retries++;return {ok:true as const,id,status:"pending" as const,resolutionVersion:1}},outbox:async()=>({counts:{total:0,pending:0,processing:0,unknown:0,sent:0,failed:0,unrecognized:0},items:[],itemsLimit:20}),overview:async()=>({revision,telegram:{chatId:"123",tokenConfigured:true,connected:true,messageThreadId:null},groups:[],rules:[],history:[],meta:{servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}}),writeTelegram:async()=>{writes++;return {chatId:"123",tokenConfigured:true,messageThreadId:null}},readTelegram:async()=>{reads++;return {botToken:"fixture-only",chatId:"123",messageThreadId:null}},send:async()=>{sends++;return {ok:true}},replace:async()=>{writes++;return {ok:true as const,revision,groups:[],rules:[]}}}),{prefix:"/api"}) for(const permission of ["mm:alerts:read","mm:alerts:write","mm:alerts:admin","mm:settings:admin"]){ const token=await new SignJWT({sub:"fixture",apps:["mm"],permissions:[permission]}).setProtectedHeader({alg:"HS256"}).setIssuer("https://auth.test.local").setExpirationTime("1h").sign(new TextEncoder().encode("evolution-fixture-secret")) - for(const [method,url] of [["GET","/api/alerts"],["GET","/api/alerts/outbox"],["PUT","/api/alerts/telegram"],["POST","/api/alerts/telegram/test"],["PUT","/api/alerts/rules"]] as const){const permitted=permission==="mm:alerts:write"||permission==="mm:alerts:admin"||method==="GET"&&permission==="mm:alerts:read";const response=await app.inject({method,url,headers:{authorization:"Bearer "+token,"if-match":'"'+revision+'"'},...(method==="PUT"?{payload:url.endsWith("rules")?{rules:[],groups:[]}:{chatId:"123"}}:{})});assert.equal(response.statusCode,permitted?200:403);assert.equal(response.body.includes("fixture-only"),false)} + for(const [method,url] of [["GET","/api/alerts"],["GET","/api/alerts/outbox"],["PUT","/api/alerts/telegram"],["POST","/api/alerts/telegram/test"],["POST","/api/alerts/outbox/a/retry"],["PUT","/api/alerts/rules"]] as const){const permitted=permission==="mm:alerts:write"||permission==="mm:alerts:admin"||method==="GET"&&permission==="mm:alerts:read";const response=await app.inject({method,url,headers:{authorization:"Bearer "+token,"if-match":'"'+revision+'"'},...(url.endsWith("/retry")?{payload:{expectedVersion:0,acknowledgedDuplicateRisk:true}}:method==="PUT"?{payload:url.endsWith("rules")?{rules:[],groups:[]}:{chatId:"123"}}:{})});assert.equal(response.statusCode,permitted?200:403);assert.equal(response.body.includes("fixture-only"),false)} } -assert.equal(writes,4);assert.equal(sends,2);assert.equal(reads,2) +assert.equal(retries,2);assert.equal(writes,4);assert.equal(sends,2);assert.equal(reads,2) for(const [method,url] of [["GET","/api/alerts"],["GET","/api/alerts/outbox"],["POST","/api/alerts/telegram/test"]] as const)assert.equal((await app.inject({method,url})).statusCode,401) assert.equal(sends,2);await app.close();console.log("Alerts JWT: real read/write/admin gates before save/send/read, no token echo") diff --git a/backend/src/routes/alerts.test.ts b/backend/src/routes/alerts.test.ts index 4d15474c..28078872 100644 --- a/backend/src/routes/alerts.test.ts +++ b/backend/src/routes/alerts.test.ts @@ -1,12 +1,13 @@ +import {OutboxRetryConflict} from "../services/alerts-outbox-retry.js" import {alertsConfigRevision} from "../services/alerts-revision.js" const revision=alertsConfigRevision({groups:[],rules:[]}) import assert from "node:assert/strict" import Fastify from "fastify" import {createAlertsRoutes,defaultAlertsDependencies} from "./alerts.js" import {executeTelegramSettingsUpdate} from "../services/alerts-telegram-settings.js" -let replacements=0,replaceConflict=false,replaceStale=false,overviewMalformed=false,outboxMalformed=false +let replacements=0,replaceConflict=false,replaceStale=false,overviewMalformed=false,outboxMalformed=false,retries=0,retryConflict=false,retryMalformed=false let stored={botToken:"fixture-secret",chatId:"123",messageThreadId:10 as number|null},reads=0,writes=0,sends=0,failSend=false,failWrite=false -const app=Fastify();await app.register(createAlertsRoutes({...defaultAlertsDependencies,outbox:async()=>outboxMalformed?{}:({counts:{total:0,pending:0,processing:0,unknown:0,sent:0,failed:0,unrecognized:0},items:[],itemsLimit:20}),overview:async()=>overviewMalformed?{}:({revision,telegram:{chatId:stored.chatId,tokenConfigured:!!stored.botToken,connected:!!stored.botToken&&!!stored.chatId.trim(),messageThreadId:stored.messageThreadId},groups:[],rules:[],history:[],meta:{servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}}),replace:async()=>{replacements++;if(replaceStale)throw Error("Правила изменились на сервере. Обновите снимок.");if(replaceConflict)throw Error("Сохранение правил уже выполняется");return {ok:true as const,revision,groups:[],rules:[]}},writeTelegram:async patch=>executeTelegramSettingsUpdate(patch,{ensure:async()=>{},update:async values=>{writes++;if(failWrite)return undefined;stored={...stored,...values};return {...stored}}}),readTelegram:async()=>{reads++;return {...stored}},send:async input=>{sends++;assert.equal(input.token,"fixture-secret");assert.equal(input.chatId,"123");assert.equal(input.messageThreadId,10);return failSend?{ok:false,error:"send unknown"}:{ok:true}}}),{prefix:"/api"}) +const app=Fastify();await app.register(createAlertsRoutes({...defaultAlertsDependencies,retryOutbox:async(id,input)=>{retries++;if(retryConflict)throw new OutboxRetryConflict();return retryMalformed?({} as never):{ok:true as const,id,status:"pending" as const,resolutionVersion:(input as {expectedVersion:number}).expectedVersion+1}},outbox:async()=>outboxMalformed?{}:({counts:{total:0,pending:0,processing:0,unknown:0,sent:0,failed:0,unrecognized:0},items:[],itemsLimit:20}),overview:async()=>overviewMalformed?{}:({revision,telegram:{chatId:stored.chatId,tokenConfigured:!!stored.botToken,connected:!!stored.botToken&&!!stored.chatId.trim(),messageThreadId:stored.messageThreadId},groups:[],rules:[],history:[],meta:{servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}}),replace:async()=>{replacements++;if(replaceStale)throw Error("Правила изменились на сервере. Обновите снимок.");if(replaceConflict)throw Error("Сохранение правил уже выполняется");return {ok:true as const,revision,groups:[],rules:[]}},writeTelegram:async patch=>executeTelegramSettingsUpdate(patch,{ensure:async()=>{},update:async values=>{writes++;if(failWrite)return undefined;stored={...stored,...values};return {...stored}}}),readTelegram:async()=>{reads++;return {...stored}},send:async input=>{sends++;assert.equal(input.token,"fixture-secret");assert.equal(input.chatId,"123");assert.equal(input.messageThreadId,10);return failSend?{ok:false,error:"send unknown"}:{ok:true}}}),{prefix:"/api"}) const test=()=>app.inject({method:"POST",url:"/api/alerts/telegram/test",payload:{}}) let response=await app.inject({method:"GET",url:"/api/alerts"});assert.equal(response.statusCode,200);assert.equal(response.body.includes("fixture-secret"),false);assert.deepEqual(response.json().rules,[]);assert.equal(response.headers["cache-control"],"no-store");overviewMalformed=true;response=await app.inject({method:"GET",url:"/api/alerts"});assert.equal(response.statusCode,500);assert.equal(response.json().ok,undefined);overviewMalformed=false response=await test();assert.equal(response.statusCode,200);assert.deepEqual(response.json(),{ok:true});assert.equal(reads,1);assert.equal(sends,1) @@ -22,4 +23,7 @@ replaceConflict=true;response=await app.inject({method:"PUT",url:"/api/alerts/ru replaceConflict=false;for(const headers of [{},{"if-match":"*"},{"if-match":revision}]){response=await app.inject({method:"PUT",url:"/api/alerts/rules",headers,payload:{groups:[],rules:[]}});assert.equal(response.statusCode,Object.keys(headers).length?400:428)}assert.equal(replacements,1) replaceStale=true;response=await app.inject({method:"PUT",url:"/api/alerts/rules",headers:{"if-match":'"'+revision+'"'},payload:{groups:[],rules:[]}});assert.equal(response.statusCode,412);assert.equal(replacements,2);assert.equal(response.json().ok,undefined) response=await app.inject({method:"GET",url:"/api/alerts/outbox"});assert.equal(response.statusCode,200);assert.equal(response.headers["cache-control"],"no-store");assert.equal(response.json().counts.total,0);outboxMalformed=true;response=await app.inject({method:"GET",url:"/api/alerts/outbox"});assert.equal(response.statusCode,500);assert.equal(response.json().counts,undefined); +for(const payload of [{},{expectedVersion:0,acknowledgedDuplicateRisk:false}]){response=await app.inject({method:"POST",url:"/api/alerts/outbox/a/retry",payload});assert.equal(response.statusCode,400)}assert.equal(retries,0) +const retryPayload={expectedVersion:0,acknowledgedDuplicateRisk:true};response=await app.inject({method:"POST",url:"/api/alerts/outbox/a/retry",payload:retryPayload});assert.equal(response.statusCode,200);assert.deepEqual(response.json(),{ok:true,id:"a",status:"pending",resolutionVersion:1});assert.equal(response.headers["cache-control"],"no-store") +retryConflict=true;response=await app.inject({method:"POST",url:"/api/alerts/outbox/a/retry",payload:retryPayload});assert.equal(response.statusCode,409);retryConflict=false;retryMalformed=true;response=await app.inject({method:"POST",url:"/api/alerts/outbox/a/retry",payload:retryPayload});assert.equal(response.statusCode,500);assert.equal(response.json().ok,undefined);assert.equal(retries,3) await app.close();console.log("Alerts HTTP: one stored test snapshot, strict numeric thread, selective save, unknown send/no retry, missing returning/no success, no token echo") diff --git a/backend/src/routes/alerts.ts b/backend/src/routes/alerts.ts index 4b16a096..1c6e60c6 100644 --- a/backend/src/routes/alerts.ts +++ b/backend/src/routes/alerts.ts @@ -1,4 +1,4 @@ -import {getAlertsOutboxOverview} from "../services/alert-engine/outbox.js" +import {getAlertsOutboxOverview,retryUnknownOutbox} from "../services/alert-engine/outbox.js" import {sendTelegramRequest} from "../services/alerts-telegram-send.js" import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod" import { @@ -9,7 +9,7 @@ import { updateTelegramSettings, } from "../services/alerts-service.js" import { - alertsResponseSchema, alertsOutboxOverviewSchema, + alertsResponseSchema, alertsOutboxOverviewSchema,alertsOutboxRetrySchema,alertsOutboxRetryReceiptSchema, putRulesSchema, normalizeAlertsConfig, putTelegramSchema, @@ -17,7 +17,9 @@ import { telegramSettingsReceiptSchema, } from "@mmapp/contracts/alerts" +import {OutboxRetryConflict} from "../services/alerts-outbox-retry.js" interface Dependencies { + retryOutbox:typeof retryUnknownOutbox; outbox:()=>Promise;overview:()=>Promise;replace:typeof replaceAlertsConfig;writeTelegram:typeof updateTelegramSettings readTelegram:()=>Promise<{botToken:string;chatId:string;messageThreadId:number|null}> send:(input:{token:string;chatId:string;text:string;messageThreadId?:number})=>Promise<{ok:true}|{ok:false;error:string}> @@ -35,6 +37,12 @@ export function createAlertsRoutes(deps:Dependencies):FastifyPluginAsyncZod {ret return reply.header("Cache-Control","no-store").send(parsed.data) }) + app.post<{Params:{id:string}}>("/alerts/outbox/:id/retry",async(req,reply)=>{ + const parsed=alertsOutboxRetrySchema.safeParse(req.body) + if(!parsed.success||!req.params.id.trim()||req.params.id.length>200)return reply.status(400).send({error:"Проверьте подтверждение и версию записи очереди"}) + try{const receipt=alertsOutboxRetryReceiptSchema.safeParse(await deps.retryOutbox(req.params.id,parsed.data));if(!receipt.success||receipt.data.id!==req.params.id||receipt.data.resolutionVersion!==parsed.data.expectedVersion+1)return reply.status(500).send({error:"Постановка сообщения в очередь не подтверждена"});return reply.header("Cache-Control","no-store").send(receipt.data)}catch(error){if(error instanceof OutboxRetryConflict)return reply.status(409).send({error:error.message});throw error} + }) + app.put("/alerts/rules", async (req, reply) => { const parsed = putRulesSchema.safeParse(req.body) if (!parsed.success) { @@ -118,5 +126,5 @@ export function createAlertsRoutes(deps:Dependencies):FastifyPluginAsyncZod {ret } } -export const defaultAlertsDependencies:Dependencies={outbox:getAlertsOutboxOverview,overview:getAlertsOverview,replace:replaceAlertsConfig,writeTelegram:updateTelegramSettings,readTelegram:getTelegramSettingsRow,send:input=>sendTelegramRequest(input.token,{chat_id:input.chatId,text:input.text,...(input.messageThreadId!==undefined?{message_thread_id:input.messageThreadId}:{})})} +export const defaultAlertsDependencies:Dependencies={retryOutbox:retryUnknownOutbox,outbox:getAlertsOutboxOverview,overview:getAlertsOverview,replace:replaceAlertsConfig,writeTelegram:updateTelegramSettings,readTelegram:getTelegramSettingsRow,send:input=>sendTelegramRequest(input.token,{chat_id:input.chatId,text:input.text,...(input.messageThreadId!==undefined?{message_thread_id:input.messageThreadId}:{})})} export default createAlertsRoutes(defaultAlertsDependencies) diff --git a/backend/src/services/alert-engine/outbox.ts b/backend/src/services/alert-engine/outbox.ts index d93f99ee..6f880465 100644 --- a/backend/src/services/alert-engine/outbox.ts +++ b/backend/src/services/alert-engine/outbox.ts @@ -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[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[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}, }) } diff --git a/backend/src/services/alerts-outbox-overview.test.ts b/backend/src/services/alerts-outbox-overview.test.ts index 7ddf5bb9..47fe358a 100644 --- a/backend/src/services/alerts-outbox-overview.test.ts +++ b/backend/src/services/alerts-outbox-overview.test.ts @@ -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/) }) diff --git a/backend/src/services/alerts-outbox-retry.test.ts b/backend/src/services/alerts-outbox-retry.test.ts new file mode 100644 index 00000000..eec2be0d --- /dev/null +++ b/backend/src/services/alerts-outbox-retry.test.ts @@ -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>>)=>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/) +}) diff --git a/backend/src/services/alerts-outbox-retry.ts b/backend/src/services/alerts-outbox-retry.ts new file mode 100644 index 00000000..521987d7 --- /dev/null +++ b/backend/src/services/alerts-outbox-retry.ts @@ -0,0 +1,13 @@ +import {alertsOutboxRetrySchema,alertsOutboxRetryReceiptSchema} from "@mmapp/contracts/alerts" +export class OutboxRetryConflict extends Error{constructor(){super("Запись очереди изменилась или не имеет неизвестного результата. Обновите снимок.")}} +export async function executeOutboxRetry(id:string,raw:unknown,deps:{transaction:(work:(context:C)=>Promise>)=>Promise>;update:(context:C,id:string,expectedVersion:number)=>Promise}){ + 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 + }) +} diff --git a/docs/ux-ui-plan.md b/docs/ux-ui-plan.md index 7af7fa52..932d3b88 100644 --- a/docs/ux-ui-plan.md +++ b/docs/ux-ui-plan.md @@ -4,7 +4,7 @@ ## Актуальная точка продолжения — 2026-10-03 -Этап 11Б в работе; последний проверенный блок 11Б.4.4.3.3; System DB принят локально с внешними ограничениями. Следующий блок 11Б.4.4.3.4: контролируемый повтор только unknown после явной проверки оператором и защиты от конкуренции; processing удерживать до отдельной проверки остановки отправителя. Затем общий dashboard permission reader, адаптивность/pickers, Releases/probe scheduling и приёмка 11Б/12. +Этап 11Б в работе; последний проверенный блок 11Б.4.4.3.4; System DB принят локально с внешними ограничениями. Следующий блок 11Б.4.4.3.5: подключить UI ручного повтора unknown с подтверждением риска дубликата, общим checkpoint при неизвестном POST и сохранением формы/источника; затем общий dashboard permission reader, адаптивность/pickers, Releases/probe scheduling и приёмка 11Б/12. Реальные pg_dump/pg_restore, PG rollback и остановка UDP worker требуют внешней приёмки. Датированные продолжения ниже — история; точный следующий шаг также указан в начале [журнала](ux-ui-progress.md). diff --git a/docs/ux-ui-progress.md b/docs/ux-ui-progress.md index 70d8fb28..d81f3082 100644 --- a/docs/ux-ui-progress.md +++ b/docs/ux-ui-progress.md @@ -28,7 +28,7 @@ | 9 | VPN, Firewall и Filters | Завершён в границах локальной приёмки 2026-09-30; блоки 9.1–9.64, ограничения в 9.64 | | 10 | Маршрутизация, диагностика, карта и терминал | Завершён в границах локальной приёмки 2026-10-01; блоки 10.1–10.71, внешние ограничения в 10.71 | | 11А | Пользователи, контейнеры, сертификаты и резервные копии | Завершён в границах локальной приёмки 2026-10-01; 11А.1–11А.33, внешние ограничения в 11А.33 | -| 11Б | Оповещения, сбор данных, интеграции, релизы и настройки | В работе: 11Б.1–11Б.4.4.3.3; scheduler и GeoIP приняты локально; NetFlow принят локально; Evolution принят локально; далее общие Settings | +| 11Б | Оповещения, сбор данных, интеграции, релизы и настройки | В работе: 11Б.1–11Б.4.4.3.4; scheduler и GeoIP приняты локально; NetFlow принят локально; Evolution принят локально; далее общие Settings | | 12 | Итоговая доступность, адаптивность, производительность и регрессия | Не начат | @@ -36,6 +36,14 @@ +## Блок 11Б.4.4.3.4 — атомарный ручной повтор unknown (2026-10-03) + +- POST /api/alerts/outbox/:id/retry под mm:alerts:write/admin требует literal acknowledgedDuplicateRisk:true и expectedVersion. Conditional UPDATE только telegram+unknown+resolution_version переводит запись в pending и увеличивает version. Processing/sent/failed/pending, устаревший снимок и повторный запрос получают 409; endpoint не отправляет Telegram и не пишет ложный успешный history. +- Receipt pending/id/version+1 проверяется внутри transaction и на HTTP границе. Ошибка returning отклоняет callback; сохранённый retryCount/maxRetries остаётся неизменным. 0009 migration добавляет resolution_version и nonnegative CHECK; GET выдаёт version для held items. Версия растёт при каждом ручном повторе, поэтому новый unknown не совпадает с предыдущей генерацией даже при одинаковом timestamp. +- 2 DI проверки concurrent retry/stale/processing/validation/receipt rollback, полный Alerts backend профиль, HTTP 400/409/500/no-store и реальные JWT gates прошли; backend build, nonincremental frontend typecheck, профильный lint и queue browser 2/2 прошли. +- Реальный PG CAS/rollback/0009 migration не проверен локально. При выпуске migrations 0008/0009 и backend/contracts/frontend обновить вместе; held GET требует resolutionVersion. UI операции ещё не подключён, processing не имеет ручного повтора до доказуемой остановки отправителя. +- Следующий точный шаг 11Б.4.4.3.5: подключить UI ручного повтора unknown с подтверждением риска дубликата, общим checkpoint при неизвестном POST и сохранением формы/источника; затем общий dashboard permission reader, адаптивность/pickers, Releases/probe scheduling и приёмка 11Б/12. + ## Блок 11Б.4.4.3.3 — подтверждённый снимок очереди Telegram (2026-10-03) - Отдельный GET /api/alerts/outbox под mm:alerts:read, no-store и строгим DTO. Counts и последние 20 processing/unknown читаются в одной repeatable read/read only transaction; сумма, безопасные числа, количество/статусы/уникальность элементов проверяются. Неизвестные legacy статусы учитываются отдельно. Ответ не содержит payload, token или lastError. diff --git a/docs/ux-ui-stage-11b.md b/docs/ux-ui-stage-11b.md index 2c493184..75db77b0 100644 --- a/docs/ux-ui-stage-11b.md +++ b/docs/ux-ui-stage-11b.md @@ -292,3 +292,7 @@ Conditional atomic claim, unknown/processing hold, transactional status/history ### Продолжение 2026-10-03 — 11Б.4.4.3.3 GET/UI очереди принят локально: независимый strict snapshot, known/unknown, сохранение при сбое и ABA; 32 browser + 12 context, API/DI/HTTP/JWT/build/typecheck/lint прошли. Реальный PG/Telegram и ручное разрешение held не приняты. Следующий блок 11Б.4.4.3.4: контролируемый повтор только unknown после явной проверки оператором и защиты от конкуренции; processing удерживать до отдельной проверки остановки отправителя. Затем общий dashboard permission reader, адаптивность/pickers, Releases/probe scheduling и приёмка 11Б/12. + +### Продолжение 2026-10-03 — 11Б.4.4.3.4 + +Backend manual unknown retry принят локально: acknowledged risk + atomic version CAS, transaction receipt, no send/history; DI/HTTP/JWT/build/typecheck/lint/queue browser прошли. Реальный PG/0009 и UI действия не приняты. Следующий блок 11Б.4.4.3.5: подключить UI ручного повтора unknown с подтверждением риска дубликата, общим checkpoint при неизвестном POST и сохранением формы/источника; затем общий dashboard permission reader, адаптивность/pickers, Releases/probe scheduling и приёмка 11Б/12. diff --git a/packages/contracts/src/alerts.ts b/packages/contracts/src/alerts.ts index 8dec583a..d3b69d9f 100644 --- a/packages/contracts/src/alerts.ts +++ b/packages/contracts/src/alerts.ts @@ -133,5 +133,8 @@ export function normalizeAlertsConfig(input:PutRulesBody){ } const outboxCount=z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER) -export const alertsOutboxOverviewSchema=z.object({counts:z.object({total:outboxCount,pending:outboxCount,processing:outboxCount,unknown:outboxCount,sent:outboxCount,failed:outboxCount,unrecognized:outboxCount}).strict(),items:z.array(z.object({id:z.string().min(1),status:z.enum(["processing","unknown"]),ruleName:z.string().nullable(),createdAt:alertTimestampSchema,attemptedAt:alertTimestampSchema}).strict()).max(20),itemsLimit:z.literal(20)}).strict().refine(value=>value.counts.total===value.counts.pending+value.counts.processing+value.counts.unknown+value.counts.sent+value.counts.failed+value.counts.unrecognized&&value.items.length===Math.min(20,value.counts.processing+value.counts.unknown)&&value.items.filter(item=>item.status==="processing").length<=value.counts.processing&&value.items.filter(item=>item.status==="unknown").length<=value.counts.unknown&&new Set(value.items.map(item=>item.id)).size===value.items.length,"Неполное состояние очереди") +export const alertsOutboxOverviewSchema=z.object({counts:z.object({total:outboxCount,pending:outboxCount,processing:outboxCount,unknown:outboxCount,sent:outboxCount,failed:outboxCount,unrecognized:outboxCount}).strict(),items:z.array(z.object({id:z.string().min(1),resolutionVersion:z.number().int().nonnegative().max(2147483647),status:z.enum(["processing","unknown"]),ruleName:z.string().nullable(),createdAt:alertTimestampSchema,attemptedAt:alertTimestampSchema}).strict()).max(20),itemsLimit:z.literal(20)}).strict().refine(value=>value.counts.total===value.counts.pending+value.counts.processing+value.counts.unknown+value.counts.sent+value.counts.failed+value.counts.unrecognized&&value.items.length===Math.min(20,value.counts.processing+value.counts.unknown)&&value.items.filter(item=>item.status==="processing").length<=value.counts.processing&&value.items.filter(item=>item.status==="unknown").length<=value.counts.unknown&&new Set(value.items.map(item=>item.id)).size===value.items.length,"Неполное состояние очереди") export type AlertsOutboxOverview=z.output + +export const alertsOutboxRetrySchema=z.object({expectedVersion:z.number().int().nonnegative().max(2147483646),acknowledgedDuplicateRisk:z.literal(true)}).strict() +export const alertsOutboxRetryReceiptSchema=z.object({ok:z.literal(true),id:z.string().min(1),status:z.literal("pending"),resolutionVersion:z.number().int().min(1).max(2147483647)}).strict() diff --git a/tests/e2e/alerts-context.spec.ts b/tests/e2e/alerts-context.spec.ts index 72bcf620..adb9d374 100644 --- a/tests/e2e/alerts-context.spec.ts +++ b/tests/e2e/alerts-context.spec.ts @@ -44,7 +44,7 @@ for(const failed of [false,true])test("Queue source: ABA ignores old held list o let old=true,started=0,completed=0,release!:()=>void;const gate=new Promise(resolve=>release=resolve) await page.route("**/health",r=>r.fulfill({headers,json:{status:"ok"}}));await page.route("**/api/**",async r=>{ const url=new URL(r.request().url()),path=url.pathname - if(path==="/api/alerts/outbox"&&old&&url.port==="8000"){started++;await gate;try{await r.fulfill(failed?{headers,status:503,json:{error:"OLD QUEUE FAILED"}}:{headers,json:{...emptyOutbox,counts:{...emptyOutbox.counts,total:1,unknown:1},items:[{id:"old",status:"unknown",ruleName:"OLD HELD",createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T00:00:00Z"}]}})}catch{}finally{completed++};return} + if(path==="/api/alerts/outbox"&&old&&url.port==="8000"){started++;await gate;try{await r.fulfill(failed?{headers,status:503,json:{error:"OLD QUEUE FAILED"}}:{headers,json:{...emptyOutbox,counts:{...emptyOutbox.counts,total:1,unknown:1},items:[{id:"old",resolutionVersion:0,status:"unknown",ruleName:"OLD HELD",createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T00:00:00Z"}]}})}catch{}finally{completed++};return} return r.fulfill({headers,json:path==="/api/auth/config"?{required:false}:path==="/api/alerts"?emptyAlerts:path==="/api/alerts/outbox"?emptyOutbox:path==="/api/filters/gre-tunnels"?{tunnels:[],failures:[]}:[]})}) try{await page.goto("/alerts");await expect.poll(()=>started).toBeGreaterThan(0);await expect(page.getByText("Всего: 0",{exact:true})).toBeVisible();await page.goto("/settings");await source(page,"http://localhost:8101");old=false;await source(page,"http://localhost:8000");await page.goto("/alerts");const panel=page.getByRole("region",{name:"Очередь Telegram",exact:true});await expect(panel).toContainText("Ожидают: 0");release();await expect.poll(()=>completed).toBe(started);await expect(panel).not.toContainText("OLD QUEUE");await expect(panel).not.toContainText("OLD HELD");await expect(panel).toContainText("Ожидают: 0")}finally{release()} }) diff --git a/tests/e2e/alerts-outbox.spec.ts b/tests/e2e/alerts-outbox.spec.ts index 12adc34b..fc70dfd9 100644 --- a/tests/e2e/alerts-outbox.spec.ts +++ b/tests/e2e/alerts-outbox.spec.ts @@ -7,5 +7,5 @@ test("Queue initial failure is unknown, retry confirms zero",async({page})=>{ let fail=true;await setup(page,()=>({json:fail?{}:emptyOutbox}));const panel=page.getByRole("region",{name:"Очередь Telegram",exact:true});await expect(panel.getByRole("alert")).toContainText("не подтверждено");await expect(panel).not.toContainText("Ожидают: 0");await expect(page.getByText("Всего: 0",{exact:true})).toBeVisible();fail=false;await panel.getByRole("button",{name:"Обновить очередь Telegram"}).click();await expect(panel).toContainText("Ожидают: 0");await expect(panel.getByRole("alert")).toHaveCount(0) }) test("Queue held states retain confirmed snapshot after failure and fit mobile",async({page})=>{ - await page.setViewportSize({width:320,height:1000});let fail=false;const held={...emptyOutbox,counts:{...emptyOutbox.counts,total:4,pending:1,processing:1,unknown:1,unrecognized:1},items:[{id:"a",status:"unknown",ruleName:"Held fixture",createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T01:00:00Z"},{id:"b",status:"processing",ruleName:null,createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T02:00:00Z"}]};await setup(page,()=>fail?{status:503,json:{error:"QUEUE UNAVAILABLE"}}:{json:held});const panel=page.getByRole("region",{name:"Очередь Telegram",exact:true});await expect(panel).toContainText("Held fixture");await expect(panel).toContainText("автоматически не переотправляются");fail=true;await panel.getByRole("button",{name:"Обновить очередь Telegram"}).click();await expect(panel.getByRole("alert")).toContainText("Показан последний успешный снимок");await expect(panel).toContainText("Ожидают: 1");const bounds=await panel.boundingBox();expect(bounds!.x).toBeGreaterThanOrEqual(0);expect(bounds!.x+bounds!.width).toBeLessThanOrEqual(320);await panel.screenshot({path:"test-results/alerts-outbox-320.png"}) + await page.setViewportSize({width:320,height:1000});let fail=false;const held={...emptyOutbox,counts:{...emptyOutbox.counts,total:4,pending:1,processing:1,unknown:1,unrecognized:1},items:[{id:"a",resolutionVersion:0,status:"unknown",ruleName:"Held fixture",createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T01:00:00Z"},{id:"b",resolutionVersion:0,status:"processing",ruleName:null,createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T02:00:00Z"}]};await setup(page,()=>fail?{status:503,json:{error:"QUEUE UNAVAILABLE"}}:{json:held});const panel=page.getByRole("region",{name:"Очередь Telegram",exact:true});await expect(panel).toContainText("Held fixture");await expect(panel).toContainText("автоматически не переотправляются");fail=true;await panel.getByRole("button",{name:"Обновить очередь Telegram"}).click();await expect(panel.getByRole("alert")).toContainText("Показан последний успешный снимок");await expect(panel).toContainText("Ожидают: 1");const bounds=await panel.boundingBox();expect(bounds!.x).toBeGreaterThanOrEqual(0);expect(bounds!.x+bounds!.width).toBeLessThanOrEqual(320);await panel.screenshot({path:"test-results/alerts-outbox-320.png"}) })