fix: cross-link tag manifests to digest keys and add fetch lock to prevent thundering herd

Tag manifests (e.g. library/nginx/manifests/latest) and their sha256-addressed
counterparts were stored at separate S3 keys with no cross-reference, so a
sha256 manifest request always missed cache even when the identical content had
just been stored under the tag key.

After serving any mutable (tag) manifest, compute the sha256 of the response
body and write it under the digest key (manifests/sha256:<hex>) if absent. The
next sha256-addressed pull hits cache immediately.

Also adds a short-lived Redis distributed lock (SET NX EX 30) around upstream
fetches so that concurrent pods racing for the same cold key poll storage for
up to 5 s before issuing a duplicate upstream request, eliminating the
thundering herd on deploy events.

Includes unit tests for both the lock primitives (acquire/release, fail-open
when Redis is unavailable) and the docker proxy behaviour (cross-link written
on tag hit, not written for sha256 requests, lock acquired/released, poll path
serves from cache without upstream fetch, fallback fetch when poll times out).

README.md

@@ -390,7 +390,7 @@ If a member is unreachable and has no cached index, it is skipped and a warning
 
 **Caching:**
 
-The merged index is cached using `min(mutable_ttl)` across all members. Each member's raw index is cached in S3 under its own remote key by the normal proxy rules; the virtual handler reuses those copies when available.
+The merged index is cached using `min(mutable_ttl)` across all members. Each member's raw index is cached in S3 under its own remote key; the virtual handler reuses those copies when available. On rebuild, each member's parsed index is also stored as a compact msgpack file (`index.msgpack`) alongside the raw YAML, eliminating the YAML parse cost on subsequent rebuilds.
 
 **Helm example:**
 

examples/single-file/remotes.yaml

@@ -34,7 +34,8 @@
 #
 remotes:
   github:
-    base_url: "https://github.com"    package: "generic"
+    base_url: "https://github.com"
+    package: "generic"
     description: "GitHub releases and files"
     immutable_patterns:
       - "gruntwork-io/terragrunt/.*terragrunt_linux_amd64.*"
@@ -64,7 +65,8 @@ remotes:
       mutable_ttl: 0
 
   github-archive:
-    base_url: "https://github.com"    package: "generic"
+    base_url: "https://github.com"
+    package: "generic"
     description: "GitHub repository archive tarballs"
     immutable_patterns:
       # Tag archives are immutable — a tag never changes
@@ -82,7 +84,8 @@ remotes:
       mutable_ttl: 86400     # Branch archives refreshed after 1 day
 
   gitea-dl:
-    base_url: "https://dl.gitea.com"    package: "generic"
+    base_url: "https://dl.gitea.com"
+    package: "generic"
     description: "Gitea download site"
     immutable_patterns:
       - "act_runner/.*/act_runner-.*-linux-amd64$"
@@ -91,7 +94,8 @@ remotes:
       mutable_ttl: 0
 
   hashicorp-releases:
-    base_url: "https://releases.hashicorp.com"    package: "generic"
+    base_url: "https://releases.hashicorp.com"
+    package: "generic"
     description: "HashiCorp product releases"
     immutable_patterns:
       - "terraform/.*terraform_.*_linux_amd64\\.zip$"
@@ -110,7 +114,8 @@ remotes:
       mutable_ttl: 0
 
   alpine:
-    base_url: "https://dl-cdn.alpinelinux.org"    package: "alpine"
+    base_url: "https://dl-cdn.alpinelinux.org"
+    package: "alpine"
     description: "Alpine Linux APK package repository"
     immutable_patterns:
       - ".*/x86_64/.*\\.apk$"
@@ -122,7 +127,8 @@ remotes:
       mutable_ttl: 7200  # Index files (APKINDEX.tar.gz) cached for 2 hours
 
   almalinux:
-    base_url: "https://gsl-syd.mm.fcix.net/almalinux"    package: "rpm"
+    base_url: "https://gsl-syd.mm.fcix.net/almalinux"
+    package: "rpm"
     description: "AlmaLinux RPM package repository"
     immutable_patterns:
       - ".*/x86_64/.*\\.rpm$"
@@ -137,7 +143,8 @@ remotes:
       mutable_ttl: 7200  # Metadata files cached for 2 hours
 
   epel:
-    base_url: "http://mirror.aarnet.edu.au/pub/epel"    package: "rpm"
+    base_url: "http://mirror.aarnet.edu.au/pub/epel"
+    package: "rpm"
     description: "EPEL (Extra Packages for Enterprise Linux)"
     immutable_patterns:
       - "8/Everything/x86_64/.*\\.rpm$"
@@ -150,7 +157,8 @@ remotes:
       mutable_ttl: 7200  # Metadata files cached for 2 hours
 
   fedora:
-    base_url: "https://gsl-syd.mm.fcix.net/fedora/linux"    package: "rpm"
+    base_url: "https://gsl-syd.mm.fcix.net/fedora/linux"
+    package: "rpm"
     description: "Fedora Linux RPM package repository"
     immutable_patterns:
       - "releases/.*/Everything/x86_64/.*\\.rpm$"
@@ -163,7 +171,8 @@ remotes:
       mutable_ttl: 300  # Metadata files cached for 5 minutes
 
   ghcr:
-    base_url: "https://ghcr.io"    package: "docker"
+    base_url: "https://ghcr.io"
+    package: "docker"
     description: "GitHub Container Registry"
     # username: "your-github-username"
     # password: "your-github-pat"  # needs read:packages scope
@@ -175,14 +184,16 @@ remotes:
       mutable_ttl: 300
 
   dockerhub:
-    base_url: "https://registry-1.docker.io"    package: "docker"
+    base_url: "https://registry-1.docker.io"
+    package: "docker"
     description: "Docker Hub registry"
     cache:
       immutable_ttl: 0
       mutable_ttl: 300
 
   pypi:
-    base_url: "https://files.pythonhosted.org"    package: "pypi"
+    base_url: "https://files.pythonhosted.org"
+    package: "pypi"
     description: "Python Package Index — simple index and package files via a single remote"
     # simple/ requests are transparently fetched from pypi.org; package files come from
     # files.pythonhosted.org (base_url). URLs in the simple index are rewritten to this remote.
@@ -203,7 +214,8 @@ remotes:
       mutable_ttl: 600  # Simple index pages refreshed after 10 minutes
 
   pypi-gitea:
-    base_url: "https://gitea.example.com/api/packages/myorg/pypi"    package: "pypi"
+    base_url: "https://gitea.example.com/api/packages/myorg/pypi"
+    package: "pypi"
     description: "Private Gitea PyPI registry — simple index and files at the same host"
     # username: "your-gitea-username"
     # password: "your-personal-access-token"  # needs package:read scope
@@ -219,7 +231,8 @@ remotes:
       mutable_ttl: 600
 
   npm:
-    base_url: "https://registry.npmjs.org"    package: "npm"
+    base_url: "https://registry.npmjs.org"
+    package: "npm"
     description: "npm registry — package metadata with tarball URL rewriting"
     check_mutable_updates: true
     immutable_patterns:
@@ -231,7 +244,8 @@ remotes:
       mutable_ttl: 600  # Package metadata refreshed after 10 minutes
 
   hashicorp-helm:
-    base_url: "https://helm.releases.hashicorp.com"    package: "helm"
+    base_url: "https://helm.releases.hashicorp.com"
+    package: "helm"
     description: "HashiCorp Helm chart repository (Vault, Consul, Nomad, etc.)"
     check_mutable_updates: true
     immutable_patterns:
@@ -241,7 +255,8 @@ remotes:
       mutable_ttl: 3600   # index.yaml refreshed after 1 hour
 
   metallb:
-    base_url: "https://metallb.github.io/metallb"    package: "helm"
+    base_url: "https://metallb.github.io/metallb"
+    package: "helm"
     description: "MetalLB load balancer Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -251,7 +266,8 @@ remotes:
       mutable_ttl: 3600
 
   jetstack:
-    base_url: "https://charts.jetstack.io"    package: "helm"
+    base_url: "https://charts.jetstack.io"
+    package: "helm"
     description: "Jetstack Helm charts (cert-manager)"
     check_mutable_updates: true
     immutable_patterns:
@@ -261,7 +277,8 @@ remotes:
       mutable_ttl: 3600
 
   rancher-stable:
-    base_url: "https://releases.rancher.com/server-charts/stable"    package: "helm"
+    base_url: "https://releases.rancher.com/server-charts/stable"
+    package: "helm"
     description: "Rancher stable Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -271,7 +288,8 @@ remotes:
       mutable_ttl: 3600
 
   purelb:
-    base_url: "https://gitlab.com/api/v4/projects/20400619/packages/helm/stable"    package: "helm"
+    base_url: "https://gitlab.com/api/v4/projects/20400619/packages/helm/stable"
+    package: "helm"
     description: "PureLB load balancer Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -281,7 +299,8 @@ remotes:
       mutable_ttl: 3600
 
   istio:
-    base_url: "https://istio-release.storage.googleapis.com/charts"    package: "helm"
+    base_url: "https://istio-release.storage.googleapis.com/charts"
+    package: "helm"
     description: "Istio service mesh Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -291,7 +310,8 @@ remotes:
       mutable_ttl: 3600
 
   cnpg:
-    base_url: "https://cloudnative-pg.github.io/charts"    package: "helm"
+    base_url: "https://cloudnative-pg.github.io/charts"
+    package: "helm"
     description: "CloudNativePG operator Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -301,7 +321,8 @@ remotes:
       mutable_ttl: 3600
 
   ceph-csi:
-    base_url: "https://ceph.github.io/csi-charts"    package: "helm"
+    base_url: "https://ceph.github.io/csi-charts"
+    package: "helm"
     description: "Ceph CSI driver Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -311,7 +332,8 @@ remotes:
       mutable_ttl: 3600
 
   external-dns:
-    base_url: "https://kubernetes-sigs.github.io/external-dns/"    package: "helm"
+    base_url: "https://kubernetes-sigs.github.io/external-dns/"
+    package: "helm"
     description: "ExternalDNS Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -321,7 +343,8 @@ remotes:
       mutable_ttl: 3600
 
   intel-helm:
-    base_url: "https://intel.github.io/helm-charts/"    package: "helm"
+    base_url: "https://intel.github.io/helm-charts/"
+    package: "helm"
     description: "Intel Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -331,7 +354,8 @@ remotes:
       mutable_ttl: 3600
 
   elastic:
-    base_url: "https://helm.elastic.co"    package: "helm"
+    base_url: "https://helm.elastic.co"
+    package: "helm"
     description: "Elastic stack Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -341,7 +365,8 @@ remotes:
       mutable_ttl: 3600
 
   k8up-io:
-    base_url: "https://k8up-io.github.io/k8up"    package: "helm"
+    base_url: "https://k8up-io.github.io/k8up"
+    package: "helm"
     description: "K8up backup operator Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -351,7 +376,8 @@ remotes:
       mutable_ttl: 3600
 
   victoriametrics:
-    base_url: "https://victoriametrics.github.io/helm-charts/"    package: "helm"
+    base_url: "https://victoriametrics.github.io/helm-charts/"
+    package: "helm"
     description: "VictoriaMetrics observability Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -361,7 +387,8 @@ remotes:
       mutable_ttl: 3600
 
   grafana:
-    base_url: "https://grafana.github.io/helm-charts"    package: "helm"
+    base_url: "https://grafana.github.io/helm-charts"
+    package: "helm"
     description: "Grafana observability Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -371,7 +398,8 @@ remotes:
       mutable_ttl: 3600
 
   helm-openldap:
-    base_url: "https://jp-gouin.github.io/helm-openldap/"    package: "helm"
+    base_url: "https://jp-gouin.github.io/helm-openldap/"
+    package: "helm"
     description: "OpenLDAP Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -381,7 +409,8 @@ remotes:
       mutable_ttl: 3600
 
   woodpecker:
-    base_url: "https://woodpecker-ci.org/"    package: "helm"
+    base_url: "https://woodpecker-ci.org/"
+    package: "helm"
     description: "Woodpecker CI Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -391,7 +420,8 @@ remotes:
       mutable_ttl: 3600
 
   stakater:
-    base_url: "https://stakater.github.io/stakater-charts"    package: "helm"
+    base_url: "https://stakater.github.io/stakater-charts"
+    package: "helm"
     description: "Stakater Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -401,7 +431,8 @@ remotes:
       mutable_ttl: 3600
 
   jfrog:
-    base_url: "https://charts.jfrog.io/"    package: "helm"
+    base_url: "https://charts.jfrog.io/"
+    package: "helm"
     description: "JFrog Helm charts"
     check_mutable_updates: true
     immutable_patterns:
@@ -411,7 +442,8 @@ remotes:
       mutable_ttl: 3600
 
   openvox:
-    base_url: "https://openvoxproject.github.io/openvox-helm-chart"    package: "helm"
+    base_url: "https://openvoxproject.github.io/openvox-helm-chart"
+    package: "helm"
     description: "OpenVox Helm charts"
     check_mutable_updates: true
     immutable_patterns:

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "lxml>=4.9.0",
     "prometheus-client>=0.19.0",
     "python-multipart>=0.0.6",
+    "msgpack>=1.0.0",
 ]
 requires-python = ">=3.11"
 readme = "README.md"

src/artifactapi/artifact/docker.py

@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import json
 import logging
@@ -47,23 +48,39 @@ async def proxy(request: Request, remote_name: str, path: str, storage, cache, c
             if not await _proxy.handle_expired_mutable(remote_name, path, remote_url, config, cache, storage):
                 cached_key = None
 
+    lock_acquired = False
+    if not cached_key:
+        lock_acquired = cache.acquire_fetch_lock(remote_name, path)
+        if not lock_acquired:
+            # Another pod is already fetching — poll storage briefly before issuing a duplicate upstream request
+            for _ in range(10):
+                await asyncio.sleep(0.5)
+                probe_key = storage.get_object_key(remote_name, path)
+                if storage.exists(probe_key):
+                    cached_key = probe_key
+                    break
+
     if not cached_key:
         logger.info(f"Cache MISS: {remote_name}/{path} - fetching from remote: {remote_url}")
-        result = await _proxy.cache_single_artifact(remote_url, remote_name, path, storage, remote_config)
+        try:
-        if result["status"] == "error":
+            result = await _proxy.cache_single_artifact(remote_url, remote_name, path, storage, remote_config)
-            raise HTTPException(status_code=502, detail=f"Failed to fetch: {result['error']}")
+            if result["status"] == "error":
-        if result["status"] == "cached" and is_mutable:
+                raise HTTPException(status_code=502, detail=f"Failed to fetch: {result['error']}")
-            cache_config = config.get_cache_config(remote_name)
+            if result["status"] == "cached" and is_mutable:
-            mutable_ttl = cache_config.get("mutable_ttl", 3600)
+                cache_config = config.get_cache_config(remote_name)
-            cache.mark_index_cached(remote_name, path, mutable_ttl)
+                mutable_ttl = cache_config.get("mutable_ttl", 3600)
-            logger.info(f"Mutable file cached with TTL: {remote_name}/{path} (ttl: {mutable_ttl}s)")
+                cache.mark_index_cached(remote_name, path, mutable_ttl)
-            if result.get("etag") or result.get("last_modified"):
+                logger.info(f"Mutable file cached with TTL: {remote_name}/{path} (ttl: {mutable_ttl}s)")
-                cache.store_mutable_meta(remote_name, path, result.get("etag"), result.get("last_modified"))
+                if result.get("etag") or result.get("last_modified"):
-        if not is_mutable:
+                    cache.store_mutable_meta(remote_name, path, result.get("etag"), result.get("last_modified"))
-            published = result.get("last_modified")
+            if not is_mutable:
-            if published:
+                published = result.get("last_modified")
-                cache.store_artifact_published(remote_name, path, published)
+                if published:
-            _proxy._check_quarantine(remote_name, published, config)
+                    cache.store_artifact_published(remote_name, path, published)
+                _proxy._check_quarantine(remote_name, published, config)
+        finally:
+            if lock_acquired:
+                cache.release_fetch_lock(remote_name, path)
     elif not is_mutable:
         published = cache.get_artifact_published(remote_name, path)
         if not published:
@@ -90,6 +107,14 @@ async def proxy(request: Request, remote_name: str, path: str, storage, cache, c
             content_type = "application/vnd.oci.image.manifest.v1+json"
 
     digest = f"sha256:{hashlib.sha256(artifact_data).hexdigest()}"
+
+    # Cross-link tag manifests to their sha256 digest key so digest-addressed pulls hit cache
+    if is_mutable and "/manifests/" in path:
+        digest_path = re.sub(r"/manifests/[^/]+$", f"/manifests/{digest}", path)
+        digest_key = storage.get_object_key(remote_name, digest_path)
+        if not storage.exists(digest_key):
+            storage.upload(digest_key, artifact_data)
+
     headers = {
         "Docker-Distribution-Api-Version": "registry/2.0",
         "Docker-Content-Digest": digest,

src/artifactapi/artifact/virtual.py

@@ -6,15 +6,21 @@ from datetime import UTC, date, datetime
 from typing import Protocol, runtime_checkable
 
 import httpx
+import msgpack as _msgpack
 import yaml
 from fastapi import HTTPException, Request, Response
 
-from ..remote import helm as _helm
-
 logger = logging.getLogger(__name__)
 
+try:
+    _YamlLoader = yaml.CSafeLoader
+    _YamlDumperBase = yaml.CDumper
+except AttributeError:
+    _YamlLoader = yaml.SafeLoader
+    _YamlDumperBase = yaml.Dumper
 
-class _HelmDumper(yaml.Dumper):
+
+class _HelmDumper(_YamlDumperBase):
     """YAML dumper that serializes datetime/date objects back to ISO 8601 strings.
 
     yaml.safe_load converts timestamp-shaped YAML scalars (e.g. chart `created`
@@ -37,21 +43,43 @@ _HelmDumper.add_representer(datetime, _repr_datetime)
 _HelmDumper.add_representer(date, _repr_date)
 
 
+def _entries_to_msgpack_safe(entries: dict) -> dict:
+    """Convert datetime/date values to ISO strings for msgpack serialization."""
+    result = {}
+    for chart, versions in entries.items():
+        safe_versions = []
+        for v in versions:
+            safe_v = {}
+            for k, val in v.items():
+                if isinstance(val, datetime):
+                    safe_v[k] = val.isoformat()
+                elif isinstance(val, date):
+                    safe_v[k] = val.isoformat()
+                else:
+                    safe_v[k] = val
+            safe_versions.append(safe_v)
+        result[chart] = safe_versions
+    return result
+
+
 async def _get_member_index(
     member_name: str,
     member_cfg: dict,
     path: str,
     storage,
     cache,
-) -> tuple[str, dict, int, bytes | None]:
+) -> tuple[str, dict, int, bytes | None, dict | None]:
     """Fetch or retrieve cached index.yaml for one member remote.
 
-    Returns (member_name, member_cfg, ttl, raw_bytes).
+    Returns (member_name, member_cfg, ttl, raw_bytes, parsed_entries).
     raw_bytes is None if the member is unreachable and not in S3.
+    parsed_entries is the pre-parsed entries dict (from msgpack cache), or None.
     """
     member_ttl = member_cfg.get("cache", {}).get("mutable_ttl", 3600)
     s3_key = storage.get_object_key(member_name, path)
+    msgpack_key = storage.get_object_key(member_name, "index.msgpack")
     raw_data: bytes | None = None
+    parsed_entries: dict | None = None
 
     if storage.exists(s3_key) and cache.is_index_valid(member_name, path):
         try:
@@ -59,6 +87,13 @@ async def _get_member_index(
             logger.info(f"Virtual: cache hit for member '{member_name}'")
         except Exception:
             raw_data = None
+        if raw_data is not None and storage.exists(msgpack_key):
+            try:
+                packed = storage.download_object(msgpack_key)
+                parsed_entries = _msgpack.unpackb(packed, raw=False)
+                logger.debug(f"Virtual: msgpack hit for member '{member_name}'")
+            except Exception:
+                parsed_entries = None
 
     if raw_data is None:
         base_url = member_cfg.get("base_url", "").rstrip("/")
@@ -76,35 +111,74 @@ async def _get_member_index(
                 raw_data = response.content
         except Exception as e:
             logger.warning(f"Virtual: failed to fetch index.yaml from member '{member_name}': {e}")
-            return member_name, member_cfg, member_ttl, None
+            return member_name, member_cfg, member_ttl, None, None
         try:
             storage.upload(s3_key, raw_data)
             cache.mark_index_cached(member_name, path, member_ttl)
         except Exception as e:
             logger.warning(f"Virtual: failed to cache index.yaml for member '{member_name}': {e}")
 
-    return member_name, member_cfg, member_ttl, raw_data
+    if parsed_entries is None and raw_data is not None:
+        try:
+            index = yaml.load(raw_data, Loader=_YamlLoader)
+            safe_entries = _entries_to_msgpack_safe(index.get("entries") or {})
+            storage.upload(msgpack_key, _msgpack.packb(safe_entries, use_bin_type=True))
+            parsed_entries = safe_entries
+        except Exception as e:
+            logger.warning(f"Virtual: failed to build msgpack cache for '{member_name}': {e}")
+
+    return member_name, member_cfg, member_ttl, raw_data, parsed_entries
 
 
-def _merge_helm_indexes(raw_indexes: list[bytes], member_names: list[str], member_configs: list[dict], proxy_base: str) -> bytes:
+def _rewrite_urls(urls: list, base_url: str, proxy_base: str, member_name: str) -> list:
+    proxy_remote = f"{proxy_base}/api/v1/remote/{member_name}"
+    rewritten = []
+    for url in urls:
+        if url.startswith(("http://", "https://")):
+            if base_url and url.startswith(base_url):
+                url = proxy_remote + url[len(base_url) :]
+        else:
+            url = f"{proxy_remote}/{url.lstrip('/')}"
+        rewritten.append(url)
+    return rewritten
+
+
+def _merge_helm_indexes(
+    raw_indexes: list[bytes],
+    parsed_entries_list: list[dict | None],
+    member_names: list[str],
+    member_configs: list[dict],
+    proxy_base: str,
+) -> bytes:
     """Merge helm index.yaml files with per-member URL rewriting.
 
     Priority is determined by position in member_names: earlier members win
     when the same chart name + version appears in multiple remotes.
+    Uses pre-parsed msgpack entries when available to skip YAML parsing.
     """
     merged_entries: dict[str, list] = {}
 
-    for raw_data, member_name, member_cfg in zip(raw_indexes, member_names, member_configs):
+    for raw_data, pre_parsed, member_name, member_cfg in zip(raw_indexes, parsed_entries_list, member_names, member_configs):
         base_url = member_cfg.get("base_url", "").rstrip("/")
-        rewritten, _ = _helm.resolve_content(raw_data, "index.yaml", "index.yaml", base_url, proxy_base, member_name)
 
-        try:
+        if pre_parsed is not None:
-            index = yaml.safe_load(rewritten)
+            entries = pre_parsed
-        except Exception as e:
+        else:
-            logger.warning(f"Virtual: failed to parse index.yaml from member '{member_name}': {e}")
+            try:
-            continue
+                index = yaml.load(raw_data, Loader=_YamlLoader)
+            except Exception as e:
+                logger.warning(f"Virtual: failed to parse index.yaml from member '{member_name}': {e}")
+                continue
+            entries = index.get("entries") or {}
 
-        for chart_name, versions in (index.get("entries") or {}).items():
+        for chart_name, versions in entries.items():
+            for version_entry in versions:
+                version_entry["urls"] = _rewrite_urls(
+                    version_entry.get("urls") or [],
+                    base_url,
+                    proxy_base,
+                    member_name,
+                )
             if chart_name not in merged_entries:
                 merged_entries[chart_name] = list(versions)
             else:
@@ -126,7 +200,14 @@ def _merge_helm_indexes(raw_indexes: list[bytes], member_names: list[str], membe
 @runtime_checkable
 class _VirtualHandler(Protocol):
     def accepts_path(self, path: str) -> bool: ...
-    def merge(self, raw_indexes: list[bytes], member_names: list[str], member_configs: list[dict], proxy_base: str) -> bytes: ...
+    def merge(
+        self,
+        raw_indexes: list[bytes],
+        parsed_entries: list[dict | None],
+        member_names: list[str],
+        member_configs: list[dict],
+        proxy_base: str,
+    ) -> bytes: ...
     def path_error(self) -> str: ...
 
 
@@ -134,8 +215,15 @@ class _HelmHandler:
     def accepts_path(self, path: str) -> bool:
         return path == "index.yaml"
 
-    def merge(self, raw_indexes: list[bytes], member_names: list[str], member_configs: list[dict], proxy_base: str) -> bytes:
+    def merge(
-        return _merge_helm_indexes(raw_indexes, member_names, member_configs, proxy_base)
+        self,
+        raw_indexes: list[bytes],
+        parsed_entries: list[dict | None],
+        member_names: list[str],
+        member_configs: list[dict],
+        proxy_base: str,
+    ) -> bytes:
+        return _merge_helm_indexes(raw_indexes, parsed_entries, member_names, member_configs, proxy_base)
 
     def path_error(self) -> str:
         return "Virtual helm repositories only serve index.yaml; chart tarballs are served directly by member remotes"
@@ -186,17 +274,19 @@ async def handle(request: Request, virtual_name: str, path: str, storage, cache,
     fetch_ms = int((time.perf_counter() - t_fetch) * 1000)
 
     raw_indexes: list[bytes] = []
+    used_parsed: list[dict | None] = []
     used_members: list[str] = []
     used_configs: list[dict] = []
     min_ttl: int | None = None
 
-    for member_name, member_cfg, member_ttl, raw_data in results:
+    for member_name, member_cfg, member_ttl, raw_data, parsed_entries in results:
         if min_ttl is None or member_ttl < min_ttl:
             min_ttl = member_ttl
         if raw_data is None:
             logger.warning(f"Virtual '{virtual_name}': skipping unreachable member '{member_name}'")
             continue
         raw_indexes.append(raw_data)
+        used_parsed.append(parsed_entries)
         used_members.append(member_name)
         used_configs.append(member_cfg)
 
@@ -207,7 +297,7 @@ async def handle(request: Request, virtual_name: str, path: str, storage, cache,
         min_ttl = 3600
 
     t_merge = time.perf_counter()
-    merged = handler.merge(raw_indexes, used_members, used_configs, proxy_base)
+    merged = await asyncio.to_thread(handler.merge, raw_indexes, used_parsed, used_members, used_configs, proxy_base)
     merge_ms = int((time.perf_counter() - t_merge) * 1000)
 
     try:
@@ -215,9 +305,11 @@ async def handle(request: Request, virtual_name: str, path: str, storage, cache,
         storage.upload(virtual_key, merged)
         cache.mark_index_cached(virtual_name, path, min_ttl)
         store_ms = int((time.perf_counter() - t_store) * 1000)
+        msgpack_hits = sum(1 for p in used_parsed if p is not None)
         logger.info(
             f"Virtual MISS: {virtual_name}/{path} rebuilt from {used_members} "
-            f"(fetch={fetch_ms}ms merge={merge_ms}ms store={store_ms}ms ttl={min_ttl}s)"
+            f"(fetch={fetch_ms}ms merge={merge_ms}ms store={store_ms}ms ttl={min_ttl}s "
+            f"msgpack={msgpack_hits}/{len(used_members)})"
         )
     except Exception as e:
         logger.warning(f"Virtual: failed to store merged index for '{virtual_name}': {e}")

src/artifactapi/cache/redis.py

@@ -99,6 +99,25 @@ class RedisCache:
         except Exception:
             return None
 
+    def acquire_fetch_lock(self, remote_name: str, path: str, ttl: int = 30) -> bool:
+        """Try to acquire a short-lived fetch lock. Returns True if acquired, False if held by another caller."""
+        if not self.available:
+            return True  # fail open: no Redis → behave as if we always hold the lock
+        key = f"fetchlock:{remote_name}:{hashlib.sha256(path.encode()).hexdigest()[:16]}"
+        try:
+            return bool(self.client.set(key, 1, nx=True, ex=ttl))
+        except Exception:
+            return True
+
+    def release_fetch_lock(self, remote_name: str, path: str) -> None:
+        if not self.available:
+            return
+        key = f"fetchlock:{remote_name}:{hashlib.sha256(path.encode()).hexdigest()[:16]}"
+        try:
+            self.client.delete(key)
+        except Exception:
+            pass
+
     def cleanup_expired_index(self, storage, remote_name: str, path: str) -> None:
         if not self.available:
             return

tests/test_cache.py

@@ -327,3 +327,74 @@ class TestArtifactPublished:
 
     def test_get_returns_none_when_unavailable(self, unavailable_cache):
         assert unavailable_cache.get_artifact_published("remote", "path") is None
+
+
+# ---------------------------------------------------------------------------
+# fetch lock (thundering-herd deduplication)
+# ---------------------------------------------------------------------------
+
+
+class TestFetchLock:
+    def test_acquire_returns_true_when_lock_obtained(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        result = cache_with_redis.acquire_fetch_lock("myremote", "library/nginx/manifests/latest")
+        assert result is True
+
+    def test_acquire_calls_set_nx_with_ttl(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        cache_with_redis.acquire_fetch_lock("myremote", "library/nginx/manifests/latest", ttl=15)
+        _, kwargs = mock_redis_client.set.call_args
+        assert kwargs.get("nx") is True
+        assert kwargs.get("ex") == 15
+
+    def test_acquire_returns_false_when_lock_already_held(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = None  # Redis SET NX → None when key exists
+        result = cache_with_redis.acquire_fetch_lock("myremote", "library/nginx/manifests/latest")
+        assert result is False
+
+    def test_acquire_fails_open_when_unavailable(self, unavailable_cache):
+        # caller must be allowed to proceed when Redis is down
+        assert unavailable_cache.acquire_fetch_lock("myremote", "some/path") is True
+
+    def test_acquire_fails_open_on_redis_exception(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.side_effect = Exception("connection reset")
+        assert cache_with_redis.acquire_fetch_lock("myremote", "some/path") is True
+
+    def test_lock_key_embeds_path_hash(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        path = "library/nginx/manifests/latest"
+        cache_with_redis.acquire_fetch_lock("myremote", path)
+        args, _ = mock_redis_client.set.call_args
+        expected_hash = hashlib.sha256(path.encode()).hexdigest()[:16]
+        assert args[0] == f"fetchlock:myremote:{expected_hash}"
+
+    def test_lock_key_hash_is_16_chars(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        cache_with_redis.acquire_fetch_lock("myremote", "some/long/path/file.tar.gz")
+        args, _ = mock_redis_client.set.call_args
+        # key format: fetchlock:<remote>:<16-char hash>
+        parts = args[0].split(":")
+        assert len(parts) == 3
+        assert len(parts[2]) == 16
+
+    def test_different_paths_produce_different_lock_keys(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.set.return_value = True
+        cache_with_redis.acquire_fetch_lock("myremote", "path/a/manifests/latest")
+        key_a = mock_redis_client.set.call_args[0][0]
+        mock_redis_client.set.reset_mock()
+        cache_with_redis.acquire_fetch_lock("myremote", "path/b/manifests/latest")
+        key_b = mock_redis_client.set.call_args[0][0]
+        assert key_a != key_b
+
+    def test_release_deletes_correct_key(self, cache_with_redis, mock_redis_client):
+        path = "library/nginx/manifests/latest"
+        cache_with_redis.release_fetch_lock("myremote", path)
+        expected_hash = hashlib.sha256(path.encode()).hexdigest()[:16]
+        mock_redis_client.delete.assert_called_once_with(f"fetchlock:myremote:{expected_hash}")
+
+    def test_release_no_op_when_unavailable(self, unavailable_cache):
+        unavailable_cache.release_fetch_lock("myremote", "some/path")  # must not raise
+
+    def test_release_no_op_on_redis_exception(self, cache_with_redis, mock_redis_client):
+        mock_redis_client.delete.side_effect = Exception("timeout")
+        cache_with_redis.release_fetch_lock("myremote", "some/path")  # must not raise

tests/test_routes.py

@@ -260,6 +260,133 @@ class TestDockerProxy:
         mock_fetch.assert_called_once()
         assert response.status_code == 200
 
+    # --- Issue 1: sha256 digest cross-linking ---
+
+    def test_tag_manifest_is_stored_under_digest_key_on_cache_hit(self, client, patched_deps):
+        # When serving a cached tag manifest the handler must also write the content
+        # under the sha256 digest key so subsequent sha256-addressed pulls hit cache.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # First exists call (tag manifest): hit. Second (digest key): miss → triggers upload.
+        deps["storage"].exists.side_effect = [True, False]
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+
+        response = client.get("/v2/docker-test/library/nginx/manifests/v1.25.3")
+
+        assert response.status_code == 200
+        deps["storage"].upload.assert_called_once_with(deps["storage"].get_object_key.return_value, manifest)
+
+    def test_tag_manifest_digest_key_not_written_when_already_exists(self, client, patched_deps):
+        # When the digest key already exists in storage upload must not be called.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # Both the tag key and the digest key already present.
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+
+        client.get("/v2/docker-test/library/nginx/manifests/v1.25.3")
+
+        deps["storage"].upload.assert_not_called()
+
+    def test_sha256_manifest_request_is_not_cross_linked(self, client, patched_deps):
+        # sha256-addressed manifests are immutable — the cross-link logic must not apply.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.return_value = True
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = False  # sha256 manifest is immutable
+
+        with patch("artifactapi.artifact.proxy._fetch_last_modified", new_callable=AsyncMock, return_value=None):
+            client.get("/v2/docker-test/library/nginx/manifests/sha256:" + "a" * 64)
+
+        deps["storage"].upload.assert_not_called()
+
+    # --- Issue 2: thundering herd distributed lock ---
+
+    def test_lock_acquired_and_released_on_upstream_fetch(self, client, patched_deps):
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        deps["storage"].exists.side_effect = [False, False]  # initial miss; digest key also absent
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = True
+
+        with patch(
+            "artifactapi.artifact.proxy.cache_single_artifact",
+            new_callable=AsyncMock,
+            return_value={"status": "cached"},
+        ):
+            response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        deps["cache"].acquire_fetch_lock.assert_called_once()
+        deps["cache"].release_fetch_lock.assert_called_once()
+        assert response.status_code == 200
+
+    def test_lock_released_even_when_fetch_returns_error(self, client, patched_deps):
+        deps = patched_deps
+        deps["storage"].exists.return_value = False
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = True
+
+        with patch(
+            "artifactapi.artifact.proxy.cache_single_artifact",
+            new_callable=AsyncMock,
+            return_value={"status": "error", "error": "upstream down"},
+        ):
+            response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        deps["cache"].release_fetch_lock.assert_called_once()
+        assert response.status_code == 502
+
+    def test_thundering_herd_polls_storage_when_lock_not_acquired(self, client, patched_deps):
+        # When the lock is held by another pod the handler must poll storage and serve
+        # from cache once the competing fetch completes, without issuing its own upstream request.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # Initial cache check: miss. First poll iteration: another pod has written it.
+        # Third call is for the digest cross-link check (is_mutable=True path); digest key exists.
+        deps["storage"].exists.side_effect = [False, True, True]
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].is_index_valid.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = False  # lock held by peer
+
+        with patch("artifactapi.artifact.docker.asyncio.sleep", new_callable=AsyncMock):
+            with patch(
+                "artifactapi.artifact.proxy.cache_single_artifact",
+                new_callable=AsyncMock,
+            ) as mock_fetch:
+                response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        mock_fetch.assert_not_called()
+        assert response.status_code == 200
+
+    def test_thundering_herd_falls_through_to_fetch_if_poll_times_out(self, client, patched_deps):
+        # If the item never appears in storage during the poll window the handler must
+        # still issue its own upstream fetch as a fallback.
+        deps = patched_deps
+        manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
+        # All exists calls return False — item never appears during polling.
+        deps["storage"].exists.return_value = False
+        deps["storage"].download_object.return_value = manifest
+        deps["cache"].is_mutable_file.return_value = True
+        deps["cache"].acquire_fetch_lock.return_value = False  # lock held by peer
+
+        with patch("artifactapi.artifact.docker.asyncio.sleep", new_callable=AsyncMock):
+            with patch(
+                "artifactapi.artifact.proxy.cache_single_artifact",
+                new_callable=AsyncMock,
+                return_value={"status": "cached"},
+            ) as mock_fetch:
+                response = client.get("/v2/docker-test/library/nginx/manifests/latest")
+
+        mock_fetch.assert_called_once()
+        assert response.status_code == 200
+
 
 # ---------------------------------------------------------------------------
 # Generic artifact route  /api/v1/remote/{remote}/{path}

tests/test_virtual.py

@@ -8,11 +8,15 @@ import yaml
 
 from artifactapi.artifact.virtual import (
     _HANDLERS,
+    _entries_to_msgpack_safe,
     _get_member_index,
     _HelmDumper,
     _HelmHandler,
     _merge_helm_indexes,
+    _rewrite_urls,
     _VirtualHandler,
+    _YamlDumperBase,
+    _YamlLoader,
 )
 
 # ---------------------------------------------------------------------------
@@ -66,12 +70,47 @@ entries:
 generated: "2023-01-01T00:00:00.000Z"
 """
 
+_INDEX_RELATIVE = b"""\
+apiVersion: v1
+entries:
+  rancher:
+    - name: rancher
+      version: "2.13.1"
+      urls:
+        - rancher-2.13.1.tgz
+generated: "2023-01-01T00:00:00.000Z"
+"""
+
 _CFG_A = {"base_url": "https://helm.releases.hashicorp.com", "cache": {"mutable_ttl": 3600}}
 _CFG_B = {"base_url": "https://charts.example.com", "cache": {"mutable_ttl": 1800}}
 
 
-def _identity_resolve(data, *args, **kwargs):
+# ---------------------------------------------------------------------------
-    return data, None
+# _YamlLoader / _YamlDumperBase — C extension selection
+# ---------------------------------------------------------------------------
+
+
+class TestYamlExtensionSelection:
+    def test_loader_is_a_class(self):
+        assert isinstance(_YamlLoader, type)
+
+    def test_dumper_base_is_a_class(self):
+        assert isinstance(_YamlDumperBase, type)
+
+    def test_helm_dumper_uses_selected_base(self):
+        assert issubclass(_HelmDumper, _YamlDumperBase)
+
+    def test_c_extensions_used_when_available(self):
+        try:
+            assert _YamlLoader is yaml.CSafeLoader
+            assert _YamlDumperBase is yaml.CDumper
+        except AttributeError:
+            assert _YamlLoader is yaml.SafeLoader
+            assert _YamlDumperBase is yaml.Dumper
+
+    def test_loader_can_parse_yaml(self):
+        result = yaml.load(b"key: value", Loader=_YamlLoader)
+        assert result == {"key": "value"}
 
 
 # ---------------------------------------------------------------------------
@@ -135,14 +174,13 @@ class TestHelmHandler:
         assert isinstance(msg, str) and len(msg) > 0
 
     def test_merge_returns_bytes(self):
-        with patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve):
+        result = self.handler.merge([_INDEX_A], [None], ["member-a"], [_CFG_A], "http://proxy.example.com")
-            result = self.handler.merge([_INDEX_A], ["member-a"], [_CFG_A], "http://proxy.example.com")
         assert isinstance(result, bytes)
 
     def test_merge_delegates_to_merge_helm_indexes(self):
         with patch("artifactapi.artifact.virtual._merge_helm_indexes", return_value=b"merged") as mock_fn:
-            result = self.handler.merge([b"data"], ["m"], [{}], "http://proxy")
+            result = self.handler.merge([b"data"], [None], ["m"], [{}], "http://proxy")
-        mock_fn.assert_called_once_with([b"data"], ["m"], [{}], "http://proxy")
+        mock_fn.assert_called_once_with([b"data"], [None], ["m"], [{}], "http://proxy")
         assert result == b"merged"
 
 
@@ -160,6 +198,41 @@ class TestHandlersRegistry:
         assert isinstance(_HANDLERS["helm"], _VirtualHandler)
 
 
+# ---------------------------------------------------------------------------
+# _rewrite_urls
+# ---------------------------------------------------------------------------
+
+
+class TestRewriteUrls:
+    def _rewrite(self, urls, base_url="https://upstream.example.com", proxy_base="http://proxy.example.com", member_name="my-remote"):
+        return _rewrite_urls(urls, base_url, proxy_base, member_name)
+
+    def test_absolute_url_matching_base_is_rewritten(self):
+        result = self._rewrite(["https://upstream.example.com/chart-1.0.0.tgz"])
+        assert result == ["http://proxy.example.com/api/v1/remote/my-remote/chart-1.0.0.tgz"]
+
+    def test_relative_url_is_prepended_with_proxy_remote(self):
+        result = self._rewrite(["chart-1.0.0.tgz"])
+        assert result == ["http://proxy.example.com/api/v1/remote/my-remote/chart-1.0.0.tgz"]
+
+    def test_relative_url_with_leading_slash(self):
+        result = self._rewrite(["/chart-1.0.0.tgz"])
+        assert result == ["http://proxy.example.com/api/v1/remote/my-remote/chart-1.0.0.tgz"]
+
+    def test_absolute_url_not_matching_base_is_unchanged(self):
+        result = self._rewrite(["https://other.example.com/chart-1.0.0.tgz"])
+        assert result == ["https://other.example.com/chart-1.0.0.tgz"]
+
+    def test_empty_url_list_returns_empty(self):
+        assert self._rewrite([]) == []
+
+    def test_multiple_urls_all_rewritten(self):
+        urls = ["https://upstream.example.com/a-1.0.0.tgz", "b-2.0.0.tgz"]
+        result = self._rewrite(urls)
+        assert result[0] == "http://proxy.example.com/api/v1/remote/my-remote/a-1.0.0.tgz"
+        assert result[1] == "http://proxy.example.com/api/v1/remote/my-remote/b-2.0.0.tgz"
+
+
 # ---------------------------------------------------------------------------
 # _merge_helm_indexes
 # ---------------------------------------------------------------------------
@@ -167,8 +240,7 @@ class TestHandlersRegistry:
 
 class TestMergeHelmIndexes:
     def _merge(self, raw_indexes, member_names, member_configs, proxy_base="http://proxy.example.com"):
-        with patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve):
+        return _merge_helm_indexes(raw_indexes, [None] * len(raw_indexes), member_names, member_configs, proxy_base)
-            return _merge_helm_indexes(raw_indexes, member_names, member_configs, proxy_base)
 
     def _parse(self, raw):
         return yaml.safe_load(raw)
@@ -187,7 +259,18 @@ class TestMergeHelmIndexes:
     def test_first_member_wins_on_duplicate_name_and_version(self):
         index = self._parse(self._merge([_INDEX_A, _INDEX_B], ["member-a", "member-b"], [_CFG_A, _CFG_B]))
         v027 = next(e for e in index["entries"]["vault"] if e["version"] == "0.27.0")
-        assert "helm.releases.hashicorp.com" in v027["urls"][0]
+        assert "member-a" in v027["urls"][0]
+
+    def test_absolute_urls_rewritten_to_proxy(self):
+        index = self._parse(self._merge([_INDEX_A], ["member-a"], [_CFG_A]))
+        url = index["entries"]["vault"][0]["urls"][0]
+        assert url == "http://proxy.example.com/api/v1/remote/member-a/vault-0.27.0.tgz"
+
+    def test_relative_urls_rewritten_to_proxy(self):
+        cfg = {"base_url": "https://releases.rancher.com/server-charts/stable", "cache": {"mutable_ttl": 3600}}
+        index = self._parse(self._merge([_INDEX_RELATIVE], ["rancher-stable"], [cfg]))
+        url = index["entries"]["rancher"][0]["urls"][0]
+        assert url == "http://proxy.example.com/api/v1/remote/rancher-stable/rancher-2.13.1.tgz"
 
     def test_different_versions_of_same_chart_both_included(self):
         index = self._parse(self._merge([_INDEX_A, _INDEX_B], ["member-a", "member-b"], [_CFG_A, _CFG_B]))
@@ -260,7 +343,7 @@ class TestGetMemberIndex:
         storage.exists.return_value = True
         cache.is_index_valid.return_value = True
 
-        _, _, _, raw_data = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+        _, _, _, raw_data, _ = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
 
         assert raw_data == b"cached bytes"
 
@@ -283,7 +366,7 @@ class TestGetMemberIndex:
             mock_cls.return_value.__aenter__.return_value = mock_client
             mock_client.get.return_value = self._fake_response(b"fresh bytes")
 
-            _, _, _, raw_data = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+            _, _, _, raw_data, _ = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
 
         assert raw_data == b"fresh bytes"
 
@@ -293,7 +376,7 @@ class TestGetMemberIndex:
             mock_cls.return_value.__aenter__.return_value = mock_client
             mock_client.get.return_value = self._fake_response()
 
-            _, _, _, raw_data = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+            _, _, _, raw_data, _ = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
 
         assert raw_data == b"upstream bytes"
 
@@ -352,7 +435,7 @@ class TestGetMemberIndex:
             mock_cls.return_value.__aenter__.return_value = mock_client
             mock_client.get.side_effect = Exception("connection refused")
 
-            _, _, _, raw_data = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+            _, _, _, raw_data, _ = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
 
         assert raw_data is None
 
@@ -364,7 +447,7 @@ class TestGetMemberIndex:
             mock_cls.return_value.__aenter__.return_value = mock_client
             mock_client.get.return_value = self._fake_response()
 
-            _, _, _, raw_data = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+            _, _, _, raw_data, _ = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
 
         assert raw_data == b"upstream bytes"
 
@@ -375,7 +458,7 @@ class TestGetMemberIndex:
             mock_cls.return_value.__aenter__.return_value = mock_client
             mock_client.get.return_value = self._fake_response()
 
-            _, _, ttl, _ = await _get_member_index("m", cfg, "index.yaml", storage, cache)
+            _, _, ttl, _, _ = await _get_member_index("m", cfg, "index.yaml", storage, cache)
 
         assert ttl == 900
 
@@ -386,7 +469,7 @@ class TestGetMemberIndex:
             mock_cls.return_value.__aenter__.return_value = mock_client
             mock_client.get.return_value = self._fake_response()
 
-            _, _, ttl, _ = await _get_member_index("m", cfg, "index.yaml", storage, cache)
+            _, _, ttl, _, _ = await _get_member_index("m", cfg, "index.yaml", storage, cache)
 
         assert ttl == 3600
 
@@ -484,22 +567,16 @@ class TestVirtualRoute:
         mock_get.assert_not_called()
 
     def test_cache_miss_returns_200_with_yaml_content_type(self, client, patched_virtual_deps):
-        with (
+        with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
+            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None)
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
-        ):
-            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE)
             response = client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
         assert response.status_code == 200
         assert "text/yaml" in response.headers["content-type"]
 
     def test_cache_miss_response_contains_merged_entries(self, client, patched_virtual_deps):
-        with (
+        with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
+            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None)
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
-        ):
-            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE)
             response = client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
         index = yaml.safe_load(response.content)
@@ -507,35 +584,26 @@ class TestVirtualRoute:
 
     def test_cache_miss_stores_result_in_s3(self, client, patched_virtual_deps):
         deps = patched_virtual_deps
-        with (
+        with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
+            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None)
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
-        ):
-            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE)
             client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
         deps["storage"].upload.assert_called_once()
 
     def test_cache_miss_marks_index_cached(self, client, patched_virtual_deps):
         deps = patched_virtual_deps
-        with (
+        with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
+            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None)
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
-        ):
-            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE)
             client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
         deps["cache"].mark_index_cached.assert_called_once()
 
     def test_cache_miss_uses_min_ttl_across_members(self, client, patched_virtual_deps):
         deps = patched_virtual_deps
-        with (
+        with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
-        ):
             mock_get.side_effect = [
-                ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE),
+                ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None),
-                ("helm-member-2", _CFG_B, 1800, _INDEX_SIMPLE),
+                ("helm-member-2", _CFG_B, 1800, _INDEX_SIMPLE, None),
             ]
             client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
@@ -544,19 +612,16 @@ class TestVirtualRoute:
 
     def test_all_members_unreachable_returns_502(self, client, patched_virtual_deps):
         with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            mock_get.return_value = ("helm-test", _CFG_A, 3600, None)
+            mock_get.return_value = ("helm-test", _CFG_A, 3600, None, None)
             response = client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
         assert response.status_code == 502
 
     def test_one_member_unreachable_still_returns_200(self, client, patched_virtual_deps):
-        with (
+        with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
-        ):
             mock_get.side_effect = [
-                ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE),
+                ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None),
-                ("helm-member-2", _CFG_B, 1800, None),
+                ("helm-member-2", _CFG_B, 1800, None, None),
             ]
             response = client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
@@ -572,10 +637,9 @@ class TestVirtualRoute:
 
         with (
             patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
             patch.object(main_mod.config, "get_remote_config", side_effect=patched_get),
         ):
-            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE)
+            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None)
             response = client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
         # only helm-test was available — should succeed
@@ -586,11 +650,181 @@ class TestVirtualRoute:
         deps = patched_virtual_deps
         deps["storage"].upload.side_effect = Exception("S3 write error")
 
-        with (
+        with patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get:
-            patch("artifactapi.artifact.virtual._get_member_index", new_callable=AsyncMock) as mock_get,
+            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE, None)
-            patch("artifactapi.artifact.virtual._helm.resolve_content", side_effect=_identity_resolve),
-        ):
-            mock_get.return_value = ("helm-test", _CFG_A, 3600, _INDEX_SIMPLE)
             response = client.get("/api/v1/virtual/helm-virtual-test/index.yaml")
 
         assert response.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# _entries_to_msgpack_safe
+# ---------------------------------------------------------------------------
+
+
+class TestEntriesToMsgpackSafe:
+    def test_plain_string_values_pass_through(self):
+        entries = {"chart": [{"name": "chart", "version": "1.0.0", "urls": ["http://x/c.tgz"]}]}
+        result = _entries_to_msgpack_safe(entries)
+        assert result["chart"][0]["version"] == "1.0.0"
+
+    def test_datetime_converted_to_iso_string(self):
+        dt = datetime(2023, 6, 15, 12, 0, 0, tzinfo=UTC)
+        entries = {"chart": [{"name": "chart", "version": "1.0.0", "created": dt}]}
+        result = _entries_to_msgpack_safe(entries)
+        assert isinstance(result["chart"][0]["created"], str)
+        assert "2023-06-15" in result["chart"][0]["created"]
+
+    def test_date_converted_to_iso_string(self):
+        entries = {"chart": [{"name": "chart", "version": "1.0.0", "created": date(2023, 6, 15)}]}
+        result = _entries_to_msgpack_safe(entries)
+        assert result["chart"][0]["created"] == "2023-06-15"
+
+    def test_empty_entries_returns_empty_dict(self):
+        assert _entries_to_msgpack_safe({}) == {}
+
+    def test_multiple_versions_all_converted(self):
+        dt = datetime(2023, 1, 1, tzinfo=UTC)
+        entries = {
+            "chart": [
+                {"name": "chart", "version": "1.0.0", "created": dt},
+                {"name": "chart", "version": "2.0.0", "created": dt},
+            ]
+        }
+        result = _entries_to_msgpack_safe(entries)
+        for v in result["chart"]:
+            assert isinstance(v["created"], str)
+
+    def test_result_is_msgpack_serializable(self):
+        import msgpack
+
+        dt = datetime(2023, 6, 15, 12, 0, 0, tzinfo=UTC)
+        entries = {"chart": [{"name": "chart", "version": "1.0.0", "created": dt, "urls": ["http://x/c.tgz"]}]}
+        safe = _entries_to_msgpack_safe(entries)
+        packed = msgpack.packb(safe, use_bin_type=True)
+        unpacked = msgpack.unpackb(packed, raw=False)
+        assert unpacked["chart"][0]["created"] == safe["chart"][0]["created"]
+
+
+# ---------------------------------------------------------------------------
+# _merge_helm_indexes — pre-parsed entries path
+# ---------------------------------------------------------------------------
+
+
+class TestMergeHelmIndexesWithParsed:
+    """Verify that pre-parsed entries (from msgpack) produce the same output as raw YAML."""
+
+    def _parse_entries(self, raw: bytes) -> dict:
+        index = yaml.safe_load(raw)
+        return index.get("entries") or {}
+
+    def test_parsed_entries_produce_same_charts_as_raw(self):
+        parsed = self._parse_entries(_INDEX_A)
+        raw_result = yaml.safe_load(_merge_helm_indexes([_INDEX_A], [None], ["member-a"], [_CFG_A], "http://proxy.example.com"))
+        parsed_result = yaml.safe_load(_merge_helm_indexes([_INDEX_A], [parsed], ["member-a"], [_CFG_A], "http://proxy.example.com"))
+        assert set(raw_result["entries"].keys()) == set(parsed_result["entries"].keys())
+
+    def test_parsed_entries_urls_are_rewritten(self):
+        parsed = self._parse_entries(_INDEX_A)
+        result = yaml.safe_load(_merge_helm_indexes([_INDEX_A], [parsed], ["member-a"], [_CFG_A], "http://proxy.example.com"))
+        url = result["entries"]["vault"][0]["urls"][0]
+        assert "member-a" in url
+        assert "proxy.example.com" in url
+
+    def test_none_parsed_falls_back_to_raw_bytes(self):
+        result = yaml.safe_load(_merge_helm_indexes([_INDEX_A], [None], ["member-a"], [_CFG_A], "http://proxy.example.com"))
+        assert "vault" in result["entries"]
+
+    def test_mixed_parsed_and_raw_merge_correctly(self):
+        parsed_a = self._parse_entries(_INDEX_A)
+        result = yaml.safe_load(
+            _merge_helm_indexes(
+                [_INDEX_A, _INDEX_B],
+                [parsed_a, None],
+                ["member-a", "member-b"],
+                [_CFG_A, _CFG_B],
+                "http://proxy.example.com",
+            )
+        )
+        assert "vault" in result["entries"]
+        assert "nginx" in result["entries"]
+
+
+# ---------------------------------------------------------------------------
+# _get_member_index — msgpack cache behaviour
+# ---------------------------------------------------------------------------
+
+
+class TestGetMemberIndexMsgpack:
+    @pytest.fixture
+    def storage(self):
+        m = MagicMock()
+        m.get_object_key.side_effect = lambda name, path: f"{name}/{path}"
+        m.exists.return_value = False
+        m.download_object.return_value = _INDEX_SIMPLE
+        return m
+
+    @pytest.fixture
+    def cache(self):
+        m = MagicMock()
+        m.is_index_valid.return_value = False
+        return m
+
+    @pytest.fixture
+    def member_cfg(self):
+        return {"base_url": "https://helm.releases.hashicorp.com", "cache": {"mutable_ttl": 3600}}
+
+    def _fake_response(self, content=_INDEX_SIMPLE):
+        r = MagicMock()
+        r.content = content
+        r.raise_for_status = MagicMock()
+        return r
+
+    async def test_cache_hit_with_msgpack_returns_parsed_entries(self, storage, cache, member_cfg):
+        import msgpack
+
+        entries = {"mychart": [{"name": "mychart", "version": "1.0.0", "urls": ["http://x/c.tgz"]}]}
+        packed = msgpack.packb(entries, use_bin_type=True)
+
+        storage.exists.side_effect = lambda key: True
+        cache.is_index_valid.return_value = True
+        storage.download_object.side_effect = lambda key: packed if key.endswith("index.msgpack") else _INDEX_SIMPLE
+
+        _, _, _, raw_data, parsed = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+
+        assert parsed == entries
+
+    async def test_cache_miss_builds_msgpack_and_returns_parsed(self, storage, cache, member_cfg):
+        with patch("artifactapi.artifact.virtual.httpx.AsyncClient") as mock_cls:
+            mock_client = AsyncMock()
+            mock_cls.return_value.__aenter__.return_value = mock_client
+            mock_client.get.return_value = self._fake_response()
+
+            _, _, _, raw_data, parsed = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+
+        assert raw_data == _INDEX_SIMPLE
+        assert isinstance(parsed, dict)
+        assert "mychart" in parsed
+
+    async def test_broken_msgpack_rebuilds_from_raw_yaml(self, storage, cache, member_cfg):
+        storage.exists.side_effect = lambda key: True
+        cache.is_index_valid.return_value = True
+        storage.download_object.side_effect = lambda key: b"not-valid-msgpack" if key.endswith("index.msgpack") else _INDEX_SIMPLE
+
+        _, _, _, raw_data, parsed = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+
+        assert raw_data == _INDEX_SIMPLE
+        # Falls back to YAML parse and rebuilds msgpack — entries are returned
+        assert isinstance(parsed, dict)
+        assert "mychart" in parsed
+
+    async def test_upstream_failure_returns_none_for_both(self, storage, cache, member_cfg):
+        with patch("artifactapi.artifact.virtual.httpx.AsyncClient") as mock_cls:
+            mock_client = AsyncMock()
+            mock_cls.return_value.__aenter__.return_value = mock_client
+            mock_client.get.side_effect = Exception("timeout")
+
+            _, _, _, raw_data, parsed = await _get_member_index("m", member_cfg, "index.yaml", storage, cache)
+
+        assert raw_data is None
+        assert parsed is None