feature: воркер удаляет профили, не использованные PROFILE_TTL_DAYS дней (#20)

Отметка последнего использования — mtime файла в профиле, ставится при открытии и
закрытии окна. Очистка при старте воркера и раз в час; профили с открытым окном
(на этом воркере или по ключу сессии в Redis) не трогаются.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
protokeyandClaude Opus 5.5 committed 2026-10-13 20:58:56 +04:00
1 parent 1075267d27
commit 4c8394eb4c
6 files changed
+97 -8

No files matched your search

+1 -2
View File
@@ -13,7 +13,6 @@ Not implemented yet, though planned for v1:
- API-key auth (`X-API-Key` is accepted but **not checked**) — [#4](https://redmine.sokolab.xyz/issues/4) - API-key auth (`X-API-Key` is accepted but **not checked**) — [#4](https://redmine.sokolab.xyz/issues/4)
- persisting request outcomes (success rate) — [#17](https://redmine.sokolab.xyz/issues/17) - persisting request outcomes (success rate) — [#17](https://redmine.sokolab.xyz/issues/17)
- cleanup of old browser profiles — [#20](https://redmine.sokolab.xyz/issues/20)
## Architecture ## Architecture
@@ -41,7 +40,7 @@ Client → Caddy (TLS) → api (FastAPI) ──task──▶ RabbitMQ ──▶
reCAPTCHA, hCaptcha) are solved via the client's 2captcha key. reCAPTCHA, hCaptcha) are solved via the client's 2captcha key.
- **Proxies**: not managed by the service — the client may supply their own proxy per request. Browser locale and - **Proxies**: not managed by the service — the client may supply their own proxy per request. Browser locale and
timezone are derived from the proxy's exit IP. timezone are derived from the proxy's exit IP.
- **Sessions**: `session_id` cookie → persistent Chrome profile on disk. A fingerprint (screen, DPR) is generated on - **Sessions**: `session_id` cookie → persistent Chrome profile on disk, removed after `PROFILE_TTL_DAYS` (7) unused days. A fingerprint (screen, DPR) is generated on
the first request of a session; cookies passed in a request replace the profile's cookies. the first request of a session; cookies passed in a request replace the profile's cookies.
- **Browser identity is not configurable**: user agent and window size are the real browser's own — a spoofed UA - **Browser identity is not configurable**: user agent and window size are the real browser's own — a spoofed UA
mismatches `userAgentData` and Cloudflare blocks it. Clients take `user_agent` from the response. mismatches `userAgentData` and Cloudflare blocks it. Clients take `user_agent` from the response.
+4 -1
View File
@@ -52,7 +52,9 @@ API доступно на `http://localhost:8000`.
- `extra/user_data_dir/<session_id>` — профили браузера (cookies, local storage, fingerprint) по `session_id`; - `extra/user_data_dir/<session_id>` — профили браузера (cookies, local storage, fingerprint) по `session_id`;
- `extra/geoip_cache.json` — кэш геолокации прокси. - `extra/geoip_cache.json` — кэш геолокации прокси.
Профили со временем накапливаются — чистить вручную: Воркер удаляет профили, которыми не пользовались `PROFILE_TTL_DAYS` дней (по умолчанию 7): при старте и затем раз
в час. Профиль с открытым окном не трогается. Клиент, вернувшийся позже, получает новый профиль и заново проходит
антибот. Удалить все профили вручную:
```bash ```bash
docker compose exec worker sh -c 'rm -rf /app/extra/user_data_dir/*' docker compose exec worker sh -c 'rm -rf /app/extra/user_data_dir/*'
@@ -78,6 +80,7 @@ docker compose exec worker sh -c 'rm -rf /app/extra/user_data_dir/*'
| `WORKER_THREADS` | `2 × BROWSER_SLOTS` | worker: потоков dramatiq | | `WORKER_THREADS` | `2 × BROWSER_SLOTS` | worker: потоков dramatiq |
| `WORKER_ID` | hostname | worker: имя воркера и его персональной очереди | | `WORKER_ID` | hostname | worker: имя воркера и его персональной очереди |
| `RESULT_TTL_HOURS` | `24` | сколько хранится результат задачи | | `RESULT_TTL_HOURS` | `24` | сколько хранится результат задачи |
| `PROFILE_TTL_DAYS` | `7` | worker: через сколько дней без использования удаляется профиль браузера |
| `V1_QUEUE_WAIT` | `30` | api: сколько `/v1/solve` ждёт, пока воркер возьмёт задачу; дальше `503` | | `V1_QUEUE_WAIT` | `30` | api: сколько `/v1/solve` ждёт, пока воркер возьмёт задачу; дальше `503` |
| `NO_SLOTS_RETRY_AFTER` | `5` | api: значение `Retry-After` (с) в ответе `503` | | `NO_SLOTS_RETRY_AFTER` | `5` | api: значение `Retry-After` (с) в ответе `503` |
| `RABBITMQ_URL` | `amqp://guest:guest@localhost:5672/` | api, worker; в compose задаётся сам | | `RABBITMQ_URL` | `amqp://guest:guest@localhost:5672/` | api, worker; в compose задаётся сам |
+2
View File
@@ -10,6 +10,8 @@ BROWSER_SLOTS=4
RABBITMQ_PASSWORD=robotex RABBITMQ_PASSWORD=robotex
# сколько часов хранится результат задачи в Redis # сколько часов хранится результат задачи в Redis
RESULT_TTL_HOURS=24 RESULT_TTL_HOURS=24
# через сколько дней без использования удаляется профиль браузера
PROFILE_TTL_DAYS=7
# сколько /v1/solve ждёт, пока воркер возьмёт задачу, прежде чем ответить 503 # сколько /v1/solve ждёт, пока воркер возьмёт задачу, прежде чем ответить 503
V1_QUEUE_WAIT=30 V1_QUEUE_WAIT=30
NO_SLOTS_RETRY_AFTER=5 NO_SLOTS_RETRY_AFTER=5
+17 -5
View File
@@ -31,10 +31,11 @@ BROWSER_SLOTS = int(os.getenv("BROWSER_SLOTS", "4"))
# через сколько задача без свободного слота снова попадёт в очередь # через сколько задача без свободного слота снова попадёт в очередь
NO_SLOT_DELAY_MS = 2000 NO_SLOT_DELAY_MS = 2000
HEARTBEAT_INTERVAL = 10 HEARTBEAT_INTERVAL = 10
PROFILE_CLEANUP_INTERVAL = 3600
TASK_TIME_LIMIT_MS = task_store.RUNNING_TTL * 1000 TASK_TIME_LIMIT_MS = task_store.RUNNING_TTL * 1000
_pool: WindowPool | None = None _pool: WindowPool | None = None
_background: asyncio.Task | None = None _background: list[asyncio.Task] = []
def _run(coro): def _run(coro):
@@ -52,17 +53,28 @@ async def _heartbeat_loop(pool: WindowPool) -> None:
await asyncio.sleep(HEARTBEAT_INTERVAL) await asyncio.sleep(HEARTBEAT_INTERVAL)
async def _cleanup_loop(pool: WindowPool) -> None:
"""Удаление старых профилей: сразу при старте и затем раз в час."""
while True:
try:
await pool.cleanup_profiles()
except Exception:
log.warning("profile cleanup failed", exc_info=True)
await asyncio.sleep(PROFILE_CLEANUP_INTERVAL)
async def _start() -> None: async def _start() -> None:
global _pool, _background global _pool
_pool = WindowPool(task_store.TaskStore(task_store.connect()), WORKER_ID, BROWSER_SLOTS) _pool = WindowPool(task_store.TaskStore(task_store.connect()), WORKER_ID, BROWSER_SLOTS)
await _pool.report() await _pool.report()
_background = asyncio.create_task(_heartbeat_loop(_pool)) _background.extend([asyncio.create_task(_heartbeat_loop(_pool)), asyncio.create_task(_cleanup_loop(_pool))])
log.info("worker %s: %d browser slots, queues %s, %s", WORKER_ID, BROWSER_SLOTS, COMMON_QUEUE, personal_queue(WORKER_ID)) log.info("worker %s: %d browser slots, queues %s, %s", WORKER_ID, BROWSER_SLOTS, COMMON_QUEUE, personal_queue(WORKER_ID))
async def _stop() -> None: async def _stop() -> None:
if _background: for task in _background:
_background.cancel() task.cancel()
_background.clear()
await _pool.close_all() await _pool.close_all()
await _pool.store.remove_worker(WORKER_ID) await _pool.store.remove_worker(WORKER_ID)
+42
View File
@@ -16,6 +16,8 @@ Redis: TTL ключа `session:<id>`, каждая задача в окно ст
import asyncio import asyncio
import base64 import base64
import logging import logging
import os
import shutil
import time import time
from contextlib import asynccontextmanager, suppress from contextlib import asynccontextmanager, suppress
from dataclasses import dataclass, field from dataclasses import dataclass, field
@@ -30,11 +32,16 @@ from api.geoip import resolve_geo
from api.logs import safe_request_dump, safe_response_dump from api.logs import safe_request_dump, safe_response_dump
from api.schemas import TaskRequest from api.schemas import TaskRequest
from engine import AsyncStealthySession from engine import AsyncStealthySession
from tasks.sessions import parse_session_id
from tasks.store import RUNNING_TTL, Status, TaskStore from tasks.store import RUNNING_TTL, Status, TaskStore
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
USER_DATA_ROOT = (Path(__file__).parent / "../../extra/user_data_dir").resolve() USER_DATA_ROOT = (Path(__file__).parent / "../../extra/user_data_dir").resolve()
# профиль, которым не пользовались столько дней, удаляется (#20)
PROFILE_TTL_DAYS = float(os.getenv("PROFILE_TTL_DAYS", "7"))
# время последнего использования профиля — mtime этого файла
LAST_USED_FILE = ".robotex_last_used"
class Outcome: class Outcome:
@@ -82,6 +89,18 @@ def session_dir(session_id: str) -> Path:
return user_data_dir return user_data_dir
def touch_profile(user_data_dir: Path) -> None:
(user_data_dir / LAST_USED_FILE).touch()
def profile_last_used(user_data_dir: Path) -> float:
try:
return (user_data_dir / LAST_USED_FILE).stat().st_mtime
except OSError:
# профиль из версии без отметки — по времени изменения каталога
return user_data_dir.stat().st_mtime
def make_page_action(payload: TaskRequest): def make_page_action(payload: TaskRequest):
"""page_action для AsyncStealthySession.fetch()/run(): выполняет очередь действий (вместе с """page_action для AsyncStealthySession.fetch()/run(): выполняет очередь действий (вместе с
прохождением антибота, api.actions.execute) и достаёт со страницы всё, что нужно для ответа. прохождением антибота, api.actions.execute) и достаёт со страницы всё, что нужно для ответа.
@@ -275,6 +294,7 @@ class WindowPool:
async def _open(self, session_id: str, payload: TaskRequest, nav_kwargs: dict): async def _open(self, session_id: str, payload: TaskRequest, nav_kwargs: dict):
user_data_dir = session_dir(session_id) user_data_dir = session_dir(session_id)
user_data_dir.mkdir(parents=True, exist_ok=True) user_data_dir.mkdir(parents=True, exist_ok=True)
touch_profile(user_data_dir)
# отпечаток задаётся профилю один раз (см. api/profile.py) # отпечаток задаётся профилю один раз (см. api/profile.py)
profile_kwargs = profile.browser_kwargs(profile.load_or_create(user_data_dir)) profile_kwargs = profile.browser_kwargs(profile.load_or_create(user_data_dir))
@@ -312,6 +332,9 @@ class WindowPool:
async def _close(self, window: Window) -> None: async def _close(self, window: Window) -> None:
"""Закрыть окно и освободить его слот.""" """Закрыть окно и освободить его слот."""
if window.engine is not None:
with suppress(OSError):
touch_profile(session_dir(window.session_id))
await self._close_engine(window) await self._close_engine(window)
if self.windows.get(window.session_id) is window: if self.windows.get(window.session_id) is window:
del self.windows[window.session_id] del self.windows[window.session_id]
@@ -346,3 +369,22 @@ class WindowPool:
await self._close(window) await self._close(window)
with suppress(Exception): with suppress(Exception):
await self.store.delete_session(session_id) await self.store.delete_session(session_id)
async def cleanup_profiles(self, ttl_days: float = PROFILE_TTL_DAYS) -> list[str]:
"""Удалить профили, которыми не пользовались ttl_days дней (#20). Профиль с открытым окном —
на этом воркере или на другом (ключ session:<id> в Redis) — не трогается."""
if not USER_DATA_ROOT.is_dir():
return []
deadline = time.time() - ttl_days * 86400
removed = []
for user_data_dir in USER_DATA_ROOT.iterdir():
session_id = parse_session_id(user_data_dir.name)
if not session_id or not user_data_dir.is_dir() or profile_last_used(user_data_dir) > deadline:
continue
if session_id in self.windows or self._busy(session_id) or await self.store.get_session(session_id):
continue
await asyncio.to_thread(shutil.rmtree, user_data_dir, True)
removed.append(session_id)
if removed:
log.info("removed %d browser profiles unused for %s days", len(removed), ttl_days)
return removed
+31
View File
@@ -178,3 +178,34 @@ def test_dead_worker_window_goes_to_common_queue(env):
env.auto = False env.auto = False
create(env, url="https://example.com") create(env, url="https://example.com")
assert env.messages[-1].queue_name == COMMON_QUEUE assert env.messages[-1].queue_name == COMMON_QUEUE
def test_old_profiles_cleaned_up(env, tmp_path):
"""#20: профили старше PROFILE_TTL_DAYS удаляются, кроме тех, где открыто окно."""
import os
import time
from worker.windows import LAST_USED_FILE
create(env, url="https://example.com", browser_idle_timeout=5)
held = env.client.cookies["session_id"]
elsewhere, old, fresh = (str(uuid.uuid4()) for _ in range(3))
env.redis.hset(session_key(elsewhere), mapping={"worker_id": "w2", "idle_timeout": 5})
long_ago = time.time() - 8 * 86400
for sid in (held, elsewhere, old, fresh, "not-a-session"):
(tmp_path / sid).mkdir(exist_ok=True)
marker = tmp_path / sid / LAST_USED_FILE
marker.touch()
if sid != fresh:
os.utime(marker, (long_ago, long_ago))
assert env.call(env.pool.cleanup_profiles) == [old]
assert sorted(p.name for p in tmp_path.iterdir()) == sorted([held, elsewhere, fresh, "not-a-session"])
def test_profile_marked_as_used(env, tmp_path):
from worker.windows import LAST_USED_FILE
create(env, url="https://example.com")
assert (tmp_path / env.client.cookies["session_id"] / LAST_USED_FILE).exists()