fix(bitbucket-cloud): stop permanent stream failures, cut sync time - #2136
Conversation
Bitbucket answers 400 for a pull request whose source and destination share no ancestry. The diff is undefined rather than empty and no retry makes it computable, so the stream now records the snapshot as unavailable instead of failing the repository on every sync. Other 400s still surface. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
A branch listing can advertise a head that the commits endpoint cannot resolve; the range then 404s and the retry without excludes 404s again, costing the repository its whole history. The error names the offending shas, so they are dropped from the range and the remaining heads still sync. Reachability skips only the affected branch. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
The stored heads are only ever the exclude side of the next commit range, so a stale one can suppress nothing but a sha already reported. Overwriting them with an empty listing cost a full history re-read as soon as any branch reappeared. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
A repository whose updated_on predates start_date can hold nothing the window asks for: commits, branches and tags only appear by being pushed, and a push moves updated_on. Those repositories are now skipped without a request. On a first read of a repository still in the window, branch heads parked well before the window are left out of the commit range — they can only reach commits the date filter discards. Tags older than start_date are no longer emitted. Pull request, pipeline and issue selections stay ungated: comments, approvals and scheduled runs happen without a push. Refs #1998 for deployments, whose timestamp field needs verifying against a payload. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Streams are read in catalog order, so the ten the dbt models read now run before the ten that nothing reads yet. A sync cut short then still delivers the data the transform layer builds on. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
The catalog holds every branch of every repository for the length of a sync and nothing reads the API object behind a BranchRef, so it keeps the four fields the streams use. Memoised catalog fills are now guarded by a lock and the cache size is logged per stream. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
A bucket spends its time waiting on the API, and the per-repository reads are independent, so a bounded pool now fetches several at once while records still leave in submission order. Each worker buffers into a bounded queue, so a repository with a long history parks instead of holding it in memory, and failures reach the same classification they did serially. bitbucket_concurrency configures the width; 1 keeps the original loop. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Closing the read mid-bucket left the signal until after the pool had been asked to shut down, so a worker blocked on a full buffer was waited on by the very shutdown that would have freed it. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
…hole The cursor and the head set were stored even when the pass had not seen everything: a head the API refused to resolve was recorded as synced, and an empty branch listing advanced the timestamp the idle gate reads. Either way the next sync skipped the repository and the missing history stayed missing. A head that could not be read is now left out of the stored set and holds the cursor, so the next pass looks again; a head deliberately left out of the range still counts as read. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Branches are current state, not dated history: a repository last pushed before the window still has branches, and gating them left it with no branch snapshot at all. The idle gate already holds a dormant repository to a single listing, so the window bought nothing here. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Repository content must not carry claims about a real deployment: the schedule comment named job ids, dates and worker topology, and two test docstrings asserted behaviour had been seen in a live environment. The technical reasons are kept, stated as conditions rather than sightings. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
|
Warning Review limit reached
Next review available in: 29 minutes You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughBitbucket Cloud ingestion now supports bounded repository concurrency, thread-safe client access, structured API errors, conservative branch snapshots, start-date filtering, unresolved-head repair, explicit API field projections, and completeness-aware state persistence. ChangesBitbucket Cloud ingestion
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant SourceBitbucketCloud
participant BitbucketStream
participant ThreadPoolExecutor
participant BitbucketClient
participant RepositoryCatalog
SourceBitbucketCloud->>BitbucketStream: configure concurrency
BitbucketStream->>ThreadPoolExecutor: submit repository reads
ThreadPoolExecutor->>BitbucketClient: fetch repository data
BitbucketClient->>RepositoryCatalog: update shared caches
ThreadPoolExecutor-->>BitbucketStream: return ordered records or errors
BitbucketStream-->>SourceBitbucketCloud: persist complete state
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
A first read of a commit range is bounded only by how much history a repository holds, so putting it ahead of the watermark-bounded streams let it starve them — a run that ended early then delivered neither. The streams the transform layer reads still come first, bounded ones first. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
A snapshot replaces the one before it, so publishing an empty one deletes every branch the repository had — and a listing that finds them again the next pass emits no correction. A repository known to have branches now publishes nothing when the listing comes back empty, and holds its cursor so the listing is retried. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
An empty listing marked every stored branch deleted, from a read already treated as incomplete, with no later correction. It now emits nothing at all in that case. The range behind each branch is also no longer materialised: only its head is held, which is where a stale exclude is rejected, so recovery still sees an unreported range while a long history streams. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
target is the commit a tag points at, which can be far older than the tag itself, so a release cut today against old history was dropped from a snapshot reported as complete. Tags carrying no date of their own are kept rather than judged by their commit. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
State built without the setter running carried no version, and reading it back matched the pre-rewrite shape: it was then reshaped into something addressing nothing, silently discarding every field the current streams keep. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py (1)
266-291: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftSignal when a tolerated failure truncates pagination.
paginate_optionalreturnspresent=Trueonce the first page succeeds. If_optional_requesttolerates a 403, 404, or tolerated 400 on a later page,records()returns fewer items while callers still passpresenttocomplete(). Streams such asrepository_recordsand_diffstatmark the result as available with a partialsnapshot_item_count, and consumers that keep only the latest snapshot treat the missing entities as deleted.Return a mutable availability flag that stays
Trueonly if pagination drains without refusal, or raise when a later page is refused so the sync fails instead of publishing a partial complete snapshot.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py` around lines 266 - 291, The records() pagination flow must signal when a later page is refused instead of returning a partial successful result. Update paginate_optional and records() to share a mutable availability state that remains true only when all pages are fetched, or propagate an error from _optional_request on later-page refusal; ensure callers’ present value is updated before complete() publishes the snapshot.
🧹 Nitpick comments (16)
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py (1)
101-101: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAnnotate the remaining parameters and the return type of
_changes.The signature was changed here, and
repo,branch, and the return type carry no hints. The coding guidelines require type hints on every function and method signature.♻️ Proposed change
- def _changes(self, repo, branch, include: str, exclude: str | None, action: str, unresolved: set[str]): + def _changes( + self, + repo: RepositoryRef, + branch: BranchRef, + include: str, + exclude: str | None, + action: str, + unresolved: set[str], + ) -> Iterator[Mapping[str, Any]]:
RepositoryRefandBranchRefcome fromsource_bitbucket_cloud.client, which this module already imports from.As per coding guidelines: "Put type hints on every function and method signature; do not allow bare
Anyto escape."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py` at line 101, Update the `_changes` method signature to annotate `repo` with `RepositoryRef`, `branch` with `BranchRef`, and add the appropriate return type, reusing the existing symbols imported from `source_bitbucket_cloud.client` and preserving the current parameter behavior.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py (1)
77-98: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReduce the nesting in the repair loop.
The body of
new_commitsreaches four indentation levels (for→try/except→if→if). The coding guidelines ask for at most two levels and for named helpers for repetition. Extract the pruning decision into a small pure helper that maps(includes, excludes, missing)to the next(includes, excludes)pair, then keep the loop flat.Also note that
last_erroris bound only inside theexceptblock. The binding is safe today because every loop iteration returns, raises, or assigns it, but the invariant is implicit. Initialize it before the loop to make the finalraiseobviously well defined.♻️ Proposed refactor
includes = list(current_heads) excludes = list(previous_heads) + last_error: BitbucketApiError | None = None for _ in range(RANGE_REPAIR_ATTEMPTS):As per coding guidelines: "Prefer early returns over nested conditionals; aim to keep function bodies to at most two indentation levels."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py` around lines 77 - 98, Reduce nesting in new_commits by extracting the includes/excludes pruning decision into a pure named helper that accepts (includes, excludes, missing) and returns the next pair, preserving the existing empty-includes and excludes-clearing behavior. Initialize last_error before the repair loop so the final raise is explicitly defined, then use early returns/continues in the loop to keep its body within two indentation levels.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py (2)
20-24: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd type hints to the new helpers.
read_all_buckets,synced_state,build, andknownhave no parameter or return annotations. The coding guidelines require hints on every function and method signature, and the path pattern covers test modules undersrc/ingestion/.As per coding guidelines: "Put type hints on every function and method signature; do not allow bare
Anyto escape."Also applies to: 35-41
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py` around lines 20 - 24, Add type annotations to the helper functions read_all_buckets, synced_state, build, and known, covering every parameter and return value. Use specific existing types where available and avoid allowing bare Any to escape, while preserving their current behavior.Source: Coding guidelines
27-32: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winImport production constants in the new tests instead of copying their values. Three new test modules hardcode values that the connector already exports. If a constant changes, a seeded state can silently stop matching the runtime shape, or a bound assertion stops testing the documented rule.
src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py#L27-L32: replace"bucket_count": 8with the already-importedBUCKET_COUNT.src/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.py#L169-L173: replace"bucket_count": 8with the already-importedBUCKET_COUNT.src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py#L146-L151: importRANGE_REPAIR_ATTEMPTSfromsource_bitbucket_cloud.streams.git_rangesand assertlen(client.commit_calls) <= RANGE_REPAIR_ATTEMPTS.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py` around lines 27 - 32, Use exported production constants in all three tests: update synced_state in src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py#L27-L32 and the corresponding state setup in src/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.py#L169-L173 to use the already-imported BUCKET_COUNT instead of 8; in src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py#L146-L151, import RANGE_REPAIR_ATTEMPTS from source_bitbucket_cloud.streams.git_ranges and bound commit_calls using that constant instead of a literal.src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py (1)
27-47: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMove the shared range pass into
CommitRangeMixin.This block is now byte-identical to
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.pylines 28-48, except for the per-commit emission. Both copies compute the snapshot, choose cold includes, trackunresolved, filter retained heads, and write the same two state keys. Any future change to the completeness rule must be applied in two places.Consider adding one helper to
CommitRangeMixinthat owns the traversal and the state write, and let each stream supply only the record projection.♻️ Sketch
# git_ranges.py def range_pass(self, repo, prior, branches, current_heads): """Yields raw commits and returns the state dict to commit.""" ...As per coding guidelines: "Use named helpers for repetition" and one concern per module.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py` around lines 27 - 47, Extract the duplicated snapshot, commit traversal, unresolved tracking, retained-head filtering, completeness calculation, and state-payload construction from the commits stream into a named helper on CommitRangeMixin, such as range_pass. Update the commits stream to call that helper and retain only its per-commit record projection, date filtering, yielding, and final state commit; update file_changes.py to use the same helper while preserving its existing projection behavior.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py (1)
19-22: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueShorten the explanatory comments to one line.
The coding guidelines limit explanatory comments to one line. Several comments added in this file span two or three lines: Lines 19-21, Lines 68-69, Lines 86-88, and Lines 155-156. The reasons are valuable. Condense each to a single line.
As per coding guidelines: "Add comments only when code cannot express the reason, such as intentional redundancy, cross-function invariants, or reasoned workarounds; keep them to one line."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py` around lines 19 - 22, Condense the explanatory comments associated with UNCOMPUTABLE_DIFF and the other referenced changes in the Bitbucket Cloud client into single-line comments, preserving each comment’s rationale and meaning without changing the implementation.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py (2)
380-396: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueCollapse the three
BUCKET_COUNT - 1checks into one guard.
finish_bucketnow testsbucket_id == BUCKET_COUNT - 1three times. Use one early return so the last-bucket work reads as one block.♻️ Proposed refactor
def finish_bucket(self, bucket_id: int, repositories: Sequence[RepositoryRef]) -> None: del repositories - if bucket_id == BUCKET_COUNT - 1: - cached_repositories, cached_branches = self._catalog.branch_cache_size - logger.info( - f"{self.name}: branch_cache repositories={cached_repositories} branches={cached_branches}" - ) - if bucket_id == BUCKET_COUNT - 1 and self._skipped_repositories: + if bucket_id != BUCKET_COUNT - 1: + return + + cached_repositories, cached_branches = self._catalog.branch_cache_size + logger.info(f"{self.name}: branch_cache repositories={cached_repositories} branches={cached_branches}") + + if 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: + if self._failed_repositories: raise RuntimeError( f"{self.name}: {len(self._failed_repositories)} repositories failed this sync: " + ", ".join(self._failed_repositories[:10]) )As per coding guidelines: "Prefer early returns over nested conditionals".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py` around lines 380 - 396, Update finish_bucket to return immediately when bucket_id is not BUCKET_COUNT - 1, then place the branch-cache logging, skipped-repository logging, and failed-repository error handling under the single remaining last-bucket guard. Preserve the existing behavior and ordering of all three operations.Source: Coding guidelines
246-256: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winThe refill bound must not exceed
max_workers. Make that explicit.The design is correct today because
len(pending) < self._concurrencysubmits at mostself._concurrencytasks and the pool has exactlyself._concurrencyworkers. Every submitted task therefore runs immediately.That equality is load-bearing. If the refill bound grows above
max_workers, a submitted task can sit in the executor queue behind workers that are parked in_offeron a full buffer. Those workers only unpark when the consumer drains them, and the consumer drainspendingin order. When the queued task reaches the head ofpending, its_collecthas never started, so_consumespins onqueue.Emptywithstopclear and never returns. The read hangs.Bind the two values together so a later change cannot separate them.
🛡️ Proposed guard
stop = threading.Event() pending: deque[_PendingRead] = deque() waiting = iter(repositories) with ThreadPoolExecutor(max_workers=self._concurrency) as pool: + # A parked worker holds its pool thread, so never submit more reads + # than the pool can run at once or the head of `pending` may never start. + in_flight = self._concurrency try: while True: - while len(pending) < self._concurrency: + while len(pending) < in_flight:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py` around lines 246 - 256, Update the refill loop around _submit and _consume so its submission limit is explicitly bound to the executor’s max_workers, preserving the invariant that pending never exceeds available workers. Bind the concurrency setting and pool worker count together or validate their equality before submitting, so future changes cannot allow queued tasks beyond max_workers.src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py (3)
25-34: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueBound the search loop so a bucketing change fails instead of hanging.
fleet_in_one_bucketloops until it findssizerepositories in the target bucket. Nothing limitsindex. Ifrepository_bucketever stops mapping any candidate tobucket, the test suite hangs with no diagnostic.♻️ Proposed refactor
def fleet_in_one_bucket(size: int, bucket: int = 0): """Repositories that all land in the same slice, so one read_records call really does run several workers.""" repos, index = [], 0 - while len(repos) < size: + limit = size * BUCKET_COUNT * 20 + while len(repos) < size and index < limit: repo = repository(slug=f"same{index:03d}", uuid=f"{{s-{index}}}") if repository_bucket(repo_state_key(repo)) == bucket: repos.append(repo) index += 1 + assert len(repos) == size, f"found {len(repos)} of {size} repositories for bucket {bucket}" return repos🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py` around lines 25 - 34, Bound the candidate-generation loop in fleet_in_one_bucket with a finite maximum index or attempt count, and fail explicitly with a useful assertion when size repositories are not found in the requested bucket. Preserve the existing repository generation and successful return behavior.
139-159: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert that the healthy repositories still advance their state.
The suite proves the victim does not advance (Line 178). It does not prove the complement: that one failing repository leaves the other repositories in the same bucket free to commit their state. That is the isolation guarantee the
_read_concurrentlydocstring claims.Add the assertion to
test_a_transient_failure_still_fails_the_sync, or add a test named for the rule.The parametrized asserts at Lines 147-149 also carry no message. The guidelines ask for the failing case in the message.
💚 Proposed test changes
`@pytest.mark.parametrize`("status", [403, 404]) def test_a_denied_repository_is_skipped_not_failed(self, status): repos = fleet() client = self.denied_client(repos, "repo05", status) stream = build(repos, client, 4) records = read_all_buckets(stream) - assert stream._failed_repositories == [] - assert "ws/repo05" in stream._skipped_repositories - assert {r["repo_slug"] for r in records} == {repo.slug for repo in repos} - {"repo05"} + assert stream._failed_repositories == [], f"HTTP {status} must not count as a failure" + assert "ws/repo05" in stream._skipped_repositories, f"HTTP {status} must skip the repository" + assert {r["repo_slug"] for r in records} == {repo.slug for repo in repos} - {"repo05"}, ( + f"every other repository must still be read after HTTP {status}" + )def test_a_transient_failure_still_fails_the_sync(self): repos = fleet() client = self.denied_client(repos, "repo05", 500) stream = build(repos, client, 4) with pytest.raises(RuntimeError, match="repositories failed"): read_all_buckets(stream) assert stream._failed_repositories == ["ws/repo05"] + + def test_one_failure_does_not_hold_back_its_neighbours_state(self): + repos = fleet() + victim = next(repo for repo in repos if repo.slug == "repo05") + neighbours = [ + repo + for repo in repos + if repo is not victim + and repository_bucket(repo_state_key(repo)) == repository_bucket(repo_state_key(victim)) + ] + stream = build(repos, self.denied_client(repos, "repo05", 500), 4) + + with pytest.raises(RuntimeError): + read_all_buckets(stream) + + for repo in neighbours: + assert repo_state_key(repo) in stream.state["repositories"], ( + f"a failure in repo05 must not block: {repo.slug}" + )As per path instructions: "include the failing case in assertion messages".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py` around lines 139 - 159, Extend test_a_transient_failure_still_fails_the_sync to assert that healthy repositories in the same bucket still advance their state despite repo05’s transient failure, matching the isolation guarantee documented by _read_concurrently. Add failing-case context to the parametrized assertions in test_a_denied_repository_is_skipped_not_failed, including the status and relevant repository details in each assertion message.Source: Coding guidelines
21-80: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd type hints to the new helpers and fake methods.
The path instructions require type hints on every function and method signature under
src/ingestion/**/*.{py,pyi}, and this test file is in that tree. The existing fakes intests/conftest.pyalready follow it, for exampledef branches(self, repo: RepositoryRef) -> list[BranchRef].These signatures have no hints or incomplete hints:
fleet(Line 21),fleet_in_one_bucket(Line 25),FleetClient.__init__repos(Line 41),FleetClient.branches(Line 49),FleetClient.commits_between(Line 56),read_all_buckets(Line 62),buildreposandclient(Line 69), andcomparable(Line 77).♻️ Proposed change for the module-level helpers
-def fleet(size: int = 12): +def fleet(size: int = 12) -> list[RepositoryRef]: return [repository(slug=f"repo{index:02d}", uuid=f"{{r-{index}}}") for index in range(size)] -def fleet_in_one_bucket(size: int, bucket: int = 0): +def fleet_in_one_bucket(size: int, bucket: int = 0) -> list[RepositoryRef]:-def read_all_buckets(stream): +def read_all_buckets(stream: CommitsStream) -> list[Mapping[str, Any]]: records = []-def build(repos, client, concurrency: int): +def build(repos: Sequence[RepositoryRef], client: FakeClient, concurrency: int) -> CommitsStream:-def comparable(records): +def comparable(records: Iterable[Mapping[str, Any]]) -> list[tuple[tuple[str, str], ...]]:As per path instructions: "Put type hints on every function and method signature; do not allow bare
Anyto escape."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py` around lines 21 - 80, Add complete, non-bare type annotations to the listed helpers and FleetClient methods: annotate collection, repository, client, stream, record, and return types using the existing project symbols and conventions from tests/conftest.py. Update fleet, fleet_in_one_bucket, FleetClient.__init__, branches, commits_between, read_all_buckets, build, and comparable without changing their behavior, and ensure no untyped Any escapes.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py (1)
32-38: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFlatten the range loop to two indentation levels.
The block reaches three levels:
if→for→if. Extract the start-window filter into a named helper so the loop body stays flat and the rule it applies is named.As per coding guidelines: "Prefer early returns over nested conditionals; aim to keep function bodies to at most two indentation levels."
♻️ Proposed refactor
+ def _in_window(self, committed_date: Any) -> bool: + return not (self._start_date and committed_date and str(committed_date)[:10] < self._start_date)- if current_head_shas != previous_head_shas: - includes = current_head_shas if previous_head_shas else self.cold_includes(branches) - for commit in self.new_commits(repo, includes, previous_head_shas, unresolved): - committed_date = commit.get("date") - 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) + if current_head_shas != previous_head_shas: + includes = current_head_shas if previous_head_shas else self.cold_includes(branches) + commits = self.new_commits(repo, includes, previous_head_shas, unresolved) + in_window = (commit for commit in commits if self._in_window(commit.get("date"))) + for commit in in_window: + yield from self._diffstat(repo, str(commit.get("hash") or ""), commit.get("date"))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py` around lines 32 - 38, Extract the start-date filtering condition from the commit loop in the relevant stream method into a named helper, such as a predicate for whether a commit is before the configured start window. Use that helper with an early continue so the loop body remains at two indentation levels, preserving the existing behavior for missing dates, unset _start_date, and qualifying commits passed to _diffstat.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py (2)
87-89: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winInclude
bodyin the assertion message.Line 89 omits the failing parameter value. Add
body!rto the assertion message.Proposed fix
- assert BitbucketApiError(400, "u", body).error_message == "" + assert BitbucketApiError(400, "u", body).error_message == "", f"should reject: {body!r}"As per coding guidelines, “Use
pytest.mark.parametrizefor copy-pasted test cases and include the failing case in assertion messages.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py` around lines 87 - 89, Update test_unparseable_bodies_have_no_message so its assertion message includes the parametrized body value using body!r, while preserving the existing expectation that error_message is empty.Source: Coding guidelines
13-125: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd annotations to all new callables.
The new helpers, fake methods, and tests omit parameter or return annotations, including
read_all_buckets(stream),client_with(routes),pull_request(), the fake classes,build(diffstat_response), and the test methods in both files. Add concrete argument annotations and use-> Nonefor test methods.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py` around lines 13 - 125, Add concrete parameter and return annotations to every new callable in src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py (lines 13-125), including read_all_buckets, FakeResponse methods, FakeSession methods, client_with, pull_request, build, and all test methods; annotate test methods with -> None. Apply the same annotation update to every affected new callable in src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py (lines 69-103), including -> None for its test methods.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py (1)
132-136: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReduce this stream-order comment to one line.
Keep the reason for the ordering, but express it in one concise line.
As per coding guidelines, “Add comments only when code cannot express the reason, such as intentional redundancy, cross-function invariants, or reasoned workarounds; keep them to one line.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py` around lines 132 - 136, Condense the multi-line stream-order comment to one concise line explaining that cost-watermark-bounded streams run before commit-range streams so interrupted syncs retain the bounded results consumed by the transform layer.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py (1)
64-65: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRemove the prose that duplicates the behavioral test names.
The test names already state the ordering rules. Keep the tests as the semantic documentation. Remove these comments and docstrings.
As per coding guidelines, “Test non-obvious semantics with a test whose name states the rule rather than adding a comment.”
Also applies to: 89-103
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py` around lines 64 - 65, Remove the explanatory comments and docstrings around the repository history ordering tests in the Bitbucket Cloud test module, including the prose near the initial repository read and the additional section identified by the review. Keep the behavioral test names and test logic unchanged as the semantic documentation.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py`:
- Around line 153-167: Ensure sessions created by the thread-local _session
property are closed when the concurrent read work completes, by tracking
worker-created requests.Session instances and adding cleanup to the
sync/stream/executor lifecycle. Alternatively, reuse a single executor per
stream so each worker’s session is reused and explicitly closed when that
executor shuts down.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py`:
- Around line 228-303: The concurrent reader should be extracted from base.py
into a new streams/concurrent_reads.py module. Move _PendingRead, RECORD_BUFFER,
QUEUE_POLL_SECONDS, _READ_DONE, and the _read_concurrently, _submit, _collect,
_consume, and _offer logic there, parameterizing it with a per-repository record
producer while preserving submission ordering, backpressure, error handling, and
cancellation. Update BitbucketStream to retain only selection between serial and
concurrent readers and use the extracted reader.
In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py`:
- Around line 182-199: Update test_a_long_history_does_not_buffer_without_bound
to use fleet_in_one_bucket so both repositories share the same bucket, then
consume that bucket directly instead of calling read_all_buckets, ensuring two
workers contend within one slice. Rename the test to reflect its verified
behavior—preserving all records when a worker parks on a full buffer—unless you
add an assertion that explicitly measures the buffer bound.
---
Outside diff comments:
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py`:
- Around line 266-291: The records() pagination flow must signal when a later
page is refused instead of returning a partial successful result. Update
paginate_optional and records() to share a mutable availability state that
remains true only when all pages are fetched, or propagate an error from
_optional_request on later-page refusal; ensure callers’ present value is
updated before complete() publishes the snapshot.
---
Nitpick comments:
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py`:
- Around line 19-22: Condense the explanatory comments associated with
UNCOMPUTABLE_DIFF and the other referenced changes in the Bitbucket Cloud client
into single-line comments, preserving each comment’s rationale and meaning
without changing the implementation.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py`:
- Around line 132-136: Condense the multi-line stream-order comment to one
concise line explaining that cost-watermark-bounded streams run before
commit-range streams so interrupted syncs retain the bounded results consumed by
the transform layer.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py`:
- Around line 380-396: Update finish_bucket to return immediately when bucket_id
is not BUCKET_COUNT - 1, then place the branch-cache logging, skipped-repository
logging, and failed-repository error handling under the single remaining
last-bucket guard. Preserve the existing behavior and ordering of all three
operations.
- Around line 246-256: Update the refill loop around _submit and _consume so its
submission limit is explicitly bound to the executor’s max_workers, preserving
the invariant that pending never exceeds available workers. Bind the concurrency
setting and pool worker count together or validate their equality before
submitting, so future changes cannot allow queued tasks beyond max_workers.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py`:
- Line 101: Update the `_changes` method signature to annotate `repo` with
`RepositoryRef`, `branch` with `BranchRef`, and add the appropriate return type,
reusing the existing symbols imported from `source_bitbucket_cloud.client` and
preserving the current parameter behavior.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py`:
- Around line 27-47: Extract the duplicated snapshot, commit traversal,
unresolved tracking, retained-head filtering, completeness calculation, and
state-payload construction from the commits stream into a named helper on
CommitRangeMixin, such as range_pass. Update the commits stream to call that
helper and retain only its per-commit record projection, date filtering,
yielding, and final state commit; update file_changes.py to use the same helper
while preserving its existing projection behavior.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py`:
- Around line 32-38: Extract the start-date filtering condition from the commit
loop in the relevant stream method into a named helper, such as a predicate for
whether a commit is before the configured start window. Use that helper with an
early continue so the loop body remains at two indentation levels, preserving
the existing behavior for missing dates, unset _start_date, and qualifying
commits passed to _diffstat.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py`:
- Around line 77-98: Reduce nesting in new_commits by extracting the
includes/excludes pruning decision into a pure named helper that accepts
(includes, excludes, missing) and returns the next pair, preserving the existing
empty-includes and excludes-clearing behavior. Initialize last_error before the
repair loop so the final raise is explicitly defined, then use early
returns/continues in the loop to keep its body within two indentation levels.
In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py`:
- Around line 25-34: Bound the candidate-generation loop in fleet_in_one_bucket
with a finite maximum index or attempt count, and fail explicitly with a useful
assertion when size repositories are not found in the requested bucket. Preserve
the existing repository generation and successful return behavior.
- Around line 139-159: Extend test_a_transient_failure_still_fails_the_sync to
assert that healthy repositories in the same bucket still advance their state
despite repo05’s transient failure, matching the isolation guarantee documented
by _read_concurrently. Add failing-case context to the parametrized assertions
in test_a_denied_repository_is_skipped_not_failed, including the status and
relevant repository details in each assertion message.
- Around line 21-80: Add complete, non-bare type annotations to the listed
helpers and FleetClient methods: annotate collection, repository, client,
stream, record, and return types using the existing project symbols and
conventions from tests/conftest.py. Update fleet, fleet_in_one_bucket,
FleetClient.__init__, branches, commits_between, read_all_buckets, build, and
comparable without changing their behavior, and ensure no untyped Any escapes.
In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py`:
- Around line 20-24: Add type annotations to the helper functions
read_all_buckets, synced_state, build, and known, covering every parameter and
return value. Use specific existing types where available and avoid allowing
bare Any to escape, while preserving their current behavior.
- Around line 27-32: Use exported production constants in all three tests:
update synced_state in
src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py#L27-L32
and the corresponding state setup in
src/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.py#L169-L173
to use the already-imported BUCKET_COUNT instead of 8; in
src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py#L146-L151,
import RANGE_REPAIR_ATTEMPTS from source_bitbucket_cloud.streams.git_ranges and
bound commit_calls using that constant instead of a literal.
In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py`:
- Around line 64-65: Remove the explanatory comments and docstrings around the
repository history ordering tests in the Bitbucket Cloud test module, including
the prose near the initial repository read and the additional section identified
by the review. Keep the behavioral test names and test logic unchanged as the
semantic documentation.
In
`@src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py`:
- Around line 87-89: Update test_unparseable_bodies_have_no_message so its
assertion message includes the parametrized body value using body!r, while
preserving the existing expectation that error_message is empty.
- Around line 13-125: Add concrete parameter and return annotations to every new
callable in
src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py
(lines 13-125), including read_all_buckets, FakeResponse methods, FakeSession
methods, client_with, pull_request, build, and all test methods; annotate test
methods with -> None. Apply the same annotation update to every affected new
callable in src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py
(lines 69-103), including -> None for its test methods.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 30962108-aa26-4b60-a5fe-6708aa22f5d0
📒 Files selected for processing (22)
src/ingestion/connectors/git/bitbucket-cloud/descriptor.yamlsrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/spec.jsonsrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_diffstat.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/conftest.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_branches.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_client.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_source.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py
Pipelines, tags, deployments, environments, issues and pull-request tasks have no model reading them, yet collecting them took most of the time a full pass spends. Ordering them last only helped a run that was cut short; not syncing them at all helps every run. The classes, schemas and tests stay, so re-enabling one is uncommenting it. Two things come with that: the platform may have dropped the state of a stream it stopped seeing, so the first pass back can be a backfill, and tags pages a repository's whole tag history every time, so it wants the idle gate branches uses before it returns. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Consuming in submission order let one deep history park every other worker on a full buffer and block new submissions behind it — the pool degraded to serial for the duration of the largest repository. Records now leave as they become ready and interleave across repositories, which the append-only, keyed bronze layer does not mind. Failures stay attributed through each repository's own queue. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
State was only emitted at bucket boundaries — roughly an eighth of the fleet apart — so a pod dying mid-bucket re-read hours of finished repositories. Incremental streams now checkpoint every 25k records; any snapshot of the state is a valid resume point because a repository only enters it once it is finished, and the property hands the platform an isolated copy since workers commit concurrently with serialisation. The state guard also rejected any bucket_count but the current one and started empty — a future granularity change would have forced the full resync the connector promises to avoid. Keys are repository-scoped and buckets are derived by hash at read time, so the count is not part of the address and the state is accepted as-is. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
The default commit payload carries the message rendered to HTML, a summary rendering of it again, and a links map per commit — several times what the streams read, on the highest-volume endpoint in the connector. Commit ranges and the PR commit/comment listings now request only the fields the records are built from (verified against a live endpoint, including the POST range form). A misspelled entry is silently dropped by the API and surfaces as a NULL column, so each projection is pinned by a test. Activity stays unprojected: its payload is a union of event shapes. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Commit messages, PR descriptions and comment bodies are kept for display, not parsed, and generated descriptions routinely run to tens of KB. The pre-rewrite connector capped these at 1KB; the cap returns at 2KB. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Tolerating a refused continuation page ended the loop quietly, so the caller published part of a collection as a complete snapshot and the unread pages' rows were deleted by replacement. Tolerance now covers the first page only — where it means the feature is absent — and a refusal after the collection has started fails the repository loudly. It is deliberately not raised as an API error: that status would mark the repository inaccessible to every later stream over one bad page. Also folds the repeated start-date comparison into a named predicate, collapses finish_bucket onto a single last-bucket guard, and annotates the reachability range helper. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py (1)
68-103: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDrain
new_commits()exhaustively before inspectingunresolved.
new_commits()records missing includes after each retry attempt, but those records only run as the generator advances. If_diffstat()suppresses output without propagating the generator, a range repair can occur while_drain_ready()treats the queue as complete after the first_READ_DONE, so later retries never updateunresolved. Ensure the commit generator is fully consumed before_read_concurrently()accepts_READ_DONEfor that repository.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py` around lines 68 - 103, Ensure the repository read flow fully consumes the new_commits generator before inspecting unresolved or accepting _READ_DONE. Update the caller that processes new_commits, including _diffstat/_drain_ready or _read_concurrently as applicable, so all retry attempts execute and missing includes are recorded before the repository is marked complete.
♻️ Duplicate comments (1)
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py (1)
231-297: 📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy liftExtract the concurrent reader into its own module (still applies).
This is the same concern raised in a prior review round on this file. The concurrent reader (
_PendingRead,_read_concurrently,_drain_ready,_submit,_collect,_offer, plus the buffer/polling constants) is self-contained and does not depend onBitbucketStreamstate beyondrepository_recordsandhandle_repository_error. The file has grown past the point the prior comment already flagged, and now also carries the incremental-state locking logic (lines 488-565). Move the producer-consumer protocol to a dedicated module (e.g.streams/concurrent_reads.py) that takes a per-repository record producer and yields records;BitbucketStreamthen only picks between the serial and concurrent readers.As per coding guidelines: "Keep one concern per Python module; split modules that grow beyond approximately 400 lines."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py` around lines 231 - 297, The concurrent producer-consumer implementation is mixed into BitbucketStream alongside unrelated stream and incremental-state logic. Extract _PendingRead, _read_concurrently, _drain_ready, _submit, _collect, _offer, and their buffer/polling constants into a dedicated concurrent_reads module with an API that accepts per-repository record producers and yields records; update BitbucketStream to retain only the serial-versus-concurrent selection and pass repository_records and handle_repository_error through the new reader.Source: Coding guidelines
🧹 Nitpick comments (4)
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py (1)
78-99: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReduce nesting in the repair loop.
The retry loop nests to four indentation levels (
for→except→if missing.intersection(...)→if unresolved is not None/if not includes). Extract the missing-sha narrowing step into a named helper to keep the loop body at two levels.♻️ Proposed refactor
+ def _narrow_after_missing( + self, missing: Collection[str], includes: list[str], excludes: list[str], unresolved: set[str] | None + ) -> tuple[list[str], list[str]] | None: + if not (missing.intersection(includes) or missing.intersection(excludes)): + return None + if unresolved is not None: + unresolved.update(missing.intersection(includes)) + return ( + [sha for sha in includes if sha not in missing], + [sha for sha in excludes if sha not in missing], + ) + def new_commits( self, repo: RepositoryRef, current_heads: Sequence[str], previous_heads: Sequence[str], unresolved: set[str] | None = None, ) -> Iterable[Mapping[str, object]]: includes = list(current_heads) excludes = list(previous_heads) last_error = BitbucketApiError(404, "", "") for _ in range(RANGE_REPAIR_ATTEMPTS): try: yield from self._client.commits_between(repo, includes, excludes) return except BitbucketApiError as exc: if exc.status_code != 404: raise last_error = exc - # Retrying re-yields whatever the failed attempt already - # emitted; bronze collapses the overlap on unique_key. - missing = exc.missing_shas - if missing.intersection(includes) or missing.intersection(excludes): - if unresolved is not None: - unresolved.update(missing.intersection(includes)) - includes = [sha for sha in includes if sha not in missing] - excludes = [sha for sha in excludes if sha not in missing] - if not includes: - return - elif excludes: - excludes = [] - else: - raise + # Retrying re-yields already-emitted records; unique_key dedups. + narrowed = self._narrow_after_missing(exc.missing_shas, includes, excludes, unresolved) + if narrowed is not None: + includes, excludes = narrowed + if not includes: + return + elif excludes: + excludes = [] + else: + raiseAs per coding guidelines, "Prefer early returns over nested conditionals; aim to keep function bodies to at most two indentation levels."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py` around lines 78 - 99, Extract the missing-SHA narrowing logic from the exception branch in the range repair loop into a named helper, including unresolved updates, filtering includes/excludes, and the empty-includes outcome. Update the loop to call that helper and use early returns or a simple result check so the retry body remains at no more than two indentation levels; preserve the existing fallback behavior for cases without missing SHAs.Source: Coding guidelines
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py (3)
25-27: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMultiple new comments span more than one line.
Several new explanatory comments run 2-4 lines (for example Lines 25-27, 248-250, 285-287, 330-332, 340-343, 489-492, 497-498, 504-507, 516-518, 521-522, 529-530). The guideline for this path asks to keep comments to one line. The reasoning in these comments is good and worth keeping; consider tightening each to a single line, or confirm with the team whether this guideline allows multi-sentence comments when a one-line comment cannot carry the necessary context.
As per coding guidelines: "Add comments only when code cannot express the reason ... keep them to one line."
Also applies to: 489-535
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py` around lines 25 - 27, Condense the explanatory comments throughout the affected stream code, especially around the commit/PR/comment body handling and the range identified by the reviewer, so each comment occupies a single line while preserving its essential rationale. Update only the comments; do not alter the ingestion behavior or remove the context they provide.Source: Coding guidelines
337-355: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse typed exceptions per failure kind.
Both
handle_repository_error(Line 344) andfinish_bucket(Line 421) raise a bareRuntimeErrorfor two different failure kinds: a credential failure and an end-of-sync repository-failure summary. Callers cannot distinguish them without string matching.As per coding guidelines: "raise one typed exception per failure kind."
♻️ Proposed typed exceptions
+class BitbucketAuthenticationError(RuntimeError): + """Raised when Bitbucket rejects the configured credentials mid-sync.""" + + +class BitbucketSyncFailedError(RuntimeError): + """Raised when one or more repositories failed during the sync.""" + + class BitbucketStream(Stream, ABC):- raise RuntimeError( + raise BitbucketAuthenticationError( "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- raise RuntimeError( + raise BitbucketSyncFailedError( f"{self.name}: {len(self._failed_repositories)} repositories failed this sync: " + ", ".join(self._failed_repositories[:10]) )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py` around lines 337 - 355, Introduce distinct typed exception classes for credential failure and end-of-sync repository-failure summary, then update handle_repository_error’s HTTP 401 branch and finish_bucket’s summary failure to raise the appropriate type instead of bare RuntimeError. Preserve the existing messages and exception chaining so callers can distinguish failures by type without string matching.Source: Coding guidelines
360-361: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAvoid a bare
Anyparameter type.
before_start_dateis only called with nullabledatevalues from commit records, which are serialized as nullable strings and handled as strings here. Type the parameter asstr | Noneso the signature matches the actual input shape.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py` around lines 360 - 361, Update the before_start_date method parameter from Any to str | None, matching the nullable string values supplied by commit records; preserve the existing boolean logic and return behavior.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py`:
- Around line 68-103: Ensure the repository read flow fully consumes the
new_commits generator before inspecting unresolved or accepting _READ_DONE.
Update the caller that processes new_commits, including _diffstat/_drain_ready
or _read_concurrently as applicable, so all retry attempts execute and missing
includes are recorded before the repository is marked complete.
---
Duplicate comments:
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py`:
- Around line 231-297: The concurrent producer-consumer implementation is mixed
into BitbucketStream alongside unrelated stream and incremental-state logic.
Extract _PendingRead, _read_concurrently, _drain_ready, _submit, _collect,
_offer, and their buffer/polling constants into a dedicated concurrent_reads
module with an API that accepts per-repository record producers and yields
records; update BitbucketStream to retain only the serial-versus-concurrent
selection and pass repository_records and handle_repository_error through the
new reader.
---
Nitpick comments:
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py`:
- Around line 25-27: Condense the explanatory comments throughout the affected
stream code, especially around the commit/PR/comment body handling and the range
identified by the reviewer, so each comment occupies a single line while
preserving its essential rationale. Update only the comments; do not alter the
ingestion behavior or remove the context they provide.
- Around line 337-355: Introduce distinct typed exception classes for credential
failure and end-of-sync repository-failure summary, then update
handle_repository_error’s HTTP 401 branch and finish_bucket’s summary failure to
raise the appropriate type instead of bare RuntimeError. Preserve the existing
messages and exception chaining so callers can distinguish failures by type
without string matching.
- Around line 360-361: Update the before_start_date method parameter from Any to
str | None, matching the nullable string values supplied by commit records;
preserve the existing boolean logic and return behavior.
In
`@src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py`:
- Around line 78-99: Extract the missing-SHA narrowing logic from the exception
branch in the range repair loop into a named helper, including unresolved
updates, filtering includes/excludes, and the empty-includes outcome. Update the
loop to call that helper and use early returns or a simple result check so the
retry body remains at no more than two indentation levels; preserve the existing
fallback behavior for cases without missing SHAs.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 3d168737-af77-4e34-b670-1d0443e6e371
📒 Files selected for processing (21)
src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_comments.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_commits.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_base.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_client.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_pr_children.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_pull_requests.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_source.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py
🚧 Files skipped from review as they are similar to previous changes (8)
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py
- src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py
- src/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.py
- src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py
A worker committed state as soon as its generator finished, which is when the last record is buffered, not when it is emitted. Since a checkpoint can be taken between any two records the consumer emits, state could claim a repository whose records were still queued: a crash in that window lost them and the idle gate skipped the repository on the next sync. Workers now stage their commits and the consumer applies them only after draining that repository's queue, so an abandoned read publishes nothing at all. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
The empty-listing guard keyed off a branch count that state written before it existed does not carry, so the first empty answer after deploying went straight through and deleted the repository's branches. One empty answer is now never enough: it is recorded and the cursor held, and only a second consecutive empty listing publishes. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Bitbucket names only the unresolvable shas it noticed, so a repository with many dead heads needs many pruning rounds. The fixed cap counted those rounds against itself and gave up mid-repair, leaving the repository to fail the same way on every future sync. The bound now comes from the head count — each round drops at least one sha — and a run that needs an unusual number of rounds says so. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Wrapping every refused continuation page hid 401 from the sync-wide abort, which would have quarantined repositories one at a time with the cause buried. Only the denial statuses are wrapped now. pull_request_tasks also leaves the required-freshness group: its stream is unwired, so the 72h SLA would start erroring three days after deploy. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
On fresh state an empty branch listing counted as a complete read, so the cursor advanced with no heads and the idle gate then skipped the repository until somebody pushed to it — the same mistake the branch snapshot already guards against, left in the three range streams. An empty listing now counts only once the same answer has been recorded before, which state written before this rule never has. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Draining one queue until it ran dry let a producer that refills between yields hold the floor for its whole repository, leaving every other worker parked on a full buffer and blocking replacement submissions. Each repository now hands over after a bounded batch. Workers also give up parking if nothing has consumed for five minutes: an abandoned read otherwise leaves non-daemon pool threads waiting on a consumer that will never return, which keeps the process alive at exit. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
…nted The queue-full timeout could not tell an absent consumer from a slow destination, and when it fired the worker set the shared stop event — after which its own completion sentinel could no longer be enqueued, so the consumer polled the orphaned queue forever. A five-minute stall downstream was enough to trigger it, for every worker in the bucket. stop belongs to the consumer, which sets it in its finally once it is finished with the bucket; until then a full buffer only means the destination is behind. Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py (1)
320-347: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueCondense the multi-line inline comment to one line.
Lines 334-337 use a 4-line inline comment split into two multi-line remarks. The repository guideline requires inline comments to stay to one line each.
♻️ Proposed condensed comment
- # The consumer has to be the slower side, otherwise it empties each - # queue before the producer can refill and rotation is not being tested. - # Closed explicitly: an abandoned read leaves workers parked, and this - # test is the one that fails while holding the generator. + # The consumer must be slower than the producer, or rotation is not exercised. + # Closed explicitly, since an abandoned read here would leave workers parked.As per coding guidelines, "Add comments only when code cannot express the reason, such as intentional redundancy, cross-function invariants, or reasoned workarounds; keep them to one line."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py` around lines 320 - 347, Condense the explanatory comments inside test_one_fast_producer_does_not_hold_the_floor into one-line comments, preserving the rationale that a slow consumer is required to exercise rotation and that the stream must be explicitly closed to release abandoned worker reads.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml`:
- Around line 47-50: Remove pull_request_tasks from the published source
contract by deleting its freshness entry in schema.yml and its promotion in
bitbucket_cloud__bronze_promoted.sql; keep the source definition aligned with
source.py and do not retain the table unless it is restored as an active stream.
---
Nitpick comments:
In `@src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py`:
- Around line 320-347: Condense the explanatory comments inside
test_one_fast_producer_does_not_hold_the_floor into one-line comments,
preserving the rationale that a slow consumer is required to exercise rotation
and that the stream must be explicitly closed to release abandoned worker reads.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: ef3992b9-e123-4b36-8034-c0f6dae58b6a
📒 Files selected for processing (12)
src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.ymlsrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.pysrc/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_client.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.pysrc/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py
🚧 Files skipped from review as they are similar to previous changes (9)
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py
- src/ingestion/connectors/git/bitbucket-cloud/tests/test_client.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py
- src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py
- src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py
Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
3 participants
Permanent API answers failed whole streams on every run, and a full pass over a
large workspace spent its time queueing serial requests.
Reliability
Throughput
bitbucket_concurrency, default 4, bounded buffers)260 tests, verified on Python 3.10 and 3.14.
Summary by CodeRabbit