Redis Lua 脚本执行时不被穿插

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 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()


compare_delete = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
end
return 0
"""

with instance() as client, client.client() as rival:
    client.set("lease", "owner-a")
    assert client.get("lease") == b"owner-a"
    rival.set("lease", "owner-b")
    assert client.eval(compare_delete, 1, "lease", "owner-a") == 0
    assert client.get("lease") == b"owner-b"
    assert client.eval(compare_delete, 1, "lease", "owner-b") == 1
    assert client.eval(compare_delete, 1, "lease", "owner-b") == 0
    print("compare-and-delete preserves a new owner's token and removes only a match")

    client.set("wrong-type", "string")
    fails_after_write = """
redis.call('SET', KEYS[1], 'before')
redis.call('LPUSH', KEYS[2], 'item')
redis.call('SET', KEYS[3], 'after')
return 1
"""
    try:
        client.eval(fails_after_write, 3, "written", "wrong-type", "not-reached")
    except redis.ResponseError as error:
        assert "WRONGTYPE" in str(error)
    else:
        raise AssertionError("script should fail on the string key")
    assert client.mget("written", "wrong-type", "not-reached") == [
        b"before", b"string", None,
    ]
    print("redis.call error stops the script but keeps the earlier write")

    catches_error = """
local reply = redis.pcall('LPUSH', KEYS[1], 'item')
if type(reply) == 'table' and reply.err then
    redis.call('SET', KEYS[2], 'handled')
    return reply.err
end
return reply
"""
    response = client.eval(catches_error, 2, "wrong-type", "handled-error")
    assert response.startswith(b"WRONGTYPE")
    assert client.get("handled-error") == b"handled"
    assert client.eval(catches_error, 2, "new-list", "unused-marker") == 1
    assert client.lrange("new-list", 0, -1) == [b"item"]
    assert client.get("unused-marker") is None
    print("redis.pcall exposes errors for handling and preserves successful command replies")

    digest = client.script_load(compare_delete)
    assert client.script_exists(digest) == [True]
    client.set("lease", "owner-c")
    assert client.evalsha(digest, 1, "lease", "owner-c") == 1
    client.set("lease", "owner-d")
    assert client.script_flush(sync_type="SYNC") is True
    assert client.script_exists(digest) == [False]
    try:
        client.evalsha(digest, 1, "lease", "owner-d")
    except redis.exceptions.NoScriptError:
        pass
    else:
        raise AssertionError("flushed script remained available")
    assert client.get("lease") == b"owner-d"
    assert client.eval(compare_delete, 1, "lease", "owner-d") == 1
    print("SCRIPT FLUSH causes NOSCRIPT; EVAL with retained source can run it again")