update bookmarking for interrupted sync

[RushiT0122]

Description of change

Consider we have following records to sync from the source,

(order_id | replication_key):

• (1, 101),
• (2, 102),
• (3, 103),
• (4, 104),
• (5, 105),
• (6, 106),
• (7, 107),
• (8, 108),
• (7, 109), (9, 109), --> order_id#7 gets updated here
• (10, 110)

Before sync starts,

• Sync starts at: 110
• Window size: 5
• Bookmark=100
• First window: 100-105

Scenario

• Sync interrupted at: since_id=8 (replication_key=108)
• Next sync starts at: 115
• Record updated before next sync: order_id=7 (replication_key=109)

On completion of next sync,

• since_id = 7
• Windows: (100-105), (105-110), (110-115)
• Bookmark = min(max_bookmark_value, stop_time) = 110

In this scenario, on resuming sync updated record id#7 with (replication_key=109) will not sync with current implementation.

Solution:

Add updated_at_max value along with since_id and on resuming interrupted sync set window between (bookmark, updated_at_max). 

• Sync interrupted at: since_id=8 (replication_key=108)
• Next sync starts at: 115
• Record updated before next sync: order_id=7 (replication_key=109)

On completion of next sync,

• since_id = 1  --> pop() the since_id=8 from the bookmark
• Starting window: (105-108), (108-113), (113-115)
• Bookmark = min(max_bookmark_value, stop_time) = 110

This solution will make sure on resuming the interrupted sync, record id#7 (replication_key=109) will be synced.

QA steps

• automated tests passing
• manual qa steps passing (list below)

Risks

Rollback steps

• revert this branch

[luandy64]

The code changes look okay, but the tests could be improved I think

[luandy64]

Style suggestion:
https://pep8.org/#break-before-or-after-binary-operator

Suggested change

	updated_at_max = last_sync_interrupted_at or updated_at_min + \
	datetime.timedelta(days=date_window_size)
	updated_at_max = (last_sync_interrupted_at
	or updated_at_min + datetime.timedelta(days=date_window_size))

The preview doesn't look great, but the line is 91 characters long, so it should be fine

[luandy64]

Why is this return value set here, but some of the other patches are set above?

[RushiT0122]

Pattern was duplicated from the existing test for conistency, will refactor the code at both the places.

[luandy64]

My opinion on consistency is only copy it when it's good. Otherwise, we should improve the code when we touch it

[luandy64]

Seems like we don't use customers after this. Can this just be

CUSTOMER_OBJECT.get_objects()

[luandy64]

Same question as before

[RushiT0122]

Fixed it.

[luandy64]

It feels like we should catch our dummy exception because in the current implementation, if an exception is thrown or not, we handle it the same.

How do we distinguish a syntax error in the test from the dummy exception, for example?

[RushiT0122]

Added dummy exception to distinguish dummy exception from code exception and used it as a side effect.

[luandy64]

This test seems poorly implemented.

1. We set the state
2. We mock all of the code that would alter the state
3. We assert the state is what we expect it to be

Shouldn't we start with an empty state, mock the API call so that state looks like

{"bookmarks": {
            "currently_sync_stream": 'customers',
            "customers": {
                "updated_at": "2021-03-27T00:00:00.000000Z",
                "since_id": 15,
                "updated_at_max": "2021-04-26T00:00:00.000000Z"}}}

then force the tap to throw an exception, and then assert the state has a since_id and an updated_at_max?

[RushiT0122]

In the code implementation, we are poping since_id and updated_at_max. This test is verifying that we are not poping these intermediate bookmarks too early before we successfully sync till updated_at_max in the bookmark. Early deletion of intermitent bookmarks will cause the undesired results in case sync is interrupted.

[luandy64]

Just one style change. It's up to you if you want to include it