TheAnachronism/docspell
1
0
Fork 0
mirror of https://github.com/TheAnachronism/docspell.git synced 2025-04-04 18:39:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve job-queue query to make sure jobs across all states show up

Browse Source
This commit is contained in:
Eike Kettner 2020-08-13 01:01:02 +02:00
parent 41ea071555
commit 69674eb485
3 changed files with 29 additions and 10 deletions
modules
backend/src/main/scala/docspell/backend/ops
OJob.scala
restserver/src/main/scala/docspell/restserver/routes
JobQueueRoutes.scala
store/src/main/scala/docspell/store/queries
QJob.scala

4
modules/backend/src/main/scala/docspell/backend/ops/OJob.scala

@ -48,7 +48,9 @@ object OJob {
def queueState(collective: Ident, maxResults: Int): F[CollectiveQueueState] =
store
.transact(QJob.queueStateSnapshot(collective).take(maxResults.toLong))
.transact(
QJob.queueStateSnapshot(collective, maxResults.toLong)
)
.map(t => JobDetail(t._1, t._2))
.compile
.toVector

2
modules/restserver/src/main/scala/docspell/restserver/routes/JobQueueRoutes.scala

@ -21,7 +21,7 @@ object JobQueueRoutes {
HttpRoutes.of {
case GET -> Root / "state" =>
for {
js <- backend.job.queueState(user.account.collective, 200)
js <- backend.job.queueState(user.account.collective, 40)
res = Conversions.mkJobQueueState(js)
resp <- Ok(res)
} yield resp

33
modules/store/src/main/scala/docspell/store/queries/QJob.scala

@ -209,7 +209,8 @@ object QJob {
store.transact(RJob.findFromIds(ids))
def queueStateSnapshot(
collective: Ident
collective: Ident,
max: Long
): Stream[ConnectionIO, (RJob, Vector[RJobLog])] = {
val JC = RJob.Columns
val waiting: Set[JobState] = Set(JobState.Waiting, JobState.Stuck, JobState.Scheduled)
@ -218,18 +219,34 @@ object QJob {
def selectJobs(now: Timestamp): Stream[ConnectionIO, RJob] = {
val refDate = now.minusHours(24)
val sql = selectSimple(
val runningJobs = (selectSimple(
JC.all,
RJob.table,
and(JC.group.is(collective), JC.state.isOneOf(running.toSeq))
) ++ orderBy(JC.submitted.desc)).query[RJob].stream
val waitingJobs = (selectSimple(
JC.all,
RJob.table,
and(
JC.group.is(collective),
or(
and(JC.state.isOneOf(done.toSeq), JC.submitted.isGt(refDate)),
JC.state.isOneOf((running ++ waiting).toSeq)
)
JC.state.isOneOf(waiting.toSeq),
JC.submitted.isGt(refDate)
)
)
(sql ++ orderBy(JC.submitted.desc)).query[RJob].stream
) ++ orderBy(JC.submitted.desc)).query[RJob].stream.take(max)
val doneJobs = (selectSimple(
JC.all,
RJob.table,
and(
JC.group.is(collective),
JC.state.isOneOf(done.toSeq),
JC.submitted.isGt(refDate)
)
) ++ orderBy(JC.submitted.desc)).query[RJob].stream.take(max)
runningJobs ++ waitingJobs ++ doneJobs
}
def selectLogs(job: RJob): ConnectionIO[Vector[RJobLog]] =