WA tiers paused

handlers/whatsapp/whatsapp.go

@@ -901,12 +901,12 @@ func sendWhatsAppMsg(rc redis.Conn, msg courier.Msg, sendPath *url.URL, payload
 
 	if rr.StatusCode == 429 || rr.StatusCode == 503 {
 		rateLimitKey := fmt.Sprintf("rate_limit:%s", msg.Channel().UUID().String())
-		rc.Do("set", rateLimitKey, "engaged")
+		rc.Do("SET", rateLimitKey, "engaged")
 
 		// The rate limit is 50 requests per second
 		// We pause sending 2 seconds so the limit count is reset
 		// TODO: In the future we should the header value when available
-		rc.Do("expire", rateLimitKey, 2)
+		rc.Do("EXPIRE", rateLimitKey, 2)
 
 		log := courier.NewChannelLogFromRR("rate limit engaged", msg.Channel(), msg.ID(), rr).WithError("Message Send Error", err)
 		return "", "", []*courier.ChannelLog{log}, err
@@ -918,6 +918,18 @@ func sendWhatsAppMsg(rc redis.Conn, msg courier.Msg, sendPath *url.URL, payload
 
 	// handle send msg errors
 	if err == nil && len(errPayload.Errors) > 0 {
+		if hasTiersError(*errPayload) {
+			rateLimitBulkKey := fmt.Sprintf("rate_limit_bulk:%s", msg.Channel().UUID().String())
+			rc.Do("SET", rateLimitBulkKey, "engaged")
+
+			// The WA tiers spam rate limit hit
+			// We pause the bulk queue for 24 hours and 5min
+			rc.Do("EXPIRE", rateLimitBulkKey, (60*60*24)+(5*60))
+
+			err := errors.Errorf("received error from send endpoint: %s", errPayload.Errors[0].Title)
+			return "", "", []*courier.ChannelLog{log}, err
+		}
+
 		if !hasWhatsAppContactError(*errPayload) {
 			err := errors.Errorf("received error from send endpoint: %s", errPayload.Errors[0].Title)
 			return "", "", []*courier.ChannelLog{log}, err
@@ -1022,6 +1034,15 @@ func buildWhatsAppHeaders(channel courier.Channel) http.Header {
 	return header
 }
 
+func hasTiersError(payload mtErrorPayload) bool {
+	for _, err := range payload.Errors {
+		if err.Code == 471 {
+			return true
+		}
+	}
+	return false
+}
+
 func hasWhatsAppContactError(payload mtErrorPayload) bool {
 	for _, err := range payload.Errors {
 		if err.Code == 1006 && err.Title == "Resource not found" && err.Details == "unknown contact" {

queue/queue.go

@@ -123,6 +123,14 @@ var luaPop = redis.NewScript(2, `-- KEYS: [EpochMS QueueType]
 
 	-- if we didn't find one, try again from our bulk queue
 	if not result[1] or isFutureResult then
+		-- check if we are rate limited for bulk queue
+		local rateLimitBulkKey = "rate_limit_bulk:" .. queueName
+		local rateLimitBulk = redis.call("get", rateLimitBulkKey)
+		if rateLimitBulk then
+			return {"retry", ""}
+		end
+
+		-- we are not pause check our bulk queue
 		local bulkQueue = queue .. "/0"
 		local bulkResult = redis.call("zrangebyscore", bulkQueue, 0, "+inf", "WITHSCORES", "LIMIT", 0, 1)
 

queue/queue_test.go

@@ -60,6 +60,19 @@ func TestLua(t *testing.T) {
 	delay := time.Second*2 - time.Duration(time.Now().UnixNano()%int64(time.Second))
 	time.Sleep(delay)
 
+	conn.Do("SET", "rate_limit_bulk:chan1", "engaged")
+	conn.Do("EXPIRE", "rate_limit_bulk:chan1", 5)
+
+	// we have the rate limit set,
+	queue, value, err := PopFromQueue(conn, "msgs")
+	assert.NoError(err)
+	if value != "" && queue != EmptyQueue {
+		t.Fatal("Should be paused")
+	}
+
+	// When the redis paused key is remove, we get the values from bulk queue/low priority
+	conn.Do("DEL", "rate_limit_bulk:chan1")
+
 	// pop 10 items off
 	for i := 0; i < 10; i++ {
 		queue, value, err := PopFromQueue(conn, "msgs")
@@ -69,7 +82,8 @@ func TestLua(t *testing.T) {
 	}
 
 	// next value should be throttled
-	queue, value, err := PopFromQueue(conn, "msgs")
+	queue, value, err = PopFromQueue(conn, "msgs")
+	assert.NoError(err)
 	if value != "" && queue != EmptyQueue {
 		t.Fatal("Should be throttled")
 	}
@@ -85,6 +99,7 @@ func TestLua(t *testing.T) {
 
 	// adding more items shouldn't change that
 	queue, value, err = PopFromQueue(conn, "msgs")
+	assert.NoError(err)
 	if value != "" && queue != EmptyQueue {
 		t.Fatal("Should be throttled")
 	}
@@ -104,11 +119,18 @@ func TestLua(t *testing.T) {
 	err = PushOntoQueue(conn, "msgs", "chan1", rate, `[{"id":31}]`, HighPriority)
 	assert.NoError(err)
 
+	// make sure pause bulk key do not prevent use to get from the high priority queue
+	conn.Do("SET", "rate_limit_bulk:chan1", "engaged")
+	conn.Do("EXPIRE", "rate_limit_bulk:chan1", 5)
+
 	queue, value, err = PopFromQueue(conn, "msgs")
 	assert.NoError(err)
 	assert.Equal(WorkerToken("msgs:chan1|10"), queue)
 	assert.Equal(`{"id":31}`, value)
 
+	// make sure paused is not present for more tests
+	conn.Do("DEL", "rate_limit_bulk:chan1")
+
 	// should get next five bulk msgs fine
 	for i := 10; i < 15; i++ {
 		queue, value, err := PopFromQueue(conn, "msgs")
@@ -119,6 +141,7 @@ func TestLua(t *testing.T) {
 
 	// push on a compound message
 	err = PushOntoQueue(conn, "msgs", "chan1", rate, `[{"id":32}, {"id":33}]`, HighPriority)
+	assert.NoError(err)
 
 	queue, value, err = PopFromQueue(conn, "msgs")
 	assert.NoError(err)
@@ -171,17 +194,19 @@ func TestLua(t *testing.T) {
 	err = PushOntoQueue(conn, "msgs", "chan1", rate, `[{"id":34}]`, HighPriority)
 	assert.NoError(err)
 
-	conn.Do("set", "rate_limit:chan1", "engaged")
+	conn.Do("SET", "rate_limit:chan1", "engaged")
 	conn.Do("EXPIRE", "rate_limit:chan1", 5)
 
 	// we have the rate limit set,
 	queue, value, err = PopFromQueue(conn, "msgs")
+	assert.NoError(err)
 	if value != "" && queue != EmptyQueue {
 		t.Fatal("Should be throttled")
 	}
 
 	time.Sleep(2 * time.Second)
 	queue, value, err = PopFromQueue(conn, "msgs")
+	assert.NoError(err)
 	if value != "" && queue != EmptyQueue {
 		t.Fatal("Should be throttled")
 	}