diff --git a/cmd/init.go b/cmd/init.go index 3317ea11..b8c25952 100644 --- a/cmd/init.go +++ b/cmd/init.go @@ -286,18 +286,21 @@ func initCampaignManager(q *Queries, cs *constants, app *App) *manager.Manager { } return manager.New(manager.Config{ - BatchSize: ko.Int("app.batch_size"), - Concurrency: ko.Int("app.concurrency"), - MessageRate: ko.Int("app.message_rate"), - MaxSendErrors: ko.Int("app.max_send_errors"), - FromEmail: cs.FromEmail, - IndividualTracking: ko.Bool("privacy.individual_tracking"), - UnsubURL: cs.UnsubURL, - OptinURL: cs.OptinURL, - LinkTrackURL: cs.LinkTrackURL, - ViewTrackURL: cs.ViewTrackURL, - MessageURL: cs.MessageURL, - UnsubHeader: ko.Bool("privacy.unsubscribe_header"), + BatchSize: ko.Int("app.batch_size"), + Concurrency: ko.Int("app.concurrency"), + MessageRate: ko.Int("app.message_rate"), + MaxSendErrors: ko.Int("app.max_send_errors"), + FromEmail: cs.FromEmail, + IndividualTracking: ko.Bool("privacy.individual_tracking"), + UnsubURL: cs.UnsubURL, + OptinURL: cs.OptinURL, + LinkTrackURL: cs.LinkTrackURL, + 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) } diff --git a/cmd/settings.go b/cmd/settings.go index 8485c5a9..69b9f5eb 100644 --- a/cmd/settings.go +++ b/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"` diff --git a/frontend/src/views/Settings.vue b/frontend/src/views/Settings.vue index 66405131..9ce43d6b 100644 --- a/frontend/src/views/Settings.vue +++ b/frontend/src/views/Settings.vue @@ -92,6 +92,45 @@ name="app.max_send_errors" type="is-light" placeholder="1999" min="0" max="100000" /> + +
+
+
+ + + +
+ +
+ + + + +
+ +
+ + + + +
+
+
diff --git a/i18n/en.json b/i18n/en.json index c1ef2727..1ba3411d 100644 --- a/i18n/en.json +++ b/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", diff --git a/internal/manager/manager.go b/internal/manager/manager.go index a166dfd2..7dda703a 100644 --- a/internal/manager/manager.go +++ b/internal/manager/manager.go @@ -47,20 +47,26 @@ type Manager struct { logger *log.Logger // Campaigns that are currently running. - camps map[int]*models.Campaign - campsMutex sync.RWMutex + camps map[int]*models.Campaign + 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 + links map[string]string + 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, @@ -90,18 +96,21 @@ type Config struct { // Number of subscribers to pull from the DB in a single iteration. BatchSize int - Concurrency int - MessageRate int - MaxSendErrors int - RequeueOnError bool - FromEmail string - IndividualTracking bool - LinkTrackURL string - UnsubURL string - OptinURL string - MessageURL string - ViewTrackURL string - UnsubHeader bool + Concurrency int + MessageRate int + MaxSendErrors int + SlidingWindow bool + SlidingWindowDuration time.Duration + SlidingWindowRate int + RequeueOnError bool + FromEmail string + IndividualTracking bool + LinkTrackURL string + UnsubURL string + OptinURL string + MessageURL string + ViewTrackURL string + UnsubHeader bool } type msgError struct { @@ -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) } diff --git a/internal/migrations/v0.9.0.go b/internal/migrations/v0.9.0.go index 05304a53..30379a47 100644 --- a/internal/migrations/v0.9.0.go +++ b/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 diff --git a/schema.sql b/schema.sql index fcff5a18..b34b72b1 100644 --- a/schema.sql +++ b/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'),