JobQueue: Allow jobs to be added before streaming starts

ts/jobs/JobQueue.ts

@@ -167,14 +167,14 @@ export abstract class JobQueue<T> {
     data: Readonly<T>,
     insert?: (job: ParsedJob<T>) => Promise<void>
   ): Promise<Job<T>> {
+    const job = this.createJob(data);
+
     if (!this.started) {
-      throw new Error(
-        `${this.logPrefix} has not started streaming. Make sure to call streamJobs().`
+      log.warn(
+        `${this.logPrefix} This queue has not started streaming, adding job ${job.id} to database only.`
       );
     }
 
-    const job = this.createJob(data);
-
     if (insert) {
       await insert(job);
     }

ts/jobs/JobQueueDatabaseStore.ts

@@ -35,21 +35,22 @@ export class JobQueueDatabaseStore implements JobQueueStore {
     );
 
     const initialFetchPromise = this.initialFetchPromises.get(job.queueType);
-    if (!initialFetchPromise) {
-      throw new Error(
-        `JobQueueDatabaseStore tried to add job for queue ${JSON.stringify(
-          job.queueType
-        )} but streaming had not yet started`
+    if (initialFetchPromise) {
+      await initialFetchPromise;
+    } else {
+      log.warn(
+        `JobQueueDatabaseStore: added job for queue "${job.queueType}" but streaming has not yet started (shouldPersist=${shouldPersist})`
       );
     }
-    await initialFetchPromise;
 
     if (shouldPersist) {
       await this.db.insertJob(formatJobForInsert(job));
     }
 
+    if (initialFetchPromise) {
       this.getQueue(job.queueType).add(job);
     }
+  }
 
   async delete(id: string): Promise<void> {
     await this.db.deleteJob(id);

ts/test-node/jobs/JobQueueDatabaseStore_test.ts

@@ -25,22 +25,23 @@ describe('JobQueueDatabaseStore', () => {
   });
 
   describe('insert', () => {
-    it("fails if streaming hasn't started yet", async () => {
+    it("adds jobs to database even if streaming hasn't started yet", async () => {
       const store = new JobQueueDatabaseStore(fakeDatabase);
 
-      let error: unknown;
-      try {
       await store.insert({
         id: 'abc',
         timestamp: 1234,
         queueType: 'test queue',
         data: { hi: 5 },
       });
-      } catch (err: unknown) {
-        error = err;
-      }
 
-      assert.instanceOf(error, Error);
+      sinon.assert.calledOnce(fakeDatabase.insertJob);
+      sinon.assert.calledWithMatch(fakeDatabase.insertJob, {
+        id: 'abc',
+        timestamp: 1234,
+        queueType: 'test queue',
+        data: { hi: 5 },
+      });
     });
 
     it('adds jobs to the database', async () => {
@@ -153,6 +154,26 @@ describe('JobQueueDatabaseStore', () => {
       sinon.assert.calledOnce(fakeDatabase.insertJob);
       assert.deepEqual(events, ['loaded jobs', 'insert']);
     });
+
+    it("adds jobs if we haven't started streaming at all", async () => {
+      const events: Array<string> = [];
+
+      fakeDatabase.insertJob.callsFake(() => {
+        events.push('insert');
+      });
+
+      const store = new JobQueueDatabaseStore(fakeDatabase);
+
+      await store.insert({
+        id: 'abc',
+        timestamp: 1234,
+        queueType: 'test queue',
+        data: { hi: 5 },
+      });
+
+      sinon.assert.calledOnce(fakeDatabase.insertJob);
+      assert.deepEqual(events, ['insert']);
+    });
   });
 
   describe('delete', () => {

ts/test-node/jobs/JobQueue_test.ts

@@ -838,7 +838,7 @@ describe('JobQueue', () => {
   });
 
   describe('add', () => {
-    it('rejects if the job queue has not started streaming', async () => {
+    it('adds even if the job queue has not started streaming', async () => {
       const fakeStore = {
         insert: sinon.stub().resolves(),
         delete: sinon.stub().resolves(),
@@ -861,7 +861,7 @@ describe('JobQueue', () => {
         maxAttempts: 99,
       });
 
-      await assert.isRejected(noopQueue.add(undefined));
+      await noopQueue.add(undefined);
 
       sinon.assert.notCalled(fakeStore.stream as sinon.SinonStub);
     });