diff --git a/.changeset/durable-queue-lifecycle.md b/.changeset/durable-queue-lifecycle.md
new file mode 100644
index 000000000..aa338be5d
--- /dev/null
+++ b/.changeset/durable-queue-lifecycle.md
@@ -0,0 +1,11 @@
+---
+"posthog-ios": minor
+---
+
+- Preserve bounded durable event, replay, and log queues across retryable upload failures instead of clearing them.
+- Limit `maxRetries` to push-subscription registration; it no longer bounds event, replay, or log queue flush attempts.
+- Acknowledge successful and terminal uploads by their exact persisted entry identities so full-queue replacements accepted during an upload are not deleted.
+- Trim persisted queues to `maxQueueSize` (or `logs.maxBufferSize` for logs) in FIFO order when loading them from disk.
+- Honor the received HTTP status and `Retry-After` when an upload also returns a transport error: successful and terminal responses remove the sent entries, while retryable responses retain them. A `3xx` left on a failed request is reported as no status, because a redirect does not confirm delivery to the final host.
+- Preserve existing queued records when writing a new record to a full queue fails.
+- Retry push-subscription requests that answer with HTTP 408 instead of treating the timeout as terminal, so a pending logout unregister survives until it succeeds.
diff --git a/PostHog/PostHogApi.swift b/PostHog/PostHogApi.swift
index 3681d93ae..7b3cd2ef2 100644
--- a/PostHog/PostHogApi.swift
+++ b/PostHog/PostHogApi.swift
@@ -10,19 +10,30 @@ import Foundation
 /// Common URLSession upload-response handler shared by `/batch`, `/snapshot`,
 /// and `/i/v1/logs`. Routes through `as?` so a missing HTTP response can't
 /// crash inside a customer process.
-private func processUploadResponse(
+func processUploadResponse(
     endpointName: String,
     data: Data?,
     response: URLResponse?,
     error: Error?,
     completion: @escaping (PostHogUploadInfo) -> Void
 ) {
+    let httpResponse = response as? HTTPURLResponse
+    // Parsed before the error branch: URLSession can deliver headers and then fail the
+    // transfer, and a rate-limited response still carries the delay the server asked for.
+    let retryAfter = httpResponse.flatMap { $0.value(forHTTPHeaderField: "Retry-After") }.flatMap(parseRetryAfter)
+
     if let error {
         hedgeLog("Error calling the \(endpointName) API: \(error).")
-        return completion(PostHogUploadInfo(statusCode: nil, error: error))
+        // A 3xx left on a failed task is the redirect URLSession was still following, not an
+        // outcome for the payload, so it's reported as no status. Honoring it would let the
+        // policies that treat 3xx as terminal (logs, push unregister) delete durable records
+        // that never reached the final host.
+        let status = httpResponse?.statusCode
+        let delivered = status.flatMap { 300 ... 399 ~= $0 ? nil : $0 }
+        return completion(PostHogUploadInfo(statusCode: delivered, error: error, retryAfter: retryAfter))
     }
 
-    guard let httpResponse = response as? HTTPURLResponse else {
+    guard let httpResponse else {
         hedgeLog("\(endpointName) API returned no HTTP response")
         return completion(PostHogUploadInfo(statusCode: nil, error: nil))
     }
@@ -34,7 +45,6 @@ private func processUploadResponse(
         hedgeLog("\(endpointName) sent successfully.")
     }
 
-    let retryAfter = httpResponse.value(forHTTPHeaderField: "Retry-After").flatMap(parseRetryAfter)
     completion(PostHogUploadInfo(statusCode: httpResponse.statusCode, error: nil, retryAfter: retryAfter))
 }
 
diff --git a/PostHog/PostHogConfig.swift b/PostHog/PostHogConfig.swift
index c32745a8a..92f463f73 100644
--- a/PostHog/PostHogConfig.swift
+++ b/PostHog/PostHogConfig.swift
@@ -89,10 +89,9 @@ public typealias BeforeSendBlock = (PostHogEvent) -> PostHogEvent?
     /// Default: `30`.
     @objc public var flushIntervalSeconds: TimeInterval = Defaults.flushIntervalSeconds
 
-    /// Maximum number of consecutive flush attempts before the entire queue is
-    /// dropped to avoid infinite retries against a permanently-broken backend.
-    /// Increments on every retriable failure including HTTP 413 cap halving;
-    /// resets on a successful 2xx response. Default 3.
+    /// Maximum number of retries for push-subscription registration failures.
+    /// Durable event, replay, and log queues retain retryable records and use
+    /// their configured capacity as the storage bound. Default: `3`.
     @objc public var maxRetries: Int = Defaults.maxRetries
 
     /// Maximum number of retries for feature flag requests after transient network errors or retryable HTTP responses.
diff --git a/PostHog/PostHogFileBackedQueue.swift b/PostHog/PostHogFileBackedQueue.swift
index bbad36801..a19e9fbad 100644
--- a/PostHog/PostHogFileBackedQueue.swift
+++ b/PostHog/PostHogFileBackedQueue.swift
@@ -8,7 +8,13 @@
 import Foundation
 
 class PostHogFileBackedQueue {
+    struct Entry {
+        let id: String
+        let data: Data
+    }
+
     let queue: URL
+    private let maxSize: Int?
     private var items = [String]()
     private let itemsLock = NSLock()
 
@@ -16,8 +22,9 @@ class PostHogFileBackedQueue {
         itemsLock.withLock { items.count }
     }
 
-    init(queue: URL, oldQueues: [URL] = []) {
+    init(queue: URL, oldQueues: [URL] = [], maxSize: Int? = nil) {
         self.queue = queue
+        self.maxSize = maxSize.map { max(1, $0) }
         setup(oldQueues: oldQueues)
     }
 
@@ -42,9 +49,7 @@ class PostHogFileBackedQueue {
         }
 
         do {
-            // when copying over buffered snapshots, content modification date will change, so we work off creation date instead.
-            let sortedItems = try FileManager.default.contentsOfDirectory(at: queue, sortedBy: .creationDateKey)
-            itemsLock.withLock { items = sortedItems }
+            try reindexFromDisk()
         } catch {
             hedgeLog("Failed to load files for queue \(error)")
             // failed to read directory – bad permissions, perhaps?
@@ -52,7 +57,11 @@ class PostHogFileBackedQueue {
     }
 
     func peek(_ count: Int) -> [Data] {
-        loadFiles(count)
+        peekEntries(count).map(\.data)
+    }
+
+    func peekEntries(_ count: Int) -> [Entry] {
+        loadEntries(count)
     }
 
     func delete(index: Int) {
@@ -70,13 +79,46 @@ class PostHogFileBackedQueue {
         deleteFiles(count)
     }
 
-    func add(_ contents: Data) {
+    func remove(ids: [String]) {
+        let ids = Set(ids)
+        let removed: [String] = itemsLock.withLock {
+            let removed = items.filter { ids.contains($0) }
+            items.removeAll { ids.contains($0) }
+            return removed
+        }
+
+        for item in removed {
+            deleteSafely(queue.appendingPathComponent(item))
+        }
+    }
+
+    /// Persists one entry and optionally enforces a FIFO capacity in the same
+    /// critical section. Returning an evicted id lets the queue report
+    /// backpressure without racing a separate depth check against other adds.
+    @discardableResult
+    func add(_ contents: Data, maxSize: Int? = nil) -> String? {
         do {
             let filename = UUID.v7String()
-            try contents.write(to: queue.appendingPathComponent(filename))
-            itemsLock.withLock { items.append(filename) }
+            let effectiveMaxSize = maxSize.map { max(1, $0) } ?? self.maxSize
+            var evicted: String?
+
+            try itemsLock.withLock {
+                try contents.write(to: queue.appendingPathComponent(filename))
+
+                if let effectiveMaxSize, items.count >= effectiveMaxSize {
+                    evicted = items.removeFirst()
+                    if let evicted {
+                        deleteSafely(queue.appendingPathComponent(evicted))
+                    }
+                }
+
+                items.append(filename)
+            }
+
+            return evicted
         } catch {
             hedgeLog("Could not write file \(error)")
+            return nil
         }
     }
 
@@ -90,15 +132,35 @@ class PostHogFileBackedQueue {
     /// Use after externally adding files to the queue directory.
     func reloadFromDisk() {
         do {
-            let sortedItems = try FileManager.default.contentsOfDirectory(at: queue, sortedBy: .creationDateKey)
-            itemsLock.withLock { items = sortedItems }
+            try reindexFromDisk()
         } catch {
             hedgeLog("Failed to reload files for queue \(error)")
         }
     }
 
-    private func loadFiles(_ count: Int) -> [Data] {
-        var results = [Data]()
+    /// Re-reads the queue directory and replaces the in-memory index with it,
+    /// enforcing the FIFO capacity. The enumeration runs inside `itemsLock` so a
+    /// filename appended by a concurrent `add` can't be dropped from the index by
+    /// the replacement while its file stays on disk.
+    private func reindexFromDisk() throws {
+        let dropped: [String] = try itemsLock.withLock {
+            // when copying over buffered snapshots, content modification date will change, so we work off creation date instead.
+            let sortedItems = try FileManager.default.contentsOfDirectory(at: queue, sortedBy: .creationDateKey)
+            let overflow = maxSize.map { max(0, sortedItems.count - $0) } ?? 0
+            items = Array(sortedItems.dropFirst(overflow))
+            return Array(sortedItems.prefix(overflow))
+        }
+
+        for item in dropped {
+            deleteSafely(queue.appendingPathComponent(item))
+        }
+        if !dropped.isEmpty {
+            hedgeLog("Dropped \(dropped.count) oldest cached records to enforce queue capacity")
+        }
+    }
+
+    private func loadEntries(_ count: Int) -> [Entry] {
+        var results = [Entry]()
         var skipped = Set<String>()
 
         let itemsCopy = itemsLock.withLock { items }
@@ -113,7 +175,7 @@ class PostHogFileBackedQueue {
                 }
                 let contents = try Data(contentsOf: itemURL)
 
-                results.append(contents)
+                results.append(Entry(id: item, data: contents))
             } catch {
                 if isTemporarilyUnavailable(error) {
                     hedgeLog("File \(itemURL) is temporarily unavailable, will retry \(error)")
@@ -191,12 +253,18 @@ private func migrateOldQueueFolder(queue: URL, oldQueueFolder: URL) {
 }
 
 private extension FileManager {
-    /// Returns filenames sorted by resource key
+    /// Returns filenames in a total order: by resource key, then by filename.
+    /// `reindexFromDisk` deletes the head of this order to enforce capacity, so a file
+    /// whose date can't be read must not sort oldest, and equal dates need a tie-breaker
+    /// because `sorted` isn't stable. UUID v7 names sort by their embedded timestamp.
     func contentsOfDirectory(at url: URL, sortedBy key: URLResourceKey) throws -> [String] {
         let urls = try contentsOfDirectory(at: url, includingPropertiesForKeys: [key])
-        return urls.sorted {
-            let date1 = (try? $0.resourceValues(forKeys: [key]).allValues[key] as? Date) ?? .distantPast
-            let date2 = (try? $1.resourceValues(forKeys: [key]).allValues[key] as? Date) ?? .distantPast
+        return urls.sorted { lhs, rhs in
+            let date1 = (try? lhs.resourceValues(forKeys: [key]).allValues[key] as? Date) ?? .distantFuture
+            let date2 = (try? rhs.resourceValues(forKeys: [key]).allValues[key] as? Date) ?? .distantFuture
+            if date1 == date2 {
+                return lhs.lastPathComponent < rhs.lastPathComponent
+            }
             return date1 < date2
         }.map(\.lastPathComponent)
     }
diff --git a/PostHog/PostHogQueue.swift b/PostHog/PostHogQueue.swift
index edc82c71f..06baf7a83 100644
--- a/PostHog/PostHogQueue.swift
+++ b/PostHog/PostHogQueue.swift
@@ -118,7 +118,8 @@ class PostHogQueue<Record> {
             rateCapWindowSeconds = endpoint.rateCapWindowSeconds(config)
             fileQueue = PostHogFileBackedQueue(
                 queue: storage.url(forKey: endpoint.storageKey),
-                oldQueues: endpoint.oldStorageKeys.map { storage.url(forKey: $0) }
+                oldQueues: endpoint.oldStorageKeys.map { storage.url(forKey: $0) },
+                maxSize: configuredMaxQueueSize
             )
             dispatchQueue = DispatchQueue(label: endpoint.dispatchQueueLabel, target: .global(qos: .utility))
         }
@@ -133,7 +134,8 @@ class PostHogQueue<Record> {
             rateCapWindowSeconds = endpoint.rateCapWindowSeconds(config)
             fileQueue = PostHogFileBackedQueue(
                 queue: storage.url(forKey: endpoint.storageKey),
-                oldQueues: endpoint.oldStorageKeys.map { storage.url(forKey: $0) }
+                oldQueues: endpoint.oldStorageKeys.map { storage.url(forKey: $0) },
+                maxSize: configuredMaxQueueSize
             )
             dispatchQueue = DispatchQueue(label: endpoint.dispatchQueueLabel, target: .global(qos: .utility))
         }
@@ -147,25 +149,13 @@ class PostHogQueue<Record> {
     }
 
     private func handleResult(_ result: PostHogUploadInfo, _ payload: PostHogConsumerPayload<Record>) {
-        // -1 means its not anything related to the API but rather network or something else, so we try again
+        // A missing response means no terminal delivery outcome is known, so
+        // the exact durable entries must remain eligible for a later flush.
         let statusCode = result.statusCode ?? -1
-
-        // Network error (-1) is universally retriable; everything else is
-        // up to the endpoint's retry policy.
         let isRetriable = statusCode == -1 || endpoint.isRetriableStatusCode(statusCode)
 
         if isRetriable {
-            let newCount = stateLock.withLock { () -> Int in
-                retryCount += 1
-                return retryCount
-            }
-            // `>` not `>=`: maxRetries is the count of retries allowed before
-            // dropping — default 3 retries attempts 1-3, drops on attempt 4.
-            if newCount > config.maxRetries {
-                dropAllQueuedRecords(reason: "max retries (\(config.maxRetries)) exceeded")
-                payload.completion(true)
-                return
-            }
+            let newCount = nextRetryCount()
             let backoffDelay = min(TimeInterval(newCount) * retryDelay, maxRetryDelay)
             let delay = max(backoffDelay, result.retryAfter ?? 0)
             pauseFor(seconds: delay)
@@ -174,27 +164,13 @@ class PostHogQueue<Record> {
             return
         }
 
-        // 413 Payload Too Large. Two paths:
-        //  - cap > 1: this is a retry. Increment retryCount, drop all if
-        //    `maxRetries` exceeded, otherwise halve cap and retry the same
-        //    records.
-        //  - cap == 1: poison drop. The offending record can't shrink any
-        //    further, so we drop the batch and apply the endpoint's poison
-        //    cap policy. Don't count it as a retry — the drop *is* the
-        //    resolution, not another attempt.
+        // A multi-record 413 is resolved by progressively shrinking the batch.
+        // Once a singleton still receives 413, only that poison entry is
+        // acknowledged for removal. Neither path clears unrelated records.
         if statusCode == 413 {
             let canHalve = batchLimitsLock.withLock { batchLimits.cap > 1 && payload.records.count > 1 }
 
             if canHalve {
-                let newCount = stateLock.withLock { () -> Int in
-                    retryCount += 1
-                    return retryCount
-                }
-                if newCount > config.maxRetries {
-                    dropAllQueuedRecords(reason: "max retries (\(config.maxRetries)) exceeded after repeated HTTP 413")
-                    payload.completion(true)
-                    return
-                }
                 let actualBatchSize = payload.records.count
                 let halvedCap = batchLimitsLock.withLock {
                     batchLimits.halve(actualBatchSize: actualBatchSize)
@@ -204,28 +180,29 @@ class PostHogQueue<Record> {
                 return
             }
 
-            // Cap stays at 1 — the offender is gone but we keep being
-            // cautious until a successful send.
-            hedgeLog("Queue: dropping batch after HTTP 413 (cap == 1)")
-            stateLock.withLock { retryCount = 0 }
+            hedgeLog("Queue: dropping singleton batch after HTTP 413")
+            resetRetryState()
             payload.completion(true)
             return
         }
 
-        // 2xx success or non-retriable 4xx (auth, malformed, etc.): pop the
-        // batch. Cap stays where it is — no ramp on success.
-        stateLock.withLock { retryCount = 0 }
+        // 2xx success or a terminal response removes the exact snapshotted
+        // entries. The adaptive cap stays where it is.
+        resetRetryState()
         payload.completion(true)
     }
 
-    /// Drops every queued record from disk and resets the retry / pause state.
-    /// Called when `retryCount` exceeds `config.maxRetries` to avoid retrying
-    /// forever against a permanently-broken backend. Cap is left where it is
-    /// — new records starting against a known-bad backend benefit from the
-    /// conservative cap until proven otherwise.
-    private func dropAllQueuedRecords(reason: String) {
-        hedgeLog("Queue: dropping all queued records — \(reason)")
-        fileQueue.clear()
+    private func nextRetryCount() -> Int {
+        stateLock.withLock {
+            let maximumBackoffStep = max(1, Int(maxRetryDelay / retryDelay))
+            if retryCount < maximumBackoffStep {
+                retryCount += 1
+            }
+            return retryCount
+        }
+    }
+
+    private func resetRetryState() {
         stateLock.withLock {
             retryCount = 0
             pausedUntil = nil
@@ -241,16 +218,7 @@ class PostHogQueue<Record> {
                 // can all receive notifications without overwriting each other.
                 reachableToken = reachability?.onReachable.subscribe { [weak self] reachability in
                     guard let self else { return }
-                    self.stateLock.withLock {
-                        if self.config.dataMode == .wifi, reachability.connection != .wifi {
-                            hedgeLog("Queue is paused because its not in WiFi mode")
-                            self.paused = true
-                        } else {
-                            self.paused = false
-                        }
-                    }
-
-                    if reachability.connection == .wifi {
+                    if self.updatePauseState(for: reachability) {
                         self.flush()
                     }
                 }
@@ -263,10 +231,21 @@ class PostHogQueue<Record> {
                     }
                 }
 
+                if let reachability {
+                    // A shared notifier may already be running, so late queue
+                    // subscribers must synchronously adopt its current state
+                    // rather than waiting for a transition that may never come.
+                    _ = updatePauseState(for: reachability)
+                }
+
                 do {
                     try reachability?.startNotifier()
                 } catch {
+                    // Without a running notifier no callback can ever clear a
+                    // reachability pause, so it must not outlive the failure.
+                    // URLSession errors and the retry backoff cover connectivity.
                     hedgeLog("Error: Unable to monitor network reachability: \(error)")
+                    stateLock.withLock { paused = false }
                 }
             #endif
         }
@@ -372,18 +351,14 @@ class PostHogQueue<Record> {
             return
         }
 
-        if fileQueue.depth >= configuredMaxQueueSize {
-            hedgeLog("Queue is full, dropping oldest record")
-            // first is always oldest
-            fileQueue.delete(index: 0)
-        }
-
         guard let data = endpoint.encode(record) else {
             hedgeLog("Tried to queue unserialisable record")
             return
         }
 
-        fileQueue.add(data)
+        if fileQueue.add(data, maxSize: configuredMaxQueueSize) != nil {
+            hedgeLog("Queue is full, dropping oldest record")
+        }
         hedgeLog("Queued \(endpoint.describe(record)). Depth: \(fileQueue.depth)")
         flushIfOverThreshold()
     }
@@ -455,12 +430,12 @@ class PostHogQueue<Record> {
                 return
             }
 
-            let items = self.fileQueue.peek(count)
+            let entries = self.fileQueue.peekEntries(count)
 
             var processing: [Record] = []
 
-            for item in items {
-                guard let record = self.endpoint.decode(item) else {
+            for entry in entries {
+                guard let record = self.endpoint.decode(entry.data) else {
                     continue
                 }
                 processing.append(record)
@@ -468,8 +443,8 @@ class PostHogQueue<Record> {
 
             completion(PostHogConsumerPayload(records: processing) { [weak self] success in
                 guard let self else { return }
-                if success, items.count > 0 {
-                    self.fileQueue.pop(items.count)
+                if success, !entries.isEmpty {
+                    self.fileQueue.remove(ids: entries.map(\.id))
                     hedgeLog("Completed!")
                 }
 
@@ -480,8 +455,28 @@ class PostHogQueue<Record> {
         }
     }
 
+    #if !os(watchOS)
+        /// Applies the latest shared reachability snapshot and returns whether
+        /// this queue may send under its configured data mode.
+        private func updatePauseState(for reachability: Reachability) -> Bool {
+            let connection = reachability.connection
+            return stateLock.withLock {
+                if connection == .unavailable {
+                    hedgeLog("Queue is paused because network is unreachable")
+                    paused = true
+                } else if config.dataMode == .wifi, connection != .wifi {
+                    hedgeLog("Queue is paused because its not in WiFi mode")
+                    paused = true
+                } else {
+                    paused = false
+                }
+                return !paused
+            }
+        }
+    #endif
+
     private func pauseFor(seconds: TimeInterval) {
-        let until = Date().addingTimeInterval(seconds)
+        let until = now().addingTimeInterval(seconds)
         stateLock.withLock { pausedUntil = until }
     }
 
@@ -491,7 +486,7 @@ class PostHogQueue<Record> {
     private func pauseReason() -> String? {
         let (isPaused, until) = stateLock.withLock { (paused, pausedUntil) }
         if isPaused { return "paused due to the reachability check" }
-        if let until, until > Date() { return "paused until `\(until)`" }
+        if let until, until > now() { return "paused until `\(until)`" }
         return nil
     }
 
@@ -513,5 +508,9 @@ class PostHogQueue<Record> {
         var currentFlushAtForTesting: Int {
             batchLimitsLock.withLock { batchLimits.flushAt }
         }
+
+        var currentRetryCountForTesting: Int {
+            stateLock.withLock { retryCount }
+        }
     }
 #endif
diff --git a/PostHog/PushNotifications/PostHogPushSubscriptionHandler.swift b/PostHog/PushNotifications/PostHogPushSubscriptionHandler.swift
index 607e7b4c0..9a9fb3ff3 100644
--- a/PostHog/PushNotifications/PostHogPushSubscriptionHandler.swift
+++ b/PostHog/PushNotifications/PostHogPushSubscriptionHandler.swift
@@ -830,9 +830,11 @@ final class PostHogPushSubscriptionHandler {
         storage.remove(key: .pushPendingUnregister)
     }
 
-    /// Transport error (no status), 429, or 5xx is retryable; everything else (4xx) is terminal.
+    /// Transport error (no status), 408, 429, or 5xx is retryable; everything else (4xx) is terminal.
+    /// 408 matches the queues' policy in `QueueEndpoint+Factories`: a request timeout can arrive as a
+    /// status alongside a URLSession error, and treating it as terminal drops the unregister intent.
     private func isRetryable(_ info: PostHogUploadInfo) -> Bool {
-        info.statusCode.map { $0 == 429 || (500 ... 599 ~= $0) } ?? true
+        info.statusCode.map { $0 == 408 || $0 == 429 || (500 ... 599 ~= $0) } ?? true
     }
 
     private func statusString(_ info: PostHogUploadInfo) -> String {
diff --git a/PostHogTests/PostHogApiTest.swift b/PostHogTests/PostHogApiTest.swift
index 891708b01..603999684 100644
--- a/PostHogTests/PostHogApiTest.swift
+++ b/PostHogTests/PostHogApiTest.swift
@@ -701,6 +701,60 @@ enum PostHogApiTests {
         }
     }
 
+    @Suite("Upload response handling", .serialized)
+    final class TestUploadResponseHandling {
+        @Test("preserves an HTTP status when URLSession also returns an error", arguments: [200, 400, 408, 429, 503])
+        func preservesHTTPStatusAlongsideError(statusCode: Int) throws {
+            let url = try #require(URL(string: "http://localhost/batch"))
+            let httpResponse = try #require(HTTPURLResponse(url: url, statusCode: statusCode, httpVersion: nil, headerFields: nil))
+            let error = URLError(.timedOut)
+            var uploadInfo: PostHogUploadInfo?
+
+            processUploadResponse(endpointName: "batch", data: nil, response: httpResponse, error: error) {
+                uploadInfo = $0
+            }
+
+            let result = try #require(uploadInfo)
+            #expect(result.statusCode == statusCode)
+            #expect((result.error as? URLError)?.code == .timedOut)
+        }
+
+        @Test("reports no status for a redirect left on a failed task", arguments: [301, 302, 307, 308])
+        func dropsRedirectStatusAlongsideError(statusCode: Int) throws {
+            let url = try #require(URL(string: "http://localhost/i/v1/logs"))
+            let httpResponse = try #require(HTTPURLResponse(url: url, statusCode: statusCode, httpVersion: nil, headerFields: ["Retry-After": "30"]))
+            let error = URLError(.httpTooManyRedirects)
+            var uploadInfo: PostHogUploadInfo?
+
+            processUploadResponse(endpointName: "logs", data: nil, response: httpResponse, error: error) {
+                uploadInfo = $0
+            }
+
+            let result = try #require(uploadInfo)
+            // no status keeps the records retryable for the logs and push-unregister
+            // policies, which classify 3xx as terminal
+            #expect(result.statusCode == nil)
+            #expect(result.retryAfter == 30)
+            #expect((result.error as? URLError)?.code == .httpTooManyRedirects)
+        }
+
+        @Test("preserves Retry-After when URLSession also returns an error")
+        func preservesRetryAfterAlongsideError() throws {
+            let url = try #require(URL(string: "http://localhost/batch"))
+            let httpResponse = try #require(HTTPURLResponse(url: url, statusCode: 429, httpVersion: nil, headerFields: ["Retry-After": "120"]))
+            let error = URLError(.networkConnectionLost)
+            var uploadInfo: PostHogUploadInfo?
+
+            processUploadResponse(endpointName: "batch", data: nil, response: httpResponse, error: error) {
+                uploadInfo = $0
+            }
+
+            let result = try #require(uploadInfo)
+            #expect(result.statusCode == 429)
+            #expect(result.retryAfter == 120)
+        }
+    }
+
     @Suite("Non-HTTPURLResponse handling", .serialized)
     final class TestNonHTTPResponseHandling {
         private func getSut() -> PostHogApi {
diff --git a/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift b/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift
index 6774e02d0..211f87368 100644
--- a/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift
+++ b/PostHogTests/PostHogFileBackedQueueAlignmentTest.swift
@@ -31,6 +31,37 @@ struct PostHogFileBackedQueueAlignmentTest {
         data.map { String(data: $0, encoding: .utf8)! }
     }
 
+    @Test("a failed write to a full queue preserves existing entries", .enabled(if: geteuid() != 0))
+    func failedWritePreservesFullQueue() throws {
+        let (queue, dir) = makeQueue()
+        defer {
+            try? FileManager.default.setAttributes([.posixPermissions: 0o755], ofItemAtPath: dir.path)
+            try? FileManager.default.removeItem(at: dir)
+        }
+
+        queue.add(Data("A".utf8), maxSize: 2)
+        queue.add(Data("B".utf8), maxSize: 2)
+        let originalIds = queue.peekEntries(2).map(\.id)
+        try FileManager.default.setAttributes([.posixPermissions: 0o555], ofItemAtPath: dir.path)
+        #expect(throws: (any Error).self) {
+            try Data("probe".utf8).write(to: dir.appendingPathComponent("write-probe"))
+        }
+
+        #expect(queue.add(Data("C".utf8), maxSize: 2) == nil)
+        #expect(queue.depth == 2)
+        #expect(queue.peekEntries(2).map(\.id) == originalIds)
+        #expect(decode(queue.peek(2)) == ["A", "B"])
+        #expect(Set(try FileManager.default.contentsOfDirectory(atPath: dir.path)) == Set(originalIds))
+
+        try FileManager.default.setAttributes([.posixPermissions: 0o755], ofItemAtPath: dir.path)
+        #expect(queue.add(Data("C".utf8), maxSize: 2) == originalIds.first)
+        #expect(queue.depth == 2)
+        #expect(decode(queue.peek(2)) == ["B", "C"])
+        let reloaded = PostHogFileBackedQueue(queue: dir, maxSize: 2)
+        #expect(Set(reloaded.peekEntries(2).map(\.id)) == Set(queue.peekEntries(2).map(\.id)))
+        #expect(reloaded.depth == 2)
+    }
+
     @Test("delivers every record once in FIFO order on the happy path")
     func happyPath() throws {
         let (queue, dir) = makeQueue()
@@ -52,6 +83,30 @@ struct PostHogFileBackedQueueAlignmentTest {
         #expect(queue.depth == 0)
     }
 
+    @Test("breaks creation-date ties by filename when trimming to capacity on load")
+    func tiedCreationDatesTrimDeterministically() throws {
+        let dir = FileManager.default.temporaryDirectory
+            .appendingPathComponent("ph-queue-tie-\(UUID().uuidString)")
+        defer { try? FileManager.default.removeItem(at: dir) }
+        try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true)
+
+        // written out of filename order so an enumeration- or write-order-dependent
+        // sort would disagree with the tie-breaker
+        let tied = Date(timeIntervalSince1970: 100)
+        for name in ["record-4", "record-1", "record-3", "record-2"] {
+            let url = dir.appendingPathComponent(name)
+            try Data(name.utf8).write(to: url)
+            try FileManager.default.setAttributes([.creationDate: tied], ofItemAtPath: url.path)
+        }
+
+        let queue = PostHogFileBackedQueue(queue: dir, maxSize: 2)
+
+        #expect(queue.depth == 2)
+        #expect(decode(queue.peek(2)) == ["record-3", "record-4"])
+        #expect(!FileManager.default.fileExists(atPath: dir.appendingPathComponent("record-1").path))
+        #expect(!FileManager.default.fileExists(atPath: dir.appendingPathComponent("record-2").path))
+    }
+
     enum UnreadableHead {
         case missing // vanished from disk: !fileExists, pruned via `continue`
         case corrupt // present but unreadable (Data(contentsOf:) throws): deleted then pruned
diff --git a/PostHogTests/PostHogFileBackedQueueTest.swift b/PostHogTests/PostHogFileBackedQueueTest.swift
index 4f1ea6573..f4f0fb2c5 100644
--- a/PostHogTests/PostHogFileBackedQueueTest.swift
+++ b/PostHogTests/PostHogFileBackedQueueTest.swift
@@ -97,6 +97,39 @@ class PostHogFileBackedQueueTest: QuickSpec {
             sut.clear()
         }
 
+        it("trims cached files to configured capacity on load") {
+            let baseUrl = applicationSupportDirectoryURL()
+            let newURL = baseUrl.appendingPathComponent("queue")
+            try FileManager.default.createDirectory(atPath: newURL.path, withIntermediateDirectories: true)
+
+            let oldestURL = newURL.appendingPathComponent("cached-oldest")
+            let middleURL = newURL.appendingPathComponent("cached-middle")
+            let newestURL = newURL.appendingPathComponent("cached-newest")
+            let oldestData = Data("cached-0".utf8)
+            let middleData = Data("cached-1".utf8)
+            let newestData = Data("cached-2".utf8)
+
+            for (url, data, creationDate, modificationDate) in [
+                (newestURL, newestData, Date(timeIntervalSince1970: 300), Date(timeIntervalSince1970: 100)),
+                (oldestURL, oldestData, Date(timeIntervalSince1970: 100), Date(timeIntervalSince1970: 300)),
+                (middleURL, middleData, Date(timeIntervalSince1970: 200), Date(timeIntervalSince1970: 200)),
+            ] {
+                try data.write(to: url)
+                try FileManager.default.setAttributes([.modificationDate: modificationDate], ofItemAtPath: url.path)
+                try FileManager.default.setAttributes([.creationDate: creationDate], ofItemAtPath: url.path)
+            }
+
+            let sut = PostHogFileBackedQueue(queue: newURL, maxSize: 2)
+
+            expect(sut.depth) == 2
+            expect(sut.peek(2)) == [middleData, newestData]
+            expect(FileManager.default.fileExists(atPath: oldestURL.path)) == false
+            expect(FileManager.default.fileExists(atPath: middleURL.path)) == true
+            expect(FileManager.default.fileExists(atPath: newestURL.path)) == true
+
+            sut.clear()
+        }
+
         it("delete from queue and disk") {
             let baseUrl = applicationSupportDirectoryURL()
             let newURL = baseUrl.appendingPathComponent("queue")
@@ -139,6 +172,51 @@ class PostHogFileBackedQueueTest: QuickSpec {
             sut.clear()
         }
 
+        it("removes exact stable entry identities") {
+            let sut = self.getSut()
+            let identicalData = self.eventJson.data(using: .utf8)!
+
+            sut.add(identicalData)
+            sut.add(identicalData)
+            sut.add(identicalData)
+
+            let entries = sut.peekEntries(3)
+            expect(Set(entries.map(\.id)).count) == 3
+            expect(entries.map(\.data)) == [identicalData, identicalData, identicalData]
+
+            sut.remove(ids: [entries[0].id, entries[2].id, "missing-entry"])
+
+            let remaining = sut.peekEntries(3)
+            expect(remaining.map(\.id)) == [entries[1].id]
+            expect(sut.depth) == 1
+            expect(FileManager.default.fileExists(atPath: sut.queue.appendingPathComponent(entries[0].id).path)) == false
+            expect(FileManager.default.fileExists(atPath: sut.queue.appendingPathComponent(entries[1].id).path)) == true
+            expect(FileManager.default.fileExists(atPath: sut.queue.appendingPathComponent(entries[2].id).path)) == false
+
+            sut.clear()
+        }
+
+        it("enforces capacity atomically across concurrent adds") {
+            let sut = self.getSut()
+            let capacity = 3
+            let group = DispatchGroup()
+            let producers = DispatchQueue(label: "com.posthog.file-queue-capacity-test", attributes: .concurrent)
+
+            for value in 0 ..< 100 {
+                group.enter()
+                producers.async {
+                    sut.add(Data("event-\(value)".utf8), maxSize: capacity)
+                    group.leave()
+                }
+            }
+
+            expect(group.wait(timeout: .now() + 5)) == .success
+            expect(sut.depth) == capacity
+            expect(try? FileManager.default.contentsOfDirectory(atPath: sut.queue.path).count) == capacity
+
+            sut.clear()
+        }
+
         it("add to queue and disk") {
             let baseUrl = applicationSupportDirectoryURL()
             let newURL = baseUrl.appendingPathComponent("queue")
diff --git a/PostHogTests/PostHogLogsQueueTest.swift b/PostHogTests/PostHogLogsQueueTest.swift
index 46bd055f2..ca8973ced 100644
--- a/PostHogTests/PostHogLogsQueueTest.swift
+++ b/PostHogTests/PostHogLogsQueueTest.swift
@@ -323,22 +323,11 @@ final class PostHogLogsQueueTests {
         #expect(queue.depth == 2)
     }
 
-    @Test("413 poison-drop does not consume the maxRetries budget — queue drains record-by-record")
+    @Test("413 halving reaches singleton poison drops and drains record-by-record")
     func handle413PoisonDropIsNotARetry() async throws {
-        // Pin down: cap=1 + 413 (poison drop) is a *resolution*, not a retry.
-        // It must not increment retryCount.
-        //
-        // Scenario: maxBatchSize=8 + 8 oversized records + default maxRetries=3.
-        // The cap halves 3 times (8→4→2→1, retryCount accumulating to 3) before
-        // reaching cap=1. If poison-drop counted as a retry, the next flush would
-        // push retryCount to 4 > 3 and fire dropAll — wiping ALL 8 records together.
-        // The correct behaviour treats poison-drop as a clean resolution: the
-        // offending record is popped, retryCount resets to 0, cap stays at 1,
-        // and the queue continues draining.
-        //
-        // Observable difference: the buggy path makes ~4 HTTP requests (3
-        // halvings + 1 dropAll). The correct path makes far more — each record
-        // costs at least one halve cycle + one poison drop.
+        // Multi-record batches are retained while the cap halves to one. At
+        // cap one, each 413 removes only that singleton and leaves unrelated
+        // durable records for subsequent flushes.
         let (queue, _) = makeQueue(maxBufferSize: 100, maxBatchSize: 8, flushAt: 8)
         defer { queue.clear()
             queue.stop()
@@ -365,43 +354,6 @@ final class PostHogLogsQueueTests {
         #expect(server.logsRequests.count > 8)
     }
 
-    @Test("drops the entire queue once retryCount exceeds maxRetries on repeated 413")
-    func handle413MaxRetriesDropsAll() async throws {
-        // Mirrors PostHogQueue's safeguard: a permanently-broken backend that
-        // keeps returning 413 should not leave the logs queue retrying forever.
-        // After config.maxRetries failed attempts, the queue drops everything
-        // and resets the retry / cap state.
-        let (queue, config) = makeQueue(maxBufferSize: 100, maxBatchSize: 64)
-        config.maxRetries = 1
-        defer { queue.clear()
-            queue.stop()
-        }
-
-        server.logsResponseHandler = { _, _ in
-            HTTPStubsResponse(jsonObject: ["error": "too large"], statusCode: 413, headers: nil)
-        }
-
-        // Enough records that cap-halving alone won't drain the queue before
-        // retryCount exceeds maxRetries.
-        for i in 0 ..< 32 {
-            queue.add(makeRecord(body: "log-\(i)"))
-        }
-        await waitUntil { queue.depth == 32 }
-
-        // Drive flushes until the queue drops everything via the maxRetries path.
-        // First flush: batchSize=32 → 413 → retryCount=1 (not > 1) → halve cap.
-        // Second flush: batchSize=16 → 413 → retryCount=2 (> 1) → drop ALL records.
-        queue.flush()
-        try? await Task.sleep(nanoseconds: 100_000_000)
-        queue.flush()
-        await waitUntil { queue.depth == 0 }
-        #expect(queue.depth == 0)
-        // Cap stays where it was when dropAll fired — no reset. Matches
-        // events / posthog-android behaviour: new records start at the
-        // halved cap until a successful send proves the backend is healthy.
-        #expect(queue.currentBatchCapForTesting == 16)
-    }
-
     @Test("halves cap from min(cap, actualBatchSize) when queue depth was below cap")
     func handle413HalvesByActualBatchSize() async throws {
         // maxBatchSize=50 but only 4 records on disk. A 413 should halve from
@@ -431,11 +383,6 @@ final class PostHogLogsQueueTests {
         #expect(queue.depth == 4)
     }
 
-    // Note: the 5xx path also goes through the maxRetries check → dropAllQueuedRecords,
-    // but testing it directly would require waiting out the 5+10+15s exponential backoff
-    // between attempts (`pausedUntil` blocks the next flush). The 413 maxRetries test
-    // above covers the shared drop logic without that latency since 413 doesn't pause.
-
     @Test("non-413 4xx drops the batch (poison-pill)")
     func handleNon413_4xxDrops() async throws {
         let (queue, _) = makeQueue(maxBufferSize: 100, maxBatchSize: 100)
@@ -743,10 +690,13 @@ final class PostHogLogsQueueTests {
         reachability.onUnreachable.invoke(reachability)
 
         queue.add(makeRecord(body: "while-offline"))
-        queue.flush()
+        for _ in 0 ..< 3 {
+            queue.flush()
+        }
         try? await Task.sleep(nanoseconds: 300_000_000)
         #expect(server.logsRequests.isEmpty)
         #expect(queue.depth == 1)
+        #expect(queue.currentRetryCountForTesting == 0)
 
         // Simulate WiFi back. The reachable callback unpauses and proactively
         // triggers a flush.
diff --git a/PostHogTests/PostHogQueueTest.swift b/PostHogTests/PostHogQueueTest.swift
index 09541661a..76f5d3dbc 100644
--- a/PostHogTests/PostHogQueueTest.swift
+++ b/PostHogTests/PostHogQueueTest.swift
@@ -11,8 +11,27 @@ import OHHTTPStubs
 import OHHTTPStubsSwift
 @testable import PostHog
 import Quick
+import Testing
 import XCTest
 
+private final class ControlledBatchSender {
+    private let lock = NSLock()
+    private var completions = [(PostHogUploadInfo) -> Void]()
+
+    var requestCount: Int {
+        lock.withLock { completions.count }
+    }
+
+    func send(_: [PostHogEvent], completion: @escaping (PostHogUploadInfo) -> Void) {
+        lock.withLock { completions.append(completion) }
+    }
+
+    func completeRequest(at index: Int, with result: PostHogUploadInfo) {
+        let completion = lock.withLock { completions[index] }
+        completion(result)
+    }
+}
+
 class PostHogQueueTest: QuickSpec {
     func getSut(flushAt: Int = 1, maxQueueSize: Int = 1000, maxBatchSize: Int = 50, maxRetries: Int = 3) -> PostHogQueue<PostHogEvent> {
         let config = PostHogConfig(projectToken: testProjectToken, host: "http://localhost:9001")
@@ -185,113 +204,44 @@ class PostHogQueueTest: QuickSpec {
             sut.clear()
         }
 
-        it("drops the entire queue once retryCount exceeds maxRetries on repeated 413") {
-            // 413 with cap > 1 increments retryCount the same way 5xx /
-            // network errors do — both paths use the same
-            // `newCount > config.maxRetries` check — so this test covers
-            // both paths' drop logic. We use 413 here because it doesn't
-            // set `pausedUntil`, letting the test drive multiple retries
-            // without waiting out the exponential backoff.
-            //
-            // 20 events with maxBatchSize=20 so halving sequence is 10 → 5
-            // → drop — cap doesn't reach 1 before maxRetries=2 is exceeded
-            // on the third attempt; the maxRetries cap fires first instead
-            // of the poison-drop path. Each flush is awaited via
-            // `currentBatchCapForTesting` so the gate inside `take()`
-            // doesn't swallow back-to-back calls.
-            let sut = self.getSut(flushAt: 100, maxBatchSize: 20, maxRetries: 2)
-            server.start(batchCount: 3)
-            server.batchResponseHandler = { _, _ in
-                HTTPStubsResponse(jsonObject: [], statusCode: 413, headers: nil)
-            }
-
-            for i in 0 ..< 20 {
-                sut.add(PostHogEvent(event: "evt\(i)", distinctId: "id"))
-            }
-
-            sut.flush()
-            expect(sut.currentBatchCapForTesting).toEventually(equal(10))
-            sut.flush()
-            expect(sut.currentBatchCapForTesting).toEventually(equal(5))
-            sut.flush()
-            expect(sut.depth).toEventually(equal(0))
-
-            sut.clear()
-        }
-
-        it("maxRetries drop wipes the entire queue, not just the current batch") {
-            // Multiple events queued. After enough 413s to trip maxRetries,
-            // the drop must clear ALL of them, not just whatever batch was
-            // in flight.
-            let sut = self.getSut(flushAt: 100, maxBatchSize: 50, maxRetries: 1)
-            server.start(batchCount: 2)
+        it("retains batch on retriable 5xx and does not change cap") {
+            let sut = self.getSut(flushAt: 2, maxBatchSize: 4)
             server.batchResponseHandler = { _, _ in
-                HTTPStubsResponse(jsonObject: [], statusCode: 413, headers: nil)
+                HTTPStubsResponse(jsonObject: [], statusCode: 500, headers: nil)
             }
 
-            for i in 0 ..< 5 {
-                sut.add(PostHogEvent(event: "event\(i)", distinctId: "id\(i)"))
-            }
+            sut.add(PostHogEvent(event: "event1", distinctId: "id1"))
+            sut.add(PostHogEvent(event: "event2", distinctId: "id2"))
 
-            // First flush: batch=5 → 413 → retryCount=1 (not > 1) → halve.
-            sut.flush()
-            expect(sut.currentBatchCapForTesting).toEventually(equal(2))
-            expect(sut.depth) == 5
+            _ = getBatchedEvents(server)
 
-            // Second flush: batch=2 → 413 → retryCount=2 (> 1) → drop ALL,
-            // not just the 2 in this batch.
-            sut.flush()
-            expect(sut.depth).toEventually(equal(0))
+            expect(sut.depth).toEventually(equal(2))
+            expect(sut.currentBatchCapForTesting).toEventually(equal(4))
 
             sut.clear()
         }
 
-        it("queue keeps working after a maxRetries drop — retryCount is reset") {
-            // After events get dropped the queue must continue to accept and
-            // flush new ones; otherwise the SDK is permanently broken until
-            // the host app restarts.
-            let sut = self.getSut(flushAt: 100, maxBatchSize: 50, maxRetries: 1)
-            var attempt = 0
-            server.start(batchCount: 3)
+        it("retains batch on HTTP 429 and does not change cap") {
+            let sut = self.getSut(flushAt: 2, maxBatchSize: 4)
             server.batchResponseHandler = { _, _ in
-                attempt += 1
-                // First two attempts fail with 413 → triggers maxRetries drop.
-                // Third attempt (the post-drop add) succeeds.
-                return attempt <= 2
-                    ? HTTPStubsResponse(jsonObject: [], statusCode: 413, headers: nil)
-                    : HTTPStubsResponse(jsonObject: ["status": "ok"], statusCode: 200, headers: nil)
-            }
-
-            for i in 0 ..< 5 {
-                sut.add(PostHogEvent(event: "doomed\(i)", distinctId: "id\(i)"))
+                HTTPStubsResponse(jsonObject: [], statusCode: 429, headers: nil)
             }
 
-            sut.flush()
-            expect(sut.currentBatchCapForTesting).toEventually(equal(2))
-            sut.flush()
-            expect(sut.depth).toEventually(equal(0))
+            sut.add(PostHogEvent(event: "event1", distinctId: "id1"))
+            sut.add(PostHogEvent(event: "event2", distinctId: "id2"))
 
-            // dropAll never resets the adaptive cap — it stays where the
-            // last 413 left it, for both events and logs. New records start
-            // against the conservative cap until a successful send proves
-            // the backend is healthy.
-            expect(sut.currentBatchCapForTesting) == 2
+            _ = getBatchedEvents(server)
 
-            // New event after the drop should flush successfully — retryCount
-            // and pausedUntil were reset by dropAllQueuedEvents.
-            sut.add(PostHogEvent(event: "after-drop", distinctId: "id"))
-            sut.flush()
-            let events = getBatchedEvents(server)
-            expect(events.contains(where: { $0.event == "after-drop" })) == true
-            expect(sut.depth).toEventually(equal(0))
+            expect(sut.depth).toEventually(equal(2))
+            expect(sut.currentBatchCapForTesting).toEventually(equal(4))
 
             sut.clear()
         }
 
-        it("retains batch on retriable 5xx and does not change cap") {
+        it("retains batch on HTTP 408 (request timeout is retriable) and does not change cap") {
             let sut = self.getSut(flushAt: 2, maxBatchSize: 4)
             server.batchResponseHandler = { _, _ in
-                HTTPStubsResponse(jsonObject: [], statusCode: 500, headers: nil)
+                HTTPStubsResponse(jsonObject: [], statusCode: 408, headers: nil)
             }
 
             sut.add(PostHogEvent(event: "event1", distinctId: "id1"))
@@ -305,44 +255,138 @@ class PostHogQueueTest: QuickSpec {
             sut.clear()
         }
 
-        it("retains batch on HTTP 429 and does not change cap") {
-            let sut = self.getSut(flushAt: 2, maxBatchSize: 4)
-            server.batchResponseHandler = { _, _ in
-                HTTPStubsResponse(jsonObject: [], statusCode: 429, headers: nil)
+        it("retains retryable HTTP failures past maxRetries and drains after recovery") {
+            let mockNow = MockDate()
+            now = { mockNow.date }
+            defer { now = { Date() } }
+
+            let sut = self.getSut(flushAt: 100, maxBatchSize: 4, maxRetries: 1)
+            server.start(batchCount: 6)
+            server.batchResponseHandler = { _, requestNumber in
+                requestNumber <= 3 || requestNumber == 5
+                    ? HTTPStubsResponse(jsonObject: [], statusCode: 503, headers: nil)
+                    : HTTPStubsResponse(jsonObject: ["status": "ok"], statusCode: 200, headers: nil)
             }
 
             sut.add(PostHogEvent(event: "event1", distinctId: "id1"))
             sut.add(PostHogEvent(event: "event2", distinctId: "id2"))
 
-            _ = getBatchedEvents(server)
+            for expectedAttempt in 1 ... 3 {
+                sut.flush()
+                expect(server.batchRequests.count).toEventually(equal(expectedAttempt))
+                expect(sut.currentRetryCountForTesting).toEventually(equal(expectedAttempt))
+                expect(sut.depth) == 2
+                mockNow.date.addTimeInterval(60)
+            }
 
-            expect(sut.depth).toEventually(equal(2))
-            expect(sut.currentBatchCapForTesting).toEventually(equal(4))
+            sut.flush()
+            expect(server.batchRequests.count).toEventually(equal(4))
+            expect(sut.depth).toEventually(equal(0))
+            expect(sut.currentRetryCountForTesting).toEventually(equal(0))
+
+            sut.add(PostHogEvent(event: "fresh", distinctId: "id3"))
+            sut.flush()
+            expect(server.batchRequests.count).toEventually(equal(5))
+            expect(sut.currentRetryCountForTesting).toEventually(equal(1))
+            expect(sut.depth) == 1
+            sut.flush()
+            expect(server.batchRequests.count) == 5
+            mockNow.date.addTimeInterval(1)
+            sut.flush()
+            expect(server.batchRequests.count).toEventually(equal(6))
+            expect(sut.depth).toEventually(equal(0))
+            expect(sut.currentRetryCountForTesting).toEventually(equal(0))
 
             sut.clear()
         }
 
-        it("retains batch on HTTP 408 (request timeout is retriable) and does not change cap") {
-            let sut = self.getSut(flushAt: 2, maxBatchSize: 4)
-            server.batchResponseHandler = { _, _ in
-                HTTPStubsResponse(jsonObject: [], statusCode: 408, headers: nil)
+        it("retains transport failures past maxRetries and drains after recovery") {
+            let mockNow = MockDate()
+            now = { mockNow.date }
+            defer { now = { Date() } }
+
+            let sut = self.getSut(flushAt: 100, maxBatchSize: 4, maxRetries: 0)
+            let networkError = NSError(domain: NSURLErrorDomain, code: NSURLErrorNetworkConnectionLost, userInfo: nil)
+            server.start(batchCount: 4)
+            server.batchResponseHandler = { _, requestNumber in
+                requestNumber <= 3
+                    ? HTTPStubsResponse(error: networkError)
+                    : HTTPStubsResponse(jsonObject: ["status": "ok"], statusCode: 200, headers: nil)
             }
 
             sut.add(PostHogEvent(event: "event1", distinctId: "id1"))
             sut.add(PostHogEvent(event: "event2", distinctId: "id2"))
 
-            _ = getBatchedEvents(server)
+            for expectedAttempt in 1 ... 3 {
+                sut.flush()
+                expect(server.batchRequests.count).toEventually(equal(expectedAttempt))
+                expect(sut.currentRetryCountForTesting).toEventually(equal(expectedAttempt))
+                expect(sut.depth) == 2
+                mockNow.date.addTimeInterval(60)
+            }
 
-            expect(sut.depth).toEventually(equal(2))
-            expect(sut.currentBatchCapForTesting).toEventually(equal(4))
+            sut.flush()
+            expect(server.batchRequests.count).toEventually(equal(4))
+            expect(sut.depth).toEventually(equal(0))
 
             sut.clear()
         }
 
+        it("late success removes exact in-flight identities after full-capacity replacement") {
+            let config = PostHogConfig(projectToken: "queue_identity_\(UUID().uuidString)", host: "http://localhost:9001")
+            config.flushAt = 100
+            config.maxQueueSize = 2
+            config.maxBatchSize = 2
+            let storage = PostHogStorage(config)
+            let sender = ControlledBatchSender()
+            let endpoint = QueueEndpoint<PostHogEvent>(
+                storageKey: .queue,
+                oldStorageKeys: [],
+                dispatchQueueLabel: "com.posthog.Queue.IdentityTest",
+                initialCap: { $0.maxBatchSize },
+                initialFlushAt: { $0.flushAt },
+                maxQueueSize: { $0.maxQueueSize },
+                flushIntervalSeconds: { $0.flushIntervalSeconds },
+                rateCapMax: { _ in 0 },
+                rateCapWindowSeconds: { _ in 0 },
+                encode: { toJSONData($0.toJSON()) },
+                decode: { PostHogEvent.fromJSON($0) },
+                describe: { $0.event },
+                send: sender.send,
+                isRetriableStatusCode: { _ in false }
+            )
+            let sut = PostHogQueue(config, storage, endpoint, nil)
+            defer { sut.clear() }
+            let identicalEvent = PostHogEvent(event: "identical", distinctId: "same-id")
+
+            sut.add(identicalEvent)
+            sut.add(identicalEvent)
+            let inFlightIds = sut.fileQueue.peekEntries(2).map(\.id)
+
+            sut.flush()
+            expect(sender.requestCount).toEventually(equal(1))
+
+            // Replace the entire in-flight batch with byte-identical payloads.
+            sut.add(identicalEvent)
+            sut.add(identicalEvent)
+            let replacementIds = sut.fileQueue.peekEntries(2).map(\.id)
+            expect(Set(inFlightIds).isDisjoint(with: replacementIds)) == true
+
+            sender.completeRequest(at: 0, with: PostHogUploadInfo(statusCode: 200, error: nil))
+
+            expect(sut.depth).toEventually(equal(2))
+            expect(sut.fileQueue.peekEntries(2).map(\.id)) == replacementIds
+
+            sut.flush()
+            expect(sender.requestCount).toEventually(equal(2))
+            sender.completeRequest(at: 1, with: PostHogUploadInfo(statusCode: 200, error: nil))
+            expect(sut.depth).toEventually(equal(0))
+        }
+
         it("halves cap repeatedly across multiple 413s and drops once cap reaches 1") {
             // flushAt is high so add() doesn't trigger an auto-flush — we drive
             // each flush manually to observe the multi-step halving sequence.
-            let sut = self.getSut(flushAt: 100, maxBatchSize: 4)
+            let sut = self.getSut(flushAt: 100, maxBatchSize: 4, maxRetries: 0)
             server.start(batchCount: 3)
             server.batchResponseHandler = { _, _ in
                 HTTPStubsResponse(jsonObject: [], statusCode: 413, headers: nil)
@@ -442,3 +486,103 @@ class PostHogQueueTest: QuickSpec {
         }
     }
 }
+
+@Suite("PostHog queue upload disposition", .serialized, .resetsGlobalState)
+struct PostHogQueueUploadDispositionTest {
+    private func makeQueue(snapshot: Bool, sender: ControlledBatchSender) -> PostHogQueue<PostHogEvent> {
+        let config = PostHogConfig(projectToken: "queue_disposition_\(UUID().uuidString)", host: "http://localhost:9001")
+        config.flushAt = 100
+        config.maxBatchSize = 4
+        config.maxRetries = 0
+        let api = PostHogApi(config)
+        let base: QueueEndpoint<PostHogEvent> = snapshot ? .snapshot(api: api) : .batch(api: api)
+        let endpoint = QueueEndpoint<PostHogEvent>(
+            storageKey: base.storageKey,
+            oldStorageKeys: [],
+            dispatchQueueLabel: base.dispatchQueueLabel,
+            initialCap: base.initialCap,
+            initialFlushAt: base.initialFlushAt,
+            maxQueueSize: base.maxQueueSize,
+            flushIntervalSeconds: base.flushIntervalSeconds,
+            rateCapMax: base.rateCapMax,
+            rateCapWindowSeconds: base.rateCapWindowSeconds,
+            encode: base.encode,
+            decode: base.decode,
+            describe: base.describe,
+            send: sender.send,
+            isRetriableStatusCode: base.isRetriableStatusCode
+        )
+        return PostHogQueue(config, PostHogStorage(config), endpoint, nil)
+    }
+
+    private func waitForRequest(_ count: Int, sender: ControlledBatchSender) async throws {
+        await waitUntil { sender.requestCount == count }
+        try #require(sender.requestCount == count)
+    }
+
+    @Test("received HTTP disposition wins over an accompanying transport error", arguments: [-1, 200, 400, 408, 429, 503], [false, true])
+    func receivedHTTPDisposition(statusCode: Int, snapshot: Bool) async throws {
+        let sender = ControlledBatchSender()
+        let queue = makeQueue(snapshot: snapshot, sender: sender)
+        defer { queue.clear() }
+        queue.add(PostHogEvent(event: "sent", distinctId: "id"))
+        let sentIds = queue.fileQueue.peekEntries(1).map(\.id)
+        queue.flush()
+        try await waitForRequest(1, sender: sender)
+        queue.add(PostHogEvent(event: "not-sent", distinctId: "id"))
+        let allIds = queue.fileQueue.peekEntries(2).map(\.id)
+        let response = statusCode == -1 ? nil : HTTPURLResponse(
+            url: try #require(URL(string: "http://localhost/batch")),
+            statusCode: statusCode, httpVersion: nil, headerFields: nil
+        )
+        processUploadResponse(endpointName: "test", data: nil, response: response, error: URLError(.networkConnectionLost)) {
+            sender.completeRequest(at: 0, with: $0)
+        }
+
+        let retryable = [-1, 408, 429, 503].contains(statusCode)
+        let expectedIds = retryable ? allIds : allIds.filter { !sentIds.contains($0) }
+        #expect(queue.fileQueue.peekEntries(2).map(\.id) == expectedIds)
+        #expect(queue.currentRetryCountForTesting == (retryable ? 1 : 0))
+        let reloaded = PostHogFileBackedQueue(queue: queue.fileQueue.queue)
+        #expect(Set(reloaded.peekEntries(2).map(\.id)) == Set(expectedIds))
+    }
+
+    @Test("413 reaches singleton after retryable failures and preserves later records", arguments: [false, true])
+    func shrinkingAfterRetryableFailures(snapshot: Bool) async throws {
+        let mockNow = MockDate()
+        now = { mockNow.date }
+        defer { now = { Date() } }
+        let sender = ControlledBatchSender()
+        let queue = makeQueue(snapshot: snapshot, sender: sender)
+        defer { queue.clear() }
+        for name in ["poison", "later-1", "later-2", "later-3"] {
+            queue.add(PostHogEvent(event: name, distinctId: "id"))
+        }
+        let originalIds = queue.fileQueue.peekEntries(4).map(\.id)
+        for attempt in 0 ..< 3 {
+            queue.flush()
+            try await waitForRequest(attempt + 1, sender: sender)
+            sender.completeRequest(at: attempt, with: PostHogUploadInfo(statusCode: 503, error: nil))
+            #expect(queue.fileQueue.peekEntries(4).map(\.id) == originalIds)
+            mockNow.date.addTimeInterval(60)
+        }
+        for (attempt, cap) in [(3, 2), (4, 1)] {
+            queue.flush()
+            try await waitForRequest(attempt + 1, sender: sender)
+            sender.completeRequest(at: attempt, with: PostHogUploadInfo(statusCode: 413, error: nil))
+            #expect(queue.currentBatchCapForTesting == cap)
+            #expect(queue.fileQueue.peekEntries(4).map(\.id) == originalIds)
+        }
+        queue.flush()
+        try await waitForRequest(6, sender: sender)
+        sender.completeRequest(at: 5, with: PostHogUploadInfo(statusCode: 413, error: nil))
+        #expect(queue.fileQueue.peekEntries(4).map(\.id) == Array(originalIds.dropFirst()))
+        #expect(queue.currentRetryCountForTesting == 0)
+        for attempt in 6 ..< 9 {
+            queue.flush()
+            try await waitForRequest(attempt + 1, sender: sender)
+            sender.completeRequest(at: attempt, with: PostHogUploadInfo(statusCode: 200, error: nil))
+        }
+        #expect(queue.depth == 0)
+    }
+}