aarondb/changefeed

changefeed — bounded, resumable, ordered delivery over a durable source

The contract is deliberately at-least-once. Consumers persist their own idempotent effect and advance their durable projection checkpoint in the same storage transaction. A cursor identifies the last acknowledged offset.

Types

Bootstrap

pub type Bootstrap {
  Bootstrap(
    snapshot: durable_log.Snapshot,
    tail: List(durable_log.Entry),
    cursor: Int,
  )
}

Constructors

Bootstrap(
  snapshot: durable_log.Snapshot,
  tail: List(durable_log.Entry),
  cursor: Int,
)

Changefeed

pub type Changefeed {
  Changefeed(
    source: durable_log.DurableLog,
    cursor: Int,
    available_credit: Int,
  )
}

Constructors

Changefeed(
  source: durable_log.DurableLog,
  cursor: Int,
  available_credit: Int,
)

ChangefeedError

pub type ChangefeedError {
  InvalidCredit(Int)
  Source(durable_log.DurableLogError)
}

Constructors

```
InvalidCredit(Int)
```
```
Source(durable_log.DurableLogError)
```

Credit

pub type Credit =
  Int

Cursor

pub type Cursor =
  Int

Values

acknowledge

pub fn acknowledge(feed: Changefeed, offset: Int) -> Changefeed

Advance only to an offset that was committed by the source. Acknowledge is idempotent for older offsets and never makes an uncommitted event observable.

bootstrap

pub fn bootstrap(
  source: durable_log.DurableLog,
  through: Int,
) -> Result(Bootstrap, ChangefeedError)

Establish a consistent snapshot boundary before opening the tail feed. The returned cursor is the snapshot position; callers must resume from it.

credits

pub fn credits(feed: Changefeed) -> Int

cursor

pub fn cursor(feed: Changefeed) -> Int

grant

pub fn grant(
  feed: Changefeed,
  credit: Int,
) -> Result(Changefeed, ChangefeedError)

Add bounded delivery credits. Zero is valid and deliberately changes nothing.

pull

pub fn pull(
  feed: Changefeed,
) -> Result(
  #(Changefeed, List(durable_log.Entry)),
  ChangefeedError,
)

Pull at most the currently granted credit. Delivery does not advance the cursor: only acknowledge does, so an interrupted consumer can see a retry.

resume

pub fn resume(
  source: durable_log.DurableLog,
  cursor: Int,
  credit: Int,
) -> Result(Changefeed, ChangefeedError)

Open a resumable feed. cursor is the last durably acknowledged offset.