Solana: find oldest block with full tx history and logs

solana-find-oldest.py

#!/usr/bin/env python3

import sys
import requests
import json

def json_rpc_request(rpc_url, method, params=None, request_id=1):
    """
    Helper function to send a JSON-RPC request to the Solana node.
    Returns the parsed JSON response.

    :param rpc_url: The endpoint of the Solana RPC node.
    :param method: The RPC method (e.g. 'getBlock', 'getSlot', ...).
    :param params: Optional list of parameters for the method.
    :param request_id: Numeric or string ID for the request.
    :return: The decoded JSON object from the HTTP response.
    """
    if params is None:
        params = []

    headers = {"Content-Type": "application/json"}
    payload = {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": method,
        "params": params
    }

    response = requests.post(rpc_url, headers=headers, data=json.dumps(payload))
    response.raise_for_status()
    return response.json()

def can_get_full_details(rpc_url, slot, debug=False):
    """
    Check if a slot has full transaction details on the node and
    verifies that log (and CPI) storage is enabled by attempting to
    fetch logs for a single transaction in the block.

    Returns True if the node returns a valid block with full transaction details
    and logs are available for at least one transaction.
    Otherwise returns False.
    """
    # Get the block with full transaction details.
    resp = json_rpc_request(
        rpc_url,
        "getBlock",
        [slot, {"transactionDetails": "full", "rewards": False, "encoding": "base64", "maxSupportedTransactionVersion": 0}]
    )

    if debug:
        print(f"Debug: getBlock({slot}) response:")
        print(json.dumps(resp, indent=2))

    # If there's an error or no result, return False.
    if "error" in resp:
        if debug:
            print(f"Debug: Slot {slot} returned error: {resp['error']}")
        return False

    if resp.get("result") is None:
        return False

    block = resp["result"]

    # Ensure there is at least one transaction in the block.
    transactions = block.get("transactions", [])
    if not transactions:
        if debug:
            print(f"Debug: Slot {slot} has no transactions to verify logs.")
        return False

    # Check logs in the first transaction (this is our sample).
    first_tx = transactions[0]
    meta = first_tx.get("meta", {})
    if meta.get("logMessages"):
        if debug:
            print(f"Debug: Logs found in block {slot} via getBlock response.")
        return True
    else:
        # If logs are missing in the block response, try fetching the transaction separately.
        tx_sig = first_tx["transaction"]["signatures"][0]
        tx_resp = json_rpc_request(
            rpc_url,
            "getTransaction",
            [tx_sig, {"encoding": "json", "maxSupportedTransactionVersion": 0}]
        )
        if debug:
            print(f"Debug: getTransaction({tx_sig}) response:")
            print(json.dumps(tx_resp, indent=2))

        if "error" in tx_resp or tx_resp.get("result") is None:
            if debug:
                print(f"Debug: getTransaction for signature {tx_sig} failed or returned no result.")
            return False

        tx_result = tx_resp["result"]
        tx_meta = tx_result.get("meta", {})
        if tx_meta.get("logMessages"):
            if debug:
                print(f"Debug: Logs found in transaction {tx_sig} via getTransaction call.")
            return True
        else:
            if debug:
                print(f"Debug: No logs found for transaction {tx_sig}.")
            return False

def main():
    if len(sys.argv) < 2:
        print("Usage: python find_oldest_block.py <rpc_url> [debug]")
        sys.exit(1)

    rpc_url = sys.argv[1]
    debug = False
    if len(sys.argv) >= 3 and sys.argv[2].lower() == "debug":
        debug = True

    # 1) Get the current slot (latest).
    current_slot_resp = json_rpc_request(rpc_url, "getSlot")
    if debug:
        print("Debug: getSlot response:")
        print(json.dumps(current_slot_resp, indent=2))

    if "result" not in current_slot_resp:
        print("Error: Could not fetch current slot")
        sys.exit(1)

    current_slot = current_slot_resp["result"]
    print(f"Current slot: {current_slot}")

    # 2) Get the first available block.
    first_block_resp = json_rpc_request(rpc_url, "getFirstAvailableBlock")
    if debug:
        print("\nDebug: getFirstAvailableBlock response:")
        print(json.dumps(first_block_resp, indent=2))

    if "result" not in first_block_resp:
        print("Error: Could not fetch first available block")
        sys.exit(1)

    first_block = first_block_resp["result"]
    print(f"First available block: {first_block}")

    # 3) Binary search between first_block and current_slot to find the earliest slot
    #    that has full transaction details and available logs.
    low = first_block
    high = current_slot
    oldest_slot_with_full_details = None

    while low <= high:
        mid = (low + high) // 2

        if can_get_full_details(rpc_url, mid, debug=debug):
            oldest_slot_with_full_details = mid
            high = mid - 1
        else:
            low = mid + 1

    if oldest_slot_with_full_details is not None:
        print(f"\nOldest slot with full transaction details and logs: {oldest_slot_with_full_details}")
    else:
        print("\nNo slot found with full transaction details and logs up to the current slot.")

if __name__ == "__main__":
    main()