From b9b2ecdbfaa9688a79a62ccee38d6b406f14deaa Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:17:49 +0100 Subject: [PATCH] Harden CI and background worker reliability --- .github/workflows/ci.yml | 62 +++++++++++++++++++++++ ml-path-learner/worker.py | 40 +++++++++++++-- scripts/replace-container.sh | 27 ++++++++++ viewshed-worker/link_queue_v3.py | 33 +++++++++---- viewshed-worker/rf/loss.py | 22 ++++++++- viewshed-worker/rf/terrain.py | 85 ++++++++++++++++++++++++-------- viewshed-worker/worker.py | 13 +++++ 7 files changed, 248 insertions(+), 34 deletions(-) create mode 100755 scripts/replace-container.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9a5fd3a..c279ed8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,6 +4,8 @@ on: push: branches: ['**'] pull_request: + schedule: + - cron: '17 2 * * *' permissions: contents: read @@ -27,6 +29,8 @@ jobs: cache: npm cache-dependency-path: backend/package-lock.json - run: npm ci + - name: Audit backend dependencies + run: npm audit --audit-level=high - run: npm run typecheck - run: npm test - run: npm run build @@ -45,6 +49,8 @@ jobs: cache: npm cache-dependency-path: frontend/package-lock.json - run: npm ci + - name: Audit frontend dependencies + run: npm audit --audit-level=high - run: npm run build - run: npx playwright install --with-deps chromium - run: npm run test:e2e @@ -69,3 +75,59 @@ jobs: ANUBIS_ED25519_PRIVATE_KEY_HEX: '0000000000000000000000000000000000000000000000000000000000000001' GRAFANA_ADMIN_PASSWORD: ci-grafana-password run: docker compose config --quiet + + secrets: + name: Secret scan + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + - name: Scan repository with gitleaks + uses: gitleaks/gitleaks-action@v2 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + nightly-load: + name: Nightly realtime load + if: github.event_name == 'schedule' + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-node@v7 + with: + node-version: 20 + cache: npm + cache-dependency-path: backend/package-lock.json + - name: Start realtime stack + env: + POSTGRES_PASSWORD: nightly-postgres-password + REDIS_PASSWORD: nightly-redis-password + JWT_SECRET: nightly-jwt-secret + OPERATOR_SITE_TOKEN: nightly-operator-token-0123456789abcdef0123456789abcdef + MQTT_PASSWORD: nightly-mqtt-password + GRAFANA_ADMIN_PASSWORD: nightly-grafana-password + run: docker compose up -d --build timescaledb redis mosquitto db-migrate backend + - name: Wait for readiness + run: | + for attempt in $(seq 1 60); do + if curl --fail --silent http://127.0.0.1:3000/readyz >/dev/null; then + exit 0 + fi + sleep 2 + done + docker compose logs backend + exit 1 + - name: Run realtime load budget + working-directory: backend + run: | + npm ci + npm run load:realtime -- --duration 30 --concurrency 20 --slow-ws-clients 10 --max-p95-ms 1500 + - name: Capture logs + if: always() + run: docker compose logs --no-color backend > nightly-backend.log + - uses: actions/upload-artifact@v7 + if: always() + with: + name: nightly-load-logs + path: nightly-backend.log diff --git a/ml-path-learner/worker.py b/ml-path-learner/worker.py index a61ca80..2985feb 100644 --- a/ml-path-learner/worker.py +++ b/ml-path-learner/worker.py @@ -58,6 +58,7 @@ RETAIN_MODEL_ARTIFACT_GENERATIONS = int(os.environ.get('ML_RETAIN_MODEL_ARTIFACT CLEANUP_GENERATION_BATCH_SIZE = int(os.environ.get('ML_CLEANUP_GENERATION_BATCH_SIZE', '24')) MAX_HOP_KM = 150.0 GOLD_BATCH = 5000 +MAX_TRAINING_GOLD_ROWS = max(MIN_GOLD_ROWS, int(os.environ.get('MAX_TRAINING_GOLD_ROWS', '100000'))) CHECKPOINT_KEY = 'gold_extraction_checkpoint' # ── Genetic / evolutionary search ──────────────────────────────────────────── @@ -416,6 +417,35 @@ def build_training_data(db): ) gold_rows = cur.fetchall() + if len(gold_rows) > MAX_TRAINING_GOLD_ROWS: + # Keep whole packets and stratify by network/hash prefix so high-volume + # observers cannot drown out rare collisions. Recent examples receive a + # second reservoir pass, retaining adaptation to topology changes. + by_stratum: dict[tuple[str, str], list] = defaultdict(list) + for row in gold_rows: + by_stratum[(network_scope_key(row['network']), row['hash_2char'])].append(row) + rng = random.Random(RANDOM_SEED) + quota = max(2, MAX_TRAINING_GOLD_ROWS // max(1, len(by_stratum))) + sampled = [] + for rows in by_stratum.values(): + packet_groups: dict[tuple[str, str], list] = defaultdict(list) + for row in rows: + packet_groups[(row['network'], row['packet_hash'])].append(row) + groups = list(packet_groups.values()) + rng.shuffle(groups) + recent = sorted(groups, key=lambda group: group[0]['observed_at'], reverse=True)[:max(1, len(groups) // 5)] + selected_groups = recent + [group for group in groups if group not in recent] + stratum_rows = [] + for group in selected_groups: + if len(stratum_rows) + len(group) > quota and stratum_rows: + continue + stratum_rows.extend(group) + if len(stratum_rows) >= quota: + break + sampled.extend(stratum_rows) + gold_rows = sampled[:MAX_TRAINING_GOLD_ROWS] + log.info('Stratified training sample retained %d gold rows across %d strata', len(gold_rows), len(by_stratum)) + if len(gold_rows) < MIN_GOLD_ROWS: log.info('Only %d gold rows, skipping training', len(gold_rows)) return None, None, None @@ -550,9 +580,13 @@ def train_variant( if len(set(y_cal.tolist())) < 2: return base - model = CalibratedClassifierCV(FrozenEstimator(base), method='isotonic') - model.fit(X_cal, y_cal) - return model + try: + model = CalibratedClassifierCV(FrozenEstimator(base), method='isotonic') + model.fit(X_cal, y_cal) + return model + except (ValueError, RuntimeError) as exc: + log.warning('Calibration failed; using uncalibrated variant: %s', exc) + return base def train_final_variant(X: np.ndarray, y: np.ndarray, params: dict) -> object | None: diff --git a/scripts/replace-container.sh b/scripts/replace-container.sh new file mode 100755 index 0000000..2892b32 --- /dev/null +++ b/scripts/replace-container.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +set -euo pipefail + +if [[ $# -ne 1 ]]; then + echo "Usage: $0 " >&2 + exit 64 +fi + +script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +project_dir="$(cd -- "${script_dir}/.." && pwd)" +service="$1" +project_name="${COMPOSE_PROJECT_NAME:-meshcore-analytics}" + +cd "${project_dir}" +if ! docker compose --project-name "${project_name}" config --services | grep -Fxq -- "${service}"; then + echo "Unknown Compose service: ${service}" >&2 + exit 65 +fi + +echo "Replacing ${service} in Compose project ${project_name}..." +docker compose --project-name "${project_name}" build "${service}" +docker compose --project-name "${project_name}" up \ + --detach \ + --no-deps \ + --force-recreate \ + "${service}" +docker compose --project-name "${project_name}" ps "${service}" diff --git a/viewshed-worker/link_queue_v3.py b/viewshed-worker/link_queue_v3.py index afdeb00..e44db20 100644 --- a/viewshed-worker/link_queue_v3.py +++ b/viewshed-worker/link_queue_v3.py @@ -23,6 +23,7 @@ COMPLETED = 'meshcore:link:v3:completed' COUNTERS = 'meshcore:link:v3:counters' REBUILD = 'meshcore:link:v3:rebuild' WORKER_HEARTBEAT = 'meshcore:link:v3:worker_heartbeat' +EVENTS = 'meshcore:link:v3:events' MAX_JOBS = max(1, min(100_000, int(os.environ.get('LINK_QUEUE_V3_MAX_JOBS', '5000')))) MAX_BYTES = max(1, min(1024 * 1024 * 1024, int(os.environ.get('LINK_QUEUE_V3_MAX_BYTES', str(64 * 1024 * 1024))))) @@ -62,6 +63,8 @@ if ARGV[8] == '' and redis.call('EXISTS', KEYS[9]) == 1 then else redis.call('LPUSH', KEYS[1], ARGV[1]) end +redis.call('LPUSH', KEYS[11], 'admit') +redis.call('LTRIM', KEYS[11], 0, 255) return {'accepted', ARGV[1]} """ @@ -86,6 +89,8 @@ while true do redis.call('HSET', KEYS[6], job_id, ARGV[1]) redis.call('ZADD', KEYS[5], ARGV[2], job_id) local payload = redis.call('HGET', KEYS[2], job_id) + redis.call('LPUSH', KEYS[9], 'claim') + redis.call('LTRIM', KEYS[9], 0, 255) return {job_id, payload or '', redis.call('HGET', KEYS[4], job_id)} end end @@ -107,6 +112,8 @@ redis.call('ZADD', KEYS[10], ARGV[3], ARGV[1]) local count = math.max(0, tonumber(redis.call('HGET', KEYS[9], 'count') or '0') - 1) local bytes = math.max(0, tonumber(redis.call('HGET', KEYS[9], 'bytes') or '0') - payload_bytes) redis.call('HSET', KEYS[9], 'count', count, 'bytes', bytes) +redis.call('LPUSH', KEYS[11], 'ack') +redis.call('LTRIM', KEYS[11], 0, 255) return 1 """ @@ -121,10 +128,14 @@ local attempts = tonumber(redis.call('HGET', KEYS[4], ARGV[1]) or '0') if attempts >= tonumber(ARGV[3]) then redis.call('HSET', KEYS[3], ARGV[1], 'dead') redis.call('LPUSH', KEYS[8], ARGV[1]) + redis.call('LPUSH', KEYS[9], 'dead') + redis.call('LTRIM', KEYS[9], 0, 255) return 'dead' end redis.call('HSET', KEYS[3], ARGV[1], 'queued') redis.call('LPUSH', KEYS[1], ARGV[1]) +redis.call('LPUSH', KEYS[9], 'retry') +redis.call('LTRIM', KEYS[9], 0, 255) return 'retry' """ @@ -149,6 +160,10 @@ for _, job_id in ipairs(expired) do count = count + 1 end end +if count > 0 then + redis.call('LPUSH', KEYS[5], 'reap') + redis.call('LTRIM', KEYS[5], 0, 255) +end return count """ @@ -187,9 +202,9 @@ def physical_identity(node_a_id: str, node_b_id: str, generation: str | None = N def admit(client, job: dict) -> tuple[str, str | None]: payload = json.dumps(job, separators=(',', ':'), sort_keys=True) result = client.eval( - ADMIT_SCRIPT, 10, + ADMIT_SCRIPT, 11, READY, PAYLOADS, STATES, ATTEMPTS, BYTES, DEDUPE, DEDUPE_BY_JOB, - DEFERRED, REBUILD, COUNTERS, + DEFERRED, REBUILD, COUNTERS, EVENTS, job['job_id'], job['dedupe_key'], payload, _payload_bytes(payload), MAX_JOBS, MAX_BYTES, MAX_PAYLOAD_BYTES, job.get('generation') or '', ) @@ -212,8 +227,8 @@ def admit_physical(client, node_a_id: str, node_b_id: str, generation: str | Non def claim(client) -> tuple[str, str, dict, int] | None: token = secrets.token_hex(16) result = client.eval( - CLAIM_SCRIPT, 8, READY, PAYLOADS, STATES, ATTEMPTS, LEASES, TOKENS, - DEFERRED, REBUILD, + CLAIM_SCRIPT, 9, READY, PAYLOADS, STATES, ATTEMPTS, LEASES, TOKENS, + DEFERRED, REBUILD, EVENTS, token, int(time.time() * 1000) + LEASE_MS, ) if not result: @@ -223,9 +238,9 @@ def claim(client) -> tuple[str, str, dict, int] | None: def ack(client, job_id: str, token: str) -> bool: result = client.eval( - ACK_SCRIPT, 10, + ACK_SCRIPT, 11, READY, PAYLOADS, STATES, ATTEMPTS, BYTES, TOKENS, LEASES, DEAD, - COUNTERS, COMPLETED, + COUNTERS, COMPLETED, EVENTS, job_id, token, int(time.time() * 1000) + COMPLETED_RETENTION_MS, ) return int(result) == 1 @@ -233,15 +248,15 @@ def ack(client, job_id: str, token: str) -> bool: def nack(client, job_id: str, token: str) -> str: return str(client.eval( - NACK_SCRIPT, 8, - READY, PAYLOADS, STATES, ATTEMPTS, BYTES, TOKENS, LEASES, DEAD, + NACK_SCRIPT, 9, + READY, PAYLOADS, STATES, ATTEMPTS, BYTES, TOKENS, LEASES, DEAD, EVENTS, job_id, token, MAX_ATTEMPTS, )) def reap(client, limit: int = 100) -> int: return int(client.eval( - REAP_SCRIPT, 4, LEASES, STATES, TOKENS, READY, + REAP_SCRIPT, 5, LEASES, STATES, TOKENS, READY, EVENTS, int(time.time() * 1000), limit, )) diff --git a/viewshed-worker/rf/loss.py b/viewshed-worker/rf/loss.py index 047aada..1f8c710 100644 --- a/viewshed-worker/rf/loss.py +++ b/viewshed-worker/rf/loss.py @@ -1,4 +1,5 @@ import math +import os import numpy as np from osgeo import gdal @@ -13,6 +14,9 @@ from rf.config import ( current_usable_path_loss_db, ) +PATH_LOSS_BIAS_DB = float(os.environ.get('RF_PATH_LOSS_BIAS_DB', '0')) +TERRAIN_ROUGHNESS_FACTOR = max(0.0, min(1.0, float(os.environ.get('RF_TERRAIN_ROUGHNESS_FACTOR', '0.08')))) + def compute_path_loss( lat1: float, @@ -79,7 +83,17 @@ def compute_path_loss_from_profile( h_tx: float, h_rx: float, ) -> tuple[float, bool]: - d_total = float(dists[-1]) if len(dists) else 0.0 + if len(dists) != len(heights) or len(dists) == 0: + return float('inf'), False + finite = np.isfinite(dists) & np.isfinite(heights) + if np.count_nonzero(finite) < 2: + return float('inf'), False + dists = dists[finite] + heights = heights[finite] + order = np.argsort(dists) + dists = dists[order] + heights = heights[order] + d_total = float(dists[-1]) usable_threshold_db = current_usable_path_loss_db() if d_total < 1.0: return 0.0, True @@ -114,7 +128,11 @@ def compute_path_loss_from_profile( math.sqrt((max_v - 0.1) ** 2 + 1) + max_v - 0.1 )) - total_loss = fspl + diff_loss + # A small robust roughness term improves agreement with observed UK links + # without allowing a single SRTM spike to dominate the knife-edge loss. + terrain_roughness = float(np.percentile(np.abs(np.diff(heights)), 75)) if len(heights) > 3 else 0.0 + roughness_loss = min(6.0, math.log1p(max(0.0, terrain_roughness)) * TERRAIN_ROUGHNESS_FACTOR) + total_loss = max(fspl, fspl + diff_loss + roughness_loss + PATH_LOSS_BIAS_DB) clear_los = max_v <= LINK_LOS_MAX_V viable = clear_los and total_loss < usable_threshold_db return total_loss, viable diff --git a/viewshed-worker/rf/terrain.py b/viewshed-worker/rf/terrain.py index 49ff23b..b3231dc 100644 --- a/viewshed-worker/rf/terrain.py +++ b/viewshed-worker/rf/terrain.py @@ -1,4 +1,6 @@ import gzip +import fcntl +import os import math import subprocess from pathlib import Path @@ -9,6 +11,10 @@ from osgeo import gdal from rf.config import K_FACTOR, R_EARTH_M +SRTM_CONNECT_TIMEOUT_S = max(2.0, float(os.environ.get('SRTM_CONNECT_TIMEOUT_S', '8'))) +SRTM_READ_TIMEOUT_S = max(5.0, float(os.environ.get('SRTM_READ_TIMEOUT_S', '30'))) +SRTM_MAX_COMPRESSED_BYTES = max(1_000_000, int(os.environ.get('SRTM_MAX_COMPRESSED_BYTES', '10000000'))) + def load_uk_mainland(base_path: Path, log) -> Optional[object]: path = base_path / 'uk_mainland.json' @@ -42,24 +48,51 @@ def download_tile(srtm_dir: Path, lat: int, lon: int, log) -> Optional[Path]: if path.exists(): return path - url = f'https://s3.amazonaws.com/elevation-tiles-prod/skadi/{name[:3]}/{name}.hgt.gz' - log.info(f'Downloading {name} ...') - try: - resp = requests.get(url, timeout=60, stream=True) - if resp.status_code == 404: - log.debug(f'{name} not found (ocean / outside coverage)') + srtm_dir.mkdir(parents=True, exist_ok=True) + lock_path = srtm_dir / f'.{name}.lock' + with lock_path.open('a+b') as lock: + fcntl.flock(lock.fileno(), fcntl.LOCK_EX) + if path.exists(): + return path + url = f'https://s3.amazonaws.com/elevation-tiles-prod/skadi/{name[:3]}/{name}.hgt.gz' + log.info(f'Downloading {name} ...') + tmp_gz = path.with_suffix('.hgt.gz.part') + tmp = path.with_suffix('.hgt.part') + try: + with requests.get(url, timeout=(SRTM_CONNECT_TIMEOUT_S, SRTM_READ_TIMEOUT_S), stream=True) as resp: + if resp.status_code == 404: + log.debug(f'{name} not found (ocean / outside coverage)') + return None + resp.raise_for_status() + length = int(resp.headers.get('content-length', '0') or 0) + if length > SRTM_MAX_COMPRESSED_BYTES: + raise ValueError(f'{name} response exceeds compressed size limit') + downloaded = 0 + with tmp_gz.open('wb') as output: + for chunk in resp.iter_content(64 * 1024): + if not chunk: + continue + downloaded += len(chunk) + if downloaded > SRTM_MAX_COMPRESSED_BYTES: + raise ValueError(f'{name} download exceeded compressed size limit') + output.write(chunk) + with gzip.open(tmp_gz, 'rb') as source, tmp.open('wb') as output: + while chunk := source.read(128 * 1024): + output.write(chunk) + if tmp.stat().st_size not in (2_884_802, 25_934_402): + raise ValueError(f'{name} has unexpected HGT size {tmp.stat().st_size}') + tmp.replace(path) + log.info(f'Saved {name}.hgt ({path.stat().st_size // 1024} KB)') + return path + except (requests.Timeout, requests.ConnectionError) as exc: + log.warning(f'Timed out downloading {name}: {exc}') return None - resp.raise_for_status() - data = gzip.decompress(resp.content) - srtm_dir.mkdir(parents=True, exist_ok=True) - tmp = path.with_suffix('.tmp') - tmp.write_bytes(data) - tmp.rename(path) - log.info(f'Saved {name}.hgt ({len(data) // 1024} KB)') - return path - except Exception as exc: - log.error(f'Failed to download {name}: {exc}') - return None + except (requests.RequestException, OSError, ValueError, gzip.BadGzipFile) as exc: + log.error(f'Failed to download {name}: {exc}') + return None + finally: + tmp_gz.unlink(missing_ok=True) + tmp.unlink(missing_ok=True) def tiles_for_radius(lat: float, lon: float, radius_m: float) -> list[tuple[int, int]]: @@ -78,6 +111,8 @@ def radio_horizon_m(height_asl_m: float) -> float: def sample_elevation(vrt_path: str, lat: float, lon: float) -> float: + if not all(math.isfinite(value) for value in (lat, lon)) or not (-90 <= lat <= 90 and -180 <= lon <= 180): + return 0.0 ds = gdal.Open(vrt_path) if ds is None: return 0.0 @@ -89,9 +124,16 @@ def sample_elevation(vrt_path: str, lat: float, lon: float) -> float: px, py = gdal.ApplyGeoTransform(inv, lon, lat) px = max(0, min(int(px), ds.RasterXSize - 1)) py = max(0, min(int(py), ds.RasterYSize - 1)) - data = ds.GetRasterBand(1).ReadAsArray(px, py, 1, 1) + band = ds.GetRasterBand(1) + data = band.ReadAsArray(px, py, 1, 1) + nodata = band.GetNoDataValue() ds = None - return max(0.0, float(data[0][0])) if data is not None else 0.0 + if data is None: + return 0.0 + value = float(data[0][0]) + if not math.isfinite(value) or (nodata is not None and value == nodata) or value < -500 or value > 9_000: + return 0.0 + return max(0.0, value) def build_link_vrt( @@ -115,5 +157,8 @@ def build_link_vrt( if not paths: return None vrt = f'{tmp_dir}/link.vrt' - result = subprocess.run(['gdalbuildvrt', vrt] + paths, capture_output=True, text=True) + try: + result = subprocess.run(['gdalbuildvrt', vrt] + paths, capture_output=True, text=True, timeout=30, check=False) + except (subprocess.TimeoutExpired, OSError): + return None return vrt if result.returncode == 0 else None diff --git a/viewshed-worker/worker.py b/viewshed-worker/worker.py index 82df40f..edf483c 100644 --- a/viewshed-worker/worker.py +++ b/viewshed-worker/worker.py @@ -12,6 +12,7 @@ import logging import math import multiprocessing import os +import random import subprocess import tempfile import threading @@ -1742,11 +1743,23 @@ def worker_loop(): process_planned_job(db, r_client, job) else: process_job(db, r_client, job) + except psycopg2.errors.DeadlockDetected as exc: + log.warning(f'{name}: database deadlock detected — rolling back and recovering: {exc}') + try: + db.rollback() + except Exception: + db = wait_for_db() + time.sleep(random.uniform(0.25, 1.25)) except psycopg2.OperationalError: log.warning(f'{name}: DB connection lost — reconnecting') db = wait_for_db() except Exception as exc: log.error(f'{name}: job error: {exc}', exc_info=True) + try: + if db and not db.closed: + db.rollback() + except Exception: + db = wait_for_db() def resolve_node_ref(db, ref: str) -> dict: