0.3.15-pre.010-fix.001
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
// file: crates/ksp-app-raw-transaction-ingest-desk/src/app_state.rs
|
||||
// version: 6
|
||||
// version: 7
|
||||
|
||||
//! Shared backend state owned by the Raw Transaction Ingest Desk Tauri application.
|
||||
|
||||
@@ -137,6 +137,11 @@ impl crate::AppState {
|
||||
return crate::start_route_runtime(std::sync::Arc::clone(&self.route_runtime), prepared).await;
|
||||
}
|
||||
|
||||
/// Returns complete latest-value monitoring projections for active and recent terminal routes in the current same-network runtime session.
|
||||
pub(crate) fn route_monitoring(&self) -> ksp_core_lib::Result<std::vec::Vec<crate::RawIngestRouteMonitoringDto>> {
|
||||
return self.route_runtime.monitoring_statuses();
|
||||
}
|
||||
|
||||
/// Requests cooperative Stop for one exact logical route and waits until its Worker cleanup is complete.
|
||||
pub(crate) async fn stop_route(&self, request: &crate::RawIngestRouteStopRequestDto) -> ksp_core_lib::Result<crate::RawIngestRouteRuntimeDto> {
|
||||
return self.route_runtime.stop_and_wait(request).await;
|
||||
@@ -175,7 +180,7 @@ impl crate::AppState {
|
||||
application_version: env!("CARGO_PKG_VERSION").to_owned(),
|
||||
config_document_count: document_count,
|
||||
fallback_logging_active: runtime.fallback_active,
|
||||
shell_phase: "pre.009-multi-route-mainnet".to_owned(),
|
||||
shell_phase: "pre.010-runtime-monitoring".to_owned(),
|
||||
startup_diagnostic: runtime.startup_diagnostic.clone(),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
// file: crates/ksp-app-raw-transaction-ingest-desk/src/lib.rs
|
||||
// version: 6
|
||||
// version: 7
|
||||
|
||||
//! Tauri desktop application scaffold for composing and supervising KSP live RAW transaction ingest routes.
|
||||
|
||||
@@ -16,6 +16,7 @@ mod errors;
|
||||
mod frontend_logging;
|
||||
mod logging_runtime;
|
||||
mod route_inventory;
|
||||
mod route_monitoring;
|
||||
mod route_runtime;
|
||||
mod route_start;
|
||||
mod splash;
|
||||
@@ -158,6 +159,14 @@ pub(crate) use self::logging_runtime::launch_identity;
|
||||
pub(crate) use self::route_inventory::build_route_inventory;
|
||||
/// Rebuilds the route inventory from one already captured Config environment snapshot.
|
||||
pub(crate) use self::route_inventory::build_route_inventory_with_environment;
|
||||
/// Stable Tauri event carrying one complete latest-value route monitoring projection.
|
||||
pub(crate) use self::route_monitoring::RAW_INGEST_ROUTE_STATUS_EVENT_NAME;
|
||||
/// Frontend-safe bounded projection of one run-local continuity gap.
|
||||
pub(crate) use self::route_monitoring::RawIngestRouteGapDto;
|
||||
/// Frontend-safe latest-value monitoring projection of one independent route Worker.
|
||||
pub(crate) use self::route_monitoring::RawIngestRouteMonitoringDto;
|
||||
/// Projects one complete concrete Worker snapshot to the safe route monitoring contract.
|
||||
pub(crate) use self::route_monitoring::project_route_monitoring;
|
||||
/// Launch ownership for one independent route Worker sharing the application Store when eligible.
|
||||
pub(crate) use self::route_runtime::RouteRuntimeLaunch;
|
||||
/// Shared multi-route runtime state with one independent Worker per route and one same-network Store.
|
||||
|
||||
@@ -0,0 +1,352 @@
|
||||
// file: crates/ksp-app-raw-transaction-ingest-desk/src/route_monitoring.rs
|
||||
// version: 1
|
||||
|
||||
//! Frontend-safe latest-value monitoring projection for Raw Transaction Ingest Desk route Workers.
|
||||
|
||||
use ts_rs::TS; // rust-rules: trait-import
|
||||
|
||||
/// Stable Tauri event carrying one complete latest-value route monitoring projection.
|
||||
pub(crate) const RAW_INGEST_ROUTE_STATUS_EVENT_NAME: &str = "ksp-raw-ingest-route-status";
|
||||
|
||||
/// Frontend-safe bounded projection of one run-local continuity gap.
|
||||
#[derive(Clone, Debug, serde::Serialize, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export, export_to = "../frontend/ts/bindings/ksp_app_raw_transaction_ingest_desk/route_monitoring/RawIngestRouteGapDto.ts")]
|
||||
pub(crate) struct RawIngestRouteGapDto {
|
||||
/// Run-local gap identifier encoded as decimal text to avoid JavaScript integer truncation.
|
||||
pub(crate) gap_id: String,
|
||||
/// Inclusive first slot encoded as decimal text.
|
||||
pub(crate) start_slot: String,
|
||||
/// Inclusive last slot encoded as decimal text.
|
||||
pub(crate) end_slot: String,
|
||||
/// Stable source-neutral gap lifecycle code.
|
||||
pub(crate) state: String,
|
||||
/// Stable source-neutral reason code.
|
||||
pub(crate) reason: String,
|
||||
/// Stable source-neutral latest repair mechanism code, when any.
|
||||
pub(crate) last_method: std::option::Option<String>,
|
||||
}
|
||||
|
||||
/// Frontend-safe latest-value monitoring projection of one independent route Worker.
|
||||
#[derive(Clone, Debug, serde::Serialize, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export, export_to = "../frontend/ts/bindings/ksp_app_raw_transaction_ingest_desk/route_monitoring/RawIngestRouteMonitoringDto.ts")]
|
||||
pub(crate) struct RawIngestRouteMonitoringDto {
|
||||
/// Confirmed/finalized commitment bound to this Worker run.
|
||||
pub(crate) commitment: crate::RawIngestCommitment,
|
||||
/// Inventory generation revalidated before this Worker started.
|
||||
pub(crate) inventory_generation: u32,
|
||||
/// Logical network shared by Worker and Store.
|
||||
pub(crate) network: String,
|
||||
/// Safe logical network-profile identifier.
|
||||
pub(crate) profile_id: String,
|
||||
/// Stable logical route identifier.
|
||||
pub(crate) route_id: crate::RawIngestRouteId,
|
||||
/// Monotone Worker snapshot sequence encoded as decimal text.
|
||||
pub(crate) sequence: String,
|
||||
/// Safe application-owned lifecycle projection.
|
||||
pub(crate) state: crate::RawIngestRouteState,
|
||||
/// Stable generic Worker health code.
|
||||
pub(crate) health: String,
|
||||
/// Stable generic Worker activity code.
|
||||
pub(crate) activity: String,
|
||||
/// Whether this latest Worker snapshot is terminal.
|
||||
pub(crate) terminal: bool,
|
||||
/// Stable terminal fault domain when lifecycle is faulted.
|
||||
pub(crate) fault_domain: std::option::Option<String>,
|
||||
/// Stable terminal fault code when lifecycle is faulted.
|
||||
pub(crate) fault_code: std::option::Option<String>,
|
||||
/// Configured bounded admission queue capacity.
|
||||
pub(crate) admission_queue_capacity: u32,
|
||||
/// Latest admission queue depth.
|
||||
pub(crate) admission_queue_depth: u32,
|
||||
/// Configured Store persistence concurrency.
|
||||
pub(crate) persistence_concurrency: u32,
|
||||
/// Latest number of in-flight Store persistence operations.
|
||||
pub(crate) in_flight_persistence: u32,
|
||||
/// Total admitted ingress entries encoded as decimal text.
|
||||
pub(crate) admitted_total: String,
|
||||
/// Total canonicalized ingress entries encoded as decimal text.
|
||||
pub(crate) canonicalized_total: String,
|
||||
/// Total successful Store persistence outcomes encoded as decimal text.
|
||||
pub(crate) persisted_total: String,
|
||||
/// Total newly inserted canonical RAW entities encoded as decimal text.
|
||||
pub(crate) entity_inserted_total: String,
|
||||
/// Total canonical RAW entities already present encoded as decimal text.
|
||||
pub(crate) entity_already_present_total: String,
|
||||
/// Total purge tombstones respected by persistence encoded as decimal text.
|
||||
pub(crate) entity_skipped_purged_total: String,
|
||||
/// Total newly inserted acquisition observations encoded as decimal text.
|
||||
pub(crate) observation_inserted_total: String,
|
||||
/// Total acquisition observations already present encoded as decimal text.
|
||||
pub(crate) observation_already_present_total: String,
|
||||
/// Total durable content conflicts encoded as decimal text.
|
||||
pub(crate) content_conflict_total: String,
|
||||
/// Total non-conflict Store failures encoded as decimal text.
|
||||
pub(crate) store_failure_total: String,
|
||||
/// Total source-task failures encoded as decimal text.
|
||||
pub(crate) source_failure_total: String,
|
||||
/// Total bounded-admission backpressure waits encoded as decimal text.
|
||||
pub(crate) backpressure_wait_total: String,
|
||||
/// Latest number of source signals awaiting hydration/admission processing.
|
||||
pub(crate) hydration_pending: u32,
|
||||
/// Run-local processing frontier encoded as decimal text when established.
|
||||
pub(crate) processing_frontier_slot: std::option::Option<String>,
|
||||
/// Oldest slot still owning pending source work encoded as decimal text when present.
|
||||
pub(crate) oldest_pending_slot: std::option::Option<String>,
|
||||
/// Whether continuity health policy has emitted a run-local projection.
|
||||
pub(crate) continuity_policy_observed: bool,
|
||||
/// Latest run-local continuity frontier encoded as decimal text when established.
|
||||
pub(crate) continuity_frontier_slot: std::option::Option<String>,
|
||||
/// Whether one or more continuity gaps are currently open.
|
||||
pub(crate) continuity_has_open_gaps: bool,
|
||||
/// Whether all failed-source continuity obligations have been reconciled.
|
||||
pub(crate) failed_source_losses_reconciled: bool,
|
||||
/// Whether current coverage proves the target can continue into future slots.
|
||||
pub(crate) future_target_coverage: bool,
|
||||
/// Stable source-neutral latest productive-source lifecycle code.
|
||||
pub(crate) source_state: std::option::Option<String>,
|
||||
/// Configured logical live-source count.
|
||||
pub(crate) source_total: u32,
|
||||
/// Latest Active source count.
|
||||
pub(crate) source_active: u32,
|
||||
/// Latest Reconnecting source count.
|
||||
pub(crate) source_reconnecting: u32,
|
||||
/// Latest Failed source count.
|
||||
pub(crate) source_failed: u32,
|
||||
/// Successful source reconnect count encoded as decimal text.
|
||||
pub(crate) source_reconnect_total: String,
|
||||
/// Replay-bearing reconnect attempt count encoded as decimal text.
|
||||
pub(crate) source_replay_attempt_total: String,
|
||||
/// Proven replay-retention continuity gap count encoded as decimal text.
|
||||
pub(crate) source_continuity_gap_total: String,
|
||||
/// Number of currently open continuity gaps.
|
||||
pub(crate) open_gap_count: u32,
|
||||
/// Number of continuity gaps currently under repair.
|
||||
pub(crate) repairing_gap_count: u32,
|
||||
/// Cumulative repaired-gap count encoded as decimal text.
|
||||
pub(crate) repaired_gap_total: String,
|
||||
/// Cumulative unresolved-gap count encoded as decimal text.
|
||||
pub(crate) unresolved_gap_total: String,
|
||||
/// Cumulative replay repair count encoded as decimal text.
|
||||
pub(crate) replay_repair_total: String,
|
||||
/// Cumulative redundant-coverage repair count encoded as decimal text.
|
||||
pub(crate) redundant_coverage_repair_total: String,
|
||||
/// Cumulative bounded HTTP scan repair count encoded as decimal text.
|
||||
pub(crate) http_scan_repair_total: String,
|
||||
/// Cumulative direct block-fetch repair count encoded as decimal text.
|
||||
pub(crate) repair_block_fetch_total: String,
|
||||
/// Cumulative transaction-hydration repair count encoded as decimal text.
|
||||
pub(crate) repair_transaction_hydration_total: String,
|
||||
/// Oldest currently open gap start slot encoded as decimal text when present.
|
||||
pub(crate) oldest_open_gap_start_slot: std::option::Option<String>,
|
||||
/// Bounded source-neutral details for current and recent continuity gaps.
|
||||
pub(crate) gaps: std::vec::Vec<crate::RawIngestRouteGapDto>,
|
||||
}
|
||||
|
||||
/// Projects one complete concrete Worker snapshot to the safe route monitoring contract.
|
||||
pub(crate) fn project_route_monitoring(
|
||||
runtime: &crate::RawIngestRouteRuntimeDto,
|
||||
snapshot: &ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot,
|
||||
) -> ksp_core_lib::Result<crate::RawIngestRouteMonitoringDto> {
|
||||
let worker = snapshot.worker_snapshot();
|
||||
let fault = worker.state().fault_code();
|
||||
let admission_queue_capacity = usize_to_u32(snapshot.admission_queue_capacity(), "admission_queue_capacity");
|
||||
let admission_queue_capacity = match admission_queue_capacity {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let admission_queue_depth = usize_to_u32(snapshot.admission_queue_depth(), "admission_queue_depth");
|
||||
let admission_queue_depth = match admission_queue_depth {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let persistence_concurrency = usize_to_u32(snapshot.persistence_concurrency(), "persistence_concurrency");
|
||||
let persistence_concurrency = match persistence_concurrency {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let in_flight_persistence = usize_to_u32(snapshot.in_flight_persistence(), "in_flight_persistence");
|
||||
let in_flight_persistence = match in_flight_persistence {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let hydration_pending = usize_to_u32(snapshot.hydration_pending(), "hydration_pending");
|
||||
let hydration_pending = match hydration_pending {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let source_total = usize_to_u32(snapshot.source_total(), "source_total");
|
||||
let source_total = match source_total {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let source_active = usize_to_u32(snapshot.source_active(), "source_active");
|
||||
let source_active = match source_active {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let source_reconnecting = usize_to_u32(snapshot.source_reconnecting(), "source_reconnecting");
|
||||
let source_reconnecting = match source_reconnecting {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let source_failed = usize_to_u32(snapshot.source_failed(), "source_failed");
|
||||
let source_failed = match source_failed {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let open_gap_count = usize_to_u32(snapshot.open_gap_count(), "open_gap_count");
|
||||
let open_gap_count = match open_gap_count {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let repairing_gap_count = usize_to_u32(snapshot.repairing_gap_count(), "repairing_gap_count");
|
||||
let repairing_gap_count = match repairing_gap_count {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let gaps = snapshot.gaps().iter().map(project_gap).collect::<std::vec::Vec<_>>();
|
||||
return std::result::Result::Ok(crate::RawIngestRouteMonitoringDto {
|
||||
commitment: runtime.commitment,
|
||||
inventory_generation: runtime.inventory_generation,
|
||||
network: runtime.network.clone(),
|
||||
profile_id: runtime.profile_id.clone(),
|
||||
route_id: runtime.route_id,
|
||||
sequence: worker.sequence().value().to_string(),
|
||||
state: project_worker_state(worker.state()),
|
||||
health: worker.health().code().to_owned(),
|
||||
activity: worker.activity().code().to_owned(),
|
||||
terminal: worker.state().is_terminal(),
|
||||
fault_domain: fault.map(|code| return code.domain().to_owned()),
|
||||
fault_code: fault.map(|code| return code.code().to_owned()),
|
||||
admission_queue_capacity,
|
||||
admission_queue_depth,
|
||||
persistence_concurrency,
|
||||
in_flight_persistence,
|
||||
admitted_total: snapshot.admitted_total().to_string(),
|
||||
canonicalized_total: snapshot.canonicalized_total().to_string(),
|
||||
persisted_total: snapshot.persisted_total().to_string(),
|
||||
entity_inserted_total: snapshot.entity_inserted_total().to_string(),
|
||||
entity_already_present_total: snapshot.entity_already_present_total().to_string(),
|
||||
entity_skipped_purged_total: snapshot.entity_skipped_purged_total().to_string(),
|
||||
observation_inserted_total: snapshot.observation_inserted_total().to_string(),
|
||||
observation_already_present_total: snapshot.observation_already_present_total().to_string(),
|
||||
content_conflict_total: snapshot.content_conflict_total().to_string(),
|
||||
store_failure_total: snapshot.store_failure_total().to_string(),
|
||||
source_failure_total: snapshot.source_failure_total().to_string(),
|
||||
backpressure_wait_total: snapshot.backpressure_wait_total().to_string(),
|
||||
hydration_pending,
|
||||
processing_frontier_slot: slot_text(snapshot.processing_frontier_slot()),
|
||||
oldest_pending_slot: slot_text(snapshot.oldest_pending_slot()),
|
||||
continuity_policy_observed: snapshot.continuity_policy_observed(),
|
||||
continuity_frontier_slot: slot_text(snapshot.continuity_frontier_slot()),
|
||||
continuity_has_open_gaps: snapshot.continuity_has_open_gaps(),
|
||||
failed_source_losses_reconciled: snapshot.failed_source_losses_reconciled(),
|
||||
future_target_coverage: snapshot.future_target_coverage(),
|
||||
source_state: snapshot.source_state().map(|state| return source_state_code(state).to_owned()),
|
||||
source_total,
|
||||
source_active,
|
||||
source_reconnecting,
|
||||
source_failed,
|
||||
source_reconnect_total: snapshot.source_reconnect_total().to_string(),
|
||||
source_replay_attempt_total: snapshot.source_replay_attempt_total().to_string(),
|
||||
source_continuity_gap_total: snapshot.source_continuity_gap_total().to_string(),
|
||||
open_gap_count,
|
||||
repairing_gap_count,
|
||||
repaired_gap_total: snapshot.repaired_gap_total().to_string(),
|
||||
unresolved_gap_total: snapshot.unresolved_gap_total().to_string(),
|
||||
replay_repair_total: snapshot.replay_repair_total().to_string(),
|
||||
redundant_coverage_repair_total: snapshot.redundant_coverage_repair_total().to_string(),
|
||||
http_scan_repair_total: snapshot.http_scan_repair_total().to_string(),
|
||||
repair_block_fetch_total: snapshot.repair_block_fetch_total().to_string(),
|
||||
repair_transaction_hydration_total: snapshot.repair_transaction_hydration_total().to_string(),
|
||||
oldest_open_gap_start_slot: slot_text(snapshot.oldest_open_gap_start_slot()),
|
||||
gaps,
|
||||
});
|
||||
}
|
||||
|
||||
fn project_gap(gap: &ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot) -> crate::RawIngestRouteGapDto {
|
||||
return crate::RawIngestRouteGapDto {
|
||||
gap_id: gap.gap_id().value().to_string(),
|
||||
start_slot: gap.start_slot().to_string(),
|
||||
end_slot: gap.end_slot().to_string(),
|
||||
state: gap_state_code(gap.state()).to_owned(),
|
||||
reason: gap_reason_code(gap.reason()).to_owned(),
|
||||
last_method: gap.last_method().map(|method| return repair_method_code(method).to_owned()),
|
||||
};
|
||||
}
|
||||
|
||||
fn project_worker_state(state: ksp_worker_api::WorkerState) -> crate::RawIngestRouteState {
|
||||
return match state {
|
||||
ksp_worker_api::WorkerState::Created | ksp_worker_api::WorkerState::Starting => crate::RawIngestRouteState::Starting,
|
||||
ksp_worker_api::WorkerState::Running => crate::RawIngestRouteState::Running,
|
||||
ksp_worker_api::WorkerState::Stopping => crate::RawIngestRouteState::Stopping,
|
||||
ksp_worker_api::WorkerState::Stopped => crate::RawIngestRouteState::Stopped,
|
||||
ksp_worker_api::WorkerState::Faulted(_) => crate::RawIngestRouteState::Faulted,
|
||||
_ => crate::RawIngestRouteState::Faulted,
|
||||
};
|
||||
}
|
||||
|
||||
fn source_state_code(state: ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState) -> &'static str {
|
||||
return match state {
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Active => "active",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Reconnecting => "reconnecting",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Closing => "closing",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Closed => "closed",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Failed => "failed",
|
||||
};
|
||||
}
|
||||
|
||||
fn gap_state_code(state: ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapState) -> &'static str {
|
||||
return match state {
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapState::Pending => "pending",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapState::Repairing => "repairing",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapState::Repaired => "repaired",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapState::Unresolved => "unresolved",
|
||||
};
|
||||
}
|
||||
|
||||
fn gap_reason_code(reason: ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason) -> &'static str {
|
||||
return match reason {
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason::HttpProducedBlockUnavailable => "http_produced_block_unavailable",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason::KnownReferenceMissing => "known_reference_missing",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason::SourceFailure => "source_failure",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason::TransportOverflow => "transport_overflow",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason::WebSocketReconnect => "websocket_reconnect",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason::YellowstoneRetention => "yellowstone_retention",
|
||||
};
|
||||
}
|
||||
|
||||
fn repair_method_code(method: ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRepairMethod) -> &'static str {
|
||||
return match method {
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRepairMethod::Replay => "replay",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRepairMethod::RedundantCoverage => "redundant_coverage",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRepairMethod::HttpScan => "http_scan",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRepairMethod::BlockFetch => "block_fetch",
|
||||
ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRepairMethod::TransactionHydration => "transaction_hydration",
|
||||
};
|
||||
}
|
||||
|
||||
fn slot_text(value: std::option::Option<u64>) -> std::option::Option<String> {
|
||||
return value.map(|slot| return slot.to_string());
|
||||
}
|
||||
|
||||
fn usize_to_u32(value: usize, field: &'static str) -> ksp_core_lib::Result<u32> {
|
||||
let converted = u32::try_from(value);
|
||||
return match converted {
|
||||
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
||||
std::result::Result::Err(error) => std::result::Result::Err(
|
||||
ksp_core_lib::Error::new(
|
||||
crate::ERROR_CODE_APP_STATE_INVALID,
|
||||
"Raw Transaction Ingest Desk cannot project a bounded Worker counter to the frontend",
|
||||
)
|
||||
.with_context("field", field)
|
||||
.with_source(error),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "../unit_tests/route_monitoring.rs"]
|
||||
mod tests;
|
||||
@@ -1,5 +1,5 @@
|
||||
// file: crates/ksp-app-raw-transaction-ingest-desk/src/route_runtime.rs
|
||||
// version: 5
|
||||
// version: 6
|
||||
|
||||
//! Multi-route Store and independent Worker lifecycle owned by Raw Transaction Ingest Desk.
|
||||
|
||||
@@ -8,6 +8,7 @@ const STORE_RECLAIM_POLL_INTERVAL: std::time::Duration = std::time::Duration::fr
|
||||
const STORE_RECLAIM_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1);
|
||||
|
||||
struct RouteRuntimeInner {
|
||||
last_monitoring: std::vec::Vec<crate::RawIngestRouteMonitoringDto>,
|
||||
last_terminals: std::vec::Vec<crate::RawIngestRouteRuntimeDto>,
|
||||
network: std::option::Option<String>,
|
||||
next_sequence: u64,
|
||||
@@ -28,6 +29,7 @@ impl crate::RouteRuntimeState {
|
||||
pub(crate) fn new() -> Self {
|
||||
return Self {
|
||||
inner: std::sync::Mutex::new(RouteRuntimeInner {
|
||||
last_monitoring: std::vec::Vec::new(),
|
||||
last_terminals: std::vec::Vec::new(),
|
||||
network: std::option::Option::None,
|
||||
next_sequence: 0,
|
||||
@@ -99,9 +101,11 @@ impl crate::RouteRuntimeState {
|
||||
route_id: prepared.route_id,
|
||||
};
|
||||
if inner.network.is_none() {
|
||||
inner.last_monitoring.clear();
|
||||
inner.last_terminals.clear();
|
||||
inner.network = std::option::Option::Some(prepared.network.clone());
|
||||
}
|
||||
inner.last_monitoring.retain(|status| return !identity.matches_monitoring(status));
|
||||
inner.last_terminals.retain(|terminal| return !identity.matches_dto(terminal));
|
||||
inner.next_sequence = sequence;
|
||||
if needs_store_open {
|
||||
@@ -196,7 +200,12 @@ impl crate::RouteRuntimeState {
|
||||
return std::result::Result::Ok(std::option::Option::None);
|
||||
}
|
||||
|
||||
fn finish(&self, token: u64, terminal: crate::RawIngestRouteRuntimeDto) -> ksp_core_lib::Result<std::option::Option<std::sync::Arc<ksp_store_lib::Store>>> {
|
||||
fn finish(
|
||||
&self,
|
||||
token: u64,
|
||||
terminal: crate::RawIngestRouteRuntimeDto,
|
||||
monitoring: std::option::Option<crate::RawIngestRouteMonitoringDto>,
|
||||
) -> ksp_core_lib::Result<std::option::Option<std::sync::Arc<ksp_store_lib::Store>>> {
|
||||
let inner = self.inner.lock();
|
||||
let mut inner = match inner {
|
||||
std::result::Result::Ok(value) => value,
|
||||
@@ -209,6 +218,10 @@ impl crate::RouteRuntimeState {
|
||||
}
|
||||
inner.last_terminals.retain(|previous| return previous.profile_id != terminal.profile_id || previous.route_id != terminal.route_id);
|
||||
inner.last_terminals.push(terminal);
|
||||
if let std::option::Option::Some(status) = monitoring {
|
||||
inner.last_monitoring.retain(|previous| return previous.profile_id != status.profile_id || previous.route_id != status.route_id);
|
||||
inner.last_monitoring.push(status);
|
||||
}
|
||||
if inner.routes.is_empty() {
|
||||
let store = inner.store.take();
|
||||
if store.is_some() {
|
||||
@@ -235,6 +248,30 @@ impl crate::RouteRuntimeState {
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns complete latest-value monitoring projections for active and recent terminal routes in the current runtime session.
|
||||
pub(crate) fn monitoring_statuses(&self) -> ksp_core_lib::Result<std::vec::Vec<crate::RawIngestRouteMonitoringDto>> {
|
||||
let inner = self.inner.lock();
|
||||
let inner = match inner {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => return std::result::Result::Err(runtime_lock_error()),
|
||||
};
|
||||
let mut values = inner.last_monitoring.clone();
|
||||
for slot in &inner.routes {
|
||||
if let RouteRuntimeSlot::Active { handle, identity, .. } = slot {
|
||||
let snapshot = handle.snapshot_source().current();
|
||||
let runtime = identity.dto(project_worker_state(snapshot.worker_snapshot().state()));
|
||||
let projected = crate::project_route_monitoring(&runtime, &snapshot);
|
||||
let projected = match projected {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
values.retain(|previous| return previous.profile_id != projected.profile_id || previous.route_id != projected.route_id);
|
||||
values.push(projected);
|
||||
}
|
||||
}
|
||||
return std::result::Result::Ok(values);
|
||||
}
|
||||
|
||||
/// Requests cooperative Stop for one exact logical route and waits for its Worker cleanup.
|
||||
pub(crate) async fn stop_and_wait(&self, request: &crate::RawIngestRouteStopRequestDto) -> ksp_core_lib::Result<crate::RawIngestRouteRuntimeDto> {
|
||||
let target = self.stop_target(request);
|
||||
@@ -389,6 +426,18 @@ impl crate::RouteRuntimeLaunch {
|
||||
return self.acknowledgement.clone();
|
||||
}
|
||||
|
||||
/// Returns the complete safe logical route identity used to project Worker monitoring snapshots.
|
||||
#[must_use]
|
||||
pub(crate) fn monitoring_identity(&self) -> crate::RawIngestRouteRuntimeDto {
|
||||
return self.acknowledgement.clone();
|
||||
}
|
||||
|
||||
/// Returns an independent latest-value source for frontend-safe monitoring relay.
|
||||
#[must_use]
|
||||
pub(crate) fn monitoring_source(&self) -> ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshotSource {
|
||||
return self.handle.snapshot_source();
|
||||
}
|
||||
|
||||
/// Waits for this Worker terminal state, removes only its route and closes Store only after the final same-network route terminates.
|
||||
pub(crate) async fn monitor(self) {
|
||||
let terminal = self.handle.wait_terminal().await;
|
||||
@@ -415,7 +464,22 @@ impl crate::RouteRuntimeLaunch {
|
||||
};
|
||||
let mut terminal = self.acknowledgement;
|
||||
terminal.state = terminal_state;
|
||||
let store = self.runtime_state.finish(self.token, terminal);
|
||||
let snapshot = self.handle.snapshot_source().current();
|
||||
let monitoring = crate::project_route_monitoring(&terminal, &snapshot);
|
||||
let monitoring = match monitoring {
|
||||
std::result::Result::Ok(value) => std::option::Option::Some(value),
|
||||
std::result::Result::Err(error) => {
|
||||
ksp_logging_lib::warn!(
|
||||
target: crate::TRACING_TARGET,
|
||||
domain = crate::TRACING_DOMAIN_ROUTE_RUNTIME,
|
||||
error_domain = error.code().domain(),
|
||||
error_code = error.code().code(),
|
||||
"Raw Transaction Ingest Desk could not retain terminal route monitoring projection"
|
||||
);
|
||||
std::option::Option::None
|
||||
},
|
||||
};
|
||||
let store = self.runtime_state.finish(self.token, terminal, monitoring);
|
||||
let store = match store {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => {
|
||||
@@ -592,6 +656,10 @@ impl RouteRuntimeIdentity {
|
||||
fn matches_dto(&self, value: &crate::RawIngestRouteRuntimeDto) -> bool {
|
||||
return self.profile_id == value.profile_id && self.route_id == value.route_id;
|
||||
}
|
||||
|
||||
fn matches_monitoring(&self, value: &crate::RawIngestRouteMonitoringDto) -> bool {
|
||||
return self.profile_id == value.profile_id && self.route_id == value.route_id;
|
||||
}
|
||||
}
|
||||
|
||||
enum RouteStopTarget {
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
// file: crates/ksp-app-raw-transaction-ingest-desk/src/tauri.rs
|
||||
// version: 6
|
||||
// version: 7
|
||||
|
||||
//! Tauri runtime assembly for the KSP Raw Transaction Ingest desktop application.
|
||||
|
||||
use tauri::Emitter; // rust-rules: trait-import
|
||||
use tauri::Manager; // rust-rules: trait-import
|
||||
|
||||
/// Runs the Raw Transaction Ingest desktop application.
|
||||
#[cfg_attr(mobile, tauri::mobile_entry_point)]
|
||||
pub fn run(arguments: &[std::ffi::OsString]) -> ksp_core_lib::Result<()> {
|
||||
@@ -78,6 +81,7 @@ fn configure_commands(builder: tauri::Builder<tauri::Wry>) -> tauri::Builder<tau
|
||||
emit_frontend_log,
|
||||
get_route_foundation,
|
||||
get_route_inventory,
|
||||
get_route_monitoring,
|
||||
get_runtime_status,
|
||||
splash_frontend_ready,
|
||||
start_route,
|
||||
@@ -121,6 +125,17 @@ fn get_route_inventory(state: tauri::State<'_, crate::AppState>) -> std::result:
|
||||
};
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
fn get_route_monitoring(
|
||||
state: tauri::State<'_, crate::AppState>,
|
||||
) -> std::result::Result<std::vec::Vec<crate::RawIngestRouteMonitoringDto>, crate::CommandErrorDto> {
|
||||
let result = state.route_monitoring();
|
||||
return match result {
|
||||
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
||||
std::result::Result::Err(error) => std::result::Result::Err(project_command_error("get_route_monitoring", crate::TRACING_DOMAIN_ROUTE_RUNTIME, &error)),
|
||||
};
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
fn get_runtime_status(state: tauri::State<'_, crate::AppState>) -> std::result::Result<crate::ShellStatusDto, crate::CommandErrorDto> {
|
||||
let result = state.shell_status();
|
||||
@@ -145,6 +160,7 @@ async fn splash_frontend_ready(
|
||||
|
||||
#[tauri::command]
|
||||
async fn start_route(
|
||||
app: tauri::AppHandle,
|
||||
request: crate::RawIngestRouteStartRequestDto,
|
||||
state: tauri::State<'_, crate::AppState>,
|
||||
) -> std::result::Result<crate::RawIngestRouteRuntimeDto, crate::CommandErrorDto> {
|
||||
@@ -156,13 +172,60 @@ async fn start_route(
|
||||
},
|
||||
};
|
||||
let acknowledgement = launch.acknowledgement();
|
||||
let monitor = tauri::async_runtime::spawn(async move {
|
||||
let monitoring_identity = launch.monitoring_identity();
|
||||
let monitoring_source = launch.monitoring_source();
|
||||
let monitoring_app = app.clone();
|
||||
let monitoring_task = tauri::async_runtime::spawn(async move {
|
||||
monitor_route_status(monitoring_app, monitoring_identity, monitoring_source).await;
|
||||
});
|
||||
std::mem::drop(monitoring_task);
|
||||
let terminal_task = tauri::async_runtime::spawn(async move {
|
||||
launch.monitor().await;
|
||||
});
|
||||
std::mem::drop(monitor);
|
||||
std::mem::drop(terminal_task);
|
||||
return std::result::Result::Ok(acknowledgement);
|
||||
}
|
||||
|
||||
async fn monitor_route_status(
|
||||
app: tauri::AppHandle,
|
||||
runtime: crate::RawIngestRouteRuntimeDto,
|
||||
source: ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshotSource,
|
||||
) {
|
||||
let mut snapshot = source.current();
|
||||
loop {
|
||||
let projected = crate::project_route_monitoring(&runtime, &snapshot);
|
||||
match projected {
|
||||
std::result::Result::Ok(status) => {
|
||||
let main = app.get_webview_window("main");
|
||||
if let std::option::Option::Some(window) = main {
|
||||
let emitted = window.emit(crate::RAW_INGEST_ROUTE_STATUS_EVENT_NAME, status);
|
||||
if emitted.is_err() {
|
||||
ksp_logging_lib::warn!(
|
||||
target: crate::TRACING_TARGET,
|
||||
domain = crate::TRACING_DOMAIN_ROUTE_RUNTIME,
|
||||
"Raw Transaction Ingest Desk could not emit latest-value route monitoring status"
|
||||
);
|
||||
}
|
||||
}
|
||||
},
|
||||
std::result::Result::Err(error) => {
|
||||
ksp_logging_lib::warn!(
|
||||
target: crate::TRACING_TARGET,
|
||||
domain = crate::TRACING_DOMAIN_ROUTE_RUNTIME,
|
||||
error_domain = error.code().domain(),
|
||||
error_code = error.code().code(),
|
||||
"Raw Transaction Ingest Desk could not project latest-value route monitoring status"
|
||||
);
|
||||
},
|
||||
}
|
||||
if snapshot.worker_snapshot().state().is_terminal() {
|
||||
return;
|
||||
}
|
||||
let observed = snapshot.worker_snapshot().sequence();
|
||||
snapshot = source.wait_for_change(observed).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
async fn stop_route(
|
||||
request: crate::RawIngestRouteStopRequestDto,
|
||||
|
||||
Reference in New Issue
Block a user