Bugfix: release multi capture memory after analysis

- track collected counts independently from retained samples
- build multi results ZIPs from persisted transaction records
- release in-memory samples after stop analysis and ZIP export
- add best-effort Linux heap trimming after cleanup
- add pytest coverage for operation memory release - 2026-03-19 01:31:54

Author: mgarcia01752

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "pypnm-docsis"
-version         = "1.5.5.0"
+version         = "1.5.5.1"
 description     = "DOCSIS 3.x/4.0 Proactive Network Maintenance Toolkit"
 readme          = "README.md"
 requires-python = ">=3.10"

src/pypnm/api/routes/advance/common/abstract/multi_capture_router.py

@@ -19,6 +19,9 @@
     MultiCaptureOperationIdResponse,
 )
 from pypnm.api.routes.common.classes.file_capture.capture_group import CaptureGroup
+from pypnm.api.routes.common.classes.file_capture.pnm_file_opearation import (
+    OperationCaptureGroupResolver,
+)
 from pypnm.config.system_config_settings import SystemConfigSettings
 from pypnm.lib.types import GroupId, OperationId
 
@@ -62,17 +65,26 @@ def _build_results_zip_response(self, operation_id: OperationId, filename_prefix
 
         Missing files are skipped with warnings.
         """
-        service = self._get_service_or_404(operation_id)
-        samples = service.results(operation_id)
-
         pnm_dir = str(SystemConfigSettings.pnm_dir())
-        cm = getattr(service, "cm", None)
-        mac = getattr(getattr(cm, "get_mac_address", None), "mac_address", "unknown")
+        resolver = OperationCaptureGroupResolver()
+        transaction_models = resolver.get_transaction_models_for_operation(operation_id)
+
+        mac = "unknown"
+        if transaction_models:
+            mac = str(transaction_models[0].mac_address)
+        else:
+            op_record = OperationManager.get_operation_record(operation_id)
+            if isinstance(op_record, dict):
+                metadata = op_record.get("metadata")
+                if isinstance(metadata, dict):
+                    md_mac = metadata.get("mac_address")
+                    if isinstance(md_mac, str) and md_mac.strip():
+                        mac = md_mac
 
         buf = io.BytesIO()
         with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zipf:
-            for sample in samples:
-                filename = getattr(sample, "filename", "")
+            for model in transaction_models:
+                filename = str(model.filename)
                 if not str(filename).strip():
                     continue
 
@@ -87,6 +99,7 @@ def _build_results_zip_response(self, operation_id: OperationId, filename_prefix
 
         buf.seek(0)
         headers = {"Content-Disposition": f"attachment; filename={filename_prefix}_{mac}_{operation_id}.zip"}
+        self._release_operation_memory(operation_id)
         return StreamingResponse(buf, media_type="application/zip", headers=headers)
 
     def _repair_capture_group_from_service_samples(self, operation_id: OperationId, capture_group_id: GroupId) -> int:
@@ -141,3 +154,15 @@ def _build_operation_id_response(self, operation_name: MultiCaptureOperation) ->
         """Return persisted operation records for a canonical multi-capture operation family."""
         operations = OperationManager.list_operation_records_by_name(operation_name.value)
         return MultiCaptureOperationIdResponse(status="success", message=None, operations=operations)
+
+    def _release_operation_memory(self, operation_id: OperationId) -> None:
+        """Release transient in-memory capture state when a service is still registered."""
+        try:
+            service = self.getService(operation_id)
+        except Exception:
+            return
+
+        try:
+            service.release_operation_memory(operation_id)
+        except Exception as exc:
+            self.logger.debug("Operation memory release failed for %s: %s", operation_id, exc)

src/pypnm/api/routes/advance/common/capture_service.py

@@ -25,6 +25,7 @@
     MessageResponseType,
 )
 from pypnm.api.routes.common.service.status_codes import ServiceStatusCode
+from pypnm.lib.memory import ProcessMemory
 from pypnm.lib.types import GroupId, OperationId, TimeStamp
 from pypnm.lib.utils import Generate
 
@@ -129,6 +130,7 @@ async def start(self) -> tuple[GroupId, OperationId]:
             "duration":         self.duration,
             "interval":         self.interval,
             "time_remaining":   self.time_remaining,
+            "collected":        0,
             "samples":          [],
             "task":             None,
         }
@@ -162,15 +164,14 @@ async def _runner() -> None:
                         if isinstance(msg_rsp.payload, list):
                             samples = self._process_captures(msg_rsp)
                             for sample in samples:
-                                self._ops[operation_id]["samples"].append(sample)
-                                if sample.transaction_id:
-                                    self._cap_group.add_transaction(sample.transaction_id)
+                                self._append_operation_sample(operation_id, sample)
                                 self.logger.debug(f"[{operation_id}] Captured sample txn={sample.transaction_id}")
                         else:
                             status_name = msg_rsp.status.name if isinstance(msg_rsp.status, ServiceStatusCode) else str(msg_rsp.status)
                             err = f"Capture response returned no transaction payload (status={status_name})"
                             self.logger.warning(f"[{operation_id}] {err}")
-                            self._ops[operation_id]["samples"].append(
+                            self._append_operation_sample(
+                                operation_id,
                                 CaptureSample(
                                     timestamp=cast(TimeStamp, iteration_ts),
                                     transaction_id="",
@@ -182,10 +183,10 @@ async def _runner() -> None:
                     except Exception as exc:
                         error_msg = str(exc)
                         self.logger.error(f"[{operation_id}] Capture error: {error_msg}", exc_info=True)
-                        self._ops[operation_id]["samples"].append(CaptureSample(timestamp       =   cast(TimeStamp, iteration_ts),
-                                                                                transaction_id  =   "",
-                                                                                filename        =   "",
-                                                                                error           =   error_msg))
+                        self._append_operation_sample(
+                            operation_id,
+                            CaptureSample(timestamp=cast(TimeStamp, iteration_ts), transaction_id="", filename="", error=error_msg),
+                        )
 
                 # Complete if still running
                 if self._ops[operation_id]["state"] == OperationState.RUNNING:
@@ -209,15 +210,14 @@ async def _runner() -> None:
                             if isinstance(msg_rsp.payload, list):
                                 samples = self._process_captures(msg_rsp)
                                 for sample in samples:
-                                    self._ops[operation_id]["samples"].append(sample)
-                                    if sample.transaction_id:
-                                        self._cap_group.add_transaction(sample.transaction_id)
+                                    self._append_operation_sample(operation_id, sample)
                                     self.logger.info(f"[{operation_id}] Captured sample txn={sample.transaction_id}")
                             else:
                                 status_name = msg_rsp.status.name if isinstance(msg_rsp.status, ServiceStatusCode) else str(msg_rsp.status)
                                 err = f"Final capture response returned no transaction payload (status={status_name})"
                                 self.logger.warning(f"[{operation_id}] {err}")
-                                self._ops[operation_id]["samples"].append(
+                                self._append_operation_sample(
+                                    operation_id,
                                     CaptureSample(
                                         timestamp=cast(TimeStamp, iteration_ts),
                                         transaction_id="",
@@ -229,11 +229,10 @@ async def _runner() -> None:
                     except Exception as exc:
                         error_msg = str(exc)
                         self.logger.error(f"[{operation_id}] Capture error: {error_msg}", exc_info=True)
-                        self._ops[operation_id]["samples"].append(
-                            CaptureSample(timestamp         =   cast(TimeStamp, iteration_ts),
-                                          transaction_id    =   "",
-                                          filename          =   "",
-                                          error             =error_msg))
+                        self._append_operation_sample(
+                            operation_id,
+                            CaptureSample(timestamp=cast(TimeStamp, iteration_ts), transaction_id="", filename="", error=error_msg),
+                        )
             except asyncio.CancelledError:
                 if operation_id in self._ops and self._ops[operation_id]["state"] == OperationState.RUNNING:
                     self._ops[operation_id]["state"] = OperationState.CANCELLED
@@ -278,6 +277,20 @@ def getOperationID(self) -> OperationId:
     def getOperation(self, operation_id:OperationId) -> dict[str, dict[str, Any]]:
         return self._ops[operation_id]
 
+    def _append_operation_sample(self, operation_id: OperationId, sample: CaptureSample) -> None:
+        """
+        Append a capture sample and update operation-level counters.
+
+        Args:
+            operation_id: Operation receiving the sample.
+            sample: Parsed capture sample to retain.
+        """
+        op = self._ops[operation_id]
+        op["samples"].append(sample)
+        op["collected"] = int(op.get("collected", 0)) + 1
+        if sample.transaction_id:
+            self._cap_group.add_transaction(sample.transaction_id)
+
     def getOperationState(self,operation_id:OperationId) -> OperationState:
         return self._ops[operation_id]["state"]
 
@@ -314,7 +327,7 @@ def status(self, operation_id: OperationId) -> dict[str, Any]:
 
         return {
             "state": op["state"],
-            "collected": len(op["samples"]),
+            "collected": int(op.get("collected", 0)),
             "time_remaining": op.get("time_remaining", 0)
         }
 
@@ -331,6 +344,26 @@ def results(self, operation_id: OperationId) -> list[CaptureSample]:
         op = self._ops.get(operation_id)
         return op["samples"] if op else []
 
+    def release_operation_memory(self, operation_id: OperationId) -> None:
+        """
+        Drop retained in-memory samples and task references for an operation.
+
+        Args:
+            operation_id: Operation whose transient in-memory state should be released.
+        """
+        op = self._ops.get(operation_id)
+        if not op:
+            return
+
+        released_samples = len(op["samples"])
+        op["samples"] = []
+        op["task"] = None
+
+        if released_samples > 0:
+            self.logger.info("[%s] Released %d retained samples from memory", operation_id, released_samples)
+
+        ProcessMemory.release_unused_memory()
+
     def stop(self, operation_id: OperationId) -> None:
         """
         Signal the background runner to stop after the current iteration.

src/pypnm/api/routes/advance/multi_ds_chan_est/router.py

@@ -182,6 +182,7 @@ def stop_capture(operationId: OperationId) -> MultiChanEstStatusResponse:
 
             service.stop(operation_id)
             status = service.status(operation_id)
+            self._release_operation_memory(operation_id)
             return MultiChanEstStatusResponse(
                 mac_address =   service.cm.get_mac_address.mac_address,
                 system_description = service.get_system_description(),
@@ -281,6 +282,7 @@ def analysis(request: MultiChanEstAnalysisRequest) -> MultiChanEstimationAnalysi
                         "system_description": analysis_result.system_description,
                     }
 
+                self._release_operation_memory(request.operation_id)
                 return MultiChanEstimationAnalysisResponse(
                     mac_address =   mac,
                     status      =   status,
@@ -292,6 +294,7 @@ def analysis(request: MultiChanEstAnalysisRequest) -> MultiChanEstimationAnalysi
                 try:
                     rpt = engine.build_report()
                     self.logger.info(f"[analysis] Built archive report for group {capture_group_id}")
+                    self._release_operation_memory(request.operation_id)
                     return PnmFileService().get_file(FileType.ARCHIVE, rpt.name)
 
                 except Exception as e:

src/pypnm/api/routes/advance/multi_rxmer/router.py

@@ -334,6 +334,7 @@ def stop_capture(operationId: OperationId) -> MultiRxMerStatusResponse:
 
             service.stop(operation_id)
             status = service.status(operation_id)
+            self._release_operation_memory(operation_id)
 
             return MultiRxMerStatusResponse(
                 mac_address=service.cm.get_mac_address.mac_address,
@@ -512,6 +513,7 @@ def analysis(request: MultiRxMerAnalysisRequest) -> MultiRxMerAnalysisResponse |
                         "mac_address": mac_address,
                         "system_description": response_system_description,
                     }
+                self._release_operation_memory(request.operation_id)
                 return MultiRxMerAnalysisResponse(
                     mac_address =   mac_address,
                     status      =   status_code,
@@ -521,6 +523,7 @@ def analysis(request: MultiRxMerAnalysisRequest) -> MultiRxMerAnalysisResponse |
 
             elif output_type == OutputType.ARCHIVE:
                 rpt = engine.build_report()
+                self._release_operation_memory(request.operation_id)
                 return PnmFileService().get_file(FileType.ARCHIVE, rpt.name)
 
             else:

src/pypnm/api/routes/advance/multi_us_ofdma_pre_eq/router.py

@@ -182,6 +182,7 @@ def stop_capture(operationId: OperationId) -> MultiUsOfdmaPreEqStatusResponse:
 
             service.stop(operation_id)
             status = service.status(operation_id)
+            self._release_operation_memory(operation_id)
             return MultiUsOfdmaPreEqStatusResponse(
                 mac_address =   service.cm.get_mac_address.mac_address,
                 system_description = service.get_system_description(),
@@ -279,6 +280,7 @@ def analysis(request: MultiUsOfdmaPreEqAnalysisRequest) -> MultiUsOfdmaPreEqAnal
                         "system_description": analysis_result.system_description,
                     }
 
+                self._release_operation_memory(request.operation_id)
                 return MultiUsOfdmaPreEqAnalysisResponse(
                     mac_address =   mac,
                     status      =   status,
@@ -290,6 +292,7 @@ def analysis(request: MultiUsOfdmaPreEqAnalysisRequest) -> MultiUsOfdmaPreEqAnal
                 try:
                     rpt = engine.build_report()
                     self.logger.info(f"[analysis] Built archive report for group {capture_group_id}")
+                    self._release_operation_memory(request.operation_id)
                     return PnmFileService().get_file(FileType.ARCHIVE, rpt.name)
 
                 except Exception as e:

src/pypnm/lib/memory.py

@@ -0,0 +1,34 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2026
+
+from __future__ import annotations
+
+import ctypes
+import gc
+import logging
+
+
+class ProcessMemory:
+    """Helpers for best-effort process memory reclamation on long-running services."""
+
+    @staticmethod
+    def release_unused_memory() -> None:
+        """
+        Trigger Python garbage collection and best-effort libc heap trimming.
+
+        This is primarily useful after large analysis/capture objects have been
+        released and the service wants to return unused heap pages to the OS on
+        Linux/glibc systems. Failures are intentionally ignored.
+        """
+        gc.collect()
+
+        try:
+            libc = ctypes.CDLL("libc.so.6")
+            malloc_trim = getattr(libc, "malloc_trim", None)
+            if callable(malloc_trim):
+                malloc_trim(0)
+        except Exception as exc:
+            logging.getLogger(ProcessMemory.__name__).debug(
+                "malloc_trim not available or failed: %s",
+                exc,
+            )

src/pypnm/version.py

@@ -6,4 +6,4 @@
 __all__ = ["__version__"]
 
 # MAJOR.MINOR.MAINTENANCE.BUILD
-__version__: str = "1.5.5.0"
+__version__: str = "1.5.5.1"

tests/test_capture_service_transaction_guard.py

@@ -13,6 +13,7 @@
     MultiCaptureOperation,
     MultiCaptureOperationModel,
 )
+from pypnm.api.routes.common.classes.file_capture.capture_sample import CaptureSample
 from pypnm.api.routes.common.extended.common_messaging_service import (
     MessageResponse,
     MessageResponseType,
@@ -176,3 +177,27 @@ async def test_capture_service_stop_cancels_background_task(monkeypatch: pytest.
 
     assert service._ops[operation_id]["state"] == capture_service.OperationState.STOPPED
     assert task.cancelled() is True or task.done() is True
+
+
+def test_release_operation_memory_keeps_collected_count(monkeypatch: pytest.MonkeyPatch) -> None:
+    service = _FakeCaptureService(duration=0, interval=0)
+    service._ops["op-1"] = {
+        "state": capture_service.OperationState.COMPLETED,
+        "collected": 2,
+        "samples": [
+            CaptureSample(timestamp=1, transaction_id="tx1", filename="a.bin", error=None),
+            CaptureSample(timestamp=2, transaction_id="tx2", filename="b.bin", error=None),
+        ],
+        "task": object(),
+        "time_remaining": 0,
+    }
+
+    released_calls: list[str] = []
+    monkeypatch.setattr(capture_service.ProcessMemory, "release_unused_memory", lambda: released_calls.append("released"))
+
+    service.release_operation_memory(OperationId("op-1"))
+
+    assert service._ops["op-1"]["samples"] == []
+    assert service._ops["op-1"]["task"] is None
+    assert service.status(OperationId("op-1"))["collected"] == 2
+    assert released_calls == ["released"]