Skip to content

fix(server): retry provider attachment CAS conflicts #3501

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
elezar merged 2 commits into from
Sep 28, 2026
Merged

fix(server): retry provider attachment CAS conflicts #3501

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ea707d2
fix(server): retry transient sandbox CAS conflicts
drew Sep 20, 2026
3ae176a
fix(server): align sandbox CAS retry with current main
drew Sep 28, 2026
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
5 changes: 3 additions & 2 deletions architecture/gateway.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -858,8 +858,9 @@ with it.
fetches the current object, applies a mutation closure, and writes with a
`MatchResourceVersion` condition. On conflict the persistence layer returns a
`Conflict` error, which gRPC handlers map to `ABORTED` status so the client
(or the next watch/reconcile event) can retry with fresh state. There is no
automatic retry loop.
(or the next watch/reconcile event) can retry with fresh state. Provider
attach/detach handlers retry bounded server-driven conflicts around this helper;
an explicit client version still fails on its first conflict.

The helper accepts an `expected_version` parameter that selects between two
modes:
Expand Down
199 changes: 139 additions & 60 deletions crates/openshell-server/src/grpc/sandbox.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ const TCP_FORWARD_CHUNK_SIZE: usize = 64 * 1024;
const NO_LOGIN_SHELL_ENV: (&str, &str) = ("OPENSHELL_NO_LOGIN_SHELL", "1");
const MAX_TEMPLATES_PER_WORKSPACE: u32 = 1000;
const MAX_CREATE_SERVICE_EXPOSURES: usize = 32;
const SANDBOX_CAS_RETRY_LIMIT: usize = 16;

#[cfg(test)]
#[path = "interactive_exec_tests.rs"]
Expand Down Expand Up @@ -1381,38 +1382,37 @@ pub(super) async fn handle_attach_sandbox_provider(
let attached_clone = attached.clone();
let mutation_id = uuid::Uuid::new_v4().to_string();

let sandbox = state
.store
.update_message_cas::<Sandbox, _>(
&sandbox_id,
request.expected_resource_version,
|sandbox| {
attached_clone.store(false, Ordering::Relaxed);
let Some(ref mut spec) = sandbox.spec else {
// Spec should always exist post-creation; if missing, fail CAS to surface error
return;
};
let sandbox = update_sandbox_cas(
state,
&sandbox_id,
request.expected_resource_version,
|sandbox| {
attached_clone.store(false, Ordering::Relaxed);
let Some(ref mut spec) = sandbox.spec else {
// Spec should always exist post-creation; if missing, fail CAS to surface error
return;
};

if spec.provider_attachment_epoch.is_empty() {
spec.provider_attachment_epoch.clone_from(&mutation_id);
}
if spec.provider_attachment_epoch.is_empty() {
spec.provider_attachment_epoch.clone_from(&mutation_id);
}

dedupe_provider_names(&mut spec.providers);
if !spec.providers.iter().any(|name| name == &provider_name)
&& spec.providers.len() < MAX_PROVIDERS
{
spec.providers.push(provider_name.clone());
spec.provider_attachment_epoch.clone_from(&mutation_id);
attached_clone.store(true, Ordering::Relaxed);
crate::compute::provisioning_deadline::attachments_changed(
sandbox,
current_time_ms(),
);
}
},
)
.await
.map_err(|e| super::persistence_error_to_status(e, "attach sandbox provider"))?;
dedupe_provider_names(&mut spec.providers);
if !spec.providers.iter().any(|name| name == &provider_name)
&& spec.providers.len() < MAX_PROVIDERS
{
spec.providers.push(provider_name.clone());
spec.provider_attachment_epoch.clone_from(&mutation_id);
attached_clone.store(true, Ordering::Relaxed);
crate::compute::provisioning_deadline::attachments_changed(
sandbox,
current_time_ms(),
);
}
},
)
.await
.map_err(|e| super::persistence_error_to_status(e, "attach sandbox provider"))?;

let attached = attached.load(Ordering::Relaxed);
let receipt = super::provider_readiness::record_provider_mutation(
Expand Down Expand Up @@ -1505,38 +1505,37 @@ pub(super) async fn handle_detach_sandbox_provider(
let detached_clone = detached.clone();
let mutation_id = uuid::Uuid::new_v4().to_string();

let sandbox = state
.store
.update_message_cas::<Sandbox, _>(
&sandbox_id,
request.expected_resource_version,
|sandbox| {
detached_clone.store(false, Ordering::Relaxed);
let Some(ref mut spec) = sandbox.spec else {
// Spec should always exist post-creation; if missing, fail CAS to surface error
return;
};
let sandbox = update_sandbox_cas(
state,
&sandbox_id,
request.expected_resource_version,
|sandbox| {
detached_clone.store(false, Ordering::Relaxed);
let Some(ref mut spec) = sandbox.spec else {
// Spec should always exist post-creation; if missing, fail CAS to surface error
return;
};

if spec.provider_attachment_epoch.is_empty() {
spec.provider_attachment_epoch.clone_from(&mutation_id);
}
if spec.provider_attachment_epoch.is_empty() {
spec.provider_attachment_epoch.clone_from(&mutation_id);
}

let before_len = spec.providers.len();
spec.providers.retain(|name| name != &provider_name);
if spec.providers.len() != before_len {
spec.provider_attachment_epoch.clone_from(&mutation_id);
detached_clone.store(true, Ordering::Relaxed);
// Only dedupe after making a change
dedupe_provider_names(&mut spec.providers);
crate::compute::provisioning_deadline::attachments_changed(
sandbox,
current_time_ms(),
);
}
},
)
.await
.map_err(|e| super::persistence_error_to_status(e, "detach sandbox provider"))?;
let before_len = spec.providers.len();
spec.providers.retain(|name| name != &provider_name);
if spec.providers.len() != before_len {
spec.provider_attachment_epoch.clone_from(&mutation_id);
detached_clone.store(true, Ordering::Relaxed);
// Only dedupe after making a change
dedupe_provider_names(&mut spec.providers);
crate::compute::provisioning_deadline::attachments_changed(
sandbox,
current_time_ms(),
);
}
},
)
.await
.map_err(|e| super::persistence_error_to_status(e, "detach sandbox provider"))?;

let detached = detached.load(Ordering::Relaxed);
let receipt = super::provider_readiness::record_provider_mutation(
Expand Down Expand Up @@ -1764,6 +1763,33 @@ async fn providers_for_sandbox(
Ok(providers)
}

/// Retry a pure server-owned mutation on a fresh resource version. Explicit
/// client versions still fail on conflict so clients retain their CAS contract.
async fn update_sandbox_cas<F>(
state: &ServerState,
sandbox_id: &str,
expected_resource_version: u64,
mut mutate: F,
) -> crate::persistence::PersistenceResult<Sandbox>
where
F: FnMut(&mut Sandbox),
{
for attempt in 1..=SANDBOX_CAS_RETRY_LIMIT {
match state
.store
.update_message_cas::<Sandbox, _>(sandbox_id, expected_resource_version, &mut mutate)
.await
{
Ok(sandbox) => return Ok(sandbox),
Err(crate::persistence::PersistenceError::Conflict { .. })
if expected_resource_version == 0 && attempt < SANDBOX_CAS_RETRY_LIMIT => {}
Err(error) => return Err(error),
}
}

unreachable!("sandbox CAS retry loop always returns on its final attempt")
}

fn dedupe_provider_names(provider_names: &mut Vec<String>) {
let mut index = 0;
while index < provider_names.len() {
Expand Down Expand Up @@ -8422,6 +8448,59 @@ mod tests {

// ---- CAS (Client-driven optimistic concurrency) tests ----

#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
async fn server_managed_sandbox_cas_retries_concurrent_updates() {
use std::sync::Barrier;

const WRITERS: usize = 5;

let state = Arc::new(test_server_state().await);
let sandbox = test_sandbox("server-cas-retry", Vec::new());
let sandbox_id = sandbox.object_id().to_string();
state.store.put_message(&sandbox).await.unwrap();

let barrier = Arc::new(Barrier::new(WRITERS));
let mut handles = Vec::with_capacity(WRITERS);
for _ in 0..WRITERS {
let state = Arc::clone(&state);
let barrier = Arc::clone(&barrier);
let sandbox_id = sandbox_id.clone();
handles.push(tokio::spawn(async move {
let mut first_attempt = true;
update_sandbox_cas(&state, &sandbox_id, 0, |sandbox| {
if first_attempt {
first_attempt = false;
barrier.wait();
}
let annotations = &mut sandbox.metadata.as_mut().unwrap().annotations;
let count = annotations
.get("internal-update-count")
.and_then(|value| value.parse::<usize>().ok())
.unwrap_or_default();
annotations
.insert("internal-update-count".to_string(), (count + 1).to_string());
})
.await
}));
}

for result in future::join_all(handles).await {
result.unwrap().unwrap();
}

let updated = state
.store
.get_message::<Sandbox>(&sandbox_id)
.await
.unwrap()
.unwrap();
assert_eq!(
updated.metadata.as_ref().unwrap().annotations["internal-update-count"],
WRITERS.to_string()
);
assert_eq!(updated.get_resource_version(), WRITERS as u64 + 1);
}

#[tokio::test]
async fn attach_sandbox_provider_client_driven_cas_succeeds_with_correct_version() {
let state = test_server_state().await;
Expand Down
Loading