livestream: read frames with readexactly to avoid truncation

[jasoncarreira]

Description

BlinkLiveStream.recv() reads the 9-byte immis frame header and then the
payload with StreamReader.read(n):

data = await self.target_reader.read(9)
if len(data) < 9:
    _LOGGER.warning("Insufficient data for header: %d bytes, expected 9", len(data))
    break

StreamReader.read(n) returns up to n bytes — it resolves as soon as any
data is buffered. When a header or payload is split across TCP segments (very
common for the larger video payloads), read() returns a short buffer, the
len(data) < n guard treats it as EOF, and the stream is torn down mid-frame
even though the connection is perfectly healthy.

This replaces both reads with readexactly(), which waits for the full frame
and raises IncompleteReadError only on a genuine EOF:

try:
    data = await self.target_reader.readexactly(9)
except asyncio.IncompleteReadError:
    _LOGGER.debug("Target closed before a full 9-byte header")
    break

How I found it

While getting live view working through Home Assistant I built a standalone
relay that parses the immis framing from a buffer (accumulate-then-slice) and it
streamed hundreds of H.264 frames reliably. A faithful port that used read(n)
instead would intermittently abort with "Insufficient data". The difference is
exactly this partial-read handling.

Note: this is independent of the recent OAuth/2FA (202) and liveview-endpoint
work in #1227/#1228/#1229/#1231 — it's a latent framing bug that surfaces once a
live view actually connects and starts delivering video.

Checklist

• Behavior verified against a live immis stream (buffered parse streams reliably; read(n) truncates)
• tox (no test currently exercises recv() framing; happy to add one if desired)

[mback2k]

Thanks a lot, this looks good! You mentioned that you are working on HA integration as well, have you seen my work-in-progress PR which was blocked due to performance issues and the recent changes to the blink API?
home-assistant/core#160708

I would like to avoid double-work.

[fronzbot]

@jasoncarreira this looks good- just fix the failing test (probably the test is expecting a certain call and your change modified that flow) and I'll merge.

[jasoncarreira]

Thanks @mback2k! No double-work to worry about — I'm not doing a competing HA-core PR. My Blink live-view work was just a local custom_components shim on my own HA while the upstream pieces land, so #160708 is the path forward and I'm glad to help it along.

A few findings from reverse-engineering the immis stream that may help the perf/stability issues:

• This PR (readexactly) fixes intermittent stream truncation — read(n) under-reads frames split across TCP segments, which shows up as aborted/janky streams.
• Blink enforces ~1 concurrent live view per account: a 2nd overlapping session (or a rapid re-tap before the ~30s server-side release) gets rejected with a control packet (msgtype 8) then EOF. That likely explains intermittent "won't start" reports — the HA side may need to serialize/preempt live views per account.
• The v6 liveview endpoint (your Fix changed liveview API endpoints #1227) + motion_event_start_time is required now; v5 returns no usable immis server.

I've got several Blink cameras + a doorbell on HA — happy to test #160708 and give performance feedback.

[mback2k]

• the HA side may need to serialize/preempt live views per account

From previous experiences with other HA core integrations the core developers will probably ask the library to take care of that, for example via a mutex lock or something similar inside the library API instance.

[FreespiritD]

I hit this independently and opened #1262 before spotting this PR, sorry for the noise. Since I had already dug into the CI failure, here is what is blocking the merge.

The CI blocker

The only red check is Lint (3.10). Everything else, including pytest on 3.10 through 3.14, is green. Ruff reports four E501 Line too long errors, all in tests/test_livestream.py:

E501 Line too long (110 > 88)  tests/test_livestream.py:340
E501 Line too long (118 > 88)  tests/test_livestream.py:379
E501 Line too long (122 > 88)  tests/test_livestream.py:451
E501 Line too long (110 > 88)  tests/test_livestream.py:495

All four are the same shape, a side_effect list written on one line:

mock_reader.readexactly.side_effect = [header_data, payload_data, asyncio.IncompleteReadError(b"", 9)]

black tests/test_livestream.py wraps them and clears it. Nothing about the fix itself is wrong.

A regression test you may want

The current tests all mock readexactly directly, so they would not fail if the read/readexactly change were reverted. It is worth having one that does, because the bug is easy to reintroduce and invisible on a fast link.

The subtlety is that the payload has to arrive while recv() is already waiting on it. If you pre-buffer the whole thing into a StreamReader, read(n) returns it all in one go and the old code passes happily. I got that wrong on the first attempt.

async def test_recv_payload_split_across_segments(self, mock_resp):
    """Test that a payload split across TCP segments is reassembled."""
    header_data = bytearray([0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0xBC])
    payload_data = bytearray([0x47] + [0x00] * 187)  # 188 bytes

    # Only the first segment is buffered when recv() starts, so the payload read
    # finds fewer bytes than the header promised. The rest arrives while recv()
    # is already waiting on it.
    reader = asyncio.StreamReader()
    reader.feed_data(header_data)
    reader.feed_data(payload_data[:100])

    mock_client = mock.Mock()
    mock_client.is_closing.return_value = False
    mock_client.write = mock.Mock()
    mock_client.drain = mock.AsyncMock()

    self.livestream.target_reader = reader
    self.livestream.target_writer = mock.Mock()
    self.livestream.clients = [mock_client]

    recv_task = asyncio.create_task(self.livestream.recv())
    await asyncio.sleep(0)  # let recv() drain the buffer and block on the rest

    reader.feed_data(payload_data[100:])
    reader.feed_eof()
    await recv_task

    mock_client.write.assert_called_once_with(payload_data)

On dev this fails with the exact production symptom (Insufficient data for payload: 100 bytes, expected 188) and no data reaching the client. With this PR applied it passes. It needs import asyncio at the top of the test module.

Happy for that to be taken as-is, or I can open a PR into the branch if that is easier. Either way, thanks for the fix, it is the difference between live view working and not on a camera with a marginal signal.