QC-1132 Correct validity in PostProcessingTasks across run restart (#2190)

This makes sure that the SOR&EOR timestamps are retrieved from ECS and passed to the user task with UserOrControl trigger, but they are not used to establish the validity of produced objects.
Also, we do not publish objects if the validity is going to be invalid, since they will not be stored anyway.
We produce appropriate warnings in such case.

The root cause for having EOR timestamp earlier than SOR for the 2nd run was discovered to come from ECS sending incorrect values.
However, with this commit, these values will not be used anyway for determining objects validity.

Author: knopers8

Framework/include/QualityControl/PostProcessingDevice.h

@@ -70,7 +70,7 @@ class PostProcessingDevice : public framework::Task
   /// \brief Callback for CallbackService::Id::Start (DPL) a.k.a. RUN transition (FairMQ)
   void start(framework::ServiceRegistryRef services);
   /// \brief Callback for CallbackService::Id::Stop (DPL) a.k.a. STOP transition (FairMQ)
-  void stop() override;
+  void stop(framework::ServiceRegistryRef services);
   /// \brief Callback for CallbackService::Id::Reset (DPL) a.k.a. RESET DEVICE transition (FairMQ)
   void reset();
 

Framework/include/QualityControl/PostProcessingRunner.h

@@ -74,7 +74,7 @@ class PostProcessingRunner
   /// \brief Start transition. Throws on errors.
   void start(framework::ServiceRegistryRef dplServices);
   /// \brief Stop transition. Throws on errors.
-  void stop();
+  void stop(framework::ServiceRegistryRef dplServices);
   /// \brief Reset transition. Throws on errors.
   void reset();
   /// \brief Runs the task over selected timestamps, performing the full start, run, stop cycle.

Framework/src/PostProcessingDevice.cxx

@@ -59,7 +59,7 @@ void PostProcessingDevice::init(framework::InitContext& ctx)
   // registering state machine callbacks
   ctx.services().get<CallbackService>().set<CallbackService::Id::Start>([this, services = ctx.services()]() mutable { start(services); });
   ctx.services().get<CallbackService>().set<CallbackService::Id::Reset>([this]() { reset(); });
-  ctx.services().get<CallbackService>().set<CallbackService::Id::Stop>([this]() { stop(); });
+  ctx.services().get<CallbackService>().set<CallbackService::Id::Stop>([this, services = ctx.services()]() mutable { stop(services); });
 }
 
 void PostProcessingDevice::run(framework::ProcessingContext& ctx)
@@ -113,9 +113,9 @@ void PostProcessingDevice::start(ServiceRegistryRef services)
   mRunner->start(services);
 }
 
-void PostProcessingDevice::stop()
+void PostProcessingDevice::stop(ServiceRegistryRef services)
 {
-  mRunner->stop();
+  mRunner->stop(services);
 }
 
 void PostProcessingDevice::reset()

Framework/src/PostProcessingRunner.cxx

@@ -168,20 +168,21 @@ void PostProcessingRunner::runOverTimestamps(const std::vector<uint64_t>& timest
 
 void PostProcessingRunner::start(framework::ServiceRegistryRef dplServices)
 {
+  Activity activityFromDriver = mTaskConfig.activity;
+  activityFromDriver.mValidity.setMin(getCurrentTimestamp());
   if (dplServices.active<framework::RawDeviceService>()) {
-    mTaskConfig.activity = computeActivity(dplServices, mTaskConfig.activity);
+    activityFromDriver = computeActivity(dplServices, activityFromDriver);
     QcInfoLogger::setPartition(mTaskConfig.activity.mPartitionName);
   }
-  mActivity = mTaskConfig.activity;
-  mActivity.mValidity = gInvalidValidityInterval;
-  QcInfoLogger::setRun(mTaskConfig.activity.mId);
-  mObjectManager->setActivity(mActivity);
+  mActivity = activityFromDriver;
+  mActivity.mValidity = gInvalidValidityInterval; // object validity shall be based on input objects, not run duration
+  QcInfoLogger::setRun(mActivity.mId);
 
   // register ourselves to the BK
   if (gSystem->Getenv("O2_QC_REGISTER_IN_BK")) { // until we are sure it works, we have to turn it on
     ILOG(Debug, Devel) << "Registering pp task to BookKeeping" << ENDM;
     try {
-      Bookkeeping::getInstance().registerProcess(mTaskConfig.activity.mId, mRunnerConfig.taskName, mRunnerConfig.detectorName, bookkeeping::DPL_PROCESS_TYPE_QC_POSTPROCESSING, "");
+      Bookkeeping::getInstance().registerProcess(mActivity.mId, mRunnerConfig.taskName, mRunnerConfig.detectorName, bookkeeping::DPL_PROCESS_TYPE_QC_POSTPROCESSING, "");
     } catch (std::runtime_error& error) {
       ILOG(Warning, Devel) << "Failed registration to the BookKeeping: " << error.what() << ENDM;
     }
@@ -190,7 +191,7 @@ void PostProcessingRunner::start(framework::ServiceRegistryRef dplServices)
   if (mTaskState == TaskState::Created || mTaskState == TaskState::Finished) {
     mInitTriggers = trigger_helpers::createTriggers(mTaskConfig.initTriggers, mTaskConfig);
     if (trigger_helpers::hasUserOrControlTrigger(mTaskConfig.initTriggers)) {
-      doInitialize({ TriggerType::UserOrControl, false, mTaskConfig.activity });
+      doInitialize({ TriggerType::UserOrControl, false, activityFromDriver, activityFromDriver.mValidity.getMin() });
     }
   } else if (mTaskState == TaskState::Running) {
     ILOG(Debug, Devel) << "Requested start, but the user task is already running - doing nothing." << ENDM;
@@ -201,11 +202,17 @@ void PostProcessingRunner::start(framework::ServiceRegistryRef dplServices)
   }
 }
 
-void PostProcessingRunner::stop()
+void PostProcessingRunner::stop(framework::ServiceRegistryRef dplServices)
 {
   if (mTaskState == TaskState::Created || mTaskState == TaskState::Running) {
     if (trigger_helpers::hasUserOrControlTrigger(mTaskConfig.stopTriggers)) {
-      doFinalize({ TriggerType::UserOrControl, false, mTaskConfig.activity });
+      // We try to get SOR and EOR times for ECS, which could be needed by the user code.
+      auto activityFromDriver = mActivity;
+      activityFromDriver.mValidity.setMax(getCurrentTimestamp());
+      if (dplServices.active<framework::RawDeviceService>()) {
+        activityFromDriver = computeActivity(dplServices, activityFromDriver);
+      }
+      doFinalize({ TriggerType::UserOrControl, false, activityFromDriver, activityFromDriver.mValidity.getMax() });
     }
   } else if (mTaskState == TaskState::Finished) {
     ILOG(Debug, Devel) << "Requested stop, but the user task is already finalized - doing nothing." << ENDM;
@@ -232,6 +239,13 @@ void PostProcessingRunner::reset()
 
 void PostProcessingRunner::updateValidity(const Trigger& trigger)
 {
+  if (trigger == TriggerType::UserOrControl) {
+    // we ignore it, because it would not make sense to use current time in tracking objects from the past,
+    // especially in asynchronous postprocessing
+    ILOG(Debug, Trace) << "Ignoring UserOrControl trigger in tracking objects validity" << ENDM;
+    mObjectManager->setValidity(mActivity.mValidity);
+    return;
+  }
   if (!trigger.activity.mValidity.isValid()) {
     ILOG(Warning, Devel) << "Not updating objects validity, because the provided trigger validity is invalid ("
                          << trigger.activity.mValidity.getMin() << ", " << trigger.activity.mValidity.getMax() << ")" << ENDM;
@@ -270,7 +284,11 @@ void PostProcessingRunner::doUpdate(const Trigger& trigger)
   mTask->update(trigger, mServices);
   updateValidity(trigger);
 
-  mPublicationCallback(mObjectManager->getNonOwningArray());
+  if (mActivity.mValidity.isValid()) {
+    mPublicationCallback(mObjectManager->getNonOwningArray());
+  } else {
+    ILOG(Warning, Support) << "Objects will not be published because their validity is invalid. This should not happen." << ENDM;
+  }
 }
 
 void PostProcessingRunner::doFinalize(const Trigger& trigger)
@@ -283,7 +301,12 @@ void PostProcessingRunner::doFinalize(const Trigger& trigger)
   mTask->finalize(trigger, mServices);
   updateValidity(trigger);
 
-  mPublicationCallback(mObjectManager->getNonOwningArray());
+  if (mActivity.mValidity.isValid()) {
+    mPublicationCallback(mObjectManager->getNonOwningArray());
+  } else {
+    // TODO: we could consider using SOR, EOR as validity in such case, so empty objects are still stored in the QCDB.
+    ILOG(Warning, Devel) << "Objects will not be published because their validity is invalid. Most likely the task's update() method was never triggered." << ENDM;
+  }
   mTaskState = TaskState::Finished;
   mObjectManager->stopPublishingAll();
 }

Framework/src/runPostProcessing.cxx

@@ -78,12 +78,12 @@ int main(int argc, const char* argv[])
 
     PostProcessingRunner runner(taskID);
     runner.init(configTree, WorkflowType::Standalone);
+    ServiceRegistry registry;
 
     if (vm.count("timestamps")) {
       // running the PP task on a set of timestamps
       runner.runOverTimestamps(vm["timestamps"].as<std::vector<uint64_t>>());
     } else {
-      ServiceRegistry registry;
       // running the PP task with an event loop
       runner.start({ registry });
 
@@ -96,7 +96,7 @@ int main(int argc, const char* argv[])
         usleep(1000000.0 * timer.getRemainingTime());
       }
     }
-    runner.stop();
+    runner.stop({ registry });
     return 0;
   } catch (const bpo::error& ex) {
     ILOG(Error, Ops) << "Exception caught: " << ex.what() << ENDM;

Framework/src/runPostProcessingOCC.cxx

@@ -118,7 +118,8 @@ class PostProcessingOCCStateMachine : public RuntimeControlledObject
 
     bool success = true;
     try {
-      mRunner->stop();
+      o2::framework::ServiceRegistry registry;
+      mRunner->stop(registry);
     } catch (const std::exception& ex) {
       ILOG(Error, Support) << "Exception caught: " << ex.what() << ENDM;
       success = false;

Framework/src/runnerUtils.cxx

@@ -65,8 +65,8 @@ Activity computeActivity(framework::ServiceRegistryRef services, const Activity&
 {
   auto runNumber = computeActivityField<int>(services, "runNumber", fallbackActivity.mId);
   auto runType = computeActivityField<int>(services, "runType", fallbackActivity.mType);
-  auto run_start_time_ms = computeActivityField<unsigned long>(services, "runStartTimeMs", fallbackActivity.mValidity.getMin());
-  auto run_stop_time_ms = computeActivityField<unsigned long>(services, "runEndTimeMs", fallbackActivity.mValidity.getMax());
+  auto runStartTimeMs = computeActivityField<unsigned long>(services, "runStartTimeMs", fallbackActivity.mValidity.getMin());
+  auto runEndTimeMs = computeActivityField<unsigned long>(services, "runEndTimeMs", fallbackActivity.mValidity.getMax());
   auto partitionName = services.get<framework::RawDeviceService>().device()->fConfig->GetProperty<std::string>("environment_id", fallbackActivity.mPartitionName);
   auto periodName = services.get<framework::RawDeviceService>().device()->fConfig->GetProperty<std::string>("lhcPeriod", "");
   if (periodName.empty()) {
@@ -82,7 +82,7 @@ Activity computeActivity(framework::ServiceRegistryRef services, const Activity&
     periodName,
     fallbackActivity.mPassName,
     fallbackActivity.mProvenance,
-    { run_start_time_ms, run_stop_time_ms },
+    { runStartTimeMs, runEndTimeMs },
     beam_type,
     partitionName,
     fillNumber);