Merge pull request #57 from cacheMon/claude/keen-volta-u9ls6v

Clarify that network traces are compressed (.csv.zst) + add regression tests

Author: 1a1a11a

docs/traces/NETWORK_EVENTS.md

@@ -28,7 +28,18 @@ from the kernel `bpf_ktime_get_ns()`) — the common cross-stream correlation
 clock. Add the manifest's `clock.mono_to_real_offset_ns` to recover wall-clock
 nanoseconds.
 
-## Connection Lifecycle — `nw_conn/nw_conn_*.csv`
+**Output Files:** Like every other trace stream, network streams are
+Zstandard-compressed on disk and on upload:
+
+- `linux_trace_v4_test/{MACHINE_ID}/{TIMESTAMP}/nw_conn/nw_conn_*.csv.zst`
+- `linux_trace_v4_test/{MACHINE_ID}/{TIMESTAMP}/nw_epoll/nw_epoll_*.csv.zst`
+- `linux_trace_v4_test/{MACHINE_ID}/{TIMESTAMP}/nw_sockopt/nw_sockopt_*.csv.zst`
+- `linux_trace_v4_test/{MACHINE_ID}/{TIMESTAMP}/nw_drop/nw_drop_*.csv.zst`
+
+(They are left uncompressed only when the optional `zstandard` library is
+unavailable — the same fallback that applies to all streams.)
+
+## Connection Lifecycle — `nw_conn/nw_conn_*.csv.zst`
 
 | # | Field | Type | Description |
 |---|-------|------|-------------|
@@ -51,7 +62,7 @@ nanoseconds.
 | 17 | return_value | `s32` | Syscall return value |
 | 18 | mono_ns | `u64` | Cross-stream correlation clock |
 
-## Multiplexing (epoll/poll/select) — `nw_epoll/nw_epoll_*.csv`
+## Multiplexing (epoll/poll/select) — `nw_epoll/nw_epoll_*.csv.zst`
 
 | # | Field | Type | Description |
 |---|-------|------|-------------|
@@ -70,7 +81,7 @@ nanoseconds.
 | 13 | latency_ns | `u64` | Wait entry→exit latency; empty otherwise |
 | 14 | mono_ns | `u64` | Cross-stream correlation clock |
 
-## Socket Options — `nw_sockopt/nw_sockopt_*.csv`
+## Socket Options — `nw_sockopt/nw_sockopt_*.csv.zst`
 
 | # | Field | Type | Description |
 |---|-------|------|-------------|
@@ -85,7 +96,7 @@ nanoseconds.
 | 9 | return_value | `s32` | Syscall return value |
 | 10 | mono_ns | `u64` | Cross-stream correlation clock |
 
-## Drops & Retransmits — `nw_drop/nw_drop_*.csv`
+## Drops & Retransmits — `nw_drop/nw_drop_*.csv.zst`
 
 | # | Field | Type | Description |
 |---|-------|------|-------------|

tests/test_writer_upload.py

@@ -345,5 +345,68 @@ def test_only_first_part_has_header(self):
         self.assertEqual(second[0], "rowB,2")
 
 
+class NetworkStreamCompressionTests(unittest.TestCase):
+    """Network streams (nw_conn/nw_epoll/nw_sockopt/nw_drop) must be compressed
+    and uploaded exactly like every other trace stream — both at shutdown
+    (force_flush) and on mid-trace rotation."""
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.output_dir = os.path.join(self.tmp, "trace")
+        self.upload = FakeUploadManager()
+        self.wm = SilentWriteManager(
+            output_dir=self.output_dir,
+            upload_manager=self.upload,
+            automatic_upload=True,
+        )
+
+    def tearDown(self):
+        import shutil
+        try:
+            self.wm.close_handles()
+        except Exception:
+            pass
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_all_network_streams_registered_for_rotation(self):
+        # Every network stream must be in the generic rotation registry so a
+        # slow stream is compressed+uploaded mid-trace, not just at shutdown.
+        for key in ("nw_conn", "nw_epoll", "nw_sockopt", "nw_drop"):
+            self.assertIn(key, self.wm._streams, key)
+
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
+    def test_force_flush_compresses_network_streams(self):
+        self.wm.append_conn_log("ts,CONNECT,1,1,proc,AF_INET")
+        self.wm.append_epoll_log("ts,EPOLL_WAIT,1,1,proc")
+        self.wm.append_sockopt_log("ts,SET,1,proc,3")
+        self.wm.append_drop_log("ts,PACKET_DROP,1,proc,TCP")
+
+        self.wm.force_flush()
+
+        # Each stream produced exactly one compressed upload under its own subdir.
+        self.assertEqual(len(self.upload.uploaded), 4)
+        subdirs = {os.path.basename(os.path.dirname(p)) for p in self.upload.uploaded}
+        for sub in ("nw_conn", "nw_epoll", "nw_sockopt", "nw_drop"):
+            self.assertIn(sub, subdirs, sub)
+        for p in self.upload.uploaded:
+            self.assertTrue(p.endswith(".csv.zst"), p)
+            self.assertTrue(os.path.exists(p))
+
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
+    def test_network_threshold_rotation_compresses(self):
+        # Dropping the threshold forces a mid-trace rotation, which must
+        # compress+upload the rotated file just like the continuous streams.
+        self.wm.nw_conn_max_events = 3
+        for i in range(7):
+            self.wm.append_conn_log(f"row{i}")
+
+        # 7 events at a threshold of 3 → 2 full rotations (2 uploads), with 1
+        # event left buffered (force_flush is not called here).
+        self.assertEqual(len(self.upload.uploaded), 2)
+        for p in self.upload.uploaded:
+            self.assertTrue(p.endswith(".csv.zst"), p)
+            self.assertEqual(os.path.basename(os.path.dirname(p)), "nw_conn")
+
+
 if __name__ == "__main__":
     unittest.main()