7b9d895ead
동시 다상품 검색 점검 중 발견. 워커 3개(소유자 6)가 포트 2개/게이트웨이를 두고 경합하는
상황을 실제 코드로 돌리니, 임대를 못 받은 워커가 **남이 쥔 포트를 그대로 집어 같은 IP 로
동시에 요청**했다:
coupang-w1 사용=70002 임대=70002
coupang-w2 사용=70002 임대=None ← 같은 IP 를 둘이 사용
원인은 _port() 의 계산식 폴백이다. 장부 모드에서 acquire 가 None 을 줘도 시간창 계산으로
포트를 하나 골라 돌려줬다. 포트 장부가 존재하는 이유("워커 N개가 같은 IP 에 요청을 몰면 그 IP 가
빨리 탄다" — port_registry.py 도입 배경)를 정면으로 무너뜨리는 경로다. 게다가 하필 **풀이 마른
상태 = IP 가 가장 귀할 때** 발동해, 남은 IP 를 두 배 속도로 태우는 악순환을 만든다.
→ 장부 모드에선 임대한 포트만 쓴다(없으면 None). 못 받으면 AdapterError 로 실패하고 잡이
백오프 후 재시도한다 — 그 사이 쿨다운이 풀린다. 풀 고갈 자체는 proxy_ports_low 가 이미 운다.
→ playwright_proxy() 도 임대가 없으면 예외. 여기서 None 을 돌려주면 **프록시 없이** 브라우저가
떠 서버 공인 IP 로 크롤하게 되는데, 그 IP 가 타면 회전으로 복구할 수 없다.
동시성 점검 결과(포트 20개/게이트웨이, 워커 3개):
정상 24건 동시 성공 24 · 포트 중복 보유 0
풀 고갈 성공 4/6(2건은 정상적으로 실패) · **같은 IP 공유 0**
전면 차단 소각이 어댑터당 2개에서 멈춤(게이트웨이당 6/20) · 브레이커 6/6 트립
테스트 3건 추가(고갈 시 None 반환·남의 포트 미사용 / 임대 없는 playwright_proxy 예외 /
검색이 깔끔히 실패), 전체 256 passed.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
244 lines
14 KiB
Python
244 lines
14 KiB
Python
"""DECODO(구 Smartproxy) residential 프록시 제공자 — 쿠팡(Akamai) 전용.
|
|
|
|
Decodo residential 은 **포트 기반 sticky** 모델이다:
|
|
gate.decodo.com : 10001..N (각 포트 = 별도 sticky 세션, 대시보드에서 지속시간 지정: 예 10분)
|
|
username/password 는 고정. → **IP 회전 = 포트를 바꾸는 것.**
|
|
|
|
매 요청 IP 변경은 Akamai 가 '쿠키-IP 불일치'로 재챌린지하므로 금물.
|
|
→ 시간창(now // window)으로 포트를 고른다: 창 안에선 같은 포트=같은 IP, 창이 지나면 다음 포트=새 IP.
|
|
자격증명/엔드포인트는 config.local.toml [DecodoConfig] 에서 로드(시크릿).
|
|
"""
|
|
|
|
import time
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
|
|
from common.logger import LOG
|
|
from config.server_configs import decodo_config
|
|
|
|
|
|
class DecodoProxy:
|
|
def __init__(self, cfg=None, host: str | None = None, registry=None, owner: str = "", store=None):
|
|
cfg = cfg if cfg is not None else decodo_config
|
|
# host 를 넘기면 그 게이트웨이를 쓴다 — 국가 타깃(kr.decodo.com)용. 포트/자격증명은 동일.
|
|
self.host = host or cfg.host
|
|
# 포트 배정 방식 3가지(우선순위 순):
|
|
# store DB 장부(PortLeaseStore) — **프로세스 간 공유**. 여러 워커 프로세스가 한 계정을
|
|
# 나눠 쓸 때 유일하게 안전하다. 프로덕션 경로.
|
|
# registry 인메모리 중재자 — 단일 프로세스 전용(테스트·단독 실행)
|
|
# 둘 다 없으면 기존 계산식(시간창+오프셋)
|
|
self._store = store
|
|
self._registry = registry
|
|
self._owner = owner or "proxy"
|
|
self._leased: int | None = None
|
|
# DB 왕복은 비동기라 검색 루프(동기 호출부)에서 곧바로 못 한다. 회전·차단은 여기에 적어두고
|
|
# 다음 ensure_port(브라우저 재기동 직전, async)에서 한 번에 반영한다.
|
|
self._pending_release: tuple[int, float] | None = None # (port, rest_sec)
|
|
self._pending_burn: list[tuple[int, float]] = [] # [(port, cooldown_sec)]
|
|
self.username = cfg.username
|
|
self.password = cfg.password
|
|
self.port_start = cfg.port_start
|
|
self.port_end = cfg.port_end
|
|
self.session_minutes = cfg.session_minutes or 10
|
|
self._rotate_offset = 0 # 봇 감지 등으로 '즉시 회전'이 필요할 때 증가
|
|
# 불탄(차단 감지된) 포트 격리 시간([DecodoConfig].port_cooldown_sec). sticky 만료(session_minutes)
|
|
# 이상이어야 쿨다운 복귀 시 같은 포트라도 사실상 새 IP 가 배정된다. 0=자동 max(sticky, 30분).
|
|
self.cooldown_sec = getattr(cfg, "port_cooldown_sec", 0) or max(self.session_minutes * 60, 1800)
|
|
# 선제 회전으로 놓은 포트의 휴식 시간. 0=자동(sticky 수명) — 그 IP 의 세션이 만료돼
|
|
# 다음에 잡힐 땐 사실상 새 IP 가 배정된다. 차단 쿨다운(cooldown_sec)보다 훨씬 짧다.
|
|
self.rest_sec = getattr(cfg, "port_rest_sec", 0) or self.session_minutes * 60
|
|
self._burned: dict[int, float] = {} # port → 쿨다운 만료 시각(monotonic)
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return all([self.host, self.username, self.password, self.port_start, self.port_end])
|
|
|
|
def rotate(self, kind: str = "rotate"):
|
|
"""시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지·예산 도달 시 호출.
|
|
|
|
kind="budget"(차단 전 선제 회전)이면 놓는 포트에 **휴식**을 준다 — 방금 쓴 IP 를
|
|
다른 워커가 곧바로 집으면 요청률이 도로 올라가 예산의 의미가 사라진다.
|
|
차단으로 태우는 건 mark_burned 가 따로 처리한다(휴식보다 훨씬 긴 쿨다운).
|
|
"""
|
|
self._rotate_offset += 1
|
|
if self._leased is None:
|
|
return
|
|
rest = self.rest_sec if kind == "budget" else 0
|
|
if self._store is not None:
|
|
self._pending_release = (self._leased, rest) # 다음 ensure_port 에서 DB 반영
|
|
self._leased = None
|
|
elif self._registry is not None:
|
|
self._registry.release(self.host, self._leased, self._owner, rest_sec=rest)
|
|
self._leased = None
|
|
|
|
def seed_offset(self, k: int):
|
|
"""워커별 시작 포트 분산용 — 동시 워커가 같은 포트(=같은 IP)를 쓰지 않도록 시작점을 벌린다."""
|
|
self._rotate_offset = k
|
|
|
|
def mark_burned(self, port: int | None, cooldown_sec: float | None = None):
|
|
"""차단 감지된 포트를 쿨다운 격리 — _port() 가 만료 전까지 건너뛴다.
|
|
선제(예산) 회전된 포트는 부르지 않는다 — 불탄 게 아니므로 로테이션 복귀 시 재사용."""
|
|
if port is None:
|
|
return
|
|
cd = cooldown_sec if cooldown_sec is not None else self.cooldown_sec
|
|
if self._store is not None:
|
|
self._pending_burn.append((port, cd))
|
|
if self._leased == port:
|
|
self._leased = None
|
|
LOG.i(f"[port] {self.host}:{port} 차단 기록 예약 {int(cd)}s ({self._owner})")
|
|
return
|
|
if self._registry is not None:
|
|
# 전역 격리 — 태운 포트를 다른 소스/워커가 곧바로 집는 걸 막는다.
|
|
self._registry.burn(self.host, port, cd, owner=self._owner, reason="block")
|
|
if self._leased == port:
|
|
self._leased = None
|
|
return
|
|
self._burned[port] = time.monotonic() + cd
|
|
LOG.i(f"[proxy] 포트 {port} 쿨다운 {int(cd)}s — 활성 {self.available_ports()}/{self.port_end - self.port_start + 1}")
|
|
|
|
async def ensure_port(self) -> int | None:
|
|
"""다음 요청에 쓸 포트를 확정한다(브라우저 재기동 직전에 await).
|
|
|
|
DB 장부 모드에서 이 함수가 유일한 I/O 지점이다: 밀린 반납·차단을 먼저 flush 하고,
|
|
임대가 없으면 새로 잡는다. 임대가 살아 있으면 연장(renew)해 sticky 수명 동안 붙잡는다
|
|
— 연장에 실패하면 남이 회수해 간 것이므로 새 포트를 잡는다.
|
|
"""
|
|
if self._store is None:
|
|
return self._port() if self.enabled else None
|
|
|
|
if self._pending_release is not None: # 선제 회전/일반 반납
|
|
port, rest = self._pending_release
|
|
self._pending_release = None
|
|
await self._store.release(self.host, port, self._owner, rest_sec=rest)
|
|
while self._pending_burn: # 차단 격리(전역)
|
|
port, cd = self._pending_burn.pop(0)
|
|
await self._store.burn(self.host, port, cd, owner=self._owner, reason="block")
|
|
|
|
lease_sec = self.session_minutes * 60
|
|
if self._leased is not None and not await self._store.renew(self.host, self._leased, self._owner, lease_sec):
|
|
self._leased = None # 만료·회수됨 → 새로 잡는다
|
|
if self._leased is None:
|
|
self._leased = await self._store.acquire(self.host, self._owner, lease_sec)
|
|
if self._leased is None:
|
|
LOG.w(f"[port] {self.host} 가용 포트 없음 — 전부 임대/휴식/쿨다운 중({self._owner})")
|
|
return self._leased
|
|
|
|
async def release(self):
|
|
"""쥐고 있는 임대를 **즉시 반납**한다(프로세스 종료 시).
|
|
|
|
이걸 안 하면 워커를 재시작해도 그 포트는 leased_until(sticky 수명, 기본 10분)까지 남의
|
|
것으로 남아 아무도 못 쓴다 — 재시작이 잦을수록 가용 풀이 계속 줄어든다.
|
|
**유휴 정리에서는 부르지 않는다**: 브라우저만 닫고 같은 IP 로 돌아와야 웜 쿠키
|
|
(cf_clearance 등)가 살고 요청 예산도 이어진다.
|
|
"""
|
|
await self.flush()
|
|
if self._leased is None:
|
|
return
|
|
port, self._leased = self._leased, None
|
|
if self._store is not None:
|
|
await self._store.release(self.host, port, self._owner)
|
|
elif self._registry is not None:
|
|
self._registry.release(self.host, port, self._owner)
|
|
|
|
async def flush(self):
|
|
"""밀린 반납·차단만 반영(검색 종료·셧다운 시). 포트를 새로 잡지는 않는다."""
|
|
if self._store is None:
|
|
return
|
|
if self._pending_release is not None:
|
|
port, rest = self._pending_release
|
|
self._pending_release = None
|
|
await self._store.release(self.host, port, self._owner, rest_sec=rest)
|
|
while self._pending_burn:
|
|
port, cd = self._pending_burn.pop(0)
|
|
await self._store.burn(self.host, port, cd, owner=self._owner, reason="block")
|
|
|
|
def available_ports(self) -> int:
|
|
"""쿨다운 중이 아닌 포트 수(관측용) — **인메모리 모드 전용**.
|
|
|
|
DB 장부(store) 모드에서는 이 값을 믿으면 안 된다: 차단은 _pending_burn 으로만 쌓였다가
|
|
DB 로 나가므로 로컬 _burned 는 늘 비어 있고, 다른 프로세스가 태운 포트도 보이지 않아
|
|
**항상 풀 크기(전량 가용)를 돌려준다**. 장부 모드의 실제 가용량은 비동기 조회라
|
|
여기서 낼 수 없다 — PortLeaseStore.snapshot() 을 쓴다(worker_main._port_pool_status).
|
|
"""
|
|
if self._registry is not None:
|
|
return self._registry.available(self.host)
|
|
self._prune_burned()
|
|
return (self.port_end - self.port_start + 1) - len(self._burned)
|
|
|
|
def _prune_burned(self):
|
|
now = time.monotonic()
|
|
self._burned = {p: t for p, t in self._burned.items() if t > now}
|
|
|
|
def _port(self) -> int | None:
|
|
"""시간창 + 수동 오프셋 기반 포트 선택. 창 안에선 동일 IP, rotate()나 창 변화 시 다음 IP.
|
|
쿨다운 중인 포트는 건너뛰고, 전 포트가 쿨다운이면 만료가 가장 임박한 포트를 쓴다(가용성 우선)."""
|
|
if self._store is not None:
|
|
# 장부 모드에선 **임대한 포트만** 쓴다(없으면 None). 계산식 폴백을 두면 안 된다 —
|
|
# 임대에 실패한 워커가 남이 쥔 포트를 그대로 집어 **같은 IP 에 요청이 겹친다**
|
|
# (2026-08-06 실측: 포트 2개·소유자 3인 상황에서 w1 과 w2 가 같은 포트로 동시 요청).
|
|
# 하필 풀이 마른 상태 = IP 가 가장 귀할 때 벌어져, 남은 IP 를 두 배 속도로 태운다.
|
|
# 이 장부가 막으려던 바로 그 문제이므로, 못 잡았으면 쓰지 않고 실패하는 게 맞다.
|
|
return self._leased
|
|
if self._registry is not None:
|
|
lease_sec = self.session_minutes * 60
|
|
if self._leased is not None and self._registry.lease_expired(self.host, self._leased, self._owner, lease_sec):
|
|
self._registry.release(self.host, self._leased, self._owner) # sticky 수명 만료 → 새 IP
|
|
self._leased = None
|
|
if self._leased is None:
|
|
# 시작 오프셋은 기존과 동일한 의미(워커별 분산). 실제 중복 방지는 registry 가 보장한다.
|
|
bucket = int(time.time() // (self.session_minutes * 60))
|
|
start = (bucket + self._rotate_offset) % (self.port_end - self.port_start + 1)
|
|
self._leased = self._registry.acquire(self.host, self._owner, start_at=start, lease_sec=lease_sec)
|
|
return self._leased
|
|
|
|
n = self.port_end - self.port_start + 1
|
|
bucket = int(time.time() // (self.session_minutes * 60))
|
|
self._prune_burned()
|
|
for k in range(n):
|
|
port = self.port_start + ((bucket + self._rotate_offset + k) % n)
|
|
if port not in self._burned:
|
|
return port
|
|
return min(self._burned, key=self._burned.get)
|
|
|
|
@property
|
|
def current_port(self):
|
|
return self._port() if self.enabled else None
|
|
|
|
def playwright_proxy(self) -> dict | None:
|
|
"""Playwright launch(proxy=...) 용 설정. 비활성 시 None(프록시 미사용)."""
|
|
if not self.enabled:
|
|
return None
|
|
port = self._port()
|
|
if port is None:
|
|
# 장부 모드에서 임대를 못 받은 상태. 여기서 None 을 돌려주면 **프록시 없이** 브라우저가
|
|
# 떠 서버 공인 IP 로 크롤하게 된다(그 IP가 타면 회전으로 복구 불가). 명시적으로 실패한다.
|
|
raise RuntimeError(f"{self.host} 가용 프록시 포트 없음 — 임대 없이 크롤할 수 없습니다")
|
|
return {
|
|
"server": f"http://{self.host}:{port}",
|
|
"username": self.username,
|
|
"password": self.password,
|
|
}
|
|
|
|
def _proxy_url(self, port: int) -> str:
|
|
return f"http://{quote(self.username)}:{quote(self.password)}@{self.host}:{port}"
|
|
|
|
async def healthcheck(self, timeout: float = 6.0) -> tuple[str | None, int | None]:
|
|
"""시작 프리플라이트: 현재 포트로 egress IP 확인, 실패하면 회전하며 살아있는 포트를 찾는다.
|
|
반환: (egress_ip, port) 성공 / (None, None) 전 포트 실패. residential IP 는 실행 중에도
|
|
죽으므로 이건 '빠른 실패+가시성'용이고, 실제 회복은 런타임 IP 회전이 담당한다."""
|
|
if not self.enabled:
|
|
return None, None
|
|
n = self.port_end - self.port_start + 1
|
|
for _ in range(n):
|
|
port = self._port()
|
|
try:
|
|
async with httpx.AsyncClient(proxy=self._proxy_url(port), timeout=timeout) as c:
|
|
r = await c.get("https://ip.decodo.com/ip")
|
|
if r.status_code == 200:
|
|
return r.text.strip(), port
|
|
except Exception as ex:
|
|
LOG.d(f"[proxy] 포트 {port} 헬스체크 실패: {type(ex).__name__} → 회전")
|
|
self.rotate()
|
|
return None, None
|