diff --git a/sql/swh-scheduler-schema.sql b/sql/swh-scheduler-schema.sql
index 4d39319..040f2e0 100644
--- a/sql/swh-scheduler-schema.sql
+++ b/sql/swh-scheduler-schema.sql
@@ -1,505 +1,506 @@
 create table dbversion
 (
   version     int primary key,
   release     timestamptz not null,
   description text not null
 );
 
 comment on table dbversion is 'Schema update tracking';
 
 insert into dbversion (version, release, description)
-       values (9, now(), 'Work In Progress');
+       values (10, now(), 'Work In Progress');
 
 create table task_type (
   type text primary key,
   description text not null,
   backend_name text not null,
   default_interval interval,
   min_interval interval,
   max_interval interval,
   backoff_factor float,
   max_queue_length bigint,
   num_retries bigint,
   retry_delay interval
 );
 
 comment on table task_type is 'Types of schedulable tasks';
 comment on column task_type.type is 'Short identifier for the task type';
 comment on column task_type.description is 'Human-readable task description';
 comment on column task_type.backend_name is 'Name of the task in the job-running backend';
 comment on column task_type.default_interval is 'Default interval for newly scheduled tasks';
 comment on column task_type.min_interval is 'Minimum interval between two runs of a task';
 comment on column task_type.max_interval is 'Maximum interval between two runs of a task';
 comment on column task_type.backoff_factor is 'Adjustment factor for the backoff between two task runs';
 comment on column task_type.max_queue_length is 'Maximum length of the queue for this type of tasks';
 comment on column task_type.num_retries is 'Default number of retries on transient failures';
 comment on column task_type.retry_delay is 'Retry delay for the task';
 
 create type task_status as enum ('next_run_not_scheduled', 'next_run_scheduled', 'completed', 'disabled');
 comment on type task_status is 'Status of a given task';
 
 create type task_policy as enum ('recurring', 'oneshot');
 comment on type task_policy is 'Recurrence policy of the given task';
 
 create type task_priority as enum('high', 'normal', 'low');
 comment on type task_priority is 'Priority of the given task';
 
 create table priority_ratio(
   id task_priority primary key,
   ratio float not null
 );
 
 comment on table priority_ratio is 'Oneshot task''s reading ratio per priority';
 comment on column priority_ratio.id is 'Task priority id';
 comment on column priority_ratio.ratio is 'Percentage of tasks to read per priority';
 
 insert into priority_ratio (id, ratio) values ('high', 0.5);
 insert into priority_ratio (id, ratio) values ('normal', 0.3);
 insert into priority_ratio (id, ratio) values ('low', 0.2);
 
 create table task (
   id bigserial primary key,
   type text not null references task_type(type),
   arguments jsonb not null,
   next_run timestamptz not null,
   current_interval interval,
   status task_status not null,
   policy task_policy not null default 'recurring',
   retries_left bigint not null default 0,
   priority task_priority references priority_ratio(id),
   check (policy <> 'recurring' or current_interval is not null)
 );
 
 comment on table task is 'Schedule of recurring tasks';
 comment on column task.arguments is 'Arguments passed to the underlying job scheduler. '
                                     'Contains two keys, ''args'' (list) and ''kwargs'' (object).';
 comment on column task.next_run is 'The next run of this task should be run on or after that time';
 comment on column task.current_interval is 'The interval between two runs of this task, '
                                            'taking into account the backoff factor';
 comment on column task.policy is 'Whether the task is one-shot or recurring';
 comment on column task.retries_left is 'The number of "short delay" retries of the task in case of '
                                        'transient failure';
 comment on column task.priority is 'Policy of the given task';
 
 create index on task(type);
 create index on task(next_run);
 create index task_args on task using btree ((arguments -> 'args'));
 create index task_kwargs on task using gin ((arguments -> 'kwargs'));
 create index on task(priority);
 
 create type task_run_status as enum ('scheduled', 'started', 'eventful', 'uneventful', 'failed', 'permfailed', 'lost');
 comment on type task_run_status is 'Status of a given task run';
 
 create table task_run (
   id bigserial primary key,
   task bigint not null references task(id),
   backend_id text,
   scheduled timestamptz,
   started timestamptz,
   ended timestamptz,
   metadata jsonb,
   status task_run_status not null default 'scheduled'
 );
 comment on table task_run is 'History of task runs sent to the job-running backend';
 comment on column task_run.backend_id is 'id of the task run in the job-running backend';
 comment on column task_run.metadata is 'Useful metadata for the given task run. '
                                        'For instance, the worker that took on the job, '
                                        'or the logs for the run.';
 
 create index on task_run(task);
 create index on task_run(backend_id);
 
 create or replace function swh_scheduler_mktemp_task ()
   returns void
   language sql
 as $$
   create temporary table tmp_task (
     like task excluding indexes
   ) on commit drop;
   alter table tmp_task
     drop column id,
     drop column current_interval,
     drop column status,
     alter column policy drop not null,
     alter column retries_left drop not null;
 $$;
 
 comment on function swh_scheduler_mktemp_task () is 'Create a temporary table for bulk task creation';
 
 
 create or replace function swh_scheduler_create_tasks_from_temp ()
   returns setof task
   language plpgsql
 as $$
 begin
   return query
   insert into task (type, arguments, next_run, status, current_interval, policy, retries_left, priority)
     select type, arguments, next_run, 'next_run_not_scheduled',
            (select default_interval from task_type tt where tt.type = t.type),
            coalesce(policy, 'recurring'),
            coalesce(retries_left, (select num_retries from task_type tt where tt.type = t.type), 0),
            coalesce(priority, null)
       from tmp_task t
       where not exists(select 1
                        from task
                        where type=t.type and
                              arguments=t.arguments and
                              policy=t.policy and
                              status='next_run_not_scheduled')
   returning task.*;
 end;
 $$;
 
 comment on function swh_scheduler_create_tasks_from_temp () is 'Create tasks in bulk from the temporary table';
 
 create or replace function swh_scheduler_peek_no_priority_tasks (task_type text, ts timestamptz default now(),
                                                                  num_tasks bigint default NULL)
   returns setof task
   language sql
   stable
 as $$
 select * from task
   where next_run <= ts
         and type = task_type
         and status = 'next_run_not_scheduled'
         and priority is null
   order by next_run
   limit num_tasks
   for update skip locked;
 $$;
 
 comment on function swh_scheduler_peek_no_priority_tasks (text, timestamptz, bigint)
 is 'Retrieve tasks without priority';
 
 create or replace function swh_scheduler_nb_priority_tasks(num_tasks_priority bigint, task_priority task_priority)
   returns numeric
   language sql stable
 as $$
   select ceil(num_tasks_priority * (select ratio from priority_ratio where id = task_priority)) :: numeric
 $$;
 
 comment on function swh_scheduler_nb_priority_tasks (bigint, task_priority)
 is 'Given a priority task and a total number, compute the number of tasks to read';
 
 create or replace function swh_scheduler_peek_tasks_with_priority (task_type text, ts timestamptz default now(),
                                                                    num_tasks_priority bigint default NULL,
                                                                    task_priority task_priority default 'normal')
   returns setof task
   language sql
   stable
 as $$
   select *
   from task t
   where t.next_run <= ts
         and t.type = task_type
         and t.status = 'next_run_not_scheduled'
         and t.priority = task_priority
   order by t.next_run
   limit num_tasks_priority
   for update skip locked;
 $$;
 
 comment on function swh_scheduler_peek_tasks_with_priority(text, timestamptz, bigint, task_priority)
 is 'Retrieve tasks with a given priority';
 
 create or replace function swh_scheduler_peek_priority_tasks (task_type text, ts timestamptz default now(),
                                                               num_tasks_priority bigint default NULL)
   returns setof task
   language plpgsql
 as $$
 declare
   r record;
   count_row bigint;
   nb_diff bigint;
   nb_high bigint;
   nb_normal bigint;
   nb_low bigint;
 begin
     -- expected values to fetch
     select swh_scheduler_nb_priority_tasks(num_tasks_priority, 'high') into nb_high;
     select swh_scheduler_nb_priority_tasks(num_tasks_priority, 'normal') into nb_normal;
     select swh_scheduler_nb_priority_tasks(num_tasks_priority, 'low') into nb_low;
     nb_diff := 0;
     count_row := 0;
 
     for r in select * from swh_scheduler_peek_tasks_with_priority(task_type, ts, nb_high, 'high')
     loop
         count_row := count_row + 1;
         return next r;
     end loop;
 
     if count_row < nb_high then
         nb_normal := nb_normal + nb_high - count_row;
     end if;
 
     count_row := 0;
     for r in select * from swh_scheduler_peek_tasks_with_priority(task_type, ts, nb_normal, 'normal')
     loop
         count_row := count_row + 1;
         return next r;
     end loop;
 
     if count_row < nb_normal then
         nb_low := nb_low + nb_normal - count_row;
     end if;
 
     return query select * from swh_scheduler_peek_tasks_with_priority(task_type, ts, nb_low, 'low');
 end
 $$;
 
 comment on function swh_scheduler_peek_priority_tasks(text, timestamptz, bigint)
 is 'Retrieve priority tasks';
 
 create or replace function swh_scheduler_peek_ready_tasks (task_type text, ts timestamptz default now(),
                                                            num_tasks bigint default NULL, num_tasks_priority bigint default NULL)
   returns setof task
   language plpgsql
 as $$
 declare
   r record;
   count_row bigint;
   nb_diff bigint;
   nb_tasks bigint;
 begin
     count_row := 0;
 
     for r in select * from swh_scheduler_peek_priority_tasks(task_type, ts, num_tasks_priority)
              order by priority, next_run
     loop
         count_row := count_row + 1;
         return next r;
     end loop;
 
     if count_row < num_tasks_priority then
        nb_tasks := num_tasks + num_tasks_priority - count_row;
     else
        nb_tasks := num_tasks;
     end if;
 
     for r in select * from swh_scheduler_peek_no_priority_tasks(task_type, ts, nb_tasks)
              order by priority, next_run
     loop
         return next r;
     end loop;
 
     return;
 end
 $$;
 
 comment on function swh_scheduler_peek_ready_tasks(text, timestamptz, bigint, bigint)
 is 'Retrieve tasks with/without priority in order';
 
 create or replace function swh_scheduler_grab_ready_tasks (task_type text, ts timestamptz default now(),
                                                            num_tasks bigint default NULL,
                                                            num_tasks_priority bigint default NULL)
   returns setof task
   language sql
 as $$
   update task
     set status='next_run_scheduled'
     from (
         select id from swh_scheduler_peek_ready_tasks(task_type, ts, num_tasks, num_tasks_priority)
     ) next_tasks
     where task.id = next_tasks.id
   returning task.*;
 $$;
 
 comment on function swh_scheduler_grab_ready_tasks (text, timestamptz, bigint, bigint)
 is 'Grab tasks ready for scheduling and change their status';
 
 create or replace function swh_scheduler_schedule_task_run (task_id bigint,
                                                             backend_id text,
                                                             metadata jsonb default '{}'::jsonb,
                                                             ts timestamptz default now())
   returns task_run
   language sql
 as $$
   insert into task_run (task, backend_id, metadata, scheduled, status)
     values (task_id, backend_id, metadata, ts, 'scheduled')
   returning *;
 $$;
 
 create or replace function swh_scheduler_mktemp_task_run ()
   returns void
   language sql
 as $$
   create temporary table tmp_task_run (
     like task_run excluding indexes
   ) on commit drop;
   alter table tmp_task_run
     drop column id,
     drop column status;
 $$;
 
 comment on function swh_scheduler_mktemp_task_run () is 'Create a temporary table for bulk task run scheduling';
 
 create or replace function swh_scheduler_schedule_task_run_from_temp ()
   returns void
   language plpgsql
 as $$
 begin
   insert into task_run (task, backend_id, metadata, scheduled, status)
     select task, backend_id, metadata, scheduled, 'scheduled'
       from tmp_task_run;
   return;
 end;
 $$;
 
 create or replace function swh_scheduler_start_task_run (backend_id text,
                                                          metadata jsonb default '{}'::jsonb,
                                                          ts timestamptz default now())
   returns task_run
   language sql
 as $$
   update task_run
     set started = ts,
         status = 'started',
         metadata = coalesce(task_run.metadata, '{}'::jsonb) || swh_scheduler_start_task_run.metadata
     where task_run.backend_id = swh_scheduler_start_task_run.backend_id
   returning *;
 $$;
 
 create or replace function swh_scheduler_end_task_run (backend_id text,
                                                        status task_run_status,
                                                        metadata jsonb default '{}'::jsonb,
                                                        ts timestamptz default now())
   returns task_run
   language sql
 as $$
   update task_run
     set ended = ts,
         status = swh_scheduler_end_task_run.status,
         metadata = coalesce(task_run.metadata, '{}'::jsonb) || swh_scheduler_end_task_run.metadata
     where task_run.backend_id = swh_scheduler_end_task_run.backend_id
   returning *;
 $$;
 
 create type task_record as (
     task_id bigint,
     task_policy task_policy,
     task_status task_status,
     task_run_id bigint,
     arguments jsonb,
     type text,
     backend_id text,
     metadata jsonb,
     scheduled timestamptz,
     started timestamptz,
-    ended timestamptz
+    ended timestamptz,
+    task_run_status task_run_status
 );
 
-create index task_run_id_asc_idx on task_run(task asc, ended asc);
+create index task_run_id_asc_idx on task_run(task asc, started asc);
 
 create or replace function swh_scheduler_task_to_archive(
   ts_after timestamptz, ts_before timestamptz, last_id bigint default -1,
   lim bigint default 10)
   returns setof task_record
   language sql stable
 as $$
    select t.id as task_id, t.policy as task_policy,
           t.status as task_status, tr.id as task_run_id,
           t.arguments, t.type, tr.backend_id, tr.metadata,
-          tr.scheduled, tr.started, tr.ended
+          tr.scheduled, tr.started, tr.ended, tr.status as task_run_status
    from task_run tr inner join task t on tr.task=t.id
-   where ((t.policy = 'oneshot' and t.status ='completed') or
-          (t.policy = 'recurring' and t.status ='disabled')) and
-          ts_after <= tr.ended  and tr.ended < ts_before and
+   where ((t.policy = 'oneshot' and t.status in ('completed', 'disabled')) or
+          (t.policy = 'recurring' and t.status = 'disabled')) and
+          ((ts_after <= tr.started and tr.started < ts_before) or tr.started is null) and
           t.id > last_id
-   order by tr.task, tr.ended
+   order by tr.task, tr.started
    limit lim;
 $$;
 
 comment on function swh_scheduler_task_to_archive is 'Read archivable tasks function';
 
 create or replace function swh_scheduler_delete_archived_tasks(
   task_ids bigint[], task_run_ids bigint[])
   returns void
   language sql
 as $$
   -- clean up task_run_ids
   delete from task_run where id in (select * from unnest(task_run_ids));
   -- clean up only tasks whose associated task_run are all cleaned up.
   -- Remaining tasks will stay there and will be cleaned up when
   -- remaining data have been indexed
   delete from task
   where id in (select t.id
                from task t left outer join task_run tr on t.id=tr.task
                where t.id in (select * from unnest(task_ids))
                and tr.task is null);
 $$;
 
 comment on function swh_scheduler_delete_archived_tasks is 'Clean up archived tasks function';
 
 create or replace function swh_scheduler_update_task_on_task_end ()
   returns trigger
   language plpgsql
 as $$
 declare
   cur_task task%rowtype;
   cur_task_type task_type%rowtype;
   adjustment_factor float;
   new_interval interval;
 begin
   select * from task where id = new.task into cur_task;
   select * from task_type where type = cur_task.type into cur_task_type;
 
   case
     when new.status = 'permfailed' then
       update task
         set status = 'disabled'
         where id = cur_task.id;
     when new.status in ('eventful', 'uneventful') then
       case
         when cur_task.policy = 'oneshot' then
           update task
             set status = 'completed'
             where id = cur_task.id;
         when cur_task.policy = 'recurring' then
           if new.status = 'uneventful' then
             adjustment_factor := 1/cur_task_type.backoff_factor;
           else
             adjustment_factor := 1/cur_task_type.backoff_factor;
           end if;
           new_interval := greatest(
             cur_task_type.min_interval,
             least(
               cur_task_type.max_interval,
               adjustment_factor * cur_task.current_interval));
           update task
             set status = 'next_run_not_scheduled',
                 next_run = now() + new_interval,
                 current_interval = new_interval,
                 retries_left = coalesce(cur_task_type.num_retries, 0)
             where id = cur_task.id;
       end case;
     else -- new.status in 'failed', 'lost'
       if cur_task.retries_left > 0 then
         update task
           set status = 'next_run_not_scheduled',
               next_run = now() + coalesce(cur_task_type.retry_delay, interval '1 hour'),
               retries_left = cur_task.retries_left - 1
           where id = cur_task.id;
       else -- no retries left
         case
           when cur_task.policy = 'oneshot' then
             update task
               set status = 'disabled'
               where id = cur_task.id;
           when cur_task.policy = 'recurring' then
             update task
               set status = 'next_run_not_scheduled',
                   next_run = now() + cur_task.current_interval,
                   retries_left = coalesce(cur_task_type.num_retries, 0)
               where id = cur_task.id;
         end case;
       end if; -- retries
   end case;
   return null;
 end;
 $$;
 
 create trigger update_task_on_task_end
   after update of status on task_run
   for each row
   when (new.status NOT IN ('scheduled', 'started'))
   execute procedure swh_scheduler_update_task_on_task_end ();
diff --git a/sql/updates/10.sql b/sql/updates/10.sql
new file mode 100644
index 0000000..bd57768
--- /dev/null
+++ b/sql/updates/10.sql
@@ -0,0 +1,47 @@
+-- SWH Scheduler Schema upgrade
+-- from_version: 09
+-- to_version: 10
+-- description: Schedule task with priority
+
+insert into dbversion (version, release, description)
+values (10, now(), 'Work In Progress');
+
+drop type task_record cascade;
+create type task_record as (
+    task_id bigint,
+    task_policy task_policy,
+    task_status task_status,
+    task_run_id bigint,
+    arguments jsonb,
+    type text,
+    backend_id text,
+    metadata jsonb,
+    scheduled timestamptz,
+    started timestamptz,
+    ended timestamptz,
+    task_run_status task_run_status
+);
+
+drop index task_run_id_asc_idx;
+create index task_run_id_started_asc_idx on task_run(task asc, started asc);
+
+create or replace function swh_scheduler_task_to_archive(
+  ts_after timestamptz, ts_before timestamptz, last_id bigint default -1,
+  lim bigint default 10)
+  returns setof task_record
+  language sql stable
+as $$
+   select t.id as task_id, t.policy as task_policy,
+          t.status as task_status, tr.id as task_run_id,
+          t.arguments, t.type, tr.backend_id, tr.metadata,
+          tr.scheduled, tr.started, tr.ended, tr.status as task_run_status
+   from task_run tr inner join task t on tr.task=t.id
+   where ((t.policy = 'oneshot' and t.status in ('completed', 'disabled')) or
+          (t.policy = 'recurring' and t.status = 'disabled')) and
+          ((ts_after <= tr.started and tr.started < ts_before) or tr.started is null) and
+          t.id > last_id
+   order by tr.task, tr.started
+   limit lim;
+$$;
+
+comment on function swh_scheduler_task_to_archive is 'Read archivable tasks function';
diff --git a/swh/scheduler/cli.py b/swh/scheduler/cli.py
index 0c0ef94..7ec6258 100644
--- a/swh/scheduler/cli.py
+++ b/swh/scheduler/cli.py
@@ -1,301 +1,308 @@
 # Copyright (C) 2016-2018  The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
 import arrow
 import click
 import csv
 import itertools
 import json
 import locale
 import logging
 
 from swh.core import utils
 from . import compute_nb_tasks_from
 from .backend_es import SWHElasticSearchClient
 
 
 locale.setlocale(locale.LC_ALL, '')
 ARROW_LOCALE = locale.getlocale(locale.LC_TIME)[0]
 
 
 class DateTimeType(click.ParamType):
     name = 'time and date'
 
     def convert(self, value, param, ctx):
         if not isinstance(value, arrow.Arrow):
             value = arrow.get(value)
 
         return value
 
 
 DATETIME = DateTimeType()
 CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
 
 
 def pretty_print_list(list, indent):
     """Pretty-print a list"""
     return ''.join('%s%s\n' % (' ' * indent, item) for item in list)
 
 
 def pretty_print_dict(dict, indent):
     """Pretty-print a list"""
     return ''.join('%s%s: %s\n' %
                    (' ' * indent, click.style(key, bold=True), value)
                    for key, value in dict.items())
 
 
 def pretty_print_task(task):
     """Pretty-print a task"""
     next_run = arrow.get(task['next_run'])
     lines = [
         '%s %s\n' % (click.style('Task', bold=True), task['id']),
         click.style('  Next run: ', bold=True),
         "%s (%s)" % (next_run.humanize(locale=ARROW_LOCALE),
                      next_run.format()),
         '\n',
         click.style('  Interval: ', bold=True),
         str(task['current_interval']), '\n',
         click.style('  Type: ', bold=True), task['type'], '\n',
         click.style('  Policy: ', bold=True), task['policy'], '\n',
         click.style('  Args:\n', bold=True),
         pretty_print_list(task['arguments']['args'], indent=4),
         click.style('  Keyword args:\n', bold=True),
         pretty_print_dict(task['arguments']['kwargs'], indent=4),
     ]
 
     return ''.join(lines)
 
 
 @click.group(context_settings=CONTEXT_SETTINGS)
 @click.option('--cls', '-c', default='local',
               help="Scheduler's class, default to 'local'")
 @click.option('--database', '-d',
               help='Scheduling database DSN')
 @click.option('--url', '-u',
               help="(Optional) Scheduler's url access")
 @click.pass_context
 def cli(ctx, cls, database, url):
     """Software Heritage Scheduler CLI interface
 
     Default to use the the local scheduler instance (plugged to the
     main scheduler db).
 
     """
     scheduler = None
     override_config = {}
     from . import get_scheduler
     if cls == 'local':
         if database:
             override_config = {'scheduling_db': database}
         scheduler = get_scheduler(cls, args=override_config)
     elif cls == 'remote':
         if url:
             override_config = {'url': url}
         scheduler = get_scheduler(cls, args=override_config)
 
     if not scheduler:
         raise ValueError('Scheduler class (local/remote) must be instantiated')
 
     ctx.obj = scheduler
 
 
 @cli.group('task')
 @click.pass_context
 def task(ctx):
     """Manipulate tasks."""
     pass
 
 
 @task.command('schedule')
 @click.option('--columns', '-c', multiple=True,
               default=['type', 'args', 'kwargs', 'next_run'],
               type=click.Choice([
                   'type', 'args', 'kwargs', 'policy', 'next_run']),
               help='columns present in the CSV file')
 @click.option('--delimiter', '-d', default=',')
 @click.argument('file', type=click.File(encoding='utf-8'))
 @click.pass_context
 def schedule_tasks(ctx, columns, delimiter, file):
     """Schedule tasks from a CSV input file.
 
     The following columns are expected, and can be set through the -c option:
 
      - type: the type of the task to be scheduled (mandatory)
 
      - args: the arguments passed to the task (JSON list, defaults to an empty
        list)
 
      - kwargs: the keyword arguments passed to the task (JSON object, defaults
        to an empty dict)
 
      - next_run: the date at which the task should run (datetime, defaults to
        now)
 
     The CSV can be read either from a named file, or from stdin (use - as
     filename).
 
     Use sample:
 
     cat scheduling-task.txt | \
         python3 -m swh.scheduler.cli \
             --database 'service=swh-scheduler-dev' \
             task schedule \
                 --columns type --columns kwargs --columns policy \
                 --delimiter ';' -
 
     """
     tasks = []
     now = arrow.utcnow()
 
     reader = csv.reader(file, delimiter=delimiter)
     for line in reader:
         task = dict(zip(columns, line))
         args = json.loads(task.pop('args', '[]'))
         kwargs = json.loads(task.pop('kwargs', '{}'))
         task['arguments'] = {
             'args': args,
             'kwargs': kwargs,
         }
         task['next_run'] = DATETIME.convert(task.get('next_run', now),
                                             None, None)
         tasks.append(task)
 
     created = ctx.obj.create_tasks(tasks)
 
     output = [
         'Created %d tasks\n' % len(created),
     ]
     for task in created:
         output.append(pretty_print_task(task))
 
     click.echo_via_pager('\n'.join(output))
 
 
 @task.command('list-pending')
 @click.option('--task-type', '-t', required=True,
               help='The tasks\' type concerned by the listing')
 @click.option('--limit', '-l', required=False, type=click.INT,
               help='The maximum number of tasks to fetch')
 @click.option('--before', '-b', required=False, type=DATETIME,
               help='List all jobs supposed to run before the given date')
 @click.pass_context
 def list_pending_tasks(ctx, task_type, limit, before):
     """List the tasks that are going to be run.
 
     You can override the number of tasks to fetch
 
     """
     num_tasks, num_tasks_priority = compute_nb_tasks_from(limit)
 
     pending = ctx.obj.peek_ready_tasks(
         task_type, timestamp=before,
         num_tasks=num_tasks, num_tasks_priority=num_tasks_priority)
     output = [
         'Found %d tasks\n' % len(pending)
     ]
     for task in pending:
         output.append(pretty_print_task(task))
 
     click.echo_via_pager('\n'.join(output))
 
 
 @task.command('archive')
 @click.option('--before', '-b', default=None,
               help='''Task whose ended date is anterior will be archived.
                       Default to current month's first day.''')
 @click.option('--after', '-a', default=None,
               help='''Task whose ended date is after the specified date will
                       be archived. Default to prior month's first day.''')
 @click.option('--batch-index', default=1000, type=click.INT,
               help='Batch size of tasks to read from db to archive')
 @click.option('--bulk-index', default=200, type=click.INT,
               help='Batch size of tasks to bulk index')
 @click.option('--batch-clean', default=1000, type=click.INT,
               help='Batch size of task to clean after archival')
 @click.option('--dry-run/--no-dry-run', is_flag=True, default=False,
               help='Default to list only what would be archived.')
 @click.option('--verbose', is_flag=True, default=False,
               help='Default to list only what would be archived.')
 @click.option('--cleanup/--no-cleanup', is_flag=True, default=True,
               help='Clean up archived tasks (default)')
 @click.option('--start-from', type=click.INT, default=-1,
               help='(Optional) default task id to start from. Default is -1.')
 @click.pass_context
 def archive_tasks(ctx, before, after, batch_index, bulk_index, batch_clean,
                   dry_run, verbose, cleanup, start_from):
     """Archive task/task_run whose (task_type is 'oneshot' and task_status
        is 'completed') or (task_type is 'recurring' and task_status is
        'disabled').
 
        With --dry-run flag set (default), only list those.
 
     """
     es_client = SWHElasticSearchClient()
     logging.basicConfig(level=logging.DEBUG if verbose else logging.INFO)
     log = logging.getLogger('swh.scheduler.cli.archive')
     logging.getLogger('urllib3').setLevel(logging.WARN)
     logging.getLogger('elasticsearch').setLevel(logging.WARN)
     if dry_run:
         log.info('**DRY-RUN** (only reading db)')
     if not cleanup:
         log.info('**NO CLEANUP**')
 
     now = arrow.utcnow()
 
     # Default to archive tasks from a rolling month starting the week
     # prior to the current one
     if not before:
         before = now.shift(weeks=-1).format('YYYY-MM-DD')
 
     if not after:
         after = now.shift(weeks=-1).shift(months=-1).format('YYYY-MM-DD')
 
     log.debug('index: %s; cleanup: %s; period: [%s ; %s]' % (
         not dry_run, not dry_run and cleanup, after, before))
 
     def group_by_index_name(data, es_client=es_client):
-        ended = data['ended']
-        return es_client.compute_index_name(ended.year, ended.month)
+        """Given a data record, determine the index's name through its ending
+           date. This varies greatly depending on the task_run's
+           status.
+
+        """
+        date = data.get('started')
+        if not date:
+            date = data['scheduled']
+        return es_client.compute_index_name(date.year, date.month)
 
     def index_data(before, last_id, batch_index, backend=ctx.obj):
         tasks_in = backend.filter_task_to_archive(
             after, before, last_id=last_id, limit=batch_index)
         for index_name, tasks_group in itertools.groupby(
                 tasks_in, key=group_by_index_name):
             log.debug('Index tasks to %s' % index_name)
             if dry_run:
                 for task in tasks_group:
                     yield task
                 continue
 
             yield from es_client.streaming_bulk(
                 index_name, tasks_group, source=['task_id', 'task_run_id'],
                 chunk_size=bulk_index, log=log)
 
     gen = index_data(before, last_id=start_from, batch_index=batch_index)
     if cleanup:
         for task_ids in utils.grouper(gen, n=batch_clean):
             task_ids = list(task_ids)
             log.info('Clean up %s tasks: [%s, ...]' % (
                 len(task_ids), task_ids[0]))
             if dry_run:  # no clean up
                 continue
             ctx.obj.delete_archived_tasks(task_ids)
     else:
         for task_ids in utils.grouper(gen, n=batch_index):
             task_ids = list(task_ids)
             log.info('Indexed %s tasks: [%s, ...]' % (
                 len(task_ids), task_ids[0]))
 
 
 @cli.group('task-run')
 @click.pass_context
 def task_run(ctx):
     """Manipulate task runs."""
     pass
 
 
 if __name__ == '__main__':
     cli()
diff --git a/swh/scheduler/tests/test_scheduler.py b/swh/scheduler/tests/test_scheduler.py
index e60facb..0c8f0e0 100644
--- a/swh/scheduler/tests/test_scheduler.py
+++ b/swh/scheduler/tests/test_scheduler.py
@@ -1,474 +1,475 @@
 # Copyright (C) 2017-2018  The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
 import copy
 import datetime
 import os
 import random
 import unittest
 import uuid
 
 from arrow import utcnow
 from collections import defaultdict
 from nose.plugins.attrib import attr
 from nose.tools import istest
 import psycopg2
 
 from swh.core.tests.db_testing import SingleDbTestFixture
 from swh.scheduler import get_scheduler
 
 
 TEST_DIR = os.path.dirname(os.path.abspath(__file__))
 TEST_DATA_DIR = os.path.join(TEST_DIR, '../../../../swh-storage-testdata')
 
 
 @attr('db')
 class CommonSchedulerTest(SingleDbTestFixture):
     TEST_DB_NAME = 'softwareheritage-scheduler-test'
     TEST_DB_DUMP = os.path.join(TEST_DATA_DIR, 'dumps/swh-scheduler.dump')
 
     def setUp(self):
         super().setUp()
 
         tt = {
             'type': 'update-git',
             'description': 'Update a git repository',
             'backend_name': 'swh.loader.git.tasks.UpdateGitRepository',
             'default_interval': datetime.timedelta(days=64),
             'min_interval': datetime.timedelta(hours=12),
             'max_interval': datetime.timedelta(days=64),
             'backoff_factor': 2,
             'max_queue_length': None,
             'num_retries': 7,
             'retry_delay': datetime.timedelta(hours=2),
         }
         tt2 = tt.copy()
         tt2['type'] = 'update-hg'
         tt2['description'] = 'Update a mercurial repository'
         tt2['backend_name'] = 'swh.loader.mercurial.tasks.UpdateHgRepository'
         tt2['max_queue_length'] = 42
         tt2['num_retries'] = None
         tt2['retry_delay'] = None
 
         self.task_types = {
             tt['type']: tt,
             tt2['type']: tt2,
         }
 
         self.task1_template = t1_template = {
             'type': tt['type'],
             'arguments': {
                 'args': [],
                 'kwargs': {},
             },
             'next_run': None,
         }
         self.task2_template = t2_template = copy.deepcopy(t1_template)
         t2_template['type'] = tt2['type']
         t2_template['policy'] = 'oneshot'
 
     def tearDown(self):
         self.backend.close_connection()
         self.empty_tables()
         super().tearDown()
 
     def empty_tables(self, whitelist=["priority_ratio"]):
         query = """SELECT table_name FROM information_schema.tables
                    WHERE table_schema = %%s and
                          table_name not in (%s)
                 """ % ','.join(map(lambda t: "'%s'" % t, whitelist))
         self.cursor.execute(query, ('public', ))
 
         tables = set(table for (table,) in self.cursor.fetchall())
 
         for table in tables:
             self.cursor.execute('truncate table %s cascade' % table)
         self.conn.commit()
 
     @istest
     def add_task_type(self):
         tt, tt2 = self.task_types.values()
         self.backend.create_task_type(tt)
         self.assertEqual(tt, self.backend.get_task_type(tt['type']))
         with self.assertRaisesRegex(psycopg2.IntegrityError,
                                     '\(type\)=\(%s\)' % tt['type']):
             self.backend.create_task_type(tt)
         self.backend.create_task_type(tt2)
         self.assertEqual(tt, self.backend.get_task_type(tt['type']))
         self.assertEqual(tt2, self.backend.get_task_type(tt2['type']))
 
     @istest
     def get_task_types(self):
         tt, tt2 = self.task_types.values()
         self.backend.create_task_type(tt)
         self.backend.create_task_type(tt2)
         self.assertCountEqual([tt2, tt], self.backend.get_task_types())
 
     @staticmethod
     def _task_from_template(template, next_run, priority, *args, **kwargs):
         ret = copy.deepcopy(template)
         ret['next_run'] = next_run
         if priority:
             ret['priority'] = priority
         if args:
             ret['arguments']['args'] = list(args)
         if kwargs:
             ret['arguments']['kwargs'] = kwargs
         return ret
 
     def _pop_priority(self, priorities):
         if not priorities:
             return None
         for priority, remains in priorities.items():
             if remains > 0:
                 priorities[priority] = remains - 1
                 return priority
         return None
 
     def _tasks_from_template(self, template, max_timestamp, num,
                              num_priority=0, priorities=None):
         if num_priority and priorities:
             priorities = {
                 priority: ratio * num_priority
                 for priority, ratio in priorities.items()
             }
 
         tasks = []
         for i in range(num + num_priority):
             priority = self._pop_priority(priorities)
             tasks.append(self._task_from_template(
                 template,
                 max_timestamp - datetime.timedelta(microseconds=i),
                 priority,
                 'argument-%03d' % i,
                 **{'kwarg%03d' % i: 'bogus-kwarg'}
             ))
         return tasks
 
     def _create_task_types(self):
         for tt in self.task_types.values():
             self.backend.create_task_type(tt)
 
     @istest
     def create_tasks(self):
         priority_ratio = self._priority_ratio()
         self._create_task_types()
         num_tasks_priority = 100
         tasks_1 = self._tasks_from_template(self.task1_template, utcnow(), 100)
         tasks_2 = self._tasks_from_template(
                 self.task2_template, utcnow(), 100,
                 num_tasks_priority, priorities=priority_ratio)
         tasks = tasks_1 + tasks_2
         # duplicate tasks are dropped at creation
         ret = self.backend.create_tasks(tasks + tasks_1 + tasks_2)
         ids = set()
         actual_priorities = defaultdict(int)
 
         for task, orig_task in zip(ret, tasks):
             task = copy.deepcopy(task)
             task_type = self.task_types[orig_task['type']]
             self.assertNotIn(task['id'], ids)
             self.assertEqual(task['status'], 'next_run_not_scheduled')
             self.assertEqual(task['current_interval'],
                              task_type['default_interval'])
             self.assertEqual(task['policy'], orig_task.get('policy',
                                                            'recurring'))
             priority = task.get('priority')
             if priority:
                 actual_priorities[priority] += 1
 
             self.assertEqual(task['retries_left'],
                              task_type['num_retries'] or 0)
             ids.add(task['id'])
             del task['id']
             del task['status']
             del task['current_interval']
             del task['retries_left']
             if 'policy' not in orig_task:
                 del task['policy']
             if 'priority' not in orig_task:
                 del task['priority']
                 self.assertEqual(task, orig_task)
 
         self.assertEqual(dict(actual_priorities), {
             priority: int(ratio * num_tasks_priority)
             for priority, ratio in priority_ratio.items()
         })
 
     @istest
     def peek_ready_tasks_no_priority(self):
         self._create_task_types()
         t = utcnow()
         task_type = self.task1_template['type']
         tasks = self._tasks_from_template(self.task1_template, t, 100)
         random.shuffle(tasks)
         self.backend.create_tasks(tasks)
 
         ready_tasks = self.backend.peek_ready_tasks(task_type)
         self.assertEqual(len(ready_tasks), len(tasks))
         for i in range(len(ready_tasks) - 1):
             self.assertLessEqual(ready_tasks[i]['next_run'],
                                  ready_tasks[i+1]['next_run'])
 
         # Only get the first few ready tasks
         limit = random.randrange(5, 5 + len(tasks)//2)
         ready_tasks_limited = self.backend.peek_ready_tasks(
             task_type, num_tasks=limit)
 
         self.assertEqual(len(ready_tasks_limited), limit)
         self.assertCountEqual(ready_tasks_limited, ready_tasks[:limit])
 
         # Limit by timestamp
         max_ts = tasks[limit-1]['next_run']
         ready_tasks_timestamped = self.backend.peek_ready_tasks(
             task_type, timestamp=max_ts)
 
         for ready_task in ready_tasks_timestamped:
             self.assertLessEqual(ready_task['next_run'], max_ts)
 
         # Make sure we get proper behavior for the first ready tasks
         self.assertCountEqual(
             ready_tasks[:len(ready_tasks_timestamped)],
             ready_tasks_timestamped,
         )
 
         # Limit by both
         ready_tasks_both = self.backend.peek_ready_tasks(
             task_type, timestamp=max_ts, num_tasks=limit//3)
         self.assertLessEqual(len(ready_tasks_both), limit//3)
         for ready_task in ready_tasks_both:
             self.assertLessEqual(ready_task['next_run'], max_ts)
             self.assertIn(ready_task, ready_tasks[:limit//3])
 
     def _priority_ratio(self):
         self.cursor.execute('select id, ratio from priority_ratio')
         priority_ratio = {}
         for row in self.cursor.fetchall():
             priority_ratio[row[0]] = row[1]
         return priority_ratio
 
     @istest
     def peek_ready_tasks_mixed_priorities(self):
         priority_ratio = self._priority_ratio()
         self._create_task_types()
         t = utcnow()
         task_type = self.task1_template['type']
         num_tasks_priority = 100
         num_tasks_no_priority = 100
         # Create tasks with and without priorities
         tasks = self._tasks_from_template(
             self.task1_template, t,
             num=num_tasks_no_priority,
             num_priority=num_tasks_priority,
             priorities=priority_ratio)
 
         random.shuffle(tasks)
         self.backend.create_tasks(tasks)
 
         # take all available tasks
         ready_tasks = self.backend.peek_ready_tasks(
             task_type)
 
         self.assertEqual(len(ready_tasks), len(tasks))
         self.assertEqual(num_tasks_priority + num_tasks_no_priority,
                          len(ready_tasks))
 
         count_tasks_per_priority = defaultdict(int)
         for task in ready_tasks:
             priority = task.get('priority')
             if priority:
                 count_tasks_per_priority[priority] += 1
 
         self.assertEqual(dict(count_tasks_per_priority), {
             priority: int(ratio * num_tasks_priority)
             for priority, ratio in priority_ratio.items()
         })
 
         # Only get some ready tasks
         num_tasks = random.randrange(5, 5 + num_tasks_no_priority//2)
         num_tasks_priority = random.randrange(5, num_tasks_priority//2)
         ready_tasks_limited = self.backend.peek_ready_tasks(
             task_type, num_tasks=num_tasks,
             num_tasks_priority=num_tasks_priority)
 
         count_tasks_per_priority = defaultdict(int)
         for task in ready_tasks_limited:
             priority = task.get('priority')
             count_tasks_per_priority[priority] += 1
 
         import math
         for priority, ratio in priority_ratio.items():
             expected_count = math.ceil(ratio * num_tasks_priority)
             actual_prio = count_tasks_per_priority[priority]
             self.assertTrue(
                 actual_prio == expected_count or
                 actual_prio == expected_count + 1)
 
         self.assertEqual(count_tasks_per_priority[None], num_tasks)
 
     @istest
     def grab_ready_tasks(self):
         priority_ratio = self._priority_ratio()
         self._create_task_types()
         t = utcnow()
         task_type = self.task1_template['type']
         num_tasks_priority = 100
         num_tasks_no_priority = 100
         # Create tasks with and without priorities
         tasks = self._tasks_from_template(
             self.task1_template, t,
             num=num_tasks_no_priority,
             num_priority=num_tasks_priority,
             priorities=priority_ratio)
         random.shuffle(tasks)
         self.backend.create_tasks(tasks)
 
         first_ready_tasks = self.backend.peek_ready_tasks(
             task_type, num_tasks=10, num_tasks_priority=10)
         grabbed_tasks = self.backend.grab_ready_tasks(
             task_type, num_tasks=10, num_tasks_priority=10)
 
         for peeked, grabbed in zip(first_ready_tasks, grabbed_tasks):
             self.assertEqual(peeked['status'], 'next_run_not_scheduled')
             del peeked['status']
             self.assertEqual(grabbed['status'], 'next_run_scheduled')
             del grabbed['status']
             self.assertEqual(peeked, grabbed)
             self.assertEqual(peeked['priority'], grabbed['priority'])
 
     @istest
     def get_tasks(self):
         self._create_task_types()
         t = utcnow()
         tasks = self._tasks_from_template(self.task1_template, t, 100)
         tasks = self.backend.create_tasks(tasks)
         random.shuffle(tasks)
         while len(tasks) > 1:
             length = random.randrange(1, len(tasks))
             cur_tasks = tasks[:length]
             tasks[:length] = []
 
             ret = self.backend.get_tasks(task['id'] for task in cur_tasks)
             self.assertCountEqual(ret, cur_tasks)
 
     @istest
     def filter_task_to_archive(self):
         """Filtering only list disabled recurring or completed oneshot tasks
 
         """
         self._create_task_types()
         _time = utcnow()
         recurring = self._tasks_from_template(self.task1_template, _time, 12)
         oneshots = self._tasks_from_template(self.task2_template, _time, 12)
         total_tasks = len(recurring) + len(oneshots)
 
         # simulate scheduling tasks
         pending_tasks = self.backend.create_tasks(recurring + oneshots)
         backend_tasks = [{
             'task': task['id'],
             'backend_id': str(uuid.uuid4()),
             'scheduled': utcnow(),
         } for task in pending_tasks]
         self.backend.mass_schedule_task_runs(backend_tasks)
 
         # we simulate the task are being done
         _tasks = []
         for task in backend_tasks:
             t = self.backend.end_task_run(
                 task['backend_id'], status='eventful')
             _tasks.append(t)
 
         # Randomly update task's status per policy
         status_per_policy = {'recurring': 0, 'oneshot': 0}
         status_choice = {
             # policy: [tuple (1-for-filtering, 'associated-status')]
             'recurring': [(1, 'disabled'),
                           (0, 'completed'),
                           (0, 'next_run_not_scheduled')],
             'oneshot': [(0, 'next_run_not_scheduled'),
-                        (0, 'disabled'),
+                        (1, 'disabled'),
                         (1, 'completed')]
         }
 
         tasks_to_update = defaultdict(list)
         _task_ids = defaultdict(list)
         # randomize 'disabling' recurring task or 'complete' oneshot task
         for task in pending_tasks:
             policy = task['policy']
             _task_ids[policy].append(task['id'])
             status = random.choice(status_choice[policy])
             if status[0] != 1:
                 continue
             # elected for filtering
             status_per_policy[policy] += status[0]
             tasks_to_update[policy].append(task['id'])
 
         self.backend.disable_tasks(tasks_to_update['recurring'])
-        # hack: change the status to something else than completed
+        # hack: change the status to something else than completed/disabled
         self.backend.set_status_tasks(
-            _task_ids['oneshot'], status='disabled')
+            _task_ids['oneshot'], status='next_run_not_scheduled')
+        # complete the tasks to update
         self.backend.set_status_tasks(
             tasks_to_update['oneshot'], status='completed')
 
         total_tasks_filtered = (status_per_policy['recurring'] +
                                 status_per_policy['oneshot'])
 
         # retrieve tasks to archive
         after = _time.shift(days=-1).format('YYYY-MM-DD')
         before = utcnow().shift(days=1).format('YYYY-MM-DD')
         tasks_to_archive = list(self.backend.filter_task_to_archive(
             after_ts=after, before_ts=before, limit=total_tasks))
 
         self.assertEqual(len(tasks_to_archive), total_tasks_filtered)
 
         actual_filtered_per_status = {'recurring': 0, 'oneshot': 0}
         for task in tasks_to_archive:
             actual_filtered_per_status[task['task_policy']] += 1
 
         self.assertEqual(actual_filtered_per_status, status_per_policy)
 
     @istest
     def delete_archived_tasks(self):
         self._create_task_types()
         _time = utcnow()
         recurring = self._tasks_from_template(
             self.task1_template, _time, 12)
         oneshots = self._tasks_from_template(
             self.task2_template, _time, 12)
         total_tasks = len(recurring) + len(oneshots)
         pending_tasks = self.backend.create_tasks(recurring + oneshots)
         backend_tasks = [{
             'task': task['id'],
             'backend_id': str(uuid.uuid4()),
             'scheduled': utcnow(),
         } for task in pending_tasks]
         self.backend.mass_schedule_task_runs(backend_tasks)
 
         _tasks = []
         percent = random.randint(0, 100)  # random election removal boundary
         for task in backend_tasks:
             t = self.backend.end_task_run(
                 task['backend_id'], status='eventful')
             c = random.randint(0, 100)
             if c <= percent:
                 _tasks.append({'task_id': t['task'], 'task_run_id': t['id']})
 
         self.backend.delete_archived_tasks(_tasks)
 
         self.cursor.execute('select count(*) from task')
         tasks_count = self.cursor.fetchone()
 
         self.cursor.execute('select count(*) from task_run')
         tasks_run_count = self.cursor.fetchone()
 
         self.assertEqual(tasks_count[0], total_tasks - len(_tasks))
         self.assertEqual(tasks_run_count[0], total_tasks - len(_tasks))
 
 
 class LocalSchedulerTest(CommonSchedulerTest, unittest.TestCase):
     def setUp(self):
         super().setUp()
         self.config = {'scheduling_db': 'dbname=' + self.TEST_DB_NAME}
         self.backend = get_scheduler('local', self.config)