lib/tobox/fetcher.rb in tobox-0.5.2 vs lib/tobox/fetcher.rb in tobox-0.6.0

@@ -91,13 +91,14 @@
 
     def with_events(event_ids, &blk)
       events, error = yield_events(event_ids, &blk)
 
       events.each do |event|
-        if error
-          event.merge!(mark_as_error(event, error))
-          handle_error_event(event, error)
+        event_error = error || event[:error]
+        if event_error
+          event.merge!(mark_as_error(event, event_error))
+          handle_error_event(event, event_error)
         else
           handle_after_event(event)
         end
       end
     end
@@ -107,13 +108,32 @@
       events = error = nil
 
       begin
         events = events_ds.all
 
-        yield events
+        unless events.empty?
+          errors_by_id = catch(:tobox_batch_errors) do
+            yield events
+            nil
+          end
 
-        events_ds.delete unless events.empty?
+          # some events from batch errored
+          if errors_by_id
+            failed = events.values_at(*errors_by_id.keys)
+            successful = events - failed
+
+            # fill in with batch error
+            failed.each do |ev|
+              ev[:error] = errors_by_id[events.index(ev)]
+            end
+
+            # delete successful
+            @ds.where(id: successful.map { |ev| ev[:id] }).delete unless successful.empty?
+          else
+            events_ds.delete
+          end
+        end
       rescue StandardError => e
         error = e
       end
 
       [events, error]
@@ -127,10 +147,10 @@
       update_params = {
         run_at: Sequel.date_add(Sequel::CURRENT_TIMESTAMP,
                                 seconds: @exponential_retry_factor**(event[:attempts] - 1)),
         # run_at: Sequel.date_add(Sequel::CURRENT_TIMESTAMP,
         #                         seconds: Sequel.function(:POWER, Sequel[@table][:attempts] + 1,  4)),
-        last_error: "#{error.message}\n#{error.backtrace.join("\n")}"
+        last_error: error.full_message(highlight: false)
       }
 
       set_event_retry_attempts(event, update_params)
     end