ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Close hostile recovery and concurrency gaps

Browse source
This commit is contained in:
ms2k 2026-07-18 05:47:15 +09:00
commit 3f9b1b3543
6 changed files with 306 additions and 113 deletions
Download patch file Download diff file Expand all files Collapse all files

60
crates/lazaret-host-coordinator/src/client_bootstrap.rs
View file

@ -183,7 +183,7 @@ fn serve_published(
loop {
while let Ok((connection_id, result)) = completions.try_recv() {
active.remove(&connection_id);
connection_completion(result, false)?;
connection_completion(result)?;
}
require_escalation_running(&escalation_completions, escalation_enabled)?;
let face = {
@ -196,7 +196,7 @@ fn serve_published(
}
drop(lifecycle);
while !active.is_empty() {
wait_for_connection_completion(&completions, &mut active, true)?;
wait_for_connection_completion(&completions, &mut active)?;
}
if let Some(escalation) = &escalation_handle {
escalation.stop();
@ -222,7 +222,7 @@ fn serve_published(
face
};
if active.len() == MAX_CONCURRENT_CONNECTIONS {
wait_for_connection_completion(&completions, &mut active, false)?;
wait_for_connection_completion(&completions, &mut active)?;
continue;
}
match face.accept_connection() {
@ -299,31 +299,26 @@ fn require_escalation_running(
fn wait_for_connection_completion(
completions: &std::sync::mpsc::Receiver<(u64, Result<(), ControlError>)>,
active: &mut BTreeMap<u64, std::os::unix::net::UnixStream>,
cancellation_requested: bool,
) -> Result<(), ProductionColdStartError> {
let (connection_id, result) = completions
.recv()
.map_err(ProductionColdStartError::service)?;
active.remove(&connection_id);
connection_completion(result, cancellation_requested)
connection_completion(result)
}
fn connection_completion(
result: Result<(), ControlError>,
cancellation_requested: bool,
) -> Result<(), ProductionColdStartError> {
fn connection_completion(result: Result<(), ControlError>) -> Result<(), ProductionColdStartError> {
match result {
Ok(()) => Ok(()),
Err(ControlError::Io(error))
if cancellation_requested
&& matches!(
error.kind(),
std::io::ErrorKind::UnexpectedEof
| std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::ConnectionAborted
| std::io::ErrorKind::BrokenPipe
| std::io::ErrorKind::NotConnected
) =>
if matches!(
error.kind(),
std::io::ErrorKind::UnexpectedEof
| std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::ConnectionAborted
| std::io::ErrorKind::BrokenPipe
| std::io::ErrorKind::NotConnected
) =>
{
Ok(())
}
@ -583,30 +578,23 @@ mod tests {
}
#[test]
fn withdrawal_suppresses_only_cancelled_socket_io() {
fn client_disconnects_and_withdrawal_are_connection_local() {
assert!(
connection_completion(
Err(ControlError::Io(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"withdrawn socket",
))),
true,
)
connection_completion(Err(ControlError::Io(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"withdrawn socket",
))))
.is_ok()
);
assert!(
connection_completion(
Err(ControlError::Io(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"ordinary socket failure",
))),
false,
)
.is_err()
connection_completion(Err(ControlError::Io(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"ordinary socket failure",
))))
.is_ok()
);
assert!(
connection_completion(Err(ControlError::Lifecycle("teardown failed".into())), true,)
.is_err()
connection_completion(Err(ControlError::Lifecycle("teardown failed".into()))).is_err()
);
}
}

10
crates/lazaret-host-coordinator/src/control.rs
View file

@ -2,10 +2,10 @@ use std::collections::BTreeMap;
use std::ffi::{OsStr, OsString};
use std::fmt;
use std::fs::{File, OpenOptions};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::io::{self, Read, Write};
use std::os::fd::{AsFd, AsRawFd};
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt};
use std::os::unix::fs::{FileExt, FileTypeExt, MetadataExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};
@ -1743,10 +1743,8 @@ impl PublishedControlFace {
"publication objects changed identity, ownership, or mode",
));
}
let mut exact_identity = self.identity_file.try_clone()?;
exact_identity.seek(SeekFrom::Start(0))?;
let mut bytes = Vec::new();
exact_identity.read_to_end(&mut bytes)?;
let mut bytes = vec![0; self.publication_bytes.len()];
self.identity_file.read_exact_at(&mut bytes, 0)?;
if bytes != self.publication_bytes {
return Err(ControlError::InvalidPublication(
"publication identity bytes changed",

49
crates/lazaret-host-coordinator/src/recovery.rs
View file

@ -630,7 +630,9 @@ where
{
recover_state_reset(configuration, store).map_err(|error| error.to_string())?;
failpoint(ProductionRecoveryBoundary::Snapshot, GenerationId::new(0))?;
let snapshot = store.snapshot().map_err(|error| error.to_string())?;
let snapshot = store
.snapshot()
.map_err(|error| format!("Domain Store snapshot failed: {error}"))?;
if snapshot.domain.domain_id != configuration.domain_id() {
return Err("recovery Domain differs from resolved configuration".into());
}
@ -651,7 +653,13 @@ where
&mut failpoint,
&mut open,
&mut delete,
)?;
)
.map_err(|error| {
format!(
"generation {} recovery failed: {error}",
generation.generation_id.get()
)
})?;
}
failpoint(
ProductionRecoveryBoundary::PrepareOperations,
@ -725,26 +733,34 @@ where
if !session_updates.is_empty() {
store
.apply(&session_updates)
.map_err(|error| error.to_string())?;
.map_err(|error| format!("Session failure update failed: {error}"))?;
}
let participants = store
.snapshot()
.map_err(|error| error.to_string())?
.map_err(|error| format!("Participant snapshot failed: {error}"))?
.participants;
let ownership = runtime_recovery_ownership(&participants, generation)?;
let mut owners = seal_runtime_owners(&ownership, open).map_err(|error| error.to_string())?;
let ownership = runtime_recovery_ownership(&participants, generation)
.map_err(|error| format!("runtime ownership inventory failed: {error}"))?;
let mut owners = seal_runtime_owners(&ownership, open)
.map_err(|error| format!("runtime owner acquisition failed: {error}"))?;
failpoint(ProductionRecoveryBoundary::Publication, generation_id)?;
recover_publication(configuration, store, generation_id)?;
recover_publication(configuration, store, generation_id)
.map_err(|error| format!("Publication recovery failed: {error}"))?;
failpoint(ProductionRecoveryBoundary::Sandbox, generation_id)?;
reconcile_runtime_before_state_anchor(&mut owners).map_err(|error| error.to_string())?;
require_sandbox_absent(configuration, store, generation_id, delete)?;
clean_role_after_absence(store, generation_id, ParticipantRole::Sandbox)?;
reconcile_runtime_before_state_anchor(&mut owners)
.map_err(|error| format!("runtime-owner reconciliation failed: {error}"))?;
require_sandbox_absent(configuration, store, generation_id, delete)
.map_err(|error| format!("Sandbox absence failed: {error}"))?;
clean_role_after_absence(store, generation_id, ParticipantRole::Sandbox)
.map_err(|error| format!("Sandbox cleanup failed: {error}"))?;
failpoint(ProductionRecoveryBoundary::HelperAndNetwork, generation_id)?;
clean_role_after_absence(store, generation_id, ParticipantRole::Helper)?;
clean_role_after_absence(store, generation_id, ParticipantRole::Network)?;
clean_role_after_absence(store, generation_id, ParticipantRole::Helper)
.map_err(|error| format!("Helper cleanup failed: {error}"))?;
clean_role_after_absence(store, generation_id, ParticipantRole::Network)
.map_err(|error| format!("Network cleanup failed: {error}"))?;
failpoint(ProductionRecoveryBoundary::GitRouting, generation_id)?;
require_mount_absent(
@ -753,13 +769,16 @@ where
"Git-routing",
)
.map_err(|error| error.to_string())?;
clean_role_after_absence(store, generation_id, ParticipantRole::GitRouting)?;
clean_role_after_absence(store, generation_id, ParticipantRole::GitRouting)
.map_err(|error| format!("Git-routing cleanup failed: {error}"))?;
failpoint(ProductionRecoveryBoundary::StateAnchor, generation_id)?;
recover_state_anchor(configuration, store, catalog, generation, &mut owners)?;
recover_state_anchor(configuration, store, catalog, generation, &mut owners)
.map_err(|error| format!("State Anchor recovery failed: {error}"))?;
failpoint(ProductionRecoveryBoundary::PreparedLower, generation_id)?;
recover_partial_prepared_lower(configuration, store, catalog, generation_id)?;
recover_partial_prepared_lower(configuration, store, catalog, generation_id)
.map_err(|error| format!("prepared-lower recovery failed: {error}"))?;
let snapshot = store.snapshot().map_err(|error| error.to_string())?;
if snapshot.participants.iter().any(|participant| {

31
crates/lazaret-host-coordinator/src/runtime.rs
View file

@ -946,6 +946,17 @@ pub struct CommittedSandboxRuntime {
}
impl CommittedSandboxRuntime {
fn receive_runtime_event(&mut self) -> Result<SupervisorEvent, SandboxCoordinatorError> {
loop {
let event = self.control.receive()?;
if event == SupervisorEvent::HelperExited {
self.network_evidence = None;
} else {
return Ok(event);
}
}
}
#[must_use]
pub const fn network_evidence(&self) -> Option<&NetworkReadinessEvidence> {
self.network_evidence.as_ref()
@ -984,11 +995,11 @@ impl CommittedSandboxRuntime {
/// # Errors
/// Returns an error if the RS cannot apply the exact retained-runsc pidfd action.
pub fn signal_attached_runsc(
&self,
&mut self,
signal: AttachedRunscSignal,
) -> Result<(), SandboxCoordinatorError> {
self.control.signal_attached_runsc(signal)?;
if self.control.receive()? == SupervisorEvent::SignalApplied(signal) {
if self.receive_runtime_event()? == SupervisorEvent::SignalApplied(signal) {
Ok(())
} else {
Err(SandboxCoordinatorError::InvalidState(
@ -1007,7 +1018,7 @@ impl CommittedSandboxRuntime {
) -> Result<(), SandboxCoordinatorError> {
coordinator.begin_release(self.generation_id)?;
self.control.stop_sandbox()?;
if self.control.receive()? != SupervisorEvent::SandboxStopped {
if self.receive_runtime_event()? != SupervisorEvent::SandboxStopped {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not stop the sandbox",
));
@ -1027,7 +1038,7 @@ impl CommittedSandboxRuntime {
.begin_release(self.generation_id)
.map_err(|error| SandboxCoordinatorError::Io(io::Error::other(error)))?;
self.control.stop_git_routing()?;
match self.control.receive()? {
match self.receive_runtime_event()? {
SupervisorEvent::GitRoutingStopped { mount } if mount == expected => coordinator
.complete_clean(self.generation_id, mount)
.map_err(|error| SandboxCoordinatorError::Io(io::Error::other(error))),
@ -1046,7 +1057,7 @@ impl CommittedSandboxRuntime {
/// Refuses an unexpected RS response or channel failure.
pub fn stop_sidecar(&mut self) -> Result<(), SandboxCoordinatorError> {
self.control.stop_sidecar()?;
if self.control.receive()? == SupervisorEvent::SidecarStopped {
if self.receive_runtime_event()? == SupervisorEvent::SidecarStopped {
self.network_evidence = None;
Ok(())
} else {
@ -1065,7 +1076,7 @@ impl CommittedSandboxRuntime {
expected_mount: MountIdentity,
) -> Result<(), SandboxCoordinatorError> {
self.control.unmount_state_anchor()?;
match self.control.receive()? {
match self.receive_runtime_event()? {
SupervisorEvent::StateAnchorUnmounted { mount } if mount == expected_mount => Ok(()),
_ => Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not confirm the exact State Anchor unmount",
@ -1085,7 +1096,7 @@ impl CommittedSandboxRuntime {
) -> Result<StateAnchorAbsenceObservation, SandboxCoordinatorError> {
self.control
.observe_state_anchor_absence(user_namespace, mount_namespace, mount)?;
match self.control.receive()? {
match self.receive_runtime_event()? {
SupervisorEvent::StateAnchorAbsenceObserved {
user_namespace: observed_user_namespace,
mount_namespace: observed_mount_namespace,
@ -1118,7 +1129,7 @@ impl CommittedSandboxRuntime {
/// Refuses an unexpected response or an RS that does not exit.
pub fn graceful_exit(mut self) -> Result<(), SandboxCoordinatorError> {
self.control.exit()?;
if self.control.receive()? != SupervisorEvent::Exited {
if self.receive_runtime_event()? != SupervisorEvent::Exited {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not acknowledge graceful exit",
));
@ -1132,7 +1143,7 @@ impl CommittedSandboxRuntime {
/// # Errors
/// Refuses any other event or an RS that does not then exit.
pub fn wait_for_attached_runsc_exit(mut self) -> Result<(), SandboxCoordinatorError> {
if self.control.receive()? != SupervisorEvent::AttachedRunscExited {
if self.receive_runtime_event()? != SupervisorEvent::AttachedRunscExited {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not report attached-runsc loss",
));
@ -1160,7 +1171,7 @@ impl CommittedSandboxRuntime {
) -> Result<(), SandboxCoordinatorError> {
coordinator.begin_release(self.generation_id)?;
self.control.stop()?;
if self.control.receive()? != SupervisorEvent::Stopped {
if self.receive_runtime_event()? != SupervisorEvent::Stopped {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not complete sandbox stop",
));

16
crates/lazaret-host-coordinator/tests/control.rs
View file

@ -516,6 +516,22 @@ fn duplicate_session_cli_and_anchor_identities_are_refused_while_registered() {
first_server.join().unwrap().unwrap();
}
#[test]
fn concurrent_connections_validate_publication_without_shared_file_offsets() {
let directory = TestDirectory::new("concurrent-publication");
let identity = generation_identity(7);
let (face, _store) = publish(&directory, &identity);
let clients = (40..56)
.map(|value| {
let face = Arc::clone(&face);
thread::spawn(move || enroll(&face, ClientId::new(bytes(value))))
})
.collect::<Vec<_>>();
for client in clients {
client.join().unwrap();
}
}
#[test]
fn anonymous_connections_are_enrollment_only_bounded_and_exact_alpn_is_required() {
let directory = TestDirectory::new("anonymous");

4
crates/lazaret-host-coordinator/tests/runtime-composition.rs
View file

@ -753,7 +753,7 @@ fn cross_bundle_handoff_refuses_target_work() -> Result<(), MatrixError> {
fn normal_stop_and_stop_resume() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let runtime = fixture.launch_committed()?;
let mut runtime = fixture.launch_committed()?;
wait_for_activity(&fixture.activity)?;
runtime
.signal_attached_runsc(AttachedRunscSignal::Stop)
@ -789,7 +789,7 @@ fn normal_stop_and_stop_resume() -> Result<(), MatrixError> {
fn attached_runsc_loss_stops_guest() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let runtime = fixture.launch_committed()?;
let mut runtime = fixture.launch_committed()?;
wait_for_activity(&fixture.activity)?;
runtime
.signal_attached_runsc(AttachedRunscSignal::Kill)