feat: Attempt to sync all streams instead of crashing on the first error

[edgarrmondragon]

Summary by Sourcery

Enhancements:

• Add a dedicated state manager instance to each stream to centralize and manage its sync state.

[sourcery-ai]

Reviewer's guide (collapsed on small PRs)

Reviewer's Guide

Implements per-stream sync outcome tracking so taps continue syncing remaining streams after individual stream failures, aggregates results into a process exit code, and tightens state advancement and logging behavior around stream sync errors.

Sequence diagram for Tap.sync_all aggregating per-stream SyncResult

sequenceDiagram
    actor CLI
    participant Tap
    participant Stream1 as Stream_success
    participant Stream2 as Stream_failure
    participant Stream3 as Stream_skipped
    participant StateMgr1 as StreamStateManager_success
    participant StateMgr2 as StreamStateManager_failure
    participant StateMgr3 as StreamStateManager_skipped

    CLI->>Tap: sync_all()
    Tap->>Stream1: sync()
    Stream1->>StateMgr1: load_state()
    StateMgr1-->>Stream1: state_loaded
    Stream1->>StateMgr1: advance_bookmark(partition_state)
    Stream1->>StateMgr1: finalize_state()
    Stream1-->>Tap: SyncResult.SUCCESS

    Tap->>Tap: combined_result = SUCCESS.combine(SUCCESS)

    Tap->>Stream2: sync()
    Stream2->>StateMgr2: load_state()
    StateMgr2-->>Stream2: state_loaded
    Stream2->>Stream2: unexpected_exception
    Stream2-->>Tap: SyncResult.FAILURE

    Tap->>Tap: combined_result = combined_result.combine(FAILURE)

    Tap->>Stream3: sync()
    Stream3->>StateMgr3: load_state()
    StateMgr3-->>Stream3: state_loaded
    Stream3-->>Tap: SyncResult.SKIPPED

    Tap->>Tap: combined_result = combined_result.combine(SKIPPED)

    Tap->>Tap: log per-stream outcomes
    Tap-->>CLI: exit_code = combined_result.to_exit_code()

Loading

Class diagram for per-stream SyncResult tracking and state management

classDiagram
    class SyncResult {
        <<enum>>
        SUCCESS
        FAILURE
        SKIPPED
        ABORTED
        +SyncResult combine(SyncResult other)
        +int to_exit_code()
        +bool is_success()
        +bool is_failure()
        +bool is_aborted()
        +bool is_skipped()
        +str log_level()
    }

    class StreamStateManager {
        +str tap_name
        +str stream_name
        +dict tap_state
        +list state_partitioning_keys
        +advance_bookmark(dict partition_state)
        +finalize_state()
        +load_state()
    }

    class Stream {
        +str name
        +Tap tap
        +dict _tap_state
        +list _state_partitioning_keys
        +list child_streams
        +SyncResult sync_result
        +StreamStateManager _state_manager
        +SyncResult sync()
        +sync_child_streams()
        +sync_incremental()
        +sync_full_table()
    }

    class Tap {
        +str name
        +list streams
        +dict state
        +SyncResult sync_all()
        +int get_exit_code(SyncResult result)
    }

    Tap "1" o-- "many" Stream : owns
    Stream "1" o-- "many" Stream : child_streams
    Stream "1" *-- "1" StreamStateManager : uses_state_manager
    Stream ..> SyncResult : returns
    Tap ..> SyncResult : aggregates
    Tap ..> StreamStateManager : finalizes_successful_state

Loading

File-Level Changes

Change	Details	Files
Add SyncResult enum and helpers to represent and combine per-stream sync outcomes and derive exit codes and logging behavior.	Define SyncResult variants to represent different sync outcomes (success, failure, abort, etc.). Implement combination logic to fold multiple per-stream results into an overall result used for process exit codes. Add logging helpers to emit standardized per-stream sync outcome messages with level driven by SyncResult. Provide utility methods/functions to map SyncResult values to appropriate exit codes.	singer_sdk/sync_result.py singer_sdk/__init__.py singer_sdk/_typing.py tests/unit/test_sync_result.py
Change Stream.sync to return and record a SyncResult while standardizing exception handling and integrating a per-stream state manager.	Update Stream.sync signature to return SyncResult and set self.sync_result accordingly. Propagate lifecycle-related abort exceptions while converting unexpected exceptions into controlled abort SyncResult values. Initialize and use a StreamStateManager instance on Stream construction instead of ad-hoc state handling. Refine error logging in Stream.sync to include exception messages and standardized lifecycle handling.	singer_sdk/streams/core.py singer_sdk/state.py tests/unit/test_stream_sync.py
Update Tap.sync_all to aggregate per-stream SyncResult values, finalize state only for successful streams, and return a combined result used for process exit codes.	Iterate over all streams, collecting each stream’s SyncResult instead of failing fast on the first error. Aggregate individual SyncResult values into an overall result that determines the tap process exit code. Finalize and emit state only for streams whose SyncResult indicates successful completion, skipping failed streams to avoid corrupting bookmarks. Log a single outcome line per stream after sync_all completes, using log level derived from SyncResult.	singer_sdk/tap_base.py singer_sdk/cli.py tests/snapshots/test_tap_sync_all_snapshots.py tests/test_tap_sync_all.py
Add tests and a test tap to verify continued syncing behavior with failing, incremental, and parent-child streams plus logging expectations.	Introduce a dedicated test tap implementing multiple streams, including incremental and parent-child relationships, that can be configured to fail. Add snapshot tests to ensure taps continue syncing remaining streams when some streams raise errors. Add targeted unit tests to validate SyncResult combination logic, logging behavior, and state advancement conditions. Update existing tests to expect non-fatal behavior when individual streams fail and to assert correct exit codes.	tests/taps/test_tap_multi_stream.py tests/snapshots/test_tap_multi_stream/* tests/unit/test_sync_result.py tests/unit/test_tap_logging.py
Tighten state advancement rules for incremental streams and update dependency selector for Python 3.13 compatibility.	Ensure state bookmarks are only advanced for incremental streams that report a successful SyncResult, preventing advancement on failure. Adjust the typing-extensions dependency selector to use python_full_version for Python 3.13 compatibility. Add or update tests to confirm that state is not advanced for streams that fail during sync.	singer_sdk/state.py pyproject.toml tests/unit/test_state_incremental.py

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[read-the-docs-community]

Documentation build overview

📚 Meltano SDK | 🛠️ Build #32409590 | 📁 Comparing 3406325 against latest (ec2cb65)

  🔍 Preview build  

17 files changed · ± 15 modified · - 2 deleted

± Modified

• genindex.html
• incremental_replication.html
• reference.html
• classes/singer_sdk.GraphQLStream.html
• classes/singer_sdk.RESTStream.html
• classes/singer_sdk.Stream.html
• classes/singer_sdk.Tap.html
• classes/singer_sdk.pagination.BaseHATEOASPaginator.html
• classes/singer_sdk.sql.SQLTap.html
• guides/pagination-classes.html
• and 5 more...

- Deleted

• classes/singer_sdk.streams.rest.HTTPRequest.html
• classes/singer_sdk.streams.rest.PageContext.html

[codspeed]

Merging this PR will not alter performance

✅ 8 untouched benchmarks

---

_{Comparing feature/continue-errors (3406325) with feat/safely-ignore-errors (f419d3b)}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 93.78%. Comparing base (f419d3b) to head (3406325).

Additional details and impacted files

@@                      Coverage Diff                      @@
##           feat/safely-ignore-errors    #3614      +/-   ##
=============================================================
- Coverage                      93.86%   93.78%   -0.09%     
=============================================================
  Files                             74       74              
  Lines                           5965     5966       +1     
  Branches                         735      735              
=============================================================
- Hits                            5599     5595       -4     
- Misses                           274      278       +4     
- Partials                          92       93       +1     

Flag	Coverage Δ
core	82.34% <100.00%> (-0.24%)	⬇️
end-to-end	75.19% <100.00%> (-0.10%)	⬇️
optional-components	42.62% <0.00%> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[sourcery-ai]

Hey - I've left some high level feedback:

• Passing self._tap_state into StreamStateManager at construction time may lead to divergence if the tap state is later mutated elsewhere; consider passing a reference or accessor so the manager always sees the latest state.
• If StreamStateManager has any expensive initialization or external dependencies, consider lazy initialization or injecting a shared manager to avoid overhead when many streams are created concurrently.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- Passing `self._tap_state` into `StreamStateManager` at construction time may lead to divergence if the tap state is later mutated elsewhere; consider passing a reference or accessor so the manager always sees the latest state.
- If `StreamStateManager` has any expensive initialization or external dependencies, consider lazy initialization or injecting a shared manager to avoid overhead when many streams are created concurrently.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}