fix(streaming): make ServerSession::CloseNow idempotent (one client drop no longer stalls others)

[mskwt]

Setup

• CARLA version: 0.9.15 (UE4 / ue4-dev)
• Platform: Windows (Win64, Shipping)
• Python version: 3.7 (reproduction client)

Describe the bug

tcp::ServerSession::CloseNow() is not idempotent. The inactivity-deadline timer, an async read/write erroring, and an explicit Close() can each enter it for the same session, so _on_closed() → Dispatcher::DeregisterSession → MultiStreamState::DisconnectSession runs twice.

A lone double-disconnect is harmless, but with two sessions on a stream, one session disconnecting twice evicts the other one:

_sessions = [V, A]
DisconnectSession(A)  # erase A -> [V], size==1 -> _session = V
DisconnectSession(A)  # size==1 branch -> _session = null; _sessions.clear()  -> V evicted

The world-snapshot stream (every client subscribes to it) is the everyday victim: when two clients are connected and one drops abruptly while a snapshot write is in flight, the survivor's broadcast goes silent and world.wait_for_tick() hangs.

Expected behavior

One client closing or dropping — even abruptly — should not stop other clients on the same stream from receiving data.

Steps to reproduce

Two clients on the world-snapshot stream; hard-kill one so its socket RSTs with a read and a snapshot write in flight. Reproduction script reproduce_min.py (one persistent client + one client connected and killed each round), with a fast tick so a write is in flight:

python reproduce_min.py --host 127.0.0.1 --port 3000 --rounds 30

• stock 0.9.15: the survivor stops receiving ticks on cycle 1
• with this fix: the survivor keeps ticking 30/30

The fix (2 files, ~14 lines, no API/ABI change)

Guard CloseNow() with an std::atomic_bool _is_closed so the close path runs exactly once:

void ServerSession::CloseNow(boost::system::error_code ec) {
  if (_is_closed.exchange(true)) return;   // first caller wins; the rest return
  ...
}

Other information

• Tested on Windows only (Win64, Shipping). The change is platform-independent LibCarla C++; a Linux make check from CI would be welcome.
• Scope: deeper multi-session teardown races under an extreme burst of simultaneous disconnects are out of scope — this is the minimal, low-risk fix for the common case.
• Prepared with AI assistance (see Co-Authored-By on the commit); the analysis was reviewed and the before/after results were measured on real 0.9.15 Win64 binaries.

---

This change is 

[update-docs]

Thanks for opening this pull request! The maintainers of this repository would appreciate it if you would update our CHANGELOG.md based on your changes.

[mskwt]

Minimal reproduction (self-contained)

Two ordinary clients on the world-snapshot stream; one is hard-killed (process terminated) so its socket RSTs with a read and a snapshot write in flight — the condition that re-enters CloseNow(). The other client just keeps calling wait_for_tick(). On stock 0.9.15 it is evicted and hangs; with the fix it keeps ticking.

Run (Python 3.7 + carla==0.9.15, server on port 3000; a fast tick rate makes the in-flight snapshot write reliable, e.g. uncapped t.MaxFPS):

python reproduce_min.py --host 127.0.0.1 --port 3000 --rounds 30

Verified results (same machine/install, only the server exe swapped):

# stock 0.9.15
[min] target 127.0.0.1:3000  rounds=30
[min] victim healthy and receiving ticks.
[min] round 1: victim_alive=False (frame=211, hb_age=0.1s)
*** REPRODUCED *** the persistent victim stopped receiving world ticks after 1 attacker connect/hard-kill cycle(s). The attacker's double-close evicted the victim's session from the stream.

# with the _is_closed fix
CURED: victim kept receiving ticks through all 30 attacker connect/hard-kill cycles.
[min] round 30: victim_alive=True (frame=493, hb_age=0.1s)

reproduce_min.py

#!/usr/bin/env python
"""
Minimal, faithful reproduction of the CARLA streaming ServerSession
double-close bug -- and proof that the one-line CloseNow() guard cures it.

Why this shape
--------------
The corruption is NOT triggered by a single session: `MultiStreamState::
DisconnectSession` makes a lone double-disconnect harmless (the 2nd call hits
`if (_sessions.size() == 0) return;`). It bites when there are >= 2 sessions on
a stream and ONE of them reaches DisconnectSession twice:

    _sessions = [V, A]                       # V = victim, A = attacker (size 2)
    DisconnectSession(A)  # 1st: erase A -> [V], size==1 -> _session = V
    DisconnectSession(A)  # 2nd: size==1 branch -> _session=null, _sessions.clear()
                          #      => V is wrongly evicted, its broadcast goes silent

The world-snapshot stream always has every client as a session, so this is the
everyday case. We therefore use the MINIMAL realistic setup -- one persistent
"victim" client plus one short-lived "attacker" client per round -- rather than
a flood of clients (which also overruns CARLA in unrelated ways and is not what
this fix targets).

The attacker is hard-killed so its socket RSTs with a read and a snapshot write
in flight, which is what re-enters CloseNow() on stock builds. We then check
whether the victim is still receiving world ticks.

    python reproduce_min.py --host 127.0.0.1 --port 3000 --rounds 20

REPRODUCED  -> the victim stopped receiving ticks (stock 0.9.15).
CURED       -> the victim kept ticking for every round (with the fix).
"""
import argparse, os, subprocess, sys, time
import carla

THIS = os.path.abspath(__file__)


def run_victim(host, port, hb_path):
    """Stay subscribed to the world-snapshot stream; write a heartbeat each tick."""
    client = carla.Client(host, port); client.set_timeout(10.0)
    world = client.get_world()
    while True:
        try:
            snap = world.wait_for_tick(5.0)
            with open(hb_path, "w") as f:
                f.write(f"{snap.frame} {time.time():.3f}")
        except RuntimeError:
            # broadcast went silent -> stop updating the heartbeat
            return


def run_attacker(host, port):
    """Become a 2nd session on the world stream, then block until hard-killed."""
    client = carla.Client(host, port); client.set_timeout(10.0)
    world = client.get_world()
    world.wait_for_tick(5.0)                     # ensure the stream session exists
    # keep a callback subscribed so snapshot writes are in flight at kill time
    world.on_tick(lambda s: None)
    while True:
        time.sleep(1.0)


def hb_age(hb_path):
    try:
        frame, ts = open(hb_path).read().split()
        return int(frame), time.time() - float(ts)
    except Exception:
        return None, 1e9


def victim_alive(hb_path, settle=6.0):
    """True if the heartbeat advanced within `settle` seconds."""
    f0, _ = hb_age(hb_path)
    deadline = time.time() + settle
    while time.time() < deadline:
        f1, age = hb_age(hb_path)
        if f1 is not None and f0 is not None and f1 > f0 and age < 2.0:
            return True
        time.sleep(0.3)
    return False


def controller(args):
    hb = os.path.join(os.path.dirname(THIS), "victim_hb.txt")
    try: os.remove(hb)
    except OSError: pass

    print(f"[min] target {args.host}:{args.port}  rounds={args.rounds}")
    victim = subprocess.Popen([sys.executable, THIS, "--victim", "--host", args.host,
                               "--port", str(args.port), "--hb", hb],
                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    time.sleep(5.0)
    if not victim_alive(hb, 10.0):
        print("[min] victim never became healthy; abort", file=sys.stderr)
        victim.kill(); return 2
    print("[min] victim healthy and receiving ticks.")

    try:
        for r in range(1, args.rounds + 1):
            atk = subprocess.Popen([sys.executable, THIS, "--attacker", "--host", args.host,
                                    "--port", str(args.port)],
                                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            time.sleep(args.stream_seconds)      # let it become a 2nd session w/ writes in flight
            atk.kill()                           # RST: read + snapshot write error together
            try: atk.wait(timeout=5)
            except Exception: pass

            alive = victim_alive(hb, args.settle)
            f, age = hb_age(hb)
            print(f"[min] round {r}: victim_alive={alive} (frame={f}, hb_age={age:.1f}s)")
            if not alive:
                print(f"\n*** REPRODUCED *** the persistent victim stopped receiving world "
                      f"ticks after {r} attacker connect/hard-kill cycle(s). The attacker's "
                      f"double-close evicted the victim's session from the stream.")
                return 1
        print(f"\nCURED: victim kept receiving ticks through all {args.rounds} "
              f"attacker connect/hard-kill cycles.")
        return 0
    finally:
        victim.kill()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--victim", action="store_true")
    ap.add_argument("--attacker", action="store_true")
    ap.add_argument("--hb", default="victim_hb.txt")
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=3000)
    ap.add_argument("--rounds", type=int, default=20)
    ap.add_argument("--stream-seconds", type=float, default=1.0)
    ap.add_argument("--settle", type=float, default=6.0)
    args = ap.parse_args()
    if args.victim:
        try: run_victim(args.host, args.port, args.hb)
        except Exception: pass
        return 0
    if args.attacker:
        try: run_attacker(args.host, args.port)
        except Exception: pass
        return 0
    return controller(args)


if __name__ == "__main__":
    sys.exit(main())

[Blyron]

Please follow our issues template

[mskwt]

Thanks for the review. I've updated this PR accordingly:

• Description rewritten to follow the issue template (Setup / Describe the bug / Expected behavior / Steps to reproduce / Other information).
• Commit message amended (force-pushed): dropped an internal validation note that was not meaningful out of context, and replaced it with the self-contained two-client reproduction — on stock 0.9.15 the surviving client stalls on the first connect/kill cycle; with this fix it keeps ticking 30/30.

[LuisPovedaCano]

Looks good to me.
Thanks for the contribution.

This may need a port also to UE5 branch to have it fixed there.
I will open an issue to track it.