Streaming Token Accounts

Stream light-token accounts via Laserstream. Same base layout as SPL Token.

This guide is for teams building custom data pipelines (aggregators, market makers). If you just need account lookups, use get_account_interface instead.

Architecture

Light token accounts share the same base layout as SPL Token (165 bytes), so you can use your existing parser. The streaming setup requires two gRPC subscriptions, both targeting the Light Token Program:

Subscription	Detects	How
Account sub (owner: cToken...)	Hot state + cold-to-hot	Pubkey cache lookup
Transaction sub (account_include: cToken...)	Hot-to-cold	Balance heuristic (pre > 0, post == 0)

The account subscription delivers all state changes while accounts are hot. The transaction subscription is needed to detect accounts going cold (compress_and_close changes the owner to System Program, which the account subscription no longer matches).

Parsing

use spl_pod::bytemuck::pod_from_bytes;
use spl_token_2022_interface::pod::PodAccount; // works for SPL-token, SPL-token-2022, and Light-token

let parsed: &PodAccount = pod_from_bytes(&data[..165])?;

For accounts with extensions, truncate to 165 bytes before parsing.

Streaming

Cargo.toml

[dependencies]
helius-laserstream = "0.1"
tokio = { version = "1", features = ["full"] }
futures = "0.3"
bs58 = "0.5"
borsh = "0.10"
light-token-interface = "0.3.0"

use futures::StreamExt;
use helius_laserstream::grpc::subscribe_request_filter_accounts_filter::Filter;
use helius_laserstream::grpc::subscribe_request_filter_accounts_filter_memcmp::Data;
use helius_laserstream::grpc::{
    SubscribeRequestFilterAccounts, SubscribeRequestFilterAccountsFilter,
    SubscribeRequestFilterAccountsFilterMemcmp, SubscribeRequestFilterTransactions,
};
use helius_laserstream::{subscribe, LaserstreamConfig};

const LIGHT_TOKEN_PROGRAM_ID: &str = "cTokenmWW8bLPjZEBAUgYy3zKxQZW6VKi7bqNFEVv3m";
const TOKEN_ACCOUNT_SIZE: u64 = 165;
const ACCOUNT_TYPE_OFFSET: u64 = 165;
const ACCOUNT_TYPE_TOKEN: u8 = 2;

1. Connect

   Mainnet

   let config = LaserstreamConfig::new(
       "https://laserstream-mainnet-ewr.helius-rpc.com".to_string(),
       std::env::var("HELIUS_API_KEY")?,
   );
   

   Devnet

   let config = LaserstreamConfig::new(
       "https://laserstream-devnet-ewr.helius-rpc.com".to_string(),
       std::env::var("HELIUS_API_KEY")?,
   );
   
2. Subscribe

   let mut request = helius_laserstream::grpc::SubscribeRequest::default();
   
   // 1. Account sub: hot state tracking + cold-to-hot detection.
   request.accounts.insert(
       "light_tokens".to_string(),
       SubscribeRequestFilterAccounts {
           owner: vec![LIGHT_TOKEN_PROGRAM_ID.to_string()],
           filters: vec![SubscribeRequestFilterAccountsFilter {
               filter: Some(Filter::Datasize(TOKEN_ACCOUNT_SIZE)),
           }],
           nonempty_txn_signature: Some(true),
           ..Default::default()
       },
   );
   request.accounts.insert(
       "light_tokens_extended".to_string(),
       SubscribeRequestFilterAccounts {
           owner: vec![LIGHT_TOKEN_PROGRAM_ID.to_string()],
           filters: vec![SubscribeRequestFilterAccountsFilter {
               filter: Some(Filter::Memcmp(SubscribeRequestFilterAccountsFilterMemcmp {
                   offset: ACCOUNT_TYPE_OFFSET,
                   data: Some(Data::Bytes(vec![ACCOUNT_TYPE_TOKEN])),
               })),
           }],
           nonempty_txn_signature: Some(true),
           ..Default::default()
       },
   );
   
   // 2. Transaction sub: hot-to-cold detection.
   request.transactions.insert(
       "light_token_txns".to_string(),
       SubscribeRequestFilterTransactions {
           vote: Some(false),
           failed: Some(false),
           account_include: vec![LIGHT_TOKEN_PROGRAM_ID.to_string()],
           ..Default::default()
       },
   );
   
   let (stream, _handle) = subscribe(config, request);
   tokio::pin!(stream);
   

Detecting transitions

Hot-to-cold

For each transaction update, find accounts whose lamport balance dropped to zero. The cache.remove call ensures only accounts you’re already tracking are processed:

Two data structures:

• cache: HashMap<[u8; 32], T> — hot account state (for quoting/routing)
• cold_cache: HashMap<[u8; 32], AccountInterface> — cold accounts with ColdContext (for building load instructions)

use helius_laserstream::grpc::subscribe_update::UpdateOneof;

Some(UpdateOneof::Transaction(tx_update)) => {
    if let Some(ref tx_info) = tx_update.transaction {
        for pubkey in find_closed_accounts(tx_info) {
            if cache.remove(&pubkey).is_some() {
                // Async: fetch AccountInterface with ColdContext.
                // Cold accounts are inactive, so this completes well
                // before anyone tries to swap through them.
                let rpc = rpc.clone();
                let cold_cache = cold_cache.clone();
                tokio::spawn(async move {
                    if let Ok(Some(iface)) = rpc.get_account_interface(&pubkey, None).await {
                        cold_cache.insert(pubkey, iface);
                    }
                });
            }
        }
    }
}

fn find_closed_accounts(
    tx_info: &helius_laserstream::grpc::SubscribeUpdateTransactionInfo,
) -> Vec<[u8; 32]> {
    let meta = match &tx_info.meta {
        Some(m) => m,
        None => return vec![],
    };
    let msg = match tx_info.transaction.as_ref().and_then(|t| t.message.as_ref()) {
        Some(m) => m,
        None => return vec![],
    };

    let mut all_keys: Vec<&[u8]> = msg.account_keys.iter().map(|k| k.as_slice()).collect();
    all_keys.extend(meta.loaded_writable_addresses.iter().map(|k| k.as_slice()));
    all_keys.extend(meta.loaded_readonly_addresses.iter().map(|k| k.as_slice()));

    let mut closed = Vec::new();
    for (i, key) in all_keys.iter().enumerate() {
        if key.len() == 32
            && meta.pre_balances.get(i).copied().unwrap_or(0) > 0
            && meta.post_balances.get(i).copied().unwrap_or(1) == 0
        {
            closed.push(<[u8; 32]>::try_from(*key).unwrap());
        }
    }
    closed
}

cache.remove filters out unrelated closures in the same transaction. No discriminator check is needed — compress_and_close always drains lamports to zero.

To build transactions that decompress cold accounts, see Router Integration.

Cold-to-hot

When a token account is decompressed, the account subscription delivers the re-created account. Match its pubkey against cold_cache:

Some(UpdateOneof::Account(account_update)) => {
    if let Some(account) = account_update.account {
        let pubkey: [u8; 32] = account.pubkey.as_slice().try_into().unwrap();
        let parsed: &PodAccount = pod_from_bytes(&account.data[..165])?;

        cold_cache.remove(&pubkey); // no longer cold
        cache.insert(pubkey, *parsed);
    }
}

Point queries

getAccountInfo returns null for cold accounts. get_account_interface() races hot and cold lookups and returns raw account bytes that work with your standard SPL parser:

use light_client::rpc::{LightClient, LightClientConfig, Rpc};
use spl_pod::bytemuck::pod_from_bytes;
use spl_token_2022_interface::pod::PodAccount;

let config = LightClientConfig::new(
    "https://api.devnet.solana.com".to_string(),
    Some("https://photon.helius.com?api-key=YOUR_KEY".to_string()),
);
let client = LightClient::new(config).await?;
let result = client.get_account_interface(&pubkey, None).await?;

if let Some(account) = result.value {
    let parsed: &PodAccount = pod_from_bytes(&account.data()[..165])?;
    if account.is_cold() {
        // Compressed -- still valid for routing.
    }
}

Data layout

165 bytes base, identical to SPL Token Account.

Field	Offset	Size
mint	0	32
owner	32	32
amount	64	8
delegate	72	36
state	108	1
is_native	109	12
delegated_amount	121	8
close_authority	129	36
account_type	165	1

account_type = 2 at byte 165 indicates extensions follow (borsh-encoded Option<Vec<ExtensionStruct>>).

Show Light extensions

These are not needed for indexing or trading.

use borsh::BorshDeserialize;
use light_token_interface::state::{Token, ExtensionStruct};

let token = Token::deserialize(&mut data.as_slice())?;

if let Some(exts) = &token.extensions {
    for ext in exts {
        if let ExtensionStruct::Compressible(info) = ext {
            // info.compression_authority, info.rent_sponsor, info.last_claimed_slot
        }
    }
}

Variant	Description
TokenMetadata(TokenMetadata)	Name, symbol, URI, additional metadata
PausableAccount(PausableAccountExtension)	Marker: mint is pausable (no data; pause state lives on mint)
PermanentDelegateAccount(PermanentDelegateAccountExtension)	Marker: mint has permanent delegate
TransferFeeAccount(TransferFeeAccountExtension)	Withheld fees from transfers
TransferHookAccount(TransferHookAccountExtension)	Marker: mint has transfer hook
CompressedOnly(CompressedOnlyExtension)	Compressed-only token (stores delegated amount)
Compressible(CompressibleExtension)	Compression config: authority, rent sponsor, timing

Source: light-token-interface

Streaming Mint Accounts