From da2bad89f9508a95e059cd1f5d20a84d1f9911f9 Mon Sep 17 00:00:00 2001 From: Edgars Date: Wed, 17 Jun 2026 21:50:23 +0100 Subject: [PATCH] chore: remove hosted ops files from studio --- .claude/skills/cloudflare-cli/SKILL.md | 107 --- .../discord-community-feedback/SKILL.md | 103 --- .claude/skills/hosted-studio-debug/SKILL.md | 150 ---- .claude/skills/manage-api-keys/SKILL.md | 128 ---- .claude/skills/manage-tiers/SKILL.md | 105 --- .claude/skills/studio-db/SKILL.md | 155 ---- scripts/dump_transactions_to_metrics.py | 430 ----------- scripts/run_dump_transactions.sh | 177 ----- scripts/run_update_error_transactions.sh | 184 ----- scripts/update_error_transactions_metrics.py | 687 ------------------ 10 files changed, 2226 deletions(-) delete mode 100644 .claude/skills/cloudflare-cli/SKILL.md delete mode 100644 .claude/skills/discord-community-feedback/SKILL.md delete mode 100644 .claude/skills/hosted-studio-debug/SKILL.md delete mode 100644 .claude/skills/manage-api-keys/SKILL.md delete mode 100644 .claude/skills/manage-tiers/SKILL.md delete mode 100644 .claude/skills/studio-db/SKILL.md delete mode 100644 scripts/dump_transactions_to_metrics.py delete mode 100755 scripts/run_dump_transactions.sh delete mode 100755 scripts/run_update_error_transactions.sh delete mode 100755 scripts/update_error_transactions_metrics.py diff --git a/.claude/skills/cloudflare-cli/SKILL.md b/.claude/skills/cloudflare-cli/SKILL.md deleted file mode 100644 index ad23baa0e..000000000 --- a/.claude/skills/cloudflare-cli/SKILL.md +++ /dev/null @@ -1,107 +0,0 @@ ---- -name: cloudflare-cli -description: Debug and manage Cloudflare DNS, cache, and proxy settings using flarectl ---- - -# Cloudflare CLI Skill - -Debug and manage Cloudflare DNS, cache, and proxy settings using flarectl. - -## Prerequisites - -- flarectl installed: `brew install cloudflare/cloudflare/flarectl` -- API token set: `export CF_API_TOKEN=` - -## Commands - -### Zones -```bash -# List all zones -flarectl zone list - -# Get zone details -flarectl zone info --zone genlayer.com -``` - -### DNS Records -```bash -# List DNS records -flarectl dns list --zone genlayer.com - -# Create record -flarectl dns create --zone genlayer.com --name subdomain --type A --content 1.2.3.4 - -# Update record (toggle proxy) -flarectl dns update --zone genlayer.com --id --proxy=true - -# Delete record -flarectl dns delete --zone genlayer.com --id -``` - -### Cache -```bash -# Purge all cache -flarectl zone purge --zone genlayer.com --everything - -# Purge specific URLs -flarectl zone purge --zone genlayer.com --files "https://studio.genlayer.com/api" -``` - -## Common Debug Patterns - -### Check if Cloudflare Proxy is Enabled -```bash -flarectl dns list --zone genlayer.com | grep studio -# Look for "proxied: true" in output -``` - -### Test Direct to Origin (bypass Cloudflare) -```bash -# Get origin IP -dig +short studio.genlayer.com - -# Test direct (if proxy disabled) -curl -v https://studio.genlayer.com/health - -# Test with Host header to specific IP -curl -H "Host: studio.genlayer.com" https:///health -k -``` - -### Diagnose 502/503 Errors -```bash -# Check if error is from Cloudflare or origin -# Cloudflare errors have cf-ray header -curl -I https://studio.genlayer.com/api 2>&1 | grep -i "cf-ray" - -# Fast response (30-60ms) = Cloudflare WAF/rate limit -# Slow response (5s+) = Origin timeout -curl -w "%{time_total}s\n" -o /dev/null -s https://studio.genlayer.com/api -``` - -### Rate Limiting Check -```bash -# 429 with fast response = Cloudflare rate limiting -for i in {1..10}; do - curl -s -o /dev/null -w "%{http_code} %{time_total}s\n" https://studio.genlayer.com/api \ - -X POST -H "Content-Type: application/json" \ - -d '{"jsonrpc":"2.0","method":"eth_chainId","params":[],"id":1}' -done -``` - -## GenLayer Domains - -| Domain | Purpose | -|--------|---------| -| studio.genlayer.com | Production studio | -| studio-dev.genlayer.com | Dev studio | -| studio-stage.genlayer.com | Staging studio | -| rally-testnet.genlayer.com | Rally production | -| rally-devnet.genlayer.com | Rally dev | - -## SSL Modes - -- **Full**: Cloudflare → Origin with any cert (allows self-signed) -- **Full (Strict)**: Cloudflare → Origin with valid cert -- **Flexible**: Cloudflare terminates SSL, HTTP to origin - -GenLayer uses Cloudflare Origin Certificates - requires proxy enabled and Full (Strict) mode. diff --git a/.claude/skills/discord-community-feedback/SKILL.md b/.claude/skills/discord-community-feedback/SKILL.md deleted file mode 100644 index 97fe07ae7..000000000 --- a/.claude/skills/discord-community-feedback/SKILL.md +++ /dev/null @@ -1,103 +0,0 @@ ---- -name: discord-community-feedback -description: Monitor Discord community channel for user-reported bugs and issues ---- - -# Discord Community Feedback Skill - -Monitor the GenLayer Discord community feedback channel for user-reported bugs, issues, and problems. - -## Prerequisites - -- Discord MCP server configured and running ([discordmcp](https://github.com/v-3/discordmcp)) -- Bot added to the GenLayer Discord server with read permissions - -## Channel Information - -| Field | Value | -|-------|-------| -| Server ID | 1237055789441487021 | -| Channel ID | 1237114454877929482 | - -## Quick Commands - -Use the Discord MCP `read-messages` tool to fetch recent messages: - -``` -Tool: read-messages -Channel: 1237114454877929482 -Limit: 100 (max allowed) -``` - -## Identifying User Problems - -When reviewing messages, look for these indicators of issues: - -### Problem Keywords -- **Errors**: error, bug, broken, crash, fail, exception -- **Functionality**: not working, doesn't work, can't, unable, stuck -- **Help requests**: help, issue, problem, wrong, weird -- **Performance**: slow, timeout, hang, freeze, unresponsive - -### Common User-Reported Issues - -| Category | Example Patterns | Related Component | -|----------|------------------|-------------------| -| Transaction failures | "tx failed", "transaction stuck" | consensus-worker | -| Contract errors | "contract not deploying", "execution error" | genvm, consensus-worker | -| UI issues | "page not loading", "button doesn't work" | frontend | -| API errors | "RPC error", "connection failed" | jsonrpc | -| Wallet issues | "can't connect wallet", "balance wrong" | frontend, jsonrpc | - -## Workflow - -1. **Fetch recent messages** from the feedback channel (up to 100) -2. **Scan for problem indicators** using keywords above -3. **Categorize issues** by component (frontend, backend, consensus, etc.) -4. **Cross-reference with logs** using the `hosted-studio-debug` skill if needed -5. **Summarize findings** for the team - -## Triage Priority - -| Priority | Indicators | -|----------|------------| -| High | Multiple users reporting same issue, "everything broken", data loss | -| Medium | Single user with reproducible issue, feature not working | -| Low | Questions, minor UI glitches, feature requests | - -## Integration with Hosted Studio Debug - -If a user reports an issue that might be related to the hosted environment: - -1. Note the approximate time of the user's report -2. Use `hosted-studio-debug` skill to check logs around that time -3. Look for correlating errors in consensus-worker or jsonrpc logs - -```bash -# Example: Check for errors around the time of a user report -argocd app logs studio-prd-workload --name studio-consensus-worker --tail 500 2>&1 | grep -iE "(error|exception|timeout)" -``` - -## Response Template - -When summarizing community feedback: - -``` -## Community Feedback Summary - -**Period**: [time range of messages reviewed] -**Total Messages**: [count] -**Issues Identified**: [count] - -### High Priority -- [Issue description] - Reported by [count] users - -### Medium Priority -- [Issue description] - -### Low Priority / Questions -- [Description] - -### No Issues -[If no problems found, note this] -``` diff --git a/.claude/skills/hosted-studio-debug/SKILL.md b/.claude/skills/hosted-studio-debug/SKILL.md deleted file mode 100644 index 52ee0871f..000000000 --- a/.claude/skills/hosted-studio-debug/SKILL.md +++ /dev/null @@ -1,150 +0,0 @@ ---- -name: hosted-studio-debug -description: Debug GenLayer Studio deployments via ArgoCD CLI ---- - -# Hosted Studio Debug Skill - -Debug GenLayer Studio deployments via ArgoCD CLI. - -## Workload Manifests - -Kubernetes manifests are in sibling repo `../devexp-apps-workload` (assume by default, ask user if not found): - -``` -devexp-apps-workload/workload/ -├── dev/ # studio-dev, rally-studio-dev -├── stg/ # studio-stg -├── prd/ # studio-prd, rally-studio-prd -``` - -Each contains Deployments, Services, Ingresses, ExternalSecrets managed by ArgoCD. - -## Prerequisites - -- Logged into ArgoCD CLI -- Access to target cluster - -## Full Status Check - -To get complete visibility into Studio status, check all components across all replicas: - -```bash -# 1. Overall app health -argocd app get -workload - -# 2. List all pods and their status -argocd app resources -workload --kind Pod - -# 3. Check consensus worker logs (all 4 replicas in prd) -argocd app logs -workload --name studio-consensus-worker --tail 500 2>&1 | grep -iE "(error|exception|timeout|failed)" - -# 4. Check JSON-RPC logs (all 2 replicas in prd) -argocd app logs -workload --name studio-jsonrpc --tail 500 2>&1 | grep -iE "(error|exception|timeout|failed)" - -# 5. Check webdriver logs -argocd app logs -workload --name studio-webdriver --tail 200 2>&1 | grep -iE "(error|exception|crash)" -``` - -## Quick Commands - -```bash -# Check app health -argocd app get -workload - -# List all resources (pods, deployments, services, etc.) -argocd app resources -workload - -# List all pods with their status -argocd app resources -workload --kind Pod - -# Tail consensus worker logs (aggregates from ALL replicas) -argocd app logs -workload --name studio-consensus-worker --tail 200 - -# Tail JSON-RPC logs (aggregates from ALL replicas) -argocd app logs -workload --name studio-jsonrpc --tail 200 - -# Check for errors across all replicas -argocd app logs -workload --name studio-consensus-worker --tail 500 2>&1 | grep -i error -argocd app logs -workload --name studio-jsonrpc --tail 500 2>&1 | grep -i error -``` - -## Multi-Replica Log Access - -**Important:** The `argocd app logs --name ` command automatically aggregates logs from ALL pod replicas. This ensures full visibility across the entire deployment. - -```bash -# Get logs from all consensus worker replicas (production has 4) -argocd app logs -workload --name studio-consensus-worker --tail 500 - -# Get logs from all JSON-RPC replicas (production has 2) -argocd app logs -workload --name studio-jsonrpc --tail 500 - -# Get logs from a specific pod (if needed for isolation) -argocd app logs -workload --pod --tail 200 - -# List pods first to get pod names -argocd app resources -workload --kind Pod -``` - -## Common Issues - -### Transaction Timeouts - -Consensus worker timeouts usually mean GenVM Manager is unresponsive. Check all 4 workers in production: - -```bash -# Check for GenVM timeouts across all consensus worker replicas -argocd app logs -workload --name studio-consensus-worker --tail 500 2>&1 | grep -E "(timeout|SocketTimeoutError|127.0.0.1:3999)" - -# Empty stdout = GenVM never started -argocd app logs -workload --name studio-consensus-worker --tail 500 2>&1 | grep "stdout=''" -``` - -**Root cause**: GenVM Manager (`genvm-modules manager --port 3999`) becomes unresponsive. -**Fix**: Worker restart (auto or manual) restarts GenVM Manager. - -Key files: -- `backend/node/genvm/origin/base_host.py:463` - where timeouts occur -- `backend/node/base.py` - Manager.create() spawns GenVM -- `backend/consensus/worker_service.py` - worker startup/health - -### Pod Restarts - -```bash -# Check restart timestamps in logs -argocd app logs -workload --name studio-consensus-worker --tail 500 2>&1 | grep -E "(Started|Uvicorn running)" - -# Via kubectl -kubectl get pods -n -o wide -kubectl get events -n --sort-by='.lastTimestamp' -``` - -### External API Failures - -Contracts calling external APIs (Twitter, etc.) through proxies: - -```bash -# Check for HTTP errors in contract execution -argocd app logs -workload --name studio-consensus-worker --tail 500 2>&1 | grep -E "(HTTP|fetch|proxy|api)" -``` - -## Environment Apps - -| Env | App | Namespace | -|-----|-----|-----------| -| dev | studio-dev-workload | studio-dev | -| stg | studio-stg-workload | studio-stg | -| prd | studio-prd-workload | studio-prd | -| rally-prd | rally-studio-prd-workload | rally-studio-prd | - -## Components - -| Component | Purpose | Prd Replicas | Common Issues | -|-----------|---------|--------------|---------------| -| studio-consensus-worker | Tx processing | 4 | Timeouts, GenVM crashes | -| studio-jsonrpc | RPC API | 2 | DB connections | -| studio-webdriver | Browser sandbox | 1 | Memory, crashes | -| database-migration | Schema updates | 1 (job) | Lock contention | - -**Note:** When debugging, always check logs from ALL replicas to get complete visibility. The `argocd app logs --name ` command handles this automatically. diff --git a/.claude/skills/manage-api-keys/SKILL.md b/.claude/skills/manage-api-keys/SKILL.md deleted file mode 100644 index 1153fc866..000000000 --- a/.claude/skills/manage-api-keys/SKILL.md +++ /dev/null @@ -1,128 +0,0 @@ ---- -name: manage-api-keys -description: Create, list, deactivate, and reactivate API keys for rate limiting -invocation: user ---- - -# Manage API Keys - -CRUD operations for API keys used in rate limiting on GenLayer Studio deployments. - -## Setup - -Before any operation, determine the target environment: - -1. **Ask the user** which environment they are targeting: - - **Local dev**: `BASE_URL=http://localhost:4000/api`, no `admin_key` needed - - **Hosted** (dev/stg/prd): `BASE_URL=https:///api`, requires `admin_key` - -2. For hosted deployments, **ask the user for the `ADMIN_API_KEY`** (stored in k8s secrets as `ADMIN_API_KEY`). - -## Operations - -Ask the user which operation to perform: **Create**, **List**, **Deactivate**, or **Reactivate**. - -### List API Keys - -**Note:** This endpoint may not exist yet. If `admin_listApiKeys` is not available, query the database directly using the `studio-db` skill, or inform the user it needs to be implemented. - -```bash -# List all keys -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_listApiKeys","params":{"admin_key":""},"id":1} -EOF - -# Filter by tier -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_listApiKeys","params":{"tier_name":"free","admin_key":""},"id":1} -EOF - -# Filter by active status -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_listApiKeys","params":{"is_active":false,"admin_key":""},"id":1} -EOF -``` - -### Create API Key - -Ask the user for: `tier_name` (suggest listing tiers first with `/manage-tiers`) and optional `description`. - -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_createApiKey","params":{"tier_name":"","description":"","admin_key":""},"id":1} -EOF -``` - -**IMPORTANT:** The full API key (e.g., `glk_abcdef1234...`) is **only returned once** at creation time. Remind the user to store it securely. Only the `key_prefix` (first 8 chars) is stored for identification. - -Response includes: -- `api_key`: Full key (store this!) -- `key_prefix`: First 8 characters (e.g., `glk_ab12`) -- `tier`: Tier name -- `description`: Optional description - -### Deactivate API Key - -Ask the user for the `key_prefix` (8 characters, e.g., `glk_ab12`). If they don't know it, list keys first. - -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_deactivateApiKey","params":{"key_prefix":"","admin_key":""},"id":1} -EOF -``` - -Deactivation takes effect immediately (Redis cache is invalidated). - -### Reactivate API Key - -**Note:** This endpoint may not exist yet. If `admin_reactivateApiKey` is not available, inform the user it needs to be implemented. - -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_reactivateApiKey","params":{"key_prefix":"","admin_key":""},"id":1} -EOF -``` - -## API Key Usage - -Clients send the API key via the `X-API-Key` HTTP header: - -```bash -curl -X POST "$BASE_URL" \ - -H "Content-Type: application/json" \ - -H "X-API-Key: glk_" \ - -d '{"jsonrpc":"2.0","method":"","params":{...},"id":1}' -``` - -- Requests without `X-API-Key` are subject to anonymous rate limits (default: 10/min, 100/hr, 1000/day). -- Invalid or deactivated keys return `-32029 "Invalid API key"`. -- When the rate limit is exceeded, the response is HTTP 429 with a JSON-RPC error including `window`, `limit`, `current`, and `retry_after_seconds`. - -## Common Errors - -| Error | Cause | -|-------|-------| -| `-32000` "Admin access required" | Missing or invalid `admin_key` on hosted deployment | -| `-32602` "Tier not found: X" | Specified `tier_name` doesn't exist | -| `-32001` "Active API key with prefix X not found" | Key doesn't exist or is already deactivated | -| `-32029` "Invalid API key" | Key is invalid or deactivated (when rate limiting is enabled) | -| `-32029` "Rate limit exceeded: N requests per minute" | Key has exceeded its tier's rate limit | - -## Key Format - -- Full key: `glk_` + 64 hex characters (68 chars total) -- Key prefix: first 8 characters (e.g., `glk_ab12`) -- Storage: only the SHA-256 hash is stored in the database; the full key cannot be recovered - -## Important Notes - -- Always use `--data-binary @-` with heredoc (`<<'EOF'`) to avoid shell expansion issues with special characters in the admin key. -- When rate limiting is disabled (`RATE_LIMIT_ENABLED=false`), keys can still be created and managed, but rate limits are not enforced and invalid keys are not rejected. -- Cache invalidation happens automatically on deactivate/reactivate (5-minute TTL otherwise). - -## Reference - -- Models: `backend/database_handler/models.py` (ApiKey class) -- Endpoints: `backend/protocol_rpc/endpoints.py` (admin_create_api_key, admin_deactivate_api_key) -- Rate limiter: `backend/protocol_rpc/rate_limiter.py` -- Middleware: `backend/protocol_rpc/rate_limit_middleware.py` diff --git a/.claude/skills/manage-tiers/SKILL.md b/.claude/skills/manage-tiers/SKILL.md deleted file mode 100644 index 38454bdd8..000000000 --- a/.claude/skills/manage-tiers/SKILL.md +++ /dev/null @@ -1,105 +0,0 @@ ---- -name: manage-tiers -description: Create, list, update, and delete API rate limiting tiers -invocation: user ---- - -# Manage API Rate Limiting Tiers - -CRUD operations for API rate limiting tiers on GenLayer Studio deployments. - -## Setup - -Before any operation, determine the target environment: - -1. **Ask the user** which environment they are targeting: - - **Local dev**: `BASE_URL=http://localhost:4000/api`, no `admin_key` needed - - **Hosted** (dev/stg/prd): `BASE_URL=https:///api`, requires `admin_key` - -2. For hosted deployments, **ask the user for the `ADMIN_API_KEY`** (stored in k8s secrets as `ADMIN_API_KEY`). - -## Operations - -Ask the user which operation to perform: **Create**, **List**, **Update**, or **Delete**. - -### List Tiers - -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_listTiers","params":{"admin_key":""},"id":1} -EOF -``` - -For local dev (no admin_key): -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" \ - -d '{"jsonrpc":"2.0","method":"admin_listTiers","params":{},"id":1}' | python3 -m json.tool -``` - -### Create Tier - -Ask the user for: `name`, `rate_limit_minute`, `rate_limit_hour`, `rate_limit_day`. - -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_createTier","params":{"name":"","rate_limit_minute":,"rate_limit_hour":,"rate_limit_day":,"admin_key":""},"id":1} -EOF -``` - -### Update Tier - -**Note:** This endpoint may not exist yet. Check by listing tiers first. If `admin_updateTier` is not available, inform the user it needs to be implemented. - -Ask the user for: `name` (existing tier) and which limits to change. - -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_updateTier","params":{"name":"","rate_limit_minute":,"rate_limit_hour":,"rate_limit_day":,"admin_key":""},"id":1} -EOF -``` - -Only include the fields that need changing. - -### Delete Tier - -**Note:** This endpoint may not exist yet. If `admin_deleteTier` is not available, inform the user it needs to be implemented. - -Deletion fails if any API keys (active or inactive) reference the tier. - -```bash -curl -s -X POST "$BASE_URL" -H "Content-Type: application/json" --data-binary @- <<'EOF' | python3 -m json.tool -{"jsonrpc":"2.0","method":"admin_deleteTier","params":{"name":"","admin_key":""},"id":1} -EOF -``` - -## Default Seeded Tiers - -These are created by the Alembic migration: - -| Tier | Requests/min | Requests/hr | Requests/day | -|------|-------------|-------------|--------------| -| free | 30 | 500 | 5,000 | -| pro | 120 | 3,000 | 50,000 | -| unlimited | 999,999 | 999,999 | 999,999 | - -## Common Errors - -| Error | Cause | -|-------|-------| -| `-32000` "Admin access required" | Missing or invalid `admin_key` on hosted deployment | -| `-32602` "Duplicate tier name" | Tier with that name already exists (unique constraint) | -| `-32602` "Cannot delete tier: N API key(s) still reference it" | Deactivate/delete keys first | -| `-32001` "Tier not found: X" | Tier name doesn't exist | - -## Important Notes - -- Always use `--data-binary @-` with heredoc (`<<'EOF'`) to avoid shell expansion issues with special characters in the admin key (e.g., `+`, `=`). -- Tier names must be unique and max 50 characters. -- Rate limits are enforced per sliding window (minute, hour, day) using Redis sorted sets. -- When rate limiting is disabled (`RATE_LIMIT_ENABLED=false`), tiers can still be managed but limits are not enforced. - -## Reference - -- Models: `backend/database_handler/models.py` (ApiTier class) -- Endpoints: `backend/protocol_rpc/endpoints.py` (admin_create_tier, admin_list_tiers) -- Migration: `backend/database_handler/migration/versions/b1c3e5f7a902_add_api_tiers_and_api_keys.py` diff --git a/.claude/skills/studio-db/SKILL.md b/.claude/skills/studio-db/SKILL.md deleted file mode 100644 index cbeec3826..000000000 --- a/.claude/skills/studio-db/SKILL.md +++ /dev/null @@ -1,155 +0,0 @@ ---- -name: studio-db -description: Query Studio deployment PostgreSQL databases for transaction debugging and analytics ---- - -# Studio Database Debugging - -Query and debug the `genlayer_state` database for transaction issues. - -## Connection - -For cluster access setup and connection workflow, see: -`devexp-apps-workload/.claude/skills/studio-db/SKILL.md` - -Quick connect (after setup): -```bash -export PATH="/opt/homebrew/share/google-cloud-sdk/bin:/opt/homebrew/bin:$PATH" -DBHOST=$(kubectl get secret database-config -n -o jsonpath='{.data.DBHOST}' | base64 -d) -DBPASSWORD=$(kubectl get secret database-password -n -o jsonpath='{.data.DBPASSWORD}' | base64 -d) - -echo "" | kubectl run pg-query --rm -i --image=postgres:16 --restart=Never \ - --namespace= --env="PGPASSWORD=$DBPASSWORD" -- \ - psql -h "$DBHOST" -p 5432 -U postgres -d genlayer_state -t -``` - -## Transaction Table Schema - -Primary table for debugging (`transactions`): - -| Column | Type | Description | -|--------|------|-------------| -| hash | VARCHAR(66) | Primary key, tx identifier | -| status | ENUM | Transaction lifecycle status | -| from_address | VARCHAR | Sender address | -| to_address | VARCHAR | Recipient/contract address | -| nonce | INTEGER | Tx sequence number | -| value | INTEGER | Transaction value | -| type | INTEGER | Tx type (0-3) | -| created_at | TIMESTAMP | When tx was created | -| input_data | JSONB | Contract call input | -| data | JSONB | Transaction data payload | -| consensus_data | JSONB | Final consensus results | -| consensus_history | JSONB | Full voting history | -| contract_snapshot | JSONB | Contract state at execution | -| appealed | BOOLEAN | Was tx appealed | -| appeal_failed | INTEGER | Appeal failure count | -| appeal_undetermined | BOOLEAN | Appeal resulted in undetermined | -| appeal_leader_timeout | BOOLEAN | Leader timeout during appeal | -| appeal_validators_timeout | BOOLEAN | Validators timeout during appeal | -| leader_timeout_validators | JSONB | Validators that timed out | -| worker_id | VARCHAR | Which worker processed tx | -| triggered_by_hash | VARCHAR(66) | Parent tx hash (for tx chains) | -| blocked_at | TIMESTAMP | When/if tx got blocked | - -## Transaction Statuses - -``` -PENDING → ACTIVATED → PROPOSING → COMMITTING → REVEALING → ACCEPTED → FINALIZED - ↘ UNDETERMINED - ↘ LEADER_TIMEOUT - ↘ VALIDATORS_TIMEOUT - ↘ CANCELED -``` - -## Other Tables - -- **current_state**: Global chain state (`id`, `data` JSONB, `balance`) -- **validators**: Validator configs (`id`, `address`, `stake`, `provider`, `model`, `plugin`) -- **llm_providers**: LLM provider configurations -- **snapshot**: Compressed state snapshots (`snapshot_id`, `state_data`, `transaction_data`) - -## Common Debugging Queries - -### Find transaction by hash -```sql -SELECT hash, status, from_address, to_address, created_at, - consensus_data, worker_id -FROM transactions WHERE hash = '0x...'; -``` - -### List stuck/problematic transactions -```sql -SELECT hash, status, created_at, worker_id -FROM transactions -WHERE status IN ('UNDETERMINED', 'LEADER_TIMEOUT', 'VALIDATORS_TIMEOUT') -ORDER BY created_at DESC LIMIT 20; -``` - -### Transaction status distribution -```sql -SELECT status, COUNT(*) as count -FROM transactions -GROUP BY status ORDER BY count DESC; -``` - -### Recent transactions -```sql -SELECT hash, status, from_address, created_at -FROM transactions ORDER BY created_at DESC LIMIT 10; -``` - -### Transactions with appeals -```sql -SELECT hash, status, appealed, appeal_undetermined, appeal_failed, - appeal_leader_timeout, appeal_validators_timeout -FROM transactions WHERE appealed = true -ORDER BY created_at DESC LIMIT 20; -``` - -### Find triggered transactions (tx chains) -```sql -SELECT t1.hash as parent, t2.hash as child, t2.status -FROM transactions t1 -JOIN transactions t2 ON t2.triggered_by_hash = t1.hash -WHERE t1.hash = '0x...'; -``` - -### Transactions by worker -```sql -SELECT worker_id, status, COUNT(*) -FROM transactions -WHERE created_at > NOW() - INTERVAL '24 hours' -GROUP BY worker_id, status -ORDER BY worker_id, count DESC; -``` - -### Check consensus history for a tx -```sql -SELECT hash, status, - jsonb_pretty(consensus_history) as history -FROM transactions WHERE hash = '0x...'; -``` - -## Write Operations - -**CAUTION**: Before any UPDATE/DELETE: -1. First SELECT to verify affected rows -2. Ask user for explicit confirmation -3. Prefer dev/stg for testing - -### Reset stuck transaction -```sql --- Verify first -SELECT hash, status FROM transactions -WHERE hash = '0x...' AND status = 'UNDETERMINED'; - --- Then update (after confirmation) -UPDATE transactions SET status = 'PENDING' WHERE hash = '0x...'; -``` - -## Full Schema - -Models: `backend/database_handler/models.py` -Migrations: `backend/database_handler/migration/versions/` -Tx processor: `backend/database_handler/transactions_processor.py` diff --git a/scripts/dump_transactions_to_metrics.py b/scripts/dump_transactions_to_metrics.py deleted file mode 100644 index 446fbd1a7..000000000 --- a/scripts/dump_transactions_to_metrics.py +++ /dev/null @@ -1,430 +0,0 @@ -#!/usr/bin/env python3 -""" -Dump historic transactions to the usage metrics service. - -This script connects to the PostgreSQL database and sends all finalized -transactions to the external usage metrics API, from the first transaction -up to (but excluding) a given transaction hash. - -Usage (from k8s pod): - python dump_transactions_to_metrics.py \ - --db-host 10.24.72.12 \ - --db-password "$DBPASSWORD" \ - --api-url "https://your-metrics-api.com" \ - --api-key "your-api-key" \ - --until-hash "0xabc123..." - - # Dry run (no API calls): - python dump_transactions_to_metrics.py \ - --db-host 10.24.72.12 \ - --db-password "$DBPASSWORD" \ - --dry-run \ - --until-hash "0xabc123..." - - # Process all finalized transactions (no hash limit): - python dump_transactions_to_metrics.py \ - --db-host 10.24.72.12 \ - --db-password "$DBPASSWORD" \ - --api-url "https://your-metrics-api.com" \ - --api-key "your-api-key" -""" - -import argparse -import json -import sys -from datetime import datetime -from typing import Optional - -import psycopg2 -import psycopg2.extras -import requests - - -# Transaction type mapping (from backend/domain/types.py) -TRANSACTION_TYPE_MAP = { - 0: "deploy", - 1: "write", - 2: "write", # SEND type treated as write - 3: "upgrade", -} - -# Transaction status mapping -TRANSACTION_STATUS_MAP = { - "ACCEPTED": "success", - "FINALIZED": "success", - "LEADER_TIMEOUT": "timeout", - "VALIDATORS_TIMEOUT": "timeout", - "UNDETERMINED": "undetermined", -} - - -def get_db_connection( - host: str, - password: str, - port: int = 5432, - user: str = "postgres", - database: str = "genlayer_state", -): - """Create database connection.""" - return psycopg2.connect( - host=host, - port=port, - user=user, - password=password, - database=database, - ) - - -def get_transaction_created_at(conn, tx_hash: str) -> Optional[datetime]: - """Get the created_at timestamp of a transaction by hash.""" - with conn.cursor() as cur: - cur.execute("SELECT created_at FROM transactions WHERE hash = %s", (tx_hash,)) - row = cur.fetchone() - if row: - return row[0] - return None - - -def fetch_finalized_transactions( - conn, - from_created_at: Optional[datetime] = None, - until_created_at: Optional[datetime] = None, - batch_size: int = 100, -): - """ - Fetch finalized transactions in batches. - - Yields batches of transaction dicts. - """ - query = """ - SELECT - hash, - from_address, - to_address, - type, - status, - created_at, - timestamp_awaiting_finalization, - appeal_processing_time, - consensus_data - FROM transactions - WHERE status = 'FINALIZED' - """ - params = [] - - if from_created_at: - query += " AND created_at > %s" - params.append(from_created_at) - - if until_created_at: - query += " AND created_at < %s" - params.append(until_created_at) - - query += " ORDER BY created_at ASC" - - # Use a named cursor for server-side cursor (avoids loading all rows into memory) - with conn.cursor( - name="tx_cursor", cursor_factory=psycopg2.extras.RealDictCursor - ) as cur: - cur.itersize = batch_size # Fetch this many rows at a time from server - cur.execute(query, params) - - batch = [] - for row in cur: - batch.append(dict(row)) - if len(batch) >= batch_size: - yield batch - batch = [] - - if batch: - yield batch - - -def calculate_processing_time_ms(tx: dict) -> int: - """Calculate processing time in milliseconds.""" - timestamp_awaiting = tx.get("timestamp_awaiting_finalization") - created_at = tx.get("created_at") - appeal_processing_time = tx.get("appeal_processing_time") or 0 - - if timestamp_awaiting is None or created_at is None: - return 0 - - try: - if isinstance(created_at, datetime): - created_at_epoch = created_at.timestamp() - else: - dt = datetime.fromisoformat(str(created_at).replace("Z", "+00:00")) - created_at_epoch = dt.timestamp() - - processing_seconds = float(timestamp_awaiting) - created_at_epoch - total_ms = int(processing_seconds * 1000) + int(appeal_processing_time * 1000) - return max(0, total_ms) - except Exception as e: - print(f"Warning: Failed to calculate processing time for {tx['hash']}: {e}") - return 0 - - -def extract_execution_result(tx: dict) -> str: - """ - Extract execution result from status and consensus_data. - - Returns one of: "success", "error", "timeout", "undetermined" - - Priority: - 1. Original status timeout (LEADER_TIMEOUT, VALIDATORS_TIMEOUT) -> "timeout" - 2. Original status undetermined (UNDETERMINED) -> "undetermined" - 3. execution_result from consensus_data.leader_receipt[0] -> "success" or "error" - 4. Default -> "success" - """ - # Check original status for timeout/undetermined - original_status = tx.get("status") - if original_status in ("LEADER_TIMEOUT", "VALIDATORS_TIMEOUT"): - return "timeout" - if original_status == "UNDETERMINED": - return "undetermined" - - # Try to get execution_result from leader receipt in consensus_data - consensus_data = tx.get("consensus_data") - if consensus_data is not None: - leader_receipts = consensus_data.get("leader_receipt", []) - if leader_receipts and len(leader_receipts) > 0: - first_receipt = leader_receipts[0] - if first_receipt is not None: - execution_result = first_receipt.get("execution_result") - if execution_result is not None: - # Handle string value - return str(execution_result).lower() - - # Default to success - return "success" - - -def extract_llm_calls(consensus_data: Optional[dict]) -> list: - """Extract LLM provider/model info from consensus_data.""" - llm_calls = [] - - if consensus_data is None: - return llm_calls - - def extract_from_receipt(receipt): - if receipt is None: - return None - - node_config = receipt.get("node_config") if isinstance(receipt, dict) else None - if node_config is None or not isinstance(node_config, dict): - return None - - primary_model = node_config.get("primary_model", {}) - if not primary_model: - return None - - provider = primary_model.get("provider", "unknown") - model = primary_model.get("model", "unknown") - - if provider == "unknown" and model == "unknown": - return None - - return { - "provider": provider, - "model": model, - "inputTokens": 0, - "outputTokens": 0, - "costUsd": 0, - } - - # Process leader receipts - leader_receipts = consensus_data.get("leader_receipt", []) - if leader_receipts: - for receipt in leader_receipts: - llm_call = extract_from_receipt(receipt) - if llm_call: - llm_calls.append(llm_call) - - # Process validator receipts - validators = consensus_data.get("validators", []) - if validators: - for receipt in validators: - llm_call = extract_from_receipt(receipt) - if llm_call: - llm_calls.append(llm_call) - - return llm_calls - - -def build_decision_payload(tx: dict) -> dict: - """Build the decision payload matching UsageMetricsService format.""" - tx_type = TRANSACTION_TYPE_MAP.get(tx.get("type"), "write") - tx_status = TRANSACTION_STATUS_MAP.get(tx.get("status"), "undetermined") - processing_time_ms = calculate_processing_time_ms(tx) - llm_calls = extract_llm_calls(tx.get("consensus_data")) - execution_result = extract_execution_result(tx) - - created_at = tx.get("created_at") - if isinstance(created_at, datetime): - created_at_iso = created_at.isoformat() - else: - created_at_iso = ( - str(created_at) if created_at else datetime.utcnow().isoformat() - ) - - return { - "externalId": tx["hash"], - "walletAddress": tx.get("from_address") - or "0x0000000000000000000000000000000000000000", - "contractAddress": tx.get("to_address"), - "type": tx_type, - "status": tx_status, - "processingTimeMs": processing_time_ms, - "createdAt": created_at_iso, - "llmCalls": llm_calls, - "result": execution_result, - } - - -def send_to_api(api_url: str, api_key: str, decisions: list) -> bool: - """Send decisions batch to the API.""" - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {api_key}", - } - - payload = {"decisions": decisions} - - try: - response = requests.post( - f"{api_url}/api/ingest", - json=payload, - headers=headers, - timeout=30, - ) - - if response.status_code != 200: - print( - f"Warning: API returned status {response.status_code}: {response.text[:200]}" - ) - return False - - return True - except requests.Timeout: - print("Warning: API request timed out") - return False - except Exception as e: - print(f"Error sending to API: {e}") - return False - - -def main(): - parser = argparse.ArgumentParser( - description="Dump historic transactions to usage metrics service" - ) - parser.add_argument("--db-host", required=True, help="Database host IP") - parser.add_argument("--db-port", type=int, default=5432, help="Database port") - parser.add_argument("--db-user", default="postgres", help="Database user") - parser.add_argument("--db-password", required=True, help="Database password") - parser.add_argument("--db-name", default="genlayer_state", help="Database name") - parser.add_argument( - "--api-url", help="Usage metrics API URL (required unless --dry-run)" - ) - parser.add_argument( - "--api-key", help="Usage metrics API key (required unless --dry-run)" - ) - parser.add_argument( - "--from-hash", help="Process transactions after this hash (exclusive)" - ) - parser.add_argument( - "--until-hash", help="Process transactions before this hash (exclusive)" - ) - parser.add_argument( - "--batch-size", type=int, default=50, help="Batch size for API calls" - ) - parser.add_argument( - "--dry-run", action="store_true", help="Print payloads without sending" - ) - parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") - - args = parser.parse_args() - - if not args.dry_run and (not args.api_url or not args.api_key): - parser.error( - "--api-url and --api-key are required unless --dry-run is specified" - ) - - # Connect to database - print(f"Connecting to database at {args.db_host}:{args.db_port}...") - conn = get_db_connection( - host=args.db_host, - port=args.db_port, - user=args.db_user, - password=args.db_password, - database=args.db_name, - ) - - # Get from_created_at if hash provided - from_created_at = None - if args.from_hash: - from_created_at = get_transaction_created_at(conn, args.from_hash) - if from_created_at is None: - print(f"Error: Transaction with hash {args.from_hash} not found") - sys.exit(1) - print( - f"Processing transactions after {from_created_at} (excluding hash: {args.from_hash})" - ) - - # Get until_created_at if hash provided - until_created_at = None - if args.until_hash: - until_created_at = get_transaction_created_at(conn, args.until_hash) - if until_created_at is None: - print(f"Error: Transaction with hash {args.until_hash} not found") - sys.exit(1) - print( - f"Processing transactions before {until_created_at} (excluding hash: {args.until_hash})" - ) - - if not args.from_hash and not args.until_hash: - print("Processing all finalized transactions") - - # Process transactions in batches - total_processed = 0 - total_sent = 0 - total_failed = 0 - - for batch in fetch_finalized_transactions( - conn, from_created_at, until_created_at, batch_size=args.batch_size - ): - decisions = [build_decision_payload(tx) for tx in batch] - total_processed += len(decisions) - - if args.dry_run: - if args.verbose: - for decision in decisions: - print(json.dumps(decision, indent=2, default=str)) - print( - f"[DRY RUN] Would send {len(decisions)} decisions (total: {total_processed})" - ) - else: - success = send_to_api(args.api_url, args.api_key, decisions) - if success: - total_sent += len(decisions) - print(f"Sent {len(decisions)} decisions (total sent: {total_sent})") - else: - total_failed += len(decisions) - print( - f"Failed to send {len(decisions)} decisions (total failed: {total_failed})" - ) - - conn.close() - - # Summary - print("\n" + "=" * 50) - print("Summary:") - print(f" Total processed: {total_processed}") - if args.dry_run: - print(" Mode: DRY RUN (no API calls made)") - else: - print(f" Successfully sent: {total_sent}") - print(f" Failed: {total_failed}") - - -if __name__ == "__main__": - main() diff --git a/scripts/run_dump_transactions.sh b/scripts/run_dump_transactions.sh deleted file mode 100755 index 286deeb0b..000000000 --- a/scripts/run_dump_transactions.sh +++ /dev/null @@ -1,177 +0,0 @@ -#!/bin/bash -# -# Run the dump_transactions_to_metrics.py script from a k8s pod. -# -# Usage: -# ./run_dump_transactions.sh [options] -# -# Examples: -# # Dry run on staging -# ./run_dump_transactions.sh studio-stg --dry-run -# -# # Process all transactions on production -# ./run_dump_transactions.sh studio-prd \ -# --api-url "https://your-metrics-api.com" \ -# --api-key "your-api-key" -# -# # Process all transactions before a specific hash (exclusive) -# ./run_dump_transactions.sh studio-prd \ -# --api-url "https://your-metrics-api.com" \ -# --api-key "your-api-key" \ -# --until-hash "0xabc123..." -# - -set -e - -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -PYTHON_SCRIPT="$SCRIPT_DIR/dump_transactions_to_metrics.py" -POD_NAME="dump-transactions-pod" - -if [ -z "$1" ]; then - echo "Usage: $0 [script-options]" - echo "" - echo "Namespaces:" - echo " studio-dev -> simulator-dev cluster (DB: 10.127.216.9)" - echo " studio-stg -> simulator-dev cluster (DB: 10.127.216.12)" - echo " rally-studio-dev -> simulator-dev cluster (DB: 10.127.216.14)" - echo " studio-prd -> simulator-prd cluster (DB: 10.24.72.12)" - echo " rally-studio-prd -> rally-prd cluster (DB: 10.24.72.22)" - echo "" - echo "Script options (passed to Python script):" - echo " --api-url URL Usage metrics API URL" - echo " --api-key KEY Usage metrics API key" - echo " --until-hash HASH Process transactions before this hash (exclusive)" - echo " --batch-size N Batch size for API calls (default: 50)" - echo " --dry-run Print payloads without sending" - echo " --verbose Verbose output" - exit 1 -fi - -ENV="$1" -shift # Remove namespace from args, rest are script options - -# Select k8s context and DB IP based on namespace -case "$ENV" in - studio-dev) - CTX="gke_simulator-dev-473709_europe-west4_simulator-dev" - DB_IP="10.127.216.9" - ;; - studio-stg) - CTX="gke_simulator-dev-473709_europe-west4_simulator-dev" - DB_IP="10.127.216.12" - ;; - rally-studio-dev) - CTX="gke_simulator-dev-473709_europe-west4_simulator-dev" - DB_IP="10.127.216.14" - ;; - studio-prd) - CTX="gke_simulator-440803_europe-west4_simulator-prd" - DB_IP="10.24.72.12" - ;; - rally-studio-prd) - CTX="gke_simulator-440803_europe-west4_rally-prd" - DB_IP="10.24.72.22" - ;; - *) - echo "Error: Unknown namespace '$ENV'" - echo "Valid namespaces: studio-dev, studio-stg, rally-studio-dev, studio-prd, rally-studio-prd" - exit 1 - ;; -esac - -echo "=== Configuration ===" -echo "Namespace: $ENV" -echo "Context: $CTX" -echo "DB IP: $DB_IP" -echo "====================" - -# Switch context and namespace -echo "Switching k8s context to: $CTX" -kubectx "$CTX" || exit 1 - -echo "Switching namespace to: $ENV" -kubens "$ENV" || exit 1 - -# Get DB password from secret -echo "Fetching DB password from 'database-password' secret..." -DBPASSWORD=$(kubectl get secret database-password -o jsonpath='{.data.DBPASSWORD}' | base64 -d) || { - echo "Failed to fetch DB password" - exit 1 -} - -# Clean up any existing pod -echo "Cleaning up any existing '$POD_NAME' pod..." -kubectl delete pod "$POD_NAME" --ignore-not-found >/dev/null 2>&1 || true -kubectl wait pod "$POD_NAME" --for=delete --timeout=30s >/dev/null 2>&1 || true - -# Create a ConfigMap with the Python script -echo "Creating ConfigMap with Python script..." -kubectl delete configmap dump-transactions-script --ignore-not-found >/dev/null 2>&1 || true -kubectl create configmap dump-transactions-script --from-file=script.py="$PYTHON_SCRIPT" - -# Build the script arguments (escape for shell) -SCRIPT_ARGS="--db-host $DB_IP --db-password \$DBPASSWORD $*" - -echo "Running script in pod..." -echo "Arguments: $SCRIPT_ARGS" -echo "" - -# Create temporary pod manifest -POD_MANIFEST=$(mktemp) -cat > "$POD_MANIFEST" </dev/null || true - -# Stream logs (will block until pod completes) -kubectl logs -f "$POD_NAME" || true - -# Wait for completion and get exit code -echo "" -echo "Waiting for pod to complete..." -kubectl wait pod "$POD_NAME" --for=jsonpath='{.status.phase}'=Succeeded --timeout=3600s 2>/dev/null || \ -kubectl wait pod "$POD_NAME" --for=jsonpath='{.status.phase}'=Failed --timeout=10s 2>/dev/null || true - -# Get exit code -EXIT_CODE=$(kubectl get pod "$POD_NAME" -o jsonpath='{.status.containerStatuses[0].state.terminated.exitCode}' 2>/dev/null || echo "unknown") -echo "Pod exit code: $EXIT_CODE" - -# Cleanup -echo "Cleaning up..." -kubectl delete pod "$POD_NAME" --ignore-not-found >/dev/null 2>&1 || true -kubectl delete configmap dump-transactions-script --ignore-not-found >/dev/null 2>&1 || true - -echo "Done!" diff --git a/scripts/run_update_error_transactions.sh b/scripts/run_update_error_transactions.sh deleted file mode 100755 index b559e868b..000000000 --- a/scripts/run_update_error_transactions.sh +++ /dev/null @@ -1,184 +0,0 @@ -#!/bin/bash -# -# Run the update_error_transactions_metrics.py script from a k8s pod. -# -# Usage: -# ./run_update_error_transactions.sh [options] -# -# Examples: -# # Dry run on staging -# ./run_update_error_transactions.sh studio-stg --dry-run -# -# # Process all error/undetermined transactions on production -# ./run_update_error_transactions.sh studio-prd \ -# --api-url "https://your-metrics-api.com" \ -# --api-key "your-api-key" -# -# # Process transactions after a specific hash (exclusive) -# ./run_update_error_transactions.sh studio-prd \ -# --api-url "https://your-metrics-api.com" \ -# --api-key "your-api-key" \ -# --from-hash "0xabc123..." -# -# # Process transactions before a specific hash (exclusive) -# ./run_update_error_transactions.sh studio-prd \ -# --api-url "https://your-metrics-api.com" \ -# --api-key "your-api-key" \ -# --until-hash "0xabc123..." -# - -set -e - -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -PYTHON_SCRIPT="$SCRIPT_DIR/update_error_transactions_metrics.py" -POD_NAME="update-error-transactions-pod" - -if [ -z "$1" ]; then - echo "Usage: $0 [script-options]" - echo "" - echo "Namespaces:" - echo " studio-dev -> simulator-dev cluster (DB: 10.127.216.9)" - echo " studio-stg -> simulator-dev cluster (DB: 10.127.216.12)" - echo " rally-studio-dev -> simulator-dev cluster (DB: 10.127.216.14)" - echo " studio-prd -> simulator-prd cluster (DB: 10.24.72.12)" - echo " rally-studio-prd -> rally-prd cluster (DB: 10.24.72.22)" - echo "" - echo "Script options (passed to Python script):" - echo " --api-url URL Usage metrics API URL" - echo " --api-key KEY Usage metrics API key" - echo " --from-hash HASH Process transactions after this hash (exclusive)" - echo " --until-hash HASH Process transactions before this hash (exclusive)" - echo " --batch-size N Batch size for API calls (default: 50)" - echo " --dry-run Print payloads without sending" - echo " --verbose Verbose output" - exit 1 -fi - -ENV="$1" -shift # Remove namespace from args, rest are script options - -# Select k8s context and DB IP based on namespace -case "$ENV" in - studio-dev) - CTX="gke_simulator-dev-473709_europe-west4_simulator-dev" - DB_IP="10.127.216.9" - ;; - studio-stg) - CTX="gke_simulator-dev-473709_europe-west4_simulator-dev" - DB_IP="10.127.216.12" - ;; - rally-studio-dev) - CTX="gke_simulator-dev-473709_europe-west4_simulator-dev" - DB_IP="10.127.216.14" - ;; - studio-prd) - CTX="gke_simulator-440803_europe-west4_simulator-prd" - DB_IP="10.24.72.12" - ;; - rally-studio-prd) - CTX="gke_simulator-440803_europe-west4_rally-prd" - DB_IP="10.24.72.22" - ;; - *) - echo "Error: Unknown namespace '$ENV'" - echo "Valid namespaces: studio-dev, studio-stg, rally-studio-dev, studio-prd, rally-studio-prd" - exit 1 - ;; -esac - -echo "=== Configuration ===" -echo "Namespace: $ENV" -echo "Context: $CTX" -echo "DB IP: $DB_IP" -echo "====================" - -# Switch context and namespace -echo "Switching k8s context to: $CTX" -kubectx "$CTX" || exit 1 - -echo "Switching namespace to: $ENV" -kubens "$ENV" || exit 1 - -# Get DB password from secret -echo "Fetching DB password from 'database-password' secret..." -DBPASSWORD=$(kubectl get secret database-password -o jsonpath='{.data.DBPASSWORD}' | base64 -d) || { - echo "Failed to fetch DB password" - exit 1 -} - -# Clean up any existing pod -echo "Cleaning up any existing '$POD_NAME' pod..." -kubectl delete pod "$POD_NAME" --ignore-not-found >/dev/null 2>&1 || true -kubectl wait pod "$POD_NAME" --for=delete --timeout=30s >/dev/null 2>&1 || true - -# Create a ConfigMap with the Python script -echo "Creating ConfigMap with Python script..." -kubectl delete configmap update-error-transactions-script --ignore-not-found >/dev/null 2>&1 || true -kubectl create configmap update-error-transactions-script --from-file=script.py="$PYTHON_SCRIPT" - -# Build the script arguments (escape for shell) -SCRIPT_ARGS="--db-host $DB_IP --db-password \$DBPASSWORD $*" - -echo "Running script in pod..." -echo "Arguments: $SCRIPT_ARGS" -echo "" - -# Create temporary pod manifest -POD_MANIFEST=$(mktemp) -cat > "$POD_MANIFEST" </dev/null || true - -# Stream logs (will block until pod completes) -kubectl logs -f "$POD_NAME" || true - -# Wait for completion and get exit code -echo "" -echo "Waiting for pod to complete..." -kubectl wait pod "$POD_NAME" --for=jsonpath='{.status.phase}'=Succeeded --timeout=86400s 2>/dev/null || \ -kubectl wait pod "$POD_NAME" --for=jsonpath='{.status.phase}'=Failed --timeout=60s 2>/dev/null || true - -# Get exit code -EXIT_CODE=$(kubectl get pod "$POD_NAME" -o jsonpath='{.status.containerStatuses[0].state.terminated.exitCode}' 2>/dev/null || echo "unknown") -echo "Pod exit code: $EXIT_CODE" - -# Cleanup -echo "Cleaning up..." -kubectl delete pod "$POD_NAME" --ignore-not-found >/dev/null 2>&1 || true -kubectl delete configmap update-error-transactions-script --ignore-not-found >/dev/null 2>&1 || true - -echo "Done!" diff --git a/scripts/update_error_transactions_metrics.py b/scripts/update_error_transactions_metrics.py deleted file mode 100755 index 8bcedf223..000000000 --- a/scripts/update_error_transactions_metrics.py +++ /dev/null @@ -1,687 +0,0 @@ -#!/usr/bin/env python3 -""" -Update error/timeout/undetermined transactions in the usage metrics service. - -This script connects to the PostgreSQL database and sends transactions that have -result="error", "timeout", or "undetermined" to the external usage metrics API -to update their records with the new "result" field. - -Optimized for large datasets (70k+ transactions) with: -- SQL-side filtering to only load error/undetermined transactions -- Pagination with garbage collection -- Progress reporting and retry logic -- Real-time output flushing - -Usage (from k8s pod): - python update_error_transactions_metrics.py \ - --db-host 10.24.72.12 \ - --db-password "$DBPASSWORD" \ - --api-url "https://your-metrics-api.com" \ - --api-key "your-api-key" - - # Dry run (no API calls): - python update_error_transactions_metrics.py \ - --db-host 10.24.72.12 \ - --db-password "$DBPASSWORD" \ - --dry-run - - # Process transactions after a specific hash (exclusive): - python update_error_transactions_metrics.py \ - --db-host 10.24.72.12 \ - --db-password "$DBPASSWORD" \ - --api-url "https://your-metrics-api.com" \ - --api-key "your-api-key" \ - --from-hash "0xabc123..." - - # Process transactions before a specific hash (exclusive): - python update_error_transactions_metrics.py \ - --db-host 10.24.72.12 \ - --db-password "$DBPASSWORD" \ - --api-url "https://your-metrics-api.com" \ - --api-key "your-api-key" \ - --until-hash "0xabc123..." -""" - -import argparse -import json -import sys -import time -from datetime import datetime -from typing import Optional - -import psycopg2 -import psycopg2.extras -import requests - - -def log(message: str): - """Print message with timestamp and flush immediately.""" - timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - print(f"[{timestamp}] {message}", flush=True) - - -def format_duration(seconds: float) -> str: - """Format duration in human-readable format.""" - if seconds < 60: - return f"{seconds:.1f}s" - elif seconds < 3600: - minutes = int(seconds // 60) - secs = int(seconds % 60) - return f"{minutes}m {secs}s" - else: - hours = int(seconds // 3600) - minutes = int((seconds % 3600) // 60) - return f"{hours}h {minutes}m" - - -# Transaction type mapping (from backend/domain/types.py) -TRANSACTION_TYPE_MAP = { - 0: "deploy", - 1: "write", - 2: "write", # SEND type treated as write - 3: "upgrade", -} - - -def get_db_connection( - host: str, - password: str, - port: int = 5432, - user: str = "postgres", - database: str = "genlayer_state", -): - """Create database connection.""" - return psycopg2.connect( - host=host, - port=port, - user=user, - password=password, - database=database, - ) - - -def get_transaction_created_at(conn, tx_hash: str) -> Optional[datetime]: - """Get the created_at timestamp of a transaction by hash.""" - with conn.cursor() as cur: - cur.execute("SELECT created_at FROM transactions WHERE hash = %s", (tx_hash,)) - row = cur.fetchone() - if row: - return row[0] - return None - - -def get_total_transaction_count( - conn, - from_created_at: Optional[datetime] = None, - until_created_at: Optional[datetime] = None, -) -> int: - """Get total count of finalized transactions to scan.""" - query = "SELECT COUNT(*) FROM transactions WHERE status = 'FINALIZED'" - params = [] - - if from_created_at: - query += " AND created_at > %s" - params.append(from_created_at) - - if until_created_at: - query += " AND created_at < %s" - params.append(until_created_at) - - with conn.cursor() as cur: - cur.execute(query, params) - return cur.fetchone()[0] - - -def is_error_transaction(tx: dict) -> bool: - """ - Check if a transaction has an error result (Python-side filtering). - - Returns True if the transaction has: - - ERROR execution result in leader receipt - - NULL/empty leader receipt - - Undetermined consensus round in history - """ - consensus_data = tx.get("consensus_data") - consensus_history = tx.get("consensus_history") - - # Check for ERROR execution result in leader receipt - if consensus_data is not None: - leader_receipts = consensus_data.get("leader_receipt", []) - # Handle both list and dict formats for leader_receipt - if isinstance(leader_receipts, list): - if leader_receipts and len(leader_receipts) > 0: - first_receipt = leader_receipts[0] - if first_receipt is not None and isinstance(first_receipt, dict): - execution_result = first_receipt.get("execution_result") - if execution_result is not None: - if str(execution_result).upper() == "ERROR": - return True - else: - # leader_receipt[0] is NULL - this is an edge case error - return True - else: - # leader_receipt is empty or missing - return True - elif isinstance(leader_receipts, dict): - # leader_receipt is a dict - check execution_result directly - execution_result = leader_receipts.get("execution_result") - if execution_result is not None: - if str(execution_result).upper() == "ERROR": - return True - # Also check if it's empty or has no execution_result - if not leader_receipts: - return True - else: - # Unknown format - treat as error - return True - else: - # No consensus_data - edge case - return True - - # Check for Undetermined consensus round in history - if ( - consensus_history is not None - and "consensus_results" in consensus_history - and len(consensus_history["consensus_results"]) > 0 - ): - last_round = consensus_history["consensus_results"][-1] - if last_round.get("consensus_round") == "Undetermined": - return True - - return False - - -def fetch_error_transactions( - conn, - from_created_at: Optional[datetime] = None, - until_created_at: Optional[datetime] = None, - batch_size: int = 50, - total_to_scan: int = 0, - page_size: int = 100, -): - """ - Fetch finalized transactions that have error results or undetermined consensus. - - Uses a server-side cursor with Python-side filtering for memory efficiency. - The SQL query is simple and fast, filtering is done in Python. - - Yields tuples of (batch, scanned_count) where batch is a list of transaction dicts. - """ - import gc - - # Simple SQL query - just fetch finalized transactions - # Filtering is done in Python which is much faster than JSONB text conversion - base_query = """ - SELECT - hash, - from_address, - to_address, - type, - status, - created_at, - timestamp_awaiting_finalization, - appeal_processing_time, - consensus_data, - consensus_history - FROM transactions - WHERE status = 'FINALIZED' - """ - base_params = [] - - if from_created_at: - base_query += " AND created_at > %s" - base_params.append(from_created_at) - - if until_created_at: - base_query += " AND created_at < %s" - base_params.append(until_created_at) - - base_query += " ORDER BY created_at ASC" - - log(" Using server-side cursor with Python-side filtering...") - - batch = [] - matched_count = 0 - scanned_count = 0 - last_progress_report = 0 - - # Use a named server-side cursor to stream results without loading all into memory - with conn.cursor( - name="error_tx_cursor", cursor_factory=psycopg2.extras.RealDictCursor - ) as cur: - cur.itersize = page_size # Fetch this many rows at a time from server - cur.execute(base_query, base_params) - - for row in cur: - scanned_count += 1 - tx = dict(row) - - # Filter in Python - check if this is an error transaction - if is_error_transaction(tx): - matched_count += 1 - batch.append(tx) - - if len(batch) >= batch_size: - yield batch, matched_count - batch = [] - gc.collect() - - # Progress report every 5,000 transactions - if scanned_count - last_progress_report >= 5000: - progress_pct = ( - (scanned_count / total_to_scan * 100) if total_to_scan > 0 else 0 - ) - log( - f" Scanned {scanned_count:,}/{total_to_scan:,} ({progress_pct:.1f}%) - Found {matched_count:,} error transactions" - ) - last_progress_report = scanned_count - gc.collect() - - log( - f" Scan complete: {scanned_count:,} transactions scanned, {matched_count:,} error transactions found" - ) - - if batch: - yield batch, matched_count - - -def extract_execution_result(tx: dict) -> str: - """ - Extract execution result from status, consensus_data, and consensus_history. - - Returns one of: "success", "error", "timeout", "undetermined" - - Priority: - 1. Original status timeout (LEADER_TIMEOUT, VALIDATORS_TIMEOUT) -> "timeout" - 2. Original status undetermined (UNDETERMINED) -> "undetermined" - 3. consensus_history.consensus_results[-1].consensus_round == "Undetermined" -> "undetermined" - 4. execution_result from consensus_data.leader_receipt[0] -> "success" or "error" - 5. Default -> "success" - """ - consensus_data = tx.get("consensus_data") - consensus_history = tx.get("consensus_history") - - # Check original status for timeout/undetermined - original_status = tx.get("status") - if original_status in ("LEADER_TIMEOUT", "VALIDATORS_TIMEOUT"): - return "timeout" - if original_status == "UNDETERMINED": - return "undetermined" - - # Check if consensus was undetermined from consensus_history - if ( - consensus_history is not None - and "consensus_results" in consensus_history - and len(consensus_history["consensus_results"]) > 0 - ): - last_round = consensus_history["consensus_results"][-1] - if last_round.get("consensus_round") == "Undetermined": - return "undetermined" - - # Try to get execution_result from leader receipt - if consensus_data is not None: - leader_receipts = consensus_data.get("leader_receipt", []) - # Handle both list and dict formats for leader_receipt - if isinstance(leader_receipts, list): - if leader_receipts and len(leader_receipts) > 0: - first_receipt = leader_receipts[0] - if first_receipt is not None and isinstance(first_receipt, dict): - execution_result = first_receipt.get("execution_result") - if execution_result is not None: - # Convert to lowercase for consistency (stored as "SUCCESS" or "ERROR") - return str(execution_result).lower() - elif isinstance(leader_receipts, dict): - # leader_receipt is a dict - check execution_result directly - execution_result = leader_receipts.get("execution_result") - if execution_result is not None: - return str(execution_result).lower() - - # Default to success - return "success" - - -def calculate_processing_time_ms(tx: dict) -> int: - """Calculate processing time in milliseconds.""" - timestamp_awaiting = tx.get("timestamp_awaiting_finalization") - created_at = tx.get("created_at") - appeal_processing_time = tx.get("appeal_processing_time") or 0 - - if timestamp_awaiting is None or created_at is None: - return 0 - - try: - if isinstance(created_at, datetime): - created_at_epoch = created_at.timestamp() - else: - dt = datetime.fromisoformat(str(created_at).replace("Z", "+00:00")) - created_at_epoch = dt.timestamp() - - processing_seconds = float(timestamp_awaiting) - created_at_epoch - total_ms = int(processing_seconds * 1000) + int(appeal_processing_time * 1000) - return max(0, total_ms) - except Exception as e: - print(f"Warning: Failed to calculate processing time for {tx['hash']}: {e}") - return 0 - - -def extract_llm_calls(consensus_data: Optional[dict]) -> list: - """Extract LLM provider/model info from consensus_data.""" - llm_calls = [] - - if consensus_data is None: - return llm_calls - - def extract_from_receipt(receipt): - if receipt is None: - return None - - node_config = receipt.get("node_config") if isinstance(receipt, dict) else None - if node_config is None or not isinstance(node_config, dict): - return None - - primary_model = node_config.get("primary_model", {}) - if not primary_model: - return None - - provider = primary_model.get("provider", "unknown") - model = primary_model.get("model", "unknown") - - if provider == "unknown" and model == "unknown": - return None - - return { - "provider": provider, - "model": model, - "inputTokens": 0, - "outputTokens": 0, - "costUsd": 0, - } - - # Process leader receipts - leader_receipts = consensus_data.get("leader_receipt", []) - if leader_receipts: - for receipt in leader_receipts: - llm_call = extract_from_receipt(receipt) - if llm_call: - llm_calls.append(llm_call) - - # Process validator receipts - validators = consensus_data.get("validators", []) - if validators: - for receipt in validators: - llm_call = extract_from_receipt(receipt) - if llm_call: - llm_calls.append(llm_call) - - return llm_calls - - -def build_decision_payload(tx: dict) -> dict: - """Build the decision payload matching UsageMetricsService format.""" - tx_type = TRANSACTION_TYPE_MAP.get(tx.get("type"), "write") - # Status is always "success" since we only process finalized transactions - tx_status = "success" - processing_time_ms = calculate_processing_time_ms(tx) - llm_calls = extract_llm_calls(tx.get("consensus_data")) - # Result can be: success, error, timeout, undetermined - execution_result = extract_execution_result(tx) - - created_at = tx.get("created_at") - if isinstance(created_at, datetime): - created_at_iso = created_at.isoformat() - else: - created_at_iso = ( - str(created_at) if created_at else datetime.utcnow().isoformat() - ) - - return { - "externalId": tx["hash"], - "walletAddress": tx.get("from_address") - or "0x0000000000000000000000000000000000000000", - "contractAddress": tx.get("to_address"), - "type": tx_type, - "status": tx_status, - "processingTimeMs": processing_time_ms, - "createdAt": created_at_iso, - "llmCalls": llm_calls, - "result": execution_result, - } - - -def send_to_api( - api_url: str, api_key: str, decisions: list, max_retries: int = 3 -) -> tuple[bool, str]: - """ - Send decisions batch to the API with retry logic. - - Returns tuple of (success, error_message). - """ - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {api_key}", - } - - payload = {"decisions": decisions} - - for attempt in range(1, max_retries + 1): - try: - response = requests.post( - f"{api_url}/api/ingest", - json=payload, - headers=headers, - timeout=60, # Increased timeout for large batches - ) - - if response.status_code == 200: - return True, "" - - error_msg = ( - f"API returned status {response.status_code}: {response.text[:200]}" - ) - if attempt < max_retries: - log(f" Retry {attempt}/{max_retries}: {error_msg}") - time.sleep(2**attempt) # Exponential backoff - else: - return False, error_msg - - except requests.Timeout: - error_msg = "API request timed out" - if attempt < max_retries: - log(f" Retry {attempt}/{max_retries}: {error_msg}") - time.sleep(2**attempt) - else: - return False, error_msg - - except Exception as e: - error_msg = f"Error: {e}" - if attempt < max_retries: - log(f" Retry {attempt}/{max_retries}: {error_msg}") - time.sleep(2**attempt) - else: - return False, error_msg - - return False, "Max retries exceeded" - - -def main(): - parser = argparse.ArgumentParser( - description="Update error/timeout/undetermined transactions in usage metrics service" - ) - parser.add_argument("--db-host", required=True, help="Database host IP") - parser.add_argument("--db-port", type=int, default=5432, help="Database port") - parser.add_argument("--db-user", default="postgres", help="Database user") - parser.add_argument("--db-password", required=True, help="Database password") - parser.add_argument("--db-name", default="genlayer_state", help="Database name") - parser.add_argument( - "--api-url", help="Usage metrics API URL (required unless --dry-run)" - ) - parser.add_argument( - "--api-key", help="Usage metrics API key (required unless --dry-run)" - ) - parser.add_argument( - "--from-hash", help="Process transactions after this hash (exclusive)" - ) - parser.add_argument( - "--until-hash", help="Process transactions before this hash (exclusive)" - ) - parser.add_argument( - "--batch-size", type=int, default=50, help="Batch size for API calls" - ) - parser.add_argument( - "--dry-run", action="store_true", help="Print payloads without sending" - ) - parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") - - args = parser.parse_args() - - if not args.dry_run and (not args.api_url or not args.api_key): - parser.error( - "--api-url and --api-key are required unless --dry-run is specified" - ) - - # Start timing - start_time = time.time() - - # Connect to database - log(f"Connecting to database at {args.db_host}:{args.db_port}...") - conn = get_db_connection( - host=args.db_host, - port=args.db_port, - user=args.db_user, - password=args.db_password, - database=args.db_name, - ) - log("Database connection established") - - # Get from_created_at if hash provided - from_created_at = None - if args.from_hash: - from_created_at = get_transaction_created_at(conn, args.from_hash) - if from_created_at is None: - log(f"ERROR: Transaction with hash {args.from_hash} not found") - sys.exit(1) - log( - f"Processing transactions after {from_created_at} (excluding hash: {args.from_hash})" - ) - - # Get until_created_at if hash provided - until_created_at = None - if args.until_hash: - until_created_at = get_transaction_created_at(conn, args.until_hash) - if until_created_at is None: - log(f"ERROR: Transaction with hash {args.until_hash} not found") - sys.exit(1) - log( - f"Processing transactions before {until_created_at} (excluding hash: {args.until_hash})" - ) - - # Get total count for progress tracking - log("Counting total transactions to scan...") - total_to_scan = get_total_transaction_count(conn, from_created_at, until_created_at) - log(f"Total finalized transactions to scan: {total_to_scan:,}") - - if total_to_scan == 0: - log("No transactions to process") - conn.close() - return - - log( - "Scanning for transactions with result='error', 'timeout', or 'undetermined'..." - ) - if args.dry_run: - log("MODE: DRY RUN (no API calls will be made)") - - # Process transactions in batches - total_matched = 0 - total_sent = 0 - total_failed = 0 - batch_count = 0 - failed_hashes = [] - result_counts = {"error": 0, "timeout": 0, "undetermined": 0} - - for batch, matched_count in fetch_error_transactions( - conn, - from_created_at, - until_created_at, - batch_size=args.batch_size, - total_to_scan=total_to_scan, - ): - batch_count += 1 - total_matched = matched_count - decisions = [build_decision_payload(tx) for tx in batch] - - # Count results by type - for tx in batch: - result = extract_execution_result(tx) - if result in result_counts: - result_counts[result] += 1 - - # Calculate progress - elapsed = time.time() - start_time - - if args.dry_run: - if args.verbose: - for decision in decisions: - print(json.dumps(decision, indent=2, default=str), flush=True) - log( - f"[DRY RUN] Batch {batch_count}: Would send {len(decisions)} decisions | " - f"Total matched: {total_matched:,} | Elapsed: {format_duration(elapsed)}" - ) - else: - success, error_msg = send_to_api(args.api_url, args.api_key, decisions) - if success: - total_sent += len(decisions) - log( - f"Batch {batch_count}: Sent {len(decisions)} decisions | " - f"Total sent: {total_sent:,} | Total matched: {total_matched:,} | " - f"Elapsed: {format_duration(elapsed)}" - ) - else: - total_failed += len(decisions) - # Track failed transaction hashes - for tx in batch: - failed_hashes.append(tx["hash"]) - log( - f"Batch {batch_count}: FAILED to send {len(decisions)} decisions | " - f"Error: {error_msg} | Total failed: {total_failed:,}" - ) - - conn.close() - - # Final timing - total_elapsed = time.time() - start_time - - # Summary - print("\n" + "=" * 60, flush=True) - log("SUMMARY") - print("=" * 60, flush=True) - print( - f" Total matched (error/timeout/undetermined): {total_matched:,}", flush=True - ) - print(f" - Errors: {result_counts['error']:,}", flush=True) - print(f" - Timeouts: {result_counts['timeout']:,}", flush=True) - print(f" - Undetermined: {result_counts['undetermined']:,}", flush=True) - print(f" Total batches processed: {batch_count}", flush=True) - print(f" Total time: {format_duration(total_elapsed)}", flush=True) - - if args.dry_run: - print(" Mode: DRY RUN (no API calls made)", flush=True) - else: - print(f" Successfully sent: {total_sent:,}", flush=True) - print(f" Failed: {total_failed:,}", flush=True) - if total_matched > 0: - success_rate = (total_sent / total_matched) * 100 - print(f" Success rate: {success_rate:.1f}%", flush=True) - - if failed_hashes: - print(f"\n Failed transaction hashes (first 10):", flush=True) - for h in failed_hashes[:10]: - print(f" - {h}", flush=True) - if len(failed_hashes) > 10: - print(f" ... and {len(failed_hashes) - 10} more", flush=True) - - print("=" * 60, flush=True) - - -if __name__ == "__main__": - main()