ES reindex: fix connection-hang root causes + parallelize into scatter-gather [sc-45337]

[yodem]

Summary

The weekly full Elasticsearch reindex CronJob took ~26h and intermittently died with a "connection error," abandoning the entire run. This PR fixes the four root causes (not masks them) and re-architects the job into an isolated, parallel scatter-gather so no section can stall another.

Story: sc-45337 (Team Platform · Devops). Built via a written plan + 12 task-by-task subagent implementations, each individually reviewed, plus an Opus whole-branch review.

Investigation (production logs + code)

Decoded the last completed run (~26h): ~14.5h real work + ~11.3h of silent multi-hour stalls. Log hours 07/08/10–14 were entirely absent; the 6800→6900 stall was 2h17m ≈ Linux TCP keepalive (7200s); silent gaps had zero ES traffic → the process was blocked in Mongo reads. Bottleneck is Mongo/Python-side, not ES ingest.

Root causes fixed

1. 🔴 Mongo dead-socket hangs (primary killer) — MongoClient set no socketTimeoutMS (pymongo default = infinite), so a dead Mongo TCP socket blocked reads ~2h until OS keepalive. Now socketTimeoutMS=300s / connectTimeoutMS=20s / serverSelectionTimeoutMS=60s (global client; a generous ceiling, safe for the online path).
2. 🔴 Un-hardened ES client — the alias swap and all 67k sheet writes used a client with no retries. Hardened (request_timeout/retry_on_timeout/max_retries/http_compress); bulk flush now retries 429s.
3. 🟠 Sheet regression — commit 12e9a845f made optional sheet fields hard-required, silently dropping ~48,053 of 67,389 public sheets (71%) from search. Restored graceful fallbacks (only owner required) → sheets searchable again.
4. 🟠 Single 26h failure domain — split into a scatter-gather: reindex_init (pagesheetrank + create indexes + bulk-load settings) → K8s Indexed Job of N=8 --mode shard pods (deterministic size-aware sharding, per-shard backoffLimitPerIndex retry) → reindex_finalize (restore settings, doc-count sanity gate, atomic alias swap). A stall in one shard no longer holds the others hostage; the alias only swaps onto a verified-complete index. The orchestrator pod holds no ES/Mongo connection during the long phase.

Also included

• Bulk-load tuning (refresh_interval:-1 / replicas:0 during load, restore + _refresh in finalize).
• Sanity gate fails closed on an unreadable current-index count (final-review hardening).
• Helm: orchestrator RBAC + cronjob wiring that propagates ES/Mongo env, secrets, and the local_settings.py volume to shard pods.

Safety / rollout

• CronJob ships disabled (reindexElasticSearch.enabled: false) — merging changes nothing in prod.
• 20 unit tests pass; helm lint clean. GKE cluster is v1.35.5 (Indexed-Job per-index retry fields are GA).

⚠️ Before enabling in prod (must do)

• Staging dry-run on -debug indexes end-to-end: --mode init → --mode shard for shard-index 0..N-1 → --mode finalize; confirm shards index disjoint groups and text-debug doc count ≈ a monolith run.
• Validate _index_size_map resolves live VersionState counts; if it excepts it degrades to count-balanced (safe, just less head-balancing). Log a sample size-map.
• Confirm shard pods connect (ES secret + mounted local_settings.py) and that the orchestrator exits non-zero and leaves the alias untouched on an injected shard failure.
• Known limitation (gate-mitigated): cross-pod shard determinism assumes stable Mongo reads during the run; sub-10% group loss could slip past the 90% gate — ticket as a follow-up (consider a coverage assert in finalize).

🤖 Generated with Claude Code

[gitvelocity-reviewer]

I'll analyze this PR systematically before scoring.

PR Overview

This PR implements a comprehensive Elasticsearch reindex resilience and scatter-gather sharding system for the Sefaria project. It addresses four root causes of a failing 26-hour weekly reindex job:

1. MongoDB dead-socket hangs (infinite timeout)
2. Un-hardened ES client (no retries)
3. ~48k public sheets dropped due to overly strict validation
4. Single failure domain (no parallelism/isolation)

---

Code Review

1. sefaria/system/database.py

Good:

• MONGO_CLIENT_TIMEOUT_KWARGS is defined at module scope before the if hasattr(sys, '_doc_build') guard — correct placement.
• Values are well-reasoned (5min socket timeout, 20s connect, 60s server selection).

Issue:

• The MONGO_CLIENT_TIMEOUT_KWARGS dict is defined outside the if hasattr(sys, '_doc_build') guard, but the client variable is only created inside the else branch. This is intentional (the dict needs to exist even in doc-build mode), but the comment in the plan says "Place the dict assignment before the if hasattr(sys, '_doc_build'): guard" — which is what was done. ✓

Minor concern:

• socketTimeoutMS=300_000 (5 minutes) on the online request path could cause issues for very slow admin queries that legitimately take >5 minutes. The plan acknowledges this tradeoff, but it's worth monitoring.

2. sefaria/helper/search.py

Good:

• get_elasticsearch_client() now has retry/timeout config.
• http_compress=True is a nice addition for bandwidth reduction.

Issue:

• The test test_default_es_client_is_hardened accesses private attributes (es._max_retries, es._retry_on_timeout, es._request_timeout). These are implementation details of elasticsearch-py 8.x and could break on minor version updates. The plan itself acknowledges this risk. A more stable approach would be to test behavior (e.g., mock the transport and verify retry behavior) rather than internal state.

3. sefaria/search.py

Sheet indexing fix (Task 4):

• The fix is correct — removing the overly strict validation that dropped 71% of public sheets.
• make_sheet_text now gracefully handles missing title/summary.

Potential issue in index_sheet:

monkeypatch.setattr(search.es_client, "create",
                    lambda index, id, body: created.update({"id": id, "body": body}))

The es_client.create signature in the test uses keyword args, but the actual call in search.py line 193 is:

es_client.create(index=index_name, id=id, body=doc)

This is fine since both use keyword args.

_index_doc_count function:

def _index_doc_count(index_name):
    try:
        if not index_client.exists(index=index_name):
            return 0
    except Exception as e:
        logger.warning(...)
        return None
    try:
        stats = index_client.stats(index=index_name)
        return stats.get('_all', {}).get('primaries', {}).get('docs', {}).get('count', 0)
    except Exception as e:
        logger.warning(...)
        return None

Good defensive coding — returns None on transient errors so callers can fail closed.

reindex_finalize sanity gate:

if new_count is None:
    raise ValueError(...)
if current_count is None:
    raise ValueError(...)

Good — fails closed on unreadable counts.

_select_shard_groups snake distribution:
The algorithm is correct and elegant. The snake distribution ensures large books are spread across shards rather than all landing in shard 0.

Potential issue in reindex_index_shard:

def reindex_index_shard(type, shard_index=None, shard_count=None, debug=False):
    names = get_new_and_current_index_names(type=type, debug=debug)
    if type == 'text':
        TextIndexer.clear_cache()
        TextIndexer.index_all(names['new'], debug=debug, shard_index=shard_index, shard_count=shard_count)

If reindex_init hasn't been called first (e.g., in a retry scenario), this will try to index into a non-existent index. The plan says reindex_init is idempotent, but there's no guard here.

index_all_of_type backward compatibility:

def index_all_of_type(type, skip=0, debug=False):
    """...
    Note: the ``skip`` parameter is accepted for backward compatibility but is no longer
    forwarded; resume-from-skip is a no-op now that indexing is sharded/phase-split.
    """
    reindex_init(type, debug=debug)
    reindex_index_shard(type, debug=debug)
    reindex_finalize(type, debug=debug)

The skip parameter silently becomes a no-op. This could be surprising for callers who relied on it. The docstring documents this, which is good.

index_all_of_type removes the 10-second countdown:
The original code had a 10-second countdown before indexing. This has been removed in the new implementation. While the countdown was probably just a safety measure, removing it without explicit mention could be a concern.

4. scripts/scheduled/reindex_orchestrator.py

Good:

• Pure functions (job_terminal_state, build_shard_job_manifest) are separated from I/O.
• Heavy imports are lazy (inside main()), enabling unit testing without kubernetes/django.
• Env propagation to shard pods is thorough.

Issue — maxFailedIndexes: 0:

"maxFailedIndexes": 0,

With maxFailedIndexes: 0, if any shard fails (even after backoffLimitPerIndex: 2 retries), the entire job fails immediately. This is intentional (the plan says "NOT finalizing (alias unchanged)"), but it means a single flaky shard aborts the whole reindex. Consider whether maxFailedIndexes: 1 or higher would be more resilient.

Issue — pip install numpy kubernetes in the command:

command = [
    "bash", "-c",
    "mkdir -p /log && touch /log/sefaria_book_errors.log && pip install numpy && /app/run /app/scripts/scheduled/reindex_elasticsearch_cronjob.py --mode shard --type text",
]

Installing packages at runtime in production pods is an anti-pattern. These should be baked into the Docker image. The kubernetes package is added to requirements.txt, but numpy is installed at runtime. This is a pre-existing issue but worth noting.

Issue — time.sleep(10) after job deletion:

batch.delete_namespaced_job(job_name, namespace, propagation_policy="Background")
time.sleep(10)

Background propagation is async — 10 seconds may not be enough for the old job to fully terminate before creating the new one. A more robust approach would be to poll until the job is gone.

Issue — No timeout on the polling loop:

while True:
    job = batch.read_namespaced_job_status(job_name, namespace)
    ...
    time.sleep(60)

If the job gets stuck (neither completing nor failing), this loop runs forever. A maximum wait time should be added.

5. helm-chart/sefaria/templates/rbac/reindex-orchestrator-rbac.yaml

Good:

• Minimal RBAC — only jobs, jobs/status, pods, pods/log in the namespace.
• Namespace-scoped Role (not ClusterRole) — good security practice.

Minor issue:

• The jobs/status subresource is listed in the Role, but the orchestrator uses read_namespaced_job_status which reads the main jobs resource. The jobs/status entry is harmless but slightly redundant.

6. helm-chart/sefaria/values.yaml

Good:

• shardResources is defined in values but not actually used in the CronJob template (it would be used by the dynamically-created shard Job). The orchestrator passes resources directly in build_shard_job_manifest. This is a minor inconsistency — the Helm values define shard resources but they're hardcoded in the Python script.

7. requirements.txt

kubernetes==31.*

Using a wildcard version pin (31.*) is unusual for requirements.txt. Most projects use >=31,<32 or a specific version. This could cause issues with pip's dependency resolver.

8. Test Coverage

sefaria/system/tests/database_timeout_test.py:

• Tests both the dict values and the live client options. Good.
• test_live_client_has_timeouts_applied requires a live MongoDB connection — this may fail in CI without a Mongo instance.

sefaria/system/tests/reindex_orchestrator_test.py:

• Good use of importlib to avoid heavy imports.
• Tests job_terminal_state and build_shard_job_manifest thoroughly.
• Missing test: what happens when job_terminal_state is called with succeeded=0, failed=0 (job not started yet)?

sefaria/tests/search_test.py:

• Good coverage of the new functions.
• test_index_all_calls_select_shard_groups_when_sharding is well-structured.
• test_reindex_finalize_sanity_gate and test_reindex_finalize_fails_closed_on_unreadable_current cover important edge cases.

Missing tests:

• reindex_init is not directly tested (only indirectly through index_all_of_type).
• reindex_index_shard with type='sheet' is not tested.
• The _index_size_map method is not tested.
• index_all_of_type composition (init → shard → finalize) is not tested.

9. The Plan Document (docs/superpowers/plans/...)

This is a 1141-line AI agent execution plan. It's detailed and well-structured, but it's a documentation/planning artifact, not production code. It's included in the diff but shouldn't affect the code quality score significantly.

---

Summary of Issues

Critical:

• None

High:

• maxFailedIndexes: 0 means any single shard failure aborts the entire reindex — consider whether this is the right tradeoff
• No timeout on the orchestrator polling loop (could run forever)

Medium:

• Test for test_live_client_has_timeouts_applied requires live MongoDB — may fail in CI
• Tests access private elasticsearch-py attributes (fragile)
• pip install numpy at runtime in shard pods (anti-pattern)
• time.sleep(10) after job deletion may be insufficient
• shardResources in values.yaml is unused by the Helm templates (hardcoded in Python)
• kubernetes==31.* version pin syntax is unusual

Low:

• skip parameter silently becomes no-op in index_all_of_type
• jobs/status RBAC entry is slightly redundant
• Missing tests for reindex_init, _index_size_map, sheet shard path

---

Code Quality Score

Scope (S): This PR touches 12 files across multiple subsystems: database layer, search/indexing engine, Kubernetes orchestration, Helm charts, and test infrastructure. It introduces new public functions (reindex_init, reindex_index_shard, reindex_finalize), a new script, new RBAC resources, and a new dependency. S = 17

Architecture (A): Major architectural shift — from a monolithic serial reindex to a scatter-gather parallel architecture with init/shard/finalize phases, Kubernetes Indexed Jobs, and an orchestrator pattern. New service boundary (orchestrator pod → shard pods). New dependency (kubernetes client). A = 17

Implementation (I): Snake distribution algorithm for balanced sharding, deterministic shard assignment, sanity gate with fail-closed semantics, env propagation to dynamically-created pods, lazy imports for testability, pure functions separated from I/O. Complex state management across phases. I = 16

Risk (R): Changes the global MongoDB client (affects online path), modifies the alias swap logic (must remain atomic), introduces Kubernetes Job creation (new operational surface), changes the weekly reindex entrypoint. The sanity gate mitigates alias swap risk. backoffLimit: 0 on the CronJob is a risk (no retry at the CronJob level). R = 14

Quality (Q): Good test coverage for the new logic (sharding determinism, sanity gate, manifest building, bulk retry kwargs, sheet fallback). Tests use monkeypatching appropriately. Some fragile private-attribute assertions. Missing tests for reindex_init, _index_size_map. The plan document is thorough. Q = 11

Performance/Security (P): Bulk-load settings (disable refresh/replicas during ingest), snake distribution for load balancing, minimal RBAC (namespace-scoped), retry with backoff for 429s. P = 4

Base Score: 17 + 17 + 16 + 14 + 11 + 4 = 79

Effort Scale:

• Effective Lines: 856 → Extra Large tier (ESF: 1.0x)
• File Count: 12 → Small tier
• Gap: Small - XL = negative → No bump
• Final ESF: 1.0x

Final Score: 79 × 1.0 = 79

Code Quality Data (JSON)

{
  "_schema": "code_quality_v5",
  "total_score": 79,
  "total_factors": "79 × 1.0 (Extra Large ESF) = 79",
  "scope_score": 17,
  "scope_factors": "12 files across database, search/indexing, Kubernetes orchestration, Helm charts, test infrastructure; new public API (reindex_init/shard/finalize); new script; new RBAC resources; new kubernetes dependency",
  "architecture_score": 17,
  "architecture_factors": "Major shift from monolithic serial reindex to scatter-gather parallel architecture; new orchestrator pattern; Kubernetes Indexed Job creation; init/shard/finalize phase decomposition; new service boundary between orchestrator and shard pods",
  "implementation_score": 16,
  "implementation_factors": "Snake distribution algorithm for balanced sharding; deterministic shard assignment without coordination; sanity gate with fail-closed semantics; env propagation to dynamically-created pods; lazy imports for testability; pure functions separated from I/O; complex multi-phase state management",
  "risk_score": 14,
  "risk_factors": "Global MongoDB client change affects online path (+4); alias swap logic modified (must remain atomic) (+4); new Kubernetes Job creation surface (+3); backoffLimit:0 on CronJob (+2); sanity gate mitigates alias swap risk (-2); feature is gated by enabled=false default (-1)",
  "quality_score": 11,
  "quality_factors": "Good coverage of sharding determinism, sanity gate, manifest building, bulk retry kwargs, sheet fallback; appropriate monkeypatching; fragile private-attribute assertions in ES client test; missing tests for reindex_init, _index_size_map, sheet shard path; live-Mongo test may fail in CI",
  "perf_security_score": 4,
  "perf_security_factors": "Bulk-load settings (