circuit breaker

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.uid2</groupId>
     <artifactId>uid2-optout</artifactId>
-    <version>4.5.0</version>
+    <version>4.5.8-alpha-122-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/Main.java

@@ -1,6 +1,8 @@
 package com.uid2.optout;
 
 import com.uid2.optout.vertx.*;
+import com.uid2.optout.vertx.OptOutTrafficFilter.MalformedTrafficFilterConfigException;
+import com.uid2.optout.vertx.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,23 @@ 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("Failed to initialize SQS log producer, delta production will be disabled: " + e.getMessage(), e);
+            } catch (MalformedTrafficFilterConfigException e) {
+                LOGGER.error("The traffic filter config is malformed, refusing to process messages, delta production will be disabled: " + e.getMessage(), e);
+            } catch (MalformedTrafficCalcConfigException e) {
+                LOGGER.error("The traffic calc config is malformed, refusing to process messages, delta production will be disabled: " + e.getMessage(), e);
             }
         }
 

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

@@ -1,23 +1,32 @@
 package com.uid2.optout.vertx;
 
+import io.vertx.core.json.JsonObject;
+
 /**
- * Result object containing statistics from delta production.
+ * Data class containing statistics from delta production.
+ * 
+ * This class holds the counts and provides JSON encoding methods.
+ * API response status is determined by the caller based on these statistics.
  */
 public class DeltaProductionResult {
     private final int deltasProduced;
     private final int entriesProcessed;
-
+    private final int droppedRequestFilesProduced;
+    private final int droppedRequestsProcessed;
+
     /* 
      * 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;
+    private final boolean stoppedDueToRecentMessages;
 
-    public DeltaProductionResult(int deltasProduced, int entriesProcessed, boolean stoppedDueToMessagesTooRecent) {
+    public DeltaProductionResult(int deltasProduced, int entriesProcessed, int droppedRequestFilesProduced, int droppedRequestsProcessed, boolean stoppedDueToRecentMessages) {
         this.deltasProduced = deltasProduced;
         this.entriesProcessed = entriesProcessed;
-        this.stoppedDueToMessagesTooRecent = stoppedDueToMessagesTooRecent;
+        this.droppedRequestFilesProduced = droppedRequestFilesProduced;
+        this.droppedRequestsProcessed = droppedRequestsProcessed;
+        this.stoppedDueToRecentMessages = stoppedDueToRecentMessages;
     }
 
     public int getDeltasProduced() {
@@ -28,8 +37,40 @@ public int getEntriesProcessed() {
         return entriesProcessed;
     }
 
-    public boolean stoppedDueToMessagesTooRecent() {
-        return stoppedDueToMessagesTooRecent;
+    public boolean stoppedDueToRecentMessages() {
+        return stoppedDueToRecentMessages;
+    }
+
+    public int getDroppedRequestFilesProduced() {
+        return droppedRequestFilesProduced;
+    }
+
+    public int getDroppedRequestsProcessed() {
+        return droppedRequestsProcessed;
+    }
+
+    /**
+     * Convert to JSON with just the production counts.
+     */
+    public JsonObject toJson() {
+        return new JsonObject()
+            .put("deltas_produced", deltasProduced)
+            .put("entries_processed", entriesProcessed)
+            .put("dropped_request_files_produced", droppedRequestFilesProduced)
+            .put("dropped_requests_processed", droppedRequestsProcessed);
+    }
+
+    /**
+     * Convert to JSON with status and counts.
+     */
+    public JsonObject toJsonWithStatus(String status) {
+        return toJson().put("status", status);
     }
-}
 
+    /**
+     * Convert to JSON with status, reason/error, and counts.
+     */
+    public JsonObject toJsonWithStatus(String status, String reasonKey, String reasonValue) {
+        return toJsonWithStatus(status).put(reasonKey, reasonValue);
+    }
+}

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

@@ -365,21 +365,23 @@ private void handleQueue(RoutingContext routingContext) {
         String email = body != null ? body.getString(EMAIL) : null;
         String phone = body != null ? body.getString(PHONE) : null;
 
-        HttpServerResponse resp = routingContext.response();
-
         // while old delta production is enabled, response is handled by replicate logic
 
         // Validate parameters - same as replicate
         if (identityHash == null || params.getAll(IDENTITY_HASH).size() != 1) {
+            LOGGER.warn("handleQueue: Invalid identity_hash parameter");
             // this.sendBadRequestError(resp);
             return;
         }
         if (advertisingId == null || params.getAll(ADVERTISING_ID).size() != 1) {
+            LOGGER.warn("handleQueue: Invalid advertising_id parameter");
+
             // this.sendBadRequestError(resp);
             return;
         }
 
         if (!this.isGetOrPost(req)) {
+            LOGGER.warn("handleQueue: Invalid HTTP method: {}", req.method());
             // this.sendBadRequestError(resp);
             return;
         }
@@ -408,8 +410,6 @@ private void handleQueue(RoutingContext routingContext) {
                 }
             }, res -> {
                 if (res.failed()) {
-                    // this.sendInternalServerError(resp, "Failed to queue message: " + res.cause().getMessage());
-                    LOGGER.error("Failed to queue message: " + res.cause().getMessage());
                 } else {
                     String messageId = (String) res.result();
 
@@ -426,8 +426,7 @@ private void handleQueue(RoutingContext routingContext) {
                 }
             });
         } catch (Exception ex) {
-            // this.sendInternalServerError(resp, ex.getMessage());
-            LOGGER.error("Error processing queue request: " + ex.getMessage(), ex);
+            LOGGER.error("handleQueue: Error processing queue request");
         }
     }