@@ -26,15 +26,21 @@ class MavenCentralService(
2626 extends LazyLogging :
2727 private val system = summon[ActorSystem ]
2828
29+ private val groupIdPageSize = 50
30+ private val artifactRefPageSize = 1000
31+ private val artifactIdPageSize = 20
32+ private val pageDelay = 500 .millis
33+ private val publishDelay = 100 .millis
34+
2935 def findNonStandard (): Future [String ] =
3036 val nonStandardLibs = NonStandardLib .load(dataPaths)
3137 for result <- nonStandardLibs.mapSync { lib =>
3238 val groupId = Artifact .GroupId (lib.groupId)
3339 // get should not throw: it is a fixed set of artifactIds
3440 val artifactId = Artifact .ArtifactId (lib.artifactId)
3541 for
36- knownRefs <- database.getArtifactRefs (groupId)
37- inserted <- findAndIndexMissingArtifacts(groupId, artifactId, knownRefs.toSet )
42+ knownRefs <- loadKnownRefs (groupId)
43+ inserted <- findAndIndexMissingArtifacts(groupId, artifactId, knownRefs)
3844 yield inserted
3945 }
4046 yield s " Inserted ${result.sum} missing poms "
@@ -53,9 +59,8 @@ class MavenCentralService(
5359 missingPomFiles <- missingVersions.mapSync(ref => mavenCentralClient.getPomFile(ref).map(_.map(ref -> _)))
5460 publishResult <- missingPomFiles.flatten.mapSync {
5561 case (mavenRef, (pomFile, creationDate)) =>
56- // Add a small delay between publishes to avoid overwhelming the database connection pool
5762 for
58- _ <- delayBetweenPublishes( )
63+ _ <- delay(publishDelay )
5964 result <- publishProcess.publishPom(mavenRef.toString(), pomFile, creationDate, None )
6065 yield result
6166 }
@@ -64,36 +69,69 @@ class MavenCentralService(
6469 case _ => false
6570 }
6671
67- private def delayBetweenPublishes (): Future [Unit ] =
68- // Small delay between publishes to avoid overwhelming the database connection pool
69- after(100 .millis, system.scheduler)(Future .successful(()))
70-
7172 def findMissing (): Future [String ] =
72- for
73- // Load group IDs only, then known refs per group — avoid loading the entire artifacts table
74- groupIds <- database.getGroupIds().map(_.sorted)
75- // we sort just to estimate through the logs the percentage of progress
76- result <- groupIds.mapSync(findAndIndexMissingArtifacts(_, None ))
77- yield s " Inserted ${result.sum} missing poms "
73+ def loop (page : Int , totalInserted : Int ): Future [Int ] =
74+ for
75+ batch <- database.getGroupIds(limit = groupIdPageSize, offset = page * groupIdPageSize)
76+ _ = logger.info(s " Processing group ID page $page ( ${batch.size} groups) " )
77+ inserted <- batch.mapSync(g => findAndIndexMissingArtifacts(g, None )).map(_.sum)
78+ total = totalInserted + inserted
79+ result <-
80+ if batch.size == groupIdPageSize then delay(pageDelay).flatMap(_ => loop(page + 1 , total))
81+ else Future .successful(total)
82+ yield result
83+
84+ loop(0 , 0 ).map(n => s " Inserted $n missing poms " )
7885
7986 private def findAndIndexMissingArtifacts (
8087 groupId : GroupId ,
8188 artifactNameOpt : Option [Artifact .Name ]
8289 ): Future [Int ] =
8390 for
84- knownRefs <- database.getArtifactRefs (groupId).map(_.toSet )
91+ knownRefs <- loadKnownRefs (groupId)
8592 artifactIds <- mavenCentralClient.getAllArtifactIds(groupId)
8693 scalaArtifactIds = artifactIds.filter(artifact =>
8794 artifactNameOpt.forall(_ == artifact.name) && artifact.isScala && artifact.binaryVersion.isValid
8895 )
89- result <- scalaArtifactIds
90- .mapSync(id => findAndIndexMissingArtifacts(groupId, id, knownRefs))
91- yield result.sum
96+ result <- processPages(scalaArtifactIds, artifactIdPageSize) { batch =>
97+ batch.mapSync(id => findAndIndexMissingArtifacts(groupId, id, knownRefs)).map(_.sum)
98+ }
99+ yield result
92100
93101 def syncOne (groupId : GroupId , artifactNameOpt : Option [Artifact .Name ]): Future [String ] =
94102 for result <- findAndIndexMissingArtifacts(groupId, artifactNameOpt)
95103 yield s " Inserted $result poms "
96104
105+ /** Load known refs for a group in pages to keep each DB query small. */
106+ private def loadKnownRefs (groupId : GroupId ): Future [Set [Artifact .Reference ]] =
107+ def loop (page : Int , acc : Set [Artifact .Reference ]): Future [Set [Artifact .Reference ]] =
108+ for
109+ batch <- database.getArtifactRefs(groupId, limit = artifactRefPageSize, offset = page * artifactRefPageSize)
110+ next = acc ++ batch
111+ result <-
112+ if batch.size == artifactRefPageSize then loop(page + 1 , next)
113+ else Future .successful(next)
114+ yield result
115+ loop(0 , Set .empty)
116+
117+ /** Process items in pages, with a short delay between full pages. */
118+ private def processPages [A ](items : Seq [A ], pageSize : Int )(process : Seq [A ] => Future [Int ]): Future [Int ] =
119+ def loop (page : Int , total : Int ): Future [Int ] =
120+ val batch = items.slice(page * pageSize, (page + 1 ) * pageSize)
121+ if batch.isEmpty then Future .successful(total)
122+ else
123+ for
124+ inserted <- process(batch)
125+ next = total + inserted
126+ result <-
127+ if batch.size == pageSize then delay(pageDelay).flatMap(_ => loop(page + 1 , next))
128+ else Future .successful(next)
129+ yield result
130+ loop(0 , 0 )
131+
132+ private def delay (duration : FiniteDuration ): Future [Unit ] =
133+ after(duration, system.scheduler)(Future .successful(()))
134+
97135 def republishArtifacts (): Future [String ] =
98136 for
99137 projectStatuses <- database.getAllProjectsStatuses()
0 commit comments