fix topology adjust corner case (#15053)

Author: aalkin

Framework/Core/src/ArrowSupport.cxx

@@ -581,7 +581,6 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
       auto spawner = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-spawner"); });
       auto analysisCCDB = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-ccdb"); });
       auto builder = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-index-builder"); });
-      auto reader = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-reader"); });
       auto writer = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-writer"); });
       auto& dec = ctx.services().get<DanglingEdgesContext>();
       dec.requestedAODs.clear();
@@ -659,6 +658,9 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         workflow.erase(writer);
       }
 
+      // removing writer would invalidate the reader iterator if it was created before
+      auto reader = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-reader"); });
+
       if (reader != workflow.end()) {
         // If reader and/or builder were adjusted, remove unneeded outputs
         // update currently requested AODs

run/o2sim_kine_publisher.cxx

@@ -13,7 +13,6 @@
 #include "Framework/AnalysisTask.h"
 #include "Monitoring/Monitoring.h"
 #include "Framework/CommonDataProcessors.h"
-#include "SimulationDataFormat/MCTrack.h"
 #include "Steer/MCKinematicsReader.h"
 
 #include "Framework/runDataProcessing.h"
@@ -64,6 +63,6 @@ WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
   spec.outputs.emplace_back("MC", "MCHEADER", 0, Lifetime::Timeframe);
   spec.outputs.emplace_back("MC", "MCTRACKS", 0, Lifetime::Timeframe);
   spec.requiredServices.push_back(o2::framework::ArrowSupport::arrowBackendSpec());
-  spec.algorithm = CommonDataProcessors::wrapWithRateLimiting(spec.algorithm);
+  spec.algorithm = CommonDataProcessors::wrapWithTimesliceConsumption(spec.algorithm);
   return {spec};
 }