aarondb/projection

projection — supervised, checkpointed derived-state reference

This pure module models the durable projection contract. Delivery is at-least-once; apply deduplicates offsets and commits its checkpoint only after the derived state has accepted the entry.

Types

Failure

pub type Failure {
  Retryable(reason: String)
  Permanent(reason: String)
  SourceUnavailable(reason: String)
  CorruptSource(offset: Int)
  PressureExceeded(limit: Int)
}

Constructors

```
Retryable(reason: String)
```
```
Permanent(reason: String)
```
```
SourceUnavailable(reason: String)
```
```
CorruptSource(offset: Int)
```
```
PressureExceeded(limit: Int)
```

Metrics

pub type Metrics {
  Metrics(
    applied: Int,
    duplicates: Int,
    restarts: Int,
    rebuilds: Int,
  )
}

Constructors

Metrics(
  applied: Int,
  duplicates: Int,
  restarts: Int,
  rebuilds: Int,
)

Projection

pub type Projection {
  Projection(
    name: String,
    generation: Int,
    max_batch: Int,
    max_restarts: Int,
    state: ProjectionState,
    last_applied_offset: Int,
    applied_offsets: List(Int),
    values: List(String),
    failure: option.Option(Failure),
    metrics: Metrics,
  )
}

Constructors

Projection(
  name: String,
  generation: Int,
  max_batch: Int,
  max_restarts: Int,
  state: ProjectionState,
  last_applied_offset: Int,
  applied_offsets: List(Int),
  values: List(String),
  failure: option.Option(Failure),
  metrics: Metrics,
)

ProjectionError

pub type ProjectionError {
  BoundedPressure(limit: Int)
  PermanentlyFailed(Failure)
  Source(durable_log.DurableLogError)
  Checkpoint(durable_log.DurableLogError)
}

Constructors

```
BoundedPressure(limit: Int)
```
```
PermanentlyFailed(Failure)
```
```
Source(durable_log.DurableLogError)
```
```
Checkpoint(durable_log.DurableLogError)
```

ProjectionState

pub type ProjectionState {
  Running
  Paused
  Replaying
  Failed
  Rebuilding
  Stale
}

Constructors

```
Running
```
```
Paused
```
```
Replaying
```
```
Failed
```
```
Rebuilding
```
```
Stale
```

Status

pub type Status {
  Status(
    state: ProjectionState,
    last_applied_offset: Int,
    lag: Int,
    failure: option.Option(Failure),
    generation: Int,
    metrics: Metrics,
  )
}

Constructors

Status(
  state: ProjectionState,
  last_applied_offset: Int,
  lag: Int,
  failure: option.Option(Failure),
  generation: Int,
  metrics: Metrics,
)

Values

catch_up

pub fn catch_up(
  projection: Projection,
  source: durable_log.DurableLog,
) -> Result(
  #(Projection, durable_log.DurableLog),
  ProjectionError,
)

Apply a bounded source tail, preserving atomic state/checkpoint ordering.

new

pub fn new(
  name: String,
  max_batch: Int,
  max_restarts: Int,
) -> Projection

rebuild

pub fn rebuild(
  projection: Projection,
  source: durable_log.DurableLog,
  snapshot: durable_log.Snapshot,
) -> Result(
  #(Projection, durable_log.DurableLog),
  ProjectionError,
)

Rebuild creates an isolated next generation from a snapshot boundary then swaps it.

restart

pub fn restart(
  projection: Projection,
  reason: String,
) -> Projection

A retryable crash increments supervision state; exhaustion is permanent and visible.

resume

pub fn resume(projection: Projection) -> Projection

status

pub fn status(
  projection: Projection,
  source: durable_log.DurableLog,
) -> Status

values

pub fn values(projection: Projection) -> List(String)