From 158916b664300c24e5952494c16f4b8a0d5e3d03 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Mon, 12 Oct 2026 12:25:14 -0400 Subject: [PATCH] Preserve and retry telemetry match boundaries across restarts --- README.md | 66 +++- t5/plugins/README.md | 45 ++- .../src/component/telemetry.cpp | 75 +++++ .../tws-t5-gsc-utils/src/utils/telemetry.cpp | 199 ++++++++++++ .../tws-t5-gsc-utils/src/utils/telemetry.hpp | 45 +++ t5/scripts/README.md | 71 ++--- t5/scripts/tws_t5_telemetry.gsc | 194 ++++++++++-- t6/plugins/README.md | 45 ++- .../src/component/telemetry.cpp | 75 +++++ .../tws-t6-gsc-utils/src/utils/telemetry.cpp | 199 ++++++++++++ .../tws-t6-gsc-utils/src/utils/telemetry.hpp | 45 +++ t6/scripts/README.md | 28 +- t6/scripts/tws_t6_telemetry.gsc | 197 +++++++++--- tools/check_telemetry.mjs | 283 ++++++++++++++++++ tools/test_plugin_telemetry.py | 52 ++++ tools/tests/telemetry_cases.cpp | 126 ++++++++ 16 files changed, 1603 insertions(+), 142 deletions(-) create mode 100644 t5/plugins/tws-t5-gsc-utils/src/component/telemetry.cpp create mode 100644 t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.cpp create mode 100644 t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.hpp create mode 100644 t6/plugins/tws-t6-gsc-utils/src/component/telemetry.cpp create mode 100644 t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.cpp create mode 100644 t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.hpp create mode 100644 tools/check_telemetry.mjs create mode 100644 tools/test_plugin_telemetry.py create mode 100644 tools/tests/telemetry_cases.cpp diff --git a/README.md b/README.md index b214218..7807166 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,69 @@ Both plugins now default to a 5-second connection timeout and 15-second total transfer timeout. Request options can override these with positive millisecond values. Telemetry explicitly sets those limits and logs native HTTP status and timings. See the [T5](t5/plugins/README.md) and [T6](t6/plugins/README.md) plugin notes. -Requests are still sent once, in order, with no retries or event timestamp changes. +Telemetry retries and durable replay require the updated plugins; see the +[delivery and rollout notes](#telemetry-delivery-and-rollout-t5-and-t6). + +## Telemetry delivery and rollout (T5 and T6) + +**Deploy Electric Cherry [#144](https://gitea.thewrightserver.net/josh/electric-cherry/issues/144) +before these reporters.** Replay requires the receiver to associate reports with +`match` and deduplicate `seq`; the legacy receiver can apply an old end to a newer +match. Build and install both updated TWS plugins, restart the server processes, +then deploy the corresponding scripts. The new `telemetryNewMatch`, +`telemetryLoad`, `telemetryStore`, `telemetryRemove` and `telemetryResponse` +functions are required even if telemetry is disabled by dvar. + +Every captured report includes a `match` key (random native process token plus +an increasing match counter, at most 64 characters) and a positive `seq`. +`game_start` is sequence 1. Retries and replay send the exact original JSON; +sequence numbers are never reassigned. Player snapshots still reflect capture +time; no client timestamp has been added and API receipt times retain their +existing meaning. + +A single sender keeps `game_start` and `game_end` at the queue head on transport +failure, native timeout, HTTP 429 or 5xx. Delays are 1, 2, 4, 8, 16, then 30 seconds +between attempts, capped at 30 seconds without a retry-count limit. Other 4xx +(and redirects) are logged and dropped. A 2xx with a JSON boolean `"ok": true` +acknowledges delivery; empty, malformed or negative 2xx replies and missing HTTP +diagnostics are unacknowledged and retried for boundaries. Accepted reports log +Electric Cherry's `telemetry.timing.matchId` as `ec_match_id` when supplied. +Heartbeats and round changes remain best effort. Waiting heartbeats are coalesced +to the newest snapshot, preserving round reports and any in-flight request. +Intentional coalescing can produce gaps in `seq`. + +The native outbox saves start/end snapshots synchronously before sending, and +removes them only after acknowledgment or a terminal HTTP rejection. Files live +under `fs_homepath/tws-telemetry/t5/` or `t6/`, scoped to the exact server name and +endpoint. Keep this directory on persistent writable storage across process or +container restarts; it contains player snapshots but no authorization token. +Use a single process per server/endpoint and storage directory. Endpoint/server +changes use a separate outbox, so restore the original configuration to replay +its backlog. Keep `tws_server_id` stable between maps. On map/process startup, +recovered boundaries precede any new `game_start`. An accepted report whose +acknowledgment/removal was lost is safely replayed with the same identity. + +Writes replace the committed file only after a complete temporary write. A +storage error blocks the affected boundary before sending and logs an error; +unsaved in-memory reports cannot survive a restart. Unreadable/corrupt saved data +disables telemetry until storage is repaired and the map restarted, preserving +the file for recovery. Each outbox is limited to 4,096 reports / 16 MiB, with +individual report bodies below 20 KiB; reaching a limit fails visibly without +discarding the saved backlog. Disk removal failures retry cleanup without sending +an already resolved report again during the same map. The plugin cannot recover +a `game_end` that was never captured (for example, a process crash during play). + +Run `node tools/check_telemetry.mjs` for both reporters' queue/retry fixtures and +`python3 tools/test_plugin_telemetry.py` for native storage, acknowledgment and +process-restart checks with sanitizers. The native test uses each restored JSON +dependency; alternatively pass `--json-header '/path/{game}-json.hpp'` with the +single headers from each plugin's pinned revision. These tests use engine shims. +Validate Windows DLL linking and both games in Plutonium before rollout: force a +timeout/429/503 at start and end, confirm permanent 400/401 failures do not retry, +restart a map and then a process with pending boundaries, and confirm their +original match IDs complete before the next start. Also verify a 2xx without +`ok: true` stays pending, an outage retains only the latest waiting heartbeat, +and a disk failure is visible without advancing past an unsaved boundary. ## Parity @@ -141,7 +203,7 @@ cleanup. Disable the leaderboard module and verify `.leaders` is absent from hel Both reporters default to `https://ec.thewrightserver.net/api/telemetry` and send `Authorization: Bearer ` plus `Content-Type: application/json` on every -POST. Existing event fields and queue behavior are unchanged. An explicit +POST. Authentication is also used for retried/replayed reports. An explicit `tws_telemetry_url` override must also use HTTPS; remove any old HTTP override. Before upgrading the API or loading these reporters: diff --git a/t5/plugins/README.md b/t5/plugins/README.md index 706f6b3..600af41 100644 --- a/t5/plugins/README.md +++ b/t5/plugins/README.md @@ -19,7 +19,7 @@ These are libcurl transfer deadlines, starting when a worker begins the request. Time waiting for a worker or the game thread to deliver completion is measured separately. A native timeout releases the request and emits `done` with `success = false`, a transport error, and diagnostics. Each request is sent once. -The telemetry sender then logs the failed event and advances its ordered queue. +The telemetry sender retries start/end reports at the head of its ordered queue. A timeout does not prove the API did not already accept the event. The existing completion values remain in the same positions. An optional fourth @@ -31,8 +31,8 @@ request waittill("done", response, success, error, diagnostics); `success` continues to mean transport success, including completed HTTP 4xx/5xx responses. Existing scripts that read only the first three values still work. -An updated reporter on an old plugin logs that diagnostics and native deadlines -are unverified; deploy the matching DLL to enable the fix. +Current reporters also require the native telemetry helpers below; deploy the +matching DLL and restart the process before loading the updated scripts. | Field | Meaning | | --- | --- | @@ -55,20 +55,45 @@ included in diagnostics. Responses retain the existing 20 KiB truncation behavio Telemetry logs the status and timings for each completion at log level 1. Non-2xx responses log at level 0 with a bounded response preview. Health counters keep transport `completed`/`failed` and track `http_rejected` separately; an HTTP -rejection is a completed transport. Producers, queue ordering and payloads are -unchanged. Heartbeats are not coalesced, failed reports are not retried, and API -receipt timestamps remain the existing behavior. +rejection is a completed transport. `accepted` counts only 2xx replies with the +JSON boolean `ok: true`. Report identities, retry behavior, heartbeat coalescing +and persistence are described in the shared +[delivery and rollout notes](../../README.md#telemetry-delivery-and-rollout-t5-and-t6). + +## Durable telemetry helpers + +The paired `component/telemetry.cpp` and `utils/telemetry.*` provide: + +- `telemetryNewMatch()`: a random process token plus a process-local match counter. +- `telemetryLoad(server, endpoint)`: ordered saved boundary entries, or undefined + on an unreadable/corrupt outbox. An empty outbox returns an empty GSC array. +- `telemetryStore(server, endpoint, body)`: idempotently commit an exact boundary + snapshot; false on storage, validation or capacity failure. +- `telemetryRemove(server, endpoint, match, seq)`: atomically remove a resolved + entry; false on failure. The old committed file remains available on failure. +- `telemetryResponse(body)`: parse bounded JSON without throwing on malformed + responses, returning `accepted` (strict boolean `ok`) and `match_id` (a string). + The GSC reporter separately checks the HTTP status before accepting it. + +The native helpers avoid the generic JSON-to-GSC converter's boolean handling. +Outbox files are scoped by game, server and endpoint under `fs_homepath/tws-telemetry`. +They contain snapshots, never authentication headers. One writer per scope is +required. Writes replace the old file with `MoveFileExW`; stale `.tmp` files are +ignored during recovery. Native memory supplies identity only; pending reports +come from disk after every script or process restart. ## Validation and rollout Run `python3 tools/test_plugin_http.py` from the repository root for portable -native HTTP tests. Build the x86 DLL on Windows, then on a test game server check: +native HTTP tests, and `python3 tools/test_plugin_telemetry.py` for outbox and +acknowledgment tests (requires the pinned JSON headers). Deploy Electric Cherry +#144 first as described in the shared rollout notes. Build the x86 DLL on Windows, then on a test game server check: 1. Updated telemetry reports HTTP status and all timing fields on normal requests. 2. An endpoint that accepts a connection but does not reply produces a native - timeout near 15 seconds, followed by the next queued request. -3. HTTP 401/429/500 responses increment `http_rejected` without blocking the queue; - transport errors increment `failed` instead. + timeout near 15 seconds; a start/end is retried before the next queued report. +3. HTTP 401 is terminal; 429/500 retain start/end at the queue head and retry. + Non-2xx replies increment `http_rejected`; transport errors increment `failed`. 4. Leaderboard and older three-value HTTP consumers still receive completion. 5. After a full server-process restart, verify all four telemetry event types in Electric Cherry. A successful transport alone is not API acceptance. diff --git a/t5/plugins/tws-t5-gsc-utils/src/component/telemetry.cpp b/t5/plugins/tws-t5-gsc-utils/src/component/telemetry.cpp new file mode 100644 index 0000000..38713c1 --- /dev/null +++ b/t5/plugins/tws-t5-gsc-utils/src/component/telemetry.cpp @@ -0,0 +1,75 @@ +#include +#include "loader/component_loader.hpp" +#include "gsc.hpp" +#include "game/scripting/array.hpp" +#include + +namespace telemetry +{ + namespace + { + utils::telemetry::outbox spool(const std::string& server, const std::string& endpoint) + { + return {"tws-telemetry/t5", server, endpoint}; + } + + void storage_error() + { + // Do not echo parse exceptions: their text can contain player data. + printf("^1[tws_telemetry] Outbox operation failed; check tws-telemetry/t5 files, disk space and permissions. Existing reports retained.\n"); + } + } + + class component final : public component_interface + { + public: + void on_startup([[maybe_unused]] plugin::plugin* plugin) override + { + gsc::function::add("telemetrynewmatch", []() -> scripting::script_value + { + try { return utils::telemetry::new_match_key(); } + catch (...) { return {}; } + }); + gsc::function::add("telemetryload", [](const std::string& server, const std::string& endpoint) -> scripting::script_value + { + try + { + scripting::array entries; + for (const auto& value : spool(server, endpoint).load()) + { + scripting::array entry; + entry["match"] = value.match; + entry["id"] = value.seq; + entry["event"] = value.event; + entry["round"] = value.round; + entry["body"] = value.body; + entry["stored"] = true; + entries.emplace_back(entry); + } + return entries; + } + catch (...) { storage_error(); return {}; } + }); + gsc::function::add("telemetrystore", [](const std::string& server, const std::string& endpoint, const std::string& body) + { + try { spool(server, endpoint).store(body); return true; } + catch (...) { storage_error(); return false; } + }); + gsc::function::add("telemetryremove", [](const std::string& server, const std::string& endpoint, const std::string& match, const int seq) + { + try { spool(server, endpoint).erase(match, seq); return true; } + catch (...) { storage_error(); return false; } + }); + gsc::function::add("telemetryresponse", [](const std::string& response) + { + const auto ack = utils::telemetry::parse_acknowledgement(response); + scripting::array result; + result["accepted"] = ack.accepted; + result["match_id"] = ack.match_id; + return result; + }); + } + }; +} + +REGISTER_COMPONENT(telemetry::component) diff --git a/t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.cpp b/t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.cpp new file mode 100644 index 0000000..dcf1af4 --- /dev/null +++ b/t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.cpp @@ -0,0 +1,199 @@ +#include +#include "telemetry.hpp" + +#include +#include +#include +#include +#include +#include + +namespace utils::telemetry +{ + namespace + { + constexpr size_t max_body_size = 0x5000; + constexpr size_t max_file_size = 16 * 1024 * 1024; + constexpr size_t max_reports = 4096; + + // Stable, compact filename for arbitrary server names and endpoints. + // The envelope also checks both strings, so a collision fails closed. + std::string scope_name(const std::string& scope) + { + uint64_t hash = 14695981039346656037ull; + for (const unsigned char byte : scope) + hash = (hash ^ byte) * 1099511628211ull; + std::ostringstream result; + result << std::hex << std::setfill('0') << std::setw(16) << hash; + return result.str(); + } + + std::string boot_token() + { + std::random_device random; + std::ostringstream result; + result << std::hex << std::setfill('0'); + for (int i = 0; i < 4; ++i) + result << std::setw(8) << static_cast(random()); + return result.str(); + } + + int positive_int(const nlohmann::json& value) + { + if (!value.is_number_integer() || value < 1 || value > std::numeric_limits::max()) + throw std::runtime_error("invalid telemetry sequence"); + return value.get(); + } + } + + std::string new_match_key() + { + static const auto boot = boot_token(); + static std::atomic counter{0}; + return boot + "-" + std::to_string(++counter); + } + + acknowledgement parse_acknowledgement(const std::string& response) + { + acknowledgement result; + // HTTP responses at this boundary have potentially been truncated. + if (response.empty() || response.size() >= max_body_size) + return result; + const auto body = nlohmann::json::parse(response, nullptr, false); + if (!body.is_object() || !body.contains("ok") || !body["ok"].is_boolean() || !body["ok"].get()) + return result; + result.accepted = true; + const auto telemetry = body.find("telemetry"); + if (telemetry != body.end() && telemetry->is_object()) + { + const auto timing = telemetry->find("timing"); + if (timing != telemetry->end() && timing->is_object()) + { + const auto id = timing->find("matchId"); + if (id != timing->end() && id->is_number_integer() && *id > 0) + result.match_id = id->dump(); + } + } + return result; + } + + outbox::outbox(std::filesystem::path root, std::string server, std::string endpoint) + : server_(std::move(server)), endpoint_(std::move(endpoint)) + { + file_ = root / (scope_name(server_ + '\0' + endpoint_) + ".json"); + } + + report outbox::decode(const std::string& body) const + { + if (body.empty() || body.size() >= max_body_size) + throw std::runtime_error("invalid telemetry report size"); + const auto data = nlohmann::json::parse(body); + if (!data.is_object() || data.at("server") != server_ || + !data.at("map").is_string() || !data.at("players").is_array()) + throw std::runtime_error("invalid telemetry report envelope"); + report value; + value.match = data.at("match").get(); + value.seq = positive_int(data.at("seq")); + value.event = data.at("event").get(); + if (value.match.empty() || value.match.size() > 64 || + (value.event != "game_start" && value.event != "game_end") || + (value.event == "game_start" && value.seq != 1)) + throw std::runtime_error("invalid durable telemetry identity/event"); + value.round = 0; + if (value.event == "game_end") + { + const auto& round = data.at("round"); + if (!round.is_number_integer() || round < 0 || round > std::numeric_limits::max()) + throw std::runtime_error("invalid telemetry round"); + value.round = round.get(); + } + value.body = body; + return value; + } + + std::vector outbox::load() const + { + if (!std::filesystem::exists(file_)) + return {}; + if (std::filesystem::file_size(file_) > max_file_size) + throw std::runtime_error("telemetry outbox exceeds size limit"); + std::ifstream stream(file_, std::ios::binary); + if (!stream) + throw std::runtime_error("cannot open telemetry outbox"); + const auto data = nlohmann::json::parse(stream); + if (stream.bad() || !data.is_object() || data.at("version") != 1 || + data.at("server") != server_ || data.at("endpoint") != endpoint_ || + !data.at("reports").is_array() || data.at("reports").size() > max_reports) + throw std::runtime_error("invalid telemetry outbox; preserve file for recovery"); + std::vector reports; + std::set> identities; + for (const auto& body : data.at("reports")) + { + auto value = decode(body.get()); + if (!identities.emplace(value.match, value.seq).second) + throw std::runtime_error("duplicate identity in telemetry outbox"); + reports.push_back(std::move(value)); + } + return reports; + } + + void outbox::save(const std::vector& reports) const + { + if (reports.size() > max_reports) + throw std::runtime_error("telemetry outbox is full; delivery blocked"); + nlohmann::json data = {{"version", 1}, {"server", server_}, {"endpoint", endpoint_}, + {"reports", nlohmann::json::array()}}; + for (const auto& value : reports) + data["reports"].push_back(value.body); + const auto encoded = data.dump(); + if (encoded.size() > max_file_size) + throw std::runtime_error("telemetry outbox is full; delivery blocked"); + std::filesystem::create_directories(file_.parent_path()); + auto temporary = file_; + temporary += ".tmp"; + std::ofstream stream(temporary, std::ios::binary | std::ios::trunc); + stream.write(encoded.data(), static_cast(encoded.size())); + stream.flush(); + if (!stream) + throw std::runtime_error("cannot write telemetry outbox; old file retained"); + stream.close(); + if (stream.fail()) + throw std::runtime_error("cannot close telemetry outbox; old file retained"); +#ifdef _WIN32 + if (!MoveFileExW(temporary.c_str(), file_.c_str(), MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH)) + throw std::runtime_error("cannot replace telemetry outbox; old file retained"); +#else + // Used by the portable tests; the production plugin targets Windows. + std::filesystem::rename(temporary, file_); +#endif + } + + void outbox::store(const std::string& body) const + { + const auto incoming = decode(body); + auto reports = load(); + for (const auto& value : reports) + { + if (value.match != incoming.match || value.seq != incoming.seq) + continue; + if (value.body != body) + throw std::runtime_error("telemetry identity reused for a different snapshot"); + return; + } + reports.push_back(incoming); + save(reports); + } + + void outbox::erase(const std::string& match, const int seq) const + { + auto reports = load(); + const auto end = std::remove_if(reports.begin(), reports.end(), [&](const report& value) + { + return value.match == match && value.seq == seq; + }); + if (end == reports.end()) + return; + reports.erase(end, reports.end()); + save(reports); + } +} diff --git a/t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.hpp b/t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.hpp new file mode 100644 index 0000000..ee32b9f --- /dev/null +++ b/t5/plugins/tws-t5-gsc-utils/src/utils/telemetry.hpp @@ -0,0 +1,45 @@ +#pragma once + +#include +#include +#include + +namespace utils::telemetry +{ + struct report + { + std::string match; + int seq; + std::string event; + int round; + std::string body; + }; + + struct acknowledgement + { + bool accepted = false; + std::string match_id; + }; + + // Native state survives script/map shutdown; a new process gets a new token. + std::string new_match_key(); + acknowledgement parse_acknowledgement(const std::string& response); + + // One game-thread writer per server/endpoint. Commit a replacement before + // discarding the previous file; a failed write leaves it available for replay. + class outbox + { + public: + outbox(std::filesystem::path root, std::string server, std::string endpoint); + std::vector load() const; + void store(const std::string& body) const; + void erase(const std::string& match, int seq) const; + + private: + std::filesystem::path file_; + std::string server_; + std::string endpoint_; + report decode(const std::string& body) const; + void save(const std::vector& reports) const; + }; +} diff --git a/t5/scripts/README.md b/t5/scripts/README.md index d5954c1..e49ced5 100644 --- a/t5/scripts/README.md +++ b/t5/scripts/README.md @@ -333,6 +333,8 @@ The reporter emits exactly four event types: | `heartbeat` | Every 30 seconds after `game_start`, until game end | Same fields as `round_change`, with current round and player stats | | `game_end` | First stock `end_game`, `game_ended`, or `intermission` notification | `server`, `map`, final `round`, full player stats | +All four events also include `match` and `seq` (see delivery below). + Each player in round/heartbeat/end reports has `name`, `guid`, `kills`, `headshots`, `downs`, `revives`, and `score`. Names and GUIDs remain strings; GUIDs never pass through numeric conversion. Round and stats are nonnegative integers, @@ -350,14 +352,14 @@ serializes an empty GSC array as `null`; player objects and strings still use `jsonSerialize` for escaping. ```json -{"event":"game_start","server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","players":[{"name":"Josh","guid":"12345678901234567"}]} -{"event":"round_change","server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","round":17,"players":[{"name":"Josh","guid":"12345678901234567","kills":420,"headshots":160,"downs":1,"revives":3,"score":32000}]} -{"event":"heartbeat","server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","round":17,"players":[{"name":"Josh","guid":"12345678901234567","kills":430,"headshots":165,"downs":1,"revives":3,"score":32500}]} -{"event":"game_end","server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","round":18,"players":[{"name":"Josh","guid":"12345678901234567","kills":445,"headshots":170,"downs":2,"revives":3,"score":33500}]} +{"event":"game_start","match":"process-token-1","seq":1,"server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","players":[{"name":"Josh","guid":"12345678901234567"}]} +{"event":"round_change","match":"process-token-1","seq":2,"server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","round":17,"players":[{"name":"Josh","guid":"12345678901234567","kills":420,"headshots":160,"downs":1,"revives":3,"score":32000}]} +{"event":"heartbeat","match":"process-token-1","seq":3,"server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","round":17,"players":[{"name":"Josh","guid":"12345678901234567","kills":430,"headshots":165,"downs":1,"revives":3,"score":32500}]} +{"event":"game_end","match":"process-token-1","seq":4,"server":"t5_zombie_prototype_1","map":"zombie_cod5_prototype","round":18,"players":[{"name":"Josh","guid":"12345678901234567","kills":445,"headshots":170,"downs":2,"revives":3,"score":33500}]} ``` Events are captured immediately into an in-memory queue. One HTTP worker -POSTs each snapshot once, in gameplay order, with `Content-Type: +POSTs snapshots in gameplay order, with `Content-Type: application/json` and `Authorization: Bearer `, waiting for completion before sending the next. Gameplay and round detection continue while requests are pending. Repeated joins or @@ -370,35 +372,33 @@ The first heartbeat is captured 30 seconds after match start, then every changes and later spawns do not reset the interval or add heartbeat workers. Heartbeats use the same snapshot builder and ordered sender as round reports; pending HTTP requests can delay transmission. Game end stops the heartbeat -worker, and already queued heartbeats drain before `game_end`. +worker. Only the newest waiting heartbeat is retained; round reports and the +in-flight request retain their order before `game_end`. -There is no discovery, beta token, session ID, sequence field, custom event, -external helper, or separate beta acknowledgement. Delivery is best effort: -no retries or persistence across map reloads, and no cross-game exactly-once -guarantee. The [TWS plugin build](../plugins/README.md) enforces a 5-second -connection and 15-second total transfer timeout. A timed-out report is logged -and skipped so later reports can proceed. A map reload can discard queued reports. +Each report includes a native-generated `match` key and a positive `seq`, starting +at 1 for `game_start`. Start/end snapshots are written to disk before sending and +replayed before a new match after map/process restarts. Retryable failures retain +the queue head with 1–30 second backoff; other 4xx are terminal. Acceptance requires +a 2xx response containing the JSON boolean `ok: true`, and logs include Electric +Cherry's `telemetry.timing.matchId`. The native 5-second connect / 15-second total +transfer limits still apply. Ordinary round/heartbeat failures are best effort. -Transport failures and non-2xx HTTP responses log to the server console. The TWS -plugin exposes HTTP status and cumulative DNS/TCP/TLS/first-byte/total timings, -plus worker-queue and completion-notification delays. Transport success alone -does not establish API acceptance. -Verify acceptance at Electric Cherry: success is HTTP 200 with `ok: true` -and telemetry containing `receivedAt`; invalid payloads return 400 and -non-JSON content types return 415. +**Deploy Electric Cherry #144 and the updated TWS plugin before these scripts.** +See the shared [delivery and rollout notes](../../README.md#telemetry-delivery-and-rollout-t5-and-t6) +for outbox location, persistence requirements, capacity/error handling and tests. **Diagnosing a lobby that stops reporting** Default logging includes initialization, match start/end, each request's event/round and local ID, queue wait and request duration in milliseconds, -transport failures/recovery, HTTP rejection/status/timings, and the final queue drain. IDs reset on map -load and only appear in logs; payloads are unchanged. Every log includes +transport failures/recovery, HTTP rejection/status/timings, and the final queue drain. +Each log identifies the original match key and payload sequence, including replays. Every log includes the server, map, and game time in milliseconds. An independent watchdog prints health every 30 seconds, even before match start and while HTTP is pending or the game-end queue is draining. It shows -the sender stage, current snapshot capture, pending queue size (excluding -the in-flight request), total queued, completed transports, transport failures, +the sender stage, current snapshot capture, pending queue size (including +the retained in-flight head), total captured, accepted reports, completed transports, transport failures, HTTP rejections, and time since the last completion (`-1` means none). Warnings remain enabled at log level `0`: @@ -406,7 +406,7 @@ at log level `0`: - `HTTP stalled`: a request has been pending for at least 30 seconds. The warning repeats on watchdog ticks with its ID, event, age, and queue size. - `sender is not consuming`: the oldest queued report has waited at least - 30 seconds with no in-flight request. + 30 seconds with no in-flight request outside an intentional retry backoff. - `round watcher has stopped advancing`: no polling progress for over 5 seconds, detected on the next watchdog tick. - `no snapshot captured`: an active match has gone at least 65 seconds @@ -420,17 +420,12 @@ indicate a proxy error. Previews can include player identities echoed by the API. Restore `1` afterward, or use `0` to retain only warnings, errors, and transport recovery messages. -Missing request handles and reported transport failures are counted and -skipped so the sender can continue with the next event. Empty responses -produce a warning. Arbitrary response bodies are not parsed as JSON, -because malformed/proxy responses must not break the sender. The watchdog -does not launch overlapping requests or retry failed events. Native timeouts -release stalled transfers. A request still pending after 30 seconds warrants -checking the installed plugin version, worker-queue delay and completion delivery. -If diagnostics are absent, the reporter warns that native deadlines are unverified; -install the TWS plugin build and restart the server process. - -Native deadlines and completion diagnostics require the [TWS plugin build](../plugins/README.md). +Missing handles, transport errors, empty/invalid acknowledgments and retryable +HTTP responses keep start/end pending. The watchdog never launches overlapping +requests. A request pending beyond 30 seconds warrants checking the installed +plugin version, worker-queue delay and completion delivery. Native deadlines, +diagnostics and the telemetry outbox/acknowledgment helpers require the updated +[TWS plugin build](../plugins/README.md). The upstream baseline contains [`http::request`](https://github.com/alicealys/t5-gsc-utils/blob/main/src/component/http.cpp) and [`jsonSerialize`](https://github.com/alicealys/t5-gsc-utils/blob/main/src/component/json.cpp). @@ -465,10 +460,8 @@ and [score tracking](https://github.com/plutoniummod/t5-scripts/blob/main/ZM/Com though transport completes. Also check an empty or HTML response. 7. Point a test server at an endpoint that accepts a connection but does not reply. With the TWS plugin, verify a native timeout near 15 seconds and - continued queue consumption. With an older plugin, expect recurring - `HTTP stalled` warnings and increasing - pending counts while gameplay continues, including after game over. - Release the delayed reply and verify the queue resumes in order and + a retained start/end retry with bounded backoff while gameplay continues. + Restore the endpoint and verify the queue resumes in order and the watchdog stops after the final drain. Test a refused connection separately to check failure counts and the recovery log once restored. diff --git a/t5/scripts/tws_t5_telemetry.gsc b/t5/scripts/tws_t5_telemetry.gsc index 4129d53..216e287 100644 --- a/t5/scripts/tws_t5_telemetry.gsc +++ b/t5/scripts/tws_t5_telemetry.gsc @@ -45,15 +45,26 @@ init() return; } + // Recover before registering producers, so old boundaries precede this match. + recovered = telemetryLoad(level.telemetry_server, level.telemetry_url); + if (!isDefined(recovered)) + { + telemetry_log(0, "ERROR: cannot recover durable outbox; reporting disabled until storage is repaired and the map restarted"); + return; + } + level.telemetry_initialized = true; level.telemetry_started = false; level.telemetry_ended = false; level.telemetry_last_round = -1; - level.telemetry_queue = []; + level.telemetry_queue = recovered; level.telemetry_queue_read = 0; - level.telemetry_queue_write = 0; + level.telemetry_queue_write = recovered.size; + for (i = 0; i < recovered.size; i++) + level.telemetry_queue[i]["captured"] = getTime(); level.telemetry_sequence = 0; level.telemetry_completed = 0; + level.telemetry_accepted = 0; level.telemetry_failed = 0; level.telemetry_http_rejected = 0; level.telemetry_missing_diagnostics_warned = false; @@ -65,7 +76,7 @@ init() level.telemetry_sender_state = "starting"; level.telemetry_inflight = undefined; - telemetry_log(1, "Initialized; waiting for match start. log_level=" + telemetry_log_level()); + telemetry_log(1, "Initialized; recovered=" + recovered.size + " waiting for match start. log_level=" + telemetry_log_level()); // Use the shared dispatcher; a second connected listener breaks T5 SP. scripts\sp\tws_t5_common::register_spawn_handler(::telemetry_game_start); @@ -83,6 +94,12 @@ telemetry_game_start() if (level.telemetry_started || level.telemetry_ended) return; + level.telemetry_match = telemetryNewMatch(); + if (!isDefined(level.telemetry_match)) + { + telemetry_log(0, "ERROR: cannot allocate match identity"); + return; + } level.telemetry_started = true; level.telemetry_last_capture = getTime(); telemetry_log(1, "Match start detected"); @@ -146,6 +163,8 @@ telemetry_game_end_watcher() // Preserve order even if the match ends before the next polling tick. telemetry_game_start(); + if (!level.telemetry_started) + return; telemetry_report_round(); level.telemetry_ended = true; level notify("telemetry_finished"); @@ -227,30 +246,78 @@ telemetry_enqueue(event_name) level.telemetry_capture_event = event_name; telemetry_log(2, "Capturing event=" + event_name); // Snapshot and serialize before yielding, including during game over. + level.telemetry_sequence++; body = "{\"event\":" + jsonSerialize(event_name); body += ",\"server\":" + jsonSerialize(level.telemetry_server); body += ",\"map\":" + jsonSerialize(GetDvar("mapname")); + body += ",\"match\":" + jsonSerialize(level.telemetry_match); + body += ",\"seq\":" + level.telemetry_sequence; if (event_name != "game_start") body += ",\"round\":" + telemetry_nonnegative_int(level.round_number); body += ",\"players\":" + telemetry_players_json(event_name != "game_start") + "}"; entry = []; - level.telemetry_sequence++; + entry["match"] = level.telemetry_match; entry["id"] = level.telemetry_sequence; entry["event"] = event_name; entry["round"] = telemetry_nonnegative_int(level.round_number); entry["captured"] = getTime(); entry["body"] = body; - level.telemetry_queue[level.telemetry_queue_write] = entry; - level.telemetry_queue_write++; + // Persist boundaries before yielding or handing any request to the plugin. + // A failed save stays in memory and blocks sending until storage recovers. + entry["stored"] = false; + telemetry_queue_push(entry); + if (telemetry_critical(entry)) + telemetry_persist_pending(); level.telemetry_last_capture = getTime(); level.telemetry_capture_event = "none"; telemetry_log(2, "Queued " + telemetry_event_label(entry) + " pending=" + telemetry_pending()); level notify("telemetry_queued"); } -// A single worker keeps one request in flight. It survives game-over notifies -// to drain the final snapshot; no retries, discovery, or beta-specific calls. +// If storage was unavailable at an earlier capture, save it first. Otherwise a +// newly captured end could be committed ahead of its unsaved start on restart. +telemetry_persist_pending() +{ + for (i = level.telemetry_queue_read; i < level.telemetry_queue_write; i++) + { + entry = level.telemetry_queue[i]; + if (!telemetry_critical(entry) || entry["stored"]) + continue; + entry["stored"] = telemetryStore(level.telemetry_server, level.telemetry_url, entry["body"]); + if (!entry["stored"]) + return; + } +} + +telemetry_critical(entry) +{ + return entry["event"] == "game_start" || entry["event"] == "game_end"; +} + +// Remove obsolete waiting heartbeats, keeping round reports and the in-flight +// head in order. The new heartbeat keeps its capture-time sequence and snapshot. +telemetry_queue_push(entry) +{ + queue = []; + count = 0; + for (i = level.telemetry_queue_read; i < level.telemetry_queue_write; i++) + { + previous = level.telemetry_queue[i]; + inflight = i == level.telemetry_queue_read && isDefined(level.telemetry_inflight); + if (entry["event"] == "heartbeat" && previous["event"] == "heartbeat" && !inflight) + continue; + queue[count] = previous; + count++; + } + queue[count] = entry; + level.telemetry_queue = queue; + level.telemetry_queue_read = 0; + level.telemetry_queue_write = count + 1; +} + +// One worker retains the head through retries. No later report can overtake an +// unacknowledged boundary. Retry delays grow 1, 2, 4, 8, 16, then at most 30s. telemetry_sender() { for (;;) @@ -263,7 +330,7 @@ telemetry_sender() if (level.telemetry_ended) { level.telemetry_sender_state = "drained"; - telemetry_log(1, "Queue drained after game end; completed=" + level.telemetry_completed + " failed=" + level.telemetry_failed); + telemetry_log(1, "Queue drained after game end; accepted=" + level.telemetry_accepted + " failed=" + level.telemetry_failed); level notify("telemetry_drained"); return; } @@ -274,12 +341,53 @@ telemetry_sender() } entry = level.telemetry_queue[level.telemetry_queue_read]; + if (!telemetry_process(entry)) + { + delay = telemetry_retry_delay(entry); + level.telemetry_sender_state = "retry_backoff"; + telemetry_log(0, "Retry pending " + telemetry_event_label(entry) + " delay_s=" + delay); + wait (delay); + continue; + } level.telemetry_queue[level.telemetry_queue_read] = undefined; level.telemetry_queue_read++; - telemetry_send(entry); } } +telemetry_retry_delay(entry) +{ + if (!isDefined(entry["retry_delay"])) + entry["retry_delay"] = 1; + else + entry["retry_delay"] *= 2; + if (entry["retry_delay"] > 30) + entry["retry_delay"] = 30; + return entry["retry_delay"]; +} + +telemetry_process(entry) +{ + critical = telemetry_critical(entry); + if (critical && !entry["stored"]) + { + entry["stored"] = telemetryStore(level.telemetry_server, level.telemetry_url, entry["body"]); + if (!entry["stored"]) + return false; + } + // If removing an acknowledged/terminal entry fails, retry the disk update + // without POSTing again. A restart before removal safely replays its key/seq. + if (!isDefined(entry["resolved"])) + { + outcome = telemetry_send(entry); + if (critical && outcome == "retry") + return false; + entry["resolved"] = true; + } + if (critical) + return telemetryRemove(level.telemetry_server, level.telemetry_url, entry["match"], entry["id"]); + return true; +} + telemetry_send(entry) { level.telemetry_inflight = entry; @@ -302,10 +410,10 @@ telemetry_send(entry) request = http::request(level.telemetry_url, options); if (!isDefined(request)) { - telemetry_transport_failed(label + " request handle is undefined; skipping event"); + telemetry_transport_failed(label + " request handle is undefined"); level.telemetry_inflight = undefined; level.telemetry_sender_state = "idle"; - return; + return "retry"; } level.telemetry_sender_state = "waiting_http"; @@ -319,8 +427,8 @@ telemetry_send(entry) telemetry_log(0, "WARNING: plugin returned no HTTP diagnostics; native timeouts are unverified. Install the TWS plugin build and restart the server process"); } - // The plugin's success flag reports transport completion, not HTTP status. - // No automatic retry: the receiver may already have accepted the event. + // The plugin's success flag reports transport completion, not API acceptance. + outcome = "retry"; if (!isDefined(success) || !success) { if (!isDefined(error)) @@ -330,29 +438,49 @@ telemetry_send(entry) else { level.telemetry_completed++; - if (level.telemetry_failure_streak > 0) - telemetry_log(0, "Transport recovered after " + level.telemetry_failure_streak + " failures; " + label); - level.telemetry_failure_streak = 0; - if (isDefined(diagnostics) && isDefined(diagnostics["http_status"])) + outcome = telemetry_response_outcome(response, diagnostics); + if (outcome == "accepted") { - if (diagnostics["http_status"] < 200 || diagnostics["http_status"] >= 300) - { - level.telemetry_http_rejected++; - telemetry_log(0, "ERROR: HTTP rejected " + label + " elapsed_ms=" + elapsed + timing + " body=" + telemetry_preview(response)); - } - else - telemetry_log(1, "HTTP completed " + label + " elapsed_ms=" + elapsed + timing); + ack = telemetryResponse(response); + level.telemetry_accepted++; + if (level.telemetry_failure_streak > 0) + telemetry_log(0, "API recovered after " + level.telemetry_failure_streak + " failures; " + label); + level.telemetry_failure_streak = 0; + telemetry_log(1, "API accepted " + label + " ec_match_id=" + ack["match_id"] + " elapsed_ms=" + elapsed + timing); } else - telemetry_log(1, "Transport completed " + label + " elapsed_ms=" + elapsed + timing + " (API acceptance unverified)"); - if (!isDefined(response) || response == "") - telemetry_log(0, "WARNING: empty API response for " + label); - else - telemetry_log(2, "Response " + label + " body=" + telemetry_preview(response)); + { + level.telemetry_failure_streak++; + level.telemetry_http_rejected++; + telemetry_log(0, "ERROR: API unacknowledged " + label + " outcome=" + outcome + " elapsed_ms=" + elapsed + timing + " body=" + telemetry_preview(response)); + } + telemetry_log(2, "Response " + label + " body=" + telemetry_preview(response)); } + if (outcome != "accepted" && (!telemetry_critical(entry) || outcome == "drop")) + telemetry_log(0, "Dropping unaccepted " + label + " outcome=" + outcome); level.telemetry_inflight = undefined; level.telemetry_sender_state = "idle"; + return outcome; +} + +telemetry_response_outcome(response, diagnostics) +{ + if (!isDefined(diagnostics) || !isDefined(diagnostics["http_status"])) + return "retry"; + status = diagnostics["http_status"]; + if (status == 0 || status == 429 || (status >= 500 && status < 600)) + return "retry"; + if (status < 200 || status >= 300) + return "drop"; + if (!isDefined(response)) + return "retry"; + // Native nonthrowing JSON parsing requires an actual boolean true, rejects + // truncated/proxy bodies, and preserves large Electric Cherry IDs as strings. + ack = telemetryResponse(response); + if (ack["accepted"]) + return "accepted"; + return "retry"; } // The native plugin appends diagnostics while preserving the old done arguments. @@ -387,7 +515,7 @@ telemetry_pending() telemetry_event_label(entry) { - return "id=" + entry["id"] + " event=" + entry["event"] + " round=" + entry["round"]; + return "match=" + entry["match"] + " seq=" + entry["id"] + " event=" + entry["event"] + " round=" + entry["round"]; } // Read live so diagnostics can be increased without restarting the map. @@ -439,7 +567,7 @@ telemetry_watchdog() telemetry_log(1, "Health started=" + level.telemetry_started + " ended=" + level.telemetry_ended + " sender=" + level.telemetry_sender_state + " capturing=" + level.telemetry_capture_event + " pending=" + pending + " queued_total=" + level.telemetry_sequence + - " completed=" + level.telemetry_completed + " failed=" + level.telemetry_failed + " http_rejected=" + level.telemetry_http_rejected + + " accepted=" + level.telemetry_accepted + " completed=" + level.telemetry_completed + " failed=" + level.telemetry_failed + " http_rejected=" + level.telemetry_http_rejected + " last_done_ms_ago=" + completion_age); if (isDefined(level.telemetry_inflight)) @@ -450,7 +578,7 @@ telemetry_watchdog() " elapsed_ms=" + age + " pending=" + pending + " stage=" + level.telemetry_sender_state + "; native deadline exceeded; check plugin version, worker queue and completion delivery"); } - else if (pending > 0) + else if (pending > 0 && level.telemetry_sender_state != "retry_backoff") { oldest = level.telemetry_queue[level.telemetry_queue_read]; if (now - oldest["captured"] >= 30000) diff --git a/t6/plugins/README.md b/t6/plugins/README.md index 91e68c4..089d4cc 100644 --- a/t6/plugins/README.md +++ b/t6/plugins/README.md @@ -19,7 +19,7 @@ These are libcurl transfer deadlines, starting when a worker begins the request. Time waiting for a worker or the game thread to deliver completion is measured separately. A native timeout releases the request and emits `done` with `success = false`, a transport error, and diagnostics. Each request is sent once. -The telemetry sender then logs the failed event and advances its ordered queue. +The telemetry sender retries start/end reports at the head of its ordered queue. A timeout does not prove the API did not already accept the event. The existing completion values remain in the same positions. An optional fourth @@ -31,8 +31,8 @@ request waittill("done", response, success, error, diagnostics); `success` continues to mean transport success, including completed HTTP 4xx/5xx responses. Existing scripts that read only the first three values still work. -An updated reporter on an old plugin logs that diagnostics and native deadlines -are unverified; deploy the matching DLL to enable the fix. +Current reporters also require the native telemetry helpers below; deploy the +matching DLL and restart the process before loading the updated scripts. | Field | Meaning | | --- | --- | @@ -55,20 +55,45 @@ included in diagnostics. Responses retain the existing 20 KiB truncation behavio Telemetry logs the status and timings for each completion at log level 1. Non-2xx responses log at level 0 with a bounded response preview. Health counters keep transport `completed`/`failed` and track `http_rejected` separately; an HTTP -rejection is a completed transport. Producers, queue ordering and payloads are -unchanged. Heartbeats are not coalesced, failed reports are not retried, and API -receipt timestamps remain the existing behavior. +rejection is a completed transport. `accepted` counts only 2xx replies with the +JSON boolean `ok: true`. Report identities, retry behavior, heartbeat coalescing +and persistence are described in the shared +[delivery and rollout notes](../../README.md#telemetry-delivery-and-rollout-t5-and-t6). + +## Durable telemetry helpers + +The paired `component/telemetry.cpp` and `utils/telemetry.*` provide: + +- `telemetryNewMatch()`: a random process token plus a process-local match counter. +- `telemetryLoad(server, endpoint)`: ordered saved boundary entries, or undefined + on an unreadable/corrupt outbox. An empty outbox returns an empty GSC array. +- `telemetryStore(server, endpoint, body)`: idempotently commit an exact boundary + snapshot; false on storage, validation or capacity failure. +- `telemetryRemove(server, endpoint, match, seq)`: atomically remove a resolved + entry; false on failure. The old committed file remains available on failure. +- `telemetryResponse(body)`: parse bounded JSON without throwing on malformed + responses, returning `accepted` (strict boolean `ok`) and `match_id` (a string). + The GSC reporter separately checks the HTTP status before accepting it. + +The native helpers avoid the generic JSON-to-GSC converter's boolean handling. +Outbox files are scoped by game, server and endpoint under `fs_homepath/tws-telemetry`. +They contain snapshots, never authentication headers. One writer per scope is +required. Writes replace the old file with `MoveFileExW`; stale `.tmp` files are +ignored during recovery. Native memory supplies identity only; pending reports +come from disk after every script or process restart. ## Validation and rollout Run `python3 tools/test_plugin_http.py` from the repository root for portable -native HTTP tests. Build the x86 DLL on Windows, then on a test game server check: +native HTTP tests, and `python3 tools/test_plugin_telemetry.py` for outbox and +acknowledgment tests (requires the pinned JSON headers). Deploy Electric Cherry +#144 first as described in the shared rollout notes. Build the x86 DLL on Windows, then on a test game server check: 1. Updated telemetry reports HTTP status and all timing fields on normal requests. 2. An endpoint that accepts a connection but does not reply produces a native - timeout near 15 seconds, followed by the next queued request. -3. HTTP 401/429/500 responses increment `http_rejected` without blocking the queue; - transport errors increment `failed` instead. + timeout near 15 seconds; a start/end is retried before the next queued report. +3. HTTP 401 is terminal; 429/500 retain start/end at the queue head and retry. + Non-2xx replies increment `http_rejected`; transport errors increment `failed`. 4. Leaderboard and older three-value HTTP consumers still receive completion. 5. After a full server-process restart, verify all four telemetry event types in Electric Cherry. A successful transport alone is not API acceptance. diff --git a/t6/plugins/tws-t6-gsc-utils/src/component/telemetry.cpp b/t6/plugins/tws-t6-gsc-utils/src/component/telemetry.cpp new file mode 100644 index 0000000..ae2a708 --- /dev/null +++ b/t6/plugins/tws-t6-gsc-utils/src/component/telemetry.cpp @@ -0,0 +1,75 @@ +#include +#include "loader/component_loader.hpp" +#include "gsc.hpp" +#include "game/scripting/array.hpp" +#include + +namespace telemetry +{ + namespace + { + utils::telemetry::outbox spool(const std::string& server, const std::string& endpoint) + { + return {"tws-telemetry/t6", server, endpoint}; + } + + void storage_error() + { + // Do not echo parse exceptions: their text can contain player data. + printf("^1[tws_telemetry] Outbox operation failed; check tws-telemetry/t6 files, disk space and permissions. Existing reports retained.\n"); + } + } + + class component final : public component_interface + { + public: + void on_startup([[maybe_unused]] plugin::plugin* plugin) override + { + gsc::function::add("telemetrynewmatch", []() -> scripting::script_value + { + try { return utils::telemetry::new_match_key(); } + catch (...) { return {}; } + }); + gsc::function::add("telemetryload", [](const std::string& server, const std::string& endpoint) -> scripting::script_value + { + try + { + scripting::array entries; + for (const auto& value : spool(server, endpoint).load()) + { + scripting::array entry; + entry["match"] = value.match; + entry["id"] = value.seq; + entry["event"] = value.event; + entry["round"] = value.round; + entry["body"] = value.body; + entry["stored"] = true; + entries.emplace_back(entry); + } + return entries; + } + catch (...) { storage_error(); return {}; } + }); + gsc::function::add("telemetrystore", [](const std::string& server, const std::string& endpoint, const std::string& body) + { + try { spool(server, endpoint).store(body); return true; } + catch (...) { storage_error(); return false; } + }); + gsc::function::add("telemetryremove", [](const std::string& server, const std::string& endpoint, const std::string& match, const int seq) + { + try { spool(server, endpoint).erase(match, seq); return true; } + catch (...) { storage_error(); return false; } + }); + gsc::function::add("telemetryresponse", [](const std::string& response) + { + const auto ack = utils::telemetry::parse_acknowledgement(response); + scripting::array result; + result["accepted"] = ack.accepted; + result["match_id"] = ack.match_id; + return result; + }); + } + }; +} + +REGISTER_COMPONENT(telemetry::component) diff --git a/t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.cpp b/t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.cpp new file mode 100644 index 0000000..dcf1af4 --- /dev/null +++ b/t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.cpp @@ -0,0 +1,199 @@ +#include +#include "telemetry.hpp" + +#include +#include +#include +#include +#include +#include + +namespace utils::telemetry +{ + namespace + { + constexpr size_t max_body_size = 0x5000; + constexpr size_t max_file_size = 16 * 1024 * 1024; + constexpr size_t max_reports = 4096; + + // Stable, compact filename for arbitrary server names and endpoints. + // The envelope also checks both strings, so a collision fails closed. + std::string scope_name(const std::string& scope) + { + uint64_t hash = 14695981039346656037ull; + for (const unsigned char byte : scope) + hash = (hash ^ byte) * 1099511628211ull; + std::ostringstream result; + result << std::hex << std::setfill('0') << std::setw(16) << hash; + return result.str(); + } + + std::string boot_token() + { + std::random_device random; + std::ostringstream result; + result << std::hex << std::setfill('0'); + for (int i = 0; i < 4; ++i) + result << std::setw(8) << static_cast(random()); + return result.str(); + } + + int positive_int(const nlohmann::json& value) + { + if (!value.is_number_integer() || value < 1 || value > std::numeric_limits::max()) + throw std::runtime_error("invalid telemetry sequence"); + return value.get(); + } + } + + std::string new_match_key() + { + static const auto boot = boot_token(); + static std::atomic counter{0}; + return boot + "-" + std::to_string(++counter); + } + + acknowledgement parse_acknowledgement(const std::string& response) + { + acknowledgement result; + // HTTP responses at this boundary have potentially been truncated. + if (response.empty() || response.size() >= max_body_size) + return result; + const auto body = nlohmann::json::parse(response, nullptr, false); + if (!body.is_object() || !body.contains("ok") || !body["ok"].is_boolean() || !body["ok"].get()) + return result; + result.accepted = true; + const auto telemetry = body.find("telemetry"); + if (telemetry != body.end() && telemetry->is_object()) + { + const auto timing = telemetry->find("timing"); + if (timing != telemetry->end() && timing->is_object()) + { + const auto id = timing->find("matchId"); + if (id != timing->end() && id->is_number_integer() && *id > 0) + result.match_id = id->dump(); + } + } + return result; + } + + outbox::outbox(std::filesystem::path root, std::string server, std::string endpoint) + : server_(std::move(server)), endpoint_(std::move(endpoint)) + { + file_ = root / (scope_name(server_ + '\0' + endpoint_) + ".json"); + } + + report outbox::decode(const std::string& body) const + { + if (body.empty() || body.size() >= max_body_size) + throw std::runtime_error("invalid telemetry report size"); + const auto data = nlohmann::json::parse(body); + if (!data.is_object() || data.at("server") != server_ || + !data.at("map").is_string() || !data.at("players").is_array()) + throw std::runtime_error("invalid telemetry report envelope"); + report value; + value.match = data.at("match").get(); + value.seq = positive_int(data.at("seq")); + value.event = data.at("event").get(); + if (value.match.empty() || value.match.size() > 64 || + (value.event != "game_start" && value.event != "game_end") || + (value.event == "game_start" && value.seq != 1)) + throw std::runtime_error("invalid durable telemetry identity/event"); + value.round = 0; + if (value.event == "game_end") + { + const auto& round = data.at("round"); + if (!round.is_number_integer() || round < 0 || round > std::numeric_limits::max()) + throw std::runtime_error("invalid telemetry round"); + value.round = round.get(); + } + value.body = body; + return value; + } + + std::vector outbox::load() const + { + if (!std::filesystem::exists(file_)) + return {}; + if (std::filesystem::file_size(file_) > max_file_size) + throw std::runtime_error("telemetry outbox exceeds size limit"); + std::ifstream stream(file_, std::ios::binary); + if (!stream) + throw std::runtime_error("cannot open telemetry outbox"); + const auto data = nlohmann::json::parse(stream); + if (stream.bad() || !data.is_object() || data.at("version") != 1 || + data.at("server") != server_ || data.at("endpoint") != endpoint_ || + !data.at("reports").is_array() || data.at("reports").size() > max_reports) + throw std::runtime_error("invalid telemetry outbox; preserve file for recovery"); + std::vector reports; + std::set> identities; + for (const auto& body : data.at("reports")) + { + auto value = decode(body.get()); + if (!identities.emplace(value.match, value.seq).second) + throw std::runtime_error("duplicate identity in telemetry outbox"); + reports.push_back(std::move(value)); + } + return reports; + } + + void outbox::save(const std::vector& reports) const + { + if (reports.size() > max_reports) + throw std::runtime_error("telemetry outbox is full; delivery blocked"); + nlohmann::json data = {{"version", 1}, {"server", server_}, {"endpoint", endpoint_}, + {"reports", nlohmann::json::array()}}; + for (const auto& value : reports) + data["reports"].push_back(value.body); + const auto encoded = data.dump(); + if (encoded.size() > max_file_size) + throw std::runtime_error("telemetry outbox is full; delivery blocked"); + std::filesystem::create_directories(file_.parent_path()); + auto temporary = file_; + temporary += ".tmp"; + std::ofstream stream(temporary, std::ios::binary | std::ios::trunc); + stream.write(encoded.data(), static_cast(encoded.size())); + stream.flush(); + if (!stream) + throw std::runtime_error("cannot write telemetry outbox; old file retained"); + stream.close(); + if (stream.fail()) + throw std::runtime_error("cannot close telemetry outbox; old file retained"); +#ifdef _WIN32 + if (!MoveFileExW(temporary.c_str(), file_.c_str(), MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH)) + throw std::runtime_error("cannot replace telemetry outbox; old file retained"); +#else + // Used by the portable tests; the production plugin targets Windows. + std::filesystem::rename(temporary, file_); +#endif + } + + void outbox::store(const std::string& body) const + { + const auto incoming = decode(body); + auto reports = load(); + for (const auto& value : reports) + { + if (value.match != incoming.match || value.seq != incoming.seq) + continue; + if (value.body != body) + throw std::runtime_error("telemetry identity reused for a different snapshot"); + return; + } + reports.push_back(incoming); + save(reports); + } + + void outbox::erase(const std::string& match, const int seq) const + { + auto reports = load(); + const auto end = std::remove_if(reports.begin(), reports.end(), [&](const report& value) + { + return value.match == match && value.seq == seq; + }); + if (end == reports.end()) + return; + reports.erase(end, reports.end()); + save(reports); + } +} diff --git a/t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.hpp b/t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.hpp new file mode 100644 index 0000000..ee32b9f --- /dev/null +++ b/t6/plugins/tws-t6-gsc-utils/src/utils/telemetry.hpp @@ -0,0 +1,45 @@ +#pragma once + +#include +#include +#include + +namespace utils::telemetry +{ + struct report + { + std::string match; + int seq; + std::string event; + int round; + std::string body; + }; + + struct acknowledgement + { + bool accepted = false; + std::string match_id; + }; + + // Native state survives script/map shutdown; a new process gets a new token. + std::string new_match_key(); + acknowledgement parse_acknowledgement(const std::string& response); + + // One game-thread writer per server/endpoint. Commit a replacement before + // discarding the previous file; a failed write leaves it available for replay. + class outbox + { + public: + outbox(std::filesystem::path root, std::string server, std::string endpoint); + std::vector load() const; + void store(const std::string& body) const; + void erase(const std::string& match, int seq) const; + + private: + std::filesystem::path file_; + std::string server_; + std::string endpoint_; + report decode(const std::string& body) const; + void save(const std::vector& reports) const; + }; +} diff --git a/t6/scripts/README.md b/t6/scripts/README.md index 07e192d..c0df416 100644 --- a/t6/scripts/README.md +++ b/t6/scripts/README.md @@ -129,7 +129,7 @@ It sends four event types over the life of a match, mirroring the T5 script: | `round_change` | Every new positive round | new round | | `game_end` | `end_game` / `game_ended` / `intermission` | final round | -Each event payload includes `event`, `server`, `map`, `round` (except `game_start`) +Each event payload includes `event`, `server`, `map`, `match`, `seq`, `round` (except `game_start`) and a `players` array with `name`, `guid`, `kills`, `headshots`, `downs`, `revives` and `score` per player. `guid` reads Plutonium's `player.guid` engine property (T5 uses the `player getGuid()` method — T6 has no such method, but the field is @@ -137,17 +137,21 @@ equivalent for identity purposes). `kills`/`headshots`/`downs`/`revives` read from T6's stock match counters in `player.pers[]`; missing or negative values are reported as `0`. `score` is the player's current spendable `player.score`. -Delivery is asynchronous with a sequenced FIFO queue and one in-flight request. -The queue survives the game-end signal so the final snapshot drains. There is no -automatic retry: the receiver may already have accepted the event. Transport -success reports the HTTP call finished, not that the API accepted it. The TWS -plugin enforces 5-second connection and 15-second transfer deadlines, then -releases a timed-out request so the sender can advance. Telemetry logs HTTP -status and native timing diagnostics; non-2xx responses increment `http_rejected` -separately from transport failures. Older plugins trigger a warning that native -deadlines are unverified. Increase -`tws_telemetry_log_level` to `2` for per-event capture/queue/response logging; -set to `0` to silence non-error output. +Delivery uses one in-flight request and a FIFO queue. Every match receives a +native-generated key; `seq` starts at 1 for `game_start` and increases at capture. +Start/end snapshots are saved to disk before sending and replayed before the next +start after map/process restarts. Transport failures, native timeouts, HTTP 429 +and 5xx retain the head with 1–30 second backoff. Other 4xx are terminal. Round +and heartbeat failures remain best effort; waiting heartbeats collapse to the +newest snapshot. Acceptance requires 2xx plus the JSON boolean `ok: true`; logs +include `telemetry.timing.matchId` as `ec_match_id` when available. + +The TWS plugin still enforces 5-second connection and 15-second transfer deadlines. +Increase `tws_telemetry_log_level` to `2` for capture/queue/response logging; set to +`0` for errors and recovery. **Deploy Electric Cherry #144 and the updated native +plugin before these scripts.** See the shared +[delivery and rollout notes](../../README.md#telemetry-delivery-and-rollout-t5-and-t6) +for outbox storage, error handling, limits and live validation. ## `.join` for newcomers diff --git a/t6/scripts/tws_t6_telemetry.gsc b/t6/scripts/tws_t6_telemetry.gsc index 30c1fb4..2a4da06 100644 --- a/t6/scripts/tws_t6_telemetry.gsc +++ b/t6/scripts/tws_t6_telemetry.gsc @@ -50,15 +50,26 @@ init() return; } + // Recover before registering producers, so old boundaries precede this match. + recovered = telemetryLoad(level.telemetry_server, level.telemetry_url); + if (!isDefined(recovered)) + { + telemetry_log(0, "ERROR: cannot recover durable outbox; reporting disabled until storage is repaired and the map restarted"); + return; + } + level.telemetry_initialized = true; level.telemetry_started = false; level.telemetry_ended = false; level.telemetry_last_round = -1; - level.telemetry_queue = []; + level.telemetry_queue = recovered; level.telemetry_queue_read = 0; - level.telemetry_queue_write = 0; + level.telemetry_queue_write = recovered.size; + for (i = 0; i < recovered.size; i++) + level.telemetry_queue[i]["captured"] = getTime(); level.telemetry_sequence = 0; level.telemetry_completed = 0; + level.telemetry_accepted = 0; level.telemetry_failed = 0; level.telemetry_http_rejected = 0; level.telemetry_missing_diagnostics_warned = false; @@ -70,7 +81,7 @@ init() level.telemetry_sender_state = "starting"; level.telemetry_inflight = undefined; - telemetry_log(1, "Initialized; waiting for match start. log_level=" + telemetry_log_level()); + telemetry_log(1, "Initialized; recovered=" + recovered.size + " waiting for match start. log_level=" + telemetry_log_level()); // Route through the shared dispatcher; a duplicate connected listener // would fight the T6 common spawn dispatcher. @@ -89,6 +100,12 @@ telemetry_game_start() if (level.telemetry_started || level.telemetry_ended) return; + level.telemetry_match = telemetryNewMatch(); + if (!isDefined(level.telemetry_match)) + { + telemetry_log(0, "ERROR: cannot allocate match identity"); + return; + } level.telemetry_started = true; level.telemetry_last_capture = getTime(); telemetry_log(1, "Match start detected"); @@ -153,6 +170,8 @@ telemetry_game_end_watcher() // Preserve order even if the match ends before the next polling tick. telemetry_game_start(); + if (!level.telemetry_started) + return; telemetry_report_round(); level.telemetry_ended = true; level notify("telemetry_finished"); @@ -240,30 +259,78 @@ telemetry_enqueue(event_name) level.telemetry_capture_event = event_name; telemetry_log(2, "Capturing event=" + event_name); // Snapshot and serialize before yielding, including during game over. + level.telemetry_sequence++; body = "{\"event\":" + jsonSerialize(event_name); body += ",\"server\":" + jsonSerialize(level.telemetry_server); body += ",\"map\":" + jsonSerialize(GetDvar("mapname")); + body += ",\"match\":" + jsonSerialize(level.telemetry_match); + body += ",\"seq\":" + level.telemetry_sequence; if (event_name != "game_start") body += ",\"round\":" + telemetry_nonnegative_int(level.round_number); body += ",\"players\":" + telemetry_players_json(event_name != "game_start") + "}"; entry = []; - level.telemetry_sequence++; + entry["match"] = level.telemetry_match; entry["id"] = level.telemetry_sequence; entry["event"] = event_name; entry["round"] = telemetry_nonnegative_int(level.round_number); entry["captured"] = getTime(); entry["body"] = body; - level.telemetry_queue[level.telemetry_queue_write] = entry; - level.telemetry_queue_write++; + // Persist boundaries before yielding or handing any request to the plugin. + // A failed save stays in memory and blocks sending until storage recovers. + entry["stored"] = false; + telemetry_queue_push(entry); + if (telemetry_critical(entry)) + telemetry_persist_pending(); level.telemetry_last_capture = getTime(); level.telemetry_capture_event = "none"; telemetry_log(2, "Queued " + telemetry_event_label(entry) + " pending=" + telemetry_pending()); level notify("telemetry_queued"); } -// A single worker keeps one request in flight. It survives game-over notifies -// to drain the final snapshot; no retries, discovery, or beta-specific calls. +// If storage was unavailable at an earlier capture, save it first. Otherwise a +// newly captured end could be committed ahead of its unsaved start on restart. +telemetry_persist_pending() +{ + for (i = level.telemetry_queue_read; i < level.telemetry_queue_write; i++) + { + entry = level.telemetry_queue[i]; + if (!telemetry_critical(entry) || entry["stored"]) + continue; + entry["stored"] = telemetryStore(level.telemetry_server, level.telemetry_url, entry["body"]); + if (!entry["stored"]) + return; + } +} + +telemetry_critical(entry) +{ + return entry["event"] == "game_start" || entry["event"] == "game_end"; +} + +// Remove obsolete waiting heartbeats, keeping round reports and the in-flight +// head in order. The new heartbeat keeps its capture-time sequence and snapshot. +telemetry_queue_push(entry) +{ + queue = []; + count = 0; + for (i = level.telemetry_queue_read; i < level.telemetry_queue_write; i++) + { + previous = level.telemetry_queue[i]; + inflight = i == level.telemetry_queue_read && isDefined(level.telemetry_inflight); + if (entry["event"] == "heartbeat" && previous["event"] == "heartbeat" && !inflight) + continue; + queue[count] = previous; + count++; + } + queue[count] = entry; + level.telemetry_queue = queue; + level.telemetry_queue_read = 0; + level.telemetry_queue_write = count + 1; +} + +// One worker retains the head through retries. No later report can overtake an +// unacknowledged boundary. Retry delays grow 1, 2, 4, 8, 16, then at most 30s. telemetry_sender() { for (;;) @@ -276,7 +343,7 @@ telemetry_sender() if (level.telemetry_ended) { level.telemetry_sender_state = "drained"; - telemetry_log(1, "Queue drained after game end; completed=" + level.telemetry_completed + " failed=" + level.telemetry_failed); + telemetry_log(1, "Queue drained after game end; accepted=" + level.telemetry_accepted + " failed=" + level.telemetry_failed); level notify("telemetry_drained"); return; } @@ -287,15 +354,53 @@ telemetry_sender() } entry = level.telemetry_queue[level.telemetry_queue_read]; + if (!telemetry_process(entry)) + { + delay = telemetry_retry_delay(entry); + level.telemetry_sender_state = "retry_backoff"; + telemetry_log(0, "Retry pending " + telemetry_event_label(entry) + " delay_s=" + delay); + wait (delay); + continue; + } level.telemetry_queue[level.telemetry_queue_read] = undefined; level.telemetry_queue_read++; - telemetry_send(entry); } } -// v3.1.1 global alias: httpPost(url, options) with options["method"|"headers"|"body"]. -// The plugin exposes no cancellation API and no HTTP status code; treat the -// success flag as transport completion only and never overlap requests. +telemetry_retry_delay(entry) +{ + if (!isDefined(entry["retry_delay"])) + entry["retry_delay"] = 1; + else + entry["retry_delay"] *= 2; + if (entry["retry_delay"] > 30) + entry["retry_delay"] = 30; + return entry["retry_delay"]; +} + +telemetry_process(entry) +{ + critical = telemetry_critical(entry); + if (critical && !entry["stored"]) + { + entry["stored"] = telemetryStore(level.telemetry_server, level.telemetry_url, entry["body"]); + if (!entry["stored"]) + return false; + } + // If removing an acknowledged/terminal entry fails, retry the disk update + // without POSTing again. A restart before removal safely replays its key/seq. + if (!isDefined(entry["resolved"])) + { + outcome = telemetry_send(entry); + if (critical && outcome == "retry") + return false; + entry["resolved"] = true; + } + if (critical) + return telemetryRemove(level.telemetry_server, level.telemetry_url, entry["match"], entry["id"]); + return true; +} + telemetry_send(entry) { level.telemetry_inflight = entry; @@ -318,10 +423,10 @@ telemetry_send(entry) request = httpPost(level.telemetry_url, options); if (!isDefined(request)) { - telemetry_transport_failed(label + " request handle is undefined; skipping event"); + telemetry_transport_failed(label + " request handle is undefined"); level.telemetry_inflight = undefined; level.telemetry_sender_state = "idle"; - return; + return "retry"; } level.telemetry_sender_state = "waiting_http"; @@ -335,8 +440,8 @@ telemetry_send(entry) telemetry_log(0, "WARNING: plugin returned no HTTP diagnostics; native timeouts are unverified. Install the TWS plugin build and restart the server process"); } - // The plugin's success flag reports transport completion, not HTTP status. - // No automatic retry: the receiver may already have accepted the event. + // The plugin's success flag reports transport completion, not API acceptance. + outcome = "retry"; if (!isDefined(success) || !success) { if (!isDefined(error)) @@ -346,29 +451,49 @@ telemetry_send(entry) else { level.telemetry_completed++; - if (level.telemetry_failure_streak > 0) - telemetry_log(0, "Transport recovered after " + level.telemetry_failure_streak + " failures; " + label); - level.telemetry_failure_streak = 0; - if (isDefined(diagnostics) && isDefined(diagnostics["http_status"])) + outcome = telemetry_response_outcome(response, diagnostics); + if (outcome == "accepted") { - if (diagnostics["http_status"] < 200 || diagnostics["http_status"] >= 300) - { - level.telemetry_http_rejected++; - telemetry_log(0, "ERROR: HTTP rejected " + label + " elapsed_ms=" + elapsed + timing + " body=" + telemetry_preview(response)); - } - else - telemetry_log(1, "HTTP completed " + label + " elapsed_ms=" + elapsed + timing); + ack = telemetryResponse(response); + level.telemetry_accepted++; + if (level.telemetry_failure_streak > 0) + telemetry_log(0, "API recovered after " + level.telemetry_failure_streak + " failures; " + label); + level.telemetry_failure_streak = 0; + telemetry_log(1, "API accepted " + label + " ec_match_id=" + ack["match_id"] + " elapsed_ms=" + elapsed + timing); } else - telemetry_log(1, "Transport completed " + label + " elapsed_ms=" + elapsed + timing + " (API acceptance unverified)"); - if (!isDefined(response) || response == "") - telemetry_log(0, "WARNING: empty API response for " + label); - else - telemetry_log(2, "Response " + label + " body=" + telemetry_preview(response)); + { + level.telemetry_failure_streak++; + level.telemetry_http_rejected++; + telemetry_log(0, "ERROR: API unacknowledged " + label + " outcome=" + outcome + " elapsed_ms=" + elapsed + timing + " body=" + telemetry_preview(response)); + } + telemetry_log(2, "Response " + label + " body=" + telemetry_preview(response)); } + if (outcome != "accepted" && (!telemetry_critical(entry) || outcome == "drop")) + telemetry_log(0, "Dropping unaccepted " + label + " outcome=" + outcome); level.telemetry_inflight = undefined; level.telemetry_sender_state = "idle"; + return outcome; +} + +telemetry_response_outcome(response, diagnostics) +{ + if (!isDefined(diagnostics) || !isDefined(diagnostics["http_status"])) + return "retry"; + status = diagnostics["http_status"]; + if (status == 0 || status == 429 || (status >= 500 && status < 600)) + return "retry"; + if (status < 200 || status >= 300) + return "drop"; + if (!isDefined(response)) + return "retry"; + // Native nonthrowing JSON parsing requires an actual boolean true, rejects + // truncated/proxy bodies, and preserves large Electric Cherry IDs as strings. + ack = telemetryResponse(response); + if (ack["accepted"]) + return "accepted"; + return "retry"; } // The native plugin appends diagnostics while preserving the old done arguments. @@ -403,7 +528,7 @@ telemetry_pending() telemetry_event_label(entry) { - return "id=" + entry["id"] + " event=" + entry["event"] + " round=" + entry["round"]; + return "match=" + entry["match"] + " seq=" + entry["id"] + " event=" + entry["event"] + " round=" + entry["round"]; } // Read live so diagnostics can be increased without restarting the map. @@ -455,7 +580,7 @@ telemetry_watchdog() telemetry_log(1, "Health started=" + level.telemetry_started + " ended=" + level.telemetry_ended + " sender=" + level.telemetry_sender_state + " capturing=" + level.telemetry_capture_event + " pending=" + pending + " queued_total=" + level.telemetry_sequence + - " completed=" + level.telemetry_completed + " failed=" + level.telemetry_failed + " http_rejected=" + level.telemetry_http_rejected + + " accepted=" + level.telemetry_accepted + " completed=" + level.telemetry_completed + " failed=" + level.telemetry_failed + " http_rejected=" + level.telemetry_http_rejected + " last_done_ms_ago=" + completion_age); if (isDefined(level.telemetry_inflight)) @@ -466,7 +591,7 @@ telemetry_watchdog() " elapsed_ms=" + age + " pending=" + pending + " stage=" + level.telemetry_sender_state + "; native deadline exceeded; check plugin version, worker queue and completion delivery"); } - else if (pending > 0) + else if (pending > 0 && level.telemetry_sender_state != "retry_backoff") { oldest = level.telemetry_queue[level.telemetry_queue_read]; if (now - oldest["captured"] >= 30000) diff --git a/tools/check_telemetry.mjs b/tools/check_telemetry.mjs new file mode 100644 index 0000000..4862554 --- /dev/null +++ b/tools/check_telemetry.mjs @@ -0,0 +1,283 @@ +// Run the actual reporter queue/response functions with HTTP, disk and VM shims. +// This validates state transitions, not engine linking or live gameplay. +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import vm from 'node:vm'; + +const root = new URL('../', import.meta.url); +let checks = 0; +function check(name, run) { + try { run(); checks++; } + catch (error) { throw new Error(name, { cause: error }); } +} + +function functions(source) { + const literals = []; + const protectedSource = source.replace(/"(?:\\.|[^"\\])*"|\/\/[^\n]*|\/\*[\s\S]*?\*\//g, token => { + if (token.startsWith('/')) return ''; + literals.push(token); + return `__literal_${literals.length - 1}__`; + }); + const found = new Map(); + const headers = /^(telemetry_\w+)\(([^\n]*)\)\s*\n\{/gm; + for (let match; (match = headers.exec(protectedSource));) { + let end = headers.lastIndex, depth = 1; + while (depth && end < protectedSource.length) { + const ch = protectedSource[end++]; + if (ch === '{') depth++; + if (ch === '}') depth--; + } + assert.equal(depth, 0); + let body = protectedSource.slice(headers.lastIndex, end - 1); + const locals = new Set([...body.matchAll(/(? x[1])); + body = body.replace(/\.size\b/g, '.length').replace(/http::request/g, 'httpPost'); + body = body.replace(/level thread \w+\(\);/g, '').replace(/level notify\(/g, 'notify('); + body = body.replace(/level waittill\(([^;]+)\);/g, 'yield $1;'); + body = body.replace(/wait \(([^;]+)\);/g, 'yield $1;'); + body = body.replace(/request waittill\([^;]+\);/g, 'var [response, success, error, diagnostics] = httpComplete(request);'); + body = body.replace(/__literal_(\d+)__/g, (_, index) => literals[Number(index)]); + found.set(match[1], `function${match[1] === 'telemetry_sender' ? '*' : ''} ${match[1]}(${match[2]}) { var ${[...locals].join(', ') || 'unused'}; ${body} }`); + } + return found; +} + +function disk() { + return { entries: [], failStore: false, failRemove: false, calls: [] }; +} + +let nextMatch = 0; +function engine(game, storage = disk()) { + const source = readFileSync(new URL(`${game}/scripts/tws_${game}_telemetry.gsc`, root), 'utf8'); + const all = functions(source); + const excluded = new Set(['telemetry_players_json', 'telemetry_stat', 'telemetry_heartbeat_watcher', + 'telemetry_round_watcher', 'telemetry_game_end_watcher', 'telemetry_watchdog', 'telemetry_log']); + const sent = [], responses = [], logs = [], notifications = []; + const queue = storage.entries.map(e => ({ ...e, stored: true, captured: 1000 })); + const c = vm.createContext({ + level: { telemetry_server: 'test-server', telemetry_url: 'https://test.invalid/telemetry', telemetry_token: 'test-token', + telemetry_started: false, telemetry_ended: false, telemetry_last_round: -1, round_number: 0, + telemetry_queue: queue, telemetry_queue_read: 0, telemetry_queue_write: queue.length, + telemetry_sequence: 0, telemetry_failed: 0, telemetry_completed: 0, telemetry_accepted: 0, + telemetry_http_rejected: 0, telemetry_failure_streak: 0, telemetry_sender_state: 'starting' }, + GetDvar: key => key === 'mapname' ? 'test-map' : '', + getTime: () => 1000, + isDefined: value => value !== undefined && value !== null, + int: Math.trunc, + jsonSerialize: JSON.stringify, + getSubStr: (value, start, end) => value.slice(start, end), + strTok: (value, delimiter) => value.split(delimiter), + telemetry_players_json: () => '[]', + telemetry_log: (verbosity, text) => logs.push(text), + notify: event => notifications.push(event), + telemetryNewMatch: () => `boot-${++nextMatch}`, + telemetryStore: (server, url, body) => { + storage.calls.push('store'); + if (storage.failStore) return false; + const data = JSON.parse(body); + const existing = storage.entries.find(e => e.match === data.match && e.id === data.seq); + if (existing) assert.equal(existing.body, body); + else storage.entries.push({ match: data.match, id: data.seq, event: data.event, round: data.round ?? 0, body }); + return true; + }, + telemetryRemove: (server, url, match, seq) => { + storage.calls.push('remove'); + if (storage.failRemove) return false; + storage.entries = storage.entries.filter(e => e.match !== match || e.id !== seq); + return true; + }, + // Native parser behavior is tested separately against the compiled C++. + telemetryResponse: response => { + try { + const data = JSON.parse(response); + return { accepted: data?.ok === true && response.length < 20480, match_id: String(data?.telemetry?.timing?.matchId ?? '') }; + } catch { return { accepted: false, match_id: '' }; } + }, + httpPost: (url, options) => { + storage.calls.push('send'); + sent.push({ ...options, headers: { ...options.headers } }); + return responses[0]?.noHandle ? undefined : {}; + }, + httpComplete: () => { + assert(responses.length, 'unexpected HTTP request'); + const response = responses.shift(); + response.during?.(); + return [response.body ?? '{"ok":true,"telemetry":{"timing":{"matchId":921}}}', response.success ?? true, + response.error, response.missingDiagnostics ? undefined : { http_status: response.status ?? 200 }]; + }, + }); + vm.runInContext([...all].filter(([name]) => !excluded.has(name)).map(([, code]) => code).join('\n'), c); + return { c, all, source, sent, responses, logs, storage, notifications }; +} + +for (const game of ['t5', 't6']) { + check(`${game}: match identity, sequence and capture before send`, () => { + const { c, storage, sent, responses } = engine(game); + c.telemetry_game_start(); c.telemetry_game_start(); + c.level.round_number = 1; + c.telemetry_report_round(); c.telemetry_enqueue('heartbeat'); c.telemetry_enqueue('game_end'); + assert.equal(c.level.telemetry_queue_write, 4); + const events = c.level.telemetry_queue.map(e => JSON.parse(e.body)); + assert.deepEqual(Array.from(events, e => e.seq), [1, 2, 3, 4]); + assert(events.every(e => e.match === events[0].match)); + assert(!('round' in events[0])); + assert(events.every(e => Array.isArray(e.players))); + assert.equal(storage.entries.length, 2); + assert.equal(sent.length, 0); + responses.push({}, {}, {}, {}); + c.level.telemetry_ended = true; + assert.equal(c.telemetry_sender().next().done, true); + assert.equal(storage.entries.length, 0); + assert.equal(c.level.telemetry_accepted, 4); + assert(sent.every(e => e.connect_timeout_ms === 5000 && e.timeout_ms === 15000)); + assert.equal(sent[0].headers.Authorization, 'Bearer test-token'); + assert(storage.calls.indexOf('store') < storage.calls.indexOf('send')); + }); + + check(`${game}: retries retain FIFO head and exact snapshot, coalesce heartbeats`, () => { + const { c, sent, responses, logs } = engine(game); + c.telemetry_game_start(); + const initial = c.level.telemetry_queue[0].body; + responses.push({ success: false, error: 'timeout' }, { status: 429 }, { status: 503 }, {}, {}, {}, {}); + const sender = c.telemetry_sender(); + assert.equal(sender.next().value, 1); + c.telemetry_enqueue('heartbeat'); + c.level.round_number = 2; c.telemetry_report_round(); + c.telemetry_enqueue('heartbeat'); c.telemetry_enqueue('heartbeat'); + c.telemetry_enqueue('game_end'); c.level.telemetry_ended = true; + assert.equal(sender.next().value, 2); + assert.equal(sender.next().value, 4); + assert.equal(sender.next().done, true); + const events = sent.map(s => JSON.parse(s.body)); + assert(sent.slice(0, 4).every(s => s.body === initial)); + assert.deepEqual(events.slice(4).map(e => [e.event, e.seq]), [['round_change', 3], ['heartbeat', 5], ['game_end', 6]]); + assert(logs.some(line => line.includes('ec_match_id=921'))); + assert(logs.some(line => line.includes('API recovered'))); + }); + + check(`${game}: response classification and permanent rejection`, () => { + const { c, responses, storage } = engine(game); + for (const status of [200, 201, 204, 299]) assert.equal(c.telemetry_response_outcome('{"ok":true}', { http_status: status }), 'accepted'); + for (const status of [0, 429, 500, 502, 503, 599]) assert.equal(c.telemetry_response_outcome('{"ok":true}', { http_status: status }), 'retry'); + for (const status of [301, 400, 401, 403, 404, 408, 415, 422]) assert.equal(c.telemetry_response_outcome('{"ok":true}', { http_status: status }), 'drop'); + for (const body of ['', '', '{"ok":false}', '{"ok":1}', '{"ok":"true"}']) assert.equal(c.telemetry_response_outcome(body, { http_status: 200 }), 'retry'); + assert.equal(c.telemetry_response_outcome('{"ok":true}', undefined), 'retry'); + c.telemetry_game_start(); responses.push({ status: 401 }); + assert.equal(c.telemetry_sender().next().value, 'telemetry_queued'); + assert.equal(storage.entries.length, 0); + assert.equal(c.level.telemetry_accepted, 0); + }); + + check(`${game}: backoff is capped and a missing handle retains the boundary`, () => { + const { c, responses, storage } = engine(game); + c.telemetry_game_start(); responses.push({ noHandle: true }); + const sender = c.telemetry_sender(); + const delays = Array.from({ length: 9 }, () => sender.next().value); + assert.deepEqual(delays, [1, 2, 4, 8, 16, 30, 30, 30, 30]); + assert.equal(storage.entries.length, 1); + assert.equal(c.level.telemetry_queue_read, 0); + }); + + check(`${game}: failed ordinary report is skipped and in-flight heartbeat stays stable`, () => { + const { c, responses, sent } = engine(game); + c.level.telemetry_match = 'match'; + c.telemetry_enqueue('heartbeat'); + responses.push({ success: false, during: () => { c.telemetry_enqueue('heartbeat'); c.telemetry_enqueue('heartbeat'); } }, {}); + assert.equal(c.telemetry_sender().next().value, 'telemetry_queued'); + assert.deepEqual(sent.map(s => JSON.parse(s.body).seq), [1, 3]); + }); + + check(`${game}: replay survives map/process reset and precedes new start`, () => { + const first = engine(game); + first.c.telemetry_game_start(); first.c.telemetry_enqueue('heartbeat'); first.c.telemetry_enqueue('game_end'); + first.responses.push({ success: false }); first.c.telemetry_sender().next(); + const replay = engine(game, first.storage); + replay.c.telemetry_game_start(); + replay.responses.push({}, {}, {}); + assert.equal(replay.c.telemetry_sender().next().value, 'telemetry_queued'); + const bodies = replay.sent.map(s => JSON.parse(s.body)); + assert.deepEqual(bodies.map(b => b.event), ['game_start', 'game_end', 'game_start']); + assert.equal(bodies[0].match, bodies[1].match); + assert.notEqual(bodies[0].match, bodies[2].match); + assert.equal(bodies[2].seq, 1); + assert.equal(replay.sent[0].body, first.sent[0].body); + assert.equal(first.storage.entries.length, 0); + assert(replay.source.indexOf('telemetryLoad(') < replay.source.indexOf('register_spawn_handler(')); + }); + + check(`${game}: disk failure preserves ordering and blocks sending`, () => { + const { c, storage, sent, responses } = engine(game); + storage.failStore = true; + c.telemetry_game_start(); c.telemetry_enqueue('game_end'); + const sender = c.telemetry_sender(); + assert.equal(sender.next().value, 1); + assert.equal(sent.length, 0); + storage.failStore = false; + c.telemetry_persist_pending(); + assert.deepEqual(storage.entries.map(e => e.event), ['game_start', 'game_end']); + storage.failRemove = true; + responses.push({}); + assert.equal(sender.next().value, 2); + assert.equal(sent.length, 1); + assert.equal(sender.next().value, 4); + assert.equal(sent.length, 1); // Only retry deletion after an acknowledgment. + storage.failRemove = false; responses.push({}); + assert.equal(sender.next().value, 'telemetry_queued'); + assert.equal(storage.entries.length, 0); + assert.equal(sent.length, 2); + }); + + check(`${game}: end captured behind a stalled round survives map unload`, () => { + const first = engine(game); + first.c.telemetry_game_start(); first.responses.push({}); + first.c.telemetry_sender().next(); + assert.equal(first.storage.entries.length, 0); + first.c.level.round_number = 12; first.c.telemetry_report_round(); + const unload = new Error('map unloaded during HTTP wait'); + first.responses.push({ during: () => { + first.c.telemetry_enqueue('game_end'); + throw unload; + } }); + assert.throws(() => first.c.telemetry_sender().next(), error => error === unload); + assert.deepEqual(first.storage.entries.map(e => e.event), ['game_end']); + const replay = engine(game, first.storage); + replay.c.telemetry_game_start(); + replay.responses.push({ status: 500 }, {}, {}); + const sender = replay.c.telemetry_sender(); + assert.equal(sender.next().value, 1); + assert.equal(sender.next().value, 'telemetry_queued'); + const events = replay.sent.map(s => JSON.parse(s.body)); + assert.deepEqual(events.map(e => e.event), ['game_end', 'game_end', 'game_start']); + assert.equal(events[0].round, 12); + assert.equal(replay.sent[0].body, replay.sent[1].body); + assert.notEqual(events[0].match, events[2].match); + }); + + check(`${game}: acknowledgment lost at restart replays the identical end`, () => { + const first = engine(game); + first.c.telemetry_game_start(); first.responses.push({}); + first.c.telemetry_sender().next(); + first.c.telemetry_enqueue('game_end'); + first.storage.failRemove = true; + first.responses.push({}); + assert.equal(first.c.telemetry_sender().next().value, 1); + assert.equal(first.c.level.telemetry_accepted, 2); + first.storage.failRemove = false; + const replay = engine(game, first.storage); + replay.responses.push({}); + replay.c.telemetry_sender().next(); + assert.equal(replay.sent[0].body, first.sent[1].body); + assert.equal(first.storage.entries.length, 0); + }); +} + +check('paired queue and delivery implementations match', () => { + const t5 = engine('t5').all, t6 = engine('t6').all; + for (const name of ['telemetry_enqueue', 'telemetry_queue_push', 'telemetry_critical', 'telemetry_persist_pending', + 'telemetry_sender', 'telemetry_process', 'telemetry_retry_delay', 'telemetry_send', 'telemetry_response_outcome']) + assert.equal(t5.get(name), t6.get(name), name); + for (const file of ['telemetry.hpp', 'telemetry.cpp']) { + const get = game => readFileSync(new URL(`${game}/plugins/tws-${game}-gsc-utils/src/utils/${file}`, root), 'utf8'); + assert.equal(get('t5'), get('t6'), file); + } +}); +console.log(`PASS ${checks} telemetry queue/response checks (both games). VM linking/gameplay still require Plutonium.`); diff --git a/tools/test_plugin_telemetry.py b/tools/test_plugin_telemetry.py new file mode 100644 index 0000000..342fd27 --- /dev/null +++ b/tools/test_plugin_telemetry.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""Test native telemetry storage/acknowledgments with ASan/UBSan, without the VM. + +Uses deps/json/single_include/nlohmann/json.hpp if restored, or --json-header +pointing to the single header at the revision in the plugin's upstream.json. +""" +import argparse +import os +from pathlib import Path +import re +import shutil +import subprocess +import tempfile + +ROOT = Path(__file__).resolve().parents[1] +parser = argparse.ArgumentParser(description=__doc__) +parser.add_argument("--json-header", help="Header path; {game} expands to t5/t6 for their respective pinned headers") +args = parser.parse_args() + +with tempfile.TemporaryDirectory(prefix="tws-telemetry-") as temp: + directory = Path(temp) + for game in ("t5", "t6"): + plugin = ROOT / game / "plugins" / f"tws-{game}-gsc-utils" + header = Path(args.json_header.format(game=game)) if args.json_header else plugin / "deps/json/single_include/nlohmann/json.hpp" + if not header.is_file(): + parser.error(f"Missing {header}; restore dependencies or pass --json-header") + shutil.copyfile(header, directory / "json.hpp") + (directory / "stdinc.hpp").write_text("#include \n#include \n#include \n#include \n") + executable = directory / game + subprocess.run([os.environ.get("CXX", "c++"), "-std=c++20", "-Wall", "-Wextra", "-Werror", + "-fsanitize=address,undefined", "-I", str(directory), "-I", str(plugin / "src/utils"), + str(plugin / "src/utils/telemetry.cpp"), str(ROOT / "tools/tests/telemetry_cases.cpp"), + "-o", str(executable)], check=True) + work = directory / f"{game}-data" + subprocess.run([str(executable), str(work)], check=True) + # Separate OS processes prove snapshots/identities don't depend on VM or + # DLL memory. A newly booted producer recovers before its own start. + subprocess.run([str(executable), str(work), "produce"], check=True) + subprocess.run([str(executable), str(work), "recover"], check=True) + subprocess.run([str(executable), str(work), "empty"], check=True) + + # Compile the actual registration adapter as well, substituting only + # engine interfaces. This does not validate the Windows ABI/VM linker. + stub = (ROOT / "tools/tests/http_component_stub.hpp").read_text() + stub = stub.replace("std::vector get_keys() const", "void emplace_back(const script_value& value) { (*data)[std::to_string(data->size())] = value; }\n std::vector get_keys() const") + (directory / "component_stub.hpp").write_text(stub) + component = re.sub(r'^#include[^\n]*\n', '', (plugin / "src/component/telemetry.cpp").read_text(), flags=re.MULTILINE) + source = directory / "component.cpp" + source.write_text('#include "component_stub.hpp"\n#include "telemetry.hpp"\n' + component) + subprocess.run([os.environ.get("CXX", "c++"), "-std=c++20", "-Wall", "-Wextra", "-Werror", "-fsyntax-only", + "-I", str(directory), "-I", str(plugin / "src/utils"), str(source)], check=True) + print(f"PASS {game}: native storage, acknowledgment, process restart and component checks", flush=True) diff --git a/tools/tests/telemetry_cases.cpp b/tools/tests/telemetry_cases.cpp new file mode 100644 index 0000000..031b5a8 --- /dev/null +++ b/tools/tests/telemetry_cases.cpp @@ -0,0 +1,126 @@ +#include "telemetry.hpp" +#include +#include +#include +#include +#include + +namespace fs = std::filesystem; +using utils::telemetry::outbox; +using utils::telemetry::new_match_key; +using utils::telemetry::parse_acknowledgement; + +const std::string server = "t6_zombie_buried_1"; +const std::string endpoint = "https://example.test/api/telemetry"; + +std::string body(const std::string& match, int seq, const std::string& event) +{ + nlohmann::json value = {{"server", server}, {"map", "zm_buried"}, {"match", match}, + {"seq", seq}, {"event", event}, {"players", nlohmann::json::array()}}; + if (event != "game_start") value["round"] = 12; + return value.dump(); +} + +template void must_fail(F action) +{ + bool failed = false; + try { action(); } catch (const std::exception&) { failed = true; } + assert(failed); +} + +int main(int argc, char** argv) +{ + assert(argc >= 2); + const fs::path root(argv[1]); + const outbox queue(root, server, endpoint); + if (argc == 3) + { + const std::string mode(argv[2]); + if (mode == "produce") + { + const auto match = new_match_key(); + queue.store(body(match, 1, "game_start")); + queue.store(body(match, 14, "game_end")); + } + else if (mode == "recover") + { + const auto reports = queue.load(); + assert(reports.size() == 2); + const auto next = new_match_key(); + assert(reports[0].match != next); + queue.store(body(next, 1, "game_start")); + const auto all = queue.load(); + assert(all.size() == 3 && all[0].event == "game_start" && all[1].event == "game_end"); + assert(all[0].match == all[1].match && all[2].match == next); + for (const auto& value : all) queue.erase(value.match, value.seq); + } + else assert(mode == "empty" && queue.load().empty()); + return 0; + } + + fs::create_directories(root); + assert(queue.load().empty()); + std::set keys; + for (int i = 0; i < 1000; ++i) + { + const auto key = new_match_key(); + assert(!key.empty() && key.size() <= 64 && keys.insert(key).second); + } + const auto match = new_match_key(); + const auto start = body(match, 1, "game_start"); + const auto end = body(match, 30, "game_end"); + queue.store(start); + queue.store(end); + queue.store(start); // A retry cannot add a duplicate or reorder the queue. + const auto replay = outbox(root, server, endpoint).load(); + assert(replay.size() == 2 && replay[0].body == start && replay[1].body == end); + assert(replay[0].seq == 1 && replay[1].seq == 30 && replay[1].round == 12); + assert(outbox(root, "other-server", endpoint).load().empty()); + assert(outbox(root, server, "https://different.test/api").load().empty()); + auto changed = nlohmann::json::parse(start); + changed["map"] = "another-map"; + must_fail([&] { queue.store(changed.dump()); }); + must_fail([&] { queue.store(body(match, 0, "game_start")); }); + must_fail([&] { queue.store(body(match, 2, "game_start")); }); + must_fail([&] { queue.store(body(match, 3, "heartbeat")); }); + must_fail([&] { queue.store("error"); }); + changed = nlohmann::json::parse(end); + changed["server"] = "wrong-server"; + must_fail([&] { queue.store(changed.dump()); }); + + const auto file = fs::directory_iterator(root)->path(); + auto temporary = file; + temporary += ".tmp"; + { std::ofstream stream(temporary); stream << "interrupted write"; } + assert(queue.load().size() == 2); // Incomplete temp file never replaces committed data. + fs::remove(temporary); + fs::create_directory(temporary); // Force replacement writes to fail even under root. + must_fail([&] { queue.erase(match, 1); }); + must_fail([&] { queue.store(body(new_match_key(), 1, "game_start")); }); + assert(queue.load().size() == 2); + fs::remove(temporary); + queue.erase(match, 1); + queue.erase(match, 1); // Cleanup is idempotent after an uncertain completion. + assert(queue.load().size() == 1 && queue.load()[0].body == end); + queue.erase(match, 30); + assert(queue.load().empty()); + + { std::ofstream stream(file); stream << "corrupt"; } + must_fail([&] { queue.load(); }); + must_fail([&] { queue.store(start); }); + assert(fs::file_size(file) == 7); // Never silently overwrite corrupt retained data. + fs::remove(file); + + for (const std::string response : {"", "bad gateway", "{", "null", "[]", "true", + "{}", "{\"ok\":false}", "{\"ok\":1}", "{\"ok\":\"true\"}", "{\"ok\":true}garbage"}) + assert(!parse_acknowledgement(response).accepted); + const auto ack = parse_acknowledgement(R"({"ok":true,"telemetry":{"timing":{"matchId":921}}})"); + assert(ack.accepted && ack.match_id == "921"); + assert(parse_acknowledgement(R"({"ok":true})").accepted); + const auto large_id = parse_acknowledgement(R"({"ok":true,"telemetry":{"timing":{"matchId":9223372036854775807}}})"); + assert(large_id.match_id == "9223372036854775807"); + assert(parse_acknowledgement(R"({"ok":true,"telemetry":null})").match_id.empty()); + assert(parse_acknowledgement(R"({"ok":true,"telemetry":{"timing":{"matchId":"wrong"}}})").match_id.empty()); + assert(!parse_acknowledgement("{\"ok\":true}" + std::string(0x5000, ' ')).accepted); + std::cout << "PASS native telemetry fixtures\n"; +}