diff --git a/airbyte-integrations/connectors/source-stripe/AGENTS.md b/airbyte-integrations/connectors/source-stripe/AGENTS.md index 6d38af26c6c9..e359a80c0889 100644 --- a/airbyte-integrations/connectors/source-stripe/AGENTS.md +++ b/airbyte-integrations/connectors/source-stripe/AGENTS.md @@ -10,6 +10,8 @@ Stripe's Events API only retains events for 30 days. If the connector's state fa **Why this matters:** What looks like a simple entity read is actually two completely different data paths depending on whether state exists and how old it is. Adding a new entity stream requires defining both the direct-read retriever AND the events-based retriever with the correct event type filter strings. If the event type strings are wrong, incremental syncs will silently miss updates. +Creation events use a cursor one second earlier than their event timestamp so update events win same-second ties and the newer payload is retained. + ## 2. Silent 403/400/404 Error Ignoring The base error handler is configured to IGNORE (not fail) responses with HTTP status 403 (permission denied), 400 (bad request), and 404 (not found). When the Stripe API returns any of these errors for a specific resource or subresource, the connector silently skips that record and continues syncing. diff --git a/airbyte-integrations/connectors/source-stripe/manifest.yaml b/airbyte-integrations/connectors/source-stripe/manifest.yaml index 8c0b9819ad89..d72824c7730f 100644 --- a/airbyte-integrations/connectors/source-stripe/manifest.yaml +++ b/airbyte-integrations/connectors/source-stripe/manifest.yaml @@ -219,7 +219,7 @@ definitions: - data - object - updated - value: "{{ record.get('updated', record.get('created', now_utc().timestamp())) | int }}" + value: "{{ (record.get('updated', record.get('created', now_utc().timestamp())) | int) - (1 if record.get('type', '').endswith('.created') else 0) }}" value_type: integer condition: "{{ record.get('data', {}).get('object', False) }}" - type: DpathFlattenFields @@ -1503,7 +1503,7 @@ definitions: fields: - path: - invoice_updated - value: "{{ (record['original_record'].get('updated') or record['original_record'].get('created') or record['original_record'].get('data', {}).get('object', {}).get('updated') or record['original_record'].get('data', {}).get('object', {}).get('created') or now_utc().timestamp()) | int }}" + value: "{{ ((record['original_record'].get('updated') or record['original_record'].get('created') or record['original_record'].get('data', {}).get('object', {}).get('updated') or record['original_record'].get('data', {}).get('object', {}).get('created') or now_utc().timestamp()) | int) - (1 if record['original_record'].get('type', '').endswith('.created') else 0) }}" value_type: integer - path: - invoice_id @@ -1671,7 +1671,7 @@ definitions: fields: - path: - subscription_updated - value: "{{ (record['original_record'].get('updated') or record['original_record'].get('created') or record['original_record'].get('data', {}).get('object', {}).get('updated') or record['original_record'].get('data', {}).get('object', {}).get('created') or now_utc().timestamp()) | int }}" + value: "{{ ((record['original_record'].get('updated') or record['original_record'].get('created') or record['original_record'].get('data', {}).get('object', {}).get('updated') or record['original_record'].get('data', {}).get('object', {}).get('created') or now_utc().timestamp()) | int) - (1 if record['original_record'].get('type', '').endswith('.created') else 0) }}" value_type: integer condition: "{{ record.get('original_record') }}" - type: RemoveFields diff --git a/airbyte-integrations/connectors/source-stripe/metadata.yaml b/airbyte-integrations/connectors/source-stripe/metadata.yaml index 2fe1f5cee75c..66e55de4b4bb 100644 --- a/airbyte-integrations/connectors/source-stripe/metadata.yaml +++ b/airbyte-integrations/connectors/source-stripe/metadata.yaml @@ -10,7 +10,7 @@ data: connectorSubtype: api connectorType: source definitionId: e094cb9a-26de-4645-8761-65c0c425d1de - dockerImageTag: 6.0.13 + dockerImageTag: 6.0.14 dockerRepository: airbyte/source-stripe documentationUrl: https://docs.airbyte.com/integrations/sources/stripe erdUrl: https://dbdocs.io/airbyteio/source-stripe?view=relationships diff --git a/airbyte-integrations/connectors/source-stripe/unit_tests/integration/test_external_account_cards.py b/airbyte-integrations/connectors/source-stripe/unit_tests/integration/test_external_account_cards.py index 69ec65c58519..a1ded4b4f14a 100644 --- a/airbyte-integrations/connectors/source-stripe/unit_tests/integration/test_external_account_cards.py +++ b/airbyte-integrations/connectors/source-stripe/unit_tests/integration/test_external_account_cards.py @@ -205,6 +205,70 @@ def _read(self, config: ConfigBuilder, expecting_exception: bool = False) -> Ent @freezegun.freeze_time(_NOW.isoformat()) class IncrementalTest(TestCase): + @HttpMocker() + def test_given_created_and_updated_events_in_same_second_when_read_then_updated_event_wins_cursor_tie( + self, http_mocker: HttpMocker + ) -> None: + state_datetime = _NOW - timedelta(days=5) + event_cursor = int(state_datetime.timestamp()) + 1 + object_record = _an_external_account_card().with_id("card_same_second") + + http_mocker.get( + _events_request().with_created_gte(state_datetime).with_created_lte(_NOW).with_limit(100).with_types(_EVENT_TYPES).build(), + _events_response() + .with_record( + _an_event() + .with_id("evt_updated") + .with_cursor(event_cursor) + .with_field(FieldPath("type"), "account.external_account.updated") + .with_field(_DATA_FIELD, object_record.build()) + ) + .with_record( + _an_event() + .with_id("evt_created") + .with_cursor(event_cursor) + .with_field(FieldPath("type"), "account.external_account.created") + .with_field(_DATA_FIELD, object_record.build()) + ) + .build(), + ) + + output = self._read( + _config(), + StateBuilder().with_stream_state(_STREAM_NAME, {"updated": int(state_datetime.timestamp())}).build(), + ) + + assert output.records[0].record.data["updated"] == event_cursor + assert output.records[1].record.data["updated"] == event_cursor - 1 + + @HttpMocker() + def test_given_created_event_at_state_boundary_when_read_then_emit_record_without_regressing_state( + self, http_mocker: HttpMocker + ) -> None: + state_value = int((_NOW - timedelta(days=5)).timestamp()) + state_datetime = datetime.fromtimestamp(state_value, timezone.utc) + + http_mocker.get( + _events_request().with_created_gte(state_datetime).with_created_lte(_NOW).with_limit(100).with_types(_EVENT_TYPES).build(), + _events_response() + .with_record( + _an_event() + .with_cursor(state_value) + .with_field(FieldPath("type"), "account.external_account.created") + .with_field(_DATA_FIELD, _an_external_account_card().build()) + ) + .build(), + ) + + output = self._read( + _config(), + StateBuilder().with_stream_state(_STREAM_NAME, {"updated": state_value}).build(), + ) + + assert len(output.records) == 1 + assert output.records[0].record.data["updated"] == state_value - 1 + assert int(output.most_recent_state.stream_state.updated) >= state_value + @HttpMocker() def test_given_no_state_when_read_then_use_external_accounts_endpoint(self, http_mocker: HttpMocker) -> None: http_mocker.get( diff --git a/docs/integrations/sources/stripe.md b/docs/integrations/sources/stripe.md index f3a9a6a916b7..644746421e30 100644 --- a/docs/integrations/sources/stripe.md +++ b/docs/integrations/sources/stripe.md @@ -317,6 +317,7 @@ If you use Airbyte Cloud and your organization restricts access to specific IPs, | Version | Date | Pull Request | Subject | |:------------|:-----------|:-------------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| 6.0.14 | 2026-08-13 | [84355](https://github.com/airbytehq/airbyte/pull/84355) | Update events now win same-second cursor ties with creation events so the newer payload is kept at the destination. | | 6.0.13 | 2026-08-11 | [84134](https://github.com/airbytehq/airbyte/pull/84134) | Update dependencies | | 6.0.12 | 2026-08-04 | [83634](https://github.com/airbytehq/airbyte/pull/83634) | Update dependencies | | 6.0.11 | 2026-07-28 | [83119](https://github.com/airbytehq/airbyte/pull/83119) | Update dependencies |