Synnax v0.52.7 (#2052)

Author: emilbon99

arc/cpp/runtime/constant/constant.h

@@ -41,9 +41,13 @@ class Constant : public node::Node {
         this->initialized = true;
         const auto &o = state.output(0);
         const auto &o_time = state.output_time(0);
-        o->resize(1);
+        if (o->data_type().is_variable())
+            *o = x::telem::Series(this->value);
+        else {
+            o->resize(1);
+            o->set(0, this->value);
+        }
         o_time->resize(1);
-        o->set(0, this->value);
         o_time->set(0, x::telem::TimeStamp::now());
         ctx.mark_changed(ir::default_output_param);
         return x::errors::NIL;

arc/cpp/runtime/constant/constant_test.cpp

@@ -241,6 +241,38 @@ TEST(ConstantTest, TimestampOutputOnFirstNext) {
     EXPECT_GT(output_time->at<int64_t>(0), 0);
 }
 
+/// @brief Test that string values are correctly output.
+TEST(ConstantTest, StringValueIsOutput) {
+    const std::string val = "hello";
+    TestSetup setup(types::Kind::String, val);
+    Constant node(setup.make_node(), val, x::telem::STRING_T);
+
+    auto ctx = make_context();
+    ASSERT_NIL(node.next(ctx));
+
+    auto checker = setup.make_node();
+    const auto &output = checker.output(0);
+    EXPECT_EQ(output->size(), 1);
+    EXPECT_EQ(output->at<std::string>(0), val);
+}
+
+/// @brief Test that reset() allows the string value to be output again.
+TEST(ConstantTest, StringResetAllowsValueToBeOutputAgain) {
+    const std::string val = "hello";
+    TestSetup setup(types::Kind::String, val);
+    Constant node(setup.make_node(), val, x::telem::STRING_T);
+
+    auto ctx = make_context();
+    node.next(ctx);
+    node.reset();
+    ASSERT_NIL(node.next(ctx));
+
+    auto checker = setup.make_node();
+    const auto &output = checker.output(0);
+    EXPECT_EQ(output->size(), 1);
+    EXPECT_EQ(output->at<std::string>(0), val);
+}
+
 /// @brief Test that reset produces a new timestamp on subsequent next().
 TEST(ConstantTest, ResetProducesNewTimestamp) {
     TestSetup setup(types::Kind::F32, 42.5f);

arc/cpp/runtime/telem/telem.h

@@ -79,6 +79,20 @@ class On : public node::Node {
         return x::errors::NIL;
     }
 
+    /// @brief Advances the high water mark to the current channel alignment,
+    /// ensuring that when a stage is (re-)activated it only responds to
+    /// data that arrives after activation rather than stale pre-existing data.
+    void reset() override {
+        node::Node::reset();
+        auto [data, index_data, ok] = state.read_chan(channel_key);
+        if (!ok || data.series.empty()) return;
+        const auto &last = data.series.back();
+        const auto lower = last.alignment;
+        const auto upper_val = lower.uint64() + (last.size() > 0 ? last.size() - 1 : 0);
+        const auto upper = x::telem::Alignment(upper_val + 1);
+        if (upper.uint64() > high_water_mark.uint64()) high_water_mark = upper;
+    }
+
     [[nodiscard]] bool is_output_truthy(const std::string &param_name) const override {
         return state.is_output_truthy(param_name);
     }

arc/cpp/runtime/telem/telem_test.cpp

@@ -431,6 +431,115 @@ TEST(OnTest, NextSkipsOnAlignmentMismatch) {
     EXPECT_EQ(call_count, 0);
 }
 
+/// @brief Test that Reset() advances the watermark so stale pre-reset data does not
+/// re-trigger the source node, while new data written after reset does.
+TEST(OnTest, ResetAdvancesHighWaterMarkPreventingStaleData) {
+    ir::Param output_param;
+    output_param.name = ir::default_output_param;
+    output_param.type = types::Type(types::Kind::F32);
+
+    ir::Node ir_node;
+    ir_node.key = "source";
+    ir_node.type = "on";
+    ir_node.outputs.params.push_back(output_param);
+
+    ir::Param channel_config;
+    channel_config.name = "channel";
+    channel_config.value = static_cast<uint32_t>(10);
+    ir_node.config.params.push_back(channel_config);
+
+    ir::IR ir;
+    ir.nodes.push_back(ir_node);
+
+    state::Config cfg{.ir = ir, .channels = {{10, x::telem::FLOAT32_T, 11}}};
+    state::State s(cfg, errors::noop_handler);
+
+    Factory factory;
+    auto state_node = ASSERT_NIL_P(s.node("source"));
+    auto node = ASSERT_NIL_P(
+        factory.create(node::Config(ir, ir_node, std::move(state_node)))
+    );
+
+    // Ingest data at alignment 0.
+    x::telem::Frame frame1(2);
+    auto d1 = x::telem::Series(1.0f);
+    d1.alignment = x::telem::Alignment(0);
+    auto t1 = x::telem::Series(static_cast<int64_t>(100));
+    t1.alignment = x::telem::Alignment(0);
+    frame1.emplace(10, std::move(d1));
+    frame1.emplace(11, std::move(t1));
+    s.ingest(frame1);
+
+    // Reset: advances the watermark to 1 (past alignment 0).
+    node->reset();
+
+    // Next should NOT fire: data at alignment 0 is now below the watermark.
+    bool changed = false;
+    auto ctx = node::Context{
+        .elapsed = x::telem::SECOND,
+        .mark_changed = [&changed](const std::string &) { changed = true; },
+        .report_error = [](const x::errors::Error &) {},
+        .activate_stage = [] {},
+    };
+    ASSERT_NIL(node->next(ctx));
+    EXPECT_FALSE(changed) << "stale pre-reset data should not trigger the source";
+
+    // Ingest new data at alignment 10 (well above the watermark of 1).
+    x::telem::Frame frame2(2);
+    auto d2 = x::telem::Series(2.0f);
+    d2.alignment = x::telem::Alignment(10);
+    auto t2 = x::telem::Series(static_cast<int64_t>(200));
+    t2.alignment = x::telem::Alignment(10);
+    frame2.emplace(10, std::move(d2));
+    frame2.emplace(11, std::move(t2));
+    s.ingest(frame2);
+
+    ASSERT_NIL(node->next(ctx));
+    EXPECT_TRUE(changed) << "data written after reset should trigger the source";
+}
+
+/// @brief Test that Reset() is a no-op when the channel has no data.
+TEST(OnTest, ResetIsNoOpWithNoData) {
+    ir::Param output_param;
+    output_param.name = ir::default_output_param;
+    output_param.type = types::Type(types::Kind::F32);
+
+    ir::Node ir_node;
+    ir_node.key = "source";
+    ir_node.type = "on";
+    ir_node.outputs.params.push_back(output_param);
+
+    ir::Param channel_config;
+    channel_config.name = "channel";
+    channel_config.value = static_cast<uint32_t>(10);
+    ir_node.config.params.push_back(channel_config);
+
+    ir::IR ir;
+    ir.nodes.push_back(ir_node);
+
+    state::Config cfg{.ir = ir, .channels = {{10, x::telem::FLOAT32_T, 11}}};
+    state::State s(cfg, errors::noop_handler);
+
+    Factory factory;
+    auto state_node = ASSERT_NIL_P(s.node("source"));
+    auto node = ASSERT_NIL_P(
+        factory.create(node::Config(ir, ir_node, std::move(state_node)))
+    );
+
+    // Reset with no ingested data should not crash.
+    node->reset();
+
+    bool changed = false;
+    auto ctx = node::Context{
+        .elapsed = x::telem::SECOND,
+        .mark_changed = [&changed](const std::string &) { changed = true; },
+        .report_error = [](const x::errors::Error &) {},
+        .activate_stage = [] {},
+    };
+    ASSERT_NIL(node->next(ctx));
+    EXPECT_FALSE(changed);
+}
+
 /// @brief Test source node calls mark_changed callback.
 TEST(OnTest, NextCallsMarkChanged) {
     ir::Param output_param;