source-monday: simplify item backfill architecture and eliminate complex caching layers#3047
Conversation
3a9a55a to
b719e22
Compare
| elif "board_state_changed" in activity_log.event: | ||
| # When a board is changing state, it can potentially be moved from the trash (deleted) | ||
| # to active or archived state, so we need to backfill items from that board since the | ||
| # activity log does not show item updates and the backfill might not backfill the items | ||
| # if the board was already deleted before the capture started. | ||
| board_ids_to_backfill.add(activity_log.resource_id) |
There was a problem hiding this comment.
A board can be deleted when a capture is created and the user can restore the board within 30 days. If this happens, the Activity Log indicates a board_state_changed event, but no item events since the items were not updated. However, we never backfilled the items, so we need to fetch all items for this board as long as the state change isn't from active or archived to deleted.
| # TODO(justin): handle other events that might require fetching items (e.g., duplicating a board with items) | ||
| elif "board_state_change" in activity_log.event: |
There was a problem hiding this comment.
The Monday.com API docs do not document any of the Activity Log events. So I have had to discover this through testing and use of their API. This TODO may need to be removed or just left as a note, since right now I am unsure a straight forward way to know a board was duplicated. In testing, when I duplicate a board with all items it shows events that don't mention duplication or anything related. My only idea right now is to maybe key of some create type event and determine if a board was just created and then just trying to fetch it's items in case it was a duplicated board+items.
b719e22 to
ffccc22
Compare
There was a problem hiding this comment.
Pull Request Overview
This pull request refactors the Monday.com connector's item fetching architecture by removing complex caching layers and simplifying the backfill process. The changes improve maintainability while ensuring consistent item retrieval through restarts.
Key changes:
- Replaced complex
ItemCacheSessionand item cache logic with a simplerBoardItemIteratorclass - Introduced new constants to control GraphQL query complexity and API rate limiting
- Enhanced board processing to include
items_countfield and improved filtering logic
Reviewed Changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| source-monday/tests/test_snapshots.py | Increased test delay from 60s to 250s |
| source-monday/source_monday/resources.py | Removed item cache session dependencies and simplified resource opening |
| source-monday/source_monday/models.py | Removed complex type definitions and added items_count field to Board model |
| source-monday/source_monday/graphql/items/items.py | Deleted original items module containing complex caching logic |
| source-monday/source_monday/graphql/items/item_cache.py | Deleted item cache implementation |
| source-monday/source_monday/graphql/items/init.py | Removed module exports |
| source-monday/source_monday/graphql/items.py | Created new simplified items module with BoardItemIterator |
| source-monday/source_monday/graphql/boards.py | Added items_count field and improved pagination logic |
| source-monday/source_monday/graphql/init.py | Updated exports to use new item fetching functions |
| source-monday/source_monday/api.py | Refactored item fetching logic and added board state change handling |
| board_cursor = get_board_cursor() | ||
|
|
||
| if not board_cursor: |
There was a problem hiding this comment.
The while True: loop at first glance may be confusing. Note, though, the get_board_cursor() can only be called after the items generator is processed since under the hood this is using the IncrementalJsonProcessor and remainder processing to pull out the necessary information after all items are streamed from the response first.
Alex-Bair
left a comment
There was a problem hiding this comment.
Had a handful of non-nit comments/questions to address before approving.
| async for board in fetch_boards_minimal(http, log): | ||
| if board.updated_at <= page_cutoff and board.state != "deleted": | ||
| qualifying_board_ids.add(board.id) | ||
| if board.items_count is not None: | ||
| max_items_count = max(max_items_count, board.items_count) | ||
|
|
||
| if len(qualifying_board_ids) >= BOARDS_PER_ITEMS_PAGE: | ||
| break |
There was a problem hiding this comment.
How does this behave when an old board is updated before the items backfill has fetched items for that board? I'm thinking of a scenario like:
cutoffis2025-06-01T12:00:00Z- The
fetch_boards_minimalreturns:
[
{"id": 3, "created_at": "2025-06-01T11:59:00Z", "updated_at": "2025-06-01T11:59:00Z"},
{"id": 2, "created_at": "2025-06-01T11:58:00Z", "updated_at": "2025-06-01T11:58:00Z"},
{"id": 1, "created_at": "2025-06-01T11:57:00Z", "updated_at": "2025-06-01T11:57:00Z"},
]- The
fetch_items_pagebackfills board 3, then returns and checkpoints. - On the next
fetch_items_pageinvocation, board 1 has been updated andfetch_boards_minimalreturns:
[
{"id": 3, "created_at": "2025-06-01T11:59:00Z", "updated_at": "2025-06-01T11:59:00Z"},
{"id": 2, "created_at": "2025-06-01T11:58:00Z", "updated_at": "2025-06-01T11:58:00Z"},
{"id": 1, "created_at": "2025-06-01T11:57:00Z", "updated_at": "2025-06-01T12:01:00Z"},
]It looks like with this filtering logic, board 1's items would be missed in the backfill. Is that a valid concern?
There was a problem hiding this comment.
If so, I'd suggest using the board's created_at instead of updated_at for the backfill cursor to avoid this kind of race condition.
We'd also need to process boards in descending order of their created_at to make sure we don't skip a board. Since Monday's API returns boards in descending order of their created_at, I suspect this is already handled, but it would be good to confirm the item_iterator.get_items_from_boards doesn't shuffle the board ids around somehow.
There was a problem hiding this comment.
Hmm. I am thinking about this still, but a couple things to note about this that are challenging:
- API only allows a query that filters by
created_atin descending order - API response does NOT return
created_atonlyupdated_at
There was a problem hiding this comment.
Wow, I didn't realize Monday doesn't let us query created_at for boards. That's frustrating.
Other ideas off the top of my head that might work:
- Keep the logic you have with
updated_atinfetch_items_page, but infetch_items_changesyield all of a board's items whenever we see a board is updated. (I think there's a reason this wouldn't work from what you've mentioned before, but I don't remember exactly what that is. Maybe something about some item updates not updating the parent board?). - Instead of using
updated_atas the page cursor, store the ids of all boards as the page cursor. Then fetch the items for a single board infetch_items_page, remove that board id from the page cursor, then yield the updated list of board ids.
There are likely other approaches too, but those were ones I thought of right now.
There was a problem hiding this comment.
The first approach "could" work but the fetching/pagination of board items is costly and pretty slow in general. Items being updated/added to update the parent board updated_at too, but the parent board updated_at can be changed for other reasons too, so I would expect a lot of false positives.
I like the second approach. Is it safe to say there is not currently a limit on the PageCursor string length/format built into the CDK? I will look myself too, but was just thinking, if we have 100k+ boards that would be a large list of IDs for a cursor that we'd parse each time since the PageCursor can only be str | int | None and we'd have to have it as comma separated string for simplicity. I could probably make it an integer and use bit manipulation to extract the next board or set of boards for a page, but will keep it simple (CSV based)
There was a problem hiding this comment.
I'm not aware of limits we impose on PageCursor about the string length or format. There are checks we perform on subsequent LogCursors (are they the same type & are they increasing?), but there are no such checks on PageCursor so I think the CDK would support a comma delimited string of board ids for a PageCursor.
You're right it could be a pretty large string if there are a lot of boards, and I agree with keeping it simple. It's not the ideal approach, but it feels like we have to get creative with the API restrictions Monday.com enforces.
It may be worthwhile getting @williamhbaker's thoughts & if he sees a better approach.
There was a problem hiding this comment.
The thing to watch out for with using a really long string as a cursor like that is that it will be the "value" in a JSON key/value map...every time the connector commits a checkpoint, this gets written to the recovery log, so if it is committing checkpoints with any kind of high frequency during the backfill then every time that huge string needs to be written to the recovery log, and it seems like it would cause very large recovery logs which are problematic.
Ideally such a checkpoint would be structured such that each board ID was the key to a map (which a dummy boolean value or somethign), and you would use JSON merge patches to clear out IDs that are completed. A connector checkpoint can be very large itself without much downside, but the checkpoint updates need to be kept small and applied with JSON merge patches to manage recovery log size.
There was a problem hiding this comment.
I will address this in a separate PR since it is CDK level changes and needs more testing. I will have a separate commit though addressing other items.
| def get_cursors(self, log: Logger) -> list[str]: | ||
| """Alias for get_result.""" | ||
| return self.get_result(log) |
There was a problem hiding this comment.
nit: It doesn't look like get_cursors is used anywhere. Can it be removed?
| ): | ||
| self.http = http | ||
| self.log = log | ||
| self._items_generator_started = False |
There was a problem hiding this comment.
nit: It doesn't look like _items_generator_started is referenced anywhere, so can it be removed?
| def get_oldest_board_updated_at(self) -> datetime | None: | ||
| return self.oldest_board_updated_at |
There was a problem hiding this comment.
nit: oldest_board_updated_at is already a public property, so I'd suggest accessing it directly instead of having a get_oldest_board_updated_at getter.
|
@Alex-Bair I am marking threads as resolved. All others will be handled in a separate PR with CDK changes for the |
|
I will also force push to keep this down to 1 commit once approved |
…lex caching layers Replaced the complex multi-class item caching system with a single-class iterator approach that maintains complete board coverage while reducing code complexity. Key improvements: - Consolidated 3 classes (ItemCacheEntry, ItemIdCache, ItemCacheSession) into 1 simple BoardItemIterator class - Eliminated two-phase caching mechanism (cache item IDs → fetch full items) in favor of direct streaming - Reduced item processing code from ~240 lines across 3 files to ~123 lines in single file (48% reduction) - Simplified API from complex process_page() → populate_next_board_chunk() → stream_items_from_cache() chain to single get_items_from_boards() method - Removed unnecessary state tracking: _current_item_index, _items deque, _board_iterator, cached board lists - Maintained essential functionality: processed board tracking prevents re-processing, complete non-deleted board coverage - Added checks in items incremental task for activity log events that indicate all items for a board should be fetched The simplified implementation eliminates potential data consistency issues from the complex caching layer while ensuring every non-deleted board gets its items fully processed during backfill. Incremental sync via activity logs continues to handle post-backfill changes efficiently.
a1e384c to
e2d2f81
Compare
Description:
This pull request introduces changes to improve item fetching logic, optimize board processing, and enhance error handling in the
source-mondayAPI. The updates include replacing deprecated methods, introducing new iterators for fetching items, refining board filtering, and improving sorting and error logging mechanisms.Item fetching improvements:
Replaced
fetch_itemsandfetch_items_by_idswithfetch_items_by_idand introducedBoardItemIteratorto handle item fetching more efficiently. This allows for better complexity control and supports backfilling items from boards consistently through restarts.Updated
fetch_items_changesto handleboard_state_changeevents by backfilling items from boards transitioning out of the deleted state. Added logic to pre-filter boards before fetching items to avoid errors caused by deleted boards. There is still a TODO item sinceduplicatinga board+items seems to not show Activity Log events that clearly show this occurrence.Board processing optimizations:
Introduced
BOARDS_PER_ITEMS_PAGEconstant to limit the number of boards processed in a single fetch operation, balancing GraphQL query complexity and API constraints. (source-monday/source_monday/api.py)Modified
fetch_boards_minimalto include anitems_countfield and reduced the default limit from 10,000 to 500. Theitems_countis used in theitemsstream backfill to control the items limit per board in the query. This is useful if the batch of boards we are working on has a bunch of boards with very few items, allowing us to reduce the query complexity dynamically.Sorting of snapshot binding documents:
snapshot_resourcefunction to sort resources bycreated_atoridbefore yielding, ensuring consistent ordering.Workflow steps:
(How does one use this feature, and how has it changed)
Documentation links affected:
(list any documentation links that you created, or existing ones that you've identified as needing updates, along with a brief description)
Notes for reviewers:
updated_atfor the page of boards' items that were emitted in thefetch_pagefunction.page_cursorbeing consistent and picking up where the backfill left off before the restart.page_cursorfor theitemsstream is continually going backwards in time since the boards are queried increated_at (DESC)order.op=uandop=drecords in the collection.This change is