Harden CI and background worker reliability

.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

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:

scripts/replace-container.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ $# -ne 1 ]]; then
+  echo "Usage: $0 <compose-service>" >&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}"

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,
     ))
 

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

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

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: