Skip to content
GitLab
Open
Issue created by Martin Aarhus Gregersen@aarhusgregersen

Cron-only queues running immediately on deploy (process.processImmediate (node:internal/timers))

Description

I have recently implemented Bull in our product and this is working dandy in development. However, when deploying to production I'm seeing a weird thing happen. The process is firing immediately, despite only being defined as a repeat: { cron: .. } job.

This ends up in the deploy process terminating, with the folllwing log entry

2023-03-14T10:39:47+01:00 Nothing listening on 0.0.0.0:8080 yet. If the deployment fails after this message, please update your configuration and redeploy.
2023-03-14T10:39:47+01:00 [PROD] Using REDACTED email transport
2023-03-14T10:39:49+01:00 Current environment is: production
2023-03-14T10:39:49+01:00 Introdus backend listening on port REDACTED
2023-03-14T10:39:49+01:00 Connected to redis
2023-03-14T10:39:49+01:00 Mar 14 09:39:48 app[4682] INFO: ./d/app: Listening on REDACTED on production
2023-03-14T10:39:51+01:00 at process.processImmediate (node:internal/timers:442:9)
2023-03-14T10:39:51+01:00 at runNextTicks (node:internal/process/task_queues:60:5)
2023-03-14T10:39:51+01:00 ^
2023-03-14T10:39:51+01:00 at updateDBAndSendMessageOthers (/home/bas/app_cfe13b28-05ef-4b24-9077-1dcd2ed7b1ea/dist/src/common/jobs/processes/sendMessage.process.js:37:15)
2023-03-14T10:39:51+01:00 Error: JOB.USER_NOT_FOUND
2023-03-14T10:39:51+01:00 /home/bas/app_cfe13b28-05ef-4b24-9077-1dcd2ed7b1ea/dist/src/common/jobs/processes/sendMessage.process.js:37
2023-03-14T10:39:51+01:00 Node.js v18.11.0
2023-03-14T10:39:51+01:00 throw new Error('JOB.USER_NOT_FOUND');

I don't even think an error in this manner should be cancelling the build steps, but its weirder yet, because it shouldn't be running at this point. As you'll see below the queues are only being added as repeatable jobs.

Minimal, Working Test code to reproduce the issue.

(An easy to reproduce test case will dramatically decrease the resolution time.)

Currently this setup is part of a larger monolithic application. The relevant codebits are included below:

jobrunner.ts

import { userActivationRemindersQueue } from './queues/userActivationReminders.queue';
import { remindersQueue } from './queues/reminders.queue';
import { newTimelineSectionsQueue } from './queues/newTimelineSection.queue';
import { networkUnreadsQueue } from './queues/networkUnreads.queue';
import { taskRemindersQueue } from './queues/taskReminders.queue';
// import { runTaskDigestJob } from './queues/taskDigests.queue';
import { sendMessagesQueue } from './queues/sendMessage.queue';

const env = process.env.NODE_ENV || 'development';

export const jobRunner = () => {
	try {
		if (env === 'production') {
			scheduledAssignmentsQueue.add({ repeat: { cron: '0 2 * * 1-5' } });
			userActivationRemindersQueue.add({ repeat: { cron: '30 2 * * 1-5' } });
			remindersQueue.add({ repeat: { cron: '0 3 * * 1-5' } });
			newTimelineSectionsQueue.add({ repeat: { cron: '30 3 * * 1-5' } });
			networkUnreadsQueue.add({ repeat: { cron: '0 4 * * 1-5' } });
			taskRemindersQueue.add({ repeat: { cron: '30 4 * * 1-5' } });
			// runTaskDigestJob();
			sendMessagesQueue.add({ repeat: { cron: '30 7 * * 1-5' } });
		} 
	} catch (err) {
		console.log(`${env.toUpperCase}: Could not start job runner.`);
		console.log(err);
	}
};

app.ts

import express from 'express';
import path from 'path';
import dotenv from 'dotenv';
import session from 'express-session';
const RedisStore = require('connect-redis')(session);
import useragent from 'express-useragent';
import cors from 'cors';
import passport from 'passport';
const helmet = require('helmet');
import morgan from 'morgan';
import * as Sentry from '@sentry/node';
import * as Tracing from '@sentry/tracing';

dotenv.config({ debug: true });
const app: express.Application = express();
const env = process.env.NODE_ENV || 'development';
const utils = require('./src/common/helpers/utils');
const auth = require('./src/common/auth/helpers/authHelper');

import { redisClient } from './src/common/helpers/db';
import { trimmer } from './src/common/utilities/trimmer';
import { corsSettings } from './src/common/auth/helpers/corsSettings';
import whilelistDomains from './src/common/auth/helpers/whitelist.json';
import { createBullBoard } from '@bull-board/api';
import { BullAdapter } from '@bull-board/api/bullAdapter';
import { ExpressAdapter } from '@bull-board/express';
import { networkUnreadsQueue } from './src/common/jobs/queues/networkUnreads.queue';
import { jobRunner } from './src/common/jobs/jobRunner';
import { taskDigestsQueue } from './src/common/jobs/queues/taskDigests.queue';
import { remindersQueue } from './src/common/jobs/queues/reminders.queue';
import { userActivationRemindersQueue } from './src/common/jobs/queues/userActivationReminders.queue';
import { sendMessagesQueue } from './src/common/jobs/queues/sendMessage.queue';

// Configure Sentry (Error reporting tool)
Sentry.init({
	dsn: process.env.SENTRY_DSN,
	integrations: [
		// enable HTTP calls tracing
		new Sentry.Integrations.Http({ tracing: true }),
		new Sentry.Integrations.Console(),
		// enable Express.js middleware tracing
		new Tracing.Integrations.Express({ app }),
	],
	tracesSampleRate: 0.5,
	environment: env,
});

app.use(Sentry.Handlers.requestHandler() as express.RequestHandler);
app.use(Sentry.Handlers.tracingHandler());

app.use('/__checkstatus__', (_req: express.Request, res: express.Response) => {
	res.status(200).send('');
});

app.set('trust proxy', true);
app.set('views', path.join(__dirname, 'views'));

// Improve security
app.use(helmet());

// Prep useragent
app.use(useragent.express());

// Support application/scim+json
app.use(
	express.json({
		type: 'application/*+json',
	})
);

// Support application/json
app.use(express.json());

app.use(
	express.urlencoded({
		extended: true,
	})
);

app.use('/docs/', express.static(path.join(__dirname, '../public/docs/')));
app.get('/docs', (req: express.Request, res: express.Response) => {
	res.sendFile('public/docs/index.html', {
		root: __dirname,
	});
});

// Splitting express session definition to allow unsecure cookies in development
const sess: session.SessionOptions = {
	store: new RedisStore({ client: redisClient }),
	secret: process.env.SESSION_SECRET,
	resave: false,
	saveUninitialized: false,
	cookie: {},
};

app.use(express.static(path.join(__dirname, 'public')));
app.use(session(sess));

app.use(passport.initialize());
app.use(passport.session());

const corsOptions: cors.CorsOptions = {
	allowedHeaders: [
		'Origin',
		'X-Requested-With',
		'Content-Type',
		'Accept',
		'X-Access-Token',
		'x-user_id',
		'sentry-trace',
	],
	credentials: true,
	methods: 'GET,HEAD,OPTIONS,PUT,PATCH,POST,DELETE',
	preflightContinue: false,
};

// Setup CORS
app.use(cors(corsOptions));
// Manual cors configuration for subdomains etc.
app.use(corsSettings);

// Connect to redis
redisClient.on('connect', () => {
	console.log('Connected to redis');
});

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
	queues: [
		new BullAdapter(networkUnreadsQueue),
		new BullAdapter(taskDigestsQueue),
		new BullAdapter(remindersQueue),
		new BullAdapter(userActivationRemindersQueue),
		new BullAdapter(sendMessagesQueue),
	],
	serverAdapter: serverAdapter,
});

app.use('/admin/queues', serverAdapter.getRouter());

jobRunner();

All of our queues and processes are formatted following a similar pattern, here's a sample queue:

import Queue from 'bull';
import { networkUnreadsProcess } from '../processes/networkUnreads.process';

export const networkUnreadsQueue = new Queue(
	'Network Unreads Notifications',
	process.env.REDIS_URL,
	{
		redis: { maxRetriesPerRequest: null, enableReadyCheck: false },
	}
);

networkUnreadsQueue.process(networkUnreadsProcess);

export const runNetworkUnreadsJob = (data?: any) => {
	networkUnreadsQueue.add(data);
};

The method runNetworkUnreadsJob is there to run on-demand jobs, but isn't currently exported or in use anywhere.

And a process:

import { Job } from 'bull';
import organizationService from '../../../organization/organization.service';
import { JobsService } from '../jobs.service';
import { NetworkService } from '../../../network/network.service';

export const networkUnreadsProcess = async (job?: Job) => {
	try {
		await getNetworkUnreads(job);
	} catch (err) {
		console.log(err);
	}
};

async function getNetworkUnreads(job?: Job) {
    // some heavy lifting ...
}

Bull version

4.10.2

Additional information

I'm using Typescript. I previously had an issue with bull-board but that has now been resolved. I don't understand this irregular behaviour and why it is attempting to run the code on deploy. The only place the JOB.USER_NOT_FOUND error is used is from within sendMessagesQueue.

The hosting provider (Clever Cloud) does provide some cron tools, but it seems that it's not possible for me to execute just the bull processes/queues in an isolated environment, as it needs context from the rest of the application.

Am I missing an implication of sorts? That somehow the jobs will run right away but also on a repeatable schedule? I do not believe that is the case, but please let me know if I am wrong.