Skip to content

Commit 685e700

Browse files
committed
fix(cli): retry provider attachment on a stale resource version
Attach and detach send the resource version the command itself just read, so a concurrent status write, such as a supervisor configuration result, could fail the command with "sandbox was modified by another operation". Retry those conflicts with a fresh read, up to five times. Mutations whose save is uncertain still never retry. Signed-off-by: Piotr Mlocek <pmlocek@nvidia.com>
1 parent 0c4ba41 commit 685e700

2 files changed

Lines changed: 208 additions & 70 deletions

File tree

‎crates/openshell-cli/src/commands/provider.rs‎

Lines changed: 90 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,10 @@ pub async fn sandbox_provider_list(
129129
Ok(())
130130
}
131131

132+
/// Attempts for an attach or detach whose resource version went stale between
133+
/// the command's read and its write.
134+
const PROVIDER_ATTACHMENT_CAS_ATTEMPTS: u32 = 5;
135+
132136
/// Save a provider attachment and optionally wait for its installed authority.
133137
pub async fn sandbox_provider_attach(
134138
server: &str,
@@ -141,45 +145,53 @@ pub async fn sandbox_provider_attach(
141145
readiness.validate()?;
142146
let mut client = grpc_client(server, tls).await?;
143147

144-
// Fetch current sandbox to get resource_version for CAS
145-
let sandbox = client
146-
.get_sandbox(GetSandboxRequest {
147-
name: (name).to_string(),
148-
workspace_scope: Some(openshell_core::proto::workspace_selector(
149-
(workspace).to_string(),
150-
)),
151-
})
152-
.await
153-
.map_err(|status| miette!("provider attachment lookup failed ({})", status.code()))?
154-
.into_inner()
155-
.sandbox
156-
.ok_or_else(|| miette::miette!("sandbox not found"))?;
148+
// The resource version only guards this command's own read, so a
149+
// conflict from a concurrent status write is retried with a fresh read.
150+
let mut attempt = 1;
151+
let (sandbox, response) = loop {
152+
let sandbox = client
153+
.get_sandbox(GetSandboxRequest {
154+
name: (name).to_string(),
155+
workspace_scope: Some(openshell_core::proto::workspace_selector(
156+
(workspace).to_string(),
157+
)),
158+
})
159+
.await
160+
.map_err(|status| miette!("provider attachment lookup failed ({})", status.code()))?
161+
.into_inner()
162+
.sandbox
163+
.ok_or_else(|| miette::miette!("sandbox not found"))?;
157164

158-
let resource_version = sandbox.metadata.as_ref().map_or(0, |m| m.resource_version);
165+
let resource_version = sandbox.metadata.as_ref().map_or(0, |m| m.resource_version);
159166

160-
let response = match client
161-
.attach_sandbox_provider(AttachSandboxProviderRequest {
162-
request_id: String::new(),
163-
sandbox: (name).to_string(),
164-
workspace_scope: Some(openshell_core::proto::workspace_selector(
165-
(workspace).to_string(),
166-
)),
167-
provider: provider.to_string(),
168-
expected_resource_version: resource_version,
169-
})
170-
.await
171-
{
172-
Ok(response) => response.into_inner(),
173-
// Explicit post-save uncertainty takes precedence over a generic retry hint.
174-
Err(status)
175-
if status.code() == Code::Aborted && !provider_mutation_is_uncertain(&status) =>
167+
match client
168+
.attach_sandbox_provider(AttachSandboxProviderRequest {
169+
request_id: String::new(),
170+
sandbox: (name).to_string(),
171+
workspace_scope: Some(openshell_core::proto::workspace_selector(
172+
(workspace).to_string(),
173+
)),
174+
provider: provider.to_string(),
175+
expected_resource_version: resource_version,
176+
})
177+
.await
176178
{
177-
return Err(miette::miette!(
178-
"Failed to attach provider: sandbox was modified by another operation.\n\
179-
Please retry the command."
180-
));
179+
Ok(response) => break (sandbox, response.into_inner()),
180+
// Explicit post-save uncertainty takes precedence over a generic retry hint.
181+
Err(status)
182+
if status.code() == Code::Aborted && !provider_mutation_is_uncertain(&status) =>
183+
{
184+
if attempt < PROVIDER_ATTACHMENT_CAS_ATTEMPTS {
185+
attempt += 1;
186+
continue;
187+
}
188+
return Err(miette::miette!(
189+
"Failed to attach provider: sandbox was modified by another operation.\n\
190+
Please retry the command."
191+
));
192+
}
193+
Err(error) => return Err(provider_mutation_error(&error, "attachment")),
181194
}
182-
Err(error) => return Err(provider_mutation_error(&error, "attachment")),
183195
};
184196

185197
let receipt = response.receipt.ok_or_else(|| {
@@ -216,45 +228,53 @@ pub async fn sandbox_provider_detach(
216228
readiness.validate()?;
217229
let mut client = grpc_client(server, tls).await?;
218230

219-
// Fetch current sandbox to get resource_version for CAS
220-
let sandbox = client
221-
.get_sandbox(GetSandboxRequest {
222-
name: (name).to_string(),
223-
workspace_scope: Some(openshell_core::proto::workspace_selector(
224-
(workspace).to_string(),
225-
)),
226-
})
227-
.await
228-
.map_err(|status| miette!("provider detachment lookup failed ({})", status.code()))?
229-
.into_inner()
230-
.sandbox
231-
.ok_or_else(|| miette::miette!("sandbox not found"))?;
231+
// The resource version only guards this command's own read, so a
232+
// conflict from a concurrent status write is retried with a fresh read.
233+
let mut attempt = 1;
234+
let (sandbox, response) = loop {
235+
let sandbox = client
236+
.get_sandbox(GetSandboxRequest {
237+
name: (name).to_string(),
238+
workspace_scope: Some(openshell_core::proto::workspace_selector(
239+
(workspace).to_string(),
240+
)),
241+
})
242+
.await
243+
.map_err(|status| miette!("provider detachment lookup failed ({})", status.code()))?
244+
.into_inner()
245+
.sandbox
246+
.ok_or_else(|| miette::miette!("sandbox not found"))?;
232247

233-
let resource_version = sandbox.metadata.as_ref().map_or(0, |m| m.resource_version);
248+
let resource_version = sandbox.metadata.as_ref().map_or(0, |m| m.resource_version);
234249

235-
let response = match client
236-
.detach_sandbox_provider(DetachSandboxProviderRequest {
237-
request_id: String::new(),
238-
sandbox: (name).to_string(),
239-
workspace_scope: Some(openshell_core::proto::workspace_selector(
240-
(workspace).to_string(),
241-
)),
242-
provider: provider.to_string(),
243-
expected_resource_version: resource_version,
244-
})
245-
.await
246-
{
247-
Ok(response) => response.into_inner(),
248-
// Explicit post-save uncertainty takes precedence over a generic retry hint.
249-
Err(status)
250-
if status.code() == Code::Aborted && !provider_mutation_is_uncertain(&status) =>
250+
match client
251+
.detach_sandbox_provider(DetachSandboxProviderRequest {
252+
request_id: String::new(),
253+
sandbox: (name).to_string(),
254+
workspace_scope: Some(openshell_core::proto::workspace_selector(
255+
(workspace).to_string(),
256+
)),
257+
provider: provider.to_string(),
258+
expected_resource_version: resource_version,
259+
})
260+
.await
251261
{
252-
return Err(miette::miette!(
253-
"Failed to detach provider: sandbox was modified by another operation.\n\
254-
Please retry the command."
255-
));
262+
Ok(response) => break (sandbox, response.into_inner()),
263+
// Explicit post-save uncertainty takes precedence over a generic retry hint.
264+
Err(status)
265+
if status.code() == Code::Aborted && !provider_mutation_is_uncertain(&status) =>
266+
{
267+
if attempt < PROVIDER_ATTACHMENT_CAS_ATTEMPTS {
268+
attempt += 1;
269+
continue;
270+
}
271+
return Err(miette::miette!(
272+
"Failed to detach provider: sandbox was modified by another operation.\n\
273+
Please retry the command."
274+
));
275+
}
276+
Err(error) => return Err(provider_mutation_error(&error, "detachment")),
256277
}
257-
Err(error) => return Err(provider_mutation_error(&error, "detachment")),
258278
};
259279

260280
let receipt = response.receipt.ok_or_else(|| miette!("gateway did not return a provider receipt; saved detachment cannot establish revocation"))?;

‎crates/openshell-cli/tests/provider_commands_integration.rs‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,8 @@ struct ProviderState {
110110
readiness_sequence: Arc<AtomicU64>,
111111
corrupt_mutation_receipt: Arc<Mutex<Option<ReceiptCorruption>>>,
112112
fail_mutation_after_save: Arc<Mutex<Option<Status>>>,
113+
/// Plain resource-version conflicts to return before an attachment saves.
114+
sandbox_provider_conflicts: Arc<AtomicU64>,
113115
global_settings: Arc<Mutex<HashMap<String, SettingValue>>>,
114116
}
115117

@@ -428,6 +430,18 @@ impl OpenShell for TestOpenShell {
428430
sandbox_name: sandbox_name.clone(),
429431
provider: request.provider.clone(),
430432
});
433+
if self
434+
.state
435+
.sandbox_provider_conflicts
436+
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
437+
left.checked_sub(1)
438+
})
439+
.is_ok()
440+
{
441+
return Err(Status::aborted(
442+
"sandbox was modified concurrently (current resource_version: 2)",
443+
));
444+
}
431445
if !self
432446
.state
433447
.providers
@@ -490,6 +504,18 @@ impl OpenShell for TestOpenShell {
490504
sandbox_name: sandbox_name.clone(),
491505
provider: request.provider.clone(),
492506
});
507+
if self
508+
.state
509+
.sandbox_provider_conflicts
510+
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
511+
left.checked_sub(1)
512+
})
513+
.is_ok()
514+
{
515+
return Err(Status::aborted(
516+
"sandbox was modified concurrently (current resource_version: 2)",
517+
));
518+
}
493519
let mut sandbox_providers = self.state.sandbox_providers.lock().await;
494520
let providers = sandbox_providers.entry(sandbox_name.clone()).or_default();
495521
let before_len = providers.len();
@@ -6035,3 +6061,95 @@ async fn provider_create_from_gcloud_adc_missing_client_secret() {
60356061
"no provider must be created when ADC validation fails"
60366062
);
60376063
}
6064+
6065+
#[tokio::test]
6066+
async fn provider_attachment_retries_resource_version_conflicts() {
6067+
let server = run_server().await;
6068+
seed_readiness_provider(&server).await;
6069+
for action in ["attach", "detach"] {
6070+
let sandbox_name = "conflicted";
6071+
server.state.sandbox_providers.lock().await.insert(
6072+
sandbox_name.to_string(),
6073+
if action == "attach" {
6074+
Vec::new()
6075+
} else {
6076+
vec![READINESS_PROVIDER.to_string()]
6077+
},
6078+
);
6079+
server.state.sandbox_provider_requests.lock().await.clear();
6080+
server
6081+
.state
6082+
.sandbox_provider_conflicts
6083+
.store(2, Ordering::SeqCst);
6084+
let output = run_readiness_cli(
6085+
&server,
6086+
&[
6087+
"sandbox",
6088+
"provider",
6089+
action,
6090+
sandbox_name,
6091+
READINESS_PROVIDER,
6092+
],
6093+
)
6094+
.await;
6095+
assert!(
6096+
output.status.success(),
6097+
"{action}: {}",
6098+
String::from_utf8_lossy(&output.stderr)
6099+
);
6100+
assert_eq!(
6101+
server.state.sandbox_provider_requests.lock().await.len(),
6102+
3,
6103+
"{action} retried each conflict once"
6104+
);
6105+
assert_eq!(
6106+
server
6107+
.state
6108+
.sandbox_providers
6109+
.lock()
6110+
.await
6111+
.get(sandbox_name)
6112+
.unwrap()
6113+
.contains(&READINESS_PROVIDER.to_string()),
6114+
action == "attach"
6115+
);
6116+
}
6117+
}
6118+
6119+
#[tokio::test]
6120+
async fn provider_attachment_reports_persistent_conflicts() {
6121+
let server = run_server().await;
6122+
seed_readiness_provider(&server).await;
6123+
server
6124+
.state
6125+
.sandbox_providers
6126+
.lock()
6127+
.await
6128+
.insert("contended".to_string(), Vec::new());
6129+
server
6130+
.state
6131+
.sandbox_provider_conflicts
6132+
.store(u64::MAX, Ordering::SeqCst);
6133+
let output = run_readiness_cli(
6134+
&server,
6135+
&[
6136+
"sandbox",
6137+
"provider",
6138+
"attach",
6139+
"contended",
6140+
READINESS_PROVIDER,
6141+
],
6142+
)
6143+
.await;
6144+
assert!(!output.status.success());
6145+
assert!(
6146+
String::from_utf8_lossy(&output.stderr).contains("Please retry the command"),
6147+
"{}",
6148+
String::from_utf8_lossy(&output.stderr)
6149+
);
6150+
assert_eq!(server.state.sandbox_provider_requests.lock().await.len(), 5);
6151+
server
6152+
.state
6153+
.sandbox_provider_conflicts
6154+
.store(0, Ordering::SeqCst);
6155+
}

0 commit comments

Comments
 (0)