Repository: meltano/sdk
Pull request #1436 (2023-02-21 08:38:03)

fix: handle sync abort, reduce duplicate `STATE` messages, rename `_MAX_RECORD_LIMIT` as `ABORT_AT_RECORD_COUNT`
This is proposed to target PR #1399, either to merge into that PR or replace #1399 and merge to `main`.

Importantly, this adds handling for how "abort" exceptions should be handled. There are two scenarios: either the stream successfully pauses the sync operation or else it raises a failure exception if pausing is not possible.

Notes:

- The private member `Stream._MAX_RECORD_LIMIT` is renamed to `Stream.ABORT_AT_RECORD_COUNT`. This rename properly documents the original and valid purpose for this limit - specifically this is a record count after which to trigger abort of the stream sync.
- Tests which _actually_ want a `sync_all()` behavior can call `sync_all()` directly, as this change reverts the behavior so that `sync_all()` will once again sync all records. This is the only way to ensure tests will get valid and finalized `state` messages. Otherwise, the aborted streams will have bookmarks left in an unfinalized, non-resumeable state.
- For test scenarios that should silence abort exceptions, `run_connection_test()` has been refactored to call a new and more generic `run_sync_dry_run()`, which can accept an arbitrary max record abort threshold. Both `run_connection_test()` and `run_sync_dry_run()` suppress the abort exceptions regardless of resumeability. In contrast, `sync_all()` will trigger the same exceptions if `ABORT_AT_RECORD_COUNT` is set and breached, but it will not suppress the exceptions as the two test-focused methods would do
- When aborting stream sync due to a user or s, we raise an exception that dispositions the caller according to whether the operation is still Singer compliant, namely that we will sync all records "at least once" without data loss.
- If we still have a valid stream sync after abort, `AbortedSyncPausedException` specifies that we are _mostly_ meeting Singer Spec, with the one exception that we left records on the source server.
  - We are Singer compliant _if_ the user runs again, and repeats until all records are drained.
- If we do not have a valid stream sync after abort, due to non-resumable streams setting up an infinite loop on subsequent restarts, then `AbortSyncFailedException` will be raised.

Other fixes:

1. A prior PR added a `SQLStream` implementation with a SQL `limit()` constraint. This has been updated to be `max limit + 1` in order for the SDK to know whether or not there are more records not synced, so we can properly inform the caller whether the sync operation was prematurely canceled or not. (If `ABORT_AT_RECORD_COUNT=100` and we have exactly 100 records in the source, then the sync operation can wrap with success. But if there are >=101 records in the source, we will report that the operation was aborted prematurely before syncing all records.
1. The properties `TapTestRunner.tap` and `TargetTestRunner.target` were creating a new tap or target upon each call to the property, which violates the expected behavior of a property returning a pointer to the same object on each subsequent call. I've renamed these to be regular methods, `new_tap()` and `new_target()`, since that reflects the behavior as implemented.
   - Alternatively, we could cache the output of `.tap` and `.target` properties, but this seemed likely to leak state across multiple tests that otherwise would be atomic. Hence, I think `new_tap()` and `new_target()` as methods probably best describe the desired behavior here.
1. Due to misapplied modulo math on zero-based record index, we previously were sending a `STATE` message after every first `RECORD` message. This PR fixes the modulo math, which breaks pre-recorded tests that expected a `STATE` message after every first `RECORD` message.
1. Previously, some cases would duplicate-send a final `STATE` message. This PR adds an internal private member `Stream._needs_state_flush` which is used to track whether one or more `RECORD` message has been written since the last `STATE` message being sent. The final duplicate `STATE` message would no longer to be sent after this PR.

Clarifications:

- There was previously some confusion regarding whether or not `_MAX_RECORD_LIMIT` was applied to certain subclasses of `Stream`. Given that the base class implementation of `Stream._sync_records()` was applying the limit (and still does here in this PR), this should fully cover `RESTStream`, `SQLStream`, and all other subclasses of `Stream`.


Future work:

- #1409 - would allow us to return a specific non-zero exit code to inform the caller that abort thresholds were met, and that one of these are true, either:
  - (a) All streams successfully paused their sync operations before shutting down, and further sync invocations should be able to eventually bring in all data which is valid, OR
  - (b) One or more streams failed to pause their sync operations in a valid state. Rerunning/retrying is not advised, since it will only lead to an infinite loop, where data loss is occurring for one or more streams, due to inability to reach a valid bookmark state.


codecov[bot]: ## [Codecov](https://codecov.io/gh/meltano/sdk/pull/1436?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
> :exclamation: No coverage uploaded for pull request base (`kgpayne/apply-max-records-limit-during-testing@5655ba8`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#section-missing-base-commit).
> The diff coverage is `n/a`.

```diff
@@                                Coverage Diff                                @@
##             kgpayne/apply-max-records-limit-during-testing    #1436   +/-   ##
=================================================================================
  Coverage                                                  ?   85.34%           
=================================================================================
  Files                                                     ?       57           
  Lines                                                     ?     4797           
  Branches                                                  ?      813           
=================================================================================
  Hits                                                      ?     4094           
  Misses                                                    ?      508           
  Partials                                                  ?      195           
```



:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano)

kgpayne: @aaronsteers this is epic 🙌

I have:
-  Fixed the failing test by moving the `_MAX_RECORD_AGE_IN_MINUTES` check onto the `_process_record_message` and `_process_batch_message` methods. Previously this check was under `_process_state` and so was triggering in the target as a side-effect of the redundant `STATE` message sent by the tap (which you fixed). Now we trigger the check on actual records arriving to the target, which I think is the expected behaviour?
- Updated the `tap-snowflake` [test PR](https://github.com/MeltanoLabs/tap-snowflake/pull/15) to point to this branch, verifying the behaviour and test speedup 🚀
- Updated this PR to target `main`. I agree we can close #1399 in favour of this PR 👍

Patch:
diff --git a/poetry.lock b/poetry.lock
index 93418055..33adb680 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.4.0 and should not be changed by hand.
 
 [[package]]
 name = "alabaster"
@@ -173,18 +173,18 @@ uvloop = ["uvloop (>=0.15.2)"]
 
 [[package]]
 name = "boto3"
-version = "1.26.70"
+version = "1.26.75"
 description = "The AWS SDK for Python"
 category = "main"
 optional = true
 python-versions = ">= 3.7"
 files = [
-    {file = "boto3-1.26.70-py3-none-any.whl", hash = "sha256:5324c2a9dbc271d2b25eb79f7469b422de411665f9ab1c0b410e8ac820859b1a"},
-    {file = "boto3-1.26.70.tar.gz", hash = "sha256:123cf34f3cc58772b4f806dfbb1ae9ffae47459de5088e971be9d3cd2b198975"},
+    {file = "boto3-1.26.75-py3-none-any.whl", hash = "sha256:0bbc318e8c4a6006de0069b99810780962f53fed591830fee9ab670aa4ec56ef"},
+    {file = "boto3-1.26.75.tar.gz", hash = "sha256:702efaf333ddd9a1520283a22bd74b6c3a890ab38df5afcf4e821a2f3d707688"},
 ]
 
 [package.dependencies]
-botocore = ">=1.29.70,<1.30.0"
+botocore = ">=1.29.75,<1.30.0"
 jmespath = ">=0.7.1,<2.0.0"
 s3transfer = ">=0.6.0,<0.7.0"
 
@@ -193,14 +193,14 @@ crt = ["botocore[crt] (>=1.21.0,<2.0a0)"]
 
 [[package]]
 name = "botocore"
-version = "1.29.70"
+version = "1.29.75"
 description = "Low-level, data-driven core of boto 3."
 category = "main"
 optional = true
 python-versions = ">= 3.7"
 files = [
-    {file = "botocore-1.29.70-py3-none-any.whl", hash = "sha256:7d9abef42846c1c2f31dacaa559f8450f6fbda74c2c9e3dc6630e06e882ad026"},
-    {file = "botocore-1.29.70.tar.gz", hash = "sha256:caaa144f49ef0d01b5e8812c9afa729def2c3358d9c4d9204789be2b56c5e849"},
+    {file = "botocore-1.29.75-py3-none-any.whl", hash = "sha256:b6e50fc7aacdcc7fa345cc2c55f53e61db8165bdca4d8ca57323e22cac2671c6"},
+    {file = "botocore-1.29.75.tar.gz", hash = "sha256:eef47ca90d02dbc92296208e24ac5e02bdf5cfa45f10d160fdc19612e141bce2"},
 ]
 
 [package.dependencies]
@@ -1579,14 +1579,14 @@ files = [
 
 [[package]]
 name = "pygithub"
-version = "1.57"
+version = "1.58.0"
 description = "Use the full Github API v3"
 category = "dev"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "PyGithub-1.57-py3-none-any.whl", hash = "sha256:5822febeac2391f1306c55a99af2bc8f86c8bf82ded000030cd02c18f31b731f"},
-    {file = "PyGithub-1.57.tar.gz", hash = "sha256:c273f252b278fb81f1769505cc6921bdb6791e1cebd6ac850cc97dad13c31ff3"},
+    {file = "PyGithub-1.58.0-py3-none-any.whl", hash = "sha256:b7bac601492a2b6c876ef326e4ffa3c1923e32707e415da76bfb8307ee8ffb7e"},
+    {file = "PyGithub-1.58.0.tar.gz", hash = "sha256:e325f5ca2c111d19dea2796c8f1e7ae11da77287007736582196341390d31d35"},
 ]
 
 [package.dependencies]
@@ -1934,14 +1934,14 @@ crt = ["botocore[crt] (>=1.20.29,<2.0a.0)"]
 
 [[package]]
 name = "setuptools"
-version = "67.2.0"
+version = "67.3.3"
 description = "Easily download, build, install, upgrade, and uninstall Python packages"
 category = "main"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "setuptools-67.2.0-py3-none-any.whl", hash = "sha256:16ccf598aab3b506593c17378473978908a2734d7336755a8769b480906bec1c"},
-    {file = "setuptools-67.2.0.tar.gz", hash = "sha256:b440ee5f7e607bb8c9de15259dba2583dd41a38879a7abc1d43a71c59524da48"},
+    {file = "setuptools-67.3.3-py3-none-any.whl", hash = "sha256:9d3de8591bd6f6522594406fa46a6418eabd0562dacb267f8556675762801514"},
+    {file = "setuptools-67.3.3.tar.gz", hash = "sha256:ed4e75fafe103c79b692f217158ba87edf38d31004b9dbc1913debb48793c828"},
 ]
 
 [package.extras]
@@ -2070,14 +2070,14 @@ files = [
 
 [[package]]
 name = "soupsieve"
-version = "2.3.2.post1"
+version = "2.4"
 description = "A modern CSS selector implementation for Beautiful Soup."
 category = "main"
 optional = true
-python-versions = ">=3.6"
+python-versions = ">=3.7"
 files = [
-    {file = "soupsieve-2.3.2.post1-py3-none-any.whl", hash = "sha256:3b2503d3c7084a42b1ebd08116e5f81aadfaea95863628c80a3b774a11b7c759"},
-    {file = "soupsieve-2.3.2.post1.tar.gz", hash = "sha256:fc53893b3da2c33de295667a0e19f078c14bf86544af307354de5fcf12a3f30d"},
+    {file = "soupsieve-2.4-py3-none-any.whl", hash = "sha256:49e5368c2cda80ee7e84da9dbe3e110b70a4575f196efb74e51b94549d921955"},
+    {file = "soupsieve-2.4.tar.gz", hash = "sha256:e28dba9ca6c7c00173e34e4ba57448f0688bb681b7c5e8bf4971daafc093d69a"},
 ]
 
 [[package]]
@@ -2335,7 +2335,7 @@ files = [
 ]
 
 [package.dependencies]
-greenlet = {version = "!=0.4.17", markers = "python_version >= \"3\" and (platform_machine == \"aarch64\" or platform_machine == \"ppc64le\" or platform_machine == \"x86_64\" or platform_machine == \"amd64\" or platform_machine == \"AMD64\" or platform_machine == \"win32\" or platform_machine == \"WIN32\")"}
+greenlet = {version = "!=0.4.17", markers = "python_version >= \"3\" and platform_machine == \"aarch64\" or python_version >= \"3\" and platform_machine == \"ppc64le\" or python_version >= \"3\" and platform_machine == \"x86_64\" or python_version >= \"3\" and platform_machine == \"amd64\" or python_version >= \"3\" and platform_machine == \"AMD64\" or python_version >= \"3\" and platform_machine == \"win32\" or python_version >= \"3\" and platform_machine == \"WIN32\""}
 importlib-metadata = {version = "*", markers = "python_version < \"3.8\""}
 
 [package.extras]
@@ -2530,14 +2530,14 @@ files = [
 
 [[package]]
 name = "types-urllib3"
-version = "1.26.25.5"
+version = "1.26.25.7"
 description = "Typing stubs for urllib3"
 category = "dev"
 optional = false
 python-versions = "*"
 files = [
-    {file = "types-urllib3-1.26.25.5.tar.gz", hash = "sha256:5630e578246d170d91ebe3901788cd28d53c4e044dc2e2488e3b0d55fb6895d8"},
-    {file = "types_urllib3-1.26.25.5-py3-none-any.whl", hash = "sha256:e8f25c8bb85cde658c72ee931e56e7abd28803c26032441eea9ff4a4df2b0c31"},
+    {file = "types-urllib3-1.26.25.7.tar.gz", hash = "sha256:df4d3e5472bf8830bd74eac12d56e659f88662ba040c7d106bf3a5bee26fff28"},
+    {file = "types_urllib3-1.26.25.7-py3-none-any.whl", hash = "sha256:28d2d7f5c31ff8ed4d9d2e396ce906c49d37523c3ec207d03d3b1695755a7199"},
 ]
 
 [[package]]
@@ -2673,14 +2673,14 @@ tests-strict = ["codecov (==2.0.15)", "pytest (==4.6.0)", "pytest (==4.6.0)", "p
 
 [[package]]
 name = "zipp"
-version = "3.13.0"
+version = "3.14.0"
 description = "Backport of pathlib-compatible object wrapper for zip files"
 category = "main"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "zipp-3.13.0-py3-none-any.whl", hash = "sha256:e8b2a36ea17df80ffe9e2c4fda3f693c3dad6df1697d3cd3af232db680950b0b"},
-    {file = "zipp-3.13.0.tar.gz", hash = "sha256:23f70e964bc11a34cef175bc90ba2914e1e4545ea1e3e2f67c079671883f9cb6"},
+    {file = "zipp-3.14.0-py3-none-any.whl", hash = "sha256:188834565033387710d046e3fe96acfc9b5e86cbca7f39ff69cf21a4128198b7"},
+    {file = "zipp-3.14.0.tar.gz", hash = "sha256:9e5421e176ef5ab4c0ad896624e87a7b2f07aca746c9b2aa305952800cb8eecb"},
 ]
 
 [package.extras]
@@ -2688,7 +2688,7 @@ docs = ["furo", "jaraco.packaging (>=9)", "jaraco.tidelift (>=1.4)", "rst.linker
 testing = ["flake8 (<5)", "func-timeout", "jaraco.functools", "jaraco.itertools", "more-itertools", "pytest (>=6)", "pytest-black (>=0.3.7)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=1.3)", "pytest-flake8", "pytest-mypy (>=0.9.1)"]
 
 [extras]
-docs = ["sphinx", "furo", "sphinx-copybutton", "myst-parser", "sphinx-autobuild", "sphinx-reredirects"]
+docs = ["furo", "myst-parser", "sphinx", "sphinx-autobuild", "sphinx-copybutton", "sphinx-reredirects"]
 s3 = ["fs-s3fs"]
 testing = ["pytest", "pytest-durations"]
 
diff --git a/singer_sdk/exceptions.py b/singer_sdk/exceptions.py
index c3b9bda4..221a26f4 100644
--- a/singer_sdk/exceptions.py
+++ b/singer_sdk/exceptions.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import abc
 import typing as t
 
 if t.TYPE_CHECKING:
@@ -24,8 +25,59 @@ class MapExpressionError(Exception):
     """Failed map expression evaluation."""
 
 
-class MaxRecordsLimitException(Exception):
-    """Exception to raise if the maximum number of allowable records is exceeded."""
+class RequestedAbortException(Exception):
+    """Base class for abort and interrupt requests.
+
+    Whenever this exception is raised, streams will attempt to shut down gracefully and
+    will emit a final resumable `STATE` message if it is possible to do so.
+    """
+
+
+class MaxRecordsLimitException(RequestedAbortException):
+    """Exception indicating the sync aborted due to too many records."""
+
+
+class AbortedSyncExceptionBase(Exception, metaclass=abc.ABCMeta):
+    """Base exception to raise when a stream sync is aborted.
+
+    Developers should not raise this directly, and instead should use:
+    1. `FatalAbortedSyncException` - Indicates the stream aborted abnormally and was not
+       able to reach a stable and resumable state.
+    2. `PausedSyncException` - Indicates the stream aborted abnormally and successfully
+       reached a 'paused' and resumable state.
+
+    Notes:
+    - `FULL_TABLE` sync operations cannot be paused and will always trigger a fatal
+      exception if aborted.
+    - `INCREMENTAL` and `LOG_BASED` streams are able to be paused only if a number of
+      preconditions are met, specifically, `state_partitioning_keys` cannot be
+      overridden and the stream must be declared with `is_sorted=True`.
+    """
+
+
+class AbortedSyncFailedException(AbortedSyncExceptionBase):
+    """Exception to raise when sync is aborted and unable to reach a stable state.
+
+    This signifies that `FULL_TABLE` streams (if applicable) were successfully
+    completed, and any bookmarks from `INCREMENTAL` and `LOG_BASED` streams were
+    advanced and finalized successfully.
+    """
+
+
+class AbortedSyncPausedException(AbortedSyncExceptionBase):
+    """Exception to raise when an aborted sync operation is paused successfully.
+
+    This exception indicates the stream aborted abnormally and successfully
+       reached a 'paused' status, and emitted a resumable state artifact before exiting.
+
+    Streams synced with `FULL_TABLE` replication can never have partial success or
+    'paused' status.
+
+    If this exception is raised, this signifies that additional records were left
+    on the source system and the sync operation aborted before reaching the end of the
+    stream. This exception signifies that bookmarks from `INCREMENTAL`
+    and `LOG_BASED` streams were successfully emitted and are resumable.
+    """
 
 
 class RecordsWithoutSchemaException(Exception):
diff --git a/singer_sdk/helpers/_state.py b/singer_sdk/helpers/_state.py
index c01ff29b..aad15961 100644
--- a/singer_sdk/helpers/_state.py
+++ b/singer_sdk/helpers/_state.py
@@ -234,12 +234,24 @@ def _greater_than_signpost(
     return new_value > signpost
 
 
+def is_state_non_resumable(stream_or_partition_state: dict) -> bool:
+    """Return True when state is non-resumable.
+
+    This is determined by checking for a "progress marker" tag in the state artifact.
+    """
+    return PROGRESS_MARKERS in stream_or_partition_state
+
+
 def finalize_state_progress_markers(stream_or_partition_state: dict) -> dict | None:
-    """Promote or wipe progress markers once sync is complete."""
+    """Promote or wipe progress markers once sync is complete.
+
+    This marks any non-resumable progress markers as finalized. If there are
+    valid bookmarks present, they will be promoted to be resumable.
+    """
     signpost_value = stream_or_partition_state.pop(SIGNPOST_MARKER, None)
     stream_or_partition_state.pop(STARTING_MARKER, None)
     if (
-        PROGRESS_MARKERS in stream_or_partition_state
+        is_state_non_resumable(stream_or_partition_state)
         and "replication_key" in stream_or_partition_state[PROGRESS_MARKERS]
     ):
         # Replication keys valid (only) after sync is complete
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index e006d0c7..222131c7 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -27,7 +27,12 @@ import pendulum
 
 import singer_sdk._singerlib as singer
 from singer_sdk import metrics
-from singer_sdk.exceptions import InvalidStreamSortException, MaxRecordsLimitException
+from singer_sdk.exceptions import (
+    AbortedSyncFailedException,
+    AbortedSyncPausedException,
+    InvalidStreamSortException,
+    MaxRecordsLimitException,
+)
 from singer_sdk.helpers._batch import (
     BaseBatchFileEncoding,
     BatchConfig,
@@ -42,6 +47,7 @@ from singer_sdk.helpers._state import (
     get_state_partitions_list,
     get_writeable_state_dict,
     increment_state,
+    is_state_non_resumable,
     log_sort_error,
     reset_state_progress_markers,
     write_replication_key_signpost,
@@ -96,7 +102,10 @@ class Stream(metaclass=abc.ABCMeta):
     STATE_MSG_FREQUENCY = 10000
     """Number of records between state messages."""
 
-    _MAX_RECORDS_LIMIT: int | None = None
+    ABORT_AT_RECORD_COUNT: int | None = None
+    """
+    If set, raise `MaxRecordsLimitException` if the limit is exceeded.
+    """
 
     TYPE_CONFORMANCE_LEVEL = TypeConformanceLevel.RECURSIVE
     """Type conformance level for this stream.
@@ -166,6 +175,7 @@ class Stream(metaclass=abc.ABCMeta):
         self._metadata: singer.MetadataMapping | None = None
         self._mask: singer.SelectionMask | None = None
         self._schema: dict
+        self._is_state_flushed: bool = True
         self.child_streams: list[Stream] = []
         if schema:
             if isinstance(schema, (PathLike, str)):
@@ -776,7 +786,9 @@ class Stream(metaclass=abc.ABCMeta):
 
     def _write_state_message(self) -> None:
         """Write out a STATE message with the latest state."""
-        singer.write_message(singer.StateMessage(value=self.tap_state))
+        if not self._is_state_flushed:
+            singer.write_message(singer.StateMessage(value=self.tap_state))
+            self._is_state_flushed = True
 
     def _generate_schema_messages(self) -> Generator[singer.SchemaMessage, None, None]:
         """Generate schema messages from stream maps.
@@ -857,6 +869,8 @@ class Stream(metaclass=abc.ABCMeta):
         for record_message in self._generate_record_messages(record):
             singer.write_message(record_message)
 
+        self._is_state_flushed = False
+
     def _write_batch_message(
         self,
         encoding: BaseBatchFileEncoding,
@@ -875,6 +889,7 @@ class Stream(metaclass=abc.ABCMeta):
                 manifest=manifest,
             ),
         )
+        self._is_state_flushed = False
 
     def _log_metric(self, point: metrics.Point) -> None:
         """Log a single measurement.
@@ -896,23 +911,70 @@ class Stream(metaclass=abc.ABCMeta):
             msg = f"Total Sync costs for stream {self.name}: {self._sync_costs}"
             self.logger.info(msg)
 
-    def _check_max_record_limit(self, record_count: int) -> None:
-        """TODO.
+    def _check_max_record_limit(self, current_record_index: int) -> None:
+        """Raise an exception if dry run record limit exceeded.
+
+        Raised if we find dry run record limit exceeded,
+        aka `current_record_index > self.ABORT_AT_RECORD_COUNT - 1`.
 
         Args:
-            record_count: TODO.
+            current_record_index: The zero-based index of the current record.
 
         Raises:
-            MaxRecordsLimitException: TODO.
+            AbortedSyncFailedException: Raised if sync could not reach a valid state.
+            AbortedSyncPausedException: Raised if sync was able to be transitioned into
+                a valid state without data loss or corruption.
         """
         if (
-            self._MAX_RECORDS_LIMIT is not None
-            and record_count >= self._MAX_RECORDS_LIMIT
+            self.ABORT_AT_RECORD_COUNT is not None
+            and current_record_index > self.ABORT_AT_RECORD_COUNT - 1
         ):
-            raise MaxRecordsLimitException(
-                "Stream prematurely aborted due to the stream's max record "
-                f"limit ({self._MAX_RECORDS_LIMIT}) being reached.",
-            )
+            try:
+                self._abort_sync(
+                    abort_reason=MaxRecordsLimitException(
+                        "Stream prematurely aborted due to the stream's max dry run "
+                        f"record limit ({self.ABORT_AT_RECORD_COUNT}) being reached.",
+                    ),
+                )
+            except (AbortedSyncFailedException, AbortedSyncPausedException) as ex:
+                raise ex
+
+    def _abort_sync(self, abort_reason: Exception) -> None:
+        """Handle a sync operation being aborted.
+
+        This method will attempt to close out the sync operation as gracefully as
+        possible - for instance, if a max runtime or record count is reached. This can
+        also be called for `SIGTERM` and KeyboardInterrupt events.
+
+        If a state message is pending, we will attempt to write it to STDOUT before
+        shutting down.
+
+        If the stream can reach a valid resumable state, then we will raise
+        `AbortedSyncPausedException`. Otherwise, `AbortedSyncFailedException` will be
+        raised.
+
+        Args:
+            abort_reason: The exception that triggered the sync to be aborted.
+
+        Raises:
+            AbortedSyncFailedException: Raised if sync could not reach a valid state.
+            AbortedSyncPausedException: Raised if sync was able to be transitioned into
+                a valid state without data loss or corruption.
+        """
+        self._write_state_message()  # Write out state message if pending.
+
+        if self.replication_method == "FULL_TABLE":
+            raise AbortedSyncFailedException(
+                "Sync operation aborted for stream in 'FULL_TABLE' replication mode.",
+            ) from abort_reason
+
+        if is_state_non_resumable(self.stream_state):
+            raise AbortedSyncFailedException(
+                "Sync operation aborted and state is not in a resumable state.",
+            ) from abort_reason
+
+        # Else, the sync operation can be assumed to be in a valid resumable state.
+        raise AbortedSyncPausedException from abort_reason
 
     # Handle interim stream state
 
@@ -997,7 +1059,8 @@ class Stream(metaclass=abc.ABCMeta):
             write_messages: Whether to write Singer messages to stdout.
 
         Raises:
-            InvalidStreamSortException: TODO
+            InvalidStreamSortException: Raised if sorting errors are found while
+                syncing the records.
 
         Yields:
             Each record from the source.
@@ -1006,7 +1069,7 @@ class Stream(metaclass=abc.ABCMeta):
         record_counter = metrics.record_counter(self.name)
         timer = metrics.sync_timer(self.name)
 
-        record_count = 0
+        record_index = 0
         current_context: dict | None
         context_list: list[dict] | None
         context_list = [context] if context is not None else self.partitions
@@ -1017,7 +1080,7 @@ class Stream(metaclass=abc.ABCMeta):
                 record_counter.context = current_context
                 timer.context = current_context
 
-                partition_record_count = 0
+                partition_record_index = 0
                 current_context = current_context or None
                 state = self.get_context_state(current_context)
                 state_partition_context = self._get_state_partition_context(
@@ -1029,6 +1092,8 @@ class Stream(metaclass=abc.ABCMeta):
                 )
 
                 for record_result in self.get_records(current_context):
+                    self._check_max_record_limit(current_record_index=record_index)
+
                     if isinstance(record_result, tuple):
                         # Tuple items should be the record and the child context
                         record, child_context = record_result
@@ -1044,30 +1109,29 @@ class Stream(metaclass=abc.ABCMeta):
                         log_sort_error(
                             log_fn=self.logger.error,
                             ex=ex,
-                            record_count=record_count + 1,
-                            partition_record_count=partition_record_count + 1,
+                            record_count=record_index + 1,
+                            partition_record_count=partition_record_index + 1,
                             current_context=current_context,
                             state_partition_context=state_partition_context,
                             stream_name=self.name,
                         )
                         raise ex
 
-                    self._check_max_record_limit(record_count)
-
                     if selected:
-                        if (
-                            record_count - 1
-                        ) % self.STATE_MSG_FREQUENCY == 0 and write_messages:
-                            self._write_state_message()
                         if write_messages:
                             self._write_record_message(record)
+
                         self._increment_stream_state(record, context=current_context)
+                        if (
+                            record_index + 1
+                        ) % self.STATE_MSG_FREQUENCY == 0 and write_messages:
+                            self._write_state_message()
 
+                        record_counter.increment()
                         yield record
 
-                        record_counter.increment()
-                        record_count += 1
-                        partition_record_count += 1
+                    record_index += 1
+                    partition_record_index += 1
 
                 if current_context == state_partition_context:
                     # Finalize per-partition state only if 1:1 with context
diff --git a/singer_sdk/streams/sql.py b/singer_sdk/streams/sql.py
index 8fb1a2dc..dfe64f53 100644
--- a/singer_sdk/streams/sql.py
+++ b/singer_sdk/streams/sql.py
@@ -197,8 +197,12 @@ class SQLStream(Stream, metaclass=abc.ABCMeta):
                     ),
                 )
 
-        if self._MAX_RECORDS_LIMIT is not None:
-            query = query.limit(self._MAX_RECORDS_LIMIT)
+        if self.ABORT_AT_RECORD_COUNT is not None:
+            # Limit record count to one greater than the abort threshold. This ensures
+            # `MaxRecordsLimitException` exception is properly raised by caller
+            # `Stream._sync_records()` if more records are available than can be
+            # processed.
+            query = query.limit(self.ABORT_AT_RECORD_COUNT + 1)
 
         with self.connector._connect() as conn:
             for record in conn.execute(query):
diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py
index f75bed38..c6bb3df2 100644
--- a/singer_sdk/tap_base.py
+++ b/singer_sdk/tap_base.py
@@ -1,5 +1,6 @@
 """Tap abstract class."""
 
+
 from __future__ import annotations
 
 import abc
@@ -7,13 +8,13 @@ import contextlib
 import json
 from enum import Enum
 from pathlib import Path, PurePath
-from typing import TYPE_CHECKING, Any, Callable, Sequence, cast
+from typing import TYPE_CHECKING, Any, Callable, Iterable, Sequence, cast
 
 import click
 
 from singer_sdk._singerlib import Catalog
 from singer_sdk.cli import common_options
-from singer_sdk.exceptions import MaxRecordsLimitException
+from singer_sdk.exceptions import AbortedSyncFailedException, AbortedSyncPausedException
 from singer_sdk.helpers import _state
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers._compat import final
@@ -179,20 +180,46 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
             PluginCapabilities.FLATTENING,
         ]
 
-    # Connection test:
+    # Connection and sync tests:
 
     @final
     def run_connection_test(self) -> bool:
+        """Run connection test, aborting each stream after 1 record.
+
+        Returns:
+            True if the test succeeded.
+        """
+        return self.run_sync_dry_run(
+            dry_run_record_limit=1,
+            streams=self.streams.values(),
+        )
+
+    @final
+    def run_sync_dry_run(
+        self,
+        dry_run_record_limit: int | None = 1,
+        streams: Iterable[Stream] | None = None,
+    ) -> bool:
         """Run connection test.
 
+        Exceptions of type `MaxRecordsLimitException` and
+        `PartialSyncSuccessException` will be ignored.
+
+        Args:
+            dry_run_record_limit: The max number of records to sync per stream object.
+            streams: The streams to test. If omitted, all streams will be tested.
+
         Returns:
             True if the test succeeded.
         """
-        for stream in self.streams.values():
+        if streams is None:
+            streams = self.streams.values()
+
+        for stream in streams:
             # Initialize streams' record limits before beginning the sync test.
-            stream._MAX_RECORDS_LIMIT = 1
+            stream.ABORT_AT_RECORD_COUNT = dry_run_record_limit
 
-        for stream in self.streams.values():
+        for stream in streams:
             if stream.parent_stream_type:
                 self.logger.debug(
                     f"Child stream '{type(stream).__name__}' should be called by "
@@ -200,7 +227,10 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
                     "Skipping direct invocation.",
                 )
                 continue
-            with contextlib.suppress(MaxRecordsLimitException):
+            with contextlib.suppress(
+                AbortedSyncFailedException,
+                AbortedSyncPausedException,
+            ):
                 stream.sync()
         return True
 
diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index c3319aca..0364ec2e 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -265,6 +265,15 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
 
     # Message handling
 
+    def _handle_max_record_age(self) -> None:
+        """Check if _MAX_RECORD_AGE_IN_MINUTES reached, and if so trigger drain."""
+        if self._max_record_age_in_minutes > self._MAX_RECORD_AGE_IN_MINUTES:
+            self.logger.info(
+                "One or more records have exceeded the max age of "
+                f"{self._MAX_RECORD_AGE_IN_MINUTES} minutes. Draining all sinks.",
+            )
+            self.drain_all()
+
     def _process_lines(self, file_input: IO[str]) -> Counter[str]:
         """Internal method to process jsonl lines from a Singer tap.
 
@@ -332,6 +341,8 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
                 )
                 self.drain_one(sink)
 
+        self._handle_max_record_age()
+
     def _process_schema_message(self, message_dict: dict) -> None:
         """Process a SCHEMA messages.
 
@@ -401,12 +412,6 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
         if self._latest_state == state:
             return
         self._latest_state = state
-        if self._max_record_age_in_minutes > self._MAX_RECORD_AGE_IN_MINUTES:
-            self.logger.info(
-                "One or more records have exceeded the max age of "
-                f"{self._MAX_RECORD_AGE_IN_MINUTES} minutes. Draining all sinks.",
-            )
-            self.drain_all()
 
     def _process_activate_version_message(self, message_dict: dict) -> None:
         """Handle the optional ACTIVATE_VERSION message extension.
@@ -431,6 +436,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
             encoding,
             message_dict["manifest"],
         )
+        self._handle_max_record_age()
 
     # Sink drain methods
 


Test patch:
diff --git a/singer_sdk/testing/config.py b/singer_sdk/testing/config.py
index f6e010d3..ebdc81ff 100644
--- a/singer_sdk/testing/config.py
+++ b/singer_sdk/testing/config.py
@@ -10,11 +10,13 @@ class SuiteConfig:
     """Test Suite Config, passed to each test.
 
     Args:
+        max_records_limit: Max records to fetch during tap testing.
         ignore_no_records: Ignore stream test failures if stream returns no records,
             for all streams.
         ignore_no_records_for_streams: Ignore stream test failures if stream returns
              no records, for named streams.
     """
 
+    max_records_limit: int | None = None
     ignore_no_records: bool = False
     ignore_no_records_for_streams: list[str] = field(default_factory=list)
diff --git a/singer_sdk/testing/factory.py b/singer_sdk/testing/factory.py
index ca589235..45dc3d1c 100644
--- a/singer_sdk/testing/factory.py
+++ b/singer_sdk/testing/factory.py
@@ -34,7 +34,6 @@ def get_test_class(
     Returns:
         A test class usable by pytest.
     """
-    suite_config = suite_config or SuiteConfig()
 
     class BaseTestClass:
         """Base test class."""
@@ -79,7 +78,7 @@ def get_test_class(
                 setattr(BaseTestClass, f"test_{suite.kind}_{test.name}", test.run)
 
         if suite.kind in {"tap_stream", "tap_stream_attribute"}:
-            streams = list(test_runner.tap.streams.values())
+            streams = list(test_runner.new_tap().streams.values())
 
             if suite.kind == "tap_stream":
                 params = [
@@ -188,10 +187,13 @@ def get_tap_test_class(
     if "parse_env_config" not in kwargs:
         kwargs["parse_env_config"] = True
 
-    suite_config = suite_config or SuiteConfig()
-
     return get_test_class(
-        test_runner=TapTestRunner(tap_class=tap_class, config=config, **kwargs),
+        test_runner=TapTestRunner(
+            tap_class=tap_class,
+            config=config,
+            suite_config=suite_config,
+            **kwargs,
+        ),
         test_suites=suites,
         suite_config=suite_config,
     )
@@ -224,8 +226,6 @@ def get_target_test_class(
     if "parse_env_config" not in kwargs:
         kwargs["parse_env_config"] = True
 
-    suite_config = suite_config or SuiteConfig()
-
     return get_test_class(
         test_runner=TargetTestRunner(
             target_class=target_class,
diff --git a/singer_sdk/testing/runners.py b/singer_sdk/testing/runners.py
index c83a3f31..5967b657 100644
--- a/singer_sdk/testing/runners.py
+++ b/singer_sdk/testing/runners.py
@@ -105,8 +105,7 @@ class TapTestRunner(SingerTestRunner):
             **kwargs,
         )
 
-    @property
-    def tap(self) -> Tap:
+    def new_tap(self) -> Tap:
         """Get new Tap instance.
 
         Returns:
@@ -120,7 +119,7 @@ class TapTestRunner(SingerTestRunner):
         Returns:
             The catalog as a string.
         """
-        return self.tap.run_discovery()
+        return self.new_tap().run_discovery()
 
     def run_connection_test(self) -> bool:
         """Run tap connection test.
@@ -128,7 +127,21 @@ class TapTestRunner(SingerTestRunner):
         Returns:
             True if connection test passes, else False.
         """
-        return self.tap.run_connection_test()
+        new_tap = self.new_tap()
+        return new_tap.run_connection_test()
+
+    def run_sync_dry_run(self) -> bool:
+        """Run tap sync dry run.
+
+        Returns:
+            True if dry run test passes, else False.
+        """
+        new_tap = self.new_tap()
+        dry_run_record_limit = None
+        if self.suite_config.max_records_limit is not None:
+            dry_run_record_limit = self.suite_config.max_records_limit
+
+        return new_tap.run_sync_dry_run(dry_run_record_limit=dry_run_record_limit)
 
     def sync_all(self, **kwargs: Any) -> None:  # noqa: ARG002
         """Run a full tap sync, assigning output to the runner object.
@@ -171,7 +184,7 @@ class TapTestRunner(SingerTestRunner):
         stdout_buf = io.StringIO()
         stderr_buf = io.StringIO()
         with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
-            self.tap.sync_all()
+            self.run_sync_dry_run()
         stdout_buf.seek(0)
         stderr_buf.seek(0)
         return stdout_buf.read(), stderr_buf.read()
@@ -211,8 +224,7 @@ class TargetTestRunner(SingerTestRunner):
         self.input_io = input_io
         self._input: IO[str] | None = None
 
-    @property
-    def target(self) -> Target:
+    def new_target(self) -> Target:
         """Get new Target instance.
 
         Returns:
@@ -246,7 +258,7 @@ class TargetTestRunner(SingerTestRunner):
                 False to keep the sink operation open for further records.
             kwargs: Unused keyword arguments.
         """
-        target = cast(Target, self.create())
+        target = self.new_target()
         stdout, stderr = self._execute_sync(
             target=target,
             target_input=self.target_input,
diff --git a/singer_sdk/testing/templates.py b/singer_sdk/testing/templates.py
index 942f3027..5de510a2 100644
--- a/singer_sdk/testing/templates.py
+++ b/singer_sdk/testing/templates.py
@@ -143,7 +143,7 @@ class TapTestTemplate(TestTemplate):
             resource: A generic external resource, provided by a pytest fixture.
             runner: A Tap or Target runner instance, to use with this test.
         """
-        self.tap = runner.tap
+        self.tap = runner.new_tap()
         super().run(config, resource, runner)
 
 
@@ -279,7 +279,7 @@ class TargetTestTemplate(TestTemplate):
             resource: A generic external resource, provided by a pytest fixture.
             runner: A Tap runner instance, to use with this test.
         """
-        self.target = runner.target
+        self.target = runner.new_target()
         super().run(config, resource, runner)
 
     @property
diff --git a/tests/core/test_parent_child.py b/tests/core/test_parent_child.py
index e361579d..b648d73d 100644
--- a/tests/core/test_parent_child.py
+++ b/tests/core/test_parent_child.py
@@ -111,8 +111,8 @@ def test_parent_context_fields_in_child(tap: MyTap):
     assert messages[1]["stream"] == child_stream.name
     assert messages[1]["schema"] == child_stream.schema
 
-    # Child records are emitted, skip state message in between
-    child_record_messages = messages[2], *messages[4:6]
+    # Child records are emitted
+    child_record_messages = messages[2:5]
     assert child_record_messages
     assert all(msg["type"] == SingerMessageType.RECORD for msg in child_record_messages)
     assert all(msg["stream"] == child_stream.name for msg in child_record_messages)
@@ -141,8 +141,8 @@ def test_child_deselected_parent(tap_with_deselected_parent: MyTap):
     assert messages[0]["stream"] == child_stream.name
     assert messages[0]["schema"] == child_stream.schema
 
-    # Child records are emitted, skip state message in between
-    child_record_messages = messages[1], *messages[3:5]
+    # Child records are emitted
+    child_record_messages = messages[1:4]
     assert child_record_messages
     assert all(msg["type"] == SingerMessageType.RECORD for msg in child_record_messages)
     assert all(msg["stream"] == child_stream.name for msg in child_record_messages)
diff --git a/tests/samples/test_tap_countries.py b/tests/samples/test_tap_countries.py
index c4cc1181..56b2c35f 100644
--- a/tests/samples/test_tap_countries.py
+++ b/tests/samples/test_tap_countries.py
@@ -15,6 +15,7 @@ from singer_sdk.helpers._catalog import (
     pop_deselected_record_properties,
 )
 from singer_sdk.testing import get_tap_test_class
+from singer_sdk.testing.config import SuiteConfig
 
 SAMPLE_CONFIG = {}
 SAMPLE_CONFIG_BAD = {"not": "correct"}
@@ -23,6 +24,7 @@ SAMPLE_CONFIG_BAD = {"not": "correct"}
 TestSampleTapCountries = get_tap_test_class(
     tap_class=SampleTapCountries,
     config=SAMPLE_CONFIG,
+    suite_config=SuiteConfig(max_records_limit=5),
 )
 
 
@@ -133,4 +135,4 @@ def test_batch_mode(outdir):
     assert counter["SCHEMA", "countries"] == 1
     assert counter["BATCH", "countries"] == 1
 
-    assert counter[("STATE",)] == 4
+    assert counter[("STATE",)] == 2
diff --git a/tests/snapshots/mapped_stream/aliased_stream.jsonl b/tests/snapshots/mapped_stream/aliased_stream.jsonl
index 9750aa87..49089dda 100644
--- a/tests/snapshots/mapped_stream/aliased_stream.jsonl
+++ b/tests/snapshots/mapped_stream/aliased_stream.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "aliased_stream", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "aliased_stream", "record": {"email": "alice@example.com", "count": 21, "user": {"id": 1, "sub": {"num": 1}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "aliased_stream", "record": {"email": "bob@example.com", "count": 13, "user": {"id": 2, "sub": {"num": 2}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "aliased_stream", "record": {"email": "charlie@example.com", "count": 19, "user": {"id": 3, "sub": {"num": 3}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/changed_key_properties.jsonl b/tests/snapshots/mapped_stream/changed_key_properties.jsonl
index 72effee8..f62049ce 100644
--- a/tests/snapshots/mapped_stream/changed_key_properties.jsonl
+++ b/tests/snapshots/mapped_stream/changed_key_properties.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"type": "object", "properties": {"email_hash": {"type": ["string", "null"]}}}, "key_properties": ["email_hash"]}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "c160f8cc69a4f0bf2b0362752353d060"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "4b9bb80620f03eb3719e0a061c14283d"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "426b189df1e2f359efe6ee90f2d2030f"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/drop_property.jsonl b/tests/snapshots/mapped_stream/drop_property.jsonl
index bd87afff..f4b0b6f7 100644
--- a/tests/snapshots/mapped_stream/drop_property.jsonl
+++ b/tests/snapshots/mapped_stream/drop_property.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"properties": {"count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 21, "user": {"id": 1, "sub": {"num": 1}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 13, "user": {"id": 2, "sub": {"num": 2}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 19, "user": {"id": 3, "sub": {"num": 3}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/drop_property_null_string.jsonl b/tests/snapshots/mapped_stream/drop_property_null_string.jsonl
index bd87afff..f4b0b6f7 100644
--- a/tests/snapshots/mapped_stream/drop_property_null_string.jsonl
+++ b/tests/snapshots/mapped_stream/drop_property_null_string.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"properties": {"count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 21, "user": {"id": 1, "sub": {"num": 1}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 13, "user": {"id": 2, "sub": {"num": 2}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 19, "user": {"id": 3, "sub": {"num": 3}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/flatten_all.jsonl b/tests/snapshots/mapped_stream/flatten_all.jsonl
index 4a88fa2c..03e7af0c 100644
--- a/tests/snapshots/mapped_stream/flatten_all.jsonl
+++ b/tests/snapshots/mapped_stream/flatten_all.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user__id": {"type": ["integer", "null"]}, "user__sub__num": {"type": ["integer", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "alice@example.com", "count": 21, "user__id": 1, "user__sub__num": 1}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "bob@example.com", "count": 13, "user__id": 2, "user__sub__num": 2}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "charlie@example.com", "count": 19, "user__id": 3, "user__sub__num": 3}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/flatten_depth_1.jsonl b/tests/snapshots/mapped_stream/flatten_depth_1.jsonl
index 36e8fbbc..6765ccde 100644
--- a/tests/snapshots/mapped_stream/flatten_depth_1.jsonl
+++ b/tests/snapshots/mapped_stream/flatten_depth_1.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user__id": {"type": ["integer", "null"]}, "user__sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "alice@example.com", "count": 21, "user__id": 1, "user__sub": "{\"num\": 1}"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "bob@example.com", "count": 13, "user__id": 2, "user__sub": "{\"num\": 2}"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "charlie@example.com", "count": 19, "user__id": 3, "user__sub": "{\"num\": 3}"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/keep_all_fields.jsonl b/tests/snapshots/mapped_stream/keep_all_fields.jsonl
index f6cea281..2be7c110 100644
--- a/tests/snapshots/mapped_stream/keep_all_fields.jsonl
+++ b/tests/snapshots/mapped_stream/keep_all_fields.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}, "email_hash": {"type": ["string", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "alice@example.com", "count": 21, "user": {"id": 1, "sub": {"num": 1}}, "email_hash": "c160f8cc69a4f0bf2b0362752353d060"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "bob@example.com", "count": 13, "user": {"id": 2, "sub": {"num": 2}}, "email_hash": "4b9bb80620f03eb3719e0a061c14283d"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "charlie@example.com", "count": 19, "user": {"id": 3, "sub": {"num": 3}}, "email_hash": "426b189df1e2f359efe6ee90f2d2030f"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/map_and_flatten.jsonl b/tests/snapshots/mapped_stream/map_and_flatten.jsonl
index 8309730b..40ecc7e4 100644
--- a/tests/snapshots/mapped_stream/map_and_flatten.jsonl
+++ b/tests/snapshots/mapped_stream/map_and_flatten.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user__id": {"type": ["integer", "null"]}, "user__sub__num": {"type": ["integer", "null"]}, "email_hash": {"type": ["string", "null"]}}, "type": "object"}, "key_properties": ["email_hash"]}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "alice@example.com", "count": 21, "user__id": 1, "user__sub__num": 1, "email_hash": "c160f8cc69a4f0bf2b0362752353d060"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "bob@example.com", "count": 13, "user__id": 2, "user__sub__num": 2, "email_hash": "4b9bb80620f03eb3719e0a061c14283d"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "charlie@example.com", "count": 19, "user__id": 3, "user__sub__num": 3, "email_hash": "426b189df1e2f359efe6ee90f2d2030f"}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/no_map.jsonl b/tests/snapshots/mapped_stream/no_map.jsonl
index 4adf43f6..f2eeb77f 100644
--- a/tests/snapshots/mapped_stream/no_map.jsonl
+++ b/tests/snapshots/mapped_stream/no_map.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "alice@example.com", "count": 21, "user": {"id": 1, "sub": {"num": 1}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "bob@example.com", "count": 13, "user": {"id": 2, "sub": {"num": 2}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email": "charlie@example.com", "count": 19, "user": {"id": 3, "sub": {"num": 3}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/non_pk_passthrough.jsonl b/tests/snapshots/mapped_stream/non_pk_passthrough.jsonl
index 4d58bd19..4c2490c6 100644
--- a/tests/snapshots/mapped_stream/non_pk_passthrough.jsonl
+++ b/tests/snapshots/mapped_stream/non_pk_passthrough.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"type": "object", "properties": {"count": {"type": ["integer", "null"]}}}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 21}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 13}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"count": 19}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/only_mapped_fields.jsonl b/tests/snapshots/mapped_stream/only_mapped_fields.jsonl
index 97af5796..d5341c62 100644
--- a/tests/snapshots/mapped_stream/only_mapped_fields.jsonl
+++ b/tests/snapshots/mapped_stream/only_mapped_fields.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"type": "object", "properties": {"email_hash": {"type": ["string", "null"]}, "fixed_count": {"type": ["integer", "null"]}}}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "c160f8cc69a4f0bf2b0362752353d060", "fixed_count": 20}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "4b9bb80620f03eb3719e0a061c14283d", "fixed_count": 12}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "426b189df1e2f359efe6ee90f2d2030f", "fixed_count": 18}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/only_mapped_fields_null_string.jsonl b/tests/snapshots/mapped_stream/only_mapped_fields_null_string.jsonl
index 97af5796..d5341c62 100644
--- a/tests/snapshots/mapped_stream/only_mapped_fields_null_string.jsonl
+++ b/tests/snapshots/mapped_stream/only_mapped_fields_null_string.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "mystream", "schema": {"type": "object", "properties": {"email_hash": {"type": ["string", "null"]}, "fixed_count": {"type": ["integer", "null"]}}}, "key_properties": []}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "c160f8cc69a4f0bf2b0362752353d060", "fixed_count": 20}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "4b9bb80620f03eb3719e0a061c14283d", "fixed_count": 12}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "mystream", "record": {"email_hash": "426b189df1e2f359efe6ee90f2d2030f", "fixed_count": 18}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/sourced_stream_1.jsonl b/tests/snapshots/mapped_stream/sourced_stream_1.jsonl
index 95d230f9..3d83ea30 100644
--- a/tests/snapshots/mapped_stream/sourced_stream_1.jsonl
+++ b/tests/snapshots/mapped_stream/sourced_stream_1.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "sourced_stream_1", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "sourced_stream_1", "record": {"email": "alice@example.com", "count": 21, "user": {"id": 1, "sub": {"num": 1}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "sourced_stream_1", "record": {"email": "bob@example.com", "count": 13, "user": {"id": 2, "sub": {"num": 2}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "sourced_stream_1", "record": {"email": "charlie@example.com", "count": 19, "user": {"id": 3, "sub": {"num": 3}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/sourced_stream_1_null_string.jsonl b/tests/snapshots/mapped_stream/sourced_stream_1_null_string.jsonl
index 95d230f9..3d83ea30 100644
--- a/tests/snapshots/mapped_stream/sourced_stream_1_null_string.jsonl
+++ b/tests/snapshots/mapped_stream/sourced_stream_1_null_string.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "sourced_stream_1", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "sourced_stream_1", "record": {"email": "alice@example.com", "count": 21, "user": {"id": 1, "sub": {"num": 1}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "sourced_stream_1", "record": {"email": "bob@example.com", "count": 13, "user": {"id": 2, "sub": {"num": 2}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "sourced_stream_1", "record": {"email": "charlie@example.com", "count": 19, "user": {"id": 3, "sub": {"num": 3}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
diff --git a/tests/snapshots/mapped_stream/sourced_stream_2.jsonl b/tests/snapshots/mapped_stream/sourced_stream_2.jsonl
index e32f10da..6a378a5c 100644
--- a/tests/snapshots/mapped_stream/sourced_stream_2.jsonl
+++ b/tests/snapshots/mapped_stream/sourced_stream_2.jsonl
@@ -1,7 +1,5 @@
 {"type": "SCHEMA", "stream": "sourced_stream_2", "schema": {"properties": {"email": {"type": ["string", "null"]}, "count": {"type": ["integer", "null"]}, "user": {"properties": {"id": {"type": ["integer", "null"]}, "sub": {"properties": {"num": {"type": ["integer", "null"]}}, "type": ["object", "null"]}}, "type": ["object", "null"]}}, "type": "object"}, "key_properties": []}
 {"type": "RECORD", "stream": "sourced_stream_2", "record": {"email": "alice@example.com", "count": 21, "user": {"id": 1, "sub": {"num": 1}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {"starting_replication_value": null}}}}
 {"type": "RECORD", "stream": "sourced_stream_2", "record": {"email": "bob@example.com", "count": 13, "user": {"id": 2, "sub": {"num": 2}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "RECORD", "stream": "sourced_stream_2", "record": {"email": "charlie@example.com", "count": 19, "user": {"id": 3, "sub": {"num": 3}}}, "time_extracted": "2022-01-01T00:00:00+00:00"}
 {"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}
-{"type": "STATE", "value": {"bookmarks": {"mystream": {}}}}