Cache in front of a slow service · Redis 8.10

1  $ cat /app/catalog.py; ls -la /app; pip list 2>/dev/null | grep -i redis; redis-cli ping; redis-server --version

2  $ grep -vE '^\s*(#|$)' /etc/redis/redis.conf; redis-cli info server | head -20; redis-cli config get maxmemory*

3  $ time curl -s http://127.0.0.1:8000/products/1; echo; curl -s -i http://127.0.0.1:8000/products/999999 | head -3

4  $ cp /app/catalog.py /tmp/catalog.py.orig
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 Redis (127.0.0.1:6379), shared by all storefront
processes, for CACHE_TTL seconds.  update_price() invalidates the cached
entry so the next get_product() in any process sees the new price.

To avoid a race where a reader fetches the old product from the inventory
service, a writer updates the price and invalidates, and then the reader
writes the stale product back into the cache, every product has a version
counter in Redis.  update_price() bumps the version; get_product() only
stores its result if the version is unchanged since before it fetched.

If Redis is unavailable, get_product() falls back to the inventory service.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL = 300  # seconds (5 minutes)
_KEY_PREFIX = "catalog:product:"

_redis = redis.Redis.from_url(
    REDIS_URL, socket_timeout=0.5, socket_connect_timeout=0.5
)

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


def _keys(product_id: int):
    base = f"{_KEY_PREFIX}{int(product_id)}"
    return base, base + ":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, ver_key = _keys(product_id)
    try:
        cached, version = _redis.mget(key, ver_key)
    except redis.RedisError:
        return _fetch(product_id)
    if cached is not None:
        return json.loads(cached)

    product = _fetch(product_id)
    try:
        _SET_IF_VERSION(
            keys=[key, ver_key],
            args=[(version or b"0").decode(), json.dumps(product), CACHE_TTL],
        )
    except redis.RedisError:
        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 after the inventory service has accepted the change.
    # Errors are propagated: silently skipping invalidation would leave a
    # stale price cached for up to CACHE_TTL.
    key, ver_key = _keys(product_id)
    pipe = _redis.pipeline(transaction=True)
    pipe.incr(ver_key)
    pipe.delete(key)
    pipe.execute()
EOF
cd /app && python3 - <<'EOF'
import time, catalog, redis, subprocess
r=redis.Redis()
t=time.time(); p=catalog.get_product(1); print(p, time.time()-t)
t=time.time(); p2=catalog.get_product(1); print(p2, time.time()-t, r.ttl("catalog:product:1"))
old=p["price"]
catalog.update_price(1, 12.34)
print(subprocess.run(["python3","-c","import catalog;print(catalog.get_product(1))"],capture_output=True,text=True).stdout)
print(catalog.get_product(1))
catalog.update_price(1, old); print(catalog.get_product(1))
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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ cat /app/catalog.py; ls -la /app; redis-cli -h 127.0.0.1 ping; redis-cli INFO server | head -20; pip list 2>/dev/null | grep -i redis; grep -v '^#' /etc/redis/redis.conf | grep -v '^$'

2  $ redis-cli CONFIG GET maxmemory*; redis-cli CONFIG GET '*rename*'; redis-cli DBSIZE; redis-cli ACL LIST; python3 --version; curl -s -w ' %{time_total}\n' http://127.0.0.1:8000/products/1

3  $ cp /app/catalog.py /tmp/catalog.py.orig; 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 Redis (shared by all storefront processes) for
CACHE_TTL_SECONDS.  update_price() invalidates the cached entry, and a per-product
version key guarded by WATCH/MULTI prevents a reader that fetched the old data
before the update from writing it back into the cache afterwards.
"""
import json
import logging
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL_SECONDS = 300
_KEY_PREFIX = "catalog:product:"

log = logging.getLogger(__name__)

# redis-py's connection pool is thread-safe and fork-aware (it resets itself
# in a child process), so one module-level client is fine for every process.
_redis = redis.Redis.from_url(
    REDIS_URL, socket_timeout=1, socket_connect_timeout=1, decode_responses=True
)


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


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


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


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

    try:
        cached = _redis.get(key)
    except redis.RedisError as exc:
        log.warning("Redis unavailable, reading product %s from inventory: %s", product_id, exc)
        return _fetch_product(product_id)
    if cached is not None:
        return json.loads(cached)

    # Cache miss: fetch from inventory and store it, but only if no
    # update_price() happened for this product while we were fetching.
    try:
        with _redis.pipeline(transaction=True) as pipe:
            pipe.watch(_version_key(product_id))
            product = _fetch_product(product_id)
            pipe.multi()
            pipe.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
            pipe.execute()
    except redis.WatchError:
        pass  # price changed concurrently; don't cache possibly stale data
    except redis.RedisError as exc:
        log.warning("Could not cache product %s: %s", product_id, exc)
        if "product" not in locals():
            return _fetch_product(product_id)
    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 cached product and bump its version so any in-flight
    # cache fill that read the old price is discarded (WATCH fails).
    with _redis.pipeline(transaction=True) as pipe:
        pipe.incr(_version_key(product_id))
        pipe.delete(_cache_key(product_id))
        pipe.execute()
EOF
cd /app && python3 - <<'EOF'
import time, catalog, subprocess, redis
r = redis.Redis()
r.delete("catalog:product:1")
t=time.time(); p=catalog.get_product(1); print(p, round(time.time()-t,3))
t=time.time(); p=catalog.get_product(1); print(p, round(time.time()-t,4))
print("ttl", r.ttl("catalog:product:1"))
old=p["price"]
catalog.update_price(1, 12.34)
out = subprocess.check_output(["python3","-c","import catalog;print(catalog.get_product(1))"])
print("other proc:", out.decode().strip())
print("this proc:", catalog.get_product(1))
catalog.update_price(1, old); print("restored:", catalog.get_product(1))
EOF

4  $ cd /app && python3 - <<'EOF'
import threading, time, catalog, redis
r = redis.Redis(); r.delete("catalog:product:2")
old = catalog._fetch_product(2)["price"]
th = threading.Thread(target=catalog.get_product, args=(2,)); th.start()
time.sleep(0.05); catalog.update_price(2, 99.99); th.join()
print("cached after race:", r.get("catalog:product:2"))
print("get:", catalog.get_product(2))
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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ cat /app/catalog.py; ls -la /app; pip list 2>/dev/null | grep -i redis; redis-cli ping; redis-cli INFO server | head -5; grep -v '^#' /etc/redis/redis.conf | grep -v '^$'

2  $ redis-cli CONFIG GET maxmemory*; redis-cli CONFIG GET save; redis-cli DBSIZE; redis-cli INFO keyspace; curl -s -w ' %{time_total}\n' 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 Redis (shared by all storefront processes) for
CACHE_TTL seconds. update_price() invalidates the cached entry so the next
get_product() in any process sees the new price. A per-product version
counter prevents a reader that fetched stale data *before* an update from
re-populating the cache with that stale data *after* the invalidation.
If Redis is unavailable, calls fall through to the inventory service.
"""
import json
import logging
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL = 300  # seconds (5 minutes)

log = logging.getLogger(__name__)

_redis = redis.Redis.from_url(
    REDIS_URL, socket_timeout=0.5, socket_connect_timeout=0.5, decode_responses=True
)

# Store the product only if no update_price() happened since we read the version.
_SET_IF_VERSION = _redis.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 _key(product_id: int) -> str:
    return f"catalog:product:{int(product_id)}"


def _ver_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 = _key(product_id), _ver_key(product_id)
    version = None
    try:
        cached, version = _redis.mget(key, vkey)
        if cached is not None:
            return json.loads(cached)
        version = version or "0"
    except redis.RedisError as exc:
        log.warning("redis read failed, bypassing cache: %s", exc)
        version = None

    product = _fetch(product_id)

    if version is not None:
        try:
            _SET_IF_VERSION(keys=[key, vkey], args=[version, json.dumps(product), CACHE_TTL])
        except redis.RedisError as exc:
            log.warning("redis write failed: %s", 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 (blocks in-flight stale writes) and drop the entry.
    key, vkey = _key(product_id), _ver_key(product_id)
    pipe = _redis.pipeline(transaction=True)
    pipe.incr(vkey)
    pipe.delete(key)
    pipe.execute()
EOF
cd /app && python3 - <<'EOF'
import time, catalog, subprocess
t=time.time(); p=catalog.get_product(1); print(p, round(time.time()-t,3))
t=time.time(); p=catalog.get_product(1); print(p, round(time.time()-t,3))
print(subprocess.run(["redis-cli","TTL","catalog:product:1"],capture_output=True,text=True).stdout)
old=p["price"]; catalog.update_price(1, 12.34)
out=subprocess.run(["python3","-c","import catalog;print(catalog.get_product(1))"],capture_output=True,text=True,cwd="/app").stdout
print("other process:", out)
catalog.update_price(1, old); print(catalog.get_product(1))
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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ ls -la /app && sed -n '1,240p' /app/catalog.py && find /app -maxdepth 2 -type f | head -80

2  $ redis-cli PING; grep -E '^(loadmodule|bind|port|requirepass|protected-mode)' /etc/redis/redis.conf; python - <<'PY'
try: import redis; print('redis-py',redis.__version__)
except Exception as e: print(repr(e))
PY

3  $ python3 - <<'PY'
try: import redis; print('redis-py',redis.__version__)
except Exception as e: print(repr(e))
PY
ls /app -la; curl -sS -i http://127.0.0.1:8000/products/1 | head -30; which pip3 || true; ls /usr/local/lib/python*/site-packages/redis* 2>/dev/null | head

4  $ 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 redis

INVENTORY_URL = "http://127.0.0.1:8000"
CACHE_TTL_SECONDS = 300
_cache = redis.Redis(host="127.0.0.1", port=6379, db=0)


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


def _product_lock(product_id: int):
    # Serialize misses and writes across storefront processes. In particular,
    # a read started before a price update cannot refill the cache afterward.
    return _cache.lock(f"catalog:product-lock:{product_id}", timeout=30)


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

    with _product_lock(product_id):
        # Another process may have populated the cache while we waited.
        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.setex(key, CACHE_TTL_SECONDS, json.dumps(product))
        return product


def update_price(product_id: int, price: float) -> None:
    product_id = int(product_id)
    with _product_lock(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 only after a successful update. The next read will fetch
        # the new price; the lock prevents an older in-flight miss from winning.
        _cache.delete(_product_key(product_id))
PY
cd /app && python3 - <<'PY'
import catalog, redis, subprocess, sys
p = catalog.get_product(1)
key = 'catalog:product:1'
r = redis.Redis()
assert r.ttl(key) > 0 and r.ttl(key) <= 300
assert catalog.get_product(1) == p
# Check the cache is shared with a separate interpreter.
out = subprocess.check_output([sys.executable, '-c', 'import catalog; print(catalog.get_product(1)["price"])'], text=True)
assert float(out) == p['price']
new_price = p['price'] + 1
try:
    assert catalog.update_price(1, new_price) is None
    assert r.get(key) is None
    out = subprocess.check_output([sys.executable, '-c', 'import catalog; print(catalog.get_product(1)["price"])'], text=True)
    assert float(out) == new_price
finally:
    catalog.update_price(1, p['price'])
    assert catalog.get_product(1)['price'] == p['price']
print('Cache TTL, cross-process read, and price invalidation verified')
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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ ls -la /app && cat /app/catalog.py && ls -la /app && cat /etc/redis/redis.conf | head -65

2  $ python - <<'PY'
try:
 import redis; print(redis.__version__,redis.__file__)
except Exception as e: print(type(e).__name__,e)
PY
which redis-cli && redis-cli ping

3  $ ls / && find /app -maxdepth 2 -type f -print && ps aux | grep -E 'inventory|redis' | head

4  $ python3 - <<'PY'
try:
 import redis; print(redis.__version__,redis.__file__)
except Exception as e: print(type(e).__name__,e)
PY
python3 -m pip --version; ls -la /usr/local/lib/python*/site-packages | head; curl -s -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8000/products/1

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
"""
import json
import urllib.request

import redis

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


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


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

    # Serialize misses with writes across processes: a read started before a
    # price change must not populate the cache after that change is committed.
    with _cache.lock(f"{key}:lock", timeout=30):
        cached = _cache.get(key)
        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)
        _cache.setex(key, CACHE_TTL_SECONDS, json.dumps(product))
        return product


def update_price(product_id: int, price: float) -> None:
    key = _product_key(product_id)
    with _cache.lock(f"{key}:lock", timeout=30):
        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()
        # Only invalidate after the inventory service has accepted the update.
        _cache.delete(key)
PY
python3 - <<'PY'
import sys
sys.path.insert(0, '/app')
import catalog
print(catalog.get_product(1))
print('TTL', catalog._cache.ttl(catalog._product_key(1)))
print(catalog.get_product(1))
PY

6  $ python3 - <<'PY'
import sys, subprocess
sys.path.insert(0, '/app')
import catalog
product = catalog.get_product(1)
original = product['price']
try:
    catalog.update_price(1, original + 0.01)
    subprocess.run(['python3', '-c', 'import sys; sys.path.insert(0, "/app"); import catalog; p=catalog.get_product(1); print(p); assert p["price"] == '+repr(round(original+0.01, 2))], check=True)
finally:
    catalog.update_price(1, original)
    assert catalog.get_product(1)['price'] == original
print('restored original price')
PY

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:v1: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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ ls -la /app && cat /app/catalog.py && cat /etc/redis/redis.conf | head -70 && ls -la /app

2  $ python - <<'PY'
try:
 import redis;print(redis.__version__,redis.__file__)
except Exception as e:print(repr(e))
PY
redis-cli ping; curl -s -i http://127.0.0.1:8000/products/1 | head; ls /app /tmp | head

3  $ command -v python3; python3 - <<'PY'
try:
 import redis;print(redis.__version__,redis.__file__)
except Exception as e:print(repr(e))
PY
ls /usr/local/lib/python*/site-packages | head; ls /usr/bin/redis*; python3 -m pip --version || true; ls -la /

4  $ 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 redis

INVENTORY_URL = "http://127.0.0.1:8000"
_CACHE_TTL = 300
_cache = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

# Only populate the cache if no price update completed while the inventory
# request was in flight. Checking and setting must be one atomic operation.
_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
"""
_INVALIDATE = """
redis.call('INCR', KEYS[2])
redis.call('DEL', KEYS[1])
"""


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


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

        version = _cache.get(version_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, version_key,
            version, json.dumps(product), _CACHE_TTL,
        ):
            return product
        # An update won the race against this read: do not return old data.


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

    # The version is intentionally persistent: an in-flight read must never
    # mistake an expired/recreated version for the one it observed earlier.
    key, version_key = _keys(product_id)
    _cache.eval(_INVALIDATE, 2, key, version_key)
PY
python3 -m py_compile /app/catalog.py
cd /app && python3 - <<'PY'
import catalog, json, subprocess
pid = 1
key, version = catalog._keys(pid)
original = catalog.get_product(pid)
print('initial:', original)
print('cached TTL:', catalog._cache.ttl(key))
again = catalog.get_product(pid)
assert again == original
try:
    catalog.update_price(pid, original['price'] + 0.01)
    assert catalog._cache.get(key) is None
    updated = catalog.get_product(pid)
    assert updated['price'] == round(original['price'] + 0.01, 2), updated
    print('updated:', updated, 'TTL:', catalog._cache.ttl(key))
finally:
    catalog.update_price(pid, original['price'])
    assert catalog.get_product(pid)['price'] == original['price']
PY

5  $ cd /app && python3 - <<'PY'
import catalog, subprocess, sys
key, _ = catalog._keys(1)
catalog.get_product(1)
result = subprocess.check_output([sys.executable, '-c', 'import catalog; print(catalog.get_product(1))'], text=True)
assert 'Pro kettle' in result
assert catalog._cache.ttl(key) > 0
print('Cross-process cache read OK; TTL:', catalog._cache.ttl(key))
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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

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

2  $ ls -la /app; find /app -name "*.py" | head -50

3  $ python3 --version; which python3 pip3 redis-cli; pip3 list 2>/dev/null | grep -i -E "redis|flask|fastapi|requests"

4  $ redis-cli -h 127.0.0.1 -p 6379 ping; echo "---"; redis-cli -h 127.0.0.1 -p 6379 info server | head -20; echo "---config---"; grep -E "^(bind|port|requirepass|maxmemory|daemonize)" /etc/redis/redis.conf

5  $ python3 -c "import redis; print(redis.__version__)"; echo "---"; pip3 show redis 2>/dev/null | head -5

6  $ ss -lntp 2>/dev/null | grep -E "8000|6379"; echo "---"; curl -s -m 3 http://127.0.0.1:8000/products/1; echo; echo "---procs---"; ps aux | grep -i -E "inventory|uvicorn|gunicorn|flask" | grep -v grep

7  $ grep -vE "^\s*#|^\s*$" /etc/redis/redis.conf

8  $ echo "GET 1:"; curl -s -m 3 http://127.0.0.1:8000/products/1; echo
echo "PUT 1 -> 99.5:"; curl -s -m 3 -X PUT -H "Content-Type: application/json" -d '{"price": 99.5}' http://127.0.0.1:8000/products/1; echo
echo "GET 1:"; curl -s -m 3 http://127.0.0.1:8000/products/1; echo
echo "GET 2:"; curl -s -m 3 http://127.0.0.1:8000/products/2; echo
echo "GET 999:"; curl -s -m 3 -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/products/999

9  $ cd /app && python3 -c "
import redis, json
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
print('ping', r.ping())
r.set('t:key', json.dumps({'a':1}), ex=300)
print('get', r.get('t:key'), 'ttl', r.ttl('t:key'))
print('module list', r.module_list())
r.delete('t:key')
"

10  $ python3 -c "import redis.connection as c, inspect; src=inspect.getsource(c.ConnectionPool); import re; print('\n'.join(l for l in src.splitlines() if 'pid' in l.lower() or 'def reset' in l or 'def get_connection' in l or 'def _checkpid' in l))"

11  $ 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 are cached in Redis so that every storefront process shares one cache
instead of hitting the (slow, ~300 ms) inventory service on every call.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

# --- Redis cache configuration --------------------------------------------
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL = 300  # seconds: a cached product expires after 5 minutes

# Version markers live longer than the entries they guard, so a version
# counter can never be reset (and then reused) while an entry written under
# the previous version could still be cached.
_VERSION_TTL = CACHE_TTL * 12

# A single module-level client shared by all threads/processes.  redis-py's
# connection pool is fork-safe (it detects a changed PID and reconnects), so
# importing this module in a forked worker just works.
_redis = redis.Redis.from_url(REDIS_URL, decode_responses=True, socket_timeout=1)


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


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


def _get_version(product_id: int) -> int:
    """Current cache generation for a product (0 when never invalidated)."""
    value = _redis.get(_version_key(product_id))
    return int(value) if value is not None else 0


def _bump_version(product_id: int) -> int:
    """Invalidate every cached entry for a product, returning the new version."""
    pipe = _redis.pipeline()
    pipe.incr(_version_key(product_id))
    pipe.expire(_version_key(product_id), _VERSION_TTL)
    version, _ = pipe.execute()
    return int(version)


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)

    # Read the current version and the entry cached under it.  A cache failure
    # must not take the storefront down, so fall back to the inventory service.
    try:
        version = _get_version(product_id)
        cached = _redis.get(_cache_key(product_id, version))
    except redis.RedisError:
        return _fetch_product(product_id)

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

    product = _fetch_product(product_id)
    try:
        _redis.set(_cache_key(product_id, version), json.dumps(product), ex=CACHE_TTL)
    except redis.RedisError:
        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/{product_id}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        raw = resp.read()
    try:
        updated = json.loads(raw)
    except (ValueError, TypeError):
        updated = None

    # Invalidate *after* the inventory write has completed, so no process can
    # keep serving the old price.  Bumping the version makes every entry cached
    # under the previous version unreachable, which also stops a reader that is
    # concurrently fetching the old value from resurrecting it in the cache.
    try:
        version = _bump_version(product_id)
        if updated is not None:
            # Warm the cache with the value we already have.
            _redis.set(_cache_key(product_id, version), json.dumps(updated), ex=CACHE_TTL)
    except redis.RedisError:
        # Best effort: at minimum try to drop whatever is cached right now.
        try:
            _redis.delete(_cache_key(product_id, _get_version(product_id)))
        except redis.RedisError:
            pass
PY
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

12  $ tr '\0' ' ' < /proc/10/cmdline; echo; echo "---"; ls -la /proc/10/cwd 2>/dev/null; cat /proc/10/cwd/*.py 2>/dev/null | head -60

13  $ redis-cli --scan --pattern 'catalog:*' | head; echo "(existing catalog keys above, if any)"

14  $ cat /opt/upstream/server.py

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 are cached in Redis so that all storefront processes share one cache
instead of hitting the (slow, ~300 ms) inventory service on every call.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

# --- Redis cache configuration --------------------------------------------
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL = 300  # seconds: a cached product expires after 5 minutes

# Version markers live longer than the entries they guard, so a version
# counter can never be reset (and then reused) while an entry written under
# the previous version could still be cached.
_VERSION_TTL = CACHE_TTL * 12

# A single module-level client shared by all threads in the process.  redis-py's
# connection pool is fork-safe (it notices a changed PID and reconnects), so
# importing this module in a forked worker just works.
_redis = redis.Redis.from_url(REDIS_URL, decode_responses=True, socket_timeout=1)


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


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


def _get_version(product_id: int) -> int:
    """Current cache generation for a product (0 when it was never invalidated)."""
    value = _redis.get(_version_key(product_id))
    return int(value) if value is not None else 0


def _bump_version(product_id: int) -> int:
    """Invalidate every cached entry for a product, returning the new version."""
    pipe = _redis.pipeline()
    pipe.incr(_version_key(product_id))
    pipe.expire(_version_key(product_id), _VERSION_TTL)
    version, _ = pipe.execute()
    return int(version)


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)

    # Read the current version together with the entry cached under it.  A
    # cache failure must never take the storefront down, so fall back to the
    # inventory service if Redis is unavailable.
    try:
        version = _get_version(product_id)
        cached = _redis.get(_cache_key(product_id, version))
    except redis.RedisError:
        return _fetch_product(product_id)

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

    product = _fetch_product(product_id)
    try:
        _redis.set(_cache_key(product_id, version), json.dumps(product), ex=CACHE_TTL)
    except redis.RedisError:
        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/{product_id}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate the cache *after* the inventory write has completed, so no
    # process keeps serving the old price.  Bumping the version makes every
    # entry cached under the previous version unreachable; that also prevents a
    # reader that is concurrently fetching the old value from resurrecting it,
    # because it would store it under the now-obsolete version.
    try:
        _bump_version(product_id)
    except redis.RedisError:
        # Best effort: at least drop whatever is cached right now.
        try:
            _redis.delete(_cache_key(product_id, _get_version(product_id)))
        except redis.RedisError:
            pass
PY
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

16  $ cat > /tmp/test_cache.py <<'PY'
import json, sys, time, urllib.request, subprocess, threading
sys.path.insert(0, "/app")
import catalog

LOG = "/var/log/upstream/requests.log"

def reset():
    urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:8000/__reset", data=b"", method="POST")).read()
    r = catalog._redis
    for k in r.scan_iter("catalog:*"):
        r.delete(k)

def ngets():
    with open(LOG) as f:
        return sum(1 for line in f if line.startswith("GET"))

def nputs():
    with open(LOG) as f:
        return sum(1 for line in f if line.startswith("PUT"))

def true_price(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))["price"]

def check(name, cond, extra=""):
    print(("PASS " if cond else "FAIL ") + name + ("  " + extra if extra else ""))
    if not cond:
        check.failed = True
check.failed = False

# ---- 1. cold read hits upstream once; warm read served from Redis ----
reset()
g0 = ngets()
p1 = catalog.get_product(1)
g1 = ngets()
p1b = catalog.get_product(1)
g2 = ngets()
check("cold read calls inventory once", g1 - g0 == 1, f"(GETs={g1-g0})")
check("warm read is served from Redis (no upstream GET)", g2 - g1 == 0, f"(GETs={g2-g1})")
check("cached value matches", p1 == p1b and p1["id"] == 1)
check("return shape intact", set(p1) == {"id", "name", "price", "currency"})

# ---- 2. TTL is 5 minutes ----
v = catalog._get_version(1)
ttl = catalog._redis.ttl(catalog._cache_key(1, v))
check("cached entry TTL is 5 minutes", 0 < ttl <= 300, f"(ttl={ttl})")
check("TTL is ~300s, not less", ttl >= 295, f"(ttl={ttl})")

# ---- 3. cache shared across processes (not per-process) ----
reset()
g0 = ngets()
catalog.get_product(2)
code = ("import sys; sys.path.insert(0,'/app'); import catalog; "
        "p=catalog.get_product(2); assert p['id']==2; print('child ok')")
out = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
g1 = ngets()
check("second *process* served from shared Redis cache", g1 - g0 == 1 and "child ok" in out.stdout,
      f"(GETs={g1-g0}) {out.stderr.strip()}")

# ---- 4. update_price invalidates cache, in any process ----
reset()
catalog.get_product(3)                     # populate cache
old = catalog.get_product(3)["price"]
catalog.update_price(3, 123.45)            # change price in this process
new = catalog.get_product(3)["price"]      # same process
check("same-process get after update returns new price", new == 123.45, f"(got {new})")
check("upstream really has the new price", true_price(3) == 123.45)

# cross-process: warm cache in a child, update here, read in another child
reset()
child_warm = ("import sys; sys.path.insert(0,'/app'); import catalog; "
              "print(catalog.get_product(4)['price'])")
child_read = ("import sys; sys.path.insert(0,'/app'); import catalog; "
              "print(catalog.get_product(4)['price'])")
subprocess.run([sys.executable, "-c", child_warm], capture_output=True, text=True)  # warms shared cache
catalog.update_price(4, 55.55)
out = subprocess.run([sys.executable, "-c", child_read], capture_output=True, text=True)
got = float(out.stdout.strip())
check("other process sees new price after update", got == 55.55, f"(got {got}) {out.stderr.strip()}")

# ---- 5. update_price affects next get even without a prior read (nothing cached) ----
reset()
catalog.update_price(5, 7.77)
p = catalog.get_product(5)
check("get after update with empty cache returns new price", p["price"] == 7.77, f"(got {p['price']})")

# ---- 6. concurrent readers don't stampede (single-flight-ish) ----
reset()
g0 = ngets()
threads = [threading.Thread(target=catalog.get_product, args=(6,)) for _ in range(40)]
for t in threads: t.start()
for t in threads: t.join()
g1 = ngets()
check("concurrent reads share one cache entry (<=2 upstream GETs)", g1 - g0 <= 2, f"(GETs={g1-g0})")

# ---- 7. invalidation is race-safe: reader started before update must not resurrect stale ----
reset()
catalog.get_product(7)
res = {}
def slow_reader():
    time.sleep(0.05)  # start after the update fires
    res["val"] = catalog.get_product(7)["price"]
# force a stale write window: monkeypatch fetch to return old price slowly
orig = catalog._fetch_product
def slow_fetch(pid):
    time.sleep(0.2)
    return orig(pid)
catalog._fetch_product = slow_fetch
catalog._redis.delete(catalog._cache_key(7, catalog._get_version(7)))  # force refetch
t = threading.Thread(target=lambda: res.update(stale=catalog.get_product(7)["price"]))
t.start()
time.sleep(0.05)
catalog.update_price(7, 42.42)  # fires while the slow reader is in flight
t.join()
catalog._fetch_product = orig
after = catalog.get_product(7)["price"]
check("stale in-flight read cannot resurrect old price", after == 42.42, f"(got {after})")

reset()
print("\nRESULT:", "FAILURES PRESENT" if check.failed else "ALL PASS")
sys.exit(1 if check.failed else 0)
PY
cd /app && python3 /tmp/test_cache.py

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 are cached in Redis so that all storefront processes share one cache
instead of hitting the (slow, ~300 ms) inventory service on every call.
"""
import json
import time
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

# --- Redis cache configuration --------------------------------------------
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL = 300  # seconds: a cached product expires after 5 minutes

# Version markers live longer than the entries they guard, so a version
# counter can never be reset (and then reused) while an entry written under
# the previous version could still be cached.
_VERSION_TTL = CACHE_TTL * 12

# Single-flight: while one process fetches a cold key the others wait up to
# this long for the shared cache instead of hammering the inventory service.
_LOCK_TTL = 10.0   # seconds the fetch lock may live before it expires
_WAIT = 1.0        # seconds a follower waits for the leader before fetching
_POLL = 0.02       # seconds between follower cache checks

# A single module-level client shared by all threads in the process.  redis-py's
# connection pool is fork-safe (it notices a changed PID and reconnects), so
# importing this module in a forked worker just works.
_redis = redis.Redis.from_url(REDIS_URL, decode_responses=True, socket_timeout=1)


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


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


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


def _get_version(product_id: int) -> int:
    """Current cache generation for a product (0 when it was never invalidated)."""
    value = _redis.get(_version_key(product_id))
    return int(value) if value is not None else 0


def _bump_version(product_id: int) -> int:
    """Invalidate every cached entry for a product, returning the new version."""
    pipe = _redis.pipeline()
    pipe.incr(_version_key(product_id))
    pipe.expire(_version_key(product_id), _VERSION_TTL)
    version, _ = pipe.execute()
    return int(version)


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)

    # Read the current version together with the entry cached under it.  A
    # cache failure must never take the storefront down, so fall back to the
    # inventory service if Redis is unavailable.
    try:
        version = _get_version(product_id)
        key = _cache_key(product_id, version)
        cached = _redis.get(key)
    except redis.RedisError:
        return _fetch_product(product_id)

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

    # Cache miss.  Let a single process fetch while the others briefly wait for
    # it, so a cold (or just-invalidated) key does not stampede the inventory
    # service when many storefront processes ask for it at once.
    lock_key = _lock_key(product_id, version)
    try:
        is_leader = _redis.set(lock_key, "1", nx=True, ex=_LOCK_TTL)
    except redis.RedisError:
        return _fetch_product(product_id)

    if not is_leader:
        deadline = time.monotonic() + _WAIT
        while time.monotonic() < deadline:
            time.sleep(_POLL)
            try:
                cached = _redis.get(key)
            except redis.RedisError:
                break
            if cached is not None:
                return json.loads(cached)
        # The leader is slow or gone; don't block the caller any longer.
        return _fetch_product(product_id)

    try:
        product = _fetch_product(product_id)
    except BaseException:
        # Release the lock so waiters do not have to time out.
        try:
            _redis.delete(lock_key)
        except redis.RedisError:
            pass
        raise
    try:
        _redis.set(key, json.dumps(product), ex=CACHE_TTL)
    except redis.RedisError:
        pass
    finally:
        try:
            _redis.delete(lock_key)
        except redis.RedisError:
            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/{product_id}",
        data=body,
        method="PUT",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        resp.read()

    # Invalidate the cache *after* the inventory write has completed, so no
    # process keeps serving the old price.  Bumping the version makes every
    # entry cached under the previous version unreachable; that also prevents a
    # reader that is concurrently fetching the old value from resurrecting it,
    # because it would store it under the now-obsolete version.
    try:
        _bump_version(product_id)
    except redis.RedisError:
        # Best effort: at least drop whatever is cached right now.
        try:
            _redis.delete(_cache_key(product_id, _get_version(product_id)))
        except redis.RedisError:
            pass
PY
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

18  $ 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 are cached in Redis so that every storefront process shares one cache
instead of hitting the (slow, ~300 ms) inventory service on every call.
"""
import json
import time
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

# --- Redis cache configuration --------------------------------------------
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL = 300  # seconds: a cached product expires after 5 minutes

# Version markers live longer than the entries they guard, so a version
# counter can never be reset (and then reused) while an entry written under
# the previous version could still be cached.
_VERSION_TTL = CACHE_TTL * 12

# Single-flight: while one process fetches a cold key the others wait (briefly)
# for the shared cache instead of all hammering the inventory service.
_LOCK_TTL = 10.0  # seconds the fetch lock may live before it expires
_WAIT = 1.0       # seconds a waiter waits for the leader before fetching itself
_POLL = 0.02      # seconds between waiter cache checks
_RETRIES = 4      # times to restart a read that was invalidated mid-flight

# A single module-level client shared by all threads in the process.  redis-py's
# connection pool is fork-safe (it notices a changed PID and reconnects), so
# importing this module in a forked worker just works.
_redis = redis.Redis.from_url(
    REDIS_URL, decode_responses=True, socket_timeout=2, socket_connect_timeout=2
)


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


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


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


def _get_version(product_id: int) -> int:
    """Current cache generation for a product (0 when it was never invalidated)."""
    value = _redis.get(_version_key(product_id))
    return int(value) if value is not None else 0


def _bump_version(product_id: int) -> int:
    """Invalidate every cached entry for a product, returning the new version."""
    pipe = _redis.pipeline()
    pipe.incr(_version_key(product_id))
    pipe.expire(_version_key(product_id), _VERSION_TTL)
    version, _ = pipe.execute()
    return int(version)


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)

    for attempt in range(_RETRIES):
        # Read the current version together with the entry cached under it.  A
        # cache failure must never take the storefront down, so fall back to
        # the inventory service if Redis is unavailable.
        try:
            version = _get_version(product_id)
            key = _cache_key(product_id, version)
            cached = _redis.get(key)
        except redis.RedisError:
            return _fetch_product(product_id)

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

        # Cache miss.  Let a single process fetch while the others briefly wait
        # for it, so a cold (or just-invalidated) key does not stampede the
        # inventory service when many processes ask for it at once.
        lock_key = _lock_key(product_id, version)
        try:
            is_leader = _redis.set(lock_key, "1", nx=True, ex=_LOCK_TTL)
        except redis.RedisError:
            return _fetch_product(product_id)

        if not is_leader:
            deadline = time.monotonic() + _WAIT
            while time.monotonic() < deadline:
                time.sleep(_POLL)
                try:
                    cached = _redis.get(key)
                    if cached is not None:
                        return json.loads(cached)
                    if _get_version(product_id) != version:
                        break  # invalidated while we waited: restart the read
                    if not _redis.exists(lock_key):
                        break  # leader gave up: fetch it ourselves
                except redis.RedisError:
                    break
            else:
                # Waited the full time and the leader is still not done.
                return _fetch_product(product_id)
            continue  # retry with the (possibly new) version

        # Leader: fetch once and populate the shared cache.
        try:
            product = _fetch_product(product_id)
        except BaseException:
            try:
                _redis.delete(lock_key)  # let waiters retry promptly
            except redis.RedisError:
                pass
            raise
        try:
            _redis.set(key, json.dumps(product), ex=CACHE_TTL)
        except redis.RedisError:
            pass
        finally:
            try:
                _redis.delete(lock_key)
            except redis.RedisError:
                pass
        return product

    return _fetch_product(product_id)


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 cache *after* the inventory write has completed, so no
    # process keeps serving the old price.  Bumping the version makes every
    # entry cached under the previous version unreachable; that also prevents a
    # reader that is concurrently fetching the old value from resurrecting it,
    # because it would store it under the now-obsolete version.
    try:
        _bump_version(product_id)
    except redis.RedisError:
        # Best effort: at least drop whatever is cached right now.
        try:
            _redis.delete(_cache_key(product_id, _get_version(product_id)))
        except redis.RedisError:
            pass
PY
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax OK')"

19  $ cd /app && python3 /tmp/test_cache.py

20  $ cd /app && python3 -c "
import redis
r = redis.Redis(decode_responses=True)
for ex in (10, 10.0, 300, 300.0):
    try:
        print(ex, '->', r.set('dbg', '1', nx=True, ex=ex), r.ttl('dbg'))
    except Exception as e:
        print(ex, '-> ERROR', type(e).__name__, e)
    r.delete('dbg')
"

21  $ cd /app && sed -i 's/^_LOCK_TTL = 10\.0 .*/_LOCK_TTL = 10    # seconds the fetch lock may live before it expires/' catalog.py && grep -n "_LOCK_TTL = " catalog.py && python3 /tmp/test_cache.py

22  $ cat > /tmp/test_stress.py <<'PY'
import json, sys, time, urllib.request, subprocess, random, os
sys.path.insert(0, "/app")
import catalog

LOG = "/var/log/upstream/requests.log"
def reset():
    urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:8000/__reset", data=b"", method="POST")).read()
    for k in catalog._redis.scan_iter("catalog:*"): catalog._redis.delete(k)
def ngets():
    with open(LOG) as f: return sum(1 for l in f if l.startswith("GET"))
def true_price(pid):
    r = urllib.request.Request(f"http://127.0.0.1:8000/products/{pid}", headers={"X-Inventory-Audit":"1"})
    return json.load(urllib.request.urlopen(r))["price"]
fails=[]
def check(n,c,e=""):
    print(("PASS " if c else "FAIL ")+n+("  "+e if e else "")); 
    if not c: fails.append(n)

# --- A. many PROCESSES cold-read the same product: should be ~1 upstream GET ---
reset(); g0=ngets()
child=("import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(10)['price'])")
ps=[subprocess.Popen([sys.executable,"-c",child],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) for _ in range(20)]
outs=[p.communicate() for p in ps]
g1=ngets()
ok=all("Traceback" not in o[1] for o in outs)
check("20 processes cold-read same product -> 1 upstream GET", g1-g0==1, f"(GETs={g1-g0})")
check("all 20 processes got a value", ok and all(o[0].strip() for o in outs))

# --- B. mixed stress: writer flips price while readers read; never a stale value ---
reset()
prices=[10.0,20.0,30.0,40.0,50.0]
def reader(res, idx):
    for _ in range(30):
        res.append(catalog.get_product(11)["price"]); time.sleep(0.001)
def writer():
    for p in prices:
        catalog.update_price(11, p); time.sleep(0.05)
import threading
res=[]
ts=[threading.Thread(target=reader,args=(res,i)) for i in range(8)]
w=threading.Thread(target=writer)
for t in ts: t.start()
w.start(); w.join()
for t in ts: t.join()
# final settle
catalog.update_price(11, 999.0)
final=catalog.get_product(11)["price"]
bad=[x for x in res if x not in prices]
check("no reader ever saw a value the writer never set", not bad, f"(bad={set(bad)})")
check("after last update, reads return the final price", final==999.0, f"(got {final})")
check("inventory holds final price", true_price(11)==999.0)

# --- C. Redis-down fallback: get_product still works via upstream ---
reset()
good=catalog._redis
class Broken:
    def __getattr__(self, name):
        def f(*a, **k): raise catalog.redis.RedisError("down")
        return f
catalog._redis=Broken()
try:
    g0=ngets(); p=catalog.get_product(12); g1=ngets()
    check("get_product works when Redis is down (falls back upstream)", p["id"]==12 and g1-g0==1, f"(GETs={g1-g0})")
    catalog.update_price(12, 1.23)  # must not raise
    check("update_price works when Redis is down", true_price(12)==1.23)
finally:
    catalog._redis=good

# --- D. no leftover lock keys, and cache key present with ~300s TTL ---
reset(); catalog.get_product(13)
locks=list(catalog._redis.scan_iter("catalog:*:lock"))
check("no leftover fetch-lock keys", not locks, f"({locks})")
v=catalog._get_version(13)
check("cache entry present after read", catalog._redis.exists(catalog._cache_key(13,v))==1)

reset()
print("\nRESULT:", "FAILURES: "+", ".join(fails) if fails else "ALL PASS")
sys.exit(1 if fails else 0)
PY
cd /app && python3 /tmp/test_stress.py

23  $ cd /app && python3 -c "
import inspect, catalog
print('get_product', inspect.signature(catalog.get_product))
print('update_price', inspect.signature(catalog.update_price))
import typing
print('get_product annotations:', catalog.get_product.__annotations__)
print('update_price annotations:', catalog.update_price.__annotations__)
# module import must not require Redis to be up (lazy) - it is, but ensure no connect at import
print('module imported cleanly')
"

24  $ cd /app && python3 - <<'PY'
# Demonstrate a warm-read benchmark: thousands of reads, zero upstream GETs.
import sys, time, urllib.request, json
sys.path.insert(0, "/app")
import catalog
LOG="/var/log/upstream/requests.log"
urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:8000/__reset", data=b"", method="POST")).read()
for k in catalog._redis.scan_iter("catalog:*"): catalog._redis.delete(k)
def ngets():
    with open(LOG) as f: return sum(1 for l in f if l.startswith("GET"))
catalog.get_product(1)  # warm
g0=ngets()
N=5000
t=time.perf_counter()
for _ in range(N): catalog.get_product(1)
dt=time.perf_counter()-t
g1=ngets()
print(f"{N} warm reads in {dt*1000:.0f} ms ({N/dt:,.0f} reads/s), upstream GETs during loop = {g1-g0}")
print("stored key:", [k for k in catalog._redis.scan_iter('catalog:*')])
PY

25  $ cd /app && python3 - <<'PY'
import urllib.request, redis
# restore inventory original data + empty request log
urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:8000/__reset", data=b"", method="POST")).read()
r = redis.Redis(decode_responses=True)
n = 0
for k in r.scan_iter("catalog:*"):
    r.delete(k); n += 1
print(f"deleted {n} catalog keys; remaining:", list(r.scan_iter("catalog:*")))
PY
rm -f /tmp/test_cache.py /tmp/test_stress.py
echo "--- final sanity: cold get + warm get on clean state ---"
cd /app && python3 -c "
import sys; sys.path.insert(0,'/app'); import catalog
print('get:', catalog.get_product(1))
print('get again (cached):', catalog.get_product(1))
"
echo "--- upstream request log ---"; cat /var/log/upstream/requests.log
echo "--- catalog.py ---"; wc -l /app/catalog.py

26  $ cd /app && python3 -c "
import urllib.request, redis
urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:8000/__reset', data=b'', method='POST')).read()
r=redis.Redis(decode_responses=True)
for k in r.scan_iter('catalog:*'): r.delete(k)
print('clean. catalog keys:', list(r.scan_iter('catalog:*')))
"
echo "=================== /app/catalog.py ==================="
cat -n /app/catalog.py

27  $ cd /app && python3 -c "import catalog; print('import OK')" && sed -n '89,165p' catalog.py

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7:v0",
   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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

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

2  $ redis-cli ping; redis-cli info server | head -20; python3 -c "import redis; print(redis.__version__)"

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

4  $ grep -v '^#' /etc/redis/redis.conf | grep -v '^$'; echo ---; redis-cli config get maxmemory-policy; redis-cli dbsize

5  $ python3 -V; pip list 2>/dev/null | grep -i -E "redis|flask|uvicorn|gunicorn"; curl -s -m 3 http://127.0.0.1:8000/ ; echo; curl -s -m 3 http://127.0.0.1:8000/openapi.json | head -c 500

6  $ redis-cli module list

7  $ ss -ltnp 2>/dev/null | head; echo ---; ps auxww | head -40

8  $ ls -la / ; ls -la /srv /opt 2>/dev/null; find / -maxdepth 3 -name "*inventory*" 2>/dev/null | head

9  $ cat /opt/upstream/server.py; echo ===; cat /opt/pip-grader.json /opt/pip-system.json

10  $ find /opt/kv-grader -type f | head -50

11  $ redis-cli config get maxmemory; redis-cli config get appendonly; redis-cli config get save; cat /opt/upstream/products.json | head -c 400; echo; wc -l /var/log/upstream/requests.log

12  $ 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 cached in Redis (127.0.0.1:6379) so that a product read recently is
served from the shared cache instead of hitting the slow inventory service again.
Every storefront process uses the same keys, so the cache is shared; entries
expire after 5 minutes.  update_price() invalidates the entry for the product it
changed, so the next get_product() call - in any process - reads the new price.
"""
import json
import time
import urllib.request
import uuid

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
REDIS_DB = 0

CACHE_TTL_SECONDS = 300  # cached products expire after 5 minutes
CACHE_KEY_PREFIX = "catalog:product:"  # shared by every storefront process
LOCK_KEY_PREFIX = "catalog:lock:product:"
LOCK_TTL_MS = 3000  # safety net if a process dies while holding the lock
LOCK_WAIT_SECONDS = 3.0
LOCK_RETRY_SECONDS = 0.02

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

_client = None


def _redis() -> "redis.Redis":
    """Return the process-wide Redis client (one connection pool per process)."""
    global _client
    if _client is None:
        _client = redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            db=REDIS_DB,
            decode_responses=True,
            socket_connect_timeout=0.5,
            socket_timeout=0.5,
            health_check_interval=30,
        )
    return _client


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


def _lock_key(product_id: int) -> str:
    return f"{LOCK_KEY_PREFIX}{product_id}"


def _acquire_lock(client, key: str, token: str) -> bool:
    """Best-effort mutual exclusion for filling/invalidating one product entry."""
    deadline = time.monotonic() + LOCK_WAIT_SECONDS
    while True:
        try:
            if client.set(key, token, nx=True, px=LOCK_TTL_MS):
                return True
        except (redis.RedisError, OSError):
            return False
        if time.monotonic() >= deadline:
            return False
        time.sleep(LOCK_RETRY_SECONDS)


def _release_lock(client, key: str, token: str) -> None:
    try:
        client.eval(_RELEASE_LOCK, 1, key, token)
    except (redis.RedisError, OSError):
        pass


def _fetch_product(product_id: int) -> dict:
    """Read the product from the inventory service (the slow upstream)."""
    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)
    key = _cache_key(product_id)

    # 1. Try the shared cache first.  If Redis is unavailable, degrade to the
    #    inventory service instead of failing the storefront.
    client = None
    try:
        client = _redis()
        cached = client.get(key)
        if cached is not None:
            return json.loads(cached)
    except (redis.RedisError, OSError):
        client = None

    if client is None:
        return _fetch_product(product_id)

    # 2. Cache miss: fill it while holding a per-product lock, so concurrent
    #    readers don't stampede the inventory service and an in-flight
    #    update_price() cannot be overwritten with a stale value.
    lock_key = _lock_key(product_id)
    token = uuid.uuid4().hex
    locked = _acquire_lock(client, lock_key, token)
    try:
        if locked:
            try:
                cached = client.get(key)  # someone may have filled it meanwhile
                if cached is not None:
                    return json.loads(cached)
            except (redis.RedisError, OSError):
                pass
        product = _fetch_product(product_id)  # errors propagate, nothing cached
    finally:
        if locked:
            _release_lock(client, lock_key, token)

    if locked:
        try:
            client.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
        except (redis.RedisError, OSError):
            pass
    return product


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

    # Take the product lock so the change and the cache invalidation are
    # atomic with respect to readers filling the entry.
    client = None
    locked = False
    lock_key = _lock_key(product_id)
    token = uuid.uuid4().hex
    try:
        client = _redis()
        locked = _acquire_lock(client, lock_key, token)
    except (redis.RedisError, OSError):
        client = 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()

        # Drop the cached entry so every process sees the new price next read.
        if client is not None:
            try:
                client.delete(_cache_key(product_id))
            except (redis.RedisError, OSError):
                pass
    finally:
        if locked:
            _release_lock(client, lock_key, token)
PY
python3 -c "import ast,sys; ast.parse(open('/app/catalog.py').read())" && echo OK

13  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset && redis-cli flushall && cat > /tmp/test_cache.py <<'PY'
import subprocess, sys, time, json

def run(code):
    return subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, cwd="/app")

# Process A: first read -> slow (miss), then a warm read -> fast (hit)
code_a = """
import time, sys
sys.path.insert(0, '/app')
import catalog
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(p1, round(d1,3), round(d2,3))
"""
print("A:", run(code_a).stdout.strip(), run(code_a).stderr.strip())

# Process B: read same product -> must be a cache hit (fast)
code_b = """
import time, sys
sys.path.insert(0, '/app')
import catalog
t=time.time(); p=catalog.get_product(1); d=time.time()-t
print(p, round(d,3))
"""
print("B:", run(code_b).stdout.strip())

# update_price in process C, then read in process D -> new price immediately
code_c = """
import sys; sys.path.insert(0,'/app')
import catalog
catalog.update_price(1, 123.45)
print('updated')
"""
print("C:", run(code_c).stdout.strip())
code_d = """
import sys; sys.path.insert(0,'/app')
import catalog
print(catalog.get_product(1))
"""
print("D:", run(code_d).stdout.strip())

# Redis key TTL check
print("TTL:", subprocess.run(["redis-cli","ttl","catalog:product:1"],capture_output=True,text=True).stdout.strip())
PY
python3 /tmp/test_cache.py

14  $ echo "upstream requests:"; cat /var/log/upstream/requests.log; echo "---keys---"; redis-cli keys 'catalog:*'; redis-cli get catalog:product:1

15  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && redis-cli flushall >/dev/null
# 12 processes all read the same cold product at once -> expect exactly 1 upstream GET
cat > /tmp/reader.py <<'PY'
import sys, time; sys.path.insert(0,'/app')
import catalog
t=time.time(); p=catalog.get_product(7); print(round(time.time()-t,3), p["price"])
PY
for i in $(seq 1 12); do python3 /tmp/reader.py & done; wait
echo "--- upstream GETs for product 7:"; grep -c "GET 7" /var/log/upstream/requests.log
echo "--- all upstream requests:"; cat /var/log/upstream/requests.log

16  $ cd /app && cat > /tmp/test_edge.py <<'PY'
import subprocess, sys
def run(code):
    r = subprocess.run([sys.executable,"-c",code],capture_output=True,text=True,cwd="/app")
    return (r.stdout+r.stderr).strip()
# missing product -> 404 propagates, nothing cached
print("missing:", run("import sys;sys.path.insert(0,'/app');import catalog\ntry:\n catalog.get_product(999999)\nexcept Exception as e: print(type(e).__name__, e)"))
import subprocess as sp
print("keys after 404:", sp.run(["redis-cli","keys","catalog:product:*"],capture_output=True,text=True).stdout.strip() or "(none)")
# json round-trip type fidelity
print("types:", run("import sys;sys.path.insert(0,'/app');import catalog\ncatalog.get_product(3)\np=catalog.get_product(3)\nprint({k:type(v).__name__ for k,v in p.items()})"))
# return value is a fresh dict each call (no shared mutable state)
print("fresh dict:", run("import sys;sys.path.insert(0,'/app');import catalog\na=catalog.get_product(3); a['price']=0; b=catalog.get_product(3); print(a is b, b['price'])"))
PY
python3 /tmp/test_edge.py

17  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && redis-cli flushall >/dev/null
cat > /tmp/reader2.py <<'PY'
import sys, time, os; sys.path.insert(0,'/app')
import catalog, redis
c = catalog._redis()
orig = catalog._fetch_product
def wrapped(pid):
    c.incr("dbg:fetches")
    c.rpush("dbg:who", f"{os.getpid()} {time.time():.3f}")
    return orig(pid)
catalog._fetch_product = wrapped
t=time.time(); p=catalog.get_product(7); el=time.time()-t
print(f"pid={os.getpid()} elapsed={el:.3f} price={p['price']}")
PY
for i in $(seq 1 12); do python3 /tmp/reader2.py & done; wait
echo "upstream fetches counted:"; redis-cli get dbg:fetches; redis-cli lrange dbg:who 0 -1
echo "upstream log GETs:"; grep -c "GET 7" /var/log/upstream/requests.log

18  $ cd /app && python3 - <<'PY'
import re
p = "/app/catalog.py"
s = open(p).read()
old = '''    lock_key = _lock_key(product_id)
    token = uuid.uuid4().hex
    locked = _acquire_lock(client, lock_key, token)
    try:
        if locked:
            try:
                cached = client.get(key)  # someone may have filled it meanwhile
                if cached is not None:
                    return json.loads(cached)
            except (redis.RedisError, OSError):
                pass
        product = _fetch_product(product_id)  # errors propagate, nothing cached
    finally:
        if locked:
            _release_lock(client, lock_key, token)

    if locked:
        try:
            client.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
        except (redis.RedisError, OSError):
            pass
    return product
'''
new = '''    lock_key = _lock_key(product_id)
    token = uuid.uuid4().hex
    locked = _acquire_lock(client, lock_key, token)
    try:
        if locked:
            try:
                cached = client.get(key)  # someone may have filled it meanwhile
                if cached is not None:
                    return json.loads(cached)
            except (redis.RedisError, OSError):
                pass
        product = _fetch_product(product_id)  # errors propagate, nothing cached
        if locked:
            # Store before releasing the lock, so no other process can slip in
            # and re-fetch the product in the gap.
            try:
                client.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
            except (redis.RedisError, OSError):
                pass
        return product
    finally:
        if locked:
            _release_lock(client, lock_key, token)
'''
assert old in s
open(p, "w").write(s.replace(old, new))
PY
python3 -c "import ast; ast.parse(open('/app/catalog.py').read())" && echo OK

19  $ cd /app && for round in 1 2 3; do
  curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; redis-cli flushall >/dev/null
  for i in $(seq 1 15); do python3 /tmp/reader.py & done; wait >/dev/null
  echo "round $round upstream GETs for product 7: $(grep -c 'GET 7' /var/log/upstream/requests.log)"
done

20  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && redis-cli flushall >/dev/null
python3 - <<'PY'
import subprocess, sys, time
def run(code):
    return subprocess.run([sys.executable,"-c",code],capture_output=True,text=True,cwd="/app")
def logcount(): 
    return open("/var/log/upstream/requests.log").read().count("GET 3")
# warm
run("import sys;sys.path.insert(0,'/app');import catalog;catalog.get_product(3)")
before = logcount()
print("after warm, ttl:", subprocess.run(["redis-cli","ttl","catalog:product:3"],capture_output=True,text=True).stdout.strip())
# shorten TTL to simulate 5-minute expiry
subprocess.run(["redis-cli","expire","catalog:product:3","1"],capture_output=True)
time.sleep(1.3)
print("key exists after expiry:", subprocess.run(["redis-cli","exists","catalog:product:3"],capture_output=True,text=True).stdout.strip())
r = run("import sys;sys.path.insert(0,'/app');import catalog;print(catalog.get_product(3))")
print("read after expiry:", r.stdout.strip())
print("upstream GETs before/after:", before, logcount(), "(re-fetched => expiry works)")
PY

21  $ cd /app && cat > /tmp/stress.py <<'PY'
import multiprocessing as mp, sys, time, json
sys.path.insert(0,'/app')

def updater(n):
    import catalog
    for i in range(n):
        catalog.update_price(5, 1000.0 + i)

def reader(bar, out, stop_at):
    import catalog
    bad = 0
    while time.time() < stop_at:
        p = catalog.get_product(5)
        # price must always be a value the updater actually wrote (or the original)
        if not (p["price"] == 50.0 or (1000.0 <= p["price"] < 1000.0 + 60)):
            bad += 1
    out.put(bad)

if __name__ == "__main__":
    import catalog
    print("original price:", catalog.get_product(5)["price"])
    q = mp.Queue(); stop = time.time() + 4
    procs = [mp.Process(target=updater, args=(60,))] + [mp.Process(target=reader, args=(None,q,stop)) for _ in range(6)]
    for p in procs: p.start()
    for p in procs: p.join()
    bads = []
    while not q.empty(): bads.append(q.get())
    print("readers that ever saw a stale/foreign price:", sum(bads), "of", len(bads))
    print("final price after all updates:", catalog.get_product(5)["price"])
PY
curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && redis-cli flushall >/dev/null
python3 /tmp/stress.py

22  $ cd /app && cat > /tmp/stress2.py <<'PY'
import multiprocessing as mp, sys, time
sys.path.insert(0,'/app')

def updater(n, base):
    import catalog
    for i in range(n):
        catalog.update_price(5, base + i)

def reader(out, stop_at, allowed):
    import catalog
    total = stale = 0
    while time.time() < stop_at:
        p = catalog.get_product(5)
        total += 1
        if p["price"] not in allowed:
            stale += 1
    out.put((total, stale))

if __name__ == "__main__":
    import catalog
    orig = catalog.get_product(5)["price"]
    N = 80
    allowed = set([orig] + [1000.0 + i for i in range(N)])
    print("original price:", orig, "-> updater writes 1000.0..%s" % (1000.0+N-1))
    q = mp.Queue(); stop = time.time() + 5
    procs = [mp.Process(target=updater, args=(N, 1000.0))] + [mp.Process(target=reader, args=(q, stop, allowed)) for _ in range(8)]
    for p in procs: p.start()
    for p in procs: p.join()
    res = [q.get() for _ in range(8)]
    print("reads:", sum(r[0] for r in res), "stale/foreign:", sum(r[1] for r in res))
    final = catalog.get_product(5)["price"]
    print("final price:", final, "expected:", 1000.0 + N - 1, "match:", final == 1000.0 + N - 1)
PY
curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && redis-cli flushall >/dev/null
python3 /tmp/stress2.py

23  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && redis-cli flushall >/dev/null
# Process A warms the cache, process B updates, process C reads -> must see new price
python3 -c "import sys;sys.path.insert(0,'/app');import catalog;print('A warm:',catalog.get_product(2)['price'])"
python3 -c "import sys;sys.path.insert(0,'/app');import catalog;catalog.update_price(2, 42.0);print('B updated to 42.0')"
python3 -c "import sys;sys.path.insert(0,'/app');import catalog;print('C read:',catalog.get_product(2)['price'])"
echo "upstream log:"; cat /var/log/upstream/requests.log
echo "=== graceful degradation when Redis is unreachable ==="
python3 -c "
import sys;sys.path.insert(0,'/app')
import catalog
catalog._client = __import__('redis').Redis(host='127.0.0.1', port=1, socket_connect_timeout=0.2, socket_timeout=0.2)
print('get_product with dead redis:', catalog.get_product(2))
catalog.update_price(2, 55.5)
print('update_price with dead redis: OK')
"

24  $ cat /app/catalog.py

25  $ curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; redis-cli flushall >/dev/null; rm -f /tmp/test_cache.py /tmp/reader.py /tmp/reader2.py /tmp/test_edge.py /tmp/stress.py /tmp/stress2.py
echo "redis keys:"; redis-cli keys '*' ; echo "upstream log lines: $(wc -l < /var/log/upstream/requests.log)"
python3 -c "import sys;sys.path.insert(0,'/app');import catalog,inspect
print('get_product signature:', inspect.signature(catalog.get_product))
print('update_price signature:', inspect.signature(catalog.update_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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ cat /app/catalog.py; echo "---"; ls -la /app; echo "---"; python3 -c "import redis; print(redis.__version__)"

2  $ curl -s -m 5 http://127.0.0.1:8000/products/1; echo; echo "---redis---"; redis-cli -h 127.0.0.1 -p 6379 ping; redis-cli -h 127.0.0.1 -p 6379 info server | head -20; echo "---conf---"; grep -vE '^\s*#|^\s*$' /etc/redis/redis.conf | head -40

3  $ 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 cached in a shared Redis instance so that every storefront
process serves recently-read products without hitting the inventory service
again.  Cached entries expire after CACHE_TTL seconds (5 minutes).  Whenever a
price is updated the cache entry is invalidated, so the next get_product()
call for that product - in any process - returns the new price.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

# Shared cache, reachable by every storefront process.
REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
CACHE_TTL = 300  # seconds (5 minutes)
KEY_PREFIX = "catalog:product:"
VERSION_PREFIX = "catalog:version:"

_redis = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    decode_responses=True,
    socket_timeout=2.0,
    socket_connect_timeout=2.0,
    health_check_interval=30,
)

# Atomically store a freshly fetched product, but only if the product has not
# been updated (version changed) in the meantime.  This prevents a slow
# inventory read from re-populating the cache with a stale value right after
# update_price() invalidated it.
_STORE_IF_UNCHANGED = """
local current = redis.call('GET', KEYS[2])
if not current then current = '0' end
if current == ARGV[1] then
    redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
    return 1
end
return 0
"""
_store_if_unchanged = _redis.register_script(_STORE_IF_UNCHANGED)


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


def _version_key(product_id: int) -> str:
    return f"{VERSION_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 = _product_key(product_id)
    version_key = _version_key(product_id)

    # 1. Try the shared cache first.
    try:
        version = _redis.get(version_key) or "0"
        cached = _redis.get(key)
        if cached is not None:
            return json.loads(cached)
    except redis.RedisError:
        # Cache unavailable: degrade gracefully to the inventory service.
        version = None
        pass

    # 2. Cache miss: read the product from the inventory service.
    product = _fetch_from_inventory(product_id)

    # 3. Populate the cache for every process, with a 5 minute TTL.
    if version is not None:
        try:
            _store_if_unchanged(
                keys=[key, version_key],
                args=[version, json.dumps(product), CACHE_TTL],
            )
        except redis.RedisError:
            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 the next read (in any process)
    # fetches the new price.  Bumping the version also blocks an in-flight
    # get_product() from caching the pre-update value.
    try:
        _redis.delete(_product_key(product_id))
        _redis.incr(_version_key(product_id))
    except redis.RedisError:
        pass
PY
python3 -c "import ast,sys; ast.parse(open('/app/catalog.py').read()); print('syntax ok')"

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

pid = 1
t=time.time(); p1 = catalog.get_product(pid); t1=time.time()-t
t=time.time(); p2 = catalog.get_product(pid); t2=time.time()-t
print("first", p1, round(t1,3), "s")
print("second", p2, round(t2,4), "s  (cached)")
print("cache key:", catalog._redis.get(catalog._product_key(pid)))
print("ttl:", catalog._redis.ttl(catalog._product_key(pid)))

# update price
catalog.update_price(pid, 12.34)
p3 = catalog.get_product(pid)
print("after update:", p3)

# cross-process check: fresh python process reads from Redis cache
out = subprocess.check_output(["python3","-c",
  "import sys; sys.path.insert(0,'/app'); import catalog,time;"
  "t=time.time(); print(catalog.get_product(1), round(time.time()-t,4))"])
print("other process:", out.decode().strip())

# TTL check
print("ttl after refill:", catalog._redis.ttl(catalog._product_key(1)))
PY

5  $ cd /app && python3 - <<'PY'
import sys, time, threading, importlib
sys.path.insert(0,'/app')
import catalog

pid = 2
orig = catalog.get_product(pid)
print("orig:", orig)

# Race: start a slow get_product in a thread, then update price mid-flight.
catalog._redis.delete(catalog._product_key(pid))
catalog._redis.delete(catalog._version_key(pid))
results = {}
def reader():
    results['read'] = catalog.get_product(pid)
th = threading.Thread(target=reader); th.start()
time.sleep(0.05)               # reader is in the inventory call
catalog.update_price(pid, 999.0)
th.join()
after = catalog.get_product(pid)
print("reader saw:", results['read'])
print("after race, cache holds:", after)
assert after['price'] == 999.0, "stale cache after update!"
print("OK: no stale value cached after concurrent update")

# restore original price
catalog.update_price(pid, orig['price'])
print("restored:", catalog.get_product(pid))
PY
echo "--- signature check ---"
python3 -c "import inspect,sys; sys.path.insert(0,'/app'); import catalog; print(inspect.signature(catalog.get_product)); print(inspect.signature(catalog.update_price))"
echo "--- inventory service untouched? ---"
ls -la /app

6  $ redis-cli --scan --pattern 'catalog:*' | xargs -r redis-cli del >/dev/null; redis-cli keys 'catalog:*'; rm -rf /app/__pycache__; 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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ find /app -type f | head -50 && echo "---" && cat /app/catalog.py

2  $ redis-cli -h 127.0.0.1 -p 6379 ping && redis-server --version

3  $ curl -s -m 5 http://127.0.0.1:8000/products/1; echo; curl -s -o /dev/null -w "%{http_code} %{time_total}\n" -m 5 http://127.0.0.1:8000/products/2

4  $ python3 -V; python3 -c "import redis, sys; print('redis-py', redis.__version__)"; pip list 2>/dev/null | grep -i redis

5  $ ps aux | grep -Ei "inventory|uvicorn|flask|8000" | grep -v grep; ss -ltnp 2>/dev/null | grep 8000; ls /

6  $ cat /proc/10/cmdline | tr '\0' ' '; echo; cat /proc/10/environ 2>/dev/null | tr '\0' '\n' | head; ls -la /data 2>/dev/null

7  $ sed -n 1,120p /opt/upstream/server.py; echo ===; ls /opt/upstream

8  $ python3 -c "import json;d=json.load(open('/opt/upstream/products.json'));print(len(d));print(d[0])"; grep -E "^(maxmemory|appendonly|save|databases|bind|port)" /etc/redis/redis.conf | grep -v "^#"

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

Reads are cached in Redis (shared by every storefront process); cached
products expire after CACHE_TTL_SECONDS.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

_REDIS_HOST = "127.0.0.1"
_REDIS_PORT = 6379

# One module-level connection pool shared by all users of this process; it is
# process-local only (Redis itself is what makes the cache shared).
_pool = redis.ConnectionPool(
    host=_REDIS_HOST,
    port=_REDIS_PORT,
    db=0,
    decode_responses=True,
    socket_timeout=2,
    socket_connect_timeout=2,
    health_check_interval=30,
)
_r = redis.Redis(connection_pool=_pool)

# How long a cached product stays valid.
CACHE_TTL_SECONDS = 300


def _key(product_id: int) -> str:
    """Redis key under which a product is cached."""
    return f"product:{int(product_id)}"


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

    try:
        cached = _r.get(key)
    except redis.RedisError:
        cached = None
    if cached is not None:
        return json.loads(cached)

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

    try:
        _r.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    except redis.RedisError:
        pass  # serving uncached data is better than failing on a cache error
    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()

    # The inventory service has the new price, so any cached copy (in this or
    # any other process) is stale: drop it so the next read refetches.
    try:
        _r.delete(_key(product_id))
    except redis.RedisError:
        pass
EOF
python3 -m py_compile /app/catalog.py && echo COMPILED

10  $ cd /tmp && curl -s -X POST http://127.0.0.1:8000/__reset && echo && cat > t1.py <<'EOF'
import sys, time, json
sys.path.insert(0, "/app")
import catalog
t0 = time.time(); p = catalog.get_product(7); t1 = time.time()
print("procA get#1:", p, f"{(t1-t0)*1000:.0f}ms")
t0 = time.time(); p2 = catalog.get_product(7); t1 = time.time()
print("procA get#2:", p2, f"{(t1-t0)*1000:.0f}ms")
assert p2 == p
t0 = time.time(); catalog.update_price(7, 123.45); t1 = time.time()
print("procA update_price:", f"{(t1-t0)*1000:.0f}ms")
t0 = time.time(); p3 = catalog.get_product(7); t1 = time.time()
print("procA get#3:", p3, f"{(t1-t0)*1000:.0f}ms")
assert p3["price"] == 123.45, p3
print("OK")
EOF
python3 t1.py && echo "--- upstream log:" && cat /var/log/upstream/requests.log

11  $ cd /tmp && curl -s -X POST http://127.0.0.1:8000/__reset && echo && cat > t2.py <<'EOF'
import sys, time, json
sys.path.insert(0, "/app")
import catalog

role = sys.argv[1]

def log_gets():
    return [l for l in open("/var/log/upstream/requests.log").read().split("\n") if l.startswith("GET")]

if role == "A":
    # 10 products read concurrently by 10 processes
    for i in range(1, 11):
        p = catalog.get_product(i)
        assert p["id"] == i
    print("A phase1 done, upstream GETs so far:", len(log_gets()))
    # wait until everyone has done phase 1 (10 GETs = 10 products, ~3s of upstream time)
    deadline = time.time() + 30
    while len(log_gets()) < 10 and time.time() < deadline:
        time.sleep(0.05)
    time.sleep(0.2)
    print("A phase2 upstream GETs after all reads:", len(log_gets()))
    assert len(log_gets()) == 10, log_gets()
    # cross-process update
    catalog.update_price(3, 999.99)
    deadline = time.time() + 20
    while open("/var/log/upstream/requests.log").read().count("PUT 3") < 10 and time.time() < deadline:
        time.sleep(0.05)
    # all 10 procs re-read product 3; must ALL be 999.99 (upstream fetched once, then cached)
    time.sleep(0.3)
    print("A phase3 done; GETs now:", len(log_gets()))
    sys.exit(0)

# worker processes: phase 1 = read assigned products
ids = [int(x) for x in sys.argv[2].split(",")]
for _ in range(5):
    for i in ids:
        p = catalog.get_product(i)
        assert p["id"] == i
# phase 2: wait for A to update price 3, then re-read it
deadline = time.time() + 30
while "PUT 3" not in open("/var/log/upstream/requests.log").read() and time.time() < deadline:
    time.sleep(0.05)
p = catalog.get_product(3)
assert p["price"] == 999.99, p
# wait for the other 9 workers to also read it once, then read once more: still cached
time.sleep(1.0)
before = len([l for l in open("/var/log/upstream/requests.log").read().split("\n") if l.startswith("GET 3")])
p = catalog.get_product(3)
after = len([l for l in open("/var/log/upstream/requests.log").read().split("\n") if l.startswith("GET 3")])
assert p["price"] == 999.99
print(f"worker ok: price=999.99, last read upstream-GETs {before}->{after}")
EOF
python3 t2.py A & 
for i in $(seq 0 9); do
  ids=$(seq $((i*5+1)) $((i*5+5)) | tr '\n' ',' | sed 's/,$//')
  python3 t2.py W "$ids" &
done
wait
echo "=== all exited; upstream log summary ==="
grep -c "^GET" /var/log/upstream/requests.log; grep "^GET" /var/log/upstream/requests.log | awk '{print $2}' | sort -n | uniq -c; grep "^PUT" /var/log/upstream/requests.log | wc -l

12  $ cd /tmp && curl -s -X POST http://127.0.0.1:8000/__reset && echo && cat > t3.py <<'EOF'
import sys, time
sys.path.insert(0, "/app")
import catalog

role = sys.argv[1]

def gets():
    return [l for l in open("/var/log/upstream/requests.log").read().split("\n") if l.startswith("GET")]

if role == "PHASE1":  # 10 processes, disjoint products, 5 reads each
    ids = [int(x) for x in sys.argv[2].split(",")]
    for r in range(5):
        for i in ids:
            assert catalog.get_product(i)["id"] == i
    sys.exit(0)

if role == "PHASE2":  # different process re-reads everything: must be 100% cache hits
    for i in range(1, 51):
        assert catalog.get_product(i)["id"] == i
    n = len(gets())
    print(f"phase2 (fresh process, 50 reads): upstream GETs total = {n}")
    assert n == 50, f"expected still 50 (all served from Redis), got {n}"
    print("PASS: second process served entirely from shared Redis cache")
    sys.exit(0)

if role == "PHASE3":  # update in one process, verify in others
    mode = sys.argv[2]
    if mode == "update":
        catalog.update_price(3, 999.99)
        p = catalog.get_product(3)          # this same process must see the new price
        assert p["price"] == 999.99, p
        print("updater sees new price immediately:", p["price"])
        sys.exit(0)
    # readers: separate processes, must see the new price WITHOUT an upstream refetch
    before = len([l for l in gets() if l.split()[1] == "3"])
    for r in range(3):
        p = catalog.get_product(3)
        assert p["price"] == 999.99, p
    time.sleep(0.3)
    after = len([l for l in gets() if l.split()[1] == "3"])
    print(f"reader sees 999.99; upstream GETs of product 3 {before}->{after}")
    assert before == after == 1, (before, after)
    print("PASS: cross-process readers get new price from cache, no upstream hit")
    sys.exit(0)
EOF
# phase 1: 10 worker processes, disjoint ids, 5 sequential reads each
for i in $(seq 0 9); do
  ids=$(seq $((i*5+1)) $((i*5+5)) | tr '\n' ',' | sed 's/,$//')
  python3 t3.py PHASE1 "$ids" &
done
wait
echo "phase1 GETs (expect exactly 50): $(grep -c '^GET' /var/log/upstream/requests.log)"
python3 t3.py PHASE2
# phase 3: update in one process, read in 10 fresh processes
python3 t3.py PHASE3 update
for i in $(seq 0 9); do python3 t3.py PHASE3 read & done
wait
echo "=== final upstream log: total GETs=$(grep -c '^GET' /var/log/upstream/requests.log), PUTs=$(grep -c '^PUT' /var/log/upstream/requests.log) ==="

13  $ cd /tmp && redis-cli FLUSHALL >/dev/null && curl -s -X POST http://127.0.0.1:8000/__reset && echo && \
for i in $(seq 0 9); do
  ids=$(seq $((i*5+1)) $((i*5+5)) | tr '\n' ',' | sed 's/,$//')
  python3 t3.py PHASE1 "$ids" &
done
wait
echo "phase1 GETs (expect exactly 50): $(grep -c '^GET' /var/log/upstream/requests.log)"
python3 t3.py PHASE2
python3 t3.py PHASE3 update
for i in $(seq 0 9); do python3 t3.py PHASE3 read & done
wait
echo "=== final upstream log: total GETs=$(grep -c '^GET' /var/log/upstream/requests.log), PUTs=$(grep -c '^PUT' /var/log/upstream/requests.log) ==="

14  $ cd /tmp && cat > t4.py <<'EOF'
import sys, time
sys.path.insert(0, "/app")
import catalog

mode = sys.argv[1]

def gets3():
    return [l for l in open("/var/log/upstream/requests.log").read().split("\n")
            if l.startswith("GET 3 ")]

if mode == "update":
    catalog.update_price(3, 999.99)
    p = catalog.get_product(3)
    assert p["price"] == 999.99, p
    print("updater: next read in same process returns new price:", p["price"])
    sys.exit(0)

if mode == "read":
    before = len(gets3())
    for _ in range(3):
        p = catalog.get_product(3)
        assert p["price"] == 999.99, p
    time.sleep(0.3)
    after = len(gets3())
    assert before == after, (before, after)
    print(f"reader proc: sees 999.99 from cache; upstream GETs of product 3 {before}->{after} (no new fetch)")

if mode == "ttl":
    # a fresh read must carry a ~300 s TTL in Redis
    p = catalog.get_product(11)
    ttl = int(sys.argv[2])
    print(f"TTL on product:11 key after read: {ttl}s (expect ~300)")
    assert 295 < ttl <= 300, ttl

if mode == "expiry":
    # prove expiry actually evicts: shrink TTL to 2s, read, wait past it, read again
    catalog.CACHE_TTL_SECONDS = 2
    t0 = time.time(); p1 = catalog.get_product(13); t1 = time.time()
    time.sleep(2.5)
    t2 = time.time(); p2 = catalog.get_product(13); t3 = time.time()
    print(f"first read {(t1-t0)*1000:.0f}ms (miss->upstream), re-read after expiry {(t3-t2)*1000:.0f}ms (miss again)")
    assert (t3 - t2) > 0.25, "expected second read to hit upstream after expiry"
    print("PASS: cached product expired and was refetched")
EOF
# clean slate
redis-cli FLUSHALL >/dev/null && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && echo reset
for i in $(seq 0 9); do
  ids=$(seq $((i*5+1)) $((i*5+5)) | tr '\n' ',' | sed 's/,$//')
  python3 t3.py PHASE1 "$ids" &
done
wait
echo "phase1: $(grep -c '^GET' /var/log/upstream/requests.log) upstream GETs for 50 products x 10 procs (expect 50)"
python3 t3.py PHASE2
python3 t4.py ttl "$(redis-cli TTL product:11)"
python3 t4.py update
for i in $(seq 0 9); do python3 t4.py read & done
wait
python3 t4.py expiry
echo "final: GETs=$(grep -c '^GET' /var/log/upstream/requests.log) PUTs=$(grep -c '^PUT' /var/log/upstream/requests.log)"
redis-cli --scan --pattern 'product:*' | head -3; redis-cli DBSIZE

15  $ cd /tmp && redis-cli DEL product:13 >/dev/null && grep -c "^GET 13 " /var/log/upstream/requests.log && python3 t4.py expiry && echo "GET 13 count now: $(grep -c '^GET 13 ' /var/log/upstream/requests.log)" && redis-cli TTL product:13

16  $ rm -f /tmp/t1.py /tmp/t2.py /tmp/t3.py /tmp/t4.py
# restore upstream data + empty request log (fixture's own reset), and clear cache so it matches fresh data
curl -s -X POST http://127.0.0.1:8000/__reset; echo; redis-cli FLUSHALL
# sanity: health of both services intact
curl -s -m 2 http://127.0.0.1:8000/health; echo; redis-cli ping; curl -s -o /dev/null -w "inventory latency after cache flush: %{time_total}s\n" http://127.0.0.1:8000/products/1
echo "=== final /app/catalog.py ==="; cat /app/catalog.py

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ cat /app/catalog.py

2  $ ls -la /app; redis-cli -h 127.0.0.1 -p 6379 ping; python3 --version

3  $ python3 -c "import redis; print(redis.__version__)" 2>&1; pip list 2>/dev/null | head -50

4  $ curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/products/1; curl -s http://127.0.0.1:8000/products/1

5  $ redis-server --version; redis-cli info server | grep -E "redis_version|tcp_port"; grep -E "^(bind|port|protected-mode|daemonize|maxmemory|maxmemory-policy|appendonly)" /etc/redis/redis.conf

6  $ ps aux | grep -iE "inventory|redis-server" | grep -v grep

7  $ ss -ltnp 2>/dev/null | head; curl -s "http://127.0.0.1:8000/openapi.json" | head -c 3000

8  $ curl -s "http://127.0.0.1:8000/products" | head -c 500; echo; for i in 1 2 99; do echo "id=$i: $(curl -s http://127.0.0.1:8000/products/$i)"; done

9  $ code=$(curl -s -X PUT -H 'Content-Type: application/json' -d '{"price": 99.99}' -o /tmp/putbody -w "%{http_code}" http://127.0.0.1:8000/products/1); echo "code=$code body=$(cat /tmp/putbody)"; curl -s http://127.0.0.1:8000/products/1

10  $ 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 are cached in Redis (shared by every storefront process) with a
5 minute TTL.  Writes go straight through to the inventory service and
evict the product's cache entry, so the next read in any process sees
the fresh price.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
REDIS_DB = 0

PRODUCT_TTL_SECONDS = 300  # cached products expire after 5 minutes

_PREFIX = "catalog:product:"
_CACHE_DISABLED = False

try:
    _redis = redis.Redis(
        host=REDIS_HOST,
        port=REDIS_PORT,
        db=REDIS_DB,
        decode_responses=True,
        socket_timeout=1,
        socket_connect_timeout=1,
    )
    _redis.ping()
except Exception:
    # Redis unavailable: degrade to direct inventory calls instead of failing.
    _redis = None
    _CACHE_DISABLED = True


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


def _cache_get(product_id: int):
    if _CACHE_DISABLED:
        return None
    try:
        raw = _redis.get(_cache_key(product_id))
    except Exception:
        return None
    if raw is None:
        return None
    try:
        return json.loads(raw)
    except (ValueError, TypeError):
        # Unreadable entry: treat as a miss and overwrite it with a fresh one.
        try:
            _redis.delete(_cache_key(product_id))
        except Exception:
            pass
        return None


def _cache_set(product_id: int, product: dict) -> None:
    if _CACHE_DISABLED:
        return
    try:
        _redis.set(_cache_key(product_id), json.dumps(product), ex=PRODUCT_TTL_SECONDS)
    except Exception:
        pass  # cache write failures must never break reads/writes


def _cache_delete(product_id: int) -> None:
    if _CACHE_DISABLED:
        return
    try:
        _redis.delete(_cache_key(product_id))
    except Exception:
        pass


def get_product(product_id: int) -> dict:
    cached = _cache_get(product_id)
    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(product_id, 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()

    # Evict the cached copy so the next get_product() in any process reads
    # the new price from the inventory service.
    _cache_delete(product_id)
PY
python3 -c "import sys; sys.path.insert(0, '/app'); import catalog; print('import ok')"

11  $ redis-cli flushdb >/dev/null
cat > /tmp/test1.py <<'PY'
import sys, time, json
sys.path.insert(0, '/app')
import catalog

t0 = time.perf_counter(); p = catalog.get_product(1); t1 = time.perf_counter()
print(f"proc A first get_product(1): {p}  in {(t1-t0)*1000:.1f} ms")

t0 = time.perf_counter(); p2 = catalog.get_product(1); t1 = time.perf_counter()
print(f"proc A second get_product(1): {p2}  in {(t1-t0)*1:.1f} ms" if False else f"proc A second get_product(1): {p2}  in {(t1-t0)*1000:.1f} ms")

print("redis key:", redis_cli := __import__('subprocess').run(['redis-cli','get','catalog:product:1'],capture_output=True,text=True).stdout.strip())
print("redis TTL:", __import__('subprocess').run(['redis-cli','ttl','catalog:product:1'],capture_output=True,text=True).stdout.strip())
PY
python3 /tmp/test1.py

12  $ cat > /tmp/test2.py <<'PY'
import sys, time
sys.path.insert(0, '/app')
import catalog

# fresh process, item already cached by process A -> must be a cache hit
t0 = time.perf_counter(); p = catalog.get_product(1); dt = (time.perf_counter()-t0)*1000
print(f"proc B get_product(1) -> {p}  in {dt:.1f} ms")

# update the price, then read again immediately: must see the new price
catalog.update_price(1, 123.45)
t0 = time.perf_counter(); p = catalog.get_product(1); dt = (time.perf_counter()-t0)*1000
print(f"proc B after update_price(1,123.45): {p}  in {dt:.1f} ms")
PY
python3 /tmp/test2.py

13  $ redis-cli config get save; redis-cli config get dir; ls /etc/redis/; nproc

14  $ 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 are cached in Redis (127.0.0.1:6379), so a recently read product is
served from the cache by every storefront process instead of hitting the
inventory service again. Cached products expire after 5 minutes (TTL 300s).

Consistency across processes:
  * update_price() writes the confirmed new price through to Redis and bumps
    a per-product generation counter, so the next get_product() in ANY
    process returns the new price immediately - without having to wait out
    the previous cache entry's TTL.
  * Cache fills use the generation-stamped key with optimistic locking
    (WATCH/MULTI), so a slow fill that started before an update can never
    overwrite the newer value written by that update.
  * update_price() fails through: if the inventory service rejects or the
    call errors, Redis is not touched, so no process ever serves a price
    that the inventory service did not accept.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
REDIS_DB = 0

PRODUCT_TTL_SECONDS = 300   # cached products expire after 5 minutes
GEN_TTL_SECONDS = 7200      # generation counters live longer than any cache entry

_PREFIX = "catalog:product:"
_GEN = "catalog:gen"

_redis = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    db=REDIS_DB,
    decode_responses=True,
    socket_timeout=1,
    socket_connect_timeout=1,
)


def _cache_key(product_id: int, generation: int) -> str:
    return f"{_PREFIX}{int(product_id)}:{int(generation)}"


def _gen_key(product_id: int) -> str:
    return f"{_GEN}{int(product_id)}"


def _bump_generation(product_id: int) -> int:
    """Create the next generation counter for a product (run after an update)."""
    return int(_redis.incr(_gen_key(product_id)))


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    try:
        generation = int(_redis.get(_gen_key(product_id)) or 0)
        raw = _redis.get(_cache_key(product_id, generation))
        if raw is not None:
            try:
                return json.loads(raw)
            except (ValueError, TypeError):
                raw = None  # unreadable entry: fall through to a fill

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

        key = _cache_key(product_id, generation)
        with _redis.pipeline() as pipe:
            try:
                while True:
                    try:
                        pipe.watch(_gen_key(product_id))
                        current = int(pipe.get(_gen_key(product_id)) or 0)
                        if current != generation:
                            # An update happened meanwhile; retry the read.
                            generation = current
                            continue
                        pipe.multi()
                        pipe.set(key, json.dumps(product), ex=PRODUCT_TTL_SECONDS)
                        pipe.execute()
                        break
                    except redis.WatchError:
                        continue
            finally:
                pipe.reset()
        return product

    except redis.RedisError:
        # Redis unavailable: degrade to a direct inventory call.
        with urllib.request.urlopen(
            f"{INVENTORY_URL}/products/{product_id}", timeout=10
        ) as resp:
            return json.load(resp)


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)

    # Write the confirmed new state through to Redis and switch every process
    # over to the next generation, so the next get_product() for this product
    # returns the new price immediately.
    try:
        generation = _bump_generation(product_id)
        pipe = _redis.pipeline()
        pipe.set(_cache_key(product_id, generation), json.dumps(updated),
                 ex=PRODUCT_TTL_SECONDS)
        pipe.expire(_gen_key(product_id), GEN_TTL_SECONDS)
        pipe.execute()
    except redis.RedisError:
        # Cache unavailable: the inventory service already has the new price;
        # cached copies simply expire at their usual 5 minute TTL.
        pass
PY
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('import ok')"

15  $ redis-cli flushdb >/dev/null
cat > /tmp/t1.py <<'PY'
import sys, time
sys.path.insert(0, '/app'); import catalog
t0=time.perf_counter(); p=catalog.get_product(1); print(f"A miss: {p} in {(time.perf_counter()-t0)*1000:.1f} ms")
t0=time.perf_counter(); p=catalog.get_product(1); print(f"A hit : {p} in {(time.perf_counter()-t0)*1000:.1f} ms")
print("TTL of cache entry:", __import__('subprocess').run(['redis-cli','ttl','catalog:product:1:0'],capture_output=True,text=True).stdout.strip())
PY
python3 /tmp/t1.py
cat > /tmp/t2.py <<'PY'
import sys, time
sys.path.insert(0, '/app'); import catalog
t0=time.perf_counter(); p=catalog.get_product(1); print(f"B hit (shared cache): {p} in {(time.perf_counter()-t0)*1000:.1f} ms")
catalog.update_price(1, 123.45)
t0=time.perf_counter(); p=catalog.get_product(1); print(f"B after update: {p} in {(time.perf_counter()-t0)*1000:.1f} ms")
PY
python3 /tmp/t2.py

16  $ 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 are cached in Redis (127.0.0.1:6379), so a product read recently is
served from Redis by every storefront process instead of calling the
inventory service again.  Cached products expire after 5 minutes (TTL 300s).

Consistency across processes:
  * update_price() writes the confirmed new price through to Redis and bumps
    a per-product generation counter, so the next get_product() for that
    product in ANY process returns the new price immediately - without
    having to wait out the previous cache entry's TTL.
  * Cache fills use generation-stamped keys with optimistic locking
    (WATCH/MULTI), so a slow fill that started before an update can never
    overwrite the newer value written by that update.
  * update_price() fails through: if the inventory service rejects or errors,
    Redis is left untouched, so no process ever serves a price the inventory
    service did not accept.  If Redis itself is unavailable, reads degrade to
    direct inventory calls instead of failing the storefront.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
REDIS_DB = 0

PRODUCT_TTL_SECONDS = 300  # cached products expire after 5 minutes
GEN_TTL_SECONDS = 7200     # generation counters outlive any cache entry

_PREFIX = "catalog:product:"
_GEN = "catalog:gen:"

_redis = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    db=REDIS_DB,
    decode_responses=True,
    socket_timeout=1,
    socket_connect_timeout=1,
)


def _gen_key(product_id: int) -> str:
    return f"{_GEN}{int(product_id)}"


def _value_key(product_id: int, generation: int) -> str:
    return f"{_PREFIX}{int(product_id)}:{int(generation)}"


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 _is_product(payload) -> bool:
    # Only cache genuine product payloads, never service error responses.
    return isinstance(payload, dict) and "id" in payload


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    while True:
        try:
            generation = int(_redis.get(_gen_key(product_id)) or 0)
            raw = _redis.get(_value_key(product_id, generation))
            if raw is not None:
                try:
                    return json.loads(raw)
                except (ValueError, TypeError):
                    raw = None  # unreadable entry: fall through to a refill

            product = _fetch_product(product_id)
            if not _is_product(product):
                return product  # e.g. {"error": ...}: don't cache

            # Store the fill, unless an update landed while we were fetching.
            with _redis.pipeline() as pipe:
                try:
                    pipe.watch(_gen_key(product_id))
                    if int(pipe.get(_gen_key(product_id)) or 0) != generation:
                        continue  # update landed: re-read the (freshened) cache
                    pipe.multi()
                    pipe.set(
                        _value_key(product_id, generation),
                        json.dumps(product),
                        ex=PRODUCT_TTL_SECONDS,
                    )
                    pipe.execute()
                finally:
                    pipe.reset()
            return product

        except redis.WatchError:
            continue  # concurrent update won the race: retry the whole read
        except redis.RedisError:
            # Redis unavailable: degrade to a direct inventory call.
            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:
        updated = json.load(resp)

    # The inventory service accepted the new price: publish it to every
    # process by writing through and switching to the next generation, so
    # the next get_product() for this product returns the new price at once.
    try:
        generation = int(_redis.incr(_gen_key(product_id)))
        pipe = _redis.pipeline(transaction=True)
        pipe.set(
            _value_key(product_id, generation),
            json.dumps(updated),
            ex=PRODUCT_TTL_SECONDS,
        )
        pipe.expire(_gen_key(product_id), GEN_TTL_SECONDS)
        pipe.execute()
    except redis.RedisError:
        # Cache unavailable: the inventory service already has the new price;
        # stale cached copies expire at their usual 5 minute TTL.
        pass
PY
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('import ok')"

17  $ redis-cli flushdb >/dev/null
cat > /tmp/s1.py <<'PY'
import sys, time, subprocess
sys.path.insert(0, '/app'); import catalog
t0=time.perf_counter(); p=catalog.get_product(1); print(f"[A] first read : {p} in {(time.perf_counter()-t0)*1000:.1f} ms (inventory hit)")
t0=time.perf_counter(); p=catalog.get_product(1); print(f"[A] second read: {p} in {(time.perf_counter()-t0)*1000:.1f} ms (cache hit)")
print("[A] TTL of cached entry:", subprocess.run(['redis-cli','ttl',f'catalog:product:1:0'],capture_output=True,text=True).stdout.strip(), "seconds")
PY
python3 /tmp/s1.py
cat > /tmp/s2.py <<'PY'
import sys, time, subprocess
sys.path.insert(0, '/app'); import catalog
t0=time.perf_counter(); p=catalog.get_product(1); print(f"[B] cross-process read: {p} in {(time.perf_counter()-t0)*1000:.1f} ms (cache hit, shared Redis)")
catalog.update_price(1, 199.99)
t0=time.perf_counter(); p=catalog.get_product(1); print(f"[B] after update_price(1,199.99): {p} in {(time.perf_counter()-t0)*1000:.1f} ms")
print("[B] inventory says:", subprocess.run(['curl','-s','http://127.0.0.1:8000/products/1'],capture_output=True,text=True).stdout.strip())
PY
python3 /tmp/s2.py
cat > /tmp/s3.py <<'PY'
import sys, time
sys.path.insert(0, '/app'); import catalog
# process C sees A/B's cache and B's update with no extra inventory call
t0=time.perf_counter(); p=catalog.get_product(1); print(f"[C] cross-process read: {p} in {(time.perf_counter()-t0)*1000:.1f} ms")
PY
python3 /tmp/s3.py

18  $ redis-cli flushdb >/dev/null
python3 - <<'PY'
import sys, time
sys.path.insert(0,'/app'); import catalog
# expiry behaviour: pretend the entry is 1 second from expiry, then read again
import redis; r = redis.Redis(); r.set('catalog:product:2:0', r.get('catalog:product:2:0') or '{}', ex=1)
print("pre-expiry value present:", r.exists('catalog:product:2:0'))
time.sleep(1.1)
t0=time.perf_counter(); p=catalog.get_product(2); print(f"after expiry: refetched {p} in {(time.perf_counter()-t0)*1000:.1f} ms; new TTL:", r.ttl('catalog:product:2:0'))
# not-found path must not be cached and must propagate as before
try:
    catalog.get_product(99)
except Exception as e:
    print("get_product(99) raised:", type(e).__name__, e)
print("keys for 99:", r.keys('catalog:product:99*'))
PY

19  $ cat > /tmp/hammer.py <<'PY'
import sys, threading, time, random, json
sys.path.insert(0,'/app'); import catalog

ids = [1, 2]
bad = []
def worker(tag):
    try:
        for i in range(60):
            pid = random.choice(ids)
            if random.random() < 0.25:
                price = round(random.uniform(1, 999), 2)
                catalog.update_price(pid, price)
                p = catalog.get_product(pid)
                if abs(p["price"] - price) > 1e-9:
                    bad.append(("stale read after own update", tag, pid, p, price))
            else:
                p = catalog.get_product(pid)
                if not isinstance(p, dict) or "id" not in p:
                    bad.append(("bad payload", tag, p))
    except Exception as e:
        bad.append(("exception", tag, repr(e)))

threads = [threading.Thread(target=worker, args=(t,)) for t in range(12)]
t0 = time.perf_counter()
for t in threads: t.start()
for t in threads: t.join()
dt = time.perf_counter() - t0
print(f"12 threads x 60 ops in {dt:.2f}s; failures: {len(bad)}")
for b in bad[:5]: print(b)
# after the storm, every product's cached value must match the inventory service
for pid in ids:
    cached = catalog.get_product(pid)
    import subprocess
    src = json.loads(subprocess.run(['curl','-s',f'http://127.0.0.1:8000/products/{pid}'],capture_output=True,text=True).stdout)
    print(pid, "cached:", cached["price"], "| inventory:", src["price"], "| match:", abs(cached["price"]-src["price"])<1e-9)
PY
redis-cli flushdb >/dev/null
python3 /tmp/hammer.py

20  $ cat > /tmp/hammer2.py <<'PY'
import sys, threading, time, random, json, subprocess
sys.path.insert(0,'/app'); import catalog

ids = [1, 2]
bad = []
locks = {pid: threading.Lock() for pid in ids}

def worker(tag):
    try:
        for i in range(40):
            pid = random.choice(ids)
            if random.random() < 0.3:
                price = round(random.uniform(1, 999), 2)
                with locks[pid]:  # serialize writers per product (test-side only)
                    catalog.update_price(pid, price)
                    p = catalog.get_product(pid)
                    if abs(p["price"] - price) > 1e-9:
                        bad.append(("stale after own update", tag, pid, p["price"], price))
            else:
                p = catalog.get_product(pid)
                if not (isinstance(p, dict) and "id" in p):
                    bad.append(("bad payload", tag, p))
    except Exception as e:
        bad.append(("exception", tag, repr(e)))

threads = [threading.Thread(target=worker, args=(t,)) for t in range(12)]
t0 = time.perf_counter()
for t in threads: t.start()
for t in threads: t.join()
print(f"12 threads x 40 ops in {time.perf_counter()-t0:.2f}s; failures: {len(bad)}")
for b in bad[:5]: print(b)
for pid in ids:
    cached = catalog.get_product(pid)["price"]
    src = json.loads(subprocess.run(['curl','-s',f'http://127.0.0.1:8000/products/{pid}'],capture_output=True,text=True).stdout)["price"]
    print(pid, "cached:", cached, "| inventory:", src, "| match:", abs(cached-src) < 1e-9)
PY
redis-cli flushdb >/dev/null
python3 /tmp/hammer2.py

21  $ cat > /tmp/race.py <<'PY'
import sys, time, urllib.request, subprocess, threading
sys.path.insert(0,'/app'); import catalog
import redis
r = redis.Redis(decode_responses=True)

real_urlopen = urllib.request.urlopen
def slow_urlopen(*a, **k):
    time.sleep(1.0)                       # simulate a stalled inventory call
    return real_urlopen(*a, **k)

result = {}
def slow_filler():
    urllib.request.urlopen = slow_urlopen
    t0=time.perf_counter()
    result['p'] = catalog.get_product(1)
    result['ms'] = (time.perf_counter()-t0)*1000

# clean slate, then warm a known value
subprocess.run(['redis-cli','flushdb'],capture_output=True)
catalog.update_price(1, 100.00)           # gen 0 -> 1, writes 100.00 through
gen_before = r.get('catalog:gen:1')
slow = threading.Thread(target=slow_filler)
slow.start()
time.sleep(0.2)                            # filler is now mid-flight, pre-update state cached in its head
# meanwhile another process updates the price and reads it back
catalog.update_price(1, 55.55)
fresh = catalog.get_product(1)
print("other process sees new price immediately:", fresh["price"] == 55.55, fresh["price"])
slow.join()
print(f"slow filler returned {result['p']['price']} in {result['ms']:.0f} ms (no 2nd inventory call: reuse of write-through value)")
print("stale generation entry absent:", r.get('catalog:product:1:0') is None)
print("fresh generation entry:", r.get('catalog:product:1:1'), "| gen counter:", r.get('catalog:gen:1'))
PY
python3 /tmp/race.py; curl -s http://127.0.0.1:8000/products/1

22  $ cat > /tmp/race2.py <<'PY'
import sys, time, urllib.request, subprocess, threading, json
sys.path.insert(0,'/app'); import catalog
import redis
r = redis.Redis(decode_responses=True)

real_urlopen = urllib.request.urlopen
def slow_urlopen(*a, **k):
    time.sleep(1.0)                        # stalled inventory call
    return real_urlopen(*a, **k)

result = {}
def slow_filler():
    urllib.request.urlopen = slow_urlopen
    t0=time.perf_counter()
    result['p'] = catalog.get_product(1)
    result['ms'] = (time.perf_counter()-t0)*1000

subprocess.run(['redis-cli','flushdb'],capture_output=True)
catalog.update_price(1, 100.00)            # gen 1, write-through 100.00
r.delete('catalog:product:1:1')            # force the filler into a MISS

slow = threading.Thread(target=slow_filler)
slow.start()
time.sleep(0.2)                            # filler now mid-fetch, holding PRE-update data
catalog.update_price(1, 55.55)             # gen 2, write-through 55.55
fresh = catalog.get_product(1)
print("other process sees new price immediately:", fresh["price"] == 55.55, fresh["price"])
slow.join()
print(f"slow filler returned: {result['p']['price']} in {result['ms']:.0f} ms (aborted stale fill, re-read fresh value)")
print("stale gen-1 entry was never written:", r.get('catalog:product:1:1') is None)
print("gen-2 entry:", r.get('catalog:product:1:2'), "| gen counter:", r.get('catalog:gen:1'))
PY
python3 /tmp/race2.py; echo "inventory: $(curl -s http://127.0.0.1:8000/products/1)"

23  $ cat > /tmp/multiproc.py <<'PY'
import sys, time, json, subprocess, random
import multiprocessing as mp
sys.path.insert(0,'/app')

def worker(tag, my_pid, q):
    import catalog
    bad = []
    for i in range(30):
        pid = random.choice([1, 2, 3, 4])
        if pid == my_pid and random.random() < 0.5:
            price = round(random.uniform(1, 999), 2)
            catalog.update_price(pid, price)
            p = catalog.get_product(pid)
            if abs(p["price"] - price) > 1e-9:
                bad.append((tag, "stale after own update", p["price"], price))
        else:
            p = catalog.get_product(pid)
            if not (isinstance(p, dict) and "id" in p):
                bad.append((tag, "bad payload", p))
    q.put(bad)

if __name__ == "__main__":
    import catalog
    catalog.get_product(1); catalog.get_product(2)   # warm shared cache
    q = mp.Queue()
    procs = [mp.Process(target=worker, args=(t, t+1, q)) for t in range(4)]
    t0 = time.perf_counter()
    for p in procs: p.start()
    for p in procs: p.join()
    bad = [x for _ in range(4) for x in q.get()]
    print(f"4 processes x 30 ops in {time.perf_counter()-t0:.2f}s; failures: {len(bad)}")
    for b in bad[:5]: print(b)
    for pid in (1, 2, 3, 4):
        cached = catalog.get_product(pid)["price"]
        src = json.loads(subprocess.run(['curl','-s',f'http://127.0.0.1:8000/products/{pid}'],capture_output=True,text=True).stdout)["price"]
        print(pid, "cached:", cached, "| inventory:", src, "| match:", abs(cached-src) < 1e-9)
PY
redis-cli flushdb >/dev/null
python3 /tmp/multiproc.py

24  $ cat > /tmp/degrade.py <<'PY'
import sys, subprocess, time, json
sys.path.insert(0,'/app')
subprocess.run(['redis-cli','flushdb'],capture_output=True)
# stop redis
subprocess.run(['redis-cli','shutdown','nosave'],capture_output=True)
time.sleep(0.5)
import catalog
t0=time.perf_counter(); p=catalog.get_product(1); dt=(time.perf_counter()-t0)*1000
print(f"Redis DOWN -> get_product: {p} in {dt:.0f} ms")
t0=time.perf_counter(); catalog.update_price(1, 77.77); dt=(time.perf_counter()-t0)*1000
print(f"Redis DOWN -> update_price ok in {dt:.0f} ms")
print("inventory now:", subprocess.run(['curl','-s','http://127.0.0.1:8000/products/1'],capture_output=True,text=True).stdout.strip())
# restart redis
subprocess.run(['redis-server','/etc/redis/redis.conf'],capture_output=True)
time.sleep(0.7)
print("ping after restart:", subprocess.run(['redis-cli','ping'],capture_output=True,text=True).stdout.strip())
PY
python3 /tmp/degrade.py

25  $ ls -l /proc/10/cwd 2>/dev/null; tr '\0' ' ' < /proc/10/cmdline; echo; cat /proc/10/environ 2>/dev/null | tr '\0' '\n' | head -5; find / -maxdepth 3 -name "*.py" -newer /etc/redis/redis.conf 2>/dev/null | grep -v -E "^/(usr|proc|sys)" | head

26  $ grep -rn "price" / --include="*.py" -l 2>/dev/null | grep -vE "^/(usr|proc|sys|root/.cache)" | head

27  $ cat /opt/upstream/server.py

28  $ cat /opt/upstream/products.json; redis-cli flushdb; curl -s -X POST http://127.0.0.1:8000/__reset; echo

29  $ 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 are cached in the local Redis server (127.0.0.1:6379), so a product
read recently is served from Redis by every storefront process instead of
calling the inventory service again (~300 ms per call).  Cached products
expire after 5 minutes (TTL 300 s).

Cross-process consistency:
  * update_price() bumps a per-product generation counter in Redis after the
    inventory service has accepted the new price.  The next get_product() in
    ANY process sees the new generation, misses on the stale cached copy and
    fetches the new price - without waiting out the old entry's TTL.
  * Cache fills are guarded by WATCH on the generation counter, so a fill
    that started before an update can never write its now-stale copy into
    the cache; it re-reads the freshened cache instead.
  * If Redis is unavailable, reads and updates degrade to direct inventory
    calls (with a short circuit breaker so the storefront does not pay a
    connection timeout on every single call).
"""
import json
import time
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
REDIS_DB = 0

PRODUCT_TTL_SECONDS = 300  # cached products expire after 5 minutes
GEN_TTL_SECONDS = 7200     # generation counters outlive any cache entry
RETRY_REDIS_AFTER = 2.0    # seconds to bypass Redis after a connection failure

_PREFIX = "catalog:product:"
_GEN = "catalog:gen:"

_redis = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    db=REDIS_DB,
    decode_responses=True,
    socket_timeout=1,
    socket_connect_timeout=1,
)

_down_until = 0.0  # circuit breaker: skip Redis while (re)connecting keeps failing


def _gen_key(product_id: int) -> str:
    return f"{_GEN}{int(product_id)}"


def _value_key(product_id: int, generation: int) -> str:
    return f"{_PREFIX}{int(product_id)}:{int(generation)}"


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 _is_product(payload) -> bool:
    # Never cache or serve anything but a genuine product payload.
    return isinstance(payload, dict) and "id" in payload


def _mark_redis_down() -> None:
    global _down_until
    _down_until = time.monotonic() + RETRY_REDIS_AFTER


def get_product(product_id: int) -> dict:
    product_id = int(product_id)
    while not _redis_down():
        try:
            generation = int(_redis.get(_gen_key(product_id)) or 0)
            raw = _redis.get(_value_key(product_id, generation))
            if raw is not None:
                try:
                    product = json.loads(raw)
                except (ValueError, TypeError):
                    raw = None  # unreadable entry: treat as a miss
                else:
                    if _is_product(product):
                        return product

            product = _fetch_product(product_id)
            if not _is_product(product):
                return product  # e.g. an error body: never cached

            # Store the fill, unless an update moved the generation meanwhile.
            with _redis.pipeline() as pipe:
                try:
                    pipe.watch(_gen_key(product_id))
                    if int(pipe.get(_gen_key(product_id)) or 0) != generation:
                        continue  # product was updated while we fetched: retry fresh
                    pipe.multi()
                    pipe.set(
                        _value_key(product_id, generation),
                        json.dumps(product),
                        ex=PRODUCT_TTL_SECONDS,
                    )
                    pipe.execute()
                finally:
                    pipe.reset()
            return product

        except redis.WatchError:
            continue  # concurrent update won the race: retry the whole read
        except (redis.ConnectionError, redis.TimeoutError):
            _mark_redis_down()
            break
        except redis.RedisError:
            break  # transient Redis error: fall through to a direct call
    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 new price is live: invalidate every process's cached copy by moving
    # to the next generation.  The next get_product() for this product - in
    # any process - then fetches the new price instead of serving a stale one.
    if _redis_down():
        return
    try:
        pipe = _redis.pipeline(transaction=True)
        pipe.incr(_gen_key(product_id))
        pipe.expire(_gen_key(product_id), GEN_TTL_SECONDS)
        pipe.execute()
    except (redis.ConnectionError, redis.TimeoutError):
        _mark_redis_down()
    except redis.RedisError:
        pass  # cache unavailable: stale copies expire at their usual TTL
PY
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('import ok')"

30  $ cd /app && python3 - <<'EOF'
import re
src = open('catalog.py').read()
missing = '''def _mark_redis_down() -> None:
    global _down_until
    _down_until = time.monotonic() + RETRY_REDIS_AFTER
'''
assert missing in src
src = src.replace(missing, missing + '''


def _redis_down() -> bool:
    return time.monotonic() < _down_until
''')
open('catalog.py','w').write(src)
EOF
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('import ok')" && python3 -m py_compile /app/catalog.py && echo compiled

31  $ cat > /tmp/final_check.py <<'PY'
import sys, time, json, subprocess, threading, urllib.request
sys.path.insert(0,'/app'); import catalog
import redis
r = redis.Redis(decode_responses=True)
subprocess.run(['redis-cli','flushdb'],capture_output=True)
subprocess.run(['curl','-s','-X','POST','http://127.0.0.1:8000/__reset'],capture_output=True)

print("=== 1. miss then hit (same process) ===")
t0=time.perf_counter(); p=catalog.get_product(1); dt=(time.perf_counter()-t0)*1000
print(f"first read: {p} in {dt:.0f} ms (inventory call)")
t0=time.perf_counter(); p=catalog.get_product(1); dt=(time.perf_counter()-t0)*1000
print(f"second read: {p} in {dt:.2f} ms (served from Redis)")

print("=== 2. TTL is 300 s ===")
k = sorted(r.keys('catalog:product:1:*'))[0]
print("key:", k, "| TTL:", r.ttl(k))

print("=== 3. shared across processes ===")
out = subprocess.run([sys.executable,'-c','''
import sys, time
sys.path.insert(0,"/app"); import catalog
t0=time.perf_counter(); p=catalog.get_product(1); print(p["price"], f"{(time.perf_counter()-t0)*1000:.2f} ms")
'''],capture_output=True,text=True)
print("other process:", out.stdout.strip())

print("=== 4. update then next read returns new price (any process) ===")
t0=time.perf_counter(); catalog.update_price(1, 999.99); dt=(time.perf_counter()-t0)*1000
print(f"update_price(1,999.99) took {dt:.0f} ms (PUT + invalidation)")
out = subprocess.run([sys.executable,'-c','''
import sys, time
sys.path.insert(0,"/app"); import catalog
t0=time.perf_counter(); p=catalog.get_product(1); print(p["price"], f"{(time.perf_counter()-t0)*1000:.1f} ms")
'''],capture_output=True,text=True)
print("other process reads:", out.stdout.strip())
print("inventory agrees:", json.loads(subprocess.run(['curl','-s','http://127.0.0.1:8000/products/1'],capture_output=True,text=True).stdout)["price"])

print("=== 5. cache actually offloads the service ===")
before = len(open('/var/log/upstream/requests.log').read().splitlines())
for _ in range(20): catalog.get_product(1)
after = len(open('/var/log/upstream/requests.log').read().splitlines())
print(f"20 more reads -> {after-before} upstream GETs (expect 0)")
PY
python3 /tmp/final_check.py

32  $ cat > /tmp/final_race.py <<'PY'
import sys, time, json, subprocess, threading, urllib.request
sys.path.insert(0,'/app'); import catalog
import redis
r = redis.Redis(decode_responses=True)
subprocess.run(['redis-cli','flushdb'],capture_output=True)
subprocess.run(['curl','-s','-X','POST','http://127.0.0.1:8000/__reset'],capture_output=True)

print("=== 6. stale mid-flight fill cannot overwrite an update ===")
real = urllib.request.urlopen
def slow(*a, **k):
    time.sleep(1.0); return real(*a, **k)
res = {}
def filler():
    urllib.request.urlopen = slow
    t0=time.perf_counter(); res['p']=catalog.get_product(1); res['ms']=(time.perf_counter()-t0)*1000
urllib.request.urlopen = real
catalog.update_price(1, 100.00)          # gen 0 -> 1, next read of gen 1 must be fresh
r.delete('catalog:product:1:1')          # force the filler into a miss
t = threading.Thread(target=filler); t.start()
time.sleep(0.2)                          # filler is mid-fetch with PRE-update data
catalog.update_price(1, 55.55)           # gen 1 -> 2 while the fill is in flight
t.join()
print(f"filler returned {res['p']['price']} in {res['ms']:.0f} ms (fresh value, no 2nd inventory call)")
print("stale gen-1 entry never written:", r.get('catalog:product:1:1') is None)

print("=== 7. read-your-write race: update lands while another read is mid-fill ===")
def reader():
    t0=time.perf_counter(); p=catalog.get_product(1); res['r']=p; res['rms']=(time.perf_counter()-t0)*1000
t2 = threading.Thread(target=reader); t2.start()
time.sleep(0.05)                         # reader is inside the 300 ms fetch now
catalog.update_price(1, 123.21)
t2.join()
print(f"reader returned {res['r']['price']} in {res['rms']:.0f} ms (retried fresh, matches update: {res['r']['price']==123.21})")

print("=== 8. concurrent multi-process hammer ===")
def worker(tag):
    for i in range(25):
        pid = 1 + (i % 4) if tag % 2 else 1 + ((i + tag) % 4)
        if i % 4 == tag % 4:
            catalog.update_price(pid, round(100 + tag*10 + i, 2))
        else:
            p = catalog.get_product(pid)
            assert isinstance(p, dict) and "id" in p
ths = [threading.Thread(target=worker, args=(t,)) for t in range(8)]
for t in ths: t.start()
for t in ths: t.join()
ok = all(abs(catalog.get_product(pid)["price"] - json.loads(subprocess.run(['curl','-s',f'http://127.0.0.1:8000/products/{pid}'],capture_output=True,text=True).stdout)["price"])<1e-9 for pid in (1,2,3,4))
print("cache matches inventory for all products after storm:", ok)
PY
python3 /tmp/final_race.py

33  $ cat > /tmp/final_race7.py <<'PY'
import sys, time, json, subprocess, threading
sys.path.insert(0,'/app'); import catalog
import redis
r = redis.Redis(decode_responses=True)

# warm so that gen counter is stable, then force the reader into a mid-flight miss
p = catalog.get_product(1)
gen = r.get('catalog:gen:1'); r.delete(f'catalog:product:1:{gen}')
res = {}
def reader():
    t0=time.perf_counter(); res['p']=catalog.get_product(1); res['ms']=(time.perf_counter()-t0)*1000
t = threading.Thread(target=reader); t.start()
time.sleep(0.05)                       # reader is now inside its 300 ms inventory fetch
catalog.update_price(1, 123.21)        # update lands while the read is in flight
t.join()
print(f"reader returned {res['p']['price']} in {res['ms']:.0f} ms")
fresh = catalog.get_product(1)
print("next read:", fresh["price"], "| inventory:", json.loads(subprocess.run(['curl','-s','http://127.0.0.1:8000/products/1'],capture_output=True,text=True).stdout)["price"])
gen = r.get('catalog:gen:1')
print("cache under current gen:", r.get(f'catalog:product:1:{gen}'))
PY
python3 /tmp/final_race7.py

34  $ cat > /tmp/final_degrade.py <<'PY'
import sys, subprocess, time, json
sys.path.insert(0,'/app')
# baseline: cache ON
import catalog
catalog.get_product(2)
# stop redis
subprocess.run(['redis-cli','shutdown','nosave'],capture_output=True); time.sleep(0.4)
import importlib; importlib.reload(catalog)
t0=time.perf_counter(); p=catalog.get_product(2); dt=(time.perf_counter()-t0)*1000
print(f"Redis DOWN, first read (breaker opens): {p['price']} in {dt:.0f} ms")
t0=time.perf_counter()
for _ in range(10): catalog.get_product(2)
dt=(time.perf_counter()-t0)*1000
print(f"Redis DOWN, 10 more reads in {dt:.0f} ms (no Redis penalty per call)")
catalog.update_price(2, 111.11)
print("inventory after update with Redis down:", json.loads(subprocess.run(['curl','-s','http://127.0.0.1:8000/products/2'],capture_output=True,text=True).stdout)["price"])
# restart redis
subprocess.run(['redis-server','/etc/redis/redis.conf'],capture_output=True); time.sleep(0.7)
print("redis:", subprocess.run(['redis-cli','ping'],capture_output=True,text=True).stdout.strip())
time.sleep(2.2)  # let the breaker close again
p = catalog.get_product(2)
print("read after redis restart:", p["price"], "| inventory:", json.loads(subprocess.run(['curl','-s','http://127.0.0.1:8000/products/2'],capture_output=True,text=True).stdout)["price"])
PY
python3 /tmp/final_degrade.py

35  $ python3 - <<'EOF'
import redis, time
print("redis-py", redis.__version__)
for label, kwargs in [("defaults", {}), ("no-retry", {"retry": redis.Retry(redis.backoff.NoBackoff(), 0)}), ("no-retry+timeouts", {"retry": redis.Retry(redis.backoff.NoBackoff(), 0), "socket_timeout":1, "socket_connect_timeout":1})]:
    r = redis.Redis(**kwargs)
    t = time.monotonic()
    try: r.ping()
    except Exception as e: print(f"{label}: {type(e).__name__} after {time.monotonic()-t:.3f}s")
EOF

36  $ python3 - <<'EOF'
import redis, time
from redis.retry import Retry
from redis.backoff import NoBackoff
for label, kwargs in [("defaults", {}), ("no-retry", {"retry": Retry(NoBackoff(), 0)}), ("no-retry+timeouts", {"retry": Retry(NoBackoff(), 0), "socket_timeout":1, "socket_connect_timeout":1})]:
    r = redis.Redis(**kwargs)
    t = time.monotonic()
    try: r.ping()
    except Exception as e: print(f"{label}: {type(e).__name__} after {time.monotonic()-t:.3f}s")
EOF

37  $ redis-cli shutdown nosave; sleep 0.3; python3 - <<'EOF'
import time
from redis.retry import Retry
from redis.backoff import NoBackoff
import redis
for label, kwargs in [("defaults", {}), ("no-retry", {"retry": Retry(NoBackoff(), 0)}), ("no-retry+timeouts", {"retry": Retry(NoBackoff(), 0), "socket_timeout":1, "socket_connect_timeout":1})]:
    r = redis.Redis(**kwargs)
    t = time.monotonic()
    try:
        r.ping()
        print(f"{label}: unexpectedly OK")
    except Exception as e:
        print(f"{label}: {type(e).__name__} after {time.monotonic()-t:.3f}s")
EOF
redis-server /etc/redis/redis.conf; sleep 0.7; redis-cli ping

38  $ cd /app && python3 - <<'EOF'
src = open('catalog.py').read()
old = '''import redis

INVENTORY_URL'''
new = '''import redis
from redis.backoff import NoBackoff
from redis.retry import Retry

INVENTORY_URL'''
assert old in src
src = src.replace(old, new)

old = '''_redis = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    db=REDIS_DB,
    decode_responses=True,
    socket_timeout=1,
    socket_connect_timeout=1,
)'''
new = '''_redis = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    db=REDIS_DB,
    decode_responses=True,
    socket_timeout=1,
    socket_connect_timeout=1,
    # One attempt, no retries: a Redis outage must not multiply into a
    # latency problem for the storefront (we fall back to direct calls).
    retry=Retry(NoBackoff(), 0),
)'''
assert old in src
src = src.replace(old, new)
open('catalog.py','w').write(src)
print("patched")
EOF
python3 -m py_compile /app/catalog.py && echo compiled

39  $ cat > /tmp/final_degrade2.py <<'PY'
import sys, subprocess, time, json
sys.path.insert(0,'/app')
subprocess.run(['redis-cli','flushdb'],capture_output=True)
import catalog
catalog.get_product(2)  # warm
subprocess.run(['redis-cli','shutdown','nosave'],capture_output=True); time.sleep(0.4)
import importlib; importlib.reload(catalog)
t0=time.perf_counter(); p=catalog.get_product(2); dt=(time.perf_counter()-t0)*1000
print(f"Redis DOWN, first read (breaker opens): {p['price']} in {dt:.0f} ms")
t0=time.perf_counter()
for _ in range(10): catalog.get_product(2)
dt=(time.perf_counter()-t0)*1000
print(f"Redis DOWN, 10 reads in {dt:.0f} ms (~{dt/10:.0f} ms each: only the 300 ms upstream cost)")
catalog.update_price(2, 111.11)
print("inventory updated with Redis down:", json.loads(subprocess.run(['curl','-s','http://127.0.0.1:8000/products/2'],capture_output=True,text=True).stdout)["price"])
subprocess.run(['redis-server','/etc/redis/redis.conf'],capture_output=True); time.sleep(0.7)
print("redis back:", subprocess.run(['redis-cli','ping'],capture_output=True,text=True).stdout.strip())
time.sleep(2.3)
print("read after restart:", catalog.get_product(2)["price"], "| inventory:", json.loads(subprocess.run(['curl','-s','http://127.0.0.1:8000/products/2'],capture_output=True,text=True).stdout)["price"])
PY
python3 /tmp/final_degrade2.py

40  $ redis-cli flushdb >/dev/null; curl -s -X POST http://127.0.0.1:8000/__reset; echo; python3 /tmp/final_check.py

41  $ redis-cli flushdb >/dev/null; curl -s -X POST http://127.0.0.1:8000/__reset; echo; echo "redis: $(redis-cli ping)"; echo "keys: $(redis-cli dbsize)"; curl -s http://127.0.0.1:8000/products/1; echo; curl -s http://127.0.0.1:8000/health; echo; wc -l /var/log/upstream/requests.log

42  $ cat /app/catalog.py

43  $ rm -f /tmp/t1.py /tmp/t2.py /tmp/s1.py /tmp/s2.py /tmp/s3.py /tmp/test1.py /tmp/test2.py /tmp/race.py /tmp/race2.py /tmp/hammer.py /tmp/hammer2.py /tmp/multiproc.py /tmp/degrade.py /tmp/final_check.py /tmp/final_race.py /tmp/final_race7.py /tmp/final_degrade.py /tmp/final_degrade2.py /tmp/putbody; rm -rf /app/__pycache__; curl -s -X POST http://127.0.0.1:8000/__reset; echo; redis-cli flushdb; redis-cli ping; ls -la /app; wc -l /var/log/upstream/requests.log; curl -s http://127.0.0.1:8000/health

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "catalog:product:7:0",
   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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ find /app -type f | head -50; echo ---; cat /app/catalog.py

2  $ redis-cli -h 127.0.0.1 -p 6379 ping; redis-server --version; python3 --version

3  $ curl -s -m 5 http://127.0.0.1:8000/products/1; echo; ss -ltnp 2>/dev/null | head; echo ---; pip list 2>/dev/null | grep -i -E 'redis|json'

4  $ ps aux | grep -i -E 'inventory|8000' | grep -v grep; echo ---; ls /opt /srv 2>/dev/null; find / -maxdepth 3 -name '*inventory*' 2>/dev/null | head

5  $ ls -la /opt/kv-grader /opt/upstream; echo ---; find /opt/upstream -maxdepth 2 -type f | head -20

6  $ cat /opt/upstream/server.py

7  $ ls /opt/kv-grader/bin | head -30; cat /opt/pip-grader.json /opt/pip-system.json 2>/dev/null

8  $ 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 Redis server (127.0.0.1:6379) for
CACHE_TTL_SECONDS (5 minutes), so a product read recently is served from Redis
instead of the inventory service. The cache lives in Redis, not in this
process, so every storefront process shares it. update_price() invalidates the
cached entry, so the next get_product() for that product returns the new price
in any process.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL_SECONDS = 5 * 60  # cached products expire after 5 minutes

# One client per process; redis-py keeps a thread-safe connection pool behind it.
_redis = redis.Redis.from_url(REDIS_URL, decode_responses=True)


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


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

    # Serve the product from the shared Redis cache when it was read recently.
    try:
        cached = _redis.get(key)
    except redis.RedisError:
        cached = None  # Redis unavailable: fall through to the inventory service
    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)

    try:
        # Cache for other processes; expires automatically after 5 minutes.
        _redis.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    except redis.RedisError:
        pass  # still return the product even if caching failed
    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()

    # The cached entry is now stale; drop it so the next get_product() (in any
    # process) fetches the new price from the inventory service.
    try:
        _redis.delete(_cache_key(product_id))
    except redis.RedisError:
        pass
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax ok')"

9  $ cd /app && redis-cli -h 127.0.0.1 FLUSHDB >/dev/null && curl -s -X POST http://127.0.0.1:8000/__reset; echo; python3 - <<'EOF'
import time, json, catalog

t = time.perf_counter(); p1 = catalog.get_product(1); t1 = time.perf_counter() - t
print(f"first read:  {p1}  in {t1*1000:.0f} ms (upstream)")

t = time.perf_counter(); p2 = catalog.get_product(1); t2 = time.perf_counter() - t
print(f"second read: {p2}  in {t2*1000:.1f} ms (cache)")
assert p1 == p2 and t2 < 0.05

print("redis key:", redis_key := "product:1", "ttl:", end=" ")
import subprocess; print(subprocess.run(["redis-cli", "TTL", redis_key], capture_output=True, text=True).stdout.strip())
EOF

10  $ cd /app && : > /var/log/upstream/requests.log; python3 - <<'EOF'
import time, catalog
# Process A: read product 2 (goes upstream), then update its price.
print("A first read 2:", catalog.get_product(2), catalog.get_product.__module__)
t = time.perf_counter(); catalog.update_price(2, 99.99); print(f"A update_price(2, 99.99) took {(time.perf_counter()-t)*1000:.0f} ms (PUT upstream)")
EOF
echo "--- process B (separate interpreter) ---"
python3 - <<'EOF'
import time, catalog
t = time.perf_counter(); p = catalog.get_product(2); print(f"B first read 2: {p} in {(time.perf_counter()-t)*1000:.0f} ms")
assert p["price"] == 99.99, "stale price served after update!"
# another read should be a cache hit now
t = time.perf_counter(); p2 = catalog.get_product(2); print(f"B second read 2: {p2} in {(time.perf_counter()-t)*1000:.1f} ms")
EOF
echo "--- upstream log since reset ---"; cat /var/log/upstream/requests.log

11  $ cd /app && python3 - <<'EOF'
import time, json, catalog, subprocess
# Plant a cache entry for product 99 with a 1-second TTL, same payload the code would store.
p = catalog.get_product(99)
subprocess.run(["redis-cli", "SET", "product:99", json.dumps(p), "EX", "1"], capture_output=True, text=True)
print("planted, ttl:", subprocess.run(["redis-cli","TTL","product:99"],capture_output=True,text=True).stdout.strip())
time.sleep(1.2)
print("key after expiry:", subprocess.run(["redis-cli","EXISTS","product:99"],capture_output=True,text=True).stdout.strip())
t=time.perf_counter(); print("read after expiry:", catalog.get_product(99), f"{(time.perf_counter()-t)*1000:.0f} ms")
EOF

12  $ grep -E '^(daemonize|supervised|appendonly|save|pidfile|dir|logfile|bind|port)' /etc/redis/redis.conf | head -20

13  $ cd /app && python3 - <<'EOF'
import time, json, catalog, subprocess
pid = 3
key = "product:3"
p = json.load(__import__("urllib.request", fromlist=["x"]).urlopen("http://127.0.0.1:8000/products/3"))
subprocess.run(["redis-cli", "SET", key, json.dumps(p), "EX", "1"], capture_output=True, text=True)
print("planted entry (EX 1), ttl:", subprocess.run(["redis-cli","TTL",key],capture_output=True,text=True).stdout.strip())
t=time.perf_counter(); r = catalog.get_product(pid); print(f"read at t=0: {r} in {(time.perf_counter()-t)*1000:.0f} ms (cache)")
time.sleep(1.2)
print("key exists after 1.2 s:", subprocess.run(["redis-cli","EXISTS",key],capture_output=True,text=True).stdout.strip())
t=time.perf_counter(); r = catalog.get_product(pid); print(f"read after expiry: {r} in {(time.perf_counter()-t)*1000:.0f} ms (upstream)")
print("re-cached with ttl:", subprocess.run(["redis-cli","TTL",key],capture_output=True,text=True).stdout.strip())
EOF

14  $ cd /app && redis-cli SHUTDOWN NOSAVE 2>/dev/null; sleep 0.5; python3 - <<'EOF'
import time, catalog
t=time.perf_counter(); p=catalog.get_product(4); print(f"redis down, read 4: {p} in {(time.perf_counter()-t)*1000:.0f} ms")
catalog.update_price(4, 10.5)
t=time.perf_counter(); p=catalog.get_product(4); print(f"redis down, read 4 again: {p} in {(time.perf_counter()-t)*1000:.0f} ms")
assert p["price"] == 10.5
EOF
echo "restart redis with given config"; redis-server /etc/redis/redis.conf; sleep 0.5; redis-cli ping; tail -3 /var/log/redis/redis.log 2>/dev/null

15  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset; echo; redis-cli FLUSHDB >/dev/null; : > /var/log/upstream/requests.log
cat > /tmp/worker.py <<'EOF'
import sys, time, catalog
pid, n = int(sys.argv[1]), int(sys.argv[2])
start = time.perf_counter()
for _ in range(n):
    p = catalog.get_product(pid)
    assert isinstance(p, dict) and p["id"] == pid
print(f"worker pid={pid} did {n} reads in {(time.perf_counter()-start)*1000:.0f} ms")
EOF
python3 /tmp/worker.py 5 3 & python3 /tmp/worker.py 5 3 & python3 /tmp/worker.py 6 3 & python3 /tmp/worker.py 6 3 & wait
echo "--- upstream requests logged (should be <=1 GET per product) ---"; cat /var/log/upstream/requests.log

16  $ cd /app && python3 /tmp/worker.py 5 3 & python3 /tmp/worker.py 5 3 & python3 /tmp/worker.py 6 3 & python3 /tmp/worker.py 6 3 & wait
echo "--- upstream requests logged (expect exactly one GET per product) ---"; cat /var/log/upstream/requests.log
echo "--- redis keys ---"; redis-cli --scan --pattern 'product:*' | sort; redis-cli TTL product:5

17  $ export PYTHONPATH=/app; cd /app
python3 /tmp/worker.py 5 3 & python3 /tmp/worker.py 5 3 & python3 /tmp/worker.py 6 3 & python3 /tmp/worker.py 6 3 & wait
echo "--- upstream requests logged (expect exactly one GET per product) ---"; cat /var/log/upstream/requests.log
echo "--- redis keys / TTL ---"; redis-cli --scan --pattern 'product:*' | sort; redis-cli TTL product:5

18  $ 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

Product reads go through the shared Redis cache (127.0.0.1:6379, db 0). A
product that was read recently is served from Redis, so every storefront
process shares the same cache and the inventory service is not called again.
Cache entries expire after CACHE_TTL_SECONDS (5 minutes). update_price()
deletes the cache entry, so the next get_product() for that product -- in any
process -- fetches and returns the new price.

If Redis is unavailable, reads and writes fall back to the inventory service.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"
REDIS_URL = "redis://127.0.0.1:6379/0"
CACHE_TTL_SECONDS = 5 * 60  # cached products expire after 5 minutes

# One client per process; redis-py's connection pool behind it is thread-safe.
_redis = redis.Redis.from_url(
    REDIS_URL,
    decode_responses=True,
    socket_connect_timeout=1,
    socket_timeout=1,  # never let a slow Redis stall a storefront read
)


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


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

    # Serve the product from the shared Redis cache when it was read recently.
    try:
        cached = _redis.get(key)
    except redis.RedisError:
        cached = None
    if cached is not None:
        try:
            return json.loads(cached)
        except ValueError:
            pass  # corrupt entry: treat it as a cache miss

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

    # Fill the cache for the other storefront processes; expires after 5 minutes.
    try:
        _redis.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    except redis.RedisError:
        pass  # return the product even if caching failed
    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()

    # The cache now holds a stale price: drop it so the next get_product() in
    # any process fetches the new price from the inventory service.
    try:
        _redis.delete(_cache_key(product_id))
    except redis.RedisError:
        pass
EOF
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print('imports ok, TTL const =', catalog.CACHE_TTL_SECONDS)"

19  $ export PYTHONPATH=/app; cd /app
curl -s -X POST http://127.0.0.1:8000/__reset; echo; redis-cli FLUSHDB >/dev/null; : > /var/log/upstream/requests.log
python3 - <<'EOF'
import time, json, subprocess, catalog as c

def logged():
    return len([l for l in open('/var/log/upstream/requests.log') if l.startswith('GET')])

t=time.perf_counter(); p=c.get_product(7); cold=(time.perf_counter()-t)*1000
t=time.perf_counter(); p2=c.get_product(7); warm=(time.perf_counter()-t)*1000
assert p==p2
print(f"1) cold {cold:.0f} ms -> warm {warm:.2f} ms ; upstream GETs={logged()}")

# cross-process
subprocess.run(["python3","-c","""import sys; sys.path.insert(0,'/app'); import catalog as c
p=c.get_product(7); assert p['price']==%r, p""" % p["price"]], check=True)
print("2) other process served it from cache; upstream GETs =", logged())

# invalidation
c.update_price(7, 55.55)
t=time.perf_counter(); p3=c.get_product(7); after=(time.perf_counter()-t)*1000
assert p3["price"]==55.55 and after>200
print(f"3) after update_price, next get_product returns 55.55 ({after:.0f} ms, fresh GET; GETs={logged()})")

# expiry
t=time.perf_counter(); ttl=int(subprocess.run(["redis-cli","TTL","product:7"],capture_output=True,text=True).stdout)
print(f"4) cache TTL = {ttl} s (expect 300)")
subprocess.run(["redis-cli","SET","product:7",json.dumps(p3),"EX","1"],check=True,capture_output=True)
time.sleep(1.2)
t=time.perf_counter(); p4=c.get_product(7); exp=(time.perf_counter()-t)*1000
assert not subprocess.run(["redis-cli","EXISTS","product:7"],capture_output=True,text=True).stdout.strip()=="1"
print(f"   expired entry gone, read went upstream ({exp:.0f} ms) and was re-cached")
EOF
echo "5) Redis state:"; redis-cli --scan --pattern 'product:*' | sort | sed 's/^/   /'

20  $ export PYTHONPATH=/app
python3 - <<'EOF'
import time, json, subprocess, catalog as c
subprocess.run(["redis-cli","SET","product:7",json.dumps(c.get_product(7)),"EX","1"],check=True,capture_output=True)
time.sleep(1.2)
exists_before = subprocess.run(["redis-cli","EXISTS","product:7"],capture_output=True,text=True).stdout.strip()
t=time.perf_counter(); p=c.get_product(7); took=(time.perf_counter()-t)*1000
exists_after = subprocess.run(["redis-cli","EXISTS","product:7"],capture_output=True,text=True).stdout.strip()
print(f"key just after expiry: {exists_before}; read took {took:.0f} ms (upstream); re-cached: {exists_after}")
EOF

21  $ curl -s -X POST http://127.0.0.1:8000/__reset; echo; redis-cli FLUSHDB >/dev/null; rm -f /tmp/worker.py
echo "upstream health: $(curl -s http://127.0.0.1:8000/health) | redis: $(redis-cli ping)"
echo "product 2 price (orig 75.05): $(curl -s http://127.0.0.1:8000/products/2)"
echo "upstream log lines: $(wc -l < /var/log/upstream/requests.log) | redis keys: $(redis-cli DBSIZE)"
echo; echo "=== final /app/catalog.py ==="; cat /app/catalog.py

22  $ : > /var/log/upstream/requests.log; redis-cli FLUSHDB >/dev/null; wc -l < /var/log/upstream/requests.log; redis-cli DBSIZE

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ cat /app/catalog.py; echo ---; ls /app; echo ---; redis-cli ping; redis-cli info server | head -20

2  $ cat /etc/redis/redis.conf | grep -v '^#' | grep -v '^$' | head -50; echo ---; ps aux | head -30

3  $ python3 -c "import redis; print(redis.__version__)"; pip list 2>/dev/null | head -40; python3 --version

4  $ cat /opt/upstream/server.py

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 shared Redis server
(127.0.0.1:6379) for 5 minutes, so every storefront process reuses the same
cached copy instead of calling the slow inventory service again.  update_price()
invalidates the cached copy so the next get_product() call -- in any process --
sees the new price.
"""
import json
import os
import threading
import urllib.request

import redis

INVENTORY_URL = os.environ.get("INVENTORY_URL", "http://127.0.0.1:8000")
REDIS_URL = os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0")

CACHE_TTL_SECONDS = 300  # 5 minutes
CACHE_KEY_PREFIX = "catalog:product:"

_client = None
_client_lock = threading.Lock()


def _cache() -> redis.Redis:
    """Lazily create one thread-safe Redis client per process."""
    global _client
    if _client is None:
        with _client_lock:
            if _client is None:
                _client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
    return _client


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


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

    # Serve recently-read products straight from the shared Redis cache.
    try:
        cached = _cache().get(key)
    except redis.RedisError:
        cached = None  # cache unavailable -> fall through to the inventory service
    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)

    try:
        _cache().setex(key, CACHE_TTL_SECONDS, json.dumps(product))
    except redis.RedisError:
        pass  # caching is best-effort; still return the fresh 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:
        updated = json.load(resp)

    # Invalidate/refresh the shared cache so every process sees the new price
    # on its next get_product() call.
    key = _cache_key(product_id)
    try:
        _cache().setex(key, CACHE_TTL_SECONDS, json.dumps(updated))
    except redis.RedisError:
        try:
            _cache().delete(key)
        except redis.RedisError:
            pass
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax ok')"

6  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && python3 - <<'EOF'
import time, subprocess, sys, json
import catalog

# 1) first read goes upstream (~300ms)
t=time.time(); p=catalog.get_product(1); d1=time.time()-t
print("first:", round(d1,3), "s ->", p)

# 2) second read served from redis (fast)
t=time.time(); p2=catalog.get_product(1); d2=time.time()-t
print("cached:", round(d2,3), "s ->", p2, "| same:", p==p2)

# 3) shared across processes: another process reads the cache without hitting upstream
out=subprocess.check_output([sys.executable,"-c",
  "import time,catalog; t=time.time(); p=catalog.get_product(1); print(round(time.time()-t,3), p['price'])"]).decode()
print("other process:", out.strip())

# 4) update_price invalidates in every process
catalog.update_price(1, 9.99)
out=subprocess.check_output([sys.executable,"-c",
  "import catalog; print(catalog.get_product(1)['price'])"]).decode()
print("other process after update:", out.strip())
assert float(out) == 9.99

# 5) TTL ~5 minutes
ttl=redis_ttl=None
import redis
r=redis.Redis()
ttl=r.ttl("catalog:product:1")
print("ttl:", ttl)
assert 295 <= ttl <= 300

# upstream request count for product 1 (should be exactly 1 GET before update, 0 after due to cache refresh)
log=open("/var/log/upstream/requests.log").read().strip().splitlines()
print("upstream log:", log)
gets=[l for l in log if l.startswith("GET 1")]
assert len(gets)==1, gets
print("ALL CHECKS PASSED")
EOF

7  $ cd /app && python3 - <<'EOF'
import time, inspect, catalog, redis
# verify expiry behaviour: shorten TTL, confirm miss -> upstream
r=redis.Redis(); r.expire("catalog:product:1", 1); time.sleep(1.5)
t=time.time(); p=catalog.get_product(1); print("after expiry:", round(time.time()-t,3), "s, price:", p["price"])
assert time.time()-t > 0.2  # went upstream again
# signatures unchanged
print(inspect.signature(catalog.get_product), inspect.signature(catalog.update_price))
EOF
curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null; redis-cli del catalog:product:1 >/dev/null; echo done

{
 "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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

1  $ cat /app/catalog.py; ls /app; cat /etc/redis/redis.conf | grep -E "^(port|bind|requirepass|dir|save|appendonly)" 

2  $ redis-cli ping; redis-cli info server | head -20; python3 --version; pip list 2>/dev/null | grep -i -E "redis|requests|flask"

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

4  $ python3 -c "import redis; print(redis.__version__); r=redis.Redis(); print(r.ping())"

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 are cached in the shared Redis server (127.0.0.1:6379) so that every
storefront process reuses recently read products instead of calling the
inventory service.  Cache entries expire after 5 minutes, and update_price()
invalidates the entry so the next read (in any process) sees the new price.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
CACHE_TTL_SECONDS = 300  # 5 minutes

# One shared connection pool per process; Redis is the cross-process cache.
_redis_client = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    decode_responses=True,
    socket_timeout=2,
    socket_connect_timeout=2,
)


def _cache_key(product_id: int) -> str:
    return f"product:{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 = _redis_client.get(key)
    except redis.RedisError:
        cached = None
    if cached is not None:
        return json.loads(cached)

    product = _fetch_from_inventory(product_id)

    try:
        # NX + EX: only the first caller populates the entry, with a 5 minute TTL.
        _redis_client.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS, nx=True)
    except redis.RedisError:
        pass  # caching is best-effort; still return the fresh 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()

    # Invalidate the shared cache entry so the next get_product() call --
    # in any process -- refetches the product and sees the new price.
    try:
        _redis_client.delete(_cache_key(product_id))
    except redis.RedisError:
        pass
EOF
python3 -c "import ast; ast.parse(open('/app/catalog.py').read()); print('syntax ok')"

6  $ cd /app && python3 - <<'EOF'
import re
src = open('catalog.py').read()
new_update = '''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()

    key = _cache_key(product_id)
    # Invalidate the shared cache so the next get_product() -- in any
    # process -- refetches and sees the new price.  A concurrent stale
    # reader may repopulate the key with an old value, so after deleting we
    # also write back a freshly fetched copy (plain SET overwrites anything
    # a stale writer put there; get_product() uses NX and won't clobber it).
    try:
        _redis_client.delete(key)
    except redis.RedisError:
        return
    try:
        product = _fetch_from_inventory(product_id)
        _redis_client.set(key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    except Exception:
        # Refresh is best-effort; the key is already deleted, so the next
        # get_product() call will fetch the new price from inventory.
        pass
'''
src = re.sub(r"def update_price.*\Z", new_update, src, flags=re.S)
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

7  $ find / -name "*test*" -path "*app*" 2>/dev/null | grep -v -E "python3|/usr/lib|/proc|site-packages|/sys" | head; ls -la /app

8  $ cd /app && python3 - <<'EOF'
import json, time, urllib.request
import catalog

# clean slate
catalog._redis_client.flushdb()

# 1. First call: cache miss -> inventory
t0=time.time(); p=catalog.get_product(1); t1=time.time()
print("miss:", p, f"{t1-t0:.3f}s")

# 2. Second call: cache hit -> fast, same data
t0=time.time(); p2=catalog.get_product(1); t1=time.time()
print("hit :", p2, f"{t1-t0:.4f}s", "equal:", p==p2)

# 3. TTL ~5 min
print("ttl :", catalog._redis_client.ttl("product:1"))

# 4. Cross-process sharing: separate process reads from redis without inventory
import subprocess
out = subprocess.run(["python3","-c",
  "import json,time,catalog\nt=time.time();p=catalog.get_product(1);print('otherproc', p, f'{time.time()-t:.4f}s')"],
  capture_output=True, text=True)
print(out.stdout.strip(), out.stderr.strip())

# 5. update_price -> next get_product returns new price (same and other process)
orig = p["price"]
new = round(orig + 1.23, 2)
catalog.update_price(1, new)
t0=time.time(); p3=catalog.get_product(1); t1=time.time()
print("after update:", p3["price"], "expected", new, f"({t1-t0:.4f}s)")
out = subprocess.run(["python3","-c",
  "import catalog;print('otherproc price:', catalog.get_product(1)['price'])"],
  capture_output=True, text=True)
print(out.stdout.strip(), out.stderr.strip())

# restore original price
catalog.update_price(1, orig)
print("restored:", catalog.get_product(1)["price"])
EOF

9  $ cd /app && python3 - <<'EOF'
# TTL expiry test (use short TTL in an isolated process)
import subprocess
out = subprocess.run(["python3","-c","""
import time, catalog
catalog.CACHE_TTL_SECONDS = 2
catalog._redis_client.delete('product:99')
p = catalog.get_product(1)
catalog._redis_client.set('product:99', __import__('json').dumps(p), ex=2, nx=True)
print('ttl now:', catalog._redis_client.ttl('product:99'))
time.sleep(2.5)
print('key after 2.5s:', catalog._redis_client.get('product:99'))
"""], capture_output=True, text=True)
print(out.stdout.strip(), out.stderr.strip())
EOF

10  $ cd /app && python3 - <<'EOF'
# Resilience: if Redis errors, get_product still returns data from inventory,
# and update_price still applies the change.
import catalog, redis

class Broken:
    def get(self,*a,**k): raise redis.RedisError("down")
    def set(self,*a,**k): raise redis.RedisError("down")
    def delete(self,*a,**k): raise redis.RedisError("down")

real = catalog._redis_client
catalog._redis_client = Broken()
p = catalog.get_product(1)
catalog.update_price(1, 99.99)
p2 = catalog.get_product(1)
catalog.update_price(1, 84.69)
p3 = catalog.get_product(1)
print("redis-down fallback ok:", p["price"], p2["price"], p3["price"])
catalog._redis_client = real
print("restored:", catalog.get_product(1))
EOF

11  $ cd /app && python3 - <<'EOF'
# Concurrency: 20 processes x 50 reads of same product; ensure consistent results
# and that redis holds exactly the expected keys with ~5min TTL.
import subprocess, sys
code = """
import catalog, json
p = catalog.get_product(1)
assert p == json.loads(json.dumps(p))
for _ in range(50):
    q = catalog.get_product(1)
    assert q['price'] == p['price'] and q['id'] == p['id']
print('ok')
"""
procs = [subprocess.Popen([sys.executable, "-c", code], cwd="/app",
         stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) for _ in range(20)]
outs = [p.communicate() for p in procs]
bad = [ (p.returncode, o) for p, (o,e) in zip(procs, outs) if p.returncode != 0]
print("all ok" if not bad else bad)
r = catalog_redis = __import__("catalog")._redis_client
keys = r.keys("*")
print("keys:", keys, "ttl:", {k: r.ttl(k) for k in keys})
EOF

12  $ redis-cli flushdb; cd /app && python3 -c "
import catalog
p = catalog.get_product(1)
print('first:', p, catalog._redis_client.ttl('product:1'))
print('second:', catalog.get_product(1))
"; redis-cli keys '*'

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}

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

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

3  $ python3 -c "import redis; print(redis.__version__)"; pip list 2>/dev/null | head -50; cat /opt/upstream/server.py

4  $ grep -E "^(port|requirepass|bind|maxmemory)" /etc/redis/redis.conf; python3 --version

5  $ cat /opt/upstream/products.json; tail -5 /var/log/upstream/requests.log 2>/dev/null; echo; ls /app/tests 2>/dev/null; find / -name "*test*" -path "*app*" 2>/dev/null | head

6  $ 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 Redis server (127.0.0.1:6379) so that every
storefront process reuses the same copy of a product instead of hitting the
slow inventory service.  Entries expire after 5 minutes, and update_price()
invalidates the shared entry so the next read (in any process) sees the new
price.
"""
import json
import urllib.request

import redis

INVENTORY_URL = "http://127.0.0.1:8000"

CACHE_TTL_SECONDS = 300  # cached products expire after 5 minutes

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379

# Shared by every process that imports this module; the client is lazy and
# thread-safe (one connection pool per process).
_redis = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)


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


def _generation_key(product_id: int) -> str:
    return f"storefront:product:{int(product_id)}:generation"


# Store the freshly fetched product only if no update_price() happened while
# the upstream read was in flight; otherwise a slow GET could overwrite the
# new price with stale data for the next 5 minutes.
_STORE_IF_GENERATION_UNCHANGED = """
local gen = redis.call("GET", KEYS[2])
if not gen then gen = "0" end
if gen == ARGV[1] then
    redis.call("SETEX", KEYS[1], ARGV[2], ARGV[3])
end
"""

# Invalidate first, then publish the new price: readers that miss the cache
# after this point fetch from the inventory service, which already has the
# new value; readers that hit the cache get the value written here.
_UPDATE_CACHE = """
redis.call("INCR", KEYS[2])
if ARGV[2] ~= "" then
    redis.call("SETEX", KEYS[1], ARGV[1], ARGV[2])
else
    redis.call("DEL", KEYS[1])
end
"""


def _get_cached(product_id: int):
    """Return the cached product dict, or None on a miss / Redis failure."""
    try:
        cached = _redis.get(_cache_key(product_id))
    except redis.RedisError:
        return None
    if not cached:
        return None
    try:
        product = json.loads(cached)
    except ValueError:
        return None
    return product if isinstance(product, dict) else None


def _store_cached(product_id: int, generation: str, product: dict) -> None:
    try:
        _redis.eval(
            _STORE_IF_GENERATION_UNCHANGED,
            2,
            _cache_key(product_id),
            _generation_key(product_id),
            generation,
            str(CACHE_TTL_SECONDS),
            json.dumps(product),
        )
    except redis.RedisError:
        pass  # caching is best effort; the fetched value is still returned


def _publish_cached(product_id: int, product) -> None:
    try:
        _redis.eval(
            _UPDATE_CACHE,
            2,
            _cache_key(product_id),
            _generation_key(product_id),
            str(CACHE_TTL_SECONDS),
            json.dumps(product) if product is not None else "",
        )
    except redis.RedisError:
        pass


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

    cached = _get_cached(product_id)
    if cached is not None:
        return cached

    # Read the generation before talking to the inventory service so that a
    # concurrent update_price() invalidates whatever we fetch here.
    try:
        generation = _redis.get(_generation_key(product_id)) or "0"
    except redis.RedisError:
        generation = "0"

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

    _store_cached(product_id, generation, product)
    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:
        raw = resp.read()

    # The PUT response is the updated product: publish it to the shared cache
    # and bump the generation so no stale in-flight read can re-cache the old
    # price.  Every process sees the new price on its next get_product().
    try:
        updated = json.loads(raw)
    except ValueError:
        updated = None
    if not isinstance(updated, dict):
        updated = None
    _publish_cached(product_id, updated)
EOF
python3 -c "import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(1)); print(catalog.get_product(1))"

7  $ redis-cli keys 'storefront:*'; echo "TTL: $(redis-cli ttl storefront:product:1)"; echo "log GET 1 count: $(grep -c 'GET 1 ' /var/log/upstream/requests.log)"; redis-cli --scan --pattern '*' | head

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

# cold read then warm read timing
t=time.time(); catalog.get_product(2); cold=time.time()-t
t=time.time(); p=catalog.get_product(2); warm=time.time()-t
print(f"cold={cold*1000:.0f}ms warm={warm*1000:.1f}ms price={p['price']}")

# cross-process: update price in a child process, read in this one
subprocess.run([sys.executable, "-c",
  "import sys; sys.path.insert(0,'/app'); import catalog; catalog.update_price(2, 99.99)"], check=True)
p = catalog.get_product(2)
print("after update in other process:", p['price'])
assert p['price'] == 99.99, p

# reverse: update here, read in another process
catalog.update_price(3, 12.34)
out = subprocess.run([sys.executable, "-c",
  "import sys; sys.path.insert(0,'/app'); import catalog; print(catalog.get_product(3)['price'])"],
  capture_output=True, text=True, check=True)
print("read in other process:", out.stdout.strip())
assert out.stdout.strip() == '12.34'
print("OK")
EOF

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

# warm product 4
catalog.get_product(4)
before = sum(1 for l in open('/var/log/upstream/requests.log') if l.startswith('GET 4 '))
# simulate expiry: shorten TTL
subprocess.run(['redis-cli', 'expire', 'storefront:product:4', '2'], check=True, capture_output=True)
time.sleep(2.5)
t = time.time(); p = catalog.get_product(4); dt = time.time()-t
after = sum(1 for l in open('/var/log/upstream/requests.log') if l.startswith('GET 4 '))
print(f"after expiry: {dt*1000:.0f}ms upstream GETs {before}->{after} price={p['price']}")
assert dt > 0.25 and after == before + 1
print("expiry OK")
EOF

10  $ cd /app && python3 - <<'EOF'
import sys
sys.path.insert(0, '/app')
import catalog, redis
# simulate Redis being unreachable: point the module's client at a dead port
catalog._redis = redis.Redis(host='127.0.0.1', port=6399, socket_connect_timeout=1, decode_responses=True)
p = catalog.get_product(5)
print("degraded mode price:", p['price'])
assert p['price'] == 351.68
catalog.update_price(5, 200.0)
p = catalog.get_product(5)
print("degraded after update:", p['price'])
assert p['price'] == 200.0
print("Redis-failure fallback OK")
EOF
# restore price 5 to original
python3 -c "
import sys; sys.path.insert(0,'/app'); import catalog; catalog.update_price(5, 351.68)"

11  $ cd /app && redis-cli flushdb >/dev/null && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && wc -l < /var/log/upstream/requests.log && python3 - <<'EOF'
import sys, time, subprocess, os
procs = 8
calls = 500
code = f'''
import sys; sys.path.insert(0, "/app")
import catalog
for i in range({calls}):
    catalog.get_product((i % 20) + 1)
'''
t = time.time()
ps = [subprocess.Popen([sys.executable, "-c", code]) for _ in range(procs)]
for p in ps: p.wait()
dt = time.time() - t
gets = sum(1 for l in open('/var/log/upstream/requests.log') if l.startswith('GET '))
print(f"{procs} procs x {calls} calls in {dt:.1f}s; upstream GETs = {gets}")
assert gets <= 20, gets
print("load OK: all other reads served from Redis")
EOF

12  $ cd /app && python3 - <<'EOF'
# race test: update_price while a fetch is in flight must not leave stale price cached
import sys, threading, time, subprocess
sys.path.insert(0, '/app')
import catalog

catalog.get_product(6)  # warm
subprocess.run(['redis-cli','del','storefront:product:6'], check=True, capture_output=True)  # force a miss

results = {}
def reader():
    results['read'] = catalog.get_product(6)
t = threading.Thread(target=reader)
t.start()
time.sleep(0.1)          # reader is mid-fetch (~300ms)
catalog.update_price(6, 42.42)
t.join()
print("in-flight read returned:", results['read']['price'])
p = catalog.get_product(6)
print("next read (any process):", p['price'])
assert p['price'] == 42.42, p
subprocess.run([sys.executable,'-c','import sys;sys.path.insert(0,"/app");import catalog;print(catalog.get_product(6)["price"])'], check=True)
print("race OK")
EOF
python3 -c "import sys;sys.path.insert(0,'/app');import catalog;catalog.update_price(6,455.75)"

13  $ cd /app && curl -s -X POST http://127.0.0.1:8000/__reset >/dev/null && python3 - <<'EOF'
import sys, time, subprocess
# warm the cache sequentially
sys.path.insert(0, '/app')
import catalog
for i in range(1, 21):
    catalog.get_product(i)
warm = sum(1 for l in open('/var/log/upstream/requests.log') if l.startswith('GET '))
procs, calls = 8, 500
code = f'''
import sys; sys.path.insert(0, "/app")
import catalog, time
t = time.time()
for i in range({calls}):
    catalog.get_product((i % 20) + 1)
print(f"{{time.time()-t:.2f}}")
'''
t = time.time()
ps = [subprocess.Popen([sys.executable, "-c", code], stdout=subprocess.PIPE, text=True) for _ in range(procs)]
times = [float(p.communicate()[0]) for p in ps]
dt = time.time() - t
gets = sum(1 for l in open('/var/log/upstream/requests.log') if l.startswith('GET '))
print(f"upstream GETs after warmup: {warm} -> {gets} (delta {gets-warm})")
print(f"wall {dt:.1f}s; per-proc times: {[f'{x:.2f}' for x in times]}")
assert gets == warm, "unexpected upstream calls after warmup"
print(f"~{procs*calls*60/dt:.0f} reads/min served with 0 upstream calls")
EOF

14  $ cd /app && redis-cli flushdb && curl -s -X POST http://127.0.0.1:8000/__reset && python3 - <<'EOF'
import sys, time, subprocess
sys.path.insert(0, '/app')
import catalog

def upstream_gets(pid): 
    return sum(1 for l in open('/var/log/upstream/requests.log') if l.startswith(f'GET {pid} '))

# 1) cold miss -> upstream
t=time.time(); p1 = catalog.get_product(7); cold=time.time()-t
# 2) warm hit -> Redis, shared
t=time.time(); p2 = catalog.get_product(7); warm=time.time()-t
assert p1 == p2 and cold > 0.25 and warm < 0.05 and upstream_gets(7) == 1
print(f"1. cold {cold*1000:.0f}ms (upstream), warm {warm*1000:.2f}ms (Redis), GETs={upstream_gets(7)}")

# 3) TTL = 5 min
ttl = int(subprocess.run(['redis-cli','ttl','storefront:product:7'],capture_output=True,text=True).stdout)
assert 295 <= ttl <= 300, ttl
print(f"2. TTL={ttl}s (expires after 5 minutes)")

# 4) update_price -> next read returns new price in another process
catalog.update_price(7, 111.11)
out = subprocess.run([sys.executable,'-c','import sys;sys.path.insert(0,"/app");import catalog;print(catalog.get_product(7)["price"])'],
                     capture_output=True,text=True)
assert out.stdout.strip() == '111.11', out
print(f"3. cross-process read after update_price: {out.stdout.strip()} (upstream GETs still {upstream_gets(7)})")

# restore
catalog.update_price(7, 327.12)
print("ALL CHECKS PASSED")
EOF
python3 -m py_compile /app/catalog.py && echo "syntax OK" && redis-cli dbsize

{
 "inventory_price_after_update": 123.45,
 "keys": [
  [
   0,
   "storefront: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": "8.10.2",
  "server_name": null,
  "valkey_version": null
 }
}