fix: preserve IO::Async accepted sockets and binary payloads

docs/about/changelog.md

@@ -6,9 +6,9 @@ priorities and future plans.
 
 ## Work in progress
 
-- Preserve IO::Async thread callback results in scalar and list context on
-  both execution backends, and align its notifier-loop refcount expectation
-  with native Perl.
+- Preserve IO::Async thread callback results and accepted listener sockets on
+  both execution backends, retain binary channel payload octets, and align its
+  notifier-loop refcount expectation with native Perl.
 
 - Fix parsing of dense Mo::Inline expressions that use `::` as a bareword.
 

src/main/java/org/perlonjava/runtime/io/InternalPipeHandle.java

@@ -190,7 +190,28 @@ public RuntimeScalar write(String string) {
         }
 
         try {
-            byte[] bytes = string.getBytes(StandardCharsets.UTF_8);
+            // An unlayered Perl handle writes octets, not UTF-8-encoded text.
+            // RuntimeScalar represents byte strings as chars in the 0..255
+            // range, so UTF-8 encoding here expands octets such as 0x8a to
+            // two bytes. That corrupts framed binary protocols (notably
+            // Storable over IO::Async::Channel) because their length prefix
+            // still describes the original octet count.
+            boolean hasWideChars = false;
+            for (int i = 0; i < string.length(); i++) {
+                if (string.charAt(i) > 0xFF) {
+                    hasWideChars = true;
+                    break;
+                }
+            }
+            byte[] bytes;
+            if (hasWideChars) {
+                bytes = string.getBytes(StandardCharsets.UTF_8);
+            } else {
+                bytes = new byte[string.length()];
+                for (int i = 0; i < string.length(); i++) {
+                    bytes[i] = (byte) string.charAt(i);
+                }
+            }
             if (!blocking && inputStream != null) {
                 int free = pipeSize - inputStream.available();
                 if (free <= 0) {

src/main/java/org/perlonjava/runtime/operators/IOOperator.java

@@ -2302,13 +2302,15 @@ public static RuntimeScalar accept(int ctx, RuntimeBase... args) {
 
             if (targetGlob != null) {
                 targetGlob.setIO(clientRuntimeIO);
+                targetGlob.acceptedSocket = true;
                 MyVarCleanupStack.retainLiveIoGlobOwners(targetGlob);
                 RuntimeScalar.retainUnstashedIoForDurableSlot(newSocketHandle);
             } else {
                 // Create a new anonymous GLOB and assign it to the lvalue
                 RuntimeScalar newGlob = new RuntimeScalar();
                 newGlob.type = RuntimeScalarType.GLOBREFERENCE;
                 RuntimeGlob anonGlob = new RuntimeGlob(null).setIO(clientRuntimeIO);
+                anonGlob.acceptedSocket = true;
                 newGlob.value = anonGlob;
                 RuntimeIO.registerGlobForFdRecycling(anonGlob, clientRuntimeIO);
                 RuntimeScalar assignedHandle = newSocketHandle.set(newGlob);

src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeGlob.java

@@ -100,6 +100,9 @@ public static RuntimeArray localizedUnderscoreArrayForCurrentCall() {
     /** Number of scalar wrappers currently pointing at this anonymous IO glob. */
     public int ioHolderCount = 0;
 
+    /** True when this glob's current IO was created by accept(). */
+    public boolean acceptedSocket;
+
     /**
      * Constructor for RuntimeGlob.
      * Initializes a new instance of the RuntimeGlob class with the specified glob name.
@@ -1205,6 +1208,7 @@ public RuntimeArray getGlobArray() {
 
     public RuntimeGlob setIO(RuntimeScalar io) {
         GlobalVariable.markStashEntryVisible(this.globName);
+        acceptedSocket = false;
         // Check if the current IO is the selected handle - if so, update it
         RuntimeIO oldIO = null;
         if (this.IO.value instanceof RuntimeIO) {
@@ -1232,6 +1236,7 @@ public RuntimeGlob setIO(RuntimeScalar io) {
 
     public RuntimeGlob setIO(RuntimeIO io) {
         GlobalVariable.markStashEntryVisible(this.globName);
+        acceptedSocket = false;
         // Set the glob name in the RuntimeIO for proper stringification
         io.globName = this.globName;
         // Check if the current IO is the selected handle - if so, update it

src/main/java/org/perlonjava/runtime/runtimetypes/RuntimeScalar.java

@@ -1975,6 +1975,8 @@ private RuntimeScalar setLargeRefCounted(RuntimeScalar value) {
         boolean assignedFromArgumentAlias = RuntimeCode.isCurrentArgumentAlias(value)
                 || (!value.ioOwner
                     && RuntimeCode.isArgumentFrameActive(value.copiedFromArgumentFrame));
+        boolean assignedFromAcceptedSocketArgument = assignedFromArgumentAlias
+                && isAcceptedSocket(value);
         boolean transferDetachedIoOwner = durableIoDestination
                 && this != value
                 && !assignedFromArgumentAlias
@@ -1994,7 +1996,7 @@ && isUnstashedIoGlob(transferGlob)
                 && isUnstashedIoGlob(newGlob)
                 && hasLiveIo(newGlob)
                 && durableIoDestination
-                && !assignedFromArgumentAlias
+                && (!assignedFromArgumentAlias || assignedFromAcceptedSocketArgument)
                 && !transferDetachedIoOwner) {
             newGlob.ioHolderCount++;
         }
@@ -2099,7 +2101,7 @@ && isSocketIOHandle(oldIo.ioHandle)) {
             value.ioOwner = false;
             this.ioOwner = true;
         } else if (durableIoDestination
-                && !assignedFromArgumentAlias
+                && (!assignedFromArgumentAlias || assignedFromAcceptedSocketArgument)
                 && this != value
                 && value.type == GLOBREFERENCE
                 && value.value instanceof RuntimeGlob assignedGlob
@@ -4068,6 +4070,15 @@ private static boolean isStreamSocketIOHandle(IOHandle handle) {
         return handle instanceof SocketIO socket && !socket.isDatagramSocket();
     }
 
+    private static boolean isAcceptedSocket(RuntimeScalar scalar) {
+        if (scalar == null || scalar.type != GLOBREFERENCE
+                || !(scalar.value instanceof RuntimeGlob glob)
+                || !glob.acceptedSocket) {
+            return false;
+        }
+        return hasLiveIo(glob) && isSocketIOHandle(((RuntimeIO) glob.getIO().value).ioHandle);
+    }
+
     private static boolean isUnstashedIoGlob(RuntimeGlob glob) {
         return glob.globName == null || !GlobalVariable.existsGlobalIO(glob.globName);
     }

src/main/perl/lib/PerlOnJava/CpanPatches/IO-Async-0.805/SkipUnsupportedSocketTests.patch

@@ -9,16 +9,6 @@
  use IO::Async::OS;
 
  use Socket qw(
---- t/05notifier-loop.t.orig
-+++ t/05notifier-loop.t
-@@ -140,6 +140,11 @@
-       '$loop->remove decrements notifiers count' );
- }
-
--is_refcount( $loop, 2, '$loop has refcount 2 finally' );
-+is_refcount( $loop, 2, '$loop has refcount 2 finally' );
-
- done_testing;
 --- t/10loop-poll-io.t.orig
 +++ t/10loop-poll-io.t
 @@ -4,5 +4,6 @@

src/test/resources/unit/accepted_socket_callback_lifetime.t

@@ -0,0 +1,37 @@
+use strict;
+use warnings;
+use IO::Socket::INET;
+use Test::More;
+
+my $listener = IO::Socket::INET->new(
+    LocalAddr => '127.0.0.1',
+    LocalPort => 0,
+    Listen    => 1,
+    ReuseAddr => 1,
+);
+
+plan skip_all => "loopback listener unavailable: $!" unless $listener;
+plan tests => 3;
+
+my $client = IO::Socket::INET->new(
+    PeerAddr => '127.0.0.1',
+    PeerPort => $listener->sockport,
+) or die "client connect: $!";
+
+accept(my $accepted, $listener) or die "accept: $!";
+
+my $callback_socket;
+sub invoke_callback {
+    my ($callback, $socket) = @_;
+    $callback->($socket);
+}
+
+invoke_callback(sub { $callback_socket = $_[0] }, $accepted);
+undef $accepted;
+
+ok defined fileno($callback_socket),
+    'accepted socket remains open after callback argument assignment';
+is length(getpeername($callback_socket)), 16,
+    'accepted socket retains its IPv4 peer address';
+is length($client->sockname), 16,
+    'connected client remains valid';

src/test/resources/unit/pipe_storable_binary_roundtrip.t

@@ -0,0 +1,24 @@
+use strict;
+use warnings;
+use Test::More;
+use Storable qw(freeze thaw);
+
+# IO::Async::Channel frames Storable data over pipes. Its binary stream can
+# contain high-bit octets, which must remain single octets on an unlayered
+# pipe; otherwise the frame length and payload diverge.
+pipe(my $reader, my $writer) or die "pipe: $!";
+binmode $reader;
+binmode $writer;
+
+my $frozen = freeze([10, 20]);
+print {$writer} pack('I', length $frozen), $frozen or die "write: $!";
+close $writer or die "close writer: $!";
+
+read($reader, my $header, 4) == 4 or die "read header: $!";
+my $length = unpack('I', $header);
+read($reader, my $payload, $length) == $length or die "read payload: $!";
+
+is_deeply(thaw($payload), [10, 20],
+    'binary Storable payload round-trips through a pipe');
+
+done_testing;