Add rrpd-max-delta-count option to limit the delta sequence length.

doc/routinator.1

@@ -289,6 +289,12 @@ is silently increased to that value.
 The actual time is chosen at random between the refresh time and this value
 in order to spread out load on the rsync server. 
 
+.TP
+.BI --rrdp-max-delta-count= count
+If the number of deltas necessary to update an RRDP repository is
+larger than the value provided by this option, the snapshot is used instead.
+If the option is missing, the default of 100 is used.
+
 .TP
 .BI --rrdp-timeout= seconds
 Sets the timeout in seconds for any RRDP-related network operation, i.e.,
@@ -1059,6 +1065,12 @@ using rsync. The default in case the value is missing is 3600 seconds. If
 the value provided is smaller than twice the refresh time, it is silently
 increased to that value.
 
+.TP
+.BI rrdp-max-delta-count
+An integer value that specifies the maximum number of deltas necessary to
+update an RRDP repository before using the snapshot instead.
+If the value is missing, the default of 100 is used.
+
 .TP
 .B rrdp-timeout
 An integer value that provides a timeout in seconds for all individual

src/collector/rrdp.rs

@@ -73,6 +73,9 @@ pub struct Collector {
 
     /// The maximum allowed size for published objects.
     max_object_size: Option<u64>,
+
+    /// The maximum number of deltas we process before using a snapshot.
+    max_delta_count: usize,
 }
 
 impl Collector {
@@ -120,6 +123,7 @@ impl Collector {
             filter_dubious: !config.allow_dubious_hosts,
             fallback_time: FallbackTime::from_config(config),
             max_object_size: config.max_object_size,
+            max_delta_count: config.rrdp_max_delta_count,
         }))
     }
 
@@ -736,6 +740,14 @@ impl Repository {
             Err(reason) => return Ok(Some(reason)),
         };
 
+        if deltas.len() > run.collector.max_delta_count {
+            debug!(
+                "RRDP: {}: Too many delta steps required ({})",
+                self.rpki_notify, deltas.len()
+            );
+            return Ok(Some(SnapshotReason::TooManyDeltas))
+        }
+
         if !deltas.is_empty() {
             let count = deltas.len();
             for (i, info) in deltas.iter().enumerate() {
@@ -2129,6 +2141,9 @@ pub enum SnapshotReason {
 
     /// A delta file was conflicting with locally stored data.
     ConflictingDelta,
+
+    /// There were too many deltas to process.
+    TooManyDeltas,
 }
 
 impl SnapshotReason {
@@ -2143,6 +2158,7 @@ impl SnapshotReason {
             LargeSerial => "large-serial",
             OutdatedLocal => "outdate-local",
             ConflictingDelta => "conflicting-delta",
+            TooManyDeltas => "too-many-deltas",
         }
     }
 }

src/config.rs

@@ -50,6 +50,9 @@ const DEFAULT_RRDP_TIMEOUT: Duration = Duration::from_secs(300);
 /// The default for the RRDP fallback time.
 const DEFAULT_RRDP_FALLBACK_TIME: Duration = Duration::from_secs(3600);
 
+/// The default for the maximum number of deltas.
+const DEFAULT_RRDP_MAX_DELTA_COUNT: usize = 100;
+
 /// The default RRDP HTTP User Agent header value to send.
 const DEFAULT_RRDP_USER_AGENT: &str = concat!("Routinator/", crate_version!());
 
@@ -172,6 +175,9 @@ pub struct Config {
     /// Time since last update of an RRDP repository before fallback to rsync.
     pub rrdp_fallback_time: Duration,
 
+    /// The maxmimm number of deltas we allow before using snapshot.
+    pub rrdp_max_delta_count: usize,
+
     /// RRDP timeout in seconds.
     ///
     /// If this is None, no timeout is set.
@@ -368,6 +374,12 @@ impl Config {
             .long("disable-rrdp")
             .help("Disable RRDP and only use rsync")
         )
+        .arg(Arg::with_name("rrdp-max-delta-count")
+            .long("rrdp-max-delta-count")
+            .takes_value(true)
+            .value_name("COUNT")
+            .help("Maximum number of RRDP deltas before using snapshot")
+        )
         .arg(Arg::with_name("rrdp-fallback-time")
             .long("rrdp-fallback-time")
             .takes_value(true)
@@ -699,6 +711,13 @@ impl Config {
             self.rrdp_fallback_time = Duration::from_secs(value)
         }
 
+        // rrdp_max_delta_count
+        if let Some(value) = from_str_value_of(
+            matches, "rrdp-max-delta-count"
+        )? {
+            self.rrdp_max_delta_count = value
+        }
+
         // rrdp_timeout
         if let Some(value) = from_str_value_of(matches, "rrdp-timeout")? {
             self.rrdp_timeout = if value == 0 {
@@ -1035,6 +1054,10 @@ impl Config {
                 .map(Duration::from_secs)
                 .unwrap_or(DEFAULT_RRDP_FALLBACK_TIME)
             },
+            rrdp_max_delta_count: {
+                file.take_usize("rrdp-max-delta-count")?
+                .unwrap_or(DEFAULT_RRDP_MAX_DELTA_COUNT)
+            },
             rrdp_timeout: {
                 match file.take_u64("rrdp-timeout")? {
                     Some(0) => None,
@@ -1233,6 +1256,7 @@ impl Config {
             rsync_timeout: Duration::from_secs(DEFAULT_RSYNC_TIMEOUT),
             disable_rrdp: false,
             rrdp_fallback_time: DEFAULT_RRDP_FALLBACK_TIME,
+            rrdp_max_delta_count: DEFAULT_RRDP_MAX_DELTA_COUNT,
             rrdp_timeout: Some(DEFAULT_RRDP_TIMEOUT), 
             rrdp_connect_timeout: None,
             rrdp_local_addr: None,
@@ -1378,6 +1402,10 @@ impl Config {
             "rrdp-fallback-time".into(),
             (self.rrdp_fallback_time.as_secs() as i64).into()
         );
+        res.insert(
+            "rrdp-max-delta-count".into(),
+            i64::try_from(self.rrdp_max_delta_count).unwrap_or(i64::MAX).into()
+        );
         res.insert(
             "rrdp-timeout".into(),
             match self.rrdp_timeout {
@@ -1774,7 +1802,7 @@ impl ConfigFile {
             None => Ok(None)
         }
     }
-    
+
     /// Takes an unsigned integer value from the config file.
     ///
     /// The value is taken from the given `key`. Returns `Ok(None)` if there
@@ -1809,6 +1837,37 @@ impl ConfigFile {
         }
     }
 
+    /// Takes an unsigned integer value from the config file.
+    ///
+    /// The value is taken from the given `key`. Returns `Ok(None)` if there
+    /// is no such key. Returns an error if the key exists but the value
+    /// isn’t an integer or if it is negative.
+    fn take_usize(&mut self, key: &str) -> Result<Option<usize>, Failed> {
+        match self.content.remove(key) {
+            Some(value) => {
+                if let toml::Value::Integer(res) = value {
+                    usize::try_from(res).map(Some).map_err(|_| {
+                        error!(
+                            "Failed in config file {}: \
+                            '{}' expected to be a positive integer.",
+                            self.path.display(), key
+                        );
+                        Failed
+                    })
+                }
+                else {
+                    error!(
+                        "Failed in config file {}: \
+                         '{}' expected to be an integer.",
+                        self.path.display(), key
+                    );
+                    Err(Failed)
+                }
+            }
+            None => Ok(None)
+        }
+    }
+
     /// Takes a small unsigned integer value from the config file.
     ///
     /// While the result is returned as an `usize`, it must be in the