diff --git a/swh/scheduler/backend.py b/swh/scheduler/backend.py
--- a/swh/scheduler/backend.py
+++ b/swh/scheduler/backend.py
@@ -317,6 +317,9 @@
         policy: str,
         timestamp: Optional[datetime.datetime] = None,
         scheduled_cooldown: Optional[datetime.timedelta] = datetime.timedelta(days=7),
+        failed_cooldown: Optional[datetime.timedelta] = datetime.timedelta(days=14),
+        notfound_cooldown: Optional[datetime.timedelta] = datetime.timedelta(days=31),
+        tablesample: Optional[float] = None,
         db=None,
         cur=None,
     ) -> List[ListedOrigin]:
@@ -354,6 +357,24 @@
             query_args.append(timestamp)
             query_args.append(scheduled_cooldown)
 
+        if failed_cooldown:
+            # Don't retry failed origins too often
+            where_clauses.append(
+                "origin_visit_stats.last_failed is null "
+                "or origin_visit_stats.last_failed < %s - %s"
+            )
+            query_args.append(timestamp)
+            query_args.append(failed_cooldown)
+
+        if notfound_cooldown:
+            # Don't retry not found origins too often
+            where_clauses.append(
+                "origin_visit_stats.last_notfound is null "
+                "or origin_visit_stats.last_notfound < %s - %s"
+            )
+            query_args.append(timestamp)
+            query_args.append(notfound_cooldown)
+
         if policy == "oldest_scheduled_first":
             order_by = "origin_visit_stats.last_scheduled NULLS FIRST"
         elif policy == "never_visited_oldest_update_first":
@@ -363,6 +384,15 @@
             # order by increasing last_update (oldest first)
             where_clauses.append("listed_origins.last_update IS NOT NULL")
             order_by = "listed_origins.last_update"
+        elif policy == "never_visited_unknown_last_update":
+            # never visited origins have a NULL last_snapshot
+            where_clauses.append("origin_visit_stats.last_snapshot IS NULL")
+
+            # Unknown last update
+            where_clauses.append("listed_origins.last_update IS NULL")
+
+            # Try to get at the oldest origins first
+            order_by = "listed_origins.first_seen"
         elif policy == "already_visited_order_by_lag":
             # TODO: store "visit lag" in a materialized view?
 
@@ -393,11 +423,17 @@
         else:
             raise UnknownPolicy(f"Unknown scheduling policy {policy}")
 
+        if tablesample:
+            table = "listed_origins tablesample SYSTEM (%s)"
+            query_args.insert(0, tablesample)
+        else:
+            table = "listed_origins"
+
         select_query = f"""
           SELECT
             {origin_select_cols}
           FROM
-            listed_origins
+            {table}
           LEFT JOIN
             origin_visit_stats USING (url, visit_type)
           WHERE
diff --git a/swh/scheduler/cli/origin.py b/swh/scheduler/cli/origin.py
--- a/swh/scheduler/cli/origin.py
+++ b/swh/scheduler/cli/origin.py
@@ -145,6 +145,52 @@
     click.echo_via_pager("\n".join(output))
 
 
+@origin.command("send-to-celery")
+@click.option(
+    "--policy", "-p", default="oldest_scheduled_first", help="Scheduling policy"
+)
+@click.option(
+    "--queue", "-q", help="Target celery queue", type=str,
+)
+@click.option(
+    "--tablesample", help="Table sampling percentage", type=float,
+)
+@click.argument("type", type=str)
+@click.pass_context
+def send_to_celery(
+    ctx, policy: str, queue: Optional[str], tablesample: Optional[float], type: str
+):
+    """Send the next origin visits of the TYPE loader to celery, filling the queue."""
+    from kombu.utils.uuid import uuid
+
+    from swh.scheduler.celery_backend.config import app, get_available_slots
+
+    scheduler = ctx.obj["scheduler"]
+
+    task_type = scheduler.get_task_type(f"load-{type}")
+
+    task_name = task_type["backend_name"]
+    queue_name = queue or task_name
+
+    num_tasks = get_available_slots(app, queue_name, task_type["max_queue_length"])
+
+    print(num_tasks, "slots available in celery queue")
+    origins = scheduler.grab_next_visits(
+        type, num_tasks, policy=policy, tablesample=tablesample
+    )
+
+    print(len(origins), "visits to send to celery")
+    for origin in origins:
+        task_dict = origin.as_task_dict()
+        app.send_task(
+            task_name,
+            task_id=uuid(),
+            args=task_dict["arguments"]["args"],
+            kwargs=task_dict["arguments"]["kwargs"],
+            queue=queue_name,
+        )
+
+
 @origin.command("update-metrics")
 @click.option("--lister", default=None, help="Only update metrics for this lister")
 @click.option(
diff --git a/swh/scheduler/interface.py b/swh/scheduler/interface.py
--- a/swh/scheduler/interface.py
+++ b/swh/scheduler/interface.py
@@ -390,6 +390,9 @@
         policy: str,
         timestamp: Optional[datetime.datetime] = None,
         scheduled_cooldown: Optional[datetime.timedelta] = datetime.timedelta(days=7),
+        failed_cooldown: Optional[datetime.timedelta] = datetime.timedelta(days=14),
+        notfound_cooldown: Optional[datetime.timedelta] = datetime.timedelta(days=31),
+        tablesample: Optional[float] = None,
     ) -> List[ListedOrigin]:
         """Get at most the `count` next origins that need to be visited with
         the `visit_type` loader according to the given scheduling `policy`.
@@ -405,6 +408,12 @@
             being scheduled (defaults to the current time)
           scheduled_cooldown: the minimal interval before which we can schedule
             the same origin again
+          failed_cooldown: the minimal interval before which we can reschedule a
+            failed origin
+          notfound_cooldown: the minimal interval before which we can reschedule a
+            notfound origin
+          tablesample: the percentage of the table on which we run the query
+            (None: no sampling)
         """
         ...