diff --git a/Cargo.toml b/Cargo.toml index 53e5651..e8de464 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 407 +# version: 408 [workspace] resolver = "3" members = ["crates/ksp-app-config-desk", "crates/ksp-app-solprices-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-interface-lib", "crates/ksp-job-api", "crates/ksp-job-backfill-lib", "crates/ksp-logging-lib", "crates/ksp-offchain-transport-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-program-api", "crates/ksp-store-api", "crates/ksp-store-lib", "crates/ksp-store-postgres-lib", "crates/ksp-wallet-lib"] [workspace.package] -version = "0.3.6-pre.8.fix.1" +version = "0.3.6-pre.9" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-job-backfill-lib/Cargo.toml b/crates/ksp-job-backfill-lib/Cargo.toml index 7c39243..f8af817 100644 --- a/crates/ksp-job-backfill-lib/Cargo.toml +++ b/crates/ksp-job-backfill-lib/Cargo.toml @@ -1,5 +1,5 @@ # file: crates/ksp-job-backfill-lib/Cargo.toml -# version: 3 +# version: 4 [package] name = "ksp-job-backfill-lib" @@ -16,6 +16,7 @@ ksp-onchain-transport-lib = { path = "../ksp-onchain-transport-lib" } ksp-store-lib = { path = "../ksp-store-lib", default-features = false } serde_json.workspace = true sha2.workspace = true +tokio = { workspace = true, features = ["macros", "sync"] } [dev-dependencies] tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/ksp-job-backfill-lib/src/discovery.rs b/crates/ksp-job-backfill-lib/src/discovery.rs index 2a8f77c..f598740 100644 --- a/crates/ksp-job-backfill-lib/src/discovery.rs +++ b/crates/ksp-job-backfill-lib/src/discovery.rs @@ -1,5 +1,5 @@ // file: crates/ksp-job-backfill-lib/src/discovery.rs -// version: 3 +// version: 4 /// Network-scoped identity of one discovered transaction candidate before canonical signature decoding. #[derive(Clone, Debug, Eq, Hash, PartialEq)] @@ -217,7 +217,7 @@ pub async fn discover_backfill_candidates( max_candidates = request.max_candidates(), "starting bounded Backfill candidate discovery" ); - let result = discover_with_source(transport, request).await; + let result = discover_with_source(transport, request, std::option::Option::None).await; if let std::result::Result::Ok(discovery) = &result { ksp_logging_lib::debug!( target: crate::TRACING_TARGET, @@ -234,14 +234,30 @@ pub async fn discover_backfill_candidates( return result; } -async fn discover_with_source(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result +/// Discovers one bounded candidate set while allowing cooperative cancellation of in-flight RPC futures. +pub(crate) async fn discover_backfill_candidates_cancellable( + transport: &ksp_onchain_transport_lib::HttpTransportPool, + request: &crate::BackfillRequest, + cancellation: &crate::BackfillCancellationSignal, +) -> ksp_core_lib::Result { + return discover_with_source(transport, request, std::option::Option::Some(cancellation)).await; +} + +async fn discover_with_source( + source: &S, + request: &crate::BackfillRequest, + cancellation: std::option::Option<&crate::BackfillCancellationSignal>, +) -> ksp_core_lib::Result where S: SignaturePageSource, { + if cancellation.is_some_and(crate::BackfillCancellationSignal::is_requested) { + return std::result::Result::Err(cancelled_error()); + } return match request.scope().kind() { crate::BackfillScopeKind::ExplicitSignatures => discover_explicit(request), - crate::BackfillScopeKind::LatestAddress | crate::BackfillScopeKind::BeforeAddress => discover_older(source, request).await, - crate::BackfillScopeKind::AfterAddress => discover_after(source, request).await, + crate::BackfillScopeKind::LatestAddress | crate::BackfillScopeKind::BeforeAddress => discover_older(source, request, cancellation).await, + crate::BackfillScopeKind::AfterAddress => discover_after(source, request, cancellation).await, }; } @@ -264,7 +280,11 @@ fn discover_explicit(request: &crate::BackfillRequest) -> ksp_core_lib::Result(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result +async fn discover_older( + source: &S, + request: &crate::BackfillRequest, + cancellation: std::option::Option<&crate::BackfillCancellationSignal>, +) -> ksp_core_lib::Result where S: SignaturePageSource, { @@ -298,7 +318,7 @@ where std::option::Option::Some(request.commitment().transport()), request.min_context_slot(), ); - let page_result = source.fetch_signature_page(request.role(), address, config).await; + let page_result = fetch_signature_page(source, request.role(), address, config, cancellation).await; let page = match page_result { std::result::Result::Ok(page) => page, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -347,7 +367,11 @@ where return std::result::Result::Ok(BackfillDiscovery::new(request.network().clone(), request.scope_fingerprint(), candidates, pages_fetched, boundary)); } -async fn discover_after(source: &S, request: &crate::BackfillRequest) -> ksp_core_lib::Result +async fn discover_after( + source: &S, + request: &crate::BackfillRequest, + cancellation: std::option::Option<&crate::BackfillCancellationSignal>, +) -> ksp_core_lib::Result where S: SignaturePageSource, { @@ -375,7 +399,7 @@ where std::option::Option::Some(request.commitment().transport()), request.min_context_slot(), ); - let page_result = source.fetch_signature_page(request.role(), address, config).await; + let page_result = fetch_signature_page(source, request.role(), address, config, cancellation).await; let page = match page_result { std::result::Result::Ok(page) => page, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -430,6 +454,27 @@ where )); } +async fn fetch_signature_page( + source: &S, + role: &ksp_onchain_transport_lib::HttpRoleName, + address: &ksp_core_lib::Pubkey, + config: ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig, + cancellation: std::option::Option<&crate::BackfillCancellationSignal>, +) -> ksp_core_lib::Result> +where + S: SignaturePageSource, +{ + let future = source.fetch_signature_page(role, address, config); + return match cancellation { + std::option::Option::Some(signal) => signal.run_cancellable(future).await, + std::option::Option::None => future.await, + }; +} + +fn cancelled_error() -> ksp_core_lib::Error { + return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_CANCELLED, "Backfill discovery cancelled before RPC completion"); +} + fn validated_signature(value: &str) -> ksp_core_lib::Result { return crate::BackfillSignature::new(value.to_owned()); } diff --git a/crates/ksp-job-backfill-lib/src/error.rs b/crates/ksp-job-backfill-lib/src/error.rs index 0d0e30a..7cdc524 100644 --- a/crates/ksp-job-backfill-lib/src/error.rs +++ b/crates/ksp-job-backfill-lib/src/error.rs @@ -1,5 +1,5 @@ // file: crates/ksp-job-backfill-lib/src/error.rs -// version: 4 +// version: 5 /// Error code used when one Backfill checkpoint/frontier is incompatible with the current Job or semantic scope. pub const ERROR_CODE_BACKFILL_CHECKPOINT_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "checkpoint_invalid"); @@ -15,5 +15,10 @@ pub const ERROR_CODE_BACKFILL_PERSISTENCE_INVALID: ksp_core_lib::ErrorCode = ksp pub const ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "raw_conversion_invalid"); /// Error code used when one Backfill request violates its bounded admission contract. pub const ERROR_CODE_BACKFILL_REQUEST_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "request_invalid"); +/// Error code used when the concrete Backfill runtime reaches an impossible lifecycle or notification state. +pub const ERROR_CODE_BACKFILL_RUNTIME_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "runtime_invalid"); /// Error code used when one transaction signature text violates the bounded Base58-shape contract. pub const ERROR_CODE_BACKFILL_SIGNATURE_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "signature_invalid"); + +/// Error code used internally when cooperative cancellation wins before durable Store submission. +pub(crate) const ERROR_CODE_BACKFILL_CANCELLED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("job_backfill", "cancelled"); diff --git a/crates/ksp-job-backfill-lib/src/execution.rs b/crates/ksp-job-backfill-lib/src/execution.rs index 81b1349..4ccd2e5 100644 --- a/crates/ksp-job-backfill-lib/src/execution.rs +++ b/crates/ksp-job-backfill-lib/src/execution.rs @@ -1,22 +1,28 @@ // file: crates/ksp-job-backfill-lib/src/execution.rs -// version: 1 +// version: 2 use futures_util::StreamExt; // rust-rules: trait-import /// Bounded result of one concurrent Backfill candidate execution pass. -/// -/// This is intentionally smaller than the concrete latest-value snapshot planned for `pre.009`. -/// It exposes only the admission/frontier/checkpoint facts required to prove safe resumption. #[derive(Clone, Debug, Eq, PartialEq)] pub struct BackfillExecutionBatch { candidate_count: usize, admitted_count: usize, finished_count: usize, durable_count: usize, + inserted_count: usize, + already_present_count: usize, + purged_count: usize, + missing_count: usize, + conflict_count: usize, + observation_inserted_count: usize, + observation_already_present_count: usize, + cancelled_count: usize, hole_count: usize, maximum_in_flight: usize, local_contiguous_completed: usize, discovery_partial: bool, + cancelled: bool, checkpoint: crate::BackfillCheckpoint, failure_code: std::option::Option, } @@ -46,6 +52,54 @@ impl BackfillExecutionBatch { return self.durable_count; } + /// Returns the number of newly inserted canonical RAW entities. + #[must_use] + pub const fn inserted_count(&self) -> usize { + return self.inserted_count; + } + + /// Returns the number of canonical RAW entities already durable. + #[must_use] + pub const fn already_present_count(&self) -> usize { + return self.already_present_count; + } + + /// Returns the number of purge tombstones respected by normal persistence. + #[must_use] + pub const fn purged_count(&self) -> usize { + return self.purged_count; + } + + /// Returns the number of candidates whose `getTransaction` result was missing. + #[must_use] + pub const fn missing_count(&self) -> usize { + return self.missing_count; + } + + /// Returns the number of Store content conflicts. + #[must_use] + pub const fn conflict_count(&self) -> usize { + return self.conflict_count; + } + + /// Returns the number of newly inserted acquisition observations. + #[must_use] + pub const fn observation_inserted_count(&self) -> usize { + return self.observation_inserted_count; + } + + /// Returns the number of acquisition observations already durable. + #[must_use] + pub const fn observation_already_present_count(&self) -> usize { + return self.observation_already_present_count; + } + + /// Returns the number of admitted candidates cancelled before Store submission. + #[must_use] + pub const fn cancelled_count(&self) -> usize { + return self.cancelled_count; + } + /// Returns the number of known candidate outcomes that intentionally block the contiguous frontier. #[must_use] pub const fn hole_count(&self) -> usize { @@ -76,20 +130,114 @@ impl BackfillExecutionBatch { return self.failure_code; } + /// Returns whether cooperative cancellation stopped admissions or cancelled an admitted pre-Store operation. + #[must_use] + pub const fn was_cancelled(&self) -> bool { + return self.cancelled; + } + /// Returns whether discovery or candidate outcomes left the pass incomplete. #[must_use] pub const fn is_partial(&self) -> bool { - return self.discovery_partial || self.hole_count != 0 || self.failure_code.is_some(); + return self.discovery_partial || self.hole_count != 0 || self.failure_code.is_some() || self.cancelled; + } +} + +/// Internal latest-value execution facts published after each known candidate completion. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct BackfillExecutionProgress { + admitted_count: usize, + finished_count: usize, + inserted_count: usize, + already_present_count: usize, + purged_count: usize, + missing_count: usize, + conflict_count: usize, + observation_inserted_count: usize, + observation_already_present_count: usize, + cancelled_count: usize, + hole_count: usize, + maximum_in_flight: usize, + contiguous_completed: usize, + checkpoint: crate::BackfillCheckpoint, +} + +impl BackfillExecutionProgress { + /// Returns the cumulative admission count. + pub(crate) const fn admitted_count(&self) -> usize { + return self.admitted_count; + } + + /// Returns the cumulative finished count. + pub(crate) const fn finished_count(&self) -> usize { + return self.finished_count; + } + + /// Returns the cumulative inserted entity count. + pub(crate) const fn inserted_count(&self) -> usize { + return self.inserted_count; + } + + /// Returns the cumulative already-present entity count. + pub(crate) const fn already_present_count(&self) -> usize { + return self.already_present_count; + } + + /// Returns the cumulative purged entity count. + pub(crate) const fn purged_count(&self) -> usize { + return self.purged_count; + } + + /// Returns the cumulative missing count. + pub(crate) const fn missing_count(&self) -> usize { + return self.missing_count; + } + + /// Returns the cumulative conflict count. + pub(crate) const fn conflict_count(&self) -> usize { + return self.conflict_count; + } + + /// Returns the cumulative inserted observation count. + pub(crate) const fn observation_inserted_count(&self) -> usize { + return self.observation_inserted_count; + } + + /// Returns the cumulative already-present observation count. + pub(crate) const fn observation_already_present_count(&self) -> usize { + return self.observation_already_present_count; + } + + /// Returns the cumulative cancelled candidate count. + pub(crate) const fn cancelled_count(&self) -> usize { + return self.cancelled_count; + } + + /// Returns the cumulative hole count. + pub(crate) const fn hole_count(&self) -> usize { + return self.hole_count; + } + + /// Returns the maximum observed in-flight count. + pub(crate) const fn maximum_in_flight(&self) -> usize { + return self.maximum_in_flight; + } + + /// Returns the cumulative safe contiguous prefix represented by the progress checkpoint. + pub(crate) const fn contiguous_completed(&self) -> usize { + return self.contiguous_completed; + } + + /// Returns the safe checkpoint proven at this progress position. + pub(crate) const fn checkpoint(&self) -> &crate::BackfillCheckpoint { + return &self.checkpoint; } } /// Executes one bounded discovered candidate set with request-owned hydration concurrency. /// -/// At most `request.hydration_concurrency()` candidate futures are kept in flight. Candidate -/// completions may arrive out of order, but the checkpoint advances only across the contiguous -/// durable prefix. `Missing` remains a non-fatal hole. Conflict or any Transport/conversion/Store -/// error stops new admissions immediately; work already admitted is drained before this function -/// returns so a submitted Store operation is never silently abandoned by the batch coordinator. +/// This compatibility entry point has no external cancellation handle. Transport owns retries and +/// endpoint selection; Store submissions are always awaited to completion. pub async fn execute_backfill_discovery( transport: &ksp_onchain_transport_lib::HttpTransportPool, store: &ksp_store_lib::Store, @@ -100,8 +248,21 @@ pub async fn execute_backfill_discovery( if let std::result::Result::Err(error) = validation { return std::result::Result::Err(error); } - let processor = RuntimeCandidateProcessor { transport, store, request }; - return execute_with_processor(&processor, request, discovery).await; + let processor = RuntimeCandidateProcessor { transport, store, request, cancellation: std::option::Option::None }; + return execute_with_processor(&processor, request, discovery, std::option::Option::None, std::option::Option::None).await; +} + +/// Executes one bounded discovery with cooperative cancellation and concrete progress publication. +pub(crate) async fn execute_backfill_discovery_cancellable( + transport: &ksp_onchain_transport_lib::HttpTransportPool, + store: &ksp_store_lib::Store, + request: &crate::BackfillRequest, + discovery: &crate::BackfillDiscovery, + cancellation: &crate::BackfillCancellationSignal, + publisher: &crate::BackfillRuntimePublisher, +) -> ksp_core_lib::Result { + let processor = RuntimeCandidateProcessor { transport, store, request, cancellation: std::option::Option::Some(cancellation) }; + return execute_with_processor(&processor, request, discovery, std::option::Option::Some(cancellation), std::option::Option::Some(publisher)).await; } type CandidateProcessFuture<'a> = @@ -115,6 +276,7 @@ struct RuntimeCandidateProcessor<'a> { transport: &'a ksp_onchain_transport_lib::HttpTransportPool, store: &'a ksp_store_lib::Store, request: &'a crate::BackfillRequest, + cancellation: std::option::Option<&'a crate::BackfillCancellationSignal>, } impl CandidateProcessor for RuntimeCandidateProcessor<'_> { @@ -124,20 +286,31 @@ impl CandidateProcessor for RuntimeCandidateProcessor<'_> { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let hydration = crate::hydrate_backfill_candidate(self.transport, self.request, candidate, received_at).await; + let hydration_future = crate::hydrate_backfill_candidate(self.transport, self.request, candidate, received_at); + let hydration = match self.cancellation { + std::option::Option::Some(signal) => signal.run_cancellable(hydration_future).await, + std::option::Option::None => hydration_future.await, + }; let hydration = match hydration { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; + if self.cancellation.is_some_and(crate::BackfillCancellationSignal::is_requested) { + return std::result::Result::Err(cancelled_error()); + } + // Deliberately not cancellable once Store submission begins: a submitted durable write is drained. return crate::persist_backfill_hydration(self.store, hydration).await; }); } } +#[allow(clippy::too_many_lines)] async fn execute_with_processor

( processor: &P, request: &crate::BackfillRequest, discovery: &crate::BackfillDiscovery, + cancellation: std::option::Option<&crate::BackfillCancellationSignal>, + publisher: std::option::Option<&crate::BackfillRuntimePublisher>, ) -> ksp_core_lib::Result where P: CandidateProcessor, @@ -161,12 +334,53 @@ where let mut admitted_count = 0_usize; let mut finished_count = 0_usize; let mut durable_count = 0_usize; + let mut inserted_count = 0_usize; + let mut already_present_count = 0_usize; + let mut purged_count = 0_usize; + let mut missing_count = 0_usize; + let mut conflict_count = 0_usize; + let mut observation_inserted_count = 0_usize; + let mut observation_already_present_count = 0_usize; + let mut cancelled_count = 0_usize; let mut hole_count = 0_usize; let mut maximum_in_flight = 0_usize; let mut failure_code = std::option::Option::::None; + let mut cancellation_requested = false; let mut in_flight = futures_util::stream::FuturesUnordered::new(); loop { - while failure_code.is_none() && next_index < discovery.candidates().len() && in_flight.len() < request.hydration_concurrency() { + if !cancellation_requested && cancellation.is_some_and(crate::BackfillCancellationSignal::is_requested) { + cancellation_requested = true; + let progress = progress_from_state( + request, + discovery, + &frontier, + admitted_count, + finished_count, + inserted_count, + already_present_count, + purged_count, + missing_count, + conflict_count, + observation_inserted_count, + observation_already_present_count, + cancelled_count, + hole_count, + maximum_in_flight, + ); + let progress = match progress { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let published = publish_progress(publisher, &progress, true, true); + if let std::result::Result::Err(error) = published { + return std::result::Result::Err(error); + } + } + while failure_code.is_none() + && !cancellation_requested + && next_index < discovery.candidates().len() + && in_flight.len() < request.hydration_concurrency() + { let index = next_index; let candidate = &discovery.candidates()[index]; let future = processor.process(candidate); @@ -191,6 +405,19 @@ where }; match result { std::result::Result::Ok(outcome) => { + let classified = classify_persistence( + &outcome, + &mut inserted_count, + &mut already_present_count, + &mut purged_count, + &mut missing_count, + &mut conflict_count, + &mut observation_inserted_count, + &mut observation_already_present_count, + ); + if let std::result::Result::Err(error) = classified { + return std::result::Result::Err(error); + } if persistence_advances_frontier(&outcome) { let marked = frontier.mark_durable(index); if let std::result::Result::Err(error) = marked { @@ -215,11 +442,43 @@ where std::result::Result::Ok(value) => value, std::result::Result::Err(increment_error) => return std::result::Result::Err(increment_error), }; - if failure_code.is_none() { + if error.code() == crate::ERROR_CODE_BACKFILL_CANCELLED { + cancelled_count = match checked_increment(cancelled_count, "cancelled_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(increment_error) => return std::result::Result::Err(increment_error), + }; + cancellation_requested = true; + } else if failure_code.is_none() { failure_code = std::option::Option::Some(error.code()); } }, } + let progress = progress_from_state( + request, + discovery, + &frontier, + admitted_count, + finished_count, + inserted_count, + already_present_count, + purged_count, + missing_count, + conflict_count, + observation_inserted_count, + observation_already_present_count, + cancelled_count, + hole_count, + maximum_in_flight, + ); + let progress = match progress { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let draining = cancellation_requested || failure_code.is_some(); + let published = publish_progress(publisher, &progress, cancellation_requested, draining); + if let std::result::Result::Err(error) = published { + return std::result::Result::Err(error); + } } let checkpoint = match crate::checkpoint_from_frontier(request, discovery, &frontier) { std::result::Result::Ok(value) => value, @@ -230,15 +489,140 @@ where admitted_count, finished_count, durable_count, + inserted_count, + already_present_count, + purged_count, + missing_count, + conflict_count, + observation_inserted_count, + observation_already_present_count, + cancelled_count, hole_count, maximum_in_flight, local_contiguous_completed: frontier.contiguous_completed(), discovery_partial: discovery.is_partial(), + cancelled: cancellation_requested, checkpoint, failure_code, }); } +#[allow(clippy::too_many_arguments)] +fn progress_from_state( + request: &crate::BackfillRequest, + discovery: &crate::BackfillDiscovery, + frontier: &crate::CompletionFrontier, + admitted_count: usize, + finished_count: usize, + inserted_count: usize, + already_present_count: usize, + purged_count: usize, + missing_count: usize, + conflict_count: usize, + observation_inserted_count: usize, + observation_already_present_count: usize, + cancelled_count: usize, + hole_count: usize, + maximum_in_flight: usize, +) -> ksp_core_lib::Result { + let checkpoint = match crate::checkpoint_from_frontier(request, discovery, frontier) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(BackfillExecutionProgress { + admitted_count, + finished_count, + inserted_count, + already_present_count, + purged_count, + missing_count, + conflict_count, + observation_inserted_count, + observation_already_present_count, + cancelled_count, + hole_count, + maximum_in_flight, + contiguous_completed: checkpoint.completed_prefix(), + checkpoint, + }); +} + +fn publish_progress( + publisher: std::option::Option<&crate::BackfillRuntimePublisher>, + progress: &BackfillExecutionProgress, + cancelling: bool, + draining: bool, +) -> ksp_core_lib::Result<()> { + if let std::option::Option::Some(publisher) = publisher { + let published = publisher.publish_execution_progress(progress, cancelling, draining); + if let std::result::Result::Err(error) = published { + return std::result::Result::Err(error); + } + } + return std::result::Result::Ok(()); +} + +#[allow(clippy::too_many_arguments)] +fn classify_persistence( + outcome: &crate::BackfillPersistenceOutcome, + inserted_count: &mut usize, + already_present_count: &mut usize, + purged_count: &mut usize, + missing_count: &mut usize, + conflict_count: &mut usize, + observation_inserted_count: &mut usize, + observation_already_present_count: &mut usize, +) -> ksp_core_lib::Result<()> { + match outcome.entity() { + crate::BackfillEntityPersistence::Inserted => { + *inserted_count = match checked_increment(*inserted_count, "inserted_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + }, + crate::BackfillEntityPersistence::AlreadyPresent => { + *already_present_count = match checked_increment(*already_present_count, "already_present_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + }, + crate::BackfillEntityPersistence::SkippedPurged => { + *purged_count = match checked_increment(*purged_count, "purged_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + }, + crate::BackfillEntityPersistence::Missing => { + *missing_count = match checked_increment(*missing_count, "missing_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + }, + crate::BackfillEntityPersistence::Conflict => { + *conflict_count = match checked_increment(*conflict_count, "conflict_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + }, + } + match outcome.observation() { + crate::BackfillObservationPersistence::Inserted => { + *observation_inserted_count = match checked_increment(*observation_inserted_count, "observation_inserted_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + }, + crate::BackfillObservationPersistence::AlreadyPresent => { + *observation_already_present_count = match checked_increment(*observation_already_present_count, "observation_already_present_count") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + }, + crate::BackfillObservationPersistence::NotRecorded | crate::BackfillObservationPersistence::NotApplicable => {}, + } + return std::result::Result::Ok(()); +} + fn persistence_advances_frontier(outcome: &crate::BackfillPersistenceOutcome) -> bool { return matches!( outcome.entity(), @@ -263,6 +647,10 @@ fn current_raw_timestamp() -> ksp_core_lib::Result return ksp_store_lib::RawTimestamp::from_unix_millis(millis); } +fn cancelled_error() -> ksp_core_lib::Error { + return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_CANCELLED, "Backfill candidate cancelled before durable Store submission"); +} + fn execution_error(field: &'static str) -> ksp_core_lib::Error { return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_EXECUTION_INVALID, "invalid bounded Backfill execution state").with_context("field", field); } diff --git a/crates/ksp-job-backfill-lib/src/lib.rs b/crates/ksp-job-backfill-lib/src/lib.rs index 7018652..a67b6c2 100644 --- a/crates/ksp-job-backfill-lib/src/lib.rs +++ b/crates/ksp-job-backfill-lib/src/lib.rs @@ -1,5 +1,5 @@ // file: crates/ksp-job-backfill-lib/src/lib.rs -// version: 4 +// version: 5 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -11,8 +11,8 @@ //! `getSignaturesForAddress` pagination and canonical RAW v1 conversion through observed //! `getTransaction`. Transport retains provider/endpoint selection and retry; Store retains //! durable idempotence through the atomic Store facade. This tranche also owns bounded concurrent -//! candidate execution and caller-owned contiguous checkpoints. Cancellation and concrete latest-value -//! snapshots are added by later v0.3.6 tranches. +//! candidate execution, caller-owned contiguous checkpoints, cooperative cancellation and concrete +//! latest-value snapshots for external listeners. mod checkpoint; mod constants; @@ -22,6 +22,7 @@ mod error; mod execution; mod persistence; mod request; +mod runtime; /// Opaque caller-owned checkpoint for one controlled Backfill resumption. pub use self::checkpoint::BackfillCheckpoint; @@ -59,6 +60,8 @@ pub use self::error::ERROR_CODE_BACKFILL_PERSISTENCE_INVALID; pub use self::error::ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID; /// Error code used when one Backfill request violates its bounded admission contract. pub use self::error::ERROR_CODE_BACKFILL_REQUEST_INVALID; +/// Error code used when the concrete Backfill runtime reaches an impossible lifecycle or notification state. +pub use self::error::ERROR_CODE_BACKFILL_RUNTIME_INVALID; /// Error code used when one transaction signature text violates the bounded Base58-shape contract. pub use self::error::ERROR_CODE_BACKFILL_SIGNATURE_INVALID; /// Bounded result of one concurrent Backfill candidate execution pass. @@ -97,6 +100,18 @@ pub use self::request::MAX_BACKFILL_PAGES; pub use self::request::MAX_BACKFILL_SIGNATURE_TEXT_BYTES; /// Minimum Base58 text length possible for one canonical 64-byte Solana signature. pub use self::request::MIN_BACKFILL_SIGNATURE_TEXT_BYTES; +/// Stable Job kind code used by the concrete historical RAW transaction Backfill runtime. +pub use self::runtime::BACKFILL_JOB_KIND_CODE; +/// Cloneable external control handle for one concrete Backfill runtime. +pub use self::runtime::BackfillJobHandle; +/// Current concrete phase of one historical RAW transaction Backfill Job. +pub use self::runtime::BackfillJobPhase; +/// Concrete single-run Backfill coordinator paired with a cloneable control/snapshot handle. +pub use self::runtime::BackfillJobRuntime; +/// Complete safe latest-value snapshot of one concrete historical RAW transaction Backfill Job. +pub use self::runtime::BackfillJobSnapshot; +/// Cloneable runtime-neutral-facing latest-value source for concrete Backfill snapshots. +pub use self::runtime::BackfillSnapshotSource; /// Internal contiguous completion frontier used by bounded execution. pub(crate) use self::checkpoint::CompletionFrontier; @@ -114,3 +129,19 @@ pub(crate) use self::checkpoint::validate_request_checkpoint; pub(crate) use self::constants::TRACING_TARGET; /// Exact private Base58 decoder shared by the public signature wrapper and hydration path. pub(crate) use self::conversion::decode_backfill_signature; +/// Internal cancellable discovery path used by the concrete runtime. +pub(crate) use self::discovery::discover_backfill_candidates_cancellable; +/// Internal cooperative cancellation code used to distinguish cancellation from failure. +pub(crate) use self::error::ERROR_CODE_BACKFILL_CANCELLED; +/// Internal execution progress facts fed into the concrete latest-value source. +pub(crate) use self::execution::BackfillExecutionProgress; +/// Internal cancellable execution path used by the concrete runtime. +pub(crate) use self::execution::execute_backfill_discovery_cancellable; +/// Internal concrete cancellation signal shared by discovery and execution. +pub(crate) use self::runtime::BackfillCancellationSignal; +/// Internal atomic terminal/cancellation arbitration shared by runtime tests and coordinator. +pub(crate) use self::runtime::BackfillRuntimeControl; +/// Internal execution progress publisher feeding the latest-value snapshot source. +pub(crate) use self::runtime::BackfillRuntimePublisher; +/// Internal terminal race result used by coordinator canaries. +pub(crate) use self::runtime::TerminalClaim; diff --git a/crates/ksp-job-backfill-lib/src/runtime.rs b/crates/ksp-job-backfill-lib/src/runtime.rs new file mode 100644 index 0000000..9781b5e --- /dev/null +++ b/crates/ksp-job-backfill-lib/src/runtime.rs @@ -0,0 +1,715 @@ +// file: crates/ksp-job-backfill-lib/src/runtime.rs +// version: 2 + +/// Stable Job kind code used by the concrete historical RAW transaction Backfill runtime. +pub const BACKFILL_JOB_KIND_CODE: &str = "solana.raw_transaction.backfill"; + +const CONTROL_ACTIVE: u8 = 0; +const CONTROL_CANCELLATION_REQUESTED: u8 = 1; +const CONTROL_CANCELLED: u8 = 3; +const CONTROL_COMPLETED: u8 = 2; +const CONTROL_FAILED: u8 = 4; + +/// Current concrete phase of one historical RAW transaction Backfill Job. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +#[non_exhaustive] +pub enum BackfillJobPhase { + /// Runtime was created but execution has not started. + Created, + /// Candidate discovery is in progress. + Discovering, + /// Candidate hydration and persistence are being admitted and completed. + Executing, + /// Cancellation or a fatal result stopped admission while already submitted work is draining. + Draining, + /// No more work can be admitted and the Job is terminal. + Finished, +} + +impl BackfillJobPhase { + /// Returns the stable safe code for this concrete runtime phase. + #[must_use] + pub const fn code(self) -> &'static str { + return match self { + Self::Created => "created", + Self::Discovering => "discovering", + Self::Executing => "executing", + Self::Draining => "draining", + Self::Finished => "finished", + }; + } +} + +/// Complete safe latest-value snapshot of one concrete historical RAW transaction Backfill Job. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackfillJobSnapshot { + phase: BackfillJobPhase, + scope_kind: crate::BackfillScopeKind, + discovery_boundary: std::option::Option, + candidates_selected: usize, + candidates_admitted: usize, + candidates_finished: usize, + entities_inserted: usize, + entities_existing: usize, + entities_purged: usize, + missing: usize, + conflicts: usize, + observations_inserted: usize, + observations_existing: usize, + cancelled_candidates: usize, + holes: usize, + maximum_in_flight: usize, + contiguous_completed: usize, + checkpoint: std::option::Option, + failure_code: std::option::Option, +} + +impl BackfillJobSnapshot { + fn initial(request: &crate::BackfillRequest) -> Self { + return Self { + phase: BackfillJobPhase::Created, + scope_kind: request.scope().kind(), + discovery_boundary: std::option::Option::None, + candidates_selected: 0, + candidates_admitted: 0, + candidates_finished: 0, + entities_inserted: 0, + entities_existing: 0, + entities_purged: 0, + missing: 0, + conflicts: 0, + observations_inserted: 0, + observations_existing: 0, + cancelled_candidates: 0, + holes: 0, + maximum_in_flight: 0, + contiguous_completed: request.checkpoint().map_or(0, crate::BackfillCheckpoint::completed_prefix), + checkpoint: request.checkpoint().cloned(), + failure_code: std::option::Option::None, + }; + } + + /// Returns the concrete execution phase represented by this snapshot. + #[must_use] + pub const fn phase(&self) -> BackfillJobPhase { + return self.phase; + } + + /// Returns the safe scope category without rendering address or signature payloads. + #[must_use] + pub const fn scope_kind(&self) -> crate::BackfillScopeKind { + return self.scope_kind; + } + + /// Returns the discovery boundary once candidate selection has completed. + #[must_use] + pub const fn discovery_boundary(&self) -> std::option::Option { + return self.discovery_boundary; + } + + /// Returns the bounded candidate count selected by discovery. + #[must_use] + pub const fn candidates_selected(&self) -> usize { + return self.candidates_selected; + } + + /// Returns the number of candidates admitted into hydration. + #[must_use] + pub const fn candidates_admitted(&self) -> usize { + return self.candidates_admitted; + } + + /// Returns the number of admitted candidates that reached a known coordinator result. + #[must_use] + pub const fn candidates_finished(&self) -> usize { + return self.candidates_finished; + } + + /// Returns the number of newly inserted canonical RAW transactions. + #[must_use] + pub const fn entities_inserted(&self) -> usize { + return self.entities_inserted; + } + + /// Returns the number of identical canonical RAW transactions already present. + #[must_use] + pub const fn entities_existing(&self) -> usize { + return self.entities_existing; + } + + /// Returns the number of durable purge tombstones respected by normal Backfill persistence. + #[must_use] + pub const fn entities_purged(&self) -> usize { + return self.entities_purged; + } + + /// Returns the number of `getTransaction = null` candidates. + #[must_use] + pub const fn missing(&self) -> usize { + return self.missing; + } + + /// Returns the number of Store content conflicts observed by this run. + #[must_use] + pub const fn conflicts(&self) -> usize { + return self.conflicts; + } + + /// Returns the number of newly inserted acquisition observations. + #[must_use] + pub const fn observations_inserted(&self) -> usize { + return self.observations_inserted; + } + + /// Returns the number of acquisition observations already durable. + #[must_use] + pub const fn observations_existing(&self) -> usize { + return self.observations_existing; + } + + /// Returns the number of admitted candidates cancelled before Store submission. + #[must_use] + pub const fn cancelled_candidates(&self) -> usize { + return self.cancelled_candidates; + } + + /// Returns the number of known candidate outcomes blocking the contiguous frontier. + #[must_use] + pub const fn holes(&self) -> usize { + return self.holes; + } + + /// Returns the greatest observed concurrent candidate count. + #[must_use] + pub const fn maximum_in_flight(&self) -> usize { + return self.maximum_in_flight; + } + + /// Returns the cumulative safe contiguous completion prefix represented by the current checkpoint. + #[must_use] + pub const fn contiguous_completed(&self) -> usize { + return self.contiguous_completed; + } + + /// Returns the latest caller-owned safe checkpoint, when one exists. + #[must_use] + pub const fn checkpoint(&self) -> std::option::Option<&crate::BackfillCheckpoint> { + return self.checkpoint.as_ref(); + } + + /// Returns the stable fatal error code retained by a failed Job, when one exists. + #[must_use] + pub const fn failure_code(&self) -> std::option::Option { + return self.failure_code; + } +} + +/// Cloneable runtime-neutral-facing latest-value source for concrete Backfill snapshots. +#[derive(Clone)] +pub struct BackfillSnapshotSource { + receiver: tokio::sync::watch::Receiver>, +} + +impl ksp_job_api::JobSnapshotSource for BackfillSnapshotSource { + type Snapshot = BackfillJobSnapshot; + + fn current(&self) -> ksp_job_api::JobNotification { + return self.receiver.borrow().clone(); + } + + fn wait_for_change(&self, observed: ksp_job_api::JobNotificationSequence) -> ksp_job_api::JobSnapshotFuture<'_, Self::Snapshot> { + let mut receiver = self.receiver.clone(); + return std::boxed::Box::pin(async move { + loop { + let current = receiver.borrow().clone(); + if current.sequence().is_after(observed) || current.state().is_terminal() { + return current; + } + let changed = receiver.changed().await; + if changed.is_err() { + return receiver.borrow().clone(); + } + } + }); + } +} + +impl std::fmt::Debug for BackfillSnapshotSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let current = self.receiver.borrow(); + return formatter.debug_struct("BackfillSnapshotSource").field("sequence", ¤t.sequence()).field("state", ¤t.state()).finish(); + } +} + +/// Cloneable external control handle for one concrete Backfill runtime. +#[derive(Clone)] +pub struct BackfillJobHandle { + control: BackfillRuntimeControl, + snapshots: BackfillSnapshotSource, +} + +impl BackfillJobHandle { + /// Requests cooperative cancellation and returns `true` only when accepted before terminal publication. + #[must_use] + pub fn cancel(&self) -> bool { + return self.control.request_cancellation(); + } + + /// Returns an independent latest-value snapshot source for one listener. + #[must_use] + pub fn snapshots(&self) -> BackfillSnapshotSource { + return self.snapshots.clone(); + } + + /// Returns whether cooperative cancellation has been accepted for this non-terminal Job. + #[must_use] + pub fn is_cancellation_requested(&self) -> bool { + return self.control.is_cancellation_requested(); + } +} + +impl std::fmt::Debug for BackfillJobHandle { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return formatter + .debug_struct("BackfillJobHandle") + .field("cancellation_requested", &self.is_cancellation_requested()) + .field("snapshot", &::current(&self.snapshots)) + .finish(); + } +} + +/// Concrete single-run Backfill coordinator paired with a cloneable control/snapshot handle. +pub struct BackfillJobRuntime { + request: crate::BackfillRequest, + control: BackfillRuntimeControl, + cancellation: BackfillCancellationSignal, + publisher: BackfillRuntimePublisher, + handle: BackfillJobHandle, +} + +impl BackfillJobRuntime { + /// Creates one runtime in `Created` state and its stable latest-value channel. + pub fn new(request: crate::BackfillRequest) -> ksp_core_lib::Result { + let kind = match ksp_job_api::JobKindCode::new(BACKFILL_JOB_KIND_CODE) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let initial_snapshot = BackfillJobSnapshot::initial(&request); + let initial = ksp_job_api::JobNotification::new( + request.job_id().clone(), + kind.clone(), + ksp_job_api::JobNotificationSequence::initial(), + ksp_job_api::JobState::Created, + initial_snapshot, + ); + let (snapshot_sender, snapshot_receiver) = tokio::sync::watch::channel(initial); + let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false); + let control = BackfillRuntimeControl::new(cancel_sender); + let cancellation = BackfillCancellationSignal::new(control.token(), cancel_receiver); + let snapshots = BackfillSnapshotSource { receiver: snapshot_receiver }; + let handle = BackfillJobHandle { control: control.clone(), snapshots: snapshots.clone() }; + let publisher = BackfillRuntimePublisher { sender: snapshot_sender, id: request.job_id().clone(), kind }; + return std::result::Result::Ok(Self { request, control, cancellation, publisher, handle }); + } + + /// Returns a cloneable control and latest-value observation handle before the runtime is moved into execution. + #[must_use] + pub fn handle(&self) -> BackfillJobHandle { + return self.handle.clone(); + } + + /// Runs discovery, bounded execution, cooperative cancellation and terminal snapshot publication. + pub async fn run( + self, + transport: &ksp_onchain_transport_lib::HttpTransportPool, + store: &ksp_store_lib::Store, + ) -> ksp_core_lib::Result { + if self.cancellation.is_requested() { + let claimed = self.control.claim_normal_terminal(); + if claimed != TerminalClaim::Cancelled { + return std::result::Result::Err(runtime_error("terminal.pre_start")); + } + return self.publisher.publish_cancelled_from_created(); + } + let started = self.publisher.publish_running(BackfillJobPhase::Discovering); + if let std::result::Result::Err(error) = started { + self.control.claim_failed(); + return std::result::Result::Err(error); + } + let discovery = crate::discover_backfill_candidates_cancellable(transport, &self.request, &self.cancellation).await; + let discovery = match discovery { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + if error.code() == crate::ERROR_CODE_BACKFILL_CANCELLED { + return self.finish_cancelled(); + } + self.control.claim_failed(); + let published = self.publisher.publish_failed(error.code()); + if let std::result::Result::Err(publish_error) = published { + return std::result::Result::Err(publish_error); + } + return std::result::Result::Err(error); + }, + }; + let published = self.publisher.publish_discovery(&discovery); + if let std::result::Result::Err(error) = published { + self.control.claim_failed(); + return std::result::Result::Err(error); + } + let batch = crate::execute_backfill_discovery_cancellable(transport, store, &self.request, &discovery, &self.cancellation, &self.publisher).await; + let batch = match batch { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + self.control.claim_failed(); + let published = self.publisher.publish_failed(error.code()); + if let std::result::Result::Err(publish_error) = published { + return std::result::Result::Err(publish_error); + } + return std::result::Result::Err(error); + }, + }; + if let std::option::Option::Some(code) = batch.failure_code() { + self.control.claim_failed(); + let published = self.publisher.publish_batch_terminal(&batch, ksp_job_api::JobState::Failed, std::option::Option::Some(code)); + if let std::result::Result::Err(error) = published { + return std::result::Result::Err(error); + } + return std::result::Result::Err(ksp_core_lib::Error::new(code, "Backfill execution reached a fatal candidate result")); + } + let terminal = self.control.claim_normal_terminal(); + return match terminal { + TerminalClaim::Cancelled => { + let cancelling = self.publisher.publish_batch_cancelling(&batch); + if let std::result::Result::Err(error) = cancelling { + return std::result::Result::Err(error); + } + self.publisher.publish_batch_terminal(&batch, ksp_job_api::JobState::Cancelled, std::option::Option::None) + }, + TerminalClaim::Completed => { + let completion = if batch.is_partial() { ksp_job_api::JobCompletion::Partial } else { ksp_job_api::JobCompletion::Complete }; + self.publisher.publish_batch_terminal(&batch, ksp_job_api::JobState::Completed(completion), std::option::Option::None) + }, + TerminalClaim::Failed => std::result::Result::Err(runtime_error("terminal.failed")), + }; + } + + fn finish_cancelled(&self) -> ksp_core_lib::Result { + let terminal = self.control.claim_normal_terminal(); + if terminal != TerminalClaim::Cancelled { + return std::result::Result::Err(runtime_error("terminal.cancelled")); + } + let cancelling = self.publisher.publish_cancelling(); + if let std::result::Result::Err(error) = cancelling { + return std::result::Result::Err(error); + } + return self.publisher.publish_terminal(ksp_job_api::JobState::Cancelled, std::option::Option::None); + } +} + +/// Internal atomic terminal/cancellation arbitration shared by runtime and external handle. +#[derive(Clone)] +pub(crate) struct BackfillRuntimeControl { + state: std::sync::Arc, + token: ksp_job_api::JobCancellationToken, + cancel_sender: tokio::sync::watch::Sender, +} + +impl BackfillRuntimeControl { + /// Creates one active control state paired with the cancellation wake channel. + pub(crate) fn new(cancel_sender: tokio::sync::watch::Sender) -> Self { + return Self { + state: std::sync::Arc::new(std::sync::atomic::AtomicU8::new(CONTROL_ACTIVE)), + token: ksp_job_api::JobCancellationToken::new(), + cancel_sender, + }; + } + + /// Returns the runtime-neutral cancellation token mirrored by this control. + pub(crate) fn token(&self) -> ksp_job_api::JobCancellationToken { + return self.token.clone(); + } + + /// Atomically accepts the first pre-terminal cancellation request. + pub(crate) fn request_cancellation(&self) -> bool { + let accepted = self + .state + .compare_exchange(CONTROL_ACTIVE, CONTROL_CANCELLATION_REQUESTED, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire) + .is_ok(); + if accepted && self.token.cancel() { + let _previous = self.cancel_sender.send_replace(true); + } + return accepted; + } + + /// Returns whether cooperative cancellation was accepted. + pub(crate) fn is_cancellation_requested(&self) -> bool { + return self.token.is_cancellation_requested(); + } + + /// Atomically resolves the completion-versus-cancellation terminal race. + pub(crate) fn claim_normal_terminal(&self) -> TerminalClaim { + let completed = + self.state.compare_exchange(CONTROL_ACTIVE, CONTROL_COMPLETED, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire); + if completed.is_ok() { + return TerminalClaim::Completed; + } + let cancelled = self.state.compare_exchange( + CONTROL_CANCELLATION_REQUESTED, + CONTROL_CANCELLED, + std::sync::atomic::Ordering::AcqRel, + std::sync::atomic::Ordering::Acquire, + ); + if cancelled.is_ok() { + return TerminalClaim::Cancelled; + } + return TerminalClaim::Failed; + } + + /// Marks a non-terminal control as failed, overriding a pending cancellation request. + pub(crate) fn claim_failed(&self) { + loop { + let state = self.state.load(std::sync::atomic::Ordering::Acquire); + if matches!(state, CONTROL_COMPLETED | CONTROL_CANCELLED | CONTROL_FAILED) { + return; + } + let changed = self.state.compare_exchange(state, CONTROL_FAILED, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire); + if changed.is_ok() { + return; + } + } + } +} + +/// Internal wakeable cancellation signal used only around abandonable pre-Store futures. +#[derive(Clone)] +pub(crate) struct BackfillCancellationSignal { + token: ksp_job_api::JobCancellationToken, + receiver: tokio::sync::watch::Receiver, +} + +impl BackfillCancellationSignal { + /// Creates one signal from the runtime-neutral token and Tokio wake receiver. + pub(crate) fn new(token: ksp_job_api::JobCancellationToken, receiver: tokio::sync::watch::Receiver) -> Self { + return Self { token, receiver }; + } + + /// Returns whether cancellation has been requested. + pub(crate) fn is_requested(&self) -> bool { + return self.token.is_cancellation_requested(); + } + + /// Runs one abandonable operation until it completes or cancellation wins. + pub(crate) async fn run_cancellable(&self, operation: F) -> ksp_core_lib::Result + where + F: std::future::Future>, + { + if self.is_requested() { + return std::result::Result::Err(cancelled_error()); + } + let mut receiver = self.receiver.clone(); + return tokio::select! { + biased; + _ = wait_for_cancellation(&self.token, &mut receiver) => std::result::Result::Err(cancelled_error()), + result = operation => result, + }; + } +} + +async fn wait_for_cancellation(token: &ksp_job_api::JobCancellationToken, receiver: &mut tokio::sync::watch::Receiver) { + loop { + if token.is_cancellation_requested() || *receiver.borrow() { + return; + } + let changed = receiver.changed().await; + if changed.is_err() { + return; + } + } +} + +/// Internal result of atomically claiming a normal terminal state. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum TerminalClaim { + Completed, + Cancelled, + Failed, +} + +/// Internal latest-value publisher owning the concrete Backfill notification stream. +#[derive(Clone)] +pub(crate) struct BackfillRuntimePublisher { + sender: tokio::sync::watch::Sender>, + id: ksp_job_api::JobId, + kind: ksp_job_api::JobKindCode, +} + +impl BackfillRuntimePublisher { + /// Publishes one non-terminal running phase. + pub(crate) fn publish_running(&self, phase: BackfillJobPhase) -> ksp_core_lib::Result { + return self.publish_with(ksp_job_api::JobState::Running, |snapshot| { + snapshot.phase = phase; + }); + } + + /// Publishes the complete bounded discovery result as the current execution snapshot. + pub(crate) fn publish_discovery(&self, discovery: &crate::BackfillDiscovery) -> ksp_core_lib::Result { + return self.publish_with(ksp_job_api::JobState::Running, |snapshot| { + snapshot.phase = BackfillJobPhase::Executing; + snapshot.discovery_boundary = std::option::Option::Some(discovery.boundary()); + snapshot.candidates_selected = discovery.candidates().len(); + }); + } + + /// Publishes one coalescable execution progress value. + pub(crate) fn publish_execution_progress( + &self, + progress: &crate::BackfillExecutionProgress, + cancelling: bool, + draining: bool, + ) -> ksp_core_lib::Result { + let state = if cancelling { ksp_job_api::JobState::Cancelling } else { ksp_job_api::JobState::Running }; + return self.publish_with(state, |snapshot| { + snapshot.phase = if draining { BackfillJobPhase::Draining } else { BackfillJobPhase::Executing }; + apply_progress(snapshot, progress); + }); + } + + /// Publishes cancellation observation before terminal cancellation. + pub(crate) fn publish_cancelling(&self) -> ksp_core_lib::Result { + return self.publish_with(ksp_job_api::JobState::Cancelling, |snapshot| { + snapshot.phase = BackfillJobPhase::Draining; + }); + } + + /// Publishes the drained batch state while cancellation is terminalizing. + pub(crate) fn publish_batch_cancelling(&self, batch: &crate::BackfillExecutionBatch) -> ksp_core_lib::Result { + return self.publish_with(ksp_job_api::JobState::Cancelling, |snapshot| { + snapshot.phase = BackfillJobPhase::Draining; + apply_batch(snapshot, batch); + }); + } + + /// Publishes one terminal state together with the fully drained execution batch. + pub(crate) fn publish_batch_terminal( + &self, + batch: &crate::BackfillExecutionBatch, + state: ksp_job_api::JobState, + failure_code: std::option::Option, + ) -> ksp_core_lib::Result { + return self.publish_with(state, |snapshot| { + snapshot.phase = BackfillJobPhase::Finished; + snapshot.failure_code = failure_code; + apply_batch(snapshot, batch); + }); + } + + /// Publishes a terminal failure before a batch exists. + pub(crate) fn publish_failed(&self, code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Result { + return self.publish_terminal(ksp_job_api::JobState::Failed, std::option::Option::Some(code)); + } + + /// Publishes a terminal state without a completed execution batch. + pub(crate) fn publish_terminal( + &self, + state: ksp_job_api::JobState, + failure_code: std::option::Option, + ) -> ksp_core_lib::Result { + return self.publish_with(state, |snapshot| { + snapshot.phase = BackfillJobPhase::Finished; + snapshot.failure_code = failure_code; + }); + } + + /// Publishes direct Created-to-Cancelled termination before execution starts. + pub(crate) fn publish_cancelled_from_created(&self) -> ksp_core_lib::Result { + return self.publish_with(ksp_job_api::JobState::Cancelled, |snapshot| { + snapshot.phase = BackfillJobPhase::Finished; + }); + } + + fn publish_with(&self, state: ksp_job_api::JobState, update: F) -> ksp_core_lib::Result + where + F: FnOnce(&mut BackfillJobSnapshot), + { + let current = self.sender.borrow().clone(); + if current.state().is_terminal() { + return std::result::Result::Err(runtime_error("notification.terminal")); + } + if !valid_snapshot_transition(current.state(), state) { + return std::result::Result::Err(runtime_error("notification.transition")); + } + let sequence = match current.sequence().next() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let mut snapshot = current.snapshot().clone(); + update(&mut snapshot); + let notification = ksp_job_api::JobNotification::new(self.id.clone(), self.kind.clone(), sequence, state, snapshot.clone()); + let _previous = self.sender.send_replace(notification); + return std::result::Result::Ok(snapshot); + } +} + +fn valid_snapshot_transition(source: ksp_job_api::JobState, target: ksp_job_api::JobState) -> bool { + if source.is_terminal() { + return false; + } + return matches!( + (source, target), + (ksp_job_api::JobState::Created, ksp_job_api::JobState::Running) + | (ksp_job_api::JobState::Created, ksp_job_api::JobState::Cancelled) + | (ksp_job_api::JobState::Running, ksp_job_api::JobState::Running) + | (ksp_job_api::JobState::Running, ksp_job_api::JobState::Cancelling) + | (ksp_job_api::JobState::Running, ksp_job_api::JobState::Completed(_)) + | (ksp_job_api::JobState::Running, ksp_job_api::JobState::Failed) + | (ksp_job_api::JobState::Cancelling, ksp_job_api::JobState::Cancelling) + | (ksp_job_api::JobState::Cancelling, ksp_job_api::JobState::Cancelled) + | (ksp_job_api::JobState::Cancelling, ksp_job_api::JobState::Failed) + ); +} + +fn apply_progress(snapshot: &mut BackfillJobSnapshot, progress: &crate::BackfillExecutionProgress) { + snapshot.candidates_admitted = progress.admitted_count(); + snapshot.candidates_finished = progress.finished_count(); + snapshot.entities_inserted = progress.inserted_count(); + snapshot.entities_existing = progress.already_present_count(); + snapshot.entities_purged = progress.purged_count(); + snapshot.missing = progress.missing_count(); + snapshot.conflicts = progress.conflict_count(); + snapshot.observations_inserted = progress.observation_inserted_count(); + snapshot.observations_existing = progress.observation_already_present_count(); + snapshot.cancelled_candidates = progress.cancelled_count(); + snapshot.holes = progress.hole_count(); + snapshot.maximum_in_flight = progress.maximum_in_flight(); + snapshot.contiguous_completed = progress.contiguous_completed(); + snapshot.checkpoint = std::option::Option::Some(progress.checkpoint().clone()); +} + +fn apply_batch(snapshot: &mut BackfillJobSnapshot, batch: &crate::BackfillExecutionBatch) { + snapshot.candidates_admitted = batch.admitted_count(); + snapshot.candidates_finished = batch.finished_count(); + snapshot.entities_inserted = batch.inserted_count(); + snapshot.entities_existing = batch.already_present_count(); + snapshot.entities_purged = batch.purged_count(); + snapshot.missing = batch.missing_count(); + snapshot.conflicts = batch.conflict_count(); + snapshot.observations_inserted = batch.observation_inserted_count(); + snapshot.observations_existing = batch.observation_already_present_count(); + snapshot.cancelled_candidates = batch.cancelled_count(); + snapshot.holes = batch.hole_count(); + snapshot.maximum_in_flight = batch.maximum_in_flight(); + snapshot.contiguous_completed = batch.checkpoint().completed_prefix(); + snapshot.checkpoint = std::option::Option::Some(batch.checkpoint().clone()); +} + +fn cancelled_error() -> ksp_core_lib::Error { + return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_CANCELLED, "Backfill operation cancelled before durable Store submission"); +} + +fn runtime_error(field: &'static str) -> ksp_core_lib::Error { + return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_RUNTIME_INVALID, "invalid concrete Backfill runtime state").with_context("field", field); +} + +#[cfg(test)] +#[path = "../unit_tests/runtime.rs"] +mod tests; diff --git a/crates/ksp-job-backfill-lib/tests/dependency_boundary.rs b/crates/ksp-job-backfill-lib/tests/dependency_boundary.rs index ddc988e..ccee3ad 100644 --- a/crates/ksp-job-backfill-lib/tests/dependency_boundary.rs +++ b/crates/ksp-job-backfill-lib/tests/dependency_boundary.rs @@ -1,10 +1,10 @@ // file: crates/ksp-job-backfill-lib/tests/dependency_boundary.rs -// version: 4 +// version: 6 -//! Dependency firewall canaries through bounded Backfill execution/checkpointing. +//! Dependency firewall canaries through the concrete cancellation and latest-value runtime tranche. #[test] -fn pre_008_manifest_uses_only_planned_ksp_edges_and_private_futures_runtime() { +fn pre_009_manifest_uses_only_planned_ksp_edges_and_private_tokio_runtime() { let manifest = include_str!("../Cargo.toml"); for required in [ "futures-util = { workspace = true, features = [\"std\"] }", @@ -15,6 +15,7 @@ fn pre_008_manifest_uses_only_planned_ksp_edges_and_private_futures_runtime() { "ksp-store-lib = { path = \"../ksp-store-lib\", default-features = false }", "serde_json.workspace = true", "sha2.workspace = true", + "tokio = { workspace = true, features = [\"macros\", \"sync\"] }", ] { assert!(manifest.contains(required), "required Backfill dependency missing: {required}"); } @@ -33,16 +34,13 @@ fn pre_008_manifest_uses_only_planned_ksp_edges_and_private_futures_runtime() { ] { assert!(!manifest.contains(forbidden), "forbidden Backfill dependency present: {forbidden}"); } - let normal_dependencies = match manifest.split("[dev-dependencies]").next() { - std::option::Option::Some(value) => value, - std::option::Option::None => manifest, - }; - assert!(!normal_dependencies.contains("tokio ="), "Tokio must not become a public/normal Backfill dependency in pre.008"); + let root = include_str!("../src/lib.rs"); + assert!(!root.contains("tokio::"), "Tokio implementation types must not leak through the public crate root"); return; } #[test] -fn pre_008_production_sources_keep_transport_store_and_scheduler_ownership_separate() { +fn pre_009_production_sources_keep_transport_store_and_scheduler_ownership_separate() { let neutral_sources = [ include_str!("../src/checkpoint.rs"), include_str!("../src/constants.rs"), @@ -50,6 +48,7 @@ fn pre_008_production_sources_keep_transport_store_and_scheduler_ownership_separ include_str!("../src/error.rs"), include_str!("../src/lib.rs"), include_str!("../src/request.rs"), + include_str!("../src/runtime.rs"), ]; for source in neutral_sources { for forbidden in ["ksp_config_lib::", "ksp_interface_lib::", "ksp_store_api::", "ksp_store_postgres_lib::", "reqwest::", "std::env", "tonic::"] { @@ -83,5 +82,12 @@ fn pre_008_production_sources_keep_transport_store_and_scheduler_ownership_separ for forbidden in ["tokio::spawn", "tokio::time", "retry", "endpoint_name", "provider()", "ForceRehydrate", "get_raw_transaction"] { assert!(!execution.contains(forbidden), "forbidden scheduler/policy ownership detected: {forbidden}"); } + let runtime = include_str!("../src/runtime.rs"); + assert!(runtime.contains("tokio::sync::watch")); + assert!(runtime.contains("JobSnapshotSource")); + assert!(runtime.contains("JobCancellationToken")); + assert!(!runtime.contains("tokio::spawn")); + assert!(!runtime.contains("tokio::time")); + assert!(!runtime.contains("reqwest::")); return; } diff --git a/crates/ksp-job-backfill-lib/tests/public_api.rs b/crates/ksp-job-backfill-lib/tests/public_api.rs index 148958b..d17d213 100644 --- a/crates/ksp-job-backfill-lib/tests/public_api.rs +++ b/crates/ksp-job-backfill-lib/tests/public_api.rs @@ -1,7 +1,7 @@ // file: crates/ksp-job-backfill-lib/tests/public_api.rs -// version: 4 +// version: 6 -//! Public API canaries through bounded Backfill execution and caller-owned checkpoints. +//! Public API canaries through the concrete Backfill cancellation and latest-value runtime tranche. #[test] fn pre_005_request_scope_and_discovery_contracts_are_available_from_crate_root() { @@ -106,3 +106,20 @@ fn pre_008_checkpoint_and_bounded_execution_contracts_are_available_from_crate_r assert_eq!(ksp_job_backfill_lib::ERROR_CODE_BACKFILL_EXECUTION_INVALID, ksp_core_lib::ErrorCode::new("job_backfill", "execution_invalid")); return; } + +#[test] +fn pre_009_concrete_runtime_snapshot_and_control_contracts_are_available_from_crate_root() { + fn assert_source() + where + T: ksp_job_api::JobSnapshotSource, + { + return; + } + assert_source::(); + let _runtime_new = ksp_job_backfill_lib::BackfillJobRuntime::new; + let _handle: std::option::Option = std::option::Option::None; + let _phase = ksp_job_backfill_lib::BackfillJobPhase::Discovering; + assert_eq!(ksp_job_backfill_lib::BACKFILL_JOB_KIND_CODE, "solana.raw_transaction.backfill"); + assert_eq!(ksp_job_backfill_lib::ERROR_CODE_BACKFILL_RUNTIME_INVALID, ksp_core_lib::ErrorCode::new("job_backfill", "runtime_invalid")); + return; +} diff --git a/crates/ksp-job-backfill-lib/tests/release_completeness.rs b/crates/ksp-job-backfill-lib/tests/release_completeness.rs index 8828cd7..ee46540 100644 --- a/crates/ksp-job-backfill-lib/tests/release_completeness.rs +++ b/crates/ksp-job-backfill-lib/tests/release_completeness.rs @@ -1,10 +1,10 @@ // file: crates/ksp-job-backfill-lib/tests/release_completeness.rs -// version: 4 +// version: 7 -//! Completeness canaries through the `pre.008` bounded execution/checkpoint tranche. +//! Completeness canaries through the `pre.009` concrete cancellation and latest-value runtime tranche. #[test] -fn pre_008_production_module_inventory_is_exact() -> std::io::Result<()> { +fn pre_009_production_module_inventory_is_exact() -> std::io::Result<()> { let source_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"); let entries = match std::fs::read_dir(source_root) { std::result::Result::Ok(value) => value, @@ -34,13 +34,24 @@ fn pre_008_production_module_inventory_is_exact() -> std::io::Result<()> { names.sort_unstable(); assert_eq!( names, - std::vec!["checkpoint.rs", "constants.rs", "conversion.rs", "discovery.rs", "error.rs", "execution.rs", "lib.rs", "persistence.rs", "request.rs"] + std::vec![ + "checkpoint.rs", + "constants.rs", + "conversion.rs", + "discovery.rs", + "error.rs", + "execution.rs", + "lib.rs", + "persistence.rs", + "request.rs", + "runtime.rs", + ] ); return std::result::Result::Ok(()); } #[test] -fn pre_008_surface_adds_bounded_execution_and_checkpoint_without_pre_009_runtime() { +fn pre_009_surface_closes_concrete_cancellation_and_latest_value_runtime() { let root = include_str!("../src/lib.rs"); for required in [ "BackfillCandidate", @@ -65,11 +76,18 @@ fn pre_008_surface_adds_bounded_execution_and_checkpoint_without_pre_009_runtime "execute_backfill_discovery", "ERROR_CODE_BACKFILL_CHECKPOINT_INVALID", "ERROR_CODE_BACKFILL_EXECUTION_INVALID", + "BackfillJobHandle", + "BackfillJobPhase", + "BackfillJobRuntime", + "BackfillJobSnapshot", + "BackfillSnapshotSource", + "BACKFILL_JOB_KIND_CODE", + "ERROR_CODE_BACKFILL_RUNTIME_INVALID", ] { - assert!(root.contains(required), "required pre.008 public contract missing: {required}"); + assert!(root.contains(required), "required pre.009 public contract missing: {required}"); } - for forbidden in ["BackfillJobHandle", "BackfillSnapshot", "JobSnapshotSource", "tokio::", "FuturesUnordered"] { - assert!(!root.contains(forbidden), "pre.009 or runtime implementation detail leaked into public root: {forbidden}"); + for forbidden in ["tokio::", "FuturesUnordered", "watch::Receiver", "watch::Sender"] { + assert!(!root.contains(forbidden), "runtime implementation detail leaked into public root: {forbidden}"); } assert!(!root.contains("pub mod ")); return; diff --git a/crates/ksp-job-backfill-lib/unit_tests/discovery.rs b/crates/ksp-job-backfill-lib/unit_tests/discovery.rs index c6c57b7..358e12a 100644 --- a/crates/ksp-job-backfill-lib/unit_tests/discovery.rs +++ b/crates/ksp-job-backfill-lib/unit_tests/discovery.rs @@ -1,5 +1,5 @@ // file: crates/ksp-job-backfill-lib/unit_tests/discovery.rs -// version: 2 +// version: 5 #[derive(Clone, Debug, Eq, PartialEq)] struct PageCall { @@ -15,6 +15,34 @@ struct FakeSource { calls: std::sync::Mutex>, } +struct PendingSource { + calls: std::sync::atomic::AtomicUsize, +} + +impl PendingSource { + fn new() -> Self { + return Self { calls: std::sync::atomic::AtomicUsize::new(0) }; + } + + fn calls(&self) -> usize { + return self.calls.load(std::sync::atomic::Ordering::Acquire); + } +} + +impl super::SignaturePageSource for PendingSource { + fn fetch_signature_page<'a>( + &'a self, + _role: &'a ksp_onchain_transport_lib::HttpRoleName, + _address: &'a ksp_core_lib::Pubkey, + _config: ksp_onchain_transport_lib::SolanaSignaturesForAddressConfig, + ) -> super::SignaturePageFuture<'a> { + self.calls.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + return std::boxed::Box::pin(async { + return std::future::pending::>>().await; + }); + } +} + impl FakeSource { fn new(pages: std::vec::Vec>) -> Self { return Self { pages: std::sync::Mutex::new(pages.into()), calls: std::sync::Mutex::new(std::vec::Vec::new()) }; @@ -130,7 +158,7 @@ async fn pre_005_latest_paginates_newest_first_and_deduplicates_pages_stably() { std::vec![page_entry('6', 60), page_entry('5', 50), page_entry('5', 50)], std::vec![page_entry('4', 40), page_entry('3', 30)], ]); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; assert!(result.is_ok()); let discovery = match result { std::result::Result::Ok(value) => value, @@ -164,7 +192,7 @@ async fn pre_005_before_uses_exclusive_anchor_then_advances_rpc_cursor() { std::option::Option::None => return, }; let source = FakeSource::new(std::vec![std::vec![page_entry('6', 60), page_entry('5', 50)], std::vec![page_entry('4', 40)]]); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; let discovery = match result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, @@ -194,7 +222,7 @@ async fn pre_005_after_keeps_only_nearest_newer_window_and_preserves_rpc_order() std::vec![page_entry('7', 70), page_entry('6', 60), page_entry('5', 50)], std::vec![page_entry('4', 40), page_entry('3', 30)], ]); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; let discovery = match result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, @@ -223,7 +251,7 @@ async fn pre_005_after_page_bound_is_partial_and_does_not_claim_anchor_completio std::option::Option::None => return, }; let source = FakeSource::new(std::vec![std::vec![page_entry('7', 70), page_entry('6', 60)], std::vec![page_entry('5', 50), page_entry('4', 40)],]); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; let discovery = match result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, @@ -243,7 +271,7 @@ async fn pre_005_latest_page_bound_is_partial_when_full_pages_leave_more_history std::option::Option::None => return, }; let source = FakeSource::new(std::vec![std::vec![page_entry('7', 70), page_entry('6', 60)]]); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; let discovery = match result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, @@ -293,7 +321,7 @@ async fn pre_005_explicit_scope_never_calls_transport_and_preserves_network_scop std::result::Result::Err(_) => return, }; let source = FakeSource::new(std::vec::Vec::new()); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; let discovery = match result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, @@ -327,7 +355,7 @@ async fn pre_008_before_resume_uses_checkpoint_cursor_instead_of_original_anchor std::result::Result::Err(_) => return, }; let source = FakeSource::new(std::vec![std::vec![page_entry('4', 40)]]); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; assert!(result.is_ok()); let calls = source.calls(); assert_eq!(calls.len(), 1); @@ -348,10 +376,36 @@ async fn pre_008_latest_resume_restarts_from_current_latest_without_rpc_cursor() std::result::Result::Err(_) => return, }; let source = FakeSource::new(std::vec![std::vec![page_entry('7', 70)]]); - let result = super::discover_with_source(&source, &request).await; + let result = super::discover_with_source(&source, &request, std::option::Option::None).await; assert!(result.is_ok()); let calls = source.calls(); assert_eq!(calls.len(), 1); assert_eq!(calls[0].before, std::option::Option::None); return; } + +#[tokio::test] +async fn pre_009_discovery_rpc_wait_is_cancelled_cooperatively() { + let scope = crate::BackfillScope::latest_address(ksp_core_lib::Pubkey::new_from_array([9_u8; 32])); + let request = match request(scope, 2, 2, 5) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let source = PendingSource::new(); + let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false); + let control = crate::BackfillRuntimeControl::new(cancel_sender); + let cancellation = crate::BackfillCancellationSignal::new(control.token(), cancel_receiver); + let discovery = super::discover_with_source(&source, &request, std::option::Option::Some(&cancellation)); + let cancel = async { + tokio::task::yield_now().await; + assert!(control.request_cancellation()); + }; + let (result, ()) = tokio::join!(discovery, cancel); + let error = match result { + std::result::Result::Ok(_) => return, + std::result::Result::Err(error) => error, + }; + assert_eq!(error.code(), crate::ERROR_CODE_BACKFILL_CANCELLED); + assert_eq!(source.calls(), 1); + return; +} diff --git a/crates/ksp-job-backfill-lib/unit_tests/execution.rs b/crates/ksp-job-backfill-lib/unit_tests/execution.rs index 4315a2f..fc0dbb8 100644 --- a/crates/ksp-job-backfill-lib/unit_tests/execution.rs +++ b/crates/ksp-job-backfill-lib/unit_tests/execution.rs @@ -1,5 +1,5 @@ // file: crates/ksp-job-backfill-lib/unit_tests/execution.rs -// version: 1 +// version: 2 #[derive(Clone, Copy)] enum FakeDisposition { @@ -164,7 +164,7 @@ async fn pre_008_execution_is_bounded_and_reconciles_out_of_order_durable_comple FakePlan { pending_polls: 2, disposition: FakeDisposition::Durable }, FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, ]); - let result = super::execute_with_processor(&processor, &request, &discovery).await; + let result = super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await; let batch = match result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, @@ -196,7 +196,7 @@ async fn pre_008_missing_is_non_fatal_but_blocks_frontier_while_later_candidates FakePlan { pending_polls: 0, disposition: FakeDisposition::Missing }, FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, ]); - let batch = match super::execute_with_processor(&processor, &request, &discovery).await { + let batch = match super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, }; @@ -227,7 +227,7 @@ async fn pre_008_conflict_stops_new_admissions_and_drains_already_in_flight_work FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, ]); - let batch = match super::execute_with_processor(&processor, &request, &discovery).await { + let batch = match super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, }; @@ -263,7 +263,7 @@ async fn pre_008_explicit_resume_skips_only_the_checkpointed_contiguous_prefix() FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, ]); - let batch = match super::execute_with_processor(&processor, &request, &discovery).await { + let batch = match super::execute_with_processor(&processor, &request, &discovery, std::option::Option::None, std::option::Option::None).await { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, }; @@ -274,3 +274,47 @@ async fn pre_008_explicit_resume_skips_only_the_checkpointed_contiguous_prefix() assert_eq!(batch.failure_code(), std::option::Option::None); return; } + +#[tokio::test] +async fn pre_009_cancellation_stops_admission_and_drains_already_admitted_candidate_work() { + let request = match explicit_request(&['1', '2', '3', '4'], 2) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let discovery = match discovery(&request, &['1', '2', '3', '4']) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let processor = FakeProcessor::new(std::vec![ + FakePlan { pending_polls: 12, disposition: FakeDisposition::Durable }, + FakePlan { pending_polls: 12, disposition: FakeDisposition::Durable }, + FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, + FakePlan { pending_polls: 0, disposition: FakeDisposition::Durable }, + ]); + let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false); + let control = crate::runtime::BackfillRuntimeControl::new(cancel_sender); + let signal = crate::runtime::BackfillCancellationSignal::new(control.token(), cancel_receiver); + let execution = super::execute_with_processor(&processor, &request, &discovery, std::option::Option::Some(&signal), std::option::Option::None); + let cancellation = async { + loop { + if processor.calls() >= 2 { + break; + } + tokio::task::yield_now().await; + } + assert!(control.request_cancellation()); + }; + let (result, ()) = tokio::join!(execution, cancellation); + let batch = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(processor.calls(), 2); + assert_eq!(batch.admitted_count(), 2); + assert_eq!(batch.finished_count(), 2); + assert_eq!(batch.durable_count(), 2); + assert_eq!(batch.cancelled_count(), 0); + assert!(batch.was_cancelled()); + assert_eq!(batch.checkpoint().completed_prefix(), 2); + return; +} diff --git a/crates/ksp-job-backfill-lib/unit_tests/runtime.rs b/crates/ksp-job-backfill-lib/unit_tests/runtime.rs new file mode 100644 index 0000000..c9170d4 --- /dev/null +++ b/crates/ksp-job-backfill-lib/unit_tests/runtime.rs @@ -0,0 +1,158 @@ +// file: crates/ksp-job-backfill-lib/unit_tests/runtime.rs +// version: 2 + +use ksp_job_api::JobSnapshotSource; // rust-rules: trait-import + +fn request() -> std::option::Option { + let signature = match crate::BackfillSignature::new("1".repeat(crate::MIN_BACKFILL_SIGNATURE_TEXT_BYTES)) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + let scope = match crate::BackfillScope::explicit_signatures(std::vec![signature]) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + let job_id = match ksp_job_api::JobId::new("backfill:runtime-test") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + let network = match ksp_store_lib::RawNetworkId::new("devnet") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + return crate::BackfillRequest::new( + job_id, + network, + ksp_onchain_transport_lib::HttpRoleName::new("history"), + crate::BackfillCommitment::Confirmed, + scope, + 100, + 10, + 1, + 1, + std::option::Option::None, + ) + .ok(); +} + +#[tokio::test] +async fn pre_009_latest_value_source_coalesces_progress_for_slow_independent_listeners() { + let request = match request() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let runtime = match crate::BackfillJobRuntime::new(request) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let handle = runtime.handle(); + let listener_a = handle.snapshots(); + let listener_b = handle.snapshots(); + let initial = listener_a.current(); + assert_eq!(initial.sequence().value(), 0); + assert_eq!(initial.state(), ksp_job_api::JobState::Created); + let first = runtime.publisher.publish_running(crate::BackfillJobPhase::Discovering); + assert!(first.is_ok()); + let second = runtime.publisher.publish_running(crate::BackfillJobPhase::Executing); + assert!(second.is_ok()); + let coalesced = listener_a.wait_for_change(initial.sequence()).await; + assert_eq!(coalesced.sequence().value(), 2); + assert_eq!(coalesced.snapshot().phase(), crate::BackfillJobPhase::Executing); + let listener_b_current = listener_b.current(); + assert_eq!(listener_b_current.sequence().value(), 2); + assert_eq!(listener_b_current.snapshot().phase(), crate::BackfillJobPhase::Executing); + return; +} + +#[tokio::test] +async fn pre_009_terminal_snapshot_is_retained_and_late_cancellation_is_rejected() { + let request = match request() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let runtime = match crate::BackfillJobRuntime::new(request) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let handle = runtime.handle(); + let listener = handle.snapshots(); + let started = runtime.publisher.publish_running(crate::BackfillJobPhase::Discovering); + assert!(started.is_ok()); + assert_eq!(runtime.control.claim_normal_terminal(), crate::TerminalClaim::Completed); + let terminal = runtime.publisher.publish_terminal(ksp_job_api::JobState::Completed(ksp_job_api::JobCompletion::Complete), std::option::Option::None); + assert!(terminal.is_ok()); + assert!(!handle.cancel()); + let current = listener.current(); + assert!(current.state().is_terminal()); + assert_eq!(current.snapshot().phase(), crate::BackfillJobPhase::Finished); + let retained = listener.wait_for_change(current.sequence()).await; + assert_eq!(retained.sequence(), current.sequence()); + assert_eq!(retained.state(), current.state()); + return; +} + +#[test] +fn pre_009_terminal_race_is_first_decision_wins_for_cancellation_vs_completion() { + let (cancel_sender, _) = tokio::sync::watch::channel(false); + let cancellation_first = crate::BackfillRuntimeControl::new(cancel_sender); + assert!(cancellation_first.request_cancellation()); + assert_eq!(cancellation_first.claim_normal_terminal(), crate::TerminalClaim::Cancelled); + assert!(!cancellation_first.request_cancellation()); + let (cancel_sender, _) = tokio::sync::watch::channel(false); + let completion_first = crate::BackfillRuntimeControl::new(cancel_sender); + assert_eq!(completion_first.claim_normal_terminal(), crate::TerminalClaim::Completed); + assert!(!completion_first.request_cancellation()); + return; +} + +#[test] +fn pre_009_fatal_failure_overrides_pending_cancellation_before_terminal_publication() { + let (cancel_sender, _) = tokio::sync::watch::channel(false); + let control = crate::BackfillRuntimeControl::new(cancel_sender); + assert!(control.request_cancellation()); + control.claim_failed(); + assert_eq!(control.claim_normal_terminal(), crate::TerminalClaim::Failed); + assert!(!control.request_cancellation()); + return; +} + +#[tokio::test] +async fn pre_009_long_running_pre_store_future_is_cancelled_cooperatively() { + let (cancel_sender, cancel_receiver) = tokio::sync::watch::channel(false); + let control = crate::BackfillRuntimeControl::new(cancel_sender); + let signal = crate::BackfillCancellationSignal::new(control.token(), cancel_receiver); + let operation = std::future::pending::>(); + let wait = signal.run_cancellable(operation); + let cancel = async { + tokio::task::yield_now().await; + assert!(control.request_cancellation()); + }; + let (result, ()) = tokio::join!(wait, cancel); + let error = match result { + std::result::Result::Ok(_) => return, + std::result::Result::Err(error) => error, + }; + assert_eq!(error.code(), crate::ERROR_CODE_BACKFILL_CANCELLED); + return; +} + +#[test] +fn pre_009_snapshot_debug_and_public_shape_do_not_include_transport_or_raw_payloads() { + let request = match request() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let runtime = match crate::BackfillJobRuntime::new(request) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let snapshot = runtime.handle().snapshots().current(); + let debug = format!("{:?}", snapshot.snapshot()); + for forbidden in ["http://", "https://", "endpoint", "provider", "raw_payload", "transaction_data"] { + assert!(!debug.contains(forbidden), "unsafe snapshot diagnostic leaked: {forbidden}"); + } + assert_eq!(snapshot.snapshot().scope_kind(), crate::BackfillScopeKind::ExplicitSignatures); + assert_eq!(snapshot.snapshot().candidates_selected(), 0); + assert!(snapshot.snapshot().failure_code().is_none()); + return; +} diff --git a/deltas/0.3.6/pre.009.md b/deltas/0.3.6/pre.009.md new file mode 100644 index 0000000..2056364 --- /dev/null +++ b/deltas/0.3.6/pre.009.md @@ -0,0 +1,272 @@ + + + +# Delta `0.3.6-pre.009` — runtime Backfill, annulation coopérative et snapshots latest-value + +## Base requise + +```text +0.3.6-pre.008-fix.001 appliquée +workspace.package.version = 0.3.6-pre.8.fix.1 +``` + +Le gate opérateur fourni pour cette base confirme : + +```text +cargo fmt --all PASS +python3 scripts/audit_rust_workspace_rules.py PASS / clean +python3 scripts/audit_markdown_tables.py ... PASS / clean (264 tables / 154 fichiers) +cargo check --workspace PASS +cargo clippy --workspace --all-targets PASS / aucun warning +cargo test -p ksp-job-backfill-lib PASS + unitaires 39 PASS + dependency_boundary 2 PASS + public_api 5 PASS + release_completeness 2 PASS +``` + +Les arbres Cargo ne sont pas rejoués pour le fix `pre.008-fix.001`, qui ne modifie ni dépendance ni feature. La présente tranche modifie en revanche le manifeste de `ksp-job-backfill-lib` en promouvant Tokio en dépendance normale privée ; les arbres Cargo redeviennent donc pertinents pour son gate. + +## Objectif + +Matérialiser exclusivement `pre.009` du plan 027 : fournir le runtime concret du premier Backfill RAW avec annulation coopérative, arbitrage terminal déterministe et diffusion latest-value de snapshots sûrs via le contrat runtime-neutral de `ksp-job-api`. + +La tranche ne crée pas de scheduler global, ne persiste pas les checkpoints, ne modifie pas Store/Transport/Job API et ne prépare pas encore le hardening externe final de `pre.010`. + +## Runtime concret + +La crate expose désormais : + +```text +BACKFILL_JOB_KIND_CODE +BackfillJobPhase +BackfillJobSnapshot +BackfillSnapshotSource +BackfillJobHandle +BackfillJobRuntime +ERROR_CODE_BACKFILL_RUNTIME_INVALID +``` + +`BackfillJobRuntime::new` construit un état initial `Created`, un canal latest-value privé et un canal de réveil d'annulation privé. `handle()` est obtenu avant de déplacer le runtime dans `run`, afin que l'hôte conserve une surface cloneable de contrôle et d'observation pendant l'exécution. + +Le kind stable est : + +```text +solana.raw_transaction.backfill +``` + +Aucun type Tokio n'est réexporté à la crate root. + +## Snapshot latest-value + +`BackfillJobSnapshot` expose uniquement des faits sûrs et bornés : + +```text +phase +scope_kind +discovery_boundary +candidates_selected +candidates_admitted +candidates_finished +entities_inserted +entities_existing +entities_purged +missing +conflicts +observations_inserted +observations_existing +cancelled_candidates +holes +maximum_in_flight +contiguous_completed +checkpoint +failure_code +``` + +Il n'expose ni URL, provider, endpoint, payload RAW, transaction encodée, credential ou message humain libre. + +`BackfillSnapshotSource` implémente le trait `ksp_job_api::JobSnapshotSource` avec un `tokio::sync::watch` entièrement privé. La source est latest-value : un listener lent peut coalescer les transitions intermédiaires et récupère toujours la valeur la plus récente. Les listeners clonés sont indépendants et la valeur terminale reste retenue après fermeture logique du Job. + +Chaque publication incrémente `JobNotificationSequence`. Une source observant déjà la séquence terminale retourne immédiatement le snapshot terminal retenu. + +## Annulation coopérative + +`BackfillJobHandle::cancel` accepte atomiquement la première annulation tant que la décision terminale n'a pas été prise. L'annulation runtime-neutral reste portée par `JobCancellationToken`; un `watch` privé sert uniquement à réveiller immédiatement les futures Tokio en attente. + +Les opérations abandonnables pré-Store utilisent une sélection coopérative : + +```text +pagination getSignaturesForAddress +hydratation getTransaction +future pré-Store encore non soumise +``` + +Une annulation déjà demandée retourne immédiatement le code interne stable : + +```text +job_backfill.cancelled +``` + +Ce code reste crate-internal et sert à distinguer l'annulation d'une erreur fatale avant projection vers `JobState::Cancelled`. + +## Drainage après soumission Store + +La frontière critique reste celle figée par le plan : une opération peut être abandonnée avant soumission durable, mais une persistance Store déjà soumise n'est jamais annulée aveuglément. + +L'exécuteur : + +1. arrête les nouvelles admissions lorsqu'une annulation est observée ; +2. peut annuler les candidats encore dans une phase pré-Store abandonnable ; +3. laisse les persistances déjà soumises produire un résultat connu ; +4. draine toutes les futures déjà admises ; +5. réconcilie la frontier et le checkpoint avant publication terminale. + +`BackfillExecutionBatch` expose maintenant les compteurs détaillés nécessaires au snapshot concret : inserted, already-present, purged, missing, conflicts, observations, cancelled, holes et état `was_cancelled`. + +## Arbitrage terminal + +Un état atomique privé sérialise la course entre complétion normale, annulation et échec fatal. + +Règles : + +```text +annulation acceptée avant claim terminal normal -> Cancelled +claim Completed avant annulation -> annulation tardive rejetée +échec fatal avant publication terminale -> Failed, même si annulation était pending +``` + +La publication latest-value suit uniquement des transitions autorisées entre `Created`, `Running`, `Cancelling`, `Completed`, `Cancelled` et `Failed`. Une publication après état terminal est rejetée par `ERROR_CODE_BACKFILL_RUNTIME_INVALID`. + +## Projection des résultats + +En absence d'échec fatal : + +- batch complet sans trou -> `Completed(Complete)` ; +- batch partiel, notamment `Missing` -> `Completed(Partial)` ; +- annulation gagnante après drainage -> `Cancelled` ; +- conflit ou erreur Transport/conversion/Store terminale -> `Failed` avec code stable conservé dans le snapshot. + +Le checkpoint terminal est celui du dernier préfixe contigu durable produit par `pre.008`; l'annulation ne peut pas faire avancer la frontier au-delà d'un trou ou d'un candidat annulé. + +## Dépendances + +La tranche promeut Tokio de dev-only à dépendance normale privée dans `ksp-job-backfill-lib` : + +```text +tokio = { workspace = true, features = ["macros", "sync"] } +``` + +L'entrée versionnée existe déjà dans `[workspace.dependencies]`; aucune nouvelle version externe n'est introduite. Les tests conservent `rt-multi-thread` uniquement dans `[dev-dependencies]`. + +`futures-util` reste la dépendance normale privée introduite par `pre.008`. Aucun type Tokio/Futures n'entre dans l'API publique. + +## Tests matérialisés + +La tranche ajoute **8 tests unitaires**, portant le total à **47 unitaires**. Ils couvrent notamment : + +- latest-value coalescé pour listener lent et indépendance des listeners clonés ; +- rétention du snapshot terminal et rejet d'une annulation tardive ; +- arbitrage déterministe annulation contre complétion ; +- échec fatal gagnant sur annulation pending avant publication terminale ; +- future pré-Store longue abandonnée coopérativement ; +- annulation pendant une page de découverte RPC pendante ; +- arrêt des admissions et drainage du travail déjà admis ; +- forme sûre du snapshot sans provenance Transport ni payload RAW. + +Une canarie publique supplémentaire porte les canaries d'intégration à **10** : + +```text +dependency_boundary 2 +public_api 6 +release_completeness 2 +``` + +Total matérialisé attendu au gate : + +```text +47 unitaires +10 canaries d'intégration +``` + +## Fichiers ajoutés + +```text +crates/ksp-job-backfill-lib/src/runtime.rs +crates/ksp-job-backfill-lib/unit_tests/runtime.rs +deltas/0.3.6/pre.009.md +``` + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-job-backfill-lib/Cargo.toml +crates/ksp-job-backfill-lib/src/discovery.rs +crates/ksp-job-backfill-lib/src/error.rs +crates/ksp-job-backfill-lib/src/execution.rs +crates/ksp-job-backfill-lib/src/lib.rs +crates/ksp-job-backfill-lib/tests/dependency_boundary.rs +crates/ksp-job-backfill-lib/tests/public_api.rs +crates/ksp-job-backfill-lib/tests/release_completeness.rs +crates/ksp-job-backfill-lib/unit_tests/discovery.rs +crates/ksp-job-backfill-lib/unit_tests/execution.rs +docs/plans/027-V0_3_6_JOB_API_BACKFILL_PLAN.md +docs/validation/023-V0_3_6_JOB_API_BACKFILL.md +``` + +Aucune suppression. + +## Versions d'en-tête + +Les versions finales reflètent chaque modification réellement enregistrée pendant la matérialisation. Certains fichiers de tests ont donc avancé de plusieurs unités plutôt que d'être artificiellement ramenés à `N + 1` : + +```text +Cargo.toml 407 -> 408 +crates/ksp-job-backfill-lib/Cargo.toml 3 -> 4 +crates/ksp-job-backfill-lib/src/discovery.rs 3 -> 4 +crates/ksp-job-backfill-lib/src/error.rs 4 -> 5 +crates/ksp-job-backfill-lib/src/execution.rs 1 -> 2 +crates/ksp-job-backfill-lib/src/lib.rs 4 -> 5 +crates/ksp-job-backfill-lib/tests/dependency_boundary.rs 4 -> 6 +crates/ksp-job-backfill-lib/tests/public_api.rs 4 -> 6 +crates/ksp-job-backfill-lib/tests/release_completeness.rs 4 -> 7 +crates/ksp-job-backfill-lib/unit_tests/discovery.rs 2 -> 5 +crates/ksp-job-backfill-lib/unit_tests/execution.rs 1 -> 2 +docs/plans/027-V0_3_6_JOB_API_BACKFILL_PLAN.md 15 -> 16 +docs/validation/023-V0_3_6_JOB_API_BACKFILL.md 15 -> 16 +``` + +Les deux nouveaux fichiers Rust ont été créés puis révisés pendant la tranche et sont livrés à `version: 2`. Le présent delta commence à `version: 1`. + +## Version workspace + +```text +workspace.package.version = 0.3.6-pre.9 +delivery = 0.3.6-pre.009 +``` + +## Validations exécutées dans l'environnement d'assemblage + +```text +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas/0.3.6 +``` + +Les audits statiques sont exécutés avant et après création du présent delta. L'environnement d'assemblage ne fournit ni Cargo, ni Rustc, ni Rustfmt ; la compilation, Clippy et les tests Rust `pre.009` restent donc à exécuter par l'opérateur. + +## Gate opérateur demandé + +Le manifeste normal de `ksp-job-backfill-lib` change dans cette tranche, donc les arbres Cargo sont à rejouer : + +```bash +cargo fmt --all +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas/0.3.6 +cargo check --workspace +cargo clippy --workspace --all-targets +cargo test -p ksp-job-backfill-lib +cargo tree -p ksp-job-backfill-lib --edges normal +cargo tree -p ksp-job-backfill-lib -e features +``` + +Attendu : 47 unitaires et 10 canaries d'intégration, sans warning Clippy. Un `cargo clean` n'est pas requis par `cargo tree`; il constitue séparément une raison valable de rejouer les arbres lorsqu'une reconstruction propre est volontairement effectuée. diff --git a/docs/plans/027-V0_3_6_JOB_API_BACKFILL_PLAN.md b/docs/plans/027-V0_3_6_JOB_API_BACKFILL_PLAN.md index ecf54db..befd109 100644 --- a/docs/plans/027-V0_3_6_JOB_API_BACKFILL_PLAN.md +++ b/docs/plans/027-V0_3_6_JOB_API_BACKFILL_PLAN.md @@ -1,5 +1,5 @@ - + # Plan v0.3.6 — Job API et premier backfill RAW @@ -482,7 +482,7 @@ La tranche matérialise 29 tests unitaires et 8 canaries d'intégration au total ### `pre.008` — Concurrence, frontier et checkpoint -**Statut : réalisé ; corrigé par `pre.008-fix.001`, gate opérateur du fix à rejouer.** +**Statut : clôturé ; corrigé par `pre.008-fix.001`, gate opérateur du fix vert.** Budget cible : **15-20 min**. Entrée : résultats Store typés et gate `pre.007` vert. La tranche ajoute `BackfillCheckpoint`, `BackfillExecutionBatch` et `execute_backfill_discovery`. Le checkpoint est opaque, caller-owned et validé contre `JobId` + fingerprint de scope ; aucune table ou persistance durable de Job n'est créée. `BackfillRequest::with_checkpoint` reste additif afin de préserver les constructeurs existants. @@ -494,13 +494,17 @@ La reprise suit les quatre sémantiques figées : Latest repart de la vue couran **Statut : matérialisé ; gate opérateur à rejouer.** -Le gate opérateur de `pre.008` confirme `cargo fmt`, les audits Rust/Markdown et `cargo check --workspace`, puis la compilation des targets de test échoue en `E0618`/`E0282` : dans `pre_008_checkpoint_is_bound_to_job_and_scope_fingerprint`, la variable locale `request` masque le helper homonyme avant la construction du second scope. Le fix renomme uniquement cette liaison locale en `primary_request` et synchronise ses usages ; aucun contrat, test, comportement runtime, checkpoint, dépendance ou feature ne change. Comme un fichier Rust est corrigé, la version workspace devient `0.3.6-pre.8.fix.1`. Les arbres Cargo de `pre.008` ont déjà été inspectés puisque cette tranche ajoutait `futures-util`; ils n'ont pas à être rejoués pour ce fix sans changement de graphe. +Le gate opérateur de `pre.008` confirme `cargo fmt`, les audits Rust/Markdown et `cargo check --workspace`, puis la compilation des targets de test échoue en `E0618`/`E0282` : dans `pre_008_checkpoint_is_bound_to_job_and_scope_fingerprint`, la variable locale `request` masque le helper homonyme avant la construction du second scope. Le fix renomme uniquement cette liaison locale en `primary_request` et synchronise ses usages ; aucun contrat, test, comportement runtime, checkpoint, dépendance ou feature ne change. Comme un fichier Rust est corrigé, la version workspace devient `0.3.6-pre.8.fix.1`. Le gate réduit du fix est ensuite intégralement vert : audits, `cargo check`, Clippy sans warning, 39 unitaires et 9 canaries d'intégration passent. Les arbres Cargo de `pre.008` avaient déjà été inspectés puisque cette tranche ajoutait `futures-util`; ils n'avaient pas à être rejoués pour ce fix sans changement de graphe. ### `pre.009` — Annulation et snapshots concrets -**Statut : planifié.** +**Statut : matérialisé ; gate opérateur à rejouer.** -Budget cible : **15-20 min**. Entrée : frontière prouvée. Intégrer annulation sur les phases critiques et progression Backfill latest-value. Sortie : courses terminales, drainage Store et listener lent couverts. +Budget cible : **15-20 min**. Entrée : frontière prouvée et gate `pre.008-fix.001` vert. La tranche matérialise `BackfillJobRuntime`, `BackfillJobHandle`, `BackfillJobSnapshot` et `BackfillSnapshotSource` comme runtime concret du vertical RAW transaction. `ksp-job-api` reste inchangée, passive et runtime-neutral ; Tokio devient uniquement un détail d'implémentation normal de `ksp-job-backfill-lib` (`macros` + `sync`) et aucun type Tokio n'est exposé dans la racine publique. Le canal concret est un `watch` latest-value O(1) : les progressions intermédiaires peuvent être coalescées, chaque listener possède son receiver cloné, et la dernière valeur terminale reste lisible tant que le handle/source existe. + +L'annulation réutilise `JobCancellationToken` et ajoute un signal réveillable privé. Les attentes de découverte et d'hydratation pré-Store sont abandonnables ; une vérification supplémentaire intervient après hydratation et avant soumission Store. Dès que `persist_backfill_hydration` est appelée, l'écriture atomique n'est plus sélectionnée contre l'annulation : toute opération Store déjà soumise est drainée jusqu'à un résultat connu. L'exécuteur arrête ensuite les nouvelles admissions, draine l'in-flight et publie frontier/checkpoint sûrs. Une arbitration atomique sépare `ACTIVE`, demande d'annulation et états terminaux : une annulation acceptée avant la décision terminale gagne sur une complétion normale, une complétion déjà réclamée refuse l'annulation tardive, et une erreur fatale peut remplacer une annulation pendante afin de ne pas masquer un échec Transport/Store. + +Le snapshot complet reste sûr : phase, catégorie de scope, borne de découverte, compteurs d'admission/fin, dispositions entité/observation, missing/conflits/annulations/trous, maximum in-flight, préfixe contigu, checkpoint et éventuel code d'erreur stable ; aucune URL, endpoint/provider, payload RAW ou message fournisseur n'y entre. La tranche matérialise 47 tests unitaires et 10 canaries d'intégration au total, dont listener lent/multi-listeners, rétention terminale, courses terminales, attente pré-Store longue, attente de page RPC pendante et drainage après annulation. Leur exécution Rust reste à confirmer par le gate opérateur. `pre.010` reste fermé au hardening et aux canaries externes. ### `pre.010` — Hardening et canaries externes diff --git a/docs/validation/023-V0_3_6_JOB_API_BACKFILL.md b/docs/validation/023-V0_3_6_JOB_API_BACKFILL.md index d8e8180..c25553a 100644 --- a/docs/validation/023-V0_3_6_JOB_API_BACKFILL.md +++ b/docs/validation/023-V0_3_6_JOB_API_BACKFILL.md @@ -1,5 +1,5 @@ - + # Validation v0.3.6 — Job API et premier backfill RAW @@ -90,7 +90,7 @@ Aucune entrée absolue, traversée, avec séparateur inversé ou lien symbolique ## 6. Notifications multi-listeners -`pre.003` fige le contrat public et ses canaries de resynchronisation. Les cases portant sur le stockage/réveil concret restent volontairement ouvertes jusqu'au producer `ksp-job-backfill-lib` ; un fake de test n'est pas traité comme preuve runtime de production. +`pre.003` fige le contrat public et ses canaries de resynchronisation. `pre.009` matérialise maintenant le producer concret `ksp-job-backfill-lib` avec un `tokio::sync::watch` privé, un handle clonable et des canaries latest-value ; leur gate Rust opérateur reste à rejouer, donc les critères comportementaux ci-dessous restent ouverts jusqu'à exécution. - [ ] Stockage O(1) de la dernière valeur, sans historique non borné. - [ ] Un listener lent ne bloque jamais le producteur. @@ -106,14 +106,14 @@ Aucune entrée absolue, traversée, avec séparateur inversé ou lien symbolique - [X] `LatestAddress`, `BeforeAddress`, `AfterAddress` et `ExplicitSignatures` couverts. - [X] Adresse validée par le type Core existant. -- [ ] Signature Base58 décodée vers exactement 64 octets. +- [X] Signature Base58 décodée vers exactement 64 octets, prouvée par le gate `pre.006-fix.002`. - [X] Engagement limité à `Confirmed` ou `Finalized`. - [X] Taille de page entre 1 et 1 000. - [X] Nombre de pages entre 1 et 10 000. - [X] Nombre de candidats entre 1 et 10 000. - [X] Concurrence d'hydratation entre 1 et 64. - [X] `min_context_slot` transmis sans en changer le sens. -- [ ] Checkpoint rejeté si son JobId ou son fingerprint de scope diffère ; canarie corrigée par `pre.008-fix.001`, gate opérateur du fix à rejouer. +- [X] Checkpoint rejeté si son JobId ou son fingerprint de scope diffère ; canarie `pre.008-fix.001` exécutée avec succès. - [X] Aucun défaut d'application silencieux dans la bibliothèque. ## 8. Pagination, ordre et déduplication @@ -175,20 +175,22 @@ Aucune entrée absolue, traversée, avec séparateur inversé ou lien symbolique ## 12. Concurrence, frontière et checkpoint -`pre.008` matérialise un checkpoint opaque caller-owned, sans table Store ni promesse de reprise crash-safe. `BackfillRequest::with_checkpoint` valide `JobId` et fingerprint avant usage. L'exécuteur utilise une collection `FuturesUnordered` strictement bornée par `hydration_concurrency`, réconcilie les fins hors ordre par index stable et draine les futures déjà admises après la première erreur fatale. Le gate opérateur confirme format, audits et `cargo check`, mais la compilation des tests échoue en `E0618`/`E0282` parce qu'une liaison locale `request` masque le helper homonyme dans la canarie checkpoint. `pre.008-fix.001` renomme uniquement cette liaison en `primary_request`; le gate du fix reste à rejouer. +`pre.008` matérialise un checkpoint opaque caller-owned, sans table Store ni promesse de reprise crash-safe. `BackfillRequest::with_checkpoint` valide `JobId` et fingerprint avant usage. L'exécuteur utilise une collection `FuturesUnordered` strictement bornée par `hydration_concurrency`, réconcilie les fins hors ordre par index stable et draine les futures déjà admises après la première erreur fatale. Le gate initial révèle uniquement un shadowing de helper dans une canarie ; `pre.008-fix.001` le corrige sans changer le contrat. Le gate réduit du fix est intégralement vert : audits, `cargo check`, Clippy sans warning, 39 unitaires et 9 canaries d'intégration passent sur `0.3.6-pre.8.fix.1`. -- [ ] Nombre d'hydratations en vol borné par la requête. -- [ ] Aucune file de candidats ou notifications non bornée. -- [ ] Résultats hors ordre réconciliés par index stable. -- [ ] Frontière avance sur insert, already present et purged skip durables. -- [ ] Missing, conflit, conversion, Transport, Store et annulation créent un trou. -- [ ] Frontière ne saute jamais un trou même si des résultats ultérieurs terminent. -- [ ] Latest rejoue la vue courante de façon idempotente. -- [ ] Before reprend au curseur contigu. -- [ ] After et explicite rejouent le scope borné avec préfixe validé. +- [X] Nombre d'hydratations en vol borné par la requête, prouvé par `pre.008-fix.001`. +- [X] Aucune file de candidats non bornée dans l'exécuteur `pre.008`; les notifications concrètes restent à valider avec `pre.009`. +- [X] Résultats hors ordre réconciliés par index stable. +- [X] Frontière avance sur insert, already present et purged skip durables. +- [X] Missing, conflit, conversion, Transport et Store créent un trou ; la branche annulation est matérialisée dans `pre.009` et reste à exécuter. +- [X] Frontière ne saute jamais un trou même si des résultats ultérieurs terminent. +- [X] Latest rejoue la vue courante de façon idempotente. +- [X] Before reprend au curseur contigu. +- [X] After et explicite rejouent le scope borné avec préfixe validé. ## 13. Annulation et résultats terminaux +`pre.009` matérialise le runtime concret, le canal latest-value et les chemins d'annulation/drainage. Les canaries couvrent l'attente RPC de découverte, une future pré-Store longue, le drainage des candidats déjà admis, les listeners lents/indépendants, la rétention terminale et les arbitrations terminales ; le gate Rust opérateur reste à rejouer avant de cocher ces critères comportementaux. + - [ ] Annulation idempotente et état `Cancelling` observable. - [ ] Admissions arrêtées après observation du token. - [ ] Attente de page et hydratation RPC longues abandonnables. @@ -204,10 +206,10 @@ Aucune entrée absolue, traversée, avec séparateur inversé ou lien symbolique - [X] `ksp-job-api` dépend exactement de Core et n'a aucune feature, dev-dependency ou build-dependency. - [X] `ksp-job-backfill-lib` ne dépend pas de Config, App, Wallet, Program, Interface ou backend Store ; `ksp-store-lib` est consommée avec `default-features = false`. - [ ] Aucun retour de dépendance vers Job depuis les couches basses. -- [X] Versions/features auditées : aucune nouvelle version externe ; `sha2`, `futures-util` et `tokio` réutilisent les entrées workspace existantes. `futures-util` devient un détail normal privé de `pre.008`, Tokio reste dev-only. +- [X] Versions/features auditées : aucune nouvelle version externe ; `sha2`, `futures-util` et `tokio` réutilisent les entrées workspace existantes. `futures-util` reste un détail normal privé depuis `pre.008` et `pre.009` promeut Tokio en dépendance normale privée limitée à `macros` + `sync`; aucun type Tokio n'est exposé. - [X] `pre.005` matérialise 11 tests unitaires et 6 canaries d’intégration ; le gate opérateur les exécute tous avec succès. - [X] `pre.006` matérialise 19 tests unitaires et 7 canaries d’intégration ; le gate opérateur les exécute tous avec succès, avec uniquement le warning Clippy corrigé par `pre.006-fix.001`. -- [X] `cargo clippy --workspace --all-targets` warning-free sur `pre.007` ; `pre.008` atteint la compilation des tests puis échoue sur le shadowing corrigé par `pre.008-fix.001`. +- [X] `cargo clippy --workspace --all-targets` warning-free sur `pre.008-fix.001` avec 39 unitaires et 9 canaries vertes ; le gate `pre.009` reste à rejouer. - [ ] `cargo fmt --all -- --check` vert. - [X] `scripts/audit_rust_workspace_rules.py` vert sur `pre.005` dans l'environnement d'assemblage et dans le gate opérateur. - [X] `scripts/audit_markdown_tables.py` vert sur `pre.005` dans le gate opérateur (264 tables / 147 fichiers, delta inclus). @@ -219,7 +221,7 @@ Aucune entrée absolue, traversée, avec séparateur inversé ou lien symbolique ## 15. Preuves d'intégration et fermeture -- [X] Fake Transport et fake Store déterministes sans backend direct exécutés avec succès au gate `pre.007`; fake processor concurrent `pre.008` matérialisé, gate à rejouer. +- [X] Fake Transport et fake Store déterministes sans backend direct exécutés avec succès au gate `pre.007`; fake processor concurrent `pre.008-fix.001` exécuté avec succès. - [ ] Vertical découverte, hydratation, conversion, persistance et observation couvert. - [ ] Smoke Devnet plus PostgreSQL configuré exécuté si l'environnement explicite est disponible. - [ ] Aucun endpoint payant, credential ou donnée sensible requis par les tests normaux.