Fix BinaryRecording file indexing when streams are disabled

Author: jsiegle

Source/Processors/RecordNode/BinaryFormat/BinaryRecording.cpp

@@ -110,12 +110,9 @@ void BinaryRecording::openFiles (File rootFolder, int experimentNumber, int reco
             singleStreamJSON.clear();
         }
 
-        LOGD ("Recording channel: ", channelInfo->getName(), " from stream ", streamId, " with global index ", globalIndex, " and local index ", indexWithinStream);
-
         m_fileIndexes.set (ch, streamIndex);
         m_channelIndexes.set (ch, indexWithinStream++);
 
-
         DynamicObject::Ptr singleChannelJSON = new DynamicObject();
 
         singleChannelJSON->setProperty ("channel_name", channelInfo->getName());
@@ -787,6 +784,7 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
                                                 int numSamples,
                                                 int fileIndex)
 {
+    ignoreUnused (fileIndex);
     if (numSamples == 0 || numChannels == 0)
         return;
 
@@ -821,8 +819,12 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
         m_batchIntBufferPtrs.resize (numChannels);
     }
 
+    // Resolve the file index from recorded channel mapping.
+    // This is robust even when some source streams have zero recorded channels.
+    const int resolvedFileIndex = m_fileIndexes[writeChannels[0]];
+
     // Get file and validate
-    if (fileIndex < 0 || fileIndex >= m_continuousFiles.size() || ! m_continuousFiles[fileIndex])
+    if (resolvedFileIndex < 0 || resolvedFileIndex >= m_continuousFiles.size() || ! m_continuousFiles[resolvedFileIndex])
         return;
 
     // Setup scale factors and output buffer pointers for each channel
@@ -844,12 +846,12 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
     // Get starting sample position (all channels in a stream have same position)
     uint64 startPos = m_samplesWritten[writeChannels[0]];
 
-    //LOGD("BinaryRecording::writeContinuousDataBatch: Writing ", numSamples, " samples for ", numChannels, " channels at position ", startPos, " to file index ", fileIndex);
+    //LOGD("BinaryRecording::writeContinuousDataBatch: Writing ", numSamples, " samples for ", numChannels, " channels at position ", startPos, " to file index ", resolvedFileIndex);
 
     // Try batch interleaving if we have all channels for this file
     // The file's channel count is determined by the stream's channel count
     // If we have a partial batch, fall back to per-channel writes
-    bool useBatchWrite = m_continuousFiles[fileIndex] != nullptr && m_continuousFiles[fileIndex]->writeChannelBatch (startPos, m_batchIntBufferPtrs.data(), numChannels, numSamples);
+    bool useBatchWrite = m_continuousFiles[resolvedFileIndex] != nullptr && m_continuousFiles[resolvedFileIndex]->writeChannelBatch (startPos, m_batchIntBufferPtrs.data(), numChannels, numSamples);
 
     if (! useBatchWrite)
     {
@@ -858,7 +860,7 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
         {
             int writeChannel = writeChannels[i];
             int channelIdx = m_channelIndexes[writeChannel];
-            m_continuousFiles[fileIndex]->writeChannel (
+            m_continuousFiles[resolvedFileIndex]->writeChannel (
                 startPos,
                 channelIdx,
                 m_batchIntBufferPtrs[i],
@@ -884,11 +886,11 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
             /* Generate sequential sample numbers using SIMD-optimized fill */
             SIMDConverter::fillSequentialInt64 (reinterpret_cast<int64_t*> (m_sampleNumberBuffer.getData()), baseSampleNumber, numSamples);
 
-            m_dataTimestampFiles[fileIndex]->writeData (m_sampleNumberBuffer, numSamples * sizeof (int64));
-            m_dataTimestampFiles[fileIndex]->increaseRecordCount (numSamples);
+            m_dataTimestampFiles[resolvedFileIndex]->writeData (m_sampleNumberBuffer, numSamples * sizeof (int64));
+            m_dataTimestampFiles[resolvedFileIndex]->increaseRecordCount (numSamples);
 
-            m_dataSyncTimestampFiles[fileIndex]->writeData (timestampBuffer, numSamples * sizeof (double));
-            m_dataSyncTimestampFiles[fileIndex]->increaseRecordCount (numSamples);
+            m_dataSyncTimestampFiles[resolvedFileIndex]->writeData (timestampBuffer, numSamples * sizeof (double));
+            m_dataSyncTimestampFiles[resolvedFileIndex]->increaseRecordCount (numSamples);
             break;
         }
     }

Source/Processors/RecordNode/BinaryFormat/SIMDConverter.h

@@ -31,6 +31,7 @@
 
 #include <cstdint>
 #include <string>
+#include "../../../TestableExport.h"
 
 /**
  * SIMD-optimized float-to-int16 conversion utilities.
@@ -47,7 +48,7 @@
  * (FloatVectorOperations::copyWithMultiply + AudioDataConverters::convertFloatToInt16LE)
  * into a single pass, eliminating the intermediate buffer and improving cache utilization.
  */
-class SIMDConverter
+class TESTABLE SIMDConverter
 {
 public:
     /**

Tests/Processors/RecordNodeTests.cpp

@@ -820,6 +820,97 @@ TEST_F(RecordNodeTests, WriteContinuousDataBatch_EmptyBatch_ZeroSamples_NoDataWr
     ASSERT_EQ(persistedData.size(), 0u);
 }
 
+class MultiStream_RecordNodeTests : public testing::Test {
+protected:
+    void SetUp() override {
+        tester = std::make_unique<ProcessorTester>(TestSourceNodeBuilder
+                                                   (FakeSourceNodeParams{
+            4,      // channels per stream
+            30000,  // sample rate
+            1.0f,   // bitVolts
+            3       // streams
+        }));
+
+        parentRecordingDir = std::filesystem::temp_directory_path() / "record_node_multi_stream_tests";
+        if (std::filesystem::exists(parentRecordingDir)) {
+            std::filesystem::remove_all(parentRecordingDir);
+        }
+        std::filesystem::create_directory(parentRecordingDir);
+
+        tester->setRecordingParentDirectory(parentRecordingDir.string());
+        processor = tester->createProcessor<RecordNode>(Plugin::Processor::RECORD_NODE);
+    }
+
+    void TearDown() override {
+        std::error_code ec;
+        std::filesystem::remove_all(parentRecordingDir, ec);
+    }
+
+    bool getRecordingPath(std::filesystem::path* outPath) const {
+        auto dirIter = std::filesystem::directory_iterator(parentRecordingDir);
+        if (dirIter == std::filesystem::directory_iterator()) {
+            return false;
+        }
+
+        auto recordingDir = dirIter->path();
+        std::stringstream ss;
+        ss << "Record Node " << processor->getNodeId();
+        *outPath = recordingDir / ss.str() / "experiment1" / "recording1";
+        return true;
+    }
+
+    std::unique_ptr<ProcessorTester> tester;
+    RecordNode* processor = nullptr;
+    std::filesystem::path parentRecordingDir;
+};
+
+TEST_F(MultiStream_RecordNodeTests, DisabledMiddleStream_DoesNotBlockSubsequentStreamWrites) {
+    // Disable all channels on middle stream; keep streams 0 and 2 enabled.
+    auto streams = processor->getDataStreams();
+    ASSERT_EQ(streams.size(), 3);
+
+    Array<var> noChannelsSelected;
+    auto* middleStreamMask = static_cast<MaskChannelsParameter*>(streams[1]->getParameter("channels"));
+    ASSERT_NE(middleStreamMask, nullptr);
+    middleStreamMask->setNextValue(noChannelsSelected, false);
+    processor->updateSettings();
+
+    tester->startAcquisition(true);
+
+    AudioBuffer<float> inputBuffer(12, 256); // 3 streams * 4 channels/stream
+    for (int ch = 0; ch < inputBuffer.getNumChannels(); ch++) {
+        for (int s = 0; s < inputBuffer.getNumSamples(); s++) {
+            inputBuffer.setSample(ch, s, static_cast<float>(ch * 1000 + s));
+        }
+    }
+
+    for (int i = 0; i < 5; i++) {
+        tester->processBlock(processor, inputBuffer);
+    }
+
+    tester->stopAcquisition();
+
+    std::filesystem::path recordingPath;
+    ASSERT_TRUE(getRecordingPath(&recordingPath));
+    auto continuousPath = recordingPath / "continuous";
+    ASSERT_TRUE(std::filesystem::exists(continuousPath));
+
+    std::vector<std::filesystem::path> continuousDatFiles;
+    for (const auto& entry : std::filesystem::recursive_directory_iterator(continuousPath)) {
+        if (entry.is_regular_file() && entry.path().filename() == "continuous.dat") {
+            continuousDatFiles.push_back(entry.path());
+        }
+    }
+
+    // Only streams with recorded channels should have files: stream 0 and stream 2.
+    ASSERT_EQ(continuousDatFiles.size(), 2u);
+
+    for (const auto& datFile : continuousDatFiles) {
+        ASSERT_GT(std::filesystem::file_size(datFile), 0u)
+            << "Expected non-empty data file for recorded stream: " << datFile.string();
+    }
+}
+
 /**
  * Test that writeContinuousDataBatch correctly resizes internal buffers.
  * This tests the buffer reallocation logic by writing progressively larger blocks.