diff --git a/internal/backup/decode_test.go b/internal/backup/decode_test.go index 22337893f..05ef76b7d 100644 --- a/internal/backup/decode_test.go +++ b/internal/backup/decode_test.go @@ -439,8 +439,8 @@ func TestDecodeSnapshot_ChunkBlobGCKeyspacesDropAsInternal(t *testing.T) { t.Parallel() root := t.TempDir() b := newSnapBuilder(0) - b.WriteEntry([]byte(S3ChunkRefRCPrefix+"sha"), 1, []byte("rc"), false, 0, snapshotEncStateCleartx) b.WriteEntry([]byte(S3ChunkBlobGCQueuePrefix+"ts|sha"), 2, []byte{}, false, 0, snapshotEncStateCleartx) + b.WriteEntry([]byte(S3ChunkRefRCPrefix+"sha"), 1, []byte("rc"), false, 0, snapshotEncStateCleartx) res, err := DecodeSnapshot(bytes.NewReader(b.Bytes()), DecodeOptions{ OutRoot: root, Adapters: AllAdapters(), diff --git a/jepsen/src/elastickv/filesystem_workload.clj b/jepsen/src/elastickv/filesystem_workload.clj index 43b186fe3..524e3a735 100644 --- a/jepsen/src/elastickv/filesystem_workload.clj +++ b/jepsen/src/elastickv/filesystem_workload.clj @@ -18,10 +18,12 @@ The backend is optimistically concurrent: open/write/close on a file that another mount is writing can fail with EAGAIN (a write conflict), and a follower mount that lost its forwarded proposal may report EIO. Both are - retried a bounded number of times (`--max-retries`); a write that exhausts - its retries is recorded as indeterminate because an attempt may already - have been committed. Reads additionally retry ENOENT, short and malformed - records: FUSE splits open(2) into LOOKUP + OPEN, so a reader racing a + retried a bounded number of times (`--max-retries`) for reads and for + writes that provably failed before mutation. A possibly applied write is + immediately indeterminate: replaying it could overwrite a newer value and + give one history operation multiple effects. Reads additionally retry + ENOENT, short and malformed records: FUSE splits open(2) into LOOKUP + OPEN, + so a reader racing a rename on another mount can briefly miss the replaced inode. A main-phase read that exhausts those retries is still a plain :fail, because under write contention a bounded retry window cannot tell that race from loss. @@ -102,7 +104,7 @@ 10) (def ^:const default-max-retries - "Attempts per operation before giving up on EAGAIN/EIO." + "Attempt limit for reads/setup and pre-mutation write failures." 8) (def ^:const retry-backoff-ms 20) @@ -474,12 +476,9 @@ (instance? FileAlreadyExistsException e))) (defn invoke-write! - "Executes a write, retrying retryable errors. The outcome is classified - over every attempt, not just the last exception: an op is :fail only when - all attempts failed before touching the backend, because a single earlier - write/close/rename error may already have been committed and a later - EAGAIN at open (which RandomAccessFile reports as FileNotFoundException) - must not downgrade it to a definite failure." + "Retries only errors known to precede mutation. A write/close/rename or + SSH failure may follow a committed write, so it ends the operation as + indeterminate instead of replaying an old value over a concurrent write." [{:keys [local? mount-dir write-mode max-retries]} test node op k v] (let [maybe-applied? (volatile! false) attempt! (fn [] @@ -497,7 +496,9 @@ (vreset! maybe-applied? true)) (throw e))))] (try - (with-retries max-retries attempt!) + (with-retries max-retries + #(and (pre-mutation-error? %) (retryable-error? %)) + attempt!) (assoc op :type :ok) (catch Exception e (if (and (not @maybe-applied?) (pre-mutation-error? e)) @@ -778,7 +779,7 @@ [nil "--concurrency N" "Number of worker threads (a multiple of --threads-per-key; at least the node count to final-read every mount)." :default default-concurrency :parse-fn #(Integer/parseInt %)] - [nil "--max-retries N" "Attempts per operation before an EAGAIN/EIO is reported." + [nil "--max-retries N" "Attempt limit; possibly applied writes are never replayed." :default default-max-retries :parse-fn #(Integer/parseInt %)] [nil "--allow-non-fuse-mount" "Skip the check that every --mount-dirs entry is a FUSE mount (local runs only)." diff --git a/jepsen/test/elastickv/filesystem_workload_test.clj b/jepsen/test/elastickv/filesystem_workload_test.clj index c8a3dea0d..6a78a034d 100644 --- a/jepsen/test/elastickv/filesystem_workload_test.clj +++ b/jepsen/test/elastickv/filesystem_workload_test.clj @@ -415,7 +415,7 @@ (finally (delete-tree! dir))))) -(deftest local-client-write-retries-exhausted-is-indeterminate +(deftest local-client-ambiguous-write-is-indeterminate (let [dir (scratch-dir)] (try (let [test-map (workload/elastickv-filesystem-test {:nodes ["n1"] :mount-dirs [dir] @@ -432,7 +432,56 @@ (finally (delete-tree! dir))))) -(deftest write-classification-uses-every-attempt +(deftest an-ambiguous-write-cannot-replay-over-a-newer-write + (doseq [mode [:overwrite :rename]] + (let [dir (scratch-dir)] + (try + (workload/local-setup! dir [0]) + (let [path (workload/register-path dir 0) + calls (atom 0) + observed (atom []) + overwrite! workload/local-overwrite! + rename! workload/local-rename-write! + apply! (fn [v] + (case mode + :overwrite (overwrite! path v) + :rename (rename! dir 0 v))) + attempt! (fn [v] + (swap! calls inc) + (apply! v) + (when (= 1 @calls) + (swap! observed conj (workload/local-read path)) + (apply! 29) + (swap! observed conj (workload/local-read path)) + (throw (java.io.IOException. "Resource temporarily unavailable"))))] + (with-redefs [workload/local-overwrite! (fn [_ v] (attempt! v)) + workload/local-rename-write! (fn [_ _ v] (attempt! v))] + (let [result (workload/invoke-write! + {:local? true :mount-dir dir :write-mode mode :max-retries 3} + {} "n1" {:type :invoke :f :write :value (independent/tuple 0 24)} 0 24)] + (is (= [24 29] @observed)) + (is (= :info (:type result)) (str mode)) + (is (= 1 @calls) (str mode " must not replay a possibly applied write")) + (is (= 29 (workload/local-read path)) (str mode " must retain the newer write"))))) + (finally + (delete-tree! dir)))))) + +(deftest remote-ambiguous-writes-are-not-replayed + (doseq [mode [:overwrite :rename]] + (let [calls (atom 0)] + (with-redefs-fn + {#'workload/remote-exec! (fn [& _] + (swap! calls inc) + (throw (ex-info "SSH command failed" + {:exit 1 :err "Input/output error"})))} + (fn [] + (let [result (workload/invoke-write! + {:local? false :mount-dir "/mnt/fs" :write-mode mode :max-retries 3} + {} "n1" {:type :invoke :f :write :value (independent/tuple 0 24)} 0 24)] + (is (= :info (:type result))) + (is (= 1 @calls)))))))) + +(deftest writes-retry-only-pre-mutation-errors (let [dir (scratch-dir)] (try (let [test-map (workload/elastickv-filesystem-test {:nodes ["n1"] :mount-dirs [dir] @@ -445,8 +494,7 @@ ;; Every attempt fails at open: nothing reached the backend -> :fail. (with-redefs [workload/local-overwrite! (fn [_ _] (throw (eagain)))] (is (= :fail (:type (write))))) - ;; First attempt fails after open (write/close EIO), later attempts - ;; fail at open: the first one may have committed -> :info. + ;; A write/close EIO may follow application: never try it again. (let [calls (atom 0)] (with-redefs [workload/local-overwrite! (fn [_ _] (if (= 1 (swap! calls inc)) @@ -454,8 +502,15 @@ (throw (eagain))))] (let [w (write)] (is (= :info (:type w))) - (is (= 3 @calls)) - (is (str/includes? (:error w) "Resource temporarily unavailable"))))) + (is (= 1 @calls)) + (is (str/includes? (:error w) "Input/output error"))))) + ;; Retrying an open failure remains safe because it had no effect. + (let [calls (atom 0)] + (with-redefs [workload/local-overwrite! (fn [_ _] + (when (< (swap! calls inc) 3) + (throw (eagain))))] + (is (= :ok (:type (write)))) + (is (= 3 @calls)))) ;; A genuinely missing file is still a definite failure. (with-redefs [workload/local-overwrite! (fn [_ _] (throw (java.nio.file.NoSuchFileException. "reg-0")))] (is (= :fail (:type (write)))))) diff --git a/main.go b/main.go index 95129d84a..f4f9387c4 100644 --- a/main.go +++ b/main.go @@ -1032,6 +1032,9 @@ func runFilesystemLeaseReaper( ticker := time.NewTicker(interval) defer ticker.Stop() for { + if ctx.Err() != nil { + return nil + } stats, err := reaper.ReapExpiredOpenHandleLeases(ctx, 0) if err != nil && ctx.Err() == nil { slog.WarnContext(ctx, "filesystem lease reaper failed", "err", err) diff --git a/main_filesystem_reaper_test.go b/main_filesystem_reaper_test.go new file mode 100644 index 000000000..42c32565e --- /dev/null +++ b/main_filesystem_reaper_test.go @@ -0,0 +1,29 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestRunFilesystemLeaseReaperSkipsCanceledContext(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + reaper := &testFilesystemLeaseReaper{cancel: cancel} + + require.NoError(t, runFilesystemLeaseReaper(ctx, reaper, time.Nanosecond)) + require.Zero(t, reaper.calls) +} + +func TestRunFilesystemLeaseReaperCancellationWinsOverReadyTick(t *testing.T) { + for range 50 { + ctx, cancel := context.WithCancel(t.Context()) + t.Cleanup(cancel) + reaper := &testFilesystemLeaseReaper{cancel: cancel} + + require.NoError(t, runFilesystemLeaseReaper(ctx, reaper, time.Nanosecond)) + require.Equal(t, 2, reaper.calls, "a ready tick must not start another reap after cancellation") + } +}