Merge pull request #36 from cacheMon/claude/fervent-noether-nrf2fq

Fix correctness bugs in BPF prober and trace pipeline

Author: 1a1a11a

.github/scripts/bpf_smoke.py

@@ -59,7 +59,31 @@ def main():
         ("kretprobe", "vfs_read", "trace_vfs_read_ret"),
         ("kprobe", "vfs_write", "trace_vfs_write"),
         ("kretprobe", "vfs_write", "trace_vfs_write_ret"),
+        # fsync de-dup pair: the kretprobe clears the nested-call marker.
+        ("kprobe", "vfs_fsync", "trace_vfs_fsync"),
+        ("kretprobe", "vfs_fsync", "trace_vfs_fsync_ret"),
+        ("kprobe", "vfs_fsync_range", "trace_vfs_fsync_range"),
     ]
+
+    # Symbol-conditional probes. These validate that the pt_regs-unwrapping
+    # *_x64 variants and the DIO direction entry probes were compiled in and
+    # attach — a guard mismatch would otherwise pass CI (BPF() load succeeds
+    # without them) and abort the tracer at startup instead.
+    conditional_probes = [
+        (b"__x64_sys_mremap", [("kprobe", "__x64_sys_mremap", "trace_mremap_entry_x64"),
+                               ("kretprobe", "__x64_sys_mremap", "trace_mremap_ret")]),
+        (b"__x64_sys_openat", [("kprobe", "__x64_sys_openat", "trace_openat_entry_x64")]),
+        (b"__x64_sys_io_uring_enter", [("kprobe", "__x64_sys_io_uring_enter", "trace_io_uring_enter_x64")]),
+        (b"iomap_dio_rw", [("kprobe", "iomap_dio_rw", "trace_dio_entry_iomap"),
+                           ("kretprobe", "iomap_dio_rw", "trace_dio_return")]),
+        (b"__blockdev_direct_IO", [("kprobe", "__blockdev_direct_IO", "trace_dio_entry_blockdev")]),
+    ]
+    for symbol, symbol_probes in conditional_probes:
+        if BPF.get_kprobe_functions(symbol):
+            probes.extend(symbol_probes)
+        else:
+            print(f"SKIP: {symbol.decode()} not present on this kernel")
+
     for kind, event, fn in probes:
         if kind == "kprobe":
             b.attach_kprobe(event=event, fn_name=fn)

src/tracer/FlagMapper.py

@@ -697,7 +697,7 @@ def format_vfs_flags(self, op_name, flags):
             str: Human-readable flags, or an empty string when the operation
                  does not define flag semantics for the current value.
         """
-        if op_name in {"OPEN", "READ", "WRITE", "CLOSE", "FSYNC", "READDIR"}:
+        if op_name in {"OPEN", "READ", "WRITE", "CLOSE", "FSYNC", "FDATASYNC", "READDIR"}:
             return self.format_fs_flags(flags)
         if op_name == "MKDIR":
             return self.format_mode_flags(flags)

src/tracer/IOTracer.py

@@ -1110,10 +1110,10 @@ def trace(self):
                 while remaining > 0 and self.running:
                     sleep_time = min(0.1, remaining)
                     time.sleep(sleep_time)
-                    
+
                     current = time.time()
                     remaining = end_time - current # type: ignore
-                    
+
                     if self.verbose and int(current) % 10 == 0 and int(current) > int(current - sleep_time):
                         elapsed = current - start
                         logger("info", f"Progress: {elapsed:.1f}s/{duration_target}s") # type: ignore
@@ -1123,7 +1123,7 @@ def trace(self):
                 # Run indefinitely until Ctrl+C
                 while self.running:
                     time.sleep(0.1)
-                    
+
                     if self.verbose:
                         current = time.time()
                         if int(current) % 30 == 0:  # Every 30 seconds

src/tracer/KernelProbeTracker.py

@@ -174,10 +174,13 @@ def attach_probes(self):
             # Capture the user-provided filename before the kernel resolves it.
             # Must be registered BEFORE vfs_open so the path is staged in time.
             # Uses the same fallback chain as the kretprobe.
+            # NOTE: __x64_sys_* wrappers take a single pt_regs* holding the
+            # user registers, so they need the *_x64 probe variants that
+            # unwrap it; reading PARM1-4 directly there yields garbage.
             if BPF.get_kprobe_functions(b'do_sys_openat2'):
                 self.add_kprobe("do_sys_openat2", "trace_do_sys_openat2_entry")
             elif BPF.get_kprobe_functions(b'__x64_sys_openat'):
-                self.add_kprobe("__x64_sys_openat", "trace_do_sys_openat2_entry")
+                self.add_kprobe("__x64_sys_openat", "trace_openat_entry_x64")
             else:
                 self.add_kprobe("sys_openat", "trace_do_sys_openat2_entry")
             self.add_kprobe("vfs_open", "trace_vfs_open")
@@ -191,6 +194,9 @@ def attach_probes(self):
             else:
                 self.add_kretprobe("sys_openat", "trace_sys_openat_ret")
             self.add_kprobe("vfs_fsync", "trace_vfs_fsync")
+            # Return probe clears the marker that suppresses the duplicate
+            # event from the nested vfs_fsync -> vfs_fsync_range call.
+            self.add_kretprobe("vfs_fsync", "trace_vfs_fsync_ret")
             self.add_kprobe("ksys_sync", "trace_ksys_sync")
             self.add_kprobe("vfs_fsync_range", "trace_vfs_fsync_range")
             self.add_kprobe("__fput", "trace_fput")
@@ -200,9 +206,10 @@ def attach_probes(self):
             self.add_kretprobe("do_mmap", "trace_mmap_ret")
             self.add_kprobe("__vm_munmap", "trace_munmap")
 
-            # mremap probes — kernel may export the arch wrapper or the generic symbol
+            # mremap probes — kernel may export the arch wrapper or the generic
+            # symbol. The wrapper needs the pt_regs-unwrapping variant.
             if BPF.get_kprobe_functions(b'__x64_sys_mremap'):
-                self.add_kprobe("__x64_sys_mremap", "trace_mremap_entry")
+                self.add_kprobe("__x64_sys_mremap", "trace_mremap_entry_x64")
                 self.add_kretprobe("__x64_sys_mremap", "trace_mremap_ret")
             elif BPF.get_kprobe_functions(b'sys_mremap'):
                 self.add_kprobe("sys_mremap", "trace_mremap_entry")
@@ -251,12 +258,16 @@ def attach_probes(self):
             # else:
             #     logger("warning", "filemap_fault not available - mmap I/O tracking disabled")
 
-            # Direct I/O probe for bypass detection (return probe only - no latency tracking)
+            # Direct I/O probes. The entry probe stages the I/O direction from
+            # the iov_iter (the return value alone cannot distinguish a read
+            # from a write); the return probe emits the completion event.
             if BPF.get_kprobe_functions(b'iomap_dio_rw'):
+                self.add_kprobe("iomap_dio_rw", "trace_dio_entry_iomap")
                 self.add_kretprobe("iomap_dio_rw", "trace_dio_return")
                 if self.developer_mode:
                     logger("info", "Direct I/O tracing enabled via iomap_dio_rw")
             elif BPF.get_kprobe_functions(b'__blockdev_direct_IO'):
+                self.add_kprobe("__blockdev_direct_IO", "trace_dio_entry_blockdev")
                 self.add_kretprobe("__blockdev_direct_IO", "trace_dio_return")
                 if self.developer_mode:
                     logger("info", "Direct I/O tracing enabled via __blockdev_direct_IO")
@@ -376,7 +387,8 @@ def attach_probes(self):
                 if self.developer_mode:
                     logger("info", "io_uring tracing enabled via __io_uring_enter")
             elif BPF.get_kprobe_functions(b'__x64_sys_io_uring_enter'):
-                self.add_kprobe("__x64_sys_io_uring_enter", "trace_io_uring_enter")
+                # Syscall wrapper: needs the pt_regs-unwrapping variant.
+                self.add_kprobe("__x64_sys_io_uring_enter", "trace_io_uring_enter_x64")
                 if self.developer_mode:
                     logger("info", "io_uring tracing enabled via __x64_sys_io_uring_enter")
             elif BPF.get_kprobe_functions(b'__sys_io_uring_enter'):

src/tracer/PathResolver.py

@@ -220,23 +220,29 @@ def resolve_path(self, inode: int, pid: int | None = None, filename: str | None
     def cleanup_old_cache(self):
         """
         Remove old entries from cache to prevent memory bloat.
-        
+
         Removes:
         - Process entries older than cache_timeout * 10 seconds
         - Limits inode cache to 5000 most recent entries
+
+        Thread-safety: this runs on the polling thread (from the perf-buffer
+        callbacks via cache maintenance), the same thread that mutates these
+        dicts. Iteration still works on list() snapshots and removals
+        tolerate missing entries as defense in depth.
         """
         current_time = time.time()
-        
+
         # Clean up process cache
         pids_to_remove = []
-        for pid, last_time in self.last_update.items():
+        for pid, last_time in list(self.last_update.items()):
             if current_time - last_time > self.cache_timeout * 10:
                 pids_to_remove.append(pid)
-        
+
         for pid in pids_to_remove:
             self.pid_to_files.pop(pid, None)
             self.last_update.pop(pid, None)
-        
+
+
         # Optionally limit inode cache size
         if len(self.inode_to_path) > 10000:
             # Keep only the most recent 5000 entries