113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""Local process ownership, heartbeat and shutdown evidence."""
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import time
|
|
import uuid
|
|
|
|
_last_status_warning = {}
|
|
|
|
|
|
def utc_now():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def read_status(path):
|
|
for attempt in range(3):
|
|
try:
|
|
value = json.loads(Path(path).read_text(encoding='utf-8'))
|
|
return value if isinstance(value, dict) else {}
|
|
except FileNotFoundError:
|
|
return {}
|
|
except (OSError, ValueError):
|
|
if attempt < 2:
|
|
time.sleep(.02)
|
|
return {}
|
|
|
|
|
|
def write_status(path, value):
|
|
"""Best-effort telemetry; Windows readers/scanners may briefly block rename.
|
|
|
|
A status-file failure must never terminate a healthy server or supervisor.
|
|
Trading intents remain durable in SQLite, independent of these JSON files.
|
|
"""
|
|
path = Path(path)
|
|
temp = path.with_name(path.name + f'.{os.getpid()}.tmp')
|
|
payload = json.dumps(value, ensure_ascii=False)
|
|
error = None
|
|
try:
|
|
for attempt in range(6):
|
|
try:
|
|
temp.write_text(payload, encoding='utf-8')
|
|
os.replace(temp, path)
|
|
return True
|
|
except OSError as exc:
|
|
error = exc
|
|
if attempt < 5:
|
|
time.sleep(min(.05 * 2 ** attempt, .4))
|
|
key = str(path)
|
|
now = time.monotonic()
|
|
if now - _last_status_warning.get(key, -float('inf')) >= 60:
|
|
_last_status_warning[key] = now
|
|
logging.getLogger('mt5supervisor' if path.name == 'supervisor.json' else 'mt5desk').warning(
|
|
'status_write_failed path=%s; continuing without stopping trading processes: %s', path, error)
|
|
return False
|
|
finally:
|
|
try:
|
|
temp.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
@contextmanager
|
|
def process_lock(path):
|
|
"""OS releases this lock even when the process is killed externally."""
|
|
handle = Path(path).open('a+b')
|
|
try:
|
|
if handle.tell() == 0:
|
|
handle.write(b'0')
|
|
handle.flush()
|
|
handle.seek(0)
|
|
try:
|
|
if os.name == 'nt':
|
|
import msvcrt
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
|
|
else:
|
|
import fcntl
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError as exc:
|
|
raise RuntimeError('Another MT5 desk process already owns ' + str(path)) from exc
|
|
yield
|
|
finally:
|
|
handle.close()
|
|
|
|
|
|
class Runtime:
|
|
def __init__(self, settings, log):
|
|
self.path = settings.state_dir / 'runtime.json'
|
|
self.log = log
|
|
previous = read_status(self.path)
|
|
self.value = {'pid': os.getpid(), 'parent_pid': os.getppid(),
|
|
'instance': uuid.uuid4().hex, 'started_at': utc_now(),
|
|
'supervised': os.environ.get('MT5_SUPERVISED') == '1', 'state': 'starting',
|
|
'launch_mode': os.environ.get('MT5_LAUNCH_MODE', 'direct')}
|
|
if previous.get('state') in ('starting', 'running'):
|
|
self.value['previous_interruption'] = {
|
|
'pid': previous.get('pid'), 'started_at': previous.get('started_at'),
|
|
'last_heartbeat_at': previous.get('heartbeat_at'),
|
|
'reason': 'Process ended without a recorded clean shutdown; cause unknown'}
|
|
log.warning('previous_process_interrupted %s', self.value['previous_interruption'])
|
|
self.beat('starting')
|
|
|
|
def beat(self, state='running', **extra):
|
|
self.value.update(state=state, heartbeat_at=utc_now(), **extra)
|
|
write_status(self.path, self.value)
|
|
|
|
def finish(self, reason, error=None):
|
|
self.beat('failed' if error else 'stopped', stopped_at=utc_now(), reason=reason,
|
|
error=error)
|
|
self.log.info('shutdown pid=%s reason=%s error=%s', os.getpid(), reason, error)
|