Add support for rate limiting messages with a sliding window.

Certain SMTP hosts limit the total number of messages that can be
sent within a window, for instance, X / 24 hours. The concurrency
and message rate controls can only limit that to a max of
1 messages / second, without a global cap.

This commit introduces a simple sliding window rate limit feature
that counts the number of messages sent in a specific window, and
upon reaching that limit, waits for the window to reset before
any more messages are pushed out globally across any number of
campaigns.

Context: https://github.com/knadh/listmonk/issues/119

cmd/init.go

@@ -298,6 +298,9 @@ func initCampaignManager(q *Queries, cs *constants, app *App) *manager.Manager {
 		ViewTrackURL:          cs.ViewTrackURL,
 		MessageURL:            cs.MessageURL,
 		UnsubHeader:           ko.Bool("privacy.unsubscribe_header"),
+		SlidingWindow:         ko.Bool("app.message_sliding_window"),
+		SlidingWindowDuration: ko.Duration("app.message_sliding_window_duration"),
+		SlidingWindowRate:     ko.Int("app.message_sliding_window_rate"),
 	}, newManagerDB(q), campNotifCB, app.i18n, lo)
 
 }

cmd/settings.go

@@ -26,6 +26,10 @@ type settings struct {
 	AppMaxSendErrors int `json:"app.max_send_errors"`
 	AppMessageRate   int `json:"app.message_rate"`
 
+	AppMessageSlidingWindow         bool   `json:"app.message_sliding_window"`
+	AppMessageSlidingWindowDuration string `json:"app.message_sliding_window_duration"`
+	AppMessageSlidingWindowRate     int    `json:"app.message_sliding_window_rate"`
+
 	PrivacyIndividualTracking bool     `json:"privacy.individual_tracking"`
 	PrivacyUnsubHeader        bool     `json:"privacy.unsubscribe_header"`
 	PrivacyAllowBlocklist     bool     `json:"privacy.allow_blocklist"`

frontend/src/views/Settings.vue

@@ -92,6 +92,45 @@
                     name="app.max_send_errors" type="is-light"
                     placeholder="1999" min="0" max="100000" />
               </b-field>
+
+              <div>
+                <div class="columns">
+                  <div class="column is-6">
+                    <b-field :label="$t('settings.performance.slidingWindow')"
+                      :message="$t('settings.performance.slidingWindowHelp')">
+                      <b-switch v-model="form['app.message_sliding_window']"
+                          name="app.message_sliding_window" />
+                    </b-field>
+                  </div>
+
+                  <div class="column is-3"
+                    :class="{'disabled': !form['app.message_sliding_window']}">
+                    <b-field :label="$t('settings.performance.slidingWindowRate')"
+                      label-position="on-border"
+                      :message="$t('settings.performance.slidingWindowRateHelp')">
+
+                      <b-numberinput v-model="form['app.message_sliding_window_rate']"
+                        name="sliding_window_rate" type="is-light"
+                        controls-position="compact"
+                        :disabled="!form['app.message_sliding_window']"
+                        placeholder="25" min="1" max="10000000" />
+                    </b-field>
+                  </div>
+
+                  <div class="column is-3"
+                    :class="{'disabled': !form['app.message_sliding_window']}">
+                    <b-field :label="$t('settings.performance.slidingWindowDuration')"
+                      label-position="on-border"
+                      :message="$t('settings.performance.slidingWindowDurationHelp')">
+
+                      <b-input v-model="form['app.message_sliding_window_duration']"
+                        name="sliding_window_duration"
+                        :disabled="!form['app.message_sliding_window']"
+                        placeholder="1h" :pattern="regDuration" :maxlength="10" />
+                    </b-field>
+                  </div>
+                </div>
+              </div><!-- sliding window -->
             </div>
           </b-tab-item><!-- performance -->
 

i18n/en.json

@@ -282,6 +282,14 @@
     "settings.performance.messageRate": "Message rate",
     "settings.performance.messageRateHelp": "Maximum number of messages to be sent out per second per worker in a second. If concurrency = 10 and message_rate = 10, then up to 10x10=100 messages may be pushed out every second. This, along with concurrency, should be tweaked to keep the net messages going out per second under the target message servers rate limits if any.",
     "settings.performance.name": "Performance",
+
+    "settings.performance.slidingWindow": "Enable sliding window limit",
+    "settings.performance.slidingWindowHelp": "Limit the total number of messages that are sent out in given period. On reaching this limit, messages are be held from sending until the time window clears.",
+    "settings.performance.slidingWindowDuration": "Duration",
+    "settings.performance.slidingWindowDurationHelp": "Duration of the sliding window period (m for minute, h for hour)",
+    "settings.performance.slidingWindowRate": "Max. messages",
+    "settings.performance.slidingWindowRateHelp": "Maximum number of messages to send within the window duration",
+
     "settings.privacy.allowBlocklist": "Allow blocklisting",
     "settings.privacy.allowBlocklistHelp": "Allow subscribers to unsubscribe from all mailing lists and mark themselves as blocklisted?",
     "settings.privacy.allowExport": "Allow exporting",

internal/manager/manager.go

@@ -48,19 +48,25 @@ type Manager struct {
 
 	// Campaigns that are currently running.
 	camps    map[int]*models.Campaign
-	campsMutex sync.RWMutex
+	campsMut sync.RWMutex
 
 	// Links generated using Track() are cached here so as to not query
 	// the database for the link UUID for every message sent. This has to
 	// be locked as it may be used externally when previewing campaigns.
 	links    map[string]string
-	linksMutex sync.RWMutex
+	linksMut sync.RWMutex
 
 	subFetchQueue      chan *models.Campaign
 	campMsgQueue       chan CampaignMessage
 	campMsgErrorQueue  chan msgError
 	campMsgErrorCounts map[int]int
 	msgQueue           chan Message
+
+	// Sliding window keeps track of the total number of messages sent in a period
+	// and on reaching the specified limit, waits until the window is over before
+	// sending further messages.
+	slidingWindowNumMsg int
+	slidingWindowStart  time.Time
 }
 
 // CampaignMessage represents an instance of campaign message to be pushed out,
@@ -93,6 +99,9 @@ type Config struct {
 	Concurrency           int
 	MessageRate           int
 	MaxSendErrors         int
+	SlidingWindow         bool
+	SlidingWindowDuration time.Duration
+	SlidingWindowRate     int
 	RequeueOnError        bool
 	FromEmail             string
 	IndividualTracking    bool
@@ -135,6 +144,7 @@ func New(cfg Config, src DataSource, notifCB models.AdminNotifCallback, i *i18n.
 		msgQueue:           make(chan Message, cfg.Concurrency),
 		campMsgErrorQueue:  make(chan msgError, cfg.MaxSendErrors),
 		campMsgErrorCounts: make(map[int]int),
+		slidingWindowStart: time.Now(),
 	}
 }
 
@@ -185,8 +195,8 @@ func (m *Manager) HasMessenger(id string) bool {
 
 // HasRunningCampaigns checks if there are any active campaigns.
 func (m *Manager) HasRunningCampaigns() bool {
-	m.campsMutex.Lock()
-	defer m.campsMutex.Unlock()
+	m.campsMut.Lock()
+	defer m.campsMut.Unlock()
 	return len(m.camps) > 0
 }
 
@@ -422,28 +432,28 @@ func (m *Manager) addCampaign(c *models.Campaign) error {
 	}
 
 	// Add the campaign to the active map.
-	m.campsMutex.Lock()
+	m.campsMut.Lock()
 	m.camps[c.ID] = c
-	m.campsMutex.Unlock()
+	m.campsMut.Unlock()
 	return nil
 }
 
 // getPendingCampaignIDs returns the IDs of campaigns currently being processed.
 func (m *Manager) getPendingCampaignIDs() []int64 {
 	// Needs to return an empty slice in case there are no campaigns.
-	m.campsMutex.RLock()
+	m.campsMut.RLock()
 	ids := make([]int64, 0, len(m.camps))
 	for _, c := range m.camps {
 		ids = append(ids, int64(c.ID))
 	}
-	m.campsMutex.RUnlock()
+	m.campsMut.RUnlock()
 	return ids
 }
 
 // nextSubscribers processes the next batch of subscribers in a given campaign.
-// If returns a bool indicating whether there any subscribers were processed
-// in the current batch or not. This can happen when all the subscribers
-// have been processed, or if a campaign has been paused or cancelled abruptly.
+// It returns a bool indicating whether any subscribers were processed
+// in the current batch or not. A false indicates that all subscribers
+// have been processed, or that a campaign has been paused or cancelled.
 func (m *Manager) nextSubscribers(c *models.Campaign, batchSize int) (bool, error) {
 	// Fetch a batch of subscribers.
 	subs, err := m.src.NextSubscribers(c.ID, batchSize)
@@ -456,8 +466,14 @@ func (m *Manager) nextSubscribers(c *models.Campaign, batchSize int) (bool, erro
 		return false, nil
 	}
 
+	// Is there a sliding window limit configured?
+	hasSliding := m.cfg.SlidingWindow &&
+		m.cfg.SlidingWindowRate > 0 &&
+		m.cfg.SlidingWindowDuration.Seconds() > 1
+
 	// Push messages.
 	for _, s := range subs {
+		// Send the message.
 		msg := m.NewCampaignMessage(c, s)
 		if err := msg.Render(); err != nil {
 			m.logger.Printf("error rendering message (%s) (%s): %v", c.Name, s.Email, err)
@@ -467,6 +483,33 @@ func (m *Manager) nextSubscribers(c *models.Campaign, batchSize int) (bool, erro
 		// Push the message to the queue while blocking and waiting until
 		// the queue is drained.
 		m.campMsgQueue <- msg
+
+		// Check if the sliding window is active.
+		if hasSliding {
+			diff := time.Now().Sub(m.slidingWindowStart)
+
+			// Window has expired. Reset the clock.
+			if diff >= m.cfg.SlidingWindowDuration {
+				m.slidingWindowStart = time.Now()
+				m.slidingWindowNumMsg = 0
+				continue
+			}
+
+			// Have the messages exceeded the limit?
+			m.slidingWindowNumMsg++
+			if m.slidingWindowNumMsg >= m.cfg.SlidingWindowRate {
+				wait := m.cfg.SlidingWindowDuration - diff
+
+				m.logger.Printf("messages exceeded (%d) for the window (%v since %s). Sleeping for %s.",
+					m.slidingWindowNumMsg,
+					m.cfg.SlidingWindowDuration,
+					m.slidingWindowStart.Format(time.RFC822Z),
+					wait.Round(time.Second)*1)
+
+				m.slidingWindowNumMsg = 0
+				time.Sleep(wait)
+			}
+		}
 	}
 
 	return true, nil
@@ -474,16 +517,16 @@ func (m *Manager) nextSubscribers(c *models.Campaign, batchSize int) (bool, erro
 
 // isCampaignProcessing checks if the campaign is bing processed.
 func (m *Manager) isCampaignProcessing(id int) bool {
-	m.campsMutex.RLock()
+	m.campsMut.RLock()
 	_, ok := m.camps[id]
-	m.campsMutex.RUnlock()
+	m.campsMut.RUnlock()
 	return ok
 }
 
 func (m *Manager) exhaustCampaign(c *models.Campaign, status string) (*models.Campaign, error) {
-	m.campsMutex.Lock()
+	m.campsMut.Lock()
 	delete(m.camps, c.ID)
-	m.campsMutex.Unlock()
+	m.campsMut.Unlock()
 
 	// A status has been passed. Change the campaign's status
 	// without further checks.
@@ -520,12 +563,12 @@ func (m *Manager) exhaustCampaign(c *models.Campaign, status string) (*models.Ca
 // trackLink register a URL and return its UUID to be used in message templates
 // for tracking links.
 func (m *Manager) trackLink(url, campUUID, subUUID string) string {
-	m.linksMutex.RLock()
+	m.linksMut.RLock()
 	if uu, ok := m.links[url]; ok {
-		m.linksMutex.RUnlock()
+		m.linksMut.RUnlock()
 		return fmt.Sprintf(m.cfg.LinkTrackURL, uu, campUUID, subUUID)
 	}
-	m.linksMutex.RUnlock()
+	m.linksMut.RUnlock()
 
 	// Register link.
 	uu, err := m.src.CreateLink(url)
@@ -536,9 +579,9 @@ func (m *Manager) trackLink(url, campUUID, subUUID string) string {
 		return url
 	}
 
-	m.linksMutex.Lock()
+	m.linksMut.Lock()
 	m.links[url] = uu
-	m.linksMutex.Unlock()
+	m.linksMut.Unlock()
 
 	return fmt.Sprintf(m.cfg.LinkTrackURL, uu, campUUID, subUUID)
 }

internal/migrations/v0.9.0.go

@@ -9,7 +9,11 @@ import (
 // V0_9_0 performs the DB migrations for v.0.9.0.
 func V0_9_0(db *sqlx.DB, fs stuffbin.FileSystem, ko *koanf.Koanf) error {
 	_, err := db.Exec(`
-	INSERT INTO settings (key, value) VALUES ('app.lang', '"en"')
+	INSERT INTO settings (key, value) VALUES
+		('app.lang', '"en"'),
+		('app.message_sliding_window', 'false'),
+		('app.message_sliding_window_duration', '"1h"'),
+		('app.message_sliding_window_rate', '10000')
 		ON CONFLICT DO NOTHING;
 	`)
 	return err

schema.sql

@@ -173,6 +173,9 @@ INSERT INTO settings (key, value) VALUES
     ('app.message_rate', '10'),
     ('app.batch_size', '1000'),
     ('app.max_send_errors', '1000'),
+    ('app.message_sliding_window', 'false'),
+    ('app.message_sliding_window_duration', '"1h"'),
+    ('app.message_sliding_window_rate', '10000'),
     ('app.notify_emails', '["admin1@mysite.com", "admin2@mysite.com"]'),
     ('app.lang', '"en"'),
     ('privacy.individual_tracking', 'false'),