Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/libs/sbx/mgr/server/src/lifecycle_metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ pub(crate) async fn run(db: crate::db::SharedConnection, shutdown_token: Cancell
continue;
}
};
crate::db::sandbox_lifecycle_counts(&db)
crate::db::read::counts(&db, crate::db::read::ByLifecycle)
};
let counts = match counts {
Ok(counts) => counts,
Expand Down
29 changes: 23 additions & 6 deletions crates/libs/sbx/mgr/server/src/local_artifacts.rs
Original file line number Diff line number Diff line change
Expand Up @@ -207,7 +207,10 @@ impl LocalArtifacts {
local_artifact_id: LocalArtifactId,
) -> crate::db::Result<Option<LocalArtifactRow>> {
let conn = self.db_conn()?;
crate::db::get_local_artifact(&conn, local_artifact_id)
crate::db::read::local_artifact(
&conn,
crate::db::read::LocalArtifactKey::Id(local_artifact_id),
)
}

/// Read-only fetch by `(sandbox_id, snapshot_name)` for the
Expand All @@ -218,7 +221,13 @@ impl LocalArtifacts {
snapshot_name: &SnapshotName,
) -> crate::db::Result<Option<LocalArtifactRow>> {
let conn = self.db_conn()?;
crate::db::get_named_local_artifact(&conn, sandbox_id, snapshot_name)
crate::db::read::local_artifact(
&conn,
crate::db::read::LocalArtifactKey::Named {
sandbox_id,
snapshot_name,
},
)
}

/// Every sandbox whose **current lifecycle artifact** this node cannot
Expand Down Expand Up @@ -330,7 +339,7 @@ impl LocalArtifacts {
/// with no report, which every consumer already has to handle.
pub fn list_pending_backups(&self) -> crate::db::Result<Vec<PendingBackup>> {
let conn = self.db_conn()?;
leases::list_pending_backups(&conn)
crate::db::scan::pending_backups(&conn)
}

/// Inventory-only observation. Callers must still acquire a checkout before
Expand All @@ -340,7 +349,9 @@ impl LocalArtifacts {
sandbox_id: &str,
) -> Result<Vec<NamedArtifactPublication>, crate::db::inventory_read::InventoryReadError> {
let conn = self.db_conn()?;
leases::list_named_publications(&conn, sandbox_id)
crate::db::read::sandbox_view(&conn, sandbox_id)
.artifacts()
.named()
}

/// Mark the artifact's remote backup as durably completed. This
Expand Down Expand Up @@ -603,7 +614,13 @@ impl LocalArtifacts {
) -> Result<DeleteLease, CheckoutError> {
let id = {
let conn = self.db_conn()?;
let row = crate::db::get_named_local_artifact(&conn, sandbox_id, snapshot_name)?;
let row = crate::db::read::local_artifact(
&conn,
crate::db::read::LocalArtifactKey::Named {
sandbox_id,
snapshot_name,
},
)?;
match row {
Some(r) => r.local_artifact_id,
None => {
Expand Down Expand Up @@ -775,7 +792,7 @@ impl LocalArtifacts {
/// Retain rows in every state, including historical deletion conflicts.
pub(crate) fn gc_staging_is_unowned(&self, id: LocalArtifactId) -> crate::db::Result<bool> {
let conn = self.db_conn()?;
leases::gc_staging_is_unowned(&conn, &id.to_string())
crate::db::read::local_artifact_unowned(&conn, id)
}

/// Examine at most `max_records` raw rows, advancing past owned, present,
Expand Down
2 changes: 1 addition & 1 deletion crates/libs/sbx/mgr/server/src/node_state_sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,7 @@ pub(crate) fn start_dirty_outbox_worker(
let Ok(db) = connection_or_signal_failure(&db) else {
return;
};
match crate::db::list_node_dirty_refs(&db, batch_limit) {
match crate::db::scan::node_dirty_batch(&db, batch_limit) {
Ok(rows) => rows,
Err(error) => {
tracing::warn!(%error, "failed to read node dirty outbox");
Expand Down
10 changes: 7 additions & 3 deletions crates/libs/sbx/mgr/server/src/route_activation_claims.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
use crate::db::Connection;

use crate::db::route_activation_claims::{OpenRouteActivationClaim, claim_matches};
use crate::db::route_activation_claims::OpenRouteActivationClaim;

#[derive(Debug, thiserror::Error)]
pub(crate) enum OpenRouteActivationAdmissionError {
Expand Down Expand Up @@ -39,10 +39,14 @@ pub(crate) fn validate_cluster_fenced_activation(
sandbox_id: &str,
claim: &OpenRouteActivationClaim,
) -> Result<(), OpenRouteActivationAdmissionError> {
let provenance = crate::db::activation_route_provenance(conn, sandbox_id)?;
let provenance =
crate::db::read::sandbox_view(conn, sandbox_id).activation_route_provenance()?;
let retained = observed_generation_floor(provenance);
let requested = claim.requested_generation();
if claim_matches(conn, sandbox_id, claim)? {
if crate::db::read::sandbox_view(conn, sandbox_id)
.route_claim()
.matches(claim)?
{
return Ok(());
}
if let Some(retained) = retained
Expand Down
40 changes: 26 additions & 14 deletions crates/libs/sbx/mgr/server/src/sandbox_eviction.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
//! Cold-sandbox eviction: free `/space/cache` held by parked suspended

Check warning on line 1 in crates/libs/sbx/mgr/server/src/sandbox_eviction.rs

View workflow job for this annotation

GitHub Actions / rust_static / Rust clippy

[archcheck] reported by reviewdog 🐶 crates/libs/sbx/mgr/server/src/sandbox_eviction.rs grew from 1253 to 1275 code lines. Once this rule is required, split the file before adding to it. Raw Output: message:"crates/libs/sbx/mgr/server/src/sandbox_eviction.rs grew from 1253 to 1275 code lines. Once this rule is required, split the file before adding to it." location:{path:"crates/libs/sbx/mgr/server/src/sandbox_eviction.rs" range:{start:{line:1}}} severity:WARNING source:{name:"archcheck"} code:{value:"file_size.listed_no_growth"}
//! sandboxes whose state is durably restorable from a ready remote
//! snapshot.
//!
Expand Down Expand Up @@ -96,7 +96,6 @@
use uuid::Uuid;

pub(crate) use crate::auto_lifecycle::now_secs;
use crate::db::eviction_scan::{count_malformed_suspended_updated_at, count_resident_rows};
use crate::service::SandboxManagerService;

/// Runtime configuration for the eviction loop. Loaded from env on
Expand Down Expand Up @@ -533,12 +532,14 @@
cursor.map(|cursor| i64::try_from(cursor.idle_witness_secs).unwrap_or(i64::MAX));
let cursor_sandbox_id = cursor.map(|cursor| cursor.sandbox_id.as_str());
let cutoff = i64::try_from(idle_cutoff_epoch_secs).unwrap_or(i64::MAX);
let rows = crate::db::eviction_scan::idle_suspended_candidates(
let rows = crate::db::scan::eviction_candidates(
conn,
cutoff,
cursor_witness,
cursor_sandbox_id,
limit,
crate::db::scan::Proofable {
cutoff,
cursor_witness,
cursor_sandbox_id,
limit,
},
)?;
Ok(rows
.into_iter()
Expand Down Expand Up @@ -579,7 +580,13 @@
idle_cutoff_epoch_secs: u64,
limit: usize,
) -> crate::db::Result<Vec<String>> {
crate::db::eviction_scan::legacy_reupload_candidates(conn, idle_cutoff_epoch_secs, limit)
crate::db::scan::eviction_candidates(
conn,
crate::db::scan::ProoflessLegacy {
idle_cutoff_epoch_secs,
limit,
},
)
}

/// Gather the local half of the eviction evidence plus the sandbox's
Expand All @@ -598,14 +605,18 @@
conn: &Connection,
sandbox_id: &str,
) -> crate::db::Result<Option<(LocalEvidence, u64)>> {
let row = crate::db::eviction_scan::artifact_pin_and_updated_at(conn, sandbox_id)?;
let row = crate::db::read::sandbox_view(conn, sandbox_id)
.pinned_artifact()
.pin_and_updated_at()?;
let Some((pin, updated_at)) = row else {
return Ok(None);
};
let updated_at_secs = updated_at.unsigned();

if let Some(pin) = pin {
let artifact = crate::db::eviction_scan::owned_artifact_evidence(conn, &pin, sandbox_id)?;
let artifact = crate::db::read::sandbox_view(conn, sandbox_id)
.pinned_artifact()
.owned_evidence(&pin)?;
let evidence = match artifact {
Some((kind, id, backup_marker)) => {
match (id.parse(), kind.parse()) {
Expand All @@ -627,7 +638,7 @@
// Unpinned: the newest upload job regardless of status (a failed
// newest upload must still count — it proves newer-than-backed
// local state exists).
let job = crate::db::eviction_scan::latest_upload_job(conn, sandbox_id)?;
let job = crate::db::read::sandbox_view(conn, sandbox_id).latest_upload()?;
let latest_job = job.and_then(|(snapshot_id, created_at)| {
let snapshot_id = snapshot_id?.trim().parse::<SnapshotId>().ok()?;
// Invalid time cannot prove recency, even when the sandbox itself
Expand Down Expand Up @@ -1239,10 +1250,11 @@
let conn = service_for_counts
.db_conn()
.map_err(|error| format!("eviction row counts db_conn: {error}"))?;
let malformed = count_malformed_suspended_updated_at(&conn)
.map_err(|error| format!("count_malformed_suspended_updated_at: {error}"))?;
let resident =
count_resident_rows(&conn).map_err(|error| format!("count_resident_rows: {error}"))?;
let malformed =
crate::db::read::counts(&conn, crate::db::read::MalformedSuspendedUpdatedAt)
.map_err(|error| format!("count_malformed_suspended_updated_at: {error}"))?;
let resident = crate::db::read::counts(&conn, crate::db::read::Resident)
.map_err(|error| format!("count_resident_rows: {error}"))?;
Ok::<_, String>((malformed, resident))
})
.await
Expand Down
1 change: 1 addition & 0 deletions crates/libs/sbx/mgr/server/src/sandbox_eviction/tests.rs
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use crate::db::eviction_scan::{count_malformed_suspended_updated_at, count_resident_rows};
use crate::db::fixture::params;
use pplx_space_common_core::types::{LocalArtifactId, SnapshotId};

Expand Down
20 changes: 12 additions & 8 deletions crates/libs/sbx/mgr/server/src/service/activation_support.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use super::network_runtime::{network_policy_refresh_error_to_status, refresh_network_policy};

Check warning on line 1 in crates/libs/sbx/mgr/server/src/service/activation_support.rs

View workflow job for this annotation

GitHub Actions / rust_static / Rust clippy

[archcheck] reported by reviewdog 🐶 crates/libs/sbx/mgr/server/src/service/activation_support.rs grew from 1574 to 1582 code lines. Once this rule is required, split the file before adding to it. Raw Output: message:"crates/libs/sbx/mgr/server/src/service/activation_support.rs grew from 1574 to 1582 code lines. Once this rule is required, split the file before adding to it." location:{path:"crates/libs/sbx/mgr/server/src/service/activation_support.rs" range:{start:{line:1}}} severity:WARNING source:{name:"archcheck"} code:{value:"file_size.listed_no_growth"}
use super::spaced_bridge::cleanup_restored_snapshot_dir_async;
use super::start_support::ActivationCleanupGuard;
use super::*;
Expand Down Expand Up @@ -364,7 +364,7 @@
}
let provenance = {
let db = self.db_conn()?;
match crate::db::activation_route_provenance(&db, sandbox_id) {
match crate::db::read::sandbox_view(&db, sandbox_id).activation_route_provenance() {
Ok(provenance) => {
if requested_can_open_claim
&& let Some(claim) = crate::db::route_activation_claims::OpenRouteActivationClaim::for_cluster_fenced_identity(
Expand Down Expand Up @@ -454,7 +454,7 @@
}
let provenance = {
let db = self.db_conn()?;
match crate::db::activation_route_provenance(&db, sandbox_id) {
match crate::db::read::sandbox_view(&db, sandbox_id).activation_route_provenance() {
Ok(provenance) => provenance,
Err(error) if error.is_no_rows() => {
return Err(
Expand Down Expand Up @@ -1003,10 +1003,12 @@
let read = || {
let conn = crate::node_state_sync::connection_or_signal_failure(db)
.map_err(|error| Status::internal(error.to_string()))?;
let sandbox = crate::db::get_sandbox(&conn, sandbox_id)
let sandbox = crate::db::read::sandbox(&conn, sandbox_id)
.map_err(|error| Status::internal(format!("db error: {error}")))?
.ok_or_else(|| Status::not_found("activation sandbox disappeared"))?;
let episode = crate::db::cleanup_episode(&conn, sandbox_id)
let episode = crate::db::read::sandbox_view(&conn, sandbox_id)
.cleanup_episode()
.strict()
.map_err(|error| Status::internal(format!("db error: {error}")))?;
// Start and restore pre-arm rollback before guest setup. Its presence
// is valid until cleanup is requested or another activation owns it.
Expand Down Expand Up @@ -1116,11 +1118,13 @@
.lock()
.map_err(|error| Status::internal(format!("database mutex poisoned: {error}")))
.and_then(|db| {
let row = crate::db::get_sandbox(&db, sandbox_id)
let row = crate::db::read::sandbox(&db, sandbox_id)
.map_err(|error| Status::internal(format!("db error: {error}")))?;
let episode = cleanup_id
.map(|cleanup_id| {
crate::db::cleanup_episode(&db, sandbox_id)
crate::db::read::sandbox_view(&db, sandbox_id)
.cleanup_episode()
.strict()
.map(|episode| {
episode.is_some_and(|episode| {
episode.id == cleanup_id && episode.requested_at.is_none()
Expand Down Expand Up @@ -1295,7 +1299,7 @@
);
return Err(activation_completion_lost_status(SandboxState::Starting));
}
if let Ok(Some(sandbox)) = crate::db::get_sandbox(&db, sandbox_id) {
if let Ok(Some(sandbox)) = crate::db::read::sandbox(&db, sandbox_id) {
emit_guest_observed_specs(
sandbox_id,
&sandbox.template,
Expand Down Expand Up @@ -1481,7 +1485,7 @@
let db = db
.lock()
.map_err(|error| Status::internal(format!("database mutex poisoned: {error}")))?;
crate::db::get_sandbox(&db, sandbox_id)
crate::db::read::sandbox(&db, sandbox_id)
.map_err(|error| Status::internal(format!("read restoring sandbox: {error}")))?
.ok_or_else(|| activation_completion_lost_status(SandboxState::Restoring))?
.pool_template
Expand Down
35 changes: 24 additions & 11 deletions crates/libs/sbx/mgr/server/src/service/auth_lookup.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use super::activation_support::{

Check warning on line 1 in crates/libs/sbx/mgr/server/src/service/auth_lookup.rs

View workflow job for this annotation

GitHub Actions / rust_static / Rust clippy

[archcheck] reported by reviewdog 🐶 crates/libs/sbx/mgr/server/src/service/auth_lookup.rs grew from 1221 to 1233 code lines. Once this rule is required, split the file before adding to it. Raw Output: message:"crates/libs/sbx/mgr/server/src/service/auth_lookup.rs grew from 1221 to 1233 code lines. Once this rule is required, split the file before adding to it." location:{path:"crates/libs/sbx/mgr/server/src/service/auth_lookup.rs" range:{start:{line:1}}} severity:WARNING source:{name:"archcheck"} code:{value:"file_size.listed_no_growth"}
GuestActivityKind, IdentityFreeDirectActivationAuthority, configure_process_env_for_activation,
spawn_resume_clock_sync, wait_ready_guest_uptime_ms, wait_ready_via_spaced_host,
};
Expand Down Expand Up @@ -548,16 +548,17 @@
/// Returns error if not found or ambiguous.
pub(super) fn lookup_sandbox(&self, sandbox_id: &str) -> Result<crate::db::SandboxRow, Status> {
let db = self.db_conn()?;
if let Some(sandbox) = crate::db::get_sandbox(&db, sandbox_id)
if let Some(sandbox) = crate::db::read::sandbox(&db, sandbox_id)
.map_err(|e| Status::internal(format!("db error: {e}")))?
{
return Ok(sandbox);
}

let matches = crate::db::find_by_prefix(&db, sandbox_id)
.map_err(|e| Status::internal(format!("db error: {e}")))?;
let matches =
crate::db::scan::resolve_selector(&db, crate::db::scan::RpcSelector, sandbox_id)
.map_err(|e| Status::internal(format!("db error: {e}")))?;
match matches.as_slice() {
[id] => crate::db::get_sandbox(&db, id)
[id] => crate::db::read::sandbox(&db, id)
.map_err(|error| Status::internal(format!("db error: {error}")))?
.ok_or_else(|| Status::not_found(format!("sandbox not found: {id}"))),
[] => Err(Status::not_found(format!(
Expand Down Expand Up @@ -764,7 +765,7 @@
// transition is deferred until after the gate succeeds.
let sandbox = {
let db = self.db_conn()?;
let sandbox = crate::db::get_sandbox(&db, sandbox_id)
let sandbox = crate::db::read::sandbox(&db, sandbox_id)
.map_err(|e| Status::internal(format!("db error: {e}")))?
.ok_or_else(|| Status::not_found(format!("sandbox not found: {sandbox_id}")))?;
sandbox
Expand Down Expand Up @@ -1031,9 +1032,15 @@
uid: u32,
) -> Result<crate::db::SandboxRow, Status> {
let db = self.db_conn()?;
let mut matches = crate::db::list_by_spaced_host_pid_uid(&db, pid, uid)
.map_err(|e| Status::internal(format!("db error: {e}")))?
.into_iter();
let mut matches = crate::db::scan::by_peer(
&db,
crate::db::scan::Peer::SpacedHost {
spaced_host_pid: pid,
uid,
},
)
.map_err(|e| Status::internal(format!("db error: {e}")))?
.into_iter();

match (matches.next(), matches.next()) {
(Some(sandbox), None) => Ok(sandbox),
Expand All @@ -1059,9 +1066,15 @@
// The exact PID/owner lookup below remains mandatory in either case.
.unwrap_or(uid);
let db = self.db_conn()?;
let mut matches = crate::db::list_by_netgtw_pid_uid(&db, pid, uid)
.map_err(|e| Status::internal(format!("db error: {e}")))?
.into_iter();
let mut matches = crate::db::scan::by_peer(
&db,
crate::db::scan::Peer::NetGateway {
netgtw_pid: pid,
uid,
},
)
.map_err(|e| Status::internal(format!("db error: {e}")))?
.into_iter();

match (matches.next(), matches.next()) {
(Some(sandbox), None) => Ok(sandbox),
Expand Down
18 changes: 13 additions & 5 deletions crates/libs/sbx/mgr/server/src/service/cache_cleanup.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use pplx_space_common_core::types::LocalArtifactId;

Check warning on line 1 in crates/libs/sbx/mgr/server/src/service/cache_cleanup.rs

View workflow job for this annotation

GitHub Actions / rust_static / Rust clippy

[archcheck] reported by reviewdog 🐶 crates/libs/sbx/mgr/server/src/service/cache_cleanup.rs grew from 1452 to 1460 code lines. Once this rule is required, split the file before adding to it. Raw Output: message:"crates/libs/sbx/mgr/server/src/service/cache_cleanup.rs grew from 1452 to 1460 code lines. Once this rule is required, split the file before adding to it." location:{path:"crates/libs/sbx/mgr/server/src/service/cache_cleanup.rs" range:{start:{line:1}}} severity:WARNING source:{name:"archcheck"} code:{value:"file_size.listed_no_growth"}

#[cfg(test)]
use super::cleanup_support::CleanupLockGuard;
Expand Down Expand Up @@ -832,7 +832,9 @@
Retirement: Fn(&str) -> Option<RetirementGuard>,
RetirementGuard: FnOnce(),
{
let rows = db.with_connection(crate::db::list_sandboxes_for_recovery)??;
let rows = db.with_connection(|conn| {
crate::db::scan::all_sandboxes(conn, crate::db::scan::SkipUndecodableRows)
})??;
reconcile_inventory_batch(
db,
paths,
Expand Down Expand Up @@ -1123,7 +1125,9 @@
// Periodic DB and directory cursors advance independently: absence from
// this DB page is not absence from SQLite. Avoid receiver RPCs for rows
// owned by another page; recheck again in the insertion transaction.
if db.with_connection(|conn| crate::db::sandbox_exists(conn, &sandbox_id.to_string()))?? {
if db.with_connection(|conn| {
crate::db::read::sandbox_view(conn, &sandbox_id.to_string()).exists()
})?? {
continue;
}
let observed_presence = SandboxInventoryPresence {
Expand Down Expand Up @@ -1231,10 +1235,14 @@
return Ok(OrphanRowMaterialization::InventoryChanged);
}
let tx = crate::db::Transaction::immediate(conn)?;
if crate::db::get_sandbox(&tx, &sandbox_id_string)?.is_some() {
if crate::db::read::sandbox(&tx, &sandbox_id_string)?.is_some() {
return Ok(OrphanRowMaterialization::DatabaseRowAppeared);
}
if crate::db::restore_transfer_scope(&tx, &sandbox_id_string)?.is_some() {
if crate::db::read::sandbox_view(&tx, &sandbox_id_string)
.restore_scope()
.scope()?
.is_some()
{
return Ok(OrphanRowMaterialization::RestoreTransferScope);
}
let committed = insert_orphaned_sandbox_row_in_transaction(
Expand Down Expand Up @@ -1473,7 +1481,7 @@
) -> crate::db::Result<()> {
// The caller keeps SQLite locked through retirement, excluding a
// same-ID insertion and its new policy obligation.
if !crate::db::sandbox_exists(db, sandbox_id)? {
if !crate::db::read::sandbox_view(db, sandbox_id).exists()? {
self.network_policy_reconcile_queue.retire(sandbox_id);
}
Ok(())
Expand Down
Loading
Loading