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 diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 25947ae6..347366e0 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -90,10 +90,20 @@ Two stdlib-only reference implementations ship in-tree: retains the full record history, enabling time-travel below. - **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance subdirectories, each an append-only journal, an atomic checkpoint, an - idempotency ledger, and a dispatched-effect log. Each append flushes to stable - storage; each checkpoint uses write-temp-plus-rename for crash-safe atomicity. + idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file. + Each append flushes to stable storage; each checkpoint uses + write-temp-plus-rename for crash-safe atomicity. Use it for durability across restarts without a database. +A write can carry a fence token (`WithFence`, `WithCheckpointFence`). A `Store` +keeps the highest token it has seen for an instance and refuses a lower one with +`ErrFenced`, writing nothing, so an owner that lost an instance is refused once the +new owner has written. `FileStore` fencing holds only within one open store. +`WithOwnerFence` sets the token a `Runner` carries on every `Append` and +`Checkpoint`. `Append` also rejects a different record at a recorded step with +`ErrStepConflict`. An out-of-tree `Store` reads the token with `AppendFence` and +`CheckpointFence`. + Persistent database backends (PostgreSQL, DynamoDB, and the like) implement `Store` out of tree, so their drivers never burden this module's dependency or vulnerability surface. diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index ed258652..74a844a0 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -7,9 +7,37 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. ## [Unreleased] +### Added + +- **Fencing.** `WithFence` on `Store.Append` and `WithCheckpointFence` on + `Store.Checkpoint` carry a caller-issued token, and a Store refuses a token lower + than the highest it has seen for the instance with `ErrFenced`, writing nothing. + `WithOwnerFence` sets the token a Runner carries on every Append and Checkpoint; + `MarkDispatched` is not fenced. `AppendFence`, + `CheckpointFence` and `AppendIdempotencyKey` let a backend outside this package + read its options. `MemStore` and `FileStore` enforce fencing; `FileStore` persists the + highest token in a `fence` file and holds its fencing within one open store. A + Store first sees a token on that owner's first write, so `Recover` alone doesn't + fence the old owner: a stale write before then is accepted, and the new owner's + write at that step gets `ErrStepConflict`. + ### Changed - Requires Go 1.27.3. +- **A `Store` implementation has new work to do.** It must enforce fence tokens and + return `ErrStepConflict`, or `WithOwnerFence` protects nothing. Existing + implementations still compile, so nothing flags this. +- **`MemStore.Append` without `WithIdempotencyKey` refuses a Record JSON can't encode,** + as `FileStore` already did. + +- **`Store.Append` rejects a different record for a recorded step.** It used to keep + the first and report success, which acknowledged a step the journal didn't hold. + It now returns `ErrStepConflict`, including for a step a checkpoint has compacted. + The same record is still an idempotent no-op, and an Append with + `WithIdempotencyKey` still dedups on the key alone. A `FileStore` ledger line from + before this change has no digest, so its step keeps the old no-op behavior. A + recovered instance that reuses a step number now fails with `ErrStepConflict` + instead of silently dropping the step. ### Fixed diff --git a/durable/README.md b/durable/README.md index 45db9a1e..faad6cd2 100644 --- a/durable/README.md +++ b/durable/README.md @@ -100,10 +100,20 @@ Two stdlib-only reference implementations ship in-tree: retains the full record history (enabling time-travel below). - **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance subdirectories, each an append-only journal, an atomic checkpoint, an - idempotency ledger, and a dispatched-effect log. Each append flushes to stable - storage; each checkpoint uses write-temp-plus-rename for crash-safe atomicity. + idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file. + Each append flushes to stable storage; each checkpoint uses + write-temp-plus-rename for crash-safe atomicity. Use it for durability across restarts without a database. +A write can carry a fence token (`WithFence`, `WithCheckpointFence`). A `Store` +keeps the highest token it has seen for an instance and refuses a lower one with +`ErrFenced`, writing nothing, so an owner that lost an instance is refused once the +new owner has written. `FileStore` fencing holds only within one open store. +`WithOwnerFence` sets the token a `Runner` carries on every `Append` and +`Checkpoint`. `Append` also rejects a different record at a recorded step with +`ErrStepConflict`. An out-of-tree `Store` reads the token with `AppendFence` and +`CheckpointFence`. + Persistent database backends (PostgreSQL, DynamoDB, ...) implement `Store` out of tree, so their drivers never burden this module's dependency or vulnerability surface. 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/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..7b9d0752 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -2,29 +2,54 @@ package durable import "errors" +// Sentinel error text. The Err values below and ErrStepOutOfRange build on it. +const ( + errorPrefix = "durable: " + + errorInstanceNotFound = "instance not found" + errorStepOutOfOrder = "record step out of order" + errorStepConflict = "step already recorded with a different record" + errorFenced = "write fenced by a newer token" + errorCheckpointNotAdvancing = "checkpoint does not advance" + errorInstanceExists = "instance already exists" + errorEffectDispatch = "effect dispatch failed" + errorStepOutOfRange = "time-travel step out of range" +) + // Sentinel errors a Store reports. Callers match them with errors.Is. var ( // ErrInstanceNotFound is reported by Store.Load for an instance that has // never been written. - ErrInstanceNotFound = errors.New("crucible/durable: instance not found") + ErrInstanceNotFound = errors.New(errorPrefix + errorInstanceNotFound) // ErrStepOutOfOrder is reported by Store.Append when a Record's Step does not // strictly follow the instance's highest recorded Step (and is not an // idempotent re-append of an already-recorded Step). - ErrStepOutOfOrder = errors.New("crucible/durable: record step out of order") + ErrStepOutOfOrder = errors.New(errorPrefix + errorStepOutOfOrder) + + // ErrStepConflict is reported by Store.Append when a Step is already recorded + // with a different Record: the writer's view of the instance has diverged from + // the journal, usually because another writer got there first. The Record was + // not written. + ErrStepConflict = errors.New(errorPrefix + errorStepConflict) + + // ErrFenced is reported by Store.Append and Store.Checkpoint for a write whose + // fence token is lower than the highest the Store has seen for the instance: + // a newer owner has taken the instance over. Nothing is written. + ErrFenced = errors.New(errorPrefix + errorFenced) // ErrCheckpointNotAdvancing is reported by Store.Checkpoint when throughStep // does not advance beyond the instance's current checkpoint. - ErrCheckpointNotAdvancing = errors.New("crucible/durable: checkpoint does not advance") + ErrCheckpointNotAdvancing = errors.New(errorPrefix + errorCheckpointNotAdvancing) // ErrInstanceExists is reported by Runner.Start when the InstanceID is already // present in the Store, so a fresh Start does not clobber a recorded instance. - ErrInstanceExists = errors.New("crucible/durable: instance already exists") + ErrInstanceExists = errors.New(errorPrefix + errorInstanceExists) // ErrEffectDispatch wraps a failure returned by the caller-supplied effect // handler (WithEffectHandler) while applying an emitted effect. The effect was // persisted (write-ahead) but not marked dispatched, so a later recovery // retries it; callers match this with errors.Is to distinguish a dispatch // failure from a kernel transition or persistence error. - ErrEffectDispatch = errors.New("crucible/durable: effect dispatch failed") + ErrEffectDispatch = errors.New(errorPrefix + errorEffectDispatch) ) diff --git a/durable/fence_test.go b/durable/fence_test.go new file mode 100644 index 00000000..d921bb8c --- /dev/null +++ b/durable/fence_test.go @@ -0,0 +1,552 @@ +package durable_test + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "sync" + "sync/atomic" + "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) + } +} + +// 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) { + 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) + } +} + +// 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 aab6d5cd..b7a9462d 100644 --- a/durable/filestore.go +++ b/durable/filestore.go @@ -27,8 +27,9 @@ 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 // // 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{} } @@ -97,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...), @@ -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 @@ -167,21 +177,24 @@ 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 { 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 { + return false, err + } if err := fi.loadCheckpoint(); err != nil { return false, err } @@ -194,7 +207,52 @@ 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 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 +// 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("durable: reading fence for %q: %w", fi.dir, err) + } + if err := json.Unmarshal(b, &fi.fence); err != nil { + return fmt.Errorf("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("durable: marshaling fence for %q: %w", id, err) + } + if err := writeAtomic(filepath.Join(fi.dir, fenceFile), b); err != nil { + return fmt.Errorf("durable: writing fence for %q: %w", id, err) + } + fi.fence = token + return nil } // loadCheckpoint reads the latest checkpoint Snapshot bytes and its throughStep, @@ -205,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 { @@ -213,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 @@ -236,7 +294,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 } @@ -284,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 { @@ -298,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 } @@ -310,18 +368,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 +385,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", @@ -344,25 +406,25 @@ 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}) + 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)) inst.maxStep = rec.Step inst.seq = seq - inst.applied[key] = seq + inst.applied[key] = appliedRecord{seq: seq, digest: digest} return seq, nil } @@ -394,7 +456,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 +465,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) @@ -415,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. @@ -433,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...) @@ -468,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{}{} } @@ -496,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 } @@ -534,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/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/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..289245fa 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,10 +143,17 @@ 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 } +// 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) { @@ -143,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) } @@ -170,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) } @@ -195,16 +213,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 +298,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("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..0fa6c994 100644 --- a/durable/options.go +++ b/durable/options.go @@ -17,12 +17,15 @@ 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 // 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 != "" { @@ -31,6 +34,32 @@ 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 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 { + 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 +69,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 the same as no token, which a fenced instance refuses. +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..7d5e4d60 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,24 @@ 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 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. 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] { + 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..13a35a03 100644 --- a/durable/store.go +++ b/durable/store.go @@ -16,11 +16,18 @@ 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). +// +// 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 - // 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 +87,36 @@ 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. 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, 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. 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 @@ -103,10 +132,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 +150,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 } 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, 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=