Fix clearAddress leaves primary-key deduplication state (#7038)

Co-authored-by: Tim Smart <hello@timsmart.co>

Author: fubhy

.changeset/fix-message-storage-clear-address-dedup.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Clear in-memory message primary-key indexes when clearing an entity address.

packages/effect/src/unstable/cluster/MessageStorage.ts

@@ -991,6 +991,14 @@ export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluste
       resetAddress: () => Effect.void,
       clearAddress: (address) =>
         Effect.sync(() => {
+          for (const [primaryKey, entry] of requestsByPrimaryKey) {
+            const envelope = entry.envelope
+            const sameAddress = address.entityType === envelope.address.entityType &&
+              address.entityId === envelope.address.entityId
+            if (sameAddress) {
+              requestsByPrimaryKey.delete(primaryKey)
+            }
+          }
           for (let i = journal.length - 1; i >= 0; i--) {
             const envelope = journal[i]
             const sameAddress = address.entityType === envelope.address.entityType &&

packages/effect/test/cluster/MessageStorage.test.ts

@@ -23,6 +23,32 @@ const MemoryLive = MessageStorage.layerMemory.pipe(
 
 describe("MessageStorage", () => {
   describe("memory", () => {
+    it.effect("removes the primary-key index when clearing an address", () =>
+      Effect.gen(function*() {
+        const driver = yield* MessageStorage.MemoryDriver
+        const address = EntityAddress.make({
+          shardId: ShardId.make("default", 1),
+          entityType: EntityType.make("Repro"),
+          entityId: EntityId.make("one")
+        })
+        const envelope: Envelope.PartialRequestEncoded = {
+          _tag: "Request",
+          requestId: "1",
+          address: { shardId: { group: "default", id: 1 }, entityType: "Repro", entityId: "one" },
+          tag: "Repro",
+          payload: {},
+          headers: {}
+        }
+        yield* driver.encoded.saveEnvelope({ envelope, primaryKey: "dedup-key", deliverAt: null })
+        yield* driver.encoded.clearAddress(address)
+        const result = yield* driver.encoded.saveEnvelope({
+          envelope: { ...envelope, requestId: "2" },
+          primaryKey: "dedup-key",
+          deliverAt: null
+        })
+        expect(result._tag).toEqual("Success")
+      }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer)))
+
     it.effect("saves a request", () =>
       Effect.gen(function*() {
         const storage = yield* MessageStorage.MessageStorage