Sasedev/khadhroony-solana-project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.3.12-pre.002

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-09 07:12:37 +02:00
parent 08e9a34183
commit 91438e8214
12 changed files with 937 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
// version: 8
// version: 9
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -11,13 +11,14 @@
//! and the caller-runtime-owned lifecycle with private child-task supervision. This tranche also
//! owns bounded source-neutral admission, common RAW canonicalization/assembly and backend-neutral
//! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API;
//! no live source or Transport dependency exists.
//! the first validated Yellowstone/HTTP runtime-resource contract exists, but no live stream is opened yet.
mod admission;
mod error;
mod identity;
mod persistence;
mod runtime;
mod runtime_resources;
mod settings;
mod snapshot;
@@ -43,6 +44,10 @@ pub use self::runtime::RawTransactionIngestHandle;
pub use self::runtime::RawTransactionIngestTerminalFuture;
/// Entry point owning synchronous validation and task launch for one RAW transaction ingest Worker run.
pub use self::runtime::RawTransactionIngestWorker;
/// Caller-composed runtime resources for the first Yellowstone + HTTP source family.
pub use self::runtime_resources::RawTransactionIngestRuntimeResources;
/// Validated Yellowstone + HTTP source contract owned by the continuous RAW transaction ingest Worker.
pub use self::runtime_resources::RawTransactionIngestYellowstoneSource;
/// Default bounded admission queue capacity for one RAW transaction ingest Worker.
pub use self::settings::DEFAULT_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY;
/// Default number of concurrent Store persistence operations for one RAW transaction ingest Worker.

22
crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
// version: 7
// version: 8
type PersistencePort = std::sync::Arc<dyn crate::RawTransactionIngestPersistencePort + 'static>;
type PersistenceTasks = tokio::task::JoinSet<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>>;
@@ -95,6 +95,26 @@ impl crate::RawTransactionIngestWorker {
}
return start_foundation(settings, runtime, std::option::Option::Some(store));
}
/// Starts one Worker with caller-composed runtime resources after synchronous network validation, without opening the live source before its dedicated tranche.
pub fn start_with_runtime_resources(
settings: crate::RawTransactionIngestSettings,
store: std::sync::Arc<ksp_store_lib::Store>,
runtime_resources: crate::RawTransactionIngestRuntimeResources,
) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle> {
let runtime = match current_runtime_handle() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let store_snapshot = store.runtime_snapshot();
if let std::result::Result::Err(error) = validate_store_network(&settings, store_snapshot.network()) {
return std::result::Result::Err(error);
}
if let std::result::Result::Err(error) = runtime_resources.validate_network(settings.network()) {
return std::result::Result::Err(error);
}
return start_foundation(settings, runtime, std::option::Option::Some(store));
}
}
fn begin_stopping(

136
crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs Normal file
View File

@@ -0,0 +1,136 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
// version: 1
/// Validated Yellowstone plus HTTP runtime source owned by the continuous RAW transaction ingest Worker.
///
/// The Transport-owned channel, subscribe request, HTTP pool and hydration role remain private. Construction validates only deterministic source-composition
/// invariants and performs no network I/O.
pub struct RawTransactionIngestYellowstoneSource {
yellowstone_channel: ksp_onchain_transport_lib::YellowstoneGrpcChannel,
subscribe_request: ksp_onchain_transport_lib::YellowstoneSubscribeRequest,
http_pool: ksp_onchain_transport_lib::HttpTransportPool,
hydration_role: ksp_onchain_transport_lib::HttpRoleName,
}
impl crate::RawTransactionIngestYellowstoneSource {
/// Creates one validated Yellowstone source contract without opening a stream or issuing HTTP requests.
pub fn new(
yellowstone_channel: ksp_onchain_transport_lib::YellowstoneGrpcChannel,
subscribe_request: ksp_onchain_transport_lib::YellowstoneSubscribeRequest,
http_pool: ksp_onchain_transport_lib::HttpTransportPool,
hydration_role: ksp_onchain_transport_lib::HttpRoleName,
) -> ksp_core_lib::Result<Self> {
if subscribe_request.validate().is_err() {
return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_request_invalid"));
}
if ingestion_filter_count(&subscribe_request) == 0 {
return std::result::Result::Err(crate::runtime_error("runtime_resources.ingestion_filter_missing"));
}
match subscribe_request.commitment() {
std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed)
| std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => {},
std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed) | std::option::Option::None => {
return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_commitment_invalid"));
},
}
let method = match ksp_onchain_transport_lib::find_http_rpc_method("getTransaction") {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_method_missing")),
};
let compatible_http_routes = compatible_http_route_count(&http_pool, &hydration_role, method.request_kind(), yellowstone_channel.cluster().as_str());
let compatible_http_routes = match compatible_http_routes {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if compatible_http_routes == 0 {
return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_role_unsupported"));
}
return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role });
}
}
impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let http_snapshot = self.http_pool.snapshot();
return formatter
.debug_struct("RawTransactionIngestYellowstoneSource")
.field("yellowstone_endpoint_name", &self.yellowstone_channel.endpoint_name())
.field("yellowstone_provider", &self.yellowstone_channel.provider().as_str())
.field("network", &self.yellowstone_channel.cluster().as_str())
.field("transaction_filter_count", &self.subscribe_request.transaction_filter_count())
.field("transaction_status_filter_count", &self.subscribe_request.transaction_status_filter_count())
.field("block_filter_count", &self.subscribe_request.block_filter_count())
.field("commitment", &self.subscribe_request.commitment())
.field("has_from_slot", &self.subscribe_request.from_slot().is_some())
.field("hydration_role", &self.hydration_role.as_str())
.field("http_endpoint_count", &http_snapshot.endpoint_count())
.finish();
}
}
/// Caller-composed runtime resources accepted by the continuous RAW transaction ingest Worker.
///
/// This first runtime-resource contract owns exactly one Yellowstone source. It deliberately does not expose a provider enum, source collection, callback,
/// enqueue surface or lower-layer client escape hatch.
pub struct RawTransactionIngestRuntimeResources {
yellowstone_source: crate::RawTransactionIngestYellowstoneSource,
}
impl crate::RawTransactionIngestRuntimeResources {
/// Owns the first validated productive-source contract for a future runtime start.
#[must_use]
pub fn new(yellowstone_source: crate::RawTransactionIngestYellowstoneSource) -> Self {
return Self { yellowstone_source };
}
/// Validates that caller-owned Worker settings target the same logical network as the composed Yellowstone/HTTP source.
pub(crate) fn validate_network(&self, network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> {
if self.yellowstone_source.yellowstone_channel.cluster().as_str() != network.as_str() {
return std::result::Result::Err(crate::runtime_error("runtime_resources.worker_network_mismatch"));
}
return std::result::Result::Ok(());
}
}
impl std::fmt::Debug for crate::RawTransactionIngestRuntimeResources {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter.debug_struct("RawTransactionIngestRuntimeResources").field("yellowstone_source", &self.yellowstone_source).finish();
}
}
fn compatible_http_route_count(
pool: &ksp_onchain_transport_lib::HttpTransportPool,
hydration_role: &ksp_onchain_transport_lib::HttpRoleName,
request_kind: &str,
expected_cluster: &str,
) -> ksp_core_lib::Result<usize> {
let snapshot = pool.snapshot();
let mut compatible = 0_usize;
for endpoint in snapshot.endpoints() {
if !endpoint.enabled() {
continue;
}
for role in endpoint.roles() {
if !role.enabled() || role.role() != hydration_role.as_str() {
continue;
}
let supports_request = role.request_kinds().iter().any(|kind| return kind.as_str() == "*" || kind.as_str() == request_kind);
if !supports_request {
continue;
}
if endpoint.cluster() != expected_cluster {
return std::result::Result::Err(crate::runtime_error("runtime_resources.transport_network_mismatch"));
}
compatible = compatible.saturating_add(1);
}
}
return std::result::Result::Ok(compatible);
}
fn ingestion_filter_count(request: &ksp_onchain_transport_lib::YellowstoneSubscribeRequest) -> usize {
return request.transaction_filter_count().saturating_add(request.transaction_status_filter_count()).saturating_add(request.block_filter_count());
}
#[cfg(test)]
#[path = "../unit_tests/runtime_resources.rs"]
mod tests;