Replace polling with database-tick waits in PlatformAPI #88
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| import Foundation | ||
| import IMDatabase | ||
| import IMessageCore | ||
| import PlatformSDK | ||
|
|
||
| private let sentMessageLinkWaitTimeout: TimeInterval = 1.5 | ||
|
|
||
| // Re-query at least this often even without a tick: FSEvents notifications can be | ||
| // dropped or coalesced, so a missed tick costs ~1s instead of the full timeout. | ||
| private let databaseTickBackstopInterval: TimeInterval = 1.0 | ||
|
|
||
| enum DatabaseTickWaits { | ||
| typealias SentMessageID = (rowID: Int, guid: String) | ||
|
|
||
| static func sentMessageIDs( | ||
| text: String?, | ||
| timeout: TimeInterval, | ||
| changes: Topic<Void>, | ||
| linkTimeout: TimeInterval = sentMessageLinkWaitTimeout, | ||
| backstopInterval: TimeInterval = databaseTickBackstopInterval, | ||
| querySentMessageIDs: @escaping @Sendable () throws -> [SentMessageID] | ||
| ) async throws -> [SentMessageID] { | ||
| let startedAt = Date() | ||
| let timeoutDeadline = startedAt.addingTimeInterval(timeout) | ||
| let linkDeadline = startedAt.addingTimeInterval(linkTimeout) | ||
| let expectedNewMessageIDCount = text.map { max($0.linkCount, 1) } ?? 1 | ||
|
|
||
| while true { | ||
| let changeStream = changes.subscribe() | ||
| let sentMessageIDs = try querySentMessageIDs() | ||
| if sentMessageIDs.count == expectedNewMessageIDCount { | ||
| return sentMessageIDs | ||
| } | ||
| if text != nil, !sentMessageIDs.isEmpty, Date() >= linkDeadline { | ||
| return sentMessageIDs | ||
| } | ||
| if Date() >= timeoutDeadline { | ||
| throw ErrorMessage("timed out waiting for sent messages") | ||
| } | ||
|
|
||
| let wakeDeadline: Date | ||
| if text != nil, !sentMessageIDs.isEmpty { | ||
| wakeDeadline = min(timeoutDeadline, linkDeadline) | ||
| } else { | ||
| wakeDeadline = timeoutDeadline | ||
| } | ||
| _ = try await waitForChange(on: changeStream, until: wakeDeadline, backstopInterval: backstopInterval) | ||
| } | ||
| } | ||
|
|
||
| static func sentThreadIDs( | ||
| timeout: TimeInterval, | ||
| changes: Topic<Void>, | ||
| backstopInterval: TimeInterval = databaseTickBackstopInterval, | ||
| querySentThreadIDs: @escaping @Sendable () throws -> [String?] | ||
| ) async throws -> [String?] { | ||
| let deadline = Date().addingTimeInterval(timeout) | ||
|
|
||
| while true { | ||
| let changeStream = changes.subscribe() | ||
| let threadIDs = try querySentThreadIDs() | ||
| if !threadIDs.contains(where: { $0 == nil }) || Date() >= deadline { | ||
| return threadIDs | ||
| } | ||
|
|
||
| _ = try await waitForChange(on: changeStream, until: deadline, backstopInterval: backstopInterval) | ||
| } | ||
| } | ||
|
|
||
| static func loadedAttachment( | ||
| messageID: String, | ||
| timeout: TimeInterval, | ||
| changes: Topic<Void>, | ||
| backstopInterval: TimeInterval = databaseTickBackstopInterval, | ||
| loadMessage: @escaping @Sendable () async throws -> PlatformSDK.Message?, | ||
| terminalAttachmentFailureState: @escaping @Sendable () async throws -> Attachment.IMFileTransferState? | ||
| ) async throws -> PlatformSDK.Message { | ||
| let deadline = Date().addingTimeInterval(timeout) | ||
| var isFirstRead = true | ||
|
|
||
| while true { | ||
| let changeStream = changes.subscribe() | ||
| let message = try await loadMessage() | ||
| .orThrow(ErrorMessage("Could not find message \(messageID)")) | ||
| let attachments = message.attachments ?? [] | ||
| if isFirstRead { | ||
| guard !attachments.isEmpty else { | ||
| throw ErrorMessage("Message \(messageID) has no attachments") | ||
| } | ||
| isFirstRead = false | ||
| } | ||
| if !attachments.isEmpty, !attachments.contains(where: { $0.loading == true }) { | ||
| return message | ||
| } | ||
|
|
||
| if let failureState = try await terminalAttachmentFailureState() { | ||
| throw ErrorMessage("Attachment in message \(messageID) failed to load (transfer state: \(failureState.rawValue))") | ||
| } | ||
|
|
||
| guard Date() < deadline else { | ||
| throw ErrorMessage("Timed out waiting for attachment in message \(messageID) to load") | ||
| } | ||
|
|
||
| _ = try await waitForChange(on: changeStream, until: deadline, backstopInterval: backstopInterval) | ||
| } | ||
| } | ||
|
Comment on lines
+131
to
+150
Contributor
There was a problem hiding this comment. Subscription leak when query succeeds without waiting. When Over time, these dangling subscriptions accumulate: each Proposed fix: Wrap stream in RAII-style cleanupIntroduce a small wrapper that ensures the stream is consumed/cancelled on scope exit: + private struct ScopedSubscription {
+ let stream: AsyncStream<Void>
+ private var iterator: AsyncStream<Void>.AsyncIterator?
+
+ init(_ stream: AsyncStream<Void>) {
+ self.stream = stream
+ }
+
+ mutating func consume() async {
+ if iterator == nil {
+ iterator = stream.makeAsyncIterator()
+ }
+ _ = await iterator?.next()
+ }
+ }
+
private static func waitForDatabaseResult<T>(
changes: Topic<Void>,
backstopInterval: TimeInterval,
query: `@escaping` `@Sendable` () async throws -> T,
evaluate: (T) async throws -> WaitResult<T>
) async throws -> T {
while true {
- let changeStream = changes.subscribe()
+ var subscription = ScopedSubscription(changes.subscribe())
+ defer {
+ // Start iteration so onTermination fires when scope exits
+ Task { [subscription] in
+ var sub = subscription
+ _ = await sub.stream.makeAsyncIterator().next()
+ }
+ }
let result = try await query()
switch try await evaluate(result) {
case let .finished(value):
return value
case let .waitingUntil(deadline):
- try await waitForChange(on: changeStream, until: deadline, backstopInterval: backstopInterval)
+ await subscription.consume()
+ try await waitForChange(on: subscription.stream, until: deadline, backstopInterval: backstopInterval)
}
}
}Alternatively, add explicit unsubscribe support to 🤖 Prompt for AI Agents |
||
|
|
||
| private static func waitForChange(on stream: AsyncStream<Void>, until deadline: Date, backstopInterval: TimeInterval) async throws -> Bool { | ||
| let remainingTime = deadline.timeIntervalSinceNow | ||
| guard remainingTime > 0 else { return false } | ||
|
|
||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| let sleepTime = min(remainingTime, backstopInterval) | ||
|
|
||
| return try await withThrowingTaskGroup(of: Bool.self) { group in | ||
| group.addTask { | ||
| var iterator = stream.makeAsyncIterator() | ||
| return await iterator.next() != nil | ||
| } | ||
| group.addTask { | ||
| try await Task.sleep(forTimeInterval: sleepTime) | ||
| return false | ||
| } | ||
|
|
||
| do { | ||
| let changed = try await group.next() ?? false | ||
| group.cancelAll() | ||
| return changed | ||
| } catch { | ||
| group.cancelAll() | ||
| throw error | ||
| } | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.