Skip to content

source-monday: simplify item backfill architecture and eliminate complex caching layers#3047

Merged
JustinASmith merged 1 commit into
mainfrom
js/source-monday-items
Jul 17, 2025
Merged

source-monday: simplify item backfill architecture and eliminate complex caching layers#3047
JustinASmith merged 1 commit into
mainfrom
js/source-monday-items

Conversation

@JustinASmith

@JustinASmith JustinASmith commented Jul 14, 2025

Copy link
Copy Markdown
Contributor

Description:

This pull request introduces changes to improve item fetching logic, optimize board processing, and enhance error handling in the source-monday API. 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_items and fetch_items_by_ids with fetch_items_by_id and introduced BoardItemIterator to handle item fetching more efficiently. This allows for better complexity control and supports backfilling items from boards consistently through restarts.

  • Updated fetch_items_changes to handle board_state_change events 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 since duplicating a board+items seems to not show Activity Log events that clearly show this occurrence.

Board processing optimizations:

  • Introduced BOARDS_PER_ITEMS_PAGE constant 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_minimal to include an items_count field and reduced the default limit from 10,000 to 500. The items_count is used in the items stream 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:

  • Changed the snapshot_resource function to sort resources by created_at or id before 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:

  • Tested on a local stack with a large account. Verified that backfilling items accurately checkpoints a ISO 8601 representing the oldest updated_at for the page of boards' items that were emitted in the fetch_page function.
  • Tested restarting the connector many times and the page_cursor being consistent and picking up where the backfill left off before the restart.
  • Also verified that the page_cursor for the items stream is continually going backwards in time since the boards are queried in created_at (DESC) order.
  • Tested that the snapshot bindings still work and data is emitted in a consistent order and that inconsistent ordering isn't showing multiple op=u and op=d records in the collection.

This change is Reviewable

@JustinASmith JustinASmith force-pushed the js/source-monday-items branch from 3a9a55a to b719e22 Compare July 15, 2025 18:21
Comment thread source-monday/source_monday/api.py
Comment thread source-monday/source_monday/api.py Outdated
Comment on lines +227 to +240
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)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +234 to +235
# 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:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread source-monday/tests/test_snapshots.py
@JustinASmith JustinASmith requested a review from Copilot July 15, 2025 18:38

This comment was marked as outdated.

@JustinASmith JustinASmith force-pushed the js/source-monday-items branch from b719e22 to ffccc22 Compare July 15, 2025 18:43
@JustinASmith JustinASmith marked this pull request as ready for review July 15, 2025 18:43
@JustinASmith JustinASmith requested a review from Copilot July 15, 2025 18:44

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 ItemCacheSession and item cache logic with a simpler BoardItemIterator class
  • Introduced new constants to control GraphQL query complexity and API rate limiting
  • Enhanced board processing to include items_count field 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

Comment thread source-monday/source_monday/graphql/items.py
Comment thread source-monday/source_monday/api.py
Comment on lines +292 to +294
board_cursor = get_board_cursor()

if not board_cursor:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@JustinASmith JustinASmith requested a review from Alex-Bair July 15, 2025 18:49

@Alex-Bair Alex-Bair left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Had a handful of non-nit comments/questions to address before approving.

Comment thread source-monday/source_monday/graphql/__init__.py
Comment thread source-monday/source_monday/api.py Outdated
Comment thread source-monday/source_monday/api.py Outdated
Comment thread source-monday/source_monday/resources.py Outdated
Comment thread source-monday/source_monday/graphql/boards.py Outdated
Comment on lines +345 to +352
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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:

  1. cutoff is 2025-06-01T12:00:00Z
  2. The fetch_boards_minimal returns:
[
  {"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"},
]
  1. The fetch_items_page backfills board 3, then returns and checkpoints.
  2. On the next fetch_items_page invocation, board 1 has been updated and fetch_boards_minimal returns:
[
  {"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?

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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_at in descending order
  • API response does NOT return created_at only updated_at

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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_at in fetch_items_page, but in fetch_items_changes yield 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_at as the page cursor, store the ids of all boards as the page cursor. Then fetch the items for a single board in fetch_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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +84 to +86
def get_cursors(self, log: Logger) -> list[str]:
"""Alias for get_result."""
return self.get_result(log)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: It doesn't look like _items_generator_started is referenced anywhere, so can it be removed?

Comment on lines +88 to +89
def get_oldest_board_updated_at(self) -> datetime | None:
return self.oldest_board_updated_at

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread source-monday/source_monday/graphql/items.py Outdated
@JustinASmith

Copy link
Copy Markdown
Contributor Author

@Alex-Bair I am marking threads as resolved. All others will be handled in a separate PR with CDK changes for the items stream. It was easier to exclude a few changes I already made that need more testing instead of cherry picking parts of what was changed in some files already.

@JustinASmith

Copy link
Copy Markdown
Contributor Author

I will also force push to keep this down to 1 commit once approved

@JustinASmith JustinASmith requested a review from Alex-Bair July 17, 2025 13:35

@Alex-Bair Alex-Bair left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM!

Comment thread source-monday/source_monday/models.py Outdated
…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.
@JustinASmith JustinASmith force-pushed the js/source-monday-items branch from a1e384c to e2d2f81 Compare July 17, 2025 14:05
@JustinASmith JustinASmith merged commit 235f3a8 into main Jul 17, 2025
97 of 103 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants