0.3.15-pre.013

crates/ksp-worker-raw-transaction-ingest-lib/README.md

@@ -1,5 +1,5 @@
 <!-- file: crates/ksp-worker-raw-transaction-ingest-lib/README.md -->
-<!-- version: 14 -->
+<!-- version: 15 -->
 
 # ksp-worker-raw-transaction-ingest-lib
 
@@ -80,12 +80,14 @@ La construction est sans I/O et refuse notamment :
 - un commitment `Processed` ou implicite ;
 - un réseau Yellowstone non représentable ;
 - une provenance provider/endpoint non représentable ;
-- l'absence d'une route HTTP compatible pour `getTransaction` sur le même réseau.
+- l'absence d'une route HTTP compatible avec le mode retenu : `getTransaction` pour les signaux transactionnels, `getBlock` pour un abonnement Yellowstone Block pur, sur le même réseau.
 
 Le runtime-resource aggregate public accepte une collection validée de 1 à 32 sources logiques et les démarre simultanément sous un supervisor privé. La collection est validée entièrement avant spawn ; aucun sous-ensemble silencieux, source primaire implicite ou standby n'est choisi. La collection interne, les `source_key`, les URLs, les filtres et les clients inférieurs ne sont pas exposés.
 
 Le supervisor possède toutes les tâches source. Une perte de source n'est plus assimilée automatiquement à un fault : lorsqu'elle porte une plage de continuité bornée, le Worker l'inscrit dans son ledger run-local puis n'autorise la continuation des siblings que si la coverage passée de cette perte est réconciliée et si les sources encore actives couvrent explicitement tout le `TargetCoverage` futur. Une perte sans plage sûre, une coverage insuffisante ou un gap encore ouvert reste terminal. Le Worker ne respawn jamais lui-même une source Transport.
 
+Pour un abonnement Yellowstone `Block` pur, la notification gRPC est un trigger de slot et ne bloque plus la boucle de réception pendant l'hydration HTTP. Le slot entre dans un coordinateur privé borné ; les `getBlock` puis l'admission complète du bloc s'exécutent dans au plus le quota `in-flight` attribué à cette source, tandis que le nombre de slots en attente reste borné par son quota `pending`. La boucle Yellowstone continue donc à consommer `next_update()` tant qu'une capacité pending existe. Aucun `unbounded_channel`, aucun task détaché et aucun silent drop n'est introduit ; si le débit aval reste durablement inférieur au débit de chaîne jusqu'à saturation de toutes les bornes, la politique reste fail-closed.
+
 Un inventaire privé `source_key -> latest processing/source state`, borné à 32 entrées, agrège la projection run-local. La frontier agrégée reste conservative : elle n'expose un `processing_frontier_slot` que lorsque toutes les sources en possèdent un, choisit le minimum des frontiers connus et le plus ancien pending. Les sources reference-bearing partagent en plus un registre global d'hydration borné : une même clé `(network, signature, commitment)` ne déclenche qu'un leader HTTP, puis chaque signal source conserve sa propre provenance lors de la finalisation.
 
 ## Contrat de source Standard Logs + HTTP
@@ -171,17 +173,17 @@ Le shutdown est borné par `shutdown_drain_timeout` (10 s par défaut). Le super
 
 La queue centrale est un `tokio::sync::mpsc` privé borné par `admission_queue_capacity`. Les sources internes subissent la backpressure ; aucune queue non bornée ni silent drop n'est autorisé.
 
-Le Worker possède des coordinateurs source-neutral Yellowstone, Standard Logs et Helius Transaction branchés sur un registre global d'hydration partagé. Standard Block et HTTP Block Polling n'entrent pas dans ce registre lorsqu'une transaction est direct-qualified.
+Le Worker possède des coordinateurs bornés pour Yellowstone, Standard Logs et Helius Transaction. Les voies Yellowstone/Standard Logs/Helius basées sur des références partagent le registre global d'hydration `(network, signature, commitment)`. Le mode Yellowstone Block utilise un coordinateur de slots séparé mais reçoit sa part des mêmes budgets globaux pending/in-flight avant spawn ; Standard Block et HTTP Block Polling n'entrent pas dans ces quotas lorsqu'une transaction est direct-qualified.
 
-Les sources reference-bearing reçoivent des quotas déterministes dont la somme reste exactement dans les bornes techniques configurées :
+Les sources nécessitant une hydration HTTP asynchrone, y compris Yellowstone Block, reçoivent des quotas déterministes dont la somme reste exactement dans les bornes techniques configurées :
 
 ```text
 somme pending source signals <= admission_queue_capacity
 somme hydration tasks in flight <= persistence_concurrency
-source reference-bearing active => quota pending >= 1 et quota in-flight >= 1
+source hydratante active => quota pending >= 1 et quota in-flight >= 1
 ```
 
-Pour garantir simultanément ces bornes et l'absence de starvation structurelle, le démarrage échoue avant spawn si le nombre de sources reference-bearing dépasse `admission_queue_capacity` ou `persistence_concurrency`. Un sémaphore global protège en plus l'ouverture effective des hydrations HTTP.
+Pour garantir simultanément ces bornes et l'absence de starvation structurelle, le démarrage échoue avant spawn si le nombre de sources hydratantes dépasse `admission_queue_capacity` ou `persistence_concurrency`. Le registre des hydrations transactionnelles protège en plus l'ouverture effective des `getTransaction`; Yellowstone Block reste borné par sa partition déterministe et son `JoinSet` possédé.
 
 Les signaux partageant le même `(network, signature, commitment)` sont coalescés cross-source avant le fan-out HTTP. La publication du résultat partagé notifie les followers sous le verrou de registry avant de retirer la clé : une nouvelle génération de leader ne peut donc pas s'intercaler entre retrait et notification. Après canonicalisation, une cache run-local bornée sérialise les acquisitions de même `(network, signature)` : la première passe par l'écriture atomique entity + observation, les suivantes de contenu canonique identique ajoutent uniquement leur observation déterministe. Une divergence de slot, block time, format ou hash canonique devient un content conflict terminal ; aucune majorité, préférence provider ou overwrite n'est appliqué. Le Store conserve son guard durable final.
 

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 47
+// version: 48
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -424,7 +424,7 @@ impl RawTransactionIngestLiveSource {
     fn uses_hydration(&self) -> bool {
         return match self {
             Self::HeliusTransaction(_) | Self::StandardLogs(_) => true,
-            Self::Yellowstone(source) => source.uses_transaction_hydration(),
+            Self::Yellowstone(_) => true,
             Self::HttpBlockPolling(_) | Self::StandardBlock(_) => false,
         };
     }
@@ -1014,6 +1014,123 @@ struct RawTransactionIngestHydrationContext {
     source_key_domain: &'static [u8],
 }
 
+#[derive(Clone)]
+struct RawTransactionIngestYellowstoneBlockHydrationContext {
+    capture_session: ksp_store_lib::RawProvenanceCode,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    http_pool: ksp_onchain_transport_lib::HttpTransportPool,
+    hydration_role: ksp_onchain_transport_lib::HttpRoleName,
+    network: ksp_store_lib::RawNetworkId,
+    route: RawTransactionIngestSourceRoute,
+    source_key: [u8; 32],
+}
+
+type RawTransactionIngestYellowstoneBlockHydrationTasks = tokio::task::JoinSet<ksp_core_lib::Result<std::option::Option<u64>>>;
+
+struct RawTransactionIngestYellowstoneBlockHydrationCoordinator {
+    max_in_flight: usize,
+    max_pending_slots: usize,
+    pending: std::collections::BTreeMap<u64, bool>,
+    tasks: RawTransactionIngestYellowstoneBlockHydrationTasks,
+}
+
+impl RawTransactionIngestYellowstoneBlockHydrationCoordinator {
+    fn new(max_pending_slots: usize, max_in_flight: usize) -> Self {
+        return Self {
+            max_in_flight,
+            max_pending_slots,
+            pending: std::collections::BTreeMap::new(),
+            tasks: RawTransactionIngestYellowstoneBlockHydrationTasks::new(),
+        };
+    }
+
+    fn can_receive(&self) -> bool {
+        return self.pending.len() < self.max_pending_slots;
+    }
+
+    fn queue_slot(&mut self, slot: u64, processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter) -> ksp_core_lib::Result<()> {
+        if self.pending.contains_key(&slot) {
+            return std::result::Result::Ok(());
+        }
+        if self.max_pending_slots == 0 || self.pending.len() >= self.max_pending_slots {
+            return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_hydration_pending_saturated"));
+        }
+        if let std::result::Result::Err(error) = processing_frontier.observe_pending(slot) {
+            return std::result::Result::Err(error);
+        }
+        let _previous = self.pending.insert(slot, false);
+        return std::result::Result::Ok(());
+    }
+
+    fn start_hydrations(
+        &mut self,
+        hydration: &RawTransactionIngestYellowstoneBlockHydrationContext,
+        admission_sender: &tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
+        stop_receiver: &tokio::sync::watch::Receiver<bool>,
+    ) -> ksp_core_lib::Result<()> {
+        if self.max_in_flight == 0 {
+            return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_hydration_concurrency_missing"));
+        }
+        while self.tasks.len() < self.max_in_flight {
+            let slot = self.pending.iter().find_map(|(slot, in_flight)| {
+                if *in_flight {
+                    return std::option::Option::None;
+                }
+                return std::option::Option::Some(*slot);
+            });
+            let slot = match slot {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => break,
+            };
+            let in_flight = match self.pending.get_mut(&slot) {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => {
+                    return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_hydration_pending_missing"));
+                },
+            };
+            *in_flight = true;
+            let task_hydration = hydration.clone();
+            let task_admission_sender = admission_sender.clone();
+            let task_stop_receiver = stop_receiver.clone();
+            let _abort_handle = self.tasks.spawn(async move {
+                return hydrate_yellowstone_block(task_hydration, slot, task_admission_sender, task_stop_receiver).await;
+            });
+        }
+        return std::result::Result::Ok(());
+    }
+
+    fn handle_joined(
+        &mut self,
+        joined: std::result::Result<ksp_core_lib::Result<std::option::Option<u64>>, tokio::task::JoinError>,
+        processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter,
+    ) -> ksp_core_lib::Result<bool> {
+        let completed_slot = 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.yellowstone_block_hydration_task_join_failed")),
+        };
+        let slot = match completed_slot {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return std::result::Result::Ok(false),
+        };
+        if self.pending.remove(&slot).is_none() {
+            return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_hydration_result_without_pending"));
+        }
+        if let std::result::Result::Err(error) = processing_frontier.settle_pending(slot) {
+            return std::result::Result::Err(error);
+        }
+        return std::result::Result::Ok(true);
+    }
+
+    async fn abort_all(&mut self, processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter) {
+        self.tasks.abort_all();
+        while self.tasks.join_next().await.is_some() {}
+        self.pending.clear();
+        processing_frontier.discard_all_pending();
+        return;
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum RawTransactionIngestYellowstoneMode {
     TransactionHydration,
@@ -1101,10 +1218,6 @@ impl crate::RawTransactionIngestYellowstoneSource {
         return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, mode, network, route, source_key });
     }
 
-    fn uses_transaction_hydration(&self) -> bool {
-        return self.mode == RawTransactionIngestYellowstoneMode::TransactionHydration;
-    }
-
     fn hydration_context(&self) -> RawTransactionIngestHydrationContext {
         let commitment = match self.subscribe_request.commitment() {
             std::option::Option::Some(value) => value,
@@ -1123,6 +1236,31 @@ impl crate::RawTransactionIngestYellowstoneSource {
         };
     }
 
+    fn block_hydration_context(
+        &self,
+        settings: &crate::RawTransactionIngestSettings,
+    ) -> ksp_core_lib::Result<RawTransactionIngestYellowstoneBlockHydrationContext> {
+        let commitment = match self.subscribe_request.commitment() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.yellowstone_commitment_missing")),
+        };
+        let capture_session = match ksp_store_lib::RawProvenanceCode::new(settings.worker_id().as_str()) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_capture_session_unrepresentable"));
+            },
+        };
+        return std::result::Result::Ok(RawTransactionIngestYellowstoneBlockHydrationContext {
+            capture_session,
+            commitment,
+            http_pool: self.http_pool.clone(),
+            hydration_role: self.hydration_role.clone(),
+            network: self.network.clone(),
+            route: self.route.clone(),
+            source_key: self.source_key,
+        });
+    }
+
     /// Runs the productive Yellowstone source task until cooperative stop or one safe terminal source failure.
     async fn run(
         self,
@@ -1133,7 +1271,7 @@ impl crate::RawTransactionIngestYellowstoneSource {
         shared: RawTransactionIngestSourceRuntimeShared,
     ) -> ksp_core_lib::Result<()> {
         if self.mode == RawTransactionIngestYellowstoneMode::BlockHydration {
-            return self.run_block_hydration(settings, stop_receiver, admission_sender, processing_frontier_sender).await;
+            return self.run_block_hydration(settings, stop_receiver, admission_sender, processing_frontier_sender, shared).await;
         }
         let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } =
             shared;
@@ -1269,7 +1407,13 @@ impl crate::RawTransactionIngestYellowstoneSource {
         mut stop_receiver: tokio::sync::watch::Receiver<bool>,
         admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
         processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
+        shared: RawTransactionIngestSourceRuntimeShared,
     ) -> ksp_core_lib::Result<()> {
+        let RawTransactionIngestSourceRuntimeShared { hydration_in_flight_limit, hydration_pending_limit, .. } = shared;
+        let hydration = match self.block_hydration_context(&settings) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
         let opened = tokio::select! {
             biased;
             _ = stop_receiver.changed() => {
@@ -1281,6 +1425,7 @@ impl crate::RawTransactionIngestYellowstoneSource {
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())),
         };
+        let mut coordinator = RawTransactionIngestYellowstoneBlockHydrationCoordinator::new(hydration_pending_limit, hydration_in_flight_limit);
         let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender);
         let snapshot_source = session.snapshot_source();
         if let std::result::Result::Err(error) = processing_frontier.observe_session_snapshot(snapshot_source.current()) {
@@ -1294,6 +1439,15 @@ impl crate::RawTransactionIngestYellowstoneSource {
             if *stop_receiver.borrow() {
                 break;
             }
+            if let std::result::Result::Err(error) = coordinator.start_hydrations(&hydration, &admission_sender, &stop_receiver) {
+                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.yellowstone_block_hydration_stalled"));
+                break;
+            }
             tokio::select! {
                 biased;
                 _ = stop_receiver.changed() => {
@@ -1310,7 +1464,24 @@ impl crate::RawTransactionIngestYellowstoneSource {
                         std::option::Option::None => session_snapshot_source = std::option::Option::None,
                     }
                 }
-                update = session.next_update() => {
+                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.yellowstone_block_hydration_join_missing"));
+                            break;
+                        },
+                    };
+                    match coordinator.handle_joined(joined, &mut processing_frontier) {
+                        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) => {
@@ -1322,42 +1493,16 @@ impl crate::RawTransactionIngestYellowstoneSource {
                             break;
                         },
                     };
-                    if let ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(value) = update {
-                        let fetched = fetch_yellowstone_block_ingresses(&self, &settings, value.slot(), &mut stop_receiver).await;
-                        let ingresses = match fetched {
-                            std::result::Result::Ok(std::option::Option::Some(value)) => value,
-                            std::result::Result::Ok(std::option::Option::None) => break,
-                            std::result::Result::Err(error) => {
-                                fault = std::option::Option::Some(error);
-                                break;
-                            },
-                        };
-                        for ingress in ingresses {
-                            let sent = tokio::select! {
-                                biased;
-                                _ = stop_receiver.changed() => {
-                                    break;
-                                }
-                                result = admission_sender.send(ingress) => result,
-                            };
-                            if sent.is_err() {
-                                if !*stop_receiver.borrow() {
-                                    fault = std::option::Option::Some(crate::runtime_error("source.admission_closed"));
-                                }
-                                break;
-                            }
-                        }
-                        if fault.is_some() || *stop_receiver.borrow() {
-                            break;
-                        }
-                        if let std::result::Result::Err(error) = processing_frontier.observe_settled(value.slot()) {
-                            fault = std::option::Option::Some(error);
-                            break;
-                        }
+                    if let ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(value) = update
+                        && let std::result::Result::Err(error) = coordinator.queue_slot(value.slot(), &mut processing_frontier)
+                    {
+                        fault = std::option::Option::Some(error);
+                        break;
                     }
                 }
             }
         }
+        coordinator.abort_all(&mut processing_frontier).await;
         processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closing);
         let closed = session.close().await;
         if let std::option::Option::Some(error) = fault {
@@ -3376,16 +3521,42 @@ fn validate_http_block_polling_discovery(next_scan_slot: u64, discovered: &[u64]
     return std::result::Result::Ok(());
 }
 
+async fn hydrate_yellowstone_block(
+    hydration: RawTransactionIngestYellowstoneBlockHydrationContext,
+    slot: u64,
+    admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
+    mut stop_receiver: tokio::sync::watch::Receiver<bool>,
+) -> ksp_core_lib::Result<std::option::Option<u64>> {
+    let fetched = fetch_yellowstone_block_ingresses(&hydration, slot, &mut stop_receiver).await;
+    let ingresses = match fetched {
+        std::result::Result::Ok(std::option::Option::Some(value)) => value,
+        std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Ok(std::option::Option::None),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    for ingress in ingresses {
+        let sent = tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => {
+                return std::result::Result::Ok(std::option::Option::None);
+            }
+            result = admission_sender.send(ingress) => result,
+        };
+        if sent.is_err() {
+            if *stop_receiver.borrow() {
+                return std::result::Result::Ok(std::option::Option::None);
+            }
+            return std::result::Result::Err(crate::runtime_error("source.admission_closed"));
+        }
+    }
+    return std::result::Result::Ok(std::option::Option::Some(slot));
+}
+
 async fn fetch_yellowstone_block_ingresses(
-    source: &crate::RawTransactionIngestYellowstoneSource,
-    settings: &crate::RawTransactionIngestSettings,
+    hydration: &RawTransactionIngestYellowstoneBlockHydrationContext,
     slot: u64,
     stop_receiver: &mut tokio::sync::watch::Receiver<bool>,
 ) -> ksp_core_lib::Result<std::option::Option<std::vec::Vec<crate::RawTransactionIngress>>> {
-    let commitment = match source.subscribe_request.commitment() {
-        std::option::Option::Some(value) => value,
-        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.yellowstone_commitment_missing")),
-    };
+    let commitment = hydration.commitment;
     let config = ksp_onchain_transport_lib::SolanaGetBlockConfig::new(
         std::option::Option::Some(commitment),
         std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64),
@@ -3401,7 +3572,7 @@ async fn fetch_yellowstone_block_ingresses(
             _ = stop_receiver.changed() => {
                 return std::result::Result::Ok(std::option::Option::None);
             }
-            result = source.http_pool.get_block_observed(&source.hydration_role, slot, std::option::Option::Some(&config)) => result,
+            result = hydration.http_pool.get_block_observed(&hydration.hydration_role, slot, std::option::Option::Some(&config)) => result,
         };
         match request {
             std::result::Result::Ok(value) if value.value().is_some() => break value,
@@ -3434,36 +3605,35 @@ async fn fetch_yellowstone_block_ingresses(
             return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_transactions_missing"));
         },
     };
-    let provenance = build_yellowstone_block_provenance(source, settings, observed.endpoint_name(), observed.provider(), received_at);
+    let provenance = build_yellowstone_block_provenance(hydration, observed.endpoint_name(), observed.provider(), received_at);
     let provenance = match provenance {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(error) => return std::result::Result::Err(error),
     };
     let mut ingresses = std::vec::Vec::with_capacity(transactions.len());
     for (position, transaction) in transactions.iter().enumerate() {
-        let material = build_http_block_polling_material_from_transaction(&source.network, slot, block.block_time(), transaction, position);
+        let material = build_http_block_polling_material_from_transaction(&hydration.network, slot, block.block_time(), transaction, position);
         let material = match material {
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => return std::result::Result::Err(error),
         };
         ingresses.push(crate::RawTransactionIngress {
             material,
-            network: source.network.clone(),
+            network: hydration.network.clone(),
             provenance: provenance.clone(),
-            source_key: source.source_key,
+            source_key: hydration.source_key,
         });
     }
     return std::result::Result::Ok(std::option::Option::Some(ingresses));
 }
 
 fn build_yellowstone_block_provenance(
-    source: &crate::RawTransactionIngestYellowstoneSource,
-    settings: &crate::RawTransactionIngestSettings,
+    hydration: &RawTransactionIngestYellowstoneBlockHydrationContext,
     endpoint_name: &str,
     provider_name: &ksp_onchain_transport_lib::HttpProviderName,
     received_at: ksp_store_lib::RawTimestamp,
 ) -> ksp_core_lib::Result<ksp_store_lib::RawAcquisitionProvenance> {
-    let (provider, endpoint_id) = match composite_provenance_codes(&source.route, "ys", provider_name.as_str(), endpoint_name) {
+    let (provider, endpoint_id) = match composite_provenance_codes(&hydration.route, "ys", provider_name.as_str(), endpoint_name) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(error) => return std::result::Result::Err(error),
     };
@@ -3475,15 +3645,7 @@ fn build_yellowstone_block_provenance(
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_method_unrepresentable")),
     };
-    let capture_session = match ksp_store_lib::RawProvenanceCode::new(settings.worker_id().as_str()) {
-        std::result::Result::Ok(value) => value,
-        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_capture_session_unrepresentable")),
-    };
-    let commitment = match source.subscribe_request.commitment() {
-        std::option::Option::Some(value) => value,
-        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.yellowstone_commitment_missing")),
-    };
-    let commitment = match ksp_store_lib::RawProvenanceCode::new(commitment.as_str()) {
+    let commitment = match ksp_store_lib::RawProvenanceCode::new(hydration.commitment.as_str()) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.yellowstone_block_commitment_unrepresentable")),
     };
@@ -3491,7 +3653,7 @@ fn build_yellowstone_block_provenance(
         ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, acquisition_method, ksp_store_lib::RawAcquisitionOrigin::Live, received_at)
             .with_endpoint_id(endpoint_id)
             .with_commitment(commitment)
-            .with_capture_session_id(capture_session),
+            .with_capture_session_id(hydration.capture_session.clone()),
     );
 }
 

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs
-// version: 38
+// version: 39
 
 //! External public, security, redaction and release-boundary hardening canaries through `v0.3.15-pre.004`.
 
@@ -1413,3 +1413,39 @@ fn v0_3_15_pre_004_websocket_source_constructors_enforce_transport_capabilities_
     assert!(!resources.contains("match ws_endpoint.provider()"));
     return;
 }
+
+#[test]
+fn v0_3_15_pre_013_yellowstone_block_hydration_is_bounded_concurrent_and_stop_preemptible() {
+    let resources = include_str!("../src/runtime_resources.rs");
+    let block_run = match resources.split_once("async fn run_block_hydration(") {
+        std::option::Option::Some((_, tail)) => match tail.split_once("/// Validated Helius") {
+            std::option::Option::Some((body, _)) => body,
+            std::option::Option::None => tail,
+        },
+        std::option::Option::None => "",
+    };
+    for required in [
+        "RawTransactionIngestYellowstoneBlockHydrationCoordinator::new",
+        "coordinator.start_hydrations",
+        "joined = coordinator.tasks.join_next()",
+        "update = session.next_update(), if can_receive",
+        "coordinator.queue_slot(value.slot()",
+        "coordinator.abort_all(&mut processing_frontier).await",
+    ] {
+        assert!(block_run.contains(required), "required pre.013 bounded Yellowstone block hydration guard missing: {required}");
+    }
+    assert!(!block_run.contains("fetch_yellowstone_block_ingresses(&self"));
+    assert!(!block_run.contains("admission_sender.send(ingress)"));
+    for required in [
+        "hydrate_yellowstone_block",
+        "result = admission_sender.send(ingress)",
+        "source.yellowstone_block_hydration_pending_saturated",
+        "source.yellowstone_block_hydration_concurrency_missing",
+    ] {
+        assert!(resources.contains(required), "required pre.013 Yellowstone block hydration bound missing: {required}");
+    }
+    for forbidden in ["unbounded_channel", "tokio::spawn(", "std::thread::spawn("] {
+        assert!(!block_run.contains(forbidden), "pre.013 introduced forbidden unbounded/detached execution: {forbidden}");
+    }
+    return;
+}

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs
-// version: 34
+// version: 35
 
 //! Release-completeness canaries through the `v0.3.15-pre.004` WebSocket capability-enforcement tranche.
 
@@ -484,3 +484,15 @@ fn v0_3_15_pre_004_websocket_capability_enforcement_canaries_are_complete_withou
     assert!(!root.contains("supports_subscription"));
     return;
 }
+
+#[test]
+fn v0_3_15_pre_013_yellowstone_block_backpressure_fix_adds_no_public_surface() {
+    let resource_tests = include_str!("../unit_tests/runtime_resources.rs");
+    let hardening = include_str!("hardening.rs");
+    let root = include_str!("../src/lib.rs");
+    assert!(resource_tests.contains("v0_3_15_pre_013_yellowstone_block_hydration_coordinator_is_bounded_before_http_work"));
+    assert!(hardening.contains("v0_3_15_pre_013_yellowstone_block_hydration_is_bounded_concurrent_and_stop_preemptible"));
+    assert!(!root.contains("YellowstoneBlockHydrationCoordinator"));
+    assert!(!root.contains("YellowstoneBlockHydrationContext"));
+    return;
+}

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs
-// version: 38
+// version: 39
 
 fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
     return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -879,7 +879,31 @@ async fn pre_009_yellowstone_block_mode_uses_get_block_without_transaction_hydra
         std::result::Result::Err(_) => return,
     };
     assert_eq!(source.mode, super::RawTransactionIngestYellowstoneMode::BlockHydration);
-    assert!(!source.uses_transaction_hydration());
+    assert!(super::RawTransactionIngestLiveSource::Yellowstone(source).uses_hydration());
+    return;
+}
+
+#[test]
+fn v0_3_15_pre_013_yellowstone_block_hydration_coordinator_is_bounded_before_http_work() {
+    let (sender, receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
+    let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(sender);
+    let mut coordinator = super::RawTransactionIngestYellowstoneBlockHydrationCoordinator::new(2, 1);
+    assert!(coordinator.can_receive());
+    assert!(coordinator.queue_slot(100, &mut processing_frontier).is_ok());
+    assert!(coordinator.queue_slot(101, &mut processing_frontier).is_ok());
+    assert!(!coordinator.can_receive());
+    assert_eq!(receiver.borrow().hydration_pending(), 2);
+    assert_eq!(receiver.borrow().oldest_pending_slot(), std::option::Option::Some(100));
+    let saturated = coordinator.queue_slot(102, &mut processing_frontier);
+    let error = match saturated {
+        std::result::Result::Ok(()) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+    assert_eq!(error.context()[0].value(), "source.yellowstone_block_hydration_pending_saturated");
+    assert_eq!(coordinator.pending.len(), 2);
+    assert_eq!(coordinator.tasks.len(), 0);
+    return;
 }
 
 #[tokio::test(flavor = "current_thread")]