tracking: one ledger, one authority — prune protection and replace-vs-add answered from the same records, fail-safe on unreadable state

src/core/wire/sample_usage.cpp

@@ -122,6 +122,21 @@ UsagePublishPlan planUsagePublish(const std::optional<std::string>& existing,
     return plan;
 }
 
+namespace {
+
+// The liveness rule, in one place so the path fold and the attribution fold can
+// never disagree about which records counted. `protectAll` is the caller's
+// zero-identified net (see usageHeldPaths).
+bool recordCounts(const UsageRecord& rec,
+                  const std::unordered_set<std::string>& liveTrackGuids,
+                  bool anyInstanceLive, bool protectAll) {
+    if (protectAll) return true;
+    if (rec.trackGuid.empty()) return anyInstanceLive;
+    return liveTrackGuids.count(rec.trackGuid) != 0;
+}
+
+}  // namespace
+
 std::vector<std::string> usageHeldPaths(
     const std::vector<UsageRecord>& records,
     const std::unordered_set<std::string>& liveTrackGuids,
@@ -133,11 +148,7 @@ std::vector<std::string> usageHeldPaths(
     // paths rather than none (zero-identified must never degrade toward delete).
     const bool protectAll = !records.empty() && !anyInstanceLive;
     for (const UsageRecord& rec : records) {
-        const bool live = protectAll ||
-                          (rec.trackGuid.empty()
-                               ? anyInstanceLive
-                               : (liveTrackGuids.count(rec.trackGuid) != 0));
-        if (!live) continue;
+        if (!recordCounts(rec, liveTrackGuids, anyInstanceLive, protectAll)) continue;
         for (const UsageHold& h : rec.holds) {
             if (h.relativePath.empty()) continue;
             if (seen.insert(h.relativePath).second) out.push_back(h.relativePath);
@@ -147,33 +158,44 @@ std::vector<std::string> usageHeldPaths(
 }
 
 UsageFoldResult foldUsageRecords(
-    const std::vector<std::optional<UsageRecord>>& decoded,
+    const std::vector<DecodedUsage>& decoded,
     const std::unordered_set<std::string>& liveTrackGuids,
     bool anyInstanceLive) {
     UsageFoldResult result;
+    for (const DecodedUsage& entry : decoded) {
+        if (entry.record) continue;
+        // Present-but-unreadable record: it may protect anything, so halt.
+        result.abortPrune = true;
+        result.offendingKeys.push_back(entry.key);
+    }
+    if (result.abortPrune) {
+        // Belt-and-braces: return the protect-all set (every readable record's
+        // paths, bypassing the liveness filter) so the fail-safe holds even if a
+        // future caller forgets to check abortPrune first. `counted` stays empty —
+        // attribution is exactly what an unreadable record makes unknowable.
+        std::unordered_set<std::string> seen;
+        for (const DecodedUsage& entry : decoded) {
+            if (!entry.record) continue;
+            for (const UsageHold& h : entry.record->holds) {
+                if (h.relativePath.empty()) continue;
+                if (seen.insert(h.relativePath).second)
+                    result.heldPaths.push_back(h.relativePath);
+            }
+        }
+        return result;
+    }
+
     std::vector<UsageRecord> records;
     records.reserve(decoded.size());
-    for (const std::optional<UsageRecord>& rec : decoded) {
-        if (!rec) {
-            // Present-but-unreadable record: it may protect anything, so halt.
-            // Belt-and-braces: also return the protect-all set (every readable
-            // record's paths, bypassing the liveness filter) so the fail-safe
-            // holds even if a future caller forgets to check abortPrune first.
-            result.abortPrune = true;
-            std::unordered_set<std::string> seen;
-            for (const std::optional<UsageRecord>& r : decoded) {
-                if (!r) continue;
-                for (const UsageHold& h : r->holds) {
-                    if (h.relativePath.empty()) continue;
-                    if (seen.insert(h.relativePath).second)
-                        result.heldPaths.push_back(h.relativePath);
-                }
-            }
-            return result;
-        }
-        records.push_back(*rec);
-    }
+    for (const DecodedUsage& entry : decoded) records.push_back(*entry.record);
     result.heldPaths = usageHeldPaths(records, liveTrackGuids, anyInstanceLive);
+
+    const bool protectAll = !records.empty() && !anyInstanceLive;
+    for (const DecodedUsage& entry : decoded) {
+        if (!recordCounts(*entry.record, liveTrackGuids, anyInstanceLive, protectAll))
+            continue;
+        result.counted.push_back(CountedUsage{entry.key, *entry.record});
+    }
     return result;
 }