Move constraint filtering down into JobStorage to improve perf.

app/src/androidTest/java/org/thoughtcrime/securesms/jobs/JobManagerPerformanceTests.kt

@@ -32,20 +32,20 @@ class JobManagerPerformanceTests {
 
   @Test
   fun testPerformance_singleQueue() {
-    runTest(2000) { TestJob(queue = "queue") }
+    runTest("singleQueue", 2000) { TestJob(queue = "queue") }
   }
 
   @Test
   fun testPerformance_fourQueues() {
-    runTest(2000) { TestJob(queue = "queue-${Random.nextInt(1, 5)}") }
+    runTest("fourQueues", 2000) { TestJob(queue = "queue-${Random.nextInt(1, 5)}") }
   }
 
   @Test
   fun testPerformance_noQueues() {
-    runTest(2000) { TestJob(queue = null) }
+    runTest("noQueues", 2000) { TestJob(queue = null) }
   }
 
-  private fun runTest(count: Int, jobCreator: () -> TestJob) {
+  private fun runTest(name: String, count: Int, jobCreator: () -> TestJob) {
     val context = AppDependencies.application
     val jobManager = testJobManager(context)
 
@@ -66,17 +66,17 @@ class JobManagerPerformanceTests {
         eventTimer.emit("job")
         latch.countDown()
         if (latch.count % 100 == 0L) {
-          Log.d(TAG, "Finished ${count - latch.count}/$count jobs")
+          Log.d(TAG, "[$name] Finished ${count - latch.count}/$count jobs")
         }
       }
     }
 
-    Log.i(TAG, "Adding jobs...")
+    Log.i(TAG, "[$name] Adding jobs...")
     jobManager.addAll((1..count).map { jobCreator() })
 
-    Log.i(TAG, "Waiting for jobs to complete...")
+    Log.i(TAG, "[$name] Waiting for jobs to complete...")
     latch.await()
-    Log.i(TAG, "Jobs complete!")
+    Log.i(TAG, "[$name] Jobs complete!")
     Log.i(TAG, eventTimer.stop().summary)
   }
 

app/src/main/java/org/thoughtcrime/securesms/database/JobDatabase.kt

@@ -197,28 +197,6 @@ class JobDatabase(
       .readToList { it.toJobSpec() }
   }
 
-  @Synchronized
-  fun getJobSpecsByKeys(keys: Collection<String>): List<JobSpec> {
-    if (keys.isEmpty()) {
-      return emptyList()
-    }
-
-    val output: MutableList<JobSpec> = ArrayList(keys.size)
-
-    for (query in SqlUtil.buildCollectionQuery(Jobs.JOB_SPEC_ID, keys)) {
-      readableDatabase
-        .select()
-        .from(Jobs.TABLE_NAME)
-        .where(query.where, query.whereArgs)
-        .run()
-        .forEach {
-          output += it.toJobSpec()
-        }
-    }
-
-    return output
-  }
-
   @Synchronized
   fun getMostEligibleJobInQueue(queue: String): JobSpec? {
     return readableDatabase

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobController.java

@@ -15,6 +15,7 @@ import org.thoughtcrime.securesms.jobmanager.persistence.DependencySpec;
 import org.thoughtcrime.securesms.jobmanager.persistence.FullSpec;
 import org.thoughtcrime.securesms.jobmanager.persistence.JobSpec;
 import org.thoughtcrime.securesms.jobmanager.persistence.JobStorage;
+import org.thoughtcrime.securesms.jobs.MinimalJobSpec;
 import org.thoughtcrime.securesms.util.Debouncer;
 
 import java.util.ArrayList;
@@ -319,7 +320,7 @@ class JobController {
    * When the job returned from this method has been run, you must call {@link #onJobFinished(Job)}.
    */
   @WorkerThread
-  synchronized @NonNull Job pullNextEligibleJobForExecution(@NonNull JobPredicate predicate) {
+  synchronized @NonNull Job pullNextEligibleJobForExecution(@NonNull Predicate<MinimalJobSpec> predicate) {
     try {
       Job job;
 
@@ -479,24 +480,27 @@ class JobController {
   }
 
   @WorkerThread
-  private @Nullable Job getNextEligibleJobForExecution(@NonNull JobPredicate predicate) {
+  private @Nullable Job getNextEligibleJobForExecution(@NonNull Predicate<MinimalJobSpec> predicate) {
-    List<JobSpec> jobSpecs = Stream.of(jobStorage.getPendingJobsWithNoDependenciesInCreatedOrder(System.currentTimeMillis()))
+    JobSpec jobSpec = jobStorage.getNextEligibleJob(System.currentTimeMillis(), minimalJobSpec -> {
-                                   .filter(predicate::shouldRun)
+      if (!predicate.test(minimalJobSpec)) {
-                                   .toList();
+        return false;
+      }
 
-    for (JobSpec jobSpec : jobSpecs) {
+      List<ConstraintSpec> constraintSpecs = jobStorage.getConstraintSpecs(minimalJobSpec.getId());
-      List<ConstraintSpec> constraintSpecs = jobStorage.getConstraintSpecs(jobSpec.getId());
       List<Constraint>     constraints     = Stream.of(constraintSpecs)
                                                    .map(ConstraintSpec::getFactoryKey)
                                                    .map(constraintInstantiator::instantiate)
                                                    .toList();
 
-      if (Stream.of(constraints).allMatch(Constraint::isMet)) {
+      return Stream.of(constraints).allMatch(Constraint::isMet);
-        return createJob(jobSpec, constraintSpecs);
+    });
-      }
+
+    if (jobSpec == null) {
+      return null;
     }
 
-    return null;
+    List<ConstraintSpec> constraintSpecs = jobStorage.getConstraintSpecs(jobSpec.getId());
+    return createJob(jobSpec, constraintSpecs);
   }
 
   private @NonNull Job createJob(@NonNull JobSpec jobSpec, @NonNull List<ConstraintSpec> constraintSpecs) {

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobManager.java

@@ -15,6 +15,7 @@ import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.jobmanager.impl.DefaultExecutorFactory;
 import org.thoughtcrime.securesms.jobmanager.persistence.JobSpec;
 import org.thoughtcrime.securesms.jobmanager.persistence.JobStorage;
+import org.thoughtcrime.securesms.jobs.MinimalJobSpec;
 import org.thoughtcrime.securesms.util.Debouncer;
 import org.thoughtcrime.securesms.util.TextSecurePreferences;
 import org.thoughtcrime.securesms.util.Util;
@@ -46,6 +47,8 @@ public class JobManager implements ConstraintObserver.Notifier {
 
   public static final int CURRENT_VERSION = 12;
 
+  private static final Predicate<MinimalJobSpec> NO_PREDICATE = spec -> true;
+
   private final Application   application;
   private final Configuration configuration;
   private final Executor      executor;
@@ -109,10 +112,10 @@ public class JobManager implements ConstraintObserver.Notifier {
       int id = 0;
 
       for (int i = 0; i < configuration.getJobThreadCount(); i++) {
-        new JobRunner(application, ++id, jobController, JobPredicate.NONE).start();
+        new JobRunner(application, ++id, jobController, NO_PREDICATE).start();
       }
 
-      for (JobPredicate predicate : configuration.getReservedJobRunners()) {
+      for (Predicate<MinimalJobSpec> predicate : configuration.getReservedJobRunners()) {
         new JobRunner(application, ++id, jobController, predicate).start();
       }
 
@@ -586,7 +589,7 @@ public class JobManager implements ConstraintObserver.Notifier {
     private final JobStorage                      jobStorage;
     private final JobMigrator                     jobMigrator;
     private final JobTracker                      jobTracker;
-    private final List<JobPredicate>       reservedJobRunners;
+    private final List<Predicate<MinimalJobSpec>> reservedJobRunners;
 
     private Configuration(int jobThreadCount,
                           @NonNull ExecutorFactory executorFactory,
@@ -596,7 +599,7 @@ public class JobManager implements ConstraintObserver.Notifier {
                           @NonNull JobStorage jobStorage,
                           @NonNull JobMigrator jobMigrator,
                           @NonNull JobTracker jobTracker,
-                          @NonNull List<JobPredicate> reservedJobRunners)
+                          @NonNull List<Predicate<MinimalJobSpec>> reservedJobRunners)
     {
       this.executorFactory        = executorFactory;
       this.jobThreadCount         = jobThreadCount;
@@ -642,7 +645,7 @@ public class JobManager implements ConstraintObserver.Notifier {
       return jobTracker;
     }
 
-    @NonNull List<JobPredicate> getReservedJobRunners() {
+    @NonNull List<Predicate<MinimalJobSpec>> getReservedJobRunners() {
       return reservedJobRunners;
     }
 
@@ -656,14 +659,14 @@ public class JobManager implements ConstraintObserver.Notifier {
       private JobStorage                      jobStorage          = null;
       private JobMigrator                     jobMigrator         = null;
       private JobTracker                      jobTracker          = new JobTracker();
-      private List<JobPredicate>              reservedJobRunners  = new ArrayList<>();
+      private List<Predicate<MinimalJobSpec>> reservedJobRunners  = new ArrayList<>();
 
       public @NonNull Builder setJobThreadCount(int jobThreadCount) {
         this.jobThreadCount = jobThreadCount;
         return this;
       }
 
-      public @NonNull Builder addReservedJobRunner(@NonNull JobPredicate predicate) {
+      public @NonNull Builder addReservedJobRunner(@NonNull Predicate<MinimalJobSpec> predicate) {
         this.reservedJobRunners.add(predicate);
         return this;
       }

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobPredicate.java

@@ -1,11 +0,0 @@
-package org.thoughtcrime.securesms.jobmanager;
-
-import androidx.annotation.NonNull;
-
-import org.thoughtcrime.securesms.jobmanager.persistence.JobSpec;
-
-public interface JobPredicate {
-  JobPredicate NONE = jobSpec -> true;
-
-  boolean shouldRun(@NonNull JobSpec jobSpec);
-}

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobRunner.java

@@ -8,10 +8,12 @@ import androidx.annotation.NonNull;
 import com.annimon.stream.Stream;
 
 import org.signal.core.util.logging.Log;
+import org.thoughtcrime.securesms.jobs.MinimalJobSpec;
 import org.thoughtcrime.securesms.util.WakeLockUtil;
 
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 
 /**
  * A thread that constantly checks for available {@link Job}s owned by the {@link JobController}.
@@ -30,9 +32,9 @@ class JobRunner extends Thread {
   private final Application               application;
   private final int                       id;
   private final JobController             jobController;
-  private final JobPredicate  jobPredicate;
+  private final Predicate<MinimalJobSpec> jobPredicate;
 
-  JobRunner(@NonNull Application application, int id, @NonNull JobController jobController, @NonNull JobPredicate predicate) {
+  JobRunner(@NonNull Application application, int id, @NonNull JobController jobController, @NonNull Predicate<MinimalJobSpec> predicate) {
     super("signal-JobRunner-" + id);
 
     this.application   = application;

app/src/main/java/org/thoughtcrime/securesms/jobmanager/impl/FactoryJobPredicate.java

@@ -1,18 +1,16 @@
 package org.thoughtcrime.securesms.jobmanager.impl;
 
-import androidx.annotation.NonNull;
+import org.thoughtcrime.securesms.jobs.MinimalJobSpec;
-
-import org.thoughtcrime.securesms.jobmanager.JobPredicate;
-import org.thoughtcrime.securesms.jobmanager.persistence.JobSpec;
 
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.function.Predicate;
 
 /**
- * A {@link JobPredicate} that will only run jobs with the provided factory keys.
+ * A {@link Predicate} that will only run jobs with the provided factory keys.
  */
-public final class FactoryJobPredicate implements JobPredicate {
+public final class FactoryJobPredicate implements Predicate<MinimalJobSpec> {
 
   private final Set<String> factories;
 
@@ -21,7 +19,7 @@ public final class FactoryJobPredicate implements JobPredicate {
   }
 
   @Override
-  public boolean shouldRun(@NonNull JobSpec jobSpec) {
+  public boolean test(MinimalJobSpec minimalJobSpec) {
-    return factories.contains(jobSpec.getFactoryKey());
+    return factories.contains(minimalJobSpec.getFactoryKey());
   }
 }

app/src/main/java/org/thoughtcrime/securesms/jobmanager/persistence/JobStorage.kt

@@ -1,6 +1,7 @@
 package org.thoughtcrime.securesms.jobmanager.persistence
 
 import androidx.annotation.WorkerThread
+import org.thoughtcrime.securesms.jobs.MinimalJobSpec
 import java.util.function.Predicate
 
 interface JobStorage {
@@ -17,7 +18,7 @@ interface JobStorage {
   fun getAllMatchingFilter(predicate: Predicate<JobSpec>): List<JobSpec>
 
   @WorkerThread
-  fun getPendingJobsWithNoDependenciesInCreatedOrder(currentTime: Long): List<JobSpec>
+  fun getNextEligibleJob(currentTime: Long, filter: (MinimalJobSpec) -> Boolean): JobSpec?
 
   @WorkerThread
   fun getJobsInQueue(queue: String): List<JobSpec>

app/src/main/java/org/thoughtcrime/securesms/jobs/FastJobStorage.kt

@@ -27,7 +27,7 @@ class FastJobStorage(private val jobDatabase: JobDatabase) : JobStorage {
 
   /**
    * We keep a set of job specs in memory to facilitate fast retrieval. This is important because the most common job storage pattern is
-   * [getPendingJobsWithNoDependenciesInCreatedOrder], which needs to return full specs.
+   * [getNextEligibleJob], which needs to return full specs.
    */
   private val jobSpecCache: LRUCache<String, JobSpec> = LRUCache(JOB_CACHE_LIMIT)
 
@@ -41,7 +41,7 @@ class FastJobStorage(private val jobDatabase: JobDatabase) : JobStorage {
   /** We keep every dependency in memory, since there aren't that many, and managing a limited subset would be very complicated. */
   private val dependenciesByJobId: MutableMap<String, MutableList<DependencySpec>> = hashMapOf()
 
-  /** The list of jobs eligible to be returned from [getPendingJobsWithNoDependenciesInCreatedOrder], kept sorted in the appropriate order. */
+  /** The list of jobs eligible to be returned from [getNextEligibleJob], kept sorted in the appropriate order. */
   private val eligibleJobs: TreeSet<MinimalJobSpec> = TreeSet(EligibleMinJobComparator)
 
   /** All migration-related jobs, kept in the appropriate order. */
@@ -124,16 +124,16 @@ class FastJobStorage(private val jobDatabase: JobDatabase) : JobStorage {
   }
 
   @Synchronized
-  override fun getPendingJobsWithNoDependenciesInCreatedOrder(currentTime: Long): List<JobSpec> {
+  override fun getNextEligibleJob(currentTime: Long, filter: (MinimalJobSpec) -> Boolean): JobSpec? {
     val stopwatch = debugStopwatch("get-pending")
     val migrationJob: MinimalJobSpec? = migrationJobs.firstOrNull()
 
     return if (migrationJob != null && !migrationJob.isRunning && migrationJob.hasEligibleRunTime(currentTime)) {
-      listOf(migrationJob.toJobSpec())
+      migrationJob.toJobSpec()
     } else if (migrationJob != null) {
-      emptyList()
+      null
     } else {
-      val minJobs: List<MinimalJobSpec> = eligibleJobs
+      eligibleJobs
         .asSequence()
         .filter { job ->
           // Filter out all jobs with unmet dependencies
@@ -141,9 +141,8 @@ class FastJobStorage(private val jobDatabase: JobDatabase) : JobStorage {
         }
         .filterNot { it.isRunning }
         .filter { job -> job.hasEligibleRunTime(currentTime) }
-        .toList()
+        .firstOrNull(filter)
-
+        ?.toJobSpec()
-      getFullJobs(minJobs)
     }.also {
       stopwatch?.stop(TAG)
     }
@@ -521,21 +520,6 @@ class FastJobStorage(private val jobDatabase: JobDatabase) : JobStorage {
     }
   }
 
-  private fun getFullJobs(minJobs: Collection<MinimalJobSpec>): List<JobSpec> {
-    val requestedKeys = minJobs.map { it.id }.toSet()
-    val cachedKeys = jobSpecCache.keys.intersect(requestedKeys)
-    val uncachedKeys = requestedKeys.subtract(cachedKeys)
-
-    val cachedJobs = cachedKeys.map { jobSpecCache[it]!! }
-    val fetchedJobs = jobDatabase.getJobSpecsByKeys(uncachedKeys)
-
-    val sorted = TreeSet(EligibleFullJobComparator).apply {
-      addAll(cachedJobs)
-      addAll(fetchedJobs)
-    }
-    return sorted.toList()
-  }
-
   private object EligibleMinJobComparator : Comparator<MinimalJobSpec> {
     override fun compare(o1: MinimalJobSpec, o2: MinimalJobSpec): Int {
       // We want to sort by priority descending, then createTime ascending

app/src/test/java/org/thoughtcrime/securesms/jobs/FastJobStorageTest.kt

@@ -5,6 +5,9 @@ import io.mockk.mockk
 import io.mockk.verify
 import org.junit.Test
 import org.thoughtcrime.securesms.assertIs
+import org.thoughtcrime.securesms.assertIsNot
+import org.thoughtcrime.securesms.assertIsNotNull
+import org.thoughtcrime.securesms.assertIsNull
 import org.thoughtcrime.securesms.database.JobDatabase
 import org.thoughtcrime.securesms.jobmanager.Job
 import org.thoughtcrime.securesms.jobmanager.persistence.ConstraintSpec
@@ -15,6 +18,11 @@ import org.thoughtcrime.securesms.testutil.TestHelpers
 import java.nio.charset.Charset
 
 class FastJobStorageTest {
+
+  companion object {
+    val NO_PREDICATE: (MinimalJobSpec) -> Boolean = { true }
+  }
+
   @Test
   fun `init - all stored data available`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
@@ -313,97 +321,100 @@ class FastJobStorageTest {
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - none when earlier item in queue is running`() {
+  fun `getNextEligibleJob - none when earlier item in queue is running`() {
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q", isRunning = true), emptyList(), emptyList())
     val fullSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = "q"), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1, fullSpec2)))
     subject.init()
 
-    subject.getPendingJobsWithNoDependenciesInCreatedOrder(1).size assertIs 0
+    subject.getNextEligibleJob(1, NO_PREDICATE) assertIs null
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - none when all jobs are running`() {
+  fun `getNextEligibleJob - none when all jobs are running`() {
     val fullSpec = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q", isRunning = true), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec)))
     subject.init()
 
-    subject.getPendingJobsWithNoDependenciesInCreatedOrder(10).size assertIs 0
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs null
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - none when next run time is after current time`() {
+  fun `getNextEligibleJob - none when next run time is after current time`() {
     val currentTime = 0L
     val fullSpec = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q", lastRunAttemptTime = 0, nextBackoffInterval = 10), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec)))
     subject.init()
 
-    subject.getPendingJobsWithNoDependenciesInCreatedOrder(currentTime).size assertIs 0
+    subject.getNextEligibleJob(currentTime, NO_PREDICATE) assertIs null
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - none when dependent on another job`() {
+  fun `getNextEligibleJob - none when dependent on another job`() {
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", isRunning = true), emptyList(), emptyList())
     val fullSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2"), emptyList(), listOf(DependencySpec("2", "1", false)))
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1, fullSpec2)))
     subject.init()
 
-    subject.getPendingJobsWithNoDependenciesInCreatedOrder(0).size assertIs 0
+    subject.getNextEligibleJob(0, NO_PREDICATE) assertIs null
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - single eligible job`() {
+  fun `getNextEligibleJob - single eligible job`() {
     val fullSpec = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q"), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec)))
     subject.init()
 
-    subject.getPendingJobsWithNoDependenciesInCreatedOrder(10).size assertIs 1
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec.jobSpec
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - multiple eligible jobs`() {
+  fun `getNextEligibleJob - multiple eligible jobs`() {
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1"), emptyList(), emptyList())
     val fullSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2"), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1, fullSpec2)))
     subject.init()
 
-    subject.getPendingJobsWithNoDependenciesInCreatedOrder(10).size assertIs 2
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec1.jobSpec
+    subject.deleteJob(fullSpec1.jobSpec.id)
+
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec2.jobSpec
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - single eligible job in mixed list`() {
+  fun `getNextEligibleJob - single eligible job in mixed list`() {
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", isRunning = true), emptyList(), emptyList())
     val fullSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2"), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1, fullSpec2)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    val job = subject.getNextEligibleJob(10, NO_PREDICATE)
-    jobs.size assertIs 1
+    job.assertIsNotNull()
-    jobs[0].id assertIs "2"
+    job.id assertIs fullSpec2.jobSpec.id
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - first item in queue`() {
+  fun `getNextEligibleJob - first item in queue`() {
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q"), emptyList(), emptyList())
     val fullSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = "q"), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1, fullSpec2)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    val job = subject.getNextEligibleJob(10, NO_PREDICATE)
-    jobs.size assertIs 1
+    job.assertIsNotNull()
-    jobs[0].id assertIs "1"
+    job.id assertIs fullSpec1.jobSpec.id
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - first item in queue with priority`() {
+  fun `getNextEligibleJob - first item in queue with priority`() {
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q", createTime = 1, priority = Job.Parameters.PRIORITY_LOW), emptyList(), emptyList())
     val fullSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = "q", createTime = 2, priority = Job.Parameters.PRIORITY_HIGH), emptyList(), emptyList())
     val fullSpec3 = FullSpec(jobSpec(id = "3", factoryKey = "f3", queueKey = "q", createTime = 3, priority = Job.Parameters.PRIORITY_DEFAULT), emptyList(), emptyList())
@@ -411,13 +422,13 @@ class FastJobStorageTest {
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1, fullSpec2, fullSpec3)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    val job = subject.getNextEligibleJob(10, NO_PREDICATE)
-    jobs.size assertIs 1
+    job.assertIsNotNull()
-    jobs[0].id assertIs "2"
+    job.id assertIs fullSpec2.jobSpec.id
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - complex priority`() {
+  fun `getNextEligibleJob - complex priority`() {
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q1", createTime = 1, priority = Job.Parameters.PRIORITY_LOW), emptyList(), emptyList())
     val fullSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = "q1", createTime = 2, priority = Job.Parameters.PRIORITY_HIGH), emptyList(), emptyList())
     val fullSpec3 = FullSpec(jobSpec(id = "3", factoryKey = "f3", queueKey = "q2", createTime = 3, priority = Job.Parameters.PRIORITY_DEFAULT), emptyList(), emptyList())
@@ -431,18 +442,35 @@ class FastJobStorageTest {
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1, fullSpec2, fullSpec3, fullSpec4, fullSpec5, fullSpec6, fullSpec7, fullSpec8, fullSpec9)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec2.jobSpec
-    jobs.size assertIs 6
+    subject.deleteJob(fullSpec2.jobSpec.id)
-    jobs[0].id assertIs "2"
+
-    jobs[1].id assertIs "6"
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec6.jobSpec
-    jobs[2].id assertIs "3"
+    subject.deleteJob(fullSpec6.jobSpec.id)
-    jobs[3].id assertIs "9"
+
-    jobs[4].id assertIs "7"
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec3.jobSpec
-    jobs[5].id assertIs "8"
+    subject.deleteJob(fullSpec3.jobSpec.id)
+
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec5.jobSpec
+    subject.deleteJob(fullSpec5.jobSpec.id)
+
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec9.jobSpec
+    subject.deleteJob(fullSpec9.jobSpec.id)
+
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec1.jobSpec
+    subject.deleteJob(fullSpec1.jobSpec.id)
+
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec4.jobSpec
+    subject.deleteJob(fullSpec4.jobSpec.id)
+
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec7.jobSpec
+    subject.deleteJob(fullSpec7.jobSpec.id)
+
+    subject.getNextEligibleJob(10, NO_PREDICATE) assertIs fullSpec8.jobSpec
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - lastRunAttemptTime in the future runs right away`() {
+  fun `getNextEligibleJob - lastRunAttemptTime in the future runs right away`() {
     val currentTime = 10L
 
     val fullSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q", lastRunAttemptTime = 100, nextBackoffInterval = 5), emptyList(), emptyList())
@@ -450,63 +478,61 @@ class FastJobStorageTest {
     val subject = FastJobStorage(mockDatabase(listOf(fullSpec1)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(currentTime)
+    val job = subject.getNextEligibleJob(currentTime, NO_PREDICATE)
-    jobs.size assertIs 1
+    job.assertIsNotNull()
-    jobs[0].id assertIs "1"
+    job.id assertIs fullSpec1.jobSpec.id
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - migration job takes precedence`() {
+  fun `getNextEligibleJob - migration job takes precedence`() {
     val plainSpec = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q", createTime = 0), emptyList(), emptyList())
     val migrationSpec = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = Job.Parameters.MIGRATION_QUEUE_KEY, createTime = 5), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(plainSpec, migrationSpec)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    val job = subject.getNextEligibleJob(10, NO_PREDICATE)
-    jobs.size assertIs 1
+    job.assertIsNotNull()
-    jobs[0].id assertIs "2"
+    job.id assertIs migrationSpec.jobSpec.id
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - running migration blocks normal jobs`() {
+  fun `getNextEligibleJob - running migration blocks normal jobs`() {
     val plainSpec = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = "q", createTime = 0), emptyList(), emptyList())
     val migrationSpec = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = Job.Parameters.MIGRATION_QUEUE_KEY, createTime = 5, isRunning = true), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(plainSpec, migrationSpec)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    subject.getNextEligibleJob(10, NO_PREDICATE).assertIsNull()
-    jobs.size assertIs 0
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - running migration blocks later migration jobs`() {
+  fun `getNextEligibleJob - running migration blocks later migration jobs`() {
     val migrationSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = Job.Parameters.MIGRATION_QUEUE_KEY, createTime = 0, isRunning = true), emptyList(), emptyList())
     val migrationSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = Job.Parameters.MIGRATION_QUEUE_KEY, createTime = 5), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(migrationSpec1, migrationSpec2)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    subject.getNextEligibleJob(10, NO_PREDICATE).assertIsNull()
-    jobs.size assertIs 0
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - only return first eligible migration job`() {
+  fun `getNextEligibleJob - only return first eligible migration job`() {
     val migrationSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = Job.Parameters.MIGRATION_QUEUE_KEY, createTime = 0), emptyList(), emptyList())
     val migrationSpec2 = FullSpec(jobSpec(id = "2", factoryKey = "f2", queueKey = Job.Parameters.MIGRATION_QUEUE_KEY, createTime = 5), emptyList(), emptyList())
 
     val subject = FastJobStorage(mockDatabase(listOf(migrationSpec1, migrationSpec2)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(10)
+    val job = subject.getNextEligibleJob(10, NO_PREDICATE)
-    jobs.size assertIs 1
+    job.assertIsNotNull()
-    jobs[0].id assertIs "1"
+    job.id assertIs migrationSpec1.jobSpec.id
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - migration job that isn't scheduled to run yet blocks later migration jobs`() {
+  fun `getNextEligibleJob - migration job that isn't scheduled to run yet blocks later migration jobs`() {
     val currentTime = 10L
 
     val migrationSpec1 = FullSpec(jobSpec(id = "1", factoryKey = "f1", queueKey = Job.Parameters.MIGRATION_QUEUE_KEY, createTime = 0, lastRunAttemptTime = 0, nextBackoffInterval = 999), emptyList(), emptyList())
@@ -515,26 +541,22 @@ class FastJobStorageTest {
     val subject = FastJobStorage(mockDatabase(listOf(migrationSpec1, migrationSpec2)))
     subject.init()
 
-    val jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(currentTime)
+    subject.getNextEligibleJob(currentTime, NO_PREDICATE).assertIsNull()
-    jobs.size assertIs 0
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - after deleted, no longer is in eligible list`() {
+  fun `getNextEligibleJob - after deleted, no longer is in eligible list`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs true
+    subject.deleteJob(DataSet1.JOB_1.id)
 
-    subject.deleteJobs(listOf("id1"))
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIsNot DataSet1.JOB_1
-
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
-    jobs.contains(DataSet1.JOB_1) assertIs false
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - after deleted, next item in queue is eligible`() {
+  fun `getNextEligibleJob - after deleted, next item in queue is eligible`() {
     // Two jobs in the same queue but with different create times
     val firstJob = DataSet1.JOB_1
     val secondJob = DataSet1.JOB_1.copy(id = "id2", createTime = 2)
@@ -548,129 +570,101 @@ class FastJobStorageTest {
     )
     subject.init()
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs firstJob
-    jobs.size assertIs 1
+    subject.deleteJob(firstJob.id)
-    jobs.contains(firstJob) assertIs true
 
-    subject.deleteJobs(listOf("id1"))
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs secondJob
-
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
-    jobs.size assertIs 1
-    jobs.contains(firstJob) assertIs false
-    jobs.contains(secondJob) assertIs true
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - after marked running, no longer is in eligible list`() {
+  fun `getNextEligibleJob - after marked running, no longer is in eligible list`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs true
+    subject.markJobAsRunning(DataSet1.JOB_1.id, 1)
 
-    subject.markJobAsRunning("id1", 1)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIsNot DataSet1.JOB_1
-
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
-    jobs.contains(DataSet1.JOB_1) assertIs false
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - after updateJobAfterRetry to be invalid, no longer is in eligible list`() {
+  fun `getNextEligibleJob - after updateJobAfterRetry to be invalid, no longer is in eligible list`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs true
+    subject.updateJobAfterRetry(DataSet1.JOB_1.id, 1, 1000, 1_000_000, null)
 
-    subject.updateJobAfterRetry("id1", 1, 1000, 1_000_000, null)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIsNot DataSet1.JOB_1
-
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
-    jobs.contains(DataSet1.JOB_1) assertIs false
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - after invalid then marked pending, is in eligible list`() {
+  fun `getNextEligibleJob - after invalid then marked pending, is in eligible list`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
-    subject.markJobAsRunning("id1", 1)
+    subject.markJobAsRunning(DataSet1.JOB_1.id, 1)
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIsNot DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs false
 
     subject.updateAllJobsToBePending()
 
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE)?.id assertIs DataSet1.JOB_1.id // The last run attempt time changes, so some fields will be different
-    jobs.filter { it.id == DataSet1.JOB_1.id }.size assertIs 1 // The last run attempt time changes, so some fields will be different
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - after updateJobs to be invalid, no longer is in eligible list`() {
+  fun `getNextEligibleJob - after updateJobs to be invalid, no longer is in eligible list`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs true
-
     subject.updateJobs(listOf(DataSet1.JOB_1.copy(isRunning = true)))
 
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIsNot DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs false
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - newly-inserted higher-priority job in queue replaces old`() {
+  fun `getNextEligibleJob - newly-inserted higher-priority job in queue replaces old`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs true
 
     val higherPriorityJob = DataSet1.JOB_1.copy(id = "id-bigboi", priority = Job.Parameters.PRIORITY_HIGH)
     subject.insertJobs(listOf(FullSpec(jobSpec = higherPriorityJob, constraintSpecs = emptyList(), dependencySpecs = emptyList())))
 
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs higherPriorityJob
-    jobs.contains(DataSet1.JOB_1) assertIs false
-    jobs.contains(higherPriorityJob) assertIs true
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - updating job to have a higher priority replaces lower priority in queue`() {
+  fun `getNextEligibleJob - updating job to have a higher priority replaces lower priority in queue`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
     val lowerPriorityJob = DataSet1.JOB_1.copy(id = "id-bigboi", priority = Job.Parameters.PRIORITY_LOW)
     subject.insertJobs(listOf(FullSpec(jobSpec = lowerPriorityJob, constraintSpecs = emptyList(), dependencySpecs = emptyList())))
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs true
-    jobs.contains(lowerPriorityJob) assertIs false
 
     val higherPriorityJob = lowerPriorityJob.copy(priority = Job.Parameters.PRIORITY_HIGH)
     subject.updateJobs(listOf(higherPriorityJob))
 
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs higherPriorityJob
-    jobs.contains(DataSet1.JOB_1) assertIs false
-    jobs.contains(higherPriorityJob) assertIs true
   }
 
   @Test
-  fun `getPendingJobsWithNoDependenciesInCreatedOrder - updating job to have an older createTime replaces newer in queue`() {
+  fun `getNextEligibleJob - updating job to have an older createTime replaces newer in queue`() {
     val subject = FastJobStorage(mockDatabase(DataSet1.FULL_SPECS))
     subject.init()
 
     val newerJob = DataSet1.JOB_1.copy(id = "id-bigboi", createTime = 1000)
     subject.insertJobs(listOf(FullSpec(jobSpec = newerJob, constraintSpecs = emptyList(), dependencySpecs = emptyList())))
 
-    var jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs DataSet1.JOB_1
-    jobs.contains(DataSet1.JOB_1) assertIs true
-    jobs.contains(newerJob) assertIs false
 
     val olderJob = newerJob.copy(createTime = 0)
     subject.updateJobs(listOf(olderJob))
 
-    jobs = subject.getPendingJobsWithNoDependenciesInCreatedOrder(100)
+    subject.getNextEligibleJob(100, NO_PREDICATE) assertIs olderJob
-    jobs.contains(DataSet1.JOB_1) assertIs false
-    jobs.contains(olderJob) assertIs true
   }
 
   @Test
@@ -825,10 +819,6 @@ class FastJobStorageTest {
     every { mock.getAllDependencySpecs() } returns dependencies
     every { mock.getConstraintSpecsForJobs(any()) } returns constraints
     every { mock.getJobSpec(any()) } answers { jobs.first { it.id == firstArg() } }
-    every { mock.getJobSpecsByKeys(any()) } answers {
-      val ids: Collection<String> = firstArg()
-      jobs.filter { ids.contains(it.id) }
-    }
     every { mock.insertJobs(any()) } answers {
       val inserts: List<FullSpec> = firstArg()
       for (insert in inserts) {

app/src/testShared/org/thoughtcrime/securesms/KotlinAssertsUtil.kt

@@ -7,12 +7,22 @@ package org.thoughtcrime.securesms
 
 import org.hamcrest.MatcherAssert
 import org.hamcrest.Matchers
+import kotlin.contracts.ExperimentalContracts
+import kotlin.contracts.contract
 
+@OptIn(ExperimentalContracts::class)
 fun <T : Any?> T.assertIsNull() {
+  contract {
+    returns() implies (this@assertIsNull == null)
+  }
   MatcherAssert.assertThat(this, Matchers.nullValue())
 }
 
+@OptIn(ExperimentalContracts::class)
 fun <T : Any?> T.assertIsNotNull() {
+  contract {
+    returns() implies (this@assertIsNotNull != null)
+  }
   MatcherAssert.assertThat(this, Matchers.notNullValue())
 }
 
@@ -20,7 +30,7 @@ infix fun <T : Any?> T.assertIs(expected: T) {
   MatcherAssert.assertThat(this, Matchers.`is`(expected))
 }
 
-infix fun <T : Any> T.assertIsNot(expected: T) {
+infix fun <T : Any?> T.assertIsNot(expected: T) {
   MatcherAssert.assertThat(this, Matchers.not(Matchers.`is`(expected)))
 }