Redis AOF 用速度换持久性

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 signal
import subprocess
from tempfile import TemporaryDirectory
import time

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


@contextmanager
def instance(directory, aof):
    endpoint = str(Path(directory) / "redis.sock")
    process = subprocess.Popen([
        os.environ["REDIS_SERVER"], "--port", "0",
        "--unixsocket", endpoint, "--unixsocketperm", "700",
        "--dir", directory, "--dbfilename", "dump.rdb", "--save", "",
        "--appendonly", "yes" if aof else "no", "--appendfsync", "always",
        "--appenddirname", "appendonlydir", "--appendfilename", "appendonly.aof",
        "--aof-use-rdb-preamble", "yes", "--no-appendfsync-on-rewrite", "no",
        "--auto-aof-rewrite-percentage", "0",
    ], 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, process
    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 in (0, -signal.SIGKILL), errors.decode()


def crash(process):
    assert process.poll() is None
    process.kill()
    assert process.wait(timeout=5) == -signal.SIGKILL


with TemporaryDirectory(prefix="redis-aof-", dir="/tmp") as directory:
    with instance(directory, aof=True) as (client, process):
        client.set("counter", 0)
        for expected in range(1, 5):
            assert client.incr("counter") == expected
        assert client.hset("profile", mapping={"name": "Ada", "level": "2"}) == 2
        client.set("deleted", "not retained")
        assert client.delete("deleted") == 1
        rewrites = client.info("persistence")["aof_rewrites"]
        assert client.bgrewriteaof()
        deadline = time.monotonic() + 5
        while True:
            state = client.info("persistence")
            if (state["aof_rewrites"] > rewrites
                    and not state["aof_rewrite_in_progress"]
                    and not state["aof_rewrite_scheduled"]):
                assert state["aof_last_bgrewrite_status"] == "ok"
                break
            assert time.monotonic() < deadline, "AOF rewrite did not finish"
            time.sleep(0.02)
        assert client.incr("counter") == 5
        assert client.info("persistence")["aof_last_write_status"] == "ok"
        assert not (Path(directory) / "dump.rdb").exists()
        print("rewrite completed; a later INCR returned 5; no standalone RDB")
        crash(process)

    aof_dir = Path(directory) / "appendonlydir"
    entries = []
    for line in (aof_dir / "appendonly.aof.manifest").read_text().splitlines():
        parts = line.split()
        entries.append(dict(zip(parts[::2], parts[1::2], strict=True)))
    active = [entry for entry in entries if entry["type"] in {"b", "i"}]
    assert {entry["type"] for entry in active} == {"b", "i"}
    for entry in active:
        assert (aof_dir / entry["file"]).is_file()
    base = next(aof_dir / entry["file"] for entry in active if entry["type"] == "b")
    assert base.read_bytes().startswith(b"REDIS")
    print("manifest references an RDB-format AOF base and an incremental AOF")

    with instance(directory, aof=True) as (client, process):
        assert client.get("counter") == b"5"
        assert client.hgetall("profile") == {b"name": b"Ada", b"level": b"2"}
        assert client.get("deleted") is None
        assert client.dbsize() == 2
        print("after SIGKILL and restart: counter=5, hash restored, deletion retained")