From 276325b4c099900c845585ff95400d25e706d8a3 Mon Sep 17 00:00:00 2001 From: "Aleksandr B." Date: Wed, 29 Jul 2026 13:42:43 +0200 Subject: [PATCH] fix(bitbucket-cloud): tolerate denied repos, cut request budget Refs #2011 (cherry picked from commit 81baac8602afee86ab5a70021c58a6ff4e220daa) Signed-off-by: Aleksandr Barkhatov --- .../bitbucket_cloud__repository_branches.sql | 15 +- .../source_bitbucket_cloud/client.py | 57 +- .../source_bitbucket_cloud/streams/base.py | 60 ++- .../streams/branches.py | 92 ++-- .../streams/commit_branch_reachability.py | 10 +- .../source_bitbucket_cloud/streams/commits.py | 10 +- .../streams/file_changes.py | 22 +- .../streams/metric_events.py | 74 ++- .../source_bitbucket_cloud/streams/pr_base.py | 50 +- .../streams/pull_requests.py | 3 + .../git/bitbucket-cloud/tests/conftest.py | 14 + .../git/bitbucket-cloud/tests/test_commits.py | 2 +- .../tests/test_file_changes.py | 6 +- .../tests/test_inaccessible_repos.py | 491 ++++++++++++++++++ .../bitbucket-cloud/tests/test_reliability.py | 10 +- .../tests/test_request_budget.py | 216 ++++++++ 16 files changed, 1064 insertions(+), 68 deletions(-) create mode 100644 src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py create mode 100644 src/ingestion/connectors/git/bitbucket-cloud/tests/test_request_budget.py diff --git a/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__repository_branches.sql b/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__repository_branches.sql index ca2daab18..eb304ddd3 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__repository_branches.sql +++ b/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__repository_branches.sql @@ -8,28 +8,33 @@ tags=['bitbucket-cloud', 'silver:class_git_repository_branches'] ) }} +-- Generations are per repository (workspace, repo_slug), matching the stream: +-- a repository that is denied or fails simply has no new generation and keeps +-- its previous branches, without freezing the other repositories. WITH generations AS ( SELECT tenant_id, source_id, - bucket_id, + workspace, + repo_slug, generation_id, countIf(record_type = 'item') AS observed_count, maxIf(snapshot_item_count, record_type = 'snapshot_complete') AS expected_count, maxIf(_airbyte_extracted_at, record_type = 'snapshot_complete') AS completed_at, countIf(record_type = 'snapshot_complete' AND snapshot_available) AS completion_count FROM {{ source('bronze_bitbucket_cloud', 'branches') }} FINAL - GROUP BY tenant_id, source_id, bucket_id, generation_id + GROUP BY tenant_id, source_id, workspace, repo_slug, generation_id HAVING completion_count > 0 AND observed_count = expected_count ), latest AS ( SELECT tenant_id, source_id, - bucket_id, + workspace, + repo_slug, argMax(generation_id, completed_at) AS generation_id FROM generations - GROUP BY tenant_id, source_id, bucket_id + GROUP BY tenant_id, source_id, workspace, repo_slug ) SELECT tenant_id, @@ -45,5 +50,5 @@ SELECT toUnixTimestamp64Milli(now64()) AS _version, _airbyte_extracted_at FROM {{ source('bronze_bitbucket_cloud', 'branches') }} AS branch FINAL -INNER JOIN latest USING (tenant_id, source_id, bucket_id, generation_id) +INNER JOIN latest USING (tenant_id, source_id, workspace, repo_slug, generation_id) WHERE record_type = 'item' diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py index 83650b659..7acdc56f4 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py @@ -49,6 +49,32 @@ def __init__(self, client: BitbucketClient, workspaces: Sequence[str], skip_fork self._skip_forks = skip_forks self._repositories: list[RepositoryRef] | None = None self._branches: dict[str, list[BranchRef]] = {} + self._inaccessible: set[str] = set() + # Shared per-sync selection caches (see streams/pr_base.py). Keyed by + # (repository, watermark) and holding SLIM projections only — a handful + # of scalar fields per entity, never the raw API objects. The raw list + # for the whole workspace would cost hundreds of MB held across the six + # sequential PR streams; the slim form is ~100 bytes per entity. + self.pr_selections: dict[tuple[str, str], tuple[list, dict]] = {} + self.pipeline_selections: dict[tuple[str, str], tuple[bool, list, dict]] = {} + self.issue_selections: dict[tuple[str, str], tuple[bool, list, dict]] = {} + + def mark_inaccessible(self, repo: RepositoryRef) -> None: + """Record that this repository denies access, for the rest of the sync. + + A repository can appear in the workspace listing and still refuse every + request under it (403) — normal with repo-scoped tokens or per-repository + permissions. The catalog is shared by every stream, so the first stream + to discover it saves the others from rediscovering it repo by repo. + """ + self._inaccessible.add(repo.uuid) + + def is_inaccessible(self, repo: RepositoryRef) -> bool: + return repo.uuid in self._inaccessible + + @property + def inaccessible_count(self) -> int: + return len(self._inaccessible) def repositories(self) -> list[RepositoryRef]: if self._repositories is None: @@ -233,19 +259,32 @@ def branches(self, repo: RepositoryRef) -> list[BranchRef]: ) return branches + # Bitbucket documents no ceiling on include/exclude counts, and per + # BCLOUD-13229 its limits tend to surface as unexplained 400s. A repository + # with hundreds of branches would otherwise send them all in one form, so + # includes are chunked; the union of the chunked ranges is the same commit + # set (the full exclude list rides along with every chunk), and bronze + # dedups any overlap by unique_key. + COMMITS_INCLUDE_CHUNK = 100 + def commits_between( self, repo: RepositoryRef, current_heads: Sequence[str], previous_heads: Sequence[str] ) -> Iterable[Mapping[str, Any]]: - form = [("include", head) for head in sorted(set(current_heads))] - form.extend(("exclude", head) for head in sorted(set(previous_heads))) - if not form: + includes = sorted(set(current_heads)) + excludes = [("exclude", head) for head in sorted(set(previous_heads))] + # With no include the endpoint falls back to every branch, so excludes + # alone would page a whole history out; nothing is newly reachable. + if not includes: return - yield from self.paginate( - self.repo_path(repo, "commits"), - method="POST", - params={"pagelen": "100"}, - data=form, - ) + for start in range(0, len(includes), self.COMMITS_INCLUDE_CHUNK): + chunk = includes[start : start + self.COMMITS_INCLUDE_CHUNK] + form = [("include", head) for head in chunk] + excludes + yield from self.paginate( + self.repo_path(repo, "commits"), + method="POST", + params={"pagelen": "100"}, + data=form, + ) def repo_path(self, repo: RepositoryRef, suffix: str) -> str: workspace = quote(repo.workspace, safe="") diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py index 7c7ee8cd2..fb19c98c8 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py @@ -13,7 +13,12 @@ from airbyte_cdk.models import SyncMode from airbyte_cdk.sources.streams import CheckpointMixin, Stream -from source_bitbucket_cloud.client import BitbucketClient, RepositoryCatalog, RepositoryRef +from source_bitbucket_cloud.client import ( + BitbucketApiError, + BitbucketClient, + RepositoryCatalog, + RepositoryRef, +) logger = logging.getLogger("airbyte") @@ -22,6 +27,10 @@ # Bumped from 2 when entity and state keys moved back to workspace/slug: a # version-2 state is keyed by repository uuid and no longer addresses anything. STATE_VERSION = 3 +# Statuses that mean "this token will never read this repository": no retry +# helps, so the repository is skipped instead of failing the sync. 404 is here +# too — a repository listed at the start of a sync can be deleted mid-run. +DENIED_STATUSES = frozenset({403, 404}) def now_iso() -> str: @@ -165,6 +174,7 @@ def __init__( self._catalog = catalog or RepositoryCatalog(self._client, self._workspaces, self._skip_forks) self._repositories_by_bucket: dict[int, list[RepositoryRef]] = {} self._failed_repositories: list[str] = [] + self._skipped_repositories: list[str] = [] def stream_slices( self, @@ -188,8 +198,29 @@ def read_records( del sync_mode, cursor_field, stream_state bucket_id, repositories = self.bucket(stream_slice) for repo in repositories: + if self._catalog.is_inaccessible(repo): + # Discovered by an earlier stream; still counts toward THIS + # stream's end-of-sync skipped summary. + self._skipped_repositories.append(f"{repo.workspace}/{repo.slug}") + continue try: yield from self.repository_records(repo, bucket_id) + except BitbucketApiError as error: + if error.status_code == 401: + # Credential failure is global, not per-repository: every + # remaining repo would fail identically, drowning the log in + # quarantine noise before a generic end-of-sync error. Abort + # now with the actionable cause instead. + raise RuntimeError( + "Bitbucket authentication failed mid-sync (HTTP 401): the token was " + "rejected. If bitbucket_username is unset, Atlassian API tokens are " + "sent as Bearer and refused — set the username, or the token has " + "expired/been rotated." + ) from error + if error.status_code in DENIED_STATUSES: + self.skip_repository(repo, error.status_code) + else: + self.record_failure(repo) except Exception: self.record_failure(repo) self.finish_bucket(bucket_id, repositories) @@ -202,12 +233,39 @@ def bucket(self, stream_slice: Mapping[str, Any] | None) -> tuple[int, list[Repo return bucket_id, self.repositories_for_slice(stream_slice) def record_failure(self, repo: RepositoryRef) -> None: + """A failure worth surfacing: transient, so retrying the sync may fix it.""" name = f"{repo.workspace}/{repo.slug}" self._failed_repositories.append(name) logger.exception(f"{self.name}: repository {name} failed; its state was not advanced, continuing") + def skip_repository(self, repo: RepositoryRef, status_code: int) -> None: + """A repository the token cannot read: skip it without failing the sync. + + A repository can be listed for the workspace and still deny every request + under it, which is routine with repo-scoped tokens and per-repository + permissions. That is a configuration fact, not an incident: retrying will + never change it, so counting it as a failure would leave the sync red + forever and bury the transient failures that do deserve attention. The + repository is marked on the shared catalog so the remaining streams skip + it instead of each rediscovering the same 403. + """ + name = f"{repo.workspace}/{repo.slug}" + already_known = self._catalog.is_inaccessible(repo) + self._catalog.mark_inaccessible(repo) + if not already_known: + logger.warning( + f"{self.name}: repository {name} denied access (HTTP {status_code}); " + "skipping it for the rest of this sync" + ) + self._skipped_repositories.append(name) + def finish_bucket(self, bucket_id: int, repositories: Sequence[RepositoryRef]) -> None: del repositories + if bucket_id == BUCKET_COUNT - 1 and self._skipped_repositories: + logger.info( + f"{self.name}: skipped {len(self._skipped_repositories)} inaccessible " + f"repositories: {', '.join(sorted(set(self._skipped_repositories))[:10])}" + ) if bucket_id == BUCKET_COUNT - 1 and self._failed_repositories: raise RuntimeError( f"{self.name}: {len(self._failed_repositories)} repositories failed this sync: " diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py index c33043cba..cca0e581f 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py @@ -3,57 +3,69 @@ from collections.abc import Iterable, Mapping from typing import Any -from airbyte_cdk.models import SyncMode +from source_bitbucket_cloud.streams.base import BitbucketIncrementalStream, repo_scope, schema, unique_key -from source_bitbucket_cloud.streams.base import BitbucketStream, repo_scope, schema, unique_key +class BranchesStream(BitbucketIncrementalStream): + """Current branches per repository, as per-repository snapshots. + + The generation is scoped to one repository, not to a bucket: a repository + that is denied (403) or fails simply produces no marker this sync, so dbt + keeps its previous branch generation, while every other repository updates + independently. A bucket-scoped generation would freeze branch updates for a + whole bucket over one denied repository — and workspaces where unreadable + repositories are common (observed in production) would freeze every bucket. + + Incremental only in the cheapest sense: the per-repository state holds the + repository's updated_on from the workspace listing, and a repository that + has not been pushed to since the last pass is skipped without a request — + its previous generation simply stays the newest complete one. + + Trade-off: a repository deleted from the workspace stops producing + generations, so its last branch snapshot lingers in silver. That is bounded + (the repository is gone) and preferable to fleet-wide starvation. + """ -class BranchesStream(BitbucketStream): name = "branches" + cursor_field = "updated_on" - def read_records( - self, - sync_mode: SyncMode, - cursor_field: list[str] | None = None, - stream_slice: Mapping[str, Any] | None = None, - stream_state: Mapping[str, Any] | None = None, - ) -> Iterable[Mapping[str, Any]]: - del sync_mode, cursor_field, stream_state - bucket_id, repositories = self.bucket(stream_slice) - generation = self.generation("branches", bucket_id) + def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]]: + prior = self.repository_state(repo) + repo_updated_on = str(repo.raw.get("updated_on") or "") + if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on: + return + generation = self.generation("branches", *repo_scope(repo)) entity_keys: set[str] = set() - failures_before = len(self._failed_repositories) - for repo in repositories: - try: - for branch in self._catalog.branches(repo): - entity_key = unique_key(self._tenant_id, self._source_id, *repo_scope(repo), branch.name) - entity_keys.add(entity_key) - yield self.item( - entity_key=entity_key, - generation_id=generation, - bucket_id=bucket_id, - repository_uuid=repo.uuid, - workspace_uuid=repo.workspace_uuid, - workspace=repo.workspace, - repo_slug=repo.slug, - name=branch.name, - target_hash=branch.head_sha, - target_date=branch.target_date, - mainbranch_name=repo.mainbranch_name, - default_branch_name=repo.mainbranch_name, - is_default=branch.is_default, - updated_on=repo.raw.get("updated_on"), - ) - except Exception: - self.record_failure(repo) + for branch in self._catalog.branches(repo): + entity_key = unique_key(self._tenant_id, self._source_id, *repo_scope(repo), branch.name) + entity_keys.add(entity_key) + yield self.item( + entity_key=entity_key, + generation_id=generation, + bucket_id=bucket_id, + repository_uuid=repo.uuid, + workspace_uuid=repo.workspace_uuid, + workspace=repo.workspace, + repo_slug=repo.slug, + name=branch.name, + target_hash=branch.head_sha, + target_date=branch.target_date, + mainbranch_name=repo.mainbranch_name, + default_branch_name=repo.mainbranch_name, + is_default=branch.is_default, + updated_on=repo.raw.get("updated_on"), + ) yield self.complete( - scope_parts=["branches", bucket_id], + scope_parts=["branches", *repo_scope(repo)], generation_id=generation, item_count=len(entity_keys), bucket_id=bucket_id, - available=len(self._failed_repositories) == failures_before, + repository_uuid=repo.uuid, + workspace_uuid=repo.workspace_uuid, + workspace=repo.workspace, + repo_slug=repo.slug, ) - self.finish_bucket(bucket_id, repositories) + self.commit_repository_state(repo, {"repo_updated_on": repo_updated_on}) def get_json_schema(self) -> Mapping[str, Any]: nullable_string = {"type": ["null", "string"]} diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py index 2f0c8b457..e06145e8a 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py @@ -15,6 +15,14 @@ class CommitBranchReachabilityStream(CommitRangeMixin, BitbucketIncrementalStrea def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]]: del bucket_id prior = self.repository_state(repo) + repo_updated_on = str(repo.raw.get("updated_on") or "") + if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on: + # The repository has not been pushed to since the last successful + # pass (updated_on comes free with the workspace listing), so the + # branch heads cannot have moved: skip the branch listing and the + # range fetch entirely. This is what keeps the per-repository + # request budget at zero for the idle majority of a large fleet. + return branches, current_heads = self.branch_snapshot(repo) previous_heads = prior.get("heads") or {} branch_by_name = {branch.name: branch for branch in branches} @@ -61,7 +69,7 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any] committed_at=None, reachability_action="branch_deleted", ) - self.commit_repository_state(repo, {"heads": current_heads}) + self.commit_repository_state(repo, {"heads": current_heads, "repo_updated_on": repo_updated_on}) def _changes(self, repo, branch, include: str, exclude: str | None, action: str): try: diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py index d885213e3..909079522 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py @@ -14,6 +14,14 @@ class CommitsStream(CommitRangeMixin, BitbucketIncrementalStream): def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]]: del bucket_id prior = self.repository_state(repo) + repo_updated_on = str(repo.raw.get("updated_on") or "") + if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on: + # The repository has not been pushed to since the last successful + # pass (updated_on comes free with the workspace listing), so the + # branch heads cannot have moved: skip the branch listing and the + # range fetch entirely. This is what keeps the per-repository + # request budget at zero for the idle majority of a large fleet. + return _, current_heads = self.branch_snapshot(repo) current_head_shas = sorted(set(current_heads.values())) previous_head_shas = prior.get("head_shas") or [] @@ -23,7 +31,7 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any] if self._start_date and record.get("date") and str(record["date"])[:10] < self._start_date: continue yield record - self.commit_repository_state(repo, {"head_shas": current_head_shas}) + self.commit_repository_state(repo, {"head_shas": current_head_shas, "repo_updated_on": repo_updated_on}) def _record(self, repo, commit: Mapping[str, Any]) -> Mapping[str, Any]: sha = str(commit.get("hash") or "") diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py index 87eb98e2a..d6e70e552 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py @@ -14,6 +14,14 @@ class FileChangesStream(CommitRangeMixin, BitbucketIncrementalStream): def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]]: del bucket_id prior = self.repository_state(repo) + repo_updated_on = str(repo.raw.get("updated_on") or "") + if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on: + # The repository has not been pushed to since the last successful + # pass (updated_on comes free with the workspace listing), so the + # branch heads cannot have moved: skip the branch listing and the + # range fetch entirely. This is what keeps the per-repository + # request budget at zero for the idle majority of a large fleet. + return _, current_heads = self.branch_snapshot(repo) current_head_shas = sorted(set(current_heads.values())) previous_head_shas = prior.get("head_shas") or [] @@ -23,14 +31,23 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any] if self._start_date and committed_date and str(committed_date)[:10] < self._start_date: continue yield from self._diffstat(repo, str(commit.get("hash") or ""), committed_date) - self.commit_repository_state(repo, {"head_shas": current_head_shas}) + self.commit_repository_state(repo, {"head_shas": current_head_shas, "repo_updated_on": repo_updated_on}) def _diffstat(self, repo, sha: str, committed_date: Any) -> Iterable[Mapping[str, Any]]: if not sha: return generation = self.generation(repo.uuid, sha) entity_keys: set[str] = set() - for entry in self._client.paginate(self._client.repo_path(repo, f"diffstat/{sha}"), params={"pagelen": "100"}): + # A commit's diffstat can be permanently gone (orphaned merge parents, + # rewritten history) — the pre-rewrite connector tolerated exactly this + # ("commit diffstat gone", ignore_404). Raising here would fail the + # repository on every sync forever. The marker records the denial so the + # completeness gate keeps whatever was known before instead of treating + # the empty read as "this commit changed nothing". + present, entries = self._client.paginate_optional( + self._client.repo_path(repo, f"diffstat/{sha}"), params={"pagelen": "100"} + ) + for entry in entries: new_file = entry.get("new") or {} old_file = entry.get("old") or {} filename = new_file.get("path") or old_file.get("path") @@ -61,6 +78,7 @@ def _diffstat(self, repo, sha: str, committed_date: Any) -> Iterable[Mapping[str scope_parts=[repo.uuid, sha, "diffstat"], generation_id=generation, item_count=len(entity_keys), + available=present, repository_uuid=repo.uuid, workspace_uuid=repo.workspace_uuid, source_type="commit", diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py index 0a8d8f911..8fc633eff 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py @@ -4,7 +4,14 @@ from datetime import datetime, timedelta from typing import Any -from source_bitbucket_cloud.streams.base import BitbucketIncrementalStream, BitbucketStream, repo_scope, schema, unique_key +from source_bitbucket_cloud.streams.base import ( + BitbucketIncrementalStream, + BitbucketStream, + repo_scope, + repo_state_key, + schema, + unique_key, +) from source_bitbucket_cloud.streams.pr_base import PullRequestStateStream @@ -85,9 +92,23 @@ class DeploymentsStream(RepositorySnapshotStream): resource = "deployments" +def slim_pipeline(pipeline: Mapping[str, Any]) -> dict[str, Any]: + """Only what the child streams read; the full object is never cached.""" + return { + "uuid": pipeline.get("uuid"), + "created_on": pipeline.get("created_on"), + "completed_on": pipeline.get("completed_on"), + "state": {"name": ((pipeline.get("state") or {}).get("name"))}, + } + + class PipelineStateStream(BitbucketIncrementalStream): cursor_field = "created_on" + # The stream emitting full pipeline records (pipelines) always fetches and + # fills the slim cache; steps and test reports read it. + reads_selection_cache = True + def repository_records(self, repo, bucket_id: int): del bucket_id present, pipelines, new_state = self.pipeline_candidates(repo, self.repository_state(repo)) @@ -101,6 +122,21 @@ def pipeline_records(self, repo, pipeline: Mapping[str, Any]): raise NotImplementedError def pipeline_candidates(self, repo, prior: Mapping[str, Any]): + cache_key = (repo_state_key(repo), str(prior.get("created_on") or "")) + if self.reads_selection_cache: + cached = self._catalog.pipeline_selections.get(cache_key) + if cached is not None: + present, slim, cached_state = cached + return present, slim, dict(cached_state) + present, pipelines, new_state = self._fetch_pipelines(repo, prior) + self._catalog.pipeline_selections[cache_key] = ( + present, + [slim_pipeline(p) for p in pipelines], + dict(new_state), + ) + return present, pipelines, new_state + + def _fetch_pipelines(self, repo, prior: Mapping[str, Any]): watermark = str(prior.get("created_on") or "") floor = None if watermark: @@ -119,8 +155,14 @@ def pipeline_candidates(self, repo, prior: Mapping[str, Any]): if pipeline_uuid: pipelines[pipeline_uuid] = pipeline for pipeline_uuid in prior.get("open") or []: + # 403 is tolerated here, not raised: Pipelines is a per-repository + # feature, so a denial means "no pipelines visible", not "this + # repository is unreadable". Letting it escape would mark the whole + # repository inaccessible and suppress its commits and pull requests. response = self._client.request( - "GET", self._client.repo_path(repo, f"pipelines/{pipeline_uuid}"), allow_not_found=True + "GET", + self._client.repo_path(repo, f"pipelines/{pipeline_uuid}"), + allow_statuses={403, 404}, ) if response is not None: pipeline = response.json() @@ -140,6 +182,7 @@ def pipeline_candidates(self, repo, prior: Mapping[str, Any]): class PipelinesStream(PipelineStateStream): name = "pipelines" + reads_selection_cache = False def pipeline_records(self, repo, pipeline: Mapping[str, Any]): pipeline_uuid = pipeline.get("uuid") @@ -230,7 +273,10 @@ def pipeline_records(self, repo, pipeline: Mapping[str, Any]): step_uuid = step.get("uuid") generation = self.generation(repo.uuid, pipeline_uuid, step_uuid, "test_reports") path = self._client.repo_path(repo, f"pipelines/{pipeline_uuid}/steps/{step_uuid}/test_reports") - response = self._client.request("GET", path, allow_not_found=True) + # Tolerates 403 for the same reason as the pipeline refetch above: a + # feature-level denial must mark this snapshot unavailable, not the + # repository. `available` below already records it. + response = self._client.request("GET", path, allow_statuses={403, 404}) count = 0 if response is not None: payload = response.json() @@ -279,9 +325,15 @@ def get_json_schema(self) -> Mapping[str, Any]: ) +def slim_issue(issue: Mapping[str, Any]) -> dict[str, Any]: + return {"id": issue.get("id"), "updated_on": issue.get("updated_on")} + + class IssueStateStream(BitbucketIncrementalStream): cursor_field = "updated_on" + reads_selection_cache = True + def repository_records(self, repo, bucket_id: int): del bucket_id if not repo.has_issues: @@ -298,6 +350,21 @@ def issue_records(self, repo, issue: Mapping[str, Any]): raise NotImplementedError def selected_issues(self, repo, prior): + cache_key = (repo_state_key(repo), str(prior.get("updated_on") or "")) + if self.reads_selection_cache: + cached = self._catalog.issue_selections.get(cache_key) + if cached is not None: + present, slim, cached_state = cached + return present, slim, dict(cached_state) + present, issues, new_state = self._fetch_issues(repo, prior) + self._catalog.issue_selections[cache_key] = ( + present, + [slim_issue(i) for i in issues], + dict(new_state), + ) + return present, issues, new_state + + def _fetch_issues(self, repo, prior): watermark = str(prior.get("updated_on") or "") floor = self._start_date if watermark: @@ -314,6 +381,7 @@ def selected_issues(self, repo, prior): class IssuesStream(IssueStateStream): name = "issues" + reads_selection_cache = False def issue_records(self, repo, issue: Mapping[str, Any]): issue_id = issue.get("id") diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_base.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_base.py index 38cc5db01..3cf2929d0 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_base.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_base.py @@ -5,7 +5,7 @@ from typing import Any from source_bitbucket_cloud.client import RepositoryRef -from source_bitbucket_cloud.streams.base import BitbucketIncrementalStream +from source_bitbucket_cloud.streams.base import BitbucketIncrementalStream, repo_state_key PR_STATES = ("OPEN", "MERGED", "DECLINED", "SUPERSEDED") TERMINAL_PR_STATES = ("MERGED", "DECLINED", "SUPERSEDED") @@ -13,9 +13,32 @@ OVERLAP_MINUTES = 5 +def slim_pull_request(pr: Mapping[str, Any]) -> dict[str, Any]: + """The four fields the child streams actually read, in the original shape. + + Anything larger must not be cached: the selection cache lives across the six + sequential PR streams for every repository at once, and raw PR objects + (description, reviewers, participants) would cost hundreds of MB where this + costs ~100 bytes per pull request. + """ + source = pr.get("source") or {} + destination = pr.get("destination") or {} + return { + "id": pr.get("id"), + "updated_on": pr.get("updated_on"), + "source": {"commit": {"hash": (source.get("commit") or {}).get("hash")}}, + "destination": {"commit": {"hash": (destination.get("commit") or {}).get("hash")}}, + } + + class PullRequestStateStream(BitbucketIncrementalStream): cursor_field = "updated_on" + # The stream that emits full PR records (pull_requests) must never consume + # the slim cache — it needs every field. It always fetches, and its fetch + # fills the cache for the five child streams that follow it. + reads_selection_cache = True + def repository_records(self, repo, bucket_id): del bucket_id selected, new_state = self.selected_pull_requests(repo, self.repository_state(repo)) @@ -28,6 +51,31 @@ def pull_request_records(self, repo, pr: Mapping[str, Any]): def selected_pull_requests( self, repo: RepositoryRef, prior: Mapping[str, Any] + ) -> tuple[list[Mapping[str, Any]], Mapping[str, Any]]: + """One PR selection per (repository, watermark) per sync, shared. + + Each of the six PR streams used to list the repository's pull requests + itself — up to three requests each, ~18 per repository per sync, the + single largest consumer of the rate limit. The selection is deterministic + given the watermark, so it is fetched once and shared through the + catalog; a stream whose watermark diverged (e.g. it failed last sync) + misses the cache and fetches its own. + """ + cache_key = (repo_state_key(repo), str(prior.get("updated_on") or ""), str(prior.get("reconcile_after_id") or 0)) + if self.reads_selection_cache: + cached = self._catalog.pr_selections.get(cache_key) + if cached is not None: + slim_selected, cached_state = cached + return slim_selected, dict(cached_state) + selected, new_state = self._fetch_pull_requests(repo, prior) + self._catalog.pr_selections[cache_key] = ( + [slim_pull_request(pr) for pr in selected], + dict(new_state), + ) + return selected, new_state + + def _fetch_pull_requests( + self, repo: RepositoryRef, prior: Mapping[str, Any] ) -> tuple[list[Mapping[str, Any]], Mapping[str, Any]]: selected: dict[int, Mapping[str, Any]] = {} watermark = str(prior.get("updated_on") or "") diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pull_requests.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pull_requests.py index 518c3556d..830182a88 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pull_requests.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pull_requests.py @@ -9,6 +9,9 @@ class PullRequestsStream(PullRequestStateStream): name = "pull_requests" + # Emits full PR records, so it must fetch everything itself; its fetch + # populates the slim cache the five child streams read. + reads_selection_cache = False def pull_request_records(self, repo, pr: Mapping[str, Any]) -> Iterable[Mapping[str, Any]]: yield self._record(repo, pr) diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py index 81fe66e9c..47c58d0c3 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py @@ -39,6 +39,20 @@ class FakeCatalog: def __init__(self, repositories: Iterable[RepositoryRef], client: FakeClient | None = None): self._repositories = list(repositories) self._client = client + self._inaccessible: set[str] = set() + self.pr_selections: dict = {} + self.pipeline_selections: dict = {} + self.issue_selections: dict = {} + + def mark_inaccessible(self, repo: RepositoryRef) -> None: + self._inaccessible.add(repo.uuid) + + def is_inaccessible(self, repo: RepositoryRef) -> bool: + return repo.uuid in self._inaccessible + + @property + def inaccessible_count(self) -> int: + return len(self._inaccessible) def repositories(self) -> list[RepositoryRef]: return self._repositories diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py index a9870c93f..62997b585 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py @@ -34,7 +34,7 @@ def test_changed_heads_fetch_range_and_map_commit(commits_stream, client, repo): assert records[0]["branch_name"] is None assert records[0]["parent_hashes"] == ["p1"] assert set(records[0]) <= set(commits_stream.get_json_schema()["properties"]) - assert commits_stream.state["repositories"][repo_state_key(repo)] == {"head_shas": ["head", "head2"]} + assert commits_stream.state["repositories"][repo_state_key(repo)] == {"head_shas": ["head", "head2"], "repo_updated_on": "2026-06-01T00:00:00+00:00"} def test_unchanged_heads_make_no_commit_request(commits_stream, client, repo): diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_file_changes.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_file_changes.py index 3198553ce..4e1c0b06a 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_file_changes.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_file_changes.py @@ -11,7 +11,7 @@ def test_file_changes_independently_walk_commits(file_changes_stream, client, re client.branch_values[repo.uuid] = [branch("main", "head")] client.commit_values = [{"hash": "c1", "date": "2026-06-01"}] path = client.repo_path(repo, "diffstat/c1") - client.page_values[path] = [ + client.optional_values[path] = (True, [ { "status": "renamed", "old": {"path": "old.py"}, @@ -20,7 +20,7 @@ def test_file_changes_independently_walk_commits(file_changes_stream, client, re "lines_removed": 2, }, {"status": "removed", "old": {"path": "gone.py"}, "lines_added": 0, "lines_removed": 8}, - ] + ]) records = read(file_changes_stream, repo) items, complete = records[:-1], records[-1] assert client.commit_calls == [(["head"], [])] @@ -36,7 +36,7 @@ def test_file_change_snapshot_counts_distinct_paths(file_changes_stream, client, client.commit_values = [{"hash": "c1", "date": "2026-06-01"}] path = client.repo_path(repo, "diffstat/c1") entry = {"status": "modified", "new": {"path": "a.py"}} - client.page_values[path] = [entry, entry] + client.optional_values[path] = (True, [entry, entry]) records = read(file_changes_stream, repo) assert records[-1]["snapshot_item_count"] == 1 diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py new file mode 100644 index 000000000..f27530103 --- /dev/null +++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py @@ -0,0 +1,491 @@ +"""A repository the token cannot read must not fail the sync. + +A repository can be listed for a workspace and still answer 403 to every request +under it — routine with repo-scoped tokens and per-repository permissions, and +observed in production. Retrying never changes it, so treating it as a failure +leaves the sync red on every run and buries the transient failures that do need +attention. These tests pin the distinction: denied is skipped, everything else +still fails loudly. + +The matrix classes at the bottom run EVERY stream the source wires — the list is +derived from `SourceBitbucketCloud.streams()` itself, so a stream added later is +covered automatically instead of depending on this file being remembered. +""" + +from __future__ import annotations + +import pytest +from airbyte_cdk.models import SyncMode + +from source_bitbucket_cloud.client import BitbucketApiError +from source_bitbucket_cloud.source import SourceBitbucketCloud +from source_bitbucket_cloud.streams.base import BUCKET_COUNT, repo_state_key, repository_bucket +from source_bitbucket_cloud.streams.branches import BranchesStream +from source_bitbucket_cloud.streams.commits import CommitsStream +from tests.conftest import SHARED, FakeCatalog, FakeClient, branch, repository + + +def every_stream_class(): + """All stream classes, from the production wiring — not a hand list.""" + source = SourceBitbucketCloud() + streams = source.streams( + { + "bitbucket_token": "t", + "bitbucket_workspaces": ["ws"], + "insight_tenant_id": "T", + "insight_source_id": "S", + } + ) + return [type(stream) for stream in streams] + + +def denied(status: int): + class DeniedClient(FakeClient): + def branches(self, repo): + raise BitbucketApiError(status, "https://api.bitbucket.org/2.0/x", "no access") + + return DeniedClient() + + +def read_all_buckets(stream): + records, error = [], None + for bucket in range(BUCKET_COUNT): + try: + records.extend(stream.read_records(SyncMode.incremental, stream_slice={"bucket_id": bucket})) + except RuntimeError as exc: + error = exc + return records, error + + +def build(cls, repos, client): + catalog = FakeCatalog(repos, client) + return cls(**{**SHARED, "client": client, "catalog": catalog}), catalog + + +class TestDeniedRepositoryIsSkipped: + def test_403_does_not_fail_the_sync(self): + stream, _ = build(CommitsStream, [repository()], denied(403)) + stream.state = {} + + records, error = read_all_buckets(stream) + + assert error is None, "a permanently denied repository must not fail the sync" + assert records == [] + + def test_404_does_not_fail_the_sync(self): + """A repository deleted between the listing and the fetch.""" + stream, _ = build(CommitsStream, [repository()], denied(404)) + stream.state = {} + + _, error = read_all_buckets(stream) + + assert error is None + + def test_denied_repository_state_is_not_advanced(self): + stream, _ = build(CommitsStream, [repository()], denied(403)) + stream.state = {} + + read_all_buckets(stream) + + assert stream.state["repositories"] == {} + + def test_it_is_recorded_on_the_shared_catalog(self): + """So the remaining streams skip it instead of rediscovering the 403.""" + repo = repository() + stream, catalog = build(CommitsStream, [repo], denied(403)) + stream.state = {} + + read_all_buckets(stream) + + assert catalog.is_inaccessible(repo) + assert catalog.inaccessible_count == 1 + + def test_a_stream_started_later_skips_it_without_a_request(self): + repo = repository() + client = denied(403) + catalog = FakeCatalog([repo], client) + catalog.mark_inaccessible(repo) + later = CommitsStream(**{**SHARED, "client": client, "catalog": catalog}) + later.state = {} + + records, error = read_all_buckets(later) + + assert (records, error) == ([], None) + assert later._skipped_repositories == [f"{repo.workspace}/{repo.slug}"], ( + "a pre-known inaccessible repository must still appear in this stream's skipped summary" + ) + + def test_other_repositories_still_sync(self): + good, bad = repository(slug="good"), repository(slug="bad", uuid="{bad}") + + class MixedClient(FakeClient): + def branches(self, repo): + if repo.slug == "bad": + raise BitbucketApiError(403, "https://api.bitbucket.org/2.0/x", "no access") + return [branch("main", "a1")] + + client = MixedClient() + client.commit_values = [{"hash": "a1", "date": "2026-06-01T00:00:00+00:00"}] + stream, _ = build(CommitsStream, [good, bad], client) + stream.state = {} + + records, error = read_all_buckets(stream) + + assert error is None + assert [r["hash"] for r in records] == ["a1"] + assert stream.state["repositories"] == {repo_state_key(good): {"head_shas": ["a1"], "repo_updated_on": "2026-06-01T00:00:00+00:00"}} + + +class TestTransientFailuresStillFail: + def test_500_still_fails_the_sync(self): + stream, catalog = build(CommitsStream, [repository()], denied(500)) + stream.state = {} + + _, error = read_all_buckets(stream) + + assert error is not None, "a transient failure must still surface" + assert not catalog.is_inaccessible(repository()), "and must not mark the repository denied" + + def test_non_api_errors_still_fail_the_sync(self): + class BrokenClient(FakeClient): + def branches(self, repo): + raise RuntimeError("boom") + + stream, _ = build(CommitsStream, [repository()], BrokenClient()) + stream.state = {} + + _, error = read_all_buckets(stream) + + assert error is not None + + +class TestBranchesSnapshotStaysSafe: + """branches is a per-repository, deletion-aware snapshot. + + A denied repository must produce NO marker — its previous generation then + stays the newest complete one and its branches are retained. Emitting an + available marker instead would read as "every branch of that repository was + deleted". Per-repository (not bucket) scope matters at fleet scale: with + denied repositories scattered across buckets, a bucket-scoped generation + would freeze branch updates for every repository, permanently. + """ + + def _bucket_of(self, repo): + return repository_bucket(repo_state_key(repo)) + + def test_denied_repository_produces_no_marker_and_keeps_its_generation(self): + repo = repository() + stream, _ = build(BranchesStream, [repo], denied(403)) + + records = list( + stream.read_records(SyncMode.full_refresh, stream_slice={"bucket_id": self._bucket_of(repo)}) + ) + + assert records == [], "a denied repository must contribute nothing — absence keeps its previous generation" + + def test_denied_repository_does_not_freeze_its_neighbours(self): + """The fleet-scale property: other repositories keep updating.""" + readable = repository(slug="readable") + stream_denied = repository(slug="denied", uuid="{denied}") + # force both into the same bucket so the old bucket-scope design would couple them + while repository_bucket(repo_state_key(readable)) != repository_bucket(repo_state_key(stream_denied)): + readable = repository(slug=readable.slug + "x") + + class MixedClient(FakeClient): + def branches(self, repo): + if repo.slug.startswith("denied"): + raise BitbucketApiError(403, "https://api.bitbucket.org/2.0/x", "no access") + return [branch("main", "a1")] + + stream, _ = build(BranchesStream, [readable, stream_denied], MixedClient()) + records = list( + stream.read_records(SyncMode.full_refresh, stream_slice={"bucket_id": self._bucket_of(readable)}) + ) + + markers = [r for r in records if r.get("record_type") == "snapshot_complete"] + assert len(markers) == 1, "exactly the readable repository closes a generation" + assert markers[0]["repo_slug"] == readable.slug + assert markers[0]["snapshot_available"] is True + assert markers[0]["snapshot_item_count"] == 1 + + def test_marker_is_available_when_every_repository_was_read(self): + repo = repository() + client = FakeClient() + client.branch_values[repo.uuid] = [branch("main", "a1")] + stream, _ = build(BranchesStream, [repo], client) + + records = list( + stream.read_records(SyncMode.full_refresh, stream_slice={"bucket_id": self._bucket_of(repo)}) + ) + + assert records[-1]["snapshot_available"] is True + assert records[-1]["snapshot_item_count"] == 1 + + def test_a_denied_repository_does_not_fail_the_branches_stream(self): + stream, _ = build(BranchesStream, [repository()], denied(403)) + + _, error = read_all_buckets(stream) + + assert error is None + + +class FullyDeniedClient(FakeClient): + """Faithful mirror of the real client against an all-403 repository. + + Raising paths raise BitbucketApiError(403); tolerant paths behave as the + real client does — paginate_optional answers (False, ()) and a request made + with allow_statuses covering 403 answers None. + """ + + def _deny(self): + raise BitbucketApiError(403, "https://api.bitbucket.org/2.0/x", "no access") + + def branches(self, repo): + self._deny() + + def commits_between(self, repo, include, exclude): + self._deny() + + def paginate(self, path, **kwargs): + self._deny() + + def paginate_optional(self, path, **kwargs): + return False, iter(()) + + def request(self, method, path, **kwargs): + allowed = kwargs.get("allow_statuses") or () + if 403 in allowed: + return None + self._deny() + + +class BrokenClient(FakeClient): + """Every request fails with a retry-exhausted 500 — a transient outage.""" + + def _boom(self): + raise BitbucketApiError(500, "https://api.bitbucket.org/2.0/x", "server error") + + branches = lambda self, repo: self._boom() # noqa: E731 + commits_between = lambda self, repo, include, exclude: self._boom() # noqa: E731 + paginate = lambda self, path, **kw: self._boom() # noqa: E731 + paginate_optional = lambda self, path, **kw: self._boom() # noqa: E731 + request = lambda self, method, path, **kw: self._boom() # noqa: E731 + + +@pytest.mark.parametrize("stream_class", every_stream_class(), ids=lambda c: c.__name__) +class TestEveryStreamSurvivesADeniedRepository: + """The no-gaps guarantee: no stream may fail the sync over a 403 repository.""" + + def test_denied_repository_never_fails_the_sync(self, stream_class): + repo = repository() + client = FullyDeniedClient() + stream = stream_class(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)}) + if hasattr(stream, "state"): + stream.state = {} + + records, error = read_all_buckets(stream) + + assert error is None, f"{stream_class.__name__} failed the sync over a denied repository" + items = [r for r in records if r.get("record_type") == "item"] + if stream_class.__name__ == "RepositoriesStream": + # Its data comes from the workspace listing, which succeeded — the + # repository is visible, only its contents are not. Emitting the + # metadata is correct. + assert items, "the workspace listing was readable; the repository row should be emitted" + else: + assert items == [], f"{stream_class.__name__} emitted items from a repository it could not read" + # Any marker touching the denied repository must say unavailable — + # otherwise dbt treats the denied read as a legitimate empty collection + # and deletes rows. Markers of unrelated empty buckets may stay + # available: their partitions contain nothing to delete. + repo_bucket = repository_bucket(repo_state_key(repo)) + touching = [ + m + for m in records + if m.get("record_type") == "snapshot_complete" + and (m.get("repository_uuid") == repo.uuid or m.get("bucket_id") == repo_bucket) + ] + if stream_class.__name__ != "RepositoriesStream": + assert all(m["snapshot_available"] is False for m in touching), ( + f"{stream_class.__name__} marked a denied read as an available snapshot" + ) + + def test_denied_repository_state_never_advances(self, stream_class): + repo = repository() + client = FullyDeniedClient() + stream = stream_class(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)}) + if not hasattr(stream, "state"): + pytest.skip("full-refresh stream keeps no state") + stream.state = {} + + read_all_buckets(stream) + + assert stream.state["repositories"].get(repo_state_key(repo), {}) in ({}, None) or ( + "head_shas" not in stream.state["repositories"].get(repo_state_key(repo), {}) + and "updated_on" not in stream.state["repositories"].get(repo_state_key(repo), {}) + ), f"{stream_class.__name__} advanced state for a repository it never read" + + +class TestCredentialFailureAbortsLoudly: + """401 is global, not per-repository: quarantining every repo one by one + would drown the log and end in a generic message. Abort at the first one + with the actionable cause.""" + + def test_401_aborts_immediately_with_the_cause(self): + good, other = repository(slug="one"), repository(slug="two", uuid="{two}") + stream, catalog = build(CommitsStream, [good, other], denied(401)) + stream.state = {} + + with pytest.raises(RuntimeError, match="authentication failed"): + for bucket in range(BUCKET_COUNT): + list(stream.read_records(SyncMode.incremental, stream_slice={"bucket_id": bucket})) + + assert catalog.inaccessible_count == 0, "401 must not mark repositories denied — the token is the problem" + assert stream._failed_repositories == [], "and must not be recorded as per-repository failures" + + +class TestVanishedDiffstatIsTolerated: + """A commit's diffstat can be permanently gone (orphaned merge parents, + rewritten history) — the pre-rewrite connector tolerated exactly this + (`ignore_404`). It must mark that commit's snapshot unavailable, not fail + the repository on every sync forever.""" + + def test_missing_diffstat_marks_snapshot_unavailable_not_the_sync(self, repo): + client = FakeClient() + client.branch_values[repo.uuid] = [branch("main", "head")] + client.commit_values = [{"hash": "gone", "date": "2026-06-01T00:00:00+00:00"}] + # the diffstat endpoint answers 404 -> paginate_optional -> (False, ()) + client.optional_values[client.repo_path(repo, "diffstat/gone")] = (False, []) + from source_bitbucket_cloud.streams.file_changes import FileChangesStream + + stream, catalog = build(FileChangesStream, [repo], client) + stream.state = {} + + records, error = read_all_buckets(stream) + + assert error is None, "one vanished diffstat must not fail the repository forever" + assert not catalog.is_inaccessible(repo) + markers = [r for r in records if r.get("record_type") == "snapshot_complete"] + assert markers and markers[0]["snapshot_available"] is False, ( + "the denial must be recorded — an available empty snapshot would read as " + "'this commit changed nothing' and zero its line counts" + ) + assert stream.state["repositories"][repo_state_key(repo)]["head_shas"] == ["head"], ( + "the repository still advances past the bad commit" + ) + + +class TestManyBranchRepositoriesChunkTheCommitRange: + """Bitbucket's include/exclude ceiling is undocumented (BCLOUD-13229); a + repository with hundreds of branches must not send them in one form.""" + + def test_includes_are_chunked_and_excludes_ride_along(self): + from source_bitbucket_cloud.client import BitbucketClient + + client = BitbucketClient("tok") + calls: list[list[tuple[str, str]]] = [] + + def fake_paginate(path, *, params=None, method="GET", data=None, **kwargs): + calls.append(list(data or [])) + return iter(()) + + client.paginate = fake_paginate + includes = [f"new{i:04d}" for i in range(250)] + excludes = [f"old{i:04d}" for i in range(30)] + + list(client.commits_between(repository(), includes, excludes)) + + assert len(calls) == 3 # 250 heads / 100 per chunk + for form in calls: + chunk_includes = [v for k, v in form if k == "include"] + chunk_excludes = sorted(v for k, v in form if k == "exclude") + assert len(chunk_includes) <= 100 + assert chunk_excludes == sorted(excludes), "every chunk must carry the FULL exclude set" + fetched = sorted(v for form in calls for k, v in form if k == "include") + assert fetched == sorted(includes), "the union of chunks must cover every head exactly once" + + +class TestFeatureLevelDenialStaysFeatureLevel: + """Pipelines is a per-repository feature: a 403 there means "no pipelines + visible", not "this repository is unreadable". It must not mark the whole + repository inaccessible — that would suppress its commits and pull requests. + + These paths only run with pre-existing pipeline state, which the all-denied + matrix above never reaches, so they are pinned separately. + """ + + def test_open_pipeline_refetch_403_does_not_poison_the_repository(self): + from source_bitbucket_cloud.streams.metric_events import PipelinesStream + + repo = repository() + + class PipelinesDeniedClient(FakeClient): + def request(self, method, path, **kwargs): + allowed = kwargs.get("allow_statuses") or () + if 403 in allowed: + return None # what the real client answers for a tolerated 403 + raise BitbucketApiError(403, path, "no access") + + client = PipelinesDeniedClient() + client.optional_values["repositories/ws/repo/pipelines"] = (True, []) + catalog = FakeCatalog([repo], client) + stream = PipelinesStream(**{**SHARED, "client": client, "catalog": catalog}) + stream.state = { + "version": 3, + "bucket_count": 8, + "repositories": {repo_state_key(repo): {"created_on": "2026-06-01T00:00:00+00:00", "open": ["p1"]}}, + } + + _, error = read_all_buckets(stream) + + assert error is None + assert not catalog.is_inaccessible(repo), ( + "a pipelines-only denial must not mark the repository inaccessible" + ) + + def test_test_reports_403_marks_the_snapshot_not_the_repository(self): + from source_bitbucket_cloud.streams.metric_events import PipelineStepTestReportsStream + + repo = repository() + + class ReportsDeniedClient(FakeClient): + def request(self, method, path, **kwargs): + allowed = kwargs.get("allow_statuses") or () + if 403 in allowed: + return None + raise BitbucketApiError(403, path, "no access") + + client = ReportsDeniedClient() + client.optional_values["repositories/ws/repo/pipelines"] = ( + True, + [{"uuid": "p1", "created_on": "2026-06-02T00:00:00+00:00", "state": {"name": "COMPLETED"}}], + ) + client.optional_values["repositories/ws/repo/pipelines/p1/steps"] = (True, [{"uuid": "s1"}]) + catalog = FakeCatalog([repo], client) + stream = PipelineStepTestReportsStream(**{**SHARED, "client": client, "catalog": catalog}) + stream.state = {} + + records, error = read_all_buckets(stream) + + assert error is None + assert not catalog.is_inaccessible(repo) + markers = [r for r in records if r.get("record_type") == "snapshot_complete"] + assert markers and all(m["snapshot_available"] is False for m in markers) + + +@pytest.mark.parametrize("stream_class", every_stream_class(), ids=lambda c: c.__name__) +class TestEveryStreamSurfacesTransientFailures: + """The counterpart: a real outage must never be silently absorbed.""" + + def test_500_fails_the_sync_loudly(self, stream_class): + repo = repository() + client = BrokenClient() + stream = stream_class(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)}) + if hasattr(stream, "state"): + stream.state = {} + + _, error = read_all_buckets(stream) + + if type(stream).__name__ == "RepositoriesStream": + pytest.skip("reads only the already-fetched catalog; no per-repository request to fail") + assert error is not None, f"{stream_class.__name__} silently swallowed a 500" diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py index 2767fa804..bb71f4baa 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py +++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py @@ -73,7 +73,7 @@ def test_no_failures_no_error(self): assert error is None assert len(records) == 1 - assert stream.state["repositories"][repo_state_key(good)] == {"head_shas": ["a1"]} + assert stream.state["repositories"][repo_state_key(good)] == {"head_shas": ["a1"], "repo_updated_on": "2026-06-01T00:00:00+00:00"} class TestIssuesDisabledRepos: @@ -154,6 +154,14 @@ def fake_paginate(path, *, params=None, method="GET", data=None, **kwargs): assert seen["params"] == {"pagelen": "100"} assert ("include", "new1") in seen["data"] and ("exclude", "old1") in seen["data"] + def test_commits_between_without_current_heads_asks_nothing(self): + client = self.make_client() + calls = [] + + client.paginate = lambda *args, **kwargs: calls.append(kwargs) or iter(()) + assert list(client.commits_between(repository(), [], ["old1"])) == [] + assert calls == [] + class TestNewCommits404Recovery: def make_stream(self, client, repo): diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_request_budget.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_request_budget.py new file mode 100644 index 000000000..8dde28ef2 --- /dev/null +++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_request_budget.py @@ -0,0 +1,216 @@ +"""Request-budget guarantees for large fleets (rate-limit survival). + +Two mechanisms, pinned separately: + +* idle gate — a repository whose `updated_on` (free with the workspace listing) + has not changed since the last pass costs the push-driven streams ZERO + requests: no branch listing, no commit range, no diffstat. +* shared selections — the PR / pipeline / issue listing happens once per + repository per sync and is shared through the catalog as a SLIM projection, + instead of each stream in the family re-listing (six times for PRs). + +Both matter at ~1,400 repositories against Bitbucket's ~1,000 req/h budget. +""" + +from __future__ import annotations + +from airbyte_cdk.models import SyncMode + +from source_bitbucket_cloud.streams.base import BUCKET_COUNT, repo_state_key +from source_bitbucket_cloud.streams.branches import BranchesStream +from source_bitbucket_cloud.streams.commits import CommitsStream +from source_bitbucket_cloud.streams.file_changes import FileChangesStream +from source_bitbucket_cloud.streams.pr_comments import PRCommentsStream +from source_bitbucket_cloud.streams.pr_commits import PRCommitsStream +from source_bitbucket_cloud.streams.pull_requests import PullRequestsStream +from tests.conftest import SHARED, FakeCatalog, FakeClient, branch + +UPDATED = "2026-06-01T00:00:00+00:00" + + +class CountingClient(FakeClient): + def __init__(self): + super().__init__() + self.branch_calls = 0 + self.pr_list_calls = 0 + + def branches(self, repo): + self.branch_calls += 1 + return self.branch_values.get(repo.uuid, []) + + def paginate(self, path, **kwargs): + if path.endswith("pullrequests"): + self.pr_list_calls += 1 + return super().paginate(path, **kwargs) + + +def read_all(stream, repo): + records = [] + for bucket in range(BUCKET_COUNT): + records.extend(stream.read_records(SyncMode.incremental, stream_slice={"bucket_id": bucket})) + return records + + +class TestIdleGate: + def _synced_state(self, repo, extra=None): + return { + "version": 3, + "bucket_count": 8, + "repositories": {repo_state_key(repo): {"repo_updated_on": UPDATED, **(extra or {})}}, + } + + def test_unchanged_repository_costs_zero_requests(self, repo): + client = CountingClient() + client.branch_values[repo.uuid] = [branch("main", "a1")] + for cls, extra in ( + (CommitsStream, {"head_shas": ["a1"]}), + (FileChangesStream, {"head_shas": ["a1"]}), + (BranchesStream, {}), + ): + stream = cls(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)}) + stream.state = self._synced_state(repo, extra) + + records = read_all(stream, repo) + + assert records == [], f"{cls.__name__} emitted for an idle repository" + assert client.branch_calls == 0, "an idle repository must not be listed at all" + assert client.commit_calls == [] + + def test_changed_updated_on_syncs_again(self, repo): + client = CountingClient() + client.branch_values[repo.uuid] = [branch("main", "a2")] + client.commit_values = [{"hash": "a2", "date": "2026-07-01T00:00:00+00:00"}] + stream = CommitsStream(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)}) + stream.state = { + "version": 3, + "bucket_count": 8, + "repositories": { + repo_state_key(repo): {"head_shas": ["a1"], "repo_updated_on": "2026-05-01T00:00:00+00:00"} + }, + } + + records = read_all(stream, repo) + + assert [r["hash"] for r in records] == ["a2"] + assert stream.state["repositories"][repo_state_key(repo)]["repo_updated_on"] == UPDATED + + def test_first_sync_is_never_gated(self, repo): + client = CountingClient() + client.branch_values[repo.uuid] = [branch("main", "a1")] + client.commit_values = [{"hash": "a1", "date": UPDATED}] + stream = CommitsStream(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)}) + stream.state = {} + + records = read_all(stream, repo) + + assert len(records) == 1 + + def test_legacy_state_without_the_field_is_not_gated(self, repo): + """Migrated pre-rewrite state has head_shas but no repo_updated_on: the + first pass must run (and thereby stamp the field).""" + client = CountingClient() + client.branch_values[repo.uuid] = [branch("main", "new")] + client.commit_values = [{"hash": "new", "date": UPDATED}] + stream = CommitsStream(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)}) + stream.state = {"ws/repo/main": {"head_sha": "old", "date": "2026-05-01T00:00:00+00:00"}} + + records = read_all(stream, repo) + + assert [r["hash"] for r in records] == ["new"] + assert client.commit_calls == [(["new"], ["old"])], "still resumes from the migrated heads" + + +def pr(pr_id=42): + return { + "id": pr_id, + "title": "t", + "description": "x" * 5000, + "state": "MERGED", + "updated_on": "2026-06-30T00:00:00+00:00", + "created_on": UPDATED, + "source": {"branch": {"name": "f"}, "commit": {"hash": "src"}}, + "destination": {"branch": {"name": "main"}, "commit": {"hash": "dst"}}, + "participants": [{"user": {"uuid": "{u}"}, "role": "REVIEWER"}] * 20, + } + + +class TestSharedPrSelection: + def _run(self, cls, repo, client, catalog): + stream = cls(**{**SHARED, "client": client, "catalog": catalog}) + stream.state = {} + return read_all(stream, repo) + + def test_children_reuse_the_parents_listing(self, repo): + client = CountingClient() + client.pr_values = [pr()] + catalog = FakeCatalog([repo], client) + + self._run(PullRequestsStream, repo, client, catalog) + after_parent = client.pr_list_calls + self._run(PRCommentsStream, repo, client, catalog) + self._run(PRCommitsStream, repo, client, catalog) + + assert after_parent >= 1 + assert client.pr_list_calls == after_parent, ( + "child streams re-listed pull requests instead of reusing the shared selection" + ) + + def test_cache_holds_slim_projections_only(self, repo): + """The memory guard: raw PR objects (description, participants, …) held + for every repository across six sequential streams would cost hundreds + of MB. Only four whitelisted fields may be cached.""" + client = CountingClient() + client.pr_values = [pr()] + catalog = FakeCatalog([repo], client) + + self._run(PullRequestsStream, repo, client, catalog) + + assert catalog.pr_selections, "the parent must fill the cache" + for slim_list, _state in catalog.pr_selections.values(): + for entry in slim_list: + assert set(entry) == {"id", "updated_on", "source", "destination"}, entry + assert set(entry["source"]) == {"commit"} + + def test_children_produce_identical_records_from_the_slim_cache(self, repo): + """Equivalence: a child fed from the cache emits exactly what it would + have emitted from its own fetch.""" + client_cached = CountingClient() + client_cached.pr_values = [pr()] + comments = (True, [{"id": 7, "content": {"raw": "lgtm"}, "user": {"uuid": "{u}"}}]) + client_cached.optional_values["repositories/ws/repo/pullrequests/42/comments"] = comments + catalog = FakeCatalog([repo], client_cached) + self._run(PullRequestsStream, repo, client_cached, catalog) # fills cache + from_cache = self._run(PRCommentsStream, repo, client_cached, catalog) + + client_fresh = CountingClient() + client_fresh.pr_values = [pr()] + client_fresh.optional_values["repositories/ws/repo/pullrequests/42/comments"] = ( + True, [{"id": 7, "content": {"raw": "lgtm"}, "user": {"uuid": "{u}"}}], + ) + fresh = self._run(PRCommentsStream, repo, client_fresh, FakeCatalog([repo], client_fresh)) + + # generation_id is derived from the stream instance's run id, so it (and + # unique_key, which embeds it) legitimately differs between two runs; + # the equivalence claim is about entity content. + volatile = {"collected_at", "generation_id", "unique_key"} + strip = lambda rows: [{k: v for k, v in r.items() if k not in volatile} for r in rows] + assert strip(from_cache) == strip(fresh) + + def test_divergent_watermark_fetches_its_own(self, repo): + """A stream whose state lags (failed last sync) must not reuse a + narrower selection.""" + client = CountingClient() + client.pr_values = [pr()] + catalog = FakeCatalog([repo], client) + self._run(PullRequestsStream, repo, client, catalog) + after_parent = client.pr_list_calls + + lagging = PRCommentsStream(**{**SHARED, "client": client, "catalog": catalog}) + lagging.state = { + "version": 3, + "bucket_count": 8, + "repositories": {repo_state_key(repo): {"updated_on": "2026-01-01T00:00:00+00:00", "reconcile_after_id": 0}}, + } + read_all(lagging, repo) + + assert client.pr_list_calls > after_parent, "a divergent watermark must trigger its own fetch"