Merge branch 'ian-UID2-6146-update-traffic-calculator-to-hardcoded-baseline' into ian-UID2-6151-add-traffic-filter-class

Author: Ian-Nara

src/main/java/com/uid2/optout/Const.java

@@ -24,6 +24,7 @@ public static class Config extends com.uid2.shared.Const.Config {
         public static final String OptOutSqsVisibilityTimeoutProp = "optout_sqs_visibility_timeout";
         public static final String OptOutDeltaJobTimeoutSecondsProp = "optout_delta_job_timeout_seconds";
         public static final String OptOutS3BucketDroppedRequestsProp = "optout_s3_bucket_dropped_requests";
+        public static final String OptOutMaxMessagesPerFileProp = "optout_max_messages_per_file";
         public static final String TrafficFilterConfigPathProp = "traffic_filter_config_path";
         public static final String TrafficCalcConfigPathProp = "traffic_calc_config_path";
         public static final String ManualOverrideS3PathProp = "manual_override_s3_path";

src/main/java/com/uid2/optout/vertx/DeltaProductionResult.java

@@ -6,12 +6,18 @@
 public class DeltaProductionResult {
     private final int deltasProduced;
     private final int entriesProcessed;
-    private final boolean stoppedDueToRecentMessages;
 
-    public DeltaProductionResult(int deltasProduced, int entriesProcessed, boolean stoppedDueToRecentMessages) {
+    /* 
+     * indicates that there are still messages in the queue, however,
+     * not enough time has elapsed to produce a delta file.
+     * We produce in batches of (5 minutes) 
+     */
+    private final boolean stoppedDueToMessagesTooRecent;
+
+    public DeltaProductionResult(int deltasProduced, int entriesProcessed, boolean stoppedDueToMessagesTooRecent) {
         this.deltasProduced = deltasProduced;
         this.entriesProcessed = entriesProcessed;
-        this.stoppedDueToRecentMessages = stoppedDueToRecentMessages;
+        this.stoppedDueToMessagesTooRecent = stoppedDueToMessagesTooRecent;
     }
 
     public int getDeltasProduced() {
@@ -22,8 +28,8 @@ public int getEntriesProcessed() {
         return entriesProcessed;
     }
 
-    public boolean stoppedDueToRecentMessages() {
-        return stoppedDueToRecentMessages;
+    public boolean stoppedDueToMessagesTooRecent() {
+        return stoppedDueToMessagesTooRecent;
     }
 }
 

src/main/java/com/uid2/optout/vertx/OptOutSqsLogProducer.java

@@ -85,6 +85,7 @@ public class OptOutSqsLogProducer extends AbstractVerticle {
     private final int visibilityTimeout;
     private final int deltaWindowSeconds; // Time window for each delta file (5 minutes = 300 seconds)
     private final int jobTimeoutSeconds;
+    private final int maxMessagesPerFile; // Memory protection: max messages per delta file
     private final int listenPort;
     private final String internalApiKey;
     private final InternalAuthMiddleware internalAuth;
@@ -138,6 +139,7 @@ public OptOutSqsLogProducer(JsonObject jsonConfig, ICloudStorage cloudStorage, O
         this.visibilityTimeout = jsonConfig.getInteger(Const.Config.OptOutSqsVisibilityTimeoutProp, 240); // 4 minutes default
         this.deltaWindowSeconds = 300; // Fixed 5 minutes for all deltas
         this.jobTimeoutSeconds = jsonConfig.getInteger(Const.Config.OptOutDeltaJobTimeoutSecondsProp, 10800); // 3 hours default
+        this.maxMessagesPerFile = jsonConfig.getInteger(Const.Config.OptOutMaxMessagesPerFileProp, 10000); // Memory protection limit
 
         // HTTP server configuration - use port offset + 1 to avoid conflicts
         this.listenPort = Const.Port.ServicePortForOptOut + Utils.getPortOffset() + 1;
@@ -149,11 +151,12 @@ public OptOutSqsLogProducer(JsonObject jsonConfig, ICloudStorage cloudStorage, O
         int bufferSize = jsonConfig.getInteger(Const.Config.OptOutProducerBufferSizeProp);
         this.buffer = ByteBuffer.allocate(bufferSize).order(ByteOrder.LITTLE_ENDIAN);
 
-        // Initialize window reader
+        // Initialize window reader with memory protection limit
         this.windowReader = new SqsWindowReader(
             this.sqsClient, this.queueUrl, this.maxMessagesPerPoll, 
-            this.visibilityTimeout, this.deltaWindowSeconds
+            this.visibilityTimeout, this.deltaWindowSeconds, this.maxMessagesPerFile
         );
+        LOGGER.info("OptOutSqsLogProducer initialized with maxMessagesPerFile: {}", this.maxMessagesPerFile);
     }
 
     @Override
@@ -344,7 +347,7 @@ private JsonObject produceDeltasBlocking() throws Exception {
         DeltaProductionResult deltaResult = this.produceBatchedDeltas();
 
         // Determine status based on results
-        if (deltaResult.getDeltasProduced() == 0 && deltaResult.stoppedDueToRecentMessages()) {
+        if (deltaResult.getDeltasProduced() == 0 && deltaResult.stoppedDueToMessagesTooRecent()) {
             // No deltas produced because all messages were too recent
             result.put("status", "skipped");
             result.put("reason", "All messages too recent");
@@ -365,30 +368,31 @@ private JsonObject produceDeltasBlocking() throws Exception {
     /**
      * Reads messages from SQS and produces delta files in 5 minute batches.
      * Continues until queue is empty or messages are too recent.
+     * Windows are limited to maxMessagesPerFile for memory protection.
      * 
      * @return DeltaProductionResult with counts and stop reason
      * @throws IOException if delta production fails
      */
     private DeltaProductionResult produceBatchedDeltas() throws IOException {
         int deltasProduced = 0;
         int totalEntriesProcessed = 0;
-        boolean stoppedDueToRecentMessages = false;
+        boolean stoppedDueToMessagesTooRecent = false;
 
         long jobStartTime = OptOutUtils.nowEpochSeconds();
-        LOGGER.info("Starting delta production from SQS queue");
+        LOGGER.info("Starting delta production from SQS queue (maxMessagesPerFile: {})", this.maxMessagesPerFile);
 
         // Read and process windows until done
         while (true) {
             if(checkJobTimeout(jobStartTime)){
                 break;
             }
 
-            // Read one complete 5-minute window
+            // Read one complete 5-minute window (limited to maxMessagesPerFile)
             SqsWindowReader.WindowReadResult windowResult = windowReader.readWindow();
 
             // If no messages, we're done (queue empty or messages too recent)
             if (windowResult.isEmpty()) {
-                stoppedDueToRecentMessages = windowResult.stoppedDueToRecentMessages();
+                stoppedDueToMessagesTooRecent = windowResult.stoppedDueToMessagesTooRecent();
                 LOGGER.info("Delta production complete - no more eligible messages");
                 break;
             }
@@ -422,7 +426,7 @@ private DeltaProductionResult produceBatchedDeltas() throws IOException {
         LOGGER.info("Delta production complete: took {}s, produced {} deltas, processed {} entries",
             totalDuration, deltasProduced, totalEntriesProcessed);
 
-        return new DeltaProductionResult(deltasProduced, totalEntriesProcessed, stoppedDueToRecentMessages);
+        return new DeltaProductionResult(deltasProduced, totalEntriesProcessed, stoppedDueToMessagesTooRecent);
     }
 
     /**

src/main/java/com/uid2/optout/vertx/OptOutTrafficCalculator.java

@@ -179,6 +179,20 @@ List<List<Long>> parseAllowlistRanges(JsonObject config) throws MalformedTraffic
 
             ranges.sort(Comparator.comparing(range -> range.get(0)));
 
+            // Validate no overlapping ranges
+            for (int i = 0; i < ranges.size() - 1; i++) {
+                long currentEnd = ranges.get(i).get(1);
+                long nextStart = ranges.get(i + 1).get(0);
+                if (currentEnd >= nextStart) {
+                    LOGGER.error("Overlapping allowlist ranges detected: [{}, {}] overlaps with [{}, {}]",
+                            ranges.get(i).get(0), currentEnd, nextStart, ranges.get(i + 1).get(1));
+                    throw new MalformedTrafficCalcConfigException(
+                            "Overlapping allowlist ranges detected at indices " + i + " and " + (i + 1));
+                }
+            }
+            
+        } catch (MalformedTrafficCalcConfigException e) {
+            throw e;
         } catch (Exception e) {
             LOGGER.error("Failed to parse allowlist ranges", e);
             throw new MalformedTrafficCalcConfigException("Failed to parse allowlist ranges: " + e.getMessage());
@@ -215,8 +229,10 @@ public TrafficStatus calculateStatus(List<Message> sqsMessages) {
             long oldestQueueTs = findOldestQueueTimestamp(sqsMessages);
             LOGGER.info("Traffic calculation: oldestQueueTs={}", oldestQueueTs);
 
-            // Define start time of the delta evaluation window [newestDeltaTs - 24h, newestDeltaTs]
-            long deltaWindowStart = newestDeltaTs - this.evaluationWindowSeconds - getAllowlistDuration(newestDeltaTs, newestDeltaTs - this.evaluationWindowSeconds);
+            // Define start time of the delta evaluation window
+            // We need evaluationWindowSeconds of non-allowlisted time, so we iteratively extend
+            // the window to account for any allowlist ranges in the extended portion
+            long deltaWindowStart = calculateWindowStartWithAllowlist(newestDeltaTs, this.evaluationWindowSeconds);
 
             // Evict old cache entries (older than delta window start)
             evictOldCacheEntries(deltaWindowStart);
@@ -391,6 +407,31 @@ long getAllowlistDuration(long t, long windowStart) {
         return totalDuration;
     }
 
+    /**
+     * Calculate the window start time that provides evaluationWindowSeconds of non-allowlisted time.
+     * Iteratively extends the window to account for allowlist ranges that may fall in extended portions.
+     */
+    long calculateWindowStartWithAllowlist(long newestDeltaTs, int evaluationWindowSeconds) {
+        long allowlistDuration = getAllowlistDuration(newestDeltaTs, newestDeltaTs - evaluationWindowSeconds);
+        
+        // Each iteration discovers at least one new allowlist range, so max iterations = number of ranges
+        int maxIterations = this.allowlistRanges.size() + 1;
+        
+        for (int i = 0; i < maxIterations && allowlistDuration > 0; i++) {
+            long newWindowStart = newestDeltaTs - evaluationWindowSeconds - allowlistDuration;
+            long newAllowlistDuration = getAllowlistDuration(newestDeltaTs, newWindowStart);
+            
+            if (newAllowlistDuration == allowlistDuration) {
+                // No new allowlist time in extended portion, we've converged
+                break;
+            }
+            
+            allowlistDuration = newAllowlistDuration;
+        }
+        
+        return newestDeltaTs - evaluationWindowSeconds - allowlistDuration;
+    }
+    
     /**
      * Find the oldest SQS queue message timestamp
      */

src/main/java/com/uid2/optout/vertx/SqsWindowReader.java

@@ -1,5 +1,7 @@
 package com.uid2.optout.vertx;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.sqs.SqsClient;
 import software.amazon.awssdk.services.sqs.model.Message;
 
@@ -9,23 +11,30 @@
 /**
  * Reads messages from SQS for complete 5-minute time windows.
  * Handles accumulation of all messages for a window before returning.
+ * Limits messages per window to prevent memory issues.
  */
 public class SqsWindowReader {
+    private static final Logger LOGGER = LoggerFactory.getLogger(SqsWindowReader.class);
+    
     private final SqsClient sqsClient;
     private final String queueUrl;
     private final int maxMessagesPerPoll;
     private final int visibilityTimeout;
     private final int deltaWindowSeconds;
+    private final int maxMessagesPerFile;
     private final SqsBatchProcessor batchProcessor;
-
+    
     public SqsWindowReader(SqsClient sqsClient, String queueUrl, int maxMessagesPerPoll, 
-                          int visibilityTimeout, int deltaWindowSeconds) {
+                          int visibilityTimeout, int deltaWindowSeconds, int maxMessagesPerFile) {
         this.sqsClient = sqsClient;
         this.queueUrl = queueUrl;
         this.maxMessagesPerPoll = maxMessagesPerPoll;
         this.visibilityTimeout = visibilityTimeout;
         this.deltaWindowSeconds = deltaWindowSeconds;
+        this.maxMessagesPerFile = maxMessagesPerFile;
         this.batchProcessor = new SqsBatchProcessor(sqsClient, queueUrl, deltaWindowSeconds);
+        LOGGER.info("SqsWindowReader initialized with: maxMessagesPerFile: {}, maxMessagesPerPoll: {}, visibilityTimeout: {}, deltaWindowSeconds: {}",
+                        maxMessagesPerFile, maxMessagesPerPoll, visibilityTimeout, deltaWindowSeconds);
     }
 
     /**
@@ -34,18 +43,19 @@ public SqsWindowReader(SqsClient sqsClient, String queueUrl, int maxMessagesPerP
     public static class WindowReadResult {
         private final List<SqsParsedMessage> messages;
         private final long windowStart;
-        private final boolean stoppedDueToRecentMessages;
+        private final boolean stoppedDueToMessagesTooRecent;
 
-        public WindowReadResult(List<SqsParsedMessage> messages, long windowStart, boolean stoppedDueToRecentMessages) {
+        public WindowReadResult(List<SqsParsedMessage> messages, long windowStart, 
+                                boolean stoppedDueToMessagesTooRecent) {
             this.messages = messages;
             this.windowStart = windowStart;
-            this.stoppedDueToRecentMessages = stoppedDueToRecentMessages;
+            this.stoppedDueToMessagesTooRecent = stoppedDueToMessagesTooRecent;
         }
 
         public List<SqsParsedMessage> getMessages() { return messages; }
         public long getWindowStart() { return windowStart; }
         public boolean isEmpty() { return messages.isEmpty(); }
-        public boolean stoppedDueToRecentMessages() { return stoppedDueToRecentMessages; }
+        public boolean stoppedDueToMessagesTooRecent() { return stoppedDueToMessagesTooRecent; }
     }
 
     /**
@@ -54,6 +64,7 @@ public WindowReadResult(List<SqsParsedMessage> messages, long windowStart, boole
      * - We discover the next window
      * - Queue is empty (no more messages)
      * - Messages are too recent (all messages younger than 5 minutes)
+     * - Message limit is reached (memory protection)
      * 
      * @return WindowReadResult with messages for the window, or empty if done
      */
@@ -62,6 +73,13 @@ public WindowReadResult readWindow() {
         long currentWindowStart = 0;
 
         while (true) {
+            // Check if we've hit the message limit
+            if (windowMessages.size() >= this.maxMessagesPerFile) {
+                LOGGER.warn("Window message limit reached ({} messages). Truncating window starting at {} for memory protection.",
+                    this.maxMessagesPerFile, currentWindowStart);
+                return new WindowReadResult(windowMessages, currentWindowStart, false);
+            }
+            
             // Read one batch from SQS (up to 10 messages)
             List<Message> rawBatch = SqsMessageOperations.receiveMessagesFromSqs(
                 this.sqsClient, this.queueUrl, this.maxMessagesPerPoll, this.visibilityTimeout);