0020.yml

version: 20
description: "queue tasks phase 2 step 1 (slow migration: 700-2000μs per row in table `queue_tasks_entities`)"
migrationScript: 0020-migration.sql
downgradeScript: 0020-downgrade.sql
methods:

  ####
  # queue_tasks_entities

  queue_tasks_entities_load:
    deprecated: true
    description: See taskcluster-lib-entities
    mode: read
    serviceName: queue
    args: partition_key text, row_key text
    returns: table (partition_key_out text, row_key_out text, value jsonb, version integer, etag uuid)
    body: |-
      begin
        return query
        select
          queue_tasks_entities_load.partition_key,
          queue_tasks_entities_load.row_key,
          entity_buf_encode(
            entity_buf_encode(
              entity_buf_encode(
                entity_buf_encode(
                  entity_buf_encode(
                    entity_buf_encode(
                      entity_buf_encode(
                        entity_buf_encode(
                          jsonb_build_object(
                            'PartitionKey', task_id,
                            'RowKey', 'task',
                            'taskId', slugid_to_uuid(task_id),
                            'provisionerId', provisioner_id,
                            'workerType', worker_type,
                            'schedulerId', scheduler_id,
                            'taskGroupId', slugid_to_uuid(task_group_id),
                            'requires', requires::text,
                            'priority', priority::text,
                            'retries', retries,
                            'retriesLeft', retries_left,
                            'created', created,
                            'deadline', deadline,
                            'expires', expires,
                            'takenUntil', coalesce(taken_until, '1970-01-01 00:00:00+00'::timestamptz)),
                          'dependencies', dependencies::text),
                        'routes', routes::text),
                      'scopes', scopes::text),
                    'payload', payload::text),
                  'metadata', metadata::text),
                'tags', tags::text),
              'extra', extra::text),
            'runs', runs::text) as value,
          1 as version,
          tasks.etag as etag
        from tasks
        where
          tasks.task_id = partition_key;
      end
  queue_tasks_entities_create:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: pk text, rk text, properties jsonb, overwrite boolean, version integer
    returns: uuid
    body: |-
      declare
        new_row tasks%ROWTYPE;
      begin
        select
          uuid_to_slugid(properties ->> 'taskId'),
          (properties ->> 'provisionerId')::text,
          (properties ->> 'workerType')::text,
          (properties ->> 'schedulerId')::text,
          uuid_to_slugid(properties ->> 'taskGroupId')::text,
          entity_buf_decode(properties, 'dependencies')::jsonb,
          (properties ->> 'requires')::task_requires,
          entity_buf_decode(properties, 'routes')::jsonb,
          (properties ->> 'priority')::task_priority,
          (properties ->> 'retries')::int,
          (properties ->> 'retriesLeft')::int,
          (properties ->> 'created')::timestamptz,
          (properties ->> 'deadline')::timestamptz,
          (properties ->> 'expires')::timestamptz,
          entity_buf_decode(properties, 'scopes')::jsonb,
          entity_buf_decode(properties, 'payload')::jsonb,
          entity_buf_decode(properties, 'metadata')::jsonb,
          entity_buf_decode(properties, 'tags')::jsonb,
          entity_buf_decode(properties, 'extra')::jsonb,
          entity_buf_decode(properties, 'runs')::jsonb,
          case (properties ->> 'takenUntil')::timestamptz
            when '1970-01-01 00:00:00+00'::timestamptz then null
            else (properties ->> 'takenUntil')::timestamptz
          end,
          false,
          public.gen_random_uuid()
        into new_row;
        if overwrite then
          raise exception 'overwrite not implemented';
        else
          execute 'insert into tasks select $1.*' using new_row;
        end if;
        return new_row.etag;
      end
  queue_tasks_entities_remove:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: partition_key text, row_key text
    returns: table (etag uuid)
    body: |-
      begin
        return query delete from tasks
        where
          tasks.task_id = partition_key
        returning tasks.etag;
      end
  queue_tasks_entities_modify:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: partition_key text, row_key text, properties jsonb, version integer, old_etag uuid
    returns: table (etag uuid)
    body: |-
      declare
        new_etag uuid;
      begin
        -- NOTE: queue only updates runs, retriesLeft, and takenUntil, so only those fields are
        -- supported here.
        new_etag = public.gen_random_uuid();
        update tasks
        set (
          runs,
          retries_left,
          taken_until,
          etag
        ) = (
          entity_buf_decode(properties, 'runs')::jsonb,
          (properties ->> 'retriesLeft')::int,
          case (properties ->> 'takenUntil')::timestamptz
            when '1970-01-01 00:00:00+00'::timestamptz then null
            else (properties ->> 'takenUntil')::timestamptz
          end,
          new_etag
        )
        where
          tasks.task_id = partition_key and
          tasks.etag = queue_tasks_entities_modify.old_etag;
        if found then
          return query select new_etag;
          return;
        end if;
        perform tasks.etag from tasks
        where
          tasks.task_id = partition_key;
        if found then
          raise exception 'unsuccessful update' using errcode = 'P0004';
        else
          raise exception 'no such row' using errcode = 'P0002';
        end if;
      end
  queue_tasks_entities_scan:
    deprecated: true
    description: See taskcluster-lib-entities
    mode: read
    serviceName: queue
    args: pk text, rk text, condition text, size integer, page integer
    returns: table (partition_key text, row_key text, value jsonb, version integer, etag uuid)
    body: |-
      declare
        cond text[];
        exp_cond_field text;
        exp_cond_operator text;
        exp_cond_operand timestamptz;
      begin
        if not condition is null then
          cond := regexp_split_to_array(condition, '\s+');
          exp_cond_field := trim(cond[3], '''');
          exp_cond_operator := cond[4];
          exp_cond_operand := cond[5] :: timestamptz;

          if not (exp_cond_field || exp_cond_operator) in ('takenUntil=', 'deadline=', 'expires<') then
            raise exception 'queue_tasks_entities_scan only supports certain takenUntil, deadline, and expires conditions';
          end if;
        end if;

        return query select
          task_id as partition_key,
          'task' as row_key,
          entity_buf_encode(
            entity_buf_encode(
              entity_buf_encode(
                entity_buf_encode(
                  entity_buf_encode(
                    entity_buf_encode(
                      entity_buf_encode(
                        entity_buf_encode(
                          jsonb_build_object(
                            'PartitionKey', task_id,
                            'RowKey', 'task',
                            'taskId', slugid_to_uuid(task_id),
                            'provisionerId', provisioner_id,
                            'workerType', worker_type,
                            'schedulerId', scheduler_id,
                            'taskGroupId', slugid_to_uuid(task_group_id),
                            'requires', requires::text,
                            'priority', priority::text,
                            'retries', retries,
                            'retriesLeft', retries_left,
                            'created', created,
                            'deadline', deadline,
                            'expires', expires,
                            'takenUntil', coalesce(taken_until, '1970-01-01 00:00:00+00'::timestamptz)),
                          'dependencies', dependencies::text),
                        'routes', routes::text),
                      'scopes', scopes::text),
                    'payload', payload::text),
                  'metadata', metadata::text),
                'tags', tags::text),
              'extra', extra::text),
            'runs', runs::text) as value,
          1 as version,
          tasks.etag as etag from tasks
        where
          (pk is NULL or pk = task_id) and
          case
            when exp_cond_field = 'deadline' then deadline = exp_cond_operand
            -- note that queue never queries for takenUntil = new Date(0)
            when exp_cond_field = 'takenUntil' then taken_until = exp_cond_operand
            when exp_cond_field = 'expires' then expires < exp_cond_operand
            else true
          end
        order by tasks.task_id
        limit case
          when (size is not null and size > 0) then size + 1
          else null
        end
        offset case
          when (page is not null and page > 0) then page
          else 0
        end;
      end

  ####
  # queue_task_group_members_entities
  #
  # This table is just an index into tasks by taskGroupId, and is now read-only -- writes
  # are just ignored.

  queue_task_group_members_entities_load:
    deprecated: true
    description: See taskcluster-lib-entities
    mode: read
    serviceName: queue
    args: partition_key text, row_key text
    returns: table (partition_key_out text, row_key_out text, value jsonb, version integer, etag uuid)
    body: |-
      begin
        return query
        select
          task_group_id,
          task_id,
          jsonb_build_object(
            'PartitionKey', task_group_id,
            'RowKey', task_id,
            'taskId', slugid_to_uuid(task_id),
            'taskGroupId', slugid_to_uuid(task_group_id),
            'expires', expires) as value,
          1 as version,
          tasks.etag as etag
        from tasks
        where
          tasks.task_group_id = partition_key and
          tasks.task_id = row_key;
      end
  queue_task_group_members_entities_create:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: pk text, rk text, properties jsonb, overwrite boolean, version integer
    returns: uuid
    body: |-
      begin
        return public.gen_random_uuid();
      end
  queue_task_group_members_entities_remove:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: partition_key text, row_key text
    returns: table (etag uuid)
    body: |-
      begin
        return query
        select tasks.etag
        from tasks
        where
          tasks.task_group_id = partition_key and
          tasks.task_id = row_key;
      end
  queue_task_group_members_entities_modify:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: partition_key text, row_key text, properties jsonb, version integer, old_etag uuid
    returns: table (etag uuid)
    body: |-
      begin
        raise exception 'modify not implemented for queue_task_group_members_entities';
      end
  queue_task_group_members_entities_scan:
    deprecated: true
    description: See taskcluster-lib-entities
    mode: read
    serviceName: queue
    args: pk text, rk text, condition text, size integer, page integer
    returns: table (partition_key text, row_key text, value jsonb, version integer, etag uuid)
    body: |-
      declare
        cond text[];
        exp_cond_field text;
        exp_cond_operator text;
        exp_cond_operand timestamptz;
      begin
        if pk is null then
          -- if pk is null, then this is the expiration scan for the members table; since
          -- there is no members table, we stub this out by simply returning an empty set,
          -- ignoring the condition
          return;
        end if;

        if not condition is null then
          cond := regexp_split_to_array(condition, '\s+');
          exp_cond_field := trim(cond[3], '''');
          exp_cond_operator := cond[4];
          exp_cond_operand := cond[5] :: timestamptz;

          if exp_cond_operator != '>=' or exp_cond_field != 'expires' then
            raise exception 'queue_task_group_memberss_entities_scan only supports `expires >= <timestamp>` conditions';
          end if;
        end if;

        return query
        select
          task_group_id,
          task_id,
          jsonb_build_object(
            'PartitionKey', task_group_id,
            'RowKey', task_id,
            'taskId', slugid_to_uuid(task_id),
            'taskGroupId', slugid_to_uuid(task_group_id),
            'expires', expires) as value,
          1 as version,
          tasks.etag as etag
        from tasks
        where
          tasks.task_group_id = pk and
          (exp_cond_operand is NULL or expires >= exp_cond_operand)
        order by tasks.task_id
        limit case
          when (size is not null and size > 0) then size + 1
          else null
        end
        offset case
          when (page is not null and page > 0) then page
          else 0
        end;
      end

  ####
  # queue_task_group_active_sets_entities
  #
  # This table is just an index into tasks by taskGroupId, and is now read-only -- writes
  # and removals are translated to changes to the `ever_resolved` column of the tasks table.

  queue_task_group_active_sets_entities_load:
    deprecated: true
    description: See taskcluster-lib-entities
    mode: read
    serviceName: queue
    args: partition_key text, row_key text
    returns: table (partition_key_out text, row_key_out text, value jsonb, version integer, etag uuid)
    body: |-
      begin
        return query
        select
          task_group_id,
          task_id,
          jsonb_build_object(
            'PartitionKey', task_group_id,
            'RowKey', task_id,
            'taskId', slugid_to_uuid(task_id),
            'taskGroupId', slugid_to_uuid(task_group_id),
            'expires', expires) as value,
          1 as version,
          tasks.etag as etag
        from tasks
        where
          not ever_resolved and
          tasks.task_group_id = partition_key and
          tasks.task_id = row_key;
      end
  queue_task_group_active_sets_entities_create:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: pk text, rk text, properties jsonb, overwrite boolean, version integer
    returns: uuid
    body: |-
      begin
        -- this method is only called before the task is created, and tasks are
        -- created in an unresolved state, so there's nothing to do here.  Note
        -- that the service does not mark a task as unresolved when rerunning it
        return public.gen_random_uuid();
      end
  queue_task_group_active_sets_entities_remove:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: partition_key text, row_key text
    returns: table (etag uuid)
    body: |-
      begin
        -- DependencyTracker uses this to mark a task as ever_resolved
        update tasks
        set ever_resolved = true
        where
          tasks.task_group_id = partition_key and
          tasks.task_id = row_key;
        return query select public.gen_random_uuid() as etag;
      end
  queue_task_group_active_sets_entities_modify:
    deprecated: true
    serviceName: queue
    description: See taskcluster-lib-entities
    mode: write
    args: partition_key text, row_key text, properties jsonb, version integer, old_etag uuid
    returns: table (etag uuid)
    body: |-
      begin
        raise exception 'modify not supported';
      end
  queue_task_group_active_sets_entities_scan:
    deprecated: true
    description: See taskcluster-lib-entities
    mode: read
    serviceName: queue
    args: pk text, rk text, condition text, size integer, page integer
    returns: table (partition_key text, row_key text, value jsonb, version integer, etag uuid)
    body: |-
      declare
        cond text[];
        exp_cond_field text;
        exp_cond_operator text;
        exp_cond_operand timestamptz;
      begin
        if not condition is null then
          cond := regexp_split_to_array(condition, '\s+');
          exp_cond_field := trim(cond[3], '''');

          if exp_cond_field != 'expires' then
            raise exception 'queue_task_group_active_sets_entities_scan only supports filtering for expired rows';
          end if;

          -- there's no distinct table to expire, so just return an empty set to make the
          -- expiration crontask finish immediately
          return;
        end if;

        -- the idea here is to use the tasks_task_group_id_unresolved_idx index for a quick response
        return query select
          task_group_id as partition_key,
          task_id as row_key,
          jsonb_build_object(
            'PartitionKey', task_group_id,
            'RowKey', task_id,
            'taskGroupId', slugid_to_uuid(task_group_id),
            'taskId', slugid_to_uuid(task_id),
            'expires', expires) as value,
          1 as version,
          public.gen_random_uuid() as etag
        from tasks
        where
          (pk is NULL or pk = task_group_id) and
          not ever_resolved
        order by tasks.task_group_id, tasks.task_id
        limit case
          when (size is not null and size > 0) then size + 1
          else null
        end
        offset case
          when (page is not null and page > 0) then page
          else 0
        end;
      end