Gateway: harden downlink freshness, uplink verify ordering, and drain

Fixes the confirmed downlink/uplink defects from the PR #1384 review +
Codex findings:

- Downlink age + #g gate (Codex P2 / review #1): rebroadcastRelayEvent
  now drops events outside the same freshness window receivers enforce
  and whose #g tag mismatches the carrier geohash, BEFORE spending any
  budget — so a 1h/200-event channel-resubscribe backfill no longer
  burns the 30/min BLE budget on events every receiver drops.
- Rate-limit + dedup before Schnorr (review #2): handleUplinkDeposit now
  runs cheap structural checks + carried-ID dedup + rate-token consume
  before isValidSignature(), so a replay flood is bounded by cheap work
  instead of unbounded main-actor verifies.
- Quota-dropped deposits not rendered (review #3): enqueueUplink reports
  acceptance and injectInbound only fires for events actually
  published/queued, ending the local-timeline divergence.
- Drain timer + mark-after-send (Codex P2 / review #4): a burst beyond
  budget now arms a timer to drain when the window frees; rebroadcast
  IDs are marked only after an event is actually sent, so overflow-
  dropped events stay retryable.
- Symmetric publish path (review #5): the gateway publish closure now
  refuses when no geo relay is known, matching the local send path
  instead of publishing dead traffic to default relays.
- Loop-rule doc (review #7): softened to reflect that rule 3 is a
  call-site convention with unit-tested backstops; added tests for the
  publishedEventIDs backstop, downlink freshness/mismatch, drain timer,
  and quota-drop non-injection.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bitchat/Services/Gateway/GatewayService.swift

@@ -30,7 +30,7 @@ import Foundation
 /// - Carried contents are public geohash chat, already plaintext on Nostr,
 ///   so the mesh carrier adds no confidentiality loss.
 ///
-/// Loop-prevention rules (each enforced here and unit-tested):
+/// Loop-prevention rules:
 /// 1. An event learned from a `fromGateway` mesh broadcast is never
 ///    re-published to relays, never re-uplinked, and never rebroadcast
 ///    (`meshBroadcastEventIDs`), so a second gateway on the same mesh cannot
@@ -41,7 +41,10 @@ import Foundation
 ///    repeat deposits and relay echoes are absorbed.
 /// 3. Uplink is only attempted for locally composed events at the send site
 ///    (`GeohashSubscriptionManager.sendGeohash`); events received over the
-///    carrier never re-enter the uplink path.
+///    carrier never re-enter the uplink path. This is a call-site convention;
+///    the `meshBroadcastEventIDs`/`publishedEventIDs` backstops in
+///    `uplinkViaMesh` enforce it defensively and are unit-tested.
+/// Rules 1 and 2 are enforced here and unit-tested.
 /// Rebroadcast storms at the mesh layer are additionally bounded by the BLE
 /// `MessageDeduplicator` and packet TTL, and receivers dedup carried events
 /// against their own relay subscriptions via the Nostr event-ID cache in
@@ -59,9 +62,9 @@ final class GatewayService: ObservableObject {
         /// Uplink deposits accepted per depositor per minute.
         static let uplinkEventsPerMinutePerDepositor = 10
         /// Downlink mesh rebroadcasts per minute — BLE airtime is precious.
-        /// Beyond the budget events queue (bounded, drop-oldest) and drain as
-        /// the window frees; a busy channel triggers frequent drains, a quiet
-        /// one never queues.
+        /// Beyond the budget events queue (bounded, drop-oldest) and drain on
+        /// a scheduled timer once the window frees (also re-driven by the next
+        /// inbound relay event); a quiet channel does not strand its backlog.
         static let downlinkEventsPerMinute = 30
         static let maxPendingDownlinks = 30
         /// Accepted clock skew for a carried ephemeral event; anything older
@@ -105,6 +108,9 @@ final class GatewayService: ObservableObject {
     /// Fired on toggle changes (advertise/withdraw the capability bit and
     /// force a re-announce).
     var onEnabledChanged: (@MainActor (Bool) -> Void)?
+    /// Schedules a downlink-drain closure to run after a delay. Injected so
+    /// the drain timer is deterministic in tests; nil arms a real `Task`.
+    var scheduleDrainTimer: (@MainActor (TimeInterval, @escaping @MainActor () -> Void) -> Void)?
 
     // MARK: State
 
@@ -119,6 +125,8 @@ final class GatewayService: ObservableObject {
     private var uplinkDepositTimes: [PeerID: [Date]] = [:]
     private var downlinkSendTimes: [Date] = []
     private var pendingDownlinks: [(event: NostrEvent, geohash: String)] = []
+    /// True while a drain timer is armed, so a burst schedules at most one.
+    private var downlinkDrainScheduled = false
 
     private let defaults: UserDefaults
     private let now: () -> Date
@@ -174,32 +182,49 @@ final class GatewayService: ObservableObject {
 
     private func handleUplinkDeposit(_ carrier: NostrCarrierPacket, from depositor: PeerID) {
         guard isEnabled else { return }
-        guard let event = validatedEvent(from: carrier) else {
+        // Cheap structural checks first (parse, size, geohash, kind, #g tag,
+        // age) — no crypto — so junk and stale replays are dropped before we
+        // ever pay for a MainActor Schnorr verify.
+        guard let event = structurallyValidEvent(from: carrier) else {
             SecureLogger.debug("🌐 Gateway: rejected uplink deposit from \(depositor.id.prefix(8))… (failed validation)", category: .security)
             return
         }
-        // Loop rule 1: events learned from a fromGateway broadcast are
-        // mesh-carried; a gateway must never re-publish them.
-        guard !meshBroadcastEventIDs.contains(event.id) else { return }
-        // Loop rule 2: repeat deposits of an already handled event are absorbed.
-        guard !publishedEventIDs.contains(event.id),
+        // Dedup by the carried event ID BEFORE verification. Loop rule 1: a
+        // fromGateway-learned event is mesh-carried and must never be
+        // re-published. Loop rule 2: repeat deposits of an already handled
+        // event are absorbed. A replay of one valid deposit is short-circuited
+        // here without a per-packet signature verify.
+        guard !meshBroadcastEventIDs.contains(event.id),
+              !publishedEventIDs.contains(event.id),
               !queuedUplinks.contains(where: { $0.event.id == event.id }) else {
             return
         }
+        // Consume the per-depositor rate token BEFORE the expensive verify so
+        // a flood of distinct forged/junk deposits is bounded by cheap work,
+        // not by main-actor Schnorr verifications.
         guard allowUplinkDeposit(from: depositor) else {
             SecureLogger.debug("🌐 Gateway: rate-limited uplink deposit from \(depositor.id.prefix(8))…", category: .session)
             return
         }
-
-        if relaysConnected?() ?? false {
-            publish(event, geohash: carrier.geohash)
-        } else {
-            enqueueUplink(QueuedUplink(depositor: depositor, geohash: carrier.geohash, event: event, queuedAt: now()))
+        // Only now pay for cryptographic verification; receivers verify again.
+        guard event.isValidSignature() else {
+            SecureLogger.debug("🌐 Gateway: rejected uplink deposit from \(depositor.id.prefix(8))… (bad signature)", category: .security)
+            return
         }
 
-        // Show the carried message on our own timeline when we're viewing
-        // that geohash; the relay echo (if any) dedups in the pipeline.
-        if currentGeohash?() == carrier.geohash {
+        let accepted: Bool
+        if relaysConnected?() ?? false {
+            publish(event, geohash: carrier.geohash)
+            accepted = true
+        } else {
+            accepted = enqueueUplink(QueuedUplink(depositor: depositor, geohash: carrier.geohash, event: event, queuedAt: now()))
+        }
+
+        // Only render on our own timeline what we actually accepted for
+        // publish or queue: a quota-dropped deposit is never published and,
+        // being directed, no other peer will ever see it, so showing it would
+        // diverge our timeline permanently from what reached the channel.
+        if accepted, currentGeohash?() == carrier.geohash {
             injectInbound?(event)
         }
     }
@@ -221,16 +246,19 @@ final class GatewayService: ObservableObject {
         SecureLogger.info("🌐 Gateway: published carried event \(event.id.prefix(8))… to relays for #\(geohash)", category: .session)
     }
 
-    private func enqueueUplink(_ item: QueuedUplink) {
+    /// Returns true when the item was actually stored for later publish.
+    @discardableResult
+    private func enqueueUplink(_ item: QueuedUplink) -> Bool {
         let fromDepositor = queuedUplinks.filter { $0.depositor == item.depositor }.count
         guard fromDepositor < Limits.maxQueuedUplinksPerDepositor else {
             SecureLogger.debug("🌐 Gateway: uplink queue quota reached for \(item.depositor.id.prefix(8))…", category: .session)
-            return
+            return false
         }
         if queuedUplinks.count >= Limits.maxQueuedUplinks {
             queuedUplinks.removeFirst(queuedUplinks.count - Limits.maxQueuedUplinks + 1)
         }
         queuedUplinks.append(item)
+        return true
     }
 
     private func allowUplinkDeposit(from depositor: PeerID) -> Bool {
@@ -258,19 +286,35 @@ final class GatewayService: ObservableObject {
     func rebroadcastRelayEvent(_ event: NostrEvent, geohash: String) {
         guard isEnabled, broadcastToMesh != nil else { return }
         guard event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue else { return }
+        // Freshness + geohash gate BEFORE spending any budget. A channel
+        // (re)subscribe backfills up to an hour of history (limit 200), but
+        // every receiver's `validatedEvent` drops anything older than the
+        // same window — so rebroadcasting backfill would burn the whole
+        // per-minute budget on events no mesh peer accepts. Also require the
+        // event's own `#g` tag to match the carrier geohash.
+        guard isFresh(event),
+              event.tags.contains(where: { $0.count >= 2 && $0[0] == "g" && $0[1] == geohash }) else {
+            return
+        }
         // Loop rule 1: never rebroadcast mesh-carried events back onto the
-        // mesh. Loop rule 2: rebroadcast each relay event at most once.
+        // mesh. Loop rule 2: rebroadcast each relay event at most once — but
+        // mark only AFTER it is actually sent (in `drainPendingDownlinks`), so
+        // an event dropped by the queue overflow stays retryable on relay
+        // redelivery. Guard against a redelivery re-queueing an event that is
+        // still waiting to be sent.
         guard !meshBroadcastEventIDs.contains(event.id),
-              !rebroadcastEventIDs.contains(event.id) else {
+              !rebroadcastEventIDs.contains(event.id),
+              !pendingDownlinks.contains(where: { $0.event.id == event.id }) else {
             return
         }
         // Verify before spending BLE airtime; receivers verify again.
         guard event.isValidSignature() else { return }
-        rebroadcastEventIDs.insert(event.id)
 
         pendingDownlinks.append((event, geohash))
         if pendingDownlinks.count > Limits.maxPendingDownlinks {
-            // Bandwidth guard: drop-oldest — fresher chat is worth more.
+            // Bandwidth guard: drop-oldest — fresher chat is worth more. The
+            // dropped event is not yet in `rebroadcastEventIDs`, so a later
+            // relay redelivery can still carry it.
             pendingDownlinks.removeFirst(pendingDownlinks.count - Limits.maxPendingDownlinks)
         }
         drainPendingDownlinks()
@@ -282,11 +326,44 @@ final class GatewayService: ObservableObject {
         while !pendingDownlinks.isEmpty,
               downlinkSendTimes.count < Limits.downlinkEventsPerMinute {
             let (event, geohash) = pendingDownlinks.removeFirst()
+            // A queued event may have aged past the window while it waited;
+            // don't burn airtime on what receivers would now drop.
+            guard isFresh(event) else { continue }
             guard let carrier = NostrCarrierPacket(direction: .fromGateway, geohash: geohash, event: event),
                   let payload = carrier.encode() else { continue }
             broadcastToMesh?(payload)
+            // Mark-after-send: only now is the relay event definitively
+            // rebroadcast (loop rule 2).
+            rebroadcastEventIDs.insert(event.id)
             downlinkSendTimes.append(now())
         }
+        // Budget exhausted with events still queued: arm a timer to drain when
+        // the window frees, instead of stranding them until the next inbound
+        // relay event (which may never come on a channel that went quiet).
+        scheduleDownlinkDrainIfNeeded()
+    }
+
+    /// Arms a single timer to drain the backlog once the per-minute window
+    /// frees. No-op when nothing is pending or a drain is already scheduled.
+    private func scheduleDownlinkDrainIfNeeded() {
+        guard !pendingDownlinks.isEmpty, !downlinkDrainScheduled else { return }
+        // The window frees when the oldest recorded send ages out of 60s.
+        let oldest = downlinkSendTimes.min() ?? now()
+        let delay = max(0.05, 60 - now().timeIntervalSince(oldest))
+        downlinkDrainScheduled = true
+        let fire: @MainActor () -> Void = { [weak self] in
+            guard let self else { return }
+            self.downlinkDrainScheduled = false
+            self.drainPendingDownlinks()
+        }
+        if let scheduleDrainTimer {
+            scheduleDrainTimer(delay, fire)
+        } else {
+            Task { @MainActor in
+                try? await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
+                fire()
+            }
+        }
     }
 
     // MARK: - Downlink (receiver role: carried event arrives over mesh)
@@ -341,18 +418,36 @@ final class GatewayService: ObservableObject {
     /// before a gateway publishes it or a receiver displays it. Ordered
     /// cheap-first; Schnorr verification runs last.
     private func validatedEvent(from carrier: NostrCarrierPacket) -> NostrEvent? {
+        guard let event = structurallyValidEvent(from: carrier),
+              event.isValidSignature() else {
+            return nil
+        }
+        return event
+    }
+
+    /// The cheap half of `validatedEvent`: parse + size + geohash + kind +
+    /// `#g` tag + freshness, with NO signature verification. Callers that can
+    /// dedup or rate-limit on the carried ID run this first so the expensive
+    /// Schnorr verify is reached only for events that survive the cheap gates.
+    private func structurallyValidEvent(from carrier: NostrCarrierPacket) -> NostrEvent? {
         guard carrier.eventJSON.count <= NostrCarrierPacket.maxEventJSONBytes,
               Self.isValidGeohash(carrier.geohash),
               let event = carrier.event(),
               event.kind == NostrProtocol.EventKind.ephemeralEvent.rawValue,
               event.tags.contains(where: { $0.count >= 2 && $0[0] == "g" && $0[1] == carrier.geohash }),
-              abs(now().timeIntervalSince1970 - TimeInterval(event.created_at)) <= Limits.maxEventAgeSeconds,
-              event.isValidSignature() else {
+              isFresh(event) else {
             return nil
         }
         return event
     }
 
+    /// True when `event.created_at` is within the accepted clock skew — the
+    /// SAME freshness window receivers enforce, so a gateway never spends
+    /// airtime on events every receiver would drop as stale.
+    private func isFresh(_ event: NostrEvent) -> Bool {
+        abs(now().timeIntervalSince1970 - TimeInterval(event.created_at)) <= Limits.maxEventAgeSeconds
+    }
+
     static func isValidGeohash(_ geohash: String) -> Bool {
         let allowed = Set("0123456789bcdefghjkmnpqrstuvwxyz")
         return (1...NostrCarrierPacket.maxGeohashLength).contains(geohash.count)

bitchat/ViewModels/ChatViewModelBootstrapper.swift

@@ -260,11 +260,15 @@ private extension ChatViewModelBootstrapper {
                 toGeohash: geohash,
                 count: TransportConfig.nostrGeoRelayCount
             )
-            if relays.isEmpty {
-                NostrRelayManager.shared.sendEvent(event)
-            } else {
-                NostrRelayManager.shared.sendEvent(event, to: relays)
+            // Symmetric with the local send path (GeohashSubscriptionManager
+            // .sendGeohash): with no known geo relay, refuse rather than
+            // publish to default relays no geo subscriber reads — that would
+            // be silent dead traffic, not delivery.
+            guard !relays.isEmpty else {
+                SecureLogger.warning("🌐 Gateway: no geo relays for #\(geohash); not publishing carried event", category: .session)
+                return
             }
+            NostrRelayManager.shared.sendEvent(event, to: relays)
         }
         gateway.broadcastToMesh = { [weak bleService] payload in
             bleService?.broadcastNostrCarrier(payload)

bitchatTests/Services/GatewayServiceTests.swift

@@ -34,6 +34,7 @@ struct GatewayServiceTests {
         private(set) var injected: [NostrEvent] = []
         private(set) var uplinkSends: [(payload: Data, peer: PeerID)] = []
         private(set) var enabledChanges: [Bool] = []
+        private(set) var scheduledDrains: [(delay: TimeInterval, work: @MainActor () -> Void)] = []
 
         private let clock = ClockBox()
         let defaults: UserDefaults
@@ -60,6 +61,11 @@ struct GatewayServiceTests {
             service.currentGeohash = { [weak self] in self?.currentGeohash }
             service.injectInbound = { [weak self] event in self?.injected.append(event) }
             service.onEnabledChanged = { [weak self] enabled in self?.enabledChanges.append(enabled) }
+            // Capture drain timers instead of arming a real Task so the drain
+            // is deterministic under the fake clock.
+            service.scheduleDrainTimer = { [weak self] delay, work in
+                self?.scheduledDrains.append((delay, work))
+            }
             if enabled {
                 service.setEnabled(true)
             }
@@ -68,6 +74,14 @@ struct GatewayServiceTests {
         func advance(_ seconds: TimeInterval) {
             clock.now = clock.now.addingTimeInterval(seconds)
         }
+
+        /// Fires every currently-scheduled drain timer (simulating the window
+        /// freeing), as the real Task would after its delay.
+        func fireScheduledDrains() {
+            let due = scheduledDrains
+            scheduledDrains.removeAll()
+            for item in due { item.work() }
+        }
     }
 
     // MARK: Event helpers
@@ -258,6 +272,27 @@ struct GatewayServiceTests {
         #expect(fixture.published.count == 1)
     }
 
+    @Test("a quota-dropped deposit is not rendered on the local timeline")
+    func quotaDroppedDepositNotInjected() throws {
+        let fixture = Fixture()
+        fixture.relaysConnected = false
+        let depositor = PeerID(str: "1111111111111111")
+
+        // Fill this depositor's offline queue to its per-depositor cap; each
+        // accepted deposit is rendered locally (we view that geohash).
+        for _ in 0..<GatewayService.Limits.maxQueuedUplinksPerDepositor {
+            try deposit(try makeEvent(), into: fixture, from: depositor)
+        }
+        #expect(fixture.service.queuedUplinks.count == GatewayService.Limits.maxQueuedUplinksPerDepositor)
+        #expect(fixture.injected.count == GatewayService.Limits.maxQueuedUplinksPerDepositor)
+
+        // One past the cap: dropped by the queue quota, so it is neither
+        // queued nor shown — no phantom local message.
+        try deposit(try makeEvent(), into: fixture, from: depositor)
+        #expect(fixture.service.queuedUplinks.count == GatewayService.Limits.maxQueuedUplinksPerDepositor)
+        #expect(fixture.injected.count == GatewayService.Limits.maxQueuedUplinksPerDepositor)
+    }
+
     // MARK: - Loop prevention
 
     @Test("never re-publishes or re-carries an event learned from a fromGateway broadcast")
@@ -338,6 +373,47 @@ struct GatewayServiceTests {
         #expect(fixture.broadcasts.count == overBudget + 1)
     }
 
+    @Test("does not spend downlink budget on stale backfill or geohash mismatch")
+    func downlinkDropsStaleAndMismatched() throws {
+        let fixture = Fixture()
+
+        // A fresh, matching event rebroadcasts.
+        let fresh = try makeEvent()
+        // An event whose #g tag disagrees with the carrier geohash.
+        let mismatched = try makeEvent(geohash: "9q8yyk")
+        // An event that will age past the receiver window before we offer it.
+        let willBeStale = try makeEvent()
+
+        fixture.service.rebroadcastRelayEvent(fresh, geohash: Self.geohash)
+        #expect(fixture.broadcasts.count == 1)
+
+        fixture.service.rebroadcastRelayEvent(mismatched, geohash: Self.geohash)
+        #expect(fixture.broadcasts.count == 1) // geohash mismatch: dropped pre-budget
+
+        fixture.advance(GatewayService.Limits.maxEventAgeSeconds + 60)
+        fixture.service.rebroadcastRelayEvent(willBeStale, geohash: Self.geohash)
+        #expect(fixture.broadcasts.count == 1) // stale backfill: dropped pre-budget
+    }
+
+    @Test("drains a downlink backlog on the scheduled timer when the channel goes quiet")
+    func downlinkDrainsOnTimer() throws {
+        let fixture = Fixture()
+        let overBudget = GatewayService.Limits.downlinkEventsPerMinute + 5
+
+        for _ in 0..<overBudget {
+            fixture.service.rebroadcastRelayEvent(try makeEvent(), geohash: Self.geohash)
+        }
+        // Budget spent; the tail is queued and a drain timer was armed.
+        #expect(fixture.broadcasts.count == GatewayService.Limits.downlinkEventsPerMinute)
+        #expect(!fixture.scheduledDrains.isEmpty)
+
+        // The channel goes quiet — no new inbound event. The window frees and
+        // the timer fires on its own, draining the remaining backlog.
+        fixture.advance(61)
+        fixture.fireScheduledDrains()
+        #expect(fixture.broadcasts.count == overBudget)
+    }
+
     // MARK: - Receiver downlink handling
 
     @Test("injects a carried event once, even when broadcast twice")
@@ -411,6 +487,25 @@ struct GatewayServiceTests {
         #expect(!fixture.service.uplinkViaMesh(event: try makeEvent(), geohash: Self.geohash))
     }
 
+    @Test("uplinkViaMesh backstop refuses an event this gateway already published")
+    func uplinkViaMeshBackstopsPublished() throws {
+        let fixture = Fixture()
+        let event = try makeEvent()
+
+        // This gateway publishes the event to relays (loop rule 2 records it
+        // in publishedEventIDs).
+        try deposit(event, into: fixture)
+        #expect(fixture.published.map(\.event.id) == [event.id])
+
+        // Relays then drop and a gateway peer appears. The same event must not
+        // be re-uplinked from this device — the publishedEventIDs backstop in
+        // uplinkViaMesh catches it even though it was never a mesh broadcast.
+        fixture.relaysConnected = false
+        fixture.gatewayPeers = [PeerID(str: "8877665544332211")]
+        #expect(!fixture.service.uplinkViaMesh(event: event, geohash: Self.geohash))
+        #expect(fixture.uplinkSends.isEmpty)
+    }
+
     // MARK: - Toggle
 
     @Test("persists the toggle and reports changes")