fix: harden MultiprocessingWriter shutdown and error handling

CHANGELOG.md

@@ -12,6 +12,11 @@
 
 2. [#237](https://github.com/InfluxCommunity/influxdb3-python/pull/237): Makes the writing API simpler and more consistent with other v3 clients:
     - Further simplifies the `WriteApi` request path by constructing v2/v3 requests directly through `RestClient`, while preserving existing write behavior.
+1. [#241](https://github.com/InfluxCommunity/influxdb3-python/pull/241): Harden `MultiprocessingWriter` shutdown and error handling:
+    - Replaces assertion-based runtime state validation with explicit exceptions.
+    - Guarantees queue task completion when worker writes fail.
+    - Adds idempotent `close()` with bounded worker shutdown and a configurable `close_timeout`.
+    - Ensures `on_shutdown` is invoked at most once.
 
 ## 0.21.0 [2026-08-27]
 

influxdb_client_3/write_client/client/util/multiprocessing_helper.py

@@ -6,6 +6,7 @@
 """
 import logging
 import multiprocessing
+import os
 import queue
 
 from influxdb_client_3 import write_client_options
@@ -121,12 +122,11 @@ def main():
 
     """
 
-    __started__ = False
-
     def __init__(self,
                  start_method='spawn',
                  process_ttl=300,
                  on_shutdown=None,
+                 close_timeout=60,
                  **kwargs
                  ) -> None:
         """
@@ -139,6 +139,7 @@ def __init__(self,
         :param process_ttl: The timeout in seconds for waiting for data in the underlying queue.
         :param on_shutdown: The callback function called when the worker process is shut down
                or when `MultiprocessingWriter` class start closing.
+        :param close_timeout: The timeout in seconds for waiting for the worker to shut down gracefully.
         :param kwargs: Arguments are passed into the ``WriteApi`` and ``write_client_options``.
             Common arguments include: `host`, `token`, `database`, `org`, `write_options`, `success_callback`,
             `error_callback`, `retry_callback`, `default_header`, and `rest_client`.
@@ -170,7 +171,11 @@ def __init__(self,
 
         self.ctx = multiprocessing.get_context(start_method)
         self.on_shutdown = on_shutdown
+        self.close_timeout = close_timeout
         self.disposed = self.ctx.Value('i', 0)
+        self._shutdown_called = self.ctx.Value('i', 0)
+        self.__started__ = False
+        self._closed = False
         self.process = self.ctx.Process(target=self.run, args=(write_api, self.disposed, process_ttl, self.on_shutdown))
         self.kwargs = kwargs
         self.queue_ = self.ctx.JoinableQueue()
@@ -184,11 +189,35 @@ def write(self, **kwargs) -> None:
         :param kwargs: arguments are passed into the `` write `` function of ``WriteApi``
         :return: None
         """
-        assert self.__started__ is True, 'Cannot write data: the writer is not started.'
-        if self.disposed.value == 0:
-            self.queue_.put(kwargs)
-        else:
-            raise Exception('Cannot write data: the writer is closed.')
+        if self.disposed.value != 0 or self._closed:
+            raise RuntimeError('Cannot write data: the writer is closed.')
+        if not self.__started__:
+            raise RuntimeError('Cannot write data: the writer is not started.')
+        self.queue_.put(kwargs)
+
+    def _call_on_shutdown(self, callback=None) -> None:
+        """Invoke the shutdown callback once across the parent and worker processes."""
+        callback = self.on_shutdown if callback is None else callback
+        if callback is None:
+            return
+
+        with self._shutdown_called.get_lock():
+            if self._shutdown_called.value != 0:
+                return
+            self._shutdown_called.value = 1
+
+        try:
+            callback()
+        except Exception:
+            logger.exception("The multiprocessing writer shutdown callback failed")
+
+    @staticmethod
+    def _close_write_api(write_api: WriteApi) -> None:
+        """Close the worker's WriteApi without preventing process shutdown."""
+        try:
+            write_api.close()
+        except Exception:
+            logger.exception("The multiprocessing writer failed to close the WriteApi")
 
     def run(self, write_api: WriteApi, disposed, process_ttl, on_shutdown) -> None:
         """
@@ -214,24 +243,32 @@ def run(self, write_api: WriteApi, disposed, process_ttl, on_shutdown) -> None:
                 next_record = self.queue_.get(timeout=process_ttl)
             except queue.Empty:
                 if disposed.value == 0:
-                    write_api.close()
+                    self._close_write_api(write_api)
                     disposed.value = 1
-                    if on_shutdown is not None:
-                        on_shutdown()
+                    self._call_on_shutdown(on_shutdown)
+                break
+
+            try:
+                if type(next_record) is _PoisonPill:
+                    # Poison pill means break the loop
+                    logger.info("flushing data...")
+                    self._close_write_api(write_api)
+                    logger.info("closed")
                     break
 
-            if type(next_record) is _PoisonPill:
-                # Poison pill means break the loop
-                logger.info("flushing data...")
-                write_api.close()
-                logger.info("closed")
+                try:
+                    write_api.write(**next_record)
+                except Exception:
+                    logger.exception("The multiprocessing writer failed to write a record")
+            finally:
                 self.queue_.task_done()
-                break
-            write_api.write(**next_record)
-            self.queue_.task_done()
 
     def start(self) -> None:
         """Start an independent process for writing data into InfluxDB."""
+        if self._closed or self.disposed.value != 0:
+            raise RuntimeError('Cannot start the writer after it has been closed.')
+        if self.__started__:
+            raise RuntimeError('The writer is already started.')
         self.process.start()
         self.__started__ = True
 
@@ -245,16 +282,33 @@ def __enter__(self):
 
     def __exit__(self, exc_type, exc_value, traceback):
         """Exit the runtime context related to this object."""
-        self.__del__()
+        self.close()
+
+    def close(self) -> None:
+        """Flush queued writes and close the worker process once."""
+        if self._closed:
+            return
+
+        self._closed = True
+        is_worker_process = getattr(self.process, 'pid', None) == os.getpid()
+        try:
+            if self.__started__ and not is_worker_process:
+                if self.disposed.value == 0:
+                    self.queue_.put(_PoisonPill())
+                self.process.join(timeout=self.close_timeout)
+                if self.process.is_alive():
+                    logger.warning("The multiprocessing writer worker did not shut down before the timeout")
+                    self.process.terminate()
+                    self.process.join(timeout=self.close_timeout)
+        finally:
+            self.__started__ = False
+            self.disposed.value = 1
+            if not is_worker_process:
+                self._call_on_shutdown()
 
     def __del__(self):
-        """Dispose of the client and write_api."""
-        if self.__started__ and self.disposed.value == 0:
-            self.queue_.put(_PoisonPill())
-            self.queue_.join()
-            self.process.join()
-            self.queue_ = None
-        self.__started__ = False
-        self.disposed.value = 1
-        if self.on_shutdown is not None:
-            self.on_shutdown()
+        """Best-effort cleanup for writers that were not explicitly closed."""
+        try:
+            self.close()
+        except Exception:
+            logger.debug("The multiprocessing writer cleanup failed", exc_info=True)

tests/test_influxdb_client_3_integration.py

@@ -672,17 +672,6 @@ def test_query_timeout(self):
         with self.assertRaisesRegex(InfluxDB3ClientQueryError, ".*Deadline Exceeded.*"):
             localClient.query("SELECT * FROM data")
 
-    def test_query_timeout_per_call_override(self):
-        localClient = InfluxDBClient3(
-            host=self.host,
-            token=self.token,
-            database=self.database,
-            query_timeout=3,
-        )
-
-        with self.assertRaisesRegex(InfluxDB3ClientQueryError, ".*Deadline Exceeded.*"):
-            localClient.query("SELECT * FROM data", timeout=0.000001)
-
     def test_write_timeout_per_call_override(self):
 
         ErrorResult = {"rt": None, "rd": None, "rx": None}