Merge branch 'pr4/install-marker-and-identity' into pr5/surface-headers

integrations/agent-plugin-core/python/telemetry.py

@@ -167,6 +167,18 @@ def _install_salt() -> str:
         return _salt_cache
 
     path = _salt_path()
+    # Read before writing. Hooks are separate processes firing on every tool
+    # call, so all but the first find the salt already published; going straight
+    # to create-fsync-link-unlink meant every one of them paid an fsync to
+    # discover that, on a path whose whole promise is appending a line and
+    # returning.
+    try:
+        _salt_cache = path.read_text(encoding="utf-8").strip()
+        if _salt_cache:
+            return _salt_cache
+    except OSError:
+        pass
+
     temporary = path.with_name(f"{path.name}.{os.getpid()}.tmp")
     try:
         path.parent.mkdir(parents=True, exist_ok=True)
@@ -617,10 +629,16 @@ def _claim_spool() -> Path | None:
 
 
 def _sweep_debris(directory: Path) -> None:
-    """Remove temp files orphaned by a crash between write and rename.
+    """Remove files nothing else will ever pick up again.
 
-    Neither glob in this module matches *.partial, so nothing else would ever
-    clean them up.
+    *.partial is a temp file orphaned by a crash between write and rename.
+    *.corrupt is a batch quarantined for undecodable content. No glob in this
+    module matches either, so without this they accumulate on disk for the life
+    of the install.
+
+    Quarantined batches are kept far longer than debris: they are the only
+    evidence left of events that could not be delivered, and someone diagnosing
+    a report of missing telemetry has to be able to find one.
     """
     now = time.time()
     for debris in directory.glob("telemetry-*.partial"):
@@ -629,6 +647,12 @@ def _sweep_debris(directory: Path) -> None:
                 debris.unlink()
         except OSError:
             continue
+    for quarantined in directory.glob("telemetry-*.corrupt"):
+        try:
+            if now - quarantined.stat().st_mtime > CLAIM_EXPIRY_SECONDS:
+                quarantined.unlink()
+        except OSError:
+            continue
 
 
 def _claim_parked(directory: Path) -> Path | None:

integrations/agent-plugin-core/tests/test_spool_delivery.py

@@ -409,3 +409,24 @@ def test_partial_files_are_swept(telemetry):
 
     telemetry.flush()
     assert not debris.exists()
+
+
+def test_quarantined_batches_are_eventually_collected(telemetry):
+    """Nothing re-globs .corrupt, so without a sweep they live on disk forever.
+
+    Kept much longer than .partial debris on purpose: a quarantined batch is the
+    only remaining evidence of events that could not be delivered.
+    """
+    directory = telemetry.memory_core.data_dir()
+    directory.mkdir(parents=True, exist_ok=True)
+    fresh = directory / "telemetry-1-aaaaaaaa-a0.corrupt"
+    old = directory / "telemetry-2-bbbbbbbb-a0.corrupt"
+    for path in (fresh, old):
+        path.write_text("torn", encoding="utf-8")
+    expired = time.time() - (telemetry.CLAIM_EXPIRY_SECONDS + 60)
+    os.utime(old, (expired, expired))
+
+    telemetry._sweep_debris(directory)
+
+    assert fresh.exists(), "a recent quarantine was discarded before anyone could look at it"
+    assert not old.exists(), "an expired quarantine was left on disk forever"