fix: MXL_ERR_FLOW_INVALID reconnect, nullptr→"" options, null guards

Readers (ndiout, fakesink) now handle MXL_ERR_FLOW_INVALID by releasing
and recreating the flow reader, then realigning the index to current time.
This lets consumer nodes survive a producer restart without exiting.

All mxlCreateInstance/FlowWriter/FlowReader options args changed from
nullptr to "" to match MXL reference implementation style.

mxlReleaseFlowReader calls guarded with null checks so cleanup is safe
when a mid-run reconnect attempt fails.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

nodes/fakesink/main.cpp

@@ -23,9 +23,9 @@ class FakeSinkNode : public dmf::NodeBase {
         log("flow active — starting read");
 
         mxlFlowReader reader{};
-        mxlStatus st = mxlCreateFlowReader(instance(), flow_id.c_str(), nullptr, &reader);
+        mxlStatus st = mxlCreateFlowReader(instance(), flow_id.c_str(), "", &reader);
         if (st != MXL_STATUS_OK) {
-            log("mxlCreateFlowReader failed (status=%d)", st);
+            log("mxlCreateFlowReader failed (%s)", dmf::mxl_status_str(st));
             return;
         }
 
@@ -59,8 +59,18 @@ class FakeSinkNode : public dmf::NodeBase {
                 mxlFlowReaderGetRuntimeInfo(reader, &ri);
                 index = ri.headIndex;
 
+            } else if (st == MXL_ERR_FLOW_INVALID) {
+                log("flow invalidated — reconnecting...");
+                if (reader) mxlReleaseFlowReader(instance(), reader);
+                reader = nullptr;
+                mxlSleepForNs(100'000'000);
+                if (mxlCreateFlowReader(instance(), flow_id.c_str(), "", &reader) == MXL_STATUS_OK) {
+                    log("flow reconnected");
+                    index = mxlGetCurrentIndex(&rate);
+                }
+
             } else {
-                log("unexpected status=%d on index=%llu", st, index);
+                log("unexpected status (%s) on index=%llu", dmf::mxl_status_str(st), index);
                 break;
             }
 
@@ -77,7 +87,7 @@ class FakeSinkNode : public dmf::NodeBase {
 
         log("stopped — total frames=%llu  invalid=%llu  late=%llu",
             frame_count, invalid_count, late_count);
-        mxlReleaseFlowReader(instance(), reader);
+        if (reader) mxlReleaseFlowReader(instance(), reader);
     }
 };
 

nodes/ndiin/main.cpp

@@ -41,8 +41,8 @@ class NDIInNode : public dmf::NodeBase {
         mxlStatus st = mxlCreateFlowWriter(
             instance(),
             dmf::make_video_flow_def(video_flow_id, node_id(), width, height, fps_num, fps_den).c_str(),
-            nullptr, &video_writer, &video_cfg, &created);
+            "", &video_writer, &video_cfg, &created);
-        if (st != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (status=%d)", st); return; }
+        if (st != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(st)); return; }
 
         const uint32_t video_stride = video_cfg.discrete.sliceSizes[0];
         log("video stride=%u B/line  grain=%u B  ring=%u grains",
@@ -69,9 +69,9 @@ class NDIInNode : public dmf::NodeBase {
                 instance(),
                 dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, bit_depth,
                                          fps_num, fps_den).c_str(),
-                nullptr, &audio_writer, &audio_cfg, &created);
+                "", &audio_writer, &audio_cfg, &created);
             if (ast != MXL_STATUS_OK) {
-                log("audio mxlCreateFlowWriter failed (status=%d) — continuing without audio", ast);
+                log("audio mxlCreateFlowWriter failed (%s) — continuing without audio", dmf::mxl_status_str(ast));
                 has_audio = false;
             } else {
                 log("audio channels=%u  buffer=%u samples",
@@ -146,7 +146,7 @@ class NDIInNode : public dmf::NodeBase {
                     grain.validSlices = grain.totalSlices;
                     mxlFlowWriterCommitGrain(video_writer, &grain);
                 } else {
-                    log("video OpenGrain failed (status=%d) at index=%llu", st, video_index);
+                    log("video OpenGrain failed (%s) at index=%llu", dmf::mxl_status_str(st), video_index);
                 }
                 video_index = current + 1;
             }

nodes/ndiout/main.cpp

@@ -64,8 +64,8 @@ class NDIOutNode : public dmf::NodeBase {
             log("flow active — starting read");
 
             mxlFlowConfigInfo video_cfg{};
-            mxlStatus vst = mxlCreateFlowReader(instance(), flow_id.c_str(), nullptr, &video_reader);
+            mxlStatus vst = mxlCreateFlowReader(instance(), flow_id.c_str(), "", &video_reader);
-            if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowReader failed (status=%d)", vst); return; }
+            if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowReader failed (%s)", dmf::mxl_status_str(vst)); return; }
             mxlFlowReaderGetConfigInfo(video_reader, &video_cfg);
             video_stride = video_cfg.discrete.sliceSizes[0];
         }
@@ -77,10 +77,11 @@ class NDIOutNode : public dmf::NodeBase {
         int  channels          = 0;
         int  samples_per_frame = 0;
         bool has_audio         = config().contains("audio_flow_id");
+        std::string audio_flow_id;
 
         if (has_audio) {
             const auto audio_flow_info = config().at("audio_flow_id");
-            const auto audio_flow_id   = audio_flow_info.at("id").get<std::string>();
+            audio_flow_id = audio_flow_info.at("id").get<std::string>();
             sample_rate       = audio_flow_info.value("sample_rate", 48000);
             channels          = audio_flow_info.value("channels", 2);
             samples_per_frame = sample_rate * fps_den / fps_num;
@@ -88,9 +89,9 @@ class NDIOutNode : public dmf::NodeBase {
             log("audio flow=%s  %d Hz  %dch  %d samples/frame",
                 audio_flow_id.c_str(), sample_rate, channels, samples_per_frame);
 
-            mxlStatus ast = mxlCreateFlowReader(instance(), audio_flow_id.c_str(), nullptr, &audio_reader);
+            mxlStatus ast = mxlCreateFlowReader(instance(), audio_flow_id.c_str(), "", &audio_reader);
             if (ast != MXL_STATUS_OK) {
-                log("audio mxlCreateFlowReader failed (status=%d) — continuing without audio", ast);
+                log("audio mxlCreateFlowReader failed (%s) — continuing without audio", dmf::mxl_status_str(ast));
                 has_audio = false;
             } else {
                 mxlFlowReaderGetConfigInfo(audio_reader, &audio_cfg);
@@ -188,6 +189,16 @@ class NDIOutNode : public dmf::NodeBase {
                     mxlFlowRuntimeInfo ari{};
                     mxlFlowReaderGetRuntimeInfo(audio_reader, &ari);
                     audio_index = ari.headIndex;
+                } else if (ast == MXL_ERR_FLOW_INVALID) {
+                    log("audio flow invalidated — reconnecting...");
+                    mxlReleaseFlowReader(instance(), audio_reader);
+                    audio_reader = nullptr;
+                    mxlSleepForNs(100'000'000);
+                    if (mxlCreateFlowReader(instance(), audio_flow_id.c_str(), "", &audio_reader) == MXL_STATUS_OK) {
+                        log("audio flow reconnected");
+                        const mxlRational r = {sample_rate, 1};
+                        audio_index = mxlGetCurrentIndex(&r);
+                    }
                 }
             }
 
@@ -223,8 +234,19 @@ class NDIOutNode : public dmf::NodeBase {
                     mxlFlowReaderGetRuntimeInfo(video_reader, &ri);
                     video_index = ri.headIndex;
 
+                } else if (vst == MXL_ERR_FLOW_INVALID) {
+                    log("video flow invalidated — reconnecting...");
+                    mxlReleaseFlowReader(instance(), video_reader);
+                    video_reader = nullptr;
+                    mxlSleepForNs(100'000'000);
+                    if (mxlCreateFlowReader(instance(), flow_id.c_str(), "", &video_reader) == MXL_STATUS_OK) {
+                        log("video flow reconnected");
+                        const mxlRational r = {fps_num, fps_den};
+                        video_index = mxlGetCurrentIndex(&r);
+                    }
+
                 } else {
-                    log("unexpected video status=%d on index=%llu", vst, video_index);
+                    log("unexpected video status (%s) on index=%llu", dmf::mxl_status_str(vst), video_index);
                     break;
                 }
 
@@ -248,8 +270,8 @@ class NDIOutNode : public dmf::NodeBase {
         else
             log("stopped");
 
-        if (has_video) mxlReleaseFlowReader(instance(), video_reader);
+        if (has_video && video_reader) mxlReleaseFlowReader(instance(), video_reader);
-        if (has_audio) mxlReleaseFlowReader(instance(), audio_reader);
+        if (has_audio && audio_reader) mxlReleaseFlowReader(instance(), audio_reader);
     }
 };
 

nodes/testpattern/main.cpp

@@ -27,8 +27,8 @@ class TestPatternNode : public dmf::NodeBase {
         mxlStatus vst = mxlCreateFlowWriter(
             instance(),
             dmf::make_video_flow_def(flow_id, node_id(), width, height, fps_num, fps_den).c_str(),
-            nullptr, &video_writer, &video_cfg, &created);
+            "", &video_writer, &video_cfg, &created);
-        if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (status=%d)", vst); return; }
+        if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(vst)); return; }
 
         const uint32_t video_stride = video_cfg.discrete.sliceSizes[0];
         log("video stride=%u B/line  grain=%u B  ring=%u grains",
@@ -53,9 +53,9 @@ class TestPatternNode : public dmf::NodeBase {
                 instance(),
                 dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, 32,
                                          fps_num, fps_den).c_str(),
-                nullptr, &audio_writer, &audio_cfg, &created);
+                "", &audio_writer, &audio_cfg, &created);
             if (ast != MXL_STATUS_OK) {
-                log("audio mxlCreateFlowWriter failed (status=%d) — continuing without audio", ast);
+                log("audio mxlCreateFlowWriter failed (%s) — continuing without audio", dmf::mxl_status_str(ast));
                 has_audio = false;
             } else {
                 log("audio channels=%u  buffer=%u samples",
@@ -85,7 +85,7 @@ class TestPatternNode : public dmf::NodeBase {
             uint8_t*     video_buf = nullptr;
             vst = mxlFlowWriterOpenGrain(video_writer, video_index, &video_grain, &video_buf);
             if (vst != MXL_STATUS_OK) {
-                log("OpenGrain failed (status=%d), skipping index=%llu", vst, video_index);
+                log("OpenGrain failed (%s) at index=%llu", dmf::mxl_status_str(vst), video_index);
                 video_index++;
                 continue;
             }

shared/FlowDef.hpp

@@ -59,7 +59,7 @@ inline std::string make_audio_flow_def(
         {"format",      "urn:x-nmos:format:audio"},
         {"label",       label},
         {"description", label + " MXL Audio Flow"},
-        {"media_type",  "audio/L" + std::to_string(bit_depth)},
+        {"media_type",  bit_depth == 32 ? "audio/float32" : "audio/L" + std::to_string(bit_depth)},
         {"parents",     json::array()},
         {"grain_rate",  {{"numerator", grain_rate_num}, {"denominator", grain_rate_den}}},
         {"sample_rate", {{"numerator", sample_rate},    {"denominator", 1}}},

shared/NodeBase.hpp

@@ -10,6 +10,35 @@
 
 namespace dmf {
 
+inline const char* mxl_status_str(mxlStatus s) noexcept {
+    switch (s) {
+        case MXL_STATUS_OK:                    return "MXL_STATUS_OK";
+        case MXL_ERR_UNKNOWN:                  return "MXL_ERR_UNKNOWN";
+        case MXL_ERR_FLOW_NOT_FOUND:           return "MXL_ERR_FLOW_NOT_FOUND";
+        case MXL_ERR_OUT_OF_RANGE_TOO_LATE:    return "MXL_ERR_OUT_OF_RANGE_TOO_LATE";
+        case MXL_ERR_OUT_OF_RANGE_TOO_EARLY:   return "MXL_ERR_OUT_OF_RANGE_TOO_EARLY";
+        case MXL_ERR_INVALID_FLOW_READER:      return "MXL_ERR_INVALID_FLOW_READER";
+        case MXL_ERR_INVALID_FLOW_WRITER:      return "MXL_ERR_INVALID_FLOW_WRITER";
+        case MXL_ERR_TIMEOUT:                  return "MXL_ERR_TIMEOUT";
+        case MXL_ERR_INVALID_ARG:              return "MXL_ERR_INVALID_ARG";
+        case MXL_ERR_CONFLICT:                 return "MXL_ERR_CONFLICT";
+        case MXL_ERR_PERMISSION_DENIED:        return "MXL_ERR_PERMISSION_DENIED";
+        case MXL_ERR_FLOW_INVALID:             return "MXL_ERR_FLOW_INVALID";
+        case MXL_ERR_STRLEN:                   return "MXL_ERR_STRLEN";
+        case MXL_ERR_INTERRUPTED:              return "MXL_ERR_INTERRUPTED";
+        case MXL_ERR_NO_FABRIC:                return "MXL_ERR_NO_FABRIC";
+        case MXL_ERR_INVALID_STATE:            return "MXL_ERR_INVALID_STATE";
+        case MXL_ERR_INTERNAL:                 return "MXL_ERR_INTERNAL";
+        case MXL_ERR_NOT_READY:                return "MXL_ERR_NOT_READY";
+        case MXL_ERR_NOT_FOUND:                return "MXL_ERR_NOT_FOUND";
+        case MXL_ERR_EXISTS:                   return "MXL_ERR_EXISTS";
+        case MXL_ERR_UNSUPPORTED_OPERATION:    return "MXL_ERR_UNSUPPORTED_OPERATION";
+        default:                               return "MXL_ERR_UNRECOGNIZED";
+    }
+}
+
+
+
 // Base class for all DMF node binaries.
 //
 // Handles the boilerplate every node needs:
@@ -51,7 +80,7 @@ public:
 
         log("domain=%s", domain_.c_str());
 
-        inst_ = mxlCreateInstance(domain_.c_str(), nullptr);
+        inst_ = mxlCreateInstance(domain_.c_str(), "");
         if (!inst_) {
             log("mxlCreateInstance failed at %s", domain_.c_str());
             return 1;

studio-manager/main.cpp

@@ -10,6 +10,7 @@
 #include <fcntl.h>
 #include <sys/wait.h>
 #include <unistd.h>
+#include <mxl/mxl.h>
 #include <mxl/time.h>
 #include "Signal.hpp"
 #include "FlowGraph.hpp"
@@ -152,6 +153,16 @@ int main(int argc, char* argv[]) {
     }
     fprintf(stderr, "[studio-manager] domain: %s\n", domain.c_str());
 
+    // Clean up stale flow directories left by any previous crashed run
+    {
+        mxlInstance gc = mxlCreateInstance(domain.c_str(), "");
+        if (gc) {
+            mxlGarbageCollectFlows(gc);
+            mxlDestroyInstance(gc);
+            fprintf(stderr, "[studio-manager] garbage collected stale flows\n");
+        }
+    }
+
     // --- Build and launch the pipeline graph ---
     const dmf::FlowGraph graph = build_graph();