core: Add a backoff for repolling failed offchain data sources

Author: lutter

core/src/polling_monitor/mod.rs

@@ -169,7 +169,9 @@ where
                         debug!(logger, "not found on polling"; "object_id" => id.to_string());
 
                         metrics.not_found.inc();
-                        queue.push_back(id);
+
+                        // We'll try again after a backoff.
+                        backoff(id, &queue, &mut backoffs);
                     }
 
                     // Error polling, log it and push the id to the back of the queue.
@@ -182,12 +184,7 @@ where
                         // Requests that return errors could mean there is a permanent issue with
                         // fetching the given item, or could signal the endpoint is overloaded.
                         // Either way a backoff makes sense.
-                        let queue = queue.cheap_clone();
-                        let backoff = backoffs.next_backoff(id.clone());
-                        graph::spawn(async move {
-                            backoff.await;
-                            queue.push_back(id);
-                        });
+                        backoff(id, &queue, &mut backoffs);
                     }
                 }
             }
@@ -197,6 +194,18 @@ where
     PollingMonitor { queue }
 }
 
+fn backoff<ID>(id: ID, queue: &Arc<Queue<ID>>, backoffs: &mut Backoffs<ID>)
+where
+    ID: Eq + Hash + Clone + Send + 'static,
+{
+    let queue = queue.cheap_clone();
+    let backoff = backoffs.next_backoff(id.clone());
+    graph::spawn(async move {
+        backoff.await;
+        queue.push_back(id);
+    });
+}
+
 /// Handle for adding objects to be monitored.
 pub struct PollingMonitor<ID> {
     queue: Arc<Queue<ID>>,