secs-gem/tests/test_thread_safety.cpp

// Thread-safety contract test.
//
// EquipmentDataModel is single-threaded by design — there are zero
// locks anywhere in the store hierarchy.  The contract documented in
// INTEGRATION.md §3 is: all access (reads from the dispatcher, writes
// from the application) must run on the io_context that drives the
// HSMS connection.  Cross-thread updates marshal through `asio::post`.
//
// This test exercises the canonical pattern: N producer threads post
// sensor updates onto an io_context that one worker thread runs.  All
// the actual reads/writes against the model happen on the worker.
// We assert (a) no updates are lost, (b) all values arrive, and
// (c) the final state is internally consistent.  A passing run isn't
// proof of race-freedom under ThreadSanitizer, but it nails down the
// pattern that customers should follow and catches obvious regressions
// (e.g. someone adding a "convenience" cross-thread mutator).

#include <doctest/doctest.h>

#include <asio.hpp>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <thread>
#include <vector>

#include "secsgem/gem/data_model.hpp"
#include "secsgem/secs2/item.hpp"

using namespace secsgem;
using namespace std::chrono_literals;

TEST_CASE("Threading: cross-thread updates land via asio::post") {
  asio::io_context io;
  auto work = asio::make_work_guard(io);

  // Pre-register the SVID on the worker thread so the producers'
  // set_value finds it.  Done synchronously before workers start.
  gem::EquipmentDataModel model;
  model.svids.add({/*id=*/100, "ChamberPressure", "Torr",
                   secs2::Item::f4(0.0f)});

  std::thread worker([&] { io.run(); });

  constexpr int kProducers = 4;
  constexpr int kUpdatesPer = 250;
  std::atomic<int> applied{0};

  std::vector<std::thread> producers;
  for (int p = 0; p < kProducers; ++p) {
    producers.emplace_back([&, p] {
      for (int i = 0; i < kUpdatesPer; ++i) {
        const float reading = static_cast<float>(p * 1000 + i);
        asio::post(io.get_executor(), [&model, &applied, reading] {
          // This block runs on the worker thread — same thread as any
          // dispatcher would.  No race possible.
          model.svids.set_value(100, secs2::Item::f4(reading));
          applied.fetch_add(1, std::memory_order_relaxed);
        });
      }
    });
  }
  for (auto& t : producers) t.join();

  // Drain — every posted update must run before we tear down.
  while (applied.load(std::memory_order_relaxed) < kProducers * kUpdatesPer) {
    std::this_thread::sleep_for(1ms);
  }

  work.reset();
  worker.join();

  CHECK(applied.load() == kProducers * kUpdatesPer);

  // Final read must also happen on the io_context thread per the
  // contract; we asio::post a final read into a fresh io and pull the
  // result back out.  This proves the contract scales to read-side
  // marshalling too.
  asio::io_context io2;
  std::optional<secs2::Item> last;
  asio::post(io2, [&] { last = model.svids.value(100); });
  io2.run();

  REQUIRE(last.has_value());
  // The final value depends on the asio::post ordering across
  // producers, but the SVID must hold *some* F4 we wrote (i.e., the
  // store didn't corrupt the variant).
  CHECK(last->format() == secs2::Format::F4);
}

TEST_CASE("Threading: posted alarm toggles never lose set/clear pairs") {
  // Mirrors the realistic case where two distinct sensor threads each
  // toggle their own alarm.  Every set+clear pair must be observable;
  // a lost set or clear would leave the alarm registry in the wrong
  // state.
  asio::io_context io;
  auto work = asio::make_work_guard(io);

  gem::EquipmentDataModel model;
  model.alarms.add({/*id=*/1, "Chiller", /*category=*/4});
  model.alarms.add({/*id=*/2, "Door",    /*category=*/1});

  std::thread worker([&] { io.run(); });

  constexpr int kCycles = 200;
  std::atomic<int> applied{0};
  auto cycle = [&](uint32_t alid) {
    for (int i = 0; i < kCycles; ++i) {
      asio::post(io.get_executor(), [&model, &applied, alid] {
        model.alarms.set_active(alid);
        applied.fetch_add(1, std::memory_order_relaxed);
      });
      asio::post(io.get_executor(), [&model, &applied, alid] {
        model.alarms.clear_active(alid);
        applied.fetch_add(1, std::memory_order_relaxed);
      });
    }
  };

  std::thread t1(cycle, 1u);
  std::thread t2(cycle, 2u);
  t1.join();
  t2.join();

  while (applied.load(std::memory_order_relaxed) < 4 * kCycles) {
    std::this_thread::sleep_for(1ms);
  }

  work.reset();
  worker.join();

  CHECK(applied.load() == 4 * kCycles);

  // Both alarms end inactive (last op in each cycle is set_inactive).
  asio::io_context io2;
  bool a1_active = true, a2_active = true;
  asio::post(io2, [&] {
    a1_active = model.alarms.active(1);
    a2_active = model.alarms.active(2);
  });
  io2.run();

  CHECK_FALSE(a1_active);
  CHECK_FALSE(a2_active);
}