[pbs-devel] [PATCH proxmox-backup v2] task tracking: improve pruning and fix accounting for missing entries

[Hannes Laimer]

Refactor the active operation tracking to use the
`check_process_running_pstart` helper. This simplifies the logic for
identifying stale entries caused by PID reuse and aligns the update path
with the read path.

Additionally, fix a logic bug where decrementing the operation count for
a non-existent entry would incorrectly create a new entry with a
positive count of 1. Now, such operations are ignored, and new entries
are only created when the count is positive.

Suggested-by: Fabian Grünbichler <f.gruenbichler at proxmox.com>
Signed-off-by: Hannes Laimer <h.laimer at proxmox.com>
---
v2, thanks @Fabian!:
- use `check_process_running_pstart` instead of `check_process_running` + starttime comparison
- improve readability by flattening match logic
- fix bug for decrements when no entry exists (led to new entry with 1
  before)

 pbs-datastore/src/task_tracking.rs | 52 ++++++++++++++++--------------
 1 file changed, 27 insertions(+), 25 deletions(-)

diff --git a/pbs-datastore/src/task_tracking.rs b/pbs-datastore/src/task_tracking.rs
index 44a4522d..10afebbe 100644
--- a/pbs-datastore/src/task_tracking.rs
+++ b/pbs-datastore/src/task_tracking.rs
@@ -103,43 +103,45 @@ pub fn update_active_operations(
     let pid = std::process::id();
     let starttime = procfs::PidStat::read_from_pid(Pid::from_raw(pid as pid_t))?.starttime;
 
-    let mut updated_active_operations = match operation {
-        Operation::Read => ActiveOperationStats { read: 1, write: 0 },
-        Operation::Write => ActiveOperationStats { read: 0, write: 1 },
-        Operation::Lookup => ActiveOperationStats { read: 0, write: 0 },
-    };
+    let mut updated_active_operations = ActiveOperationStats::default();
     let mut found_entry = false;
     let mut updated_tasks: Vec<TaskOperations> = match file_read_optional_string(&path)? {
         Some(data) => serde_json::from_str::<Vec<TaskOperations>>(&data)?
-            .iter_mut()
-            .filter_map(
-                |task| match procfs::check_process_running(task.pid as pid_t) {
-                    Some(stat) if pid == task.pid && stat.starttime != task.starttime => None,
-                    Some(_) => {
-                        if pid == task.pid {
-                            found_entry = true;
-                            match operation {
-                                Operation::Read => task.active_operations.read += count,
-                                Operation::Write => task.active_operations.write += count,
-                                Operation::Lookup => (), // no IO must happen there
-                            };
-                            updated_active_operations = task.active_operations;
-                        }
-                        Some(task.clone())
+            .into_iter()
+            .filter_map(|mut task| {
+                match procfs::check_process_running_pstart(task.pid as pid_t, task.starttime) {
+                    // update entry for current PID
+                    Some(_stat) if pid == task.pid => {
+                        found_entry = true;
+                        match operation {
+                            Operation::Read => task.active_operations.read += count,
+                            Operation::Write => task.active_operations.write += count,
+                            Operation::Lookup => (), // no IO must happen there
+                        };
+                        updated_active_operations = task.active_operations;
+                        Some(task)
                     }
-                    _ => None,
-                },
-            )
+                    // keep other entries
+                    Some(_stat) => Some(task),
+                    // drop entries for PIDs which are not running or have been recycled
+                    None => None,
+                }
+            })
             .collect(),
         None => Vec::new(),
     };
 
-    if !found_entry {
+    if !found_entry && count > 0 {
+        match operation {
+            Operation::Read => updated_active_operations.read = count,
+            Operation::Write => updated_active_operations.write = count,
+            Operation::Lookup => (),
+        };
         updated_tasks.push(TaskOperations {
             pid,
             starttime,
             active_operations: updated_active_operations,
-        })
+        });
     }
     replace_file(
         &path,
-- 
2.47.3