docs: Explain Progress is not resumable if interrupted in docs FAQ (#…

…2168)

* docs: Explain `Progress is not resumable if interrupted` in docs FAQ

* Use root package logger

* Add `pragma: no cover`

* Add test

docs/faq.md

@@ -26,3 +26,12 @@ class MyStream(Stream):
         super().__init__(tap)
         self.conn...
 ```
+
+## I'm seeing `Note: Progress is not resumable if interrupted.` in my state files
+
+If the stream attribute [`is_sorted`](singer_sdk.Stream.is_sorted) (default: `False`) is not set to `True`, the records are assumed not to be monotonically increasing, i.e. there's no guarantee that newer records always come later and the tap has to run to completion so the state can safely reflect the largest replication value seen. If you know that the records are monotonically increasing, you can set `is_sorted` to `True` and the sync will be resumable if it's interrupted at any point and the state file will reflect this.
+
+```python
+class MyStream(Stream):
+    is_sorted = True
+```

singer_sdk/helpers/_state.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import typing as t
 
 from singer_sdk.exceptions import InvalidStreamSortException
@@ -17,6 +18,8 @@
 SIGNPOST_MARKER = "replication_key_signpost"
 STARTING_MARKER = "starting_replication_value"
 
+logger = logging.getLogger("singer_sdk")
+
 
 def get_state_if_exists(
     tap_state: dict,
@@ -207,6 +210,11 @@ def increment_state(
             stream_or_partition_state[PROGRESS_MARKERS] = {
                 PROGRESS_MARKER_NOTE: "Progress is not resumable if interrupted.",
             }
+            logger.warning(
+                "Stream is assumed to be unsorted, progress is not resumable if "
+                "interrupted",
+                extra={"replication_key": replication_key},
+            )
         progress_dict = stream_or_partition_state[PROGRESS_MARKERS]
     old_rk_value = to_json_compatible(progress_dict.get("replication_key_value"))
     new_rk_value = to_json_compatible(latest_record[replication_key])

tests/core/test_state_handling.py

@@ -103,3 +103,27 @@ def test_state_finalize(dirty_state, finalized_state):
         for partition_state in stream_state.get("partitions", []):
             _state.finalize_state_progress_markers(partition_state)
     assert state == finalized_state
+
+
+def test_irresumable_state():
+    stream_state = {}
+    latest_record = {"updated_at": "2021-05-17T20:41:16Z"}
+    replication_key = "updated_at"
+    is_sorted = False
+    check_sorted = False
+
+    _state.increment_state(
+        stream_state,
+        latest_record=latest_record,
+        replication_key=replication_key,
+        is_sorted=is_sorted,
+        check_sorted=check_sorted,
+    )
+
+    assert stream_state == {
+        "progress_markers": {
+            "Note": "Progress is not resumable if interrupted.",
+            "replication_key": "updated_at",
+            "replication_key_value": "2021-05-17T20:41:16Z",
+        },
+    }