Fix snapshot and filesystem integration regressions

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(),

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)."

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,17 +494,23 @@
         ;; 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))
                                                       (throw (java.io.IOException. "Input/output error"))
                                                       (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))))))

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)

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")
+	}
+}