Skip to content

Refactor the JobRunner to use actors and be async/await #538

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mpretty-cyro merged 27 commits into from
Feb 2, 2026
Merged

Refactor the JobRunner to use actors and be async/await #538

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
f9b8ece
WIP of refactoring the JobRunner to be async/await
mpretty-cyro Aug 29, 2025
6e4d3dd
Refactored a couple of jobs
mpretty-cyro Aug 29, 2025
37e1110
Updating a bunch of jobs to conform to new structure, awaitResult tweaks
mpretty-cyro Aug 31, 2025
033717a
Updated ConfigSyncJob to use new convention
mpretty-cyro Sep 1, 2025
3963d84
Merge remote-tracking branch 'upstream/dev' into feature/job-runner-r…
mpretty-cyro Jan 16, 2026
b83876c
A bunch of logic simplification in the JobRunner
mpretty-cyro Jan 19, 2026
561f797
Finished updating core JobRunner and Job logic
mpretty-cyro Jan 21, 2026
ae19062
Fixed a number of issues found during testing
mpretty-cyro Jan 23, 2026
00ef5a3
Fixed a bunch of issues found during testing
mpretty-cyro Jan 26, 2026
0c122c4
Resolved some remaining TODOs
mpretty-cyro Jan 27, 2026
fcd56e0
Fix CI build errors
mpretty-cyro Jan 27, 2026
cc36a1e
Another CI build error...
mpretty-cyro Jan 27, 2026
6a8ac8f
More CI errors...
mpretty-cyro Jan 27, 2026
4a298f9
Merge branch 'dev' into feature/job-runner-refactor
mpretty-cyro Jan 28, 2026
258882e
Renamed the `canStart` function to `canRunConcurrentlyWith` to improv…
mpretty-cyro Jan 28, 2026
18489b4
Fixed an issue where the default community display pictures could fai…
mpretty-cyro Jan 28, 2026
83eab39
Fixed the disabled input states (change in master to a different file)
mpretty-cyro Jan 28, 2026
8e11069
Pulled Mocking system refactor across from the Network Refactor branch
mpretty-cyro Jan 30, 2026
f1d4fd6
Fixed the remaining unit tests
mpretty-cyro Feb 2, 2026
0abe3c2
Fixed an issue where removing the display pic could appear as though …
mpretty-cyro Feb 2, 2026
34f6536
Removed some empty migrations
mpretty-cyro Feb 2, 2026
1b675ac
Missed a commit
mpretty-cyro Feb 2, 2026
6a85676
Comment copy change
mpretty-cyro Feb 2, 2026
3d936a3
Fixed a couple of issues found during regression tests
mpretty-cyro Feb 2, 2026
32ab295
Fixed a CI build issue
mpretty-cyro Feb 2, 2026
7b87b15
Fixed an issue where the "Leave Group" button was missing
mpretty-cyro Feb 2, 2026
b02375c
Fixed a crash which could occur when deleting a conversation while an…
mpretty-cyro Feb 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
930 changes: 723 additions & 207 deletions Session.xcodeproj/project.pbxproj
View file
Open in desktop

Large diffs are not rendered by default.

35 changes: 31 additions & 4 deletions Session.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

12 changes: 9 additions & 3 deletions Session/Calls/Call Management/SessionCallManager.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,9 +191,15 @@ public final class SessionCallManager: NSObject, CallManagerProtocol {
Log.info(.calls, "Called suspendDatabaseIfCallEndedInBackground.")

if dependencies[singleton: .appContext].isInBackground {
// Stop all jobs except for message sending and when completed suspend the database
dependencies[singleton: .jobRunner].stopAndClearPendingJobs(exceptForVariant: .messageSend) { [dependencies] _ in
if self.currentCall?.hasEnded != false {
/// Stop all jobs except for message sending and when completed suspend the database
Task.detached(priority: .userInitiated) { [weak self, dependencies] in
await dependencies[singleton: .jobRunner].stopAndClearJobs(
filters: JobRunner.Filters(
exclude: [.variant(.messageSend)]
)
)

if self?.currentCall?.hasEnded != false {
dependencies.mutate(cache: .libSessionNetwork) { $0.suspendNetworkAccess() }
dependencies[singleton: .storage].suspendDatabaseAccess()
Log.flush()
Expand Down
156 changes: 75 additions & 81 deletions Session/Closed Groups/EditGroupViewModel.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -605,48 +605,45 @@ class EditGroupViewModel: SessionTableViewModel, NavigatableStateHolder, Observa
)

/// Actually trigger the sending process
MessageSender
.addGroupMembers(
groupSessionId: threadId,
members: memberInfo,
allowAccessToHistoricMessages: dependencies[feature: .updatedGroupsAllowHistoricAccessOnInvite],
using: dependencies
)
.subscribe(on: DispatchQueue.global(qos: .userInitiated), using: dependencies)
.receive(on: DispatchQueue.main, using: dependencies)
.sinkUntilComplete(
receiveCompletion: { [weak self, threadId, dependencies] result in
switch result {
case .finished: break
case .failure:
let memberIds: [String] = memberInfo.map(\.id)

/// Flag the members as failed
dependencies[singleton: .storage].writeAsync { db in
try? GroupMember
.filter(GroupMember.Columns.groupId == threadId)
.filter(memberIds.contains(GroupMember.Columns.profileId))
.updateAllAndConfig(
db,
GroupMember.Columns.roleStatus.set(to: GroupMember.RoleStatus.failed),
using: dependencies
)
Task.detached(priority: .userInitiated) { [weak self, threadId, dependencies] in
do {
try await MessageSender.addGroupMembers(
groupSessionId: threadId,
members: memberInfo,
allowAccessToHistoricMessages: dependencies[feature: .updatedGroupsAllowHistoricAccessOnInvite],
using: dependencies
)
}
catch {
let memberIds: [String] = memberInfo.map(\.id)

/// Flag the members as failed
try await dependencies[singleton: .storage].writeAsync { db in
try? GroupMember
.filter(GroupMember.Columns.groupId == threadId)
.filter(memberIds.contains(GroupMember.Columns.profileId))
.updateAllAndConfig(
db,
GroupMember.Columns.roleStatus.set(to: GroupMember.RoleStatus.failed),
using: dependencies
)
}

/// Show a toast that the invitations failed to send
await MainActor.run { [weak self] in
self?.showToast(
text: GroupInviteMemberJob.failureMessage(
groupName: currentGroupName,
memberIds: memberIds,
profileInfo: memberInfo.reduce(into: [:]) { result, next in
result[next.id] = next.profile
}

/// Show a toast that the invitations failed to send
self?.showToast(
text: GroupInviteMemberJob.failureMessage(
groupName: currentGroupName,
memberIds: memberIds,
profileInfo: memberInfo.reduce(into: [:]) { result, next in
result[next.id] = next.profile
}
),
backgroundColor: .backgroundSecondary
)
}
),
backgroundColor: .backgroundSecondary
)
}
)
}
}
}

private func resendInvitations(
Expand All @@ -664,45 +661,42 @@ class EditGroupViewModel: SessionTableViewModel, NavigatableStateHolder, Observa
/// Actually trigger the sending process
let memberIds: [String] = memberInfo.map { $0.id }

MessageSender
.resendInvitations(
groupSessionId: threadId,
memberIds: memberIds,
using: dependencies
)
.subscribe(on: DispatchQueue.global(qos: .userInitiated), using: dependencies)
.receive(on: DispatchQueue.main, using: dependencies)
.sinkUntilComplete(
receiveCompletion: { [weak self, threadId, dependencies] result in
switch result {
case .finished: break
case .failure:
/// Flag the members as failed
dependencies[singleton: .storage].writeAsync { db in
try? GroupMember
.filter(GroupMember.Columns.groupId == threadId)
.filter(memberIds.contains(GroupMember.Columns.profileId))
.updateAllAndConfig(
db,
GroupMember.Columns.roleStatus.set(to: GroupMember.RoleStatus.failed),
using: dependencies
)
Task.detached(priority: .userInitiated) { [weak self, threadId, dependencies] in
do {
try await MessageSender.resendInvitations(
groupSessionId: threadId,
memberIds: memberIds,
using: dependencies
)
}
catch {
/// Flag the members as failed
_ = try? await dependencies[singleton: .storage].writeAsync { db in
try GroupMember
.filter(GroupMember.Columns.groupId == threadId)
.filter(memberIds.contains(GroupMember.Columns.profileId))
.updateAllAndConfig(
db,
GroupMember.Columns.roleStatus.set(to: GroupMember.RoleStatus.failed),
using: dependencies
)
}

/// Show a toast that the invitations failed to send
await MainActor.run { [weak self] in
self?.showToast(
text: GroupInviteMemberJob.failureMessage(
groupName: currentGroupName,
memberIds: memberIds,
profileInfo: memberInfo.reduce(into: [:]) { result, next in
result[next.id] = next.profile
}

/// Show a toast that the invitations failed to send
self?.showToast(
text: GroupInviteMemberJob.failureMessage(
groupName: currentGroupName,
memberIds: memberIds,
profileInfo: memberInfo.reduce(into: [:]) { result, next in
result[next.id] = next.profile
}
),
backgroundColor: .backgroundSecondary
)
}
),
backgroundColor: .backgroundSecondary
)
}
)
}
}
}

private func removeMembers(currentGroupName: String, memberIds: Set<String>) {
Expand Down Expand Up @@ -756,16 +750,16 @@ class EditGroupViewModel: SessionTableViewModel, NavigatableStateHolder, Observa
cancelStyle: .alert_text,
dismissOnConfirm: false,
onConfirm: { [weak self, threadId, dependencies] modal in
MessageSender
.removeGroupMembers(
Task.detached(priority: .userInitiated) { [weak self] in
try await MessageSender.removeGroupMembers(
groupSessionId: threadId,
memberIds: memberIds,
removeTheirMessages: dependencies[feature: .updatedGroupsRemoveMessagesOnKick],
sendMemberChangedMessage: true,
using: dependencies
)
.subscribe(on: DispatchQueue.global(qos: .userInitiated), using: dependencies)
.sinkUntilComplete()
}

self?.selectedIdsSubject.send((currentGroupName, []))
modal.dismiss(animated: true)
}
Expand Down
86 changes: 39 additions & 47 deletions Session/Conversations/ConversationVC+Interaction.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1380,8 +1380,7 @@ extension ConversationVC:
details: AttachmentDownloadJob.Details(
attachmentId: mediaView.attachment.id
)
),
canStartJob: true
)
)
}
break
Expand Down Expand Up @@ -2225,58 +2224,51 @@ extension ConversationVC:
return presentingViewController.present(errorModal, animated: true, completion: nil)
}

dependencies[singleton: .storage]
.writePublisher { db in
dependencies[singleton: .communityManager].add(
db,
roomToken: room,
server: server,
publicKey: publicKey,
joinedAt: (dependencies[cache: .snodeAPI].currentOffsetTimestampMs() / 1000),
forceVisible: false
)
}
.flatMap { successfullyAddedGroup in
dependencies[singleton: .communityManager].performInitialRequestsAfterAdd(
queue: DispatchQueue.global(qos: .userInitiated),
Task.detached(priority: .userInitiated) {
do {
let successfullyAddedGroup: Bool = try await dependencies[singleton: .storage].writeAsync { db in
dependencies[singleton: .communityManager].add(
db,
roomToken: room,
server: server,
publicKey: publicKey,
joinedAt: (dependencies[cache: .snodeAPI].currentOffsetTimestampMs() / 1000),
forceVisible: false
)
}
try await dependencies[singleton: .communityManager].performInitialRequestsAfterAdd(
successfullyAddedGroup: successfullyAddedGroup,
roomToken: room,
server: server,
publicKey: publicKey
)
}
.subscribe(on: DispatchQueue.global(qos: .userInitiated))
.receive(on: DispatchQueue.main)
.sinkUntilComplete(
receiveCompletion: { result in
switch result {
case .finished: break
case .failure(let error):
// If there was a failure then the group will be in invalid state until
// the next launch so remove it (the user will be left on the previous
// screen so can re-trigger the join)
dependencies[singleton: .storage].writeAsync { db in
try dependencies[singleton: .communityManager].delete(
db,
openGroupId: OpenGroup.idFor(roomToken: room, server: server),
skipLibSessionUpdate: false
)
}

// Show the user an error indicating they failed to properly join the group
let errorModal: ConfirmationModal = ConfirmationModal(
info: ConfirmationModal.Info(
title: "communityJoinError".localized(),
body: .text("\(error)"),
cancelTitle: "okay".localized(),
cancelStyle: .alert_text
)
)

presentingViewController.present(errorModal, animated: true, completion: nil)
}
catch {
/// If there was a failure then the group will be in invalid state until the next launch so remove it (the
/// user will be left on the previous screen so can re-trigger the join)
try? await dependencies[singleton: .storage].writeAsync { db in
try dependencies[singleton: .communityManager].delete(
db,
openGroupId: OpenGroup.idFor(roomToken: room, server: server),
skipLibSessionUpdate: false
)
}
)

/// Show the user an error indicating they failed to properly join the group
await MainActor.run {
let errorModal: ConfirmationModal = ConfirmationModal(
info: ConfirmationModal.Info(
title: "communityJoinError".localized(),
body: .text("\(error)"),
cancelTitle: "okay".localized(),
cancelStyle: .alert_text
)
)

presentingViewController.present(errorModal, animated: true, completion: nil)
}
}
}
}
)
)
Expand Down
Loading