circuit breaker refactored

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.uid2</groupId>
     <artifactId>uid2-optout</artifactId>
-    <version>4.5.0</version>
+    <version>4.5.13-alpha-128-SNAPSHOT</version>
     <name>uid2-optout</name>
     <url>https://github.com/IABTechLab/uid2-optout</url>
 
@@ -160,6 +160,11 @@
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>sqs</artifactId>
         </dependency>
+        <dependency>
+            <groupId>commons-logging</groupId>
+            <artifactId>commons-logging</artifactId>
+            <version>1.2</version>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -32,6 +32,7 @@ public static class Config extends com.uid2.shared.Const.Config {
         public static final String OptOutTrafficCalcThresholdMultiplierProp = "traffic_calc_threshold_multiplier";
         public static final String OptOutTrafficCalcEvaluationWindowSecondsProp = "traffic_calc_evaluation_window_seconds";
         public static final String OptOutTrafficCalcAllowlistRangesProp = "traffic_calc_allowlist_ranges";
+        public static final String OptOutSqsDeltaWindowSecondsProp = "optout_sqs_delta_window_seconds";
     }
 
     public static class Event {

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

@@ -1,6 +1,8 @@
 package com.uid2.optout;
 
 import com.uid2.optout.vertx.*;
+import com.uid2.optout.traffic.OptOutTrafficFilter.MalformedTrafficFilterConfigException;
+import com.uid2.optout.traffic.OptOutTrafficCalculator.MalformedTrafficCalcConfigException;
 import com.uid2.shared.ApplicationVersion;
 import com.uid2.shared.Utils;
 import com.uid2.shared.attest.AttestationResponseHandler;
@@ -27,7 +29,6 @@
 import io.vertx.config.ConfigRetriever;
 import io.vertx.core.*;
 import io.vertx.core.http.HttpServerOptions;
-import io.vertx.core.http.impl.HttpUtils;
 import io.vertx.core.json.JsonObject;
 import io.vertx.micrometer.MetricsDomain;
 import org.slf4j.Logger;
@@ -296,14 +297,22 @@ public void run(String[] args) throws IOException {
                     fsSqs = CloudUtils.createStorage(optoutBucket, sqsConfig);
                 }
 
-                // Deploy SQS log producer with its own storage instance
-                OptOutSqsLogProducer sqsLogProducer = new OptOutSqsLogProducer(this.config, fsSqs, sqsCs);
+                // Create SQS-specific cloud storage instance for dropped requests (different bucket)
+                String optoutBucketDroppedRequests = this.config.getString(Const.Config.OptOutS3BucketDroppedRequestsProp);
+                ICloudStorage fsSqsDroppedRequests = CloudUtils.createStorage(optoutBucketDroppedRequests, config);
+
+                // Deploy SQS log producer with its own storage instance    
+                OptOutSqsLogProducer sqsLogProducer = new OptOutSqsLogProducer(this.config, fsSqs, fsSqsDroppedRequests, sqsCs, Const.Event.DeltaProduce, null);
                 futs.add(this.deploySingleInstance(sqsLogProducer));
 
                 LOGGER.info("SQS log producer deployed - bucket: {}, folder: {}", 
                     this.config.getString(Const.Config.OptOutS3BucketProp), sqsFolder);
             } catch (IOException e) {
-                LOGGER.error("Failed to initialize SQS log producer: " + e.getMessage(), e);
+                LOGGER.error("circuit_breaker_config_error: failed to initialize SQS log producer, delta production will be disabled: {}", e.getMessage(), e);
+            } catch (MalformedTrafficFilterConfigException e) {
+                LOGGER.error("circuit_breaker_config_error: traffic filter config is malformed, delta production will be disabled: {}", e.getMessage(), e);
+            } catch (MalformedTrafficCalcConfigException e) {
+                LOGGER.error("circuit_breaker_config_error: traffic calc config is malformed, delta production will be disabled: {}", e.getMessage(), e);
             }
         }
 

src/main/java/com/uid2/optout/delta/DeltaFileWriter.java

@@ -0,0 +1,113 @@
+package com.uid2.optout.delta;
+
+import com.uid2.shared.optout.OptOutConst;
+import com.uid2.shared.optout.OptOutEntry;
+import com.uid2.shared.optout.OptOutUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+/**
+ * Handles binary writing of delta file entries.
+ * 
+ * Delta files have the following format:
+ * - Start entry: null hash (32 bytes) + null hash (32 bytes) + timestamp (8 bytes)
+ * - Opt-out entries: hash (32 bytes) + id (32 bytes) + timestamp (7 bytes) + metadata (1 byte)
+ * - End entry: ones hash (32 bytes) + ones hash (32 bytes) + timestamp (8 bytes)
+ * 
+ * Each entry is 72 bytes (OptOutConst.EntrySize)
+ */
+public class DeltaFileWriter {
+    private static final Logger LOGGER = LoggerFactory.getLogger(DeltaFileWriter.class);
+
+    private ByteBuffer buffer;
+
+    /**
+     * Create a DeltaFileWriter with the specified initial buffer size.
+     * 
+     * @param bufferSize Initial buffer size in bytes
+     */
+    public DeltaFileWriter(int bufferSize) {
+        this.buffer = ByteBuffer.allocate(bufferSize).order(ByteOrder.LITTLE_ENDIAN);
+    }
+
+    /**
+     * Write the start-of-delta sentinel entry.
+     * Uses null hash bytes and the window start timestamp.
+     * 
+     * @param stream Output stream to write to
+     * @param windowStart Window start timestamp (epoch seconds)
+     * @throws IOException if write fails
+     */
+    public void writeStartOfDelta(ByteArrayOutputStream stream, long windowStart) throws IOException {
+        ensureCapacity(OptOutConst.EntrySize);
+
+        buffer.put(OptOutUtils.nullHashBytes);
+        buffer.put(OptOutUtils.nullHashBytes);
+        buffer.putLong(windowStart);
+
+        flushToStream(stream);
+    }
+
+    /**
+     * Write a single opt-out entry.
+     * 
+     * @param stream Output stream to write to
+     * @param hashBytes Hash bytes (32 bytes)
+     * @param idBytes ID bytes (32 bytes)
+     * @param timestamp Entry timestamp (epoch seconds)
+     * @throws IOException if write fails
+     */
+    public void writeOptOutEntry(ByteArrayOutputStream stream, byte[] hashBytes, byte[] idBytes, long timestamp) throws IOException {
+        ensureCapacity(OptOutConst.EntrySize);
+
+        OptOutEntry.writeTo(buffer, hashBytes, idBytes, timestamp);
+
+        flushToStream(stream);
+    }
+
+    /**
+     * Write the end-of-delta sentinel entry.
+     * Uses ones hash bytes and the window end timestamp.
+     * 
+     * @param stream Output stream to write to
+     * @param windowEnd Window end timestamp (epoch seconds)
+     * @throws IOException if write fails
+     */
+    public void writeEndOfDelta(ByteArrayOutputStream stream, long windowEnd) throws IOException {
+        ensureCapacity(OptOutConst.EntrySize);
+
+        buffer.put(OptOutUtils.onesHashBytes);
+        buffer.put(OptOutUtils.onesHashBytes);
+        buffer.putLong(windowEnd);
+
+        flushToStream(stream);
+    }
+
+    /**
+     * Flush the buffer contents to the output stream and clear the buffer.
+     */
+    private void flushToStream(ByteArrayOutputStream stream) throws IOException {
+        buffer.flip();
+        byte[] entry = new byte[buffer.remaining()];
+        buffer.get(entry);
+        stream.write(entry);
+        buffer.clear();
+    }
+
+    /**
+     * Ensure buffer has sufficient capacity, expanding if necessary.
+     */
+    private void ensureCapacity(int dataSize) {
+        if (buffer.capacity() < dataSize) {
+            int newCapacity = Integer.highestOneBit(dataSize) << 1;
+            LOGGER.info("expanding buffer size: current {}, need {}, new {}", buffer.capacity(), dataSize, newCapacity);
+            this.buffer = ByteBuffer.allocate(newCapacity).order(ByteOrder.LITTLE_ENDIAN);
+        }
+    }
+}
+

src/main/java/com/uid2/optout/vertx/DeltaProduceJobStatus.java → src/main/java/com/uid2/optout/delta/DeltaProductionJobStatus.java

@@ -1,4 +1,4 @@
-package com.uid2.optout.vertx;
+package com.uid2.optout.delta;
 
 import io.vertx.core.json.JsonObject;
 import java.time.Instant;
@@ -10,7 +10,7 @@
  * (running, completed, failed), timing information, and result or error details.
  * 
  */
-public class DeltaProduceJobStatus {
+public class DeltaProductionJobStatus {
     private final Instant startTime;
     private volatile JobState state;
     private volatile JsonObject result;
@@ -23,7 +23,7 @@ public enum JobState {
         FAILED
     }
 
-    public DeltaProduceJobStatus() {
+    public DeltaProductionJobStatus() {
         this.startTime = Instant.now();
         this.state = JobState.RUNNING;
     }

src/main/java/com/uid2/optout/delta/DeltaProductionMetrics.java

@@ -0,0 +1,60 @@
+package com.uid2.optout.delta;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Metrics;
+
+/**
+ * Metrics counters for delta production operations.
+ * 
+ * Tracks:
+ * - Number of delta files produced
+ * - Number of opt-out entries processed
+ * - Number of dropped request files produced
+ * - Number of dropped requests processed
+ */
+public class DeltaProductionMetrics {
+
+    private final Counter deltasProduced;
+    private final Counter entriesProcessed;
+    private final Counter droppedRequestFilesProduced;
+    private final Counter droppedRequestsProcessed;
+
+    public DeltaProductionMetrics() {
+        this.deltasProduced = Counter
+            .builder("uid2_optout_sqs_delta_produced_total")
+            .description("counter for how many optout delta files are produced from SQS")
+            .register(Metrics.globalRegistry);
+
+        this.entriesProcessed = Counter
+            .builder("uid2_optout_sqs_entries_processed_total")
+            .description("counter for how many optout entries are processed from SQS")
+            .register(Metrics.globalRegistry);
+
+        this.droppedRequestFilesProduced = Counter
+            .builder("uid2_optout_sqs_dropped_request_files_produced_total")
+            .description("counter for how many optout dropped request files are produced from SQS")
+            .register(Metrics.globalRegistry);
+
+        this.droppedRequestsProcessed = Counter
+            .builder("uid2_optout_sqs_dropped_requests_processed_total")
+            .description("counter for how many optout dropped requests are processed from SQS")
+            .register(Metrics.globalRegistry);
+    }
+
+    /**
+     * Record that a delta file was produced with the given number of entries.
+     */
+    public void recordDeltaProduced(int entryCount) {
+        deltasProduced.increment();
+        entriesProcessed.increment(entryCount);
+    }
+
+    /**
+     * Record that a dropped requests file was produced with the given number of entries.
+     */
+    public void recordDroppedRequestsProduced(int requestCount) {
+        droppedRequestFilesProduced.increment();
+        droppedRequestsProcessed.increment(requestCount);
+    }
+}
+