589 lines
27 KiB
Python
589 lines
27 KiB
Python
"""Background DB Writer Worker
|
|
|
|
ARCHITECTURE (CLAUDE.md 3.2):
|
|
- 실시간/콜백 스레드에서는 절대 DB 세션을 직접 열지 않는다.
|
|
- 모든 DB 쓰기 작업은 이 모듈의 db_task_queue 로 위임한다.
|
|
- 워커가 큐를 소비하며 단일 세션에서 순차 처리 → 락 경합 제거.
|
|
|
|
Supported task types:
|
|
- insert_trade_history : dict → TradeHistory INSERT
|
|
- update_trade_history : dict → TradeHistory UPDATE by id (체결 확인 후 슬리피지/수량 반영)
|
|
- update_trade_history_by_order_id : dict → TradeHistory UPDATE by order_id (KIS 주문번호 기준)
|
|
- upsert_stock_analysis : dict → StockAnalysis single UPSERT
|
|
- bulk_upsert_stock_analysis : list → StockAnalysis batch UPSERT
|
|
- upsert_stock_ban : dict → StockBan UPSERT (loss_tracker)
|
|
- insert_news : dict → News INSERT (뉴스 감성 분석 결과 저장)
|
|
- log_mcp_deny : dict → MCP 거부 이벤트 로그 (DB 저장 없이 로그만)
|
|
- upsert_portfolio : dict → Portfolio INSERT/UPDATE (strategy_type 포함)
|
|
- upsert_active_trade : dict → ActiveTrade UPSERT (strategy_name 포함)
|
|
- close_active_trade : dict → ActiveTrade 상태를 'closed'로 변경
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.models.models import TradeHistory
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# DLQ (Dead Letter Queue) — 3회 재시도 실패 시 파일로 백업
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
DLQ_DIR = Path.home() / ".cache" / "auto-stock" / "dlq"
|
|
|
|
|
|
async def _save_to_dlq(task: dict, error: str) -> None:
|
|
"""3회 재시도 실패한 task를 DLQ 파일로 저장.
|
|
|
|
저장 위치: ~/.cache/auto-stock/dlq/{timestamp}_{task_type}_{uuid}.json
|
|
파일 I/O는 asyncio.to_thread로 감싸 이벤트 루프 블로킹을 방지한다.
|
|
"""
|
|
try:
|
|
task_type = task.get("type", "unknown")
|
|
filename = (
|
|
f"{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
|
f"_{task_type}"
|
|
f"_{uuid.uuid4().hex[:8]}.json"
|
|
)
|
|
dlq_entry = {
|
|
"task": task,
|
|
"error": str(error),
|
|
"failed_at": datetime.now().isoformat(),
|
|
"retry_count": task.get("_retry_count", 0),
|
|
}
|
|
|
|
def _write():
|
|
DLQ_DIR.mkdir(parents=True, exist_ok=True)
|
|
(DLQ_DIR / filename).write_text(
|
|
json.dumps(dlq_entry, default=str, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
await asyncio.to_thread(_write)
|
|
logger.critical(
|
|
"[DLQ] Task saved to DLQ: %s | type=%s | error=%s",
|
|
filename, task_type, error,
|
|
)
|
|
except Exception as e:
|
|
logger.critical(
|
|
"[DLQ] Failed to save to DLQ: %s | task_type=%s",
|
|
e, task.get("type", "unknown"),
|
|
)
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 전역 큐 — 앱 어디서나 import 하여 put() 호출
|
|
# [이슈8 수정] asyncio.Queue()를 모듈 로드 시점에 생성하면 이벤트 루프가 없는
|
|
# 환경(pytest, 스크립트 임포트 등)에서 RuntimeError가 발생할 수 있음.
|
|
# _db_task_queue를 None으로 두고, 첫 접근 시 lazy 초기화하는 패턴으로 변경.
|
|
# 외부 코드는 기존처럼 db_task_queue 이름으로 접근 가능 (get_db_task_queue() 내부 참조).
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
_db_task_queue: asyncio.Queue | None = None
|
|
|
|
|
|
def get_db_task_queue() -> asyncio.Queue:
|
|
"""이벤트 루프 기동 후 최초 호출 시 Queue를 생성(lazy init)하여 반환."""
|
|
global _db_task_queue
|
|
if _db_task_queue is None:
|
|
_db_task_queue = asyncio.Queue()
|
|
return _db_task_queue
|
|
|
|
|
|
# 하위 호환: 기존 코드가 `db_task_queue.put()` 형태로 직접 참조하는 경우를 위해
|
|
# 프록시 객체 대신 get_db_task_queue()를 통해 접근하도록 모듈 수준 alias 제공.
|
|
# 단, 모듈 임포트 시점에 이벤트 루프가 없어도 오류가 발생하지 않음.
|
|
class _QueueProxy:
|
|
"""asyncio.Queue를 lazy init으로 감싸는 투명 프록시."""
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(get_db_task_queue(), name)
|
|
|
|
async def put(self, item):
|
|
await get_db_task_queue().put(item)
|
|
|
|
async def get(self):
|
|
return await get_db_task_queue().get()
|
|
|
|
def task_done(self):
|
|
get_db_task_queue().task_done()
|
|
|
|
def qsize(self):
|
|
return get_db_task_queue().qsize()
|
|
|
|
def empty(self):
|
|
return get_db_task_queue().empty()
|
|
|
|
|
|
db_task_queue: asyncio.Queue = _QueueProxy() # type: ignore[assignment]
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 편의 함수 — 호출 측에서 dict 구성 없이 바로 사용
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
async def enqueue_trade_history(data: dict) -> None:
|
|
"""매매일지 INSERT 를 큐에 등록."""
|
|
await db_task_queue.put({"type": "insert_trade_history", "data": data})
|
|
|
|
|
|
async def enqueue_update_trade_history(trade_id: int, updates: dict) -> None:
|
|
"""기존 TradeHistory 레코드 UPDATE 를 큐에 등록.
|
|
|
|
주로 체결 확인(verify_and_sync) 후 executed_price/quantity/status/slippage 반영 시 사용.
|
|
|
|
Args:
|
|
trade_id: 업데이트할 TradeHistory.id
|
|
updates: 업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
|
|
"""
|
|
await db_task_queue.put({
|
|
"type": "update_trade_history",
|
|
"data": {"id": trade_id, **updates},
|
|
})
|
|
|
|
|
|
async def enqueue_upsert_analysis(row: dict) -> None:
|
|
"""단건 StockAnalysis UPSERT 를 큐에 등록."""
|
|
await db_task_queue.put({"type": "upsert_stock_analysis", "data": row})
|
|
|
|
|
|
async def enqueue_bulk_upsert_analysis(rows: list[dict]) -> None:
|
|
"""다건 StockAnalysis batch UPSERT 를 큐에 등록."""
|
|
if rows:
|
|
await db_task_queue.put({"type": "bulk_upsert_stock_analysis", "data": rows})
|
|
|
|
|
|
async def enqueue_db_write(task: dict) -> None:
|
|
"""범용 DB 쓰기 태스크를 큐에 등록.
|
|
|
|
task 는 반드시 'type' 키를 포함해야 하며, 워커가 지원하는 task type 중 하나여야 한다.
|
|
예: {"type": "log_mcp_deny", "reason": "...", "trade_plan": {...}}
|
|
"""
|
|
await db_task_queue.put(task)
|
|
|
|
|
|
async def enqueue_trade_journal_raw(data: dict) -> None:
|
|
"""TradeJournal INSERT 를 큐에 등록 (순수 dict 전달용)."""
|
|
await db_task_queue.put({"type": "insert_trade_journal", "data": data})
|
|
|
|
|
|
async def enqueue_stock_ban_upsert(stock_code: str, consecutive_losses: int, ban_until) -> None:
|
|
"""StockBan UPSERT 를 큐에 등록 (loss_tracker에서 호출)."""
|
|
await db_task_queue.put({
|
|
"type": "upsert_stock_ban",
|
|
"data": {
|
|
"stock_code": stock_code,
|
|
"consecutive_losses": consecutive_losses,
|
|
"ban_until": ban_until,
|
|
}
|
|
})
|
|
|
|
|
|
async def enqueue_portfolio_upsert(data: dict) -> None:
|
|
"""Portfolio INSERT/UPDATE 를 큐에 등록.
|
|
|
|
data 필수 키: stock_code, quantity, average_price
|
|
data 선택 키: stock_name, current_price, profit_loss, profit_loss_rate,
|
|
market, strategy_type ('swing' | 'daytrading')
|
|
"""
|
|
await db_task_queue.put({"type": "upsert_portfolio", "data": data})
|
|
|
|
|
|
async def enqueue_active_trade_upsert(data: dict) -> None:
|
|
"""ActiveTrade UPSERT 를 큐에 등록.
|
|
|
|
data 필수 키: stock_code, quantity, entry_price
|
|
data 선택 키: strategy_name ('swing' | 'day_trade'), stop_loss,
|
|
take_profit, entry_reason, highest_since_entry
|
|
"""
|
|
await db_task_queue.put({"type": "upsert_active_trade", "data": data})
|
|
|
|
|
|
async def enqueue_close_active_trade(stock_code: str, strategy_name: str | None = None) -> None:
|
|
"""ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록."""
|
|
await db_task_queue.put({
|
|
"type": "close_active_trade",
|
|
"data": {"stock_code": stock_code, "strategy_name": strategy_name},
|
|
})
|
|
|
|
|
|
async def enqueue_update_trade_history_by_order_id(order_id: str, updates: dict) -> None:
|
|
"""KIS 주문번호(order_id)로 TradeHistory UPDATE 를 큐에 등록.
|
|
|
|
trade_id 대신 order_id 로 레코드를 조회하여 업데이트한다.
|
|
enqueue_trade_history() 로 INSERT 한 레코드의 id 를 모르는 경우 사용.
|
|
|
|
Args:
|
|
order_id: KIS 주문번호 (TradeHistory.order_id)
|
|
updates: 업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
|
|
"""
|
|
await db_task_queue.put({
|
|
"type": "update_trade_history_by_order_id",
|
|
"data": {"order_id": order_id, **updates},
|
|
})
|
|
|
|
|
|
async def enqueue_trade_journal(trade, content: str) -> None:
|
|
"""TradeJournal INSERT 를 큐에 등록 (main.py post_market_scheduler 용).
|
|
|
|
CLAUDE.md 3.2: 스케줄러(실시간 계층)에서 직접 DB 커밋 금지 → 큐 경유.
|
|
"""
|
|
from datetime import datetime
|
|
from app.services.trade_analyzer import _to_float, _parse_strategy, _parse_exit_type
|
|
|
|
pl_rate = _to_float(trade.profit_loss_rate)
|
|
strategy = _parse_strategy(trade.entry_reason)
|
|
exit_type = _parse_exit_type(trade.exit_reason)
|
|
mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative")
|
|
tags = [strategy, exit_type]
|
|
if abs(pl_rate) > 5.0:
|
|
tags.append("significant")
|
|
|
|
await db_task_queue.put({
|
|
"type": "insert_trade_journal",
|
|
"data": {
|
|
"trade_id": trade.id,
|
|
"stock_code": trade.stock_code,
|
|
"stock_name": trade.stock_name,
|
|
"journal_date": trade.executed_at or trade.created_at or datetime.utcnow(),
|
|
"content": content,
|
|
"strategy_name": strategy,
|
|
"mood": mood,
|
|
"tags": tags,
|
|
},
|
|
})
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 워커 본체
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
async def db_writer_worker() -> None:
|
|
"""lifespan 에서 asyncio.create_task() 로 실행되는 영구 백그라운드 워커.
|
|
|
|
큐에서 task 를 꺼내 단일 AsyncSession 안에서 처리한 뒤
|
|
commit / rollback 을 자동 관리한다.
|
|
"""
|
|
logger.info("🛠️ Background DB Writer Worker started")
|
|
queue = get_db_task_queue() # 워커 진입 시 큐 인스턴스 고정
|
|
while True:
|
|
task = await queue.get()
|
|
try:
|
|
task_type = task.get("type")
|
|
data = task.get("data")
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
# ── 1. 매매일지 INSERT ─────────────────────────────────────
|
|
if task_type == "insert_trade_history":
|
|
session.add(TradeHistory(**data))
|
|
await session.commit()
|
|
logger.info(
|
|
"✅ [Worker] TradeHistory saved: %s",
|
|
data.get("stock_code", "?"),
|
|
)
|
|
|
|
# ── 2. TradeHistory UPDATE (체결 확인 후 슬리피지/수량 반영) ──
|
|
elif task_type == "update_trade_history":
|
|
trade_id = data.get("id")
|
|
if trade_id:
|
|
row = await session.execute(
|
|
select(TradeHistory).where(TradeHistory.id == trade_id)
|
|
)
|
|
trade = row.scalar_one_or_none()
|
|
if trade:
|
|
for key, val in data.items():
|
|
if key != "id" and val is not None and hasattr(trade, key):
|
|
setattr(trade, key, val)
|
|
await session.commit()
|
|
logger.info(
|
|
"✅ [Worker] TradeHistory updated: id=%s status=%s",
|
|
trade_id,
|
|
data.get("status", "?"),
|
|
)
|
|
|
|
# ── 2b. TradeHistory UPDATE by order_id (KIS 주문번호 기준) ──
|
|
elif task_type == "update_trade_history_by_order_id":
|
|
order_id = data.get("order_id")
|
|
if order_id:
|
|
row = await session.execute(
|
|
select(TradeHistory).where(TradeHistory.order_id == order_id)
|
|
)
|
|
trade = row.scalar_one_or_none()
|
|
if trade:
|
|
for key, val in data.items():
|
|
if key != "order_id" and val is not None and hasattr(trade, key):
|
|
setattr(trade, key, val)
|
|
await session.commit()
|
|
logger.info(
|
|
"✅ [Worker] TradeHistory updated by order_id=%s status=%s",
|
|
order_id,
|
|
data.get("status", "?"),
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"⚠️ [Worker] TradeHistory not found for order_id=%s",
|
|
order_id,
|
|
)
|
|
|
|
# ── 3. 단건 StockAnalysis UPSERT ──────────────────────────
|
|
elif task_type == "upsert_stock_analysis":
|
|
from app.services.analysis_db_service import _execute_upsert
|
|
await _execute_upsert(session, [data])
|
|
await session.commit()
|
|
logger.debug(
|
|
"✅ [Worker] StockAnalysis upsert: %s",
|
|
data.get("stock_code", "?"),
|
|
)
|
|
|
|
# ── 3. 다건 StockAnalysis batch UPSERT ────────────────────
|
|
elif task_type == "bulk_upsert_stock_analysis":
|
|
from app.services.analysis_db_service import _execute_upsert
|
|
await _execute_upsert(session, data) # data = list[dict]
|
|
await session.commit()
|
|
logger.info(
|
|
"✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data)
|
|
)
|
|
|
|
# ── 4. StockBan UPSERT ────────────────────────────────────────
|
|
elif task_type == "upsert_stock_ban":
|
|
from app.models.models import StockBan
|
|
|
|
stock_code = data.get("stock_code")
|
|
consecutive_losses = data.get("consecutive_losses")
|
|
ban_until = data.get("ban_until")
|
|
|
|
result = await session.execute(select(StockBan).where(StockBan.stock_code == stock_code))
|
|
record = result.scalar_one_or_none()
|
|
|
|
if record is None:
|
|
record = StockBan(
|
|
stock_code=stock_code,
|
|
consecutive_losses=consecutive_losses,
|
|
ban_until=ban_until,
|
|
)
|
|
session.add(record)
|
|
else:
|
|
record.consecutive_losses = consecutive_losses
|
|
record.ban_until = ban_until
|
|
|
|
await session.commit()
|
|
logger.debug(
|
|
"✅ [Worker] StockBan upsert: %s (losses=%d, ban_until=%s)",
|
|
stock_code,
|
|
consecutive_losses,
|
|
ban_until.isoformat() if ban_until else "None",
|
|
)
|
|
|
|
# ── 4b. News INSERT (중복 URL 무시) ────────────────────────────
|
|
elif task_type == "insert_news":
|
|
from app.models.models import News
|
|
|
|
news_url = data.get("url")
|
|
is_dup = False
|
|
# URL 기반 중복 방지 (동일 URL은 skip)
|
|
if news_url:
|
|
dup_result = await session.execute(
|
|
select(News.id).where(News.url == news_url).limit(1)
|
|
)
|
|
if dup_result.scalar_one_or_none() is not None:
|
|
is_dup = True
|
|
logger.debug(
|
|
"⏭️ [Worker] News skip (dup URL): %s",
|
|
(data.get("title") or "?")[:40],
|
|
)
|
|
|
|
if not is_dup:
|
|
session.add(News(**data))
|
|
await session.commit()
|
|
logger.debug(
|
|
"✅ [Worker] News saved: %s",
|
|
(data.get("title") or "?")[:40],
|
|
)
|
|
|
|
# ── 5. MCP 거부 이벤트 로그 ───────────────────────────────────
|
|
elif task_type == "log_mcp_deny":
|
|
# DB 저장 없이 로그만 기록 (필요 시 별도 테이블 확장 가능)
|
|
logger.warning(
|
|
"🛑 [Worker] MCP DENY logged: reason=%s, trade_plan=%s",
|
|
task.get("reason", "unknown"),
|
|
task.get("trade_plan"),
|
|
)
|
|
|
|
# ── 6. Portfolio UPSERT ────────────────────────────────────────
|
|
elif task_type == "upsert_portfolio":
|
|
from app.models.models import Portfolio
|
|
|
|
stock_code = data.get("stock_code")
|
|
result = await session.execute(
|
|
select(Portfolio).where(Portfolio.stock_code == stock_code)
|
|
)
|
|
record = result.scalar_one_or_none()
|
|
|
|
if record is None:
|
|
record = Portfolio(**data)
|
|
session.add(record)
|
|
else:
|
|
for key, val in data.items():
|
|
if key != "stock_code" and val is not None:
|
|
setattr(record, key, val)
|
|
|
|
await session.commit()
|
|
logger.debug(
|
|
"✅ [Worker] Portfolio upsert: %s (strategy=%s)",
|
|
stock_code,
|
|
data.get("strategy_type", "swing"),
|
|
)
|
|
|
|
# ── 7. ActiveTrade UPSERT ──────────────────────────────────────
|
|
elif task_type == "upsert_active_trade":
|
|
from app.models.models import ActiveTrade
|
|
|
|
stock_code = data.get("stock_code")
|
|
strategy_name = data.get("strategy_name")
|
|
|
|
query = select(ActiveTrade).where(
|
|
ActiveTrade.stock_code == stock_code,
|
|
ActiveTrade.status == "active",
|
|
)
|
|
if strategy_name:
|
|
query = query.where(ActiveTrade.strategy_name == strategy_name)
|
|
|
|
result = await session.execute(query)
|
|
record = result.scalar_one_or_none()
|
|
|
|
if record is None:
|
|
# remaining_pct_multiply는 신규 생성 시 무시
|
|
create_data = {k: v for k, v in data.items() if k != "remaining_pct_multiply"}
|
|
record = ActiveTrade(**create_data)
|
|
session.add(record)
|
|
else:
|
|
# remaining_pct_multiply: 기존 remaining_pct에 비율을 곱함
|
|
# (MDD 축소 시 기존 분할매도 이력 반영)
|
|
multiply_ratio = data.pop("remaining_pct_multiply", None)
|
|
if multiply_ratio is not None:
|
|
current_remaining = getattr(record, "remaining_pct", 1.0) or 1.0
|
|
record.remaining_pct = current_remaining * float(multiply_ratio)
|
|
for key, val in data.items():
|
|
if key != "stock_code" and val is not None:
|
|
setattr(record, key, val)
|
|
|
|
await session.commit()
|
|
logger.debug(
|
|
"✅ [Worker] ActiveTrade upsert: %s (strategy=%s)",
|
|
stock_code,
|
|
strategy_name,
|
|
)
|
|
|
|
# ── 8. ActiveTrade close ───────────────────────────────────────
|
|
elif task_type == "close_active_trade":
|
|
from app.models.models import ActiveTrade
|
|
from datetime import datetime as _dt
|
|
|
|
stock_code = data.get("stock_code")
|
|
strategy_name = data.get("strategy_name")
|
|
|
|
query = select(ActiveTrade).where(
|
|
ActiveTrade.stock_code == stock_code,
|
|
ActiveTrade.status == "active",
|
|
)
|
|
if strategy_name:
|
|
query = query.where(ActiveTrade.strategy_name == strategy_name)
|
|
|
|
result = await session.execute(query)
|
|
record = result.scalar_one_or_none()
|
|
|
|
if record:
|
|
record.status = "closed"
|
|
record.updated_at = _dt.utcnow()
|
|
await session.commit()
|
|
logger.debug(
|
|
"✅ [Worker] ActiveTrade closed: %s (strategy=%s)",
|
|
stock_code,
|
|
strategy_name,
|
|
)
|
|
|
|
# ── 9. TradeJournal INSERT ────────────────────────────────────
|
|
elif task_type == "insert_trade_journal":
|
|
from app.models.models import TradeJournal
|
|
|
|
session.add(TradeJournal(**data))
|
|
await session.commit()
|
|
logger.debug(
|
|
"✅ [Worker] TradeJournal saved: %s",
|
|
data.get("stock_code", "?"),
|
|
)
|
|
|
|
else:
|
|
logger.warning("⚠️ [Worker] Unknown task type: %s", task_type)
|
|
|
|
except asyncio.CancelledError:
|
|
# 큐 drain: 종료 전 남은 태스크 처리하여 데이터 손실 방지
|
|
remaining = 0
|
|
while not queue.empty():
|
|
try:
|
|
drain_task = queue.get_nowait()
|
|
drain_type = drain_task.get("type", "unknown")
|
|
drain_data = drain_task.get("data")
|
|
async with AsyncSessionLocal() as drain_session:
|
|
if drain_type == "insert_trade_history":
|
|
drain_session.add(TradeHistory(**drain_data))
|
|
await drain_session.commit()
|
|
elif drain_type in (
|
|
"update_trade_history",
|
|
"update_trade_history_by_order_id",
|
|
"upsert_stock_analysis",
|
|
"bulk_upsert_stock_analysis",
|
|
"upsert_stock_ban",
|
|
"upsert_portfolio",
|
|
"upsert_active_trade",
|
|
"close_active_trade",
|
|
"insert_trade_journal",
|
|
):
|
|
# 단순 재큐잉 대신, 중요 태스크만 커밋 시도
|
|
# insert_trade_history 외에는 복잡한 로직이므로 DLQ 저장
|
|
await _save_to_dlq(drain_task, "shutdown_drain")
|
|
else:
|
|
await _save_to_dlq(drain_task, "shutdown_drain_unknown_type")
|
|
remaining += 1
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
except Exception as drain_err:
|
|
logger.error(f"[DBWorker] Drain 실패: {drain_err}")
|
|
try:
|
|
await _save_to_dlq(drain_task, str(drain_err))
|
|
except Exception:
|
|
pass
|
|
remaining += 1
|
|
finally:
|
|
queue.task_done()
|
|
if remaining:
|
|
logger.info(f"[DBWorker] 종료 시 {remaining}건 drain 완료")
|
|
logger.info("⏹️ DB Writer Worker stopped")
|
|
break
|
|
except Exception as exc:
|
|
retry_count = task.get("_retry_count", 0)
|
|
max_retries = 3
|
|
task_type = task.get("type", "unknown")
|
|
|
|
if retry_count < max_retries:
|
|
task["_retry_count"] = retry_count + 1
|
|
delay = min(2 ** retry_count, 8) # 1, 2, 4초 지수 백오프
|
|
logger.warning(
|
|
"⚠️ [Worker] DB task retry %d/%d (type=%s): %s — %ds 후 재시도",
|
|
retry_count + 1, max_retries, task_type, exc, delay,
|
|
)
|
|
await asyncio.sleep(delay)
|
|
await queue.put(task) # 큐에 재등록
|
|
else:
|
|
logger.error(
|
|
"❌ [Worker] DB task DROPPED after %d retries (type=%s): %s",
|
|
max_retries, task_type, exc, exc_info=True,
|
|
)
|
|
# DLQ: 최종 실패 task를 파일로 백업하여 데이터 손실 방지
|
|
await _save_to_dlq(task, str(exc))
|
|
finally:
|
|
queue.task_done()
|