feat(protocol): add a redis health check

Author: stakach

shard.yml

@@ -1,5 +1,5 @@
 name: placeos-driver
-version: 7.13.3
+version: 7.14.0
 
 dependencies:
   action-controller:

spec/helper.cr

@@ -15,6 +15,8 @@ class PlaceOS::Driver::ProcessManagerMock
 
   class_getter callbacks : Hash(String, Proc(Request, Request?)) = {} of String => Request -> Request?
 
+  getter terminated : Channel(Nil) { Channel(Nil).new }
+
   def self.register(name : String, &callback : Request -> Request?)
     ProcessManagerMock.callbacks[name] = callback
   end

src/placeos-driver/protocol.cr

@@ -195,6 +195,7 @@ class PlaceOS::Driver::Protocol
 
   private def produce_io(timeout_period)
     spawn(same_thread: true) { self.process! }
+    spawn { self.redis_health_check }
 
     # Ensures all outgoing event processing is done on the same thread
     @timeouts = Tasker.every(timeout_period) do
@@ -276,4 +277,30 @@ class PlaceOS::Driver::Protocol
     @processor.close
     @timeouts.try &.cancel
   end
+
+  private def redis_health_check
+    failures = 0
+    loop do
+      return if @process_manager.terminated.closed?
+
+      time = (50 + rand(10)).seconds
+      select
+      when @process_manager.terminated.receive?
+        return
+      when timeout(time)
+        # perform health check
+        begin
+          ::PlaceOS::Driver::RedisStorage.with_redis(&.ping)
+          failures = 0
+        rescue error
+          failures += 1
+          raise error if failures >= 2
+          Log.warn(exception: error) { "redis healthcheck failed - retrying" }
+        end
+      end
+    end
+  rescue error
+    Log.fatal(exception: error) { "redis healthcheck failed - terminating process" }
+    @process_manager.terminate
+  end
 end