Redis 流按顺序保管事件供消费

redis_lab=$(mktemp -d)
curl --fail --location --output "$redis_lab/redis.tar.gz" https://download.redis.io/releases/redis-8.10.1.tar.gz
tar -xzf "$redis_lab/redis.tar.gz" -C "$redis_lab"
make -C "$redis_lab/redis-8.10.1" build redis -j2 MALLOC=libc
export REDIS_SERVER="$redis_lab/redis-8.10.1/src/redis-server"
python3 -m venv "$redis_lab/venv"
"$redis_lab/venv/bin/python" -m pip install redis==8.1.0

from concurrent.futures import ThreadPoolExecutor

from contextlib import contextmanager
import os
from pathlib import Path
import subprocess
from tempfile import TemporaryDirectory
import time

import redis
from redis.backoff import NoBackoff
from redis.retry import Retry


@contextmanager
def instance():
    with TemporaryDirectory(prefix="redis-lab-", dir="/tmp") as directory:
        endpoint = str(Path(directory) / "redis.sock")
        process = subprocess.Popen([
            os.environ["REDIS_SERVER"], "--port", "0",
            "--unixsocket", endpoint, "--unixsocketperm", "700",
            "--save", "", "--appendonly", "no", "--dir", directory,
            "--maxmemory-policy", "noeviction",
        ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
        client = redis.Redis(unix_socket_path=endpoint, decode_responses=False,
                             socket_timeout=2, retry=Retry(NoBackoff(), 0))
        try:
            deadline = time.monotonic() + 5
            while True:
                assert process.poll() is None, "Redis exited during startup"
                try:
                    if client.ping():
                        break
                except redis.ConnectionError:
                    if time.monotonic() >= deadline:
                        raise
                    time.sleep(0.02)
            yield client
        finally:
            client.close()
            if process.poll() is None:
                process.terminate()
            try:
                _, errors = process.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
                raise
            assert process.returncode == 0, errors.decode()


def ids(entries):
    return [entry[0] for entry in entries]


with instance() as client:
    for entry_id in ("1-0", "1-1", "2-0"):
        assert client.xadd("events", {"value": entry_id}, id=entry_id) == entry_id.encode()
    try:
        client.xadd("events", {"value": "duplicate"}, id="1-1")
    except redis.ResponseError as error:
        assert "equal or smaller" in str(error)
    else:
        raise AssertionError("XADD must reject an ID below the stream's last ID")
    generated = [client.xadd("auto", {"value": "same"}) for _ in range(2)]
    assert tuple(map(int, generated[0].split(b"-"))) < tuple(map(int, generated[1].split(b"-")))
    assert client.xlen("auto") == 2
    print("explicit IDs advance; duplicate/lower ID rejected; equal payloads get distinct auto IDs")

    assert ids(client.xrange("events", "1-1", "+")) == [b"1-1", b"2-0"]
    assert ids(client.xrange("events", "(1-1", "+")) == [b"2-0"]
    first = client.xread({"events": "0-0"}, count=2)
    second = client.xread({"events": "0-0"}, count=2)
    assert first == second and ids(first[0][1]) == [b"1-0", b"1-1"]
    saved = first[0][1][-1][0]
    remaining = client.xread({"events": saved})
    assert ids(remaining[0][1]) == [b"2-0"]
    saved = remaining[0][1][-1][0]
    print("XRANGE includes its start; XREAD excludes the saved ID; independent reads replay")

    client.xadd("events", {"value": "between-calls"}, id="3-0")
    assert client.xread({"events": "$"}, block=100) == []
    assert ids(client.xread({"events": saved})[0][1]) == [b"3-0"]

    def read_live():
        with client.client() as connection:
            connection.client_setname("stream-lab-reader")
            return connection.xread({"events": "$"}, block=1000)

    with ThreadPoolExecutor(max_workers=1) as worker:
        future = worker.submit(read_live)
        deadline = time.monotonic() + 1
        while not any(
            entry["name"] == "stream-lab-reader" and "b" in entry["flags"]
            for entry in client.client_list()
        ):
            assert time.monotonic() < deadline, "reader never blocked"
            time.sleep(0.01)
        client.xadd("events", {"value": "live"}, id="4-0")
        assert ids(future.result(timeout=2)[0][1]) == [b"4-0"]
    print("$ skips existing entries; a saved ID catches up; observed blocking receives a new entry")

    assert client.xdel("events", "2-0") == 1
    assert client.xrange("events", "2-0", "2-0") == []
    assert client.xtrim("events", maxlen=2, approximate=False) == 2
    assert ids(client.xrange("events")) == [b"3-0", b"4-0"]
    assert ids(client.xread({"events": "0-0"})[0][1]) == [b"3-0", b"4-0"]
    print("XDEL removes a record; exact MAXLEN keeps the newest two; old IDs cannot restore deleted data")