Correctly send http end when no upgrade header (#1626)

Author: bmaidics

runtime/binding-ws/src/main/java/io/aklivity/zilla/runtime/binding/ws/internal/stream/WsServerFactory.java

@@ -210,7 +210,8 @@ public MessageConsumer newStream(
             doHttpBegin(sender, originId, routedId, newReplyId, 0L, 0L, 0, traceId, authorization, affinity,
                 hs -> hs.item(h -> h.name(":status").value("400"))
                         .item(h -> h.name("connection").value("close")));
-            doHttpEnd(sender, originId, routedId, newReplyId, traceId);
+            doWindow(sender, originId, routedId, initialId, 0, 0, 0, traceId, authorization, 0, 0);
+            doHttpEnd(sender, originId, routedId, newReplyId, 0L, 0L, 0, traceId, authorization);
             newStream = (t, b, o, l) -> {};
         }
         else if (key != null &&
@@ -1487,18 +1488,55 @@ private void doHttpEnd(
         long originId,
         long routedId,
         long streamId,
-        long traceId)
+        long sequence,
+        long acknowledge,
+        int maximum,
+        long traceId,
+        long authorization)
     {
         final EndFW end = endRW.wrap(writeBuffer, 0, writeBuffer.capacity())
-                .originId(originId)
-                .routedId(routedId)
-                .streamId(streamId)
-                .traceId(traceId)
-                .build();
+            .originId(originId)
+            .routedId(routedId)
+            .streamId(streamId)
+            .sequence(sequence)
+            .acknowledge(acknowledge)
+            .maximum(maximum)
+            .traceId(traceId)
+            .authorization(authorization)
+            .build();
 
         receiver.accept(end.typeId(), end.buffer(), end.offset(), end.sizeof());
     }
 
+    private void doWindow(
+        MessageConsumer receiver,
+        long originId,
+        long routedId,
+        long streamId,
+        long sequence,
+        long acknowledge,
+        int maximum,
+        long traceId,
+        long authorization,
+        long budgetId,
+        int padding)
+    {
+        final WindowFW window = windowRW.wrap(writeBuffer, 0, writeBuffer.capacity())
+            .originId(originId)
+            .routedId(routedId)
+            .streamId(streamId)
+            .sequence(sequence)
+            .acknowledge(acknowledge)
+            .maximum(maximum)
+            .traceId(traceId)
+            .authorization(authorization)
+            .budgetId(budgetId)
+            .padding(padding)
+            .build();
+
+        receiver.accept(window.typeId(), window.buffer(), window.offset(), window.sizeof());
+    }
+
     private Flyweight.Builder.Visitor visitWsDataEx(
         int flags)
     {

runtime/binding-ws/src/test/java/io/aklivity/zilla/runtime/binding/ws/internal/streams/server/AdvisoryIT.java

@@ -91,4 +91,13 @@ public void shouldReceiveClientSentChallenge() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${net}/client.sent.no.upgrade/client" })
+    public void shouldReceiveClientEndWhenNoUpdate() throws Exception
+    {
+        k3po.finish();
+    }
 }

specs/binding-ws.spec/src/main/scripts/io/aklivity/zilla/specs/binding/ws/streams/network/advise/client.sent.no.upgrade/client.rpt

@@ -0,0 +1,36 @@
+#
+# Copyright 2021-2024 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+property key ${ws:handshakeKey()}
+
+connect "zilla://streams/net0"
+  option zilla:window 8192
+  option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${http:beginEx()
+                              .typeId(zilla:id("http"))
+                              .header(":method", "GET")
+                              .build()}
+
+connected
+
+read zilla:begin.ext ${http:beginEx()
+                             .typeId(zilla:id("http"))
+                             .header(":status", "400")
+                             .header("connection", "close")
+                             .build()}
+
+read closed

specs/binding-ws.spec/src/main/scripts/io/aklivity/zilla/specs/binding/ws/streams/network/advise/client.sent.no.upgrade/server.rpt

@@ -0,0 +1,35 @@
+#
+# Copyright 2021-2024 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+accept "zilla://streams/net0"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+accepted
+
+read zilla:begin.ext ${http:beginEx()
+                              .typeId(zilla:id("http"))
+                              .header(":method", "GET")
+                              .build()}
+
+write zilla:begin.ext ${http:beginEx()
+                             .typeId(zilla:id("http"))
+                             .header(":status", "400")
+                             .header("connection", "close")
+                             .build()}
+
+connected
+
+write close

specs/binding-ws.spec/src/test/java/io/aklivity/zilla/specs/binding/ws/streams/network/AdvisoryIT.java

@@ -72,4 +72,13 @@ public void shouldReceiveClientSentChallenge() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Specification({
+        "${net}/client.sent.no.upgrade/client",
+        "${net}/client.sent.no.upgrade/server" })
+    public void shouldReceiveClientEndWhenNoUpdate() throws Exception
+    {
+        k3po.finish();
+    }
 }