// Copyright © 2022 Rangeproof Pty Ltd. All rights reserved.
//
// stringlint:disable
import Foundation
import GRDB
public protocol JobRunnerType {
// MARK: - Configuration
func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant)
func canStart(queue: JobQueue?) -> Bool
func afterBlockingQueue(callback: @escaping () -> ())
// MARK: - State Management
func jobInfoFor(jobs: [Job]?, state: JobRunner.JobState, variant: Job.Variant?) -> [Int64: JobRunner.JobInfo]
func appDidFinishLaunching(using dependencies: Dependencies)
func appDidBecomeActive(using dependencies: Dependencies)
func startNonBlockingQueues(using dependencies: Dependencies)
func stopAndClearPendingJobs(exceptForVariant: Job.Variant?, onComplete: (() -> ())?)
// MARK: - Job Scheduling
@discardableResult func add(_ db: Database, job: Job?, canStartJob: Bool, using dependencies: Dependencies) -> Job?
func upsert(_ db: Database, job: Job?, canStartJob: Bool, using dependencies: Dependencies)
@discardableResult func insert(_ db: Database, job: Job?, before otherJob: Job) -> (Int64, Job)?
}
// MARK: - JobRunnerType Convenience
public extension JobRunnerType {
func allJobInfo() -> [Int64: JobRunner.JobInfo] { return jobInfoFor(jobs: nil, state: .any, variant: nil) }
func jobInfoFor(jobs: [Job]) -> [Int64: JobRunner.JobInfo] {
return jobInfoFor(jobs: jobs, state: .any, variant: nil)
}
func jobInfoFor(jobs: [Job], state: JobRunner.JobState) -> [Int64: JobRunner.JobInfo] {
return jobInfoFor(jobs: jobs, state: state, variant: nil)
}
func jobInfoFor(state: JobRunner.JobState) -> [Int64: JobRunner.JobInfo] {
return jobInfoFor(jobs: nil, state: state, variant: nil)
}
func jobInfoFor(state: JobRunner.JobState, variant: Job.Variant) -> [Int64: JobRunner.JobInfo] {
return jobInfoFor(jobs: nil, state: state, variant: variant)
}
func jobInfoFor(variant: Job.Variant) -> [Int64: JobRunner.JobInfo] {
return jobInfoFor(jobs: nil, state: .any, variant: variant)
}
func isCurrentlyRunning(_ job: Job?) -> Bool {
guard let job: Job = job else { return false }
return !jobInfoFor(jobs: [job], state: .running).isEmpty
}
func hasJob<T: Encodable>(
of variant: Job.Variant? = nil,
inState state: JobRunner.JobState = .any,
with jobDetails: T
) -> Bool {
guard
let detailsData: Data = try? JSONEncoder()
.with(outputFormatting: .sortedKeys) // Needed for deterministic comparison
.encode(jobDetails)
else { return false }
return jobInfoFor(jobs: nil, state: state, variant: variant)
.values
.contains(where: { $0.detailsData == detailsData })
}
func stopAndClearPendingJobs(exceptForVariant: Job.Variant? = nil, onComplete: (() -> ())? = nil) {
stopAndClearPendingJobs(exceptForVariant: exceptForVariant, onComplete: onComplete)
}
}
// MARK: - JobExecutor
public protocol JobExecutor {
/// The maximum number of times the job can fail before it fails permanently
///
/// **Note:** A value of `-1` means it will retry indefinitely
static var maxFailureCount: Int { get }
static var requiresThreadId: Bool { get }
static var requiresInteractionId: Bool { get }
/// This method contains the logic needed to complete a job
///
/// **Note:** The code in this method should run synchronously and the various
/// "result" blocks should not be called within a database closure
///
/// - Parameters:
/// - job: The job which is being run
/// - success: The closure which is called when the job succeeds (with an
/// updated `job` and a flag indicating whether the job should forcibly stop running)
/// - failure: The closure which is called when the job fails (with an updated
/// `job`, an `Error` (if applicable) and a flag indicating whether it was a permanent
/// failure)
/// - deferred: The closure which is called when the job is deferred (with an
/// updated `job`)
static func run(
_ job: Job,
queue: DispatchQueue,
success: @escaping (Job, Bool, Dependencies) -> (),
failure: @escaping (Job, Error?, Bool, Dependencies) -> (),
deferred: @escaping (Job, Dependencies) -> (),
using dependencies: Dependencies
)
}
// MARK: - JobRunner
public final class JobRunner: JobRunnerType {
public struct JobState: OptionSet, Hashable {
public let rawValue: UInt8
public init(rawValue: UInt8) {
self.rawValue = rawValue
}
public static let pending: JobState = JobState(rawValue: 1 << 0)
public static let running: JobState = JobState(rawValue: 1 << 1)
public static let any: JobState = [ .pending, .running ]
}
public enum JobResult {
case succeeded
case failed
case deferred
case notFound
}
public struct JobInfo: Equatable, CustomDebugStringConvertible {
public let variant: Job.Variant
public let threadId: String?
public let interactionId: Int64?
public let detailsData: Data?
public var debugDescription: String {
let dataDescription: String = detailsData
.map { data in "Data(hex: \(data.toHexString()), \(data.bytes.count) bytes" }
.defaulting(to: "nil")
return [
"JobRunner.JobInfo(",
"variant: \(variant),",
" threadId: \(threadId ?? "nil"),",
" interactionId: \(interactionId.map { "\($0)" } ?? "nil"),",
" detailsData: \(dataDescription)",
")"
].joined()
}
}
// MARK: - Variables
private let allowToExecuteJobs: Bool
private let blockingQueue: Atomic<JobQueue?>
private let queues: Atomic<[Job.Variant: JobQueue]>
private var blockingQueueDrainCallback: Atomic<[() -> ()]> = Atomic([])
internal var appReadyToStartQueues: Atomic<Bool> = Atomic(false)
internal var appHasBecomeActive: Atomic<Bool> = Atomic(false)
internal var perSessionJobsCompleted: Atomic<Set<Int64>> = Atomic([])
internal var hasCompletedInitialBecomeActive: Atomic<Bool> = Atomic(false)
internal var shutdownBackgroundTask: Atomic<OWSBackgroundTask?> = Atomic(nil)
private var canStartNonBlockingQueue: Bool {
blockingQueue.wrappedValue?.hasStartedAtLeastOnce.wrappedValue == true &&
blockingQueue.wrappedValue?.isRunning.wrappedValue != true &&
appHasBecomeActive.wrappedValue
}
// MARK: - Initialization
init(
isTestingJobRunner: Bool = false,
variantsToExclude: [Job.Variant] = [],
using dependencies: Dependencies = Dependencies()
) {
var jobVariants: Set<Job.Variant> = Job.Variant.allCases
.filter { !variantsToExclude.contains($0) }
.asSet()
self.allowToExecuteJobs = (
isTestingJobRunner || (
HasAppContext() &&
CurrentAppContext().isMainApp &&
!SNUtilitiesKit.isRunningTests
)
)
self.blockingQueue = Atomic(
JobQueue(
type: .blocking,
executionType: .serial,
qos: .default,
isTestingJobRunner: isTestingJobRunner,
jobVariants: []
)
)
self.queues = Atomic([
// MARK: -- Message Send Queue
JobQueue(
type: .messageSend,
executionType: .concurrent, // Allow as many jobs to run at once as supported by the device
qos: .default,
isTestingJobRunner: isTestingJobRunner,
jobVariants: [
jobVariants.remove(.attachmentUpload),
jobVariants.remove(.messageSend),
jobVariants.remove(.notifyPushServer),
jobVariants.remove(.sendReadReceipts),
jobVariants.remove(.groupLeaving),
jobVariants.remove(.configurationSync)
].compactMap { $0 }
),
// MARK: -- Message Receive Queue
JobQueue(
type: .messageReceive,
// Explicitly serial as executing concurrently means message receives getting processed at
// different speeds which can result in:
// • Small batches of messages appearing in the UI before larger batches
// • Closed group messages encrypted with updated keys could start parsing before it's key
// update message has been processed (ie. guaranteed to fail)
executionType: .serial,
qos: .default,
isTestingJobRunner: isTestingJobRunner,
jobVariants: [
jobVariants.remove(.messageReceive),
jobVariants.remove(.configMessageReceive)
].compactMap { $0 }
),
// MARK: -- Attachment Download Queue
JobQueue(
type: .attachmentDownload,
executionType: .serial,
qos: .utility,
isTestingJobRunner: isTestingJobRunner,
jobVariants: [
jobVariants.remove(.attachmentDownload)
].compactMap { $0 }
),
// MARK: -- General Queue
JobQueue(
type: .general(number: 0),
executionType: .serial,
qos: .utility,
isTestingJobRunner: isTestingJobRunner,
jobVariants: Array(jobVariants)
)
].reduce(into: [:]) { prev, next in
next.jobVariants.forEach { variant in
prev[variant] = next
}
})
// Now that we've finished setting up the JobRunner, update the queue closures
self.blockingQueue.mutate {
$0?.canStart = { [weak self] queue -> Bool in (self?.canStart(queue: queue) == true) }
$0?.onQueueDrained = { [weak self] in
// Once all blocking jobs have been completed we want to start running
// the remaining job queues
self?.startNonBlockingQueues(using: dependencies)
self?.blockingQueueDrainCallback.mutate {
$0.forEach { $0() }
$0 = []
}
}
}
self.queues.mutate {
$0.values.forEach { queue in
queue.canStart = { [weak self] targetQueue -> Bool in (self?.canStart(queue: targetQueue) == true) }
}
}
}
// MARK: - Configuration
public func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant) {
blockingQueue.wrappedValue?.setExecutor(executor, for: variant) // The blocking queue can run any job
queues.wrappedValue[variant]?.setExecutor(executor, for: variant)
}
public func canStart(queue: JobQueue?) -> Bool {
return (
allowToExecuteJobs &&
appReadyToStartQueues.wrappedValue && (
queue?.type == .blocking ||
canStartNonBlockingQueue
)
)
}
public func afterBlockingQueue(callback: @escaping () -> ()) {
guard
(blockingQueue.wrappedValue?.hasStartedAtLeastOnce.wrappedValue != true) ||
(blockingQueue.wrappedValue?.isRunning.wrappedValue == true)
else { return callback() }
blockingQueueDrainCallback.mutate { $0.append(callback) }
}
// MARK: - State Management
public func jobInfoFor(
jobs: [Job]?,
state: JobRunner.JobState,
variant: Job.Variant?
) -> [Int64: JobRunner.JobInfo] {
var result: [(Int64, JobRunner.JobInfo)] = []
let targetKeys: [JobQueue.JobKey] = (jobs?.compactMap { JobQueue.JobKey($0) } ?? [])
let targetVariants: [Job.Variant] = (variant.map { [$0] } ?? jobs?.map { $0.variant })
.defaulting(to: [])
// Insert the state of any pending jobs
if state.contains(.pending) {
func infoFor(queue: JobQueue?, variants: [Job.Variant]) -> [(Int64, JobRunner.JobInfo)] {
return (queue?.pendingJobsQueue.wrappedValue
.filter { variants.isEmpty || variants.contains($0.variant) }
.compactMap { job -> (Int64, JobRunner.JobInfo)? in
guard let jobKey: JobQueue.JobKey = JobQueue.JobKey(job) else { return nil }
guard
targetKeys.isEmpty ||
targetKeys.contains(jobKey)
else { return nil }
return (
jobKey.id,
JobRunner.JobInfo(
variant: job.variant,
threadId: job.threadId,
interactionId: job.interactionId,
detailsData: job.details
)
)
})
.defaulting(to: [])
}
result.append(contentsOf: infoFor(queue: blockingQueue.wrappedValue, variants: targetVariants))
queues.wrappedValue
.filter { key, _ -> Bool in targetVariants.isEmpty || targetVariants.contains(key) }
.map { _, queue in queue }
.asSet()
.forEach { queue in result.append(contentsOf: infoFor(queue: queue, variants: targetVariants)) }
}
// Insert the state of any running jobs
if state.contains(.running) {
func infoFor(queue: JobQueue?, variants: [Job.Variant]) -> [(Int64, JobRunner.JobInfo)] {
return (queue?.infoForAllCurrentlyRunningJobs()
.filter { variants.isEmpty || variants.contains($0.value.variant) }
.compactMap { jobId, info -> (Int64, JobRunner.JobInfo)? in
guard
targetKeys.isEmpty ||
targetKeys.contains(JobQueue.JobKey(id: jobId, variant: info.variant))
else { return nil }
return (jobId, info)
})
.defaulting(to: [])
}
result.append(contentsOf: infoFor(queue: blockingQueue.wrappedValue, variants: targetVariants))
queues.wrappedValue
.filter { key, _ -> Bool in targetVariants.isEmpty || targetVariants.contains(key) }
.map { _, queue in queue }
.asSet()
.forEach { queue in result.append(contentsOf: infoFor(queue: queue, variants: targetVariants)) }
}
return result
.reduce(into: [:]) { result, next in
result[next.0] = next.1
}
}
public func appDidFinishLaunching(using dependencies: Dependencies) {
// Flag that the JobRunner can start it's queues
appReadyToStartQueues.mutate { $0 = true }
// Note: 'appDidBecomeActive' will run on first launch anyway so we can
// leave those jobs out and can wait until then to start the JobRunner
let jobsToRun: (blocking: [Job], nonBlocking: [Job]) = dependencies.storage
.read { db in
let blockingJobs: [Job] = try Job
.filter(
[
Job.Behaviour.recurringOnLaunch,
Job.Behaviour.runOnceNextLaunch
].contains(Job.Columns.behaviour)
)
.filter(Job.Columns.shouldBlock == true)
.order(
Job.Columns.priority.desc,
Job.Columns.id
)
.fetchAll(db)
let nonblockingJobs: [Job] = try Job
.filter(
[
Job.Behaviour.recurringOnLaunch,
Job.Behaviour.runOnceNextLaunch
].contains(Job.Columns.behaviour)
)
.filter(Job.Columns.shouldBlock == false)
.order(
Job.Columns.priority.desc,
Job.Columns.id
)
.fetchAll(db)
return (blockingJobs, nonblockingJobs)
}
.defaulting(to: ([], []))
// Add and start any blocking jobs
blockingQueue.wrappedValue?.appDidFinishLaunching(
with: jobsToRun.blocking,
canStart: true,
using: dependencies
)
// Add any non-blocking jobs (we don't start these incase there are blocking "on active"
// jobs as well)
let jobsByVariant: [Job.Variant: [Job]] = jobsToRun.nonBlocking.grouped(by: \.variant)
let jobQueues: [Job.Variant: JobQueue] = queues.wrappedValue
jobsByVariant.forEach { variant, jobs in
jobQueues[variant]?.appDidFinishLaunching(
with: jobs,
canStart: false,
using: dependencies
)
}
}
public func appDidBecomeActive(using dependencies: Dependencies) {
// Flag that the JobRunner can start it's queues and start queueing non-launch jobs
appReadyToStartQueues.mutate { $0 = true }
appHasBecomeActive.mutate { $0 = true }
// If we have a running "sutdownBackgroundTask" then we want to cancel it as otherwise it
// can result in the database being suspended and us being unable to interact with it at all
shutdownBackgroundTask.mutate {
$0?.cancel()
$0 = nil
}
// Retrieve any jobs which should run when becoming active
let hasCompletedInitialBecomeActive: Bool = self.hasCompletedInitialBecomeActive.wrappedValue
let jobsToRun: [Job] = dependencies.storage
.read { db in
return try Job
.filter(Job.Columns.behaviour == Job.Behaviour.recurringOnActive)
.order(
Job.Columns.priority.desc,
Job.Columns.id
)
.fetchAll(db)
}
.defaulting(to: [])
.filter { hasCompletedInitialBecomeActive || !$0.shouldSkipLaunchBecomeActive }
// Store the current queue state locally to avoid multiple atomic retrievals
let jobQueues: [Job.Variant: JobQueue] = queues.wrappedValue
let blockingQueueIsRunning: Bool = (blockingQueue.wrappedValue?.isRunning.wrappedValue == true)
guard !jobsToRun.isEmpty else {
if !blockingQueueIsRunning {
jobQueues.map { _, queue in queue }.asSet().forEach { $0.start(using: dependencies) }
}
return
}
// Add and start any non-blocking jobs (if there are no blocking jobs)
//
// We only want to trigger the queue to start once so we need to consolidate the
// queues to list of jobs (as queues can handle multiple job variants), this means
// that 'onActive' jobs will be queued before any standard jobs
let jobsByVariant: [Job.Variant: [Job]] = jobsToRun.grouped(by: \.variant)
jobQueues
.reduce(into: [:]) { result, variantAndQueue in
result[variantAndQueue.value] = (result[variantAndQueue.value] ?? [])
.appending(contentsOf: (jobsByVariant[variantAndQueue.key] ?? []))
}
.forEach { queue, jobs in
queue.appDidBecomeActive(
with: jobs,
canStart: !blockingQueueIsRunning,
using: dependencies
)
}
self.hasCompletedInitialBecomeActive.mutate { $0 = true }
}
public func startNonBlockingQueues(using dependencies: Dependencies) {
queues.wrappedValue.map { _, queue in queue }.asSet().forEach { queue in
queue.start(using: dependencies)
}
}
public func stopAndClearPendingJobs(
exceptForVariant: Job.Variant?,
onComplete: (() -> ())?
) {
// Inform the JobRunner that it can't start any queues (this is to prevent queues from
// rescheduling themselves while in the background, when the app restarts or becomes active
// the JobRunenr will update this flag)
appReadyToStartQueues.mutate { $0 = false }
appHasBecomeActive.mutate { $0 = false }
// Stop all queues except for the one containing the `exceptForVariant`
queues.wrappedValue
.map { _, queue in queue }
.asSet()
.filter { queue -> Bool in
guard let exceptForVariant: Job.Variant = exceptForVariant else { return true }
return !queue.jobVariants.contains(exceptForVariant)
}
.forEach { $0.stopAndClearPendingJobs() }
// Ensure the queue is actually running (if not the trigger the callback immediately)
guard
let exceptForVariant: Job.Variant = exceptForVariant,
let queue: JobQueue = queues.wrappedValue[exceptForVariant],
queue.isRunning.wrappedValue == true
else {
onComplete?()
return
}
let oldQueueDrained: (() -> ())? = queue.onQueueDrained
// Create a backgroundTask to give the queue the chance to properly be drained
shutdownBackgroundTask.mutate {
$0 = OWSBackgroundTask(labelStr: #function) { [weak queue] state in
// If the background task didn't succeed then trigger the onComplete (and hope we have
// enough time to complete it's logic)
guard state != .cancelled else {
queue?.onQueueDrained = oldQueueDrained
return
}
guard state != .success else { return }
onComplete?()
queue?.onQueueDrained = oldQueueDrained
queue?.stopAndClearPendingJobs()
}
}
// Add a callback to be triggered once the queue is drained
queue.onQueueDrained = { [weak self, weak queue] in
oldQueueDrained?()
queue?.onQueueDrained = oldQueueDrained
onComplete?()
self?.shutdownBackgroundTask.mutate { $0 = nil }
}
}
// MARK: - Execution
@discardableResult public func add(
_ db: Database,
job: Job?,
canStartJob: Bool,
using dependencies: Dependencies
) -> Job? {
// Store the job into the database (getting an id for it)
guard let updatedJob: Job = try? job?.inserted(db) else {
SNLog("[JobRunner] Unable to add \(job.map { "\($0.variant)" } ?? "unknown") job")
return nil
}
guard !canStartJob || updatedJob.id != nil else {
SNLog("[JobRunner] Not starting \(job.map { "\($0.variant)" } ?? "unknown") job due to missing id")
return nil
}
// Don't add to the queue if the JobRunner isn't ready (it's been saved to the db so it'll be loaded
// once the queue actually get started later)
guard canAddToQueue(updatedJob) else { return updatedJob }
queues.wrappedValue[updatedJob.variant]?.add(db, job: updatedJob, canStartJob: canStartJob, using: dependencies)
// Don't start the queue if the job can't be started
guard canStartJob else { return updatedJob }
// Start the job runner if needed
db.afterNextTransactionNestedOnce(dedupeId: "JobRunner-Start: \(updatedJob.variant)") { [weak self] _ in
self?.queues.wrappedValue[updatedJob.variant]?.start(using: dependencies)
}
return updatedJob
}
public func upsert(
_ db: Database,
job: Job?,
canStartJob: Bool,
using dependencies: Dependencies
) {
guard let job: Job = job else { return } // Ignore null jobs
guard job.id != nil else {
add(db, job: job, canStartJob: canStartJob, using: dependencies)
return
}
// Don't add to the queue if the JobRunner isn't ready (it's been saved to the db so it'll be loaded
// once the queue actually get started later)
guard canAddToQueue(job) else { return }
queues.wrappedValue[job.variant]?.upsert(db, job: job, canStartJob: canStartJob, using: dependencies)
// Don't start the queue if the job can't be started
guard canStartJob else { return }
// Start the job runner if needed
db.afterNextTransactionNestedOnce(dedupeId: "JobRunner-Start: \(job.variant)") { [weak self] _ in
self?.queues.wrappedValue[job.variant]?.start(using: dependencies)
}
}
@discardableResult public func insert(
_ db: Database,
job: Job?,
before otherJob: Job
) -> (Int64, Job)? {
switch job?.behaviour {
case .recurringOnActive, .recurringOnLaunch, .runOnceNextLaunch:
SNLog("[JobRunner] Attempted to insert \(job.map { "\($0.variant)" } ?? "unknown") job before the current one even though it's behaviour is \(job.map { "\($0.behaviour)" } ?? "unknown")")
return nil
default: break
}
// Store the job into the database (getting an id for it)
guard let updatedJob: Job = try? job?.inserted(db) else {
SNLog("[JobRunner] Unable to add \(job.map { "\($0.variant)" } ?? "unknown") job")
return nil
}
guard let jobId: Int64 = updatedJob.id else {
SNLog("[JobRunner] Unable to add \(job.map { "\($0.variant)" } ?? "unknown") job due to missing id")
return nil
}
queues.wrappedValue[updatedJob.variant]?.insert(updatedJob, before: otherJob)
return (jobId, updatedJob)
}
internal func afterCurrentlyRunningJob(_ job: Job?, callback: @escaping (JobResult) -> ()) {
guard let job: Job = job, let jobId: Int64 = job.id, let queue: JobQueue = queues.wrappedValue[job.variant] else {
callback(.notFound)
return
}
queue.afterCurrentlyRunningJob(jobId, callback: callback)
}
internal func removePendingJob(_ job: Job?) {
guard let job: Job = job, let jobId: Int64 = job.id else { return }
queues.wrappedValue[job.variant]?.removePendingJob(jobId)
}
// MARK: - Convenience
fileprivate static func getRetryInterval(for job: Job) -> TimeInterval {
// Arbitrary backoff factor...
// try 1 delay: 0.5s
// try 2 delay: 1s
// ...
// try 5 delay: 16s
// ...
// try 11 delay: 512s
let maxBackoff: Double = 10 * 60 // 10 minutes
return 0.25 * min(maxBackoff, pow(2, Double(job.failureCount)))
}
fileprivate func canAddToQueue(_ job: Job) -> Bool {
// We can only start the job if it's an "on launch" job or the app has become active
return (
job.behaviour == .runOnceNextLaunch ||
job.behaviour == .recurringOnLaunch ||
appHasBecomeActive.wrappedValue
)
}
}
// MARK: - JobQueue
public final class JobQueue: Hashable {
fileprivate enum QueueType: Hashable {
case blocking
case general(number: Int)
case messageSend
case messageReceive
case attachmentDownload
var name: String {
switch self {
case .blocking: return "Blocking"
case .general(let number): return "General-\(number)"
case .messageSend: return "MessageSend"
case .messageReceive: return "MessageReceive"
case .attachmentDownload: return "AttachmentDownload"
}
}
}
fileprivate enum ExecutionType {
/// A serial queue will execute one job at a time until the queue is empty, then will load any new/deferred
/// jobs and run those one at a time
case serial
/// A concurrent queue will execute as many jobs as the device supports at once until the queue is empty,
/// then will load any new/deferred jobs and try to start them all
case concurrent
}
private class Trigger {
private var timer: Timer?
fileprivate var fireTimestamp: TimeInterval = 0
static func create(
queue: JobQueue,
timestamp: TimeInterval,
using dependencies: Dependencies
) -> Trigger? {
/// Setup the trigger (wait at least 1 second before triggering)
///
/// **Note:** We use the `Timer.scheduledTimerOnMainThread` method because running a timer
/// on our random queue threads results in the timer never firing, the `start` method will redirect itself to
/// the correct thread
let trigger: Trigger = Trigger()
trigger.fireTimestamp = max(1, (timestamp - dependencies.dateNow.timeIntervalSince1970))
trigger.timer = Timer.scheduledTimerOnMainThread(
withTimeInterval: trigger.fireTimestamp,
repeats: false,
using: dependencies,
block: { [weak queue] _ in
queue?.start(using: dependencies)
}
)
return trigger
}
func invalidate() {
// Need to do this to prevent a strong reference cycle
timer?.invalidate()
timer = nil
}
}
fileprivate struct JobKey: Equatable, Hashable {
fileprivate let id: Int64
fileprivate let variant: Job.Variant
fileprivate init(id: Int64, variant: Job.Variant) {
self.id = id
self.variant = variant
}
fileprivate init?(_ job: Job?) {
guard let id: Int64 = job?.id, let variant: Job.Variant = job?.variant else { return nil }
self.id = id
self.variant = variant
}
}
private static let deferralLoopThreshold: Int = 3
private let id: UUID = UUID()
fileprivate let type: QueueType
private let executionType: ExecutionType
private let qosClass: DispatchQoS
private let queueKey: DispatchSpecificKey = DispatchSpecificKey<String>()
private let queueContext: String
fileprivate let jobVariants: [Job.Variant]
private lazy var internalQueue: DispatchQueue = {
let result: DispatchQueue = DispatchQueue(
label: self.queueContext,
qos: self.qosClass,
attributes: (self.executionType == .concurrent ? [.concurrent] : []),
autoreleaseFrequency: .inherit,
target: nil
)
result.setSpecific(key: queueKey, value: queueContext)
return result
}()
private var executorMap: Atomic<[Job.Variant: JobExecutor.Type]> = Atomic([:])
fileprivate var canStart: ((JobQueue?) -> Bool)?
fileprivate var onQueueDrained: (() -> ())?
fileprivate var hasStartedAtLeastOnce: Atomic<Bool> = Atomic(false)
fileprivate var isRunning: Atomic<Bool> = Atomic(false)
fileprivate var pendingJobsQueue: Atomic<[Job]> = Atomic([])
private var nextTrigger: Atomic<Trigger?> = Atomic(nil)
private var jobCallbacks: Atomic<[Int64: [(JobRunner.JobResult) -> ()]]> = Atomic([:])
private var currentlyRunningJobIds: Atomic<Set<Int64>> = Atomic([])
private var currentlyRunningJobInfo: Atomic<[Int64: JobRunner.JobInfo]> = Atomic([:])
private var deferLoopTracker: Atomic<[Int64: (count: Int, times: [TimeInterval])]> = Atomic([:])
private let maxDeferralsPerSecond: Int
fileprivate var hasPendingJobs: Bool { !pendingJobsQueue.wrappedValue.isEmpty }
// MARK: - Initialization
fileprivate init(
type: QueueType,
executionType: ExecutionType,
qos: DispatchQoS,
isTestingJobRunner: Bool,
jobVariants: [Job.Variant]
) {
self.type = type
self.executionType = executionType
self.queueContext = "JobQueue-\(type.name)"
self.qosClass = qos
self.maxDeferralsPerSecond = (isTestingJobRunner ? 10 : 1) // Allow for tripping the defer loop in tests
self.jobVariants = jobVariants
}
// MARK: - Hashable
public func hash(into hasher: inout Hasher) {
id.hash(into: &hasher)
}
public static func == (lhs: JobQueue, rhs: JobQueue) -> Bool {
return (lhs.id == rhs.id)
}
// MARK: - Configuration
fileprivate func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant) {
executorMap.mutate { $0[variant] = executor }
}
// MARK: - Execution
fileprivate func add(
_ db: Database,
job: Job,
canStartJob: Bool,
using dependencies: Dependencies
) {
// Check if the job should be added to the queue
guard
canStartJob,
job.behaviour != .runOnceNextLaunch,
job.nextRunTimestamp <= dependencies.dateNow.timeIntervalSince1970
else { return }
guard job.id != nil else {
SNLog("[JobRunner] Prevented attempt to add \(job.variant) job without id to queue")
return
}
pendingJobsQueue.mutate { $0.append(job) }
// If this is a concurrent queue then we should immediately start the next job
guard executionType == .concurrent else { return }
// Ensure that the database commit has completed and then trigger the next job to run (need
// to ensure any interactions have been correctly inserted first)
db.afterNextTransactionNestedOnce(dedupeId: "JobRunner-Add: \(job.variant)") { [weak self] _ in
self?.runNextJob(using: dependencies)
}
}
/// Upsert a job onto the queue, if the queue isn't currently running and 'canStartJob' is true then this will start
/// the JobRunner
///
/// **Note:** If the job has a `behaviour` of `runOnceNextLaunch` or the `nextRunTimestamp`
/// is in the future then the job won't be started
fileprivate func upsert(
_ db: Database,
job: Job,
canStartJob: Bool,
using dependencies: Dependencies
) {
guard let jobId: Int64 = job.id else {
SNLog("[JobRunner] Prevented attempt to upsert \(job.variant) job without id to queue")
return
}
// Lock the pendingJobsQueue while checking the index and inserting to ensure we don't run into
// any multi-threading shenanigans
//
// Note: currently running jobs are removed from the pendingJobsQueue so we don't need to check
// the 'jobsCurrentlyRunning' set
var didUpdateExistingJob: Bool = false
pendingJobsQueue.mutate { queue in
if let jobIndex: Array<Job>.Index = queue.firstIndex(where: { $0.id == jobId }) {
queue[jobIndex] = job
didUpdateExistingJob = true
}
}
// If we didn't update an existing job then we need to add it to the pendingJobsQueue
guard !didUpdateExistingJob else { return }
add(db, job: job, canStartJob: canStartJob, using: dependencies)
}
fileprivate func insert(_ job: Job, before otherJob: Job) {
guard job.id != nil else {
SNLog("[JobRunner] Prevented attempt to insert \(job.variant) job without id to queue")
return
}
// Insert the job before the current job (re-adding the current job to
// the start of the pendingJobsQueue if it's not in there) - this will mean the new
// job will run and then the otherJob will run (or run again) once it's
// done
pendingJobsQueue.mutate {
guard let otherJobIndex: Int = $0.firstIndex(of: otherJob) else {
$0.insert(contentsOf: [job, otherJob], at: 0)
return
}
$0.insert(job, at: otherJobIndex)
}
}
fileprivate func appDidFinishLaunching(
with jobs: [Job],
canStart: Bool,
using dependencies: Dependencies
) {
pendingJobsQueue.mutate { $0.append(contentsOf: jobs) }
// Start the job runner if needed
if canStart && !isRunning.wrappedValue {
start(using: dependencies)
}
}
fileprivate func appDidBecomeActive(
with jobs: [Job],
canStart: Bool,
using dependencies: Dependencies
) {
let currentlyRunningJobIds: Set<Int64> = currentlyRunningJobIds.wrappedValue
pendingJobsQueue.mutate { queue in
// Avoid re-adding jobs to the queue that are already in it (this can
// happen if the user sends the app to the background before the 'onActive'
// jobs and then brings it back to the foreground)
let jobsNotAlreadyInQueue: [Job] = jobs
.filter { job in
!currentlyRunningJobIds.contains(job.id ?? -1) &&
!queue.contains(where: { $0.id == job.id })
}
queue.append(contentsOf: jobsNotAlreadyInQueue)
}
// Start the job runner if needed
if canStart && !isRunning.wrappedValue {
start(using: dependencies)
}
}
fileprivate func infoForAllCurrentlyRunningJobs() -> [Int64: JobRunner.JobInfo] {
return currentlyRunningJobInfo.wrappedValue
}
fileprivate func afterCurrentlyRunningJob(_ jobId: Int64, callback: @escaping (JobRunner.JobResult) -> ()) {
guard currentlyRunningJobIds.wrappedValue.contains(jobId) else { return callback(.notFound) }
jobCallbacks.mutate { jobCallbacks in
jobCallbacks[jobId] = (jobCallbacks[jobId] ?? []).appending(callback)
}
}
fileprivate func hasPendingOrRunningJobWith(
threadId: String? = nil,
interactionId: Int64? = nil,
detailsData: Data? = nil
) -> Bool {
let pendingJobs: [Job] = pendingJobsQueue.wrappedValue
let currentlyRunningJobInfo: [Int64: JobRunner.JobInfo] = currentlyRunningJobInfo.wrappedValue
var possibleJobIds: Set<Int64> = Set(currentlyRunningJobInfo.keys)
.inserting(contentsOf: pendingJobs.compactMap { $0.id }.asSet())
// Remove any which don't have the matching threadId (if provided)
if let targetThreadId: String = threadId {
let pendingJobIdsWithWrongThreadId: Set<Int64> = pendingJobs
.filter { $0.threadId != targetThreadId }
.compactMap { $0.id }
.asSet()
let runningJobIdsWithWrongThreadId: Set<Int64> = currentlyRunningJobInfo
.filter { _, info -> Bool in info.threadId != targetThreadId }
.map { key, _ in key }
.asSet()
possibleJobIds = possibleJobIds
.subtracting(pendingJobIdsWithWrongThreadId)
.subtracting(runningJobIdsWithWrongThreadId)
}
// Remove any which don't have the matching interactionId (if provided)
if let targetInteractionId: Int64 = interactionId {
let pendingJobIdsWithWrongInteractionId: Set<Int64> = pendingJobs
.filter { $0.interactionId != targetInteractionId }
.compactMap { $0.id }
.asSet()
let runningJobIdsWithWrongInteractionId: Set<Int64> = currentlyRunningJobInfo
.filter { _, info -> Bool in info.interactionId != targetInteractionId }
.map { key, _ in key }
.asSet()
possibleJobIds = possibleJobIds
.subtracting(pendingJobIdsWithWrongInteractionId)
.subtracting(runningJobIdsWithWrongInteractionId)
}
// Remove any which don't have the matching details (if provided)
if let targetDetailsData: Data = detailsData {
let pendingJobIdsWithWrongDetailsData: Set<Int64> = pendingJobs
.filter { $0.details != targetDetailsData }
.compactMap { $0.id }
.asSet()
let runningJobIdsWithWrongDetailsData: Set<Int64> = currentlyRunningJobInfo
.filter { _, info -> Bool in info.detailsData != detailsData }
.map { key, _ in key }
.asSet()
possibleJobIds = possibleJobIds
.subtracting(pendingJobIdsWithWrongDetailsData)
.subtracting(runningJobIdsWithWrongDetailsData)
}
return !possibleJobIds.isEmpty
}
fileprivate func removePendingJob(_ jobId: Int64) {
pendingJobsQueue.mutate { queue in
queue = queue.filter { $0.id != jobId }
}
}
// MARK: - Job Running
fileprivate func start(
forceWhenAlreadyRunning: Bool = false,
using dependencies: Dependencies
) {
// Only start if the JobRunner is allowed to start the queue
guard canStart?(self) == true else { return }
guard forceWhenAlreadyRunning || !isRunning.wrappedValue else { return }
// The JobRunner runs synchronously we need to ensure this doesn't start
// on the main thread (if it is on the main thread then swap to a different thread)
guard DispatchQueue.with(key: queueKey, matches: queueContext, using: dependencies) else {
internalQueue.async(using: dependencies) { [weak self] in
self?.start(using: dependencies)
}
return
}
// Flag the JobRunner as running (to prevent something else from trying to start it
// and messing with the execution behaviour)
var wasAlreadyRunning: Bool = false
isRunning.mutate { isRunning in
wasAlreadyRunning = isRunning
isRunning = true
}
hasStartedAtLeastOnce.mutate { $0 = true }
// Get any pending jobs
let jobVariants: [Job.Variant] = self.jobVariants
let jobIdsAlreadyRunning: Set<Int64> = currentlyRunningJobIds.wrappedValue
let jobsAlreadyInQueue: Set<Int64> = pendingJobsQueue.wrappedValue.compactMap { $0.id }.asSet()
let jobsToRun: [Job] = dependencies.storage.read(using: dependencies) { db in
try Job
.filterPendingJobs(
variants: jobVariants,
excludeFutureJobs: true,
includeJobsWithDependencies: false
)
.filter(!jobIdsAlreadyRunning.contains(Job.Columns.id)) // Exclude jobs already running
.filter(!jobsAlreadyInQueue.contains(Job.Columns.id)) // Exclude jobs already in the queue
.fetchAll(db)
}
.defaulting(to: [])
// Determine the number of jobs to run
var jobCount: Int = 0
pendingJobsQueue.mutate { queue in
queue.append(contentsOf: jobsToRun)
jobCount = queue.count
}
// If there are no pending jobs and nothing in the queue then schedule the JobRunner
// to start again when the next scheduled job should start
guard jobCount > 0 else {
if jobIdsAlreadyRunning.isEmpty {
isRunning.mutate { $0 = false }
scheduleNextSoonestJob(using: dependencies)
}
return
}
// Run the first job in the pendingJobsQueue
if !wasAlreadyRunning {
SNLogNotTests("[JobRunner] Starting \(queueContext) with (\(jobCount) job\(jobCount != 1 ? "s" : ""))")
}
runNextJob(using: dependencies)
}
fileprivate func stopAndClearPendingJobs() {
isRunning.mutate { $0 = false }
pendingJobsQueue.mutate { $0 = [] }
deferLoopTracker.mutate { $0 = [:] }
}
private func runNextJob(using dependencies: Dependencies) {
// Ensure the queue is running (if we've stopped the queue then we shouldn't start the next job)
guard isRunning.wrappedValue else { return }
// Ensure this is running on the correct queue
guard DispatchQueue.with(key: queueKey, matches: queueContext, using: dependencies) else {
internalQueue.async(using: dependencies) { [weak self] in
self?.runNextJob(using: dependencies)
}
return
}
guard let (nextJob, numJobsRemaining): (Job, Int) = pendingJobsQueue.mutate({ queue in queue.popFirst().map { ($0, queue.count) } }) else {
// If it's a serial queue, or there are no more jobs running then update the 'isRunning' flag
if executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty {
isRunning.mutate { $0 = false }
}
// Always attempt to schedule the next soonest job (otherwise if enough jobs get started in rapid
// succession then pending/failed jobs in the database may never get re-started in a concurrent queue)
scheduleNextSoonestJob(using: dependencies)
return
}
guard let jobExecutor: JobExecutor.Type = executorMap.wrappedValue[nextJob.variant] else {
SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing executor")
handleJobFailed(
nextJob,
error: JobRunnerError.executorMissing,
permanentFailure: true,
using: dependencies
)
return
}
guard !jobExecutor.requiresThreadId || nextJob.threadId != nil else {
SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing required threadId")
handleJobFailed(
nextJob,
error: JobRunnerError.requiredThreadIdMissing,
permanentFailure: true,
using: dependencies
)
return
}
guard !jobExecutor.requiresInteractionId || nextJob.interactionId != nil else {
SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing required interactionId")
handleJobFailed(
nextJob,
error: JobRunnerError.requiredInteractionIdMissing,
permanentFailure: true,
using: dependencies
)
return
}
guard nextJob.id != nil else {
SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing id")
handleJobFailed(
nextJob,
error: JobRunnerError.jobIdMissing,
permanentFailure: false,
using: dependencies
)
return
}
// If the 'nextRunTimestamp' for the job is in the future then don't run it yet
guard nextJob.nextRunTimestamp <= dependencies.dateNow.timeIntervalSince1970 else {
handleJobDeferred(nextJob, using: dependencies)
return
}
// Check if the next job has any dependencies
let dependencyInfo: (expectedCount: Int, jobs: Set<Job>) = dependencies.storage.read(using: dependencies) { db in
let expectedDependencies: Set<JobDependencies> = try JobDependencies
.filter(JobDependencies.Columns.jobId == nextJob.id)
.fetchSet(db)
let jobDependencies: Set<Job> = try Job
.filter(ids: expectedDependencies.compactMap { $0.dependantId })
.fetchSet(db)
return (expectedDependencies.count, jobDependencies)
}
.defaulting(to: (0, []))
guard dependencyInfo.jobs.count == dependencyInfo.expectedCount else {
SNLog("[JobRunner] \(queueContext) found job with missing dependencies, removing the job")
handleJobFailed(
nextJob,
error: JobRunnerError.missingDependencies,
permanentFailure: true,
using: dependencies
)
return
}
guard dependencyInfo.jobs.isEmpty else {
SNLog("[JobRunner] \(queueContext) found job with \(dependencyInfo.jobs.count) dependencies, running those first")
/// Remove all jobs this one is dependant on that aren't currently running from the queue and re-insert them at the start
/// of the queue
///
/// **Note:** We don't add the current job back the the queue because it should only be re-added if it's dependencies
/// are successfully completed
let currentlyRunningJobIds: [Int64] = Array(currentlyRunningJobIds.wrappedValue)
let dependencyJobsNotCurrentlyRunning: [Job] = dependencyInfo.jobs
.filter { job in !currentlyRunningJobIds.contains(job.id ?? -1) }
.sorted { lhs, rhs in (lhs.id ?? -1) < (rhs.id ?? -1) }
pendingJobsQueue.mutate { queue in
queue = queue
.filter { !dependencyJobsNotCurrentlyRunning.contains($0) }
.inserting(contentsOf: dependencyJobsNotCurrentlyRunning, at: 0)
}
handleJobDeferred(nextJob, using: dependencies)
return
}
// Update the state to indicate the particular job is running
//
// Note: We need to store 'numJobsRemaining' in it's own variable because
// the 'SNLog' seems to dispatch to it's own queue which ends up getting
// blocked by the JobRunner's queue becuase 'jobQueue' is Atomic
var numJobsRunning: Int = 0
nextTrigger.mutate { trigger in
trigger?.invalidate() // Need to invalidate to prevent a memory leak
trigger = nil
}
currentlyRunningJobIds.mutate { currentlyRunningJobIds in
currentlyRunningJobIds = currentlyRunningJobIds.inserting(nextJob.id)
numJobsRunning = currentlyRunningJobIds.count
}
currentlyRunningJobInfo.mutate { currentlyRunningJobInfo in
currentlyRunningJobInfo = currentlyRunningJobInfo.setting(
nextJob.id,
JobRunner.JobInfo(
variant: nextJob.variant,
threadId: nextJob.threadId,
interactionId: nextJob.interactionId,
detailsData: nextJob.details
)
)
}
SNLog("[JobRunner] \(queueContext) started \(nextJob.variant) job (\(executionType == .concurrent ? "\(numJobsRunning) currently running, " : "")\(numJobsRemaining) remaining)")
/// As it turns out Combine doesn't plat too nicely with concurrent Dispatch Queues, in Combine events are dispatched asynchronously to
/// the queue which means an odd situation can occasionally occur where the `finished` event can actually run before the `output`
/// event - this can result in unexpected behaviours (for more information see https://github.com/groue/GRDB.swift/issues/1334)
///
/// Due to this if a job is meant to run on a concurrent queue then we actually want to create a temporary serial queue just for the execution
/// of that job
let targetQueue: DispatchQueue = {
guard executionType == .concurrent else { return internalQueue }
return DispatchQueue(
label: "\(self.queueContext)-serial",
qos: self.qosClass,
attributes: [],
autoreleaseFrequency: .inherit,
target: nil
)
}()
jobExecutor.run(
nextJob,
queue: targetQueue,
success: handleJobSucceeded,
failure: handleJobFailed,
deferred: handleJobDeferred,
using: dependencies
)
// If this queue executes concurrently and there are still jobs remaining then immediately attempt
// to start the next job
if executionType == .concurrent && numJobsRemaining > 0 {
internalQueue.async(using: dependencies) { [weak self] in
self?.runNextJob(using: dependencies)
}
}
}
private func scheduleNextSoonestJob(using dependencies: Dependencies) {
let jobVariants: [Job.Variant] = self.jobVariants
let jobIdsAlreadyRunning: Set<Int64> = currentlyRunningJobIds.wrappedValue
let nextJobTimestamp: TimeInterval? = dependencies.storage.read(using: dependencies) { db in
try Job
.filterPendingJobs(
variants: jobVariants,
excludeFutureJobs: false,
includeJobsWithDependencies: false
)
.select(.nextRunTimestamp)
.filter(!jobIdsAlreadyRunning.contains(Job.Columns.id)) // Exclude jobs already running
.asRequest(of: TimeInterval.self)
.fetchOne(db)
}
// If there are no remaining jobs or the JobRunner isn't allowed to start any queues then trigger
// the 'onQueueDrained' callback and stop
guard let nextJobTimestamp: TimeInterval = nextJobTimestamp, canStart?(self) == true else {
if executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty {
self.onQueueDrained?()
}
return
}
// If the next job isn't scheduled in the future then just restart the JobRunner immediately
let secondsUntilNextJob: TimeInterval = (nextJobTimestamp - dependencies.dateNow.timeIntervalSince1970)
guard secondsUntilNextJob > 0 else {
// Only log that the queue is getting restarted if this queue had actually been about to stop
if executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty {
let timingString: String = (nextJobTimestamp == 0 ?
"that should be in the queue" :
"scheduled \(Int(ceil(abs(secondsUntilNextJob)))) second\(Int(ceil(abs(secondsUntilNextJob))) == 1 ? "" : "s") ago"
)
SNLog("[JobRunner] Restarting \(queueContext) immediately for job \(timingString)")
}
// Trigger the 'start' function to load in any pending jobs that aren't already in the
// queue (for concurrent queues we want to force them to load in pending jobs and add
// them to the queue regardless of whether the queue is already running)
internalQueue.async(using: dependencies) { [weak self] in
self?.start(forceWhenAlreadyRunning: (self?.executionType == .concurrent), using: dependencies)
}
return
}
// Only schedule a trigger if this queue has actually completed
guard executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty else { return }
// Setup a trigger
SNLog("[JobRunner] Stopping \(queueContext) until next job in \(Int(ceil(abs(secondsUntilNextJob)))) second\(Int(ceil(abs(secondsUntilNextJob))) == 1 ? "" : "s")")
nextTrigger.mutate { trigger in
trigger?.invalidate() // Need to invalidate the old trigger to prevent a memory leak
trigger = Trigger.create(queue: self, timestamp: nextJobTimestamp, using: dependencies)
}
}
// MARK: - Handling Results
/// This function is called when a job succeeds
private func handleJobSucceeded(
_ job: Job,
shouldStop: Bool,
using dependencies: Dependencies
) {
/// Retrieve the dependant jobs first (the `JobDependecies` table has cascading deletion when the original `Job` is
/// removed so we need to retrieve these records before that happens)
let dependantJobs: [Job] = dependencies.storage
.read(using: dependencies) { db in try job.dependantJobs.fetchAll(db) }
.defaulting(to: [])
switch job.behaviour {
case .runOnce, .runOnceNextLaunch:
dependencies.storage.write(using: dependencies) { db in
/// Since this job has been completed we can update the dependencies so other job that were dependant
/// on this one can be run
_ = try JobDependencies
.filter(JobDependencies.Columns.dependantId == job.id)
.deleteAll(db)
_ = try job.delete(db)
}
case .recurring where shouldStop == true:
dependencies.storage.write(using: dependencies) { db in
/// Since this job has been completed we can update the dependencies so other job that were dependant
/// on this one can be run
_ = try JobDependencies
.filter(JobDependencies.Columns.dependantId == job.id)
.deleteAll(db)
_ = try job.delete(db)
}
/// For `recurring` jobs which have already run, they should automatically run again but we want at least 1 second
/// to pass before doing so - the job itself should really update it's own `nextRunTimestamp` (this is just a safety net)
case .recurring where job.nextRunTimestamp <= dependencies.dateNow.timeIntervalSince1970:
guard let jobId: Int64 = job.id else { break }
dependencies.storage.write(using: dependencies) { db in
_ = try Job
.filter(id: jobId)
.updateAll(
db,
Job.Columns.failureCount.set(to: 0),
Job.Columns.nextRunTimestamp.set(to: (dependencies.dateNow.timeIntervalSince1970 + 1))
)
}
/// For `recurringOnLaunch/Active` jobs which have already run but failed once, we need to clear their
/// `failureCount` and `nextRunTimestamp` to prevent them from endlessly running over and over again
case .recurringOnLaunch, .recurringOnActive:
guard
let jobId: Int64 = job.id,
job.failureCount != 0 &&
job.nextRunTimestamp > TimeInterval.leastNonzeroMagnitude
else { break }
dependencies.storage.write(using: dependencies) { db in
_ = try Job
.filter(id: jobId)
.updateAll(
db,
Job.Columns.failureCount.set(to: 0),
Job.Columns.nextRunTimestamp.set(to: 0)
)
}
default: break
}
/// Now that the job has been completed we want to insert any jobs that were dependant on it, that aren't already running
/// to the start of the queue (the most likely case is that we want an entire job chain to be completed at the same time rather
/// than being blocked by other unrelated jobs)
///
/// **Note:** If any of these `dependantJobs` have other dependencies then when they attempt to start they will be
/// removed from the queue, replaced by their dependencies
if !dependantJobs.isEmpty {
let currentlyRunningJobIds: [Int64] = Array(currentlyRunningJobIds.wrappedValue)
let dependantJobsNotCurrentlyRunning: [Job] = dependantJobs
.filter { job in !currentlyRunningJobIds.contains(job.id ?? -1) }
.sorted { lhs, rhs in (lhs.id ?? -1) < (rhs.id ?? -1) }
pendingJobsQueue.mutate { queue in
queue = queue
.filter { !dependantJobsNotCurrentlyRunning.contains($0) }
.inserting(contentsOf: dependantJobsNotCurrentlyRunning, at: 0)
}
}
// Perform job cleanup and start the next job
performCleanUp(for: job, result: .succeeded, using: dependencies)
internalQueue.async(using: dependencies) { [weak self] in
self?.runNextJob(using: dependencies)
}
}
/// This function is called when a job fails, if it's wasn't a permanent failure then the 'failureCount' for the job will be incremented and it'll
/// be re-run after a retry interval has passed
private func handleJobFailed(
_ job: Job,
error: Error?,
permanentFailure: Bool,
using dependencies: Dependencies
) {
guard dependencies.storage.read(using: dependencies, { db in try Job.exists(db, id: job.id ?? -1) }) == true else {
SNLog("[JobRunner] \(queueContext) \(job.variant) job canceled")
performCleanUp(for: job, result: .failed, using: dependencies)
internalQueue.async(using: dependencies) { [weak self] in
self?.runNextJob(using: dependencies)
}
return
}
// If this is the blocking queue and a "blocking" job failed then rerun it
// immediately (in this case we don't trigger any job callbacks because the
// job isn't actually done, it's going to try again immediately)
if self.type == .blocking && job.shouldBlock {
SNLog("[JobRunner] \(queueContext) \(job.variant) job failed; retrying immediately")
// If it was a possible deferral loop then we don't actually want to
// retry the job (even if it's a blocking one, this gives a small chance
// that the app could continue to function)
let wasPossibleDeferralLoop: Bool = {
if let error = error, case JobRunnerError.possibleDeferralLoop = error { return true }
return false
}()
performCleanUp(
for: job,
result: .failed,
shouldTriggerCallbacks: wasPossibleDeferralLoop,
using: dependencies
)
// Only add it back to the queue if it wasn't a deferral loop
if !wasPossibleDeferralLoop {
pendingJobsQueue.mutate { $0.insert(job, at: 0) }
}
internalQueue.async(using: dependencies) { [weak self] in
self?.runNextJob(using: dependencies)
}
return
}
// Get the max failure count for the job (a value of '-1' means it will retry indefinitely)
let maxFailureCount: Int = (executorMap.wrappedValue[job.variant]?.maxFailureCount ?? 0)
let nextRunTimestamp: TimeInterval = (dependencies.dateNow.timeIntervalSince1970 + JobRunner.getRetryInterval(for: job))
var dependantJobIds: [Int64] = []
var failureText: String = "failed"
dependencies.storage.write(using: dependencies) { db in
/// Retrieve a list of dependant jobs so we can clear them from the queue
dependantJobIds = try job.dependantJobs
.select(.id)
.asRequest(of: Int64.self)
.fetchAll(db)
/// Delete/update the failed jobs and any dependencies
let updatedFailureCount: UInt = (job.failureCount + 1)
guard
!permanentFailure && (
maxFailureCount < 0 ||
updatedFailureCount <= maxFailureCount
)
else {
failureText = (maxFailureCount >= 0 && updatedFailureCount > maxFailureCount ?
"failed permanently; too many retries" :
"failed permanently"
)
// If the job permanently failed or we have performed all of our retry attempts
// then delete the job and all of it's dependant jobs (it'll probably never succeed)
_ = try job.dependantJobs
.deleteAll(db)
_ = try job.delete(db)
return
}
failureText = "failed; scheduling retry (failure count is \(updatedFailureCount))"
_ = try job
.with(
failureCount: updatedFailureCount,
nextRunTimestamp: nextRunTimestamp
)
.saved(db)
// Update the failureCount and nextRunTimestamp on dependant jobs as well (update the
// 'nextRunTimestamp' value to be 1ms later so when the queue gets regenerated they'll
// come after the dependency)
try job.dependantJobs
.updateAll(
db,
Job.Columns.failureCount.set(to: updatedFailureCount),
Job.Columns.nextRunTimestamp.set(to: (nextRunTimestamp + (1 / 1000)))
)
}
/// Remove any dependant jobs from the queue (shouldn't be in there but filter the queue just in case so we don't try
/// to run a deleted job or get stuck in a loop of trying to run dependencies indefinitely)
if !dependantJobIds.isEmpty {
pendingJobsQueue.mutate { queue in
queue = queue.filter { !dependantJobIds.contains($0.id ?? -1) }
}
}
SNLog("[JobRunner] \(queueContext) \(job.variant) job \(failureText)")
performCleanUp(for: job, result: .failed, using: dependencies)
internalQueue.async(using: dependencies) { [weak self] in
self?.runNextJob(using: dependencies)
}
}
/// This function is called when a job neither succeeds or fails (this should only occur if the job has specific logic that makes it dependant
/// on other jobs, and it should automatically manage those dependencies)
public func handleJobDeferred(
_ job: Job,
using dependencies: Dependencies
) {
var stuckInDeferLoop: Bool = false
deferLoopTracker.mutate {
guard let lastRecord: (count: Int, times: [TimeInterval]) = $0[job.id] else {
$0 = $0.setting(
job.id,
(1, [dependencies.dateNow.timeIntervalSince1970])
)
return
}
let timeNow: TimeInterval = dependencies.dateNow.timeIntervalSince1970
stuckInDeferLoop = (
lastRecord.count >= JobQueue.deferralLoopThreshold &&
(timeNow - lastRecord.times[0]) < CGFloat(lastRecord.count * maxDeferralsPerSecond)
)
$0 = $0.setting(
job.id,
(
lastRecord.count + 1,
// Only store the last 'deferralLoopThreshold' times to ensure we aren't running faster
// than one loop per second
lastRecord.times.suffix(JobQueue.deferralLoopThreshold - 1) + [timeNow]
)
)
}
// It's possible (by introducing bugs) to create a loop where a Job tries to run and immediately
// defers itself but then attempts to run again (resulting in an infinite loop); this won't block
// the app since it's on a background thread but can result in 100% of a CPU being used (and a
// battery drain)
//
// This code will maintain an in-memory store for any jobs which are deferred too quickly (ie.
// more than 'deferralLoopThreshold' times within 'deferralLoopThreshold' seconds)
guard !stuckInDeferLoop else {
deferLoopTracker.mutate { $0 = $0.removingValue(forKey: job.id) }
handleJobFailed(
job,
error: JobRunnerError.possibleDeferralLoop,
permanentFailure: false,
using: dependencies
)
return
}
performCleanUp(for: job, result: .deferred, using: dependencies)
internalQueue.async(using: dependencies) { [weak self] in
self?.runNextJob(using: dependencies)
}
}
private func performCleanUp(
for job: Job,
result: JobRunner.JobResult,
shouldTriggerCallbacks: Bool = true,
using dependencies: Dependencies
) {
// The job is removed from the queue before it runs so all we need to to is remove it
// from the 'currentlyRunning' set
currentlyRunningJobIds.mutate { $0 = $0.removing(job.id) }
currentlyRunningJobInfo.mutate { $0 = $0.removingValue(forKey: job.id) }
guard shouldTriggerCallbacks else { return }
// Run any job callbacks now that it's done
var jobCallbacksToRun: [(JobRunner.JobResult) -> ()] = []
jobCallbacks.mutate { jobCallbacks in
jobCallbacksToRun = (jobCallbacks[job.id] ?? [])
jobCallbacks = jobCallbacks.removingValue(forKey: job.id)
}
DispatchQueue.global(qos: .default).async(using: dependencies) {
jobCallbacksToRun.forEach { $0(result) }
}
}
}
// MARK: - JobRunner Singleton
// FIXME: Remove this once the jobRunner is dependency injected everywhere correctly
public extension JobRunner {
internal static let instance: JobRunner = JobRunner()
// MARK: - Static Access
static func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant) {
instance.setExecutor(executor, for: variant)
}
static func appDidFinishLaunching(using dependencies: Dependencies = Dependencies()) {
instance.appDidFinishLaunching(using: dependencies)
}
static func appDidBecomeActive(using dependencies: Dependencies = Dependencies()) {
instance.appDidBecomeActive(using: dependencies)
}
static func afterBlockingQueue(callback: @escaping () -> ()) {
instance.afterBlockingQueue(callback: callback)
}
/// Add a job onto the queue, if the queue isn't currently running and 'canStartJob' is true then this will start
/// the JobRunner
///
/// **Note:** If the job has a `behaviour` of `runOnceNextLaunch` or the `nextRunTimestamp`
/// is in the future then the job won't be started
static func add(
_ db: Database,
job: Job?,
canStartJob: Bool = true,
using dependencies: Dependencies = Dependencies()
) { instance.add(db, job: job, canStartJob: canStartJob, using: dependencies) }
/// Upsert a job onto the queue, if the queue isn't currently running and 'canStartJob' is true then this will start
/// the JobRunner
///
/// **Note:** If the job has a `behaviour` of `runOnceNextLaunch` or the `nextRunTimestamp`
/// is in the future then the job won't be started
static func upsert(
_ db: Database,
job: Job?,
canStartJob: Bool = true,
using dependencies: Dependencies = Dependencies()
) { instance.upsert(db, job: job, canStartJob: canStartJob, using: dependencies) }
@discardableResult static func insert(
_ db: Database,
job: Job?,
before otherJob: Job
) -> (Int64, Job)? { instance.insert(db, job: job, before: otherJob) }
/// Calling this will clear the JobRunner queues and stop it from running new jobs, any currently executing jobs will continue to run
/// though (this means if we suspend the database it's likely that any currently running jobs will fail to complete and fail to record their
/// failure - they _should_ be picked up again the next time the app is launched)
static func stopAndClearPendingJobs(
exceptForVariant: Job.Variant? = nil,
onComplete: (() -> ())? = nil
) { instance.stopAndClearPendingJobs(exceptForVariant: exceptForVariant, onComplete: onComplete) }
static func isCurrentlyRunning(_ job: Job?) -> Bool {
return instance.isCurrentlyRunning(job)
}
static func afterCurrentlyRunningJob(_ job: Job?, callback: @escaping (JobResult) -> ()) {
instance.afterCurrentlyRunningJob(job, callback: callback)
}
static func removePendingJob(_ job: Job?) {
instance.removePendingJob(job)
}
}