Rework daily cron jobs to hit the API instead of working directly on the Registry

app-e2e/src/Test/E2E/Endpoint/Scheduler.purs

@@ -0,0 +1,148 @@
+-- | E2E tests for the Scheduler, covering:
+-- | - scheduleLegacyImports: Detects new package versions via GitHub tags
+-- | - scheduleTransfers: Detects packages that moved to new GitHub locations
+-- | - schedulePackageSetUpdates: Detects recent uploads for package set inclusion
+-- |
+-- | IMPORTANT: These tests must run BEFORE resetTestState is called, since
+-- | the scheduler runs at server startup and creates jobs that would be cleared.
+module Test.E2E.Endpoint.Scheduler (spec) where
+
+import Registry.App.Prelude
+
+import Data.Array as Array
+import Data.Map as Map
+import Registry.API.V1 (Job(..))
+import Registry.Location (Location(..))
+import Registry.Operation (AuthenticatedPackageOperation(..))
+import Registry.Operation as Operation
+import Registry.PackageName as PackageName
+import Registry.Test.Assert as Assert
+import Registry.Version as Version
+import Test.E2E.Support.Client as Client
+import Test.E2E.Support.Env (E2ESpec)
+import Test.Spec as Spec
+
+spec :: E2ESpec
+spec = do
+  Spec.describe "scheduleLegacyImports" do
+    Spec.it "enqueues publish jobs for new package versions discovered via GitHub tags" do
+      -- The scheduler runs at server startup and should have already
+      -- fetched tags for packages in the registry metadata.
+      -- prelude has v6.0.1 published but v6.0.2 in tags (per wiremock config)
+      jobs <- Client.getJobs
+
+      -- Find publish jobs for prelude
+      let
+        isPreludePublishJob :: Job -> Boolean
+        isPreludePublishJob = case _ of
+          PublishJob { packageName, packageVersion } ->
+            packageName == unsafeFromRight (PackageName.parse "prelude")
+              && packageVersion
+              == unsafeFromRight (Version.parse "6.0.2")
+          _ -> false
+
+        preludeJob = Array.find isPreludePublishJob jobs
+
+      case preludeJob of
+        Just (PublishJob { payload }) -> do
+          -- Verify the compiler is from previous version (prelude@6.0.1 has compilers [0.15.10, 0.15.11])
+          -- The scheduler should use the lowest compiler from the previous version for compatibility
+          let expectedCompiler = unsafeFromRight (Version.parse "0.15.10")
+          when (payload.compiler /= expectedCompiler) do
+            Assert.fail $ "Expected compiler 0.15.10 but got " <> Version.print payload.compiler
+        Just _ -> Assert.fail "Expected PublishJob but got different job type"
+        Nothing -> do
+          -- Log what jobs we did find for debugging
+          let publishJobs = Array.filter isPublishJob jobs
+          Assert.fail $ "Expected to find a publish job for prelude@6.0.2 but found "
+            <> show (Array.length publishJobs)
+            <> " publish jobs: "
+            <> show (map formatPublishJob publishJobs)
+
+    Spec.it "does not enqueue jobs for already-published versions" do
+      jobs <- Client.getJobs
+
+      -- prelude v6.0.1 is already published, should NOT have a new job
+      let
+        isDuplicateJob :: Job -> Boolean
+        isDuplicateJob = case _ of
+          PublishJob { packageName, packageVersion } ->
+            packageName == unsafeFromRight (PackageName.parse "prelude")
+              && packageVersion
+              == unsafeFromRight (Version.parse "6.0.1")
+          _ -> false
+
+        duplicateJob = Array.find isDuplicateJob jobs
+
+      case duplicateJob of
+        Nothing -> pure unit -- Good, no duplicate job
+        Just _ -> Assert.fail "Found unexpected publish job for already-published prelude@6.0.1"
+
+  Spec.describe "scheduleTransfers" do
+    Spec.it "enqueues transfer jobs when package location changes" do
+      jobs <- Client.getJobs
+      let
+        isTransferredJob :: Job -> Boolean
+        isTransferredJob = case _ of
+          TransferJob { packageName } ->
+            packageName == unsafeFromRight (PackageName.parse "transferred")
+          _ -> false
+      case Array.find isTransferredJob jobs of
+        Just (TransferJob { packageName, payload }) -> do
+          -- Verify packageName
+          when (packageName /= unsafeFromRight (PackageName.parse "transferred")) do
+            Assert.fail $ "Wrong package name: " <> PackageName.print packageName
+          -- Verify newLocation in payload
+          case payload.payload of
+            Transfer { newLocation } ->
+              case newLocation of
+                GitHub { owner } ->
+                  when (owner /= "new-owner") do
+                    Assert.fail $ "Expected owner 'new-owner' but got '" <> owner <> "'"
+                _ -> Assert.fail "Expected GitHub location"
+            _ -> Assert.fail "Expected Transfer payload"
+        Just _ -> Assert.fail "Expected TransferJob but got different job type"
+        Nothing -> do
+          let transferJobs = Array.filter isTransferJob jobs
+          Assert.fail $ "Expected to find a transfer job for 'transferred' but found "
+            <> show (Array.length transferJobs)
+            <> " transfer jobs"
+
+  Spec.describe "schedulePackageSetUpdates" do
+    Spec.it "enqueues package set update for recent uploads not in set" do
+      jobs <- Client.getJobs
+      let packageSetJobs = Array.filter isPackageSetJob jobs
+      case Array.head packageSetJobs of
+        Just (PackageSetJob { payload }) ->
+          case payload of
+            Operation.PackageSetUpdate { packages } ->
+              case Map.lookup (unsafeFromRight $ PackageName.parse "type-equality") packages of
+                Just (Just _) -> pure unit
+                _ -> Assert.fail "Expected type-equality in package set update"
+        Just _ -> Assert.fail "Expected PackageSetJob but got different job type"
+        Nothing -> Assert.fail "Expected package set job to be enqueued"
+
+-- | Check if a job is a PublishJob
+isPublishJob :: Job -> Boolean
+isPublishJob = case _ of
+  PublishJob _ -> true
+  _ -> false
+
+-- | Format a PublishJob for debugging output
+formatPublishJob :: Job -> String
+formatPublishJob = case _ of
+  PublishJob { packageName, packageVersion } ->
+    PackageName.print packageName <> "@" <> Version.print packageVersion
+  _ -> "<not a publish job>"
+
+-- | Check if a job is a TransferJob
+isTransferJob :: Job -> Boolean
+isTransferJob = case _ of
+  TransferJob _ -> true
+  _ -> false
+
+-- | Check if a job is a PackageSetJob
+isPackageSetJob :: Job -> Boolean
+isPackageSetJob = case _ of
+  PackageSetJob _ -> true
+  _ -> false

app-e2e/src/Test/Main.purs

@@ -6,6 +6,7 @@ import Data.Time.Duration (Milliseconds(..))
 import Test.E2E.Endpoint.Jobs as Jobs
 import Test.E2E.Endpoint.PackageSets as PackageSets
 import Test.E2E.Endpoint.Publish as Publish
+import Test.E2E.Endpoint.Scheduler as Scheduler
 import Test.E2E.Endpoint.Transfer as Transfer
 import Test.E2E.Endpoint.Unpublish as Unpublish
 import Test.E2E.GitHubIssue as GitHubIssue
@@ -21,6 +22,10 @@ main :: Effect Unit
 main = do
   env <- mkTestEnv
   runSpecAndExitProcess' config [ consoleReporter ] $ hoistE2E env do
+    -- The scheduler runs at startup and enqueues a bunch of jobs in the DB,
+    -- so we need to run these tests without cleaning out the state first
+    Spec.describe "Scheduler" Scheduler.spec
+
     Spec.before_ resetTestState $ Spec.after_ assertReposClean $ Spec.describe "E2E Tests" do
       Spec.describe "Endpoints" do
         Spec.describe "Publish" Publish.spec

app/fixtures/registry-index/tr/an/transferred

@@ -0,0 +1 @@
+{"name":"transferred","version":"1.0.0","license":"MIT","location":{"githubOwner":"old-owner","githubRepo":"purescript-transferred"},"ref":"v1.0.0","dependencies":{}}

app/fixtures/registry/metadata/transferred.json

@@ -0,0 +1,16 @@
+{
+  "location": {
+    "githubOwner": "old-owner",
+    "githubRepo": "purescript-transferred"
+  },
+  "published": {
+    "1.0.0": {
+      "bytes": 1000,
+      "compilers": ["0.15.10"],
+      "hash": "sha256-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+      "publishedTime": "2022-01-01T00:00:00.000Z",
+      "ref": "v1.0.0"
+    }
+  },
+  "unpublished": {}
+}

app/src/App/API.purs

@@ -195,11 +195,18 @@ packageSetUpdate details = do
 
   let changeSet = candidates.accepted <#> maybe Remove Update
   Log.notice "Attempting to build package set update."
-  PackageSets.upgradeAtomic latestPackageSet (fromMaybe prevCompiler payload.compiler) changeSet >>= case _ of
-    Left error ->
-      Except.throw $ "The package set produced from this suggested update does not compile:\n\n" <> error
-    Right packageSet -> do
-      let commitMessage = PackageSets.commitMessage latestPackageSet changeSet (un PackageSet packageSet).version
+  PackageSets.upgradeSequential latestPackageSet (fromMaybe prevCompiler payload.compiler) changeSet >>= case _ of
+    Nothing ->
+      Except.throw "No packages could be added to the package set. All packages failed to compile."
+    Just { failed, succeeded, result: packageSet } -> do
+      unless (Map.isEmpty failed) do
+        let
+          formatFailed = String.joinWith "\n" $ Array.catMaybes $ flip map (Map.toUnfoldable failed) \(Tuple name change) ->
+            case change of
+              PackageSets.Update version -> Just $ "  - " <> formatPackageVersion name version
+              PackageSets.Remove -> Nothing
+        Log.warn $ "Some packages could not be added to the set:\n" <> formatFailed
+      let commitMessage = PackageSets.commitMessage latestPackageSet succeeded (un PackageSet packageSet).version
       Registry.writePackageSet packageSet commitMessage
       Log.notice "Built and released a new package set! Now mirroring to the package-sets repo..."
       Registry.mirrorPackageSet packageSet

app/src/App/Main.purs

@@ -8,22 +8,33 @@ import Effect.Aff as Aff
 import Effect.Class.Console as Console
 import Fetch.Retry as Fetch.Retry
 import Node.Process as Process
-import Registry.App.Server.Env (ServerEnv, createServerEnv)
+import Registry.App.Effect.Log as Log
+import Registry.App.Effect.Registry as Registry
+import Registry.App.Server.Env (createServerEnv, runEffects)
 import Registry.App.Server.JobExecutor as JobExecutor
 import Registry.App.Server.Router as Router
+import Registry.App.Server.Scheduler as Scheduler
 
 main :: Effect Unit
-main = do
-  createServerEnv # Aff.runAff_ case _ of
-    Left error -> do
+main = Aff.launchAff_ do
+  Aff.attempt createServerEnv >>= case _ of
+    Left error -> liftEffect do
       Console.log $ "Failed to start server: " <> Aff.message error
       Process.exit' 1
     Right env -> do
-      case env.vars.resourceEnv.healthchecksUrl of
-        Nothing -> Console.log "HEALTHCHECKS_URL not set, healthcheck pinging disabled"
-        Just healthchecksUrl -> Aff.launchAff_ $ healthcheck healthchecksUrl
-      Aff.launchAff_ $ jobExecutor env
-      Router.runRouter env
+      -- Initialize registry repo before launching parallel processes, to avoid
+      -- race condition where both Scheduler and Job Executor try to clone the
+      -- Registry at the same time
+      void $ runEffects env do
+        Log.info "Initializing registry repo..."
+        Registry.readAllMetadata
+      liftEffect do
+        case env.vars.resourceEnv.healthchecksUrl of
+          Nothing -> Console.log "HEALTHCHECKS_URL not set, healthcheck pinging disabled"
+          Just healthchecksUrl -> Aff.launchAff_ $ healthcheck healthchecksUrl
+        Aff.launchAff_ $ withRetryLoop "Scheduler" $ Scheduler.runScheduler env
+        Aff.launchAff_ $ withRetryLoop "Job executor" $ JobExecutor.runJobExecutor env
+        Router.runRouter env
   where
   healthcheck :: String -> Aff Unit
   healthcheck healthchecksUrl = loop limit
@@ -63,20 +74,22 @@ main = do
         Succeeded _ -> do
           Console.error "Healthchecks returned non-200 status and failure limit reached, will not retry."
 
-  jobExecutor :: ServerEnv -> Aff Unit
-  jobExecutor env = do
-    loop initialRestartDelay
+  -- | Run an Aff action in a loop with exponential backoff on failure.
+  -- | If the action runs for longer than 60 seconds before failing,
+  -- | the restart delay resets to the initial value (heuristic for stability).
+  withRetryLoop :: String -> Aff (Either Aff.Error Unit) -> Aff Unit
+  withRetryLoop name action = loop initialRestartDelay
     where
     initialRestartDelay = Milliseconds 100.0
 
     loop restartDelay = do
       start <- nowUTC
-      result <- JobExecutor.runJobExecutor env
+      result <- action
       end <- nowUTC
 
       Console.error case result of
-        Left error -> "Job executor failed: " <> Aff.message error
-        Right _ -> "Job executor exited for no reason."
+        Left error -> name <> " failed: " <> Aff.message error
+        Right _ -> name <> " exited for no reason."
 
       -- This is a heuristic: if the executor keeps crashing immediately, we
       -- restart with an exponentially increasing delay, but once the executor