0059.yml

version: 59
description: Add task_queue_id column to tasks table and populate with online migration
migrationScript: |-
  begin
    alter table tasks add column task_queue_id text;

    create function online_migration_v59_batch(batch_size_in integer, state_in jsonb)
    returns table (count integer, state jsonb) as $$
    declare
      item record;
      count integer;
      last_task_id text;
    begin
      count := 0;
      for item in
        select task_id
        from tasks
        where
          (state_in ->> 'task_id' is null or task_id > state_in ->> 'task_id') and
          task_queue_id is null
        order by task_id
        limit batch_size_in
      loop
        update tasks
        set task_queue_id = tasks.provisioner_id || '/' || tasks.worker_type
        where tasks.task_id = item.task_id;
        count := count + 1;
      end loop;
      return query select
        count as count,
        to_jsonb(item) as state;
    end
    $$ language plpgsql;

    create function online_migration_v59_is_complete() returns boolean as $$
    begin
      perform * from tasks where task_queue_id is null limit 1;
      return not found;
    end
    $$ language plpgsql;
  end
downgradeScript: |-
  begin
    alter table tasks drop column task_queue_id;

  end
methods:
  create_task:
    description: |-
      Create a new task, without scheduling it, and with empty values
      for the status information.
    mode: write
    serviceName: queue
    args: |-
      task_id text,
      provisioner_id text,
      worker_type text,
      scheduler_id text,
      task_group_id text,
      dependencies jsonb,
      requires task_requires,
      routes jsonb,
      priority task_priority,
      retries integer,
      created timestamptz,
      deadline timestamptz,
      expires timestamptz,
      scopes jsonb,
      payload jsonb,
      metadata jsonb,
      tags jsonb,
      extra jsonb
    returns: void
    body: |-
      begin
        insert
        into tasks (
          task_id,
          provisioner_id,
          worker_type,
          task_queue_id,
          scheduler_id,
          task_group_id,
          dependencies,
          requires,
          routes,
          priority,
          retries,
          created,
          deadline,
          expires,
          scopes,
          payload,
          metadata,
          tags,
          extra,
          retries_left,
          runs,
          taken_until,
          ever_resolved
        )
        values (
          task_id,
          provisioner_id,
          worker_type,
          provisioner_id || '/' || worker_type,
          scheduler_id,
          task_group_id,
          dependencies,
          requires,
          routes,
          priority,
          retries,
          created,
          deadline,
          expires,
          scopes,
          payload,
          metadata,
          tags,
          extra,
          -- default values for the mutable bits
          retries,
          jsonb_build_array(),
          null, -- not taken
          false
        );
      end
  get_task:
    description: |-
      Get all properties of a task.  Note that all properties but `runs`,
      `retries_left`, and `taken_until` are immutable.
    mode: read
    serviceName: queue
    args: task_id_in text
    returns: |-
      table (
        task_id text,
        provisioner_id text,
        worker_type text,
        scheduler_id text,
        task_group_id text,
        dependencies jsonb,
        requires task_requires,
        routes jsonb,
        priority task_priority,
        retries integer,
        retries_left int,
        created timestamptz,
        deadline timestamptz,
        expires timestamptz,
        scopes jsonb,
        payload jsonb,
        metadata jsonb,
        tags jsonb,
        extra jsonb,
        runs jsonb,
        taken_until timestamptz
      )
    body: |-
      begin
        return query
        select
          tasks.task_id,
          coalesce(tasks.provisioner_id, split_part(tasks.task_queue_id, '/', 1)),
          coalesce(tasks.worker_type, split_part(tasks.task_queue_id, '/', 2)),
          tasks.scheduler_id,
          tasks.task_group_id,
          tasks.dependencies,
          tasks.requires,
          tasks.routes,
          tasks.priority,
          tasks.retries,
          tasks.retries_left,
          tasks.created,
          tasks.deadline,
          tasks.expires,
          tasks.scopes,
          tasks.payload,
          tasks.metadata,
          tasks.tags,
          tasks.extra,
          tasks.runs,
          tasks.taken_until
        from tasks
        where
          tasks.task_id = task_id_in;
      end
  get_tasks_by_task_group:
    description: |-
      Get all properties of all tasks in the given task group.
    deprecated: true
    mode: read
    serviceName: queue
    args: task_group_id_in text, page_size_in integer, page_offset_in integer
    returns: |-
      table (
        task_id text,
        provisioner_id text,
        worker_type text,
        scheduler_id text,
        task_group_id text,
        dependencies jsonb,
        requires task_requires,
        routes jsonb,
        priority task_priority,
        retries integer,
        retries_left int,
        created timestamptz,
        deadline timestamptz,
        expires timestamptz,
        scopes jsonb,
        payload jsonb,
        metadata jsonb,
        tags jsonb,
        extra jsonb,
        runs jsonb,
        taken_until timestamptz
      )
    body: |-
      begin
        return query
        select
          tasks.task_id,
          coalesce(tasks.provisioner_id, split_part(tasks.task_queue_id, '/', 1)),
          coalesce(tasks.worker_type, split_part(tasks.task_queue_id, '/', 2)),
          tasks.scheduler_id,
          tasks.task_group_id,
          tasks.dependencies,
          tasks.requires,
          tasks.routes,
          tasks.priority,
          tasks.retries,
          tasks.retries_left,
          tasks.created,
          tasks.deadline,
          tasks.expires,
          tasks.scopes,
          tasks.payload,
          tasks.metadata,
          tasks.tags,
          tasks.extra,
          tasks.runs,
          tasks.taken_until
        from tasks
        where tasks.task_group_id = task_group_id_in
        limit get_page_limit(page_size_in)
        offset get_page_offset(page_offset_in);
      end