Merge #347f6831: sync: tier live coverage under connection pressure

nostr:nevent1qgsx2lyl2e4zvfadwcvkd9fkrcwczj7mf858hy85mwqclwgut8wpg2spz3mhxue69uhhyetvv9ujumn8d96zuer9wcq3yamnwvaz7tm8d96xummnw3ezucm0d5q3kamnwvaz7tmwva5hgtnyv9hxxmmwwashjer9wchxxmmdqqsrglmgx8q2u69zk6zppglyvn9q45cfv5juy2yv7khq0a23xlpl5hg54dczz

PR-Author: DanConwayDev's Agent
nostr:npub1v47f74n2ycn66asev62nv8sas99akj0g0wg0fkup37u3ckwuzs4q7cwtp0

CoverNote:

Reference coverage now has ordered capacity tiers instead of a binary core/descendant split. Announcements, state, canonical repository `a`, and canonical root `e` coverage remain essential; root `E`, core `q`/`A`, descendant `e`/`a`, and descendant `q` are admitted as complete priority tiers when the shared NIP-11-aware ledger has room. Descendant `E`/`A` and every tier below the cutoff use the existing cursor-overlapped REQ+EOSE rotation.

The four commits are deliberately reviewable in isolation:

1. Name and test the ordered filter vocabulary without changing wire behaviour.
2. Apply the complete-tier live cutoff and paced packed fallback through the existing ledger and five-second scheduler.
3. Union core and descendant reference values before historic grouping, reducing overlapping delivery without changing coverage.
4. Preserve the complete auxiliary frontier when any admitted live subscription receives CLOSED, avoiding a reconciliation-window gap.

## Validation

- `nix develop -c cargo test --lib`: 686 passed before the final focused lifecycle assertion; the final tip's focused lifecycle test also passes.
- All three descendant live/fallback/addressable integration scenarios pass together.
- All three adaptive-pagination scenarios pass.
- `nix build .#ngit-grasp` passes; the final small lifecycle commit was then built successfully by the archive's remote Nix deployment.
- A full `--test sync` run retained timing failures that reproduce on master under the already-merged per-relay query pacer. The branch-specific descendant failures exposed by that run were fixed by rotating one relay-compatible packed REQ rather than singleton filters.

## Archive production evidence

Exact proposal tip `27f31e50ed86ce83f12bd247c4bb7be8b8cbf2b0` was deployed to the disposable archive burn-in instance at 10:13:51 UTC on 2026-08-10 and observed through 10:41:48 UTC. The service remained active with the original PID and zero restarts. Memory was 611 MiB current / 613 MiB peak at the final trend sample while archive hydration was still active.

The archive exercised both sides of the policy across real relay sizes and advertised capacities:

- `git.shakespeare.diy` (fallback budget 20): 4,587 event IDs plus 5 coordinates; 36 filters stayed live in 14 subscriptions and 11 filters became 5 packed rotating REQs.
- `nos.lol` (advertised budget 20): 2,636 event IDs plus 12 coordinates; 24 filters stayed live in 8 subscriptions and 7 filters became 3 packed rotating REQs.
- `relay.primal.net` (advertised budget 20): 850 event IDs plus 3 coordinates; 16 filters stayed live in 5 subscriptions and 3 filters became one rotating REQ.
- `relay.kcbitcoiners.com` (advertised budget 2,000): ordinary auxiliary tiers fit live; only the deliberately historic-only tier rotated.

Across the window, 17 distinct relays exercised persistent tier admission and 17 exercised fallback rotation. The service started 457 packed fallback batches and confirmed 472 descendant-purpose batches. Five early confirmations inherited a broader batch failure flag but each corresponding cursor reached terminal success and later cycles continued; no auxiliary live subscription received CLOSED. There were zero query-rate-limit responses, zero subscription-limit responses, zero panics/OOM signatures, and no unexplained auxiliary-coverage retirement (the sole retirement followed a real frontier change and was reinstalled).

The observed watchdogs were two pre-existing empty historic pages from `relay.nostrcheck.me`; neither involved the new descendant rotation. Four policy refusals came from the access-restricted `relay.getalby.com/v1` and followed the existing 24-hour deferral path. External authentication failures and peer resets remained isolated to their established recovery paths.

## Recommendation

Ready to merge. The archive demonstrated live and rotating coverage concurrently across fallback and advertised capacities from 20 to 2,000, with packed multi-filter rotation completing repeated full cursor laps on the largest relays. The remaining remote failures are pre-existing peer policy, authentication, transport, and historic-page recovery cases rather than regressions in this proposal.

docs/explanation/sync-scaling-constraints.md

@@ -304,8 +304,9 @@ Derived from the tightest commonly observed values; all sizing below assumes:
 Each relay connection owns one implemented budget ledger of B subscription
 slots. Four consumers share it, in priority order:
 
-1. **Core live subscriptions** (persistent, `limit: 0`) — the product; sized
-   first.
+1. **Essential live subscriptions** (persistent, `limit: 0`) — announcements,
+   repository states, canonical repository `a` references and canonical root
+   `e` references are never demoted.
 2. **Reserved margin** (2 slots) — control-plane safety capacity kept beyond
    the live set (which includes Layer-1) for ad-hoc operations and recovery.
 3. **Historic sync and dependency recovery** (transient) — at least one usable
@@ -313,21 +314,24 @@ slots. Four consumers share it, in priority order:
    REQ+EOSE pages/fallbacks/retries, and exact-ID purgatory polls draw from
    the remainder. NEG retains its four-round class cap and transient REQ its
    five-request class cap, but neither can exceed the shared residual.
-4. **Descendant live coverage** (auxiliary persistent) — admitted only when
-   its complete separately grouped set fits after core coverage while still
-   preserving the margin and a transient slot. Otherwise each constrained
-   relay advances one cursor-overlapped REQ+EOSE filter per five-second tick
-   through the same transient queue. Direct thread members contribute their
-   event IDs; replaceable and addressable members also contribute their NIP-01
-   coordinates. The resulting `e`/`E`/`q` and `a`/`A`/coordinate-`q` filters
-   cover one descendant generation without recursively expanding the frontier.
+4. **Priority-tiered reference coverage** — remaining filters are considered
+   in this order: root `E`; core compatibility `q`/`A`; descendant canonical
+   `e`/`a`; descendant `q`. A complete tier remains persistent only when it
+   fits after essential coverage while preserving the margin and a transient
+   slot. Lower tiers advance one relay-compatible, cursor-overlapped REQ+EOSE
+   filter group per five-second tick through the same transient queue. Descendant uppercase
+   `E`/`A` references are historic-only. Direct thread members contribute
+   event IDs and replaceable/addressable coordinates, covering one descendant
+   generation without recursively expanding the frontier.
 
 NIP-11 `max_subscriptions` sets B for each new connection session; when it is
 absent B falls back to 20. Advertised values below that floor are honoured
-(notably nostream's default 10). Two slots remain reserved. Core live filter
-groups are packed first and admitted atomically against the advertised
-subscription-count budget. Incremental five-second batches preserve full core
-REQs and separately owned descendant REQs. The planner uses the same
+(notably nostream's default 10). Two slots remain reserved. Essential live
+filter groups are packed first and admitted atomically against the advertised
+subscription-count budget. Incremental five-second batches preserve full
+essential REQs and separately owned tiered reference REQs. Tier filters are
+packed across boundaries, while admission stops at a complete-tier boundary.
+The planner uses the same
 filter-count and serialized-byte grouping rules as wire submission. It repacks
 the complete mutable core tail with the new filters when that releases at least
 one slot; otherwise it retires only the smallest useful subset which reduces
@@ -335,14 +339,20 @@ the incremental slot cost. Thus byte-bound groups are not rebuilt merely
 because they contain fewer than the maximum filter count. Repository and
 identifier inputs are sorted before byte chunking so equivalent coverage has
 stable group identity.
+Historic repository batches union root and direct-descendant values into one
+`e`/`E`/`q` family, and repository and addressable-descendant coordinates into
+one `a`/`A`/`q` family, before ordinary count/byte grouping. This lets one REQ
+deduplicate events matching both core and descendant references without
+creating a subscription per repository.
 If the changed tail cannot fit the count or learned byte budget, no extension
 is opened and historic recovery remains available. Capacity pressure is the
 backstop which may schedule a complete regroup after outstanding historic
 batches drain; an ordinary tail update never rebuilds stable full groups.
 
-Reconnect and exceptional full restoration still build Layer 1, 2, and 3
-coverage together and reserve the whole grouped set before opening it. Both
-full replacement and tail replacement remember the exact previous grouping:
+Reconnect and exceptional full restoration rebuild essential coverage first;
+the five-second reconciler then admits the largest complete prefix of reference
+tiers which fits the refreshed session budget. Both full replacement and tail
+replacement remember the exact previous grouping:
 a failure while opening a replacement closes every newly opened group and
 restores the retired groups. A partial CLOSE failure likewise reopens any tail
 groups which were already closed before reporting the failure. Multi-connection
@@ -360,7 +370,10 @@ purgatory polling uses the same transient class bound and shared ledger as
 historic pagination. Transient subscription IDs and their permits are
 registered locally before the REQ is sent; this ordering is required because
 an empty or cached response can deliver EOSE/CLOSED before the SDK subscribe
-call returns. Subscribe failure rolls that pre-registration back. Unexpected
+call returns. Negentropy hydration also registers the complete paced chunk set
+and its requested-event accounting in the pending batch before sending the
+first REQ, so early deliveries cannot become an artificial missing residual.
+Subscribe failure rolls both forms of pre-registration back. Unexpected
 CLOSED for a persistent live subscription is
 reported to the manager, which recomputes and transactionally reopens complete
 live coverage. Each reconnect closes the retired ledger and creates a new
@@ -394,9 +407,12 @@ The per-relay event processor retains its 1,000-message bounded data queue.
 Permit release does not depend on that queue draining: a separate listener on
 rust-nostr's broadcast relay notifications consumes only EOSE/CLOSED terminals
 and closes/releases transient ownership. The processor-facing listener still
-delivers ordered EVENT and lifecycle work to the sync actor, but sustained page
-traffic cannot hide terminal accounting behind EVENT backpressure. The data
-queue remains finite as a memory-safety boundary for non-conforming peers.
+delivers ordered EVENT and lifecycle work to the sync actor. EOSE and CLOSED
+use non-blocking actor inboxes: their production is bounded by the session
+subscription ledger, and a large paced historic batch can keep the actor busy
+longer than a fixed lifecycle inbox could safely absorb. This prevents actor
+backpressure from stopping the ordered EVENT processor, while the EVENT queue
+remains finite as the memory-safety boundary for non-conforming peers.
 
 The levers, in the order we reach for them:
 

src/sync/filters.rs

@@ -13,6 +13,165 @@ use std::collections::HashSet;
 
 use nostr_sdk::prelude::*;
 
+/// Ordered live-coverage priorities. Lower-priority tiers may be served by
+/// paced REQ+EOSE rotation when the connection cannot keep every filter live.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum CoverageTier {
+    EssentialCore,
+    RootUppercase,
+    CoreCompatibility,
+    DescendantCanonical,
+    DescendantQuote,
+    HistoricOnly,
+}
+
+#[derive(Debug, Clone)]
+pub struct TieredFilter {
+    pub tier: CoverageTier,
+    pub filter: Filter,
+}
+
+/// Build the live coverage that is never demoted under connection pressure.
+pub fn build_essential_live_filters(
+    full_repos: &HashSet<String>,
+    state_only_repos: &HashSet<String>,
+    root_events: &HashSet<EventId>,
+    since: Option<Timestamp>,
+) -> Vec<Filter> {
+    let all_repos: HashSet<String> = full_repos.union(state_only_repos).cloned().collect();
+    let mut filters = state_event_filters_for_our_repos(&all_repos, since);
+    if !full_repos.is_empty() {
+        filters.extend(
+            tiered_repo_event_filters(full_repos, false, since)
+                .into_iter()
+                .filter(|entry| entry.tier == CoverageTier::EssentialCore)
+                .map(|entry| entry.filter),
+        );
+    }
+    filters.extend(
+        tiered_root_event_filters(root_events, false, since)
+            .into_iter()
+            .filter(|entry| entry.tier == CoverageTier::EssentialCore)
+            .map(|entry| entry.filter),
+    );
+    filters
+}
+
+/// Build the priority-ordered core reference coverage that can move between a
+/// persistent subscription and paced REQ+EOSE rotation.
+pub fn tiered_auxiliary_core_filters(
+    full_repos: &HashSet<String>,
+    root_events: &HashSet<EventId>,
+    since: Option<Timestamp>,
+) -> Vec<TieredFilter> {
+    let mut filters = tiered_root_event_filters(root_events, false, since);
+    filters.extend(tiered_repo_event_filters(full_repos, false, since));
+    filters.retain(|entry| entry.tier != CoverageTier::EssentialCore);
+    filters.sort_by_key(|entry| entry.tier);
+    filters
+}
+
+fn tagged_filters<T: AsRef<str>>(
+    values: &[T],
+    tag: SingleLetterTag,
+    since: Option<Timestamp>,
+) -> Vec<Filter> {
+    chunk_values_by_bytes(values)
+        .into_iter()
+        .map(|chunk| {
+            let mut filter = Filter::new();
+            for value in chunk {
+                filter = filter.custom_tag(tag, value.as_ref());
+            }
+            match since {
+                Some(timestamp) => filter.since(timestamp),
+                None => filter,
+            }
+        })
+        .collect()
+}
+
+fn tiered_tagged_filters<T: AsRef<str>>(
+    values: &[T],
+    tiers: &[(CoverageTier, SingleLetterTag)],
+    since: Option<Timestamp>,
+) -> Vec<TieredFilter> {
+    tiers
+        .iter()
+        .flat_map(|(tier, tag)| {
+            tagged_filters(values, *tag, since)
+                .into_iter()
+                .map(|filter| TieredFilter {
+                    tier: *tier,
+                    filter,
+                })
+        })
+        .collect()
+}
+
+/// Priority-labelled filters for repository coordinates.
+pub fn tiered_repo_event_filters(
+    repos: &HashSet<String>,
+    descendant: bool,
+    since: Option<Timestamp>,
+) -> Vec<TieredFilter> {
+    let mut values: Vec<_> = repos.iter().collect();
+    values.sort_unstable();
+    let tiers = if descendant {
+        [
+            (
+                CoverageTier::DescendantCanonical,
+                SingleLetterTag::LOWERCASE_A,
+            ),
+            (CoverageTier::DescendantQuote, SingleLetterTag::LOWERCASE_Q),
+            (CoverageTier::HistoricOnly, SingleLetterTag::UPPERCASE_A),
+        ]
+    } else {
+        [
+            (CoverageTier::EssentialCore, SingleLetterTag::LOWERCASE_A),
+            (
+                CoverageTier::CoreCompatibility,
+                SingleLetterTag::UPPERCASE_A,
+            ),
+            (
+                CoverageTier::CoreCompatibility,
+                SingleLetterTag::LOWERCASE_Q,
+            ),
+        ]
+    };
+    tiered_tagged_filters(&values, &tiers, since)
+}
+
+/// Priority-labelled filters for event references.
+pub fn tiered_root_event_filters(
+    root_events: &HashSet<EventId>,
+    descendant: bool,
+    since: Option<Timestamp>,
+) -> Vec<TieredFilter> {
+    let mut values: Vec<String> = root_events.iter().map(EventId::to_hex).collect();
+    values.sort_unstable();
+    let tiers = if descendant {
+        [
+            (
+                CoverageTier::DescendantCanonical,
+                SingleLetterTag::LOWERCASE_E,
+            ),
+            (CoverageTier::DescendantQuote, SingleLetterTag::LOWERCASE_Q),
+            (CoverageTier::HistoricOnly, SingleLetterTag::UPPERCASE_E),
+        ]
+    } else {
+        [
+            (CoverageTier::EssentialCore, SingleLetterTag::LOWERCASE_E),
+            (CoverageTier::RootUppercase, SingleLetterTag::UPPERCASE_E),
+            (
+                CoverageTier::CoreCompatibility,
+                SingleLetterTag::LOWERCASE_Q,
+            ),
+        ]
+    };
+    tiered_tagged_filters(&values, &tiers, since)
+}
+
 /// Serialized byte budget of tag values per filter chunk.
 ///
 /// Relay filter limits are byte caps, not item counts: strfry bounds the
@@ -364,6 +523,61 @@ mod tests {
         assert_eq!(filters.len(), 3);
     }
 
+    #[test]
+    fn core_reference_filters_have_stable_priority_tiers() {
+        let repos = HashSet::from(["30617:pubkey:repo".to_string()]);
+        let roots = HashSet::from([EventId::from_byte_array([0; 32])]);
+
+        let repo_tiers: Vec<_> = tiered_repo_event_filters(&repos, false, None)
+            .into_iter()
+            .map(|entry| entry.tier)
+            .collect();
+        assert_eq!(
+            repo_tiers,
+            [
+                CoverageTier::EssentialCore,
+                CoverageTier::CoreCompatibility,
+                CoverageTier::CoreCompatibility,
+            ]
+        );
+
+        let root_tiers: Vec<_> = tiered_root_event_filters(&roots, false, None)
+            .into_iter()
+            .map(|entry| entry.tier)
+            .collect();
+        assert_eq!(
+            root_tiers,
+            [
+                CoverageTier::EssentialCore,
+                CoverageTier::RootUppercase,
+                CoverageTier::CoreCompatibility,
+            ]
+        );
+    }
+
+    #[test]
+    fn descendant_reference_filters_put_uppercase_tags_in_history_only() {
+        let repos = HashSet::from(["1621:pubkey:patch".to_string()]);
+        let roots = HashSet::from([EventId::from_byte_array([0; 32])]);
+
+        for tiers in [
+            tiered_repo_event_filters(&repos, true, None),
+            tiered_root_event_filters(&roots, true, None),
+        ] {
+            assert_eq!(
+                tiers
+                    .into_iter()
+                    .map(|entry| entry.tier)
+                    .collect::<Vec<_>>(),
+                [
+                    CoverageTier::DescendantCanonical,
+                    CoverageTier::DescendantQuote,
+                    CoverageTier::HistoricOnly,
+                ]
+            );
+        }
+    }
+
     #[test]
     fn test_repo_filters_batching() {
         // Each ref serializes to exactly 1024 bytes (1021 + quoting overhead),

src/sync/relay_connection.rs

@@ -490,10 +490,10 @@ where
 /// Events from a relay connection
 #[derive(Debug)]
 pub enum RelayEvent {
-    /// A new event was received (event, subscription_id)
-    Event(Box<Event>, SubscriptionId),
+    /// A new event was received (event, subscription_id, data-lane arrival).
+    Event(Box<Event>, SubscriptionId, std::time::Instant),
     /// End of stored events for a subscription
-    EndOfStoredEvents(SubscriptionId),
+    EndOfStoredEvents(SubscriptionId, std::time::Instant),
     /// NOTICE message from relay
     Notice(String),
     /// Connection was closed
@@ -1201,7 +1201,7 @@ impl RelayConnection {
         filter_groups: Vec<Vec<Filter>>,
     ) -> Result<Vec<SubscriptionId>, String> {
         self.replace_live_filter_groups_with(filter_groups, |filters, permit| {
-            self.subscribe_filters_with_live_permit(filters, None, Some(permit))
+            self.subscribe_filters_with_live_permit(filters, None, Some(permit), None)
         })
         .await
     }
@@ -1229,7 +1229,9 @@ impl RelayConnection {
                     .map(|_| ())
                     .map_err(|error| format!("{subscription_id}: {error}"))
             },
-            |filters, permit| self.subscribe_filters_with_live_permit(filters, None, Some(permit)),
+            |filters, permit| {
+                self.subscribe_filters_with_live_permit(filters, None, Some(permit), None)
+            },
         )
         .await
     }
@@ -1653,6 +1655,7 @@ impl RelayConnection {
                     event,
                     subscription_id,
                 } => {
+                    let data_lane_arrival = std::time::Instant::now();
                     self.record_transient_req_event(&subscription_id);
                     tracing::trace!(
                         relay = %url,
@@ -1661,7 +1664,11 @@ impl RelayConnection {
                         "Received event"
                     );
                     if event_sender
-                        .send(RelayEvent::Event(Box::new(*event), subscription_id.clone()))
+                        .send(RelayEvent::Event(
+                            Box::new(*event),
+                            subscription_id.clone(),
+                            data_lane_arrival,
+                        ))
                         .await
                         .is_err()
                     {
@@ -1671,11 +1678,15 @@ impl RelayConnection {
                 }
                 RelayNotification::Message { message } => match *message {
                     RelayMessage::EndOfStoredEvents(sub_id) => {
+                        let data_lane_arrival = std::time::Instant::now();
                         tracing::debug!(relay = %url, sub_id = ?sub_id, "Received EOSE");
                         // Convert Cow<SubscriptionId> to owned SubscriptionId
                         let owned_sub_id = sub_id.into_owned();
                         if event_sender
-                            .send(RelayEvent::EndOfStoredEvents(owned_sub_id))
+                            .send(RelayEvent::EndOfStoredEvents(
+                                owned_sub_id,
+                                data_lane_arrival,
+                            ))
                             .await
                             .is_err()
                         {
@@ -1816,6 +1827,24 @@ impl RelayConnection {
         self.subscribe_filters(vec![filter], request_class).await
     }
 
+    /// Open a transient subscription with an ID already registered by its
+    /// caller. Pending-batch owners use this to make an immediate EOSE visible
+    /// before the wire request can be answered.
+    pub async fn subscribe_filter_with_id(
+        &self,
+        filter: Filter,
+        request_class: TransientRequestClass,
+        subscription_id: SubscriptionId,
+    ) -> Result<SubscriptionId, String> {
+        self.subscribe_filters_with_live_permit(
+            vec![filter],
+            Some(request_class),
+            None,
+            Some(subscription_id),
+        )
+        .await
+    }
+
     /// Subscribe to several OR filters under one NIP-01 subscription ID.
     ///
     /// Relays apply active-REQ limits to subscription IDs, not to the filters
@@ -1826,7 +1855,7 @@ impl RelayConnection {
         filters: Vec<Filter>,
         request_class: TransientRequestClass,
     ) -> Result<SubscriptionId, String> {
-        self.subscribe_filters_with_live_permit(filters, Some(request_class), None)
+        self.subscribe_filters_with_live_permit(filters, Some(request_class), None, None)
             .await
     }
 
@@ -1838,7 +1867,7 @@ impl RelayConnection {
         filter_groups: Vec<Vec<Filter>>,
     ) -> Result<Vec<SubscriptionId>, String> {
         self.subscribe_live_filter_groups_with(filter_groups, |filters, permit| {
-            self.subscribe_filters_with_live_permit(filters, None, Some(permit))
+            self.subscribe_filters_with_live_permit(filters, None, Some(permit), None)
         })
         .await
     }
@@ -1897,6 +1926,7 @@ impl RelayConnection {
         filters: Vec<Filter>,
         transient_class: Option<TransientRequestClass>,
         reserved_live_permit: Option<SessionPermit>,
+        requested_subscription_id: Option<SubscriptionId>,
     ) -> Result<SubscriptionId, String> {
         if filters.is_empty() {
             return Err("Cannot subscribe with an empty filter set".to_string());
@@ -1969,7 +1999,9 @@ impl RelayConnection {
         // The relay can answer an empty or cached query before `subscribe`
         // returns. Register transient ownership against a caller-chosen ID
         // first so an immediate EOSE/CLOSED cannot race past local accounting.
-        let transient_sub_id = transient_class.map(|_| SubscriptionId::generate());
+        let transient_sub_id = transient_class.map(|_| {
+            requested_subscription_id.unwrap_or_else(SubscriptionId::generate)
+        });
         if let (Some(sub_id), Some(permit)) = (&transient_sub_id, transient_permit) {
             self.hold_transient_req_permit(sub_id.clone(), permit);
         }
@@ -3704,7 +3736,7 @@ mod tests {
                     }
                 },
                 |filters, permit| {
-                    connection.subscribe_filters_with_live_permit(filters, None, Some(permit))
+                    connection.subscribe_filters_with_live_permit(filters, None, Some(permit), None)
                 },
             )
             .await

tests/sync/descendant_sync.rs

@@ -139,7 +139,7 @@ async fn descendant_live_coverage_is_preferred_when_capacity_remains() {
     assert!(
         wait_for_log(
             &syncing.log_path(),
-            "Installed auxiliary descendant live coverage",
+            "Installed priority-bounded auxiliary live coverage",
             Duration::from_secs(20),
         )
         .await,