Account initial GAP independently of first delivered stream record

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) {

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()