Fix Windows file deletion by closing stream before removeFile (#287)

Follow-up to #284

Move `storage.removeFile()` call outside the `use{}` block to ensure
`InputStream` is fully closed before attempting file deletion. On Windows,
files cannot be deleted while file handles remain open, causing silent
deletion failures and infinite upload retries of the same batch files.

Author: apuig

core/src/main/java/com/segment/analytics/kotlin/core/platform/EventPipeline.kt

@@ -132,8 +132,8 @@ open class EventPipeline(
             val fileUrlList = parseFilePaths(storage.read(Storage.Constants.Events))
             for (url in fileUrlList) {
                 // upload event file
+                var shouldCleanup = true
                 storage.readAsStream(url)?.use { data ->
-                    var shouldCleanup = true
                     try {
                         val connection = httpClient.upload(apiHost)
                         connection.outputStream?.let {
@@ -150,10 +150,10 @@ open class EventPipeline(
                         analytics.reportInternalError(e)
                         shouldCleanup = handleUploadException(e, url)
                     }
+                }
 
-                    if (shouldCleanup) {
-                        storage.removeFile(url)
-                    }
+                if (shouldCleanup) {
+                    storage.removeFile(url)
                 }
             }
         }

core/src/test/kotlin/com/segment/analytics/kotlin/core/platform/EventPipelineTest.kt

@@ -234,4 +234,93 @@ internal class EventPipelineTest {
         // Verify that close() was called on the InputStream even when exception occurred
         assertTrue(isClosed, "InputStream should have been closed when exception occurs")
     }
+
+    @Test
+    fun `removeFile is called after InputStream is closed`() {
+        // Track order of operations to ensure stream is closed before file deletion
+        val operationOrder = mutableListOf<String>()
+        var streamClosed = false
+
+        val trackableInputStream = object : java.io.InputStream() {
+            override fun read(): Int = -1
+            override fun close() {
+                streamClosed = true
+                operationOrder.add("stream_closed")
+                super.close()
+            }
+        }
+
+        every { storage.readAsStream(any()) } returns trackableInputStream
+        every { storage.removeFile(any()) } answers {
+            operationOrder.add("file_removed")
+            // Verify stream was closed before removeFile is called (Windows file locking requirement)
+            assertTrue(streamClosed, "InputStream must be closed before removeFile is called")
+            true
+        }
+
+        pipeline.put(event1)
+        pipeline.put(event2)
+
+        // Give some time for async processing
+        Thread.sleep(500)
+
+        // Verify correct operation order
+        coVerify {
+            storage.removeFile(any())
+        }
+        assertTrue(operationOrder == listOf("stream_closed", "file_removed"),
+            "Expected order: [stream_closed, file_removed], but got: $operationOrder")
+    }
+
+    @Test
+    fun `removeFile is called when readAsStream returns null`() {
+        every { storage.readAsStream(any()) } returns null
+
+        pipeline.put(event1)
+        pipeline.put(event2)
+
+        // Give some time for async processing
+        Thread.sleep(500)
+
+        // Verify file is still deleted even when stream is null
+        coVerify {
+            storage.removeFile(any())
+        }
+    }
+
+    @Test
+    fun `InputStream is closed before removeFile even when upload throws exception`() {
+        val operationOrder = mutableListOf<String>()
+        var streamClosed = false
+
+        val trackableInputStream = object : java.io.InputStream() {
+            override fun read(): Int = -1
+            override fun close() {
+                streamClosed = true
+                operationOrder.add("stream_closed")
+                super.close()
+            }
+        }
+
+        every { storage.readAsStream(any()) } returns trackableInputStream
+        every { anyConstructed<HTTPClient>().upload(any()) } throws HTTPException(400, "", "", mutableMapOf())
+        every { storage.removeFile(any()) } answers {
+            operationOrder.add("file_removed")
+            assertTrue(streamClosed, "InputStream must be closed before removeFile even when upload fails")
+            true
+        }
+
+        pipeline.put(event1)
+        pipeline.put(event2)
+
+        // Give some time for async processing
+        Thread.sleep(500)
+
+        // Verify correct operation order even with exception
+        coVerify {
+            storage.removeFile(any())
+        }
+        assertTrue(operationOrder == listOf("stream_closed", "file_removed"),
+            "Expected order: [stream_closed, file_removed], but got: $operationOrder")
+    }
 }