Please consider the following formatting changes to #12451

Framework/Core/src/CompletionPolicyHelpers.cxx

@@ -15,6 +15,7 @@
 #include "Framework/DeviceSpec.h"
 #include "Framework/CompilerBuiltins.h"
 #include "Framework/Logger.h"
+#include "Framework/TimesliceIndex.h"
 #include "Framework/TimingInfo.h"
 #include "DecongestionService.h"
 
@@ -107,10 +108,34 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAll(const char* name, Compl
 {
   auto callback = [](InputSpan const& inputs, std::vector<InputSpec> const& specs, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp {
     assert(inputs.size() == specs.size());
+
+    size_t si = 0;
+    bool missingSporadic = false;
+    bool missingTimeframe = false;
+    size_t currentTimeslice = -1;
     for (auto& input : inputs) {
-      if (input.header == nullptr) {
+      assert(si < specs.size());
+      auto& spec = specs[si++];
+      if (input.header == nullptr && spec.lifetime != Lifetime::Sporadic) {
         return CompletionPolicy::CompletionOp::Wait;
       }
+      if (input.header == nullptr && spec.lifetime == Lifetime::Sporadic) {
+        missingSporadic |= true;
+      }
+      if (input.header != nullptr && currentTimeslice == -1) {
+        auto const* dph = framework::DataRefUtils::getHeader<o2::framework::DataProcessingHeader*>(input);
+        if (dph && !TimingInfo::timesliceIsTimer(dph->startTime)) {
+          currentTimeslice = dph->startTime;
+        }
+      }
+    }
+    // If some sporadic inputs are missing, we wait for them util we are sure they will not come,
+    // i.e. until the oldest possible timeslice is beyond the timeslice of the input.
+    auto& timesliceIndex = ref.get<TimesliceIndex>();
+    auto oldestPossibleTimeslice = timesliceIndex.getOldestPossibleInput().timeslice.value;
+
+    if (missingSporadic && currentTimeslice >= oldestPossibleTimeslice) {
+      return CompletionPolicy::CompletionOp::Retry;
     }
     return CompletionPolicy::CompletionOp::Consume;
   };

Framework/Core/test/test_StaggeringWorkflow.cxx

@@ -53,7 +53,7 @@ void customize(std::vector<o2::framework::CompletionPolicy>& policies)
                         // search for spec names starting with "processor"
                         return spec.name.find("processor") == 0;
                       },
-                      [](auto const&, auto const&, auto &) { return o2::framework::CompletionPolicy::CompletionOp::Consume; }});
+                      [](auto const&, auto const&, auto&) { return o2::framework::CompletionPolicy::CompletionOp::Consume; }});
 }
 
 #include "Framework/runDataProcessing.h"

Framework/TestWorkflows/CMakeLists.txt

@@ -12,9 +12,14 @@
 o2_add_dpl_workflow(dummy-workflow
                   SOURCES src/o2DummyWorkflow.cxx
                   COMPONENT_NAME TestWorkflows)
+
 o2_add_dpl_workflow(detect-missing-timeframe
                   SOURCES test/test_DetectMissingTimeframe.cxx
                   COMPONENT_NAME TestWorkflows)
+
+o2_add_dpl_workflow(wait-until-possible
+                  SOURCES test/test_WaitUntilPossible.cxx
+                  COMPONENT_NAME TestWorkflows)
 
 o2_add_dpl_workflow(o2rootmessage-workflow
                   SOURCES "src/test_o2RootMessageWorkflow.cxx"

Framework/TestWorkflows/test/test_WaitUntilPossible.cxx

@@ -0,0 +1,82 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/DataTakingContext.h"
+#include "Framework/CompletionPolicyHelpers.h"
+#include "Framework/DeviceSpec.h"
+#include "Framework/RawDeviceService.h"
+#include "Framework/ControlService.h"
+#include "Framework/Configurable.h"
+#include "Framework/RunningWorkflowInfo.h"
+#include "Framework/RateLimiter.h"
+#include <fairmq/Device.h>
+
+#include <iostream>
+#include <chrono>
+#include <thread>
+#include <vector>
+
+using namespace o2::framework;
+
+#include "Framework/runDataProcessing.h"
+
+// This is how you can define your processing in a declarative way
+WorkflowSpec defineDataProcessing(ConfigContext const& specs)
+{
+  DataProcessorSpec a{
+    .name = "A",
+    .outputs = {OutputSpec{{"data"}, "TST", "A1", 0}},
+    .algorithm = AlgorithmSpec{adaptStateless(
+      [](DataAllocator& outputs, RawDeviceService& device, DataTakingContext& context, ProcessingContext& pcx) {
+        LOG(info) << "Data TST/A1/0 created";
+        outputs.make<int>(OutputRef{"data"}, 1);
+      })},
+  };
+  DataProcessorSpec b{
+    .name = "B",
+    .outputs = {OutputSpec{{"sporadic"}, "TST", "B1", 0, Lifetime::Sporadic}},
+    .algorithm = AlgorithmSpec{adaptStateless(
+      [](DataAllocator& outputs, RawDeviceService& device, DataTakingContext& context, ProcessingContext& pcx) {
+        // This will always be late, however since the oldest possible timeframe
+        // will be used to decide the scheduling, it will not be dropped.
+        sleep(1);
+        // We also create it only every second time, so that we can check that
+        // the sporadic output is not mandatory.
+        static int i = 0;
+        if (i++ % 2 == 0) {
+          LOG(info) << "Data TST/B1/0 created";
+          outputs.make<int>(OutputRef{"sporadic"}, 1);
+        }
+      })},
+  };
+  DataProcessorSpec d{
+    .name = "D",
+    .inputs = {InputSpec{"a1", "TST", "A1", 0, Lifetime::Timeframe},
+               InputSpec{"b1", "TST", "B1", 0, Lifetime::Sporadic}},
+    .algorithm = AlgorithmSpec{adaptStateless(
+      [](InputRecord& inputs) {
+        auto refA = inputs.get("a1");
+        auto headerA = o2::header::get<const DataProcessingHeader*>(refA.header);
+        LOG(info) << "Start time: " << headerA->startTime;
+        auto refB = inputs.get("b1");
+        if (!refB.header) {
+          LOG(info) << "No sporadic input for start time " << headerA->startTime;
+          return;
+        }
+        auto headerB = o2::header::get<const DataProcessingHeader*>(refB.header);
+        LOG(info) << "Start time: " << headerB->startTime;
+      })},
+  };
+
+  return workflow::concat(WorkflowSpec{a},
+                          WorkflowSpec{b},
+                          WorkflowSpec{d});
+}