From 123b8d407a5ae33f9b39fda8bf091c390b3450f5 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sat, 7 Nov 2026 06:50:12 +0000 Subject: [PATCH 1/4] Test that usage sharing pauses sending after failed sends Usage sharing in Python has no back-off. Every package is sent while the service is failing, and is lost. These tests expect the back-off .NET is being given, at its default settings of ten failed sends within 100 seconds and a first pause of two seconds: - after ten failed sends nothing is sent during the pause, and the evidence processed during it is all sent once the pause ends and the service answers, with no more requests processed; - closing while paused returns at once and sends nothing. The test receiver can now answer with another status, and records every request it gets as well as the data it accepts. The tests fail until the back-off is added. --- .../tests/test_usage.py | 128 +++++++++++++++++- 1 file changed, 125 insertions(+), 3 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py index 4f1bea4..60bc104 100644 --- a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py +++ b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py @@ -21,8 +21,10 @@ # ********************************************************************* 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 +61,44 @@ 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 @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.status + + +def client_ips(received): + """! + The client IPs in the usage data received. + """ + ips = [] + for data in received: + ips.extend(re.findall("([^<]*)", 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 + self.requests = 0 @classmethod def tearDownClass(cls): @@ -97,6 +118,107 @@ 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_data_received(self): """! Check that data is received from the share usage element. From 94d89a8cadb47c3a1a2bf96300391522cdd2a70c Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sat, 7 Nov 2026 06:53:34 +0000 Subject: [PATCH 2/4] Pause usage sharing sends after repeated failures Python now has the same back-off as .NET, with the same five settings and defaults, as constructor arguments. - After failures_to_enter_recovery sends fail within failures_window_seconds, sending pauses for initial_recovery_delay_seconds. A send fails when it raises or the answer is not 200. - When a pause ends one send is tried. If it fails the delay is multiplied by recovery_multiplier, up to max_recovery_delay_seconds. If it succeeds the back-off ends. - Processing is not affected. Evidence is queued during a pause, and new evidence is dropped only when the queue is full. The new max_queue_size argument sets the size, by default 1000 or ten times the requested package size if that is larger, as in .NET. The queue had no limit before, which a pause would have needed. - One thread sends at a time, as in .NET and Java. It sends until fewer than requested_package_size entries are queued. A timer that does not keep the process running starts sending when a pause ends, so the queue drains without more requests being processed. - Closing cancels the timer and sends nothing while paused, logging how many entries were not sent. ShareUsageBackOff holds the rules and takes the time from the caller, so its tests set the time. send_thread now returns whether the service accepted the message. --- .../share_usage.py | 217 ++++++++++++++++-- .../share_usage_backoff.py | 154 +++++++++++++ .../tests/test_share_usage_backoff.py | 122 ++++++++++ .../tests/test_usage.py | 75 ++++++ 4 files changed, 550 insertions(+), 18 deletions(-) create mode 100644 pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_backoff.py create mode 100644 pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_backoff.py 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..4064ef3 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,14 @@ # 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 + +# 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 +71,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 +119,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 +180,72 @@ def __init__( self.tracker = ShareUsageTracker(interval = interval) - # Initialise share data list + # The entries waiting to be sent, oldest first. self.share_data = [] + 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. + 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) + with self._lock: + queued = len(self.share_data) + if queued > 0 and self.backoff.remaining(time.monotonic()) > 0: + self._log( + "info", + f"{queued} usage sharing entries were not sent, as sending " + "was paused after failed sends when usage sharing closed.") def get_constant_xml(self): if self.constant_xml is None: @@ -224,22 +306,110 @@ 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. + """ - data = f'{"".join(share_data)}' + 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() - self.thread_pool.submit(self.send_thread, self.endpoint, data) + 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. + """ + + 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.requested_package_size] + del self.share_data[:self.requested_package_size] + if not share_data: + self._sending = False + return + + data = f'{"".join(share_data)}' + 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) @@ -247,6 +417,14 @@ def send_thread(self, endpoint, data): "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}.") + return False + return True def add_to_share_usage(self, data): """! @@ -299,13 +477,16 @@ def add_to_share_usage(self, data): xml += f"" xml += "" - - 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: + return + self.share_data.append(xml) # 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..219ea3a --- /dev/null +++ b/pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage_backoff.py @@ -0,0 +1,154 @@ +# ********************************************************************* +# 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 (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/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_usage.py b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py index 60bc104..a9cfbec 100644 --- a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py +++ b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py @@ -219,6 +219,81 @@ def test_back_off_close_while_paused(self): 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_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_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. From c152ba70f89330899719673968968927072cf40f Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:46:42 +0000 Subject: [PATCH 3/4] Test the full queue log, the failed answer log, message size, close and the repeat boundary --- .../tests/test_share_usage_tracker.py | 82 ++++++++++++ .../tests/test_usage.py | 120 +++++++++++++++++- 2 files changed, 201 insertions(+), 1 deletion(-) create mode 100644 pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_share_usage_tracker.py 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 a9cfbec..4ed90d2 100644 --- a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py +++ b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py @@ -20,6 +20,8 @@ # such notice(s) shall fulfill the requirements of that article. # ********************************************************************* +import contextlib +import io import platform import re import sys @@ -65,6 +67,8 @@ def get_received(self): 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(): @@ -72,7 +76,7 @@ def server(): data = gzip.decompress(request.data).decode("utf8") if app.status == 200: app.received.append(data) - return "", app.status + return app.body, app.status def client_ips(received): @@ -98,7 +102,9 @@ def setUp(self): app.received = [] app.attempts = [] app.status = 200 + app.body = "" self.requests = 0 + self.logs = [] @classmethod def tearDownClass(cls): @@ -275,6 +281,118 @@ def test_back_off_full_queue_drops_new_evidence(self): 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. From 6a09047c7dc0bd4cd560635bd4ed3f9736044272 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:47:35 +0000 Subject: [PATCH 4/4] Log a full queue and the content of a failed answer, send up to twice the package size, send what is queued on close and share a repeat at the interval --- .../share_usage.py | 82 +++++++++++++++---- .../share_usage_backoff.py | 3 +- .../share_usage_tracker.py | 2 +- .../tests/test_usage.py | 1 + 4 files changed, 72 insertions(+), 16 deletions(-) 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 4064ef3..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 @@ -54,6 +54,13 @@ # 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 @@ -183,6 +190,12 @@ def __init__( # 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: @@ -225,10 +238,10 @@ def __del__(self): 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. Closing - does not wait for a pause after failed sends, and sends nothing - while paused. + 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. """ @@ -239,13 +252,37 @@ def close(self): self._resume_timer.cancel() self._resume_timer = None self.thread_pool.shutdown(wait=True) - with self._lock: - queued = len(self.share_data) - if queued > 0 and self.backoff.remaining(time.monotonic()) > 0: + 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'{"".join(share_data)}' + 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 " - "was paused after failed sends when usage sharing closed.") + "had failed when usage sharing closed.") def get_constant_xml(self): if self.constant_xml is None: @@ -363,8 +400,8 @@ def _send_while_queued(self): def _send_messages(self): while True: with self._lock: - share_data = self.share_data[:self.requested_package_size] - del self.share_data[:self.requested_package_size] + share_data = self.share_data[:self._message_size] + del self.share_data[:self._message_size] if not share_data: self._sending = False return @@ -412,7 +449,6 @@ def send_thread(self, endpoint, 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}" @@ -422,7 +458,8 @@ def send_thread(self, endpoint, data): self._log( "warning", f"ShareUsage sent {len(data)} bytes of data and the service " - f"answered {response.status_code}.") + f"answered {response.status_code} with content " + f"'{response.text[:LOGGED_CONTENT_LENGTH]}'.") return False return True @@ -481,8 +518,25 @@ def add_to_share_usage(self, data): with self._lock: # When the queue is full, the newest entries are dropped. if len(self.share_data) >= self.max_queue_size: - return - self.share_data.append(xml) + 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 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 index 219ea3a..bfaec05 100644 --- 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 @@ -63,7 +63,8 @@ def __init__( max_recovery_delay_seconds = MAX_RECOVERY_DELAY_SECONDS_DEFAULT, recovery_multiplier = RECOVERY_MULTIPLIER_DEFAULT): - if not (FAILURES_TO_ENTER_RECOVERY_MIN <= failures_to_enter_recovery + 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 " 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_usage.py b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py index 4ed90d2..f2973bf 100644 --- a/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py +++ b/pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py @@ -401,6 +401,7 @@ def test_back_off_invalid_settings(self): 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},