Make the ShardPositions model into a ShardPositionsService actor (#4156)

Its purpose is to encapsulate all of the logic used to maintained a
distributed eventually consistent view of the published shard positions
over the cluster.

From the user point of view, after instantiation
- indexing pipelines need to feed it with updates. (this happens on
  suggest_truncate). This is done by publishing
  `LocalShardPositionsUpdates` to the event broker.
- clients interested in updates can just subscript the
  `ShardPositionsUpdate` object in the event broker. The event
  received can come from a local indexing pipeline or anywhere in the
  cluster.

The service takes care of deduping/ignoring updates when necessary.

The two object (Local and not) are very similar, but different in
semantics.

Related to #4056

quickwit/quickwit-cluster/src/cluster.rs

@@ -69,7 +69,7 @@ pub struct Cluster {
     cluster_id: String,
     self_chitchat_id: ChitchatId,
     /// Socket address (UDP) the node listens on for receiving gossip messages.
-    gossip_listen_addr: SocketAddr,
+    pub gossip_listen_addr: SocketAddr,
     inner: Arc<RwLock<InnerCluster>>,
 }
 
@@ -379,7 +379,7 @@ impl Cluster {
         Ok(())
     }
 
-    async fn chitchat(&self) -> Arc<Mutex<Chitchat>> {
+    pub async fn chitchat(&self) -> Arc<Mutex<Chitchat>> {
         self.inner.read().await.chitchat_handle.chitchat()
     }
 }

quickwit/quickwit-common/src/shared_consts.rs

@@ -36,8 +36,5 @@ pub const DELETION_GRACE_PERIOD: Duration = Duration::from_secs(60 * 32); // 32
 /// being requested.
 pub const SCROLL_BATCH_LEN: usize = 1_000;
 
-/// Prefix used in chitchat to broadcast the positions of the shards assigned to an indexer.
-pub const INDEXER_ASSIGNED_SHARDS_POSITIONS_PREFIX: &str = "indexer.assigned_shards_positions:";
-
 /// Prefix used in chitchat to broadcast the list of primary shards hosted by a leader.
 pub const INGESTER_PRIMARY_SHARDS_PREFIX: &str = "ingester.primary_shards:";