fix: Coordinator Not Updating Redis Indexer Config (#398)

Coordinator pulls indexer config from registry upon initialization.
While it runs, it continues to check each block for registry update
calls. When the registry is updated, it kicks off a historical process
if necessary, and updates its local indexer configurations to reflect
the new config for that particular indexer.

It turns out that the local indexer config was in fact not correctly
updating. It was setting the wrong key for storing the config. As a
result, existing indexer updates were not being captured while new
indexers would have the wrong config. Coordinator would attempt to set
various status such as provisioning but the incorrect keys would lead to
errors around that. And, failing to capture indexer config updates leads
to runner never seeing those updates and continuing to run stale code.
Restarting coordinator resolved these issues since the config is read
freshly from registry, hiding these issues for some time, before they
would once again reappear.

In addition to the above problem, the configuration is only updated in
redis when the next matching block appears. So, if a new indexer is
published without a start block, its redis config and stream will not be
set until a new matching block appears. This can be problematic for
developers who are the producers of blocks matching their contract
filter. In that case, matching blocks may be rare, leading to a long
delay before updated code is read and executed. Setting the config
immediately after an update is seen eliminates such a problem.

I've fixed the particular line which was in correctly setting the key
wrong, as well as set the redis config immediately after an update, to
resolve the above issues.

indexer/queryapi_coordinator/src/indexer_registry.rs

@@ -176,7 +176,20 @@ async fn index_and_process_register_calls(
                         streamers_lock.insert(new_indexer_function.get_full_name(), streamer);
                     }
 
-                    fns.insert(update.method_name.clone(), new_indexer_function);
+                    storage::set(
+                        context.redis_connection_manager,
+                        storage::generate_real_time_storage_key(
+                            &new_indexer_function.get_full_name(),
+                        ),
+                        serde_json::to_string(&new_indexer_function.clone())?,
+                        None,
+                    )
+                    .await?;
+
+                    fns.insert(
+                        new_indexer_function.function_name.clone(),
+                        new_indexer_function,
+                    );
                 }
             };
         }