feat(command-dump): add REDIRECT control frame support (#1759)

* feat(command-dump): add REDIRECT control frame support

Capture the engine's REDIRECT frame (type id 0x40000005, introduced
in #1755) in the dump command's pcap output and surface it in the
zilla.lua Wireshark dissector. Like RESET, REDIRECT terminates the
stream, so it is mapped to a TCP PSH segment. Like BEGIN, the frame
carries an int64 affinity and a round-tripped extension, so each
binding's extension dissector treats REDIRECT the same as BEGIN.

Adds k3po scripts for the client-sent redirect scenario; the IT method
is wired but @ignore'd until its tshark-generated expected fixture is
regenerated on a docker-enabled host.

https://claude.ai/code/session_01K4wd5z8Ejgq4ejTsmL88oh

* fix(engine): align REDIRECT teardown with RESET in k3po-zilla transport

Per the engine ABI from #1755, REDIRECT terminates the stream like RESET
(EngineWorker dispatcher.remove + Target streams[].remove for both
RedirectFW.TYPE_ID and ResetFW.TYPE_ID). The k3po-zilla test transport
was leaving the channel in an inconsistent state on both ends, so
follow-up cleanup or script steps could emit END / ABORT frames against
a stream the engine had already torn down. This produced inconsistent
zilla dump pcap output for redirect scenarios.

Mirror the existing RESET pattern exactly:

- `ZillaSource.doAdviseInput` (ADVISORY_REDIRECT branch) — the caller of
  `streamFactory.doRedirect` now closes the local READ side after the
  frame is written, mirroring `doAbortInputAfterBegin` for RESET. The
  frame writer (`ZillaTarget.doRedirect`) stays state-free, mirroring
  `ZillaTarget.doReset` / `doAbortInput`.
- `Throttle.onRedirect` — on the receiver side, mirror `Throttle.onReset`
  exactly: `setWriteAborted` + `setWriteClosed` + `fireOutputAborted` +
  the disconnect / unbind / close events. The receiver is the source of
  the stream (REDIRECT travels in the throttle direction, opposite the
  BEGIN), so its WRITE side is what the redirect terminates — same as
  RESET semantics.

The pre-existing `binding-http` `RedirectIT` (`read advise zilla:redirect`
+ `write aborted` on the application side, added in #1758) was authored
against the previous buggy transport behavior and has been failing on
develop ever since. With this fix in place it will need to be revisited
in a follow-up — that test is unaffected by this commit and remains in
the same failure state as on develop's HEAD.

Verified: engine surefire 320/320 + failsafe 204/204
(incl. SimplexIT 37/37 with the PR #1755 REDIRECT test).

https://claude.ai/code/session_01K4wd5z8Ejgq4ejTsmL88oh

* test(command-dump): enable shouldReceiveClientSentRedirect

Pair the redirect rpt scripts with `read abort` (client) and
`write aborted` (server) so post-REDIRECT teardown is deterministic
once the k3po-zilla transport aligns REDIRECT with RESET (#13ad202a1).
Add the tshark fixture and unignore the IT method.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: jfallows

incubator/command-dump/src/main/java/io/aklivity/zilla/runtime/command/dump/internal/airline/ZillaDumpCommand.java

@@ -79,6 +79,7 @@
 import io.aklivity.zilla.runtime.command.dump.internal.types.stream.ExtensionFW;
 import io.aklivity.zilla.runtime.command.dump.internal.types.stream.FlushFW;
 import io.aklivity.zilla.runtime.command.dump.internal.types.stream.FrameFW;
+import io.aklivity.zilla.runtime.command.dump.internal.types.stream.RedirectFW;
 import io.aklivity.zilla.runtime.command.dump.internal.types.stream.ResetFW;
 import io.aklivity.zilla.runtime.command.dump.internal.types.stream.SignalFW;
 import io.aklivity.zilla.runtime.command.dump.internal.types.stream.WindowFW;
@@ -216,6 +217,7 @@ public final class ZillaDumpCommand extends ZillaCommand
     private final FlushFW flushRO = new FlushFW();
     private final SignalFW signalRO = new SignalFW();
     private final ChallengeFW challengeRO = new ChallengeFW();
+    private final RedirectFW redirectRO = new RedirectFW();
     private final PcapGlobalHeaderFW.Builder pcapGlobalHeaderRW = new PcapGlobalHeaderFW.Builder();
     private final PcapPacketHeaderFW.Builder pcapPacketHeaderRW = new PcapPacketHeaderFW.Builder();
     private final BeginFW.Builder beginRW = new BeginFW.Builder();
@@ -225,6 +227,7 @@ public final class ZillaDumpCommand extends ZillaCommand
     private final ResetFW.Builder resetRW = new ResetFW.Builder();
     private final FlushFW.Builder flushRW = new FlushFW.Builder();
     private final ChallengeFW.Builder challengeRW = new ChallengeFW.Builder();
+    private final RedirectFW.Builder redirectRW = new RedirectFW.Builder();
     private final IPv6HeaderFW.Builder ipv6HeaderRW = new IPv6HeaderFW.Builder();
     private final IPv6JumboHeaderFW.Builder ipv6JumboHeaderRW = new IPv6JumboHeaderFW.Builder();
     private final TcpHeaderFW.Builder tcpHeaderRW = new TcpHeaderFW.Builder();
@@ -595,6 +598,9 @@ private boolean handleFrame(
             case ChallengeFW.TYPE_ID:
                 onChallenge(challengeRO.wrap(buffer, index, index + length));
                 break;
+            case RedirectFW.TYPE_ID:
+                onRedirect(redirectRO.wrap(buffer, index, index + length));
+                break;
             default:
                 break;
             }
@@ -740,6 +746,21 @@ private void onChallenge(
             }
         }
 
+        private void onRedirect(
+            RedirectFW redirect)
+        {
+            if (allowedBinding.test(redirect.routedId()))
+            {
+                int offset = redirect.offset() - HEADER_LENGTH;
+                final RedirectFW newRedirect = redirectRW.wrap(patchBuffer, 0, redirect.sizeof()).set(redirect).build();
+                final ExtensionFW extension = newRedirect.extension().get(extensionRO::tryWrap);
+                patchExtension(patchBuffer, extension, RedirectFW.FIELD_OFFSET_EXTENSION);
+
+                writeFrame(RedirectFW.TYPE_ID, worker, offset, newRedirect.originId(), newRedirect.routedId(),
+                    newRedirect.streamId(), newRedirect.timestamp(), newRedirect, PSH);
+            }
+        }
+
         private void patchExtension(
             MutableDirectBuffer buffer,
             ExtensionFW extension,

incubator/command-dump/src/main/resources/io/aklivity/zilla/runtime/command/dump/internal/airline/zilla.lua

@@ -41,6 +41,7 @@ RESET_ID = 0x40000001
 WINDOW_ID = 0x40000002
 SIGNAL_ID = 0x40000003
 CHALLENGE_ID = 0x40000004
+REDIRECT_ID = 0x40000005
 
 AMQP_ID = 0x112dc182
 FILESYSTEM_ID = 0xe4e6aa9e
@@ -867,6 +868,8 @@ function zilla_protocol.dissector(buffer, pinfo, tree)
         handle_signal_frame(buffer, next_offset, subtree, pinfo, info)
     elseif frame_type_id == ABORT_ID or frame_type_id == RESET_ID or frame_type_id == CHALLENGE_ID then
         handle_extension(buffer, subtree, pinfo, info, next_offset, frame_type_id)
+    elseif frame_type_id == REDIRECT_ID then
+        handle_redirect_frame(buffer, next_offset, subtree, pinfo, info)
     end
 end
 
@@ -876,6 +879,12 @@ function handle_begin_frame(buffer, offset, subtree, pinfo, info)
     handle_extension(buffer, subtree, pinfo, info, offset + 8, BEGIN_ID)
 end
 
+function handle_redirect_frame(buffer, offset, subtree, pinfo, info)
+    local slice_affinity = buffer(offset, 8)
+    subtree:add_le(fields.affinity, slice_affinity)
+    handle_extension(buffer, subtree, pinfo, info, offset + 8, REDIRECT_ID)
+end
+
 function handle_data_frame(buffer, offset, tree, subtree, sequence, acknowledge, maximum, pinfo, info, protocol_type)
     local slice_flags = buffer(offset, 1)
     local flags_label = string.format("Flags: 0x%02x", slice_flags:le_uint())
@@ -974,6 +983,7 @@ function resolve_frame_type(frame_type_id)
     elseif frame_type_id == WINDOW_ID    then frame_type = "WINDOW"
     elseif frame_type_id == SIGNAL_ID    then frame_type = "SIGNAL"
     elseif frame_type_id == CHALLENGE_ID then frame_type = "CHALLENGE"
+    elseif frame_type_id == REDIRECT_ID  then frame_type = "REDIRECT"
     end
     return frame_type
 end
@@ -1280,7 +1290,8 @@ function add_proxy_string_as_subtree(buffer, tree, label_format, slice_type_id,
 end
 
 function handle_http_extension(buffer, offset, ext_subtree, frame_type_id)
-    if frame_type_id == BEGIN_ID or frame_type_id == RESET_ID or frame_type_id == CHALLENGE_ID then
+    if frame_type_id == BEGIN_ID or frame_type_id == RESET_ID or frame_type_id == CHALLENGE_ID or
+            frame_type_id == REDIRECT_ID then
         dissect_and_add_http_headers(buffer, offset, ext_subtree, "Headers", "Header")
     elseif frame_type_id == END_ID then
         dissect_and_add_http_headers(buffer, offset, ext_subtree, "Trailers", "Trailer")
@@ -1311,7 +1322,7 @@ function dissect_and_add_http_headers(buffer, offset, tree, plural_name, singula
 end
 
 function handle_grpc_extension(buffer, offset, ext_subtree, frame_type_id)
-    if frame_type_id == BEGIN_ID then
+    if frame_type_id == BEGIN_ID or frame_type_id == REDIRECT_ID then
         handle_grpc_begin_extension(buffer, offset, ext_subtree)
     elseif frame_type_id == DATA_ID then
         handle_grpc_data_extension(buffer, offset, ext_subtree)
@@ -1464,7 +1475,7 @@ function decode_varuint32(buffer, offset)
 end
 
 function handle_sse_extension(buffer, offset, ext_subtree, frame_type_id)
-    if frame_type_id == BEGIN_ID then
+    if frame_type_id == BEGIN_ID or frame_type_id == REDIRECT_ID then
         handle_sse_begin_extension(buffer, offset, ext_subtree)
     elseif frame_type_id == DATA_ID then
         handle_sse_data_extension(buffer, offset, ext_subtree)
@@ -1521,7 +1532,7 @@ function handle_sse_end_extension(buffer, offset, ext_subtree)
 end
 
 function handle_ws_extension(buffer, offset, ext_subtree, frame_type_id)
-    if frame_type_id == BEGIN_ID then
+    if frame_type_id == BEGIN_ID or frame_type_id == REDIRECT_ID then
         handle_ws_begin_extension(buffer, offset, ext_subtree)
     elseif frame_type_id == DATA_ID then
         handle_ws_data_extension(buffer, offset, ext_subtree)
@@ -1629,12 +1640,13 @@ function handle_filesystem_extension(buffer, offset, ext_subtree)
 end
 
 function handle_mqtt_extension(buffer, offset, ext_subtree, frame_type_id)
-    if frame_type_id == BEGIN_ID or frame_type_id == DATA_ID or frame_type_id == FLUSH_ID then
+    if frame_type_id == BEGIN_ID or frame_type_id == DATA_ID or frame_type_id == FLUSH_ID or
+            frame_type_id == REDIRECT_ID then
         local kind_length = 1
         local slice_kind = buffer(offset, kind_length)
         local kind = mqtt_ext_kinds[slice_kind:le_int()]
         ext_subtree:add_le(fields.mqtt_ext_kind, slice_kind)
-        if frame_type_id == BEGIN_ID then
+        if frame_type_id == BEGIN_ID or frame_type_id == REDIRECT_ID then
             if kind == "PUBLISH" then
                 handle_mqtt_begin_publish_extension(buffer, offset + kind_length, ext_subtree)
             elseif kind == "SUBSCRIBE" then
@@ -2046,12 +2058,13 @@ function handle_mqtt_reset_extension(buffer, offset, ext_subtree)
 end
 
 function handle_kafka_extension(buffer, offset, ext_subtree, frame_type_id)
-    if frame_type_id == BEGIN_ID or frame_type_id == DATA_ID or frame_type_id == FLUSH_ID then
+    if frame_type_id == BEGIN_ID or frame_type_id == DATA_ID or frame_type_id == FLUSH_ID or
+            frame_type_id == REDIRECT_ID then
         local api_length = 1
         local slice_api = buffer(offset, api_length)
         local api = kafka_ext_apis[slice_api:le_uint()]
         ext_subtree:add_le(fields.kafka_ext_api, slice_api)
-        if frame_type_id == BEGIN_ID then
+        if frame_type_id == BEGIN_ID or frame_type_id == REDIRECT_ID then
             if api == "CONSUMER" then
                 handle_kafka_begin_consumer_extension(buffer, offset + api_length, ext_subtree)
             elseif api == "GROUP" then

incubator/command-dump/src/main/scripts/io/aklivity/zilla/runtime/command/dump/binding/ws/streams/application/advise/client.sent.redirect/client.rpt

@@ -0,0 +1,34 @@
+#
+# Copyright 2021-2024 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+connect "zilla://streams/app0"
+        option zilla:ephemeral "app0"
+        option zilla:timestamps "false"
+        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 advised zilla:redirect 0x00000000000000b1L
+
+read abort

incubator/command-dump/src/main/scripts/io/aklivity/zilla/runtime/command/dump/binding/ws/streams/application/advise/client.sent.redirect/server.rpt

@@ -0,0 +1,34 @@
+#
+# Copyright 2021-2024 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES 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:timestamps "false"
+    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 advise zilla:redirect 0x00000000000000b1L
+
+write aborted

incubator/command-dump/src/test/java/io/aklivity/zilla/runtime/command/dump/internal/WsAdvisoryApplicationIT.java

@@ -49,4 +49,13 @@ public void shouldReceiveClientSentChallenge() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Specification({
+        "${app}/client.sent.redirect/client",
+        "${app}/client.sent.redirect/server" })
+    public void shouldReceiveClientSentRedirect() throws Exception
+    {
+        k3po.finish();
+    }
 }