diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index f98bfd7..5866472 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -8,6 +8,10 @@ env: REGISTRY: ghcr.io IMAGE_PREFIX: ${{ github.repository_owner }}/crypto-sentiment +concurrency: + group: ${{ github.repository }}-production-deploy + cancel-in-progress: false + jobs: build-and-push: name: Build and Push Images @@ -49,9 +53,7 @@ jobs: context: . target: crawler push: true - tags: | - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-crawler:${{ steps.meta.outputs.version }} - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-crawler:latest + tags: ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-crawler:${{ steps.meta.outputs.version }} build-args: | TORCH_BASE_IMAGE=${{ env.REGISTRY }}/protostatis/crypto-sentiment-torch-base:latest cache-from: type=gha @@ -63,9 +65,7 @@ jobs: context: . target: api push: true - tags: | - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-api:${{ steps.meta.outputs.version }} - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-api:latest + tags: ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-api:${{ steps.meta.outputs.version }} cache-from: type=gha cache-to: type=gha,mode=max @@ -75,9 +75,7 @@ jobs: context: ./dashboard-frontend target: production push: true - tags: | - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-frontend:${{ steps.meta.outputs.version }} - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-frontend:latest + tags: ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-frontend:${{ steps.meta.outputs.version }} cache-from: type=gha cache-to: type=gha,mode=max @@ -87,9 +85,7 @@ jobs: context: ./dashboard-frontend/game file: ./dashboard-frontend/game/server/Dockerfile push: true - tags: | - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-game-server:${{ steps.meta.outputs.version }} - ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-game-server:latest + tags: ${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-game-server:${{ steps.meta.outputs.version }} cache-from: type=gha cache-to: type=gha,mode=max @@ -109,7 +105,7 @@ jobs: host: ${{ secrets.EC2_HOST }} username: ${{ secrets.EC2_USERNAME }} key: ${{ secrets.EC2_SSH_KEY }} - command_timeout: 20m + command_timeout: 45m script: | set -e @@ -168,21 +164,20 @@ jobs: if [ -f /opt/crypto-sentiment/.env ]; then CRAWLER_ENV_ARGS="--env-file /opt/crypto-sentiment/.env" else - echo "WARNING: /opt/crypto-sentiment/.env not found; crawler will run without local env config" + echo "ERROR: /opt/crypto-sentiment/.env is required for production" + exit 1 fi - # ========== PRE-DEPLOYMENT BACKUP ========== - if [ -f /opt/crypto-sentiment/data/sentiment.db ]; then - BACKUP_NAME="sentiment_predeploy_$(date +%Y%m%d_%H%M%S).db" - cp /opt/crypto-sentiment/data/sentiment.db /opt/crypto-sentiment/backups/$BACKUP_NAME - echo "Created pre-deploy backup: $BACKUP_NAME" - # Keep only last 14 backups - ls -t /opt/crypto-sentiment/backups/sentiment_*.db 2>/dev/null | tail -n +15 | xargs -r rm + # Existing production data is mandatory. Never bootstrap an empty + # database or repository state during a release deployment. + if [ ! -s /opt/crypto-sentiment/data/sentiment.db ]; then + echo "ERROR: Production database is missing or empty" + exit 1 + fi + if [ ! -s /opt/crypto-sentiment/data/orchestrator_state.json ]; then + echo "ERROR: Production orchestrator state is missing or empty" + exit 1 fi - - # Seed state files if missing - [ ! -f /opt/crypto-sentiment/data/orchestrator_state.json ] && cp data/orchestrator_state.json /opt/crypto-sentiment/data/ && echo "Seeded orchestrator_state.json" - [ ! -f /opt/crypto-sentiment/data/discovery_state.json ] && cp data/discovery_state.json /opt/crypto-sentiment/data/ 2>/dev/null || true # ========== DISK SPACE CHECK & CLEANUP ========== echo "Pre-cleanup disk usage:" @@ -230,6 +225,19 @@ jobs: # ========== ENSURE NETWORK EXISTS ========== docker network create --subnet=172.18.0.0/16 crypto-sentiment_crypto-net 2>/dev/null || echo "Network already exists" + # OpenRouter is fail-closed in the crawler. Prove the candidate image + # can make a real uncached request before replacing healthy services. + echo "Checking candidate OpenRouter embedding access..." + docker run --rm \ + --network crypto-sentiment_crypto-net \ + $CRAWLER_ENV_ARGS \ + crypto-sentiment-crawler:current \ + uv run python -m crypto_sentiment_crawler.maintenance.deployment_checks \ + openrouter \ + --expected-model qwen/qwen3-embedding-8b \ + --expected-dimensions 4096 + echo "Candidate OpenRouter canary passed." + # The local Mac's solver is forwarded as a mode-0600 Unix socket. # A solver outage degrades only Reddit collection; it must not block # unrelated API, frontend, or security releases. @@ -282,6 +290,68 @@ jobs: echo "WARNING: Reddit cookie solver socket is unavailable or has unsafe permissions ($SOCKET_METADATA; expected $EXPECTED_SOCKET_METADATA); deploying with standard Reddit fetching." fi + # ========== COHERENT PRE-DEPLOYMENT BACKUP ========== + # Use the candidate API image so backup behavior does not depend on + # the EC2 host's Python version. Retry if a live belief publication + # advances between the SQLite snapshot and the JSON state copy. + BACKUP_SUFFIX=$(date -u +%Y%m%dT%H%M%SZ) + BACKUP_NAME="sentiment_predeploy_${BACKUP_SUFFIX}.db" + STATE_BACKUP_NAME="orchestrator_state_predeploy_${BACKUP_SUFFIX}.json" + BACKUP_PATH="/opt/crypto-sentiment/backups/$BACKUP_NAME" + STATE_BACKUP_PATH="/opt/crypto-sentiment/backups/$STATE_BACKUP_NAME" + BACKUP_READY=0 + for attempt in $(seq 1 5); do + rm -f "$BACKUP_PATH" "$STATE_BACKUP_PATH" + docker run --rm \ + -v /opt/crypto-sentiment/data:/app/data:ro \ + -v /opt/crypto-sentiment/backups:/app/backups \ + crypto-sentiment-api:current \ + python -m crypto_sentiment_crawler.maintenance.deployment_checks \ + backup /app/data/sentiment.db "/app/backups/$BACKUP_NAME" + cp /opt/crypto-sentiment/data/orchestrator_state.json "$STATE_BACKUP_PATH" + if docker run --rm \ + -v /opt/crypto-sentiment/backups:/app/backups:ro \ + crypto-sentiment-api:current \ + python -m crypto_sentiment_crawler.maintenance.deployment_checks \ + publication \ + --db "/app/backups/$BACKUP_NAME" \ + --state "/app/backups/$STATE_BACKUP_NAME"; then + BACKUP_READY=1 + break + fi + echo "Backup pair changed during snapshot; retrying ($attempt/5)..." + sleep 1 + done + if [ "$BACKUP_READY" -ne 1 ]; then + echo "ERROR: Could not create a coherent database/state backup" + rm -f "$BACKUP_PATH" "$STATE_BACKUP_PATH" + exit 1 + fi + sha256sum "$BACKUP_PATH" "$STATE_BACKUP_PATH" + + if [ -f /opt/crypto-sentiment/data/discovery_state.json ]; then + DISCOVERY_BACKUP_PATH="/opt/crypto-sentiment/backups/discovery_state_predeploy_${BACKUP_SUFFIX}.json" + DISCOVERY_BACKUP_READY=0 + for attempt in $(seq 1 3); do + cp /opt/crypto-sentiment/data/discovery_state.json "$DISCOVERY_BACKUP_PATH" + if python3 -m json.tool "$DISCOVERY_BACKUP_PATH" >/dev/null 2>&1; then + DISCOVERY_BACKUP_READY=1 + break + fi + rm -f "$DISCOVERY_BACKUP_PATH" + sleep 1 + done + if [ "$DISCOVERY_BACKUP_READY" -ne 1 ]; then + echo "ERROR: Could not create a valid discovery-state backup" + exit 1 + fi + sha256sum "$DISCOVERY_BACKUP_PATH" + fi + echo "Created and verified pre-deploy recovery set: $BACKUP_SUFFIX" + ls -t /opt/crypto-sentiment/backups/sentiment_predeploy_*.db 2>/dev/null | tail -n +15 | xargs -r rm + ls -t /opt/crypto-sentiment/backups/orchestrator_state_predeploy_*.json 2>/dev/null | tail -n +15 | xargs -r rm + ls -t /opt/crypto-sentiment/backups/discovery_state_predeploy_*.json 2>/dev/null | tail -n +15 | xargs -r rm + # ========== SYNC CONFIG FILES ========== mkdir -p /opt/crypto-sentiment/dashboard-frontend cp dashboard-frontend/blocklist.conf /opt/crypto-sentiment/dashboard-frontend/blocklist.conf @@ -290,16 +360,34 @@ jobs: # Keep stopped containers until the replacement set passes health # checks so a failed rollout can restore the known-good set. SERVICES="crypto-crawler crypto-api crypto-frontend crypto-signals crypto-game-server" + RESTORE_ORDER="crypto-api crypto-game-server crypto-frontend crypto-crawler crypto-signals crypto-belief-auto" PREVIOUS_SERVICES="" for service in $SERVICES; do if docker inspect "$service-previous" >/dev/null 2>&1; then echo "ERROR: Stale rollback container exists: $service-previous" exit 1 fi - if docker inspect "$service" >/dev/null 2>&1; then - PREVIOUS_SERVICES="$PREVIOUS_SERVICES $service" + if ! docker inspect "$service" >/dev/null 2>&1; then + echo "ERROR: Cannot deploy safely without rollback container $service" + exit 1 fi + PREVIOUS_STATE=$(docker inspect --format '{{.State.Status}}' "$service") + if [ "$PREVIOUS_STATE" != "running" ]; then + echo "ERROR: Existing rollback container $service is not running ($PREVIOUS_STATE)" + exit 1 + fi + PREVIOUS_SERVICES="$PREVIOUS_SERVICES $service" done + if docker inspect crypto-belief-auto-previous >/dev/null 2>&1; then + echo "ERROR: Stale rollback container exists: crypto-belief-auto-previous" + exit 1 + fi + if docker inspect crypto-belief-auto >/dev/null 2>&1; then + LEGACY_STATE=$(docker inspect --format '{{.State.Status}}' crypto-belief-auto) + if [ "$LEGACY_STATE" = "running" ]; then + PREVIOUS_SERVICES="$PREVIOUS_SERVICES crypto-belief-auto" + fi + fi ROLLBACK_ACTIVE=1 NEW_CONTAINERS_STARTED=0 @@ -311,32 +399,79 @@ jobs: echo "Deployment failed; restoring previous containers..." ROLLBACK_ACTIVE=0 + trap - EXIT INT TERM + set +e + ROLLBACK_FAILED=0 if [ "$NEW_CONTAINERS_STARTED" -eq 1 ]; then for service in $SERVICES; do - docker rm -f "$service" >/dev/null 2>&1 || true + if docker inspect "$service" >/dev/null 2>&1; then + docker rm -f "$service" >/dev/null 2>&1 || ROLLBACK_FAILED=1 + fi done fi - for service in $PREVIOUS_SERVICES; do + for service in $RESTORE_ORDER; do if docker inspect "$service-previous" >/dev/null 2>&1; then - docker rename "$service-previous" "$service" || true - docker start "$service" >/dev/null 2>&1 || true - elif docker inspect "$service" >/dev/null 2>&1; then - docker start "$service" >/dev/null 2>&1 || true + if ! docker rename "$service-previous" "$service"; then + echo "ROLLBACK ERROR: Could not rename $service-previous" + ROLLBACK_FAILED=1 + continue + fi + if ! docker start "$service"; then + echo "ROLLBACK ERROR: Could not start $service" + ROLLBACK_FAILED=1 + fi + else + case " $PREVIOUS_SERVICES " in + *" $service "*) + if ! docker start "$service"; then + echo "ROLLBACK ERROR: Could not restart $service" + ROLLBACK_FAILED=1 + fi + ;; + esac + fi + done + sleep 5 + for service in $PREVIOUS_SERVICES; do + RESTORED_STATE=$(docker inspect --format '{{.State.Status}}' "$service" 2>/dev/null || echo "missing") + if [ "$RESTORED_STATE" != "running" ]; then + echo "ROLLBACK ERROR: $service is not running after restore ($RESTORED_STATE)" + ROLLBACK_FAILED=1 fi done + if [ "$ROLLBACK_FAILED" -ne 0 ]; then + echo "CRITICAL: Automatic rollback was incomplete; operator action required" + docker ps -a + else + echo "Previous service set restored." + fi return "$exit_code" } - trap 'rollback_on_exit' 0 - - # The current service set no longer includes belief-auto. - docker stop crypto-belief-auto 2>/dev/null || true - docker rm crypto-belief-auto 2>/dev/null || true + trap 'rollback_on_exit' EXIT + trap 'exit 130' INT + trap 'exit 143' TERM echo "Stopping old containers..." for service in $PREVIOUS_SERVICES; do docker stop "$service" docker rename "$service" "$service-previous" done + + HEARTBEAT_WATERMARK_JSON=$(docker run --rm \ + -v /opt/crypto-sentiment/data:/app/data:ro \ + crypto-sentiment-api:current \ + python -m crypto_sentiment_crawler.maintenance.deployment_checks \ + heartbeat-watermark --db /app/data/sentiment.db) + HEARTBEAT_WATERMARK=$(printf '%s' "$HEARTBEAT_WATERMARK_JSON" | \ + python3 -c 'import json, sys; print(json.load(sys.stdin)["heartbeat_id"])') + case "$HEARTBEAT_WATERMARK" in + ''|*[!0-9]*) + echo "ERROR: Invalid heartbeat watermark" + exit 1 + ;; + esac + CANDIDATE_STARTED_AT=$(date -u +%Y-%m-%dT%H:%M:%SZ) + echo "Candidate heartbeat watermark: $HEARTBEAT_WATERMARK" NEW_CONTAINERS_STARTED=1 # ========== START SERVICES ========== @@ -428,6 +563,46 @@ jobs: crypto-sentiment-crawler:current \ uv run signals bot + # Keep rollback protection active until every service is stable and + # the new crawler has completed its initial runtime jobs. + echo "Waiting for stable services and fresh runtime heartbeats..." + RUNTIME_CHECK_OUT=/tmp/panicradar-runtime-check.out + RUNTIME_CHECK_ERR=/tmp/panicradar-runtime-check.err + RUNTIME_READY=0 + for i in $(seq 1 360); do + CRAWLER_STATE=$(docker inspect --format '{{.State.Status}}' crypto-crawler 2>/dev/null || echo "missing") + CRAWLER_RESTARTS=$(docker inspect --format '{{.RestartCount}}' crypto-crawler 2>/dev/null || echo "999") + if [ "$CRAWLER_STATE" = "restarting" ] || [ "$CRAWLER_STATE" = "exited" ] || [ "$CRAWLER_STATE" = "dead" ] || [ "$CRAWLER_RESTARTS" -ne 0 ]; then + echo "ERROR: Crawler is unstable (state=$CRAWLER_STATE restarts=$CRAWLER_RESTARTS)" + docker logs crypto-crawler --tail 100 + exit 1 + fi + if [ "$CRAWLER_STATE" = "running" ] && docker exec crypto-api \ + python -m crypto_sentiment_crawler.maintenance.deployment_checks \ + runtime \ + --db /app/data/sentiment.db \ + --state /app/data/orchestrator_state.json \ + --since "$CANDIDATE_STARTED_AT" \ + --after-heartbeat-id "$HEARTBEAT_WATERMARK" \ + >"$RUNTIME_CHECK_OUT" 2>"$RUNTIME_CHECK_ERR"; then + cat "$RUNTIME_CHECK_OUT" + RUNTIME_READY=1 + break + fi + if [ $i -eq 360 ]; then + echo "ERROR: Runtime checks did not pass within 12 minutes" + cat "$RUNTIME_CHECK_ERR" 2>/dev/null || true + docker logs crypto-crawler --tail 100 + exit 1 + fi + sleep 2 + done + rm -f "$RUNTIME_CHECK_OUT" "$RUNTIME_CHECK_ERR" + if [ "$RUNTIME_READY" -ne 1 ]; then + echo "ERROR: Runtime readiness was not established" + exit 1 + fi + # ========== POST-DEPLOY CLEANUP ========== echo "Cleaning up unused images..." docker image prune -af @@ -446,20 +621,41 @@ jobs: exit 1 fi + for service in $SERVICES; do + SERVICE_STATE=$(docker inspect --format '{{.State.Status}}' "$service") + SERVICE_RESTARTS=$(docker inspect --format '{{.RestartCount}}' "$service") + if [ "$SERVICE_STATE" != "running" ] || [ "$SERVICE_RESTARTS" -ne 0 ]; then + echo "ERROR: $service is unstable (state=$SERVICE_STATE restarts=$SERVICE_RESTARTS)" + docker logs "$service" --tail 100 2>/dev/null || true + exit 1 + fi + done + + echo "Checking frontend and API through Nginx..." + if ! curl -ksf -o /dev/null https://localhost; then + echo "ERROR: Frontend HTTPS check failed" + docker logs crypto-frontend --tail 100 + exit 1 + fi + if ! curl -ksf -o /dev/null https://localhost/api/dashboard/summary; then + echo "ERROR: API proxy check failed" + docker logs crypto-frontend --tail 100 + docker logs crypto-api --tail 100 + exit 1 + fi + ROLLBACK_ACTIVE=0 for service in $PREVIOUS_SERVICES; do docker rm "$service-previous" >/dev/null 2>&1 || true done - trap - 0 + docker rm crypto-belief-auto >/dev/null 2>&1 || true + trap - EXIT INT TERM for component in crawler api frontend game-server; do docker tag "crypto-sentiment-$component:current" "crypto-sentiment-$component:latest" done - echo "All 5 containers running. Checking frontend..." - if ! curl -sf -o /dev/null https://localhost --insecure 2>/dev/null; then - echo "WARNING: Frontend HTTPS check failed (may be normal if cert issue)" - fi + echo "All 5 containers passed runtime and proxy checks." # ========== SETUP CRON ========== chmod +x deploy/backup-db.sh @@ -473,3 +669,33 @@ jobs: echo "========== DEPLOYMENT COMPLETE ==========" docker ps + + promote-latest: + name: Promote Verified Images + runs-on: ubuntu-latest + needs: [build-and-push, deploy] + permissions: + contents: read + packages: write + + steps: + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Log in to GitHub Container Registry + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Promote deployed release to latest + run: | + set -e + VERSION="${{ needs.build-and-push.outputs.version }}" + for component in crawler api frontend game-server; do + IMAGE="${{ env.REGISTRY }}/${{ env.IMAGE_PREFIX }}-$component" + docker buildx imagetools create \ + --tag "$IMAGE:latest" \ + "$IMAGE:$VERSION" + done diff --git a/crypto_sentiment_crawler/maintenance/deployment_checks.py b/crypto_sentiment_crawler/maintenance/deployment_checks.py new file mode 100644 index 0000000..6e0d9ad --- /dev/null +++ b/crypto_sentiment_crawler/maintenance/deployment_checks.py @@ -0,0 +1,429 @@ +"""Fail-closed checks used by the supervised production deployment.""" + +from __future__ import annotations + +import argparse +import json +import sqlite3 +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Sequence + +DEFAULT_HEARTBEAT_COMPONENTS = ("price", "crawl", "belief_update") + + +class DeploymentCheckError(RuntimeError): + """Raised when a deployment safety invariant is not satisfied.""" + + +def _parse_timestamp(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def create_verified_backup(source: Path, destination: Path) -> dict[str, Any]: + """Create and validate an online SQLite backup. + + SQLite's backup API takes a transactionally consistent snapshot while the + crawler remains online. A failed or corrupt destination is removed. + """ + + if not source.is_file(): + raise DeploymentCheckError(f"SQLite source does not exist: {source}") + if destination.exists(): + raise DeploymentCheckError(f"Backup destination already exists: {destination}") + + destination.parent.mkdir(parents=True, exist_ok=True) + source_uri = f"file:{source.resolve()}?mode=ro" + source_connection = sqlite3.connect(source_uri, uri=True, timeout=30.0) + destination_connection: sqlite3.Connection | None = None + verified = False + + try: + source_connection.execute("PRAGMA busy_timeout = 30000") + destination_connection = sqlite3.connect(destination, timeout=30.0) + source_connection.backup(destination_connection) + destination_connection.commit() + + result = destination_connection.execute("PRAGMA quick_check").fetchall() + if result != [("ok",)]: + raise DeploymentCheckError(f"Backup quick_check failed: {result!r}") + verified = True + finally: + if destination_connection is not None: + destination_connection.close() + source_connection.close() + if not verified: + destination.unlink(missing_ok=True) + + return { + "backup": str(destination), + "bytes": destination.stat().st_size, + "quick_check": "ok", + } + + +def _open_read_transaction(db_path: Path) -> sqlite3.Connection: + """Open a query-only transaction with a stable SQLite read snapshot.""" + + if not db_path.is_file(): + raise DeploymentCheckError(f"SQLite database does not exist: {db_path}") + + source_uri = f"file:{db_path.resolve()}?mode=ro" + connection = sqlite3.connect(source_uri, uri=True, timeout=30.0) + try: + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA busy_timeout = 30000") + connection.execute("PRAGMA query_only = ON") + connection.execute("BEGIN") + except Exception: + connection.close() + raise + return connection + + +def _check_integrity(connection: sqlite3.Connection, *, label: str) -> None: + rows = connection.execute("PRAGMA quick_check").fetchall() + results = [row[0] for row in rows] + if results != ["ok"]: + raise DeploymentCheckError(f"{label} quick_check failed: {results!r}") + + +def _read_state_version(state_path: Path) -> int: + if not state_path.is_file(): + raise DeploymentCheckError(f"Orchestrator state does not exist: {state_path}") + try: + state = json.loads(state_path.read_text()) + except (OSError, json.JSONDecodeError) as exc: + raise DeploymentCheckError(f"Invalid orchestrator state: {exc}") from exc + if not isinstance(state, dict): + raise DeploymentCheckError("Orchestrator state root is not an object") + state_version = state.get("belief_version") + if type(state_version) is not int: + raise DeploymentCheckError("State belief_version is not an integer") + return state_version + + +def _check_publication( + connection: sqlite3.Connection, + *, + state_path: Path, +) -> dict[str, Any]: + """Validate published source weights against one state snapshot.""" + + publication = connection.execute( + "SELECT belief_version FROM belief_publications WHERE id = 1" + ).fetchone() + if publication is None: + raise DeploymentCheckError("No published source-weight version") + belief_version = int(publication["belief_version"]) + + current_count = int( + connection.execute("SELECT COUNT(*) FROM source_weights").fetchone()[0] + ) + snapshot_count = int( + connection.execute( + "SELECT COUNT(*) FROM source_weight_snapshots WHERE belief_version = ?", + (belief_version,), + ).fetchone()[0] + ) + extra_count = int( + connection.execute( + """ + SELECT COUNT(*) + FROM source_weights current + WHERE NOT EXISTS ( + SELECT 1 FROM source_weight_snapshots snapshot + WHERE snapshot.belief_version = ? + AND snapshot.source = current.source + ) + """, + (belief_version,), + ).fetchone()[0] + ) + missing_count = int( + connection.execute( + """ + SELECT COUNT(*) + FROM source_weight_snapshots snapshot + WHERE snapshot.belief_version = ? + AND NOT EXISTS ( + SELECT 1 FROM source_weights current + WHERE current.source = snapshot.source + ) + """, + (belief_version,), + ).fetchone()[0] + ) + mismatch_count = int( + connection.execute( + """ + SELECT COUNT(*) + FROM source_weights current + JOIN source_weight_snapshots snapshot + ON snapshot.belief_version = ? + AND snapshot.source = current.source + WHERE current.belief_version IS NOT snapshot.belief_version + OR current.weight IS NOT snapshot.weight + OR current.accuracy IS NOT snapshot.accuracy + OR current.is_contrarian IS NOT snapshot.is_contrarian + OR current.alpha IS NOT snapshot.alpha + OR current.beta IS NOT snapshot.beta + OR current.sample_size IS NOT snapshot.sample_size + """, + (belief_version,), + ).fetchone()[0] + ) + + state_version = _read_state_version(state_path) + if state_version != belief_version: + raise DeploymentCheckError( + f"State version {state_version} != published version {belief_version}" + ) + if current_count <= 0: + raise DeploymentCheckError("Published source-weight mirror is empty") + if current_count != snapshot_count or extra_count or missing_count or mismatch_count: + raise DeploymentCheckError( + "Source-weight mirror mismatch: " + f"current={current_count}, snapshot={snapshot_count}, extra={extra_count}, " + f"missing={missing_count}, values={mismatch_count}" + ) + + return { + "belief_version": belief_version, + "source_weights": current_count, + "mirror_exact": True, + } + + +def check_publication_database( + *, + db_path: Path, + state_path: Path, +) -> dict[str, Any]: + """Verify a database/state backup pair is coherent and restorable.""" + + connection = _open_read_transaction(db_path) + try: + _check_integrity(connection, label="Backup") + publication = _check_publication(connection, state_path=state_path) + finally: + connection.close() + return {"quick_check": "ok", **publication} + + +def get_heartbeat_watermark(db_path: Path) -> dict[str, int]: + """Return the highest heartbeat row ID before candidate startup.""" + + connection = _open_read_transaction(db_path) + try: + row = connection.execute( + "SELECT COALESCE(MAX(id), 0) AS heartbeat_id FROM pipeline_heartbeats" + ).fetchone() + heartbeat_id = int(row["heartbeat_id"]) + finally: + connection.close() + return {"heartbeat_id": heartbeat_id} + + +def check_openrouter( + *, + expected_model: str, + expected_dimensions: int, +) -> dict[str, Any]: + """Make one uncached embedding request using the candidate image config.""" + + import numpy as np + + from ..config import settings + from ..processing.embedding_providers import OpenRouterEmbeddingProvider + + backend = (settings.embedding_backend or "").strip().lower() + model = (settings.embedding_model or "").strip() + if backend != "openrouter": + raise DeploymentCheckError( + f"Expected EMBEDDING_BACKEND=openrouter, got {backend or ''}" + ) + if model != expected_model: + raise DeploymentCheckError( + f"Expected EMBEDDING_MODEL={expected_model}, got {model or ''}" + ) + if not settings.openrouter_api_key: + raise DeploymentCheckError("OPENROUTER_API_KEY is empty") + + provider = OpenRouterEmbeddingProvider( + model=model, + api_key=settings.openrouter_api_key, + cache_path=None, + max_retries=1, + ) + vectors = provider.encode( + [f"panicradar deployment canary {time.time_ns()}"], + normalize=True, + ) + expected_shape = (1, expected_dimensions) + if vectors.shape != expected_shape: + raise DeploymentCheckError( + f"Expected embedding shape {expected_shape}, got {vectors.shape}" + ) + if not np.all(np.isfinite(vectors)): + raise DeploymentCheckError("OpenRouter returned a non-finite embedding") + norm = float(np.linalg.norm(vectors[0])) + if not np.isclose(norm, 1.0, atol=1e-5): + raise DeploymentCheckError(f"Embedding is not normalized (norm={norm})") + + return { + "backend": backend, + "model": model, + "shape": list(vectors.shape), + "normalized": True, + } + + +def check_runtime_database( + *, + db_path: Path, + state_path: Path, + since: datetime, + heartbeat_after_id: int, + components: Sequence[str] = DEFAULT_HEARTBEAT_COMPONENTS, +) -> dict[str, Any]: + """Verify candidate heartbeats and exact source-weight publication state.""" + + if heartbeat_after_id < 0: + raise DeploymentCheckError("Heartbeat watermark must be non-negative") + if since.tzinfo is None: + since = since.replace(tzinfo=timezone.utc) + since = since.astimezone(timezone.utc) + + connection = _open_read_transaction(db_path) + try: + heartbeat_times: dict[str, str] = {} + heartbeat_ids: dict[str, int] = {} + for component in components: + row = connection.execute( + """ + SELECT id, last_success_at, last_error_at, last_error_message + FROM pipeline_heartbeats + WHERE component = ? + ORDER BY id DESC + LIMIT 1 + """, + (component,), + ).fetchone() + if row is None: + raise DeploymentCheckError(f"Missing heartbeat for {component}") + heartbeat_id = int(row["id"]) + if heartbeat_id <= heartbeat_after_id: + raise DeploymentCheckError( + f"Heartbeat for {component} did not come from the candidate " + f"(id={heartbeat_id}, watermark={heartbeat_after_id})" + ) + if row["last_error_at"] is not None: + raise DeploymentCheckError( + f"Latest {component} heartbeat is an error: " + f"{row['last_error_message'] or ''}" + ) + if row["last_success_at"] is None: + raise DeploymentCheckError(f"Heartbeat has no success time: {component}") + success_at = _parse_timestamp(row["last_success_at"]) + if success_at <= since: + raise DeploymentCheckError( + f"Heartbeat for {component} predates candidate startup: " + f"{success_at.isoformat()}" + ) + heartbeat_times[component] = success_at.isoformat() + heartbeat_ids[component] = heartbeat_id + + # Integrity can be expensive on the production database. Run it only + # after the lightweight heartbeat gate proves candidate startup is done. + _check_integrity(connection, label="Runtime") + publication = _check_publication(connection, state_path=state_path) + finally: + connection.close() + + return { + "quick_check": "ok", + "heartbeats": heartbeat_times, + "heartbeat_ids": heartbeat_ids, + "heartbeat_watermark": heartbeat_after_id, + **publication, + } + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + backup = subparsers.add_parser("backup", help="Create a verified SQLite backup") + backup.add_argument("source", type=Path) + backup.add_argument("destination", type=Path) + + publication = subparsers.add_parser( + "publication", help="Check a database/state publication pair" + ) + publication.add_argument("--db", type=Path, required=True) + publication.add_argument("--state", type=Path, required=True) + + watermark = subparsers.add_parser( + "heartbeat-watermark", help="Read the latest heartbeat row ID" + ) + watermark.add_argument("--db", type=Path, required=True) + + openrouter = subparsers.add_parser("openrouter", help="Run an embedding canary") + openrouter.add_argument("--expected-model", required=True) + openrouter.add_argument("--expected-dimensions", type=int, required=True) + + runtime = subparsers.add_parser("runtime", help="Check post-start runtime state") + runtime.add_argument("--db", type=Path, required=True) + runtime.add_argument("--state", type=Path, required=True) + runtime.add_argument("--since", required=True) + runtime.add_argument("--after-heartbeat-id", type=int, required=True) + runtime.add_argument( + "--component", + action="append", + dest="components", + help="Required fresh heartbeat component (repeatable)", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + try: + if args.command == "backup": + result = create_verified_backup(args.source, args.destination) + elif args.command == "publication": + result = check_publication_database( + db_path=args.db, + state_path=args.state, + ) + elif args.command == "heartbeat-watermark": + result = get_heartbeat_watermark(args.db) + elif args.command == "openrouter": + result = check_openrouter( + expected_model=args.expected_model, + expected_dimensions=args.expected_dimensions, + ) + else: + result = check_runtime_database( + db_path=args.db, + state_path=args.state, + since=_parse_timestamp(args.since), + heartbeat_after_id=args.after_heartbeat_id, + components=args.components or DEFAULT_HEARTBEAT_COMPONENTS, + ) + except Exception as exc: + print(f"DEPLOYMENT CHECK FAILED: {exc}", file=sys.stderr) + return 1 + + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_deployment_checks.py b/tests/test_deployment_checks.py new file mode 100644 index 0000000..cb873d0 --- /dev/null +++ b/tests/test_deployment_checks.py @@ -0,0 +1,253 @@ +"""Tests for deployment backup and runtime acceptance checks.""" + +import json +import sqlite3 +from datetime import datetime, timedelta, timezone +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pytest + +from crypto_sentiment_crawler.maintenance.deployment_checks import ( + DeploymentCheckError, + check_openrouter, + check_publication_database, + check_runtime_database, + create_verified_backup, + get_heartbeat_watermark, +) + + +def test_create_verified_backup_copies_a_consistent_database(tmp_path: Path) -> None: + source = tmp_path / "source.db" + destination = tmp_path / "backup.db" + connection = sqlite3.connect(source) + assert connection.execute("PRAGMA journal_mode = WAL").fetchone()[0] == "wal" + connection.execute("CREATE TABLE events (value TEXT)") + connection.execute("INSERT INTO events VALUES ('ready')") + connection.commit() + connection.close() + + result = create_verified_backup(source, destination) + + backup = sqlite3.connect(destination) + try: + value = backup.execute("SELECT value FROM events").fetchone()[0] + finally: + backup.close() + assert value == "ready" + assert result["quick_check"] == "ok" + assert result["bytes"] > 0 + source_connection = sqlite3.connect(source) + try: + assert source_connection.execute("PRAGMA journal_mode").fetchone()[0] == "wal" + finally: + source_connection.close() + + +def _create_runtime_state(tmp_path: Path) -> tuple[Path, Path, datetime]: + db_path = tmp_path / "sentiment.db" + state_path = tmp_path / "orchestrator_state.json" + since = datetime.now(timezone.utc) - timedelta(seconds=1) + heartbeat_time = datetime.now(timezone.utc).isoformat() + + connection = sqlite3.connect(db_path) + connection.executescript( + """ + CREATE TABLE pipeline_heartbeats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + component TEXT, + last_success_at TEXT, + last_error_at TEXT, + last_error_message TEXT + ); + CREATE TABLE belief_publications ( + id INTEGER PRIMARY KEY, + belief_version INTEGER NOT NULL + ); + CREATE TABLE source_weights ( + source TEXT PRIMARY KEY, + weight REAL, + accuracy REAL, + is_contrarian INTEGER, + alpha REAL, + beta REAL, + sample_size INTEGER, + belief_version INTEGER + ); + CREATE TABLE source_weight_snapshots ( + belief_version INTEGER, + source TEXT, + weight REAL, + accuracy REAL, + is_contrarian INTEGER, + alpha REAL, + beta REAL, + sample_size INTEGER, + PRIMARY KEY (belief_version, source) + ); + """ + ) + connection.executemany( + """ + INSERT INTO pipeline_heartbeats ( + component, last_success_at, last_error_at, last_error_message + ) VALUES (?, ?, NULL, NULL) + """, + [(component, heartbeat_time) for component in ("price", "crawl", "belief_update")], + ) + connection.execute("INSERT INTO belief_publications VALUES (1, 7)") + values = ("reddit_bitcoin", 0.4, 0.6, 0, 3.0, 2.0, 5, 7) + connection.execute("INSERT INTO source_weights VALUES (?, ?, ?, ?, ?, ?, ?, ?)", values) + connection.execute( + "INSERT INTO source_weight_snapshots VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (7, *values[:-1]), + ) + connection.commit() + connection.close() + state_path.write_text(json.dumps({"belief_version": 7})) + return db_path, state_path, since + + +def test_runtime_check_accepts_fresh_exact_publication(tmp_path: Path) -> None: + db_path, state_path, since = _create_runtime_state(tmp_path) + + result = check_runtime_database( + db_path=db_path, + state_path=state_path, + since=since, + heartbeat_after_id=0, + ) + + assert result["belief_version"] == 7 + assert result["source_weights"] == 1 + assert result["mirror_exact"] is True + assert set(result["heartbeats"]) == {"price", "crawl", "belief_update"} + + +def test_runtime_check_rejects_extra_current_weight(tmp_path: Path) -> None: + db_path, state_path, since = _create_runtime_state(tmp_path) + connection = sqlite3.connect(db_path) + connection.execute( + "INSERT INTO source_weights VALUES ('ghost', 0.9, NULL, 0, NULL, NULL, NULL, 7)" + ) + connection.commit() + connection.close() + + with pytest.raises(DeploymentCheckError, match="mirror mismatch"): + check_runtime_database( + db_path=db_path, + state_path=state_path, + since=since, + heartbeat_after_id=0, + ) + + +def test_runtime_check_rejects_error_heartbeat(tmp_path: Path) -> None: + db_path, state_path, since = _create_runtime_state(tmp_path) + connection = sqlite3.connect(db_path) + connection.execute( + """ + INSERT INTO pipeline_heartbeats ( + component, last_success_at, last_error_at, last_error_message + ) VALUES ('crawl', ?, ?, 'database is locked') + """, + (datetime.now(timezone.utc).isoformat(), datetime.now(timezone.utc).isoformat()), + ) + connection.commit() + connection.close() + + with pytest.raises(DeploymentCheckError, match="crawl heartbeat is an error"): + check_runtime_database( + db_path=db_path, + state_path=state_path, + since=since, + heartbeat_after_id=0, + ) + + +def test_runtime_check_rejects_pre_cutover_heartbeats(tmp_path: Path) -> None: + db_path, state_path, since = _create_runtime_state(tmp_path) + watermark = get_heartbeat_watermark(db_path)["heartbeat_id"] + + with pytest.raises(DeploymentCheckError, match="did not come from the candidate"): + check_runtime_database( + db_path=db_path, + state_path=state_path, + since=since, + heartbeat_after_id=watermark, + ) + + +def test_publication_check_rejects_mismatched_state_backup(tmp_path: Path) -> None: + db_path, state_path, _ = _create_runtime_state(tmp_path) + assert check_publication_database( + db_path=db_path, + state_path=state_path, + )["belief_version"] == 7 + + state_path.write_text(json.dumps({"belief_version": 8})) + with pytest.raises(DeploymentCheckError, match="State version 8"): + check_publication_database(db_path=db_path, state_path=state_path) + + +def test_publication_check_rejects_invalid_state_json(tmp_path: Path) -> None: + db_path, state_path, _ = _create_runtime_state(tmp_path) + state_path.write_text("{") + + with pytest.raises(DeploymentCheckError, match="Invalid orchestrator state"): + check_publication_database(db_path=db_path, state_path=state_path) + + +def test_openrouter_check_uses_candidate_configuration(monkeypatch) -> None: + from crypto_sentiment_crawler import config + from crypto_sentiment_crawler.processing import embedding_providers + + configured = SimpleNamespace( + embedding_backend="openrouter", + embedding_model="qwen/qwen3-embedding-8b", + openrouter_api_key="secret-test-value", + ) + + class FakeProvider: + def __init__(self, **kwargs): + assert kwargs["model"] == configured.embedding_model + assert kwargs["api_key"] == configured.openrouter_api_key + assert kwargs["cache_path"] is None + + def encode(self, texts, normalize=True): + assert len(texts) == 1 + assert normalize is True + return np.ones((1, 4), dtype=np.float32) / 2 + + monkeypatch.setattr(config, "settings", configured) + monkeypatch.setattr(embedding_providers, "OpenRouterEmbeddingProvider", FakeProvider) + + result = check_openrouter( + expected_model="qwen/qwen3-embedding-8b", + expected_dimensions=4, + ) + + assert result["shape"] == [1, 4] + assert result["normalized"] is True + + +def test_openrouter_check_rejects_local_backend(monkeypatch) -> None: + from crypto_sentiment_crawler import config + + monkeypatch.setattr( + config, + "settings", + SimpleNamespace( + embedding_backend="local", + embedding_model="all-MiniLM-L6-v2", + openrouter_api_key="", + ), + ) + + with pytest.raises(DeploymentCheckError, match="EMBEDDING_BACKEND=openrouter"): + check_openrouter( + expected_model="qwen/qwen3-embedding-8b", + expected_dimensions=4096, + )