ref(server): Re-authenticate at regular intervals

CHANGELOG.md

@@ -5,6 +5,7 @@
 **Features**:
 
 - Support chunked formdata keys for event payloads on the Minidump endpoint. Since crashpad has a limit for the length of custom attributes, the sentry event payload can be split up into `sentry__1`, `sentry__2`, etc. ([#721](https://github.com/getsentry/relay/pull/721))
+- Add periodic re-authentication with the upstream server, previously there was only one initial authentication. ([#731](https://github.com/getsentry/relay/pull/731))
 
 **Internal**:
 

relay-config/src/config.rs

@@ -524,6 +524,12 @@ struct Http {
     max_retry_interval: u32,
     /// The custom HTTP Host header to send to the upstream.
     host_header: Option<String>,
+    /// The number of seconds after a successful authentication after which
+    /// a Relay tries to re-authenticate with the upstream server.
+    auth_interval: u64,
+    /// The number of seconds allowed for a Relay to re-authenticate after which
+    /// it is considered that re-authentication has failed.
+    auth_grace_period: u64,
 }
 
 impl Default for Http {
@@ -533,6 +539,8 @@ impl Default for Http {
             connection_timeout: 3,
             max_retry_interval: 60,
             host_header: None,
+            auth_interval: 60,
+            auth_grace_period: 10,
         }
     }
 }
@@ -1100,6 +1108,18 @@ impl Config {
         self.values.relay.tls_identity_password.as_deref()
     }
 
+    /// Returns the interval at which Realy should try to re-authenticate with the upstream
+    pub fn http_auth_interval(&self) -> Duration {
+        Duration::from_secs(self.values.http.auth_interval)
+    }
+
+    /// The maximum amount of time that a Relay is allowed to take to re-authenticate with
+    /// the upstream after which it is declared as un-authenticated (if it is not able to
+    /// authenticate).
+    pub fn http_auth_grace_period(&self) -> Duration {
+        Duration::from_secs(self.values.http.auth_grace_period)
+    }
+
     /// Returns whether this Relay should emit outcomes.
     ///
     /// This is `true` either if `outcomes.emit_outcomes` is explicitly enabled, or if this Relay is

relay-server/src/actors/upstream.rs

@@ -22,6 +22,7 @@ use std::collections::VecDeque;
 use std::fmt;
 use std::str;
 use std::sync::Arc;
+use std::time::Instant;
 
 use ::actix::fut;
 use ::actix::prelude::*;
@@ -75,12 +76,20 @@ pub enum UpstreamRequestError {
     ChannelClosed,
 }
 
+impl UpstreamRequestError {
+    fn is_network_error(&self) -> bool {
+        match self {
+            Self::SendFailed(_) | Self::PayloadFailed(_) => true,
+            Self::ResponseError(code, _) => matches!(code.as_u16(), 502 | 503),
+            _ => false,
+        }
+    }
+}
+
 /// Represents the current auth state.
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
 enum AuthState {
     Unknown,
-    RegisterRequestChallenge,
-    RegisterChallengeResponse,
     Registered,
     Error,
 }
@@ -195,6 +204,7 @@ struct UpstreamRequest {
 
 pub struct UpstreamRelay {
     backoff: RetryBackoff,
+    first_error: Option<Instant>,
     config: Arc<Config>,
     auth_state: AuthState,
     max_inflight_requests: usize,
@@ -257,11 +267,12 @@ impl UpstreamRelay {
         UpstreamRelay {
             backoff: RetryBackoff::new(config.http_max_retry_interval()),
             max_inflight_requests: config.max_concurrent_requests(),
-            config,
             auth_state: AuthState::Unknown,
             num_inflight_requests: 0,
             high_prio_requests: VecDeque::new(),
             low_prio_requests: VecDeque::new(),
+            first_error: None,
+            config,
         }
     }
 
@@ -273,6 +284,17 @@ impl UpstreamRelay {
         }
     }
 
+    fn handle_network_error(&mut self) {
+        let now = Instant::now();
+        let first_error = *self.first_error.get_or_insert(now);
+        if first_error + self.config.http_auth_grace_period() > now {
+            return;
+        }
+
+        self.auth_state = AuthState::Error;
+        // TODO: initiate authentication loop if not already running
+    }
+
     fn send_request(
         &mut self,
         request: UpstreamRequest,
@@ -451,7 +473,7 @@ impl Message for Authenticate {
 /// message was successfully handled).
 ///
 /// **Note:** Relay has retry functionality, outside this actor, that periodically sends Authenticate
-/// messages until successful Authentication with the upstream server was achieved.    
+/// messages until successful Authentication with the upstream server was achieved.
 impl Handler<Authenticate> for UpstreamRelay {
     type Result = ResponseActFuture<Self, (), ()>;
 
@@ -466,29 +488,37 @@ impl Handler<Authenticate> for UpstreamRelay {
             self.config.upstream_descriptor()
         );
 
-        self.auth_state = AuthState::RegisterRequestChallenge;
         let request = RegisterRequest::new(&credentials.id, &credentials.public_key);
 
         let future = self
             .send_query(request)
             .into_actor(self)
             .and_then(|challenge, slf, ctx| {
                 log::debug!("got register challenge (token = {})", challenge.token());
-                slf.auth_state = AuthState::RegisterChallengeResponse;
                 let challenge_response = challenge.create_response();
 
                 log::debug!("sending register challenge response");
                 let fut = slf.send_query(challenge_response).into_actor(slf);
                 ctx.notify(PumpHttpMessageQueue);
                 fut
             })
-            .map(|_, slf, _ctx| {
+            .map(|_, slf, ctx| {
                 log::info!("relay successfully registered with upstream");
                 slf.auth_state = AuthState::Registered;
+
+                slf.backoff.reset();
+                slf.first_error = None;
+
+                ctx.notify_later(Authenticate, slf.config.http_auth_interval());
             })
             .map_err(|err, slf, ctx| {
                 log::error!("authentication encountered error: {}", LogError(&err));
-                slf.auth_state = AuthState::Error;
+
+                if err.is_network_error() {
+                    slf.handle_network_error();
+                } else {
+                    slf.auth_state = AuthState::Error;
+                }
 
                 // Do not retry client errors including authentication failures since client errors
                 // are usually permanent. This allows the upstream to reject unsupported Relays