Skip to content

Commit 29bbc99

Browse files
even-weiclaude
even-wei
and
claude
committed
fix(cli): address review feedback on per_node.db emitter
Critical fixes: - primary_key divergence from DbtAdapter.get_model: derive primary_key by scanning columns in catalog order (which matches warehouse column order from `dbt docs generate`) rather than manifest child_map iteration order. Adds tests with multiple unique tests to pin the tiebreaker. - Scratch tempdir leaked on every cloud upload failure path. Move creation to right before the emit block, wrap emit+upload in try/finally, and unconditionally `shutil.rmtree` the scratch. Adds tests covering HTTP 500 on per_node.db PUT and get_upload_urls_by_session_id raising. - Scale-fixture path was hardcoded to a developer workstation, making the scale class dead code for everyone else. Read RECCE_SCALE_FIXTURE_DIR with a committed-fallback at tests/fixtures/jaffle-shop-expand. Warnings: - Dropped `PRAGMA journal_mode = WAL` on the writer. One-shot writers that are immediately uploaded as a single file must be self-contained at close time; WAL would require an explicit truncating checkpoint for that and offers no benefit without concurrent readers. Test now asserts rollback journal mode and that no -wal/-shm sidecars survive. - Removed `assert per_node_scratch is not None` (stripped under `python -O`) in favor of structurally guaranteeing the assignment inside `if is_cloud`. - Added a rollback test for PerNodeDbWriter.__exit__. - Added a CLI test for the emit-failure graceful-degradation path. Docstring: dropped hard-coded line numbers pointing at DbtAdapter.get_model. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: even-wei <evenwei@infuseai.io>
1 parent 8e04a9e commit 29bbc99

4 files changed

Lines changed: 571 additions & 178 deletions

File tree

‎recce/cli.py‎

Lines changed: 144 additions & 140 deletions
Original file line numberDiff line numberDiff line change
@@ -350,17 +350,10 @@ def init(cache_db, **kwargs):
350350
cloud_client = None
351351
cloud_org_id = None
352352
cloud_project_id = None
353-
# per_node.db is a throwaway per-task artifact (each cloud task regenerates
354-
# it fresh from the uploaded manifest/catalog). It lives in its own tempdir
355-
# that is cleaned up after a successful upload. cll_cache.db, by contrast,
356-
# persists at ~/.recce/cll_cache.db to serve cross-session warm-cache reuse.
357-
per_node_scratch: Optional[Path] = None
358353

359354
if is_cloud:
360355
from recce.util.recce_cloud import RecceCloud, RecceCloudException
361356

362-
per_node_scratch = Path(tempfile.mkdtemp(prefix="recce-per-node-"))
363-
364357
cloud_token = kwargs.get("cloud_token") or kwargs.get("api_token")
365358
if not cloud_token:
366359
console.print("[[red]Error[/red]] --cloud requires --cloud-token or --api-token (or GITHUB_TOKEN env var).")
@@ -687,155 +680,166 @@ def _stream_download_to_file(url: str, dest: Path) -> int:
687680

688681
# In cloud mode, emit per_node.db — a pure-artifact SQLite that Cloud
689682
# streams to serve lineage without proxying to an ephemeral Recce instance.
690-
per_node_db_path: Optional[Path] = None
683+
# The scratch dir is always cleaned up, even on upload failure, so
684+
# long-lived Cloud deploys don't accumulate recce-per-node-* directories
685+
# in /tmp on retries.
691686
if is_cloud:
692-
assert per_node_scratch is not None
693-
per_node_db_path = per_node_scratch / "per_node.db"
694-
console.print("\n[bold]Emitting per-node SQLite...[/bold]")
695-
t_pn_start = time.perf_counter()
687+
per_node_scratch = Path(tempfile.mkdtemp(prefix="recce-per-node-"))
696688
try:
697-
with PerNodeDbWriter(per_node_db_path) as writer:
698-
writer.write_meta(
699-
schema_version=str(PER_NODE_DB_SCHEMA_VERSION),
700-
session_id=session_id or "",
701-
recce_version=__version__,
702-
generated_at=str(int(time.time())),
703-
)
704-
envs_to_emit = [
705-
("current", dbt_adapter.curr_manifest, dbt_adapter.curr_catalog),
706-
("base", dbt_adapter.base_manifest, dbt_adapter.base_catalog),
707-
]
708-
for env_name, manifest, catalog in envs_to_emit:
709-
if manifest is None:
710-
continue
711-
manifest_dict = manifest.to_dict() if hasattr(manifest, "to_dict") else manifest
712-
catalog_dict = (
713-
catalog.to_dict() if (catalog is not None and hasattr(catalog, "to_dict")) else catalog
714-
)
715-
node_rows, column_rows, edge_rows, test_rows = extract_rows_from_artifacts(
716-
manifest_dict, catalog_dict, env_name
689+
per_node_db_path: Optional[Path] = per_node_scratch / "per_node.db"
690+
console.print("\n[bold]Emitting per-node SQLite...[/bold]")
691+
t_pn_start = time.perf_counter()
692+
try:
693+
with PerNodeDbWriter(per_node_db_path) as writer:
694+
writer.write_meta(
695+
schema_version=str(PER_NODE_DB_SCHEMA_VERSION),
696+
session_id=session_id or "",
697+
recce_version=__version__,
698+
generated_at=str(int(time.time())),
717699
)
718-
writer.write_nodes(node_rows)
719-
writer.write_columns(column_rows)
720-
writer.write_edges(edge_rows)
721-
writer.write_tests(test_rows)
722-
pn_elapsed = time.perf_counter() - t_pn_start
723-
pn_size_mb = per_node_db_path.stat().st_size / 1024 / 1024
724-
console.print(
725-
f" per_node.db saved to [bold]{per_node_db_path}[/bold] ({pn_size_mb:.1f} MB, {pn_elapsed:.1f}s)"
726-
)
727-
except Exception as e:
728-
logger.warning("[recce init] Failed to emit per_node.db: %s", e)
729-
console.print(f" [[yellow]Warning[/yellow]] Failed to emit per_node.db: {e}")
730-
per_node_db_path = None
731-
732-
# Upload results to Cloud if in cloud mode
733-
if is_cloud and cloud_client:
734-
console.print("\n[bold]Uploading results to Cloud...[/bold]")
735-
upload_failures = []
736-
upload_succeeded = False
737-
try:
738-
upload_urls = cloud_client.get_upload_urls_by_session_id(cloud_org_id, cloud_project_id, session_id)
739-
740-
# Upload CLL map
741-
cll_map_upload_url = upload_urls.get("cll_map_url")
742-
if cll_map_upload_url and cll_map_path.is_file():
743-
try:
744-
with open(cll_map_path, "rb") as f:
745-
resp = requests.put(
746-
cll_map_upload_url,
747-
data=f,
748-
headers={"Content-Type": "application/json"},
749-
timeout=_UPLOAD_TIMEOUT,
700+
envs_to_emit = [
701+
("current", dbt_adapter.curr_manifest, dbt_adapter.curr_catalog),
702+
("base", dbt_adapter.base_manifest, dbt_adapter.base_catalog),
703+
]
704+
for env_name, manifest, catalog in envs_to_emit:
705+
if manifest is None:
706+
continue
707+
manifest_dict = manifest.to_dict() if hasattr(manifest, "to_dict") else manifest
708+
catalog_dict = (
709+
catalog.to_dict() if (catalog is not None and hasattr(catalog, "to_dict")) else catalog
750710
)
751-
if resp.status_code in (200, 204):
752-
console.print(f" Uploaded cll_map.json ({cll_map_path.stat().st_size / 1024 / 1024:.1f} MB)")
753-
else:
754-
upload_failures.append("cll_map.json")
755-
console.print(
756-
f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: HTTP {resp.status_code}"
711+
node_rows, column_rows, edge_rows, test_rows = extract_rows_from_artifacts(
712+
manifest_dict, catalog_dict, env_name
757713
)
758-
except requests.RequestException as e:
759-
upload_failures.append("cll_map.json")
760-
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: {e}")
761-
elif not cll_map_upload_url:
714+
writer.write_nodes(node_rows)
715+
writer.write_columns(column_rows)
716+
writer.write_edges(edge_rows)
717+
writer.write_tests(test_rows)
718+
pn_elapsed = time.perf_counter() - t_pn_start
719+
pn_size_mb = per_node_db_path.stat().st_size / 1024 / 1024
762720
console.print(
763-
" [[yellow]Warning[/yellow]] No cll_map_url in upload URLs (Cloud server may need update)"
721+
f" per_node.db saved to [bold]{per_node_db_path}[/bold] ({pn_size_mb:.1f} MB, {pn_elapsed:.1f}s)"
764722
)
765-
766-
# Upload per_node.db. Graceful degradation: if Cloud hasn't added
767-
# the per_node_db_url key yet, log a warning and continue — this
768-
# keeps old CLI versions compatible with new Cloud and vice versa.
769-
per_node_db_upload_url = upload_urls.get("per_node_db_url")
770-
if per_node_db_upload_url and per_node_db_path and per_node_db_path.is_file():
723+
except Exception as e:
724+
logger.warning("[recce init] Failed to emit per_node.db: %s", e)
725+
console.print(f" [[yellow]Warning[/yellow]] Failed to emit per_node.db: {e}")
726+
per_node_db_path = None
727+
728+
# Upload results to Cloud.
729+
if cloud_client:
730+
console.print("\n[bold]Uploading results to Cloud...[/bold]")
731+
upload_failures: list[str] = []
771732
try:
772-
with open(per_node_db_path, "rb") as f:
773-
resp = requests.put(
774-
per_node_db_upload_url,
775-
data=f,
776-
headers={"Content-Type": "application/octet-stream"},
777-
timeout=_UPLOAD_TIMEOUT,
778-
)
779-
if resp.status_code in (200, 204):
733+
upload_urls = cloud_client.get_upload_urls_by_session_id(cloud_org_id, cloud_project_id, session_id)
734+
735+
# Upload CLL map
736+
cll_map_upload_url = upload_urls.get("cll_map_url")
737+
if cll_map_upload_url and cll_map_path.is_file():
738+
try:
739+
with open(cll_map_path, "rb") as f:
740+
resp = requests.put(
741+
cll_map_upload_url,
742+
data=f,
743+
headers={"Content-Type": "application/json"},
744+
timeout=_UPLOAD_TIMEOUT,
745+
)
746+
if resp.status_code in (200, 204):
747+
console.print(
748+
f" Uploaded cll_map.json ({cll_map_path.stat().st_size / 1024 / 1024:.1f} MB)"
749+
)
750+
else:
751+
upload_failures.append("cll_map.json")
752+
console.print(
753+
f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: "
754+
f"HTTP {resp.status_code}"
755+
)
756+
except requests.RequestException as e:
757+
upload_failures.append("cll_map.json")
758+
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: {e}")
759+
elif not cll_map_upload_url:
780760
console.print(
781-
f" Uploaded per_node.db ({per_node_db_path.stat().st_size / 1024 / 1024:.1f} MB)"
761+
" [[yellow]Warning[/yellow]] No cll_map_url in upload URLs "
762+
"(Cloud server may need update)"
782763
)
783-
else:
784-
upload_failures.append("per_node.db")
764+
765+
# Upload per_node.db. Graceful degradation: if Cloud hasn't added
766+
# the per_node_db_url key yet, log a warning and continue — this
767+
# keeps old CLI versions compatible with new Cloud and vice versa.
768+
per_node_db_upload_url = upload_urls.get("per_node_db_url")
769+
if per_node_db_upload_url and per_node_db_path and per_node_db_path.is_file():
770+
try:
771+
with open(per_node_db_path, "rb") as f:
772+
resp = requests.put(
773+
per_node_db_upload_url,
774+
data=f,
775+
headers={"Content-Type": "application/octet-stream"},
776+
timeout=_UPLOAD_TIMEOUT,
777+
)
778+
if resp.status_code in (200, 204):
779+
console.print(
780+
f" Uploaded per_node.db "
781+
f"({per_node_db_path.stat().st_size / 1024 / 1024:.1f} MB)"
782+
)
783+
else:
784+
upload_failures.append("per_node.db")
785+
console.print(
786+
f" [[yellow]Warning[/yellow]] Failed to upload per_node.db: "
787+
f"HTTP {resp.status_code}"
788+
)
789+
except requests.RequestException as e:
790+
upload_failures.append("per_node.db")
791+
console.print(f" [[yellow]Warning[/yellow]] Failed to upload per_node.db: {e}")
792+
elif not per_node_db_upload_url:
785793
console.print(
786-
f" [[yellow]Warning[/yellow]] Failed to upload per_node.db: HTTP {resp.status_code}"
794+
" [[yellow]Warning[/yellow]] No per_node_db_url in upload URLs "
795+
"(Cloud server may need update)"
787796
)
788-
except requests.RequestException as e:
789-
upload_failures.append("per_node.db")
790-
console.print(f" [[yellow]Warning[/yellow]] Failed to upload per_node.db: {e}")
791-
elif not per_node_db_upload_url:
792-
console.print(
793-
" [[yellow]Warning[/yellow]] No per_node_db_url in upload URLs (Cloud server may need update)"
794-
)
795797

796-
# Upload CLL cache. cll_cache.db is load-bearing across sessions —
797-
# build_full_cll_map reuses its warm entries on subsequent runs —
798-
# so Cloud uploads it alongside per_node.db.
799-
cll_cache_upload_url = upload_urls.get("cll_cache_url")
800-
if cll_cache_upload_url and Path(cache_db).is_file():
801-
try:
802-
with open(cache_db, "rb") as f:
803-
resp = requests.put(
804-
cll_cache_upload_url,
805-
data=f,
806-
headers={"Content-Type": "application/octet-stream"},
807-
timeout=_UPLOAD_TIMEOUT,
808-
)
809-
if resp.status_code in (200, 204):
810-
console.print(f" Uploaded cll_cache.db ({Path(cache_db).stat().st_size / 1024 / 1024:.1f} MB)")
811-
else:
812-
upload_failures.append("cll_cache.db")
798+
# Upload CLL cache. cll_cache.db is load-bearing across sessions —
799+
# build_full_cll_map reuses its warm entries on subsequent runs —
800+
# so Cloud uploads it alongside per_node.db.
801+
cll_cache_upload_url = upload_urls.get("cll_cache_url")
802+
if cll_cache_upload_url and Path(cache_db).is_file():
803+
try:
804+
with open(cache_db, "rb") as f:
805+
resp = requests.put(
806+
cll_cache_upload_url,
807+
data=f,
808+
headers={"Content-Type": "application/octet-stream"},
809+
timeout=_UPLOAD_TIMEOUT,
810+
)
811+
if resp.status_code in (200, 204):
812+
console.print(
813+
f" Uploaded cll_cache.db "
814+
f"({Path(cache_db).stat().st_size / 1024 / 1024:.1f} MB)"
815+
)
816+
else:
817+
upload_failures.append("cll_cache.db")
818+
console.print(
819+
f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: "
820+
f"HTTP {resp.status_code}"
821+
)
822+
except requests.RequestException as e:
823+
upload_failures.append("cll_cache.db")
824+
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: {e}")
825+
elif not cll_cache_upload_url:
826+
logger.debug("No cll_cache_url in upload URLs — cache upload not supported yet")
827+
828+
if upload_failures:
813829
console.print(
814-
f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: HTTP {resp.status_code}"
830+
f"[bold yellow]Cloud upload completed with warnings[/bold yellow] "
831+
f"(failed: {', '.join(upload_failures)})"
815832
)
816-
except requests.RequestException as e:
817-
upload_failures.append("cll_cache.db")
818-
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: {e}")
819-
elif not cll_cache_upload_url:
820-
logger.debug("No cll_cache_url in upload URLs — cache upload not supported yet")
821-
822-
if upload_failures:
823-
console.print(
824-
f"[bold yellow]Cloud upload completed with warnings[/bold yellow] "
825-
f"(failed: {', '.join(upload_failures)})"
826-
)
827-
else:
828-
console.print("[bold green]Cloud upload complete.[/bold green]")
829-
upload_succeeded = True
830-
except Exception as e:
831-
logger.warning("[recce init] Cloud upload failed: %s", e)
832-
console.print(f" [[yellow]Warning[/yellow]] Cloud upload failed: {e}")
833+
else:
834+
console.print("[bold green]Cloud upload complete.[/bold green]")
835+
except Exception as e:
836+
logger.warning("[recce init] Cloud upload failed: %s", e)
837+
console.print(f" [[yellow]Warning[/yellow]] Cloud upload failed: {e}")
833838
finally:
834-
# Clean up per_node.db scratch dir only. cll_cache.db lives at
835-
# ~/.recce/cll_cache.db (or the user-provided --cache-db path) and
836-
# must persist across invocations for warm-cache reuse.
837-
if upload_succeeded and per_node_scratch is not None:
838-
shutil.rmtree(per_node_scratch, ignore_errors=True)
839+
# Always remove the per_node.db scratch dir — it is throwaway per
840+
# invocation. cll_cache.db lives at ~/.recce/cll_cache.db (or the
841+
# user-provided --cache-db) and is NOT touched here.
842+
shutil.rmtree(per_node_scratch, ignore_errors=True)
839843
else:
840844
console.print("Run [bold]recce server --enable-cll-cache[/bold] to use the cached lineage.")
841845

0 commit comments

Comments
 (0)