Always end the transaction, even when a commit or discard raises

docs/source/changelog.rst

@@ -9,6 +9,9 @@ Fixes
 - Avoid mutating live ``BlockCache`` and ``BackgroundBlockCache`` instances
   when pickling (#2102)
 
+- End the transaction even when a commit or discard raises, so the filesystem
+  is not left in transaction mode and deferred temporary files are cleaned up
+
 2026.7.0
 --------
 

fsspec/implementations/tests/test_local.py

@@ -1,5 +1,6 @@
 import bz2
 import errno
+import glob
 import gzip
 import os
 import os.path
@@ -16,7 +17,12 @@
 import fsspec
 from fsspec import compression
 from fsspec.core import OpenFile, get_fs_token_paths, open_files
-from fsspec.implementations.local import LocalFileSystem, get_umask, make_path_posix
+from fsspec.implementations.local import (
+    LocalFileOpener,
+    LocalFileSystem,
+    get_umask,
+    make_path_posix,
+)
 from fsspec.tests.test_utils import WIN
 
 files = {
@@ -533,6 +539,37 @@ def test_transaction_with_compression(tmpdir):
         assert f.read() == "data"
 
 
+def test_transaction_ends_when_a_commit_fails(tmpdir):
+    # A failed commit must not leave the filesystem mid-transaction: instances
+    # are cached, so every later write would be deferred into a temporary file
+    # that nothing commits, and would vanish without an error.
+    fs = LocalFileSystem()
+    real_commit = LocalFileOpener.commit
+    tmp_before = set(glob.glob(os.path.join(tempfile.gettempdir(), "tmp*")))
+
+    def commit(self):
+        if os.path.basename(self.path) == "b":
+            raise PermissionError(self.path)
+        real_commit(self)
+
+    with patch.object(LocalFileOpener, "commit", commit):
+        with pytest.raises(PermissionError):
+            with fs.transaction:
+                for name in ("a", "b", "c"):
+                    with fs.open(str(tmpdir / name), "wb") as f:
+                        f.write(b"data")
+
+    assert fs._intrans is False
+    assert fs._transaction is None
+    assert not set(glob.glob(os.path.join(tempfile.gettempdir(), "tmp*"))) - tmp_before
+
+    # the instance is still usable, and a later write is not swallowed
+    path = str(tmpdir / "later")
+    with fs.open(path, "wb") as f:
+        f.write(b"data")
+    assert fs.cat(path) == b"data"
+
+
 def test_same_permissions_with_and_without_transaction(tmpdir):
     tmpdir = str(tmpdir)
 

fsspec/transaction.py

@@ -1,5 +1,8 @@
+import logging
 from collections import deque
 
+logger = logging.getLogger("fsspec")
+
 
 class Transaction:
     """Filesystem transaction write context
@@ -38,15 +41,32 @@ def start(self):
 
     def complete(self, commit=True):
         """Finish transaction: commit or discard all deferred files"""
-        while self.files:
-            f = self.files.popleft()
-            if commit:
-                f.commit()
-            else:
-                f.discard()
-        self.fs._intrans = False
-        self.fs._transaction = None
-        self.fs = None
+        f = None
+        try:
+            while self.files:
+                f = self.files.popleft()
+                if commit:
+                    f.commit()
+                else:
+                    f.discard()
+                f = None
+        finally:
+            # the file being processed when the error was raised is already
+            # off the queue; put it back so its temporary file is cleaned up
+            if f is not None:
+                self.files.appendleft(f)
+            # A failed commit or discard must still end the transaction.
+            # Leaving _intrans set would defer every later write on this
+            # filesystem into a temporary file that nothing ever commits,
+            # and instances are cached, so that would persist process-wide.
+            while self.files:
+                try:
+                    self.files.popleft().discard()
+                except Exception:
+                    logger.debug("Discarding deferred file failed", exc_info=True)
+            self.fs._intrans = False
+            self.fs._transaction = None
+            self.fs = None
 
 
 class FileActor:
@@ -82,9 +102,12 @@ def __init__(self, fs):
 
     def complete(self, commit=True):
         """Finish transaction: commit or discard all deferred files"""
-        if commit:
-            self.files.commit().result()
-        else:
-            self.files.discard().result()
-        self.fs._intrans = False
-        self.fs = None
+        try:
+            if commit:
+                self.files.commit().result()
+            else:
+                self.files.discard().result()
+        finally:
+            self.fs._intrans = False
+            self.fs._transaction = None
+            self.fs = None