fix(bitbucket-cloud): tolerate denied repos, cut request budget

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'

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="")

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: "

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"]}

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:

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 "")