factos_pog - v2.0.0

Durable Effects

Durable effects carry a configured subscription’s work across the transaction boundary into an application worker. Start with How factos_pog works for the complete lifecycle and Durable Subscriptions for producing these effects atomically with events.

The durable envelope

An effect codec receives the application-owned typed effect and returns the non-generic opaque ProposedEffect representation: consumer, key, target, type, metadata, and payload. Workers receive that persisted representation as an OutboxMessage:

Factos Pog intentionally has no effect decoder. The application decodes its own type_, metadata, and opaque payload before executing the external work.

A worker that needs the full source fact can load it through the dispatcher that owns the event codec:

read_outbox_source_event(
  dispatcher:,
  connection:,
  message:,
) -> Result(factos.Recorded(event), factos_pog.Error(Nil))

read_outbox_source_event matches the stored position and event id, then decodes the row with the dispatcher’s event codec. It returns SourceEventNotFound(event_id:) when that identity no longer resolves and propagates normal decode and StoreError failures.

Configure a consumer

Factos Pog owns retry timing, attempt limits, age limits, and dead-letter transitions. Start from sensible defaults and build one validated RetryPolicy:

let assert Ok(policy) =
  factos_pog.new_retry_policy()
  |> factos_pog.build

The defaults are 12 maximum attempts, a 30-second initial delay, a 15-minute maximum delay, a 24-hour maximum delivery age, and 20 percent jitter. Override only the values that differ:

let assert Ok(fast_policy) =
  factos_pog.new_retry_policy()
  |> factos_pog.with_maximum_attempts(attempts: 3)
  |> factos_pog.with_initial_delay(milliseconds: 1000)
  |> factos_pog.with_maximum_delay(milliseconds: 60_000)
  |> factos_pog.with_maximum_age(milliseconds: 3_600_000)
  |> factos_pog.with_jitter(percent: 10)
  |> factos_pog.build

build validates the final combination and returns:

Bind that policy to the exact consumer and target emitted by the subscription codec:

let assert Ok(consumer) =
  factos_pog.outbox_consumer(
    name: "wallets",
    target: "ledgers",
    policy:,
  )

A blank identity after trimming returns InvalidConsumer or InvalidTarget. One OutboxConsumer now owns the policy for the ("wallets", "ledgers") delivery identity.

Lease work

Lease only pending, available messages for the configured consumer and target:

Leasing atomically increments each message’s one-based attempt, assigns an unguessable lock_token, and hides the row from competing workers until the lease expires. A non-positive limit returns Ok([]). A lease duration below one millisecond is clamped to one millisecond.

Every PostgreSQL operation in this lifecycle may return StoreError. Keep each leased message available until its settlement result is known; an unknown settlement result is not permission to execute the external effect again.

Execute and settle

Decode and execute each message in application code, classify only the outcome, and finalize it through settle_outbox:

let assert Ok(messages) =
  factos_pog.lease_outbox(
    connection,
    consumer:,
    limit: 10,
    lease_for_milliseconds: 30_000,
  )

messages
|> list.each(fn(message) {
  let outcome = case execute_effect(message) {
    Ok(Nil) -> factos_pog.delivered()
    Error(Temporary) ->
      factos_pog.retryable_failure(reason: "ledger_unavailable")
    Error(Permanent) ->
      factos_pog.permanent_failure(reason: "invalid_ledger_effect")
  }

  case
    factos_pog.settle_outbox(
      connection,
      consumer:,
      message:,
      outcome:,
    )
  {
    Ok(factos_pog.DeliveryAcknowledged) ->
      record_acknowledged(message)
    Ok(factos_pog.RetryScheduled(attempt:, delay_milliseconds:)) ->
      record_retry(message, attempt, delay_milliseconds)
    Ok(factos_pog.DeadLettered(attempt:, reason:)) ->
      record_dead_letter(message, attempt, reason)
    Ok(factos_pog.StaleLease) ->
      discard_lost_lease(message)
    Error(error) ->
      retain_unsettled_message(message, error)
  }
})

DeliveryAcknowledged means the row is delivered. RetryScheduled(attempt:, delay_milliseconds:) means the same effect will become available after Factos Pog’s computed delay. DeadLettered(attempt:, reason:) means automated delivery stopped. StaleLease means this worker lost ownership; do not execute or settle from that message again. The Error(error) branch must retain enough message context to reconcile the unknown settlement result rather than silently dropping it.

The three outcome constructors classify one already-executed leased message; they do not control whether a worker actor continues. Failure constructors keep the supplied reason unchanged. settle_outbox trims it and returns InvalidDeliveryReason before SQL when it is blank. Applications choose whether a failure is retryable or permanent, but cannot supply a retry delay or bypass the configured policy.

Retry and dead-letter policy

Retryable failures use capped exponential backoff. The initial delay doubles by attempt until it reaches the configured maximum, then deterministic jitter is applied from stable message identity and attempt. The result remains capped by the maximum delay. The same message and attempt therefore receive the same delay.

A retryable failure becomes DeadLettered when either maximum_attempts or maximum_age_milliseconds is reached. permanent_failure(reason:) dead-letters immediately, regardless of the remaining retry budget.

Stale leases and concurrency

Settlement is one PostgreSQL update guarded by message id, configured consumer, configured target, lock token, pending status, and unexpired lease. Expired, superseded, consumer-mismatched, target-mismatched, already-finalized, and concurrent losing settlements all return StaleLease without mutating the row. Exactly one current lease owner can win settlement.

Delivery is at least once. A worker can complete external work and then lose its lease before acknowledgement, allowing the effect to be delivered again. Effect handlers must therefore be idempotent or deduplicate with the stable (message.consumer, message.key) identity.

After StaleLease, never execute or settle from that leased message again. Let the current lease owner handle any redelivery.

Inspect dead letters

Operators can list dead-letter metadata with optional consumer and target filters:

list_dead_letters(
  connection,
  consumer: option.Option(String),
  target: option.Option(String),
  limit: Int,
) -> Result(List(factos_pog.DeadLetter), factos_pog.Error(Nil))

Pass option.None for an unfiltered dimension or option.Some(identity) to select it. A non-positive limit returns Ok([]). Results expose safe envelope identity, attempts, the terminal reason and timestamps, plus replay count and last replay time; the original opaque payload stays in the outbox row.

For example:

let assert Ok(dead_letters) =
  factos_pog.list_dead_letters(
    connection,
    consumer: option.Some("wallets"),
    target: option.Some("ledgers"),
    limit: 100,
  )

Replay a dead letter

Replay one row explicitly:

replay_dead_letter(
  connection,
  id: Int,
  attempts: factos_pog.ReplayAttempts,
) -> Result(factos_pog.ReplayResult, factos_pog.Error(Nil))

Choose PreserveAttempts to retain the previous delivery count or ResetAttempts to restart it at zero before the next one-based lease. A successful replay moves the row back to pending, preserves its original payload, increments replay_count, and sets last_replayed_at.

let assert Ok(factos_pog.Replayed) =
  factos_pog.replay_dead_letter(
    connection,
    id: dead_letter.id,
    attempts: factos_pog.PreserveAttempts,
  )

A row that is absent or no longer dead-lettered returns NotDeadLettered. Replaying does not execute the effect directly; it makes the preserved row available to its configured consumer again.

Operational rules

✨ Search Document