[promises] Fix some behaviors exposed by api_fuzzer (#34795)

1. we shouldn't spawn multiple promises to deal with cancellations (one
is enough)
2. lame_client needs to express that it will never read a message

src/core/lib/surface/call.cc

@@ -2766,6 +2766,9 @@ class ClientPromiseBasedCall final : public PromiseBasedCall {
   }
 
   void CancelWithError(absl::Status error) override {
+    if (cancel_with_error_called_.exchange(true, std::memory_order_relaxed)) {
+      return;
+    }
     if (!started_.exchange(true, std::memory_order_relaxed)) {
       // Initial metadata not sent yet, so we can just fail the call.
       Spawn(
@@ -2844,6 +2847,9 @@ class ClientPromiseBasedCall final : public PromiseBasedCall {
   // In the latter case real world code sometimes does not sent the initial
   // metadata, and so gating based upon that does not work out.
   std::atomic<bool> started_{false};
+  // True after the first CancelWithError call - prevents spamming cancels from
+  // overflowing the party.
+  std::atomic<bool> cancel_with_error_called_{false};
   // TODO(ctiller): delete when we remove the filter based API (may require some
   // cleanup in wrapped languages: they depend on this to hold slice refs)
   ServerMetadataHandle recv_initial_metadata_;

src/core/lib/surface/lame_client.cc

@@ -78,7 +78,13 @@ ArenaPromise<ServerMetadataHandle> LameClientFilter::MakeCallPromise(
   // TODO(ctiller): remove if check once promise_based_filter is removed (Close
   // is still needed)
   if (args.server_to_client_messages != nullptr) {
-    args.server_to_client_messages->Close();
+    args.server_to_client_messages->CloseWithError();
+  }
+  if (args.client_to_server_messages != nullptr) {
+    args.client_to_server_messages->CloseWithError();
+  }
+  if (args.server_initial_metadata != nullptr) {
+    args.server_initial_metadata->CloseWithError();
   }
   args.client_initial_metadata_outstanding.Complete(true);
   return Immediate(ServerMetadataFromStatus(error_));

test/core/end2end/fuzzers/BUILD

@@ -45,7 +45,6 @@ grpc_cc_library(
 
 grpc_proto_fuzzer(
     name = "api_fuzzer",
-    size = "enormous",
     srcs = ["api_fuzzer.cc"],
     corpus = "api_fuzzer_corpus",
     language = "C++",

test/core/end2end/fuzzers/api_fuzzer_corpus/testcase-4947272136720384

@@ -0,0 +1,445 @@
+actions {
+  create_server {
+  }
+}
+actions {
+  create_channel {
+  }
+}
+actions {
+  create_call {
+    propagation_mask: 37
+    method {
+      value: "\020\000"
+    }
+    host {
+      value: "1"
+    }
+    timeout: 27489
+  }
+}
+actions {
+  queue_batch {
+  }
+}
+actions {
+  ping {
+  }
+}
+actions {
+  destroy_call {
+  }
+}
+actions {
+  ping {
+  }
+}
+actions {
+  create_call {
+    propagation_mask: 16777216
+    method {
+      value: " 1"
+    }
+    host {
+      value: "\000\000\000\321\000\000\000\000\000\321\250"
+    }
+    timeout: 67108864
+  }
+}
+actions {
+  poll_cq {
+  }
+}
+actions {
+  ping {
+  }
+}
+actions {
+  check_connectivity: true
+}
+actions {
+  queue_batch {
+    operations {
+    }
+  }
+}
+actions {
+  queue_batch {
+  }
+}
+actions {
+  change_active_call {
+  }
+}
+actions {
+  change_active_call {
+  }
+}
+actions {
+  shutdown_server {
+  }
+}
+actions {
+  ping {
+  }
+}
+actions {
+  ping {
+  }
+}
+actions {
+  create_call {
+    propagation_mask: 253
+    method {
+      value: "\020\000"
+    }
+    host {
+      value: "["
+    }
+    timeout: 67108864
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      send_close_from_client {
+      }
+    }
+    operations {
+      send_close_from_client {
+      }
+    }
+    operations {
+      receive_initial_metadata {
+      }
+    }
+    operations {
+      send_initial_metadata {
+      }
+      flags: 4
+    }
+    operations {
+      receive_message {
+      }
+      flags: 10158080
+    }
+  }
+}
+actions {
+  queue_batch {
+    operations {
+    }
+  }
+}
+actions {
+  cancel_call {
+  }
+}
+actions {
+  get_peer {
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      receive_status_on_client {
+      }
+      flags: 813367296
+    }
+    operations {
+    }
+  }
+}
+actions {
+  poll_cq {
+  }
+}
+actions {
+  create_call {
+    propagation_mask: 1862270976
+    method {
+      value: "http"
+    }
+    timeout: 64
+  }
+}
+actions {
+  get_target {
+  }
+}
+actions {
+  ping {
+  }
+}
+actions {
+  change_active_call {
+  }
+}
+actions {
+  cancel_all_calls_if_shutdown {
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      receive_status_on_client {
+      }
+    }
+    operations {
+      receive_message {
+      }
+    }
+    operations {
+      send_initial_metadata {
+      }
+      flags: 4
+    }
+    operations {
+      flags: 536870912
+    }
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      send_close_from_client {
+      }
+    }
+    operations {
+      receive_initial_metadata {
+      }
+    }
+    operations {
+      send_initial_metadata {
+      }
+      flags: 4
+    }
+    operations {
+      receive_message {
+      }
+      flags: 10158080
+    }
+  }
+}
+actions {
+  disable_tracer: ""
+}
+actions {
+  ping {
+  }
+}
+actions {
+  resize_resource_quota: 813367296
+}
+actions {
+  poll_cq {
+  }
+}
+actions {
+  destroy_call {
+  }
+}
+actions {
+  check_connectivity: true
+}
+actions {
+  check_connectivity: true
+}
+actions {
+  destroy_call {
+  }
+}
+actions {
+  cancel_all_calls_if_shutdown {
+  }
+}
+actions {
+  disable_tracer: "+"
+}
+actions {
+  create_call {
+    propagation_mask: 4
+    method {
+      value: " 1"
+    }
+    host {
+      value: "["
+    }
+    timeout: 1801545216
+  }
+}
+actions {
+  create_call {
+    method {
+      value: "["
+    }
+    timeout: 7012449
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      receive_status_on_client {
+      }
+      flags: 6619136
+    }
+    operations {
+      flags: 1862270976
+    }
+  }
+}
+actions {
+  change_active_call {
+  }
+}
+actions {
+  create_call {
+    propagation_mask: 253
+    method {
+      value: "\020\000"
+    }
+    host {
+      value: "["
+    }
+    timeout: 67108864
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      send_close_from_client {
+      }
+    }
+    operations {
+      receive_initial_metadata {
+      }
+    }
+    operations {
+      send_initial_metadata {
+      }
+      flags: 4
+    }
+    operations {
+      receive_message {
+      }
+      flags: 10158080
+    }
+  }
+}
+actions {
+  change_active_call {
+  }
+}
+actions {
+  cancel_all_calls_if_shutdown {
+  }
+}
+actions {
+  ping {
+  }
+}
+actions {
+  destroy_call {
+  }
+}
+actions {
+  get_target {
+  }
+}
+actions {
+  destroy_call {
+  }
+}
+actions {
+  check_connectivity: true
+}
+actions {
+  watch_connectivity: 1023410176
+}
+actions {
+  enable_tracer: "grpc.defaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaault_compression_algorithm"
+}
+actions {
+  change_active_call {
+  }
+}
+actions {
+  cancel_all_calls_if_shutdown {
+  }
+}
+actions {
+  get_peer {
+  }
+}
+actions {
+  watch_connectivity: 121
+}
+actions {
+  create_call {
+    method {
+      value: "\000?"
+    }
+    timeout: 67108864
+  }
+}
+actions {
+  enable_tracer: "\000?"
+}
+actions {
+  cancel_call {
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      send_initial_metadata {
+      }
+      flags: 6291456
+    }
+  }
+}
+actions {
+  queue_batch {
+  }
+}
+actions {
+}
+actions {
+}
+actions {
+  get_target {
+  }
+}
+actions {
+  resize_resource_quota: 2
+}
+actions {
+  destroy_call {
+  }
+}
+actions {
+}
+actions {
+}
+actions {
+  poll_cq {
+  }
+}
+actions {
+}
+actions {
+  resize_resource_quota: 536870912
+}
+actions {
+  shutdown_server {
+  }
+}
+actions {
+  queue_batch {
+  }
+}
+config_vars {
+  enable_fork_support: true
+  verbosity: "."
+  experiments: 16357073846609641471
+}

test/core/util/fuzzer_corpus_test.cc

@@ -53,7 +53,7 @@ TEST_P(FuzzerCorpusTest, RunOneExample) {
   // down before calling LLVMFuzzerTestOneInput(), because most
   // implementations of that function will initialize and shutdown gRPC
   // internally.
-  gpr_log(GPR_INFO, "Example file: %s", GetParam().c_str());
+  fprintf(stderr, "Example file: %s\n", GetParam().c_str());
   grpc_slice buffer;
   squelch = false;
   GPR_ASSERT(GRPC_LOG_IF_ERROR("load_file",