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

v0.3.12-pre.006

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-09 13:30:15 +02:00
parent 50c01a19c9
commit 901a0515b7
10 changed files with 1011 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
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: 12
// version: 13
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -11,10 +11,10 @@
//! 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;
//! the first validated Yellowstone/HTTP runtime-resource contract also closes the deterministic hydration
//! qualification path in test configuration. Transaction/TransactionStatus/Block signal projection, BlockMeta/Slot
//! continuity projection and HTTP hydration remain inactive in production until the productive source task is wired;
//! no live stream is opened yet.
//! the validated Yellowstone/HTTP runtime-resource contract now drives one productive supervised source task.
//! Transaction/TransactionStatus/Block updates feed bounded HTTP `getTransaction` hydration and the existing central
//! admission path; BlockMeta/Slot remain continuity-only signals. Hydration is coalesced by network/signature/commitment
//! under bounded in-flight and pending budgets, while reconnect/frontier/replay interpretation remains deferred.
mod admission;
mod error;

12
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: 8
// version: 9
type PersistencePort = std::sync::Arc<dyn crate::RawTransactionIngestPersistencePort + 'static>;
type PersistenceTasks = tokio::task::JoinSet<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>>;
@@ -96,7 +96,7 @@ 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.
/// Starts one Worker with caller-composed runtime resources and one supervised productive Yellowstone source task.
pub fn start_with_runtime_resources(
settings: crate::RawTransactionIngestSettings,
store: std::sync::Arc<ksp_store_lib::Store>,
@@ -113,7 +113,13 @@ impl crate::RawTransactionIngestWorker {
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));
let source = runtime_resources.into_yellowstone_source();
let source_settings = settings.clone();
return start_foundation_with_source_spawner(settings, runtime, std::option::Option::Some(store), move |children, stop_receiver, admission_sender| {
let _abort_handle = children.spawn(async move {
return source.run(source_settings, stop_receiver, admission_sender).await;
});
});
}
}

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

@@ -1,17 +1,12 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
// version: 7
// version: 8
#[cfg(test)]
use sha2::Digest; // rust-rules: trait-import
#[cfg(test)]
const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0";
#[cfg(test)]
const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL: &str = "yellowstone_http";
#[cfg(test)]
const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone_http.source_key.v1\0";
#[cfg(test)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RawTransactionIngestSourceFamily {
Block,
@@ -19,14 +14,12 @@ enum RawTransactionIngestSourceFamily {
TransactionStatus,
}
#[cfg(test)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RawTransactionIngestContinuityFamily {
BlockMeta,
Slot,
}
#[cfg(test)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RawTransactionIngestContinuityStatus {
Completed,
@@ -38,7 +31,6 @@ enum RawTransactionIngestContinuityStatus {
Processed,
}
#[cfg(test)]
#[derive(Clone, Eq, PartialEq)]
struct RawTransactionIngestContinuitySignal {
created_at: std::option::Option<RawTransactionIngestSourceTimestamp>,
@@ -53,7 +45,6 @@ struct RawTransactionIngestContinuitySignal {
status: std::option::Option<RawTransactionIngestContinuityStatus>,
}
#[cfg(test)]
impl std::fmt::Debug for RawTransactionIngestContinuitySignal {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter
@@ -78,21 +69,18 @@ struct RawTransactionIngestSourceRoute {
provider: ksp_store_lib::RawProvenanceCode,
}
#[cfg(test)]
#[derive(Clone, Copy, Eq, PartialEq)]
struct RawTransactionIngestSourceTimestamp {
nanos: u32,
seconds: i64,
}
#[cfg(test)]
impl std::fmt::Debug for RawTransactionIngestSourceTimestamp {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter.debug_struct("RawTransactionIngestSourceTimestamp").field("nanos", &self.nanos).field("seconds", &self.seconds).finish();
}
}
#[cfg(test)]
#[derive(Clone, Eq, PartialEq)]
struct RawTransactionIngestSourceSignal {
created_at: std::option::Option<RawTransactionIngestSourceTimestamp>,
@@ -107,7 +95,6 @@ struct RawTransactionIngestSourceSignal {
transaction_index: std::option::Option<u64>,
}
#[cfg(test)]
impl std::fmt::Debug for RawTransactionIngestSourceSignal {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter
@@ -126,7 +113,6 @@ impl std::fmt::Debug for RawTransactionIngestSourceSignal {
}
}
#[cfg(test)]
trait RawTransactionIngestYellowstoneSignalView {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp>;
@@ -141,7 +127,6 @@ trait RawTransactionIngestYellowstoneSignalView {
fn slot(&self) -> u64;
}
#[cfg(test)]
impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionUpdate {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
return ksp_onchain_transport_lib::YellowstoneTransactionUpdate::created_at(self);
@@ -168,7 +153,6 @@ impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::Ye
}
}
#[cfg(test)]
impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
return ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate::created_at(self);
@@ -195,7 +179,6 @@ impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::Ye
}
}
#[cfg(test)]
trait RawTransactionIngestYellowstoneBlockView {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp>;
@@ -208,7 +191,6 @@ trait RawTransactionIngestYellowstoneBlockView {
fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)>;
}
#[cfg(test)]
impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::YellowstoneBlockUpdate {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
return ksp_onchain_transport_lib::YellowstoneBlockUpdate::created_at(self);
@@ -235,7 +217,6 @@ impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::Yel
}
}
#[cfg(test)]
trait RawTransactionIngestYellowstoneContinuityView {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp>;
@@ -250,7 +231,6 @@ trait RawTransactionIngestYellowstoneContinuityView {
fn status(&self) -> std::option::Option<RawTransactionIngestContinuityStatus>;
}
#[cfg(test)]
impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
return ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate::created_at(self);
@@ -277,7 +257,6 @@ impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib
}
}
#[cfg(test)]
impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib::YellowstoneSlotUpdate {
fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
return ksp_onchain_transport_lib::YellowstoneSlotUpdate::created_at(self);
@@ -304,7 +283,6 @@ impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib
}
}
#[cfg(test)]
impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)>
for RawTransactionIngestSourceSignal
{
@@ -313,7 +291,6 @@ impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onc
}
}
#[cfg(test)]
impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)>
for RawTransactionIngestSourceSignal
{
@@ -324,8 +301,8 @@ impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onc
/// 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.
/// The Transport-owned channel, subscribe request, HTTP pool and hydration role remain private. Construction validates deterministic source-composition
/// invariants without network I/O; runtime execution opens exactly one Transport-owned Yellowstone session.
pub struct RawTransactionIngestYellowstoneSource {
yellowstone_channel: ksp_onchain_transport_lib::YellowstoneGrpcChannel,
subscribe_request: ksp_onchain_transport_lib::YellowstoneSubscribeRequest,
@@ -383,6 +360,92 @@ impl crate::RawTransactionIngestYellowstoneSource {
}
return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, network, route });
}
/// Runs the productive Yellowstone source task until cooperative stop or one safe terminal source failure.
pub(crate) async fn run(
self,
settings: crate::RawTransactionIngestSettings,
mut stop_receiver: tokio::sync::watch::Receiver<bool>,
admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
) -> ksp_core_lib::Result<()> {
let opened = tokio::select! {
biased;
_ = stop_receiver.changed() => {
return std::result::Result::Ok(());
}
result = self.yellowstone_channel.open_standard_subscribe(self.subscribe_request.clone()) => result,
};
let mut session = match opened {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())),
};
let mut coordinator = RawTransactionIngestHydrationCoordinator::new(&settings);
let mut fault = std::option::Option::None;
loop {
if *stop_receiver.borrow() {
break;
}
if let std::result::Result::Err(error) = coordinator.start_hydrations(&self) {
fault = std::option::Option::Some(error);
break;
}
let can_receive = coordinator.can_receive();
if !can_receive && coordinator.tasks.is_empty() {
fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled"));
break;
}
tokio::select! {
biased;
_ = stop_receiver.changed() => {
break;
}
joined = coordinator.tasks.join_next(), if !coordinator.tasks.is_empty() => {
let joined = match joined {
std::option::Option::Some(value) => value,
std::option::Option::None => {
fault = std::option::Option::Some(crate::runtime_error("source.hydration_join_missing"));
break;
},
};
let handled = coordinator.handle_joined(joined, &self, &settings, &admission_sender, &mut stop_receiver).await;
match handled {
std::result::Result::Ok(true) => {},
std::result::Result::Ok(false) => break,
std::result::Result::Err(error) => {
fault = std::option::Option::Some(error);
break;
},
}
}
update = session.next_update(), if can_receive => {
let update = match update {
std::result::Result::Ok(std::option::Option::Some(value)) => value,
std::result::Result::Ok(std::option::Option::None) => {
fault = std::option::Option::Some(crate::runtime_error("source.session_closed"));
break;
},
std::result::Result::Err(error) => {
fault = std::option::Option::Some(source_transport_error(error.code()));
break;
},
};
if let std::result::Result::Err(error) = route_yellowstone_update(&self, &mut coordinator, update) {
fault = std::option::Option::Some(error);
break;
}
}
}
}
coordinator.abort_all().await;
let closed = session.close().await;
if let std::option::Option::Some(error) = fault {
return std::result::Result::Err(error);
}
return match closed {
std::result::Result::Ok(()) => std::result::Result::Ok(()),
std::result::Result::Err(error) => std::result::Result::Err(source_transport_error(error.code())),
};
}
}
impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource {
@@ -416,7 +479,7 @@ pub struct RawTransactionIngestRuntimeResources {
}
impl crate::RawTransactionIngestRuntimeResources {
/// Owns the first validated productive-source contract for a future runtime start.
/// Owns the validated productive Yellowstone source contract consumed by `start_with_runtime_resources`.
#[must_use]
pub fn new(yellowstone_source: crate::RawTransactionIngestYellowstoneSource) -> Self {
return Self { yellowstone_source };
@@ -429,6 +492,11 @@ impl crate::RawTransactionIngestRuntimeResources {
}
return std::result::Result::Ok(());
}
/// Consumes the runtime-resource aggregate into its single validated Yellowstone source for private supervisor wiring.
pub(crate) fn into_yellowstone_source(self) -> crate::RawTransactionIngestYellowstoneSource {
return self.yellowstone_source;
}
}
impl std::fmt::Debug for crate::RawTransactionIngestRuntimeResources {
@@ -497,7 +565,6 @@ fn ingestion_filter_count(request: &ksp_onchain_transport_lib::YellowstoneSubscr
return request.transaction_filter_count().saturating_add(request.transaction_status_filter_count()).saturating_add(request.block_filter_count());
}
#[cfg(test)]
fn matched_filter_direct_id(filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]) -> std::option::Option<ksp_store_lib::RawProvenanceCode> {
let mut names = filters.iter().map(ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::as_str).collect::<std::vec::Vec<_>>();
names.sort_unstable();
@@ -511,7 +578,6 @@ fn matched_filter_direct_id(filters: &[ksp_onchain_transport_lib::YellowstoneSub
};
}
#[cfg(test)]
fn matched_filter_fingerprint(filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]) -> [u8; 32] {
let mut names = filters.iter().map(ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::as_str).collect::<std::vec::Vec<_>>();
names.sort_unstable();
@@ -526,7 +592,6 @@ fn matched_filter_fingerprint(filters: &[ksp_onchain_transport_lib::YellowstoneS
return hasher.finalize().into();
}
#[cfg(test)]
fn project_yellowstone_block_signals<T: RawTransactionIngestYellowstoneBlockView>(
source: &crate::RawTransactionIngestYellowstoneSource,
update: &T,
@@ -558,7 +623,6 @@ fn project_yellowstone_block_signals<T: RawTransactionIngestYellowstoneBlockView
return std::result::Result::Ok(signals);
}
#[cfg(test)]
fn project_yellowstone_continuity_signal<T: RawTransactionIngestYellowstoneContinuityView>(
source: &crate::RawTransactionIngestYellowstoneSource,
update: &T,
@@ -578,7 +642,6 @@ fn project_yellowstone_continuity_signal<T: RawTransactionIngestYellowstoneConti
};
}
#[cfg(test)]
fn map_yellowstone_slot_status(status: ksp_onchain_transport_lib::YellowstoneSlotStatus) -> RawTransactionIngestContinuityStatus {
return match status {
ksp_onchain_transport_lib::YellowstoneSlotStatus::Processed => RawTransactionIngestContinuityStatus::Processed,
@@ -591,7 +654,6 @@ fn map_yellowstone_slot_status(status: ksp_onchain_transport_lib::YellowstoneSlo
};
}
#[cfg(test)]
fn map_yellowstone_source_timestamp(
created_at: std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp>,
) -> std::option::Option<RawTransactionIngestSourceTimestamp> {
@@ -600,7 +662,6 @@ fn map_yellowstone_source_timestamp(
});
}
#[cfg(test)]
fn project_yellowstone_signal<T: RawTransactionIngestYellowstoneSignalView>(
source: &crate::RawTransactionIngestYellowstoneSource,
update: &T,
@@ -623,52 +684,326 @@ fn project_yellowstone_signal<T: RawTransactionIngestYellowstoneSignalView>(
};
}
#[cfg(test)]
enum RawTransactionIngestHydrationOutcome {
Available(std::boxed::Box<crate::RawTransactionIngress>),
Missing(ksp_store_lib::RawTransactionReference),
fn route_yellowstone_update(
source: &crate::RawTransactionIngestYellowstoneSource,
coordinator: &mut RawTransactionIngestHydrationCoordinator,
update: ksp_onchain_transport_lib::YellowstoneSubscribeUpdate,
) -> ksp_core_lib::Result<()> {
match update {
ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Transaction(value) => {
let received_at = match current_raw_timestamp() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, value.as_ref())), received_at);
},
ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::TransactionStatus(value) => {
let received_at = match current_raw_timestamp() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, &value)), received_at);
},
ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(value) => {
let received_at = match current_raw_timestamp() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let signals = match project_yellowstone_block_signals(source, &value) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
for signal in signals {
if let std::result::Result::Err(error) = coordinator.queue_signal(source, signal, received_at) {
return std::result::Result::Err(error);
}
}
return std::result::Result::Ok(());
},
ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::BlockMeta(value) => {
let _continuity_signal = project_yellowstone_continuity_signal(source, &value);
return std::result::Result::Ok(());
},
ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Slot(value) => {
let _continuity_signal = project_yellowstone_continuity_signal(source, &value);
return std::result::Result::Ok(());
},
ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Account(_)
| ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Ping(_)
| ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Pong(_)
| ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Entry(_) => std::result::Result::Ok(()),
}
}
#[cfg(test)]
async fn hydrate_yellowstone_signal(
fn current_raw_timestamp() -> ksp_core_lib::Result<ksp_store_lib::RawTimestamp> {
let duration = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
let duration = match duration {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.clock_before_epoch")),
};
let millis = match u64::try_from(duration.as_millis()) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.clock_unrepresentable")),
};
return match ksp_store_lib::RawTimestamp::from_unix_millis(millis) {
std::result::Result::Ok(value) => std::result::Result::Ok(value),
std::result::Result::Err(_) => std::result::Result::Err(crate::runtime_error("source.clock_out_of_bounds")),
};
}
fn source_transport_error(code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Error {
return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED, "RAW transaction ingest Yellowstone source transport failed")
.with_context("transport_domain", code.domain())
.with_context("transport_code", code.code());
}
#[derive(Clone, Eq, Ord, PartialEq, PartialOrd)]
struct RawTransactionIngestHydrationKey {
commitment: &'static str,
network: ksp_store_lib::RawNetworkId,
signature: ksp_store_lib::RawTransactionSignature,
}
struct RawTransactionIngestPendingSignal {
received_at: ksp_store_lib::RawTimestamp,
signal: RawTransactionIngestSourceSignal,
}
struct RawTransactionIngestPendingHydration {
in_flight: bool,
signals: std::vec::Vec<RawTransactionIngestPendingSignal>,
}
type RawTransactionIngestObservedTransaction =
ksp_onchain_transport_lib::HttpObservedValue<std::option::Option<ksp_onchain_transport_lib::SolanaConfirmedTransaction>>;
struct RawTransactionIngestHydrationFetch {
key: RawTransactionIngestHydrationKey,
observed: RawTransactionIngestObservedTransaction,
}
type RawTransactionIngestHydrationTasks = tokio::task::JoinSet<ksp_core_lib::Result<RawTransactionIngestHydrationFetch>>;
struct RawTransactionIngestHydrationCoordinator {
max_in_flight: usize,
max_pending_signals: usize,
pending_signal_count: usize,
pending: std::collections::BTreeMap<RawTransactionIngestHydrationKey, RawTransactionIngestPendingHydration>,
tasks: RawTransactionIngestHydrationTasks,
}
impl RawTransactionIngestHydrationCoordinator {
fn new(settings: &crate::RawTransactionIngestSettings) -> Self {
return Self {
max_in_flight: settings.persistence_concurrency(),
max_pending_signals: crate::MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY,
pending_signal_count: 0,
pending: std::collections::BTreeMap::new(),
tasks: RawTransactionIngestHydrationTasks::new(),
};
}
fn can_receive(&self) -> bool {
return self.pending_signal_count < self.max_pending_signals;
}
fn queue_signal(
&mut self,
source: &crate::RawTransactionIngestYellowstoneSource,
signal: RawTransactionIngestSourceSignal,
received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<()> {
if self.pending_signal_count >= self.max_pending_signals {
return std::result::Result::Err(crate::runtime_error("source.hydration_pending_saturated"));
}
let key = match hydration_key(source, &signal) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
match self.pending.entry(key) {
std::collections::btree_map::Entry::Occupied(mut entry) => {
entry.get_mut().signals.push(RawTransactionIngestPendingSignal { received_at, signal });
},
std::collections::btree_map::Entry::Vacant(entry) => {
entry.insert(RawTransactionIngestPendingHydration {
in_flight: false,
signals: std::vec![RawTransactionIngestPendingSignal { received_at, signal }],
});
},
}
self.pending_signal_count = match self.pending_signal_count.checked_add(1) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_pending_counter_exhausted")),
};
return self.start_hydrations(source);
}
fn start_hydrations(&mut self, source: &crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result<()> {
while self.tasks.len() < self.max_in_flight {
let key = self
.pending
.iter()
.find_map(|(key, pending)| if pending.in_flight { std::option::Option::None } else { std::option::Option::Some(key.clone()) });
let key = match key {
std::option::Option::Some(value) => value,
std::option::Option::None => break,
};
let commitment = match hydration_commitment(source) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let pending = match self.pending.get_mut(&key) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_pending_missing")),
};
pending.in_flight = true;
let pool = source.http_pool.clone();
let role = source.hydration_role.clone();
let expected_network = source.network.clone();
let task_key = key.clone();
let _abort_handle = self.tasks.spawn(async move {
return fetch_yellowstone_hydration(pool, role, expected_network, task_key, commitment).await;
});
}
return std::result::Result::Ok(());
}
async fn handle_joined(
&mut self,
joined: std::result::Result<ksp_core_lib::Result<RawTransactionIngestHydrationFetch>, tokio::task::JoinError>,
source: &crate::RawTransactionIngestYellowstoneSource,
settings: &crate::RawTransactionIngestSettings,
admission_sender: &tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
stop_receiver: &mut tokio::sync::watch::Receiver<bool>,
) -> ksp_core_lib::Result<bool> {
let fetched = match joined {
std::result::Result::Ok(std::result::Result::Ok(value)) => value,
std::result::Result::Ok(std::result::Result::Err(error)) => return std::result::Result::Err(error),
std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.hydration_task_join_failed")),
};
let pending = match self.pending.remove(&fetched.key) {
std::option::Option::Some(value) => value,
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_result_without_pending")),
};
if pending.signals.len() > self.pending_signal_count {
return std::result::Result::Err(crate::runtime_error("source.hydration_pending_counter_invalid"));
}
self.pending_signal_count -= pending.signals.len();
for pending_signal in pending.signals {
let ingress = finalize_yellowstone_hydration(source, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed);
let ingress = match ingress {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let ingress = match ingress {
std::option::Option::Some(value) => value,
std::option::Option::None => continue,
};
let sent = tokio::select! {
biased;
_ = stop_receiver.changed() => {
return std::result::Result::Ok(false);
}
result = admission_sender.send(ingress) => result,
};
if sent.is_err() {
if *stop_receiver.borrow() {
return std::result::Result::Ok(false);
}
return std::result::Result::Err(crate::runtime_error("source.admission_closed"));
}
}
if let std::result::Result::Err(error) = self.start_hydrations(source) {
return std::result::Result::Err(error);
}
return std::result::Result::Ok(true);
}
async fn abort_all(&mut self) {
self.tasks.abort_all();
while self.tasks.join_next().await.is_some() {}
self.pending.clear();
self.pending_signal_count = 0;
return;
}
}
fn hydration_commitment(source: &crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result<ksp_onchain_transport_lib::SolanaCommitment> {
return match source.subscribe_request.commitment() {
std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) => {
std::result::Result::Ok(ksp_onchain_transport_lib::SolanaCommitment::Confirmed)
},
std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => {
std::result::Result::Ok(ksp_onchain_transport_lib::SolanaCommitment::Finalized)
},
std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed) | std::option::Option::None => {
std::result::Result::Err(crate::runtime_error("hydration.commitment_invalid"))
},
};
}
fn hydration_key(
source: &crate::RawTransactionIngestYellowstoneSource,
signal: &RawTransactionIngestSourceSignal,
) -> ksp_core_lib::Result<RawTransactionIngestHydrationKey> {
let commitment = match hydration_commitment(source) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
return std::result::Result::Ok(RawTransactionIngestHydrationKey {
commitment: commitment.as_str(),
network: signal.network.clone(),
signature: signal.signature,
});
}
async fn fetch_yellowstone_hydration(
http_pool: ksp_onchain_transport_lib::HttpTransportPool,
hydration_role: ksp_onchain_transport_lib::HttpRoleName,
expected_network: ksp_store_lib::RawNetworkId,
key: RawTransactionIngestHydrationKey,
commitment: ksp_onchain_transport_lib::SolanaCommitment,
) -> ksp_core_lib::Result<RawTransactionIngestHydrationFetch> {
if key.network != expected_network {
return std::result::Result::Err(crate::runtime_error("hydration.network_key_mismatch"));
}
if key.commitment != commitment.as_str() {
return std::result::Result::Err(crate::runtime_error("hydration.commitment_key_mismatch"));
}
let config = ksp_onchain_transport_lib::SolanaGetTransactionConfig::new(
std::option::Option::Some(commitment),
std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64),
std::option::Option::Some(0),
);
let signature_text = ksp_raw_transaction_lib::format_raw_transaction_signature(&key.signature);
let observed = http_pool.get_transaction_observed(&hydration_role, signature_text.as_str(), std::option::Option::Some(&config)).await;
let observed = match observed {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(hydration_transport_error(error.code())),
};
return std::result::Result::Ok(RawTransactionIngestHydrationFetch { key, observed });
}
fn finalize_yellowstone_hydration(
source: &crate::RawTransactionIngestYellowstoneSource,
settings: &crate::RawTransactionIngestSettings,
signal: RawTransactionIngestSourceSignal,
received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<RawTransactionIngestHydrationOutcome> {
observed: &RawTransactionIngestObservedTransaction,
) -> ksp_core_lib::Result<std::option::Option<crate::RawTransactionIngress>> {
if &signal.network != settings.network() || signal.network != source.network {
return std::result::Result::Err(crate::runtime_error("hydration.network_mismatch"));
}
if signal.route != source.route {
return std::result::Result::Err(crate::runtime_error("hydration.source_route_mismatch"));
}
let commitment = match source.subscribe_request.commitment() {
std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) => ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => 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("hydration.commitment_invalid"));
},
};
let config = ksp_onchain_transport_lib::SolanaGetTransactionConfig::new(
std::option::Option::Some(commitment),
std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64),
std::option::Option::Some(0),
);
let signature_text = ksp_raw_transaction_lib::format_raw_transaction_signature(&signal.signature);
let observed = source.http_pool.get_transaction_observed(&source.hydration_role, signature_text.as_str(), std::option::Option::Some(&config)).await;
let observed = match observed {
let commitment = match hydration_commitment(source) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(hydration_transport_error(error.code())),
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let http_provider = observed.provider().as_str().to_owned();
let http_endpoint = observed.endpoint_name().to_owned();
let transaction = observed.into_value();
let transaction = match transaction {
let transaction = match observed.value().as_ref() {
std::option::Option::Some(value) => value,
std::option::Option::None => {
let reference = ksp_store_lib::RawTransactionReference::new(signal.network, signal.signature);
return std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Missing(reference));
},
std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
};
if transaction.slot() != signal.slot {
return std::result::Result::Err(crate::runtime_error("hydration.slot_mismatch"));
@@ -699,7 +1034,7 @@ async fn hydrate_yellowstone_signal(
if embedded_signature != signal.signature {
return std::result::Result::Err(crate::runtime_error("hydration.signature_mismatch"));
}
let provenance = build_hydration_provenance(settings, &signal, http_provider.as_str(), http_endpoint.as_str(), commitment, received_at);
let provenance = build_hydration_provenance(settings, &signal, observed.provider().as_str(), observed.endpoint_name(), commitment, received_at);
let provenance = match provenance {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
@@ -720,15 +1055,46 @@ async fn hydrate_yellowstone_signal(
}),
map_hydration_wire_field(transaction.transaction_index(), |value| return *value),
);
return std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Available(std::boxed::Box::new(crate::RawTransactionIngress {
material,
network: signal.network,
provenance,
source_key,
})));
return std::result::Result::Ok(std::option::Option::Some(crate::RawTransactionIngress { material, network: signal.network, provenance, source_key }));
}
#[cfg(test)]
enum RawTransactionIngestHydrationOutcome {
Available(std::boxed::Box<crate::RawTransactionIngress>),
Missing(ksp_store_lib::RawTransactionReference),
}
#[cfg(test)]
async fn hydrate_yellowstone_signal(
source: &crate::RawTransactionIngestYellowstoneSource,
settings: &crate::RawTransactionIngestSettings,
signal: RawTransactionIngestSourceSignal,
received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<RawTransactionIngestHydrationOutcome> {
let reference = ksp_store_lib::RawTransactionReference::new(signal.network.clone(), signal.signature);
let key = match hydration_key(source, &signal) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let commitment = match hydration_commitment(source) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let fetched = fetch_yellowstone_hydration(source.http_pool.clone(), source.hydration_role.clone(), source.network.clone(), key, commitment).await;
let fetched = match fetched {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let ingress = finalize_yellowstone_hydration(source, settings, signal, received_at, &fetched.observed);
return match ingress {
std::result::Result::Ok(std::option::Option::Some(value)) => {
std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Available(std::boxed::Box::new(value)))
},
std::result::Result::Ok(std::option::Option::None) => std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Missing(reference)),
std::result::Result::Err(error) => std::result::Result::Err(error),
};
}
fn build_hydration_provenance(
settings: &crate::RawTransactionIngestSettings,
signal: &RawTransactionIngestSourceSignal,
@@ -779,7 +1145,6 @@ fn build_hydration_provenance(
return std::result::Result::Ok(provenance);
}
#[cfg(test)]
fn hydration_method_code(family: RawTransactionIngestSourceFamily) -> &'static str {
return match family {
RawTransactionIngestSourceFamily::Block => "block_get_transaction",
@@ -788,7 +1153,6 @@ fn hydration_method_code(family: RawTransactionIngestSourceFamily) -> &'static s
};
}
#[cfg(test)]
fn fingerprint_filter_code(fingerprint: &[u8; 32]) -> ksp_core_lib::Result<ksp_store_lib::RawProvenanceCode> {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut value = std::string::String::with_capacity(71);
@@ -803,7 +1167,6 @@ fn fingerprint_filter_code(fingerprint: &[u8; 32]) -> ksp_core_lib::Result<ksp_s
};
}
#[cfg(test)]
fn representable_observed_at(
created_at: std::option::Option<RawTransactionIngestSourceTimestamp>,
received_at: ksp_store_lib::RawTimestamp,
@@ -830,7 +1193,6 @@ fn representable_observed_at(
return std::option::Option::Some(observed_at);
}
#[cfg(test)]
fn hydration_source_key(provenance: &ksp_store_lib::RawAcquisitionProvenance) -> [u8; 32] {
let mut hasher = sha2::Sha256::new();
hasher.update(RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN);
@@ -852,21 +1214,18 @@ fn hydration_source_key(provenance: &ksp_store_lib::RawAcquisitionProvenance) ->
return hasher.finalize().into();
}
#[cfg(test)]
fn hash_hydration_source_key(hasher: &mut sha2::Sha256, value: &[u8]) {
hasher.update((value.len() as u64).to_be_bytes());
hasher.update(value);
return;
}
#[cfg(test)]
fn hydration_transport_error(code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Error {
return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED, "RAW transaction ingest hydration transport failed")
.with_context("transport_domain", code.domain())
.with_context("transport_code", code.code());
}
#[cfg(test)]
fn map_hydration_wire_field<T, U, F>(
field: &ksp_onchain_transport_lib::SolanaWireField<T>,
mut map_value: F,

56
crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs
// version: 12
// version: 13
//! Dependency firewall canaries for the RAW transaction ingest Worker foundation.
@@ -64,7 +64,7 @@ fn v0_3_12_pre_002_manifest_opens_only_the_onchain_transport_live_source_edge()
}
#[test]
fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_offline_and_transport_facade_only() {
fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_and_transport_facade_only() {
let root = include_str!("../src/lib.rs");
let resources = include_str!("../src/runtime_resources.rs");
for required in [
@@ -85,8 +85,6 @@ fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_offline_and_trans
"pub struct RawTransactionIngestSourceSignal",
"pub(crate) struct RawTransactionIngestSourceSignal",
"pub use self::runtime_resources::RawTransactionIngestSourceSignal",
"open_standard_subscribe",
"next_update",
"get_block_observed",
"ksp_store_postgres_lib::",
"reqwest::",
@@ -127,10 +125,9 @@ fn v0_3_12_pre_004_hydration_contract_uses_only_transport_facade_common_raw_and_
] {
assert!(resources.contains(required), "required pre.004 hydration contract missing: {required}");
}
assert!(resources.contains("#[cfg(test)]\nasync fn hydrate_yellowstone_signal"));
assert!(resources.contains("fn finalize_yellowstone_hydration"));
assert!(resources.contains("fn fetch_yellowstone_hydration"));
for forbidden in [
"open_standard_subscribe",
"next_update",
"get_block_observed",
"ksp_config_lib::",
"ksp_job_backfill_lib::",
@@ -145,7 +142,7 @@ fn v0_3_12_pre_004_hydration_contract_uses_only_transport_facade_common_raw_and_
}
#[test]
fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_offline_and_transport_facade_only() {
fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_facade_only() {
let root = include_str!("../src/lib.rs");
let resources = include_str!("../src/runtime_resources.rs");
for required in [
@@ -161,8 +158,6 @@ fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_offline_and_tran
assert!(resources.contains(required), "required pre.005 private adapter contract missing: {required}");
}
for forbidden in [
"open_standard_subscribe",
"next_update",
"get_block_observed",
"ksp_config_lib::",
"ksp_job_backfill_lib::",
@@ -177,7 +172,46 @@ fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_offline_and_tran
}
#[test]
fn v0_3_12_pre_002_source_surface_hardens_shutdown_and_faults_without_backend_or_premature_live_io() {
fn v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_and_existing_admission() {
let runtime = include_str!("../src/runtime.rs");
let resources = include_str!("../src/runtime_resources.rs");
for required in ["runtime_resources.into_yellowstone_source()", "children.spawn", "source.run(source_settings, stop_receiver, admission_sender)"] {
assert!(runtime.contains(required), "required pre.006 supervisor wiring missing: {required}");
}
for required in [
"open_standard_subscribe",
"next_update",
"RawTransactionIngestHydrationCoordinator",
"std::collections::BTreeMap",
"tokio::task::JoinSet",
"MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY",
"get_transaction_observed",
"admission_sender.send(ingress)",
"tasks.abort_all()",
"session.close().await",
"commitment: commitment.as_str()",
"network: signal.network.clone()",
"signature: signal.signature",
] {
assert!(resources.contains(required), "required pre.006 productive-source contract missing: {required}");
}
for forbidden in [
"get_block_observed",
"ksp_config_lib::",
"ksp_job_backfill_lib::",
"ksp_store_postgres_lib::",
"reqwest::",
"tonic::",
"yellowstone_grpc_proto::",
"unbounded_channel",
] {
assert!(!runtime.contains(forbidden) && !resources.contains(forbidden), "pre.006 crossed a forbidden productive-source boundary: {forbidden}");
}
return;
}
#[test]
fn v0_3_12_pre_006_source_surface_hardens_shutdown_and_faults_without_backend_edges() {
let root = include_str!("../src/lib.rs");
let runtime = include_str!("../src/runtime.rs");
let admission = include_str!("../src/admission.rs");

38
crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs
// version: 7
// version: 8
//! External public, security, redaction and release-boundary hardening canaries for `pre.010`.
@@ -382,17 +382,39 @@ fn v0_3_12_pre_005_block_and_continuity_shapes_are_bounded_redacted_and_raw_sepa
}
#[test]
fn v0_3_12_pre_002_runtime_resource_contract_performs_no_live_io_or_source_spawn() {
fn v0_3_12_pre_006_runtime_resource_contract_opens_one_supervised_transport_source() {
let resources = include_str!("../src/runtime_resources.rs");
let runtime = include_str!("../src/runtime.rs");
for forbidden in ["open_standard_subscribe", "next_update", "get_block_observed", "tokio::spawn", "JoinSet"] {
assert!(!resources.contains(forbidden), "runtime-resource contract opened premature live source behavior: {forbidden}");
for required in ["open_standard_subscribe", "next_update", "get_transaction_observed", "RawTransactionIngestHydrationCoordinator", "session.close().await"]
{
assert!(resources.contains(required), "productive runtime-resource source behavior missing: {required}");
}
assert!(resources.contains("#[cfg(test)]\nasync fn hydrate_yellowstone_signal"));
assert!(resources.contains("get_transaction_observed"));
assert!(runtime.contains("start_with_runtime_resources"));
for forbidden in ["open_standard_subscribe", "next_update", "get_transaction_observed", "get_block_observed"] {
assert!(!runtime.contains(forbidden), "pre.002 runtime start opened premature live behavior: {forbidden}");
assert!(runtime.contains("source.run(source_settings, stop_receiver, admission_sender)"));
for forbidden in ["get_block_observed", "ksp_config_lib::", "ksp_store_postgres_lib::", "reqwest::", "tonic::", "yellowstone_grpc_proto::"] {
assert!(!resources.contains(forbidden) && !runtime.contains(forbidden), "pre.006 runtime source crossed a forbidden boundary: {forbidden}");
}
return;
}
#[test]
fn v0_3_12_pre_006_source_coalescence_is_bounded_stop_preemptible_and_redacted() {
let resources = include_str!("../src/runtime_resources.rs");
for required in [
"max_in_flight: settings.persistence_concurrency()",
"max_pending_signals: crate::MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY",
"pending_signal_count",
"stop_receiver.changed()",
"tasks.abort_all()",
"source.hydration_pending_saturated",
"source.hydration_task_join_failed",
"hydration_transport_error(error.code())",
"source_transport_error(error.code())",
] {
assert!(resources.contains(required), "required pre.006 bounded/stop-safe source guard missing: {required}");
}
for forbidden in ["remote_message", "response_body", "dead_error", "TransactionStatus.error", ".error()", "Authorization", "Bearer "] {
assert!(!resources.contains(forbidden), "remote/secret material leaked into productive source: {forbidden}");
}
return;
}

12
crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs
// version: 5
// version: 6
//! Release-completeness canaries through the `pre.010` public/release/security hardening tranche.
@@ -104,7 +104,8 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() {
"pre_010_source_visibility_contract_uses_crate_root_for_shared_items",
"pre_010_production_surface_has_no_historical_backfill_or_retriever_contract",
"v0_3_12_pre_002_production_sources_keep_transport_confined_to_runtime_resources",
"v0_3_12_pre_002_runtime_resource_contract_performs_no_live_io_or_source_spawn",
"v0_3_12_pre_006_runtime_resource_contract_opens_one_supervised_transport_source",
"v0_3_12_pre_006_source_coalescence_is_bounded_stop_preemptible_and_redacted",
"pre_010_lower_layers_have_no_dependency_return_to_concrete_worker",
"v0_3_12_pre_002_public_root_exposes_contract_types_without_transport_implementation_paths",
] {
@@ -112,11 +113,12 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() {
}
let dependency_boundary = include_str!("dependency_boundary.rs");
assert!(dependency_boundary.contains("pre_002_manifest_dependency_surface_is_exact"));
assert!(dependency_boundary.contains("v0_3_12_pre_002_source_surface_hardens_shutdown_and_faults_without_backend_or_premature_live_io"));
assert!(dependency_boundary.contains("v0_3_12_pre_006_source_surface_hardens_shutdown_and_faults_without_backend_edges"));
assert!(dependency_boundary.contains("v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_and_existing_admission"));
assert!(dependency_boundary.contains("v0_3_12_pre_002_manifest_opens_only_the_onchain_transport_live_source_edge"));
assert!(dependency_boundary.contains("v0_3_12_pre_003_transaction_and_status_adapters_are_private_offline_and_transport_facade_only"));
assert!(dependency_boundary.contains("v0_3_12_pre_003_transaction_and_status_adapters_are_private_and_transport_facade_only"));
assert!(dependency_boundary.contains("v0_3_12_pre_004_hydration_contract_uses_only_transport_facade_common_raw_and_existing_ingress"));
assert!(dependency_boundary.contains("v0_3_12_pre_005_block_and_continuity_adapters_remain_private_offline_and_transport_facade_only"));
assert!(dependency_boundary.contains("v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_facade_only"));
let public_api = include_str!("public_api.rs");
assert!(public_api.contains("pre_003_kind_code_and_settings_are_consumable_from_crate_root"));
assert!(public_api.contains("pre_004_start_handle_and_terminal_future_are_consumable_without_public_join_handle"));

78
crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs
// version: 5
// version: 6
fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -1245,3 +1245,79 @@ async fn pre_004_status_provenance_and_future_source_timestamp_are_bounded_and_r
assert!(!debug.contains(PRE_004_ZERO_SIGNATURE_TEXT));
return;
}
#[tokio::test(flavor = "current_thread")]
async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanout() {
let source = match signal_source() {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let settings = match pre_004_settings() {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let transaction = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::Transaction, 91, 3, &["transaction-filter"], 0x44) {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let status = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::TransactionStatus, 91, 3, &["status-filter"], 0x44) {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let transaction_key = match super::hydration_key(&source, &transaction) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return,
};
let status_key = match super::hydration_key(&source, &status) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return,
};
assert!(transaction_key == status_key);
let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_760_000_200_000) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return,
};
let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings);
if coordinator.queue_signal(&source, transaction, received_at).is_err() {
return;
}
if coordinator.queue_signal(&source, status, received_at).is_err() {
return;
}
assert_eq!(coordinator.pending.len(), 1);
assert_eq!(coordinator.pending_signal_count, 2);
assert_eq!(coordinator.tasks.len(), 1);
coordinator.abort_all().await;
assert_eq!(coordinator.pending_signal_count, 0);
assert!(coordinator.pending.is_empty());
assert!(coordinator.tasks.is_empty());
return;
}
#[test]
fn pre_006_coalescence_key_separates_network_signature_and_commitment() {
let source = match signal_source() {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let first = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::Transaction, 12, 1, &["same"], 0x10) {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let second = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::Block, 12, 1, &["same"], 0x11) {
std::option::Option::Some(value) => value,
std::option::Option::None => return,
};
let first_key = match super::hydration_key(&source, &first) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return,
};
let second_key = match super::hydration_key(&source, &second) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return,
};
assert!(first_key != second_key);
assert_eq!(first_key.commitment, "confirmed");
assert_eq!(first_key.network.as_str(), "devnet");
return;
}