1586 lines
68 KiB
Python
1586 lines
68 KiB
Python
"""常驻监听任务（resident watch）。
|
||
|
||
与现有三样东西的关系，先说清楚，避免把旧语义冒充成常驻：
|
||
|
||
- 任务状态机（cloud/server.py）：queued -> running ->
|
||
completed / failed / interrupted / merged / merge-failed，
|
||
其中 /pause 只是把进行中回合任务标 interrupted 并置
|
||
paused_by_user=1（见 server._pause_task），不是一个可续跑
|
||
的“已暂停常驻态”。本模块绝不复用那条路：常驻态存在
|
||
resident_watches 表里，与任务表互不干扰。
|
||
- 云端自动化（cloud/automations.py）：定义层只有 active /
|
||
cancelled，到期靠外部触发器调 run_due_automation 抢单建任务，
|
||
服务端本身不起定时器。本模块复用它的两道口径而不是另起
|
||
一套：抢单用条件 UPDATE（同一间隔内只有一个执行器能拿到
|
||
rowcount 1），配额门直接调 automations._automation_quota_
|
||
exceeded（与 server / worker 同口径，改口径时只改那一处）。
|
||
- goal 循环（host.py run_goal_loop，max_rounds=0 不限轮次）：
|
||
目标是“做完一个目标”，结束条件是模型宣告达成；本模块的
|
||
目标是“一直监听外部变化”，结束条件只能是用户结束（或
|
||
配额耗尽转暂停，见下），两者不是一回事。
|
||
|
||
状态机（本模块自有，与任务状态机不冲突）：
|
||
running -> paused -> running（恢复）
|
||
running / paused -> ended（终态）
|
||
ended -> 任何操作一律拒绝（恢复必须新建任务，见 end_watch）
|
||
|
||
暂停保留现场：暂停只改 status / paused_reason / next_poll_at，
|
||
last_content_hash、last_poll_at、last_change_at、last_change_text
|
||
一律不动；暂停期间 due 为 False，驱动绝不轮询。恢复时
|
||
next_poll_at 置为 now（立刻复查一次），用旧游标比对：暂停
|
||
期间网站变了，恢复后第一次轮询就会把那次变化报出来，而不是
|
||
丢掉——“从游标续跑”就是这个意思。
|
||
|
||
结束是终态：ended 后 next_poll_at 为 NULL，驱动的 due 查询
|
||
只认 status='running'，所以结束后绝不会静默重启；ended 行
|
||
保留在库里供审计与查询，名字可重用（与自动化 cancelled 的
|
||
名字重用规则同款）。
|
||
|
||
变化才花钱：轮询本身是廉价抓取 + sha256 比对，不跑模型。
|
||
只有三类轮询产出：baseline（首次抓取，只建游标，不上报）、
|
||
unchanged（游标不变，不建任务）、changed（游标变了，才建
|
||
一个 queued 任务，prompt = 用户的处理提示 + 变化摘录）。
|
||
每次轮询都在 resident_polls 留一行，可查。
|
||
|
||
配额边界：配额门只在“变化要建模型任务”那一刻过，不在
|
||
轮询那一刻过（轮询不花模型额度）。配额耗尽时不结束、不丢
|
||
变化：状态转 paused、paused_reason='quota_exceeded'，游标不
|
||
推进（恢复后复查仍会报出同一变化），轮询记
|
||
changed_quota_paused。配额检查本身失败（fail-closed 文本）
|
||
时不转暂停，按瞬时失败退避处理，游标同样不推进。
|
||
|
||
驱动：服务端本身不起定时器（与自动化同款结论）。真正的
|
||
驱动是 cloud/worker.py 的 run_loop 轮次边界调
|
||
process_due_resident_watches（见 worker 里的调用点），或
|
||
任何外部触发器直接调本模块的 run_due_resident_watch /
|
||
process_due_resident_watches；两者都走 claim 门，两个驱动
|
||
同时跑也不会在同一间隔内重复轮询。
|
||
|
||
S2 二选一（本批选批量总预算，不新增独立驱动进程，论证
|
||
见验证摘要）：轮询间隔只是下限、不是 SLA——模型任务一轮
|
||
最长 1800 秒仍会拖后轮询，任何对外口径都必须写“尽力而
|
||
为、延迟无界”。批量总预算（拟定默认 120 秒，见
|
||
resident_config）把单批最坏耗时 bound 在“预算 + 单条超时
|
||
上限 60 秒”，剩余到期监听保持 due、下批再跑；延迟本身由
|
||
resident_metrics 的 due_count / max_overdue_seconds 度量，
|
||
进 /metrics，不靠间隔数值自证送达。独立驱动进程的运维
|
||
成本（新 unit、独立 DB 连接、启停监管）是审核第 7 节留给
|
||
用户的拍板项，本批不擅自新增；claim 门已保证未来加驱动
|
||
时多驱动不重复轮询。
|
||
|
||
本模块只依赖标准库 + cloud.automations 的复用函数
|
||
（row_get、_audit_row、_automation_quota_exceeded）与
|
||
cloud.db 的完整性错误类型（仅用于把唯一索引冲突转成
|
||
ValueError，不 import 其它 cloud 模块），避免循环 import。
|
||
"""
|
||
|
||
import hashlib
|
||
import http.client
|
||
import ipaddress
|
||
import os
|
||
import re
|
||
import socket
|
||
import sqlite3
|
||
import ssl
|
||
import time
|
||
import uuid
|
||
from urllib.parse import urljoin
|
||
from urllib.parse import urlsplit
|
||
|
||
from cloud.automations import _audit_row as _auto_audit_row
|
||
from cloud.automations import _automation_quota_exceeded
|
||
from cloud.peer_messages import row_get
|
||
from cloud.db import PostgresIntegrityError as _PgIntegrityError
|
||
|
||
MIN_INTERVAL_SECONDS = 60
|
||
MAX_INTERVAL_SECONDS = 30 * 86400
|
||
MAX_NAME_LEN = 100
|
||
MAX_PROMPT_LEN = 20000
|
||
MAX_URL_LEN = 2000
|
||
MIN_FETCH_TIMEOUT_SECONDS = 1
|
||
MAX_FETCH_TIMEOUT_SECONDS = 60
|
||
DEFAULT_FETCH_TIMEOUT_SECONDS = 10
|
||
MAX_FETCH_BYTES = 1000000
|
||
MAX_CHANGE_TEXT_LEN = 2000
|
||
MAX_POLL_DETAIL_LEN = 500
|
||
POLL_LIMIT = 50
|
||
MAX_REDIRECTS = 10
|
||
RESIDENT_USER_AGENT = "codex-app-server-resident-watch/1.0"
|
||
|
||
# P0/P1 拟定默认值（全部是拟定值、供最终验收改，不是用户已
|
||
# 拍板；数值只在这里与 resident_config() 一处定义，环境变量
|
||
# 可覆盖，见下文各 CODEX_RESIDENT_* 变量）：
|
||
# - 每监听每日变化任务上限 24 个：3600 秒间隔的监听一天最多
|
||
# 24 次变化，本值即“每小时最多 1 个任务”的量级；60 秒间隔
|
||
# 的抖动页面一天最多也只建 24 个任务，超限的变化记
|
||
# changed_capped 轮询、游标不推进（次日/下次仍可报，不丢）。
|
||
# - 变化抖动冷却 300 秒：上次建任务 300 秒内的新变化记
|
||
# changed_cooldown、游标不推进，冷却过后复查仍可报。
|
||
# - polls 保留期 90 天：批量驱动每批开头 prune 一次，过期行
|
||
# 删除、未过期保留；当日上限计数只依赖当日轮询行，保留期
|
||
# 远大于 1 天，不影响上限计数。
|
||
# - 批量总预算 120 秒：见 S2 论证（模块头“驱动”一段）。
|
||
DEFAULT_MAX_DAILY_CHANGE_TASKS = 24
|
||
DEFAULT_CHANGE_COOLDOWN_SECONDS = 300
|
||
DEFAULT_POLL_RETENTION_DAYS = 90
|
||
DEFAULT_BATCH_BUDGET_SECONDS = 120
|
||
# 结构化失败原因口径（落 poll.failure_reason 列，detail 前缀
|
||
# 同样写 reason=<原因>，老读法只看 detail 也不丢信息）：
|
||
# dns / proxy / timeout / http_status / ssrf_blocked / tls /
|
||
# redirect / network / invalid_url / unknown。
|
||
FAILURE_REASONS = frozenset(
|
||
{"dns", "proxy", "timeout", "http_status", "ssrf_blocked", "tls",
|
||
"redirect", "network", "invalid_url", "unknown"})
|
||
|
||
|
||
def _env_int(name, default):
|
||
"""环境变量整数：解析失败或 <=0 时回默认值，绝不因此炸掉
|
||
驱动（配置错误按默认跑，见摘要已知偏差）。"""
|
||
try:
|
||
value = int(os.environ.get(name) or default)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
return value if value > 0 else default
|
||
|
||
|
||
def resident_config():
|
||
"""常驻监听的可配置口径（每次调用时读环境变量，重启与
|
||
改环境变量即生效，不需改代码；数值本身是拟定默认值）：
|
||
- proxy：CODEX_RESIDENT_PROXY，未设/空/direct = 直连（默认
|
||
直连，不跟随进程 http_proxy/https_proxy）；env = 显式跟随
|
||
进程环境代理（旧 urllib 默认行为，须显式 opt-in）；其它值
|
||
= 显式代理 URL（http/https），no_proxy 仍读进程环境。
|
||
- max_daily_change_tasks：CODEX_RESIDENT_MAX_DAILY_CHANGE_TASKS。
|
||
- change_cooldown_seconds：CODEX_RESIDENT_CHANGE_COOLDOWN_SECONDS。
|
||
- poll_retention_days：CODEX_RESIDENT_POLL_RETENTION_DAYS。
|
||
- batch_budget_seconds：CODEX_RESIDENT_BATCH_BUDGET_SECONDS。
|
||
"""
|
||
raw = os.environ.get("CODEX_RESIDENT_PROXY")
|
||
if raw is None or not raw.strip():
|
||
proxy = {"mode": "direct", "url": None}
|
||
elif raw.strip().lower() == "direct":
|
||
proxy = {"mode": "direct", "url": None}
|
||
elif raw.strip().lower() == "env":
|
||
proxy = {"mode": "env", "url": None}
|
||
else:
|
||
proxy = {"mode": "url", "url": raw.strip()}
|
||
return {
|
||
"proxy": proxy,
|
||
"max_daily_change_tasks": _env_int(
|
||
"CODEX_RESIDENT_MAX_DAILY_CHANGE_TASKS",
|
||
DEFAULT_MAX_DAILY_CHANGE_TASKS),
|
||
"change_cooldown_seconds": _env_int(
|
||
"CODEX_RESIDENT_CHANGE_COOLDOWN_SECONDS",
|
||
DEFAULT_CHANGE_COOLDOWN_SECONDS),
|
||
"poll_retention_days": _env_int(
|
||
"CODEX_RESIDENT_POLL_RETENTION_DAYS",
|
||
DEFAULT_POLL_RETENTION_DAYS),
|
||
"batch_budget_seconds": _env_int(
|
||
"CODEX_RESIDENT_BATCH_BUDGET_SECONDS",
|
||
DEFAULT_BATCH_BUDGET_SECONDS),
|
||
}
|
||
|
||
|
||
def resolve_proxy_config(raw=None):
|
||
"""解析代理配置：非法代理 URL 在这里就抛 FetchError
|
||
（reason=proxy），调用方按结构化失败记轮询，不炸驱动。"""
|
||
cfg = resident_config()
|
||
if raw is not None:
|
||
# 显式传入的 raw 只用于测试与未来调用方：与环境变量
|
||
# 同款三态（direct/env/URL），不接受其它形态。
|
||
if raw.strip().lower() in ("direct", "env"):
|
||
return {"mode": raw.strip().lower(), "url": None}
|
||
if raw.strip():
|
||
cfg = dict(cfg)
|
||
cfg["proxy"] = {"mode": "url", "url": raw.strip()}
|
||
else:
|
||
cfg["proxy"] = {"mode": "direct", "url": None}
|
||
proxy = cfg["proxy"]
|
||
if proxy["mode"] == "url":
|
||
parts = urlsplit(proxy["url"])
|
||
if parts.scheme not in ("http", "https") or not parts.hostname:
|
||
raise FetchError(
|
||
"proxy",
|
||
"invalid resident proxy URL: %s" % proxy["url"])
|
||
return proxy
|
||
|
||
|
||
def _host_in_no_proxy(host, no_proxy_value):
|
||
"""no_proxy 匹配：精确、后缀（.example.com 匹配子域）、*
|
||
全匹配；主机名统一小写去尾点，与 validate 口径一致。"""
|
||
host = (host or "").strip().lower().rstrip(".")
|
||
for entry in (no_proxy_value or "").split(","):
|
||
entry = entry.strip().lower().rstrip(".")
|
||
if not entry:
|
||
continue
|
||
if entry == "*":
|
||
return True
|
||
if host == entry or host.endswith("." + entry):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _proxy_for_target(host, scheme, proxy):
|
||
"""本跳的目标代理决策：返回 None = 直连，否则
|
||
(proxy_host, proxy_port)。env 模式读进程环境
|
||
http_proxy/https_proxy（大小写均认）；url 模式用配置的
|
||
URL；两种模式都认进程 no_proxy/NO_PROXY。直连模式永远
|
||
返回 None——旧代码跟随进程代理的行为至此结束，改动见
|
||
摘要已知偏差（部署时若宿主出口必须走代理，须显式配
|
||
CODEX_RESIDENT_PROXY=env 或代理 URL）。"""
|
||
no_proxy = os.environ.get("no_proxy") or os.environ.get("NO_PROXY") or ""
|
||
if _host_in_no_proxy(host, no_proxy):
|
||
return None
|
||
if proxy["mode"] == "direct":
|
||
return None
|
||
if proxy["mode"] == "env":
|
||
if scheme == "https":
|
||
raw = os.environ.get("https_proxy") \
|
||
or os.environ.get("HTTPS_PROXY") or ""
|
||
else:
|
||
raw = os.environ.get("http_proxy") \
|
||
or os.environ.get("HTTP_PROXY") or ""
|
||
if not raw.strip():
|
||
return None
|
||
parts = urlsplit(raw.strip())
|
||
if parts.scheme not in ("http", "https") or not parts.hostname:
|
||
raise FetchError("proxy",
|
||
"invalid proxy URL in environment")
|
||
return (parts.hostname,
|
||
parts.port or (443 if parts.scheme == "https" else 80))
|
||
parts = urlsplit(proxy["url"])
|
||
return (parts.hostname,
|
||
parts.port or (443 if parts.scheme == "https" else 80))
|
||
|
||
RUNNING = "running"
|
||
PAUSED = "paused"
|
||
ENDED = "ended"
|
||
|
||
# 变化判定口径（v1）：对抓取文本做空白折叠后取 sha256。
|
||
# 动态页面（广告、时间戳）本身会抖动，v1 不做选择器/区域
|
||
# 提取，抖动风险写在验证摘要的已知偏差里，不假装已解决。
|
||
_WS_RE = re.compile(r"\s+")
|
||
|
||
|
||
WATCH_COLUMNS = (
|
||
"id", "project_id", "name", "target_url", "prompt",
|
||
"interval_seconds", "fetch_timeout_seconds", "status",
|
||
"last_poll_at", "last_claimed_at", "next_poll_at",
|
||
"last_content_hash", "last_change_at", "last_change_text",
|
||
"consecutive_failures", "paused_reason", "ended_at",
|
||
"created_by", "created_at", "updated_at")
|
||
|
||
POLL_COLUMNS = (
|
||
"id", "watch_id", "project_id", "status", "observed_hash",
|
||
"detail", "task_id", "created_at", "failure_reason")
|
||
|
||
|
||
def _row_dict(row, columns):
|
||
if row is None or isinstance(row, dict):
|
||
return row
|
||
try:
|
||
row["id"]
|
||
return row
|
||
except (KeyError, IndexError, TypeError):
|
||
# sqlite3.Row 缺列报 IndexError（不是 TypeError）：只有
|
||
# SELECT * 的行能原样用，单列查询（如 claim 取 interval）
|
||
# 走 zip 回落，columns 必须与 SELECT 列顺序一致。
|
||
return dict(zip(columns, row))
|
||
|
||
|
||
def _existing_columns(conn, table):
|
||
try:
|
||
return {row[1] for row in conn.execute(
|
||
"PRAGMA table_info(%s)" % table).fetchall()}
|
||
except Exception:
|
||
return set()
|
||
|
||
|
||
def ensure_resident_tables(conn):
|
||
"""幂等建常驻监听表，补齐新增列（仿 automations 写法）。
|
||
|
||
调用约定（G1）：本函数只在启动/迁移/API 路径调——
|
||
server init_db、worker 首次批量、create/list/get 等 API
|
||
函数；热路径（claim / run_due / due 列表 / 批量循环）绝
|
||
不调它，批量内每条的 CREATE INDEX DDL 锁与 claim UPDATE
|
||
的锁交互正是旧双 worker 死锁的疑似根因。热路径只在
|
||
表确实缺失时（missing-table 错误）ensure 后重试一次。
|
||
"""
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS resident_watches("
|
||
"id TEXT PRIMARY KEY, "
|
||
"project_id TEXT, "
|
||
"name TEXT, "
|
||
"target_url TEXT, "
|
||
"prompt TEXT, "
|
||
"interval_seconds INTEGER, "
|
||
"fetch_timeout_seconds INTEGER, "
|
||
"status TEXT, "
|
||
"last_poll_at DOUBLE PRECISION, "
|
||
"last_claimed_at DOUBLE PRECISION, "
|
||
"next_poll_at DOUBLE PRECISION, "
|
||
"last_content_hash TEXT, "
|
||
"last_change_at DOUBLE PRECISION, "
|
||
"last_change_text TEXT, "
|
||
"consecutive_failures INTEGER, "
|
||
"paused_reason TEXT, "
|
||
"ended_at DOUBLE PRECISION, "
|
||
"created_by TEXT, "
|
||
"created_at DOUBLE PRECISION, "
|
||
"updated_at DOUBLE PRECISION)")
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS resident_polls("
|
||
"id TEXT PRIMARY KEY, "
|
||
"watch_id TEXT, "
|
||
"project_id TEXT, "
|
||
"status TEXT, "
|
||
"observed_hash TEXT, "
|
||
"detail TEXT, "
|
||
"task_id TEXT, "
|
||
"created_at DOUBLE PRECISION, "
|
||
"failure_reason TEXT)")
|
||
existing = _existing_columns(conn, "resident_watches")
|
||
for col, ddl in (
|
||
("fetch_timeout_seconds", "INTEGER"),
|
||
("next_poll_at", "DOUBLE PRECISION"),
|
||
("last_content_hash", "TEXT"),
|
||
("last_change_at", "DOUBLE PRECISION"),
|
||
("last_change_text", "TEXT"),
|
||
("consecutive_failures", "INTEGER"),
|
||
("paused_reason", "TEXT"),
|
||
("ended_at", "DOUBLE PRECISION")):
|
||
if col not in existing:
|
||
try:
|
||
conn.execute(
|
||
"ALTER TABLE resident_watches ADD COLUMN %s %s"
|
||
% (col, ddl))
|
||
except Exception:
|
||
pass
|
||
existing_polls = _existing_columns(conn, "resident_polls")
|
||
if "failure_reason" not in existing_polls:
|
||
try:
|
||
conn.execute(
|
||
"ALTER TABLE resident_polls"
|
||
" ADD COLUMN failure_reason TEXT")
|
||
except Exception:
|
||
pass
|
||
conn.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_resident_watches_project"
|
||
" ON resident_watches(project_id, status)")
|
||
# 同项目未结束名唯一：先查后插挡不住并发，唯一索引才是
|
||
# 真正的关口（server environments 的同款部分唯一索引，
|
||
# WHERE status != 'archived' 的 resident 版）；已结束名
|
||
# 不在索引内，可重用。已存在重复名的旧库建索引会失败，
|
||
# 此时不炸掉整次建表调用，SELECT 查重仍挡正常路径。
|
||
try:
|
||
conn.execute(
|
||
"CREATE UNIQUE INDEX IF NOT EXISTS"
|
||
" idx_resident_watches_name"
|
||
" ON resident_watches(project_id, name)"
|
||
" WHERE status != 'ended'")
|
||
except Exception:
|
||
pass
|
||
conn.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_resident_polls_watch"
|
||
" ON resident_polls(watch_id, created_at)")
|
||
|
||
|
||
def _backoff_seconds(failures):
|
||
"""失败退避增量：30、60、120……秒，上限 3600 秒。
|
||
调用方把它加在间隔之上（下次 = now + interval + backoff），
|
||
失败后绝不比正常间隔更频繁地重试，行为见失败分支测试。"""
|
||
if failures <= 0:
|
||
return 0
|
||
return min(3600, 30 * (2 ** (failures - 1)))
|
||
|
||
|
||
def watch_to_dict(row, now=None):
|
||
row = _row_dict(row, WATCH_COLUMNS)
|
||
if now is None:
|
||
now = time.time()
|
||
status = row["status"]
|
||
nxt = row_get(row, "next_poll_at")
|
||
due = bool(status == RUNNING and (nxt is None or nxt <= now))
|
||
return {
|
||
"id": row["id"],
|
||
"project_id": row["project_id"],
|
||
"name": row["name"],
|
||
"target_url": row["target_url"],
|
||
"prompt": row["prompt"],
|
||
"interval_seconds": row["interval_seconds"],
|
||
"fetch_timeout_seconds": row_get(row, "fetch_timeout_seconds")
|
||
or DEFAULT_FETCH_TIMEOUT_SECONDS,
|
||
"status": status,
|
||
"last_poll_at": row_get(row, "last_poll_at"),
|
||
"last_claimed_at": row_get(row, "last_claimed_at"),
|
||
"next_poll_at": nxt,
|
||
"last_content_hash": row_get(row, "last_content_hash"),
|
||
"last_change_at": row_get(row, "last_change_at"),
|
||
"last_change_text": row_get(row, "last_change_text"),
|
||
"consecutive_failures": row_get(row, "consecutive_failures") or 0,
|
||
"paused_reason": row_get(row, "paused_reason"),
|
||
"ended_at": row_get(row, "ended_at"),
|
||
"created_by": row["created_by"],
|
||
"created_at": row["created_at"],
|
||
"updated_at": row_get(row, "updated_at"),
|
||
"due": due,
|
||
}
|
||
|
||
|
||
def _is_noncanonical_numeric_host(host):
|
||
"""非规范数字 IP 写法：单整数（2130706433）、十六进制
|
||
（0x7f000001）、前导零八进制（0177.0.0.1）、含十六进制段
|
||
的点分写法——urllib/socket 的 inet_aton 会把它们解析成
|
||
数字 IP，ipaddress 却认不出来。规范点分十进制（4 段、
|
||
每段 0-255、无前导零，0 本身除外）不在此列；普通域名
|
||
（含字母的标签）也不在此列。"""
|
||
if re.fullmatch(r"0[xX][0-9a-fA-F]+", host):
|
||
return True
|
||
if re.fullmatch(r"[0-9]+", host):
|
||
return True
|
||
if "." not in host:
|
||
return False
|
||
labels = host.split(".")
|
||
if not all(re.fullmatch(r"0[xX][0-9a-fA-F]+|[0-9]+", lab)
|
||
for lab in labels):
|
||
return False
|
||
if len(labels) != 4:
|
||
return True
|
||
for lab in labels:
|
||
if lab.startswith("0x") or lab.startswith("0X"):
|
||
return True
|
||
if len(lab) > 1 and lab.startswith("0"):
|
||
return True
|
||
if int(lab) > 255:
|
||
return True
|
||
return False
|
||
|
||
|
||
def validate_target_url(url):
|
||
"""目标 URL 字面校验：仅 http/https、无 userinfo、主机非空、
|
||
字面 localhost（含尾点、*.localhost）/ 私有 / 回环 /
|
||
链路本地 IP / 非规范数字 IP 写法拒绝（与
|
||
websearch._is_blocked_fetch_url 的字面 IP 口径同款）。
|
||
本函数只做字面层：创建监听时调它，域名解析到私网的
|
||
情形不在创建时拦（DNS 会变，创建时放行不等于抓取时放
|
||
行）；抓取时的 DNS 级门在 default_fetch 里逐跳执行
|
||
（resolve_public_addresses + pin），那才是权威关口。
|
||
私网抓取在抓取层直接失败、不产生摘录、不建任务。"""
|
||
if not url or not isinstance(url, str) or not url.strip():
|
||
raise ValueError("target_url required")
|
||
url = url.strip()
|
||
if len(url) > MAX_URL_LEN:
|
||
raise ValueError("target_url too long")
|
||
parts = urlsplit(url)
|
||
if parts.scheme not in ("http", "https"):
|
||
raise ValueError("target_url must be http or https")
|
||
if parts.username or parts.password:
|
||
raise ValueError("target_url must not contain credentials")
|
||
host = (parts.hostname or "").strip().lower().rstrip(".")
|
||
if not host:
|
||
raise ValueError("target_url host required")
|
||
if host == "localhost" or host.endswith(".localhost"):
|
||
raise ValueError("localhost URLs are blocked")
|
||
if _is_noncanonical_numeric_host(host):
|
||
raise ValueError("non-canonical numeric IP URLs are blocked")
|
||
try:
|
||
addr = ipaddress.ip_address(host)
|
||
except ValueError:
|
||
addr = None
|
||
if addr is not None and (addr.is_private or addr.is_loopback
|
||
or addr.is_link_local or addr.is_reserved
|
||
or addr.is_multicast):
|
||
raise ValueError("private / loopback URLs are blocked")
|
||
return url
|
||
|
||
|
||
def _validate_fields(name, prompt, target_url, interval_seconds,
|
||
fetch_timeout_seconds):
|
||
if not name or not isinstance(name, str) or not name.strip():
|
||
raise ValueError("name required")
|
||
name = name.strip()
|
||
if len(name) > MAX_NAME_LEN:
|
||
raise ValueError("name too long")
|
||
if not prompt or not isinstance(prompt, str) or not prompt.strip():
|
||
raise ValueError("prompt required")
|
||
prompt = prompt.strip()
|
||
if len(prompt) > MAX_PROMPT_LEN:
|
||
raise ValueError("prompt too long")
|
||
target_url = validate_target_url(target_url)
|
||
if isinstance(interval_seconds, bool) \
|
||
or not isinstance(interval_seconds, int):
|
||
raise ValueError("interval_seconds must be an integer")
|
||
if interval_seconds < MIN_INTERVAL_SECONDS:
|
||
raise ValueError(
|
||
"interval_seconds must be at least %d" % MIN_INTERVAL_SECONDS)
|
||
if interval_seconds > MAX_INTERVAL_SECONDS:
|
||
raise ValueError("interval_seconds too large")
|
||
if fetch_timeout_seconds is None:
|
||
fetch_timeout_seconds = DEFAULT_FETCH_TIMEOUT_SECONDS
|
||
if isinstance(fetch_timeout_seconds, bool) \
|
||
or not isinstance(fetch_timeout_seconds, int):
|
||
raise ValueError("fetch_timeout_seconds must be an integer")
|
||
if not MIN_FETCH_TIMEOUT_SECONDS <= fetch_timeout_seconds \
|
||
<= MAX_FETCH_TIMEOUT_SECONDS:
|
||
raise ValueError(
|
||
"fetch_timeout_seconds must be between %d and %d"
|
||
% (MIN_FETCH_TIMEOUT_SECONDS, MAX_FETCH_TIMEOUT_SECONDS))
|
||
return name, prompt, target_url, interval_seconds, fetch_timeout_seconds
|
||
|
||
|
||
def create_watch(conn, project_id, name, target_url, prompt, created_by,
|
||
interval_seconds=3600, fetch_timeout_seconds=None,
|
||
now=None):
|
||
"""新建常驻监听：建完即到期（next_poll_at=now），驱动下
|
||
一轮即可认领；同项目内未结束（running/paused）的名字唯一，
|
||
已结束的名字可重用。创建本身不走配额门（轮询不花模型额度，
|
||
配额只在变化建任务时过，见 run_due_resident_watch）。"""
|
||
(name, prompt, target_url, interval_seconds,
|
||
fetch_timeout_seconds) = _validate_fields(
|
||
name, prompt, target_url, interval_seconds, fetch_timeout_seconds)
|
||
if now is None:
|
||
now = time.time()
|
||
ensure_resident_tables(conn)
|
||
clash = conn.execute(
|
||
"SELECT id FROM resident_watches WHERE project_id = ?"
|
||
" AND name = ? AND status != ?",
|
||
(project_id, name, ENDED)).fetchone()
|
||
if clash is not None:
|
||
raise ValueError(
|
||
"resident watch name already exists in this project")
|
||
watch_id = uuid.uuid4().hex
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO resident_watches(id, project_id, name, target_url,"
|
||
" prompt, interval_seconds, fetch_timeout_seconds, status,"
|
||
" last_poll_at, last_claimed_at, next_poll_at,"
|
||
" last_content_hash, last_change_at, last_change_text,"
|
||
" consecutive_failures, paused_reason, ended_at,"
|
||
" created_by, created_at, updated_at)"
|
||
" VALUES(?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?,"
|
||
" NULL, NULL, NULL, 0, NULL, NULL, ?, ?, ?)",
|
||
(watch_id, project_id, name, target_url, prompt,
|
||
interval_seconds, fetch_timeout_seconds, RUNNING, now,
|
||
created_by, now, now))
|
||
except (sqlite3.IntegrityError, _PgIntegrityError) as exc:
|
||
# 并发下 SELECT 查重与 INSERT 之间可能插进同名行，
|
||
# 唯一索引是真正的关口：冲突转回与查重同款 ValueError，
|
||
# 调用方（server）据此返回 400，不把数据库错误外泄。
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
raise ValueError(
|
||
"resident watch name already exists in this project") from exc
|
||
conn.commit()
|
||
row = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
|
||
return watch_to_dict(row, now)
|
||
|
||
|
||
def list_watches(conn, project_id, now=None):
|
||
if now is None:
|
||
now = time.time()
|
||
ensure_resident_tables(conn)
|
||
rows = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE project_id = ?"
|
||
" ORDER BY created_at, id",
|
||
(project_id,)).fetchall()
|
||
return [watch_to_dict(r, now) for r in rows]
|
||
|
||
|
||
def get_watch(conn, watch_id, now=None):
|
||
if now is None:
|
||
now = time.time()
|
||
ensure_resident_tables(conn)
|
||
row = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
|
||
if row is None:
|
||
return None
|
||
return watch_to_dict(row, now)
|
||
|
||
|
||
def list_polls(conn, watch_id, limit=POLL_LIMIT):
|
||
ensure_resident_tables(conn)
|
||
rows = conn.execute(
|
||
"SELECT * FROM resident_polls WHERE watch_id = ?"
|
||
" ORDER BY created_at DESC, id LIMIT ?",
|
||
(watch_id, limit)).fetchall()
|
||
out = []
|
||
for raw in rows:
|
||
r = _row_dict(raw, POLL_COLUMNS)
|
||
out.append({
|
||
"id": r["id"],
|
||
"watch_id": r["watch_id"],
|
||
"project_id": r["project_id"],
|
||
"status": r["status"],
|
||
"observed_hash": row_get(r, "observed_hash"),
|
||
"detail": row_get(r, "detail") or "",
|
||
"task_id": row_get(r, "task_id"),
|
||
"created_at": r["created_at"],
|
||
"failure_reason": row_get(r, "failure_reason"),
|
||
})
|
||
return out
|
||
|
||
|
||
def _transition(conn, watch_id, from_statuses, to_status, now,
|
||
extra_sets=(), extra_vals=()):
|
||
"""条件状态迁移：返回 (ok, dict)。行不存在时 (False, None)；
|
||
状态不对时 (False, 旧 dict)，调用方据此区分 404 与 409。"""
|
||
ensure_resident_tables(conn)
|
||
row = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE id = ?",
|
||
(watch_id,)).fetchone()
|
||
if row is None:
|
||
return False, None
|
||
row = _row_dict(row, WATCH_COLUMNS)
|
||
if row["status"] not in from_statuses:
|
||
return False, watch_to_dict(row, now)
|
||
placeholders = ",".join("?" for _ in from_statuses)
|
||
sets = ["status = ?", "updated_at = ?"] + list(extra_sets)
|
||
vals = [to_status, now] + list(extra_vals) + [watch_id] \
|
||
+ list(from_statuses)
|
||
cur = conn.execute(
|
||
"UPDATE resident_watches SET %s WHERE id = ? AND status IN (%s)"
|
||
% (", ".join(sets), placeholders), tuple(vals))
|
||
conn.commit()
|
||
if cur.rowcount != 1:
|
||
return False, watch_to_dict(row, now)
|
||
row = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
|
||
return True, watch_to_dict(row, now)
|
||
|
||
|
||
def pause_watch(conn, watch_id, now=None, reason="user_paused"):
|
||
"""用户暂停：仅 running 能暂停；保留全部现场，只把
|
||
next_poll_at 清掉，驱动即停。"""
|
||
if now is None:
|
||
now = time.time()
|
||
return _transition(conn, watch_id, (RUNNING,), PAUSED, now,
|
||
extra_sets=("paused_reason = ?",
|
||
"next_poll_at = NULL"),
|
||
extra_vals=(reason,))
|
||
|
||
|
||
def resume_watch(conn, watch_id, now=None):
|
||
"""恢复：仅 paused 能恢复；现场不动，next_poll_at=now
|
||
立刻复查一次（用旧游标比对，暂停期间的变化不丢）。"""
|
||
if now is None:
|
||
now = time.time()
|
||
return _transition(conn, watch_id, (PAUSED,), RUNNING, now,
|
||
extra_sets=("paused_reason = NULL",
|
||
"next_poll_at = ?"),
|
||
extra_vals=(now,))
|
||
|
||
|
||
def end_watch(conn, watch_id, now=None):
|
||
"""结束：running / paused -> ended，终态。ended 后
|
||
pause / resume / end 一律 (False, 旧 dict)，调用方转 409；
|
||
恢复已结束的监听必须新建任务，不复活本行。"""
|
||
if now is None:
|
||
now = time.time()
|
||
return _transition(conn, watch_id, (RUNNING, PAUSED), ENDED, now,
|
||
extra_sets=("ended_at = ?",
|
||
"next_poll_at = NULL"),
|
||
extra_vals=(now,))
|
||
|
||
|
||
def update_watch(conn, watch_id, fields, now=None):
|
||
"""更新定义：只认 name / prompt / target_url /
|
||
interval_seconds / fetch_timeout_seconds；ended 的行 409
|
||
（终态不能改，改它必须重建）；改间隔不重置游标、
|
||
不立即触发：已排好的 next_poll_at 不变，下一次轮询
|
||
完成后才按新间隔重算（想立即复查请暂停后恢复）。
|
||
改目标 URL 则强制走一次 baseline：last_content_hash
|
||
置 NULL，下一轮只建基线、不建任务——旧 hash 与新站
|
||
内容比对必然报一次假变化的旧行为至此结束；改间隔、
|
||
改提示、改名一律不重置游标（语义保留，见摘要）。"""
|
||
if now is None:
|
||
now = time.time()
|
||
ensure_resident_tables(conn)
|
||
row = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
|
||
if row is None:
|
||
return False, None
|
||
row = _row_dict(row, WATCH_COLUMNS)
|
||
if row["status"] == ENDED:
|
||
return False, watch_to_dict(row, now)
|
||
if not isinstance(fields, dict):
|
||
raise ValueError("fields must be an object")
|
||
if not any(k in fields for k in (
|
||
"name", "prompt", "target_url", "interval_seconds",
|
||
"fetch_timeout_seconds")):
|
||
raise ValueError("no updatable field given")
|
||
current = watch_to_dict(row, now)
|
||
(name, prompt, target_url, interval, timeout) = _validate_fields(
|
||
fields.get("name", current["name"]),
|
||
fields.get("prompt", current["prompt"]),
|
||
fields.get("target_url", current["target_url"]),
|
||
fields.get("interval_seconds", current["interval_seconds"]),
|
||
fields.get("fetch_timeout_seconds",
|
||
current["fetch_timeout_seconds"]))
|
||
if name != current["name"]:
|
||
clash = conn.execute(
|
||
"SELECT id FROM resident_watches WHERE project_id = ?"
|
||
" AND name = ? AND status != ? AND id != ?",
|
||
(row["project_id"], name, ENDED, watch_id)).fetchone()
|
||
if clash is not None:
|
||
raise ValueError(
|
||
"resident watch name already exists in this project")
|
||
target_changed = target_url != current["target_url"]
|
||
try:
|
||
if target_changed:
|
||
conn.execute(
|
||
"UPDATE resident_watches SET name = ?, prompt = ?,"
|
||
" target_url = ?, interval_seconds = ?,"
|
||
" fetch_timeout_seconds = ?, updated_at = ?,"
|
||
" last_content_hash = NULL WHERE id = ?",
|
||
(name, prompt, target_url, interval, timeout, now,
|
||
watch_id))
|
||
else:
|
||
conn.execute(
|
||
"UPDATE resident_watches SET name = ?, prompt = ?,"
|
||
" target_url = ?, interval_seconds = ?,"
|
||
" fetch_timeout_seconds = ?, updated_at = ? WHERE id = ?",
|
||
(name, prompt, target_url, interval, timeout, now,
|
||
watch_id))
|
||
except (sqlite3.IntegrityError, _PgIntegrityError) as exc:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
raise ValueError(
|
||
"resident watch name already exists in this project") from exc
|
||
conn.commit()
|
||
row = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
|
||
return True, watch_to_dict(row, now)
|
||
|
||
|
||
def _is_missing_table_error(exc):
|
||
"""缺表错误判定（SQLite 与 PG 两种文案）：热路径只在这种
|
||
错误上 ensure 后重试一次，其它错误一律不触发 DDL。"""
|
||
msg = str(exc).lower()
|
||
return "no such table" in msg or "does not exist" in msg
|
||
|
||
|
||
def due_watches(conn, now=None):
|
||
"""全部项目当前到期的 running 监听（纯读，不抢单），
|
||
供驱动列出待处理项；执行入口是 claim + run_due。
|
||
热路径不调 ensure（见 ensure_resident_tables 的约定）；
|
||
表缺失时由批量入口的 fallback 统一处理。"""
|
||
if now is None:
|
||
now = time.time()
|
||
rows = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE status = ?"
|
||
" AND (next_poll_at IS NULL OR next_poll_at <= ?)"
|
||
" ORDER BY next_poll_at, created_at, id",
|
||
(RUNNING, now)).fetchall()
|
||
return [watch_to_dict(r, now) for r in rows]
|
||
|
||
|
||
def _due_watches_with_fallback(conn, now):
|
||
"""批量入口的 due 列表：正常路径零 DDL；表缺失时
|
||
ensure 一次后重试，仍失败则把原始错误抛给调用方。"""
|
||
try:
|
||
return due_watches(conn, now)
|
||
except Exception as exc:
|
||
if not _is_missing_table_error(exc):
|
||
raise
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
ensure_resident_tables(conn)
|
||
return due_watches(conn, now)
|
||
|
||
|
||
def claim_due_watch(conn, watch_id, now=None):
|
||
"""抢单：条件 UPDATE 只有一行能成功。抢到即把
|
||
last_poll_at / last_claimed_at 写成 now，并把 next_poll_at
|
||
预置为 now + interval（崩在抓取中途时，下个间隔仍会重试，
|
||
不会因认领锁死而永久停摆；抓取完成后 run_due 会按成败
|
||
重算 next_poll_at）。抢不到时返回 False，调用方必须放弃。"""
|
||
if now is None:
|
||
now = time.time()
|
||
row = conn.execute(
|
||
"SELECT interval_seconds FROM resident_watches WHERE id = ?",
|
||
(watch_id,)).fetchone()
|
||
if row is None:
|
||
return False
|
||
interval = row_get(_row_dict(row, ("interval_seconds",)),
|
||
"interval_seconds") or MIN_INTERVAL_SECONDS
|
||
cur = conn.execute(
|
||
"UPDATE resident_watches SET last_poll_at = ?,"
|
||
" last_claimed_at = ?, next_poll_at = ?"
|
||
" WHERE id = ? AND status = ?"
|
||
" AND (next_poll_at IS NULL OR next_poll_at <= ?)",
|
||
(now, now, now + interval, watch_id, RUNNING, now))
|
||
conn.commit()
|
||
return cur.rowcount == 1
|
||
|
||
|
||
def _content_hash(text):
|
||
normalized = _WS_RE.sub(" ", text or "").strip()
|
||
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
|
||
|
||
|
||
class FetchError(ValueError):
|
||
"""结构化抓取失败：reason 见 FAILURE_REASONS，status 仅
|
||
http_status 时有值。ValueError 子类，故旧调用方按
|
||
ValueError 捕获的行为不变；run_due 据此写 failure_reason
|
||
与 reason= 前缀的 detail，不把原始内网响应写进任何记录。"""
|
||
|
||
def __init__(self, reason, message, status=None):
|
||
super().__init__(message)
|
||
self.reason = reason if reason in FAILURE_REASONS else "unknown"
|
||
self.status = status
|
||
|
||
|
||
class SSRFBlockedError(FetchError):
|
||
"""SSRF 拦截：字面或解析后地址命中内网/私有/回环/
|
||
链路本地/保留/组播/未指定地址，或非规范数字写法。"""
|
||
|
||
def __init__(self, message):
|
||
super().__init__("ssrf_blocked", message)
|
||
|
||
|
||
class HttpStatusFetchError(FetchError):
|
||
"""非 200 响应：不把错误页当成“变化”处理，也不把它
|
||
的正文写进变化任务（正文在 default_fetch 里直接丢弃）。"""
|
||
|
||
def __init__(self, status, message=None):
|
||
super().__init__(
|
||
"http_status",
|
||
message or ("HTTP status %s" % status), status=status)
|
||
|
||
|
||
def classify_fetch_error(exc):
|
||
"""异常 -> 结构化原因：FetchError 自带 reason；超时、
|
||
TLS、连接类按类型分；其余一律 unknown（不谎称是网络
|
||
错误，也不把异常文本当原因）。"""
|
||
if isinstance(exc, FetchError):
|
||
return exc.reason
|
||
if isinstance(exc, (socket.timeout, TimeoutError)):
|
||
return "timeout"
|
||
if isinstance(exc, ssl.SSLError):
|
||
return "tls"
|
||
if isinstance(exc, (ConnectionError, OSError)):
|
||
return "network"
|
||
if isinstance(exc, ValueError):
|
||
return "invalid_url"
|
||
return "unknown"
|
||
|
||
|
||
def _default_resolver(host, port):
|
||
"""默认解析器：socket.getaddrinfo，去重保序。返回 IP
|
||
字符串列表；测试可注入同签名的 resolver（见 default_fetch
|
||
的 resolver 参数——生产路径一律不传，默认即本函数）。"""
|
||
infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
|
||
ips = []
|
||
for info in infos:
|
||
try:
|
||
ip = info[4][0]
|
||
except (IndexError, TypeError):
|
||
continue
|
||
if ip not in ips:
|
||
ips.append(ip)
|
||
return ips
|
||
|
||
|
||
def _address_is_blocked(addr):
|
||
"""单个解析地址是否命中拦截面：私有/回环/链路本地/
|
||
保留/组播/未指定；IPv6 的 IPv4 映射地址同样按映射后
|
||
的 IPv4 地址再判一次（::ffff:127.0.0.1 必须拦住）。"""
|
||
if addr.is_private or addr.is_loopback or addr.is_link_local \
|
||
or addr.is_reserved or addr.is_multicast \
|
||
or addr.is_unspecified:
|
||
return True
|
||
mapped = getattr(addr, "ipv4_mapped", None)
|
||
if mapped is not None and (
|
||
mapped.is_private or mapped.is_loopback
|
||
or mapped.is_link_local or mapped.is_reserved
|
||
or mapped.is_multicast or mapped.is_unspecified):
|
||
return True
|
||
return False
|
||
|
||
|
||
def resolve_public_addresses(host, port, resolver=None):
|
||
"""DNS 级 SSRF 门：解析全部 A/AAAA，任一地址命中拦截面
|
||
即整批拒绝（只放行“全部解析结果都是公网地址”的目标，
|
||
不搞“取第一个公网地址蒙混过关”）；解析失败、空结果、
|
||
非 IP 结果一律按 dns 失败处理，不放行、不重试建连。
|
||
注意：本门只在直连模式下配合下文的 pin 才是完整封口；
|
||
代理模式下建连由代理方二次解析，见 default_fetch 的
|
||
残留风险说明。"""
|
||
try:
|
||
ips = (resolver or _default_resolver)(host, port)
|
||
except FetchError:
|
||
raise
|
||
except Exception as exc:
|
||
raise FetchError(
|
||
"dns", "DNS resolution failed for %s: %s" % (host, exc))
|
||
if not ips:
|
||
raise FetchError("dns", "DNS resolution returned no address"
|
||
" for %s" % host)
|
||
out = []
|
||
for ipstr in ips:
|
||
try:
|
||
addr = ipaddress.ip_address(str(ipstr))
|
||
except ValueError:
|
||
raise FetchError(
|
||
"dns", "DNS resolution returned non-IP %r for %s"
|
||
% (ipstr, host))
|
||
if _address_is_blocked(addr):
|
||
raise SSRFBlockedError(
|
||
"resolved address %s for %s is not public"
|
||
% (ipstr, host))
|
||
out.append(str(ipstr))
|
||
return out
|
||
|
||
|
||
class _PinnedHTTPConnection(http.client.HTTPConnection):
|
||
"""直连时 pin 住解析结果的连接：建连目标是解析出的 IP，
|
||
而不是再次按主机名解析（解析与建连之间的 TOCTOU 窗口
|
||
至此关闭）；Host 头仍用原始主机名（http.client 按
|
||
self.host 生成），服务端看到的 Host 与用户填的目标一致。"""
|
||
|
||
def __init__(self, host, port=None, pinned_ips=(), timeout=None):
|
||
super().__init__(host, port, timeout=timeout)
|
||
self._pinned_ips = list(pinned_ips)
|
||
|
||
def connect(self):
|
||
last = None
|
||
for ip in self._pinned_ips:
|
||
try:
|
||
self.sock = socket.create_connection(
|
||
(ip, self.port), self.timeout)
|
||
return
|
||
except OSError as exc:
|
||
last = exc
|
||
if last is not None:
|
||
raise last
|
||
raise OSError("no pinned address for %s" % self.host)
|
||
|
||
|
||
class _PinnedHTTPSConnection(_PinnedHTTPConnection,
|
||
http.client.HTTPSConnection):
|
||
"""直连 https 的 pin 连接：先按 IP 建裸 socket，再用
|
||
默认校验上下文 wrap，server_hostname 必须是原始主机名
|
||
（不是 IP），否则 SNI 与证书主机名校验都会错——这是
|
||
“为 pin 而破坏 TLS 校验”的反模式，本实现明确不这么做：
|
||
证书仍按原始主机名校验，IP 只决定 TCP 对端。"""
|
||
|
||
def connect(self):
|
||
_PinnedHTTPConnection.connect(self)
|
||
context = ssl.create_default_context()
|
||
self.sock = context.wrap_socket(self.sock,
|
||
server_hostname=self.host)
|
||
|
||
|
||
def _fetch_single_hop(url, timeout, pinned_ips, proxy):
|
||
"""单跳抓取（不跟随重定向）：返回 (status, location, raw)。
|
||
直连走 pin 连接；代理走 http.client 的代理连接（https 用
|
||
set_tunnel，CPython 按 tunnel 主机名做 SNI，与直连同款
|
||
证书主机名校验）。transport 异常由调用方按 _wrap 处理，
|
||
本函数不把异常文本加工成原因。"""
|
||
parts = urlsplit(url)
|
||
host = (parts.hostname or "").strip().lower().rstrip(".")
|
||
port = parts.port or (443 if parts.scheme == "https" else 80)
|
||
path = parts.path or "/"
|
||
if parts.query:
|
||
path = path + "?" + parts.query
|
||
headers = {"User-Agent": RESIDENT_USER_AGENT}
|
||
if proxy is None:
|
||
if parts.scheme == "https":
|
||
conn = _PinnedHTTPSConnection(host, port, pinned_ips,
|
||
timeout=timeout)
|
||
else:
|
||
conn = _PinnedHTTPConnection(host, port, pinned_ips,
|
||
timeout=timeout)
|
||
conn.request("GET", path, headers=headers)
|
||
elif parts.scheme == "https":
|
||
conn = http.client.HTTPSConnection(
|
||
proxy[0], proxy[1], timeout=timeout,
|
||
context=ssl.create_default_context())
|
||
conn.set_tunnel(host, port)
|
||
conn.request("GET", path, headers=headers)
|
||
else:
|
||
conn = http.client.HTTPConnection(proxy[0], proxy[1],
|
||
timeout=timeout)
|
||
conn.request("GET", url, headers=headers)
|
||
try:
|
||
resp = conn.getresponse()
|
||
raw = resp.read(MAX_FETCH_BYTES + 1)[:MAX_FETCH_BYTES]
|
||
return resp.status, resp.getheader("Location"), raw
|
||
finally:
|
||
try:
|
||
conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _wrap_transport_error(exc, via_proxy):
|
||
"""transport 异常 -> FetchError：代理模式下的连接失败记
|
||
proxy（代理本身不可达/拒绝都算代理侧失败，不算目标站
|
||
失败）；直连按类型分超时/TLS/网络。"""
|
||
if isinstance(exc, FetchError):
|
||
return exc
|
||
if via_proxy and isinstance(exc, (ConnectionError, OSError,
|
||
socket.timeout, TimeoutError)):
|
||
return FetchError("proxy", "proxy connection failed: %s" % exc)
|
||
if isinstance(exc, (socket.timeout, TimeoutError)):
|
||
return FetchError("timeout", "fetch timed out: %s" % exc)
|
||
if isinstance(exc, ssl.SSLError):
|
||
return FetchError("tls", "TLS failed: %s" % exc)
|
||
if isinstance(exc, (ConnectionError, OSError)):
|
||
return FetchError("network", "fetch failed: %s" % exc)
|
||
return FetchError("unknown", "fetch failed: %s" % exc)
|
||
|
||
|
||
def default_fetch(url, timeout, resolver=None, proxy_config=None):
|
||
"""默认廉价抓取：超时可配，读满 1MB 即断，解码失败按
|
||
replace 处理；只在 200 时返回正文，其它状态一律抛
|
||
HttpStatusFetchError（错误页正文直接丢弃，绝不进入
|
||
变化判定与任务摘录）。
|
||
|
||
SSRF 封口（直连模式，完整）：
|
||
1. 每跳先过 validate_target_url 的字面校验；
|
||
2. 再解析全部 A/AAAA 并过 resolve_public_addresses；
|
||
3. 建连只连解析出的 IP（pin），https 的 SNI 与证书主机
|
||
名仍是原始主机名，TLS 校验不被 pin 破坏；
|
||
4. 重定向每跳重复 1-3 步，相对跳转先 urljoin 再校验；
|
||
5. 超过 MAX_REDIRECTS 跳抛 redirect 失败。
|
||
残留风险（诚实写明，不假装已封口）：
|
||
- 代理模式下：客户端的解析 gate 只是前置检查，真正建连
|
||
的是代理方、由代理方二次解析；若代理方解析结果与客
|
||
户端不一致（DNS 变化、代理方走不同解析器），pin 不生
|
||
效，须靠代理自身的 egress allowlist 兜底。生产部署若
|
||
用代理模式，必须同时给代理配内网地址拦截。
|
||
- 解析与建连之间仍有理论 TOCTOU：直连模式下 pin 已把
|
||
窗口关到“解析结果本身”这一层（建连不再二次解析），
|
||
但 DNS 记录在解析后、建连前被改写不影响本次建连目
|
||
标（目标就是解析时刻的 IP），残留的是“解析时刻的
|
||
IP 在之后被重新分配给内网用途”的极端情形，不在本次
|
||
封口承诺内。
|
||
- resolver 参数是测试注入点（与 run_due 的 fetcher 注入
|
||
同款）：生产路径一律不传；传了自定义 resolver 的调用
|
||
方自行承担该解析器的可信责任。
|
||
- 代理配置默认值是直连（见 resident_config）：旧代码跟
|
||
随进程 http_proxy/https_proxy 的行为已改，部署时若宿
|
||
主出口必须走代理，须显式配 CODEX_RESIDENT_PROXY=env
|
||
或代理 URL，否则公网抓取会直连失败（记 proxy/timeout/
|
||
network 失败轮询，不建任务）。
|
||
调用方可注入 fetcher 替换本函数（测试与未来选择器提取
|
||
都走注入点，不改本函数签名；注入的 fetcher 绕过本门，
|
||
生产驱动不用注入）。"""
|
||
if proxy_config is None:
|
||
proxy = resolve_proxy_config()
|
||
elif isinstance(proxy_config, dict):
|
||
proxy = resolve_proxy_config() if proxy_config.get("mode") is None \
|
||
else proxy_config
|
||
if proxy.get("mode") == "url":
|
||
resolve_proxy_config(proxy["url"])
|
||
else:
|
||
proxy = resolve_proxy_config(proxy_config)
|
||
current = url
|
||
for _hop in range(MAX_REDIRECTS + 1):
|
||
try:
|
||
current = validate_target_url(current)
|
||
except ValueError as exc:
|
||
if "blocked" in str(exc):
|
||
raise SSRFBlockedError(str(exc))
|
||
raise FetchError("invalid_url", str(exc))
|
||
parts = urlsplit(current)
|
||
host = (parts.hostname or "").strip().lower().rstrip(".")
|
||
try:
|
||
port = parts.port or (443 if parts.scheme == "https" else 80)
|
||
except ValueError:
|
||
raise FetchError("invalid_url",
|
||
"invalid port in URL: %s" % current)
|
||
ips = resolve_public_addresses(host, port, resolver=resolver)
|
||
target_proxy = _proxy_for_target(host, parts.scheme, proxy)
|
||
try:
|
||
status, location, raw = _fetch_single_hop(
|
||
current, timeout, ips, target_proxy)
|
||
except Exception as exc:
|
||
raise _wrap_transport_error(exc, target_proxy is not None)
|
||
if status in (301, 302, 303, 307, 308):
|
||
if not location:
|
||
raise FetchError(
|
||
"http_status", "redirect without Location")
|
||
current = urljoin(current, location)
|
||
continue
|
||
if status != 200:
|
||
raise HttpStatusFetchError(status)
|
||
return raw.decode("utf-8", errors="replace")
|
||
raise FetchError("redirect", "too many redirects")
|
||
|
||
|
||
def _record_poll(conn, watch_id, project_id, status, observed_hash,
|
||
detail, task_id, now, failure_reason=None):
|
||
"""记轮询行：failure_reason 是结构化失败原因（见
|
||
FAILURE_REASONS），成功/变化类轮询为 None；detail 前缀
|
||
同样写 reason=<原因>（失败类），老读法只看 detail 也不
|
||
丢原因。失败轮询的 detail 只含异常文本，绝不含抓取正文
|
||
（SSRF 拦截时根本没有正文可写，见 default_fetch）。"""
|
||
poll_id = uuid.uuid4().hex
|
||
conn.execute(
|
||
"INSERT INTO resident_polls(id, watch_id, project_id, status,"
|
||
" observed_hash, detail, task_id, created_at, failure_reason)"
|
||
" VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(poll_id, watch_id, project_id, status, observed_hash,
|
||
(detail or "")[:MAX_POLL_DETAIL_LEN], task_id, now,
|
||
failure_reason))
|
||
return {
|
||
"id": poll_id,
|
||
"watch_id": watch_id,
|
||
"project_id": project_id,
|
||
"status": status,
|
||
"observed_hash": observed_hash,
|
||
"detail": (detail or "")[:MAX_POLL_DETAIL_LEN],
|
||
"task_id": task_id,
|
||
"created_at": now,
|
||
"failure_reason": failure_reason,
|
||
}
|
||
|
||
|
||
def _utc_day_start(now):
|
||
"""now 所在 UTC 日的起点 epoch 秒：每日变化任务上限按
|
||
UTC 日切分（不按本地时区，避免时区与夏令时切分歧义）。"""
|
||
import datetime
|
||
dt = datetime.datetime.fromtimestamp(now, tz=datetime.timezone.utc)
|
||
start = dt.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
return start.timestamp()
|
||
|
||
|
||
def _changed_tasks_today(conn, watch_id, now):
|
||
"""本 UTC 日内已建的变化任务数：只数 status='changed'
|
||
且 task_id 非空的轮询行（changed_capped / changed_cooldown /
|
||
changed_quota_paused 都不占名额）。表缺失时按 0 计（热路
|
||
径 fallback 之前已建表，正常路径不会走到这里）。"""
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT COUNT(*) FROM resident_polls WHERE watch_id = ?"
|
||
" AND status = ? AND task_id IS NOT NULL AND created_at >= ?",
|
||
(watch_id, "changed", _utc_day_start(now))).fetchone()
|
||
except Exception:
|
||
return 0
|
||
try:
|
||
return int(row[0])
|
||
except (TypeError, ValueError, IndexError):
|
||
return 0
|
||
|
||
|
||
def prune_old_polls(conn, now=None, retention_days=None):
|
||
"""删过期轮询行：created_at 早于 now - 保留期的行全部删，
|
||
返回删除行数；保留期内（包括当日）的行一律不动。
|
||
本函数自己 commit（批量驱动每批开头调它，失败由调用方
|
||
兜底，不炸批量）。"""
|
||
if now is None:
|
||
now = time.time()
|
||
if retention_days is None:
|
||
retention_days = resident_config()["poll_retention_days"]
|
||
cutoff = now - retention_days * 86400
|
||
cur = conn.execute(
|
||
"DELETE FROM resident_polls WHERE created_at < ?", (cutoff,))
|
||
conn.commit()
|
||
try:
|
||
return int(cur.rowcount)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
|
||
def _create_change_task(conn, watch, excerpt, now):
|
||
"""变化建任务：列补齐口径与 automations.run_due_automation
|
||
同款；任务 prompt = 用户处理提示 + 变化摘录。"""
|
||
existing_tasks = _existing_columns(conn, "tasks")
|
||
for col, ddl in (
|
||
("executor", "TEXT"),
|
||
("model", "TEXT"),
|
||
("effort", "TEXT"),
|
||
("environment_id", "TEXT"),
|
||
("parent_task_id", "TEXT"),
|
||
("coordinator_part", "TEXT")):
|
||
if col not in existing_tasks:
|
||
try:
|
||
conn.execute(
|
||
"ALTER TABLE tasks ADD COLUMN %s %s" % (col, ddl))
|
||
except Exception:
|
||
pass
|
||
task_id = uuid.uuid4().hex
|
||
prompt = ("%s\n\n[常驻监听变化] %s %s\n变化摘录：\n%s"
|
||
% (watch["prompt"], watch["name"], watch["target_url"],
|
||
excerpt))
|
||
conn.execute(
|
||
"INSERT INTO tasks(id, project_id, prompt, status,"
|
||
" idempotency_key, depends_on, priority, created_by,"
|
||
" created_at, updated_at, last_heartbeat, executor,"
|
||
" model, effort, environment_id, parent_task_id,"
|
||
" coordinator_part)"
|
||
" VALUES(?, ?, ?, 'queued', NULL, '[]', 0, ?, ?, ?,"
|
||
" NULL, NULL, NULL, NULL, NULL, NULL, NULL)",
|
||
(task_id, watch["project_id"], prompt, watch["created_by"],
|
||
now, now))
|
||
return task_id
|
||
|
||
|
||
def _guarded_watch_update(conn, watch_id, sets, vals):
|
||
"""终局 UPDATE 带 status='running' 条件：抓取中途被用户
|
||
暂停/结束时，条件不满足、rowcount 为 0，调用方据此回滚
|
||
并放弃本轮（不建任务、不记轮询），而不是把 paused/ended
|
||
行的 next_poll_at 改回未来值。返回是否成功。"""
|
||
cur = conn.execute(
|
||
"UPDATE resident_watches SET %s WHERE id = ? AND status = ?"
|
||
% ", ".join(sets),
|
||
tuple(vals) + (watch_id, RUNNING))
|
||
return cur.rowcount == 1
|
||
|
||
|
||
def run_due_resident_watch(conn, watch_id, fetcher=None, now=None,
|
||
resolver=None):
|
||
"""单条常驻监听的驱动唯一入口：抢单 -> 廉价抓取 -> hash
|
||
比对 -> 记轮询 -> 仅变化时建任务。
|
||
|
||
抢不到返回 None（不存在、已暂停/已结束、未到期、已被
|
||
别的驱动抢走），不记轮询、不建任务。抓取失败记 failed
|
||
（failure_reason 结构化，见 classify_fetch_error）并退避，
|
||
游标不动。首次抓取记 baseline，只建游标。未变化记
|
||
unchanged，不建任务。变化时依次过三道门：每日上限
|
||
（changed_capped）、抖动冷却（changed_cooldown）、配额门
|
||
（changed_quota_paused / changed_quota_check_failed）——
|
||
前两道被拦时游标不推进（变化不丢，冷却/次日仍可报），
|
||
配额耗尽转 paused（paused_reason='quota_exceeded'，游标
|
||
不推进）；全部通过才建 queued 任务并推进游标。
|
||
热路径不调 ensure；抢单时表缺失才 ensure 后重试一次。
|
||
fetcher 为 None 时用 default_fetch（resolver 透传给它，
|
||
生产路径 resolver 为 None 即默认 DNS 门）。"""
|
||
if now is None:
|
||
now = time.time()
|
||
try:
|
||
claimed = claim_due_watch(conn, watch_id, now)
|
||
except Exception as exc:
|
||
if not _is_missing_table_error(exc):
|
||
raise
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
ensure_resident_tables(conn)
|
||
claimed = claim_due_watch(conn, watch_id, now)
|
||
if not claimed:
|
||
return None
|
||
row = conn.execute(
|
||
"SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
|
||
if row is None:
|
||
return None
|
||
watch = watch_to_dict(row, now)
|
||
if watch["status"] != RUNNING:
|
||
# 认领后、抓取前被用户暂停/结束：不抓取、不建任务、
|
||
# 不改行（暂停/结束已把 next_poll_at 清掉，保持原样）。
|
||
return None
|
||
interval = watch["interval_seconds"]
|
||
timeout = watch["fetch_timeout_seconds"]
|
||
if fetcher is not None:
|
||
fetch = fetcher
|
||
else:
|
||
def fetch(url, timeout, _resolver=resolver):
|
||
return default_fetch(url, timeout, resolver=_resolver)
|
||
try:
|
||
text = fetch(watch["target_url"], timeout)
|
||
except Exception as exc:
|
||
reason = classify_fetch_error(exc)
|
||
failures = watch["consecutive_failures"] + 1
|
||
nxt = now + interval + _backoff_seconds(failures)
|
||
if not _guarded_watch_update(
|
||
conn, watch_id,
|
||
("consecutive_failures = ?", "next_poll_at = ?",
|
||
"updated_at = ?"),
|
||
(failures, nxt, now)):
|
||
conn.rollback()
|
||
return None
|
||
detail = "reason=%s; %s" % (reason, str(exc))
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"failed", None, detail,
|
||
None, now, failure_reason=reason)
|
||
conn.commit()
|
||
_auto_audit_row(conn, watch["project_id"],
|
||
"resident_watch_poll_failed",
|
||
"%s: %s" % (watch_id, str(exc)[:200]), now)
|
||
return poll
|
||
if not isinstance(text, str):
|
||
text = str(text)
|
||
observed = _content_hash(text)
|
||
excerpt = text[:MAX_CHANGE_TEXT_LEN]
|
||
old_hash = watch["last_content_hash"]
|
||
if old_hash is None:
|
||
if not _guarded_watch_update(
|
||
conn, watch_id,
|
||
("last_content_hash = ?", "consecutive_failures = 0",
|
||
"next_poll_at = ?", "updated_at = ?"),
|
||
(observed, now + interval, now)):
|
||
conn.rollback()
|
||
return None
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"baseline", observed, "first poll", None, now)
|
||
conn.commit()
|
||
return poll
|
||
if observed == old_hash:
|
||
if not _guarded_watch_update(
|
||
conn, watch_id,
|
||
("consecutive_failures = 0", "next_poll_at = ?",
|
||
"updated_at = ?"),
|
||
(now + interval, now)):
|
||
conn.rollback()
|
||
return None
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"unchanged", observed, "", None, now)
|
||
conn.commit()
|
||
return poll
|
||
# 真实变化，先过成本上限与抖动冷却（拟定值，见模块头）：
|
||
# 两道门被拦时都只记轮询、不建任务、不推进游标（变化不
|
||
# 丢：下次复查内容仍与旧游标不同，冷却过后/次日仍可报），
|
||
# next_poll_at 按正常间隔重算（不是失败退避——这不是失败）。
|
||
cfg = resident_config()
|
||
if _changed_tasks_today(conn, watch_id, now) \
|
||
>= cfg["max_daily_change_tasks"]:
|
||
if not _guarded_watch_update(
|
||
conn, watch_id,
|
||
("consecutive_failures = 0", "next_poll_at = ?",
|
||
"updated_at = ?"),
|
||
(now + interval, now)):
|
||
conn.rollback()
|
||
return None
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"changed_capped", observed,
|
||
"daily change task cap reached", None, now)
|
||
conn.commit()
|
||
return poll
|
||
last_change_at = watch["last_change_at"]
|
||
if last_change_at is not None \
|
||
and now - last_change_at < cfg["change_cooldown_seconds"]:
|
||
if not _guarded_watch_update(
|
||
conn, watch_id,
|
||
("consecutive_failures = 0", "next_poll_at = ?",
|
||
"updated_at = ?"),
|
||
(now + interval, now)):
|
||
conn.rollback()
|
||
return None
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"changed_cooldown", observed,
|
||
"change cooldown active", None, now)
|
||
conn.commit()
|
||
return poll
|
||
# 真实变化：配额门只在这里过。配额门契约是前缀
|
||
# （quota_check_failed: / quota_exceeded: / None），不用
|
||
# 子串判定，以免超限文本含该子串时误判。
|
||
gate = _automation_quota_exceeded(conn, watch["created_by"])
|
||
if gate is not None and not gate.startswith("quota_check_failed"):
|
||
cur = conn.execute(
|
||
"UPDATE resident_watches SET status = ?,"
|
||
" paused_reason = ?, next_poll_at = NULL, updated_at = ?"
|
||
" WHERE id = ? AND status = ?",
|
||
(PAUSED, "quota_exceeded", now, watch_id, RUNNING))
|
||
if cur.rowcount != 1:
|
||
conn.rollback()
|
||
return None
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"changed_quota_paused", observed, gate,
|
||
None, now)
|
||
conn.commit()
|
||
_auto_audit_row(conn, watch["project_id"],
|
||
"resident_watch_paused_quota",
|
||
"%s: %s" % (watch_id, gate[:200]), now)
|
||
return poll
|
||
if gate is not None:
|
||
failures = watch["consecutive_failures"] + 1
|
||
nxt = now + interval + _backoff_seconds(failures)
|
||
if not _guarded_watch_update(
|
||
conn, watch_id,
|
||
("consecutive_failures = ?", "next_poll_at = ?",
|
||
"updated_at = ?"),
|
||
(failures, nxt, now)):
|
||
conn.rollback()
|
||
return None
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"changed_quota_check_failed", observed, gate,
|
||
None, now)
|
||
conn.commit()
|
||
return poll
|
||
# 先受保护地推进游标，再建任务：抓取中途被暂停/结束时，
|
||
# 条件不满足即回滚，本轮不建任务、不记轮询。
|
||
if not _guarded_watch_update(
|
||
conn, watch_id,
|
||
("last_content_hash = ?", "last_change_at = ?",
|
||
"last_change_text = ?", "consecutive_failures = 0",
|
||
"next_poll_at = ?", "updated_at = ?"),
|
||
(observed, now, excerpt, now + interval, now)):
|
||
conn.rollback()
|
||
return None
|
||
task_id = _create_change_task(conn, watch, excerpt, now)
|
||
poll = _record_poll(conn, watch_id, watch["project_id"],
|
||
"changed", observed, excerpt[:MAX_POLL_DETAIL_LEN],
|
||
task_id, now)
|
||
conn.commit()
|
||
_auto_audit_row(conn, watch["project_id"],
|
||
"resident_watch_change_task_created",
|
||
"%s -> %s" % (watch_id, task_id), now)
|
||
return poll
|
||
|
||
|
||
def _is_retryable_lock_error(exc):
|
||
"""可重试的锁错误：PG 死锁/锁不可用（适配层 locked=True）
|
||
与 SQLite 的 database is locked。其它错误重试无用。"""
|
||
if getattr(exc, "locked", False):
|
||
return True
|
||
return "database is locked" in str(exc).lower()
|
||
|
||
|
||
def _run_one_isolated(conn, watch_id, fetcher, now, resolver, errors):
|
||
"""批量内单条隔离：锁错误回滚后重试一次（第一次死锁时
|
||
claim 可能已提交——重试抢不到即干净返回 None；claim 未
|
||
提交时重试可成功），仍失败则记入 errors 并继续下一条，
|
||
绝不让一条死锁 abort 整批剩余监听。非锁异常同样隔离
|
||
（抓取类失败已在 run_due 内记 failed 轮询，不会走到
|
||
这里；走到这里的是 DB/建任务类异常）。"""
|
||
for attempt in range(2):
|
||
try:
|
||
return run_due_resident_watch(conn, watch_id,
|
||
fetcher=fetcher, now=now,
|
||
resolver=resolver)
|
||
except Exception as exc:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
if _is_retryable_lock_error(exc) and attempt == 0:
|
||
continue
|
||
if errors is not None:
|
||
errors.append((watch_id, str(exc)[:200]))
|
||
return None
|
||
return None
|
||
|
||
|
||
def process_due_resident_watches(conn, fetcher=None, now=None, limit=50,
|
||
budget_seconds=None,
|
||
monotonic_fn=None, resolver=None,
|
||
errors=None):
|
||
"""驱动批量入口：列出到期监听，逐条走 run_due（每条自带
|
||
claim 门，批量内重复与并发驱动都不会重复轮询）。返回
|
||
本轮实际产出的轮询记录列表（抢不到的条目不在内）。
|
||
|
||
批量总预算（S2）：budget_seconds 缺省取拟定默认值
|
||
（resident_config，120 秒）；每条开始前检查单调时钟，
|
||
elapsed >= 预算即停——剩余到期监听保持 due 状态，下批
|
||
再跑，不丢。预算不含“正在跑的最后一条”：最坏单批耗时
|
||
= 预算 + 单条超时上限（60 秒），相对旧的 50×60=3000 秒
|
||
已有界；单条超时本身仍由 fetch_timeout_seconds 卡。
|
||
每批开头 prune 一次过期轮询（失败只回滚、不炸批量）。
|
||
errors 为调用方传入的列表时，逐条隔离的异常与预算耗
|
||
尽信号（watch_id=None 的条目）记入其中，供 worker 记
|
||
日志；不传时静默丢弃（旧调用方行为不变）。"""
|
||
if now is None:
|
||
now = time.time()
|
||
if budget_seconds is None:
|
||
budget_seconds = resident_config()["batch_budget_seconds"]
|
||
mono = monotonic_fn or time.monotonic
|
||
started = mono()
|
||
due = _due_watches_with_fallback(conn, now)[:limit]
|
||
try:
|
||
prune_old_polls(conn, now=now)
|
||
except Exception:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
out = []
|
||
for watch in due:
|
||
if budget_seconds is not None \
|
||
and mono() - started >= budget_seconds:
|
||
if errors is not None:
|
||
errors.append(
|
||
(None, "batch_budget_exceeded after %d polls"
|
||
% len(out)))
|
||
break
|
||
poll = _run_one_isolated(conn, watch["id"], fetcher, now,
|
||
resolver, errors)
|
||
if poll is not None:
|
||
out.append(poll)
|
||
return out
|
||
|
||
|
||
def resident_metrics(conn, project_ids, now=None):
|
||
"""resident 可观测口径（供 /metrics 用，只读不建表）：
|
||
返回 dict，表缺失时 {"available": False}（老库未跑过
|
||
init_db/ensure 时不炸 /metrics）。字段：
|
||
- watches_by_status：限定项目内的监听状态计数；
|
||
- due_count：running 且已到期（next_poll_at <= now）的条数；
|
||
- max_overdue_seconds：到期监听的最大逾期秒数（无到期为 0）；
|
||
- polls_last_24h / failed_last_24h / changed_last_24h /
|
||
change_tasks_last_24h：近 24h 轮询、失败、变化、变化任务数；
|
||
- config：当前拟定配置值（见 resident_config）。
|
||
延迟口径说明：due_count 与 max_overdue_seconds 直接度量
|
||
“到期未跑”的积压——驱动被模型派发拖住时，这两项会如
|
||
实变大，不存在“间隔写 60 秒就等于 60 秒送达”的谎言。"""
|
||
if now is None:
|
||
now = time.time()
|
||
ids = list(project_ids or [])
|
||
if not ids:
|
||
return {"available": True, "watches_by_status": {},
|
||
"due_count": 0, "max_overdue_seconds": 0,
|
||
"polls_last_24h": 0, "failed_last_24h": 0,
|
||
"changed_last_24h": 0, "change_tasks_last_24h": 0,
|
||
"config": resident_config()}
|
||
placeholders = ",".join("?" for _ in ids)
|
||
try:
|
||
watch_rows = conn.execute(
|
||
"SELECT status, next_poll_at FROM resident_watches"
|
||
" WHERE project_id IN (%s)" % placeholders,
|
||
tuple(ids)).fetchall()
|
||
except Exception:
|
||
return {"available": False}
|
||
by_status = {}
|
||
due_count = 0
|
||
max_overdue = 0
|
||
for row in watch_rows:
|
||
try:
|
||
status = row["status"]
|
||
nxt = row["next_poll_at"]
|
||
except (KeyError, IndexError, TypeError):
|
||
status, nxt = row[0], row[1]
|
||
by_status[status] = by_status.get(status, 0) + 1
|
||
if status == RUNNING and nxt is not None and nxt <= now:
|
||
due_count += 1
|
||
max_overdue = max(max_overdue, now - nxt)
|
||
try:
|
||
poll_rows = conn.execute(
|
||
"SELECT status, task_id FROM resident_polls"
|
||
" WHERE project_id IN (%s) AND created_at >= ?"
|
||
% placeholders,
|
||
tuple(ids) + (now - 86400,)).fetchall()
|
||
except Exception:
|
||
poll_rows = []
|
||
polls_24h = len(poll_rows)
|
||
failed_24h = 0
|
||
changed_24h = 0
|
||
tasks_24h = 0
|
||
for row in poll_rows:
|
||
try:
|
||
status = row["status"]
|
||
task_id = row["task_id"]
|
||
except (KeyError, IndexError, TypeError):
|
||
status, task_id = row[0], row[1]
|
||
if status == "failed":
|
||
failed_24h += 1
|
||
if status == "changed":
|
||
changed_24h += 1
|
||
if task_id:
|
||
tasks_24h += 1
|
||
return {"available": True, "watches_by_status": by_status,
|
||
"due_count": due_count,
|
||
"max_overdue_seconds": max_overdue,
|
||
"polls_last_24h": polls_24h, "failed_last_24h": failed_24h,
|
||
"changed_last_24h": changed_24h,
|
||
"change_tasks_last_24h": tasks_24h,
|
||
"config": resident_config()}
|