0022.yml

version: 22
description: "queue task dependencies phase 2 step 1 (slow migration: 10μs per row in table `queue_task_dependency_entities`)"
migrationScript: 0022-migration.sql
downgradeScript: 0022-downgrade.sql
methods:

  ####
  # queue_task_dependency_entities

  queue_task_dependency_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
          required_task_id,
          dependent_task_id,
          jsonb_build_object(
            'PartitionKey', required_task_id,
            'RowKey', dependent_task_id,
            'taskId', slugid_to_uuid(required_task_id),
            'dependentTaskId', slugid_to_uuid(dependent_task_id),
            'require', replace(requires::text, 'all-', ''),
            'expires', expires) as value,
          1 as version,
          task_dependencies.etag as etag
        from task_dependencies
        where
          required_task_id = partition_key and dependent_task_id = row_key;
      end
  queue_task_dependency_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_etag uuid;
      begin
        -- note that this function always overwrites (queue always calls it that way anyway)
        new_etag = public.gen_random_uuid();
        insert into task_dependencies select
          rk,
          pk,
          ('all-' || (properties ->> 'require'))::task_requires,
          true,
          (properties ->> 'expires')::timestamptz,
          new_etag
        -- if the row already exists, that's because it was created by
        -- queue_task_requirement_entities_create, so just update the requires value
        on conflict (required_task_id, dependent_task_id) do
          update
          set requires = ('all-' || (properties ->> 'require'))::task_requires
          where
            task_dependencies.required_task_id = pk and
            task_dependencies.dependent_task_id = rk;
        return new_etag;
      end
  queue_task_dependency_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 task_dependencies
        where
          task_dependencies.required_task_id = partition_key and
          task_dependencies.dependent_task_id = row_key
        returning task_dependencies.etag;
      end
  queue_task_dependency_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 'not implemented';
      end
  queue_task_dependency_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 ('expires<', 'expires>=') then
            raise exception 'scan only supports filtering for expired rows, not on %', exp_cond_field || exp_cond_operator;
          end if;
        end if;

        return query select
          required_task_id as partition_key,
          dependent_task_id as row_key,
          jsonb_build_object(
            'PartitionKey', required_task_id,
            'RowKey', dependent_task_id,
            'taskId', slugid_to_uuid(required_task_id),
            'dependentTaskId', slugid_to_uuid(dependent_task_id),
            'require', replace(requires::text, 'all-', ''),
            'expires', expires) as value,
          1 as version,
          task_dependencies.etag as etag from task_dependencies
        where
          (pk is null or required_task_id = pk) and
          case
            when exp_cond_field = 'expires' and exp_cond_operator = '<' then expires < exp_cond_operand
            when exp_cond_field = 'expires' and exp_cond_operator = '>=' then expires >= exp_cond_operand
            else true
          end
        order by task_dependencies.required_task_id, task_dependencies.dependent_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_dependency_entities
  #
  # This is synthesized from the task_dependencies table

  queue_task_requirement_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
          dependent_task_id,
          required_task_id,
          jsonb_build_object(
            'PartitionKey', dependent_task_id,
            'RowKey', required_task_id,
            'taskId', slugid_to_uuid(dependent_task_id),
            'requiredTaskId', slugid_to_uuid(required_task_id),
            'expires', expires) as value,
          1 as version,
          task_dependencies.etag as etag
        from task_dependencies
        where
          dependent_task_id = partition_key and required_task_id = row_key and not satisfied;
      end
  queue_task_requirement_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_etag uuid;
      begin
        -- note that this function always overwrites (queue always calls it that way anyway)
        new_etag = public.gen_random_uuid();
        insert into task_dependencies select
          pk,
          rk,
          -- arbitrary value; in practice TaskDependency.create is called shortly
          -- after this and will set the value correctly
          'all-completed',
          -- if a TaskRequirement exists, then the dependency is not satisified
          false,
          (properties ->> 'expires')::timestamptz,
          new_etag
        -- if the dependency was created already, just update satisfied
        on conflict (required_task_id, dependent_task_id) do
          update
          set satisfied = false
          where
            task_dependencies.required_task_id = rk and
            task_dependencies.dependent_task_id = pk;
        return new_etag;
      end
  queue_task_requirement_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
        -- removing the requirement means that this dep is satisfied
        return query
        update task_dependencies
        set satisfied = true
        where
          task_dependencies.dependent_task_id = partition_key and
          task_dependencies.required_task_id = row_key and
          not task_dependencies.satisfied
        returning task_dependencies.etag;
      end
  queue_task_requirement_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 'not implemented';
      end
  queue_task_requirement_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 exp_cond_field || exp_cond_operator != 'expires<' then
            raise exception 'scan only supports filtering for expired rows';
          end if;

          -- if this is the expiration crontask, return an empty set -- expiration of TaskDependency
          -- is sufficient
          return;
        end if;

        return query select
          dependent_task_id as partition_key,
          required_task_id as row_key,
          jsonb_build_object(
            'PartitionKey', dependent_task_id,
            'RowKey', required_task_id,
            'taskId', slugid_to_uuid(dependent_task_id),
            'requiredTaskId', slugid_to_uuid(required_task_id),
            'expires', expires) as value,
          1 as version,
          task_dependencies.etag as etag from task_dependencies
        where
          (pk is null or dependent_task_id = pk) and
          not satisfied
        order by task_dependencies.dependent_task_id, task_dependencies.required_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