status_poll_policy.cpp

Go to the documentation of this file.

/// @file status_poll_policy.cpp
/// @brief StatusPollPolicy implementation.
/// @ingroup hioc_hub
 
#include "status_poll_policy.h"
 
#include "hub_decisions.h"
 
#include <algorithm>
#include <array>
 
namespace esphome {
namespace home_io_control {
 
namespace {
/// Stamp value for "now": millis() == 0 would read as "never", so it is nudged to 1.
uint32_t stamp(uint32_t now) { return now == 0 ? 1 : now; }
/// True when @p t has been reached at @p now (wrap-safe; @p t is assumed within half the range).
bool reached(uint32_t now, uint32_t t) { return static_cast<int32_t>(now - t) > 0; }
 
/// Forget everything about the predicted-end read: the aim, the count it came from and the mark.
void drop_predicted_end(PollTracking &t) {
  t.settle_update = 0;
  t.last_remaining_s.reset();
  t.stale_reports = 0;
  t.predicted_end_read = false;
}
}  // namespace
 
void StatusPollPolicy::set_interval(const std::string &device_id, uint32_t interval_ms) {
  tracking_[device_id].interval_ms = interval_ms;
}
 
uint32_t StatusPollPolicy::get_interval(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return (it != tracking_.end()) ? it->second.interval_ms : 0;
}
 
void StatusPollPolicy::begin_tracking(const std::string &device_id, uint32_t initial_delay_ms, uint32_t now) {
  state_unknown_.erase(device_id);
  PollTracking &t = tracking_[device_id];
  t.poll_deadline = now + MAX_TRACKED_STATUS_POLL_WINDOW_MS;
  t.next_update = (initial_delay_ms == 0) ? 0 : (now + initial_delay_ms);
  t.status_poll_failures = 0;
  t.auth_poll_failures = 0;
  t.moving_poll_misses = 0;
  t.stop_settle = false;
  t.confirm_stopped = false;
  drop_predicted_end(t);
}
 
void StatusPollPolicy::clear(const std::string &device_id) {
  state_unknown_.erase(device_id);
  auto it = tracking_.find(device_id);
  if (it == tracking_.end())
    return;
  PollTracking &t = it->second;
  t.next_update = 0;
  t.poll_deadline = 0;
  t.status_poll_failures = 0;
  t.auth_poll_failures = 0;
  t.moving_poll_misses = 0;
  t.stop_settle = false;
  t.confirm_stopped = false;
  drop_predicted_end(t);
  // interval_ms is configuration — NOT cleared; remains set for future tracking cycles.
}
 
MovingReportAction StatusPollPolicy::on_moving_report(const std::string &device_id, uint32_t now, bool may_open) {
  if (is_tracking_active(device_id, now))
    return MovingReportAction::FOLLOW;
  if (get_poll_deadline(device_id) != 0) {
    // The window ran out before this report: a final outcome, so it is closed rather than reopened.
    note_ended_(device_id);
    clear(device_id);
    return MovingReportAction::END;
  }
  if (!may_open)
    return MovingReportAction::NONE;
  begin_tracking(device_id, 0, now);
  return MovingReportAction::OPEN;
}
 
void StatusPollPolicy::on_trusted_report(const std::string &device_id, std::optional<uint16_t> remaining_s,
                                         uint32_t now) {
  auto it = tracking_.find(device_id);
  if (it == tracking_.end())
    return;
  PollTracking &t = it->second;
  t.predicted_end_read = false;
  if (!remaining_s.has_value()) {
    drop_predicted_end(t);
    return;
  }
  if (t.last_remaining_s.has_value() && *remaining_s >= *t.last_remaining_s) {
    if (t.stale_reports < UINT8_MAX)
      t.stale_reports++;
  } else {
    t.stale_reports = 0;
  }
  t.last_remaining_s = remaining_s;
  if (t.stale_reports >= PREDICTED_END_STALE_REPORTS) {
    t.settle_update = 0;
    return;
  }
  const uint32_t aim = now + (*remaining_s * 1000U) + PREDICTED_END_MARGIN_MS;
  // A read the window would not live to see is not planned: the progress reads and the window's own
  // end handling stand.
  const bool beyond_window = t.poll_deadline != 0 && static_cast<int32_t>(aim - t.poll_deadline) > 0;
  t.settle_update = beyond_window ? 0 : stamp(aim);
}
 
uint32_t StatusPollPolicy::frozen_count_delay_ms(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  if (it == tracking_.end() || it->second.stale_reports < PREDICTED_END_STALE_REPORTS)
    return 0;
  switch (it->second.stale_reports - PREDICTED_END_STALE_REPORTS) {
    case 0:
      return FROZEN_COUNT_BACKOFF_FIRST_MS;
    case 1:
      return FROZEN_COUNT_BACKOFF_SECOND_MS;
    default:
      return FROZEN_COUNT_BACKOFF_MAX_MS;
  }
}
 
uint32_t StatusPollPolicy::get_settle_update(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return (it != tracking_.end()) ? it->second.settle_update : 0;
}
 
void StatusPollPolicy::set_next_update(const std::string &device_id, uint32_t abs_time) {
  tracking_[device_id].next_update = abs_time;
}
 
uint32_t StatusPollPolicy::get_next_update(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return (it != tracking_.end()) ? it->second.next_update : 0;
}
 
uint32_t StatusPollPolicy::get_poll_deadline(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return (it != tracking_.end()) ? it->second.poll_deadline : 0;
}
 
bool StatusPollPolicy::is_tracking_active(const std::string &device_id, uint32_t now) const {
  auto it = tracking_.find(device_id);
  if (it == tracking_.end())
    return false;
  const PollTracking &t = it->second;
  return t.poll_deadline != 0 && now <= t.poll_deadline;
}
 
uint32_t StatusPollPolicy::on_exchange_failed(const std::string &device_id, bool auth_like, uint32_t now,
                                              bool device_moving) {
  auto it = tracking_.find(device_id);
  if (it == tracking_.end())
    return 0;
  PollTracking &t = it->second;
  // A one-off poll of an untracked device fails here too; only a window that was open can run out.
  const bool was_tracking = t.poll_deadline != 0;
 
  // A device believed travelling keeps its normal cadence for its first misses rather than leave a
  // gap (MOVING_POLL_MISSES_SKIPPED), but never waits longer than the ladder's first step would.
  // Only a miss outside a failure streak is skipped: a streak already runs the ladder.
  if (device_moving && t.moving_poll_misses < MOVING_POLL_MISSES_SKIPPED && t.status_poll_failures == 0 &&
      t.auth_poll_failures == 0 && is_tracking_active(device_id, now)) {
    const uint32_t delay_ms =
        std::min(settle_delay_ms(t.interval_ms, 0, /*cap_for_stop=*/false), STATUS_RETRY_AFTER_FAIL_MS);
    if (now + delay_ms <= t.poll_deadline) {
      t.moving_poll_misses++;
      t.next_update = now + delay_ms;
      return delay_ms;
    }
  }
 
  uint32_t backoff_ms;
  if (auth_like) {
    t.status_poll_failures = 0;
    if (t.auth_poll_failures < UINT8_MAX)
      t.auth_poll_failures++;
    backoff_ms = retry_delay_ms(t.auth_poll_failures, true);
  } else {
    t.auth_poll_failures = 0;
    if (t.status_poll_failures < UINT8_MAX)
      t.status_poll_failures++;
    backoff_ms = retry_delay_ms(t.status_poll_failures, false);
  }
 
  if (is_tracking_active(device_id, now) && now + backoff_ms <= t.poll_deadline) {
    t.next_update = now + backoff_ms;
    return backoff_ms;
  }
 
  if (was_tracking)
    note_ended_(device_id);
  clear(device_id);
  return 0;
}
 
void StatusPollPolicy::clear_failure_streaks(const std::string &device_id) {
  auto it = tracking_.find(device_id);
  if (it == tracking_.end())
    return;
  it->second.status_poll_failures = 0;
  it->second.auth_poll_failures = 0;
  it->second.moving_poll_misses = 0;
}
 
void StatusPollPolicy::mark_stop_settle(const std::string &device_id) {
  auto it = tracking_.find(device_id);
  if (it != tracking_.end())
    it->second.stop_settle = true;
}
 
bool StatusPollPolicy::take_stop_settle(const std::string &device_id) {
  auto it = tracking_.find(device_id);
  if (it == tracking_.end())
    return false;
  const bool marked = it->second.stop_settle;
  it->second.stop_settle = false;
  return marked;
}
 
void StatusPollPolicy::set_confirm_stopped(const std::string &device_id, bool needed) {
  auto it = tracking_.find(device_id);
  if (it != tracking_.end())
    it->second.confirm_stopped = needed;
}
 
bool StatusPollPolicy::take_confirm_stopped(const std::string &device_id) {
  auto it = tracking_.find(device_id);
  if (it == tracking_.end())
    return false;
  const bool marked = it->second.confirm_stopped;
  it->second.confirm_stopped = false;
  return marked;
}
 
bool StatusPollPolicy::is_predicted_end_read(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return it != tracking_.end() && it->second.predicted_end_read;
}
 
bool StatusPollPolicy::take_settling_read(const std::string &device_id) {
  const bool after_stop = take_stop_settle(device_id);
  const bool confirms_stop = take_confirm_stopped(device_id);
  bool predicted_end = false;
  auto it = tracking_.find(device_id);
  if (it != tracking_.end()) {
    predicted_end = it->second.predicted_end_read;
    it->second.predicted_end_read = false;
  }
  return after_stop || confirms_stop || predicted_end;
}
 
uint8_t StatusPollPolicy::get_status_poll_failures(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return (it != tracking_.end()) ? it->second.status_poll_failures : 0;
}
 
uint8_t StatusPollPolicy::get_auth_poll_failures(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return (it != tracking_.end()) ? it->second.auth_poll_failures : 0;
}
 
uint8_t StatusPollPolicy::get_moving_poll_misses(const std::string &device_id) const {
  auto it = tracking_.find(device_id);
  return (it != tracking_.end()) ? it->second.moving_poll_misses : 0;
}
 
namespace {
/// True when @p t is 0 (never) or at least @p span_ms old (wrap-safe).
bool older_than(uint32_t now, uint32_t t, uint32_t span_ms) { return t == 0 || now - t >= span_ms; }
}  // namespace
 
bool StatusPollPolicy::is_remote_poll_pending(const std::string &device_id, uint32_t now) const {
  auto it = foreign_.find(device_id);
  if (it == foreign_.end() || it->second.pending_due_ms == 0)
    return false;
  // Wrap-safe: before the due time `now - due` is a huge unsigned value (a negative distance), so
  // it is still pending; more than the delay past the due time means the "ran" signal was lost.
  const uint32_t since_due = now - it->second.pending_due_ms;
  return since_due > INT32_MAX || since_due <= REMOTE_ACTIVITY_STATUS_POLL_DELAY_MS;
}
 
void StatusPollPolicy::on_remote_poll_ran(const std::string &device_id) {
  auto it = foreign_.find(device_id);
  if (it != foreign_.end())
    it->second.pending_due_ms = 0;
}
 
ForeignAction StatusPollPolicy::on_foreign_frame(const std::string &device_id, ForeignFrameKind kind,
                                                 const std::string &controller, uint8_t cmd, uint32_t now,
                                                 uint32_t last_status_ms) {
  ForeignTraffic &f = foreign_[device_id];
  const bool pending = is_remote_poll_pending(device_id, now);
  const ForeignTraffic before = f;
 
  if (kind == ForeignFrameKind::REQUEST_TO_DEVICE) {
    f.last_request_cmd = cmd;
    f.last_request_from = controller;
    f.last_request_ms = stamp(now);
  }
 
  if (pending) {
    f.pending_due_ms = stamp(now + REMOTE_ACTIVITY_STATUS_POLL_DELAY_MS);
    return ForeignAction::PUSH_PENDING;
  }
 
  bool arm;
  if (kind == ForeignFrameKind::REQUEST_TO_DEVICE) {
    if (decisions::overheard_command_writes(cmd)) {
      arm = true;
    } else {
      arm = older_than(now, last_status_ms, FOREIGN_READ_RESYNC_MS) &&
            older_than(now, f.last_read_poll_ms, FOREIGN_READ_RESYNC_MS);
      if (arm)
        f.last_read_poll_ms = stamp(now);
    }
  } else {
    // A challenge answers the request just before it: ignore it only when that request, from this
    // same controller, was a read.
    const bool follows_read = before.last_request_ms != 0 && before.last_request_from == controller &&
                              now - before.last_request_ms <= FOREIGN_CHALLENGE_AFTER_REQUEST_MS &&
                              !decisions::overheard_command_writes(before.last_request_cmd);
    arm = !follows_read;
  }
 
  if (!arm)
    return ForeignAction::IGNORE;
  f.pending_due_ms = stamp(now + REMOTE_ACTIVITY_STATUS_POLL_DELAY_MS);
  return ForeignAction::ARM;
}
 
std::optional<std::string> StatusPollPolicy::pop_due_device(uint32_t now) {
  for (auto &pair : tracking_) {
    PollTracking &t = pair.second;
    const bool predicted_end_due = t.settle_update != 0 && reached(now, t.settle_update);
    const bool progress_due = t.next_update != 0 && now > t.next_update;
    if (!predicted_end_due && !progress_due)
      continue;
    if (!is_tracking_active(pair.first, now)) {
      if (t.poll_deadline != 0)
        note_ended_(pair.first);
      clear(pair.first);
      continue;
    }
    if (predicted_end_due) {
      t.settle_update = 0;
      t.next_update = 0;
      t.predicted_end_read = true;
      return pair.first;
    }
    // A progress read right in front of the predicted-end read is not worth its own exchange.
    if (t.settle_update != 0 &&
        static_cast<int32_t>(t.settle_update - now) <= static_cast<int32_t>(PREDICTED_END_DROPS_PROGRESS_WITHIN_MS)) {
      t.next_update = 0;
      continue;
    }
    t.next_update = 0;
    return pair.first;
  }
  return std::nullopt;
}
 
std::optional<std::string> StatusPollPolicy::pop_ended_device() {
  if (ended_.empty())
    return std::nullopt;
  std::string id = ended_.front();
  ended_.erase(ended_.begin());
  return id;
}
 
void StatusPollPolicy::on_linked_remote_activity() { catchup_burst_open_ = true; }
 
void StatusPollPolicy::on_exchange_ended(uint32_t now, uint32_t delay_ms) {
  if (!catchup_burst_open_)
    return;
  if (delay_ms == 0) {
    catchup_burst_open_ = false;
    catchup_due_ms_ = 0;
    return;
  }
  catchup_due_ms_ = stamp(now + delay_ms);
}
 
void StatusPollPolicy::mark_state_unknown(const std::string &device_id) { state_unknown_.insert(device_id); }
 
bool StatusPollPolicy::take_state_unknown(const std::string &device_id) { return state_unknown_.erase(device_id) != 0; }
 
bool StatusPollPolicy::take_due_catchup(uint32_t now) {
  // Wrap-safe: before the due time `now - due` is a huge unsigned value (a negative distance).
  if (!catchup_burst_open_ || catchup_due_ms_ == 0 || now - catchup_due_ms_ > INT32_MAX)
    return false;
  catchup_burst_open_ = false;
  catchup_due_ms_ = 0;
  return true;
}
 
void StatusPollPolicy::note_ended_(const std::string &device_id) {
  if (std::find(ended_.begin(), ended_.end(), device_id) == ended_.end())
    ended_.push_back(device_id);
}
 
namespace {
 
/// The two backoff ladders, one delay per consecutive failure; the last entry repeats.
constexpr std::array<uint32_t, 5> SILENT_RETRY_LADDER_MS{
    STATUS_RETRY_AFTER_FAIL_MS, STATUS_RETRY_AFTER_FAIL_STEP2_MS, STATUS_RETRY_AFTER_FAIL_STEP3_MS,
    STATUS_RETRY_AFTER_FAIL_STEP4_MS, STATUS_RETRY_AFTER_FAIL_MAX_MS};
constexpr std::array<uint32_t, 4> AUTH_RETRY_LADDER_MS{
    STATUS_AUTH_RETRY_AFTER_FAIL_MS, STATUS_AUTH_RETRY_AFTER_FAIL_STEP2_MS, STATUS_AUTH_RETRY_AFTER_FAIL_STEP3_MS,
    STATUS_AUTH_RETRY_AFTER_FAIL_MAX_MS};
 
/// The delay for the @p consecutive_failures-th failure in a row (0 is read as 1) on @p ladder.
template<size_t N> uint32_t ladder_step(const std::array<uint32_t, N> &ladder, uint8_t consecutive_failures) {
  const size_t index = consecutive_failures <= 1 ? 0 : consecutive_failures - 1;
  return ladder[std::min(index, N - 1)];
}
 
}  // namespace
 
uint32_t StatusPollPolicy::retry_delay_ms(uint8_t consecutive_failures, bool auth_like) {
  return auth_like ? ladder_step(AUTH_RETRY_LADDER_MS, consecutive_failures)
                   : ladder_step(SILENT_RETRY_LADDER_MS, consecutive_failures);
}
 
}  // namespace home_io_control
}  // namespace esphome