{"id":477645,"date":"2026-04-27T15:55:21","date_gmt":"2026-04-27T15:55:21","guid":{"rendered":"https:\/\/savepearlharbor.com\/?p=477645"},"modified":"-0001-11-30T00:00:00","modified_gmt":"-0001-11-29T21:00:00","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=477645","title":{"rendered":"At-least-once. \u042d\u0442\u043e \u043d\u0435 \u0431\u0430\u0433 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440\u0430. \u042d\u0442\u043e \u0432\u0430\u0448\u0430 \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u043d\u0430\u044f \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0430"},"content":{"rendered":"
\n
\n

\u041a\u043e\u0434 \u0432 \u0441\u0442\u0430\u0442\u044c\u0435 \u0438\u043b\u043b\u044e\u0441\u0442\u0440\u0430\u0442\u0438\u0432\u043d\u044b\u0439, \u043f\u043e\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u0442 \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u043d\u044b\u0435 \u0440\u0435\u0448\u0435\u043d\u0438\u044f \u0438 \u043e\u0431\u044a\u044f\u0441\u043d\u044f\u0435\u0442 \u043f\u043e\u0447\u0435\u043c\u0443 \u0438\u043c\u0435\u043d\u043d\u043e \u0442\u0430\u043a. \u041d\u0435 \u043f\u0440\u0435\u0434\u043d\u0430\u0437\u043d\u0430\u0447\u0435\u043d \u0434\u043b\u044f copy-paste \u0432 \u043f\u0440\u043e\u0434 \u0431\u0435\u0437 \u0430\u0434\u0430\u043f\u0442\u0430\u0446\u0438\u0438 \u043f\u043e\u0434 \u0432\u0430\u0448\u0443 \u0438\u043d\u0444\u0440\u0430\u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443, \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433 \u0438 \u0442\u0440\u0435\u0431\u043e\u0432\u0430\u043d\u0438\u044f. <\/p>\n<\/blockquote>\n

\u0414\u0443\u043c\u0430\u043b, \u0437\u0430\u0439\u0434\u0443 \u0432 \u043a\u0440\u0438\u043f\u0442\u0443 \u0438 \u0431\u0443\u0434\u0443 \u043f\u0440\u043e\u0441\u0442\u043e \u0434\u0451\u0440\u0433\u0430\u0442\u044c API \u0431\u043b\u043e\u043a\u0447\u0435\u0439\u043d\u0430. \u041d\u0435 \u0432\u044b\u0448\u043b\u043e.<\/p>\n

\u0417\u0430\u0445\u043e\u0436\u0443 \u0432 \u043f\u0440\u043e\u0435\u043a\u0442. \u0421\u0442\u0435\u043a: FastAPI, PostgreSQL, Redis \u043a\u0430\u043a Celery broker, Celery workers, Docker, Web3. \u0421\u0442\u0430\u0440\u0442\u0430\u043f \u043d\u0430 \u0445\u0430\u0439\u043f\u0435, \u0434\u0435\u043d\u044c\u0433\u0438 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u0435, \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u0430 \u0441\u043e\u0431\u0440\u0430\u043d\u0430 \u043d\u0430 \u043a\u043e\u043b\u0435\u043d\u043a\u0435. \u0421\u043c\u043e\u0442\u0440\u044e \u043d\u0430 \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u0443 \u043f\u043b\u0430\u0442\u0451\u0436\u043d\u043e\u0433\u043e \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0438\u043d\u0433\u0430 \u0438 \u043f\u0435\u0440\u0432\u0430\u044f \u043c\u044b\u0441\u043b\u044c: \u0440\u0435\u0431\u044f\u0442\u0430, \u0432\u044b \u0441\u0435\u0440\u044c\u0451\u0437\u043d\u043e? \u0424\u0438\u043d\u0430\u043d\u0441\u043e\u0432\u044b\u0435 \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u0438 \u0441 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u043c\u0438 \u0434\u0435\u043d\u044c\u0433\u0430\u043c\u0438, \u0431\u0435\u0437 idempotency \u0432\u043e\u043e\u0431\u0449\u0435, Redis \u043a\u0430\u043a \u0431\u0440\u043e\u043a\u0435\u0440 \u0431\u0435\u0437 persistence, Web3.py \u0441\u0438\u043d\u0445\u0440\u043e\u043d\u043d\u044b\u0435 \u0432\u044b\u0437\u043e\u0432\u044b \u0432\u043d\u0443\u0442\u0440\u0438 Celery \u0442\u0430\u0441\u043a\u043e\u0432.<\/p>\n

\u0420\u0430\u0437\u0433\u043e\u0432\u043e\u0440 \u0431\u044b\u043b \u043a\u043e\u0440\u043e\u0442\u043a\u0438\u0439: \u0437\u0430\u0434\u0430\u0447\u0430 \u0442\u0430\u043a\u0430\u044f, \u0447\u0438\u043d\u0438 \u0447\u0442\u043e \u0435\u0441\u0442\u044c. \u0421\u0440\u043e\u043a\u0438 \u0433\u043e\u0440\u0435\u043b\u0438.<\/p>\n

\u0427\u0442\u043e \u0431\u044b\u043b\u043e \u0441\u043b\u043e\u043c\u0430\u043d\u043e <\/h2>\n

\u041f\u0435\u0440\u0432\u044b\u0439 \u043c\u0435\u0441\u044f\u0446 \u043f\u0440\u043e\u0434\u0430. \u041f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u043f\u0438\u0448\u0435\u0442 \u0432 \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u043a\u0443: \u0437\u0430\u0447\u0438\u0441\u043b\u0438\u043b\u0438 \u0434\u0432\u0430\u0436\u0434\u044b, \u0432\u044b\u0432\u0435\u043b \u0434\u0432\u043e\u0439\u043d\u0443\u044e \u0441\u0443\u043c\u043c\u0443. \u041e\u0442\u043a\u0440\u044b\u0432\u0430\u044e \u043b\u043e\u0433\u0438, \u0447\u0438\u0441\u0442\u043e. \u0414\u0432\u0430 \u0438\u0434\u0435\u043d\u0442\u0438\u0447\u043d\u044b\u0445 \u0441\u043e\u0431\u044b\u0442\u0438\u044f, \u043e\u0431\u0430 200, \u0440\u0430\u0437\u043d\u0438\u0446\u0430 \u0447\u0435\u0442\u044b\u0440\u0435 \u0441\u0435\u043a\u0443\u043d\u0434\u044b. \u041e\u0431\u0430 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u0430\u043d\u044b. \u041f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u043f\u043e\u043b\u0443\u0447\u0438\u043b \u0434\u0432\u043e\u0439\u043d\u043e\u0439 \u0431\u0430\u043b\u0430\u043d\u0441.<\/p>\n

\u0415\u0436\u0435\u0434\u043d\u0435\u0432\u043d\u0430\u044f \u0441\u0432\u0435\u0440\u043a\u0430 \u0441 on-chain \u0434\u0430\u043d\u043d\u044b\u043c\u0438 \u043f\u043e\u043a\u0430\u0437\u0430\u043b\u0430 \u0440\u0430\u0441\u0445\u043e\u0436\u0434\u0435\u043d\u0438\u0435: \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0430\u043a\u043a\u0430\u0443\u043d\u0442\u043e\u0432 \u0441 \u0431\u0430\u043b\u0430\u043d\u0441\u043e\u043c \u0431\u043e\u043b\u044c\u0448\u0435 \u0447\u0435\u043c \u0434\u043e\u043b\u0436\u043d\u043e \u0431\u044b\u0442\u044c \u043f\u043e confirmed \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f\u043c. \u0417\u0430 \u043f\u0435\u0440\u0432\u044b\u0439 \u043c\u0435\u0441\u044f\u0446 \u043d\u0430\u0448\u043b\u0438 23 \u0434\u0443\u0431\u043b\u0438\u0440\u0443\u044e\u0449\u0438\u0445 \u0437\u0430\u0447\u0438\u0441\u043b\u0435\u043d\u0438\u044f \u043d\u0430 ~180k \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0439, \u043e\u043a\u043e\u043b\u043e 0.013% error rate. 23 \u0434\u0432\u043e\u0439\u043d\u044b\u0445 \u0437\u0430\u0447\u0438\u0441\u043b\u0435\u043d\u0438\u044f \u0437\u0430 \u043c\u0435\u0441\u044f\u0446. \u0416\u0438\u0432\u044b\u0435 \u0434\u0435\u043d\u044c\u0433\u0438, \u043d\u0435 \u043c\u0435\u0442\u0440\u0438\u043a\u0430.<\/p>\n

\u041f\u0435\u0440\u0432\u043e\u0435, \u0447\u0442\u043e \u0432\u044b\u043b\u0435\u0437\u043b\u043e: \u0434\u0443\u0431\u043b\u0438 \u043e\u0442 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440\u0430. Alchemy, Infura \u0438 \u0432\u0441\u0435 \u043e\u0441\u0442\u0430\u043b\u044c\u043d\u044b\u0435 \u0431\u043b\u043e\u043a\u0447\u0435\u0439\u043d-\u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440\u044b \u0440\u0430\u0431\u043e\u0442\u0430\u044e\u0442 \u043f\u043e at-least-once delivery. \u041f\u0440\u0438 \u0441\u0435\u0442\u0435\u0432\u043e\u043c \u0441\u0431\u043e\u0435, \u0440\u0435\u0441\u0442\u0430\u0440\u0442\u0435, \u043f\u043e\u0434 \u043d\u0430\u0433\u0440\u0443\u0437\u043a\u043e\u0439 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u043f\u043e\u0432\u0442\u043e\u0440\u044f\u0435\u0442 \u0434\u043e\u0441\u0442\u0430\u0432\u043a\u0443. \u041f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u0442\u0430\u043a \u0438 \u043d\u0430\u043f\u0438\u0441\u0430\u043b \u0432 \u0434\u043e\u043a\u0430\u0445. \u042d\u0442\u043e \u043d\u0435 \u0431\u0430\u0433, \u044d\u0442\u043e \u0443\u0441\u043b\u043e\u0432\u0438\u044f \u0438\u0433\u0440\u044b. \u041f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u043f\u043e\u0432\u0442\u043e\u0440\u044f\u0435\u0442 \u0434\u043e\u0441\u0442\u0430\u0432\u043a\u0443, \u0442\u0432\u043e\u0439 \u043a\u043e\u0434 \u0434\u043e\u043b\u0436\u0435\u043d \u044d\u0442\u043e \u043f\u0435\u0440\u0435\u0436\u0438\u0432\u0430\u0442\u044c \u0431\u0435\u0437 \u043f\u043e\u0441\u043b\u0435\u0434\u0441\u0442\u0432\u0438\u0439. \u041d\u0430\u0448 \u043d\u0435 \u043f\u0435\u0440\u0435\u0436\u0438\u0432\u0430\u043b.<\/p>\n

\u0414\u0430\u043b\u044c\u0448\u0435 \u0445\u0443\u0436\u0435. \u0414\u0432\u0430 \u043f\u0430\u0440\u0430\u043b\u043b\u0435\u043b\u044c\u043d\u044b\u0445 \u0437\u0430\u043f\u0440\u043e\u0441\u0430 \u043d\u0430 \u0432\u044b\u0432\u043e\u0434 \u0447\u0438\u0442\u0430\u043b\u0438 \u0431\u0430\u043b\u0430\u043d\u0441 \u043e\u0434\u043d\u043e\u0432\u0440\u0435\u043c\u0435\u043d\u043d\u043e, \u043e\u0431\u0430 \u0432\u0438\u0434\u0435\u043b\u0438 \u0434\u043e\u0441\u0442\u0430\u0442\u043e\u0447\u043d\u043e \u0441\u0440\u0435\u0434\u0441\u0442\u0432, \u043e\u0431\u0430 \u043f\u0440\u043e\u0445\u043e\u0434\u0438\u043b\u0438 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e. \u0414\u0432\u0430 \u0437\u0430\u043f\u0440\u043e\u0441\u0430 \u0447\u0438\u0442\u0430\u044e\u0442 \u0431\u0430\u043b\u0430\u043d\u0441 \u043e\u0434\u043d\u043e\u0432\u0440\u0435\u043c\u0435\u043d\u043d\u043e, \u043e\u0431\u0430 \u0432\u0438\u0434\u044f\u0442 \u0447\u0442\u043e \u0434\u0435\u043d\u0435\u0433 \u0445\u0432\u0430\u0442\u0430\u0435\u0442, \u043e\u0431\u0430 \u0441\u043f\u0438\u0441\u044b\u0432\u0430\u044e\u0442. \u0428\u043a\u043e\u043b\u044c\u043d\u0430\u044f \u0433\u043e\u043d\u043a\u0430.<\/p>\n

async def withdraw(conn, user_id: int, amount: Decimal):    balance = await conn.fetchval(        \"SELECT balance FROM users WHERE id = $1\", user_id    )    if balance >= amount:        await conn.execute(            \"UPDATE users SET balance = balance - $1 WHERE id = $2\",            amount, user_id        )<\/code>
<\/a><\/div><\/pre>\n
\u0414\u0430\u043b\u044c\u0448\u0435. Celery \u0441 \u0434\u0435\u0444\u043e\u043b\u0442\u043d\u044b\u043c\u0438 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0430\u043c\u0438 \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0430\u0435\u0442 \u0437\u0430\u0434\u0430\u0447\u0443 \u0431\u0440\u043e\u043a\u0435\u0440\u0443 \u0432 \u043c\u043e\u043c\u0435\u043d\u0442 \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f. \u0412\u043e\u0440\u043a\u0435\u0440 \u043f\u0430\u0434\u0430\u0435\u0442 \u0432 \u0441\u0435\u0440\u0435\u0434\u0438\u043d\u0435 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438, \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u043e, \u0434\u043e \u0437\u0430\u043f\u0438\u0441\u0438 \u0432 \u0411\u0414 \u043d\u0435 \u0434\u043e\u0448\u043b\u043e. \u041d\u0438\u043a\u0430\u043a\u043e\u0433\u043e retry, \u043d\u0438\u043a\u0430\u043a\u043e\u0433\u043e DLQ. \u0412\u043e\u0440\u043a\u0435\u0440 \u0443\u043f\u0430\u043b, \u0437\u0430\u0434\u0430\u0447\u0430 \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u0430, \u0434\u0435\u043d\u044c\u0433\u0438 \u043d\u0435 \u043f\u0440\u0438\u0448\u043b\u0438. \u041f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u0436\u0434\u0451\u0442 \u0438 \u043d\u0435 \u043f\u043e\u043d\u0438\u043c\u0430\u0435\u0442 \u0447\u0442\u043e \u0441\u043b\u0443\u0447\u0438\u043b\u043e\u0441\u044c.<\/p>\n
\u0418 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 \u0442\u0438\u0445\u0438\u0439 \u0443\u0431\u0438\u0439\u0446\u0430: amount<\/code> \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0443\u0435\u0442\u0441\u044f \u0432 JSON \u0447\u0435\u0440\u0435\u0437 Celery broker \u043a\u0430\u043a float<\/code>. Decimal(\"50.1\")<\/code> \u043f\u0440\u0435\u0432\u0440\u0430\u0449\u0430\u0435\u0442\u0441\u044f \u0432 JSON float, \u0442\u043e \u0435\u0441\u0442\u044c \u0432 50.099999999999994<\/code>. \u041d\u0430 \u043c\u0430\u0441\u0448\u0442\u0430\u0431\u0435 \u044d\u0442\u043e \u043a\u043e\u043f\u0438\u0442\u0441\u044f \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u0439 \u0443\u0431\u044b\u0442\u043e\u043a. \u041d\u0438\u043a\u0442\u043e \u043d\u0435 \u0437\u0430\u043c\u0435\u0442\u0438\u043b, \u043f\u043e\u043a\u0430 \u043d\u0435 \u043f\u043e\u0441\u0447\u0438\u0442\u0430\u043b\u0438.<\/p>\n
\u041f\u043e\u0441\u043b\u0435\u0434\u043d\u0435\u0435: \u043f\u0440\u044f\u043c\u043e\u0439 \u0432\u044b\u0437\u043e\u0432 .delay()<\/code> \u0438\u0437 webhook handler \u0441\u043e\u0437\u0434\u0430\u0451\u0442 \u043e\u043a\u043d\u043e \u043c\u0435\u0436\u0434\u0443 \u0437\u0430\u043f\u0438\u0441\u044c\u044e \u0432 \u0411\u0414 \u0438 \u043f\u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u043e\u0439 \u0432 \u043e\u0447\u0435\u0440\u0435\u0434\u044c. \u0415\u0441\u043b\u0438 \u043f\u0440\u043e\u0446\u0435\u0441\u0441 \u0443\u043f\u0430\u043b \u0432 \u044d\u0442\u043e\u0442 \u043c\u043e\u043c\u0435\u043d\u0442, \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0437\u0430\u0432\u0438\u0441\u043d\u0435\u0442 \u0432 pending<\/code> \u0431\u0435\u0437 \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u043e\u0433\u043e \u0432\u043e\u0441\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d\u0438\u044f.<\/p>\n
\u0418\u0442\u043e\u0433\u043e \u043f\u044f\u0442\u044c \u043f\u0440\u043e\u0431\u043b\u0435\u043c. \u041d\u0430\u0447\u0430\u043b \u0447\u0438\u043d\u0438\u0442\u044c.<\/p>\n
\u041f\u0435\u0440\u0432\u044b\u0439 \u0438\u043d\u0441\u0442\u0438\u043d\u043a\u0442: Redis distributed lock  <\/h2>\n
SET NX EX<\/code> \u043d\u0430 user_id<\/code>. \u041f\u0430\u0442\u0442\u0435\u0440\u043d \u043e\u043f\u0438\u0441\u0430\u043d \u0443 Antirez, \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d \u0437\u0430 20 \u043c\u0438\u043d\u0443\u0442. \u041d\u0435 \u0432\u0437\u043b\u0435\u0442\u0435\u043b\u043e.<\/p>\n
\u0412\u043e\u0442 \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u044b\u0439 \u0441\u0446\u0435\u043d\u0430\u0440\u0438\u0439, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0432\u0441\u043a\u0440\u044b\u043b\u0441\u044f \u0432 \u043b\u043e\u0433\u0430\u0445. \u0412\u043e\u0440\u043a\u0435\u0440 \u0431\u0435\u0440\u0451\u0442 \u043b\u043e\u043a \u0432 Redis. \u041d\u0430\u0447\u0438\u043d\u0430\u0435\u0442 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044e \u0432 PostgreSQL. \u041c\u0435\u0436\u0434\u0443 \u044d\u0442\u0438\u043c\u0438 \u0434\u0432\u0443\u043c\u044f \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u044f\u043c\u0438 OOM killer \u0443\u0431\u0438\u0432\u0430\u0435\u0442 \u043f\u0440\u043e\u0446\u0435\u0441\u0441. PostgreSQL \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u043e\u0442\u043a\u0430\u0442\u0438\u043b\u0430\u0441\u044c \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438, \u0431\u0430\u043b\u0430\u043d\u0441 \u043d\u0435 \u0438\u0437\u043c\u0435\u043d\u0438\u043b\u0441\u044f. Redis \u043b\u043e\u043a \u0432\u0438\u0441\u0438\u0442 30 \u0441\u0435\u043a\u0443\u043d\u0434 \u0434\u043e TTL. \u0427\u0435\u0440\u0435\u0437 30 \u0441\u0435\u043a\u0443\u043d\u0434 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0439 \u0432\u043e\u0440\u043a\u0435\u0440 \u0431\u0435\u0440\u0451\u0442 \u043b\u043e\u043a, \u0432\u0438\u0434\u0438\u0442 \u0447\u0442\u043e  idempotency_key <\/code>\u043d\u0435 \u0437\u0430\u043f\u0438\u0441\u0430\u043d (\u0437\u0430\u043f\u0438\u0441\u0430\u0442\u044c \u0431\u044b\u043b\u043e \u043d\u0435\u043a\u043e\u043c\u0443, \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u043e\u0442\u043a\u0430\u0442\u0438\u043b\u0430\u0441\u044c) \u0438 \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0435\u0442 \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0437\u0430\u043d\u043e\u0432\u043e. \u0414\u0432\u043e\u0439\u043d\u043e\u0435 \u0437\u0430\u0447\u0438\u0441\u043b\u0435\u043d\u0438\u0435. \u0412 \u043b\u043e\u0433\u0430\u0445 \u043e\u0431\u0430 \u0432\u043e\u0440\u043a\u0435\u0440\u0430 \u0447\u0438\u0441\u0442\u044b\u0435.<\/p>\n
\u041f\u0440\u043e\u0431\u043b\u0435\u043c\u0430 \u043d\u0435 \u0432 \u0440\u0430\u0437\u043c\u0435\u0440\u0435 TTL. \u041f\u0440\u043e\u0431\u043b\u0435\u043c\u0430 \u0432 \u043e\u0442\u0441\u0443\u0442\u0441\u0442\u0432\u0438\u0438 cross-system \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e\u0441\u0442\u0438 \u043c\u0435\u0436\u0434\u0443 Redis \u0438 PostgreSQL. Redis \u043d\u0435 \u043f\u043e\u0434\u0445\u043e\u0434\u0438\u0442, \u043d\u0435\u0442 \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e\u0441\u0442\u0438 \u0441 PostgreSQL. \u041f\u0440\u043e\u0432\u0435\u0440\u043a\u0430 \u0432 \u043a\u043e\u0434\u0435 \u0442\u043e\u0436\u0435 \u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442, \u0434\u0432\u0430 \u0432\u043e\u0440\u043a\u0435\u0440\u0430 \u043e\u0431\u0430 \u043f\u0440\u043e\u0439\u0434\u0443\u0442 SELECT \u0434\u043e INSERT. \u0415\u0434\u0438\u043d\u0441\u0442\u0432\u0435\u043d\u043d\u043e\u0435 \u0447\u0442\u043e \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e \u043f\u043e \u043e\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u0438\u044e  unique constraint. \u0421 \u0434\u0435\u043d\u044c\u0433\u0430\u043c\u0438 \u043d\u0435\u0442 «\u043f\u043e\u0447\u0442\u0438 \u043f\u0440\u0430\u0432\u0438\u043b\u044c\u043d\u043e».<\/p>\n
 \u0421\u0445\u0435\u043c\u0430 \u0431\u0430\u0437\u044b \u0434\u0430\u043d\u043d\u044b\u0445  <\/h2>\n
CREATE TABLE payment_events (    event_id     TEXT PRIMARY KEY,    user_id      INTEGER NOT NULL REFERENCES users(id),    amount       NUMERIC(38, 18) NOT NULL,    event_type   TEXT NOT NULL,    status       TEXT NOT NULL DEFAULT 'pending',    retry_count  INTEGER NOT NULL DEFAULT 0,    created_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),    updated_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),    CONSTRAINT valid_status CHECK (        status IN ('pending', 'enqueued', 'processing', 'confirmed', 'failed')    ));CREATE TABLE balance_events (    id              BIGSERIAL PRIMARY KEY,    user_id         INTEGER NOT NULL REFERENCES users(id),    amount          NUMERIC(38, 18) NOT NULL,    event_type      TEXT NOT NULL,    source_event_id TEXT NOT NULL,    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),    CONSTRAINT uq_balance_events_source UNIQUE (source_event_id, event_type));CREATE TABLE processed_events (    idempotency_key TEXT PRIMARY KEY,    outcome         TEXT NOT NULL DEFAULT 'pending',    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW());CREATE TABLE dead_letter_queue (    id         BIGSERIAL PRIMARY KEY,    event_id   TEXT NOT NULL,    event_type TEXT NOT NULL,    user_id    INTEGER NOT NULL,    amount     NUMERIC(38, 18) NOT NULL,    error      TEXT NOT NULL,    attempt    INTEGER NOT NULL DEFAULT 1,    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW());ALTER TABLE users    ADD COLUMN IF NOT EXISTS initial_balance NUMERIC(38, 18) NOT NULL DEFAULT 0,    ADD CONSTRAINT balance_non_negative CHECK (balance >= 0);CREATE INDEX idx_payment_events_pending    ON payment_events (updated_at, created_at)    WHERE status = 'pending';CREATE INDEX idx_payment_events_enqueued    ON payment_events (updated_at)    WHERE status = 'enqueued';CREATE INDEX idx_payment_events_processing    ON payment_events (updated_at)    WHERE status = 'processing';CREATE INDEX idx_balance_events_user_id    ON balance_events (user_id);CREATE INDEX idx_balance_events_created_at    ON balance_events (created_at DESC);CREATE INDEX idx_processed_events_created    ON processed_events (created_at);CREATE INDEX idx_processed_events_pending_stale    ON processed_events (created_at)    WHERE outcome = 'pending';CREATE INDEX idx_dlq_event_id    ON dead_letter_queue (event_id);CREATE INDEX idx_dlq_created_at    ON dead_letter_queue (created_at DESC);<\/code>
<\/a><\/div><\/pre>\n
\u0412 \u0441\u0445\u0435\u043c\u0435 \u0442\u0440\u0438 \u043d\u0435\u043e\u0447\u0435\u0432\u0438\u0434\u043d\u044b\u0445 \u0440\u0435\u0448\u0435\u043d\u0438\u044f.<\/p>\n
NUMERIC(38, 18)<\/code>, \u043d\u0435 NUMERIC(20, 8)<\/code>. \u041a\u043e\u043b\u043e\u043d\u043a\u0430 amount<\/code> \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f \u0432 ETH, \u043d\u0435 \u0432 wei. Webhook-\u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u043f\u0440\u0438\u0441\u044b\u043b\u0430\u0435\u0442 \u0443\u0436\u0435 \u0441\u043a\u043e\u043d\u0432\u0435\u0440\u0442\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0435 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435. \u0415\u0441\u043b\u0438 \u0432\u0430\u0448 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0430\u0435\u0442 wei, \u043a\u043e\u043d\u0432\u0435\u0440\u0442\u0438\u0440\u0443\u0439\u0442\u0435 \u043d\u0430 \u0432\u0445\u043e\u0434\u0435: amount_eth = Decimal(wei_str) \/ Decimal(10**18)<\/code> \u0434\u043e \u043f\u0435\u0440\u0435\u0434\u0430\u0447\u0438 \u0432 _validate_amount<\/code>  . ERC-20 \u0441\u0430\u043c\u0438 \u043e\u0431\u044a\u044f\u0432\u043b\u044f\u044e\u0442 decimals()<\/code>: USDC\/USDT — 6, WBTC — 8, DAI\/WETH\/MKR — 18. ETH \u0432 wei \u0442\u043e\u0436\u0435 10^18. NUMERIC(20, 8)<\/code> \u0432\u044b\u0434\u0435\u0440\u0436\u0438\u0442 USDC\/USDT, \u043d\u043e \u0444\u0438\u0437\u0438\u0447\u0435\u0441\u043a\u0438 \u043d\u0435 \u0432\u043c\u0435\u0449\u0430\u0435\u0442 18-decimal \u0442\u043e\u043a\u0435\u043d\u044b, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 \u0431\u0435\u0440\u0451\u043c worst case, NUMERIC(38, 18)<\/code>.<\/p>\n
initial_balance<\/code> \u043d\u0443\u0436\u043d\u0430 \u0434\u043b\u044f \u0441\u0432\u0435\u0440\u043a\u0438. \u041f\u0440\u0438 \u043c\u0438\u0433\u0440\u0430\u0446\u0438\u0438 \u0437\u0430\u043f\u043e\u043b\u043d\u044f\u0435\u0442\u0435 \u0435\u0451 \u0442\u0435\u043a\u0443\u0449\u0438\u043c \u0431\u0430\u043b\u0430\u043d\u0441\u043e\u043c, UPDATE users SET initial_balance = balance WHERE <\u0443\u0441\u043b\u043e\u0432\u0438\u0435><\/code>. \u042d\u0442\u043e \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442 \u0447\u0442\u043e balance_events<\/code> \u043d\u0430\u0447\u0438\u043d\u0430\u044e\u0442 \u043d\u0430\u043f\u043e\u043b\u043d\u044f\u0442\u044c\u0441\u044f \u0441 \u043d\u0443\u043b\u044f, \u0438 hot_path_balance_check<\/code> \u043a\u043e\u0440\u0440\u0435\u043a\u0442\u043d\u043e \u0441\u0447\u0438\u0442\u0430\u0435\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u0442\u0435\u0445 \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u0435\u0439, \u0443 \u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u0432\u0441\u0435 \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u0438 \u043f\u0440\u043e\u0448\u043b\u0438 \u0447\u0435\u0440\u0435\u0437 balance_events<\/code>. \u0414\u043b\u044f \u043d\u043e\u0432\u044b\u0445 \u0441\u0438\u0441\u0442\u0435\u043c initial_balance<\/code> \u043e\u0441\u0442\u0430\u0451\u0442\u0441\u044f 0.<\/p>\n
\u041e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0435 \u0438\u043d\u0434\u0435\u043a\u0441\u044b \u0434\u043b\u044f pending\/enqueued\/processing \u0432\u043c\u0435\u0441\u0442\u043e \u043e\u0434\u043d\u043e\u0433\u043e status IN (...)<\/code>, \u0442\u0430\u043a \u043a\u0430\u043a poller’\u044b \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044e\u0442 \u0440\u0430\u0437\u043d\u044b\u0435 \u043f\u0430\u0442\u0442\u0435\u0440\u043d\u044b \u0434\u043e\u0441\u0442\u0443\u043f\u0430. idx_payment_events_pending<\/code>, partial index \u0441 (updated_at, created_at)<\/code> \u0434\u043b\u044f ORDER BY created_at<\/code> \u0432 enqueue_pending_events<\/code>, \u0438\u043d\u0430\u0447\u0435 \u043f\u043b\u0430\u043d\u0438\u0440\u043e\u0432\u0449\u0438\u043a \u0441\u043e\u0440\u0442\u0438\u0440\u0443\u0435\u0442 \u0431\u0435\u0437 \u0438\u043d\u0434\u0435\u043a\u0441\u0430.<\/p>\n
retry_count<\/code> \u0432 payment_events<\/code> \u0434\u043e\u0431\u0430\u0432\u043b\u0435\u043d \u0434\u043b\u044f \u043f\u0440\u0435\u0434\u043e\u0442\u0432\u0440\u0430\u0449\u0435\u043d\u0438\u044f \u0431\u0435\u0441\u043a\u043e\u043d\u0435\u0447\u043d\u043e\u0433\u043e \u0446\u0438\u043a\u043b\u0430 pending -> enqueued<\/code> \u043f\u0440\u0438 durable outage Redis, \u043e\u0431 \u044d\u0442\u043e\u043c \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u0435\u0435 \u0432 \u0441\u0435\u043a\u0446\u0438\u0438 \u043f\u0440\u043e \u0434\u0435\u0433\u0440\u0430\u0434\u0430\u0446\u0438\u044e.<\/p>\n
\u041a\u0430\u043a \u044d\u0442\u043e \u043f\u043e\u0447\u0438\u043d\u0438\u043b\u043e\u0441\u044c  <\/h2>\n
\u0418\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f<\/h4>\n
import osimport uuidimport jsonimport hmacimport randomimport hashlibimport secretsimport threadingimport structlogimport psycopg2import psycopg2.extrasimport psycopg2.poolimport redis as redis_libfrom typing import Literal, Optionalimport refrom decimal import Decimal, InvalidOperationfrom contextvars import ContextVarfrom dataclasses import dataclass, fieldfrom datetime import datetime, timezone, timedeltafrom celery import shared_taskfrom celery.exceptions import Ignore, MaxRetriesExceededErrorlogger = structlog.get_logger()@dataclass(frozen=True)class Settings:    DATABASE_URL:   str    WEBHOOK_SECRET: str    ETH_RPC_URL:    str    ALERT_EMAIL:    str    REDIS_URL:      str = \"redis:\/\/localhost:6379\/0\"    @classmethod    def from_env(cls) -> \"Settings\":        required = (\"DATABASE_URL\", \"WEBHOOK_SECRET\", \"ETH_RPC_URL\", \"ALERT_EMAIL\")        missing  = [k for k in required if not os.environ.get(k)]        if missing:            raise RuntimeError(f\"Missing required env vars: {', '.join(missing)}\")        return cls(            DATABASE_URL   = os.environ[\"DATABASE_URL\"],            WEBHOOK_SECRET = os.environ[\"WEBHOOK_SECRET\"],            ETH_RPC_URL    = os.environ[\"ETH_RPC_URL\"],            ALERT_EMAIL    = os.environ[\"ALERT_EMAIL\"],            REDIS_URL      = os.environ.get(\"REDIS_URL\", \"redis:\/\/localhost:6379\/0\"),        )settings = Settings.from_env()_redis_client = redis_lib.Redis.from_url(    settings.REDIS_URL,    decode_responses=True,    socket_connect_timeout=2,    socket_timeout=2,    retry_on_timeout=False,)<\/code>
<\/a><\/div><\/pre>\n
send_alert<\/code>: rate-limited \u043e\u0431\u0451\u0440\u0442\u043a\u0430 \u043d\u0430\u0434 \u043b\u043e\u0433\u0433\u0435\u0440\u043e\u043c. \u0412 \u043f\u0440\u043e\u0434\u0435 \u0437\u0430\u043c\u0435\u043d\u044f\u0435\u0442\u0441\u044f \u043d\u0430 PagerDuty\/OpsGenie SDK. \u041e\u0434\u0438\u043d\u0430\u043a\u043e\u0432\u044b\u0435 alert_key<\/code> \u0432\u043d\u0443\u0442\u0440\u0438 cooldown \u043e\u043a\u043d\u0430 \u043f\u043e\u0434\u0430\u0432\u043b\u044f\u044e\u0442\u0441\u044f. \u0415\u0441\u043b\u0438 \u043a\u043b\u044e\u0447 \u043d\u0435 \u0437\u0430\u0434\u0430\u043d, \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0430 \u0431\u0435\u0437 rate-limit, \u0434\u043b\u044f \u043e\u0434\u043d\u043e\u0440\u0430\u0437\u043e\u0432\u044b\u0445 \u043a\u0440\u0438\u0442\u0438\u0447\u043d\u044b\u0445 \u0430\u043b\u0435\u0440\u0442\u043e\u0432. \u041d\u0435 \u0431\u0440\u043e\u0441\u0430\u0435\u0442 \u0438\u0441\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u0439 \u043d\u0438\u043a\u043e\u0433\u0434\u0430.<\/p>\n
_alert_last_sent<\/code> \u0440\u0430\u0441\u0442\u0451\u0442 \u043f\u0440\u0438 \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0445 \u043a\u043b\u044e\u0447\u0430\u0445. \u0415\u0441\u043b\u0438 \u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043a\u043b\u044e\u0447\u0438 per-event-id (\u0430 \u043c\u044b \u0442\u0430\u043a \u0438 \u0434\u0435\u043b\u0430\u0435\u043c \u0434\u043b\u044f orphan-\u0430\u043b\u0435\u0440\u0442\u043e\u0432), \u0437\u0430 \u043c\u0435\u0441\u044f\u0446 \u044d\u0442\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043c\u0438\u043b\u043b\u0438\u043e\u043d\u043e\u0432 \u0437\u0430\u043f\u0438\u0441\u0435\u0439. \u041f\u043e\u044d\u0442\u043e\u043c\u0443 \u043f\u0440\u0438 \u043f\u0435\u0440\u0435\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u0438 \u0441\u043d\u0430\u0447\u0430\u043b\u0430 \u0447\u0438\u0441\u0442\u0438\u043c \u0443\u0441\u0442\u0430\u0440\u0435\u0432\u0448\u0438\u0435 \u043a\u043b\u044e\u0447\u0438, \u0430 \u0435\u0441\u043b\u0438 \u043f\u043e\u0441\u043b\u0435 \u0447\u0438\u0441\u0442\u043a\u0438 \u043c\u0435\u0441\u0442\u043e \u0432\u0441\u0451 \u0440\u0430\u0432\u043d\u043e \u043d\u0435 \u043e\u0441\u0432\u043e\u0431\u043e\u0434\u0438\u043b\u043e\u0441\u044c, \u043f\u043e\u0434\u0430\u0432\u043b\u044f\u0435\u043c \u043d\u043e\u0432\u044b\u0435. \u041a\u043e\u0441\u0442\u044b\u043b\u044c, \u0434\u0430. \u041d\u043e \u0437\u0430 \u0432\u043e\u0441\u0435\u043c\u044c \u043c\u0435\u0441\u044f\u0446\u0435\u0432 \u043d\u0435 \u043e\u0442\u0432\u0430\u043b\u0438\u043b\u0441\u044f.<\/p>\n
_alert_lock = threading.Lock()_alert_last_sent: dict = {}MAX_ALERT_KEYS = 1_000def send_alert(message: str, alert_key: Optional[str] = None,               cooldown_seconds: int = 300) -> None:    try:        if alert_key is None:            logger.critical(\"ALERT\", message=message)            return        with _alert_lock:            now = datetime.now(timezone.utc)            if len(_alert_last_sent) >= MAX_ALERT_KEYS:                stale_cutoff = now - timedelta(seconds=cooldown_seconds * 2)                stale = [k for k, v in _alert_last_sent.items() if v < stale_cutoff]                for k in stale:                    del _alert_last_sent[k]                if len(_alert_last_sent) >= MAX_ALERT_KEYS and alert_key not in _alert_last_sent:                    logger.warning(\"send_alert suppressed: rate limit dict full\",                                   alert_key=alert_key)                    return            last = _alert_last_sent.get(alert_key)            if last and (now - last).total_seconds() < cooldown_seconds:                return            _alert_last_sent[alert_key] = now        logger.critical(\"ALERT\", message=message, alert_key=alert_key)    except Exception as e:        logger.error(\"send_alert failed\", error=str(e))class ImproperlyConfigured(RuntimeError):    pass<\/code>
<\/a><\/div><\/pre>\n
Trace ID \u0447\u0435\u0440\u0435\u0437 \u0432\u0441\u044e \u0446\u0435\u043f\u043e\u0447\u043a\u0443<\/h4>\n
\u041a\u0430\u0436\u0434\u044b\u0439 \u0432\u043e\u0440\u043a\u0435\u0440 \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u0442 \u0441\u0432\u043e\u0439 ContextVar<\/code> \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438, \u0448\u0430\u0440\u0438\u0442\u044c \u0435\u0433\u043e \u043c\u0435\u0436\u0434\u0443 \u043f\u043e\u0442\u043e\u043a\u0430\u043c\u0438 \u043d\u0435\u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e.  <\/p>\n
_trace_id: ContextVar[str] = ContextVar('trace_id', default='')def get_trace_id() -> str:    return _trace_id.get() or 'no-trace'def set_trace_id(tid: str) -> None:    _trace_id.set(tid)def new_trace_id() -> str:    tid = str(uuid.uuid4())    _trace_id.set(tid)    return tidstructlog.configure(    processors=[        structlog.processors.add_log_level,        lambda _, __, event_dict: {**event_dict, \"trace_id\": get_trace_id()},        structlog.processors.JSONRenderer(),    ])<\/code>
<\/a><\/div><\/pre>\n
Idempotency key \u0447\u0435\u0440\u0435\u0437 DB unique constraint<\/h4>\n
\u041a\u043b\u044e\u0447 \u0441\u0442\u0440\u043e\u0438\u0442\u0441\u044f \u0438\u0437 event_id<\/code> \u0438 event_type<\/code>, \u043f\u0438\u0448\u0435\u0442\u0441\u044f \u0432 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u0443\u044e \u0442\u0430\u0431\u043b\u0438\u0446\u0443 \u0441 unique constraint \u0432 \u0442\u043e\u0439 \u0436\u0435 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0438 \u0447\u0442\u043e \u0438 \u0438\u0437\u043c\u0435\u043d\u0435\u043d\u0438\u0435 \u0431\u0430\u043b\u0430\u043d\u0441\u0430.<\/p>\n
Redis \u043d\u0435 \u043f\u043e\u0434\u0445\u043e\u0434\u0438\u0442, \u043d\u0435\u0442 \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e\u0441\u0442\u0438 \u0441 PostgreSQL. \u041f\u0440\u043e\u0432\u0435\u0440\u043a\u0430 \u0432 \u043a\u043e\u0434\u0435 \u0442\u043e\u0436\u0435 \u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442, \u0434\u0432\u0430 \u0432\u043e\u0440\u043a\u0435\u0440\u0430 \u043e\u0431\u0430 \u043f\u0440\u043e\u0439\u0434\u0443\u0442 SELECT \u0434\u043e INSERT. \u0415\u0434\u0438\u043d\u0441\u0442\u0432\u0435\u043d\u043d\u043e\u0435, \u0447\u0442\u043e \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e \u043f\u043e \u043e\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u0438\u044e: unique constraint.<\/p>\n
\u0421\u043d\u0430\u0447\u0430\u043b\u0430 \u0434\u0435\u043b\u0430\u043b \u043a\u043e\u043d\u043a\u0430\u0442\u0435\u043d\u0430\u0446\u0438\u044e f\"{event_id}::{event_type}\"<\/code>. \u0421\u043b\u043e\u0432\u0438\u043b \u043a\u043e\u043b\u043b\u0438\u0437\u0438\u044e \u043f\u0440\u0438 ::<\/code> \u0432 event_id<\/code>. \u041f\u043e\u043f\u0440\u043e\u0431\u043e\u0432\u0430\u043b NUL-\u0440\u0430\u0437\u0434\u0435\u043b\u0438\u0442\u0435\u043b\u044c: f\"{event_id}\\0{event_type}\".encode()<\/code>. \u0422\u043e\u0436\u0435 \u043a\u043e\u043b\u043b\u0438\u0437\u0438\u044f:   _idempotency_key(\"a\\x00b\", \"c\") == _idempotency_key(\"a\", \"b\\x00c\")<\/code>, \u043e\u0431\u0430 \u0434\u0430\u044e\u0442 \u0431\u0430\u0439\u0442\u044b b\"a\\x00b\\x00c\"<\/code>. \u0424\u0438\u043d\u0430\u043b\u044c\u043d\u044b\u0439 \u0432\u0430\u0440\u0438\u0430\u043d\u0442, length-prefix encoding: \u043a\u0430\u0436\u0434\u043e\u0435 \u043f\u043e\u043b\u0435 \u043f\u0440\u0435\u0434\u0432\u0430\u0440\u044f\u0435\u0442\u0441\u044f 4-\u0431\u0430\u0439\u0442\u043e\u0432\u043e\u0439 \u0434\u043b\u0438\u043d\u043e\u0439, \u043a\u043e\u043b\u043b\u0438\u0437\u0438\u044f \u043c\u0435\u0436\u0434\u0443 \u043f\u043e\u043b\u044f\u043c\u0438 \u043f\u0440\u0438\u043d\u0446\u0438\u043f\u0438\u0430\u043b\u044c\u043d\u043e \u043d\u0435\u0432\u043e\u0437\u043c\u043e\u0436\u043d\u0430.<\/p>\n
class RetryableError(Exception):    passclass AlreadyProcessedError(Exception):    passMAX_AMOUNT = Decimal(\"10\") ** 20_AMOUNT_RE = re.compile(r\"^[0-9]+(\\.[0-9]+)?([eE][+-]?[0-9]+)?$\")def _validate_amount(amount) -> Decimal:    if isinstance(amount, float):        raise ValueError(            f\"float \u043d\u0435 \u0434\u043e\u043f\u0443\u0441\u043a\u0430\u0435\u0442\u0441\u044f \u2014 \u043f\u0435\u0440\u0435\u0434\u0430\u0432\u0430\u0439\u0442\u0435 amount \u043a\u0430\u043a str \u0438\u0437 JSON payload. \"            f\"\u041f\u043e\u043b\u0443\u0447\u0435\u043d\u043e: {amount!r}\"        )    if isinstance(amount, str) and amount != amount.strip():        raise ValueError(            f\"amount \u0441\u043e\u0434\u0435\u0440\u0436\u0438\u0442 whitespace: {amount!r}. \"            f\"\u041f\u0435\u0440\u0435\u0434\u0430\u0432\u0430\u0439\u0442\u0435 amount \u0431\u0435\u0437 \u043f\u0440\u043e\u0431\u0435\u043b\u043e\u0432.\"        )    if isinstance(amount, str) and not _AMOUNT_RE.fullmatch(amount):        raise ValueError(f\"invalid amount format: {amount!r}\")    try:        amount_decimal = Decimal(str(amount))        if not amount_decimal.is_finite():            raise ValueError(f\"amount must be finite, got {amount_decimal}\")        if amount_decimal <= 0:            raise ValueError(f\"amount must be positive, got {amount_decimal}\")        if amount_decimal.normalize().as_tuple().exponent < -18:            raise ValueError(f\"amount precision exceeds 18 decimals: {amount_decimal}\")        if amount_decimal >= MAX_AMOUNT:            raise ValueError(                f\"amount exceeds NUMERIC(38,18) capacity: {amount_decimal} >= 10^20\"            )        return amount_decimal    except InvalidOperation:        raise ValueError(f\"invalid amount format: {amount!r}\")def _idempotency_key(event_id: str, event_type: str) -> str:    a = event_id.encode(\"utf-8\")    b = event_type.encode(\"utf-8\")    payload = (        len(a).to_bytes(4, \"big\") + a +        len(b).to_bytes(4, \"big\") + b    )    return hashlib.sha256(payload).hexdigest()<\/code>
<\/a><\/div><\/pre>\n
\u0414\u0432\u0430 \u043c\u0435\u0441\u0442\u0430 \u0432 \u043f\u0435\u0440\u0432\u043e\u0439 \u0432\u0435\u0440\u0441\u0438\u0438 \u0431\u044b\u043b\u0438 \u0441\u043b\u043e\u043c\u0430\u043d\u044b \u043d\u0430 \u0433\u0440\u0430\u043d\u0438\u0447\u043d\u044b\u0445 \u0441\u043b\u0443\u0447\u0430\u044f\u0445: MAX_AMOUNT = Decimal(\"10\")**20 - 1<\/code> \u043e\u0442\u0432\u0435\u0440\u0433\u0430\u043b\u0430 \u0432\u0430\u043b\u0438\u0434\u043d\u0443\u044e \u0441\u0443\u043c\u043c\u0443, \u0430 50.0000000000000000000<\/code> \u0443\u0445\u043e\u0434\u0438\u043b\u043e \u0432 \u043e\u0442\u043a\u0430\u0437 \u043a\u0430\u043a exponent=-19<\/code> \u0445\u043e\u0442\u044f \u0437\u043d\u0430\u0447\u0430\u0449\u0438\u0445 \u0446\u0438\u0444\u0440 \u0442\u0430\u043c \u043d\u0435\u0442. \u041f\u043e\u0447\u0438\u043d\u0435\u043d\u043e: 10**20<\/code> \u0431\u0435\u0437 -1<\/code>, \u0438 normalize()<\/code> \u043f\u0435\u0440\u0435\u0434 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u043e\u0439 exponent. \u041d\u0430 \u0433\u0440\u0430\u043d\u0438\u0447\u043d\u044b\u0435 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u044f \u0441\u0432\u043e\u0438\u0445 \u0432\u0430\u043b\u0438\u0434\u0430\u0442\u043e\u0440\u043e\u0432 \u0441\u0442\u043e\u0438\u0442 \u043f\u0438\u0441\u0430\u0442\u044c \u0442\u0435\u0441\u0442\u044b, \u0443\u0437\u043d\u0430\u0451\u0448\u044c \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u043d\u043e\u0435.<\/p>\n
\u0415\u0449\u0451 \u043e\u0434\u0438\u043d \u0441\u044e\u0440\u043f\u0440\u0438\u0437 \u0438\u0437 \u0442\u043e\u0439 \u0436\u0435 \u0441\u0435\u0440\u0438\u0438: _validate_amount(\"+50.1\")<\/code>,_validate_amount(\"1_000\")<\/code> \u0438 _validate_amount(\"\u0661\u0662\u0663\")<\/code> \u0432\u0441\u0435 \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0430\u044e\u0442 \u043a\u043e\u0440\u0440\u0435\u043a\u0442\u043d\u044b\u0439 Decimal<\/code>. Python \u0442\u043e\u043b\u0435\u0440\u0430\u043d\u0442\u0435\u043d \u043a underscore-\u043d\u043e\u0442\u0430\u0446\u0438\u0438, leading +<\/code> \u0438 \u0430\u0440\u0430\u0431\u043e-\u0438\u043d\u0434\u0438\u0439\u0441\u043a\u0438\u043c \u0446\u0438\u0444\u0440\u0430\u043c. \u0414\u043b\u044f \u0444\u0438\u043d\u0430\u043d\u0441\u043e\u0432\u043e\u0433\u043e \u0432\u0430\u043b\u0438\u0434\u0430\u0442\u043e\u0440\u0430 \u044d\u0442\u043e \u043d\u0435\u0436\u0435\u043b\u0430\u0442\u0435\u043b\u044c\u043d\u043e\u0435 \u043f\u043e\u0432\u0435\u0434\u0435\u043d\u0438\u0435, \u043d\u0430 \u0432\u0445\u043e\u0434\u0435 \u043e\u0436\u0438\u0434\u0430\u0435\u0442\u0441\u044f \u0441\u0442\u0440\u043e\u0433\u043e [\u0446\u0438\u0444\u0440\u044b].[\u0446\u0438\u0444\u0440\u044b]<\/code>. \u0414\u043e\u0431\u0430\u0432\u043b\u0435\u043d regex ^[0-9]+(\\.[0-9]+)?([eE][+-]?[0-9]+)?$<\/code> \u043f\u0435\u0440\u0435\u0434 Decimal()<\/code>, \u043e\u0442\u043a\u043b\u043e\u043d\u044f\u0435\u0442 \u0432\u0441\u0451 \u043d\u0435\u0441\u0442\u0430\u043d\u0434\u0430\u0440\u0442\u043d\u043e\u0435.<\/p>\n
FSM \u043f\u0435\u0440\u0435\u0445\u043e\u0434\u043e\u0432, \u043e\u0434\u043d\u0430 \u0442\u043e\u0447\u043a\u0430 \u043f\u0440\u0430\u0432\u0434\u044b  <\/h4>\n
\u0421\u0442\u0430\u0442\u0443\u0441 \u0441\u043e\u0431\u044b\u0442\u0438\u044f — \u0434\u0435\u0442\u0435\u0440\u043c\u0438\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u044b\u0439 \u043a\u043e\u043d\u0435\u0447\u043d\u044b\u0439 \u0430\u0432\u0442\u043e\u043c\u0430\u0442. \u0421\u043d\u0430\u0447\u0430\u043b\u0430 \u0431\u044b\u043b\u043e \u0442\u0440\u0438 \u043c\u0435\u0441\u0442\u0430 \u0441 raw UPDATE payment_events SET status = ...<\/code>. \u042d\u0442\u043e \u043d\u0430\u0440\u0443\u0448\u0430\u043b\u043e \u0438\u043d\u0432\u0430\u0440\u0438\u0430\u043d\u0442 FSM.<\/p>\n
\u041f\u0435\u0440\u0435\u0445\u043e\u0434\u044b: pending<\/code> \u0438\u0434\u0451\u0442 \u0432 enqueued<\/code> \u0447\u0435\u0440\u0435\u0437 poller. Enqueued<\/code> \u0432 processing<\/code> \u043a\u043e\u0433\u0434\u0430 \u0432\u043e\u0440\u043a\u0435\u0440 \u0432\u0437\u044f\u043b \u0437\u0430\u0434\u0430\u0447\u0443. \u0418\u0437 processing<\/code> \u0442\u043e\u043b\u044c\u043a\u043e \u0432 confirmed<\/code> \u0438\u043b\u0438 failed<\/code>. \u041e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 edge, enqueued<\/code> \u0441\u0440\u0430\u0437\u0443 \u0432 confirmed<\/code>, \u043d\u0443\u0436\u0435\u043d \u0434\u043b\u044f replay path: \u043a\u043e\u0433\u0434\u0430 processed_events<\/code> \u0443\u0436\u0435 \u0441\u043e\u0434\u0435\u0440\u0436\u0438\u0442 outcome=success<\/code>, \u043d\u043e \u0432\u043e\u0440\u043a\u0435\u0440 \u0443\u043f\u0430\u043b \u043f\u043e\u0441\u043b\u0435 \u0442\u043e\u0433\u043e \u043a\u0430\u043a \u0437\u0430\u043f\u0438\u0441\u0430\u043b \u044d\u0442\u043e \u0438 \u0434\u043e \u0442\u043e\u0433\u043e \u043a\u0430\u043a \u0443\u0441\u043f\u0435\u043b \u043f\u0435\u0440\u0435\u0432\u0435\u0441\u0442\u0438 payment_events<\/code> \u0432 confirmed<\/code>. \u0411\u0435\u0437 \u044d\u0442\u043e\u0433\u043e edge \u0441\u043e\u0431\u044b\u0442\u0438\u044f \u0437\u0430\u0432\u0438\u0441\u0430\u043b\u0438 \u0431\u044b \u0432 enqueued<\/code> \u0432\u0435\u0447\u043d\u043e.<\/p>\n
VALID_TRANSITIONS: dict[str, set[str]] = {    \"pending\":    {\"enqueued\", \"failed\"},    \"enqueued\":   {\"processing\", \"failed\", \"pending\", \"confirmed\"},    \"processing\": {\"confirmed\", \"failed\"},}TERMINAL_STATUSES = frozenset({\"confirmed\", \"failed\"})def transition_event_status(cur, event_id, from_status, to_status):    if to_status not in VALID_TRANSITIONS.get(from_status, set()):        raise ValueError(f\"\u043d\u0435\u0432\u0430\u043b\u0438\u0434\u043d\u044b\u0439 \u043f\u0435\u0440\u0435\u0445\u043e\u0434: {from_status} -> {to_status}\")    cur.execute(        \"UPDATE payment_events SET status = %s, updated_at = NOW() \"        \"WHERE event_id = %s AND status = %s\",        (to_status, event_id, from_status),    )    if cur.rowcount == 0:        cur.execute(\"SELECT status FROM payment_events WHERE event_id = %s\", (event_id,))        actual = cur.fetchone()        actual_status = actual[\"status\"] if actual else \"not found\"        if actual_status == to_status:            logger.info(\"status already set\", event_id=event_id, status=to_status)            return        if actual_status in TERMINAL_STATUSES:            raise AlreadyProcessedError(f\"event already terminal: {actual_status}\")        raise RetryableError(            f\"concurrent status transition event_id={event_id} \"            f\"expected={from_status} actual={actual_status}\"        )<\/code>
<\/a><\/div><\/pre>\n
_mark_event_failed<\/code>: \u0431\u0435\u0437\u043e\u043f\u0430\u0441\u043d\u044b\u0439 \u043f\u0435\u0440\u0435\u0432\u043e\u0434 \u0432 failed<\/code> \u0438\u0437 \u043b\u044e\u0431\u043e\u0433\u043e \u043d\u0435\u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u044c\u043d\u043e\u0433\u043e \u0441\u0442\u0430\u0442\u0443\u0441\u0430. \u041a\u043e\u043c\u043c\u0438\u0442\u0438\u0442 \u0441\u0430\u043c\u0430, \u043f\u0440\u0430\u0432\u0438\u043b\u043e \u043e\u0434\u043d\u043e, \u0441\u0442\u0430\u0442\u0443\u0441 failed<\/code> \u0434\u043e\u043b\u0436\u0435\u043d \u043b\u0435\u0447\u044c \u0432 \u0411\u0414 \u043d\u0435\u0441\u043c\u043e\u0442\u0440\u044f \u043d\u0438 \u043d\u0430 \u0447\u0442\u043e. \u0412\u0441\u0451 \u043e\u0441\u0442\u0430\u043b\u044c\u043d\u043e\u0435 \u043f\u043e\u0442\u043e\u043c. \u0412\u044b\u0437\u044b\u0432\u0430\u0442\u044c \u0442\u043e\u043b\u044c\u043a\u043e \u043d\u0430 \u0447\u0438\u0441\u0442\u043e\u043c \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0438, \u043f\u043e\u0441\u043b\u0435 rollback.<\/p>\n
_mark_event_failed<\/strong><\/code> \u043a\u043e\u043c\u043c\u0438\u0442\u0438\u0442 \u0441\u0430\u043c\u0430, \u0435\u0441\u043b\u0438 \u0431\u0443\u0434\u0435\u0448\u044c \u0440\u0435\u0444\u0430\u043a\u0442\u043e\u0440\u0438\u0442\u044c, \u044d\u0442\u043e \u0442\u0435\u0431\u044f \u0443\u043a\u0443\u0441\u0438\u0442.   <\/strong><\/p>\n
def _mark_event_failed(conn, event_id) -> bool:    try:        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:            cur.execute(\"SET LOCAL lock_timeout = '2s'\")            cur.execute(                \"SELECT status FROM payment_events WHERE event_id = %s FOR UPDATE NOWAIT\",                (event_id,)            )            row = cur.fetchone()            if row is None:                conn.rollback()                return False            current = row[\"status\"]            if current in TERMINAL_STATUSES:                conn.rollback()                return False            try:                transition_event_status(cur, event_id, current, \"failed\")                conn.commit()                return True            except (ValueError, RetryableError, AlreadyProcessedError):                conn.rollback()                return False    except (psycopg2.errors.LockNotAvailable, psycopg2.errors.QueryCanceled):        try:            conn.rollback()        except Exception:            pass        return False    except Exception:        try:            conn.rollback()        except Exception:            pass        raise<\/code>
<\/a><\/div><\/pre>\n
SELECT FOR UPDATE NOWAIT + lock_timeout  <\/h4>\n
\u0421 \u043e\u0431\u044b\u0447\u043d\u044b\u043c FOR UPDATE<\/code> \u0432\u043e\u0440\u043a\u0435\u0440 \u043c\u043e\u043b\u0447\u0430 \u0436\u0434\u0451\u0442 \u043b\u043e\u043a, \u0431\u043b\u043e\u043a\u0438\u0440\u0443\u044f thread. NOWAIT<\/code> \u044d\u0442\u043e \u0443\u0431\u0438\u0440\u0430\u0435\u0442.<\/p>\n
\u041c\u0438\u0433\u0440\u0430\u0446\u0438\u044f \u0441 ALTER TABLE<\/code> \u0431\u043b\u043e\u043a\u0438\u0440\u0443\u0435\u0442 \u0442\u0430\u0431\u043b\u0438\u0446\u0443 \u0446\u0435\u043b\u0438\u043a\u043e\u043c \u0438 \u043f\u0440\u0438 \u044d\u0442\u043e\u043c lock_timeout = '2s'<\/code> \u043d\u0435 \u0434\u0430\u0451\u0442 \u0432\u043e\u0440\u043a\u0435\u0440\u0443 \u0432\u0438\u0441\u0435\u0442\u044c \u0432\u0441\u0451 \u044d\u0442\u043e \u0432\u0440\u0435\u043c\u044f.<\/p>\n
\u041f\u0440\u043e \u043a\u043e\u0434\u044b \u043e\u0448\u0438\u0431\u043e\u043a: lock_timeout<\/code> \u0431\u0440\u043e\u0441\u0430\u0435\u0442 LockNotAvailable<\/code> (pgcode 55P03<\/code>), statement_timeout<\/code> \u0431\u0440\u043e\u0441\u0430\u0435\u0442 QueryCanceled<\/code> (pgcode 57014<\/code>). \u041f\u0443\u0442\u0430\u043d\u0438\u0446\u0430 \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u0442 \u043a \u043d\u0435\u043f\u043e\u0439\u043c\u0430\u043d\u043d\u043e\u043c\u0443 \u0438\u0441\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u044e \u0432 \u043f\u0440\u043e\u0434\u0430\u043a\u0448\u0435\u043d\u0435. DeadlockDetected<\/code> (pgcode 40P01<\/code>): \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u0443\u0431\u0438\u0442\u0430 PostgreSQL \u0438\u0437-\u0437\u0430 \u0446\u0438\u043a\u043b\u0430 \u0431\u043b\u043e\u043a\u0438\u0440\u043e\u0432\u043e\u043a, \u0442\u043e\u0436\u0435 transient, \u0442\u043e\u0436\u0435 retryable. PostgreSQL \u0441\u0430\u043c \u0432\u044b\u0431\u0438\u0440\u0430\u0435\u0442 \u0436\u0435\u0440\u0442\u0432\u0443 \u0438 \u043e\u0442\u043a\u0430\u0442\u044b\u0432\u0430\u0435\u0442 \u0435\u0451 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044e, \u043f\u043e\u0432\u0442\u043e\u0440 \u0440\u0435\u0448\u0430\u0435\u0442 \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0443. \u0412\u0441\u0435 \u0442\u0440\u0438 \u043d\u0443\u0436\u043d\u043e \u043b\u043e\u0432\u0438\u0442\u044c \u0432\u043c\u0435\u0441\u0442\u0435.<\/p>\n
\u041f\u0443\u043b \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0439 \u0441 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u0435\u0439  <\/h4>\n
\u0421\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0432 \u043f\u0443\u043b\u0435 \u043c\u043e\u0436\u0435\u0442 \u0431\u044b\u0442\u044c \u043c\u0451\u0440\u0442\u0432\u044b\u043c: PostgreSQL \u0437\u0430\u043a\u0440\u044b\u0432\u0430\u0435\u0442 idle-\u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u044f \u0447\u0435\u0440\u0435\u0437 tcp_keepalives_idle<\/code> \u0438\u043b\u0438 idle_in_transaction_session_timeout<\/code>. \u0411\u0435\u0437 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438 \u0432\u043e\u0440\u043a\u0435\u0440 \u043f\u043e\u043b\u0443\u0447\u0438\u0442 \u0440\u0430\u0437\u043e\u0440\u0432\u0430\u043d\u043d\u044b\u0439 TCP \u0438 \u0443\u043f\u0430\u0434\u0451\u0442 \u0441 InterfaceError<\/code> \u0432 \u0441\u043b\u0443\u0447\u0430\u0439\u043d\u044b\u0439 \u043c\u043e\u043c\u0435\u043d\u0442.<\/p>\n
_PooledConn<\/code>: \u043e\u0431\u0451\u0440\u0442\u043a\u0430 \u043d\u0430\u0434 \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435\u043c, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u0437\u043d\u0430\u0435\u0442 \u043a\u0430\u043a \u0432\u0435\u0440\u043d\u0443\u0442\u044c \u0435\u0433\u043e \u0432\u043b\u0430\u0434\u0435\u043b\u044c\u0446\u0443. \u0427\u0435\u0440\u0435\u0437 putconn()<\/code> \u043e\u0431\u0440\u0430\u0442\u043d\u043e \u0432 \u043f\u0443\u043b \u0438\u043b\u0438 \u0447\u0435\u0440\u0435\u0437 close()<\/code> \u0435\u0441\u043b\u0438 \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0441\u043e\u0437\u0434\u0430\u0432\u0430\u043b\u043e\u0441\u044c \u043d\u0430\u043f\u0440\u044f\u043c\u0443\u044e. putconn()<\/code> \u0438\u0434\u0435\u043c\u043f\u043e\u0442\u0435\u043d\u0442\u0435\u043d, \u0432\u0442\u043e\u0440\u043e\u0439 \u0432\u044b\u0437\u043e\u0432 no-op. getattr<\/code> \u043d\u0435 \u043f\u0440\u043e\u043a\u0441\u0438\u0440\u0443\u0435\u0442 dunder-\u043c\u0435\u0442\u043e\u0434\u044b, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 _PooledConn<\/code> \u043d\u0435\u043b\u044c\u0437\u044f \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u043a\u0430\u043a context manager. \u0412\u0435\u0441\u044c \u043a\u043e\u0434 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442 \u0447\u0435\u0440\u0435\u0437 conn.cursor()<\/code>.<\/p>\n
get_validated_conn<\/code> \u0434\u0435\u043b\u0430\u0435\u0442 \u0442\u0440\u0438 \u0443\u0440\u043e\u0432\u043d\u044f \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438 \u0431\u0435\u0437 I\/O \u0432 \u043d\u043e\u0440\u043c\u0430\u043b\u044c\u043d\u043e\u043c \u043f\u043e\u0442\u043e\u043a\u0435: \u0441\u043d\u0430\u0447\u0430\u043b\u0430 conn.closed<\/code> (in-memory \u0444\u043b\u0430\u0433), \u043f\u043e\u0442\u043e\u043c conn.status<\/code> (\u0433\u0440\u044f\u0437\u043d\u0430\u044f \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u0438\u0437 \u043f\u0440\u0435\u0434\u044b\u0434\u0443\u0449\u0435\u0433\u043e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043d\u0438\u044f), \u0438 \u0442\u043e\u043b\u044c\u043a\u043e \u0435\u0441\u043b\u0438 \u0441\u0442\u0430\u0442\u0443\u0441 \u043d\u0435 STATUS_READY<\/code>, \u0434\u0435\u043b\u0430\u0435\u0442 SELECT 1<\/code>.<\/p>\n
class _PooledConn:    def __init__(self, conn, pool=None):        self._conn = conn        self._pool = pool    def putconn(self, close=False):        if self._pool is None:            try:                self._conn.close()            except Exception:                pass            return        pool, self._pool = self._pool, None  # idempotent: second call is a no-op        try:            pool.putconn(self._conn, close=close)        except Exception:            pass    def __getattr__(self, name):        return getattr(self._conn, name)def get_validated_conn(pool: psycopg2.pool.SimpleConnectionPool) -> \"_PooledConn\":    try:        conn = pool.getconn()    except psycopg2.pool.PoolError as e:        raise RetryableError(f\"DB connection pool exhausted: {e}\")    if conn.closed != 0:        try:            pool.putconn(conn, close=True)        except Exception:            pass        direct = psycopg2.connect(dsn=settings.DATABASE_URL)        return _PooledConn(direct, pool=None)    if conn.status == psycopg2.extensions.STATUS_IN_TRANSACTION:        try:            conn.rollback()            logger.warning(\"get_validated_conn: rolled back dirty connection\")        except Exception:            try:                pool.putconn(conn, close=True)            except Exception:                pass            direct = psycopg2.connect(dsn=settings.DATABASE_URL)            return _PooledConn(direct, pool=None)    if conn.status != psycopg2.extensions.STATUS_READY:        try:            with conn.cursor() as cur:                cur.execute(\"SELECT 1\")        except Exception:            try:                pool.putconn(conn, close=True)            except Exception:                pass            direct = psycopg2.connect(dsn=settings.DATABASE_URL)            return _PooledConn(direct, pool=None)    return _PooledConn(conn, pool)<\/code>
<\/a><\/div><\/pre>\n
Deposit \u0438 withdrawal  <\/h4>\n
\u041f\u0440\u0438 INSERT INTO processed_events<\/code> \u0434\u0432\u0430 \u0438\u0441\u0445\u043e\u0434\u0430: \u0443\u0441\u043f\u0435\u0445, \u0438\u0434\u0451\u043c \u0434\u0430\u043b\u044c\u0448\u0435 (first-time path). UniqueViolation<\/code>, \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0443\u0436\u0435 \u0432\u0438\u0434\u0435\u043b\u0438 (replay path).<\/p>\n
\u041d\u0430 replay path \u0441\u043c\u043e\u0442\u0440\u0438\u043c outcome<\/code>. \u0415\u0441\u043b\u0438 success<\/code>, \u0441\u0438\u043d\u0445\u0440\u043e\u043d\u0438\u0437\u0438\u0440\u0443\u0435\u043c \u0441\u0442\u0430\u0442\u0443\u0441 payment_events<\/code> \u0441 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u0441\u0442\u044c\u044e. \u0415\u0441\u043b\u0438 pending<\/code>, \u0434\u0440\u0443\u0433\u043e\u0439 \u0432\u043e\u0440\u043a\u0435\u0440 \u0435\u0449\u0451 \u0432 \u0441\u0435\u0440\u0435\u0434\u0438\u043d\u0435 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0438, \u0431\u0440\u043e\u0441\u0430\u0435\u043c RetryableError<\/code> \u0434\u043b\u044f \u043d\u0435\u043c\u0435\u0434\u043b\u0435\u043d\u043d\u043e\u0433\u043e retry \u0432\u043c\u0435\u0441\u0442\u043e \u043e\u0436\u0438\u0434\u0430\u043d\u0438\u044f recover_stale_enqueued_events<\/code> \u0447\u0435\u0440\u0435\u0437 3 \u043c\u0438\u043d\u0443\u0442\u044b.<\/p>\n
retry_count<\/code> \u0441\u0431\u0440\u0430\u0441\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0432 0 \u043f\u0440\u0438 \u0443\u0441\u043f\u0435\u0448\u043d\u043e\u0439 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0435. \u0411\u0435\u0437 \u044d\u0442\u043e\u0433\u043e: \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u043f\u043e\u043f\u0430\u043b\u043e \u0432 retry, \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u0430\u043b\u043e\u0441\u044c \u0443\u0441\u043f\u0435\u0448\u043d\u043e \u0441 retry_count=7<\/code>, \u0447\u0435\u0440\u0435\u0437 14+ \u0434\u043d\u0435\u0439 processed_events<\/code> \u0432\u044b\u0447\u0438\u0441\u0442\u0438\u043b\u0441\u044f \u043f\u043e TTL, \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u043f\u0440\u0438\u0448\u043b\u043e \u0441\u043d\u043e\u0432\u0430 \u0447\u0435\u0440\u0435\u0437 reorg compensation. \u0421\u0442\u0430\u0440\u0442\u0443\u0435\u0442 \u0443\u0436\u0435 \u0441 7\/10 \u0434\u043e DLQ \u0432\u043c\u0435\u0441\u0442\u043e 0\/10.<\/p>\n
\u041e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 \u0441\u043b\u0443\u0447\u0430\u0439 \u0432 replay: processed_events<\/code> \u0433\u043e\u0432\u043e\u0440\u0438\u0442 \u0447\u0442\u043e \u0432\u0441\u0451 \u043e\u043a, \u0430 payment_events<\/code> \u043e\u0431 \u044d\u0442\u043e\u043c \u0441\u043e\u0431\u044b\u0442\u0438\u0438 \u043d\u0438\u0447\u0435\u0433\u043e \u043d\u0435 \u0437\u043d\u0430\u0435\u0442. \u0422\u0430\u043a\u043e\u0433\u043e \u0431\u044b\u0442\u044c \u043d\u0435 \u0434\u043e\u043b\u0436\u043d\u043e. \u041b\u043e\u0433\u0438\u0440\u0443\u0435\u043c, \u0430\u043b\u0435\u0440\u0442\u0438\u043c, \u043d\u0435 \u043f\u0430\u043d\u0438\u043a\u0443\u0435\u043c.<\/p>\n
def process_deposit_sync(conn, event_id, event_type, user_id, amount):    amount_decimal  = _validate_amount(amount)    idempotency_key = _idempotency_key(event_id, event_type)    with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:        cur.execute(\"SET LOCAL lock_timeout = '2s'\")        cur.execute(\"SET LOCAL statement_timeout = '5s'\")        try:            cur.execute(                \"INSERT INTO processed_events (idempotency_key, outcome) \"                \"VALUES (%s, 'pending')\",                (idempotency_key,),            )        except psycopg2.errors.UniqueViolation:            conn.rollback()            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur2:                cur2.execute(\"SET LOCAL lock_timeout = '2s'\")                cur2.execute(\"SET LOCAL statement_timeout = '5s'\")                cur2.execute(                    \"SELECT outcome FROM processed_events WHERE idempotency_key = %s\",                    (idempotency_key,),                )                row = cur2.fetchone()                outcome = row[\"outcome\"] if row else \"pending\"                if outcome == \"success\":                    cur2.execute(                        \"SELECT status FROM payment_events WHERE event_id = %s\",                        (event_id,)                    )                    r = cur2.fetchone()                    current = r[\"status\"] if r else None                    if current is None:                        logger.error(                            \"deposit_replay: orphan event, processed_events \"                            \"exists but payment_event not found\",                            event_id=event_id,                        )                        send_alert(                            f\"[CRITICAL] orphan deposit event: {event_id}\",                            alert_key=f\"orphan_deposit:{event_id}\",                        )                    elif current == \"enqueued\":                        transition_event_status(cur2, event_id, \"enqueued\", \"confirmed\")                    elif current == \"processing\":                        transition_event_status(cur2, event_id, \"processing\", \"confirmed\")                    elif current == \"confirmed\":                        pass                    else:                        conn.rollback()                        raise RetryableError(                            f\"deposit_replay FSM violation: \"                            f\"payment_event.status={current!r} with outcome=success \"                            f\"for event_id={event_id}\"                        )                    conn.commit()                    return            conn.rollback()            raise RetryableError(                f\"deposit idempotency hit with outcome={outcome!r} for event_id={event_id}\"            )        except (psycopg2.errors.LockNotAvailable, psycopg2.errors.QueryCanceled,                psycopg2.errors.DeadlockDetected) as e:            conn.rollback()            raise RetryableError(f\"timeout on deposit for user {user_id}: {e}\")        try:            transition_event_status(cur, event_id, \"enqueued\", \"processing\")            cur.execute(\"SELECT id FROM users WHERE id = %s\", (user_id,))            if cur.fetchone() is None:                conn.rollback()                raise ValueError(f\"user {user_id} not found\")            cur.execute(                \"UPDATE users SET balance = balance + %s WHERE id = %s\",                (amount_decimal, user_id),            )            if cur.rowcount == 0:                conn.rollback()                raise ValueError(f\"user {user_id} disappeared between SELECT and UPDATE\")        except (psycopg2.errors.LockNotAvailable, psycopg2.errors.QueryCanceled,                psycopg2.errors.DeadlockDetected) as e:            conn.rollback()            raise RetryableError(f\"lock\/timeout on deposit first-time path for user {user_id}: {e}\")        try:            cur.execute(                \"INSERT INTO balance_events \"                \"(user_id, amount, event_type, source_event_id, created_at) \"                \"VALUES (%s, %s, %s, %s, NOW())\",                (user_id, amount_decimal, event_type, event_id),            )        except psycopg2.errors.UniqueViolation:            conn.rollback()            raise Exception(                f\"balance_events duplicate without idempotency key violation \"                f\"event_id={event_id}, investigate immediately\"            )        cur.execute(            \"UPDATE processed_events SET outcome = 'success' WHERE idempotency_key = %s\",            (idempotency_key,),        )        cur.execute(            \"UPDATE payment_events SET retry_count = 0 WHERE event_id = %s\",            (event_id,),        )        transition_event_status(cur, event_id, \"processing\", \"confirmed\")        conn.commit()WithdrawalOutcome = Literal[\"success\", \"insufficient_funds\"]def process_withdrawal_sync(conn, event_id, event_type, user_id, amount) -> WithdrawalOutcome:    amount_decimal  = _validate_amount(amount)    idempotency_key = _idempotency_key(event_id, event_type)    with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:        cur.execute(\"SET LOCAL lock_timeout = '2s'\")        cur.execute(\"SET LOCAL statement_timeout = '5s'\")        try:            cur.execute(                \"INSERT INTO processed_events (idempotency_key, outcome) \"                \"VALUES (%s, 'pending')\",                (idempotency_key,),            )        except psycopg2.errors.UniqueViolation:            conn.rollback()            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as lc:                lc.execute(\"SET LOCAL lock_timeout = '2s'\")                lc.execute(\"SET LOCAL statement_timeout = '5s'\")                lc.execute(                    \"SELECT outcome FROM processed_events WHERE idempotency_key = %s\",                    (idempotency_key,),                )                stored = (lc.fetchone() or {\"outcome\": \"pending\"})[\"outcome\"]                if stored == \"success\":                    lc.execute(                        \"SELECT status FROM payment_events WHERE event_id = %s\",                        (event_id,)                    )                    r = lc.fetchone()                    current = r[\"status\"] if r else None                    if current is None:                        logger.error(                            \"withdrawal_replay: orphan event, processed_events \"                            \"exists but payment_event not found\",                            event_id=event_id,                        )                        send_alert(                            f\"[CRITICAL] orphan withdrawal event: {event_id}\",                            alert_key=f\"orphan_withdrawal:{event_id}\",                        )                    elif current == \"enqueued\":                        transition_event_status(lc, event_id, \"enqueued\", \"confirmed\")                    elif current == \"processing\":                        transition_event_status(lc, event_id, \"processing\", \"confirmed\")                    elif current == \"confirmed\":                        pass                    else:                        conn.rollback()                        raise RetryableError(                            f\"withdrawal_replay FSM violation: \"                            f\"payment_event.status={current!r} with outcome=success \"                            f\"for event_id={event_id}\"                        )                    conn.commit()                    return \"success\"                elif stored == \"insufficient_funds\":                    lc.execute(                        \"SELECT status FROM payment_events WHERE event_id = %s\",                        (event_id,)                    )                    r = lc.fetchone()                    current = r[\"status\"] if r else None                    if current == \"enqueued\":                        transition_event_status(lc, event_id, \"enqueued\", \"failed\")                    elif current == \"processing\":                        transition_event_status(lc, event_id, \"processing\", \"failed\")                    elif current is None:                        logger.error(                            \"withdrawal_replay insufficient_funds: orphan event\",                            event_id=event_id,                        )                        send_alert(                            f\"[CRITICAL] orphan withdrawal (insufficient_funds): {event_id}\",                            alert_key=f\"orphan_withdrawal_insuf:{event_id}\",                        )                    conn.commit()                    return \"insufficient_funds\"                else:                    conn.rollback()                    raise RetryableError(                        f\"withdrawal idempotency hit with outcome={stored!r} \"                        f\"for event_id={event_id}\"                    )        except (psycopg2.errors.LockNotAvailable, psycopg2.errors.QueryCanceled,                psycopg2.errors.DeadlockDetected) as e:            conn.rollback()            raise RetryableError(f\"lock\/timeout on withdrawal INSERT for user {user_id}: {e}\")        try:            cur.execute(                \"SELECT balance FROM users WHERE id = %s FOR UPDATE NOWAIT\",                (user_id,),            )        except (psycopg2.errors.LockNotAvailable, psycopg2.errors.QueryCanceled,                psycopg2.errors.DeadlockDetected) as e:            conn.rollback()            raise RetryableError(f\"lock\/timeout\/deadlock on user lock for user {user_id}: {e}\")        row = cur.fetchone()        if row is None:            conn.rollback()            raise ValueError(f\"user {user_id} not found\")        try:            if row[\"balance\"] < amount_decimal:                cur.execute(                    \"UPDATE processed_events SET outcome = 'insufficient_funds' \"                    \"WHERE idempotency_key = %s\",                    (idempotency_key,),                )                transition_event_status(cur, event_id, \"enqueued\", \"failed\")                conn.commit()                return \"insufficient_funds\"            transition_event_status(cur, event_id, \"enqueued\", \"processing\")            cur.execute(                \"UPDATE users SET balance = balance - %s WHERE id = %s\",                (amount_decimal, user_id),            )            try:                cur.execute(                    \"INSERT INTO balance_events \"                    \"(user_id, amount, event_type, source_event_id, created_at) \"                    \"VALUES (%s, %s, %s, %s, NOW())\",                    (user_id, -amount_decimal, event_type, event_id),                )            except psycopg2.errors.UniqueViolation:                conn.rollback()                raise Exception(                    f\"balance_events duplicate without idempotency key violation \"                    f\"event_id={event_id}, investigate immediately\"                )            cur.execute(                \"UPDATE processed_events SET outcome = 'success' WHERE idempotency_key = %s\",                (idempotency_key,),            )            cur.execute(                \"UPDATE payment_events SET retry_count = 0 WHERE event_id = %s\",                (event_id,),            )            transition_event_status(cur, event_id, \"processing\", \"confirmed\")            conn.commit()            return \"success\"        except (psycopg2.errors.LockNotAvailable, psycopg2.errors.QueryCanceled,                psycopg2.errors.DeadlockDetected) as e:            conn.rollback()            raise RetryableError(f\"lock\/timeout on withdrawal path for user {user_id}: {e}\")<\/code>
<\/a><\/div><\/pre>\n
Webhook: outbox pattern \u0432\u043c\u0435\u0441\u0442\u043e \u043f\u0440\u044f\u043c\u043e\u0433\u043e .delay()  <\/h4>\n
\u041f\u0440\u044f\u043c\u043e\u0439 \u0432\u044b\u0437\u043e\u0432 .delay()<\/code> \u0438\u0437 webhook handler \u0441\u043e\u0437\u0434\u0430\u0451\u0442 \u043e\u043a\u043d\u043e \u043c\u0435\u0436\u0434\u0443 \u0437\u0430\u043f\u0438\u0441\u044c\u044e \u0432 \u0411\u0414 \u0438 \u043f\u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u043e\u0439 \u0432 \u043e\u0447\u0435\u0440\u0435\u0434\u044c. \u0415\u0441\u043b\u0438 \u043f\u0440\u043e\u0446\u0435\u0441\u0441 \u0443\u043f\u0430\u043b \u0432 \u044d\u0442\u043e\u0442 \u043c\u043e\u043c\u0435\u043d\u0442, \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0437\u0430\u0432\u0438\u0441\u043d\u0435\u0442 \u0432 pending<\/code> \u043d\u0430\u0432\u0441\u0435\u0433\u0434\u0430.<\/p>\n
\u0420\u0435\u0448\u0435\u043d\u0438\u0435: webhook \u0442\u043e\u043b\u044c\u043a\u043e \u043f\u0438\u0448\u0435\u0442 \u0432 \u0411\u0414. \u041e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 poller \u043a\u0430\u0436\u0434\u044b\u0435 5 \u0441\u0435\u043a\u0443\u043d\u0434 \u0431\u0435\u0440\u0451\u0442 pending \u0441\u043e\u0431\u044b\u0442\u0438\u044f, \u0441\u043d\u0430\u0447\u0430\u043b\u0430 \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e \u043c\u0435\u043d\u044f\u0435\u0442 \u0441\u0442\u0430\u0442\u0443\u0441\u044b \u0438 \u043a\u043e\u043c\u043c\u0438\u0442\u0438\u0442, \u0438 \u0442\u043e\u043b\u044c\u043a\u043e \u043f\u043e\u0441\u043b\u0435 \u044d\u0442\u043e\u0433\u043e \u0441\u0442\u0430\u0432\u0438\u0442 \u0432 Celery \u043e\u0447\u0435\u0440\u0435\u0434\u044c. \u0421\u043d\u0430\u0447\u0430\u043b\u0430 \u043a\u043e\u043c\u043c\u0438\u0442\u0438\u043c, \u043f\u043e\u0442\u043e\u043c \u043a\u043b\u0430\u0434\u0451\u043c \u0432 \u043e\u0447\u0435\u0440\u0435\u0434\u044c, \u0438\u043d\u0430\u0447\u0435 \u0432\u043e\u0440\u043a\u0435\u0440 \u0441\u0442\u0430\u0440\u0442\u0443\u0435\u0442 \u0440\u0430\u043d\u044c\u0448\u0435 \u0447\u0435\u043c \u0411\u0414 \u0437\u043d\u0430\u0435\u0442 \u043e\u0431 enqueued<\/code>.<\/p>\n
Alchemy \u0438 Infura \u0437\u043d\u0430\u044e\u0442 \u0442\u043e\u043b\u044c\u043a\u043e tx-hash \u0438 \u0430\u0434\u0440\u0435\u0441 \u043f\u043e\u043b\u0443\u0447\u0430\u0442\u0435\u043b\u044f. \u041c\u0430\u043f\u043f\u0438\u043d\u0433 to_address<\/code> \u0432 user_id<\/code> \u0434\u0435\u043b\u0430\u0435\u0442\u0441\u044f \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u043c \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u043c \u043a \u0442\u0430\u0431\u043b\u0438\u0446\u0435 deposit_addresses<\/code>. \u042d\u0442\u043e\u0442 \u0441\u043b\u043e\u0439 \u0432\u044b\u043d\u0435\u0441\u0435\u043d \u0438\u0437 \u0441\u0442\u0430\u0442\u044c\u0438, \u043d\u043e \u0431\u0435\u0437 \u043d\u0435\u0433\u043e \u0437\u043b\u043e\u0443\u043c\u044b\u0448\u043b\u0435\u043d\u043d\u0438\u043a \u0441 HMAC-\u043a\u043b\u044e\u0447\u043e\u043c \u0437\u0430\u0447\u0438\u0441\u043b\u0438\u0442 \u0434\u0435\u043d\u044c\u0433\u0438 \u043d\u0430 \u043b\u044e\u0431\u043e\u0439 user_id<\/code>. \u0421\u0442\u043e\u0438\u0442 \u0434\u0435\u0440\u0436\u0430\u0442\u044c \u044d\u0442\u043e \u0432 \u0433\u043e\u043b\u043e\u0432\u0435.<\/p>\n
verify_webhook_signature<\/code> \u043f\u0440\u0438\u043d\u0438\u043c\u0430\u0435\u0442 raw_body<\/code> \u0431\u0430\u0439\u0442\u0430\u043c\u0438 \u0434\u043e JSON-\u043f\u0430\u0440\u0441\u0438\u043d\u0433\u0430, \u043f\u043e\u0434\u043f\u0438\u0441\u044c \u0441\u0447\u0438\u0442\u0430\u0435\u0442\u0441\u044f \u043f\u043e \u0438\u0441\u0445\u043e\u0434\u043d\u044b\u043c \u0431\u0430\u0439\u0442\u0430\u043c. secrets.compare_digest<\/code> \u0437\u0430\u0449\u0438\u0449\u0430\u0435\u0442 \u043e\u0442 timing attack.<\/p>\n
import asyncpgfrom fastapi import FastAPI, Request, HTTPExceptionfrom slowapi import Limiter, _rate_limit_exceeded_handlerfrom slowapi.util import get_remote_addressfrom slowapi.errors import RateLimitExceededapp = FastAPI()limiter = Limiter(key_func=get_remote_address, storage_uri=settings.REDIS_URL)app.state.limiter = limiterapp.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)ALLOWED_EVENT_TYPES = frozenset({\"deposit\", \"airdrop\", \"withdrawal\", \"withdrawal_fee\"})def verify_webhook_signature(raw_body, signature_header, signing_key):    if not signing_key:        raise ImproperlyConfigured(\"WEBHOOK_SECRET is not set\")    if len(signing_key) < 32:        raise ImproperlyConfigured(            f\"WEBHOOK_SECRET too short: {len(signing_key)} chars, minimum 32\"        )    if not signature_header:        return False    mac = hmac.new(        key=signing_key.encode(\"utf-8\"),        msg=raw_body,        digestmod=hashlib.sha256,    )    return secrets.compare_digest(mac.hexdigest(), signature_header)@app.post(\"\/webhook\/payments\")@limiter.limit(\"300\/minute\")@limiter.limit(\"30\/second\")async def payment_webhook(request: Request):    raw_body  = await request.body()    signature = request.headers.get(\"X-Alchemy-Signature\", \"\")    if not verify_webhook_signature(raw_body, signature, settings.WEBHOOK_SECRET):        raise HTTPException(status_code=401, detail=\"invalid signature\")    trace_id = (        request.headers.get(\"X-Request-ID\")        or request.headers.get(\"X-Alchemy-Request-ID\")        or new_trace_id()    )    set_trace_id(trace_id)    try:        payload    = json.loads(raw_body)        event_id   = payload[\"event_id\"]        event_type = payload[\"event_type\"]        user_id    = payload[\"user_id\"]        if not isinstance(payload.get(\"amount\"), str):            raise HTTPException(status_code=400, detail=\"amount must be a JSON string, not a number\")        amount_str = payload[\"amount\"]    except (json.JSONDecodeError, KeyError) as e:        raise HTTPException(status_code=400, detail=f\"invalid payload: {e}\")    if event_type not in ALLOWED_EVENT_TYPES:        raise HTTPException(status_code=400, detail=f\"unknown event_type: {event_type!r}\")    try:        _validate_amount(amount_str)    except ValueError as e:        raise HTTPException(status_code=400, detail=f\"invalid amount: {e}\")    db = request.app.state.db    try:        async with db.transaction():            await db.fetchrow(                \"INSERT INTO payment_events (event_id, user_id, amount, event_type, status) \"                \"VALUES ($1, $2, $3, $4, 'pending') ON CONFLICT (event_id) DO NOTHING\",                event_id, user_id, amount_str, event_type,            )    except asyncpg.exceptions.ForeignKeyViolationError:        logger.warning(            \"orphan webhook event (user not found)\",            event_id=event_id, user_id=user_id,        )        raise HTTPException(status_code=400, detail=\"user not found\")    return {\"status\": \"accepted\", \"trace_id\": trace_id}<\/code>
<\/a><\/div><\/pre>\n
FastAPI \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442 \u0432 async event loop, \u0431\u043b\u043e\u043a\u0438\u0440\u0443\u044e\u0449\u0438\u0439 psycopg2 \u0442\u0430\u043c \u0443\u0431\u044c\u0451\u0442 throughput, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 asyncpg \u0432 webhook. Celery workers, \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0435 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u044b \u0431\u0435\u0437 event loop, asyncpg \u0442\u0430\u043c \u0442\u043e\u043b\u044c\u043a\u043e \u0434\u043e\u0431\u0430\u0432\u0438\u0442 \u0441\u043b\u043e\u0436\u043d\u043e\u0441\u0442\u044c.<\/p>\n
enqueue_pending_events<\/code> \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 \u0442\u043e\u0442 \u0436\u0435 \u043f\u0430\u0442\u0442\u0435\u0440\u043d SAVEPOINT<\/code> \u0447\u0442\u043e \u0438 recover_stale_enqueued_events<\/code>. \u0411\u0435\u0437 \u043d\u0435\u0433\u043e \u043e\u0434\u0438\u043d event \u0441 AlreadyProcessedError<\/code> \u043e\u0442 race \u0441 recover’\u043e\u043c \u043e\u0442\u043a\u0430\u0442\u044b\u0432\u0430\u043b \u0432\u0435\u0441\u044c \u0431\u0430\u0442\u0447 \u0438\u0437 100 \u0441\u043e\u0431\u044b\u0442\u0438\u0439. \u041e\u043d\u0438 \u043e\u0441\u0442\u0430\u0432\u0430\u043b\u0438\u0441\u044c \u0432 pending<\/code> \u0438 \u043f\u043e\u0434\u0445\u0432\u0430\u0442\u044b\u0432\u0430\u043b\u0438\u0441\u044c \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u043c \u0442\u0438\u043a\u043e\u043c, \u043d\u043e \u044d\u0442\u043e \u0432\u0438\u0434\u043d\u043e \u0432 \u043b\u043e\u0433\u0430\u0445 \u043a\u0430\u043a \u043f\u043e\u0442\u0435\u0440\u044f\u043d\u043d\u044b\u0439 \u0442\u0438\u043a. SAVEPOINT sp_enq<\/code> \u043d\u0430 \u043a\u0430\u0436\u0434\u044b\u0439 event \u0438\u0437\u043e\u043b\u0438\u0440\u0443\u0435\u0442 \u043e\u0448\u0438\u0431\u043a\u0443.<\/p>\n
@shared_task(name=\"enqueue_pending_events\")def enqueue_pending_events() -> dict:    conn = get_validated_conn(db_pool)    events_to_enqueue = []    try:        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:            cur.execute(\"\"\"                SELECT event_id, event_type, user_id, amount                FROM payment_events                WHERE status = 'pending'                  AND updated_at < NOW() - INTERVAL '5 seconds'                ORDER BY created_at                LIMIT 100                FOR UPDATE SKIP LOCKED            \"\"\")            events = cur.fetchall()            events_ok = []            for event in events:                try:                    cur.execute(\"SAVEPOINT sp_enq\")                    transition_event_status(cur, event['event_id'], \"pending\", \"enqueued\")                    cur.execute(\"RELEASE SAVEPOINT sp_enq\")                    events_ok.append(event)                except (ValueError, RetryableError, AlreadyProcessedError) as e:                    cur.execute(\"ROLLBACK TO SAVEPOINT sp_enq\")                    cur.execute(\"RELEASE SAVEPOINT sp_enq\")                    logger.warning(\"enqueue: skipped event\",                                   event_id=event['event_id'], error=str(e))            conn.commit()            events_to_enqueue = list(events_ok)    except Exception:        try:            conn.rollback()        except Exception:            pass        logger.exception(\"enqueue_pending_events: transition failed\")        raise    finally:        try:            conn.putconn()        except Exception:            pass    enqueued = 0    for event in events_to_enqueue:        try:            process_payment_event.apply_async(                args=[event['event_id'], event['event_type'],                      event['user_id'], str(event['amount'])],                kwargs={\"trace_id\": str(uuid.uuid4())},            )            enqueued += 1        except Exception:            logger.exception(\"apply_async failed\", event_id=event['event_id'])    logger.info(\"enqueue_pending_events done\", enqueued=enqueued)    return {\"enqueued\": enqueued}<\/code>
<\/a><\/div><\/pre>\n
recover_stale_enqueued_events<\/code> \u043a\u0430\u0436\u0434\u044b\u0435 2 \u043c\u0438\u043d\u0443\u0442\u044b \u043d\u0430\u0445\u043e\u0434\u0438\u0442 \u0441\u043e\u0431\u044b\u0442\u0438\u044f, \u0437\u0430\u0441\u0442\u0440\u044f\u0432\u0448\u0438\u0435 \u0432 enqueued<\/code>. \u041f\u043e\u0441\u043b\u0435 MAX_RECOVERY_ATTEMPTS<\/code> \u043f\u043e\u043f\u044b\u0442\u043e\u043a \u043f\u0435\u0440\u0435\u0432\u043e\u0434\u0438\u0442 \u0432 failed<\/code> + DLQ. SAVEPOINT<\/code> \u043d\u0430 \u043a\u0430\u0436\u0434\u043e\u0435 \u0441\u043e\u0431\u044b\u0442\u0438\u0435, \u043e\u0448\u0438\u0431\u043a\u0430 \u0432 \u043e\u0434\u043d\u043e\u043c \u043d\u0435 \u043e\u0442\u043a\u0430\u0442\u044b\u0432\u0430\u0435\u0442 \u0432\u0435\u0441\u044c \u0431\u0430\u0442\u0447.  <\/p>\n
MAX_RECOVERY_ATTEMPTS = 10@shared_task(name=\"recover_stale_enqueued_events\")def recover_stale_enqueued_events() -> dict:    conn      = get_validated_conn(db_pool)    recovered = 0    dlqed     = 0    try:        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:            cur.execute(\"\"\"                SELECT event_id, event_type, user_id, amount, retry_count                FROM payment_events                WHERE status = 'enqueued'                  AND updated_at < NOW() - INTERVAL '3 minutes'                ORDER BY created_at                LIMIT 50                FOR UPDATE SKIP LOCKED            \"\"\")            stale = cur.fetchall()            for event in stale:                try:                    cur.execute(\"SAVEPOINT sp_recover\")                    if event['retry_count'] >= MAX_RECOVERY_ATTEMPTS:                        transition_event_status(cur, event['event_id'], \"enqueued\", \"failed\")                        cur.execute(                            \"INSERT INTO dead_letter_queue \"                            \"(event_id, event_type, user_id, amount, error) \"                            \"VALUES (%s, %s, %s, %s, %s)\",                            (event['event_id'], event['event_type'], event['user_id'],                             str(event['amount']),                             f\"exhausted recovery attempts ({MAX_RECOVERY_ATTEMPTS})\")                        )                        dlqed += 1                    else:                        transition_event_status(cur, event['event_id'], \"enqueued\", \"pending\")                        cur.execute(                            \"UPDATE payment_events SET retry_count = retry_count + 1 \"                            \"WHERE event_id = %s\",                            (event['event_id'],)                        )                        recovered += 1                    cur.execute(\"RELEASE SAVEPOINT sp_recover\")                except Exception as sp_exc:                    try:                        cur.execute(\"ROLLBACK TO SAVEPOINT sp_recover\")                        cur.execute(\"RELEASE SAVEPOINT sp_recover\")                    except Exception:                        pass                    logger.error(\"recover: event skipped due to error\",                                 event_id=event['event_id'], error=str(sp_exc))            conn.commit()        if recovered or dlqed:            logger.warning(\"recover_stale_enqueued_events\",                          recovered=recovered, dlqed=dlqed)        if dlqed:            send_alert(                f\"[WARNING] {dlqed} events exhausted recovery attempts, check DLQ\",                alert_key=\"recovery_exhausted\",            )        return {\"recovered\": recovered, \"dlqed\": dlqed}    except Exception:        try:            conn.rollback()        except Exception:            pass        logger.exception(\"recover_stale_enqueued_events failed\")        raise    finally:        try:            conn.putconn()        except Exception:            pass<\/code>
<\/a><\/div><\/pre>\n
Celery: acks_late + reject_on_worker_lost + \u043f\u0443\u043b \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0439  <\/h4>\n
acks_late=True<\/code>: \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u0438\u0435 \u0431\u0440\u043e\u043a\u0435\u0440\u0443 \u043f\u043e\u0441\u043b\u0435 \u0437\u0430\u0432\u0435\u0440\u0448\u0435\u043d\u0438\u044f \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438, \u043d\u0435 \u0432 \u043c\u043e\u043c\u0435\u043d\u0442 \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f. \u041f\u043e \u0443\u043c\u043e\u043b\u0447\u0430\u043d\u0438\u044e Celery \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0430\u0435\u0442 \u0441\u0440\u0430\u0437\u0443: \u0432\u043e\u0440\u043a\u0435\u0440 \u043f\u0430\u0434\u0430\u0435\u0442 \u0432 \u0441\u0435\u0440\u0435\u0434\u0438\u043d\u0435, \u0437\u0430\u0434\u0430\u0447\u0430 \u043f\u043e\u0442\u0435\u0440\u044f\u043d\u0430.<\/p>\n
reject_on_worker_lost=True<\/code>: \u043f\u0440\u0438 SIGKILL\/OOM \u0437\u0430\u0434\u0430\u0447\u0430 \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0430\u0435\u0442\u0441\u044f \u0432 \u043e\u0447\u0435\u0440\u0435\u0434\u044c.<\/p>\n
\u0415\u0441\u043b\u0438 \u0441\u043e\u0437\u0434\u0430\u0442\u044c \u043f\u0443\u043b \u0434\u043e fork, \u0432\u0441\u0435 \u0432\u043e\u0440\u043a\u0435\u0440\u044b \u043d\u0430\u0441\u043b\u0435\u0434\u0443\u044e\u0442 \u043e\u0434\u043d\u0438 \u0438 \u0442\u0435 \u0436\u0435 \u0444\u0430\u0439\u043b\u043e\u0432\u044b\u0435 \u0434\u0435\u0441\u043a\u0440\u0438\u043f\u0442\u043e\u0440\u044b. \u0414\u0432\u0430 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0430 \u0448\u043b\u044e\u0442 \u0437\u0430\u043f\u0440\u043e\u0441\u044b \u0432 \u043e\u0434\u0438\u043d \u0441\u043e\u043a\u0435\u0442 \u0438 \u043e\u0442\u0432\u0435\u0442\u044b \u043f\u0435\u0440\u0435\u043c\u0435\u0448\u0438\u0432\u0430\u044e\u0442\u0441\u044f, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 \u043f\u0443\u043b \u0441\u043e\u0437\u0434\u0430\u0451\u0442\u0441\u044f \u0432 worker_process_init<\/code>.<\/p>\n
import osfrom celery.signals import worker_process_initdb_pool = None_local_breaker = None@worker_process_init.connectdef init_worker(**kwargs):    global db_pool, _local_breaker    _local_breaker = _InProcessBreaker()    worker_pool = os.environ.get(\"CELERY_WORKER_POOL\", \"prefork\")    is_threaded = worker_pool in (\"gevent\", \"eventlet\")    pool_class  = (        psycopg2.pool.ThreadedConnectionPool if is_threaded        else psycopg2.pool.SimpleConnectionPool    )    db_pool = pool_class(minconn=2, maxconn=10, dsn=settings.DATABASE_URL)    logger.info(\"worker init done\", pool=pool_class.__name__, worker_pool=worker_pool)def _get_local_breaker() -> \"_InProcessBreaker\":    global _local_breaker    if _local_breaker is None:        _local_breaker = _InProcessBreaker()    return _local_breakerBACKOFF_BASE_SEC = 1BACKOFF_CAP_SEC  = 60def jittered_backoff(attempt: int) -> float:    cap = min(BACKOFF_CAP_SEC, BACKOFF_BASE_SEC * (2 ** attempt))    return random.uniform(0, cap)<\/code>
<\/a><\/div><\/pre>\n
\u0414\u0432\u0443\u0445\u0443\u0440\u043e\u0432\u043d\u0435\u0432\u044b\u0439 DLQ  <\/h4>\n
\u0421\u043d\u0430\u0447\u0430\u043b\u0430 \u0434\u0435\u043b\u0430\u043b LPUSH<\/code> \u0438 EXPIRE<\/code> \u0434\u0432\u0443\u043c\u044f \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u043c\u0438 \u043a\u043e\u043c\u0430\u043d\u0434\u0430\u043c\u0438. \u041c\u0435\u0436\u0434\u0443 \u043d\u0438\u043c\u0438 \u0432\u043e\u0437\u043c\u043e\u0436\u0435\u043d crash, \u043a\u043b\u044e\u0447 \u043e\u0441\u0442\u0430\u0451\u0442\u0441\u044f \u0431\u0435\u0437 TTL, \u0436\u0438\u0432\u0451\u0442 \u0432\u0435\u0447\u043d\u043e. \u0424\u0438\u043a\u0441 \u0447\u0435\u0440\u0435\u0437 pipeline<\/code>.<\/p>\n
\u041f\u0440\u043e \u0441\u0445\u0435\u043c\u0443 DLQ: \u0442\u0430\u0431\u043b\u0438\u0446\u0430 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 BIGSERIAL PRIMARY KEY<\/code>, \u043d\u0435 event_id PRIMARY KEY<\/code>. \u042d\u0442\u043e \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0445\u0440\u0430\u043d\u0438\u0442\u044c \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043f\u043e\u043f\u044b\u0442\u043e\u043a \u0434\u043b\u044f \u043e\u0434\u043d\u043e\u0433\u043e event_id<\/code>. \u0421\u043b\u0435\u0434\u0441\u0442\u0432\u0438\u0435: ON CONFLICT (event_id) DO NOTHING<\/code> \u043d\u0435\u0434\u043e\u043f\u0443\u0441\u0442\u0438\u043c, event_id<\/code> \u043d\u0435 \u0438\u043c\u0435\u0435\u0442 UNIQUE<\/code> constraint. \u041a\u0430\u0436\u0434\u044b\u0439 INSERT<\/code> \u0441\u043e\u0437\u0434\u0430\u0451\u0442 \u043d\u043e\u0432\u0443\u044e \u0437\u0430\u043f\u0438\u0441\u044c \u0441 \u043f\u043e\u043b\u043d\u043e\u0439 \u0438\u0441\u0442\u043e\u0440\u0438\u0435\u0439 \u043f\u043e\u043f\u044b\u0442\u043a\u0438.<\/p>\n
DLQ_REDIS_KEY = \"dlq:payment_events\"DLQ_REDIS_TTL = 7 * 24 * 3600  # 7 \u0434\u043d\u0435\u0439def save_to_dlq_sync(conn, event_id, event_type, user_id, amount, error):    payload = {        \"event_id\": event_id, \"event_type\": event_type,        \"user_id\": user_id, \"amount\": str(amount),        \"error\": error, \"trace_id\": get_trace_id(),    }    db_ok = False    try:        with conn.cursor() as cur:            cur.execute(                \"INSERT INTO dead_letter_queue \"                \"(event_id, event_type, user_id, amount, error, created_at) \"                \"VALUES (%s, %s, %s, %s, %s, NOW())\",                (event_id, event_type, user_id, str(amount), error)            )            conn.commit()        db_ok = True    except Exception as db_exc:        logger.error(\"DLQ postgres write failed\", event_id=event_id, error=str(db_exc))        try:            conn.rollback()        except Exception:            pass    if db_ok:        return    try:        pipe = _redis_client.pipeline()        pipe.lpush(DLQ_REDIS_KEY, json.dumps(payload))        pipe.expire(DLQ_REDIS_KEY, DLQ_REDIS_TTL)        pipe.execute()        logger.warning(\"DLQ saved to Redis fallback\", event_id=event_id)        return    except redis_lib.RedisError as e:        logger.error(\"DLQ redis write failed\", event_id=event_id, error=str(e))    logger.critical(        \"DLQ_UNRECOVERABLE\",        event_id=event_id, event_type=event_type,        user_id=user_id, amount=str(amount),        error=error, trace_id=get_trace_id(),        dlq_payload=payload,    )<\/code>
<\/a><\/div><\/pre>\n
drain_redis_dlq<\/code> \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0435\u0442\u0441\u044f \u043f\u043e \u0440\u0430\u0441\u043f\u0438\u0441\u0430\u043d\u0438\u044e \u043f\u043e\u0441\u043b\u0435 \u0438\u043d\u0446\u0438\u0434\u0435\u043d\u0442\u0430 \u0441 \u0411\u0414. failed<\/code> \u0441\u0431\u0440\u0430\u0441\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0432 0 \u043f\u0440\u0438 \u043a\u0430\u0436\u0434\u043e\u043c \u0443\u0441\u043f\u0435\u0448\u043d\u043e\u043c INSERT, \u0441\u0447\u0451\u0442\u0447\u0438\u043a consecutive, \u043d\u0435 total. \u0427\u0435\u0440\u0435\u0434\u0443\u044e\u0449\u0438\u0435\u0441\u044f \u0443\u0441\u043f\u0435\u0445\u0438 \u0438 \u043e\u0448\u0438\u0431\u043a\u0438 \u043d\u0435 \u0442\u0440\u0438\u0433\u0433\u0435\u0440\u044f\u0442 \u0430\u0432\u0430\u0440\u0438\u0439\u043d\u044b\u0439 break, \u043d\u043e drained > 0<\/code> \u043f\u0440\u0438 \u044d\u0442\u043e\u043c.  <\/p>\n
DRAIN_BATCH_SIZE = 500@shared_task(name=\"drain_redis_dlq\")def drain_redis_dlq() -> dict:    drained = 0    failed  = 0    conn    = get_validated_conn(db_pool)    try:        for _ in range(DRAIN_BATCH_SIZE):            raw = _redis_client.rpop(DLQ_REDIS_KEY)            if raw is None:                break            try:                payload = json.loads(raw)            except json.JSONDecodeError:                logger.critical(                    \"drain_redis_dlq: malformed JSON in DLQ, item discarded\",                    raw=raw[:200],                )                continue            try:                with conn.cursor() as cur:                    cur.execute(                        \"INSERT INTO dead_letter_queue \"                        \"(event_id, event_type, user_id, amount, error, created_at) \"                        \"VALUES (%s, %s, %s, %s, %s, NOW())\",                        (payload['event_id'], payload['event_type'],                         payload['user_id'], payload['amount'], payload['error'])                    )                    conn.commit()                drained += 1                failed = 0            except Exception as e:                try:                    conn.rollback()                except Exception:                    pass                pipe = _redis_client.pipeline()                pipe.lpush(DLQ_REDIS_KEY, raw)                pipe.expire(DLQ_REDIS_KEY, DLQ_REDIS_TTL)                pipe.execute()                failed += 1                logger.error(                    \"drain failed, requeued to head\",                    event_id=payload.get('event_id'), error=str(e),                )                if failed >= 10:                    logger.error(\"drain aborted after 10 consecutive failures\")                    break    finally:        conn.putconn()    logger.info(\"drain_redis_dlq done\", drained=drained, failed=failed)    return {\"drained\": drained, \"failed\": failed}<\/code>
<\/a><\/div><\/pre>\n
Celery task: \u043f\u043e\u043b\u043d\u0430\u044f \u0432\u0435\u0440\u0441\u0438\u044f  <\/h4>\n
@shared_task(name=\"process_payment_event\", bind=True, max_retries=5, acks_late=True, reject_on_worker_lost=True)def process_payment_event(self, event_id, event_type, user_id, amount, trace_id=\"\"):    set_trace_id(trace_id or new_trace_id())    conn      = get_validated_conn(db_pool)    committed = False    conn_ok   = True    try:        if event_type in (\"deposit\", \"airdrop\"):            process_deposit_sync(conn, event_id, event_type, user_id, amount)            committed = True        elif event_type in (\"withdrawal\", \"withdrawal_fee\"):            outcome = process_withdrawal_sync(conn, event_id, event_type, user_id, amount)            committed = True            if outcome == \"insufficient_funds\":                notify_user_insufficient_funds(user_id)        else:            logger.error(\"unknown event_type\", event_type=event_type, event_id=event_id)            try:                conn.rollback()            except Exception:                pass            try:                _mark_event_failed(conn, event_id)            except Exception as mark_exc:                logger.error(\"_mark_event_failed raised\",                            event_id=event_id, error=str(mark_exc))            save_to_dlq_sync(                conn, event_id, event_type, user_id, amount,                f\"unknown event_type: {event_type!r}\"            )            raise Ignore()    except AlreadyProcessedError as exc:        logger.info(\"event already processed\", event_id=event_id, reason=str(exc))        raise Ignore()    except RetryableError as exc:        delay = jittered_backoff(self.request.retries)        logger.warning(            \"retrying\", event_id=event_id,            attempt=self.request.retries, delay=delay, reason=str(exc),        )        try:            raise self.retry(exc=exc, countdown=delay)        except MaxRetriesExceededError:            conn.rollback()            _mark_event_failed(conn, event_id)            save_to_dlq_sync(conn, event_id, event_type, user_id, amount,                             f\"retries exhausted: {exc}\")            raise Ignore()    except Ignore:        raise    except Exception as exc:        logger.exception(\"unhandled error\", event_id=event_id)        try:            conn.rollback()        except Exception:            pass        try:            _mark_event_failed(conn, event_id)        except Exception as mark_exc:            logger.error(\"_mark_event_failed raised in catch-all\",                        event_id=event_id, error=str(mark_exc))        try:            save_to_dlq_sync(conn, event_id, event_type, user_id, amount, str(exc))        except Exception as dlq_exc:            logger.critical(                \"DLQ write failed, manual recovery required\",                event_id=event_id, trace_id=get_trace_id(),                original_error=str(exc), dlq_error=str(dlq_exc),            )        self.update_state(state=\"FAILURE\", meta={\"error\": str(exc)})        raise Ignore()    finally:        if not committed:            try:                conn.rollback()            except Exception as rb_exc:                logger.error(\"rollback failed\", event_id=event_id, error=str(rb_exc))                conn_ok = False        conn.putconn(close=not conn_ok)def notify_user_insufficient_funds(user_id: int) -> None:    pass<\/code>
<\/a><\/div><\/pre>\n
notify_user_insufficient_funds<\/code>: \u0437\u0430\u0433\u043b\u0443\u0448\u043a\u0430. \u0412 \u043f\u0440\u043e\u0434\u0435 \u043d\u0443\u0436\u043d\u0430 outbox-\u0437\u0430\u043f\u0438\u0441\u044c \u0432\u043d\u0443\u0442\u0440\u0438 process_withdrawal_sync<\/code> \u0434\u043e \u0444\u0438\u043d\u0430\u043b\u044c\u043d\u043e\u0433\u043e commit. \u0412\u044b\u0437\u043e\u0432 \u043e\u0442\u0441\u044e\u0434\u0430 (\u043f\u043e\u0441\u043b\u0435 commit) \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442 out-of-band: \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u0430\u044f \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f, \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0435 \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u0438 \u0434\u043e\u0441\u0442\u0430\u0432\u043a\u0438.<\/p>\n
\u0417\u0434\u0435\u0441\u044c \u0441\u043a\u0440\u044b\u0442\u0430 \u043b\u043e\u0432\u0443\u0448\u043a\u0430 \u0441\u043f\u0435\u0446\u0438\u0444\u0438\u0447\u043d\u0430\u044f \u0438\u043c\u0435\u043d\u043d\u043e \u0434\u043b\u044f at-least-once. \u041f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u043f\u0440\u0438\u0441\u044b\u043b\u0430\u0435\u0442 \u043e\u0434\u043d\u043e \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0442\u0440\u0438\u0436\u0434\u044b  process_payment_event<\/code> \u043e\u0442\u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442 \u0442\u0440\u0438\u0436\u0434\u044b. \u0411\u0430\u043b\u0430\u043d\u0441 \u043d\u0435 \u0438\u0437\u043c\u0435\u043d\u0438\u0442\u0441\u044f (idempotency \u0447\u0435\u0440\u0435\u0437 unique constraint). \u041d\u043e notify<\/code> \u0432\u044b\u0437\u043e\u0432\u0435\u0442\u0441\u044f \u0442\u0440\u0438\u0436\u0434\u044b, \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u043f\u043e\u043b\u0443\u0447\u0438\u0442 \u0442\u0440\u0438 \u0443\u0432\u0435\u0434\u043e\u043c\u043b\u0435\u043d\u0438\u044f \u00ab\u043d\u0435\u0434\u043e\u0441\u0442\u0430\u0442\u043e\u0447\u043d\u043e \u0441\u0440\u0435\u0434\u0441\u0442\u0432\u00bb \u0432\u043c\u0435\u0441\u0442\u043e \u043e\u0434\u043d\u043e\u0433\u043e. \u0418\u0434\u0435\u043c\u043f\u043e\u0442\u0435\u043d\u0442\u043d\u043e\u0441\u0442\u044c \u0411\u0414 \u043d\u0435 \u0440\u0430\u0441\u043f\u0440\u043e\u0441\u0442\u0440\u0430\u043d\u044f\u0435\u0442\u0441\u044f \u043d\u0430 side-effects \u0432\u043d\u0435 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0438 \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438.<\/p>\n
\u0412\u0442\u043e\u0440\u0430\u044f \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0430 \u0441 \u044d\u0442\u0438\u043c placement: \u0435\u0441\u043b\u0438 notify<\/code> \u0431\u0440\u043e\u0441\u0438\u0442 \u0438\u0441\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u0435, \u043e\u043d \u043f\u043e\u043f\u0430\u0434\u0451\u0442 \u0432 catch-all except Exception<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0437\u0430\u043f\u0438\u0448\u0435\u0442 \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0432 DLQ, \u0445\u043e\u0442\u044f \u0434\u0435\u043d\u044c\u0433\u0438 \u0443\u0436\u0435 \u0441\u043f\u0438\u0441\u0430\u043d\u044b \u043a\u043e\u0440\u0440\u0435\u043a\u0442\u043d\u043e \u0438 business-\u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u0437\u0430\u043a\u043e\u043c\u043c\u0438\u0447\u0435\u043d\u0430. \u042d\u0442\u043e \u0448\u0443\u043c \u0432 DLQ \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0431\u0443\u0434\u0435\u0442 \u0441\u043a\u0440\u044b\u0432\u0430\u0442\u044c \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u0435 \u0438\u043d\u0446\u0438\u0434\u0435\u043d\u0442\u044b.<\/p>\n
Circuit breaker \u0434\u043b\u044f Web3 RPC  <\/h4>\n
\u0422\u0440\u0451\u0445\u0444\u0430\u0437\u043d\u044b\u0439:  closed \u2192 open \u2192 half-open \u2192 closed. Redis \u043a\u0430\u043a shared state \u043c\u0435\u0436\u0434\u0443 \u0438\u043d\u0441\u0442\u0430\u043d\u0441\u0430\u043c\u0438, in-process breaker \u043a\u0430\u043a fallback \u043a\u043e\u0433\u0434\u0430 Redis \u043d\u0435\u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d.<\/p>\n
_InProcessBreaker<\/code>: \u043f\u0440\u043e\u0441\u0442\u043e\u0439 per-process \u0441\u0447\u0451\u0442\u0447\u0438\u043a \u0441 lock. \u041e\u0442\u043a\u0440\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u043f\u0440\u0438RPC_ERROR_THRESHOLD<\/code> \u043e\u0448\u0438\u0431\u043a\u0430\u0445, \u0437\u0430\u043a\u0440\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u043f\u043e\u0441\u043b\u0435 RPC_COOLDOWN_SEC<\/code> \u0441\u0435\u043a\u0443\u043d\u0434. \u041d\u0443\u0436\u0435\u043d \u0438\u043c\u0435\u043d\u043d\u043e \u043a\u0430\u043a fallback: \u0435\u0441\u043b\u0438 Redis \u0441\u0430\u043c \u043d\u0435\u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d, circuit breaker \u043d\u0435 \u0434\u043e\u043b\u0436\u0435\u043d \u043f\u0435\u0440\u0435\u0441\u0442\u0430\u0442\u044c \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c.<\/p>\n
\u041d\u0443\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0442\u043e\u043b\u044c\u043a\u043e \u043e\u0434\u0438\u043d \u0442\u0435\u0441\u0442\u043e\u0432\u044b\u0439 \u0437\u0430\u043f\u0440\u043e\u0441 \u043f\u043e\u043a\u0430 breaker \u043e\u0442\u043a\u0440\u044b\u0442, \u0438 \u0442\u043e\u0433\u0434\u0430 SET nx=True<\/code> \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u0440\u0443\u0435\u0442, \u0447\u0442\u043e \u0442\u043e\u043b\u044c\u043a\u043e \u043f\u0435\u0440\u0432\u044b\u0439 \u0432\u043e\u0440\u043a\u0435\u0440 \u043f\u043e\u043b\u0443\u0447\u0438\u0442 \u043f\u0440\u0430\u0432\u043e, \u0430 \u043e\u0441\u0442\u0430\u043b\u044c\u043d\u044b\u0435 \u0443\u0432\u0438\u0434\u044f\u0442 \u0437\u0430\u043d\u044f\u0442\u044b\u0439 probe key.<\/p>\n
import web3from prometheus_client import Counterrpc_errors_total = Counter(\"web3_rpc_errors_total\", \"Web3 RPC failures\", [\"method\"])_w3 = web3.Web3(web3.HTTPProvider(settings.ETH_RPC_URL))FINALIZED_CACHE_TTL   = 86_400PENDING_CACHE_TTL     = 30CACHE_PREFIX          = \"eth:fin:\"RPC_ERROR_THRESHOLD   = 5RPC_COOLDOWN_SEC      = 60RPC_ERROR_WINDOW_SEC  = 30HALF_OPEN_PROBE_KEY   = \"circuit:web3:half_open_probe\"HALF_OPEN_PROBE_TTL   = 10@dataclassclass _InProcessBreaker:    _lock:       threading.Lock = field(default_factory=threading.Lock)    _errors:     int            = 0    _open_until: \"datetime | None\" = None    def is_open(self) -> bool:        with self._lock:            if self._open_until is None:                return False            if datetime.now(timezone.utc) > self._open_until:                self._open_until = None                self._errors = 0                return False            return True    def record_error(self) -> None:        with self._lock:            self._errors += 1            if self._errors >= RPC_ERROR_THRESHOLD:                self._open_until = datetime.now(timezone.utc) + timedelta(seconds=RPC_COOLDOWN_SEC)    def record_success(self) -> None:        with self._lock:            self._errors     = 0            self._open_until = Nonedef _is_circuit_open() -> bool:    try:        if _redis_client.get(\"circuit:web3:open\"):            is_probe = _redis_client.set(                HALF_OPEN_PROBE_KEY, \"1\", nx=True, ex=HALF_OPEN_PROBE_TTL            )            if is_probe:                return False            return True    except redis_lib.RedisError:        pass    return _get_local_breaker().is_open()def _record_rpc_error(method: str) -> None:    rpc_errors_total.labels(method=method).inc()    _get_local_breaker().record_error()    try:        pipe  = _redis_client.pipeline()        pipe.incr(\"circuit:web3:errors\")        pipe.expire(\"circuit:web3:errors\", RPC_ERROR_WINDOW_SEC)        count, _ = pipe.execute()        if count >= RPC_ERROR_THRESHOLD:            open_pipe = _redis_client.pipeline()            open_pipe.setex(\"circuit:web3:open\", RPC_COOLDOWN_SEC, \"1\")            open_pipe.set(HALF_OPEN_PROBE_KEY, \"1\", ex=HALF_OPEN_PROBE_TTL)            open_pipe.execute()            logger.critical(\"web3 circuit breaker opened\", count=count)            send_alert(                f\"[CRITICAL] Web3 RPC circuit breaker OPEN, \"                f\"{count} failures in {RPC_ERROR_WINDOW_SEC}s.\",                alert_key=\"web3_breaker_open\",            )        else:            if not _redis_client.exists(\"circuit:web3:open\"):                _redis_client.delete(HALF_OPEN_PROBE_KEY)    except redis_lib.RedisError as e:        logger.warning(\"circuit breaker state write failed\", error=str(e))def _record_rpc_success() -> None:    _get_local_breaker().record_success()    try:        _redis_client.delete(\"circuit:web3:errors\")        _redis_client.delete(HALF_OPEN_PROBE_KEY)        if _redis_client.delete(\"circuit:web3:open\"):            logger.info(\"web3 circuit breaker closed\")            send_alert(                \"[INFO] Web3 RPC circuit breaker CLOSED, recovered\",                alert_key=\"web3_breaker_closed\",            )    except redis_lib.RedisError:        passdef is_transaction_finalized(tx_hash: str) -> bool:    if _is_circuit_open():        logger.warning(\"web3 circuit open, skipping\", tx_hash=tx_hash)        return False    cache_key = f\"{CACHE_PREFIX}{tx_hash}\"    try:        cached = _redis_client.get(cache_key)        if cached == \"1\":            return True        if cached == \"0\":            return False    except redis_lib.RedisError as e:        logger.warning(\"redis cache unavailable\", tx_hash=tx_hash, error=str(e))    method = \"get_transaction_receipt\"    try:        receipt = _w3.eth.get_transaction_receipt(tx_hash)        if receipt is None:            return False        method = \"get_block_finalized\"        finalized_block = _w3.eth.get_block(\"finalized\")[\"number\"]        result = receipt[\"blockNumber\"] <= finalized_block        _record_rpc_success()    except Exception as e:        logger.error(\"eth rpc error\", tx_hash=tx_hash, error=str(e))        _record_rpc_error(method)        return False    try:        ttl = FINALIZED_CACHE_TTL if result else PENDING_CACHE_TTL        _redis_client.setex(cache_key, ttl, \"1\" if result else \"0\")    except redis_lib.RedisError as e:        logger.warning(\"redis cache write failed\", tx_hash=tx_hash, error=str(e))    return result<\/code>
<\/a><\/div><\/pre>\n
\u041c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433  <\/h4>\n
hot_path_balance_check<\/code>: \u043d\u0435 \u043f\u043e\u043b\u043d\u043e\u0446\u0435\u043d\u043d\u0430\u044f \u0444\u0438\u043d\u0430\u043d\u0441\u043e\u0432\u0430\u044f \u0441\u0432\u0435\u0440\u043a\u0430, \u0430 \u0442\u043e\u043b\u044c\u043a\u043e \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433 \u0433\u043e\u0440\u044f\u0447\u0435\u0433\u043e \u043f\u0443\u0442\u0438 \u0437\u0430 \u043f\u043e\u0441\u043b\u0435\u0434\u043d\u0438\u0435 10 \u043c\u0438\u043d\u0443\u0442. users.balance<\/code> \u0438 SUM(balance_events)<\/code> \u0447\u0438\u0442\u0430\u044e\u0442\u0441\u044f \u043e\u0434\u043d\u0438\u043c JOIN-\u0437\u0430\u043f\u0440\u043e\u0441\u043e\u043c, \u0441\u043d\u0438\u043c\u043e\u043a \u0434\u0430\u043d\u043d\u044b\u0445 \u0431\u0435\u0440\u0451\u0442\u0441\u044f \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e. REPEATABLE_READ<\/code> \u0437\u0434\u0435\u0441\u044c defensive engineering: \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u0440\u0443\u0435\u0442 \u043a\u043e\u043d\u0441\u0438\u0441\u0442\u0435\u043d\u0442\u043d\u044b\u0439 \u0441\u043d\u0438\u043c\u043e\u043a \u043d\u0430 \u0443\u0440\u043e\u0432\u043d\u0435 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0438 \u0438 \u0437\u0430\u0449\u0438\u0449\u0430\u0435\u0442 \u043e\u0442 phantom reads \u0435\u0441\u043b\u0438 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u0432\u044b\u0440\u0430\u0441\u0442\u0435\u0442 \u0434\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u0438\u0445 \u043e\u043f\u0435\u0440\u0430\u0442\u043e\u0440\u043e\u0432 \u0432 \u0431\u0443\u0434\u0443\u0449\u0435\u043c. \u0414\u043b\u044f \u043f\u043e\u043b\u043d\u043e\u0439 \u0438\u0441\u0442\u043e\u0440\u0438\u0447\u0435\u0441\u043a\u043e\u0439 \u0441\u0432\u0435\u0440\u043a\u0438 \u043d\u0443\u0436\u043d\u0430 \u043d\u043e\u0447\u043d\u0430\u044f \u0437\u0430\u0434\u0430\u0447\u0430, \u044d\u0442\u043e \u0432 \u0431\u044d\u043a\u043b\u043e\u0433\u0435.<\/p>\n
set_isolation_level<\/code> \u043c\u043e\u0436\u0435\u0442 \u0431\u0440\u043e\u0441\u0438\u0442\u044c \u0435\u0441\u043b\u0438 \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0440\u0430\u0437\u043e\u0440\u0432\u0430\u043d\u043e. \u0411\u0435\u0437 try\/except<\/code> \u0432\u043e\u043a\u0440\u0443\u0433 \u043d\u0435\u0433\u043e \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0432\u0435\u0440\u043d\u0451\u0442\u0441\u044f \u0432 \u043f\u0443\u043b \u0441 REPEATABLE_READ<\/code>, \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0439 \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u043d\u0435 \u0431\u0443\u0434\u0435\u0442 \u043e\u0436\u0438\u0434\u0430\u0442\u044c \u0442\u0430\u043a\u043e\u0433\u043e \u043f\u043e\u0432\u0435\u0434\u0435\u043d\u0438\u044f.<\/p>\n
from prometheus_client import Counterhot_path_runs = Counter(\"hot_path_balance_check_runs_total\", \"Runs of hot path balance check\")@shared_task(name=\"hot_path_balance_check\")def hot_path_balance_check() -> None:    conn    = get_validated_conn(db_pool)    conn_ok = True    try:        conn.set_isolation_level(            psycopg2.extensions.ISOLATION_LEVEL_REPEATABLE_READ        )        try:            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:                cur.execute(\"SET LOCAL statement_timeout = '10s'\")                cur.execute(\"\"\"                    SELECT                        u.id,                        u.balance                                        AS actual_balance,                        u.initial_balance + COALESCE(SUM(be.amount), 0) AS calculated_balance                    FROM users u                    INNER JOIN (                        SELECT DISTINCT user_id FROM balance_events                        WHERE created_at > NOW() - INTERVAL '10 minutes'                    ) recent ON recent.user_id = u.id                    LEFT JOIN balance_events be ON be.user_id = u.id                    GROUP BY u.id, u.balance, u.initial_balance                    HAVING u.balance != u.initial_balance + COALESCE(SUM(be.amount), 0)                \"\"\")                mismatches = cur.fetchall()            conn.commit()        except Exception:            conn.rollback()            raise        finally:            try:                conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_DEFAULT)            except Exception:                conn_ok = False        hot_path_runs.inc()        if mismatches:            send_alert(                f\"[CRITICAL] balance mismatch: {[dict(m) for m in mismatches]}\",                alert_key=\"balance_mismatch\",            )    except Exception:        logger.exception(\"hot_path_balance_check failed\")        try:            conn.rollback()        except Exception:            pass        send_alert(            \"[WARNING] hot_path_balance_check failed, check worker logs\",            alert_key=\"balance_check_failed\",        )        raise    finally:        conn.putconn(close=not conn_ok)@shared_task(name=\"alert_zombie_events\")def alert_zombie_events() -> None:    conn = get_validated_conn(db_pool)    try:        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:            cur.execute(\"\"\"                SELECT event_id, user_id, status, updated_at FROM payment_events                WHERE (status = 'processing' AND updated_at < NOW() - INTERVAL '5 minutes')                   OR (status IN ('pending', 'enqueued') AND updated_at < NOW() - INTERVAL '15 minutes')            \"\"\")            zombies = cur.fetchall()            cur.execute(\"\"\"                SELECT COUNT(*) AS dlq_size FROM dead_letter_queue                WHERE created_at > NOW() - INTERVAL '1 hour'            \"\"\")            recent_dlq = cur.fetchone()['dlq_size']            cur.execute(\"\"\"                SELECT COUNT(*) AS stuck FROM processed_events                WHERE outcome = 'pending' AND created_at < NOW() - INTERVAL '10 minutes'            \"\"\")            stuck_pending = cur.fetchone()['stuck']        if zombies:            send_alert(                f\"[WARNING] zombie events: {[z['event_id'] for z in zombies[:20]]}\",                alert_key=\"zombie_events\",            )        if recent_dlq > 10:            send_alert(                f\"[WARNING] {recent_dlq} events in DLQ last hour, investigate\",                alert_key=\"dlq_flood\",            )        if stuck_pending:            send_alert(                f\"[CRITICAL] {stuck_pending} stuck 'pending' rows in processed_events, \"                f\"architectural invariant broken, investigate urgently\",                alert_key=\"processed_events_stuck_pending\",            )    finally:        try:            conn.rollback()        except Exception:            pass        conn.putconn()<\/code>
<\/a><\/div><\/pre>\n
Prometheus-\u0430\u043b\u0435\u0440\u0442\u044b:  <\/p>\n
- alert: HotPathBalanceCheckNotRunning  expr: increase(hot_path_balance_check_runs_total[6m]) == 0  for: 0m- alert: Web3RPCCircuitOpen  expr: increase(web3_rpc_errors_total[5m]) > 5  for: 0m- alert: CeleryHighRetryRate  expr: rate(celery_tasks_total{state=\"retry\"}[5m])      \/ rate(celery_tasks_total{state=\"success\"}[5m]) > 0.1- alert: CeleryQueueDepth  expr: celery_queue_depth > 500  for: 5m<\/code>
<\/a><\/div><\/pre>\n
Beat \u0440\u0430\u0441\u043f\u0438\u0441\u0430\u043d\u0438\u0435:  <\/p>\n
from celery.schedules import crontabbeat_schedule = {    \"enqueue-pending-events\":        {\"task\": \"enqueue_pending_events\",        \"schedule\": 5.0},    \"recover-stale-enqueued-events\": {\"task\": \"recover_stale_enqueued_events\", \"schedule\": 120.0},    \"cleanup-processed-events\":      {\"task\": \"cleanup_processed_events\",      \"schedule\": crontab(hour=3, minute=0)},    \"hot-path-balance-check\":        {\"task\": \"hot_path_balance_check\",        \"schedule\": 60.0},    \"alert-zombie-events\":           {\"task\": \"alert_zombie_events\",           \"schedule\": 60.0},    \"drain-redis-dlq\":               {\"task\": \"drain_redis_dlq\",               \"schedule\": crontab(hour=4, minute=0)},}<\/code>
<\/a><\/div><\/pre>\n
Cleanup  <\/h4>\n
processed_events<\/code> \u0447\u0438\u0441\u0442\u0438\u0442 \u0442\u043e\u043b\u044c\u043a\u043e terminal outcomes. outcome='pending'<\/code> \u043d\u0435 \u0442\u0440\u043e\u0433\u0430\u0435\u0442\u0441\u044f \u043d\u0438\u043a\u043e\u0433\u0434\u0430. \u042d\u0442\u043e \u0437\u0430\u0449\u0438\u0442\u0430 \u043e\u0442 \u0442\u0438\u0445\u043e\u0433\u043e \u0443\u0434\u0430\u043b\u0435\u043d\u0438\u044f \u0441\u043e\u0431\u044b\u0442\u0438\u0439 \u0432 \u0441\u0435\u0440\u0435\u0434\u0438\u043d\u0435 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438. \u0421\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0431\u0435\u0440\u0451\u0442\u0441\u044f \u0438 \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0430\u0435\u0442\u0441\u044f \u0432 \u043f\u0443\u043b \u043d\u0430 \u043a\u0430\u0436\u0434\u044b\u0439 \u0431\u0430\u0442\u0447, \u0430 \u043d\u0435 \u0443\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0435\u0442\u0441\u044f \u043d\u0430 \u0432\u0441\u0451 \u0432\u0440\u0435\u043c\u044f \u0437\u0430\u0434\u0430\u0447\u0438.<\/p>\n
TTL 14 \u0434\u043d\u0435\u0439 \u0431\u0435\u0437\u043e\u043f\u0430\u0441\u0435\u043d \u0431\u043b\u0430\u0433\u043e\u0434\u0430\u0440\u044f UNIQUE (source_event_id, event_type)<\/code> \u043d\u0430 balance_events<\/code>: \u0434\u0430\u0436\u0435 \u0435\u0441\u043b\u0438 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u0440\u0435\u043f\u043b\u0435\u0438\u0442 \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0441\u043f\u0443\u0441\u0442\u044f 14+ \u0434\u043d\u0435\u0439 \u0438 processed_events<\/code> \u0443\u0436\u0435 \u043e\u0447\u0438\u0449\u0435\u043d, \u043f\u043e\u0432\u0442\u043e\u0440\u043d\u044b\u0439 INSERT \u0432 balance_events<\/code> \u043e\u0442\u043a\u043b\u043e\u043d\u044f\u0435\u0442\u0441\u044f unique constraint. \u0414\u0430\u0436\u0435 \u0435\u0441\u043b\u0438 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0435 \u043e\u0431\u043b\u0430\u0436\u0430\u0435\u0442\u0441\u044f, unique constraint \u043d\u0430 balance_events<\/code> \u043d\u0435 \u043f\u0440\u043e\u043f\u0443\u0441\u0442\u0438\u0442 \u0434\u0443\u0431\u043b\u044c. \u0411\u0430\u0437\u0430 \u0431\u0443\u0434\u0435\u0442 \u043f\u043e\u0441\u043b\u0435\u0434\u043d\u0438\u043c \u0440\u0443\u0431\u0435\u0436\u043e\u043c.<\/p>\n
CLEANUP_BATCH_SIZE    = 5_000CLEANUP_BATCH_PAUSE   = 0.1CLEANUP_MAX_BATCHES   = 200CLEANUP_TTL_DAYS      = 14CLEANUP_SAFE_STATUSES = (\"success\", \"insufficient_funds\")@shared_task(name=\"cleanup_processed_events\")def cleanup_processed_events() -> dict:    import time    total_deleted = 0    batches = 0    for _ in range(CLEANUP_MAX_BATCHES):        conn = get_validated_conn(db_pool)        try:            with conn.cursor() as cur:                cur.execute(\"\"\"                    DELETE FROM processed_events                    WHERE idempotency_key IN (                        SELECT idempotency_key FROM processed_events                        WHERE outcome = ANY(%s)                          AND created_at < NOW() - (%s * INTERVAL '1 day')                        LIMIT %s                        FOR UPDATE SKIP LOCKED                    )                \"\"\", (list(CLEANUP_SAFE_STATUSES), CLEANUP_TTL_DAYS, CLEANUP_BATCH_SIZE))                deleted = cur.rowcount                conn.commit()        except Exception:            try:                conn.rollback()            except Exception:                pass            raise        finally:            conn.putconn()        total_deleted += deleted        batches += 1        if deleted < CLEANUP_BATCH_SIZE:            break        time.sleep(CLEANUP_BATCH_PAUSE)    return {\"batches\": batches, \"deleted\": total_deleted}<\/code>
<\/a><\/div><\/pre>\n
\u0422\u0435\u0441\u0442\u044b \u0438 \u0447\u0442\u043e \u043e\u043d\u0438 \u043d\u0435 \u043f\u043e\u043a\u0440\u044b\u0432\u0430\u044e\u0442  <\/h2>\n
Idempotency \u043d\u0430 \u043c\u043e\u043a\u0430\u0445 \u043d\u0435 \u043f\u0440\u043e\u0432\u0435\u0440\u0438\u0448\u044c. Unique constraint \u0434\u043e\u043b\u0436\u0435\u043d \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c \u043a\u0430\u043a \u0432 \u043f\u0440\u043e\u0434\u0435, \u0437\u043d\u0430\u0447\u0438\u0442 \u043d\u0443\u0436\u043d\u0430 \u0440\u0435\u0430\u043b\u044c\u043d\u0430\u044f \u0411\u0414. threading<\/code> \u043d\u0435 \u043f\u043e\u0434\u0445\u043e\u0434\u0438\u0442, GIL \u043c\u0435\u0448\u0430\u0435\u0442 \u043d\u043e\u0440\u043c\u0430\u043b\u044c\u043d\u043e \u0432\u043e\u0441\u043f\u0440\u043e\u0438\u0437\u0432\u0435\u0441\u0442\u0438 \u0433\u043e\u043d\u043a\u0443. \u041f\u043e\u044d\u0442\u043e\u043c\u0443 multiprocessing<\/code>.<\/p>\n
\u0411\u0430\u043b\u0430\u043d\u0441 \u043c\u043e\u0436\u0435\u0442 \u0441\u043e\u0439\u0442\u0438\u0441\u044c, \u0430 balance_events<\/code> \u043f\u0440\u0438 \u044d\u0442\u043e\u043c \u0437\u0430\u0434\u0443\u0431\u043b\u0438\u0440\u043e\u0432\u0430\u043d. \u041f\u0440\u043e\u0432\u0435\u0440\u044f\u0439 \u043e\u0431\u0430. \u0418\u043c\u0435\u043d\u043d\u043e \u0442\u0430\u043a \u0438 \u0431\u044b\u0432\u0430\u0435\u0442, \u0431\u0430\u043b\u0430\u043d\u0441 \u0441\u0445\u043e\u0434\u0438\u0442\u0441\u044f, \u0432\u0441\u0451 \u0432\u044b\u0433\u043b\u044f\u0434\u0438\u0442 \u0447\u0438\u0441\u0442\u043e, \u0430 \u0434\u0443\u0431\u043b\u0438 \u0432 balance_events<\/code> \u0432\u0441\u043f\u043b\u044b\u0432\u0430\u044e\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u043a\u043e\u0433\u0434\u0430 \u043f\u0440\u0438\u0445\u043e\u0434\u0438\u0442 \u0430\u0443\u0434\u0438\u0442\u043e\u0440.<\/p>\n
import pytestimport psycopg2import psycopg2.extrasimport multiprocessingfrom decimal import DecimalTEST_DSN = \"host=localhost port=5433 dbname=testdb user=testuser password=testuser\"@pytest.fixture(scope=\"session\")def db_conn_session():    conn = psycopg2.connect(TEST_DSN)    yield conn    conn.close()@pytest.fixturedef db_conn(db_conn_session):    conn = db_conn_session    conn.rollback()    with conn.cursor() as cur:        cur.execute(\"\"\"            TRUNCATE balance_events, processed_events, payment_events,                     dead_letter_queue, users            RESTART IDENTITY CASCADE        \"\"\")        cur.execute(            \"INSERT INTO users (id, balance, initial_balance) VALUES (1, 100.0, 100.0)\"        )    conn.commit()    yield conn    try:        conn.rollback()    except Exception:        passdef _deposit_worker(dsn, event_id, event_type, user_id, amount, barrier, q):    conn = psycopg2.connect(dsn)    try:        barrier.wait(timeout=10)        process_deposit_sync(conn, event_id, event_type, user_id, amount)        q.put((\"ok\", None))    except Exception as e:        q.put((\"err\", f\"{type(e).__name__}: {e}\"))    finally:        conn.close()def test_duplicate_deposits_produce_single_credit(db_conn):    \"\"\"10 \u0432\u043e\u0440\u043a\u0435\u0440\u043e\u0432, \u041e\u0414\u0418\u041d event_id, \u0438\u043c\u0438\u0442\u0430\u0446\u0438\u044f at-least-once delivery.\"\"\"    with db_conn.cursor() as cur:        cur.execute(            \"INSERT INTO payment_events (event_id, user_id, amount, event_type, status) \"            \"VALUES ('evt_dup', 1, '50.0', 'deposit', 'enqueued')\"        )    db_conn.commit()    N = 10    barrier = multiprocessing.Barrier(N)    q = multiprocessing.Queue()    workers = [        multiprocessing.Process(            target=_deposit_worker,            args=(TEST_DSN, \"evt_dup\", \"deposit\", 1, \"50.0\", barrier, q),        ) for _ in range(N)    ]    for w in workers: w.start()    for w in workers: w.join(timeout=20)    with db_conn.cursor() as cur:        cur.execute(\"SELECT balance FROM users WHERE id = 1\")        balance = cur.fetchone()[0]        cur.execute(\"SELECT COUNT(*) FROM balance_events\")        be_count = cur.fetchone()[0]        cur.execute(\"SELECT COUNT(*) FROM processed_events\")        pe_count = cur.fetchone()[0]    assert balance == Decimal(\"150.0\"), f\"\u0434\u0443\u0431\u043b\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435! balance={balance}\"    assert be_count == 1, f\"balance_events \u0434\u0443\u0431\u043b\u0438\u0440\u043e\u0432\u0430\u043d: {be_count}\"    assert pe_count == 1<\/code>
<\/a><\/div><\/pre>\n
\u0427\u0442\u043e \u0442\u0435\u0441\u0442\u044b \u043d\u0435 \u043f\u043e\u043a\u0440\u044b\u0432\u0430\u044e\u0442<\/h2>\n
Throughput \u043f\u043e\u0434 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u0439 \u043d\u0430\u0433\u0440\u0443\u0437\u043a\u043e\u0439. NOWAIT<\/code> \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0443\u0435\u0442 \u0434\u043e\u0441\u0442\u0443\u043f, throughput \u0434\u043e\u0441\u0442\u0438\u0433\u0430\u0435\u0442\u0441\u044f \u0447\u0435\u0440\u0435\u0437 Celery retry \u0441 jittered backoff. \u0414\u043b\u044f \u043d\u0430\u0433\u0440\u0443\u0437\u043e\u0447\u043d\u043e\u0433\u043e \u0442\u0435\u0441\u0442\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f \u043d\u0443\u0436\u0435\u043d \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 \u0441\u0442\u0435\u043d\u0434 \u0441 Celery workers.<\/p>\n
\u041f\u0430\u0434\u0435\u043d\u0438\u0435 \u0432\u043e\u0440\u043a\u0435\u0440\u0430 \u043f\u043e\u0441\u0440\u0435\u0434\u0438 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0438. reject_on_worker_lost=True<\/code> \u0442\u0440\u0435\u0431\u0443\u0435\u0442 \u0443\u0431\u0438\u0439\u0441\u0442\u0432\u0430 Celery worker \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0430 \u0438 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438 \u0447\u0442\u043e \u0437\u0430\u0434\u0430\u0447\u0430 \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0435\u043d\u0430 \u0432 \u0431\u0440\u043e\u043a\u0435\u0440. \u042d\u0442\u043e integration test, \u0436\u0438\u0432\u0451\u0442 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u043e.<\/p>\n
Redis DLQ fallback. \u0422\u0440\u0435\u0431\u0443\u0435\u0442 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u0433\u043e Redis \u0438 \u0438\u043c\u0438\u0442\u0430\u0446\u0438\u0438 \u043f\u0430\u0434\u0435\u043d\u0438\u044f PostgreSQL. \u041f\u0440\u043e\u0432\u0435\u0440\u044f\u0435\u0442\u0441\u044f \u0447\u0435\u0440\u0435\u0437 chaos testing.<\/p>\n
Backpressure \u0438 \u0434\u0435\u0433\u0440\u0430\u0434\u0430\u0446\u0438\u044f  <\/h2>\n
\u041f\u043e\u0434 \u0432\u044b\u0441\u043e\u043a\u043e\u0439 \u043a\u043e\u043d\u043a\u0443\u0440\u0435\u043d\u0446\u0438\u0435\u0439 \u043d\u0430 \u043e\u0434\u043d\u043e\u0433\u043e \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044f RetryableError<\/code> \u043e\u0442 NOWAIT<\/code> \u043d\u0430\u043a\u0430\u043f\u043b\u0438\u0432\u0430\u044e\u0442\u0441\u044f. max_retries=5<\/code> \u0434\u0430\u0451\u0442 6 \u043f\u043e\u043f\u044b\u0442\u043e\u043a (initial + 5 retry), \u0441\u0443\u043c\u043c\u0430\u0440\u043d\u044b\u0439 worst case delay \u0434\u043e ~63\u0441. \u0415\u0441\u043b\u0438 throughput \u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0445 webhooks \u043f\u0440\u0435\u0432\u044b\u0448\u0430\u0435\u0442 \u0441\u043f\u043e\u0441\u043e\u0431\u043d\u043e\u0441\u0442\u044c \u0440\u0430\u0437\u0433\u0440\u0435\u0431\u0430\u0442\u044c retry, CeleryQueueDepth<\/code> \u043d\u0430\u0447\u043d\u0451\u0442 \u0440\u0430\u0441\u0442\u0438.<\/p>\n
\u041a\u043e\u0433\u0434\u0430 DLQ \u043d\u0430\u0447\u0438\u043d\u0430\u0435\u0442 \u0437\u0430\u043f\u043e\u043b\u043d\u044f\u0442\u044c\u0441\u044f: alert_zombie_events<\/code> \u0437\u0430\u0444\u0438\u043a\u0441\u0438\u0440\u0443\u0435\u0442 dlq_size > 10 events\/hour<\/code>, \u043f\u0435\u0440\u0432\u044b\u0439 \u0441\u0438\u0433\u043d\u0430\u043b. PostgreSQL dead_letter_queue<\/code> \u0440\u0430\u0441\u0442\u0451\u0442 \u0431\u0435\u0437 \u043e\u0433\u0440\u0430\u043d\u0438\u0447\u0435\u043d\u0438\u0439, \u0442\u0435\u043e\u0440\u0435\u0442\u0438\u0447\u0435\u0441\u043a\u0438 \u0434\u043e \u0440\u0430\u0437\u043c\u0435\u0440\u0430 \u0434\u0438\u0441\u043a\u0430. \u0414\u0430\u043b\u044c\u0448\u0435 \u0440\u0443\u0447\u043d\u0430\u044f \u0440\u0430\u0431\u043e\u0442\u0430: \u0440\u0430\u0437\u043e\u0431\u0440\u0430\u0442\u044c \u043f\u0440\u0438\u0447\u0438\u043d\u0443, \u043f\u043e\u0447\u0438\u043d\u0438\u0442\u044c, \u0441\u0434\u0435\u043b\u0430\u0442\u044c replay \u0438\u0437 DLQ.<\/p>\n
\u0410\u0432\u0442\u043e\u0440\u0435\u043f\u043b\u0435\u0439 \u043d\u0435 \u0441\u0434\u0435\u043b\u0430\u043d \u0441\u043e\u0437\u043d\u0430\u0442\u0435\u043b\u044c\u043d\u043e. \u0421\u043e\u0431\u044b\u0442\u0438\u0435 \u043f\u043e\u043f\u0430\u043b\u043e \u0432 DLQ, \u0437\u043d\u0430\u0447\u0438\u0442 \u0447\u0442\u043e-\u0442\u043e \u043f\u043e\u0448\u043b\u043e \u043d\u0435 \u0442\u0430\u043a. \u041f\u0443\u0441\u0442\u044c \u0447\u0435\u043b\u043e\u0432\u0435\u043a \u0440\u0430\u0437\u0431\u0435\u0440\u0451\u0442\u0441\u044f \u043f\u0440\u0435\u0436\u0434\u0435, \u0447\u0435\u043c \u0433\u043d\u0430\u0442\u044c \u0435\u0433\u043e \u043e\u0431\u0440\u0430\u0442\u043d\u043e. \u041d\u0435 \u0445\u043e\u0447\u0443, \u0447\u0442\u043e\u0431\u044b \u0441\u0438\u0441\u0442\u0435\u043c\u0430 \u0441\u0430\u043c\u0430 \u0440\u0435\u0448\u0430\u043b\u0430 \u0447\u0442\u043e \u0434\u0435\u043b\u0430\u0442\u044c \u0441 \u0434\u0435\u043d\u044c\u0433\u0430\u043c\u0438, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0443\u0436\u0435 \u043e\u0434\u0438\u043d \u0440\u0430\u0437 \u0443\u043f\u0430\u043b\u0438.<\/p>\n
\u0415\u0441\u043b\u0438 Redis broker \u043d\u0435\u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d, apply_async<\/code> \u0432\u0441\u0435\u0433\u0434\u0430 \u043f\u0430\u0434\u0430\u0435\u0442. \u0421\u043e\u0431\u044b\u0442\u0438\u0435 \u043e\u0441\u0442\u0430\u0451\u0442\u0441\u044f \u0432 enqueued<\/code>, \u0447\u0435\u0440\u0435\u0437 3 \u043c\u0438\u043d\u0443\u0442\u044b recover_stale_enqueued_events<\/code> \u043f\u0435\u0440\u0435\u0432\u043e\u0434\u0438\u0442 \u043e\u0431\u0440\u0430\u0442\u043d\u043e \u0432 pending<\/code> \u0438 \u0438\u043d\u043a\u0440\u0435\u043c\u0435\u043d\u0442\u0438\u0442 retry_count<\/code>. \u041f\u043e\u0441\u043b\u0435 MAX_RECOVERY_ATTEMPTS<\/code> (10 \u043f\u043e\u043f\u044b\u0442\u043e\u043a ~ 30 \u043c\u0438\u043d\u0443\u0442) \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0443\u0445\u043e\u0434\u0438\u0442 \u0432 DLQ + \u0430\u043b\u0435\u0440\u0442.<\/p>\n
acks_late<\/code> \u0438 reject_on_worker_lost<\/code> \u0437\u0430\u0449\u0438\u0449\u0430\u044e\u0442 \u043e\u0442 \u043f\u0430\u0434\u0435\u043d\u0438\u044f \u0432\u043e\u0440\u043a\u0435\u0440\u0430, \u043d\u0435 \u0431\u0440\u043e\u043a\u0435\u0440\u0430. \u0415\u0441\u043b\u0438 master Redis \u0443\u043f\u0430\u0434\u0451\u0442, in-flight \u0437\u0430\u0434\u0430\u0447\u0438 \u043f\u043e\u0442\u0435\u0440\u044f\u043d\u044b. appendonly yes<\/code> + appendfsync everysec<\/code>, \u043c\u0438\u043d\u0438\u043c\u0443\u043c \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0434\u043e\u043b\u0436\u0435\u043d \u0431\u044b\u0442\u044c. \u0415\u0441\u043b\u0438 \u0441\u043e\u0432\u0441\u0435\u043c \u043d\u0435 \u0445\u043e\u0447\u0435\u0448\u044c \u0442\u0435\u0440\u044f\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435, appendfsync always<\/code>, \u043d\u043e throughput \u043f\u0440\u043e\u0441\u044f\u0434\u0435\u0442.<\/p>\n
\u041f\u0440\u043e blockchain reorg  <\/h2>\n
\u0414\u043b\u044f Ethereum \u043f\u043e\u0441\u043b\u0435 \u043f\u0435\u0440\u0435\u0445\u043e\u0434\u0430 \u043d\u0430 PoS \u0444\u0438\u043d\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f \u0447\u0435\u0440\u0435\u0437 \u0434\u0432\u0430 checkpoint \u044d\u043f\u043e\u0445\u0438 (~12-15 \u043c\u0438\u043d\u0443\u0442). \u042d\u0432\u0440\u0438\u0441\u0442\u0438\u043a\u0430 \u00ab12 \u0431\u043b\u043e\u043a\u043e\u0432\u00bb \u0438\u0437 \u044d\u043f\u043e\u0445\u0438 PoW \u0441\u0435\u0439\u0447\u0430\u0441 \u043d\u0435 \u043f\u0440\u0438\u043c\u0435\u043d\u0438\u043c\u0430. \u0414\u043b\u044f L2 (Arbitrum, Optimism) \u043f\u0440\u0430\u0432\u0438\u043b\u0430 \u0434\u0440\u0443\u0433\u0438\u0435, \u0437\u0434\u0435\u0441\u044c \u0442\u043e\u043b\u044c\u043a\u043e \u043f\u0440\u043e Ethereum L1.<\/p>\n
\u0422\u0435\u043a\u0443\u0449\u0430\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f reorg \u043d\u0435 \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0435\u0442, \u044d\u0442\u043e \u0441\u043a\u043e\u0443\u043f \u043f\u0435\u0440\u0432\u043e\u0433\u043e \u0440\u0435\u043b\u0438\u0437\u0430. \u041f\u0440\u0438 \u0440\u0435\u043e\u0440\u0433\u0435 \u0431\u043b\u043e\u043a\u0447\u0435\u0439\u043d \u00ab\u043e\u0442\u043a\u0430\u0442\u044b\u0432\u0430\u0435\u0442\u00bb \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0431\u043b\u043e\u043a\u043e\u0432. \u0422\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u043c\u043e\u0436\u0435\u0442 \u043f\u043e\u043f\u0430\u0441\u0442\u044c \u0432 \u043d\u043e\u0432\u044b\u0439 \u0431\u043b\u043e\u043a \u0431\u0435\u0437 \u0438\u0437\u043c\u0435\u043d\u0435\u043d\u0438\u0439 \u0438\u043b\u0438 \u043d\u0435 \u043f\u043e\u043f\u0430\u0441\u0442\u044c \u043d\u0438\u043a\u0443\u0434\u0430, \u044d\u0444\u0444\u0435\u043a\u0442\u0438\u0432\u043d\u043e \u043e\u0442\u043c\u0435\u043d\u0435\u043d\u0430.<\/p>\n
\u0421\u0430\u043c\u044b\u0439 \u043f\u0440\u043e\u0441\u0442\u043e\u0439 \u0432\u0430\u0440\u0438\u0430\u043d\u0442: \u043a\u043e\u043c\u043f\u0435\u043d\u0441\u0438\u0440\u0443\u044e\u0449\u0430\u044f \u0437\u0430\u043f\u0438\u0441\u044c \u0432 balance_events<\/code> \u0441\u043e \u0437\u043d\u0430\u043a\u043e\u043c \u043c\u0438\u043d\u0443\u0441, \u0441 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u043c idempotency key \u0447\u0442\u043e\u0431\u044b \u043d\u0435 \u043a\u043e\u043d\u0444\u043b\u0438\u043a\u0442\u043e\u0432\u0430\u0442\u044c \u0441 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b\u044c\u043d\u044b\u043c \u0441\u043e\u0431\u044b\u0442\u0438\u0435\u043c:<\/p>\n
def handle_reorg_event(original_tx_hash: str, user_id: int, amount: Decimal) -> None:    reorg_event_id  = f\"reorg:{original_tx_hash}\"    idempotency_key = _idempotency_key(reorg_event_id, \"reorg_compensation\")    # \u0434\u0430\u043b\u0435\u0435 \u0441\u0442\u0430\u043d\u0434\u0430\u0440\u0442\u043d\u044b\u0439 flow \u0447\u0435\u0440\u0435\u0437 processed_events + balance_events<\/code>
<\/a><\/div><\/pre>\n
\u0427\u0442\u043e \u0432 \u0431\u044d\u043a\u043b\u043e\u0433\u0435  <\/h2>\n
\u0421\u0430\u043c\u043e\u0435 \u0431\u043e\u043b\u0435\u0437\u043d\u0435\u043d\u043d\u043e\u0435 \u043f\u0440\u044f\u043c\u043e \u0441\u0435\u0439\u0447\u0430\u0441: notify_user_insufficient_funds<\/code> \u0432\u044b\u0437\u044b\u0432\u0430\u0435\u0442\u0441\u044f \u0432\u043d\u0435 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0438, \u043d\u0443\u0436\u0435\u043d outbox pattern, \u0437\u0430\u043f\u0438\u0441\u044c \u0432 outbox-\u0442\u0430\u0431\u043b\u0438\u0446\u0443 \u0432\u043d\u0443\u0442\u0440\u0438 \u0442\u043e\u0439 \u0436\u0435 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0438 \u0447\u0442\u043e \u0438 UPDATE users<\/code>. \u0411\u0435\u0437 \u044d\u0442\u043e\u0433\u043e \u043f\u0440\u0438 at-least-once delivery \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u0442 N \u0443\u0432\u0435\u0434\u043e\u043c\u043b\u0435\u043d\u0438\u0439 \u043d\u0430 \u043e\u0434\u0438\u043d \u043e\u0442\u043a\u0430\u0437, \u0430 \u043f\u0440\u0438 \u0441\u0431\u043e\u0435 notify<\/code> \u0432 DLQ \u043f\u043e\u043f\u0430\u0434\u0430\u044e\u0442 \u0437\u0430\u043f\u0438\u0441\u0438 \u043e\u0431 \u0443\u0441\u043f\u0435\u0448\u043d\u043e \u0437\u0430\u0432\u0435\u0440\u0448\u0451\u043d\u043d\u044b\u0445 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f\u0445. \u041f\u0440\u0438 >50 \u0432\u043e\u0440\u043a\u0435\u0440\u043e\u0432 \u043f\u0440\u044f\u043c\u044b\u0435 \u0441\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u044f \u0443\u043f\u0438\u0440\u0430\u044e\u0442\u0441\u044f \u0432 max_connections<\/code>, \u043d\u0443\u0436\u0435\u043d PgBouncer. \u041c\u0438\u043d\u0430 \u043a\u043e\u0442\u043e\u0440\u0443\u044e \u0441\u0442\u043e\u0438\u0442 \u0443\u0447\u0435\u0441\u0442\u044c \u0434\u043e \u0435\u0433\u043e \u0432\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u044f \u0432 transaction pooling: hot_path_balance_check<\/code> \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442 conn.set_isolation_level(REPEATABLE_READ)<\/code>, \u044d\u0442\u043e session-level \u043a\u043e\u043c\u0430\u043d\u0434\u0430, PgBouncer \u0432 transaction pooling \u0435\u0451 \u043d\u0435 \u0441\u043e\u0445\u0440\u0430\u043d\u044f\u0435\u0442 \u043c\u0435\u0436\u0434\u0443 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f\u043c\u0438. SET LOCAL lock_timeout\/statement_timeout<\/code> \u0440\u0430\u0431\u043e\u0442\u0430\u044e\u0442 \u043d\u043e\u0440\u043c\u0430\u043b\u044c\u043d\u043e (\u043e\u043d\u0438 tx-scoped), \u0430 isolation level \u043f\u0440\u0438\u0434\u0451\u0442\u0441\u044f \u043f\u0435\u0440\u0435\u043f\u0438\u0441\u0430\u0442\u044c \u043d\u0430 BEGIN ISOLATION LEVEL REPEATABLE READ<\/code> \u043f\u0440\u044f\u043c\u043e \u0432 \u0437\u0430\u043f\u0440\u043e\u0441\u0435 \u0438\u043b\u0438 \u043e\u0441\u0442\u0430\u0432\u0438\u0442\u044c session pooling \u0434\u043b\u044f \u044d\u0442\u043e\u0439 \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u043e\u0439 \u0442\u0430\u0441\u043a\u0438. \u042d\u0442\u043e \u043a\u043b\u0430\u0441\u0441\u0438\u0447\u0435\u0441\u043a\u0430\u044f \u0441\u0435\u043d\u044c\u043e\u0440\u0441\u043a\u0430\u044f \u043c\u0438\u043d\u0430: \u0432 dev \u0432\u0441\u0451 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442, \u043b\u043e\u043c\u0430\u0435\u0442\u0441\u044f \u0442\u043e\u043b\u044c\u043a\u043e \u043f\u043e\u0441\u043b\u0435 \u0432\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u044f PgBouncer \u0432 \u043f\u0440\u043e\u0434\u0435. processed_events<\/code> \u0440\u0430\u0441\u0442\u0451\u0442, cleanup \u0437\u0430\u043a\u0440\u044b\u0432\u0430\u0435\u0442 \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0443 \u043d\u0430 \u0431\u043b\u0438\u0436\u0430\u0439\u0448\u0435\u0435 \u0432\u0440\u0435\u043c\u044f, \u043d\u043e \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u043e\u043d\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u0435 \u043f\u043e \u0434\u0430\u0442\u0435 \u043f\u0440\u0438 \u043f\u0435\u0440\u0435\u0445\u043e\u0434\u0435 \u0437\u0430 \u0441\u043e\u0442\u043d\u0438 \u043c\u0438\u043b\u043b\u0438\u043e\u043d\u043e\u0432 \u0441\u0442\u0440\u043e\u043a \u043d\u0435\u0438\u0437\u0431\u0435\u0436\u043d\u043e.<\/p>\n
\u0414\u0430\u043b\u044c\u0448\u0435: Redis broker \u0441 appendfsync always<\/code>, \u043f\u043e\u043b\u043d\u0430\u044f \u0438\u0441\u0442\u043e\u0440\u0438\u0447\u0435\u0441\u043a\u0430\u044f \u0441\u0432\u0435\u0440\u043a\u0430 \u0447\u0435\u0440\u0435\u0437 materialized view, \u043f\u043e\u043b\u043d\u0430\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f reorg handling, Admin UI \u0434\u043b\u044f DLQ \u0432\u043c\u0435\u0441\u0442\u043e \u0440\u0443\u0447\u043d\u043e\u0433\u043e SQL. \u041e\u0434\u043d\u0430 \u043b\u043e\u0432\u0443\u0448\u043a\u0430 \u0432 \u043a\u043e\u0434\u0435 \u043a\u043e\u0442\u043e\u0440\u0443\u044e \u044f \u043f\u043e\u043a\u0430 \u043d\u0435 \u0442\u0440\u043e\u0433\u0430\u043b: _mark_event_failed<\/code> \u043a\u043e\u043c\u043c\u0438\u0442\u0438\u0442 \u0441\u0430\u043c\u0430, \u043f\u0440\u0438 \u0431\u0443\u0434\u0443\u0449\u0435\u043c \u0440\u0435\u0444\u0430\u043a\u0442\u043e\u0440\u0438\u043d\u0433\u0435 \u044d\u0442\u043e \u0442\u0435\u0431\u044f \u0443\u043a\u0443\u0441\u0438\u0442.<\/p>\n
\u0420\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442  <\/h2>\n
\u041e\u0447\u0435\u0440\u0435\u0434\u044c \u0440\u0430\u0437\u0433\u0440\u0435\u0431\u0430\u0435\u0442\u0441\u044f \u0438 \u0441\u0443\u043c\u043c\u044b \u0441\u0445\u043e\u0434\u044f\u0442\u0441\u044f, \u044d\u0442\u043e \u0440\u0430\u0437\u043d\u044b\u0435 \u0432\u0435\u0449\u0438. Prometheus \u043f\u043e\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u0442 \u043f\u0435\u0440\u0432\u043e\u0435, hot_path_balance_check<\/code> \u0432\u0442\u043e\u0440\u043e\u0435. \u041d\u0443\u0436\u043d\u044b \u043e\u0431\u0430, \u043e\u0434\u043d\u0438\u043c \u043d\u0435 \u043e\u0431\u043e\u0439\u0442\u0438\u0441\u044c.<\/p>\n
8 \u043c\u0435\u0441\u044f\u0446\u0435\u0432 \u0432 \u043f\u0440\u043e\u0434\u0435. 0 \u0434\u0443\u0431\u043b\u0438\u0440\u0443\u044e\u0449\u0438\u0445 \u0437\u0430\u0447\u0438\u0441\u043b\u0435\u043d\u0438\u0439 \u043f\u043e\u0441\u043b\u0435 \u0434\u0435\u043f\u043b\u043e\u044f \u0444\u0438\u043a\u0441\u0430, \u043f\u0440\u043e\u0442\u0438\u0432 23 \u0437\u0430 \u043f\u0435\u0440\u0432\u044b\u0439 \u043c\u0435\u0441\u044f\u0446 \u043d\u0430 ~180k \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0439. Webhook-\u0434\u043e\u0441\u0442\u0430\u0432\u043a\u0430 100% \u0441 \u0443\u0447\u0451\u0442\u043e\u043c retry-\u043b\u043e\u0433\u0438\u043a\u0438 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440\u0430.<\/p>\n
\u0417\u0434\u0435\u0441\u044c \u043d\u0435\u0442 \u043d\u0438 \u043e\u0434\u043d\u043e\u0433\u043e \u0440\u0435\u0448\u0435\u043d\u0438\u044f \u043a\u043e\u0442\u043e\u0440\u043e\u0435 \u044f \u043f\u0440\u0438\u0434\u0443\u043c\u0430\u043b \u0437\u0430\u0440\u0430\u043d\u0435\u0435. \u041a\u0430\u0436\u0434\u043e\u0435 \u0437\u0430\u043a\u0440\u044b\u0432\u0430\u0435\u0442 \u0434\u044b\u0440\u0443 \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u0443\u0436\u0435 \u0441\u0442\u0440\u0435\u043b\u044c\u043d\u0443\u043b\u0430. Idempotency \u0447\u0435\u0440\u0435\u0437 DB unique constraint, TOCTOU \u0447\u0435\u0440\u0435\u0437 SELECT FOR UPDATE NOWAIT<\/code>, \u043f\u043e\u0442\u0435\u0440\u044f \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0439 \u0447\u0435\u0440\u0435\u0437 acks_late<\/code> + outbox, FSM \u0447\u0435\u0440\u0435\u0437 VALID_TRANSITIONS<\/code>. \u041a\u0430\u0436\u0434\u043e\u0435 \u0438\u0437 \u044d\u0442\u0438\u0445 \u0440\u0435\u0448\u0435\u043d\u0438\u0439 \u043f\u043e \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u043e\u0441\u0442\u0438 \u043d\u0435 \u0433\u0430\u0440\u0430\u043d\u0442\u0438\u044f. \u0412\u043c\u0435\u0441\u0442\u0435 \u0437\u0430\u043a\u0440\u044b\u0432\u0430\u044e\u0442 \u0434\u0440\u0443\u0433 \u0434\u0440\u0443\u0433\u0430. \u0412\u043e\u0441\u0435\u043c\u044c \u043c\u0435\u0441\u044f\u0446\u0435\u0432 \u0431\u0435\u0437 \u0438\u043d\u0446\u0438\u0434\u0435\u043d\u0442\u043e\u0432.<\/p>\n<\/div>\n
\u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 https:\/\/habr.com\/ru\/articles\/1028708\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"
\u041a\u043e\u0434 \u0432 \u0441\u0442\u0430\u0442\u044c\u0435 \u0438\u043b\u043b\u044e\u0441\u0442\u0440\u0430\u0442\u0438\u0432\u043d\u044b\u0439, \u043f\u043e\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u0442 \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u043d\u044b\u0435 \u0440\u0435\u0448\u0435\u043d\u0438\u044f \u0438 \u043e\u0431\u044a\u044f\u0441\u043d\u044f\u0435\u0442 \u043f\u043e\u0447\u0435\u043c\u0443 \u0438\u043c\u0435\u043d\u043d\u043e \u0442\u0430\u043a. \u041d\u0435 \u043f\u0440\u0435\u0434\u043d\u0430\u0437\u043d\u0430\u0447\u0435\u043d \u0434\u043b\u044f copy-paste \u0432 \u043f\u0440\u043e\u0434 \u0431\u0435\u0437 \u0430\u0434\u0430\u043f\u0442\u0430\u0446\u0438\u0438 \u043f\u043e\u0434 \u0432\u0430\u0448\u0443 \u0438\u043d\u0444\u0440\u0430\u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443, \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433 \u0438 \u0442\u0440\u0435\u0431\u043e\u0432\u0430\u043d\u0438\u044f.  \u0414\u0443\u043c\u0430\u043b, \u0437\u0430\u0439\u0434\u0443 \u0432 \u043a\u0440\u0438\u043f\u0442\u0443 \u0438 \u0431\u0443\u0434\u0443 \u043f\u0440\u043e\u0441\u0442\u043e \u0434\u0451\u0440\u0433\u0430\u0442\u044c API \u0431\u043b\u043e\u043a\u0447\u0435\u0439\u043d\u0430. \u041d\u0435 \u0432\u044b\u0448\u043b\u043e.\u0417\u0430\u0445\u043e\u0436\u0443 \u0432 \u043f\u0440\u043e\u0435\u043a\u0442. \u0421\u0442\u0435\u043a: FastAPI, PostgreSQL, Redis \u043a\u0430\u043a Celery broker, Celery workers, Docker, Web3. \u0421\u0442\u0430\u0440\u0442\u0430\u043f \u043d\u0430 \u0445\u0430\u0439\u043f\u0435, \u0434\u0435\u043d\u044c\u0433\u0438 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u0435, \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u0430 \u0441\u043e\u0431\u0440\u0430\u043d\u0430 \u043d\u0430 \u043a\u043e\u043b\u0435\u043d\u043a\u0435. \u0421\u043c\u043e\u0442\u0440\u044e \u043d\u0430 \u0430\u0440\u0445\u0438\u0442\u0435\u043a\u0442\u0443\u0440\u0443 \u043f\u043b\u0430\u0442\u0451\u0436\u043d\u043e\u0433\u043e \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0438\u043d\u0433\u0430 \u0438 \u043f\u0435\u0440\u0432\u0430\u044f \u043c\u044b\u0441\u043b\u044c: \u0440\u0435\u0431\u044f\u0442\u0430, \u0432\u044b \u0441\u0435\u0440\u044c\u0451\u0437\u043d\u043e? \u0424\u0438\u043d\u0430\u043d\u0441\u043e\u0432\u044b\u0435 \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u0438 \u0441 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u043c\u0438 \u0434\u0435\u043d\u044c\u0433\u0430\u043c\u0438, \u0431\u0435\u0437 idempotency \u0432\u043e\u043e\u0431\u0449\u0435, Redis \u043a\u0430\u043a \u0431\u0440\u043e\u043a\u0435\u0440 \u0431\u0435\u0437 persistence, Web3.py \u0441\u0438\u043d\u0445\u0440\u043e\u043d\u043d\u044b\u0435 \u0432\u044b\u0437\u043e\u0432\u044b \u0432\u043d\u0443\u0442\u0440\u0438 Celery \u0442\u0430\u0441\u043a\u043e\u0432.\u0420\u0430\u0437\u0433\u043e\u0432\u043e\u0440 \u0431\u044b\u043b \u043a\u043e\u0440\u043e\u0442\u043a\u0438\u0439: \u0437\u0430\u0434\u0430\u0447\u0430 \u0442\u0430\u043a\u0430\u044f, \u0447\u0438\u043d\u0438 \u0447\u0442\u043e \u0435\u0441\u0442\u044c. \u0421\u0440\u043e\u043a\u0438 \u0433\u043e\u0440\u0435\u043b\u0438.\u0427\u0442\u043e \u0431\u044b\u043b\u043e \u0441\u043b\u043e\u043c\u0430\u043d\u043e  \u041f\u0435\u0440\u0432\u044b\u0439 \u043c\u0435\u0441\u044f\u0446 \u043f\u0440\u043e\u0434\u0430. \u041f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u043f\u0438\u0448\u0435\u0442 \u0432 \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u043a\u0443: \u0437\u0430\u0447\u0438\u0441\u043b\u0438\u043b\u0438 \u0434\u0432\u0430\u0436\u0434\u044b, \u0432\u044b\u0432\u0435\u043b \u0434\u0432\u043e\u0439\u043d\u0443\u044e \u0441\u0443\u043c\u043c\u0443. \u041e\u0442\u043a\u0440\u044b\u0432\u0430\u044e \u043b\u043e\u0433\u0438, \u0447\u0438\u0441\u0442\u043e. \u0414\u0432\u0430 \u0438\u0434\u0435\u043d\u0442\u0438\u0447\u043d\u044b\u0445 \u0441\u043e\u0431\u044b\u0442\u0438\u044f, \u043e\u0431\u0430 200, \u0440\u0430\u0437\u043d\u0438\u0446\u0430 \u0447\u0435\u0442\u044b\u0440\u0435 \u0441\u0435\u043a\u0443\u043d\u0434\u044b. \u041e\u0431\u0430 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u0430\u043d\u044b. \u041f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u043f\u043e\u043b\u0443\u0447\u0438\u043b \u0434\u0432\u043e\u0439\u043d\u043e\u0439 \u0431\u0430\u043b\u0430\u043d\u0441.\u0415\u0436\u0435\u0434\u043d\u0435\u0432\u043d\u0430\u044f \u0441\u0432\u0435\u0440\u043a\u0430 \u0441 on-chain \u0434\u0430\u043d\u043d\u044b\u043c\u0438 \u043f\u043e\u043a\u0430\u0437\u0430\u043b\u0430 \u0440\u0430\u0441\u0445\u043e\u0436\u0434\u0435\u043d\u0438\u0435: \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0430\u043a\u043a\u0430\u0443\u043d\u0442\u043e\u0432 \u0441 \u0431\u0430\u043b\u0430\u043d\u0441\u043e\u043c \u0431\u043e\u043b\u044c\u0448\u0435 \u0447\u0435\u043c \u0434\u043e\u043b\u0436\u043d\u043e \u0431\u044b\u0442\u044c \u043f\u043e confirmed \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f\u043c. \u0417\u0430 \u043f\u0435\u0440\u0432\u044b\u0439 \u043c\u0435\u0441\u044f\u0446 \u043d\u0430\u0448\u043b\u0438 23 \u0434\u0443\u0431\u043b\u0438\u0440\u0443\u044e\u0449\u0438\u0445 \u0437\u0430\u0447\u0438\u0441\u043b\u0435\u043d\u0438\u044f \u043d\u0430 ~180k \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u0439, \u043e\u043a\u043e\u043b\u043e 0.013% error rate. 23 \u0434\u0432\u043e\u0439\u043d\u044b\u0445 \u0437\u0430\u0447\u0438\u0441\u043b\u0435\u043d\u0438\u044f \u0437\u0430 \u043c\u0435\u0441\u044f\u0446. \u0416\u0438\u0432\u044b\u0435 \u0434\u0435\u043d\u044c\u0433\u0438, \u043d\u0435 \u043c\u0435\u0442\u0440\u0438\u043a\u0430.\u041f\u0435\u0440\u0432\u043e\u0435, \u0447\u0442\u043e \u0432\u044b\u043b\u0435\u0437\u043b\u043e: \u0434\u0443\u0431\u043b\u0438 \u043e\u0442 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440\u0430. Alchemy, Infura \u0438 \u0432\u0441\u0435 \u043e\u0441\u0442\u0430\u043b\u044c\u043d\u044b\u0435 \u0431\u043b\u043e\u043a\u0447\u0435\u0439\u043d-\u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440\u044b \u0440\u0430\u0431\u043e\u0442\u0430\u044e\u0442 \u043f\u043e at-least-once delivery. \u041f\u0440\u0438 \u0441\u0435\u0442\u0435\u0432\u043e\u043c \u0441\u0431\u043e\u0435, \u0440\u0435\u0441\u0442\u0430\u0440\u0442\u0435, \u043f\u043e\u0434 \u043d\u0430\u0433\u0440\u0443\u0437\u043a\u043e\u0439 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u043f\u043e\u0432\u0442\u043e\u0440\u044f\u0435\u0442 \u0434\u043e\u0441\u0442\u0430\u0432\u043a\u0443. \u041f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u0442\u0430\u043a \u0438 \u043d\u0430\u043f\u0438\u0441\u0430\u043b \u0432 \u0434\u043e\u043a\u0430\u0445. \u042d\u0442\u043e \u043d\u0435 \u0431\u0430\u0433, \u044d\u0442\u043e \u0443\u0441\u043b\u043e\u0432\u0438\u044f \u0438\u0433\u0440\u044b. \u041f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u043f\u043e\u0432\u0442\u043e\u0440\u044f\u0435\u0442 \u0434\u043e\u0441\u0442\u0430\u0432\u043a\u0443, \u0442\u0432\u043e\u0439 \u043a\u043e\u0434 \u0434\u043e\u043b\u0436\u0435\u043d \u044d\u0442\u043e \u043f\u0435\u0440\u0435\u0436\u0438\u0432\u0430\u0442\u044c \u0431\u0435\u0437 \u043f\u043e\u0441\u043b\u0435\u0434\u0441\u0442\u0432\u0438\u0439. \u041d\u0430\u0448 \u043d\u0435 \u043f\u0435\u0440\u0435\u0436\u0438\u0432\u0430\u043b.\u0414\u0430\u043b\u044c\u0448\u0435 \u0445\u0443\u0436\u0435. \u0414\u0432\u0430 \u043f\u0430\u0440\u0430\u043b\u043b\u0435\u043b\u044c\u043d\u044b\u0445 \u0437\u0430\u043f\u0440\u043e\u0441\u0430 \u043d\u0430 \u0432\u044b\u0432\u043e\u0434 \u0447\u0438\u0442\u0430\u043b\u0438 \u0431\u0430\u043b\u0430\u043d\u0441 \u043e\u0434\u043d\u043e\u0432\u0440\u0435\u043c\u0435\u043d\u043d\u043e, \u043e\u0431\u0430 \u0432\u0438\u0434\u0435\u043b\u0438 \u0434\u043e\u0441\u0442\u0430\u0442\u043e\u0447\u043d\u043e \u0441\u0440\u0435\u0434\u0441\u0442\u0432, \u043e\u0431\u0430 \u043f\u0440\u043e\u0445\u043e\u0434\u0438\u043b\u0438 \u0432\u0430\u043b\u0438\u0434\u0430\u0446\u0438\u044e. \u0414\u0432\u0430 \u0437\u0430\u043f\u0440\u043e\u0441\u0430 \u0447\u0438\u0442\u0430\u044e\u0442 \u0431\u0430\u043b\u0430\u043d\u0441 \u043e\u0434\u043d\u043e\u0432\u0440\u0435\u043c\u0435\u043d\u043d\u043e, \u043e\u0431\u0430 \u0432\u0438\u0434\u044f\u0442 \u0447\u0442\u043e \u0434\u0435\u043d\u0435\u0433 \u0445\u0432\u0430\u0442\u0430\u0435\u0442, \u043e\u0431\u0430 \u0441\u043f\u0438\u0441\u044b\u0432\u0430\u044e\u0442. \u0428\u043a\u043e\u043b\u044c\u043d\u0430\u044f \u0433\u043e\u043d\u043a\u0430.async def withdraw(conn, user_id: int, amount: Decimal):    balance = await conn.fetchval(        «SELECT balance FROM users WHERE id = $1», user_id    )    if balance >= amount:        await conn.execute(            «UPDATE users SET balance = balance — $1 WHERE id = $2»,            amount, user_id        )\u0414\u0430\u043b\u044c\u0448\u0435. Celery \u0441 \u0434\u0435\u0444\u043e\u043b\u0442\u043d\u044b\u043c\u0438 \u043d\u0430\u0441\u0442\u0440\u043e\u0439\u043a\u0430\u043c\u0438 \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0430\u0435\u0442 \u0437\u0430\u0434\u0430\u0447\u0443 \u0431\u0440\u043e\u043a\u0435\u0440\u0443 \u0432 \u043c\u043e\u043c\u0435\u043d\u0442 \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f. \u0412\u043e\u0440\u043a\u0435\u0440 \u043f\u0430\u0434\u0430\u0435\u0442 \u0432 \u0441\u0435\u0440\u0435\u0434\u0438\u043d\u0435 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438, \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u043e, \u0434\u043e \u0437\u0430\u043f\u0438\u0441\u0438 \u0432 \u0411\u0414 \u043d\u0435 \u0434\u043e\u0448\u043b\u043e. \u041d\u0438\u043a\u0430\u043a\u043e\u0433\u043e retry, \u043d\u0438\u043a\u0430\u043a\u043e\u0433\u043e DLQ. \u0412\u043e\u0440\u043a\u0435\u0440 \u0443\u043f\u0430\u043b, \u0437\u0430\u0434\u0430\u0447\u0430 \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u0430, \u0434\u0435\u043d\u044c\u0433\u0438 \u043d\u0435 \u043f\u0440\u0438\u0448\u043b\u0438. \u041f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c \u0436\u0434\u0451\u0442 \u0438 \u043d\u0435 \u043f\u043e\u043d\u0438\u043c\u0430\u0435\u0442 \u0447\u0442\u043e \u0441\u043b\u0443\u0447\u0438\u043b\u043e\u0441\u044c.\u0418 \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0439 \u0442\u0438\u0445\u0438\u0439 \u0443\u0431\u0438\u0439\u0446\u0430: amount \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0443\u0435\u0442\u0441\u044f \u0432 JSON \u0447\u0435\u0440\u0435\u0437 Celery broker \u043a\u0430\u043a float. Decimal(«50.1») \u043f\u0440\u0435\u0432\u0440\u0430\u0449\u0430\u0435\u0442\u0441\u044f \u0432 JSON float, \u0442\u043e \u0435\u0441\u0442\u044c \u0432 50.099999999999994. \u041d\u0430 \u043c\u0430\u0441\u0448\u0442\u0430\u0431\u0435 \u044d\u0442\u043e \u043a\u043e\u043f\u0438\u0442\u0441\u044f \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u0439 \u0443\u0431\u044b\u0442\u043e\u043a. \u041d\u0438\u043a\u0442\u043e \u043d\u0435 \u0437\u0430\u043c\u0435\u0442\u0438\u043b, \u043f\u043e\u043a\u0430 \u043d\u0435 \u043f\u043e\u0441\u0447\u0438\u0442\u0430\u043b\u0438.\u041f\u043e\u0441\u043b\u0435\u0434\u043d\u0435\u0435: \u043f\u0440\u044f\u043c\u043e\u0439 \u0432\u044b\u0437\u043e\u0432 .delay() \u0438\u0437 webhook handler \u0441\u043e\u0437\u0434\u0430\u0451\u0442 \u043e\u043a\u043d\u043e \u043c\u0435\u0436\u0434\u0443 \u0437\u0430\u043f\u0438\u0441\u044c\u044e \u0432 \u0411\u0414 \u0438 \u043f\u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u043e\u0439 \u0432 \u043e\u0447\u0435\u0440\u0435\u0434\u044c. \u0415\u0441\u043b\u0438 \u043f\u0440\u043e\u0446\u0435\u0441\u0441 \u0443\u043f\u0430\u043b \u0432 \u044d\u0442\u043e\u0442 \u043c\u043e\u043c\u0435\u043d\u0442, \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0437\u0430\u0432\u0438\u0441\u043d\u0435\u0442 \u0432 pending \u0431\u0435\u0437 \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u043e\u0433\u043e \u0432\u043e\u0441\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d\u0438\u044f.\u0418\u0442\u043e\u0433\u043e \u043f\u044f\u0442\u044c \u043f\u0440\u043e\u0431\u043b\u0435\u043c. \u041d\u0430\u0447\u0430\u043b \u0447\u0438\u043d\u0438\u0442\u044c.\u041f\u0435\u0440\u0432\u044b\u0439 \u0438\u043d\u0441\u0442\u0438\u043d\u043a\u0442: Redis distributed lock  SET NX EX \u043d\u0430 user_id. \u041f\u0430\u0442\u0442\u0435\u0440\u043d \u043e\u043f\u0438\u0441\u0430\u043d \u0443 Antirez, \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u043d \u0437\u0430 20 \u043c\u0438\u043d\u0443\u0442. \u041d\u0435 \u0432\u0437\u043b\u0435\u0442\u0435\u043b\u043e.\u0412\u043e\u0442 \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u044b\u0439 \u0441\u0446\u0435\u043d\u0430\u0440\u0438\u0439, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0432\u0441\u043a\u0440\u044b\u043b\u0441\u044f \u0432 \u043b\u043e\u0433\u0430\u0445. \u0412\u043e\u0440\u043a\u0435\u0440 \u0431\u0435\u0440\u0451\u0442 \u043b\u043e\u043a \u0432 Redis. \u041d\u0430\u0447\u0438\u043d\u0430\u0435\u0442 \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044e \u0432 PostgreSQL. \u041c\u0435\u0436\u0434\u0443 \u044d\u0442\u0438\u043c\u0438 \u0434\u0432\u0443\u043c\u044f \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u044f\u043c\u0438 OOM killer \u0443\u0431\u0438\u0432\u0430\u0435\u0442 \u043f\u0440\u043e\u0446\u0435\u0441\u0441. PostgreSQL \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u043e\u0442\u043a\u0430\u0442\u0438\u043b\u0430\u0441\u044c \u0430\u0432\u0442\u043e\u043c\u0430\u0442\u0438\u0447\u0435\u0441\u043a\u0438, \u0431\u0430\u043b\u0430\u043d\u0441 \u043d\u0435 \u0438\u0437\u043c\u0435\u043d\u0438\u043b\u0441\u044f. Redis \u043b\u043e\u043a \u0432\u0438\u0441\u0438\u0442 30 \u0441\u0435\u043a\u0443\u043d\u0434 \u0434\u043e TTL. \u0427\u0435\u0440\u0435\u0437 30 \u0441\u0435\u043a\u0443\u043d\u0434 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0439 \u0432\u043e\u0440\u043a\u0435\u0440 \u0431\u0435\u0440\u0451\u0442 \u043b\u043e\u043a, \u0432\u0438\u0434\u0438\u0442 \u0447\u0442\u043e  idempotency_key \u043d\u0435 \u0437\u0430\u043f\u0438\u0441\u0430\u043d (\u0437\u0430\u043f\u0438\u0441\u0430\u0442\u044c \u0431\u044b\u043b\u043e \u043d\u0435\u043a\u043e\u043c\u0443, \u0442\u0440\u0430\u043d\u0437\u0430\u043a\u0446\u0438\u044f \u043e\u0442\u043a\u0430\u0442\u0438\u043b\u0430\u0441\u044c) \u0438 \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0435\u0442 \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u0437\u0430\u043d\u043e\u0432\u043e. \u0414\u0432\u043e\u0439\u043d\u043e\u0435 \u0437\u0430\u0447\u0438\u0441\u043b\u0435\u043d\u0438\u0435. \u0412 \u043b\u043e\u0433\u0430\u0445 \u043e\u0431\u0430 \u0432\u043e\u0440\u043a\u0435\u0440\u0430 \u0447\u0438\u0441\u0442\u044b\u0435.\u041f\u0440\u043e\u0431\u043b\u0435\u043c\u0430 \u043d\u0435 \u0432 \u0440\u0430\u0437\u043c\u0435\u0440\u0435 TTL. \u041f\u0440\u043e\u0431\u043b\u0435\u043c\u0430 \u0432 \u043e\u0442\u0441\u0443\u0442\u0441\u0442\u0432\u0438\u0438 cross-system \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e\u0441\u0442\u0438 \u043c\u0435\u0436\u0434\u0443 Redis \u0438 PostgreSQL. Redis \u043d\u0435 \u043f\u043e\u0434\u0445\u043e\u0434\u0438\u0442, \u043d\u0435\u0442 \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e\u0441\u0442\u0438 \u0441 PostgreSQL. \u041f\u0440\u043e\u0432\u0435\u0440\u043a\u0430 \u0432 \u043a\u043e\u0434\u0435 \u0442\u043e\u0436\u0435 \u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442, \u0434\u0432\u0430 \u0432\u043e\u0440\u043a\u0435\u0440\u0430 \u043e\u0431\u0430 \u043f\u0440\u043e\u0439\u0434\u0443\u0442 SELECT \u0434\u043e INSERT. \u0415\u0434\u0438\u043d\u0441\u0442\u0432\u0435\u043d\u043d\u043e\u0435 \u0447\u0442\u043e \u0430\u0442\u043e\u043c\u0430\u0440\u043d\u043e \u043f\u043e \u043e\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u0438\u044e  unique constraint. \u0421 \u0434\u0435\u043d\u044c\u0433\u0430\u043c\u0438 \u043d\u0435\u0442 «\u043f\u043e\u0447\u0442\u0438 \u043f\u0440\u0430\u0432\u0438\u043b\u044c\u043d\u043e». \u0421\u0445\u0435\u043c\u0430 \u0431\u0430\u0437\u044b \u0434\u0430\u043d\u043d\u044b\u0445  CREATE TABLE payment_events (    event_id     TEXT PRIMARY KEY,    user_id      INTEGER NOT NULL REFERENCES users(id),    amount       NUMERIC(38, 18) NOT NULL,    event_type   TEXT NOT NULL,    status       TEXT NOT NULL DEFAULT ‘pending’,    retry_count  INTEGER NOT NULL DEFAULT 0,    created_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),    updated_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),    CONSTRAINT valid_status CHECK (        status IN (‘pending’, ‘enqueued’, ‘processing’, ‘confirmed’, ‘failed’)    ));CREATE TABLE balance_events (    id              BIGSERIAL PRIMARY KEY,    user_id         INTEGER NOT NULL REFERENCES users(id),    amount          NUMERIC(38, 18) NOT NULL,    event_type      TEXT NOT NULL,    source_event_id TEXT NOT NULL,    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),    CONSTRAINT uq_balance_events_source UNIQUE (source_event_id, event_type));CREATE TABLE processed_events (    idempotency_key TEXT PRIMARY KEY,    outcome         TEXT NOT NULL DEFAULT ‘pending’,    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW());CREATE TABLE dead_letter_queue (    id         BIGSERIAL PRIMARY KEY,    event_id   TEXT NOT NULL,    event_type TEXT NOT NULL,    user_id    INTEGER NOT NULL,    amount     NUMERIC(38, 18) NOT NULL,    error      TEXT NOT NULL,    attempt    INTEGER NOT NULL DEFAULT 1,    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW());ALTER TABLE users    ADD COLUMN IF NOT EXISTS initial_balance NUMERIC(38, 18) NOT NULL DEFAULT 0,    ADD CONSTRAINT balance_non_negative CHECK (balance >= 0);CREATE INDEX idx_payment_events_pending    ON payment_events (updated_at, created_at)    WHERE status = ‘pending’;CREATE INDEX idx_payment_events_enqueued    ON payment_events (updated_at)    WHERE status = ‘enqueued’;CREATE INDEX idx_payment_events_processing    ON payment_events (updated_at)    WHERE status = ‘processing’;CREATE INDEX idx_balance_events_user_id    ON balance_events (user_id);CREATE INDEX idx_balance_events_created_at    ON balance_events (created_at DESC);CREATE INDEX idx_processed_events_created    ON processed_events (created_at);CREATE INDEX idx_processed_events_pending_stale    ON processed_events (created_at)    WHERE outcome = ‘pending’;CREATE INDEX idx_dlq_event_id    ON dead_letter_queue (event_id);CREATE INDEX idx_dlq_created_at    ON dead_letter_queue (created_at DESC);\u0412 \u0441\u0445\u0435\u043c\u0435 \u0442\u0440\u0438 \u043d\u0435\u043e\u0447\u0435\u0432\u0438\u0434\u043d\u044b\u0445 \u0440\u0435\u0448\u0435\u043d\u0438\u044f.NUMERIC(38, 18), \u043d\u0435 NUMERIC(20, 8). \u041a\u043e\u043b\u043e\u043d\u043a\u0430 amount \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f \u0432 ETH, \u043d\u0435 \u0432 wei. Webhook-\u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u043f\u0440\u0438\u0441\u044b\u043b\u0430\u0435\u0442 \u0443\u0436\u0435 \u0441\u043a\u043e\u043d\u0432\u0435\u0440\u0442\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0435 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435. \u0415\u0441\u043b\u0438 \u0432\u0430\u0448 \u043f\u0440\u043e\u0432\u0430\u0439\u0434\u0435\u0440 \u0432\u043e\u0437\u0432\u0440\u0430\u0449\u0430\u0435\u0442 wei, \u043a\u043e\u043d\u0432\u0435\u0440\u0442\u0438\u0440\u0443\u0439\u0442\u0435 \u043d\u0430 \u0432\u0445\u043e\u0434\u0435: amount_eth = Decimal(wei_str) \/ Decimal(10**18) \u0434\u043e \u043f\u0435\u0440\u0435\u0434\u0430\u0447\u0438 \u0432 _validate_amount  . ERC-20 \u0441\u0430\u043c\u0438 \u043e\u0431\u044a\u044f\u0432\u043b\u044f\u044e\u0442 decimals(): USDC\/USDT — 6, WBTC — 8, DAI\/WETH\/MKR — 18. ETH \u0432 wei \u0442\u043e\u0436\u0435 10^18. NUMERIC(20, 8) \u0432\u044b\u0434\u0435\u0440\u0436\u0438\u0442 USDC\/USDT, \u043d\u043e \u0444\u0438\u0437\u0438\u0447\u0435\u0441\u043a\u0438 \u043d\u0435 \u0432\u043c\u0435\u0449\u0430\u0435\u0442 18-decimal \u0442\u043e\u043a\u0435\u043d\u044b, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 \u0431\u0435\u0440\u0451\u043c worst case, NUMERIC(38, 18).initial_balance \u043d\u0443\u0436\u043d\u0430 \u0434\u043b\u044f \u0441\u0432\u0435\u0440\u043a\u0438. \u041f\u0440\u0438 \u043c\u0438\u0433\u0440\u0430\u0446\u0438\u0438 \u0437\u0430\u043f\u043e\u043b\u043d\u044f\u0435\u0442\u0435 \u0435\u0451 \u0442\u0435\u043a\u0443\u0449\u0438\u043c \u0431\u0430\u043b\u0430\u043d\u0441\u043e\u043c, UPDATE users SET initial_balance = balance WHERE <\u0443\u0441\u043b\u043e\u0432\u0438\u0435>. \u042d\u0442\u043e \u043e\u0437\u043d\u0430\u0447\u0430\u0435\u0442 \u0447\u0442\u043e balance_events \u043d\u0430\u0447\u0438\u043d\u0430\u044e\u0442 \u043d\u0430\u043f\u043e\u043b\u043d\u044f\u0442\u044c\u0441\u044f \u0441 \u043d\u0443\u043b\u044f, \u0438 hot_path_balance_check \u043a\u043e\u0440\u0440\u0435\u043a\u0442\u043d\u043e \u0441\u0447\u0438\u0442\u0430\u0435\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u0442\u0435\u0445 \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u0435\u0439, \u0443 \u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u0432\u0441\u0435 \u043e\u043f\u0435\u0440\u0430\u0446\u0438\u0438 \u043f\u0440\u043e\u0448\u043b\u0438 \u0447\u0435\u0440\u0435\u0437 balance_events. \u0414\u043b\u044f \u043d\u043e\u0432\u044b\u0445 \u0441\u0438\u0441\u0442\u0435\u043c initial_balance \u043e\u0441\u0442\u0430\u0451\u0442\u0441\u044f 0.\u041e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0435 \u0438\u043d\u0434\u0435\u043a\u0441\u044b \u0434\u043b\u044f pending\/enqueued\/processing \u0432\u043c\u0435\u0441\u0442\u043e \u043e\u0434\u043d\u043e\u0433\u043e status IN (…), \u0442\u0430\u043a \u043a\u0430\u043a poller’\u044b \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044e\u0442 \u0440\u0430\u0437\u043d\u044b\u0435 \u043f\u0430\u0442\u0442\u0435\u0440\u043d\u044b \u0434\u043e\u0441\u0442\u0443\u043f\u0430. idx_payment_events_pending, partial index \u0441 (updated_at, created_at) \u0434\u043b\u044f ORDER BY created_at \u0432 enqueue_pending_events, \u0438\u043d\u0430\u0447\u0435 \u043f\u043b\u0430\u043d\u0438\u0440\u043e\u0432\u0449\u0438\u043a \u0441\u043e\u0440\u0442\u0438\u0440\u0443\u0435\u0442 \u0431\u0435\u0437 \u0438\u043d\u0434\u0435\u043a\u0441\u0430.retry_count \u0432 payment_events \u0434\u043e\u0431\u0430\u0432\u043b\u0435\u043d \u0434\u043b\u044f \u043f\u0440\u0435\u0434\u043e\u0442\u0432\u0440\u0430\u0449\u0435\u043d\u0438\u044f \u0431\u0435\u0441\u043a\u043e\u043d\u0435\u0447\u043d\u043e\u0433\u043e \u0446\u0438\u043a\u043b\u0430 pending -> enqueued \u043f\u0440\u0438 durable outage Redis, \u043e\u0431 \u044d\u0442\u043e\u043c \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u0435\u0435 \u0432 \u0441\u0435\u043a\u0446\u0438\u0438 \u043f\u0440\u043e \u0434\u0435\u0433\u0440\u0430\u0434\u0430\u0446\u0438\u044e.\u041a\u0430\u043a \u044d\u0442\u043e \u043f\u043e\u0447\u0438\u043d\u0438\u043b\u043e\u0441\u044c  \u0418\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044fimport osimport uuidimport jsonimport hmacimport randomimport hashlibimport secretsimport threadingimport structlogimport psycopg2import psycopg2.extrasimport psycopg2.poolimport redis as redis_libfrom typing import Literal, Optionalimport refrom decimal import Decimal, InvalidOperationfrom contextvars import ContextVarfrom dataclasses import dataclass, fieldfrom datetime import datetime, timezone, timedeltafrom celery import shared_taskfrom celery.exceptions import Ignore, MaxRetriesExceededErrorlogger = structlog.get_logger()@dataclass(frozen=True)class Settings:    DATABASE_URL:   str    WEBHOOK_SECRET: str    ETH_RPC_URL:    str    ALERT_EMAIL:    str    REDIS_URL:      str = «redis:\/\/localhost:6379\/0»    @classmethod    def from_env(cls) -> «Settings»:        required = («DATABASE_URL», «WEBHOOK_SECRET», «ETH_RPC_URL», «ALERT_EMAIL»)        missing  = [k for k in required if not os.environ.get(k)]        if missing:            raise RuntimeError(f»Missing required env vars: {‘, ‘.join(missing)}»)        return cls(            DATABASE_URL   = os.environ[«DATABASE_URL»],            WEBHOOK_SECRET = os.environ[«WEBHOOK_SECRET»],            ETH_RPC_URL    = os.environ[«ETH_RPC_URL»],            ALERT_EMAIL    = os.environ[«ALERT_EMAIL»],            REDIS_URL      = os.environ.get(«REDIS_URL», «redis:\/\/localhost:6379\/0»),        )settings = Settings.from_env()_redis_client = redis_lib.Redis.from_url(    settings.REDIS_URL,    decode_responses=True,    socket_connect_timeout=2,    socket_timeout=2,    retry_on_timeout=False,)send_alert: rate-limited \u043e\u0431\u0451\u0440\u0442\u043a\u0430 \u043d\u0430\u0434 \u043b\u043e\u0433\u0433\u0435\u0440\u043e\u043c. \u0412 \u043f\u0440\u043e\u0434\u0435 \u0437\u0430\u043c\u0435\u043d\u044f\u0435\u0442\u0441\u044f \u043d\u0430 PagerDuty\/OpsGenie SDK. \u041e\u0434\u0438\u043d\u0430\u043a\u043e\u0432\u044b\u0435 alert_key \u0432\u043d\u0443\u0442\u0440\u0438 cooldown \u043e\u043a\u043d\u0430 \u043f\u043e\u0434\u0430\u0432\u043b\u044f\u044e\u0442\u0441\u044f. \u0415\u0441\u043b\u0438 \u043a\u043b\u044e\u0447 \u043d\u0435 \u0437\u0430\u0434\u0430\u043d, \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0430 \u0431\u0435\u0437 rate-limit, \u0434\u043b\u044f \u043e\u0434\u043d\u043e\u0440\u0430\u0437\u043e\u0432\u044b\u0445 \u043a\u0440\u0438\u0442\u0438\u0447\u043d\u044b\u0445 \u0430\u043b\u0435\u0440\u0442\u043e\u0432. \u041d\u0435 \u0431\u0440\u043e\u0441\u0430\u0435\u0442 \u0438\u0441\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u0439 \u043d\u0438\u043a\u043e\u0433\u0434\u0430._alert_last_sent \u0440\u0430\u0441\u0442\u0451\u0442 \u043f\u0440\u0438 \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0445 \u043a\u043b\u044e\u0447\u0430\u0445. \u0415\u0441\u043b\u0438 \u0433\u0435\u043d\u0435\u0440\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043a\u043b\u044e\u0447\u0438 per-event-id (\u0430 \u043c\u044b \u0442\u0430\u043a \u0438 \u0434\u0435\u043b\u0430\u0435\u043c \u0434\u043b\u044f orphan-\u0430\u043b\u0435\u0440\u0442\u043e\u0432), \u0437\u0430 \u043c\u0435\u0441\u044f\u0446 \u044d\u0442\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043c\u0438\u043b\u043b\u0438\u043e\u043d\u043e\u0432 \u0437\u0430\u043f\u0438\u0441\u0435\u0439. \u041f\u043e\u044d\u0442\u043e\u043c\u0443 \u043f\u0440\u0438 \u043f\u0435\u0440\u0435\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u0438 \u0441\u043d\u0430\u0447\u0430\u043b\u0430 \u0447\u0438\u0441\u0442\u0438\u043c \u0443\u0441\u0442\u0430\u0440\u0435\u0432\u0448\u0438\u0435 \u043a\u043b\u044e\u0447\u0438, \u0430 \u0435\u0441\u043b\u0438 \u043f\u043e\u0441\u043b\u0435 \u0447\u0438\u0441\u0442\u043a\u0438 \u043c\u0435\u0441\u0442\u043e \u0432\u0441\u0451 \u0440\u0430\u0432\u043d\u043e \u043d\u0435 \u043e\u0441\u0432\u043e\u0431\u043e\u0434\u0438\u043b\u043e\u0441\u044c, \u043f\u043e\u0434\u0430\u0432\u043b\u044f\u0435\u043c \u043d\u043e\u0432\u044b\u0435. \u041a\u043e\u0441\u0442\u044b\u043b\u044c, \u0434\u0430. \u041d\u043e \u0437\u0430 \u0432\u043e\u0441\u0435\u043c\u044c \u043c\u0435\u0441\u044f\u0446\u0435\u0432 \u043d\u0435 \u043e\u0442\u0432\u0430\u043b\u0438\u043b\u0441\u044f._alert_lock = threading.Lock()_alert_last_sent: dict = {}MAX_ALERT_KEYS = 1_000def send_alert(message: str, alert_key: Optional[str] = None,               cooldown_seconds: int = 300) -> None:    try:        if alert_key is None: …<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-477645","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/477645","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=477645"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/477645\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=477645"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=477645"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=477645"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}