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

v0.3.14-pre.008-fix.002

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-12 09:45:58 +02:00
parent ec94001ab8
commit bf394e0e4b
4 changed files with 203 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -158,15 +158,11 @@ impl RawTransactionIngestLiveSource {
(true, false, false, http_block_scan, true)
},
Self::HttpBlockPolling(source) => {
let known_reference_hydration = match http_role_supports_rpc_method(
&source.http_pool,
&source.polling_role,
"getTransaction",
source.network.as_str(),
) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let known_reference_hydration =
match http_role_supports_rpc_method(&source.http_pool, &source.polling_role, "getTransaction", source.network.as_str()) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
(false, true, false, true, known_reference_hydration)
},
Self::StandardBlock(_) => (false, true, false, false, false),
@@ -230,21 +226,13 @@ impl RawTransactionIngestLiveSource {
tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
let mut source_future = std::boxed::Box::pin(async move {
return match self {
Self::HeliusTransaction(source) => {
source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await
},
Self::HeliusTransaction(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await,
Self::HttpBlockPolling(source) => {
source
.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared.continuity_contracts)
.await
source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared.continuity_contracts).await
},
Self::StandardBlock(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender).await,
Self::StandardLogs(source) => {
source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await
},
Self::Yellowstone(source) => {
source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await
},
Self::StandardLogs(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await,
Self::Yellowstone(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await,
};
});
loop {
@@ -825,12 +813,8 @@ impl crate::RawTransactionIngestYellowstoneSource {
processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
shared: RawTransactionIngestSourceRuntimeShared,
) -> ksp_core_lib::Result<()> {
let RawTransactionIngestSourceRuntimeShared {
continuity_contracts,
global_hydration_registry,
hydration_in_flight_limit,
hydration_pending_limit,
} = shared;
let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
shared;
let opened = tokio::select! {
biased;
_ = stop_receiver.changed() => {
@@ -1071,12 +1055,8 @@ impl crate::RawTransactionIngestHeliusTransactionSource {
processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
shared: RawTransactionIngestSourceRuntimeShared,
) -> ksp_core_lib::Result<()> {
let RawTransactionIngestSourceRuntimeShared {
continuity_contracts,
global_hydration_registry,
hydration_in_flight_limit,
hydration_pending_limit,
} = shared;
let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
shared;
let connected = tokio::select! {
biased;
_ = stop_receiver.changed() => {
@@ -1489,9 +1469,7 @@ impl crate::RawTransactionIngestHttpBlockPollingSource {
break 'source;
}
}
if !blocked_by_null
&& let std::option::Option::Some(proven_end_slot) = discovery.proven_end_slot
{
if !blocked_by_null && let std::option::Option::Some(proven_end_slot) = discovery.proven_end_slot {
let coverage_result = {
let mut contracts = match continuity_contracts.lock() {
std::result::Result::Ok(value) => value,
@@ -1882,12 +1860,8 @@ impl crate::RawTransactionIngestStandardLogsSource {
processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
shared: RawTransactionIngestSourceRuntimeShared,
) -> ksp_core_lib::Result<()> {
let RawTransactionIngestSourceRuntimeShared {
continuity_contracts,
global_hydration_registry,
hydration_in_flight_limit,
hydration_pending_limit,
} = shared;
let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
shared;
let connected = tokio::select! {
biased;
_ = stop_receiver.changed() => {
@@ -2336,9 +2310,7 @@ impl crate::RawTransactionIngestRuntimeResources {
};
let source_key = source.source_key();
let _abort_handle = children.spawn(async move {
let result = source
.run(source_settings, source_stop_receiver, source_admission_sender, publisher, source_shared)
.await;
let result = source.run(source_settings, source_stop_receiver, source_admission_sender, publisher, source_shared).await;
return (source_key, result);
});
}
@@ -2564,9 +2536,7 @@ fn source_loss_continuity_range(error: &ksp_core_lib::Error) -> ksp_core_lib::Re
}
return match (start_slot, end_slot) {
(std::option::Option::None, std::option::Option::None) => std::result::Result::Ok(std::option::Option::None),
(std::option::Option::Some(start), std::option::Option::Some(end)) if end >= start => {
std::result::Result::Ok(std::option::Option::Some((start, end)))
},
(std::option::Option::Some(start), std::option::Option::Some(end)) if end >= start => std::result::Result::Ok(std::option::Option::Some((start, end))),
_ => std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")),
};
}
@@ -2584,10 +2554,7 @@ fn source_loss_is_reconcilable(error: &ksp_core_lib::Error) -> bool {
}
return matches!(
context.value(),
"source.configured_source_closed"
| "source.continuity_gap_proven"
| "source.replay_coverage_unproven"
| "source.websocket_incident_unbounded"
"source.configured_source_closed" | "source.continuity_gap_proven" | "source.replay_coverage_unproven" | "source.websocket_incident_unbounded"
);
});
}
@@ -2836,10 +2803,7 @@ fn validate_http_block_discovery_result(
}
let proven_end_slot = discovered.last().copied().map(|slot| return slot.min(end_slot));
let produced_slots = discovered.iter().copied().take_while(|slot| return *slot <= end_slot).collect();
std::result::Result::Ok(RawTransactionIngestHttpDiscoveryWindow {
produced_slots,
proven_end_slot,
})
std::result::Result::Ok(RawTransactionIngestHttpDiscoveryWindow { produced_slots, proven_end_slot })
},
};
}
@@ -3231,9 +3195,7 @@ fn standard_logs_live_source_key(
return hasher.finalize().into();
}
fn yellowstone_coverage_scope_fingerprint(
request: &ksp_onchain_transport_lib::YellowstoneSubscribeRequest,
) -> ksp_core_lib::Result<[u8; 32]> {
fn yellowstone_coverage_scope_fingerprint(request: &ksp_onchain_transport_lib::YellowstoneSubscribeRequest) -> ksp_core_lib::Result<[u8; 32]> {
let mut normalized = request.clone();
normalized.set_commitment(std::option::Option::None);
normalized.set_ping(std::option::Option::None);
@@ -4049,11 +4011,7 @@ impl RawTransactionIngestProcessingFrontierReporter {
}
fn observe_websocket_session_snapshot(&mut self, snapshot: ksp_onchain_transport_lib::WsSessionSnapshot) -> ksp_core_lib::Result<()> {
return self.observe_websocket_continuity(
map_websocket_source_state(snapshot.state()),
snapshot.continuity_gap_count(),
snapshot.overflow_count(),
);
return self.observe_websocket_continuity(map_websocket_source_state(snapshot.state()), snapshot.continuity_gap_count(), snapshot.overflow_count());
}
fn observe_websocket_continuity(
@@ -4094,19 +4052,15 @@ impl RawTransactionIngestProcessingFrontierReporter {
}
},
std::option::Option::None => {
let anchor = match crate::RawTransactionIngestWebSocketIncidentAnchor::new(
start_slot,
reconnect_total,
overflow_total,
saw_reconnect,
saw_overflow,
) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
self.publish();
return std::result::Result::Err(error);
},
};
let anchor =
match crate::RawTransactionIngestWebSocketIncidentAnchor::new(start_slot, reconnect_total, overflow_total, saw_reconnect, saw_overflow)
{
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
self.publish();
return std::result::Result::Err(error);
},
};
self.websocket_incident_anchor = std::option::Option::Some(anchor);
},
}
@@ -4425,13 +4379,7 @@ impl RawTransactionIngestHydrationCoordinator {
self.pending_signal_count -= pending.signals.len();
for pending_signal in pending.signals {
let signal_slot = pending_signal.signal.slot;
let resolution = resolve_known_reference_hydration(
hydration,
settings,
pending_signal.signal,
pending_signal.received_at,
&fetched.observed,
);
let resolution = resolve_known_reference_hydration(hydration, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed);
let resolution = match resolution {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
@@ -4639,16 +4587,9 @@ fn resolve_known_reference_hydration(
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if let std::option::Option::Some(value) = ingress {
return std::result::Result::Ok(RawTransactionIngestKnownReferenceHydrationResolution::Available(std::boxed::Box::new(
value,
)));
return std::result::Result::Ok(RawTransactionIngestKnownReferenceHydrationResolution::Available(std::boxed::Box::new(value)));
}
let block_slot_supported = match http_role_supports_rpc_method(
&hydration.http_pool,
&hydration.hydration_role,
"getBlock",
hydration.network.as_str(),
) {
let block_slot_supported = match http_role_supports_rpc_method(&hydration.http_pool, &hydration.hydration_role, "getBlock", hydration.network.as_str()) {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};

4
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: 29
// version: 30
fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -143,7 +143,7 @@ fn supervisor_contracts(
let (scope, reference_bearing, block_material) = if *family == "full_ledger" {
(crate::RawTransactionIngestCoverageScope::full_ledger_transactions(), false, true)
} else {
(crate::RawTransactionIngestCoverageScope::exact_source_scope(*family, [*fingerprint_byte; 32]), true, false)
(crate::RawTransactionIngestCoverageScope::exact_source_scope(family, [*fingerprint_byte; 32]), true, false)
};
let capability = crate::RawTransactionIngestContinuityCapabilityDescriptor::new(
[*source_key_byte; 32],