fix: use TAI-time-based grain index alignment in passthrough

Instead of chasing headIndex from mxlFlowReaderGetRuntimeInfo (which
points to the NEXT grain to be written, causing perpetual TOO_EARLY),
the passthrough now uses mxlTimestampToIndex + mxlGetNsUntilIndex for
proper timing alignment, matching the pattern used by mxl-gst-sink.

Key changes:
- realign() computes read_index from current TAI time minus 1 grain delay
- process() uses mxlSleepForNs to wait until the target grain is due
- on_add_reader fetches grain_rate from mxlFlowConfigInfo
- Removed separate write_index_ (uses grain_info.index for writer)

nodes/passthrough/src/passthrough_node.cpp

@@ -2,6 +2,7 @@
 
 #include <mxl/flow.h>
 #include <mxl/mxl.h>
+#include <mxl/time.h>
 
 #include <spdlog/spdlog.h>
 
@@ -19,10 +20,14 @@ void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter wr
 void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
     if (port_id == "video_in") {
         reader_ = reader;
-        mxlFlowRuntimeInfo runtime{};
+
-        mxlFlowReaderGetRuntimeInfo(*reader_, &runtime);
+        mxlFlowConfigInfo config{};
-        read_index_ = runtime.headIndex;
+        mxlFlowReaderGetConfigInfo(*reader_, &config);
-        spdlog::info("Passthrough: reader added on video_in, starting at index {}", read_index_);
+        grain_rate_ = config.common.grainRate;
+
+        realign();
+        spdlog::info("Passthrough: reader added on video_in, grain_rate={}/{}, read_index={}",
+                     grain_rate_.numerator, grain_rate_.denominator, read_index_);
     }
 }
 
@@ -40,21 +45,32 @@ void PassthroughNode::on_remove_reader(const std::string& port_id) {
     }
 }
 
+void PassthroughNode::realign() {
+    auto now = mxlGetTime();
+    auto current_index = mxlTimestampToIndex(&grain_rate_, now);
+    constexpr int64_t read_delay_grains = 1;
+    read_index_ = current_index - read_delay_grains;
+    aligned_ = true;
+}
+
 void PassthroughNode::process() {
-    if (!reader_ || !writer_) {
+    if (!reader_ || !writer_ || !aligned_) {
         return;
     }
 
+    auto wait_ns = mxlGetNsUntilIndex(read_index_, &grain_rate_);
+    if (wait_ns > 0 && wait_ns < 500000000ULL) {
+        mxlSleepForNs(wait_ns);
+    }
+
     mxlGrainInfo grain_info{};
     uint8_t* payload = nullptr;
 
     auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload);
     if (status != MXL_STATUS_OK) {
         if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
-            mxlFlowRuntimeInfo runtime{};
+            spdlog::warn("Passthrough: index {} out of range ({}), realigning", read_index_, static_cast<int>(status));
-            mxlFlowReaderGetRuntimeInfo(*reader_, &runtime);
+            realign();
-            spdlog::warn("Passthrough: index {} out of range, jumping to head {}", read_index_, runtime.headIndex);
-            read_index_ = runtime.headIndex;
         }
         return;
     }

nodes/passthrough/src/passthrough_node.hpp

@@ -4,7 +4,6 @@
 
 #include <mxl/flow.h>
 
-#include <unordered_map>
 #include <optional>
 
 namespace dmf_node {
@@ -34,12 +33,14 @@ public:
     nlohmann::json status() const override;
 
 private:
+    void realign();
+
     std::optional<mxlFlowReader> reader_;
     std::optional<mxlFlowWriter> writer_;
-    mxlInstance mxl_instance_ = nullptr;
+    mxlRational grain_rate_{50, 1};
     uint64_t read_index_ = 0;
-    uint64_t write_index_ = 0;
     uint64_t grains_processed_ = 0;
+    bool aligned_ = false;
 };
 
 } // namespace dmf_node