Fix send_all() silently truncating typed memoryviews after partial writes

newsfragments/3543.bugfix.rst

@@ -0,0 +1,2 @@
+Fix ``SocketStream.send_all()`` and POSIX ``FdStream.send_all()`` silently truncating typed
+memoryviews (``itemsize > 1``) when ``send()``/``os.write()`` performs a partial write.

src/trio/_highlevel_socket.py

@@ -117,11 +117,16 @@ async def send_all(self, data: bytes | bytearray | memoryview) -> None:
                             raise trio.ClosedResourceError("socket was already closed")
                         await trio.lowlevel.checkpoint()
                         return
-                    total_sent = 0
-                    while total_sent < len(data):
-                        with data[total_sent:] as remaining:
-                            sent = await self.socket.send(remaining)
-                        total_sent += sent
+                    # send() returns a byte count, so the buffer must be
+                    # sliced in bytes: cast typed views (itemsize > 1) to a
+                    # one-dimensional byte view first, otherwise a partial
+                    # write silently skips data (gh-3543).
+                    with data.cast("B") as data:
+                        total_sent = 0
+                        while total_sent < len(data):
+                            with data[total_sent:] as remaining:
+                                sent = await self.socket.send(remaining)
+                            total_sent += sent
 
     async def wait_send_all_might_not_block(self) -> None:
         with self._send_conflict_detector:

src/trio/_tests/test_highlevel_socket.py

@@ -119,6 +119,47 @@ async def receiver() -> None:
         assert await b.receive_some(10) == b""
 
 
+async def test_SocketStream_send_all_typed_memoryview(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Regression test for gh-3543: send_all() must not silently truncate a
+    # typed memoryview (itemsize > 1) when send() performs partial writes.
+    a_sock, b_sock = tsocket.socketpair()
+    with a_sock, b_sock:
+        a = SocketStream(a_sock)
+        b = SocketStream(b_sock)
+
+        # Force partial writes: at most 3 bytes per send() call.
+        real_send = a_sock.send
+
+        def limited_send(buf: bytes | bytearray | memoryview, *args: object) -> int:
+            return real_send(memoryview(buf)[:3])  # type: ignore[return-value]
+
+        monkeypatch.setattr(a_sock, "send", limited_send)
+
+        payload = bytes(range(256)) * 64
+        received = bytearray()
+
+        async def sender() -> None:
+            with memoryview(payload).cast("H") as typed:
+                assert typed.itemsize == 2
+                await a.send_all(typed)
+            await a.send_eof()
+
+        async def receiver() -> None:
+            while True:
+                chunk = await b.receive_some(65536)
+                if not chunk:
+                    break
+                received.extend(chunk)
+
+        async with _core.open_nursery() as nursery:
+            nursery.start_soon(sender)
+            nursery.start_soon(receiver)
+
+        assert bytes(received) == payload
+
+
 async def fill_stream(s: SocketStream) -> None:
     async def sender() -> None:
         while True:

src/trio/_tests/test_unix_pipes.py

@@ -66,6 +66,39 @@ async def test_send_pipe() -> None:
         os.close(r)
 
 
+async def test_send_all_typed_memoryview(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Regression test for gh-3543: FdStream.send_all() must not silently
+    # truncate a typed memoryview (itemsize > 1) on partial writes.
+    write, read = await make_pipe()
+
+    # Force partial writes: at most 3 bytes per os.write() call.
+    real_write = os.write
+
+    def limited_write(fd: int, buf: bytes | bytearray | memoryview) -> int:
+        return real_write(fd, memoryview(buf)[:3])
+
+    monkeypatch.setattr(os, "write", limited_write)
+
+    payload = bytes(range(256)) * 64
+    with memoryview(payload).cast("H") as typed:
+        assert typed.itemsize == 2
+        await write.send_all(typed)
+    # Empty views (even strided ones, which cast("B") would reject) are
+    # still accepted.
+    await write.send_all(b"")
+    await write.send_all(memoryview(b"")[::2])
+    await write.aclose()
+
+    received = bytearray()
+    while True:
+        chunk = await read.receive_some(65536)
+        if not chunk:
+            break
+        received.extend(chunk)
+    await read.aclose()
+    assert bytes(received) == payload
+
+
 async def test_receive_pipe() -> None:
     r, w = os.pipe()
     async with FdStream(r) as recv:

src/trio/_unix_pipes.py

@@ -134,23 +134,30 @@ async def send_all(self, data: bytes | bytearray | memoryview) -> None:
             if self._fd_holder.closed:
                 raise trio.ClosedResourceError("file was already closed")
             await trio.lowlevel.checkpoint()
-            length = len(data)
             # adapted from the SocketStream code
             with memoryview(data) as view:
-                sent = 0
-                while sent < length:
-                    with view[sent:] as remaining:
-                        try:
-                            sent += os.write(self._fd_holder.fd, remaining)
-                        except BlockingIOError:
-                            await trio.lowlevel.wait_writable(self._fd_holder.fd)
-                        except OSError as e:
-                            if e.errno == errno.EBADF:
-                                raise trio.ClosedResourceError(
-                                    "file was already closed",
-                                ) from None
-                            else:
-                                raise trio.BrokenResourceError from e
+                if not view:
+                    return
+                # os.write() returns a byte count, so the buffer must be
+                # sliced in bytes: cast typed views (itemsize > 1) to a
+                # one-dimensional byte view first, otherwise a partial write
+                # silently skips data (gh-3543). The empty case is handled
+                # above because cast() rejects strided views, even empty ones.
+                with view.cast("B") as view:
+                    sent = 0
+                    while sent < len(view):
+                        with view[sent:] as remaining:
+                            try:
+                                sent += os.write(self._fd_holder.fd, remaining)
+                            except BlockingIOError:
+                                await trio.lowlevel.wait_writable(self._fd_holder.fd)
+                            except OSError as e:
+                                if e.errno == errno.EBADF:
+                                    raise trio.ClosedResourceError(
+                                        "file was already closed",
+                                    ) from None
+                                else:
+                                    raise trio.BrokenResourceError from e
 
     async def wait_send_all_might_not_block(self) -> None:
         with self._send_conflict_detector: