Fix WS large message bug

runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java

@@ -972,7 +972,7 @@ private int decodePacketTypeV5(
                 server.onDecodeError(traceId, authorization, PACKET_TOO_LARGE);
                 server.decoder = decodeIgnoreAll;
             }
-            else if (length >= 0)
+            else if (limit - packet.limit() >= length || server.decodeBudget() == 0)
             {
                 server.decodeablePacketBytes = packet.sizeof() + length;
                 server.decoder = decoder;
@@ -4975,6 +4975,11 @@ private boolean validContent(
                     payload.sizeof(), ValueConsumer.NOP);
         }
 
+        private int decodeBudget()
+        {
+            return decodeMax - (int) (decodeSeq - decodeAck);
+        }
+
         private final class Subscription
         {
             private int id = 0;

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

@@ -717,8 +717,8 @@ private final class WsClient
             private MutableDirectBuffer header;
             private int headerLength;
 
-            private int payloadProgress;
-            private int payloadLength;
+            private long payloadProgress;
+            private long payloadLength;
             private int maskingKey;
 
             private int statusLength;
@@ -1306,7 +1306,7 @@ private int decodeContinuation(
 
                 // TODO: limit acceptReply bytes by acceptReply window, or RESET on overflow?
 
-                final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
                 final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
                 doAppData(decodeTraceId, 0x80, maskingKey, payload);
@@ -1331,7 +1331,7 @@ private int decodeText(
 
                 // TODO: limit acceptReply bytes by acceptReply window, or RESET on overflow?
 
-                final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
                 final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
                 doAppData(decodeTraceId, 0x81, maskingKey, payload);
@@ -1354,7 +1354,7 @@ private int decodeBinary(
             {
                 // TODO: limit acceptReply bytes by acceptReply window, or RESET on overflow?
 
-                final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
                 final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
                 doAppData(decodeTraceId, 0x82, maskingKey, payload);
@@ -1403,7 +1403,7 @@ private int decodeClose(
                 }
                 else
                 {
-                    final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                    final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
                     payloadProgress += decodeBytes;
 
                     int remaining = Math.min(length, 2 - statusLength);
@@ -1443,7 +1443,7 @@ private int decodePing(
                 }
                 else
                 {
-                    final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                    final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
                     OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
 
@@ -1481,7 +1481,7 @@ private int decodePong(
                 }
                 else
                 {
-                    final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                    final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
                     payloadRO.wrap(buffer, offset, offset + decodeBytes);
 

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

@@ -281,8 +281,8 @@ private final class WsServer
         private MutableDirectBuffer status;
         private int statusLength;
 
-        private int payloadProgress;
-        private int payloadLength;
+        private long payloadProgress;
+        private long payloadLength;
         private int maskingKey;
 
         private long initialSeq;
@@ -809,7 +809,7 @@ private int decodeContinuation(
         {
             // TODO: limit acceptReply bytes by acceptReply window, or RESET on overflow?
 
-            final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+            final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
             final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
             stream.doAppData(decodeTraceId, decodeAuthorization, 0x80, maskingKey, payload);
@@ -834,7 +834,7 @@ private int decodeText(
 
             // TODO: limit acceptReply bytes by acceptReply window, or RESET on overflow?
 
-            final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+            final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
             final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
             stream.doAppData(decodeTraceId, decodeAuthorization, 0x81, maskingKey, payload);
@@ -857,7 +857,7 @@ private int decodeBinary(
         {
             // TODO: limit acceptReply bytes by acceptReply window, or RESET on overflow?
 
-            final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+            final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
             final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
             stream.doAppData(decodeTraceId, decodeAuthorization, 0x82, maskingKey, payload);
@@ -886,7 +886,7 @@ private int decodeClose(
             }
             else
             {
-                final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
                 payloadProgress += decodeBytes;
 
                 int remaining = Math.min(length, 2 - statusLength);
@@ -926,7 +926,7 @@ private int decodePong(
             }
             else
             {
-                final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
                 payloadRO.wrap(buffer, offset, offset + decodeBytes);
 
@@ -955,7 +955,7 @@ private int decodePing(
             }
             else
             {
-                final int decodeBytes = Math.min(length, payloadLength - payloadProgress);
+                final int decodeBytes = Math.min(length, (int) Math.min(payloadLength - payloadProgress, Integer.MAX_VALUE));
 
                 OctetsFW payload = payloadRO.wrap(buffer, offset, offset + decodeBytes);
 

runtime/binding-ws/src/main/java/io/aklivity/zilla/runtime/binding/ws/internal/types/codec/WsHeaderFW.java

@@ -84,11 +84,12 @@ public int maskingKey()
         return buffer().getInt(offset() + FIELD_OFFSET_FLAGS_AND_OPCODE + FIELD_SIZE_FLAGS_AND_OPCODE + lengthSize());
     }
 
-    public int length()
+    public long length()
     {
         return length(buffer(), offset());
     }
 
+    //TODO: add 10k tests with 8k buffer
     @Override
     public int limit()
     {
@@ -230,7 +231,7 @@ private int lengthSize()
         }
     }
 
-    private static int length(DirectBuffer buffer, int offset)
+    private static long length(DirectBuffer buffer, int offset)
     {
         int length = buffer.getByte(offset + FIELD_OFFSET_MASK_AND_LENGTH) & 0x7f;
 
@@ -241,23 +242,12 @@ private static int length(DirectBuffer buffer, int offset)
 
         case 0x7f:
             long length8bytes = buffer.getLong(offset + FIELD_OFFSET_MASK_AND_LENGTH + 1, ByteOrder.BIG_ENDIAN);
-            validateLength(length8bytes);
-            return (int) length8bytes & 0xffffffff;
-
+            return length8bytes & 0x7fff_ffff_ffff_ffffL;
         default:
             return length;
         }
     }
 
-    private static void validateLength(
-            long length8bytes)
-    {
-        if (length8bytes >> 17L != 0L)
-        {
-            throw new IllegalStateException(String.format("frame payload=%d too long", length8bytes));
-        }
-    }
-
     private static int lengthSize0(byte b)
     {
         switch (b & 0x7f)

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

@@ -72,6 +72,18 @@ public void shouldEchoBinaryFrameWithPayloadLength125() throws Exception
         k3po.finish();
     }
 
+    @Ignore
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${net}/echo.binary.payload.length.10k/handshake.request.and.frame",
+        "${app}/echo.binary.payload.length.10k/handshake.response.and.frame" })
+    //@Configure(name = ENGINE_BUFFER_SLOT_CAPACITY_NAME, value = "8192")
+    public void shouldEchoBinaryFrameWithPayloadLength10k() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("server.yaml")
     @Specification({

specs/binding-ws.spec/src/main/scripts/io/aklivity/zilla/specs/binding/ws/streams/application/framing/echo.binary.payload.length.10k/handshake.request.and.frame.rpt

@@ -0,0 +1,37 @@
+#
+# Copyright 2021-2023 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()}
+property writeMask ${http:randomBytes(4)}
+property client10240 ${http:randomBytes(10240)}
+
+connect "zilla://streams/app0"
+        option zilla:window 8192
+        option zilla:transmission "duplex"
+
+write zilla:begin.ext ${ws:beginEx()
+                            .typeId(zilla:id("ws"))
+                            .protocol(null)
+                            .scheme("http")
+                            .authority("localhost:8080")
+                            .path("/echo")
+                            .build()}
+
+connected
+
+write ${client10240}
+
+read ${client10240}

specs/binding-ws.spec/src/main/scripts/io/aklivity/zilla/specs/binding/ws/streams/application/framing/echo.binary.payload.length.10k/handshake.response.and.frame.rpt

@@ -0,0 +1,34 @@
+#
+# Copyright 2021-2023 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/app0"
+  option zilla:window 8192
+  option zilla:transmission "duplex"
+accepted
+
+read zilla:begin.ext ${ws:beginEx()
+                           .typeId(zilla:id("ws"))
+                           .protocol(null)
+                           .scheme("http")
+                           .authority("localhost:8080")
+                           .path("/echo")
+                           .build()}
+
+connected
+
+read ([0..10240] :server10240)
+
+write ${server10240}

specs/binding-ws.spec/src/main/scripts/io/aklivity/zilla/specs/binding/ws/streams/network/framing/echo.binary.payload.length.10k/handshake.request.and.frame.rpt

@@ -0,0 +1,53 @@
+#
+# Copyright 2021-2023 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 location 'http://localhost:8080/echo'
+property key ${ws:handshakeKey()}
+property writeMask ${http:randomBytes(4)}
+property client10240 ${http:randomBytes(10240)}
+
+connect "zilla://streams/net0"
+  option zilla:window 8192
+  option zilla:transmission "duplex"
+
+write zilla:begin.ext ${http:beginEx()
+                              .typeId(zilla:id("http"))
+                              .header(":method", "GET")
+                              .header(":scheme", "http")
+                              .header(":authority", "localhost:8080")
+                              .header(":path", "/echo")
+                              .header("upgrade", "websocket")
+                              .header("connection", "upgrade")
+                              .header("sec-websocket-key", key)
+                              .header("sec-websocket-version", "13")
+                              .build()}
+
+read zilla:begin.ext ${http:beginEx()
+                             .typeId(zilla:id("http"))
+                             .header(":status", "101")
+                             .header("upgrade", "websocket")
+                             .header("connection", "upgrade")
+                             .header("sec-websocket-accept", ws:handshakeHash(key))
+                             .build()}
+
+connected
+
+write [0x82 0x7F 0x28] ${writeMask}
+write option mask ${writeMask}
+write ${client10240}
+write option mask [0x00 0x00 0x00 0x00]
+
+read [0x82 0x7d] ${client10240}

specs/binding-ws.spec/src/main/scripts/io/aklivity/zilla/specs/binding/ws/streams/network/framing/echo.binary.payload.length.10k/handshake.response.and.frame.rpt

@@ -0,0 +1,51 @@
+#
+# Copyright 2021-2023 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 location 'http://localhost:8080/echo'
+
+accept "zilla://streams/net0"
+  option zilla:window 8192
+  option zilla:transmission "duplex"
+accepted
+
+read zilla:begin.ext ${zilla:id("http")}
+                       [0xb4 0x00 0x00 0x00]
+                       [0x08 0x00 0x00 0x00]
+                       [0x07] ":method" [0x03 0x00] "GET"
+                       [0x07] ":scheme" [0x04 0x00] "http"
+                       [0x0a] ":authority" [0x0e 0x00] "localhost:8080"
+                       [0x05] ":path" [0x05 0x00] "/echo"
+                       [0x07] "upgrade" [0x09 0x00] "websocket"
+                       [0x0a] "connection" [0x07 0x00] "upgrade"
+                       [0x11] "sec-websocket-key" [0x18 0x00] /(?<key>[a-zA-Z0-9+\/=]{24})/
+                       [0x15] "sec-websocket-version" [0x02 0x00] "13"
+
+write zilla:begin.ext ${http:beginEx()
+                              .typeId(zilla:id("http"))
+                              .header(":status", "101")
+                              .header("upgrade", "websocket")
+                              .header("connection", "upgrade")
+                              .header("sec-websocket-accept", ws:handshakeHash(key))
+                              .build()}
+
+connected
+
+read [0x82 0xfd] ([0..4] :readMask)
+read option mask ${readMask}
+read ([0..125] :server125)
+read option mask [0x00 0x00 0x00 0x00]
+
+write [0x82 0x7d] ${server125}