Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ name: CI

on:
pull_request:
branches: [main]

permissions:
contents: read
Expand Down
14 changes: 12 additions & 2 deletions docs/src/content/docs/durable/overview.md
Original file line number Diff line number Diff line change
Expand Up @@ -90,10 +90,20 @@ Two stdlib-only reference implementations ship in-tree:
retains the full record history, enabling time-travel below.
- **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance
subdirectories, each an append-only journal, an atomic checkpoint, an
idempotency ledger, and a dispatched-effect log. Each append flushes to stable
storage; each checkpoint uses write-temp-plus-rename for crash-safe atomicity.
idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file.
Each append flushes to stable storage; each checkpoint uses
write-temp-plus-rename for crash-safe atomicity.
Use it for durability across restarts without a database.

A write can carry a fence token (`WithFence`, `WithCheckpointFence`). A `Store`
keeps the highest token it has seen for an instance and refuses a lower one with
`ErrFenced`, writing nothing, so an owner that lost an instance is refused once the
new owner has written. `FileStore` fencing holds only within one open store.
`WithOwnerFence` sets the token a `Runner` carries on every `Append` and
`Checkpoint`. `Append` also rejects a different record at a recorded step with
`ErrStepConflict`. An out-of-tree `Store` reads the token with `AppendFence` and
`CheckpointFence`.

Persistent database backends (PostgreSQL, DynamoDB, and the like) implement
`Store` out of tree, so their drivers never burden this module's dependency or
vulnerability surface.
Expand Down
28 changes: 28 additions & 0 deletions durable/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,37 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.

## [Unreleased]

### Added

- **Fencing.** `WithFence` on `Store.Append` and `WithCheckpointFence` on
`Store.Checkpoint` carry a caller-issued token, and a Store refuses a token lower
than the highest it has seen for the instance with `ErrFenced`, writing nothing.
`WithOwnerFence` sets the token a Runner carries on every Append and Checkpoint;
`MarkDispatched` is not fenced. `AppendFence`,
`CheckpointFence` and `AppendIdempotencyKey` let a backend outside this package
read its options. `MemStore` and `FileStore` enforce fencing; `FileStore` persists the
highest token in a `fence` file and holds its fencing within one open store. A
Store first sees a token on that owner's first write, so `Recover` alone doesn't
fence the old owner: a stale write before then is accepted, and the new owner's
write at that step gets `ErrStepConflict`.

### Changed

- Requires Go 1.27.3.
- **A `Store` implementation has new work to do.** It must enforce fence tokens and
return `ErrStepConflict`, or `WithOwnerFence` protects nothing. Existing
implementations still compile, so nothing flags this.
- **`MemStore.Append` without `WithIdempotencyKey` refuses a Record JSON can't encode,**
as `FileStore` already did.

- **`Store.Append` rejects a different record for a recorded step.** It used to keep
the first and report success, which acknowledged a step the journal didn't hold.
It now returns `ErrStepConflict`, including for a step a checkpoint has compacted.
The same record is still an idempotent no-op, and an Append with
`WithIdempotencyKey` still dedups on the key alone. A `FileStore` ledger line from
before this change has no digest, so its step keeps the old no-op behavior. A
recovered instance that reuses a step number now fails with `ErrStepConflict`
instead of silently dropping the step.

### Fixed

Expand Down
14 changes: 12 additions & 2 deletions durable/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -100,10 +100,20 @@ Two stdlib-only reference implementations ship in-tree:
retains the full record history (enabling time-travel below).
- **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance
subdirectories, each an append-only journal, an atomic checkpoint, an
idempotency ledger, and a dispatched-effect log. Each append flushes to stable
storage; each checkpoint uses write-temp-plus-rename for crash-safe atomicity.
idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file.
Each append flushes to stable storage; each checkpoint uses
write-temp-plus-rename for crash-safe atomicity.
Use it for durability across restarts without a database.

A write can carry a fence token (`WithFence`, `WithCheckpointFence`). A `Store`
keeps the highest token it has seen for an instance and refuses a lower one with
`ErrFenced`, writing nothing, so an owner that lost an instance is refused once the
new owner has written. `FileStore` fencing holds only within one open store.
`WithOwnerFence` sets the token a `Runner` carries on every `Append` and
`Checkpoint`. `Append` also rejects a different record at a recorded step with
`ErrStepConflict`. An out-of-tree `Store` reads the token with `AppendFence` and
`CheckpointFence`.

Persistent database backends (PostgreSQL, DynamoDB, ...) implement `Store` out of
tree, so their drivers never burden this module's dependency or vulnerability
surface.
Expand Down
15 changes: 15 additions & 0 deletions durable/digest_internal_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
package durable

import "testing"

// TestRecordDigest_Golden pins the digest format. A retried Append after an upgrade is
// compared against digests already on disk, so this value must not change.
func TestRecordDigest_Golden(t *testing.T) {
got, err := recordDigest(Record{Step: 1, Event: []byte(`"go"`)})
if err != nil {
t.Fatalf("recordDigest: %v", err)
}
if want := "b56f3a37e67187ccd02a5b465d0c6533fb4ef737b6dd135641c3d34c38e491b7"; got != want {
t.Fatalf("recordDigest = %s, want %s", got, want)
}
}
3 changes: 2 additions & 1 deletion durable/doc.go
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,8 @@
//
// FileStore is the on-disk reference implementation: a directory of per-instance
// subdirectories, each holding an append-only journal, an atomic checkpoint, an
// idempotency ledger, and a dispatched-effect log. Each Append flushes to stable
// idempotency ledger, a dispatched-effect log and, once fenced, a fence file. Each
// Append flushes to stable
// storage; each Checkpoint uses write-temp+rename for crash-safe atomicity. Use
// FileStore when you need durability across restarts without a database.
//
Expand Down
9 changes: 5 additions & 4 deletions durable/effect_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -432,9 +432,10 @@ func TestRunner_EffectDispatch_NonDispatchStore_NoOp(t *testing.T) {
}
}

// timerMachine emits a kernel driver effect (a delayed `after` transition) on the
// same step a domain effect fires, so the dispatcher must route the driver effect
// to the Scheduler and only the domain effect to the handler.
// timerNotifyMachine emits a kernel driver effect (a delayed `after` transition) on
// the same step a domain effect fires, so the dispatcher must route the driver
// effect to the Scheduler and only the domain effect to the handler. The timer
// transition emits a second notify, which only a test that ticks observes.
func timerNotifyMachine() *state.Machine[string, string, *runCtx] {
return state.ForgeFor[*runCtx]("timer").
Action("notify", func(state.ActionCtx[*runCtx]) (state.Effect, error) {
Expand All @@ -445,7 +446,7 @@ func timerNotifyMachine() *state.Machine[string, string, *runCtx] {
State("done").Final().
Initial("idle").
Transition("idle").On("go").GoTo("waiting").Do("notify").
Transition("waiting").After(time.Second).GoTo("done").
Transition("waiting").After(time.Second).GoTo("done").Do("notify").
Quench()
}

Expand Down
35 changes: 30 additions & 5 deletions durable/errors.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,29 +2,54 @@ package durable

import "errors"

// Sentinel error text. The Err values below and ErrStepOutOfRange build on it.
const (
errorPrefix = "durable: "

errorInstanceNotFound = "instance not found"
errorStepOutOfOrder = "record step out of order"
errorStepConflict = "step already recorded with a different record"
errorFenced = "write fenced by a newer token"
errorCheckpointNotAdvancing = "checkpoint does not advance"
errorInstanceExists = "instance already exists"
errorEffectDispatch = "effect dispatch failed"
errorStepOutOfRange = "time-travel step out of range"
)

// Sentinel errors a Store reports. Callers match them with errors.Is.
var (
// ErrInstanceNotFound is reported by Store.Load for an instance that has
// never been written.
ErrInstanceNotFound = errors.New("crucible/durable: instance not found")
ErrInstanceNotFound = errors.New(errorPrefix + errorInstanceNotFound)

// ErrStepOutOfOrder is reported by Store.Append when a Record's Step does not
// strictly follow the instance's highest recorded Step (and is not an
// idempotent re-append of an already-recorded Step).
ErrStepOutOfOrder = errors.New("crucible/durable: record step out of order")
ErrStepOutOfOrder = errors.New(errorPrefix + errorStepOutOfOrder)

// ErrStepConflict is reported by Store.Append when a Step is already recorded
// with a different Record: the writer's view of the instance has diverged from
// the journal, usually because another writer got there first. The Record was
// not written.
ErrStepConflict = errors.New(errorPrefix + errorStepConflict)

// ErrFenced is reported by Store.Append and Store.Checkpoint for a write whose
// fence token is lower than the highest the Store has seen for the instance:
// a newer owner has taken the instance over. Nothing is written.
ErrFenced = errors.New(errorPrefix + errorFenced)

// ErrCheckpointNotAdvancing is reported by Store.Checkpoint when throughStep
// does not advance beyond the instance's current checkpoint.
ErrCheckpointNotAdvancing = errors.New("crucible/durable: checkpoint does not advance")
ErrCheckpointNotAdvancing = errors.New(errorPrefix + errorCheckpointNotAdvancing)

// ErrInstanceExists is reported by Runner.Start when the InstanceID is already
// present in the Store, so a fresh Start does not clobber a recorded instance.
ErrInstanceExists = errors.New("crucible/durable: instance already exists")
ErrInstanceExists = errors.New(errorPrefix + errorInstanceExists)

// ErrEffectDispatch wraps a failure returned by the caller-supplied effect
// handler (WithEffectHandler) while applying an emitted effect. The effect was
// persisted (write-ahead) but not marked dispatched, so a later recovery
// retries it; callers match this with errors.Is to distinguish a dispatch
// failure from a kernel transition or persistence error.
ErrEffectDispatch = errors.New("crucible/durable: effect dispatch failed")
ErrEffectDispatch = errors.New(errorPrefix + errorEffectDispatch)
)
Loading
Loading