fix server ready

.github/workflows/ci.yml

@@ -206,3 +206,56 @@ jobs:
           name: integration-tests-reports
           path: ~/data/reports.tar
           retention-days: 7
+
+  distributed-tests:
+    needs: build-and-test
+    runs-on: ubuntu-latest
+    name: distributed-tests
+    strategy:
+      fail-fast: false
+
+    steps:
+      - name: retrieve saved docker image
+        uses: actions/download-artifact@v3
+        with:
+          name: image-testing
+          path: ~/data
+
+      - name: docker load
+        run: docker load -i ~/data/new_hstream_image.tar
+
+      - name: fetch distributed tests source code
+        uses: actions/checkout@v2
+        with:
+          repository: "hstreamdb/distributed-tests"
+          submodules: "recursive"
+          path: distributed-tests
+
+      - uses: actions/setup-java@v2
+        with:
+          distribution: "adopt"
+          java-version: 11
+          cache: "gradle"
+
+      - uses: gradle/wrapper-validation-action@v1
+
+      - name: run distributed tests
+        run: |
+          cd distributed-tests
+          export HSTREAM_IMAGE_NAME=$NEW_HSTREAM_IMAGE
+          ./gradlew test --info --fail-fast
+
+      - name: tar tests reports
+        if: ${{ success() }} || ${{ failure() }}
+        run: |
+          mv distributed-tests/.logs distributed-tests/logs
+          mv distributed-tests/app/build/reports distributed-tests/reports
+          tar -cvf ~/data/reports.tar distributed-tests/logs distributed-tests/reports
+
+      - name: upload tests-reports
+        uses: actions/upload-artifact@v2
+        if: ${{ success() }} || ${{ failure() }}
+        with:
+          name: distributed-tests-reports
+          path: ~/data/reports.tar
+          retention-days: 7

hstream-gossip/src/HStream/Gossip/Core.hs

@@ -7,7 +7,7 @@
 module HStream.Gossip.Core where
 
 import           Control.Concurrent               (killThread, newEmptyMVar,
-                                                   putMVar, takeMVar,
+                                                   putMVar, readMVar, takeMVar,
                                                    tryPutMVar)
 import           Control.Concurrent.STM           (atomically, check, dupTChan,
                                                    flushTQueue, modifyTVar,
@@ -214,7 +214,8 @@ handleEventMessage gc@GossipContext{..} msg@(EventMessage eName lpTime bs) = do
             then do
               Log.info . Log.buildString $ "[Server Node" <> show (I.serverNodeId serverSelf)
                                         <> "] Handling Internal Event" <> show eName <> " with lamport " <> show lpInt
-              handleINITEvent gc bs
+              (isSeed, _, _) <- readMVar seedsInfo
+              when isSeed $ handleINITEvent gc bs
             else Log.info $ "Action dealing with event " <> Log.buildString' eName <> " not found"
           Just action -> do
             action bs
@@ -228,7 +229,7 @@ handleINITEvent gc@GossipContext{..} payload = do
       void $ tryPutMVar clusterInited Gossip
       atomically $ do
         mWorkers <- readTVar workers
-        check $ Map.size mWorkers == length seeds
+        check $ (Map.size mWorkers + 1) == length seeds
       broadCastUserEvent gc eventNameINITED (BL.toStrict $ PT.toLazyByteString serverSelf)
 
 broadCastUserEvent :: GossipContext -> EventName -> EventPayload -> IO ()

hstream-gossip/src/HStream/Gossip/Start.hs

@@ -68,14 +68,14 @@ initGossipContext gossipOpts _eventHandlers serverSelf seeds = do
   clusterInited <- newEmptyMVar
   clusterReady  <- newEmptyMVar
   numInited     <- newEmptyMVar
+  seedsInfo     <- newEmptyMVar
   let eventHandlers = Map.insert eventNameINITED (handleINITEDEvent numInited (length seeds) clusterReady) _eventHandlers
   return GossipContext {..}
 
 startGossip :: GossipContext -> IO ()
 startGossip gc@GossipContext {..} = do
-  seedInfo <- newEmptyMVar
-  void $ startListeners (I.serverNodeHost serverSelf) seedInfo gc
-  (isSeed, seeds', wasIDead) <- readMVar seedInfo
+  void $ startListeners (I.serverNodeHost serverSelf) gc
+  (isSeed, seeds', wasIDead) <- readMVar seedsInfo
   if isSeed && not wasIDead
     then newTVarIO 0 >>= putMVar numInited . Just
       >> threadDelay 1000000
@@ -137,13 +137,13 @@ handleINITEDEvent initedM l ready payload = readMVar initedM >>= \case
       if x == l then putMVar ready () >> Log.info "All servers have been initialized"
         else when (x > l) $ Log.warning "More seeds has been initiated, something went wrong"
 
-startListeners ::  ByteString -> MVar (Bool, [(ByteString, Int)], Bool) -> GossipContext -> IO (Async ())
-startListeners grpcHost seedInfo gc@GossipContext {..} = do
+startListeners ::  ByteString -> GossipContext -> IO (Async ())
+startListeners grpcHost gc@GossipContext {..} = do
   let grpcOpts = GRPC.defaultServiceOptions {
       GRPC.serverHost = GRPC.Host grpcHost
     , GRPC.serverPort = GRPC.Port $ fromIntegral $ I.serverNodeGossipPort serverSelf
     , GRPC.serverOnStarted = Just $ do
-        void . forkIO $ amIASeed serverSelf seeds >>= putMVar seedInfo
+        void . forkIO $ amIASeed serverSelf seeds >>= putMVar seedsInfo
         Log.debug . Log.buildString $ "Server node " <> show serverSelf <> " started"
     }
   let api = handlers gc

hstream-gossip/src/HStream/Gossip/Types.hs

@@ -73,6 +73,7 @@ data GossipContext = GossipContext
   { serverSelf    :: I.ServerNode
   , eventHandlers :: EventHandlers
   , seeds         :: [(ByteString, Int)]
+  , seedsInfo     :: MVar (Bool, [(ByteString, Int)], Bool)
   , serverList    :: TVar ServerList
   , actionChan    :: TChan RequestAction
   , statePool     :: TQueue G.StateMessage