feat(sqs): clear SNS FIFO dedup cache on PurgeQueue

Clear SNS FIFO topic deduplication for topics with SQS subscriptions targeting a purged queue when `clearFifoDeduplicationCacheOnPurge` is enabled.

Author: r-belakapytski

docs/configuration/application-yml.md

@@ -84,7 +84,7 @@ floci:
       enabled: true
       default-visibility-timeout: 30         # Seconds
       max-message-size: 262144               # Bytes (256 KB)
-      clear-fifo-deduplication-cache-on-purge: false  # When true, PurgeQueue also clears the FIFO deduplication cache
+      clear-fifo-deduplication-cache-on-purge: false  # When true, PurgeQueue clears SQS FIFO dedup and SNS FIFO topic dedup for topics subscribed to that queue
 
     s3:
       enabled: true
@@ -229,7 +229,7 @@ All keys in this table are declared on `EmulatorConfig` and accept environment v
 | `FLOCI_SERVICES_SSM_MAX_PARAMETER_HISTORY`         | `5`              | Max parameter versions kept                                   |
 | `FLOCI_SERVICES_SQS_DEFAULT_VISIBILITY_TIMEOUT`    | `30`             | Default visibility timeout (seconds)                          |
 | `FLOCI_SERVICES_SQS_MAX_MESSAGE_SIZE`              | `262144`         | Max message size (bytes)                                      |
-| `FLOCI_SERVICES_SQS_CLEAR_FIFO_DEDUPLICATION_CACHE_ON_PURGE` | `false` | When `true`, `PurgeQueue` also clears the FIFO 5-minute deduplication cache for the target queue |
+| `FLOCI_SERVICES_SQS_CLEAR_FIFO_DEDUPLICATION_CACHE_ON_PURGE` | `false` | When `true`, `PurgeQueue` clears the FIFO 5-minute deduplication cache for the target queue and matching SNS FIFO topic dedup entries |
 | `FLOCI_SERVICES_S3_DEFAULT_PRESIGN_EXPIRY_SECONDS` | `3600`           | Pre-signed URL expiry                                         |
 | `FLOCI_SERVICES_DOCKER_NETWORK`                    | *(unset)*        | Shared Docker network for Lambda, RDS, ElastiCache containers |
 | `FLOCI_SERVICES_ECS_MOCK`                          | `false`          | Skip Docker; tasks go straight to RUNNING (useful for CI)     |

docs/services/sqs.md

@@ -37,7 +37,7 @@ floci:
       enabled: true
       default-visibility-timeout: 30  # Seconds
       max-message-size: 262144        # 256 KB
-      clear-fifo-deduplication-cache-on-purge: false  # When true, PurgeQueue also clears the FIFO 5-minute deduplication cache
+      clear-fifo-deduplication-cache-on-purge: false  # When true, PurgeQueue clears the FIFO deduplication cache for the queue and for any SNS FIFO topics that subscribe to that queue (SNS in-memory dedup)
 ```
 
 ## Examples

src/main/java/io/github/hectorvent/floci/services/sns/SnsService.java

@@ -567,6 +567,54 @@ private boolean isDuplicate(String topicArn, String deduplicationId) {
         return false;
     }
 
+    /**
+     * Removes all FIFO deduplication cache entries for SNS topics that have an SQS subscription
+     * whose endpoint resolves to the same queue path as {@code queueUrl} (used when purging SQS
+     * with {@code clearFifoDeduplicationCacheOnPurge}).
+     */
+    public void clearFifoDeduplicationCacheForSqsQueueSubscriptions(String queueUrl, String region) {
+        String queuePath = extractQueuePathFromUrl(queueUrl);
+        if (queuePath.isEmpty()) {
+            return;
+        }
+        String subPrefix = "sub::" + region + "::";
+        subscriptionStore.keys().stream()
+                .filter(key -> key.startsWith(subPrefix))
+                .map(key -> subscriptionStore.get(key).orElse(null))
+                .filter(Objects::nonNull)
+                .filter(sub -> "sqs".equals(sub.getProtocol()))
+                .filter(sub -> sqsSubscriptionEndpointMatchesQueuePath(sub.getEndpoint(), queuePath))
+                .map(Subscription::getTopicArn)
+                .forEach(topicArn ->
+                        fifoDeduplicationCache.keySet().removeIf(cacheKey -> cacheKey.startsWith(topicArn + ":")));
+    }
+
+    private boolean sqsSubscriptionEndpointMatchesQueuePath(String endpoint, String queuePath) {
+        if (endpoint == null) {
+            return false;
+        }
+        String asUrl = sqsArnToUrl(endpoint);
+        return extractQueuePathFromUrl(asUrl).equals(queuePath);
+    }
+
+    /**
+     * Same path extraction as {@code SqsService} queue URL normalization ({@code /accountId/queueName}).
+     */
+    private static String extractQueuePathFromUrl(String url) {
+        if (url == null) {
+            return "";
+        }
+        int schemeEnd = url.indexOf("://");
+        if (schemeEnd < 0) {
+            return url;
+        }
+        int pathStart = url.indexOf('/', schemeEnd + 3);
+        if (pathStart < 0) {
+            return url;
+        }
+        return url.substring(pathStart);
+    }
+
     private List<Subscription> subscriptionsByTopic(String topicArn, String region) {
         List<Subscription> result = new ArrayList<>();
         String prefix = "sub::" + region + "::";
@@ -739,4 +787,4 @@ private static String topicKey(String region, String arn) {
     private static String subKey(String region, String subscriptionArn) {
         return "sub::" + region + "::" + subscriptionArn;
     }
-}
+}

src/main/java/io/github/hectorvent/floci/services/sqs/SqsService.java

@@ -5,6 +5,7 @@
 import io.github.hectorvent.floci.core.common.RegionResolver;
 import io.github.hectorvent.floci.core.storage.StorageBackend;
 import io.github.hectorvent.floci.core.storage.StorageFactory;
+import io.github.hectorvent.floci.services.sns.SnsService;
 import io.github.hectorvent.floci.services.sqs.model.Message;
 import io.github.hectorvent.floci.services.sqs.model.Queue;
 import com.fasterxml.jackson.core.type.TypeReference;
@@ -41,9 +42,11 @@ private record RedrivePolicy(int maxReceiveCount, String deadLetterTargetArn) {
     private final String baseUrl;
     private final RegionResolver regionResolver;
     private final boolean clearFifoDeduplicationCacheOnPurge;
+    private final SnsService snsService;
 
     @Inject
-    public SqsService(StorageFactory storageFactory, EmulatorConfig config, RegionResolver regionResolver) {
+    public SqsService(StorageFactory storageFactory, EmulatorConfig config, RegionResolver regionResolver,
+                      SnsService snsService) {
         this(
                 storageFactory.create("sqs", "sqs-queues.json",
                         new TypeReference<Map<String, Queue>>() {
@@ -58,7 +61,8 @@ public SqsService(StorageFactory storageFactory, EmulatorConfig config, RegionRe
                 config.services().sqs().maxMessageSize(),
                 config.effectiveBaseUrl(),
                 regionResolver,
-                config.services().sqs().clearFifoDeduplicationCacheOnPurge()
+                config.services().sqs().clearFifoDeduplicationCacheOnPurge(),
+                snsService
         );
     }
 
@@ -68,21 +72,22 @@ public SqsService(StorageFactory storageFactory, EmulatorConfig config, RegionRe
     SqsService(StorageBackend<String, Queue> queueStore,
                int defaultVisibilityTimeout, int maxMessageSize, String baseUrl) {
         this(queueStore, null, null, defaultVisibilityTimeout, maxMessageSize, baseUrl,
-                new RegionResolver("us-east-1", "000000000000"), false);
+                new RegionResolver("us-east-1", "000000000000"), false, null);
     }
 
     SqsService(StorageBackend<String, Queue> queueStore, StorageBackend<String, List<Message>> messageStore,
                StorageBackend<String, Map<String, Long>> dedupStore,
                int defaultVisibilityTimeout, int maxMessageSize, String baseUrl,
                RegionResolver regionResolver) {
         this(queueStore, messageStore, dedupStore, defaultVisibilityTimeout, maxMessageSize, baseUrl,
-                regionResolver, false);
+                regionResolver, false, null);
     }
 
     SqsService(StorageBackend<String, Queue> queueStore, StorageBackend<String, List<Message>> messageStore,
                StorageBackend<String, Map<String, Long>> dedupStore,
                int defaultVisibilityTimeout, int maxMessageSize, String baseUrl,
-               RegionResolver regionResolver, boolean clearFifoDeduplicationCacheOnPurge) {
+               RegionResolver regionResolver, boolean clearFifoDeduplicationCacheOnPurge,
+               SnsService snsService) {
         this.queueStore = queueStore;
         this.messageStore = messageStore;
         this.dedupStore = dedupStore;
@@ -91,6 +96,7 @@ public SqsService(StorageFactory storageFactory, EmulatorConfig config, RegionRe
         this.baseUrl = baseUrl;
         this.regionResolver = regionResolver;
         this.clearFifoDeduplicationCacheOnPurge = clearFifoDeduplicationCacheOnPurge;
+        this.snsService = snsService;
         loadPersistedMessages();
         loadPersistedDedup();
     }
@@ -618,6 +624,9 @@ public void purgeQueue(String queueUrl, String region) {
             if (dedupStore != null) {
                 dedupStore.delete(storageKey);
             }
+            if (snsService != null) {
+                snsService.clearFifoDeduplicationCacheForSqsQueueSubscriptions(queueUrl, region);
+            }
         }
         LOG.infov("Purged queue{0}: {1}",
                 clearFifoDeduplicationCacheOnPurge ? " (dedup cache cleared)" : "", queueUrl);

src/test/java/io/github/hectorvent/floci/services/sns/SnsSqsFanoutFifoDeliveryTest.java

@@ -164,4 +164,34 @@ void publish_duplicateMessage_isDeduplicatedAtTopicLevel() {
         assertEquals(1, messages.size());
         assertTrue(messages.get(0).getBody().contains("first"));
     }
+
+    @Test
+    void clearFifoDedupForSubscribedQueue_thenRepublishWithSameDedupId_deliversAgain() {
+        RegionResolver regionResolver = new RegionResolver(REGION, ACCOUNT);
+        SqsService purgeSqsService = SqsServiceFactory.createInMemoryWithFifoDedupPurgeAndSns(
+                BASE_URL, regionResolver, snsService);
+        sqsService.createQueue("manual-sns-dedup-queue.fifo", Map.of("FifoQueue", "true"), REGION);
+        purgeSqsService.createQueue("manual-sns-dedup-queue.fifo", Map.of("FifoQueue", "true"), REGION);
+        String queueArn = "arn:aws:sqs:" + REGION + ":" + ACCOUNT + ":manual-sns-dedup-queue.fifo";
+
+        snsService.createTopic("manual-sns-dedup-topic.fifo", Map.of("FifoTopic", "true"), null, REGION);
+        String topicArn = "arn:aws:sns:" + REGION + ":" + ACCOUNT + ":manual-sns-dedup-topic.fifo";
+        snsService.subscribe(topicArn, "sqs", queueArn, REGION, Map.of());
+
+        String queueUrl = BASE_URL + "/" + ACCOUNT + "/manual-sns-dedup-queue.fifo";
+
+        snsService.publish(topicArn, null, null, "before-clear", null, null, "group-1", "shared-dedup", REGION);
+        List<Message> first = sqsService.receiveMessage(queueUrl, 10, 30, 0, REGION);
+        assertEquals(1, first.size());
+        for (Message m : first) {
+            sqsService.deleteMessage(queueUrl, m.getReceiptHandle(), REGION);
+        }
+
+        purgeSqsService.purgeQueue(queueUrl, REGION);
+
+        snsService.publish(topicArn, null, null, "after-clear", null, null, "group-1", "shared-dedup", REGION);
+        List<Message> after = sqsService.receiveMessage(queueUrl, 10, 30, 0, REGION);
+        assertEquals(1, after.size());
+        assertTrue(after.getFirst().getBody().contains("after-clear"));
+    }
 }

src/test/java/io/github/hectorvent/floci/services/sqs/SqsServiceFactory.java

@@ -2,6 +2,7 @@
 
 import io.github.hectorvent.floci.core.common.RegionResolver;
 import io.github.hectorvent.floci.core.storage.InMemoryStorage;
+import io.github.hectorvent.floci.services.sns.SnsService;
 
 /**
  * Test helper to create SqsService instances
@@ -12,4 +13,10 @@ public static SqsService createInMemory(String baseUrl, RegionResolver regionRes
         return new SqsService(new InMemoryStorage<>(), new InMemoryStorage<>(), new InMemoryStorage<>(),
                 30, 262144, baseUrl, regionResolver);
     }
+
+    public static SqsService createInMemoryWithFifoDedupPurgeAndSns(String baseUrl, RegionResolver regionResolver,
+                                                                    SnsService snsService) {
+        return new SqsService(new InMemoryStorage<>(), new InMemoryStorage<>(), new InMemoryStorage<>(),
+                30, 262144, baseUrl, regionResolver, true, snsService);
+    }
 }

src/test/java/io/github/hectorvent/floci/services/sqs/SqsServiceTest.java

@@ -3,6 +3,7 @@
 import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.core.common.RegionResolver;
 import io.github.hectorvent.floci.core.storage.InMemoryStorage;
+import io.github.hectorvent.floci.services.sns.SnsService;
 import io.github.hectorvent.floci.services.sqs.model.Message;
 import io.github.hectorvent.floci.services.sqs.model.Queue;
 import org.junit.jupiter.api.BeforeEach;
@@ -12,6 +13,8 @@
 import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 
 class SqsServiceTest {
 
@@ -353,7 +356,7 @@ void fifoQueueIgnoresPerMessageDelaySeconds() {
     void purgeQueueClearsFifoDeduplicationCacheWhenEnabled() {
         final var service = new SqsService(
                 new InMemoryStorage<>(), new InMemoryStorage<>(), new InMemoryStorage<>(),
-                30, 262144, BASE_URL, new RegionResolver("us-east-1", "000000000000"), true);
+                30, 262144, BASE_URL, new RegionResolver("us-east-1", "000000000000"), true, null);
 
         final var queue = service.createQueue("dedup-clear.fifo", Map.of("ContentBasedDeduplication", "true"));
 
@@ -406,7 +409,7 @@ void purgeQueueClearsDedupStoreWhenEnabled() {
         final var dedupStore = new InMemoryStorage<String, Map<String, Long>>();
         final var service = new SqsService(
                 new InMemoryStorage<>(), new InMemoryStorage<>(), dedupStore,
-                30, 262144, BASE_URL, new RegionResolver("us-east-1", "000000000000"), true);
+                30, 262144, BASE_URL, new RegionResolver("us-east-1", "000000000000"), true, null);
 
         final var queue = service.createQueue("dedup-store-clear.fifo",
                 Map.of("ContentBasedDeduplication", "true"));
@@ -421,4 +424,16 @@ void purgeQueueClearsDedupStoreWhenEnabled() {
         assertTrue(dedupStore.keys().isEmpty(),
                 "Dedup store must be empty after purge with clearFifoDeduplicationCacheOnPurge=true");
     }
+
+    @Test
+    void purgeQueueWithClearFifoDelegatesToSnsForFifoDedupOnSubscribedTopics() {
+        final var sns = mock(SnsService.class);
+        final var service = new SqsService(
+                new InMemoryStorage<>(), new InMemoryStorage<>(), new InMemoryStorage<>(),
+                30, 262144, BASE_URL, new RegionResolver("us-east-1", "000000000000"), true, sns);
+        final var queue = service.createQueue("sns-dedup-delegate.fifo", Map.of("FifoQueue", "true"));
+        service.purgeQueue(queue.getQueueUrl());
+        verify(sns).clearFifoDeduplicationCacheForSqsQueueSubscriptions(
+                queue.getQueueUrl(), "us-east-1");
+    }
 }