UN-3479 [FIX] Address review: prefer success over cached failure; guard None group listing

- _cached: a raced _FETCH_FAILED no longer shadows a peer's successful
  listing (would silently no-op all share replication for the run).
- list_groups: 'or {}' guard mirrors list_users so a None response can't
  raise AttributeError.
- Add regression tests for both race directions.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: chandrasekharan-zipstack

src/unstract/clone/client.py

@@ -123,7 +123,7 @@ def list_users(self) -> list[dict[str, Any]]:
     def list_groups(self) -> list[dict[str, Any]]:
         """List org groups; no server-side name filter — callers match in memory."""
         result = self._request("GET", "groups/")
-        return result if isinstance(result, list) else result.get("results", [])
+        return result if isinstance(result, list) else (result or {}).get("results", [])
 
     def create_group(self, payload: dict[str, Any]) -> dict[str, Any]:
         """Create a group; response has no ``id`` — re-list to learn the pk."""

src/unstract/clone/sharing.py

@@ -46,15 +46,22 @@ def _cached(ctx: CloneContext, key: str, build: Callable[[], Any]) -> Any:
     with ctx.share_cache_lock:
         if key in ctx.share_cache:
             return ctx.share_cache[key]
-    # Build outside the lock (HTTP call); worst case two threads race and
-    # one result wins — the listings are read-only so that's harmless.
+    # Build outside the lock (HTTP call). Two threads may race; the merge
+    # below keeps a real result over a failure sentinel (listings are
+    # read-only, so two successes are interchangeable).
     try:
         value = build()
     except Exception as e:
         logger.warning("share replication: %s listing failed: %s", key, e)
         value = _FETCH_FAILED
     with ctx.share_cache_lock:
-        ctx.share_cache.setdefault(key, value)
+        # Prefer a real result if we raced: a cached failure must not shadow a
+        # peer's success, else share replication silently no-ops for the run.
+        cached = ctx.share_cache.get(key, _FETCH_FAILED)
+        if key not in ctx.share_cache or (
+            cached is _FETCH_FAILED and value is not _FETCH_FAILED
+        ):
+            ctx.share_cache[key] = value
         return ctx.share_cache[key]
 
 

tests/clone/test_sharing.py

@@ -13,7 +13,7 @@
 from unstract.clone.context import CloneContext, CloneOptions, RemapTable
 from unstract.clone.phases.base import build_post_payload
 from unstract.clone.report import PhaseResult
-from unstract.clone.sharing import apply_share_state
+from unstract.clone.sharing import _FETCH_FAILED, _cached, apply_share_state
 
 
 class FakeClient:
@@ -208,6 +208,29 @@ def test_share_users_listing_caches_across_resources():
     assert len(tgt_client.share_posts) == 2
 
 
+def test_cached_failure_does_not_shadow_a_raced_success():
+    # A peer thread committing _FETCH_FAILED mid-build must not shadow our
+    # success, else share replication silently no-ops for the rest of the run.
+    ctx = _ctx(FakeClient(), FakeClient())
+
+    def build_while_peer_fails():
+        ctx.share_cache["k"] = _FETCH_FAILED  # racing thread commits first
+        return {"ok": 1}
+
+    assert _cached(ctx, "k", build_while_peer_fails) == {"ok": 1}
+    assert ctx.share_cache["k"] == {"ok": 1}
+
+
+def test_cached_success_is_not_overwritten_by_a_raced_failure():
+    ctx = _ctx(FakeClient(), FakeClient())
+
+    def build_fails_while_peer_succeeds():
+        ctx.share_cache["k"] = {"ok": 1}  # racing thread commits success first
+        raise RuntimeError("down")
+
+    assert _cached(ctx, "k", build_fails_while_peer_succeeds) == {"ok": 1}
+
+
 def test_share_post_failure_lands_in_errors():
     src_client = FakeClient(users=[{"id": "7", "email": "alice@x.com"}])
     tgt_client = FakeClient(users=[{"id": "70", "email": "alice@x.com"}])