feat(declarative): add RESET_SYNC_TOKEN recovery for SyncTokenCursor (2/3)

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2878,14 +2878,21 @@ definitions:
         enum: [HttpResponseFilter]
       action:
         title: Action
-        description: Action to execute if a response matches the filter.
+        description: >-
+          Action to execute if a response matches the filter. `RESET_SYNC_TOKEN` works only on streams whose
+          `incremental_sync` is a `SyncTokenCursor`. It drops the partition's stored token and re-reads the partition
+          from page 1 without it, at most once per attempt. The destination is not cleared: append connections receive
+          the partition's records again, and records the API deleted or hid while the token was invalid stay in the
+          destination. When the API requires the client to clear its copy (for example, Google Calendar's HTTP 410),
+          use `FAIL` with an instruction to refresh the stream instead.
         type: string
         enum:
           - SUCCESS
           - FAIL
           - RETRY
           - IGNORE
           - RESET_PAGINATION
+          - RESET_SYNC_TOKEN
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
           - REDUCE_PAGE_SIZE
@@ -2896,6 +2903,7 @@ definitions:
           - RETRY
           - IGNORE
           - RESET_PAGINATION
+          - RESET_SYNC_TOKEN
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
           - REDUCE_PAGE_SIZE

airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py

@@ -35,6 +35,7 @@
 from airbyte_cdk.sources.streams.concurrent.cursor import Cursor, CursorField
 from airbyte_cdk.sources.streams.concurrent.partitions.partition import Partition
 from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState
+from airbyte_cdk.utils.analytics_message import create_analytics_message
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 logger = logging.getLogger("airbyte")
@@ -97,6 +98,7 @@ def __init__(
         self._warned_no_token = False
         self._warned_cap = False
         self._partitions_over_cap: set[str] = set()
+        self._reset_partitions: set[str] = set()
         self._invalid_state_reason: Optional[str] = None
         self._load_state(stream_state or {})
 
@@ -210,12 +212,17 @@ def on_page_completed(self, page: PageCompleted) -> None:
             )
         else:
             complete = True
-        if not complete:
-            self._warn_missing_token()
-            return
-
-        token = self._coerce_token(token_value, path)
+        token = self._coerce_token(token_value, path) if complete else None
         if token is None:
+            if partition_key in self._reset_partitions:
+                message = (
+                    f"Stream {self._stream_name}: the full re-read after a sync token reset ended "
+                    "without a new token, so the next sync would be rejected the same way. "
+                    "Check token_path and completion_condition."
+                )
+                raise AirbyteTracedException(
+                    message=message, internal_message=message, failure_type=FailureType.system_error
+                )
             self._warn_missing_token()
             return
 
@@ -226,6 +233,46 @@ def on_page_completed(self, page: PageCompleted) -> None:
                 .isoformat(timespec="milliseconds")
                 .replace("+00:00", "Z"),
             }
+            if partition_key in self._reset_partitions:
+                self._pending[partition_key]["after_reset"] = True
+
+    def restart_without_token(
+        self, stream_slice: StreamSlice, reason: Optional[str] = None
+    ) -> StreamSlice:
+        if "token" not in stream_slice.cursor_slice:
+            message = (
+                f"Stream {self._stream_name}: the API rejected a read without a sync token. "
+                "Check the RESET_SYNC_TOKEN response filter."
+            )
+            raise AirbyteTracedException(
+                internal_message=message,
+                message=message,
+                failure_type=FailureType.system_error,
+            )
+
+        partition_key = self._serializer.to_partition_key(stream_slice.partition or {})
+        with self._lock:
+            if (self._committed.get(partition_key) or {}).get("after_reset"):
+                message = (
+                    f"Stream {self._stream_name}: the API rejected the sync token that the previous "
+                    f"sync committed after a full re-read ({reason or 'no reason given'}). Two "
+                    "rejections in a row usually mean token_path, token_option or the "
+                    "RESET_SYNC_TOKEN predicate is wrong. Refresh the stream after fixing it."
+                )
+                raise AirbyteTracedException(
+                    message=message, internal_message=message, failure_type=FailureType.system_error
+                )
+            self._reset_partitions.add(partition_key)
+
+        self._message_repository.emit_message(
+            create_analytics_message("sync-token-reset", self._stream_name)
+        )
+
+        return StreamSlice(
+            partition=stream_slice.partition,
+            cursor_slice={},
+            extra_fields=stream_slice.extra_fields,
+        )
 
     def _coerce_token(self, value: Any, path: list[str]) -> Optional[str]:
         if value is None or value == "":
@@ -397,8 +444,16 @@ def _load_state(self, stream_state: StreamState) -> None:
             return
         if self._partitioned:
             states = stream_state.get("states")
-            if not isinstance(states, list) or "token" in stream_state:
-                if "token" not in stream_state and "states" not in stream_state:
+            if (
+                not isinstance(states, list)
+                or "token" in stream_state
+                or "after_reset" in stream_state
+            ):
+                if (
+                    "token" not in stream_state
+                    and "states" not in stream_state
+                    and "after_reset" not in stream_state
+                ):
                     self._log_bootstrap("none")
                 else:
                     self._invalid_state_reason = "unpartitioned shape on a partitioned stream"
@@ -410,29 +465,45 @@ def _load_state(self, stream_state: StreamState) -> None:
                     return
                 partition = state.get("partition")
                 cursor = state.get("cursor")
+                after_reset = cursor.get("after_reset") if isinstance(cursor, Mapping) else None
                 if (
                     isinstance(partition, Mapping)
                     and isinstance(cursor, Mapping)
                     and isinstance(cursor.get("token"), str)
                     and cursor.get("token")
+                    and ("after_reset" not in cursor or isinstance(after_reset, bool))
                 ):
-                    committed[self._serializer.to_partition_key(partition)] = {
+                    committed_state = {
                         "token": cursor["token"],
                         "committed_at": cursor.get("committed_at", ""),
                     }
+                    if "after_reset" in cursor:
+                        committed_state["after_reset"] = after_reset
+                    committed[self._serializer.to_partition_key(partition)] = committed_state
                 else:
-                    self._invalid_state_reason = "a partition entry has no valid token"
+                    self._invalid_state_reason = (
+                        "a partition entry has an invalid token or after_reset"
+                    )
                     return
             self._committed.update(committed)
         else:
             if "states" in stream_state:
                 self._invalid_state_reason = "partitioned shape on an unpartitioned stream"
                 return
             token = stream_state.get("token")
+            after_reset = stream_state.get("after_reset")
+            if "after_reset" in stream_state and not isinstance(after_reset, bool):
+                self._invalid_state_reason = "after_reset is not a boolean"
+                return
             if isinstance(token, str) and token:
-                self._committed["{}"] = {
+                committed_state = {
                     "token": token,
                     "committed_at": stream_state.get("committed_at", ""),
                 }
+                if "after_reset" in stream_state:
+                    committed_state["after_reset"] = after_reset
+                self._committed["{}"] = committed_state
             elif "token" in stream_state:
                 self._invalid_state_reason = "the token is not a non-empty string"
+            elif "after_reset" in stream_state:
+                self._invalid_state_reason = "after_reset is present without a token"

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -614,6 +614,7 @@ class Action(Enum):
     RETRY = "RETRY"
     IGNORE = "IGNORE"
     RESET_PAGINATION = "RESET_PAGINATION"
+    RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN"
     RATE_LIMITED = "RATE_LIMITED"
     REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY"
     REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE"
@@ -630,13 +631,21 @@ class HttpResponseFilter(BaseModel):
     type: Literal["HttpResponseFilter"]
     action: Optional[Action] = Field(
         None,
-        description="Action to execute if a response matches the filter.",
+        description=(
+            "Action to execute if a response matches the filter. `RESET_SYNC_TOKEN` works only on streams whose "
+            "`incremental_sync` is a `SyncTokenCursor`. It drops the partition's stored token and re-reads the "
+            "partition from page 1 without it, at most once per attempt. The destination is not cleared: append "
+            "connections receive the partition's records again, and records the API deleted or hid while the token "
+            "was invalid stay in the destination. When the API requires the client to clear its copy (for example, "
+            "Google Calendar's HTTP 410), use `FAIL` with an instruction to refresh the stream instead."
+        ),
         examples=[
             "SUCCESS",
             "FAIL",
             "RETRY",
             "IGNORE",
             "RESET_PAGINATION",
+            "RESET_SYNC_TOKEN",
             "RATE_LIMITED",
             "REFRESH_TOKEN_THEN_RETRY",
             "REDUCE_PAGE_SIZE",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1257,6 +1257,9 @@ def create_session_token_authenticator(
         self._reject_reduce_page_size_action(
             model.login_requester, f"`login_requester` of the SessionTokenAuthenticator of {name}"
         )
+        self._reject_reset_sync_token_action(
+            model.login_requester, f"`login_requester` of the SessionTokenAuthenticator of {name}"
+        )
         self._reject_split_request_window_action(
             model.login_requester, f"`login_requester` of the SessionTokenAuthenticator of {name}"
         )
@@ -2137,6 +2140,13 @@ def create_default_stream(
         self, model: DeclarativeStreamModel, config: Config, is_parent: bool = False, **kwargs: Any
     ) -> AbstractStream:
         primary_key = model.primary_key.__root__ if model.primary_key else None
+        requester = getattr(model.retriever, "requester", None)
+        if self._uses_reset_sync_token_action(
+            getattr(requester, "error_handler", None)
+        ) and not isinstance(model.incremental_sync, SyncTokenCursorModel):
+            raise ValueError(
+                f"Stream {model.name}: RESET_SYNC_TOKEN requires a SyncTokenCursor incremental_sync."
+            )
         self._validate_sync_token_stream(model)
         self._migrate_state(model, config)
         self._warn_on_ineffective_incremental_dependency(model)
@@ -4235,6 +4245,9 @@ def _get_url(req: Requester) -> str:
             page_observer=(
                 cursor.on_page_completed if isinstance(cursor, SyncTokenCursor) else None
             ),
+            token_reset_handler=(
+                cursor.restart_without_token if isinstance(cursor, SyncTokenCursor) else None
+            ),
             parameters=model.parameters or {},
         )
 
@@ -4539,6 +4552,29 @@ def _uses_reduce_page_size_action(self, error_handler: Any) -> bool:
         # A CustomErrorHandler can return any action and we cannot inspect it, so we do not validate it.
         return False
 
+    def _reject_reset_sync_token_action(self, requester: Any, description: str) -> None:
+        if requester is None:
+            return
+        if self._uses_reset_sync_token_action(getattr(requester, "error_handler", None)):
+            raise ValueError(
+                f"The RESET_SYNC_TOKEN response action is not supported on the {description}: only a "
+                "DeclarativeStream configured with a SyncTokenCursor can re-read a partition without "
+                "its stored sync token."
+            )
+
+    def _uses_reset_sync_token_action(self, error_handler: Any) -> bool:
+        if isinstance(error_handler, CompositeErrorHandlerModel):
+            return any(
+                self._uses_reset_sync_token_action(nested)
+                for nested in error_handler.error_handlers
+            )
+        if isinstance(error_handler, DefaultErrorHandlerModel):
+            return any(
+                response_filter.action == HttpResponseFilterActionModel.RESET_SYNC_TOKEN
+                for response_filter in error_handler.response_filters or []
+            )
+        return False
+
     def _create_request_window_splitting(
         self,
         model: SimpleRetrieverModel,
@@ -5026,6 +5062,10 @@ def create_async_retriever(
                 getattr(model, requester_field, None),
                 f"`{requester_field}` of the AsyncRetriever of stream {name}",
             )
+            self._reject_reset_sync_token_action(
+                getattr(model, requester_field, None),
+                f"`{requester_field}` of the AsyncRetriever of stream {name}",
+            )
             self._reject_split_request_window_action(
                 getattr(model, requester_field, None),
                 f"`{requester_field}` of the AsyncRetriever of stream {name}",
@@ -5673,6 +5713,7 @@ def create_file_uploader(
         self, model: FileUploaderModel, config: Config, **kwargs: Any
     ) -> FileUploader:
         self._reject_reduce_page_size_action(model.requester, "requester of a `file_uploader`")
+        self._reject_reset_sync_token_action(model.requester, "requester of a `file_uploader`")
         self._reject_split_request_window_action(model.requester, "requester of a `file_uploader`")
         name = "File Uploader"
         requester = self._create_component_from_model(

airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py

@@ -71,6 +71,7 @@ def interpret_response(
                 ResponseAction.RETRY,
                 ResponseAction.IGNORE,
                 ResponseAction.RESET_PAGINATION,
+                ResponseAction.RESET_SYNC_TOKEN,
                 ResponseAction.REDUCE_PAGE_SIZE,
                 ResponseAction.SPLIT_REQUEST_WINDOW,
             ]:

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -68,6 +68,10 @@
     RequestWindowSplitNotSupportedException,
     RequestWindowSplitRequiredException,
 )
+from airbyte_cdk.sources.streams.http.sync_token_reset_exception import (
+    SyncTokenResetNotSupportedException,
+    SyncTokenResetRequiredException,
+)
 from airbyte_cdk.sources.types import Config, Record, StreamSlice
 from airbyte_cdk.utils.mapping_helpers import combine_mappings
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
@@ -158,6 +162,7 @@ class SimpleRetriever(Retriever):
         Callable[[StreamSlice, Optional[datetime.timedelta]], Optional[List[StreamSlice]]]
     ] = None
     page_observer: Optional[Callable[["PageCompleted"], None]] = None
+    token_reset_handler: Optional[Callable[[StreamSlice, Optional[str]], StreamSlice]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._paginator = self.paginator or NoPagination(parameters=parameters)
@@ -707,6 +712,25 @@ def _read_records_or_split_request_window(
                 yield from self._read_records_or_split_request_window(
                     records_schema, child, original_slice, depth + 1
                 )
+        except SyncTokenResetRequiredException as exception:
+            if self.token_reset_handler is None:
+                raise SyncTokenResetNotSupportedException(stream_name=self.name) from exception
+
+            reset_slice = self.token_reset_handler(stream_slice, exception.error_message)
+            reason = exception.error_message or "the API rejected the stored sync token"
+            if stream_slice.partition:
+                LOGGER.warning(
+                    f"Stream {self.name}: {reason}; re-reading partition {stream_slice.partition} in full "
+                    "without a sync token. All of its records will be emitted again."
+                )
+            else:
+                LOGGER.warning(
+                    f"Stream {self.name}: {reason}; re-reading the stream in full "
+                    "without a sync token. All of its records will be emitted again."
+                )
+            yield from self._read_records_or_split_request_window(
+                records_schema, reset_slice, original_slice, depth + 1
+            )
 
     def _with_request_window_failure_message(self, message: str) -> str:
         """

airbyte_cdk/sources/streams/http/error_handlers/response_models.py

@@ -17,6 +17,7 @@ class ResponseAction(Enum):
     FAIL = "FAIL"
     IGNORE = "IGNORE"
     RESET_PAGINATION = "RESET_PAGINATION"
+    RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN"
     RATE_LIMITED = "RATE_LIMITED"
     REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY"
     REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE"

airbyte_cdk/sources/streams/http/http_client.py

@@ -65,6 +65,9 @@
     ResponseAwareAuthenticator,
     TokenRotatingAuthenticator,
 )
+from airbyte_cdk.sources.streams.http.sync_token_reset_exception import (
+    SyncTokenResetRequiredException,
+)
 from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter
 from airbyte_cdk.sources.utils.types import JsonType
 from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
@@ -620,6 +623,9 @@ def _handle_error_resolution(
         if error_resolution.response_action == ResponseAction.RESET_PAGINATION:
             raise PaginationResetRequiredException()
 
+        if error_resolution.response_action == ResponseAction.RESET_SYNC_TOKEN:
+            raise SyncTokenResetRequiredException(error_resolution.error_message)
+
         if error_resolution.response_action == ResponseAction.REDUCE_PAGE_SIZE:
             raise PageSizeReductionRequiredException(
                 stream_name=self._name, error_message=error_resolution.error_message