[Core] Placement-group resource cleanup is abandoned after five transient RPC failures

[logical-misha]

What happened + What you expected to happen

GCS permanently stops RemovePlacementGroupBundles cleanup after five failed
RPC attempts, even when the target node remains registered as alive.

DestroyPlacementGroupCommittedBundleResources also erases the GCS-local
committed bundle indexes immediately after starting the RPC. Once the fixed
retry budget is exhausted, no cleanup obligation remains in GCS, while the
raylet may still hold the placement-group resources.

Actual behavior: five injected transient UNAVAILABLE responses produce five
requests, then GCS logs that the max retry count was reached and stops.

Expected behavior: while the GCS source of truth still says the node is alive,
cleanup should remain outstanding and continue with backoff. It is reasonable
to stop once the node is known dead, because that raylet can no longer hold
live resources.

Versions / Dependencies

• Ray: 3.0.0.dev0, commit
  3d026f5b65e52955cb8b3e21d88941bb0ff5392c
• OS: Ubuntu 24.04, Linux 6.17, x86-64
• Python: 3.12.3
• Bazel: 7.5.0
• GCC: 13.3.0

Reproduction script

Apply the complete regression-test patch included at the end, then run:

bazel test //src/ray/gcs/tests:gcs_placement_group_scheduler_test \
  --test_filter=GcsPlacementGroupSchedulerTest.DestroyPlacementGroupKeepsRetryingLiveNode \
  --test_output=errors

The test creates and commits a placement group, starts destruction, leaves the
node alive, and returns five transient RPC failures:

auto node = GenNodeInfo();
AddNode(node);

auto pg = std::make_shared<GcsPlacementGroup>(
    GenCreatePlacementGroupRequest(), "", counter_, clock_);
scheduler_->ScheduleUnplacedBundles(SchedulePgRequest{
    pg,
    [this](auto pg, bool) {
      absl::MutexLock lock(&placement_group_requests_mutex_);
      failure_placement_groups_.push_back(std::move(pg));
    },
    [this](auto pg) {
      absl::MutexLock lock(&placement_group_requests_mutex_);
      success_placement_groups_.push_back(std::move(pg));
    }});

ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources());
WaitPendingDone(raylet_clients_[0]->commit_callbacks, 1);
ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources());
WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS);

scheduler_->DestroyPlacementGroupBundleResourcesIfExists(
    pg->GetPlacementGroupID());

for (int attempt = 0; attempt < 5; ++attempt) {
  WaitPendingDone(raylet_clients_[0]->remove_pg_bundles_callbacks, 1);
  ASSERT_TRUE(raylet_clients_[0]->GrantRemovePlacementGroupBundles(
      Status::RpcError("unavailable", grpc::StatusCode::UNAVAILABLE)));
}

WaitPendingDone(raylet_clients_[0]->remove_pg_bundles_callbacks, 1);
ASSERT_GT(raylet_clients_[0]->num_remove_pg_bundles_requested, 5);

Current master fails with:

Failed to remove 2 bundle(s) ... because the max retry count is reached.
Expected: (raylet_clients_[0]->num_remove_pg_bundles_requested) > (5),
  actual: 5 vs 5

Relevant implementation

RemovePlacementGroupBundles returns immediately when
max_retry == current_retry_count. That branch does not consult
GcsNodeManager to distinguish a dead node from a live node experiencing
transient RPC failures. DestroyPlacementGroupCommittedBundleResources then
erases both bundle-location indexes independently of the RPC outcome.

A minimal fix can recheck node liveness when the bounded retry count is reached
and continue retrying with backoff while the node remains alive. A stronger
design would retain an explicit cleanup obligation until either removal is
acknowledged or GCS observes node death. The tested live-node continuation fix
passes this regression and nearby placement-group destruction tests.

Open issue #54332 states the broader policy that network failures are transient
and retries should stop only when the downstream node is known dead. It tracks
several concrete RPCs, but does not list this RemovePlacementGroupBundles
retry-exhaustion path.

Issue Severity

High: a finite run of transient failures can permanently leak placement-group
resources on a live node, reducing cluster capacity until the raylet exits.

Complete regression-test patch

diff --git a/src/ray/gcs/tests/gcs_placement_group_scheduler_test.cc b/src/ray/gcs/tests/gcs_placement_group_scheduler_test.cc
index 8c7b8a4..ee1419f 100644
--- a/src/ray/gcs/tests/gcs_placement_group_scheduler_test.cc
+++ b/src/ray/gcs/tests/gcs_placement_group_scheduler_test.cc
@@ -593,6 +593,44 @@ TEST_F(GcsPlacementGroupSchedulerTest, DestroyPlacementGroup) {
   ASSERT_FALSE(raylet_clients_[0]->GrantRemovePlacementGroupBundles());
 }
 
+TEST_F(GcsPlacementGroupSchedulerTest, DestroyPlacementGroupKeepsRetryingLiveNode) {
+  auto node = GenNodeInfo();
+  AddNode(node);
+  ASSERT_EQ(1, gcs_node_manager_->GetAllAliveNodes().size());
+
+  auto placement_group = std::make_shared<GcsPlacementGroup>(
+      GenCreatePlacementGroupRequest(), "", counter_, clock_);
+
+  scheduler_->ScheduleUnplacedBundles(SchedulePgRequest{
+      placement_group,
+      [this](std::shared_ptr<GcsPlacementGroup> placement_group, bool is_insfeasble) {
+        absl::MutexLock lock(&placement_group_requests_mutex_);
+        failure_placement_groups_.emplace_back(std::move(placement_group));
+      },
+      [this](std::shared_ptr<GcsPlacementGroup> placement_group) {
+        absl::MutexLock lock(&placement_group_requests_mutex_);
+        success_placement_groups_.emplace_back(std::move(placement_group));
+      }});
+  ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources());
+  WaitPendingDone(raylet_clients_[0]->commit_callbacks, 1);
+  ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources());
+  WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS);
+
+  scheduler_->DestroyPlacementGroupBundleResourcesIfExists(
+      placement_group->GetPlacementGroupID());
+
+  for (int attempt = 0; attempt < 5; ++attempt) {
+    WaitPendingDone(raylet_clients_[0]->remove_pg_bundles_callbacks, 1);
+    ASSERT_TRUE(raylet_clients_[0]->GrantRemovePlacementGroupBundles(
+        Status::RpcError("unavailable", grpc::StatusCode::UNAVAILABLE)));
+  }
+
+  // The node is still alive. Cleanup should not be permanently abandoned after
+  // the fixed retry budget is exhausted.
+  WaitPendingDone(raylet_clients_[0]->remove_pg_bundles_callbacks, 1);
+  ASSERT_GT(raylet_clients_[0]->num_remove_pg_bundles_requested, 5);
+}
+
 TEST_F(GcsPlacementGroupSchedulerTest, DestroyCancelledPlacementGroup) {
   auto node0 = GenNodeInfo(0);
   auto node1 = GenNodeInfo(1);
diff --git a/src/ray/raylet_rpc_client/fake_raylet_client.h b/src/ray/raylet_rpc_client/fake_raylet_client.h
index 0d16b48..8b74090 100644
--- a/src/ray/raylet_rpc_client/fake_raylet_client.h
+++ b/src/ray/raylet_rpc_client/fake_raylet_client.h
@@ -222,8 +222,7 @@ class FakeRayletClient : public RayletClientInterface {
     }
   }
 
-  bool GrantRemovePlacementGroupBundles(bool success = true) {
-    Status status = Status::OK();
+  bool GrantRemovePlacementGroupBundles(const Status &status = Status::OK()) {
     RemovePlacementGroupBundlesReply reply;
     if (remove_pg_bundles_callbacks.size() == 0) {
       return false;