Tari Mining Pool Block Height Sync Checker

gistfile1.txt

#!/usr/bin/env python3
"""
Quick Pool Sync Checker

A simple, self-contained script to check mining pool block height sync status.
No external dependencies required - only uses standard library + requests.

Usage:
    python quick_pool_check.py              # Default tolerance: ±2 blocks
    python quick_pool_check.py --tolerance 3 # Custom tolerance: ±3 blocks
    python quick_pool_check.py --help       # Show help message
"""

import requests
import argparse
import sys
import socket
import json
import asyncio
import time
from datetime import datetime

def get_main_network_block_height():
    """
    Get the current mainnet block height from Tari explorer.
    
    Returns:
        int: Current block height
    """
    url = "https://textexplore.tari.com/?json"
    response = requests.get(url)
    response.raise_for_status()
    data = response.json()
    return int(data['tipInfo']['metadata']['best_block_height'])

def test_pool_connectivity(hostname, port=1338, timeout=5):
    """
    Test TCP connectivity to a mining pool and attempt JSON-RPC communication.
    
    Args:
        hostname (str): Pool hostname
        port (int): Pool port (default: 1338 for Tari pools)
        timeout (int): Connection timeout in seconds
        
    Returns:
        dict: Connection test results
    """
    result = {
        'hostname': hostname,
        'port': port,
        'tcp_connect': False,
        'jsonrpc_response': False,
        'latency_ms': None,
        'error': None
    }
    
    try:
        start_time = time.time()
        
        # Test TCP connection
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        
        connect_result = sock.connect_ex((hostname, port))
        if connect_result == 0:
            result['tcp_connect'] = True
            result['latency_ms'] = round((time.time() - start_time) * 1000, 2)
            
            # Test JSON-RPC communication (SHA3x mining pool login request)
            login_request = {
                "id": 1,
                "jsonrpc": "2.0",
                "method": "login",
                "params": {
                    "agent": "sha3x-miner/3.0",
                    "algo": ["sha3x"],
                    "login": "12BS6NPHQWS7Xg66dDj5o7gC1t3dSvwZAqgxXsuEwwgZWUVY5LYDeWCRpbm1bTjdbvpyKJr3S1tDry6NXMfkLDZacJR",
                    "pass": "worker1"
                }
            }
            
            try:
                message = json.dumps(login_request) + "\n"
                sock.send(message.encode('utf-8'))
                
                # Try to read response (with short timeout)
                sock.settimeout(2)
                response_data = sock.recv(1024)
                
                if response_data:
                    try:
                        response_text = response_data.decode('utf-8').strip()
                        if response_text and ('"jsonrpc"' in response_text or '"result"' in response_text or '"error"' in response_text):
                            result['jsonrpc_response'] = True
                    except UnicodeDecodeError:
                        pass  # Binary response, but at least we got something
                        
            except (socket.timeout, socket.error):
                pass  # JSON-RPC test failed, but TCP worked
                
        else:
            result['error'] = f"TCP connection failed (code: {connect_result})"
            
        sock.close()
        
    except socket.gaierror as e:
        result['error'] = f"DNS resolution failed: {e}"
    except socket.timeout:
        result['error'] = "Connection timeout"
    except Exception as e:
        result['error'] = f"Unexpected error: {e}"
    
    return result

def extract_pool_hostnames(pool_data):
    """
    Extract unique hostnames from pool data for connectivity testing.
    
    Args:
        pool_data (dict): Pool data from API
        
    Returns:
        list: List of unique hostnames
    """
    hostnames = set()
    for pool in pool_data.get('pplns', []):
        host_info = pool.get('host', {})
        hostname = host_info.get('hostname')
        if hostname and hostname != 'Unknown':
            hostnames.add(hostname)
    return list(hostnames)

def quick_check(tolerance=2, verbose=False, test_connectivity=False):
    """Quick check of pool sync status and optionally test connectivity."""
    try:
        # Get mainnet block height
        if verbose:
            print("🔍 Fetching mainnet block height...")
        main_block_height = get_main_network_block_height()
        
        # Get pool data
        if verbose:
            print("🏊 Fetching mining pool data...")
        response = requests.get("https://backend.sha3x.supportxtm.com/api//pool/ports")
        response.raise_for_status()
        pool_data = response.json()
        
        print(f"Mainnet: {main_block_height:,}")
        if tolerance != 2:
            print(f"Tolerance: ±{tolerance} blocks")
        print("=" * 60)
        
        out_of_sync = 0
        connectivity_issues = 0
        
        # Test connectivity if requested
        connectivity_results = {}
        if test_connectivity:
            if verbose:
                print("🔌 Testing pool connectivity...")
            hostnames = extract_pool_hostnames(pool_data)
            for hostname in hostnames:
                connectivity_results[hostname] = test_pool_connectivity(hostname)
                if not connectivity_results[hostname]['tcp_connect']:
                    connectivity_issues += 1
        
        for pool in pool_data.get('pplns', []):
            host_info = pool.get('host', {})
            pool_height = host_info.get('blockID', 0)
            hostname = host_info.get('hostname', 'Unknown')
            short_hostname = hostname.replace('.sha3x.supportxtm.com', '')
            miners = pool.get('miners', 0)
            
            diff = main_block_height - pool_height
            sync_status = '✅' if abs(diff) <= tolerance else '❌'
            
            if abs(diff) > tolerance:
                out_of_sync += 1
            
            # Prepare output line
            if test_connectivity and hostname in connectivity_results:
                conn_result = connectivity_results[hostname]
                if conn_result['tcp_connect']:
                    conn_status = '🟢' if conn_result['jsonrpc_response'] else '🟡'
                    latency = f"{conn_result['latency_ms']}ms" if conn_result['latency_ms'] else "N/A"
                else:
                    conn_status = '🔴'
                    latency = "FAIL"
                
                if verbose:
                    print(f"{sync_status}{conn_status} {short_hostname}: {pool_height:,} ({diff:+d}) | {latency} | Miners: {miners}")
                else:
                    print(f"{sync_status}{conn_status} {short_hostname}: {pool_height:,} ({diff:+d}) | {latency}")
            else:
                if verbose:
                    print(f"{sync_status} {short_hostname}: {pool_height:,} ({diff:+d}) | Miners: {miners}")
                else:
                    print(f"{sync_status} {short_hostname}: {pool_height:,} ({diff:+d})")
        
        print("=" * 60)
        
        # Summary
        if test_connectivity:
            if connectivity_issues > 0:
                print(f"⚠️  {connectivity_issues} pool(s) have connectivity issues.")
        
        if out_of_sync == 0:
            if test_connectivity and connectivity_issues == 0:
                print("✅ All pools in sync and connectable!")
            elif test_connectivity:
                print("✅ All pools in sync (but some connectivity issues)")
                return 1
            else:
                print("✅ All pools in sync!")
            return 0
        else:
            if test_connectivity and connectivity_issues > 0:
                print(f"❌ {out_of_sync} pool(s) out of sync, {connectivity_issues} connectivity issues.")
            else:
                print(f"❌ {out_of_sync} pool(s) out of sync!")
            return 1
            
    except requests.RequestException as e:
        print(f"❌ Network Error: {e}")
        return 2
    except Exception as e:
        print(f"❌ Error: {e}")
        return 2

def main():
    """Main function with command line argument parsing."""
    parser = argparse.ArgumentParser(
        description="Check Tari mining pool block height sync status",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python quick_pool_check.py              # Default check (±2 blocks tolerance)
  python quick_pool_check.py -t 3         # Custom tolerance (±3 blocks)
  python quick_pool_check.py --verbose    # Show additional details
  python quick_pool_check.py -c           # Test connectivity to pools
  python quick_pool_check.py -t 1 -c -v   # Full check: strict tolerance + connectivity + verbose

Exit Codes:
  0 - All pools are in sync (and connectable if tested)
  1 - Some pools are out of sync or have connectivity issues
  2 - Error occurred (network, API, etc.)
        """
    )
    
    parser.add_argument(
        '-t', '--tolerance', 
        type=int, 
        default=2, 
        help='Block height tolerance (default: 2)'
    )
    
    parser.add_argument(
        '-v', '--verbose', 
        action='store_true', 
        help='Show verbose output including miner counts'
    )
    
    parser.add_argument(
        '-c', '--connectivity', 
        action='store_true', 
        help='Test TCP connectivity and JSON-RPC communication to pools'
    )
    
    args = parser.parse_args()
    
    # Validate tolerance
    if args.tolerance < 0:
        print("❌ Error: Tolerance must be non-negative")
        sys.exit(2)
    
    # Run the check
    exit_code = quick_check(tolerance=args.tolerance, verbose=args.verbose, test_connectivity=args.connectivity)
    sys.exit(exit_code)

if __name__ == "__main__":
    main()

Tari Mining Pool Block Height Sync Checker

A simple, self-contained Python script to monitor Tari mining pool synchronization status.

Features

• ✅ Self-contained - No external project dependencies
• ✅ Configurable tolerance - Set custom block height difference thresholds
• ✅ Verbose mode - Show detailed information including miner counts
• ✅ Professional CLI - Complete argument parsing and help system
• ✅ Proper exit codes - Integration-friendly for monitoring systems
• ✅ Real-time data - Fetches live data from Tari network and pools

Requirements

• Python 3.6+
• requests library (pip install requests)

Usage

# Basic check (±2 blocks tolerance)
python tari_pool_checker.py

# Custom tolerance
python tari_pool_checker.py -t 3

# Verbose output with miner counts
python tari_pool_checker.py --verbose

# Strict monitoring with detailed output
python tari_pool_checker.py -t 1 -v

# Show help
python tari_pool_checker.py --help