lightspeed/pipeline

Integrated data-pipeline core contracts for M31.

Types

Boundary

Stage boundary contracts.

pub type Boundary {
  Boundary(input_contract: String, output_contract: String)
}

Constructors

Boundary(input_contract: String, output_contract: String)

Lifecycle

Pipeline lifecycle.

pub type Lifecycle {
  Idle
  Scheduled(at_ms: Int)
  Triggered(reason: String, at_ms: Int)
  Running(run_id: String, started_ms: Int)
  Failed(run_id: String, reason: String)
  Crashed(run_id: String, reason: String)
  Replaying(run_id: String, from_stage: String, started_ms: Int)
  Completed(run_id: String, finished_ms: Int)
}

Constructors

```
Idle
```
```
Scheduled(at_ms: Int)
```
```
Triggered(reason: String, at_ms: Int)
```

Running(run_id: String, started_ms: Int)

```
Failed(run_id: String, reason: String)
```
```
Crashed(run_id: String, reason: String)
```

Replaying(run_id: String, from_stage: String, started_ms: Int)

Completed(run_id: String, finished_ms: Int)

Pipeline

Pipeline contract.

pub type Pipeline {
  Pipeline(name: String, schedule: Schedule, stages: List(Stage))
}

Constructors

Pipeline(name: String, schedule: Schedule, stages: List(Stage))

ProcessResult

Stage processing result.

pub type ProcessResult {
  Applied(checkpoint: checkpoint.Checkpoint)
  DuplicateSuppressed(stage: String, idempotency_key: String)
  Rejected(reason: String)
}

Constructors

Applied(checkpoint: checkpoint.Checkpoint)

DuplicateSuppressed(stage: String, idempotency_key: String)

```
Rejected(reason: String)
```

Runtime

opaque

Pipeline runtime.

pub opaque type Runtime

Schedule

Pipeline schedule profile.

pub type Schedule {
  Manual
  Interval(interval_ms: Int)
  Cron(expression: String)
}

Constructors

```
Manual
```
```
Interval(interval_ms: Int)
```
```
Cron(expression: String)
```

Stage

One pipeline stage.

pub type Stage {
  Stage(name: String, kind: StageKind, boundary: Boundary)
}

Constructors

Stage(name: String, kind: StageKind, boundary: Boundary)

StageKind

Stage category.

pub type StageKind {
  Source
  Transform
  Sink
}

Constructors

```
Source
```
```
Transform
```
```
Sink
```

Values

checkpoints

pub fn checkpoints(
  runtime: Runtime,
) -> List(checkpoint.Checkpoint)

Stable checkpoints in emit order.

complete

pub fn complete(runtime: Runtime, now_ms: Int) -> Runtime

Complete the active run.

crash

pub fn crash(runtime: Runtime, reason: String) -> Runtime

Mark active run as crashed.

fail

pub fn fail(runtime: Runtime, reason: String) -> Runtime

Mark active run as failed.

lifecycle

pub fn lifecycle(runtime: Runtime) -> Lifecycle

Runtime lifecycle accessor.

lifecycle_label

pub fn lifecycle_label(lifecycle: Lifecycle) -> String

Lifecycle label.

new

pub fn new(pipeline: Pipeline) -> Runtime

Build one runtime.

pipeline

pub fn pipeline(
  name: String,
  schedule: Schedule,
  stages: List(Stage),
) -> Pipeline

Build one pipeline contract.

pipeline_signature

pub fn pipeline_signature(pipeline: Pipeline) -> String

Stable pipeline signature.

process

pub fn process(
  runtime: Runtime,
  stage_name: String,
  processed_records: Int,
  lag_ms: Int,
  idempotency_key: String,
  now_ms: Int,
) -> #(Runtime, ProcessResult)

Process one stage event.

process_result_label

pub fn process_result_label(result: ProcessResult) -> String

Stable process-result label.

record_dead_letter

pub fn record_dead_letter(
  runtime: Runtime,
  count: Int,
) -> Runtime

Record dead-letter increments for the active run.

record_retry

pub fn record_retry(runtime: Runtime, count: Int) -> Runtime

Record retries for the active run.

resume_from_latest_checkpoint

pub fn resume_from_latest_checkpoint(
  runtime: Runtime,
  now_ms: Int,
) -> Result(Runtime, String)

Resume from the latest checkpoint.

runtime_pipeline

pub fn runtime_pipeline(runtime: Runtime) -> Pipeline

Runtime pipeline accessor.

runtime_telemetry

pub fn runtime_telemetry(runtime: Runtime) -> telemetry.Snapshot

Runtime telemetry accessor.

schedule

pub fn schedule(runtime: Runtime, at_ms: Int) -> Runtime

Schedule one runtime.

schedule_label

pub fn schedule_label(schedule: Schedule) -> String

Schedule label.

signature

pub fn signature(runtime: Runtime) -> String

Stable runtime signature.

sink_idempotency_keys

pub fn sink_idempotency_keys(runtime: Runtime) -> List(String)

Sink idempotency keys in application order.

sink_stage

pub fn sink_stage(name: String, input_contract: String) -> Stage

Build a sink stage.

source_stage

pub fn source_stage(
  name: String,
  output_contract: String,
) -> Stage

Build a source stage.

stage_kind_label

pub fn stage_kind_label(kind: StageKind) -> String

Stage kind label.

start

pub fn start(runtime: Runtime, now_ms: Int) -> Runtime

Start one run.

transform_stage

pub fn transform_stage(
  name: String,
  input_contract: String,
  output_contract: String,
) -> Stage

Build a transform stage.

trigger

pub fn trigger(
  runtime: Runtime,
  reason: String,
  at_ms: Int,
) -> Runtime

Trigger one runtime.

valid

pub fn valid(runtime: Runtime) -> Bool

Validate pipeline and runtime invariants.

valid_pipeline

pub fn valid_pipeline(pipeline: Pipeline) -> Bool

Validate one pipeline contract.