Repository: meltano/sdk
Pull request #1164 (2022-11-09 18:59:07)

fix: Write a valid final state message at the end of each stream sync
Refer to this [slack thread](https://meltano.slack.com/archives/C01PKLU5D1R/p1668012985498759).

This PR adds a call to issue a final (valid) state message at the end of each stream sync.

In some cases, messages sent by the tap were invalid. For instance, if `state_partitioning_keys` is overridden in a stream, the stream would never issue a valid state message, making incremental syncs impossible.

Here is an example of the last message issued by `tap-github` when running on the `issue_comments` stream (which has `state_partitioning_keys` overridden). `progress_markers` should not be present in the output, and `replication_key_value` should have been promoted one level up.

```json
{"type": "STATE", "value": {"bookmarks": {
    "tempStream": {},
    "repositories": {"partitions": [{"context": {"org": "nextcloud", "repo": "server", "repo_id": 60243197}}]},
    "issue_comments": {
        "partitions": [{"context": {"org": "nextcloud", "repo": "server"}, 
        "replication_key_signpost": "2022-11-09T16:44:57.220041+00:00",
        "starting_replication_value": "2022-11-07",
        "progress_markers": {"Note": "Progress is not resumable if interrupted.", "replication_key": "updated_at", "replication_key_value": "2022-11-09T16:41:17Z"}
        }]
    }
}}}
```

There are probably some other use cases where the final state message is never sent.

I will try to add a useful test case around this, but it would also be helpful to come up with some more general tests for taps using the sdk, to validate messages they produce.

<!-- readthedocs-preview meltano-sdk start -->
----
:books: Documentation preview :books:: https://meltano-sdk--1164.org.readthedocs.build/en/1164/

<!-- readthedocs-preview meltano-sdk end -->

edgarrmondragon: Ah, there's probably some tests that tally message types that will have to be updated.
aaronsteers: @laurentS - I looked through the codebase for other references context, and yes, I think your implementation looks correct.

As @edgarrmondragon notes, this might break other tests which are counting number of state messages emitted.

I did evaluate whether this should be run at the very end of `sync_all()`, outside the streams loop, but your implementation is better (inside the loop) because it provides earlier notification to the target per stream, and because it avoids an unnecessary duplicate double-write of the state message at the end of the sync operation.

I did also evaluate if this should be calling a private member, but I think this is appropriate given that we define `sync_all()` as `@final`.

Anyway - thanks for submitting this. I'll let @edgarrmondragon take from here in regards to official review/approval, etc.
codecov[bot]: # [Codecov](https://codecov.io/gh/meltano/sdk/pull/1164?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
> Merging [#1164](https://codecov.io/gh/meltano/sdk/pull/1164?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (eb6e7a7) into [main](https://codecov.io/gh/meltano/sdk/commit/59b5b3d8f779c28b42d938092128129fab8803b5?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (59b5b3d) will **increase** coverage by `0.05%`.
> The diff coverage is `100.00%`.

```diff
@@            Coverage Diff             @@
##             main    #1164      +/-   ##
==========================================
+ Coverage   83.52%   83.57%   +0.05%     
==========================================
  Files          42       42              
  Lines        3872     3873       +1     
  Branches      657      657              
==========================================
+ Hits         3234     3237       +3     
+ Misses        474      473       -1     
+ Partials      164      163       -1     
```


| [Impacted Files](https://codecov.io/gh/meltano/sdk/pull/1164?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) | Coverage Δ | |
|---|---|---|
| [singer\_sdk/tap\_base.py](https://codecov.io/gh/meltano/sdk/pull/1164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay90YXBfYmFzZS5weQ==) | `68.66% <100.00%> (+0.14%)` | :arrow_up: |
| [singer\_sdk/target\_base.py](https://codecov.io/gh/meltano/sdk/pull/1164/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay90YXJnZXRfYmFzZS5weQ==) | `86.44% <0.00%> (+0.93%)` | :arrow_up: |

: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)

ericboucher: I might be missing something but I have one concern.

If we update the state too frequently while we are going through a stream in descending order, there is a risk that we send a state update before we actually reached our expected final "STATE".

"Random" example, we are updating issue comments on GitHub, which we can only navigate in descending order. Say our initial state is 2022-10-01, and today is 2022-11-09. We want to go through the whole stream until we reach data from 2022-10-01. Only then do we want to update the state to our new one of "2022-11-09" aka today.

If we send a state update too early, and the stream fails before actually finishing, we would have no way to know, and would then assume that we got all the data we needed until 2022-11-09. No?
aaronsteers: @ericboucher - your example is exactly the reason why the SDK decoupled _sending_ the state message from _finalizing_ the state message. 

If we've written the internal logic correctly, it should be (near) impossible to send an invalid or too-soon / too-frequent state message. Any bookmarks that aren't yet resumable should have the progress-marker tag which distinguishes between resumable and non-resumable bookmarks.
laurentS: I spent a few hours tracking state messages today :face_with_spiral_eyes: and, at least for the use case I was interested in (`tap-github`'s `issue_comments` stream, which behaves like @ericboucher describes above), the sdk seems to behave correctly (with the fix in this PR):
- while it does emit `STATE` messages as it progresses, those messages are invalid (see the example I copied at the top of this PR: the `replication_key` and `replication_key_value` are under `progress_marker`, which is the wrong place) so these have no effect
- at the very end, the call to `finalize_state_progress_markers` fixes the state object to a valid format (in memory).
- what was missing was actually sending this valid state across the line to the target. This is what the PR adds.

I can't speak for other use cases, but for this one, I think we're good :)
laurentS: > @laurentS do you feel confident writing a test for this change? It's probably OK if you don't, given the comments above.

@edgarrmondragon I looked around the sdk testing code a bit, but I feel that the most relevant test would be something like what I [outlined](https://github.com/meltano/sdk/discussions/1162#discussioncomment-4105811) in @kgpayne 's discussion.

For this PR, we would need to run `tap.sync_all()` on an unsorted stream with overridden state partitioning keys to actually test the behaviour, and verify that the last message sent is a valid `STATE` message. Maybe we can add that to `SampleTapCountries`, if you think that's useful.

Otherwise, I feel comfortable with merging as is, the worst case scenario is that taps might send a duplicate final STATE message with the same content, which should not hurt.
edgarrmondragon: > Otherwise, I feel comfortable with merging as is, the worst case scenario is that taps might send a duplicate final STATE message with the same content, which should not hurt.

I agree 👍.

Thanks @laurentS!
edgarrmondragon: @laurentS can you merge `main` into your PR branch?
laurentS: > @laurentS can you merge `main` into your PR branch?

I won't get to my computer until Tuesday at best. Feel free to take over from here if you want/can. Otherwise I'll finish it next week.

Patch:
diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py
index 6e7220ed..9a1b5ccb 100644
--- a/singer_sdk/tap_base.py
+++ b/singer_sdk/tap_base.py
@@ -380,6 +380,7 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
 
             stream.sync()
             stream.finalize_state_progress_markers()
+            stream._write_state_message()
 
         # this second loop is needed for all streams to print out their costs
         # including child streams which are otherwise skipped in the loop above


Test patch:
diff --git a/tests/core/test_countries_sync.py b/tests/core/test_countries_sync.py
index 9161e2c0..6022d000 100644
--- a/tests/core/test_countries_sync.py
+++ b/tests/core/test_countries_sync.py
@@ -130,4 +130,4 @@ def test_batch_mode(monkeypatch, outdir):
     assert counter["SCHEMA", "countries"] == 1
     assert counter["BATCH", "countries"] == 1
 
-    assert counter[("STATE",)] == 2
+    assert counter[("STATE",)] == 4