Preserve and retry telemetry match boundaries across restarts

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
 

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)