constellation/worker_pool

Resilient OTP worker pools backed by Constellation demand.

The pool owns the Stage protocol and worker lifecycle. Workers request a bounded prefetch window and renew demand only after their handler returns. If a worker fails after receiving a batch, that batch is not retried; processing is at-most-once and recovery policy remains application-owned.

Types

Config

opaque </>

Worker pool configuration.

pub opaque type Config(event, worker_state)

ConfigError

</>

Invalid worker pool settings.

pub type ConfigError {
  InvalidSize(Int)
  InvalidPrefetch(Int)
  InvalidTimeout(Int)
}

Constructors

```
InvalidSize(Int)
```
```
InvalidPrefetch(Int)
```
```
InvalidTimeout(Int)
```

Event

</>

Lifecycle and processing events emitted asynchronously by a pool.

pub type Event {
  WorkerStarted(id: WorkerId, slot: Int)
  WorkerStopped(id: WorkerId, slot: Int)
  WorkerReplaced(
    previous: WorkerId,
    replacement: WorkerId,
    slot: Int,
  )
  BatchStarted(id: WorkerId, event_count: Int)
  BatchCompleted(id: WorkerId, event_count: Int)
  PoolStopping
  PoolStopped
}

Constructors

```
WorkerStarted(id: WorkerId, slot: Int)
```
```
WorkerStopped(id: WorkerId, slot: Int)
```

WorkerReplaced(
  previous: WorkerId,
  replacement: WorkerId,
  slot: Int,
)

BatchStarted(id: WorkerId, event_count: Int)

BatchCompleted(id: WorkerId, event_count: Int)

```
PoolStopping
```
```
PoolStopped
```

Pool

opaque </>

A running worker pool capability.

pub opaque type Pool(event, worker_state)

PoolError

</>

Runtime errors returned by pool operations.

pub type PoolError {
  PoolUnavailable
  PoolTimeout
  AlreadyStopping
  SourceManaged
  StageProtocol(stage_error.StageError)
}

Constructors

```
PoolUnavailable
```
```
PoolTimeout
```
```
AlreadyStopping
```
```
SourceManaged
```
```
StageProtocol(stage_error.StageError)
```

Snapshot

</>

Current observable pool state.

pub type Snapshot {
  Snapshot(
    workers: List(WorkerId),
    buffered_events: Int,
    stopping: Bool,
  )
}

Constructors

Snapshot(
  workers: List(WorkerId),
  buffered_events: Int,
  stopping: Bool,
)

StartError

</>

Errors that can prevent a pool from starting.

pub type StartError {
  InvalidConfig(ConfigError)
  ActorStart(actor.StartError)
}

Constructors

```
InvalidConfig(ConfigError)
```
```
ActorStart(actor.StartError)
```

WorkerId

opaque </>

Stable identity assigned to one worker incarnation.

pub opaque type WorkerId

Values

each

</>

pub fn each(
  size size: Int,
  prefetch prefetch: Int,
  initial_state initial_state: fn(Int) -> worker_state,
  handle_event handle_event: fn(worker_state, event) -> worker_state,
) -> Config(event, worker_state)

Builds a configuration whose handler processes one event at a time.

new

</>

pub fn new(
  size size: Int,
  prefetch prefetch: Int,
  initial_state initial_state: fn(Int) -> worker_state,
  handle_batch handle_batch: fn(worker_state, List(event)) -> worker_state,
) -> Config(event, worker_state)

Builds a batch-processing worker pool configuration.

push

</>

pub fn push(
  pool: Pool(event, worker_state),
  events: List(event),
) -> Result(Nil, PoolError)

Pushes events into a push-driven pool.

snapshot

</>

pub fn snapshot(
  pool: Pool(event, worker_state),
) -> Result(Snapshot, PoolError)

Returns worker identities and current buffered event count.

start

</>

pub fn start(
  config: Config(event, worker_state),
) -> Result(Pool(event, worker_state), StartError)

Starts a push-driven worker pool.

start_with_source

</>

pub fn start_with_source(
  config: Config(event, worker_state),
  notify_source: fn(source.Event) -> Nil,
) -> Result(
  #(Pool(event, worker_state), source.Source(event)),
  StartError,
)

Starts a pool connected to an asynchronous demand source.

Source callbacks run in a dedicated notifier process and never block the pool’s Stage process.

stop

</>

pub fn stop(
  pool: Pool(event, worker_state),
) -> Result(Nil, PoolError)

Gracefully stops workers after already delivered batches finish.

with_reporter

</>

pub fn with_reporter(
  config: Config(event, worker_state),
  reporter: fn(Event) -> Nil,
) -> Config(event, worker_state)

Installs a non-blocking lifecycle and telemetry reporter.

with_timeout

</>

pub fn with_timeout(
  config: Config(event, worker_state),
  milliseconds: Int,
) -> Config(event, worker_state)

Sets the timeout used by synchronous pool and source calls.

worker_id_to_int

</>

pub fn worker_id_to_int(id: WorkerId) -> Int

Returns the integer representation of a worker identity.