Class: Procrastinator::TaskMetaData

Inherits:
Object
  • Object
show all
Defined in:
lib/procrastinator/task_meta_data.rb

Overview

TaskMetaData objects are State Patterns that record information about the work done on a particular task.

It contains the specific information needed to run a task instance. Users define a task handler class, which describes the “how” of a task and TaskMetaData represents the “what” and “when”.

It contains task-specific data, timing information, and error records.

All of its state is read-only.

Author:

  • Robin Miller

Constant Summary collapse

EXPECTED_DATA =

These are the attributes expected to be in the persistence mechanism

[:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(id: nil, queue: nil, data: nil, run_at: nil, initial_run_at: nil, expire_at: nil, attempts: 0, last_error: nil, last_fail_at: nil) ⇒ TaskMetaData

Returns a new instance of TaskMetaData.



39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/procrastinator/task_meta_data.rb', line 39

def initialize(id: nil, queue: nil, data: nil,
               run_at: nil, initial_run_at: nil, expire_at: nil,
               attempts: 0, last_error: nil, last_fail_at: nil)
   @id             = id
   @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
   @run_at         = get_time(run_at)
   @initial_run_at = get_time(initial_run_at) || @run_at
   @expire_at      = get_time(expire_at)
   @attempts       = (attempts || 0).to_i
   @last_error     = last_error
   @last_fail_at   = get_time(last_fail_at)
   @data           = data ? JSON.parse(data, symbolize_names: true) : nil
end

Instance Attribute Details

#:attempts(: attempts) ⇒ Integer (readonly)

Returns The number of times this task has been attempted.

Returns:

  • (Integer) —

    The number of times this task has been attempted



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

#:data(: data) ⇒ String (readonly)

Returns App-provided JSON data.

Returns:

  • (String) —

    App-provided JSON data



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

#:expire_at(: expire_at) ⇒ Integer (readonly)

Returns Linux epoch timestamp of when to consider this task obsolete.

Returns:

  • (Integer) —

    Linux epoch timestamp of when to consider this task obsolete



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

#:id(: id) ⇒ Integer (readonly)

Returns the unique identifier for this task.

Returns:

  • (Integer) —

    the unique identifier for this task



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

#:initial_run_at(: initial_run_at) ⇒ Integer (readonly)

Returns Linux epoch timestamp of the original value for run_at.

Returns:

  • (Integer) —

    Linux epoch timestamp of the original value for run_at



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

#:last_error(: last_error) ⇒ String (readonly)

Returns The message and stack trace of the error encountered on the most recent failed attempt.

Returns:

  • (String) —

    The message and stack trace of the error encountered on the most recent failed attempt



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

#:last_fail_at(: last_fail_at) ⇒ Integer (readonly)

Returns Linux epoch timestamp of when the last_error was recorded.

Returns:

  • (Integer) —

    Linux epoch timestamp of when the last_error was recorded



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

#:run_at(: run_at) ⇒ Integer (readonly)

Returns Linux epoch timestamp of when to attempt this task next.

Returns:

  • (Integer) —

    Linux epoch timestamp of when to attempt this task next



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/procrastinator/task_meta_data.rb', line 33

class TaskMetaData
   # These are the attributes expected to be in the persistence mechanism
   EXPECTED_DATA = [:id, :run_at, :initial_run_at, :expire_at, :attempts, :last_error, :last_fail_at, :data].freeze

   attr_reader(*EXPECTED_DATA, :queue)

   def initialize(id: nil, queue: nil, data: nil,
                  run_at: nil, initial_run_at: nil, expire_at: nil,
                  attempts: 0, last_error: nil, last_fail_at: nil)
      @id             = id
      @queue          = queue || raise(ArgumentError, 'queue cannot be nil')
      @run_at         = get_time(run_at)
      @initial_run_at = get_time(initial_run_at) || @run_at
      @expire_at      = get_time(expire_at)
      @attempts       = (attempts || 0).to_i
      @last_error     = last_error
      @last_fail_at   = get_time(last_fail_at)
      @data           = data ? JSON.parse(data, symbolize_names: true) : nil
   end

   # Increases the number of attempts on this task by one, unless the limit has been reached.
   #
   # @raise [Task::AttemptsExhaustedError] when the number of attempts has exceeded the Queue's defined maximum.
   def add_attempt
      raise Task::AttemptsExhaustedError unless attempts_left?

      @attempts += 1
   end

   # Records a failure on this task
   #
   # @param error [StandardError] The error to record
   def failure(error)
      @last_fail_at = Time.now
      @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

      if retryable?
         reschedule
         :fail
      else
         @run_at = nil
         :final_fail
      end
   end

   # @return [Boolean] whether the task has attempts left and is not expired
   def retryable?
      attempts_left? && !expired?
   end

   # @return [Boolean] whether the task is expired
   def expired?
      !@expire_at.nil? && @expire_at < Time.now
   end

   # @return [Boolean] whether there are attempts left until the Queue's defined maximum is reached (if any)
   def attempts_left?
      @queue.max_attempts.nil? || @attempts < @queue.max_attempts
   end

   # @return [Boolean] whether the task's run_at is exceeded
   def runnable?
      !@run_at.nil? && @run_at <= Time.now
   end

   # @return [Boolean] whether the task's last execution completed successfully.
   # @raise [RuntimeError] when the task has not been attempted yet or when it is expired
   def successful?
      raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

      !expired? && @last_error.nil? && @last_fail_at.nil?
   end

   # Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling
   # calculation algorithm.
   #
   # @param run_at - the new time to run this task
   # @param expire_at - the new time to expire this task
   def reschedule(run_at: nil, expire_at: nil)
      validate_run_at(run_at, expire_at)

      @expire_at = expire_at if expire_at

      if run_at
         @run_at = @initial_run_at = get_time(run_at)
         clear_fails
         @attempts = 0
      end

      return if run_at || expire_at

      # (30 + n_attempts^4) seconds is chosen to rapidly expand
      # but with the baseline of 30s to avoid hitting the disk too frequently.
      @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
   end

   # @return [Hash] representation of the task metadata as a hash
   def to_h
      {id:             @id,
       queue:          @queue.name.to_s,
       run_at:         @run_at,
       initial_run_at: @initial_run_at,
       expire_at:      @expire_at,
       attempts:       @attempts,
       last_fail_at:   @last_fail_at,
       last_error:     @last_error,
       data:           serialized_data}
   end

   # @return [String] :data serialized as a JSON string
   def serialized_data
      JSON.dump(@data)
   end

   # Resets the last failure time and error.
   def clear_fails
      @last_error   = nil
      @last_fail_at = nil
   end

   private

   def get_time(data)
      case data
      when NilClass
         nil
      when Numeric
         Time.at data
      when String
         Time.parse data
      when Time
         data
      else
         return data.to_time if data.respond_to? :to_time

         raise ArgumentError, "Unknown data type: #{ data.class } (#{ data })"
      end
   end

   def validate_run_at(run_at, expire_at)
      return unless run_at

      if expire_at && run_at > expire_at
         raise ArgumentError, "new run_at (#{ run_at }) is later than new expire_at (#{ expire_at })"
      end

      return unless @expire_at && run_at > @expire_at

      raise ArgumentError, "new run_at (#{ run_at }) is later than existing expire_at (#{ @expire_at })"
   end
end

Instance Method Details

#add_attempt ⇒ Object

Increases the number of attempts on this task by one, unless the limit has been reached.

Raises:



56
57
58
59
60
 
# File 'lib/procrastinator/task_meta_data.rb', line 56

def add_attempt
   raise Task::AttemptsExhaustedError unless attempts_left?

   @attempts += 1
end

#attempts_left? ⇒ Boolean

Returns whether there are attempts left until the Queue’s defined maximum is reached (if any).

Returns:

  • (Boolean) —

    whether there are attempts left until the Queue’s defined maximum is reached (if any)



89
90
91
 
# File 'lib/procrastinator/task_meta_data.rb', line 89

def attempts_left?
   @queue.max_attempts.nil? || @attempts < @queue.max_attempts
end

#clear_fails ⇒ Object

Resets the last failure time and error.



148
149
150
151
 
# File 'lib/procrastinator/task_meta_data.rb', line 148

def clear_fails
   @last_error   = nil
   @last_fail_at = nil
end

#expired? ⇒ Boolean

Returns whether the task is expired.

Returns:

  • (Boolean) —

    whether the task is expired



84
85
86
 
# File 'lib/procrastinator/task_meta_data.rb', line 84

def expired?
   !@expire_at.nil? && @expire_at < Time.now
end

#failure(error) ⇒ Object

Records a failure on this task

Parameters:

  • error (StandardError) —

    The error to record



65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/procrastinator/task_meta_data.rb', line 65

def failure(error)
   @last_fail_at = Time.now
   @last_error   = %[Task failed: #{ error.message }\n#{ error.backtrace&.join("\n") }]

   if retryable?
      reschedule
      :fail
   else
      @run_at = nil
      :final_fail
   end
end

#reschedule(run_at: nil, expire_at: nil) ⇒ Object

Updates the run and/or expiry time. If neither is provided, will reschedule based on the rescheduling calculation algorithm.

Parameters:

  • run_at (defaults to: nil) —

    • the new time to run this task

  • expire_at (defaults to: nil) —

    • the new time to expire this task



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/procrastinator/task_meta_data.rb', line 111

def reschedule(run_at: nil, expire_at: nil)
   validate_run_at(run_at, expire_at)

   @expire_at = expire_at if expire_at

   if run_at
      @run_at = @initial_run_at = get_time(run_at)
      clear_fails
      @attempts = 0
   end

   return if run_at || expire_at

   # (30 + n_attempts^4) seconds is chosen to rapidly expand
   # but with the baseline of 30s to avoid hitting the disk too frequently.
   @run_at += 30 + (@attempts ** 4) unless @run_at.nil?
end

#retryable? ⇒ Boolean

Returns whether the task has attempts left and is not expired.

Returns:

  • (Boolean) —

    whether the task has attempts left and is not expired



79
80
81
 
# File 'lib/procrastinator/task_meta_data.rb', line 79

def retryable?
   attempts_left? && !expired?
end

#runnable? ⇒ Boolean

Returns whether the task’s run_at is exceeded.

Returns:

  • (Boolean) —

    whether the task’s run_at is exceeded



94
95
96
 
# File 'lib/procrastinator/task_meta_data.rb', line 94

def runnable?
   !@run_at.nil? && @run_at <= Time.now
end

#serialized_data ⇒ String

Returns :data serialized as a JSON string.

Returns:

  • (String) —

    :data serialized as a JSON string



143
144
145
 
# File 'lib/procrastinator/task_meta_data.rb', line 143

def serialized_data
   JSON.dump(@data)
end

#successful? ⇒ Boolean

Returns whether the task’s last execution completed successfully.

Returns:

  • (Boolean) —

    whether the task’s last execution completed successfully.

Raises:

  • (RuntimeError) —

    when the task has not been attempted yet or when it is expired



100
101
102
103
104
 
# File 'lib/procrastinator/task_meta_data.rb', line 100

def successful?
   raise 'you cannot check for success before running #work' if !expired? && @attempts <= 0

   !expired? && @last_error.nil? && @last_fail_at.nil?
end

#to_h ⇒ Hash

Returns representation of the task metadata as a hash.

Returns:

  • (Hash) —

    representation of the task metadata as a hash



130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/procrastinator/task_meta_data.rb', line 130

def to_h
   {id:             @id,
    queue:          @queue.name.to_s,
    run_at:         @run_at,
    initial_run_at: @initial_run_at,
    expire_at:      @expire_at,
    attempts:       @attempts,
    last_fail_at:   @last_fail_at,
    last_error:     @last_error,
    data:           serialized_data}
end