connector

Connector and retry/dead-letter contracts for M32 orchestration parity.

Types

ConnectorPlan

Full connector plan contract.

pub type ConnectorPlan {
  ConnectorPlan(
    name: String,
    source: SourceConnector,
    sink: SinkConnector,
    retry_policy: RetryPolicy,
    dead_letter_policy: DeadLetterPolicy,
    reprocess_window: option.Option(ReprocessWindow),
    max_batch_records: Int,
  )
}

Constructors

ConnectorPlan(
  name: String,
  source: SourceConnector,
  sink: SinkConnector,
  retry_policy: RetryPolicy,
  dead_letter_policy: DeadLetterPolicy,
  reprocess_window: option.Option(ReprocessWindow),
  max_batch_records: Int,
)

DeadLetterPolicy

Dead-letter policy after retry exhaustion.

pub type DeadLetterPolicy {
  DeadLetterPolicy(
    enabled: Bool,
    destination: String,
    max_payload_bytes: Int,
  )
}

Constructors

DeadLetterPolicy(
  enabled: Bool,
  destination: String,
  max_payload_bytes: Int,
)

FailureAction

Deterministic failure action.

pub type FailureAction {
  Retry(next_attempt: Int, at_ms: Int)
  DeadLetter(destination: String, reason: String)
  Reject(reason: String)
}

Constructors

```
Retry(next_attempt: Int, at_ms: Int)
```

DeadLetter(destination: String, reason: String)

```
Reject(reason: String)
```

ReprocessWindow

Optional reprocess-window contract.

pub type ReprocessWindow {
  ReprocessWindow(
    start_sequence: Int,
    end_sequence: Int,
    reason: String,
  )
}

Constructors

ReprocessWindow(
  start_sequence: Int,
  end_sequence: Int,
  reason: String,
)

RetryPolicy

Retry policy for connector failures.

pub type RetryPolicy {
  RetryPolicy(
    max_attempts: Int,
    base_backoff_ms: Int,
    max_backoff_ms: Int,
  )
}

Constructors

RetryPolicy(
  max_attempts: Int,
  base_backoff_ms: Int,
  max_backoff_ms: Int,
)

SinkConnector

Sink-side connector abstraction.

pub type SinkConnector {
  DatabaseSink(name: String, table: String, upsert_key: String)
  PubSubSink(name: String, topic: String, ordering_key: String)
}

Constructors

DatabaseSink(name: String, table: String, upsert_key: String)

PubSubSink(name: String, topic: String, ordering_key: String)

SourceConnector

Source-side connector abstraction.

pub type SourceConnector {
  DatabaseSource(name: String, query: String, batch_size: Int)
  FileSource(
    name: String,
    path: String,
    format: String,
    batch_size: Int,
  )
  QueueSource(name: String, topic: String, prefetch: Int)
}

Constructors

DatabaseSource(name: String, query: String, batch_size: Int)

FileSource(
  name: String,
  path: String,
  format: String,
  batch_size: Int,
)

QueueSource(name: String, topic: String, prefetch: Int)

Values

classify_failure

pub fn classify_failure(
  plan: ConnectorPlan,
  attempt: Int,
  reason: String,
  now_ms: Int,
) -> FailureAction

Determine failure action from retry and dead-letter policy.

connector_plan

pub fn connector_plan(
  name: String,
  source: SourceConnector,
  sink: SinkConnector,
  retry_policy: RetryPolicy,
  dead_letter_policy: DeadLetterPolicy,
  reprocess_window: option.Option(ReprocessWindow),
  max_batch_records: Int,
) -> ConnectorPlan

Build a connector plan.

database_sink

pub fn database_sink(
  name: String,
  table: String,
  upsert_key: String,
) -> SinkConnector

Build a database sink connector.

database_source

pub fn database_source(
  name: String,
  query: String,
  batch_size: Int,
) -> SourceConnector

Build a database source connector.

dead_letter_policy

pub fn dead_letter_policy(
  plan: ConnectorPlan,
) -> DeadLetterPolicy

Access dead-letter policy.

dead_letter_policy_label

pub fn dead_letter_policy_label(
  policy: DeadLetterPolicy,
) -> String

Dead-letter policy label.

default_plan

pub fn default_plan() -> ConnectorPlan

Default M32 connector plan.

failure_action_label

pub fn failure_action_label(action: FailureAction) -> String

Failure action label.

file_source

pub fn file_source(
  name: String,
  path: String,
  format: String,
  batch_size: Int,
) -> SourceConnector

Build a file source connector.

max_batch_records

pub fn max_batch_records(plan: ConnectorPlan) -> Int

Access max batch records.

pubsub_sink

pub fn pubsub_sink(
  name: String,
  topic: String,
  ordering_key: String,
) -> SinkConnector

Build a pubsub sink connector.

queue_source

pub fn queue_source(
  name: String,
  topic: String,
  prefetch: Int,
) -> SourceConnector

Build a queue source connector.

retry_policy

pub fn retry_policy(plan: ConnectorPlan) -> RetryPolicy

Access retry policy.

retry_policy_label

pub fn retry_policy_label(policy: RetryPolicy) -> String

Retry policy label.

signature

pub fn signature(plan: ConnectorPlan) -> String

Stable connector-plan signature.

sink

pub fn sink(plan: ConnectorPlan) -> SinkConnector

Access sink connector.

sink_label

pub fn sink_label(sink: SinkConnector) -> String

Sink connector label.

source

pub fn source(plan: ConnectorPlan) -> SourceConnector

Access source connector.

source_label

pub fn source_label(source: SourceConnector) -> String

Source connector label.

valid

pub fn valid(plan: ConnectorPlan) -> Bool

Validate connector plan invariants.

within_reprocess_window

pub fn within_reprocess_window(
  plan: ConnectorPlan,
  sequence: Int,
) -> Bool

Check whether one sequence is allowed by the optional reprocess window.