Preserve UAP ping phase and add configurable retry scheduling

.gitignore

@@ -66,3 +66,6 @@ projects/intel_x86/linux/gcc/aether-client-cpp
 /build-windows*/
 *.log
 
+
+.artifacts/
+

CMakeLists.txt

@@ -64,6 +64,7 @@ option(AE_BUILD_TESTS "Build tests" ${AE_ROOT_PORJECT})
 option(AE_BUILD_ANDROID_SMOKE "Build Android NDK smoke shared library and runner" Off)
 option(AE_ADDRESS_SANITIZE "Enable address sanitizer" Off)
 option(AE_NO_STRIP_ALL "Do not apply --strip_all, useful for bloaty and similar tools " Off)
+option(AE_ENABLE_PING_TEST_FAULTS "Enable test-only ping request/response fault injection" Off)
 
 set(UTM_ID "0" CACHE STRING "User Tracking Measurement ID, must be a uint32 value")
 set(USER_CONFIG "" CACHE PATH "Path to user provided configuration header file")
@@ -81,6 +82,7 @@ message(STATUS "Aether build options:
    AE_BUILD_ANDROID_SMOKE=${AE_BUILD_ANDROID_SMOKE}
    AE_ADDRESS_SANITIZE=${AE_ADDRESS_SANITIZE}
    AE_NO_STRIP_ALL=${AE_NO_STRIP_ALL}
+   AE_ENABLE_PING_TEST_FAULTS=${AE_ENABLE_PING_TEST_FAULTS}
    UTM_ID=${UTM_ID}
    USER_CONFIG=${USER_CONFIG}
    FS_INIT=${FS_INIT}
@@ -268,6 +270,9 @@ endif()
 if (AE_FILTRATION)
   target_compile_definitions(${TARGET_NAME} PUBLIC "AE_FILTRATION=1")
 endif()
+if (AE_ENABLE_PING_TEST_FAULTS)
+  target_compile_definitions(${TARGET_NAME} PUBLIC "AE_ENABLE_PING_TEST_FAULTS=1")
+endif()
 
 # for debug purposes only, set registration server ip address
 if(NOT "${AE_REG_CLOUD_ADDR}" STREQUAL "")
@@ -320,7 +325,7 @@ target_compile_options(${TARGET_NAME} PRIVATE
 target_compile_options(${TARGET_NAME} PUBLIC
   $<$<CXX_COMPILER_ID:MSVC>:
       /wd4100 /wd4101 /wd4127 /wd4244 /wd4324
-      /wd4456 /wd4459 /wd4714
+      /wd4456 /wd4459 /wd4702 /wd4714
       >
 )
 
@@ -386,6 +391,10 @@ if(AE_BUILD_EXAMPLES)
   add_subdirectory(examples/capi/oddity)
   add_subdirectory(examples/benches/send_message_delays)
   add_subdirectory(examples/benches/send_messages_bandwidth)
+  add_subdirectory(examples/benches/aether_uap_delivery_timing_bench)
+  add_subdirectory(examples/aether_uap_peer_deadline_test)
+  add_subdirectory(examples/aether_uap_ping_retry_window_test)
+  add_subdirectory(examples/aether_uap_1s_timing_characterization)
 endif()
 
 if(AE_BUILD_TESTS)

aether/CMakeLists.txt

@@ -71,7 +71,9 @@ list(APPEND aether_srcs
             "ae_actions/ping.cpp"
             "ae_actions/check_access_for_send_message.cpp"
             "ae_actions/telemetry.cpp"
-            "ae_actions/select_client.cpp")
+            "ae_actions/select_client.cpp"
+            "ae_actions/query_peer_receive_schedule.cpp"
+            "ae_actions/announce_next_ping_unknown.cpp")
 
 list(APPEND aether_srcs
             "registration/api/client_reg_api_safe.cpp"

aether/actions/action_pool.h

@@ -18,6 +18,7 @@
 #define AETHER_ACTIONS_ACTION_POLL_H_
 
 #include <variant>
+#include <vector>
 #include <type_traits>
 
 #include "aether/warning_disable.h"
@@ -59,11 +60,12 @@ class ActionPool : public etl::pool<T, Capacity> {
 
  private:
   void Destroy(T* p) {
-    ts_ = ac_.scheduler().Task([&, p]() { base_t::template destroy<T>(p); });
+    destroy_tasks_.push_back(
+        ac_.scheduler().Task([this, p]() { base_t::template destroy<T>(p); }));
   }
 
   AC ac_;
-  TaskSubscription ts_;
+  std::vector<TaskSubscription> destroy_tasks_;
 };
 
 template <ActionContext AC, typename... T, std::size_t Capacity>
@@ -96,11 +98,12 @@ class ActionPool<AC, std::variant<T...>, Capacity>
 
  private:
   void Destroy(Action* p) {
-    ts_ = ac_.scheduler().Task([&, p]() { base_t::destroy(p); });
+    destroy_tasks_.push_back(
+        ac_.scheduler().Task([this, p]() { base_t::destroy(p); }));
   }
 
   AC ac_;
-  TaskSubscription ts_;
+  std::vector<TaskSubscription> destroy_tasks_;
 };
 }  // namespace ae
 

aether/ae_actions/announce_next_ping_unknown.cpp

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2026 Aethernet Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "aether/ae_actions/announce_next_ping_unknown.h"
+
+#include <cassert>
+
+#include "aether/client.h"
+#include "aether/cloud_connections/ping_cloud_servers.h"
+#include "aether/config.h"
+
+namespace ae {
+
+AnnounceNextPingUnknown::AnnounceNextPingUnknown(AeContext const& ae_context,
+                                                 Client& client)
+    : ae_context_{ae_context}, client_{&client} {
+  start_sub_ = ae_context_.scheduler().Task([this]() { Start(); });
+  if (!start_sub_) {
+    assert(false && "Task allocation failed");
+    Fail(static_cast<int>(AnnounceNextPingUnknownError::kAnnounceFailed));
+  }
+}
+
+AnnounceNextPingUnknown::~AnnounceNextPingUnknown() { finished_ = true; }
+
+AnnounceNextPingUnknown::ResultEvent::Subscriber
+AnnounceNextPingUnknown::result_event() noexcept {
+  return EventSubscriber{result_event_};
+}
+
+void AnnounceNextPingUnknown::Start() {
+  if (finished_ || client_ == nullptr) {
+    return;
+  }
+#if AE_ENABLE_PING
+  (void)client_->cloud_connection();
+  auto* pings = client_->ping_cloud_servers();
+  if (pings == nullptr) {
+    Fail(static_cast<int>(AnnounceNextPingUnknownError::kNoPingManager));
+    return;
+  }
+  announce_sub_ = pings->announce_event().Subscribe(
+      [this](Result<std::monostate, int> const& res) {
+        if (!res) {
+          Fail(res.error() == 0
+                   ? static_cast<int>(
+                         AnnounceNextPingUnknownError::kAnnounceFailed)
+                   : res.error());
+          return;
+        }
+        CompleteOk();
+      });
+  pings->BeginAnnounceUnknown();
+#else
+  Fail(static_cast<int>(AnnounceNextPingUnknownError::kPingDisabled));
+#endif
+}
+
+void AnnounceNextPingUnknown::CompleteOk() {
+  if (finished_) {
+    return;
+  }
+  finished_ = true;
+  result_event_.Emit(Ok{std::monostate{}});
+  Finish();
+}
+
+void AnnounceNextPingUnknown::Fail(int code) {
+  if (finished_) {
+    return;
+  }
+  finished_ = true;
+  result_event_.Emit(Error{code});
+  Finish();
+}
+
+}  // namespace ae

aether/ae_actions/announce_next_ping_unknown.h

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2026 Aethernet Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef AETHER_AE_ACTIONS_ANNOUNCE_NEXT_PING_UNKNOWN_H_
+#define AETHER_AE_ACTIONS_ANNOUNCE_NEXT_PING_UNKNOWN_H_
+
+#include "aether/config.h"
+
+#include <variant>
+
+#include "aether-miscpp/types/result.h"
+
+#include "aether/ae_context.h"
+#include "aether/actions/action.h"
+#include "aether/events/event_subscription.h"
+#include "aether/events/events.h"
+#include "aether/tasks/details/task_subsctiption.h"
+
+namespace ae {
+class Client;
+
+enum class AnnounceNextPingUnknownError : int {
+  kPingDisabled = 1,
+  kNoPingManager = 2,
+  kAnnounceFailed = 3,
+};
+
+class AnnounceNextPingUnknown final : public Action {
+ public:
+  using ResultEvent = Event<void(Result<std::monostate, int>)>;
+
+  AnnounceNextPingUnknown(AeContext const& ae_context, Client& client);
+  ~AnnounceNextPingUnknown() override;
+
+  AE_CLASS_NO_COPY_MOVE(AnnounceNextPingUnknown)
+
+  ResultEvent::Subscriber result_event() noexcept;
+
+ private:
+  void Start();
+  void CompleteOk();
+  void Fail(int code);
+
+  AeContext ae_context_;
+  Client* client_{nullptr};
+  ResultEvent result_event_;
+  Subscription announce_sub_;
+  TaskSubscription start_sub_;
+  bool finished_{false};
+};
+
+}  // namespace ae
+
+#endif  // AETHER_AE_ACTIONS_ANNOUNCE_NEXT_PING_UNKNOWN_H_

aether/ae_actions/ping.cpp

@@ -25,7 +25,11 @@
 #  include "aether/server.h"
 
 #  include "aether/cloud_connections/cloud_server_connection.h"
+#  include "aether/cloud_connections/ping_schedule_guard.h"
 #  include "aether/work_cloud_api/work_server_api/authorized_api.h"
+#  if AE_ENABLE_PING_TEST_FAULTS
+#    include "aether/ae_actions/ping_test_faults.h"
+#  endif
 
 #  include "aether/ae_actions/ae_actions_tele.h"
 
@@ -89,16 +93,39 @@ void Ping::Start(TimePoint current_time) {
   }
   state_ = RequestState::kPending;
 
+#if AE_ENABLE_PING_TEST_FAULTS
+  if (test_fault_mode_ ==
+      static_cast<std::uint8_t>(PingFaultMode::kDropRequest)) {
+    request_start_ = current_time;
+    timeout_sub_ = ae_context_.scheduler().DelayedTask(
+        [this]() { PingResponseTimeout(RequestId{}); },
+        current_time + timeout_);
+    if (state_ == RequestState::kPending && !timeout_sub_) {
+      AE_TELE_ERROR(
+          kPingTimeoutError,
+          "Ping timeout task allocation failed server id {} request {}",
+          server_id_, RequestId{});
+      state_ = RequestState::kFinished;
+      ResetRequestSubscriptions();
+      result_event_.Emit(PingResult{Error{5}});
+    }
+    return;
+  }
+#endif
+
   auto& write_action = cc->AuthorizedApiCall(
       SubApi{[this, current_time](ApiContext<AuthorizedApi>& auth_api) {
-        auto next_ping_hint_ms = static_cast<std::uint64_t>(
-            std::chrono::duration_cast<std::chrono::milliseconds>(
-                next_ping_hint_)
-                .count());
-        auto rx_window_ms = static_cast<std::uint64_t>(
-            std::chrono::duration_cast<std::chrono::milliseconds>(rx_window_)
-                .count());
-
+        auto next_ping_hint_ms =
+            next_ping_hint_.count() == 0
+                ? std::int64_t{0}
+                : FloorDurationToPositiveInt64Ms(next_ping_hint_);
+        auto rx_window_ms = CeilDurationToSaturatedInt64Ms(rx_window_);
+
+        // ping() is the full schedule contract: nextConnectMsDuration and
+        // rxWindowMs. Do not follow with set_next_read_delay(interval).
+#if AE_ENABLE_PING_TEST_FAULTS
+        PingTestFaults::Instance().OnAuthPing();
+#endif
         auto pong_promise = auth_api->ping(next_ping_hint_ms, rx_window_ms);
         auto req_id = pong_promise.request_id();
 
@@ -110,6 +137,13 @@ void Ping::Start(TimePoint current_time) {
 
         auto wait_result_sub =
             pong_promise.Subscribe([this, req_id](auto&& res) {
+#if AE_ENABLE_PING_TEST_FAULTS
+              PingTestFaults::Instance().OnProtocolResponse();
+              if (test_fault_mode_ ==
+                  static_cast<std::uint8_t>(PingFaultMode::kIgnoreResponse)) {
+                return;
+              }
+#endif
               if (res) {
                 PingResponse(req_id);
               } else {

aether/ae_actions/ping.h

@@ -21,6 +21,7 @@
 
 #if AE_ENABLE_PING
 
+#  include <cstdint>
 #  include <variant>
 
 #  include "aether-miscpp/types/result.h"
@@ -55,6 +56,10 @@ class Ping {
 
   void Start(TimePoint current_time);
 
+#if AE_ENABLE_PING_TEST_FAULTS
+  void ApplyTestFault(std::uint8_t mode) noexcept { test_fault_mode_ = mode; }
+#endif
+
  private:
   void PingResponse(RequestId request_id);
   void PingResponseError(RequestId request_id, std::int32_t error_code);
@@ -82,6 +87,9 @@ class Ping {
 
   ResultEvent result_event_;
   RequestState state_{RequestState::kCreated};
+#if AE_ENABLE_PING_TEST_FAULTS
+  std::uint8_t test_fault_mode_{0};
+#endif
 };
 }  // namespace ae
 #endif  // AE_ENABLE_PING