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

v0.3.10-pre.003

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-06 10:41:47 +02:00
parent aabe8dedb2
commit 8309d829cc
12 changed files with 546 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

327
crates/ksp-job-backfill-lib/src/conversion.rs
View File

@@ -1,18 +1,13 @@
// file: crates/ksp-job-backfill-lib/src/conversion.rs
// version: 4
// version: 5
use sha2::Digest; // rust-rules: trait-import
/// KSP-owned source-independent RAW transaction format identifier produced by this Backfill vertical.
pub const RAW_TRANSACTION_FORMAT_ID: &str = "ksp.solana.raw_transaction";
/// Initial KSP-owned RAW transaction format version produced by this Backfill vertical.
pub const RAW_TRANSACTION_FORMAT_VERSION: u32 = 1;
const RAW_TRANSACTION_METHOD_CODE: &str = "getTransaction";
const RAW_TRANSACTION_OBSERVATION_CONTRACT_VERSION: u32 = 1;
const RAW_TRANSACTION_PROTOCOL_CODE: &str = "solana.http.json_rpc";
/// Complete in-memory RAW transaction acquisition ready for the later Store persistence tranche.
/// Complete in-memory RAW transaction acquisition ready for Store persistence.
#[derive(Debug)]
pub struct BackfillRawAcquisition {
inner: Box<BackfillRawAcquisitionInner>,
@@ -20,35 +15,34 @@ pub struct BackfillRawAcquisition {
#[derive(Debug)]
struct BackfillRawAcquisitionInner {
transaction: ksp_store_lib::RawTransaction,
observation: ksp_store_lib::RawTransactionObservation,
acquisition: ksp_raw_transaction_lib::RawTransactionAcquisition,
}
impl crate::BackfillRawAcquisition {
/// Returns the canonical RAW transaction produced from the typed Transport response.
/// Returns the canonical RAW transaction produced through the source-neutral common layer.
#[must_use]
pub const fn transaction(&self) -> &ksp_store_lib::RawTransaction {
return &self.inner.transaction;
pub fn transaction(&self) -> &ksp_store_lib::RawTransaction {
return self.inner.acquisition.transaction();
}
/// Returns the acquisition observation whose provenance records the actual successful endpoint.
/// Returns the Backfill-owned acquisition observation linked by the common assembly layer.
#[must_use]
pub const fn observation(&self) -> &ksp_store_lib::RawTransactionObservation {
return &self.inner.observation;
pub fn observation(&self) -> &ksp_store_lib::RawTransactionObservation {
return self.inner.acquisition.observation();
}
/// Consumes the in-memory acquisition into the canonical transaction and its observation.
/// Consumes the in-memory acquisition into the canonical transaction and its Backfill-owned observation.
#[must_use]
pub fn into_parts(self) -> (ksp_store_lib::RawTransaction, ksp_store_lib::RawTransactionObservation) {
let inner = *self.inner;
return (inner.transaction, inner.observation);
return inner.acquisition.into_parts();
}
}
/// Result of hydrating one deterministic Backfill candidate through observed `getTransaction`.
#[derive(Debug)]
pub enum BackfillHydrationOutcome {
/// The RPC returned one complete transaction and conversion produced canonical RAW plus provenance.
/// The RPC returned one complete transaction and common conversion produced canonical RAW plus Backfill provenance.
Available(crate::BackfillRawAcquisition),
/// The RPC returned JSON `null`; only the canonical transaction identity exists and no provenance is fabricated.
Missing(ksp_store_lib::RawTransactionReference),
@@ -71,11 +65,12 @@ impl crate::BackfillHydrationOutcome {
}
}
/// Hydrates one candidate with the typed observed Transport path and converts it to canonical RAW v1.
/// Hydrates one candidate with the typed observed Transport path and converts it through the common RAW v1 layer.
///
/// The caller supplies the local receipt timestamp because wall-clock ownership remains outside this
/// pure conversion tranche. Transport retains endpoint selection and retry. This function never
/// persists to Store; persistence begins in `pre.007`.
/// pure conversion path. Transport retains endpoint selection and retry. Backfill retains campaign
/// provenance and observation-key ownership; the common crate owns signature parsing, RAW v1
/// canonicalization, content hashing and transaction/observation assembly.
pub async fn hydrate_backfill_candidate(
transport: &ksp_onchain_transport_lib::HttpTransportPool,
request: &crate::BackfillRequest,
@@ -105,12 +100,12 @@ pub async fn hydrate_backfill_candidate(
std::option::Option::None => return std::result::Result::Ok(crate::BackfillHydrationOutcome::Missing(reference)),
};
let fields = CanonicalTransactionFields {
slot: transaction.slot(),
block_time: transaction.block_time(),
transaction: transaction.transaction(),
meta: transaction.meta(),
version: transaction.version(),
slot: transaction.slot(),
transaction: transaction.transaction(),
transaction_index: transaction.transaction_index(),
version: transaction.version(),
};
let acquisition = convert_available_fields(request, reference, fields, provider.as_str(), endpoint.as_str(), received_at);
return match acquisition {
@@ -119,49 +114,22 @@ pub async fn hydrate_backfill_candidate(
};
}
/// Decodes one validated Base58 signature to exactly 64 canonical bytes without a Solana SDK dependency.
/// Decodes one validated Base58 signature through the source-neutral common parser while preserving Backfill error semantics.
pub(crate) fn decode_backfill_signature(signature: &crate::BackfillSignature) -> ksp_core_lib::Result<ksp_store_lib::RawTransactionSignature> {
let text = signature.as_str().as_bytes();
let mut decoded = [0_u8; 64];
let mut leading_zeroes = 0_usize;
for byte in text {
if *byte != b'1' {
break;
}
leading_zeroes += 1;
}
for byte in text {
let digit = match base58_digit(*byte) {
std::option::Option::Some(digit) => digit,
std::option::Option::None => return std::result::Result::Err(conversion_error("signature")),
};
let mut carry = u32::from(digit);
for output in decoded.iter_mut().rev() {
let value = (u32::from(*output) * 58) + carry;
*output = (value & 0xff) as u8;
carry = value >> 8;
}
if carry != 0 {
return std::result::Result::Err(conversion_error("signature"));
}
}
let significant_len = match decoded.iter().position(|byte| return *byte != 0) {
std::option::Option::Some(index) => decoded.len() - index,
std::option::Option::None => 0,
let parsed = ksp_raw_transaction_lib::parse_raw_transaction_signature(signature.as_str());
return match parsed {
std::result::Result::Ok(signature) => std::result::Result::Ok(signature),
std::result::Result::Err(_) => std::result::Result::Err(conversion_error("signature")),
};
if leading_zeroes + significant_len != decoded.len() {
return std::result::Result::Err(conversion_error("signature"));
}
return std::result::Result::Ok(ksp_store_lib::RawTransactionSignature::new(decoded));
}
struct CanonicalTransactionFields<'a> {
slot: u64,
block_time: std::option::Option<i64>,
transaction: &'a ksp_onchain_transport_lib::SolanaEncodedTransaction,
meta: &'a ksp_onchain_transport_lib::SolanaWireField<serde_json::Value>,
version: &'a ksp_onchain_transport_lib::SolanaWireField<ksp_onchain_transport_lib::SolanaTransactionVersion>,
slot: u64,
transaction: &'a ksp_onchain_transport_lib::SolanaEncodedTransaction,
transaction_index: &'a ksp_onchain_transport_lib::SolanaWireField<u32>,
version: &'a ksp_onchain_transport_lib::SolanaWireField<ksp_onchain_transport_lib::SolanaTransactionVersion>,
}
fn canonical_reference(request: &crate::BackfillRequest, candidate: &crate::BackfillCandidate) -> ksp_core_lib::Result<ksp_store_lib::RawTransactionReference> {
@@ -184,30 +152,9 @@ fn convert_available_fields(
endpoint: &str,
received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<crate::BackfillRawAcquisition> {
let block_time = convert_block_time(fields.block_time);
let block_time = match block_time {
std::result::Result::Ok(block_time) => block_time,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let bytes = canonical_payload_bytes(&fields);
let bytes = match bytes {
std::result::Result::Ok(bytes) => bytes,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let hash: [u8; 32] = sha2::Sha256::digest(bytes.as_slice()).into();
let format_id = ksp_store_lib::RawFormatId::new(crate::RAW_TRANSACTION_FORMAT_ID);
let format_id = match format_id {
std::result::Result::Ok(format_id) => format_id,
std::result::Result::Err(_) => return std::result::Result::Err(conversion_error("payload.format_id")),
};
let payload = ksp_store_lib::RawPayload::try_new(
format_id,
crate::RAW_TRANSACTION_FORMAT_VERSION,
bytes.into_boxed_slice(),
ksp_store_lib::RawContentHash::new(hash),
);
let payload = match payload {
std::result::Result::Ok(payload) => payload,
let transaction = canonical_transaction(&reference, fields);
let transaction = match transaction {
std::result::Result::Ok(transaction) => transaction,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let provenance = build_provenance(request, provider, endpoint, received_at);
@@ -215,182 +162,60 @@ fn convert_available_fields(
std::result::Result::Ok(provenance) => provenance,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
let observation_key = observation_key(request, &reference, provider, endpoint);
let transaction = ksp_store_lib::RawTransaction::new(reference.clone(), fields.slot, block_time, payload);
let observation = ksp_store_lib::RawTransactionObservation::new(observation_key, reference, provenance);
return std::result::Result::Ok(crate::BackfillRawAcquisition { inner: Box::new(BackfillRawAcquisitionInner { transaction, observation }) });
let observation_key = observation_key(request, transaction.reference(), provider, endpoint);
let acquisition = ksp_raw_transaction_lib::assemble_raw_transaction_acquisition(transaction, observation_key, provenance);
return std::result::Result::Ok(crate::BackfillRawAcquisition { inner: Box::new(BackfillRawAcquisitionInner { acquisition }) });
}
fn convert_block_time(value: std::option::Option<i64>) -> ksp_core_lib::Result<std::option::Option<ksp_store_lib::RawTimestamp>> {
let seconds = match value {
std::option::Option::Some(seconds) => seconds,
std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
};
let seconds = match u64::try_from(seconds) {
std::result::Result::Ok(seconds) => seconds,
std::result::Result::Err(_) => return std::result::Result::Err(conversion_error("block_time")),
};
let millis = match seconds.checked_mul(1_000) {
std::option::Option::Some(millis) => millis,
std::option::Option::None => return std::result::Result::Err(conversion_error("block_time")),
};
let timestamp = ksp_store_lib::RawTimestamp::from_unix_millis(millis);
return match timestamp {
std::result::Result::Ok(timestamp) => std::result::Result::Ok(std::option::Option::Some(timestamp)),
std::result::Result::Err(_) => std::result::Result::Err(conversion_error("block_time")),
};
}
fn canonical_payload_bytes(fields: &CanonicalTransactionFields<'_>) -> ksp_core_lib::Result<std::vec::Vec<u8>> {
let (transaction_data, transaction_encoding) = match fields.transaction {
ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { data, encoding } => {
if *encoding != ksp_onchain_transport_lib::SolanaTransactionBinaryEncoding::Base64 {
return std::result::Result::Err(conversion_error("transaction.encoding"));
}
(data.as_str(), "base64")
fn canonical_transaction(
reference: &ksp_store_lib::RawTransactionReference,
fields: CanonicalTransactionFields<'_>,
) -> ksp_core_lib::Result<ksp_store_lib::RawTransaction> {
let transaction_data = match fields.transaction {
ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { data, encoding }
if *encoding == ksp_onchain_transport_lib::SolanaTransactionBinaryEncoding::Base64 =>
{
data.clone()
},
ksp_onchain_transport_lib::SolanaEncodedTransaction::LegacyBinary(_) | ksp_onchain_transport_lib::SolanaEncodedTransaction::Json(_) => {
ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { .. }
| ksp_onchain_transport_lib::SolanaEncodedTransaction::LegacyBinary(_)
| ksp_onchain_transport_lib::SolanaEncodedTransaction::Json(_) => {
return std::result::Result::Err(conversion_error("transaction.encoding"));
},
};
let mut bytes = std::vec::Vec::new();
bytes.extend_from_slice(b"{\"transaction\":[");
if let std::result::Result::Err(error) = append_json_string(&mut bytes, transaction_data) {
return std::result::Result::Err(error);
}
bytes.push(b',');
if let std::result::Result::Err(error) = append_json_string(&mut bytes, transaction_encoding) {
return std::result::Result::Err(error);
}
bytes.push(b']');
if let std::result::Result::Err(error) = append_wire_value(&mut bytes, "meta", fields.meta, |output, value| return append_canonical_json(output, value)) {
return std::result::Result::Err(error);
}
let version_result = append_wire_value(&mut bytes, "version", fields.version, |output, value| {
let meta = map_wire_field(fields.meta, |value| return value.clone());
let version = map_wire_field(fields.version, |value| {
return match value {
ksp_onchain_transport_lib::SolanaTransactionVersion::Legacy => append_json_string(output, "legacy"),
ksp_onchain_transport_lib::SolanaTransactionVersion::Number(number) => {
output.extend_from_slice(number.to_string().as_bytes());
std::result::Result::Ok(())
},
ksp_onchain_transport_lib::SolanaTransactionVersion::Legacy => ksp_raw_transaction_lib::RawTransactionVersion::Legacy,
ksp_onchain_transport_lib::SolanaTransactionVersion::Number(number) => ksp_raw_transaction_lib::RawTransactionVersion::Number(*number),
};
});
if let std::result::Result::Err(error) = version_result {
return std::result::Result::Err(error);
}
let transaction_index_result = append_wire_value(&mut bytes, "transactionIndex", fields.transaction_index, |output, value| {
output.extend_from_slice(value.to_string().as_bytes());
return std::result::Result::Ok(());
});
if let std::result::Result::Err(error) = transaction_index_result {
return std::result::Result::Err(error);
}
bytes.push(b'}');
return std::result::Result::Ok(bytes);
let transaction_index = map_wire_field(fields.transaction_index, |value| return *value);
let material = ksp_raw_transaction_lib::RawTransactionMaterial::binary_base64(
reference.network().clone(),
reference.signature(),
fields.slot,
fields.block_time,
transaction_data,
meta,
version,
transaction_index,
);
let canonical = ksp_raw_transaction_lib::canonicalize_raw_transaction(material);
return match canonical {
std::result::Result::Ok(transaction) => std::result::Result::Ok(transaction),
std::result::Result::Err(_) => std::result::Result::Err(conversion_error("common.canonicalization")),
};
}
fn append_wire_value<T, F>(
output: &mut std::vec::Vec<u8>,
key: &str,
field: &ksp_onchain_transport_lib::SolanaWireField<T>,
mut append_value: F,
) -> ksp_core_lib::Result<()>
fn map_wire_field<T, U, F>(field: &ksp_onchain_transport_lib::SolanaWireField<T>, mut map_value: F) -> ksp_raw_transaction_lib::RawTransactionWireField<U>
where
F: FnMut(&mut std::vec::Vec<u8>, &T) -> ksp_core_lib::Result<()>,
F: FnMut(&T) -> U,
{
return match field {
ksp_onchain_transport_lib::SolanaWireField::Omitted => std::result::Result::Ok(()),
ksp_onchain_transport_lib::SolanaWireField::Null => {
output.push(b',');
let key_result = append_json_string(output, key);
if let std::result::Result::Err(error) = key_result {
return std::result::Result::Err(error);
}
output.extend_from_slice(b":null");
std::result::Result::Ok(())
},
ksp_onchain_transport_lib::SolanaWireField::Value(value) => {
output.push(b',');
let key_result = append_json_string(output, key);
if let std::result::Result::Err(error) = key_result {
return std::result::Result::Err(error);
}
output.push(b':');
append_value(output, value)
},
};
}
fn append_canonical_json(output: &mut std::vec::Vec<u8>, value: &serde_json::Value) -> ksp_core_lib::Result<()> {
return match value {
serde_json::Value::Null => {
output.extend_from_slice(b"null");
std::result::Result::Ok(())
},
serde_json::Value::Bool(value) => {
if *value {
output.extend_from_slice(b"true");
} else {
output.extend_from_slice(b"false");
}
std::result::Result::Ok(())
},
serde_json::Value::Number(value) => {
output.extend_from_slice(value.to_string().as_bytes());
std::result::Result::Ok(())
},
serde_json::Value::String(value) => append_json_string(output, value.as_str()),
serde_json::Value::Array(values) => {
output.push(b'[');
for (index, item) in values.iter().enumerate() {
if index != 0 {
output.push(b',');
}
let item_result = append_canonical_json(output, item);
if let std::result::Result::Err(error) = item_result {
return std::result::Result::Err(error);
}
}
output.push(b']');
std::result::Result::Ok(())
},
serde_json::Value::Object(values) => {
output.push(b'{');
let mut keys = values.keys().collect::<std::vec::Vec<_>>();
keys.sort_unstable();
for (index, key) in keys.iter().enumerate() {
if index != 0 {
output.push(b',');
}
let key_result = append_json_string(output, key.as_str());
if let std::result::Result::Err(error) = key_result {
return std::result::Result::Err(error);
}
output.push(b':');
let item = values.get(key.as_str());
let item = match item {
std::option::Option::Some(item) => item,
std::option::Option::None => return std::result::Result::Err(conversion_error("payload.meta")),
};
let item_result = append_canonical_json(output, item);
if let std::result::Result::Err(error) = item_result {
return std::result::Result::Err(error);
}
}
output.push(b'}');
std::result::Result::Ok(())
},
};
}
fn append_json_string(output: &mut std::vec::Vec<u8>, value: &str) -> ksp_core_lib::Result<()> {
let encoded = serde_json::to_vec(value);
return match encoded {
std::result::Result::Ok(encoded) => {
output.extend_from_slice(encoded.as_slice());
std::result::Result::Ok(())
},
std::result::Result::Err(_) => std::result::Result::Err(conversion_error("payload.json")),
ksp_onchain_transport_lib::SolanaWireField::Omitted => ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
ksp_onchain_transport_lib::SolanaWireField::Null => ksp_raw_transaction_lib::RawTransactionWireField::Null,
ksp_onchain_transport_lib::SolanaWireField::Value(value) => ksp_raw_transaction_lib::RawTransactionWireField::Value(map_value(value)),
};
}
@@ -463,18 +288,6 @@ fn hash_bytes(hasher: &mut sha2::Sha256, value: &[u8]) {
hasher.update(value);
}
fn base58_digit(byte: u8) -> std::option::Option<u8> {
return match byte {
b'1'..=b'9' => std::option::Option::Some(byte - b'1'),
b'A'..=b'H' => std::option::Option::Some((byte - b'A') + 9),
b'J'..=b'N' => std::option::Option::Some((byte - b'J') + 17),
b'P'..=b'Z' => std::option::Option::Some((byte - b'P') + 22),
b'a'..=b'k' => std::option::Option::Some((byte - b'a') + 33),
b'm'..=b'z' => std::option::Option::Some((byte - b'm') + 44),
_ => std::option::Option::None,
};
}
fn conversion_error(field: &'static str) -> ksp_core_lib::Error {
return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID, "invalid deterministic Backfill RAW conversion")
.with_context("field", field);