Report endpoint_payload.dropped when the orchestrator removes a stored payload

Sources/DatadogSDKTesting/Telemetry/ARCHITECTURE.md

@@ -144,10 +144,14 @@ methods are the protocol requirements; no-observer convenience methods are
 
 ### Gathered (3776)
 - **`endpoint_payload.*`** — `requests`, `requests_ms`, `bytes` (request size),
-  `requests_errors`, `events_count`, `events_serialization_ms`, tagged
+  `requests_errors`, `events_count`, `events_serialization_ms`, `dropped`, tagged
   `test_cycle` (spans) / `code_coverage` (coverage). Wired in
   `DDTracer.endpointPayloadObservers(...)`. This family has **no feature call
-  site** — the observers are its only home.
+  site** — the observers are its only home. `dropped` is reported from
+  `FilesOrchestrator` (via its `onDrop` callback → `UploadObserver.uploadDropped`)
+  when a stored batch is removed without being uploaded: too old
+  (`maxFileAgeForRead`) or purged to keep the directory under `maxDirectorySize`.
+  Successful upload deletions (`delete(readableFile:)`) are **not** drops.
 - **API request families** — `git_requests.{settings,search_commits,objects_pack}`
   (+ `_ms`, `_errors`, and `objects_pack_bytes`), `itr_skippable_tests.{request,
   request_ms,request_errors,response_bytes}`, `known_tests.{request,request_ms,
@@ -164,12 +168,7 @@ methods are the protocol requirements; no-observer convenience methods are
    - `itr_skippable_tests.response_tests` / `response_suites`
    - `known_tests.response_tests`
    - `test_management_tests.response_tests`
-2. **`endpoint_payload.dropped`** — the `UploadObserver.uploadDropped` hook exists
-   but the worker has no retry-exhaustion drop today; failed batches stay on disk
-   and are age-purged by `FilesOrchestrator`. Wire `dropped` from the purge path
-   (or add an explicit drop) — see `DDTracer.endpointPayloadObservers` where
-   `onDropped` is already mapped to `endpointPayload.dropped`.
-3. **Local feature metrics** — emitted directly via `SessionConfig.telemetry` /
+2. **Local feature metrics** — emitted directly via `SessionConfig.telemetry` /
    the feature's injected `Telemetry`, at the feature instrumentation sites:
    - `events.created` / `events.finished` (+ all their tags: `event_type`,
      `test_framework`, `is_new`, `is_modified`, `is_retry`, `retry_reason`,
@@ -182,9 +181,9 @@ methods are the protocol requirements; no-observer convenience methods are
    - `git.commit_sha_match` / `git.commit_sha_discrepancy` — git info providers.
    - `itr.skipped` / `itr.unskippable` / `itr.forced_run` — `TestImpactAnalysis`.
    - `code_coverage.{started,finished,is_empty,errors,files}` — coverage feature.
-4. **`impacted_tests_detection.*`** — no feature/API exists yet; instruments are
+3. **`impacted_tests_detection.*`** — no feature/API exists yet; instruments are
    defined but unused.
-5. **`error_type` granularity** — `Telemetry.errorType(statusCode:)` maps `nil`
+4. **`error_type` granularity** — `Telemetry.errorType(statusCode:)` maps `nil`
    status to `.network`; it cannot distinguish `timeout` from `network` without
    the underlying `URLError`. Refine if needed.
 

Sources/EventsExporter/CoverageExporter/CoverageExporter.swift

@@ -40,10 +40,12 @@ internal final class CoverageExporter: CoverageExporterType {
          observers: ExporterObservers.Feature = .init()) throws {
         self.configuration = config
 
+        let uploadObserver = observers.upload
         let filesOrchestrator = FilesOrchestrator(
             directory: try storage.createSubdirectory(path: "v1"),
             performance: PerformancePreset.instantDataDelivery,
-            dateProvider: SystemDateProvider()
+            dateProvider: SystemDateProvider(),
+            onDrop: uploadObserver.map { obs in { obs.uploadDropped(payloadBytes: $0) } }
         )
 
         let encoder = api.encoder

Sources/EventsExporter/Persistence/FilesOrchestrator.swift

@@ -148,47 +148,55 @@ internal final class FilesOrchestrator: FilesOrchestratorType {
 
         // MARK: - Reader
 
+        /// Reader results paired with the byte sizes of any files dropped
+        /// (deleted for exceeding `maxFileAgeForRead`) during the scan. The
+        /// caller reports the drops *outside* the state lock.
         func oldestReadableFile(directory: borrowing Directory,
                                 performance: borrowing StoragePerformancePreset,
-                                dateProvider: borrowing DateProvider) throws -> ReadableFile?
+                                dateProvider: borrowing DateProvider) throws -> (file: ReadableFile?, droppedBytes: [Int])
         {
-            guard let oldest = try fileInfos(directory: directory,
-                                             performance: performance,
-                                             dateProvider: dateProvider).first
-            else { return nil }
+            let (infos, droppedBytes) = try fileInfos(directory: directory,
+                                                      performance: performance,
+                                                      dateProvider: dateProvider)
+            guard let oldest = infos.first else { return (nil, droppedBytes) }
             let age = dateProvider.currentDate().timeIntervalSince(oldest.creationDate)
-            return age >= performance.minFileAgeForRead ? oldest.file : nil
+            return (age >= performance.minFileAgeForRead ? oldest.file : nil, droppedBytes)
         }
 
         func allReadableFiles(directory: borrowing Directory,
                               performance: borrowing StoragePerformancePreset,
-                              dateProvider: borrowing DateProvider) throws -> [ReadableFile]
+                              dateProvider: borrowing DateProvider) throws -> (files: [ReadableFile], droppedBytes: [Int])
         {
-            try fileInfos(directory: directory,
-                          performance: performance,
-                          dateProvider: dateProvider).map { $0.file }
+            let (infos, droppedBytes) = try fileInfos(directory: directory,
+                                                      performance: performance,
+                                                      dateProvider: dateProvider)
+            return (infos.map { $0.file }, droppedBytes)
         }
 
         private func fileInfos(directory: borrowing Directory,
                                performance: borrowing StoragePerformancePreset,
-                               dateProvider: borrowing DateProvider) throws -> [FileInfo]
+                               dateProvider: borrowing DateProvider) throws -> (files: [FileInfo], droppedBytes: [Int])
         {
             let allFiles = try directory.files()
                 .filter { !activeWrites.contains($0.name) }
                 .map { FileInfo(file: $0) }
 
             var readableFiles: [FileInfo] = []
             readableFiles.reserveCapacity(allFiles.count)
+            var droppedBytes: [Int] = []
             for info in allFiles {
                 let fileAge = dateProvider.currentDate().timeIntervalSince(info.creationDate)
                 if fileAge > performance.maxFileAgeForRead {
+                    // Too old to ever upload — count it as a dropped payload.
+                    let size = (try? info.file.size()).map(Int.init) ?? 0
                     try info.file.delete()
+                    droppedBytes.append(size)
                 } else {
                     readableFiles.append(info)
                 }
             }
 
-            return readableFiles.sorted()
+            return (readableFiles.sorted(), droppedBytes)
         }
     }
 
@@ -200,14 +208,20 @@ internal final class FilesOrchestrator: FilesOrchestratorType {
     private let directory: Directory
     private let dateProvider: DateProvider
     private let performance: StoragePerformancePreset
+    /// Invoked (outside the state lock) with the byte size of each file removed
+    /// without being uploaded — too old (`maxFileAgeForRead`) or purged to keep
+    /// the directory under `maxDirectorySize`. Wired to `endpoint_payload.dropped`.
+    private let onDrop: (@Sendable (Int) -> Void)?
 
     init(directory: Directory,
          performance: StoragePerformancePreset,
-         dateProvider: DateProvider)
+         dateProvider: DateProvider,
+         onDrop: (@Sendable (Int) -> Void)? = nil)
     {
         self.directory = directory
         self.dateProvider = dateProvider
         self.performance = performance
+        self.onDrop = onDrop
         self.state = Synced(.init())
     }
 
@@ -245,15 +259,30 @@ internal final class FilesOrchestrator: FilesOrchestratorType {
     // MARK: - `ReadableFile` orchestration
 
     func getReadableFile() throws -> ReadableFile? {
-        try state.use { try $0.oldestReadableFile(directory: directory,
-                                                  performance: performance,
-                                                  dateProvider: dateProvider) }
+        let (file, droppedBytes) = try state.use {
+            try $0.oldestReadableFile(directory: directory,
+                                      performance: performance,
+                                      dateProvider: dateProvider)
+        }
+        reportDrops(droppedBytes)
+        return file
     }
 
     func getAllReadableFiles() throws -> [ReadableFile] {
-        try state.use { try $0.allReadableFiles(directory: directory,
-                                                performance: performance,
-                                                dateProvider: dateProvider) }
+        let (files, droppedBytes) = try state.use {
+            try $0.allReadableFiles(directory: directory,
+                                    performance: performance,
+                                    dateProvider: dateProvider)
+        }
+        reportDrops(droppedBytes)
+        return files
+    }
+
+    /// Report dropped-payload sizes. Called outside the state lock so the
+    /// observer (which may touch its own locks) can't contend with file ops.
+    private func reportDrops(_ droppedBytes: [Int]) {
+        guard let onDrop else { return }
+        droppedBytes.forEach(onDrop)
     }
 
     func delete(readableFile: ReadableFile) throws {
@@ -272,7 +301,10 @@ internal final class FilesOrchestrator: FilesOrchestratorType {
             .compactMap { (info) -> FileInfo? in
                 let fileAge = dateProvider.currentDate().timeIntervalSince(info.creationDate)
                 if fileAge > performance.maxFileAgeForRead {
+                    // Too old to ever upload — count it as a dropped payload.
+                    let size = (try? info.file.size()).map(Int.init) ?? 0
                     try info.file.delete()
+                    onDrop?(size)
                     return nil
                 }
                 return info
@@ -286,6 +318,8 @@ internal final class FilesOrchestrator: FilesOrchestratorType {
                 let fileWithSize = filesWithSize.removeFirst()
                 try fileWithSize.file.delete()
                 sizeFreed += fileWithSize.size
+                // Purged to stay under the directory size limit — also a drop.
+                onDrop?(Int(fileWithSize.size))
             }
         }
     }

Sources/EventsExporter/Spans/SpansExporter.swift

@@ -8,19 +8,21 @@
 import OpenTelemetrySdk
 
 internal final class SpansExporter: SpanExporter {
     let configuration: ExporterConfiguration
     let runtimeId: String
     let spansStorage: FeatureStoreAndUpload
     private let encoder: JSONEncoder
 
     init(config: ExporterConfiguration, storage: Directory, api: SpansApi,
          observers: ExporterObservers.Feature = .init()) throws {
         self.configuration = config
 
+        let uploadObserver = observers.upload
         let filesOrchestrator = FilesOrchestrator(
             directory: try storage.createSubdirectory(path: "v1"),
             performance: configuration.performancePreset,
-            dateProvider: SystemDateProvider()
+            dateProvider: SystemDateProvider(),
+            onDrop: uploadObserver.map { obs in { obs.uploadDropped(payloadBytes: $0) } }
         )
 
         var metadata = SpanSanitizer().sanitize(metadata: config.metadata)

Tests/EventsExporter/Persistence/FilesOrchestratorTests.swift

@@ -223,6 +223,47 @@ class FilesOrchestratorTests: XCTestCase {
         XCTAssertEqual(try temporaryDirectory.files().count, 0)
     }
 
+    func testGivenFileTooOld_whenScanned_itReportsDropWithSize() throws {
+        final class Box: @unchecked Sendable { var sizes: [Int] = [] }
+        let box = Box()
+        let dateProvider = RelativeDateProvider()
+        let orchestrator = FilesOrchestrator(
+            directory: temporaryDirectory,
+            performance: performance,
+            dateProvider: dateProvider,
+            onDrop: { box.sizes.append($0) }
+        )
+        let file = try temporaryDirectory.createFile(named: dateProvider.currentDate().toFileName)
+        try file.append(data: Data("hello".utf8)) // 5 bytes
+
+        dateProvider.advance(bySeconds: 2 * performance.maxFileAgeForRead)
+
+        // Scanning for a readable file finds it too old, deletes it, and reports
+        // the drop with the file's size.
+        XCTAssertNil(try orchestrator.getReadableFile())
+        XCTAssertEqual(try temporaryDirectory.files().count, 0)
+        XCTAssertEqual(box.sizes, [5])
+    }
+
+    func testGivenSuccessfulRead_whenDeleted_itDoesNotReportDrop() throws {
+        final class Box: @unchecked Sendable { var sizes: [Int] = [] }
+        let box = Box()
+        let dateProvider = RelativeDateProvider()
+        let orchestrator = FilesOrchestrator(
+            directory: temporaryDirectory,
+            performance: performance,
+            dateProvider: dateProvider,
+            onDrop: { box.sizes.append($0) }
+        )
+        _ = try temporaryDirectory.createFile(named: dateProvider.currentDate().toFileName)
+        dateProvider.advance(bySeconds: 1 + performance.minFileAgeForRead)
+
+        let readableFile = try orchestrator.getReadableFile().unwrapOrThrow()
+        try orchestrator.delete(readableFile: readableFile) // marked-as-read, not a drop
+
+        XCTAssertEqual(box.sizes, [], "Successful upload deletion must not be reported as a drop")
+    }
+
     func testItDeletesReadableFile() throws {
         let dateProvider = RelativeDateProvider()
         let orchestrator = configureOrchestrator(using: dateProvider)