Fix timeout bug in api fuzzer (#28385)

* Fix timeout bug in api fuzzer

* add example

test/core/end2end/fuzzers/api_fuzzer.cc

@@ -220,8 +220,8 @@ static void do_connect(void* arg, grpc_error_handle error) {
     grpc_endpoint* server;
     grpc_passthru_endpoint_create(&client, &server, nullptr, true);
     *fc->ep = client;
-    start_scheduling_grpc_passthru_endpoint_channel_effects(
-        client, g_channel_actions, [&]() { g_channel_force_delete = true; });
+    start_scheduling_grpc_passthru_endpoint_channel_effects(client,
+                                                            g_channel_actions);
 
     grpc_core::Server* core_server = grpc_core::Server::FromC(g_server);
     grpc_transport* transport = grpc_create_chttp2_transport(

test/core/end2end/fuzzers/api_fuzzer_corpus/timeout1

@@ -0,0 +1,105 @@
+actions {
+  create_server {
+  }
+}
+actions {
+  create_channel {
+    target: "unix::::::::::::::::::::::::::;:::::;:::::::::::>:::::::::::::::::::::::9:::::\026I:::::::c::,::::\001::::::::::::::::"
+    channel_actions {
+      add_n_bytes_writable: 55
+      add_n_bytes_readable: 106652627894272
+    }
+  }
+}
+actions {
+  advance_time: 257
+}
+actions {
+  create_call {
+    propagation_mask: 83968879
+    method {
+      value: "u-bin"
+    }
+    host {
+      value: "\361\243\275\203\361\226\215\265\363\274\273\202\363\234\232\265\360\243\230\252\361\241\227\253\360\233\262\234\363\242\252\237\362\231\230\250\360\260\203\220\362\225\271\250\360\264\257\267\361\201\262\227\363\235\256\253\360\250\251\207\362\254\245\212\362\270\225\220\362\227\235\250\363\227\272\241\362\254\272\206\361\251\216\231\361\200\275\233\363\242\271\201\361\252\244\252\361\253\210\274\361\214\232\240\362\272\276\276\362\234\247\242\345\225\225"
+      intern: true
+    }
+    timeout: 1000000000
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      send_initial_metadata {
+        metadata {
+          key {
+            value: "oo"
+          }
+          value {
+            value: "u-bin"
+            intern: true
+          }
+        }
+        metadata {
+          key {
+            value: "u-bio"
+          }
+          value {
+            value: "u-bin"
+          }
+        }
+        metadata {
+          key {
+            value: "u-biogrpc.lame_filter_error"
+            intern: true
+          }
+          value {
+            value: "u-bin"
+            intern: true
+          }
+        }
+        metadata {
+          key {
+            value: "u-bin"
+            intern: true
+          }
+          value {
+            value: "unix::pehttp,\014,,Sac,,t"
+          }
+        }
+      }
+    }
+    operations {
+      flags: 53
+    }
+    operations {
+      flags: 1886680168
+    }
+  }
+}
+actions {
+  request_call {
+  }
+}
+actions {
+  advance_time: 9869440
+}
+actions {
+  create_server {
+  }
+}
+actions {
+}
+actions {
+  get_target {
+  }
+}
+actions {
+  enable_tracer: "u-bin"
+}
+actions {
+}
+actions {
+}
+actions {
+}

test/core/util/passthru_endpoint.cc

@@ -307,9 +307,7 @@ static void me_add_to_pollset_set(grpc_endpoint* /*ep*/,
 static void me_delete_from_pollset_set(grpc_endpoint* /*ep*/,
                                        grpc_pollset_set* /*pollset*/) {}
 
-static void me_shutdown(grpc_endpoint* ep, grpc_error_handle why) {
-  half* m = reinterpret_cast<half*>(ep);
-  gpr_mu_lock(&m->parent->mu);
+static void shutdown_locked(half* m, grpc_error_handle why) {
   m->parent->shutdown = true;
   flush_pending_ops_locked(m, GRPC_ERROR_NONE);
   if (m->on_read) {
@@ -326,6 +324,12 @@ static void me_shutdown(grpc_endpoint* ep, grpc_error_handle why) {
         GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING("Shutdown", &why, 1));
     m->on_read = nullptr;
   }
+}
+
+static void me_shutdown(grpc_endpoint* ep, grpc_error_handle why) {
+  half* m = reinterpret_cast<half*>(ep);
+  gpr_mu_lock(&m->parent->mu);
+  shutdown_locked(m, why);
   gpr_mu_unlock(&m->parent->mu);
   GRPC_ERROR_UNREF(why);
 }
@@ -472,7 +476,10 @@ static void do_next_sched_channel_action(void* arg, grpc_error_handle error) {
 
 static void sched_next_channel_action_locked(half* m) {
   if (m->parent->channel_effects->actions.empty()) {
-    m->parent->channel_effects->on_complete();
+    auto* err =
+        GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel actions complete");
+    shutdown_locked(m, err);
+    GRPC_ERROR_UNREF(err);
     return;
   }
   grpc_timer_init(&m->parent->channel_effects->timer,
@@ -484,8 +491,7 @@ static void sched_next_channel_action_locked(half* m) {
 
 void start_scheduling_grpc_passthru_endpoint_channel_effects(
     grpc_endpoint* ep,
-    const std::vector<grpc_passthru_endpoint_channel_action>& actions,
-    std::function<void()> on_complete) {
+    const std::vector<grpc_passthru_endpoint_channel_action>& actions) {
   half* m = reinterpret_cast<half*>(ep);
   gpr_mu_lock(&m->parent->mu);
   if (!m->parent->simulate_channel_actions || m->parent->shutdown) {
@@ -493,7 +499,6 @@ void start_scheduling_grpc_passthru_endpoint_channel_effects(
     return;
   }
   m->parent->channel_effects->actions = actions;
-  m->parent->channel_effects->on_complete = std::move(on_complete);
   sched_next_channel_action_locked(m);
   gpr_mu_unlock(&m->parent->mu);
 }

test/core/util/passthru_endpoint.h

@@ -48,7 +48,6 @@ void grpc_passthru_endpoint_stats_destroy(grpc_passthru_endpoint_stats* stats);
 
 void start_scheduling_grpc_passthru_endpoint_channel_effects(
     grpc_endpoint* ep,
-    const std::vector<grpc_passthru_endpoint_channel_action>& actions,
-    std::function<void()> on_complete);
+    const std::vector<grpc_passthru_endpoint_channel_action>& actions);
 
 #endif  // PASSTHRU_ENDPOINT_H