workers: land host/pid/address columns with idempotent migration and pid_verified stale classification
- ensure_workers_table: new tables include host TEXT, pid BIGINT, address TEXT; old tables gain them via _ensure_hostpid_columns (PRAGMA column check, missing-only ALTER, old rows keep NULL, never fabricates host/pid). SQLite and Postgres share this path (PG via the db adapter's PRAGMA translation); POSTGRES_SCHEMA deliberately still excludes workers, which is volatile heartbeat data in UNMIGRATED_TABLES (adding it broke the P2 migration alignment tests, verified and reverted). - heartbeat/register: given host/pid/address refresh, omitted values preserve the old row; address has no derived source. Local worker passes host_name + own pid; HTTP heartbeat only stores caller-supplied values, never the server's own. - classify_stale_workers: kill only when age strictly exceeds the threshold AND host matches local host AND pid is proven dead; equal threshold, foreign host, missing pid, or live/unknown pid are list_only. Classification only, never kills, never matches by name/prefix, never touches infra containers.
This commit is contained in:
1 parent
26630bafec
commit
32e264054c
4 files changed
+524
-24
No files matched your search
+26
-4
@@ -9128,9 +9128,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
|
||||
def _workers_heartbeat(self, user_id):
|
||||
"""worker 心跳:body 含 worker_id(必填,缺了 400)与可选
|
||||
executor/capacity/running。心跳本身就是注册:worker_id
|
||||
不存在时自动注册,绝不 404。running 是 worker 上报的
|
||||
当前占用槽数(启动中占位也算);未给时保留旧值。"""
|
||||
executor/capacity/running/host/pid/address。心跳本身就是
|
||||
注册:worker_id 不存在时自动注册,绝不 404。running 是
|
||||
worker 上报的当前占用槽数(启动中占位也算);未给时保留
|
||||
旧值。host/pid/address 只存调用方显式给的值:远程 worker
|
||||
的 host/pid 绝不由 server 本机推导(server 的主机名/进程
|
||||
号不是该 worker 的),未给时保留旧值。"""
|
||||
body, err = self._read_json()
|
||||
if err:
|
||||
self._send_json(400, {"error": err})
|
||||
@@ -9157,10 +9160,29 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
if running < 0:
|
||||
self._send_json(400, {"error": "running must be >= 0"})
|
||||
return
|
||||
host = body.get("host")
|
||||
if host is not None and not isinstance(host, str):
|
||||
self._send_json(400, {"error": "host must be a string"})
|
||||
return
|
||||
pid = body.get("pid")
|
||||
if pid is not None:
|
||||
try:
|
||||
pid = int(pid)
|
||||
except (TypeError, ValueError):
|
||||
self._send_json(400, {"error": "pid must be an integer"})
|
||||
return
|
||||
if pid <= 0:
|
||||
self._send_json(400, {"error": "pid must be > 0"})
|
||||
return
|
||||
address = body.get("address")
|
||||
if address is not None and not isinstance(address, str):
|
||||
self._send_json(400, {"error": "address must be a string"})
|
||||
return
|
||||
try:
|
||||
worker = _registry_heartbeat(
|
||||
self.server.db_path, worker_id,
|
||||
executor=executor, capacity=capacity, running=running)
|
||||
executor=executor, capacity=capacity, running=running,
|
||||
host=host, pid=pid, address=address)
|
||||
except ValueError as exc:
|
||||
self._send_json(400, {"error": str(exc)})
|
||||
return
|
||||
|
||||
Reference in new issue
Block a user