feat(platform): add heartbeat module for replica health monitoring

[lterrac]

Draft — depends on #53. Do not merge until PR #53 (feat/platform-config) is merged into main.

What this adds

platform/include/modules/heartbeat/module.hpp — a module that monitors replica liveness via periodic heartbeat messages.

• Replicas call addOutput and the module's service() loop sends a heartbeat pulse at every interval
• The coordinator (or any monitor) calls addInput and the dispatcher routes incoming heartbeats to heartbeatMessageHandler
• Health transitions through three states: unknown (no heartbeat yet) → healthy (recent pulse received) → unhealthy (tolerance window exceeded without a pulse)
• A healthChangeCallback fires on every state transition, giving the coordinator or autoscaler an immediate signal to react

Dependencies

PR	Status	Required
#53 — configuration types, channelController, service, module base	open	✅ must merge first

Test plan

• PR feat(platform): add configuration types, module layer, and host control plane #53 merged into main
• ninja -C build compiles clean
• Manual: co-located coordinator+replica with heartbeat wired, kill replica → callback fires with unhealthy

🤖 Generated with Claude Code

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 4e4ec0cd-abc6-49d2-a255-843a7eb314dc

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[gemini-code-assist]

Code Review

This pull request introduces the host-side control plane for PyPTO Serving, adding documentation, examples, and core platform modules (such as broadcast deployment, channel controller, service, and heartbeat), alongside Python bindings for the coordinator-replica bridge. The code review identified critical concurrency and correctness issues, including data races in the heartbeat module's shared maps and a race condition in the coordinator's response handler when processing multiple outputs concurrently. Additionally, feedback was provided to address potential undefined behavior from unchecked topology collections in runtime initialization and a resource leak vulnerability in the replica module's cleanup guard.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Data race vulnerability: heartbeatMessageHandler and checkTimeouts run concurrently on different TaskR worker threads and both read/write to _health and _lastSeen maps without any synchronization. Additionally, dynamic calls to addInput or removeInput can modify these maps and cause undefined behavior or crashes due to concurrent map rehashing. A mutex must be introduced to protect all accesses to these shared maps, and getHealthSnapshot should return the map by value under the lock to prevent data races on the returned reference.

[gemini-code-assist]

Race condition and correctness bug: In replicaResponseHandler, _jobsMutex is released before satisfying the job's output dependencies and checking job->isFinished(). If a task has multiple outputs, the replica sends separate messages for each output, which can be processed concurrently on different threads. This leads to concurrent writes to the same Job object and a race where both threads can see isFinished() == true, causing the completion callback to be called multiple times and _jobs / _replicaJobs to be erased multiple times. To prevent this, the dependency satisfaction, finished check, and map erasure must be performed atomically under the _jobsMutex lock, while releasing the lock before invoking the user completion callback to avoid deadlocks.

    std::shared_ptr<Job> job = nullptr;
    bool isFinished = false;
    std::vector<JobOutput> outputs;

    {
      std::lock_guard lock(_jobsMutex);
      if (!_jobs.contains(jobId)) return;
      job = _jobs.at(jobId);

      auto &dependency = job->getOutputDependency(outputName);
      if (dependency.isSatisfied()) HICR_THROW_RUNTIME("Output dependency '%s' for job %lu is already satisfied.", outputName.c_str(), jobId);
      dependency.storeData(message.getData(), message.getSize());
      dependency.setSatisfied(true);

      if (job->isFinished())
      {
        isFinished = true;
        outputs.reserve(job->getOutputDependencies().size());
        for (auto &[_, outputDependency] : job->getOutputDependencies())
        {
          outputs.push_back(JobOutput{
            .name = outputDependency.getName(),
            .data = outputDependency.getData(),
          });
        }
        _jobs.erase(jobId);

        std::lock_guard lockReplica(_replicaJobsMutex);
        _replicaJobs.erase(replicaId);
      }
    }

    if (!isFinished) return;

[gemini-code-assist]

Defensive programming check: topology.getDevices() and device->getMemorySpaceList() can be empty, which would cause undefined behavior when dereferencing .begin(). Add explicit checks to ensure these collections are not empty before accessing their elements.

  const auto topology       = hwlocTopologyManager.queryTopology();
  if (topology.getDevices().empty()) { HICR_THROW_RUNTIME("No devices found in topology."); }
  auto       device         = *topology.getDevices().begin();
  auto       memorySpaces   = device->getMemorySpaceList();
  if (memorySpaces.empty()) { HICR_THROW_RUNTIME("No memory spaces found on device."); }
  runtime.bufferMemorySpace = *memorySpaces.begin();

[gemini-code-assist]

Resource leak vulnerability: guard.dismissed = true; is set before pushing the response messages to the output channels. If pushMessageLocking throws an exception (e.g., due to a closed channel or network error), the CleanupGuard destructor will not free the allocated output memory slots, leading to a memory leak. The guard should only be dismissed after all output messages have been successfully dispatched.

    for (const auto &output : context.getOutputs())
    {
      const auto response = message_t(
        output.data == nullptr ? nullptr : static_cast<const uint8_t *>(output.data->getPointer()), output.data == nullptr ? 0 : output.data->getSize(), activeJob.metadata);
      _outputChannels.at(output.name)->pushMessageLocking(response);
    }

    // All validation and dispatch passed. Dismiss the guard so outputs stay alive.
    guard.dismissed = true;

    // Input slots are no longer needed once outputs have been dispatched.
    freeInputSlots(activeJob);