diff --git a/Makefile b/Makefile index 9b8c02ace..a7044e51f 100644 --- a/Makefile +++ b/Makefile @@ -1,3 +1,8 @@ +.PHONY: build test + +build: + go build ./... + test: go test -v -race ./... diff --git a/docs/design/2026_05_28_implemented_tla_safety_spec.md b/docs/design/2026_05_28_implemented_tla_safety_spec.md index ed259bf77..1a6e8cf54 100644 --- a/docs/design/2026_05_28_implemented_tla_safety_spec.md +++ b/docs/design/2026_05_28_implemented_tla_safety_spec.md @@ -753,15 +753,14 @@ extra expected-PASS configs. | Module | Was | Now | Configs | |---|---|---|---| | HLC | `BecomeLeader` Observes `maxAppliedHLC` (strategy (c)); the fence was a toggle; logical overflow could issue `ceiling + 1`; leader-only issuance | `IssueTimestamp` is `NextFenced` (no fence before the first ceiling, fail closed at `wall >= ceiling` or when the next value's physical half exceeds the ceiling); no election-time handoff; `IssuanceMode` selects TSO (allocation-floor handoff), legacy `Coordinate` (leader-only, no handoff) or legacy `ShardedCoordinator` (any node stamps) | `MCHLC.cfg` (TSO, pass); `MCHLC_gap.cfg` (TSO without the floor, HLC-4); `MCHLC_legacy_single_gap.cfg` (HLC-4); `MCHLC_legacy_sharded_gap.cfg` (HLC-3, G10) | -| OCC | Fresh `start_ts`; no read-set validation (snapshot isolation); atomic commit | `start_ts` = applied watermark for reading transactions, fresh for write-only ones; one-phase Allocate / Apply with the commit-ts fence, lock check and read + write validation; 2PC with commit-ts settled before PREPARE, PREPARE validating read and write keys, COMMIT without fence or revalidation; raw writes; new `OCC8_NoWriteSkew` | `MCOCC.cfg` (OCC-1/2/4/5, pass); `MCOCC_onephase.cfg` (single shard, one Raft group, every invariant incl. OCC-3 / OCC-8, pass); `MCOCC_gap.cfg` (OCC-1); `MCOCC_g0_gap.cfg` (OCC-3, G0); `MCOCC_g1_gap.cfg` (OCC-8, G1); `MCOCC_emptystore_gap.cfg` (OCC-3b: an empty-store start is MaxUint64, read-latest); `MCOCC_multigroup_gap.cfg` (OCC-3: G0 across groups, `MultiGroup = TRUE`, `ReadAdvance = FALSE`); `MCOCC_multigroup.cfg` (the same with `ReadAdvance = TRUE`, every invariant, pass); `MCOCC_live.cfg` | +| OCC | Fresh `start_ts`; no read-set validation (snapshot isolation); atomic commit | `start_ts` = applied watermark for reading transactions, fresh for write-only ones; one-phase Allocate / Apply with the commit-ts fence, lock check and read + write validation; 2PC with commit-ts settled before PREPARE, PREPARE validating read and write keys, COMMIT without fence or revalidation; raw writes; new `OCC8_NoWriteSkew` | `MCOCC.cfg` (OCC-1/2/4/5, pass); `MCOCC_onephase.cfg` (single shard, one Raft group, `CommitFence = TRUE`, every invariant incl. OCC-3 / OCC-8, pass); `MCOCC_raw_gap.cfg` (the same with `CommitFence = FALSE`: OCC-3, raw entries applying out of timestamp order); `MCOCC_gap.cfg` (OCC-1); `MCOCC_g0_gap.cfg` (OCC-3, G0, `CommitFence = FALSE`); `MCOCC_g0.cfg` (full model with `CommitFence = TRUE`: raw and PREPARE fenced, OCC-1..5 pass); `MCOCC_g1_gap.cfg` (OCC-8, G1); `MCOCC_emptystore_gap.cfg` (OCC-3b: an empty-store start is MaxUint64, read-latest); `MCOCC_multigroup_gap.cfg` (OCC-3: G0 across groups, `MultiGroup = TRUE`, `ReadAdvance = FALSE`); `MCOCC_multigroup.cfg` (the same with `ReadAdvance = TRUE`, every invariant, pass); `MCOCC_live.cfg` | | MVCC | Atomic `Compact` keeping `>= newMin` plus the newest `< newMin`; unique timestamps assumed | Resumable compaction (pending watermark, per-key prune, finish) keeping `> minTS` plus the newest `<= minTS`; reads below the effective watermark fail closed; `UniqueTs = FALSE` admits a colliding write, which the raw apply path writes over the existing version; new `MVCC5_AckedWriteNotOverwritten` | `MCMVCC.cfg` (pass); `MCMVCC_gap.cfg` (MVCC-4); `MCMVCC_collision_gap.cfg` (MVCC-5, G10) | | Routes | Comments claimed the watcher always reads the head snapshot | Comments describe the bounded delta batches of `SyncOnce` and stale-delta rejection; the actions were already general enough | unchanged | | Composed | Every transaction pins its catalog version; Composed-2 justified by "SplitRange is same-group only" | `BeginTxn` chooses pinned or unpinned (`ObservedRouteVersion` absent, so `verifyComposed1` short-circuits), and `AutoPin` pins at the Dispatch-time version (`maybeAutoPinObservedRouteVersion`); Composed-1 asserted for pinned transactions only; Composed-2 documented as not modelled, since cross-group migration has shipped | unchanged | Known abstractions that remain: one PREPARE step per transaction rather than per shard, COMMIT drains every lock at once (no asynchronous -secondary commit, so G12 / G14 are out of scope), the raw write path's -own reordering window, the Composed migration fence / staged-visibility +secondary commit, so G12 / G14 are out of scope), the Composed migration fence / staged-visibility phases, the 2PC COMMIT write-fence bypass, and the single Raft group of the HLC model. The OCC model's `MultiGroup` mode gives every key its own group: the one-phase fence compares against the applying group's @@ -779,4 +778,10 @@ advance's physical half, a second ceiling source next to the lease that HLC.tla's `ApplyCeiling` does not cover. The leader admits an advance only up to its wall clock plus two lease windows, so the ceiling can sit one window further ahead of the wall clock than a lease alone allows; -HLC-2 (ceiling monotonicity) still holds because every raise is a max. +HLC-2 (ceiling monotonicity) still holds because every raise is a max. Raw writes are split into +allocation and apply (`AllocateRaw`, `ApplyRaw` / `RejectRaw`), so the +raw path's reordering window is modelled; `CommitFence` +(`--commitTSFence`) fences raw applies and PREPAREs against the group +watermark. Without it `MCOCC_raw_gap.cfg` and `MCOCC_g0_gap.cfg` fail +OCC-3; with it `MCOCC_onephase.cfg` and `MCOCC_g0.cfg` pass, and so does +`MCOCC_multigroup.cfg` (one-phase and raw across groups). diff --git a/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md index ebc9a09e8..b720af03c 100644 --- a/docs/design/2026_09_26_partial_serializable_isolation_audit.md +++ b/docs/design/2026_09_26_partial_serializable_isolation_audit.md @@ -697,16 +697,76 @@ idempotent but cost log entries and latency); skipping the repeated no-op forward a follower sends on every page of a long scan while its local apply still lags `ts`; and the read-side benchmark. +**Raw and 2PC apply-order fence (branch +`design/serializable-audit-a0-raw-commit-fence`).** Extends the one-phase +commit-ts fence, which is all `main` carries of A0's fence, to the other two +paths G0 names, behind a replicated activation. + +- **Raw.** `handleRawRequest` applies through `ApplyMutationsRaftAtFenced`, + so a raw entry whose timestamp is at or below the group's watermark is + rejected with a write conflict. Raw timestamps are coordinator-allocated + (never caller-chosen), so the fence only fires on reordering; + `ShardedCoordinator` re-dispatches a fenced single-group raw write with a + fresh timestamp (raw writes carry no read set, and nothing was applied). + Single-group eligibility requires both the group set actually dispatched + on that attempt and the current route to remain single-group. The same + rule applies to the existing raw Composed-1 version-GC retry: a partially + successful multi-group write must never be reissued after a route change. + `DEL_PREFIX` stays unfenced: a multi-prefix broadcast shares one + timestamp and would fence itself. +- **2PC: fence the PREPARE, not the COMMIT.** The section above moves the + commit-timestamp allocation after the PREPAREs and fences the primary + COMMIT. This branch keeps the allocation where it is (settled before + PREPARE, carried in `TxnMeta.CommitTS` or, when durable PREPARE evidence + is enabled, `TxnMeta.PrepareCommitTS`) and fences each PREPARE against + the same timestamp used for its route floor and durable lock proof. A + PREPARE whose settled timestamp is at or below its group's + watermark is rejected with a write conflict, and the coordinator aborts + the transaction as for any PREPARE conflict. Once a PREPARE applies, its + write locks (at `start_ts < CommitTS`) are visible, so a reader at + `s >= start_ts` meets the lock and resolves it, and a reader that + completed before the PREPARE read at `s` no higher than the watermark + the PREPARE was checked against, below `CommitTS`. Neither can see the + COMMIT land below its snapshot, so the COMMIT needs no fence and keeps + its idempotent re-apply. With the read watermark advance on, a reader + advances a secondary group before reading, so the same check covers 2PC + across groups. TLC agrees: OCC.tla's `CommitFence` constant adds the + check to `Prepare`; `MCOCC_g0.cfg` (fence on) passes OCC-3 where + `MCOCC_g0_gap.cfg` (fence off) fails it, and with `MultiGroup` and 2PC + the check passes with `ReadAdvance` and fails without it. +- **Activation.** A fence changes the FSM verdict, so a per-node flag would + let replicas of one group disagree. A replicated activation entry (FSM + tag `0x2f`, an invalid protobuf prefix distinct from the guarded bootstrap + control prefix `0x1f`; `0x18` would be a legacy + `Request`'s `ts` tag) persists a marker key per group, reloaded at open and after + a snapshot restore; the FSM fences raw entries and PREPAREs only once the + marker is set. Production startup accepts `--commitTSFence` only with the + protected Raft catalog bootstrap configured. The coordinator waits until + every local protected group reports DataReady before proposing activation + through maintenance admission. Public listeners stay closed until every + local KV data group has applied the marker; DataReady alone does not open + the write path before the renewal worker's first activation tick. The + dedicated TSO group is excluded from that wait: it applies no KV entries, + so its control store never carries the marker. Legacy + startup rejects both the flag and a previously persisted activation marker: + its non-Raft catalog writes could + otherwise advance the watermark on only one replica. Enablement also + requires every replica to understand tag `0x2f`; an older replica fails to + decode the entry and does not fence. +- **Still open:** `DEL_PREFIX`; the 2PC ABORT timestamp; and G1, which the + fence does not touch. + +For multi-group 2PC, a secondary read that finishes before PREPARE raises +that group's replicated watermark to its snapshot timestamp. A PREPARE with +an older settled commit timestamp is then rejected before the coordinator +commits the primary. If PREPARE applies first, the read meets its lock. +`MCOCC_multigroup_2pc.cfg` checks both interleavings with `ReadAdvance` and +`CommitFence`, and `TestMultiGroupPrepareRejectsCommitBelowSecondaryReadWatermark` +checks the rejection and abort on two real Raft groups. This does not cover +deployments that leave read-watermark advancement disabled. + Open after this branch: -- **2PC commit timestamps versus advanced secondaries.** The commit - timestamp is allocated on the primary group's leader, whose HLC and - watermark may sit below an `s` that a secondary group was advanced to; - the secondary COMMIT is unfenced, so it can land at or below `s` on that - group if its PREPARE applied after the read. Fix: the primary allocates - the commit timestamp above every participant's watermark (each PREPARE - reply returns its group's replicated watermark; the allocation goes - through the HLC's after-through path). A merge blocker for A0. - **Non-transactional reads stay plain.** Redis `GET` and the scans, DynamoDB `GetItem` / `Query` / `Scan`, S3 `GET` / `LIST`, and the SQS peek read at the local watermark without the capability. A single-key diff --git a/kv/commit_ts_fence.go b/kv/commit_ts_fence.go new file mode 100644 index 000000000..8c5c847a8 --- /dev/null +++ b/kv/commit_ts_fence.go @@ -0,0 +1,117 @@ +package kv + +import ( + "bytes" + "context" + + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" +) + +// raftEncodeCommitTSFenceActivation turns on the raw and PREPARE apply-order +// fence for the group whose log carries it (docs/design/ +// 2026_09_26_partial_serializable_isolation_audit.md, "Raw and 2PC +// apply-order fence"). The entry has no payload. Its apply persists +// commitTSFenceEnabledKey, so the fence state is a function of the applied +// log and every replica of the group reaches the same verdict on every later +// raw entry and PREPARE. A per-node switch could not give that: replicas +// with different settings would accept and reject the same entry. +// +// 0x2f (field 5, wire type 7) is an invalid protobuf prefix, so it cannot be +// mistaken for a legacy bare Request (whose first byte is often 0x18, field 3 +// "ts" as a varint) or a RaftCommand. 0x1f belongs to guarded bootstrap. +const raftEncodeCommitTSFenceActivation byte = 0x2f + +var ( + commitTSFenceEnabledKey = []byte(txnMetaPrefix + "commit_ts_fence/enabled") + commitTSFenceEnabledValue = []byte{1} +) + +var ErrInvalidCommitTSFenceActivation = errors.New("invalid commit-ts fence activation") + +// EncodeCommitTSFenceActivation returns the Raft payload that turns the raw +// and PREPARE apply-order fence on for a group. +func EncodeCommitTSFenceActivation() []byte { + return []byte{raftEncodeCommitTSFenceActivation} +} + +// CommitTSFenceEnabled reports whether st has applied the fence activation. +func CommitTSFenceEnabled(ctx context.Context, st store.MVCCStore) (bool, error) { + if st == nil { + return false, nil + } + value, err := st.GetAt(ctx, commitTSFenceEnabledKey, ^uint64(0)) + if errors.Is(err, store.ErrKeyNotFound) { + return false, nil + } + if err != nil { + return false, errors.WithStack(err) + } + if !bytes.Equal(value, commitTSFenceEnabledValue) { + return false, errors.Wrap(ErrInvalidCommitTSFenceActivation, "invalid marker value") + } + return true, nil +} + +func (f *kvFSM) reloadCommitTSFence(ctx context.Context) error { + enabled, err := CommitTSFenceEnabled(ctx, f.store) + if err != nil { + return err + } + f.commitTSFenceEnabled.Store(enabled) + return nil +} + +// applyCommitTSFenceActivation persists the marker at the current watermark, +// like the prepare evidence activation, so it writes no version above any +// timestamp a client was handed. A replay finds the marker set and is a +// no-op. A failure to persist halts the apply loop: a replica that skipped +// the activation would keep accepting entries its peers fence. +func (f *kvFSM) applyCommitTSFenceActivation(ctx context.Context, body []byte) any { + if len(body) != 0 { + return errors.Wrapf(ErrInvalidCommitTSFenceActivation, "unexpected %d-byte payload", len(body)) + } + if f.commitTSFenceEnabled.Load() { + return nil + } + stamp := max(uint64(1), f.store.LastCommitTS()) + muts := []*store.KVPairMutation{{Op: store.OpTypePut, Key: commitTSFenceEnabledKey, Value: commitTSFenceEnabledValue}} + if err := f.store.ApplyMutationsRaftAt(ctx, muts, nil, stamp, stamp, f.pendingApplyIdx); err != nil { + return haltErr(errors.Wrap(errors.Mark(err, ErrRaftEntryApply), "kv/fsm: persist commit-ts fence activation")) + } + f.commitTSFenceEnabled.Store(true) + return nil +} + +// checkPrepareCommitTSFence rejects a PREPARE whose settled commit timestamp +// is at or below the group's watermark, once the fence is active. A reader +// that finished before this PREPARE read at a snapshot no higher than that +// watermark; a reader after it meets the PREPARE's locks. Either way the +// COMMIT cannot land below a snapshot that was already read, which is why +// the COMMIT itself is not fenced. A transaction without a settled commit +// timestamp, or a PREPARE that writes nothing, is not checked. +func (f *kvFSM) checkPrepareCommitTSFence(meta TxnMeta, storeMuts []*store.KVPairMutation) error { + commitTS := prepareRouteFloorTimestamp(meta, 0) + if !f.commitTSFenceEnabled.Load() || commitTS == 0 { + return nil + } + if commitTS > f.store.LastCommitTS() { + return nil + } + for _, mut := range storeMuts { + if mut != nil && len(mut.Key) > 0 { + return errors.WithStack(store.NewWriteConflictError(mut.Key)) + } + } + return nil +} + +// applyRawMutations applies a raw entry's mutations, fenced once the +// activation has applied. Raw requests commit against the latest state, so +// commitTS is both the validation snapshot and the commit timestamp. +func (f *kvFSM) applyRawMutations(ctx context.Context, muts []*store.KVPairMutation, commitTS uint64) error { + if f.commitTSFenceEnabled.Load() { + return errors.WithStack(f.store.ApplyMutationsRaftAtFenced(ctx, muts, nil, commitTS, commitTS, f.pendingApplyIdx)) + } + return errors.WithStack(f.store.ApplyMutationsRaftAt(ctx, muts, nil, commitTS, commitTS, f.pendingApplyIdx)) +} diff --git a/kv/commit_ts_fence_test.go b/kv/commit_ts_fence_test.go new file mode 100644 index 000000000..e0ac32391 --- /dev/null +++ b/kv/commit_ts_fence_test.go @@ -0,0 +1,381 @@ +package kv + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/bootjp/elastickv/distribution" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func newCommitTSFenceFSM(t *testing.T) (*kvFSM, store.MVCCStore) { + t.Helper() + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { _ = st.Close() }) + fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM) + require.True(t, ok) + return fsm, st +} + +func rawPut(ts uint64, key, value string) *pb.Request { + return &pb.Request{Ts: ts, Mutations: []*pb.Mutation{{Op: pb.Op_PUT, Key: []byte(key), Value: []byte(value)}}} +} + +func prepareRequest(startTS, commitTS uint64, primary string, keys ...string) *pb.Request { + muts := []*pb.Mutation{{ + Op: pb.Op_PUT, + Key: []byte(txnMetaPrefix), + Value: EncodeTxnMeta(TxnMeta{PrimaryKey: []byte(primary), LockTTLms: defaultTxnLockTTLms, CommitTS: commitTS}), + }} + for _, k := range keys { + muts = append(muts, &pb.Mutation{Op: pb.Op_PUT, Key: []byte(k), Value: []byte("v")}) + } + return &pb.Request{IsTxn: true, Phase: pb.Phase_PREPARE, Ts: startTS, Mutations: muts} +} + +func activateCommitTSFence(t *testing.T, fsm *kvFSM) { + t.Helper() + require.Nil(t, fsm.Apply(EncodeCommitTSFenceActivation())) +} + +// Two raw writes allocated c1 < c2 apply in the opposite order. A reader +// at the watermark c2 misses c1's key, and c1 then lands below that +// snapshot (G0 on the raw path). The default build, without the fence +// activation, still admits it; this pins the gap the activation closes. +func TestRawApplyOutOfOrderLandsBelowSnapshotWithoutFence(t *testing.T) { + t.Parallel() + ctx := context.Background() + fsm, st := newCommitTSFenceFSM(t) + + require.NoError(t, applyFSMRequest(t, fsm, rawPut(20, "a", "x"))) + s := st.LastCommitTS() + _, err := st.GetAt(ctx, []byte("b"), s) + require.ErrorIs(t, err, store.ErrKeyNotFound) + + require.NoError(t, applyFSMRequest(t, fsm, rawPut(10, "b", "late"))) + got, err := st.GetAt(ctx, []byte("b"), s) + require.NoError(t, err) + require.Equal(t, "late", string(got), "without the fence the late raw write lands below the snapshot") +} + +// Once the activation has applied, the late raw write is rejected as a +// write conflict and the snapshot at the watermark stays stable. +func TestRawApplyFencedOnceActivated(t *testing.T) { + t.Parallel() + ctx := context.Background() + fsm, st := newCommitTSFenceFSM(t) + activateCommitTSFence(t, fsm) + + require.NoError(t, applyFSMRequest(t, fsm, rawPut(20, "a", "x"))) + s := st.LastCommitTS() + + err := applyFSMRequest(t, fsm, rawPut(10, "b", "late")) + require.ErrorIs(t, err, store.ErrWriteConflict) + _, err = st.GetAt(ctx, []byte("b"), s) + require.ErrorIs(t, err, store.ErrKeyNotFound, "the fenced raw write must not land below the snapshot") + + // A write above the watermark still applies. + require.NoError(t, applyFSMRequest(t, fsm, rawPut(s+1, "b", "ok"))) +} + +// A PREPARE whose settled commit timestamp is at or below the watermark is +// rejected once the fence is active, and installs no lock; above it, the +// PREPARE applies. +func TestPrepareFencedOnceActivated(t *testing.T) { + t.Parallel() + ctx := context.Background() + fsm, st := newCommitTSFenceFSM(t) + + require.NoError(t, applyFSMRequest(t, fsm, rawPut(20, "a", "x"))) + // Default build: the stale PREPARE applies. + require.NoError(t, applyFSMRequest(t, fsm, prepareRequest(5, 15, "p1", "p1"))) + + activateCommitTSFence(t, fsm) + err := applyFSMRequest(t, fsm, prepareRequest(6, 18, "p2", "p2")) + require.ErrorIs(t, err, store.ErrWriteConflict) + _, err = st.GetAt(ctx, txnLockKey([]byte("p2")), ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound, "a fenced PREPARE must not install a lock") + + require.NoError(t, applyFSMRequest(t, fsm, prepareRequest(7, 25, "p3", "p3"))) +} + +func TestPreparedEvidenceCommitTimestampIsFenced(t *testing.T) { + t.Parallel() + fsm, st := newCommitTSFenceFSM(t) + require.Nil(t, fsm.Apply(EncodePrepareEvidenceFence(0))) + activateCommitTSFence(t, fsm) + require.NoError(t, applyFSMRequest(t, fsm, rawPut(20, "watermark", "v"))) + + for _, hint := range []uint64{0, 999} { + stale := preparedEvidenceRequest(pb.Phase_PREPARE, 18) + meta, _, err := extractTxnMeta(stale.Mutations) + require.NoError(t, err) + meta.CommitTS = hint + stale.Mutations[0].Value = EncodeTxnMeta(meta) + err = applyFSMRequest(t, fsm, stale) + require.ErrorIs(t, err, store.ErrWriteConflict) + _, err = st.GetAt(t.Context(), txnLockKey([]byte("primary")), ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound) + } + + fresh := preparedEvidenceRequest(pb.Phase_PREPARE, 25) + require.NoError(t, applyFSMRequest(t, fsm, fresh)) +} + +func TestMultiGroupPrepareRejectsCommitBelowSecondaryReadWatermark(t *testing.T) { + t.Parallel() + ctx := context.Background() + f := newTwoGroupReadFixture(t) + for _, gid := range []uint64{1, 2} { + _, err := f.groups[gid].Proposer().Propose(ctx, EncodeCommitTSFenceActivation()) + require.NoError(t, err) + require.Eventually(t, func() bool { + enabled, err := CommitTSFenceEnabled(ctx, f.stores[gid]) + return err == nil && enabled + }, 5*time.Second, 10*time.Millisecond) + } + + // A completed read at s=30 on the secondary precedes its PREPARE. + // The primary may have settled commit_ts=20 earlier, but it must not + // commit after the secondary rejects that stale PREPARE. + receipt, err := f.shards.AdvanceReadWatermarkForGroup(ctx, 2, 30) + require.NoError(t, err) + require.GreaterOrEqual(t, receipt.AppliedTimestamp, uint64(30)) + + coord := NewShardedCoordinator(f.shards.engine, f.groups, 1, NewHLC(), f.shards) + grouped := map[uint64][]*pb.Mutation{ + 1: {{Op: pb.Op_PUT, Key: []byte("a-k"), Value: []byte("primary")}}, + 2: {{Op: pb.Op_PUT, Key: []byte("b-k"), Value: []byte("secondary")}}, + } + _, err = coord.dispatchMultiShardTxn(ctx, 10, 20, 0, []byte("a-k"), grouped, []uint64{1, 2}, nil, 0, nil) + require.ErrorIs(t, err, store.ErrWriteConflict) + for gid, key := range map[uint64]string{1: "a-k", 2: "b-k"} { + _, err = f.stores[gid].GetAt(ctx, []byte(key), ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound) + _, err = f.stores[gid].GetAt(ctx, txnLockKey([]byte(key)), ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound) + } +} + +// The activation persists: a new FSM over the same store fences without +// seeing the entry again, and a replay of the entry is a no-op. +func TestCommitTSFenceActivationPersists(t *testing.T) { + t.Parallel() + ctx := context.Background() + fsm, st := newCommitTSFenceFSM(t) + + enabled, err := CommitTSFenceEnabled(ctx, st) + require.NoError(t, err) + require.False(t, enabled) + + activateCommitTSFence(t, fsm) + activateCommitTSFence(t, fsm) + enabled, err = CommitTSFenceEnabled(ctx, st) + require.NoError(t, err) + require.True(t, enabled) + + reopened, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM) + require.True(t, ok) + require.True(t, reopened.commitTSFenceEnabled.Load()) + require.NoError(t, applyFSMRequest(t, reopened, rawPut(20, "a", "x"))) + require.ErrorIs(t, applyFSMRequest(t, reopened, rawPut(10, "b", "late")), store.ErrWriteConflict) +} + +func TestCommitTSFenceActivationRejectsPayload(t *testing.T) { + t.Parallel() + fsm, _ := newCommitTSFenceFSM(t) + resp := fsm.Apply([]byte{raftEncodeCommitTSFenceActivation, 1}) + respErr, ok := resp.(error) + require.True(t, ok) + require.ErrorIs(t, respErr, ErrInvalidCommitTSFenceActivation) + require.False(t, fsm.commitTSFenceEnabled.Load()) +} + +// A replica that cannot persist the activation halts rather than skip it. +func TestCommitTSFenceActivationHaltsOnPersistFailure(t *testing.T) { + t.Parallel() + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { _ = st.Close() }) + fsm, ok := NewKvFSMWithHLC(failingRaftApplyStore{MVCCStore: st}, NewHLC()).(*kvFSM) + require.True(t, ok) + + require.Error(t, haltApplyOf(fsm.Apply(EncodeCommitTSFenceActivation()))) + require.False(t, fsm.commitTSFenceEnabled.Load()) +} + +// The leader of a group proposes the activation when the coordinator is +// built with WithCommitTSFenceActivation, and stops once it has applied. +func TestShardedCoordinatorActivatesCommitTSFence(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { _ = st.Close() }) + r, stop := newSingleRaft(t, "fence", NewKvFSMWithHLC(st, NewHLC())) + t.Cleanup(stop) + + engine := distribution.NewEngine() + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{{RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}}, + })) + groups := map[uint64]*ShardGroup{1: {Engine: r, Store: st, Txn: NewLeaderProxyWithEngine(r)}} + coord := NewShardedCoordinator(engine, groups, 1, NewHLC(), nil).WithCommitTSFenceActivation() + + coord.activateCommitTSFences(ctx) + require.Eventually(t, func() bool { + enabled, err := CommitTSFenceEnabled(ctx, st) + return err == nil && enabled + }, 5*time.Second, 10*time.Millisecond) + + coord.activateCommitTSFences(ctx) + _, done := coord.commitTSFenceActiveGroups.Load(uint64(1)) + require.True(t, done, "an activated group is remembered") +} + +func TestShardedCoordinatorCommitTSFenceWaitsForReadinessAndAdmission(t *testing.T) { + ctx := context.Background() + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { _ = st.Close() }) + r, stop := newSingleRaft(t, "fence-ready", NewKvFSMWithHLC(st, NewHLC())) + t.Cleanup(stop) + + groups := map[uint64]*ShardGroup{1: {Engine: r, Store: st, Txn: NewLeaderProxyWithEngine(r)}} + coord := NewShardedCoordinator(distribution.NewEngine(), groups, 1, NewHLC(), nil). + WithCommitTSFenceActivation().WithCommitTSFenceActivationReady(func() bool { return false }) + coord.activateCommitTSFences(ctx) + enabled, err := CommitTSFenceEnabled(ctx, st) + require.NoError(t, err) + require.False(t, enabled) + + var admitted bool + coord.WithCommitTSFenceActivationReady(func() bool { return true }) + coord.WithOperationAdmission(func(ctx context.Context, group uint64, purpose AdmissionPurpose) (context.Context, func(error) error, error) { + require.Equal(t, uint64(1), group) + require.Equal(t, AdmissionMaintenance, purpose) + admitted = true + return ctx, func(err error) error { return err }, nil + }) + coord.activateCommitTSFences(ctx) + require.True(t, admitted) + require.Eventually(t, func() bool { + enabled, err := CommitTSFenceEnabled(ctx, st) + return err == nil && enabled + }, 5*time.Second, 10*time.Millisecond) +} + +// Without WithCommitTSFenceActivation the coordinator proposes nothing. +func TestShardedCoordinatorCommitTSFenceActivationIsOptIn(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { _ = st.Close() }) + r, stop := newSingleRaft(t, "nofence", NewKvFSMWithHLC(st, NewHLC())) + t.Cleanup(stop) + + coord := NewShardedCoordinator(distribution.NewEngine(), map[uint64]*ShardGroup{ + 1: {Engine: r, Store: st, Txn: NewLeaderProxyWithEngine(r)}, + }, 1, NewHLC(), nil) + require.Nil(t, coord.activateCommitTSFencesAsync(ctx), "an opted-out coordinator starts no activation pass") + time.Sleep(50 * time.Millisecond) + enabled, err := CommitTSFenceEnabled(ctx, st) + require.NoError(t, err) + require.False(t, enabled) +} + +// fencingOnceTransactional rejects the first raw commit as fenced and +// accepts the next, recording the timestamps it saw. +type fencingOnceTransactional struct { + mu sync.Mutex + tss []uint64 +} + +func (s *fencingOnceTransactional) Commit(_ context.Context, reqs []*pb.Request) (*TransactionResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + for _, r := range reqs { + s.tss = append(s.tss, r.GetTs()) + } + if len(s.tss) == 1 { + return nil, store.NewWriteConflictError([]byte("k")) + } + return &TransactionResponse{CommitIndex: uint64(len(s.tss))}, nil +} + +func (s *fencingOnceTransactional) Abort(context.Context, []*pb.Request) (*TransactionResponse, error) { + return &TransactionResponse{}, nil +} + +// A raw write the apply fenced is re-dispatched with a fresh timestamp, and +// the client sees the retry's success. +func TestShardedCoordinatorRetriesFencedRawWrite(t *testing.T) { + t.Parallel() + engine := distribution.NewEngine() + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{{RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}}, + })) + txn := &fencingOnceTransactional{} + coord := NewShardedCoordinator(engine, map[uint64]*ShardGroup{1: {Txn: txn}}, 1, NewHLC(), nil) + + _, err := coord.Dispatch(context.Background(), &OperationGroup[OP]{ + Elems: []*Elem[OP]{{Op: Put, Key: []byte("k"), Value: []byte("v")}}, + }) + require.NoError(t, err) + require.Len(t, txn.tss, 2) + require.Greater(t, txn.tss[1], txn.tss[0], "the retry must carry a fresh, higher timestamp") +} + +// RunHLCLeaseRenewal must not return while an activation pass it started is +// still running: shutdown closes the group stores right after it returns, and +// the pass reads them. A cancelled pass also stops visiting further groups. +func TestRunHLCLeaseRenewalJoinsCommitTSFenceActivation(t *testing.T) { + t.Parallel() + entered := make(chan struct{}) + release := make(chan struct{}) + var calls atomic.Int32 + coord := NewShardedCoordinator(distribution.NewEngine(), map[uint64]*ShardGroup{1: {}}, 1, NewHLC(), nil). + WithCommitTSFenceActivation(). + WithCommitTSFenceActivationReady(func() bool { + // The first check runs on the renewal loop; the second runs + // inside the background pass, which blocks here. + if calls.Add(1) == 2 { + close(entered) + <-release + } + return true + }) + + ctx, cancel := context.WithCancel(context.Background()) + returned := make(chan struct{}) + go func() { + coord.RunHLCLeaseRenewal(ctx) + close(returned) + }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("activation pass did not start") + } + cancel() + select { + case <-returned: + t.Fatal("RunHLCLeaseRenewal returned while the activation pass was still running") + case <-time.After(100 * time.Millisecond): + } + close(release) + select { + case <-returned: + case <-time.After(5 * time.Second): + t.Fatal("RunHLCLeaseRenewal did not return after the activation pass finished") + } +} diff --git a/kv/fsm.go b/kv/fsm.go index 7b4b971a0..01ca556b4 100644 --- a/kv/fsm.go +++ b/kv/fsm.go @@ -131,6 +131,7 @@ type kvFSM struct { replicatedPins *replicatedPinJournal replicatedPinLoadErr error prepareEvidenceEnabled atomic.Bool + commitTSFenceEnabled atomic.Bool prepareCreationSealed atomic.Bool prepareEvidenceReleaseFloor atomic.Uint64 prepareEvidenceTimestampFloor atomic.Uint64 @@ -592,6 +593,8 @@ func (f *kvFSM) applyReservedOpcode(ctx context.Context, data []byte) (any, bool switch { case data[0] == raftEncodeHLCLease: return f.applyHLCLease(data[1:]), true + case data[0] == raftEncodeCommitTSFenceActivation: + return f.applyCommitTSFenceActivation(ctx, data[1:]), true case data[0] == raftEncodeWatermarkAdvance: return f.applyWatermarkAdvance(ctx, data[1:]), true case data[0] >= fsmwire.OpEncryptionMin && data[0] <= fsmwire.OpEncryptionMax: @@ -875,10 +878,8 @@ func (f *kvFSM) handleRawRequest(ctx context.Context, r *pb.Request, commitTS ui if err := f.recordMigrationWrite(ctx, r.Mutations, commitTS); err != nil { return err } - // Raw requests always commit against the latest state; use commitTS as both - // the validation snapshot and the commit timestamp. - if err := f.store.ApplyMutationsRaftAt(ctx, muts, nil, commitTS, commitTS, f.pendingApplyIdx); err != nil { - return errors.WithStack(err) + if err := f.applyRawMutations(ctx, muts, commitTS); err != nil { + return err } f.notifyApplyObservers(commitTS, r.Mutations) return nil @@ -2101,6 +2102,9 @@ func (f *kvFSM) handlePrepareRequest(ctx context.Context, r *pb.Request) error { if err != nil { return err } + if err := f.checkPrepareCommitTSFence(meta, storeMuts); err != nil { + return err + } return f.recordAndApplyPrepare(ctx, uniq, storeMuts, r.ReadKeys, floorTS, startTS) } diff --git a/kv/fsm_backup.go b/kv/fsm_backup.go index cdc86e3c6..1bf646653 100644 --- a/kv/fsm_backup.go +++ b/kv/fsm_backup.go @@ -122,6 +122,9 @@ func (f *kvFSM) reloadBackupTimestampFloor(ctx context.Context) error { if err := f.reloadPrepareEvidence(ctx); err != nil { return err } + if err := f.reloadCommitTSFence(ctx); err != nil { + return err + } value, err := f.store.GetAt(ctx, backupTimestampFloorKey, ^uint64(0)) if err != nil { if errors.Is(err, store.ErrKeyNotFound) { diff --git a/kv/fsm_reserved_opcode_test.go b/kv/fsm_reserved_opcode_test.go index d75de770c..73c5f76d9 100644 --- a/kv/fsm_reserved_opcode_test.go +++ b/kv/fsm_reserved_opcode_test.go @@ -25,6 +25,7 @@ func TestKVFSMReservedOpcodesAreDistinct(t *testing.T) { "read watermark": raftEncodeWatermarkAdvance, "target readiness": raftEncodeTargetReadiness, "replicated backup": raftEncodeReplicatedBackup, + "commit-ts fence": raftEncodeCommitTSFenceActivation, } owner := make(map[byte]string, len(opcodes)) for name, op := range opcodes { @@ -35,4 +36,6 @@ func TestKVFSMReservedOpcodesAreDistinct(t *testing.T) { } owner[op] = name } + require.NotEqual(t, byte(0x1f), raftEncodeCommitTSFenceActivation, + "the guarded bootstrap control prefix must not dispatch as a legacy opcode") } diff --git a/kv/shard_router.go b/kv/shard_router.go index c8b97a867..af0b87981 100644 --- a/kv/shard_router.go +++ b/kv/shard_router.go @@ -167,7 +167,12 @@ func (s *ShardRouter) Register(group uint64, tm Transactional, st store.MVCCStor } func (s *ShardRouter) Commit(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error) { - return s.process(reqs, func(g *routerGroup, rs []*pb.Request) (*TransactionResponse, error) { + resp, _, err := s.commitWithGroupCount(ctx, reqs) + return resp, err +} + +func (s *ShardRouter) commitWithGroupCount(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, int, error) { + return s.processWithGroupCount(reqs, func(g *routerGroup, rs []*pb.Request) (*TransactionResponse, error) { return g.tm.Commit(ctx, rs) }) } @@ -180,20 +185,26 @@ func (s *ShardRouter) Abort(ctx context.Context, reqs []*pb.Request) (*Transacti } func (s *ShardRouter) process(reqs []*pb.Request, fn func(*routerGroup, []*pb.Request) (*TransactionResponse, error)) (*TransactionResponse, error) { + resp, _, err := s.processWithGroupCount(reqs, fn) + return resp, err +} + +func (s *ShardRouter) processWithGroupCount(reqs []*pb.Request, fn func(*routerGroup, []*pb.Request) (*TransactionResponse, error)) (*TransactionResponse, int, error) { if len(reqs) == 0 { - return nil, errors.WithStack(ErrInvalidRequest) + return nil, 0, errors.WithStack(ErrInvalidRequest) } grouped, err := s.groupRequests(reqs) if err != nil { - return nil, errors.WithStack(err) + return nil, 0, errors.WithStack(err) } if err := validateShardBatch(reqs[0], grouped); err != nil { - return nil, errors.WithStack(err) + return nil, 0, errors.WithStack(err) } - return s.processGrouped(grouped, fn) + resp, err := s.processGrouped(grouped, fn) + return resp, len(grouped), err } func validateShardBatch(first *pb.Request, grouped map[uint64][]*pb.Request) error { diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go index aa8f34768..2102e4435 100644 --- a/kv/sharded_coordinator.go +++ b/kv/sharded_coordinator.go @@ -555,7 +555,14 @@ type ShardedCoordinator struct { // hlcRenewalBlocked lets startup code temporarily suppress background HLC // lease proposals while another startup-only Raft mutation must run first. hlcRenewalBlocked func() bool - hlcRecoveryMu sync.Mutex + // commitTSFenceActivation makes the coordinator propose the raw and + // PREPARE apply-order fence activation on groups it leads + // (WithCommitTSFenceActivation). + commitTSFenceActivation bool + commitTSFenceActivationReady func() bool + commitTSFenceActivationInFlight atomic.Bool + commitTSFenceActiveGroups sync.Map + hlcRecoveryMu sync.Mutex // hlcRenewalInFlight prevents a slow or quorum-stalled group from stacking // another background HLC lease proposal for the same group on the next tick. hlcRenewalMu sync.Mutex @@ -1402,22 +1409,23 @@ func isComposed1RetryableError(err error) bool { // dispatchNonTxn routes a non-transactional operation group through the // shard router. Extracted from Dispatch to keep that method's branch // count within the cyclop budget after the 7a registration gate landed. -func (c *ShardedCoordinator) dispatchNonTxn(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, error) { +func (c *ShardedCoordinator) dispatchNonTxn(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, rawDispatchShape, error) { if err := c.rejectWriteFencedPointElems(reqs.Elems); err != nil { - return nil, err + return nil, rawDispatchShape{}, err } if hasExplicitGroupElem(reqs.Elems) { return c.dispatchExplicitGroupNonTxn(ctx, reqs) } logs, err := c.requestLogs(ctx, reqs) if err != nil { - return nil, err + return nil, rawDispatchShape{}, err } - r, err := c.router.Commit(ctx, logs) + r, groupCount, err := c.router.commitWithGroupCount(ctx, logs) + shape := rawDispatchShape{groupCount: groupCount, requestCount: len(logs)} if err != nil { - return nil, errors.WithStack(err) + return nil, shape, errors.WithStack(err) } - return &CoordinateResponse{CommitIndex: r.CommitIndex}, nil + return &CoordinateResponse{CommitIndex: r.CommitIndex}, shape, nil } func hasExplicitGroupElem(elems []*Elem[OP]) bool { @@ -1429,41 +1437,40 @@ func hasExplicitGroupElem(elems []*Elem[OP]) bool { return false } -func (c *ShardedCoordinator) dispatchExplicitGroupNonTxn(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, error) { +func (c *ShardedCoordinator) dispatchExplicitGroupNonTxn(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, rawDispatchShape, error) { logs, gids, err := c.rawLogsWithGroups(ctx, reqs) if err != nil { - return nil, err + return nil, rawDispatchShape{}, err } + shape := rawDispatchShape{groupCount: len(gids), requestCount: len(logs)} var maxIndex uint64 for i, gid := range gids { g, err := c.txnGroupForID(gid) if err != nil { - return nil, err + return nil, shape, err } resp, err := g.Txn.Commit(ctx, []*pb.Request{logs[i]}) if err != nil { - return nil, errors.WithStack(err) + return nil, shape, errors.WithStack(err) } if resp != nil && resp.CommitIndex > maxIndex { maxIndex = resp.CommitIndex } } - return &CoordinateResponse{CommitIndex: maxIndex}, nil + return &CoordinateResponse{CommitIndex: maxIndex}, shape, nil } func (c *ShardedCoordinator) dispatchRawWithComposed1Retry(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, error) { for attempt := 0; attempt <= composed1RetryAttempts; attempt++ { - resp, handled, err := c.dispatchBeforeShardRouting(ctx, reqs) - if !handled { - resp, err = c.dispatchNonTxn(ctx, reqs) - } + resp, shape, err := c.dispatchRawWithFenceRetry(ctx, reqs) if err == nil { return resp, nil } - if !errors.Is(err, ErrComposed1VersionGCd) || attempt == composed1RetryAttempts || c.engine == nil { + if errors.Is(err, ErrTransactionOutcomeUnknown) || + !errors.Is(err, ErrComposed1VersionGCd) || attempt == composed1RetryAttempts || c.engine == nil { return resp, err } - if !c.canRetryRawVersionGC(reqs) { + if !shape.singleRequest() || !c.canRetryRawVersionGC(reqs) { return resp, err } reqs.ObservedRouteVersion = c.engine.Version() @@ -3840,6 +3847,13 @@ func (c *ShardedCoordinator) RunHLCLeaseRenewal(ctx context.Context) { defer children.Done() <-done }() + if activation := c.activateCommitTSFencesAsync(ctx); activation != nil { + children.Add(1) + go func() { + defer children.Done() + <-activation + }() + } timer.Reset(hlcRenewalInterval) case <-ctx.Done(): return diff --git a/kv/sharded_coordinator_commit_ts_fence.go b/kv/sharded_coordinator_commit_ts_fence.go new file mode 100644 index 000000000..ed48a2ec8 --- /dev/null +++ b/kv/sharded_coordinator_commit_ts_fence.go @@ -0,0 +1,152 @@ +package kv + +import ( + "context" + "log/slog" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" +) + +// rawFenceRetryAttempts bounds how often a fenced raw write is re-dispatched +// with a fresh timestamp. A fence rejection means another entry with a later +// timestamp applied first; the fresh timestamp is above it unless that race +// repeats. +const rawFenceRetryAttempts = 3 + +type rawDispatchShape struct { + groupCount int + requestCount int +} + +func (s rawDispatchShape) singleRequest() bool { + return s.groupCount == 1 && s.requestCount == 1 +} + +// WithCommitTSFenceActivation makes this coordinator, while it leads a group, +// propose the raw and PREPARE apply-order fence activation for that group +// until the activation has applied (--commitTSFence). Enable it only once +// every node runs a binary that understands the activation entry. +func (c *ShardedCoordinator) WithCommitTSFenceActivation() *ShardedCoordinator { + if c != nil { + c.commitTSFenceActivation = true + } + return c +} + +// WithCommitTSFenceActivationReady delays proposals until the process has +// loaded its protected catalog and all groups have reached DataReady. +func (c *ShardedCoordinator) WithCommitTSFenceActivationReady(ready func() bool) *ShardedCoordinator { + if c != nil { + c.commitTSFenceActivationReady = ready + } + return c +} + +// activateCommitTSFencesAsync runs one activation pass in the background so +// a stalled group cannot delay the HLC lease renewal that shares the tick. +// At most one pass runs at a time. It returns a channel closed when the pass +// it started finishes, or nil when it started none; the caller must wait on +// it before the group stores the pass reads can be closed. +func (c *ShardedCoordinator) activateCommitTSFencesAsync(ctx context.Context) <-chan struct{} { + if c == nil || !c.commitTSFenceActivation || (c.commitTSFenceActivationReady != nil && !c.commitTSFenceActivationReady()) { + return nil + } + if !c.commitTSFenceActivationInFlight.CompareAndSwap(false, true) { + return nil + } + done := make(chan struct{}) + go func() { + defer close(done) + defer c.commitTSFenceActivationInFlight.Store(false) + c.activateCommitTSFences(ctx) + }() + return done +} + +// activateCommitTSFences proposes the activation on every group this node +// leads whose store has not applied it yet. A group whose store shows the +// marker is remembered and not read again. A cancelled pass stops before the +// next group. +func (c *ShardedCoordinator) activateCommitTSFences(ctx context.Context) { + if c.commitTSFenceActivationReady != nil && !c.commitTSFenceActivationReady() { + return + } + for gid, group := range c.groups { + if ctx.Err() != nil { + return + } + if c.commitTSFenceActivationReady != nil && !c.commitTSFenceActivationReady() { + return + } + c.activateCommitTSFenceForGroup(ctx, gid, group) + } +} + +func (c *ShardedCoordinator) activateCommitTSFenceForGroup(ctx context.Context, gid uint64, group *ShardGroup) { + if _, done := c.commitTSFenceActiveGroups.Load(gid); done { + return + } + if group == nil || group.Engine == nil || group.Store == nil || group.Engine.State() != raftengine.StateLeader { + return + } + enabled, err := CommitTSFenceEnabled(ctx, group.Store) + if err != nil { + c.logger().WarnContext(ctx, "commit-ts fence activation: read marker", slog.Uint64("group_id", gid), slog.Any("err", err)) + return + } + if enabled { + c.commitTSFenceActiveGroups.Store(gid, struct{}{}) + return + } + pctx, cancel := context.WithTimeout(ctx, hlcRenewalProposalTimeout) + // The wrap-aware proposer chain, as for HLC lease entries, so the + // entry follows the group's Raft envelope cutover. + _, err = admittedOperation(pctx, c.operationAdmission, gid, AdmissionMaintenance, func(owned context.Context) (*raftengine.ProposalResult, error) { + return group.Proposer().Propose(owned, EncodeCommitTSFenceActivation()) + }) + cancel() + if err != nil { + c.logger().WarnContext(ctx, "commit-ts fence activation: propose", slog.Uint64("group_id", gid), slog.Any("err", err)) + } +} + +// retryableRawFenceConflict reports whether a raw dispatch that failed with +// err can be re-dispatched with a fresh timestamp. Only a write conflict +// qualifies (a fenced raw entry, or one below a newer version, which is the +// same reordering seen per key); nothing of a rejected raw entry applied, +// and a raw write carries no read set. A dispatch with multiple requests +// is not retried, even if they reached one group, because another request +// may have applied. A route +// write-timestamp floor rejection also reads as a write conflict; it keeps +// its own contract (the caller sees it) and is not retried here. +func (c *ShardedCoordinator) retryableRawFenceConflict(reqs *OperationGroup[OP], shape rawDispatchShape, err error) bool { + return errors.Is(err, store.ErrWriteConflict) && + !errors.Is(err, ErrTransactionOutcomeUnknown) && + !errors.Is(err, ErrRouteWriteTimestampTooLow) && + shape.singleRequest() && c.canRetryRawVersionGC(reqs) +} + +// dispatchRawWithFenceRetry dispatches a raw request, and re-dispatches it +// with a fresh timestamp when the apply rejected it as out of timestamp +// order (retryableRawFenceConflict). +func (c *ShardedCoordinator) dispatchRawWithFenceRetry(ctx context.Context, reqs *OperationGroup[OP]) (*CoordinateResponse, rawDispatchShape, error) { + var ( + resp *CoordinateResponse + shape rawDispatchShape + err error + ) + for attempt := 0; attempt <= rawFenceRetryAttempts; attempt++ { + var handled bool + shape = rawDispatchShape{} + resp, handled, err = c.dispatchBeforeShardRouting(ctx, reqs) + if !handled { + resp, shape, err = c.dispatchNonTxn(ctx, reqs) + } + if err == nil || !c.retryableRawFenceConflict(reqs, shape, err) { + return resp, shape, err + } + } + return resp, shape, err +} diff --git a/kv/sharded_coordinator_raw_retry_groups_test.go b/kv/sharded_coordinator_raw_retry_groups_test.go new file mode 100644 index 000000000..817692461 --- /dev/null +++ b/kv/sharded_coordinator_raw_retry_groups_test.go @@ -0,0 +1,128 @@ +package kv + +import ( + "context" + "errors" + "testing" + + "github.com/bootjp/elastickv/distribution" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func TestRawRetryDoesNotReplayUnknownOutcome(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + err error + }{ + {name: "write conflict", err: errors.Join(ErrTransactionOutcomeUnknown, store.NewWriteConflictError([]byte("k")))}, + {name: "version GC", err: errors.Join(ErrTransactionOutcomeUnknown, ErrComposed1VersionGCd)}, + } { + t.Run(tc.name, func(t *testing.T) { + engine := distribution.NewEngine() + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{ + {RouteID: 1, GroupID: 1, State: distribution.RouteStateActive}, + }, + })) + txn := &recordingTransactional{errs: []error{tc.err}} + coord := NewShardedCoordinator(engine, map[uint64]*ShardGroup{ + 1: {Txn: txn}, + }, 1, NewHLC(), nil) + + _, err := coord.Dispatch(context.Background(), &OperationGroup[OP]{Elems: []*Elem[OP]{ + {Op: Put, Key: []byte("k"), Value: []byte("value")}, + }}) + require.ErrorIs(t, err, ErrTransactionOutcomeUnknown) + require.ErrorIs(t, err, tc.err) + require.Len(t, txn.requests, 1, "an unknown outcome must never be replayed") + }) + } +} + +func TestRawRetryDoesNotResendPartiallyAppliedGroupsAfterRouteChange(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + err error + }{ + {name: "write conflict", err: store.NewWriteConflictError([]byte("z"))}, + {name: "version GC", err: ErrComposed1VersionGCd}, + } { + t.Run(tc.name, func(t *testing.T) { + engine := distribution.NewEngine() + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{ + {RouteID: 1, End: []byte("m"), GroupID: 1, State: distribution.RouteStateActive}, + {RouteID: 2, Start: []byte("m"), GroupID: 2, State: distribution.RouteStateActive}, + }, + })) + + succeeded := &recordingTransactional{} + conflicted := &recordingTransactional{ + errs: []error{tc.err}, + onCommit: func(_ int, _ *pb.Request) { + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 2, + Routes: []distribution.RouteDescriptor{ + {RouteID: 1, GroupID: 1, State: distribution.RouteStateActive}, + }, + })) + }, + } + coord := NewShardedCoordinator(engine, map[uint64]*ShardGroup{ + 1: {Txn: succeeded}, + 2: {Txn: conflicted}, + }, 1, NewHLC(), nil) + _, err := coord.Dispatch(context.Background(), &OperationGroup[OP]{Elems: []*Elem[OP]{ + {Op: Put, Key: []byte("a"), Value: []byte("first")}, + {Op: Put, Key: []byte("z"), Value: []byte("second")}, + }}) + require.ErrorIs(t, err, tc.err) + require.Len(t, succeeded.requests, 1, "a successful group must not be sent the same write again") + require.Len(t, conflicted.requests, 1) + }) + } +} + +func TestRawRetryRequiresSingleAppliedRequestAfterRouteConvergence(t *testing.T) { + t.Parallel() + engine := distribution.NewEngine() + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{ + {RouteID: 1, End: []byte("m"), GroupID: 1, State: distribution.RouteStateActive}, + {RouteID: 2, Start: []byte("m"), GroupID: 2, State: distribution.RouteStateActive}, + }, + })) + coord := NewShardedCoordinator(engine, map[uint64]*ShardGroup{ + 1: {Txn: &recordingTransactional{}}, + 2: {Txn: &recordingTransactional{}}, + }, 1, NewHLC(), nil) + reqs := &OperationGroup[OP]{Elems: []*Elem[OP]{ + {Op: Put, Key: []byte("a"), Value: []byte("first")}, + {Op: Put, Key: []byte("z"), Value: []byte("second")}, + }} + logs, err := coord.requestLogs(context.Background(), reqs) + require.NoError(t, err) + require.Len(t, logs, 2) + + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 2, + Routes: []distribution.RouteDescriptor{ + {RouteID: 1, GroupID: 1, State: distribution.RouteStateActive}, + }, + })) + grouped, err := coord.router.groupRequests(logs) + require.NoError(t, err) + require.Len(t, grouped, 1) + require.Len(t, grouped[1], 2) + shape := rawDispatchShape{groupCount: len(grouped), requestCount: len(logs)} + require.False(t, coord.retryableRawFenceConflict(reqs, shape, store.NewWriteConflictError([]byte("z"))), + "one request can apply while the other conflicts in the same destination group") + require.False(t, shape.singleRequest(), "version-GC retry must obey the same partial-success guard") +} diff --git a/kv/txn_keys.go b/kv/txn_keys.go index b785bdf9a..5dc3f30dd 100644 --- a/kv/txn_keys.go +++ b/kv/txn_keys.go @@ -97,6 +97,7 @@ func isTxnPeerLocalControlKey(key []byte) bool { bytes.Equal(key, migrationConfigurationIndexKey) || bytes.Equal(key, backupTimestampFloorKey) || bytes.Equal(key, prepareEvidenceEnabledKey) || + bytes.Equal(key, commitTSFenceEnabledKey) || bytes.Equal(key, prepareEvidenceGateKey) || bytes.Equal(key, prepareEvidenceTimestampFloorKey) } diff --git a/main.go b/main.go index 3f5df7321..582847006 100644 --- a/main.go +++ b/main.go @@ -149,6 +149,7 @@ var ( tsoBatchSize = flag.Int("tsoBatchSize", defaultTSOBatchSize, "Timestamp batch size used by TSO cutover and shadow validation") tsoModeFile = flag.String("tsoModeFile", "", "Path to a runtime-reloadable TSO mode file containing legacy, shadow, cutover, or phase-d") tsoModeReloadInterval = flag.Duration("tsoModeReloadInterval", defaultTSOReload, "Polling interval for tsoModeFile") + commitTSFence = flag.Bool("commitTSFence", false, "Reserved for replicated catalog bootstrap; activation is rejected until the startup catalog no longer writes outside Raft") readWatermarkAdvance = flag.Bool("readWatermarkAdvance", false, "Raise a lagging group's commit watermark before a cross-group snapshot read so late one-phase writes cannot land below the snapshot; enable only after every node runs a binary that understands the advance entry") leaderBalance = flag.Bool("leaderBalance", false, "Enable automatic count-based Raft-group leader balancing on the default-group leader") leaderBalanceInterval = flag.Duration("leaderBalanceInterval", defaultLeaderBalanceInterval, "Interval between leader-balance scheduler evaluations") @@ -645,6 +646,7 @@ func run() error { WithKeyVizLabelsEnabled(*keyVizFlags.LabelsEnabled). WithAllShardGroups(dataGroupIDs(cfg.groups)...). WithPartitionResolver(sqsPartitionResolver) + configureCommitTSFenceActivation(coordinate, *commitTSFence, runtimes) installBootstrapCoordinator(bootstrapDomain, coordinate, &cleanup) tsoWiring, err := configureCoordinatorTSOWithObserver( coordinate, @@ -770,6 +772,14 @@ func newStartupShardStore(engine *distribution.Engine, groups map[uint64]*kv.Sha return shardStore } +func configureCommitTSFenceActivation(coordinate *kv.ShardedCoordinator, enabled bool, runtimes []*raftGroupRuntime) { + if enabled { + coordinate.WithCommitTSFenceActivation().WithCommitTSFenceActivationReady(func() bool { + return commitTSFenceRuntimesReady(runtimes) + }) + } +} + func waitRunGroup(eg *errgroup.Group) error { if err := eg.Wait(); err != nil { return errors.Wrapf(err, "failed to serve") @@ -793,6 +803,9 @@ func startTSOModeReloader(ctx context.Context, eg *errgroup.Group, wiring coordi } func resolveValidatedRuntimeInputs() (runtimeConfig, raftEngineType, raftBootstrapConfig, bool, error) { + if err := validateCommitTSFenceStartup(*commitTSFence, *raftBootstrapPeerConfigDir != ""); err != nil { + return runtimeConfig{}, "", raftBootstrapConfig{}, false, err + } cfg, engineType, bootstrapCfg, bootstrap, err := resolveRuntimeInputs() if err != nil { return runtimeConfig{}, "", raftBootstrapConfig{}, false, err @@ -804,6 +817,13 @@ func resolveValidatedRuntimeInputs() (runtimeConfig, raftEngineType, raftBootstr return cfg, engineType, bootstrapCfg, bootstrap, err } +func validateCommitTSFenceStartup(enabled, protected bool) error { + if enabled && !protected { + return errors.New("commitTSFence requires protected Raft catalog bootstrap; legacy startup catalog writes outside Raft") + } + return nil +} + func validateLiveBackupConfig( defaultTTL time.Duration, maxTTL time.Duration, @@ -2990,8 +3010,14 @@ func startServersAfterStartupRotation(waitRotateOnStartup startupRotationWaiter, if err := waitRotateOnStartup.Wait(in.ctx); err != nil { return runner.startupFailure(errors.Wrap(err, "encryption rotate-on-startup: wait before serving")) } + return startPublicServicesAfterCommitTSFence(in, &runner, publicKVGate, *commitTSFence) +} + +func startPublicServicesAfterCommitTSFence(in serversInput, runner *runtimeServerRunner, gate *startupPublicKVGate, fenceEnabled bool) error { startHLCLeaseRenewal(in.ctx, in.eg, in.coordinate) - publicKVGate.markReady() + if err := markPublicGateReadyAfterCommitTSFence(in.ctx, gate, in.runtimes, fenceEnabled); err != nil { + return runner.startupFailure(errors.Wrap(err, "wait for commit-ts fence activation before public service")) + } startDistributionSplitJobRunner(in.ctx, in.eg, in.distServer) startMigrationOwnershipRefresh(in.ctx, in.eg, in.distServer) if err := runner.startPublicServices(); err != nil { @@ -4694,6 +4720,9 @@ func setupDistributionCatalog( runtimes []*raftGroupRuntime, engine *distribution.Engine, ) (*distribution.CatalogStore, error) { + if err := validateNoExistingCommitTSFenceActivation(ctx, runtimes, bootstrapDomainForRuntimes(runtimes) != nil); err != nil { + return nil, err + } catalogGroupID, err := distributionCatalogGroupID(engine) if err != nil { return nil, errors.Wrapf(err, "resolve distribution catalog group") @@ -4728,6 +4757,22 @@ func setupDistributionCatalog( return distCatalog, nil } +func validateNoExistingCommitTSFenceActivation(ctx context.Context, runtimes []*raftGroupRuntime, protected bool) error { + for _, rt := range runtimes { + if rt == nil || rt.store == nil { + continue + } + active, err := kv.CommitTSFenceEnabled(ctx, rt.store) + if err != nil { + return errors.Wrapf(err, "read commit-ts fence activation for group %d", rt.spec.id) + } + if active && !protected { + return errors.WithStack(errors.Newf("group %d has an active commit-ts fence; protected Raft catalog bootstrap is required before startup", rt.spec.id)) + } + } + return nil +} + func distributionCatalogGroupID(engine *distribution.Engine) (uint64, error) { if engine == nil { return 0, errors.New("distribution engine is required") diff --git a/main_bootstrap_payload.go b/main_bootstrap_payload.go index 3ffd22061..f580d3be7 100644 --- a/main_bootstrap_payload.go +++ b/main_bootstrap_payload.go @@ -160,6 +160,9 @@ func validateBootstrapCatalogPayload(payload []byte) error { } func validateBootstrapDataPayload(payload []byte) error { + if bytes.Equal(payload, kv.EncodeCommitTSFenceActivation()) { + return nil + } if _, matched, err := kv.DecodeWatermarkAdvance(payload); matched { return errors.Wrap(err, "decode data read watermark") } diff --git a/main_bootstrap_payload_test.go b/main_bootstrap_payload_test.go index 100c4c006..91348b111 100644 --- a/main_bootstrap_payload_test.go +++ b/main_bootstrap_payload_test.go @@ -85,6 +85,7 @@ func bootstrapPayloadFixtures(t testing.TB) []bootstrapPayloadCase { {"single", bootstrapPayloadProto(t, bootstrapPayloadSingle, bootstrapPayloadRequest()), data}, {"batch", bootstrapPayloadProto(t, bootstrapPayloadBatch, &pb.RaftCommand{Requests: []*pb.Request{bootstrapPayloadRequest(), bootstrapPayloadRequest()}}), data}, {"watermark", kv.EncodeWatermarkAdvance(42), data}, + {"commit timestamp fence activation", kv.EncodeCommitTSFenceActivation(), data}, {"pin", kv.EncodeBackupPinEntry(pin), data}, {"reserve", kv.EncodeBackupReserveEntry(pin), data}, {"extend", kv.EncodeBackupExtendEntry(kv.BackupExtendEntry{PinID: pin.PinID, Deadline: pin.Deadline}), data}, diff --git a/main_commit_ts_fence_readiness.go b/main_commit_ts_fence_readiness.go new file mode 100644 index 000000000..f7fd52ca6 --- /dev/null +++ b/main_commit_ts_fence_readiness.go @@ -0,0 +1,84 @@ +package main + +import ( + "context" + "time" + + "github.com/bootjp/elastickv/kv" + "github.com/cockroachdb/errors" +) + +// Activation must not infer readiness from a missing runtime callback. The +// legacy public gate permits that shape, but a fence changes Raft apply verdicts. +func commitTSFenceRuntimesReady(runtimes []*raftGroupRuntime) bool { + if len(runtimes) == 0 { + return false + } + domain := bootstrapDomainForRuntimes(runtimes) + if domain == nil { + return false + } + for _, runtime := range runtimes { + if runtime == nil || runtime.bootstrapDomain != domain || runtime.publicReady == nil || !runtime.publicReady() { + return false + } + } + return true +} + +// Keep public listeners closed until every local replica has applied the +// activation entry. The renewal worker may propose it after DataReady, but +// DataReady alone does not imply that the fence is active yet. +func markPublicGateReadyAfterCommitTSFence(ctx context.Context, gate *startupPublicKVGate, runtimes []*raftGroupRuntime, enabled bool) error { + if enabled { + if err := waitForCommitTSFenceActivation(ctx, runtimes); err != nil { + return err + } + } + gate.markReady() + return nil +} + +func waitForCommitTSFenceActivation(ctx context.Context, runtimes []*raftGroupRuntime) error { + ticker := time.NewTicker(bootstrapReadinessRetryInterval) + defer ticker.Stop() + for { + active, err := commitTSFenceRuntimesActive(ctx, runtimes) + if err != nil { + return err + } + if active { + return nil + } + select { + case <-ctx.Done(): + return errors.WithStack(ctx.Err()) + case <-ticker.C: + } + } +} + +// Only KV data groups carry the activation marker. The dedicated TSO group's +// runtime store is its control store and its ShardGroup has no KV Store, so +// the coordinator never proposes the activation there; waiting on it would +// keep public traffic closed forever. At least one data group is required. +func commitTSFenceRuntimesActive(ctx context.Context, runtimes []*raftGroupRuntime) (bool, error) { + dataGroups := 0 + for _, runtime := range runtimes { + if runtime == nil || runtime.store == nil { + return false, nil + } + if runtime.spec.id == dedicatedTSORaftGroupID { + continue + } + dataGroups++ + active, err := kv.CommitTSFenceEnabled(ctx, runtime.store) + if err != nil { + return false, errors.Wrapf(err, "read commit-ts fence activation for group %d", runtime.spec.id) + } + if !active { + return false, nil + } + } + return dataGroups > 0, nil +} diff --git a/main_commit_ts_fence_readiness_test.go b/main_commit_ts_fence_readiness_test.go new file mode 100644 index 000000000..ea02bcc3e --- /dev/null +++ b/main_commit_ts_fence_readiness_test.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func TestCommitTSFenceRuntimesReady(t *testing.T) { + domain := &bootstrapDomain{} + ready := func() bool { return true } + blocked := func() bool { return false } + cases := []struct { + name string + runtimes []*raftGroupRuntime + want bool + }{ + {name: "no runtimes"}, + {name: "nil runtime", runtimes: []*raftGroupRuntime{nil}}, + {name: "legacy runtime", runtimes: []*raftGroupRuntime{{publicReady: ready}}}, + {name: "missing callback", runtimes: []*raftGroupRuntime{{bootstrapDomain: domain}}}, + {name: "not ready", runtimes: []*raftGroupRuntime{{bootstrapDomain: domain, publicReady: blocked}}}, + {name: "mixed domains", runtimes: []*raftGroupRuntime{{bootstrapDomain: domain, publicReady: ready}, {bootstrapDomain: &bootstrapDomain{}, publicReady: ready}}}, + {name: "missing group", runtimes: []*raftGroupRuntime{{bootstrapDomain: domain, publicReady: ready}, nil}}, + {name: "all protected groups ready", runtimes: []*raftGroupRuntime{{bootstrapDomain: domain, publicReady: ready}, {bootstrapDomain: domain, publicReady: ready}}, want: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, commitTSFenceRuntimesReady(tc.runtimes)) + }) + } +} + +func TestPublicGateWaitsForEveryCommitTSFenceActivation(t *testing.T) { + runtimes := make([]*raftGroupRuntime, 2) + fsms := make([]kv.FSM, 2) + for i := range runtimes { + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + runtimes[i] = &raftGroupRuntime{spec: groupSpec{id: uint64(i + 1)}, store: st} + fsms[i] = kv.NewKvFSMWithHLC(st, kv.NewHLC()) + } + gate := &startupPublicKVGate{} + active, err := commitTSFenceRuntimesActive(t.Context(), runtimes) + require.NoError(t, err) + require.False(t, active) + require.Nil(t, fsms[0].Apply(kv.EncodeCommitTSFenceActivation())) + active, err = commitTSFenceRuntimesActive(t.Context(), runtimes) + require.NoError(t, err) + require.False(t, active, "one active group cannot open public traffic") + + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- markPublicGateReadyAfterCommitTSFence(ctx, gate, runtimes, true) }() + select { + case err := <-done: + t.Fatalf("gate opened before all groups activated: %v", err) + case <-time.After(2 * bootstrapReadinessRetryInterval): + } + require.True(t, gate.blocked()) + require.Nil(t, fsms[1].Apply(kv.EncodeCommitTSFenceActivation())) + require.NoError(t, <-done) + require.False(t, gate.blocked()) +} + +func TestCommitTSFenceActivationWaitFailsClosed(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + require.ErrorIs(t, waitForCommitTSFenceActivation(ctx, nil), context.Canceled) + require.ErrorIs(t, waitForCommitTSFenceActivation(ctx, []*raftGroupRuntime{{}}), context.Canceled) + gate := &startupPublicKVGate{} + require.ErrorIs(t, markPublicGateReadyAfterCommitTSFence(ctx, gate, nil, true), context.Canceled) + require.True(t, gate.blocked()) + require.NoError(t, markPublicGateReadyAfterCommitTSFence(ctx, gate, nil, false)) + require.False(t, gate.blocked()) +} + +// The dedicated TSO group never applies the KV activation entry: its runtime +// store is the TSO control store and its ShardGroup has no KV Store, so the +// coordinator never proposes the marker there. Readiness must only wait for +// the KV data groups, but it must still require at least one of them. +func TestCommitTSFenceReadinessIgnoresDedicatedTSOGroup(t *testing.T) { + newRuntime := func(id uint64) (*raftGroupRuntime, kv.FSM) { + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + return &raftGroupRuntime{spec: groupSpec{id: id}, store: st}, kv.NewKvFSMWithHLC(st, kv.NewHLC()) + } + tso, _ := newRuntime(dedicatedTSORaftGroupID) + data, dataFSM := newRuntime(1) + + active, err := commitTSFenceRuntimesActive(t.Context(), []*raftGroupRuntime{tso}) + require.NoError(t, err) + require.False(t, active, "a TSO-only node has no KV data group to fence") + + runtimes := []*raftGroupRuntime{tso, data} + active, err = commitTSFenceRuntimesActive(t.Context(), runtimes) + require.NoError(t, err) + require.False(t, active) + + require.Nil(t, dataFSM.Apply(kv.EncodeCommitTSFenceActivation())) + active, err = commitTSFenceRuntimesActive(t.Context(), runtimes) + require.NoError(t, err) + require.True(t, active, "the TSO control store never carries the KV marker") + + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + gate := &startupPublicKVGate{} + require.NoError(t, markPublicGateReadyAfterCommitTSFence(ctx, gate, runtimes, true)) + require.False(t, gate.blocked()) +} diff --git a/main_commit_ts_fence_test.go b/main_commit_ts_fence_test.go new file mode 100644 index 000000000..bc0c433d6 --- /dev/null +++ b/main_commit_ts_fence_test.go @@ -0,0 +1,44 @@ +package main + +import ( + "testing" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func TestCommitTSFenceStartupRequiresProtectedCatalogBootstrap(t *testing.T) { + require.NoError(t, validateCommitTSFenceStartup(false, false)) + require.NoError(t, validateCommitTSFenceStartup(false, true)) + require.NoError(t, validateCommitTSFenceStartup(true, true)) + require.ErrorContains(t, validateCommitTSFenceStartup(true, false), "requires protected Raft catalog bootstrap") +} + +func TestRuntimeInputsRejectCommitTSFenceBeforeOpeningStores(t *testing.T) { + previous := *commitTSFence + *commitTSFence = true + t.Cleanup(func() { *commitTSFence = previous }) + + _, _, _, _, err := resolveValidatedRuntimeInputs() + require.ErrorContains(t, err, "requires protected Raft catalog bootstrap") +} + +func TestCatalogSetupRejectsPreviouslyActivatedCommitTSFence(t *testing.T) { + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + fsm := kv.NewKvFSMWithHLC(st, kv.NewHLC()) + require.Nil(t, fsm.Apply(kv.EncodeCommitTSFenceActivation())) + + engine := distribution.NewEngine() + engine.UpdateRoute(nil, nil, 1) + _, err = setupDistributionCatalog(t.Context(), []*raftGroupRuntime{{spec: groupSpec{id: 1}, store: st}}, engine) + require.ErrorContains(t, err, "active commit-ts fence") + + version, err := distribution.NewCatalogStore(st).Version(t.Context()) + require.NoError(t, err) + require.Zero(t, version) + require.NoError(t, validateNoExistingCommitTSFenceActivation(t.Context(), []*raftGroupRuntime{{spec: groupSpec{id: 1}, store: st}}, true)) +} diff --git a/scripts/tla-check.sh b/scripts/tla-check.sh index 9e4ee69b2..5411d4cb6 100755 --- a/scripts/tla-check.sh +++ b/scripts/tla-check.sh @@ -56,6 +56,7 @@ gap_configs_for() { printf '_g1_gap|Invariant OCC8_NoWriteSkew is violated\n' printf '_emptystore_gap|Invariant OCC3b_EmptyStoreReadStability is violated\n' printf '_multigroup_gap|Invariant OCC3_ReadSnapshotStability is violated\n' + printf '_raw_gap|Invariant OCC3_ReadSnapshotStability is violated\n' ;; mvcc) printf '_gap|Invariant MVCC4_NoLostCommitOnSnapshotInstall is violated\n' @@ -80,7 +81,7 @@ gap_configs_for() { # because the full model (with 2PC) pins known OCC-3 / OCC-8 gaps. pass_configs_for() { case "$1" in - occ) printf '_onephase\n_multigroup\n' ;; + occ) printf '_onephase\n_multigroup\n_multigroup_2pc\n_g0\n' ;; hlc|mvcc|routes|composed) ;; *) echo "ERROR: no pass-config list registered for module '$1'." \ diff --git a/tla/README.md b/tla/README.md index c567696f9..450f8cbcc 100644 --- a/tla/README.md +++ b/tla/README.md @@ -162,10 +162,12 @@ take the one-phase path: `AllocateOnePhase` picks `commit_ts`, and write or read key has a version newer than `start_ts`; otherwise `RejectOnePhase`. Multi-shard transactions settle `commit_ts` first, then `Prepare` validates read and write keys and locks the write keys, -and `Commit` applies with no fence and no revalidation. `RawWrite` -is a non-transactional write that raises the watermark (refused on a -locked key); `Prepare` raises the watermark to `start_ts`, as the lock -rows' apply does. The lock +and `Commit` applies with no fence and no revalidation. A raw write is +allocated (`AllocateRaw`) and applied later in any order (`ApplyRaw` / +`RejectRaw`; refused on a locked key or below a newer version); +`CommitFence` (`--commitTSFence`) fences raw applies and PREPAREs +against the group watermark. `Prepare` raises the watermark to +`start_ts`, as the lock rows' apply does. The lock resolver is abstracted into the atomic lock drain inside `Commit`/`Abort`; timestamps come from one global counter (unique, leader-issued). `EnableSafety` gates the commit-ts > start-ts @@ -195,7 +197,9 @@ Invariants asserted: | Config | Checks | Expected | |---|---|---| | `MCOCC.cfg` | full model; OCC-1, OCC-2, OCC-4, OCC-5 | PASS | -| `MCOCC_onephase.cfg` | single-shard path only, one Raft group; every invariant incl. OCC-3, OCC-8 | PASS | +| `MCOCC_onephase.cfg` | single-shard path only, one Raft group, `CommitFence = TRUE`; every invariant incl. OCC-3, OCC-8 | PASS | +| `MCOCC_raw_gap.cfg` | the same with `CommitFence = FALSE` (known gap: raw writes unfenced in the default build) | FAIL on `OCC3_ReadSnapshotStability` (two raw entries apply out of timestamp order) | +| `MCOCC_g0.cfg` | full model, `CommitFence = TRUE` (raw and PREPARE fenced); OCC-1..5 | PASS | | `MCOCC_gap.cfg` | `EnableSafety = FALSE` | FAIL on `OCC1_CommitTsAboveStart` (depth ≈ 5) | | `MCOCC_g0_gap.cfg` | full model (known gap G0) | FAIL on `OCC3_ReadSnapshotStability` (depth 9: 2PC allocates, a raw write raises the watermark, a reader reads, the 2PC commits below its snapshot) | | `MCOCC_g1_gap.cfg` | full model (known gap G1) | FAIL on `OCC8_NoWriteSkew` (a one-phase commit applies between the other txn's PREPARE and COMMIT) | diff --git a/tla/occ/MCOCC.cfg b/tla/occ/MCOCC.cfg index 2aa944589..2e959dbb8 100644 --- a/tla/occ/MCOCC.cfg +++ b/tla/occ/MCOCC.cfg @@ -17,6 +17,7 @@ CONSTANTS EnableTwoPhase = TRUE MultiGroup = FALSE ReadAdvance = FALSE + CommitFence = FALSE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC.tla b/tla/occ/MCOCC.tla index dbdbbcad6..f11658060 100644 --- a/tla/occ/MCOCC.tla +++ b/tla/occ/MCOCC.tla @@ -17,6 +17,10 @@ (* one-phase path (TLC fails on OCC-3) *) (* MCOCC_multigroup.cfg — the same with ReadAdvance (read watermark *) (* advance enabled; TLC passes everything) *) +(* MCOCC_g0.cfg — full model with CommitFence (raw and PREPARE *) +(* fenced; TLC passes OCC-3) *) +(* MCOCC_raw_gap.cfg — known gap: raw writes unfenced in the default *) +(* build (TLC fails on OCC-3) *) (* MCOCC_live.cfg — OCC-L1 liveness under SpecLive (TLC passes) *) (* *) (* Run via `make tla-check` from the repository root. *) diff --git a/tla/occ/MCOCC_emptystore_gap.cfg b/tla/occ/MCOCC_emptystore_gap.cfg index 2a1e8a575..efe39fca8 100644 --- a/tla/occ/MCOCC_emptystore_gap.cfg +++ b/tla/occ/MCOCC_emptystore_gap.cfg @@ -20,6 +20,7 @@ CONSTANTS EnableTwoPhase = FALSE MultiGroup = FALSE ReadAdvance = FALSE + CommitFence = FALSE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_g0.cfg b/tla/occ/MCOCC_g0.cfg new file mode 100644 index 000000000..4b4de0d88 --- /dev/null +++ b/tla/occ/MCOCC_g0.cfg @@ -0,0 +1,41 @@ +\* TLC config: G0 closed by the apply-order fence (--commitTSFence; +\* docs/design/2026_09_26_partial_serializable_isolation_audit.md, "Raw +\* and 2PC apply-order fence"). The full model (one-phase, 2PC, raw), one +\* Raft group, CommitFence = TRUE: a PREPARE whose settled commit_ts is at +\* or below the group watermark is rejected, and so is a raw entry at or +\* below it. Expected outcome: TLC passes every invariant listed, including +\* OCC-3. OCC-8 is not listed: G1 is independent of the fence and stays +\* pinned by MCOCC_g1_gap.cfg. + +SPECIFICATION Spec + +CONSTANTS + Keys = {k1, k2} + TxnIds = {t1, t2} + MaxTs = 6 + MaxOps = 3 + EnableSafety = TRUE + EnableTwoPhase = TRUE + MultiGroup = FALSE + ReadAdvance = FALSE + CommitFence = TRUE + +SYMMETRY Symmetry + +INVARIANTS + TypeOK + OCC1_CommitTsAboveStart + OCC2_NoWriteWriteConflict + OCC3_ReadSnapshotStability + OCC4_NoStrandedLockAtQuiescence + OCC5_StartTsConsistency + +PROPERTIES + OCC5_Action + CommitTsAssignedOnce + +CONSTRAINT StateConstraint + +\* Quiescence is a valid end state, not a deadlock. +CHECK_DEADLOCK + FALSE diff --git a/tla/occ/MCOCC_g0_gap.cfg b/tla/occ/MCOCC_g0_gap.cfg index 08962643d..f9b9ab8eb 100644 --- a/tla/occ/MCOCC_g0_gap.cfg +++ b/tla/occ/MCOCC_g0_gap.cfg @@ -6,9 +6,8 @@ \* below its snapshot. Expected outcome: TLC FAILS \* OCC3_ReadSnapshotStability. \* -\* Kept as an expected failure so that the fix (A0) turns this run into -\* "unexpectedly passed" and forces the model and this config to be -\* updated. +\* CommitFence = FALSE: the default build (--commitTSFence off). The +\* fenced build is checked by MCOCC_g0.cfg, which passes OCC-3. SPECIFICATION Spec @@ -21,6 +20,7 @@ CONSTANTS EnableTwoPhase = TRUE MultiGroup = FALSE ReadAdvance = FALSE + CommitFence = FALSE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_g1_gap.cfg b/tla/occ/MCOCC_g1_gap.cfg index bb170cf89..888889d2f 100644 --- a/tla/occ/MCOCC_g1_gap.cfg +++ b/tla/occ/MCOCC_g1_gap.cfg @@ -20,6 +20,7 @@ CONSTANTS EnableTwoPhase = TRUE MultiGroup = FALSE ReadAdvance = FALSE + CommitFence = FALSE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_gap.cfg b/tla/occ/MCOCC_gap.cfg index a3c7f145f..4ed2c1651 100644 --- a/tla/occ/MCOCC_gap.cfg +++ b/tla/occ/MCOCC_gap.cfg @@ -21,6 +21,7 @@ CONSTANTS EnableTwoPhase = TRUE MultiGroup = FALSE ReadAdvance = FALSE + CommitFence = FALSE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_live.cfg b/tla/occ/MCOCC_live.cfg index 4d6fa92c9..fee2a4974 100644 --- a/tla/occ/MCOCC_live.cfg +++ b/tla/occ/MCOCC_live.cfg @@ -21,6 +21,7 @@ CONSTANTS EnableTwoPhase = TRUE MultiGroup = FALSE ReadAdvance = FALSE + CommitFence = FALSE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_multigroup.cfg b/tla/occ/MCOCC_multigroup.cfg index fde88f7d3..b94851cbd 100644 --- a/tla/occ/MCOCC_multigroup.cfg +++ b/tla/occ/MCOCC_multigroup.cfg @@ -17,6 +17,7 @@ CONSTANTS EnableTwoPhase = FALSE MultiGroup = TRUE ReadAdvance = TRUE + CommitFence = TRUE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_multigroup_2pc.cfg b/tla/occ/MCOCC_multigroup_2pc.cfg new file mode 100644 index 000000000..0e1a4b8ce --- /dev/null +++ b/tla/occ/MCOCC_multigroup_2pc.cfg @@ -0,0 +1,34 @@ +\* Multi-group 2PC with per-group read advancement and the PREPARE fence. +\* Checks whether a secondary can commit below a completed read snapshot. + +SPECIFICATION Spec + +CONSTANTS + Keys = {k1, k2} + TxnIds = {t1, t2} + MaxTs = 6 + MaxOps = 3 + EnableSafety = TRUE + EnableTwoPhase = TRUE + MultiGroup = TRUE + ReadAdvance = TRUE + CommitFence = TRUE + +SYMMETRY Symmetry + +INVARIANTS + TypeOK + OCC1_CommitTsAboveStart + OCC2_NoWriteWriteConflict + OCC3_ReadSnapshotStability + OCC4_NoStrandedLockAtQuiescence + OCC5_StartTsConsistency + +PROPERTIES + OCC5_Action + CommitTsAssignedOnce + +CONSTRAINT StateConstraint + +CHECK_DEADLOCK + FALSE diff --git a/tla/occ/MCOCC_multigroup_gap.cfg b/tla/occ/MCOCC_multigroup_gap.cfg index a7102c535..b8929edba 100644 --- a/tla/occ/MCOCC_multigroup_gap.cfg +++ b/tla/occ/MCOCC_multigroup_gap.cfg @@ -21,6 +21,7 @@ CONSTANTS EnableTwoPhase = FALSE MultiGroup = TRUE ReadAdvance = FALSE + CommitFence = TRUE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_onephase.cfg b/tla/occ/MCOCC_onephase.cfg index 76634b1fb..8adc0e0a5 100644 --- a/tla/occ/MCOCC_onephase.cfg +++ b/tla/occ/MCOCC_onephase.cfg @@ -17,6 +17,7 @@ CONSTANTS EnableTwoPhase = FALSE MultiGroup = FALSE ReadAdvance = FALSE + CommitFence = TRUE SYMMETRY Symmetry diff --git a/tla/occ/MCOCC_raw_gap.cfg b/tla/occ/MCOCC_raw_gap.cfg new file mode 100644 index 000000000..5b6c3d607 --- /dev/null +++ b/tla/occ/MCOCC_raw_gap.cfg @@ -0,0 +1,42 @@ +\* TLC config: known implementation gap: raw writes are not fenced in +\* the default build (--commitTSFence off). Single-shard path, one Raft +\* group, CommitFence = FALSE: the coordinator allocates a raw timestamp +\* before proposing, so two raw entries can apply out of timestamp order +\* and the later-allocated one raises the watermark first; a reader at +\* that watermark misses the earlier one, which then lands below its +\* snapshot. Expected outcome: TLC FAILS OCC3_ReadSnapshotStability. +\* MCOCC_onephase.cfg is the same with the fence on, and passes. + +SPECIFICATION Spec + +CONSTANTS + Keys = {k1, k2} + TxnIds = {t1, t2} + MaxTs = 6 + MaxOps = 3 + EnableSafety = TRUE + EnableTwoPhase = FALSE + MultiGroup = FALSE + ReadAdvance = FALSE + CommitFence = FALSE + +SYMMETRY Symmetry + +INVARIANTS + TypeOK + OCC1_CommitTsAboveStart + OCC2_NoWriteWriteConflict + OCC3_ReadSnapshotStability + OCC4_NoStrandedLockAtQuiescence + OCC5_StartTsConsistency + OCC8_NoWriteSkew + +PROPERTIES + OCC5_Action + CommitTsAssignedOnce + +CONSTRAINT StateConstraint + +\* Quiescence is a valid end state, not a deadlock. +CHECK_DEADLOCK + FALSE diff --git a/tla/occ/OCC.tla b/tla/occ/OCC.tla index c29d499e3..b267f81ae 100644 --- a/tla/occ/OCC.tla +++ b/tla/occ/OCC.tla @@ -70,12 +70,14 @@ CONSTANTS Keys, \* finite set of keys exercised in the model TxnIds, \* finite set of transaction identifiers MaxTs, \* upper bound on the abstract HLC counter (state-space) - MaxOps, \* upper bound on Write + RawWrite actions (state-space) + MaxOps, \* upper bound on WriteIntent + AllocateRaw actions (state-space) EnableSafety, \* TRUE: commit_ts > start_ts discipline; FALSE: gap config EnableTwoPhase, \* TRUE: multi-shard 2PC path enabled; FALSE: one-phase only MultiGroup, \* TRUE: one Raft group per key; FALSE: a single group - ReadAdvance \* TRUE: --readWatermarkAdvance; a snapshot read first + ReadAdvance, \* TRUE: --readWatermarkAdvance; a snapshot read first \* raises the read key's group watermark to start_ts + CommitFence \* TRUE: --commitTSFence; raw applies and PREPAREs are + \* fenced against the group watermark \* Transaction lifecycle states. \* Proposed — one-phase entry with an allocated commit_ts, not applied yet @@ -99,10 +101,12 @@ VARIABLES readObs, \* TxnIds -> [Keys -> Nat] (commit_ts observed; 0 = no version) lockMap, \* Keys -> [owner: TxnIds \cup {"none"}, lockTs: Nat] versions, \* Keys -> SUBSET [commitTs: Nat, owner: TxnIds \cup {"raw"}] - opCount \* state-space bound + opCount, \* state-space bound + rawPending \* SUBSET [key: Keys, ts: Nat]: raw entries allocated + \* but not yet applied vars == <> + writeSet, readSet, readObs, lockMap, versions, opCount, rawPending>> \* === HELPERS === HasLock(k) == lockMap[k] # NoLock @@ -185,6 +189,7 @@ Init == /\ lockMap = [k \in Keys |-> NoLock] /\ versions = [k \in Keys |-> {}] /\ opCount = 0 + /\ rawPending = {} \* === ACTIONS === @@ -204,7 +209,7 @@ BeginTxn(t) == /\ snapshotRead' = [snapshotRead EXCEPT ![t] = TRUE] /\ startTs' = [startTs EXCEPT ![t] = IF GlobalWatermark = 0 THEN MaxTs ELSE GlobalWatermark] /\ UNCHANGED <> + readObs, lockMap, versions, opCount, rawPending>> (***************************************************************************) (* BeginWriteOnly: a coordinator-stamped transaction (no caller start_ts) *) @@ -220,7 +225,7 @@ BeginWriteOnly(t) == /\ snapshotRead' = [snapshotRead EXCEPT ![t] = FALSE] /\ startTs' = [startTs EXCEPT ![t] = tsCounter + 1] /\ UNCHANGED <> + lockMap, versions, opCount, rawPending>> (***************************************************************************) (* ReadKey: snapshot read on key k at startTs[t]. Blocked while a lock *) @@ -248,7 +253,7 @@ ReadKey(t, k) == /\ readSet' = [readSet EXCEPT ![t] = @ \cup {k}] /\ readObs' = [readObs EXCEPT ![t] = [@ EXCEPT ![k] = LatestVisible(k, startTs[t])]] /\ UNCHANGED <> + commitTs, writeSet, lockMap, versions, opCount, rawPending>> (***************************************************************************) (* WriteIntent: buffer a write to k locally in the txn's writeSet. *) @@ -260,26 +265,49 @@ WriteIntent(t, k) == /\ writeSet' = [writeSet EXCEPT ![t] = @ \cup {k}] /\ opCount' = opCount + 1 /\ UNCHANGED <> + commitTs, readSet, readObs, lockMap, versions, rawPending>> (***************************************************************************) -(* RawWrite(k): a non-transactional write (handleRawRequest). It is *) -(* rejected while a txn lock sits on k (validateRawMutationForApply → *) -(* assertNoConflictingTxnLock). Allocation *) -(* and apply are one step here, so the raw path's own reordering window *) -(* is not modelled; its role is to raise the watermark. *) +(* AllocateRaw / ApplyRaw / RejectRaw: a non-transactional write *) +(* (handleRawRequest). The coordinator allocates the timestamp and *) +(* proposes the entry; entries apply in any order. The apply rejects the *) +(* write while a txn lock sits on k (assertNoConflictingTxnLock), when k *) +(* already has a newer version (checkApplyConflicts at start = commit), *) +(* and, with CommitFence, when the timestamp is at or below the group *) +(* watermark (ApplyMutationsRaftAtFenced). A rejected raw write is *) +(* retried by the coordinator with a fresh timestamp, which is a new *) +(* AllocateRaw here. *) (***************************************************************************) -RawWrite(k) == +AllocateRaw(k) == /\ opCount < MaxOps - /\ ~HasLock(k) /\ tsCounter < MaxTs - /\ LET ct == tsCounter + 1 IN - /\ tsCounter' = ct - /\ versions' = [versions EXCEPT ![k] = @ \cup {[commitTs |-> ct, owner |-> "raw"]}] - /\ watermark' = RaiseWatermark({k}, ct) - /\ opCount' = opCount + 1 - /\ UNCHANGED <> + /\ tsCounter' = tsCounter + 1 + /\ rawPending' = rawPending \cup {[key |-> k, ts |-> tsCounter + 1]} + /\ opCount' = opCount + 1 + /\ UNCHANGED <> + +RawApplies(e) == + /\ ~HasLock(e.key) + /\ \A v \in versions[e.key] : v.commitTs <= e.ts + /\ CommitFence => e.ts > watermark[GroupOf(e.key)] + +ApplyRaw(e) == + /\ e \in rawPending + /\ RawApplies(e) + /\ rawPending' = rawPending \ {e} + /\ versions' = [versions EXCEPT ![e.key] = @ \cup {[commitTs |-> e.ts, owner |-> "raw"]}] + /\ watermark' = RaiseWatermark({e.key}, e.ts) + /\ UNCHANGED <> + +RejectRaw(e) == + /\ e \in rawPending + /\ ~RawApplies(e) + /\ rawPending' = rawPending \ {e} + /\ UNCHANGED <> \* --- One-phase path --------------------------------------------------- @@ -297,7 +325,7 @@ AllocateOnePhase(t) == /\ commitTs' = [commitTs EXCEPT ![t] = FreshCommitTs] /\ txnState' = [txnState EXCEPT ![t] = "Proposed"] /\ UNCHANGED <> + readObs, lockMap, versions, opCount, rawPending>> \* Every check the one-phase apply performs. OnePhaseApplies(t) == @@ -319,14 +347,14 @@ ApplyOnePhase(t) == /\ versions' = AddVersions(t, commitTs[t]) /\ watermark' = RaiseWatermark(writeSet[t], commitTs[t]) /\ UNCHANGED <> + readSet, readObs, lockMap, opCount, rawPending>> RejectOnePhase(t) == /\ txnState[t] = "Proposed" /\ ~OnePhaseApplies(t) /\ txnState' = [txnState EXCEPT ![t] = "Aborted"] /\ UNCHANGED <> + writeSet, readSet, readObs, lockMap, versions, opCount, rawPending>> \* --- Two-phase path --------------------------------------------------- @@ -344,7 +372,7 @@ AllocateTwoPhase(t) == /\ commitTs' = [commitTs EXCEPT ![t] = FreshCommitTs] /\ txnState' = [txnState EXCEPT ![t] = "Allocated"] /\ UNCHANGED <> + readObs, lockMap, versions, opCount, rawPending>> (***************************************************************************) (* Prepare: handlePrepareRequest validates the write and read keys *) @@ -353,8 +381,16 @@ AllocateTwoPhase(t) == (* are applied at start_ts through ApplyMutationsRaftAt, which raises the *) (* group's LastCommitTS to at least start_ts. *) (***************************************************************************) +\* With CommitFence, handlePrepareRequest rejects a PREPARE whose settled +\* commit_ts is at or below the watermark of a group it writes to; the +\* coordinator then aborts the transaction (CancelTxn here: nothing of a +\* rejected PREPARE persists). +PrepareFenceOk(t) == + CommitFence => \A k \in writeSet[t] : commitTs[t] > watermark[GroupOf(k)] + Prepare(t) == /\ txnState[t] = "Allocated" + /\ PrepareFenceOk(t) /\ NoForeignLock(t) /\ NoNewerVersion(t, writeSet[t]) /\ NoNewerVersion(t, readSet[t]) @@ -365,7 +401,7 @@ Prepare(t) == ELSE lockMap[k]] /\ watermark' = RaiseWatermark(writeSet[t], startTs[t]) /\ UNCHANGED <> + writeSet, readSet, readObs, versions, opCount, rawPending>> (***************************************************************************) (* Commit: handleCommitRequest promotes every lock owned by t into a *) @@ -380,7 +416,7 @@ Commit(t) == /\ lockMap' = ReleaseLocks(t) /\ watermark' = RaiseWatermark(writeSet[t], commitTs[t]) /\ UNCHANGED <> + readSet, readObs, opCount, rawPending>> (***************************************************************************) (* Abort: three ways a transaction ends without committing. *) @@ -403,7 +439,7 @@ CancelTxn(t) == /\ txnState[t] \in {"Active", "Allocated"} /\ txnState' = [txnState EXCEPT ![t] = "Aborted"] /\ UNCHANGED <> + writeSet, readSet, readObs, lockMap, versions, opCount, rawPending>> PersistRollback(t, abortTs) == /\ txnState' = [txnState EXCEPT ![t] = "Aborted"] @@ -416,13 +452,13 @@ AbortPrepared(t) == /\ tsCounter' = tsCounter + 1 /\ PersistRollback(t, tsCounter + 1) /\ UNCHANGED <> + readObs, versions, opCount, rawPending>> ResolverAbort(t) == /\ txnState[t] = "Prepared" /\ PersistRollback(t, startTs[t] + 1) /\ UNCHANGED <> + readSet, readObs, versions, opCount, rawPending>> Abort(t) == CancelTxn(t) \/ AbortPrepared(t) \/ ResolverAbort(t) @@ -444,7 +480,9 @@ Next == \/ \E t \in TxnIds : BeginWriteOnly(t) \/ \E t \in TxnIds, k \in Keys : ReadKey(t, k) \/ \E t \in TxnIds, k \in Keys : WriteIntent(t, k) - \/ \E k \in Keys : RawWrite(k) + \/ \E k \in Keys : AllocateRaw(k) + \/ \E e \in rawPending : ApplyRaw(e) + \/ \E e \in rawPending : RejectRaw(e) \/ \E t \in TxnIds : AllocateOnePhase(t) \/ \E t \in TxnIds : ApplyOnePhase(t) \/ \E t \in TxnIds : RejectOnePhase(t) @@ -492,6 +530,7 @@ TypeOK == /\ readSet \in [TxnIds -> SUBSET Keys] /\ readObs \in [TxnIds -> [Keys -> 0..MaxTs]] /\ opCount \in 0..MaxOps + /\ rawPending \subseteq [key: Keys, ts: 1..MaxTs] /\ \A k \in Keys : /\ lockMap[k].owner \in TxnIds \cup {"none"} /\ lockMap[k].lockTs \in 0..MaxTs