fix: identify RequestError::Transport as infrastructure error

examples/watcher/watcher-config.toml

@@ -1,7 +1,9 @@
 testnet = false
 homeserver = "8um71us3fyw6h8wbcxb5ar3rwusy1a6u49956ikzojg3gcwd1dty"
 events_limit = 50
+key_based_events_limit = 50
 watcher_sleep = 5000
+hs_resolver_sleep = 10000
 # Initial backoff duration (in seconds) after the first failure of a homeserver
 initial_backoff_secs = 60
 # Maximum backoff duration (in seconds) for a failing homeserver
@@ -18,6 +20,21 @@ moderated_tags = [
     "il_adult_nu_sex_act",
 ]
 
+# Event retry configuration
+[retry]
+# Transient error retry limit before dead-letter
+max_retries = 10
+# Safety net for homeservers that disappear silently
+max_dependency_retries = 50
+# Base for exponential backoff on transient retries (seconds)
+initial_backoff_secs = 10
+# Backoff ceiling for transient retries (seconds)
+max_backoff_secs = 3600
+# Base for MissingDependency polling backoff (seconds)
+initial_missing_dep_backoff_secs = 60
+# Backoff ceiling for MissingDependency (seconds)
+max_missing_dep_backoff_secs = 3600
+
 [stack]
 # Logging, options: error, warn, info, debug and trace
 log_level = "debug"

nexus-common/default.config.toml

@@ -14,11 +14,16 @@ testnet = false
 testnet_host = "localhost"
 # Synonym homeserver pubky
 homeserver = "8um71us3fyw6h8wbcxb5ar3rwusy1a6u49956ikzojg3gcwd1dty"
-# Maximum number of events to fetch per run from each homeserver
+# Maximum number of events to fetch per run from the default homeserver (max: 1000)
 events_limit = 50
+# Maximum events per user per run for key-based (non-default) homeservers (max: 100)
+key_based_events_limit = 50
 # Maximum number of monitored homeservers. If set to 1, only the default homeserver is monitored.
 monitored_homeservers_limit = 50
 watcher_sleep = 5000
+hs_resolver_sleep = 10000
+# Minimum time (ms) before a user's homeserver mapping is considered stale and therefore eligible to be re-resolved (default: 1 hour)
+hs_resolver_ttl = 3600000
 # Initial backoff duration (in seconds) after the first failure of a homeserver
 initial_backoff_secs = 60
 # Maximum backoff duration (in seconds) for a failing homeserver
@@ -35,6 +40,20 @@ moderated_tags = [
     "il_adult_nu_sex_act",
 ]
 
+# Event retry configuration
+[watcher.retry]
+# Transient error retry limit before dead-letter
+max_retries = 10
+# Safety net for homeservers that disappear silently
+max_dependency_retries = 50
+# Base for exponential backoff on transient retries (seconds)
+initial_backoff_secs = 10
+# Backoff ceiling for transient retries (seconds)
+max_backoff_secs = 3600
+# Base for MissingDependency polling backoff (seconds)
+initial_missing_dep_backoff_secs = 60
+# Backoff ceiling for MissingDependency (seconds)
+max_missing_dep_backoff_secs = 3600
 
 [stack]
 # Logging, options: error, warn, info, debug and trace

nexus-common/src/config/daemon.rs

@@ -84,7 +84,9 @@ mod tests {
             PubkyId::try_from("8um71us3fyw6h8wbcxb5ar3rwusy1a6u49956ikzojg3gcwd1dty").unwrap()
         );
         assert_eq!(c.watcher.events_limit, 50);
+        assert_eq!(c.watcher.key_based_events_limit, 50);
         assert_eq!(c.watcher.watcher_sleep, 5_000);
+        assert_eq!(c.watcher.hs_resolver_sleep, 10_000);
         assert_eq!(
             c.watcher.moderation_id,
             PubkyId::try_from(DEFAULT_MODERATION_ID).unwrap()

nexus-common/src/config/mod.rs

@@ -32,13 +32,15 @@ mod api;
 mod daemon;
 pub mod file;
 mod stack;
-mod watcher;
+pub mod watcher;
 
 pub use api::ApiConfig;
 pub use daemon::DaemonConfig;
 pub use stack::{default_stack, OtlpConfig, StackConfig};
-pub use watcher::WatcherConfig;
-pub use watcher::{DEFAULT_INITIAL_BACKOFF_SECS, DEFAULT_MAX_BACKOFF_SECS};
+pub use watcher::{
+    EventRetryConfig, WatcherConfig, DEFAULT_HS_RESOLVER_TTL, DEFAULT_INITIAL_BACKOFF_SECS,
+    DEFAULT_MAX_BACKOFF_SECS, MAX_EVENTS_LIMIT, MAX_KEY_BASED_EVENTS_LIMIT,
+};
 
 use crate::file::validate_and_expand_path;
 

nexus-common/src/config/watcher.rs

@@ -1,24 +1,51 @@
+use crate::models::event::EventProcessorError;
+
 use super::file::ConfigLoader;
 use super::{default_stack, DaemonConfig, StackConfig};
 use async_trait::async_trait;
 use pubky_app_specs::PubkyId;
-use serde::{Deserialize, Serialize};
+use serde::{de::Error, Deserialize, Deserializer, Serialize};
 use std::fmt::Debug;
 
 pub const TESTNET: bool = false;
 pub const DEFAULT_TESTNET_HOST: &str = "localhost";
 // Testnet homeserver key
 pub const HOMESERVER_PUBKY: &str = "8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo";
 /// Default for [WatcherConfig::events_limit]
-pub const DEFAULT_EVENTS_LIMIT: u32 = 1_000;
+pub const DEFAULT_EVENTS_LIMIT: u16 = 1_000;
+/// Default for [WatcherConfig::key_based_events_limit]
+pub const DEFAULT_KEY_BASED_EVENTS_LIMIT: u16 = 50;
+/// Upper bound for [WatcherConfig::events_limit]
+pub const MAX_EVENTS_LIMIT: u16 = 1_000;
+/// Upper bound for [WatcherConfig::key_based_events_limit]
+pub const MAX_KEY_BASED_EVENTS_LIMIT: u16 = 100;
 /// Default for [WatcherConfig::monitored_homeservers_limit]
 pub const DEFAULT_MONITORED_HOMESERVERS_LIMIT: usize = 50;
 /// Default for [WatcherConfig::watcher_sleep]
 pub const DEFAULT_WATCHER_SLEEP: u64 = 5_000;
+/// Default for [WatcherConfig::hs_resolver_sleep]
+pub const DEFAULT_HS_RESOLVER_SLEEP: u64 = 10_000;
+/// Default for [WatcherConfig::hs_resolver_ttl]: 1 hour in milliseconds
+pub const DEFAULT_HS_RESOLVER_TTL: u64 = 3_600_000;
 /// Default for [WatcherConfig::initial_backoff_secs]
 pub const DEFAULT_INITIAL_BACKOFF_SECS: u64 = 60;
 /// Default for [WatcherConfig::max_backoff_secs]
 pub const DEFAULT_MAX_BACKOFF_SECS: u64 = 3_600;
+
+// Retry configuration defaults
+/// Default for [EventRetryConfig::max_retries]
+pub const DEFAULT_MAX_RETRIES: u32 = 10;
+/// Default for [EventRetryConfig::max_dependency_retries]
+pub const DEFAULT_MAX_DEPENDENCY_RETRIES: u32 = 50;
+/// Default for [EventRetryConfig::initial_backoff_secs] (transient errors)
+pub const DEFAULT_INITIAL_TRANSIENT_BACKOFF_SECS: u64 = 10;
+/// Default for [EventRetryConfig::max_backoff_secs] (transient errors)
+pub const DEFAULT_MAX_TRANSIENT_BACKOFF_SECS: u64 = 3_600;
+/// Default for [EventRetryConfig::initial_missing_dep_backoff_secs]
+pub const DEFAULT_INITIAL_MISSING_DEP_BACKOFF_SECS: u64 = 60;
+/// Default for [EventRetryConfig::max_missing_dep_backoff_secs]
+pub const DEFAULT_MAX_MISSING_DEP_BACKOFF_SECS: u64 = 3_600;
+
 // Default moderation service key (test user key, overridden by config.toml value)
 pub const DEFAULT_MODERATION_ID: &str = "uo7jgkykft4885n8cruizwy6khw71mnu5pq3ay9i8pw1ymcn85ko";
 // Moderation service key
@@ -31,19 +58,89 @@ pub const MODERATED_TAGS: [&str; 6] = [
     "il_adult_nu_sex_act",
 ];
 
+/// Retry configuration settings
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
+#[serde(default)]
+pub struct EventRetryConfig {
+    /// Transient error retry limit before dead-letter
+    pub max_retries: u32,
+    /// Safety net for homeservers that disappear silently (no DEL events, content just gone)
+    pub max_dependency_retries: u32,
+    /// Base for exponential backoff on transient retries (seconds)
+    pub initial_backoff_secs: u64,
+    /// Backoff ceiling for transient retries (seconds)
+    pub max_backoff_secs: u64,
+    /// Base for MissingDependency polling backoff (seconds)
+    pub initial_missing_dep_backoff_secs: u64,
+    /// Backoff ceiling for MissingDependency (seconds)
+    pub max_missing_dep_backoff_secs: u64,
+}
+
+impl Default for EventRetryConfig {
+    fn default() -> Self {
+        Self {
+            max_retries: DEFAULT_MAX_RETRIES,
+            max_dependency_retries: DEFAULT_MAX_DEPENDENCY_RETRIES,
+            initial_backoff_secs: DEFAULT_INITIAL_TRANSIENT_BACKOFF_SECS,
+            max_backoff_secs: DEFAULT_MAX_TRANSIENT_BACKOFF_SECS,
+            initial_missing_dep_backoff_secs: DEFAULT_INITIAL_MISSING_DEP_BACKOFF_SECS,
+            max_missing_dep_backoff_secs: DEFAULT_MAX_MISSING_DEP_BACKOFF_SECS,
+        }
+    }
+}
+
+impl EventRetryConfig {
+    /// Returns (initial_backoff, max_backoff) values, in seconds, for the given error
+    pub fn get_backoff_params(&self, error: &EventProcessorError) -> (u64, u64) {
+        let initial = match error.is_missing_dependency() {
+            true => self.initial_missing_dep_backoff_secs,
+            false => self.initial_backoff_secs,
+        };
+        let max = match error.is_missing_dependency() {
+            true => self.max_missing_dep_backoff_secs,
+            false => self.max_backoff_secs,
+        };
+        (initial, max)
+    }
+}
+
 /// Configuration settings for the Nexus Watcher service
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
 pub struct WatcherConfig {
     pub testnet: bool,
     pub testnet_host: String,
+
     /// Default homeserver. Other homeservers may be ingested in addition, but this one is prioritized.
     pub homeserver: PubkyId,
-    /// Maximum number of events to fetch per run from each homeserver
-    pub events_limit: u32,
+
+    /// Maximum number of events to fetch per run from the default homeserver.
+    /// Must not exceed [MAX_EVENTS_LIMIT].
+    #[serde(deserialize_with = "deserialize_events_limit")]
+    pub events_limit: u16,
+
+    /// Maximum events per user per run for key-based (non-default) homeservers.
+    /// Must not exceed [MAX_KEY_BASED_EVENTS_LIMIT].
+    #[serde(
+        default = "default_key_based_events_limit",
+        deserialize_with = "deserialize_key_based_events_limit"
+    )]
+    pub key_based_events_limit: u16,
+
     /// Maximum number of monitored homeservers
     pub monitored_homeservers_limit: usize,
+
     /// Sleep between every full run (over all monitored homeservers), in milliseconds
     pub watcher_sleep: u64,
+
+    /// Sleep between every run of the user HS resolver periodic task, in milliseconds
+    #[serde(default = "default_hs_resolver_sleep")]
+    pub hs_resolver_sleep: u64,
+
+    /// Minimum time (ms) before a user's homeserver mapping is re-resolved.
+    /// Users whose `HOSTED_BY.resolved_at` is newer than this TTL are skipped.
+    #[serde(default = "default_hs_resolver_ttl")]
+    pub hs_resolver_ttl: u64,
+
     /// Initial backoff duration (in seconds) after the first failure of a homeserver
     #[serde(default = "default_initial_backoff_secs")]
     pub initial_backoff_secs: u64,
@@ -52,6 +149,11 @@ pub struct WatcherConfig {
     pub max_backoff_secs: u64,
     #[serde(default = "default_stack")]
     pub stack: StackConfig,
+
+    // Retry configuration
+    #[serde(default)]
+    pub retry: EventRetryConfig,
+
     // Moderation
     pub moderation_id: PubkyId,
     pub moderated_tags: Vec<String>,
@@ -72,16 +174,69 @@ impl Default for WatcherConfig {
             testnet_host: DEFAULT_TESTNET_HOST.to_string(),
             homeserver,
             events_limit: DEFAULT_EVENTS_LIMIT,
+            key_based_events_limit: DEFAULT_KEY_BASED_EVENTS_LIMIT,
             monitored_homeservers_limit: DEFAULT_MONITORED_HOMESERVERS_LIMIT,
             watcher_sleep: DEFAULT_WATCHER_SLEEP,
+            hs_resolver_sleep: DEFAULT_HS_RESOLVER_SLEEP,
+            hs_resolver_ttl: DEFAULT_HS_RESOLVER_TTL,
             initial_backoff_secs: DEFAULT_INITIAL_BACKOFF_SECS,
             max_backoff_secs: DEFAULT_MAX_BACKOFF_SECS,
+            retry: EventRetryConfig::default(),
             moderation_id,
             moderated_tags: MODERATED_TAGS.iter().map(|s| s.to_string()).collect(),
         }
     }
 }
 
+fn default_hs_resolver_sleep() -> u64 {
+    DEFAULT_HS_RESOLVER_SLEEP
+}
+
+fn default_key_based_events_limit() -> u16 {
+    DEFAULT_KEY_BASED_EVENTS_LIMIT
+}
+
+fn deserialize_events_limit<'de, D>(deserializer: D) -> Result<u16, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let val = u16::deserialize(deserializer)?;
+    if val == 0 {
+        return Err(D::Error::custom("events_limit must be at least 1"));
+    }
+
+    if val > MAX_EVENTS_LIMIT {
+        let err_msg = format!("events_limit ({val}) exceeds max ({MAX_EVENTS_LIMIT})");
+        Err(D::Error::custom(err_msg))
+    } else {
+        Ok(val)
+    }
+}
+
+fn deserialize_key_based_events_limit<'de, D>(deserializer: D) -> Result<u16, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let val = u16::deserialize(deserializer)?;
+    if val == 0 {
+        return Err(D::Error::custom(
+            "key_based_events_limit must be at least 1",
+        ));
+    }
+
+    if val > MAX_KEY_BASED_EVENTS_LIMIT {
+        let err_msg =
+            format!("key_based_events_limit ({val}) exceeds max ({MAX_KEY_BASED_EVENTS_LIMIT})");
+        Err(D::Error::custom(err_msg))
+    } else {
+        Ok(val)
+    }
+}
+
+fn default_hs_resolver_ttl() -> u64 {
+    DEFAULT_HS_RESOLVER_TTL
+}
+
 /// Converts a [`DaemonConfig`] into an [`WatcherConfig`], extracting only the Watcher-related settings
 /// and the shared application stack
 impl From<DaemonConfig> for WatcherConfig {