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

0.3.15-pre.012

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-16 10:58:29 +02:00
parent 2f587e22be
commit 31ae38853d
21 changed files with 815 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
crates/ksp-app-raw-transaction-ingest-desk/src/route_runtime.rs
View File

@@ -1,9 +1,10 @@
// file: crates/ksp-app-raw-transaction-ingest-desk/src/route_runtime.rs
// version: 6
// version: 7
//! Multi-route Store and independent Worker lifecycle owned by Raw Transaction Ingest Desk.
const ROUTE_STOP_CLEANUP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
const ROUTE_SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
const ROUTE_STOP_CLEANUP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
const STORE_RECLAIM_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(5);
const STORE_RECLAIM_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1);
@@ -13,6 +14,7 @@ struct RouteRuntimeInner {
network: std::option::Option<String>,
next_sequence: u64,
routes: std::vec::Vec<RouteRuntimeSlot>,
shutting_down: bool,
store: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
store_closing_token: std::option::Option<u64>,
store_opening_token: std::option::Option<u64>,
@@ -34,6 +36,7 @@ impl crate::RouteRuntimeState {
network: std::option::Option::None,
next_sequence: 0,
routes: std::vec::Vec::new(),
shutting_down: false,
store: std::option::Option::None,
store_closing_token: std::option::Option::None,
store_opening_token: std::option::Option::None,
@@ -41,12 +44,34 @@ impl crate::RouteRuntimeState {
};
}
/// Rejects new route Start work after application shutdown has closed admission.
pub(crate) fn ensure_start_admission_open(&self) -> ksp_core_lib::Result<()> {
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()),
};
if inner.shutting_down {
return std::result::Result::Err(ksp_core_lib::Error::new(
crate::ERROR_CODE_ROUTE_RUNTIME_SHUTTING_DOWN,
"Raw Transaction Ingest Desk cannot start a route while application shutdown is in progress",
));
}
return std::result::Result::Ok(());
}
fn reserve(&self, prepared: &crate::PreparedRouteStart) -> ksp_core_lib::Result<RouteRuntimeReservation> {
let inner = self.inner.lock();
let mut inner = match inner {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(runtime_lock_error()),
};
if inner.shutting_down {
return std::result::Result::Err(ksp_core_lib::Error::new(
crate::ERROR_CODE_ROUTE_RUNTIME_SHUTTING_DOWN,
"Raw Transaction Ingest Desk cannot start a route while application shutdown is in progress",
));
}
if inner.store_closing_token.is_some() {
return std::result::Result::Err(ksp_core_lib::Error::new(
crate::ERROR_CODE_ROUTE_RUNTIME_ACTIVE,
@@ -111,7 +136,7 @@ impl crate::RouteRuntimeState {
if needs_store_open {
inner.store_opening_token = std::option::Option::Some(sequence);
}
inner.routes.push(RouteRuntimeSlot::Starting { identity: identity.clone(), token: sequence });
inner.routes.push(RouteRuntimeSlot::Starting { identity: identity.clone(), stop_requested: false, token: sequence });
return std::result::Result::Ok(RouteRuntimeReservation { identity, needs_store_open, token: sequence, worker_id });
}
@@ -173,7 +198,19 @@ impl crate::RouteRuntimeState {
));
},
};
let state = project_worker_state(handle.snapshot_source().current().worker_snapshot().state());
let stop_requested = match &inner.routes[position] {
RouteRuntimeSlot::Starting { stop_requested, .. } => *stop_requested,
RouteRuntimeSlot::Active { .. } => false,
};
let should_stop = stop_requested || inner.shutting_down;
if should_stop {
let _accepted = handle.request_stop();
}
let state = if should_stop {
crate::RawIngestRouteState::Stopping
} else {
project_worker_state(handle.snapshot_source().current().worker_snapshot().state())
};
inner.routes[position] = RouteRuntimeSlot::Active { handle, identity: reservation.identity.clone(), token: reservation.token };
return std::result::Result::Ok(reservation.identity.dto(state));
}
@@ -184,7 +221,12 @@ impl crate::RouteRuntimeState {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(runtime_lock_error()),
};
let cancelled_terminal = inner.routes.iter().find(|slot| return slot.token() == token).and_then(RouteRuntimeSlot::cancelled_start_terminal);
inner.routes.retain(|slot| return slot.token() != token);
if let std::option::Option::Some(terminal) = cancelled_terminal {
inner.last_terminals.retain(|previous| return previous.profile_id != terminal.profile_id || previous.route_id != terminal.route_id);
inner.last_terminals.push(terminal);
}
if inner.store_opening_token == std::option::Option::Some(token) {
inner.store_opening_token = std::option::Option::None;
}
@@ -272,8 +314,71 @@ impl crate::RouteRuntimeState {
return std::result::Result::Ok(values);
}
/// Closes route Start admission exactly once and requests cooperative Stop for all active or still-starting routes.
pub(crate) fn begin_shutdown(&self) -> ksp_core_lib::Result<bool> {
let inner = self.inner.lock();
let mut inner = match inner {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(runtime_lock_error()),
};
if inner.shutting_down {
return std::result::Result::Ok(false);
}
inner.shutting_down = true;
for slot in &mut inner.routes {
match slot {
RouteRuntimeSlot::Starting { stop_requested, .. } => *stop_requested = true,
RouteRuntimeSlot::Active { handle, .. } => {
let _accepted = handle.request_stop();
},
}
}
return std::result::Result::Ok(true);
}
/// Waits until all owned route Workers and the shared Store have completed bounded application-shutdown cleanup.
pub(crate) async fn shutdown_and_wait(&self) -> ksp_core_lib::Result<()> {
let begin = self.begin_shutdown();
if let std::result::Result::Err(error) = begin {
return std::result::Result::Err(error);
}
let started = std::time::Instant::now();
loop {
let complete = self.shutdown_complete();
let complete = match complete {
std::result::Result::Ok(value) => value,
std::result::Result::Err(error) => return std::result::Result::Err(error),
};
if complete {
return std::result::Result::Ok(());
}
if started.elapsed() >= ROUTE_SHUTDOWN_TIMEOUT {
return std::result::Result::Err(ksp_core_lib::Error::new(
crate::ERROR_CODE_ROUTE_RUNTIME_SHUTDOWN_FAILED,
"Raw Transaction Ingest Desk application shutdown did not release all route Workers and shared Store state before the bounded deadline",
));
}
tokio::time::sleep(STORE_RECLAIM_POLL_INTERVAL).await;
}
}
fn shutdown_complete(&self) -> ksp_core_lib::Result<bool> {
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()),
};
return std::result::Result::Ok(
inner.routes.is_empty() && inner.store.is_none() && inner.store_opening_token.is_none() && inner.store_closing_token.is_none(),
);
}
/// 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 identity = request.validate_logical_identity();
if let std::result::Result::Err(error) = identity {
return std::result::Result::Err(error);
}
let target = self.stop_target(request);
let target = match target {
std::result::Result::Ok(value) => value,
@@ -281,6 +386,7 @@ impl crate::RouteRuntimeState {
};
let (handle, identity, token) = match target {
RouteStopTarget::Active { handle, identity, token } => (handle, identity, token),
RouteStopTarget::Starting { identity, token } => return self.wait_route_cleanup(token, &identity).await,
RouteStopTarget::Terminal(value) => return std::result::Result::Ok(value),
RouteStopTarget::TerminalClosing => return self.wait_terminal_store_cleanup(request).await,
};
@@ -296,33 +402,30 @@ impl crate::RouteRuntimeState {
fn stop_target(&self, request: &crate::RawIngestRouteStopRequestDto) -> ksp_core_lib::Result<RouteStopTarget> {
let inner = self.inner.lock();
let inner = match inner {
let mut inner = match inner {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => return std::result::Result::Err(runtime_lock_error()),
};
let slot = inner.routes.iter().find(|slot| return slot.matches_identity(request.profile_id.as_str(), request.route_id));
return match slot {
std::option::Option::Some(RouteRuntimeSlot::Active { handle, identity, token }) => {
std::result::Result::Ok(RouteStopTarget::Active { handle: handle.clone(), identity: identity.clone(), token: *token })
},
std::option::Option::Some(RouteRuntimeSlot::Starting { .. }) => std::result::Result::Err(ksp_core_lib::Error::new(
let position = inner.routes.iter().position(|slot| return slot.matches_identity(request.profile_id.as_str(), request.route_id));
if let std::option::Option::Some(position) = position {
return match &mut inner.routes[position] {
RouteRuntimeSlot::Active { handle, identity, token } => {
std::result::Result::Ok(RouteStopTarget::Active { handle: handle.clone(), identity: identity.clone(), token: *token })
},
RouteRuntimeSlot::Starting { identity, stop_requested, token } => {
*stop_requested = true;
std::result::Result::Ok(RouteStopTarget::Starting { identity: identity.clone(), token: *token })
},
};
}
let terminal = inner.last_terminals.iter().find(|terminal| return terminal.profile_id == request.profile_id && terminal.route_id == request.route_id);
return match terminal {
std::option::Option::Some(value) if inner.store_closing_token.is_none() => std::result::Result::Ok(RouteStopTarget::Terminal(value.clone())),
std::option::Option::Some(_) => std::result::Result::Ok(RouteStopTarget::TerminalClosing),
std::option::Option::None => std::result::Result::Err(ksp_core_lib::Error::new(
crate::ERROR_CODE_ROUTE_RUNTIME_NOT_ACTIVE,
"Raw Transaction Ingest Desk route Worker is still starting and cannot be stopped yet",
"Raw Transaction Ingest Desk has no active Worker for the selected logical route",
)),
std::option::Option::None => {
let terminal =
inner.last_terminals.iter().find(|terminal| return terminal.profile_id == request.profile_id && terminal.route_id == request.route_id);
match terminal {
std::option::Option::Some(value) if inner.store_closing_token.is_none() => {
std::result::Result::Ok(RouteStopTarget::Terminal(value.clone()))
},
std::option::Option::Some(_) => std::result::Result::Ok(RouteStopTarget::TerminalClosing),
std::option::Option::None => std::result::Result::Err(ksp_core_lib::Error::new(
crate::ERROR_CODE_ROUTE_RUNTIME_NOT_ACTIVE,
"Raw Transaction Ingest Desk has no active Worker for the selected logical route",
)),
}
},
};
}
@@ -668,6 +771,10 @@ enum RouteStopTarget {
identity: RouteRuntimeIdentity,
token: u64,
},
Starting {
identity: RouteRuntimeIdentity,
token: u64,
},
Terminal(crate::RawIngestRouteRuntimeDto),
TerminalClosing,
}
@@ -682,6 +789,7 @@ struct RouteRuntimeReservation {
enum RouteRuntimeSlot {
Starting {
identity: RouteRuntimeIdentity,
stop_requested: bool,
token: u64,
},
Active {
@@ -706,6 +814,13 @@ impl RouteRuntimeSlot {
};
}
fn cancelled_start_terminal(&self) -> std::option::Option<crate::RawIngestRouteRuntimeDto> {
return match self {
Self::Starting { identity, stop_requested: true, .. } => std::option::Option::Some(identity.dto(crate::RawIngestRouteState::Stopped)),
Self::Starting { stop_requested: false, .. } | Self::Active { .. } => std::option::Option::None,
};
}
fn token(&self) -> u64 {
return match self {
Self::Starting { token, .. } | Self::Active { token, .. } => *token,