feat: Cancel historical backfill process before starting anew #363
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
morgsmccauley
force-pushed
the
200-stop-historical-processing-on-indexer-redeployment
branch
from
bf90a5f
to
de1e10b
Compare
morgsmccauley
changed the base branch from
main
to
201-clean-up-sqs-code-in-coordinator-1
morgsmccauley
changed the title
morgsmccauley
commented
Nov 7, 2023
| } | ||
|
|
||
| pub(crate) async fn process_historical_messages_or_handle_error( | ||
| block_height: BlockHeight, | ||
| current_block_height: BlockHeight, |
There was a problem hiding this comment.
I got confused about what block_height this actually represents, so ended up renaming it
morgsmccauley
commented
Nov 7, 2023
| use unescape::unescape; | ||
|
|
||
| use crate::indexer_reducer; | ||
| use crate::indexer_reducer::FunctionCallInfo; | ||
| use crate::indexer_types::{IndexerFunction, IndexerRegistry}; | ||
| use indexer_rule_type::indexer_rule::{IndexerRule, IndexerRuleKind, MatchingRule, Status}; | ||
|
|
||
| pub(crate) fn registry_as_vec_of_indexer_functions( |
There was a problem hiding this comment.
Now done inline
gabehamilton
reviewed
Nov 7, 2023
| .lock() | ||
| .await | ||
| .entry(function_invocation.account_id.clone()) | ||
| { |
There was a problem hiding this comment.
Should we also clean up streams here?
There was a problem hiding this comment.
Yes, good call. I'll do that in this issue #346
gabehamilton
approved these changes
Nov 7, 2023
Base automatically changed from
201-clean-up-sqs-code-in-coordinator-1
to
main
November 7, 2023 18:13
morgsmccauley
force-pushed
the
200-stop-historical-processing-on-indexer-redeployment
branch
from
53d599c
to
b636916
Compare
morgsmccauley
deleted the
200-stop-historical-processing-on-indexer-redeployment
branch
Merged
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR updates Coordinator such that only one Historical Backfill process can exist for a given Indexer Function. Currently, these processes do not have awareness of each other, meaning a developer can trigger multiple simultaneous backfills, with them all writing to the same Redis Stream. The major changes are detailed below.
Streamer/Historical backfill tasks
In this PR, I've introduced the
historical_block_processing::Streamerstruct. This is used to encapsulate the async task which is spawned for handling the Historical Backfill. I went with "Streamer", as I envision this becoming the single process which handles both historical/real-time messages for a given Indexer, as described in #216 (comment).Streameris responsible for starting theasynctask, and provides a basic API for interacting with it. Right now, that only includes cancelling it. Cancelling aStreamerwill drop the existing future, preventing any new messages from being added, and then delete the Redis Stream removing all existing messages.Indexer Registry Mutex/Async
indexer_registryis the in-memory representation of the given Indexer Registry. Currently, we pass around theMutexGuardto mutate this data, resulting in the lock being held for much longer than is needed, blocking all other consumers from taking the lock. There isn't much contention for this lock, so it isn't a problem right now, but could end up being one in future.Adding in the
StreamerMutexmade mitigating this problem trivial, so I went ahead and made some changes. I've updated the code so that we pass around theMutexitself, rather than the guard, allowing us to the hold the lock for the least amount of time possible. This required updating most methods withinindexer_registrytoasync, so that we could take theasynclock.With the
Mutexbeing used rather than theMutexGuard, it seemed sensible to addindexer_registrytoQueryAPIContext, rather than passing it round individually.