diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc
index ded57cc..65a0c7e 100644
--- a/http2/adapter/nghttp2_adapter_test.cc
+++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -24,6 +24,24 @@
   WINDOW_UPDATE,
 };
 
+// This send callback assumes |source|'s pointer is a TestDataSource, and
+// |user_data| is a Http2VisitorInterface.
+int TestSendCallback(nghttp2_session*,
+                     nghttp2_frame* frame,
+                     const uint8_t* framehd,
+                     size_t length,
+                     nghttp2_data_source* source,
+                     void* user_data) {
+  auto* visitor = static_cast<Http2VisitorInterface*>(user_data);
+  // Send the frame header via the visitor.
+  visitor->OnReadyToSend(ToStringView(framehd, 9));
+  auto* test_source = static_cast<TestDataSource*>(source->ptr);
+  absl::string_view payload = test_source->ReadNext(length);
+  // Send the frame payload via the visitor.
+  visitor->OnReadyToSend(payload);
+  return 0;
+}
+
 TEST(NgHttp2AdapterTest, ClientConstruction) {
   testing::StrictMock<MockHttp2Visitor> visitor;
   auto adapter = NgHttp2Adapter::CreateClientAdapter(visitor);
@@ -269,22 +287,8 @@
   TestDataSource body1{kBody};
   // The TestDataSource is wrapped in the nghttp2_data_provider data type.
   nghttp2_data_provider provider = body1.MakeDataProvider();
+  nghttp2_send_data_callback send_callback = &TestSendCallback;
 
-  // This send callback assumes |source|'s pointer is a TestDataSource, which we
-  // know is true because we just converted |body1| into a nghttp2_data_provider
-  // above.
-  nghttp2_send_data_callback send_callback =
-      [](nghttp2_session*, nghttp2_frame* frame, const uint8_t* framehd,
-         size_t length, nghttp2_data_source* source, void* user_data) {
-        auto* visitor = static_cast<Http2VisitorInterface*>(user_data);
-        // Send the frame header via the visitor.
-        visitor->OnReadyToSend(ToStringView(framehd, 9));
-        auto* test_source = static_cast<TestDataSource*>(source->ptr);
-        absl::string_view payload = test_source->ReadNext(length);
-        // Send the frame payload via the visitor.
-        visitor->OnReadyToSend(payload);
-        return 0;
-      };
   // This call transforms it back into a DataFrameSource, which is compatible
   // with the Http2Adapter API.
   std::unique_ptr<DataFrameSource> frame_source =
@@ -304,6 +308,53 @@
   EXPECT_FALSE(adapter->session().want_write());
 }
 
+// This test verifies how nghttp2 behaves when a data source becomes
+// read-blocked.
+TEST(NgHttp2AdapterTest, ClientSubmitRequestWithDataProviderAndReadBlock) {
+  DataSavingVisitor visitor;
+  auto adapter = NgHttp2Adapter::CreateClientAdapter(visitor);
+
+  const absl::string_view kBody = "This is an example request body.";
+  // This test will use TestDataSource as the source of the body payload data.
+  TestDataSource body1{kBody};
+  body1.set_is_data_available(false);
+  // The TestDataSource is wrapped in the nghttp2_data_provider data type.
+  nghttp2_data_provider provider = body1.MakeDataProvider();
+  nghttp2_send_data_callback send_callback = &TestSendCallback;
+
+  // This call transforms it back into a DataFrameSource, which is compatible
+  // with the Http2Adapter API.
+  std::unique_ptr<DataFrameSource> frame_source =
+      MakeZeroCopyDataFrameSource(provider, &visitor, std::move(send_callback));
+  int stream_id =
+      adapter->SubmitRequest(ToHeaders({{":method", "POST"},
+                                        {":scheme", "http"},
+                                        {":authority", "example.com"},
+                                        {":path", "/this/is/request/one"}}),
+                             frame_source.get(), nullptr);
+  EXPECT_GT(stream_id, 0);
+  EXPECT_TRUE(adapter->session().want_write());
+
+  adapter->Send();
+  // Client preface does not appear to include the mandatory SETTINGS frame.
+  absl::string_view serialized = visitor.data();
+  EXPECT_THAT(serialized,
+              testing::StartsWith(spdy::kHttp2ConnectionHeaderPrefix));
+  serialized.remove_prefix(strlen(spdy::kHttp2ConnectionHeaderPrefix));
+  EXPECT_THAT(serialized, EqualsFrames({spdy::SpdyFrameType::HEADERS}));
+  visitor.Clear();
+  EXPECT_FALSE(adapter->session().want_write());
+
+  // Resume the deferred stream.
+  body1.set_is_data_available(true);
+  nghttp2_session_resume_data(adapter->session().raw_ptr(), stream_id);
+  EXPECT_TRUE(adapter->session().want_write());
+
+  adapter->Send();
+  EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::DATA}));
+  EXPECT_FALSE(adapter->session().want_write());
+}
+
 TEST(NgHttp2AdapterTest, ServerConstruction) {
   testing::StrictMock<MockHttp2Visitor> visitor;
   auto adapter = NgHttp2Adapter::CreateServerAdapter(visitor);
diff --git a/http2/adapter/nghttp2_data_provider.cc b/http2/adapter/nghttp2_data_provider.cc
index 7a70965..83037e4 100644
--- a/http2/adapter/nghttp2_data_provider.cc
+++ b/http2/adapter/nghttp2_data_provider.cc
@@ -18,6 +18,7 @@
                                     uint32_t* data_flags,
                                     nghttp2_data_source* source,
                                     void* /* user_data */) {
+  *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
   auto* frame_source = static_cast<DataFrameSource*>(source->ptr);
   auto [result_length, done] = frame_source->SelectPayloadLength(length);
   if (result_length == DataFrameSource::kBlocked) {
@@ -25,7 +26,6 @@
   } else if (result_length == DataFrameSource::kError) {
     return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
   }
-  *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
   if (done) {
     *data_flags |= NGHTTP2_DATA_FLAG_EOF;
   }
diff --git a/http2/adapter/nghttp2_util.cc b/http2/adapter/nghttp2_util.cc
index d97c64a..812bbec 100644
--- a/http2/adapter/nghttp2_util.cc
+++ b/http2/adapter/nghttp2_util.cc
@@ -133,11 +133,13 @@
     ssize_t result = provider_.read_callback(
         nullptr /* session */, stream_id, nullptr /* buf */, max_length,
         &data_flags, &provider_.source, nullptr /* user_data */);
-    if (result < 0) {
-      return {-1, false};
+    if (result == NGHTTP2_ERR_DEFERRED) {
+      return {kBlocked, false};
+    } else if (result < 0) {
+      return {kError, false};
     } else if ((data_flags & NGHTTP2_DATA_FLAG_NO_COPY) == 0) {
       QUICHE_LOG(ERROR) << "Source did not use the zero-copy API!";
-      return {-1, false};
+      return {kError, false};
     } else {
       if (data_flags & NGHTTP2_DATA_FLAG_NO_END_STREAM) {
         send_fin_ = false;
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc
index 330476c..8c10656 100644
--- a/http2/adapter/oghttp2_session.cc
+++ b/http2/adapter/oghttp2_session.cc
@@ -127,7 +127,7 @@
     const Http2StreamId stream_id = write_scheduler_.PopNextReadyStream();
     // TODO(birenroy): Add a return value to indicate write blockage, so streams
     // aren't woken unnecessarily.
-    WriteForStream(stream_id);
+    continue_writing = WriteForStream(stream_id);
   }
   if (continue_writing) {
     SendQueuedFrames();
@@ -152,12 +152,12 @@
   return true;
 }
 
-void OgHttp2Session::WriteForStream(Http2StreamId stream_id) {
+bool OgHttp2Session::WriteForStream(Http2StreamId stream_id) {
   auto it = stream_map_.find(stream_id);
   if (it == stream_map_.end()) {
     QUICHE_LOG(ERROR) << "Can't find stream " << stream_id
                       << " which is ready to write!";
-    return;
+    return true;
   }
   StreamState& state = it->second;
   if (state.outbound_body == nullptr) {
@@ -171,13 +171,22 @@
         MaybeCloseWithRstStream(stream_id, state);
       }
     }
-    return;
+    return true;
   }
+  bool source_can_produce = true;
   int32_t available_window =
       std::min(std::min(peer_window_, state.send_window), max_frame_payload_);
   while (available_window > 0 && state.outbound_body != nullptr) {
     auto [length, end_data] =
         state.outbound_body->SelectPayloadLength(available_window);
+    if (length == DataFrameSource::kBlocked) {
+      source_can_produce = false;
+      break;
+    } else if (length == DataFrameSource::kError) {
+      source_can_produce = false;
+      visitor_.OnCloseStream(stream_id, Http2ErrorCode::INTERNAL_ERROR);
+      break;
+    }
     const bool fin = end_data ? state.outbound_body->send_fin() : false;
     spdy::SpdyDataIR data(stream_id);
     data.set_fin(fin);
@@ -208,9 +217,11 @@
   }
   // If the stream still has data to send, it should be marked as ready in the
   // write scheduler.
-  if (state.send_window > 0 && state.outbound_body != nullptr) {
+  if (source_can_produce && state.send_window > 0 &&
+      state.outbound_body != nullptr) {
     write_scheduler_.MarkStreamReady(stream_id, false);
   }
+  return available_window > 0;
 }
 
 int32_t OgHttp2Session::SubmitRequest(absl::Span<const Header> headers,
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h
index 30bc9d8..c5b9efc 100644
--- a/http2/adapter/oghttp2_session.h
+++ b/http2/adapter/oghttp2_session.h
@@ -160,7 +160,10 @@
   // Sends queued frames, returning true if all frames were flushed.
   bool SendQueuedFrames();
 
-  void WriteForStream(Http2StreamId stream_id);
+  // Returns false if the connection is write-blocked (due to flow control or
+  // some other reason).
+  bool WriteForStream(Http2StreamId stream_id);
+
   void SendTrailers(Http2StreamId stream_id, spdy::SpdyHeaderBlock trailers);
 
   // Encapsulates the RST_STREAM NO_ERROR behavior described in RFC 7540
diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc
index 185fd3a..636fda1 100644
--- a/http2/adapter/oghttp2_session_test.cc
+++ b/http2/adapter/oghttp2_session_test.cc
@@ -221,6 +221,41 @@
   EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::HEADERS}));
 }
 
+// This test exercises the case where the client request body source is read
+// blocked.
+TEST(OgHttp2SessionTest, ClientSubmitRequestWithReadBlock) {
+  DataSavingVisitor visitor;
+  OgHttp2Session session(
+      visitor, OgHttp2Session::Options{.perspective = Perspective::kClient});
+  EXPECT_FALSE(session.want_write());
+
+  const char* kSentinel1 = "arbitrary pointer 1";
+  TestDataFrameSource body1(visitor, "This is an example request body.");
+  body1.set_is_data_available(false);
+  int stream_id =
+      session.SubmitRequest(ToHeaders({{":method", "POST"},
+                                       {":scheme", "http"},
+                                       {":authority", "example.com"},
+                                       {":path", "/this/is/request/one"}}),
+                            &body1, const_cast<char*>(kSentinel1));
+  EXPECT_GT(stream_id, 0);
+  EXPECT_TRUE(session.want_write());
+  EXPECT_EQ(kSentinel1, session.GetStreamUserData(stream_id));
+  session.Send();
+  absl::string_view serialized = visitor.data();
+  EXPECT_THAT(serialized,
+              testing::StartsWith(spdy::kHttp2ConnectionHeaderPrefix));
+  serialized.remove_prefix(strlen(spdy::kHttp2ConnectionHeaderPrefix));
+  EXPECT_THAT(serialized,
+              EqualsFrames({SpdyFrameType::SETTINGS, SpdyFrameType::HEADERS}));
+  // No data frame, as body1 was read blocked.
+  visitor.Clear();
+  EXPECT_FALSE(session.want_write());
+
+  // Currently there is no way to indicate that the first stream is no longer
+  // read blocked.
+}
+
 TEST(OgHttp2SessionTest, ClientStartShutdown) {
   DataSavingVisitor visitor;
   OgHttp2Session session(
diff --git a/http2/adapter/test_utils.cc b/http2/adapter/test_utils.cc
index 3ccd63e..c6e98a2 100644
--- a/http2/adapter/test_utils.cc
+++ b/http2/adapter/test_utils.cc
@@ -36,13 +36,16 @@
 
 std::pair<ssize_t, bool> TestDataFrameSource::SelectPayloadLength(
     size_t max_length) {
+  if (!is_data_available_) {
+    return {kBlocked, false};
+  }
   // The stream is done if there's no more data, or if |max_length| is at least
   // as large as the remaining data.
   const bool end_data =
       current_fragment_.empty() || (payload_fragments_.size() == 1 &&
                                     max_length >= current_fragment_.size());
   const ssize_t length = std::min(max_length, current_fragment_.size());
-  return std::make_pair(length, end_data);
+  return {length, end_data};
 }
 
 void TestDataFrameSource::Send(absl::string_view frame_header,
diff --git a/http2/adapter/test_utils.h b/http2/adapter/test_utils.h
index 0ea7ec4..de001a1 100644
--- a/http2/adapter/test_utils.h
+++ b/http2/adapter/test_utils.h
@@ -50,11 +50,14 @@
   void Send(absl::string_view frame_header, size_t payload_length) override;
   bool send_fin() const override { return has_fin_; }
 
+  void set_is_data_available(bool value) { is_data_available_ = value; }
+
  private:
   Http2VisitorInterface& visitor_;
   std::vector<std::string> payload_fragments_;
   absl::string_view current_fragment_;
   const bool has_fin_;
+  bool is_data_available_ = true;
 };
 
 // A simple class that can easily be adapted to act as a nghttp2_data_source.
@@ -78,20 +81,27 @@
         .source = {.ptr = this},
         .read_callback = [](nghttp2_session*, int32_t, uint8_t*, size_t length,
                             uint32_t* data_flags, nghttp2_data_source* source,
-                            void*) {
+                            void*) -> ssize_t {
+          *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
           auto* s = static_cast<TestDataSource*>(source->ptr);
+          if (!s->is_data_available()) {
+            return NGHTTP2_ERR_DEFERRED;
+          }
           const ssize_t ret = s->SelectPayloadLength(length);
           if (ret < length) {
             *data_flags |= NGHTTP2_DATA_FLAG_EOF;
           }
-          *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
           return ret;
         }};
   }
 
+  bool is_data_available() const { return is_data_available_; }
+  void set_is_data_available(bool value) { is_data_available_ = value; }
+
  private:
   const std::string data_;
   absl::string_view remaining_ = data_;
+  bool is_data_available_ = true;
 };
 
 // These matchers check whether a string consists entirely of HTTP/2 frames of