From 284b603e6e731a9ed46233605c1e369ef658a810 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 16:59:04 +0200 Subject: [PATCH] Account initial GAP independently of first delivered stream record --- tests/stream_netns/node.c | 9 ++++++--- tests/stream_netns/scenario.py | 6 ++++-- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/stream_netns/node.c b/tests/stream_netns/node.c index 28d09546..51e95fd2 100644 --- a/tests/stream_netns/node.c +++ b/tests/stream_netns/node.c @@ -29,7 +29,7 @@ struct lab_stream { struct STREAM_HANDLE *handle; uint64_t source, generated, received, bytes, gaps, gap_batches, errors, ends, closed, next, first, latency_max; - uint64_t latency_sum, histogram[LAB_HISTOGRAM], last_data_ns; + uint64_t latency_sum, histogram[LAB_HISTOGRAM], last_data_ns, start_seq; uint64_t seed, interval_ns, due_ns; uint8_t *payload; size_t size; @@ -95,6 +95,7 @@ static void event(struct UTUN_INSTANCE *inst, const struct stream_event *event, if (event->type == STREAM_EVENT_CLOSED) { stream->closed++; return; } if (event->type == STREAM_EVENT_END) { stream->ends++; return; } if (event->type == STREAM_EVENT_GAP) { + if (!stream->started) stream->start_seq = event->first_seq; stream->gaps++; stream->gap_batches += event->last_seq - event->first_seq + 1; if (stream->started && event->first_seq != stream->next) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: GAP cursor expected=%" PRIu64 " first=%" PRIu64, stream->next, event->first_seq); @@ -118,6 +119,7 @@ static void event(struct UTUN_INSTANCE *inst, const struct stream_event *event, " expected=%" PRIu64 " bytes=%zu checksum=%d", stream->source, event->first_seq, stream->next, event->length, valid); stream->errors++; } + if (!stream->started) stream->start_seq = event->first_seq; if (!stream->received) stream->first = event->first_seq; stream->started = 1; stream->next = event->last_seq + 1; stream->received++; stream->bytes += event->length; stream->last_data_ns = now_ns(); @@ -172,11 +174,12 @@ static void statistics(struct lab *lab, uint64_t id, int loads) { printf("%s{\"source\":\"%" PRIu64 "\",\"publisher\":%d,\"generated\":%" PRIu64 ",\"received\":%" PRIu64 ",\"bytes\":%" PRIu64 ",\"gaps\":%" PRIu64 ",\"gap_batches\":%" PRIu64 ",\"errors\":%" PRIu64 ",\"ends\":%" PRIu64 ",\"closed\":%" PRIu64 ",\"first_seq\":%" PRIu64 ",\"next_seq\":%" PRIu64 - ",\"age_ms\":%" PRIu64 ",\"latency_p50_ms\":%" PRIu64 ",\"latency_p95_ms\":%" PRIu64 + ",\"start_seq\":%" PRIu64 ",\"age_ms\":%" PRIu64 ",\"latency_p50_ms\":%" PRIu64 ",\"latency_p95_ms\":%" PRIu64 ",\"latency_p99_ms\":%" PRIu64 ",\"latency_max_ms\":%" PRIu64, separator++ ? "," : "", stream->source, stream->publisher, stream->generated, stream->received, stream->bytes, stream->gaps, stream->gap_batches, stream->errors, stream->ends, stream->closed, - stream->first, stream->next, stream->last_data_ns ? (now_ns() - stream->last_data_ns) / UINT64_C(1000000) : 0, + stream->first, stream->next, stream->start_seq, + stream->last_data_ns ? (now_ns() - stream->last_data_ns) / UINT64_C(1000000) : 0, percentile(stream, 50), percentile(stream, 95), percentile(stream, 99), stream->latency_max); struct stream_info info; if (stream->handle && stream_get_info(stream->handle, &info) == STREAM_OK) { diff --git a/tests/stream_netns/scenario.py b/tests/stream_netns/scenario.py index 6c1b7229..66e6a97f 100644 --- a/tests/stream_netns/scenario.py +++ b/tests/stream_netns/scenario.py @@ -183,8 +183,9 @@ class Scenario: stream = self.stream(final[name], source_id) if stream["next_seq"] != generated: raise RuntimeError(f"{name}: END delivered before final cursor: {stream['next_seq']} != {generated}") - if stream["received"] + stream["gap_batches"] + stream["first_seq"] != generated: - raise RuntimeError(f"{name}: unaccounted missing records") + if stream["received"] + stream["gap_batches"] + stream["start_seq"] != generated: + raise RuntimeError(f"{name}: unaccounted records: received={stream['received']}, gaps={stream['gap_batches']}, " + f"start={stream['start_seq']}, generated={generated}") self.report(final, source_id) return {"pass": True, "group": group, "publisher": self.args.publisher, "generated": generated, "phases": self.phases, "subscribers_checked": len(names)} @@ -204,6 +205,7 @@ class Scenario: (self.network.directory / "final-stats.json").write_text(json.dumps(final, indent=2) + "\n") def close(self): + (self.network.directory / "last-stats.json").write_text(json.dumps(self.latest, indent=2) + "\n") failures = {} for name, agent in self.agents.items(): status = agent.close()