From 3314c5349fae9e2f0ab305c3291e18baa7116c8e Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Tue, 7 Oct 2025 12:57:56 +0200
Subject: [PATCH 01/14] Delayed postprocessing timeout logic

---
 .../models/DataHandlingModel.hpp              |  6 +++
 .../models/detail/DataHandlingModel.hxx       | 44 ++++++++++++++-----
 2 files changed, 38 insertions(+), 12 deletions(-)

diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index 2fb6d1a..114f90b 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -163,6 +163,12 @@ class DataHandlingModel : public DataHandlingConcept
     return { reinterpret_cast<RDT&>(original) };
   }
 
+  // Actions postprocess scheduler takes if no data arrives in a configured time
+  virtual void invoke_postprocess_schedule_timeout_policy() const
+  {
+    return; // No-op for this class
+  }    
+
   // Operational monitoring
   virtual void generate_opmon_data() override;
 
diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 3841506..6bf88fa 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -322,10 +322,15 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {
   auto now = last_post_proc_time;
   std::chrono::milliseconds milliseconds;
   RDT processed_element;
+  int consecutive_timeouts = 0;
+  const timestamp_t max_wait_in_ticks = m_post_processing_delay_max_wait * 62500;  
 
   // Deferral of the post processing, to allow elements being reordered in the LB
   // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value  
   while (m_run_marker.load()) {
+    bool timeout = false;
+    bool postprocess = false;
+
     try {
       co_await folly::coro::timeout(
         m_baton.operator co_await(),
@@ -333,22 +338,13 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {
         m_timekeeper.get());
       m_baton.reset();
     } catch (const folly::FutureTimeout&) {
-      ++m_num_post_processing_delay_max_waits;
+      timeout = true;
     }
 
     if (m_latency_buffer_impl->occupancy() == 0) {
       continue;
     }
 
-    now = std::chrono::system_clock::now();
-    milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_post_proc_time);
-
-    if (milliseconds.count() <= m_post_processing_delay_min_wait) {
-      continue;
-    }
-
-    last_post_proc_time = now;
-
     // Get the LB boundaries
     auto tail = m_latency_buffer_impl->back();
     newest_ts = tail->get_timestamp();
@@ -360,8 +356,26 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {
       TLOG() << "***** First pass post processing *****";
     }
 
-    if (newest_ts - processed_element.get_timestamp() > m_processing_delay_ticks) {
-      end_win_ts = newest_ts - m_processing_delay_ticks;
+    now = std::chrono::system_clock::now();
+
+    if (timeout) {      
+      ++m_num_post_processing_delay_max_waits;
+      ++consecutive_timeouts;
+      const timestamp_t timeout_accumulated = consecutive_timeouts * max_wait_in_ticks;      
+      end_win_ts = newest_ts - m_processing_delay_ticks + timeout_accumulated;
+      postprocess = true;
+    } else {
+      consecutive_timeouts = 0;
+      milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_post_proc_time);
+      if (milliseconds.count() > m_post_processing_delay_min_wait) {
+        if (newest_ts - processed_element.get_timestamp() > m_processing_delay_ticks) {
+          end_win_ts = newest_ts - m_processing_delay_ticks;
+          postprocess = true;
+        }
+      }
+    }
+
+    if (postprocess) {
       auto start_iter = m_latency_buffer_impl->lower_bound(processed_element, false);
       processed_element.set_timestamp(end_win_ts);
       auto end_iter = m_latency_buffer_impl->lower_bound(processed_element, false);
@@ -372,6 +386,12 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {
         ++m_sum_payloads;
         ++m_stats_packet_count;
       }
+
+      last_post_proc_time = now;
+
+      if (timeout) {
+        invoke_postprocess_schedule_timeout_policy();
+      }
     }
   }
 }

From ad8c6ec9f764b5f39ac92504a18a07326f6a5b5c Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Fri, 10 Oct 2025 16:52:18 +0200
Subject: [PATCH 02/14] Nested class PostprocessManager with unit test

---
 CMakeLists.txt                                |  1 +
 include/datahandlinglibs/ReadoutTypes.hpp     | 13 +++
 .../models/DataHandlingModel.hpp              | 96 +++++++++++++++++++
 .../models/detail/DataHandlingModel.hxx       | 70 ++------------
 .../testutils/UnitTestUtilities.hpp           | 32 +++++++
 ...atahandlinglibs_DataHandlingModel_test.cxx | 70 ++++++++++++++
 6 files changed, 221 insertions(+), 61 deletions(-)
 create mode 100644 include/datahandlinglibs/testutils/UnitTestUtilities.hpp
 create mode 100644 unittest/datahandlinglibs_DataHandlingModel_test.cxx

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 14e7a8c..f0069e9 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -83,6 +83,7 @@ daq_add_application(datahandlinglibs_test_composite_key test_composite_key_app.c
 #daq_add_unit_test(datahandlinglibs_BufferedReadWrite_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 #daq_add_unit_test(datahandlinglibs_VariableSizeElementQueue_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 daq_add_unit_test(datahandlinglibs_DataMoveCallbackRegistry_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
+daq_add_unit_test(datahandlinglibs_DataHandlingModel_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 
 ##############################################################################
 # Installation
diff --git a/include/datahandlinglibs/ReadoutTypes.hpp b/include/datahandlinglibs/ReadoutTypes.hpp
index 1845f9c..03c0522 100644
--- a/include/datahandlinglibs/ReadoutTypes.hpp
+++ b/include/datahandlinglibs/ReadoutTypes.hpp
@@ -8,6 +8,8 @@
 #ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_READOUTTYPES_HPP_
 #define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_READOUTTYPES_HPP_
 
+#include "daqdataformats/FragmentHeader.hpp"
+
 #include <cstdint> // uint_t types
 #include <memory>  // unique_ptr
 #include <tuple>   // std::tie
@@ -43,6 +45,12 @@ struct DUMMY_FRAME_STRUCT
     return timestamp;
   }
 
+  size_t get_num_frames() const { return frames_per_element; }
+
+  size_t get_frame_size() const { return frame_size; }
+
+  size_t get_payload_size() const { return get_num_frames() * get_frame_size(); }
+
   void set_another_key(uint64_t compkey)
   {
     another_key = compkey; 
@@ -65,6 +73,11 @@ struct DUMMY_FRAME_STRUCT
   static const constexpr size_t frame_size = DUMMY_FRAME_SIZE;
   static const constexpr uint8_t frames_per_element = 1; // NOLINT(build/unsigned)
   static const constexpr size_t element_size = DUMMY_FRAME_SIZE;
+  static const constexpr dunedaq::daqdataformats::SourceID::Subsystem subsystem =
+    dunedaq::daqdataformats::SourceID::Subsystem::kUnknown;
+  static const constexpr dunedaq::daqdataformats::FragmentType fragment_type =
+    dunedaq::daqdataformats::FragmentType::kUnknown;
+  static const constexpr uint64_t expected_tick_difference = 1;
 };
 
 } // namespace types
diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index 114f90b..c1fb14e 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -132,6 +132,102 @@ class DataHandlingModel : public DataHandlingConcept
   std::function<void(IDT&&)> m_consume_callback;
 
 protected:
+  class PostprocessManager {
+  public:
+    PostprocessManager(
+      LatencyBufferType& latency_buffer_impl, RawDataProcessorType& raw_processor_impl,
+      uint64_t processing_delay_ticks, uint64_t post_processing_delay_min_wait, uint64_t post_processing_delay_max_wait) : 
+      m_latency_buffer_impl{latency_buffer_impl},
+      m_raw_processor_impl{raw_processor_impl},
+      m_processing_delay_ticks{processing_delay_ticks},
+      m_post_processing_delay_min_wait{post_processing_delay_min_wait},
+      m_post_processing_delay_max_wait{post_processing_delay_max_wait},
+      m_first_cycle{true},
+      m_last_post_proc_time{std::chrono::system_clock::now()},
+      m_consecutive_timeouts{0},
+      m_max_wait_in_ticks{post_processing_delay_max_wait * 62500}
+    {
+    }
+
+    // Deferral of the post processing, to allow elements being reordered in the LB
+    // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value      
+    int perform_postprocessing(bool timeout) {
+      if (m_latency_buffer_impl.occupancy() == 0) {
+        return 0;
+      }
+
+      if (m_first_cycle) {
+        auto head = m_latency_buffer_impl.front();
+        m_unprocessed_element.set_timestamp(head->get_timestamp());
+        m_first_cycle = false;
+        TLOG() << "***** First pass post processing *****";
+      }
+
+      // Get the LB boundaries
+      auto tail = m_latency_buffer_impl.back();
+      auto newest_ts = tail->get_timestamp();
+          
+      timestamp_t end_win_ts = 0;
+      std::chrono::time_point<std::chrono::system_clock> now;
+
+      if (timeout) {      
+        ++m_consecutive_timeouts;
+        timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;  
+
+        // Cap to prevent end_win_ts from becoming unnecessarily large
+        timestamp_t timeout_cap = newest_ts + 1;
+        timeout_accumulated = std::min(timeout_accumulated, timeout_cap);  
+
+        end_win_ts = newest_ts - m_processing_delay_ticks + timeout_accumulated;
+
+      } else {
+        m_consecutive_timeouts = 0;
+        now = std::chrono::system_clock::now();
+        auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_last_post_proc_time);
+
+        if (milliseconds.count() > m_post_processing_delay_min_wait) {
+          if (newest_ts - m_unprocessed_element.get_timestamp() > m_processing_delay_ticks) {
+            end_win_ts = newest_ts - m_processing_delay_ticks;
+          }
+        }
+      }
+
+      if (end_win_ts == 0) {
+        return 0;
+      }
+
+      auto start_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, false);
+      m_unprocessed_element.set_timestamp(end_win_ts);
+      auto end_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, false);
+
+      if (start_iter == end_iter) {
+        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess";
+        return 0;
+      }
+
+      int processed = 0;
+      for (auto it = start_iter; it != end_iter; ++it) {
+        m_raw_processor_impl.postprocess_item(&(*it));
+        ++processed;
+      }
+
+      m_last_post_proc_time = now;
+
+      return processed;
+    }  
+
+  private:
+    LatencyBufferType& m_latency_buffer_impl;
+    RawDataProcessorType& m_raw_processor_impl;
+    const uint64_t m_processing_delay_ticks;
+    const uint64_t m_post_processing_delay_min_wait;
+    const uint64_t m_post_processing_delay_max_wait;  
+    bool m_first_cycle;
+    RDT m_unprocessed_element;
+    int m_consecutive_timeouts;
+    const timestamp_t m_max_wait_in_ticks;  
+    std::chrono::time_point<std::chrono::system_clock> m_last_post_proc_time;
+  };
 
   // Perform processing operations on payload
   void process_item(RDT&& payload);
diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 6bf88fa..5dc0b8a 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -315,23 +315,13 @@ folly::coro::Task<void>
 DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {  
 
   TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
-  timestamp_t newest_ts = 0;
-  timestamp_t end_win_ts = 0;
-  bool first_cycle = true;
-  auto last_post_proc_time = std::chrono::system_clock::now();
-  auto now = last_post_proc_time;
-  std::chrono::milliseconds milliseconds;
-  RDT processed_element;
-  int consecutive_timeouts = 0;
-  const timestamp_t max_wait_in_ticks = m_post_processing_delay_max_wait * 62500;  
-
-  // Deferral of the post processing, to allow elements being reordered in the LB
-  // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value  
+  PostprocessManager manager{
+    *m_latency_buffer_impl, *m_raw_processor_impl, m_processing_delay_ticks, m_post_processing_delay_min_wait, m_post_processing_delay_max_wait};
+
   while (m_run_marker.load()) {
     bool timeout = false;
-    bool postprocess = false;
 
-    try {
+     try {
       co_await folly::coro::timeout(
         m_baton.operator co_await(),
         std::chrono::milliseconds{m_post_processing_delay_max_wait},
@@ -339,59 +329,17 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {
       m_baton.reset();
     } catch (const folly::FutureTimeout&) {
       timeout = true;
-    }
-
-    if (m_latency_buffer_impl->occupancy() == 0) {
-      continue;
-    }
-
-    // Get the LB boundaries
-    auto tail = m_latency_buffer_impl->back();
-    newest_ts = tail->get_timestamp();
-
-    if (first_cycle) {
-      auto head = m_latency_buffer_impl->front();
-      processed_element.set_timestamp(head->get_timestamp());
-      first_cycle = false;
-      TLOG() << "***** First pass post processing *****";
-    }
-
-    now = std::chrono::system_clock::now();
-
-    if (timeout) {      
       ++m_num_post_processing_delay_max_waits;
-      ++consecutive_timeouts;
-      const timestamp_t timeout_accumulated = consecutive_timeouts * max_wait_in_ticks;      
-      end_win_ts = newest_ts - m_processing_delay_ticks + timeout_accumulated;
-      postprocess = true;
-    } else {
-      consecutive_timeouts = 0;
-      milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_post_proc_time);
-      if (milliseconds.count() > m_post_processing_delay_min_wait) {
-        if (newest_ts - processed_element.get_timestamp() > m_processing_delay_ticks) {
-          end_win_ts = newest_ts - m_processing_delay_ticks;
-          postprocess = true;
-        }
-      }
     }
 
-    if (postprocess) {
-      auto start_iter = m_latency_buffer_impl->lower_bound(processed_element, false);
-      processed_element.set_timestamp(end_win_ts);
-      auto end_iter = m_latency_buffer_impl->lower_bound(processed_element, false);
-
-      for (auto it = start_iter; it != end_iter; ++it) {
-        m_raw_processor_impl->postprocess_item(&(*it));
-        ++m_num_payloads;
-        ++m_sum_payloads;
-        ++m_stats_packet_count;
-      }
-
-      last_post_proc_time = now;
+    if (auto processed = manager.perform_postprocessing(timeout); processed > 0) {
+      m_num_payloads += processed;
+      m_sum_payloads += processed;
+      m_stats_packet_count += processed;
 
       if (timeout) {
         invoke_postprocess_schedule_timeout_policy();
-      }
+      }      
     }
   }
 }
diff --git a/include/datahandlinglibs/testutils/UnitTestUtilities.hpp b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
new file mode 100644
index 0000000..5f31564
--- /dev/null
+++ b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
@@ -0,0 +1,32 @@
+#ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
+#define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
+
+#include "datahandlinglibs/models/DataHandlingModel.hpp"
+#include "datahandlinglibs/models/DefaultRequestHandlerModel.hpp"
+#include "datahandlinglibs/models/TaskRawDataProcessorModel.hpp"
+
+namespace dunedaq {
+namespace datahandlinglibs {
+namespace unittest {
+
+template<typename ReadoutType,
+         typename RequestHandlerType,
+         typename LatencyBufferType,
+         typename RawDataProcessorType,
+         typename InputDataType = ReadoutType>
+class MockDataHandlingModel
+  : public 
+      DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>
+{
+public:
+  using Base = 
+    DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>;
+  using Base::Base;
+  using Base::PostprocessManager;
+};
+
+}
+}
+}
+
+#endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
diff --git a/unittest/datahandlinglibs_DataHandlingModel_test.cxx b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
new file mode 100644
index 0000000..3e37c14
--- /dev/null
+++ b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
@@ -0,0 +1,70 @@
+/**
+ * @file datahandlinglibs_DataHandlingModel_test.cxx Unit Tests for DataHandlingModel
+ *
+ * This is part of the DUNE DAQ Application Framework, copyright 2020.
+ * Licensing/copyright details are in the COPYING file that you should have
+ * received with this code.
+ */
+
+#define BOOST_TEST_MODULE datahandlinglibs_DataHandlingModel_test // NOLINT
+
+#include "boost/test/unit_test.hpp"
+
+#include "datahandlinglibs/testutils/UnitTestUtilities.hpp"
+#include "datahandlinglibs/ReadoutTypes.hpp"
+#include "datahandlinglibs/models/SkipListLatencyBufferModel.hpp"
+
+BOOST_AUTO_TEST_SUITE(datahandlinglibs_DataHandlingModel_test)
+
+using namespace dunedaq::datahandlinglibs;
+
+using ReadoutType = types::DUMMY_FRAME_STRUCT;
+
+BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBufferModel)
+{
+  std::atomic<bool> run_marker = true;
+
+  auto model = unittest::MockDataHandlingModel<
+    ReadoutType,
+    DefaultRequestHandlerModel<ReadoutType, SkipListLatencyBufferModel<ReadoutType>>,
+    SkipListLatencyBufferModel<ReadoutType>,
+    TaskRawDataProcessorModel<ReadoutType>>(run_marker);
+
+  auto buffer = std::make_shared<SkipListLatencyBufferModel<ReadoutType>>();
+
+  for (int i = 1; i < 6; i++) {
+    ReadoutType frame;
+    frame.timestamp = i * 62500;
+    buffer->write(std::move(frame));
+  }
+
+  const bool post_processing_enabled = true;
+  auto error_registry = std::make_unique<FrameErrorRegistry>();
+  
+  auto raw_processor = 
+    std::make_shared<TaskRawDataProcessorModel<ReadoutType>>(error_registry, post_processing_enabled);
+
+  const uint64_t delay_ticks = 4 * 62500;
+  const uint64_t delay_min_wait = 1;
+  const uint64_t delay_max_wait = 2;
+  
+  typename decltype(model)::PostprocessManager manager{
+    *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait};     
+    
+  // First pass
+  bool timeout = false;
+  int processed_count = manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 0);
+
+  timeout = true;
+  processed_count += manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 2);
+
+  processed_count += manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 4);
+  
+  processed_count += manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 5);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

From 7ad14bb9754d73303cf2b9055d3261594e7da219 Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Mon, 13 Oct 2025 10:58:31 +0200
Subject: [PATCH 03/14] Rename and cap fix

---
 .../models/DataHandlingModel.hpp               | 16 +++++++---------
 .../models/detail/DataHandlingModel.hxx        |  4 ++--
 .../testutils/UnitTestUtilities.hpp            |  2 +-
 ...datahandlinglibs_DataHandlingModel_test.cxx | 18 +++++++++++++-----
 4 files changed, 23 insertions(+), 17 deletions(-)

diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index c1fb14e..8c36c05 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -132,9 +132,9 @@ class DataHandlingModel : public DataHandlingConcept
   std::function<void(IDT&&)> m_consume_callback;
 
 protected:
-  class PostprocessManager {
+  class PostprocessScheduleAlgorithm {
   public:
-    PostprocessManager(
+    PostprocessScheduleAlgorithm(
       LatencyBufferType& latency_buffer_impl, RawDataProcessorType& raw_processor_impl,
       uint64_t processing_delay_ticks, uint64_t post_processing_delay_min_wait, uint64_t post_processing_delay_max_wait) : 
       m_latency_buffer_impl{latency_buffer_impl},
@@ -151,8 +151,9 @@ class DataHandlingModel : public DataHandlingConcept
 
     // Deferral of the post processing, to allow elements being reordered in the LB
     // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value      
-    int perform_postprocessing(bool timeout) {
+    int run(bool timeout) {
       if (m_latency_buffer_impl.occupancy() == 0) {
+        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (empty buffer)";
         return 0;
       }
 
@@ -174,12 +175,8 @@ class DataHandlingModel : public DataHandlingConcept
         ++m_consecutive_timeouts;
         timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;  
 
-        // Cap to prevent end_win_ts from becoming unnecessarily large
-        timestamp_t timeout_cap = newest_ts + 1;
-        timeout_accumulated = std::min(timeout_accumulated, timeout_cap);  
-
         end_win_ts = newest_ts - m_processing_delay_ticks + timeout_accumulated;
-
+        end_win_ts = std::min(end_win_ts, newest_ts + 1); // Cap to prevent end_win_ts from becoming unnecessarily large
       } else {
         m_consecutive_timeouts = 0;
         now = std::chrono::system_clock::now();
@@ -193,6 +190,7 @@ class DataHandlingModel : public DataHandlingConcept
       }
 
       if (end_win_ts == 0) {
+        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (end_win_ts == 0)";
         return 0;
       }
 
@@ -201,7 +199,7 @@ class DataHandlingModel : public DataHandlingConcept
       auto end_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, false);
 
       if (start_iter == end_iter) {
-        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess";
+        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (start_iter == end_iter)";
         return 0;
       }
 
diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 5dc0b8a..09cc1da 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -315,7 +315,7 @@ folly::coro::Task<void>
 DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {  
 
   TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
-  PostprocessManager manager{
+  PostprocessScheduleAlgorithm algorithm{
     *m_latency_buffer_impl, *m_raw_processor_impl, m_processing_delay_ticks, m_post_processing_delay_min_wait, m_post_processing_delay_max_wait};
 
   while (m_run_marker.load()) {
@@ -332,7 +332,7 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {
       ++m_num_post_processing_delay_max_waits;
     }
 
-    if (auto processed = manager.perform_postprocessing(timeout); processed > 0) {
+    if (auto processed = algorithm.run(timeout); processed > 0) {
       m_num_payloads += processed;
       m_sum_payloads += processed;
       m_stats_packet_count += processed;
diff --git a/include/datahandlinglibs/testutils/UnitTestUtilities.hpp b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
index 5f31564..7531d6e 100644
--- a/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
+++ b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
@@ -22,7 +22,7 @@ class MockDataHandlingModel
   using Base = 
     DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>;
   using Base::Base;
-  using Base::PostprocessManager;
+  using Base::PostprocessScheduleAlgorithm;
 };
 
 }
diff --git a/unittest/datahandlinglibs_DataHandlingModel_test.cxx b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
index 3e37c14..b3732dd 100644
--- a/unittest/datahandlinglibs_DataHandlingModel_test.cxx
+++ b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
@@ -48,22 +48,30 @@ BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBuffe
   const uint64_t delay_min_wait = 1;
   const uint64_t delay_max_wait = 2;
   
-  typename decltype(model)::PostprocessManager manager{
+  typename decltype(model)::PostprocessScheduleAlgorithm algorithm{
     *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait};     
     
   // First pass
   bool timeout = false;
-  int processed_count = manager.perform_postprocessing(timeout);
+  int processed_count = algorithm.run(timeout);
+  // Buffer = {1, 2, 3, 4, 5} delay_ticks = 4
+  // 5 - 1 > 4 is false => no postprocessing
   BOOST_REQUIRE_EQUAL(processed_count, 0);
 
   timeout = true;
-  processed_count += manager.perform_postprocessing(timeout);
+  // 1st timeout => timeout_accumulated = 1 * 2 (delay_max_wait = 2)
+  // end_win_ts = 5 - 4 + 2 => postprocess until 3 {1, 2}
+  processed_count += algorithm.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 2);
 
-  processed_count += manager.perform_postprocessing(timeout);
+  // 2nd timeout => timeout_accumulated = 2 * 2
+  // end_win_ts = 5 - 4 + 4 => postprocess until 5 {3, 4}  
+  processed_count += algorithm.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 4);
   
-  processed_count += manager.perform_postprocessing(timeout);
+  // 3rd timeout => timeout_accumulated = 3 * 2
+  // end_win_ts = 5 - 4 + 6 => postprocess until 6 (capped to newest_ts + 1) {5}    
+  processed_count += algorithm.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 5);
 }
 

From 6bcf425cec376f53a676ecd8bfc744955b991d1a Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Mon, 13 Oct 2025 11:05:21 +0200
Subject: [PATCH 04/14] Variable rename

---
 .../models/detail/DataHandlingModel.hxx                |  4 ++--
 unittest/datahandlinglibs_DataHandlingModel_test.cxx   | 10 +++++-----
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 09cc1da..9adb4c0 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -315,7 +315,7 @@ folly::coro::Task<void>
 DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {  
 
   TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
-  PostprocessScheduleAlgorithm algorithm{
+  PostprocessScheduleAlgorithm sched_algo{
     *m_latency_buffer_impl, *m_raw_processor_impl, m_processing_delay_ticks, m_post_processing_delay_min_wait, m_post_processing_delay_max_wait};
 
   while (m_run_marker.load()) {
@@ -332,7 +332,7 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {
       ++m_num_post_processing_delay_max_waits;
     }
 
-    if (auto processed = algorithm.run(timeout); processed > 0) {
+    if (auto processed = sched_algo.run(timeout); processed > 0) {
       m_num_payloads += processed;
       m_sum_payloads += processed;
       m_stats_packet_count += processed;
diff --git a/unittest/datahandlinglibs_DataHandlingModel_test.cxx b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
index b3732dd..5075d07 100644
--- a/unittest/datahandlinglibs_DataHandlingModel_test.cxx
+++ b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
@@ -48,12 +48,12 @@ BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBuffe
   const uint64_t delay_min_wait = 1;
   const uint64_t delay_max_wait = 2;
   
-  typename decltype(model)::PostprocessScheduleAlgorithm algorithm{
+  typename decltype(model)::PostprocessScheduleAlgorithm sched_algo{
     *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait};     
     
   // First pass
   bool timeout = false;
-  int processed_count = algorithm.run(timeout);
+  int processed_count = sched_algo.run(timeout);
   // Buffer = {1, 2, 3, 4, 5} delay_ticks = 4
   // 5 - 1 > 4 is false => no postprocessing
   BOOST_REQUIRE_EQUAL(processed_count, 0);
@@ -61,17 +61,17 @@ BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBuffe
   timeout = true;
   // 1st timeout => timeout_accumulated = 1 * 2 (delay_max_wait = 2)
   // end_win_ts = 5 - 4 + 2 => postprocess until 3 {1, 2}
-  processed_count += algorithm.run(timeout);
+  processed_count += sched_algo.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 2);
 
   // 2nd timeout => timeout_accumulated = 2 * 2
   // end_win_ts = 5 - 4 + 4 => postprocess until 5 {3, 4}  
-  processed_count += algorithm.run(timeout);
+  processed_count += sched_algo.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 4);
   
   // 3rd timeout => timeout_accumulated = 3 * 2
   // end_win_ts = 5 - 4 + 6 => postprocess until 6 (capped to newest_ts + 1) {5}    
-  processed_count += algorithm.run(timeout);
+  processed_count += sched_algo.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 5);
 }
 

From 2209892518b650757ef00fd232c98604a25f4c65 Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Mon, 13 Oct 2025 11:34:30 +0200
Subject: [PATCH 05/14] Lint updates

---
 include/datahandlinglibs/ReadoutTypes.hpp     |  2 +-
 .../models/DataHandlingModel.hpp              | 47 ++++++++++---------
 .../models/detail/DataHandlingModel.hxx       | 14 ++++--
 .../detail/SkipListLatencyBufferModel.hxx     |  2 +
 .../testutils/UnitTestUtilities.hpp           | 19 +++++---
 ...atahandlinglibs_DataHandlingModel_test.cxx | 38 ++++++++-------
 6 files changed, 72 insertions(+), 50 deletions(-)

diff --git a/include/datahandlinglibs/ReadoutTypes.hpp b/include/datahandlinglibs/ReadoutTypes.hpp
index 03c0522..bc1b085 100644
--- a/include/datahandlinglibs/ReadoutTypes.hpp
+++ b/include/datahandlinglibs/ReadoutTypes.hpp
@@ -77,7 +77,7 @@ struct DUMMY_FRAME_STRUCT
     dunedaq::daqdataformats::SourceID::Subsystem::kUnknown;
   static const constexpr dunedaq::daqdataformats::FragmentType fragment_type =
     dunedaq::daqdataformats::FragmentType::kUnknown;
-  static const constexpr uint64_t expected_tick_difference = 1;
+  static const constexpr uint64_t expected_tick_difference = 1; // NOLINT(build/unsigned)
 };
 
 } // namespace types
diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index 8c36c05..bcbc07e 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -49,6 +49,7 @@
 #include <folly/coro/Task.h>
 #include <folly/futures/ThreadWheelTimekeeper.h>
 
+#include <algorithm>
 #include <functional>
 #include <memory>
 #include <string>
@@ -132,26 +133,30 @@ class DataHandlingModel : public DataHandlingConcept
   std::function<void(IDT&&)> m_consume_callback;
 
 protected:
-  class PostprocessScheduleAlgorithm {
+  class PostprocessScheduleAlgorithm
+  {
   public:
-    PostprocessScheduleAlgorithm(
-      LatencyBufferType& latency_buffer_impl, RawDataProcessorType& raw_processor_impl,
-      uint64_t processing_delay_ticks, uint64_t post_processing_delay_min_wait, uint64_t post_processing_delay_max_wait) : 
-      m_latency_buffer_impl{latency_buffer_impl},
-      m_raw_processor_impl{raw_processor_impl},
-      m_processing_delay_ticks{processing_delay_ticks},
-      m_post_processing_delay_min_wait{post_processing_delay_min_wait},
-      m_post_processing_delay_max_wait{post_processing_delay_max_wait},
-      m_first_cycle{true},
-      m_last_post_proc_time{std::chrono::system_clock::now()},
-      m_consecutive_timeouts{0},
-      m_max_wait_in_ticks{post_processing_delay_max_wait * 62500}
+    PostprocessScheduleAlgorithm(LatencyBufferType& latency_buffer_impl,
+                                 RawDataProcessorType& raw_processor_impl,
+                                 uint64_t processing_delay_ticks, // NOLINT(build/unsigned)
+                                 uint64_t post_processing_delay_min_wait, // NOLINT(build/unsigned)
+                                 uint64_t post_processing_delay_max_wait) // NOLINT(build/unsigned)
+      : m_latency_buffer_impl{ latency_buffer_impl }
+      , m_raw_processor_impl{ raw_processor_impl }
+      , m_processing_delay_ticks{ processing_delay_ticks }
+      , m_post_processing_delay_min_wait{ post_processing_delay_min_wait }
+      , m_post_processing_delay_max_wait{ post_processing_delay_max_wait }
+      , m_first_cycle{ true }
+      , m_last_post_proc_time{ std::chrono::system_clock::now() }
+      , m_consecutive_timeouts{ 0 }
+      , m_max_wait_in_ticks{ post_processing_delay_max_wait * 62500 }
     {
     }
 
     // Deferral of the post processing, to allow elements being reordered in the LB
-    // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value      
-    int run(bool timeout) {
+    // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value
+    int run(bool timeout)
+    {
       if (m_latency_buffer_impl.occupancy() == 0) {
         TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (empty buffer)";
         return 0;
@@ -217,9 +222,9 @@ class DataHandlingModel : public DataHandlingConcept
   private:
     LatencyBufferType& m_latency_buffer_impl;
     RawDataProcessorType& m_raw_processor_impl;
-    const uint64_t m_processing_delay_ticks;
-    const uint64_t m_post_processing_delay_min_wait;
-    const uint64_t m_post_processing_delay_max_wait;  
+    const uint64_t m_processing_delay_ticks; // NOLINT(build/unsigned)
+    const uint64_t m_post_processing_delay_min_wait; // NOLINT(build/unsigned)
+    const uint64_t m_post_processing_delay_max_wait; // NOLINT(build/unsigned)
     bool m_first_cycle;
     RDT m_unprocessed_element;
     int m_consecutive_timeouts;
@@ -277,9 +282,9 @@ class DataHandlingModel : public DataHandlingConcept
   int m_current_fake_trigger_id;
   daqdataformats::SourceID m_sourceid;
   daqdataformats::run_number_t m_run_number;
-  uint64_t m_processing_delay_ticks;
-  uint64_t m_post_processing_delay_min_wait;
-  uint64_t m_post_processing_delay_max_wait;
+  uint64_t m_processing_delay_ticks; // NOLINT(build/unsigned)
+  uint64_t m_post_processing_delay_min_wait; // NOLINT(build/unsigned)
+  uint64_t m_post_processing_delay_max_wait; // NOLINT(build/unsigned)
 
   // STATS
   using metric_t = dunedaq::datahandlinglibs::opmon::DataHandlerInfo;
diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 9adb4c0..2300b5d 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -312,19 +312,23 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::run_consume()
 
 template<class RDT, class RHT, class LBT, class RPT, class IDT>
 folly::coro::Task<void>
-DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {  
+DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule()
+{
 
   TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
-  PostprocessScheduleAlgorithm sched_algo{
-    *m_latency_buffer_impl, *m_raw_processor_impl, m_processing_delay_ticks, m_post_processing_delay_min_wait, m_post_processing_delay_max_wait};
+  PostprocessScheduleAlgorithm sched_algo{ *m_latency_buffer_impl,
+                                           *m_raw_processor_impl,
+                                           m_processing_delay_ticks,
+                                           m_post_processing_delay_min_wait,
+                                           m_post_processing_delay_max_wait };
 
   while (m_run_marker.load()) {
     bool timeout = false;
 
-     try {
+    try {
       co_await folly::coro::timeout(
         m_baton.operator co_await(),
-        std::chrono::milliseconds{m_post_processing_delay_max_wait},
+        std::chrono::milliseconds{ m_post_processing_delay_max_wait },
         m_timekeeper.get());
       m_baton.reset();
     } catch (const folly::FutureTimeout&) {
diff --git a/include/datahandlinglibs/models/detail/SkipListLatencyBufferModel.hxx b/include/datahandlinglibs/models/detail/SkipListLatencyBufferModel.hxx
index efd5e67..bfe2a73 100644
--- a/include/datahandlinglibs/models/detail/SkipListLatencyBufferModel.hxx
+++ b/include/datahandlinglibs/models/detail/SkipListLatencyBufferModel.hxx
@@ -1,5 +1,7 @@
 // Declarations for SkipListLatencyBufferModel
 
+#include "datahandlinglibs/opmon/datahandling_info.pb.h"
+
 namespace dunedaq {
 namespace datahandlinglibs {
 
diff --git a/include/datahandlinglibs/testutils/UnitTestUtilities.hpp b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
index 7531d6e..16163f7 100644
--- a/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
+++ b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
@@ -1,3 +1,11 @@
+/**
+ * @file UnitTestUtilities.hpp Unit test helper classes
+ *
+ * This is part of the DUNE DAQ Application Framework, copyright 2020.
+ * Licensing/copyright details are in the COPYING file that you should have
+ * received with this code.
+ */
+
 #ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
 #define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
 
@@ -15,18 +23,17 @@ template<typename ReadoutType,
          typename RawDataProcessorType,
          typename InputDataType = ReadoutType>
 class MockDataHandlingModel
-  : public 
-      DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>
+  : public DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>
 {
 public:
-  using Base = 
+  using Base =
     DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>;
   using Base::Base;
   using Base::PostprocessScheduleAlgorithm;
 };
 
-}
-}
-}
+} // namespace unittest
+} // namespace datahandlinglibs
+} // namespace dunedaq
 
 #endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
diff --git a/unittest/datahandlinglibs_DataHandlingModel_test.cxx b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
index 5075d07..c4d4b1a 100644
--- a/unittest/datahandlinglibs_DataHandlingModel_test.cxx
+++ b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
@@ -10,9 +10,12 @@
 
 #include "boost/test/unit_test.hpp"
 
-#include "datahandlinglibs/testutils/UnitTestUtilities.hpp"
 #include "datahandlinglibs/ReadoutTypes.hpp"
 #include "datahandlinglibs/models/SkipListLatencyBufferModel.hpp"
+#include "datahandlinglibs/testutils/UnitTestUtilities.hpp"
+
+#include <memory>
+#include <utility>
 
 BOOST_AUTO_TEST_SUITE(datahandlinglibs_DataHandlingModel_test)
 
@@ -24,11 +27,11 @@ BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBuffe
 {
   std::atomic<bool> run_marker = true;
 
-  auto model = unittest::MockDataHandlingModel<
-    ReadoutType,
-    DefaultRequestHandlerModel<ReadoutType, SkipListLatencyBufferModel<ReadoutType>>,
-    SkipListLatencyBufferModel<ReadoutType>,
-    TaskRawDataProcessorModel<ReadoutType>>(run_marker);
+  auto model =
+    unittest::MockDataHandlingModel<ReadoutType,
+                                    DefaultRequestHandlerModel<ReadoutType, SkipListLatencyBufferModel<ReadoutType>>,
+                                    SkipListLatencyBufferModel<ReadoutType>,
+                                    TaskRawDataProcessorModel<ReadoutType>>(run_marker);
 
   auto buffer = std::make_shared<SkipListLatencyBufferModel<ReadoutType>>();
 
@@ -40,17 +43,18 @@ BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBuffe
 
   const bool post_processing_enabled = true;
   auto error_registry = std::make_unique<FrameErrorRegistry>();
-  
-  auto raw_processor = 
+
+  auto raw_processor =
     std::make_shared<TaskRawDataProcessorModel<ReadoutType>>(error_registry, post_processing_enabled);
 
-  const uint64_t delay_ticks = 4 * 62500;
-  const uint64_t delay_min_wait = 1;
-  const uint64_t delay_max_wait = 2;
-  
+  const uint64_t delay_ticks = 4 * 62500; // NOLINT(build/unsigned)
+  const uint64_t delay_min_wait = 1; // NOLINT(build/unsigned)
+  const uint64_t delay_max_wait = 2; // NOLINT(build/unsigned)
+
   typename decltype(model)::PostprocessScheduleAlgorithm sched_algo{
-    *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait};     
-    
+    *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait
+  };
+
   // First pass
   bool timeout = false;
   int processed_count = sched_algo.run(timeout);
@@ -65,12 +69,12 @@ BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBuffe
   BOOST_REQUIRE_EQUAL(processed_count, 2);
 
   // 2nd timeout => timeout_accumulated = 2 * 2
-  // end_win_ts = 5 - 4 + 4 => postprocess until 5 {3, 4}  
+  // end_win_ts = 5 - 4 + 4 => postprocess until 5 {3, 4}
   processed_count += sched_algo.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 4);
-  
+
   // 3rd timeout => timeout_accumulated = 3 * 2
-  // end_win_ts = 5 - 4 + 6 => postprocess until 6 (capped to newest_ts + 1) {5}    
+  // end_win_ts = 5 - 4 + 6 => postprocess until 6 (capped to newest_ts + 1) {5}
   processed_count += sched_algo.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 5);
 }

From deddaa8e50c6e3d28f1193186394eb436114e252 Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Mon, 13 Oct 2025 17:04:27 +0200
Subject: [PATCH 06/14] Uninitialized variables fix

---
 .../models/DataHandlingModel.hpp                | 17 +++++++++--------
 .../datahandlinglibs_DataHandlingModel_test.cxx |  4 ++--
 2 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index bcbc07e..096458f 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -147,6 +147,7 @@ class DataHandlingModel : public DataHandlingConcept
       , m_post_processing_delay_min_wait{ post_processing_delay_min_wait }
       , m_post_processing_delay_max_wait{ post_processing_delay_max_wait }
       , m_first_cycle{ true }
+      , m_unprocessed_element{}
       , m_last_post_proc_time{ std::chrono::system_clock::now() }
       , m_consecutive_timeouts{ 0 }
       , m_max_wait_in_ticks{ post_processing_delay_max_wait * 62500 }
@@ -174,31 +175,31 @@ class DataHandlingModel : public DataHandlingConcept
       auto newest_ts = tail->get_timestamp();
           
       timestamp_t end_win_ts = 0;
-      std::chrono::time_point<std::chrono::system_clock> now;
+      std::chrono::time_point<std::chrono::system_clock> now{ std::chrono::system_clock::now() };
 
       if (timeout) {      
         ++m_consecutive_timeouts;
         timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;  
 
         end_win_ts = newest_ts - m_processing_delay_ticks + timeout_accumulated;
-        end_win_ts = std::min(end_win_ts, newest_ts + 1); // Cap to prevent end_win_ts from becoming unnecessarily large
+        end_win_ts = std::min(end_win_ts, newest_ts + 1); // Cap to prevent end_win_ts from becoming unnecessarily large 
       } else {
         m_consecutive_timeouts = 0;
-        now = std::chrono::system_clock::now();
         auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_last_post_proc_time);
 
         if (milliseconds.count() > m_post_processing_delay_min_wait) {
           if (newest_ts - m_unprocessed_element.get_timestamp() > m_processing_delay_ticks) {
             end_win_ts = newest_ts - m_processing_delay_ticks;
+          } else {
+            TLOG_DEBUG(TLVL_WORK_STEPS) << "Not ready to postprocess (m_processing_delay_ticks is greater)";
+            return 0;
           }
+        } else {
+          TLOG_DEBUG(TLVL_WORK_STEPS) << "Not ready to postprocess (too fast)";
+          return 0;
         }
       }
 
-      if (end_win_ts == 0) {
-        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (end_win_ts == 0)";
-        return 0;
-      }
-
       auto start_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, false);
       m_unprocessed_element.set_timestamp(end_win_ts);
       auto end_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, false);
diff --git a/unittest/datahandlinglibs_DataHandlingModel_test.cxx b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
index c4d4b1a..ccde460 100644
--- a/unittest/datahandlinglibs_DataHandlingModel_test.cxx
+++ b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
@@ -23,7 +23,7 @@ using namespace dunedaq::datahandlinglibs;
 
 using ReadoutType = types::DUMMY_FRAME_STRUCT;
 
-BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBufferModel)
+BOOST_AUTO_TEST_CASE(datahandlinglibs_DataHandlingModel_PostprocessScheduleAlgorithm_timeout)
 {
   std::atomic<bool> run_marker = true;
 
@@ -36,7 +36,7 @@ BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBuffe
   auto buffer = std::make_shared<SkipListLatencyBufferModel<ReadoutType>>();
 
   for (int i = 1; i < 6; i++) {
-    ReadoutType frame;
+    ReadoutType frame{};
     frame.timestamp = i * 62500;
     buffer->write(std::move(frame));
   }

From 7b4e158374e91b04cf6dbcf3359f5d70380d664b Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Wed, 15 Oct 2025 11:20:39 +0200
Subject: [PATCH 07/14] Added SupportsDelayedPostprocessing concept Renamed
 m_unprocessed_element Optimization in timeout logic

---
 .../concepts/LatencyBufferConcept.hpp         |  4 ++++
 .../models/DataHandlingModel.hpp              | 19 ++++++++++++-------
 .../models/SkipListLatencyBufferModel.hpp     |  2 ++
 .../models/detail/DataHandlingModel.hxx       |  7 +++++++
 ...atahandlinglibs_DataHandlingModel_test.cxx |  5 +++++
 5 files changed, 30 insertions(+), 7 deletions(-)

diff --git a/include/datahandlinglibs/concepts/LatencyBufferConcept.hpp b/include/datahandlinglibs/concepts/LatencyBufferConcept.hpp
index 3143883..b36b40f 100644
--- a/include/datahandlinglibs/concepts/LatencyBufferConcept.hpp
+++ b/include/datahandlinglibs/concepts/LatencyBufferConcept.hpp
@@ -13,10 +13,14 @@
 #include "opmonlib/MonitorableObject.hpp"
 
 #include <cstddef>
+#include <concepts>
 
 namespace dunedaq {
 namespace datahandlinglibs {
 
+template<typename T>
+concept SupportsDelayedPostprocessing = T::supports_delayed_postprocessing;
+
 /**
  * Concept of a LatencyBuffer.
  *
diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index 096458f..e4cbb6c 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -147,7 +147,7 @@ class DataHandlingModel : public DataHandlingConcept
       , m_post_processing_delay_min_wait{ post_processing_delay_min_wait }
       , m_post_processing_delay_max_wait{ post_processing_delay_max_wait }
       , m_first_cycle{ true }
-      , m_unprocessed_element{}
+      , m_processed_up_to{}
       , m_last_post_proc_time{ std::chrono::system_clock::now() }
       , m_consecutive_timeouts{ 0 }
       , m_max_wait_in_ticks{ post_processing_delay_max_wait * 62500 }
@@ -165,7 +165,7 @@ class DataHandlingModel : public DataHandlingConcept
 
       if (m_first_cycle) {
         auto head = m_latency_buffer_impl.front();
-        m_unprocessed_element.set_timestamp(head->get_timestamp());
+        m_processed_up_to.set_timestamp(head->get_timestamp());
         m_first_cycle = false;
         TLOG() << "***** First pass post processing *****";
       }
@@ -178,6 +178,11 @@ class DataHandlingModel : public DataHandlingConcept
       std::chrono::time_point<std::chrono::system_clock> now{ std::chrono::system_clock::now() };
 
       if (timeout) {      
+        if (m_processed_up_to.get_timestamp() >= newest_ts + 1) {
+          TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (at or past cap)";
+          return 0;
+        }        
+
         ++m_consecutive_timeouts;
         timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;  
 
@@ -188,7 +193,7 @@ class DataHandlingModel : public DataHandlingConcept
         auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_last_post_proc_time);
 
         if (milliseconds.count() > m_post_processing_delay_min_wait) {
-          if (newest_ts - m_unprocessed_element.get_timestamp() > m_processing_delay_ticks) {
+          if (newest_ts - m_processed_up_to.get_timestamp() > m_processing_delay_ticks) {
             end_win_ts = newest_ts - m_processing_delay_ticks;
           } else {
             TLOG_DEBUG(TLVL_WORK_STEPS) << "Not ready to postprocess (m_processing_delay_ticks is greater)";
@@ -200,9 +205,9 @@ class DataHandlingModel : public DataHandlingConcept
         }
       }
 
-      auto start_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, false);
-      m_unprocessed_element.set_timestamp(end_win_ts);
-      auto end_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, false);
+      auto start_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);
+      m_processed_up_to.set_timestamp(end_win_ts);
+      auto end_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);
 
       if (start_iter == end_iter) {
         TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (start_iter == end_iter)";
@@ -227,7 +232,7 @@ class DataHandlingModel : public DataHandlingConcept
     const uint64_t m_post_processing_delay_min_wait; // NOLINT(build/unsigned)
     const uint64_t m_post_processing_delay_max_wait; // NOLINT(build/unsigned)
     bool m_first_cycle;
-    RDT m_unprocessed_element;
+    RDT m_processed_up_to;
     int m_consecutive_timeouts;
     const timestamp_t m_max_wait_in_ticks;  
     std::chrono::time_point<std::chrono::system_clock> m_last_post_proc_time;
diff --git a/include/datahandlinglibs/models/SkipListLatencyBufferModel.hpp b/include/datahandlinglibs/models/SkipListLatencyBufferModel.hpp
index dd064be..b31cab6 100644
--- a/include/datahandlinglibs/models/SkipListLatencyBufferModel.hpp
+++ b/include/datahandlinglibs/models/SkipListLatencyBufferModel.hpp
@@ -35,6 +35,8 @@ class SkipListLatencyBufferModel : public LatencyBufferConcept<T>
   using SkipListTAcc = typename folly::ConcurrentSkipList<T>::Accessor; // SKL Accessor
   using SkipListTSkip = typename folly::ConcurrentSkipList<T>::Skipper; // Skipper accessor
 
+  static constexpr bool supports_delayed_postprocessing = true;
+
   // Constructor
   SkipListLatencyBufferModel()
     : m_skip_list(folly::ConcurrentSkipList<T>::createInstance(unconfigured_head_height))
diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 2300b5d..6640ab4 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -85,6 +85,13 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::init(const appmodel::DataHandlerModu
   m_post_processing_delay_min_wait = mcfg->get_module_configuration()->get_post_processing_delay_min_wait();
   m_post_processing_delay_max_wait = mcfg->get_module_configuration()->get_post_processing_delay_max_wait();
   
+  if (m_processing_delay_ticks) {
+    if constexpr (!SupportsDelayedPostprocessing<LBT>) {
+      ers::error(ConfigurationError(ERS_HERE, m_sourceid,
+        "Delayed postprocessing (post_processing_delay_ticks > 0) requires a sorted buffer (SkipList). "
+        "Queue buffers (FixedRateQueue, BinarySearchQueue) expect in-order data and must use post_processing_delay_ticks = 0."));
+    }
+  }
 
   // Configure implementations:
   m_raw_processor_impl->conf(mcfg);
diff --git a/unittest/datahandlinglibs_DataHandlingModel_test.cxx b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
index ccde460..e9aff7a 100644
--- a/unittest/datahandlinglibs_DataHandlingModel_test.cxx
+++ b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
@@ -77,6 +77,11 @@ BOOST_AUTO_TEST_CASE(datahandlinglibs_DataHandlingModel_PostprocessScheduleAlgor
   // end_win_ts = 5 - 4 + 6 => postprocess until 6 (capped to newest_ts + 1) {5}
   processed_count += sched_algo.run(timeout);
   BOOST_REQUIRE_EQUAL(processed_count, 5);
+
+  // 4th timeout
+  // m_processed_up_to.timestamp = newest_ts + 1 => nothing to postprocess (at cap)
+  processed_count += sched_algo.run(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 5);
 }
 
 BOOST_AUTO_TEST_SUITE_END()

From 6e71a70efcdf1eba46553c5c4cc5627d1ac23efd Mon Sep 17 00:00:00 2001
From: Deniz Tuana Ergonul Uzun <deniz.tuana.ergonul@cern.ch>
Date: Fri, 17 Oct 2025 13:09:30 +0200
Subject: [PATCH 08/14] Postprocess coroutine timeout test

---
 .../models/DataHandlingModel.hpp              |  4 +-
 .../models/detail/DataHandlingModel.hxx       | 17 +++++-
 .../testutils/UnitTestUtilities.hpp           | 22 ++++++++
 ...atahandlinglibs_DataHandlingModel_test.cxx | 52 +++++++++++++++++--
 4 files changed, 87 insertions(+), 8 deletions(-)

diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index e4cbb6c..5e5e8cb 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -47,7 +47,7 @@
 
 #include <folly/coro/Baton.h>
 #include <folly/coro/Task.h>
-#include <folly/futures/ThreadWheelTimekeeper.h>
+#include <folly/futures/Future.h>
 
 #include <algorithm>
 #include <functional>
@@ -339,7 +339,7 @@ class DataHandlingModel : public DataHandlingConcept
   // POSTPROCESS SCHEDULER
   utilities::ReusableThread m_postprocess_scheduler_thread;
   folly::coro::Baton m_baton;
-  std::unique_ptr<folly::ThreadWheelTimekeeper> m_timekeeper;
+  std::unique_ptr<folly::Timekeeper> m_timekeeper;
 
   // LATENCY BUFFER
   std::shared_ptr<LatencyBufferType> m_latency_buffer_impl;
diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 6640ab4..dad6454 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -2,6 +2,9 @@
 
 #include <folly/coro/BlockingWait.h>
 #include <folly/coro/Timeout.h>
+#include <folly/futures/ThreadWheelTimekeeper.h>
+#include <folly/coro/CurrentExecutor.h>
+#include <folly/CancellationToken.h>
 
 #include <typeinfo>
 
@@ -323,25 +326,35 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule()
 {
 
   TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
+
   PostprocessScheduleAlgorithm sched_algo{ *m_latency_buffer_impl,
                                            *m_raw_processor_impl,
                                            m_processing_delay_ticks,
                                            m_post_processing_delay_min_wait,
                                            m_post_processing_delay_max_wait };
+                                           
+  const auto wait_data = [this]() -> folly::coro::Task<void> {
+    // folly::coro::timeout cancels the task on timeout.
+    // Baton is not cancellable, so we attach a callback to resume the coroutine.
+    auto token = co_await folly::coro::co_current_cancellation_token;
+    folly::CancellationCallback cb(token, [this] { m_baton.post(); });
+    co_await m_baton; // Wait data
+  };
 
   while (m_run_marker.load()) {
     bool timeout = false;
 
     try {
       co_await folly::coro::timeout(
-        m_baton.operator co_await(),
+        wait_data(),
         std::chrono::milliseconds{ m_post_processing_delay_max_wait },
         m_timekeeper.get());
-      m_baton.reset();
+
     } catch (const folly::FutureTimeout&) {
       timeout = true;
       ++m_num_post_processing_delay_max_waits;
     }
+    m_baton.reset();
 
     if (auto processed = sched_algo.run(timeout); processed > 0) {
       m_num_payloads += processed;
diff --git a/include/datahandlinglibs/testutils/UnitTestUtilities.hpp b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
index 16163f7..16673fd 100644
--- a/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
+++ b/include/datahandlinglibs/testutils/UnitTestUtilities.hpp
@@ -30,6 +30,28 @@ class MockDataHandlingModel
     DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>;
   using Base::Base;
   using Base::PostprocessScheduleAlgorithm;
+  using typename Base::num_post_processing_delay_max_waits_t;
+
+  void test_run_postprocess_scheduler(
+    std::shared_ptr<LatencyBufferType> latency_buffer_impl, std::shared_ptr<RawDataProcessorType> raw_processor_impl,
+    std::unique_ptr<folly::Timekeeper> timekeeper, uint64_t post_processing_delay_max_wait)
+  {
+    this->m_latency_buffer_impl = latency_buffer_impl;
+    this->m_raw_processor_impl = raw_processor_impl;
+    this->m_timekeeper = std::move(timekeeper);
+    this->m_post_processing_delay_max_wait = post_processing_delay_max_wait;
+    this->run_postprocess_scheduler();
+  }
+
+  num_post_processing_delay_max_waits_t get_num_post_processing_delay_max_waits()
+  {
+    return this->m_num_post_processing_delay_max_waits.load();
+  }
+
+  void set_run_marker(bool run_marker)
+  {
+    return this->m_run_marker.store(run_marker);
+  }
 };
 
 } // namespace unittest
diff --git a/unittest/datahandlinglibs_DataHandlingModel_test.cxx b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
index e9aff7a..ebe8d48 100644
--- a/unittest/datahandlinglibs_DataHandlingModel_test.cxx
+++ b/unittest/datahandlinglibs_DataHandlingModel_test.cxx
@@ -14,6 +14,8 @@
 #include "datahandlinglibs/models/SkipListLatencyBufferModel.hpp"
 #include "datahandlinglibs/testutils/UnitTestUtilities.hpp"
 
+#include <folly/futures/ManualTimekeeper.h>
+
 #include <memory>
 #include <utility>
 
@@ -23,6 +25,48 @@ using namespace dunedaq::datahandlinglibs;
 
 using ReadoutType = types::DUMMY_FRAME_STRUCT;
 
+BOOST_AUTO_TEST_CASE(datahandlinglibs_DataHandlingModel_run_postprocess_scheduler_timeout)
+{
+  std::atomic<bool> run_marker = true;
+
+  auto model =
+    unittest::MockDataHandlingModel<ReadoutType,
+                                    DefaultRequestHandlerModel<ReadoutType, SkipListLatencyBufferModel<ReadoutType>>,
+                                    SkipListLatencyBufferModel<ReadoutType>,
+                                    TaskRawDataProcessorModel<ReadoutType>>(run_marker);
+
+  auto buffer = std::make_shared<SkipListLatencyBufferModel<ReadoutType>>(); // Empty buffer
+
+  constexpr bool post_processing_enabled = true;
+  auto error_registry = std::make_unique<FrameErrorRegistry>();
+
+  auto raw_processor =
+    std::make_shared<TaskRawDataProcessorModel<ReadoutType>>(error_registry, post_processing_enabled);
+
+  auto timekeeper = std::make_unique<folly::ManualTimekeeper>();
+  auto* timekeeper_ptr = timekeeper.get();
+
+  constexpr uint64_t delay_max_wait = 2; // NOLINT(build/unsigned)
+
+  std::thread coro_thread([&]() {
+    model.test_run_postprocess_scheduler(buffer, raw_processor, std::move(timekeeper), delay_max_wait);
+  });
+
+  // Wait for coroutine to start then timeout to get registered
+  while (timekeeper_ptr->numScheduled() == 0) {
+    std::this_thread::sleep_for(1ms);
+  }
+  // Safe-guard for the delay between timeout registration and coroutine suspension  
+  // If the test is failing, consider a longer sleep or a better way to synchronize
+  std::this_thread::sleep_for(1ms);
+  timekeeper_ptr->advance(std::chrono::milliseconds{ delay_max_wait }); // Trigger a timeout
+
+  model.set_run_marker(false); // Let coroutine end
+  coro_thread.join(); // The test will stuck here if timeout is not triggered (because of folly::coro::blockingWait)
+
+  BOOST_REQUIRE_EQUAL(model.get_num_post_processing_delay_max_waits(), 1);
+}
+
 BOOST_AUTO_TEST_CASE(datahandlinglibs_DataHandlingModel_PostprocessScheduleAlgorithm_timeout)
 {
   std::atomic<bool> run_marker = true;
@@ -41,15 +85,15 @@ BOOST_AUTO_TEST_CASE(datahandlinglibs_DataHandlingModel_PostprocessScheduleAlgor
     buffer->write(std::move(frame));
   }
 
-  const bool post_processing_enabled = true;
+  constexpr bool post_processing_enabled = true;
   auto error_registry = std::make_unique<FrameErrorRegistry>();
 
   auto raw_processor =
     std::make_shared<TaskRawDataProcessorModel<ReadoutType>>(error_registry, post_processing_enabled);
 
-  const uint64_t delay_ticks = 4 * 62500; // NOLINT(build/unsigned)
-  const uint64_t delay_min_wait = 1; // NOLINT(build/unsigned)
-  const uint64_t delay_max_wait = 2; // NOLINT(build/unsigned)
+  constexpr uint64_t delay_ticks = 4 * 62500; // NOLINT(build/unsigned)
+  constexpr uint64_t delay_min_wait = 1; // NOLINT(build/unsigned)
+  constexpr uint64_t delay_max_wait = 2; // NOLINT(build/unsigned)
 
   typename decltype(model)::PostprocessScheduleAlgorithm sched_algo{
     *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait

From 665b0be7d45d3c00b7f95b51050fda5f7377f1eb Mon Sep 17 00:00:00 2001
From: Alessandro Thea <alessandro.thea@cern.ch>
Date: Fri, 17 Oct 2025 21:48:01 +0200
Subject: [PATCH 09/14] Skipping timeout coroutine if no max wait set.

---
 .../models/detail/DataHandlingModel.hxx       | 23 +++++++++++--------
 1 file changed, 14 insertions(+), 9 deletions(-)

diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index dad6454..c6d19d6 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -344,16 +344,21 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule()
   while (m_run_marker.load()) {
     bool timeout = false;
 
-    try {
-      co_await folly::coro::timeout(
-        wait_data(),
-        std::chrono::milliseconds{ m_post_processing_delay_max_wait },
-        m_timekeeper.get());
-
-    } catch (const folly::FutureTimeout&) {
-      timeout = true;
-      ++m_num_post_processing_delay_max_waits;
+    if ( m_post_processing_delay_max_wait > 0 ) {
+      try {
+        co_await folly::coro::timeout(
+          wait_data(),
+          std::chrono::milliseconds{ m_post_processing_delay_max_wait },
+          m_timekeeper.get());
+
+      } catch (const folly::FutureTimeout&) {
+        timeout = true;
+        ++m_num_post_processing_delay_max_waits;
+      }
+    } else {
+      co_await m_baton;
     }
+
     m_baton.reset();
 
     if (auto processed = sched_algo.run(timeout); processed > 0) {

From 3bd3e893289b7921a5a7f7eabcdf8138fe7718db Mon Sep 17 00:00:00 2001
From: Alessandro Thea <alessandro.thea@cern.ch>
Date: Mon, 20 Oct 2025 11:57:53 +0200
Subject: [PATCH 10/14] Adding a toy test app to examine the timeout coroutine
 logic in isolation

---
 test/apps/test_co_timeout_app.cxx | 106 ++++++++++++++++++++++++++++++
 1 file changed, 106 insertions(+)
 create mode 100644 test/apps/test_co_timeout_app.cxx

diff --git a/test/apps/test_co_timeout_app.cxx b/test/apps/test_co_timeout_app.cxx
new file mode 100644
index 0000000..6eb6633
--- /dev/null
+++ b/test/apps/test_co_timeout_app.cxx
@@ -0,0 +1,106 @@
+/**
+ * @file test_ratelimiter_app.cxx Test application for
+ * ratelimiter implementation
+ *
+ * This is part of the DUNE DAQ Application Framework, copyright 2020.
+ * Licensing/copyright details are in the COPYING file that you should have
+ * received with this code.
+ */
+#
+
+#include <thread>
+#include <atomic>
+#include <chrono>
+#include <iostream>
+#include <folly/coro/Baton.h>
+#include <folly/coro/Task.h>
+#include <folly/futures/Future.h>
+#include <folly/coro/CurrentExecutor.h>
+#include <folly/coro/Timeout.h>
+#include <folly/futures/ThreadWheelTimekeeper.h>
+#include <folly/coro/BlockingWait.h>
+
+
+// using namespace dunedaq::datahandlinglibs;
+using namespace std::chrono_literals;
+
+folly::coro::Baton baton{0};
+uint32_t max_wait = 500;
+
+folly::coro::Task<void>
+postprocess_schedule() {
+  
+  folly::ThreadWheelTimekeeper tk;
+
+  const auto wait_data = [&baton]() -> folly::coro::Task<void> {
+    // folly::coro::timeout cancels the task on timeout.
+    // Baton is not cancellable, so we attach a callback to resume the coroutine.
+    auto token = co_await folly::coro::co_current_cancellation_token;
+    folly::CancellationCallback cb(token, [&baton] { baton.post(); });
+    co_await baton; // Wait data
+  };
+
+
+  uint64_t n_timeouts = 0;
+  uint64_t n_process = 0;
+
+  while(true) {
+    try {
+      co_await folly::coro::timeout(
+        wait_data(),
+        std::chrono::milliseconds{ max_wait },
+        &tk);
+      ++n_process;
+    } catch (const folly::FutureTimeout&) {
+      // timeout = true;
+      std::cout << "Timeout " << ++n_timeouts << std::endl;
+    }
+    baton.reset();
+  }
+
+
+  co_return;
+}
+
+int
+main(int /*argc*/, char** /*argv[]*/)
+{
+  std::atomic<bool> run_marker;
+
+
+
+
+  // A sleepy worker thread
+  std::jthread sleepy_worker(
+    [&baton](std::stop_token stoken)
+    {
+
+      while(!stoken.stop_requested()) {
+        baton.post();
+      }
+        // for (int i = 10; i; --i)
+        // {
+        //     std::this_thread::sleep_for(300ms);
+        //     if (stoken.stop_requested())
+        //     {
+        //         print("Sleepy worker is requested to stop\n");
+        //         return;
+        //     }
+        //     print("Sleepy worker goes back to sleep\n");
+        // }
+    });
+
+    // std::cout << "Sleeping for 3s" << std::endl;
+
+    // std::this_thread::sleep_for(3s);
+
+    std::cout << "Starting the coroutine" << std::endl;
+    folly::coro::blockingWait(postprocess_schedule());
+
+    std::cout << "Requesting stop" << std::endl;
+    // sleepy_worker.request_stop();
+    std::cout << "Thread stopped" << std::endl;
+
+
+  return 0;
+} // NOLINT(readability/fn_size)

From ed37033963f9a8b58b772ca50cb9c629a22c6810 Mon Sep 17 00:00:00 2001
From: Alessandro Thea <alessandro.thea@cern.ch>
Date: Mon, 20 Oct 2025 12:07:40 +0200
Subject: [PATCH 11/14] Adding new test app to makefile

---
 CMakeLists.txt | 1 +
 1 file changed, 1 insertion(+)

diff --git a/CMakeLists.txt b/CMakeLists.txt
index f0069e9..19bf7ab 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -76,6 +76,7 @@ daq_add_application(datahandlinglibs_test_bufferedfilewriter test_bufferedfilewr
 daq_add_application(datahandlinglibs_test_bufferedfilereader test_bufferedfilereader_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 daq_add_application(datahandlinglibs_test_skiplist test_skiplist_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 daq_add_application(datahandlinglibs_test_composite_key test_composite_key_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
+daq_add_application(datahandlinglibs_test_co_timeout test_co_timeout_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 
 ##############################################################################
 # Unit Tests

From 6148b2110039608cda91283c71b6a94be708670d Mon Sep 17 00:00:00 2001
From: Alessandro Thea <alessandro.thea@cern.ch>
Date: Mon, 20 Oct 2025 12:57:12 +0200
Subject: [PATCH 12/14] Logging pp timeout at run start for testing

---
 include/datahandlinglibs/models/detail/DataHandlingModel.hxx | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index c6d19d6..1c661bf 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -326,6 +326,8 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule()
 {
 
   TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
+  TLOG() << "***** Starting post-process coroutine with timout " << m_post_processing_delay_max_wait << " *****";
+
 
   PostprocessScheduleAlgorithm sched_algo{ *m_latency_buffer_impl,
                                            *m_raw_processor_impl,

From e7df040faecdfbd474660b59d17c7436975e53aa Mon Sep 17 00:00:00 2001
From: Alessandro Thea <alessandro.thea@cern.ch>
Date: Wed, 12 Nov 2025 14:20:47 +0100
Subject: [PATCH 13/14] Re-arranged postprocessing timeout handling

---
 .../concepts/RawDataProcessorConcept.hpp      |  3 +++
 .../models/DataHandlingModel.hpp              | 26 +++++++++++++++----
 .../models/TaskRawDataProcessorModel.hpp      |  3 +++
 .../models/detail/DataHandlingModel.hxx       |  3 ---
 4 files changed, 27 insertions(+), 8 deletions(-)

diff --git a/include/datahandlinglibs/concepts/RawDataProcessorConcept.hpp b/include/datahandlinglibs/concepts/RawDataProcessorConcept.hpp
index 1ecb69c..1313057 100644
--- a/include/datahandlinglibs/concepts/RawDataProcessorConcept.hpp
+++ b/include/datahandlinglibs/concepts/RawDataProcessorConcept.hpp
@@ -47,6 +47,9 @@ class RawDataProcessorConcept : public opmonlib::MonitorableObject
   virtual void preprocess_item(ReadoutType* item) = 0;
   //! Postprocess one element
   virtual void postprocess_item(const ReadoutType* item) = 0;
+  //! Handle postprocess timeout event
+  virtual void invoke_postprocess_schedule_timeout_policy(std::uint64_t accumulated_timeout_ticks) = 0;
+
 };
 
 } // namespace datahandlinglibs
diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index 5e5e8cb..241e9bc 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -150,13 +150,27 @@ class DataHandlingModel : public DataHandlingConcept
       , m_processed_up_to{}
       , m_last_post_proc_time{ std::chrono::system_clock::now() }
       , m_consecutive_timeouts{ 0 }
-      , m_max_wait_in_ticks{ post_processing_delay_max_wait * 62500 }
+      , m_max_wait_in_ticks{ post_processing_delay_max_wait * 62500 } // FIXME: hardcoded clock frequency
     {
     }
 
+    // High-level interface
+    // Schedule deferred post-processing and notify timeout expiration to the processor
+    int run(bool timeout) {
+      int processed = this->do_run(timeout);
+
+      if (timeout) {
+        timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;  
+        m_raw_processor_impl.invoke_postprocess_schedule_timeout_policy(timeout_accumulated);
+      }
+
+      return processed;
+    }
+
+
     // Deferral of the post processing, to allow elements being reordered in the LB
     // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value
-    int run(bool timeout)
+    int do_run(bool timeout)
     {
       if (m_latency_buffer_impl.occupancy() == 0) {
         TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (empty buffer)";
@@ -169,7 +183,7 @@ class DataHandlingModel : public DataHandlingConcept
         m_first_cycle = false;
         TLOG() << "***** First pass post processing *****";
       }
-
+      
       // Get the LB boundaries
       auto tail = m_latency_buffer_impl.back();
       auto newest_ts = tail->get_timestamp();
@@ -177,7 +191,9 @@ class DataHandlingModel : public DataHandlingConcept
       timestamp_t end_win_ts = 0;
       std::chrono::time_point<std::chrono::system_clock> now{ std::chrono::system_clock::now() };
 
-      if (timeout) {      
+      if (timeout) {
+        // Return if the last processed timestamp is greater than the newest timestamp
+        // This condition occurs after a timeout
         if (m_processed_up_to.get_timestamp() >= newest_ts + 1) {
           TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (at or past cap)";
           return 0;
@@ -204,7 +220,7 @@ class DataHandlingModel : public DataHandlingConcept
           return 0;
         }
       }
-
+      auto old_process_up_to = m_processed_up_to;
       auto start_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);
       m_processed_up_to.set_timestamp(end_win_ts);
       auto end_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);
diff --git a/include/datahandlinglibs/models/TaskRawDataProcessorModel.hpp b/include/datahandlinglibs/models/TaskRawDataProcessorModel.hpp
index 694d64a..c60616b 100644
--- a/include/datahandlinglibs/models/TaskRawDataProcessorModel.hpp
+++ b/include/datahandlinglibs/models/TaskRawDataProcessorModel.hpp
@@ -80,6 +80,9 @@ class TaskRawDataProcessorModel : public RawDataProcessorConcept<ReadoutType>
   // Registers ReadoutType item pointer to to the post-processing queue
   void postprocess_item(const ReadoutType* item) override;
 
+  // Handle a timeout event 
+  void invoke_postprocess_schedule_timeout_policy(std::uint64_t accumilated_timeout_ticks) override {}
+
   // Registers a pre-processing task to the pre-processor pipeline
   template<typename Task>
   void add_preprocess_task(Task&& task);
diff --git a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
index 1c661bf..4d102b3 100644
--- a/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
+++ b/include/datahandlinglibs/models/detail/DataHandlingModel.hxx
@@ -368,9 +368,6 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule()
       m_sum_payloads += processed;
       m_stats_packet_count += processed;
 
-      if (timeout) {
-        invoke_postprocess_schedule_timeout_policy();
-      }      
     }
   }
 }

From fb18b44a4b3dd386b95a86149592413fcaface00 Mon Sep 17 00:00:00 2001
From: Alessandro Thea <alessandro.thea@cern.ch>
Date: Wed, 12 Nov 2025 15:04:07 +0100
Subject: [PATCH 14/14] Removing unused variable

---
 include/datahandlinglibs/models/DataHandlingModel.hpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/include/datahandlinglibs/models/DataHandlingModel.hpp b/include/datahandlinglibs/models/DataHandlingModel.hpp
index 241e9bc..6711390 100644
--- a/include/datahandlinglibs/models/DataHandlingModel.hpp
+++ b/include/datahandlinglibs/models/DataHandlingModel.hpp
@@ -220,7 +220,7 @@ class DataHandlingModel : public DataHandlingConcept
           return 0;
         }
       }
-      auto old_process_up_to = m_processed_up_to;
+
       auto start_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);
       m_processed_up_to.set_timestamp(end_win_ts);
       auto end_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);