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

v0.3.14-pre.008-fix.001

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-12 09:44:36 +02:00
parent 505cc8c3f6
commit ec94001ab8
3 changed files with 277 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
// version: 35
// version: 36
use sha2::Digest; // rust-rules: trait-import
@@ -158,11 +158,15 @@ 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),
@@ -226,13 +230,21 @@ 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 {
@@ -813,8 +825,12 @@ 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() => {
@@ -1055,8 +1071,12 @@ 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() => {
@@ -1469,7 +1489,9 @@ 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,
@@ -1860,8 +1882,12 @@ 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() => {
@@ -2310,7 +2336,9 @@ 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);
});
}
@@ -2454,18 +2482,19 @@ async fn supervise_live_source_tasks(
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await;
}
let (active_source_keys, processing_frontier_slot) = {
let supervisor_state = {
let inventory = match inventory.lock() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(poisoned) => poisoned.into_inner(),
};
match inventory.supervisor_state() {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
},
}
inventory.supervisor_state()
};
let (active_source_keys, processing_frontier_slot) = match supervisor_state {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => {
source_stop_sender.send_replace(true);
return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
},
};
let continuity_range = match source_loss_continuity_range(&source_fault) {
std::result::Result::Ok(value) => value,
@@ -2535,7 +2564,9 @@ 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")),
};
}
@@ -2553,7 +2584,10 @@ 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"
);
});
}
@@ -2802,7 +2836,10 @@ 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,
})
},
};
}
@@ -3194,7 +3231,9 @@ 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);
@@ -4010,7 +4049,11 @@ 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(
@@ -4051,15 +4094,19 @@ 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);
},
}
@@ -4378,7 +4425,13 @@ 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),
@@ -4586,9 +4639,16 @@ 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),
};