Merge remote-tracking branch 'origin/adinapoli/issue-198-part-2' into dev-merge

cabal.project.freeze

@@ -1211,7 +1211,7 @@ constraints: any.AC-Angle ==1.0,
              any.hspec-contrib ==0.5.1,
              any.hspec-core ==2.7.10,
              any.hspec-discover ==2.7.10,
-             any.hspec-expectations ==0.8.2,
+             any.hspec-expectations ==0.8.3,
              any.hspec-expectations-json ==1.0.0.4,
              any.hspec-expectations-lifted ==0.10.0,
              any.hspec-expectations-pretty-diff ==0.7.2.6,
@@ -2019,6 +2019,7 @@ constraints: any.AC-Angle ==1.0,
              any.record-hasfield ==1.0,
              any.record-wrangler ==0.1.1.0,
              any.records-sop ==0.1.1.0,
+             any.recover-rtti ==0.4.3,
              any.recursion-schemes ==5.2.2.2,
              any.reducers ==3.12.4,
              any.ref-fd ==0.5,

gargantext.cabal

@@ -868,11 +868,12 @@ test-suite garg-test
       Parsers.Date
       Parsers.Types
       Parsers.WOS
+      Utils
       Utils.Crypto
       Utils.Jobs
       Paths_gargantext
   hs-source-dirs:
-      src-test
+      test
   default-extensions:
       DataKinds
       DeriveGeneric
@@ -912,6 +913,7 @@ test-suite garg-test
     , gargantext
     , gargantext-prelude
     , hspec
+    , hspec-expectations >= 0.8.3
     , http-client
     , http-client-tls
     , mtl
@@ -920,6 +922,7 @@ test-suite garg-test
     , patches-map
     , quickcheck-instances
     , raw-strings-qq
+    , recover-rtti
     , servant-job
     , stm
     , tasty

package.yaml

@@ -505,7 +505,7 @@ executables:
 tests:
    garg-test:
      main: Main.hs
-     source-dirs: src-test
+     source-dirs: test
      default-extensions:
      - DataKinds
      - DeriveGeneric

src/Gargantext/Utils/Jobs/Map.hs

@@ -97,6 +97,7 @@ lookupJob jid (JobMap mvar) = Map.lookup jid <$> readTVarIO mvar
 gcThread :: Ord jid => JobSettings -> JobMap jid w a -> IO ()
 gcThread js (JobMap mvar) = go
   where go = do
+          threadDelay (jsGcPeriod js * 1000000)
           now <- getCurrentTime
           candidateEntries <- Map.filter (expired now) <$> readTVarIO mvar
           forM_ candidateEntries $ \je -> do
@@ -108,7 +109,6 @@ gcThread js (JobMap mvar) = go
             case mrunningjob of
               Nothing -> return ()
               Just a  -> killJ a
-          threadDelay (jsGcPeriod js * 1000000)
           go
 
         expired now jobentry = case jTimeoutAfter jobentry of

stack.yaml

@@ -116,6 +116,7 @@ extra-deps:
 - hgal-2.0.0.2@sha256:13d58afd0668b9cb881c612eff8488a0e289edd4bbffa893df4beee60cfeb73b,653
 - hsparql-0.3.8
 - hstatistics-0.3.1
+- hspec-expectations-0.8.3
 - json-stream-0.4.2.4@sha256:8b7f17d54a6e1e6311756270f8bcf51e91bab4300945400de66118470dcf51b9,4716
 - located-base-0.1.1.1@sha256:7c6395f2b6fbf2d5f76c3514f774423838c0ea94e1c6a5530dd3c94b30c9d1c8,1904
 - logging-effect-1.3.12@sha256:72d168dd09887649ba9501627219b6027cbec2d5541931555b7885b133785ce3,1679
@@ -123,6 +124,7 @@ extra-deps:
 - monoid-extras-0.5.1@sha256:438dbfd7b4dce47d8f0ca577f56caf94bd1e21391afa545cad09fe7cf2e5793d,2333
 - rake-0.0.1@sha256:3380f6567fb17505d1095b7f32222c0b631fa04126ad39726c84262da99c08b3,2025
 - random-1.2.1
+- recover-rtti-0.4.3
 - servant-cassava-0.10.1@sha256:07e7b6ca67cf57dcb4a0041a399a25d058844505837c6479e01d62be59d01fdf,1665
 - servant-ekg-0.3.1@sha256:19bd9dc3943983da8e79d6f607614c68faea4054fb889d508c8a2b67b6bdd448,2203
 - servant-flatten-0.2@sha256:276896f7c5cdec5b8f8493f6205fded0cc602d050b58fdb09a6d7c85c3bb0837,1234

test/Utils.hs

+{-# LANGUAGE ScopedTypeVariables #-}
+module Utils where
+
+import Prelude
+import Control.Exception
+import Test.Tasty.HUnit
+
+-- | Marks the input 'Assertion' as pending, by ignoring any exception
+-- thrown by it.
+pending :: String -> Assertion -> Assertion
+pending reason act = act `catch` (\(e :: SomeException) -> do
+  putStrLn $ "PENDING: " <> reason
+  putStrLn (displayException e))

src-test/Utils/Jobs.hs → test/Utils/Jobs.hs

@@ -17,11 +17,13 @@ import Data.Either
 import Data.List
 import Data.Sequence (Seq, (|>), fromList)
 import Data.Time
+import Debug.RecoverRTTI (anythingToString)
 import Prelude
 import System.IO.Unsafe
 import Network.HTTP.Client.TLS (newTlsManager)
 import Network.HTTP.Client (Manager)
-import Test.Hspec hiding (pending)
+import Test.Hspec
+import Test.Hspec.Expectations.Contrib (annotate)
 import qualified Servant.Job.Types as SJ
 import qualified Servant.Job.Core  as SJ
 
@@ -33,6 +35,7 @@ import Gargantext.Utils.Jobs.State
 import Gargantext.API.Prelude
 import Gargantext.API.Admin.EnvTypes as EnvTypes
 import Gargantext.API.Admin.Orchestrator.Types
+import Control.Concurrent.Async
 
 
 data JobT = A
@@ -53,18 +56,29 @@ addJobToSchedule jobt mvar = do
 data Counts = Counts { countAs :: Int, countBs :: Int }
   deriving (Eq, Show)
 
-jobDuration, initialDelay :: Int
+jobDuration :: Int
 jobDuration = 100000
-initialDelay = 20000
+
+type Timer = TVar Bool
 
 -- | Use in conjuction with 'registerDelay' to create an 'STM' transaction
 -- that will simulate the duration of a job by waiting the timeout registered
 -- by 'registerDelay' before continuing.
-waitJobSTM :: TVar Bool -> STM ()
-waitJobSTM tv = do
+waitTimerSTM :: Timer -> STM ()
+waitTimerSTM tv = do
   v <- readTVar tv
   check v
 
+-- | Samples the running jobs from the first 'TVar' and write them
+-- in the queue.
+sampleRunningJobs :: Timer -> TVar [String] -> TQueue [String]-> STM ()
+sampleRunningJobs timer runningJs samples = do
+  waitTimerSTM timer
+  runningNow <- readTVar runningJs
+  case runningNow of
+    [] -> pure () -- ignore empty runs, when the system is kickstarting.
+    xs -> writeTQueue samples xs
+
 -- | The aim of this test is to ensure that the \"max runners\" setting is
 -- respected, i.e. we have no more than \"N\" jobs running at the same time.
 testMaxRunners :: IO ()
@@ -76,13 +90,27 @@ testMaxRunners = do
   st :: JobsState JobT [String] () <- newJobsState settings defaultPrios
   now <- getCurrentTime
   runningJs   <- newTVarIO []
+  samples     <- newTQueueIO
   remainingJs <- newTVarIO num_jobs
+
+  -- Not the most elegant solution, but in order to test the \"max runners\"
+  -- parameter we start an asynchronous computation that continuously reads the content
+  -- of 'runningJs' and at the end ensures that this value was
+  -- always <= \"max_runners" (but crucially not 0).
+
+  asyncReader <- async $ forever $ do 
+    samplingFrequency <- registerDelay 100_000
+    atomically $ sampleRunningJobs samplingFrequency runningJs samples
+  
   let duration = 1_000_000
       j num _jHandle _inp _l = do
         durationTimer <- registerDelay duration
+        -- NOTE: We do the modification of the 'runningJs' and the rest
+        -- in two transactions on purpose, to give a chance to the async
+        -- sampler to sample the status of the world.
+        atomically $  modifyTVar runningJs (\xs -> ("Job #" ++ show num) : xs)
         atomically $ do
-          modifyTVar runningJs (\xs -> ("Job #" ++ show num) : xs)
-          waitJobSTM durationTimer
+          waitTimerSTM durationTimer
           modifyTVar runningJs (\xs -> filter (/=("Job #" ++ show num)) xs)
           modifyTVar remainingJs pred
       jobs = [ (A, j n) | n <- [1..num_jobs::Int] ]
@@ -94,16 +122,19 @@ testMaxRunners = do
         x <- readTVar remainingJs
         check (x == 0)
 
+  -- Wait for the jobs to finish, then stop the sampler.
   waitFinished
+  cancel asyncReader
 
-  r1 <- readTVarIO runningJs
-  sort r1 `shouldBe` ["Job #1", "Job #2"]
-  threadDelay jobDuration
-  r2 <- readTVarIO runningJs
-  sort r2 `shouldBe` ["Job #3", "Job #4"]
-  threadDelay jobDuration
-  r3 <- readTVarIO runningJs
-  r3 `shouldBe` []
+  -- Check that we got /some/ samples and for each of them,
+  -- let's check only two runners at max were alive.
+  allSamples <- atomically $ flushTQueue samples
+  length allSamples `shouldSatisfy` (> 0)
+
+  forM_ allSamples $ \runLog -> do
+    annotate "predicate to satisfy: (x `isInfixOf` [\"Job #1\", \"Job #2\"] || x `isInfixOf` [\"Job #3\", \"Job #4\"]" $
+      shouldSatisfy (sort runLog)
+                    (\x -> x `isInfixOf` ["Job #1", "Job #2"] || x `isInfixOf` ["Job #3", "Job #4"])
 
 testPrios :: IO ()
 testPrios = do
@@ -136,18 +167,19 @@ testPrios = do
 testExceptions :: IO ()
 testExceptions = do
   k <- genSecret
-  let settings = defaultJobSettings 2 k
+  let settings = defaultJobSettings 1 k
   st :: JobsState JobT [String] () <- newJobsState settings defaultPrios
   jid <- pushJob A ()
     (\_jHandle _inp _log -> readFile "/doesntexist.txt" >>= putStrLn)
     settings st
-  threadDelay initialDelay
+  -- Wait 1 second to make sure the job is finished.
+  threadDelay $ 1_000_000
   mjob <- lookupJob jid (jobsData st)
   case mjob of
-    Nothing ->  error "boo"
+    Nothing ->  fail "lookupJob failed, job not found!"
     Just je ->  case jTask je of
-      DoneJ _ r -> isLeft r `shouldBe` True
-      _         -> error "boo2"
+      DoneJ _ r   -> isLeft r `shouldBe` True
+      unexpected  -> fail $ "Expected job to be done, but got: " <> anythingToString unexpected
   return ()
 
 testFairness :: IO ()
@@ -373,26 +405,20 @@ testMarkProgress = do
                         ]
                         }
 
-pending :: String -> IO () -> IO ()
-pending reason act = act `catch` (\(e :: SomeException) -> do
-  putStrLn $ "PENDING: " <> reason
-  putStrLn (displayException e))
-
-
 test :: Spec
 test = do
   describe "job queue" $ do
     it "respects max runners limit" $
-      pending "Ticket #198" testMaxRunners
+      testMaxRunners
     it "respects priorities" $
       testPrios
     it "can handle exceptions" $
-      pending "Ticket #198" testExceptions
+      testExceptions
     it "fairly picks equal-priority-but-different-kind jobs" $
       testFairness
   describe "job status update and tracking" $ do
     it "can fetch the latest job status" $
-      pending "Ticket #198" testFetchJobStatus
+      testFetchJobStatus
     it "can spin two separate jobs and track their status separately" $
       testFetchJobStatusNoContention
     it "marking stuff behaves as expected" $