core(M2): машина состояний сессии Ayla LAN + mock-модуль + интеграционные сценарии
- session.{hpp,cpp}: state machine (idle/registering/online/recovering/
offline/key_error); httpd-обработчики key_exchange (200/426/412, re-key
прозрачно), commands (одна команда, 206/200, envelope, глобальный seq_no),
datapoint (unpack -> PropertyEvent / 401+тишина 50с для re-key-восстановления);
сессионный поток: local_reg POST?dsn/PUT (local_ip_for), keep-alive, backoff
x1.6->60с, 503->offline/NoSlot, activation-timeout->recovering, delete_session
с ожиданием выдачи; очередь с coalescing + batch; телеметрия; колбэки из
двух потоков с задокументированным контрактом; буферы datapoint-пути в Impl.
- platform: local_ip_for (UDP-connect) posix+esp-idf; стек httpd 24576
(переполнение 16КБ поймано gdb на Release).
- mock_ac.py: мок-модуль, stdlib-only чистый python AES-256 (свёрстан с
pycryptodome); сценарии: 503, no-poll, rekey-every, stale-gap (эмуляция
'вернувшегося' приложения), fail-pushes (битая подпись), garbage-pushes
(обрыв блока), break-outbound (исходящий десинк -> модуль ре-кает на
local_reg, как probe1-3), push-every, fail-first-ke.
- session_runner + test_session_mock.py: 9 сценариев через ctest, включая
самосинхронизацию CBC и восстановление после исходящего десинка.
- Прибор AP-WC1E: активация <=1с; re-key семантика ИСПРАВЛЕНА по живым
тестам: re-key при зазоре local_reg >= ~44-50с (не по возрасту сессии!);
при честном keep-alive 15с сессия стабильна без re-key; PROTOCOL/LEGACY/
PLAN обновлены; восстановление = тишина >порога + возврат.
- CI: 7/7 x3 (gcc-Rel, gcc-ASan/UBSan, clang); ESP-IDF esp32 build complete.
Ревью под-агентом: 2 круга (стек httpd, залипание состояний, dangling cfg,
физика десинка) — APPROVED.
This commit is contained in:
@@ -121,7 +121,7 @@ void parse_target(const char* full_target, HttpRequest* req) {
|
||||
}
|
||||
}
|
||||
|
||||
constexpr uint32_t kHttpdThreadStack = 8192;
|
||||
constexpr uint32_t kHttpdThreadStack = 24576; // commands-путь: envelope+crypto ~10КБ поверх буферов запроса
|
||||
constexpr uint32_t kAcceptPollMs = 100;
|
||||
constexpr uint32_t kClientRxTimeoutMs = 30000;
|
||||
constexpr uint32_t kClientTxTimeoutMs = 10000;
|
||||
|
||||
@@ -159,6 +159,35 @@ int tcp_connect(const char* host, uint16_t port, uint32_t timeout_ms) {
|
||||
return fd;
|
||||
}
|
||||
|
||||
bool local_ip_for(const char* host, char* out, size_t out_cap) {
|
||||
struct addrinfo hints {};
|
||||
hints.ai_family = AF_INET;
|
||||
hints.ai_socktype = SOCK_DGRAM;
|
||||
struct addrinfo* list = nullptr;
|
||||
if (lwip_getaddrinfo(host, "80", &hints, &list) != 0 || list == nullptr) {
|
||||
return false;
|
||||
}
|
||||
int fd = lwip_socket(AF_INET, SOCK_DGRAM, 0);
|
||||
if (fd < 0) {
|
||||
lwip_freeaddrinfo(list);
|
||||
return false;
|
||||
}
|
||||
bool ok = lwip_connect(fd, list->ai_addr, list->ai_addrlen) == 0;
|
||||
struct sockaddr_in local {};
|
||||
socklen_t slen = sizeof(local);
|
||||
if (ok && lwip_getsockname(fd, reinterpret_cast<struct sockaddr*>(&local),
|
||||
&slen) == 0) {
|
||||
const char* s = inet_ntop(AF_INET, &local.sin_addr, out,
|
||||
static_cast<socklen_t>(out_cap));
|
||||
ok = s != nullptr;
|
||||
} else {
|
||||
ok = false;
|
||||
}
|
||||
lwip_freeaddrinfo(list);
|
||||
lwip_close(fd);
|
||||
return ok;
|
||||
}
|
||||
|
||||
long tcp_send(int fd, const void* buf, size_t len) {
|
||||
const uint8_t* p = static_cast<const uint8_t*>(buf);
|
||||
size_t done = 0;
|
||||
|
||||
@@ -43,6 +43,9 @@ uint16_t tcp_local_port(int fd);
|
||||
int tcp_accept(int listen_fd, uint32_t* peer_ip, uint16_t* peer_port);
|
||||
// Подключается к host:port (host — DNS-имя или dotted-quad) с таймаутом.
|
||||
int tcp_connect(const char* host, uint16_t port, uint32_t timeout_ms);
|
||||
// Локальный IP-адрес (dotted) интерфейса, которым достигается host
|
||||
// (без реального трафика: UDP connect). Для local_reg.
|
||||
bool local_ip_for(const char* host, char* out, size_t out_cap);
|
||||
long tcp_send(int fd, const void* buf, size_t len); // >0 / -1
|
||||
// Блокирующее чтение; 0 — EOF, -1 — ошибка/таймаут.
|
||||
long tcp_recv(int fd, void* buf, size_t len);
|
||||
|
||||
@@ -60,7 +60,12 @@ bool thread_create(void (*fn)(void*), void* ctx, const char* name,
|
||||
pthread_t tid;
|
||||
pthread_attr_t attr;
|
||||
pthread_attr_init(&attr);
|
||||
if (stack_bytes > 0) pthread_attr_setstacksize(&attr, stack_bytes);
|
||||
if (stack_bytes > 0) {
|
||||
if (pthread_attr_setstacksize(&attr, stack_bytes) != 0) {
|
||||
// glibc отвергает < PTHREAD_STACK_MIN; остаётся дефолт (больше — не меньше)
|
||||
// логируем только: ядро запрашивает >= PTHREAD_STACK_MIN.
|
||||
}
|
||||
}
|
||||
int rc = pthread_create(&tid, &attr, thread_trampoline, start);
|
||||
pthread_attr_destroy(&attr);
|
||||
if (rc != 0) {
|
||||
@@ -168,6 +173,35 @@ int tcp_connect(const char* host, uint16_t port, uint32_t timeout_ms) {
|
||||
return fd;
|
||||
}
|
||||
|
||||
bool local_ip_for(const char* host, char* out, size_t out_cap) {
|
||||
struct addrinfo hints {};
|
||||
hints.ai_family = AF_INET;
|
||||
hints.ai_socktype = SOCK_DGRAM;
|
||||
struct addrinfo* list = nullptr;
|
||||
if (::getaddrinfo(host, "80", &hints, &list) != 0 || list == nullptr) {
|
||||
return false;
|
||||
}
|
||||
int fd = ::socket(AF_INET, SOCK_DGRAM, 0);
|
||||
if (fd < 0) {
|
||||
::freeaddrinfo(list);
|
||||
return false;
|
||||
}
|
||||
bool ok = ::connect(fd, list->ai_addr, list->ai_addrlen) == 0;
|
||||
struct sockaddr_in local {};
|
||||
socklen_t slen = sizeof(local);
|
||||
if (ok && ::getsockname(fd, reinterpret_cast<struct sockaddr*>(&local),
|
||||
&slen) == 0) {
|
||||
const char* s = inet_ntop(AF_INET, &local.sin_addr, out,
|
||||
static_cast<socklen_t>(out_cap));
|
||||
ok = s != nullptr;
|
||||
} else {
|
||||
ok = false;
|
||||
}
|
||||
::freeaddrinfo(list);
|
||||
::close(fd);
|
||||
return ok;
|
||||
}
|
||||
|
||||
long tcp_send(int fd, const void* buf, size_t len) {
|
||||
const uint8_t* p = static_cast<const uint8_t*>(buf);
|
||||
size_t done = 0;
|
||||
|
||||
909
src/ayla/session.cpp
Normal file
909
src/ayla/session.cpp
Normal file
@@ -0,0 +1,909 @@
|
||||
#include "ayla/session.hpp"
|
||||
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <mutex>
|
||||
#include <new>
|
||||
|
||||
#include "ayla/envelope.hpp"
|
||||
#include "ayla/httpc.hpp"
|
||||
#include "ayla/json.hpp"
|
||||
#include "ayla/log.hpp"
|
||||
#include "ayla/platform/platform.hpp"
|
||||
|
||||
namespace fgl::ayla {
|
||||
|
||||
namespace {
|
||||
|
||||
constexpr uint32_t kLoopTickMs = 20;
|
||||
constexpr size_t kMaxName = 40;
|
||||
constexpr size_t kMaxQueue = 32;
|
||||
|
||||
struct Command {
|
||||
uint8_t type; // 1=GET, 2=SET, 3=DELETE
|
||||
char name[kMaxName];
|
||||
int64_t value;
|
||||
char base_type[10];
|
||||
int cmd_id;
|
||||
};
|
||||
|
||||
bool gen_random_token(char* out, size_t len) {
|
||||
static const char kAlpha[] =
|
||||
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
|
||||
uint8_t rnd[32];
|
||||
size_t need = len <= sizeof(rnd) ? len : sizeof(rnd);
|
||||
if (!plat::random(rnd, need)) return false;
|
||||
for (size_t i = 0; i < len; i++) {
|
||||
out[i] = kAlpha[rnd[i % need] % 62];
|
||||
}
|
||||
out[len] = '\0';
|
||||
return true;
|
||||
}
|
||||
|
||||
// "cmd_id=5&status=200" -> значения; отсутствующие остаются прежними.
|
||||
void parse_query(const char* query, int* cmd_id, int* status) {
|
||||
const char* p = query;
|
||||
while (*p != '\0') {
|
||||
const char* eq = strchr(p, '=');
|
||||
const char* amp = strchr(p, '&');
|
||||
const char* end = (amp != nullptr) ? amp : p + strlen(p);
|
||||
if (eq != nullptr && eq < end) {
|
||||
size_t klen = static_cast<size_t>(eq - p);
|
||||
int v = atoi(eq + 1);
|
||||
if (klen == 6 && strncmp(p, "cmd_id", 6) == 0) *cmd_id = v;
|
||||
if (klen == 6 && strncmp(p, "status", 6) == 0) *status = v;
|
||||
}
|
||||
if (amp == nullptr) break;
|
||||
p = amp + 1;
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
struct Session::Impl {
|
||||
SessionConfig cfg{};
|
||||
SessionTimings timings{};
|
||||
SessionCallbacks cbs{};
|
||||
|
||||
char host[64] = {};
|
||||
char dsn[40] = {};
|
||||
char lanip_key[64] = {};
|
||||
|
||||
// ---- httpd-поток (используется только из httpd-потока) ----
|
||||
HttpServer httpd;
|
||||
SessionCrypto crypto;
|
||||
bool crypto_ready = false;
|
||||
int64_t out_seq = 0;
|
||||
char ke_random2[17] = {};
|
||||
// Рабочие буферы datapoint-пути (httpd однопоточен; вынесены из стека —
|
||||
// экономия ~7КБ стека потока httpd).
|
||||
char dp_body[kHttpdMaxBody + 1];
|
||||
char dp_enc_b64[kEnvelopeMaxB64];
|
||||
char dp_sign_b64[96];
|
||||
char dp_plain[kEnvelopeMaxPlain];
|
||||
char dp_data[kEnvelopeMaxPlain];
|
||||
|
||||
// ---- разделяемое ----
|
||||
std::mutex queue_mu;
|
||||
Command queue[kMaxQueue] = {};
|
||||
uint8_t queue_len = 0;
|
||||
int next_cmd_id = 1;
|
||||
bool batch_open = false;
|
||||
Command batch[kMaxQueue] = {};
|
||||
uint8_t batch_len = 0;
|
||||
|
||||
std::atomic<uint8_t> state{static_cast<uint8_t>(SessionState::kIdle)};
|
||||
std::atomic<int> last_error{static_cast<int>(SessionError::kNone)};
|
||||
std::atomic<uint64_t> ke_time_ms{0}; // время ответа на KE (0 — не было)
|
||||
std::atomic<uint64_t> last_local_reg_ms{0};
|
||||
std::atomic<uint64_t> quiet_until_ms{0}; // пауза local_reg (восстановление)
|
||||
std::atomic<uint64_t> retry_at_ms{0};
|
||||
std::atomic<bool> had_poll_since_ke{false};
|
||||
std::atomic<bool> ever_active{false}; // был online хотя бы раз
|
||||
std::atomic<bool> want_notify{false}; // после batch commit / перехода online
|
||||
std::atomic<bool> delete_pending{false};
|
||||
std::atomic<bool> delete_served{false};
|
||||
std::atomic<bool> decrypt_failed{false};
|
||||
std::atomic<bool> running{false};
|
||||
|
||||
std::atomic<uint32_t> rekeys{0};
|
||||
std::atomic<uint32_t> pushes_ok{0};
|
||||
std::atomic<uint32_t> pushes_bad{0};
|
||||
std::atomic<uint32_t> cmds_served{0};
|
||||
|
||||
plat::ThreadId thread = nullptr;
|
||||
uint8_t backoff_attempts = 0;
|
||||
uint32_t backoff_ms = 0;
|
||||
bool reg_ok = false; // последний local_reg принят (202/200)
|
||||
|
||||
uint16_t listen_port_actual = 0;
|
||||
|
||||
// ---------- helpers (вызывается из обоих потоков) ----------
|
||||
void set_state(SessionState st, SessionError err) {
|
||||
uint8_t prev = state.exchange(static_cast<uint8_t>(st),
|
||||
std::memory_order_acq_rel);
|
||||
last_error.store(static_cast<int>(err), std::memory_order_release);
|
||||
if (prev != static_cast<uint8_t>(st) && cbs.on_state != nullptr) {
|
||||
cbs.on_state(cbs.ctx, st, err); // без дублирования одинаковых состояний
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- очередь (mutex) ----------
|
||||
bool enqueue_locked(const Command& cmd) {
|
||||
// coalescing: SET замещает незабранный SET того же свойства;
|
||||
// GET-дубликат отбрасывается; DELETE — единственный.
|
||||
if (cmd.type == 3) {
|
||||
for (uint8_t i = 0; i < queue_len; i++) {
|
||||
if (queue[i].type == 3) return true;
|
||||
}
|
||||
} else {
|
||||
for (uint8_t i = 0; i < queue_len; i++) {
|
||||
if (queue[i].type == cmd.type &&
|
||||
strncmp(queue[i].name, cmd.name, kMaxName) == 0) {
|
||||
if (cmd.type == 2) {
|
||||
queue[i].value = cmd.value; // замещаем
|
||||
return true;
|
||||
}
|
||||
if (cmd.type == 1) return true; // дубликат GET
|
||||
}
|
||||
}
|
||||
}
|
||||
if (queue_len >= cfg.max_queue || queue_len >= kMaxQueue) return false;
|
||||
queue[queue_len++] = cmd;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool submit(Command cmd) {
|
||||
std::lock_guard<std::mutex> lk(queue_mu);
|
||||
bool ok;
|
||||
if (batch_open && cmd.type != 3) {
|
||||
if (batch_len >= kMaxQueue) return false;
|
||||
// coalescing внутри batch
|
||||
for (uint8_t i = 0; i < batch_len; i++) {
|
||||
if (batch[i].type == cmd.type &&
|
||||
strncmp(batch[i].name, cmd.name, kMaxName) == 0) {
|
||||
if (cmd.type == 2) {
|
||||
batch[i].value = cmd.value;
|
||||
return true;
|
||||
}
|
||||
if (cmd.type == 1) return true;
|
||||
}
|
||||
}
|
||||
batch[batch_len++] = cmd;
|
||||
ok = true;
|
||||
} else {
|
||||
ok = enqueue_locked(cmd);
|
||||
}
|
||||
return ok;
|
||||
}
|
||||
|
||||
// ---------- httpd-обработчики (httpd-поток) ----------
|
||||
static bool http_handler(const HttpRequest& req, HttpResponse& resp,
|
||||
void* ctx);
|
||||
|
||||
void handle_key_exchange(const HttpRequest& req, HttpResponse& resp);
|
||||
void handle_commands(HttpResponse& resp);
|
||||
void handle_datapoint(const HttpRequest& req, HttpResponse& resp);
|
||||
|
||||
void build_get_payload(char* out, size_t out_cap, const Command& c,
|
||||
int* seq_out);
|
||||
void build_set_payload(char* out, size_t out_cap, const Command& c);
|
||||
|
||||
// ---------- session-поток ----------
|
||||
static void session_thread_trampoline(void* ctx) {
|
||||
static_cast<Impl*>(ctx)->session_loop();
|
||||
}
|
||||
void session_loop();
|
||||
bool send_local_reg(bool notify, bool first);
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// httpd
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
bool Session::Impl::http_handler(const HttpRequest& req, HttpResponse& resp,
|
||||
void* ctx) {
|
||||
auto* impl = static_cast<Impl*>(ctx);
|
||||
if (strcmp(req.method, "POST") == 0) {
|
||||
if (strcmp(req.target, "/local_lan/key_exchange.json") == 0) {
|
||||
impl->handle_key_exchange(req, resp);
|
||||
return true;
|
||||
}
|
||||
if (strcmp(req.target, "/local_lan/property/datapoint.json") == 0) {
|
||||
impl->handle_datapoint(req, resp);
|
||||
return true;
|
||||
}
|
||||
if (strcmp(req.target, "/local_lan/property/datapoint/ack.json") == 0 ||
|
||||
strcmp(req.target, "/local_lan/node/property/datapoint.json") == 0 ||
|
||||
strcmp(req.target, "/local_lan/node/property/datapoint/ack.json") == 0) {
|
||||
FGL_LOGD("session: ack/node datapoint (пусто ok)");
|
||||
resp.status = 200;
|
||||
return true;
|
||||
}
|
||||
} else if (strcmp(req.method, "GET") == 0) {
|
||||
if (strcmp(req.target, "/local_lan/commands.json") == 0) {
|
||||
if (!impl->crypto_ready) {
|
||||
resp.status = 401;
|
||||
return true;
|
||||
}
|
||||
impl->handle_commands(resp);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
resp.status = 404;
|
||||
return true;
|
||||
}
|
||||
|
||||
void Session::Impl::handle_key_exchange(const HttpRequest& req,
|
||||
HttpResponse& resp) {
|
||||
// Тело: {"key_exchange":{"ver":1,"proto":1,"key_id":N,"random_1":..,"time_1":N,"sec":""}}
|
||||
char body[kHttpdMaxBody + 1];
|
||||
size_t n = req.body_len < kHttpdMaxBody ? req.body_len : kHttpdMaxBody;
|
||||
memcpy(body, req.body, n);
|
||||
body[n] = '\0';
|
||||
|
||||
json::Doc doc;
|
||||
if (!doc.parse(body)) {
|
||||
resp.status = 400;
|
||||
return;
|
||||
}
|
||||
int64_t ver = 0, proto = 0, key_id = 0, time_1 = 0;
|
||||
char random_1[32] = {}, sec[8] = {};
|
||||
bool have_r1 = doc.get_string("random_1", random_1, sizeof(random_1));
|
||||
bool have_t1 = doc.get_int("time_1", &time_1);
|
||||
bool have_sec = doc.get_string("sec", sec, sizeof(sec));
|
||||
bool have_ver = doc.get_int("ver", &ver);
|
||||
bool have_proto = doc.get_int("proto", &proto);
|
||||
bool have_kid = doc.get_int("key_id", &key_id);
|
||||
if (!have_r1 || !have_t1 || !have_ver || !have_proto || !have_kid) {
|
||||
FGL_LOGW("session: key_exchange неполный");
|
||||
resp.status = 400;
|
||||
return;
|
||||
}
|
||||
if (ver != 1 || proto != 1 || (have_sec && sec[0] != '\0')) {
|
||||
FGL_LOGW("session: key_exchange ver/proto/sec не поддержаны");
|
||||
set_state(SessionState::kKeyError, SessionError::kBadKeyExchange);
|
||||
resp.status = 426;
|
||||
return;
|
||||
}
|
||||
{
|
||||
// sec длиннее буфера -> get_string=false, но поле есть: setup-режим
|
||||
// не поддерживаем — тоже 426 (PROTOCOL §3.1).
|
||||
const char* sec_start = nullptr;
|
||||
size_t sec_len = 0;
|
||||
jsmntype_t sec_type;
|
||||
if (doc.find("sec", &sec_start, &sec_len, &sec_type) && sec_type == JSMN_STRING &&
|
||||
sec_len >= sizeof(sec)) {
|
||||
set_state(SessionState::kKeyError, SessionError::kBadKeyExchange);
|
||||
resp.status = 426;
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (static_cast<uint32_t>(key_id) != cfg.lanip_key_id) {
|
||||
FGL_LOGE("session: key_id %lld != %u — ротация ключа?",
|
||||
static_cast<long long>(key_id),
|
||||
static_cast<unsigned>(cfg.lanip_key_id));
|
||||
set_state(SessionState::kKeyError, SessionError::kKeyMismatch);
|
||||
resp.status = 412;
|
||||
return;
|
||||
}
|
||||
|
||||
if (!gen_random_token(ke_random2, 16)) {
|
||||
resp.status = 500;
|
||||
return;
|
||||
}
|
||||
int64_t time_2 = static_cast<int64_t>(plat::now_ms()) * 1000000ll;
|
||||
crypto_ready = crypto.init(lanip_key, random_1, ke_random2, time_1,
|
||||
time_2); // цепочки сбрасываются тут же
|
||||
if (!crypto_ready) {
|
||||
resp.status = 500;
|
||||
return;
|
||||
}
|
||||
had_poll_since_ke.store(false, std::memory_order_release);
|
||||
ke_time_ms.store(plat::now_ms(), std::memory_order_release);
|
||||
bool rekey = ever_active.load(std::memory_order_acquire);
|
||||
rekeys.fetch_add(1, std::memory_order_relaxed);
|
||||
if (!rekey && state.load(std::memory_order_acquire) !=
|
||||
static_cast<uint8_t>(SessionState::kRegistering)) {
|
||||
set_state(SessionState::kRegistering, SessionError::kNone);
|
||||
}
|
||||
FGL_LOGI("session: key exchange #%u (rekey=%d)",
|
||||
static_cast<unsigned>(rekeys.load(std::memory_order_relaxed)), rekey);
|
||||
|
||||
static thread_local char out[128];
|
||||
json::Writer w(out, sizeof(out));
|
||||
w.begin_object();
|
||||
w.key("random_2");
|
||||
w.string(ke_random2);
|
||||
w.key("time_2");
|
||||
w.integer(time_2);
|
||||
w.end_object();
|
||||
resp.status = 200;
|
||||
resp.body = reinterpret_cast<const uint8_t*>(out);
|
||||
resp.body_len = strlen(out);
|
||||
// буфер out живёт до конца ответа (send_response в handle_connection
|
||||
// выполняется синхронно в том же кадре стека httpd-потока).
|
||||
}
|
||||
|
||||
void Session::Impl::build_get_payload(char* out, size_t out_cap,
|
||||
const Command& c, int* seq_out) {
|
||||
(void)seq_out;
|
||||
json::Writer w(out, out_cap);
|
||||
w.begin_object();
|
||||
w.key("cmds");
|
||||
w.begin_array();
|
||||
w.begin_object();
|
||||
w.key("cmd");
|
||||
w.begin_object();
|
||||
w.key("cmd_id");
|
||||
w.integer(c.cmd_id);
|
||||
w.key("method");
|
||||
w.string("GET");
|
||||
w.key("resource");
|
||||
char res[80];
|
||||
snprintf(res, sizeof(res), "property.json?name=%s", c.name);
|
||||
w.string(res);
|
||||
w.key("data");
|
||||
w.string("");
|
||||
w.key("uri");
|
||||
w.string("/local_lan/property/datapoint.json");
|
||||
w.end_object();
|
||||
w.end_object();
|
||||
w.end_array();
|
||||
w.end_object();
|
||||
}
|
||||
|
||||
void Session::Impl::build_set_payload(char* out, size_t out_cap,
|
||||
const Command& c) {
|
||||
json::Writer w(out, out_cap);
|
||||
w.begin_object();
|
||||
w.key("properties");
|
||||
w.begin_array();
|
||||
w.begin_object();
|
||||
w.key("property");
|
||||
w.begin_object();
|
||||
w.key("base_type");
|
||||
w.string(c.base_type);
|
||||
w.key("name");
|
||||
w.string(c.name);
|
||||
w.key("value");
|
||||
if (strcmp(c.base_type, "boolean") == 0) {
|
||||
w.boolean(c.value != 0);
|
||||
} else {
|
||||
w.integer(c.value);
|
||||
}
|
||||
w.key("id");
|
||||
char id[9];
|
||||
gen_random_token(id, 8);
|
||||
w.string(id);
|
||||
w.end_object();
|
||||
w.end_object();
|
||||
w.end_array();
|
||||
w.end_object();
|
||||
}
|
||||
|
||||
void Session::Impl::handle_commands(HttpResponse& resp) {
|
||||
Command head{};
|
||||
bool have = false;
|
||||
uint8_t remaining = 0;
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(queue_mu);
|
||||
if (queue_len > 0) {
|
||||
head = queue[0];
|
||||
have = true;
|
||||
queue_len--;
|
||||
memmove(queue, queue + 1, queue_len * sizeof(Command));
|
||||
remaining = queue_len;
|
||||
}
|
||||
}
|
||||
char payload[512];
|
||||
if (!have) {
|
||||
payload[0] = '{';
|
||||
payload[1] = '}';
|
||||
payload[2] = '\0';
|
||||
} else if (head.type == 1) {
|
||||
build_get_payload(payload, sizeof(payload), head, nullptr);
|
||||
} else if (head.type == 2) {
|
||||
build_set_payload(payload, sizeof(payload), head);
|
||||
} else { // DELETE session
|
||||
json::Writer w(payload, sizeof(payload));
|
||||
w.begin_object();
|
||||
w.key("cmds");
|
||||
w.begin_array();
|
||||
w.begin_object();
|
||||
w.key("cmd");
|
||||
w.begin_object();
|
||||
w.key("cmd_id");
|
||||
w.integer(0);
|
||||
w.key("method");
|
||||
w.string("DELETE");
|
||||
w.key("resource");
|
||||
w.string("local_reg.json");
|
||||
w.key("data");
|
||||
w.string("delete_session");
|
||||
w.key("uri");
|
||||
w.string("/local_lan");
|
||||
w.end_object();
|
||||
w.end_object();
|
||||
w.end_array();
|
||||
w.end_object();
|
||||
delete_served.store(true, std::memory_order_release);
|
||||
delete_pending.store(false, std::memory_order_release);
|
||||
FGL_LOGI("session: delete_session выдан модулю");
|
||||
}
|
||||
|
||||
static thread_local char envelope[kEnvelopeMaxB64];
|
||||
int64_t seq = out_seq++;
|
||||
if (!envelope_pack(crypto.app, seq, payload, envelope, sizeof(envelope))) {
|
||||
resp.status = 500;
|
||||
return;
|
||||
}
|
||||
cmds_served.fetch_add(1, std::memory_order_relaxed);
|
||||
had_poll_since_ke.store(true, std::memory_order_release);
|
||||
uint8_t st8 = state.load(std::memory_order_acquire);
|
||||
if (st8 == static_cast<uint8_t>(SessionState::kRegistering) ||
|
||||
st8 == static_cast<uint8_t>(SessionState::kRecovering) ||
|
||||
st8 == static_cast<uint8_t>(SessionState::kOffline)) {
|
||||
// Опрос команд = сессия жива (в т.ч. после re-key при recovering/offline).
|
||||
ever_active.store(true, std::memory_order_release);
|
||||
decrypt_failed.store(false, std::memory_order_release);
|
||||
set_state(SessionState::kOnline, SessionError::kNone);
|
||||
want_notify.store(true, std::memory_order_release);
|
||||
}
|
||||
resp.status = remaining > 0 ? 206 : 200;
|
||||
resp.body = reinterpret_cast<const uint8_t*>(envelope);
|
||||
resp.body_len = strlen(envelope);
|
||||
}
|
||||
|
||||
void Session::Impl::handle_datapoint(const HttpRequest& req,
|
||||
HttpResponse& resp) {
|
||||
size_t n = req.body_len < kHttpdMaxBody ? req.body_len : kHttpdMaxBody;
|
||||
memcpy(dp_body, req.body, n);
|
||||
dp_body[n] = '\0';
|
||||
json::Doc wrap;
|
||||
if (!wrap.parse(dp_body)) {
|
||||
resp.status = 400;
|
||||
return;
|
||||
}
|
||||
if (!wrap.get_string("enc", dp_enc_b64, sizeof(dp_enc_b64)) ||
|
||||
!wrap.get_string("sign", dp_sign_b64, sizeof(dp_sign_b64))) {
|
||||
resp.status = 400;
|
||||
return;
|
||||
}
|
||||
|
||||
int64_t seq_no = -1;
|
||||
if (!envelope_unpack(crypto.dev, dp_enc_b64, dp_sign_b64, dp_plain,
|
||||
sizeof(dp_plain), &seq_no)) {
|
||||
pushes_bad.fetch_add(1, std::memory_order_relaxed);
|
||||
FGL_LOGW("session: push не расшифрован/подпись (401); пауза и re-key");
|
||||
if (state.load(std::memory_order_acquire) ==
|
||||
static_cast<uint8_t>(SessionState::kOnline)) {
|
||||
decrypt_failed.store(true, std::memory_order_release);
|
||||
set_state(SessionState::kRecovering, SessionError::kDecryptFailed);
|
||||
// Стратегия восстановления (PROTOCOL §4.4): замолчать на > порога
|
||||
// «возврата» (~44-50с) — следующий local_reg заставит модуль re-key.
|
||||
quiet_until_ms.store(plat::now_ms() + timings.recovering_quiet_ms,
|
||||
std::memory_order_release);
|
||||
}
|
||||
resp.status = 401;
|
||||
return;
|
||||
}
|
||||
|
||||
// Восстановление после 401 с живой цепочкой (бракованная подпись):
|
||||
// сообщение расшифровано — отменяем тишину и возвращаем online.
|
||||
if (decrypt_failed.exchange(false, std::memory_order_acq_rel)) {
|
||||
quiet_until_ms.store(0, std::memory_order_release);
|
||||
set_state(SessionState::kOnline, SessionError::kNone);
|
||||
FGL_LOGI("session: цепочка восстановлена (успешный push после 401)");
|
||||
}
|
||||
|
||||
// {"seq_no":N,"data":{"name":..,"value":..}}
|
||||
json::Doc top;
|
||||
if (!top.parse(dp_plain)) {
|
||||
pushes_ok.fetch_add(1, std::memory_order_relaxed);
|
||||
resp.status = 200;
|
||||
return;
|
||||
}
|
||||
const char* data_start = nullptr;
|
||||
size_t data_len = 0;
|
||||
jsmntype_t data_type;
|
||||
PropertyEvent ev{};
|
||||
ev.seq_no = seq_no;
|
||||
if (top.find("data", &data_start, &data_len, &data_type) &&
|
||||
data_type == JSMN_OBJECT && data_len < sizeof(dp_data)) {
|
||||
memcpy(dp_data, data_start, data_len);
|
||||
dp_data[data_len] = '\0';
|
||||
{
|
||||
json::Doc data_doc;
|
||||
if (data_doc.parse(dp_data)) {
|
||||
char name[kMaxName];
|
||||
if (data_doc.get_string("name", name, sizeof(name))) {
|
||||
snprintf(ev.name, sizeof(ev.name), "%s", name);
|
||||
int64_t iv = 0;
|
||||
bool bv = false;
|
||||
const char* sv = nullptr;
|
||||
size_t sv_len = 0;
|
||||
jsmntype_t vt;
|
||||
if (data_doc.get_int("value", &iv)) {
|
||||
ev.is_int = true;
|
||||
ev.int_value = iv;
|
||||
} else if (data_doc.get_bool("value", &bv)) {
|
||||
ev.is_bool = true;
|
||||
ev.bool_value = bv;
|
||||
} else if (data_doc.find("value", &sv, &sv_len, &vt) &&
|
||||
vt == JSMN_STRING && sv_len + 1 <= sizeof(ev.str_value)) {
|
||||
memcpy(ev.str_value, sv, sv_len);
|
||||
ev.str_value[sv_len] = '\0';
|
||||
}
|
||||
int cmd_id = -1, status = 0;
|
||||
parse_query(req.query, &cmd_id, &status);
|
||||
ev.cmd_id = cmd_id;
|
||||
ev.status = status;
|
||||
if (cbs.on_property != nullptr) {
|
||||
cbs.on_property(cbs.ctx, ev);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
pushes_ok.fetch_add(1, std::memory_order_relaxed);
|
||||
resp.status = 200;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// session-поток: local_reg / keep-alive / backoff / таймауты
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
bool Session::Impl::send_local_reg(bool notify, bool first) {
|
||||
char path[96];
|
||||
if (first) {
|
||||
snprintf(path, sizeof(path), "/local_reg.json?dsn=%s", dsn);
|
||||
} else {
|
||||
snprintf(path, sizeof(path), "/local_reg.json");
|
||||
}
|
||||
char local_ip[24];
|
||||
if (!plat::local_ip_for(host, local_ip, sizeof(local_ip))) {
|
||||
FGL_LOGW("session: local_ip_for(%s) failed", host);
|
||||
return false;
|
||||
}
|
||||
char body[160];
|
||||
json::Writer w(body, sizeof(body));
|
||||
w.begin_object();
|
||||
w.key("local_reg");
|
||||
w.begin_object();
|
||||
w.key("ip");
|
||||
w.string(local_ip);
|
||||
w.key("notify");
|
||||
w.boolean(notify);
|
||||
w.key("port");
|
||||
w.integer(listen_port_actual);
|
||||
w.key("uri");
|
||||
w.string("/local_lan");
|
||||
w.end_object();
|
||||
w.end_object();
|
||||
if (!w.ok()) return false;
|
||||
|
||||
HttpcRequest req;
|
||||
req.method = first ? "POST" : "PUT";
|
||||
req.host = host;
|
||||
req.port = cfg.device_port;
|
||||
req.path = path;
|
||||
req.body = reinterpret_cast<const uint8_t*>(body);
|
||||
req.body_len = strlen(body);
|
||||
req.timeout_ms = 5000;
|
||||
HttpcResponse resp;
|
||||
if (!httpc_perform(req, &resp)) {
|
||||
return false;
|
||||
}
|
||||
last_local_reg_ms.store(plat::now_ms(), std::memory_order_release);
|
||||
if (resp.status == 503) {
|
||||
set_state(SessionState::kOffline, SessionError::kNoSlot);
|
||||
retry_at_ms.store(plat::now_ms() + timings.no_slot_retry_ms,
|
||||
std::memory_order_release);
|
||||
FGL_LOGW("session: local_reg -> 503 (нет слотов)");
|
||||
backoff_attempts = 0;
|
||||
backoff_ms = 0;
|
||||
reg_ok = false; // сессия не активировалась — следующий local_reg POST?dsn
|
||||
return true; // транспорт ок — это протокольный ответ
|
||||
}
|
||||
if (resp.status != 202 && resp.status != 200) {
|
||||
FGL_LOGW("session: local_reg -> %d", resp.status);
|
||||
return false;
|
||||
}
|
||||
if (state.load(std::memory_order_acquire) ==
|
||||
static_cast<uint8_t>(SessionState::kIdle)) {
|
||||
set_state(SessionState::kRegistering, SessionError::kNone);
|
||||
}
|
||||
backoff_attempts = 0;
|
||||
backoff_ms = 0;
|
||||
reg_ok = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
void Session::Impl::session_loop() {
|
||||
bool first_reg = true;
|
||||
while (running.load(std::memory_order_acquire)) {
|
||||
uint64_t now = plat::now_ms();
|
||||
SessionState st =
|
||||
static_cast<SessionState>(state.load(std::memory_order_acquire));
|
||||
|
||||
// kKeyError — устойчивая ошибка: конфиг менять вручную, не дёргаем модуль.
|
||||
if (st == SessionState::kKeyError) {
|
||||
plat::sleep_ms(kLoopTickMs);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Тишина (восстановление после «KE без poll»).
|
||||
if (now < quiet_until_ms.load(std::memory_order_acquire)) {
|
||||
plat::sleep_ms(kLoopTickMs);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Активация: KE отвечен, но опроса нет.
|
||||
uint64_t ke_time = ke_time_ms.load(std::memory_order_acquire);
|
||||
if (st == SessionState::kRegistering && ke_time != 0 &&
|
||||
!had_poll_since_ke.load(std::memory_order_acquire) &&
|
||||
now - ke_time > timings.activation_timeout_ms) {
|
||||
FGL_LOGW("session: активация не наступила (KE без poll) — пауза %ums",
|
||||
static_cast<unsigned>(timings.recovering_quiet_ms));
|
||||
ke_time_ms.store(0, std::memory_order_release);
|
||||
quiet_until_ms.store(now + timings.recovering_quiet_ms,
|
||||
std::memory_order_release);
|
||||
set_state(SessionState::kRecovering, SessionError::kActivationTimeout);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Backoff / отложенный повтор.
|
||||
if (now < retry_at_ms.load(std::memory_order_acquire)) {
|
||||
plat::sleep_ms(kLoopTickMs);
|
||||
continue;
|
||||
}
|
||||
|
||||
// local_reg: по keep-alive, по notify (batch/online) или первичный.
|
||||
bool queue_nonempty;
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(queue_mu);
|
||||
queue_nonempty = queue_len > 0;
|
||||
}
|
||||
bool notify = queue_nonempty || want_notify.exchange(false,
|
||||
std::memory_order_acq_rel);
|
||||
uint64_t last_reg = last_local_reg_ms.load(std::memory_order_acquire);
|
||||
bool due = notify || last_reg == 0 ||
|
||||
now - last_reg >= timings.keepalive_ms;
|
||||
if (!due) {
|
||||
plat::sleep_ms(kLoopTickMs);
|
||||
continue;
|
||||
}
|
||||
|
||||
bool first = first_reg || !reg_ok;
|
||||
if (!send_local_reg(queue_nonempty, first)) {
|
||||
// Транспортная ошибка: backoff.
|
||||
backoff_ms = backoff_ms == 0 ? timings.backoff_base_ms
|
||||
: (backoff_ms * 8) / 5; // x1.6
|
||||
if (backoff_ms > timings.backoff_max_ms) {
|
||||
backoff_ms = timings.backoff_max_ms;
|
||||
}
|
||||
backoff_attempts++;
|
||||
if (backoff_attempts >= timings.backoff_attempts) {
|
||||
set_state(SessionState::kOffline, SessionError::kUnreachable);
|
||||
retry_at_ms.store(now + timings.backoff_max_ms,
|
||||
std::memory_order_release);
|
||||
backoff_attempts = 0;
|
||||
backoff_ms = 0;
|
||||
} else {
|
||||
retry_at_ms.store(now + backoff_ms, std::memory_order_release);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (reg_ok) first_reg = false;
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Session (публичный класс)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
Session* Session::create(const SessionConfig& cfg, const SessionCallbacks& cbs) {
|
||||
if (cfg.host == nullptr || cfg.dsn == nullptr || cfg.lanip_key == nullptr) {
|
||||
return nullptr;
|
||||
}
|
||||
if (strlen(cfg.lanip_key) >= sizeof(Impl::lanip_key) ||
|
||||
strlen(cfg.dsn) >= sizeof(Impl::dsn) ||
|
||||
strlen(cfg.host) >= sizeof(Impl::host)) {
|
||||
return nullptr; // не помещается во внутренние копии
|
||||
}
|
||||
auto* impl = new (std::nothrow) Impl();
|
||||
if (impl == nullptr) return nullptr;
|
||||
impl->cfg = cfg;
|
||||
impl->cbs = cbs;
|
||||
if (impl->cfg.max_queue == 0) impl->cfg.max_queue = 16;
|
||||
if (impl->cfg.keepalive_ms == 0) impl->cfg.keepalive_ms = 15000;
|
||||
impl->timings.keepalive_ms = impl->cfg.keepalive_ms;
|
||||
snprintf(impl->host, sizeof(impl->host), "%s", cfg.host);
|
||||
snprintf(impl->dsn, sizeof(impl->dsn), "%s", cfg.dsn);
|
||||
snprintf(impl->lanip_key, sizeof(impl->lanip_key), "%s", cfg.lanip_key);
|
||||
auto* s = new (std::nothrow) Session(cfg, cbs);
|
||||
if (s == nullptr) {
|
||||
delete impl;
|
||||
return nullptr;
|
||||
}
|
||||
s->impl_ = impl;
|
||||
return s;
|
||||
}
|
||||
|
||||
Session::Session(const SessionConfig&, const SessionCallbacks&) : impl_(nullptr) {}
|
||||
|
||||
SessionError Session::last_error() const {
|
||||
if (impl_ == nullptr) return SessionError::kNone;
|
||||
return static_cast<SessionError>(impl_->last_error.load(std::memory_order_acquire));
|
||||
}
|
||||
|
||||
SessionState Session::state() const {
|
||||
if (impl_ == nullptr) return SessionState::kIdle;
|
||||
return static_cast<SessionState>(impl_->state.load(std::memory_order_acquire));
|
||||
}
|
||||
|
||||
uint16_t Session::listen_port() const {
|
||||
return impl_ != nullptr ? impl_->listen_port_actual : 0;
|
||||
}
|
||||
|
||||
uint32_t Session::rekey_count() const {
|
||||
return impl_ != nullptr ? impl_->rekeys.load(std::memory_order_relaxed) : 0;
|
||||
}
|
||||
uint32_t Session::pushes_ok() const {
|
||||
return impl_ != nullptr ? impl_->pushes_ok.load(std::memory_order_relaxed) : 0;
|
||||
}
|
||||
uint32_t Session::pushes_bad() const {
|
||||
return impl_ != nullptr ? impl_->pushes_bad.load(std::memory_order_relaxed) : 0;
|
||||
}
|
||||
uint32_t Session::commands_served() const {
|
||||
return impl_ != nullptr ? impl_->cmds_served.load(std::memory_order_relaxed) : 0;
|
||||
}
|
||||
bool Session::had_activity() const {
|
||||
return impl_ != nullptr &&
|
||||
impl_->had_poll_since_ke.load(std::memory_order_acquire);
|
||||
}
|
||||
Session::~Session() {
|
||||
stop();
|
||||
delete impl_;
|
||||
}
|
||||
|
||||
bool Session::start() {
|
||||
if (impl_ == nullptr || impl_->running.load(std::memory_order_acquire)) {
|
||||
return false;
|
||||
}
|
||||
if (!impl_->httpd.start(impl_->cfg.listen_port, Impl::http_handler, impl_,
|
||||
"fgl_session")) {
|
||||
return false;
|
||||
}
|
||||
impl_->listen_port_actual = impl_->httpd.port();
|
||||
impl_->running.store(true, std::memory_order_release);
|
||||
if (!plat::thread_create(Impl::session_thread_trampoline, impl_,
|
||||
"fgl_sess", 8192, &impl_->thread)) {
|
||||
impl_->running.store(false, std::memory_order_release);
|
||||
impl_->httpd.stop();
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
void Session::stop() {
|
||||
if (impl_ == nullptr || !impl_->running.load(std::memory_order_acquire)) {
|
||||
return;
|
||||
}
|
||||
// Штатное завершение: DELETE-команда + notify local_reg. Сессионный поток
|
||||
// ещё работает и доставит notify; ждём выдачи команды модулю.
|
||||
if (impl_->state.load(std::memory_order_acquire) !=
|
||||
static_cast<uint8_t>(SessionState::kKeyError)) {
|
||||
delete_session();
|
||||
uint64_t deadline = plat::now_ms() + impl_->timings.delete_wait_ms;
|
||||
while (plat::now_ms() < deadline &&
|
||||
!impl_->delete_served.load(std::memory_order_acquire)) {
|
||||
plat::sleep_ms(10);
|
||||
}
|
||||
}
|
||||
impl_->running.store(false, std::memory_order_release);
|
||||
if (impl_->thread != nullptr) {
|
||||
plat::thread_join(impl_->thread);
|
||||
impl_->thread = nullptr;
|
||||
}
|
||||
impl_->httpd.stop();
|
||||
// Сброс для возможного рестарта.
|
||||
impl_->delete_served.store(false, std::memory_order_release);
|
||||
impl_->delete_pending.store(false, std::memory_order_release);
|
||||
impl_->quiet_until_ms.store(0, std::memory_order_release);
|
||||
impl_->retry_at_ms.store(0, std::memory_order_release);
|
||||
impl_->ke_time_ms.store(0, std::memory_order_release);
|
||||
impl_->set_state(SessionState::kIdle, SessionError::kNone);
|
||||
}
|
||||
|
||||
bool Session::get_property(const char* name) {
|
||||
if (impl_ == nullptr || name == nullptr || strlen(name) >= kMaxName) {
|
||||
return false;
|
||||
}
|
||||
Command cmd{};
|
||||
cmd.type = 1;
|
||||
snprintf(cmd.name, sizeof(cmd.name), "%s", name);
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(impl_->queue_mu);
|
||||
cmd.cmd_id = impl_->next_cmd_id++;
|
||||
}
|
||||
if (!impl_->submit(cmd)) return false;
|
||||
impl_->want_notify.store(true, std::memory_order_release);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Session::set_property(const char* name, int64_t value,
|
||||
const char* base_type) {
|
||||
if (impl_ == nullptr || name == nullptr || strlen(name) >= kMaxName) {
|
||||
return false;
|
||||
}
|
||||
Command cmd{};
|
||||
cmd.type = 2;
|
||||
snprintf(cmd.name, sizeof(cmd.name), "%s", name);
|
||||
cmd.value = value;
|
||||
snprintf(cmd.base_type, sizeof(cmd.base_type), "%s", base_type);
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(impl_->queue_mu);
|
||||
cmd.cmd_id = impl_->next_cmd_id++;
|
||||
}
|
||||
if (!impl_->submit(cmd)) return false;
|
||||
impl_->want_notify.store(true, std::memory_order_release);
|
||||
return true;
|
||||
}
|
||||
|
||||
void Session::set_timings_for_test(const SessionTimings& t) {
|
||||
if (impl_ == nullptr) return;
|
||||
SessionTimings tmp = t;
|
||||
if (tmp.keepalive_ms < 100) tmp.keepalive_ms = 100; // анти-спам
|
||||
impl_->timings = tmp;
|
||||
}
|
||||
|
||||
bool Session::begin_batch() {
|
||||
std::lock_guard<std::mutex> lk(impl_->queue_mu);
|
||||
if (impl_->batch_open) return false;
|
||||
impl_->batch_open = true;
|
||||
impl_->batch_len = 0;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Session::commit_batch() {
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(impl_->queue_mu);
|
||||
if (!impl_->batch_open) return false;
|
||||
bool ok = true;
|
||||
for (uint8_t i = 0; i < impl_->batch_len; i++) {
|
||||
if (!impl_->enqueue_locked(impl_->batch[i])) ok = false;
|
||||
}
|
||||
impl_->batch_open = false;
|
||||
impl_->batch_len = 0;
|
||||
if (!ok) return false;
|
||||
}
|
||||
impl_->want_notify.store(true, std::memory_order_release);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Session::abort_batch() {
|
||||
std::lock_guard<std::mutex> lk(impl_->queue_mu);
|
||||
if (!impl_->batch_open) return false;
|
||||
impl_->batch_open = false;
|
||||
impl_->batch_len = 0;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Session::delete_session() {
|
||||
if (impl_ == nullptr) return false;
|
||||
Command cmd{};
|
||||
cmd.type = 3;
|
||||
snprintf(cmd.name, sizeof(cmd.name), "local_reg.json");
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(impl_->queue_mu);
|
||||
if (!impl_->enqueue_locked(cmd)) return false;
|
||||
}
|
||||
impl_->delete_pending.store(true, std::memory_order_release);
|
||||
impl_->want_notify.store(true, std::memory_order_release);
|
||||
return true;
|
||||
}
|
||||
|
||||
} // namespace fgl::ayla
|
||||
126
src/ayla/session.hpp
Normal file
126
src/ayla/session.hpp
Normal file
@@ -0,0 +1,126 @@
|
||||
// Сессия Ayla LAN (сторона «приложения»). docs/PROTOCOL.md §4-6, §4.4.
|
||||
// Потоки: httpd (входящие от модуля: key exchange/commands/datapoint) и
|
||||
// session (исходящие local_reg, таймеры, backoff). Крипто-цепочки и выдача
|
||||
// команд — только в httpd-потоке; session-поток читает очередь под mutex.
|
||||
#pragma once
|
||||
|
||||
#include <atomic>
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
|
||||
#include "ayla/crypto.hpp"
|
||||
#include "ayla/httpd.hpp"
|
||||
|
||||
namespace fgl::ayla {
|
||||
|
||||
enum class SessionState : uint8_t {
|
||||
kIdle = 0, // создан, не запущен
|
||||
kRegistering, // local_reg отправлен, ждём key exchange
|
||||
kOnline, // сессия активна
|
||||
kRecovering, // ожидание самолечения (re-key по keep-alive / активация)
|
||||
kOffline, // модуль недоступен (backoff) или нет слотов
|
||||
kKeyError, // lanip_key_id не совпал — требуется смена конфига
|
||||
};
|
||||
|
||||
enum class SessionError : int {
|
||||
kNone = 0,
|
||||
kNoSlot = 1, // 503: оба слота модуля заняты
|
||||
kUnreachable = 2, // transport/backoff
|
||||
kKeyMismatch = 3, // key_id != lanip_key_id (state = kKeyError)
|
||||
kBadKeyExchange = 4, // ver/proto/sec не поддержаны
|
||||
kActivationTimeout = 5,// KE прошёл, опроса commands.json нет (>5 c)
|
||||
kDecryptFailed = 6, // подпись/расшифровка push не сошлись (ждём re-key)
|
||||
};
|
||||
|
||||
// Событие обновления свойства (push модуля). Значение — какой-то один тип.
|
||||
struct PropertyEvent {
|
||||
char name[40];
|
||||
bool is_int = false;
|
||||
int64_t int_value = 0;
|
||||
bool is_bool = false;
|
||||
bool bool_value = false;
|
||||
char str_value[64]; // используется, если !is_int && !is_bool
|
||||
int cmd_id = -1; // из ?cmd_id=N (ответ на GET), иначе -1
|
||||
int status = 0; // из ?status=200
|
||||
int64_t seq_no = 0;
|
||||
};
|
||||
|
||||
struct SessionConfig {
|
||||
const char* host = nullptr; // DNS-имя или IP модуля
|
||||
uint16_t device_port = 80; // порт local_reg модуля
|
||||
const char* dsn = nullptr; // "AC000W00XXXXXXX"
|
||||
const char* lanip_key = nullptr; // base64-строка как есть
|
||||
uint32_t lanip_key_id = 0;
|
||||
uint16_t listen_port = 10275; // 0 — любой свободный
|
||||
uint32_t keepalive_ms = 15000;
|
||||
uint8_t max_queue = 16; // лимит очереди команд
|
||||
};
|
||||
|
||||
struct SessionCallbacks {
|
||||
// КОНТРАКТ: колбэки приходят из потоков ядра (httpd и/или session),
|
||||
// возможно перекрытие во времени; обязаны быть быстрыми и реентерабельными.
|
||||
// Вызывать stop() из колбэка запрещено (deadlock на join).
|
||||
void (*on_state)(void* ctx, SessionState st, SessionError err);
|
||||
void (*on_property)(void* ctx, const PropertyEvent& ev);
|
||||
void* ctx = nullptr;
|
||||
};
|
||||
|
||||
// Тайминги поведения (PROTOCOL §4.3-4.4; проверено на приборе).
|
||||
struct SessionTimings {
|
||||
uint32_t keepalive_ms = 15000; // период local_reg
|
||||
uint32_t activation_timeout_ms = 5000; // нет poll после KE
|
||||
uint32_t recovering_quiet_ms = 50000; // пауза при десинке/«KE без poll»
|
||||
// (> порога возврата модуля ~44-50с)
|
||||
uint32_t no_slot_retry_ms = 60000; // повтор после 503
|
||||
uint32_t backoff_base_ms = 1000; // transport backoff, шаг x1.6
|
||||
uint32_t backoff_max_ms = 60000;
|
||||
uint8_t backoff_attempts = 6;
|
||||
uint32_t delete_wait_ms = 2000;
|
||||
};
|
||||
|
||||
class Session {
|
||||
public:
|
||||
static Session* create(const SessionConfig& cfg, const SessionCallbacks& cbs);
|
||||
~Session();
|
||||
|
||||
Session(const Session&) = delete;
|
||||
Session& operator=(const Session&) = delete;
|
||||
|
||||
bool start();
|
||||
// Штатное завершение: DELETE-команда + local_reg notify, ожидание выдачи,
|
||||
// остановка потоков. state -> kIdle.
|
||||
void stop();
|
||||
|
||||
SessionState state() const;
|
||||
SessionError last_error() const;
|
||||
uint16_t listen_port() const;
|
||||
// Телеметрия (диагностика).
|
||||
uint32_t rekey_count() const;
|
||||
uint32_t pushes_ok() const;
|
||||
uint32_t pushes_bad() const;
|
||||
uint32_t commands_served() const;
|
||||
bool had_activity() const;
|
||||
|
||||
// ---- Команды (потокобезопасны; кладутся в очередь с coalescing) ----
|
||||
// GET-команда: свойство придёт on_property (cmd_id совпадает).
|
||||
bool get_property(const char* name);
|
||||
// SET-команда (integer/boolean как int64).
|
||||
bool set_property(const char* name, int64_t value,
|
||||
const char* base_type = "integer");
|
||||
// Пакет: собрать несколько команд, один notify на commit.
|
||||
bool begin_batch();
|
||||
bool commit_batch();
|
||||
bool abort_batch();
|
||||
// DELETE local_reg.json/delete_session (для stop() и ручного завершения).
|
||||
bool delete_session();
|
||||
|
||||
// Тест-хук: тайминги. ТОЛЬКО до start() (после — читаются потоками ядра).
|
||||
void set_timings_for_test(const SessionTimings& t);
|
||||
|
||||
private:
|
||||
Session(const SessionConfig& cfg, const SessionCallbacks& cbs);
|
||||
struct Impl;
|
||||
Impl* impl_;
|
||||
};
|
||||
|
||||
} // namespace fgl::ayla
|
||||
Reference in New Issue
Block a user