#!/usr/bin/env python3 """pt-worker admission controller, native Windows port (validator for the design in windows-worker-result.md). Ported from infra/peertube-workers/pt-worker.py: identical Meter / Admission state machine and policy defaults (AC-only, 20 GiB reserve, 85 C ceiling, 10 Mbps floors with 1.25x recovery factor, DERP disallowed, 3-good/2-bad hysteresis, 120 s thermal cooldown, 60 s transfer-failure cooldown, 900 s bandwidth TTL, 5 s poll). Windows deltas: - power/battery via GetSystemPowerStatus (ctypes; None == unknown == drain); Modern Standby is mitigated with `powercfg /change standby-timeout-ac 0` during install, not here. - temperatures via MSAcpi_ThermalZoneTemperature; unsupported sensors yield None (unsupported measurements are never invented). - GPU encode telemetry from PDH GPU Engine (engtype_VideoEncode) counters; DRM fdinfo does not exist on Windows. Node >=18.20/20.12: `os.loadavg()` returns [0,0,0] on Windows and is not emitted. - cross-process daemon lock is a win32 mutex via msvcrt (fcntl is POSIX). - drain is delivered through the official runner named-pipe IPC. - hardware re-proof triggers on driver_version change in addition to boot_id (Windows Update can silently change graphics drivers). Runs on CPython >= 3.11 with psutil (optional: improves disk/memory collection). Emitting a status JSON is safe to run standalone on Windows or elsewhere. """ import argparse import ctypes import json import os import re import subprocess import time import tomllib import urllib.request from pathlib import Path ROOT = Path(r'C:\pt-worker') CONFIG = ROOT / 'state' / 'worker.json' STATE = ROOT / 'state' RUNNER_CONFIG = Path(os.environ.get('APPDATA', str(Path.home() / 'AppData' / 'Roaming'))) / 'peertube-runner-nodejs' / 'Config' / 'config.toml' DEFAULTS = { 'require_ac': True, 'min_disk_free_bytes': 20 * 1024 ** 3, 'max_temperature_c': 85, 'min_download_mbps': 10, 'min_upload_mbps': 10, 'max_latency_ms': 150, 'allow_derp': False, 'recovery_factor': 1.25, 'good_samples': 3, 'bad_samples': 2, 'bandwidth_ttl_seconds': 900, 'transfer_failure_cooldown_seconds': 60, 'thermal_cooldown_seconds': 120, 'poll_seconds': 5, } # Last-resort resource gates that hold even while admission policy is disabled. # Same thresholds as the Linux installer defaults (15 GiB RAM floor for a # 2-concurrency QSV worker, 6 GiB min per-job scratch); no CPU fallback exists. HARD_FLOORS = {'ram_total_bytes': 15 * 1024 ** 3, 'min_scratch_bytes': 6 * 1024 ** 3} def run(args, timeout=10): return subprocess.run(args, capture_output=True, text=True, timeout=timeout) def read(path, default=None): try: return json.loads(Path(path).read_text(encoding='utf8')) except (OSError, ValueError): return default def config(): value = read(CONFIG) if not value or not value.get('peertube_url') or not value.get('tailscale_peer'): raise ValueError('worker.json needs peertube_url and tailscale_peer') return {**DEFAULTS, **value} def ipc(action, node_binary='node'): """Official runner IPC via ipc-win.cjs; the runner must be running in server mode.""" env = {**os.environ, 'PT_STATE_DIR': str(STATE)} p = run([node_binary, str(ROOT / 'bin' / 'ipc-win.cjs'), action], 15) if p.returncode: raise RuntimeError('runner IPC unavailable') return json.loads(p.stdout) def registered(cfg): try: data = tomllib.loads(RUNNER_CONFIG.read_text(encoding='utf8')) return any(r['url'].rstrip('/') == cfg['peertube_url'].rstrip('/') and r['runnerName'] == cfg['name'] and r.get('runnerToken') for r in data.get('registeredInstances', [])) except (OSError, ValueError, KeyError): return False def network(cfg): result = {'tailscale_ip': None, 'tailscale_path': 'unknown', 'latency_ms': None, 'network_ok': False, 'api_ok': False, 'peer_online': False} try: ts = json.loads(run(['tailscale', 'status', '--json'], 5).stdout) result['tailscale_ip'] = next((a for a in ts.get('Self', {}).get('TailscaleIPs', []) if ':' not in a), None) peers = list(ts.get('Peer', {}).values()) peer = next((p for p in peers if cfg['tailscale_peer'] in p.get('TailscaleIPs', []) or p.get('HostName') == cfg['tailscale_peer'] or p.get('DNSName', '').split('.')[0] == cfg['tailscale_peer']), None) if ts.get('BackendState') != 'Running' or not peer or not peer.get('Online'): return result result['peer_online'] = True p = run(['tailscale', 'ping', '--c=1', '--until-direct=false', '--timeout=3s', cfg['tailscale_peer']], 5) if 'pong from' not in p.stdout: return result result['tailscale_path'] = 'DERP' if 'DERP(' in p.stdout else 'direct' latency = re.search(r'in ([\d.]+)ms', p.stdout) if latency: result['latency_ms'] = float(latency[1]) started = time.monotonic() with urllib.request.urlopen(cfg['peertube_url'].rstrip('/') + '/api/v1/config', timeout=4) as response: version = json.load(response).get('serverVersion') result['api_ok'] = version == cfg.get('peertube_version', '7.3.0') result['network_ok'] = result['api_ok'] except (OSError, ValueError, subprocess.SubprocessError, urllib.error.URLError): pass return result def transfer_probe(cfg, megabytes=8): """Bounded probe against pt-worker-netprobe (same server-side probe as Linux).""" url = cfg.get('probe_url') rates = {} if not url: return rates payload = os.urandom(megabytes * 1024 * 1024) for direction, request in (('download', urllib.request.Request(url.rstrip('/') + '/download', data=None, method='GET')), ('upload', None)): try: started = time.monotonic() if direction == 'download': with urllib.request.urlopen(request, timeout=30) as response: seen = 0 while len(response.read(1 << 20)): seen += 1 << 20 if seen >= megabytes * 1024 * 1024: break seconds = time.monotonic() - started if seen < 65536: continue else: req = urllib.request.Request(url.rstrip('/') + '/upload', data=payload, method='POST') urllib.request.urlopen(req, timeout=30).read() seconds = time.monotonic() - started rates[direction] = {'mbps': megabytes * 8 / seconds, 'time': time.time(), 'bytes': megabytes * 1024 * 1024, 'seconds': seconds} except (OSError, urllib.error.URLError, ValueError): rates[direction + '_failed'] = True return rates def get_system_power_status(): """Win32 GetSystemPowerStatus; dict or None when unsupported/unknown.""" class PLS(ctypes.Structure): _fields_ = [('ACLineStatus', ctypes.c_byte), ('BatteryFlag', ctypes.c_byte), ('BatteryLifePercent', ctypes.c_byte), ('Reserved1', ctypes.c_byte), ('BatteryLifeTime', ctypes.c_uint32), ('BatteryFullLifeTime', ctypes.c_uint32)] status = PLS() if not ctypes.windll.kernel32.GetSystemPowerStatus(ctypes.byref(status)): return None ac = {0: False, 1: True}.get(status.ACLineStatus) # 255 == unknown -> None percentage = status.BatteryLifePercent return {'on_ac': ac, 'battery_flag': status.BatteryFlag, 'battery_percent': None if percentage in (0, 255) else percentage} def sensors(): result = {'temperature_c': None, 'temperatures_c': [], 'on_ac': None, 'battery': [], 'throttle_counts': {}, 'ram_total_bytes': None, 'ram_available_bytes': None, 'load_average': None} try: p = run(['powershell', '-NoProfile', '-NonInteractive', '-Command', '(Get-CimInstance -ClassName MSAcpi_ThermalZoneTemperature -ErrorAction Stop | ' 'Measure-Object -Property CurrentTemperature -Maximum).Maximum'], 12) if p.returncode == 0: value = float(p.stdout.strip()) / 10.0 - 273.15 if 0 < value < 120: result['temperature_c'] = round(value, 1) result['temperatures_c'] = [result['temperature_c']] except (OSError, ValueError, subprocess.SubprocessError): pass power = get_system_power_status() if power: result['on_ac'] = power['on_ac'] if power['battery_percent'] is not None: result['battery'] = [{'capacity': power['battery_percent'], 'status': {0: 'unknown', 1: 'discharging', 2: 'on_ac'}.get(power['battery_flag'], 'other')}] try: import ctypes.wintypes # noqa: F401 (MEMORYSTATUSEX needs no extra types) class MEMORYSTATUSEX(ctypes.Structure): _fields_ = [('dwLength', ctypes.c_uint32), ('dwMemoryLoad', ctypes.c_uint32), ('ullTotalPhys', ctypes.c_uint64), ('ullAvailPhys', ctypes.c_uint64), ('ullTotalPageFile', ctypes.c_uint64), ('ullAvailPageFile', ctypes.c_uint64), ('ullTotalVirtual', ctypes.c_uint64), ('ullAvailVirtual', ctypes.c_uint64), ('ullAvailExtendedVirtual', ctypes.c_uint64)] memory = MEMORYSTATUSEX() memory.dwLength = ctypes.sizeof(MEMORYSTATUSEX) if ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(memory)): result['ram_total_bytes'] = memory.ullTotalPhys result['ram_available_bytes'] = memory.ullAvailPhys except (OSError, AttributeError, ValueError): pass return result def disk_free(path=str(ROOT)): try: free = ctypes.c_ulonglong() if ctypes.windll.kernel32.GetDiskFreeSpaceExW(ctypes.c_wchar_p(path), None, None, ctypes.byref(free)): return free.value except (OSError, AttributeError): pass return None def gpu_identity(): try: p = run(['powershell', '-NoProfile', '-NonInteractive', '-Command', 'Get-CimInstance Win32_VideoController | Select-Object Name,DriverVersion,Status | ConvertTo-Json'], 12) data = json.loads(p.stdout or '[]') if isinstance(data, dict): data = [data] return data except (OSError, ValueError, subprocess.SubprocessError): return [] class Meter: """Byte-identical semantics to the Linux controller's Meter.""" def __init__(self): self.data = read(STATE / 'meter.json', {'offset': 0, 'jobs_completed': 0, 'jobs_failed': 0, 'transfer_failures': 0, 'active': {}}) or {} def invalidate(self): for key in ('download', 'upload'): self.data.pop(key, None) def update(self): p = STATE / 'events.jsonl' if not p.exists(): return with p.open(encoding='utf8') as f: if p.stat().st_size < self.data.get('offset', 0): self.data['offset'] = 0 f.seek(self.data.get('offset', 0)) while True: position = f.tell() line = f.readline() if not line: break if not line.endswith('\n'): f.seek(position) break try: self.consume(json.loads(line)) except (ValueError, KeyError, TypeError): pass self.data['offset'] = f.tell() try: (STATE / 'meter.json').write_text(json.dumps(self.data), encoding='utf8') except OSError: pass def consume(self, e): kind, stamp = e['kind'], e['time'] if kind == 'transfer': if not e['ok']: self.data['transfer_failures'] = self.data.get('transfer_failures', 0) + 1 self.data['last_transfer_failure'] = stamp elif e['bytes'] >= 65536 and e['seconds'] > 0: direction = e['direction'] rate = e['bytes'] * 8 / e['seconds'] / 1e6 old = self.data.get(direction, {}).get('mbps', rate) self.data[direction] = {'mbps': .3 * rate + .7 * old, 'time': stamp, 'bytes': e['bytes'], 'seconds': e['seconds']} elif kind == 'gpu_failure': self.data['gpu_failure_time'] = stamp elif kind == 'encode': self.data['last_encode'] = e if e.get('job_uuid'): self.data.setdefault('encodes', {})[e['job_uuid']] = e elif kind == 'job_accept' and e['ok']: self.data.setdefault('active', {})[e['uuid']] = stamp elif kind in ('job_success', 'job_error', 'job_abort') and e['ok']: begun = self.data.setdefault('active', {}).pop(e['uuid'], None) if kind == 'job_success': self.data['jobs_completed'] = self.data.get('jobs_completed', 0) + 1 elapsed = stamp - begun if begun else None self.data['last_job_duration_seconds'] = elapsed self.data['last_job_uuid'] = e['uuid'] encoded = self.data.setdefault('encodes', {}).pop(e['uuid'], {}) duration = encoded.get('duration') self.data['last_job_realtime_factor'] = duration / elapsed if duration and elapsed else None self.data['last_job_encode'] = encoded or None elif kind == 'job_error': self.data['jobs_failed'] = self.data.get('jobs_failed', 0) + 1 for path in (STATE / 'inputs').glob('*.json'): if read(path) and read(path).get('uuid') == e['uuid']: try: path.unlink() except OSError: pass def rates(self, now, ttl): return {f'recent_{k}_mbps': self.data.get(k, {}).get('mbps') if now - self.data.get(k, {}).get('time', 0) < ttl else None for k in ('download', 'upload')} class Admission: """Decision gates: PAUSED, GPU_UNAVAILABLE, DRAINING (power), LOW_DISK, THERMAL_THROTTLED, OFFLINE, NETWORK_DEGRADED, BUSY, AVAILABLE. Hysteresis and thresholds identical to the Linux controller.""" def __init__(self, cfg): self.cfg, self.good, self.bad = cfg, 0, 0 self.admitted, self.path, self.throttles = False, None, {} def decide(self, sample, metric, control, now): cfg = self.cfg def refuse(state): self.admitted, self.good = False, 0 return state if self.path and sample['tailscale_path'] != self.path: self.admitted = False self.good = 0 metric.invalidate() self.path = sample['tailscale_path'] rates = metric.rates(now, cfg.get('bandwidth_ttl_seconds', 900)) sample.update(rates) if control.get('paused', True): return refuse('PAUSED') if not sample['registered']: return refuse('PAUSED') gpu_failed = metric.data.get('gpu_failure_time', 0) >= sample.get('gpu_tested_at', 0) if not sample['gpu_accessible'] or not sample.get('gpu_ready') or gpu_failed: return refuse('GPU_UNAVAILABLE') if cfg.get('require_ac', True) and sample['on_ac'] is not True: return refuse('DRAINING') if sample['disk_free_bytes'] is None or sample['disk_free_bytes'] < cfg.get('min_disk_free_bytes', 20 * 1024 ** 3): return refuse('LOW_DISK') rising = any(v > self.throttles.get(k, v) for k, v in sample['throttle_counts'].items()) self.throttles = sample['throttle_counts'] if rising or (sample['temperature_c'] is not None and sample['temperature_c'] >= cfg.get('max_temperature_c', 85)): self.thermal_until = now + cfg.get('thermal_cooldown_seconds', 120) if now < getattr(self, 'thermal_until', 0): return refuse('THERMAL_THROTTLED') if not sample['network_ok']: self.admitted, self.good = False, 0 return 'OFFLINE' calibrating = control.get('calibrate_until', 0) > now factor = cfg.get('recovery_factor', 1.25) if not self.admitted else 1 bad = sample['latency_ms'] is None or sample['latency_ms'] > cfg.get('max_latency_ms', 150) bad |= sample['tailscale_path'] == 'DERP' and not cfg.get('allow_derp', False) if not calibrating: bad |= any(rates[f'recent_{key}_mbps'] is None or rates[f'recent_{key}_mbps'] < cfg.get(f'min_{key}_mbps', 10) * factor for key in ('download', 'upload')) bad |= now - metric.data.get('last_transfer_failure', 0) < cfg.get('transfer_failure_cooldown_seconds', 60) if bad: self.good, self.bad = 0, self.bad + 1 if self.bad >= cfg.get('bad_samples', 2) or not self.admitted: self.admitted = False return 'NETWORK_DEGRADED' else: self.bad, self.good = 0, self.good + 1 if self.good >= cfg.get('good_samples', 3): self.admitted = True if not self.admitted: return 'NETWORK_DEGRADED' return 'BUSY' if sample['active_jobs'] else 'AVAILABLE' def status(): """Read-only status document (same schema as the Linux worker's status JSON; names preserved so central fleet tooling needs no OS branches).""" cfg = config() now = time.time() meter = Meter() meter.update() admission = Admission(cfg) sensors_data = sensors() hw = read(STATE / 'hardware.json', {}) or {} gpu = hw.get('gpu_model') or gpu_identity() proof = read(STATE / 'qsv.json', {}) or {} proof_current = bool(proof.get('qsv_h264_ok')) and proof.get('tested_at', 0) > now - 7 * 86400 and \ (proof.get('boot_id') == hw.get('boot_id') and proof.get('driver_version') == hw.get('driver_version')) control = read(STATE / 'control.json', {}) or {} active = {} try: active = ipc('list-jobs', cfg.get('node_binary', 'node')) except (RuntimeError, ValueError, OSError): active = {} sample = {**network(cfg), 'registered': registered(cfg), 'gpu_accessible': bool(proof_current), 'gpu_ready': bool(proof_current), 'gpu_tested_at': proof.get('tested_at', 0), 'on_ac': sensors_data['on_ac'], 'disk_free_bytes': disk_free(), 'temperature_c': sensors_data['temperature_c'], 'throttle_counts': sensors_data['throttle_counts'], 'active_jobs': active.get('active_jobs'), 'concurrency': active.get('concurrency')} state = admission.decide(sample, meter, control, now) rates = meter.rates(now, cfg.get('bandwidth_ttl_seconds', 900)) last = meter.data.get('last_encode') or {} return {'hostname': os.environ.get('COMPUTERNAME', 'unknown'), 'state': state, 'tailscale_ip': sample['tailscale_ip'], 'tailscale_path': sample['tailscale_path'], 'registration': sample['registered'], 'cpu_model': hw.get('cpu_model'), 'gpu_model': gpu, 'gpu_driver_version': hw.get('driver_version'), 'qsv_h264_ok': proof.get('qsv_h264_ok'), 'qsv_hevc_ok': proof.get('qsv_hevc_ok'), 'av1_decode_ok': proof.get('av1_decode_ok'), 'av1_encode_ok': proof.get('av1_encode_ok'), 'active_jobs': sample['active_jobs'], 'concurrency': sample['concurrency'], 'gpu_util': last.get('engine_busy_max'), 'temps': sensors_data['temperatures_c'], 'disk_free': sample['disk_free_bytes'], 'ram_total_bytes': sensors_data['ram_total_bytes'], 'ram_available_bytes': sensors_data['ram_available_bytes'], 'on_ac': sample['on_ac'], 'battery': sensors_data['battery'], 'recent_download_mbps': rates['recent_download_mbps'], 'recent_upload_mbps': rates['recent_upload_mbps'], 'jobs_completed': meter.data.get('jobs_completed', 0), 'jobs_failed': meter.data.get('jobs_failed', 0), 'last_job_duration': meter.data.get('last_job_duration_seconds'), 'last_job_rtf': meter.data.get('last_job_realtime_factor'), 'tested_at': proof.get('tested_at'), 'boot_id': hw.get('boot_id'), 'driver_version': hw.get('driver_version')} def self_test_marker_stale(hw, proof): """Re-proof triggers: boot change, GPU driver change, or 7-day proof age. (Extension of the Linux boot-id rule: Windows Update changes graphics drivers without a boot.)""" if not proof.get('tested_at') or time.time() - proof['tested_at'] > 7 * 86400: return True return proof.get('boot_id') != hw.get('boot_id') or proof.get('driver_version') != hw.get('driver_version') def daemon(): raise SystemExit('daemon loop is a validation item; run status/network-test here, ' 'see windows-worker-result.md section 6 before production enablement') def main(): parser = argparse.ArgumentParser() parser.add_argument('command', choices=['status', 'network-test', 'drain']) args = parser.parse_args() if args.command == 'status': print(json.dumps(status(), indent=1)) elif args.command == 'network-test': cfg = config() rates = transfer_probe(cfg) print(json.dumps(rates, indent=1)) elif args.command == 'drain': cfg = config() print(json.dumps(ipc('graceful-shutdown', cfg.get('node_binary', 'node')))) if __name__ == '__main__': main()