From a848231ae0f346dc7cc000973fbeb65b0894ee92 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 10 Apr 2024 21:59:03 +0200 Subject: Adding upstream version 3.8.5. Signed-off-by: Daniel Baumann --- src/oqmgr/qmgr_active.c | 607 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 607 insertions(+) create mode 100644 src/oqmgr/qmgr_active.c (limited to 'src/oqmgr/qmgr_active.c') diff --git a/src/oqmgr/qmgr_active.c b/src/oqmgr/qmgr_active.c new file mode 100644 index 0000000..b1c1a4a --- /dev/null +++ b/src/oqmgr/qmgr_active.c @@ -0,0 +1,607 @@ +/*++ +/* NAME +/* qmgr_active 3 +/* SUMMARY +/* active queue management +/* SYNOPSIS +/* #include "qmgr.h" +/* +/* void qmgr_active_feed(scan_info, queue_id) +/* QMGR_SCAN *scan_info; +/* const char *queue_id; +/* +/* void qmgr_active_drain() +/* +/* int qmgr_active_done(message) +/* QMGR_MESSAGE *message; +/* DESCRIPTION +/* These functions maintain the active message queue: the set +/* of messages that the queue manager is actually working on. +/* The active queue is limited in size. Messages are drained +/* from the active queue by allocating a delivery process and +/* by delivering mail via that process. Messages leak into the +/* active queue only when the active queue is small enough. +/* Damaged message files are saved to the "corrupt" directory. +/* +/* qmgr_active_feed() inserts the named message file into +/* the active queue. Message files with the wrong name or +/* with other wrong properties are skipped but not removed. +/* The following queue flags are recognized, other flags being +/* ignored: +/* .IP QMGR_SCAN_ALL +/* Examine all queue files. Normally, deferred queue files with +/* future time stamps are ignored, and incoming queue files with +/* future time stamps are frowned upon. +/* .PP +/* qmgr_active_drain() allocates one delivery process. +/* Process allocation is asynchronous. Once the delivery +/* process is available, an attempt is made to deliver +/* a message via it. Message delivery is asynchronous, too. +/* +/* qmgr_active_done() deals with a message after delivery +/* has been tried for all in-core recipients. If the message +/* was bounced, a bounce message is sent to the sender, or +/* to the Errors-To: address if one was specified. +/* If there are more on-file recipients, a new batch of +/* in-core recipients is read from the queue file. Otherwise, +/* if a delivery agent marked the queue file as corrupt, +/* the queue file is moved to the "corrupt" queue (surprise); +/* if at least one delivery failed, the message is moved +/* to the deferred queue. The time stamps of a deferred queue +/* file are set to the nearest wakeup time of its recipient +/* sites (if delivery failed due to a problem with a next-hop +/* host), are set into the future by the amount of time the +/* message was queued (per-message exponential backoff), or are set +/* into the future by a minimal backoff time, whichever is more. +/* The minimal_backoff_time parameter specifies the minimal +/* amount of time between delivery attempts; maximal_backoff_time +/* specifies an upper limit. +/* DIAGNOSTICS +/* Fatal: queue file access failures, out of memory. +/* Panic: interface violations, internal consistency errors. +/* Warnings: corrupt message file. A corrupt message is saved +/* to the "corrupt" queue for further inspection. +/* LICENSE +/* .ad +/* .fi +/* The Secure Mailer license must be distributed with this software. +/* AUTHOR(S) +/* Wietse Venema +/* IBM T.J. Watson Research +/* P.O. Box 704 +/* Yorktown Heights, NY 10598, USA +/* +/* Wietse Venema +/* Google, Inc. +/* 111 8th Avenue +/* New York, NY 10011, USA +/*--*/ + +/* System library. */ + +#include +#include +#include +#include +#include +#include +#include +#include + +#ifndef S_IRWXU /* What? no POSIX system? */ +#define S_IRWXU 0700 +#endif + +/* Utility library. */ + +#include +#include +#include +#include +#include + +/* Global library. */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* Application-specific. */ + +#include "qmgr.h" + + /* + * A bunch of call-back routines. + */ +static void qmgr_active_done_2_bounce_flush(int, void *); +static void qmgr_active_done_2_generic(QMGR_MESSAGE *); +static void qmgr_active_done_25_trace_flush(int, void *); +static void qmgr_active_done_25_generic(QMGR_MESSAGE *); +static void qmgr_active_done_3_defer_flush(int, void *); +static void qmgr_active_done_3_defer_warn(int, void *); +static void qmgr_active_done_3_generic(QMGR_MESSAGE *); + +/* qmgr_active_corrupt - move corrupted file out of the way */ + +static void qmgr_active_corrupt(const char *queue_id) +{ + const char *myname = "qmgr_active_corrupt"; + + if (mail_queue_rename(queue_id, MAIL_QUEUE_ACTIVE, MAIL_QUEUE_CORRUPT)) { + if (errno != ENOENT) + msg_fatal("%s: save corrupt file queue %s id %s: %m", + myname, MAIL_QUEUE_ACTIVE, queue_id); + } else { + msg_warn("saving corrupt file \"%s\" from queue \"%s\" to queue \"%s\"", + queue_id, MAIL_QUEUE_ACTIVE, MAIL_QUEUE_CORRUPT); + } +} + +/* qmgr_active_defer - defer queue file */ + +static void qmgr_active_defer(const char *queue_name, const char *queue_id, + const char *dest_queue, int delay) +{ + const char *myname = "qmgr_active_defer"; + const char *path; + struct utimbuf tbuf; + + if (msg_verbose) + msg_info("wakeup %s after %ld secs", queue_id, (long) delay); + + tbuf.actime = tbuf.modtime = event_time() + delay; + path = mail_queue_path((VSTRING *) 0, queue_name, queue_id); + if (utime(path, &tbuf) < 0 && errno != ENOENT) + msg_fatal("%s: update %s time stamps: %m", myname, path); + if (mail_queue_rename(queue_id, queue_name, dest_queue)) { + if (errno != ENOENT) + msg_fatal("%s: rename %s from %s to %s: %m", myname, + queue_id, queue_name, dest_queue); + msg_warn("%s: rename %s from %s to %s: %m", myname, + queue_id, queue_name, dest_queue); + } else if (msg_verbose) { + msg_info("%s: defer %s", myname, queue_id); + } +} + +/* qmgr_active_feed - feed one message into active queue */ + +int qmgr_active_feed(QMGR_SCAN *scan_info, const char *queue_id) +{ + const char *myname = "qmgr_active_feed"; + QMGR_MESSAGE *message; + struct stat st; + const char *path; + + if (strcmp(scan_info->queue, MAIL_QUEUE_ACTIVE) == 0) + msg_panic("%s: bad queue %s", myname, scan_info->queue); + if (msg_verbose) + msg_info("%s: queue %s", myname, scan_info->queue); + + /* + * Make sure this is something we are willing to open. + */ + if (mail_open_ok(scan_info->queue, queue_id, &st, &path) == MAIL_OPEN_NO) + return (0); + + if (msg_verbose) + msg_info("%s: %s", myname, path); + + /* + * Skip files that have time stamps into the future. They need to cool + * down. Incoming and deferred files can have future time stamps. + */ + if ((scan_info->flags & QMGR_SCAN_ALL) == 0 + && st.st_mtime > time((time_t *) 0) + 1) { + if (msg_verbose) + msg_info("%s: skip %s (%ld seconds)", myname, queue_id, + (long) (st.st_mtime - event_time())); + return (0); + } + + /* + * Move the message to the active queue. File access errors are fatal. + */ + if (mail_queue_rename(queue_id, scan_info->queue, MAIL_QUEUE_ACTIVE)) { + if (errno != ENOENT) + msg_fatal("%s: %s: rename from %s to %s: %m", myname, + queue_id, scan_info->queue, MAIL_QUEUE_ACTIVE); + msg_warn("%s: %s: rename from %s to %s: %m", myname, + queue_id, scan_info->queue, MAIL_QUEUE_ACTIVE); + return (0); + } + + /* + * Extract envelope information: sender and recipients. At this point, + * mail addresses have been processed by the cleanup service so they + * should be in canonical form. Generate requests to deliver this + * message. + * + * Throwing away queue files seems bad, especially when they made it this + * far into the mail system. Therefore we save bad files to a separate + * directory for further inspection. + * + * After queue manager restart it is possible that a queue file is still + * being delivered. In that case (the file is locked), defer delivery by + * a minimal amount of time. + */ +#define QMGR_FLUSH_AFTER (QMGR_FLUSH_EACH | QMGR_FLUSH_DFXP) +#define MAYBE_FLUSH_AFTER(mode) \ + (((mode) & MAIL_QUEUE_STAT_UNTHROTTLE) ? QMGR_FLUSH_AFTER : 0) +#define MAYBE_FORCE_EXPIRE(mode) \ + (((mode) & MAIL_QUEUE_STAT_EXPIRE) ? QMGR_FORCE_EXPIRE : 0) +#define MAYBE_UPDATE_MODE(mode) \ + (((mode) & MAIL_QUEUE_STAT_UNTHROTTLE) ? \ + (mode) & ~MAIL_QUEUE_STAT_UNTHROTTLE : 0) + + if ((message = qmgr_message_alloc(MAIL_QUEUE_ACTIVE, queue_id, + scan_info->flags + | MAYBE_FLUSH_AFTER(st.st_mode) + | MAYBE_FORCE_EXPIRE(st.st_mode), + MAYBE_UPDATE_MODE(st.st_mode))) == 0) { + qmgr_active_corrupt(queue_id); + return (0); + } else if (message == QMGR_MESSAGE_LOCKED) { + qmgr_active_defer(MAIL_QUEUE_ACTIVE, queue_id, MAIL_QUEUE_INCOMING, 60); + return (0); + } else { + + /* + * Special case if all recipients were already delivered. Send any + * bounces and clean up. + */ + if (message->refcount == 0) + qmgr_active_done(message); + return (1); + } +} + +/* qmgr_active_done - dispose of message after recipients have been tried */ + +void qmgr_active_done(QMGR_MESSAGE *message) +{ + const char *myname = "qmgr_active_done"; + struct stat st; + + if (msg_verbose) + msg_info("%s: %s", myname, message->queue_id); + + /* + * During a previous iteration, an attempt to bounce this message may + * have failed, so there may still be a bounce log lying around. XXX By + * groping around in the bounce queue, we're trespassing on the bounce + * service's territory. But doing so is more robust than depending on the + * bounce daemon to do the lookup for us, and for us to do the deleting + * after we have received a successful status from the bounce service. + * The bounce queue directory blocks are most likely in memory anyway. If + * these lookups become a performance problem we will have to build an + * in-core cache into the bounce daemon. + * + * Don't bounce when the bounce log is empty. The bounce process obviously + * failed, and the delivery agent will have requested that the message be + * deferred. + * + * Bounces are sent asynchronously to avoid stalling while the cleanup + * daemon waits for the qmgr to accept the "new mail" trigger. + * + * See also code in cleanup_bounce.c. + */ + if (stat(mail_queue_path((VSTRING *) 0, MAIL_QUEUE_BOUNCE, message->queue_id), &st) == 0) { + if (st.st_size == 0) { + if (mail_queue_remove(MAIL_QUEUE_BOUNCE, message->queue_id)) + msg_fatal("remove %s %s: %m", + MAIL_QUEUE_BOUNCE, message->queue_id); + } else { + if (msg_verbose) + msg_info("%s: bounce %s", myname, message->queue_id); + if (message->verp_delims == 0 || var_verp_bounce_off) + abounce_flush(BOUNCE_FLAG_KEEP, + message->queue_name, + message->queue_id, + message->encoding, + message->smtputf8, + message->sender, + message->dsn_envid, + message->dsn_ret, + qmgr_active_done_2_bounce_flush, + (void *) message); + else + abounce_flush_verp(BOUNCE_FLAG_KEEP, + message->queue_name, + message->queue_id, + message->encoding, + message->smtputf8, + message->sender, + message->dsn_envid, + message->dsn_ret, + message->verp_delims, + qmgr_active_done_2_bounce_flush, + (void *) message); + return; + } + } + + /* + * Asynchronous processing does not reach this point. + */ + qmgr_active_done_2_generic(message); +} + +/* qmgr_active_done_2_bounce_flush - process abounce_flush() status */ + +static void qmgr_active_done_2_bounce_flush(int status, void *context) +{ + QMGR_MESSAGE *message = (QMGR_MESSAGE *) context; + + /* + * Process abounce_flush() status and continue processing. + */ + message->flags |= status; + qmgr_active_done_2_generic(message); +} + +/* qmgr_active_done_2_generic - continue processing */ + +static void qmgr_active_done_2_generic(QMGR_MESSAGE *message) +{ + const char *path; + struct stat st; + + /* + * A delivery agent marks a queue file as corrupt by changing its + * attributes, and by pretending that delivery was deferred. + */ + if (message->flags + && mail_open_ok(MAIL_QUEUE_ACTIVE, message->queue_id, &st, &path) == MAIL_OPEN_NO) { + qmgr_active_corrupt(message->queue_id); + qmgr_message_free(message); + return; + } + + /* + * If we did not read all recipients from this file, go read some more, + * but remember whether some recipients have to be tried again. + * + * Throwing away queue files seems bad, especially when they made it this + * far into the mail system. Therefore we save bad files to a separate + * directory for further inspection by a human being. + */ + if (message->rcpt_offset > 0) { + if (qmgr_message_realloc(message) == 0) { + qmgr_active_corrupt(message->queue_id); + qmgr_message_free(message); + } else { + if (message->refcount == 0) + qmgr_active_done(message); /* recurse for consistency */ + } + return; + } + + /* + * XXX With multi-recipient mail, some recipients may have NOTIFY=SUCCESS + * and others not. Depending on what subset of recipients are delivered, + * a trace file may or may not be created. Even when the last partial + * delivery attempt had no NOTIFY=SUCCESS recipients, a trace file may + * still exist from a previous partial delivery attempt. So as long as + * any recipient has NOTIFY=SUCCESS we have to always look for the trace + * file and be prepared for the file not to exist. + * + * See also comments in bounce/bounce_notify_util.c. + */ + if ((message->tflags & (DEL_REQ_FLAG_USR_VRFY | DEL_REQ_FLAG_RECORD + | DEL_REQ_FLAG_REC_DLY_SENT)) + || (message->rflags & QMGR_READ_FLAG_NOTIFY_SUCCESS)) { + atrace_flush(message->tflags, + message->queue_name, + message->queue_id, + message->encoding, + message->smtputf8, + message->sender, + message->dsn_envid, + message->dsn_ret, + qmgr_active_done_25_trace_flush, + (void *) message); + return; + } + + /* + * Asynchronous processing does not reach this point. + */ + qmgr_active_done_25_generic(message); +} + +/* qmgr_active_done_25_trace_flush - continue after atrace_flush() completion */ + +static void qmgr_active_done_25_trace_flush(int status, void *context) +{ + QMGR_MESSAGE *message = (QMGR_MESSAGE *) context; + + /* + * Process atrace_flush() status and continue processing. + */ + if (status == 0 && message->tflags_offset) + qmgr_message_kill_record(message, message->tflags_offset); + message->flags |= status; + qmgr_active_done_25_generic(message); +} + +/* qmgr_active_done_25_generic - continue processing */ + +static void qmgr_active_done_25_generic(QMGR_MESSAGE *message) +{ + const char *myname = "qmgr_active_done_25_generic"; + const char *expire_status = 0; + + /* + * If we get to this point we have tried all recipients for this message. + * If the message is too old, try to bounce it. + * + * Bounces are sent asynchronously to avoid stalling while the cleanup + * daemon waits for the qmgr to accept the "new mail" trigger. + */ + if (message->flags) { + if ((message->qflags & QMGR_FORCE_EXPIRE) != 0) { + expire_status = "force-expired"; + } else if (event_time() >= message->create_time + + (*message->sender ? var_max_queue_time : var_dsn_queue_time)) { + expire_status = "expired"; + } else { + expire_status = 0; + } + if (expire_status != 0) { + msg_info("%s: from=<%s>, status=%s, returned to sender", + message->queue_id, info_log_addr_form_sender(message->sender), + expire_status); + if (message->verp_delims == 0 || var_verp_bounce_off) + adefer_flush(BOUNCE_FLAG_KEEP, + message->queue_name, + message->queue_id, + message->encoding, + message->smtputf8, + message->sender, + message->dsn_envid, + message->dsn_ret, + qmgr_active_done_3_defer_flush, + (void *) message); + else + adefer_flush_verp(BOUNCE_FLAG_KEEP, + message->queue_name, + message->queue_id, + message->encoding, + message->smtputf8, + message->sender, + message->dsn_envid, + message->dsn_ret, + message->verp_delims, + qmgr_active_done_3_defer_flush, + (void *) message); + return; + } else if (message->warn_time > 0 + && event_time() >= message->warn_time - 1) { + if (msg_verbose) + msg_info("%s: sending defer warning for %s", myname, message->queue_id); + adefer_warn(BOUNCE_FLAG_KEEP, + message->queue_name, + message->queue_id, + message->encoding, + message->smtputf8, + message->sender, + message->dsn_envid, + message->dsn_ret, + qmgr_active_done_3_defer_warn, + (void *) message); + return; + } + } + + /* + * Asynchronous processing does not reach this point. + */ + qmgr_active_done_3_generic(message); +} + +/* qmgr_active_done_3_defer_warn - continue after adefer_warn() completion */ + +static void qmgr_active_done_3_defer_warn(int status, void *context) +{ + QMGR_MESSAGE *message = (QMGR_MESSAGE *) context; + + /* + * Process adefer_warn() completion status and continue processing. + */ + if (status == 0) + qmgr_message_update_warn(message); + qmgr_active_done_3_generic(message); +} + +/* qmgr_active_done_3_defer_flush - continue after adefer_flush() completion */ + +static void qmgr_active_done_3_defer_flush(int status, void *context) +{ + QMGR_MESSAGE *message = (QMGR_MESSAGE *) context; + + /* + * Process adefer_flush() status and continue processing. + */ + message->flags = status; + qmgr_active_done_3_generic(message); +} + +/* qmgr_active_done_3_generic - continue processing */ + +static void qmgr_active_done_3_generic(QMGR_MESSAGE *message) +{ + const char *myname = "qmgr_active_done_3_generic"; + int delay; + + /* + * Some recipients need to be tried again. Move the queue file time + * stamps into the future by the amount of time that the message is + * delayed, and move the message to the deferred queue. Impose minimal + * and maximal backoff times. + * + * Since we look at actual time in queue, not time since last delivery + * attempt, backoff times will be distributed. However, we can still see + * spikes in delivery activity because the interval between deferred + * queue scans is finite. + */ + if (message->flags) { + if (message->create_time > 0) { + delay = event_time() - message->create_time; + if (delay > var_max_backoff_time) + delay = var_max_backoff_time; + if (delay < var_min_backoff_time) + delay = var_min_backoff_time; + } else { + delay = var_min_backoff_time; + } + qmgr_active_defer(message->queue_name, message->queue_id, + MAIL_QUEUE_DEFERRED, delay); + } + + /* + * All recipients done. Remove the queue file. + */ + else { + if (mail_queue_remove(message->queue_name, message->queue_id)) { + if (errno != ENOENT) + msg_fatal("%s: remove %s from %s: %m", myname, + message->queue_id, message->queue_name); + msg_warn("%s: remove %s from %s: %m", myname, + message->queue_id, message->queue_name); + } else { + /* Same format as logged by postsuper. */ + msg_info("%s: removed", message->queue_id); + } + } + + /* + * Finally, delete the in-core message structure. + */ + qmgr_message_free(message); +} + +/* qmgr_active_drain - drain active queue by allocating a delivery process */ + +void qmgr_active_drain(void) +{ + QMGR_TRANSPORT *transport; + + /* + * Allocate one delivery process for every transport with pending mail. + * The process allocation completes asynchronously. + */ + while ((transport = qmgr_transport_select()) != 0) { + if (msg_verbose) + msg_info("qmgr_active_drain: allocate %s", transport->name); + qmgr_transport_alloc(transport, qmgr_deliver); + } +} -- cgit v1.2.3