FIX Assign shard logic

[fulmicoton]

We need a way for the indexer node (this is done in the IndexingService) to apply the indexing plan that is assigned to us by the control plane.

For the new ingest v2 indexing pipelines, we added a concept of shards.

One pipeline is now source_uid + set of shards.

The MultiFetchStream offers the flexibility of making it possible to add new shards dynamically (without restarting the pipeline).

Shards that are removed, but have reached EOF can also be removed safely.
(ask Adrien: right now when a stream has reached EOF, do we need to do something in the multi fetch stream?).

A quick and dirty solution could be :

• out of the existing pipelines, identify the one matching the given list of shards the best.
• send the Indexing pipeline assign shards.
• assign shard updates the state of the indexing pipeline params (this is not done today).
  The point is that we want the list of shards to be remember if the pipeline has to be respawned due to regular supervision, and also we need a state to expose through chitchat.
• We update the running pipeline without restarting it:
  If there was only shard additions and EOF shard deletion
• the source received a AddShards message and subscribes to the missing shards.
• if the published position was EOF nothing needs to be done (need check from Adrien)

If there is a removal of a shard that has not reached EOF. As a temporary fix, we respawn the entire pipeline.
With the current scheduling logic, this should be rare, but it can happen when the load associated to shards has increased. That temporary fix could still yield to a publish error if the race condition gods are against us:

• new shard owner acquires the publish locks and starts indexing before old shard owner 2 have received the memo, and old shard owner attempts to publish.

Ideally we would like to make it entirely impossible and have the control plane close/reopen shard when that happens, but keep the shard assigned to the same pipeline until it reaches EOF.

---

Existing bug today

• fails if it happens in the middle of a pipeline respawn. (handle_opt is none)
• in presence of a change in the list of shards, a pipeline respawn does not know the list of shards (unsure)
• in presence of a change in the list of shards, the pipeline is not restarted. The problem is complicated, but I suspect we end up reindexing from the last commit for the shard we used to index, and would hit the publish lock in the likely presence of a conflict. (unsure)