Fix permanent shard loss when RevokeShards times out - #3766
Fix permanent shard loss when RevokeShards times out#3766Aditya1404Sal wants to merge 3 commits into
Conversation
✅ Deploy Preview for golemcloud canceled.
|
9267d90 to
4849499
Compare
4849499 to
84e0086
Compare
|
I've asked some detailed explanation from the agent to understand the findings, I think they are useful so copying all 3 findings with detailed explanation in the next 3 comments. |
In-flight worker creation is invisible to the drainIntended invariantAfter
The new drain only proves a narrower statement:
Those are not equivalent. 1. A request checks ownership before creating the workerIn let owned_agent_id = self
.canonicalize_owned_agent_id(&OwnedAgentId::new(environment_id, &agent_id))
.await?;
self.ensure_worker_belongs_to_this_executor(&agent_id)?;
let metadata = Worker::<Ctx>::get_latest_metadata(self, &owned_agent_id).await;
// More asynchronous work...
Worker::get_or_create_suspended_with_freshness(
self,
&owned_agent_id,
// ...
)
.awaitThe ownership check is a point-in-time check. There is no guard held between it and the subsequent worker creation. 2. Worker creation is hidden as a pending cache entry
let active_agent = self
.agents
.get_or_insert_simple(&cache_key, || {
Box::pin(async move {
let worker = Worker::new(
&deps,
// ...
)
.await;
worker.map(|worker| Arc::new(ActiveAgent::new(Arc::new(worker))))
})
})
.await?;While But match value {
Item::Cached { value, .. } => {
snapshotted_pairs.push((key.clone(), value.clone()));
}
Item::Pending { .. } => {}
}Therefore, a worker creation can be actively running while being completely invisible to the drain. 3. Revoke treats an empty snapshot as completionThe PR’s drain loop does: let mut lost = Vec::new();
for (agent_id, worker) in self.active_agents().snapshot().await {
if self.shard_service().check_worker(&agent_id).is_err()
&& worker.is_loaded().await
{
lost.push((agent_id, worker));
}
}
if lost.is_empty() {
break;
}If the only affected worker is still a pending cache computation, The fact that this is inside a loop does not help: it breaks on the first empty observation and does not know that pending creations exist. 4. The pending worker can still perform durable writesBefore let oplog = oplog_service
.create(
owned_agent_id,
agent_mode,
initial_oplog_entry,
// ...
)
.await;
this.worker_service()
.update_cached_status(
owned_agent_id,
None,
initial_status_value.clone(),
)
.await;It can also enqueue and commit agent initialization: worker
.enqueue_worker_invocation(
AgentInvocation::AgentInitialization {
// ...
},
)
.await
.expect("Failed enqueuing initial agent invocations to worker");
self.add_and_commit_oplog_internal(&instance_guard, entry, None)
.await;All of this can happen after revoke has returned. Failing timelineWhy the invocation-loop check is too lateThe PR adds: 'outer: loop {
if let Err(err @ WorkerExecutorError::InvalidShardId { .. }) =
self.parent.shard_service().check_worker(&agent_id)
{
self.parent.acknowledge_interruption();
self.stop_unloaded(Some(err)).await;
self.parent.remove_from_active_agents().await;
break;
}
let (instance, store) = match self.create_instance().await {
// ...
};
}This prevents the old executor from creating a Wasmtime instance after ownership is lost, which is good. But it runs only after:
So it prevents guest execution, but not old-owner durable writes. Required shape of the fixA plain second ownership check is still racy: The operation needs an admission permit or epoch fence: Alternatively, storage writes need an ownership epoch/fencing token that makes writes from the previous owner fail. |
Failed assignment reconciliation happens after reassignmentThis is a separate ordering problem in the shard manager. What happens after an assignment failsThe PR correctly recognizes that a failed RPC may have been applied: // The executor may have applied the assignment even though the call failed
// (for example a timeout), so it always gets the authoritative assignment next.
{
let mut updates_guard = updates.lock().await;
for (pod, _) in &rebalance_failures.failed_assignments {
updates_guard.retry_full_assignment(*pod);
}
}However, The next pass assigns first and reconciles laterAt the beginning of the next pass: let (new_pods, removed_pods, retry_full_assignment_pods) =
updates.lock().await.reset();
// ...
for pod in retry_full_assignment_pods {
if current_routing_table.has_pod(pod) {
full_assignment_pods.insert(pod);
}
}But then the code executes the new rebalance before sending those authoritative assignments: let rebalance_failures =
Self::execute_rebalance(worker_executors.clone(), &mut rebalance).await;
Only afterward does the manager send let failed_full_assignments = set_shard_assignments(
worker_executors.clone(),
routing_table_snapshot.number_of_shards,
&full_assignments,
)
.await;Thus the actual pass order is: The safe order is 3 before 2. Concrete reproduced sequenceSuppose: C is removed, so shards 0 and 1 need new owners. First pass: The manager records shard 1 as unassigned because the RPC failed, while B locally owns it. The PR queues B for reconciliation. Next pass, I observed: That creates this interval: During that overlap, B’s background recovery or a routed request can run the agent while A is also starting it. Why the existing failed-revoke handling is betterFor failed revocations, the PR adds the pod directly to the current pass: for (pod, _) in &rebalance_failures.failed_unassignments {
full_assignment_pods.insert(*pod);
}That means the old pod gets Failed assignments should follow the same pattern: add their pods to the current If that authoritative reconciliation also times out completely, the system must make an explicit safety-versus-liveness decision. Without storage epochs, immediately assigning elsewhere can still create two owners. |
timed-out shard RPCs leave duplicate drain tasks runningThe PR deliberately detaches shard draining from the RPC lifetime. That makes draining cancellation-safe, but every retry creates another independent drain task. If an agent never acknowledges interruption, none of those tasks finish. 1. Every RPC spawns a new drain taskBoth async fn drain_lost_agents(
&self,
change: ShardAssignmentChange,
) -> Result<(), WorkerExecutorError> {
let origin = TraceOrigin::capture_current();
let this = self.clone();
let handle = tokio::spawn(async move {
let span = related_span!(
origin,
Level::INFO,
"shard_drain",
trigger = change.trigger()
);
this.drain_lost_agents_to_completion(change)
.instrument(span)
.await
});
handle.await.map_err(|err| {
WorkerExecutorError::runtime(format!(
"Draining the agents of the revoked shards failed: {err}"
))
})?
}The important distinction is:
That is intentional: otherwise a timeout could stop draining halfway through. The problem is that the task is spawned once per RPC attempt, rather than once per logical assignment change. 2. The drain has an unbounded acknowledgement waitFor each lost agent, the drain requests let receivers: Vec<_> =
join_all(lost.iter().map(|(agent_id, worker)| async move {
worker
.set_interrupting(InterruptKind::Restart)
.await
.map(|receiver| (agent_id.clone(), receiver))
}))
.await
.into_iter()
.flatten()
.collect();Then all receivers are awaited without any deadline: let acks = join_all(receivers.into_iter().map(
|(agent_id, mut receiver)| async move {
match receiver.recv().await {
Ok(()) => true,
Err(err) => {
debug!(
agent_id = %agent_id,
error = %err,
"Agent went away before acknowledging its interruption"
);
false
}
}
},
))
.await;If one agent is stuck somewhere where it cannot process the interrupt, There is a shutdown branch around the complete drain: tokio::select! {
(...) = drain => {
// success
}
_ = shutdown.cancelled() => {
Err(WorkerExecutorError::runtime(
"Executor is shutting down"
))
}
}So “immortal” here means until one of these occurs:
There is no ordinary per-drain timeout or cancellation once the RPC has gone away. 3. A retry subscribes again instead of joining the original drainSuppose Drain #1 already moved the worker into A later drain calls ExecutionStatus::Interrupting {
interrupt_kind: current_kind,
await_interruption,
agent_mode,
timestamp,
} => {
let receiver = await_interruption.subscribe();
if matches!(current_kind, InterruptKind::Restart)
&& !matches!(interrupt_kind, InterruptKind::Restart)
{
*execution_status = ExecutionStatus::Interrupting {
interrupt_kind,
await_interruption,
agent_mode,
timestamp,
};
}
Some(receiver)
}The pending interrupt queue also accepts another Self::Pending(current) => {
if matches!(interrupt.kind, InterruptKind::Restart) {
interrupt.reacquire_permits |= current.reacquire_permits;
}
*self = Self::Pending(interrupt);
true
}Consequently, repeated The broadcast channel itself is appropriate for multiple legitimate waiters. What is missing is deduplication at the logical drain operation level. 4. The shard manager retries these calls
with_retriable_errors(
"worker_executor",
"revoke_shards",
Some(format!("{pod}")),
&self.config.retries,
&(pod, shard_ids),
|(pod, shard_ids)| {
Box::pin(self.revoke_shards_internal(pod, shard_ids))
},
)
.awaitEach individual attempt has a 60-second timeout: let revoke_shards_response = timeout(
self.config.revoke_shards_timeout,
self.client.call(
"revoke_shards",
pod.uri(self.config.client_config.tls_enabled()),
move |client| {
let request = revoke_shards_request.clone();
Box::pin(client.revoke_shards(request))
},
),
)
.await
.map_err(|_: Elapsed| ShardManagerError::Timeout)?;The default configuration is: revoke_shards_timeout: Duration::from_secs(60),
retries: RetryConfig::max_attempts_5(),
let set_shard_assignment_response = timeout(
self.config.revoke_shards_timeout,
self.client.call(
"set_shard_assignment",
// ...
),
)
.await
.map_err(|_: Elapsed| ShardManagerError::Timeout)?;A concrete sequence can therefore look like this: The exact count depends on which calls fail and how reconciliation proceeds, but it is not bounded to one logical drain. 5. What each blocked task retainsEvery spawned task owns: let this = self.clone();It also holds, while waiting:
These tasks are asleep rather than busy-looping, so one or two are inexpensive. The issue is unbounded accumulation across retries and reconciliation passes. A persistently wedged but otherwise healthy executor can retain more tasks and worker/service state indefinitely. |
…ound, leave failed-revoke shards unassigned and reconcile the old executor
…ntil each is unloaded, raise the revoke deadline to 60s
|
All three findings were real -- thanks, these were sharp. Fixes for each are in the latest commits, plus some results from running the same kind of adversarial pass myself.
Separately, my own adversarial pass over the same blast radius turned up a few more things: Fixed here: ActiveAgents::remove could delete a pending cache entry (a new creation racing the removal's awaits), hiding an in-flight Worker::new from the drain and letting the finished creation re-insert itself afterwards — removal now uses remove_if_cached, which never touches pending entries. And the pre-existing wedge where a caller cancelled mid-Worker::new orphans its pending entry forever — which the drain would now inherit as a permanently hung revoke — is closed by moving creation to get_or_insert_simple_spawned, so the producer survives caller cancellation and the entry always settles. Found but deliberately not fixed here: (i) RPC handlers that write durably to an already-cached agent (invocation enqueue, update, cancel-invocation, revert, plugin activation, card delivery, promise completion, worker deletion) check ownership at RPC entry only — a handler already past its check when the revoke lands can perform its durable write after the drain returns, and the drain cannot see it (the agent is neither loaded nor pending); (ii) two post-is_loaded status writes — the final suspend checkpoint when an idle/eviction stop races the revoke, and the background status flusher retrying a failed stop-time flush (both self-heal on refold, which validates the baseline). Any fix for these today would be yet another point-in-time check in exactly the territory the oplog fence covers — every one of these paths funnels through the storage sites the fence guards — so the fixes depend on the fence and I'd land them with it rather than half-close them now. |
…econcile failed assignments before reassigning, single-flight the drain; survive caller-cancelled creations, never remove pending cache entries, guard the drain owner slot
e87dcf1 to
11a3064
Compare
| .collect(); | ||
|
|
||
| Ok(()) | ||
| if lost.is_empty() && pending_lost.is_empty() { |
There was a problem hiding this comment.
P1: Keep the revoke barrier open for already-admitted durable operations
This completion condition now handles pending Worker::new and Running workers, but it still misses cached workers in Unloaded or WaitingForPermit state and RPCs already past their ownership check.
For example, start_if_needed can validate ownership and install WaitingForPermit; because is_loaded() only recognizes Running, this drain can then return. The admitted invocation can subsequently reach enqueue_worker_invocation_with_effect and commit its pending oplog entry without another ownership check. Other durable handlers have the same gap.
Please extend the admission/quiescence barrier through durable enqueue/write completion, or add the atomic storage epoch fence here. Deferring that fence leaves RevokeShards able to report success while the old owner can still write.
| if !failed_pre_sets.is_empty() { | ||
| warn!( | ||
| failed_pods = failed_pre_sets.iter().map(|(pod, _)| pod).join(", "), | ||
| "Some pods could not receive their authoritative shard assignment before the rebalance; retrying after it" |
There was a problem hiding this comment.
P1: Do not rebalance shards whose authoritative reconciliation failed
Retrying this SetShardAssignment after the rebalance does not resolve the ownership uncertainty. If Assign(B, {5}) was applied but timed out, the same-pass reconciliation failed, and this pre-rebalance reconciliation also fails, B may still own shard 5. The code nevertheless proceeds to execute_rebalance, which can assign shard 5 to A before any successful Set(B, ...).
vigoo reproduced that sequence deterministically: A received shard 5 while B’s local assignment still contained shard 5. Please keep these in-doubt shards unassigned/suppress their reassignment until the authoritative Set succeeds, unless an atomic ownership epoch makes stale-owner writes impossible.
2 participants
fixes GOL-474
Problem
Chaos run on golem-dev (2026-08-18): 460 of 1024 shards had no owner for 43 minutes and the cluster never repaired itself.
RevokeShardson the executor removed the shards from its local assignment and then drained the affected agents one at a time, each step paying the agent's teardown (oplog commit + status flush + checkpoint). With enough agents that exceeded the shard manager's 5 s deadline. On failure the shard manager calledRebalance::remove_shards, which dropped the shards from both halves of the plan, so the routing table kept crediting the old executor (which had already dropped them locally) and the loop replayed the identical plan forever.A first version of this PR detached the drain into a background task. That was wrong (thanks @vigoo): returning before the agents are stopped lets an agent keep running on the old executor while the new owner recovers it, and both write the same oplog. Without an epoch fence the synchronous drain is the ordering barrier.
Change
Executor (
grpc/mod.rs) — the drain stays on the RPC, but:Worker::is_loaded()turning false, not the interrupt acknowledgement. The ack is sent fromset_suspendedbefore the invocation loop commits the oplog and flushes status;is_loaded()only flips insidestop_internal_lockedafter that commit and flush, on every exit path (including the ones that never remove the worker fromActiveAgents). Loaded-but-idle agents get no ack receiver but are woken and unload through the same path, so the same barrier covers them;SetShardAssignmentgets the same synchronous drain for the shards it removes (it is the reconcile the shard manager sends after a failed revoke); recovery of gained shards stays detached for bothAssignShardsandSetShardAssignment;InterruptKind::Restart→RetryDecision::Immediateused to re-instantiate on the losing executor), andon_shard_assignment_changedre-checks ownership before each restart.Shard manager
remove_assignment_shards, one-sided) and sends the old pod its authoritativeSetShardAssignmentin the same pass, before the next pass hands the shards to another pod. A failed assignment also queues the pod for reconciliation.revoke_shards_timeout5 s → 60 s. It bounds a drain whose duration is one agent's suspend, not the agent count; withretriesit is a liveness backstop (5 × 60 s is the longest a wedged-but-alive executor can hold one pass before its shards are released).SetShardAssignmentis wrapped inrevoke_shards_timeout(it shared the 5 sassign_shards_timeout).Tests
golem-worker-executor/tests/sharding.rs(new, group2, sequential):revoke_shards_returns_only_after_lost_agents_are_unloaded,revoke_shards_drains_lost_agents_concurrently(per-agent teardown commits delayed through a one-shot test-utils seam inTestOplog::commit; asserts the teardowns overlap),revoke_shards_unloads_idle_agents,set_shard_assignment_drains_lost_agents_before_returning. All four fail against the background drain (the RPC returned in <1 ms with the agents still loaded).golem-shard-manager/tests/shard_management.rs:revoke_timeout_after_executor_applied_it_does_not_strand_shards(the production failure shape; never converges onmain),failed_revoke_reconciles_old_executor_before_reassigning,unreachable_executor_still_releases_its_revoked_shards.Accepted behaviour / not in scope
Set, then its shards are released. That wait is the point.retries_on_unavailableloop retries the 10 s connect timeout inside each of the 5 attempts, so a black-holed pod can hold one pass for several minutes (before: ~25 s, cut off by the 5 s deadline). Left as is to keep this change minimal; lowering the client's connect timeout or its inner retries is a config-only knob if it shows up.claim_dueand processing; the rebalance loop has no backoff.Verification
cargo test -p golem-shard-manager --lib: 51 passed;--test integration(Postgres testcontainer): 15 passed, including the three revoke-failure tests.cargo test -p golem-worker-executor --test integration -- sharding::: 4 passed. Against the previous background drain the same four fail (... was still loaded when revoke_shards returned (after 0.5–0.8 ms)).Clocksagents drain in ~12 ms; 5 idle agents in ~260 ms; with a 200 ms artificial teardown commit per agent the concurrent drain takes ~330 ms (sequential would be ≥ 1 s).api::interruption,api::simulated_crash,wasi::sleep_*): 7 passed.cargo clippy --all-targetsongolem-shard-manager,golem-worker-executor,golem-worker-executor-test-utils: clean;cargo fmt --all -- --check: clean.golem-shard-manager/config/*regenerated from the binary: onlyrevoke_shards_timeout = "1m"changes.integration-tests --test sharding(coordinated_scenario_01_02,service_is_responsive_to_shard_changes) against rebuilt service binaries: 2 passed (81 s).