Pause usage sharing sends after repeated failures

pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java

@@ -47,9 +47,12 @@
  *     {@link Constants#FIFTYONE_COOKIE_PREFIX} or has been configured to be
  *     shared using the constructor parameter.
  * 5) Not a header, cookie, session value or query string parameter.
+ * When all Evidence is shared, everything is included. When all query string
+ * parameters are shared, every query string parameter is included.
  * The prefix of a key is everything before its first '.', and the field
  * everything after it. An email address, the salt for hashed email
- * identifiers, the query string and the path are never included.
+ * identifiers, the query string, the path and the Cookie header are never
+ * included, whatever the configuration.
  */
 public class EvidenceKeyFilterShareUsage implements EvidenceKeyFilter {
 
@@ -65,6 +68,17 @@ public class EvidenceKeyFilterShareUsage implements EvidenceKeyFilter {
      */
     private final String sessionCookieName;
 
+    /**
+     * If true then every key is included, apart from those that are never
+     * shared.
+     */
+    private final boolean shareAllEvidence;
+
+    /**
+     * If true then every query string parameter is included.
+     */
+    private final boolean shareAllQueryStringParameters;
+
     /**
      * The content of HTTP headers in this array will not be included in the
      * request information sent to 51degrees. Any header names added here are
@@ -113,6 +127,40 @@ public EvidenceKeyFilterShareUsage(
         List<String> includedQueryStringParams,
         boolean includeSession,
         String sessionCookieName) {
+        this(
+            blockedHttpHeaders,
+            includedQueryStringParams,
+            includeSession,
+            sessionCookieName,
+            false,
+            false);
+    }
+
+    /**
+     * Constructor.
+     * @param blockedHttpHeaders a list of the names of the HTTP headers that
+     *                           share usage should not send to 51Degrees
+     * @param includedQueryStringParams a list of the names of query string
+     *                                  parameters that share usage should send
+     *                                  to 51Degrees
+     * @param includeSession if true then the session cookie will be included in
+     *                       the filter
+     * @param sessionCookieName the name of the cookie that contains the session
+     *                          id
+     * @param shareAllEvidence if true then all Evidence is included, apart from
+     *                         the keys that are never shared
+     * @param shareAllQueryStringParameters if true then all query string
+     *                                      parameters are included
+     */
+    public EvidenceKeyFilterShareUsage(
+        List<String> blockedHttpHeaders,
+        List<String> includedQueryStringParams,
+        boolean includeSession,
+        String sessionCookieName,
+        boolean shareAllEvidence,
+        boolean shareAllQueryStringParameters) {
+        this.shareAllEvidence = shareAllEvidence;
+        this.shareAllQueryStringParameters = shareAllQueryStringParameters;
         // The Cookie header carries every cookie, including the ones the
         // cookie rule leaves out.
         this.blockedHttpHeaders.add("cookies");
@@ -144,6 +192,10 @@ public boolean include(String key) {
             }
         }
 
+        if (shareAllEvidence) {
+            return true;
+        }
+
         boolean result;
         int separator = lowerKey.indexOf(EVIDENCE_SEPERATOR);
         if (separator > 0) {
@@ -169,8 +221,9 @@ public boolean include(String key) {
                 case EVIDENCE_QUERY_PREFIX:
                     // Only include query string parameters that start with the
                     // 51Degrees prefix or have been specified in the
-                    // constructor.
-                    result = field.startsWith(Constants.FIFTYONE_COOKIE_PREFIX) ||
+                    // constructor, unless all of them are shared.
+                    result = shareAllQueryStringParameters ||
+                        field.startsWith(Constants.FIFTYONE_COOKIE_PREFIX) ||
                         includedQueryStringParams.contains(field);
                     break;
                 default:

pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java

@@ -57,6 +57,38 @@ public EvidenceKeyFilterShareUsageTracker(
             sessionCookieName);
     }
 
+    /**
+     * Constructor.
+     * @param blockedHttpHeaders a list of the names of the HTTP headers that
+     *                           share usage should not send to 51Degrees
+     * @param includedQueryStringParams a list of the names of query string
+     *                                  parameters that share usage should send
+     *                                  to 51Degrees
+     * @param includeSession if true then the session cookie will be included in
+     *                       the filter
+     * @param sessionCookieName the name of the cookie that contains the session
+     *                          id
+     * @param shareAllEvidence if true then all Evidence is included, apart from
+     *                         the keys that are never shared
+     * @param shareAllQueryStringParameters if true then all query string
+     *                                      parameters are included
+     */
+    public EvidenceKeyFilterShareUsageTracker(
+        List<String> blockedHttpHeaders,
+        List<String> includedQueryStringParams,
+        boolean includeSession,
+        String sessionCookieName,
+        boolean shareAllEvidence,
+        boolean shareAllQueryStringParameters) {
+        super(
+            blockedHttpHeaders,
+            includedQueryStringParams,
+            includeSession,
+            sessionCookieName,
+            shareAllEvidence,
+            shareAllQueryStringParameters);
+    }
+
     @Override
     public boolean include(String key) {
         if(key.equalsIgnoreCase(EVIDENCE_SESSIONID) ||

pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java

@@ -93,6 +93,13 @@ public class Constants {
      * Default lost data count for share usage before it wraps
      */
     static final int LOST_DATA_RESET_DEFAULT = 5000000;
+
+    /**
+     * 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.
+     */
+    static final int QUEUE_FULL_LOG_EVERY = 1000;
     /**
      * The default URL to send usage data to.
      */
@@ -104,4 +111,43 @@ public class Constants {
      */
     static final String EVIDENCE_HTTPHEADER_COOKIE_SUFFIX = "cookie";
 
+    /**
+     * The default number of failed sends within the failures window that
+     * pauses sending.
+     */
+    public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT = 10;
+
+    /**
+     * The fewest failed sends that can be set to pause sending.
+     */
+    public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN = 1;
+
+    /**
+     * The most failed sends that can be set to pause sending.
+     */
+    public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX = 100;
+
+    /**
+     * The default time in seconds over which failed sends are counted.
+     */
+    public static final int SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT = 100;
+
+    /**
+     * The default length in seconds of the first pause after failed sends.
+     */
+    public static final double
+        SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT = 2.0;
+
+    /**
+     * The default longest pause in seconds.
+     */
+    public static final double
+        SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT = 300.0;
+
+    /**
+     * The default multiplier applied to the pause each time the send tried
+     * when a pause ends fails.
+     */
+    public static final double
+        SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT = 2.0;
 }

pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java

@@ -0,0 +1,165 @@
+/* *********************************************************************
+ * 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.
+ * ********************************************************************* */
+
+package fiftyone.pipeline.engines.fiftyone.flowelements;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/**
+ * 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 milliseconds on
+ * the clock that {@link #now()} reads, so a change to the system time cannot
+ * lengthen or skip a pause.
+ */
+class ShareUsageBackOff {
+
+    private final int failuresToEnterRecovery;
+    private final long failuresWindowMillis;
+    private final double initialDelaySeconds;
+    private final double maxDelaySeconds;
+    private final double multiplier;
+
+    /**
+     * When each recent failed send happened, oldest first. Empty while
+     * backing off.
+     */
+    private final Deque<Long> failures = new ArrayDeque<>();
+
+    /**
+     * The length of the current pause in seconds, or zero when not backing
+     * off.
+     */
+    private double delaySeconds = 0;
+
+    /**
+     * When sending may start again.
+     */
+    private long resumeAt = Long.MIN_VALUE;
+
+    /**
+     * @return the time in milliseconds on a monotonic clock, for the caller to
+     * pass to the other methods. Its zero is not a calendar time.
+     */
+    static long now() {
+        return System.nanoTime() / 1000000;
+    }
+
+    ShareUsageBackOff(
+        int failuresToEnterRecovery,
+        int failuresWindowSeconds,
+        double initialDelaySeconds,
+        double maxDelaySeconds,
+        double multiplier) {
+        if (failuresToEnterRecovery <
+                Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN ||
+            failuresToEnterRecovery >
+                Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX) {
+            throw new IllegalArgumentException(
+                "failuresToEnterRecovery must be from " +
+                    Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN +
+                    " to " +
+                    Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX + ".");
+        }
+        if (failuresWindowSeconds < 1) {
+            throw new IllegalArgumentException(
+                "failuresWindowSeconds must be 1 or more.");
+        }
+        if ((initialDelaySeconds > 0) == false) {
+            throw new IllegalArgumentException(
+                "initialDelaySeconds must be more than 0.");
+        }
+        if ((maxDelaySeconds > 0) == false) {
+            throw new IllegalArgumentException(
+                "maxDelaySeconds must be more than 0.");
+        }
+        if ((multiplier >= 1) == false) {
+            throw new IllegalArgumentException(
+                "multiplier must be 1 or more.");
+        }
+        this.failuresToEnterRecovery = failuresToEnterRecovery;
+        this.failuresWindowMillis = failuresWindowSeconds * 1000L;
+        this.initialDelaySeconds = initialDelaySeconds;
+        this.maxDelaySeconds = maxDelaySeconds;
+        this.multiplier = multiplier;
+    }
+
+    /**
+     * @return true from the start of a pause until a send succeeds
+     */
+    synchronized boolean isBackingOff() {
+        return delaySeconds > 0;
+    }
+
+    /**
+     * @param now the time now in milliseconds
+     * @return the milliseconds left before sending may start again, or zero
+     * if it may start now
+     */
+    synchronized long remaining(long now) {
+        return resumeAt > now ? resumeAt - now : 0;
+    }
+
+    /**
+     * Records a failed send.
+     * @param now the time now in milliseconds
+     * @return the new pause in milliseconds if this failure started or
+     * lengthened one, otherwise zero
+     */
+    synchronized long failed(long now) {
+        if (delaySeconds > 0) {
+            // The send tried when the last pause ended failed.
+            delaySeconds = Math.min(
+                delaySeconds * multiplier,
+                maxDelaySeconds);
+        } else {
+            failures.addLast(now);
+            while (failures.peekFirst() < now - failuresWindowMillis) {
+                failures.removeFirst();
+            }
+            if (failures.size() < failuresToEnterRecovery) {
+                return 0;
+            }
+            failures.clear();
+            delaySeconds = Math.min(initialDelaySeconds, maxDelaySeconds);
+        }
+        long pause = Math.max(1, Math.round(delaySeconds * 1000));
+        resumeAt = now + pause;
+        return pause;
+    }
+
+    /**
+     * Records a successful send.
+     * @return true if this ended a back-off
+     */
+    synchronized boolean succeeded() {
+        if (delaySeconds == 0) {
+            return false;
+        }
+        delaySeconds = 0;
+        resumeAt = Long.MIN_VALUE;
+        return true;
+    }
+}