Made upstream endpoints more resilient to failures by checking health/reconnecting at regular intervals.

1 files changed, 34 insertions, 10 deletions

diff --git a/zenserver/upstream/upstreamcache.h b/zenserver/upstream/upstreamcache.h
index 142fe260f..96ee8bddc 100644
--- a/zenserver/upstream/upstreamcache.h
+++ b/zenserver/upstream/upstreamcache.h
@@ -6,6 +6,7 @@
 #include <zencore/iohash.h>
 #include <zencore/zencore.h>
 
+#include <chrono>
 #include <memory>
 
 namespace zen {
@@ -35,18 +36,33 @@ struct UpstreamCacheRecord
 
 struct UpstreamCacheOptions
 {
-	uint32_t ThreadCount   = 4;
-	bool	 ReadUpstream  = true;
-	bool	 WriteUpstream = true;
+	std::chrono::seconds HealthCheckInterval{5};
+	uint32_t			 ThreadCount   = 4;
+	bool				 ReadUpstream  = true;
+	bool				 WriteUpstream = true;
+};
+
+enum class UpstreamStatusCode : uint8_t
+{
+	Ok,
+	Error
+};
+
+struct UpstreamError
+{
+	UpstreamStatusCode StatusCode = UpstreamStatusCode::Ok;
+	std::string		   Reason;
+
+	explicit operator bool() const { return StatusCode != UpstreamStatusCode::Ok; }
 };
 
 struct GetUpstreamCacheResult
 {
-	IoBuffer	Value;
-	std::string Reason;
-	int64_t		Bytes		   = {};
-	double		ElapsedSeconds = {};
-	bool		Success		   = false;
+	IoBuffer	  Value;
+	UpstreamError Error;
+	int64_t		  Bytes			 = {};
+	double		  ElapsedSeconds = {};
+	bool		  Success		 = false;
 };
 
 struct PutUpstreamCacheResult
@@ -57,6 +73,12 @@ struct PutUpstreamCacheResult
 	bool		Success		   = false;
 };
 
+struct UpstreamEndpointHealth
+{
+	std::string Reason;
+	bool		Ok = false;
+};
+
 /**
  * The upstream endpont is responsible for handling upload/downloading of cache records.
  */
@@ -65,7 +87,9 @@ class UpstreamEndpoint
 public:
 	virtual ~UpstreamEndpoint() = default;
 
-	virtual bool Initialize() = 0;
+	virtual bool IsHealthy() const = 0;
+
+	virtual UpstreamEndpointHealth CheckHealth() = 0;
 
 	virtual std::string_view DisplayName() const = 0;
 
@@ -88,7 +112,7 @@ public:
 
 	virtual bool Initialize() = 0;
 
-	virtual void AddEndpoint(std::unique_ptr<UpstreamEndpoint> Endpoint) = 0;
+	virtual void RegisterEndpoint(std::unique_ptr<UpstreamEndpoint> Endpoint) = 0;
 
 	virtual GetUpstreamCacheResult GetCacheRecord(UpstreamCacheKey CacheKey, ZenContentType Type) = 0;