Fix bug where pushed SPDY streams were never actually closed.

The fix uncovers a few things.  Previously, when we had an unclaimed
push stream, we'd leave it as an "active" stream, even if the EOF had
already been received on that stream.  I changed it so that the push
stream is removed from the active stream as soon as it is inactive,
but it remains on the unclaimed_pushed_streams list.  This seems more
correct.

The hardest part of the test was getting the test to verify the fix.
To verify the fix, I modified the Push tests so that we leave the
session open (e.g. terminate the list of reads with an ERR_IO_PENDING
rather than an EOF so that it sits open), which gives us the opportunity
to check if there are active streams left on the session when we 
completed the test.  If so, then we'll pop an error.

BUG=52898
TEST=Updated existing tests.


Review URL: http://codereview.chromium.org/3108038

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@57023 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 271 insertions, 137 deletions

diff --git a/net/spdy/spdy_http_stream.cc b/net/spdy/spdy_http_stream.cc
index 41b18aa..622d2c0 100644
--- a/net/spdy/spdy_http_stream.cc
+++ b/net/spdy/spdy_http_stream.cc
@@ -391,7 +391,8 @@ void SpdyHttpStream::OnDataReceived(const char* data, int length) {
   // Note that data may be received for a SpdyStream prior to the user calling
   // ReadResponseBody(), therefore user_buffer_ may be NULL.  This may often
   // happen for server initiated streams.
-  if (length > 0 && !stream_->closed()) {
+  DCHECK(!stream_->closed() || stream_->pushed());
+  if (length > 0) {
     // Save the received data.
     IOBufferWithSize* io_buffer = new IOBufferWithSize(length);
     memcpy(io_buffer->data(), data, length);
diff --git a/net/spdy/spdy_network_transaction_unittest.cc b/net/spdy/spdy_network_transaction_unittest.cc
index 6c0973e..49f41ed 100644
--- a/net/spdy/spdy_network_transaction_unittest.cc
+++ b/net/spdy/spdy_network_transaction_unittest.cc
@@ -10,6 +10,7 @@
 #include "net/http/http_transaction_unittest.h"
 #include "net/spdy/spdy_http_stream.h"
 #include "net/spdy/spdy_session.h"
+#include "net/spdy/spdy_session_pool.h"
 #include "net/spdy/spdy_test_util.h"
 #include "net/url_request/url_request_unittest.h"
 #include "testing/platform_test.h"
@@ -82,6 +83,18 @@ class SpdyNetworkTransactionTest
             }
           }
 
+    ~NormalSpdyTransactionHelper() {
+      // Any test which doesn't close the socket by sending it an EOF will
+      // have a valid session left open, which leaks the entire session pool.
+      // This is just fine - in fact, some of our tests intentionally do this
+      // so that we can check consistency of the SpdySessionPool as the test
+      // finishes.  If we had put an EOF on the socket, the SpdySession would
+      // have closed and we wouldn't be able to check the consistency.
+
+      // Forcefully close existing sessions here.
+      session()->spdy_session_pool()->CloseAllSessions();
+    }
+
     void SetDeterministic() {
       session_ = SpdySessionDependencies::SpdyCreateSessionDeterministic(
           session_deps_.get());
@@ -247,12 +260,15 @@ class SpdyNetworkTransactionTest
     HttpNetworkTransaction* trans() { return trans_.get(); }
     void ResetTrans() { trans_.reset(); }
     TransactionHelperResult& output() { return output_; }
-    HttpRequestInfo& request() { return request_; }
-    scoped_refptr<HttpNetworkSession>& session() { return session_; }
+    const HttpRequestInfo& request() const { return request_; }
+    const scoped_refptr<HttpNetworkSession>& session() const {
+      return session_;
+    }
     scoped_ptr<SpdySessionDependencies>& session_deps() {
       return session_deps_;
     }
-    int port() { return port_; }
+    int port() const { return port_; }
+    SpdyNetworkTransactionTestTypes test_type() const { return test_type_; }
 
    private:
     typedef std::vector<StaticSocketDataProvider*> DataVector;
@@ -307,28 +323,54 @@ class SpdyNetworkTransactionTest
     return google_post_request_;
   }
 
-  class RunServerPushTestCallback : public CallbackRunner< Tuple1<int> > {
-   public:
-    RunServerPushTestCallback(scoped_refptr<net::IOBufferWithSize> buffer,
-                              std::string& result, bool& need_read_callback)
-        : buffer_(buffer), result_(result),
-        need_read_callback_(need_read_callback) {}
-
-    virtual void RunWithParams(const Tuple1<int>& params) {
-      // Indicates some type of error.
-      if(params.a <= 0)
-        return;
+  // Read the result of a particular transaction, knowing that we've got
+  // multiple transactions in the read pipeline; so as we read, we may have
+  // to skip over data destined for other transactions while we consume
+  // the data for |trans|.
+  int ReadResult(HttpNetworkTransaction* trans,
+                 OrderedSocketData* data,
+                 std::string* result) {
+    const int kSize = 3000;
 
-      std::string temp(buffer_->data(), params.a);
-      result_.append(temp);
-      need_read_callback_ = true;
+    int bytes_read = 0;
+    scoped_refptr<net::IOBufferWithSize> buf = new net::IOBufferWithSize(kSize);
+    TestCompletionCallback callback;
+    while (true) {
+      int rv = trans->Read(buf, kSize, &callback);
+      if (rv == ERR_IO_PENDING) {
+        // Multiple transactions may be in the data set.  Keep pulling off
+        // reads until we complete our callback.
+        while (!callback.have_result()) {
+          data->CompleteRead();
+          MessageLoop::current()->RunAllPending();
+        }
+        rv = callback.WaitForResult();
+      } else if (rv <= 0) {
+        break;
+      }
+      result->append(buf->data(), rv);
+      bytes_read += rv;
     }
+    return bytes_read;
+  }
 
-   private:
-    scoped_refptr<net::IOBufferWithSize> buffer_;
-    std::string& result_;
-    bool need_read_callback_;
-  };
+  void VerifyStreamsClosed(const NormalSpdyTransactionHelper& helper) {
+    // This lengthy block is reaching into the pool to dig out the active
+    // session.  Once we have the session, we verify that the streams are
+    // all closed and not leaked at this point.
+    const GURL& url = helper.request().url;
+    int port = helper.test_type() == SPDYNPN ? 443 : 80;
+    HostPortPair host_port_pair(url.host(), port);
+    HostPortProxyPair pair(host_port_pair, "DIRECT");
+    BoundNetLog log;
+    const scoped_refptr<HttpNetworkSession>& session = helper.session();
+    scoped_refptr<SpdySessionPool> pool(session->spdy_session_pool());
+    EXPECT_TRUE(pool->HasSession(pair));
+    scoped_refptr<SpdySession> spdy_session(pool->Get(pair, session, log));
+    ASSERT_TRUE(spdy_session.get() != NULL);
+    EXPECT_EQ(0u, spdy_session->num_active_streams());
+    EXPECT_EQ(0u, spdy_session->num_unclaimed_pushed_streams());
+  }
 
   void RunServerPushTest(MockWrite writes[], int writes_length,
                          MockRead reads[], int reads_length,
@@ -353,8 +395,6 @@ class SpdyNetworkTransactionTest
     rv = callback.WaitForResult();
 
     // Request the pushed path.
-    const int kSize = 3000;
-    scoped_refptr<net::IOBufferWithSize> buf = new net::IOBufferWithSize(kSize);
     scoped_ptr<HttpNetworkTransaction> trans2(
         new HttpNetworkTransaction(helper.session()));
     rv = trans2->Start(&CreateGetPushRequest(), &callback, BoundNetLog());
@@ -363,37 +403,30 @@ class SpdyNetworkTransactionTest
 
     // The data for the pushed path may be coming in more than 1 packet. Compile
     // the results into a single string.
+
+    // Read the server push body.
+    std::string result2;
+    ReadResult(trans2.get(), data, &result2);
+    // Read the response body.
     std::string result;
-    bool need_read_callback = true;
-    RunServerPushTestCallback callback2(buf, result, need_read_callback);
-    while(!data->at_read_eof())
-    {
-      if(need_read_callback) {
-        while((rv = trans2->Read(buf, kSize, &callback2)) > 0) {
-          std::string result1(buf->data(),rv);
-          result.append(result1);
-        }
-        need_read_callback = false;
-      }
-      else
-        data->CompleteRead();
-      MessageLoop::current()->RunAllPending();
-    }
+    ReadResult(trans, data, &result);
 
     // Verify that we consumed all test data.
     EXPECT_TRUE(data->at_read_eof());
     EXPECT_TRUE(data->at_write_eof());
 
     // Verify that the received push data is same as the expected push data.
-    EXPECT_EQ(result.compare(expected),0) << "Received data: "
-                                          << result
-                                          << "||||| Expected data: "
-                                          << expected;
+    EXPECT_EQ(result2.compare(expected), 0) << "Received data: "
+                                            << result2
+                                            << "||||| Expected data: "
+                                            << expected;
 
     // Verify the SYN_REPLY.
     // Copy the response info, because trans goes away.
     *response = *trans->GetResponseInfo();
     *response2 = *trans2->GetResponseInfo();
+
+    VerifyStreamsClosed(helper);
   }
 
  private:
@@ -1766,7 +1799,7 @@ class SpdyNetworkTransactionTest::StartTransactionCallback
     : public CallbackRunner< Tuple1<int> > {
  public:
   explicit StartTransactionCallback(
-      scoped_refptr<HttpNetworkSession>& session,
+      const scoped_refptr<HttpNetworkSession>& session,
       NormalSpdyTransactionHelper& helper)
       : session_(session), helper_(helper) {}
 
@@ -1785,7 +1818,7 @@ class SpdyNetworkTransactionTest::StartTransactionCallback
   }
 
  private:
-  scoped_refptr<HttpNetworkSession>& session_;
+  const scoped_refptr<HttpNetworkSession>& session_;
   NormalSpdyTransactionHelper& helper_;
 };
 
@@ -2156,29 +2189,33 @@ TEST_P(SpdyNetworkTransactionTest, RedirectServerPush) {
 TEST_P(SpdyNetworkTransactionTest, ServerPushSingleDataFrame) {
   static const unsigned char kPushBodyFrame[] = {
     0x00, 0x00, 0x00, 0x02,                                      // header, ID
-    0x01, 0x00, 0x00, 0x05,                                      // FIN, length
-    'h', 'e', 'l', 'l', 'o',                                     // "hello"
+    0x01, 0x00, 0x00, 0x06,                                      // FIN, length
+    'p', 'u', 's', 'h', 'e', 'd'                                 // "pushed"
   };
-
-  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_syn(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_body(ConstructSpdyBodyFrame(1, true));
   MockWrite writes[] = {
-    CreateMockWrite(*req, 1),
+    CreateMockWrite(*stream1_syn, 1),
   };
 
-  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
-  scoped_ptr<spdy::SpdyFrame> rep(ConstructSpdyPush(NULL, 0, 2, 1, "/foo.dat"));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_reply(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_syn(ConstructSpdyPush(NULL, 0, 2, 1, "/foo.dat"));
   MockRead reads[] = {
-    CreateMockRead(*resp, 2),
-    CreateMockRead(*rep, 3),
+    CreateMockRead(*stream1_reply, 2),
+    CreateMockRead(*stream2_syn, 3),
+    CreateMockRead(*stream1_body, 4, false),
     MockRead(true, reinterpret_cast<const char*>(kPushBodyFrame),
-             arraysize(kPushBodyFrame), 4),
-    MockRead(true, ERR_IO_PENDING, 5),  // Force a pause
-    MockRead(true, 0, 0, 6)  // EOF
+             arraysize(kPushBodyFrame), 5),
+    MockRead(true, ERR_IO_PENDING, 6),  // Force a pause
   };
 
   HttpResponseInfo response;
   HttpResponseInfo response2;
-  std::string expected_push_result("hello");
+  std::string expected_push_result("pushed");
   RunServerPushTest(writes, arraysize(writes), reads, arraysize(reads),
                     &response, &response2, expected_push_result);
 
@@ -2191,27 +2228,88 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushSingleDataFrame) {
   EXPECT_EQ("HTTP/1.1 200 OK", response2.headers->GetStatusLine());
 }
 
+TEST_P(SpdyNetworkTransactionTest, ServerPushServerAborted) {
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_syn(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_body(ConstructSpdyBodyFrame(1, true));
+  MockWrite writes[] = {
+    CreateMockWrite(*stream1_syn, 1),
+  };
+
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_reply(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_syn(ConstructSpdyPush(NULL, 0, 2, 1, "/foo.dat"));
+  scoped_ptr<spdy::SpdyFrame>
+    stream2_rst(ConstructSpdyRstStream(2, spdy::PROTOCOL_ERROR));
+  MockRead reads[] = {
+    CreateMockRead(*stream1_reply, 2),
+    CreateMockRead(*stream2_syn, 3),
+    CreateMockRead(*stream2_rst, 4),
+    CreateMockRead(*stream1_body, 5, false),
+    MockRead(true, ERR_IO_PENDING, 6),  // Force a pause
+  };
+
+  scoped_refptr<OrderedSocketData> data(
+      new OrderedSocketData(reads, arraysize(reads),
+                            writes, arraysize(writes)));
+  NormalSpdyTransactionHelper helper(CreateGetRequest(),
+                                     BoundNetLog(), GetParam());
+
+  helper.RunPreTestSetup();
+  helper.AddData(data.get());
+
+  HttpNetworkTransaction* trans = helper.trans();
+
+  // Start the transaction with basic parameters.
+  TestCompletionCallback callback;
+  int rv = trans->Start(&CreateGetRequest(), &callback, BoundNetLog());
+  EXPECT_EQ(ERR_IO_PENDING, rv);
+  rv = callback.WaitForResult();
+  EXPECT_EQ(OK, rv);
+
+  // Verify that we consumed all test data.
+  EXPECT_TRUE(data->at_read_eof()) << "Read count: "
+                                   << data->read_count()
+                                   << " Read index: "
+                                   << data->read_index();
+  EXPECT_TRUE(data->at_write_eof()) << "Write count: "
+                                    << data->write_count()
+                                    << " Write index: "
+                                    << data->write_index();
+
+  // Verify the SYN_REPLY.
+  HttpResponseInfo response = *trans->GetResponseInfo();
+  EXPECT_TRUE(response.headers != NULL);
+  EXPECT_EQ("HTTP/1.1 200 OK", response.headers->GetStatusLine());
+}
+
 TEST_P(SpdyNetworkTransactionTest, ServerPushMultipleDataFrame) {
   static const unsigned char kPushBodyFrame1[] = {
     0x00, 0x00, 0x00, 0x02,                                      // header, ID
-    0x01, 0x00, 0x00, 0x1E,                                      // FIN, length
-    'h', 'e', 'l', 'l', 'o',                                     // "hello"
+    0x01, 0x00, 0x00, 0x1F,                                      // FIN, length
+    'p', 'u', 's', 'h', 'e', 'd'                                 // "pushed"
   };
   static const char kPushBodyFrame2[] = " my darling";
   static const char kPushBodyFrame3[] = " hello";
   static const char kPushBodyFrame4[] = " my baby";
 
-  scoped_ptr<spdy::SpdyFrame> req(
-      ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_syn(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_body(ConstructSpdyBodyFrame(1, true));
   MockWrite writes[] = {
-    CreateMockWrite(*req, 1),
+    CreateMockWrite(*stream1_syn, 1),
   };
 
-  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
-  scoped_ptr<spdy::SpdyFrame> rep(ConstructSpdyPush(NULL, 0, 2, 1, "/foo.dat"));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_reply(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_syn(ConstructSpdyPush(NULL, 0, 2, 1, "/foo.dat"));
   MockRead reads[] = {
-    CreateMockRead(*resp, 2),
-    CreateMockRead(*rep, 3),
+    CreateMockRead(*stream1_reply, 2),
+    CreateMockRead(*stream2_syn, 3),
     MockRead(true, reinterpret_cast<const char*>(kPushBodyFrame1),
              arraysize(kPushBodyFrame1), 4),
     MockRead(true, reinterpret_cast<const char*>(kPushBodyFrame2),
@@ -2220,13 +2318,13 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushMultipleDataFrame) {
              arraysize(kPushBodyFrame3) - 1, 6),
     MockRead(true, reinterpret_cast<const char*>(kPushBodyFrame4),
              arraysize(kPushBodyFrame4) - 1, 7),
-    MockRead(true, ERR_IO_PENDING, 8),  // Force a pause
-    MockRead(true, 0, 0, 9)  // EOF
+    CreateMockRead(*stream1_body, 8, false),
+    MockRead(true, ERR_IO_PENDING, 9),  // Force a pause
   };
 
   HttpResponseInfo response;
   HttpResponseInfo response2;
-  std::string expected_push_result("hello my darling hello my baby");
+  std::string expected_push_result("pushed my darling hello my baby");
   RunServerPushTest(writes, arraysize(writes), reads, arraysize(reads),
                     &response, &response2, expected_push_result);
 
@@ -2242,24 +2340,28 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushMultipleDataFrame) {
 TEST_P(SpdyNetworkTransactionTest, ServerPushMultipleDataFrameInterrupted) {
   static const unsigned char kPushBodyFrame1[] = {
     0x00, 0x00, 0x00, 0x02,                                      // header, ID
-    0x01, 0x00, 0x00, 0x1E,                                      // FIN, length
-    'h', 'e', 'l', 'l', 'o',                                     // "hello"
+    0x01, 0x00, 0x00, 0x1F,                                      // FIN, length
+    'p', 'u', 's', 'h', 'e', 'd'                                 // "pushed"
   };
   static const char kPushBodyFrame2[] = " my darling";
   static const char kPushBodyFrame3[] = " hello";
   static const char kPushBodyFrame4[] = " my baby";
 
-  scoped_ptr<spdy::SpdyFrame> req(
-      ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_syn(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_body(ConstructSpdyBodyFrame(1, true));
   MockWrite writes[] = {
-    CreateMockWrite(*req, 1),
+    CreateMockWrite(*stream1_syn, 1),
   };
 
-  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
-  scoped_ptr<spdy::SpdyFrame> rep(ConstructSpdyPush(NULL, 0, 2, 1, "/foo.dat"));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_reply(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_syn(ConstructSpdyPush(NULL, 0, 2, 1, "/foo.dat"));
   MockRead reads[] = {
-    CreateMockRead(*resp, 2),
-    CreateMockRead(*rep, 3),
+    CreateMockRead(*stream1_reply, 2),
+    CreateMockRead(*stream2_syn, 3),
     MockRead(true, reinterpret_cast<const char*>(kPushBodyFrame1),
              arraysize(kPushBodyFrame1), 4),
     MockRead(true, reinterpret_cast<const char*>(kPushBodyFrame2),
@@ -2269,12 +2371,13 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushMultipleDataFrameInterrupted) {
              arraysize(kPushBodyFrame3) - 1, 7),
     MockRead(true, reinterpret_cast<const char*>(kPushBodyFrame4),
              arraysize(kPushBodyFrame4) - 1, 8),
-    MockRead(true, 0, 0, 9)  // EOF
+    CreateMockRead(*stream1_body.get(), 9, false),
+    MockRead(true, ERR_IO_PENDING, 10)  // Force a pause.
   };
 
   HttpResponseInfo response;
   HttpResponseInfo response2;
-  std::string expected_push_result("hello my darling hello my baby");
+  std::string expected_push_result("pushed my darling hello my baby");
   RunServerPushTest(writes, arraysize(writes), reads, arraysize(reads),
                     &response, &response2, expected_push_result);
 
@@ -2288,20 +2391,26 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushMultipleDataFrameInterrupted) {
 }
 
 TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID0) {
-  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
-  scoped_ptr<spdy::SpdyFrame> res(
-      ConstructSpdyRstStream(2, spdy::INVALID_STREAM));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_syn(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_body(ConstructSpdyBodyFrame(1, true));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_rst(ConstructSpdyRstStream(2, spdy::INVALID_STREAM));
   MockWrite writes[] = {
-    CreateMockWrite(*req, 1),
-    CreateMockWrite(*res, 4),
+    CreateMockWrite(*stream1_syn, 1),
+    CreateMockWrite(*stream2_rst, 4),
   };
 
-  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
-  scoped_ptr<spdy::SpdyFrame> rep(ConstructSpdyPush(NULL, 0, 2, 0, "/foo.dat"));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_reply(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_syn(ConstructSpdyPush(NULL, 0, 2, 0, "/foo.dat"));
   MockRead reads[] = {
-    CreateMockRead(*resp, 2),
-    CreateMockRead(*rep, 3),
-    MockRead(true, 0, 0, 5)  // EOF
+    CreateMockRead(*stream1_reply, 2),
+    CreateMockRead(*stream2_syn, 3),
+    CreateMockRead(*stream1_body, 4),
+    MockRead(true, ERR_IO_PENDING, 5)  // Force a pause
   };
 
   scoped_refptr<OrderedSocketData> data(
@@ -2320,7 +2429,6 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID0) {
   int rv = trans->Start(&CreateGetRequest(), &callback, BoundNetLog());
   EXPECT_EQ(ERR_IO_PENDING, rv);
   rv = callback.WaitForResult();
-  data->CompleteRead();
   EXPECT_EQ(OK, rv);
 
   // Verify that we consumed all test data.
@@ -2340,20 +2448,26 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID0) {
 }
 
 TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID9) {
-  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
-  scoped_ptr<spdy::SpdyFrame> res(
-      ConstructSpdyRstStream(2, spdy::INVALID_ASSOCIATED_STREAM));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_syn(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_body(ConstructSpdyBodyFrame(1, true));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_rst(ConstructSpdyRstStream(2, spdy::INVALID_ASSOCIATED_STREAM));
   MockWrite writes[] = {
-    CreateMockWrite(*req, 1),
-    CreateMockWrite(*res, 4),
+    CreateMockWrite(*stream1_syn, 1),
+    CreateMockWrite(*stream2_rst, 4),
   };
 
-  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
-  scoped_ptr<spdy::SpdyFrame> rep(ConstructSpdyPush(NULL, 0, 2, 9, "/foo.dat"));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_reply(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_syn(ConstructSpdyPush(NULL, 0, 2, 9, "/foo.dat"));
   MockRead reads[] = {
-    CreateMockRead(*resp, 2),
-    CreateMockRead(*rep, 3),
-    MockRead(true, 0, 0, 5)  // EOF
+    CreateMockRead(*stream1_reply, 2),
+    CreateMockRead(*stream2_syn, 3),
+    CreateMockRead(*stream1_body, 4),
+    MockRead(true, ERR_IO_PENDING, 5),  // Force a pause
   };
 
   scoped_refptr<OrderedSocketData> data(
@@ -2372,7 +2486,6 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID9) {
   int rv = trans->Start(&CreateGetRequest(), &callback, BoundNetLog());
   EXPECT_EQ(ERR_IO_PENDING, rv);
   rv = callback.WaitForResult();
-  data->CompleteRead();
   EXPECT_EQ(OK, rv);
 
   // Verify that we consumed all test data.
@@ -2392,20 +2505,26 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushInvalidAssociatedStreamID9) {
 }
 
 TEST_P(SpdyNetworkTransactionTest, ServerPushNoURL) {
-  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
-  scoped_ptr<spdy::SpdyFrame> res(
-      ConstructSpdyRstStream(2, spdy::PROTOCOL_ERROR));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_syn(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_body(ConstructSpdyBodyFrame(1, true));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_rst(ConstructSpdyRstStream(2, spdy::PROTOCOL_ERROR));
   MockWrite writes[] = {
-    CreateMockWrite(*req, 1),
-    CreateMockWrite(*res, 4),
+    CreateMockWrite(*stream1_syn, 1),
+    CreateMockWrite(*stream2_rst, 4),
   };
 
-  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
-  scoped_ptr<spdy::SpdyFrame> rep(ConstructSpdyPush(NULL, 0, 2, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream1_reply(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame>
+      stream2_syn(ConstructSpdyPush(NULL, 0, 2, 1));
   MockRead reads[] = {
-    CreateMockRead(*resp, 2),
-    CreateMockRead(*rep, 3),
-    MockRead(true, 0, 0, 5)  // EOF
+    CreateMockRead(*stream1_reply, 2),
+    CreateMockRead(*stream2_syn, 3),
+    CreateMockRead(*stream1_body, 4),
+    MockRead(true, ERR_IO_PENDING, 5)  // Force a pause
   };
 
   scoped_refptr<OrderedSocketData> data(
@@ -2424,7 +2543,6 @@ TEST_P(SpdyNetworkTransactionTest, ServerPushNoURL) {
   int rv = trans->Start(&CreateGetRequest(), &callback, BoundNetLog());
   EXPECT_EQ(ERR_IO_PENDING, rv);
   rv = callback.WaitForResult();
-  data->CompleteRead();
   EXPECT_EQ(OK, rv);
   // Verify that we consumed all test data.
   EXPECT_TRUE(data->at_read_eof()) << "Read count: "
@@ -3743,8 +3861,8 @@ TEST_P(SpdyNetworkTransactionTest, ProxyConnect) {
                                      BoundNetLog(), GetParam());
   helper.session_deps().reset(new SpdySessionDependencies(
       net::SpdyCreateFixedProxyService("myproxy:70")));
-  helper.session() = SpdySessionDependencies::SpdyCreateSession(
-      helper.session_deps().get());
+  helper.SetSession(SpdySessionDependencies::SpdyCreateSession(
+      helper.session_deps().get()));
   helper.RunPreTestSetup();
   HttpNetworkTransaction* trans = helper.trans();
 
@@ -3970,7 +4088,7 @@ TEST_P(SpdyNetworkTransactionTest, DirectConnectProxyReconnect) {
   NormalSpdyTransactionHelper helper_proxy(request_proxy,
                                            BoundNetLog(), GetParam());
   helper_proxy.session_deps().swap(ssd_proxy);
-  helper_proxy.session() = session_proxy;
+  helper_proxy.SetSession(session_proxy);
   helper_proxy.RunPreTestSetup();
   helper_proxy.AddData(data_proxy.get());
 
diff --git a/net/spdy/spdy_session.cc b/net/spdy/spdy_session.cc
index c90bdd1..64a04cc 100644
--- a/net/spdy/spdy_session.cc
+++ b/net/spdy/spdy_session.cc
@@ -198,6 +198,10 @@ SpdySession::~SpdySession() {
     connection_->socket()->Disconnect();
   }
 
+  // Streams should all be gone now.
+  DCHECK_EQ(0u, num_active_streams());
+  DCHECK_EQ(0u, num_unclaimed_pushed_streams());
+
   RecordHistograms();
 
   net_log_.EndEvent(NetLog::TYPE_SPDY_SESSION, NULL);
@@ -781,6 +785,7 @@ void SpdySession::CloseAllStreams(net::Error status) {
   if (!unclaimed_pushed_streams_.empty()) {
     streams_abandoned_count_ += unclaimed_pushed_streams_.size();
     abandoned_push_streams.Add(unclaimed_pushed_streams_.size());
+    unclaimed_pushed_streams_.clear();
   }
 
   for (int i = 0;i < NUM_PRIORITIES;++i) {
@@ -854,14 +859,19 @@ void SpdySession::ActivateStream(SpdyStream* stream) {
 
 void SpdySession::DeleteStream(spdy::SpdyStreamId id, int status) {
   DLOG(INFO) << "Removing SpdyStream " << id << " from active stream list.";
-  // Remove the stream from unclaimed_pushed_streams_ and active_streams_.
-  PushedStreamMap::iterator it;
-  for (it = unclaimed_pushed_streams_.begin();
-       it != unclaimed_pushed_streams_.end(); ++it) {
-    scoped_refptr<SpdyStream> curr = it->second;
-    if (id == curr->stream_id()) {
-      unclaimed_pushed_streams_.erase(it);
-      break;
+
+  // For push streams, if they are being deleted normally, we leave
+  // the stream in the unclaimed_pushed_streams_ list.  However, if
+  // the stream is errored out, clean it up entirely.
+  if (status != OK) {
+    PushedStreamMap::iterator it;
+    for (it = unclaimed_pushed_streams_.begin();
+         it != unclaimed_pushed_streams_.end(); ++it) {
+      scoped_refptr<SpdyStream> curr = it->second;
+      if (id == curr->stream_id()) {
+        unclaimed_pushed_streams_.erase(it);
+        break;
+      }
     }
   }
 
diff --git a/net/spdy/spdy_session.h b/net/spdy/spdy_session.h
index 6fe8d37..480f9ad 100644
--- a/net/spdy/spdy_session.h
+++ b/net/spdy/spdy_session.h
@@ -162,6 +162,13 @@ class SpdySession : public base::RefCounted<SpdySession>,
 
   void set_in_session_pool(bool val) { in_session_pool_ = val; }
 
+  // Access to the number of active and pending streams.  These are primarily
+  // available for testing and diagnostics.
+  size_t num_active_streams() const { return active_streams_.size(); }
+  size_t num_unclaimed_pushed_streams() const {
+      return unclaimed_pushed_streams_.size();
+  }
+
  private:
   friend class base::RefCounted<SpdySession>;
   FRIEND_TEST_ALL_PREFIXES(SpdySessionTest, GetActivePushStream);
diff --git a/net/spdy/spdy_stream.cc b/net/spdy/spdy_stream.cc
index 71c0901..f7a1676 100644
--- a/net/spdy/spdy_stream.cc
+++ b/net/spdy/spdy_stream.cc
@@ -29,11 +29,12 @@ SpdyStream::SpdyStream(
       response_status_(OK),
       cancelled_(false),
       send_bytes_(0),
-      recv_bytes_(0),
-      histograms_recorded_(false) {}
+      recv_bytes_(0) {
+}
 
 SpdyStream::~SpdyStream() {
   DLOG(INFO) << "Deleting SpdyStream for stream " << stream_id_;
+  UpdateHistograms();
 }
 
 void SpdyStream::SetDelegate(Delegate* delegate) {
@@ -183,9 +184,12 @@ void SpdyStream::OnDataReceived(const char* data, int length) {
       IOBufferWithSize* buf = new IOBufferWithSize(length);
       memcpy(buf->data(), data, length);
       pending_buffers_.push_back(buf);
-    }
-    else
+    } else {
       pending_buffers_.push_back(NULL);
+      metrics_.StopStream();
+      session_->CloseStream(stream_id_, net::OK);
+      // Note: |this| may be deleted after calling CloseStream.
+    }
     return;
   }
 
@@ -202,9 +206,8 @@ void SpdyStream::OnDataReceived(const char* data, int length) {
   // A zero-length read means that the stream is being closed.
   if (!length) {
     metrics_.StopStream();
-    scoped_refptr<SpdyStream> self(this);
     session_->CloseStream(stream_id_, net::OK);
-    UpdateHistograms();
+    // Note: |this| may be deleted after calling CloseStream.
     return;
   }
 
@@ -419,11 +422,6 @@ int SpdyStream::DoOpen(int result) {
 }
 
 void SpdyStream::UpdateHistograms() {
-  if (histograms_recorded_)
-    return;
-
-  histograms_recorded_ = true;
-
   // We need all timers to be filled in, otherwise metrics can be bogus.
   if (send_time_.is_null() || recv_first_byte_time_.is_null() ||
       recv_last_byte_time_.is_null())