diff --git a/.github/workflows/jepsen-test-scheduled-dedup.yml b/.github/workflows/jepsen-test-scheduled-dedup.yml index 27c197252..527158f35 100644 --- a/.github/workflows/jepsen-test-scheduled-dedup.yml +++ b/.github/workflows/jepsen-test-scheduled-dedup.yml @@ -1,23 +1,22 @@ # Jepsen Scheduled Stress Test — Option-2 Dedup Mode # # Daily run with ELASTICKV_REDIS_ONEPHASE_DEDUP=1 and -# ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=1 so the demo cluster exercises the -# option-2 idempotency path on both adapters. The criterion for -# default-on (docs/design/2026_05_21 §M4 and -# docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md §M2) is 7 -# consecutive days of green runs with no :duplicate-elements / -# :G-single-item-realtime anomalies in each workload's analysis output. +# ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=1 pinned so the demo cluster exercises the +# option-2 idempotency path on both adapters, independent of future default +# changes. This preserves the high-pressure dedup signal for +# :duplicate-elements / :G-single-item-realtime anomalies in each workload's +# analysis output. # -# Scope: Redis + DynamoDB workloads. The dedup feature ships behind the +# Scope: Redis + DynamoDB workloads. The dedup feature is controlled by the # Redis adapter's onePhaseTxnDedup flag (RPUSH/LPUSH, MULTI/EXEC, # standalone SET) and the DynamoDB adapter's onePhaseTxnDedup flag -# (single-item UpdateItem/PutItem/DeleteItem via PR #920). S3 / SQS do +# (single-item UpdateItem/PutItem/DeleteItem). S3 / SQS do # not yet route through the dedup loop, so re-running them here would add # hours of CI for no signal on the new code path. # # Cadence: 03:17 UTC daily (off-peak; non-zero minute per ScheduleWakeup # guidance). The general 6-hourly scheduled workflow continues to run -# without the dedup gate so the legacy path also stays covered. +# with explicit dedup opt-outs so the legacy path also stays covered. on: schedule: @@ -72,8 +71,8 @@ jobs: # state. Anomalies in :duplicate-elements / :G-single-item-realtime # under this flag indicate a regression in option-2 plumbing. ELASTICKV_REDIS_ONEPHASE_DEDUP: "1" - # Enable the DynamoDB adapter option-2 dedup gate (PR #920). demo.go - # wires it via adapter.NewDynamoDBServer, which reads this env var; + # Pin the DynamoDB adapter option-2 dedup path on. demo.go wires it + # via adapter.NewDynamoDBServer, which reads this env var; # the single-item write path (UpdateItem/PutItem/DeleteItem) then # routes through retryItemWriteWithGenerationDedup on the leader, # exercising the same FSM exact-ts probe as the Redis path. The @@ -156,7 +155,7 @@ jobs: # The env vars are set at the JOB level above and inherited by # all `run:` steps; nothing in demo.go can intercept or unset # them before the adapters read os.Getenv. So if they are "1" - # here, the dedup gates ARE active in the cluster. We print + # here, the dedup paths ARE active in the cluster. We print # them explicitly so a failed run's log makes the configuration # unambiguous (vs the general 6-hourly workflow whose runs would # have empty values here). @@ -165,7 +164,7 @@ jobs: exit 2 fi if [ "${ELASTICKV_DYNAMODB_ONEPHASE_DEDUP:-}" != "1" ]; then - echo "FATAL: ELASTICKV_DYNAMODB_ONEPHASE_DEDUP is not '1' — this workflow runs only with the dedup gate on" + echo "FATAL: ELASTICKV_DYNAMODB_ONEPHASE_DEDUP is not '1' — this workflow pins the dedup path on" exit 2 fi @@ -199,7 +198,7 @@ jobs: working-directory: jepsen timeout-minutes: 10 # --local connects to the already-running demo cluster's dynamo - # ports (gate ON via ELASTICKV_DYNAMODB_ONEPHASE_DEDUP), so the + # ports (dedup ON via ELASTICKV_DYNAMODB_ONEPHASE_DEDUP), so the # single-item list_append writes exercise the option-2 reuse + # exact-ts probe path. A :duplicate-elements anomaly here is the # regression this workflow is meant to catch. diff --git a/.github/workflows/jepsen-test-scheduled.yml b/.github/workflows/jepsen-test-scheduled.yml index 0ea6634f1..9bc2869c1 100644 --- a/.github/workflows/jepsen-test-scheduled.yml +++ b/.github/workflows/jepsen-test-scheduled.yml @@ -39,19 +39,16 @@ jobs: runs-on: ubuntu-latest env: GOCACHE: /tmp/go-build - # Explicit dedup-OFF control baseline. The Redis adapter's - # onePhaseTxnDedup flipped to default-on in - # docs/design/2026_06_10_proposed_redis_onephase_dedup_default_on.md; - # this workflow is preserved as the legacy-path coverage so anomalies - # the dedup gate prevents (`:duplicate-elements`, `:future-read`, - # `:G-single-item-realtime`) continue to be measured against an - # unprotected build. Pair with the dedup-ON workflow - # (.github/workflows/jepsen-test-scheduled-dedup.yml) which sets - # ELASTICKV_REDIS_ONEPHASE_DEDUP=1 explicitly. Retirement of this - # workflow is a follow-up after 30 days of post-flip data; until - # then, do NOT remove this env var — without it the two workflows - # would exercise the same path under the new default. + # Explicit dedup-OFF control baseline. The Redis and DynamoDB adapter + # onePhaseTxnDedup flags are default-on, so this workflow is preserved + # as legacy-path coverage. Pair with the dedup-ON workflow + # (.github/workflows/jepsen-test-scheduled-dedup.yml) which pins both + # env vars to 1. Retirement of this workflow is a follow-up after 30 + # days of post-flip data; until then, do NOT remove these env vars — + # without them the two workflows would exercise the same path under + # the new defaults. ELASTICKV_REDIS_ONEPHASE_DEDUP: "0" + ELASTICKV_DYNAMODB_ONEPHASE_DEDUP: "0" steps: - uses: actions/checkout@v6 with: diff --git a/adapter/dynamodb.go b/adapter/dynamodb.go index b0ca73c44..f53ed238b 100644 --- a/adapter/dynamodb.go +++ b/adapter/dynamodb.go @@ -131,10 +131,9 @@ type DynamoDBServer struct { // retryable write error, the retry REUSES the failed attempt's write set // under a fresh commit_ts and carries prev_commit_ts so the FSM no-ops a // commit that already landed under leadership churn (the :duplicate-elements - // anomaly). It MUST stay off until every node runs a probe-aware binary — - // see R5 in docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md. - // Default off; enabled via WithDynamoOnePhaseTxnDedup or the - // ELASTICKV_DYNAMODB_ONEPHASE_DEDUP env var. + // anomaly). The FSM probe now ships on every node, so this defaults on. + // Set ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0 or pass + // WithDynamoOnePhaseTxnDedup(false) for an operator rollback. onePhaseTxnDedup bool } @@ -163,9 +162,11 @@ func WithDynamoDBLeaderMap(m map[string]string) DynamoDBServerOption { } } -// WithDynamoOnePhaseTxnDedup enables the option-2 one-phase idempotency dedup on -// the single-item write retry path (see DynamoDBServer.onePhaseTxnDedup). Off by -// default; enable only after the whole cluster runs a probe-aware binary. +// WithDynamoOnePhaseTxnDedup enables or disables the option-2 one-phase +// idempotency dedup on the single-item write retry path (see +// DynamoDBServer.onePhaseTxnDedup). It defaults on; pass false to opt out from +// code, or set ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0 to opt out from the +// environment. The constructor option trumps the env var. func WithDynamoOnePhaseTxnDedup(enabled bool) DynamoDBServerOption { return func(server *DynamoDBServer) { server.onePhaseTxnDedup = enabled @@ -259,7 +260,7 @@ func NewDynamoDBServer(listen net.Listener, st store.MVCCStore, coordinate kv.Co listen: listen, store: st, coordinator: coordinate, - onePhaseTxnDedup: os.Getenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP") == "1", + onePhaseTxnDedup: os.Getenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP") != "0", } d.targetHandlers = map[string]func(http.ResponseWriter, *http.Request){ createTableTarget: d.createTable, diff --git a/adapter/dynamodb_item_write.go b/adapter/dynamodb_item_write.go index a56822591..31cbeec48 100644 --- a/adapter/dynamodb_item_write.go +++ b/adapter/dynamodb_item_write.go @@ -119,11 +119,12 @@ func (d *DynamoDBServer) retryItemWriteWithGeneration( exhaustedMessage string, prepare func(readTS uint64) (*itemWritePlan, error), ) (*itemWritePlan, error) { - // Option-2 one-phase dedup (gated, default off): on a retryable write error, - // reuse the failed attempt's write set under a fresh commit_ts + prev_commit_ts - // so the FSM no-ops a commit that already landed under leadership churn, - // instead of re-reading and re-appending (the :duplicate-elements anomaly). - // See docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md. + // Option-2 one-phase dedup (default on, with an explicit rollback switch): + // on a retryable write error, reuse the failed attempt's write set under a + // fresh commit_ts + prev_commit_ts so the FSM no-ops a commit that already + // landed under leadership churn, instead of re-reading and re-appending (the + // :duplicate-elements anomaly). See + // docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md. // // Leader-only (codex P1, PR #920): the dedup path allocates commit_ts from // the LOCAL HLC and carries it as prev_commit_ts, so that timestamp MUST be @@ -143,7 +144,7 @@ func (d *DynamoDBServer) retryItemWriteWithGeneration( // retryItemWriteWithGenerationLegacy is the pre-dedup retry loop: it recomputes // the write set from a fresh read on every retryable error. It is the active -// path whenever the dedup gate is off or this node is not the leader, so it +// path whenever dedup is explicitly off or this node is not the leader, so it // stays byte-identical to the pre-feature behavior. func (d *DynamoDBServer) retryItemWriteWithGenerationLegacy( ctx context.Context, diff --git a/adapter/dynamodb_onephase_dedup_test.go b/adapter/dynamodb_onephase_dedup_test.go index b65aee1a3..a3b5d0bc2 100644 --- a/adapter/dynamodb_onephase_dedup_test.go +++ b/adapter/dynamodb_onephase_dedup_test.go @@ -74,11 +74,7 @@ func readListValues(t *testing.T, server *DynamoDBServer, schema *dynamoTableSch } func newDedupItemWriteServer(st store.MVCCStore, coord kv.Coordinator, dedup bool) (*dynamoTableSchema, *DynamoDBServer) { - var opts []DynamoDBServerOption - if dedup { - opts = append(opts, WithDynamoOnePhaseTxnDedup(true)) - } - server := NewDynamoDBServer(nil, st, coord, opts...) + server := NewDynamoDBServer(nil, st, coord, WithDynamoOnePhaseTxnDedup(dedup)) return dedupItemTable(), server } @@ -251,8 +247,32 @@ func TestItemWriteDedup_NonLeaderFallsBackToLegacy(t *testing.T) { "non-leader falls back to the legacy recompute path (leader allocates commit_ts via redirect)") } +func TestItemWriteDedup_DefaultOn(t *testing.T) { + t.Setenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP", "") + server := NewDynamoDBServer(nil, store.NewMVCCStore(), newDedupTestCoordinator(store.NewMVCCStore(), 0, false)) + require.True(t, server.onePhaseTxnDedup) +} + +func TestItemWriteDedup_EnvOptOut(t *testing.T) { + t.Setenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP", "0") + server := NewDynamoDBServer(nil, store.NewMVCCStore(), newDedupTestCoordinator(store.NewMVCCStore(), 0, false)) + require.False(t, server.onePhaseTxnDedup) +} + +func TestItemWriteDedup_OptionOverridesEnv(t *testing.T) { + t.Setenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP", "0") + server := NewDynamoDBServer(nil, store.NewMVCCStore(), newDedupTestCoordinator(store.NewMVCCStore(), 0, false), WithDynamoOnePhaseTxnDedup(true)) + require.True(t, server.onePhaseTxnDedup) +} + +func TestItemWriteDedup_OptionOverridesEnvToDisable(t *testing.T) { + t.Setenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP", "1") + server := NewDynamoDBServer(nil, store.NewMVCCStore(), newDedupTestCoordinator(store.NewMVCCStore(), 0, false), WithDynamoOnePhaseTxnDedup(false)) + require.False(t, server.onePhaseTxnDedup) +} + // TestItemWriteDedup_DisabledKeepsLegacyPath pins that the gate is load-bearing: -// with onePhaseTxnDedup OFF (the default), the legacy retry RE-READS and +// with onePhaseTxnDedup explicitly OFF, the legacy retry RE-READS and // recomputes, so a landed-then-ambiguous attempt 1 is double-applied — the // :duplicate-elements bug. This characterizes the pre-fix behavior the gate // closes; flipping the gate on (the headline test) eliminates the duplicate. diff --git a/adapter/redis.go b/adapter/redis.go index df76dbc50..a19248e15 100644 --- a/adapter/redis.go +++ b/adapter/redis.go @@ -391,11 +391,16 @@ func (c *redisMetricsConn) WriteError(msg string) { // kv.isLeadershipLossError so any sentinel those classifiers already // recognize as transient also flips a Redis reply to NOTLEADER. func writeRedisError(conn redcon.Conn, err error) { + msg := err.Error() if isTransientLeaderRedisError(err) { - conn.WriteError("NOTLEADER " + err.Error()) + if strings.HasPrefix(strings.ToUpper(msg), "NOTLEADER ") { + conn.WriteError(msg) + return + } + conn.WriteError("NOTLEADER " + msg) return } - conn.WriteError(err.Error()) + conn.WriteError(msg) } // isTransientLeaderRedisError reports whether err is a transient @@ -1983,6 +1988,9 @@ func (r *RedisServer) proxyTransactionToLeader(conn redcon.Conn, queue []redcon. if handleProxyTxnError(conn, err) { return } + if handleProxyTxnCommandError(conn, cmds) { + return + } writeProxyCmdsResult(conn, cmds) } @@ -2033,7 +2041,8 @@ func handleProxyTxnError(conn redcon.Conn, err error) bool { // Fatal transport / context error: per-command results are unreliable. if err != nil { var netErr net.Error - if errors.Is(err, context.DeadlineExceeded) || + if isTransientLeaderRedisError(err) || + errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || @@ -2045,6 +2054,24 @@ func handleProxyTxnError(conn redcon.Conn, err error) bool { return false } +// handleProxyTxnCommandError promotes transient leadership failures returned +// by the proxied EXEC target to a top-level EXEC error. go-redis can surface a +// target's EXEC-level error on the queued command handles; writing those as +// EXEC array elements makes clients treat leadership churn like command data. +func handleProxyTxnCommandError(conn redcon.Conn, cmds []*redis.Cmd) bool { + for _, cmd := range cmds { + err := cmd.Err() + if err == nil || errors.Is(err, redis.Nil) { + continue + } + if isTransientLeaderRedisError(err) { + writeRedisError(conn, err) + return true + } + } + return false +} + // writeProxyCmdsResult writes an EXEC-style array reply for the given pipeline // command handles. For any other non-nil per-command errors, each cmd carries // its own result, which is the correct Redis EXEC semantics. diff --git a/adapter/redis_error_prefix_test.go b/adapter/redis_error_prefix_test.go index 670df0303..245f4a9c6 100644 --- a/adapter/redis_error_prefix_test.go +++ b/adapter/redis_error_prefix_test.go @@ -1,6 +1,7 @@ package adapter import ( + "context" "fmt" "io" "testing" @@ -8,17 +9,23 @@ import ( "github.com/bootjp/elastickv/internal/raftengine" "github.com/bootjp/elastickv/kv" "github.com/cockroachdb/errors" + "github.com/redis/go-redis/v9" "github.com/tidwall/redcon" ) -// captureConn is the minimal redcon.Conn surface writeRedisError uses; -// only WriteError needs a real implementation. +// captureConn is the minimal redcon.Conn surface these tests inspect. type captureConn struct { redcon.Conn - lastErr string + lastErr string + lastArray int + wroteArray bool } func (c *captureConn) WriteError(msg string) { c.lastErr = msg } +func (c *captureConn) WriteArray(count int) { + c.lastArray = count + c.wroteArray = true +} func TestIsTransientLeaderRedisError(t *testing.T) { t.Parallel() @@ -105,6 +112,9 @@ func TestWriteRedisError(t *testing.T) { {"grpc-wrapped leader-not-found gains NOTLEADER prefix", errors.New("rpc error: code = Unknown desc = leader not found"), "NOTLEADER rpc error: code = Unknown desc = leader not found"}, + {"already NOTLEADER-prefixed error is not double-prefixed", + errors.New("NOTLEADER leader not found"), + "NOTLEADER leader not found"}, // Regression: address-mapping gap errors (raft leader known // but raft→redis address missing in r.leaderRedis) must be // ERR-prefixed at the source so Carmine maps to :prefix :err @@ -131,6 +141,69 @@ func TestWriteRedisError(t *testing.T) { } } +func TestHandleProxyTxnError(t *testing.T) { + t.Parallel() + t.Run("transient leader error is top-level NOTLEADER", func(t *testing.T) { + t.Parallel() + c := &captureConn{} + handled := handleProxyTxnError(c, errors.New("rpc error: code = Unknown desc = leader not found")) + if !handled { + t.Fatal("handleProxyTxnError returned false") + } + if c.lastErr != "NOTLEADER rpc error: code = Unknown desc = leader not found" { + t.Fatalf("last error = %q", c.lastErr) + } + if c.wroteArray { + t.Fatalf("unexpected array reply %d", c.lastArray) + } + }) + + t.Run("redis transaction abort remains null array", func(t *testing.T) { + t.Parallel() + c := &captureConn{} + handled := handleProxyTxnError(c, redis.TxFailedErr) + if !handled { + t.Fatal("handleProxyTxnError returned false") + } + if !c.wroteArray || c.lastArray != -1 { + t.Fatalf("array reply = (%v, %d), want (-1)", c.wroteArray, c.lastArray) + } + if c.lastErr != "" { + t.Fatalf("unexpected error reply %q", c.lastErr) + } + }) +} + +func TestHandleProxyTxnCommandError(t *testing.T) { + t.Parallel() + t.Run("transient command error is promoted to top-level NOTLEADER", func(t *testing.T) { + t.Parallel() + cmd := redis.NewCmd(context.Background(), "LRANGE", "k", 0, -1) + cmd.SetErr(errors.New("rpc error: code = FailedPrecondition desc = raft engine: not leader")) + c := &captureConn{} + handled := handleProxyTxnCommandError(c, []*redis.Cmd{cmd}) + if !handled { + t.Fatal("handleProxyTxnCommandError returned false") + } + if c.lastErr != "NOTLEADER rpc error: code = FailedPrecondition desc = raft engine: not leader" { + t.Fatalf("last error = %q", c.lastErr) + } + }) + + t.Run("ordinary command error stays in EXEC result array", func(t *testing.T) { + t.Parallel() + cmd := redis.NewCmd(context.Background(), "LRANGE", "k", 0, -1) + cmd.SetErr(errors.New("WRONGTYPE Operation against a key holding the wrong kind of value")) + c := &captureConn{} + if handleProxyTxnCommandError(c, []*redis.Cmd{cmd}) { + t.Fatal("ordinary command error was handled as terminal") + } + if c.lastErr != "" { + t.Fatalf("unexpected error reply %q", c.lastErr) + } + }) +} + // TestHasTransientLeaderSuffix_PinsSentinels closes the gap noted // at kv/coordinator.go:529 ("A symmetric pin lives in the adapter // test package"): the adapter's redisLeaderErrorPhrases set must diff --git a/docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md b/docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md index 323bb79ae..c36281c63 100644 --- a/docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md +++ b/docs/design/2026_06_03_partial_dynamodb_onephase_dedup.md @@ -296,22 +296,24 @@ fixed), so the adapter returns the remembered `plan` on a dedup no-op — no store re-read. This is the parent doc's R1 invariance, instantiated for the item write. -### Gating (R5) — default off, ship the reader before the writer +### Gating (R5) — default on, rollback switch remains -Emission is gated behind a new DynamoDB flag, mirroring -`RedisServer.onePhaseTxnDedup`: +Emission is controlled by a DynamoDB flag, mirroring +`RedisServer.onePhaseTxnDedup`. The probe-aware FSM reader now ships on every +node, so the writer side can be the default behavior while retaining an +operator rollback switch: - `DynamoDBServer.onePhaseTxnDedup bool`, set from - `os.Getenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP") == "1"` and/or a + `os.Getenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP") != "0"` and/or a `WithDynamoOnePhaseTxnDedup(bool)` server option (distinct symbol name to avoid colliding with the Redis `WithOnePhaseTxnDedup` in the same `adapter` package). -- **Off** → `retryItemWriteWithGeneration` keeps its current +- **On (default)** → the dedup loop above. +- **Off (`ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0` or + `WithDynamoOnePhaseTxnDedup(false)`)** → + `retryItemWriteWithGeneration` keeps its current recompute-on-retry behavior byte-for-byte; no `PrevCommitTS` is ever emitted; - the FSM probe never fires. No mixed-version divergence window. -- **On** → the dedup loop above. Enable only after the whole cluster runs a - probe-aware binary (the FSM probe code is already everywhere as of the parent - doc's M2b, so this is satisfied on current `main`). + the FSM probe never fires. The FSM-determinism argument (R5 in the parent doc) holds unchanged: the probe decision depends only on the replicated `PrevCommitTS` and the deterministic @@ -373,16 +375,16 @@ every replica applying the same log entry. ## Milestones -### M1 — DynamoDB adapter reuse-dedup (gated) +### M1 — DynamoDB adapter reuse-dedup (default on) - Adapter `commit_ts` allocation in the prepare step (`d.coordinator.Clock().NextFenced()` → `plan.req.CommitTS`). - `reusableItemWrite` struct + `dispatchItemWriteReuse` (mirror `reusableListPush` / `dispatchListPushReuse`). -- `retryItemWriteWithGeneration` gains the gated dedup loop; the gate-off path - is unchanged. +- `retryItemWriteWithGeneration` gains the dedup loop; the explicit opt-out + path is unchanged. - `DynamoDBServer.onePhaseTxnDedup` + `WithDynamoOnePhaseTxnDedup` + env var; - wire the env var in the server constructor. + wire the env var in the server constructor with `0` as the rollback value. - **Tests (`adapter/dynamodb_*_dedup_test.go`), all three outcomes against the real OCC + the real probe, plus the gate-off legacy path:** - `LandedPriorAttempt_ReturnsCachedResult` — attempt 1 lands then errors; @@ -406,20 +408,21 @@ every replica applying the same log entry. - Local reproduction: 3-node demo under CPU pressure / shortened election timeouts (`cmd/server/demo.go`) so leadership flaps during the DynamoDB - workload, with `ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=1`. + workload. The default path has DynamoDB dedup enabled; set + `ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0` only to reproduce the legacy path. - **CI — LANDED.** The DynamoDB list-append workload is added to the dedup-mode workflow (`.github/workflows/jepsen-test-scheduled-dedup.yml`) with - `ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=1` set at the job env (read by + `ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=1` pinned at the job env (read by `adapter.NewDynamoDBServer` in the demo cluster), a fail-closed gate assertion before the listeners come up (mirroring the Redis assertion), and the launch step now also waits on the dynamo listeners (63801-63803). The - general workflow (`.github/workflows/jepsen-test-scheduled.yml`) continues to - run DynamoDB **gate-off** so the legacy path stays covered until default-on. + general workflow (`.github/workflows/jepsen-test-scheduled.yml`) explicitly + sets `ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0` so the legacy path stays covered + as a control baseline after default-on. - Criterion to default-on: 7 consecutive days without `:duplicate-elements` in - the dedup-mode DynamoDB workload, both workflows green. **Default-on is a - separate follow-up PR** flipping `DynamoDBServer.onePhaseTxnDedup`'s default - (and the env-var sense) only after the 7-day criterion is met — not done in - the workflow PR, to keep the ship-reader-before-writer sequencing (R5). + the dedup-mode DynamoDB workload, both workflows green. **Satisfied; this PR + flips `DynamoDBServer.onePhaseTxnDedup`'s default and the env-var sense to + default-on with `0` as the rollback value.** Scope estimate: M1 is ~120–180 LOC Go in `adapter/dynamodb.go` + tests; no FSM change (the probe already exists), no proto change, no new store primitive. @@ -461,12 +464,17 @@ change (the probe already exists), no proto change, no new store primitive. `firstWriteKey`; for list-push the first write key and the min key coincide on the seq-addressed layout, but for the item write they need not.) - **R5 — FSM determinism across rolling upgrade.** Unchanged from the parent - doc: V1-default wire format (no `PrevCommitTS` ⇒ V1 meta), fail-loud - unknown-flag rejection, default-off emission. The probe-aware FSM is already - on `main`. + doc: V1-compatible opt-out wire format (no `PrevCommitTS` ⇒ V1 meta), + fail-loud unknown-flag rejection, and default-on emission only after the + probe-aware FSM ships on every node. ## Decision log +- (2026-06-18) Default-on follow-up: `DynamoDBServer.onePhaseTxnDedup` now + defaults on because the probe-aware FSM reader is everywhere. Operators can + still set `ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0` or + `WithDynamoOnePhaseTxnDedup(false)` for rollback; the general scheduled + Jepsen workflow pins that opt-out to keep legacy-path coverage. - (2026-06-03, PR #920 round-1) **Leader-only dedup guard added** per codex P1: the adapter-local `commitTS` allocation is only safe on the leader, so the dedup path is gated on `d.coordinator.IsLeader()` (+ `NextFenced` ceiling @@ -481,4 +489,4 @@ change (the probe already exists), no proto change, no new store primitive. (`2026_05_21_proposed_txn_secondary_idempotency.md`); the DynamoDB single-item write recomputes its write set on a self-inflicted `WriteConflict` under leadership churn and double-applies. Fix: extend - option-2 reuse-dedup to `retryItemWriteWithGeneration`, gated off by default. + option-2 reuse-dedup to `retryItemWriteWithGeneration`. diff --git a/jepsen/src/elastickv/redis_workload.clj b/jepsen/src/elastickv/redis_workload.clj index b2821d371..20283a64c 100644 --- a/jepsen/src/elastickv/redis_workload.clj +++ b/jepsen/src/elastickv/redis_workload.clj @@ -18,6 +18,62 @@ (def default-nodes ["n1" "n2" "n3" "n4" "n5"]) +(defn- redis-exception? [x] + (instance? Throwable x)) + +(defn- first-exception [xs] + (some (fn [x] + (when (redis-exception? x) + x)) + xs)) + +(defn- mop-result-error [mops] + (first-exception (map (fn [[_ _ v]] v) mops))) + +(defn- redis-error-label [e] + (let [data (ex-data e) + prefix (:prefix data) + msg (or (.getMessage e) "") + upper (str/upper-case msg)] + (cond + prefix prefix + (str/starts-with? upper "NOTLEADER") :notleader + (str/starts-with? upper "MOVED") :moved + (str/starts-with? upper "ASK") :ask + (str/starts-with? upper "WRONGTYPE") :wrongtype + (str/starts-with? upper "ERR") :err + :else :exec-error))) + +(defn- exec-array-error-op [op e] + (assoc op :type :info + :error [:exec-array-error (redis-error-label e)])) + +(defn- complete-mops-or-error [conn op mops] + (if-let [e (mop-result-error mops)] + (exec-array-error-op op e) + (->> mops + (mapv (partial redis-append/parse-read conn)) + (assoc op :type :ok, :value)))) + +(defn- complete-exec-results [conn op exec-results] + (if-let [e (first-exception exec-results)] + (exec-array-error-op op e) + (let [mops (mapv (fn [[f k v] r] + [f k (if (= f :r) r v)]) + (:value op) + exec-results)] + (complete-mops-or-error conn op mops)))) + +(defn- apply-multi-op! [conn op] + (let [exec-results (rc/with-txn conn + (mapv (partial redis-append/apply-mop! conn) (:value op)))] + (complete-exec-results conn op exec-results))) + +(defn- apply-single-op! [conn op] + (->> (:value op) + (mapv (partial redis-append/apply-mop! conn)) + (complete-mops-or-error conn op))) + (defrecord ElastickvRedisClient [node->port conn] client/Client (open! [this test node] @@ -38,17 +94,9 @@ (let [conn (:conn this)] (rc/with-exceptions op #{} (rc/with-conn conn - (->> (if (< 1 (count (:value op))) - (->> (:value op) - (mapv (partial redis-append/apply-mop! conn)) - (rc/with-txn conn) - (mapv (fn [[f k v] r] - [f k (if (= f :r) r v)]) - (:value op))) - (->> (:value op) - (mapv (partial redis-append/apply-mop! conn)))) - (mapv (partial redis-append/parse-read conn)) - (assoc op :type :ok, :value))))))) + (if (< 1 (count (:value op))) + (apply-multi-op! conn op) + (apply-single-op! conn op))))))) (defn elastickv-append-workload "Append workload that reuses jepsen-io/redis logic but targets elastickv." diff --git a/jepsen/test/elastickv/redis_workload_test.clj b/jepsen/test/elastickv/redis_workload_test.clj new file mode 100644 index 000000000..c7ec7d5e2 --- /dev/null +++ b/jepsen/test/elastickv/redis_workload_test.clj @@ -0,0 +1,27 @@ +(ns elastickv.redis-workload-test + (:require [clojure.test :refer :all] + [elastickv.redis-workload :as workload])) + +(deftest complete-mops-parses-reads + (let [op {:type :invoke + :f :txn + :value [[:r 1 nil] [:append 1 3]]} + result (#'workload/complete-mops-or-error + nil + op + [[:r 1 ["1" "2"]] [:append 1 3]])] + (is (= :ok (:type result))) + (is (= [[:r 1 [1 2]] [:append 1 3]] (:value result))))) + +(deftest exec-array-error-becomes-info + (let [op {:type :invoke + :f :txn + :value [[:r 1 nil] [:append 1 3]]} + err (ex-info "NOTLEADER leader not found" {:prefix :notleader}) + result (#'workload/complete-exec-results + nil + op + [["1"] err])] + (is (= :info (:type result))) + (is (= [:exec-array-error :notleader] (:error result))) + (is (= (:value op) (:value result)))))