Fix stream desync when a non-blocking read hits a partial message ("Invalid offset" disconnects) - #43

What happened

I've been building a bot with RLGym v2 (a PPO policy trained in RocketSim, run in RLBot through rlgym_compat). I play it in RLBot v5 through the GUI on Windows 11, Python 3.12, rlbot 2.0.0b55, rlbot_flatbuffers 0.19.0.

When I started a 2v2 (two copies of my bot vs two Nextos), my bots kept disconnecting about a second after spawning. Nexto was fine every time. It was random: sometimes one of my bots dropped, sometimes both, and they'd just sit there for the rest of the match. This is what the log looked like:

13:45:27  INFO RLBotServer       MatchStarter: Spawning Indonesian Coconut (index 0, team 0, aid indonesian_coconut/indonesian_coconut)
13:45:27  INFO RLBotServer       MatchStarter: Spawning Indonesian Coconut (2) (index 1, team 0, aid indonesian_coconut/indonesian_coconut)
13:45:27  INFO RLBotServer       MatchStarter: Spawning Nexto (index 2, team 1, aid rlgym/nexto)
13:45:27  INFO RLBotServer       MatchStarter: Spawning Nexto (2) (index 3, team 1, aid rlgym/nexto)
13:45:27  INFO RLBotServer FlatBuffersSession: Closing session for client 632 (w/o aid)
13:45:27    ERROR:       rlbot[    interface.py:266 -  handle_incoming_messages() ] Error while unpacking message (0 bytes): Can't make flatbuffer @ "...":
  In <[CorePacketRef]@0>::read_as_root(): Invalid offset
13:45:27    ERROR:       rlbot[    interface.py:266 -  handle_incoming_messages() ] Error while unpacking message (0 bytes): Can't make flatbuffer @ "...":
  In <[CorePacketRef]@0>::read_as_root(): Invalid offset
13:45:27  INFO RLBotServer FlatBuffersSession: Closing session for client 1064 (index 1, team 0, aid indonesian_coconut/indonesian_coconut)
13:45:27  INFO RLBotServer FlatBuffersSession: Closing session for client 1044 (index 0, team 0, aid indonesian_coconut/indonesian_coconut)

Sometimes the server also logged Session N's outbound message queue is full (… messages missed) right before closing the session.

The weird part was that the same bot worked fine on other people's machines, so at first I assumed it was my setup. I updated rlbot and rlbot_flatbuffers to the latest versions, did a full uninstall and reinstall of RLBot v5 (which pulled RLBotServer.exe v5.0.0-rc17 and rlbotgui.exe beta23), and turned off my antivirus. None of it helped. The problem turned out to be in SocketRelay itself.

What's going on

The bot manager reads from the socket like this (bot.py#L167-L185; hivemind.py and script.py do the same):

match self._game_interface.handle_incoming_messages(
    blocking=self._latest_packet is None
):

So whenever the bot has a packet waiting to be processed, which means it's falling behind, the next read is non-blocking. In interface.py:

try:
    self.socket.setblocking(blocking)
    incoming_message = self.read_message()
except BlockingIOError:
    # No incoming messages and blocking==False
    return MsgHandlingResult.NO_INCOMING_MSGS

read_message() reads a 2-byte length and then _read_exact(size). The except BlockingIOError assumes the error means "nothing to read", but it can also be raised partway through a message:

1. The length header and part of the payload have arrived, but the rest is still on its way.
2. The header read succeeds, maybe a few recv_into calls in _read_exact succeed too, then the next one raises BlockingIOError.
3. handle_incoming_messages returns NO_INCOMING_MSGS, and the bytes it already read are thrown away.
4. The next call starts in the middle of that payload and reads two random payload bytes as the length. FlatBuffers payloads are full of zero bytes, so you often get a length of 0. _read_exact(0) returns b"", CorePacket.unpack(b"") raises InvalidFlatbuffer, and the relay returns TERMINATED. That's the Error while unpacking message (0 bytes) ... Invalid offset above. A non-zero garbage length just pushes the reader further out of step until it hits one of those.

The same thing happens if only one of the two header bytes has arrived: the stream ends up shifted by one byte. In my test that case doesn't even log an error. Messages just stop arriving.

It's not the server sending empty messages. SpecStreamWriter.Write refuses to send an empty payload, so a 0-byte message can only come from the client losing its place in the stream.

Why it mostly hits at spawn, and only on some machines

The non-blocking read only happens while the bot is behind. At spawn:

• GamePackets start coming in at 240 Hz.
• With wants_ball_predictions=True (the default), each BallPrediction with 720 slices is about 37.5 KB, compared with about 1.1 KB for a 2v2 GamePacket. Big messages are much more likely to be only partly there when you read.
• The first few get_output calls are usually the slowest (model warm-up and so on), and Rocket League plus every bot process are all competing for the CPU.

My bot does real work on every packet, so on my machine it fell behind right at spawn. On faster machines it keeps up and never takes the non-blocking path, which explains why it worked for other people.

The fix

read_message() now does only its first recv non-blocking. Once any byte of a message has arrived, it reads the rest of that message blocking, so it never throws a half-read message away:

def read_message(self) -> bytes:
    assert self.socket is not None, "Socket has not been established"

    # Only this first recv may raise BlockingIOError when the socket is non-blocking.
    # Once any byte of a message has been consumed, the rest must be read blocking,
    # otherwise the consumed bytes are lost and the stream loses its framing.
    header = self.socket.recv(2)
    if len(header) == 0:
        raise EOFError
    self.socket.setblocking(True)

    if len(header) == 1:
        header += self._read_exact(1)
    size = int.from_bytes(header, "big")
    return self._read_exact(size)

• Nothing has arrived yet: recv(2) raises BlockingIOError before reading anything, so you still get NO_INCOMING_MSGS and nothing is lost.
• Part of a message has arrived: the socket blocks just long enough to read the rest of that message. handle_incoming_messages already calls setblocking(blocking) at the start of every call, so the caller's mode is back on the next call. On localhost the wait is tiny, and it beats dropping the connection.
• Only one header byte has arrived: the second byte gets read blocking.
• The server disconnects: behaviour is unchanged. An empty recv raises EOFError, which ends up as "SocketRelay disconnected unexpectedly!" / TERMINATED like before.
• _read_int was only used by read_message, so I removed it.

Since Bot, Hivemind and Script all go through this path, they all get the fix. No public API changes.

This fixed it for me. The same 2v2 that broke on almost every launch now plays normally, with all four bots.

Reproducing it

To show it deterministically, I used a socket.socketpair() to feed a SocketRelay real CorePacket frames in pieces, with a delay in between, while calling handle_incoming_messages(blocking=False) the way the managers do.

import socket
import sys
import threading
import time

sys.path.insert(0, sys.argv[1])  # path to a python-interface checkout
import rlbot.interface as interface
from rlbot import flat
from rlbot.interface import MsgHandlingResult, SocketRelay

print("rlbot loaded from:", interface.__file__)


def frame(n):
    payload = flat.CorePacket(flat.GamePacket(match_info=flat.MatchInfo(frame_num=n))).pack()
    return len(payload).to_bytes(2, "big") + payload


def make_relay():
    server, client = socket.socketpair()
    relay = SocketRelay("test")
    relay.socket = client
    relay.is_connected = True
    got = []
    relay.packet_handlers.append(lambda p: got.append(p.match_info.frame_num))
    return server, relay, got


def drain(relay, got, want):
    results = []
    for _ in range(8):
        r = relay.handle_incoming_messages(blocking=False)
        results.append(r.name)
        if r == MsgHandlingResult.TERMINATED or len(got) == want:
            break
        time.sleep(0.15)
    return results


def split_send(server, chunks):
    def run():
        for c in chunks:
            server.sendall(c)
            time.sleep(0.3)

    t = threading.Thread(target=run)
    t.start()
    time.sleep(0.1)
    return t


def case_split_payload():
    server, relay, got = make_relay()
    a, b = frame(111), frame(222)
    t = split_send(server, [a[: len(a) // 2], a[len(a) // 2 :] + b])
    results = drain(relay, got, 2)
    t.join()
    return got == [111, 222], got, results


def case_split_header():
    server, relay, got = make_relay()
    a, b = frame(111), frame(222)
    t = split_send(server, [a[:1], a[1:] + b])
    results = drain(relay, got, 2)
    t.join()
    return got == [111, 222], got, results


def case_eof():
    server, relay, got = make_relay()
    server.sendall(frame(111))
    server.close()
    time.sleep(0.1)
    results = drain(relay, got, 99)
    return got == [111] and results[-1] == "TERMINATED", got, results


for case in (case_split_payload, case_split_header, case_eof):
    ok, got, results = case()
    print(f"{'PASS' if ok else 'FAIL'} {case.__name__}: frames={got} results={results}")

On master (cc246ed):

ERROR: interface[ interface.py:266 - handle_incoming_messages() ] Error while unpacking message (12 bytes): ... Invalid offset
FAIL case_split_payload: frames=[] results=['NO_INCOMING_MSGS', 'NO_INCOMING_MSGS', 'TERMINATED']
FAIL case_split_header: frames=[] results=['NO_INCOMING_MSGS', 'NO_INCOMING_MSGS', 'NO_INCOMING_MSGS', 'NO_INCOMING_MSGS', 'NO_INCOMING_MSGS', 'NO_INCOMING_MSGS', 'NO_INCOMING_MSGS', 'NO_INCOMING_MSGS']
PASS case_eof: frames=[111] results=['MORE_MSGS_QUEUED', 'TERMINATED']

On this branch:

PASS case_split_payload: frames=[111, 222] results=['MORE_MSGS_QUEUED', 'MORE_MSGS_QUEUED']
PASS case_split_header: frames=[111, 222] results=['MORE_MSGS_QUEUED', 'MORE_MSGS_QUEUED']
PASS case_eof: frames=[111] results=['MORE_MSGS_QUEUED', 'TERMINATED']

On master, the split-payload case gives the exact same Invalid offset crash I was getting in real matches. The split-header case is the silent version: the stream is off by one byte and neither message ever gets delivered.

Happy to turn the repro into a proper test if you'd like one in the repo.