diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py
index 210dfb0..d4173c3 100644
--- a/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py
+++ b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py
@@ -25,12 +25,18 @@
 import gzip
 import platform
 import datetime
+import threading
+import time
 from concurrent.futures import ThreadPoolExecutor
 from importlib.metadata import version
 
 import requests
 from fiftyone_pipeline_engines.engine import Engine
 
+from .share_usage_backoff import ShareUsageBackOff, \
+    FAILURES_TO_ENTER_RECOVERY_DEFAULT, FAILURES_WINDOW_SECONDS_DEFAULT, \
+    INITIAL_RECOVERY_DELAY_SECONDS_DEFAULT, \
+    MAX_RECOVERY_DELAY_SECONDS_DEFAULT, RECOVERY_MULTIPLIER_DEFAULT
 from .share_usage_evidencekeyfilter import ShareUsageEvidenceKeyFilter, \
     ShareUsageTrackerEvidenceKeyFilter
 from .share_usage_tracker import ShareUsageTracker
@@ -44,6 +50,21 @@
 # The default number of seconds to wait for the server connection before failing the request when sending usage data
 REQUEST_TIMEOUT = 60
 
+# The default maximum number of entries waiting to be sent, unless ten
+# times the requested package size is larger.
+MAX_QUEUE_SIZE_DEFAULT = 1000
+
+# While the queue is full, a message is logged for the first entry discarded
+# and then once for every this many more, never one for each entry.
+QUEUE_FULL_LOG_EVERY = 1000
+
+# The most characters of an answer's content that are logged.
+LOGGED_CONTENT_LENGTH = 500
+
+# The longest single wait during a pause. Longer pauses are waited out in
+# parts, as a timer cannot wait for any length of time on every platform.
+LONGEST_WAIT_SECONDS = 86400
+
 
 class ShareUsage(Engine):
     def __init__(
@@ -57,7 +78,13 @@ def __init__(
         endpoint = "https://devices-v4.51degrees.com/new.ashx",
         request_timeout=REQUEST_TIMEOUT,
         share_all = False,
-        track_session = False):
+        track_session = False,
+        max_queue_size = None,
+        failures_to_enter_recovery = FAILURES_TO_ENTER_RECOVERY_DEFAULT,
+        failures_window_seconds = FAILURES_WINDOW_SECONDS_DEFAULT,
+        initial_recovery_delay_seconds = INITIAL_RECOVERY_DELAY_SECONDS_DEFAULT,
+        max_recovery_delay_seconds = MAX_RECOVERY_DELAY_SECONDS_DEFAULT,
+        recovery_multiplier = RECOVERY_MULTIPLIER_DEFAULT):
         """!
         Constructor for ShareUsage element
         
@@ -99,6 +126,26 @@ def __init__(
         @param share_all: If True, all evidence is shared apart from the
         email address, the salt used for hashed email identifiers, and the
         request's query string and path. False by default.
+        @type max_queue_size: int
+        @param max_queue_size: The most entries held while waiting to be
+        sent. When the queue is full, new entries are dropped. The default
+        is 1000, or ten times requested_package_size if that is larger.
+        @type failures_to_enter_recovery: int
+        @param failures_to_enter_recovery: The number of failed sends within
+        failures_window_seconds that pauses sending, from 1 to 100. Usage
+        data is still queued while sending is paused. 10 by default.
+        @type failures_window_seconds: int
+        @param failures_window_seconds: The time in seconds over which
+        failed sends are counted. 100 by default.
+        @type initial_recovery_delay_seconds: float
+        @param initial_recovery_delay_seconds: The length in seconds of the
+        first pause after failed sends. 2.0 by default.
+        @type max_recovery_delay_seconds: float
+        @param max_recovery_delay_seconds: The longest a pause can be, in
+        seconds. 300.0 by default.
+        @type recovery_multiplier: float
+        @param recovery_multiplier: What the pause is multiplied by each time
+        the send tried when a pause ends fails. 2.0 by default.
         """
 
         super(ShareUsage, self).__init__()
@@ -140,30 +187,102 @@ def __init__(
 
         self.tracker = ShareUsageTracker(interval = interval)
 
-        # Initialise share data list 
+        # The entries waiting to be sent, oldest first.
         self.share_data = []
 
+        # The entries discarded since the queue last had room.
+        self._discarded = 0
+
+        # A message holds at most twice the requested package size.
+        self._message_size = 2 * requested_package_size
+
+        self.max_queue_size = max_queue_size if max_queue_size is not None \
+            else max(MAX_QUEUE_SIZE_DEFAULT, 10 * requested_package_size)
+        if self.max_queue_size < requested_package_size:
+            raise ValueError(
+                "max_queue_size cannot be smaller than "
+                "requested_package_size.")
+
+        # Pauses sending after repeated failed sends.
+        self.backoff = ShareUsageBackOff(
+            failures_to_enter_recovery,
+            failures_window_seconds,
+            initial_recovery_delay_seconds,
+            max_recovery_delay_seconds,
+            recovery_multiplier)
+
+        # Guards the queue, the sending flag and the resume timer.
+        self._lock = threading.Lock()
+        # True while a thread is sending.
+        self._sending = False
+        # Starts sending again when a pause ends.
+        self._resume_timer = None
+        # Set when the element closes.
+        self._closing = threading.Event()
+
         self.constant_xml = None
 
-        # Reusable thread pool for sending usage in background
-        # According to the doc on ThreadPoolExecutor - all threads will be joined before the interpreter can exit
-        # this ensures that all the data will be sent correctly before the actual exit happens
+        # One thread sends at a time, so a failing service gets one request
+        # at a time. According to the doc on ThreadPoolExecutor - all threads
+        # will be joined before the interpreter can exit, so a message being
+        # sent is finished before the actual exit happens.
 
-        self.thread_pool = ThreadPoolExecutor(thread_name_prefix='ShareUsage')
+        self.thread_pool = ThreadPoolExecutor(
+            max_workers=1, thread_name_prefix='ShareUsage')
 
     def __del__(self):
-        self.thread_pool.shutdown(wait=True)
+        thread_pool = getattr(self, "thread_pool", None)
+        if thread_pool is not None:
+            thread_pool.shutdown(wait=True)
 
     def close(self):
 
         """!
-        Wait for any usage data already being sent, then stop the threads
-        that send it. The pipeline calls this when it is closed.
+        Wait for any usage data already being sent, then send what is still
+        queued and stop the threads that send it. The pipeline calls this
+        when it is closed. Closing does not wait for a pause after failed
+        sends, and sends nothing while paused.
 
         """
 
         super(ShareUsage, self).close()
+        self._closing.set()
+        with self._lock:
+            if self._resume_timer is not None:
+                self._resume_timer.cancel()
+                self._resume_timer = None
         self.thread_pool.shutdown(wait=True)
+        self._send_remaining()
+
+    def _send_remaining(self):
+        """!
+        Send the entries still queued when closing, even if there are fewer
+        than the requested package size. Nothing is sent while paused, or
+        after a message fails, and the entries left are discarded and
+        counted in the log.
+        """
+
+        while True:
+            with self._lock:
+                queued = len(self.share_data)
+                if queued == 0:
+                    return
+                paused = self.backoff.remaining(time.monotonic()) > 0
+                if not paused:
+                    share_data = self.share_data[:self._message_size]
+                    del self.share_data[:self._message_size]
+            if paused:
+                break
+            data = f'<Devices version="{SHARE_USAGE_VERSION}">{"".join(share_data)}</Devices>'
+            if not self.send_thread(self.endpoint, data):
+                with self._lock:
+                    queued = len(self.share_data)
+                break
+        if queued > 0:
+            self._log(
+                "info",
+                f"{queued} usage sharing entries were not sent, as sending "
+                "had failed when usage sharing closed.")
 
     def get_constant_xml(self):
         if self.constant_xml is None:
@@ -224,29 +343,125 @@ def get_evidence_key_filter(self):
 
     def share_send_usage(self):
         """!
-        Internal method to send the share usage bundle to the 51Degrees servers
+        Internal method to start sending the queued usage data to the
+        51Degrees servers. Nothing starts if a thread is already sending,
+        if fewer than requested_package_size entries are queued, or while
+        sending is paused after failed sends, in which case a timer starts
+        sending when the pause ends.
         """
 
-        share_data = self.share_data
-        self.share_data = []
+        with self._lock:
+            if self._sending or self._closing.is_set():
+                return
+            pause = self.backoff.remaining(time.monotonic())
+            if pause > 0:
+                self._schedule_resume(pause)
+                return
+            if len(self.share_data) < self.requested_package_size:
+                return
+            self._sending = True
+        try:
+            self.thread_pool.submit(self._send_while_queued)
+        except RuntimeError:
+            # The thread pool has been shut down, so nothing more is sent.
+            with self._lock:
+                self._sending = False
+
+    def _schedule_resume(self, pause):
+        """!
+        Start sending when the pause ends. Called with the lock held.
+        """
+
+        if self._resume_timer is None and not self._closing.is_set():
+            self._resume_timer = threading.Timer(
+                min(pause, LONGEST_WAIT_SECONDS), self._resume)
+            # The timer does not keep the process running.
+            self._resume_timer.daemon = True
+            self._resume_timer.start()
 
-        data = f'<Devices version="{SHARE_USAGE_VERSION}">{"".join(share_data)}</Devices>'
+    def _resume(self):
+        with self._lock:
+            self._resume_timer = None
+        self.share_send_usage()
+
+    def _send_while_queued(self):
+        """!
+        Send messages of requested_package_size entries until fewer are
+        queued or a failed send pauses sending.
+        """
 
-        self.thread_pool.submit(self.send_thread, self.endpoint, data)
+        try:
+            self._send_messages()
+        except Exception as e:
+            with self._lock:
+                self._sending = False
+            self._log("error", f"ShareUsage failed sending data: {e}")
+
+    def _send_messages(self):
+        while True:
+            with self._lock:
+                share_data = self.share_data[:self._message_size]
+                del self.share_data[:self._message_size]
+                if not share_data:
+                    self._sending = False
+                    return
+
+            data = f'<Devices version="{SHARE_USAGE_VERSION}">{"".join(share_data)}</Devices>'
+            accepted = self.send_thread(self.endpoint, data)
+
+            with self._lock:
+                resumed = False
+                pause = 0
+                if accepted:
+                    resumed = self.backoff.succeeded()
+                else:
+                    pause = self.backoff.failed(time.monotonic())
+                    if pause > 0:
+                        self._schedule_resume(pause)
+                done = pause > 0 or \
+                    len(self.share_data) < self.requested_package_size
+                if done:
+                    self._sending = False
+
+            if resumed:
+                self._log("info", "Usage sharing resumed after a send succeeded.")
+            if pause > 0:
+                self._log(
+                    "warning",
+                    f"Usage sharing paused for {pause} seconds after failed "
+                    "sends. Usage data is still queued, up to the maximum "
+                    "queue size, and sending starts again when the pause "
+                    "ends.")
+            if done:
+                return
 
     def send_thread(self, endpoint, data):
+        """!
+        Send one message of usage data.
+
+        @return: True if the service accepted it
+        """
+
         data = gzip.compress(bytearray(data, encoding='utf8'))
 
         # setting a reasonable time out and catching any exceptions
         try:
-            requests.post(endpoint, headers={"Content-Encoding": "gzip", "Content-Type": "text/xml"}, data=data,
+            response = requests.post(endpoint, headers={"Content-Encoding": "gzip", "Content-Type": "text/xml"}, data=data,
                           timeout=self.request_timeout)
         except Exception as e:
-            print(e)
             self._log(
                 "error",
                 f"ShareUsage failed sending {len(data)} bytes of data: {e}"
             )
+            return False
+        if response.status_code != 200:
+            self._log(
+                "warning",
+                f"ShareUsage sent {len(data)} bytes of data and the service "
+                f"answered {response.status_code} with content "
+                f"'{response.text[:LOGGED_CONTENT_LENGTH]}'.")
+            return False
+        return True
 
     def add_to_share_usage(self, data):
         """!
@@ -299,13 +514,33 @@ def add_to_share_usage(self, data):
                     xml += f"</{entry_key}>"
         
         xml += "</Device>"
- 
-        self.share_data.append(xml)
+
+        with self._lock:
+            # When the queue is full, the newest entries are dropped.
+            if len(self.share_data) >= self.max_queue_size:
+                self._discarded += 1
+                discarded = self._discarded
+                log_full = (discarded == 1 or
+                            discarded % QUEUE_FULL_LOG_EVERY == 0)
+            else:
+                discarded = 0
+                log_full = False
+                self._discarded = 0
+                self.share_data.append(xml)
+
+        if discarded > 0:
+            if log_full:
+                self._log(
+                    "warning",
+                    f"The usage sharing queue is full at "
+                    f"{self.max_queue_size} entries, so usage data is being "
+                    f"discarded. {discarded} entries have been discarded "
+                    "since the queue last had room.")
+            return
 
         # Send share usage data if data size greater than requested
         # package size
-        if len(self.share_data) >= self.requested_package_size:
-            self.share_send_usage()
+        self.share_send_usage()
 
     def process_internal(self, flow_data):
 
diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_backoff.py b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_backoff.py
new file mode 100644
index 0000000..bfaec05
--- /dev/null
+++ b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_backoff.py
@@ -0,0 +1,155 @@
+# *********************************************************************
+# This Original Work is copyright of 51 Degrees Mobile Experts Limited.
+# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House,
+# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU.
+#
+# This Original Work is licensed under the European Union Public Licence
+# (EUPL) v.1.2 and is subject to its terms as set out below.
+#
+# If a copy of the EUPL was not distributed with this file, You can obtain
+# one at https://opensource.org/licenses/EUPL-1.2.
+#
+# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be
+# amended by the European Commission) shall be deemed incompatible for
+# the purposes of the Work and the provisions of the compatibility
+# clause in Article 5 of the EUPL shall not apply.
+#
+# If using the Work as, or as part of, a network application, by
+# including the attribution notice(s) required under Article 5 of the EUPL
+# in the end user terms of the application under an appropriate heading,
+# such notice(s) shall fulfill the requirements of that article.
+# *********************************************************************
+
+from collections import deque
+import threading
+
+# The default number of failed sends within the failures window that
+# pauses sending.
+FAILURES_TO_ENTER_RECOVERY_DEFAULT = 10
+
+# The range of failed sends that can be set to pause sending.
+FAILURES_TO_ENTER_RECOVERY_MIN = 1
+FAILURES_TO_ENTER_RECOVERY_MAX = 100
+
+# The default time in seconds over which failed sends are counted.
+FAILURES_WINDOW_SECONDS_DEFAULT = 100
+
+# The default length in seconds of the first pause after failed sends.
+INITIAL_RECOVERY_DELAY_SECONDS_DEFAULT = 2.0
+
+# The default longest pause in seconds.
+MAX_RECOVERY_DELAY_SECONDS_DEFAULT = 300.0
+
+# The default multiplier applied to the pause each time the send tried
+# when a pause ends fails.
+RECOVERY_MULTIPLIER_DEFAULT = 2.0
+
+
+class ShareUsageBackOff:
+    """!
+    Decides when usage sharing may send. Sending pauses for the initial
+    delay once the set number of sends fail within the window. A failed
+    send after a pause multiplies the delay, up to the maximum, and a
+    successful send ends the back-off. Only sending pauses, so evidence is
+    still queued while paused. Each method takes the time from the caller,
+    in seconds.
+    """
+
+    def __init__(
+        self,
+        failures_to_enter_recovery = FAILURES_TO_ENTER_RECOVERY_DEFAULT,
+        failures_window_seconds = FAILURES_WINDOW_SECONDS_DEFAULT,
+        initial_recovery_delay_seconds = INITIAL_RECOVERY_DELAY_SECONDS_DEFAULT,
+        max_recovery_delay_seconds = MAX_RECOVERY_DELAY_SECONDS_DEFAULT,
+        recovery_multiplier = RECOVERY_MULTIPLIER_DEFAULT):
+
+        if not (isinstance(failures_to_enter_recovery, int)
+                and FAILURES_TO_ENTER_RECOVERY_MIN <= failures_to_enter_recovery
+                <= FAILURES_TO_ENTER_RECOVERY_MAX):
+            raise ValueError(
+                "failures_to_enter_recovery must be from "
+                f"{FAILURES_TO_ENTER_RECOVERY_MIN} to "
+                f"{FAILURES_TO_ENTER_RECOVERY_MAX}. "
+                f"Received: {failures_to_enter_recovery}.")
+        if not failures_window_seconds >= 1:
+            raise ValueError("failures_window_seconds must be 1 or more.")
+        if not initial_recovery_delay_seconds > 0:
+            raise ValueError(
+                "initial_recovery_delay_seconds must be more than 0.")
+        if not max_recovery_delay_seconds > 0:
+            raise ValueError(
+                "max_recovery_delay_seconds must be more than 0.")
+        if not recovery_multiplier >= 1:
+            raise ValueError("recovery_multiplier must be 1 or more.")
+
+        self.failures_to_enter_recovery = failures_to_enter_recovery
+        self.failures_window_seconds = failures_window_seconds
+        self.initial_recovery_delay_seconds = initial_recovery_delay_seconds
+        self.max_recovery_delay_seconds = max_recovery_delay_seconds
+        self.recovery_multiplier = recovery_multiplier
+
+        self._lock = threading.Lock()
+        # When each recent failed send happened, oldest first. Empty while
+        # backing off.
+        self._failures = deque()
+        # The length of the current pause, or zero when not backing off.
+        self._delay = 0
+        # When sending may start again.
+        self._resume_at = None
+
+    @property
+    def backing_off(self):
+        """!
+        True from the start of a pause until a send succeeds.
+        """
+        with self._lock:
+            return self._delay > 0
+
+    def remaining(self, now):
+        """!
+        The seconds left before sending may start again, or zero if it may
+        start now.
+        """
+        with self._lock:
+            if self._resume_at is None or self._resume_at <= now:
+                return 0
+            return self._resume_at - now
+
+    def failed(self, now):
+        """!
+        Record a failed send.
+
+        @return: the new pause in seconds if this failure started or
+        lengthened one, otherwise zero
+        """
+        with self._lock:
+            if self._delay > 0:
+                # The send tried when the last pause ended failed.
+                self._delay = min(
+                    self._delay * self.recovery_multiplier,
+                    self.max_recovery_delay_seconds)
+            else:
+                self._failures.append(now)
+                while self._failures[0] < now - self.failures_window_seconds:
+                    self._failures.popleft()
+                if len(self._failures) < self.failures_to_enter_recovery:
+                    return 0
+                self._failures.clear()
+                self._delay = min(
+                    self.initial_recovery_delay_seconds,
+                    self.max_recovery_delay_seconds)
+            self._resume_at = now + self._delay
+            return self._delay
+
+    def succeeded(self):
+        """!
+        Record a successful send.
+
+        @return: True if this ended a back-off
+        """
+        with self._lock:
+            if self._delay == 0:
+                return False
+            self._delay = 0
+            self._resume_at = None
+            return True
diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_tracker.py b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_tracker.py
index 694c3de..ef7f3d9 100644
--- a/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_tracker.py
+++ b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_tracker.py
@@ -59,7 +59,7 @@ def match(self, key, value):
 
         difference = time.time() - value
 
-        if difference > self.interval:
+        if difference >= self.interval:
             self.set_cache_value(key, value)
             return True
         else:
diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_backoff.py b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_backoff.py
new file mode 100644
index 0000000..762b309
--- /dev/null
+++ b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_backoff.py
@@ -0,0 +1,122 @@
+# *********************************************************************
+# This Original Work is copyright of 51 Degrees Mobile Experts Limited.
+# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House,
+# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU.
+#
+# This Original Work is licensed under the European Union Public Licence
+# (EUPL) v.1.2 and is subject to its terms as set out below.
+#
+# If a copy of the EUPL was not distributed with this file, You can obtain
+# one at https://opensource.org/licenses/EUPL-1.2.
+#
+# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be
+# amended by the European Commission) shall be deemed incompatible for
+# the purposes of the Work and the provisions of the compatibility
+# clause in Article 5 of the EUPL shall not apply.
+#
+# If using the Work as, or as part of, a network application, by
+# including the attribution notice(s) required under Article 5 of the EUPL
+# in the end user terms of the application under an appropriate heading,
+# such notice(s) shall fulfill the requirements of that article.
+# *********************************************************************
+
+import unittest
+
+from fiftyone_pipeline_engines_fiftyone.share_usage_backoff import \
+    ShareUsageBackOff
+
+
+def create(failures=3, initial=2, maximum=5):
+    """!
+    Three failures within a ten second window start a two second pause,
+    rising by two each time up to five.
+    """
+    return ShareUsageBackOff(failures, 10, initial, maximum, 2)
+
+
+class ShareUsageBackOffTests(unittest.TestCase):
+    """!
+    The rules that decide when usage sharing may send after failed sends,
+    with each test setting the time.
+    """
+
+    def test_pauses_after_failures_within_window(self):
+        backoff = create()
+
+        self.assertEqual(backoff.failed(0), 0)
+        self.assertEqual(backoff.failed(1), 0)
+        self.assertEqual(backoff.remaining(1), 0,
+            "Sending paused before enough sends failed.")
+        self.assertFalse(backoff.backing_off)
+
+        self.assertEqual(backoff.failed(2), 2)
+        self.assertTrue(backoff.backing_off)
+        self.assertEqual(backoff.remaining(3), 1)
+        self.assertEqual(backoff.remaining(4), 0)
+
+    def test_failures_outside_window_do_not_count(self):
+        backoff = create()
+
+        backoff.failed(0)
+        backoff.failed(1)
+
+        self.assertEqual(backoff.failed(11.5), 0,
+            "Failures older than the window were counted.")
+        self.assertFalse(backoff.backing_off)
+
+    def test_successes_between_failures_do_not_reset(self):
+        """!
+        The window counts failures, not failures in a row, so a high
+        failure rate pauses sending even when some sends succeed.
+        """
+        backoff = create()
+
+        backoff.failed(0)
+        self.assertFalse(backoff.succeeded())
+        backoff.failed(1)
+        self.assertFalse(backoff.succeeded())
+
+        self.assertEqual(backoff.failed(2), 2)
+
+    def test_failed_send_after_pause_multiplies_delay(self):
+        backoff = create(failures=1)
+
+        self.assertEqual(backoff.failed(0), 2)
+        self.assertEqual(backoff.failed(2), 4)
+        self.assertEqual(backoff.failed(6), 5,
+            "The delay grew past the maximum.")
+        self.assertEqual(backoff.failed(11), 5)
+        self.assertEqual(backoff.remaining(11), 5)
+
+    def test_success_ends_back_off(self):
+        backoff = create(failures=2)
+        backoff.failed(0)
+        backoff.failed(1)
+        backoff.failed(3)
+
+        self.assertTrue(backoff.succeeded(),
+            "A success did not end the back-off.")
+        self.assertFalse(backoff.backing_off)
+        self.assertEqual(backoff.remaining(4), 0)
+
+        # The next pause needs the full number of failures again, and
+        # starts from the initial delay.
+        self.assertEqual(backoff.failed(8), 0)
+        self.assertEqual(backoff.failed(9), 2)
+
+    def test_maximum_below_initial_limits_first_pause(self):
+        backoff = create(failures=1, initial=10, maximum=3)
+
+        self.assertEqual(backoff.failed(0), 3)
+
+    def test_invalid_settings_are_rejected(self):
+        for settings in [
+                (0, 10, 2, 5, 2),
+                (101, 10, 2, 5, 2),
+                (1, 0, 2, 5, 2),
+                (1, 10, 0, 5, 2),
+                (1, 10, float("nan"), 5, 2),
+                (1, 10, 2, 0, 2),
+                (1, 10, 2, 5, 0.5)]:
+            with self.assertRaises(ValueError, msg=str(settings)):
+                ShareUsageBackOff(*settings)
diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_tracker.py b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_tracker.py
new file mode 100644
index 0000000..63cc973
--- /dev/null
+++ b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_tracker.py
@@ -0,0 +1,82 @@
+# *********************************************************************
+# This Original Work is copyright of 51 Degrees Mobile Experts Limited.
+# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House,
+# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU.
+#
+# This Original Work is licensed under the European Union Public Licence
+# (EUPL) v.1.2 and is subject to its terms as set out below.
+#
+# If a copy of the EUPL was not distributed with this file, You can obtain
+# one at https://opensource.org/licenses/EUPL-1.2.
+#
+# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be
+# amended by the European Commission) shall be deemed incompatible for
+# the purposes of the Work and the provisions of the compatibility
+# clause in Article 5 of the EUPL shall not apply.
+#
+# If using the Work as, or as part of, a network application, by
+# including the attribution notice(s) required under Article 5 of the EUPL
+# in the end user terms of the application under an appropriate heading,
+# such notice(s) shall fulfill the requirements of that article.
+
+import unittest
+from unittest import mock
+
+from fiftyone_pipeline_engines_fiftyone.share_usage_tracker import \
+    ShareUsageTracker
+
+# The tracker reads the time through this name.
+TIME = "fiftyone_pipeline_engines_fiftyone.share_usage_tracker.time.time"
+
+INTERVAL_SECONDS = 60
+START = 1000000.0
+
+
+class ShareUsageTrackerTests(unittest.TestCase):
+
+    def track_at(self, tracker, key, seconds):
+        """!
+        Track the key with the clock at the given seconds after the start.
+        """
+        with mock.patch(TIME, return_value=START + seconds):
+            return tracker.track(key)
+
+    def share_at(self, tracker, key, seconds):
+        """!
+        Track the key as the element does, storing the time when it is shared.
+        """
+        with mock.patch(TIME, return_value=START + seconds):
+            shared = tracker.track(key)
+            if shared:
+                tracker.set_cache_value(key)
+            return shared
+
+    def test_repeat_inside_the_interval_is_not_shared(self):
+        tracker = ShareUsageTracker(interval=INTERVAL_SECONDS)
+        self.assertTrue(self.share_at(tracker, "a", 0))
+
+        self.assertFalse(
+            self.track_at(tracker, "a", INTERVAL_SECONDS - 1))
+
+    def test_repeat_at_the_interval_is_shared(self):
+        tracker = ShareUsageTracker(interval=INTERVAL_SECONDS)
+        self.assertTrue(self.share_at(tracker, "a", 0))
+
+        self.assertTrue(self.track_at(tracker, "a", INTERVAL_SECONDS))
+
+    def test_repeat_after_the_interval_is_shared(self):
+        tracker = ShareUsageTracker(interval=INTERVAL_SECONDS)
+        self.assertTrue(self.share_at(tracker, "a", 0))
+
+        self.assertTrue(self.track_at(tracker, "a", INTERVAL_SECONDS + 1))
+
+    def test_repeat_inside_the_interval_leaves_the_time(self):
+        tracker = ShareUsageTracker(interval=INTERVAL_SECONDS)
+        self.assertTrue(self.share_at(tracker, "a", 0))
+        self.assertFalse(self.share_at(tracker, "a", 30))
+
+        self.assertTrue(self.track_at(tracker, "a", INTERVAL_SECONDS))
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py
index 4f1bea4..f2973bf 100644
--- a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py
+++ b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py
@@ -20,9 +20,13 @@
 # such notice(s) shall fulfill the requirements of that article.
 # *********************************************************************
 
+import contextlib
+import io
 import platform
+import re
 import sys
 import threading
+import time
 from time import sleep
 import unittest
 from wsgiref.simple_server import make_server
@@ -59,25 +63,48 @@ def get_received(self):
 
 app = Flask(__name__)
 app.received = []
+# When each request arrived, whatever the answer.
+app.attempts = []
+# The status the receiver answers with. Only data it accepts is received.
+app.status = 200
+# The content the receiver answers with.
+app.body = ""
 
 @app.route('/', methods=['POST'])
 def server():
+    app.attempts.append(time.monotonic())
     data = gzip.decompress(request.data).decode("utf8")
-    str(app.received.append(data))
-    return ""
+    if app.status == 200:
+        app.received.append(data)
+    return app.body, app.status
+
+
+def client_ips(received):
+    """!
+    The client IPs in the usage data received.
+    """
+    ips = []
+    for data in received:
+        ips.extend(re.findall("<ClientIP>([^<]*)</ClientIP>", data))
+    return ips
 
 
 ######################################
 # The Tests
 
 class UsageTests(unittest.TestCase):
-    
+
     @classmethod
     def setUpClass(cls):
         cls.startReceiver()
 
     def setUp(self):
         app.received = []
+        app.attempts = []
+        app.status = 200
+        app.body = ""
+        self.requests = 0
+        self.logs = []
 
     @classmethod
     def tearDownClass(cls):
@@ -97,6 +124,295 @@ def waitForUsageThreads(self):
         # allowing the thread pool in ShareUsage to start sending data and release the GIL
         sleep(0.1)
 
+    def wait_for(self, condition, message):
+        """!
+        Wait for a condition, failing the test with the message if it is not
+        met within 30 seconds.
+        """
+        end = time.monotonic() + 30
+        while not condition():
+            if time.monotonic() > end:
+                self.fail(message())
+            sleep(0.02)
+
+    def back_off_pipeline(self, **settings):
+        """!
+        A pipeline whose usage sharing element sends each entry on its own.
+        The element is closed when the test ends.
+        """
+        usage = ShareUsage(
+            requested_package_size=1,
+            endpoint=testEndpoint,
+            **settings)
+        pipeline = PipelineBuilder().add(usage).build()
+        self.addCleanup(usage.close)
+        return usage, pipeline
+
+    def process_unique(self, pipeline):
+        """!
+        Process a request whose evidence no other request has, and return
+        its client IP.
+        """
+        ip = f"10.0.{self.requests // 250}.{self.requests % 250 + 1}"
+        self.requests += 1
+        flowdata = pipeline.create_flowdata()
+        flowdata.evidence.add("server.client-ip", ip)
+        flowdata.process()
+        return ip
+
+    def fail_sends(self, pipeline, count):
+        """!
+        Process requests one at a time, waiting for each to be sent, so each
+        is a separate failed send.
+        """
+        app.status = 500
+        for sent in range(1, count + 1):
+            self.process_unique(pipeline)
+            self.wait_for(
+                lambda: len(app.attempts) == sent,
+                lambda: f"Send {sent} was not made.")
+
+    def test_back_off_pauses_then_drains_queue(self):
+        """!
+        Ten failed sends pause sending for two seconds, the defaults.
+        Nothing is sent during the pause, the evidence processed during it
+        is queued, and all of it is sent once the pause ends and the
+        service answers, with no more requests processed.
+        """
+
+        usage, pipeline = self.back_off_pipeline()
+        self.fail_sends(pipeline, 10)
+        paused_at = time.monotonic()
+
+        queued = [self.process_unique(pipeline) for i in range(5)]
+        sleep(0.7)
+        self.assertEqual(
+            len(app.attempts), 10,
+            "Sends were made during the pause that ten failed sends "
+            "should have started.")
+
+        app.status = 200
+        self.wait_for(
+            lambda: all(ip in client_ips(app.received) for ip in queued),
+            lambda: "The evidence processed during the pause was not all "
+                f"sent once the service answered. Sent: "
+                f"{client_ips(app.received)}")
+        waited = app.attempts[10] - paused_at
+        self.assertGreaterEqual(
+            waited, 1.5,
+            f"Sending resumed {waited:.2f} seconds after the pause started, "
+            "before the two second pause ended.")
+
+    def test_back_off_close_while_paused(self):
+        """!
+        Closing while paused returns at once rather than waiting for the
+        pause to end, and sends nothing.
+        """
+
+        usage, pipeline = self.back_off_pipeline()
+        self.fail_sends(pipeline, 10)
+        self.process_unique(pipeline)
+        self.process_unique(pipeline)
+
+        start = time.monotonic()
+        usage.close()
+        took = time.monotonic() - start
+
+        self.assertLess(
+            took, 1.5,
+            f"Closing took {took:.2f} seconds, as it waited for the pause.")
+        self.assertEqual(
+            len(app.attempts), 10,
+            "A send was made during the pause or when closing.")
+
+    def test_back_off_delay_grows_to_maximum(self):
+        """!
+        Each failed send after a pause multiplies the delay, and the delay
+        stops growing at the maximum.
+        """
+
+        usage, pipeline = self.back_off_pipeline(
+            failures_to_enter_recovery=1,
+            initial_recovery_delay_seconds=0.5,
+            max_recovery_delay_seconds=1.0)
+        app.status = 500
+        for i in range(40):
+            self.process_unique(pipeline)
+
+        self.wait_for(
+            lambda: len(app.attempts) >= 4,
+            lambda: "Fewer than four sends were made.")
+        gaps = [app.attempts[i] - app.attempts[i - 1] for i in range(1, 4)]
+        shown = ", ".join(f"{gap:.2f}" for gap in gaps)
+        self.assertGreaterEqual(
+            gaps[0], 0.4,
+            f"The first pause was not the initial delay. Gaps: {shown}")
+        self.assertGreaterEqual(
+            gaps[1], 0.9,
+            f"The second pause was not doubled. Gaps: {shown}")
+        self.assertTrue(
+            0.9 <= gaps[2] < 1.9,
+            f"The third pause was not held at the maximum. Gaps: {shown}")
+
+    def test_back_off_full_queue_drops_new_evidence(self):
+        """!
+        When the queue fills during a pause, new evidence is dropped and
+        what was queued first is sent once sending resumes.
+        """
+
+        usage, pipeline = self.back_off_pipeline(
+            max_queue_size=4,
+            failures_to_enter_recovery=1,
+            initial_recovery_delay_seconds=2,
+            max_recovery_delay_seconds=2)
+        self.fail_sends(pipeline, 1)
+
+        kept = [self.process_unique(pipeline) for i in range(4)]
+        dropped = [self.process_unique(pipeline) for i in range(6)]
+        app.status = 200
+
+        self.wait_for(
+            lambda: all(ip in client_ips(app.received) for ip in kept),
+            lambda: "The evidence queued during the pause was not all sent. "
+                f"Sent: {client_ips(app.received)}")
+        sleep(0.5)
+        for ip in dropped:
+            self.assertNotIn(
+                ip, client_ips(app.received),
+                "Evidence processed when the queue was full was sent.")
+
+    def test_full_queue_logs_without_a_message_per_entry(self):
+        """!
+        Entries discarded because the queue is full are logged, but not with
+        a message for each entry discarded.
+        """
+
+        usage, pipeline = self.back_off_pipeline(
+            max_queue_size=4,
+            failures_to_enter_recovery=1,
+            initial_recovery_delay_seconds=2,
+            max_recovery_delay_seconds=2)
+        usage._log = lambda level, message: self.logs.append((level, message))
+        self.fail_sends(pipeline, 1)
+        self.logs.clear()
+
+        for i in range(10):
+            self.process_unique(pipeline)
+
+        full = [log for log in self.logs if "full" in log[1]]
+        self.assertGreaterEqual(
+            len(full), 1, "Discarding entries was not logged.")
+        self.assertLess(
+            len(full), 6, "A message was logged for each entry discarded.")
+        self.assertEqual("warning", full[0][0])
+
+    def test_failed_answer_logs_status_and_content(self):
+        """!
+        A send answered with anything other than 200 logs the status code and
+        the content of the answer.
+        """
+
+        usage, pipeline = self.back_off_pipeline()
+        usage._log = lambda level, message: self.logs.append((level, message))
+        app.body = "the service says no"
+
+        self.fail_sends(pipeline, 1)
+
+        self.wait_for(
+            lambda: any("500" in log[1] for log in self.logs),
+            lambda: f"The status code was not logged. Logged: {self.logs}")
+        failed = [log for log in self.logs if "500" in log[1]][0]
+        self.assertIn("the service says no", failed[1])
+
+    def test_failed_send_prints_nothing(self):
+        """!
+        A send that cannot be made is logged and does not print.
+        """
+
+        usage, pipeline = self.back_off_pipeline(request_timeout=1)
+        usage._log = lambda level, message: self.logs.append((level, message))
+        usage.endpoint = "http://127.0.0.1:1"
+        printed = io.StringIO()
+
+        with contextlib.redirect_stdout(printed):
+            self.process_unique(pipeline)
+            self.wait_for(
+                lambda: any("failed sending" in log[1] for log in self.logs),
+                lambda: f"The failed send was not logged. Logged: {self.logs}")
+
+        self.assertEqual("", printed.getvalue())
+
+    def test_message_holds_up_to_twice_the_package_size(self):
+        """!
+        A message holds the requested package size or more entries, up to
+        twice the package size, so a queue that filled during a pause drains
+        in fewer messages.
+        """
+
+        usage = ShareUsage(
+            requested_package_size=2,
+            endpoint=testEndpoint,
+            failures_to_enter_recovery=1,
+            initial_recovery_delay_seconds=1,
+            max_recovery_delay_seconds=1)
+        pipeline = PipelineBuilder().add(usage).build()
+        self.addCleanup(usage.close)
+        app.status = 500
+        self.process_unique(pipeline)
+        self.process_unique(pipeline)
+        self.wait_for(
+            lambda: len(app.attempts) == 1,
+            lambda: "The first send was not made.")
+        for i in range(6):
+            self.process_unique(pipeline)
+
+        app.status = 200
+        self.wait_for(
+            lambda: len(client_ips(app.received)) == 6,
+            lambda: f"Not all entries were sent: {client_ips(app.received)}")
+
+        sizes = [len(client_ips([data])) for data in app.received]
+        self.assertEqual(
+            [4, 2], sizes,
+            "The messages did not hold up to twice the package size.")
+
+    def test_close_sends_fewer_entries_than_the_package_size(self):
+        """!
+        Closing sends the entries still queued, even when there are fewer
+        than the requested package size.
+        """
+
+        usage = ShareUsage(
+            requested_package_size=5,
+            endpoint=testEndpoint)
+        pipeline = PipelineBuilder().add(usage).build()
+        ips = [self.process_unique(pipeline) for i in range(2)]
+        self.assertEqual([], app.received)
+
+        usage.close()
+
+        self.assertEqual(ips, client_ips(app.received))
+
+    def test_back_off_invalid_settings(self):
+        """!
+        Settings the pause cannot work with are rejected.
+        """
+
+        for settings in [
+                {"failures_to_enter_recovery": 0},
+                {"failures_to_enter_recovery": 101},
+                {"failures_to_enter_recovery": 2.5},
+                {"failures_window_seconds": 0},
+                {"initial_recovery_delay_seconds": 0},
+                {"max_recovery_delay_seconds": -1},
+                {"recovery_multiplier": 0.5},
+                {"max_queue_size": 0}]:
+            with self.assertRaises(ValueError, msg=str(settings)):
+                ShareUsage(
+                    requested_package_size=1,
+                    endpoint=testEndpoint,
+                    **settings)
+
     def test_data_received(self):
         """!
         Check that data is received from the share usage element.