Revert "Report endpoint_payload.dropped when the orchestrator removes a stored payload"

This reverts commit 5441bb5.

Author: ypopovych

Sources/DatadogSDKTesting/Telemetry/ARCHITECTURE.md

@@ -144,14 +144,10 @@ 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`, `dropped`, tagged
+  `requests_errors`, `events_count`, `events_serialization_ms`, tagged
   `test_cycle` (spans) / `code_coverage` (coverage). Wired in
   `DDTracer.endpointPayloadObservers(...)`. This family has **no feature call
-  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.
+  site** — the observers are its only home.
 - **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,
@@ -168,7 +164,12 @@ 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. **Local feature metrics** — emitted directly via `SessionConfig.telemetry` /
+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` /
    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`,
@@ -181,9 +182,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.
-3. **`impacted_tests_detection.*`** — no feature/API exists yet; instruments are
+4. **`impacted_tests_detection.*`** — no feature/API exists yet; instruments are
    defined but unused.
-4. **`error_type` granularity** — `Telemetry.errorType(statusCode:)` maps `nil`
+5. **`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,12 +40,10 @@ 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(),
-            onDrop: uploadObserver.map { obs in { obs.uploadDropped(payloadBytes: $0) } }
+            dateProvider: SystemDateProvider()
         )
 
         let encoder = api.encoder

Sources/EventsExporter/Persistence/FilesOrchestrator.swift

@@ -148,55 +148,47 @@ 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 -> (file: ReadableFile?, droppedBytes: [Int])
+                                dateProvider: borrowing DateProvider) throws -> ReadableFile?
         {
-            let (infos, droppedBytes) = try fileInfos(directory: directory,
-                                                      performance: performance,
-                                                      dateProvider: dateProvider)
-            guard let oldest = infos.first else { return (nil, droppedBytes) }
+            guard let oldest = try fileInfos(directory: directory,
+                                             performance: performance,
+                                             dateProvider: dateProvider).first
+            else { return nil }
             let age = dateProvider.currentDate().timeIntervalSince(oldest.creationDate)
-            return (age >= performance.minFileAgeForRead ? oldest.file : nil, droppedBytes)
+            return age >= performance.minFileAgeForRead ? oldest.file : nil
         }
 
         func allReadableFiles(directory: borrowing Directory,
                               performance: borrowing StoragePerformancePreset,
-                              dateProvider: borrowing DateProvider) throws -> (files: [ReadableFile], droppedBytes: [Int])
+                              dateProvider: borrowing DateProvider) throws -> [ReadableFile]
         {
-            let (infos, droppedBytes) = try fileInfos(directory: directory,
-                                                      performance: performance,
-                                                      dateProvider: dateProvider)
-            return (infos.map { $0.file }, droppedBytes)
+            try fileInfos(directory: directory,
+                          performance: performance,
+                          dateProvider: dateProvider).map { $0.file }
         }
 
         private func fileInfos(directory: borrowing Directory,
                                performance: borrowing StoragePerformancePreset,
-                               dateProvider: borrowing DateProvider) throws -> (files: [FileInfo], droppedBytes: [Int])
+                               dateProvider: borrowing DateProvider) throws -> [FileInfo]
         {
             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(), droppedBytes)
+            return readableFiles.sorted()
         }
     }
 
@@ -208,20 +200,14 @@ 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,
-         onDrop: (@Sendable (Int) -> Void)? = nil)
+         dateProvider: DateProvider)
     {
         self.directory = directory
         self.dateProvider = dateProvider
         self.performance = performance
-        self.onDrop = onDrop
         self.state = Synced(.init())
     }
 
@@ -259,30 +245,15 @@ internal final class FilesOrchestrator: FilesOrchestratorType {
     // MARK: - `ReadableFile` orchestration
 
     func getReadableFile() throws -> ReadableFile? {
-        let (file, droppedBytes) = try state.use {
-            try $0.oldestReadableFile(directory: directory,
-                                      performance: performance,
-                                      dateProvider: dateProvider)
-        }
-        reportDrops(droppedBytes)
-        return file
+        try state.use { try $0.oldestReadableFile(directory: directory,
+                                                  performance: performance,
+                                                  dateProvider: dateProvider) }
     }
 
     func getAllReadableFiles() throws -> [ReadableFile] {
-        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)
+        try state.use { try $0.allReadableFiles(directory: directory,
+                                                performance: performance,
+                                                dateProvider: dateProvider) }
     }
 
     func delete(readableFile: ReadableFile) throws {
@@ -301,10 +272,7 @@ 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
@@ -318,8 +286,6 @@ 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

@@ -17,12 +17,10 @@ internal final class SpansExporter: SpanExporter {
          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(),
-            onDrop: uploadObserver.map { obs in { obs.uploadDropped(payloadBytes: $0) } }
+            dateProvider: SystemDateProvider()
         )
 
         var metadata = SpanSanitizer().sanitize(metadata: config.metadata)

Tests/EventsExporter/Persistence/FilesOrchestratorTests.swift

@@ -223,47 +223,6 @@ 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)