From be8b39cdc9ce4a6067f521cf72f738dd6fcce5c0 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Thu, 3 Sep 2026 13:50:16 -0400
Subject: [PATCH 01/13] perf(crawl): thread the crawl over input files, drop
 the crawlspace

Pairs with the dug change that makes expand_concept fetch concurrently.
That bounds out at roughly the number of real TranQL queries per concept,
measured at ~4, so the rest of the concurrency has to come from here: the
crawl_tranql loop over concept files, which is one iteration per annotated
input and tens of thousands of them for a dbGaP dataset.

Files are independent -- each decodes its own concepts, expands them and
writes its own output dir -- and the work is nearly all http wait, so
threads scale it despite the GIL. The loop body moves to crawl_one_file
unchanged. Failures are re-raised rather than left in the pool, because a
file that raises produced no output and the task must not report success.

indexing.crawl_workers and indexing.crawl_file_workers multiply, and their
product should not exceed what TranQL can serve at once -- past that,
requests only queue at its gunicorn. search-chart carried workerCount: 1
with a sync worker, which is why the service sat at 2.5% of its two-core
limit while being the bottleneck; that goes to 16 alongside this.

The crawlspace wiring goes away with the file cache it fed: crawl_concepts
no longer creates or assigns it, and crawl_tranql no longer clears the
crawl_output dir, since nothing writes there now.
---
 src/roger/config/__init__.py          |   7 ++
 src/roger/config/config.yaml          |   2 +
 src/roger/pipelines/base.py           | 144 ++++++++++++++------------
 tests/unit/test_crawl_file_workers.py |  83 +++++++++++++++
 4 files changed, 170 insertions(+), 66 deletions(-)
 create mode 100644 tests/unit/test_crawl_file_workers.py

diff --git a/src/roger/config/__init__.py b/src/roger/config/__init__.py
index a7732b0..a986732 100644
--- a/src/roger/config/__init__.py
+++ b/src/roger/config/__init__.py
@@ -232,6 +232,13 @@ class IndexingConfig(DictLike):
         "anat_to_pheno": ["anatomical_entity", "phenotypic_feature"],
     })
     tranql_endpoint: str = "http://tranql-service/tranql/query?dynamic_id_resolution=true&asynchronous=false"
+    # Concurrency for the crawl. `crawl_workers` threads the TranQL fetches
+    # within one concept; `crawl_file_workers` threads whole input files.
+    # They multiply, and the product should not exceed what the TranQL
+    # service can serve at once (its gunicorn worker count) -- past that,
+    # requests only queue.
+    crawl_workers: int = 4
+    crawl_file_workers: int = 4
     # by default skips node to element queries
     node_to_element_queries: dict = field(default_factory=lambda: {})
     element_mapping: str = ""
diff --git a/src/roger/config/config.yaml b/src/roger/config/config.yaml
index 3d883d1..42435ef 100644
--- a/src/roger/config/config.yaml
+++ b/src/roger/config/config.yaml
@@ -120,6 +120,8 @@ indexing:
     "chemical_mixture_to_disease": ["chemical_mixture", "disease"]
     "phen_to_anat": ["phenotypic_feature", "anatomical_entity"]
   tranql_endpoint: "http://tranql-service/tranql/query?dynamic_id_resolution=true&asynchronous=false"
+  crawl_workers: 4
+  crawl_file_workers: 4
   node_to_element_queries:
     enabled: false
     cde:
diff --git a/src/roger/pipelines/base.py b/src/roger/pipelines/base.py
index f3af15f..f930254 100644
--- a/src/roger/pipelines/base.py
+++ b/src/roger/pipelines/base.py
@@ -778,10 +778,6 @@ def crawl_concepts(self, concepts, data_set_name, output_path=None):
         :param data_set_name:
         :return:
         """
-        # TODO crawl dir seems to be storaing crawling info to avoid
-        # re-crawling, but is that consting us much? , it was when tranql was
-        # slow, but might right to consider getting rid of it.
-        crawl_dir = storage.dug_crawl_path('crawl_output')
         output_file_name = os.path.join(data_set_name,
                                         'expanded_concepts.txt')
         extracted_dug_elements_file_name = os.path.join(
@@ -796,7 +792,6 @@ def crawl_concepts(self, concepts, data_set_name, output_path=None):
             extracted_output_file = os.path.join(
                 output_path, extracted_dug_elements_file_name)
 
-        Path(crawl_dir).mkdir(parents=True, exist_ok=True)
         extracted_dug_elements = []
         log.debug("Creating Dug Crawler object")
         crawler = Crawler(
@@ -806,8 +801,8 @@ def crawl_concepts(self, concepts, data_set_name, output_path=None):
             tranqlizer=self.tranqlizer,
             tranql_queries=self.tranql_queries,
             http_session=self.cached_session,
+            crawl_workers=self.config.indexing.crawl_workers,
         )
-        crawler.crawlspace = crawl_dir
         counter = 0
         total = len(concepts)
         for concept in concepts.values():
@@ -1220,78 +1215,95 @@ def crawl_tranql(self, to_string=False, concept_files=None,
                 input_data_path, format='txt')
 
         if output_data_path:
-            crawl_dir = os.path.join(output_data_path, 'crawl_output')
             expanded_concepts_dir = os.path.join(output_data_path,
                                                  'expanded_concepts')
         else:
-            crawl_dir = storage.dug_crawl_path('crawl_output')
             expanded_concepts_dir = storage.dug_expanded_concepts_path("")
-        log.info("Clearing crawl output dir %s", crawl_dir)
-        storage.clear_dir(crawl_dir)
 
         log.info("Clearing expanded concepts dir: %s", expanded_concepts_dir)
         storage.clear_dir(expanded_concepts_dir)
 
-        log.info("Crawling Dug Concepts, found %d file(s).",
-                 len(concept_files))
-        for file_ in concept_files:
-            objects = storage.read_object(file_)
-            objects = objects or {}
-            if not objects:
-                log.info(f'no concepts in {file_}')
-            data_set =  jsonpickle.decode(objects)
-            original_variables_dataset_name = os.path.split(
-                os.path.dirname(file_))[-1]
-            self.crawl_concepts(concepts=data_set,
-                                data_set_name=original_variables_dataset_name,
-                                output_path= output_data_path)
-
-            # After expanding concepts with KG answers, update the
-            # corresponding elements' optional_terms so that KG-derived
-            # search terms are present when elements are later indexed.
-            # This mirrors what Crawler.crawl() does after concept expansion.
-            # The updated elements are written to the expanded concepts
-            # directory (alongside expanded_concepts.txt) rather than
-            # mutating the annotate step's output.
-            annotation_elements_file = os.path.join(
-                os.path.dirname(file_), 'elements.txt')
-            expanded_elements_file_name = os.path.join(
-                original_variables_dataset_name, 'elements.txt')
-            if not output_data_path:
-                expanded_elements_file = (
-                    storage.dug_expanded_concepts_path(
-                        expanded_elements_file_name))
-            else:
-                expanded_elements_file = os.path.join(
-                    output_data_path, expanded_elements_file_name)
-            if os.path.exists(annotation_elements_file):
-                log.info("Updating element optional terms from expanded "
-                         "concepts for %s", original_variables_dataset_name)
-                elements = jsonpickle.decode(
-                    storage.read_object(annotation_elements_file))
-                for element in elements:
-                    if isinstance(element, DugConcept):
-                        continue
-                    # Replace each element's concept references with
-                    # the expanded versions that now carry kg_answers.
-                    for concept_id in list(element.concepts.keys()):
-                        if concept_id in data_set:
-                            element.concepts[concept_id] = data_set[
-                                concept_id]
-                    element.set_optional_terms()
-                storage.write_object(
-                    jsonpickle.encode(elements, indent=2),
-                    expanded_elements_file)
-                log.info("Updated elements serialized to %s",
-                         expanded_elements_file)
-            else:
-                log.warning("Elements file not found at %s, skipping "
-                            "optional terms update",
-                            annotation_elements_file)
+        workers = max(1, int(self.config.indexing.crawl_file_workers))
+        workers = min(workers, len(concept_files)) or 1
+        log.info("Crawling Dug Concepts, found %d file(s) with %d worker(s).",
+                 len(concept_files), workers)
+        if workers == 1:
+            for file_ in concept_files:
+                self.crawl_one_file(file_, output_data_path)
+        else:
+            # Files are independent: each decodes its own concepts, expands
+            # them and writes its own output dir. The work is nearly all
+            # http wait on TranQL, so threads scale it despite the GIL.
+            with ThreadPoolExecutor(max_workers=workers,
+                                    thread_name_prefix='crawl') as pool:
+                futures = [pool.submit(self.crawl_one_file, file_,
+                                       output_data_path)
+                           for file_ in concept_files]
+                # surface the first failure rather than letting the pool
+                # swallow it; a raised exception means that file produced
+                # nothing and the task must not report success
+                for future in futures:
+                    future.result()
 
         output_log = self.log_stream.getvalue() if to_string else ''
         return output_log
 
+    def crawl_one_file(self, file_, output_data_path=None):
+        "Expand one annotate output file's concepts and write its outputs"
+        objects = storage.read_object(file_)
+        objects = objects or {}
+        if not objects:
+            log.info(f'no concepts in {file_}')
+        data_set =  jsonpickle.decode(objects)
+        original_variables_dataset_name = os.path.split(
+            os.path.dirname(file_))[-1]
+        self.crawl_concepts(concepts=data_set,
+                            data_set_name=original_variables_dataset_name,
+                            output_path= output_data_path)
+
+        # After expanding concepts with KG answers, update the
+        # corresponding elements' optional_terms so that KG-derived
+        # search terms are present when elements are later indexed.
+        # This mirrors what Crawler.crawl() does after concept expansion.
+        # The updated elements are written to the expanded concepts
+        # directory (alongside expanded_concepts.txt) rather than
+        # mutating the annotate step's output.
+        annotation_elements_file = os.path.join(
+            os.path.dirname(file_), 'elements.txt')
+        expanded_elements_file_name = os.path.join(
+            original_variables_dataset_name, 'elements.txt')
+        if not output_data_path:
+            expanded_elements_file = (
+                storage.dug_expanded_concepts_path(
+                    expanded_elements_file_name))
+        else:
+            expanded_elements_file = os.path.join(
+                output_data_path, expanded_elements_file_name)
+        if os.path.exists(annotation_elements_file):
+            log.info("Updating element optional terms from expanded "
+                     "concepts for %s", original_variables_dataset_name)
+            elements = jsonpickle.decode(
+                storage.read_object(annotation_elements_file))
+            for element in elements:
+                if isinstance(element, DugConcept):
+                    continue
+                # Replace each element's concept references with
+                # the expanded versions that now carry kg_answers.
+                for concept_id in list(element.concepts.keys()):
+                    if concept_id in data_set:
+                        element.concepts[concept_id] = data_set[
+                            concept_id]
+                element.set_optional_terms()
+            storage.write_object(
+                jsonpickle.encode(elements, indent=2),
+                expanded_elements_file)
+            log.info("Updated elements serialized to %s",
+                     expanded_elements_file)
+        else:
+            log.warning("Elements file not found at %s, skipping "
+                        "optional terms update",
+                        annotation_elements_file)
+
     def index_concepts(self, to_string=False,
                        input_data_path=None, output_data_path=None):
         "Index concepts from expanded concept files"
diff --git a/tests/unit/test_crawl_file_workers.py b/tests/unit/test_crawl_file_workers.py
new file mode 100644
index 0000000..7a8c4b4
--- /dev/null
+++ b/tests/unit/test_crawl_file_workers.py
@@ -0,0 +1,83 @@
+"""crawl_tranql threads whole input files; failures must not be swallowed.
+
+The dir loop is where the bulk of crawl concurrency comes from -- one file
+per annotated input, tens of thousands of them for a dbGaP dataset. Two
+things must hold: every file gets processed exactly once regardless of
+worker count, and a file that raises fails the task rather than quietly
+producing no output.
+"""
+import threading
+import types
+
+import pytest
+
+
+def make_pipeline(workers, crawl_one):
+    """A stand-in carrying only what crawl_tranql touches."""
+    from roger.pipelines.base import DugPipeline
+
+    pipeline = types.SimpleNamespace()
+    pipeline.config = types.SimpleNamespace(
+        indexing=types.SimpleNamespace(crawl_file_workers=workers))
+    pipeline.log_stream = types.SimpleNamespace(getvalue=lambda: '')
+    pipeline.crawl_one_file = crawl_one
+    pipeline.crawl_tranql = types.MethodType(
+        DugPipeline.crawl_tranql.__wrapped__
+        if hasattr(DugPipeline.crawl_tranql, '__wrapped__')
+        else DugPipeline.crawl_tranql, pipeline)
+    return pipeline
+
+
+FILES = [f"/in/file{i}/concepts.txt" for i in range(12)]
+
+
+@pytest.mark.parametrize("workers", [1, 4, 8])
+def test_every_file_processed_once(workers, monkeypatch, tmp_path):
+    import roger.pipelines.base as base
+
+    seen = []
+    lock = threading.Lock()
+
+    def crawl_one(file_, output_data_path=None):
+        with lock:
+            seen.append(file_)
+
+    monkeypatch.setattr(base.storage, 'clear_dir', lambda *a, **k: None)
+    pipeline = make_pipeline(workers, crawl_one)
+    pipeline.crawl_tranql(concept_files=list(FILES),
+                          output_data_path=str(tmp_path))
+
+    assert sorted(seen) == sorted(FILES)
+    assert len(seen) == len(FILES)
+
+
+def test_one_bad_file_fails_the_task(monkeypatch, tmp_path):
+    import roger.pipelines.base as base
+
+    def crawl_one(file_, output_data_path=None):
+        if file_.endswith("file7/concepts.txt"):
+            raise ValueError("bad pickle")
+
+    monkeypatch.setattr(base.storage, 'clear_dir', lambda *a, **k: None)
+    pipeline = make_pipeline(4, crawl_one)
+    with pytest.raises(ValueError, match="bad pickle"):
+        pipeline.crawl_tranql(concept_files=list(FILES),
+                              output_data_path=str(tmp_path))
+
+
+def test_worker_count_is_capped_by_file_count(monkeypatch, tmp_path):
+    """Two files must not spin up eight threads."""
+    import roger.pipelines.base as base
+
+    threads = set()
+    lock = threading.Lock()
+
+    def crawl_one(file_, output_data_path=None):
+        with lock:
+            threads.add(threading.current_thread().name)
+
+    monkeypatch.setattr(base.storage, 'clear_dir', lambda *a, **k: None)
+    pipeline = make_pipeline(8, crawl_one)
+    pipeline.crawl_tranql(concept_files=FILES[:2],
+                          output_data_path=str(tmp_path))
+    assert len(threads) <= 2

From c9ed8f837e4ce96a769b07055283518c23bcd326 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Thu, 3 Sep 2026 14:30:19 -0400
Subject: [PATCH 02/13] bump roger

---
 requirements.txt | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/requirements.txt b/requirements.txt
index 817aed9..2c359e4 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -5,7 +5,7 @@ git+https://github.com/falkordb/falkordb-bulk-loader.git@v1.0.6
 setuptools>=66
 pytest
 PyYAML
-git+https://github.com/helxplatform/dug@develop
+git+https://github.com/helxplatform/dug@tranql-crawl-concurrency
 orjson>=3.11
 git+https://github.com/helxplatform/kg_utils.git@v0.0.10.1
 # kg_utils/merging.py hashes an f-string; xxhash 4.0 dropped the

From 5b0f4f6abb7be4cee0c9e46c996f32e081b536dd Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Sat, 5 Sep 2026 11:36:51 -0400
Subject: [PATCH 03/13] fix(bulkload): strip bare \r from node description/name

A bare carriage return with no paired \n still reads as a line break
under universal newlines, so both our own tooling and the
falkordb_bulk_loader CSV parser split one row into two, corrupting the
column count. Hit in production on a dbGaP Study "activitybk" field
whose description had codebook text pasted in with stray CRs.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/core/bulkload.py                    | 15 ++++--
 .../test_bulkload_newline_sanitization.py     | 50 +++++++++++++++++++
 2 files changed, 61 insertions(+), 4 deletions(-)
 create mode 100644 tests/unit/test_bulkload_newline_sanitization.py

diff --git a/src/roger/core/bulkload.py b/src/roger/core/bulkload.py
index 4dc2736..a2dd0db 100644
--- a/src/roger/core/bulkload.py
+++ b/src/roger/core/bulkload.py
@@ -54,12 +54,19 @@ def create_nodes_csv_file(self, input_data_path=None, output_data_path=None):
         merged_nodes_file = storage.merged_objects('nodes', input_data_path)
         counter = 1
         for node in storage.json_line_iter(merged_nodes_file):
+            # \r alone (no paired \n) still reads as a line break under
+            # universal newlines -- both our own row count and the bulk
+            # loader's CSV parser split on it, turning one row into two
+            # and corrupting the column count. Seen in dbGaP codebook text
+            # pasted in with stray CRs (e.g. a Study "activitybk" field).
             if node.get('description'):
-                node['description'] = node['description'].replace('\n',
-                                                                  ' ')
+                node['description'] = (
+                    node['description'].replace('\r\n', ' ')
+                    .replace('\r', ' ').replace('\n', ' '))
             if node.get('name'):
-                node['name'] = node['name'].replace('\n',
-                                                    ' ')
+                node['name'] = (
+                    node['name'].replace('\r\n', ' ')
+                    .replace('\r', ' ').replace('\n', ' '))
             if not node.get('category'):
                 category_error_nodes.add(node['id'])
                 node['category'] = [BiolinkModel.root_type]
diff --git a/tests/unit/test_bulkload_newline_sanitization.py b/tests/unit/test_bulkload_newline_sanitization.py
new file mode 100644
index 0000000..c50ca3f
--- /dev/null
+++ b/tests/unit/test_bulkload_newline_sanitization.py
@@ -0,0 +1,50 @@
+"""A bare \\r (no paired \\n) in a node's description/name still reads as a
+line break under universal newlines. Both our own tooling and the
+falkordb_bulk_loader CSV parser split on it, turning one row into two and
+corrupting the column count -- seen in dbGaP codebook text pasted in with
+stray CRs (a Study "activitybk" field).
+"""
+from roger.core.bulkload import BulkLoad
+from roger.config import config
+
+
+def test_bare_cr_in_description_does_not_split_the_row(tmp_path, monkeypatch):
+    from roger.core import bulkload as bulkload_mod
+
+    node = {
+        'id': 'HDP00066:activitybk',
+        'category': ['biolink:Study'],
+        'name': 'activitybk',
+        'description': 'SECTION H. READ Scale\r Ask: do you read to your child?',
+    }
+    leaf_class = 'biolink:Study'
+    schema = {
+        leaf_class: {
+            'category': 'list', 'name': 'str',
+            'description': 'str', 'id': 'str',
+        }
+    }
+
+    monkeypatch.setattr(bulkload_mod.storage, 'merged_objects',
+                        lambda kind, path=None: 'nodes.jsonl')
+    monkeypatch.setattr(bulkload_mod.storage, 'json_line_iter',
+                        lambda path: iter([node]))
+    monkeypatch.setattr(bulkload_mod.storage, 'read_schema',
+                        lambda schema_type, path=None: schema)
+    monkeypatch.setattr(bulkload_mod.storage, 'bulk_path',
+                        lambda name, path=None: str(tmp_path / name))
+
+    class FakeBiolink:
+        def get_leaf_class(self, names):
+            return leaf_class
+
+    bulk = BulkLoad(FakeBiolink(), config=config)
+    bulk.create_nodes_csv_file(input_data_path=None, output_data_path=None)
+
+    out_file = tmp_path / 'nodes' / f"{leaf_class.replace(':', '~')}.csv-0-1"
+    lines = out_file.read_bytes().split(b'\n')
+    lines = [l for l in lines if l]
+    assert len(lines) == 2, lines  # header + one data row, not split in two
+    header, row = (l.decode().split('\x1e') for l in lines)
+    assert len(header) == len(row) == 4
+    assert b'\r' not in out_file.read_bytes()

From 69774c34d186af1b49c116e51cd1b0cfddb1b0be Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Sun, 6 Sep 2026 13:54:27 -0400
Subject: [PATCH 04/13] perf(storage): gzip crawl-stage .txt/.jsonl artifacts

expanded_concepts.txt and the per-directory expanded elements.txt embed
full TranQL knowledge-graph subgraphs per concept -- repeated CURIEs,
biolink categories, and JSON keys that compress 5-10x+. That's the
difference between a large dataset's crawl output fitting on the task
PVC and hitting ENOSPC mid-run, as bdc-recover just did at 25h/try 12.

Same filename/extension as before, so every existing glob pattern that
finds these files by name still matches. read_object detects the gzip
magic number, so artifacts already committed to lakefs before this
still read as plain text -- no forced re-migration.

migrate_pickled_classes.py reads/writes these same files directly
(bypassing storage.py), so it needs the same transparent gzip handling
to keep working -- and now also brings old plain-text artifacts up to
the current (gzip) on-disk format when it rewrites them.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 scripts/migrate_pickled_classes.py        | 34 +++++++++++++++++------
 src/roger/core/storage.py                 | 30 ++++++++++++++++++--
 tests/unit/test_storage_gzip_artifacts.py | 27 ++++++++++++++++++
 3 files changed, 80 insertions(+), 11 deletions(-)
 create mode 100644 tests/unit/test_storage_gzip_artifacts.py

diff --git a/scripts/migrate_pickled_classes.py b/scripts/migrate_pickled_classes.py
index a6e06b1..6ace171 100644
--- a/scripts/migrate_pickled_classes.py
+++ b/scripts/migrate_pickled_classes.py
@@ -16,6 +16,7 @@
 """
 
 import argparse
+import gzip
 import importlib
 import os
 import json
@@ -39,6 +40,23 @@
 PY_OBJECT = re.compile(r'"py/object":\s*"([^"]+)"')
 ARTIFACTS = ('elements.txt', 'concepts.txt', 'expanded_concepts.txt')
 
+GZIP_MAGIC = b'\x1f\x8b'
+
+
+def read_artifact_text(path):
+    """roger's storage.write_object gzips these now; older artifacts
+    committed before that are still plain text, so detect and handle both."""
+    raw = path.read_bytes()
+    if raw[:2] == GZIP_MAGIC:
+        return gzip.decompress(raw).decode('utf-8')
+    return raw.decode('utf-8')
+
+
+def write_artifact_text(path, text):
+    """Always write gzip -- migration is also the chance to bring an old
+    plain-text artifact up to the current on-disk format."""
+    path.write_bytes(gzip.compress(text.encode('utf-8')))
+
 
 def install_alias(module_path):
     """Stand in for a legacy module, resolving class names against CURRENT."""
@@ -110,7 +128,7 @@ def field_drift_paths(paths):
     """
     drift = {}
     for path in paths:
-        obj = jsonpickle.decode(path.read_text())
+        obj = jsonpickle.decode(read_artifact_text(path))
         stack, seen = [obj], set()
         while stack:
             item = stack.pop()
@@ -137,7 +155,7 @@ def scan(root):
     print(f"{len(files)} artifact file(s) under {root}")
     found = set()
     for path in files:
-        found |= classes_in(path.read_text())
+        found |= classes_in(read_artifact_text(path))
     broken = broken_modules(found)
     for cls in sorted(found):
         module_path = cls.rpartition('.')[0]
@@ -180,7 +198,7 @@ def dead_modules(sample_paths):
     """
     dead, found = set(), set()
     for path in sample_paths:
-        found |= classes_in(path.read_text())
+        found |= classes_in(read_artifact_text(path))
     for cls in found:
         module_path = cls.rpartition('.')[0]
         if module_path in dead or module_path in sys.modules:
@@ -271,7 +289,7 @@ def restamp(root, sample=20, dry_run=False):
 
     changed = 0
     for i, path in enumerate(files, 1):
-        text = path.read_text()
+        text = read_artifact_text(path)
         new_text, unmapped = restamp_text(text, dead)
         if unmapped:
             raise SystemExit(
@@ -284,7 +302,7 @@ def restamp(root, sample=20, dry_run=False):
             # write-then-rename: a pod killed mid-write must not leave a
             # truncated artifact behind, and there are 150k of them
             tmp = path.with_name(path.name + '.restamp-tmp')
-            tmp.write_text(new_text)
+            write_artifact_text(tmp, new_text)
             os.replace(tmp, path)
         if changed % 5000 == 0:
             print(f"  {changed} rewritten ({i}/{len(files)} scanned)")
@@ -296,13 +314,13 @@ def restamp(root, sample=20, dry_run=False):
 def fix(root, dry_run=False):
     files = artifact_files(root)
     for module_path in broken_modules(
-            {c for p in files for c in classes_in(p.read_text())}):
+            {c for p in files for c in classes_in(read_artifact_text(p))}):
         print(f"aliasing legacy module {module_path}")
         install_alias(module_path)
 
     changed = 0
     for path in files:
-        text = path.read_text()
+        text = read_artifact_text(path)
         obj = jsonpickle.decode(text)
         fill_defaults(obj)
         rewritten = jsonpickle.encode(obj, indent=2)
@@ -311,7 +329,7 @@ def fix(root, dry_run=False):
         changed += 1
         print(f"{'would rewrite' if dry_run else 'rewrote'} {path}")
         if not dry_run:
-            path.write_text(rewritten)
+            write_artifact_text(path, rewritten)
     print(f"{changed} of {len(files)} file(s) needed migration")
     return changed
 
diff --git a/src/roger/core/storage.py b/src/roger/core/storage.py
index d10d0d6..82d1f06 100644
--- a/src/roger/core/storage.py
+++ b/src/roger/core/storage.py
@@ -5,6 +5,7 @@
 
 import os
 import glob
+import gzip
 import time
 import pathlib
 import pickle
@@ -86,9 +87,25 @@ def read_object(path, key=None):
         with open(file=path, mode="rb") as stream:
             obj = pickle.load(stream)
     elif path.endswith(".jsonl") or path.endswith('.txt'):
-        obj = read_data(path)
+        obj = read_gzip_or_plain_text(path) if not is_web(path) \
+            else read_data(path)
     return obj
 
+# gzip magic number -- distinguishes a compressed artifact from the plain
+# text ones already committed to lakefs before this was added, so both
+# read transparently and nothing downstream needs to change.
+GZIP_MAGIC = b'\x1f\x8b'
+
+def read_gzip_or_plain_text(path):
+    """ Read a local .txt/.jsonl artifact, decompressing it if it was
+    gzip-written by write_object; falls back to plain text for artifacts
+    written before compression was added. """
+    with open(path, 'rb') as stream:
+        raw = stream.read()
+    if raw[:2] == GZIP_MAGIC:
+        return gzip.decompress(raw).decode('utf-8')
+    return raw.decode('utf-8')
+
 def is_web (uri):
     """ The URI is a web URI (starts with http or https).
     :param uri: A URI """
@@ -122,8 +139,15 @@ def write_object (obj, path, key=None):
         with open (path, "wb") as stream:
             pickle.dump(obj, file=stream)
     elif path.endswith(".jsonl") or path.endswith('.txt'):
-        with open (path, "w", encoding="utf-8") as stream:
-            stream.write(obj)
+        # gzip -- these are the crawl-stage KG-answer artifacts, and the
+        # same repeated CURIEs/biolink categories/JSON keys compress
+        # 5-10x; that's the difference between fitting a large dataset's
+        # crawl output on the PVC and hitting ENOSPC mid-run. Same
+        # filename/extension as before so every glob pattern that finds
+        # these files by name still matches; read_object detects the
+        # gzip magic number so old uncompressed artifacts still read.
+        with open(path, "wb") as stream:
+            stream.write(gzip.compress(obj.encode('utf-8')))
     else:
         # Raise an exception if invalid.
         raise ValueError (f"Unrecognized extension: {path}")
diff --git a/tests/unit/test_storage_gzip_artifacts.py b/tests/unit/test_storage_gzip_artifacts.py
new file mode 100644
index 0000000..b0f37ac
--- /dev/null
+++ b/tests/unit/test_storage_gzip_artifacts.py
@@ -0,0 +1,27 @@
+"""write_object gzips .txt/.jsonl artifacts now -- the repeated CURIEs and
+biolink categories in crawl-stage KG-answer JSON compress 5-10x, which is
+the difference between a large dataset's crawl fitting on the PVC and
+hitting ENOSPC mid-run. read_object must still read artifacts committed
+before this was added.
+"""
+import gzip
+
+from roger.core import storage
+
+
+def test_txt_artifact_round_trips_through_gzip(tmp_path):
+    path = str(tmp_path / 'concepts.txt')
+    text = '{"id": "UMLS:C1"}' * 100
+
+    storage.write_object(text, path)
+
+    assert open(path, 'rb').read(2) == b'\x1f\x8b'
+    assert storage.read_object(path) == text
+
+
+def test_txt_artifact_backward_compat_with_plain_text(tmp_path):
+    path = tmp_path / 'concepts.txt'
+    text = '{"id": "UMLS:C1"}'
+    path.write_text(text, encoding='utf-8')
+
+    assert storage.read_object(str(path)) == text

From f78aec4462a071bf8392d178cd38e695516115f5 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Mon, 7 Sep 2026 08:25:51 -0400
Subject: [PATCH 05/13] perf(storage): drop gzip to level 6, level 9 was
 throttling crawl pods

gzip.compress's default (level 9) cost 3-4x the CPU of level 6 for the
same compressed size on this repetitive JSON -- measured identical
output bytes at both levels on realistic KG-answer-shaped data. Deadly
under a crawl task's thin CPU limit: bdc-recover's pod (250m limit, 4
crawl workers) was throttled 70% of scheduling periods after the gzip
change landed, cutting throughput from ~4 dirs/min to ~3 dirs/min.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/core/storage.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/src/roger/core/storage.py b/src/roger/core/storage.py
index 82d1f06..0f9f1cc 100644
--- a/src/roger/core/storage.py
+++ b/src/roger/core/storage.py
@@ -147,7 +147,11 @@ def write_object (obj, path, key=None):
         # these files by name still matches; read_object detects the
         # gzip magic number so old uncompressed artifacts still read.
         with open(path, "wb") as stream:
-            stream.write(gzip.compress(obj.encode('utf-8')))
+            # level 9 (gzip.compress's default) burned 3-4x the CPU of
+            # level 6 for the same ratio on this repetitive JSON -- deadly
+            # under a crawl task's thin CPU limit (measured 70% of periods
+            # throttled on a 250m limit with 4 crawl workers).
+            stream.write(gzip.compress(obj.encode('utf-8'), compresslevel=6))
     else:
         # Raise an exception if invalid.
         raise ValueError (f"Unrecognized extension: {path}")

From cd9ecc98394ef4e4e63e61e0ecdd67066fe8c7f0 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Mon, 7 Sep 2026 08:40:41 -0400
Subject: [PATCH 06/13] perf(crawl): give the crawl task its own cpu limit

crawl_file_workers threads do real CPU work now (TranQL response
handling, jsonpickle encode, gzip) -- measured a crawl pod throttled
70% of its scheduling periods on the chart's default cpu limit with 4
worker threads, cutting throughput to roughly a third of what it
should be. New indexing.crawl_cpu config (default "1", a full core)
plumbed through a generalized resource_override (memory_override is
now a thin wrapper over it) the same way annotate_memory already
overrides the crawl task's memory limit.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/config/__init__.py         |  6 +++++
 src/roger/config/config.yaml         |  1 +
 src/roger/tasks.py                   | 38 ++++++++++++++++++++++------
 tests/unit/test_tasks_incremental.py | 10 ++++++++
 4 files changed, 47 insertions(+), 8 deletions(-)

diff --git a/src/roger/config/__init__.py b/src/roger/config/__init__.py
index a986732..33ae4f1 100644
--- a/src/roger/config/__init__.py
+++ b/src/roger/config/__init__.py
@@ -239,6 +239,12 @@ class IndexingConfig(DictLike):
     # requests only queue.
     crawl_workers: int = 4
     crawl_file_workers: int = 4
+    # crawl_file_workers threads doing real work (TranQL fetches, jsonpickle
+    # encode, gzip) on the chart's default cpu limit throttled a crawl pod
+    # 70% of its scheduling periods, cutting throughput to a third. One full
+    # core gives each worker thread real headroom instead of fighting over
+    # a quarter of one.
+    crawl_cpu: str = "1"
     # by default skips node to element queries
     node_to_element_queries: dict = field(default_factory=lambda: {})
     element_mapping: str = ""
diff --git a/src/roger/config/config.yaml b/src/roger/config/config.yaml
index 42435ef..ed614ec 100644
--- a/src/roger/config/config.yaml
+++ b/src/roger/config/config.yaml
@@ -122,6 +122,7 @@ indexing:
   tranql_endpoint: "http://tranql-service/tranql/query?dynamic_id_resolution=true&asynchronous=false"
   crawl_workers: 4
   crawl_file_workers: 4
+  crawl_cpu: "1"
   node_to_element_queries:
     enabled: false
     cde:
diff --git a/src/roger/tasks.py b/src/roger/tasks.py
index 37f60f6..248e742 100755
--- a/src/roger/tasks.py
+++ b/src/roger/tasks.py
@@ -127,21 +127,35 @@ def get_executor_config(data_path='/opt/airflow/share/data'):
 
 
 def memory_override(limit: str, request: str = None) -> dict:
-    """executor_config bumping only this task's memory.
+    """executor_config bumping only this task's memory. See
+    resource_override -- this is kept as a thin wrapper since it is the
+    common case and already used at several call sites."""
+    return resource_override(memory_limit=limit, memory_request=request)
+
+
+def resource_override(memory_limit: str = None, memory_request: str = None,
+                      cpu_limit: str = None, cpu_request: str = None) -> dict:
+    """executor_config bumping only this task's cpu and/or memory.
 
     Everything else (image, volumes, env, service account) is inherited from
     the chart's worker pod template; this patches the 'base' container so one
     heavy task does not force the default up for every task. Keep request
-    well under limit: the namespace quota counts requests.memory and
-    limits.memory separately.
+    well under limit: the namespace quota counts requests.memory/cpu and
+    limits.memory/cpu separately.
     """
     from kubernetes.client import models as k8s
+    requests, limits = {}, {}
+    if memory_limit:
+        limits["memory"] = memory_limit
+        requests["memory"] = memory_request or "1Gi"
+    if cpu_limit:
+        limits["cpu"] = cpu_limit
+        requests["cpu"] = cpu_request or cpu_limit
     return {"pod_override": k8s.V1Pod(spec=k8s.V1PodSpec(containers=[
         k8s.V1Container(
             name="base",
             resources=k8s.V1ResourceRequirements(
-                requests={"memory": request or "1Gi"},
-                limits={"memory": limit}))]))}
+                requests=requests, limits=limits))]))}
 
 
 def init_lakefs_client(config: RogerConfig) -> LakeFsWrapper:
@@ -772,7 +786,7 @@ def create_python_task(dag, name, a_callable, func_kwargs=None,
                        external_repos=None, pass_conf=True,
                        no_output_files=False, no_input_files=False,
                        incremental_pull=True, clear_output_prefix=False,
-                       memory=None, resumable=False):
+                       memory=None, cpu=None, resumable=False):
     """ Create a python task.
     :param func_kwargs: additional arguments for callable.
     :param dag: dag to add task to.
@@ -793,6 +807,8 @@ def create_python_task(dag, name, a_callable, func_kwargs=None,
         vary run to run (the bulk-load CSVs) and would otherwise accumulate.
     :param memory: memory limit for this task's pod, e.g. '15Gi'. Omit to
         take the chart's worker default.
+    :param cpu: cpu limit for this task's pod, e.g. '1'. Omit to take the
+        chart's worker default.
     """
 
     if external_repos is None:
@@ -814,8 +830,9 @@ def create_python_task(dag, name, a_callable, func_kwargs=None,
         # executor_config example left commented; fill if needed
         "dag": dag,
     }
-    if memory:
-        python_operator_args["executor_config"] = memory_override(memory)
+    if memory or cpu:
+        python_operator_args["executor_config"] = resource_override(
+            memory_limit=memory, cpu_limit=cpu)
 
     if config.lakefs_config.enabled:
         pre_exec_conf = {
@@ -962,6 +979,11 @@ def create_pipeline_taskgroup(
             crawl_callable,
             # expands every concept through tranql, accumulating answers
             memory=configparam.annotation.annotate_memory,
+            # crawl_file_workers threads doing real CPU work (TranQL
+            # fetches, jsonpickle encode, gzip) on the chart's thin default
+            # cpu limit throttled a crawl pod 70% of its scheduling
+            # periods, cutting throughput to a third.
+            cpu=configparam.indexing.crawl_cpu,
             pass_conf=False)
         crawl_task.set_upstream(annotate_task)
 
diff --git a/tests/unit/test_tasks_incremental.py b/tests/unit/test_tasks_incremental.py
index 7b6bbc3..9d92fce 100644
--- a/tests/unit/test_tasks_incremental.py
+++ b/tests/unit/test_tasks_incremental.py
@@ -427,6 +427,16 @@ def test_memory_override_patches_base_container():
     assert container.resources.requests == {"memory": "1Gi"}
 
 
+def test_resource_override_patches_cpu_and_memory_together():
+    pytest.importorskip("kubernetes")
+    cfg = tasks.resource_override(memory_limit="15Gi", cpu_limit="1")
+    container = cfg["pod_override"].spec.containers[0]
+    assert container.resources.limits == {"memory": "15Gi", "cpu": "1"}
+    # cpu request defaults to the limit (no separate cheap-request case,
+    # unlike memory) since cpu limits are compressible, not a quota risk
+    assert container.resources.requests == {"memory": "1Gi", "cpu": "1"}
+
+
 def test_es_taskgroup_pulls_crawl_outputs_only(monkeypatch, lakefs_env):
     """index_variables must read crawl's expanded elements.txt, not
     annotate's: only the crawl copy carries KG-derived optional_terms, and

From f00c10f9682eaf24583c85a695f589190339115f Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Mon, 7 Sep 2026 08:59:28 -0400
Subject: [PATCH 07/13] perf(crawl): bump crawl_cpu default to 4 cores

1 core still left 4 crawl_file_workers threads sharing it. Match cores
1:1 with worker threads instead. Plenty of quota headroom in both live
namespaces to absorb it.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/config/__init__.py | 8 ++++----
 src/roger/config/config.yaml | 2 +-
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/src/roger/config/__init__.py b/src/roger/config/__init__.py
index 33ae4f1..ddfe1d2 100644
--- a/src/roger/config/__init__.py
+++ b/src/roger/config/__init__.py
@@ -241,10 +241,10 @@ class IndexingConfig(DictLike):
     crawl_file_workers: int = 4
     # crawl_file_workers threads doing real work (TranQL fetches, jsonpickle
     # encode, gzip) on the chart's default cpu limit throttled a crawl pod
-    # 70% of its scheduling periods, cutting throughput to a third. One full
-    # core gives each worker thread real headroom instead of fighting over
-    # a quarter of one.
-    crawl_cpu: str = "1"
+    # 70% of its scheduling periods, cutting throughput to a third. Match
+    # crawl_file_workers 1:1 with cores so each worker thread gets its own,
+    # instead of 4 threads fighting over a fraction of one.
+    crawl_cpu: str = "4"
     # by default skips node to element queries
     node_to_element_queries: dict = field(default_factory=lambda: {})
     element_mapping: str = ""
diff --git a/src/roger/config/config.yaml b/src/roger/config/config.yaml
index ed614ec..4d802ae 100644
--- a/src/roger/config/config.yaml
+++ b/src/roger/config/config.yaml
@@ -122,7 +122,7 @@ indexing:
   tranql_endpoint: "http://tranql-service/tranql/query?dynamic_id_resolution=true&asynchronous=false"
   crawl_workers: 4
   crawl_file_workers: 4
-  crawl_cpu: "1"
+  crawl_cpu: "4"
   node_to_element_queries:
     enabled: false
     cde:

From b96c9f9fb1d7c4e3ba19bd6581a86f22acc5d8e9 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Mon, 7 Sep 2026 10:43:04 -0400
Subject: [PATCH 08/13] perf(crawl): skip files already crawled by an earlier
 try

crawl_one_file had no resume check, unlike annotation_is_complete for
annotate -- a killed/retried crawl re-walked every file from the start
of the sorted list. TranQL's response cache makes redoing an
already-crawled file cheap, but not free, and a large dataset's
never-before-seen files queue behind all of that free-but-not-instant
rework before any real new progress resumes. Mirrors the annotate
pattern: crawl_output_path/crawl_is_complete check expanded_concepts.txt
existence, crawl_tranql filters pending files down before submitting
them to the pool.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/pipelines/base.py           | 43 ++++++++++++++++++++++---
 tests/unit/test_crawl_file_workers.py | 45 +++++++++++++++++++++++++++
 2 files changed, 84 insertions(+), 4 deletions(-)

diff --git a/src/roger/pipelines/base.py b/src/roger/pipelines/base.py
index f930254..fc83edf 100644
--- a/src/roger/pipelines/base.py
+++ b/src/roger/pipelines/base.py
@@ -388,6 +388,30 @@ def annotation_is_complete(cls, parse_file, output_data_path):
                    for path in cls.annotation_output_paths(parse_file,
                                                            output_data_path))
 
+    @staticmethod
+    def crawl_output_path(concept_file, output_data_path=None):
+        "The expanded_concepts.txt file crawl_one_file writes for a concept_file"
+        data_set_name = os.path.split(os.path.dirname(concept_file))[-1]
+        output_file_name = os.path.join(data_set_name, 'expanded_concepts.txt')
+        if not output_data_path:
+            return storage.dug_expanded_concepts_path(output_file_name)
+        return os.path.join(output_data_path, output_file_name)
+
+    @classmethod
+    def crawl_is_complete(cls, concept_file, output_data_path):
+        """True if this file's crawl output is already fully written.
+
+        crawl_one_file has no partial-write hazard like annotation's two
+        files do -- expanded_concepts.txt is written in one call -- so
+        existence and non-empty is the whole check. Without this, a resumed
+        try re-crawls every file from scratch: TranQL's response cache makes
+        the already-done ones cheap, but not free, and a large dataset's
+        never-before-seen files still queue behind all the free-but-not-
+        instant redone work before any real progress resumes.
+        """
+        path = cls.crawl_output_path(concept_file, output_data_path)
+        return os.path.isfile(path) and os.path.getsize(path) > 0
+
     def annotate_one_file(self, parse_file, parser, output_data_path,
                           index=0, total=0):
         "Parse and annotate a single input file, writing pickles for it"
@@ -1223,12 +1247,23 @@ def crawl_tranql(self, to_string=False, concept_files=None,
         log.info("Clearing expanded concepts dir: %s", expanded_concepts_dir)
         storage.clear_dir(expanded_concepts_dir)
 
+        pending = [f for f in concept_files
+                  if not self.crawl_is_complete(f, output_data_path)]
+        skipped = len(concept_files) - len(pending)
+        if skipped:
+            log.info("Resuming: %d of %d files already crawled, %d to go",
+                     skipped, len(concept_files), len(pending))
+
+        if not pending:
+            output_log = self.log_stream.getvalue() if to_string else ''
+            return output_log
+
         workers = max(1, int(self.config.indexing.crawl_file_workers))
-        workers = min(workers, len(concept_files)) or 1
+        workers = min(workers, len(pending)) or 1
         log.info("Crawling Dug Concepts, found %d file(s) with %d worker(s).",
-                 len(concept_files), workers)
+                 len(pending), workers)
         if workers == 1:
-            for file_ in concept_files:
+            for file_ in pending:
                 self.crawl_one_file(file_, output_data_path)
         else:
             # Files are independent: each decodes its own concepts, expands
@@ -1238,7 +1273,7 @@ def crawl_tranql(self, to_string=False, concept_files=None,
                                     thread_name_prefix='crawl') as pool:
                 futures = [pool.submit(self.crawl_one_file, file_,
                                        output_data_path)
-                           for file_ in concept_files]
+                           for file_ in pending]
                 # surface the first failure rather than letting the pool
                 # swallow it; a raised exception means that file produced
                 # nothing and the task must not report success
diff --git a/tests/unit/test_crawl_file_workers.py b/tests/unit/test_crawl_file_workers.py
index 7a8c4b4..0e2914f 100644
--- a/tests/unit/test_crawl_file_workers.py
+++ b/tests/unit/test_crawl_file_workers.py
@@ -6,6 +6,7 @@
 worker count, and a file that raises fails the task rather than quietly
 producing no output.
 """
+import os
 import threading
 import types
 
@@ -21,6 +22,9 @@ def make_pipeline(workers, crawl_one):
         indexing=types.SimpleNamespace(crawl_file_workers=workers))
     pipeline.log_stream = types.SimpleNamespace(getvalue=lambda: '')
     pipeline.crawl_one_file = crawl_one
+    # nothing exists yet in these tests -- every file is pending, same as
+    # the pre-skip-check behavior this test suite was written against
+    pipeline.crawl_is_complete = lambda f, output_data_path: False
     pipeline.crawl_tranql = types.MethodType(
         DugPipeline.crawl_tranql.__wrapped__
         if hasattr(DugPipeline.crawl_tranql, '__wrapped__')
@@ -81,3 +85,44 @@ def crawl_one(file_, output_data_path=None):
     pipeline.crawl_tranql(concept_files=FILES[:2],
                           output_data_path=str(tmp_path))
     assert len(threads) <= 2
+
+
+def test_already_crawled_files_are_skipped(monkeypatch, tmp_path):
+    """A resumed try must not redo files an earlier try already crawled --
+    TranQL's response cache makes that cheap, not free, and it stands
+    between a large dataset and any real new progress."""
+    import roger.pipelines.base as base
+
+    seen = []
+
+    def crawl_one(file_, output_data_path=None):
+        seen.append(file_)
+
+    monkeypatch.setattr(base.storage, 'clear_dir', lambda *a, **k: None)
+    pipeline = make_pipeline(4, crawl_one)
+    already_done = set(FILES[:5])
+    pipeline.crawl_is_complete = lambda f, output_data_path: f in already_done
+
+    pipeline.crawl_tranql(concept_files=list(FILES),
+                          output_data_path=str(tmp_path))
+
+    assert sorted(seen) == sorted(set(FILES) - already_done)
+
+
+def test_crawl_is_complete_checks_the_real_pipeline(tmp_path):
+    """Exercise the actual DugPipeline method, not just the stand-in used
+    above -- this is what would have caught crawl_output_path drifting
+    from where crawl_concepts actually writes."""
+    from roger.pipelines.base import DugPipeline
+
+    concept_file = "/in/phs000123.v1.data_dict/concepts.txt"
+    output_data_path = str(tmp_path)
+
+    assert DugPipeline.crawl_is_complete(concept_file, output_data_path) is False
+
+    path = DugPipeline.crawl_output_path(concept_file, output_data_path)
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, 'w') as f:
+        f.write('{}')
+
+    assert DugPipeline.crawl_is_complete(concept_file, output_data_path) is True

From 213fe63343e9d8ca8a273705b238f640a2040ce8 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Mon, 7 Sep 2026 11:26:13 -0400
Subject: [PATCH 09/13] tools: add gzip_artifacts.py to retroactively compress
 old annotation output

crawl_bdc-parent died with ENOSPC downloading its own input: bdc-parent's
elements.txt/concepts.txt were annotated before storage.write_object
started gzipping .txt/.jsonl artifacts, so they're still ~3MB/dir raw --
61,597 dirs is ~186GB before crawl writes a single byte of its own
(gzipped) output. Pure byte-level gzip, no jsonpickle decode/re-encode
like migrate_pickled_classes.py does -- content is untouched, just
smaller, and there's no class-path migration needed here.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 scripts/gzip_artifacts.py | 126 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 126 insertions(+)
 create mode 100644 scripts/gzip_artifacts.py

diff --git a/scripts/gzip_artifacts.py b/scripts/gzip_artifacts.py
new file mode 100644
index 0000000..07b6f39
--- /dev/null
+++ b/scripts/gzip_artifacts.py
@@ -0,0 +1,126 @@
+#!/usr/bin/env python
+"""Gzip-compress annotate/crawl artifacts committed before storage.py's
+write_object started gzipping them (see roger.core.storage).
+
+A dataset annotated before that fix has its elements.txt/concepts.txt sitting
+in lakefs uncompressed -- ~3MB/dir raw for a typical dbGaP data dict pair.
+crawl's own output is gzipped going forward, but crawl still has to
+*download* that old, uncompressed annotation output as input first: for
+bdc-parent's 61,597 dirs that is ~186GB of local disk before crawl writes a
+single byte, which is most of what blew the crawl task's 200G PVC.
+
+Unlike migrate_pickled_classes.py this does no jsonpickle decode/re-encode --
+pure byte-level gzip, so the serialized content is untouched, just smaller.
+
+    lakectl local clone lakefs://<repo>/<branch>/annotate_and_index/<task>/ ./out
+    python scripts/gzip_artifacts.py ./out            # compress in place
+    lakectl local commit ./out -m "gzip annotation output"
+
+    python scripts/gzip_artifacts.py --dry-run ./out  # report only
+    python scripts/gzip_artifacts.py --self-check     # no dir needed
+"""
+
+import argparse
+import gzip
+import sys
+from pathlib import Path
+
+ARTIFACTS = ('elements.txt', 'concepts.txt', 'expanded_concepts.txt')
+GZIP_MAGIC = b'\x1f\x8b'
+# level 6 matches storage.py's write_object -- level 9 (gzip's default)
+# burned 3-4x the cpu for the same ratio on this repetitive JSON.
+COMPRESSLEVEL = 6
+
+
+def artifact_files(root):
+    return sorted(p for p in Path(root).rglob('*.txt') if p.name in ARTIFACTS)
+
+
+def compress_file(path, dry_run=False):
+    """Returns (raw_size, compressed_size) if compressed, None if skipped
+    (already gzip or empty)."""
+    raw = path.read_bytes()
+    if not raw or raw[:2] == GZIP_MAGIC:
+        return None
+    compressed = gzip.compress(raw, compresslevel=COMPRESSLEVEL)
+    if not dry_run:
+        tmp = path.with_name(path.name + '.gzip-tmp')
+        tmp.write_bytes(compressed)
+        tmp.replace(path)
+    return len(raw), len(compressed)
+
+
+def run(root, dry_run=False):
+    files = artifact_files(root)
+    total_raw = total_compressed = 0
+    changed = skipped = 0
+    for i, path in enumerate(files, 1):
+        result = compress_file(path, dry_run=dry_run)
+        if result is None:
+            skipped += 1
+            continue
+        raw_size, compressed_size = result
+        total_raw += raw_size
+        total_compressed += compressed_size
+        changed += 1
+        if changed % 5000 == 0:
+            print(f"  {changed} compressed ({i}/{len(files)} scanned)")
+    verb = "would compress" if dry_run else "compressed"
+    print(f"{verb} {changed} of {len(files)} file(s), "
+          f"{skipped} already gzip or empty")
+    if total_raw:
+        print(f"{total_raw / 1e9:.2f}GB -> {total_compressed / 1e9:.2f}GB "
+              f"({total_raw / total_compressed:.1f}x)")
+    return changed
+
+
+def self_check():
+    "Round-trip a small payload; fails loudly if compress_file regresses."
+    import tempfile
+    payload = b'{"id": "UMLS:C1"}' * 500
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / 'concepts.txt'
+        path.write_bytes(payload)
+
+        result = compress_file(path, dry_run=True)
+        assert result is not None, "dry-run should report a would-be change"
+        assert path.read_bytes() == payload, "dry-run must not touch the file"
+
+        result = compress_file(path)
+        assert result is not None
+        raw_size, compressed_size = result
+        assert raw_size == len(payload)
+        assert compressed_size < raw_size, "should have actually shrunk"
+
+        with open(path, 'rb') as f:
+            assert f.read(2) == GZIP_MAGIC
+        assert gzip.decompress(path.read_bytes()) == payload
+
+        # idempotent: running again on an already-gzipped file is a no-op
+        assert compress_file(path) is None
+        assert gzip.decompress(path.read_bytes()) == payload
+
+    print("self-check ok")
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__,
+                                     formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument('root', nargs='?', help='directory to compress in place')
+    parser.add_argument('--dry-run', action='store_true',
+                        help='report what would change without writing')
+    parser.add_argument('--self-check', action='store_true',
+                        help='round-trip a synthetic payload, no dir needed')
+    args = parser.parse_args()
+
+    if args.self_check:
+        self_check()
+        return
+    if not args.root:
+        parser.error("root directory required unless --self-check")
+    run(args.root, dry_run=args.dry_run)
+
+
+if __name__ == '__main__':
+    sys.exit(main())

From 2883bafbddf74a6b12498bbb189c502d3b7692f9 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Thu, 10 Sep 2026 14:27:27 -0400
Subject: [PATCH 10/13] perf(crawl): make crawl resumable now that it has a
 skip check

crawl_is_complete (b96c9f9) gave crawl the same per-file skip check
annotate has, but crawl_task never got resumable=True -- it was turned
off in e5c5a5e when crawl had no such check and retained output was
pure disk cost. With the skip check now in place, keeping crawl's
output on failure lets a retry resume instead of redoing the whole
dataset.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/tasks.py | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/src/roger/tasks.py b/src/roger/tasks.py
index 248e742..4605760 100755
--- a/src/roger/tasks.py
+++ b/src/roger/tasks.py
@@ -794,11 +794,13 @@ def create_python_task(dag, name, a_callable, func_kwargs=None,
     :param a_callable: The code to run in this task.
     :param no_input_files: skip the lakefs input download entirely.
     :param resumable: keep the output dir when the task fails, so a retry
-        can pick up where it left off. Only true for annotate, which is
-        the one task with a skip check (annotation_is_complete). For the
-        others the retained output is never reused and is pure disk cost:
-        three failed crawls held 47GB and filled the shared volume, which
-        is what made them fail in the first place.
+        can pick up where it left off. True for annotate and crawl, which
+        have skip checks (annotation_is_complete, crawl_is_complete) that
+        make the retained output cheap to detect and reuse. make_kgx has no
+        such check, so for it the retained output would be pure disk cost:
+        three failed crawls once held 47GB and filled the shared volume,
+        back before crawl had a skip check of its own -- keeping its output
+        made that failure permanent instead of transient.
     :param incremental_pull: when False the task always downloads its full
         inputs even if the dag runs with incremental=True (needed for tasks
         that rebuild state from scratch, e.g. ES indexing after a wipe).
@@ -984,6 +986,10 @@ def create_pipeline_taskgroup(
             # cpu limit throttled a crawl pod 70% of its scheduling
             # periods, cutting throughput to a third.
             cpu=configparam.indexing.crawl_cpu,
+            # crawl_is_complete now gives crawl the same skip check that
+            # justified resumable for annotate; retained output is reused,
+            # not dead weight (see crawl_tranql's pending-files filter)
+            resumable=True,
             pass_conf=False)
         crawl_task.set_upstream(annotate_task)
 

From 0d2a8d8c80cf162b368c104577646e0034fef0e8 Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Thu, 10 Sep 2026 14:32:23 -0400
Subject: [PATCH 11/13] perf(crawl): bump crawl_file_workers and crawl_cpu to 8

Cache is warm for most of bdc-parent's remaining files after the
mid-run eviction/restart, so file-level parallelism (I/O-wait bound on
cache hits, CPU-bound on misses) has more room to help than before.
crawl_cpu matches 1:1 so each worker thread still gets its own core.

crawl_workers stays at 4 for now -- the product with crawl_file_workers
(32) exceeds tranql's 8 gunicorn workers, but most requests are cache
hits at this point in the run so tranql itself sees little of it.
Revisit if the remaining, mostly-uncached files start queuing.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/config/config.yaml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/roger/config/config.yaml b/src/roger/config/config.yaml
index 4d802ae..2b95c2f 100644
--- a/src/roger/config/config.yaml
+++ b/src/roger/config/config.yaml
@@ -121,8 +121,8 @@ indexing:
     "phen_to_anat": ["phenotypic_feature", "anatomical_entity"]
   tranql_endpoint: "http://tranql-service/tranql/query?dynamic_id_resolution=true&asynchronous=false"
   crawl_workers: 4
-  crawl_file_workers: 4
-  crawl_cpu: "4"
+  crawl_file_workers: 8
+  crawl_cpu: "8"
   node_to_element_queries:
     enabled: false
     cde:

From f6ed27d263d5b4c508592fa3bffb9fc6233034de Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Mon, 14 Sep 2026 10:01:19 -0400
Subject: [PATCH 12/13] fix(crawl): skip a corrupted file instead of failing
 the whole task

bdc-parent's crawl_bdc-parent hit a null byte inside one committed
annotate-output file, crashing jsonpickle/yaml decode in
crawl_one_file. Input is re-pulled fresh from lakefs on every retry
(only crawl's output survives via resumable), so a task-ending
exception here means the same file kills every future retry forever --
not transient like the CPU/eviction failures before it.

crawl_one_file's per-file work now runs through crawl_one_file_safe,
which catches exceptions, logs them, and records the file to a
.crawl_failed_files.json manifest (dotfile, same convention as
tasks.py's REMOVED_FILES_MANIFEST, so storage.py's glob skips it)
instead of propagating. One bad file out of 61,597 no longer blocks
the rest; the manifest gives a starting point to investigate the
source data afterward.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/pipelines/base.py           | 35 +++++++++++++++++++++-----
 tests/unit/test_crawl_file_workers.py | 36 ++++++++++++++++++++-------
 2 files changed, 56 insertions(+), 15 deletions(-)

diff --git a/src/roger/pipelines/base.py b/src/roger/pipelines/base.py
index fc83edf..3151309 100644
--- a/src/roger/pipelines/base.py
+++ b/src/roger/pipelines/base.py
@@ -1,6 +1,7 @@
 "Base class for implementing a dataset annotate, crawl, and index pipeline"
 
 import os
+import json
 import random
 import threading
 import time
@@ -42,6 +43,10 @@
 
 log = get_logger()
 
+# Dotfile so storage.py's *.json / **/*.json globs over pulled task outputs
+# skip it, same convention as tasks.py's REMOVED_FILES_MANIFEST.
+CRAWL_FAILED_FILES_MANIFEST = ".crawl_failed_files.json"
+
 class PipelineException(Exception):
     "Exception raised from DugPipeline and related classes"
 
@@ -1262,24 +1267,42 @@ def crawl_tranql(self, to_string=False, concept_files=None,
         workers = min(workers, len(pending)) or 1
         log.info("Crawling Dug Concepts, found %d file(s) with %d worker(s).",
                  len(pending), workers)
+
+        # A single corrupt input file (e.g. a null byte from a torn write)
+        # must not sink the other tens of thousands. list.append is atomic
+        # under the GIL, so no lock is needed across worker threads.
+        failed = []
+
+        def crawl_one_file_safe(file_):
+            try:
+                self.crawl_one_file(file_, output_data_path)
+            except Exception as e:
+                log.error("Skipping %s, crawl failed: %s", file_, e)
+                failed.append({"file": file_, "error": str(e)})
+
         if workers == 1:
             for file_ in pending:
-                self.crawl_one_file(file_, output_data_path)
+                crawl_one_file_safe(file_)
         else:
             # Files are independent: each decodes its own concepts, expands
             # them and writes its own output dir. The work is nearly all
             # http wait on TranQL, so threads scale it despite the GIL.
             with ThreadPoolExecutor(max_workers=workers,
                                     thread_name_prefix='crawl') as pool:
-                futures = [pool.submit(self.crawl_one_file, file_,
-                                       output_data_path)
+                futures = [pool.submit(crawl_one_file_safe, file_)
                            for file_ in pending]
-                # surface the first failure rather than letting the pool
-                # swallow it; a raised exception means that file produced
-                # nothing and the task must not report success
                 for future in futures:
                     future.result()
 
+        if failed:
+            manifest_path = os.path.join(
+                output_data_path or storage.dug_expanded_concepts_path(""),
+                CRAWL_FAILED_FILES_MANIFEST)
+            with open(manifest_path, 'w') as f:
+                json.dump(failed, f, indent=2)
+            log.warning("%d of %d file(s) failed to crawl, see %s",
+                       len(failed), len(pending), manifest_path)
+
         output_log = self.log_stream.getvalue() if to_string else ''
         return output_log
 
diff --git a/tests/unit/test_crawl_file_workers.py b/tests/unit/test_crawl_file_workers.py
index 0e2914f..e9e840b 100644
--- a/tests/unit/test_crawl_file_workers.py
+++ b/tests/unit/test_crawl_file_workers.py
@@ -1,11 +1,14 @@
-"""crawl_tranql threads whole input files; failures must not be swallowed.
+"""crawl_tranql threads whole input files; one bad file must not sink the rest.
 
 The dir loop is where the bulk of crawl concurrency comes from -- one file
-per annotated input, tens of thousands of them for a dbGaP dataset. Two
-things must hold: every file gets processed exactly once regardless of
-worker count, and a file that raises fails the task rather than quietly
-producing no output.
+per annotated input, tens of thousands of them for a dbGaP dataset. Every
+file gets processed exactly once regardless of worker count, and a file
+that raises (e.g. a corrupted input) is logged to a manifest and skipped
+rather than failing the whole task -- input is re-pulled fresh from lakefs
+on every retry, so a task-ending exception here means the same file kills
+every future retry too.
 """
+import json
 import os
 import threading
 import types
@@ -55,18 +58,33 @@ def crawl_one(file_, output_data_path=None):
     assert len(seen) == len(FILES)
 
 
-def test_one_bad_file_fails_the_task(monkeypatch, tmp_path):
+def test_one_bad_file_is_skipped_not_fatal(monkeypatch, tmp_path):
+    """A corrupted input file must not sink the other pending files, and
+    must not fail the task -- input is re-pulled fresh from lakefs on every
+    retry, so a fatal exception here would fail identically forever."""
     import roger.pipelines.base as base
 
+    seen = []
+    lock = threading.Lock()
+
     def crawl_one(file_, output_data_path=None):
         if file_.endswith("file7/concepts.txt"):
             raise ValueError("bad pickle")
+        with lock:
+            seen.append(file_)
 
     monkeypatch.setattr(base.storage, 'clear_dir', lambda *a, **k: None)
     pipeline = make_pipeline(4, crawl_one)
-    with pytest.raises(ValueError, match="bad pickle"):
-        pipeline.crawl_tranql(concept_files=list(FILES),
-                              output_data_path=str(tmp_path))
+    pipeline.crawl_tranql(concept_files=list(FILES),
+                          output_data_path=str(tmp_path))
+
+    bad_file = "/in/file7/concepts.txt"
+    assert sorted(seen) == sorted(set(FILES) - {bad_file})
+
+    manifest_path = os.path.join(str(tmp_path), base.CRAWL_FAILED_FILES_MANIFEST)
+    with open(manifest_path) as f:
+        failed = json.load(f)
+    assert failed == [{"file": bad_file, "error": "bad pickle"}]
 
 
 def test_worker_count_is_capped_by_file_count(monkeypatch, tmp_path):

From 9f6d5a7e4979b11dfc22c6b327355cf88ba67a9b Mon Sep 17 00:00:00 2001
From: YaphetKG <kebedey@renci.org>
Date: Tue, 15 Sep 2026 10:44:49 -0400
Subject: [PATCH 13/13] fix(indexing): size index_variables_task, it OOMKilled
 on bdc-recover

index_bdc-recover_variables OOMKilled twice at the chart's bare 2Gi
default -- create_es_taskgroup never gave any ES indexing task a
memory override, unlike annotate/make_kgx/crawl. index_bdc-recover_concepts
runs fine at the same limit, so it's specifically the variables
elements that are large for this dataset. Reuses annotation.annotate_memory,
the existing knob for "this task holds a chunk of the dataset in memory",
same as make_kgx and crawl.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/roger/tasks.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/src/roger/tasks.py b/src/roger/tasks.py
index 4605760..e64b2fd 100755
--- a/src/roger/tasks.py
+++ b/src/roger/tasks.py
@@ -1064,6 +1064,10 @@ def full_pull(*paths):
                     configparam=configparam,
                     method_name='index_variables',
                     **kwargs),
+            # decodes one elements.txt at a time, but a single file can be
+            # large for a big dataset; index_bdc-recover_variables OOMKilled
+            # at the chart's 2Gi default while its concepts sibling did not
+            memory=configparam.annotation.annotate_memory,
             **full_pull(crawl_path))
 
         validate_index_variables_task = create_python_task(