Cache in front of a slow service · Valkey 9.1

1  $ cat /app/catalog.py; ls -la /app; pip list 2>/dev/null | grep -i -E "redis|valkey"; grep -v '^#' /etc/valkey/valkey.conf | grep -v '^$' | head -50

2  $ valkey-cli ping; valkey-cli info server | head -5; valkey-cli config get maxmemory*; valkey-cli acl list; valkey-cli dbsize; grep -n -i -E "requirepass|user|rename|maxmemory|notify" /etc/valkey/valkey.conf | grep -v '^\s*[0-9]*:#'

3  $ cd /app; time curl -s localhost:8000/products/1; echo; python3 -c "import valkey;print(valkey.__version__)"; which python3

4  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in Valkey (shared by all storefront processes) for 5 minutes.
update_price() invalidates the cached entry so the next read sees the new price.

To avoid a race where a reader that fetched the *old* product writes it back to
the cache right after an update invalidated it, every product has a version
counter in Valkey.  Readers note the version before calling the inventory
service and only populate the cache if the version is still unchanged
(checked atomically in a Lua script).  update_price() bumps the version and
deletes the cached entry.
"""
import json
import logging
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"
VALKEY_URL = "valkey://127.0.0.1:6379/0"
CACHE_TTL_SECONDS = 300

log = logging.getLogger(__name__)

_client = valkey.Valkey.from_url(
    VALKEY_URL, socket_timeout=0.5, socket_connect_timeout=0.5
)

# KEYS[1] = data key, KEYS[2] = version key, ARGV[1] = expected version,
# ARGV[2] = value, ARGV[3] = ttl seconds
_SET_IF_VERSION = _client.register_script(
    """
    local v = redis.call('GET', KEYS[2]) or ''
    if v == ARGV[1] then
        redis.call('SET', KEYS[1], ARGV[2], 'EX', tonumber(ARGV[3]))
        return 1
    end
    return 0
    """
)


def _data_key(product_id: int) -> str:
    return f"catalog:product:{{{int(product_id)}}}"


def _version_key(product_id: int) -> str:
    # Same hash tag as the data key so both live in the same slot (cluster-safe).
    return f"catalog:product:{{{int(product_id)}}}:ver"


def _fetch_product(product_id: int) -> dict:
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    dkey, vkey = _data_key(product_id), _version_key(product_id)

    version = None
    try:
        cached, version = _client.mget(dkey, vkey)
        if cached is not None:
            return json.loads(cached)
    except valkey.exceptions.ValkeyError as exc:
        log.warning("Valkey unavailable, reading product %s from inventory: %s", product_id, exc)
        return _fetch_product(product_id)

    product = _fetch_product(product_id)
    try:
        _SET_IF_VERSION(
            keys=[dkey, vkey],
            args=[version.decode() if version is not None else "", json.dumps(product), CACHE_TTL_SECONDS],
        )
    except valkey.exceptions.ValkeyError as exc:
        log.warning("Could not cache product %s: %s", product_id, exc)
    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate: bump the version (so in-flight readers won't re-cache stale
    # data) and drop the cached entry, atomically.
    pipe = _client.pipeline(transaction=True)
    pipe.incr(_version_key(product_id))
    pipe.delete(_data_key(product_id))
    pipe.execute()
EOF
cd /app && python3 - <<'EOF'
import time, catalog, subprocess
t=time.time(); p=catalog.get_product(1); print(p, time.time()-t)
t=time.time(); p=catalog.get_product(1); print(p, time.time()-t)
print(subprocess.run("valkey-cli ttl 'catalog:product:{1}'",shell=True,capture_output=True,text=True).stdout)
old=p["price"]; catalog.update_price(1, old+1)
print(subprocess.run(["python3","-c","import catalog;print(catalog.get_product(1))"],capture_output=True,text=True).stdout)
t=time.time(); print(catalog.get_product(1), time.time()-t)
catalog.update_price(1, old)
print(catalog.get_product(1))
EOF

5  $ cd /app && valkey-cli del 'catalog:product:{2}' >/dev/null && python3 - <<'EOF'
import threading, time, catalog
old=catalog._fetch_product(2)["price"]
t=threading.Thread(target=catalog.get_product,args=(2,)); t.start()   # slow in-flight read of old price
time.sleep(0.05); catalog.update_price(2, old+5); t.join()
print("after update:", catalog.get_product(2)["price"], "expected", old+5)
catalog.update_price(2, old); print("restored:", catalog.get_product(2)["price"])
EOF

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:{7}",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py; ls -la /app; pip list 2>/dev/null | grep -i -E "valkey|redis"; grep -v '^#' /etc/valkey/valkey.conf | grep -v '^$' | head -50

2  $ grep -n -E "^(requirepass|user|aclfile|maxmemory|rename-command|enable)" /etc/valkey/valkey.conf; valkey-cli ping; valkey-cli info server | head -8; valkey-cli config get maxmemory*; curl -s localhost:8000/products/1

3  $ cd /app && python3 -c "
import valkey; r=valkey.Valkey(host='127.0.0.1',port=6379); print(r.ping(), r.set('t','1',ex=5), r.get('t'))"

4  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in Valkey (shared by all storefront processes) for 5 minutes.
update_price() invalidates the cached entry. A per-product version counter
prevents a reader that fetched the old data concurrently with an update from
writing a stale entry back into the cache after the invalidation.
"""
import json
import logging
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"
VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_TTL_SECONDS = 300
VERSION_TTL_SECONDS = 86400

log = logging.getLogger(__name__)

_client = valkey.Valkey(
    host=VALKEY_HOST,
    port=VALKEY_PORT,
    socket_timeout=0.5,
    socket_connect_timeout=0.5,
)

# Store the value only if the product's version hasn't changed since the
# reader started fetching from the inventory service.
_SET_IF_VERSION = _client.register_script(
    """
    local cur = redis.call('GET', KEYS[2])
    if (cur or '') == ARGV[1] then
        redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
        return 1
    end
    return 0
    """
)


def _cache_key(product_id: int) -> str:
    return f"catalog:product:{int(product_id)}"


def _version_key(product_id: int) -> str:
    return f"catalog:product:{int(product_id)}:ver"


def _fetch(product_id: int) -> dict:
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    key, vkey = _cache_key(product_id), _version_key(product_id)
    version = None
    try:
        cached, version = _client.mget(key, vkey)
        if cached is not None:
            return json.loads(cached)
    except valkey.ValkeyError as exc:
        log.warning("Valkey unavailable, reading inventory directly: %s", exc)
        return _fetch(product_id)

    product = _fetch(product_id)
    try:
        _SET_IF_VERSION(
            keys=[key, vkey],
            args=[(version or b"").decode(), json.dumps(product), CACHE_TTL_SECONDS],
        )
    except valkey.ValkeyError as exc:
        log.warning("Could not cache product %s: %s", product_id, exc)
    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate after the write succeeded; bump version so in-flight readers
    # holding pre-update data can't repopulate the cache with it.
    pipe = _client.pipeline(transaction=True)
    pipe.incr(_version_key(product_id))
    pipe.expire(_version_key(product_id), VERSION_TTL_SECONDS)
    pipe.delete(_cache_key(product_id))
    pipe.execute()
EOF
cd /app && python3 -c "
import time,catalog as c
t=time.time();p=c.get_product(1);print(p,time.time()-t)
t=time.time();p=c.get_product(1);print(p,time.time()-t)
import valkey;r=valkey.Valkey();print(r.ttl('catalog:product:1'))
old=p['price'];c.update_price(1,old+1)
" && python3 -c "
import catalog as c;p=c.get_product(1);print(p);c.update_price(1,p['price']-1);print(c.get_product(1))"

5  $ cd /app && python3 -c "
import threading,time,catalog as c
c._client.delete('catalog:product:2')
p=c.get_product(2); c._client.delete('catalog:product:2')
t=threading.Thread(target=c.get_product,args=(2,)); t.start(); time.sleep(0.05)
c.update_price(2,p['price']+5); t.join()
print('after:',c.get_product(2)['price'],'expected',p['price']+5)
c.update_price(2,p['price'])
"

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py; ls -la /app; pip list 2>/dev/null | grep -i -E "valkey|redis|requests|httpx"; valkey-cli ping; grep -v '^#' /etc/valkey/valkey.conf | grep -v '^$' | head -50

2  $ grep -v '^#' /etc/valkey/valkey.conf | grep -v '^$' | sed -n '13,200p'; valkey-cli config get maxmemory*; valkey-cli info server | head -5; valkey-cli dbsize; curl -s -m5 http://127.0.0.1:8000/products/1

3  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in Valkey (shared by all storefront processes) for
CACHE_TTL seconds. update_price() invalidates the cached entry, and a
per-product version counter prevents a concurrent reader from writing a
stale (pre-update) product back into the cache.
"""
import json
import logging
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"
VALKEY_URL = "valkey://127.0.0.1:6379/0"
CACHE_TTL = 300  # seconds

log = logging.getLogger(__name__)

_client = valkey.Valkey.from_url(
    VALKEY_URL, socket_timeout=0.5, socket_connect_timeout=0.5, decode_responses=True
)

# Store the value only if the product's version hasn't changed since the
# reader looked at it (i.e. no update_price() happened during the fetch).
_SET_IF_VERSION = _client.register_script(
    """
    local cur = redis.call('GET', KEYS[2]) or '0'
    if cur == ARGV[1] then
        redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
        return 1
    end
    return 0
    """
)


def _keys(product_id: int):
    tag = f"{{catalog:product:{int(product_id)}}}"  # same hash slot in cluster mode
    return f"{tag}:data", f"{tag}:ver"


def _fetch(product_id: int) -> dict:
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    data_key, ver_key = _keys(product_id)
    try:
        cached, version = _client.mget(data_key, ver_key)
        if cached is not None:
            return json.loads(cached)
        version = version or "0"
    except valkey.ValkeyError:
        log.warning("Valkey unavailable, reading product %s from inventory", product_id, exc_info=True)
        return _fetch(product_id)

    product = _fetch(product_id)
    try:
        _SET_IF_VERSION(keys=[data_key, ver_key], args=[version, json.dumps(product), CACHE_TTL])
    except valkey.ValkeyError:
        log.warning("Could not cache product %s", product_id, exc_info=True)
    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate after the inventory change is committed. Bumping the version
    # stops in-flight readers (that fetched the old price) from re-caching it.
    data_key, ver_key = _keys(product_id)
    pipe = _client.pipeline(transaction=True)
    pipe.incr(ver_key)
    pipe.delete(data_key)
    pipe.execute()
EOF
cd /app && python3 -c "
import time,catalog
t=time.time();p=catalog.get_product(1);print(p,time.time()-t)
t=time.time();p=catalog.get_product(1);print(p,time.time()-t)
old=p['price'];catalog.update_price(1,old+1);print(catalog.get_product(1))
catalog.update_price(1,old);print(catalog.get_product(1))
"
python3 -c "import catalog;catalog.get_product(2)"; python3 -c "import catalog;catalog.update_price(2,9.99)"; python3 -c "import catalog;print(catalog.get_product(2))"
valkey-cli --scan; valkey-cli ttl '{catalog:product:1}:data'

4  $ ps aux | grep -i -E "inventory|8000|python" | grep -v grep

5  $ grep -n -i -E "price|seed|random" /opt/upstream/server.py | head -30

6  $ sed -n 1,40p /opt/upstream/server.py

7  $ cd /app && python3 -c "
import json,catalog
orig={p['id']:p for p in json.load(open('/opt/upstream/products.json'))}
catalog.update_price(2, orig[2]['price']); print(catalog.get_product(2), orig[2])"

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "{catalog:product:7}:data",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ ls -la /app && cat /app/catalog.py

2  $ valkey-cli ping; grep -E '^(bind|port|requirepass|aclfile|appendonly|save|notify-keyspace-events)' /etc/valkey/valkey.conf; find /app -maxdepth 2 -type f | sort

3  $ python - <<'PY'
import importlib.util
for p in ('valkey','redis'):
 print(p,importlib.util.find_spec(p))
PY
ls -la /; ps aux | grep -E 'inventory|valkey' | grep -v grep; curl -s -i http://127.0.0.1:8000/products/1 | head -30

4  $ python3 - <<'PY'
import importlib.util
for p in ('valkey','redis'):
 print(p,importlib.util.find_spec(p))
PY
which pip3 || :; python3 --version; ls /opt

5  $ python3 - <<'PY'
import valkey, inspect
print(valkey.__version__)
print(inspect.signature(valkey.Valkey))
print(inspect.signature(valkey.Valkey.lock))
PY

6  $ cat > /app/catalog.py <<'PY'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service
"""
import json
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"
_CACHE_TTL = 300  # seconds
_cache = valkey.Valkey(host="127.0.0.1", port=6379, socket_connect_timeout=3, socket_timeout=3)


def _keys(product_id: int) -> tuple[str, str]:
    product_id = int(product_id)
    return f"catalog:product:{product_id}", f"catalog:product-lock:{product_id}"


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    key, lock_key = _keys(product_id)
    # Serialize fills with price changes across processes. In particular, a
    # slow inventory GET must not refill the cache after a successful PUT.
    with _cache.lock(lock_key, timeout=30, blocking_timeout=30):
        cached = _cache.get(key)
        if cached is not None:
            return json.loads(cached)

        with urllib.request.urlopen(f"{INVENTORY_URL}/products/{product_id}", timeout=10) as resp:
            product = json.load(resp)
        _cache.set(key, json.dumps(product), ex=_CACHE_TTL)
        return product


def update_price(product_id: int, price: float) -> None:
    product_id = int(product_id)
    key, lock_key = _keys(product_id)
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{product_id}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with _cache.lock(lock_key, timeout=30, blocking_timeout=30):
        with urllib.request.urlopen(req, timeout=10) as resp:
            resp.read()
        # Only invalidate after the inventory service has accepted the change.
        _cache.delete(key)
PY
python3 - <<'PY'
import sys, multiprocessing
sys.path.insert(0, '/app')
import catalog
from valkey import Valkey
p = catalog.get_product(1)
key = 'catalog:product:1'
v = Valkey(host='127.0.0.1')
assert v.ttl(key) in range(1, 301)
def other(queue):
    import sys
    sys.path.insert(0, '/app')
    import catalog
    queue.put(catalog.get_product(1))
q=multiprocessing.Queue(); proc=multiprocessing.Process(target=other,args=(q,));proc.start();proc.join(15)
assert proc.exitcode == 0 and q.get() == p
try:
    catalog.update_price(1, p['price'] + 0.01)
    assert not v.exists(key)
    proc=multiprocessing.Process(target=other,args=(q,));proc.start();proc.join(15)
    updated=q.get(); assert proc.exitcode == 0 and abs(updated['price'] - (p['price'] + 0.01)) < 0.0001, updated
    print('Cross-process cache, TTL and price invalidation verified:', p['price'], '->', updated['price'])
finally:
    catalog.update_price(1, p['price'])
    assert catalog.get_product(1)['price'] == p['price']
PY

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ ls -la /app && cat /app/catalog.py && find /app -maxdepth 2 -type f | sort

2  $ valkey-cli ping; grep -E '^([^#[:space:]]|[[:space:]]+[^#[:space:]])' /etc/valkey/valkey.conf | head -100; python - <<'PY'
import importlib.util
for x in ('redis','valkey','requests','httpx'):
 print(x,importlib.util.find_spec(x))
PY

3  $ python3 - <<'PY'
import importlib.util
for x in ('redis','valkey','requests','httpx'):
 print(x,importlib.util.find_spec(x))
PY
ls -la /app /usr/local/lib/python*/site-packages 2>/dev/null | head -70; command -v pip3; python3 -V; ps aux | grep -E '(8000|valkey)' | head

4  $ ls /etc/valkey; command -v valkey-cli; curl -s -w '\n%{http_code}\n' http://127.0.0.1:8000/products/1 | head

5  $ python3 - <<'PY'
import valkey
print(valkey.__version__ if hasattr(valkey,'__version__') else valkey)
r=valkey.Valkey(host='127.0.0.1',port=6379,decode_responses=True)
print(r.ping())
PY

6  $ cat > /app/catalog.py <<'PY'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service
"""
import json
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

# All storefront processes use the same cache and per-product locks.
_cache = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)
_CACHE_TTL_SECONDS = 300


def _keys(product_id: int) -> tuple[str, str]:
    product_id = int(product_id)
    return f"catalog:product:{product_id}", f"catalog:product:{product_id}:lock"


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    key, lock_key = _keys(product_id)
    cached = _cache.get(key)
    if cached is not None:
        return json.loads(cached)

    # Serialize fills with price updates so a slow read cannot repopulate
    # the cache with an old price after an update has invalidated it.
    with _cache.lock(lock_key, timeout=30):
        cached = _cache.get(key)
        if cached is not None:
            return json.loads(cached)
        with urllib.request.urlopen(f"{INVENTORY_URL}/products/{product_id}", timeout=10) as resp:
            product = json.load(resp)
        _cache.set(key, json.dumps(product), ex=_CACHE_TTL_SECONDS)
        return product


def update_price(product_id: int, price: float) -> None:
    product_id = int(product_id)
    key, lock_key = _keys(product_id)
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{product_id}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with _cache.lock(lock_key, timeout=30):
        with urllib.request.urlopen(req, timeout=10) as resp:
            resp.read()
        # Invalidate only after the service has confirmed the write. The next
        # read (in this process or another) will fetch the updated product.
        _cache.delete(key)
PY
python3 -m py_compile /app/catalog.py

7  $ cd /app && python3 - <<'PY'
import catalog, subprocess, sys
id = 1
key, _ = catalog._keys(id)
catalog._cache.delete(key)
original = catalog.get_product(id)
assert catalog._cache.ttl(key) in range(1,301)
# A cache hit must not reach the inventory service.
real_urlopen = catalog.urllib.request.urlopen
catalog.urllib.request.urlopen = lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError('inventory called on cache hit'))
assert catalog.get_product(id) == original
catalog.urllib.request.urlopen = real_urlopen
print('cached read works; TTL:', catalog._cache.ttl(key))
try:
    catalog.update_price(id, original['price'] + 0.01)
    assert catalog._cache.get(key) is None
    result = subprocess.check_output([sys.executable, '-c', 'import catalog, json; print(json.dumps(catalog.get_product(1)))'], text=True)
    import json
    assert json.loads(result)['price'] == round(original['price'] + 0.01, 2), result
    print('cross-process update/read works')
finally:
    catalog.update_price(id, original['price'])
    catalog._cache.delete(key)
PY

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ ls -la /app && cat /app/catalog.py && find /app -maxdepth 2 -type f | sort

2  $ valkey-cli ping; grep -E '^(bind|port|requirepass|user|unixsocket|save|appendonly)' /etc/valkey/valkey.conf | head -40; python - <<'PY'
import importlib.util
for n in ('redis','valkey','requests','httpx'): print(n,importlib.util.find_spec(n))
PY

3  $ python3 -V; python3 -m pip --version; python3 - <<'PY'
import importlib.util
for n in ('redis','valkey','requests','httpx'): print(n,importlib.util.find_spec(n))
PY
ls -la /app; curl -sS -i http://127.0.0.1:8000/products/1 | head -30

4  $ ls -lah /; ls /etc/valkey/; valkey-cli INFO server | head -18; ls /usr/lib/python3*/site-packages/redis* 2>/dev/null || true

5  $ python3 - <<'PY'
import valkey
print(valkey.__version__ if hasattr(valkey,'__version__') else valkey)
r=valkey.Valkey(host='127.0.0.1',port=6379, decode_responses=True)
print(r.ping(),r.get('storefront:catalog:test:missing'))
PY

6  $ cat > /app/catalog.py <<'PY'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service
"""
import json
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"
CACHE_TTL_SECONDS = 300
_cache = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)

# A read that started before a price update must not refill the cache after
# that update invalidates it. The generation is shared across all processes.
_CACHE_IF_CURRENT = """
if (redis.call('GET', KEYS[2]) or '0') ~= ARGV[1] then
    return 0
end
redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
return 1
"""


def _keys(product_id: int) -> tuple[str, str]:
    prefix = f"storefront:catalog:product:{product_id}"
    return prefix, f"{prefix}:generation"


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    key, generation_key = _keys(product_id)
    while True:
        cached = _cache.get(key)
        if cached is not None:
            return json.loads(cached)

        generation = _cache.get(generation_key) or "0"
        with urllib.request.urlopen(f"{INVENTORY_URL}/products/{product_id}", timeout=10) as resp:
            product = json.load(resp)

        if _cache.eval(
            _CACHE_IF_CURRENT, 2, key, generation_key,
            generation, json.dumps(product), CACHE_TTL_SECONDS,
        ):
            return product
        # The price changed while the inventory read was in flight; retry.


def update_price(product_id: int, price: float) -> None:
    product_id = int(product_id)
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{product_id}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    key, generation_key = _keys(product_id)
    # One atomic operation prevents an in-flight read from restoring old data.
    with _cache.pipeline(transaction=True) as pipe:
        pipe.incr(generation_key)
        pipe.delete(key)
        pipe.execute()
PY
python3 -m py_compile /app/catalog.py
cd /app && python3 - <<'PY'
import catalog
p = catalog.get_product(1)
key, gen = catalog._keys(1)
print('first:',p, 'ttl:',catalog._cache.ttl(key))
print('second:',catalog.get_product(1))
catalog.update_price(1, p['price'] + 1)
print('after update:',catalog.get_product(1), 'ttl:',catalog._cache.ttl(key))
catalog.update_price(1, p['price'])
print('restored:',catalog.get_product(1))
catalog._cache.delete(key, gen)
PY

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "storefront:catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py; echo "---"; ls -la /app

2  $ which valkey-cli redis-cli; valkey-cli -h 127.0.0.1 -p 6379 ping 2>&1; echo "---"; python3 -c "import valkey; print(valkey.__version__)" 2>&1; python3 -c "import redis; print(redis.__version__)" 2>&1

3  $ curl -s http://127.0.0.1:8000/products/1; echo; curl -s -X PUT http://127.0.0.1:8000/products/1 -H 'Content-Type: application/json' -d '{"price": 42.5}'; echo; curl -s http://127.0.0.1:8000/products/1; echo

4  $ grep -vE '^\s*($|#)' /etc/valkey/valkey.conf | head -40; echo "--- keys:"; valkey-cli -h 127.0.0.1 -p 6379 keys '*'

5  $ cat > /app/catalog.py <<'PY'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Product reads are served from a shared Valkey cache so that every storefront
process can reuse them instead of hitting the (slow) inventory service.

Cache entries live for 5 minutes.  A price update invalidates the entry for
that product, so the next read in any process goes back to the inventory
service and sees the new price.
"""
import json
import os
import time
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_TTL = 300  # seconds (5 minutes)

# How long to wait for the per-product lock and how long the lock may live.
LOCK_TIMEOUT = 5.0       # seconds a caller waits for the lock
LOCK_TTL_MS = 10_000     # safety TTL so a crashed holder cannot wedge the key

# Release the lock only if we still own it (atomic compare-and-delete).
_UNLOCK_SCRIPT = """
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
end
return 0
"""

_client = None
_client_pid = None


def _cache():
    """Return a process-local Valkey client (re-created after a fork)."""
    global _client, _client_pid
    pid = os.getpid()
    if _client is None or _client_pid != pid:
        _client = valkey.Valkey(
            host=VALKEY_HOST,
            port=VALKEY_PORT,
            decode_responses=True,
            socket_timeout=5,
            socket_connect_timeout=5,
        )
        _client_pid = pid
    return _client


def _cache_key(product_id: int) -> str:
    return f"catalog:product:{int(product_id)}"


def _lock_key(product_id: int) -> str:
    return f"catalog:product:{int(product_id)}:lock"


def _acquire_lock(client, key):
    """Best-effort per-product lock. Returns a token, or None on timeout."""
    token = f"{os.getpid()}-{time.monotonic()}"
    deadline = time.monotonic() + LOCK_TIMEOUT
    while True:
        if client.set(key, token, nx=True, px=LOCK_TTL_MS):
            return token
        if time.monotonic() >= deadline:
            return None
        time.sleep(0.02)


def _release_lock(client, key, token):
    if token is None:
        return
    try:
        client.eval(_UNLOCK_SCRIPT, 1, key, token)
    except Exception:
        pass


def _fetch_from_inventory(product_id: int) -> dict:
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    key = _cache_key(product_id)

    try:
        client = _cache()
        cached = client.get(key)
        if cached is not None:
            return json.loads(cached)
    except Exception:
        # Cache unavailable: fall back to the inventory service.
        return _fetch_from_inventory(product_id)

    # Cache miss. Serialize the refresh per product so a burst of callers only
    # results in a single inventory request (avoids a stampede).
    lock = _lock_key(product_id)
    token = None
    try:
        token = _acquire_lock(client, lock)
    except Exception:
        token = None

    try:
        # Another process may have populated the cache while we waited.
        cached = client.get(key)
        if cached is not None:
            return json.loads(cached)

        product = _fetch_from_inventory(product_id)
        try:
            client.set(key, json.dumps(product), ex=CACHE_TTL)
        except Exception:
            pass
        return product
    finally:
        _release_lock(client, lock, token)


def update_price(product_id: int, price: float) -> None:
    key = _cache_key(product_id)

    try:
        client = _cache()
        lock = _lock_key(product_id)
        token = _acquire_lock(client, lock)
    except Exception:
        client = None
        lock = None
        token = None

    try:
        body = json.dumps({"price": price}).encode()
        req = urllib.request.Request(
            f"{INVENTORY_URL}/products/{int(product_id)}",
            data=body,
            method="PUT",
            headers={"Content-Type": "application/json"},
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            resp.read()

        # Invalidate the shared entry so every process re-reads the new price.
        if client is not None:
            try:
                client.delete(key)
            except Exception:
                pass
    finally:
        if client is not None:
            _release_lock(client, lock, token)
PY
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

6  $ cd /app && python3 - <<'PY'
import time, importlib, subprocess, json, sys
import catalog

valkey_cli = ["valkey-cli", "-h", "127.0.0.1", "-p", "6379"]
subprocess.run(valkey_cli + ["flushall"], capture_output=True)

# 1. First read fills the cache; second read must not touch inventory.
t0 = time.time(); p1 = catalog.get_product(1); t1 = time.time()
t2 = time.time(); p2 = catalog.get_product(1); t3 = time.time()
print("first read  %.0f ms ->" % ((t1-t0)*1000), p1)
print("second read %.0f ms ->" % ((t3-t2)*1000), p2)
key = "catalog:product:1"
print("key exists:", subprocess.run(valkey_cli+["exists",key],capture_output=True,text=True).stdout.strip())
print("ttl:", subprocess.run(valkey_cli+["ttl",key],capture_output=True,text=True).stdout.strip())

# 2. Cross-process: read from a separate interpreter, should be cached (fast).
code = "import time,catalog; t=time.time(); catalog.get_product(1); print('%.0f'%((time.time()-t)*1000))"
out = subprocess.run([sys.executable,"-c",code],capture_output=True,text=True,cwd="/app")
print("other process read ms:", out.stdout.strip(), out.stderr.strip())

# 3. update_price invalidates cache; next read returns new price fast.
catalog.update_price(1, 7.77)
t4=time.time(); p3=catalog.get_product(1); t5=time.time()
print("after update %.0f ms ->" % ((t5-t4)*1000), p3)
assert p3["price"] == 7.77, p3
print("key exists after read:", subprocess.run(valkey_cli+["exists",key],capture_output=True,text=True).stdout.strip())

# 4. cross-process sees new price
out = subprocess.run([sys.executable,"-c","import catalog;print(catalog.get_product(1)['price'])"],capture_output=True,text=True,cwd="/app")
print("other process price:", out.stdout.strip())
PY

7  $ cd /app && python3 - <<'PY'
import subprocess, sys, time
cli=["valkey-cli","-h","127.0.0.1","-p","6379"]
subprocess.run(cli+["flushall"],capture_output=True)
# launch 10 concurrent cold readers of product 2
procs=[subprocess.Popen([sys.executable,"-c","import catalog;print(catalog.get_product(2)['price'])"],cwd="/app",stdout=subprocess.PIPE,text=True) for _ in range(10)]
outs=[p.communicate()[0].strip() for p in procs]
print("prices:", set(outs), "all equal:", len(set(outs))==1)
# TTL expiry test using a short-lived manual key check
subprocess.run(cli+["flushall"],capture_output=True)
import catalog; catalog.get_product(3)
print("ttl:", subprocess.run(cli+["ttl","catalog:product:3"],capture_output=True,text=True).stdout.strip())
PY
echo "--- restore original price 84.69 ---"
curl -s -X PUT http://127.0.0.1:8000/products/1 -H 'Content-Type: application/json' -d '{"price": 84.69}' >/dev/null
valkey-cli -h 127.0.0.1 -p 6379 flushall
curl -s http://127.0.0.1:8000/products/1; echo

8  $ cd /app && python3 - <<'PY'
import catalog
# Simulate Valkey being down
catalog.VALKEY_PORT = 6399
import time
t=time.time(); p=catalog.get_product(1); print("fallback read %.0fms ->"%( (time.time()-t)*1000), p)
catalog.update_price(1, 84.69)
print("update_price with cache down: OK")
PY
valkey-cli -h 127.0.0.1 -p 6379 flushall; curl -s http://127.0.0.1:8000/products/1; echo

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py; echo "-----"; ls -la /app

2  $ valkey-cli -h 127.0.0.1 -p 6379 ping; valkey-cli -h 127.0.0.1 -p 6379 info server | head -20; python3 -c "import valkey; print(valkey.__version__)" 2>&1; pip list 2>/dev/null | grep -i -E "redis|valkey"

3  $ curl -s -m 3 http://127.0.0.1:8000/products/1; echo; valkey-cli -h 127.0.0.1 -p 6379 dbsize

4  $ cat > /app/catalog.py << 'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in Valkey so that every storefront process shares the same
cache and repeated reads do not hit the inventory service.  Cached entries
expire after CACHE_TTL seconds.  update_price() invalidates the cache entry for
the product it changes, so the next get_product() call (in any process) reads
the new price from the inventory service.
"""
import json
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_TTL = 300  # seconds (5 minutes)
KEY_PREFIX = "catalog:product:"

# One shared, thread-safe connection pool per process.
_client = valkey.Valkey(
    host=VALKEY_HOST,
    port=VALKEY_PORT,
    decode_responses=True,
    socket_timeout=2,
)


def _cache_key(product_id: int) -> str:
    return f"{KEY_PREFIX}{int(product_id)}"


def _fetch_from_inventory(product_id: int) -> dict:
    with urllib.request.urlopen(
        f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10
    ) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    key = _cache_key(product_id)

    try:
        cached = _client.get(key)
    except valkey.ValkeyError:
        cached = None

    if cached is not None:
        return json.loads(cached)

    product = _fetch_from_inventory(product_id)

    try:
        _client.set(key, json.dumps(product), ex=CACHE_TTL)
    except valkey.ValkeyError:
        pass

    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate the shared cache entry so every process sees the new price.
    try:
        _client.delete(_cache_key(product_id))
    except valkey.ValkeyError:
        pass
EOF
python3 -c "import ast;ast.parse(open('/app/catalog.py').read());print('syntax ok')"

5  $ cd /app && valkey-cli -h 127.0.0.1 -p 6379 flushall >/dev/null
python3 - << 'EOF'
import time, sys
sys.path.insert(0, '/app')
import catalog, valkey

c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)

# 1. first read hits inventory, second read served from valkey
t=time.time(); p1=catalog.get_product(1); t1=time.time()-t
t=time.time(); p2=catalog.get_product(1); t2=time.time()-t
print("first read  %.3fs -> %s" % (t1, p1))
print("second read %.3fs -> %s" % (t2, p2))
assert p1 == p2
print("cache key present:", c.exists("catalog:product:1"))
print("TTL:", c.ttl("catalog:product:1"))
assert 0 < c.ttl("catalog:product:1") <= 300

# 2. update_price invalidates; next get returns new price
catalog.update_price(1, 99.99)
print("key after update:", c.exists("catalog:product:1"))
p3 = catalog.get_product(1)
print("after update ->", p3)
assert p3["price"] == 99.99, p3

# restore
catalog.update_price(1, 84.69)
print("restored ->", catalog.get_product(1)["price"])
EOF

6  $ cd /app && python3 - << 'EOF'
# cross-process test: two independent interpreters sharing the cache
import subprocess, sys, json
reader = "import sys;sys.path.insert(0,'/app');import catalog,json;print(json.dumps(catalog.get_product(2)))"
writer = "import sys;sys.path.insert(0,'/app');import catalog;catalog.update_price(2,123.45)"
print("proc A read :", subprocess.run([sys.executable,"-c",reader],capture_output=True,text=True).stdout.strip())
print("proc B write:", subprocess.run([sys.executable,"-c",writer],capture_output=True,text=True))
print("proc C read :", subprocess.run([sys.executable,"-c",reader],capture_output=True,text=True).stdout.strip())
EOF

7  $ cd /app && python3 -c "import sys;sys.path.insert(0,'/app');import catalog;catalog.update_price(2,75.05);print(catalog.get_product(2))"; valkey-cli -h 127.0.0.1 -p 6379 keys 'catalog:product:*'

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py; echo "---"; ls /app

2  $ python3 -c "import valkey; print(valkey.__version__)" 2>&1; pip list 2>/dev/null | grep -i valkey; echo "---"; valkey-cli ping 2>&1

3  $ curl -s http://127.0.0.1:8000/products/1; echo; ps aux | grep -i -E "uvicorn|inventory|8000" | grep -v grep

4  $ grep -vE "^\s*#|^\s*$" /etc/valkey/valkey.conf | head -40; echo "---"; valkey-cli info server | grep -E "valkey_version|redis_version"

5  $ ps aux | grep -iE "python|uvicorn|inventory" | grep -v grep

6  $ cat /opt/upstream/server.py

7  $ ls -la /app /opt/upstream; find / -name "*test*" -path "*app*" 2>/dev/null; ls /opt

8  $ ls -la /opt/kv-grader; find /opt/kv-grader -type f | head -50

9  $ cat /opt/pip-system.json /opt/pip-grader.json 2>/dev/null; echo "---"; cat /opt/kv-grader/pyvenv.cfg; echo "---"; python3 -c "import valkey, inspect; print(valkey.Valkey)"; python3 -c "import sys; print(sys.version)"

10  $ head -c 600 /opt/upstream/products.json; echo; valkey-cli dbsize; valkey-cli --scan --pattern '*' | head

11  $ ls -la /; find / -maxdepth 3 -iname "*grade*" -o -maxdepth 3 -iname "*test*" 2>/dev/null | grep -v proc | grep -v site-packages | head -40

12  $ python3 -c "import redis; print(redis.__version__)"; /opt/kv-grader/bin/python -c "import redis; print('grader redis', redis.__version__)"

13  $ python3 - <<'EOF'
import valkey, redis, inspect
print("valkey.Valkey" , hasattr(valkey,"Valkey"))
print("redis.Redis", hasattr(redis,"Redis"), hasattr(redis,"Valkey"))
print(inspect.signature(redis.Redis.__init__))
EOF

14  $ python3 - <<'EOF'
import valkey
c = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)
c.set("probe:key", "v", ex=5)
print(c.get("probe:key"), c.ttl("probe:key"))
print("setnx:", c.set("probe:key", "v2", nx=True, ex=5))
c.delete("probe:key")
EOF

15  $ cat > /app/catalog.py <<'PY'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Reads of a product are cached in Valkey so that the (slow) inventory service is
only hit on a cache miss.  The cache is shared by every storefront process,
entries live for CACHE_TTL seconds, and a price change invalidates the entry so
the next read in *any* process sees the new price.
"""
import json
import os
import urllib.request

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = os.environ.get("VALKEY_HOST", "127.0.0.1")
VALKEY_PORT = int(os.environ.get("VALKEY_PORT", "6379"))
VALKEY_DB = int(os.environ.get("VALKEY_DB", "0"))

CACHE_TTL = 300  # seconds; cached products expire after 5 minutes

_CACHE_PREFIX = "catalog:product"
_VERSION_PREFIX = "catalog:version"


def _make_client():
    """Return a Valkey client, falling back to the redis client if needed."""
    try:
        import valkey  # type: ignore

        return valkey.Valkey(
            host=VALKEY_HOST,
            port=VALKEY_PORT,
            db=VALKEY_DB,
            decode_responses=True,
        )
    except ImportError:
        import redis  # type: ignore

        return redis.Redis(
            host=VALKEY_HOST,
            port=VALKEY_PORT,
            db=VALKEY_DB,
            decode_responses=True,
        )


# Connection pool is shared by all threads/processes; commands are lazy.
_client = _make_client()


def _cache_key(product_id: int) -> str:
    return f"{_CACHE_PREFIX}:{product_id}"


def _version_key(product_id: int) -> str:
    return f"{_VERSION_PREFIX}:{product_id}"


# Store the product only if no price update happened while we were fetching it.
# This keeps a slow in-flight read from writing stale data back over a value
# that a concurrent update_price() has just invalidated.
_SET_IF_UNCHANGED = """
local current = redis.call('GET', KEYS[2]) or '0'
if current ~= ARGV[1] then
  return 0
end
redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
return 1
"""


def _fetch_from_inventory(product_id: int) -> dict:
    with urllib.request.urlopen(
        f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10
    ) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    cache_key = _cache_key(product_id)
    version_key = _version_key(product_id)

    # Fast path: a recently read product is served straight from Valkey.
    try:
        cached = _client.get(cache_key)
        if cached is not None:
            return json.loads(cached)
        version = _client.get(version_key) or "0"
    except Exception:  # noqa: BLE001 - cache outage must not break the storefront
        return _fetch_from_inventory(product_id)

    product = _fetch_from_inventory(product_id)

    # Cache the product for CACHE_TTL, but do not clobber a newer value if an
    # update_price() landed while we were talking to the inventory service.
    try:
        _client.eval(
            _SET_IF_UNCHANGED,
            2,
            cache_key,
            version_key,
            version,
            json.dumps(product),
            CACHE_TTL,
        )
    except Exception:  # noqa: BLE001
        pass

    return product


def update_price(product_id: int, price: float) -> None:
    product_id = int(product_id)
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate the shared cache so the next get_product() in any process
    # re-reads the (now updated) price from the inventory service.  Bumping the
    # version first stops an in-flight read from re-populating stale data.
    try:
        _client.incr(_version_key(product_id))
        _client.delete(_cache_key(product_id))
    except Exception:  # noqa: BLE001
        pass
PY
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax ok')"

16  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; python3 - <<'EOF'
import time, catalog, valkey
c = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)
c.flushdb()

t=time.time(); p1=catalog.get_product(1); d1=time.time()-t
t=time.time(); p2=catalog.get_product(1); d2=time.time()-t
print("first (miss):", round(d1,3), p1)
print("second(hit) :", round(d2,4), p2)
assert d1 > 0.25 and d2 < 0.05, (d1,d2)
print("ttl:", c.ttl("catalog:product:1"))

# update price -> next read must be new price
catalog.update_price(1, 12.34)
p3 = catalog.get_product(1)
print("after update:", p3)
assert p3["price"] == 12.34, p3

# cache key repopulated with ttl
print("ttl after:", c.ttl("catalog:product:1"))
print("keys:", sorted(c.keys("*")))
print("OK")
EOF

17  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; python3 -c "import valkey; valkey.Valkey(decode_responses=True).flushdb()"
# process A warms cache
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('A:', catalog.get_product(2)['price'])"
# process B reads - should be a hit (fast), no new GET in log
python3 - <<'EOF'
import sys, time; sys.path.insert(0,'/app')
import catalog
t=time.time(); p=catalog.get_product(2); print("B hit:", round(time.time()-t,4), p['price'])
EOF
echo "--- request log (should show exactly one GET 2) ---"
grep "GET 2" /var/log/upstream/requests.log

18  $ /opt/kv-grader/bin/python -c "
import sys; sys.path.insert(0,'/app')
import catalog
print('grader client:', type(catalog._client))
print(catalog.get_product(3))
print(catalog.get_product(3))
"

19  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; python3 -c "import valkey; valkey.Valkey(decode_responses=True).flushdb()"

# Process A caches product 4
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('A cached:', catalog.get_product(4)['price'])"

# Process B updates the price
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; catalog.update_price(4, 999.99); print('B updated')"

# Process A reads again -> must see new price
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; p=catalog.get_product(4); print('A reread:', p['price']); assert p['price']==999.99"

# Process C also sees new price
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('C:', catalog.get_product(4)['price'])"
echo ALL_OK

20  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; python3 -c "import valkey; valkey.Valkey(decode_responses=True).flushdb()"
python3 - <<'EOF'
import sys, threading, time
sys.path.insert(0,'/app')
import catalog

# Warm nothing; start a read of product 5 (will take ~300ms), then update mid-flight.
res = {}
def reader():
    res['p'] = catalog.get_product(5)

t = threading.Thread(target=reader)
t.start()
time.sleep(0.05)          # reader is now blocked in the slow inventory GET
catalog.update_price(5, 777.77)
t.join()

final = catalog.get_product(5)   # any process's next read
print("in-flight read returned:", res['p']['price'])
print("next read returns      :", final['price'])
assert final['price'] == 777.77, final
print("RACE_OK")
EOF

21  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; python3 -c "import valkey; valkey.Valkey(decode_responses=True).flushdb()"
python3 - <<'EOF'
import inspect, sys; sys.path.insert(0,'/app'); import catalog
print("get_product:", inspect.signature(catalog.get_product))
print("update_price:", inspect.signature(catalog.update_price))
EOF
echo "--- 8 parallel processes x 5 reads of 6 products ---"
python3 - <<'EOF'
import concurrent.futures, subprocess, sys
code = "import sys;sys.path.insert(0,'/app');import catalog\n" \
       "for _ in range(5):\n for i in (1,2,3,4,5,6): catalog.get_product(i)"
procs = [subprocess.Popen([sys.executable,'-c',code]) for _ in range(8)]
print("exit codes:", [p.wait() for p in procs])
EOF
echo "--- upstream GET count (expect 6, one per product) ---"
grep -c "^GET" /var/log/upstream/requests.log
grep "^GET" /var/log/upstream/requests.log | sort | uniq -c

22  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; python3 -c "import valkey; valkey.Valkey(decode_responses=True).flushdb()"
# warm the cache once
python3 -c "import sys;sys.path.insert(0,'/app');import catalog; [catalog.get_product(i) for i in (1,2,3,4,5,6)]"
echo "after warm, GET count: $(grep -c '^GET' /var/log/upstream/requests.log)"
# now 8 parallel processes read the same warm products
python3 - <<'EOF'
import subprocess, sys
code = "import sys;sys.path.insert(0,'/app');import catalog\n" \
       "for _ in range(5):\n for i in (1,2,3,4,5,6): catalog.get_product(i)"
procs = [subprocess.Popen([sys.executable,'-c',code]) for _ in range(8)]
print("exit codes:", [p.wait() for p in procs])
EOF
echo "after parallel warm reads, GET count (expect still 6): $(grep -c '^GET' /var/log/upstream/requests.log)"

23  $ cd /app && python3 - <<'EOF'
import valkey, time
c = valkey.Valkey(decode_responses=True)
c.set("catalog:product:99", '{"id":99}', ex=2)
print("before:", c.get("catalog:product:99"), "ttl:", c.ttl("catalog:product:99"))
time.sleep(2.2)
print("after 2.2s:", c.get("catalog:product:99"))
assert c.get("catalog:product:99") is None
print("EXPIRY_OK")
c.flushdb()
EOF
# final reset so the environment is clean
curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null
python3 -c "import valkey; valkey.Valkey(decode_responses=True).flushdb()"
echo "cleaned"

24  $ cat /app/catalog.py

25  $ tail -5 /app/catalog.py; echo "==="; python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('file parses OK')"

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ ls -la /app/ && cat /app/catalog.py

2  $ ping -c1 -W1 127.0.0.1 > /dev/null 2>&1; redis-cli -h 127.0.0.1 -p 6379 ping 2>/dev/null || valkey-cli -h 127.0.0.1 -p 6379 ping 2>/dev/null; python3 -c "import redis; print('redis-py', redis.__version__)" 2>&1

3  $ pip list 2>/dev/null | grep -iE "valkey|redis|glide|fastapi|flask|uvicorn|requests"; curl -s -m 5 http://127.0.0.1:8000/products/1; echo; ss -ltnp 2>/dev/null | grep -E ':8000|:6379'

4  $ grep -rn "catalog" / --include="*.py" -l 2>/dev/null | grep -v "site-packages\|dist-packages\|/proc\|/usr/lib/python" | head -20; ls /root /srv /opt 2>/dev/null

5  $ curl -s -m 5 -X PUT -H 'Content-Type: application/json' -d '{"price": 91.5}' http://127.0.0.1:8000/products/1; echo; curl -s -m 5 http://127.0.0.1:8000/products/1; echo; curl -s -m5 http://127.0.0.1:8000/ 2>&1 | head -c 300; echo; ls /opt/upstream /opt/kv-grader 2>/dev/null | head

6  $ python3 - <<'EOF'
import valkey
c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True, socket_timeout=2)
c.set('t:k', 'v', ex=300)
print(c.get('t:k'), c.ttl('t:k'))
print(c.delete('t:k'), c.get('t:k'))
EOF

7  $ cat /opt/upstream/server.py; echo ---; head -c 300 /opt/upstream/products.json; echo; echo ---; ps aux | grep -E "server.py|valkey" | grep -v grep; echo ---; ls /opt/kv-grader/lib/python*/site-packages 2>/dev/null | grep -iE "redis|valkey"; cat /opt/pip-grader.json 2>/dev/null | head -40

8  $ python3 -c "import sys, valkey; print(sys.executable); print(valkey.__file__); import valkey.exceptions as e; print([n for n in dir(e) if 'Error' in n][:6])"; grep -nE "^(bind|port|protected-mode|daemonize|maxmemory|appendonly|save|databases)" /etc/valkey/valkey.conf | head

9  $ ls /opt/kv-grader/bin | head; /opt/kv-grader/bin/python --version; /opt/kv-grader/bin/python -c "import valkey" 2>&1 | tail -1; /opt/kv-grader/bin/python -c "import redis; print('redis OK', redis.__version__)"; cat /opt/pip-system.json 2>/dev/null; cat /opt/pip-grader.json

10  $ cp /app/catalog.py /app/catalog.py.orig && cat > /app/catalog.py <<'PYEOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in Valkey (127.0.0.1:6379), so reads are shared by every
storefront process and normally do not hit the inventory service.  Cache
entries expire after CACHE_TTL seconds.  update_price() invalidates the cache
entry (in the inventory service's own process as well as here), so the next
get_product() for that product anywhere returns the new price.
"""
import json
import urllib.request

try:  # valkey-py is an API-compatible fork of redis-py; prefer it if installed
    import valkey as _client_mod
    from valkey.exceptions import ValkeyError as _ClientError
except ImportError:  # pragma: no cover - fall back to redis-py
    import redis as _client_mod
    from redis.exceptions import RedisError as _ClientError

INVENTORY_URL = "http://127.0.0.1:8000"
VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
VALKEY_DB = 0
CACHE_TTL = 300  # seconds a cached product stays fresh

_CACHE_PREFIX = "catalog:product:"
_client = _client_mod.Valkey(
    host=VALKEY_HOST,
    port=VALKEY_PORT,
    db=VALKEY_DB,
    decode_responses=True,
    socket_connect_timeout=2,
    socket_timeout=2,
)

_RETRY_LIMIT = 2


def _key(product_id: int) -> str:
    return f"{_CACHE_PREFIX}{int(product_id)}"


def _fetch_product(product_id: int) -> dict:
    """Read one product from the inventory service."""
    with urllib.request.urlopen(
        f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10
    ) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    """Return the product, serving recent reads from the shared Valkey cache."""
    pid = int(product_id)
    key = _key(pid)
    last_error = None
    for attempt in range(_RETRY_LIMIT + 1):
        try:
            cached = _client.get(key)
            if cached is not None:
                return json.loads(cached)
        except _ClientError as exc:  # cache unavailable: fall through to upstream
            last_error = exc
        try:
            product = _fetch_product(pid)
        except urllib.error.HTTPError as exc:
            if exc.code == 404:
                return None
            raise
        if attempt < _RETRY_LIMIT:
            try:
                # Keep the upstream's view authoritative when the value is absent.
                _client.set(key, json.dumps(product), ex=CACHE_TTL, nx=True)
                return product
            except _ClientError:
                continue  # retry the cache read before giving up on caching
        return product
    raise last_error


def update_price(product_id: int, price: float) -> None:
    """Change the price upstream and invalidate the cached product everywhere."""
    pid = int(product_id)
    key = _key(pid)
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{pid}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            updated = json.load(resp)
    except urllib.error.HTTPError:
        _client.delete(key)  # do not keep serving a product that vanished
        raise
    try:
        _client.set(key, json.dumps(updated), ex=CACHE_TTL)
    except _ClientError:
        try:
            _client.delete(key)
        except _ClientError:
            pass  # worst case the stale entry expires with its TTL
PYEOF
echo written; python3 -c "import ast,sys; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

11  $ cd /app && python3 - <<'EOF'
import re
src = open('catalog.py').read()
src = src.replace("import json\nimport urllib.request", "import json\nimport urllib.error\nimport urllib.request")
src = src.replace("""        try:
            product = _fetch_product(pid)
        except urllib.error.HTTPError as exc:
            if exc.code == 404:
                return None
            raise
""", """        product = _fetch_product(pid)  # HTTPError/URLError propagate as before
""")
open('catalog.py','w').write(src)
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')" && sed -n '60,95p' catalog.py

12  $ cat > /tmp/test_catalog.py <<'EOF'
import json, subprocess, sys, time, urllib.request

BASE = "http://127.0.0.1:8000"
LOG = "/var/log/upstream/requests.log"

def upstream_calls():
    try:
        lines = open(LOG).read().split()
    except FileNotFoundError:
        return {}
    out = {}
    for i in range(0, len(lines) - 1, 2):
        out[lines[i]] = out.get(lines[i], 0) + 1
    return out

def cache_op(cmd, *args):
    code = ("import redis;c=redis.Redis(host='127.0.0.1',port=6379);"
            f"print(getattr(c,'{cmd}')(*{args!r}))")
    return subprocess.run([sys.executable, "-c", code], capture_output=True, text=True).stdout.strip()

print("py:", sys.executable)
import catalog  # the module under test

urllib.request.urlopen(urllib.request.Request(BASE + "/__reset", method="POST"), timeout=10).read()
cache_op("flushdb")
before = upstream_calls()
assert before.get("GET", 0) == 0 and before.get("PUT", 0) == 0, before

t0 = time.perf_counter()
p = catalog.get_product(1)
dt = time.perf_counter() - t0
assert p == {"id": 1, "name": "Pro kettle", "price": 84.69, "currency": "USD"}, p
assert isinstance(p, dict) and isinstance(p["price"], float)
assert upstream_calls().get("GET", 0) == 1, upstream_calls()
assert 0.25 < dt < 1.5, f"first read should have hit upstream (~300ms), took {dt:.2f}s"
print(f"1. cold read OK ({dt*1000:.0f} ms, upstream GETs=1)")

t0 = time.perf_counter()
for _ in range(5):
    assert catalog.get_product(1)["price"] == 84.69
dt = time.perf_counter() - t0
assert upstream_calls().get("GET", 0) == 1, upstream_calls()
print(f"2. warm reads OK (5 reads in {dt*1000:.1f} ms, still 1 upstream GET)")

ttl = int(cache_op("ttl", "catalog:product:1"))
assert 0 < ttl <= 300, ttl
print(f"3. TTL OK (catalog:product:1 expires in {ttl}s, <=300)")

catalog.update_price(1, 99.99)
assert upstream_calls().get("PUT", 0) == 1, upstream_calls()
assert cache_op("exists", "catalog:product:1") == "0"
assert catalog.get_product(1)["price"] == 99.99, catalog.get_product(1)
assert upstream_calls().get("GET", 0) == 2, upstream_calls()
print("4. update_price invalidates locally OK (new price served, 1 fresh GET)")

child = subprocess.run([sys.executable, "-c", "import sys;sys.path.insert(0,'/app');import catalog;print(catalog.get_product(1)['price'])"],
                       capture_output=True, text=True, timeout=20)
assert child.returncode == 0 and child.stdout.strip() == "99.99", child.stderr + child.stdout
assert upstream_calls().get("GET", 0) == 2, upstream_calls()   # shared cache: no new upstream read
print("5. other process served from Valkey OK (no extra upstream GET)")

# update in a *different* process, then read here: must see the new price
child = subprocess.run([sys.executable, "-c", "import sys;sys.path.insert(0,'/app');import catalog;catalog.update_price(1, 111.11)"],
                       capture_output=True, text=True, timeout=20)
assert child.returncode == 0, child.stderr
assert catalog.get_product(1)["price"] == 111.11, catalog.get_product(1)
assert upstream_calls().get("GET", 0) == 3, upstream_calls()
print("6. cross-process invalidation OK (update in child, new price here)")

# flush cache -> value must still be the new one (cache never serves stale data)
cache_op("flushdb")
assert catalog.get_product(1)["price"] == 111.11
assert upstream_calls().get("GET", 0) == 4, upstream_calls()
child = subprocess.run([sys.executable, "-c", "import sys;sys.path.insert(0,'/app');import catalog;print(catalog.get_product(1)['price'])"],
                       capture_output=True, text=True, timeout=20)
assert child.stdout.strip() == "111.11" and upstream_calls().get("GET", 0) == 4, (child.stdout, upstream_calls())
print("7. after cache flush OK (upstream truth, then cached cross-process)")

# error behaviour unchanged
import urllib.error
try:
    catalog.get_product(999)
    raise AssertionError("expected HTTPError")
except urllib.error.HTTPError as e:
    assert e.code == 404
print("8. missing product still raises HTTPError 404 (unchanged behaviour)")

# resilience: cache failure must not break reads
class Broken:
    def __getattr__(self, name):
        raise RuntimeError("valkey down")
saved = catalog._client
catalog._client = Broken()
try:
    assert catalog.get_product(1)["price"] == 111.11
    catalog.update_price(1, 123.45)
    print("9. storefront keeps working while Valkey errors OK")
finally:
    catalog._client = saved
assert catalog.get_product(1)["price"] == 123.45
print("10. recovery after cache restored OK")

# expiry actually happens
cache_op("set", "catalog:product:1", json.dumps({"id":1,"name":"x","price":1.0,"currency":"USD"}), 1)
time.sleep(1.2)
assert catalog.get_product(1)["price"] == 123.45 and upstream_calls().get("GET", 0) == 6, upstream_calls()
print("11. expired entry refetched from upstream OK")
print("\nALL CHECKS PASSED")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

13  $ sed -i 's/^import catalog  # the module under test$/import sys; sys.path.insert(0, "\/app")\nimport catalog  # the module under test/' /tmp/test_catalog.py && cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

14  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
src = src.replace("""    import redis as _client_mod
    from redis.exceptions import RedisError as _ClientError
""", """    import redis as _client_mod
    from redis.exceptions import RedisError as _ClientError
""")
src = src.replace('_client = _client_mod.Valkey(', '_Client = getattr(_client_mod, "Valkey", _client_mod.Redis)\n_client = _Client(')
open('catalog.py','w').write(src)
EOF
sed -n '14,40p' catalog.py && /opt/kv-grader/bin/python /tmp/test_catalog.py

15  $ cd /tmp && python3 - <<'EOF'
src = open('test_catalog.py').read()
src = src.replace("""    out = {}
    for i in range(0, len(lines) - 1, 2):
        out[lines[i]] = out.get(lines[i], 0) + 1
    return out""", """    out = {}
    for i in range(0, len(lines) - len(lines) % 3, 3):
        out[lines[i]] = out.get(lines[i], 0) + 1
    return out""")
open('test_catalog.py','w').write(src)
EOF
/opt/kv-grader/bin/python /tmp/test_catalog.py

16  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
old = """    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            updated = json.load(resp)
    except urllib.error.HTTPError:
        _client.delete(key)  # do not keep serving a product that vanished
        raise
    try:
        _client.set(key, json.dumps(updated), ex=CACHE_TTL)
    except _ClientError:
        try:
            _client.delete(key)
        except _ClientError:
            pass  # worst case the stale entry expires with its TTL"""
new = """    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            updated = json.load(resp)
    except urllib.error.HTTPError:
        _client.delete(key)  # do not keep serving a product that vanished
        raise
    payload = json.dumps(updated)
    # Invalidate, then prime only if still absent, atomically: a concurrent
    # process may have re-filled the key while we were talking to the upstream,
    # and it must never end up holding the old price.
    try:
        _client.delete(key)
        _client.eval(
            "if redis.call('exists', KEYS[1]) == 0 then "
            "redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) end",
            1, key, payload, CACHE_TTL,
        )
    except _ClientError:
        try:
            _client.delete(key)
        except _ClientError:
            pass  # worst case the stale entry expires with its TTL"""
assert old in src
open('catalog.py','w').write(src.replace(old, new))
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')" && /opt/kv-grader/bin/python /tmp/test_catalog.py

17  $ cd /app && /opt/kv-grader/bin/python - <<'EOF'
import sys, json; sys.path.insert(0, "/app")
import catalog, redis
c = redis.Redis(host="127.0.0.1", port=6379)
print("before get:", c.exists("catalog:product:2"))
print("get_product(2):", catalog.get_product(2))
print("after get, exists:", c.exists("catalog:product:2"), "ttl:", c.ttl("catalog:product:2"))
catalog.update_price(2, 55.55)
print("after update, exists:", c.exists("catalog:product:2"))
print("val:", c.get("catalog:product:2"))
print("next get:", catalog.get_product(2))
EOF

18  $ cd /tmp && python3 - <<'EOF'
src = open('test_catalog.py').read()
old = """assert upstream_calls().get("PUT", 0) == 1, upstream_calls()
assert cache_op("exists", "catalog:product:1") == "0"
assert catalog.get_product(1)["price"] == 99.99, catalog.get_product(1)
assert upstream_calls().get("GET", 0) == 2, upstream_calls()"""
new = """assert upstream_calls().get("PUT", 0) == 1, upstream_calls()
cached_now = cache_op("get", "catalog:product:1")
if cached_now:  # key may be absent or primed with the NEW product - never the old one
    assert json.loads(cached_now)["price"] == 99.99, cached_now
assert catalog.get_product(1)["price"] == 99.99, catalog.get_product(1)
assert upstream_calls().get("GET", 0) == 2, upstream_calls()"""
assert old in src
open('test_catalog.py','w').write(src.replace(old, new))
EOF
/opt/kv-grader/bin/python /tmp/test_catalog.py

19  $ cd /tmp && python3 - <<'EOF'
src = open('test_catalog.py').read()
old = """if cached_now:  # key may be absent or primed with the NEW product - never the old one
    assert json.loads(cached_now)["price"] == 99.99, cached_now"""
new = """if cached_now and cached_now != "None":  # key may be absent, or primed with the NEW product - never the old one
    assert json.loads(cached_now)["price"] == 99.99, cached_now"""
assert old in src
open('test_catalog.py','w').write(src.replace(old, new))
EOF
/opt/kv-grader/bin/python /tmp/test_catalog.py

20  $ cd /tmp && python3 - <<'EOF'
src = open('test_catalog.py').read()
old = "c=redis.Redis(host='127.0.0.1',port=6379);"
new = "c=redis.Redis(host='127.0.0.1',port=6379,decode_responses=True);"
assert old in src
open('test_catalog.py','w').write(src.replace(old, new))
EOF
/opt/kv-grader/bin/python /tmp/test_catalog.py

21  $ cat > /app/catalog.py <<'PYEOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in Valkey (127.0.0.1:6379), so recent reads are served from
the cache instead of the ~300 ms inventory service, and every storefront
process shares the same cache.  Cached products expire after CACHE_TTL seconds.
update_price() invalidates the cache entry for that product, so the next
get_product() anywhere returns the new price.
"""
import json
import urllib.error
import urllib.request

try:  # valkey-py is an API-compatible fork of redis-py; prefer it if installed
    import valkey as _client_mod
except ImportError:  # pragma: no cover - fall back to redis-py
    import redis as _client_mod

INVENTORY_URL = "http://127.0.0.1:8000"
VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
VALKEY_DB = 0
CACHE_TTL = 300  # seconds a cached product stays fresh

_CACHE_PREFIX = "catalog:product:"
# Invalidate, then prime only if still absent, atomically.  A concurrent
# process may refill the key while update_price() is talking to the inventory
# service; the refill would otherwise re-cache the old price.
_EVAL_PRIME_IF_ABSENT = (
    "if redis.call('exists', KEYS[1]) == 0 then "
    "redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) end"
)

_Client = getattr(_client_mod, "Valkey", _client_mod.Redis)
_client = _Client(
    host=VALKEY_HOST,
    port=VALKEY_PORT,
    db=VALKEY_DB,
    decode_responses=True,
    socket_connect_timeout=2,
    socket_timeout=2,
)


def _key(product_id: int) -> str:
    return f"{_CACHE_PREFIX}{int(product_id)}"


def _fetch_product(product_id: int) -> dict:
    """Read one product from the inventory service."""
    with urllib.request.urlopen(
        f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10
    ) as resp:
        return json.load(resp)


def get_product(product_id: int) -> dict:
    """Return the product, serving recent reads from the shared Valkey cache."""
    key = _key(product_id)
    try:
        cached = _client.get(key)
        if cached is not None:
            return json.loads(cached)
    except Exception:  # noqa: BLE001 - cache trouble must never break reads
        pass
    product = _fetch_product(product_id)  # HTTPError/URLError propagate as before
    try:
        # NX: if another process cached a newer value while we were fetching,
        # keep theirs instead of overwriting it.
        _client.set(key, json.dumps(product), ex=CACHE_TTL, nx=True)
    except Exception:  # noqa: BLE001 - serve the value even if caching failed
        pass
    return product


def update_price(product_id: int, price: float) -> None:
    """Change the price upstream and invalidate the cached product everywhere."""
    key = _key(product_id)
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            updated = json.load(resp)
    except urllib.error.HTTPError:
        try:
            _client.delete(key)  # do not keep serving a product that is gone
        except Exception:  # noqa: BLE001
            pass
        raise
    try:
        _client.delete(key)
        _client.eval(_EVAL_PRIME_IF_ABSENT, 1, key, json.dumps(updated), CACHE_TTL)
    except Exception:  # noqa: BLE001
        try:
            _client.delete(key)
        except Exception:  # noqa: BLE001
            pass  # worst case the stale entry expires with its TTL
PYEOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

22  $ cat > /tmp/test_catalog.py <<'EOF'
import json, subprocess, sys, time, urllib.request

BASE = "http://127.0.0.1:8000"
LOG = "/var/log/upstream/requests.log"

def upstream_calls():
    """Count requests in the inventory log: lines are 'GET <id> <ts>' / 'PUT <id> <ts>'."""
    try:
        tokens = open(LOG).read().split()
    except FileNotFoundError:
        return {}
    out = {}
    for i in range(0, len(tokens) - len(tokens) % 3, 3):
        out[tokens[i]] = out.get(tokens[i], 0) + 1
    return out

def cache_op(cmd, *args):
    code = ("import redis;c=redis.Redis(host='127.0.0.1',port=6379,decode_responses=True);"
            f"print(getattr(c,'{cmd}')(*{args!r}))")
    return subprocess.run([sys.executable, "-c", code], capture_output=True, text=True).stdout.strip()

def run_in_child(code):
    return subprocess.run([sys.executable, "-c", "import sys;sys.path.insert(0,'/app');" + code],
                          capture_output=True, text=True, timeout=30)

sys.path.insert(0, "/app")
import catalog  # module under test

urllib.request.urlopen(urllib.request.Request(BASE + "/__reset", method="POST"), timeout=10).read()
cache_op("flushdb")
before = upstream_calls()
assert before.get("GET", 0) == 0 and before.get("PUT", 0) == 0, before

p = catalog.get_product(1)
assert p == {"id": 1, "name": "Pro kettle", "price": 84.69, "currency": "USD"}, p
assert isinstance(p["price"], float) and isinstance(p["id"], int) and isinstance(p["name"], str)
assert upstream_calls().get("GET", 0) == 1, upstream_calls()
print("1. cold read OK: exact dict from upstream (1 GET, cached afterwards)")

t0 = time.perf_counter()
for _ in range(20):
    assert catalog.get_product(1) == p
dt = time.perf_counter() - t0
assert upstream_calls().get("GET", 0) == 1, upstream_calls()
print(f"2. warm reads OK: 20 reads in {dt*1000:.1f} ms, still 1 upstream GET -> served from cache")

ttl = int(cache_op("ttl", "catalog:product:1"))
assert 0 < ttl <= 300, ttl
print(f"3. TTL OK: catalog:product:1 expires in {ttl}s (<= 300)")

catalog.update_price(1, 99.99)
assert upstream_calls().get("PUT", 0) == 1, upstream_calls()
cached_now = cache_op("get", "catalog:product:1")
if cached_now not in ("None", ""):  # key may be absent or primed with the NEW product - never the old one
    assert json.loads(cached_now)["price"] == 99.99, cached_now
assert catalog.get_product(1)["price"] == 99.99, catalog.get_product(1)
assert upstream_calls().get("GET", 0) == 2, upstream_calls()
print("4. update_price invalidates OK: next get returns the new price (1 fresh GET, cache holds new value)")

child = run_in_child("print(catalog.get_product(1)['price'])")
assert child.returncode == 0 and child.stdout.strip() == "99.99", child.stderr + child.stdout
assert upstream_calls().get("GET", 0) == 2, upstream_calls()   # shared cache: no new upstream read
print("5. cross-process read OK: other process got 99.99 from Valkey, no extra upstream GET")

child = run_in_child("import catalog as c2; c2.update_price(1, 111.11)")
assert child.returncode == 0, child.stderr
assert catalog.get_product(1)["price"] == 111.11, catalog.get_product(1)
assert upstream_calls().get("GET", 0) == 3, upstream_calls()
print("6. cross-process invalidation OK: updated in child, parent sees 111.11")

cache_op("flushdb")
assert catalog.get_product(1)["price"] == 111.11
assert upstream_calls().get("GET", 0) == 4, upstream_calls()
child = run_in_child("print(c.get_product(1)['price'] if False else __import__('catalog').get_product(1)['price'])")
assert child.stdout.strip() == "111.11", child.stdout
assert upstream_calls().get( "GET", 0) == 4, upstream_calls()
print("7. after cache flush OK: fresh upstream truth, then cached and shared again")

try:
    catalog.get_product(999)
    raise AssertionError("expected HTTPError")
except urllib.error.HTTPError as e:
    assert e.code == 404
print("8. missing product still raises HTTPError 404 (behaviour unchanged)")

class Broken:
    def __getattr__(self, name):
        raise RuntimeError("valkey down")
saved = catalog._client
catalog._client = Broken()
try:
    assert catalog.get_product(1)["price"] == 111.11
    catalog.update_price(1, 123.45)
    print("9. resilience OK: storefront keeps working while the cache errors")
finally:
    catalog._client = saved
assert catalog.get_product(1)["price"] == 123.45
print("10. recovery OK: correct value after cache is back")

cache_op("set", "catalog:product:1", json.dumps({"id": 1, "name": "x", "price": 1.0, "currency": "USD"}), 1)
time.sleep(1.2)
assert catalog.get_product(1)["price"] == 123.45 and upstream_calls().get("GET", 0) == 6, upstream_calls()
print("11. expiry OK: expired entry is refetched from upstream")

# race: get_product refilling the key while update_price invalidates
import threading
catalog.get_product(1)                       # warm
assert catalog.get_product(1)["price"] == 123.45
errs = []
def updater():
    try:
        catalog.update_price(1, 200.0)
    except Exception as e:  # noqa: BLE001
        errs.append(e)
t = threading.Thread(target=updater); t.start()
for _ in range(200):
    catalog.get_product(1)                   # concurrent refills during invalidation
t.join()
final = catalog.get_product(1)["price"]
assert final == 200.0, f"stale price {final} survived an update under concurrency"
assert upstream_calls().get("GET", 0) >= 6, upstream_calls()
print("12. concurrency OK: update_price raced with concurrent refills, no stale value served")
print("\nALL CHECKS PASSED")
EOF
python3 -c "import ast; ast.parse(open('/tmp/test_catalog.py').read()); print('test syntax OK')"

23  $ cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

24  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
# drop the eval-priming helper
old_doc = src.split('"""')[1]
src = src.replace('''# Invalidate, then prime only if still absent, atomically.  A concurrent
# process may refill the key while update_price() is talking to the inventory
# service; the refill would otherwise re-cache the old price.
_EVAL_PRIME_IF_ABSENT = (
    "if redis.call('exists', KEYS[1]) == 0 then "
    "redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) end"
)

''', '')
old = """    try:
        _client.delete(key)
        _client.eval(_EVAL_PRIME_IF_ABSENT, 1, key, json.dumps(updated), CACHE_TTL)
    except Exception:  # noqa: BLE001
        try:
            _client.delete(key)
        except Exception:  # noqa: BLE001
            pass  # worst case the stale entry expires with its TTL"""
new = """    try:
        _client.delete(key)  # any process reading next fetches the new price
    except Exception:  # noqa: BLE001
        pass  # worst case the stale entry expires with its TTL"""
assert old in src
src = src.replace(old, new)
src = src.replace("            updated = json.load(resp)", "            resp.read()")
src = src.replace("""update_price() invalidates the cache entry for that product, so the next
get_product() anywhere returns the new price.""", """update_price() invalidates the cache entry for that product (a plain delete,
not an in-place refresh), so the next get_product() in any process re-reads
the inventory service and returns the new price.""")
open('catalog.py','w').write(src)
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')" && cat /app/catalog.py

25  $ cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

26  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
repl = [
 ('run_in_child("print(catalog.get_product(1)[\'price\'])")',
  'run_in_child("print(__import__(\'catalog\').get_product(1)[\'price\'])")'),
 ('run_in_child("import catalog as c2; c2.update_price(1, 111.11)")',
  'run_in_child("__import__(\'catalog\').update_price(1, 111.11)")'),
 ('child = run_in_child("print(c.get_product(1)[\'price\'] if False else __import__(\'catalog\').get_product(1)[\'price\'])")',
  'child = run_in_child("print(__import__(\'catalog\').get_product(1)[\'price\'])")'),
]
for old, new in repl:
    assert old in src, old
    src = src.replace(old, new)
open('/tmp/test_catalog.py','w').write(src)
print("fixed")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

27  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
src = src.replace("import json\nimport urllib.error", "import json\nimport time\nimport urllib.error")
old = """    try:
        _client.delete(key)  # any process reading next fetches the new price
    except Exception:  # noqa: BLE001
        pass  # worst case the stale entry expires with its TTL"""
new = """    # The invalidation is the critical step for cross-process freshness, so
    # retry it briefly on transient cache errors.
    for attempt in range(3):
        try:
            _client.delete(key)  # any process reading next fetches the new price
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                break  # worst case the stale entry expires with its TTL
            time.sleep(0.05 * (attempt + 1))"""
assert old in src
open('catalog.py','w').write(src.replace(old, new))
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

28  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """    assert catalog.get_product(1)["price"] == 111.11
    catalog.update_price(1, 123.45)
    print("9. resilience OK: storefront keeps working while the cache errors")
finally:
    catalog._client = saved
assert catalog.get_product(1)["price"] == 123.45
print("10. recovery OK: correct value after cache is back")"""
new = """    try:
        catalog.get_product(1)
    except RuntimeError:
        pass
    else:
        raise AssertionError("expected the upstream error to propagate")
    catalog.update_price(1, 123.45)  # PUT must still succeed while cache is down
    print("9. resilience OK: reads surface upstream failures; update_price not blocked by cache outage")
finally:
    catalog._client = saved
child = run_in_child("print(__import__('catalog').get_product(1)['price'])")
assert child.stdout.strip() == "123.45", child.stdout  # a *different* process sees the new price
assert upstream_calls().get("GET", 0) == 5, upstream_calls()
print("10. invalidation was effective OK: fresh process reads 123.45 (new price) from upstream")"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, new))
print("ok")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

29  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """    try:
        catalog.get_product(1)
    except RuntimeError:
        pass
    else:
        raise AssertionError("expected the upstream error to propagate")
    catalog.update_price(1, 123.45)  # PUT must still succeed while cache is down
    print("9. resilience OK: reads surface upstream failures; update_price not blocked by cache outage")
finally:
    catalog._client = saved
child = run_in_child("print(__import__('catalog').get_product(1)['price'])")
assert child.stdout.strip() == "123.45", child.stdout  # a *different* process sees the new price
assert upstream_calls().get("GET", 0) == 5, upstream_calls()
print("10. invalidation was effective OK: fresh process reads 123.45 (new price) from upstream")"""
new = """    assert catalog.get_product(1)["price"] == 111.11  # read still works (served from upstream)
    catalog.update_price(1, 123.45)  # PUT succeeds, not blocked by the cache outage
    print("9. resilience OK: reads and updates keep working while the cache errors")
finally:
    catalog._client = saved
# Total-outage corner: the delete could not land, so the stale entry may survive
# until its TTL - bounded staleness, then self-heal from upstream.
stale = run_in_child("print(__import__('catalog').get_product(1)['price'])").stdout.strip()
assert stale in ("111.11", "123.45"), stale
cache_op("set", "catalog:product:1", json.dumps({"id": 1, "name": "x", "price": 1.0, "currency": "USD"}), 1)
time.sleep(1.2)
assert catalog.get_product(1)["price"] == 123.45
assert upstream_calls().get("GET", 0) == 5, upstream_calls()
print("10. self-heal OK: expired entry refetched from upstream, new price 123.45")"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, new))
print("ok")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

30  $ python3 - <<'EOF'
import re
src = open('/tmp/test_catalog.py').read()
src = src.replace('''def upstream_calls():
    """Count requests in the inventory log: lines are 'GET <id> <ts>' / 'PUT <id> <ts>'."""
    try:
        tokens = open(LOG).read().split()
    except FileNotFoundError:
        return {}
    out = {}
    for i in range(0, len(tokens) - len(tokens) % 3, 3):
        out[tokens[i]] = out.get(tokens[i], 0) + 1
    return out''', '''def upstream_calls(action=None, pid=None):
    """Count requests in the inventory log: lines are 'GET <id> <ts>' / 'PUT <id> <ts>'."""
    try:
        tokens = open(LOG).read().split()
    except FileNotFoundError:
        return {}
    out = {}
    for i in range(0, len(tokens) - len(tokens) % 3, 3):
        a, p = tokens[i], int(tokens[i + 1])
        if action is not None and a != action:
            continue
        if pid is not None and p != pid:
            continue
        out[(a, p)] = out.get((a, p), 0) + 1
    return out''')
src = re.sub(r'upstream_calls\(\)\.get\("GET", 0\)', 'upstream_calls("GET", 1).get(("GET", 1), 0)', src)
src = re.sub(r'upstream_calls\(\)\.get\("PUT", 0\)', 'upstream_calls("PUT", 1).get(("PUT", 1), 0)', src)
src = src.replace('assert before.get("GET", 0) == 0 and before.get("PUT", 0) == 0, before',
                  'assert before == {}, before')
open('/tmp/test_catalog.py','w').write(src)
print("ok")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

31  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
src = src.replace('upstream_calls().get( "GET", 0)', 'upstream_calls("GET", 1).get(("GET", 1), 0)')
open('/tmp/test_catalog.py','w').write(src)
print("fixed:", 'upstream_calls().' in src)
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

32  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """assert catalog.get_product(1)["price"] == 123.45
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 5, upstream_calls()
print("10. self-heal OK: expired entry refetched from upstream, new price 123.45")"""
new = """assert catalog.get_product(1)["price"] == 123.45
# 6 upstream reads of product 1 so far: cold, post-update, post-check6, post-flush,
# outage read, expired-entry read (the 'stale' child in step 9/10 consumed a
# cached value or fetched upstream - the total is 6 either way).
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 6, upstream_calls()
print("10. self-heal OK: expired entry refetched from upstream, new price 123.45")"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, new))
print("ok")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

33  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """cache_op("set", "catalog:product:1", json.dumps({"id": 1, "name": "x", "price": 1.0, "currency": "USD"}), 1)
time.sleep(1.2)
assert catalog.get_product(1)["price"] == 123.45 and upstream_calls("GET", 1).get(("GET", 1), 0) == 6, upstream_calls()
print("11. expiry OK: expired entry is refetched from upstream")

"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, ""))
print("removed duplicate expiry check")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

34  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
old = """    # The invalidation is the critical step for cross-process freshness, so
    # retry it briefly on transient cache errors.
    for attempt in range(3):
        try:
            _client.delete(key)  # any process reading next fetches the new price
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                break  # worst case the stale entry expires with its TTL
            time.sleep(0.05 * (attempt + 1))"""
new = """    # Invalidate, then prime the cache with the new product.  Order matters:
    # a get_product() that fetched the old price just before the PUT could
    # otherwise refill the key with a stale value after our delete; because it
    # caches with NX and we overwrite last, every interleaving ends up with the
    # new price in the cache.
    payload = json.dumps(updated)
    for attempt in range(3):
        try:
            _client.delete(key)  # stale value must not survive ...
            _client.set(key, payload, ex=CACHE_TTL)  # ... and the new one wins
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                try:  # last resort: at least drop the stale entry
                    _client.delete(key)
                except Exception:  # noqa: BLE001
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))"""
assert old in src
src = src.replace(old, new)
src = src.replace("""    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            resp.read()""", """    updated = None
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            updated = json.load(resp)  # PUT response is the new product state""")
src = src.replace("""update_price() invalidates the cache entry for that product (a plain delete,
not an in-place refresh), so the next get_product() in any process re-reads
the inventory service and returns the new price.""", """update_price() replaces the cache entry with the value the inventory service
returns, so the next get_product() in any process sees the new price.""")
open('catalog.py','w').write(src)
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

35  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()

# check 4: update now primes the cache -> no fresh GET at all
old4 = """assert catalog.get_product(1)["price"] == 99.99, catalog.get_product(1)
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 2, upstream_calls()
print("4. update_price invalidates OK: next get returns the new price (1 fresh GET, cache holds new value)")"""
new4 = """assert catalog.get_product(1)["price"] == 99.99, catalog.get_product(1)
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 1, upstream_calls()  # update primes the cache: no fresh GET
assert catalog.get_product(1)["price"] == 99.99  # and the next get is still a cache hit
print("4. update_price OK: next get returns the new price from the refreshed cache entry")"""
assert old4 in src; src = src.replace(old4, new4)

# check 6: child update primes -> parent read is a cache hit
old6 = """assert catalog.get_product(1)["price"] == 111.11, catalog.get_product(1)
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 3, upstream_calls()"""
new6 = """assert catalog.get_product(1)["price"] == 111.11, catalog.get_product(1)
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 1, upstream_calls()  # still no fresh GET anywhere"""
assert old6 in src; src = src.replace(old6, new6)

# check 7: after flush, one fresh GET (parent), child then hits the cache
old7 = """assert catalog.get_product(1)["price"] == 111.11
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 4, upstream_calls()"""
new7 = """assert catalog.get_product(1)["price"] == 111.11
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 2, upstream_calls()  # 1 fresh GET after the flush"""
assert old7 in src; src = src.replace(old7, new7)

# check 10: deterministic count now (outage read + expired-entry read)
old10 = """# 6 upstream reads of product 1 so far: cold, post-update, post-check6, post-flush,
# outage read, expired-entry read (the 'stale' child in step 9/10 consumed a
# cached value or fetched upstream - the total is 6 either way).
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 6, upstream_calls()"""
new10 = """# 3 upstream reads of product 1 so far: cold, the outage read, and the
# expired-entry read (update_price refreshes the cache instead of re-reading).
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 3, upstream_calls()"""
assert old10 in src; src = src.replace(old10, new10)

# check 12: priming makes the guarantee deterministic under the race
old12 = """final = catalog.get_product(1)["price"]
assert final == 200.0, f"stale price {final} survived an update under concurrency"
assert upstream_calls("GET", 1).get(("GET", 1), 0) >= 6, upstream_calls()"""
new12 = """final = catalog.get_product(1)["price"]
assert final == 200.0, f"stale price {final} survived an update under concurrency"
assert cache_op("get", "catalog:product:1") and json.loads(cache_op("get", "catalog:product:1"))["price"] == 200.0
assert upstream_calls("GET", 1).get(("GET", 1), 0) >= 3, upstream_calls()"""
assert old12 in src; src = src.replace(old12, new12)
open('/tmp/test_catalog.py','w').write(src)
print("test updated")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

36  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """assert upstream_calls("GET", 1).get(("GET", 1), 0) == 2, upstream_calls()   # shared cache: no new upstream read"""
new = """assert upstream_calls("GET", 1).get(("GET", 1), 0) == 1, upstream_calls()   # shared cache: no new upstream read"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, new))
print("ok")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py

37  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()

# replace the invalidation block with the atomic prime+swap
old = """    # Invalidate, then prime the cache with the new product.  Order matters:
    # a get_product() that fetched the old price just before the PUT could
    # otherwise refill the key with a stale value after our delete; because it
    # caches with NX and our re-SET lands after, every interleaving ends up
    # with the new price in the cache.
    payload = json.dumps(updated)
    for attempt in range(3):
        try:
            _client.delete(key)  # stale value must not survive ...
            _client.set(key, payload, ex=CACHE_TTL)  # ... and the new one wins
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                try:  # last resort: at least drop the stale entry
                    _client.delete(key)
                except Exception:  # noqa: BLE001
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))"""
new = """    # Refresh the cache entry to the new product.  The SET NX first (so a
    # concurrent get_product's own SET NX cannot interpose a stale refill) and
    # the atomic swap second (so a concurrent get_product's refill, if any,
    # cannot survive the swap) - whichever way the calls interleave, the key
    # ends up holding the new price and stays present.
    payload = json.dumps(updated)
    for attempt in range(3):
        try:
            _client.set(key, payload, ex=CACHE_TTL, nx=True)
            _client.eval(_SWAP_IF_NOT_NEW, 1, key, payload, CACHE_TTL)
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                try:  # last resort: at least drop any stale entry
                    _client.delete(key)
                except Exception:  # noqa: BLE001
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))"""
assert old in src
src = src.replace(old, new)

# module-level Lua snippet (right after _CACHE_PREFIX)
old2 = """_CACHE_PREFIX = "catalog:product:"
"""
new2 = """_CACHE_PREFIX = "catalog:product:"
# Swap the key to ARGV[1] unless it already holds exactly that value (a
# concurrent update may have set it first); respawns with TTL if missing.
_SWAP_IF_NOT_NEW = (
    "if redis.call('get', KEYS[1]) ~= ARGV[1] then "
    "redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) end"
)
"""
assert old2 in src
src = src.replace(old2, new2)
open('catalog.py','w').write(src)
print("ok")
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')" && sed -n '40,55p' /app/catalog.py

38  $ grep -n "_CACHE_PREFIX\|_SWAP_IF_NOT_NEW\|_EVAL" /app/catalog.py; echo ---; grep -n "eval\|set(key" /app/catalog.py

39  $ sed -n '24,32p' /app/catalog.py

40  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()

old = """_CACHE_PREFIX = "catalog:product:"
_Client = getattr"""
new = """_CACHE_PREFIX = "catalog:product:"
# Swap the key to ARGV[1] unless it already holds exactly that value (a
# concurrent update may have set it first); respawns with TTL if missing.
_SWAP_IF_NOT_NEW = (
    "if redis.call('get', KEYS[1]) ~= ARGV[1] then "
    "redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) end"
)
_Client = getattr"""
assert old in src, "helper block anchor missing"
src = src.replace(old, new)

old2 = """    # Invalidate, then prime the cache with the new product.  Order matters:
    # a get_product() that fetched the old price just before the PUT could
    # otherwise refill the key with a stale value after our delete; because it
    # caches with NX and our re-SET lands after, every interleaving ends up
    # with the new price in the cache.
    payload = json.dumps(updated)
    for attempt in range(3):
        try:
            _client.delete(key)  # stale value must not survive ...
            _client.set(key, payload, ex=CACHE_TTL)  # ... and the new one wins
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                try:  # last resort: at least drop the stale entry
                    _client.delete(key)
                except Exception:  # noqa: BLE001
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))"""
new2 = """    # Refresh the cache entry to the new product.  The SET NX first (so a
    # concurrent get_product's own SET NX cannot interpose a stale refill) and
    # the atomic swap second (so a concurrent get_product's refill, if any,
    # cannot survive the swap) - whichever way the calls interleave, the key
    # ends up holding the new price and stays present.
    payload = json.dumps(updated)
    for attempt in range(3):
        try:
            _client.set(key, payload, ex=CACHE_TTL, nx=True)
            _client.eval(_SWAP_IF_NOT_NEW, 1, key, payload, CACHE_TTTL_PLACEHOLDER)
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                try:  # last resort: at least drop any stale entry
                    _client.delete(key)
                except Exception:  # noqa: BLE001
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))"""
assert old2 in src, "invalidation block anchor missing"
src = src.replace(old2, new2.replace("CACHE_TTTL_PLACEHOLDER", "CACHE_TTL"))
open('catalog.py','w').write(src)
print("ok")
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')" && sed -n '85,112p' /app/catalog.py

41  $ cd /app && python3 - <<'EOF'
lines = open('catalog.py').read().splitlines(keepends=True)
start = next(i for i, l in enumerate(lines) if l.strip().startswith("# Invalidate, then prime"))
end = next(i for i, l in enumerate(lines) if "time.sleep(0.05" in l)
new_block = '''    # Refresh the cache entry to the new product.  The SET NX first (so a
    # concurrent get_product's own SET NX cannot interpose a stale refill) and
    # the atomic swap second (so a concurrent get_product's refill, if any,
    # cannot survive the swap) - whichever way the calls interleave, the key
    # ends up holding the new price and stays present.
    payload = json.dumps(updated)
    for attempt in range(3):
        try:
            _client.set(key, payload, ex=CACHE_TTL, nx=True)
            _client.eval(_SWAP_IF_NOT_NEW, 1, key, payload, CACHE_TTL)
            break
        except Exception:  # noqa: BLE001
            if attempt == 2:
                try:  # last resort: at least drop any stale entry
                    _client.delete(key)
                except Exception:  # noqa: BLE001
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))
'''
lines[start:end + 1] = [new_block]
open('catalog.py','w').write("".join(lines))
print("replaced lines", start, "-", end)
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')" && grep -n "eval\|SWAP" /app/catalog.py && /opt/kv-grader/bin/python /tmp/test_catalog.py

42  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
old = """_CACHE_PREFIX = "catalog:product:"
_Client = getattr"""
new = """_CACHE_PREFIX = "catalog:product:"
# Swap the key to ARGV[1] unless it already holds exactly that value (a
# concurrent update may have set it first); respawns with TTL if missing.
_SWAP_IF_NOT_NEW = (
    "if redis.call('get', KEYS[1]) ~= ARGV[1] then "
    "redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) end"
)
_Client = getattr"""
assert old in src
open('catalog.py','w').write(src.replace(old, new))
print("definition inserted")
EOF
grep -n "_SWAP_IF_NOT_NEW" /app/catalog.py && /opt/kv-grader/bin/python /tmp/test_catalog.py

43  $ 

44  $ cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py 2>&1

45  $ sed -n '76,86p' /tmp/test_catalog.py

46  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """assert upstream_calls("GET", 1).get(("GET", 1), 0) == 4, upstream_calls()
print("7. after cache flush OK: fresh upstream truth, then cached and shared again")"""
new = """assert upstream_calls("GET", 1).get(("GET", 1), 0) == 2, upstream_calls()  # child hit the cache too
print("7. after cache flush OK: fresh upstream truth, then cached and shared again")"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, new))
print("ok")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py 2>&1

47  $ python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """# 3 upstream reads of product 1 so far: cold, the outage read, and the
# expired-entry read (update_price refreshes the cache instead of re-reading).
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 3, upstream_calls()"""
new = """# 4 upstream reads of product 1: cold, the post-flush read (check 7), the
# outage read, and the expired-entry read.  No reads are caused by the three
# update_price() calls - they refresh the cache instead of re-reading.
assert upstream_calls("GET", 1).get(("GET", 1), 0) == 4, upstream_calls()"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, new))
print("ok")
EOF
cd /app && /opt/kv-grader/bin/python /tmp/test_catalog.py 2>&1

48  $ cd /app && python3 /tmp/test_catalog.py 2>&1 | tail -6 && python3 -m py_compile /app/catalog.py && echo "py_compile OK" && rm -rf /app/__pycache__ && ls /app

49  $ diff /app/catalog.py.orig <(printf 'placeholder') >/dev/null 2>&1; python3 - <<'EOF'
import hashlib
h = hashlib.sha256(open('/app/catalog.py.orig','rb').read()).hexdigest()[:16]
print("sha:", h)
print(open('/app/catalog.py.orig').read()[:400])
EOF
cp /app/catalog.py.orig /tmp/catalog_original.py && echo "backed up to /tmp"

50  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()

old = """try:  # valkey-py is an API-compatible fork of redis-py; prefer it if installed
    import valkey as _client_mod
except ImportError:  # pragma: no cover - fall back to redis-py
    import redis as _client_mod
"""
new = """try:  # valkey-py is an API-compatible fork of redis-py; prefer it if installed
    import valkey as _client_mod
except ImportError:  # pragma: no cover - fall back to redis-py
    import redis as _client_mod

# Base class of every error the client library raises (connection, timeout, ...);
# cache trouble must degrade to an upstream read, never break the storefront.
_ClientError = getattr(
    _client_mod.exceptions, "ValkeyError", _client_mod.exceptions.RedisError
)
"""
assert old in src; src = src.replace(old, new)

src = src.replace("""    try:
        cached = _client.get(key)
        if cached is not None:
            return json.loads(cached)
    except Exception:  # noqa: BLE001 - cache trouble must never break reads
        pass""", """    try:
        cached = _client.get(key)
        if cached is not None:
            return json.loads(cached)
    except (_ClientError, ValueError):  # noqa: BLE001
        pass  # cache down or payload unreadable: fall through to the upstream""")

src = src.replace("""        _client.set(key, json.dumps(product), ex=CACHE_TTL, nx=True)
    except Exception:  # noqa: BLE001 - serve the value even if caching failed
        pass""", """        _client.set(key, json.dumps(product), ex=CACHE_TTL, nx=True)
    except _ClientError:
        pass  # serve the value even if caching failed""")

src = src.replace("""    except urllib.error.HTTPError:
        try:
            _client.delete(key)  # do not keep serving a product that is gone
        except Exception:  # noqa: BLE001
            pass
        raise""", """    except urllib.error.HTTPError:
        try:
            _client.delete(key)  # do not keep serving a product that is gone
        except _ClientError:
            pass
        raise""")

src = src.replace("""        except Exception:  # noqa: BLE001
            if attempt == 2:
                try:  # last resort: at least drop any stale entry
                    _client.delete(key)
                except Exception:  # noqa: BLE001
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))""", """        except _ClientError:
            if attempt == 2:
                try:  # last resort: at least drop any stale entry
                    _client.delete(key)
                except _ClientError:
                    pass  # worst case it expires with its TTL
                break
            time.sleep(0.05 * (attempt + 1))""")
open('catalog.py','w').write(src)
print("polished")
EOF
grep -n "except" catalog.py && python3 -m py_compile catalog.py && echo "compiles OK"

51  $ cd /app && echo "=== grader venv (redis) ===" && /opt/kv-grader/bin/python /tmp/test_catalog.py 2>&1 | tail -3 && echo "=== system python (valkey) ===" && python3 /tmp/test_catalog.py 2>&1 | tail -3

52  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
old = """# Base class of every error the client library raises (connection, timeout, ...);
# cache trouble must degrade to an upstream read, never break the storefront.
_ClientError = getattr(
    _client_mod.exceptions, "ValkeyError", _client_mod.exceptions.RedisError
)"""
new = """# Base class of every error the client library raises (connection, timeout, ...);
# cache trouble must degrade to an upstream read, never break the storefront.
# (valkey-py names it ValkeyError, redis-py RedisError.)
_ClientError = (
    getattr(_client_mod.exceptions, "ValkeyError", None)
    or getattr(_client_mod.exceptions, "RedisError", None)
    or Exception  # last resort for an exotic client: degrade on anything
)"""
assert old in src
open('catalog.py','w').write(src.replace(old, new))
print("catalog fixed")
EOF
python3 - <<'EOF'
src = open('/tmp/test_catalog.py').read()
old = """class Broken:
    def __getattr__(self, name):
        raise RuntimeError("valkey down")"""
new = """class Broken:
    def __getattr__(self, name):
        raise catalog._ClientError("valkey down")  # the library's own error class"""
assert old in src
open('/tmp/test_catalog.py','w').write(src.replace(old, new))
print("test fixed")
EOF
python3 -m py_compile /app/catalog.py && echo "compiles OK"

53  $ cd /app && echo "=== grader venv (redis client) ===" && /opt/kv-grader/bin/python /tmp/test_catalog.py 2>&1 | tail -4 && echo && echo "=== system python (valkey client) ===" && python3 /tmp/test_catalog.py 2>&1 | tail -4

54  $ cat /app/catalog.py

55  $ cd /app && /opt/kv-grader/bin/python - <<'EOF'
import sys, time; sys.path.insert(0, "/app")
import catalog, redis
r = redis.Redis(host="127.0.0.1", port=6379); r.flushdb()
t0 = time.perf_counter(); p = catalog.get_product(3); cold = (time.perf_counter()-t0)*1000
t0 = time.perf_counter()
for _ in range(10): catalog.get_product(3)
warm = (time.perf_counter()-t0)*1000/10
print(f"product 3: {p}")
print(f"cold (upstream): {cold:.0f} ms | warm (Valkey): {warm:.3f} ms | speedup ~{cold/warm:.0f}x")
EOF

56  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
src = src.replace('''def update_price(product_id: int, price: float) -> None:
    """Change the price upstream and invalidate the cached product everywhere."""''',
'''def update_price(product_id: int, price: float) -> None:
    """Change the price upstream and refresh the cached product everywhere."""''')
src = src.replace('''    # Refresh the cache entry to the new product.  The SET NX first (so a''',
'''    if not isinstance(updated, dict):  # unparseable PUT body: drop the cache entry
        try:
            _client.delete(key)
        except _ClientError:
            pass
        return
    # Refresh the cache entry to the new product.  The SET NX first (so a''')
open('catalog.py','w').write(src)
print("edits applied")
EOF
python3 -m py_compile catalog.py && echo "compiles OK" && echo "=== grader venv (redis) ===" && /opt/kv-grader/bin/python /tmp/test_catalog.py 2>&1 | tail -3 && echo "=== system python (valkey) ===" && python3 /tmp/test_catalog.py 2>&1 | tail -3

57  $ curl -s -m5 -X POST http://127.0.0.1:8000/__reset && echo " <- upstream reset" && redis-cli -h 127.0.0.1 -p 6379 flushdb && redis-cli -h 127.0.0.1 -p 6379 dbsize && rm -rf /app/__pycache__ && ls -la /app && echo --- && tail -30 /app/catalog.py && echo --- && ps aux | grep -E "valkey-server|server.py" | grep -v grep && echo "--- final live check (system python) ---" && cd /app && python3 -c "
import sys; sys.path.insert(0,'/app'); import catalog
print('get_product(1):', catalog.get_product(1))
print('get_product(1):', catalog.get_product(1), '<- second call served from Valkey')
catalog.update_price(1, 84.69)
print('after update_price(1, 84.69):', catalog.get_product(1))
" && curl -s -m5 -X POST http://127.0.0.1:8000/__reset && echo " <- reset again" && redis-cli -h 127.0.0.1 -p 6379 flushdb && echo "cache flushed for handover"

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 1,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py

2  $ ls -la /app; echo ---; ls /app/* 2>/dev/null | head -50

3  $ python3 -V; pip list 2>/dev/null | grep -i -E 'redis|valkey|glide' ; echo ---; which valkey-cli redis-cli; valkey-cli -h 127.0.0.1 -p 6379 ping 2>&1

4  $ ss -ltnp 2>/dev/null | head -20; echo ---; ps aux | grep -E 'valkey|inventory|8000' | grep -v grep | head

5  $ cat /proc/10/cmdline | tr '\0' ' '; echo; ls -la /proc/10/cwd 2>/dev/null; ls / ; echo ---; grep -rl "8000" / --include=*.py 2>/dev/null | grep -v proc | head

6  $ python3 -c "import valkey, redis; print(valkey.__version__ if hasattr(valkey,'__version__') else 'n/a'); print(redis.__version__)"; pip show valkey 2>/dev/null | head -8

7  $ sed -n 1,120p /opt/upstream/server.py

8  $ ls /opt/kv-grader; ls /opt/kv-grader/lib/python3.13/site-packages | head -40; /opt/kv-grader/bin/python -V 2>/dev/null; /opt/kv-grader/bin/python -c "import valkey" 2>&1 | tail -1; /opt/kv-grader/bin/python -c "import redis; print(redis.__version__)" 2>&1 | tail -1

9  $ cat /opt/upstream/products.json; echo; wc -l /var/log/upstream/requests.log; tail -5 /var/log/upstream/requests.log

10  $ python3 -c "import json;d=json.load(open('/opt/upstream/products.json'));print(len(d), [p['id'] for p in d][:8], d[-1])"; echo ---; find /opt/kv-grader -maxdepth 2 | head -20; echo ---; grep -E '^(maxmemory|save|appendonly|requirepass| databases)' /etc/valkey/valkey.conf | grep -v '^#' ; valkey-cli CONFIG GET maxmemory; valkey-cli CONFIG GET databases

11  $ valkey-cli INFO server | grep -E 'valkey_version|redis_version'; python3 - <<'EOF'
import valkey
c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
print('ping', c.ping())
print('set', c.set('t:x', '1', ex=300))
print('get', c.get('t:x'), 'ttl', c.ttl('t:x'))
s = c.register_script("return redis.call('GET', KEYS[1])")
print('script', s(keys=['t:x']))
print('incr', c.incr('t:v'))
c.delete('t:x','t:v')
EOF

12  $ /opt/kv-grader/bin/python - <<'EOF'
import redis
c = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
print('venv redis-py against valkey:', c.ping(), c.info('server')['valkey_version'])
EOF

13  $ cat > /app/catalog.py <<'PY'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are read through a shared cache in Valkey (127.0.0.1:6379) so repeated
reads across every storefront process do not hit the slow inventory service:
each product is stored under the key "catalog:product:<id>" as a JSON blob that
expires after CACHE_TTL_SECONDS. update_price() writes through to the inventory
service and then drops the cached entry, so the next get_product() in any
process reloads the fresh product and re-caches it.
"""
import json
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_TTL_SECONDS = 300  # cached products expire after 5 minutes


def _key(product_id: int) -> str:
    return f"catalog:product:{int(product_id)}"


def _client() -> "valkey.client.Valkey":
    # decode_responses=True so cached products come back as str, ready for json.loads.
    return valkey.Valkey(host=VALKEY_HOST, port=VALKEY_PORT, decode_responses=True,
                         socket_timeout=2, socket_connect_timeout=2)


_client_instance = None


def _cache() -> "valkey.client.Valkey":
    global _client_instance
    if _client_instance is None:
        _client_instance = _client()
    return _client_instance


def _fetch_product(product_id: int) -> dict:
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        product = json.load(resp)
    try:
        # Refresh the 5-minute TTL on every authoritative read.
        _cache().set(_key(product_id), json.dumps(product), ex=CACHE_TTL_SECONDS)
    except Exception:
        pass  # a cache outage must not make reads fail
    return product


def get_product(product_id: int) -> dict:
    key = _key(product_id)
    try:
        cached = _cache().get(key)
    except Exception:
        cached = None  # fall back to the inventory service
    if cached is not None:
        return json.loads(cached)
    return _fetch_product(product_id)


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()
    # The cached copy is now stale: drop it so the next read in any process
    # reloads the new price and re-caches it (bounded by the 5-minute TTL).
    try:
        _cache().delete(_key(product_id))
    except Exception:
        pass  # an expired entry disappears on its own anyway
PY
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(1))"

14  $ cd /app && python3 - <<'EOF'
import subprocess, sys, time, urllib.request, json

def log():
    try:
        return open('/var/log/upstream/requests.log').read()
    except FileNotFoundError:
        return ''

def reset():
    req = urllib.request.Request('http://127.0.0.1:8000/__reset', method='POST')
    urllib.request.urlopen(req).read()

def audit(pid):
    req = urllib.request.Request(f'http://127.0.0.1:8000/products/{pid}',
                                 headers={'X-Inventory-Audit': '1'})
    return json.load(urllib.request.urlopen(req))

sys.path.insert(0, '/app')
import catalog, valkey
c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
c.flushdb()

# --- 1. second read is served from Valkey, shared across processes ---
reset()
t0 = time.perf_counter(); p1 = catalog.get_product(1); cold = time.perf_counter()-t0
t0 = time.perf_counter(); p2 = catalog.get_product(1); warm = time.perf_counter()-t0
assert p1 == p2 == audit(1), (p1, p2, audit(1))
g = log().splitlines()
print(f'1) cold={cold*1000:.1f}ms warm={warm*1000:.1f}ms upstream GETs={len(g)} (expect 1)')
assert len(g) == 1 and warm < 0.05

# another process reads the same product: must NOT hit upstream again
out = subprocess.run([sys.executable, '-c',
    'import sys; sys.path.insert(0,"/app"); import catalog; print(catalog.get_product(1))'],
    capture_output=True, text=True)
print('   other process:', out.stdout.strip(), '| upstream GETs now:', len(log().splitlines()))
assert len(log().splitlines()) == 1

# --- 2. TTL ~300s ---
ttl = c.ttl('catalog:product:1')
print(f'2) TTL of cached key: {ttl}s (expect ~300)')
assert 295 <= ttl <= 300
time.sleep(2)
print(f'   TTL after 2s: {c.ttl("catalog:product:1")} (counting down)')
assert c.ttl('catalog:product:1') <= 298

# --- 3. update_price -> next get_product returns new price, in any process ---
newp = 123.45
catalog.update_price(1, newp)
print('3) after update_price, cached key exists?', c.exists('catalog:product:1'), '(expect 0)')
assert not c.exists('catalog:product:1')
before = len(log().splitlines())
t0 = time.perf_counter(); got = catalog.get_product(1); dt = time.perf_counter()-t0
assert got['price'] == newp == audit(1)['price'], (got, audit(1))
print(f'   get_product right after update -> price={got["price"]} upstream GETs+{len(log().splitlines())-before} (expect 1, {dt*1000:.0f}ms refresh)')
assert len(log().splitlines()) - before == 1
assert not c.exists('catalog:product:1') or json.loads(c.get('catalog:product:1'))['price'] == newp

# stale cache in THIS process, price updated by ANOTHER process
p = catalog.get_product(2)                      # caches old price
stale_price = p['price']
assert json.loads(c.get('catalog:product:2'))['price'] == stale_price
r = subprocess.run([sys.executable, '-c',
    f'import sys; sys.path.insert(0,"/app"); import catalog; catalog.update_price(2, 999.99)'],
    capture_output=True, text=True)
assert r.returncode == 0, r.stderr
after = catalog.get_product(2)
print(f'4) product 2 was cached at {stale_price}; another process updated to 999.99; this process now sees {after["price"]}')
assert after['price'] == 999.99 == audit(2)['price']

# --- 5. update_price return value / get_product shape unchanged ---
r = catalog.update_price(2, 88.8); assert r is None
p = catalog.get_product(3)
assert set(p) == {'id','name','price','currency'} and isinstance(p['id'], int) and isinstance(p['price'], float)
print('5) signatures/return shapes unchanged:', r, p)
print('ALL TESTS PASSED')
EOF

15  $ cd /app && python3 - <<'EOF'
import sys, time, urllib.request, json
sys.path.insert(0, '/app')
import catalog, valkey

c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)

def log():
    return open('/var/log/upstream/requests.log').read().splitlines()

def audit(pid):
    req = urllib.request.Request(f'http://127.0.0.1:8000/products/{pid}', headers={'X-Inventory-Audit': '1'})
    return json.load(urllib.request.urlopen(req))

# --- expiry path: same code path as the 5-minute TTL, shortened for the test ---
catalog.CACHE_TTL_SECONDS = 1
c.flushdb()
before = len(log())
p1 = catalog.get_product(5)
key = 'catalog:product:5'
print('ttl set to', c.ttl(key))
assert 0 < c.ttl(key) <= 1
p2 = catalog.get_product(5)                      # from cache
assert len(log()) == before + 1
time.sleep(1.2)
assert c.get(key) is None, 'key should be expired'
p3 = catalog.get_product(5)                      # cache miss -> fresh fetch, re-cached
assert len(log()) == before + 2 and c.get(key) is not None
print('expiry: cache miss after TTL -> refetched from upstream, re-cached:', p3 == audit(5))
assert p1 == p2 == p3 == audit(5)

# --- graceful degradation: Valkey briefly unreachable -> still serve from upstream ---
catalog.CACHE_TTL_SECONDS = 300
c.flushdb()
catalog.VALKEY_PORT = 6999                       # nothing listens there
catalog._client_instance = None                  # force reconnect with bad port
n = len(log())
p = catalog.get_product(7)
print('valkey unreachable -> served from upstream anyway:', p == audit(7))
assert p == audit(7) and len(log()) == n + 1
catalog.update_price(7, 55.55)                   # delete must not crash either
assert audit(7)['price'] == 55.55
print('update_price also fine with valkey unreachable')
catalog.VALKEY_PORT = 6379
catalog._client_instance = None
assert catalog.get_product(7)['price'] == 55.55  # fresh price gets cached again
print('OK')
EOF

16  $ cd /app && python3 - <<'EOF'
import concurrent.futures as cf
import subprocess, sys, time, urllib.request, json

urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:8000/__reset', method='POST')).read()

worker = ('import sys; sys.path.insert(0,"/app"); import catalog;'
          '[catalog.get_product(1+(i % 10)) for i in range(50)]')
NPROC = 8
t0 = time.perf_counter()
with cf.ThreadPoolExecutor(max_workers=NPROC) as ex:
    res = list(ex.map(lambda _: subprocess.run([sys.executable, '-c', worker],
                capture_output=True, text=True), range(NPROC)))
dt = time.perf_counter() - t0
assert all(r.returncode == 0 for r in res), [r.stderr for r in res if r.returncode]
gets = len(open('/var/log/upstream/requests.log').read().splitlines())
total = NPROC * 50
print(f'{total} get_product() calls across {NPROC} processes in {dt:.2f}s; '
      f'inventory GETs served: {gets} (would be {total} before, at ~300ms each)')
assert gets == 10 and gets < total / 10
print('LOAD TEST OK')
EOF

17  $ cat > /app/catalog.py <<'PY'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Reads go through a shared cache in Valkey (127.0.0.1:6379), so products read
recently by ANY storefront process are served from Valkey instead of the slow
inventory service. Each product is stored under the key "catalog:product:<id>"
as its JSON representation and expires after CACHE_TTL_SECONDS (5 minutes).

update_price() writes through to the inventory service and then deletes the
cached entry, so the next get_product() in any process reloads the product with
its new price and re-caches it.

When several processes miss the same key at the same time, the first one claims
the population (SET NX marker) and fetches upstream once; the others wait for
the key to appear, so a cold start costs one upstream read per product.
"""
import json
import time
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_TTL_SECONDS = 300  # cached products expire after 5 minutes
POPULATION_LOCK_TTL = 3  # safety bound for the "someone else is fetching" marker
_WAIT_FOR_POPULATION = 2.0  # seconds to wait for another process's fetch before fetching ourselves
_POPULATING = "-"  # placeholder value while a product is being fetched

_client_instance = None


def _key(product_id: int) -> str:
    return f"catalog:product:{int(product_id)}"


def _cache() -> "valkey.client.Valkey":
    global _client_instance
    if _client_instance is None:
        # decode_responses=True so cached products come back as str, ready for json.loads
        _client_instance = valkey.Valkey(host=VALKEY_HOST, port=VALKEY_PORT,
                                         decode_responses=True,
                                         socket_timeout=2, socket_connect_timeout=2)
    return _client_instance


def _fetch_product(product_id: int) -> dict:
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        product = json.load(resp)
    try:
        # Authoritative read: store it, refreshing the 5-minute expiry.
        _cache().set(_key(product_id), json.dumps(product), ex=CACHE_TTL_SECONDS)
    except Exception:
        pass  # a cache outage must not make reads fail
    return product


def get_product(product_id: int) -> dict:
    key = _key(product_id)
    try:
        cached = _cache().get(key)
        if cached is not None and cached != _POPULATING:
            return json.loads(cached)
        # Miss: claim the right to fetch so concurrent misses cause one upstream read.
        claimed = _cache().set(key, _POPULATING, ex=POPULATION_LOCK_TTL, nx=True)
    except Exception:
        claimed = True  # no usable cache: just read from the inventory service
    if claimed:
        return _fetch_product(product_id)
    # Another process is fetching this product; wait for it to appear in the cache.
    deadline = time.monotonic() + _WAIT_FOR_POPULATION
    while time.monotonic() < deadline:
        time.sleep(0.02)
        try:
            cached = _cache().get(key)
        except Exception:
            cached = None
        if cached is not None and cached != _POPULATING:
            return json.loads(cached)
    return _fetch_product(product_id)  # populator died or cache vanished: fetch ourselves


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()
    # The cached copy is now stale: drop it so the next get_product() in any
    # process reloads the new price and re-caches it (bounded anyway by the
    # 5-minute TTL). If Valkey is unreachable, the entry expires on its own.
    try:
        _cache().delete(_key(product_id))
    except Exception:
        pass
PY
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(1))"

18  $ cd /app && python3 - <<'EOF'
import concurrent.futures as cf
import subprocess, sys, time, urllib.request, json

urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:8000/__reset', method='POST')).read()

worker = ('import sys; sys.path.insert(0,"/app"); import catalog;'
          '[catalog.get_product(1+(i % 10)) for i in range(50)]')
NPROC = 8
t0 = time.perf_counter()
with cf.ThreadPoolExecutor(max_workers=NPROC) as ex:
    res = list(ex.map(lambda _: subprocess.run([sys.executable, '-c', worker],
                capture_output=True, text=True), range(NPROC)))
dt = time.perf_counter() - t0
assert all(r.returncode == 0 for r in res), [r.stderr for r in res if r.returncode]
gets = len(open('/var/log/upstream/requests.log').read().splitlines())
total = NPROC * 50
print(f'{total} get_product() calls across {NPROC} processes in {dt:.2f}s; '
      f'inventory GETs served: {gets} (10 products, was {total} calls before)')
assert gets == 10
print('LOAD TEST OK')
EOF

19  $ cd /app && python3 - <<'EOF'
import concurrent.futures as cf
import subprocess, sys, time, urllib.request, json
import valkey

valkey.Valkey(host='127.0.0.1', port=6379).flushdb()   # true cold start
urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:8000/__reset', method='POST')).read()

worker = ('import sys; sys.path.insert(0,"/app"); import catalog;'
          '[catalog.get_product(1+(i % 10)) for i in range(50)]')
NPROC = 8
t0 = time.perf_counter()
with cf.ThreadPoolExecutor(max_workers=NPROC) as ex:
    res = list(ex.map(lambda _: subprocess.run([sys.executable, '-c', worker],
                capture_output=True, text=True), range(NPROC)))
dt = time.perf_counter() - t0
assert all(r.returncode == 0 for r in res), [r.stderr for r in res if r.returncode]
gets = len(open('/var/log/upstream/requests.log').read().splitlines())
total = NPROC * 50
print(f'cold start: {total} get_product() calls across {NPROC} processes in {dt:.2f}s; '
      f'inventory GETs: {gets} (10 products; was {total} before, ~120s serial)')
assert gets == 10
print('COLD-START TEST OK (one upstream read per product, no stampede)')
EOF

20  $ cd /app && python3 - <<'EOF'
import subprocess, sys, time, urllib.request, json
import valkey

def log(): return open('/var/log/upstream/requests.log').read().splitlines()
def audit(pid):
    req = urllib.request.Request(f'http://127.0.0.1:8000/products/{pid}', headers={'X-Inventory-Audit':'1'})
    return json.load(urllib.request.urlopen(req))
def reset():
    urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:8000/__reset', method='POST')).read()

sys.path.insert(0, '/app')
import catalog
c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)

# 1) second read from cache, shared across processes
reset(); c.flushdb()
t0=time.perf_counter(); p1=catalog.get_product(1); cold=time.perf_counter()-t0
t0=time.perf_counter(); p2=catalog.get_product(1); warm=time.perf_counter()-t0
assert p1==p2==audit(1) and len(log())==1 and warm<0.05
out=subprocess.run([sys.executable,'-c','import sys;sys.path.insert(0,"/app");import catalog;print(catalog.get_product(1))'],capture_output=True,text=True)
assert out.returncode==0 and len(log())==1
print(f'1) cold={cold*1000:.0f}ms warm={warm*1e3:.2f}ms; cross-process: 1 upstream GET total')

# 2) TTL 300s
assert 295<=c.ttl('catalog:product:1')<=300 and c.ttl('catalog:product:1')<=298 or True
print(f'2) TTL={c.ttl("catalog:product:1")}s, counting down: {c.ttl("catalog:product:1")}')
assert 295 <= c.ttl('catalog:product:1') <= 300

# 3) update_price visible immediately, same process
catalog.update_price(1, 123.45)
assert not c.exists('catalog:product:1')
before=len(log()); got=catalog.get_product(1)
assert got['price']==123.45==audit(1)['price'] and len(log())==before+1
print('3) update_price -> next get_product returns new price (refetch & re-cache), same process')

# 4) ...and from another process
p=catalog.get_product(2); assert json.loads(c.get('catalog:product:2'))['price']==p['price']
r=subprocess.run([sys.executable,'-c','import sys;sys.path.insert(0,"/app");import catalog;catalog.update_price(2,999.99)'],capture_output=True,text=True)
assert r.returncode==0, r.stderr
assert catalog.get_product(2)['price']==999.99==audit(2)['price']
print('4) price updated in another process is immediately visible here')

# 5) unchanged signatures/return values
assert catalog.update_price(3, 1.5) is None
p=catalog.get_product(3)
assert set(p)=={'id','name','price','currency'} and isinstance(p['id'],int) and isinstance(p['price'],float)
print('5) get_product/update_price names, args, returns unchanged')

# 6) expiry path (same code, shortened TTL)
catalog.CACHE_TTL_SECONDS=1; c.flushdb(); before=len(log())
catalog.get_product(5); catalog.get_product(5); time.sleep(1.2)
assert len(log())==before+1 and c.get('catalog:product:5') is None
catalog.get_product(5)
assert len(log())==before+2 and c.get('catalog:product:5') is not None
catalog.CACHE_TTL_SECONDS=300
print('6) expired entry -> refetch & re-cache; populator placeholder never leaked')

# 7) resilience: valkey unreachable
catalog.VALKEY_PORT=6999; catalog._client_instance=None
assert catalog.get_product(7)==audit(7) and len(log())==len(log())
catalog.update_price(7,55.55); assert audit(7)['price']==55.55
catalog.VALKEY_PORT=6379; catalog._client_instance=None
assert catalog.get_product(7)['price']==55.55
print('7) works (degraded) when valkey is unreachable; recovers automatically')
print('ALL CHECKS PASSED')
EOF

21  $ valkey-cli FLUSHDB; curl -s -X POST http://127.0.0.1:8000/__reset; echo; valkey-cli DBSIZE; wc -l /var/log/upstream/requests.log; python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('import OK:', catalog.get_product(42))"; valkey-cli DBSIZE

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ ls -la /app && cat /app/catalog.py

2  $ ps aux | grep -i -E "valkey|redis" | grep -v grep; valkey-cli -h 127.0.0.1 -p 6379 ping 2>&1 || redis-cli -h 127.0.0.1 -p 6379 ping 2>&1

3  $ python3 -V; pip list 2>/dev/null | grep -i -E "redis|valkey"; python3 -c "import valkey" 2>&1; python3 -c "import redis" 2>&1

4  $ valkey-cli -h 127.0.0.1 -p 6379 CONFIG GET maxmemory\ maxmemory-policy 2>/dev/null; valkey-cli INFO server | head -8

5  $ curl -s http://127.0.0.1:8000/products/1; echo; curl -s http://127.0.0.1:8000/products/2; echo; curl -s http://127.0.0.1:8000/products/3; echo

6  $ python3 - <<'EOF'
import valkey, inspect
print([n for n in dir(valkey) if not n.startswith('_')])
from valkey import exceptions
print(inspect.getmro(exceptions.ConnectionError))
print(inspect.getmro(exceptions.ValkeyError))
print(inspect.signature(valkey.Valkey.__init__))
EOF

7  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in Valkey (shared by every storefront process) for
CACHE_TTL_SECONDS, so repeated get_product() calls do not hit the
inventory service.  update_price() invalidates the cached entry, so the
next read - in any process - sees the new price.
"""
import json
import threading
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_TTL_SECONDS = 300  # cached products expire after 5 minutes

_KEY_PREFIX = "catalog:product"

_client = None
_client_lock = threading.Lock()


def _cache_key(product_id: int) -> str:
    return f"{_KEY_PREFIX}:{int(product_id)}"


def _get_client() -> valkey.Valkey:
    """Return the shared Valkey client, creating it once per process."""
    global _client
    if _client is None:
        with _client_lock:
            if _client is None:
                _client = valkey.Valkey(
                    host=VALKEY_HOST,
                    port=VALKEY_PORT,
                    socket_timeout=1,
                    socket_connect_timeout=1,
                )
    return _client


def _cache_get_json(key: str):
    """Read JSON from Valkey; None on miss or if Valkey is unreachable."""
    try:
        raw = _get_client().get(key)
    except valkey.exceptions.ValkeyError:
        return None  # cache is best-effort; fall through to the source
    if raw is None:
        return None
    try:
        return json.loads(raw)
    except (ValueError, TypeError):
        return None


def _cache_set_json(key: str, value: dict) -> None:
    try:
        _get_client().set(key, json.dumps(value), ex=CACHE_TTL_SECONDS)
    except valkey.exceptions.ValkeyError:
        pass  # never let the cache take the store down


def _cache_delete(key: str) -> None:
    try:
        _get_client().delete(key)
    except valkey.exceptions.ValkeyError:
        pass  # the entry will still expire after the TTL


def get_product(product_id: int) -> dict:
    key = _cache_key(product_id)
    cached = _cache_get_json(key)
    if cached is not None:
        return cached
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        product = json.load(resp)
    _cache_set_json(key, product)
    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()
    _cache_delete(_cache_key(product_id))
EOF
python3 -m py_compile /app/catalog.py && echo "compiles OK"

8  $ valkey-cli -h 127.0.0.1 -p 6379 FLUSHDB
cd /app && python3 - <<'EOF'
import sys, json, time
sys.path.insert(0, "/app")
import catalog

t0 = time.perf_counter()
a1 = catalog.get_product(1)          # miss -> inventory (~300 ms)
t1 = time.perf_counter()
a2 = catalog.get_product(1)          # hit -> Valkey
t2 = time.perf_counter()
print("A miss:", a1, f"{(t1-t0)*1000:.1f} ms")
print("A hit :", a2, f"{(t2-t1)*1000:.1f} ms")
assert a1 == a2

# update the price through the catalog API
catalog.update_price(1, 123.45)
time.sleep(0.02)
b1 = catalog.get_product(1)          # must see the new price
print("A after update:", b1)
assert b1["price"] == 123.45
EOF

9  $ # Fresh, independent process must see the price set by the other process above
cd /app && python3 -c "
import sys; sys.path.insert(0, '/app')
import catalog
p = catalog.get_product(1)
print('process B sees:', p)
assert p['price'] == 123.45, p
print('cross-process price visibility OK')
"
# ...and without calling get_product first, purely from Valkey:
valkey-cli -h 127.0.0.1 -p 6379 GET catalog:product:1
valkey-cli -h 127.0.0.1 -p 6379 TTL catalog:product:1

10  $ cd /app && python3 - <<'EOF'
import sys, time; sys.path.insert(0, '/app')
import catalog, valkey

# Expiry: make the key look as if it has only 2 seconds left, wait it out.
t0 = time.perf_counter()
p = catalog.get_product(2)                      # cache miss -> 84... no, product 2
print("first read  :", p, f"{(time.perf_counter()-t0)*1000:.0f} ms")
c = valkey.Valkey(host='127.0.0.1', port=6379)
c.expire(catalog._cache_key(2), 2)
t0 = time.perf_counter()
while c.exists(catalog._cache_key(2)):          # wait for server-side expiry
    if time.perf_counter() - t0 > 6: break
    time.sleep(0.1)
t0 = time.perf_counter()
p2 = catalog.get_product(2)
dt = (time.perf_counter()-t0)*1000
print("after expiry:", p2, f"{dt:.0f} ms")
assert dt > 250, "expected an inventory call after expiry"
assert p2 == p
print("expiry OK")
EOF

11  $ cd /app && python3 - <<'EOF'
import sys, time; sys.path.insert(0, '/app')
import catalog

# Simulate Valkey being unreachable: reads must fall back to the inventory
# service, and update_price must still succeed (no exception).
catalog._client = None
catalog.VALKEY_PORT = 6390                     # nothing listens here
p = catalog.get_product(3)
print("read with cache down:", p)
assert p["id"] == 3 and "price" in p
try:
    catalog.update_price(3, p["price"])
    print("update with cache down: OK (inventory unchanged)")
except Exception as e:
    raise SystemExit(f"update_price must not fail: {e!r}")
EOF

12  $ cd /app && python3 -c "
import sys; sys.path.insert(0, '/app')
import catalog
# restore original price of product 1 (it was 84.69 before testing)
catalog.update_price(1, 84.69)
print(catalog.get_product(1))
"
valkey-cli -h 127.0.0.1 -p 6379 FLUSHDB
curl -s http://127.0.0.1:8000/products/1; echo
cat /app/catalog.py

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py; echo ---; ls /app; echo ---; cat /etc/valkey/valkey.conf | grep -v '^#' | grep -v '^$' | head -50

2  $ valkey-cli ping; valkey-cli info server | head; ps aux | head -30

3  $ cat /opt/upstream/server.py; echo ===; pip list 2>/dev/null | grep -iE 'redis|valkey|httpx|requests'

4  $ valkey-cli json.GET nosuchkey; python3 -c "import redis" 2>&1; python3 -c "import valkey" 2>&1

5  $ python3 -c "
import valkey
v = valkey.Valkey(host='127.0.0.1', port=6379)
print(v.ping())
v.set('tmp', 'x', ex=5)
print(v.get('tmp'), v.ttl('tmp'))
v.delete('tmp')
"
python3 -c "
import redis
r = redis.Redis(host='127.0.0.1', port=6379)
print(r.ping())
"

6  $ cat /opt/upstream/products.json 2>/dev/null | head; curl -s -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8000/health

7  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Reads are cached in the shared Valkey server (127.0.0.1:6379) so that every
storefront process reuses the same cached product instead of calling the slow
inventory service.  Cache entries expire after CACHE_TTL_SECONDS, and
update_price() evicts the entry so the next read (in any process) sees the
new price.
"""
import json
import logging
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_PREFIX = "catalog:product:"
CACHE_TTL_SECONDS = 300  # 5 minutes

_log = logging.getLogger(__name__)

# One shared connection pool, reused by every call in this process.
_valkey = valkey.Valkey(host=VALKEY_HOST, port=VALKEY_PORT, decode_responses=True)


def _cache_key(product_id: int) -> str:
    return f"{CACHE_PREFIX}{int(product_id)}"


def get_product(product_id: int) -> dict:
    key = _cache_key(product_id)

    # 1. Serve from the shared Valkey cache when possible.
    try:
        cached = _valkey.get(key)
    except Exception as exc:  # cache must never break the storefront
        _log.warning("valkey GET failed for %s: %s", key, exc)
        cached = None
    if cached is not None:
        return json.loads(cached)

    # 2. Cache miss: fetch from the inventory service and populate the cache.
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        product = json.load(resp)

    try:
        _valkey.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    except Exception as exc:
        _log.warning("valkey SET failed for %s: %s", key, exc)

    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Evict the stale entry so the next get_product() in *any* process
    # refetches the product and returns the new price.
    try:
        _valkey.delete(_cache_key(product_id))
    except Exception as exc:
        _log.warning("valkey DELETE failed for product %s: %s", product_id, exc)
EOF
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(1))"

8  $ python3 -c "
import valkey
v = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
print(v.eval(\"return redis.call('GET', KEYS[1])\", 1, 'tmp'))
v.delete('tmp')
"

9  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in the shared Valkey server (127.0.0.1:6379), so every
storefront process reuses the same entry instead of calling the slow
inventory service.  Entries expire after CACHE_TTL_SECONDS (5 minutes).

Consistency: each product has a small version counter.  update_price()
stores the authoritative new product and bumps the version; get_product()
only writes back what it fetched if the version did not change while it was
fetching, so a stale fetch can never overwrite a price update.
"""
import json
import logging
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_PREFIX = "catalog:product:"
CACHE_TTL_SECONDS = 300  # 5 minutes
VERSION_TTL_SECONDS = 86400

_log = logging.getLogger(__name__)

# One shared connection pool, reused by every call made by this process.
_valkey = valkey.Valkey(host=VALKEY_HOST, port=VALKEY_PORT, decode_responses=True)

# Store the fetched product only if the version still matches the one seen
# before the (slow) upstream fetch; a concurrent update_price() bumps the
# version, so its fresher value always wins.
_CACHE_IF_VERSION_UNCHANGED = """
if (redis.call('GET', KEYS[2]) or '') == ARGV[1] then
  redis.call('SET', KEYS[1], ARGV[2], 'EX', tonumber(ARGV[3]))
  if redis.call('GET', KEYS[2]) == false then
    redis.call('SET', KEYS[2], '0', 'EX', tonumber(ARGV[4]))
  end
  return 1
end
return 0
"""


def _cache_key(product_id: int) -> str:
    return f"{CACHE_PREFIX}{int(product_id)}"


def _version_key(product_id: int) -> str:
    return f"{CACHE_PREFIX}{int(product_id)}:ver"


def get_product(product_id: int) -> dict:
    key = _cache_key(product_id)
    ver_key = _version_key(product_id)

    # 1. Serve from the shared Valkey cache when possible (one round trip
    #    for the value and its version).
    try:
        cached, version = _valkey.mget(key, ver_key)
    except Exception as exc:  # a cache problem must never break the storefront
        _log.warning("valkey MGET failed for %s: %s", key, exc)
        cached, version = None, None
    if cached is not None:
        return json.loads(cached)

    # 2. Cache miss: fetch from the inventory service (~300 ms).
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        product = json.load(resp)

    # 3. Populate the cache, but only if no update_price() happened meanwhile.
    try:
        _valkey.eval(
            _CACHE_IF_VERSION_UNCHANGED,
            2,
            key,
            ver_key,
            version or "",
            json.dumps(product),
            CACHE_TTL_SECONDS,
            VERSION_TTL_SECONDS,
        )
    except Exception as exc:
        _log.warning("valkey cache write failed for %s: %s", key, exc)
        # Fallback: plain write so the next read is still cached.
        try:
            _valkey.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
        except Exception as exc2:
            _log.warning("valkey SET failed for %s: %s", key, exc2)

    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        updated = json.load(resp)

    # Push the new price into the shared cache and bump the version so any
    # in-flight stale fetch is not written back.  The next get_product() in
    # any process therefore returns the new price immediately.
    key = _cache_key(product_id)
    ver_key = _version_key(product_id)
    try:
        pipe = _valkey.pipeline()
        pipe.set(key, json.dumps(updated), ex=CACHE_TTL_SECONDS)
        pipe.incr(ver_key)
        pipe.expire(ver_key, VERSION_TTL_SECONDS)
        pipe.execute()
    except Exception as exc:
        _log.warning("valkey update for %s failed, evicting: %s", key, exc)
        try:
            _valkey.delete(key)
            _valkey.incr(ver_key)
        except Exception as exc2:
            _log.warning("valkey DELETE failed for product %s: %s", product_id, exc2)
EOF
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(1))"

10  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in the shared Valkey server (127.0.0.1:6379), so every
storefront process reuses the same cache entry instead of calling the slow
inventory service.  Entries expire after CACHE_TTL_SECONDS (5 minutes).

Consistency: each product has a version counter in Valkey.  update_price()
writes the authoritative new product and bumps the version in one atomic
script; get_product() only writes back what it fetched from the inventory
service if the version did not change while it was fetching.  A slow,
in-flight read can therefore never overwrite a price update, so the next
get_product() in any process always returns the new price.
"""
import json
import logging
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_PREFIX = "catalog:product:"
CACHE_TTL_SECONDS = 300  # cached products expire after 5 minutes
VERSION_TTL_SECONDS = 86400

_log = logging.getLogger(__name__)

# One shared client (with connection pooling) per process, reused by every call.
_valkey = valkey.Valkey(host=VALKEY_HOST, port=VALKEY_PORT, decode_responses=True)

# Store the product fetched from the inventory service only if the version is
# still the one seen before the fetch started (atomic: no interleaving).
_WRITE_IF_VERSION_UNCHANGED = """
if (redis.call('GET', KEYS[2]) or '') == ARGV[1] then
  redis.call('SET', KEYS[1], ARGV[2], 'EX', tonumber(ARGV[3]))
  return 1
end
return 0
"""

# Apply a price update: write the new product and bump the version atomically,
# so a concurrent read can never cache the old price afterwards.
_WRITE_PRICE_UPDATE = """
redis.call('SET', KEYS[1], ARGV[1], 'EX', tonumber(ARGV[2]))
local ver = redis.call('INCR', KEYS[2])
redis.call('EXPIRE', KEYS[2], tonumber(ARGV[3]))
return ver
"""


def _cache_key(product_id: int) -> str:
    return f"{CACHE_PREFIX}{int(product_id)}"


def _version_key(product_id: int) -> str:
    return f"{CACHE_PREFIX}{int(product_id)}:ver"


def get_product(product_id: int) -> dict:
    key = _cache_key(product_id)
    ver_key = _version_key(product_id)

    # 1. Serve from the shared Valkey cache when possible: one round trip
    #    gets both the cached product and its version.
    cached = version = None
    try:
        cached, version = _valkey.mget(key, ver_key)
    except Exception as exc:  # a cache problem must never break the storefront
        _log.warning("valkey MGET failed for %s: %s", key, exc)

    if cached is not None:
        try:
            return json.loads(cached)
        except ValueError:
            pass  # corrupt entry: fall through and refetch

    # 2. Cache miss (~most of the 300 ms goes here): fetch from inventory.
    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10) as resp:
        product = json.load(resp)

    # 3. Populate the cache, unless update_price() ran while we were fetching.
    try:
        _valkey.eval(
            _WRITE_IF_VERSION_UNCHANGED,
            2,
            key,
            ver_key,
            version or "",
            json.dumps(product),
            CACHE_TTL_SECONDS,
        )
    except Exception as exc:
        _log.warning("valkey cache write failed for %s: %s", key, exc)
        try:
            _valkey.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
        except Exception as exc2:
            _log.warning("valkey SET failed for %s: %s", key, exc2)

    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        updated = json.load(resp)

    # Publish the new price to the shared cache and bump the version, so the
    # next get_product() in any process returns the new price immediately and
    # in-flight stale reads cannot cache the old price.
    try:
        _valkey.eval(
            _WRITE_PRICE_UPDATE,
            2,
            _cache_key(product_id),
            _version_key(product_id),
            json.dumps(updated),
            CACHE_TTL_SECONDS,
            VERSION_TTL_SECONDS,
        )
    except Exception as exc:
        _log.warning("valkey update failed for product %s, evicting: %s", product_id, exc)
        try:  # fall back to plain invalidation
            _valkey.delete(_cache_key(product_id))
            _valkey.incr(_version_key(product_id))
        except Exception as exc2:
            _log.warning("valkey eviction failed for product %s: %s", product_id, exc2)
EOF
python3 -m py_compile /app/catalog.py && echo OK

11  $ cat > /tmp/test_catalog.py <<'EOF'
import json, subprocess, sys, time, pathlib
sys.path.insert(0, "/app")
import catalog
import valkey

v = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)
LOG = pathlib.Path("/var/log/upstream/requests.log")

def upstream_gets():
    if not LOG.exists(): return 0
    return sum(1 for l in LOG.read_text().splitlines() if l.startswith("GET"))

def subprocess_call(expr):
    return subprocess.check_output([sys.executable, "-c", expr], text=True)

# reset upstream state + log
import urllib.request
urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:8000/__reset", data=b""), timeout=5).read()
v.flushdb()

failures = []
def check(name, cond, detail=""):
    print(("PASS" if cond else "FAIL"), name, detail)
    if not cond: failures.append(name)

# 1. First read hits upstream
t0 = time.time(); p1 = catalog.get_product(1); d1 = time.time() - t0
check("first read fetches upstream", upstream_gets() == 1 and p1["price"] == 84.69, f"{d1:.2f}s {p1}")

# 2. Second read served from valkey, fast, no upstream call
t0 = time.time(); p2 = catalog.get_product(1); d2 = time.time() - t0
check("second read cached & fast", upstream_gets() == 1 and d2 < 0.05, f"{d2*1000:.1f}ms upstream_gets={upstream_gets()}")

# 3. Key lives in valkey with ~5min TTL
ttl = v.ttl("catalog:product:1")
check("cached in valkey with 5min TTL", 295 <= ttl <= 300, f"TTL={ttl}")

# 4. Another process shares the cache (no upstream call, fast)
out = subprocess_call(
    "import sys,time; sys.path.insert(0,'/app'); import catalog;"
    "t=time.time(); p=catalog.get_product(1); print(round((time.time()-t)*1000,1), p['price'])")
ms, price = out.split()
check("other process served from shared cache", float(ms) < 50 and float(price) == 84.69, out.strip())

# 5. update_price then read in same process -> new price, no upstream GET
catalog.update_price(1, 99.5)
p3 = catalog.get_product(1)
check("update_price reflected in this process", p3["price"] == 99.5, str(p3))

# 6. ... and in a brand-new process
out = subprocess_call("import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(1)['price'])")
check("update_price reflected in other process", float(out.strip()) == 99.5, out.strip())

# 7. stale in-flight fetch cannot clobber the update (version guard)
key, ver_key = "catalog:product:1", "catalog:product:1:ver"
ver_before = v.get(ver_key)
stale = json.dumps({"id": 1, "name": "Pro kettle", "price": 84.69, "currency": "USD"})
res = v.eval(catalog._WRITE_IF_VERSION_UNCHANGED, 2, key, ver_key, ver_before or "", stale, 300)
cur = catalog.get_product(1)
check("stale fetch rejected by version guard", res == 0 and cur["price"] == 99.5, f"eval={res} price={cur['price']}")

# 8. expiry: emulate the TTL elapsing -> next read refetches from upstream
v.expire(key, 1); time.sleep(1.2)
before = upstream_gets()
p4 = catalog.get_product(1)
check("expired entry refetched from upstream", upstream_gets() == before + 1, f"price={p4['price']}")
check("refetched value re-cached with 5min TTL", 295 <= v.ttl(key) <= 300, f"TTL={v.ttl(key)}")

# 9. all products work, cache misses only fetch once each
for pid in (2, 3, 4):
    catalog.get_product(pid)
gets_a = upstream_gets()
for pid in (2, 3, 4):
    catalog.get_product(pid)
check("multiple products each fetched once", upstream_gets() == gets_a, f"upstream_gets={upstream_gets()}")

print()
print("FAILURES:", failures if failures else "none")
sys.exit(1 if failures else 0)
EOF
python3 /tmp/test_catalog.py

12  $ cd /tmp && python3 - <<'EOF'
import re
src = open("/tmp/test_catalog.py").read()
# capture the version before update_price runs (step 5)
src = src.replace(
    "catalog.update_price(1, 99.5)",
    "ver_before_update = v.get(ver_key if 'ver_key' in dir() else 'catalog:product:1:ver')\ncatalog.update_price(1, 99.5)"
)
# in step 7 use the pre-update version to simulate a stale in-flight fetch
src = src.replace(
    'ver_before = v.get(ver_key)',
    'ver_before = ver_before_update'
)
open("/tmp/test_catalog.py", "w").write(src)
EOF
python3 /tmp/test_catalog.py

13  $ cat > /tmp/stress.py <<'EOF'
import subprocess, sys, time, pathlib, urllib.request
LOG = pathlib.Path("/var/log/upstream/requests.log")
urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:8000/__reset", data=b""), timeout=5).read()

READER = r"""
import sys, time
sys.path.insert(0, '/app')
import catalog
start = float(sys.argv[1])
stale = 0
for _ in range(60):
    p = catalog.get_product(1)
    if time.time() > start and p['price'] not in (101.0, 102.0, 103.0):
        stale += 1
    time.sleep(0.01)
print(stale)
"""

# readers start before updates; updates happen mid-flight at known times
u1 = time.time() + 0.7   # -> 101.0
u2 = time.time() + 1.2   # -> 102.0
u3 = time.time() + 1.7   # -> 103.0
readers = [subprocess.Popen([sys.executable, "-c", READER, str(min(u1, time.time()+0.05))],
                             stdout=subprocess.PIPE, text=True) for _ in range(12)]

import sys; sys.path.insert(0, "/app")
import catalog
catalog.update_price(1, 101.0); time.sleep(u2 - time.time()) if u2 > time.time() else None
catalog.update_price(1, 102.0); time.sleep(max(0, u3 - time.time()))
catalog.update_price(1, 103.0)

stale_total = 0
for r in readers:
    stale_total += int(r.communicate()[0])

gets = sum(1 for l in LOG.read_text().splitlines() if l.startswith("GET"))
final = catalog.get_product(1)
print("stale reads after first update:", stale_total)
print("upstream GETs during stress:", gets)
print("final price:", final["price"])
sys.exit(0 if stale_total == 0 and final["price"] == 103.0 and gets < 20 else 1)
EOF
python3 /tmp/stress.py; echo "exit=$?"

14  $ cat > /tmp/stress.py <<'EOF'
import subprocess, sys, time, pathlib, urllib.request
LOG = pathlib.Path("/var/log/upstream/logs" "x")  # placeholder
LOG = pathlib.Path("/var/log/upstream/requests.log")
urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:8000/__reset", data=b""), timeout=5).read()

sys.path.insert(0, "/app")
import catalog, valkey
v = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)
v.delete("catalog:product:1", "catalog:product:1:ver")  # start cold, like a fresh deployment

READER = r"""
import sys, time
sys.path.insert(0, '/app')
import catalog
threshold = float(sys.argv[1])   # only calls *starting* after this judge staleness
bad = 0
seen = []
for _ in range(70):
    t0 = time.time()
    p = catalog.get_product(1)
    if t0 > threshold and p['price'] not in (101.0, 102.0, 103.0):
        bad += 1
        seen.append(p['price'])
    time.sleep(0.005)
print(bad, seen)
"""

base = time.time()
u1, u2, u3 = base + 0.8, base + 1.4, base + 2.0   # update_price returns ~ these times
threshold = u1 + 0.1
readers = [subprocess.Popen([sys.executable, "-c", READER, str(threshold)],
                             stdout=subprocess.PIPE, text=True) for _ in range(16)]

time.sleep(max(0, u1 - time.time())); catalog.update_price(1, 101.0)
time.sleep(max(0, u2 - time.time())); catalog.update_price(1, 102.0)
time.sleep(max(0, u3 - time.time())); catalog.update_price(1, 103.0)

bad_total, details = 0, []
for r in readers:
    out, _ = r.communicate()
    n, seen = out.strip().split(" ", 1)
    bad_total += int(n)
    if int(n): details.append(seen)

gets = sum(1 for l in LOG.read_text().splitlines() if l.startswith("GET"))
final = catalog.get_product(1)
cached = v.get("catalog:product:1")
import json
print("stale reads (calls starting after first update):", bad_total, details)
print("upstream GETs during stress:", gets)
print("final price:", final["price"], "| cached price:", json.loads(cached)["price"])
ok = (bad_total == 0 and final["price"] == 103.0 and json.loads(cached)["price"] == 103.0 and gets <= 20)
print("RESULT:", "PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)
EOF
python3 /tmp/stress.py; echo "exit=$?"

15  $ python3 /tmp/test_catalog.py; echo "suite exit=$?"

16  $ python3 - <<'EOF'
import sys, time
sys.path.insert(0, "/app")
import catalog, valkey
# simulate valkey outage: point the client at a dead port
catalog._valkey = valkey.Valkey(host="127.0.0.1", port=6399, socket_connect_timeout=1, decode_responses=True)
t0 = time.time()
p = catalog.get_product(2)
print("degraded read ok:", p, f"{time.time()-t0:.2f}s")
catalog.update_price(2, 55.5)   # must not raise
print("degraded update ok, upstream now:", catalog.get_product(2)["price"])
EOF

17  $ curl -s -X POST http://127.0.0.1:8000/__reset; echo; valkey-cli flushdb; valkey-cli keys 'catalog:*'; python3 -c "
import sys; sys.path.insert(0,'/app'); import catalog
print(catalog.get_product(1))"

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 1,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ cat /app/catalog.py; echo ---; ls -la /app; echo ---; cat /app/*.py | head -100

2  $ valkey-cli ping; valkey-cli info server | head; python3 -c "import valkey" 2>&1; pip list 2>/dev/null | grep -i -E "redis|valkey"

3  $ python3 -c "
import valkey, inspect
print(valkey.__version__ if hasattr(valkey,'__version__') else '')
v = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
print(v.ping())
"; curl -s -m 5 http://127.0.0.1:8000/products/1; echo

4  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products are cached in the local Valkey server (127.0.0.1:6379) for 5 minutes
so every storefront process shares the cache instead of hitting the inventory
service.  update_price() refreshes/invalidates the cached entry, so the next
get_product() call (in any process) returns the new price.
"""
import json
import os
import urllib.request

import valkey as valkey_lib

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = os.environ.get("VALKEY_HOST", "127.0.0.1")
VALKEY_PORT = int(os.environ.get("VALKEY_PORT", "6379"))
CACHE_TTL_SECONDS = 300  # 5 minutes
KEY_PREFIX = "catalog:product:"

# One shared client per process (valkey clients are thread-safe / pooled).
_valkey_client = None


def _get_cache():
    """Return a connected Valkey client, or None if Valkey is unreachable."""
    global _valkey_client
    if _valkey_client is None:
        try:
            client = valkey_lib.Valkey(
                host=VALKEY_HOST, port=VALKEY_PORT, decode_responses=True
            )
            client.ping()
            _valkey_client = client
        except Exception:
            return None
    return _valkey_client


def _cache_key(product_id: int) -> str:
    return f"{KEY_PREFIX}{int(product_id)}"


def get_product(product_id: int) -> dict:
    cache = _get_cache()
    if cache is not None:
        try:
            cached = cache.get(_cache_key(product_id))
        except Exception:
            cached = None
        if cached is not None:
            return json.loads(cached)

    with urllib.request.urlopen(
        f"{INVENTORY_URL}/products/{int(product_id)}", timeout=10
    ) as resp:
        product = json.load(resp)

    if cache is not None:
        try:
            cache.set(_cache_key(product_id), json.dumps(product), ex=CACHE_TTL_SECONDS)
        except Exception:
            pass
    return product


def update_price(product_id: int, price: float) -> None:
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{int(product_id)}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate the shared cache so the next get_product() in *any* process
    # re-fetches the fresh price from the inventory service.
    cache = _get_cache()
    if cache is not None:
        try:
            cache.delete(_cache_key(product_id))
        except Exception:
            pass
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax ok')"

5  $ cd /app && python3 - <<'EOF'
import subprocess, time, json, sys

def run(code):
    out = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, cwd="/app")
    if out.returncode != 0:
        print("ERR", out.stderr); sys.exit(1)
    return out.stdout.strip()

# process 1: cold read
t0 = time.time()
p = run("import catalog, json; print(json.dumps(catalog.get_product(1)))")
print("proc1 cold:", p, f"{time.time()-t0:.3f}s")

# process 2: should hit Valkey (fast)
t0 = time.time()
p2 = run("import catalog, json; print(json.dumps(catalog.get_product(1)))")
print("proc2 cached:", p2, f"{time.time()-t0:.3f}s")

# TTL check
ttl = run("import valkey; v=valkey.Valkey(); print(v.ttl('catalog:product:1'))")
print("ttl:", ttl)

# original price
orig = json.loads(p)["price"]

# process 3: update price
new = round(orig + 7.77, 2)
run(f"import catalog; catalog.update_price(1, {new})")

# process 4: must see new price immediately
p3 = run("import catalog, json; print(json.dumps(catalog.get_product(1)))")
got = json.loads(p3)["price"]
print("after update (proc4):", p3, "-> new price visible:", got == new)

# restore
run(f"import catalog; catalog.update_price(1, {orig})")
p4 = run("import catalog, json; print(json.dumps(catalog.get_product(1)))")
print("restored:", p4, "ok:", json.loads(p4)["price"] == orig)
EOF

6  $ cd /app && python3 - <<'EOF'
import time, json, catalog, valkey

# same-process cache hit
t0=time.time(); catalog.get_product(2)
t1=time.time(); catalog.get_product(2)
print(f"proc first={t1-t0:.3f}s hit={time.time()-t1:.4f}s")

# verify key really expires (simulate by shrinking TTL to 1s)
v = valkey.Valkey(decode_responses=True)
v.expire("catalog:product:2", 1)
time.sleep(1.5)
print("key gone after expiry:", v.get("catalog:product:2") is None)
t0=time.time(); catalog.get_product(2)
print(f"refetch after expiry ok ({time.time()-t0:.3f}s), re-cached:",
      v.get("catalog:product:2") is not None, "ttl:", v.ttl("catalog:product:2"))
EOF
valkey-cli keys 'catalog:*'

7  $ valkey-cli --scan --pattern 'catalog:*' | xargs -r valkey-cli del; valkey-cli keys 'catalog:*'; echo cleaned

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}

1  $ ls -la /app && cat /app/catalog.py

2  $ cat /etc/valkey/valkey.conf | grep -v '^#' | grep -v '^$'

3  $ python3 -c "import valkey; print(valkey.__version__ if hasattr(valkey,'__version__') else 'ok')" 2>&1; python3 -c "import redis; print(redis.__version__)" 2>&1; pip list 2>/dev/null | head -50

4  $ valkey-cli ping; curl -s -m 5 http://127.0.0.1:8000/products/1

5  $ cat > /app/catalog.py <<'EOF'
"""Product catalog used by the storefront.

get_product(product_id) -> dict    {"id": int, "name": str, "price": float, "currency": str}
update_price(product_id, price)    changes the price in the inventory service

Products read through get_product() are cached in the local Valkey server
(127.0.0.1:6379) so that every storefront process shares one cache instead of
hitting the inventory service.  Cache entries expire after 5 minutes, and
update_price() invalidates the entry so the next read gets the fresh price.
"""
import json
import urllib.request

import valkey

INVENTORY_URL = "http://127.0.0.1:8000"

VALKEY_HOST = "127.0.0.1"
VALKEY_PORT = 6379
CACHE_TTL_SECONDS = 300  # 5 minutes
KEY_PREFIX = "catalog:product:"


def _cache_key(product_id: int) -> str:
    return f"{KEY_PREFIX}{int(product_id)}"


def _get_valkey() -> valkey.Valkey:
    return valkey.Valkey(host=VALKEY_HOST, port=VALKEY_PORT, decode_responses=True)


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    client = _get_valkey()

    cached = client.get(_cache_key(product_id))
    if cached is not None:
        return json.loads(cached)

    with urllib.request.urlopen(f"{INVENTORY_URL}/products/{product_id}", timeout=10) as resp:
        product = json.load(resp)

    client.set(_cache_key(product_id), json.dumps(product), ex=CACHE_TTL_SECONDS)
    return product


def update_price(product_id: int, price: float) -> None:
    product_id = int(product_id)
    body = json.dumps({"price": price}).encode()
    req = urllib.request.Request(
        f"{INVENTORY_URL}/products/{product_id}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate the shared cache so the next get_product() (in any process)
    # returns the new price.
    _get_valkey().delete(_cache_key(product_id))
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax ok')"

6  $ cd /app && valkey-cli FLUSHDB >/dev/null && python3 - <<'EOF'
import time, sys
sys.path.insert(0, '/app')
import catalog, valkey, json

orig = catalog.get_product(1)
print("first fetch:", time.time())

t0 = time.time()
p = catalog.get_product(1)
print("second fetch took %.4fs, same data: %s" % (time.time()-t0, p == orig))

v = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
print("valkey keys:", v.keys("*"))
print("ttl:", v.ttl("catalog:product:1"))

new_price = round(orig["price"] + 1.23, 2)
catalog.update_price(1, new_price)
print("after update_price, keys:", v.keys("*"))
EOF

7  $ cd /app && valkey-cli FLUSHDB >/dev/null && python3 - <<'EOF'
import subprocess, sys, time, json, urllib.request

# Process A: warm the cache
subprocess.run([sys.executable, "-c", "import sys; sys.path.insert(0,'/app'); import catalog; catalog.get_product(1)"], check=True)

# Process B: must be served from Valkey (fast), even though it never fetched before
t0 = time.time()
b1 = subprocess.run([sys.executable, "-c", "import sys,json; sys.path.insert(0,'/app'); import catalog; print(json.dumps(catalog.get_product(1)))"], check=True, capture_output=True, text=True)
print("process B read took %.3fs (shared cache)" % (time.time()-t0))
print("B sees:", b1.stdout.strip())

# Process A changes the price
new_price = 99.99
subprocess.run([sys.executable, "-c", f"import sys; sys.path.insert(0,'/app'); import catalog; catalog.update_price(1, {new_price})"], check=True)

# Process B: next read must return the new price
t0 = time.time()
b2 = subprocess.run([sys.executable, "-c", "import sys,json; sys.path.insert(0,'/app'); import catalog; print(json.dumps(catalog.get_product(1)))"], check=True, capture_output=True, text=True)
p = json.loads(b2.stdout)
print("B after update sees price:", p["price"], "-> correct:", p["price"] == new_price, "(took %.3fs)" % (time.time()-t0))

# direct check against the service
live = json.load(urllib.request.urlopen("http://127.0.0.1:8000/products/1"))
print("service price:", live["price"])
EOF

8  $ cd /app && python3 - <<'EOF'
# verify expiry actually removes the key: temporarily use a tiny TTL to exercise the mechanism
import sys, time, json, valkey
sys.path.insert(0, '/app')
import catalog

v = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
v.flushdb()
catalog.get_product(1)
print("key present after get:", v.exists("catalog:product:1"))
v.expire("catalog:product:1", 2)   # same mechanism (EXPIRE), shortened for the test
time.sleep(2.5)
print("key present after 2.5s:", v.exists("catalog:product:1"))
print("real TTL set by code:", "(checked earlier: 300s)")
EOF

9  $ cd /app && python3 -c "
import sys; sys.path.insert(0,'/app'); import catalog
catalog.update_price(1, 84.69)   # restore original price from test
print(catalog.get_product(1))
" && valkey-cli DBSIZE

10  $ grep -rn "catalog" /app --include='*.py' | grep -v catalog.py || true; ls /app

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7",
   300
  ]
 ],
 "read12": {
  "upstream_gets": 1,
  "value": {
   "currency": "USD",
   "id": 12,
   "name": "Everyday office chair",
   "price": 449.14
  }
 },
 "read7": {
  "first": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 327.12
  },
  "first_err": null,
  "second_err": null,
  "upstream_gets": 1
 },
 "read7_after_update": {
  "err": null,
  "upstream_gets_total": 2,
  "value": {
   "currency": "USD",
   "id": 7,
   "name": "Travel blender",
   "price": 123.45
  }
 },
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 }
}