feat: add debug logger for duplicate downloads from the same relay

Adds RelayDuplicateDownloadLogger, a debug-only relay listener (wired in
AppModules alongside RelaySpeedLogger/RelayReqStats) that measures how well
our filter design avoids pulling the same event from the same relay twice.

For every (relay, eventId) it tracks which subscriptions delivered it. The
first arrival is expected; any later arrival is wasted bandwidth and is
classified as:
 - REDELIVERY: the same subscription re-REQ'd the event (since/EOSE tuning), or
 - OVERLAP: two assemblers' filters overlap on the same relay (filter merging).

Each duplicate logs an immediate warning with the subscription ids involved
(cross-referenceable to the REQ filters printed by RelayLogger), the kind,
author, and bytes wasted. A rolling DuplicateDownloadStats digest aggregates
waste by offending interaction, relay, and kind to point at the filter or
assembler that should be refined.

amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt

@@ -69,6 +69,7 @@ import com.vitorpamplona.amethyst.service.playback.service.PlaybackServiceClient
 import com.vitorpamplona.amethyst.service.relayClient.CacheClientConnector
 import com.vitorpamplona.amethyst.service.relayClient.RelayProxyClientConnector
 import com.vitorpamplona.amethyst.service.relayClient.authCommand.model.AuthCoordinator
+import com.vitorpamplona.amethyst.service.relayClient.dupLogger.RelayDuplicateDownloadLogger
 import com.vitorpamplona.amethyst.service.relayClient.notifyCommand.model.NotifyCoordinator
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.RelaySubscriptionsCoordinator
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.EventFinderQueryState
@@ -511,6 +512,10 @@ class AppModules(
     val relayReqStats = if (isDebug) RelayReqStats(client) else null
     val logger = if (isDebug) RelaySpeedLogger(client) else null
 
+    // Detects events downloaded more than once from the same relay so we can
+    // refine which filter/assembler is requesting redundant data.
+    val duplicateDownloadLogger = if (isDebug) RelayDuplicateDownloadLogger(client) else null
+
     // Coordinates all subscriptions for the Nostr Client
     val sources: RelaySubscriptionsCoordinator =
         RelaySubscriptionsCoordinator(

amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/dupLogger/DownloadRecord.kt

@@ -0,0 +1,131 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.service.relayClient.dupLogger
+
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.utils.cache.LargeCache
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
+
+/**
+ * Tracks every time a single (relay, eventId) pair has been downloaded and which
+ * subscriptions delivered it. The first download is expected; every download after
+ * that is wasted bandwidth — the same relay sent us the same event again.
+ *
+ * Two kinds of waste are distinguished by looking at the subscriptions involved:
+ *  - [Category.REDELIVERY]: the SAME subscription id received the event more than
+ *    once from the relay. Usually a re-REQ that didn't carry a tight enough `since`,
+ *    or an EOSE/limit handling problem in that assembler.
+ *  - [Category.OVERLAP]: TWO OR MORE subscription ids each pulled the event from the
+ *    same relay. The filters of those assemblers overlap on this relay and could be
+ *    merged or narrowed.
+ *
+ * Records are accumulated for the whole debug session, so this is debug-only.
+ */
+@OptIn(ExperimentalAtomicApi::class)
+class DownloadRecord(
+    val relayUrl: NormalizedRelayUrl,
+    val eventId: HexKey,
+    val kind: Int,
+    val author: HexKey,
+) {
+    enum class Category {
+        REDELIVERY,
+        OVERLAP,
+    }
+
+    // total number of times this event arrived from this relay (1 = no waste yet)
+    private val downloads = AtomicInteger(0)
+
+    // total bytes of all the duplicate (2nd+) deliveries
+    private val wastedBytes = AtomicInteger(0)
+
+    // how many times each subscription id delivered this event from this relay
+    private val perSubId = LargeCache<String, AtomicInteger>()
+
+    /**
+     * Registers one delivery of this event from [relayUrl] under [subId].
+     *
+     * @return null if this is the first (expected) download, or a [Duplicate]
+     *  describing the waste when this is the 2nd+ download.
+     */
+    fun register(
+        subId: String,
+        memory: Int,
+    ): Duplicate? {
+        val count = downloads.incrementAndGet()
+
+        val subStat = perSubId.get(subId)
+        if (subStat != null) {
+            subStat.incrementAndGet()
+        } else {
+            perSubId.put(subId, AtomicInteger(1))
+        }
+
+        if (count == 1) return null
+
+        val wasted = wastedBytes.addAndGet(memory)
+
+        val subIds = subscriptionIds()
+        val category = if (subIds.size > 1) Category.OVERLAP else Category.REDELIVERY
+
+        return Duplicate(
+            relayUrl = relayUrl,
+            eventId = eventId,
+            kind = kind,
+            author = author,
+            downloads = count,
+            wastedBytesTotal = wasted,
+            lastBytes = memory,
+            subscriptions = subIds,
+            offendingSubId = subId,
+            category = category,
+        )
+    }
+
+    private fun subscriptionIds(): List<String> = perSubId.keys().sorted()
+
+    /** Snapshot describing one wasted (duplicated) download. */
+    data class Duplicate(
+        val relayUrl: NormalizedRelayUrl,
+        val eventId: HexKey,
+        val kind: Int,
+        val author: HexKey,
+        val downloads: Int,
+        val wastedBytesTotal: Int,
+        val lastBytes: Int,
+        val subscriptions: List<String>,
+        val offendingSubId: String,
+        val category: Category,
+    ) {
+        /**
+         * A stable key for the offending filter interaction. For overlaps it is the
+         * sorted set of subscription ids ("subA x subB"); for redeliveries it is the
+         * single re-requesting subscription id ("subA (re)").
+         */
+        fun interactionKey(): String =
+            when (category) {
+                Category.OVERLAP -> subscriptions.joinToString(" x ")
+                Category.REDELIVERY -> "$offendingSubId (re)"
+            }
+    }
+}

amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/dupLogger/DuplicateDownloadStats.kt

@@ -0,0 +1,159 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.service.relayClient.dupLogger
+
+import com.vitorpamplona.amethyst.service.relayClient.dupLogger.RelayDuplicateDownloadLogger.Companion.TAG
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.displayUrl
+import com.vitorpamplona.quartz.utils.Log
+import com.vitorpamplona.quartz.utils.cache.LargeCache
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.concurrent.timer
+
+/**
+ * Aggregates wasted (duplicated) downloads over a rolling window and periodically
+ * logs a digest that points at the filter/assembler interactions that are pulling
+ * the same event from the same relay more than once.
+ *
+ * The digest is grouped by the most actionable dimensions:
+ *  - by interaction (sorted subscription-id set, or a single re-requesting sub) so
+ *    the developer can map it back to a filter via the REQ that [RelayLogger] prints;
+ *  - by relay, to spot relays that we double-pull from;
+ *  - by kind, to spot event kinds whose filters overlap across assemblers.
+ */
+class DuplicateDownloadStats {
+    companion object {
+        const val KB: Int = 1024
+
+        // how often the digest is printed, in milliseconds
+        const val REPORT_PERIOD_MS: Long = 10_000
+    }
+
+    private val duplicateDownloads = AtomicInteger(0)
+    private val wastedBytes = AtomicInteger(0)
+
+    // interactionKey -> wasted downloads + a sample so we know what it is
+    private val byInteraction = LargeCache<String, InteractionTally>()
+    private val byRelay = LargeCache<NormalizedRelayUrl, AtomicInteger>()
+    private val byKind = LargeCache<Int, AtomicInteger>()
+
+    fun add(dup: DownloadRecord.Duplicate) {
+        duplicateDownloads.incrementAndGet()
+        wastedBytes.addAndGet(dup.lastBytes)
+
+        increment(byKind, dup.kind)
+        increment(byRelay, dup.relayUrl)
+
+        val key = dup.interactionKey()
+        val tally = byInteraction.get(key)
+        if (tally != null) {
+            tally.add(dup)
+        } else {
+            byInteraction.put(key, InteractionTally(dup.category).also { it.add(dup) })
+        }
+    }
+
+    private fun <K> increment(
+        cache: LargeCache<K, AtomicInteger>,
+        key: K,
+    ) {
+        val stat = cache.get(key)
+        if (stat != null) {
+            stat.incrementAndGet()
+        } else {
+            cache.put(key, AtomicInteger(1))
+        }
+    }
+
+    fun hasAnything() = duplicateDownloads.get() > 0
+
+    fun reset() {
+        duplicateDownloads.set(0)
+        wastedBytes.set(0)
+        byInteraction.clear()
+        byRelay.forEach { _, value -> value.set(0) }
+        byKind.forEach { _, value -> value.set(0) }
+    }
+
+    fun log() {
+        Log.w(TAG) {
+            "Duplicated downloads in the last ${REPORT_PERIOD_MS / 1000}s: " +
+                "${duplicateDownloads.get()} events, ${wastedBytes.get() / KB}kb wasted"
+        }
+
+        byInteraction
+            .map { key, value -> key to value }
+            .sortedByDescending { it.second.count.get() }
+            .forEach { (key, tally) ->
+                Log.w(TAG) { "-- ${tally.category}: $key -> $tally" }
+            }
+
+        Log.w(TAG) {
+            "-- By relay: " +
+                byRelay.joinToString(", ") { key, value ->
+                    if (value.get() > 0) "${key.displayUrl()} (${value.get()})" else ""
+                }
+        }
+
+        Log.w(TAG) {
+            "-- By kind: " +
+                byKind.joinToString(", ") { key, value ->
+                    if (value.get() > 0) "kind $key (${value.get()})" else ""
+                }
+        }
+    }
+
+    init {
+        timer(name = "DuplicateDownloadReporter", period = REPORT_PERIOD_MS, daemon = true) {
+            if (hasAnything()) {
+                log()
+                reset()
+            }
+        }
+    }
+
+    /** Running waste for a single offending filter interaction within the window. */
+    class InteractionTally(
+        val category: DownloadRecord.Category,
+    ) {
+        val count = AtomicInteger(0)
+        val wastedBytes = AtomicInteger(0)
+        private val kinds = LargeCache<Int, AtomicInteger>()
+        private val relays = LargeCache<NormalizedRelayUrl, AtomicInteger>()
+
+        fun add(dup: DownloadRecord.Duplicate) {
+            count.incrementAndGet()
+            wastedBytes.addAndGet(dup.lastBytes)
+
+            val kindStat = kinds.get(dup.kind)
+            if (kindStat != null) kindStat.incrementAndGet() else kinds.put(dup.kind, AtomicInteger(1))
+
+            val relayStat = relays.get(dup.relayUrl)
+            if (relayStat != null) relayStat.incrementAndGet() else relays.put(dup.relayUrl, AtomicInteger(1))
+        }
+
+        private fun printKinds() = kinds.joinToString(", ") { key, value -> "k$key x${value.get()}" }
+
+        private fun printRelays() = relays.joinToString(", ") { key, value -> "${key.displayUrl()} x${value.get()}" }
+
+        override fun toString() = "${count.get()} dups, ${wastedBytes.get() / KB}kb; kinds [${printKinds()}]; relays [${printRelays()}]"
+    }
+}

amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/dupLogger/RelayDuplicateDownloadLogger.kt

@@ -0,0 +1,119 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.service.relayClient.dupLogger
+
+import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
+import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.displayUrl
+import com.vitorpamplona.quartz.utils.Log
+import com.vitorpamplona.quartz.utils.bytesUsedInMemory
+import com.vitorpamplona.quartz.utils.cache.LargeCache
+
+/**
+ * Debug-only logger that measures how good our filter design is at NOT downloading
+ * the same event twice from the same relay.
+ *
+ * A relay normally returns each matching event only once per subscription. So a
+ * second arrival of the same (relay, eventId) means one of our filters is doing
+ * redundant work:
+ *  - if it arrives again on the SAME subscription id, that subscription re-REQ'd
+ *    without a tight enough `since` (an EOSE/since refinement opportunity);
+ *  - if it arrives on a DIFFERENT subscription id, two assemblers have overlapping
+ *    filters against the same relay (a filter-merging/narrowing opportunity).
+ *
+ * For every duplicate it prints an immediate warning that includes the subscription
+ * ids involved — those ids can be matched to the exact filter that was sent by
+ * cross-referencing the REQ frames printed by [RelayLogger]. It also keeps a rolling
+ * [DuplicateDownloadStats] digest grouped by the offending interaction, relay and kind.
+ *
+ * Memory: it remembers every (relay, eventId) seen for the lifetime of the session,
+ * which is why it must only be enabled on debug builds.
+ *
+ * Wire it up the same way as [RelaySpeedLogger]:
+ * ```
+ * val dupLogger = if (isDebug) RelayDuplicateDownloadLogger(client) else null
+ * ```
+ */
+class RelayDuplicateDownloadLogger(
+    val client: INostrClient,
+) {
+    companion object {
+        val TAG: String = RelayDuplicateDownloadLogger::class.java.simpleName
+    }
+
+    // every (relay, eventId) we have downloaded, plus the subs that delivered it.
+    private val seen = LargeCache<Int, DownloadRecord>()
+
+    private val window = DuplicateDownloadStats()
+
+    private fun key(
+        relayHash: Int,
+        eventIdHash: Int,
+    ): Int = 31 * relayHash + eventIdHash
+
+    private val clientListener =
+        object : RelayConnectionListener {
+            override fun onIncomingMessage(
+                relay: IRelayClient,
+                msgStr: String,
+                msg: Message,
+            ) {
+                if (msg is EventMessage) {
+                    record(relay, msg, msgStr.bytesUsedInMemory())
+                }
+            }
+        }
+
+    private fun record(
+        relay: IRelayClient,
+        msg: EventMessage,
+        memory: Int,
+    ) {
+        val event = msg.event
+        val record =
+            seen.getOrCreate(key(relay.url.hashCode(), event.id.hashCode())) {
+                DownloadRecord(relay.url, event.id, event.kind, event.pubKey)
+            }
+
+        val dup = record.register(msg.subId, memory) ?: return
+
+        window.add(dup)
+
+        Log.w(TAG) {
+            "DUP x${dup.downloads} [${dup.category}] ${relay.url.displayUrl()}: " +
+                "kind ${dup.kind} ev ${dup.eventId.take(8)} by ${dup.author.take(8)}; " +
+                "subs ${dup.subscriptions}; +${dup.lastBytes}b (total ${dup.wastedBytesTotal}b wasted)"
+        }
+    }
+
+    init {
+        Log.d(TAG, "Init, Subscribe")
+        client.addConnectionListener(clientListener)
+    }
+
+    fun destroy() {
+        Log.d(TAG, "Destroy, Unsubscribe")
+        client.removeConnectionListener(clientListener)
+    }
+}