From fd25bdab846c3adc1942a1ef2718fa83b1c3559b Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 14:17:04 +0900 Subject: [PATCH 1/8] Fix lost Redis appends during Raft leader handoff --- kv/fsm.go | 42 ++++++++++++++++--- kv/fsm_occ_test.go | 5 ++- kv/fsm_onephase_dedup_test.go | 30 +++++++++++++ kv/leader_routed_store.go | 7 ++++ kv/shard_store.go | 13 ++++++ store/apply_order_fence_jepsen_test.go | 56 +++++++++++++++++++++++++ store/lsm_store.go | 58 ++++++++++++++++++++++---- store/lsm_store_applied_index_test.go | 27 ++++++++++-- store/mvcc_store.go | 11 +++++ store/raft_entry.go | 45 ++++++++++++++++---- store/store.go | 15 ++++--- 11 files changed, 276 insertions(+), 33 deletions(-) create mode 100644 store/apply_order_fence_jepsen_test.go diff --git a/kv/fsm.go b/kv/fsm.go index 15eeaffc2..30f444699 100644 --- a/kv/fsm.go +++ b/kv/fsm.go @@ -1514,7 +1514,7 @@ func (f *kvFSM) handleOnePhaseTxnRequest(ctx context.Context, r *pb.Request, com // applying this log entry. The retention-window > max-retry-latency // invariant prevents the rare case where a real never-landed retry // arrives with PrevCommitTS below pebble's compacted floor. - dedup, err := f.dedupProbeOnePhase(ctx, meta, r.ReadKeys) + dedup, err := f.dedupProbeOnePhase(ctx, meta, r.ReadKeys, muts) if err != nil { return err } @@ -1531,7 +1531,7 @@ func (f *kvFSM) handleOnePhaseTxnRequest(ctx context.Context, r *pb.Request, com if err != nil { return err } - if err := f.store.ApplyMutationsRaftAt(ctx, storeMuts, r.ReadKeys, startTS, commitTS, f.pendingApplyIdx); err != nil { + if err := f.store.ApplyMutationsRaftAtFenced(ctx, storeMuts, r.ReadKeys, startTS, commitTS, f.pendingApplyIdx); err != nil { return errors.WithStack(err) } f.notifyApplyObservers(commitTS, uniq) @@ -1630,7 +1630,7 @@ func (f *kvFSM) uniqueTxnMutationsAboveFloorWithSnapshot(muts []*pb.Mutation, co // Returns (true, nil) → the entry must no-op (prior attempt landed). // Returns (false, nil) → fall through to normal apply. // Returns (false, err) → propagate err; apply must not proceed. -func (f *kvFSM) dedupProbeOnePhase(ctx context.Context, meta TxnMeta, readKeys [][]byte) (bool, error) { +func (f *kvFSM) dedupProbeOnePhase(ctx context.Context, meta TxnMeta, readKeys [][]byte, muts []*pb.Mutation) (bool, error) { if meta.PrevCommitTS == 0 { return false, nil } @@ -1639,7 +1639,7 @@ func (f *kvFSM) dedupProbeOnePhase(ctx context.Context, meta TxnMeta, readKeys [ return false, errors.WithStack(err) } if landed { - return true, nil + return f.priorListItemsMatch(ctx, muts, readKeys, meta.PrevCommitTS) } for _, stagedKey := range stagedVisibilityKeysForReadAlias(readKeys, meta.PrimaryKey, meta.PrimaryKey) { landed, err = f.store.CommittedVersionAt(ctx, stagedKey, meta.PrevCommitTS) @@ -1647,12 +1647,44 @@ func (f *kvFSM) dedupProbeOnePhase(ctx context.Context, meta TxnMeta, readKeys [ return false, errors.WithStack(err) } if landed { - return true, nil + return f.priorListItemsMatch(ctx, muts, readKeys, meta.PrevCommitTS) } } return false, nil } +// A timestamp may be issued by two leaders during a handoff. The existence +// of an item version at PrevCommitTS alone cannot identify this retry: a +// different append can have written the same list position at that timestamp. +func (f *kvFSM) priorListItemsMatch(ctx context.Context, muts []*pb.Mutation, readKeys [][]byte, prevTS uint64) (bool, error) { + for _, mut := range muts { + if mut == nil || !store.IsListItemKey(mut.Key) || mut.Op != pb.Op_PUT { + continue + } + keys := append([][]byte{mut.Key}, stagedVisibilityKeysForReadAlias(readKeys, mut.Key, mut.Key)...) + matched := false + for _, key := range keys { + landed, err := f.store.CommittedVersionAt(ctx, key, prevTS) + if err != nil { + return false, errors.WithStack(err) + } + if !landed { + continue + } + value, err := f.store.GetAt(ctx, key, prevTS) + if err != nil { + return false, errors.WithStack(err) + } + matched = bytes.Equal(value, mut.Value) + break + } + if !matched { + return false, nil + } + } + return true, nil +} + func stagedVisibilityJobIDsForReadAlias(readKeys [][]byte, aliasRawKey []byte) []uint64 { if len(readKeys) == 0 { return nil diff --git a/kv/fsm_occ_test.go b/kv/fsm_occ_test.go index f6d1fdd6c..973d12067 100644 --- a/kv/fsm_occ_test.go +++ b/kv/fsm_occ_test.go @@ -180,14 +180,15 @@ func TestOnePhaseTxnNilReadKeysSkipsReadWriteValidation(t *testing.T) { fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM) require.True(t, ok) - // nil ReadKeys: read-write check is skipped entirely. + // nil ReadKeys: read-write check is skipped entirely. Use a commit + // timestamp above the applied watermark so the apply-order fence permits it. req := &pb.Request{ IsTxn: true, Phase: pb.Phase_NONE, Ts: 100, ReadKeys: nil, Mutations: []*pb.Mutation{ - {Op: pb.Op_PUT, Key: []byte(txnMetaPrefix), Value: EncodeTxnMeta(TxnMeta{PrimaryKey: []byte("wk"), CommitTS: 120})}, + {Op: pb.Op_PUT, Key: []byte(txnMetaPrefix), Value: EncodeTxnMeta(TxnMeta{PrimaryKey: []byte("wk"), CommitTS: 220})}, {Op: pb.Op_PUT, Key: []byte("wk"), Value: []byte("val")}, }, } diff --git a/kv/fsm_onephase_dedup_test.go b/kv/fsm_onephase_dedup_test.go index c9f06c088..7e5191489 100644 --- a/kv/fsm_onephase_dedup_test.go +++ b/kv/fsm_onephase_dedup_test.go @@ -69,6 +69,36 @@ func TestOnePhaseDedup_NoOpsWhenPriorAttemptLanded(t *testing.T) { require.Equal(t, uint64(20), latest, "newest version must remain attempt 1's at 20") } +func TestOnePhaseDedup_DifferentWriteAtPriorTimestampIsNotOurAttempt(t *testing.T) { + ctx := context.Background() + st := store.NewMVCCStore() + fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM) + require.True(t, ok) + key := store.ListItemKey([]byte("jepsen-list"), 43) + require.NoError(t, applyFSMRequest(t, fsm, onePhaseReq(10, 20, 0, key, []byte("45")))) + + // A second append allocated the same timestamp, but its first attempt + // never landed. Its retry must not mistake the first append for itself. + err := applyFSMRequest(t, fsm, onePhaseReq(10, 40, 20, key, []byte("44"))) + require.ErrorIs(t, err, store.ErrWriteConflict) + got, err := st.GetAt(ctx, key, 40) + require.NoError(t, err) + require.Equal(t, []byte("45"), got) +} + +func TestOnePhaseDedup_ListItemRetryWithMatchingValueNoOps(t *testing.T) { + ctx := context.Background() + st := store.NewMVCCStore() + fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM) + require.True(t, ok) + key := store.ListItemKey([]byte("jepsen-list"), 43) + require.NoError(t, applyFSMRequest(t, fsm, onePhaseReq(10, 20, 0, key, []byte("44")))) + require.NoError(t, applyFSMRequest(t, fsm, onePhaseReq(30, 40, 20, key, []byte("44")))) + landed, err := st.CommittedVersionAt(ctx, key, 40) + require.NoError(t, err) + require.False(t, landed) +} + func TestOnePhaseDedup_NoOpsWhenPriorAttemptLandedAsStagedVersion(t *testing.T) { t.Parallel() ctx := context.Background() diff --git a/kv/leader_routed_store.go b/kv/leader_routed_store.go index af4392b7a..1615bf8d5 100644 --- a/kv/leader_routed_store.go +++ b/kv/leader_routed_store.go @@ -570,6 +570,13 @@ func (s *LeaderRoutedStore) ApplyMutationsRaftAt(ctx context.Context, mutations return errors.WithStack(s.local.ApplyMutationsRaftAt(ctx, mutations, readKeys, startTS, commitTS, appliedIndex)) } +func (s *LeaderRoutedStore) ApplyMutationsRaftAtFenced(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error { + if s == nil || s.local == nil { + return errors.WithStack(store.ErrNotSupported) + } + return errors.WithStack(s.local.ApplyMutationsRaftAtFenced(ctx, mutations, readKeys, startTS, commitTS, appliedIndex)) +} + func (s *LeaderRoutedStore) DeletePrefixAt(ctx context.Context, prefix []byte, excludePrefix []byte, commitTS uint64) error { if s == nil || s.local == nil { return errors.WithStack(store.ErrNotSupported) diff --git a/kv/shard_store.go b/kv/shard_store.go index 0ce37ce54..9c4301a99 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -5179,6 +5179,19 @@ func (s *ShardStore) ApplyMutationsRaftAt(ctx context.Context, mutations []*stor return errors.WithStack(group.Store.ApplyMutationsRaftAt(ctx, mutations, readKeys, startTS, commitTS, appliedIndex)) } +func (s *ShardStore) ApplyMutationsRaftAtFenced(ctx context.Context, mutations []*store.KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error { + group, err := s.resolveSingleShardGroup(mutations) + if err != nil || group == nil { + return err + } + if err := s.ensureMutationWriteTimestampFloors(mutations, commitTS); err != nil { + return err + } + readKeys = s.readKeysWithStagedVisibilityAliases(group, readKeys) + readKeys = s.readKeysWithStagedVisibilityMutationAliases(group, readKeys, mutations) + return errors.WithStack(group.Store.ApplyMutationsRaftAtFenced(ctx, mutations, readKeys, startTS, commitTS, appliedIndex)) +} + func ensureRouteWriteTimestampFloor(route distribution.Route, key []byte, commitTS uint64) error { if route.MinWriteTSExclusive == 0 || commitTS == 0 || commitTS > route.MinWriteTSExclusive { return nil diff --git a/store/apply_order_fence_jepsen_test.go b/store/apply_order_fence_jepsen_test.go new file mode 100644 index 000000000..a9cc1dd38 --- /dev/null +++ b/store/apply_order_fence_jepsen_test.go @@ -0,0 +1,56 @@ +package store + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestOnePhaseApplyRejectsTimestampAtAppliedWatermark(t *testing.T) { + for _, tc := range []struct { + name string + new func(*testing.T) MVCCStore + }{ + {"memory", func(*testing.T) MVCCStore { return NewMVCCStore() }}, + {"pebble", newApplyIndexPebbleStore}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + st := tc.new(t) + put := func(key, val string) []*KVPairMutation { + return []*KVPairMutation{{Op: OpTypePut, Key: []byte(key), Value: []byte(val)}} + } + require.NoError(t, st.ApplyMutationsRaftAtFenced(ctx, put("first", "45"), nil, 99, 100, 10)) + for _, ts := range []uint64{100, 99} { + err := st.ApplyMutationsRaftAtFenced(ctx, put("later", "44"), nil, 98, ts, 11) + require.ErrorIs(t, err, ErrWriteConflict) + } + _, err := st.GetAt(ctx, []byte("later"), 100) + require.ErrorIs(t, err, ErrKeyNotFound) + require.Equal(t, uint64(100), st.LastCommitTS()) + require.NoError(t, st.ApplyMutationsRaftAtFenced(ctx, put("later", "44"), nil, 100, 101, 11)) + }) + } +} + +func TestOnePhaseRaftEntryRejectsCollidingAppend(t *testing.T) { + ctx := context.Background() + st := newApplyIndexPebbleStore(t) + applier := raftEntryApplier(t, st) + key := ListItemKey([]byte("jepsen-list"), 43) + put := func(value string) []*KVPairMutation { + return []*KVPairMutation{{Op: OpTypePut, Key: key, Value: []byte(value)}} + } + require.NoError(t, applier.ApplyRaftEntry(ctx, 10, func(ctx context.Context) error { + return st.ApplyMutationsRaftAtFenced(ctx, put("45"), nil, 99, 100, 10) + })) + require.NoError(t, applier.ApplyRaftEntry(ctx, 11, func(ctx context.Context) error { + err := st.ApplyMutationsRaftAtFenced(ctx, put("44"), nil, 99, 100, 11) + require.ErrorIs(t, err, ErrWriteConflict) + return nil + })) + got, err := st.GetAt(ctx, key, 100) + require.NoError(t, err) + require.Equal(t, []byte("45"), got) +} diff --git a/store/lsm_store.go b/store/lsm_store.go index 62acb45b2..5d872a6ee 100644 --- a/store/lsm_store.go +++ b/store/lsm_store.go @@ -2389,14 +2389,14 @@ func (s *pebbleStore) ApplyMutations(ctx context.Context, mutations []*KVPairMut // registration commits. // appliedIndex=0: direct path has no raft index; the leaf treats 0 as // "do not write metaAppliedIndex" so the meta key stays unchanged. - return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.directApplyWriteOpts(), true, 0, true) + return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.directApplyWriteOpts(), true, 0, true, false) } // ApplyMutationsPreservingLastCommitTS is a direct, durable write path for // local auxiliary MVCC keys that must not advance the store-wide safe snapshot // watermark. It still uses pebble.Sync and the direct writer-registration gate. func (s *pebbleStore) ApplyMutationsPreservingLastCommitTS(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS uint64) error { - return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.directApplyWriteOpts(), true, 0, false) + return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.directApplyWriteOpts(), true, 0, false, false) } // ApplyMutationsRaft is the raft-apply commit path. Durability is governed @@ -2425,7 +2425,7 @@ func (s *pebbleStore) ApplyMutationsRaft(ctx context.Context, mutations []*KVPai // land here; their LastAppliedIndex() will stay behind the snapshot // pointer and the skip optimisation will fall back to full restore // for them. Preferred path is ApplyMutationsRaftAt. - return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.raftApplyWriteOpts(), false, 0, true) + return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.raftApplyWriteOpts(), false, 0, true, false) } // ApplyMutationsRaftAt is ApplyMutationsRaft with the raft entry @@ -2440,7 +2440,14 @@ func (s *pebbleStore) ApplyMutationsRaftAt(ctx context.Context, mutations []*KVP if scope := s.raftEntryScopeFrom(ctx); scope != nil { return s.stageMutationsInRaftEntry(ctx, scope, mutations, readKeys, startTS, commitTS, appliedIndex) } - return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.raftApplyWriteOpts(), false, appliedIndex, true) + return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.raftApplyWriteOpts(), false, appliedIndex, true, false) +} + +func (s *pebbleStore) ApplyMutationsRaftAtFenced(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error { + if scope := s.raftEntryScopeFrom(ctx); scope != nil { + return s.stageMutationsInRaftEntryWithFence(ctx, scope, mutations, readKeys, startTS, commitTS, appliedIndex, true) + } + return s.applyMutationsWithOpts(ctx, mutations, readKeys, startTS, commitTS, s.raftApplyWriteOpts(), false, appliedIndex, true, true) } func (s *pebbleStore) raftApplyAlreadyLandedLocked(mutations []*KVPairMutation, commitTS uint64) (bool, error) { @@ -2527,6 +2534,16 @@ func (s *pebbleStore) staleRaftApplyFastPathLocked(mutations []*KVPairMutation, if appliedIndex == 0 { return false, nil } + persisted, present, err := s.readAppliedIndexLocked() + if err != nil || !present { + return false, err + } + if appliedIndex < persisted { + return true, nil + } + if appliedIndex > persisted { + return false, nil + } landed, err := s.raftApplyAlreadyLandedLocked(mutations, commitTS) if err != nil || !landed { return false, err @@ -2534,6 +2551,21 @@ func (s *pebbleStore) staleRaftApplyFastPathLocked(mutations []*KVPairMutation, return true, s.markRaftApplyAlreadyLandedLocked(commitTS, appliedIndex, writeOpts) } +// checkCommitTSFence runs while the apply lock is held. Readers can use the +// applied watermark as a snapshot only if a later Raft entry cannot add a +// version at or below it. +func checkCommitTSFence(mutations []*KVPairMutation, commitTS, watermark uint64, fenced bool) error { + if !fenced || commitTS > watermark { + return nil + } + for _, mut := range mutations { + if mut != nil && len(mut.Key) > 0 { + return NewWriteConflictError(mut.Key) + } + } + return nil +} + func (s *pebbleStore) hasCommitsAfter(startTS uint64) bool { s.mtx.RLock() defer s.mtx.RUnlock() @@ -2559,7 +2591,7 @@ func (s *pebbleStore) checkApplyConflictsWith(ctx context.Context, r pebble.Read return s.checkReadConflictsWith(ctx, r, readKeys, startTS) } -func (s *pebbleStore) applyMutationsWithOpts(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS uint64, writeOpts *pebble.WriteOptions, gateRegistration bool, appliedIndex uint64, advanceLastCommitTS bool) error { +func (s *pebbleStore) applyMutationsWithOpts(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS uint64, writeOpts *pebble.WriteOptions, gateRegistration bool, appliedIndex uint64, advanceLastCommitTS, fenceCommitTS bool) error { // Serialize conflict check → batch commit so concurrent ApplyMutations // cannot both pass checkConflicts and then both commit. applyMu comes // before dbMu (see the lock ordering on pebbleStore). @@ -2576,6 +2608,9 @@ func (s *pebbleStore) applyMutationsWithOpts(ctx context.Context, mutations []*K if fastPathDone { return nil } + if err := checkCommitTSFence(mutations, commitTS, s.LastCommitTS(), fenceCommitTS); err != nil { + return err + } b := s.db.NewBatch() defer b.Close() @@ -2600,10 +2635,8 @@ func (s *pebbleStore) applyMutationsWithOpts(ctx context.Context, mutations []*K // meta key so a crash either commits all three atomically or none. // appliedIndex==0 is the legacy / non-raft callers (ApplyMutations // or ApplyMutationsRaft); they leave the key unchanged. - if appliedIndex > 0 { - if err := setPebbleUint64InBatch(b, metaAppliedIndexBytes, appliedIndex); err != nil { - return err - } + if err := stageAppliedIndexInBatch(b, appliedIndex); err != nil { + return err } if err := b.Commit(writeOpts); err != nil { return errors.WithStack(err) @@ -2615,6 +2648,13 @@ func (s *pebbleStore) applyMutationsWithOpts(ctx context.Context, mutations []*K return nil } +func stageAppliedIndexInBatch(b *pebble.Batch, appliedIndex uint64) error { + if appliedIndex == 0 { + return nil + } + return setPebbleUint64InBatch(b, metaAppliedIndexBytes, appliedIndex) +} + func (s *pebbleStore) stageLastCommitTSInBatch(b *pebble.Batch, commitTS uint64, advance bool) (uint64, func(), error) { if !advance { return 0, func() {}, nil diff --git a/store/lsm_store_applied_index_test.go b/store/lsm_store_applied_index_test.go index 2811e7150..f22265f32 100644 --- a/store/lsm_store_applied_index_test.go +++ b/store/lsm_store_applied_index_test.go @@ -98,6 +98,27 @@ func TestApplyMutationsRaftAt_BundlesMetaAppliedIndex(t *testing.T) { require.Equal(t, []byte("v1"), val) } +func TestApplyMutationsRaftAt_DistinctEntryWithSameTimestampIsNotReplay(t *testing.T) { + ctx := context.Background() + st := newApplyIndexPebbleStore(t) + key := []byte("jepsen-list") + itemKey := ListItemKey(key, 43) + deltaKey := ListMetaDeltaKey(key, 100, 0) + appendAt := func(value string) []*KVPairMutation { + return []*KVPairMutation{ + {Op: OpTypePut, Key: itemKey, Value: []byte(value)}, + {Op: OpTypePut, Key: deltaKey, Value: MarshalListMetaDelta(ListMetaDelta{LenDelta: 1})}, + } + } + + require.NoError(t, st.ApplyMutationsRaftAt(ctx, appendAt("45"), nil, 99, 100, 10)) + err := st.ApplyMutationsRaftAt(ctx, appendAt("44"), nil, 99, 100, 11) + require.ErrorIs(t, err, ErrWriteConflict, "a distinct append must not be acknowledged as a replay") + got, err := st.GetAt(ctx, itemKey, 100) + require.NoError(t, err) + require.Equal(t, []byte("45"), got) +} + func TestPromoteVersions_BundlesMetaAppliedIndex(t *testing.T) { ctx := context.Background() st := newApplyIndexPebbleStore(t) @@ -257,7 +278,7 @@ func TestRetireMigrationRaft_BundlesMetaAppliedIndex(t *testing.T) { require.Equal(t, []byte("v100"), val, "retirement must leave imported data intact") } -func TestApplyMutationsRaftAt_AlreadyLandedAdvancesStaleAppliedIndex(t *testing.T) { +func TestApplyMutationsRaftAt_EqualVersionWithoutRaftIndexIsNotReplay(t *testing.T) { ctx := context.Background() st := newApplyIndexPebbleStore(t) ps := pebbleStoreApplied(t, st) @@ -277,12 +298,12 @@ func TestApplyMutationsRaftAt_AlreadyLandedAdvancesStaleAppliedIndex(t *testing. binary.LittleEndian.PutUint64(staleAppliedIndex[:], entryIdx-1) require.NoError(t, ps.db.Set(metaAppliedIndexBytes, staleAppliedIndex[:], pebble.Sync)) - require.NoError(t, ps.ApplyMutationsRaftAt(ctx, muts, nil, startTS, commitTS, entryIdx)) + require.ErrorIs(t, ps.ApplyMutationsRaftAt(ctx, muts, nil, startTS, commitTS, entryIdx), ErrWriteConflict) got, present, err := ps.LastAppliedIndex() require.NoError(t, err) require.True(t, present) - require.Equal(t, entryIdx, got) + require.Equal(t, entryIdx-1, got) val, err := ps.GetAt(ctx, []byte("k1"), commitTS) require.NoError(t, err) diff --git a/store/mvcc_store.go b/store/mvcc_store.go index cb68752cf..71943d0f9 100644 --- a/store/mvcc_store.go +++ b/store/mvcc_store.go @@ -697,6 +697,10 @@ func (s *mvccStore) ApplyMutationsRaftAt(ctx context.Context, mutations []*KVPai return s.ApplyMutations(ctx, mutations, readKeys, startTS, commitTS) } +func (s *mvccStore) ApplyMutationsRaftAtFenced(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS, _ uint64) error { + return s.applyMutations(ctx, mutations, readKeys, startTS, commitTS, true) +} + // ApplyRaftEntry satisfies RaftEntryApplier. The in-memory store has no // durable state that a crash could tear, so the callback simply runs and // each request applies as it does on the per-request path. @@ -705,8 +709,15 @@ func (s *mvccStore) ApplyRaftEntry(ctx context.Context, _ uint64, fn func(ctx co } func (s *mvccStore) ApplyMutations(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS uint64) error { + return s.applyMutations(ctx, mutations, readKeys, startTS, commitTS, false) +} + +func (s *mvccStore) applyMutations(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS uint64, fenced bool) error { s.mtx.Lock() defer s.mtx.Unlock() + if err := checkCommitTSFence(mutations, commitTS, s.lastCommitTS, fenced); err != nil { + return err + } if s.lastCommitTS > startTS { if err := s.checkConflictsLocked(mutations, readKeys, startTS); err != nil { diff --git a/store/raft_entry.go b/store/raft_entry.go index d0b2a531e..f07fa4b20 100644 --- a/store/raft_entry.go +++ b/store/raft_entry.go @@ -222,6 +222,10 @@ func (s *pebbleStore) commitRaftEntryScope(scope *raftEntryScope) error { // indexed batch, the mutations are encoded before anything is staged so a // validation failure leaves the batch untouched, and nothing is committed. func (s *pebbleStore) stageMutationsInRaftEntry(ctx context.Context, scope *raftEntryScope, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error { + return s.stageMutationsInRaftEntryWithFence(ctx, scope, mutations, readKeys, startTS, commitTS, appliedIndex, false) +} + +func (s *pebbleStore) stageMutationsInRaftEntryWithFence(ctx context.Context, scope *raftEntryScope, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64, fenced bool) error { if scope.failed != nil { return scope.failed } @@ -234,17 +238,17 @@ func (s *pebbleStore) stageMutationsInRaftEntry(ctx context.Context, scope *raft return scope.fail(errors.WithStack(ErrRaftEntryStoreSwapped)) } - if appliedIndex > 0 { - landed, err := s.raftApplyAlreadyLandedWith(scope.batch, mutations, commitTS) - if err != nil { - return scope.fail(err) - } - if landed { + landed, observe, err := s.raftEntryMutationAlreadyLanded(scope, mutations, commitTS, appliedIndex) + if err != nil { + return scope.fail(err) + } + if landed { + if observe { scope.observe(commitTS, true) - return nil } + return nil } - if err := s.checkApplyConflictsWith(ctx, scope.batch, scope.hasCommitsAfter, mutations, readKeys, startTS, false); err != nil { + if err := s.checkRaftEntryMutationConflicts(ctx, scope, mutations, readKeys, startTS, commitTS, fenced); err != nil { return scope.failUnlessVerdict(err) } encoded, err := s.encodeMutationsForBatch(mutations, commitTS, false) @@ -258,6 +262,31 @@ func (s *pebbleStore) stageMutationsInRaftEntry(ctx context.Context, scope *raft return nil } +func (s *pebbleStore) checkRaftEntryMutationConflicts(ctx context.Context, scope *raftEntryScope, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS uint64, fenced bool) error { + if err := checkCommitTSFence(mutations, commitTS, s.LastCommitTS(), fenced); err != nil { + return err + } + return s.checkApplyConflictsWith(ctx, scope.batch, scope.hasCommitsAfter, mutations, readKeys, startTS, false) +} + +func (s *pebbleStore) raftEntryMutationAlreadyLanded(scope *raftEntryScope, mutations []*KVPairMutation, commitTS, appliedIndex uint64) (bool, bool, error) { + if appliedIndex == 0 { + return false, false, nil + } + persisted, present, err := s.readAppliedIndexLocked() + if err != nil { + return false, false, err + } + if present && appliedIndex < persisted { + return true, false, nil + } + if (present && appliedIndex == persisted) || scope.touched { + landed, err := s.raftApplyAlreadyLandedWith(scope.batch, mutations, commitTS) + return landed, landed, err + } + return false, false, nil +} + // failUnlessVerdict returns a per-request verdict untouched and fails the // scope for any other error, which is operational. func (sc *raftEntryScope) failUnlessVerdict(err error) error { diff --git a/store/store.go b/store/store.go index 4090d2340..d04bf94cf 100644 --- a/store/store.go +++ b/store/store.go @@ -267,12 +267,10 @@ type MVCCStore interface { // CommittedVersionAt reports whether a committed version stamped // EXACTLY commitTS exists for key. Unlike GetAt (newest version // <= ts) this is an exact-timestamp existence check, used by the - // one-phase transaction idempotency probe to ask "did the previous - // attempt — which committed at this exact commit_ts — land?". Because - // commit timestamps are issued by the strictly-monotonic, unique HLC - // (Clock().Next()), a version at an exact commitTS on a given key can - // only have come from the transaction that was assigned that timestamp, - // so an exact hit unambiguously identifies that attempt. A tombstone + // one-phase transaction idempotency probe to ask whether the previous + // attempt may have landed. A leader handoff can assign the same timestamp + // to distinct attempts, so an exact hit alone does not identify the retry. + // A tombstone // counts as a landed version (the attempt committed a delete). See // docs/design/2026_05_21_proposed_txn_secondary_idempotency.md. CommittedVersionAt(ctx context.Context, key []byte, commitTS uint64) (bool, error) @@ -317,6 +315,11 @@ type MVCCStore interface { // key", matching ApplyMutationsRaft semantics for callers that have // not yet been wired to the raftengine.ApplyIndexAware seam. ApplyMutationsRaftAt(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error + // ApplyMutationsRaftAtFenced rejects a live one-phase write whose + // commit timestamp is at or below the already-applied watermark. A write + // below that watermark could appear after a reader took that snapshot. + // PREPARE and 2PC COMMIT use the unfenced method above. + ApplyMutationsRaftAtFenced(ctx context.Context, mutations []*KVPairMutation, readKeys [][]byte, startTS, commitTS, appliedIndex uint64) error // DeletePrefixAt atomically deletes all visible (non-tombstone, non-expired) // keys matching prefix at commitTS by writing tombstone versions. An empty // prefix means "all keys". Keys matching excludePrefix are preserved. From 2f5e5a84b2d631df36c41c0f62cc38621571c555 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 15:33:59 +0900 Subject: [PATCH 2/8] Reject colliding writes in Raft batches and retries --- kv/fsm.go | 20 ++++++++++++-------- kv/fsm_onephase_dedup_test.go | 15 +++++++++++++++ store/apply_order_fence_jepsen_test.go | 19 +++++++++++++++++++ store/raft_entry.go | 6 +++--- 4 files changed, 49 insertions(+), 11 deletions(-) diff --git a/kv/fsm.go b/kv/fsm.go index 30f444699..3c196c1a9 100644 --- a/kv/fsm.go +++ b/kv/fsm.go @@ -1639,7 +1639,7 @@ func (f *kvFSM) dedupProbeOnePhase(ctx context.Context, meta TxnMeta, readKeys [ return false, errors.WithStack(err) } if landed { - return f.priorListItemsMatch(ctx, muts, readKeys, meta.PrevCommitTS) + return f.priorMutationValuesMatch(ctx, muts, readKeys, meta.PrevCommitTS) } for _, stagedKey := range stagedVisibilityKeysForReadAlias(readKeys, meta.PrimaryKey, meta.PrimaryKey) { landed, err = f.store.CommittedVersionAt(ctx, stagedKey, meta.PrevCommitTS) @@ -1647,18 +1647,22 @@ func (f *kvFSM) dedupProbeOnePhase(ctx context.Context, meta TxnMeta, readKeys [ return false, errors.WithStack(err) } if landed { - return f.priorListItemsMatch(ctx, muts, readKeys, meta.PrevCommitTS) + return f.priorMutationValuesMatch(ctx, muts, readKeys, meta.PrevCommitTS) } } return false, nil } -// A timestamp may be issued by two leaders during a handoff. The existence -// of an item version at PrevCommitTS alone cannot identify this retry: a -// different append can have written the same list position at that timestamp. -func (f *kvFSM) priorListItemsMatch(ctx context.Context, muts []*pb.Mutation, readKeys [][]byte, prevTS uint64) (bool, error) { - for _, mut := range muts { - if mut == nil || !store.IsListItemKey(mut.Key) || mut.Op != pb.Op_PUT { +// A timestamp may be issued by two leaders during a handoff. An exact +// version at PrevCommitTS alone cannot identify the retry, even for scalar +// writes. Compare every final PUT value in the retained write set. +func (f *kvFSM) priorMutationValuesMatch(ctx context.Context, muts []*pb.Mutation, readKeys [][]byte, prevTS uint64) (bool, error) { + uniq, err := uniqueMutations(muts) + if err != nil { + return false, err + } + for _, mut := range uniq { + if mut.Op != pb.Op_PUT { continue } keys := append([][]byte{mut.Key}, stagedVisibilityKeysForReadAlias(readKeys, mut.Key, mut.Key)...) diff --git a/kv/fsm_onephase_dedup_test.go b/kv/fsm_onephase_dedup_test.go index 7e5191489..c5bbba6f1 100644 --- a/kv/fsm_onephase_dedup_test.go +++ b/kv/fsm_onephase_dedup_test.go @@ -86,6 +86,21 @@ func TestOnePhaseDedup_DifferentWriteAtPriorTimestampIsNotOurAttempt(t *testing. require.Equal(t, []byte("45"), got) } +func TestOnePhaseDedup_DifferentScalarWriteAtPriorTimestampIsNotOurAttempt(t *testing.T) { + ctx := context.Background() + st := store.NewMVCCStore() + fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM) + require.True(t, ok) + key := []byte("jepsen-register") + require.NoError(t, applyFSMRequest(t, fsm, onePhaseReq(10, 20, 0, key, []byte("45")))) + + err := applyFSMRequest(t, fsm, onePhaseReq(10, 40, 20, key, []byte("44"))) + require.ErrorIs(t, err, store.ErrWriteConflict) + got, err := st.GetAt(ctx, key, 40) + require.NoError(t, err) + require.Equal(t, []byte("45"), got) +} + func TestOnePhaseDedup_ListItemRetryWithMatchingValueNoOps(t *testing.T) { ctx := context.Background() st := store.NewMVCCStore() diff --git a/store/apply_order_fence_jepsen_test.go b/store/apply_order_fence_jepsen_test.go index a9cc1dd38..412212237 100644 --- a/store/apply_order_fence_jepsen_test.go +++ b/store/apply_order_fence_jepsen_test.go @@ -54,3 +54,22 @@ func TestOnePhaseRaftEntryRejectsCollidingAppend(t *testing.T) { require.NoError(t, err) require.Equal(t, []byte("45"), got) } + +func TestOnePhaseRaftEntryRejectsCollidingWriteInSameBatch(t *testing.T) { + ctx := context.Background() + st := newApplyIndexPebbleStore(t) + applier := raftEntryApplier(t, st) + key := ListItemKey([]byte("jepsen-list"), 43) + put := func(value string) []*KVPairMutation { + return []*KVPairMutation{{Op: OpTypePut, Key: key, Value: []byte(value)}} + } + require.NoError(t, applier.ApplyRaftEntry(ctx, 10, func(ctx context.Context) error { + require.NoError(t, st.ApplyMutationsRaftAtFenced(ctx, put("45"), nil, 99, 100, 10)) + err := st.ApplyMutationsRaftAtFenced(ctx, put("44"), nil, 99, 100, 10) + require.ErrorIs(t, err, ErrWriteConflict) + return nil + })) + got, err := st.GetAt(ctx, key, 100) + require.NoError(t, err) + require.Equal(t, []byte("45"), got) +} diff --git a/store/raft_entry.go b/store/raft_entry.go index f07fa4b20..0dc75f759 100644 --- a/store/raft_entry.go +++ b/store/raft_entry.go @@ -238,7 +238,7 @@ func (s *pebbleStore) stageMutationsInRaftEntryWithFence(ctx context.Context, sc return scope.fail(errors.WithStack(ErrRaftEntryStoreSwapped)) } - landed, observe, err := s.raftEntryMutationAlreadyLanded(scope, mutations, commitTS, appliedIndex) + landed, observe, err := s.raftEntryMutationAlreadyLanded(scope, mutations, commitTS, appliedIndex, fenced) if err != nil { return scope.fail(err) } @@ -269,7 +269,7 @@ func (s *pebbleStore) checkRaftEntryMutationConflicts(ctx context.Context, scope return s.checkApplyConflictsWith(ctx, scope.batch, scope.hasCommitsAfter, mutations, readKeys, startTS, false) } -func (s *pebbleStore) raftEntryMutationAlreadyLanded(scope *raftEntryScope, mutations []*KVPairMutation, commitTS, appliedIndex uint64) (bool, bool, error) { +func (s *pebbleStore) raftEntryMutationAlreadyLanded(scope *raftEntryScope, mutations []*KVPairMutation, commitTS, appliedIndex uint64, fenced bool) (bool, bool, error) { if appliedIndex == 0 { return false, false, nil } @@ -280,7 +280,7 @@ func (s *pebbleStore) raftEntryMutationAlreadyLanded(scope *raftEntryScope, muta if present && appliedIndex < persisted { return true, false, nil } - if (present && appliedIndex == persisted) || scope.touched { + if (present && appliedIndex == persisted) || (scope.touched && !fenced) { landed, err := s.raftApplyAlreadyLandedWith(scope.batch, mutations, commitTS) return landed, landed, err } From afbb73da220cb5cfa7779b1d3d005b3af4bf53dd Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 17:30:38 +0900 Subject: [PATCH 3/8] Log filesystem write failures before errno conversion --- internal/filesystem/fuseadapter/adapter.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/internal/filesystem/fuseadapter/adapter.go b/internal/filesystem/fuseadapter/adapter.go index eec2aa384..dea449a92 100644 --- a/internal/filesystem/fuseadapter/adapter.go +++ b/internal/filesystem/fuseadapter/adapter.go @@ -190,6 +190,9 @@ func (a *Adapter) Write(ctx context.Context, inode uint64, fh uint64, offset uin return 0, errno } n, err := a.core.Write(ctx, inode, fh, offset, data) + if err != nil { + slog.WarnContext(ctx, "filesystem FUSE write failed", "inode", inode, "offset", offset, "size", len(data), "error", err) + } return n, Errno(err) } @@ -277,7 +280,11 @@ func (a *Adapter) StatFS(ctx context.Context, inode uint64) (filesystem.StatFS, } func (a *Adapter) refreshOpenHandleLease(ctx context.Context, inode uint64, fh uint64) syscall.Errno { - return Errno(a.core.RefreshOpenHandleLease(ctx, inode, fh, a.clientID)) + err := a.core.RefreshOpenHandleLease(ctx, inode, fh, a.clientID) + if err != nil { + slog.WarnContext(ctx, "filesystem FUSE handle refresh failed", "inode", inode, "fh", fh, "error", err) + } + return Errno(err) } func (a *Adapter) trackOpenHandle(inode uint64, fh uint64) { From 01e842493d7452bbc312fd45a74125439f71d5a7 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 17:39:09 +0900 Subject: [PATCH 4/8] Log timestamp fence conflicts --- store/lsm_store.go | 1 + 1 file changed, 1 insertion(+) diff --git a/store/lsm_store.go b/store/lsm_store.go index 5d872a6ee..b03ce4b1d 100644 --- a/store/lsm_store.go +++ b/store/lsm_store.go @@ -2560,6 +2560,7 @@ func checkCommitTSFence(mutations []*KVPairMutation, commitTS, watermark uint64, } for _, mut := range mutations { if mut != nil && len(mut.Key) > 0 { + slog.Warn("one-phase write rejected below applied watermark", "key", string(mut.Key), "commit_ts", commitTS, "last_commit_ts", watermark) return NewWriteConflictError(mut.Key) } } From aa51b35c4f6af0bf700c7c7e2a64518552c2a7ba Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 17:44:09 +0900 Subject: [PATCH 5/8] Initialize filesystem Jepsen registers once --- jepsen/src/elastickv/filesystem_workload.clj | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/jepsen/src/elastickv/filesystem_workload.clj b/jepsen/src/elastickv/filesystem_workload.clj index 970d741da..371079791 100644 --- a/jepsen/src/elastickv/filesystem_workload.clj +++ b/jepsen/src/elastickv/filesystem_workload.clj @@ -528,7 +528,7 @@ (assoc op :type :ok :value (independent/tuple k kind) :error (io-error-label e)) (fail op (io-error-label e)))))) -(defrecord FilesystemClient [node->mount-dir write-mode key-count max-retries require-fuse-mount node mount-dir local?] +(defrecord FilesystemClient [node->mount-dir write-mode key-count max-retries require-fuse-mount setup-done node mount-dir local?] client/Client (open! [this test node] (let [dir (or (get node->mount-dir node) @@ -542,11 +542,16 @@ (let [keys (range key-count)] (when (and local? require-fuse-mount) (require-local-fuse-mount! mount-dir)) - (with-retries max-retries - (fn [] - (if local? - (local-setup! mount-dir keys) - (remote-exec! test node (remote-setup-cmd mount-dir keys))))))) + (if local? + ;; Every local client sees the same mounted filesystem. Initialize its + ;; registers once, before concurrent operations start, so setup does + ;; not race with itself on the same inode and exhaust OCC retries. + (locking setup-done + (when-not @setup-done + (with-retries max-retries #(local-setup! mount-dir keys)) + (reset! setup-done true))) + (with-retries max-retries + #(remote-exec! test node (remote-setup-cmd mount-dir keys)))))) (teardown! [_this test] ;; Jepsen runs teardown! before setup! and again after the test, so the @@ -657,7 +662,7 @@ require-fuse (if (contains? opts :require-fuse-mount) (boolean (:require-fuse-mount opts)) true) node->dir (or (:node->mount-dir opts) (node->mount-dir default-nodes nil default-mount-dir)) - client (->FilesystemClient node->dir write-mode key-count max-retries require-fuse nil nil false)] + client (->FilesystemClient node->dir write-mode key-count max-retries require-fuse (atom false) nil nil false)] (when-not (contains? write-modes write-mode) (throw (ex-info "unknown write mode" {:write-mode write-mode}))) {:client client From 6d874a3744937f4fbc0f76b9e12b4aaf416b94ff Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 17:50:03 +0900 Subject: [PATCH 6/8] Serialize filesystem Jepsen setup writes --- jepsen/src/elastickv/filesystem_workload.clj | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/jepsen/src/elastickv/filesystem_workload.clj b/jepsen/src/elastickv/filesystem_workload.clj index 371079791..83b22926c 100644 --- a/jepsen/src/elastickv/filesystem_workload.clj +++ b/jepsen/src/elastickv/filesystem_workload.clj @@ -528,7 +528,7 @@ (assoc op :type :ok :value (independent/tuple k kind) :error (io-error-label e)) (fail op (io-error-label e)))))) -(defrecord FilesystemClient [node->mount-dir write-mode key-count max-retries require-fuse-mount setup-done node mount-dir local?] +(defrecord FilesystemClient [node->mount-dir write-mode key-count max-retries require-fuse-mount setup-lock node mount-dir local?] client/Client (open! [this test node] (let [dir (or (get node->mount-dir node) @@ -543,13 +543,11 @@ (when (and local? require-fuse-mount) (require-local-fuse-mount! mount-dir)) (if local? - ;; Every local client sees the same mounted filesystem. Initialize its - ;; registers once, before concurrent operations start, so setup does - ;; not race with itself on the same inode and exhaust OCC retries. - (locking setup-done - (when-not @setup-done - (with-retries max-retries #(local-setup! mount-dir keys)) - (reset! setup-done true))) + ;; Every local client sees the same mounted filesystem. Serialize + ;; setup writes to avoid self-conflicts, while still resetting the + ;; registers each time setup is called for a new test. + (locking setup-lock + (with-retries max-retries #(local-setup! mount-dir keys))) (with-retries max-retries #(remote-exec! test node (remote-setup-cmd mount-dir keys)))))) @@ -662,7 +660,7 @@ require-fuse (if (contains? opts :require-fuse-mount) (boolean (:require-fuse-mount opts)) true) node->dir (or (:node->mount-dir opts) (node->mount-dir default-nodes nil default-mount-dir)) - client (->FilesystemClient node->dir write-mode key-count max-retries require-fuse (atom false) nil nil false)] + client (->FilesystemClient node->dir write-mode key-count max-retries require-fuse (Object.) nil nil false)] (when-not (contains? write-modes write-mode) (throw (ex-info "unknown write mode" {:write-mode write-mode}))) {:client client From 78cab1baba23284ee712bdf140cece5bd019f25a Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 17:58:19 +0900 Subject: [PATCH 7/8] Remove temporary Jepsen diagnostics --- internal/filesystem/fuseadapter/adapter.go | 9 +-------- store/lsm_store.go | 1 - 2 files changed, 1 insertion(+), 9 deletions(-) diff --git a/internal/filesystem/fuseadapter/adapter.go b/internal/filesystem/fuseadapter/adapter.go index dea449a92..eec2aa384 100644 --- a/internal/filesystem/fuseadapter/adapter.go +++ b/internal/filesystem/fuseadapter/adapter.go @@ -190,9 +190,6 @@ func (a *Adapter) Write(ctx context.Context, inode uint64, fh uint64, offset uin return 0, errno } n, err := a.core.Write(ctx, inode, fh, offset, data) - if err != nil { - slog.WarnContext(ctx, "filesystem FUSE write failed", "inode", inode, "offset", offset, "size", len(data), "error", err) - } return n, Errno(err) } @@ -280,11 +277,7 @@ func (a *Adapter) StatFS(ctx context.Context, inode uint64) (filesystem.StatFS, } func (a *Adapter) refreshOpenHandleLease(ctx context.Context, inode uint64, fh uint64) syscall.Errno { - err := a.core.RefreshOpenHandleLease(ctx, inode, fh, a.clientID) - if err != nil { - slog.WarnContext(ctx, "filesystem FUSE handle refresh failed", "inode", inode, "fh", fh, "error", err) - } - return Errno(err) + return Errno(a.core.RefreshOpenHandleLease(ctx, inode, fh, a.clientID)) } func (a *Adapter) trackOpenHandle(inode uint64, fh uint64) { diff --git a/store/lsm_store.go b/store/lsm_store.go index b03ce4b1d..5d872a6ee 100644 --- a/store/lsm_store.go +++ b/store/lsm_store.go @@ -2560,7 +2560,6 @@ func checkCommitTSFence(mutations []*KVPairMutation, commitTS, watermark uint64, } for _, mut := range mutations { if mut != nil && len(mut.Key) > 0 { - slog.Warn("one-phase write rejected below applied watermark", "key", string(mut.Key), "commit_ts", commitTS, "last_commit_ts", watermark) return NewWriteConflictError(mut.Key) } } From d23ab8afc53553c6a196896b171fb1166135e563 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 00:51:34 +0900 Subject: [PATCH 8/8] test: reproduce acknowledged DynamoDB append timestamp collisions --- kv/fsm_dynamodb_timestamp_collision_test.go | 74 +++++++++++++++++++++ 1 file changed, 74 insertions(+) create mode 100644 kv/fsm_dynamodb_timestamp_collision_test.go diff --git a/kv/fsm_dynamodb_timestamp_collision_test.go b/kv/fsm_dynamodb_timestamp_collision_test.go new file mode 100644 index 000000000..3c6021a17 --- /dev/null +++ b/kv/fsm_dynamodb_timestamp_collision_test.go @@ -0,0 +1,74 @@ +package kv + +import ( + "bytes" + "context" + "path/filepath" + "testing" + + "github.com/bootjp/elastickv/internal/encryption" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +func TestDynamoAppendDistinctRaftEntryCannotAcknowledgeTimestampCollision(t *testing.T) { + for _, encrypted := range []bool{false, true} { + t.Run(map[bool]string{false: "plaintext", true: "encrypted"}[encrypted], func(t *testing.T) { + ctx := context.Background() + var opts []store.PebbleStoreOption + if encrypted { + keys := encryption.NewKeystore() + require.NoError(t, keys.Set(1, bytes.Repeat([]byte{0x37}, encryption.KeySize))) + cipher, err := encryption.NewCipher(keys) + require.NoError(t, err) + opts = append(opts, store.WithEncryption(cipher, store.NewCounterNonceFactory(1, 1), func() (uint32, bool) { return 1, true })) + } + st, err := store.NewPebbleStore(filepath.Join(t.TempDir(), "fsm.db"), opts...) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM) + require.True(t, ok) + key := []byte(DynamoItemPrefix + "amVwc2VuX2FwcGVuZA|2|16\x00\x01") + // Minimized from the same committed WAL that produced the missing + // acknowledged append37/key16: distinct entries371/372 in term5 + // carried different item values at this identical commit timestamp. + const collisionTS uint64 = 117485124856709124 + first := dynamoCollisionItem(t, "43") + lost := dynamoCollisionItem(t, "37") + fsm.SetApplyIndex(371) + require.NoError(t, applyFSMRequest(t, fsm, onePhaseReq(collisionTS-1, collisionTS, 0, key, first))) + // An actual replay of entry371 remains idempotent. + require.NoError(t, applyFSMRequest(t, fsm, onePhaseReq(collisionTS-1, collisionTS, 0, key, first))) + fsm.SetApplyIndex(372) + err = applyFSMRequest(t, fsm, onePhaseReq(collisionTS-2, collisionTS, 0, key, lost)) + require.ErrorIs(t, err, store.ErrWriteConflict, "another entry must not return success without writing its value") + got, err := st.GetAt(ctx, key, collisionTS) + require.NoError(t, err) + require.Equal(t, first, got) + // Re-read/recompute at a fresh timestamp preserves both appends. + both := dynamoCollisionItem(t, "43", "37") + fsm.SetApplyIndex(373) + require.NoError(t, applyFSMRequest(t, fsm, onePhaseReq(collisionTS, collisionTS+1, 0, key, both))) + got, err = st.GetAt(ctx, key, collisionTS+1) + require.NoError(t, err) + require.Equal(t, both, got) + }) + } +} + +func dynamoCollisionItem(t *testing.T, values ...string) []byte { + t.Helper() + list := &pb.DynamoAttributeValueList{} + for _, value := range values { + list.Values = append(list.Values, &pb.DynamoAttributeValue{Value: &pb.DynamoAttributeValue_N{N: value}}) + } + item := &pb.DynamoItem{Attributes: map[string]*pb.DynamoAttributeValue{ + "pk": {Value: &pb.DynamoAttributeValue_S{S: "16"}}, + "val": {Value: &pb.DynamoAttributeValue_L{L: list}}, + }} + body, err := (proto.MarshalOptions{Deterministic: true}).Marshal(item) + require.NoError(t, err) + return append([]byte{0, 'D', 'I', 1}, body...) +}