Replacing the same object causes premature dynamic-index promotion

import argparse
import json
import os
import pathlib
import shutil
import signal
import subprocess
import tempfile
import time
import urllib.request


UUID = "00000000-0000-0000-0000-000000000007"
THRESHOLD = 3


def request(base, method, path, body=None):
    data = None if body is None else json.dumps(body).encode()
    req = urllib.request.Request(
        base + path,
        data=data,
        headers={"Content-Type": "application/json"} if data else {},
        method=method,
    )
    with urllib.request.urlopen(req, timeout=10) as response:
        raw = response.read()
        return response.status, json.loads(raw) if raw else None


def wait_ready(base, process):
    deadline = time.time() + 60
    while time.time() < deadline:
        if process.poll() is not None:
            raise RuntimeError(f"Weaviate exited with {process.returncode}")
        try:
            if request(base, "GET", "/v1/meta")[0] == 200:
                if request(base, "GET", "/v1/schema")[0] == 200:
                    return
        except Exception:
            pass
        time.sleep(0.25)
    raise TimeoutError("Weaviate did not become ready")


def hnsw_exists(data_dir, class_name):
    class_dir = data_dir / class_name.lower()
    return any(class_dir.rglob("*.hnsw.commitlog.d"))


def run_trial(binary, trial):
    port = 18400 + trial * 10
    root = pathlib.Path(tempfile.mkdtemp(prefix=f"weaviate-dynamic-{trial}-"))
    data_dir = root / "data"
    data_dir.mkdir()
    base = f"http://127.0.0.1:{port}"

    env = os.environ.copy()
    env.update({
        "ASYNC_INDEXING": "true",
        "AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED": "true",
        "PERSISTENCE_DATA_PATH": str(data_dir),
        "GRPC_PORT": str(port + 1),
        "GO_PROFILING_PORT": str(port + 2),
        "RAFT_PORT": str(port + 3),
        "RAFT_INTERNAL_RPC_PORT": str(port + 4),
        "CLUSTER_HOSTNAME": "127.0.0.1",
        "CLUSTER_GOSSIP_BIND_PORT": str(port + 5),
        "CLUSTER_DATA_BIND_PORT": str(port + 6),
        "CLUSTER_ADVERTISE_PORT": str(port + 6),
    })

    log = (root / "server.log").open("wb")
    process = subprocess.Popen(
        [str(binary), "--host", "127.0.0.1", "--port", str(port), "--scheme", "http"],
        stdout=log,
        stderr=subprocess.STDOUT,
        env=env,
    )

    try:
        wait_ready(base, process)

        positive = f"DynReplace{trial}"
        control = f"DynControl{trial}"
        config = {
            "vectorizer": "none",
            "vectorIndexType": "dynamic",
            "vectorIndexConfig": {"threshold": THRESHOLD},
            "properties": [],
        }

        request(base, "POST", "/v1/schema", {"class": positive, **config})
        request(base, "POST", "/v1/schema", {"class": control, **config})

        request(base, "POST", "/v1/objects", {
            "class": positive,
            "id": UUID,
            "vector": [1, 0],
        })

        for value in range(2, 8):
            request(base, "PUT", f"/v1/objects/{positive}/{UUID}", {
                "class": positive,
                "id": UUID,
                "vector": [value, 0],
            })

        request(base, "POST", "/v1/objects", {
            "class": control,
            "id": "00000000-0000-0000-0000-000000000008",
            "vector": [1, 0],
        })

        deadline = time.time() + 20
        positive_hnsw = False
        while time.time() < deadline:
            positive_hnsw = hnsw_exists(data_dir, positive)
            if positive_hnsw:
                break
            time.sleep(0.25)

        control_hnsw = hnsw_exists(data_dir, control)

        print(json.dumps({
            "trial": trial,
            "positive_hnsw": positive_hnsw,
            "control_hnsw": control_hnsw,
            "live_object_count": 1,
            "threshold": THRESHOLD,
        }))

        if positive_hnsw and not control_hnsw:
            return True
        if not positive_hnsw and not control_hnsw:
            return None
        raise RuntimeError("unexpected control promotion")

    finally:
        if process.poll() is None:
            process.send_signal(signal.SIGTERM)
            try:
                process.wait(timeout=10)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait()
        log.close()
        shutil.rmtree(root, ignore_errors=True)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--binary",
        default=os.environ.get("WEAVIATE_BINARY", "./weaviate-server"),
    )
    parser.add_argument("--trials", type=int, default=10)
    args = parser.parse_args()

    binary = pathlib.Path(args.binary)
    if not binary.exists():
        print(f"Missing Weaviate binary: {binary}")
        return 2

    results = []
    try:
        for trial in range(args.trials):
            result = run_trial(binary, trial)
            results.append(result)
    except Exception as exc:
        print(f"Environment or runtime failure: {exc}")
        return 3

    if all(result is True for result in results):
        print("BUG REPRODUCED: repeated replacement promoted one live object")
        return 10

    if all(result is None for result in results):
        print("NO BUG OBSERVED")
        return 0

    print("INCONCLUSIVE")
    return 3


if __name__ == "__main__":
    raise SystemExit(main())

positive_hnsw = false
control_hnsw = false
live vector count = 1
threshold = 3

positive_hnsw: true 
control_hnsw: false 
live_object_count: 1 
threshold: 3