fix(cpworker): keep zmq heartbeats out of fwd counters; stamp stats with wall-clock time

cpctl/cmd/stats.go

@@ -26,7 +26,7 @@ func init() {
 	f.IntVarP(&statsCfg.count, "count", "n", 0,
 		"number of samples (0 = run forever)")
 	f.DurationVarP(&statsCfg.interval, "interval", "i", 2*time.Second,
-		"interval between samples")
+		"poll interval; cpworker refreshes stats every 5s, so samples are at least 5s apart")
 }
 
 var statsCmd = &cobra.Command{
@@ -109,7 +109,7 @@ func emitStatsRaw(out io.Writer, s cpworker.StatsSummary, format string) error {
 	switch format {
 	case FormatJSONL:
 		rec := statsRecord{
-			Ts:       time.Unix(s.Time.Sec, s.Time.Nsec).UTC().Format(time.RFC3339Nano),
+			Ts:       wallTime(s),
 			Kind:     "raw",
 			Counters: countersMap(s),
 		}
@@ -128,7 +128,7 @@ func emitStatsDiff(out io.Writer, s, last cpworker.StatsSummary, format string)
 	switch format {
 	case FormatJSONL:
 		rec := statsRecord{
-			Ts:        t1.UTC().Format(time.RFC3339Nano),
+			Ts:        wallTime(s),
 			Kind:      "rate",
 			IntervalS: &secs,
 			Counters:  countersMap(s),
@@ -142,6 +142,11 @@ func emitStatsDiff(out io.Writer, s, last cpworker.StatsSummary, format string)
 	}
 }
 
+// wallTime formats when cpworker took the snapshot.
+func wallTime(s cpworker.StatsSummary) string {
+	return time.Unix(s.WallTime.Sec, s.WallTime.Nsec).UTC().Format(time.RFC3339Nano)
+}
+
 func writeJSONL(out io.Writer, v any) error {
 	data, err := json.Marshal(v)
 	if err != nil {
@@ -171,6 +176,7 @@ func countersMap(s cpworker.StatsSummary) map[string]any {
 		"error_drop_packets":     packetsAsJSON(s.Output.ErrorDropPackets),
 		"ratelimit_drop_bytes":   bytesAsJSON(s.Output.RatelimitDropBytes),
 		"ratelimit_drop_packets": packetsAsJSON(s.Output.RatelimitDropPackets),
+		"heartbeat_packets":      packetsAsJSON(s.Output.HeartbeatPackets),
 	}
 }
 
@@ -205,6 +211,7 @@ func ratesMap(s, last cpworker.StatsSummary, secs float64) map[string]any {
 	addPacketsRate("error_drop_packets_per_sec", s.Output.ErrorDropPackets, last.Output.ErrorDropPackets)
 	addBytesRate("ratelimit_drop_bytes_per_sec", s.Output.RatelimitDropBytes, last.Output.RatelimitDropBytes)
 	addPacketsRate("ratelimit_drop_packets_per_sec", s.Output.RatelimitDropPackets, last.Output.RatelimitDropPackets)
+	addPacketsRate("heartbeat_packets_per_sec", s.Output.HeartbeatPackets, last.Output.HeartbeatPackets)
 	return rates
 }
 
@@ -233,6 +240,7 @@ func printRawText(out io.Writer, s cpworker.StatsSummary) {
 		{"Error Drop Packets", formatPacketsStats(s.Output.ErrorDropPackets)},
 		{"Ratelimit Drop Bytes", formatBytesStats(s.Output.RatelimitDropBytes)},
 		{"Ratelimit Drop Packets", formatPacketsStats(s.Output.RatelimitDropPackets)},
+		{"Heartbeat Packets", formatPacketsStats(s.Output.HeartbeatPackets)},
 	}
 	maxLen := 0
 	for _, r := range rows {
@@ -246,8 +254,8 @@ func printRawText(out io.Writer, s cpworker.StatsSummary) {
 }
 
 func printSummaryStats(out io.Writer, stats cpworker.StatsSummary, lastStats cpworker.StatsSummary) {
-	headers := make([]string, 0, 12)
-	row := make([]string, 0, 12)
+	headers := make([]string, 0, 13)
+	row := make([]string, 0, 13)
 
 	t1 := time.Unix(stats.Time.Sec, stats.Time.Nsec)
 	t2 := time.Unix(lastStats.Time.Sec, lastStats.Time.Nsec)
@@ -284,6 +292,7 @@ func printSummaryStats(out io.Writer, stats cpworker.StatsSummary, lastStats cpw
 	addPackets("Error Drop Packets", stats.Output.ErrorDropPackets, lastStats.Output.ErrorDropPackets)
 	addBytes("Ratelimit Drop Bytes", stats.Output.RatelimitDropBytes, lastStats.Output.RatelimitDropBytes)
 	addPackets("Ratelimit Drop Packets", stats.Output.RatelimitDropPackets, lastStats.Output.RatelimitDropPackets)
+	addPackets("Heartbeat Packets", stats.Output.HeartbeatPackets, lastStats.Output.HeartbeatPackets)
 
 	maxHeaderLen := 0
 	for _, h := range headers {

cpctl/cmd/stats_test.go

@@ -164,3 +164,82 @@ func TestPacketsStatsPerSec_ZeroSecs(t *testing.T) {
 		t.Errorf("zero seconds should return input unchanged, got %+v", got)
 	}
 }
+
+// #297: ts is the snapshot's wall-clock time, not its CLOCK_MONOTONIC time.
+func TestRunStats_TsIsSnapshotWallTime(t *testing.T) {
+	withWall := func(s cpworker.StatsSummary, sec int64) cpworker.StatsSummary {
+		s.WallTime.Sec = sec
+		s.WallTime.Nsec = 250000000
+		return s
+	}
+	cases := []struct {
+		name      string
+		count     int
+		snapshots []cpworker.StatsSummary
+		wantTs    string
+	}{
+		{"raw", 1, []cpworker.StatsSummary{withWall(makeStats(632655, 0, 0, 0, 0), 1759900000)}, "2025-10-08T05:06:40.25Z"},
+		{"rate", 2, []cpworker.StatsSummary{
+			withWall(makeStats(632655, 0, 0, 0, 0), 1759900000),
+			withWall(makeStats(632660, 0, 0, 0, 0), 1759900005),
+		}, "2025-10-08T05:06:45.25Z"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			client := &fakeStatsClient{snapshots: tc.snapshots}
+			var buf bytes.Buffer
+			if err := runStats(context.Background(), client, &buf, tc.count, time.Microsecond, FormatJSONL); err != nil {
+				t.Fatal(err)
+			}
+			var rec statsRecord
+			if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil {
+				t.Fatalf("decode: %v\n%s", err, buf.String())
+			}
+			if rec.Ts != tc.wantTs {
+				t.Errorf("ts = %q, want %q", rec.Ts, tc.wantTs)
+			}
+		})
+	}
+}
+
+// #299: heartbeats are reported in their own counter.
+func TestRunStats_ReportsHeartbeatPackets(t *testing.T) {
+	s0 := makeStats(1000, 0, 0, 0, 0)
+	s1 := makeStats(1005, 0, 0, 0, 0)
+	s1.Output.HeartbeatPackets = cpworker.PacketsStats{Packets: 5}
+
+	t.Run("jsonl", func(t *testing.T) {
+		client := &fakeStatsClient{snapshots: []cpworker.StatsSummary{s0, s1}}
+		var buf bytes.Buffer
+		if err := runStats(context.Background(), client, &buf, 2, time.Microsecond, FormatJSONL); err != nil {
+			t.Fatal(err)
+		}
+		var rec statsRecord
+		if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil {
+			t.Fatalf("decode: %v", err)
+		}
+		hb, ok := rec.Counters["heartbeat_packets"].(map[string]any)
+		if !ok || hb["packets"] != float64(5) {
+			t.Errorf("counters.heartbeat_packets = %v, want packets 5", rec.Counters["heartbeat_packets"])
+		}
+		rate, ok := rec.Rates["heartbeat_packets_per_sec"].(map[string]any)
+		if !ok || rate["packets"] != float64(1) {
+			t.Errorf("rates.heartbeat_packets_per_sec = %v, want packets 1", rec.Rates["heartbeat_packets_per_sec"])
+		}
+	})
+	t.Run("text", func(t *testing.T) {
+		for _, count := range []int{1, 2} {
+			client := &fakeStatsClient{snapshots: []cpworker.StatsSummary{s1, s1}}
+			if count == 2 {
+				client.snapshots = []cpworker.StatsSummary{s0, s1}
+			}
+			var buf bytes.Buffer
+			if err := runStats(context.Background(), client, &buf, count, time.Microsecond, FormatText); err != nil {
+				t.Fatal(err)
+			}
+			if !strings.Contains(buf.String(), "Heartbeat Packets") {
+				t.Errorf("-n %d text output has no Heartbeat Packets row:\n%s", count, buf.String())
+			}
+		}
+	})
+}

cpgolib/cpworker/stats.go

@@ -10,10 +10,17 @@ const (
 )
 
 type StatsSummary struct {
+	// Time is the snapshot's CLOCK_MONOTONIC time. Use it for the interval
+	// between two snapshots; it is not a date.
 	Time struct {
 		Sec  int64 `mapstructure:"sec"`
 		Nsec int64 `mapstructure:"nsec"`
 	} `mapstructure:"time"`
+	// WallTime is the snapshot's CLOCK_REALTIME time: when it was taken.
+	WallTime struct {
+		Sec  int64 `mapstructure:"sec"`
+		Nsec int64 `mapstructure:"nsec"`
+	} `mapstructure:"wall_time"`
 	Capture        CaptureStats        `mapstructure:"capture"`
 	Output         OutputStats         `mapstructure:"output"`
 	PipelineBuffer PipelineBufferStats `mapstructure:"pipeline_buffer"`

cpgolib/cpworker/unix_client_test.go

@@ -209,6 +209,35 @@ func TestUnixClient_Info(t *testing.T) {
 	}
 }
 
+func TestUnixClient_CollectStatsSummaryTimes(t *testing.T) {
+	srv := newFakeServer(t)
+	defer srv.Close()
+
+	srv.handlers["collect_stats_summary"] = func(args map[string]any) (map[string]any, error) {
+		return map[string]any{
+			"time":      map[string]any{"sec": 632655, "nsec": 154030865},
+			"wall_time": map[string]any{"sec": 1759900000, "nsec": 250000000},
+		}, nil
+	}
+
+	client, err := NewClient(srv.connStr())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	stats, err := client.CollectStatsSummary(context.Background())
+	if err != nil {
+		t.Fatalf("collect_stats_summary: %v", err)
+	}
+	if stats.Time.Sec != 632655 || stats.Time.Nsec != 154030865 {
+		t.Errorf("time: got %+v", stats.Time)
+	}
+	if stats.WallTime.Sec != 1759900000 || stats.WallTime.Nsec != 250000000 {
+		t.Errorf("wall_time: got %+v", stats.WallTime)
+	}
+}
+
 func TestUnixClient_UnknownCommand(t *testing.T) {
 	srv := newFakeServer(t)
 	defer srv.Close()

cpworker/src/output_zmq.c

@@ -113,27 +113,35 @@ int zmq_flush_packet(zmq_output_t *output)
         output->error_info.first_pktsec = pkts_buf->first_pktsec;
     }
 
+    // Only captured packets are counted as forwarded or dropped. A batch that carries
+    // nothing but heartbeats contributes no bytes, not even its batch header.
+    uint16_t data_num = send_num - pkts_buf->heartbeat_num;
+    uint32_t data_bytes = data_num > 0 ? pkts_buf->batch_bufpos - pkts_buf->heartbeat_bytes : 0;
+
     int rc = zmq_send(output->pusher, &(pkts_buf->buf[0]), pkts_buf->batch_bufpos, ZMQ_DONTWAIT);
     if (rc != -1)
     {
-        bytes_stats_add(&output->base.stats->fwd_bytes, pkts_buf->batch_bufpos);
-        packets_stats_add(&output->base.stats->fwd_packets, send_num);
+        bytes_stats_add(&output->base.stats->fwd_bytes, data_bytes);
+        packets_stats_add(&output->base.stats->fwd_packets, data_num);
+        packets_stats_add(&output->base.stats->heartbeat_packets, pkts_buf->heartbeat_num);
     }
     else
     {
         if (output->error_info.nb_drop_batches == 0)
             snprintf(output->error_info.send_error, ERROR_BUFFER_SIZE, "zmq_send failed: %s", zmq_strerror(errno));
 
         output->error_info.nb_drop_batches++;
-        output->error_info.nb_drop_packets += send_num;
+        output->error_info.nb_drop_packets += data_num;
 
-        bytes_stats_add(&output->base.stats->error_drop_bytes, pkts_buf->batch_bufpos);
-        packets_stats_add(&output->base.stats->error_drop_packets, send_num);
+        bytes_stats_add(&output->base.stats->error_drop_bytes, data_bytes);
+        packets_stats_add(&output->base.stats->error_drop_packets, data_num);
     }
 
     pkts_buf->first_pktsec = 0;
     pkts_buf->batch_bufpos = sizeof(zmq_pkt_batch_hdr_t);
     pkts_buf->batch_hdr.pkts_num = 0;
+    pkts_buf->heartbeat_num = 0;
+    pkts_buf->heartbeat_bytes = 0;
     return 0;
 }
 
@@ -336,14 +344,15 @@ static void zmq_send_heartbeat_packet(zmq_output_t *output, struct timeval *tv)
     memcpy(&pkts_buf->buf[pos + offsetof(struct ether_header, ether_type)], &hb_etype, sizeof(hb_etype));
     pos += sizeof(eth);
 
+    pkts_buf->heartbeat_bytes += pos - pkts_buf->batch_bufpos;
+    pkts_buf->heartbeat_num++;
     pkts_buf->batch_bufpos = pos;
     pkts_buf->batch_hdr.pkts_num++;
 
-    // Flush immediately - heartbeat triggers batch flush
+    // Flush immediately - heartbeat triggers batch flush. The flush counts the heartbeat
+    // in heartbeat_packets once it is sent.
     zmq_flush_packet(output);
 
-    // Update heartbeat stats and last packet time
-    packets_stats_add(&output->base.stats->heartbeat_packets, 1);
     output->last_pkt_tv = *tv;
 }
 

cpworker/src/output_zmq.h

@@ -43,6 +43,10 @@ typedef struct
     char buf[ZMQ_MAX_BATCH_BUF_SIZE];
     uint32_t batch_bufpos;
     long int first_pktsec;
+    // Heartbeat records in the batch. They are not captured packets, so a flush keeps
+    // them out of the fwd/error_drop counters and counts them in heartbeat_packets.
+    uint16_t heartbeat_num;
+    uint32_t heartbeat_bytes;
 } zmq_pkts_buf_t;
 
 typedef struct ZmqOptions

cpworker/src/task.c

@@ -157,7 +157,8 @@ static task_error_t *task_error_new(int index, const char *fingerprint, const ch
 
 typedef struct TaskStatsSummarySnapshot
 {
-    struct timespec tm;
+    struct timespec tm;      // CLOCK_MONOTONIC: interval between two snapshots
+    struct timespec wall_tm; // CLOCK_REALTIME: when the snapshot was taken
     capture_stats_t capture;
     output_stats_t output;
     pipeline_buffer_stats_t pipeline_buffer;
@@ -644,6 +645,7 @@ static void task_manager_update_stats_summary()
     memset(&stats, 0, sizeof(task_stats_summary_snapshot_t));
 
     clock_gettime(CLOCK_MONOTONIC, &stats.tm);
+    clock_gettime(CLOCK_REALTIME, &stats.wall_tm);
     stats.capture = this->stats_summary.capture;
     stats.output = this->stats_summary.output;
     if (this->execution_model == EXECUTION_MODEL_PIPELINE)
@@ -731,6 +733,11 @@ int task_manager_collect_stats_summary_command(cJSON *cmd_msg, cJSON *server_msg
         !cJSON_AddNumberToObject(time, "nsec", stats.tm.tv_nsec))
         return -1;
 
+    cJSON *wall_time = cJSON_AddObjectToObject(server_msg, "wall_time");
+    if (!wall_time || !cJSON_AddNumberToObject(wall_time, "sec", stats.wall_tm.tv_sec) ||
+        !cJSON_AddNumberToObject(wall_time, "nsec", stats.wall_tm.tv_nsec))
+        return -1;
+
     if (add_capture_stats(server_msg, "capture", &stats.capture) != 0 ||
         add_output_stats(server_msg, "output", &stats.output) != 0 ||
         add_pipeline_buffer_stats(server_msg, "pipeline_buffer", &stats.pipeline_buffer) != 0)

cpworker/tests/unit/misc.c

@@ -315,13 +315,13 @@ void test_zmq_heartbeat_packet_generation(void)
     // Call heartbeat - should generate a heartbeat packet
     output_heartbeat((output_base_t *)output, now_tv.tv_sec);
 
-    // Verify heartbeat stats were incremented
+    // The heartbeat is queued on the not-yet-connected pipe, so the send succeeds
     TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets);
 
-    // The ZMQ send may fail (no receiver) but the packet was still generated
-    // Either forwarded or error-dropped (depends on ZMQ connection state)
-    uint64_t total = stats.fwd_packets.packets + stats.error_drop_packets.packets;
-    TEST_ASSERT_EQUAL_UINT64(1, total);
+    // A heartbeat is not a captured packet: it is neither forwarded nor dropped
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_bytes.bytes);
+    TEST_ASSERT_EQUAL_UINT64(0, stats.error_drop_packets.packets);
 
     zmq_output_destroy((output_base_t *)output);
 }