secs-gem/apps/secs_gemd.cpp

// secs_gemd — the SECS/GEM daemon.
//
// Runs the GEM engine (EquipmentRuntime + the default handlers) on a background
// thread, owning the HSMS link to the host, and exposes the name-based gRPC
// API (proto/secsgem/v1/equipment.proto) so a tool's software — in any language
// — can drive the equipment without linking C++ or knowing SEMI.
//
// Operations (Phase E):
//   --grpc      defaults to 127.0.0.1:50051 — NEVER expose the unauthenticated
//               API on the equipment LAN. For same-host tool software prefer a
//               Unix domain socket: --grpc unix:///run/secs_gemd.sock
//   --metrics   Prometheus endpoint port (0 = disabled). Gauges: HSMS link
//               state, control state, spool depth.
//   SIGTERM/SIGINT: graceful shutdown — gRPC server drains (open streams are
//               cancelled), then the engine stops cleanly, so a supervised
//               restart (systemd / docker stop) never kills the spool journal
//               mid-write. Exit code 0.
//
// Deployment recipe: deploy/secs_gemd.service (systemd) and docs/42.

#include <grpcpp/grpcpp.h>

#include <asio.hpp>
#include <chrono>
#include <csignal>
#include <cstdint>
#include <functional>
#include <iostream>
#include <memory>
#include <string>

#include "secsgem/daemon/equipment_service.hpp"
#include "secsgem/gem/default_handlers.hpp"
#include "secsgem/gem/runtime.hpp"
#include "secsgem/metrics/prometheus.hpp"

namespace gem = secsgem::gem;
namespace metrics = secsgem::metrics;
using namespace std::chrono_literals;

namespace {
std::string arg(int argc, char** argv, const std::string& key, const std::string& def) {
  for (int i = 1; i + 1 < argc; ++i)
    if (key == argv[i]) return argv[i + 1];
  return def;
}
}  // namespace

int main(int argc, char** argv) {
  const std::string hsms_port    = arg(argc, argv, "--port", "5000");
  const std::string grpc_addr    = arg(argc, argv, "--grpc", "127.0.0.1:50051");
  const std::string cfgdir       = arg(argc, argv, "--config-dir", "data");
  const std::string spool_dir    = arg(argc, argv, "--spool-dir", "");
  const uint16_t metrics_port    =
      static_cast<uint16_t>(std::stoi(arg(argc, argv, "--metrics", "0")));

  gem::EquipmentRuntime::Config cfg;
  cfg.equipment_yaml     = cfgdir + "/equipment.yaml";
  cfg.control_state_yaml = cfgdir + "/control_state.yaml";
  cfg.process_job_yaml   = cfgdir + "/process_job_state.yaml";
  cfg.control_job_yaml   = cfgdir + "/control_job_state.yaml";
  cfg.port = static_cast<uint16_t>(std::stoi(hsms_port));
  cfg.spool_dir = spool_dir;
  cfg.log = [](const std::string& m) { std::cout << "[gemd] " << m << std::endl; };

  std::unique_ptr<gem::EquipmentRuntime> rt;
  try {
    rt = std::make_unique<gem::EquipmentRuntime>(cfg);
  } catch (const std::exception& e) {
    std::cerr << "[gemd] config error: " << e.what() << std::endl;
    return 1;
  }
  gem::register_default_handlers(*rt);

  // ---- observability (before run_async: observers must land first) -------
  std::shared_ptr<metrics::Registry> registry;
  std::shared_ptr<metrics::PrometheusServer> exporter;
  std::shared_ptr<asio::steady_timer> gauge_timer;
  if (metrics_port != 0) {
    registry = std::make_shared<metrics::Registry>();
    registry->describe("secsgem_link_selected",
                       "1 when an HSMS session is SELECTED, else 0",
                       metrics::MetricType::Gauge);
    registry->describe("secsgem_control_state",
                       "E30 control state (0=EquipOffline 1=AttemptOnline "
                       "2=HostOffline 3=OnlineLocal 4=OnlineRemote)",
                       metrics::MetricType::Gauge);
    registry->describe("secsgem_spool_depth", "Queued spool messages",
                       metrics::MetricType::Gauge);
    rt->add_link_observer([registry](bool selected) {
      registry->set_gauge("secsgem_link_selected", selected ? 1.0 : 0.0);
    });
    rt->add_control_state_observer(
        [registry](gem::ControlState, gem::ControlState to, gem::ControlEvent) {
          registry->set_gauge("secsgem_control_state", static_cast<double>(to));
        });
    registry->set_gauge("secsgem_link_selected", 0.0);
    registry->set_gauge("secsgem_control_state",
                        static_cast<double>(rt->control_state()));
    // Spool depth: sampled on the io thread (the model's owner).
    gauge_timer = std::make_shared<asio::steady_timer>(rt->io());
    auto tick = std::make_shared<std::function<void(std::error_code)>>();
    *tick = [registry, gauge_timer, tick, raw = rt.get()](std::error_code ec) {
      if (ec) return;
      registry->set_gauge("secsgem_spool_depth",
                          static_cast<double>(raw->model().spool.size()));
      gauge_timer->expires_after(5s);
      gauge_timer->async_wait(*tick);
    };
    gauge_timer->expires_after(1s);
    gauge_timer->async_wait(*tick);
    exporter = std::make_shared<metrics::PrometheusServer>(
        rt->io(), metrics_port, registry);
    exporter->start();
  }

  // Construct the service before starting the io thread: its constructor
  // snapshots the name->id/format maps and registers its observers.
  secsgem::daemon::EquipmentService service(*rt);
  rt->run_async();  // engine + HSMS link on a background thread

  grpc::ServerBuilder builder;
  builder.AddListeningPort(grpc_addr, grpc::InsecureServerCredentials());
  builder.RegisterService(&service);
  std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
  if (!server) {
    std::cerr << "[gemd] failed to start gRPC server on " << grpc_addr << std::endl;
    return 1;
  }

  // Graceful shutdown: the signal lands on the io thread, which only nudges
  // the gRPC server; Wait() then returns on the MAIN thread, which tears the
  // engine down (rt->stop() joins the io thread — it must not run on it).
  // The deadline cancels open Subscribe/WatchHealth streams instead of
  // waiting for clients to hang up.
  asio::signal_set signals(rt->io(), SIGINT, SIGTERM);
  signals.async_wait([&server, &rt](std::error_code ec, int signo) {
    if (ec) return;
    rt->log("signal " + std::to_string(signo) + ": shutting down");
    server->Shutdown(std::chrono::system_clock::now() + 2s);
  });

  std::cout << "[gemd] gRPC on " << grpc_addr
            << (metrics_port ? "; metrics on :" + std::to_string(metrics_port) +
                                   "/metrics"
                             : "")
            << "; HSMS equipment on :" << hsms_port << std::endl;
  server->Wait();
  rt->stop();
  std::cout << "[gemd] stopped cleanly" << std::endl;
  return 0;
}