1 of 100

API Docs

Welcome

What is Helius?

Solana's Leading RPC & API Platform

Helius is an enterprise-grade Solana development platform offering best-in-class RPCs, APIs, data streaming tools, and 24/7 customer support. Helius also runs the top Solana validator and offers both individual stakers and institutional staking partners 0% fees and 100% MEV rewards.

To read data from and write data to Solana, applications require an RPC node. Our globally distributed fleet of RPCs are powered by top-of-the-line hardware, offer 99.99% uptime SLAs, and are trusted by the largest wallets and applications on Solana.

We offer two types of RPC nodes:

Our battle-tested nodes are globally distributed across New York, Ashburn, Chicago, Vancouver, Los Angeles, Singapore, Tokyo, Frankfurt, and the UK. For Shared RPC nodes, we automatically send your requests to the closest node, guaranteeing the quickest and most efficient routing.

Staked Connections

Staked connections can improve transaction landing rates. As we run the largest validator on Solana, you can send transactions through our staked connections.

We offer two types of staked connections:

Data Services📈

Get the Solana data you need, how you need it. Whether you're streaming real-time updates or automating on-chain event handling, we've got you covered with tools built for speed, scale, and simplicity.

Data Streaming

We offer a variety of methods for streaming Solana data to your databases.

Geyser

Enhanced Websockets

Standard Websockets

Event Notifications

Webhooks

Helius offers various APIs for interacting with NFTs, compressed NFTs, and tokens, calculating priority fees, and parsing transactions.

Priority Fee API

Enhanced Transaction APIs

Solana DAS API and Compression

Photon API and ZK Compression

Pricing & Rate limits

Flexible pricing for all your needs. Switch plans or cancel at anytime. Start in under 5 seconds.

Access a shared RPC node fleet across New York, Ashburn, Chicago, Vancouver, Los Angeles, Singapore, Tokyo, Frankfurt, and the UK, automatically routing requests to the nearest node.

Features:

RPC calls
Webhooks
Standard WebSockets
Enhanced WebSockets (beta) — available for Business and Professional plans only
DAS APIs
Enhanced Transactions API
Priority Fee API
Token Balance API
Photon Methods (ZK Compression)
Mint API (deprecated)

Each plan has a set amount of credits and rate limits for different calls.

* See rate limit exceptions below.

Rate Limit Exceptions *

Photon Limits

Credits

A clear breakdown of the credit costs in our Standard plans.

Standard RPC Calls: 1 credit per call
- Exceptions:
  - Archival Calls: 10 credits each
  - Send Transaction:
  - Priority Fee API: 1 credit
DAS Calls: 10 credits each
Photon Calls:
- Standard Calls: 10 credits
Webhook:
- Webhook Pushes: 1 credit
Enhanced Transactions API and Mint API: 100 credits per call

Extra Credits

If you exhaust your credits within a billing cycle, you can purchase additional credits:

Crypto Plans — purchase additional prepaid credits directly from the dashboard
Card Payment Plans — enable autoscaling in the Project Settings on the dashboard

Enterprise plans are ideal for large teams that need personalized rate limits and more credits than the professional plan offers. Enterprise customers receive discounted pricing for high credit volumes and longer commitments. You can choose your support level with service level agreements (SLAs), uptime guarantees, usage alerts, and direct access to our engineering team to help you optimize your setup.

Legacy (Deprecated) Plan Limits

Prepaid Credits and Autoscaling

Running low on credits mid-cycle? Helius makes it easy to keep your projects running smoothly by allowing you to purchase additional prepaid credits or automatically scale your usage.

How to Get Additional Credits

If your credits run out during your billing cycle, you can get more credits by either purchasing prepaid credits if you paid via crypto, or enabling autoscaling if you paid by card.

1. Prepaid Credits (Crypto Plans)

If you're on a Crypto Plan, you can quickly purchase additional prepaid credits directly from your Helius Dashboard:

Select Buy More Credits.
Complete the purchase with your preferred crypto payment method.

Prepaid credits activate immediately and never expire—you keep them until they run out.

2. Autoscaling (Card Payment Plans)

Autoscaling automatically adds credits as you use them, ensuring uninterrupted service:

Navigate to Project Settings.
Enable the Autoscaling toggle.

Once enabled, Helius will automatically allocate more credits as your usage increases.

Note: Autoscaling is capped for Developer and Business plans to help manage costs and avoid unexpected charges. If you require unlimited autoscaling, consider upgrading to the Professional plan.

Pricing for Additional Credits

Additional credit pricing depends on your current plan:

If you're consistently going over your limit, consider upgrading your plan for better rates and unlimited autoscaling.

Notifications and Tracking Credit Usage

Helius keeps you informed about your credit usage to help manage your spend:

Email Alerts: You'll receive notifications via email when you're approaching your credit limits and when autoscaling occurs.

Autoscaling Limits and Enterprise Options

Developer and Business plans: Autoscaling is available but has predefined caps to avoid unexpectedly large bills.
Professional plans: Autoscaling is unlimited—ideal for rapidly scaling projects.
Enterprise plans: For credit needs exceeding 1B/month, set up an Enterprise plan tailored to your requirements.

Helius is committed to helping you manage your resources efficiently—ensuring your projects have the flexibility they need without unexpected downtime.

Pay with Crypto

Purchasing a Helius Plan with Crypto

Getting Started with Crypto Payments 🪙

Helius allows you to buy plans using USDC. You can secure your subscription in just one straightforward transaction—no hassle. Below, you’ll find an easy-to-follow guide on how to get started.

Purchasing a Plan with Crypto 💲

Initiate Crypto Payment Click Pay with Crypto and connect your preferred wallet.
Enter Billing Information Provide your name and billing address.
Confirm Payment Click Pay with USDC and wait for your transaction to process. Once it’s complete, your new plan will be active!

Important Notes:

Prepare Your Wallet for Auto-Renewal: Your wallet must have sufficient funds for upcoming billing periods.
Automatic Withdrawals: Funds will be automatically deducted from your wallet at each billing cycle.
Manage Permissions: You can revoke permissions or change your wallet anytime.
Cancel Anytime: You can cancel your subscription if needed.

Frequently Asked Questions

I want to switch between fiat and crypto payments.

I paid for a plan, but it hasn't been updated yet in my dashboard.

How do I cancel my subscription?

What does “Authorize Automatic Withdrawals” mean?

When you authorize automatic withdrawals, you’re giving the platform permission to pull the necessary amount of USDC (on Solana) directly from your wallet whenever a new billing period starts. This is often referred to as a “subscription” or “recurring payment,” where you don’t have to manually approve each individual charge.

How It Works:
- At the beginning of each billing cycle, the system checks if you have enough USDC in your Solana wallet.
- If sufficient funds are available, the required amount is automatically deducted to cover the plan’s cost.
- If you ever want to stop automatic charges, you can revoke this authorization via your wallet or billing settings.
Why Authorize?
- It saves you from having to manually make a payment every month (or billing period).
- It ensures your service continues without interruption, as the payment is handled automatically.
Things to Keep in Mind:
- Sufficient Funds: Make sure your wallet has enough USDC ahead of each billing date.
- Revoking Authorization: You can usually disable or revoke this automatic withdrawal at any time if you decide you’d rather pay manually.

How can I manually pay for my subscription instead of authorizing payments?

Manual payments are not supported. You must authorize USDC so it can be automatically deducted on billing days. However, you can revoke this authorization after each purchase by following these steps:

Select your plan (e.g., Developer).
Click Edit in the Authorization section.
Set the limit to 0 to revoke authorization.

Solana RPC Nodes

Helius RPCs Overview

Solana's most loved RPC nodes.

What is Solana RPC?

To interact with blockchain data, you must use an RPC node. Here is a brief explainer on how RPCs on Solana work:

What makes Helius RPCs Different?

At Helius, we are Solana natives. We've been here since the beginning, and unlike other providers, we focus exclusively on Solana and are dedicated to perfecting the Solana experience.

We take pride in offering the most well-rounded RPC solution in the ecosystem, ranking in the top 1% for performance, pricing, and consistent uptime. With Helius, there are no surprises—just reliability you can count on.

Should you face any challenges, our best-in-class 24/7 support team is always here to help, ensuring you stay unblocked and moving forward.

How to use Helius RPCs?

Mainnet – https://mainnet.helius-rpc.com
Devnet – https://devnet.helius-rpc.com

Shared RPC Node Locations

Helius provides services through a globally distributed set of nodes strategically positioned in key locations: New York, Ashburn, Chicago, Vancouver, Los Angeles, Singapore, Tokyo, Frankfurt, and the UK. Helius automatically directs user requests to the closest node, guaranteeing the quickest and most efficient routing.

Dedicated Nodes

Private Dedicated Nodes optimized for low latency and high performance

Get a Dedicated Node in a specific region with no rate or credit limits.

Perfect for latency-sensitive applications such as data streaming or handling a high volume of requests, it’s ideal for teams seeking a dedicated environment with consistent performance and cost management. Choose a single node in a specific location to enhance server-to-server interactions or establish a node cluster for wide geographic distribution. Enhance capabilities using advanced Geyser plugins.

Node Options:

AMD EPYC 7443p
AMD EPYC 7543p (Best for resource-intensive RPC calls)
AMD EPYC 9254 (Best for resource-intensive RPC calls)

Access Includes:

Standard RPC and WebSocket methods
Premium support
Node clients — Agave (recommended) or Jito Labs
Geyser Plugin (Optional) — Yellowstone

For trading teams and Telegram bots, we typically recommend purchasing Dedicated Nodes (lower latency) with the Yellowstone Geyser (gRPC) Plugin for streaming and Dedicated Staked Connections (optimized landing rates for sendTransaction).

Dedicated nodes cannot access staked connections, which greatly improves landing rates. You'll need access to a standard paid plan to use staked connections. Learn more here: Sending Transactions on Solana.

If your team needs a cluster of multiple Dedicated Nodes or needs to run a fleet of Pythnet nodes, contact our sales team. We'll review your requirements and help you with a custom solution.

How to Order a Dedicated Node

Dedicated Nodes can be ordered directly from the developer portal on the Dedicated Nodes tab.

Node Type

The type of node you choose depends on your specific project requirements. Reviewing each node's CPU cores and RAM is important to determine which will best support your workload. Since we don't impose any rate limits, your node's performance will rely entirely on its specifications.

If you plan to make resource-intensive calls (e.g., getProgramAccounts, getTokenLargestAccounts, getTokenAccountsByDelegate, getTokenAccountsByOwner, getLargestAccounts), the AMD EPYC 7543p and AMD EPYC 9254 offer similar performance, but the EPYC 9254 is newer and more efficient.

The AMD EPYC 7443p is better if your workload does not require resource-intensive calls.

Node Location

We offer nodes across multiple regions: in North America (Pittsburgh, Newark, Salt Lake City, Los Angeles, Vancouver); in Europe (Dublin, London, Amsterdam, Frankfurt); and in Asia (Tokyo, Singapore) to ensure the best geographic coverage with the global Solana infrastructure.

For optimal latency, choose a node closest to your server. Your node will be deployed within three hours of payment.

Node Client

You can customize your node by selecting the client type — either Agave or Jito Labs (fork of Agave with an additional method simulateBundle)

Dedicated nodes cannot send Jito Bundles on their own. To send Jito Bundles, you must use the Jito API, which handles packaging and sending the bundles through Jito’s system.

To simplify this process, our SDK provides an easy method called Send Jito Bundle.

Geyser Plugin (Optional)

Add optional features like the Yellowstone Geyser Plugin, which supports streaming of slots, blocks, transactions, and account updates.

We recommend running Geyser and RPC operations on separate nodes for optimal performance.

Payment Options

You can pay via fiat or crypto (USDC). Once your payment goes through, your node will be deployed within 3 hours.

For billing, fiat payments will receive a discount on the next month's bill for the number of days it took to provision the node. For crypto payments, the billing cycle starts once the node is delivered.

Demo

Getting Started with Dedicated Nodes

Dedicated Nodes with Increased Bandwidth and Reduced Latency

Dashboard Overview

Once you have placed an order for a Dedicated Node you will get access to the overview page which will be your Dedicated Node Hub. This will have all your pending orders and active Nodes. It can be reached by going to the Dedicated Nodes tab on the main dashboard or by clicking here. You will get an email once your node is ready and the status will change to Succeeded.

Order ID and Host ID

Order ID and Host ID are identifiers that can be used when contacting the support team with questions regarding your order (Order ID) or your node (Host ID). Think of them as names for your Node.

URL

The URL is your standard Solana RPC URL which can be used to call RPC methods or used as a websocket by changing https to wss.

gRPC URL & gRPC Token

Your gRPC URL and gRPC Token are used when working with the Geyser Plugin that you chose. There will be code examples further that will explain how to use the gRPC and stream Solana events with minimal latency. All our dedicated nodes are Shredstream enhanced.

Region

This is where your node is located.

Solana version

This is the current version that is running on your node.

Geyser type

This is the Geyser plugin you chose.

Status

Status represents the current state of your node:

Pending — signifies a pending order for your node
Underway — your node is being set up
Succeeded — your node is fully set up
Failed — your node is down

Node Deactivated

This indicates if your node is currently suspended. If a payment is missed, your node will be deactivated automatically. You have a 3-day grace period to reactivate it by completing the payment.

Action Buttons

There are also 4 action buttons on the right that give you full control over your node's lifecycle and performance monitoring.

⬆️ Update Version

Lets you update or downgrade your node's version
Use this when you want to update to a newer version
Important to keep your node up-to-date with latest features and performance updates

📊 Metrics

Shows your node's performance
Track things like your response times, requests per second, success and failure rates
Monitor health and usage with request success rates and error rates

🗑️ Delete Node

Permanently and immediately removes your node and cancels the subscription
This cannot be undone and you will have to order a new Node

Working With Your Node

RPC and Websocket

Each dedicated eode comes with an RPC URL that you can use like in the example below.

Here we are using Solana web3.js to call getSlot using our dedicated node:

// Using @solana/web3.js
const { Connection } = require('@solana/web3.js');

const connection = new Connection('https://liveried-grazings-gxzjabdgqa-dedicated.helius-rpc.com?api-key=465964ff-f718-47d2-a51c-66ddcce332d7');

// Get the current slot
const getSlot = async () => {
    const slot = await connection.getSlot();
    console.log('Current slot:', slot);
};

getSlot();

This is how you would set up a native websocket connection to stream new slots:

const Websocket = require('ws');
const ws = new Websocket('wss://liveried-grazings-gxzjabdgqa-dedicated.helius-rpc.com?api-key=465964ff-f718-47d2-a51c-66ddcce332d7	');

ws.onopen = () => {
    ws.send(JSON.stringify({
        jsonrpc: '2.0',
        id: 1,
        method: 'slotSubscribe'
    }));
};

ws.onmessage = (event) => {
    console.log(JSON.parse(event.data));
};

Set up your Geyser Plugin

To begin using the Geyser plugin you need to clone the Yellowstone repo:

git clone https://github.com/helius-labs/yellowstone-grpc.git

Using the CLI

cd yellowstone-grpc/examples/rust/
cargo run --bin client -- -e "https://liveried-grazings-gxzjabdgqa-dedicated-lb.helius-rpc.com:2053" --x-token 42f03938-1daa-4162-a457-bb551ecaf590 subscribe --slots

Once complete, you should see the terminal output new slots. Don't forget to replace the URL and Token with your own.

TypeScript Example

This example streams all Raydium transactions live in JSON format:

import Client, { CommitmentLevel, SubscribeRequest } from "@triton-one/yellowstone-grpc";
import * as bs58 from 'bs58';

const processBuffers = (obj: any): any =>
  !obj ? obj :
  Buffer.isBuffer(obj) || obj instanceof Uint8Array ? bs58.encode(obj) :
  Array.isArray(obj) ? obj.map(processBuffers) :
  typeof obj === 'object' ? Object.fromEntries(Object.entries(obj).map(([k, v]) => [k, processBuffers(v)])) :
  obj;

const main = async () => {
  const client = new Client("grpc_url", 
    "x_token", 
    { "grpc.max_receive_message_length": 64 * 1024 * 1024 });

  const stream = await client.subscribe();
  const write = (req: SubscribeRequest) => new Promise<void>((resolve, reject) => 
    stream.write(req, (err) => err ? reject(err) : resolve()));

  stream.on("data", (data) => {
    try { console.log(JSON.stringify(processBuffers(data), null, 2)); }
    catch (e) { console.error('Error:', e); }
  });

  await write({
    slots: {},
    accounts: {},
    accountsDataSlice: [],
    transactions: {
      allRaydiumTxs: {
        accountInclude: ["675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"],
        accountExclude: [],
        accountRequired: [],
      }
    },
    blocks: {},
    blocksMeta: {},
    entry: {},
    commitment: CommitmentLevel.PROCESSED,
  });

  setInterval(() => write({
    ping: { id: 1 },
    accounts: {},
    accountsDataSlice: [],
    transactions: {},
    blocks: {},
    blocksMeta: {},
    entry: {},
    slots: {},
  }).catch(console.error), 30000);

  await new Promise<void>((resolve, reject) => {
    stream.on("error", (e) => { console.error("Stream error:", e); reject(e); stream.end(); });
    stream.on("end", resolve);
    stream.on("close", resolve);
  });
};

main().catch(console.error);

Best Practices

Managing Node Load

While nodes don't have strict usage limits, they are bound by hardware constraints. A balanced workload helps maintain optimal performance and node liveliness.

Avoid mixing heavy account-based calls (e.g. getProgramAccounts, getTokenAccountsByOwner, etc.) with gRPC usage on the same node.
- It's best to move heavy account-based calls to a shared plan since we have a custom indexer that makes those calls much faster.
Monitor your node's metrics to understand its performance based on response times, successful request %, and error counts.
Consider distributing intensive workloads across multiple nodes in a dedicated fleet

If your team needs a cluster of multiple dedicated nodes, contact our sales team. We will take in your requirements and configure a dedicated fleet for your team.

System Recovery

Our nodes include an automatic recovery mechanism:

If a node falls too far behind in processing slots, it will automatically initiate a restart
During this recovery period, all requests are seamlessly routed to our shared backup pool
Once your node catches up and becomes healthy, traffic will resume to it

Questions and Support

Contact us through your Telegram group (available for dedicated node customers)
You can also get support from the chat system on our website or in our Discord server

RPC

HTTP

Solana nodes accept HTTP requests using the JSON-RPC 2.0 specification.

Info

For JavaScript applications, use the @solana/web3.js library as a convenient interface for the RPC methods to interact with a Solana node.
For a PubSub connection to a Solana node, use the WebSocket API.

Request Formatting

To make a JSON-RPC request, send an HTTP POST request with a Content-Type: application/json header. The JSON request data should contain the following fields:

Field

Type

Description

jsonrpc

string

Set to "2.0".

string | number

A unique identifier for the request, generated by the client. Can be a string, number, or null.

method

string

The name of the method to invoke.

params

array

A JSON array of ordered parameter values.

Example Request using cURL

curl https://mainnet.helius-rpc.com/?api-key=<api-key> -s -X POST -H "Content-Type: application/json" -d '
  {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "getBalance",
    "params": [
      "83astBRguLMdt2h5U1Tpdq5tjFoJ6noeGwaY3mDLVcri"
    ]
  }
'

Response Format

The response will be a JSON object with the following fields:

Field

Type

Description

jsonrpc

string

Matches the request specification.

number

Matches the request identifier.

result

array|number|object|string

Requested data or success confirmation.

Batch Requests

Requests can be sent in batches by sending an array of JSON-RPC request objects in a single POST request.

Example Request with Commitment

The commitment parameter should be included as the last element in the params array:

curl https://mainnet.helius-rpc.com/?api-key=<api-key> -s -X POST -H "Content-Type: application/json" -d '
  {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "getBalance",
    "params": [
      "83astBRguLMdt2h5U1Tpdq5tjFoJ6noeGwaY3mDLVcri",
      {
        "commitment": "finalized"
      }
    ]
  }
'

Definitions

Hash: A SHA-256 hash of a chunk of data.
Pubkey: The public key of an Ed25519 key-pair.
Transaction: A list of Solana instructions signed by a client keypair to authorize those actions.
Signature: An Ed25519 signature of a transaction's payload data, including instructions. This can be used to identify transactions.

Health Check

Although not part of the JSON-RPC API, a GET /health request at the RPC HTTP endpoint provides a health-check mechanism for load balancers or other network infrastructure.

Response Codes

Status

Description

The node is within HEALTH_CHECK_SLOT_DISTANCE slots of the latest cluster confirmed slot.

behind { distance }

The node is behind by distance slots from the latest cluster confirmed slot.

unknown

The node is unable to determine its status relative to the cluster.

Example health check response body:

ok
behind { distance }
unknown

getAccountInfo

Returns all information associated with the account of provided Pubkey

getBalance

Returns the lamport balance of the account of provided Pubkey

getBlock

Returns identity and transaction information about a confirmed block in the ledger

getBlockCommitment

Returns commitment for particular block

getBlockProduction

Returns recent block production information from the current or previous epoch.

getBlockTime

Returns the estimated production time of a block.

getBlocks

Returns a list of confirmed blocks between two slots

getEpochInfo

Returns information about the current epoch

Data Streaming

Solana APIs

Getting Started with Dedicated Nodes

Dedicated Nodes with Increased Bandwidth and Reduced Latency

Dashboard Overview

Order ID and Host ID

Order ID and Host ID are identifiers that can be used when contacting the support team with questions regarding your order (Order ID) or your node (Host ID). Think of them as names for your Node.

URL

The URL is your standard Solana RPC URL which can be used to call RPC methods or used as a websocket by changing https to wss.

gRPC URL & gRPC Token

Region

This is where your node is located.

Solana version

This is the current version that is running on your node.

Geyser type

This is the Geyser plugin you chose.

Status

Status represents the current state of your node:

Pending — signifies a pending order for your node
Underway — your node is being set up
Succeeded — your node is fully set up
Failed — your node is down

Node Deactivated

This indicates if your node is currently suspended. If a payment is missed, your node will be deactivated automatically. You have a 3-day grace period to reactivate it by completing the payment.

Action Buttons

There are also 4 action buttons on the right that give you full control over your node's lifecycle and performance monitoring.

⬆️ Update Version

Lets you update or downgrade your node's version
Use this when you want to update to a newer version
Important to keep your node up-to-date with latest features and performance updates

📊 Metrics

Shows your node's performance
Track things like your response times, requests per second, success and failure rates
Monitor health and usage with request success rates and error rates

🗑️ Delete Node

Permanently and immediately removes your node and cancels the subscription
This cannot be undone and you will have to order a new Node

Working With Your Node

RPC and Websocket

Each dedicated eode comes with an RPC URL that you can use like in the example below.

Here we are using Solana web3.js to call getSlot using our dedicated node:

// Using @solana/web3.js
const { Connection } = require('@solana/web3.js');

const connection = new Connection('https://liveried-grazings-gxzjabdgqa-dedicated.helius-rpc.com?api-key=465964ff-f718-47d2-a51c-66ddcce332d7');

// Get the current slot
const getSlot = async () => {
    const slot = await connection.getSlot();
    console.log('Current slot:', slot);
};

getSlot();

This is how you would set up a native websocket connection to stream new slots:

const Websocket = require('ws');
const ws = new Websocket('wss://liveried-grazings-gxzjabdgqa-dedicated.helius-rpc.com?api-key=465964ff-f718-47d2-a51c-66ddcce332d7	');

ws.onopen = () => {
    ws.send(JSON.stringify({
        jsonrpc: '2.0',
        id: 1,
        method: 'slotSubscribe'
    }));
};

ws.onmessage = (event) => {
    console.log(JSON.parse(event.data));
};

Set up your Geyser Plugin

To begin using the Geyser plugin you need to clone the Yellowstone repo:

git clone https://github.com/helius-labs/yellowstone-grpc.git

Using the CLI

cd yellowstone-grpc/examples/rust/
cargo run --bin client -- -e "https://liveried-grazings-gxzjabdgqa-dedicated-lb.helius-rpc.com:2053" --x-token 42f03938-1daa-4162-a457-bb551ecaf590 subscribe --slots

Once complete, you should see the terminal output new slots. Don't forget to replace the URL and Token with your own.

TypeScript Example

This example streams all Raydium transactions live in JSON format:

import Client, { CommitmentLevel, SubscribeRequest } from "@triton-one/yellowstone-grpc";
import * as bs58 from 'bs58';

const processBuffers = (obj: any): any =>
  !obj ? obj :
  Buffer.isBuffer(obj) || obj instanceof Uint8Array ? bs58.encode(obj) :
  Array.isArray(obj) ? obj.map(processBuffers) :
  typeof obj === 'object' ? Object.fromEntries(Object.entries(obj).map(([k, v]) => [k, processBuffers(v)])) :
  obj;

const main = async () => {
  const client = new Client("grpc_url", 
    "x_token", 
    { "grpc.max_receive_message_length": 64 * 1024 * 1024 });

  const stream = await client.subscribe();
  const write = (req: SubscribeRequest) => new Promise<void>((resolve, reject) => 
    stream.write(req, (err) => err ? reject(err) : resolve()));

  stream.on("data", (data) => {
    try { console.log(JSON.stringify(processBuffers(data), null, 2)); }
    catch (e) { console.error('Error:', e); }
  });

  await write({
    slots: {},
    accounts: {},
    accountsDataSlice: [],
    transactions: {
      allRaydiumTxs: {
        accountInclude: ["675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"],
        accountExclude: [],
        accountRequired: [],
      }
    },
    blocks: {},
    blocksMeta: {},
    entry: {},
    commitment: CommitmentLevel.PROCESSED,
  });

  setInterval(() => write({
    ping: { id: 1 },
    accounts: {},
    accountsDataSlice: [],
    transactions: {},
    blocks: {},
    blocksMeta: {},
    entry: {},
    slots: {},
  }).catch(console.error), 30000);

  await new Promise<void>((resolve, reject) => {
    stream.on("error", (e) => { console.error("Stream error:", e); reject(e); stream.end(); });
    stream.on("end", resolve);
    stream.on("close", resolve);
  });
};

main().catch(console.error);

Best Practices

Managing Node Load

While nodes don't have strict usage limits, they are bound by hardware constraints. A balanced workload helps maintain optimal performance and node liveliness.

Avoid mixing heavy account-based calls (e.g. getProgramAccounts, getTokenAccountsByOwner, etc.) with gRPC usage on the same node.
- It's best to move heavy account-based calls to a shared plan since we have a custom indexer that makes those calls much faster.
Monitor your node's metrics to understand its performance based on response times, successful request %, and error counts.
Consider distributing intensive workloads across multiple nodes in a dedicated fleet

If your team needs a cluster of multiple dedicated nodes, contact our sales team. We will take in your requirements and configure a dedicated fleet for your team.

System Recovery

Our nodes include an automatic recovery mechanism:

If a node falls too far behind in processing slots, it will automatically initiate a restart
During this recovery period, all requests are seamlessly routed to our shared backup pool
Once your node catches up and becomes healthy, traffic will resume to it

Questions and Support

Contact us through your Telegram group (available for dedicated node customers)
You can also get support from the chat system on our website or in our Discord server

blockSubscribe

Subscribe to receive notification anytime a new block is confirmed or finalized.

blockSubscribe is marked as unstable in the Solana documentation. As a result, Helius does not support this method.

Parameters

`filter` (string | object, required)

Description: Filter criteria for the logs to receive results based on account type.

String Options:

all: Include all transactions in the block.

Object Options:

A JSON object with the following field:

mentionsAccountOrProgram (string): Return only transactions that mention the provided public key (as a base-58 encoded string). If no mentions occur in a given block, no notification will be sent.

`object` (optional)

Description: Configuration object containing the following fields:

commitment (string, optional): Default: finalized. Note: processed is not supported.
encoding (string, optional): Default: json. Specifies the encoding format for each returned transaction. Values:
- json
- jsonParsed
- base58
- base64
Details:
- jsonParsed attempts to use program-specific instruction parsers to return more human-readable and explicit data in the transaction.message.instructions list.
- If jsonParsed is requested but a parser cannot be found, the instruction falls back to regular JSON encoding (accounts, data, and programIdIndex fields).
transactionDetails (string, optional): Default: full. Specifies the level of transaction detail to return. Values:
- full
- accounts
- signatures
- none
Details:
- If accounts is requested, transaction details only include signatures and an annotated list of accounts in each transaction.
- Transaction metadata is limited to: fee, err, pre_balances, post_balances, pre_token_balances, and post_token_balances.
maxSupportedTransactionVersion (number, optional): Specifies the maximum transaction version to return in responses.
Details:
- If the requested block contains a transaction with a higher version, an error will be returned.
- If omitted, only legacy transactions will be returned, and a block containing any versioned transaction will prompt an error.
showRewards (bool, optional): Whether to populate the rewards array. Default behavior includes rewards if this parameter is not provided.

Result

Returns an integer which serves as the subscription ID. This ID is required to unsubscribe.

Code Samples

Request Example 1 (Subscribe to all transactions in a block):

{
  "jsonrpc": "2.0",
  "id": "1",
  "method": "blockSubscribe",
  "params": ["all"]
}

Request Example 2 (Filter by account and include optional configuration):

{
  "jsonrpc": "2.0",
  "id": "1",
  "method": "blockSubscribe",
  "params": [
    {
      "mentionsAccountOrProgram": "LieKvPRE8XeX3Y2xVNHjKlpAScD12lYySBVQ4HqoJ5op"
    },
    {
      "commitment": "confirmed",
      "encoding": "base64",
      "showRewards": true,
      "transactionDetails": "full"
    }
  ]
}

Response Example:

{
  "jsonrpc": "2.0",
  "result": 0,
  "id": 1
}

Notification Format

The notification is an object with the following fields:

slot (u64): The corresponding slot.
err (object | null): Error if an issue occurs while publishing the notification; otherwise null.
block (object | null): A block object as seen in the getBlock RPC HTTP method.

Example Notification

{
  "jsonrpc": "2.0",
  "method": "blockNotification",
  "params": {
    "result": {
      "context": {
        "slot": 112301554
      },
      "value": {
        "slot": 112301554,
        "block": {
          "previousBlockhash": "GJp125YAN4ufCSUvZJVdCyWQJ7RPWMmwxoyUQySydZA",
          "blockhash": "6ojMHjctdqfB55JDpEpqfHnP96fiaHEcvzEQ2NNcxzHP",
          "parentSlot": 112301553,
          "transactions": [
            {
              "transaction": [
                "OpltwoUvWxYi1P2U8vbIdE/aPntjYo5Aa0VQ2JJyeJE2g9Vvxk8dDGgFMruYfDu8/IfUWb0REppTe7IpAuuLRgIBAAkWnj4KHRpEWWW7gvO1c0BHy06wZi2g7/DLqpEtkRsThAXIdBbhXCLvltw50ZnjDx2hzw74NVn49kmpYj2VZHQJoeJoYJqaKcvuxCi/2i4yywedcVNDWkM84Iuw+cEn9/ROCrXY4qBFI9dveEERQ1c4kdU46xjxj9Vi+QXkb2Kx45QFVkG4Y7HHsoS6WNUiw2m4ffnMNnOVdF9tJht7oeuEfDMuUEaO7l9JeUxppCvrGk3CP45saO51gkwVYEgKzhpKjCx3rgsYxNR81fY4hnUQXSbbc2Y55FkwgRBpVvQK7/+clR4Gjhd3L4y+OtPl7QF93Akg1LaU9wRMs5nvfDFlggqI9PqJl+IvVWrNRdBbPS8LIIhcwbRTkSbqlJQWxYg3Bo2CTVbw7rt1ZubuHWWp0mD/UJpLXGm2JprWTePNULzHu67sfqaWF99LwmwjTyYEkqkRt1T0Je5VzHgJs0N5jY4iIU9K3lMqvrKOIn/2zEMZ+ol2gdgjshx+sphIyhw65F3J/Dbzk04LLkK+CULmN571Y+hFlXF2ke0BIuUG6AUF+4214Cu7FXnqo3rkxEHDZAk0lRrAJ8X/Z+iwuwI5cgbd9uHXZaGT2cvhRs7reawctIXtX1s3kTqM9YV+/wCpDLAp8axcEkaQkLDKRoWxqp8XLNZSKial7Rk+ELAVVKWoWLRXRZ+OIggu0OzMExvVLE5VHqy71FNHq4gGitkiKYNFWSLIE4qGfdFLZXy/6hwS+wq9ewjikCpd//C9BcCL7Wl0iQdUslxNVCBZHnCoPYih9JXvGefOb9WWnjGy14sG9j70+RSVx6BlkFELWwFvIlWR/tHn3EhHAuL0inS2pwX7ZQTAU6gDVaoqbR2EiJ47cKoPycBNvHLoKxoY9AZaBjPl6q8SKQJSFyFd9n44opAgI6zMTjYF/8Ok4VpXEESp3QaoUyTI9sOJ6oFP6f4dwnvQelgXS+AEfAsHsKXxGAIUDQENAgMEBQAGBwgIDg8IBJCER3QXl1AVDBADCQoOAAQLERITDAjb7ugh3gOuTy==",
                "base64"
              ],
              "meta": {
                "err": null,
                "status": {
                  "Ok": null
                },
                "fee": 5000
              }
            }
          ],
          "blockTime": 1639926816,
          "blockHeight": 101210751
        },
        "err": null
      }
    },
    "subscription": 14
  }
}

Transaction Monitoring

Track Solana transactions in real-time with program filtering and detailed execution data.

Transaction Monitoring with Yellowstone gRPC

This guide demonstrates how to use Yellowstone gRPC to monitor transactions on Solana. You'll learn how to track transactions in real-time with filtering options and reliable reconnection handling.

TypeScript Implementation

import Client, { CommitmentLevel, SubscribeRequest } from "@triton-one/yellowstone-grpc";
import * as bs58 from 'bs58';

class GrpcStreamManager {
    private client: Client;
    private stream: any;
    private isConnected: boolean = false;
    private reconnectAttempts: number = 0;
    private readonly maxReconnectAttempts: number = 10;
    private readonly reconnectInterval: number = 5000; // 5 seconds
    private readonly dataHandler: (data: any) => void;

    constructor(
        endpoint: string,
        authToken: string,
        dataHandler: (data: any) => void
    ) {
        this.client = new Client(
            endpoint,
            authToken,
            { "grpc.max_receive_message_length": 64 * 1024 * 1024 }
        );
        this.dataHandler = dataHandler;
    }

    async connect(subscribeRequest: SubscribeRequest): Promise<void> {
        try {
            this.stream = await this.client.subscribe();
            this.isConnected = true;
            this.reconnectAttempts = 0;

            this.stream.on("data", this.handleData.bind(this));
            this.stream.on("error", this.handleError.bind(this));
            this.stream.on("end", () => this.handleDisconnect(subscribeRequest));
            this.stream.on("close", () => this.handleDisconnect(subscribeRequest));

            await this.write(subscribeRequest);
            this.startPing();
        } catch (error) {
            console.error("Connection error:", error);
            await this.reconnect(subscribeRequest);
        }
    }

    private async write(req: SubscribeRequest): Promise<void> {
        return new Promise((resolve, reject) => {
            this.stream.write(req, (err: any) => err ? reject(err) : resolve());
        });
    }

    private async reconnect(subscribeRequest: SubscribeRequest): Promise<void> {
        if (this.reconnectAttempts >= this.maxReconnectAttempts) {
            console.error("Max reconnection attempts reached");
            return;
        }

        this.reconnectAttempts++;
        console.log(`Reconnecting... Attempt ${this.reconnectAttempts}`);

        setTimeout(async () => {
            try {
                await this.connect(subscribeRequest);
            } catch (error) {
                console.error("Reconnection failed:", error);
                await this.reconnect(subscribeRequest);
            }
        }, this.reconnectInterval * Math.min(this.reconnectAttempts, 5));
    }

    private startPing(): void {
        setInterval(() => {
            if (this.isConnected) {
                this.write({
                    ping: { id: 1 },
                    accounts: {},
                    accountsDataSlice: [],
                    transactions: {},
                    blocks: {},
                    blocksMeta: {},
                    entry: {},
                    slots: {},
                    transactionsStatus: {},
                }).catch(console.error);
            }
        }, 30000);
    }

    private handleData(data: any): void {
        try {
            const processed = this.processBuffers(data);
            this.dataHandler(processed);
        } catch (error) {
            console.error("Error processing data:", error);
        }
    }

    private handleError(error: any): void {
        console.error("Stream error:", error);
        this.isConnected = false;
    }

    private handleDisconnect(subscribeRequest: SubscribeRequest): void {
        console.log("Stream disconnected");
        this.isConnected = false;
        this.reconnect(subscribeRequest);
    }

    private processBuffers(obj: any): any {
        if (!obj) return obj;
        if (Buffer.isBuffer(obj) || obj instanceof Uint8Array) {
            return bs58.default.encode(obj);
        }
        if (Array.isArray(obj)) {
            return obj.map(item => this.processBuffers(item));
        }
        if (typeof obj === 'object') {
            return Object.fromEntries(
                Object.entries(obj).map(([k, v]) => [k, this.processBuffers(v)])
            );
        }
        return obj;
    }
}

// Transaction monitoring implementation
async function monitorTransactions() {
    const manager = new GrpcStreamManager(
        "your-grpc-url:2053",
        "your-x-token",
        handleTransactionUpdate
    );

    // Create subscription request for monitoring program transactions
    const subscribeRequest: SubscribeRequest = {
        transactions: {
            client: {
                accountInclude: ["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"],
                accountExclude: [],
                accountRequired: [],
                vote: false,
                failed: false
            }
        },
        commitment: CommitmentLevel.CONFIRMED,
        accounts: {},
        accountsDataSlice: [],
        blocks: {},
        blocksMeta: {},
        entry: {},
        slots: {},
        transactionsStatus: {}
    };

    await manager.connect(subscribeRequest);
}

function handleTransactionUpdate(data: any): void {
    // Check if we have transaction data in the correct structure
    if (data?.transaction?.transaction) {
        const txInfo = data.transaction.transaction;
        console.log('\n=== Transaction Details ===');
        console.log(`Signature: ${txInfo.signature}`);
        console.log(`Slot: ${data.transaction.slot}`);
        console.log(`Is Vote: ${txInfo.isVote}`);
        
        if (txInfo.meta) {
            console.log(`Fee: ${txInfo.meta.fee} lamports`);
            console.log(`Compute Units: ${txInfo.meta.computeUnitsConsumed}`);
            console.log(`Status: ${txInfo.meta.err ? 'Failed' : 'Success'}`);

            // Token balance changes
            if (txInfo.meta.preTokenBalances?.length > 0) {
                console.log('\n=== Token Balance Changes ===');
                txInfo.meta.preTokenBalances.forEach((preBalance: any, index: number) => {
                    const postBalance = txInfo.meta.postTokenBalances[index];
                    if (preBalance && postBalance) {
                        console.log(`Token Account: ${preBalance.mint}`);
                        console.log(`  Pre Balance: ${preBalance.uiTokenAmount?.uiAmount}`);
                        console.log(`  Post Balance: ${postBalance.uiTokenAmount?.uiAmount}`);
                        if (preBalance.owner) {
                            console.log(`  Owner: ${preBalance.owner}`);
                        }
                    }
                });
            }

            // Log messages
            if (txInfo.meta.logMessages?.length > 0) {
                console.log('\n=== Program Logs ===');
                txInfo.meta.logMessages.forEach((log: string) => {
                    console.log(`  ${log}`);
                });
            }
        }

        // Transaction instructions
        if (txInfo.transaction?.message) {
            const message = txInfo.transaction.message;
            
            console.log('\n=== Account Keys ===');
            if (Array.isArray(message.accountKeys)) {
                message.accountKeys.forEach((key: Uint8Array, index: number) => {
                    try {
                        console.log(`  ${index}: ${bs58.default.encode(key)}`);
                    } catch (error) {
                        console.log(`  ${index}: [Unable to encode key]`);
                    }
                });
            }

            // Instructions
            // TODO: Add instructions processing logic
            if (Array.isArray(message.instructions)) {
                console.log('\n=== Instructions ===');
                message.instructions.forEach((ix: any, index: number) => {
                    try {
                        console.log(`\nInstruction ${index + 1}:`);
                        console.log(`  Program: ${bs58.default.encode(message.accountKeys[ix.programIdIndex])}`);
                        console.log(`  Accounts: ${ix.accounts.map((idx: number) => 
                            bs58.default.encode(message.accountKeys[idx])
                        )}`);
                        if (ix.data) {
                            console.log(`  Data: ${bs58.default.encode(ix.data)}`);
                        }
                    } catch (error) {
                        console.log(`  [Unable to decode instruction ${index + 1}]`);
                    }
                });
            }

            // Process inner instructions
            // TODO: Add inner instructions processing logic
            if (txInfo.meta?.innerInstructions?.length > 0) {
                console.log('\n=== Inner Instructions ===');
                txInfo.meta.innerInstructions.forEach((inner: any, index: number) => {
                    console.log(`\nInner Instruction Set ${index + 1}:`);
                    inner.instructions.forEach((ix: any, i: number) => {
                        try {
                            console.log(`  Instruction ${i + 1}:`);
                            console.log(`    Program: ${bs58.default.encode(message.accountKeys[ix.programIdIndex])}`);
                            console.log(`    Accounts: ${ix.accounts.map((idx: number) => 
                                bs58.default.encode(message.accountKeys[idx])
                            )}`);
                            if (ix.data) {
                                console.log(`    Data: ${bs58.default.encode(ix.data)}`);
                            }
                        } catch (error) {
                            console.log(`    [Unable to decode instruction]`);
                        }
                    });
                });
            }
        }

        console.log('\n' + '='.repeat(50) + '\n');
    }
}

// Start monitoring
monitorTransactions().catch(console.error);

Transaction Subscription Options

1. Monitor Program Transactions

const subscribeRequest: SubscribeRequest = {
    transactions: {
        accountInclude: ["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"], // Token program
        accountExclude: [],
        accountRequired: [],
        vote: false,
        failed: false,
        signature: undefined
    },
    commitment: CommitmentLevel.CONFIRMED,
};

2. Monitor Multiple Programs

const subscribeRequest: SubscribeRequest = {
    transactions: {
        accountInclude: [
            "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA", // Token program
            "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL", // Associated Token program
        ],
        accountExclude: [],
        accountRequired: [],
        vote: false,
        failed: true,
        signature: undefined
    },
    commitment: CommitmentLevel.CONFIRMED,
};

3. Monitor All Transactions

const subscribeRequest: SubscribeRequest = {
    transactions: {
        accountInclude: [],
        accountExclude: [],
        accountRequired: [],
        vote: true,
        failed: true,
        signature: undefined
    },
    commitment: CommitmentLevel.CONFIRMED,
};

Data Structures

Transaction Update:
- Slot number (when transaction was processed)
- Transaction signature (unique identifier)
- Transaction message
  - Account keys (involved accounts)
  - Instructions (program calls and data)
  - Recent blockhash
- Transaction metadata
  - Status (success/failure)
  - Fee amount
  - Compute units consumed
  - Log messages
  - Inner instructions
  - Balance changes
Instruction Details:
- Program ID (program being called)
- Account list (accounts used by instruction)
- Instruction data (program-specific data)
- Inner instructions (CPI calls)

Rust Implementation

use {
    anyhow::Result,
    futures::{stream::StreamExt, sink::SinkExt},
    log::error,
    std::{collections::HashMap, sync::Arc, time::Duration},
    tokio::sync::Mutex,
    tonic::{metadata::errors::InvalidMetadataValue, transport::Endpoint},
    yellowstone_grpc_client::{GeyserGrpcClient, InterceptorXToken},
    yellowstone_grpc_proto::{
        geyser::{
            geyser_client::GeyserClient,
            SubscribeRequest,
            SubscribeRequestFilterTransactions,
            subscribe_update::UpdateOneof,
            SubscribeUpdateTransaction,
        },
        prelude::{CommitmentLevel, SubscribeRequestPing},
    },
    tonic_health::pb::health_client::HealthClient,
    solana_sdk::pubkey::Pubkey,
    hex,
    tokio::sync::mpsc,
};

/// Manager for handling gRPC stream connections and transaction updates
struct GrpcStreamManager {
    client: GeyserGrpcClient<InterceptorXToken>,
    is_connected: bool,
    reconnect_attempts: u32,
    max_reconnect_attempts: u32,
    reconnect_interval: Duration,
}

impl GrpcStreamManager {
    /// Handles transaction update messages from the gRPC stream
    /// This function can be customized based on your requirements:
    /// - Store transactions in a database
    /// - Trigger specific actions based on transaction contents
    /// - Filter for specific types of transactions
    /// - Transform data into your required format
    /// 
    /// # Arguments
    /// * `transaction_update` - The transaction update containing all details
    fn handle_transaction_update(&self, transaction_update: &SubscribeUpdateTransaction) {
        if let Some(transaction) = &transaction_update.transaction {
            println!("Transaction Update:");
            println!("  Signature: {}", hex::encode(&transaction.signature));
            
            if let Some(meta) = &transaction.meta {
                println!("  Status: {}", if meta.err.is_none() { "Success" } else { "Failed" });
                println!("  Fee: {:?}", meta.fee);
                println!("  Compute Units: {:?}", meta.compute_units_consumed);
            }

            if let Some(transaction_message) = &transaction.transaction {
                if let Some(message) = &transaction_message.message {
                    println!("Account Keys:");
                    for (i, key) in message.account_keys.iter().enumerate() {
                        if let Ok(pubkey) = Pubkey::try_from(key.as_slice()) {
                            println!("  {}: {}", i, pubkey);
                        } else {
                            println!("  {}: {}", i, hex::encode(key));
                        }
                    }

                    println!("Instructions:");
                    for (i, instruction) in message.instructions.iter().enumerate() {
                        println!("Instruction {}:", i + 1);
                        if let Some(program_key) = message.account_keys.get(instruction.program_id_index as usize) {
                            if let Ok(pubkey) = Pubkey::try_from(program_key.as_slice()) {
                                println!("  Program: {}", pubkey);
                            } else {
                                println!("  Program: {}", hex::encode(program_key));
                            }
                        }
                        
                        println!("  Accounts:");
                        for &account_idx in &instruction.accounts {
                            if let Some(key) = message.account_keys.get(account_idx as usize) {
                                if let Ok(pubkey) = Pubkey::try_from(key.as_slice()) {
                                    println!("    {}", pubkey);
                                } else {
                                    println!("    {}", hex::encode(key));
                                }
                            }
                        }
                        println!("  Data: {}", hex::encode(&instruction.data));
                    }

                    if let Some(meta) = &transaction.meta {
                        if !meta.log_messages.is_empty() {
                            println!("Log Messages:");
                            for log in &meta.log_messages {
                                println!("  {}", log);
                            }
                        }

                        println!("Balance Changes:");
                        for i in 0..meta.pre_balances.len().min(meta.post_balances.len()) {
                            let pre = meta.pre_balances[i];
                            let post = meta.post_balances[i];
                            if pre != post {
                                if let Some(key) = message.account_keys.get(i) {
                                    if let Ok(pubkey) = Pubkey::try_from(key.as_slice()) {
                                        println!("  Account {}: {} -> {} (Δ{})",
                                            pubkey,
                                            pre, post, post.saturating_sub(pre));
                                    } else {
                                        println!("  Account {}: {} -> {} (Δ{})",
                                            hex::encode(key),
                                            pre, post, post.saturating_sub(pre));
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    /// Creates a new GrpcStreamManager instance
    /// 
    /// # Arguments
    /// * `endpoint` - The gRPC endpoint URL
    /// * `x_token` - Authentication token for the endpoint
    async fn new(endpoint: &str, x_token: &str) -> Result<Arc<Mutex<GrpcStreamManager>>> {
        let interceptor = InterceptorXToken {
            x_token: Some(x_token.parse().map_err(|e: InvalidMetadataValue| anyhow::Error::from(e))?),
            x_request_snapshot: true,
        };

        let channel = Endpoint::from_shared(endpoint.to_string())?
            .connect_timeout(Duration::from_secs(10))
            .timeout(Duration::from_secs(10))
            .connect()
            .await
            .map_err(|e| anyhow::Error::from(e))?;

        let client = GeyserGrpcClient::new(
            HealthClient::with_interceptor(channel.clone(), interceptor.clone()),
            GeyserClient::with_interceptor(channel, interceptor),
        );

        Ok(Arc::new(Mutex::new(GrpcStreamManager {
            client,
            is_connected: false,
            reconnect_attempts: 0,
            max_reconnect_attempts: 10,
            reconnect_interval: Duration::from_secs(5),
        })))
    }

    /// Establishes connection and handles the subscription stream
    /// 
    /// # Arguments
    /// * `request` - The subscription request containing transaction filters and other parameters
    async fn connect(&mut self, request: SubscribeRequest) -> Result<()> {
        let request = request.clone();
        let (mut subscribe_tx, mut stream) = self.client.subscribe_with_request(Some(request.clone())).await?;

        self.is_connected = true;
        self.reconnect_attempts = 0;

        // Create a channel for sending ping requests
        let (ping_sender, mut ping_receiver) = mpsc::channel(10);
        
        // Add client-initiated ping mechanism
        let ping_handle = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(30));
            loop {
                interval.tick().await;
                if let Err(e) = ping_sender.send(()).await {
                    error!("Failed to send ping signal: {:?}", e);
                    break;
                }
            }
        });

        // Process both stream messages and ping requests
        loop {
            tokio::select! {
                Some(message) = stream.next() => {
                    match message {
                        Ok(msg) => {
                            match msg.update_oneof {
                                Some(UpdateOneof::Transaction(transaction)) => {
                                    self.handle_transaction_update(&transaction);
                                }
                                Some(UpdateOneof::Ping(_)) => {
                                    subscribe_tx
                                        .send(SubscribeRequest {
                                            ping: Some(SubscribeRequestPing { id: 1 }),
                                            ..Default::default()
                                        })
                                        .await?;
                                }
                                Some(UpdateOneof::Pong(_)) => {} // Ignore pong responses
                                _ => {
                                    println!("Other update received: {:?}", msg);
                                }
                            }
                        }
                        Err(err) => {
                            error!("Error: {:?}", err);
                            self.is_connected = false;
                            ping_handle.abort(); // Cleanup ping task
                            Box::pin(self.reconnect(request.clone())).await?;
                            return Ok(());
                        }
                    }
                }
                Some(_) = ping_receiver.recv() => {
                    // Send ping when requested by the ping task
                    if let Err(e) = subscribe_tx
                        .send(SubscribeRequest {
                            ping: Some(SubscribeRequestPing { id: 1 }),
                            ..Default::default()
                        })
                        .await
                    {
                        error!("Failed to send ping: {:?}", e);
                        break;
                    }
                }
                else => break,
            }
        }

        ping_handle.abort(); // Cleanup ping task
        Ok(())
    }

    /// Attempts to reconnect when the connection is lost
    /// 
    /// # Arguments
    /// * `request` - The original subscription request to reestablish the connection
    async fn reconnect(&mut self, request: SubscribeRequest) -> Result<()> {
        if self.reconnect_attempts >= self.max_reconnect_attempts {
            println!("Max reconnection attempts reached");
            return Ok(());
        }

        self.reconnect_attempts += 1;
        println!("Reconnecting... Attempt {}", self.reconnect_attempts);

        let backoff = self.reconnect_interval * std::cmp::min(self.reconnect_attempts, 5);
        tokio::time::sleep(backoff).await;

        Box::pin(self.connect(request)).await
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // Initialize gRPC stream manager
    let manager = GrpcStreamManager::new(
        "your-grpc-url:2053",
        "your-x-token",
    ).await?;

    let mut manager_lock = manager.lock().await;

    // Create subscription request for token program transactions
    let request = SubscribeRequest {
        transactions: HashMap::from_iter(vec![(
            "transactions".to_string(),
            SubscribeRequestFilterTransactions {
                vote: Some(false),
                failed: Some(false),
                signature: None,
                account_include: vec!["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA".to_string()],
                account_exclude: vec![],
                account_required: vec![],
            },
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    };

    println!("Starting subscription for Token Program transactions");
    
    // Start the subscription
    let result = manager_lock.connect(request).await;
    if let Err(e) = &result {
        println!("Subscription error: {:?}", e);
    }
    result?;

    Ok(())
}

Go Implementation

package main

import (
	"context"
	"crypto/tls"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/mr-tron/base58"
	pb "github.com/rpcpool/yellowstone-grpc/examples/golang/proto"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)

type GrpcStreamManager struct {
	mu                   sync.Mutex
	conn                 *grpc.ClientConn
	client               pb.GeyserClient
	stream               pb.Geyser_SubscribeClient
	isConnected          bool
	reconnectAttempts    int
	maxReconnectAttempts int
	reconnectInterval    time.Duration
	dataHandler          func(*pb.SubscribeUpdate)
	xToken               string
}

func NewGrpcStreamManager(endpoint string, xToken string, dataHandler func(*pb.SubscribeUpdate)) (*GrpcStreamManager, error) {
	// Create gRPC connection with interceptor for x-token
	ctx := metadata.NewOutgoingContext(
		context.Background(),
		metadata.New(map[string]string{"x-token": xToken}),
	)

	// Configure TLS
	config := &tls.Config{
		InsecureSkipVerify: true,
	}

	conn, err := grpc.DialContext(ctx, endpoint,
		grpc.WithTransportCredentials(credentials.NewTLS(config)),
		grpc.WithInitialWindowSize(1<<30),
		grpc.WithInitialConnWindowSize(1<<30),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(64*1024*1024),
			grpc.MaxCallRecvMsgSize(64*1024*1024),
		),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to connect: %v", err)
	}

	return &GrpcStreamManager{
		conn:                 conn,
		client:               pb.NewGeyserClient(conn),
		isConnected:          false,
		reconnectAttempts:    0,
		maxReconnectAttempts: 10,
		reconnectInterval:    5 * time.Second,
		dataHandler:          dataHandler,
		xToken:               xToken,
	}, nil
}

func (m *GrpcStreamManager) Connect(ctx context.Context, req *pb.SubscribeRequest) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	log.Println("Attempting to connect...")

	// Add x-token to context
	ctx = metadata.NewOutgoingContext(
		ctx,
		metadata.New(map[string]string{"x-token": m.xToken}),
	)

	stream, err := m.client.Subscribe(ctx)
	if err != nil {
		log.Printf("Failed to subscribe: %v", err)
		return m.reconnect(ctx, req)
	}

	if err := stream.Send(req); err != nil {
		log.Printf("Failed to send request: %v", err)
		return m.reconnect(ctx, req)
	}

	m.stream = stream
	m.isConnected = true
	m.reconnectAttempts = 0
	log.Println("Connection established")

	// Start ping goroutine
	go func() {
		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if !m.isConnected {
					return
				}
				pingReq := &pb.SubscribeRequest{
					Ping: &pb.SubscribeRequestPing{Id: 1},
				}
				if err := m.stream.Send(pingReq); err != nil {
					log.Printf("Ping failed: %v", err)
					m.handleDisconnect(ctx, req)
					return
				}
			}
		}
	}()

	// Process updates
	go func() {
		for {
			update, err := m.stream.Recv()
			if err != nil {
				log.Printf("Stream error: %v", err)
				m.handleDisconnect(ctx, req)
				return
			}

			m.dataHandler(update)
		}
	}()

	return nil
}

func (m *GrpcStreamManager) reconnect(ctx context.Context, req *pb.SubscribeRequest) error {
	if m.reconnectAttempts >= m.maxReconnectAttempts {
		return fmt.Errorf("max reconnection attempts reached")
	}

	m.reconnectAttempts++
	log.Printf("Reconnecting... Attempt %d", m.reconnectAttempts)

	backoff := m.reconnectInterval * time.Duration(min(m.reconnectAttempts, 5))
	time.Sleep(backoff)

	return m.Connect(ctx, req)
}

func (m *GrpcStreamManager) handleDisconnect(ctx context.Context, req *pb.SubscribeRequest) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if !m.isConnected {
		return
	}

	m.isConnected = false
	if err := m.reconnect(ctx, req); err != nil {
		log.Printf("Failed to reconnect: %v", err)
	}
}

func (m *GrpcStreamManager) Close() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.isConnected = false
	if m.conn != nil {
		return m.conn.Close()
	}
	return nil
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

func handleAccountUpdate(update *pb.SubscribeUpdate) {
	if tx := update.GetTransaction(); tx != nil {
		log.Printf("Transaction Update:\n")
		log.Printf("  Slot: %d\n", tx.GetSlot())

		txInfo := tx.GetTransaction()
		if txInfo != nil {
			// Print signature if available
			if len(txInfo.GetSignature()) > 0 {
				log.Printf("  Signature: %s\n", base58.Encode(txInfo.GetSignature()))
			}

			// Print if it's a vote transaction
			log.Printf("  Is Vote: %v\n", txInfo.GetIsVote())

			// Print transaction index
			log.Printf("  Transaction Index: %d\n", txInfo.GetIndex())

			// Print transaction details
			if tx := txInfo.GetTransaction(); tx != nil {
				if msg := tx.GetMessage(); msg != nil {
					if accounts := msg.GetAccountKeys(); len(accounts) > 0 {
						log.Printf("  Account Keys:\n")
						for _, acc := range accounts {
							if len(acc) > 0 {
								log.Printf("    - %s\n", base58.Encode(acc))
							}
						}
					}
				}
			}

			// Print status and metadata
			if meta := txInfo.GetMeta(); meta != nil {
				log.Printf("  Status: %v\n", meta.GetErr() == nil)
				log.Printf("  Fee: %d\n", meta.GetFee())

				// Print log messages if any
				if len(meta.GetLogMessages()) > 0 {
					log.Printf("  Log Messages:\n")
					for _, msg := range meta.GetLogMessages() {
						log.Printf("    - %s\n", msg)
					}
				}
			}
		}
	}
}

func boolPtr(b bool) *bool {
	return &b
}

func main() {
	ctx := context.Background()

	// Create manager with data handler and x-token
	manager, err := NewGrpcStreamManager(
		"your-grpc-url:2053",
        	"your-x-token",
		handleAccountUpdate,
	)
	if err != nil {
		log.Fatal(err)
	}
	defer manager.Close()

	// Create subscription request for token program transactions
	transactions := make(map[string]*pb.SubscribeRequestFilterTransactions)
	transactions["transactions"] = &pb.SubscribeRequestFilterTransactions{
		Vote:            boolPtr(false),
		Failed:          boolPtr(false),
		AccountInclude:  []string{"TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"},
		AccountExclude:  []string{},
		AccountRequired: []string{},
	}

	commitment := pb.CommitmentLevel_CONFIRMED
	req := &pb.SubscribeRequest{
		Transactions: transactions,
		Commitment:   &commitment,
	}

	// Connect and handle updates
	if err := manager.Connect(ctx, req); err != nil {
		log.Fatal(err)
	}

	// Keep the main goroutine running
	select {}
}

Entry Monitoring

Stream Solana block entries with transaction batches and execution results

Entry Monitoring with Yellowstone gRPC

This guide demonstrates how to use Yellowstone gRPC to monitor block entries on Solana. Block entries represent the fundamental units of execution on the Solana blockchain, containing transaction batches and their execution results. You'll learn how to track entries with reliable reconnection handling.

TypeScript Implementation

import Client, { CommitmentLevel, SubscribeRequest } from "@triton-one/yellowstone-grpc";
import * as bs58 from 'bs58';

class GrpcStreamManager {
    private client: Client;
    private stream: any;
    private isConnected: boolean = false;
    private reconnectAttempts: number = 0;
    private readonly maxReconnectAttempts: number = 10;
    private readonly reconnectInterval: number = 5000; // 5 seconds
    private readonly dataHandler: (data: any) => void;

    constructor(
        endpoint: string,
        authToken: string,
        dataHandler: (data: any) => void
    ) {
        this.client = new Client(
            endpoint,
            authToken,
            { "grpc.max_receive_message_length": 64 * 1024 * 1024 }
        );
        this.dataHandler = dataHandler;
    }

    async connect(subscribeRequest: SubscribeRequest): Promise<void> {
        try {
            this.stream = await this.client.subscribe();
            this.isConnected = true;
            this.reconnectAttempts = 0;

            this.stream.on("data", this.handleData.bind(this));
            this.stream.on("error", this.handleError.bind(this));
            this.stream.on("end", () => this.handleDisconnect(subscribeRequest));
            this.stream.on("close", () => this.handleDisconnect(subscribeRequest));

            await this.write(subscribeRequest);
            this.startPing();
        } catch (error) {
            console.error("Connection error:", error);
            await this.reconnect(subscribeRequest);
        }
    }

    private async write(req: SubscribeRequest): Promise<void> {
        return new Promise((resolve, reject) => {
            this.stream.write(req, (err: any) => err ? reject(err) : resolve());
        });
    }

    private async reconnect(subscribeRequest: SubscribeRequest): Promise<void> {
        if (this.reconnectAttempts >= this.maxReconnectAttempts) {
            console.error("Max reconnection attempts reached");
            return;
        }

        this.reconnectAttempts++;
        console.log(`Reconnecting... Attempt ${this.reconnectAttempts}`);

        setTimeout(async () => {
            try {
                await this.connect(subscribeRequest);
            } catch (error) {
                console.error("Reconnection failed:", error);
                await this.reconnect(subscribeRequest);
            }
        }, this.reconnectInterval * Math.min(this.reconnectAttempts, 5));
    }

    private startPing(): void {
        setInterval(() => {
            if (this.isConnected) {
                this.write({
                    ping: { id: 1 },
                    accounts: {},
                    accountsDataSlice: [],
                    transactions: {},
                    blocks: {},
                    blocksMeta: {},
                    entry: {},
                    slots: {},
                    transactionsStatus: {},
                }).catch(console.error);
            }
        }, 30000);
    }

    private handleData(data: any): void {
        try {
            const processed = this.processBuffers(data);
            this.dataHandler(processed);
        } catch (error) {
            console.error("Error processing data:", error);
        }
    }

    private handleError(error: any): void {
        console.error("Stream error:", error);
        this.isConnected = false;
    }

    private handleDisconnect(subscribeRequest: SubscribeRequest): void {
        console.log("Stream disconnected");
        this.isConnected = false;
        this.reconnect(subscribeRequest);
    }

    private processBuffers(obj: any): any {
        if (!obj) return obj;
        if (Buffer.isBuffer(obj) || obj instanceof Uint8Array) {
            return bs58.default.encode(obj);
        }
        if (Array.isArray(obj)) {
            return obj.map(item => this.processBuffers(item));
        }
        if (typeof obj === 'object') {
            return Object.fromEntries(
                Object.entries(obj).map(([k, v]) => [k, this.processBuffers(v)])
            );
        }
        return obj;
    }
}

// Entry monitoring implementation
async function monitorEntries() {
    const manager = new GrpcStreamManager(
        "your-grpc-url:2053",
        "your-x-token",
        handleEntryUpdate
    );

    const subscribeRequest: SubscribeRequest = {
        entry: {
            entrySubscribe: {}  // Subscribe to all entries
        },
        accounts: {},
        accountsDataSlice: [],
        slots: {},
        blocks: {},
        blocksMeta: {},
        transactions: {},
        transactionsStatus: {},
        commitment: CommitmentLevel.CONFIRMED,
    };

    console.log('Starting entry monitoring...');
    await manager.connect(subscribeRequest);
}

function handleEntryUpdate(data: any): void {
    if (data?.entry) {
        const entry = data.entry;
        console.log('\n=== Entry Details ===');
        console.log(`Slot: ${entry.slot}`);
        
        if (entry.index !== undefined) {
            console.log(`Entry Index: ${entry.index}`);
        }

        if (entry.numHashes !== undefined) {
            console.log(`Number of Hashes: ${entry.numHashes}`);
        }

        if (entry.hash) {
            console.log(`Hash: ${entry.hash}`);
        }

        if (entry.transactions?.length > 0) {
            console.log('\n=== Entry Transactions ===');
            console.log(`Transaction Count: ${entry.transactions.length}`);
            
            entry.transactions.forEach((tx: any, index: number) => {
                console.log(`\nTransaction ${index + 1}:`);
                if (tx.signature) {
                    console.log(`  Signature: ${tx.signature}`);
                }
                if (tx.isVote !== undefined) {
                    console.log(`  Is Vote: ${tx.isVote}`);
                }
            });
        }

        if (entry.tick !== undefined) {
            console.log('\n=== Tick Information ===');
            console.log(`Is Tick: ${entry.tick}`);
        }

        console.log('\n' + '='.repeat(50) + '\n');
    }
}

// Start entry monitoring
monitorEntries().catch(console.error);

Entry Subscription Options

1. Monitor All Entries

const subscribeRequest: SubscribeRequest = {
    entry: {
        allEntries: {}
    },
    commitment: CommitmentLevel.CONFIRMED,
};

2. Monitor Filtered Entries

const subscribeRequest: SubscribeRequest = {
    entry: {
        filteredEntries: {
            filter: {
                voting: false,  // Exclude vote transactions
                failed: false  // Exclude failed transactions
            }
        }
    },
    commitment: CommitmentLevel.CONFIRMED,
};

Data Structures

Entry Update:
- Slot number (current slot being processed)
- Entry index (position in the block)
- Number of hashes (PoH count)
- Entry hash (unique identifier)
- Transaction count (number of transactions in entry)
- Transactions array (detailed transaction information)
Transaction Details:
- Signature (unique transaction identifier)
- Vote status (whether it's a vote transaction)
- Transaction message
  - Account keys (involved accounts)
  - Header information
  - Instructions (program calls and data)
  - Recent blockhash
- Transaction metadata
  - Status (success/failure)
  - Fee amount
  - Balance changes
  - Log messages

Rust Implementation

use {
    anyhow::Result, futures::{sink::SinkExt, stream::StreamExt}, hex, log::error, solana_sdk::pubkey::Pubkey, std::{collections::HashMap, sync::Arc, time::Duration}, tokio::sync::Mutex, tonic::{metadata::errors::InvalidMetadataValue, transport::Endpoint}, tonic_health::pb::health_client::HealthClient, yellowstone_grpc_client::{GeyserGrpcClient, InterceptorXToken}, yellowstone_grpc_proto::{
        geyser::{
            geyser_client::GeyserClient, subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestFilterBlocks, SubscribeRequestFilterEntry, SubscribeRequestFilterSlots, SubscribeUpdateBlock, SubscribeUpdateEntry, SubscribeUpdateSlot, CommitmentLevel
        },
        prelude::SubscribeRequestPing,
    }
};

/// Manager for handling gRPC stream connections and updates
struct GrpcStreamManager {
    client: GeyserGrpcClient<InterceptorXToken>,
    is_connected: bool,
    reconnect_attempts: u32,
    max_reconnect_attempts: u32,
    reconnect_interval: Duration,
}

impl GrpcStreamManager {
    /// Creates a new GrpcStreamManager instance
    async fn new(endpoint: &str, x_token: &str) -> Result<Arc<Mutex<Self>>> {
        let interceptor = InterceptorXToken {
            x_token: Some(x_token.parse().map_err(|e: InvalidMetadataValue| anyhow::Error::from(e))?),
            x_request_snapshot: true,
        };

        let channel = Endpoint::from_shared(endpoint.to_string())?
            .connect_timeout(Duration::from_secs(10))
            .timeout(Duration::from_secs(10))
            .connect()
            .await
            .map_err(|e| anyhow::Error::from(e))?;

        let client = GeyserGrpcClient::new(
            HealthClient::with_interceptor(channel.clone(), interceptor.clone()),
            GeyserClient::with_interceptor(channel, interceptor),
        );

        Ok(Arc::new(Mutex::new(Self {
            client,
            is_connected: false,
            reconnect_attempts: 0,
            max_reconnect_attempts: 10,
            reconnect_interval: Duration::from_secs(5),
        })))
    }

    /// Handles entry update messages from the gRPC stream
    fn handle_entry_update(&self, entry: &SubscribeUpdateEntry) {
        println!("\n=== Entry Details ===");
        println!("Slot: {}", entry.slot);
        println!("Entry Index: {}", entry.index);
        println!("Number of Hashes: {}", entry.num_hashes);

        if !entry.hash.is_empty() {
            println!("Hash: {}", hex::encode(&entry.hash));
        }

        println!("Executed Transaction Count: {}", entry.executed_transaction_count);
        if entry.starting_transaction_index > 0 {
            println!("Starting Transaction Index: {}", entry.starting_transaction_index);
        }

        println!("\n{}\n", "=".repeat(50));
    }

    /// Establishes connection and handles the subscription stream
    async fn connect(&mut self, request: SubscribeRequest) -> Result<()> {
        let request = request.clone();
        let (mut subscribe_tx, mut stream) = self.client.subscribe_with_request(Some(request.clone())).await?;

        self.is_connected = true;
        self.reconnect_attempts = 0;

        while let Some(message) = stream.next().await {
            match message {
                Ok(msg) => {
                    match msg.update_oneof {
                        Some(UpdateOneof::Entry(entry)) => {
                            self.handle_entry_update(&entry);
                        }
                        Some(UpdateOneof::Ping(_)) => {
                            subscribe_tx
                                .send(SubscribeRequest {
                                    ping: Some(SubscribeRequestPing { id: 1 }),
                                    ..Default::default()
                                })
                                .await?;
                        }
                        Some(UpdateOneof::Pong(_)) => {} // Ignore pong responses
                        _ => {
                            println!("Other update received: {:?}", msg);
                        }
                    }
                }
                Err(err) => {
                    error!("Error: {:?}", err);
                    self.is_connected = false;
                    Box::pin(self.reconnect(request.clone())).await?;
                    break;
                }
            }
        }

        Ok(())
    }

    /// Attempts to reconnect when the connection is lost
    async fn reconnect(&mut self, request: SubscribeRequest) -> Result<()> {
        if self.reconnect_attempts >= self.max_reconnect_attempts {
            println!("Max reconnection attempts reached");
            return Ok(());
        }

        self.reconnect_attempts += 1;
        println!("Reconnecting... Attempt {}", self.reconnect_attempts);

        let backoff = self.reconnect_interval * std::cmp::min(self.reconnect_attempts, 5);
        tokio::time::sleep(backoff).await;

        Box::pin(self.connect(request)).await
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // Initialize gRPC stream manager
    let manager = GrpcStreamManager::new(
        "your-grpc-url:2053",
        "your-x-token",
    ).await?;

    let mut manager_lock = manager.lock().await;

    // Create subscription request for entries only
    let request = SubscribeRequest {
        entry: HashMap::from_iter(vec![(
            "entry".to_string(),
            SubscribeRequestFilterEntry {},
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    };

    println!("Starting entry monitoring...");
    
    // Start the subscription
    let result = manager_lock.connect(request).await;
    if let Err(e) = &result {
        println!("Subscription error: {:?}", e);
    }
    result?;

    Ok(())
}

Go Implementation

package main

import (
    "context"
    "crypto/tls"
    "encoding/hex"
    "fmt"
    "log"
    "sync"
    "time"

    pb "github.com/rpcpool/yellowstone-grpc/examples/golang/proto"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/metadata"
)

func handleEntryUpdate(entry *pb.SubscribeUpdateEntry) {
    fmt.Printf("Entry Update:\n")
    fmt.Printf("  Slot: %d\n", entry.Slot)
    fmt.Printf("  Index: %d\n", entry.Index)
    fmt.Printf("  Num Hashes: %d\n", entry.NumHashes)
    fmt.Printf("  Hash: %s\n", hex.EncodeToString(entry.Hash))
    fmt.Printf("  Transaction Count: %d\n", entry.ExecutedTransactionCount)

    if entry.Transactions != nil {
        fmt.Printf("Processing %d transactions...\n", len(entry.Transactions))
        for i, tx := range entry.Transactions {
            fmt.Printf("Transaction %d:\n", i+1)
            fmt.Printf("  Signature: %s\n", hex.EncodeToString(tx.Signature))
            fmt.Printf("  Is Vote: %v\n", tx.IsVote)

            if tx.Meta != nil {
                fmt.Printf("  Status: %s\n", getTransactionStatus(tx.Meta.Err))
                fmt.Printf("  Fee: %d\n", tx.Meta.Fee)
                fmt.Printf("  Pre Balances: %v\n", tx.Meta.PreBalances)
                fmt.Printf("  Post Balances: %v\n", tx.Meta.PostBalances)
            }
        }
    }
}

func getTransactionStatus(err interface{}) string {
    if err == nil {
        return "Success"
    }
    return "Failed"
}

func main() {
    ctx := context.Background()

    // Create manager with data handler and x-token
    manager, err := NewGrpcStreamManager(
        "your-dedicated-node-url:2053",
        "x-token",
        handleEntryUpdate,
    )
    if err != nil {
        log.Fatal(err)
    }
    defer manager.Close()

    // Create subscription request for entry monitoring
    entryFilter := &pb.SubscribeRequestFilterEntry{
        Filter: &pb.EntryFilter{
            Voting: false,
            Failed: false,
        },
    }

    req := &pb.SubscribeRequest{
        Entry:      entryFilter,
        Commitment: pb.CommitmentLevel_CONFIRMED,
    }

    // Connect and handle updates
    if err := manager.Connect(ctx, req); err != nil {
        log.Fatal(err)
    }

    // Keep the main goroutine running
    select {}
}

Getting Started with Yellowstone gRPC (Geyser Plugin)

A comprehensive setup guide for connecting to Solana's real-time data stream using Yellowstone gRPC.

Introduction

With Dedicated Node, you can leverage the powerful Yellowstone gRPC interface to receive real-time updates about on-chain events. The Geyser plugin provides a high-performance, low-latency stream of Solana blockchain data.

What is gRPC and Why Use It?

gRPC (Google Remote Procedure Call) is a modern, high-performance framework that enables client and server applications to communicate transparently. For Solana blockchain data streaming, gRPC offers several key advantages:

Language Agnostic: Using Protocol Buffers (protobuf), gRPC allows you to write clients in multiple programming languages while maintaining type safety and consistency.
Binary Protocol: Unlike JSON-based REST APIs, gRPC uses a binary protocol that is more efficient in terms of bandwidth and parsing overhead.
Bi-directional Streaming: gRPC supports bi-directional streaming, perfect for receiving continuous updates about blockchain events.
Type Safety: The protobuf schema ensures type safety across all supported languages.

Getting Started

Your Helius Dedicated Node exposes the Yellowstone gRPC interface on port 2053. The service provides various subscription options for different types of blockchain data:

Account updates
Slot updates
Transaction information
Block data
Entry data

Installation & Project Setup

Before diving into specific examples in the following sections, let's set up our development environment for each supported language:

TypeScript/JavaScript

Create a new project:

mkdir ts-yellowstone-client
cd ts-yellowstone-client
npm init -y

Install dependencies:

npm install @triton-one/yellowstone-grpc bs58
npm install typescript ts-node @types/node --save-dev

Create TypeScript configuration:

npx tsc --init

Rust

Create a new project:

cargo new rust-yellowstone-client
cd rust-yellowstone-client

Add to Cargo.toml:

[dependencies]
yellowstone-grpc-client = "1.13.0"
yellowstone-grpc-proto = "1.13.0"
tokio = { version = "1.0", features = ["full"] }
anyhow = "1.0"
futures = "0.3"
tonic = "0.10"
tonic-health = "0.10"
hex = "0.4"
solana-sdk = "1.17"
log = "0.4"

Go

Create a new project:

mkdir go-yellowstone-client
cd go-yellowstone-client

Initialize Go module:

go mod init go-yellowstone-client

Create project structure:

mkdir -p cmd/client
touch cmd/client/main.go

Install required dependencies:

go get github.com/mr-tron/base58@v1.2.0
go get github.com/rpcpool/yellowstone-grpc/examples/golang@latest
go get google.golang.org/grpc@v1.67.1

Your go.mod should look similar to this:

module go-yellowstone-client

go 1.21

require (
	github.com/mr-tron/base58 v1.2.0
	github.com/rpcpool/yellowstone-grpc/examples/golang v0.0.0-20250206164228-b9f96ae944bb
	google.golang.org/grpc v1.67.1
)

require (
	golang.org/x/net v0.28.0 // indirect
	golang.org/x/sys v0.24.0 // indirect
	golang.org/x/text v0.17.0 // indirect
	google.golang.org/genproto/googleapis/rpc v0.0.0-20240814211410-ddb44dafa142 // indirect
	google.golang.org/protobuf v1.35.1 // indirect
)

These setups provide the foundation for the examples in the following sections. Each subsequent guide will build upon this basic setup to demonstrate specific monitoring capabilities.

Available Subscriptions

The Yellowstone gRPC interface allows you to subscribe to various types of data:

1. Account Updates

Subscribe to changes in account data, including:

Account balance changes
Data modifications
Ownership changes
Creation/deletion events

2. Slot Updates

Monitor slots to receive notifications about:

New slots
Confirmed slots
Finalized slots
Parent slot relationships

3. Transaction Information

Stream transactions to receive updates about:

Transaction signatures
Transaction status
Success/failure information
Account involvement

4. Block Data

Access comprehensive block information:

Block metadata
Included transactions
Account updates
Entry information

5. Entry Data

Track low-level entry information:

Entry indexes
Transaction counts
Entry hashes

Commitment Levels

Yellowstone gRPC supports three commitment levels:

PROCESSED: Fastest, but may include unconfirmed data
CONFIRMED: Balance between speed and finality
FINALIZED: Highest level of finality

Choose the appropriate commitment level based on your application's requirements for data finality versus latency.

Next Steps

Continue to the following sections to learn about the core subscription types in Yellowstone gRPC:

Account Monitoring - Monitor account updates and changes
Transaction Monitoring - Track transactions and their status
Slot and Block Monitoring - Monitor slots, blocks, and their metadata
Entry Monitoring - Track low-level block entries and execution

Each guide includes complete working examples in TypeScript, Rust, and Go, along with advanced filtering patterns and best practices.

Enhanced Websockets (beta)

Stream real-time transaction and account updates directly to your applications with our enhanced websocket integration.

What are Enhanced Websockets?

Helius provides Geyser-enhanced WebSockets in addition to Solana's standard WebSocket methods. These enhanced WebSockets offer faster response times than traditional RPC WebSockets. We support two key subscription methods: transactionSubscribe and accountSubscribe.

These Websockets are currently in beta and may occasionally experience delayed data. To address this, we're developing Laserstream, a new product that offers replay capabilities and ensures real-time delivery without delays. Join the Laserstream waitlist for updates.

This feature is available exclusively for business and professional plans. Please note that this is the Enhanced Geyser WebSocket, not the Yellowstone Geyser. For Yellowstone Geyser functionality, a Dedicated Node is required.

Subscription Endpoints

Geyser-enhanced websockets are currently available on mainnet and devnet with the following URLs

Mainnet: wss://atlas-mainnet.helius-rpc.com/?api-key=<API_KEY>

Devnet: wss://atlas-devnet.helius-rpc.com/?api-key=<API_KEY>

Websockets have a 10-minute inactivity timer; implementing health checks and sending pings every minute is heavily recommended to keep the websocket connection alive.

The transactionSubscribe websocket method enables real-time transaction events. To use it, provide a TransactionSubscribeFilter and optionally include TransactionSubscribeOptions for further customization.

TransactionSubscribeFilter

vote: A boolean flag to include/exclude vote-related transactions.

failed: A boolean flag to include/exclude transactions that failed.

signature: Filters updates to a specific transaction based on its signature.

accountInclude: A list of accounts for which you want to receive transaction updates. This means that only one of the accounts must be included in the transaction updates (e.g., Account 1 OR Account 2).

accountExclude: A list of accounts you want to exclude from transaction updates.

accountRequired: Transactions must involve these specified accounts to be included in updates. This means that all of the accounts must be included in the transaction updates (e.g., Account 1 AND Account 2).

You can include up to 50,000 addresses in the accountsInclude, accountExclude and accountRequired arrays.

TransactionSubscribeOptions (Optional)

commitment: Specifies the commitment level for fetching data, dictating at what stage of the transaction lifecycle updates are sent. The possible values are processed, confirmed and finalized

encoding: Sets the encoding format of the returned transaction data. The possible values are base58, base64 and jsonParsed

transactionDetails : Determines the level of detail for the returned transaction data. The possible values are full, signatures, accounts and none

showRewards: A boolean flag indicating if reward data should be included in the transaction updates.

maxSupportedTransactionVersion: Specifies the highest version of transactions you want to receive updates. To get Versioned Transactions, set the value to 1.

maxSupportedTransactionVersion is required to return the accounts and full-level details of a given transaction (i.e., transactionDetails: "accounts" | "full").

Example Payload

{
  "jsonrpc": "2.0",
  "id": 420,
  "method": "transactionSubscribe",
  "params": [
      {
        "vote": false,
        "failed": false,
        "signature": "2dd5zTLrSs2udfNsegFRCnzSyQcPrM9svX6m1UbEM5bSdXXFj3XpqaodtKarLYFP2mTVUsV27sRDdZCgcKhjeD9S",
        "accountInclude": ["pqx3fvvh6b2eZBfLhTtQ5KxzU3CginmgGTmDCjk8TPP"],
        "accountExclude": ["FbfwE8ZmVdwUbbEXdq4ofhuUEiAxeSk5kaoYrJJekpnZ"],
        "accountRequired": ["As1XYY9RdGkjs62isDhLKG3yxMCMatnbanXrqU85XvXW"]
      },
      {
	"commitment": "processed",
    	"encoding": "base64",
    	"transactionDetails": "full",
    	"showRewards": true,
    	"maxSupportedTransactionVersion": 0
      }
  ]
}

Example Notifications

{
    "jsonrpc": "2.0",
    "method": "transactionNotification",
    "params": {
        "subscription": 4743323479349712,
        "result": {
            "transaction": {
                "transaction": [
                    "Ae6zfSExLsJ/E1+q0jI+3ueAtSoW+6HnuDohmuFwagUo2BU4OpkSdUKYNI1dJfMOonWvjaumf4Vv1ghn9f3Avg0BAAEDGycH0OcYRpfnPNuu0DBQxTYPWpmwHdXPjb8y2P200JgK3hGiC2JyC9qjTd2lrug7O4cvSRUVWgwohbbefNgKQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA0HcpwKokfYDDAJTaF/TWRFWm0Gz5/me17PRnnywHurMBAgIAAQwCAAAAoIYBAAAAAAA=",
                    "base64"
                ],
                "meta": {
                    "err": null,
                    "status": {
                        "Ok": null
                    },
                    "fee": 5000,
                    "preBalances": [
                        28279852264,
                        158122684,
                        1
                    ],
                    "postBalances": [
                        28279747264,
                        158222684,
                        1
                    ],
                    "innerInstructions": [],
                    "logMessages": [
                        "Program 11111111111111111111111111111111 invoke [1]",
                        "Program 11111111111111111111111111111111 success"
                    ],
                    "preTokenBalances": [],
                    "postTokenBalances": [],
                    "rewards": null,
                    "loadedAddresses": {
                        "writable": [],
                        "readonly": []
                    },
                    "computeUnitsConsumed": 0
                }
            },
            "signature": "5moMXe6VW7L7aQZskcAkKGQ1y19qqUT1teQKBNAAmipzdxdqVLAdG47WrsByFYNJSAGa9TByv15oygnqYvP6Hn2p",
            "slot": 224341380
        }
    }
}

Code Example

In this example, we are subscribing to transactions that contain the Raydium account 675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8. Whenever a transaction occurs that contains 675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8in the accountKeys of the transaction, we will receive a websocket notification.

Based on the subscription options, the transaction notification will be sent at the processed commitment level,jsonParsedencoding, full transaction details, and will show rewards.

const WebSocket = require('ws');

// Create a WebSocket connection
const ws = new WebSocket('wss://atlas-mainnet.helius-rpc.com/?api-key=<API_KEY>');

// Function to send a request to the WebSocket server
function sendRequest(ws) {
    const request = {
        jsonrpc: "2.0",
        id: 420,
        method: "transactionSubscribe",
        params: [
            {
                accountInclude: ["675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"]
            },
            {
                commitment: "processed",
                encoding: "jsonParsed",
                transactionDetails: "full",
                showRewards: true,
                maxSupportedTransactionVersion: 0
            }
        ]
    };
    ws.send(JSON.stringify(request));
}

// Function to send a ping to the WebSocket server
function startPing(ws) {
    setInterval(() => {
        if (ws.readyState === WebSocket.OPEN) {
            ws.ping();
            console.log('Ping sent');
        }
    }, 30000); // Ping every 30 seconds
}

// Define WebSocket event handlers

ws.on('open', function open() {
    console.log('WebSocket is open');
    sendRequest(ws);  // Send a request once the WebSocket is open
    startPing(ws);    // Start sending pings
});

ws.on('message', function incoming(data) {
    const messageStr = data.toString('utf8');
    try {
        const messageObj = JSON.parse(messageStr);
        console.log('Received:', messageObj);
    } catch (e) {
        console.error('Failed to parse JSON:', e);
    }
});

ws.on('error', function error(err) {
    console.error('WebSocket error:', err);
});

ws.on('close', function close() {
    console.log('WebSocket is closed');
});

Solana's Websockets supports a method that allows you to subscribe to an account and receive notifications via the websocket connection whenever there are changes to the lamports or data associated with a matching account public key. This method aligns directly with the Solana Websocket API specification.

Parameters

string: The account public key, sent in base58 format (required).

object: An optional object used to pass additional parameters.

encoding: Specifies the format for data returned in the AccountNotification. Supported values: base58, base64, base64+zstd, jsonParsed (default is base58).
commitment: Defines the commitment level for the transaction. Supported values: finalized, confirmed, processed (default is finalized).

Example Payload

{
  "jsonrpc": "2.0",
  "id": 420,
  "method": "accountSubscribe",
   "params": [
    "SysvarC1ock11111111111111111111111111111111",
    {
      "encoding": "jsonParsed",
      "commitment": "finalized"
    }
  ]
}

Example Notification

{
    'jsonrpc': '2.0', 
    'method': 'accountNotification', 
    'params': 
        {
          'subscription': 237508762798666, 
          'result': 
           {
            'context': {'slot': 235781083}, 
            'value': 
             {
             'lamports': 1169280, 
             'data': 'BvEhEb6hixL3QPn41gHcyi2CDGKt381jbNKFFCQr6XDTzCTXCuSUG9D', 
             'owner': 'Sysvar1111111111111111111111111111111111111', 
             'executable': False, 
             'rentEpoch': 361, 
             'space': 40
             }
           }
        }
}

Code Example

In this example, we are subscribing to account changes for the account SysvarC1ock11111111111111111111111111111111 . We will see an update whenever a change occurs to the account data or the lamports for this account. This happens at a frequent interval for this specific account as the slot and unixTimestamp are both a part of the returned account data.

// Create a WebSocket connection
const ws = new WebSocket('wss://atlas-mainnet.helius-rpc.com?api-key=<API_KEY>');

// Function to send a request to the WebSocket server
function sendRequest(ws) {
    const request = {
        jsonrpc: "2.0",
        id: 420,
        method: "accountSubscribe",
        params: [
            "SysvarC1ock11111111111111111111111111111111", // pubkey of account we want to subscribe to
            {
                encoding: "jsonParsed", // base58, base64, base65+zstd, jsonParsed
                commitment: "confirmed", // defaults to finalized if unset
            }
        ]
    };
    ws.send(JSON.stringify(request));
}

// Function to send a ping to the WebSocket server
function startPing(ws) {
    setInterval(() => {
        if (ws.readyState === WebSocket.OPEN) {
            ws.ping();
            console.log('Ping sent');
        }
    }, 30000); // Ping every 30 seconds
}

// Define WebSocket event handlers

ws.on('open', function open() {
    console.log('WebSocket is open');
    sendRequest(ws);  // Send a request once the WebSocket is open
    startPing(ws);    // Start sending pings
});

ws.on('message', function incoming(data) {
    const messageStr = data.toString('utf8');
    try {
        const messageObj = JSON.parse(messageStr);
        console.log('Received:', messageObj);
    } catch (e) {
        console.error('Failed to parse JSON:', e);
    }
});

ws.on('error', function error(err) {
    console.error('WebSocket error:', err);
});

ws.on('close', function close() {
    console.log('WebSocket is closed');
});

package main

import (
	"context"
	"crypto/tls"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/mr-tron/base58"
	pb "github.com/rpcpool/yellowstone-grpc/examples/golang/proto"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)

type GrpcStreamManager struct {
	mu                   sync.Mutex
	conn                 *grpc.ClientConn
	client               pb.GeyserClient
	stream               pb.Geyser_SubscribeClient
	isConnected          bool
	reconnectAttempts    int
	maxReconnectAttempts int
	reconnectInterval    time.Duration
	dataHandler          func(*pb.SubscribeUpdate)
	xToken               string
}

func NewGrpcStreamManager(endpoint string, xToken string, dataHandler func(*pb.SubscribeUpdate)) (*GrpcStreamManager, error) {
	// Create gRPC connection with interceptor for x-token
	ctx := metadata.NewOutgoingContext(
		context.Background(),
		metadata.New(map[string]string{"x-token": xToken}),
	)

	// Configure TLS
	config := &tls.Config{
		InsecureSkipVerify: true,
	}

	conn, err := grpc.DialContext(ctx, endpoint,
		grpc.WithTransportCredentials(credentials.NewTLS(config)),
		grpc.WithInitialWindowSize(1<<30),
		grpc.WithInitialConnWindowSize(1<<30),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(64*1024*1024),
			grpc.MaxCallRecvMsgSize(64*1024*1024),
		),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to connect: %v", err)
	}

	return &GrpcStreamManager{
		conn:                 conn,
		client:               pb.NewGeyserClient(conn),
		isConnected:          false,
		reconnectAttempts:    0,
		maxReconnectAttempts: 10,
		reconnectInterval:    5 * time.Second,
		dataHandler:          dataHandler,
		xToken:               xToken,
	}, nil
}

func (m *GrpcStreamManager) Connect(ctx context.Context, req *pb.SubscribeRequest) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	log.Println("Attempting to connect...")

	// Add x-token to context
	ctx = metadata.NewOutgoingContext(
		ctx,
		metadata.New(map[string]string{"x-token": m.xToken}),
	)

	stream, err := m.client.Subscribe(ctx)
	if err != nil {
		log.Printf("Failed to subscribe: %v", err)
		return m.reconnect(ctx, req)
	}

	if err := stream.Send(req); err != nil {
		log.Printf("Failed to send request: %v", err)
		return m.reconnect(ctx, req)
	}

	m.stream = stream
	m.isConnected = true
	m.reconnectAttempts = 0
	log.Println("Connection established")

	// Start ping goroutine
	go func() {
		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if !m.isConnected {
					return
				}
				pingReq := &pb.SubscribeRequest{
					Ping: &pb.SubscribeRequestPing{Id: 1},
				}
				if err := m.stream.Send(pingReq); err != nil {
					log.Printf("Ping failed: %v", err)
					m.handleDisconnect(ctx, req)
					return
				}
			}
		}
	}()

	// Process updates
	go func() {
		for {
			update, err := m.stream.Recv()
			if err != nil {
				log.Printf("Stream error: %v", err)
				m.handleDisconnect(ctx, req)
				return
			}

			m.dataHandler(update)
		}
	}()

	return nil
}

func (m *GrpcStreamManager) reconnect(ctx context.Context, req *pb.SubscribeRequest) error {
	if m.reconnectAttempts >= m.maxReconnectAttempts {
		return fmt.Errorf("max reconnection attempts reached")
	}

	m.reconnectAttempts++
	log.Printf("Reconnecting... Attempt %d", m.reconnectAttempts)

	backoff := m.reconnectInterval * time.Duration(min(m.reconnectAttempts, 5))
	time.Sleep(backoff)

	return m.Connect(ctx, req)
}

func (m *GrpcStreamManager) handleDisconnect(ctx context.Context, req *pb.SubscribeRequest) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if !m.isConnected {
		return
	}

	m.isConnected = false
	if err := m.reconnect(ctx, req); err != nil {
		log.Printf("Failed to reconnect: %v", err)
	}
}

func (m *GrpcStreamManager) Close() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.isConnected = false
	if m.conn != nil {
		return m.conn.Close()
	}
	return nil
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

func handleAccountUpdate(update *pb.SubscribeUpdate) {
	switch u := update.GetUpdateOneof().(type) {
	case *pb.SubscribeUpdate_Block:
		handleBlockUpdate(u.Block)
	case *pb.SubscribeUpdate_Slot:
		handleSlotUpdate(u.Slot)
	case *pb.SubscribeUpdate_Transaction:
		handleTransactionUpdate(u.Transaction)
	}
}

func handleBlockUpdate(block *pb.SubscribeUpdateBlock) {
	log.Printf("\n=== Block Details ===\n")
	log.Printf("Slot: %d\n", block.GetSlot())
	log.Printf("Parent Slot: %d\n", block.GetParentSlot())
	log.Printf("Blockhash: %s\n", block.GetBlockhash())
	log.Printf("Previous Blockhash: %s\n", block.GetParentBlockhash())

	if txs := block.GetTransactions(); len(txs) > 0 {
		log.Printf("\n=== Block Transactions ===\n")
		log.Printf("Transaction Count: %d\n", len(txs))

		for i, tx := range txs {
			if txInfo := tx.GetTransaction(); txInfo != nil {
				log.Printf("\nTransaction %d:\n", i+1)
				if signatures := txInfo.GetSignatures(); len(signatures) > 0 {
					log.Printf("  Signature: %s\n", base58.Encode(signatures[0]))
				}

				if meta := tx.GetMeta(); meta != nil {
					log.Printf("  Status: %v\n", meta.GetErr() == nil)
					log.Printf("  Fee: %d lamports\n", meta.GetFee())
				}
			}
		}
	}

	if rewards := block.GetRewards(); rewards != nil {
		if rewardList := rewards.GetRewards(); len(rewardList) > 0 {
			log.Printf("\n=== Block Rewards ===\n")
			for _, reward := range rewardList {
				pubkey := []byte(reward.GetPubkey())
				log.Printf("  Account: %s\n", base58.Encode(pubkey))
				log.Printf("  Lamports: %d\n", reward.GetLamports())
				log.Printf("  Post Balance: %d\n", reward.GetPostBalance())
				log.Printf("  Reward Type: %d\n", reward.GetRewardType())
				log.Printf("  Commission: %d\n", reward.GetCommission())
				log.Printf("  ---\n")
			}
		}
	}
	log.Printf("\n%s\n", strings.Repeat("=", 50))
}

func handleSlotUpdate(slot *pb.SubscribeUpdateSlot) {
	log.Printf("\n=== Slot Update ===\n")
	log.Printf("Slot: %d\n", slot.GetSlot())

	if parent := slot.GetParent(); parent > 0 {
		log.Printf("Parent Slot: %d\n", parent)
	}

	log.Printf("Status: %v\n", slot.GetStatus())

	if deadError := slot.GetDeadError(); deadError != "" {
		log.Printf("Dead Error: %s\n", deadError)
	}

	log.Printf("\n%s\n", strings.Repeat("=", 50))
}

func handleTransactionUpdate(tx *pb.SubscribeUpdateTransaction) {
	log.Printf("Transaction Update:\n")
	log.Printf("  Slot: %d\n", tx.GetSlot())

	txInfo := tx.GetTransaction()
	if txInfo != nil {
		// Print signature if available
		if len(txInfo.GetSignature()) > 0 {
			log.Printf("  Signature: %s\n", base58.Encode(txInfo.GetSignature()))
		}

		// Print if it's a vote transaction
		log.Printf("  Is Vote: %v\n", txInfo.GetIsVote())

		// Print transaction index
		log.Printf("  Transaction Index: %d\n", txInfo.GetIndex())

		// Print transaction details
		if tx := txInfo.GetTransaction(); tx != nil {
			if msg := tx.GetMessage(); msg != nil {
				if accounts := msg.GetAccountKeys(); len(accounts) > 0 {
					log.Printf("  Account Keys:\n")
					for _, acc := range accounts {
						if len(acc) > 0 {
							log.Printf("    - %s\n", base58.Encode(acc))
						}
					}
				}
			}
		}

		// Print status and metadata
		if meta := txInfo.GetMeta(); meta != nil {
			log.Printf("  Status: %v\n", meta.GetErr() == nil)
			log.Printf("  Fee: %d\n", meta.GetFee())

			// Print log messages if any
			if len(meta.GetLogMessages()) > 0 {
				log.Printf("  Log Messages:\n")
				for _, msg := range meta.GetLogMessages() {
					log.Printf("    - %s\n", msg)
				}
			}
		}
	}
}

func boolPtr(b bool) *bool {
	return &b
}

func main() {
	ctx := context.Background()

	// Create manager with data handler and x-token
	manager, err := NewGrpcStreamManager(
		"your-grpc-url:2053",
        	"your-x-token",
		handleAccountUpdate,
	)
	if err != nil {
		log.Fatal(err)
	}
	defer manager.Close()

	// Create subscription request for blocks and slots
	blocks := make(map[string]*pb.SubscribeRequestFilterBlocks)
	blocks["blocks"] = &pb.SubscribeRequestFilterBlocks{
		AccountInclude: []string{
			"TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA", // Token Program
			"11111111111111111111111111111111",            // System Program
			"So11111111111111111111111111111111111111112", // Wrapped SOL
		},
		IncludeTransactions: boolPtr(true),
		IncludeAccounts:     boolPtr(true),
		IncludeEntries:      boolPtr(false),
	}

	slots := make(map[string]*pb.SubscribeRequestFilterSlots)
	slots["slots"] = &pb.SubscribeRequestFilterSlots{
		FilterByCommitment: boolPtr(true),
	}

	commitment := pb.CommitmentLevel_CONFIRMED
	req := &pb.SubscribeRequest{
		Blocks:     blocks,
		Slots:      slots,
		Commitment: &commitment,
	}

	log.Println("Starting block and slot monitoring...")
	log.Println("Monitoring blocks for Token Program, System Program, and Wrapped SOL activities...")

	// Connect and handle updates
	if err := manager.Connect(ctx, req); err != nil {
		log.Fatal(err)
	}

	// Keep the main goroutine running
	select {}
}