From 85753ecd0993cf99115dc5babdd0d01c1ccebbb1 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 07:37:29 -0700 Subject: [PATCH 1/6] fix(durable): add write fencing and reject conflicting steps Store writes can carry a fence token, and a write below the highest token seen returns ErrFenced and writes nothing. Append now rejects a different record at a recorded step with ErrStepConflict instead of acknowledging a step the journal doesn't hold. Relates to #214. The failed-Append fix and the Handle lock are not included. Signed-off-by: Ray Karnes --- durable/CHANGELOG.md | 22 ++ durable/effect_test.go | 9 +- durable/errors.go | 11 + durable/fence_test.go | 446 ++++++++++++++++++++++++++++++++++++++ durable/filestore.go | 94 ++++++-- durable/filestore_test.go | 6 +- durable/memstore.go | 91 ++++++-- durable/memstore_test.go | 9 +- durable/options.go | 56 ++++- durable/runner.go | 6 +- durable/runner_options.go | 20 ++ durable/runner_test.go | 41 ++++ durable/store.go | 50 +++-- 13 files changed, 798 insertions(+), 63 deletions(-) create mode 100644 durable/fence_test.go diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index ed258652..5fcee0bb 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -7,9 +7,31 @@ 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` implement it; `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. +- **`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 diff --git a/durable/effect_test.go b/durable/effect_test.go index c2a33a3b..bb065564 100644 --- a/durable/effect_test.go +++ b/durable/effect_test.go @@ -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) { @@ -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() } diff --git a/durable/errors.go b/durable/errors.go index 1b3ec5df..110dab0a 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -13,6 +13,17 @@ var ( // idempotent re-append of an already-recorded Step). ErrStepOutOfOrder = errors.New("crucible/durable: record step out of order") + // 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, and nothing was + // written. + ErrStepConflict = errors.New("crucible/durable: step already recorded with a different record") + + // 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("crucible/durable: write fenced by a newer token") + // 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") diff --git a/durable/fence_test.go b/durable/fence_test.go new file mode 100644 index 00000000..9c523bbd --- /dev/null +++ b/durable/fence_test.go @@ -0,0 +1,446 @@ +package durable_test + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "testing" + + "github.com/stablekernel/crucible/durable" + "github.com/stablekernel/crucible/state" +) + +// storeBackends runs a contract test against each in-tree Store. +func storeBackends(t *testing.T, test func(t *testing.T, st durable.Store)) { + t.Helper() + t.Run("MemStore", func(t *testing.T) { test(t, durable.NewMemStore()) }) + t.Run("FileStore", func(t *testing.T) { test(t, newFileStore(t)) }) +} + +func tailLen(t *testing.T, st durable.Store, id durable.InstanceID) int { + t.Helper() + _, tail, err := st.Load(context.Background(), id) + if err != nil { + t.Fatalf("Load: %v", err) + } + return len(tail) +} + +func TestStore_Fence_LowerTokenWritesNothing(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + if _, err := st.Append(ctx, id, recOf(0, "a"), durable.WithFence(2)); err != nil { + t.Fatalf("Append(2): %v", err) + } + if _, err := st.Append(ctx, id, recOf(1, "b"), durable.WithFence(1)); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("Append(1): err = %v, want ErrFenced", err) + } + // The fence comes first: even a retry of a recorded step is refused. + if _, err := st.Append(ctx, id, recOf(0, "a"), durable.WithFence(1)); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("retry at token 1: err = %v, want ErrFenced", err) + } + if err := st.Checkpoint(ctx, id, []byte("snap"), 0, durable.WithCheckpointFence(1)); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("Checkpoint(1): err = %v, want ErrFenced", err) + } + snap, tail, err := st.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if snap != nil || len(tail) != 1 { + t.Fatalf("fenced writes changed the store: snapshot %q, %d records", snap, len(tail)) + } + }) +} + +func TestStore_Fence_EqualTokenPasses(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + seq, err := st.Append(ctx, id, recOf(0, "a"), durable.WithFence(2)) + if err != nil { + t.Fatalf("Append: %v", err) + } + again, err := st.Append(ctx, id, recOf(0, "a"), durable.WithFence(2)) + if err != nil { + t.Fatalf("retry: %v", err) + } + if again != seq { + t.Fatalf("retry seq = %d, want %d", again, seq) + } + if _, err = st.Append(ctx, id, recOf(1, "b"), durable.WithFence(2)); err != nil { + t.Fatalf("next step: %v", err) + } + if err = st.Checkpoint(ctx, id, []byte("snap"), 1, durable.WithCheckpointFence(2)); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + }) +} + +func TestStore_Fence_CheckpointRaisesIt(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + if err := st.Checkpoint(ctx, id, []byte("snap"), -1, durable.WithCheckpointFence(3)); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + if _, err := st.Append(ctx, id, recOf(0, "a"), durable.WithFence(2)); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("Append(2): err = %v, want ErrFenced", err) + } + }) +} + +func TestStore_Fence_NoTokenCountsAsZero(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + if _, err := st.Append(ctx, id, recOf(0, "a")); err != nil { + t.Fatalf("unfenced Append on an unfenced instance: %v", err) + } + if _, err := st.Append(ctx, id, recOf(1, "b"), durable.WithFence(1)); err != nil { + t.Fatalf("Append(1): %v", err) + } + if _, err := st.Append(ctx, id, recOf(2, "c")); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("unfenced Append on a fenced instance: err = %v, want ErrFenced", err) + } + }) +} + +func TestStore_Append_SameRecordIsANoOpAfterCompaction(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + seq, err := st.Append(ctx, id, recOf(0, "a")) + if err != nil { + t.Fatalf("Append: %v", err) + } + if err = st.Checkpoint(ctx, id, []byte("snap"), 0); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + again, err := st.Append(ctx, id, recOf(0, "a")) + if err != nil { + t.Fatalf("retry after compaction: %v", err) + } + if again != seq { + t.Fatalf("retry seq = %d, want %d", again, seq) + } + }) +} + +func TestStore_Append_DifferentRecordConflictsAfterCompaction(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + if _, err := st.Append(ctx, id, recOf(0, "a")); err != nil { + t.Fatalf("Append: %v", err) + } + if err := st.Checkpoint(ctx, id, []byte("snap"), 0); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + if _, err := st.Append(ctx, id, recOf(0, "b")); !errors.Is(err, durable.ErrStepConflict) { + t.Fatalf("err = %v, want ErrStepConflict", err) + } + }) +} + +func TestStore_Append_EmptyAndNilFieldsAreTheSameRecord(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + if _, err := st.Append(ctx, id, durable.Record{Step: 0}); err != nil { + t.Fatalf("Append: %v", err) + } + same := durable.Record{Step: 0, Event: []byte{}, Snapshot: []byte{}, Entries: []state.JournalEntry{}} + if _, err := st.Append(ctx, id, same); err != nil { + t.Fatalf("retry: %v", err) + } + }) +} + +func TestStore_Append_ExplicitKeyVouchesForTheRecord(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + if _, err := st.Append(ctx, id, recOf(0, "a"), durable.WithIdempotencyKey("k")); err != nil { + t.Fatalf("Append: %v", err) + } + if _, err := st.Append(ctx, id, recOf(0, "b"), durable.WithIdempotencyKey("k")); err != nil { + t.Fatalf("same key, different record: %v", err) + } + if n := tailLen(t, st, id); n != 1 { + t.Fatalf("records = %d, want 1", n) + } + }) +} + +func TestFileStore_FenceAndDigestSurviveReopen(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + const id = durable.InstanceID("inst") + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + if _, err = st.Append(ctx, id, recOf(0, "a"), durable.WithFence(5)); err != nil { + t.Fatalf("Append: %v", err) + } + if err = st.Checkpoint(ctx, id, []byte("snap"), 0, durable.WithCheckpointFence(5)); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, err = reopened.Append(ctx, id, recOf(1, "b"), durable.WithFence(4)); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("Append(4) after reopen: err = %v, want ErrFenced", err) + } + if _, err = reopened.Append(ctx, id, recOf(0, "other"), durable.WithFence(5)); !errors.Is(err, durable.ErrStepConflict) { + t.Fatalf("different record after reopen: err = %v, want ErrStepConflict", err) + } +} + +// A ledger written before digests existed can't be compared, so a re-append stays +// the no-op it always was. +func TestFileStore_LedgerWithoutDigest_StaysANoOp(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + const id = durable.InstanceID("inst") + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + if _, err = st.Append(ctx, id, recOf(0, "a")); err != nil { + t.Fatalf("Append: %v", err) + } + ledger := filepath.Join(filepath.Dir(durable.JournalPath(dir, id)), "applied.log") + if err = os.WriteFile(ledger, []byte(`{"key":"step:0","seq":1}`+"\n"), 0o600); err != nil { + t.Fatalf("rewrite ledger: %v", err) + } + + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, err = reopened.Append(ctx, id, recOf(0, "other")); err != nil { + t.Fatalf("re-append over an old ledger: %v", err) + } +} + +func TestFileStore_CorruptFence_FailsLoudly(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + const id = durable.InstanceID("inst") + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + if _, err = st.Append(ctx, id, recOf(0, "a"), durable.WithFence(5)); err != nil { + t.Fatalf("Append: %v", err) + } + fence := filepath.Join(filepath.Dir(durable.JournalPath(dir, id)), "fence") + if err = os.WriteFile(fence, []byte("not a number"), 0o600); err != nil { + t.Fatalf("corrupt fence: %v", err) + } + + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, _, err = reopened.Load(ctx, id); err == nil { + t.Fatal("Load over a corrupt fence succeeded") + } +} + +// A Start that crashed after raising the fence but before its baseline landed never +// wrote the instance, so the id must stay usable. +func TestFileStore_FenceOnlyDirectory_IsNotAnInstance(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + const id = durable.InstanceID("inst") + instDir := filepath.Dir(durable.JournalPath(dir, id)) + if err := os.MkdirAll(instDir, 0o700); err != nil { + t.Fatalf("mkdir: %v", err) + } + if err := os.WriteFile(filepath.Join(instDir, "fence"), []byte("5"), 0o600); err != nil { + t.Fatalf("write fence: %v", err) + } + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + if _, _, err = st.Load(ctx, id); !errors.Is(err, durable.ErrInstanceNotFound) { + t.Fatalf("Load: err = %v, want ErrInstanceNotFound", err) + } + if err = st.Checkpoint(ctx, id, []byte("snap"), -1, durable.WithCheckpointFence(4)); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("lower token: err = %v, want ErrFenced", err) + } + if err = st.Checkpoint(ctx, id, []byte("snap"), -1, durable.WithCheckpointFence(5)); err != nil { + t.Fatalf("retried Start's checkpoint: %v", err) + } +} + +func TestStore_Fence_ComesBeforeRecordEncoding(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + if _, err := st.Append(ctx, id, recOf(0, "a"), durable.WithFence(2)); err != nil { + t.Fatalf("Append: %v", err) + } + bad := durable.Record{Step: 1, Entries: []state.JournalEntry{{Payload: json.RawMessage("{")}}} + if _, err := st.Append(ctx, id, bad, durable.WithFence(1)); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("err = %v, want ErrFenced", err) + } + }) +} + +func TestFenceAccessors(t *testing.T) { + tests := []struct { + name string + token int64 + want int64 + }{ + {"a token", 7, 7}, + {"zero is ignored", 0, 0}, + {"negative is ignored", -1, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := durable.AppendFence(durable.WithFence(tt.token)); got != tt.want { + t.Fatalf("AppendFence = %d, want %d", got, tt.want) + } + if got := durable.CheckpointFence(durable.WithCheckpointFence(tt.token)); got != tt.want { + t.Fatalf("CheckpointFence = %d, want %d", got, tt.want) + } + }) + } + if got := durable.AppendFence(); got != 0 { + t.Fatalf("AppendFence() = %d, want 0", got) + } + if got := durable.AppendIdempotencyKey(durable.WithIdempotencyKey("k")); got != "k" { + t.Fatalf("AppendIdempotencyKey = %q, want k", got) + } + if got := durable.AppendIdempotencyKey(); got != "" { + t.Fatalf("AppendIdempotencyKey() = %q, want empty", got) + } +} + +func ownerFence(token int64) durable.Option[string, string, *runCtx] { + return durable.WithOwnerFence[string, string, *runCtx](token) +} + +// fenceSpy records the token on every write the Runner makes. +type fenceSpy struct { + durable.Store + appends []int64 + checkpoints []int64 +} + +func (s *fenceSpy) Append(ctx context.Context, id durable.InstanceID, rec durable.Record, opts ...durable.AppendOption) (int64, error) { + s.appends = append(s.appends, durable.AppendFence(opts...)) + return s.Store.Append(ctx, id, rec, opts...) +} + +func (s *fenceSpy) Checkpoint(ctx context.Context, id durable.InstanceID, snapshot []byte, throughStep int, opts ...durable.CheckpointOption) error { + s.checkpoints = append(s.checkpoints, durable.CheckpointFence(opts...)) + return s.Store.Checkpoint(ctx, id, snapshot, throughStep, opts...) +} + +func TestRunner_OwnerFence_CarriedOnEveryWrite(t *testing.T) { + ctx := context.Background() + m := linearMachine() + spy := &fenceSpy{Store: durable.NewMemStore()} + opts := []durable.Option[string, string, *runCtx]{ + ownerFence(4), + durable.WithCheckpointEvery[string, string, *runCtx](1), + } + h, err := durable.NewRunner(m, spy, opts...).Start(ctx, "inst", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "go"); err != nil { + t.Fatalf("Fire: %v", err) + } + if len(spy.appends) != 1 || len(spy.checkpoints) != 2 { + t.Fatalf("appends %v, checkpoints %v: want 1 and 2 writes", spy.appends, spy.checkpoints) + } + for _, got := range append(spy.appends, spy.checkpoints...) { + if got != 4 { + t.Fatalf("write carried token %d, want 4", got) + } + } +} + +func TestRecover_NewerOwnerFencesTheOlder(t *testing.T) { + ctx := context.Background() + m := branchingMachine() + store := durable.NewMemStore() + const id = durable.InstanceID("split-1") + if _, err := durable.NewRunner(m, store, ownerFence(1)).Start(ctx, id, &runCtx{}, state.WithInitialState("start")); err != nil { + t.Fatalf("Start: %v", err) + } + stale, err := durable.Recover(ctx, m, store, id, ownerFence(1)) + if err != nil { + t.Fatalf("Recover(stale): %v", err) + } + owner, err := durable.Recover(ctx, m, store, id, ownerFence(2)) + if err != nil { + t.Fatalf("Recover(owner): %v", err) + } + if _, err = owner.Fire(ctx, "goLeft"); err != nil { + t.Fatalf("owner Fire: %v", err) + } + if _, err = stale.Fire(ctx, "goRight"); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("stale Fire: err = %v, want ErrFenced", err) + } + recovered, err := durable.Recover(ctx, m, store, id, ownerFence(2)) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := recovered.Instance().Current(); got != "left" { + t.Fatalf("recovered at %q, want left", got) + } +} + +// Nothing is written at recovery, so a stale write can land before the new owner's +// first. The owner hears about it from the step check and recovers the journal as +// written, so no acknowledged step is lost. +func TestRecover_StaleWriteBeforeOwnersFirst_ConflictsThenRecovers(t *testing.T) { + ctx := context.Background() + m := branchingMachine() + store := durable.NewMemStore() + const id = durable.InstanceID("split-1") + if _, err := durable.NewRunner(m, store, ownerFence(1)).Start(ctx, id, &runCtx{}, state.WithInitialState("start")); err != nil { + t.Fatalf("Start: %v", err) + } + stale, err := durable.Recover(ctx, m, store, id, ownerFence(1)) + if err != nil { + t.Fatalf("Recover(stale): %v", err) + } + owner, err := durable.Recover(ctx, m, store, id, ownerFence(2)) + if err != nil { + t.Fatalf("Recover(owner): %v", err) + } + if _, err = stale.Fire(ctx, "goRight"); err != nil { + t.Fatalf("stale Fire: %v", err) + } + if _, err = owner.Fire(ctx, "goLeft"); !errors.Is(err, durable.ErrStepConflict) { + t.Fatalf("owner Fire: err = %v, want ErrStepConflict", err) + } + // The owner's refused write still raised the fence. + if _, err = stale.Fire(ctx, "finish"); !errors.Is(err, durable.ErrFenced) { + t.Fatalf("stale Fire after takeover: err = %v, want ErrFenced", err) + } + owner, err = durable.Recover(ctx, m, store, id, ownerFence(2)) + if err != nil { + t.Fatalf("Recover(owner again): %v", err) + } + if got := owner.Instance().Current(); got != "right" { + t.Fatalf("recovered at %q, want right", got) + } + if _, err = owner.Fire(ctx, "finish"); err != nil { + t.Fatalf("owner Fire after recovering: %v", err) + } +} diff --git a/durable/filestore.go b/durable/filestore.go index aab6d5cd..20c0c345 100644 --- a/durable/filestore.go +++ b/durable/filestore.go @@ -29,6 +29,7 @@ import ( // //checkpoint.meta latest checkpoint throughStep // //applied.log append-only idempotency ledger (key\tseq) // //dispatched.log append-only dispatched effect ids +// //fence highest fence token seen, once there is one // // The journal is the source of truth for the post-checkpoint Record tail; the // checkpoint files hold the compacted prefix's Snapshot; the applied ledger @@ -49,6 +50,12 @@ import ( // // All methods are safe for concurrent use; a per-instance mutex serializes // writes to one instance's files while distinct instances proceed independently. +// +// A directory belongs to one open FileStore at a time. Each keeps an in-memory index +// it never re-reads, so two over one directory, in one process or two, can overwrite +// each other: conflicting steps can reach the journal and the fence can go down. +// FileStore's fencing holds only between writers sharing one FileStore. A +// multi-process host needs a networked backend. type FileStore struct { cfg fileStoreConfig root string @@ -83,9 +90,11 @@ type fileInstance struct { maxStep int // seq is the monotonic per-instance append sequence. seq int64 - // applied maps an idempotency key to the seq assigned on first append. It is + // applied maps an idempotency key to what the first append recorded. It is // seeded from applied.log so a compacted step's key still dedups. - applied map[string]int64 + applied map[string]appliedRecord + // fence is the highest fence token seen for the instance. Zero means none. + fence int64 // dispatched is the set of effect ids already applied for the instance. dispatched map[string]struct{} } @@ -127,6 +136,7 @@ const ( metaFile = "checkpoint.meta" appliedFile = "applied.log" dispatchedFile = "dispatched.log" + fenceFile = "fence" ) // instance returns the in-memory index for id, loading it from disk on first @@ -147,7 +157,7 @@ func (s *FileStore) instance(id InstanceID) (*fileInstance, bool, error) { dir: instanceDir(s.root, id), throughStep: noCheckpoint, maxStep: noRecord, - applied: make(map[string]int64), + applied: make(map[string]appliedRecord), dispatched: make(map[string]struct{}), } s.instances[id] = inst @@ -182,6 +192,9 @@ func (fi *fileInstance) load() (bool, error) { return false, fmt.Errorf("crucible/durable: instance path %q is not a directory", fi.dir) } + if err := fi.loadFence(); err != nil { + return false, err + } if err := fi.loadCheckpoint(); err != nil { return false, err } @@ -194,7 +207,51 @@ func (fi *fileInstance) load() (bool, error) { if err := fi.loadDispatched(); err != nil { return false, err } - return true, nil + // A directory holding only a fence is a Start that crashed before its baseline + // checkpoint landed, so the instance was never written: Start may run again. + return fi.checkpoint != nil || len(fi.tail) > 0 || len(fi.applied) > 0 || len(fi.dispatched) > 0, nil +} + +// loadFence reads the highest fence token seen, if any. The file is written +// atomically, so an unreadable one is corruption and fails loudly: reading it as +// zero would readmit a fenced writer. +func (fi *fileInstance) loadFence() error { + b, err := os.ReadFile(filepath.Join(fi.dir, fenceFile)) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return fmt.Errorf("crucible/durable: reading fence for %q: %w", fi.dir, err) + } + if err := json.Unmarshal(b, &fi.fence); err != nil { + return fmt.Errorf("crucible/durable: decoding fence for %q: %w", fi.dir, err) + } + return nil +} + +// admitFence refuses a token below the highest seen and durably raises the highest +// to a greater one. The raise lands before the write it guards: a crash between the +// two leaves the newer owner fenced in with its write unmade, never a stale writer +// readmitted. The caller must hold fi.mu. +func (fi *fileInstance) admitFence(id InstanceID, token int64) error { + if token < fi.fence { + return fencedError(id, token, fi.fence) + } + if token == fi.fence { + return nil + } + if err := fi.ensureDir(); err != nil { + return err + } + b, err := json.Marshal(token) + if err != nil { + return fmt.Errorf("crucible/durable: marshaling fence for %q: %w", id, err) + } + if err := writeAtomic(filepath.Join(fi.dir, fenceFile), b); err != nil { + return fmt.Errorf("crucible/durable: writing fence for %q: %w", id, err) + } + fi.fence = token + return nil } // loadCheckpoint reads the latest checkpoint Snapshot bytes and its throughStep, @@ -236,7 +293,7 @@ func (fi *fileInstance) loadApplied() error { if jerr := json.Unmarshal(line, &entry); jerr != nil { return } - fi.applied[entry.Key] = entry.Seq + fi.applied[entry.Key] = appliedRecord{seq: entry.Seq, digest: entry.Digest} if entry.Seq > fi.seq { fi.seq = entry.Seq } @@ -310,18 +367,15 @@ const maxLineBytes = 16 * 1024 * 1024 // appliedEntry is one line of the idempotency ledger. type appliedEntry struct { - Key string `json:"key"` - Seq int64 `json:"seq"` + Key string `json:"key"` + Seq int64 `json:"seq"` + Digest string `json:"digest,omitempty"` } // Append implements Store. It is atomic and idempotent per (id, key), durably // writing a complete journal line before returning. func (s *FileStore) Append(_ context.Context, id InstanceID, rec Record, opts ...AppendOption) (int64, error) { cfg := resolveAppend(opts...) - key := cfg.idempotencyKey - if key == "" { - key = fmt.Sprintf("step:%d", rec.Step) - } inst, _, err := s.instance(id) if err != nil { @@ -330,8 +384,15 @@ func (s *FileStore) Append(_ context.Context, id InstanceID, rec Record, opts .. inst.mu.Lock() defer inst.mu.Unlock() + if err = inst.admitFence(id, cfg.fence); err != nil { + return 0, err + } + key, digest, err := appendKey(cfg, rec) + if err != nil { + return 0, err + } if existing, ok := inst.applied[key]; ok { - return existing, nil + return existing.check(id, rec.Step, digest) } if rec.Step <= inst.maxStep { return 0, fmt.Errorf("%w: step %d is not greater than recorded step %d for instance %q", @@ -351,7 +412,7 @@ func (s *FileStore) Append(_ context.Context, id InstanceID, rec Record, opts .. } seq := inst.seq + 1 - appliedLine, err := json.Marshal(appliedEntry{Key: key, Seq: seq}) + appliedLine, err := json.Marshal(appliedEntry{Key: key, Seq: seq, Digest: digest}) if err != nil { return 0, fmt.Errorf("crucible/durable: marshaling applied ledger for %q: %w", id, err) } @@ -362,7 +423,7 @@ func (s *FileStore) Append(_ context.Context, id InstanceID, rec Record, opts .. inst.tail = append(inst.tail, cloneRecord(rec)) inst.maxStep = rec.Step inst.seq = seq - inst.applied[key] = seq + inst.applied[key] = appliedRecord{seq: seq, digest: digest} return seq, nil } @@ -394,7 +455,7 @@ func (s *FileStore) Load(_ context.Context, id InstanceID) ([]byte, []Record, er // (temp+rename) and compacts the on-disk journal to only the post-checkpoint // tail. throughStep must advance beyond the current checkpoint. func (s *FileStore) Checkpoint(_ context.Context, id InstanceID, snapshot []byte, throughStep int, opts ...CheckpointOption) error { - _ = resolveCheckpoint(opts...) // no per-checkpoint option defined; on-disk tail is always compacted + cfg := resolveCheckpoint(opts...) inst, _, err := s.instance(id) if err != nil { @@ -403,6 +464,9 @@ func (s *FileStore) Checkpoint(_ context.Context, id InstanceID, snapshot []byte inst.mu.Lock() defer inst.mu.Unlock() + if err = inst.admitFence(id, cfg.fence); err != nil { + return err + } if throughStep <= inst.throughStep { return fmt.Errorf("%w: throughStep %d does not advance past checkpoint step %d for instance %q", ErrCheckpointNotAdvancing, throughStep, inst.throughStep, id) diff --git a/durable/filestore_test.go b/durable/filestore_test.go index 2012f49b..8d004713 100644 --- a/durable/filestore_test.go +++ b/durable/filestore_test.go @@ -104,7 +104,7 @@ func TestFileStore_Append_PerStepIdempotency(t *testing.T) { } } -func TestFileStore_Append_FirstWriterWins(t *testing.T) { +func TestFileStore_Append_DifferentRecordConflicts(t *testing.T) { ctx := context.Background() st := newFileStore(t) const id = durable.InstanceID("inst") @@ -113,8 +113,8 @@ func TestFileStore_Append_FirstWriterWins(t *testing.T) { if _, err := st.Append(ctx, id, first); err != nil { t.Fatalf("Append first: %v", err) } - if _, err := st.Append(ctx, id, recOf(0, "usurper")); err != nil { - t.Fatalf("Append second: %v", err) + if _, err := st.Append(ctx, id, recOf(0, "usurper")); !errors.Is(err, durable.ErrStepConflict) { + t.Fatalf("Append second: err = %v, want ErrStepConflict", err) } _, tail, err := st.Load(ctx, id) diff --git a/durable/memstore.go b/durable/memstore.go index cdc9b5a5..6fd28418 100644 --- a/durable/memstore.go +++ b/durable/memstore.go @@ -2,6 +2,9 @@ package durable import ( "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" "fmt" "sync" ) @@ -60,9 +63,11 @@ type memInstance struct { // non-idempotent Append. seq int64 // applied maps an append's idempotency key (the Record Step by default, or an - // explicit WithIdempotencyKey value) to the seq assigned on first append, so - // a re-append is a no-op returning that seq. - applied map[string]int64 + // explicit WithIdempotencyKey value) to what the first append recorded, so a + // re-append is a no-op returning that seq and a different Record is a conflict. + applied map[string]appliedRecord + // fence is the highest fence token seen for the instance. Zero means none. + fence int64 // dispatched is the set of effect ids the Runner has applied through its // effect handler. It backs the exactly-once dedup: an effect whose id is // present is skipped on (re)dispatch. It survives checkpoint compaction so a @@ -88,7 +93,7 @@ func (s *MemStore) instanceLocked(id InstanceID) *memInstance { inst = &memInstance{ throughStep: noCheckpoint, maxStep: noRecord, - applied: make(map[string]int64), + applied: make(map[string]appliedRecord), dispatched: make(map[string]struct{}), } if s.cfg.initialCapacity > 0 { @@ -103,20 +108,26 @@ func (s *MemStore) instanceLocked(id InstanceID) *memInstance { // is the Record Step by default or the WithIdempotencyKey value when supplied. func (s *MemStore) Append(_ context.Context, id InstanceID, rec Record, opts ...AppendOption) (int64, error) { cfg := resolveAppend(opts...) - key := cfg.idempotencyKey - if key == "" { - key = fmt.Sprintf("step:%d", rec.Step) - } s.mu.Lock() defer s.mu.Unlock() inst := s.instanceLocked(id) + if cfg.fence < inst.fence { + return 0, fencedError(id, cfg.fence, inst.fence) + } + inst.fence = cfg.fence + + key, digest, err := appendKey(cfg, rec) + if err != nil { + return 0, err + } + // Idempotency: a re-append of an already-applied key is a no-op returning the - // original sequence. The first writer of the key wins. + // original sequence, unless it carries a different Record. if existing, ok := inst.applied[key]; ok { - return existing, nil + return existing.check(id, rec.Step, digest) } // Strict per-instance Step ordering: a new Step must exceed every recorded @@ -132,7 +143,7 @@ func (s *MemStore) Append(_ context.Context, id InstanceID, rec Record, opts ... } inst.maxStep = rec.Step inst.seq++ - inst.applied[key] = inst.seq + inst.applied[key] = appliedRecord{seq: inst.seq, digest: digest} return inst.seq, nil } @@ -195,16 +206,21 @@ func (s *MemStore) History(_ context.Context, id InstanceID) ([]byte, []Record, // Checkpoint implements Store. It persists snapshot as the instance's checkpoint at // throughStep and compacts the tail through that step. throughStep must advance // beyond the current checkpoint. Time-travel retention is a store-level capability -// (NewMemStore with WithHistory), so the per-checkpoint CheckpointOptions do not -// change what Load returns here. +// (NewMemStore with WithHistory), so no CheckpointOption changes what Load returns +// here. func (s *MemStore) Checkpoint(_ context.Context, id InstanceID, snapshot []byte, throughStep int, opts ...CheckpointOption) error { - _ = resolveCheckpoint(opts...) // no per-checkpoint option defined; history is store-level (WithHistory) + cfg := resolveCheckpoint(opts...) s.mu.Lock() defer s.mu.Unlock() inst := s.instanceLocked(id) + if cfg.fence < inst.fence { + return fencedError(id, cfg.fence, inst.fence) + } + inst.fence = cfg.fence + if throughStep <= inst.throughStep { return fmt.Errorf("%w: throughStep %d does not advance past checkpoint step %d for instance %q", ErrCheckpointNotAdvancing, throughStep, inst.throughStep, id) @@ -275,6 +291,53 @@ func (s *MemStore) Dispatched(_ context.Context, id InstanceID) (map[string]bool return out, nil } +// appliedRecord is what a Store keeps for an applied idempotency key: the append +// sequence, and a digest of the Record so a different one is caught even after a +// Checkpoint compacts the Record away. The digest is empty when the Record can't be +// compared: an explicit WithIdempotencyKey, or a ledger line from before digests. +type appliedRecord struct { + seq int64 + digest string +} + +// check resolves a re-append of an applied key: the original sequence for the same +// Record, ErrStepConflict for a different one. +func (a appliedRecord) check(id InstanceID, step int, digest string) (int64, error) { + if a.digest != "" && digest != "" && a.digest != digest { + return 0, fmt.Errorf("%w: step %d for instance %q", ErrStepConflict, step, id) + } + return a.seq, nil +} + +// appendKey returns the idempotency key for an Append, and the digest to compare +// when that key is the Step. An explicit WithIdempotencyKey gets no digest. +func appendKey(cfg appendConfig, rec Record) (key, digest string, err error) { + if cfg.idempotencyKey != "" { + return cfg.idempotencyKey, "", nil + } + digest, err = recordDigest(rec) + if err != nil { + return "", "", err + } + return fmt.Sprintf("step:%d", rec.Step), digest, nil +} + +// recordDigest returns a stable digest of rec. It digests the clone because +// cloneRecord turns an empty field into nil, so the two compare equal. +func recordDigest(rec Record) (string, error) { + b, err := json.Marshal(cloneRecord(rec)) + if err != nil { + return "", fmt.Errorf("crucible/durable: digesting record step %d: %w", rec.Step, err) + } + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]), nil +} + +// fencedError reports a write whose token is below the highest seen. +func fencedError(id InstanceID, token, highest int64) error { + return fmt.Errorf("%w: token %d is below %d for instance %q", ErrFenced, token, highest, id) +} + // cloneBytes returns a copy of b, or nil for a nil input, so a returned snapshot is // insulated from later mutation of the caller's slice (and vice versa). func cloneBytes(b []byte) []byte { diff --git a/durable/memstore_test.go b/durable/memstore_test.go index 175b3327..a5165e6f 100644 --- a/durable/memstore_test.go +++ b/durable/memstore_test.go @@ -131,7 +131,7 @@ func TestMemStore_Append_PerStepIdempotency(t *testing.T) { } } -func TestMemStore_Append_FirstWriterWins(t *testing.T) { +func TestMemStore_Append_DifferentRecordConflicts(t *testing.T) { ctx := context.Background() st := durable.NewMemStore() const id = durable.InstanceID("inst") @@ -140,9 +140,10 @@ func TestMemStore_Append_FirstWriterWins(t *testing.T) { if _, err := st.Append(ctx, id, first); err != nil { t.Fatalf("Append first: %v", err) } - // A second append at the same step with different content must not overwrite. - if _, err := st.Append(ctx, id, recOf(0, "usurper")); err != nil { - t.Fatalf("Append second: %v", err) + // A second append at the same step with different content must not overwrite, + // and must not be reported as applied. + if _, err := st.Append(ctx, id, recOf(0, "usurper")); !errors.Is(err, durable.ErrStepConflict) { + t.Fatalf("Append second: err = %v, want ErrStepConflict", err) } _, tail, err := st.Load(ctx, id) diff --git a/durable/options.go b/durable/options.go index 112212fe..b24722ed 100644 --- a/durable/options.go +++ b/durable/options.go @@ -17,6 +17,8 @@ type appendConfig struct { // idempotencyKey, when set, overrides the Record Step as the dedup key for // the append. Empty means dedup on Step alone (the default contract). idempotencyKey string + // fence is the writer's fence token. Zero means no token was supplied. + fence int64 } // WithIdempotencyKey sets an explicit idempotency key for an Append, overriding @@ -31,6 +33,31 @@ func WithIdempotencyKey(key string) AppendOption { } } +// WithFence carries the writer's fence token on an Append: an increasing number, +// issued by the host with each ownership of the instance. A Store refuses a token +// lower than the highest it has seen for the instance with ErrFenced (see Store). A +// non-positive token is ignored. WithCheckpointFence is the Checkpoint counterpart; +// a Runner sets both from WithOwnerFence. +func WithFence(token int64) AppendOption { + return func(c *appendConfig) { + if token > 0 { + c.fence = token + } + } +} + +// AppendIdempotencyKey returns the explicit key opts carry, or "" when there is +// none. It is how a Store backend outside this package honors WithIdempotencyKey. +func AppendIdempotencyKey(opts ...AppendOption) string { + return resolveAppend(opts...).idempotencyKey +} + +// AppendFence returns the fence token opts carry, or zero when there is none. It +// is how a Store backend outside this package reads the token. +func AppendFence(opts ...AppendOption) int64 { + return resolveAppend(opts...).fence +} + func resolveAppend(opts ...AppendOption) appendConfig { var c appendConfig for _, opt := range opts { @@ -40,16 +67,33 @@ func resolveAppend(opts ...AppendOption) appendConfig { } // CheckpointOption configures a single Store.Checkpoint call. It is the additive -// extension point for per-checkpoint policy a backend may layer on. No option is -// defined yet: the seam reserves a stable signature so per-checkpoint policy can -// arrive additively without breaking the Store interface. Time-travel retention, +// extension point for per-checkpoint policy a backend may layer on, so that policy +// can arrive without breaking the Store interface. Time-travel retention, // which an earlier per-checkpoint flag covered, is now a store-level capability // (the HistoryStore seam, MemStore's WithHistory). type CheckpointOption func(*checkpointConfig) -// checkpointConfig holds resolved CheckpointOption state for one Checkpoint. It is -// presently empty; the type reserves a stable seam for additive options. -type checkpointConfig struct{} +// checkpointConfig holds resolved CheckpointOption state for one Checkpoint. +type checkpointConfig struct { + // fence is the writer's fence token. Zero means no token was supplied. + fence int64 +} + +// WithCheckpointFence carries the writer's fence token on a Checkpoint, with the +// same meaning as WithFence on an Append. A non-positive token is ignored. +func WithCheckpointFence(token int64) CheckpointOption { + return func(c *checkpointConfig) { + if token > 0 { + c.fence = token + } + } +} + +// CheckpointFence returns the fence token opts carry, or zero when there is none. +// It is how a Store backend outside this package reads the token. +func CheckpointFence(opts ...CheckpointOption) int64 { + return resolveCheckpoint(opts...).fence +} func resolveCheckpoint(opts ...CheckpointOption) checkpointConfig { var c checkpointConfig diff --git a/durable/runner.go b/durable/runner.go index 05769b81..67e7ab02 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -272,7 +272,7 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt if err != nil { return nil, fmt.Errorf("durable: marshaling start checkpoint envelope for %q: %w", id, err) } - if err := r.store.Checkpoint(ctx, id, baseEnv, baselineStep); err != nil { + if err := r.store.Checkpoint(ctx, id, baseEnv, baselineStep, WithCheckpointFence(r.cfg.fence)); err != nil { return nil, fmt.Errorf("durable: checkpointing start baseline for %q: %w", id, err) } @@ -474,7 +474,7 @@ func (h *Handle[S, E, C]) appendStep(ctx context.Context, step int, rec *Record) } rec.Snapshot = env } - if _, err := h.runner.store.Append(ctx, h.id, *rec); err != nil { + if _, err := h.runner.store.Append(ctx, h.id, *rec, WithFence(h.runner.cfg.fence)); err != nil { return fmt.Errorf("durable: recording step %d for %q: %w", step, h.id, err) } return nil @@ -488,7 +488,7 @@ func (h *Handle[S, E, C]) checkpointStep(ctx context.Context, step int, rec *Rec if !h.checkpointDue(step) { return nil } - if err := h.runner.store.Checkpoint(ctx, h.id, rec.Snapshot, step); err != nil { + if err := h.runner.store.Checkpoint(ctx, h.id, rec.Snapshot, step, WithCheckpointFence(h.runner.cfg.fence)); err != nil { return fmt.Errorf("durable: checkpointing step %d for %q: %w", step, h.id, err) } return nil diff --git a/durable/runner_options.go b/durable/runner_options.go index af264e69..601f7484 100644 --- a/durable/runner_options.go +++ b/durable/runner_options.go @@ -43,6 +43,9 @@ type runnerConfig[S comparable, E comparable, C any] struct { // whose transitions emit side-effecting domain effects supplies it with // WithEffectHandler. effectHandler EffectHandler + // fence is the owner's fence token, carried on every Append and Checkpoint. Zero + // (the default) carries none. + fence int64 } // WithCheckpointEvery sets the checkpoint policy: the Runner persists a full @@ -142,6 +145,23 @@ func WithEffectHandler[S comparable, E comparable, C any](h EffectHandler) Optio } } +// WithOwnerFence sets the fence token a Runner carries on every Append and +// Checkpoint, so a Handle that has lost the instance to a newer owner gets ErrFenced +// instead of committing. Pass the token the host's lease issued; a retry by the same +// owner keeps the same token. A non-positive token is ignored. +// +// The token applies to every instance the Runner drives, so a host with a token per +// instance passes it to Recover, or builds a Runner per instance. Nothing is written +// at recovery: a Store first sees the token on the owner's first write, and a stale +// write that lands before that one surfaces to the owner as ErrStepConflict. +func WithOwnerFence[S comparable, E comparable, C any](token int64) Option[S, E, C] { + return func(c *runnerConfig[S, E, C]) { + if token > 0 { + c.fence = token + } + } +} + // resolveRunner folds opts over the working-baseline defaults. func resolveRunner[S comparable, E comparable, C any](opts ...Option[S, E, C]) runnerConfig[S, E, C] { c := runnerConfig[S, E, C]{ diff --git a/durable/runner_test.go b/durable/runner_test.go index 931af3c5..c5ea4492 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -464,3 +464,44 @@ func TestRunner_Start_RejectsDuplicate(t *testing.T) { t.Fatalf("want ErrInstanceExists, got %v", err) } } + +// TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory proves +// two Handles on one instance (a stalled worker, and the worker that recovered the +// instance after it) cannot both have their Fires acknowledged while the journal +// records something else: recovery yields a history holding every Fire +// acknowledged. Takeover on Recover and optimistic concurrency on the step +// both satisfy it. +func TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory(t *testing.T) { + ctx := context.Background() + m := branchingMachine() + store := durable.NewMemStore() + id := durable.InstanceID("split-1") + if _, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("start")); err != nil { + t.Fatalf("Start: %v", err) + } + stale, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover(stale): %v", err) + } + owner, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover(owner): %v", err) + } + _, ownerErr := owner.Fire(ctx, "goLeft") + _, staleErr := stale.Fire(ctx, "goRight") + if staleErr == nil { + _, staleErr = stale.Fire(ctx, "finish") + } + + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + got := recovered.Instance().Current() + if ownerErr == nil && owner.Instance().Current() != got { + t.Fatalf("owner's Fire was acknowledged at %q, but recovery yields %q", owner.Instance().Current(), got) + } + if staleErr == nil && stale.Instance().Current() != got { + t.Fatalf("stale Handle's Fires were acknowledged at %q, but recovery yields %q", stale.Instance().Current(), got) + } +} diff --git a/durable/store.go b/durable/store.go index 04b68565..2aefe223 100644 --- a/durable/store.go +++ b/durable/store.go @@ -16,11 +16,13 @@ type InstanceID string // (Entries), the effects the step emitted (Effects, each carrying a stamped // EffectID for idempotent dispatch), and optionally a full marshaled Snapshot // checkpoint taken at this step. A Record is identified within its instance by -// its Step ordinal; appending the same Step twice is a no-op (see Store.Append). +// its Step ordinal; appending the same Record at the same Step twice is a no-op, +// and appending a different one is rejected (see Store.Append). type Record struct { // Step is the Fire ordinal this Record was produced at, indexing the // instance's recorded Traces. It is the per-instance idempotency key: a - // Store treats a second Append at the same Step as already-applied. + // Store treats a second Append of the same Record at the same Step as + // already-applied. Step int // Event is the structured, JSON form of the driving event the Fire at this @@ -80,14 +82,31 @@ type Record struct { // that same Step order. A Store preserves the relative order of Entries and // Effects within each Record verbatim. // -// Idempotency: Append is idempotent per (InstanceID, Step). Appending a Step -// that is already present is a no-op — the stored Record is retained unchanged -// and the original append sequence is returned — so an at-least-once caller -// (a crash-and-retry between persist and acknowledge) never double-applies a -// step. The first writer of a Step wins. The idempotency record for a Step -// survives a Checkpoint that compacts that Step away, so a delayed retry of an -// already-checkpointed Step is still recognized as a no-op rather than rejected -// as out of order. +// Idempotency: Append is idempotent per (InstanceID, Step). Appending the same +// Record at a Step already present is a no-op — the stored Record is retained +// unchanged and the original append sequence is returned — so an at-least-once +// caller (a crash-and-retry between persist and acknowledge) never double-applies +// a step. Appending a different Record at that Step is rejected with +// ErrStepConflict, because two writers disagree about the step and dropping one +// would acknowledge a step the journal doesn't hold. The idempotency record for a +// Step survives a Checkpoint that compacts that Step away, so a delayed retry of an +// already-checkpointed Step is still recognized rather than rejected as out of +// order, and a different Record there is still a conflict. Two Records are the +// same when every field matches, with a nil and an empty field counting as equal. +// An Append carrying WithIdempotencyKey (a backend reads it with +// AppendIdempotencyKey) dedups on that key alone, so the caller vouches for the +// Record: no comparison is made. +// +// Fencing: an Append or Checkpoint may carry a fence token (WithFence, +// WithCheckpointFence; a backend reads it with AppendFence and CheckpointFence). A +// Store keeps the highest token it has seen per instance, durably. A write whose +// token is lower returns ErrFenced and writes nothing; this check comes before +// every other. A token that passes raises the highest even if the write then fails +// another check, so a new owner's refused first write still fences the old owner. +// A write with no token counts as token zero, so once an instance has +// been fenced an unfenced write is refused. An equal token passes, so a retry by +// the same owner still works. The token is the fence alone: two writers on one +// token are told apart by ErrStepConflict. // // Atomicity: each Append and each Checkpoint is atomic. A concurrent Load never // observes a partially written Record or a checkpoint torn against its tail; it @@ -103,10 +122,12 @@ type Record struct { // All methods are safe for concurrent use by multiple goroutines. type Store interface { // Append persists rec for the instance and returns its monotonic per-instance - // append sequence. Appending a Step already recorded for the instance is a - // no-op that returns the existing sequence (idempotency). Records must be + // append sequence. Appending the same Record at a Step already recorded for the + // instance is a no-op that returns the existing sequence (idempotency); a + // different Record there is rejected with ErrStepConflict. Records must be // appended in increasing Step order; an out-of-order Step is rejected with - // ErrStepOutOfOrder. + // ErrStepOutOfOrder. A fence token below the highest seen is rejected with + // ErrFenced. Append(ctx context.Context, id InstanceID, rec Record, opts ...AppendOption) (seq int64, err error) // Load returns the instance's latest checkpoint Snapshot bytes (nil if the @@ -119,6 +140,7 @@ type Store interface { // compacts the journal/effect tail through that step, so a subsequent Load // returns this Snapshot plus only the Records appended after throughStep. A // Checkpoint that does not advance throughStep beyond the current checkpoint - // is rejected with ErrCheckpointNotAdvancing. + // is rejected with ErrCheckpointNotAdvancing. A fence token below the highest + // seen is rejected with ErrFenced. Checkpoint(ctx context.Context, id InstanceID, snapshot []byte, throughStep int, opts ...CheckpointOption) error } From be7bd0e478cbca8868ba498463437dc9e50b375c Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 07:56:50 -0700 Subject: [PATCH 2/6] docs(durable): describe fencing and the fence file in the Stores sections Signed-off-by: Ray Karnes --- docs/src/content/docs/durable/overview.md | 13 +++++++++++-- durable/README.md | 13 +++++++++++-- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 25947ae6..5345e7bc 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -90,10 +90,19 @@ 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 can't keep writing +to it. `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. diff --git a/durable/README.md b/durable/README.md index 45db9a1e..6bfc3276 100644 --- a/durable/README.md +++ b/durable/README.md @@ -100,10 +100,19 @@ 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 can't keep writing +to it. `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. From a92f50a65f2483fe1c5134470732573cd1a1ae49 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 09:37:19 -0700 Subject: [PATCH 3/6] fix(durable): tighten the fencing contract and refused first writes Folds in the review of the fencing change. The docs no longer overclaim what a fence guarantees. The Store contract now says the token must be as durable as the Records and the check atomic with the write, and states how explicit keys behave. MemStore treats an instance with a refused first write as never written, matching FileStore, and a directory with no fence keeps its old meaning. Adds a golden digest test and a racing-token test. Relates to #214. Signed-off-by: Ray Karnes --- docs/src/content/docs/durable/overview.md | 5 +- durable/CHANGELOG.md | 8 +- durable/README.md | 5 +- durable/digest_internal_test.go | 15 +++ durable/doc.go | 3 +- durable/errors.go | 4 +- durable/fence_test.go | 108 +++++++++++++++++++++- durable/filestore.go | 20 ++-- durable/memstore.go | 11 ++- durable/options.go | 12 ++- durable/runner_options.go | 5 +- durable/store.go | 18 +++- 12 files changed, 181 insertions(+), 33 deletions(-) create mode 100644 durable/digest_internal_test.go diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 5345e7bc..347366e0 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -97,8 +97,9 @@ Two stdlib-only reference implementations ship in-tree: 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 can't keep writing -to it. `WithOwnerFence` sets the token a `Runner` carries on every `Append` and +`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`. diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 5fcee0bb..74a844a0 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -15,7 +15,7 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. `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` implement it; `FileStore` persists the + 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 @@ -24,6 +24,12 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. ### 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. diff --git a/durable/README.md b/durable/README.md index 6bfc3276..faad6cd2 100644 --- a/durable/README.md +++ b/durable/README.md @@ -107,8 +107,9 @@ Two stdlib-only reference implementations ship in-tree: 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 can't keep writing -to it. `WithOwnerFence` sets the token a `Runner` carries on every `Append` and +`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`. diff --git a/durable/digest_internal_test.go b/durable/digest_internal_test.go new file mode 100644 index 00000000..132f7f05 --- /dev/null +++ b/durable/digest_internal_test.go @@ -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) + } +} diff --git a/durable/doc.go b/durable/doc.go index 731e2859..b4d5b6c9 100644 --- a/durable/doc.go +++ b/durable/doc.go @@ -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. // diff --git a/durable/errors.go b/durable/errors.go index 110dab0a..b7d9be94 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -15,8 +15,8 @@ var ( // 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, and nothing was - // written. + // the journal, usually because another writer got there first. The Record was + // not written. ErrStepConflict = errors.New("crucible/durable: step already recorded with a different record") // ErrFenced is reported by Store.Append and Store.Checkpoint for a write whose diff --git a/durable/fence_test.go b/durable/fence_test.go index 9c523bbd..d921bb8c 100644 --- a/durable/fence_test.go +++ b/durable/fence_test.go @@ -6,6 +6,8 @@ import ( "errors" "os" "path/filepath" + "sync" + "sync/atomic" "testing" "github.com/stablekernel/crucible/durable" @@ -404,7 +406,7 @@ func TestRecover_NewerOwnerFencesTheOlder(t *testing.T) { } } -// Nothing is written at recovery, so a stale write can land before the new owner's +// Recovery writes no fence token, so a stale write can land before the new owner's // first. The owner hears about it from the step check and recovers the journal as // written, so no acknowledged step is lost. func TestRecover_StaleWriteBeforeOwnersFirst_ConflictsThenRecovers(t *testing.T) { @@ -444,3 +446,107 @@ func TestRecover_StaleWriteBeforeOwnersFirst_ConflictsThenRecovers(t *testing.T) t.Fatalf("owner Fire after recovering: %v", err) } } + +// TestStore_Fence_RacingTokens races owners with rising tokens. Once a token is +// acknowledged no lower one may be, and tokens never fall along the journal. +func TestStore_Fence_RacingTokens(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + var highest, next, violations atomic.Int64 + var wg sync.WaitGroup + for token := int64(1); token <= 8; token++ { + wg.Go(func() { + for range 40 { + step := int(next.Add(1)) + before := highest.Load() + payload, err := json.Marshal(token) + if err != nil { + t.Errorf("Marshal: %v", err) + return + } + _, err = st.Append(ctx, id, durable.Record{Step: step, Snapshot: payload}, durable.WithFence(token)) + switch { + case err == nil: + if token < before { + violations.Add(1) + } + for { + seen := highest.Load() + if token <= seen || highest.CompareAndSwap(seen, token) { + break + } + } + case errors.Is(err, durable.ErrFenced), errors.Is(err, durable.ErrStepOutOfOrder), errors.Is(err, durable.ErrStepConflict): + default: + t.Errorf("Append: %v", err) + } + } + }) + } + wg.Wait() + if n := violations.Load(); n != 0 { + t.Fatalf("%d lower tokens acknowledged after a higher one", n) + } + _, tail, err := st.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + var last int64 + for _, rec := range tail { + var token int64 + if err := json.Unmarshal(rec.Snapshot, &token); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if token < last { + t.Fatalf("token fell from %d to %d at step %d", last, token, rec.Step) + } + last = token + } + }) +} + +// TestStore_RefusedFirstWrite_IsNotAnInstance proves a first write refused after the +// fence check leaves an instance Start can still create, on every backend. +func TestStore_RefusedFirstWrite_IsNotAnInstance(t *testing.T) { + storeBackends(t, func(t *testing.T, st durable.Store) { + ctx := context.Background() + const id = durable.InstanceID("inst") + bad := durable.Record{Step: 0, Entries: []state.JournalEntry{{Payload: json.RawMessage("{")}}} + if _, err := st.Append(ctx, id, bad, durable.WithFence(1)); err == nil { + t.Fatal("Append of an unencodable Record succeeded") + } + if _, _, err := st.Load(ctx, id); !errors.Is(err, durable.ErrInstanceNotFound) { + t.Fatalf("Load = %v, want ErrInstanceNotFound", err) + } + }) +} + +// TestFileStore_EmptyDirectoryWithNoFence_StillExists proves a directory with no fence +// keeps the meaning it had before fencing. +func TestFileStore_EmptyDirectoryWithNoFence_StillExists(t *testing.T) { + root := t.TempDir() + st, err := durable.NewFileStore(root) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + const id = durable.InstanceID("inst") + if err = st.MarkDispatched(context.Background(), id, "e"); err != nil { + t.Fatalf("MarkDispatched: %v", err) + } + entries, err := os.ReadDir(root) + if err != nil || len(entries) != 1 { + t.Fatalf("ReadDir = (%v, %v)", entries, err) + } + dir := filepath.Join(root, entries[0].Name()) + if err = os.Remove(filepath.Join(dir, "dispatched.log")); err != nil { + t.Fatalf("Remove: %v", err) + } + reopened, err := durable.NewFileStore(root) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, _, err = reopened.Load(context.Background(), id); err != nil { + t.Fatalf("Load of an existing directory = %v, want nil", err) + } +} diff --git a/durable/filestore.go b/durable/filestore.go index 20c0c345..f21acea3 100644 --- a/durable/filestore.go +++ b/durable/filestore.go @@ -27,7 +27,7 @@ import ( // //journal.log append-only, one JSON Record per line // //checkpoint.snap latest checkpoint Snapshot bytes // //checkpoint.meta latest checkpoint throughStep -// //applied.log append-only idempotency ledger (key\tseq) +// //applied.log append-only idempotency ledger, one JSON line per key (key, seq, digest) // //dispatched.log append-only dispatched effect ids // //fence highest fence token seen, once there is one // @@ -177,9 +177,9 @@ func (s *FileStore) instance(id InstanceID) (*fileInstance, bool, error) { return inst, inst.existed, nil } -// load reconstructs the instance index from its on-disk files. It returns -// whether the instance directory existed (an instance ever written). A torn -// trailing journal line is skipped without error. +// load reconstructs the instance index from its on-disk files. It returns whether +// the directory exists, unless it holds only a fence. A torn trailing journal line is +// skipped without error. func (fi *fileInstance) load() (bool, error) { info, err := os.Stat(fi.dir) if err != nil { @@ -207,9 +207,10 @@ func (fi *fileInstance) load() (bool, error) { if err := fi.loadDispatched(); err != nil { return false, err } - // A directory holding only a fence is a Start that crashed before its baseline - // checkpoint landed, so the instance was never written: Start may run again. - return fi.checkpoint != nil || len(fi.tail) > 0 || len(fi.applied) > 0 || len(fi.dispatched) > 0, nil + // A directory holding only a fence is a first write that never landed, such as a + // crashed Start, so the instance was never written and Start may run again. A + // directory with no fence keeps its old meaning, so it still counts as existing. + return fi.fence == 0 || fi.checkpoint != nil || len(fi.tail) > 0 || len(fi.applied) > 0 || len(fi.dispatched) > 0, nil } // loadFence reads the highest fence token seen, if any. The file is written @@ -598,10 +599,7 @@ func appendLine(path string, line []byte) (err error) { } }() - buf := make([]byte, 0, len(line)+1) - buf = append(buf, line...) - buf = append(buf, '\n') - if _, err = f.Write(buf); err != nil { + if _, err = f.Write(append(line[:len(line):len(line)], '\n')); err != nil { return err } return f.Sync() diff --git a/durable/memstore.go b/durable/memstore.go index 6fd28418..f6b26cb2 100644 --- a/durable/memstore.go +++ b/durable/memstore.go @@ -147,6 +147,13 @@ func (s *MemStore) Append(_ context.Context, id InstanceID, rec Record, opts ... return inst.seq, nil } +// written reports whether anything reached the instance. A write refused on its first +// use still creates the entry, to hold the fence, and that doesn't make an instance. It +// tests throughStep rather than checkpoint, because an empty snapshot is stored as nil. +func (i *memInstance) written() bool { + return i.throughStep != noCheckpoint || len(i.tail) > 0 || len(i.applied) > 0 || len(i.dispatched) > 0 +} + // Load implements Store. It returns the latest checkpoint plus the post- // checkpoint Record tail in Step order, or ErrInstanceNotFound. func (s *MemStore) Load(_ context.Context, id InstanceID) ([]byte, []Record, error) { @@ -154,7 +161,7 @@ func (s *MemStore) Load(_ context.Context, id InstanceID) ([]byte, []Record, err defer s.mu.RUnlock() inst, ok := s.instances[id] - if !ok { + if !ok || !inst.written() { return nil, nil, fmt.Errorf("%w: %q", ErrInstanceNotFound, id) } @@ -181,7 +188,7 @@ func (s *MemStore) History(_ context.Context, id InstanceID) ([]byte, []Record, defer s.mu.RUnlock() inst, ok := s.instances[id] - if !ok { + if !ok || !inst.written() { return nil, nil, fmt.Errorf("%w: %q", ErrInstanceNotFound, id) } diff --git a/durable/options.go b/durable/options.go index b24722ed..0fa6c994 100644 --- a/durable/options.go +++ b/durable/options.go @@ -23,8 +23,9 @@ type appendConfig struct { // WithIdempotencyKey sets an explicit idempotency key for an Append, overriding // the default of deduplicating on the Record's Step alone. Two Appends carrying -// the same key for the same instance collapse to one. An empty key is ignored -// (the Step-based default applies). +// the same key for the same instance collapse to one, with no comparison of the +// Records. An empty key is ignored (the Step-based default applies). Keys starting +// "step:" are reserved for the Store's own per-Step key. func WithIdempotencyKey(key string) AppendOption { return func(c *appendConfig) { if key != "" { @@ -36,8 +37,9 @@ func WithIdempotencyKey(key string) AppendOption { // WithFence carries the writer's fence token on an Append: an increasing number, // issued by the host with each ownership of the instance. A Store refuses a token // lower than the highest it has seen for the instance with ErrFenced (see Store). A -// non-positive token is ignored. WithCheckpointFence is the Checkpoint counterpart; -// a Runner sets both from WithOwnerFence. +// non-positive token is the same as no token, which a fenced instance refuses. +// WithCheckpointFence is the Checkpoint counterpart; a Runner sets both from +// WithOwnerFence. func WithFence(token int64) AppendOption { return func(c *appendConfig) { if token > 0 { @@ -80,7 +82,7 @@ type checkpointConfig struct { } // WithCheckpointFence carries the writer's fence token on a Checkpoint, with the -// same meaning as WithFence on an Append. A non-positive token is ignored. +// same meaning as WithFence on an Append. A non-positive token is the same as no token, which a fenced instance refuses. func WithCheckpointFence(token int64) CheckpointOption { return func(c *checkpointConfig) { if token > 0 { diff --git a/durable/runner_options.go b/durable/runner_options.go index 601f7484..7d5e4d60 100644 --- a/durable/runner_options.go +++ b/durable/runner_options.go @@ -148,10 +148,11 @@ func WithEffectHandler[S comparable, E comparable, C any](h EffectHandler) Optio // WithOwnerFence sets the fence token a Runner carries on every Append and // Checkpoint, so a Handle that has lost the instance to a newer owner gets ErrFenced // instead of committing. Pass the token the host's lease issued; a retry by the same -// owner keeps the same token. A non-positive token is ignored. +// owner keeps the same token. A non-positive token is the same as no token, which a +// fenced instance refuses. MarkDispatched is not fenced. // // The token applies to every instance the Runner drives, so a host with a token per -// instance passes it to Recover, or builds a Runner per instance. Nothing is written +// instance passes it to Recover, or builds a Runner per instance. No fence token is written // at recovery: a Store first sees the token on the owner's first write, and a stale // write that lands before that one surfaces to the owner as ErrStepConflict. func WithOwnerFence[S comparable, E comparable, C any](token int64) Option[S, E, C] { diff --git a/durable/store.go b/durable/store.go index 2aefe223..13a35a03 100644 --- a/durable/store.go +++ b/durable/store.go @@ -18,6 +18,11 @@ type InstanceID string // checkpoint taken at this step. A Record is identified within its instance by // its Step ordinal; appending the same Record at the same Step twice is a no-op, // and appending a different one is rejected (see Store.Append). +// +// The Stores compare Records by a digest of their JSON form, taken through cloneRecord. +// A new field needs `json:",omitempty"`, or every digest changes and a retried Append +// after an upgrade reads as a conflict. It must also be copied in cloneRecord, or the +// digest ignores it. The same applies to state.JournalEntry and state.EffectEnvelope. type Record struct { // Step is the Fire ordinal this Record was produced at, indexing the // instance's recorded Traces. It is the per-instance idempotency key: a @@ -95,18 +100,23 @@ type Record struct { // same when every field matches, with a nil and an empty field counting as equal. // An Append carrying WithIdempotencyKey (a backend reads it with // AppendIdempotencyKey) dedups on that key alone, so the caller vouches for the -// Record: no comparison is made. +// Record: no comparison is made. Keys starting "step:" are reserved for the Store's +// own per-Step key. A Step recorded under an explicit key is not recognized by a +// later Append without that key, or with a different one: that Append returns +// ErrStepOutOfOrder. // // Fencing: an Append or Checkpoint may carry a fence token (WithFence, // WithCheckpointFence; a backend reads it with AppendFence and CheckpointFence). A -// Store keeps the highest token it has seen per instance, durably. A write whose +// Store keeps the highest token it has seen per instance, as durably as its Records. +// The token check must be serialized with the write it guards, against the Store's +// own state. A write whose // token is lower returns ErrFenced and writes nothing; this check comes before // every other. A token that passes raises the highest even if the write then fails // another check, so a new owner's refused first write still fences the old owner. // A write with no token counts as token zero, so once an instance has // been fenced an unfenced write is refused. An equal token passes, so a retry by -// the same owner still works. The token is the fence alone: two writers on one -// token are told apart by ErrStepConflict. +// the same owner still works. Tokens alone can't separate two writers on one token; +// ErrStepConflict does. // // Atomicity: each Append and each Checkpoint is atomic. A concurrent Load never // observes a partially written Record or a checkpoint torn against its tail; it From d2ca03de5deba8990064707b408de6b452560480 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 15:17:18 -0700 Subject: [PATCH 4/6] chore(ci): run CI only on pull requests into main A stacked PR re-ran the full suite on every push, though only the PR based on main can merge. Each PR runs CI once the stack retargets it to main, which saves GitHub Actions minutes. Signed-off-by: Ray Karnes --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 927a6923..baa938c1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,6 +2,7 @@ name: CI on: pull_request: + branches: [main] permissions: contents: read From 45f273f5032b11de8dfe82c8e3163dc05587ad79 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Fri, 6 Nov 2026 07:10:03 -0700 Subject: [PATCH 5/6] refactor(durable): move sentinel text to constants and use the durable prefix Sentinel error text now lives in private constants built on errorPrefix. Every error in the package now starts with "durable: " rather than "crucible/durable: ". Exported names are unchanged, and errors.Is matching is unaffected. Signed-off-by: Ray Karnes --- durable/errors.go | 28 +++++++++++++----- durable/filestore.go | 46 +++++++++++++++--------------- durable/filestore_dirsync_other.go | 4 +-- durable/memstore.go | 2 +- durable/timetravel.go | 2 +- 5 files changed, 48 insertions(+), 34 deletions(-) diff --git a/durable/errors.go b/durable/errors.go index b7d9be94..7b9d0752 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -2,40 +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("crucible/durable: step already recorded with a different record") + 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("crucible/durable: write fenced by a newer token") + 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) ) diff --git a/durable/filestore.go b/durable/filestore.go index f21acea3..b7a9462d 100644 --- a/durable/filestore.go +++ b/durable/filestore.go @@ -106,10 +106,10 @@ type fileInstance struct { // options; with none supplied it returns a ready store. func NewFileStore(dir string, opts ...FileStoreOption) (*FileStore, error) { if dir == "" { - return nil, errors.New("crucible/durable: FileStore dir must not be empty") + return nil, errors.New("durable: FileStore dir must not be empty") } if err := os.MkdirAll(dir, 0o700); err != nil { - return nil, fmt.Errorf("crucible/durable: creating FileStore root %q: %w", dir, err) + return nil, fmt.Errorf("durable: creating FileStore root %q: %w", dir, err) } return &FileStore{ cfg: resolveFileStore(opts...), @@ -186,10 +186,10 @@ func (fi *fileInstance) load() (bool, error) { if errors.Is(err, os.ErrNotExist) { return false, nil } - return false, fmt.Errorf("crucible/durable: stat instance dir %q: %w", fi.dir, err) + return false, fmt.Errorf("durable: stat instance dir %q: %w", fi.dir, err) } if !info.IsDir() { - return false, fmt.Errorf("crucible/durable: instance path %q is not a directory", fi.dir) + return false, fmt.Errorf("durable: instance path %q is not a directory", fi.dir) } if err := fi.loadFence(); err != nil { @@ -222,10 +222,10 @@ func (fi *fileInstance) loadFence() error { if errors.Is(err, os.ErrNotExist) { return nil } - return fmt.Errorf("crucible/durable: reading fence for %q: %w", fi.dir, err) + return fmt.Errorf("durable: reading fence for %q: %w", fi.dir, err) } if err := json.Unmarshal(b, &fi.fence); err != nil { - return fmt.Errorf("crucible/durable: decoding fence for %q: %w", fi.dir, err) + return fmt.Errorf("durable: decoding fence for %q: %w", fi.dir, err) } return nil } @@ -246,10 +246,10 @@ func (fi *fileInstance) admitFence(id InstanceID, token int64) error { } b, err := json.Marshal(token) if err != nil { - return fmt.Errorf("crucible/durable: marshaling fence for %q: %w", id, err) + return fmt.Errorf("durable: marshaling fence for %q: %w", id, err) } if err := writeAtomic(filepath.Join(fi.dir, fenceFile), b); err != nil { - return fmt.Errorf("crucible/durable: writing fence for %q: %w", id, err) + return fmt.Errorf("durable: writing fence for %q: %w", id, err) } fi.fence = token return nil @@ -263,7 +263,7 @@ func (fi *fileInstance) loadCheckpoint() error { if errors.Is(err, os.ErrNotExist) { return nil } - return fmt.Errorf("crucible/durable: reading checkpoint for %q: %w", fi.dir, err) + return fmt.Errorf("durable: reading checkpoint for %q: %w", fi.dir, err) } metaBytes, err := os.ReadFile(filepath.Join(fi.dir, metaFile)) if err != nil { @@ -271,11 +271,11 @@ func (fi *fileInstance) loadCheckpoint() error { // A snapshot without intact meta is treated as no checkpoint. return nil } - return fmt.Errorf("crucible/durable: reading checkpoint meta for %q: %w", fi.dir, err) + return fmt.Errorf("durable: reading checkpoint meta for %q: %w", fi.dir, err) } var through int if err := json.Unmarshal(metaBytes, &through); err != nil { - return fmt.Errorf("crucible/durable: decoding checkpoint meta for %q: %w", fi.dir, err) + return fmt.Errorf("durable: decoding checkpoint meta for %q: %w", fi.dir, err) } fi.checkpoint = snap fi.throughStep = through @@ -342,7 +342,7 @@ func scanLines(path, label string, fn func(line []byte)) (err error) { if errors.Is(oerr, os.ErrNotExist) { return nil } - return fmt.Errorf("crucible/durable: opening %s %q: %w", label, path, oerr) + return fmt.Errorf("durable: opening %s %q: %w", label, path, oerr) } defer func() { if cerr := f.Close(); cerr != nil && err == nil { @@ -356,7 +356,7 @@ func scanLines(path, label string, fn func(line []byte)) (err error) { fn(sc.Bytes()) } if serr := sc.Err(); serr != nil { - return fmt.Errorf("crucible/durable: scanning %s %q: %w", label, path, serr) + return fmt.Errorf("durable: scanning %s %q: %w", label, path, serr) } return nil } @@ -406,19 +406,19 @@ func (s *FileStore) Append(_ context.Context, id InstanceID, rec Record, opts .. inst.existed = true line, err := json.Marshal(cloneRecord(rec)) if err != nil { - return 0, fmt.Errorf("crucible/durable: marshaling record step %d for %q: %w", rec.Step, id, err) + return 0, fmt.Errorf("durable: marshaling record step %d for %q: %w", rec.Step, id, err) } if err = appendLine(filepath.Join(inst.dir, journalFile), line); err != nil { - return 0, fmt.Errorf("crucible/durable: appending journal for %q: %w", id, err) + return 0, fmt.Errorf("durable: appending journal for %q: %w", id, err) } seq := inst.seq + 1 appliedLine, err := json.Marshal(appliedEntry{Key: key, Seq: seq, Digest: digest}) if err != nil { - return 0, fmt.Errorf("crucible/durable: marshaling applied ledger for %q: %w", id, err) + return 0, fmt.Errorf("durable: marshaling applied ledger for %q: %w", id, err) } if err = appendLine(filepath.Join(inst.dir, appliedFile), appliedLine); err != nil { - return 0, fmt.Errorf("crucible/durable: appending applied ledger for %q: %w", id, err) + return 0, fmt.Errorf("durable: appending applied ledger for %q: %w", id, err) } inst.tail = append(inst.tail, cloneRecord(rec)) @@ -480,13 +480,13 @@ func (s *FileStore) Checkpoint(_ context.Context, id InstanceID, snapshot []byte // Persist the snapshot and its throughStep atomically (temp+rename each). meta, err := json.Marshal(throughStep) if err != nil { - return fmt.Errorf("crucible/durable: marshaling checkpoint meta for %q: %w", id, err) + return fmt.Errorf("durable: marshaling checkpoint meta for %q: %w", id, err) } if err = writeAtomic(filepath.Join(inst.dir, checkpointFile), snapshot); err != nil { - return fmt.Errorf("crucible/durable: writing checkpoint for %q: %w", id, err) + return fmt.Errorf("durable: writing checkpoint for %q: %w", id, err) } if err = writeAtomic(filepath.Join(inst.dir, metaFile), meta); err != nil { - return fmt.Errorf("crucible/durable: writing checkpoint meta for %q: %w", id, err) + return fmt.Errorf("durable: writing checkpoint meta for %q: %w", id, err) } // Compact the on-disk journal: keep only Records strictly after throughStep. @@ -498,7 +498,7 @@ func (s *FileStore) Checkpoint(_ context.Context, id InstanceID, snapshot []byte kept = append(kept, rec) } if err = inst.rewriteJournal(kept); err != nil { - return fmt.Errorf("crucible/durable: compacting journal for %q: %w", id, err) + return fmt.Errorf("durable: compacting journal for %q: %w", id, err) } inst.checkpoint = append([]byte(nil), snapshot...) @@ -533,7 +533,7 @@ func (s *FileStore) MarkDispatched(_ context.Context, id InstanceID, effectIDs . continue } if err = appendLine(filepath.Join(inst.dir, dispatchedFile), []byte(eid)); err != nil { - return fmt.Errorf("crucible/durable: appending dispatched id for %q: %w", id, err) + return fmt.Errorf("durable: appending dispatched id for %q: %w", id, err) } inst.dispatched[eid] = struct{}{} } @@ -561,7 +561,7 @@ func (s *FileStore) Dispatched(_ context.Context, id InstanceID) (map[string]boo // ensureDir creates the instance directory on first write. func (fi *fileInstance) ensureDir() error { if err := os.MkdirAll(fi.dir, 0o700); err != nil { - return fmt.Errorf("crucible/durable: creating instance dir %q: %w", fi.dir, err) + return fmt.Errorf("durable: creating instance dir %q: %w", fi.dir, err) } return nil } diff --git a/durable/filestore_dirsync_other.go b/durable/filestore_dirsync_other.go index cd689b87..dc5705a6 100644 --- a/durable/filestore_dirsync_other.go +++ b/durable/filestore_dirsync_other.go @@ -18,11 +18,11 @@ import ( func syncDir(dir string) error { d, err := os.Open(dir) if err != nil { - return fmt.Errorf("crucible/durable: opening dir %q to sync: %w", dir, err) + return fmt.Errorf("durable: opening dir %q to sync: %w", dir, err) } defer func() { _ = d.Close() }() if err := d.Sync(); err != nil { - return fmt.Errorf("crucible/durable: syncing dir %q: %w", dir, err) + return fmt.Errorf("durable: syncing dir %q: %w", dir, err) } return nil } diff --git a/durable/memstore.go b/durable/memstore.go index f6b26cb2..289245fa 100644 --- a/durable/memstore.go +++ b/durable/memstore.go @@ -334,7 +334,7 @@ func appendKey(cfg appendConfig, rec Record) (key, digest string, err error) { func recordDigest(rec Record) (string, error) { b, err := json.Marshal(cloneRecord(rec)) if err != nil { - return "", fmt.Errorf("crucible/durable: digesting record step %d: %w", rec.Step, err) + return "", fmt.Errorf("durable: digesting record step %d: %w", rec.Step, err) } sum := sha256.Sum256(b) return hex.EncodeToString(sum[:]), nil diff --git a/durable/timetravel.go b/durable/timetravel.go index 7b876a56..9bec1218 100644 --- a/durable/timetravel.go +++ b/durable/timetravel.go @@ -43,7 +43,7 @@ import ( // ErrStepOutOfRange is reported by StateAt when the requested step is below the // start baseline or beyond the instance's last recorded step. Callers match it with // errors.Is. -var ErrStepOutOfRange = errors.New("crucible/durable: time-travel step out of range") +var ErrStepOutOfRange = errors.New(errorPrefix + errorStepOutOfRange) // BaselineStep is the step that addresses an instance's start baseline — its state // immediately after Start, before any event fired. StateAt(ctx, m, st, id, From d69dca55cd7ef96068705d4dcbc28335617ec8d7 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Fri, 6 Nov 2026 07:26:58 -0700 Subject: [PATCH 6/6] chore(deps): bump klauspost/compress to v1.20.2 for GO-2026-6865 The advisory, published after this branch last passed, covers a panic on two-block frames and fails govulncheck on every PR into main. Four modules reach the vulnerable code; all 19 that carry the dependency move together. Signed-off-by: Ray Karnes --- examples/sourcedrive/go.mod | 2 +- examples/sourcedrive/go.sum | 4 ++-- sink/cloudwatch/go.mod | 2 +- sink/cloudwatch/go.sum | 4 ++-- sink/dynamo/go.mod | 2 +- sink/dynamo/go.sum | 4 ++-- sink/eventbridge/go.mod | 2 +- sink/eventbridge/go.sum | 4 ++-- sink/firehose/go.mod | 2 +- sink/firehose/go.sum | 4 ++-- sink/gcppubsub/go.mod | 2 +- sink/gcppubsub/go.sum | 4 ++-- sink/kafka/go.mod | 2 +- sink/kafka/go.sum | 4 ++-- sink/kinesis/go.mod | 2 +- sink/kinesis/go.sum | 4 ++-- sink/nats/go.mod | 2 +- sink/nats/go.sum | 4 ++-- sink/redis/go.mod | 2 +- sink/redis/go.sum | 4 ++-- sink/s3/go.mod | 2 +- sink/s3/go.sum | 4 ++-- sink/sns/go.mod | 2 +- sink/sns/go.sum | 4 ++-- sink/sqs/go.mod | 2 +- sink/sqs/go.sum | 4 ++-- sink/timestream/go.mod | 2 +- sink/timestream/go.sum | 4 ++-- source/jetstream/go.mod | 2 +- source/jetstream/go.sum | 4 ++-- source/kafka/go.mod | 2 +- source/kafka/go.sum | 2 ++ source/kafka/integration/go.mod | 2 +- source/kafka/integration/go.sum | 4 ++-- source/redis/go.mod | 2 +- source/redis/go.sum | 4 ++-- telemetry/datadog/go.mod | 2 +- telemetry/datadog/go.sum | 4 ++-- 38 files changed, 57 insertions(+), 55 deletions(-) diff --git a/examples/sourcedrive/go.mod b/examples/sourcedrive/go.mod index 1b541eb9..5956e246 100644 --- a/examples/sourcedrive/go.mod +++ b/examples/sourcedrive/go.mod @@ -34,7 +34,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.6 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/examples/sourcedrive/go.sum b/examples/sourcedrive/go.sum index c32579a4..d534d1a1 100644 --- a/examples/sourcedrive/go.sum +++ b/examples/sourcedrive/go.sum @@ -46,8 +46,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= -github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/cloudwatch/go.mod b/sink/cloudwatch/go.mod index 139e74c8..4d78060f 100644 --- a/sink/cloudwatch/go.mod +++ b/sink/cloudwatch/go.mod @@ -49,7 +49,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/cloudwatch/go.sum b/sink/cloudwatch/go.sum index 24c329bc..90abc604 100644 --- a/sink/cloudwatch/go.sum +++ b/sink/cloudwatch/go.sum @@ -79,8 +79,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/dynamo/go.mod b/sink/dynamo/go.mod index cf77bbfa..585d19ca 100644 --- a/sink/dynamo/go.mod +++ b/sink/dynamo/go.mod @@ -45,7 +45,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/dynamo/go.sum b/sink/dynamo/go.sum index 1d28fac5..bc24a751 100644 --- a/sink/dynamo/go.sum +++ b/sink/dynamo/go.sum @@ -79,8 +79,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/eventbridge/go.mod b/sink/eventbridge/go.mod index ba3fac00..45680836 100644 --- a/sink/eventbridge/go.mod +++ b/sink/eventbridge/go.mod @@ -49,7 +49,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/eventbridge/go.sum b/sink/eventbridge/go.sum index b74d9386..e6c5c58a 100644 --- a/sink/eventbridge/go.sum +++ b/sink/eventbridge/go.sum @@ -79,8 +79,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/firehose/go.mod b/sink/firehose/go.mod index 34ddc4e5..533f11b5 100644 --- a/sink/firehose/go.mod +++ b/sink/firehose/go.mod @@ -48,7 +48,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/firehose/go.sum b/sink/firehose/go.sum index 3fce78a8..8c3bb4d3 100644 --- a/sink/firehose/go.sum +++ b/sink/firehose/go.sum @@ -85,8 +85,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/gcppubsub/go.mod b/sink/gcppubsub/go.mod index 93218bc7..7ab976b9 100644 --- a/sink/gcppubsub/go.mod +++ b/sink/gcppubsub/go.mod @@ -38,7 +38,7 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect github.com/googleapis/gax-go/v2 v2.21.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/gcppubsub/go.sum b/sink/gcppubsub/go.sum index e9ffcc91..8f58be6d 100644 --- a/sink/gcppubsub/go.sum +++ b/sink/gcppubsub/go.sum @@ -106,8 +106,8 @@ github.com/googleapis/enterprise-certificate-proxy v0.3.14 h1:yh8ncqsbUY4shRD5dA github.com/googleapis/enterprise-certificate-proxy v0.3.14/go.mod h1:vqVt9yG9480NtzREnTlmGSBmFrA+bzb0yl0TxoBQXOg= github.com/googleapis/gax-go/v2 v2.21.0 h1:h45NjjzEO3faG9Lg/cFrBh2PgegVVgzqKzuZl/wMbiI= github.com/googleapis/gax-go/v2 v2.21.0/go.mod h1:But/NJU6TnZsrLai/xBAQLLz+Hc7fHZJt/hsCz3Fih4= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/kafka/go.mod b/sink/kafka/go.mod index 18fe6786..2db4830e 100644 --- a/sink/kafka/go.mod +++ b/sink/kafka/go.mod @@ -30,7 +30,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.6 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/kafka/go.sum b/sink/kafka/go.sum index b5d66cbf..fe699692 100644 --- a/sink/kafka/go.sum +++ b/sink/kafka/go.sum @@ -72,8 +72,8 @@ github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh6 github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs= github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY= github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc= -github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= -github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/kinesis/go.mod b/sink/kinesis/go.mod index f3f06e85..beffe1bf 100644 --- a/sink/kinesis/go.mod +++ b/sink/kinesis/go.mod @@ -45,7 +45,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/kinesis/go.sum b/sink/kinesis/go.sum index 42461144..38b710d9 100644 --- a/sink/kinesis/go.sum +++ b/sink/kinesis/go.sum @@ -79,8 +79,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/nats/go.mod b/sink/nats/go.mod index 834350c9..0ae915ae 100644 --- a/sink/nats/go.mod +++ b/sink/nats/go.mod @@ -34,7 +34,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/nats/go.sum b/sink/nats/go.sum index 74fab310..bead3f2d 100644 --- a/sink/nats/go.sum +++ b/sink/nats/go.sum @@ -46,8 +46,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/redis/go.mod b/sink/redis/go.mod index 1f53562a..b7c84fcf 100644 --- a/sink/redis/go.mod +++ b/sink/redis/go.mod @@ -30,7 +30,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/mdelapenya/tlscert v0.2.0 // indirect diff --git a/sink/redis/go.sum b/sink/redis/go.sum index eee7d55a..68e9e767 100644 --- a/sink/redis/go.sum +++ b/sink/redis/go.sum @@ -50,8 +50,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= diff --git a/sink/s3/go.mod b/sink/s3/go.mod index 533c7295..f48adeaf 100644 --- a/sink/s3/go.mod +++ b/sink/s3/go.mod @@ -47,7 +47,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/s3/go.sum b/sink/s3/go.sum index 8f923534..3e14a396 100644 --- a/sink/s3/go.sum +++ b/sink/s3/go.sum @@ -83,8 +83,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/sns/go.mod b/sink/sns/go.mod index 13cee599..c054a378 100644 --- a/sink/sns/go.mod +++ b/sink/sns/go.mod @@ -49,7 +49,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/sns/go.sum b/sink/sns/go.sum index 93c16c16..b0f50e59 100644 --- a/sink/sns/go.sum +++ b/sink/sns/go.sum @@ -79,8 +79,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/sqs/go.mod b/sink/sqs/go.mod index 2eb7802b..23ef92e4 100644 --- a/sink/sqs/go.mod +++ b/sink/sqs/go.mod @@ -44,7 +44,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/sqs/go.sum b/sink/sqs/go.sum index f9138b32..4f5c3f23 100644 --- a/sink/sqs/go.sum +++ b/sink/sqs/go.sum @@ -77,8 +77,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/sink/timestream/go.mod b/sink/timestream/go.mod index d8cb7f9b..cb8c4e24 100644 --- a/sink/timestream/go.mod +++ b/sink/timestream/go.mod @@ -45,7 +45,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/sink/timestream/go.sum b/sink/timestream/go.sum index 88001ca6..cca7b1bb 100644 --- a/sink/timestream/go.sum +++ b/sink/timestream/go.sum @@ -79,8 +79,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/source/jetstream/go.mod b/source/jetstream/go.mod index e58f2130..db14c83a 100644 --- a/source/jetstream/go.mod +++ b/source/jetstream/go.mod @@ -34,7 +34,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/source/jetstream/go.sum b/source/jetstream/go.sum index 74fab310..bead3f2d 100644 --- a/source/jetstream/go.sum +++ b/source/jetstream/go.sum @@ -46,8 +46,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/source/kafka/go.mod b/source/kafka/go.mod index 5c9813d0..a70da433 100644 --- a/source/kafka/go.mod +++ b/source/kafka/go.mod @@ -10,7 +10,7 @@ require ( ) require ( - github.com/klauspost/compress v1.18.6 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/pierrec/lz4/v4 v4.1.26 // indirect github.com/stablekernel/crucible/telemetry v0.0.0 // indirect ) diff --git a/source/kafka/go.sum b/source/kafka/go.sum index b5d6d347..b31eaeff 100644 --- a/source/kafka/go.sum +++ b/source/kafka/go.sum @@ -1,5 +1,7 @@ github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/twmb/franz-go v1.21.2 h1:WrvV/spF48JzcRylqDQy02Vm6V6W4lhtD9Y4BOYNMu4= diff --git a/source/kafka/integration/go.mod b/source/kafka/integration/go.mod index 854bd50b..093e3b47 100644 --- a/source/kafka/integration/go.mod +++ b/source/kafka/integration/go.mod @@ -39,7 +39,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.6 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/source/kafka/integration/go.sum b/source/kafka/integration/go.sum index c32579a4..d534d1a1 100644 --- a/source/kafka/integration/go.sum +++ b/source/kafka/integration/go.sum @@ -46,8 +46,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= -github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= diff --git a/source/redis/go.mod b/source/redis/go.mod index f3acfca1..c6321d40 100644 --- a/source/redis/go.mod +++ b/source/redis/go.mod @@ -34,7 +34,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/mdelapenya/tlscert v0.2.0 // indirect diff --git a/source/redis/go.sum b/source/redis/go.sum index eee7d55a..68e9e767 100644 --- a/source/redis/go.sum +++ b/source/redis/go.sum @@ -50,8 +50,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= diff --git a/telemetry/datadog/go.mod b/telemetry/datadog/go.mod index 657e839a..21dc9a0d 100644 --- a/telemetry/datadog/go.mod +++ b/telemetry/datadog/go.mod @@ -39,7 +39,7 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/hashicorp/go-version v1.8.0 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/compress v1.18.4 // indirect + github.com/klauspost/compress v1.20.2 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/linkdata/deadlock v0.5.5 // indirect github.com/lufia/plan9stats v0.0.0-20260216142805-b3301c5f2a88 // indirect diff --git a/telemetry/datadog/go.sum b/telemetry/datadog/go.sum index 29fdc6bb..1ad49834 100644 --- a/telemetry/datadog/go.sum +++ b/telemetry/datadog/go.sum @@ -87,8 +87,8 @@ github.com/hashicorp/go-version v1.8.0 h1:KAkNb1HAiZd1ukkxDFGmokVZe1Xy9HG6NUp+bP github.com/hashicorp/go-version v1.8.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= -github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c= -github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/klauspost/compress v1.20.2 h1:v5tiZvm1KGLyg5d5oXdXe3lsSPrbEUiKoXGia//I7Qo= +github.com/klauspost/compress v1.20.2/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=