diff --git a/.github/workflows/jepsen-test-scheduled.yml b/.github/workflows/jepsen-test-scheduled.yml index 03741b7d8..42340a978 100644 --- a/.github/workflows/jepsen-test-scheduled.yml +++ b/.github/workflows/jepsen-test-scheduled.yml @@ -7,6 +7,20 @@ on: description: "Workload runtime seconds" required: false default: "300" + long-fuse-run: + description: "Run FUSE Jepsen workloads for 30 minutes" + type: boolean + required: false + default: false + strict-mode: + description: "Use strict FUSE mode for filesystem workloads" + type: boolean + required: false + default: true + workloads-json: + description: "Optional JSON array of workload names to run" + required: false + default: "" rate: description: "Ops/sec per worker" required: false @@ -41,27 +55,7 @@ jobs: strategy: fail-fast: false matrix: - workload: - - unit - - redis - - redis-zset - - dynamodb - - dynamodb-type-string - - dynamodb-type-number - - dynamodb-type-binary - - dynamodb-type-bool - - dynamodb-type-null - - dynamodb-type-string-set - - dynamodb-type-number-set - - dynamodb-type-binary-set - - dynamodb-type-list - - dynamodb-type-map - - s3 - - filesystem-cross-mount - - filesystem-sqlite-check - - filesystem-overwrite - - filesystem-rename - - sqlite + workload: ${{ fromJSON(inputs.workloads-json || '["unit","redis","redis-zset","dynamodb","dynamodb-type-string","dynamodb-type-number","dynamodb-type-binary","dynamodb-type-bool","dynamodb-type-null","dynamodb-type-string-set","dynamodb-type-number-set","dynamodb-type-binary-set","dynamodb-type-list","dynamodb-type-map","s3","filesystem-cross-mount","filesystem-sqlite-check","filesystem-node-crash","filesystem-benchmark","filesystem-benchmark-legacy","filesystem-benchmark-multigroup","filesystem-benchmark-multigroup-legacy","filesystem-overwrite","filesystem-rename","sqlite"]') }} env: GOCACHE: /tmp/go-build WORKLOAD: ${{ matrix.workload }} @@ -159,9 +153,9 @@ jobs: - name: Run Redis Jepsen workload against elastickv if: matrix.workload == 'redis' working-directory: jepsen - timeout-minutes: 15 + timeout-minutes: 70 run: | - timeout 720 ~/lein run -m elastickv.redis-workload \ + timeout 3600 ~/lein run -m elastickv.redis-workload \ --time-limit ${{ inputs.time-limit || '300' }} \ --rate ${{ inputs.rate || '10' }} \ --concurrency ${{ inputs.concurrency || '8' }} \ @@ -173,9 +167,9 @@ jobs: - name: Run Redis ZSet safety Jepsen workload against elastickv if: matrix.workload == 'redis-zset' working-directory: jepsen - timeout-minutes: 15 + timeout-minutes: 70 run: | - timeout 720 ~/lein run -m elastickv.redis-zset-safety-workload \ + timeout 3600 ~/lein run -m elastickv.redis-zset-safety-workload \ --time-limit ${{ inputs.time-limit || '300' }} \ --rate ${{ inputs.rate || '10' }} \ --concurrency ${{ inputs.concurrency || '8' }} \ @@ -184,9 +178,9 @@ jobs: - name: Run DynamoDB Jepsen workload against elastickv if: matrix.workload == 'dynamodb' working-directory: jepsen - timeout-minutes: 15 + timeout-minutes: 70 run: | - timeout 720 ~/lein run -m elastickv.dynamodb-workload --local \ + timeout 3600 ~/lein run -m elastickv.dynamodb-workload --local \ --time-limit ${{ inputs.time-limit || '300' }} \ --rate ${{ inputs.rate || '10' }} \ --concurrency ${{ inputs.concurrency || '8' }} \ @@ -220,9 +214,9 @@ jobs: - name: Run S3 Jepsen workload against elastickv if: matrix.workload == 's3' working-directory: jepsen - timeout-minutes: 15 + timeout-minutes: 70 run: | - timeout 720 ~/lein run -m elastickv.s3-workload --local \ + timeout 3600 ~/lein run -m elastickv.s3-workload --local \ --time-limit ${{ inputs.time-limit || '300' }} \ --rate ${{ inputs.rate || '10' }} \ --concurrency ${{ inputs.concurrency || '8' }} \ @@ -244,6 +238,44 @@ jobs: RAFT_REDIS_MAP="127.0.0.1:50051=127.0.0.1:63791,127.0.0.1:50052=127.0.0.1:63792,127.0.0.1:50053=127.0.0.1:63793" RAFT_S3_MAP="127.0.0.1:50051=127.0.0.1:63901,127.0.0.1:50052=127.0.0.1:63902,127.0.0.1:50053=127.0.0.1:63903" RAFT_DYNAMO_MAP="127.0.0.1:50051=127.0.0.1:63801,127.0.0.1:50052=127.0.0.1:63802,127.0.0.1:50053=127.0.0.1:63803" + STRICT_FLAGS=(--filesystemStrict) + if [[ "${{ github.event_name }}" == 'workflow_dispatch' && "${{ inputs.strict-mode }}" == 'false' ]] || + [[ "$WORKLOAD" == 'filesystem-benchmark-legacy' || "$WORKLOAD" == 'filesystem-benchmark-multigroup-legacy' ]]; then + STRICT_FLAGS=() + fi + + if [[ "$WORKLOAD" == 'filesystem-benchmark-multigroup' || "$WORKLOAD" == 'filesystem-benchmark-multigroup-legacy' ]]; then + mkdir -p /tmp/elastickv-fs/fs-n1 + nohup /tmp/elastickv-bin \ + --address "127.0.0.1:50051" \ + --redisAddress "" --dynamoAddress "" --s3Address "" --sqsAddress "" \ + --filesystemMount /tmp/elastickv-fs/fs-n1 \ + "${STRICT_FLAGS[@]}" \ + --filesystemRootUID "$(id -u)" \ + --filesystemRootGID "$(id -g)" \ + --metricsAddress "" --pprofAddress "" \ + --raftId n1 --raftDataDir /tmp/elastickv-fs/n1 --raftBootstrap \ + --raftGroups "1=127.0.0.1:50051,2=127.0.0.1:50054" \ + --shardRanges ':!fs|op|=1,!fs|op|:=2' \ + > /tmp/elastickv-fs-n1.log 2>&1 & + echo $! > /tmp/elastickv-fs.pid + for _ in {1..90}; do + if ! kill -0 "$(cat /tmp/elastickv-fs.pid)" 2>/dev/null; then + echo "Two-group filesystem process exited before readiness" >&2 + tail -n 200 /tmp/elastickv-fs-n1.log || true + exit 1 + fi + if nc -z 127.0.0.1 50051 && nc -z 127.0.0.1 50054 \ + && mountpoint -q /tmp/elastickv-fs/fs-n1 \ + && stat /tmp/elastickv-fs/fs-n1 >/dev/null; then + echo "Two-group filesystem cluster is up" + exit 0 + fi + sleep 1 + done + tail -n 200 /tmp/elastickv-fs-n1.log || true + exit 1 + fi : > /tmp/elastickv-fs.pid for node in 1 2 3; do @@ -254,7 +286,7 @@ jobs: --dynamoAddress "127.0.0.1:6380${node}" \ --s3Address "127.0.0.1:6390${node}" \ --filesystemMount "/tmp/elastickv-fs/fs-n${node}" \ - --filesystemStrict \ + "${STRICT_FLAGS[@]}" \ --filesystemRootUID "$(id -u)" \ --filesystemRootGID "$(id -g)" \ --metricsAddress "" \ @@ -296,15 +328,47 @@ jobs: - name: Verify SQLite rollback journals without client retries if: matrix.workload == 'filesystem-sqlite-check' run: python3 scripts/test-fuse-sqlite.py /tmp/elastickv-fs/fs-n1 --writes 30 + - name: Verify strict FUSE fsync after Raft node crash + if: matrix.workload == 'filesystem-node-crash' + timeout-minutes: 5 + run: | + python3 scripts/test-fuse-node-crash.py \ + /tmp/elastickv-fs/fs-n1 \ + /tmp/elastickv-fs/fs-n2 \ + /tmp/elastickv-fs/fs-n3 \ + --pid-file /tmp/elastickv-fs.pid \ + --restart-log /tmp/elastickv-fs-n1-restarted.log + - name: Measure FUSE write and fsync latency + if: matrix.workload == 'filesystem-benchmark' || matrix.workload == 'filesystem-benchmark-legacy' || matrix.workload == 'filesystem-benchmark-multigroup' || matrix.workload == 'filesystem-benchmark-multigroup-legacy' + timeout-minutes: 10 + run: | + python3 scripts/test_benchmark_fuse.py + MODE=strict + if [[ "$WORKLOAD" == 'filesystem-benchmark-legacy' || "$WORKLOAD" == 'filesystem-benchmark-multigroup-legacy' ]]; then MODE=legacy; fi + LAYOUT=single-group + if [[ "$WORKLOAD" == 'filesystem-benchmark-multigroup' || "$WORKLOAD" == 'filesystem-benchmark-multigroup-legacy' ]]; then LAYOUT=two-group; fi + python3 scripts/benchmark-fuse.py /tmp/elastickv-fs/fs-n1 \ + --mode "$MODE" --layout "$LAYOUT" --writes 40 --workers 4 --repetitions 3 \ + --output /tmp/elastickv-fuse-benchmark.json + - name: Upload FUSE benchmark results + if: matrix.workload == 'filesystem-benchmark' || matrix.workload == 'filesystem-benchmark-legacy' || matrix.workload == 'filesystem-benchmark-multigroup' || matrix.workload == 'filesystem-benchmark-multigroup-legacy' + uses: actions/upload-artifact@v7 + with: + name: elastickv-fuse-benchmark-${{ matrix.workload }} + path: /tmp/elastickv-fuse-benchmark.json + retention-days: 14 - name: Run filesystem Jepsen workload against elastickv if: matrix.workload == 'filesystem-overwrite' || matrix.workload == 'filesystem-rename' working-directory: jepsen # Each write mode uses a fresh filesystem cluster on its own runner. - # See jepsen/src/elastickv/filesystem_workload.clj. - timeout-minutes: 30 + # The scheduled run keeps enough bounded registers to exercise strict + # mutations for the full 30-minute workload. See filesystem_budget.sh. + timeout-minutes: 70 env: - FS_REQUESTED_KEYS: ${{ inputs.key-count || '16' }} - FS_REQUESTED_WRITES: ${{ inputs.max-writes-per-key || '250' }} + FS_TIME_LIMIT: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '1800' || inputs.time-limit || '300' }} + FS_RATE: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '1' || inputs.rate || '10' }} + FS_REQUESTED_KEYS: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '512' || inputs.key-count || '16' }} + FS_REQUESTED_WRITES: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '16' || inputs.max-writes-per-key || '250' }} run: | # Knossos checks each register independently. Bound each history's # ambiguous writes while keeping the requested total write budget. @@ -323,9 +387,9 @@ jobs: MODE="${WORKLOAD#filesystem-}" filesystem_budget "$MODE" "$FS_REQUESTED_KEYS" "$FS_REQUESTED_WRITES" echo "Filesystem register budget: ${FS_KEY_COUNT} keys, ${FS_MAX_WRITES} writes per key" - timeout 720 ~/lein run -m elastickv.filesystem-workload --local \ - --time-limit ${{ inputs.time-limit || '300' }} \ - --rate ${{ inputs.rate || '10' }} \ + timeout 3600 ~/lein run -m elastickv.filesystem-workload --local \ + --time-limit "$FS_TIME_LIMIT" \ + --rate "$FS_RATE" \ --concurrency ${{ inputs.concurrency || '8' }} \ --key-count "$FS_KEY_COUNT" \ --max-writes-per-key "$FS_MAX_WRITES" \ @@ -334,29 +398,35 @@ jobs: - name: Run SQLite Jepsen workload on FUSE if: matrix.workload == 'sqlite' working-directory: jepsen - timeout-minutes: 20 + timeout-minutes: 70 + env: + FS_TIME_LIMIT: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '1800' || inputs.time-limit || '300' }} + FS_RATE: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '0.25' || inputs.rate || '10' }} + SQLITE_REQUESTED_KEYS: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '128' || '24' }} + SQLITE_REQUESTED_WRITES: ${{ (github.event_name == 'schedule' || inputs.long-fuse-run) && '16' || '64' }} run: | # Spread the longer run across databases while bounding the # ambiguous writes Knossos must explore for any one database. bash test/sqlite_budget_test.sh source ./sqlite_budget.sh - sqlite_budget 24 64 + sqlite_budget "$SQLITE_REQUESTED_KEYS" "$SQLITE_REQUESTED_WRITES" echo "SQLite register budget: ${SQLITE_KEY_COUNT} keys, ${SQLITE_MAX_WRITES} writes per key" - timeout 900 ~/lein run -m elastickv.sqlite-workload --local \ - --time-limit ${{ inputs.time-limit || '300' }} \ - --rate ${{ inputs.rate || '10' }} \ + timeout 3600 ~/lein run -m elastickv.sqlite-workload --local \ + --time-limit "$FS_TIME_LIMIT" \ + --rate "$FS_RATE" \ --concurrency 8 --key-count "$SQLITE_KEY_COUNT" --max-writes-per-key "$SQLITE_MAX_WRITES" \ --mount-dirs /tmp/elastickv-fs/fs-n1,/tmp/elastickv-fs/fs-n2,/tmp/elastickv-fs/fs-n3 - - name: Archive Jepsen store on failure - if: failure() && matrix.workload != 'unit' + - name: Archive Jepsen history + if: always() && matrix.workload != 'unit' run: bash scripts/archive-jepsen-store.sh jepsen/store /tmp/jepsen-store-types.tar.gz - - name: Upload Jepsen store on failure - if: failure() && matrix.workload != 'unit' + - name: Upload Jepsen history + if: always() && matrix.workload != 'unit' uses: actions/upload-artifact@v7 with: name: jepsen-store-${{ matrix.workload }} path: /tmp/jepsen-store-types.tar.gz retention-days: 7 + if-no-files-found: ignore - name: Dump demo cluster log on failure if: failure() && matrix.workload != 'unit' && !startsWith(matrix.workload, 'filesystem-') && matrix.workload != 'sqlite' # The previous `tail -n 500` truncated a 3-minute workload's @@ -394,6 +464,7 @@ jobs: echo "=== ${node}: full log line count ===" wc -l "$log" || true done + tail -n 200 /tmp/elastickv-fs-n1-restarted.log || true - name: Upload filesystem cluster logs on failure if: failure() && (startsWith(matrix.workload, 'filesystem-') || matrix.workload == 'sqlite') uses: actions/upload-artifact@v7 @@ -403,6 +474,7 @@ jobs: /tmp/elastickv-fs-n1.log /tmp/elastickv-fs-n2.log /tmp/elastickv-fs-n3.log + /tmp/elastickv-fs-n1-restarted.log retention-days: 14 if-no-files-found: ignore - name: Stop demo cluster diff --git a/adapter/redis.go b/adapter/redis.go index 5d0bee2e9..0d558afad 100644 --- a/adapter/redis.go +++ b/adapter/redis.go @@ -422,6 +422,11 @@ func isTransientLeaderRedisError(err error) bool { if err == nil { return false } + // A proposal may have committed before leadership was lost. An unknown + // transaction outcome must not become a NOTLEADER retry instruction. + if errors.Is(err, kv.ErrTransactionOutcomeUnknown) { + return false + } if errors.Is(err, ErrLeaderNotFound) || errors.Is(err, ErrNotLeader) || errors.Is(err, kv.ErrLeaderNotFound) || diff --git a/adapter/redis_error_prefix_test.go b/adapter/redis_error_prefix_test.go index 9ce29e0c7..e81f1728f 100644 --- a/adapter/redis_error_prefix_test.go +++ b/adapter/redis_error_prefix_test.go @@ -40,6 +40,8 @@ func TestIsTransientLeaderRedisError(t *testing.T) { {"kv.ErrLeaderProxyCircuitOpen", kv.ErrLeaderProxyCircuitOpen, true}, {"raftengine.ErrNotLeader", raftengine.ErrNotLeader, true}, {"raftengine.ErrLeadershipLost", raftengine.ErrLeadershipLost, true}, + {"unknown outcome after leadership lost", + fmt.Errorf("%w: %w", kv.ErrTransactionOutcomeUnknown, raftengine.ErrLeadershipLost), false}, {"raftengine.ErrLeadershipTransferInProgress", raftengine.ErrLeadershipTransferInProgress, true}, {"wrapped with errors.WithStack", @@ -102,6 +104,9 @@ func TestWriteRedisError(t *testing.T) { {"fmt.Errorf-wrapped raftengine error preserves wrapping in message", fmt.Errorf("verify leader: %w", raftengine.ErrNotLeader), "NOTLEADER verify leader: raft engine: not leader"}, + {"unknown outcome is not marked safe for leader retry", + fmt.Errorf("%w: %w", kv.ErrTransactionOutcomeUnknown, raftengine.ErrLeadershipLost), + "transaction outcome unknown: raft engine: leadership lost"}, {"unrelated error untouched", errors.New("WRONGTYPE op"), "WRONGTYPE op"}, {"generic io.EOF untouched", diff --git a/docs/design/2026_10_26_fuse_strict_rollout_evidence.md b/docs/design/2026_10_26_fuse_strict_rollout_evidence.md new file mode 100644 index 000000000..35b259467 --- /dev/null +++ b/docs/design/2026_10_26_fuse_strict_rollout_evidence.md @@ -0,0 +1,319 @@ +# Strict FUSE durability and rollout evidence + +Status: validation in progress. Strict mode remains opt-in. + +## Scope of the current evidence + +The mounted tests use three Linux processes, three FUSE mounts, and one +three-voter Raft group. SQLite uses rollback journals on one mount. The tests +do not establish shared-database locking across mounts or SQLite WAL safety. + +## Transaction decisions + +An operation record is the transaction primary. Its generation and fenced +attempt IDs survive restart. A committed primary can outlive the coordinator's +response. Therefore an error returned by the primary commit call is not, by +itself, an abort decision. + +The coordinator now leaves prepared secondary intents intact when the primary +response is missing. Lock resolution reads the primary decision before +committing a secondary. A definitive, typed apply verdict or pre-append +leadership rejection can still abort the +prepared intents. Nil commit responses are unknown, including on the +single-group path. +The etcd Raft engine now returns `ErrLeadershipLost` for proposals still +pending when the leader steps down. Those proposals may have entered the log, +so `ErrNotLeader` is reserved for the pre-append check. +The Redis error adapter does not convert a wrapped unknown outcome into +`NOTLEADER`, which clients may interpret as permission to retry a write. + +`TestLostPrimaryReplyResolvesPreparedSecondaryFromDurableDecision` exercises +the lost-response path with two live Raft groups and their lock resolver. +`TestStrictLostPrimaryCommitReplyDoesNotAbortPreparedParticipants` and +`TestStrictDefinitivePrimaryConflictAbortsPreparedParticipants`, +`TestPrimaryRouteVerdictAbortsPreparedParticipants`, and +`TestPrimaryCommitDecisionClassification` cover the two error classes. +`TestJournalGenerationFenceSurvivesProcessKill` verifies that a +restarted store still rejects an old attempt after its durable fence. +`TestJournalCommittedAttemptDecisionSurvivesProcessKill` verifies that a +committed attempt, stored result, and effect survive a process kill; retrying +the same attempt returns that result without publishing another effect. +`TestMultiGroupPrimaryDecisionSurvivesProcessKill` runs two independent +single-voter Raft groups with Pebble stores, kills their shared process after +the primary COMMIT but before the secondary COMMIT, then reopens both groups. +The test invokes the restarted resolver's decision step directly; it reads +the primary decision and commits the prepared secondary. The test does not +exercise the periodic lock scan. It covers a precise cross-group crash +boundary but does not +model a three-voter group or a crash during WAL persistence. + +## Crash and mounted evidence + +The `filesystem-node-crash` job writes and fsyncs a file, fsyncs its parent +directory, kills one Raft node, checks the surviving mounts, restarts the +killed node with strict session recovery, and checks the restarted mount. +This job passed in [run 40319573770](https://github.com/bootjp/elastickv/actions/runs/40319573770). +The candidate extends the same job to acknowledged rename, chmod, unlink, +mkdir/rmdir, and truncate followed by file and directory fsyncs. The expanded +job passed in [run 40328639750](https://github.com/bootjp/elastickv/actions/runs/40328639750) +at `48755a5b`. + +The test establishes durability after successful fsync through one node +process crash. It does not inject a crash at each boundary between Raft WAL +persistence, FSM application, receipt publication, FUSE reply, and fsync. It +also does not crash a multi-group deployment. Those cases remain separate +acceptance requirements. + +All 23 jobs in [run 40320716183](https://github.com/bootjp/elastickv/actions/runs/40320716183) +passed at commit `a33dad1821631f8f43a6578aa3125712161108d5`, including the +mounted crash test, SQLite and filesystem workloads, and both benchmark jobs. +All 23 jobs also passed in +[run 40322309957](https://github.com/bootjp/elastickv/actions/runs/40322309957) +at commit `daebcb77fb4715f51c6a006870ebaa9fc5ec29a7`. + +## Performance evidence + +The first paired measurement used independent GitHub runners in +[run 40320716183](https://github.com/bootjp/elastickv/actions/runs/40320716183). +Each operation wrote 4 KiB and called `fsync`. The hot-inode case used four +writers with disjoint offsets and no client retry. + +| Case | Legacy ops/s | Strict ops/s | Legacy p99 | Strict p99 | +| --- | ---: | ---: | ---: | ---: | +| One writer, 40 operations | 105.9 | 40.8 | 12.7 ms | 30.7 ms | +| Four writers, 160 operations | 86.4 | 30.4 | 82.7 ms | 475.1 ms | + +The three-trial paired measurement in +[run 40322309957](https://github.com/bootjp/elastickv/actions/runs/40322309957) +used the same workload and commit on separate fresh runners. Across the three +trials, the median serial throughput was 59.2 ops/s in legacy mode and 36.1 +ops/s in strict mode; median serial p99 was 22.7 ms and 40.0 ms. The median +hot-inode throughput was 53.4 and 32.1 ops/s; median hot-inode p99 was 130.5 +ms and 250.2 ms. These comparisons are descriptive because the runners are +separate and scheduling noise is not controlled. + +These are observations, not performance limits. The workflow retains the JSON +artifacts for all trials. A rollout +limit must use repeated matched measurements and explicit maintainer approval, +as required by the parent design. The current hot-inode result warrants review +before choosing that limit. +The benchmark now records a per-worker throughput fairness ratio for the +hot-inode case. A separate two-group layout places operation records and file +effects on different Raft groups. Its first run +[40325784929](https://github.com/bootjp/elastickv/actions/runs/40325784929) +failed with a `txn locked` FUSE error. A subsequent run +[40327671052](https://github.com/bootjp/elastickv/actions/runs/40327671052) +returned EIO while the mount was up but the catalog watcher still reported a +connection refusal on the local Raft listener. The benchmark setup now waits +for both listeners and a successful mounted `stat`. Neither failed run yields +a valid two-group performance measurement. +Run [40328639750](https://github.com/bootjp/elastickv/actions/runs/40328639750) +reached both listeners and still failed with a typed `txn locked` error during +the first write. A deterministic two-group Raft test reproduced a second local +journal commit reaching PREPARE while the first held the mount session lock. +The same test passed after serializing this service's journal commits. This +does not serialize separate mounts and does not yet establish a passing +mounted two-group workload or an acceptable performance cost. +Run [40330054840](https://github.com/bootjp/elastickv/actions/runs/40330054840) +then failed while reading a session row locked by an in-flight journal +transaction. Journal reads and definitive prewrite lock rejections now retry +the same timestamp and attempt with a bounded backoff; a typed unknown +transaction outcome is never retried. The amended two-group benchmark passed +in [run 40331001888](https://github.com/bootjp/elastickv/actions/runs/40331001888) +at `fd9a2294`. Across three trials, median serial throughput was 67.1 +write/fsync operations per second with a 28.7 ms median of trial p99 values. +Four writers on one inode achieved 59.0 operations per second, 126.5 ms p99, +and a 0.954 median minimum-to-maximum worker throughput ratio. This was one +process with two single-voter Raft groups and one FUSE mount; it is not a +multi-mount or three-voter performance result. The expanded acknowledged +mutation node-crash job also passed at this SHA. +The same run's separate single-group runners measured median legacy serial +throughput of 113.0 operations per second (p50/p95/p99 8.2/12.0/14.3 ms) +and strict serial throughput of 28.3 (34.1/42.1/45.8 ms). For the hot inode, +legacy reached 79.0 operations per second (43.4/83.4/98.4 ms) and strict +reached 25.6 (136.5/242.5/309.4 ms). These are separate runners, so the +ratios are descriptive rather than a controlled regression threshold. The +two-group result uses a different one-voter topology and cannot be compared +directly to the three-voter single-group result. A two-group legacy job is +being added to supply a same-topology baseline. +The first paired two-group run +[40332154333](https://github.com/bootjp/elastickv/actions/runs/40332154333) +did not produce valid throughput results. The legacy hot-writer job reported +a primary commit timestamp mismatch on an open-handle fence. The strict job +reported operation-record errors during acknowledgement and subsequent writes +in the hot-writer trial. A possible duplicate-retirement path was identified: +the ACK path now checks the durable session retirement fence before consuming +the record's retained-byte reservation again. Its effect on the mounted +failure remains unverified. Both jobs must pass before the two-group +comparison can be used for rollout. +The long run +[40337556125](https://github.com/bootjp/elastickv/actions/runs/40337556125) +again failed the Redis workload and both two-group benchmarks at `f2cbf5cb`. +The strict benchmark exposed a gap between secondary PREPARE and primary +PREPARE: a foreground read treated the absent primary lock as an abort and +removed the prepared secondary. The resolver now keeps that secondary pending +until its TTL expires or a durable primary decision is visible. A two-group +Raft regression reproduces concurrent journal admissions; a second regression +checks the gap directly. The legacy benchmark exposed a primary identity +collision between distinct open references. Open and release now use the +per-reference key as their transaction primary. The Redis history classified +a transaction that lost its leader after dispatch as a definite failure, +although a later read observed its writes. The Jepsen client now records +that leadership outcome as indeterminate. The corrected candidate still +requires a complete CI run before the two-group measurements can be used. +The same long run's SQLite job also became inconclusive: operation journal +ACK accounting failed, subsequent operations hit the journal quota, and the +final database reads could not open their files. It did not report a +linearizability anomaly. The new resolver behavior may address the missing +session update, but this must be established by a passing long run. +In the first rerun at `ed5efef1`, the legacy two-group benchmark still +returned a typed `txn locked` error from a concurrent open. Open now waits +and rebuilds its transaction after that transient lock, within its existing +30-second retry budget; unknown outcomes still stop without replay. +The same rerun passed the Redis, SQLite, and strict two-group jobs. Its +remaining long-running rename job was superseded by the next full run. +Both two-group benchmark jobs passed in +[run 40345042339](https://github.com/bootjp/elastickv/actions/runs/40345042339) +at `711a7ba5`. Across three trials, the median legacy serial throughput was +112.1 ops/s with a 12.6 ms median trial p99; the strict values were 80.5 +ops/s and 39.5 ms. For four writers on one inode, the median legacy +throughput was 83.9 ops/s with an 88.6 ms p99 and 0.914 worker fairness; +the strict values were 54.9 ops/s, 276.3 ms, and 0.880. These jobs use +separate runners, so the comparison is descriptive. The full stress run +failed: Redis, Redis ZSet, and DynamoDB were cut off by a 720-second shell +timeout despite their requested 1800-second workload limit. The SQLite job +became inconclusive after operation journal ACK accounting errors and quota +exhaustion. The workflow now allows 3600 seconds for these protocol jobs, +including checker time. The numbers do not establish a rollout threshold. +The push test in +[run 40346594962](https://github.com/bootjp/elastickv/actions/runs/40346594962) +also failed a mounted strict-mode quota test when a FUSE request was canceled +after admission and before publishing its attempt decision. Every journaled +mutation now uses a separate bounded decision context after BeginAttempt; +request cancellation can no longer abandon that phase immediately. The +formerly failing mounted test passed ten consecutive runs in a local Linux +container with `/dev/fuse`, and the complete FUSE adapter package passed +there. The corrected push test +[40363982074](https://github.com/bootjp/elastickv/actions/runs/40363982074) +passed at `7323b1f9`; that SHA also includes a test-only fix for an unrelated +HLC renewal test race. + +The 30-minute stress run +[40362040528](https://github.com/bootjp/elastickv/actions/runs/40362040528) +at `3e3f4ab4` passed 24 of 25 jobs, including the formerly timed-out Redis, +Redis ZSet, and DynamoDB jobs. Its SQLite job remained inconclusive. The +first server-side failure was an operation-record error at sequence 1275; +ACK retirement then reported that a record's reservation exceeded the +session's retained usage. Several final SQLite reads could not open their +database files, so the checker returned `:unknown`. No linearizability +violation was reported, but this run does not close the SQLite gate or +establish a rollout threshold. + +The next scheduled strict run on main, +[40375070553](https://github.com/bootjp/elastickv/actions/runs/40375070553) +at `a299b49a`, passed SQLite but returned `:unknown` for filesystem +overwrite and rename. Their final reads failed with FUSE I/O errors. The +overwrite server also logged an exhausted HLC ceiling; the rename server +logged operation-record errors during ACK retirement. Neither checker +reported `:valid? false`. This main run predates the post-admission request +cancellation fix in the pull request, so its failures do not establish the +outcome of the corrected candidate. + +For a mode comparison, the targeted legacy run +[40378658363](https://github.com/bootjp/elastickv/actions/runs/40378658363) +at `2f542c4f` ran the same 30-minute overwrite, rename, and SQLite workloads +with strict mode disabled. All three checkers returned `:valid? true`. +The production Go code is unchanged between this commit and strict run +`40362040528`; the intervening changes are a test, this document, and +workflow inputs. Both runs used 512 keys with 16 writes per key for each +filesystem workload, and 128 keys with 16 writes per key for SQLite. The +strict SQLite run was inconclusive while its legacy counterpart passed. +These are separate runners and one trial per mode, so they do not quantify a +failure rate or prove a successful-operation consistency regression. They +provide evidence of an operational regression that blocks default-on strict +mode. Repeat the comparison after correcting the ACK accounting failure. +Require complete final reads in every strict checker. + +The strict mount now publishes `elastickv_fs_operation_journal_sessions`, +`elastickv_fs_operation_journal_pending`, and +`elastickv_fs_operation_journal_retained_bytes` from a fenced snapshot. These +gauges make pending work and retained storage visible during a canary. A +failed scan is logged; the last gauge value must not be treated as fresh +evidence after a scan failure. + +## Investigation and integration decision + +The ACK error is a correctness blocker for strict mode. +The guard in `reserveAckRetirement` checks a non-retired record after reading +its session, then rejects a reservation larger than the session's retained +usage. +It does not establish that a duplicate ACK caused the mismatch. +Increasing the quota, clamping the counters, or removing the guard would hide +the violation without establishing a valid retirement decision. + +First, identify the earliest transition that breaks the journal's accounting +invariants. +At a transactionally coherent state, the retained count and bytes must equal +the number and summed reservations of non-retired admitted records, and the +pending count must equal the number of pending records. +Account for a committed retirement fence whose record deletion is not yet +visible on another group before reporting a mismatch. +Capture the operation ID, session and generation, record and session versions, +snapshot and commit timestamps, route and group, retirement fence, and the +transaction decision at the first divergence. +The diagnostic must observe durable state without repairing or discarding it. + +Compare one-group and split-group layouts under concurrent admission, +generation advance, duplicate ACK, mount restart, and lost replies. +Pause or crash at PREPARE, the primary decision, and secondary apply to +distinguish mixed visibility from a lost update or recovery error. +The journal writes the record and session in one transaction with both keys +in the read set; a confirmed lost update requires investigation below the +journal arithmetic, including OCC and cross-group recovery. +Fix the first proven cause before changing the quota or optimizing ACKs. + +Investigate the HLC ceiling failure independently. +Record lease renewal, leader changes, group headroom, and workload timing. +Do not relax the timestamp fence to make a workload pass. +After the ACK fix, verify that cancellation and expired decision contexts do +not leave pending or retained work that fails to drain after recovery. + +Merging an opt-in implementation is separate from approving default-on strict +mode. +The runtime default remains disabled, and the strict scheduled Jepsen lane +continues to expose failures while this investigation proceeds. +The primary-decision and Raft error changes also affect non-strict +transactions, so merge review must assess those paths independently of the +strict rollout gate. +A merge does not establish that the long-running strict workload or the +multi-group recovery gate has passed. + +## Remaining rollout gates + +1. Reproduce the ACK mismatch, identify its first invalid transition, and + verify the smallest correction with one-group and split-group tests. + Require zero accounting invariant violations and eventual drain of pending + and retained work after recovery. +2. Exercise lost replies and process or node crashes at every publication and + barrier boundary, including a multi-group secondary crash and three-voter + primary and secondary recovery. +3. Run the race suite, Linux mount tests, and 30-minute overwrite, rename, and + SQLite Jepsen workloads on the same corrected candidate SHA with multiple + seeds. + Require complete final reads and `:valid? true` from every strict checker; + `:unknown` does not close the safety gate. + The scheduled workflow archives histories for successful and failed jobs. +4. Repeat the legacy and strict tests under matched conditions and state the + trial count before comparing failure rates. + Then benchmark single- and multi-group latency, throughput, conflicts, + fairness, and retained journal bytes. + Approve numerical canary limits from repeated measurements. +5. Run an opt-in canary with a verified backup and observable pending sessions. + Stop the canary on data mismatch, unverifiable fsync receipts, or unresolved + work that does not drain after recovery. + Resolve or cancel every pending operation and retire its session before a + supported downgrade. + +Default-on strict mode is blocked until these gates close. +A successful single-group smoke run cannot substitute for the multi-group +crash campaign. diff --git a/internal/filesystem/operation_contention.go b/internal/filesystem/operation_contention.go index 69ca3966c..e4478f330 100644 --- a/internal/filesystem/operation_contention.go +++ b/internal/filesystem/operation_contention.go @@ -13,6 +13,12 @@ func waitOperationCASRetry(ctx context.Context, retry int) error { return waitWriteConflictRetry(ctx, min(retry, operationCASMaxBackoffShift)) } +// An admitted attempt needs a durable decision even if its FUSE request is +// interrupted. The bounded context keeps publication and rejection live. +func operationDecisionContext(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), writeConflictRetryBudget) +} + func cancelledOperationError(status, contentionStatus uint8) error { if status == contentionStatus { return errors.WithStack(ErrContentionLimit) diff --git a/internal/filesystem/operation_crash_test.go b/internal/filesystem/operation_crash_test.go index 4530d577f..82d672bbf 100644 --- a/internal/filesystem/operation_crash_test.go +++ b/internal/filesystem/operation_crash_test.go @@ -9,17 +9,188 @@ import ( "syscall" "testing" + "github.com/bootjp/elastickv/kv" "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" ) const operationCrashDirectory = "ELASTICKV_OPERATION_CRASH_DIRECTORY" +const operationFenceCrashDirectory = "ELASTICKV_OPERATION_FENCE_CRASH_DIRECTORY" +const operationDecisionCrashDirectory = "ELASTICKV_OPERATION_DECISION_CRASH_DIRECTORY" type operationCrashEvidence struct { IDs []OperationID `json:"ids"` Inode uint64 `json:"inode"` } +type operationFenceCrashEvidence struct { + ID OperationID `json:"id"` + Attempt [16]byte `json:"attempt"` +} + +type crashReplayCoordinator struct { + *testCoordinator + dispatches int +} + +func (c *crashReplayCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + c.dispatches++ + return c.testCoordinator.Dispatch(ctx, req) +} + +func TestJournalCommittedAttemptDecisionSurvivesProcessKill(t *testing.T) { + if dir := os.Getenv(operationDecisionCrashDirectory); dir != "" { + runOperationDecisionCrashHelper(t, dir) + return + } + dir := t.TempDir() + cmd := exec.Command(os.Args[0], "-test.run=^TestJournalCommittedAttemptDecisionSurvivesProcessKill$") + cmd.Env = append(os.Environ(), operationDecisionCrashDirectory+"="+dir) + output, err := cmd.CombinedOutput() + var exit *exec.ExitError + require.ErrorAs(t, err, &exit, string(output)) + status, ok := exit.Sys().(syscall.WaitStatus) + require.True(t, ok, string(output)) + require.Equal(t, syscall.SIGKILL, status.Signal(), string(output)) + raw, err := os.ReadFile(filepath.Join(dir, "decision-evidence.json")) + require.NoError(t, err) + var evidence operationFenceCrashEvidence + require.NoError(t, json.Unmarshal(raw, &evidence)) + + st, err := store.NewPebbleStore(filepath.Join(dir, "pebble")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + coordinator := &crashReplayCoordinator{testCoordinator: &testCoordinator{st: st}} + svc, err := NewService(st, coordinator) + require.NoError(t, err) + journal, err := NewOperationJournal(svc) + require.NoError(t, err) + outcome, record, err := journal.ResolveAttempt(t.Context(), evidence.ID, evidence.Attempt) + require.NoError(t, err) + require.Equal(t, AttemptCommitted, outcome) + require.Equal(t, []byte("done"), record.Result) + beforeTS := st.LastCommitTS() + beforeVersion, exists, err := st.LatestCommitTS(t.Context(), []byte("data")) + require.NoError(t, err) + require.True(t, exists) + beforeValue, err := st.GetAt(t.Context(), []byte("data"), beforeTS) + require.NoError(t, err) + require.Equal(t, []byte("new"), beforeValue) + duplicate, err := journal.Publish(t.Context(), evidence.ID, evidence.Attempt, operationAttemptPlan{ + Intent: []byte("write"), Effects: []*kv.Elem[kv.OP]{{Op: kv.Put, Key: []byte("data"), Value: []byte("new")}}, Result: []byte("done"), + }) + require.NoError(t, err) + require.Equal(t, record, duplicate) + require.Zero(t, coordinator.dispatches, "replay must not dispatch another mutation") + require.Equal(t, beforeTS, st.LastCommitTS(), "replay must not publish another mutation") + afterVersion, exists, err := st.LatestCommitTS(t.Context(), []byte("data")) + require.NoError(t, err) + require.True(t, exists) + require.Equal(t, beforeVersion, afterVersion) + value, err := st.GetAt(t.Context(), []byte("data"), st.LastCommitTS()) + require.NoError(t, err) + require.Equal(t, []byte("new"), value) +} + +func runOperationDecisionCrashHelper(t *testing.T, dir string) { + ctx := context.Background() + st, err := store.NewPebbleStore(filepath.Join(dir, "pebble")) + require.NoError(t, err) + svc, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(svc) + require.NoError(t, err) + id := testOperationID(t, journal, 1) + _, err = journal.Admit(ctx, id, []byte("write"), 128, 1024) + require.NoError(t, err) + attempt, err := journal.BeginAttempt(ctx, id) + require.NoError(t, err) + _, err = journal.Publish(ctx, id, attempt.AttemptID, operationAttemptPlan{ + Intent: []byte("write"), Effects: []*kv.Elem[kv.OP]{{Op: kv.Put, Key: []byte("data"), Value: []byte("new")}}, Result: []byte("done"), + }) + require.NoError(t, err) + raw, err := json.Marshal(operationFenceCrashEvidence{ID: id, Attempt: attempt.AttemptID}) + require.NoError(t, err) + fileOnDisk, err := os.OpenFile(filepath.Join(dir, "decision-evidence.json"), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + require.NoError(t, err) + _, err = fileOnDisk.Write(raw) + require.NoError(t, err) + require.NoError(t, fileOnDisk.Sync()) + require.NoError(t, fileOnDisk.Close()) + require.NoError(t, syscall.Kill(os.Getpid(), syscall.SIGKILL)) +} + +func TestJournalGenerationFenceSurvivesProcessKill(t *testing.T) { + if dir := os.Getenv(operationFenceCrashDirectory); dir != "" { + runOperationFenceCrashHelper(t, dir) + return + } + dir := t.TempDir() + cmd := exec.Command(os.Args[0], "-test.run=^TestJournalGenerationFenceSurvivesProcessKill$") + cmd.Env = append(os.Environ(), operationFenceCrashDirectory+"="+dir) + output, err := cmd.CombinedOutput() + var exit *exec.ExitError + require.ErrorAs(t, err, &exit, string(output)) + status, ok := exit.Sys().(syscall.WaitStatus) + require.True(t, ok, string(output)) + require.Equal(t, syscall.SIGKILL, status.Signal(), string(output)) + raw, err := os.ReadFile(filepath.Join(dir, "fence-evidence.json")) + require.NoError(t, err) + var evidence operationFenceCrashEvidence + require.NoError(t, json.Unmarshal(raw, &evidence)) + + st, err := store.NewPebbleStore(filepath.Join(dir, "pebble")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + svc, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(svc) + require.NoError(t, err) + outcome, record, err := journal.ResolveAttempt(t.Context(), evidence.ID, evidence.Attempt) + require.NoError(t, err) + require.Equal(t, AttemptAborted, outcome) + require.Contains(t, record.FencedAttempts, evidence.Attempt) + _, err = journal.Publish(t.Context(), evidence.ID, evidence.Attempt, operationAttemptPlan{ + Intent: []byte("write"), Effects: []*kv.Elem[kv.OP]{{Op: kv.Put, Key: []byte("data"), Value: []byte("old")}}, + }) + require.ErrorIs(t, err, ErrOperationGeneration) + _, err = st.GetAt(t.Context(), []byte("data"), st.LastCommitTS()) + require.ErrorIs(t, err, store.ErrKeyNotFound) + _, err = journal.Publish(t.Context(), evidence.ID, record.AttemptID, operationAttemptPlan{ + Intent: []byte("write"), Effects: []*kv.Elem[kv.OP]{{Op: kv.Put, Key: []byte("data"), Value: []byte("new")}}, + }) + require.NoError(t, err) + value, err := st.GetAt(t.Context(), []byte("data"), st.LastCommitTS()) + require.NoError(t, err) + require.Equal(t, []byte("new"), value) +} + +func runOperationFenceCrashHelper(t *testing.T, dir string) { + ctx := context.Background() + st, err := store.NewPebbleStore(filepath.Join(dir, "pebble")) + require.NoError(t, err) + svc, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(svc) + require.NoError(t, err) + id := testOperationID(t, journal, 1) + _, err = journal.Admit(ctx, id, []byte("write"), 128, 1024) + require.NoError(t, err) + attempt, err := journal.BeginAttempt(ctx, id) + require.NoError(t, err) + _, err = journal.AdvanceGeneration(ctx, id, attempt.AttemptID) + require.NoError(t, err) + raw, err := json.Marshal(operationFenceCrashEvidence{ID: id, Attempt: attempt.AttemptID}) + require.NoError(t, err) + fileOnDisk, err := os.OpenFile(filepath.Join(dir, "fence-evidence.json"), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + require.NoError(t, err) + _, err = fileOnDisk.Write(raw) + require.NoError(t, err) + require.NoError(t, fileOnDisk.Sync()) + require.NoError(t, fileOnDisk.Close()) + require.NoError(t, syscall.Kill(os.Getpid(), syscall.SIGKILL)) +} + func TestJournalReceiptsSurviveProcessKill(t *testing.T) { if dir := os.Getenv(operationCrashDirectory); dir != "" { runOperationCrashHelper(t, dir) diff --git a/internal/filesystem/operation_create.go b/internal/filesystem/operation_create.go index 015d4d90a..42ed89aa3 100644 --- a/internal/filesystem/operation_create.go +++ b/internal/filesystem/operation_create.go @@ -89,7 +89,7 @@ func createEffectLimit(parent uint64, name, clientID []byte) (uint32, error) { } func (s *Service) executeCreateOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, parent uint64, name []byte, typ FileType, opts CreateOptions, intent []byte) (CreateResult, error) { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { result, err := s.publishCreateOperationAttempt(deadlineCtx, journal, id, attempt.AttemptID, parent, name, typ, opts, intent) diff --git a/internal/filesystem/operation_journal.go b/internal/filesystem/operation_journal.go index 5e006910b..26d345766 100644 --- a/internal/filesystem/operation_journal.go +++ b/internal/filesystem/operation_journal.go @@ -37,12 +37,16 @@ func (b serviceOperationBackend) Snapshot(ctx context.Context) (uint64, error) { } func (b serviceOperationBackend) GetAt(ctx context.Context, key []byte, ts uint64) ([]byte, error) { - value, err := b.service.store.GetAt(ctx, key, ts) + value, err := retryJournalLocked(ctx, func(ctx context.Context) ([]byte, error) { + return b.service.store.GetAt(ctx, key, ts) + }) return value, errors.Wrap(err, "read operation journal") } func (b serviceOperationBackend) ScanAt(ctx context.Context, start, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { - page, err := b.service.store.ScanAt(ctx, start, end, limit, ts) + page, err := retryJournalLocked(ctx, func(ctx context.Context) ([]*store.KVPair, error) { + return b.service.store.ScanAt(ctx, start, end, limit, ts) + }) return page, errors.Wrap(err, "scan operation journal") } @@ -83,7 +87,39 @@ func (b serviceOperationBackend) commit(ctx context.Context, ts uint64, elems [] } primary = elem.Key } - return b.service.dispatchTxnWithPrimaryAtRoute(ctx, ts, elems, readKeys, true, primary, observed) + if err := ctx.Err(); err != nil { + return nil, errors.WithStack(err) + } + select { + case <-b.service.journalCommitPermit: + defer func() { b.service.journalCommitPermit <- struct{}{} }() + case <-ctx.Done(): + return nil, errors.WithStack(ctx.Err()) + } + return retryJournalLocked(ctx, func(ctx context.Context) (*kv.CoordinateResponse, error) { + return b.service.dispatchTxnWithPrimaryAtRoute(ctx, ts, elems, readKeys, true, primary, observed) + }) +} + +// A lock held by another journal transaction is a transient dependency. The +// caller retains the same snapshot and attempt identity while the owner +// finishes. An unknown transaction outcome is never retried here. +func retryJournalLocked[T any](ctx context.Context, call func(context.Context) (T, error)) (T, error) { + var result T + for retry := range operationCASRetries { + value, err := call(ctx) + if !errors.Is(err, kv.ErrTxnLocked) || errors.Is(err, kv.ErrTransactionOutcomeUnknown) { + return value, err + } + result = value + if retry == operationCASRetries-1 { + return result, err + } + if err = waitOperationCASRetry(ctx, retry); err != nil { + return result, err + } + } + return result, errors.WithStack(ErrContentionLimit) } func (b serviceOperationBackend) CommittedVersionAt(ctx context.Context, key []byte, ts uint64) (bool, error) { @@ -399,7 +435,7 @@ func (j *OperationJournal) loadSessionAt(ctx context.Context, id OperationSessio } session, err := decodeOperationSession(raw) if err != nil || session.ID != id { - return operationSession{}, false, errors.WithStack(ErrOperationRecordCorrupt) + return operationSession{}, false, errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session row") } return session, true, nil } @@ -414,7 +450,7 @@ func (j *OperationJournal) loadRecordAt(ctx context.Context, id OperationID, ts } record, err := decodeOperationRecord(raw) if err != nil || record.ID != id { - return operationRecord{}, false, errors.WithStack(ErrOperationRecordCorrupt) + return operationRecord{}, false, errors.Wrap(ErrOperationRecordCorrupt, "invalid operation row") } return record, true, nil } diff --git a/internal/filesystem/operation_journal_exec.go b/internal/filesystem/operation_journal_exec.go index b807cff0c..f9c675d90 100644 --- a/internal/filesystem/operation_journal_exec.go +++ b/internal/filesystem/operation_journal_exec.go @@ -461,7 +461,7 @@ func (j *OperationJournal) resolveRecordAt(ctx context.Context, id OperationID, return record, errors.WithStack(ErrOperationPending) } if record.CommitTS == 0 { - return operationRecord{}, errors.WithStack(ErrOperationRecordCorrupt) + return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, "terminal operation has no commit timestamp") } return record, nil } @@ -623,7 +623,7 @@ func (j *OperationJournal) retireUnadmittedOnce(ctx context.Context, id Operatio } raw, err := encodeOperationSession(session) if err != nil { - return err + return errors.Wrap(err, "encode unadmitted retirement session") } _, err = j.backend.Commit(ctx, ts, []*kv.Elem[kv.OP]{{Op: kv.Put, Key: session.ID.key(), Value: raw}}, [][]byte{id.key(), session.ID.key()}) @@ -644,7 +644,7 @@ func (j *OperationJournal) acknowledgeOnce(ctx context.Context, id OperationID) } raw, err := encodeOperationSession(session) if err != nil { - return err + return errors.Wrap(err, "encode acknowledged session") } elems := []*kv.Elem[kv.OP]{ {Op: kv.Del, Key: id.key()}, @@ -659,8 +659,13 @@ func (j *OperationJournal) acknowledgeOnce(ctx context.Context, id OperationID) } func reserveAckRetirement(session *operationSession, record operationRecord) error { + // The session fence may become visible before the primary record deletion + // on another group. A repeated ACK must not consume the reservation twice. + if session.retired(record.ID.Sequence) { + return errors.WithStack(ErrOperationRetired) + } if session.RetainedCount == 0 || session.RetainedBytes < record.Reservation { - return errors.WithStack(ErrOperationRecordCorrupt) + return errors.Wrap(ErrOperationRecordCorrupt, "ack reservation exceeds retained usage") } if err := advanceAckWatermark(session, record.ID.Sequence); err != nil { return err diff --git a/internal/filesystem/operation_journal_multigroup_test.go b/internal/filesystem/operation_journal_multigroup_test.go new file mode 100644 index 000000000..3ffa7f2ca --- /dev/null +++ b/internal/filesystem/operation_journal_multigroup_test.go @@ -0,0 +1,306 @@ +package filesystem + +import ( + "bytes" + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/fskeys" + "github.com/bootjp/elastickv/internal/raftengine" + etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd" + "github.com/bootjp/elastickv/kv" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +type holdPreparedTxn struct { + kv.Transactional + prepared chan struct{} + release chan struct{} + once sync.Once +} + +type synchronizeSessionReadsBackend struct { + operationJournalBackend + sessionKey []byte + reads atomic.Int32 + both chan struct{} +} + +func (b *synchronizeSessionReadsBackend) GetAt(ctx context.Context, key []byte, ts uint64) ([]byte, error) { + value, err := b.operationJournalBackend.GetAt(ctx, key, ts) + if !bytes.Equal(key, b.sessionKey) { + return value, err + } + count := b.reads.Add(1) + if count == 2 { + close(b.both) + } + if count <= 2 { + select { + case <-b.both: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return value, err +} + +func (h *holdPreparedTxn) Commit(ctx context.Context, reqs []*pb.Request) (*kv.TransactionResponse, error) { + response, err := h.Transactional.Commit(ctx, reqs) + if err == nil && len(reqs) == 1 && reqs[0].Phase == pb.Phase_PREPARE { + h.once.Do(func() { + close(h.prepared) + <-h.release + }) + } + return response, err +} + +func TestOperationJournalTwoGroupCreateOpenWrite(t *testing.T) { + routes := distribution.NewEngine() + require.NoError(t, routes.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte{}, End: []byte("!fs|op|"), GroupID: 1, State: distribution.RouteStateActive}, + {RouteID: 2, Start: []byte("!fs|op|"), GroupID: 2, State: distribution.RouteStateActive}, + }, + })) + factory := etcdraftengine.NewFactory(etcdraftengine.FactoryConfig{ + TickInterval: 10 * time.Millisecond, HeartbeatTick: 1, ElectionTick: 10, + MaxSizePerMsg: 1 << 20, MaxInflightMsg: 256, + }) + groups := make(map[uint64]*kv.ShardGroup, 2) + for _, gid := range []uint64{1, 2} { + st := store.NewMVCCStore() + id := fmt.Sprintf("fs-journal-g%d", gid) + result, err := factory.Create(raftengine.FactoryConfig{ + LocalID: id, LocalAddress: id, DataDir: t.TempDir(), Bootstrap: true, + StateMachine: kv.NewKvFSMWithHLC(st, kv.NewHLC()), + }) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, result.Engine.Close()) + if result.Close != nil { + require.NoError(t, result.Close()) + } + }) + require.Eventually(t, func() bool { return result.Engine.State() == raftengine.StateLeader }, + 5*time.Second, 10*time.Millisecond) + groups[gid] = &kv.ShardGroup{Engine: result.Engine, Store: st, Txn: kv.NewLeaderProxyWithEngine(result.Engine)} + } + shards := kv.NewShardStore(routes, groups) + coord := kv.NewShardedCoordinator(routes, groups, 1, kv.NewHLC(), shards) + svc, err := NewService(shards, coord) + require.NoError(t, err) + ctx := context.Background() + require.NoError(t, svc.InitializeRoot(ctx, testRootMode, 1000, 1000)) + journal, err := NewOperationJournal(svc) + require.NoError(t, err) + filesystemID, err := journal.EnsureFilesystem(ctx) + require.NoError(t, err) + session, err := journal.RegisterSession(ctx, filesystemID, "benchmark-mount") + require.NoError(t, err) + id := OperationID{Filesystem: session.Filesystem, Session: session.Session, Sequence: 1} + file, err := svc.CreateWithOperation(ctx, id, RootInode, []byte("benchmark.dat"), CreateOptions{Mode: testFileMode}) + require.NoError(t, err) + createID := id + id.Sequence++ + fh, err := svc.OpenWithOperation(ctx, id, file.Inode, []byte("benchmark-mount")) + require.NoError(t, err) + held := &holdPreparedTxn{Transactional: groups[1].Txn, prepared: make(chan struct{}), release: make(chan struct{})} + groups[1].Txn = held + id.Sequence++ + assertWriteAfterConcurrentAcknowledgement(t, ctx, journal, svc, createID, id, file.Inode, held) + assertConcurrentJournalAdmissions(t, ctx, journal, session) + id.Sequence++ + require.NoError(t, svc.ReleaseWithOperation(ctx, id, file.Inode, fh, []byte("benchmark-mount"))) + assertJournalCommitsDoNotCollideOnPreparedSession(t, ctx, svc, groups[1], session) +} + +func assertConcurrentJournalAdmissions(t *testing.T, ctx context.Context, journal *OperationJournal, session OperationSessionID) { + t.Helper() + backend := &synchronizeSessionReadsBackend{ + operationJournalBackend: journal.backend, + sessionKey: session.key(), + both: make(chan struct{}), + } + journal = newOperationJournalWithBackend(backend) + var sequence atomic.Uint64 + sequence.Store(4) + var workers sync.WaitGroup + errorsCh := make(chan error, 2) + for worker := range 2 { + workers.Add(1) + go func() { + defer workers.Done() + for index := range 4 { + id := OperationID{Filesystem: session.Filesystem, Session: session.Session, Sequence: sequence.Add(1)} + _, err := journal.Admit(ctx, id, []byte{byte(worker), byte(index)}, 16, 1024) + if err != nil { + errorsCh <- fmt.Errorf("admit operation %d: %w", id.Sequence, err) + return + } + } + }() + } + workers.Wait() + close(errorsCh) + for err := range errorsCh { + require.NoError(t, err) + } + ts, err := journal.backend.Snapshot(ctx) + require.NoError(t, err) + actual, exists, err := journal.loadSessionAt(ctx, session, ts) + require.NoError(t, err) + require.True(t, exists) + require.Equal(t, uint32(8), actual.Pending) +} + +func assertJournalCommitsDoNotCollideOnPreparedSession(t *testing.T, ctx context.Context, svc *Service, group *kv.ShardGroup, session OperationSessionID) { + t.Helper() + held := &holdPreparedTxn{Transactional: group.Txn, prepared: make(chan struct{}), release: make(chan struct{})} + group.Txn = held + backend := serviceOperationBackend{service: svc} + sessionKey := fskeys.OperationSessionKey(session.Filesystem, session.Session) + commitWithContext := func(commitCtx context.Context, sequence uint64) error { + ts, err := backend.Snapshot(commitCtx) + if err != nil { + return err + } + primary := fskeys.OperationKey(session.Filesystem, session.Session, sequence) + _, err = backend.Commit(commitCtx, ts, []*kv.Elem[kv.OP]{ + {Op: kv.Put, Key: primary, Value: []byte{1}}, + {Op: kv.Put, Key: sessionKey, Value: []byte{2}}, + }, [][]byte{primary, sessionKey}) + return err + } + commit := func(sequence uint64) error { return commitWithContext(ctx, sequence) } + first := make(chan error, 1) + go func() { first <- commit(1000) }() + select { + case <-held.prepared: + case <-time.After(5 * time.Second): + t.Fatal("first journal commit did not prepare the session") + } + second := make(chan error, 1) + go func() { second <- commit(1001) }() + select { + case err := <-second: + close(held.release) + t.Fatalf("second journal commit returned during prepared session lock: %v", err) + case <-time.After(100 * time.Millisecond): + } + assertJournalCommitWaiterCanCancel(t, ctx, held, commitWithContext) + close(held.release) + select { + case err := <-first: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("first journal commit did not finish") + } + select { + case err := <-second: + require.NotErrorIs(t, err, kv.ErrTxnLocked) + case <-time.After(5 * time.Second): + t.Fatal("second journal commit did not finish") + } +} + +func assertJournalCommitWaiterCanCancel(t *testing.T, ctx context.Context, held *holdPreparedTxn, commit func(context.Context, uint64) error) { + t.Helper() + cancelCtx, cancel := context.WithCancel(ctx) + cancelled := make(chan error, 1) + go func() { cancelled <- commit(cancelCtx, 1002) }() + select { + case err := <-cancelled: + close(held.release) + t.Fatalf("journal commit returned before its waiter was cancelled: %v", err) + case <-time.After(50 * time.Millisecond): + } + cancel() + select { + case err := <-cancelled: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(5 * time.Second): + close(held.release) + t.Fatal("cancelled journal commit remained blocked") + } +} + +func assertWriteAfterConcurrentAcknowledgement(t *testing.T, ctx context.Context, journal *OperationJournal, svc *Service, createID, writeID OperationID, inode uint64, held *holdPreparedTxn) { + t.Helper() + ackDone := make(chan error, 1) + go func() { ackDone <- journal.Acknowledge(ctx, createID) }() + select { + case <-held.prepared: + case <-time.After(5 * time.Second): + t.Fatal("acknowledgement did not prepare its session participant") + } + type writeResult struct { + count int + err error + } + writeDone := make(chan writeResult, 1) + go func() { + count, writeErr := svc.WriteWithOperation(ctx, writeID, inode, 0, []byte("data")) + writeDone <- writeResult{count, writeErr} + }() + select { + case result := <-writeDone: + close(held.release) + t.Fatalf("write returned while the acknowledgement held the prepared session lock: %v", result.err) + case <-time.After(100 * time.Millisecond): + } + close(held.release) + select { + case err := <-ackDone: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("acknowledgement did not complete") + } + select { + case result := <-writeDone: + require.NoError(t, result.err) + require.Equal(t, 4, result.count) + case <-time.After(5 * time.Second): + t.Fatal("write did not complete after acknowledgement") + } +} + +func TestRetryJournalLockedKeepsUnknownOutcome(t *testing.T) { + tries := 0 + value, err := retryJournalLocked(context.Background(), func(context.Context) ([]byte, error) { + tries++ + if tries < 3 { + return nil, kv.ErrTxnLocked + } + return []byte("visible"), nil + }) + require.NoError(t, err) + require.Equal(t, []byte("visible"), value) + require.Equal(t, 3, tries) + + tries = 0 + _, err = retryJournalLocked(context.Background(), func(context.Context) ([]byte, error) { + tries++ + return nil, fmt.Errorf("%w: %w", kv.ErrTransactionOutcomeUnknown, kv.ErrTxnLocked) + }) + require.ErrorIs(t, err, kv.ErrTransactionOutcomeUnknown) + require.Equal(t, 1, tries) +} + +func TestRepeatedAcknowledgementKeepsReservation(t *testing.T) { + session := operationSession{RejectBefore: 1, AckGaps: []uint64{3}, RetainedCount: 1, RetainedBytes: 100} + record := operationRecord{ID: OperationID{Sequence: 3}, Reservation: 10} + require.ErrorIs(t, reserveAckRetirement(&session, record), ErrOperationRetired) + require.Equal(t, []uint64{3}, session.AckGaps) + require.Equal(t, uint32(1), session.RetainedCount) + require.Equal(t, uint64(100), session.RetainedBytes) +} diff --git a/internal/filesystem/operation_lease.go b/internal/filesystem/operation_lease.go index 9ccf3e99a..2e4ff11aa 100644 --- a/internal/filesystem/operation_lease.go +++ b/internal/filesystem/operation_lease.go @@ -48,7 +48,7 @@ func (s *Service) RefreshOpenHandleLeaseWithOperation(ctx context.Context, id Op } func (s *Service) executeLeaseOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, inode, fh uint64, clientID, intent []byte) error { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { err := s.publishLeaseOperationAttempt(deadlineCtx, journal, id, attempt.AttemptID, inode, fh, clientID, intent) diff --git a/internal/filesystem/operation_open.go b/internal/filesystem/operation_open.go index 6ddf3496a..56e649d09 100644 --- a/internal/filesystem/operation_open.go +++ b/internal/filesystem/operation_open.go @@ -60,7 +60,7 @@ func (s *Service) OpenWithOperation(ctx context.Context, id OperationID, inode u } func (s *Service) executeOpenOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, inode uint64, clientID, intent []byte) (uint64, error) { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { fh, err := s.publishOpenOperationAttempt(deadlineCtx, journal, id, attempt.AttemptID, inode, clientID, intent) diff --git a/internal/filesystem/operation_release.go b/internal/filesystem/operation_release.go index 7ff325f89..c4848a0c0 100644 --- a/internal/filesystem/operation_release.go +++ b/internal/filesystem/operation_release.go @@ -66,7 +66,7 @@ func (s *Service) admitHandleOperation(ctx context.Context, id OperationID, inte } func (s *Service) executeReleaseOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, inode, fh uint64, clientID, intent []byte) error { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { err := s.publishReleaseOperationAttempt(deadlineCtx, journal, id, attempt.AttemptID, inode, fh, clientID, intent) diff --git a/internal/filesystem/operation_rename.go b/internal/filesystem/operation_rename.go index 9711b2e28..4c7be2b04 100644 --- a/internal/filesystem/operation_rename.go +++ b/internal/filesystem/operation_rename.go @@ -67,7 +67,7 @@ func (s *Service) RenameWithOperation(ctx context.Context, id OperationID, oldPa } func (s *Service) executeRenameOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, parent uint64, oldName, newName, intent []byte) error { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { err := s.publishRenameOperationAttempt(deadlineCtx, journal, id, attempt.AttemptID, parent, oldName, newName, intent) diff --git a/internal/filesystem/operation_session.go b/internal/filesystem/operation_session.go index ff5a561ae..0868e3131 100644 --- a/internal/filesystem/operation_session.go +++ b/internal/filesystem/operation_session.go @@ -81,8 +81,14 @@ func decodeOperationSession(raw []byte) (operationSession, error) { } func validateOperationSession(session operationSession) error { - if !validOperationSessionIdentity(session) || !validOperationSessionUsage(session) || !validOperationAckGaps(session) { - return errors.WithStack(ErrOperationRecordCorrupt) + if !validOperationSessionIdentity(session) { + return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session identity") + } + if !validOperationSessionUsage(session) { + return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session usage") + } + if !validOperationAckGaps(session) { + return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session acknowledgement gaps") } return nil } diff --git a/internal/filesystem/operation_setattr.go b/internal/filesystem/operation_setattr.go index 8ef43fe02..e5054f3a1 100644 --- a/internal/filesystem/operation_setattr.go +++ b/internal/filesystem/operation_setattr.go @@ -61,7 +61,7 @@ func (s *Service) SetAttrWithOperation(ctx context.Context, id OperationID, inod } func (s *Service) executeSetAttrOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, inode uint64, mask SetAttrMask, attrs SetAttr, intent []byte) (Stat, error) { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { stat, err := s.publishSetAttrOperationAttempt(deadlineCtx, journal, id, attempt.AttemptID, inode, mask, attrs, intent) diff --git a/internal/filesystem/operation_stats.go b/internal/filesystem/operation_stats.go new file mode 100644 index 000000000..5373971f0 --- /dev/null +++ b/internal/filesystem/operation_stats.go @@ -0,0 +1,44 @@ +package filesystem + +import ( + "context" + + "github.com/bootjp/elastickv/internal/fskeys" + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" +) + +// OperationJournalStats is a snapshot of retained strict-mode work across +// every mount session. Pending and retained bytes must drain before rollback. +type OperationJournalStats struct { + Sessions uint64 + ClosedSessions uint64 + Pending uint64 + RetainedCount uint64 + RetainedBytes uint64 + MaxBytes uint64 +} + +func (s *Service) ListOperationJournalStats(ctx context.Context) (OperationJournalStats, error) { + ts, err := s.readTS(ctx) + if err != nil { + return OperationJournalStats{}, errors.Wrap(err, "snapshot operation journal usage") + } + var stats OperationJournalStats + err = s.scanVisiblePrefix(ctx, fskeys.OperationSessionAllPrefix(), ts, func(pair *store.KVPair) error { + session, decodeErr := decodeOperationSession(pair.Value) + if decodeErr != nil { + return decodeErr + } + stats.Sessions++ + if session.Closed { + stats.ClosedSessions++ + } + stats.Pending += uint64(session.Pending) + stats.RetainedCount += uint64(session.RetainedCount) + stats.RetainedBytes += session.RetainedBytes + stats.MaxBytes += session.MaxBytes + return nil + }) + return stats, errors.Wrap(err, "scan operation journal usage") +} diff --git a/internal/filesystem/operation_stats_test.go b/internal/filesystem/operation_stats_test.go new file mode 100644 index 000000000..884125e00 --- /dev/null +++ b/internal/filesystem/operation_stats_test.go @@ -0,0 +1,44 @@ +package filesystem + +import ( + "testing" + + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func TestOperationJournalStatsTrackRetainedWorkAcrossSessions(t *testing.T) { + st := store.NewMVCCStore() + svc, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(svc) + require.NoError(t, err) + filesystemID, err := journal.EnsureFilesystem(t.Context()) + require.NoError(t, err) + first, err := journal.RegisterSession(t.Context(), filesystemID, "first") + require.NoError(t, err) + _, err = journal.RegisterSession(t.Context(), filesystemID, "second") + require.NoError(t, err) + id := OperationID{Filesystem: first.Filesystem, Session: first.Session, Sequence: 1} + _, err = journal.Admit(t.Context(), id, []byte("write"), 128, 1024) + require.NoError(t, err) + + stats, err := svc.ListOperationJournalStats(t.Context()) + require.NoError(t, err) + require.EqualValues(t, 2, stats.Sessions) + require.EqualValues(t, 1, stats.Pending) + require.EqualValues(t, 1, stats.RetainedCount) + require.Positive(t, stats.RetainedBytes) + require.EqualValues(t, 2*defaultOperationQuota, stats.MaxBytes) + + _, err = journal.Cancel(t.Context(), id) + require.NoError(t, err) + require.NoError(t, journal.Acknowledge(t.Context(), id)) + require.NoError(t, journal.CloseSession(t.Context(), first)) + stats, err = svc.ListOperationJournalStats(t.Context()) + require.NoError(t, err) + require.EqualValues(t, 1, stats.ClosedSessions) + require.Zero(t, stats.Pending) + require.Zero(t, stats.RetainedCount) + require.Zero(t, stats.RetainedBytes) +} diff --git a/internal/filesystem/operation_unlink.go b/internal/filesystem/operation_unlink.go index 49868b107..04a3c602f 100644 --- a/internal/filesystem/operation_unlink.go +++ b/internal/filesystem/operation_unlink.go @@ -64,7 +64,7 @@ func (s *Service) unlinkNodeWithOperation(ctx context.Context, id OperationID, p } func (s *Service) executeUnlinkOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, parent uint64, name []byte, directory bool, intent []byte) error { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { err := s.publishUnlinkOperationAttempt(deadlineCtx, journal, id, attempt.AttemptID, parent, name, directory, intent) diff --git a/internal/filesystem/operation_write.go b/internal/filesystem/operation_write.go index 08e97434c..b4931a451 100644 --- a/internal/filesystem/operation_write.go +++ b/internal/filesystem/operation_write.go @@ -103,7 +103,7 @@ func (s *Service) writeOperationEffectLimit(ctx context.Context, inode, offset, } func (s *Service) executeWriteOperation(ctx context.Context, journal *OperationJournal, id OperationID, attempt operationRecord, inode, offset, end uint64, data, intent []byte) (int, error) { - deadlineCtx, cancel := context.WithTimeout(ctx, writeConflictRetryBudget) + deadlineCtx, cancel := operationDecisionContext(ctx) defer cancel() for retry := range writeConflictRetryLimit { if err := deadlineCtx.Err(); err != nil { diff --git a/internal/filesystem/operation_write_test.go b/internal/filesystem/operation_write_test.go index aeb564c22..a2f2f1c01 100644 --- a/internal/filesystem/operation_write_test.go +++ b/internal/filesystem/operation_write_test.go @@ -222,6 +222,34 @@ func TestWriteWithOperationReplansAfterDefinitiveConflict(t *testing.T) { require.Equal(t, []byte("aZ"), data) } +func TestWriteWithOperationFinishesDecisionAfterRequestCancellation(t *testing.T) { + requestCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc := newTestServiceWithOptions(t, []uint64{2}) + require.NoError(t, svc.InitializeRoot(requestCtx, testRootMode, 1000, 1000)) + file, err := svc.Create(requestCtx, RootInode, []byte("file"), CreateOptions{Mode: testFileMode}) + require.NoError(t, err) + id := testWriteOperationID(t, svc, 1) + inner, ok := svc.dispatch.(*testCoordinator) + require.True(t, ok) + fault := &writeOperationFaultDispatcher{inner: inner, key: fskeys.ChunkKey(file.Inode, file.Inode, 0)} + fault.before = func(context.Context) error { + cancel() + return nil + } + svc.dispatch = fault + + n, err := svc.WriteWithOperation(requestCtx, id, file.Inode, 0, []byte("a")) + require.NoError(t, err) + require.Equal(t, 1, n) + require.True(t, fault.once) + j, err := NewOperationJournal(svc) + require.NoError(t, err) + record, err := j.Resolve(context.Background(), id) + require.NoError(t, err) + require.Equal(t, OperationCommitted, record.State) +} + func TestWriteWithOperationResolvesLostCommitReply(t *testing.T) { ctx := context.Background() svc := newTestServiceWithOptions(t, []uint64{2}) diff --git a/internal/filesystem/service.go b/internal/filesystem/service.go index d2ae05037..4d39615d3 100644 --- a/internal/filesystem/service.go +++ b/internal/filesystem/service.go @@ -252,16 +252,19 @@ type LeaseReapStats struct { } type Service struct { - store store.MVCCStore - dispatch Dispatcher - chunkSize uint64 - openLeaseTTL time.Duration - capacity uint64 - maxFiles uint64 - now func() time.Time - allocID func() (uint64, error) - homeSlot func(uint64) uint64 - observer OperationalObserver + // Journal commits within one mount share a session row. Keep a later + // two-phase prepare from meeting this mount's own prepared session lock. + journalCommitPermit chan struct{} + store store.MVCCStore + dispatch Dispatcher + chunkSize uint64 + openLeaseTTL time.Duration + capacity uint64 + maxFiles uint64 + now func() time.Time + allocID func() (uint64, error) + homeSlot func(uint64) uint64 + observer OperationalObserver } type Option func(*Service) @@ -332,16 +335,18 @@ func NewService(st store.MVCCStore, dispatch Dispatcher, opts ...Option) (*Servi return nil, ErrDispatcherRequired } s := &Service{ - store: st, - dispatch: dispatch, - chunkSize: DefaultChunkSize, - openLeaseTTL: defaultOpenHandleLeaseTTL, - now: time.Now, - allocID: randomNonZeroUint64, + journalCommitPermit: make(chan struct{}, 1), + store: st, + dispatch: dispatch, + chunkSize: DefaultChunkSize, + openLeaseTTL: defaultOpenHandleLeaseTTL, + now: time.Now, + allocID: randomNonZeroUint64, homeSlot: func(inode uint64) uint64 { return inode }, } + s.journalCommitPermit <- struct{}{} for _, opt := range opts { opt(s) } @@ -636,11 +641,12 @@ func (s *Service) openWithHandle(ctx context.Context, inode uint64, clientID []b if err != nil { return err } - return s.dispatchTxn(ctx, ts, []*kv.Elem[kv.OP]{refElem, fenceElem}, [][]byte{ + _, err = s.dispatchTxnWithPrimaryPolicy(ctx, ts, []*kv.Elem[kv.OP]{refElem, fenceElem}, [][]byte{ fskeys.InodeKey(inode), refKey, fskeys.RefFenceKey(inode), - }) + }, false, refKey) + return err } //nolint:cyclop // Read handles EOF, sparse chunks, and chunk boundary copying in one linear path. @@ -750,7 +756,8 @@ func (s *Service) retryWrite(ctx context.Context, inode uint64, offset uint64, d } func isDefinitiveWriteConflict(err error) bool { - return err != nil && errors.Is(err, store.ErrWriteConflict) && !errors.Is(err, kv.ErrTransactionOutcomeUnknown) + return err != nil && !errors.Is(err, kv.ErrTransactionOutcomeUnknown) && + (errors.Is(err, store.ErrWriteConflict) || errors.Is(err, kv.ErrTxnLocked)) } func waitWriteConflictRetry(ctx context.Context, attempt int) error { @@ -1271,7 +1278,7 @@ func (s *Service) releaseOnce(ctx context.Context, inode uint64, fh uint64, clie if err != nil { return err } - if err := s.dispatchTxn(ctx, ts, elems, readKeys); err != nil { + if _, err := s.dispatchTxnWithPrimaryPolicy(ctx, ts, elems, readKeys, false, refKey); err != nil { return err } s.reclaimCommittedOrphanBestEffort(ctx, chunkDeletes) diff --git a/internal/filesystem/service_test.go b/internal/filesystem/service_test.go index 8f96b1e1c..945a420b1 100644 --- a/internal/filesystem/service_test.go +++ b/internal/filesystem/service_test.go @@ -488,6 +488,23 @@ func TestServiceOpenRebuildsAfterDefinitiveConflicts(t *testing.T) { require.Equal(t, 6, conflicts.calls) } +func TestServiceOpenWaitsForPendingTransactionLock(t *testing.T) { + ctx := context.Background() + svc := newTestService(t, 2, 3, 4) + require.NoError(t, svc.InitializeRoot(ctx, testRootMode, 1000, 1000)) + created, err := svc.Create(ctx, RootInode, []byte("file"), CreateOptions{Mode: testFileMode}) + require.NoError(t, err) + inner, ok := svc.dispatch.(*testCoordinator) + require.True(t, ok) + locked := &transientWriteConflictCoordinator{inner: inner, remaining: 1, failure: kv.ErrTxnLocked} + svc.dispatch = locked + + fh, err := svc.Open(ctx, created.Inode, []byte("client-a")) + require.NoError(t, err) + require.NotZero(t, fh) + require.Equal(t, 2, locked.calls) +} + func TestServiceRefreshLeaseRebuildsAfterDefinitiveConflict(t *testing.T) { ctx := context.Background() svc := newTestService(t, 2, 3) @@ -732,6 +749,7 @@ func TestServiceOpenUsesRefFenceWithoutTouchingInode(t *testing.T) { require.EqualValues(t, 3, fh) require.Len(t, rec.requests, 1) require.True(t, elemTouchesKey(rec.requests[0].Elems, fskeys.RefKey(file.Inode, []byte("client-a"), fh))) + require.Equal(t, fskeys.RefKey(file.Inode, []byte("client-a"), fh), rec.requests[0].PrimaryKey) require.True(t, elemTouchesKey(rec.requests[0].Elems, fskeys.RefFenceKey(file.Inode))) require.False(t, elemTouchesKey(rec.requests[0].Elems, fskeys.InodeKey(file.Inode))) } @@ -1067,6 +1085,7 @@ func TestServiceReleaseLastOpenHandleGcsInSingleTxn(t *testing.T) { require.NoError(t, svc.Release(ctx, file.Inode, file.FH, []byte("client-a"))) require.Len(t, rec.requests, 1) require.True(t, elemTouchesKey(rec.requests[0].Elems, fskeys.RefKey(file.Inode, []byte("client-a"), file.FH))) + require.Equal(t, fskeys.RefKey(file.Inode, []byte("client-a"), file.FH), rec.requests[0].PrimaryKey) require.True(t, elemTouchesKey(rec.requests[0].Elems, fskeys.InodeKey(file.Inode))) require.True(t, elemTouchesKey(rec.requests[0].Elems, fskeys.RefFenceKey(file.Inode))) requireUsageRouteCounterWithTxn(t, rec.requests[0].Elems) diff --git a/internal/fskeys/keys.go b/internal/fskeys/keys.go index 738dc013c..8af98e82d 100644 --- a/internal/fskeys/keys.go +++ b/internal/fskeys/keys.go @@ -80,6 +80,11 @@ func IsOperationKey(key []byte) bool { len(key) == len(operationPrefix)+operationIdentityBytes+u64Bytes } +// OperationSessionAllPrefix spans durable mount incarnation rows. +func OperationSessionAllPrefix() []byte { + return []byte(operationSessionPrefix) +} + // OperationSessionKey names one durable mount incarnation. func OperationSessionKey(filesystemID, sessionID [16]byte) []byte { out := make([]byte, 0, len(operationSessionPrefix)+operationIdentityBytes) diff --git a/internal/raftengine/etcd/engine.go b/internal/raftengine/etcd/engine.go index 40ad3993f..cf5760f9c 100644 --- a/internal/raftengine/etcd/engine.go +++ b/internal/raftengine/etcd/engine.go @@ -4029,7 +4029,11 @@ func (e *Engine) refreshStatus() { e.fireLeaderAcquiredCallbacks() } if previous == raftengine.StateLeader && status.State != raftengine.StateLeader { - e.failPending(errors.WithStack(errNotLeader)) + // Pending proposals have already passed the pre-append leader check. + // Some may have entered the replicated log before this transition; + // callers must resolve their outcome instead of treating them as a + // definitive not-leader rejection. + e.failPending(errors.WithStack(raftengine.ErrLeadershipLost)) // Drop the per-peer ack map so a future re-election cannot // surface a stale majority-ack instant before the new term's // heartbeats have actually confirmed liveness. diff --git a/internal/raftengine/etcd/engine_test.go b/internal/raftengine/etcd/engine_test.go index 735e07d04..d09af4448 100644 --- a/internal/raftengine/etcd/engine_test.go +++ b/internal/raftengine/etcd/engine_test.go @@ -2489,6 +2489,28 @@ func TestErrNotLeaderMatchesRaftEngineSentinel(t *testing.T) { require.True(t, errors.Is(errors.WithStack(errLeadershipTransferConfChangePending), raftengine.ErrLeadershipTransferConfChangePending)) } +func TestLeaderLossLeavesPendingProposalOutcomeUnknown(t *testing.T) { + t.Parallel() + storage := etcdraft.NewMemoryStorage() + req := proposalRequest{id: 1, done: make(chan proposalResult, 1)} + engine := &Engine{ + nodeID: 1, + rawNode: mustRawNode(t, storage, 1), + storage: storage, + status: raftengine.Status{State: raftengine.StateLeader}, + pendingProposals: map[uint64]proposalRequest{req.id: req}, + pendingReads: map[uint64]readRequest{}, + pendingConfigs: map[uint64]adminRequest{}, + } + // The RawNode is a follower while the published status is still leader. + // refreshStatus must treat the waiting proposal as possibly appended. + engine.refreshStatus() + result := <-req.done + require.ErrorIs(t, result.err, raftengine.ErrLeadershipLost) + require.NotErrorIs(t, result.err, raftengine.ErrNotLeader) + require.Empty(t, engine.pendingProposals) +} + // TestSelectDispatchLane_LegacyThreeLane verifies that, when the opt-in // multi-lane dispatcher is disabled (default), priority control traffic uses // the heartbeat lane, heartbeat responses use their coalescing response lane, diff --git a/jepsen/src/elastickv/redis_workload.clj b/jepsen/src/elastickv/redis_workload.clj index ac4f5d3f2..a83a0e27b 100644 --- a/jepsen/src/elastickv/redis_workload.clj +++ b/jepsen/src/elastickv/redis_workload.clj @@ -74,6 +74,14 @@ (mapv (partial redis-append/apply-mop! conn)) (complete-mops-or-error conn op))) +(defn- preserve-unknown-leader-outcome [result] + ;; The upstream Redis client marks NOTLEADER as a definite failure. The + ;; reply can also arrive after EXEC committed but before its response was + ;; delivered, so a non-idempotent transaction remains indeterminate. + (if (and (= :fail (:type result)) (= :notleader (:error result))) + (assoc result :type :info) + result)) + (defrecord ElastickvRedisClient [node->port conn] client/Client (open! [this test node] @@ -92,11 +100,12 @@ (invoke! [this test op] (let [conn (:conn this)] - (rc/with-exceptions op #{} - (rc/with-conn conn - (if (< 1 (count (:value op))) - (apply-multi-op! conn op) - (apply-single-op! conn op))))))) + (preserve-unknown-leader-outcome + (rc/with-exceptions op #{} + (rc/with-conn conn + (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 index c7ec7d5e2..9b1df5a7b 100644 --- a/jepsen/test/elastickv/redis_workload_test.clj +++ b/jepsen/test/elastickv/redis_workload_test.clj @@ -1,6 +1,7 @@ (ns elastickv.redis-workload-test (:require [clojure.test :refer :all] - [elastickv.redis-workload :as workload])) + [elastickv.redis-workload :as workload] + [jepsen.redis.client :as rc])) (deftest complete-mops-parses-reads (let [op {:type :invoke @@ -25,3 +26,13 @@ (is (= :info (:type result))) (is (= [:exec-array-error :notleader] (:error result))) (is (= (:value op) (:value result))))) + +(deftest notleader-reply-after-dispatch-remains-indeterminate + (let [op {:type :invoke :f :txn :value [[:append 6 9]]} + raw (rc/with-exceptions op #{} + (throw (ex-info "NOTLEADER leadership lost" {:prefix :notleader}))) + result (#'workload/preserve-unknown-leader-outcome raw)] + (is (= :fail (:type raw))) + (is (= :info (:type result))) + (is (= :notleader (:error result))) + (is (= (:value op) (:value result))))) diff --git a/kv/lease_warmup_test.go b/kv/lease_warmup_test.go index 6eb5c62b1..afa9d8976 100644 --- a/kv/lease_warmup_test.go +++ b/kv/lease_warmup_test.go @@ -540,6 +540,8 @@ func TestShardedCoordinator_RenewHLCLeases_SkipsInFlightGroupWithoutBlockingPeer eng2 := newShardedLeaseEngine(200) entered := make(chan struct{}, 2) release := make(chan struct{}) + releaseBlocked := sync.OnceFunc(func() { close(release) }) + t.Cleanup(releaseBlocked) eng1.proposeHook = func() { entered <- struct{}{} <-release @@ -549,14 +551,17 @@ func TestShardedCoordinator_RenewHLCLeases_SkipsInFlightGroupWithoutBlockingPeer first := coord.renewHLCLeases(context.Background()) <-entered require.Eventually(t, func() bool { - return eng2.proposeCalls.Load() == 1 - }, time.Second, 10*time.Millisecond, + coord.hlcRenewalMu.Lock() + _, inFlight := coord.hlcRenewalInFlight[2] + coord.hlcRenewalMu.Unlock() + return eng2.proposeCalls.Load() == 1 && !inFlight + }, 2*time.Second, 10*time.Millisecond, "precondition: the first round must fully finish for the non-blocked group") second := coord.renewHLCLeases(context.Background()) require.Eventually(t, func() bool { return eng2.proposeCalls.Load() == 2 - }, time.Second, 10*time.Millisecond, + }, 2*time.Second, 10*time.Millisecond, "other led groups must still renew while one group is in flight") requireRenewalDone(t, second) select { @@ -569,7 +574,7 @@ func TestShardedCoordinator_RenewHLCLeases_SkipsInFlightGroupWithoutBlockingPeer require.Equal(t, int32(2), eng2.proposeCalls.Load(), "other led groups must still renew while one group is in flight") - close(release) + releaseBlocked() requireRenewalDone(t, first) third := coord.renewHLCLeases(context.Background()) diff --git a/kv/shard_store.go b/kv/shard_store.go index 836b373c6..551c3af60 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -4961,7 +4961,7 @@ func loadTxnLockAt(ctx context.Context, g *ShardGroup, key []byte, ts uint64) (t func (s *ShardStore) resolveTxnLockForKey(ctx context.Context, g *ShardGroup, key []byte, lock txnLock) error { // Check primary transaction status to decide commit/rollback. - status, commitTS, err := s.primaryTxnStatus(ctx, lock.PrimaryKey, lock.StartTS) + status, commitTS, err := s.primaryTxnStatus(ctx, lock) if err != nil { return err } @@ -5125,7 +5125,7 @@ func (s *ShardStore) cachedLockTxnStatus(ctx context.Context, plan *scanLockPlan if state, ok := plan.statusCache[txnKey]; ok { return state, nil } - status, commitTS, err := s.primaryTxnStatus(ctx, lock.PrimaryKey, lock.StartTS) + status, commitTS, err := s.primaryTxnStatus(ctx, lock) if err != nil { return lockTxnStatus{}, err } @@ -5418,7 +5418,8 @@ const ( txnStatusRolledBack ) -func (s *ShardStore) primaryTxnStatus(ctx context.Context, primaryKey []byte, startTS uint64) (txnStatus, uint64, error) { +func (s *ShardStore) primaryTxnStatus(ctx context.Context, secondaryLock txnLock) (txnStatus, uint64, error) { + primaryKey, startTS := secondaryLock.PrimaryKey, secondaryLock.StartTS status, commitTS, done, err := s.primaryTxnRecordedStatus(ctx, primaryKey, startTS) if err != nil || done { return status, commitTS, err @@ -5429,7 +5430,12 @@ func (s *ShardStore) primaryTxnStatus(ctx context.Context, primaryKey []byte, st return txnStatusPending, 0, err } if !locked { - return txnStatusRolledBack, 0, nil + // Secondaries are prepared before the primary. Its absent lock is not + // an abort decision while this secondary's lease is still live. + if !txnLockExpired(secondaryLock) { + return txnStatusPending, 0, nil + } + return s.expiredPrimaryTxnStatus(ctx, primaryKey, startTS) } if !txnLockExpired(lock) { return txnStatusPending, 0, nil diff --git a/kv/shard_store_txn_lock_test.go b/kv/shard_store_txn_lock_test.go index d0f9b314a..245411dfe 100644 --- a/kv/shard_store_txn_lock_test.go +++ b/kv/shard_store_txn_lock_test.go @@ -170,6 +170,36 @@ func TestShardStoreGetAt_ReturnsTxnLockedForPendingCrossShardTxn(t *testing.T) { require.True(t, errors.Is(err, ErrTxnLocked), "expected ErrTxnLocked for secondary key, got %v", err) } +func TestShardStoreGetAt_DoesNotAbortSecondaryBeforePrimaryPrepare(t *testing.T) { + t.Parallel() + ctx := context.Background() + shardStore, groups, cleanup := setupTwoShardStore(t) + defer cleanup() + + const startTS = uint64(10) + const commitTS = uint64(20) + primaryKey := []byte("b") + secondaryKey := []byte("x") + _, err := groups[2].Txn.Commit(ctx, []*pb.Request{ + makePrepareRequest(startTS, secondaryKey, []byte("value"), primaryKey), + }) + require.NoError(t, err) + + _, err = shardStore.GetAt(ctx, secondaryKey, ^uint64(0)) + require.ErrorIs(t, err, ErrTxnLocked) + _, err = groups[2].Store.GetAt(ctx, txnLockKey(secondaryKey), ^uint64(0)) + require.NoError(t, err, "a read cannot discard a secondary intent before the primary is prepared") + + _, err = groups[1].Txn.Commit(ctx, []*pb.Request{ + makePrepareRequest(startTS, primaryKey, []byte("primary"), primaryKey), + }) + require.NoError(t, err) + commitPrimary(t, groups[1], startTS, commitTS, primaryKey) + value, err := shardStore.GetAt(ctx, secondaryKey, commitTS) + require.NoError(t, err) + require.Equal(t, []byte("value"), value) +} + func TestShardStoreGetAt_ResolvesCommittedSecondaryLock(t *testing.T) { t.Parallel() diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go index 37a5ab2ea..a6aa9b95a 100644 --- a/kv/sharded_coordinator.go +++ b/kv/sharded_coordinator.go @@ -1990,16 +1990,20 @@ func (c *ShardedCoordinator) dispatchMultiShardTxn(ctx context.Context, startTS, primaryGid, maxIndex, err := c.commitPrimaryTxn(ctx, startTS, primaryKey, grouped, gids, commitTS, observedRouteVersion, bypassKeysByGroup) if err != nil { - // abortPreparedTxn must run even when ctx was the reason - // commitPrimaryTxn failed — otherwise prewrite intents on - // every prepared shard linger until LockResolver picks them - // up at a future tick (lease window of expensive - // keyspace-scan work). Detach cancellation but cap with - // verifyLeaderTimeout so a hung Abort cannot leak. - cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), verifyLeaderTimeout) - c.abortPreparedTxn(cleanupCtx, startTS, primaryKey, prepared, c.abortTimestamp(cleanupCtx, startTS, commitTS)) - cancel() - return nil, errors.WithStack(err) + if primaryCommitDefinitelyRejected(err) { + // A deterministic apply verdict or pre-append leadership rejection + // is a definitive non-commit. Release prepared intents even if the + // caller has since cancelled. + cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), verifyLeaderTimeout) + c.abortPreparedTxn(cleanupCtx, startTS, primaryKey, prepared, c.abortTimestamp(cleanupCtx, startTS, commitTS)) + cancel() + return nil, errors.WithStack(err) + } + // A lost primary COMMIT response may conceal a durable commit + // decision. Aborting prepared secondaries here can then turn one + // committed transaction into a partial commit. Lock resolution must + // consult the primary decision; retain the intents until it does. + return nil, fmt.Errorf("%w: %w", ErrTransactionOutcomeUnknown, err) } // commitSecondaryTxns propagates ObservedRouteVersion to per- @@ -2022,6 +2026,16 @@ func (c *ShardedCoordinator) dispatchMultiShardTxn(ctx context.Context, startTS, return &CoordinateResponse{CommitIndex: maxIndex, CommitTS: commitTS}, nil } +func primaryCommitDefinitelyRejected(err error) bool { + if errors.Is(err, ErrTransactionOutcomeUnknown) { + return false + } + return isRawApplyVerdict(err) || + errors.Is(err, ErrComposed1Violation) || + errors.Is(err, raftengine.ErrNotLeader) || + errors.Is(err, raftengine.ErrLeadershipTransferInProgress) +} + func (c *ShardedCoordinator) resolveTxnCommitTS(ctx context.Context, startTS, commitTS uint64) (uint64, error) { if commitTS == 0 { next, err := c.nextTxnTSAfter(ctx, startTS) @@ -2069,7 +2083,7 @@ func (c *ShardedCoordinator) dispatchSingleShardTxn(ctx context.Context, startTS return nil, errors.WithStack(err) } if resp == nil { - return &CoordinateResponse{}, nil + return nil, errors.WithStack(ErrTransactionOutcomeUnknown) } return &CoordinateResponse{CommitIndex: resp.CommitIndex, CommitTS: commitTS}, nil } @@ -2201,7 +2215,7 @@ func (c *ShardedCoordinator) commitPrimaryTxn(ctx context.Context, startTS uint6 return primaryGid, 0, errors.WithStack(err) } if r == nil { - return primaryGid, 0, nil + return primaryGid, 0, errors.WithStack(ErrTransactionOutcomeUnknown) } return primaryGid, r.CommitIndex, nil } diff --git a/kv/sharded_coordinator_composed1_retry_test.go b/kv/sharded_coordinator_composed1_retry_test.go index fe66ba53d..f347a5e10 100644 --- a/kv/sharded_coordinator_composed1_retry_test.go +++ b/kv/sharded_coordinator_composed1_retry_test.go @@ -48,7 +48,7 @@ func (s *composed1RetryingTransactional) Commit(_ context.Context, reqs []*pb.Re } s.prevCommitTSSeen = append(s.prevCommitTSSeen, prev) } - if n == 1 { + if n == 1 && s.firstErr != nil { return nil, s.firstErr } return &TransactionResponse{CommitIndex: uint64(n)}, nil //nolint:gosec // n is a small positive counter for test-stub responses diff --git a/kv/sharded_coordinator_crash_test.go b/kv/sharded_coordinator_crash_test.go new file mode 100644 index 000000000..939abd5b3 --- /dev/null +++ b/kv/sharded_coordinator_crash_test.go @@ -0,0 +1,135 @@ +package kv + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "syscall" + "testing" + "time" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +const multiGroupCrashDirectory = "ELASTICKV_MULTIGROUP_CRASH_DIRECTORY" + +type killAfterPrimaryCommit struct { + Transactional + marker string +} + +func (t killAfterPrimaryCommit) Commit(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error) { + response, err := t.Transactional.Commit(ctx, reqs) + if err == nil && len(reqs) == 1 && reqs[0].Phase == pb.Phase_COMMIT { + marker, openErr := os.OpenFile(t.marker, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if openErr != nil { + return nil, openErr + } + if _, writeErr := marker.Write([]byte("primary committed")); writeErr != nil { + _ = marker.Close() + return nil, writeErr + } + if syncErr := marker.Sync(); syncErr != nil { + _ = marker.Close() + return nil, syncErr + } + if closeErr := marker.Close(); closeErr != nil { + return nil, closeErr + } + _ = syscall.Kill(os.Getpid(), syscall.SIGKILL) + } + return response, err +} + +func openCrashTestGroups(t *testing.T, dir string, bootstrap bool) (*LockResolver, *ShardStore, map[uint64]*ShardGroup, func()) { + t.Helper() + router := distribution.NewEngine() + router.UpdateRoute([]byte("a"), []byte("m"), 1) + router.UpdateRoute([]byte("m"), nil, 2) + groups := make(map[uint64]*ShardGroup, 2) + var closeGroups []func() + for _, gid := range []uint64{1, 2} { + dataDir := filepath.Join(dir, fmt.Sprintf("raft-%d", gid)) + storeDir := filepath.Join(dir, fmt.Sprintf("store-%d", gid)) + st, err := store.NewPebbleStore(storeDir) + require.NoError(t, err) + id := fmt.Sprintf("crash-g%d", gid) + result, err := newSingleRaftFactory().Create(raftengine.FactoryConfig{ + LocalID: id, LocalAddress: id, DataDir: dataDir, Bootstrap: bootstrap, + StateMachine: NewKvFSMWithHLC(st, NewHLC()), + }) + require.NoError(t, err) + require.Eventually(t, func() bool { return result.Engine.State() == raftengine.StateLeader }, + 5*time.Second, 10*time.Millisecond) + groups[gid] = &ShardGroup{Engine: result.Engine, Store: st, Txn: NewLeaderProxyWithEngine(result.Engine)} + closeGroups = append(closeGroups, func() { + require.NoError(t, result.Engine.Close()) + if result.Close != nil { + require.NoError(t, result.Close()) + } + require.NoError(t, st.Close()) + }) + } + shards := NewShardStore(router, groups) + resolver := NewLockResolver(shards, groups, nil) + return resolver, shards, groups, func() { + resolver.Close() + for _, closeGroup := range closeGroups { + closeGroup() + } + } +} + +func TestMultiGroupPrimaryDecisionSurvivesProcessKill(t *testing.T) { + if dir := os.Getenv(multiGroupCrashDirectory); dir != "" { + _, shards, groups, _ := openCrashTestGroups(t, dir, true) + groups[1].Txn = killAfterPrimaryCommit{Transactional: groups[1].Txn, marker: filepath.Join(dir, "primary-committed")} + coord := NewShardedCoordinator(shards.engine, groups, 1, NewHLC(), shards) + _, err := coord.dispatchMultiShardTxn(context.Background(), 10, 20, 0, []byte("b"), + map[uint64][]*pb.Mutation{ + 1: {{Op: pb.Op_PUT, Key: []byte("b"), Value: []byte("primary")}}, + 2: {{Op: pb.Op_PUT, Key: []byte("x"), Value: []byte("secondary")}}, + }, []uint64{1, 2}, nil, 0, nil, true) + require.NoError(t, err, "the primary-commit crash hook was not reached") + return + } + + dir := t.TempDir() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestMultiGroupPrimaryDecisionSurvivesProcessKill$") + cmd.Env = append(os.Environ(), multiGroupCrashDirectory+"="+dir) + output, err := cmd.CombinedOutput() + var exit *exec.ExitError + require.ErrorAs(t, err, &exit, string(output)) + status, ok := exit.Sys().(syscall.WaitStatus) + require.True(t, ok, string(output)) + require.Equal(t, syscall.SIGKILL, status.Signal(), string(output)) + require.FileExists(t, filepath.Join(dir, "primary-committed")) + + resolver, shards, groups, cleanup := openCrashTestGroups(t, dir, false) + defer cleanup() + ctx = context.Background() + primary, err := shards.GetAt(ctx, []byte("b"), 20) + require.NoError(t, err) + require.Equal(t, []byte("primary"), primary) + _, err = groups[2].Store.GetAt(ctx, []byte("x"), 20) + require.ErrorIs(t, err, store.ErrKeyNotFound, + "the secondary must still be prepared before recovery resolves it") + rawLock, err := groups[2].Store.GetAt(ctx, txnLockKey([]byte("x")), ^uint64(0)) + require.NoError(t, err) + lock, err := decodeTxnLock(rawLock) + require.NoError(t, err) + require.NoError(t, resolver.resolveExpiredLock(ctx, groups[2], []byte("x"), lock)) + secondary, err := groups[2].Store.GetAt(ctx, []byte("x"), 20) + require.NoError(t, err) + require.Equal(t, []byte("secondary"), secondary) + _, err = groups[2].Store.GetAt(ctx, txnLockKey([]byte("x")), ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound) +} diff --git a/kv/sharded_coordinator_strict_result_test.go b/kv/sharded_coordinator_strict_result_test.go index 27c770438..feaed14a0 100644 --- a/kv/sharded_coordinator_strict_result_test.go +++ b/kv/sharded_coordinator_strict_result_test.go @@ -6,7 +6,9 @@ import ( "testing" "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" ) @@ -43,6 +45,23 @@ func TestCommitSecondaryTxnsRequiresNonNilDecision(t *testing.T) { require.ErrorIs(t, err, ErrTransactionOutcomeUnknown) } +func TestSingleShardTxnRequiresNonNilDecision(t *testing.T) { + t.Parallel() + coord := &ShardedCoordinator{groups: map[uint64]*ShardGroup{1: {Txn: nilDecisionTransactional{}}}} + _, err := coord.dispatchSingleShardTxn(context.Background(), 10, 20, 0, []byte("primary"), 1, + []*Elem[OP]{{Op: Put, Key: []byte("primary"), Value: []byte("value")}}, nil, 0, nil) + require.ErrorIs(t, err, ErrTransactionOutcomeUnknown) +} + +func TestPrimaryTxnRequiresNonNilDecision(t *testing.T) { + t.Parallel() + coord := &ShardedCoordinator{groups: map[uint64]*ShardGroup{1: {Txn: nilDecisionTransactional{}}}} + _, _, err := coord.commitPrimaryTxn(context.Background(), 10, []byte("primary"), + map[uint64][]*pb.Mutation{1: {{Op: pb.Op_PUT, Key: []byte("primary")}}}, + []uint64{1}, 20, 0, nil) + require.ErrorIs(t, err, ErrTransactionOutcomeUnknown) +} + type nilDecisionTransactional struct{} func (nilDecisionTransactional) Commit(context.Context, []*pb.Request) (*TransactionResponse, error) { @@ -53,6 +72,65 @@ func (nilDecisionTransactional) Abort(context.Context, []*pb.Request) (*Transact return nil, nil } +type lostPrimaryCommitReplyTxn struct { + Transactional +} + +func (t lostPrimaryCommitReplyTxn) Commit(ctx context.Context, reqs []*pb.Request) (*TransactionResponse, error) { + response, err := t.Transactional.Commit(ctx, reqs) + if err == nil && len(reqs) == 1 && reqs[0].Phase == pb.Phase_COMMIT { + return nil, errors.New("committed primary reply lost") + } + return response, err +} + +func TestLostPrimaryReplyResolvesPreparedSecondaryFromDurableDecision(t *testing.T) { + lr, ss, groups, cleanup := setupLockResolverEnv(t) + defer cleanup() + groups[1].Txn = lostPrimaryCommitReplyTxn{Transactional: groups[1].Txn} + coord := NewShardedCoordinator(ss.engine, groups, 1, NewHLC(), ss) + ctx := context.Background() + _, err := coord.dispatchMultiShardTxn(ctx, 10, 20, 0, []byte("b"), + map[uint64][]*pb.Mutation{ + 1: {{Op: pb.Op_PUT, Key: []byte("b"), Value: []byte("primary")}}, + 2: {{Op: pb.Op_PUT, Key: []byte("x"), Value: []byte("secondary")}}, + }, []uint64{1, 2}, nil, 0, nil, true) + require.ErrorIs(t, err, ErrTransactionOutcomeUnknown) + + primary, err := ss.GetAt(ctx, []byte("b"), 20) + require.NoError(t, err) + require.Equal(t, []byte("primary"), primary) + rawLock, err := groups[2].Store.GetAt(ctx, txnLockKey([]byte("x")), ^uint64(0)) + require.NoError(t, err) + lock, err := decodeTxnLock(rawLock) + require.NoError(t, err) + require.NoError(t, lr.resolveExpiredLock(ctx, groups[2], []byte("x"), lock)) + secondary, err := ss.GetAt(ctx, []byte("x"), 20) + require.NoError(t, err) + require.Equal(t, []byte("secondary"), secondary) +} + +func TestRepeatedTwoGroupCommitReleasesSecondaryLocks(t *testing.T) { + _, ss, groups, cleanup := setupLockResolverEnv(t) + defer cleanup() + coord := NewShardedCoordinator(ss.engine, groups, 1, NewHLC(), ss) + ctx := context.Background() + for _, tc := range []struct { + start, commit uint64 + value string + }{{10, 20, "first"}, {30, 40, "second"}} { + _, err := coord.dispatchMultiShardTxn(ctx, tc.start, tc.commit, 0, []byte("b"), + map[uint64][]*pb.Mutation{ + 1: {{Op: pb.Op_PUT, Key: []byte("b"), Value: []byte(tc.value)}}, + 2: {{Op: pb.Op_PUT, Key: []byte("x"), Value: []byte(tc.value)}}, + }, []uint64{1, 2}, nil, 0, nil, true) + require.NoError(t, err) + } + value, err := ss.GetAt(ctx, []byte("x"), 40) + require.NoError(t, err) + require.Equal(t, []byte("second"), value) +} + func TestStrictSecondaryRouteShiftNeverStartsANewTransaction(t *testing.T) { t.Parallel() router := distribution.NewEngine() @@ -81,6 +159,101 @@ func TestStrictSecondaryRouteShiftNeverStartsANewTransaction(t *testing.T) { require.Len(t, secondary.requests, 1+txnSecondaryCommitRetryAttempts) } +func TestStrictLostPrimaryCommitReplyDoesNotAbortPreparedParticipants(t *testing.T) { + t.Parallel() + router := distribution.NewEngine() + applyComposed1Snapshot(t, router, 7, []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte("a"), End: []byte("m"), GroupID: 1}, + {RouteID: 2, Start: []byte("m"), GroupID: 2}, + }) + responseLost := raftengine.ErrLeadershipLost + primary := &recordingTransactional{errs: []error{nil, responseLost}} + secondary := &recordingTransactional{} + coord := NewShardedCoordinator(router, map[uint64]*ShardGroup{ + 1: {Txn: primary}, 2: {Txn: secondary}, + }, 1, NewHLC(), nil) + _, err := coord.Dispatch(context.Background(), &OperationGroup[OP]{ + IsTxn: true, RequireAllParticipants: true, PrimaryKey: []byte("b"), + Elems: []*Elem[OP]{ + {Op: Put, Key: []byte("b"), Value: []byte("record")}, + {Op: Put, Key: []byte("x"), Value: []byte("effect")}, + }, + }) + require.ErrorIs(t, err, ErrTransactionOutcomeUnknown) + require.ErrorIs(t, err, responseLost) + require.Len(t, primary.requests, 2) + require.Equal(t, pb.Phase_PREPARE, primary.requests[0].Phase) + require.Equal(t, pb.Phase_COMMIT, primary.requests[1].Phase) + require.Len(t, secondary.requests, 1) + require.Equal(t, pb.Phase_PREPARE, secondary.requests[0].Phase) +} + +func TestStrictDefinitivePrimaryConflictAbortsPreparedParticipants(t *testing.T) { + t.Parallel() + router := distribution.NewEngine() + applyComposed1Snapshot(t, router, 7, []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte("a"), End: []byte("m"), GroupID: 1}, + {RouteID: 2, Start: []byte("m"), GroupID: 2}, + }) + primary := &recordingTransactional{errs: []error{nil, store.ErrWriteConflict}} + secondary := &recordingTransactional{} + coord := NewShardedCoordinator(router, map[uint64]*ShardGroup{ + 1: {Txn: primary}, 2: {Txn: secondary}, + }, 1, NewHLC(), nil) + _, err := coord.Dispatch(context.Background(), &OperationGroup[OP]{ + IsTxn: true, RequireAllParticipants: true, PrimaryKey: []byte("b"), + Elems: []*Elem[OP]{ + {Op: Put, Key: []byte("b"), Value: []byte("record")}, + {Op: Put, Key: []byte("x"), Value: []byte("effect")}, + }, + }) + require.ErrorIs(t, err, store.ErrWriteConflict) + require.NotErrorIs(t, err, ErrTransactionOutcomeUnknown) + require.Len(t, primary.requests, 3) + require.Equal(t, pb.Phase_ABORT, primary.requests[2].Phase) + require.Len(t, secondary.requests, 2) + require.Equal(t, pb.Phase_ABORT, secondary.requests[1].Phase) +} + +func TestPrimaryCommitDecisionClassification(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + err error + rejected bool + }{ + {"conflict", store.ErrWriteConflict, true}, + {"route version", ErrComposed1VersionGCd, true}, + {"route moved", ErrComposed1Violation, true}, + {"not leader before append", raftengine.ErrNotLeader, true}, + {"ambiguous wrapped conflict", errors.Join(ErrTransactionOutcomeUnknown, store.ErrWriteConflict), false}, + {"unclassified transport", errors.New("response lost"), false}, + } { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.rejected, primaryCommitDefinitelyRejected(tc.err)) + }) + } +} + +func TestPrimaryRouteVerdictAbortsPreparedParticipants(t *testing.T) { + t.Parallel() + primary := &recordingTransactional{errs: []error{nil, ErrComposed1VersionGCd}} + secondary := &recordingTransactional{} + coord := &ShardedCoordinator{groups: map[uint64]*ShardGroup{ + 1: {Txn: primary}, 2: {Txn: secondary}, + }} + grouped := map[uint64][]*pb.Mutation{ + 1: {{Key: []byte("b"), Value: []byte("record")}}, + 2: {{Key: []byte("x"), Value: []byte("effect")}}, + } + _, err := coord.dispatchMultiShardTxn(context.Background(), 10, 20, 0, + []byte("b"), grouped, []uint64{1, 2}, nil, 7, nil, true) + require.ErrorIs(t, err, ErrComposed1VersionGCd) + require.NotErrorIs(t, err, ErrTransactionOutcomeUnknown) + require.Equal(t, pb.Phase_ABORT, primary.requests[2].Phase) + require.Equal(t, pb.Phase_ABORT, secondary.requests[1].Phase) +} + func TestStrictTxnRejectsEffectRouteChangedAfterReceiptPlacement(t *testing.T) { t.Parallel() router := distribution.NewEngine() diff --git a/main.go b/main.go index 16fee45a7..878617495 100644 --- a/main.go +++ b/main.go @@ -1076,6 +1076,9 @@ func startFilesystemIfEnabled( } installFilesystemServerLifecycle(ctx, eg, cleanup, server, serveDone, config.mountPoint) startFilesystemPlacementCollector(ctx, eg, service, *filesystemPlacementScanInterval) + if config.strict { + startFilesystemJournalCollector(ctx, eg, service, observer, *filesystemPlacementScanInterval) + } startFilesystemLeaseReaper(ctx, eg, service, *filesystemLeaseReapInterval) slog.Info("filesystem FUSE mounted", "mount_point", config.mountPoint, "client_id", config.clientID) return nil @@ -1348,6 +1351,37 @@ func startFilesystemPlacementCollector( }) } +func startFilesystemJournalCollector( + ctx context.Context, + eg *errgroup.Group, + service *filesystem.Service, + observer monitoring.FileSystemObserver, + interval time.Duration, +) { + if eg == nil || service == nil || observer == nil || interval <= 0 { + return + } + eg.Go(func() error { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + stats, err := service.ListOperationJournalStats(ctx) + if err != nil { + if ctx.Err() == nil { + slog.WarnContext(ctx, "filesystem journal usage scan failed", "err", err) + } + } else { + observer.ObserveOperationJournalStats(stats) + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + } + } + }) +} + type filesystemLeaseReaper interface { ReapExpiredOpenHandleLeases(context.Context, int) (filesystem.LeaseReapStats, error) } diff --git a/monitoring/filesystem.go b/monitoring/filesystem.go index 2b93f28f3..208369b4f 100644 --- a/monitoring/filesystem.go +++ b/monitoring/filesystem.go @@ -25,6 +25,7 @@ type FileSystemObserver interface { ObserveInodeIDCollisionRetry() ObserveMoveJob(jobID []byte, active bool) ObservePlacementStats(filesystem.PlacementStats) + ObserveOperationJournalStats(filesystem.OperationJournalStats) } type FileSystemMetrics struct { @@ -41,6 +42,9 @@ type FileSystemMetrics struct { homeEpochConflictTotal prometheus.Counter openHandleLeaseActive prometheus.Gauge orphanInodeGCPending prometheus.Gauge + journalSessions prometheus.Gauge + journalPending prometheus.Gauge + journalRetainedBytes prometheus.Gauge inodeCollisionTotal prometheus.Counter filesByGroup *prometheus.GaugeVec activeMoveJobs map[string]struct{} @@ -107,6 +111,18 @@ func newFileSystemMetrics(registerer prometheus.Registerer) *FileSystemMetrics { Name: "elastickv_fs_inode_id_collision_retry_total", Help: "Total inode allocator retries caused by reserved or already-used IDs.", }), + journalSessions: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "elastickv_fs_operation_journal_sessions", + Help: "Durable strict FUSE mount sessions, including closed sessions awaiting retirement.", + }), + journalPending: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "elastickv_fs_operation_journal_pending", + Help: "Admitted strict FUSE operations without a terminal result.", + }), + journalRetainedBytes: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "elastickv_fs_operation_journal_retained_bytes", + Help: "Bytes reserved by durable strict FUSE journal records across mount sessions.", + }), filesByGroup: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: "elastickv_fs_files_by_group", Help: "Current active filesystem files by owning shard group.", @@ -129,11 +145,20 @@ func newFileSystemMetrics(registerer prometheus.Registerer) *FileSystemMetrics { m.openHandleLeaseActive, m.orphanInodeGCPending, m.inodeCollisionTotal, + m.journalSessions, + m.journalPending, + m.journalRetainedBytes, m.filesByGroup, ) return m } +func (m *FileSystemMetrics) ObserveOperationJournalStats(stats filesystem.OperationJournalStats) { + m.journalSessions.Set(float64(stats.Sessions)) + m.journalPending.Set(float64(stats.Pending)) + m.journalRetainedBytes.Set(float64(stats.RetainedBytes)) +} + func (m *FileSystemMetrics) ObserveChunkRead(elapsed time.Duration) { if m == nil { return diff --git a/monitoring/filesystem_test.go b/monitoring/filesystem_test.go index d82cb0489..6637c647f 100644 --- a/monitoring/filesystem_test.go +++ b/monitoring/filesystem_test.go @@ -51,6 +51,9 @@ func TestFileSystemMetricsObserveOperationsAndPlacement(t *testing.T) { OpenHandleLeases: 4, OrphanedInodes: 5, }) + metrics.ObserveOperationJournalStats(filesystem.OperationJournalStats{ + Sessions: 2, Pending: 1, RetainedBytes: 4096, + }) require.Equal(t, float64(1), testutil.ToFloat64(metrics.chunkReadOpsTotal)) require.Equal(t, float64(1), testutil.ToFloat64(metrics.chunkWriteOpsTotal)) @@ -61,6 +64,9 @@ func TestFileSystemMetricsObserveOperationsAndPlacement(t *testing.T) { require.Equal(t, float64(1), testutil.ToFloat64(metrics.fileMoveInflight)) require.Equal(t, float64(4), testutil.ToFloat64(metrics.openHandleLeaseActive)) require.Equal(t, float64(5), testutil.ToFloat64(metrics.orphanInodeGCPending)) + require.Equal(t, float64(2), testutil.ToFloat64(metrics.journalSessions)) + require.Equal(t, float64(1), testutil.ToFloat64(metrics.journalPending)) + require.Equal(t, float64(4096), testutil.ToFloat64(metrics.journalRetainedBytes)) require.Equal(t, float64(2), testutil.ToFloat64(metrics.filesByGroup.WithLabelValues("1"))) require.Equal(t, float64(3), testutil.ToFloat64(metrics.filesByGroup.WithLabelValues("2"))) diff --git a/scripts/benchmark-fuse.py b/scripts/benchmark-fuse.py new file mode 100644 index 000000000..88544e914 --- /dev/null +++ b/scripts/benchmark-fuse.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Measure FUSE write plus fsync latency on one file and a hot inode.""" + +import argparse +import json +import multiprocessing +import os +import pathlib +import time +import uuid + + +BLOCK = 4096 + + +def percentile(samples, percent): + ordered = sorted(samples) + index = max(0, (len(ordered) * percent + 99) // 100 - 1) + return round(ordered[index] / 1_000_000, 3) + + +def summary(samples, elapsed): + return { + "operations": len(samples), + "elapsed_seconds": round(elapsed, 3), + "ops_per_second": round(len(samples) / elapsed, 3), + "latency_ms": { + "p50": percentile(samples, 50), + "p95": percentile(samples, 95), + "p99": percentile(samples, 99), + "max": round(max(samples) / 1_000_000, 3), + }, + } + + +def write_and_sync(fd, offset, payload): + started = time.perf_counter_ns() + written = os.pwrite(fd, payload, offset) + if written != len(payload): + raise OSError(f"short pwrite: {written}/{len(payload)}") + os.fsync(fd) + return time.perf_counter_ns() - started + + +def hot_writer(path, worker, writes, workers, start, results): + samples = [] + try: + fd = os.open(path, os.O_RDWR) + try: + if not start.wait(30): + raise TimeoutError("hot writer start timed out") + payload = bytes([worker + 1]) * BLOCK + for index in range(writes): + samples.append(write_and_sync(fd, (index * workers + worker) * BLOCK, payload)) + finally: + os.close(fd) + results.put((samples, None)) + except Exception as exc: + results.put((samples, repr(exc))) + + +def collect_worker_results(processes, results, timeout=300): + try: + # A worker cannot exit until its queue feeder flushes. Read each + # result before join so a large sample list cannot fill the pipe. + outcomes = [results.get(timeout=timeout) for _ in processes] + except Exception: + for process in processes: + process.terminate() + process.join() + raise + for process in processes: + process.join(10) + if process.is_alive(): + process.terminate() + process.join() + raise TimeoutError("hot writer did not exit after returning results") + if process.exitcode != 0: + raise RuntimeError(f"hot writer exited with status {process.exitcode}") + return outcomes + + +def run_serial(path, writes): + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600) + samples = [] + try: + started = time.perf_counter() + for index in range(writes): + samples.append(write_and_sync(fd, index * BLOCK, b"S" * BLOCK)) + elapsed = time.perf_counter() - started + finally: + os.close(fd) + if path.read_bytes() != b"S" * BLOCK * writes: + raise AssertionError("serial writes were not retained") + return summary(samples, elapsed) + + +def run_hot(path, workers, writes): + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600) + os.close(fd) + start = multiprocessing.Event() + results = multiprocessing.Queue() + processes = [ + multiprocessing.Process(target=hot_writer, args=(str(path), worker, writes, workers, start, results)) + for worker in range(workers) + ] + for process in processes: + process.start() + started = time.perf_counter() + start.set() + outcomes = collect_worker_results(processes, results) + failures = [error for _, error in outcomes if error] + if failures: + raise RuntimeError(f"hot writer errors: {failures}") + samples = [sample for worker_samples, _ in outcomes for sample in worker_samples] + elapsed = time.perf_counter() - started + expected = b"".join(bytes([worker + 1]) * BLOCK for _ in range(writes) for worker in range(workers)) + if path.read_bytes() != expected: + raise AssertionError("hot inode writes were not retained") + result = summary(samples, elapsed) + worker_rates = [len(worker_samples) * 1_000_000_000 / sum(worker_samples) + for worker_samples, _ in outcomes] + result["worker_fairness_min_to_max"] = round(min(worker_rates) / max(worker_rates), 3) + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("mount", type=pathlib.Path) + parser.add_argument("--writes", type=int, default=40) + parser.add_argument("--workers", type=int, default=4) + parser.add_argument("--repetitions", type=int, default=3) + parser.add_argument("--mode", choices=("strict", "legacy"), required=True) + parser.add_argument("--layout", choices=("single-group", "two-group"), required=True) + parser.add_argument("--output", type=pathlib.Path, required=True) + args = parser.parse_args() + if args.writes < 1 or not 1 <= args.workers <= 255 or args.repetitions < 1: + parser.error("writes and repetitions must be positive; workers must be between 1 and 255") + prefix = uuid.uuid4().hex + result = { + "source_sha": os.environ.get("GITHUB_SHA"), + "workflow_run_id": os.environ.get("GITHUB_RUN_ID"), + "runner_os": os.environ.get("RUNNER_OS"), + "block_bytes": BLOCK, + "writes_per_worker": args.writes, + "workers": args.workers, + "mode": args.mode, + "layout": args.layout, + "trials": [], + } + for trial in range(args.repetitions): + result["trials"].append({ + "trial": trial + 1, + "serial": run_serial(args.mount / f"benchmark-serial-{prefix}-{trial}.dat", args.writes), + "hot_inode": run_hot(args.mount / f"benchmark-hot-{prefix}-{trial}.dat", args.workers, args.writes), + }) + args.output.write_text(json.dumps(result, indent=2) + "\n") + print(json.dumps(result, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/test-fuse-node-crash.py b/scripts/test-fuse-node-crash.py new file mode 100644 index 000000000..eaba46a4f --- /dev/null +++ b/scripts/test-fuse-node-crash.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +"""Check strict FUSE fsync durability through one Raft node SIGKILL/restart.""" + +import argparse +import os +import pathlib +import signal +import stat +import subprocess +import time + + +def await_contents(path, expected, seconds=90): + deadline = time.monotonic() + seconds + last_error = None + while time.monotonic() < deadline: + try: + actual = path.read_bytes() + if actual == expected: + return + last_error = AssertionError(f"{path}: {actual!r} != {expected!r}") + except OSError as exc: + last_error = exc + time.sleep(0.5) + raise AssertionError(f"{path}: did not recover: {last_error}") + + +def sync_file_and_directory(path, contents): + descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(descriptor, "wb", closefd=False) as stream: + stream.write(contents) + stream.flush() + os.fsync(descriptor) + finally: + os.close(descriptor) + sync_directory(path.parent) + + +def sync_directory(path): + directory = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + + +def await_mutation_state(mount, contents): + await_contents(mount / "strict-node-crash.dat", contents) + await_contents(mount / "strict-node-renamed.dat", b"renamed across a node crash\n") + await_contents(mount / "strict-node-directory" / "truncated.dat", b"truncated\n") + deadline = time.monotonic() + 90 + while time.monotonic() < deadline: + renamed = mount / "strict-node-renamed.dat" + removed = mount / "strict-node-removed.dat" + empty = mount / "strict-node-empty" + nested = mount / "strict-node-directory" + try: + if (stat.S_IMODE(renamed.stat().st_mode) == 0o640 + and nested.is_dir() and not removed.exists() and not empty.exists() + and not (mount / "strict-node-source.dat").exists()): + return + except OSError: + pass + time.sleep(0.5) + raise AssertionError(f"{mount}: mutation state did not recover") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("mounts", nargs=3, type=pathlib.Path) + parser.add_argument("--pid-file", type=pathlib.Path, required=True) + parser.add_argument("--restart-log", type=pathlib.Path, required=True) + args = parser.parse_args() + + name = "strict-node-crash.dat" + contents = b"committed before Raft node SIGKILL\n" * 128 + paths = [mount / name for mount in args.mounts] + sync_file_and_directory(paths[0], contents) + first = args.mounts[0] + source = first / "strict-node-source.dat" + renamed = first / "strict-node-renamed.dat" + sync_file_and_directory(source, b"renamed across a node crash\n") + os.rename(source, renamed) + os.chmod(renamed, 0o640) + with renamed.open("rb") as stream: + os.fsync(stream.fileno()) + removed = first / "strict-node-removed.dat" + sync_file_and_directory(removed, b"remove me\n") + os.unlink(removed) + empty = first / "strict-node-empty" + os.mkdir(empty) + os.rmdir(empty) + nested = first / "strict-node-directory" + os.mkdir(nested) + truncated = nested / "truncated.dat" + sync_file_and_directory(truncated, b"truncated\nthen removed\n") + os.truncate(truncated, len(b"truncated\n")) + with truncated.open("rb") as stream: + os.fsync(stream.fileno()) + sync_directory(nested) + sync_directory(first) + for mount in args.mounts[1:]: + await_mutation_state(mount, contents) + + pids = [int(line) for line in args.pid_file.read_text().splitlines() if line.strip()] + if len(pids) < len(args.mounts): + raise AssertionError("pid file does not contain every cluster node") + pid = pids[0] + # /proc cmdline ends in NUL; interior empty arguments (for example an + # intentionally disabled metrics address) must remain in their positions. + argv = [os.fsdecode(arg) for arg in pathlib.Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0")[:-1]] + if not argv or "--filesystemStrict" not in argv: + raise AssertionError("node was not started in strict mode") + os.kill(pid, signal.SIGKILL) + # The surviving quorum must still serve the acknowledged file. + for mount in args.mounts[1:]: + await_mutation_state(mount, contents) + + restart_argv = argv if "--filesystemStrictRecover" in argv else argv + ["--filesystemStrictRecover"] + with args.restart_log.open("wb") as log: + restarted = subprocess.Popen(restart_argv, stdout=log, stderr=subprocess.STDOUT) + with args.pid_file.open("a") as pid_file: + pid_file.write(f"{restarted.pid}\n") + await_mutation_state(args.mounts[0], contents) + for _ in range(10): + if restarted.poll() is not None: + raise AssertionError(f"restarted node exited with {restarted.returncode}") + time.sleep(0.5) + print("PASS: acknowledged FUSE mutations and fsync survived node SIGKILL and restart") + + +if __name__ == "__main__": + main() diff --git a/scripts/test_benchmark_fuse.py b/scripts/test_benchmark_fuse.py new file mode 100644 index 000000000..6264eb005 --- /dev/null +++ b/scripts/test_benchmark_fuse.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +"""Regression for draining large benchmark samples before worker join.""" + +import importlib.util +import multiprocessing +import pathlib +import unittest + + +SCRIPT = pathlib.Path(__file__).with_name("benchmark-fuse.py") +SPEC = importlib.util.spec_from_file_location("benchmark_fuse", SCRIPT) +benchmark = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(benchmark) + + +def send_large_result(queue): + queue.put((list(range(30000)), None)) + + +class BenchmarkWorkerResultsTest(unittest.TestCase): + def test_large_sample_queue_does_not_block_worker_exit(self): + queue = multiprocessing.Queue() + processes = [multiprocessing.Process(target=send_large_result, args=(queue,)) for _ in range(2)] + for process in processes: + process.start() + outcomes = benchmark.collect_worker_results(processes, queue, timeout=10) + self.assertEqual([30000, 30000], [len(samples) for samples, _ in outcomes]) + self.assertTrue(all(process.exitcode == 0 for process in processes)) + + +if __name__ == "__main__": + unittest.main()