8284585: PushPromiseContinuation test fails intermittently in timeout

Reviewed-by: abakhtin
Backport-of: 65da38d844760f7d17a143f8b4d5e25ea0144e27

src/java.net.http/share/classes/jdk/internal/net/http/Http2Connection.java

@@ -763,7 +763,7 @@ class Http2Connection  {
             }
 
             Stream<?> stream = getStream(streamid);
-            if (stream == null) {
+            if (stream == null && pushContinuationState == null) {
                 // Should never receive a frame with unknown stream id
 
                 if (frame instanceof HeaderFrame) {
@@ -803,7 +803,11 @@ class Http2Connection  {
             if (pushContinuationState != null) {
                 if (frame instanceof ContinuationFrame cf) {
                     try {
+                        if (streamid == pushContinuationState.pushContFrame.streamid())
                             handlePushContinuation(stream, cf);
+                        else
+                            protocolError(ErrorFrame.PROTOCOL_ERROR, "Received a Continuation Frame with an " +
+                                    "unexpected stream id");
                     } catch (UncheckedIOException e) {
                         debug.log("Error handling Push Promise with Continuation: " + e.getMessage(), e);
                         protocolError(ErrorFrame.PROTOCOL_ERROR, e.getMessage());
@@ -890,8 +894,6 @@ class Http2Connection  {
 
     private <T> void completePushPromise(int promisedStreamid, Stream<T> parent, HttpHeaders headers)
             throws IOException {
-        // Perhaps the following checks could be moved to handlePushPromise()
-        // to reset the PushPromise stream earlier?
         HttpRequestImpl parentReq = parent.request;
         if (promisedStreamid != nextPushStream) {
             resetStream(promisedStreamid, ResetFrame.PROTOCOL_ERROR);

test/jdk/java/net/httpclient/http2/PushPromiseContinuation.java

@@ -59,6 +59,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.BiPredicate;
@@ -71,7 +72,7 @@ import jdk.httpclient.test.lib.http2.OutgoingPushPromise;
 import jdk.httpclient.test.lib.http2.Http2TestServerConnection;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.*;
 
 public class PushPromiseContinuation {
 
@@ -102,7 +103,7 @@ public class PushPromiseContinuation {
 
         // Need to have a custom exchange supplier to manage the server's push
         // promise with continuation flow
-        server.setExchangeSupplier(Http2LPPTestExchangeImpl::new);
+        server.setExchangeSupplier(Http2PushPromiseContinuationExchangeImpl::new);
 
         System.err.println("PushPromiseContinuation: Server listening on port " + server.getAddress().getPort());
         server.start();
@@ -171,6 +172,31 @@ public class PushPromiseContinuation {
         verify(resp);
     }
 
+    @Test
+    public void testSendHeadersOnPushPromiseStream() throws Exception {
+        // This test server sends a push promise that should be followed by a continuation but
+        // incorrectly sends on Response Headers while the client awaits the continuation.
+        Http2TestServer faultyServer = new Http2TestServer(false, 0);
+        faultyServer.addHandler(new ServerPushHandler(), "/");
+        faultyServer.setExchangeSupplier(Http2PushPromiseHeadersExchangeImpl::new);
+        System.err.println("PushPromiseContinuation: FaultyServer listening on port " + faultyServer.getAddress().getPort());
+        faultyServer.start();
+
+        int faultyPort = faultyServer.getAddress().getPort();
+        URI faultyUri = new URI("http://localhost:" + faultyPort + "/");
+
+        HttpClient client = HttpClient.newHttpClient();
+        // Server is making a request to an incorrect URI
+        HttpRequest hreq = HttpRequest.newBuilder(faultyUri).version(HttpClient.Version.HTTP_2).GET().build();
+        CompletableFuture<HttpResponse<String>> cf =
+                client.sendAsync(hreq, HttpResponse.BodyHandlers.ofString(UTF_8), pph);
+
+        CompletionException t = expectThrows(CompletionException.class, () -> cf.join());
+        assertEquals(t.getCause().getClass(), IOException.class, "Expected an IOException but got " + t.getCause());
+        System.err.println("Client received the following expected exception: " + t.getCause());
+        faultyServer.stop();
+    }
+
     private void verify(HttpResponse<String> resp) {
         assertEquals(resp.statusCode(), 200);
         assertEquals(resp.body(), mainResponseBody);
@@ -191,12 +217,43 @@ public class PushPromiseContinuation {
         }
     }
 
-    static class Http2LPPTestExchangeImpl extends Http2TestExchangeImpl {
+    static class Http2PushPromiseHeadersExchangeImpl extends Http2TestExchangeImpl {
+
+        Http2PushPromiseHeadersExchangeImpl(int streamid, String method, HttpHeaders reqheaders, HttpHeadersBuilder rspheadersBuilder, URI uri, InputStream is, SSLSession sslSession, BodyOutputStream os, Http2TestServerConnection conn, boolean pushAllowed) {
+            super(streamid, method, reqheaders, rspheadersBuilder, uri, is, sslSession, os, conn, pushAllowed);
+        }
+
+
+        @Override
+        public void serverPush(URI uri, HttpHeaders headers, InputStream content) {
+            HttpHeadersBuilder headersBuilder = new HttpHeadersBuilder();
+            headersBuilder.setHeader(":method", "GET");
+            headersBuilder.setHeader(":scheme", uri.getScheme());
+            headersBuilder.setHeader(":authority", uri.getAuthority());
+            headersBuilder.setHeader(":path", uri.getPath());
+            for (Map.Entry<String,List<String>> entry : headers.map().entrySet()) {
+                for (String value : entry.getValue())
+                    headersBuilder.addHeader(entry.getKey(), value);
+            }
+            HttpHeaders combinedHeaders = headersBuilder.build();
+            OutgoingPushPromise pp = new OutgoingPushPromise(streamid, uri, combinedHeaders, content);
+            // Indicates to the client that a continuation should be expected
+            pp.setFlag(0x0);
+            try {
+                conn.addToOutputQ(pp);
+                // writeLoop will spin up thread to read the InputStream
+            } catch (IOException ex) {
+                System.err.println("TestServer: pushPromise exception: " + ex);
+            }
+        }
+    }
+
+    static class Http2PushPromiseContinuationExchangeImpl extends Http2TestExchangeImpl {
 
         HttpHeadersBuilder pushPromiseHeadersBuilder;
         List<ContinuationFrame> cfs;
 
-        Http2LPPTestExchangeImpl(int streamid, String method, HttpHeaders reqheaders,
+        Http2PushPromiseContinuationExchangeImpl(int streamid, String method, HttpHeaders reqheaders,
                                                  HttpHeadersBuilder rspheadersBuilder, URI uri, InputStream is,
                                                  SSLSession sslSession, BodyOutputStream os,
                                                  Http2TestServerConnection conn, boolean pushAllowed) {
@@ -253,7 +310,8 @@ public class PushPromiseContinuation {
             HttpHeaders pushPromiseHeaders = pushPromiseHeadersBuilder.build();
             testHeaders = testHeadersBuilder.build();
             // Create the Push Promise Frame
-            OutgoingPushPromise pp = new OutgoingPushPromise(streamid, uri, pushPromiseHeaders, content);
+            OutgoingPushPromise pp = new OutgoingPushPromise(streamid, uri, pushPromiseHeaders, content, cfs);
+
             // Indicates to the client that a continuation should be expected
             pp.setFlag(0x0);
 
@@ -261,10 +319,6 @@ public class PushPromiseContinuation {
                 // Schedule push promise and continuation for sending
                 conn.addToOutputQ(pp);
                 System.err.println("Server: Scheduled a Push Promise to Send");
-                for (ContinuationFrame cf : cfs) {
-                    conn.addToOutputQ(cf);
-                    System.err.println("Server: Scheduled a Continuation to Send");
-                }
             } catch (IOException ex) {
                 System.err.println("Server: pushPromise exception: " + ex);
             }

test/jdk/java/net/httpclient/lib/jdk/httpclient/test/lib/http2/Http2TestServerConnection.java

@@ -951,6 +951,11 @@ public class Http2TestServerConnection {
         nextPushStreamId += 2;
         pp.streamid(op.parentStream);
         writeFrame(pp);
+        // No need to check for END_HEADERS flag here to allow for tests to simulate bad server side
+        // behavior i.e Continuation Frames included with END_HEADERS flag set
+        for (Http2Frame cf : op.getContinuations())
+            writeFrame(cf);
+
         final InputStream ii = op.is;
         final BodyOutputStream oo = new BodyOutputStream(
                 promisedStreamid,

test/jdk/java/net/httpclient/lib/jdk/httpclient/test/lib/http2/OutgoingPushPromise.java

@@ -26,6 +26,9 @@ package jdk.httpclient.test.lib.http2;
 import java.io.InputStream;
 import java.net.URI;
 import java.net.http.HttpHeaders;
+import java.util.List;
+
+import jdk.internal.net.http.frame.ContinuationFrame;
 import jdk.internal.net.http.frame.Http2Frame;
 
 // will be converted to a PushPromiseFrame in the writeLoop
@@ -35,16 +38,29 @@ public class OutgoingPushPromise extends Http2Frame {
     final URI uri;
     final InputStream is;
     final int parentStream; // not the pushed streamid
+    private final List<Http2Frame> continuations;
 
     public OutgoingPushPromise(int parentStream,
                                URI uri,
                                HttpHeaders headers,
                                InputStream is) {
+        this(parentStream, uri, headers, is, List.of());
+    }
+
+    public OutgoingPushPromise(int parentStream,
+                               URI uri,
+                               HttpHeaders headers,
+                               InputStream is,
+                               List<ContinuationFrame> continuations) {
         super(0,0);
         this.uri = uri;
         this.headers = headers;
         this.is = is;
         this.parentStream = parentStream;
+        this.continuations = List.copyOf(continuations);
     }
 
+    public List<Http2Frame> getContinuations() {
+        return continuations;
+    }
 }