fix(ecstore): preserve active ILM source journals

crates/ecstore/src/core/pools.rs

@@ -5351,6 +5351,7 @@ impl ECStore {
         let stage = if terminal { "terminal" } else { "progress" };
         let record = validate_durable_ilm_record(path, data)
             .map_err(|err| Error::other(format!("{stage} durable ILM record is invalid at path `{path}`: {err}")))?;
+        let active_source_pool_indices = active_runs.iter().map(|(pool_idx, _)| *pool_idx).collect::<Vec<_>>();
         let mut terminal_target_pool_indices = Vec::new();
         for (source_pool_idx, run_token) in active_runs {
             let receipt_path = decommission_durable_ilm_receipt_path(&run_token, path, record.id_kind, &record.id);
@@ -5361,7 +5362,11 @@ impl ECStore {
                         .advance_durable_ilm_decommission_receipt(pool_idx, &receipt_path, &record, terminal)
                         .await?;
                     receipt_found |= found;
-                    if terminal && found && !terminal_target_pool_indices.contains(&pool_idx) {
+                    if terminal
+                        && found
+                        && !active_source_pool_indices.contains(&pool_idx)
+                        && !terminal_target_pool_indices.contains(&pool_idx)
+                    {
                         terminal_target_pool_indices.push(pool_idx);
                     }
                 }

crates/ecstore/src/store/init.rs

@@ -3287,6 +3287,10 @@ mod tests {
             .persist_decommission_durable_ilm_receipt_for_test(0, 2, &path, &record, false)
             .await
             .expect("source pool zero receipt should persist on the target");
+        store
+            .persist_decommission_durable_ilm_receipt_for_test(0, 1, &path, &record, false)
+            .await
+            .expect("source pool zero receipt may reach another active source first");
         let source_one_receipt = store
             .persist_decommission_durable_ilm_receipt_for_test(1, 2, &path, &record, false)
             .await