code: def test_pass_returned_object_ref(
      one_worker_100MiB, use_ray_put, failure
  ):
      @ray.remote
      def return_an_id():
          return [
              put_object(
                  np.zeros(20 * 1024 * 1024, dtype=np.uint8),
                  use_ray_put,
              )
          ]
  
      # TODO(edoakes): this fails with an ActorError with max_retries=1.
      @ray.remote(max_retries=0)
      def pending(ref, signal):
          ray.get(signal.wait.remote())
          ray.get(ref[0])
          if failure:
              os._exit(0)
  
      signal = SignalActor.remote()
      outer_oid = return_an_id.remote()
      inner_oid_binary = ray.get(outer_oid)[0].binary()
      pending_oid = pending.remote([outer_oid], signal)
  
      # Remove the local reference to the returned ID.
      del outer_oid
  
      # Check that the inner ID is pinned by the remote task ID and finishing
      # the task unpins the object.
      ray.get(signal.send.remote())
      try:
          # Should succeed because inner_oid is pinned if no failure.
          ray.get(pending_oid)
          assert not failure
      except ray.exceptions.WorkerCrashedError:
          assert failure
  
      def ref_not_exists():
          worker = ray._private.worker.global_worker
          inner_oid = ray.ObjectRef(inner_oid_binary)
          return not worker.core_worker.object_exists(inner_oid)
  
      wait_for_condition(ref_not_exists)
  
imports: import logging
  import os
  import copy
  import platform
  import random
  import signal
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import SignalActor, put_object, wait_for_condition, wait_for_num_actors
  import ray._private.gcs_utils
label_desc: Use image node_id 5 for calling a global function with example usage: test_pass_returned_object_ref(one_worker_100MiB, use_ray_put, failure) and returns: list, unknown