22647221264

Committed 03 Mar 2026 11:16PM UTC coverage: 63.073%. First build

Build # 22647221264

Build Type

Pull #3527

github

Committed by

web-flow

Commit Message

Merge 19c79d71e into d650c6db2

Pull Request Pull Request #3527: ES6 & ES7 updates

Coverage Stats

950 of 1422 branches covered (66.81%)

25 of 36 new or added lines in 8 files covered. (69.44%)

5418 of 8590 relevant lines covered (63.07%)

27.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.94

/outbound/queue.js

'use strict'

const child_process = require('node:child_process')
const fs = require('node:fs')
const path = require('node:path')

const async = require('async')
const { Address } = require('address-rfc2821')
const config = require('haraka-config')

const logger = require('../logger')
const TimerQueue = require('./timer_queue')
const HMailItem = require('./hmail')
const obc = require('./config')
const _qfile = require('./qfile')
const obtls = require('./tls')

exports.name = 'outbound/queue'

let queue_dir
if (config.get('queue_dir')) {
    queue_dir = path.resolve(config.get('queue_dir'))
} else if (process.env.HARAKA) {
    queue_dir = path.resolve(process.env.HARAKA, 'queue')
} else {
    queue_dir = path.resolve('test', 'test-queue')
}

exports.queue_dir = queue_dir

const load_queue = async.queue((file, cb) => {
    const hmail = new HMailItem(file, path.join(queue_dir, file))
    exports._add_hmail(hmail)
    hmail.once('ready', cb)
}, obc.cfg.concurrency_max)

let in_progress = 0
const delivery_queue = (exports.delivery_queue = async.queue((hmail, cb) => {
    in_progress++
    hmail.next_cb = () => {
        in_progress--
        cb()
    }
    if (obtls.cfg) return hmail.send()
    obtls.init(() => {
        hmail.send()
    })
}, obc.cfg.concurrency_max))

const temp_fail_queue = (exports.temp_fail_queue = new TimerQueue())

let queue_count = 0

exports.get_stats = () => `${in_progress}/${exports.delivery_queue.length()}/${exports.temp_fail_queue.length()}`

exports.list_queue = (cb) => {
    exports._load_cur_queue(null, exports._list_file, cb)
}

exports._stat_file = (file, cb) => {
    queue_count++
    setImmediate(cb)
}

exports.stat_queue = (cb) => {
    const self = exports
    exports._load_cur_queue(null, exports._stat_file, (err) => {
        if (err) return cb(err)
        return cb(null, self.stats())
    })
}

exports.load_queue = (pid) => {
    // Initialise and load queue
    // This function is called first when not running under cluster,
    exports.ensure_queue_dir()
    exports.delete_dot_files()

    exports._load_cur_queue(pid, exports._add_file, () => {
        logger.info(exports, `[pid: ${pid}] ${delivery_queue.length()} files in my delivery queue`)
        logger.info(exports, `[pid: ${pid}] ${load_queue.length()} files in my load queue`)
        logger.info(exports, `[pid: ${pid}] ${temp_fail_queue.length()} files in my temp fail queue`)
    })
}

exports._load_cur_queue = (pid, iteratee, cb) => {
    logger.info(exports, 'Loading outbound queue from ', queue_dir)
    fs.readdir(queue_dir, (err, files) => {
        if (err) {
            return logger.error(exports, `Failed to load queue directory (${queue_dir}): ${err}`)
        }

        this.cur_time = new Date() // set once so we're not calling it a lot

        this.load_queue_files(pid, files, iteratee, cb)
    })
}

exports.read_parts = (file) => {
    if (file.startsWith(_qfile.platformDOT)) {
        logger.warn(exports, `'Skipping' dot-file in queue folder: ${file}`)
        return false
    }

    if (file.startsWith('error.')) {
        logger.warn(exports, `'Skipping' error file in queue folder: ${file}`)
        return false
    }

    const parts = _qfile.parts(file)
    if (!parts) {
        logger.error(exports, `Unrecognized file in queue folder: ${file}`)
        return false
    }

    return parts
}

exports.rename_to_actual_pid = (file, parts, cb) => {
    // maintain some original details for the rename
    const new_filename = _qfile.name({
        arrival: parts.arrival,
        uid: parts.uid,
        next_attempt: parts.next_attempt,
        attempts: parts.attempts,
    })

    fs.rename(path.join(queue_dir, file), path.join(queue_dir, new_filename), (err) => {
        if (err) {
            return cb(`Unable to rename queue file: ${file} to ${new_filename} : ${err}`)
        }

        cb(null, new_filename)
    })
}

exports._add_file = (file, cb) => {
    const self = exports
    const parts = _qfile.parts(file)

    if (parts.next_attempt <= self.cur_time) {
        logger.debug(exports, `File ${file} needs processing now`)
        load_queue.push(file)
    } else {
        logger.debug(exports, `File ${file} needs processing later: ${parts.next_attempt - self.cur_time}ms`)
        temp_fail_queue.add(file, parts.next_attempt - self.cur_time, () => {
            load_queue.push(file)
        })
    }

    cb()
}

exports.load_queue_files = (pid, input_files, iteratee, callback = function () {}) => {
    const self = exports
    const searchPid = parseInt(pid)

    let stat_renamed = 0
    let stat_loaded = 0

    if (searchPid) {
        logger.info(exports, `Grabbing queue files for pid: ${pid}`)
    } else {
        logger.info(exports, 'Loading the queue...')
    }

    async.map(
        input_files,
        (file, cb) => {
            const parts = self.read_parts(file)
            if (!parts) return cb()

            if (searchPid) {
                if (parts.pid !== searchPid) return cb()

                self.rename_to_actual_pid(file, parts, (error, renamed_file) => {
                    if (error) {
                        logger.error(exports, `${error}`)
                        return cb()
                    }

                    stat_renamed++
                    stat_loaded++
                    cb(null, renamed_file)
                })
            } else {
                stat_loaded++
                cb(null, file)
            }
        },
        (err, results) => {
            if (err) logger.err(exports, `[pid: ${pid}] ${err}`)
            if (searchPid) logger.info(exports, `[pid: ${pid}] ${stat_renamed} files old PID queue fixed up`)
            logger.debug(exports, `[pid: ${pid}] ${stat_loaded} files loaded`)

            async.map(
                results.filter((i) => i),
                iteratee,
                callback,
            )
        },
    )
}

exports.stats = () => {
    return {
        queue_dir,
        queue_count,
    }
}

exports._list_file = (file, cb) => {
    const tl_reader = fs.createReadStream(path.join(queue_dir, file), {
        start: 0,
        end: 3,
    })
    tl_reader.on('error', (err) => {
        console.error(`Error reading queue file: ${file}:`, err)
    })
    tl_reader.once('data', (buf) => {
        // I'm making the assumption here we won't ever read less than 4 bytes
        // as no filesystem on the planet should be that dumb...
        tl_reader.destroy()
        const todo_len = (buf[0] << 24) + (buf[1] << 16) + (buf[2] << 8) + buf[3]
        const td_reader = fs.createReadStream(path.join(queue_dir, file), {
            encoding: 'utf8',
            start: 4,
            end: todo_len + 3,
        })
        let todo = ''
        td_reader.on('data', (str) => {
            todo += str
            if (Buffer.byteLength(todo) === todo_len) {
                // we read everything
                const todo_struct = JSON.parse(todo)
                todo_struct.rcpt_to = todo_struct.rcpt_to.map((a) => new Address(a))
                todo_struct.mail_from = new Address(todo_struct.mail_from)
                todo_struct.file = file
                todo_struct.full_path = path.join(queue_dir, file)
                const parts = _qfile.parts(file)
                todo_struct.pid = parts?.pid || null
                cb(null, todo_struct)
            }
        })
        td_reader.on('end', () => {
            if (Buffer.byteLength(todo) !== todo_len) {
                console.error("Didn't find right amount of data in todo for file:", file)
                return cb()
            }
        })
    })
}

exports.flush_queue = (domain, pid) => {
    if (domain) {
        exports.list_queue((err, qlist) => {
            if (err) return logger.error(exports, `Failed to load queue: ${err}`)
            for (const todo of qlist) {
                if (todo.domain.toLowerCase() != domain.toLowerCase()) return
                if (pid && todo.pid != pid) return
                // console.log("requeue: ", todo);
                delivery_queue.push(new HMailItem(todo.file, todo.full_path))
            }
        })
    } else {
        temp_fail_queue.drain()
    }
}

exports.load_pid_queue = (pid) => {
    logger.info(exports, `Loading queue for pid: ${pid}`)
    exports.load_queue(pid)
}

exports.ensure_queue_dir = () => {
    // this code is only run at start-up.
    if (fs.existsSync(queue_dir)) return

    logger.debug(exports, `Creating queue directory ${queue_dir}`)
    try {
        fs.mkdirSync(queue_dir, 493) // 493 == 0755
        const cfg = config.get('smtp.ini')
        let uid
        let gid
        if (cfg.user) uid = child_process.execSync(`id -u ${cfg.user}`).toString().trim()
        if (cfg.group) gid = child_process.execSync(`id -g ${cfg.group}`).toString().trim()
        if (uid && gid) {
            fs.chown(queue_dir, uid, gid)
        } else if (uid) {
            fs.chown(queue_dir, uid)
        }
    } catch (err) {
        if (err.code !== 'EEXIST') {
            logger.error(exports, `Error creating queue directory: ${err}`)
            throw err
        }
    }
}

exports.delete_dot_files = () => {
    for (const file of fs.readdirSync(queue_dir)) {
        if (file.startsWith(_qfile.platformDOT)) {
            logger.warn(exports, `Removing left over dot-file: ${file}`)
            return fs.unlinkSync(path.join(queue_dir, file))
        }
    }
}

exports._add_hmail = (hmail) => {
    if (hmail.next_process < exports.cur_time) {
        delivery_queue.push(hmail)
    } else {
        temp_fail_queue.add(hmail.filename, hmail.next_process - exports.cur_time, () => {
            delivery_queue.push(hmail)
        })
    }
}

exports.scan_queue_pids = (cb) => {
    const self = exports

    // Under cluster, this is called first by the master
    self.ensure_queue_dir()
    self.delete_dot_files()

    fs.readdir(queue_dir, (err, files) => {
        if (err) {
            logger.error(exports, `Failed to load queue directory (${queue_dir}): ${err}`)
            return cb(err)
        }

        const pids = {}

        for (const file of files) {
            const parts = self.read_parts(file)
            if (parts) pids[parts.pid] = true
        }

        return cb(null, Object.keys(pids))
    })
}

1	'use strict'	1✔
2		1✔
3	const child_process = require('node:child_process')	1✔
4	const fs = require('node:fs')	1✔
5	const path = require('node:path')	1✔
6		1✔
7	const async = require('async')	1✔
8	const { Address } = require('address-rfc2821')	1✔
9	const config = require('haraka-config')	1✔
10		1✔
11	const logger = require('../logger')	1✔
12	const TimerQueue = require('./timer_queue')	1✔
13	const HMailItem = require('./hmail')	1✔
14	const obc = require('./config')	1✔
15	const _qfile = require('./qfile')	1✔
16	const obtls = require('./tls')	1✔
17		1✔
18	exports.name = 'outbound/queue'	1✔
19		1✔
20	let queue_dir	1✔
21	if (config.get('queue_dir')) {	1!
22	queue_dir = path.resolve(config.get('queue_dir'))	×
23	} else if (process.env.HARAKA) {	1!
24	queue_dir = path.resolve(process.env.HARAKA, 'queue')	×
25	} else {	1✔
26	queue_dir = path.resolve('test', 'test-queue')	1✔
27	}	1✔
28		1✔
29	exports.queue_dir = queue_dir	1✔
30		1✔
31	const load_queue = async.queue((file, cb) => {	1✔
32	const hmail = new HMailItem(file, path.join(queue_dir, file))	×
33	exports._add_hmail(hmail)	×
34	hmail.once('ready', cb)	×
35	}, obc.cfg.concurrency_max)	1✔
36		1✔
37	let in_progress = 0	1✔
38	const delivery_queue = (exports.delivery_queue = async.queue((hmail, cb) => {	1✔
39	in_progress++	×
40	hmail.next_cb = () => {	×
41	in_progress--	×
42	cb()	×
43	}	×
44	if (obtls.cfg) return hmail.send()	×
45	obtls.init(() => {	×
46	hmail.send()	×
47	})	×
48	}, obc.cfg.concurrency_max))	1✔
49		1✔
50	const temp_fail_queue = (exports.temp_fail_queue = new TimerQueue())	1✔
51		1✔
52	let queue_count = 0	1✔
53		1✔
54	exports.get_stats = () => `${in_progress}/${exports.delivery_queue.length()}/${exports.temp_fail_queue.length()}`	1✔
55		1✔
56	exports.list_queue = (cb) => {	1✔
57	exports._load_cur_queue(null, exports._list_file, cb)	1✔
58	}	1✔
59		1✔
60	exports._stat_file = (file, cb) => {	1✔
61	queue_count++	×
62	setImmediate(cb)	×
63	}	×
64		1✔
65	exports.stat_queue = (cb) => {	1✔
66	const self = exports	×
67	exports._load_cur_queue(null, exports._stat_file, (err) => {	×
68	if (err) return cb(err)	×
69	return cb(null, self.stats())	×
70	})	×
71	}	×
72		1✔
73	exports.load_queue = (pid) => {	1✔
74	// Initialise and load queue	7✔
75	// This function is called first when not running under cluster,	7✔
76	exports.ensure_queue_dir()	7✔
77	exports.delete_dot_files()	7✔
78		7✔
79	exports._load_cur_queue(pid, exports._add_file, () => {	7✔
80	logger.info(exports, `[pid: ${pid}] ${delivery_queue.length()} files in my delivery queue`)	7✔
81	logger.info(exports, `[pid: ${pid}] ${load_queue.length()} files in my load queue`)	7✔
82	logger.info(exports, `[pid: ${pid}] ${temp_fail_queue.length()} files in my temp fail queue`)	7✔
83	})	7✔
84	}	7✔
85		1✔
86	exports._load_cur_queue = (pid, iteratee, cb) => {	1✔
87	logger.info(exports, 'Loading outbound queue from ', queue_dir)	8✔
88	fs.readdir(queue_dir, (err, files) => {	8✔
89	if (err) {	8!
90	return logger.error(exports, `Failed to load queue directory (${queue_dir}): ${err}`)	×
91	}	×
92		8✔
93	this.cur_time = new Date() // set once so we're not calling it a lot	8✔
94		8✔
95	this.load_queue_files(pid, files, iteratee, cb)	8✔
96	})	8✔
97	}	8✔
98		1✔
99	exports.read_parts = (file) => {	1✔
100	if (file.startsWith(_qfile.platformDOT)) {	1!
101	logger.warn(exports, `'Skipping' dot-file in queue folder: ${file}`)	×
102	return false	×
103	}	×
104		1✔
105	if (file.startsWith('error.')) {	1!
106	logger.warn(exports, `'Skipping' error file in queue folder: ${file}`)	×
107	return false	×
108	}	×
109		1✔
110	const parts = _qfile.parts(file)	1✔
111	if (!parts) {	1✔
112	logger.error(exports, `Unrecognized file in queue folder: ${file}`)	1✔
113	return false	1✔
114	}	1✔
115		×
116	return parts	×
117	}	1✔
118		1✔
119	exports.rename_to_actual_pid = (file, parts, cb) => {	1✔
120	// maintain some original details for the rename	×
121	const new_filename = _qfile.name({	×
122	arrival: parts.arrival,	×
123	uid: parts.uid,	×
124	next_attempt: parts.next_attempt,	×
125	attempts: parts.attempts,	×
126	})	×
127		×
128	fs.rename(path.join(queue_dir, file), path.join(queue_dir, new_filename), (err) => {	×
129	if (err) {	×
130	return cb(`Unable to rename queue file: ${file} to ${new_filename} : ${err}`)	×
131	}	×
132		×
133	cb(null, new_filename)	×
134	})	×
135	}	×
136		1✔
137	exports._add_file = (file, cb) => {	1✔
138	const self = exports	×
139	const parts = _qfile.parts(file)	×
140		×
141	if (parts.next_attempt <= self.cur_time) {	×
142	logger.debug(exports, `File ${file} needs processing now`)	×
143	load_queue.push(file)	×
144	} else {	×
145	logger.debug(exports, `File ${file} needs processing later: ${parts.next_attempt - self.cur_time}ms`)	×
146	temp_fail_queue.add(file, parts.next_attempt - self.cur_time, () => {	×
147	load_queue.push(file)	×
148	})	×
149	}	×
150		×
151	cb()	×
152	}	×
153		1✔
154	exports.load_queue_files = (pid, input_files, iteratee, callback = function () {}) => {	1✔
155	const self = exports	8✔
156	const searchPid = parseInt(pid)	8✔
157		8✔
158	let stat_renamed = 0	8✔
159	let stat_loaded = 0	8✔
160		8✔
161	if (searchPid) {	8!
162	logger.info(exports, `Grabbing queue files for pid: ${pid}`)	×
163	} else {	8✔
164	logger.info(exports, 'Loading the queue...')	8✔
165	}	8✔
166		8✔
167	async.map(	8✔
168	input_files,	8✔
169	(file, cb) => {	8✔
170	const parts = self.read_parts(file)	1✔
171	if (!parts) return cb()	1✔
172		×
173	if (searchPid) {	×
174	if (parts.pid !== searchPid) return cb()	×
175		×
176	self.rename_to_actual_pid(file, parts, (error, renamed_file) => {	×
177	if (error) {	×
178	logger.error(exports, `${error}`)	×
179	return cb()	×
180	}	×
181		×
182	stat_renamed++	×
183	stat_loaded++	×
184	cb(null, renamed_file)	×
185	})	×
186	} else {	×
187	stat_loaded++	×
188	cb(null, file)	×
189	}	×
190	},	8✔
191	(err, results) => {	8✔
192	if (err) logger.err(exports, `[pid: ${pid}] ${err}`)	8!
193	if (searchPid) logger.info(exports, `[pid: ${pid}] ${stat_renamed} files old PID queue fixed up`)	8!
194	logger.debug(exports, `[pid: ${pid}] ${stat_loaded} files loaded`)	8✔
195		8✔
196	async.map(	8✔
197	results.filter((i) => i),	8✔
198	iteratee,	8✔
199	callback,	8✔
200	)	8✔
201	},	8✔
202	)	8✔
203	}	8✔
204		1✔
205	exports.stats = () => {	1✔
206	return {	×
207	queue_dir,	×
208	queue_count,	×
209	}	×
210	}	×
211		1✔
212	exports._list_file = (file, cb) => {	1✔
213	const tl_reader = fs.createReadStream(path.join(queue_dir, file), {	×
214	start: 0,	×
215	end: 3,	×
216	})	×
217	tl_reader.on('error', (err) => {	×
218	console.error(`Error reading queue file: ${file}:`, err)	×
219	})	×
220	tl_reader.once('data', (buf) => {	×
221	// I'm making the assumption here we won't ever read less than 4 bytes	×
222	// as no filesystem on the planet should be that dumb...	×
223	tl_reader.destroy()	×
224	const todo_len = (buf[0] << 24) + (buf[1] << 16) + (buf[2] << 8) + buf[3]	×
225	const td_reader = fs.createReadStream(path.join(queue_dir, file), {	×
226	encoding: 'utf8',	×
227	start: 4,	×
228	end: todo_len + 3,	×
229	})	×
230	let todo = ''	×
231	td_reader.on('data', (str) => {	×
232	todo += str	×
233	if (Buffer.byteLength(todo) === todo_len) {	×
234	// we read everything	×
235	const todo_struct = JSON.parse(todo)	×
236	todo_struct.rcpt_to = todo_struct.rcpt_to.map((a) => new Address(a))	×
237	todo_struct.mail_from = new Address(todo_struct.mail_from)	×
238	todo_struct.file = file	×
239	todo_struct.full_path = path.join(queue_dir, file)	×
240	const parts = _qfile.parts(file)	×
241	todo_struct.pid = parts?.pid \|\| null	×
242	cb(null, todo_struct)	×
243	}	×
244	})	×
245	td_reader.on('end', () => {	×
246	if (Buffer.byteLength(todo) !== todo_len) {	×
247	console.error("Didn't find right amount of data in todo for file:", file)	×
248	return cb()	×
249	}	×
250	})	×
251	})	×
252	}	×
253		1✔
254	exports.flush_queue = (domain, pid) => {	1✔
255	if (domain) {	×
256	exports.list_queue((err, qlist) => {	×
257	if (err) return logger.error(exports, `Failed to load queue: ${err}`)	×
258	for (const todo of qlist) {	×
259	if (todo.domain.toLowerCase() != domain.toLowerCase()) return	×
260	if (pid && todo.pid != pid) return	×
261	// console.log("requeue: ", todo);	×
262	delivery_queue.push(new HMailItem(todo.file, todo.full_path))	×
263	}	×
264	})	×
265	} else {	×
266	temp_fail_queue.drain()	×
267	}	×
268	}	×
269		1✔
270	exports.load_pid_queue = (pid) => {	1✔
271	logger.info(exports, `Loading queue for pid: ${pid}`)	×
272	exports.load_queue(pid)	×
273	}	×
274		1✔
275	exports.ensure_queue_dir = () => {	1✔
276	// this code is only run at start-up.	7✔
277	if (fs.existsSync(queue_dir)) return	7✔
278		×
279	logger.debug(exports, `Creating queue directory ${queue_dir}`)	×
280	try {	×
281	fs.mkdirSync(queue_dir, 493) // 493 == 0755	×
282	const cfg = config.get('smtp.ini')	×
283	let uid	×
284	let gid	×
285	if (cfg.user) uid = child_process.execSync(`id -u ${cfg.user}`).toString().trim()	×
286	if (cfg.group) gid = child_process.execSync(`id -g ${cfg.group}`).toString().trim()	×
287	if (uid && gid) {	7!
288	fs.chown(queue_dir, uid, gid)	×
289	} else if (uid) {	×
290	fs.chown(queue_dir, uid)	×
291	}	×
292	} catch (err) {	7!
293	if (err.code !== 'EEXIST') {	×
294	logger.error(exports, `Error creating queue directory: ${err}`)	×
295	throw err	×
296	}	×
297	}	×
298	}	7✔
299		1✔
300	exports.delete_dot_files = () => {	1✔
301	for (const file of fs.readdirSync(queue_dir)) {	7!
NEW 302	if (file.startsWith(_qfile.platformDOT)) {	×
303	logger.warn(exports, `Removing left over dot-file: ${file}`)	×
304	return fs.unlinkSync(path.join(queue_dir, file))	×
305	}	×
306	}	×
307	}	7✔
308		1✔
309	exports._add_hmail = (hmail) => {	1✔
310	if (hmail.next_process < exports.cur_time) {	×
311	delivery_queue.push(hmail)	×
312	} else {	×
313	temp_fail_queue.add(hmail.filename, hmail.next_process - exports.cur_time, () => {	×
314	delivery_queue.push(hmail)	×
315	})	×
316	}	×
317	}	×
318		1✔
319	exports.scan_queue_pids = (cb) => {	1✔
320	const self = exports	×
321		×
322	// Under cluster, this is called first by the master	×
323	self.ensure_queue_dir()	×
324	self.delete_dot_files()	×
325		×
326	fs.readdir(queue_dir, (err, files) => {	×
327	if (err) {	×
328	logger.error(exports, `Failed to load queue directory (${queue_dir}): ${err}`)	×
329	return cb(err)	×
330	}	×
331		×
332	const pids = {}	×
333		×
334	for (const file of files) {	×
335	const parts = self.read_parts(file)	×
336	if (parts) pids[parts.pid] = true	×
337	}	×
338		×
339	return cb(null, Object.keys(pids))	×
340	})	×
341	}	×

haraka / Haraka / 22647221264

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous