2018-10-20 19:51:48 +02:00
//
// C o p y r i g h t ( c ) 2 0 1 8 O p e n W h i s p e r S y s t e m s . A l l r i g h t s r e s e r v e d .
//
import Foundation
// / J o b Q u e u e - A d u r a b l e w o r k q u e u e
// /
// / W h e n w o r k n e e d s t o b e d o n e , a d d i t t o t h e J o b Q u e u e .
// / T h e J o b Q u e u e w i l l p e r s i s t a J o b R e c o r d t o b e s u r e t h a t w o r k c a n b e r e s t a r t e d i f t h e a p p i s k i l l e d .
// /
2018-10-23 17:37:30 +02:00
// / T h e a c t u a l w o r k , i s c a r r i e d o u t i n a D u r a b l e O p e r a t i o n w h i c h t h e J o b Q u e u e s p i n s o f f , b a s e d o n t h e c o n t e n t s
2018-10-20 19:51:48 +02:00
// / o f a J o b R e c o r d .
// /
2018-10-23 17:37:30 +02:00
// / F o r a c o n c r e t e e x a m p l e , t a k e m e s s a g e s e n d i n g .
// / A d d a n o u t g o i n g m e s s a g e t o t h e M e s s a g e S e n d e r J o b Q u e u e , w h i c h f i r s t r e c o r d s a S S K M e s s a g e S e n d e r J o b R e c o r d .
// / T h e M e s s a g e S e n d e r J o b Q u e u e t h e n u s e s t h a t S S K M e s s a g e S e n d e r J o b R e c o r d t o c r e a t e a M e s s a g e S e n d e r O p e r a t i o n w h i c h
2018-10-20 19:51:48 +02:00
// / t a k e s c a r e o f t h e a c t u a l b u s i n e s s o f c o m m u n i c a t i n g w i t h t h e s e r v i c e .
2018-10-23 17:37:30 +02:00
// /
// / D u r a b l e O p e r a t i o n s a r e r e t r y a b l e - v i a t h e i r ` r e m a i n i n g R e t r i e s ` l o g i c . H o w e v e r , i f t h e o p e r a t i o n e n c o u n t e r s
// / a n e r r o r w h e r e ` e r r o r . i s R e t r y a b l e = = f a l s e ` , t h e o p e r a t i o n w i l l f a i l , r e g a r d l e s s o f a v a i l a b l e r e t r i e s .
public extension Error {
var isRetryable : Bool {
return ( self as NSError ) . isRetryable
}
}
2018-10-20 19:51:48 +02:00
2018-11-11 03:27:28 +01:00
extension SSKJobRecordStatus : CustomStringConvertible {
public var description : String {
switch self {
case . ready :
return " ready "
case . unknown :
return " unknown "
case . running :
return " running "
case . permanentlyFailed :
return " permanentlyFailed "
case . obsolete :
return " obsolete "
}
}
}
2018-10-20 19:51:48 +02:00
public enum JobError : Error {
case assertionFailure ( description : String )
case obsolete ( description : String )
}
public protocol DurableOperation : class {
associatedtype JobRecordType : SSKJobRecord
associatedtype DurableOperationDelegateType : DurableOperationDelegate
var jobRecord : JobRecordType { get }
var durableOperationDelegate : DurableOperationDelegateType ? { get set }
2018-10-22 18:53:14 +02:00
var operation : OWSOperation { get }
2018-10-20 19:51:48 +02:00
var remainingRetries : UInt { get set }
}
public protocol DurableOperationDelegate : class {
associatedtype DurableOperationType : DurableOperation
func durableOperationDidSucceed ( _ operation : DurableOperationType , transaction : YapDatabaseReadWriteTransaction )
func durableOperation ( _ operation : DurableOperationType , didReportError : Error , transaction : YapDatabaseReadWriteTransaction )
func durableOperation ( _ operation : DurableOperationType , didFailWithError error : Error , transaction : YapDatabaseReadWriteTransaction )
}
public protocol JobQueue : DurableOperationDelegate {
typealias DurableOperationDelegateType = Self
typealias JobRecordType = DurableOperationType . JobRecordType
// MARK: D e p e n d e n c i e s
var dbConnection : YapDatabaseConnection { get }
var finder : JobRecordFinder { get }
// MARK: D e f a u l t I m p l e m e n t a t i o n s
func add ( jobRecord : JobRecordType , transaction : YapDatabaseReadWriteTransaction )
func restartOldJobs ( )
func workStep ( )
func defaultSetup ( )
// MARK: R e q u i r e d
2018-10-22 18:53:14 +02:00
var runningOperations : [ DurableOperationType ] { get set }
2018-10-20 19:51:48 +02:00
var jobRecordLabel : String { get }
2018-10-22 17:49:45 +02:00
var isSetup : Bool { get set }
2018-10-20 19:51:48 +02:00
func setup ( )
func didMarkAsReady ( oldJobRecord : JobRecordType , transaction : YapDatabaseReadWriteTransaction )
func operationQueue ( jobRecord : JobRecordType ) -> OperationQueue
func buildOperation ( jobRecord : JobRecordType , transaction : YapDatabaseReadTransaction ) throws -> DurableOperationType
2018-10-22 18:53:14 +02:00
// / W h e n ` r e q u i r e s I n t e r n e t ` i s t r u e , w e i m m e d i a t e l y r u n a n y j o b s w h i c h a r e w a i t i n g f o r r e t r y u p o n d e t e c t i n g R e a c h a b i l i t y .
// /
// / B e c a u s e ` R e a c h a b i l i t y ` i s n ' t 1 0 0 % r e l i a b l e , t h e j o b s w i l l b e a t t e m p t e d r e g a r d l e s s o f w h a t w e t h i n k o u r c u r r e n t R e a c h a b i l i t y i s .
// / H o w e v e r , b e c a u s e t h e s e j o b s w i l l l i k e l y f a i l m a n y t i m e s i n s u c c e s s i o n , t h e i r ` r e t r y I n t e r v a l ` c o u l d b e q u i t e l o n g b y t h e t i m e w e
// / a r e b a c k o n l i n e .
var requiresInternet : Bool { get }
2018-10-20 19:51:48 +02:00
static var maxRetries : UInt { get }
}
public extension JobQueue {
2018-11-02 19:13:54 +01:00
// MARK: D e p e n d e n c i e s
2018-10-20 19:51:48 +02:00
var dbConnection : YapDatabaseConnection {
return SSKEnvironment . shared . primaryStorage . dbReadWriteConnection
}
var finder : JobRecordFinder {
return JobRecordFinder ( )
}
2018-10-22 18:53:14 +02:00
var reachabilityManager : SSKReachabilityManager {
return SSKEnvironment . shared . reachabilityManager
}
2018-10-20 19:51:48 +02:00
// MARK:
func add ( jobRecord : JobRecordType , transaction : YapDatabaseReadWriteTransaction ) {
assert ( jobRecord . status = = . ready )
2018-11-05 15:16:17 +01:00
2018-10-20 19:51:48 +02:00
jobRecord . save ( with : transaction )
2020-06-11 13:14:22 +02:00
transaction . addCompletionQueue ( DispatchQueue . global ( ) ) {
2018-11-10 20:28:57 +01:00
self . startWorkWhenAppIsReady ( )
}
}
func startWorkWhenAppIsReady ( ) {
guard ! CurrentAppContext ( ) . isRunningTests else {
2020-06-11 13:14:22 +02:00
DispatchQueue . global ( ) . async {
2018-11-10 20:28:57 +01:00
self . workStep ( )
}
return
}
AppReadiness . runNowOrWhenAppDidBecomeReady {
2020-06-11 13:14:22 +02:00
DispatchQueue . global ( ) . async {
2018-11-10 20:28:57 +01:00
self . workStep ( )
2018-11-02 19:13:54 +01:00
}
2018-10-20 19:51:48 +02:00
}
}
func workStep ( ) {
Logger . debug ( " " )
2018-10-22 17:49:45 +02:00
guard isSetup else {
2018-10-20 19:51:48 +02:00
if ! CurrentAppContext ( ) . isRunningTests {
2018-10-22 17:49:45 +02:00
owsFailDebug ( " not setup " )
2018-10-20 19:51:48 +02:00
}
return
}
2020-11-04 01:46:30 +01:00
Storage . writeSync { transaction in
2018-10-20 19:51:48 +02:00
guard let nextJob : JobRecordType = self . finder . getNextReady ( label : self . jobRecordLabel , transaction : transaction ) as ? JobRecordType else {
Logger . verbose ( " nothing left to enqueue " )
return
}
do {
try nextJob . saveAsStarted ( transaction : transaction )
let operationQueue = self . operationQueue ( jobRecord : nextJob )
let durableOperation = try self . buildOperation ( jobRecord : nextJob , transaction : transaction )
durableOperation . durableOperationDelegate = self as ? Self . DurableOperationType . DurableOperationDelegateType
assert ( durableOperation . durableOperationDelegate != nil )
let remainingRetries = self . remainingRetries ( durableOperation : durableOperation )
durableOperation . remainingRetries = remainingRetries
2018-10-22 18:53:14 +02:00
self . runningOperations . append ( durableOperation )
2018-10-20 19:51:48 +02:00
Logger . debug ( " adding operation: \( durableOperation ) with remainingRetries: \( remainingRetries ) " )
operationQueue . addOperation ( durableOperation . operation )
} catch JobError . assertionFailure ( let description ) {
owsFailDebug ( " assertion failure: \( description ) " )
nextJob . saveAsPermanentlyFailed ( transaction : transaction )
} catch JobError . obsolete ( let description ) {
// T O D O i s t h i s e v e n w o r t h w h i l e t o h a v e o b s o l e t e s t a t e ? S h o u l d w e j u s t d e l e t e t h e t a s k o u t r i g h t ?
Logger . verbose ( " marking obsolete task as such. description: \( description ) " )
nextJob . saveAsObsolete ( transaction : transaction )
} catch {
owsFailDebug ( " unexpected error " )
}
2020-06-11 13:14:22 +02:00
DispatchQueue . global ( ) . async {
2018-10-20 19:51:48 +02:00
self . workStep ( )
}
}
}
public func restartOldJobs ( ) {
2020-11-04 01:46:30 +01:00
Storage . writeSync { transaction in
2018-10-20 19:51:48 +02:00
let runningRecords = self . finder . allRecords ( label : self . jobRecordLabel , status : . running , transaction : transaction )
Logger . info ( " marking old `running` JobRecords as ready: \( runningRecords . count ) " )
for record in runningRecords {
guard let jobRecord = record as ? JobRecordType else {
owsFailDebug ( " unexpectred jobRecord: \( record ) " )
continue
}
do {
try jobRecord . saveRunningAsReady ( transaction : transaction )
self . didMarkAsReady ( oldJobRecord : jobRecord , transaction : transaction )
} catch {
owsFailDebug ( " failed to mark old running records as ready error: \( error ) " )
jobRecord . saveAsPermanentlyFailed ( transaction : transaction )
}
}
}
}
// / U n l e s s y o u n e e d s p e c i a l h a n d l i n g , y o u r s e t u p m e t h o d c a n b e a s s i m p l e a s
// /
// / f u n c s e t u p ( ) {
// / d e f a u l t S e t u p ( )
// / }
// /
// / S o y o u m i g h t a s k , w h y n o t j u s t r e n a m e t h i s m e t h o d t o ` s e t u p ` ? B e c a u s e
// / ` s e t u p ` i s c a l l e d f r o m o b j c , a n d d e f a u l t i m p l e m e n t a t i o n s f r o m a p r o t o c o l
// / c a n n o t b e m a r k e d a s @ o b j c .
func defaultSetup ( ) {
2018-10-22 17:49:45 +02:00
guard ! isSetup else {
2018-10-20 19:51:48 +02:00
owsFailDebug ( " already ready already " )
return
}
self . restartOldJobs ( )
2018-10-22 18:53:14 +02:00
if self . requiresInternet {
NotificationCenter . default . addObserver ( forName : . reachabilityChanged ,
object : self . reachabilityManager . observationContext ,
queue : nil ) { _ in
if self . reachabilityManager . isReachable {
Logger . verbose ( " isReachable: true " )
self . becameReachable ( )
} else {
Logger . verbose ( " isReachable: false " )
}
}
}
2018-10-22 17:49:45 +02:00
self . isSetup = true
2018-11-10 20:28:57 +01:00
self . startWorkWhenAppIsReady ( )
2018-10-20 19:51:48 +02:00
}
func remainingRetries ( durableOperation : DurableOperationType ) -> UInt {
let maxRetries = type ( of : self ) . maxRetries
let failureCount = durableOperation . jobRecord . failureCount
guard maxRetries > failureCount else {
return 0
}
return maxRetries - failureCount
}
2018-10-22 18:53:14 +02:00
func becameReachable ( ) {
guard requiresInternet else {
2018-10-25 20:17:58 +02:00
owsFailDebug ( " should only be called if `requiresInternet` is true " )
2018-10-22 18:53:14 +02:00
return
}
2018-11-11 03:27:28 +01:00
_ = self . runAnyQueuedRetry ( )
}
func runAnyQueuedRetry ( ) -> DurableOperationType ? {
guard let runningDurableOperation = self . runningOperations . first else {
return nil
}
runningDurableOperation . operation . runAnyQueuedRetry ( )
return runningDurableOperation
2018-10-22 18:53:14 +02:00
}
2018-10-20 19:51:48 +02:00
// MARK: D u r a b l e O p e r a t i o n D e l e g a t e
func durableOperationDidSucceed ( _ operation : DurableOperationType , transaction : YapDatabaseReadWriteTransaction ) {
2018-10-22 18:53:14 +02:00
self . runningOperations = self . runningOperations . filter { $0 !== operation }
2018-10-20 19:51:48 +02:00
operation . jobRecord . remove ( with : transaction )
}
func durableOperation ( _ operation : DurableOperationType , didReportError : Error , transaction : YapDatabaseReadWriteTransaction ) {
do {
try operation . jobRecord . addFailure ( transaction : transaction )
} catch {
owsFailDebug ( " error while addingFailure: \( error ) " )
operation . jobRecord . saveAsPermanentlyFailed ( transaction : transaction )
}
}
func durableOperation ( _ operation : DurableOperationType , didFailWithError error : Error , transaction : YapDatabaseReadWriteTransaction ) {
2018-10-22 18:53:14 +02:00
self . runningOperations = self . runningOperations . filter { $0 !== operation }
2018-10-20 19:51:48 +02:00
operation . jobRecord . saveAsPermanentlyFailed ( transaction : transaction )
}
}
@objc ( SSKJobRecordFinder )
public class JobRecordFinder : NSObject , Finder {
typealias ExtensionType = YapDatabaseSecondaryIndex
typealias TransactionType = YapDatabaseSecondaryIndexTransaction
enum JobRecordField : String {
case status , label , sortId
}
func getNextReady ( label : String , transaction : YapDatabaseReadTransaction ) -> SSKJobRecord ? {
var result : SSKJobRecord ?
self . enumerateJobRecords ( label : label , status : . ready , transaction : transaction ) { jobRecord , stopPointer in
result = jobRecord
stopPointer . pointee = true
}
return result
}
func allRecords ( label : String , status : SSKJobRecordStatus , transaction : YapDatabaseReadTransaction ) -> [ SSKJobRecord ] {
var result : [ SSKJobRecord ] = [ ]
2018-11-02 19:13:54 +01:00
self . enumerateJobRecords ( label : label , status : status , transaction : transaction ) { jobRecord , _ in
2018-10-20 19:51:48 +02:00
result . append ( jobRecord )
}
return result
}
func enumerateJobRecords ( label : String , status : SSKJobRecordStatus , transaction : YapDatabaseReadTransaction , block : @ escaping ( SSKJobRecord , UnsafeMutablePointer < ObjCBool > ) -> Void ) {
let queryFormat = String ( format : " WHERE %@ = ? AND %@ = ? ORDER BY %@ " , JobRecordField . status . rawValue , JobRecordField . label . rawValue , JobRecordField . sortId . rawValue )
let query = YapDatabaseQuery ( string : queryFormat , parameters : [ status . rawValue , label ] )
2018-11-02 19:13:54 +01:00
self . ext ( transaction : transaction ) . enumerateKeysAndObjects ( matching : query ) { _ , _ , object , stopPointer in
2018-10-20 19:51:48 +02:00
guard let jobRecord = object as ? SSKJobRecord else {
owsFailDebug ( " expecting jobRecord but found: \( object ) " )
return
}
block ( jobRecord , stopPointer )
}
}
static var dbExtensionName : String {
return " SecondaryIndexJobRecord "
}
@objc
public class func asyncRegisterDatabaseExtensionObjC ( storage : OWSStorage ) {
asyncRegisterDatabaseExtension ( storage : storage )
}
static var dbExtensionConfig : YapDatabaseSecondaryIndex {
let setup = YapDatabaseSecondaryIndexSetup ( )
setup . addColumn ( JobRecordField . sortId . rawValue , with : . integer )
setup . addColumn ( JobRecordField . status . rawValue , with : . integer )
setup . addColumn ( JobRecordField . label . rawValue , with : . text )
let block : YapDatabaseSecondaryIndexWithObjectBlock = { transaction , dict , collection , key , object in
guard let jobRecord = object as ? SSKJobRecord else {
return
}
dict [ JobRecordField . sortId . rawValue ] = jobRecord . sortId
dict [ JobRecordField . status . rawValue ] = jobRecord . status . rawValue
dict [ JobRecordField . label . rawValue ] = jobRecord . label
}
let handler = YapDatabaseSecondaryIndexHandler . withObjectBlock ( block )
let options = YapDatabaseSecondaryIndexOptions ( )
let whitelist = YapWhitelistBlacklist ( whitelist : Set ( [ SSKJobRecord . collection ( ) ] ) )
options . allowedCollections = whitelist
return YapDatabaseSecondaryIndex . init ( setup : setup , handler : handler , versionTag : " 2 " , options : options )
}
}
protocol Finder {
associatedtype ExtensionType : YapDatabaseExtension
associatedtype TransactionType : YapDatabaseExtensionTransaction
static var dbExtensionName : String { get }
static var dbExtensionConfig : ExtensionType { get }
func ext ( transaction : YapDatabaseReadTransaction ) -> TransactionType
static func asyncRegisterDatabaseExtension ( storage : OWSStorage )
static func testingOnly_ensureDatabaseExtensionRegistered ( storage : OWSStorage )
}
extension Finder {
func ext ( transaction : YapDatabaseReadTransaction ) -> TransactionType {
return transaction . ext ( type ( of : self ) . dbExtensionName ) as ! TransactionType
}
static func asyncRegisterDatabaseExtension ( storage : OWSStorage ) {
storage . asyncRegister ( dbExtensionConfig , withName : dbExtensionName )
}
// O n l y f o r t e s t i n g .
static func testingOnly_ensureDatabaseExtensionRegistered ( storage : OWSStorage ) {
guard storage . registeredExtension ( dbExtensionName ) = = nil else {
return
}
storage . register ( dbExtensionConfig , withName : dbExtensionName )
}
}