Merge pull request #5726 from ctiller/ignore_connectivity

Implement fail fast semantics for C core

11 files changed, 143 insertions, 43 deletions

diff --git a/src/core/ext/lb_policy/pick_first/pick_first.c b/src/core/ext/lb_policy/pick_first/pick_first.c
index 709db21a57..43197c1191 100644
--- a/src/core/ext/lb_policy/pick_first/pick_first.c
+++ b/src/core/ext/lb_policy/pick_first/pick_first.c
@@ -40,6 +40,7 @@
 typedef struct pending_pick {
   struct pending_pick *next;
   grpc_pollset *pollset;
+  uint32_t initial_metadata_flags;
   grpc_connected_subchannel **target;
   grpc_closure *on_complete;
 } pending_pick;
@@ -149,6 +150,31 @@ static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
   gpr_mu_unlock(&p->mu);
 }
 
+static void pf_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
+                            uint32_t initial_metadata_flags_mask,
+                            uint32_t initial_metadata_flags_eq) {
+  pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
+  pending_pick *pp;
+  gpr_mu_lock(&p->mu);
+  pp = p->pending_picks;
+  p->pending_picks = NULL;
+  while (pp != NULL) {
+    pending_pick *next = pp->next;
+    if ((pp->initial_metadata_flags & initial_metadata_flags_mask) ==
+        initial_metadata_flags_eq) {
+      grpc_pollset_set_del_pollset(exec_ctx, p->base.interested_parties,
+                                   pp->pollset);
+      grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, false, NULL);
+      gpr_free(pp);
+    } else {
+      pp->next = p->pending_picks;
+      p->pending_picks = pp;
+    }
+    pp = next;
+  }
+  gpr_mu_unlock(&p->mu);
+}
+
 static void start_picking(grpc_exec_ctx *exec_ctx, pick_first_lb_policy *p) {
   p->started_picking = 1;
   p->checking_subchannel = 0;
@@ -171,6 +197,7 @@ static void pf_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
 
 static int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
                    grpc_pollset *pollset, grpc_metadata_batch *initial_metadata,
+                   uint32_t initial_metadata_flags,
                    grpc_connected_subchannel **target,
                    grpc_closure *on_complete) {
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
@@ -199,6 +226,7 @@ static int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
     pp->next = p->pending_picks;
     pp->pollset = pollset;
     pp->target = target;
+    pp->initial_metadata_flags = initial_metadata_flags;
     pp->on_complete = on_complete;
     p->pending_picks = pp;
     gpr_mu_unlock(&p->mu);
@@ -286,11 +314,14 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
             &p->checking_connectivity, &p->connectivity_changed);
         break;
       case GRPC_CHANNEL_TRANSIENT_FAILURE:
-        grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
-                                    GRPC_CHANNEL_TRANSIENT_FAILURE,
-                                    "connecting_transient_failure");
         p->checking_subchannel =
             (p->checking_subchannel + 1) % p->num_subchannels;
+        if (p->checking_subchannel == 0) {
+          /* only trigger transient failure when we've tried all alternatives */
+          grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
+                                      GRPC_CHANNEL_TRANSIENT_FAILURE,
+                                      "connecting_transient_failure");
+        }
         p->checking_connectivity = grpc_subchannel_check_connectivity(
             p->subchannels[p->checking_subchannel]);
         if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) {
@@ -378,14 +409,9 @@ static void pf_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
 }
 
 static const grpc_lb_policy_vtable pick_first_lb_policy_vtable = {
-    pf_destroy,
-    pf_shutdown,
-    pf_pick,
-    pf_cancel_pick,
-    pf_ping_one,
-    pf_exit_idle,
-    pf_check_connectivity,
-    pf_notify_on_state_change};
+    pf_destroy,     pf_shutdown,           pf_pick,
+    pf_cancel_pick, pf_cancel_picks,       pf_ping_one,
+    pf_exit_idle,   pf_check_connectivity, pf_notify_on_state_change};
 
 static void pick_first_factory_ref(grpc_lb_policy_factory *factory) {}
 
diff --git a/src/core/ext/lb_policy/round_robin/round_robin.c b/src/core/ext/lb_policy/round_robin/round_robin.c
index 68943d808a..eac6971099 100644
--- a/src/core/ext/lb_policy/round_robin/round_robin.c
+++ b/src/core/ext/lb_policy/round_robin/round_robin.c
@@ -49,6 +49,7 @@ int grpc_lb_round_robin_trace = 0;
 typedef struct pending_pick {
   struct pending_pick *next;
   grpc_pollset *pollset;
+  uint32_t initial_metadata_flags;
   grpc_connected_subchannel **target;
   grpc_closure *on_complete;
 } pending_pick;
@@ -275,6 +276,32 @@ static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
   gpr_mu_unlock(&p->mu);
 }
 
+static void rr_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
+                            uint32_t initial_metadata_flags_mask,
+                            uint32_t initial_metadata_flags_eq) {
+  round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
+  pending_pick *pp;
+  gpr_mu_lock(&p->mu);
+  pp = p->pending_picks;
+  p->pending_picks = NULL;
+  while (pp != NULL) {
+    pending_pick *next = pp->next;
+    if ((pp->initial_metadata_flags & initial_metadata_flags_mask) ==
+        initial_metadata_flags_eq) {
+      grpc_pollset_set_del_pollset(exec_ctx, p->base.interested_parties,
+                                   pp->pollset);
+      *pp->target = NULL;
+      grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, false, NULL);
+      gpr_free(pp);
+    } else {
+      pp->next = p->pending_picks;
+      p->pending_picks = pp;
+    }
+    pp = next;
+  }
+  gpr_mu_unlock(&p->mu);
+}
+
 static void start_picking(grpc_exec_ctx *exec_ctx, round_robin_lb_policy *p) {
   size_t i;
   p->started_picking = 1;
@@ -303,6 +330,7 @@ static void rr_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
 
 static int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
                    grpc_pollset *pollset, grpc_metadata_batch *initial_metadata,
+                   uint32_t initial_metadata_flags,
                    grpc_connected_subchannel **target,
                    grpc_closure *on_complete) {
   round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
@@ -330,6 +358,7 @@ static int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
     pp->pollset = pollset;
     pp->target = target;
     pp->on_complete = on_complete;
+    pp->initial_metadata_flags = initial_metadata_flags;
     p->pending_picks = pp;
     gpr_mu_unlock(&p->mu);
     return 0;
@@ -485,14 +514,9 @@ static void rr_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
 }
 
 static const grpc_lb_policy_vtable round_robin_lb_policy_vtable = {
-    rr_destroy,
-    rr_shutdown,
-    rr_pick,
-    rr_cancel_pick,
-    rr_ping_one,
-    rr_exit_idle,
-    rr_check_connectivity,
-    rr_notify_on_state_change};
+    rr_destroy,     rr_shutdown,           rr_pick,
+    rr_cancel_pick, rr_cancel_picks,       rr_ping_one,
+    rr_exit_idle,   rr_check_connectivity, rr_notify_on_state_change};
 
 static void round_robin_factory_ref(grpc_lb_policy_factory *factory) {}
 
diff --git a/src/core/lib/channel/client_channel.c b/src/core/lib/channel/client_channel.c
index 3f7cf1cf97..55867ba80c 100644
--- a/src/core/lib/channel/client_channel.c
+++ b/src/core/lib/channel/client_channel.c
@@ -114,6 +114,22 @@ static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand,
                             grpc_lb_policy *lb_policy,
                             grpc_connectivity_state current_state);
 
+static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx,
+                                                  channel_data *chand,
+                                                  grpc_connectivity_state state,
+                                                  const char *reason) {
+  if ((state == GRPC_CHANNEL_TRANSIENT_FAILURE ||
+       state == GRPC_CHANNEL_FATAL_FAILURE) &&
+      chand->lb_policy != NULL) {
+    /* cancel fail-fast picks */
+    grpc_lb_policy_cancel_picks(
+        exec_ctx, chand->lb_policy,
+        /* mask= */ GRPC_INITIAL_METADATA_IGNORE_CONNECTIVITY,
+        /* check= */ 0);
+  }
+  grpc_connectivity_state_set(exec_ctx, &chand->state_tracker, state, reason);
+}
+
 static void on_lb_policy_state_changed_locked(
     grpc_exec_ctx *exec_ctx, lb_policy_connectivity_watcher *w) {
   grpc_connectivity_state publish_state = w->state;
@@ -127,8 +143,8 @@ static void on_lb_policy_state_changed_locked(
     GRPC_LB_POLICY_UNREF(exec_ctx, w->chand->lb_policy, "channel");
     w->chand->lb_policy = NULL;
   }
-  grpc_connectivity_state_set(exec_ctx, &w->chand->state_tracker, publish_state,
-                              "lb_changed");
+  set_channel_connectivity_state_locked(exec_ctx, w->chand, publish_state,
+                                        "lb_changed");
   if (w->state != GRPC_CHANNEL_FATAL_FAILURE) {
     watch_lb_policy(exec_ctx, w->chand, w->lb_policy, w->state);
   }
@@ -200,8 +216,8 @@ static void cc_on_config_changed(grpc_exec_ctx *exec_ctx, void *arg,
   }
 
   if (iomgr_success && chand->resolver) {
-    grpc_connectivity_state_set(exec_ctx, &chand->state_tracker, state,
-                                "new_lb+resolver");
+    set_channel_connectivity_state_locked(exec_ctx, chand, state,
+                                          "new_lb+resolver");
     if (lb_policy != NULL) {
       watch_lb_policy(exec_ctx, chand, lb_policy, state);
     }
@@ -216,8 +232,8 @@ static void cc_on_config_changed(grpc_exec_ctx *exec_ctx, void *arg,
       GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel");
       chand->resolver = NULL;
     }
-    grpc_connectivity_state_set(exec_ctx, &chand->state_tracker,
-                                GRPC_CHANNEL_FATAL_FAILURE, "resolver_gone");
+    set_channel_connectivity_state_locked(
+        exec_ctx, chand, GRPC_CHANNEL_FATAL_FAILURE, "resolver_gone");
     gpr_mu_unlock(&chand->mu_config);
   }
 
@@ -272,8 +288,8 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
   }
 
   if (op->disconnect && chand->resolver != NULL) {
-    grpc_connectivity_state_set(exec_ctx, &chand->state_tracker,
-                                GRPC_CHANNEL_FATAL_FAILURE, "disconnect");
+    set_channel_connectivity_state_locked(
+        exec_ctx, chand, GRPC_CHANNEL_FATAL_FAILURE, "disconnect");
     grpc_resolver_shutdown(exec_ctx, chand->resolver);
     GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel");
     chand->resolver = NULL;
@@ -290,6 +306,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
 
 typedef struct {
   grpc_metadata_batch *initial_metadata;
+  uint32_t initial_metadata_flags;
   grpc_connected_subchannel **connected_subchannel;
   grpc_closure *on_ready;
   grpc_call_element *elem;
@@ -298,6 +315,7 @@ typedef struct {
 
 static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
                               grpc_metadata_batch *initial_metadata,
+                              uint32_t initial_metadata_flags,
                               grpc_connected_subchannel **connected_subchannel,
                               grpc_closure *on_ready);
 
@@ -308,6 +326,7 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
   } else if (cpa->connected_subchannel == NULL) {
     /* cancelled, do nothing */
   } else if (cc_pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
+                                cpa->initial_metadata_flags,
                                 cpa->connected_subchannel, cpa->on_ready)) {
     grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, true, NULL);
   }
@@ -316,6 +335,7 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
 
 static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
                               grpc_metadata_batch *initial_metadata,
+                              uint32_t initial_metadata_flags,
                               grpc_connected_subchannel **connected_subchannel,
                               grpc_closure *on_ready) {
   grpc_call_element *elem = elemp;
@@ -349,7 +369,8 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
     GRPC_LB_POLICY_REF(lb_policy, "cc_pick_subchannel");
     gpr_mu_unlock(&chand->mu_config);
     r = grpc_lb_policy_pick(exec_ctx, lb_policy, calld->pollset,
-                            initial_metadata, connected_subchannel, on_ready);
+                            initial_metadata, initial_metadata_flags,
+                            connected_subchannel, on_ready);
     GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "cc_pick_subchannel");
     return r;
   }
@@ -362,6 +383,7 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
   }
   cpa = gpr_malloc(sizeof(*cpa));
   cpa->initial_metadata = initial_metadata;
+  cpa->initial_metadata_flags = initial_metadata_flags;
   cpa->connected_subchannel = connected_subchannel;
   cpa->on_ready = on_ready;
   cpa->elem = elem;
diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c
index deba23e21f..211f537c69 100644
--- a/src/core/lib/channel/http_client_filter.c
+++ b/src/core/lib/channel/http_client_filter.c
@@ -111,10 +111,12 @@ static void hc_mutate_op(grpc_call_element *elem,
                                elem);
     /* Send : prefixed headers, which have to be before any application
        layer headers. */
-    grpc_metadata_batch_add_head(op->send_initial_metadata, &calld->method,
-                                 op->send_idempotent_request
-                                     ? GRPC_MDELEM_METHOD_PUT
-                                     : GRPC_MDELEM_METHOD_POST);
+    grpc_metadata_batch_add_head(
+        op->send_initial_metadata, &calld->method,
+        op->send_initial_metadata_flags &
+                GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST
+            ? GRPC_MDELEM_METHOD_PUT
+            : GRPC_MDELEM_METHOD_POST);
     grpc_metadata_batch_add_head(op->send_initial_metadata, &calld->scheme,
                                  channeld->static_scheme);
     grpc_metadata_batch_add_tail(op->send_initial_metadata, &calld->te_trailers,
diff --git a/src/core/lib/channel/subchannel_call_holder.c b/src/core/lib/channel/subchannel_call_holder.c
index 22f3679bf5..a2686a380a 100644
--- a/src/core/lib/channel/subchannel_call_holder.c
+++ b/src/core/lib/channel/subchannel_call_holder.c
@@ -127,7 +127,7 @@ retry:
           break;
         case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL:
           holder->pick_subchannel(exec_ctx, holder->pick_subchannel_arg, NULL,
-                                  &holder->connected_subchannel, NULL);
+                                  0, &holder->connected_subchannel, NULL);
           break;
       }
       gpr_mu_unlock(&holder->mu);
@@ -145,7 +145,8 @@ retry:
     GRPC_CALL_STACK_REF(holder->owning_call, "pick_subchannel");
     if (holder->pick_subchannel(
             exec_ctx, holder->pick_subchannel_arg, op->send_initial_metadata,
-            &holder->connected_subchannel, &holder->next_step)) {
+            op->send_initial_metadata_flags, &holder->connected_subchannel,
+            &holder->next_step)) {
       holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
       GRPC_CALL_STACK_UNREF(exec_ctx, holder->owning_call, "pick_subchannel");
     }
diff --git a/src/core/lib/channel/subchannel_call_holder.h b/src/core/lib/channel/subchannel_call_holder.h
index 5cf291a266..769e22b061 100644
--- a/src/core/lib/channel/subchannel_call_holder.h
+++ b/src/core/lib/channel/subchannel_call_holder.h
@@ -42,6 +42,7 @@
     called when the subchannel is available) */
 typedef int (*grpc_subchannel_call_holder_pick_subchannel)(
     grpc_exec_ctx *exec_ctx, void *arg, grpc_metadata_batch *initial_metadata,
+    uint32_t initial_metadata_flags,
     grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready);
 
 typedef enum {
diff --git a/src/core/lib/client_config/lb_policy.c b/src/core/lib/client_config/lb_policy.c
index 3d23669ec2..6174162d49 100644
--- a/src/core/lib/client_config/lb_policy.c
+++ b/src/core/lib/client_config/lb_policy.c
@@ -101,10 +101,11 @@ void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx,
 int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                         grpc_pollset *pollset,
                         grpc_metadata_batch *initial_metadata,
+                        uint32_t initial_metadata_flags,
                         grpc_connected_subchannel **target,
                         grpc_closure *on_complete) {
   return policy->vtable->pick(exec_ctx, policy, pollset, initial_metadata,
-                              target, on_complete);
+                              initial_metadata_flags, target, on_complete);
 }
 
 void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
@@ -112,6 +113,14 @@ void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
   policy->vtable->cancel_pick(exec_ctx, policy, target);
 }
 
+void grpc_lb_policy_cancel_picks(grpc_exec_ctx *exec_ctx,
+                                 grpc_lb_policy *policy,
+                                 uint32_t initial_metadata_flags_mask,
+                                 uint32_t initial_metadata_flags_eq) {
+  policy->vtable->cancel_picks(exec_ctx, policy, initial_metadata_flags_mask,
+                               initial_metadata_flags_eq);
+}
+
 void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
   policy->vtable->exit_idle(exec_ctx, policy);
 }
diff --git a/src/core/lib/client_config/lb_policy.h b/src/core/lib/client_config/lb_policy.h
index a63e8e68df..60729ef8f9 100644
--- a/src/core/lib/client_config/lb_policy.h
+++ b/src/core/lib/client_config/lb_policy.h
@@ -60,9 +60,13 @@ struct grpc_lb_policy_vtable {
   /** implement grpc_lb_policy_pick */
   int (*pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
               grpc_pollset *pollset, grpc_metadata_batch *initial_metadata,
+              uint32_t initial_metadata_flags,
               grpc_connected_subchannel **target, grpc_closure *on_complete);
   void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                       grpc_connected_subchannel **target);
+  void (*cancel_picks)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
+                       uint32_t initial_metadata_flags_mask,
+                       uint32_t initial_metadata_flags_eq);
 
   void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                    grpc_closure *closure);
@@ -122,6 +126,7 @@ void grpc_lb_policy_init(grpc_lb_policy *policy,
 int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                         grpc_pollset *pollset,
                         grpc_metadata_batch *initial_metadata,
+                        uint32_t initial_metadata_flags,
                         grpc_connected_subchannel **target,
                         grpc_closure *on_complete);
 
@@ -131,6 +136,14 @@ void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
 void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                                 grpc_connected_subchannel **target);
 
+/** Cancel all pending picks which have:
+    (initial_metadata_flags & initial_metadata_flags_mask) ==
+         initial_metadata_flags_eq */
+void grpc_lb_policy_cancel_picks(grpc_exec_ctx *exec_ctx,
+                                 grpc_lb_policy *policy,
+                                 uint32_t initial_metadata_flags_mask,
+                                 uint32_t initial_metadata_flags_eq);
+
 void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
 
 void grpc_lb_policy_notify_on_state_change(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/lib/client_config/subchannel.c b/src/core/lib/client_config/subchannel.c
index 47c53a16ba..b4fd770e24 100644
--- a/src/core/lib/client_config/subchannel.c
+++ b/src/core/lib/client_config/subchannel.c
@@ -54,7 +54,7 @@
 #define STRONG_REF_MASK (~(gpr_atm)((1 << INTERNAL_REF_BITS) - 1))
 
 #define GRPC_SUBCHANNEL_MIN_CONNECT_TIMEOUT_SECONDS 20
-#define GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS 1
+#define GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS 2
 #define GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER 1.6
 #define GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS 120
 #define GRPC_SUBCHANNEL_RECONNECT_JITTER 0.2
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index 37cc724b53..ba94852f9a 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -81,11 +81,11 @@ typedef enum {
   /* Status came from the application layer overriding whatever
      the wire says */
   STATUS_FROM_API_OVERRIDE = 0,
-  /* Status was created by some internal channel stack operation */
-  STATUS_FROM_CORE,
   /* Status came from 'the wire' - or somewhere below the surface
      layer */
   STATUS_FROM_WIRE,
+  /* Status was created by some internal channel stack operation */
+  STATUS_FROM_CORE,
   /* Status came from the server sending status */
   STATUS_FROM_SERVER_STATUS,
   STATUS_SOURCE_COUNT
@@ -1110,6 +1110,9 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, bool success) {
 
   gpr_mu_lock(&call->mu);
   if (bctl->send_initial_metadata) {
+    if (!success) {
+      set_status_code(call, STATUS_FROM_CORE, GRPC_STATUS_UNAVAILABLE);
+    }
     grpc_metadata_batch_destroy(
         &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]);
   }
@@ -1232,8 +1235,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
         call->metadata_batch[0][0].deadline = call->send_deadline;
         stream_op.send_initial_metadata =
             &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */];
-        stream_op.send_idempotent_request =
-            (op->flags & GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) != 0;
+        stream_op.send_initial_metadata_flags = op->flags;
         break;
       case GRPC_OP_SEND_MESSAGE:
         if (!are_write_flags_valid(op->flags)) {
diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h
index 460e4dcedc..bb1120ee3a 100644
--- a/src/core/lib/transport/transport.h
+++ b/src/core/lib/transport/transport.h
@@ -101,9 +101,9 @@ typedef struct grpc_transport_stream_op {
   /** Send initial metadata to the peer, from the provided metadata batch.
       idempotent_request MUST be set if this is non-null */
   grpc_metadata_batch *send_initial_metadata;
-  /** Iff send_initial_metadata != NULL, flags if this is an idempotent request
-      or not */
-  bool send_idempotent_request;
+  /** Iff send_initial_metadata != NULL, flags associated with
+      send_initial_metadata: a bitfield of GRPC_INITIAL_METADATA_xxx */
+  uint32_t send_initial_metadata_flags;
 
   /** Send trailing metadata to the peer, from the provided metadata batch. */
   grpc_metadata_batch *send_trailing_metadata;