chroma: use force_fpcalc=True to prevent fd exhaustion

beetsplug/chroma.py

@@ -83,8 +83,10 @@ def acoustid_match(log, path):
     _matches, _fingerprints, and _acoustids dictionaries accordingly.
     """
     try:
-        duration, fp = acoustid.fingerprint_file(util.syspath(path))
-    except acoustid.FingerprintGenerationError as exc:
+        duration, fp = acoustid.fingerprint_file(
+            util.syspath(path), force_fpcalc=True
+        )
+    except (acoustid.FingerprintGenerationError, TypeError) as exc:
         log.error(
             "fingerprinting of {} failed: {}",
             util.displayable_path(repr(path)),
@@ -442,15 +444,17 @@ def fingerprint_item(log, item, write=False, quiet=False):
     else:
         log.info("{.filepath}: fingerprinting", item)
         try:
-            _, fp = acoustid.fingerprint_file(util.syspath(item.path))
+            _, fp = acoustid.fingerprint_file(
+                util.syspath(item.path), force_fpcalc=True
+            )
             item.acoustid_fingerprint = fp.decode()
             if write:
                 log.info("{.filepath}: writing fingerprint", item)
                 item.try_write()
             if item._db:
                 item.store()
             return item.acoustid_fingerprint
-        except acoustid.FingerprintGenerationError as exc:
+        except (acoustid.FingerprintGenerationError, TypeError) as exc:
             log.info("fingerprint generation failed: {}", exc)
     return None
 

docs/changelog.rst

@@ -21,6 +21,10 @@ New features
 Bug fixes
 ~~~~~~~~~
 
+- :doc:`plugins/chroma`: Fix file descriptor exhaustion when fingerprinting
+  large libraries. The chroma plugin now uses the ``fpcalc`` binary directly
+  (via ``force_fpcalc=True``) instead of routing through audioread's GStreamer
+  backend, which leaked fds on fingerprinting errors. :bug:`5171`
 - Add ``editor`` config option to allow users to permanently set their preferred
   editor, overriding ``$VISUAL`` and ``$EDITOR`` environment variables.
   :bug:`6641`

test/plugins/test_chroma.py

@@ -71,6 +71,58 @@ def test_chroma_search_close(self, compare_fingerprints):
         assert TEST_TITLE_1 in output.split("\n")[0]
 
 
+class TestAcoustidMatch:
+    """Tests for acoustid_match() covering the force_fpcalc fix (#5171)."""
+
+    def _make_log(self):
+        log = MagicMock()
+        log.error = MagicMock()
+        return log
+
+    @patch("beetsplug.chroma.acoustid.fingerprint_file")
+    def test_fingerprint_file_called_with_force_fpcalc(self, mock_fp):
+        """acoustid_match must pass force_fpcalc=True to avoid GStreamer fd leak."""
+        mock_fp.return_value = (30, b"FINGERPRINT")
+        with patch("beetsplug.chroma.acoustid.lookup") as mock_lookup:
+            mock_lookup.return_value = {"status": "ok", "results": []}
+            chroma.acoustid_match(self._make_log(), b"/fake/path.mp3")
+        mock_fp.assert_called_once()
+        _, kwargs = mock_fp.call_args
+        assert kwargs.get("force_fpcalc") is True
+
+    @patch("beetsplug.chroma.acoustid.fingerprint_file")
+    def test_type_error_logged_not_raised(self, mock_fp):
+        """TypeError from old pyacoustid without force_fpcalc must be caught."""
+        mock_fp.side_effect = TypeError(
+            "unexpected keyword argument 'force_fpcalc'"
+        )
+        log = self._make_log()
+        chroma.acoustid_match(log, b"/fake/path.mp3")  # must not raise
+        log.error.assert_called_once()
+
+
+class TestFingerprintItem:
+    """Tests for fingerprint_item() covering the force_fpcalc TypeError fix."""
+
+    def _make_log(self):
+        log = MagicMock()
+        log.info = MagicMock()
+        return log
+
+    @patch("beetsplug.chroma.acoustid.fingerprint_file")
+    def test_type_error_returns_none_not_raised(self, mock_fp):
+        """TypeError from old pyacoustid in fingerprint_item must be caught."""
+        mock_fp.side_effect = TypeError(
+            "unexpected keyword argument 'force_fpcalc'"
+        )
+        item = MagicMock()
+        item.length = 30
+        item.acoustid_fingerprint = None
+        log = self._make_log()
+        result = chroma.fingerprint_item(log, item)  # must not raise
+        assert result is None
+
+
 def _seed_acoustid_match(item_path: bytes = b"/fake/path.mp3") -> Item:
     """Seed the chroma module-level match cache as if acoustid had run."""
     chroma._matches[item_path] = (