Module provider_stream 

Processed provider-stream ingress types and adapters. Processed provider-stream ingress surfaces for SOF.

Use this module when the upstream source is already beyond raw shreds, for example Yellowstone gRPC or LaserStream-style processed transaction feeds. These updates bypass SOF’s packet, shred, FEC, and reconstruction stages and enter directly at the plugin/derived-state transaction layer.

Built-in mode capability summary:

• YellowstoneGrpc: built-in Yellowstone transaction feed
• YellowstoneGrpcTransactionStatus: built-in Yellowstone transaction-status feed
• YellowstoneGrpcAccounts: built-in Yellowstone account-update feed
• YellowstoneGrpcBlockMeta: built-in Yellowstone block-meta feed
• YellowstoneGrpcSlots: built-in Yellowstone slot feed
• LaserStream: built-in LaserStream transaction feed
• LaserStreamTransactionStatus: built-in LaserStream transaction-status feed
• LaserStreamAccounts: built-in LaserStream account-update feed
• LaserStreamBlockMeta: built-in LaserStream block-meta feed
• LaserStreamSlots: built-in LaserStream slot feed
• WebsocketTransaction: built-in websocket transactionSubscribe
• WebsocketLogs: built-in websocket logsSubscribe
• WebsocketAccount: built-in websocket accountSubscribe
• WebsocketProgram: built-in websocket programSubscribe

Each built-in source config can report its matching runtime mode directly through runtime_mode(). ProviderStreamMode::Generic remains the typed custom-adapter path and the fan-in mode when you want to combine multiple heterogeneous upstream sources into one runtime ingress.

Generic provider producers may still enqueue TransactionViewBatch, BlockMeta, RecentBlockhash, SlotStatus, ClusterTopology, LeaderSchedule, or Reorg updates directly.

ProviderStreamMode::Generic is SOF’s typed custom-adapter mode. Your producer ingests an upstream format and maps it into one of the ProviderStreamUpdate variants below before handing it to SOF.

Built-in provider source configs extend that same typed surface:

• websocket:
  • [websocket::WebsocketTransactionConfig] can target [websocket::WebsocketPrimaryStream::Transaction], [websocket::WebsocketPrimaryStream::Account], or [websocket::WebsocketPrimaryStream::Program]
  • [websocket::WebsocketLogsConfig] targets logsSubscribe
• Yellowstone:
  • [yellowstone::YellowstoneGrpcConfig] can target [yellowstone::YellowstoneGrpcStream::Transaction], [yellowstone::YellowstoneGrpcStream::TransactionStatus], [yellowstone::YellowstoneGrpcStream::Accounts], or [yellowstone::YellowstoneGrpcStream::BlockMeta]
  • [yellowstone::YellowstoneGrpcSlotsConfig] targets slot updates
• LaserStream:
  • [laserstream::LaserStreamConfig] can target [laserstream::LaserStreamStream::Transaction], [laserstream::LaserStreamStream::TransactionStatus], [laserstream::LaserStreamStream::Accounts], or [laserstream::LaserStreamStream::BlockMeta]
  • [laserstream::LaserStreamSlotsConfig] targets slot updates

Those source selectors do not create a second runtime API. They extend the existing provider config objects and emit the matching ProviderStreamUpdate variants into the same runtime dispatch path.

Built-in configs may also set:

• a stable source instance label for observability via with_source_instance(...)
• whether one source is readiness-gating or optional via with_readiness(...)
• an operational role via with_source_role(...)
• an explicit arbitration priority via with_source_priority(...)
• duplicate handling via with_source_arbitration(...)
  • EmitAll keeps the historical behavior and forwards overlapping provider events from every source
  • FirstSeen suppresses later overlapping duplicates across sources
  • FirstSeenThenPromote keeps the first event immediate, but allows one later higher-priority duplicate to promote through

Generic multi-source producers can reserve one stable source identity with [ProviderStreamFanIn::sender_for_source]. The returned [ReservedProviderStreamSender] automatically attributes every update it sends to that reserved provider source. Generic producers can set the same source policy directly on [ProviderSourceIdentity] with with_role(...), with_priority(...), and with_arbitration(...).

Generic readiness becomes source-aware only after a custom producer emits [ProviderStreamUpdate::Health] for that reserved source. Until then, ProviderStreamMode::Generic falls back to progress-based readiness and only knows that typed updates are flowing at all.

Fan-in duplicate arbitration is source-aware and keyed by the logical event, not just by queue position. That means two sources carrying the same transaction or control-plane item can now either:

• both dispatch (EmitAll, the default)
• dispatch once (FirstSeen)
• dispatch once immediately and then allow one later higher-priority source to promote (FirstSeenThenPromote)

Variant-to-runtime mapping:

• Transaction:
  • drives on_transaction
  • drives derived-state transaction apply when enabled
  • synthesizes on_recent_blockhash from the transaction message when that hook is requested
• SerializedTransaction:
  • same transaction-family path, but lets SOF prefilter before full decode
• TransactionLog:
  • drives on_transaction_log
• TransactionStatus:
  • drives on_transaction_status
• TransactionViewBatch:
  • drives on_transaction_view_batch
• AccountUpdate:
  • drives on_account_update
• BlockMeta:
  • drives on_block_meta
• RecentBlockhash:
  • drives on_recent_blockhash
• SlotStatus:
  • drives on_slot_status
• ClusterTopology:
  • drives on_cluster_topology
• LeaderSchedule:
  • drives on_leader_schedule
• Reorg:
  • drives on_reorg
• Health:
  • updates provider health/readiness/observability
  • does not dispatch into plugin hooks

Feed Provider Transactions Into SOF

use std::sync::Arc;

use solana_hash::Hash;
use solana_keypair::Keypair;
use solana_message::{Message, VersionedMessage};
use solana_signer::Signer;
use solana_transaction::versioned::VersionedTransaction;
use sof::{
    event::{TxCommitmentStatus, TxKind},
    framework::TransactionEvent,
    provider_stream::{
        create_provider_stream_queue, ProviderStreamMode, ProviderStreamUpdate,
    },
    runtime::ObserverRuntime,
};

let (tx, rx) = create_provider_stream_queue(128);
let signer = Keypair::new();
let message = Message::new(&[], Some(&signer.pubkey()));
let transaction = VersionedTransaction::try_new(VersionedMessage::Legacy(message), &[&signer])?;

tx.send(ProviderStreamUpdate::Transaction(TransactionEvent {
    slot: 1,
    commitment_status: TxCommitmentStatus::Processed,
    confirmed_slot: None,
    finalized_slot: None,
    signature: transaction.signatures.first().copied(),
    tx: Arc::new(transaction),
    kind: TxKind::NonVote,
}))
.await?;

ObserverRuntime::new()
    .with_provider_stream_ingress(ProviderStreamMode::Generic, rx)
    .run_until(async {})
    .await?;

Structs

ProviderSourceHealthEvent

One provider source health transition observed by SOF.

ProviderSourceIdentity

Stable provider source instance identity used in runtime health and provider-origin events.

ProviderSourceIdentityRegistrationError

Duplicate provider source identity registration failure for multi-source fan-in.

ProviderStreamFanIn

Helper for feeding one SOF provider queue from multiple provider sources.

ReservedProviderStreamSender

One reserved provider source identity plus a sender bound to that reservation.

SerializedTransactionEvent

One serialized provider-fed transaction that has not yet been materialized.

Enums

ProviderReplayMode

Replay policy for processed provider transaction streams.

ProviderSourceArbitrationMode

Duplicate arbitration mode for one provider source inside fan-in.

ProviderSourceHealthReason

Typed reason for one provider source health transition.

ProviderSourceHealthStatus

Health state for one provider source feeding SOF.

ProviderSourceId

Stable provider source identifier used in runtime health reporting.

ProviderSourceReadiness

Readiness class for one provider source observed by SOF.

ProviderSourceRole

Relative operational role for one provider source inside a fan-in graph.

ProviderStreamMode

Identifies the processed provider family driving SOF’s direct plugin ingress.

ProviderStreamUpdate

One processed provider-stream update accepted by SOF.

Constants

DEFAULT_PROVIDER_STREAM_QUEUE_CAPACITY

Default queue capacity for processed provider-stream ingress.

Functions

create_provider_stream_fan_in

Creates one bounded queue plus a typed helper for multi-source provider fan-in.

create_provider_stream_queue

Creates one bounded queue for processed provider-stream updates.

Type Aliases

ProviderSourceRef

Shared provider source identity carried by provider-origin events.

ProviderStreamReceiver

Receiver type for processed provider-stream ingress.

ProviderStreamSender

Sender type for processed provider-stream ingress.