tweaks

Author: nadav-govari

quickwit/quickwit-indexing/src/actors/indexing_service.rs

@@ -789,8 +789,7 @@ impl IndexingService {
         plan_request: ApplyIndexingPlanRequest,
         ctx: &ActorContext<Self>,
     ) -> Result<(), IndexingError> {
-        let tasks = &plan_request.indexing_tasks;
-        let pipeline_diff = self.compute_pipeline_diff(tasks);
+        let pipeline_diff = self.compute_pipeline_diff(&plan_request.indexing_tasks);
 
         if !pipeline_diff.pipelines_to_shutdown.is_empty() {
             self.shutdown_pipelines(&pipeline_diff.pipelines_to_shutdown)

quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs

@@ -1436,6 +1436,11 @@ impl MetastoreService for PostgresqlMetastore {
             .bind(&request.publish_token)
             .fetch_all(&self.connection_pool)
             .await?;
+        
+        if pg_shards.len() != request.shard_ids.len() {
+            warn_on_unacquired_shards(&request, &pg_shards);
+        }
+        
         let acquired_shards = pg_shards
             .into_iter()
             .map(|pg_shard| pg_shard.into())
@@ -2953,6 +2958,28 @@ impl PostgresqlMetastore {
     }
 }
 
+/// Best-effort diagnostics for the acquire error path: logs the shards from `request` that were not
+/// acquired — those absent from `acquired_pg_shards` because a more recent publish token owns them,
+/// or because they no longer exist. Does not touch the database.
+fn warn_on_unacquired_shards(request: &AcquireShardsRequest, acquired_pg_shards: &[PgShard]) {
+    let not_acquired_shard_ids: Vec<&ShardId> = request
+        .shard_ids
+        .iter()
+        .filter(|shard_id| {
+            !acquired_pg_shards
+                .iter()
+                .any(|pg_shard| &pg_shard.shard_id == *shard_id)
+        })
+        .collect();
+    warn!(
+        index_uid=%request.index_uid(),
+        source_id=%request.source_id,
+        shard_ids=?not_acquired_shard_ids,
+        publish_token=%request.publish_token,
+        "could not acquire shards: held by a more recent publish token, or no longer present"
+    );
+}
+
 async fn open_or_fetch_shard<'e>(
     executor: impl Executor<'e, Database = Postgres> + Clone,
     subrequest: &OpenShardSubrequest,