Turbo Pipelines (turbo)

Manage Turbo Pipelines with “turbo”

ctx.turbo lets a task create and manage Turbo pipelines and jobs from inside your Compose app. Use it to spin up a pipeline in response to an on-chain event, kick off a one-time backfill and poll it to completion, or pause and clean up pipelines your app owns. A Turbo job is a pipeline whose definition has job: true — it runs to completion and then exits, rather than streaming forever. ctx.turbo.createJob is a shortcut for createPipeline with that flag set.

How pipelines are named and cleaned up

Pipelines you create through ctx.turbo are namespaced under your Compose app. You pass a plain name like token-transfers, and the pipeline is registered internally with your app’s prefix. Every method handles the prefix for you, so you always refer to pipelines by the plain name you gave them, and listPipelines only returns pipelines that belong to your app. The upshot is that when your Compose app is deleted, the pipelines it created are deleted with it — you don’t have to track and tear them down yourself.

The namespaced name (app prefix included) has a length limit. If your app name plus the pipeline name is too long, createPipeline throws. Keep pipeline names short and the error message tells you the maximum.

Methods

turbo: {
  createPipeline(config: CreateTurboPipelineConfig, retryConfig?): Promise<TurboPipeline>;
  createJob(config: CreateTurboJobConfig, retryConfig?): Promise<TurboPipeline>;
  getPipeline(name: string, retryConfig?): Promise<TurboPipeline>;
  getStatus(name: string, retryConfig?): Promise<TurboPipelineStatus>;
  listPipelines(retryConfig?): Promise<TurboPipeline[]>;
  pausePipeline(name: string, retryConfig?): Promise<PipelineActionResult>;
  resumePipeline(name: string, retryConfig?): Promise<PipelineActionResult>;
  restartPipeline(name: string, options?: RestartOptions, retryConfig?): Promise<PipelineActionResult>;
  deletePipeline(name: string, retryConfig?): Promise<PipelineDeleteResult>;
  getLogs(name: string, options?: GetLogsOptions, retryConfig?): Promise<PipelineLogsResult>;
  getState(name: string, retryConfig?): Promise<unknown>;
  validate(definition: TurboPipelineDefinition, retryConfig?): Promise<ValidateResult>;
}

Config types

export type TurboPipelineDefinition = {
  sources: Record<string, unknown>;
  transforms: Record<string, unknown>;
  sinks: Record<string, unknown>;
  job?: boolean;
};

export type CreateTurboPipelineConfig = {
  name: string; // lowercase letters, numbers, and hyphens
  resourceSize?: string;
  description?: string;
  useDedicatedIp?: boolean;
  definition: TurboPipelineDefinition;
};

// createJob takes the same config — the client sets job: true for you.
export type CreateTurboJobConfig = Omit<CreateTurboPipelineConfig, "definition"> & {
  definition: Omit<TurboPipelineDefinition, "job">;
};

export type RestartOptions = { clearState?: boolean };

export type GetLogsOptions = {
  logLevels?: string; // comma-separated, e.g. "error,warn"
  cursor?: number;
  after?: number;
  search?: string;
  direction?: "asc" | "desc";
};

Response types

export type TurboPipeline = {
  name: string;
  status?: string;
  version?: number;
  definition?: TurboPipelineDefinition & {
    name?: string;
    resource_size?: string;
    description?: string;
    use_dedicated_ip?: boolean;
  };
  created_at: string;
  updated_at?: string;
};

export type TurboPipelineStatus = {
  name: string;
  status: string;
};

export type ValidateResult = {
  valid: boolean;
  errors?: Array<{ message: string } | string>;
  warnings?: Array<{ message: string } | string>;
};

export type PipelineActionResult = {
  name: string;
  action: string;
};

export type PipelineDeleteResult = {
  name: string;
  deleted: boolean;
};

export type PipelineLogsResult = {
  logs: unknown[];
};

The definition is the same pipeline configuration you’d write in YAML, expressed as an object. Reads (getPipeline, getStatus, listPipelines, getLogs, getState, validate) retry on transient failures by default; the mutating methods do not. Every call is recorded as a span on the task run, like ctx.fetch.

Examples

Kick off a one-time backfill (job)

import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const job = await turbo.createJob({
    name: "solana-backfill",
    resourceSize: "m",
    definition: {
      sources: {
        token_transfers: {
          type: "dataset",
          dataset_name: "solana.token_transfers",
          version: "1.0.0",
          start_block: 250000000,
          end_block: 250000002,
        },
      },
      transforms: {
        filtered: { type: "sql", primary_key: "id", sql: "SELECT * FROM token_transfers" },
      },
      sinks: {
        out: { type: "postgres", from: "filtered", schema: "public", table: "token_transfers", secret_name: "MY_PG", primary_key: "id" },
      },
    },
  });

  return job;
}

Create a streaming pipeline in response to an event

import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext, params: { contract: string }) {
  // Sanitize the contract address for use in the pipeline name
  const slug = params.contract.slice(2, 10).toLowerCase();

  return turbo.createPipeline({
    name: `transfers-${slug}`,
    definition: {
      sources: {
        logs: { type: "dataset", dataset_name: "ethereum.raw_logs", version: "1.0.0", start_at: "latest" },
      },
      transforms: {
        filtered: {
          type: "sql",
          primary_key: "id",
          sql: `SELECT * FROM logs WHERE address = lower('${params.contract}')`,
        },
      },
      sinks: {
        out: { type: "postgres", from: "filtered", schema: "public", table: "transfers", secret_name: "MY_PG", primary_key: "id" },
      },
    },
  });
}

Poll a job until it finishes

import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const { status } = await turbo.getStatus("solana-backfill");

  if (status === "STOPPED" || status === "FAILED") {
    const logs = await turbo.getLogs("solana-backfill", { logLevels: "error" });
    return { done: true, status, logs };
  }

  return { done: false, status };
}

Pair this with a cron trigger to check on a long-running job on a schedule instead of blocking a single task run.

Validate before deploying

import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const result = await turbo.validate({
    sources: { blocks: { type: "dataset", dataset_name: "ethereum.raw_blocks", version: "1.0.0" } },
    transforms: { t: { type: "sql", primary_key: "id", sql: "SELECT * FROM blocks" } },
    sinks: { out: { type: "blackhole", from: "t" } },
  });

  if (!result.valid) {
    const messages = (result.errors ?? []).map((e) =>
      typeof e === "string" ? e : e.message
    );
    throw new Error(messages.join("; "));
  }
}

List, pause, and delete the pipelines your app owns

import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const pipelines = await turbo.listPipelines();

  for (const p of pipelines) {
    await turbo.pausePipeline(p.name);
  }

  await turbo.deletePipeline("solana-backfill");
}

Compose

Guides

Reference

Manage Turbo Pipelines with “turbo”

How pipelines are named and cleaned up

Methods

Config types

Response types

Examples

Kick off a one-time backfill (job)

Create a streaming pipeline in response to an event

Poll a job until it finishes

Validate before deploying

List, pause, and delete the pipelines your app owns

Next Steps

Turbo Pipelines

Job mode

​Manage Turbo Pipelines with “turbo”

​How pipelines are named and cleaned up

​Methods

​Config types

​Response types

​Examples

​Kick off a one-time backfill (job)

​Create a streaming pipeline in response to an event

​Poll a job until it finishes

​Validate before deploying

​List, pause, and delete the pipelines your app owns

​Next Steps

Turbo Pipelines

Job mode

Manage Turbo Pipelines with “turbo”

How pipelines are named and cleaned up

Methods

Config types

Response types

Examples

Kick off a one-time backfill (job)

Create a streaming pipeline in response to an event

Poll a job until it finishes

Validate before deploying

List, pause, and delete the pipelines your app owns

Next Steps