Reduce matrix scheduling time and preserve queued work

app/src/App/API.purs

@@ -354,12 +354,6 @@ type PublishResult =
   , matrix :: Maybe PublishedPackage
   }
 
-toPublishResult :: Maybe PublishedPackage -> PublishResult
-toPublishResult matrix =
-  { disposition: if isJust matrix then Published else AlreadyPublished
-  , matrix
-  }
-
 -- | Resolve both compiler and resolutions for a publish operation.
 -- | Will come up with some sort of plan if not provided with a compiler and/or resolutions.
 resolveCompilerAndDeps
@@ -747,7 +741,7 @@ publish payload = do
       }
     _, _ -> do
       Log.info "Verifying the package build plan..."
-      compilerIndex <- MatrixBuilder.readCompilerIndex
+      { compilerIndex } <- MatrixBuilder.readMatrixSnapshot
       resolveCompilerAndDeps compilerIndex (Manifest receivedManifest) payload.compiler payload.resolutions
   Log.info $ "Using compiler " <> Version.print compiler
 
@@ -787,11 +781,12 @@ publish payload = do
       Registry.writeManifest (Manifest receivedManifest)
       Log.notice "Verified the existing package tarball and reconciled its registry metadata and manifest."
 
+    matrix = { compiler, dependencies: receivedManifest.dependencies, version: receivedManifest.version }
     reconciledResult =
       if isNothing existingManifest then
-        Just { compiler, dependencies: receivedManifest.dependencies, version: receivedManifest.version }
+        { disposition: Published, matrix: Just matrix }
       else
-        Nothing
+        { disposition: AlreadyPublished, matrix: Nothing }
 
   case existingPublishedVersion of
     Just _ | Just url <- existingPursuitUrl -> do
@@ -800,7 +795,7 @@ publish payload = do
         , url
         ]
       FS.Extra.remove tmp
-      pure $ toPublishResult Nothing
+      pure { disposition: AlreadyPublished, matrix: Nothing }
 
     -- If metadata already contains this version, first reconcile storage and
     -- both Git repositories, then retry any missing Pursuit documentation.
@@ -813,7 +808,7 @@ publish payload = do
         , ". Please try with a later compiler."
         ]
       FS.Extra.remove tmp
-      pure $ toPublishResult reconciledResult
+      pure reconciledResult
 
     Just info -> do
       reconcileExistingPublication info
@@ -849,7 +844,7 @@ publish payload = do
             Right _ -> do
               FS.Extra.remove tmp
               Log.notice "Successfully uploaded package docs to Pursuit! 🎉 🚀"
-          pure $ toPublishResult reconciledResult
+          pure reconciledResult
 
     -- In this case the package version has not been published, so we proceed
     -- with ordinary publishing.
@@ -1056,7 +1051,7 @@ publish payload = do
           ]
 
       FS.Extra.remove tmp
-      pure $ toPublishResult $ Just { compiler, dependencies: receivedManifest.dependencies, version: receivedManifest.version }
+      pure { disposition: Published, matrix: Just matrix }
 
 validateResolutions :: forall r. Manifest -> Map PackageName Version -> Run (EXCEPT String + r) Unit
 validateResolutions manifest resolutions = do
@@ -1115,7 +1110,7 @@ findAllCompilers
    . { source :: FilePath, manifest :: Manifest, compilers :: NonEmptyArray Version }
   -> Run (REGISTRY_READ + STORAGE + COMPILER_CACHE + LOG + AFF + EFFECT + EXCEPT String + r) FindAllCompilersResult
 findAllCompilers { source, manifest, compilers } = do
-  compilerIndex <- MatrixBuilder.readCompilerIndex
+  { compilerIndex } <- MatrixBuilder.readMatrixSnapshot
   checkedCompilers <- for compilers \target -> do
     Log.debug $ "Trying compiler " <> Version.print target
     case Solver.solveWithCompiler (Range.exact target) compilerIndex (un Manifest manifest).dependencies of

app/src/App/Server/JobExecutor.purs

@@ -10,7 +10,6 @@ import Data.Array as Array
 import Data.DateTime (DateTime)
 import Data.Int as Int
 import Data.Map as Map
-import Data.Set as Set
 import Data.Time.Duration as Duration
 import Effect.Aff as Aff
 import Effect.Ref as Ref
@@ -26,6 +25,7 @@ import Registry.App.Effect.Registry (REGISTRY_READ)
 import Registry.App.Effect.Registry as Registry
 import Registry.App.Server.Env (ServerEffects, ServerEnv, runEffects)
 import Registry.App.Server.Env as Env
+import Registry.App.Server.MatrixBuilder (MatrixSolverResult)
 import Registry.App.Server.MatrixBuilder as MatrixBuilder
 import Registry.ManifestIndex as ManifestIndex
 import Registry.PackageName as PackageName
@@ -157,26 +157,10 @@ executeJob _ = case _ of
       -- At this point this package has been verified with one compiler only.
       -- So we need to enqueue compilation jobs for (1) same package, all the other
       -- compilers, and (2) same compiler, all packages that depend on this one
-      -- TODO here we are building the compiler index, but we should really cache it
-      compilerIndex <- MatrixBuilder.readCompilerIndex
-      let solverData = { compiler, name, version, dependencies, compilerIndex }
-      samePackageAllCompilers <- MatrixBuilder.solveForAllCompilers solverData
-      sameCompilerAllDependants <- MatrixBuilder.solveDependantsForCompiler solverData
-      for (Array.fromFoldable $ Set.union samePackageAllCompilers sameCompilerAllDependants) \{ compiler: solvedCompiler, resolutions, name: solvedPackage, version: solvedVersion } -> do
-        Log.info $ Array.fold
-          [ "Enqueuing matrix job: compiler "
-          , Version.print solvedCompiler
-          , ", package "
-          , PackageName.print solvedPackage
-          , "@"
-          , Version.print solvedVersion
-          ]
-        Db.insertMatrixJob
-          { payload: resolutions
-          , compilerVersion: solvedCompiler
-          , packageName: solvedPackage
-          , packageVersion: solvedVersion
-          }
+      snapshot <- MatrixBuilder.readMatrixSnapshot
+      let solverData = { compiler, name, version, dependencies, snapshot }
+      void $ MatrixBuilder.solveForAllCompilers solverData enqueueMatrixJob
+      void $ MatrixBuilder.solveDependantsForCompiler solverData enqueueMatrixJob
     pure $ Just result.disposition
   UnpublishJob { payload } -> API.authenticated payload $> Nothing
   TransferJob { payload } -> API.authenticated payload $> Nothing
@@ -186,28 +170,18 @@ executeJob _ = case _ of
     -- packages trigger jobs for their dependents so it cascades through the registry)
     dependencies <- MatrixBuilder.runMatrixJob details
 
-    -- TODO here we are building the compiler index, but we should really cache it
-    compilerIndex <- MatrixBuilder.readCompilerIndex
-    let solverData = { compiler: details.compilerVersion, name: packageName, version: packageVersion, dependencies, compilerIndex }
-    sameCompilerAllDependants <- MatrixBuilder.solveDependantsForCompiler solverData
-    for_ (Array.fromFoldable sameCompilerAllDependants) \{ compiler: solvedCompiler, resolutions, name: solvedPackage, version: solvedVersion } -> do
-      Log.info $ Array.fold
-        [ "Enqueuing matrix job: compiler "
-        , Version.print solvedCompiler
-        , ", package "
-        , PackageName.print solvedPackage
-        , "@"
-        , Version.print solvedVersion
-        ]
-      Db.insertMatrixJob
-        { payload: resolutions
-        , compilerVersion: solvedCompiler
-        , packageName: solvedPackage
-        , packageVersion: solvedVersion
-        }
+    snapshot <- MatrixBuilder.readMatrixSnapshot
+    let solverData = { compiler: details.compilerVersion, name: packageName, version: packageVersion, dependencies, snapshot }
+    void $ MatrixBuilder.solveDependantsForCompiler solverData enqueueMatrixJob
     pure Nothing
   PackageSetJob payload -> API.packageSetUpdate payload $> Nothing
 
+enqueueMatrixJob :: forall r. MatrixSolverResult -> Run (DB + LOG + r) Unit
+enqueueMatrixJob { compiler, resolutions, name, version } = do
+  Log.info $ "Enqueuing matrix job: compiler " <> Version.print compiler <> ", package " <> formatPackageVersion name version
+  void $ Db.insertMatrixJob
+    { payload: resolutions, compilerVersion: compiler, packageName: name, packageVersion: version }
+
 upgradeRegistryToNewCompiler :: forall r. Version -> Run (DB + LOG + EXCEPT String + REGISTRY_READ + r) Unit
 upgradeRegistryToNewCompiler newCompilerVersion = do
   Log.info $ "New compiler found: " <> Version.print newCompilerVersion

app/src/App/Server/MatrixBuilder.purs

@@ -1,9 +1,11 @@
 module Registry.App.Server.MatrixBuilder
   ( BuildPlanEntry
+  , MatrixSnapshot
+  , MatrixSolverResult
   , checkIfNewCompiler
   , installBuildPlan
   , printCompilerFailure
-  , readCompilerIndex
+  , readMatrixSnapshot
   , resolutionsToBuildPlan
   , runMatrixJob
   , solveForAllCompilers
@@ -36,6 +38,7 @@ import Registry.App.Effect.Storage (STORAGE)
 import Registry.App.Effect.Storage as Storage
 import Registry.Foreign.FSExtra as FS.Extra
 import Registry.Foreign.Tmp as Tmp
+import Registry.ManifestIndex (ManifestIndex)
 import Registry.ManifestIndex as ManifestIndex
 import Registry.Metadata as Metadata
 import Registry.PackageName as PackageName
@@ -100,14 +103,21 @@ runMatrixJob { compilerVersion, packageName, packageVersion, payload: buildPlan
               Log.error $ "No existing metadata for " <> PackageName.print packageName <> "@" <> Version.print packageVersion
               Except.throw $ "No manifest found for " <> PackageName.print packageName <> "@" <> Version.print packageVersion
 
--- TODO feels like we should be doing this at startup and use the cache instead
--- of reading files all over again
-readCompilerIndex :: forall r. Run (REGISTRY_READ + AFF + EXCEPT String + r) Solver.CompilerIndex
-readCompilerIndex = do
+-- | A pass-local snapshot, captured after publication or compatibility writes.
+-- | The executor is serial; queued jobs cannot change it during scheduling.
+type MatrixSnapshot =
+  { compilerIndex :: Solver.CompilerIndex
+  , manifestIndex :: ManifestIndex
+  , metadata :: Map PackageName Metadata
+  , compilers :: NonEmptyArray Version
+  }
+
+readMatrixSnapshot :: forall r. Run (REGISTRY_READ + AFF + EXCEPT String + r) MatrixSnapshot
+readMatrixSnapshot = do
   metadata <- Registry.readAllMetadata
-  manifests <- Registry.readAllManifests
-  allCompilers <- PursVersions.pursVersions
-  pure $ Solver.buildCompilerIndex allCompilers manifests metadata
+  manifestIndex <- Registry.readAllManifests
+  compilers <- PursVersions.pursVersions
+  pure { compilerIndex: Solver.buildCompilerIndex compilers manifestIndex metadata, manifestIndex, metadata, compilers }
 
 -- | A build plan entry with integrity information for verification.
 type BuildPlanEntry = { version :: Version, hash :: Sha256, bytes :: Number }
@@ -168,7 +178,7 @@ printCompilerFailure compiler = case _ of
     ]
 
 type MatrixSolverData =
-  { compilerIndex :: Solver.CompilerIndex
+  { snapshot :: MatrixSnapshot
   , compiler :: Version
   , name :: PackageName
   , version :: Version
@@ -182,27 +192,32 @@ type MatrixSolverResult =
   , resolutions :: Map PackageName Version
   }
 
-solveForAllCompilers :: forall r. MatrixSolverData -> Run (AFF + EXCEPT String + LOG + r) (Set MatrixSolverResult)
-solveForAllCompilers solverData@{ compiler } = do
+-- | Emit each solved plan before continuing, so cancellation does not discard
+-- | all progress. Callers can persist plans using the existing matrix queue.
+solveForAllCompilers :: forall r. MatrixSolverData -> (MatrixSolverResult -> Run (LOG + r) Unit) -> Run (LOG + r) (Set MatrixSolverResult)
+solveForAllCompilers solverData@{ compiler, snapshot } emit = do
   -- remove the compiler we tested with from the set of all of them
-  compilers <- (Array.filter (_ /= compiler) <<< NonEmptyArray.toArray) <$> PursVersions.pursVersions
-  newJobs <- for compilers \target ->
-    trySolveForCompiler (solverData { compiler = target })
+  let compilers = Array.filter (_ /= compiler) $ NonEmptyArray.toArray snapshot.compilers
+  newJobs <- for compilers \target -> do
+    result <- trySolveForCompiler (solverData { compiler = target })
+    for_ result emit
+    pure result
   pure $ Set.fromFoldable $ Array.catMaybes newJobs
 
-solveDependantsForCompiler :: forall r. MatrixSolverData -> Run (EXCEPT String + LOG + REGISTRY_READ + r) (Set MatrixSolverResult)
-solveDependantsForCompiler { compilerIndex, name, version, compiler } = do
-  manifestIndex <- Registry.readAllManifests
+solveDependantsForCompiler :: forall r. MatrixSolverData -> (MatrixSolverResult -> Run (LOG + r) Unit) -> Run (LOG + r) (Set MatrixSolverResult)
+solveDependantsForCompiler { snapshot, name, version, compiler } emit = do
   let seed = Tuple name version
-  { results, visited } <- go manifestIndex (Set.singleton seed) name version
+  -- Sort this fixed index once, not once per already-compatible dependant.
+  let manifests = ManifestIndex.toSortedArray ManifestIndex.ConsiderRanges snapshot.manifestIndex
+  { results, visited } <- go manifests (Set.singleton seed) name version
   Log.info $ Array.fold
     [ "Cascade from "
     , PackageName.print name
     , "@"
     , Version.print version
     , ": "
     , show (Set.size results)
-    , " enqueued out of "
+    , " solved out of "
     , show (Set.size visited - 1)
     , " dependants visited"
     ]
@@ -219,17 +234,20 @@ solveDependantsForCompiler { compilerIndex, name, version, compiler } = do
   -- never be retriggered.
   -- With this recursive propagation, when C@new completes we cascade through
   -- B (already compiled) and reach A, allowing for a plan to resolve.
-  go manifestIndex visited pkgName pkgVersion = do
-    let dependentManifests = ManifestIndex.dependants manifestIndex pkgName pkgVersion
-    foldM (processManifest manifestIndex) { visited, results: Set.empty } dependentManifests
+  go manifests visited pkgName pkgVersion = do
+    let
+      dependentManifests = Array.filter
+        (\(Manifest manifest) -> maybe false (flip Range.includes pkgVersion) $ Map.lookup pkgName manifest.dependencies)
+        manifests
+    foldM (processManifest manifests) { visited, results: Set.empty } dependentManifests
 
-  processManifest manifestIndex acc (Manifest manifest) = do
+  processManifest manifests acc (Manifest manifest) = do
     let pv = Tuple manifest.name manifest.version
     if Set.member pv acc.visited then
       pure acc
     else do
       let newVisited = Set.insert pv acc.visited
-      Registry.readMetadata manifest.name >>= case _ of
+      case Map.lookup manifest.name snapshot.metadata of
         Nothing -> do
           Log.warn $ "No metadata for dependant " <> PackageName.print manifest.name <> ", skipping"
           pure { visited: newVisited, results: acc.results }
@@ -241,21 +259,23 @@ solveDependantsForCompiler { compilerIndex, name, version, compiler } = do
             Just { compilers }
               | elem compiler compilers -> do
                   -- Already has compiler: propagate through to find stranded packages
-                  sub <- go manifestIndex newVisited manifest.name manifest.version
+                  sub <- go manifests newVisited manifest.name manifest.version
                   pure { visited: sub.visited, results: acc.results <> sub.results }
               | otherwise -> do
-                  result <- trySolveForCompiler { compilerIndex, compiler, name: manifest.name, version: manifest.version, dependencies: manifest.dependencies }
-                  pure case result of
-                    Nothing -> { visited: newVisited, results: acc.results }
-                    Just entry -> { visited: newVisited, results: Set.insert entry acc.results }
+                  result <- trySolveForCompiler { snapshot, compiler, name: manifest.name, version: manifest.version, dependencies: manifest.dependencies }
+                  case result of
+                    Nothing -> pure { visited: newVisited, results: acc.results }
+                    Just entry -> do
+                      emit entry
+                      pure { visited: newVisited, results: Set.insert entry acc.results }
 
 -- | Try to solve a package's dependencies for a specific compiler. Returns
 -- | the solver result if the produced build plan targets the expected compiler,
 -- | Nothing otherwise (solver failure or compiler mismatch).
 trySolveForCompiler :: forall r. MatrixSolverData -> Run (LOG + r) (Maybe MatrixSolverResult)
-trySolveForCompiler { compilerIndex, compiler, name, version, dependencies } = do
+trySolveForCompiler { snapshot, compiler, name, version, dependencies } = do
   Log.debug $ "Trying compiler " <> Version.print compiler <> " for package " <> PackageName.print name
-  case Solver.solveWithCompiler (Range.exact compiler) compilerIndex dependencies of
+  case Solver.solveWithCompiler (Range.exact compiler) snapshot.compilerIndex dependencies of
     Left solverErrors -> do
       Log.info $ "Failed to solve with compiler " <> Version.print compiler <> ": " <> PackageName.print name <> "@" <> Version.print version
       Log.debug $ "Solver errors:\n" <> foldMapWithIndex

app/test/App/API.purs

@@ -207,8 +207,8 @@ spec = do
 
         -- Finally, publishing the same package again is an idempotent success.
         Except.runExcept (API.publish publishArgs) >>= case _ of
-          Right { disposition: V1.AlreadyPublished } -> pure unit
-          Right _ -> Except.throw $ "Expected publishing " <> formatPackageVersion name version <> " twice to report an already-published disposition."
+          Right { disposition: V1.AlreadyPublished, matrix: Nothing } -> pure unit
+          Right _ -> Except.throw $ "Expected publishing " <> formatPackageVersion name version <> " twice to report an already-published disposition without matrix follow-up."
           Left err -> Except.throw $ "Expected an idempotent publish but got error: " <> err
 
       case result of

app/test/App/SQLite.js

@@ -1,11 +1,18 @@
 import Database from "better-sqlite3";
 import { readFileSync } from "node:fs";
 
-export const withDatabase = (test) => () => {
+export const openDatabase = () => {
   const db = new Database(":memory:");
+  db.pragma("foreign_keys = ON");
+  db.exec(readFileSync("db/schema.sql", "utf8"));
+  return db;
+};
+
+export const closeDatabase = db => () => db.close();
+
+export const withDatabase = (test) => () => {
+  const db = openDatabase();
   try {
-    db.pragma("foreign_keys = ON");
-    db.exec(readFileSync("db/schema.sql", "utf8"));
     test(db)();
   } finally {
     db.close();

app/test/App/SQLite.purs

@@ -1,7 +1,8 @@
-module Test.Registry.App.SQLite (spec) where
+module Test.Registry.App.SQLite (spec, withDatabaseAff) where
 
 import Registry.App.Prelude
 
+import Effect.Aff as Aff
 import Effect.Exception as Exception
 import Registry.API.V1 (SortOrder(..))
 import Registry.API.V1 as V1
@@ -14,6 +15,12 @@ import Test.Spec as Spec
 
 foreign import withDatabase :: (SQLite -> Effect Unit) -> Effect Unit
 
+foreign import openDatabase :: Effect SQLite
+foreign import closeDatabase :: SQLite -> Effect Unit
+
+withDatabaseAff :: forall a. (SQLite -> Aff a) -> Aff a
+withDatabaseAff = Aff.bracket (liftEffect openDatabase) (liftEffect <<< closeDatabase)
+
 spec :: Spec.Spec Unit
 spec = Spec.it "allows failed publishes to be retried without losing deduplication or job history" $ liftEffect $ withDatabase \db -> do
   let