DPLT-1077 Process historical messages via Redis Streams #181
Conversation
morgsmccauley
changed the title
| new_indexer_function_copy, | ||
| Opts::parse(), | ||
| )) | ||
| tokio::spawn(async move { |
There was a problem hiding this comment.
Need to ownership of redis_connection_manager to the thread
| @@ -84,14 +92,6 @@ pub async fn xadd( | |||
| ) -> anyhow::Result<()> { | |||
| tracing::debug!(target: STORAGE, "XADD: {:?}, {:?}", stream_key, fields); | |||
|
|
|||
| // TODO: Remove stream cap when we finally start processing it | |||
| redis::cmd("XTRIM") | |||
There was a problem hiding this comment.
We now delete the stream messages after they are processed - so this can be removed
| @@ -1,157 +1,57 @@ | |||
| import { createClient } from 'redis'; | |||
There was a problem hiding this comment.
All Redis related actions have been extracted to RedisClient
morgsmccauley
force-pushed
the
DPLT-1077-process-historical-messages
branch
from
5ccc003
to
75b8d8b
Compare
| storage::sadd( | ||
| redis_connection_manager, | ||
| storage::STREAMS_SET_KEY, | ||
| storage::generate_historical_stream_key(&indexer_function.get_full_name()), |
There was a problem hiding this comment.
Should we have the ability to distinguish between different historical processes?
Won't using the indexer function's name cause the old historical processes's with the same name to be overwritten?
There was a problem hiding this comment.
Yes, and that's intentional. We don't want to have multiple concurrent historical processes for a given indexer. I'll be working on stopping existing processes soon, which is why I didn't bother creating unique streams.
This PR write historical messages to a dedicated Redis Stream per indexer. Redis keys have been slightly restructured to make it easier to distinguish which streams need to be monitored, mainly:
morgs.near/test, we store the stream name, as this indexer could potentially have two streams, historical and real-time.storage, which stores the indexer config, is now suffixed under the stream name, so we can have separate storage for both historical and real-time, preventing race conflicts between themThe last point is a bit of a weird one, but I am trying to avoid the scenario where we change the historical indexers code out from underneath it. This would be the case if the developer changes their indexer code while a historical process is still running. Keeping the indexer code consistent throughout a given historical backfill seems most correct.
It would be possible for a developer to start another historical process, while an existing one is ongoing, causing the indexer code to change, as historical processes for a given indexer share the same storage key. But I don't think it's worth actively preventing this, as our ideal state is to stop the existing historical backfill when another is kicked off.
Once an historical process has been kicked off, Runner will always start a process to listen to the stream, there's nothing telling it to stop listening. I may address this in a future PR, or just wait until we have control messages :).