155 lines
7.5 KiB
Python
155 lines
7.5 KiB
Python
"""Run from Windows Task Scheduler, independently of Codex/terminal lifetime.
|
|
|
|
Supervises this workspace's server only. It does not start a broker terminal,
|
|
change trading authorization, send orders or reset execution identifiers.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from urllib.request import ProxyHandler, build_opener
|
|
|
|
from .config import ROOT
|
|
from .runtime import process_lock, utc_now, write_status
|
|
|
|
|
|
def workspace_id():
|
|
return hashlib.sha256(str(ROOT.resolve()).casefold().encode()).hexdigest()[:16]
|
|
|
|
|
|
def health(port):
|
|
try:
|
|
opener = build_opener(ProxyHandler({}))
|
|
with opener.open(f'http://127.0.0.1:{port}/health', timeout=2) as response:
|
|
result = json.load(response)
|
|
return result if result.get('app') == 'mt5_app' and result.get('workspace_id') == workspace_id() else None
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def run(settings):
|
|
settings.state_dir.mkdir(parents=True, exist_ok=True)
|
|
log_dir = ROOT / 'logs'
|
|
log_dir.mkdir(exist_ok=True)
|
|
log = logging.getLogger('mt5supervisor')
|
|
log.setLevel(logging.INFO)
|
|
handler = RotatingFileHandler(log_dir / 'supervisor.log', maxBytes=2_000_000,
|
|
backupCount=3, encoding='utf-8')
|
|
handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
|
|
log.addHandler(handler)
|
|
if os.environ.get('MT5_LAUNCH_MODE') == 'terminal' and sys.stdout is not None:
|
|
console = logging.StreamHandler(sys.stdout)
|
|
console.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
|
|
log.addHandler(console)
|
|
path = settings.state_dir / 'supervisor.json'
|
|
stop_file = settings.state_dir / 'supervisor.stop'
|
|
state = {'pid': os.getpid(), 'parent_pid': os.getppid(), 'started_at': utc_now(),
|
|
'restarts': 0, 'workspace_id': workspace_id(),
|
|
'launch_mode': os.environ.get('MT5_LAUNCH_MODE', 'windows_task')}
|
|
child = None
|
|
python = Path(sys.executable).with_name('python.exe') if os.name == 'nt' else Path(sys.executable)
|
|
flags = getattr(subprocess, 'CREATE_NO_WINDOW', 0)
|
|
last_publish_warning = -float('inf')
|
|
|
|
def publish(status, **extra):
|
|
nonlocal last_publish_warning
|
|
state.update(state=status, heartbeat_at=utc_now(), **extra)
|
|
try:
|
|
saved = write_status(path, state)
|
|
except Exception:
|
|
# Telemetry/serialization cannot own the server's lifetime.
|
|
if time.monotonic() - last_publish_warning >= 60:
|
|
log.exception('status_publish_failed; keeping server running')
|
|
last_publish_warning = time.monotonic()
|
|
saved = False
|
|
if not saved:
|
|
state['status_write_failures'] = state.get('status_write_failures', 0) + 1
|
|
elif state.get('status_write_failures'):
|
|
log.info('status_write_recovered after_failures=%s', state['status_write_failures'])
|
|
state['status_write_failures'] = 0
|
|
return saved
|
|
|
|
def stop_child(reason):
|
|
if child is None or child.poll() is not None:
|
|
return
|
|
log.warning('stopping_owned_child pid=%s reason=%s', child.pid, reason)
|
|
# A Windows venv launcher has a Python child. Stop the owned process
|
|
# tree, never all python.exe or any terminal64.exe processes.
|
|
if os.name == 'nt':
|
|
result = subprocess.run(['taskkill', '/PID', str(child.pid), '/T', '/F'],
|
|
capture_output=True, timeout=15, creationflags=flags)
|
|
if result.returncode:
|
|
log.error('taskkill_failed returncode=%s output=%r', result.returncode, result.stderr)
|
|
else:
|
|
child.terminate()
|
|
child.wait(timeout=15)
|
|
log.warning('owned_child_stopped pid=%s exit_code=%s reason=%s', child.pid, child.returncode, reason)
|
|
|
|
with process_lock(settings.state_dir / 'supervisor.lock'):
|
|
if stop_file.exists():
|
|
log.info('supervisor_start_suppressed reason=stop_marker')
|
|
publish('stopped', stopped_at=utc_now(), reason='stop_marker')
|
|
return
|
|
log.info('supervisor_started pid=%s parent=%s', os.getpid(), os.getppid())
|
|
publish('starting')
|
|
try:
|
|
while not stop_file.exists():
|
|
if child is None:
|
|
existing = health(settings.port)
|
|
if existing:
|
|
publish('monitoring_existing', server_pid=existing.get('pid'))
|
|
time.sleep(5)
|
|
continue
|
|
environment = dict(os.environ, MT5_SUPERVISED='1', PYTHONIOENCODING='utf-8',
|
|
PYTHONUNBUFFERED='1')
|
|
with (settings.state_dir / 'server.stdout.log').open('ab') as out, \
|
|
(settings.state_dir / 'server.stderr.log').open('ab') as err:
|
|
child = subprocess.Popen([str(python), '-m', 'mt5_app', 'serve'], cwd=ROOT.parent,
|
|
env=environment, stdout=out, stderr=err, creationflags=flags)
|
|
(settings.state_dir / 'server.pid').write_text(str(child.pid), encoding='ascii')
|
|
state['child_started_at'] = utc_now()
|
|
publish('starting_server', launcher_pid=child.pid)
|
|
log.info('server_started launcher_pid=%s', child.pid)
|
|
failures = 0
|
|
code = child.poll()
|
|
if code is not None:
|
|
state['restarts'] += 1
|
|
publish('waiting_restart', last_exit_code=code, last_exit_at=utc_now())
|
|
log.error('server_exited launcher_pid=%s exit_code=%s hex=%s; restart in 15 seconds',
|
|
child.pid, code, hex(code & 0xffffffff))
|
|
child = None
|
|
for _ in range(15):
|
|
if stop_file.exists():
|
|
break
|
|
time.sleep(1)
|
|
continue
|
|
current = health(settings.port)
|
|
failures = 0 if current else failures + 1
|
|
publish('running' if current else 'waiting_health', server_pid=(current or {}).get('pid'),
|
|
health_failures=failures)
|
|
if failures >= 18: # about 90 seconds; keep startup/import time too
|
|
log.error('server_health_timeout launcher_pid=%s', child.pid)
|
|
stop_child('health_timeout')
|
|
time.sleep(5)
|
|
publish('stopping')
|
|
stop_child('supervisor_stop_requested')
|
|
publish('stopped', stopped_at=utc_now(), reason='stop_file')
|
|
log.info('supervisor_stopped reason=stop_file')
|
|
except KeyboardInterrupt:
|
|
publish('stopping', reason='keyboard_interrupt')
|
|
stop_child('keyboard_interrupt')
|
|
publish('stopped', stopped_at=utc_now(), reason='keyboard_interrupt')
|
|
except BaseException as exc:
|
|
log.exception('supervisor_failed')
|
|
publish('failed', stopped_at=utc_now(), error=f'{type(exc).__name__}: {exc}')
|
|
# Never kill a healthy trading server because its observer failed.
|
|
# The independent Windows trigger will restart/adopt supervision.
|
|
log.error('supervisor_exiting_without_stopping_server launcher_pid=%s',
|
|
child.pid if child and child.poll() is None else None)
|
|
raise
|